lib/PublicInbox/IMAPD.pm
lib/PublicInbox/IMAPTracker.pm
lib/PublicInbox/IMAPsearchqp.pm
+lib/PublicInbox/IO.pm
lib/PublicInbox/IPC.pm
lib/PublicInbox/IdxStack.pm
lib/PublicInbox/Import.pm
lib/PublicInbox/POP3.pm
lib/PublicInbox/POP3D.pm
lib/PublicInbox/PktOp.pm
-lib/PublicInbox/ProcessIO.pm
-lib/PublicInbox/ProcessIONBF.pm
lib/PublicInbox/Qspawn.pm
lib/PublicInbox/Reply.pm
lib/PublicInbox/RepoAtom.pm
use PublicInbox::Spawn qw(spawn);
use Socket qw(AF_UNIX SOCK_STREAM);
use PublicInbox::Syscall qw(EPOLLIN);
-use PublicInbox::ProcessIO;
+use PublicInbox::IO;
use autodie qw(socketpair);
# fields:
$opt->{0} = $opt->{1} = $s2;
my $cmd = [$^X, $^W ? ('-w') : (),
qw[-MPublicInbox::Gcf2 -e PublicInbox::Gcf2::loop]];
- my $pid = spawn($cmd, $env, $opt);
- my $sock = PublicInbox::ProcessIO->maybe_new($pid, $s1);
+ PublicInbox::IO::attach_pid($s1, spawn($cmd, $env, $opt));
$self->{inflight} = [];
$self->{epwatch} = \undef; # for Git->cleanup
- $self->SUPER::new($sock, EPOLLIN);
+ $self->SUPER::new($s1, EPOLLIN);
}
sub gcf2_async ($$$;$) {
use File::Glob qw(bsd_glob GLOB_NOSORT);
use File::Spec ();
use PublicInbox::Spawn qw(spawn popen_rd run_qx which);
-use PublicInbox::ProcessIONBF;
+use PublicInbox::IO;
use PublicInbox::Tmpfile;
use IO::Poll qw(POLLIN);
use Carp qw(croak carp);
my ($self, $batch, $err_c) = @_;
$self->{sock} and Carp::confess('BUG: {sock} exists');
socketpair(my $s1, my $s2, AF_UNIX, SOCK_STREAM, 0);
+ $s1->blocking(0);
my $opt = { pgid => 0, 0 => $s2, 1 => $s2 };
my $gd = $self->{git_dir};
if ($gd =~ s!/([^/]+/[^/]+)\z!/!) {
$self->fail("tmpfile($id): $!");
}
my $pid = spawn(\@cmd, undef, $opt);
- $self->{sock} = PublicInbox::ProcessIONBF->new($pid, $s1);
+ $self->{sock} = PublicInbox::IO::attach_pid($s1, $pid);
}
sub poll_in ($) { IO::Poll::_poll($RDTIMEO, fileno($_[0]), my $ev = POLLIN) }
my $ret = 0;
for my $obj ($self, ($self->{ck} // ())) {
my $sock = $obj->{sock} // next;
- my PublicInbox::ProcessIONBF $p = tied *$sock; # ProcessIONBF
- my $pid = $p->{pid} // next;
+ my $pid = $sock->attached_pid // next;
open my $fh, '<', "/proc/$pid/maps" or return cleanup($self, 1);
while (<$fh>) {
# n.b. we do not restart for unlinked multi-pack-index
--- /dev/null
+# Copyright (C) all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# supports reaping of children tied to a pipe or socket
+package PublicInbox::IO;
+use v5.12;
+use parent qw(IO::Handle);
+use PublicInbox::DS qw(awaitpid);
+
+# TODO: this can probably be the new home for read_all, try_cat
+# and maybe even buffered read/readline...
+
+sub waitcb { # awaitpid callback
+ my ($pid, $errref, $cb, @args) = @_;
+ $$errref = $?; # sets .cerr for _close
+ $cb->($pid, @args) if $cb;
+}
+
+sub attach_pid ($$;@) {
+ my ($io, $pid, @cb_arg) = @_;
+ bless $io, __PACKAGE__;
+ # we share $err (and not $self) with awaitpid to avoid a ref cycle
+ ${*$io}{pi_io_reap} = [ $$, $pid, \(my $err) ];
+ awaitpid($pid, \&waitcb, \$err, @cb_arg);
+ $io;
+}
+
+sub attached_pid {
+ my ($io) = @_;
+ ${${*$io}{pi_io_reap} // []}[1];
+}
+
+# caller cares about error result if they call close explicitly
+# reap->[2] may be set before this is called via waitcb
+sub close {
+ my ($io) = @_;
+ my $ret = $io->SUPER::close;
+ my $reap = delete ${*$io}{pi_io_reap};
+ return $ret unless $reap && $reap->[0] == $$;
+ ${$reap->[2]} // (my $w = awaitpid($reap->[1])); # sets [2]
+ ($? = ${$reap->[2]}) ? '' : $ret;
+}
+
+sub DESTROY {
+ my ($io) = @_;
+ my $reap = delete ${*$io}{pi_io_reap};
+ if ($reap && $reap->[0] == $$) {
+ $io->SUPER::close;
+ awaitpid($reap->[1]);
+ }
+ $io->SUPER::DESTROY;
+}
+
+1;
use PublicInbox::ContentHash qw(content_digest);
use PublicInbox::MDA;
use PublicInbox::Eml;
-use PublicInbox::ProcessIO;
+use PublicInbox::IO;
use POSIX qw(strftime);
use autodie qw(read close socketpair);
use Carp qw(croak);
--quiet --done --date-format=raw) ];
my $pid = spawn($gfi, undef, { 0 => $s2, 1 => $s2 });
$self->{nchg} = 0;
- $self->{io} = PublicInbox::ProcessIO->maybe_new($pid, $io);
+ $self->{io} = PublicInbox::IO::attach_pid($io, $pid);
};
if ($@) {
$self->lock_release;
if ($@) { # regular file (but not w/ select|IO::Poll backends)
$self->{-need_rq} = 1;
$self->requeue;
- } elsif (do { no warnings 'unopened'; !stat($in) }) { # ProcessIONBF
} elsif (-p _ || -S _) { # O_NONBLOCK for sockets and pipes
$in->blocking(0);
} elsif (-t $in) { # isatty(3) can't use `_' stat cache
use v5.10.1;
use parent qw(PublicInbox::IPC);
use PublicInbox::Eml;
-use PublicInbox::ProcessIO;
+use PublicInbox::IO;
use PublicInbox::Spawn qw(spawn);
use IO::Handle; # ->autoflush
use Fcntl qw(SEEK_SET SEEK_END O_CREAT O_EXCL O_WRONLY);
my $cmd = PublicInbox::MboxReader::zsfx2cmd($zsfx, undef, $lei);
my ($r, $w) = @{delete $lei->{zpipe}};
my $rdr = { 0 => $r, 1 => $lei->{1}, 2 => $lei->{2}, pgid => 0 };
- my $pid = spawn($cmd, undef, $rdr);
- $lei->{1} = PublicInbox::ProcessIO->maybe_new($pid, $w,
+ $lei->{1} = PublicInbox::IO::attach_pid($w, spawn($cmd, undef, $rdr),
\&reap_compress, $lei, $cmd, $lei->{1});
}
+++ /dev/null
-# Copyright (C) all contributors <meta@public-inbox.org>
-# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
-
-# a tied handle for auto reaping of children tied to a pipe or socket,
-# see perltie(1) for details.
-package PublicInbox::ProcessIO;
-use v5.12;
-use PublicInbox::DS qw(awaitpid);
-use Symbol qw(gensym);
-use bytes qw(length);
-
-sub maybe_new {
- my ($cls, $pid, $fh, @cb_arg) = @_;
- return ($fh, $pid) if wantarray;
- my $s = gensym;
- tie *$s, $cls, $pid, $fh, @cb_arg;
- $s;
-}
-
-sub waitcb { # awaitpid callback
- my ($pid, $err_ref, $cb, @args) = @_;
- $$err_ref = $?; # sets >{pp_chld_err} for _close
- $cb->($pid, @args) if $cb;
-}
-
-sub TIEHANDLE {
- my ($cls, $pid, $fh, @cb_arg) = @_;
- my $self = bless { pid => $pid, fh => $fh, ppid => $$ }, $cls;
- # we share $err (and not $self) with awaitpid to avoid a ref cycle
- $self->{pp_chld_err} = \(my $err);
- awaitpid($pid, \&waitcb, \$err, @cb_arg);
- $self;
-}
-
-# for IO::Uncompress::Gunzip and PublicInbox::LeiRediff
-sub BINMODE { @_ == 1 ? binmode($_[0]->{fh}) : binmode($_[0]->{fh}, $_[1]) }
-
-sub READ { read($_[0]->{fh}, $_[1], $_[2], $_[3] || 0) }
-
-sub READLINE { readline($_[0]->{fh}) }
-
-sub WRITE { syswrite($_[0]->{fh}, $_[1], $_[2] // length($_[1]), $_[3] // 0) }
-
-sub PRINT { print { $_[0]->{fh} } @_[1..$#_] }
-
-sub FILENO { fileno($_[0]->{fh}) }
-
-sub _close ($;$) {
- my ($self, $wait) = @_;
- my ($fh, $pid) = delete(@$self{qw(fh pid)});
- my $ret = (defined($fh) && $wait) ? close($fh) : ($fh = '');
- return $ret unless defined($pid) && $self->{ppid} == $$;
- if ($wait) { # caller cares about the exit status:
- # synchronous wait via defined(wantarray) on awaitpid:
- defined(${$self->{pp_chld_err}}) or $wait = awaitpid($pid);
- ($? = ${$self->{pp_chld_err}}) and $ret = '';
- } else {
- awaitpid($pid); # depends on $in_loop or not
- }
- $ret;
-}
-
-# if caller uses close(), assume they want to check $? immediately so
-# we'll waitpid() synchronously. n.b. wantarray doesn't seem to
-# propagate `undef' down to tied methods, otherwise I'd rely on that.
-sub CLOSE { _close($_[0], 1) }
-
-# if relying on DESTROY, assume the caller doesn't care about $? and
-# we can let the event loop call waitpid() whenever it gets SIGCHLD
-sub DESTROY {
- _close($_[0]);
- undef;
-}
-
-1;
+++ /dev/null
-# Copyright (C) all contributors <meta@public-inbox.org>
-# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
-
-# used to support unbuffered partial reads
-package PublicInbox::ProcessIONBF;
-use v5.12;
-use parent qw(PublicInbox::ProcessIO);
-use IO::Handle; # ->blocking
-
-sub new {
- my ($cls, $pid, $fh, @cb_arg) = @_;
- $fh->blocking(0) // die "$fh->blocking(0): $!";
- my $io = $cls->SUPER::maybe_new($pid, $fh, @cb_arg);
-}
-
-sub replace {
- my ($cls, $orig) = @_;
- my $pio = tied *$orig; # ProcessIO
- $pio->{fh}->blocking(0) // die "$pio->{fh}->blocking(0): $!";
- bless $pio, $cls;
-}
-
-sub READ { sysread($_[0]->{fh}, $_[1], $_[2], $_[3] // 0) }
-
-1;
# we can safely finalize if pipe was closed before, or if
# {_err} is defined by waitpid_err. Deleting {rpipe} will
- # trigger PublicInbox::ProcessIO::DESTROY -> waitpid_err,
+ # trigger PublicInbox::IO::DESTROY -> waitpid_err,
# but it may not fire right away if inside the event loop.
my $closed_before = !delete($self->{rpipe});
finalize($self) if $closed_before || defined($self->{_err});
sub _yield_start { # may run later, much later...
my ($self) = @_;
if ($self->{psgi_env}->{'pi-httpd.async'}) {
- require PublicInbox::ProcessIONBF;
my $rpipe = $self->{rpipe};
- PublicInbox::ProcessIONBF->replace($rpipe);
+ $rpipe->blocking(0);
PublicInbox::InputPipe::consume($rpipe, \&ipipe_cb, $self);
} else {
require PublicInbox::GetlineResponse;
use Fcntl qw(SEEK_SET);
use IO::Handle ();
use Carp qw(croak);
-use PublicInbox::ProcessIO;
+use PublicInbox::IO;
our @EXPORT_OK = qw(which spawn popen_rd popen_wr run_die run_wait run_qx);
our @RLIMITS = qw(RLIMIT_CPU RLIMIT_CORE RLIMIT_DATA);
use autodie qw(open pipe seek sysseek truncate);
my ($cmd, $env, $opt, @cb_arg) = @_;
pipe(my $r, local $opt->{1});
my $pid = spawn($cmd, $env, $opt);
- PublicInbox::ProcessIO->maybe_new($pid, $r, @cb_arg);
+ wantarray ? ($r, $pid) : PublicInbox::IO::attach_pid($r, $pid, @cb_arg)
}
sub popen_wr {
pipe(local $opt->{0}, my $w);
$w->autoflush(1);
my $pid = spawn($cmd, $env, $opt);
- PublicInbox::ProcessIO->maybe_new($pid, $w, @cb_arg)
+ wantarray ? ($w, $pid) : PublicInbox::IO::attach_pid($w, $pid, @cb_arg)
}
sub read_out_err ($) {
{
my $fh = popen_rd([qw(echo hello)]);
- ok(fileno($fh) >= 0, 'tied fileno works');
+ ok(fileno($fh) >= 0, 'fileno works');
my $l = <$fh>;
- is($l, "hello\n", 'tied readline works');
+ is($l, "hello\n", 'readline works');
$l = <$fh>;
- ok(!$l, 'tied readline works for EOF');
+ ok(!$l, 'readline works for EOF');
}
{
my $fh = popen_rd([qw(printf foo\nbar)]);
- ok(fileno($fh) >= 0, 'tied fileno works');
- my $tfh = (tied *$fh)->{fh};
- is($tfh->blocking(0), 1, '->blocking was true');
- is($tfh->blocking, 0, '->blocking is false');
- is($tfh->blocking(1), 0, '->blocking was true');
- is($tfh->blocking, 1, '->blocking is true');
+ ok(fileno($fh) >= 0, 'fileno works');
+ is($fh->blocking(0), 1, '->blocking was true');
+ is($fh->blocking, 0, '->blocking is false');
+ is($fh->blocking(1), 0, '->blocking was true');
+ is($fh->blocking, 1, '->blocking is true');
my @line = <$fh>;
is_deeply(\@line, [ "foo\n", 'bar' ], 'wantarray works on readline');
}
{
my $fh = popen_rd([qw(echo hello)]);
+ like($fh->attached_pid, qr/\A[0-9]+\z/, 'have a PID');
my $buf;
is(sysread($fh, $buf, 6), 6, 'sysread got 6 bytes');
- is($buf, "hello\n", 'tied gets works');
+ is($buf, "hello\n", 'sysread works');
is(sysread($fh, $buf, 6), 0, 'sysread got EOF');
$? = 1;
ok($fh->close, 'close succeeds');
is($?, 0, '$? set properly');
+ is($fh->attached_pid, undef, 'attached_pid cleared after close');
}
{
$fh = popen_rd(['true'], undef, undef, sub { @c = caller });
undef $fh; # ->DESTROY
ok(scalar(@c), 'callback fired by ->DESTROY');
- ok(grep(!m[/PublicInbox/ProcessIO\.pm\z], @c),
- 'callback not invoked by ProcessIO');
+ ok(grep(!m[/PublicInbox/IO\.pm\z], @c),
+ 'callback not invoked by PublicInbox::IO');
}
{ # children don't wait on siblings
my @arg;
my $fh = popen_rd(['cat'], undef, { 0 => $r },
sub { @arg = @_; warn "x=$$\n" }, 'hi');
- my $pp = tied *$fh;
my $pid = fork // BAIL_OUT $!;
local $SIG{__WARN__} = sub { _exit(1) };
if ($pid == 0) {
use IO::Handle; # ->autoflush
use PublicInbox::TestCommon;
use PublicInbox::Spawn;
-use PublicInbox::DS; # already loaded by Spawn via ProcessIO
+use PublicInbox::DS; # already loaded by Spawn via PublicInbox::IO
use Getopt::Long qw(:config gnu_getopt no_ignore_case auto_abbrev);
use Errno qw(EINTR);
use Fcntl qw(:seek);
projects / thirdparty / public-inbox.git / commitdiff
