about summary refs log tree commit homepage
diff options
context:
space:
mode:
authorEric Wong <e@80x24.org>2023-01-17 07:19:10 +0000
committerEric Wong <e@80x24.org>2023-01-18 23:26:03 +0000
commit4a2a95bbc78f99c8c5278cfe29de74bd1483903c (patch)
tree8652affab199e745cb52da49f526a2d998e755fb
parentf7f0e7bee09cfc57ecc6845a8cf9e80a0bbd64f5 (diff)
downloadpublic-inbox-4a2a95bbc78f99c8c5278cfe29de74bd1483903c.tar.gz
This avoids awkwardly stuffing an arrayref into callbacks
which expect multiple arguments.  IPC->awaitpid_init now
allows pre-registering callbacks before spawning workers.
-rw-r--r--lib/PublicInbox/IPC.pm30
-rw-r--r--lib/PublicInbox/LEI.pm8
-rw-r--r--lib/PublicInbox/LeiConvert.pm2
-rw-r--r--lib/PublicInbox/LeiInput.pm2
-rw-r--r--lib/PublicInbox/LeiMirror.pm7
-rw-r--r--lib/PublicInbox/LeiStore.pm7
-rw-r--r--lib/PublicInbox/LeiToMail.pm7
-rw-r--r--lib/PublicInbox/LeiUp.pm5
-rw-r--r--lib/PublicInbox/LeiXSearch.pm9
-rwxr-xr-xscript/public-inbox-clone2
10 files changed, 35 insertions, 44 deletions
diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
index 34e40118..edc5ba64 100644
--- a/lib/PublicInbox/IPC.pm
+++ b/lib/PublicInbox/IPC.pm
@@ -12,7 +12,7 @@ use strict;
 use v5.10.1;
 use parent qw(Exporter);
 use Carp qw(croak);
-use PublicInbox::DS qw(dwaitpid);
+use PublicInbox::DS qw(awaitpid);
 use PublicInbox::Spawn;
 use PublicInbox::OnDestroy;
 use PublicInbox::WQWorker;
@@ -133,26 +133,26 @@ sub ipc_worker_spawn {
         $self->{-ipc_req} = $w_req;
         $self->{-ipc_res} = $r_res;
         $self->{-ipc_ppid} = $$;
+        awaitpid($pid, \&ipc_worker_reap, $self);
         $self->{-ipc_pid} = $pid;
 }
 
