From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: X-Spam-Checker-Version: SpamAssassin 3.4.2 (2018-09-13) on dcvr.yhbt.net X-Spam-Level: X-Spam-Status: No, score=-4.0 required=3.0 tests=ALL_TRUSTED,BAYES_00 shortcircuit=no autolearn=ham autolearn_force=no version=3.4.2 Received: from localhost (dcvr.yhbt.net [127.0.0.1]) by dcvr.yhbt.net (Postfix) with ESMTP id EBC8A1FA19 for ; Thu, 21 Jan 2021 19:46:24 +0000 (UTC) From: Eric Wong To: meta@public-inbox.org Subject: [PATCH 05/12] lei: fix inadvertant FD sharing Date: Thu, 21 Jan 2021 19:46:17 +0000 Message-Id: <20210121194624.32002-6-e@80x24.org> In-Reply-To: <20210121194624.32002-1-e@80x24.org> References: <20210121194624.32002-1-e@80x24.org> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit List-Id: $wq->{-ipc_atfork_child_close} neededed to be initialized properly. And start setting $0 in workers to improve visibility. --- lib/PublicInbox/IPC.pm | 22 ++++++++++++++++++---- lib/PublicInbox/LEI.pm | 9 +++++---- lib/PublicInbox/LeiQuery.pm | 21 +++++++++++---------- lib/PublicInbox/LeiToMail.pm | 2 +- lib/PublicInbox/LeiXSearch.pm | 27 ++++++++++++--------------- 5 files changed, 47 insertions(+), 34 deletions(-) diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm index 8fec2e62..24f45e03 100644 --- 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 6be6d10b..2cb2bf40 100644 --- 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 941bc299..7d634b5e 100644 --- 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 3dcce9e7..87cc9c47 100644 --- 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 13611882..7b33677e 100644 --- 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