lei: fix inadvertant FD sharing

$wq->{-ipc_atfork_child_close} neededed to be initialized properly.
And start setting $0 in workers to improve visibility.

diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
index 8fec2e62a4373f4b81507403ba64245e623808b5..24f45e036d99058b0dd08d77810fa08f00322c7f 100644 (file)
--- a/lib/PublicInbox/IPC.pm
+++ b/lib/PublicInbox/IPC.pm
@@ -134,6 +134,12 @@ sub ipc_worker_reap { # dwaitpid callback
        warn "PID:$pid died with \$?=$?\n" if $? && ($? & 127) != 15;
 }
 
+sub wq_wait_old {
+       my ($self) = @_;
+       my $pids = delete $self->{"-wq_old_pids.$$"} or return;
+       dwaitpid($_, \&ipc_worker_reap, $self) for @$pids;
+}
+
 # for base class, override in sub classes
 sub ipc_atfork_prepare {}
 
@@ -370,17 +376,25 @@ sub wq_workers {
 }
 
 sub wq_close {
-       my ($self) = @_;
+       my ($self, $nohang) = @_;
        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
-       return (keys %$workers) if wantarray; # caller will reap
-       for my $pid (keys %$workers) {
-               dwaitpid($pid, \&ipc_worker_reap, $self);
+       my @pids = map { $_ + 0 } keys %$workers;
+       if ($nohang) {
+               push @{$self->{"-wq_old_pids.$$"}}, @pids;
+       } else {
+               dwaitpid($_, \&ipc_worker_reap, $self) for @pids;
        }
 }
 
+sub wq_kill_old {
+       my ($self) = @_;
+       my $pids = $self->{"-wq_old_pids.$$"} or return;
+       kill 'TERM', @$pids;
+}
+
 sub wq_kill {
        my ($self, $sig) = @_;
        my $workers = $self->{-wq_workers} or return;

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 6be6d10b541dcf44d48656e082d625a9ebb8c7ad..2cb2bf406c433f66e72c6c9e3499b0cc11fb6244 100644 (file)
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -281,11 +281,14 @@ sub fail ($$;$) {
 
 sub atfork_prepare_wq {
        my ($self, $wq) = @_;
-       my $tcafc = $wq->{-ipc_atfork_child_close};
+       my $tcafc = $wq->{-ipc_atfork_child_close} //= [];
        push @$tcafc, @TO_CLOSE_ATFORK_CHILD;
        if (my $sock = $self->{sock}) {
                push @$tcafc, @$self{qw(0 1 2)}, $sock;
        }
+       if (my $pgr = $self->{pgr}) {
+               push @$tcafc, @$pgr[1,2];
+       }
        for my $f (qw(lxs l2m)) {
                my $ipc = $self->{$f} or next;
                push @$tcafc, grep { defined }
@@ -335,9 +338,7 @@ sub atfork_parent_wq {
        my $l2m = $ret->{l2m};
        if ($l2m && $l2m != $wq) { # $wq == lxs
                $io[4] = $l2m->{-wq_s1} if $l2m->{-wq_s1};
-               if (my @pids = $l2m->wq_close) {
-                       $wq->{l2m_pids} = \@pids;
-               }
+               $l2m->wq_close(1);
        }
        ($ret, @io);
 }

diff --git a/lib/PublicInbox/LeiQuery.pm b/lib/PublicInbox/LeiQuery.pm
index 941bc2997737a294f88a52d8a3942fdc9bb1da51..7d634b5e947571071709c40cd52824858bd34050 100644 (file)
--- a/lib/PublicInbox/LeiQuery.pm
+++ b/lib/PublicInbox/LeiQuery.pm
@@ -32,24 +32,25 @@ sub lei_q {
                my $sto = $self->_lei_store(1);
                push @srcs, $sto->search;
        }
-       my $lxs = PublicInbox::LeiXSearch->new;
 
+       my $lxs = $self->{lxs} = PublicInbox::LeiXSearch->new;
        # --external is enabled by default, but allow --no-external
-       if ($opt->{external} // 1) {
+       if ($opt->{external} //= 1) {
                $self->_externals_each(\&_vivify_external, \@srcs);
        }
-       my $j = $opt->{jobs} // (scalar(@srcs) > 3 ? 3 : scalar(@srcs));
-       $j = 1 if !$opt->{thread};
-       $self->atfork_prepare_wq($lxs);
-       $lxs->wq_workers_start('lei_xsearch', $j, $self->oldset);
-       $self->{lxs} = $lxs;
-
+       my $xj = $opt->{jobs} // (scalar(@srcs) > 3 ? 3 : scalar(@srcs));
+       $xj = 1 if !$opt->{thread};
        my $ovv = PublicInbox::LeiOverview->new($self) or return;
+       $self->atfork_prepare_wq($lxs);
+       $lxs->wq_workers_start('lei_xsearch', $xj, $self->oldset);
+       delete $lxs->{-ipc_atfork_child_close};
        if (my $l2m = $self->{l2m}) {
-               $j = 4 if $j <= 4; # TODO configurable
+               my $mj = 4; # TODO: configurable
                $self->atfork_prepare_wq($l2m);
-               $l2m->wq_workers_start('lei2mail', $j, $self->oldset);
+               $l2m->wq_workers_start('lei2mail', $mj, $self->oldset);
+               delete $l2m->{-ipc_atfork_child_close};
        }
+
        # no forking workers after this
 
        my %mset_opt = map { $_ => $opt->{$_} } qw(thread limit offset);

diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm
index 3dcce9e71687f4628152277aa98be83b63238e92..87cc9c4767cbd1e9e0d243b89ce5caddc7e3729b 100644 (file)
--- a/lib/PublicInbox/LeiToMail.pm
+++ b/lib/PublicInbox/LeiToMail.pm
@@ -467,7 +467,7 @@ sub write_mail { # via ->wq_do
 
 sub ipc_atfork_prepare {
        my ($self) = @_;
-       # (qry_status_wr, stdout|mbox, stderr, 3: sock, 4: each_smsg_done_wr)
+       # (done_wr, stdout|mbox, stderr, 3: sock, 4: each_smsg_done_wr)
        $self->wq_set_recv_modes(qw[+<&= >&= >&= +<&= >&=]);
        $self->SUPER::ipc_atfork_prepare; # PublicInbox::IPC
 }

diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index 1361188242877d26c49f532b31525e03a8214d4a..7b33677e7ea159c989d18dc44ec7fd13763a217a 100644 (file)
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -110,6 +110,7 @@ sub wait_startq ($) {
 
 sub query_thread_mset { # for --thread
        my ($self, $lei, $ibxish) = @_;
+       local $0 = "$0 query_thread_mset";
        my $startq = delete $self->{5};
        my %sig = $lei->atfork_child_wq($self);
        local @SIG{keys %sig} = values %sig;
@@ -148,6 +149,7 @@ sub query_thread_mset { # for --thread
 
 sub query_mset { # non-parallel for non-"--thread" users
        my ($self, $lei, $srcs) = @_;
+       local $0 = "$0 query_mset";
        my $startq = delete $self->{5};
        my %sig = $lei->atfork_child_wq($self);
        local @SIG{keys %sig} = values %sig;
@@ -192,12 +194,10 @@ sub git {
 sub query_done { # EOF callback
        my ($self, $lei) = @_;
        my $l2m = delete $lei->{l2m};
-       if (my $pids = delete $self->{l2m_pids}) {
-               my $ipc_worker_reap = $self->can('ipc_worker_reap');
-               dwaitpid($_, $ipc_worker_reap, $l2m) for @$pids;
-       }
+       $l2m->wq_wait_old if $l2m;
+       $self->wq_wait_old;
        $lei->{ovv}->ovv_end($lei);
-       if ($l2m) { # calls LeiToMail reap_compress
+       if ($l2m) { # close() calls LeiToMail reap_compress
                close(delete($lei->{1})) if $lei->{1};
                $lei->start_mua;
        }
@@ -232,12 +232,12 @@ sub start_query { # always runs in main (lei-daemon) process
        for my $rmt (@$remotes) {
                $self->wq_do('query_thread_mbox', $io, $lei, $rmt);
        }
-       close $io->[0]; # qry_status_wr
        @$io = ();
 }
 
 sub query_prepare { # called by wq_do
        my ($self, $lei) = @_;
+       local $0 = "$0 query_prepare";
        my %sig = $lei->atfork_child_wq($self);
        -p $lei->{0} or die "BUG: \$done pipe expected";
        local @SIG{keys %sig} = values %sig;
@@ -246,11 +246,11 @@ sub query_prepare { # called by wq_do
        syswrite($lei->{0}, '.') == 1 or die "do_post_augment trigger: $!";
 }
 
-sub sigpipe_handler {
-       my ($self, $lei_orig, $pids) = @_;
-       if ($pids) { # one-shot (no event loop)
-               kill 'TERM', @$pids;
+sub sigpipe_handler { # handles SIGPIPE from wq workers
+       my ($self, $lei_orig) = @_;
+       if ($self->wq_kill_old) {
                kill 'PIPE', $$;
+               $self->wq_wait_old;
        } else {
                $self->wq_kill;
                $self->wq_close;
@@ -287,19 +287,16 @@ sub do_query {
                $io[1] = $zpipe->[1] if $zpipe;
        }
        start_query($self, \@io, $lei, $srcs);
+       $self->wq_close(1);
        unless ($in_loop) {
-               my @pids = $self->wq_close;
                # for the $lei->atfork_child_wq PIPE handler:
-               $done_op->{'!'}->[3] = \@pids;
                while ($done->{sock}) { $done->event_step }
-               my $ipc_worker_reap = $self->can('ipc_worker_reap');
-               dwaitpid($_, $ipc_worker_reap, $self) for @pids;
        }
 }
 
 sub ipc_atfork_prepare {
        my ($self) = @_;
-       # (0: qry_status_wr, 1: stdout|mbox, 2: stderr,
+       # (0: done_wr, 1: stdout|mbox, 2: stderr,
        #  3: sock, 4: $l2m->{-wq_s1}, 5: $startq)
        $self->wq_set_recv_modes(qw[+<&= >&= >&= +<&= +<&= <&=]);
        $self->SUPER::ipc_atfork_prepare; # PublicInbox::IPC