From: Eric Wong <e@80x24.org>
To: meta@public-inbox.org
Subject: [PATCH 05/12] lei: fix inadvertant FD sharing
Date: Thu, 21 Jan 2021 19:46:17 +0000 [thread overview]
Message-ID: <20210121194624.32002-6-e@80x24.org> (raw)
In-Reply-To: <20210121194624.32002-1-e@80x24.org>
$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
next prev parent reply other threads:[~2021-01-21 19:46 UTC|newest]
Thread overview: 13+ messages / expand[flat|nested] mbox.gz Atom feed top
2021-01-21 19:46 [PATCH 00/12] lei: another dump Eric Wong
2021-01-21 19:46 ` [PATCH 01/12] lei_overview: rename {relevance} => {pct} Eric Wong
2021-01-21 19:46 ` [PATCH 02/12] lei q: retrieve keywords for local, non-external messages Eric Wong
2021-01-21 19:46 ` [PATCH 03/12] lei_xsearch: eliminate some unused, commented-out code Eric Wong
2021-01-21 19:46 ` [PATCH 04/12] lei: show {pct} and {oid} in From_ lines and filenames Eric Wong
2021-01-21 19:46 ` Eric Wong [this message]
2021-01-21 19:46 ` [PATCH 06/12] lei_to_mail: avoid segfault on exit Eric Wong
2021-01-21 19:46 ` [PATCH 07/12] lei: oneshot: use client $io[2] for placeholder Eric Wong
2021-01-21 19:46 ` [PATCH 08/12] lei: remove INT/QUIT/TERM handlers, fix daemon EOF Eric Wong
2021-01-21 19:46 ` [PATCH 09/12] lei_xsearch: reduce reference paths to lxs Eric Wong
2021-01-21 19:46 ` [PATCH 10/12] lei: remove @TO_CLOSE_ATFORK_CHILD Eric Wong
2021-01-21 19:46 ` [PATCH 11/12] lei: forget-external support with canonicalization Eric Wong
2021-01-21 19:46 ` [PATCH 12/12] lei forget-external: bash completion support Eric Wong
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
Avoid top-posting and favor interleaved quoting:
https://en.wikipedia.org/wiki/Posting_style#Interleaved_style
List information: https://public-inbox.org/README
* Reply using the --to, --cc, and --in-reply-to
switches of git-send-email(1):
git send-email \
--in-reply-to=20210121194624.32002-6-e@80x24.org \
--to=e@80x24.org \
--cc=meta@public-inbox.org \
/path/to/YOUR_REPLY
https://kernel.org/pub/software/scm/git/docs/git-send-email.html
* If your mail client supports setting the In-Reply-To header
via mailto: links, try the mailto: link
Be sure your reply has a Subject: header at the top and a blank line
before the message body.
Code repositories for project(s) associated with this public inbox
https://80x24.org/public-inbox.git
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox;
as well as URLs for read-only IMAP folder(s) and NNTP newsgroup(s).