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
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;
}
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;
$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}) {
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);
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 ($;$$) {
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;
}
$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 }
}
}
+++ /dev/null
-# 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;
--- /dev/null
+# 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;