lei: keep $lei around until workers are reaped

[public-inbox.git] / lib / PublicInbox / IPC.pm

diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
index 479c4377b3e07ba12aa8bada31523ca9b4efdcd9..689f32d0dc690a74058414d419f56b23d86c9f29 100644 (file)
--- a/lib/PublicInbox/IPC.pm
+++ b/lib/PublicInbox/IPC.pm
@@ -16,7 +16,6 @@ use PublicInbox::Spawn;
 use PublicInbox::OnDestroy;
 use PublicInbox::WQWorker;
 use Socket qw(AF_UNIX MSG_EOR SOCK_STREAM);
-use Errno qw(EMSGSIZE);
 my $SEQPACKET = eval { Socket::SOCK_SEQPACKET() }; # portable enough?
 use constant PIPE_BUF => $^O eq 'linux' ? 4096 : POSIX::_POSIX_PIPE_BUF();
 my $WQ_MAX_WORKERS = 4096;
@@ -138,15 +137,17 @@ sub ipc_worker_spawn {
 }
 
 sub ipc_worker_reap { # dwaitpid callback
-       my ($self, $pid) = @_;
-       # SIGTERM (15) is our default exit signal
-       warn "PID:$pid died with \$?=$?\n" if $? && ($? & 127) != 15;
+       my ($args, $pid) = @_;
+       return if !$?;
+       # TERM(15) is our default exit signal, PIPE(13) is likely w/ pager
+       my $s = $? & 127;
+       warn "PID:$pid died with \$?=$?\n" if $s != 15 && $s != 13;
 }
 
 sub wq_wait_old {
-       my ($self) = @_;
+       my ($self, $args) = @_;
        my $pids = delete $self->{"-wq_old_pids.$$"} or return;
-       dwaitpid($_, \&ipc_worker_reap, $self) for @$pids;
+       dwaitpid($_, \&ipc_worker_reap, [$self, $args]) for @$pids;
 }
 
 # for base class, override in sub classes
@@ -163,7 +164,7 @@ sub ipc_atfork_child {
 
 # idempotent, can be called regardless of whether worker is active or not
 sub ipc_worker_stop {
-       my ($self) = @_;
+       my ($self, $args) = @_;
        my ($pid, $ppid) = delete(@$self{qw(-ipc_pid -ipc_ppid)});
        my ($w_req, $r_res) = delete(@$self{qw(-ipc_req -ipc_res)});
        if (!$w_req && !$r_res) {
@@ -174,7 +175,7 @@ sub ipc_worker_stop {
        $w_req = $r_res = undef;
 
        return if $$ != $ppid;
-       dwaitpid($pid, \&ipc_worker_reap, $self);
+       dwaitpid($pid, \&ipc_worker_reap, [$self, $args]);
 }
 
 # use this if we have multiple readers reading curl or "pigz -dc"
@@ -278,7 +279,7 @@ sub recv_and_run {
        undef $buf;
        my $sub = shift @$args;
        eval { $self->$sub(@$args) };
-       warn "$$ wq_worker: $@" if $@ && ref($@) ne 'PublicInbox::SIGPIPE';
+       warn "$$ wq_worker: $@" if $@;
        delete @$self{0..($nfd-1)};
        $n;
 }
@@ -301,8 +302,9 @@ sub wq_do { # always async
        if (my $s1 = $self->{-wq_s1}) { # run in worker
                my $fds = [ map { fileno($_) } @$ios ];
                my $n = $send_cmd->($s1, $fds, freeze([$sub, @args]), MSG_EOR);
-               return if defined($n);
-               croak "sendmsg error: $!" if $! != EMSGSIZE;
+               return if defined($n); # likely
+               croak "sendmsg: $! (check RLIMIT_NOFILE)" if $!{ETOOMANYREFS};
+               croak "sendmsg: $!" if !$!{EMSGSIZE};
                socketpair(my $r, my $w, AF_UNIX, SOCK_STREAM, 0) or
                        croak "socketpair: $!";
                my $buf = freeze([$sub, @args]);
@@ -320,7 +322,7 @@ sub wq_do { # always async
        } else {
                @$self{0..$#$ios} = @$ios;
                eval { $self->$sub(@args) };
-               warn "wq_do: $@" if $@ && ref($@) ne 'PublicInbox::SIGPIPE';
+               warn "wq_do: $@" if $@;
                delete @$self{0..$#$ios}; # don't close
        }
 }