diff options
| -rw-r--r-- | lib/PublicInbox/LEI.pm | 15 | ||||
| -rw-r--r-- | lib/PublicInbox/LeiToMail.pm | 7 | ||||
| -rw-r--r-- | lib/PublicInbox/LeiXSearch.pm | 23 | ||||
| -rw-r--r-- | xt/lei-sigpipe.t | 29 |
4 files changed, 50 insertions, 24 deletions
diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm index 4b1dc673..802d2cd9 100644 --- a/lib/PublicInbox/LEI.pm +++ b/lib/PublicInbox/LEI.pm @@ -301,10 +301,13 @@ sub atfork_child_wq { PIPE => sub { $self->x_it(13); # SIGPIPE = 13 # we need to close explicitly to avoid Perl warning on SIGPIPE - close(delete $self->{1}); - # regular files and /dev/null (-c) won't trigger SIGPIPE - close(delete $self->{2}) unless (-f $self->{2} || -c _); - syswrite($self->{0}, '!') unless $self->{sock}; # for eof_wait + for my $i (1, 2) { + next unless $self->{$i} && (-p $self->{$i} || -S _); + close(delete $self->{$i}); + } + # trigger the LeiXSearch $done OpPipe: + syswrite($self->{0}, '!') if $self->{0} && -p $self->{0}; + $SIG{PIPE} = 'DEFAULT'; die bless(\"$_[0]", 'PublicInbox::SIGPIPE'), }); } @@ -322,7 +325,7 @@ sub atfork_parent_wq { my @io = delete @$ret{0..2}; $io[3] = delete($ret->{sock}) // *STDERR{GLOB}; my $l2m = $ret->{l2m}; - if ($l2m && $l2m != $wq) { + 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; @@ -672,7 +675,7 @@ sub start_mua { @cmd = map { $_ eq '%f' ? ($replaced = $mfolder) : $_ } @cmd; push @cmd, $mfolder unless defined($replaced); $sock //= $self->{sock}; - if ($PublicInbox::DS::in_loop) { # lei(1) client process runs it + if ($sock) { # lei(1) client process runs it send($sock, exec_buf(\@cmd, {}), MSG_EOR); } else { # oneshot $self->{"mua.pid.$self.$$"} = spawn(\@cmd); diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm index a1dce550..8e58ad11 100644 --- a/lib/PublicInbox/LeiToMail.pm +++ b/lib/PublicInbox/LeiToMail.pm @@ -247,11 +247,16 @@ sub _mbox_write_cb ($$) { $dedupe->prepare_dedupe; sub { # for git_to_mail my ($buf, $oid, $kw) = @_; + return unless $out; my $eml = PublicInbox::Eml->new($buf); if (!$dedupe->is_dup($eml, $oid)) { $buf = $eml2mbox->($eml, $kw); my $lk = $ovv->lock_for_scope; - $write->($out, $buf); + eval { $write->($out, $buf) }; + if ($@) { + die $@ if ref($@) ne 'PublicInbox::SIGPIPE'; + undef $out + } } } } diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm index 73fd17f4..45a073a0 100644 --- a/lib/PublicInbox/LeiXSearch.pm +++ b/lib/PublicInbox/LeiXSearch.pm @@ -219,7 +219,7 @@ sub start_query { # always runs in main (lei-daemon) process @$io = (); } -sub query_prepare { # for wq_do, +sub query_prepare { # called by wq_do my ($self, $lei) = @_; my %sig = $lei->atfork_child_wq($self); local @SIG{keys %sig} = values %sig; @@ -227,6 +227,18 @@ sub query_prepare { # for wq_do, $lei->fail($@) if $@; } +sub sigpipe_handler { + my ($self, $lei_orig, $pids) = @_; + if ($pids) { # one-shot (no event loop) + kill 'TERM', @$pids; + kill 'PIPE', $$; + } else { + $self->wq_kill; + $self->wq_close; + } + close(delete $lei_orig->{1}) if $lei_orig->{1}; +} + sub do_query { my ($self, $lei_orig, $srcs) = @_; my ($lei, @io) = $lei_orig->atfork_parent_wq($self); @@ -234,7 +246,10 @@ sub do_query { pipe(my $done, $io[0]) or die "pipe $!"; $lei_orig->event_step_init; # wait for shutdowns - my $done_op = { '' => [ \&query_done, $self, $lei_orig ] }; + my $done_op = { + '' => [ \&query_done, $self, $lei_orig ], + '!' => [ \&sigpipe_handler, $self, $lei_orig ] + }; my $in_loop = exists $lei_orig->{sock}; $done = PublicInbox::OpPipe->new($done, $done_op, $in_loop); my $l2m = $lei->{l2m}; @@ -244,7 +259,7 @@ sub do_query { my @l2m_io = (undef, @io[1..$#io]); pipe(my $startq, $l2m_io[0]) or die "pipe: $!"; $self->wq_do('query_prepare', \@l2m_io, $lei); - $io[4] //= *STDERR{GLOB}; + $io[4] = *STDERR{GLOB}; # don't send l2m->{-wq_s1} die "BUG: unexpected \$io[5]: $io[5]" if $io[5]; fcntl($startq, 1031, 4096) if $^O eq 'linux'; # F_SETPIPE_SZ $io[5] = $startq; @@ -253,7 +268,7 @@ sub do_query { unless ($in_loop) { my @pids = $self->wq_close; # for the $lei->atfork_child_wq PIPE handler: - $done_op->{'!'} = [ \&CORE::kill, 'TERM', @pids ]; + $done_op->{'!'}->[3] = \@pids; $done->event_step; my $ipc_worker_reap = $self->can('ipc_worker_reap'); if (my $l2m_pids = delete $self->{l2m_pids}) { diff --git a/xt/lei-sigpipe.t b/xt/lei-sigpipe.t index 4d35bbb3..448bd7db 100644 --- a/xt/lei-sigpipe.t +++ b/xt/lei-sigpipe.t @@ -11,19 +11,22 @@ require_mods(qw(json DBD::SQLite Search::Xapian)); my $do_test = sub { my $env = shift // {}; - pipe(my ($r, $w)) or BAIL_OUT $!; - open my $err, '+>', undef or BAIL_OUT $!; - my $opt = { run_mode => 0, 1 => $w, 2 => $err }; - my $tp = start_script([qw(lei q -t), 'bytes:1..'], $env, $opt); - close $w; - sysread($r, my $buf, 1); - close $r; # trigger SIGPIPE - $tp->join; - ok(WIFSIGNALED($?), 'signaled'); - is(WTERMSIG($?), SIGPIPE, 'got SIGPIPE'); - seek($err, 0, 0); - my @err = grep(!m{mkdir /dev/null\b}, <$err>); - is_deeply(\@err, [], 'no errors'); + for my $out ([], [qw(-f mboxcl2)]) { + pipe(my ($r, $w)) or BAIL_OUT $!; + open my $err, '+>', undef or BAIL_OUT $!; + my $opt = { run_mode => 0, 1 => $w, 2 => $err }; + my $cmd = [qw(lei q -t), @$out, 'bytes:1..']; + my $tp = start_script($cmd, $env, $opt); + close $w; + sysread($r, my $buf, 1); + close $r; # trigger SIGPIPE + $tp->join; + ok(WIFSIGNALED($?), "signaled @$out"); + is(WTERMSIG($?), SIGPIPE, "got SIGPIPE @$out"); + seek($err, 0, 0); + my @err = grep(!m{mkdir /dev/null\b}, <$err>); + is_deeply(\@err, [], "no errors @$out"); + } }; $do_test->(); |