]> Sergey Matveev's repositories - public-inbox.git/commitdiff
lei: switch to use SEQPACKET socketpair instead of pipe
authorEric Wong <e@80x24.org>
Tue, 2 Feb 2021 11:46:47 +0000 (11:46 +0000)
committerEric Wong <e@80x24.org>
Wed, 3 Feb 2021 07:32:32 +0000 (07:32 +0000)
This will allow us to use larger messages and do progress
reporting to accumulate in the main daemon.

MANIFEST
lib/PublicInbox/LEI.pm
lib/PublicInbox/LeiXSearch.pm
lib/PublicInbox/OpPipe.pm [deleted file]
lib/PublicInbox/PktOp.pm [new file with mode: 0644]

index 017dc7f2a356a0ffc7560647eb971f31a7aa7089..bcb9d08ed7e5e5a42d99f4b19d05d9623db64610 100644 (file)
--- a/MANIFEST
+++ b/MANIFEST
@@ -205,9 +205,9 @@ lib/PublicInbox/NNTPD.pm
 lib/PublicInbox/NNTPdeflate.pm
 lib/PublicInbox/NewsWWW.pm
 lib/PublicInbox/OnDestroy.pm
-lib/PublicInbox/OpPipe.pm
 lib/PublicInbox/Over.pm
 lib/PublicInbox/OverIdx.pm
+lib/PublicInbox/PktOp.pm
 lib/PublicInbox/ProcessPipe.pm
 lib/PublicInbox/Qspawn.pm
 lib/PublicInbox/Reply.pm
