From a14bc9e6cdd70f38eaa2961d1ea775300b9577ea Mon Sep 17 00:00:00 2001 From: Eric Wong Date: Tue, 19 Jan 2021 09:34:31 +0000 Subject: lei q: fix augment of compressed mailboxes We need to delay writing out the mailbox until the compressor process is up and running, so have startq wait a bit. This means we must create the pipe early and hand it off to the workers before augmenting, despite spawning the gzip/pigz/xz/bzip2 process after augment is complete. --- lib/PublicInbox/LEI.pm | 1 + lib/PublicInbox/LeiToMail.pm | 19 +++++++++++-------- lib/PublicInbox/LeiXSearch.pm | 40 ++++++++++++++++++++++++++-------------- 3 files changed, 38 insertions(+), 22 deletions(-) (limited to 'lib/PublicInbox') diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm index e4f8bedb..f3edfe82 100644 --- a/lib/PublicInbox/LEI.pm +++ b/lib/PublicInbox/LEI.pm @@ -758,6 +758,7 @@ sub accept_dispatch { # Listener {post_accept} callback sub dclose { my ($self) = @_; delete $self->{lxs}; # stops LeiXSearch queries + close(delete $self->{1}) if $self->{1}; # may reap_compress $self->close if $self->{sock}; # PublicInbox::DS::close } diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm index 99388b5b..a6e517ea 100644 --- a/lib/PublicInbox/LeiToMail.pm +++ b/lib/PublicInbox/LeiToMail.pm @@ -200,18 +200,19 @@ sub zsfx2cmd ($$$) { } sub _post_augment_mbox { # open a compressor process - my ($self, $lei) = @_; + my ($self, $lei, $zpipe) = @_; my $zsfx = $self->{zsfx} or return; my $cmd = zsfx2cmd($zsfx, undef, $lei); - pipe(my ($r, $w)) or die "pipe: $!"; + my ($r, $w) = splice(@$zpipe, 0, 2); my $rdr = { 0 => $r, 1 => $lei->{1}, 2 => $lei->{2} }; my $pid = spawn($cmd, $lei->{env}, $rdr); - $lei->{"pid.$pid"} = $cmd; my $pp = gensym; - tie *$pp, 'PublicInbox::ProcessPipe', $pid, $w, \&reap_compress, $lei; + my $dup = bless { "pid.$pid" => $cmd }, ref($lei); + $dup->{$_} = $lei->{$_} for qw(2 sock); + tie *$pp, 'PublicInbox::ProcessPipe', $pid, $w, \&reap_compress, $dup; $lei->{1} = $pp; die 'BUG: unexpected {ovv}->{lock_path}' if $lei->{ovv}->{lock_path}; - $lei->{ovv}->ovv_out_lk_init if ($lei->{opt}->{jobs} // 2) > 1; + $lei->{ovv}->ovv_out_lk_init; } sub decompress_src ($$$) { @@ -395,7 +396,9 @@ sub _pre_augment_mbox { die "seek($dst): $!\n"; } state $zsfx_allow = join('|', keys %zsfx2cmd); - ($self->{zsfx}) = ($dst =~ /\.($zsfx_allow)\z/); + ($self->{zsfx}) = ($dst =~ /\.($zsfx_allow)\z/) or return; + pipe(my ($r, $w)) or die "pipe: $!"; + [ $r, $w ]; } sub _do_augment_mbox { @@ -433,10 +436,10 @@ sub do_augment { # slow, runs in wq worker } sub post_augment { # fast (spawn compressor or mkdir), runs in main daemon - my ($self, $lei) = @_; + my ($self, $lei, @args) = @_; # _post_augment_maildir, _post_augment_mbox my $m = "_post_augment_$self->{base_type}"; - $self->$m($lei); + $self->$m($lei, @args); } sub write_mail { # via ->wq_do diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm index 120857b8..002791c2 100644 --- a/lib/PublicInbox/LeiXSearch.pm +++ b/lib/PublicInbox/LeiXSearch.pm @@ -191,17 +191,22 @@ sub query_done { # EOF callback dwaitpid($_, $ipc_worker_reap, $l2m) for @$pids; } $lei->{ovv}->ovv_end($lei); - $lei->start_mua if $l2m; + if ($l2m) { # calls LeiToMail reap_compress + close(delete($lei->{1})) if $lei->{1}; + $lei->start_mua; + } $lei->dclose; } +sub do_post_augment { + my ($lei, $zpipe, $au_done) = @_; + my $l2m = $lei->{l2m} or die 'BUG: no {l2m}'; + $l2m->post_augment($lei, $zpipe); + close $au_done; # triggers wait_startq +} + sub start_query { # always runs in main (lei-daemon) process my ($self, $io, $lei, $srcs) = @_; - if (my $l2m = $lei->{l2m}) { - $lei->{1} = $io->[1]; - $l2m->post_augment($lei); - $io->[1] = delete $lei->{1}; - } my $remotes = $self->{remotes} // []; if ($lei->{opt}->{thread}) { for my $ibxish (@$srcs) { @@ -221,9 +226,11 @@ sub start_query { # always runs in main (lei-daemon) process sub query_prepare { # called by wq_do my ($self, $lei) = @_; my %sig = $lei->atfork_child_wq($self); + -p $lei->{0} or die "BUG: \$done pipe expected"; local @SIG{keys %sig} = values %sig; eval { $lei->{l2m}->do_augment($lei) }; $lei->fail($@) if $@; + syswrite($lei->{0}, '.') == 1 or die "do_post_augment trigger: $!"; } sub sigpipe_handler { @@ -253,26 +260,31 @@ sub do_query { $done = PublicInbox::OpPipe->new($done, $done_op, $in_loop); my $l2m = $lei->{l2m}; if ($l2m) { - $l2m->pre_augment($lei_orig); # may redirect $lei->{1} for mbox + # may redirect $lei->{1} for mbox + my $zpipe = $l2m->pre_augment($lei_orig); $io[1] = $lei_orig->{1}; - my @l2m_io = (undef, @io[1..$#io]); - pipe(my $startq, $l2m_io[0]) or die "pipe: $!"; - $self->wq_do('query_prepare', \@l2m_io, $lei); + pipe(my ($startq, $au_done)) or die "pipe: $!"; + $done_op->{'.'} = [ \&do_post_augment, $lei_orig, + $zpipe, $au_done ]; $io[4] = *STDERR{GLOB}; # don't send l2m->{-wq_s1} + $self->wq_do('query_prepare', \@io, $lei); die "BUG: unexpected \$io[5]: $io[5]" if $io[5]; fcntl($startq, 1031, 4096) if $^O eq 'linux'; # F_SETPIPE_SZ $io[5] = $startq; + $io[1] = $zpipe->[1] if $zpipe; } start_query($self, \@io, $lei, $srcs); unless ($in_loop) { my @pids = $self->wq_close; # for the $lei->atfork_child_wq PIPE handler: $done_op->{'!'}->[3] = \@pids; - $done->event_step; + # $done->event_step; + # my $ipc_worker_reap = $self->can('ipc_worker_reap'); + # if (my $l2m_pids = delete $self->{l2m_pids}) { + # dwaitpid($_, $ipc_worker_reap, $l2m) for @$l2m_pids; + # } + while ($done->{sock}) { $done->event_step } my $ipc_worker_reap = $self->can('ipc_worker_reap'); - if (my $l2m_pids = delete $self->{l2m_pids}) { - dwaitpid($_, $ipc_worker_reap, $l2m) for @$l2m_pids; - } dwaitpid($_, $ipc_worker_reap, $self) for @pids; } } -- cgit v1.2.3-24-ge0c7