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 CA41E1FA17 for ; Fri, 29 Jan 2021 07:43:00 +0000 (UTC) From: Eric Wong To: meta@public-inbox.org Subject: [PATCH 4/7] lei: less error-prone FD mapping Date: Fri, 29 Jan 2021 12:42:57 +0500 Message-Id: <20210129074300.14475-5-e@80x24.org> In-Reply-To: <20210129074300.14475-1-e@80x24.org> References: <20210129074300.14475-1-e@80x24.org> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit List-Id: Keeping track of non-standard FDs gets tricky, so make it easier by relying on st_dev/st_ino mapping in the transmitted objects. We'll keep using numbers for the standard FDs since we need to be able to easily redirect them in the producer (main daemon) process for (gzip|bzip2|xz) if writing to a compressed mbox. --- lib/PublicInbox/LEI.pm | 56 +++++++++++++++++++++------- lib/PublicInbox/LeiOverview.pm | 9 ++--- lib/PublicInbox/LeiToMail.pm | 8 +--- lib/PublicInbox/LeiXSearch.pm | 68 +++++++++++++++------------------- lib/PublicInbox/Spawn.pm | 2 +- 5 files changed, 77 insertions(+), 66 deletions(-) diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm index f5413aab..3ed330f9 100644 --- a/lib/PublicInbox/LEI.pm +++ b/lib/PublicInbox/LEI.pm @@ -335,14 +335,27 @@ sub atfork_prepare_wq { } } +sub io_restore ($$) { + my ($dst, $src) = @_; + for my $i (0..2) { # standard FDs + my $io = delete $src->{$i} or next; + $dst->{$i} = $io; + } + for my $i (3..9) { # named (non-standard) FDs + my $io = $src->{$i} or next; + my @st = stat($io) or die "stat $src.$i ($io): $!"; + my $f = delete $dst->{"dev=$st[0],ino=$st[1]"} // next; + $dst->{$f} = $io; + delete $src->{$i}; + } +} + # usage: my %sig = $lei->atfork_child_wq($wq); # local @SIG{keys %sig} = values %sig; sub atfork_child_wq { my ($self, $wq) = @_; - my ($sock, $l2m_wq_s1); - (@$self{qw(0 1 2)}, $sock, $l2m_wq_s1) = delete(@$wq{0..4}); - $self->{sock} = $sock if -S $sock; - $self->{l2m}->{-wq_s1} = $l2m_wq_s1 if $l2m_wq_s1 && -S $l2m_wq_s1; + io_restore($self, $wq); + io_restore($self->{l2m}, $wq); %PATH2CFG = (); undef $errors_log; $quit = \&CORE::exit; @@ -355,30 +368,45 @@ sub atfork_child_wq { close(delete $self->{$i}); } # trigger the LeiXSearch $done OpPipe: - syswrite($self->{0}, '!') if $self->{0} && -p $self->{0}; + syswrite($self->{op_pipe}, '!') if $self->{op_pipe}; $SIG{PIPE} = 'DEFAULT'; die bless(\"$_[0]", 'PublicInbox::SIGPIPE'), }); } +sub io_extract ($;@) { + my ($obj, @fields) = @_; + my @io; + for my $f (@fields) { + my $io = delete $obj->{$f} or next; + my @st = stat($io) or die "W: stat $obj.$f ($io): $!"; + $obj->{"dev=$st[0],ino=$st[1]"} = $f; + push @io, $io; + } + @io +} + # usage: ($lei, @io) = $lei->atfork_parent_wq($wq); sub atfork_parent_wq { my ($self, $wq) = @_; my $env = delete $self->{env}; # env is inherited at fork - my $ret = bless { %$self }, ref($self); - if (my $dedupe = delete $ret->{dedupe}) { - $ret->{dedupe} = $wq->deep_clone($dedupe); + my $lei = bless { %$self }, ref($self); + if (my $dedupe = delete $lei->{dedupe}) { + $lei->{dedupe} = $wq->deep_clone($dedupe); } $self->{env} = $env; - delete @$ret{qw(3 -lei_store cfg old_1 pgr lxs)}; # keep l2m - my @io = delete @$ret{0..2}; - $io[3] = delete($ret->{sock}) // $io[2]; - my $l2m = $ret->{l2m}; + delete @$lei{qw(3 -lei_store cfg old_1 pgr lxs)}; # keep l2m + my @io = (delete(@$lei{qw(0 1 2)}), + io_extract($lei, qw(sock op_pipe startq))); + my $l2m = $lei->{l2m}; if ($l2m && $l2m != $wq) { # $wq == lxs - $io[4] = $l2m->{-wq_s1} if $l2m->{-wq_s1}; + if (my $wq_s1 = $l2m->{-wq_s1}) { + push @io, io_extract($l2m, '-wq_s1'); + $l2m->{-wq_s1} = $wq_s1; + } $l2m->wq_close(1); } - ($ret, @io); + ($lei, @io); } sub _help ($;$) { diff --git a/lib/PublicInbox/LeiOverview.pm b/lib/PublicInbox/LeiOverview.pm index f9a28138..c67e2747 100644 --- a/lib/PublicInbox/LeiOverview.pm +++ b/lib/PublicInbox/LeiOverview.pm @@ -220,14 +220,13 @@ sub ovv_each_smsg_cb { # runs in wq worker usually }; } elsif ($l2m && $l2m->{-wq_s1}) { my ($lei_ipc, @io) = $lei->atfork_parent_wq($l2m); - # n.b. $io[0] = qry_status_wr, $io[1] = mbox|stdout, - # $io[4] becomes a notification pipe that triggers EOF + # $io[-1] becomes a notification pipe that triggers EOF # in this wq worker when all outstanding ->write_mail # calls are complete - die "BUG: \$io[4] $io[4] unexpected" if $io[4]; - pipe($l2m->{each_smsg_done}, $io[4]) or die "pipe: $!"; - fcntl($io[4], 1031, 4096) if $^O eq 'linux'; + pipe($l2m->{each_smsg_done}, $io[$#io + 1]) or die "pipe: $!"; + fcntl($io[-1], 1031, 4096) if $^O eq 'linux'; # F_SETPIPE_SZ delete @$lei_ipc{qw(l2m opt mset_opt cmd)}; + $lei_ipc->{each_smsg_not_done} = $#io; my $git = $ibxish->git; # (LeiXSearch|Inbox|ExtSearch)->git $self->{git} = $git; my $git_dir = $git->{git_dir}; diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm index 08a1570d..61b546b5 100644 --- a/lib/PublicInbox/LeiToMail.pm +++ b/lib/PublicInbox/LeiToMail.pm @@ -460,7 +460,7 @@ sub post_augment { # fast (spawn compressor or mkdir), runs in main daemon sub write_mail { # via ->wq_do my ($self, $git_dir, $smsg, $lei) = @_; - my $not_done = delete $self->{4}; # write end of {each_smsg_done} + my $not_done = delete $self->{$lei->{each_smsg_not_done}}; my $wcb = $self->{wcb} //= do { # first message my %sig = $lei->atfork_child_wq($self); @SIG{keys %sig} = values %sig; # not local @@ -471,12 +471,6 @@ sub write_mail { # via ->wq_do $git->cat_async($smsg->{blob}, \&git_to_mail, [$wcb, $smsg, $not_done]); } -sub ipc_atfork_prepare { - my ($self) = @_; - # FDs: (done_wr, stdout|mbox, stderr, 3: sock, 4: each_smsg_done_wr) - $self->SUPER::ipc_atfork_prepare; # PublicInbox::IPC -} - # We rely on OnDestroy to run this before ->DESTROY, since ->DESTROY # ordering is unstable at worker exit and may cause segfaults sub reap_gits { diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm index 9ea2b5f3..e69b637c 100644 --- a/lib/PublicInbox/LeiXSearch.pm +++ b/lib/PublicInbox/LeiXSearch.pm @@ -109,9 +109,9 @@ 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; + my $startq = delete $lei->{startq}; my ($srch, $over) = ($ibxish->search, $ibxish->over); unless ($srch && $over) { @@ -145,9 +145,9 @@ sub query_thread_mset { # for --thread sub query_mset { # non-parallel for non-"--thread" users my ($self, $lei) = @_; local $0 = "$0 query_mset"; - my $startq = delete $self->{5}; my %sig = $lei->atfork_child_wq($self); local @SIG{keys %sig} = values %sig; + my $startq = delete $lei->{startq}; my $mo = { %{$lei->{mset_opt}} }; my $mset; for my $loc (locals($self)) { @@ -173,7 +173,7 @@ sub each_eml { # callback for MboxReader->mboxrd $smsg->parse_references($eml, mids($eml)); $smsg->{$_} //= '' for qw(from to cc ds subject references mid); delete @$smsg{qw(From Subject -ds -ts)}; - if (my $startq = delete($self->{5})) { wait_startq($startq) } + if (my $startq = delete($lei->{startq})) { wait_startq($startq) } $each_smsg->($smsg, undef, $eml); } @@ -352,11 +352,12 @@ 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"; + -p $lei->{op_pipe} or die "BUG: \$done pipe expected"; local @SIG{keys %sig} = values %sig; + delete $lei->{l2m}->{-wq_s1}; eval { $lei->{l2m}->do_augment($lei) }; $lei->fail($@) if $@; - syswrite($lei->{0}, '.') == 1 or die "do_post_augment trigger: $!"; + syswrite($lei->{op_pipe}, '.') == 1 or die "do_post_augment trigger: $!" } sub sigpipe_handler { # handles SIGPIPE from l2m/lxs workers @@ -370,56 +371,45 @@ sub sigpipe_handler { # handles SIGPIPE from l2m/lxs workers } sub do_query { - my ($self, $lei_orig) = @_; - my ($lei, @io) = $lei_orig->atfork_parent_wq($self); - $io[0] = undef; - pipe(my $done, $io[0]) or die "pipe $!"; - $lei_orig->{1}->autoflush(1); + my ($self, $lei) = @_; + $lei->{1}->autoflush(1); + my ($au_done, $zpipe); + my $l2m = $lei->{l2m}; + if ($l2m) { + pipe($lei->{startq}, $au_done) or die "pipe: $!"; + # 1031: F_SETPIPE_SZ + fcntl($lei->{startq}, 1031, 4096) if $^O eq 'linux'; + $zpipe = $l2m->pre_augment($lei); + } + pipe(my $done, $lei->{op_pipe}) or die "pipe $!"; + my ($lei_ipc, @io) = $lei->atfork_parent_wq($self); + delete($lei->{op_pipe}); - $lei_orig->event_step_init; # wait for shutdowns + $lei->event_step_init; # wait for shutdowns my $done_op = { - '' => [ \&query_done, $lei_orig ], - '!' => [ \&sigpipe_handler, $lei_orig ] + '' => [ \&query_done, $lei ], + '!' => [ \&sigpipe_handler, $lei ] }; - my $in_loop = exists $lei_orig->{sock}; + my $in_loop = exists $lei->{sock}; $done = PublicInbox::OpPipe->new($done, $done_op, $in_loop); - my $l2m = $lei->{l2m}; if ($l2m) { - # may redirect $lei->{1} for mbox - my $zpipe = $l2m->pre_augment($lei_orig); - $io[1] = $lei_orig->{1}; - pipe(my ($startq, $au_done)) or die "pipe: $!"; - $done_op->{'.'} = [ \&do_post_augment, $lei_orig, - $zpipe, $au_done ]; - local $io[4] = *STDERR{GLOB}; # don't send l2m->{-wq_s1} - die "BUG: unexpected \$io[5]: $io[5]" if $io[5]; - $self->wq_do('query_prepare', \@io, $lei); - fcntl($startq, 1031, 4096) if $^O eq 'linux'; # F_SETPIPE_SZ - $io[5] = $startq; + $done_op->{'.'} = [ \&do_post_augment, $lei, $zpipe, $au_done ]; + $self->wq_do('query_prepare', \@io, $lei_ipc); $io[1] = $zpipe->[1] if $zpipe; } - start_query($self, \@io, $lei); + start_query($self, \@io, $lei_ipc); $self->wq_close(1); unless ($in_loop) { - # for the $lei->atfork_child_wq PIPE handler: + # for the $lei_ipc->atfork_child_wq PIPE handler: while ($done->{sock}) { $done->event_step } } } -sub ipc_atfork_prepare { - my ($self) = @_; - if (exists $self->{remotes}) { - require PublicInbox::MboxReader; - require IO::Uncompress::Gunzip; - } - # FDS: (0: done_wr, 1: stdout|mbox, 2: stderr, - # 3: sock, 4: $l2m->{-wq_s1}, 5: $startq) - $self->SUPER::ipc_atfork_prepare; # PublicInbox::IPC -} - sub add_uri { my ($self, $uri) = @_; if (my $curl = $self->{curl} //= which('curl') // 0) { + require PublicInbox::MboxReader; + require IO::Uncompress::Gunzip; push @{$self->{remotes}}, $uri; } else { warn "curl missing, ignoring $uri\n"; diff --git a/lib/PublicInbox/Spawn.pm b/lib/PublicInbox/Spawn.pm index ef4885c1..1842899c 100644 --- a/lib/PublicInbox/Spawn.pm +++ b/lib/PublicInbox/Spawn.pm @@ -209,7 +209,7 @@ my $fdpass = <<'FDPASS'; #include #if defined(CMSG_SPACE) && defined(CMSG_LEN) -#define SEND_FD_CAPA 6 +#define SEND_FD_CAPA 10 #define SEND_FD_SPACE (SEND_FD_CAPA * sizeof(int)) union my_cmsg { struct cmsghdr hdr;