index 17ad18b9cf09aa304b211f5bc966e6cb548454de..737db1e16f56c627fedf3b9d10b84742bf771ff4 100644 (file)
@@ -306,7 +306,7 @@ sub qerr ($;@) { $_[0]->{opt}->{quiet} or err(shift, @_) }
 sub fail ($$;$) {
        my ($self, $buf, $exit_code) = @_;
        err($self, $buf) if defined $buf;
-       syswrite($self->{op_pipe}, '!') if $self->{op_pipe}; # fail_handler
+       send($self->{pkt_op}, '!', MSG_EOR) if $self->{pkt_op}; # fail_handler
        x_it($self, ($exit_code // 1) << 8);
        undef;
 }
@@ -369,14 +369,14 @@ sub io_restore ($$) {
 sub note_sigpipe { # triggers sigpipe_handler
        my ($self, $fd) = @_;
        close(delete($self->{$fd})); # explicit close silences Perl warning
-       syswrite($self->{op_pipe}, '|') if $self->{op_pipe};
+       send($self->{pkt_op}, '|', MSG_EOR) if $self->{pkt_op};
        x_it($self, 13);
 }
 
 sub atfork_child_wq {
        my ($self, $wq) = @_;
        io_restore($self, $wq);
-       -p $self->{op_pipe} or die 'BUG: {op_pipe} expected';
+       -S $self->{pkt_op} or die 'BUG: {pkt_op} expected';
        io_restore($self->{l2m}, $wq);
        %PATH2CFG = ();
        undef $errors_log;
@@ -408,7 +408,7 @@ sub atfork_parent_wq {
        $self->{env} = $env;
        delete @$lei{qw(3 -lei_store cfg old_1 pgr lxs)}; # keep l2m
        my @io = (delete(@$lei{qw(0 1 2)}),
-                       io_extract($lei, qw(sock op_pipe startq)));
+                       io_extract($lei, qw(sock pkt_op startq)));
        my $l2m = $lei->{l2m};
        if ($l2m && $l2m != $wq) { # $wq == lxs
                if (my $wq_s1 = $l2m->{-wq_s1}) {
index f630e79a9b1cb48687e5e1c8f2fbb3d3e615180b..e577ab09a3508bf9eab3aa7396eca2161cc6a922 100644 (file)
@@ -9,10 +9,11 @@ use strict;
 use v5.10.1;
 use parent qw(PublicInbox::LeiSearch PublicInbox::IPC);
 use PublicInbox::DS qw(dwaitpid);
-use PublicInbox::OpPipe;
+use PublicInbox::PktOp;
 use PublicInbox::Import;
 use File::Temp 0.19 (); # 0.19 for ->newdir
 use File::Spec ();
+use Socket qw(MSG_EOR);
 use PublicInbox::Search qw(xap_terms);
 use PublicInbox::Spawn qw(popen_rd spawn which);
 use PublicInbox::MID qw(mids);
@@ -353,7 +354,8 @@ sub query_prepare { # called by wq_do
        delete $lei->{l2m}->{-wq_s1};
        eval { $lei->{l2m}->do_augment($lei) };
        $lei->fail($@) if $@;
-       syswrite($lei->{op_pipe}, '.') == 1 or die "do_post_augment trigger: $!"
+       send($lei->{pkt_op}, '.', MSG_EOR) == 1 or
+               die "do_post_augment trigger: $!"
 }
 
 sub fail_handler ($;$$) {
@@ -380,20 +382,19 @@ sub do_query {
                fcntl($lei->{startq}, 1031, 4096) if $^O eq 'linux';
                $zpipe = $l2m->pre_augment($lei);
        }
-       pipe(my $done, $lei->{op_pipe}) or die "pipe $!";
+       my $in_loop = exists $lei->{sock};
+       my $ops = {
+               '|' => [ \&sigpipe_handler, $lei ],
+               '!' => [ \&fail_handler, $lei ],
+               '.' => [ \&do_post_augment, $lei, $zpipe, $au_done ],
+               '' => [ \&query_done, $lei ],
+       };
+       (my $op, $lei->{pkt_op}) = PublicInbox::PktOp->pair($ops, $in_loop);
        my ($lei_ipc, @io) = $lei->atfork_parent_wq($self);
-       delete($lei->{op_pipe});
+       delete($lei->{pkt_op});
 
        $lei->event_step_init; # wait for shutdowns
-       my $done_op = {
-               '' => [ \&query_done, $lei ],
-               '|' => [ \&sigpipe_handler, $lei ],
-               '!' => [ \&fail_handler, $lei ]
-       };
-       my $in_loop = exists $lei->{sock};
-       $done = PublicInbox::OpPipe->new($done, $done_op, $in_loop);
        if ($l2m) {
-               $done_op->{'.'} = [ \&do_post_augment, $lei, $zpipe, $au_done ];
                $self->wq_do('query_prepare', \@io, $lei_ipc);
                $io[1] = $zpipe->[1] if $zpipe;
        }
@@ -401,7 +402,7 @@ sub do_query {
        $self->wq_close(1);
        unless ($in_loop) {
                # for the $lei_ipc->atfork_child_wq PIPE handler:
-               while ($done->{sock}) { $done->event_step }
+               while ($op->{sock}) { $op->event_step }
        }
 }
 
diff --git a/lib/PublicInbox/OpPipe.pm b/lib/PublicInbox/OpPipe.pm
deleted file mode 100644 (file)
index 295a8aa..0000000
+++ /dev/null
@@ -1,41 +0,0 @@
-# Copyright (C) 2021 all contributors <meta@public-inbox.org>
-# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
-
-# bytecode dispatch pipe, reads a byte, runs a sub
-# byte => [ sub, @operands ]
-package PublicInbox::OpPipe;
-use strict;
-use v5.10.1;
-use parent qw(PublicInbox::DS);
-use PublicInbox::Syscall qw(EPOLLIN);
-
-sub new {
-       my ($cls, $rd, $op_map, $in_loop) = @_;
-       my $self = bless { sock => $rd, op_map => $op_map }, $cls;
-       # 1031: F_SETPIPE_SZ, 4096: page size
-       fcntl($rd, 1031, 4096) if $^O eq 'linux';
-       if ($in_loop) { # iff using DS->EventLoop
-               $rd->blocking(0);
-               $self->SUPER::new($rd, EPOLLIN);
-       }
-       $self;
-}
-
-sub event_step {
-       my ($self) = @_;
-       my $rd = $self->{sock};
-       my $byte;
-       until (defined(sysread($rd, $byte, 1))) {
-               return if $!{EAGAIN};
-               next if $!{EINTR};
-               die "read \$rd: $!";
-       }
-       my $op = $self->{op_map}->{$byte} or die "BUG: unknown byte `$byte'";
-       if ($byte eq '') { # close on EOF
-               $rd->blocking ? delete($self->{sock}) : $self->close;
-       }
-       my ($sub, @args) = @$op;
-       $sub->(@args);
-}
-
-1;
diff --git a/lib/PublicInbox/PktOp.pm b/lib/PublicInbox/PktOp.pm
new file mode 100644 (file)
index 0000000..d5b95a7
--- /dev/null
@@ -0,0 +1,79 @@
+# Copyright (C) 2021 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# op dispatch socket, reads a message, runs a sub
+# There may be multiple producers, but (for now) only one consumer
+# Used for lei_xsearch and maybe other things
+# "literal" => [ sub, @operands ]
+# /regexp/ => [ sub, @operands ]
+package PublicInbox::PktOp;
+use strict;
+use v5.10.1;
+use parent qw(PublicInbox::DS);
+use Errno qw(EAGAIN EINTR);
+use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
+use Socket qw(AF_UNIX MSG_EOR SOCK_SEQPACKET);
+
+sub new {
+       my ($cls, $r, $ops, $in_loop) = @_;
+       my $self = bless { sock => $r, ops => $ops, re => [] }, $cls;
+       if (ref($ops) eq 'ARRAY') {
+               my %ops;
+               for my $op (@$ops) {
+                       if (ref($op->[0])) {
+                               push @{$self->{re}}, $op;
+                       } else {
+                               $ops{$op->[0]} = $op->[1];
+                       }
+               }
+               $self->{ops} = \%ops;
+       }
+       if ($in_loop) { # iff using DS->EventLoop
+               $r->blocking(0);
+               $self->SUPER::new($r, EPOLLIN|EPOLLET);
+       }
+       $self;
+}
+
+# returns a blessed object as the consumer, and a GLOB/IO for the producer
+sub pair {
+       my ($cls, $ops, $in_loop) = @_;
+       my ($c, $p);
+       socketpair($c, $p, AF_UNIX, SOCK_SEQPACKET, 0) or die "socketpair: $!";
+       (new($cls, $c, $ops, $in_loop), $p);
+}
+
+sub close {
+       my ($self) = @_;
+       my $c = $self->{sock} or return;
+       $c->blocking ? delete($self->{sock}) : $self->SUPER::close;
+}
+
+sub event_step {
+       my ($self) = @_;
+       my $c = $self->{sock};
+       my $msg;
+       do {
+               my $n = recv($c, $msg, 128, 0);
+               unless (defined $n) {
+                       return if $! == EAGAIN;
+                       next if $! == EINTR;
+                       $self->close;
+                       die "recv: $!";
+               }
+               my $op = $self->{ops}->{$msg};
+               unless ($op) {
+                       for my $re_op (@{$self->{re}}) {
+                               $msg =~ $re_op->[0] or next;
+                               $op = $re_op->[1];
+                               last;
+                       }
+               }
+               die "BUG: unknown message: `$msg'" unless $op;
+               my ($sub, @args) = @$op;
+               $sub->(@args);
+               return $self->close if $msg eq ''; # close on EOF
+       } while (1);
+}
+
+1;