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

diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
index dbb87e4e..50de1bed 100644
--- a/lib/PublicInbox/IPC.pm
+++ b/lib/PublicInbox/IPC.pm
@@ -2,16 +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(WNOHANG);
-use Socket qw(AF_UNIX MSG_EOR);
+use PublicInbox::OnDestroy;
+use PublicInbox::WQWorker;
+use Socket qw(AF_UNIX MSG_EOR SOCK_STREAM);
 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+
@@ -27,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");
 }
@@ -44,23 +53,18 @@ my $send_cmd = PublicInbox::Spawn->can('send_cmd4') // do {
 	PublicInbox::CmdIPC4->can('send_cmd4');
 };
 
-sub wq_set_recv_modes {
-	my ($self, @modes) = @_;
-	$self->{-wq_recv_modes} = \@modes;
-}
-
 sub _get_rec ($) {
 	my ($r) = @_;
 	defined(my $len = <$r>) or return;
 	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;
 }
 
@@ -105,19 +109,26 @@ sub ipc_worker_spawn {
 	pipe(my ($r_res, $w_res)) or die "pipe: $!";
 	my $sigset = $oldset // PublicInbox::DS::block_signals();
 	$self->ipc_atfork_prepare;
-	defined(my $pid = fork) or die "fork: $!";
+	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)};
+		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;
@@ -129,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;
@@ -152,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) {
@@ -163,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"
@@ -243,63 +258,99 @@ sub ipc_sibling_atfork_child {
 	$pid == $$ and die "BUG: $$ ipc_atfork_child called on itself";
 }
 
-sub wq_worker_loop ($) {
-	my ($self) = @_;
-	my $len = $self->{wq_req_len} // (4096 * 33);
-	my $s2 = $self->{-wq_s2} // die 'BUG: no -wq_s2';
-	while (1) {
-		my @fds = $recv_cmd->($s2, my $buf, $len) or return; # EOF
-		my @m = @{$self->{-wq_recv_modes} // [qw( +<&= >&= >&= )]};
-		my $nfd = 0;
-		for my $fd (@fds) {
-			my $mode = shift(@m);
-			if (open(my $cmdfh, $mode, $fd)) {
-				$self->{$nfd++} = $cmdfh;
-				$cmdfh->autoflush(1);
-			} else {
-				die "$$ open($mode$fd) (FD:$nfd): $!";
-			}
+sub recv_and_run {
+	my ($self, $s2, $len, $full_stream) = @_;
+	my @fds = $recv_cmd->($s2, my $buf, $len);
+	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)) {
+			$self->{$nfd++} = $cmdfh;
+			$cmdfh->autoflush(1);
+		} else {
+			die "$$ open(+<&=$fd) (FD:$nfd): $!";
 		}
-		# Sereal dies on truncated data, Storable returns undef
-		my $args = thaw($buf) //
-			die "thaw error on buffer of size:".length($buf);
-		my $sub = shift @$args;
-		eval { $self->$sub(@$args) };
-		warn "$$ wq_worker: $@" if $@ &&
-					ref($@) ne 'PublicInbox::SIGPIPE';
-		delete @$self{0..($nfd-1)};
 	}
+	while ($full_stream && $n < $len) {
+		my $r = sysread($s2, $buf, $len - $n, $n) // croak "read: $!";
+		croak "read EOF after $n/$len bytes" if $r == 0;
+		$n = length($buf);
+	}
+	# Sereal dies on truncated data, Storable returns undef
+	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 $@;
+	delete @$self{0..($nfd-1)};
+	$n;
+}
+
+sub wq_worker_loop ($) {
+	my ($self) = @_;
+	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);
 }
 
 sub wq_do { # always async
 	my ($self, $sub, $ios, @args) = @_;
 	if (my $s1 = $self->{-wq_s1}) { # run in worker
 		my $fds = [ map { fileno($_) } @$ios ];
-		$send_cmd->($s1, $fds, freeze([$sub, @args]), MSG_EOR);
+		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: $!";
+		$n = $send_cmd->($s1, [ fileno($r) ],
+				ipc_freeze(['do_sock_stream', length($buf)]),
+				MSG_EOR) // croak "sendmsg: $!";
+		undef $r;
+		$n = $send_cmd->($w, $fds, $buf, 0) // croak "sendmsg: $!";
+		while ($n < length($buf)) {
+			my $x = syswrite($w, $buf, length($buf) - $n, $n) //
+					croak "syswrite: $!";
+			croak "syswrite wrote 0 bytes" if $x == 0;
+			$n += $x;
+		}
 	} 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;
@@ -308,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;
@@ -320,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;
 }
 
@@ -413,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;