X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FIPC.pm;h=50de1bed22d71ba4ac72cc3a6f781368907881c3;hb=7349713101700e488231ad9ffece8ee42de0928c;hp=c52441f7bf2628d5710130d78149c049e69b9656;hpb=9dfc0b670fc634b54998c3020f173b82de1915ac;p=public-inbox.git

diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
index c52441f7..50de1bed 100644
--- a/lib/PublicInbox/IPC.pm
+++ b/lib/PublicInbox/IPC.pm
@@ -2,18 +2,24 @@
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 
 # base class for remote IPC calls and workqueues, requires Storable or Sereal
+# - ipc_do and ipc_worker_* is for a single worker/producer and uses pipes
+# - wq_do and wq_worker* is for a single producer and multiple workers,
+#   using SOCK_SEQPACKET for work distribution
+# use ipc_do when you need work done on a certain process
+# use wq_do when your work can be done on any idle worker
 package PublicInbox::IPC;
 use strict;
 use v5.10.1;
+use parent qw(Exporter);
 use Carp qw(confess croak);
 use PublicInbox::DS qw(dwaitpid);
 use PublicInbox::Spawn;
-use POSIX qw(mkfifo WNOHANG);
+use PublicInbox::OnDestroy;
+use PublicInbox::WQWorker;
 use Socket qw(AF_UNIX MSG_EOR SOCK_STREAM);
-use Errno qw(EMSGSIZE);
-use File::Temp 0.19 (); # 0.19 for ->newdir
 my $SEQPACKET = eval { Socket::SOCK_SEQPACKET() }; # portable enough?
 use constant PIPE_BUF => $^O eq 'linux' ? 4096 : POSIX::_POSIX_PIPE_BUF();
+our @EXPORT_OK = qw(ipc_freeze ipc_thaw);
 my $WQ_MAX_WORKERS = 4096;
 my ($enc, $dec);
 # ->imports at BEGIN turns sereal_*_with_object into custom ops on 5.14+
@@ -29,12 +35,13 @@ BEGIN {
 };
 
 if ($enc && $dec) { # should be custom ops
-	*freeze = sub ($) { sereal_encode_with_object $enc, $_[0] };
-	*thaw = sub ($) { sereal_decode_with_object $dec, $_[0], my $ret };
+	*ipc_freeze = sub ($) { sereal_encode_with_object $enc, $_[0] };
+	*ipc_thaw = sub ($) { sereal_decode_with_object $dec, $_[0], my $ret };
 } else {
 	eval { # some distros have Storable as a separate package from Perl
 		require Storable;
-		Storable->import(qw(freeze thaw));
+		*ipc_freeze = \&Storable::freeze;
+		*ipc_thaw = \&Storable::thaw;
 		$enc = 1;
 	} // warn("Storable (part of Perl) missing: $@\n");
 }
@@ -52,12 +59,12 @@ sub _get_rec ($) {
 	chop($len) eq "\n" or croak "no LF byte in $len";
 	defined(my $n = read($r, my $buf, $len)) or croak "read error: $!";
 	$n == $len or croak "short read: $n != $len";
-	thaw($buf);
+	ipc_thaw($buf);
 }
 
 sub _pack_rec ($) {
 	my ($ref) = @_;
-	my $buf = freeze($ref);
+	my $buf = ipc_freeze($ref);
 	length($buf) . "\n" . $buf;
 }
 
