lei/store: use SOCK_SEQPACKET rather than pipe

This has several advantages:

* no need to use ipc.lock to protect a pipe for non-atomic writes

* ability to pass FDs.  In another commit, this will let us
  simplify lei->sto_done_request and pass newly-created
  sockets to lei/store directly.

disadvantages:

- an extra pipe is required for rare messages over several
  hundred KB, this is probably a non-issue, though

The performance delta is unknown, but I expect shards
(which remain pipes) to be the primary bottleneck IPC-wise
for lei/store.

1 files changed, 8 insertions, 2 deletions

diff --git a/lib/PublicInbox/LeiStore.pm b/lib/PublicInbox/LeiStore.pm
index 08add8f5..4ec63699 100644
--- a/lib/PublicInbox/LeiStore.pm
+++ b/lib/PublicInbox/LeiStore.pm
@@ -552,6 +552,12 @@ sub ipc_atfork_child {
         $self->SUPER::ipc_atfork_child;
 }
 
+sub recv_and_run {
+        my ($self, @args) = @_;
+        local $PublicInbox::DS::in_loop = 0; # waitpid synchronously
+        $self->SUPER::recv_and_run(@args);
+}
+
 sub write_prepare {
         my ($self, $lei) = @_;
         $lei // die 'BUG: $lei not passed';
@@ -560,14 +566,14 @@ sub write_prepare {
                 require PublicInbox::PktOp;
                 my ($s2d_op_c, $s2d_op_p) = PublicInbox::PktOp->pair;
                 my $dir = $lei->store_path;
-                $self->ipc_lock_init("$dir/ipc.lock");
                 substr($dir, -length('/lei/store'), 10, '');
                 pipe(my ($r, $w)) or die "pipe: $!";
                 $w->autoflush(1);
                 # Mail we import into lei are private, so headers filtered out
                 # by -mda for public mail are not appropriate
                 local @PublicInbox::MDA::BAD_HEADERS = ();
-                $self->ipc_worker_spawn("lei/store $dir", $lei->oldset, {
+                $self->{-wq_no_bcast} = 1;
+                $self->wq_workers_start("lei/store $dir", 1, $lei->oldset, {
                                         lei => $lei,
                                         -err_wr => $w,
                                         to_close => [ $r, $s2d_op_c->{sock} ],