ipc: generic IPC dispatch based on Storable

I intend to use this with LeiStore when importing from multiple
slow sources at once (e.g. curl, IMAP, etc).  This is because
over.sqlite3 can only have a single writer, and we'll have
several slow readers running in parallel.

Watch and SearchIdxShard should also be able to use this code
in the future, but this will be proven with LeiStore, first.

2 files changed, 130 insertions, 1 deletions

diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
new file mode 100644
index 00000000..c04140ae
--- /dev/null
+++ b/lib/PublicInbox/IPC.pm
@@ -0,0 +1,129 @@
+# Copyright (C) 2020 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# base class for remote IPC calls, requires Storable
+# TODO: this ought to be usable in SearchIdxShard
+package PublicInbox::IPC;
+use strict;
+use v5.10.1;
+use Socket qw(AF_UNIX SOCK_STREAM);
+use Carp qw(confess croak);
+use PublicInbox::Sigfd;
+
+sub _get_rec ($) {
+        my ($sock) = @_;
+        local $/ = "\n";
+        defined(my $len = <$sock>) or return;
+        chop($len) eq "\n" or croak "no LF byte in $len";
+        defined(my $r = read($sock, my $buf, $len)) or croak "read error: $!";
+        $r == $len or croak "short read: $r != $len";
+        thaw($buf);
+}
+
+sub _send_rec ($$) {
+        my ($sock, $ref) = @_;
+        my $buf = freeze($ref);
+        print $sock length($buf), "\n", $buf or croak "print: $!";
+}
+
+sub ipc_return ($$$) {
+        my ($s2, $ret, $exc) = @_;
+        _send_rec($s2, $exc ? bless(\$exc, 'PublicInbox::IPC::Die') : $ret);
+}
+
+sub ipc_worker_loop ($$) {
+        my ($self, $s2) = @_;
+        $self->ipc_atfork_child if $self->can('ipc_atfork_child');
+        $s2->autoflush(1);
+        while (my $rec = _get_rec($s2)) {
+                my ($wantarray, $sub, @args) = @$rec;
+                if (!defined($wantarray)) { # no waiting if client doesn't care
+                        eval { $self->$sub(@args) };
+                        eval { warn "die: $@ (from nowait $sub)\n" } if $@;
+                } elsif ($wantarray) {
+                        my @ret = eval { $self->$sub(@args) };
+                        ipc_return($s2, \@ret, $@);
+                } else {
+                        my $ret = eval { $self->$sub(@args) };
+                        ipc_return($s2, \$ret, $@);
+                }
+        }
+}
+
+sub ipc_worker_spawn ($$$) {
+        my ($self, $ident, $oldset) = @_;
+        eval { require Storable; Storable->import(qw(freeze thaw)); };
+        if ($@) {
+                state $w //= warn "Storable (part of Perl) missing: $@\n";
+                return;
+        }
+        my $pid = $self->{-ipc_worker_pid};
+        confess "BUG: already spawned PID:$pid" if $pid;
+        confess "BUG: already have worker socket" if $self->{-ipc_sock};
+        my ($s1, $s2);
+        socketpair($s1, $s2, AF_UNIX, SOCK_STREAM, 0) or die "socketpair: $!";
+        my $sigset = $oldset // PublicInbox::Sigfd::block_signals();
+        defined($pid = fork) or die "fork: $!";
+        if ($pid == 0) {
+                undef $s1;
+                local $0 = $ident;
+                $SIG{$_} = 'IGNORE' for (qw(TERM INT QUIT));
+                PublicInbox::Sigfd::sig_setmask($oldset);
+                eval { ipc_worker_loop($self, $s2) };
+                die "worker $ident died: $@\n" if $@;
+                $self->ipc_at_worker_exit if $self->can('ipc_at_worker_exit');
+                exit;
+        }
+        PublicInbox::Sigfd::sig_setmask($sigset) unless $oldset;
+        $s1->autoflush(1);
+        $self->{-ipc_sock} = $s1;
+        $self->{-ipc_worker_pid} = $pid;
+}
+
+sub ipc_reap_worker { # dwaitpid callback
+        my ($self, $pid) = @_;
+        warn "PID:$pid died with \$?=$?\n" if $?;
+}
+
+sub ipc_worker_stop {
+        my ($self) = @_;
+        my $pid;
+        if (delete $self->{-ipc_sock}) {
+                $pid = delete $self->{-ipc_worker_pid} or die "no PID?";
+        } else {
+                $pid = delete $self->{-ipc_worker_pid} and
+                        die "unexpected PID:$pid";
+        }
+        return unless $pid;
+        eval { PublicInbox::DS::dwaitpid($pid, \&ipc_reap_worker, $self) };
+        if ($@) {
+                my $wp = waitpid($pid, 0);
+                $pid == $wp or die "waitpid($pid) returned $wp: \$?=$?";
+                ipc_reap_worker($self, $pid);
+        }
+}
+
+# use this if we have multiple readers reading curl or "pigz -dc"
+# and writing to the same store
+sub ipc_lock_init {
+        my ($self, $f) = @_;
+        require PublicInbox::Lock;
+        $self->{-ipc_lock} //= bless { lock_path => $f }, 'PublicInbox::Lock'
+}
+
+sub ipc_do {
+        my ($self, $sub, @args) = @_;
+        if (my $s1 = $self->{-ipc_sock}) {
+                my $ipc_lock = $self->{-ipc_lock};
+                my $lock = $ipc_lock ? $ipc_lock->lock_for_scope : undef;
+                _send_rec($s1, [ wantarray, $sub, @args ]);
+                return unless defined(wantarray);
+                my $ret = _get_rec($s1) // die "no response on $sub";
+                die $$ret if ref($ret) eq 'PublicInbox::IPC::Die';
+                wantarray ? @$ret : $$ret;
+        } else {
+                $self->$sub(@args);
+        }
+}
+
+1;
diff --git a/lib/PublicInbox/LeiStore.pm b/lib/PublicInbox/LeiStore.pm
index f8383d5e..2745c560 100644
--- a/lib/PublicInbox/LeiStore.pm
+++ b/lib/PublicInbox/LeiStore.pm
@@ -9,7 +9,7 @@
 package PublicInbox::LeiStore;
 use strict;
 use v5.10.1;
-use parent qw(PublicInbox::Lock);
+use parent qw(PublicInbox::Lock PublicInbox::IPC);
 use PublicInbox::SearchIdx qw(crlf_adjust);
 use PublicInbox::ExtSearchIdx;
 use PublicInbox::Import;