-sub ipc_worker_reap { # dwaitpid callback
-        my ($args, $pid) = @_;
-        my ($self, @uargs) = @$args;
+sub ipc_worker_reap { # awaitpid callback
+        my ($pid, $self) = @_;
         delete $self->{-wq_workers}->{$pid};
-        return $self->{-reap_do}->($args, $pid) if $self->{-reap_do};
+        if (my $cb_args = $self->{-reap_do}) {
+                return $cb_args->[0]->($pid, $self, @$cb_args[1..$#$cb_args]);
+        }
         return if !$?;
         my $s = $? & 127;
         # 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_async {
-        my ($self, $cb, @uargs) = @_;
-        local $PublicInbox::DS::in_loop = 1;
-        $self->{-reap_do} = $cb;
-        my @pids = keys %{$self->{-wq_workers}};
-        dwaitpid($_, \&ipc_worker_reap, [ $self, @uargs ]) for @pids;
+# register wait workers
+sub awaitpid_init {
+        my ($self, @cb_args) = @_;
+        $self->{-reap_do} = \@cb_args;
 }
 
 # for base class, override in sub classes
@@ -178,9 +178,7 @@ sub ipc_worker_stop {
         }
         die 'no PID with IPC pipes' unless $pid;
         $w_req = $r_res = undef;
-
-        return if $$ != $ppid;
-        dwaitpid($pid, \&ipc_worker_reap, [$self]);
+        awaitpid($pid) if $$ == $ppid; # for non-event loop
 }
 
 # use this if we have multiple readers reading curl or "pigz -dc"
@@ -397,6 +395,7 @@ sub _wq_worker_start ($$$$) {
                 undef $end; # trigger exit
         } else {
                 $self->{-wq_workers}->{$pid} = $bcast1;
+                awaitpid($pid, \&ipc_worker_reap, $self);
         }
 }
 
@@ -428,8 +427,7 @@ sub wq_close {
         }
         delete @$self{qw(-wq_s1 -wq_s2)} or return;
         return if $self->{-reap_do};
-        my @pids = keys %{$self->{-wq_workers}};
-        dwaitpid($_, \&ipc_worker_reap, [ $self ]) for @pids;
+        awaitpid($_) for keys %{$self->{-wq_workers}};
 }
 
 sub wq_kill {
diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index b78d70de..6ad42111 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -18,7 +18,6 @@ use IO::Handle ();
 use Fcntl qw(SEEK_SET);
 use PublicInbox::Config;
 use PublicInbox::Syscall qw(EPOLLIN);
-use PublicInbox::DS qw(dwaitpid);
 use PublicInbox::Spawn qw(spawn popen_rd);
 use PublicInbox::Lock;
 use PublicInbox::Eml;
@@ -644,12 +643,12 @@ sub workers_start {
         my $end = $lei->pkt_op_pair;
         my $ident = $wq->{-wq_ident} // "lei-$lei->{cmd} worker";
         $flds->{lei} = $lei;
+        $wq->awaitpid_init($wq->can('_wq_done_wait') // \&wq_done_wait, $lei);
         $wq->wq_workers_start($ident, $jobs, $lei->oldset, $flds);
         delete $lei->{pkt_op_p};
         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);
 }
 
@@ -1391,9 +1390,8 @@ sub DESTROY {
         # preserve $? for ->fail or ->x_it code
 }
 
-sub wq_done_wait { # dwaitpid callback
-        my ($arg, $pid) = @_;
-        my ($wq, $lei) = @$arg;
+sub wq_done_wait { # awaitpid cb (via wq_eof / IPC->awaitpid_init)
+        my ($pid, $wq, $lei) = @_;
         local $current_lei = $lei;
         my $err_type = $lei->{-err_type};
         $? and $lei->child_error($?,
diff --git a/lib/PublicInbox/LeiConvert.pm b/lib/PublicInbox/LeiConvert.pm
index 59af40de..1acd4558 100644
--- a/lib/PublicInbox/LeiConvert.pm
+++ b/lib/PublicInbox/LeiConvert.pm
@@ -30,7 +30,7 @@ sub input_maildir_cb {
 
 sub process_inputs { # via wq_do
         my ($self) = @_;
-        local $PublicInbox::DS::in_loop = 0; # force synchronous dwaitpid
+        local $PublicInbox::DS::in_loop = 0; # force synchronous awaitpid
         $self->SUPER::process_inputs;
         my $lei = $self->{lei};
         delete $lei->{1};
diff --git a/lib/PublicInbox/LeiInput.pm b/lib/PublicInbox/LeiInput.pm
index a1dcc907..c258f824 100644
--- a/lib/PublicInbox/LeiInput.pm
+++ b/lib/PublicInbox/LeiInput.pm
@@ -177,7 +177,7 @@ sub input_path_url {
                         $mbl->{fh} =
                              PublicInbox::MboxReader::zsfxcat($in, $zsfx, $lei);
                 }
-                local $PublicInbox::DS::in_loop = 0 if $zsfx; # dwaitpid
+                local $PublicInbox::DS::in_loop = 0 if $zsfx; # awaitpid
                 $self->input_fh($ifmt, $mbl->{fh}, $input, @args);
         } elsif (-d _ && (-d "$input/cur" || -d "$input/new")) {
                 return $lei->fail(<<EOM) if $ifmt && $ifmt ne 'maildir';
diff --git a/lib/PublicInbox/LeiMirror.pm b/lib/PublicInbox/LeiMirror.pm
index 87abf88c..abf66315 100644
--- a/lib/PublicInbox/LeiMirror.pm
+++ b/lib/PublicInbox/LeiMirror.pm
@@ -31,9 +31,8 @@ sub keep_going ($) {
                 $_[0]->{lei}->{opt}->{'keep-going'});
 }
 
-sub _wq_done_wait { # dwaitpid callback (via wq_eof)
-        my ($arg, $pid) = @_;
-        my ($mrr, $lei) = @$arg;
+sub _wq_done_wait { # awaitpid cb (via wq_eof / IPC->awaitpid_init)
+        my ($pid, $mrr, $lei) = @_;
         if ($?) {
                 $lei->child_error($?);
         } elsif (!$lei->{child_error}) {
@@ -236,7 +235,7 @@ sub index_cloned_inbox {
                         my ($k) = ($sw =~ /\A([\w-]+)/);
                         $opt->{$k} = $lei->{opt}->{$k};
                 }
-                # force synchronous dwaitpid for v2:
+                # force synchronous awaitpid for v2:
                 local $PublicInbox::DS::in_loop = 0;
                 my $cfg = PublicInbox::Config->new(undef, $lei->{2});
                 my $env = PublicInbox::Admin::index_prepare($opt, $cfg);
diff --git a/lib/PublicInbox/LeiStore.pm b/lib/PublicInbox/LeiStore.pm
index 57f0e013..0ecf1388 100644
--- a/lib/PublicInbox/LeiStore.pm
+++ b/lib/PublicInbox/LeiStore.pm
@@ -604,9 +604,8 @@ sub recv_and_run {
         $self->SUPER::recv_and_run(@args);
 }
 
-sub _sto_atexit { # dwaitpid callback
-        my ($args, $pid) = @_;
-        my $self = $args->[0];
+sub _sto_atexit { # awaitpid cb (via awaitpid_init)
+        my ($pid, $sto) = @_;
         warn "lei/store PID:$pid died \$?=$?\n" if $?;
 }
 
@@ -621,12 +620,12 @@ sub write_prepare {
                 # Mail we import into lei are private, so headers filtered out
                 # by -mda for public mail are not appropriate
                 local @PublicInbox::MDA::BAD_HEADERS = ();
+                $self->awaitpid_init(\&_sto_atexit); # outlives $lei
                 $self->wq_workers_start("lei/store $dir", 1, $lei->oldset, {
                                         lei => $lei,
                                         -err_wr => $w,
                                         to_close => [ $r ],
                                 });
-                $self->wq_wait_async(\&_sto_atexit); # outlives $lei
                 require PublicInbox::LeiStoreErr;
                 PublicInbox::LeiStoreErr->new($r, $lei);
         }
diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm
index 1528165a..6a4554e7 100644
--- a/lib/PublicInbox/LeiToMail.pm
+++ b/lib/PublicInbox/LeiToMail.pm
@@ -652,9 +652,8 @@ sub _do_augment_mbox {
         $dedupe->pause_dedupe if $dedupe;
 }
 
-sub v2w_done_wait { # dwaitpid callback
-        my ($arg, $pid) = @_;
-        my ($v2w, $lei) = @$arg;
+sub v2w_done_wait { # awaitpid cb (via awaitpid_init)
+        my ($pid, $v2w, $lei) = @_;
         $lei->child_error($?, "error for $v2w->{ibx}->{inboxdir}") if $?;
 }
 
@@ -680,8 +679,8 @@ sub _pre_augment_v2 {
         PublicInbox::InboxWritable->new($ibx, @creat);
         $ibx->init_inbox if @creat;
         my $v2w = $ibx->importer;
+        $v2w->awaitpid_init(\&v2w_done_wait, $lei);
         $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";
diff --git a/lib/PublicInbox/LeiUp.pm b/lib/PublicInbox/LeiUp.pm
index 49917339..3e92242e 100644
--- a/lib/PublicInbox/LeiUp.pm
+++ b/lib/PublicInbox/LeiUp.pm
@@ -165,9 +165,8 @@ sub _complete_up { # lei__complete hook
         map { $match_cb->($_) } PublicInbox::LeiSavedSearch::list($lei);
 }
 
-sub _wq_done_wait { # dwaitpid callback
-        my ($arg, $pid) = @_;
-        my ($wq, $lei) = @$arg;
+sub _wq_done_wait { # awaitpid cb (via awaitpid_init)
+        my ($pid, $wq, $lei) = @_;
         $lei->child_error($?, 'auth failure') if $?
 }
 
diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index 730df1f7..f9aa870e 100644
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -400,9 +400,8 @@ sub query_remote_mboxrd {
 
 sub git { $_[0]->{git} // die 'BUG: git uninitialized' }
 
-sub xsearch_done_wait { # dwaitpid callback
-        my ($arg, $pid) = @_;
-        my ($wq, $lei) = @$arg;
+sub xsearch_done_wait { # awaitpid cb (via awaitpid_init)
+        my ($pid, $wq, $lei) = @_;
         return if !$?;
         my $s = $? & 127;
         return $lei->child_error($?) if $s == 13 || $s == 15;
@@ -573,16 +572,16 @@ sub do_query {
                         fcntl($b_r, $F_SETPIPE_SZ, 4096) if $F_SETPIPE_SZ;
                         $l2m->{au_peers} = [ $a_r, $a_w, $b_r, $b_w ];
                 }
+                $l2m->awaitpid_init(\&xsearch_done_wait, $lei);
                 $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->awaitpid_init(\&xsearch_done_wait, $lei);
         $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 = ();
diff --git a/script/public-inbox-clone b/script/public-inbox-clone
index e93ac37b..598979bc 100755
--- a/script/public-inbox-clone
+++ b/script/public-inbox-clone
@@ -62,5 +62,5 @@ my $mrr = bless {
 
 $? = 0;
 $mrr->do_mirror;
-$mrr->can('_wq_done_wait')->([$mrr, $lei], $$);
+$mrr->can('_wq_done_wait')->($$, $mrr, $lei);
 exit(($lei->{child_error} // 0) >> 8);