about summary refs log tree commit homepage
diff options
context:
space:
mode:
authorEric Wong <e@80x24.org>2021-10-15 13:30:55 +0000
committerEric Wong <e@80x24.org>2021-10-15 15:58:24 +0000
commita1733d3406dfbde52d1468e671edd1d76893f546 (patch)
tree118763410e5c6f32776671a435595792f0555ee4
parent2ca12a7fd78d7c9c27fec4f50fdcb58f9c838003 (diff)
downloadpublic-inbox-a1733d3406dfbde52d1468e671edd1d76893f546.tar.gz
Simplify our APIs and force dwaitpid() to work in async mode for
all lei workers.  This avoids having lingering zombies for
parallel searches if one worker finishes soon before another.

The old distinction between "old" and "new" workers was
needlessly complex, error-prone, and embarrasingly bad.

We also never handled v2:// writers properly before on
Ctrl-C/Ctrl-Z (SIGINT/SIGTSTP), so add them to @WQ_KEYS
to ensure they get handled by $lei when appropropriate.
-rw-r--r--lib/PublicInbox/IPC.pm46
-rw-r--r--lib/PublicInbox/LEI.pm16
-rw-r--r--lib/PublicInbox/LeiBlob.pm2
-rw-r--r--lib/PublicInbox/LeiConvert.pm2
-rw-r--r--lib/PublicInbox/LeiImportKw.pm1
-rw-r--r--lib/PublicInbox/LeiInput.pm2
-rw-r--r--lib/PublicInbox/LeiInspect.pm2
-rw-r--r--lib/PublicInbox/LeiLsSearch.pm2
-rw-r--r--lib/PublicInbox/LeiMirror.pm2
-rw-r--r--lib/PublicInbox/LeiNoteEvent.pm5
-rw-r--r--lib/PublicInbox/LeiP2q.pm2
-rw-r--r--lib/PublicInbox/LeiPmdir.pm1
-rw-r--r--lib/PublicInbox/LeiStore.pm1
-rw-r--r--lib/PublicInbox/LeiToMail.pm10
-rw-r--r--lib/PublicInbox/LeiUp.pm2
-rw-r--r--lib/PublicInbox/LeiXSearch.pm16
16 files changed, 51 insertions, 61 deletions
diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
index 6c189b64..3e299448 100644
--- a/lib/PublicInbox/IPC.pm
+++ b/lib/PublicInbox/IPC.pm
@@ -134,16 +134,22 @@ sub ipc_worker_spawn {
 
 sub ipc_worker_reap { # dwaitpid callback
         my ($args, $pid) = @_;
+        my ($self, @uargs) = @$args;
+        delete $self->{-wq_workers}->{$pid};
+        return $self->{-reap_do}->($args, $pid) if $self->{-reap_do};
         return if !$?;
-        # TERM(15) is our default exit signal, PIPE(13) is likely w/ pager
         my $s = $? & 127;
-        warn "PID:$pid died with \$?=$?\n" if $s != 15 && $s != 13;
+        # TERM(15) is our default exit signal, PIPE(13) is likely w/ pager
+        warn "$self->{-wq_ident} PID:$pid died \$?=$?\n" if $s != 15 && $s != 13
 }
 
-sub wq_wait_old {
-        my ($self, $cb, @args) = @_;
-        my $pids = delete $self->{"-wq_old_pids.$$"} or return;
-        dwaitpid($_, $cb // \&ipc_worker_reap, [$self, @args]) for @$pids;
+sub wq_wait_async {
+        my ($self, $cb, @uargs) = @_;
+        local $PublicInbox::DS::in_loop = 1;
+        $self->{-reap_async} = 1;
+        $self->{-reap_do} = $cb;
+        my @pids = keys %{$self->{-wq_workers}};
+        dwaitpid($_, \&ipc_worker_reap, [ $self, @uargs ]) for @pids;
 }
 
 # for base class, override in sub classes
@@ -394,42 +400,24 @@ sub wq_workers_start {
 }
 
 sub wq_close {
-        my ($self, $nohang, $cb, @args) = @_;
+        my ($self) = @_;
         delete @$self{qw(-wq_s1 -wq_s2)} or return;
-        my $ppid = delete $self->{-wq_ppid} or return;
-        my $workers = delete $self->{-wq_workers} // die 'BUG: no wq_workers';
-        return if $ppid != $$; # can't reap siblings or parents
-        my @pids = map { $_ + 0 } keys %$workers;
-        if ($nohang) {
-                push @{$self->{"-wq_old_pids.$$"}}, @pids;
-        } else {
-                $cb //= \&ipc_worker_reap;
-                unshift @args, $self;
-                dwaitpid($_, $cb, \@args) for @pids;
-        }
-}
-
-sub wq_kill_old {
-        my ($self, $sig) = @_;
-        my $pids = $self->{"-wq_old_pids.$$"} or return;
-        kill($sig // 'TERM', @$pids);
+        return if $self->{-reap_async};
+        my @pids = keys %{$self->{-wq_workers}};
+        dwaitpid($_, \&ipc_worker_reap, [ $self ]) for @pids;
 }
 
 sub wq_kill {
         my ($self, $sig) = @_;
-        my $workers = $self->{-wq_workers} or return;
-        kill($sig // 'TERM', keys %$workers);
+        kill($sig // 'TERM', keys %{$self->{-wq_workers}});
 }
 
 sub DESTROY {
         my ($self) = @_;
         my $ppid = $self->{-wq_ppid};
         wq_kill($self) if $ppid && $ppid == $$;
-        my $err = $?;
         wq_close($self);
-        wq_wait_old($self);
         ipc_worker_stop($self);
-        $? = $err if $err;
 }
 
 sub detect_nproc () {
diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index b6338377..83534878 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -455,16 +455,12 @@ my %CONFIG_KEYS = (
         'leistore.dir' => 'top-level storage location',
 );
 
-my @WQ_KEYS = qw(lxs l2m ikw pmd wq1 lne); # internal workers
+my @WQ_KEYS = qw(lxs l2m ikw pmd wq1 lne v2w); # internal workers
 
 sub _drop_wq {
         my ($self) = @_;
         for my $wq (grep(defined, delete(@$self{@WQ_KEYS}))) {
-                if ($wq->wq_kill('-TERM')) {
-                        $wq->wq_close(0, undef, $self);
-                } elsif ($wq->wq_kill_old('-TERM')) {
-                        $wq->wq_wait_old(undef, $self);
-                }
+                $wq->wq_kill('-TERM');
                 $wq->DESTROY;
         }
 }
@@ -644,6 +640,7 @@ sub workers_start {
         my $op_c = delete $lei->{pkt_op_c};
         @$end = ();
         $lei->event_step_init;
+        $wq->wq_wait_async($wq->can('_wq_done_wait') // \&wq_done_wait, $lei);
         ($op_c, $ops);
 }
 
@@ -651,7 +648,7 @@ sub workers_start {
 sub wait_wq_events {
         my ($lei, $op_c, $ops) = @_;
         for my $wq (grep(defined, @$lei{qw(ikw pmd)})) { # auxiliary WQs
-                $wq->wq_close(1);
+                $wq->wq_close;
         }
         $op_c->{ops} = $ops;
 }
@@ -1150,7 +1147,7 @@ sub event_step {
                 if ($buf =~ /\A(?:STOP|CONT|TERM)\z/) {
                         my $sig = "-$buf";
                         for my $wq (grep(defined, @$self{@WQ_KEYS})) {
-                                $wq->wq_kill($sig) or $wq->wq_kill_old($sig);
+                                $wq->wq_kill($sig);
                         }
                 } else {
                         die "unrecognized client signal: $buf";
@@ -1393,8 +1390,7 @@ sub fchdir {
 sub wq_eof { # EOF callback for main daemon
         my ($lei) = @_;
         local $current_lei = $lei;
-        my $wq1 = delete $lei->{wq1} // return $lei->fail; # already failed
-        $wq1->wq_wait_old($wq1->can('_wq_done_wait') // \&wq_done_wait, $lei);
+        delete $lei->{wq1} // return $lei->fail; # already failed
 }
 
 sub watch_state_ok ($) {
diff --git a/lib/PublicInbox/LeiBlob.pm b/lib/PublicInbox/LeiBlob.pm
index b6a62d24..004b156c 100644
--- a/lib/PublicInbox/LeiBlob.pm
+++ b/lib/PublicInbox/LeiBlob.pm
@@ -166,7 +166,7 @@ sub lei_blob {
         my ($op_c, $ops) = $lei->workers_start($self, 1);
         $lei->{wq1} = $self;
         $self->wq_io_do('do_solve_blob', []);
-        $self->wq_close(1);
+        $self->wq_close;
         $lei->wait_wq_events($op_c, $ops);
 }
 
diff --git a/lib/PublicInbox/LeiConvert.pm b/lib/PublicInbox/LeiConvert.pm
index 9e98edc3..68fc7c0b 100644
--- a/lib/PublicInbox/LeiConvert.pm
+++ b/lib/PublicInbox/LeiConvert.pm
@@ -58,7 +58,7 @@ sub lei_convert { # the main "lei convert" method
         my ($op_c, $ops) = $lei->workers_start($self, 1);
         $lei->{wq1} = $self;
         $self->wq_io_do('process_inputs', []);
-        $self->wq_close(1);
+        $self->wq_close;
         $lei->wait_wq_events($op_c, $ops);
 }
 
diff --git a/lib/PublicInbox/LeiImportKw.pm b/lib/PublicInbox/LeiImportKw.pm
index 8359f338..54454511 100644
--- a/lib/PublicInbox/LeiImportKw.pm
+++ b/lib/PublicInbox/LeiImportKw.pm
@@ -50,7 +50,6 @@ sub _lei_wq_eof { # EOF callback for main lei daemon
         my ($lei) = @_;
         my $ikw = delete $lei->{ikw} or return $lei->fail;
         $lei->sto_done_request($ikw->{lei_sock});
-        $ikw->wq_wait_old($lei->can('wq_done_wait'), $lei);
 }
 
 1;
diff --git a/lib/PublicInbox/LeiInput.pm b/lib/PublicInbox/LeiInput.pm
index dd40d838..2621fc1f 100644
--- a/lib/PublicInbox/LeiInput.pm
+++ b/lib/PublicInbox/LeiInput.pm
@@ -402,7 +402,7 @@ sub input_only_atfork_child {
 sub input_only_net_merge_all_done {
         my ($self) = @_;
         $self->wq_io_do('process_inputs');
-        $self->wq_close(1);
+        $self->wq_close;
 }
 
 # like Getopt::Long, but for +kw:FOO and -kw:FOO to prepare
diff --git a/lib/PublicInbox/LeiInspect.pm b/lib/PublicInbox/LeiInspect.pm
index 5ba96056..05b6e21d 100644
--- a/lib/PublicInbox/LeiInspect.pm
+++ b/lib/PublicInbox/LeiInspect.pm
@@ -242,7 +242,7 @@ sub inspect_start ($$) {
         $lei->{wq1} = $self;
         $lei->wait_wq_events($op_c, $ops);
         $self->wq_do('inspect_argv');
-        $self->wq_close(1);
+        $self->wq_close;
 }
 
 sub ins_add { # InputPipe->consume callback
diff --git a/lib/PublicInbox/LeiLsSearch.pm b/lib/PublicInbox/LeiLsSearch.pm
index aebf0184..0193e590 100644
--- a/lib/PublicInbox/LeiLsSearch.pm
+++ b/lib/PublicInbox/LeiLsSearch.pm
@@ -75,7 +75,7 @@ sub bg_worker ($$$) {
         my ($op_c, $ops) = $lei->workers_start($self, 1);
         $lei->{wq1} = $self;
         $self->wq_io_do('do_ls_search_long', [], $pfx);
-        $self->wq_close(1);
+        $self->wq_close;
         $lei->wait_wq_events($op_c, $ops);
 }
 
diff --git a/lib/PublicInbox/LeiMirror.pm b/lib/PublicInbox/LeiMirror.pm
index a75c99c4..e20d30b4 100644
--- a/lib/PublicInbox/LeiMirror.pm
+++ b/lib/PublicInbox/LeiMirror.pm
@@ -451,7 +451,7 @@ sub start {
         my ($op_c, $ops) = $lei->workers_start($self, 1);
         $lei->{wq1} = $self;
         $self->wq_io_do('do_mirror', []);
-        $self->wq_close(1);
+        $self->wq_close;
         $lei->wait_wq_events($op_c, $ops);
 }
 
diff --git a/lib/PublicInbox/LeiNoteEvent.pm b/lib/PublicInbox/LeiNoteEvent.pm
index 1b714dae..ba4dfd49 100644
--- a/lib/PublicInbox/LeiNoteEvent.pm
+++ b/lib/PublicInbox/LeiNoteEvent.pm
@@ -14,7 +14,7 @@ our $to_flush; # { cfgpath => $lei }
 sub flush_lei ($) {
         my ($lei) = @_;
         my $lne = delete $lei->{cfg}->{-lei_note_event};
-        $lne->wq_close(1, undef, $lei) if $lne; # runs _lei_wq_eof;
+        $lne->wq_close if $lne; # runs _lei_wq_eof;
 }
 
 # we batch up writes and flush every 5s (matching Linux default
@@ -111,9 +111,8 @@ sub ipc_atfork_child {
 
 sub _lei_wq_eof { # EOF callback for main lei daemon
         my ($lei) = @_;
-        my $lne = delete $lei->{lne} or return $lei->fail;
+        delete $lei->{lne} or return $lei->fail;
         $lei->sto_done_request;
-        $lne->wq_wait_old($lei->can('wq_done_wait'), $lei);
 }
 
 1;
diff --git a/lib/PublicInbox/LeiP2q.pm b/lib/PublicInbox/LeiP2q.pm
index 5c2ce0a1..08ec81c5 100644
--- a/lib/PublicInbox/LeiP2q.pm
+++ b/lib/PublicInbox/LeiP2q.pm
@@ -191,7 +191,7 @@ sub lei_p2q { # the "lei patch-to-query" entry point
         my ($op_c, $ops) = $lei->workers_start($self, 1);
         $lei->{wq1} = $self;
         $self->wq_io_do('do_p2q', []);
-        $self->wq_close(1);
+        $self->wq_close;
         $lei->wait_wq_events($op_c, $ops);
 }
 
diff --git a/lib/PublicInbox/LeiPmdir.pm b/lib/PublicInbox/LeiPmdir.pm
index 2d3b9755..f9b68fc2 100644
--- a/lib/PublicInbox/LeiPmdir.pm
+++ b/lib/PublicInbox/LeiPmdir.pm
@@ -51,7 +51,6 @@ sub _lei_wq_eof { # EOF callback for main lei daemon
         my ($lei) = @_;
         my $pmd = delete $lei->{pmd} or return $lei->fail;
         $lei->sto_done_request($pmd->{lei_sock});
-        $pmd->wq_wait_old($lei->can('wq_done_wait'), $lei);
 }
 
 1;
diff --git a/lib/PublicInbox/LeiStore.pm b/lib/PublicInbox/LeiStore.pm
index c45380d1..82104570 100644
--- a/lib/PublicInbox/LeiStore.pm
+++ b/lib/PublicInbox/LeiStore.pm
@@ -587,6 +587,7 @@ sub write_prepare {
                                         -err_wr => $w,
                                         to_close => [ $r ],
                                 });
+                $self->wq_wait_async; # outlives $lei
                 require PublicInbox::LeiStoreErr;
                 PublicInbox::LeiStoreErr->new($r, $lei);
         }
diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm
index 9c748dea..76e103c7 100644
--- a/lib/PublicInbox/LeiToMail.pm
+++ b/lib/PublicInbox/LeiToMail.pm
@@ -637,6 +637,12 @@ sub _do_augment_mbox {
         $dedupe->pause_dedupe if $dedupe;
 }
 
+sub v2w_done_wait { # dwaitpid callback
+        my ($arg, $pid) = @_;
+        my ($v2w, $lei) = @$arg;
+        $lei->child_error($?, "error for $v2w->{ibx}->{inboxdir}") if $?;
+}
+
 sub _pre_augment_v2 {
         my ($self, $lei) = @_;
         my $dir = $self->{dst};
@@ -659,8 +665,8 @@ sub _pre_augment_v2 {
         PublicInbox::InboxWritable->new($ibx, @creat);
         $ibx->init_inbox if @creat;
         my $v2w = $ibx->importer;
-        $v2w->{-wq_no_bcast} = 1;
         $v2w->wq_workers_start("lei/v2w $dir", 1, $lei->oldset, {lei => $lei});
+        $v2w->wq_wait_async(\&v2w_done_wait, $lei);
         $lei->{v2w} = $v2w;
         return if !$lei->{opt}->{shared};
         my $d = "$lei->{ale}->{git}->{git_dir}/objects";
@@ -811,7 +817,7 @@ sub net_merge_all_done {
                                 $self->{dst}, \$self->{-au_noted});
         }
         $self->wq_broadcast('do_post_auth');
-        $self->wq_close(1);
+        $self->wq_close;
 }
 
 1;
diff --git a/lib/PublicInbox/LeiUp.pm b/lib/PublicInbox/LeiUp.pm
index df65cb9b..39604177 100644
--- a/lib/PublicInbox/LeiUp.pm
+++ b/lib/PublicInbox/LeiUp.pm
@@ -122,7 +122,7 @@ EOM
 sub net_merge_all_done {
         my ($self, $lei) = @_;
         $lei->{net} = delete($self->{-net_new}) if $self->{-net_new};
-        $self->wq_close(1);
+        $self->wq_close;
         eval { redispatch_all($self, $lei) };
         warn "E: $@" if $@;
 }
diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index 3ec75528..fd2c8a37 100644
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -413,14 +413,14 @@ sub query_done { # EOF callback for main daemon
         my ($lei) = @_;
         local $PublicInbox::LEI::current_lei = $lei;
         my $l2m = delete $lei->{l2m};
-        $l2m->wq_wait_old(\&xsearch_done_wait, $lei) if $l2m;
-        if (my $lxs = delete $lei->{lxs}) {
-                $lxs->wq_wait_old(\&xsearch_done_wait, $lei);
-        }
+        delete $lei->{lxs};
         ($lei->{opt}->{'mail-sync'} && !$lei->{sto}) and
                 warn "BUG: {sto} missing with --mail-sync";
         $lei->sto_done_request if $lei->{sto};
-        my $wait = $lei->{v2w} ? $lei->{v2w}->wq_do('done') : undef;
+        if (my $v2w = delete $lei->{v2w}) {
+                $v2w->wq_do('done');
+                $v2w->wq_close;
+        }
         $lei->{ovv}->ovv_end($lei);
         my $start_mua;
         if ($l2m) { # close() calls LeiToMail reap_compress
@@ -466,7 +466,7 @@ sub do_post_augment {
         if ($err) {
                 if (my $lxs = delete $lei->{lxs}) {
                         $lxs->wq_kill('-TERM');
-                        $lxs->wq_close(0, undef, $lei);
+                        $lxs->wq_close;
                 }
                 $lei->fail("$err");
         }
@@ -514,7 +514,7 @@ sub start_query ($$) { # always runs in main (lei-daemon) process
         if ($self->{-do_lcat}) {
                 $self->wq_io_do('lcat_dump', []);
         }
-        $self->wq_close(1); # lei_xsearch workers stop when done
+        $self->wq_close; # lei_xsearch workers stop when done
 }
 
 sub incr_start_query { # called whenever an l2m shard starts do_post_auth
@@ -569,12 +569,14 @@ sub do_query {
                 }
                 $l2m->wq_workers_start('lei2mail', undef,
                                         $lei->oldset, { lei => $lei });
+                $l2m->wq_wait_async(\&xsearch_done_wait, $lei);
                 pipe($lei->{startq}, $lei->{au_done}) or die "pipe: $!";
                 fcntl($lei->{startq}, $F_SETPIPE_SZ, 4096) if $F_SETPIPE_SZ;
                 delete $l2m->{au_peers};
         }
         $self->wq_workers_start('lei_xsearch', undef,
                                 $lei->oldset, { lei => $lei });
+        $self->wq_wait_async(\&xsearch_done_wait, $lei);
         my $op_c = delete $lei->{pkt_op_c};
         delete $lei->{pkt_op_p};
         @$end = ();