lib/PublicInbox/AdminEdit.pm
lib/PublicInbox/AltId.pm
lib/PublicInbox/Cgit.pm
-lib/PublicInbox/CmdIPC1.pm
lib/PublicInbox/CmdIPC4.pm
lib/PublicInbox/CompressNoop.pm
lib/PublicInbox/Config.pm
+++ /dev/null
-# Copyright (C) 2021 all contributors <meta@public-inbox.org>
-# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
-
-# callers should use PublicInbox::CmdIPC1->can('send_cmd1') (or recv_cmd1)
-# 2nd choice for lei(1) front-end and 3rd choice for lei internals
-package PublicInbox::CmdIPC1;
-use strict;
-use v5.10.1;
-BEGIN { eval {
-require IO::FDPass; # XS, available in all major distros
-no warnings 'once';
-
-*send_cmd1 = sub ($$$$) { # (sock, fds, buf, flags) = @_;
- my ($sock, $fds, undef, $flags) = @_;
- for my $fd (@$fds) {
- IO::FDPass::send(fileno($sock), $fd) or
- die "IO::FDPass::send: $!";
- }
- send($sock, $_[2], $flags) or die "send $!";
-};
-
-*recv_cmd1 = sub ($$$;$) {
- my ($s, undef, $len, $nfds) = @_;
- $nfds //= 3;
- my @fds = map { IO::FDPass::recv(fileno($s)) } (1..$nfds);
- recv($s, $_[1], $len, 0) // die "recv: $!";
- length($_[1]) == 0 ? () : @fds;
-};
-
-} } # /eval /BEGIN
-
-1;
} // warn("Storable (part of Perl) missing: $@\n");
}
-my $recv_cmd1; # PublicInbox::CmdIPC1::recv_cmd1;
my $recv_cmd = PublicInbox::Spawn->can('recv_cmd4');
my $send_cmd = PublicInbox::Spawn->can('send_cmd4') // do {
require PublicInbox::CmdIPC4;
$recv_cmd //= PublicInbox::CmdIPC4->can('recv_cmd4');
PublicInbox::CmdIPC4->can('send_cmd4');
-} // do {
- # IO::FDPass only allows sending a single FD at-a-time, which
- # means we can't guarantee all packets end up on the same worker,
- # so we cap WQ_MAX_WORKERS
- require PublicInbox::CmdIPC1;
- $recv_cmd1 = PublicInbox::CmdIPC1->can('recv_cmd1');
- $WQ_MAX_WORKERS = 1 if $recv_cmd1;
- wq_set_recv_fds(3);
- PublicInbox::CmdIPC1->can('send_cmd1');
};
-# needed to tell recv_cmd1 how many times to loop IO::FDPass::recv
-sub wq_set_recv_fds {
- return unless $recv_cmd1;
- my $nfds = pop;
- my $sub = sub {
- my ($sock, $fds, undef, $flags) = @_;
- $recv_cmd1->($sock, $fds, $_[2], $flags, $nfds);
- };
- my $self = pop;
- if (ref $self) {
- $self->{-wq_recv_cmd} = $sub;
- } else {
- $recv_cmd = $sub;
- }
+sub wq_set_recv_modes {
+ my ($self, @modes) = @_;
+ $self->{-wq_recv_modes} = \@modes;
}
sub _get_rec ($) {
sub _close_recvd ($) {
my ($self) = @_;
- close($_) for (grep { defined } (delete @$self{0..2}));
+ my $x = $self->{-wq_recv_modes};
+ my $end = $x ? $#$x : 2;
+ close($_) for (grep { defined } (delete @$self{0..$end}));
}
sub wq_worker_loop ($) {
local $SIG{PIPE} = sub {
my $cur_sub = $sub;
_close_recvd($self);
- die(bless(\$cur_sub, __PACKAGE__.'::PIPE')) if $cur_sub;
+ die(bless(\$cur_sub, 'PublicInbox::SIGPIPE')) if $cur_sub;
};
- my $rcv = $self->{-wq_recv_cmd} // $recv_cmd;
while (1) {
- my (@fds) = $rcv->($s2, $buf, $len) or return; # EOF
+ my (@fds) = $recv_cmd->($s2, $buf, $len) or return; # EOF
my $i = 0;
- my @m = @{$self->{wq_open_modes} // [qw( +<&= >&= >&= )]};
+ my @m = @{$self->{-wq_recv_modes} // [qw( +<&= >&= >&= )]};
for my $fd (@fds) {
my $mode = shift(@m);
if (open(my $cmdfh, $mode, $fd)) {
undef $sub; # quiet SIG{PIPE} handler
die $@ if $@;
};
- warn "$$ wq_worker: $@" if $@ && ref $@ ne __PACKAGE__.'::PIPE';
+ warn "$$ wq_worker: $@" if $@ &&
+ ref($@) ne 'PublicInbox::SIGPIPE';
# need to close explicitly to avoid warnings after SIGPIPE
_close_recvd($self);
}
} else {
@$self{0..$#$ios} = @$ios;
eval { $self->$sub(@args) };
- warn "wq_do: $@" if $@;
- delete @$self{0..$#$ios};
+ warn "wq_do: $@" if $@ && ref($@) ne 'PublicInbox::SIGPIPE';
+ delete @$self{0..$#$ios}; # don't close
}
}
# usage: local %SIG = (%SIG, $lei->atfork_child_wq($wq));
sub atfork_child_wq {
my ($self, $wq) = @_;
- $self->{sock} //= $wq->{0};
- $self->{$_} //= $wq->{$_} for (0..2);
+ return () if $self->{0}; # did not fork
+ $self->{$_} = $wq->{$_} for (0..2);
+ $self->{sock} = $wq->{3} // die 'BUG: no {sock}'; # may be undef
my $oldpipe = $SIG{PIPE};
%PATH2CFG = ();
@TO_CLOSE_ATFORK_CHILD = ();
my $env = delete $self->{env}; # env is inherited at fork
my $ret = bless { %$self }, ref($self);
$self->{env} = $env;
- delete @$ret{qw(-lei_store cfg)};
- my $in = delete $ret->{0};
- ($ret, delete($ret->{sock}) // $in, delete @$ret{1, 2});
+ delete @$ret{qw(-lei_store cfg pgr)};
+ ($ret, delete @$ret{qw(0 1 2 sock)});
} else {
- ($self, ($self->{sock} // $self->{0}), @$self{1, 2});
+ ($self, @$self{qw(0 1 2 sock)});
}
}
$new_env{MORE} = 'FRX' if $^O eq 'freebsd';
pipe(my ($r, $wpager)) or return warn "pipe: $!";
my $rdr = { 0 => $r, 1 => $self->{1}, 2 => $self->{2} };
- my $pid;
+ my $pgr = [ undef, @$rdr{1, 2} ];
if (my $sock = $self->{sock}) { # lei(1) process runs it
delete @new_env{keys %$env}; # only set iff unset
my $buf = "exec 1\0".$pager;
my $fds = [ map { fileno($_) } @$rdr{0..2} ];
$send_cmd->($sock, $fds, $buf .= "\n", 0);
} else {
- $pid = spawn([$pager], $env, $rdr);
+ $pgr->[0] = spawn([$pager], $env, $rdr);
}
$self->{1} = $wpager;
$self->{2} = $wpager if -t $self->{2};
$env->{GIT_PAGER_IN_USE} = 'true'; # we may spawn git
- [ $pid, @$rdr{1, 2} ];
+ $self->{pgr} = $pgr;
+}
+
+sub stop_pager {
+ my ($self) = @_;
+ my $pgr = delete($self->{pgr}) or return;
+ my $pid = $pgr->[0];
+ close $self->{1};
+ # {2} may not be redirected
+ $self->{1} = $pgr->[1];
+ $self->{2} = $pgr->[2];
+ dwaitpid($pid, undef, $self->{sock}) if $pid;
}
sub accept_dispatch { # Listener {post_accept} callback
my $dev_ino_expect = pack('dd', $st[0], $st[1]); # dev+ino
pipe(my ($eof_r, $eof_w)) or die "pipe: $!";
local $oldset = PublicInbox::DS::block_signals();
- if ($nfd == 1) {
- require PublicInbox::CmdIPC1;
- $send_cmd = PublicInbox::CmdIPC1->can('send_cmd1');
- $recv_cmd = PublicInbox::CmdIPC1->can('recv_cmd1');
- } elsif ($nfd == 4) {
+ if ($nfd == 4) {
$send_cmd = PublicInbox::Spawn->can('send_cmd4');
$recv_cmd = PublicInbox::Spawn->can('recv_cmd4') // do {
require PublicInbox::CmdIPC4;
};
}
$recv_cmd or die <<"";
-(Socket::MsgHdr || IO::FDPass || Inline::C) missing/unconfigured (nfd=$nfd);
+(Socket::MsgHdr || Inline::C) missing/unconfigured (nfd=$nfd);
require PublicInbox::Listener;
require PublicInbox::EOFpipe;
exit($exit_code // 0);
}
-# for users w/o IO::FDPass
+# for users w/o Socket::Msghdr
sub oneshot {
my ($main_pkg) = @_;
my $exit = $main_pkg->can('exit'); # caller may override exit()
local $quit = $exit if $exit;
local %PATH2CFG;
umask(077) // die("umask(077): $!");
- dispatch((bless {
- 0 => *STDIN{GLOB},
- 1 => *STDOUT{GLOB},
- 2 => *STDERR{GLOB},
- env => \%ENV
- }, __PACKAGE__), @ARGV);
+ local $SIG{PIPE} = sub { die(bless(\"$_[0]", 'PublicInbox::SIGPIPE')) };
+ eval {
+ my $self = bless {
+ 0 => *STDIN{GLOB},
+ 1 => *STDOUT{GLOB},
+ 2 => *STDERR{GLOB},
+ env => \%ENV
+ }, __PACKAGE__;
+ dispatch($self, @ARGV);
+ };
+ die $@ if $@ && ref($@) ne 'PublicInbox::SIGPIPE';
}
# ensures stdout hits the FS before sock disconnects so a client
sub DESTROY {
my ($self) = @_;
$self->{1}->autoflush(1);
+ stop_pager($self);
}
1;
if ($self->{sock}) {
$self->atfork_prepare_wq($lxs);
$lxs->wq_workers_start('lei_xsearch', $j, $self->oldset)
- // $self->wq_workers($j);
+ // $lxs->wq_workers($j);
}
unshift(@srcs, $sto->search) if $opt->{'local'};
my $out = $opt->{output} // '-';
$out = 'json:/dev/stdout' if $out eq '-';
my $isatty = -t $self->{1};
# no forking workers after this
- my $pid_old12 = $self->start_pager if $isatty;
+ $self->start_pager if $isatty;
my $json = substr($out, 0, 5) eq 'json:' ?
ref(PublicInbox::Config->json)->new : undef;
if ($json) {
# my $wcb = PublicInbox::LeiToMail->write_cb($out, $self);
$self->{mset_opt} = \%mset_opt;
$lxs->do_query($self, \@srcs);
- if ($pid_old12) { # [ pid, stdout, stderr ]
- my $pid = $pid_old12->[0];
- $self->{$_} = $pid_old12->[$_] for (1, 2);
- dwaitpid($pid, undef, $self->{sock}) if $pid;
- }
}
1;
$self->SUPER::ipc_atfork_child; # PublicInbox::IPC
}
+sub ipc_atfork_prepare {
+ my ($self) = @_;
+ $self->wq_set_recv_modes(qw[<&= >&= >&= +<&=]);
+ $self->SUPER::ipc_atfork_prepare; # PublicInbox::IPC
+}
+
1;
#include <sys/socket.h>
#if defined(CMSG_SPACE) && defined(CMSG_LEN)
-#define SEND_FD_CAPA 3
+#define SEND_FD_CAPA 4
#define SEND_FD_SPACE (SEND_FD_CAPA * sizeof(int))
union my_cmsg {
struct cmsghdr hdr;
my $narg = 4;
my $recv_cmd = PublicInbox::CmdIPC4->can('recv_cmd4');
my $send_cmd = PublicInbox::CmdIPC4->can('send_cmd4') // do {
- require PublicInbox::CmdIPC1; # 2nd choice
- $narg = 1;
- $recv_cmd = PublicInbox::CmdIPC1->can('recv_cmd1');
- PublicInbox::CmdIPC1->can('send_cmd1');
-} // do {
require PublicInbox::Spawn; # takes ~50ms even if built *sigh*
$narg = 4;
$recv_cmd = PublicInbox::Spawn->can('recv_cmd4');
$pwd = $cwd;
}
1;
-}) { # (Socket::MsgHdr|IO::FDPass|Inline::C), $sock, $pwd are all available:
+}) { # (Socket::MsgHdr|Inline::C), $sock, $pwd are all available:
local $ENV{PWD} = $pwd;
my $buf = join("\0", scalar(@ARGV), @ARGV);
while (my ($k, $v) = each %ENV) { $buf .= "\0$k=$v" }
die $buf;
}
}
-} else { # for systems lacking Socket::MsgHdr, IO::FDPass or Inline::C
+} else { # for systems lacking Socket::MsgHdr or Inline::C
warn $@ if $@;
require PublicInbox::LEI;
PublicInbox::LEI::oneshot(__PACKAGE__);
}
}
-SKIP: {
- require_mods('IO::FDPass', 13);
- require_ok 'PublicInbox::CmdIPC1';
- $send = PublicInbox::CmdIPC1->can('send_cmd1');
- $recv = PublicInbox::CmdIPC1->can('recv_cmd1');
- $do_test->(SOCK_STREAM, 0, 'IO::FDPass stream');
- $do_test->($SOCK_SEQPACKET, MSG_EOR, 'IO::FDPass seqpacket');
-}
-
done_testing;
use Test::More;
use PublicInbox::TestCommon;
use Fcntl qw(SEEK_SET);
+require_mods(qw(Storable||Sereal));
require_ok 'PublicInbox::IPC';
state $once = eval <<'';
package PublicInbox::IPC;
};
$test->('local');
-SKIP: {
- require_mods(qw(Storable||Sereal), 16);
+{
my $pid = $ipc->ipc_worker_spawn('test worker');
ok($pid > 0 && kill(0, $pid), 'worker spawned and running');
defined($pid) or BAIL_OUT 'no spawn, no test';
$ipc->ipc_worker_stop; # idempotent
# work queues
-$ipc->{wq_open_modes} = [qw( >&= >&= >&= )];
+$ipc->wq_set_recv_modes(qw( >&= >&= >&= ));
pipe(my ($ra, $wa)) or BAIL_OUT $!;
pipe(my ($rb, $wb)) or BAIL_OUT $!;
pipe(my ($rc, $wc)) or BAIL_OUT $!;
# wq_do works across fork (siblings can feed)
SKIP: {
- skip 'Socket::MsgHdr, IO::FDPass, Inline::C missing', 7 if !$ppids[0];
+ skip 'Socket::MsgHdr or Inline::C missing', 3 if !$ppids[0];
is_deeply(\@ppids, [$$, undef, undef],
'parent pid returned in wq_workers_start');
my $pid = fork // BAIL_OUT $!;
}
$ipc->wq_close;
-seek($warn, 0, SEEK_SET) or BAIL_OUT;
-my @warn = <$warn>;
-is(scalar(@warn), 3, 'warned 3 times');
-like($warn[0], qr/ wq_do: /, '1st warned from wq_do');
-like($warn[1], qr/ wq_worker: /, '2nd warned from wq_worker');
-is($warn[2], $warn[1], 'worker did not die');
-
-$SIG{__WARN__} = 'DEFAULT';
-is($ipc->wq_workers_start('wq', 1), $$, 'workers started again');
-is($ipc->wq_workers, 1, '1 worker started');
SKIP: {
- $ipc->WQ_MAX_WORKERS > 1 or
- skip 'Inline::C or Socket::MsgHdr not available', 4;
- $ipc->wq_worker_incr;
- is($ipc->wq_workers, 2, 'worker count bumped');
- $ipc->wq_worker_decr;
- $ipc->wq_worker_decr_wait(10);
- is($ipc->wq_workers, 1, 'worker count lowered');
- is($ipc->wq_workers(2), 2, 'worker count set');
- is($ipc->wq_workers, 2, 'worker count stayed set');
+ skip 'Socket::MsgHdr or Inline::C missing', 11 if !$ppids[0];
+ seek($warn, 0, SEEK_SET) or BAIL_OUT;
+ my @warn = <$warn>;
+ is(scalar(@warn), 3, 'warned 3 times');
+ like($warn[0], qr/ wq_do: /, '1st warned from wq_do');
+ like($warn[1], qr/ wq_worker: /, '2nd warned from wq_worker');
+ is($warn[2], $warn[1], 'worker did not die');
+
+ $SIG{__WARN__} = 'DEFAULT';
+ is($ipc->wq_workers_start('wq', 1), $$, 'workers started again');
+ is($ipc->wq_workers, 1, '1 worker started');
+ SKIP: {
+ $ipc->WQ_MAX_WORKERS > 1 or
+ skip 'Inline::C or Socket::MsgHdr not available', 4;
+ $ipc->wq_worker_incr;
+ is($ipc->wq_workers, 2, 'worker count bumped');
+ $ipc->wq_worker_decr;
+ $ipc->wq_worker_decr_wait(10);
+ is($ipc->wq_workers, 1, 'worker count lowered');
+ is($ipc->wq_workers(2), 2, 'worker count set');
+ is($ipc->wq_workers, 2, 'worker count stayed set');
+ }
+ $ipc->wq_close;
+ is($ipc->wq_workers, undef, 'workers undef after close');
}
-$ipc->wq_close;
-is($ipc->wq_workers, undef, 'workers undef after close');
done_testing;
SKIP: { # real socket
require_mods(qw(Cwd), my $nr = 105);
- my $nfd = eval { require Socket::MsgHdr; 4 } //
- eval { require IO::FDPass; 1 } // do {
+ my $nfd = eval { require Socket::MsgHdr; 4 } // do {
require PublicInbox::Spawn;
PublicInbox::Spawn->can('send_cmd4') ? 4 : undef;
} //
- skip 'Socket::MsgHdr, IO::FDPass or Inline::C missing or unconfigured',
- $nr;
+ skip 'Socket::MsgHdr or Inline::C missing or unconfigured', $nr;
local $ENV{XDG_RUNTIME_DIR} = "$home/xdg_run";
my $sock = "$ENV{XDG_RUNTIME_DIR}/lei/$nfd.sock";