@@ -107,16 +114,21 @@ sub ipc_worker_spawn {
 	if ($pid == 0) {
 		srand($seed);
 		eval { PublicInbox::DS->Reset };
-		delete @$self{qw(-wq_s1 -wq_workers -wq_ppid)};
+		delete @$self{qw(-wq_s1 -wq_s2 -wq_workers -wq_ppid)};
 		$w_req = $r_res = undef;
 		$w_res->autoflush(1);
 		$SIG{$_} = 'IGNORE' for (qw(TERM INT QUIT));
 		local $0 = $ident;
 		PublicInbox::DS::sig_setmask($sigset);
-		my $on_destroy = $self->ipc_atfork_child;
-		eval { ipc_worker_loop($self, $r_req, $w_res) };
+		# ensure we properly exit even if warn() dies:
+		my $end = PublicInbox::OnDestroy->new($$, sub { exit(!!$@) });
+		eval {
+			my $on_destroy = $self->ipc_atfork_child;
+			local %SIG = %SIG;
+			ipc_worker_loop($self, $r_req, $w_res);
+		};
 		die "worker $ident PID:$$ died: $@\n" if $@;
-		exit;
+		undef $end; # trigger exit
 	}
 	PublicInbox::DS::sig_setmask($sigset) unless $oldset;
 	$r_req = $w_res = undef;
@@ -128,20 +140,24 @@ 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
 sub ipc_atfork_prepare {}
 
+sub wq_atexit_child {}
+
 sub ipc_atfork_child {
 	my ($self) = @_;
 	my $io = delete($self->{-ipc_atfork_child_close}) or return;
@@ -151,7 +167,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) {
@@ -162,7 +178,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"
@@ -242,10 +258,11 @@ sub ipc_sibling_atfork_child {
 	$pid == $$ and die "BUG: $$ ipc_atfork_child called on itself";
 }
 
-sub _recv_and_run {
+sub recv_and_run {
 	my ($self, $s2, $len, $full_stream) = @_;
 	my @fds = $recv_cmd->($s2, my $buf, $len);
-	my $n = length($buf // '') or return;
+	return if scalar(@fds) && !defined($fds[0]);
+	my $n = length($buf) or return 0;
 	my $nfd = 0;
 	for my $fd (@fds) {
 		if (open(my $cmdfh, '+<&=', $fd)) {
@@ -261,39 +278,41 @@ sub _recv_and_run {
 		$n = length($buf);
 	}
 	# Sereal dies on truncated data, Storable returns undef
-	my $args = thaw($buf) // die "thaw error on buffer of size: $n";
+	my $args = ipc_thaw($buf) // die "thaw error on buffer of size: $n";
 	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;
 }
 
 sub wq_worker_loop ($) {
 	my ($self) = @_;
-	my $len = $self->{wq_req_len} // (4096 * 33);
-	my $s2 = $self->{-wq_s2} // die 'BUG: no -wq_s2';
-	1 while (_recv_and_run($self, $s2, $len));
+	my $wqw = PublicInbox::WQWorker->new($self);
+	PublicInbox::DS->SetPostLoopCallback(sub { $wqw->{sock} });
+	PublicInbox::DS->EventLoop;
+	PublicInbox::DS->Reset;
 }
 
 sub do_sock_stream { # via wq_do, for big requests
 	my ($self, $len) = @_;
-	_recv_and_run($self, delete $self->{0}, $len, 1);
+	recv_and_run($self, delete $self->{0}, $len, 1);
 }
 
 sub wq_do { # always async
 	my ($self, $sub, $ios, @args) = @_;
 	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;
+		my $buf = ipc_freeze([$sub, @args]);
+		my $n = $send_cmd->($s1, $fds, $buf, MSG_EOR);
+		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]);
 		$n = $send_cmd->($s1, [ fileno($r) ],
-				freeze(['do_sock_stream', length($buf)]),
+				ipc_freeze(['do_sock_stream', length($buf)]),
 				MSG_EOR) // croak "sendmsg: $!";
 		undef $r;
 		$n = $send_cmd->($w, $fds, $buf, 0) // croak "sendmsg: $!";
@@ -306,29 +325,32 @@ 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
 	}
 }
 
-sub _wq_worker_start ($$) {
-	my ($self, $oldset) = @_;
+sub _wq_worker_start ($$$) {
+	my ($self, $oldset, $fields) = @_;
 	my $seed = rand(0xffffffff);
 	my $pid = fork // die "fork: $!";
 	if ($pid == 0) {
 		srand($seed);
 		eval { PublicInbox::DS->Reset };
 		delete @$self{qw(-wq_s1 -wq_workers -wq_ppid)};
-		$SIG{$_} = 'IGNORE' for (qw(PIPE TTOU TTIN));
-		$SIG{$_} = 'DEFAULT' for (qw(TERM QUIT INT CHLD));
+		@$self{keys %$fields} = values(%$fields) if $fields;
+		$SIG{$_} = 'IGNORE' for (qw(PIPE));
+		$SIG{$_} = 'DEFAULT' for (qw(TTOU TTIN TERM QUIT INT CHLD));
 		local $0 = $self->{-wq_ident};
 		PublicInbox::DS::sig_setmask($oldset);
 		# ensure we properly exit even if warn() dies:
 		my $end = PublicInbox::OnDestroy->new($$, sub { exit(!!$@) });
-		my $on_destroy = $self->ipc_atfork_child;
-		eval { wq_worker_loop($self) };
+		eval {
+			my $on_destroy = $self->ipc_atfork_child;
+			local %SIG = %SIG;
+			wq_worker_loop($self);
+		};
 		warn "worker $self->{-wq_ident} PID:$$ died: $@" if $@;
-		undef $on_destroy;
 		undef $end; # trigger exit
 	} else {
 		$self->{-wq_workers}->{$pid} = \undef;
@@ -337,7 +359,7 @@ sub _wq_worker_start ($$) {
 
 # starts workqueue workers if Sereal or Storable is installed
 sub wq_workers_start {
-	my ($self, $ident, $nr_workers, $oldset) = @_;
+	my ($self, $ident, $nr_workers, $oldset, $fields) = @_;
 	($enc && $send_cmd && $recv_cmd && defined($SEQPACKET)) or return;
 	return if $self->{-wq_s1}; # idempotent
 	$self->{-wq_s1} = $self->{-wq_s2} = undef;
@@ -349,18 +371,18 @@ sub wq_workers_start {
 	my $sigset = $oldset // PublicInbox::DS::block_signals();
 	$self->{-wq_workers} = {};
 	$self->{-wq_ident} = $ident;
-	_wq_worker_start($self, $sigset) for (1..$nr_workers);
+	_wq_worker_start($self, $sigset, $fields) for (1..$nr_workers);
 	PublicInbox::DS::sig_setmask($sigset) unless $oldset;
 	$self->{-wq_ppid} = $$;
 }
 
 sub wq_worker_incr { # SIGTTIN handler
-	my ($self, $oldset) = @_;
+	my ($self, $oldset, $fields) = @_;
 	$self->{-wq_s2} or return;
 	return if wq_workers($self) >= $WQ_MAX_WORKERS;
 	$self->ipc_atfork_prepare;
 	my $sigset = $oldset // PublicInbox::DS::block_signals();
-	_wq_worker_start($self, $sigset);
+	_wq_worker_start($self, $sigset, $fields);
 	PublicInbox::DS::sig_setmask($sigset) unless $oldset;
 }
 
@@ -442,6 +464,6 @@ sub DESTROY {
 }
 
 # Sereal doesn't have dclone
-sub deep_clone { thaw(freeze($_[-1])) }
+sub deep_clone { ipc_thaw(ipc_freeze($_[-1])) }
 
 1;