use PublicInbox::Spawn qw(popen_rd);
require Plack::Util;
+# n.b.: we get EAGAIN with public-inbox-httpd, and EINTR on other PSGI servers
+use Errno qw(EAGAIN EINTR);
+
my $def_limiter;
# declares a command to spawn (but does not spawn it).
}
sub _do_spawn {
- my ($self, $cb) = @_;
+ my ($self, $start_cb, $limiter) = @_;
my $err;
my ($cmd, $env, $opts) = @{$self->{args}};
my %opts = %{$opts || {}};
- my $limiter = $self->{limiter};
+ $self->{limiter} = $limiter;
foreach my $k (PublicInbox::Spawn::RLIMITS()) {
if (defined(my $rlimit = $limiter->{$k})) {
$opts{$k} = $rlimit;
}
($self->{rpipe}, $self->{pid}) = popen_rd($cmd, $env, \%opts);
+
+ # drop any IO handles opt was holding open via $opt->{hold}
+ # No need to hold onto the descriptor once the child process has it.
+ $self->{args} = $cmd; # keep this around for logging
+
if (defined $self->{pid}) {
$limiter->{running}++;
} else {
$self->{err} = $!;
}
- $cb->($self->{rpipe});
+ $start_cb->($self->{rpipe});
}
sub child_err ($) {
$msg;
}
-sub finish ($) {
- my ($self) = @_;
+sub log_err ($$) {
+ my ($env, $msg) = @_;
+ $env->{'psgi.errors'}->print($msg, "\n");
+}
+
+# callback for dwaitpid
+sub waitpid_err ($$) {
+ my ($self, $pid) = @_;
+ my $xpid = delete $self->{pid};
+ my $err;
+ if ($pid > 0) { # success!
+ $err = child_err($?);
+ } elsif ($pid < 0) { # ??? does this happen in our case?
+ $err = "W: waitpid($xpid, 0) => $pid: $!";
+ } # else should not be called with pid == 0
+
+ my $env = delete $self->{env};
+
+ # done, spawn whatever's in the queue
my $limiter = $self->{limiter};
- my $running;
- if (delete $self->{rpipe}) {
- my $pid = delete $self->{pid};
- $self->{err} = $pid == waitpid($pid, 0) ? child_err($?) :
- "PID:$pid still running?";
- $running = --$limiter->{running};
+ my $running = --$limiter->{running};
+
+ if ($running < $limiter->{max}) {
+ if (my $next = shift(@{$limiter->{run_queue}})) {
+ _do_spawn(@$next, $limiter);
+ }
}
- # limiter->{max} may change dynamically
- if (($running || $limiter->{running}) < $limiter->{max}) {
- if (my $next = shift @{$limiter->{run_queue}}) {
- _do_spawn(@$next);
+ if ($err) {
+ $self->{err} = $err;
+ if ($env && !$env->{'qspawn.quiet'}) {
+ log_err($env, join(' ', @{$self->{args}}) . ": $err");
}
}
- $self->{err};
+ if (my $fin_cb = delete $self->{fin_cb}) {
+ eval { $fin_cb->() }
+ }
}
-sub start {
- my ($self, $limiter, $cb) = @_;
- $self->{limiter} = $limiter;
+sub do_waitpid ($;$$) {
+ my ($self, $env, $fin_cb) = @_;
+ my $pid = $self->{pid};
+ $self->{env} = $env;
+ $self->{fin_cb} = $fin_cb;
+ # PublicInbox::DS may not be loaded
+ eval { PublicInbox::DS::dwaitpid($pid, \&waitpid_err, $self) };
+ # done if we're running in PublicInbox::DS::EventLoop
+ if ($@) {
+ # non public-inbox-{httpd,nntpd} callers may block:
+ my $ret = waitpid($pid, 0);
+ waitpid_err($self, $ret);
+ }
+}
- if ($limiter->{running} < $limiter->{max}) {
- _do_spawn($self, $cb);
- } else {
- push @{$limiter->{run_queue}}, [ $self, $cb ];
+sub finish ($;$$) {
+ my ($self, $env, $fin_cb) = @_;
+ if (delete $self->{rpipe}) {
+ do_waitpid($self, $env, $fin_cb);
+ } elsif ($fin_cb) {
+ eval { $fin_cb->() };
}
}
-sub _psgi_finish ($$) {
- my ($self, $env) = @_;
- my $err = $self->finish;
- if ($err && !$env->{'qspawn.quiet'}) {
- $err = join(' ', @{$self->{args}->[0]}).": $err\n";
- $env->{'psgi.errors'}->print($err);
+sub start {
+ my ($self, $limiter, $start_cb) = @_;
+ if ($limiter->{running} < $limiter->{max}) {
+ _do_spawn($self, $start_cb, $limiter);
+ } else {
+ push @{$limiter->{run_queue}}, [ $self, $start_cb ];
}
}
# and safe to slurp.
sub psgi_qx {
my ($self, $env, $limiter, $qx_cb) = @_;
- my $qx = PublicInbox::Qspawn::Qx->new;
+ my $scalar = '';
+ open(my $qx, '+>', \$scalar) or die; # PerlIO::scalar
my $end = sub {
- _psgi_finish($self, $env);
- eval { $qx_cb->($qx) };
+ my $err = $_[0]; # $!
+ log_err($env, "psgi_qx: $err") if defined($err);
+ finish($self, $env, sub { $qx_cb->(\$scalar) });
$qx = undef;
};
my $rpipe; # comes from popen_rd
my $async = $env->{'pi-httpd.async'};
my $cb = sub {
- my $r = sysread($rpipe, my $buf, 8192);
+ my ($r, $buf);
+reread:
+ $r = sysread($rpipe, $buf, 65536);
if ($async) {
$async->async_pass($env->{'psgix.io'}, $qx, \$buf);
} elsif (defined $r) {
$r ? $qx->write($buf) : $end->();
} else {
- return if $!{EAGAIN} || $!{EINTR}; # loop again
- $end->();
+ return if $! == EAGAIN; # try again when notified
+ goto reread if $! == EINTR;
+ $end->($!);
}
};
$limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
- $self->start($limiter, sub { # may run later, much later...
+ $self->start($limiter, sub { # start_cb, may run later, much later...
($rpipe) = @_; # popen_rd result
if ($async) {
# PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
$async = $async->($rpipe, $cb, $end);
+ # $cb will call ->async_pass or ->close
} else { # generic PSGI
$cb->() while $qx;
}
my ($self, $env, $limiter, $parse_hdr) = @_;
my ($fh, $rpipe);
my $end = sub {
- _psgi_finish($self, $env);
+ my $err = $_[0]; # $!
+ log_err($env, "psgi_return: $err") if defined($err);
+ finish($self, $env);
$fh->close if $fh; # async-only
};
my $buf = '';
my $rd_hdr = sub {
- my $r = sysread($rpipe, $buf, 1024, length($buf));
- return if !defined($r) && ($!{EINTR} || $!{EAGAIN});
- $parse_hdr->($r, \$buf);
+ # typically used for reading CGI headers
+ # we must loop until EAGAIN for EPOLLET in HTTPD/Async.pm
+ # We also need to check EINTR for generic PSGI servers.
+ my $ret;
+ my $total_rd = 0;
+ do {
+ my $r = sysread($rpipe, $buf, 4096, length($buf));
+ if (defined($r)) {
+ $total_rd += $r;
+ $ret = $parse_hdr->($r ? $total_rd : 0, \$buf);
+ } else {
+ # caller should notify us when it's ready:
+ return if $! == EAGAIN;
+ next if $! == EINTR; # immediate retry
+ log_err($env, "error reading header: $!");
+ $ret = [ 500, [], [ "Internal error\n" ] ];
+ }
+ } until (defined $ret);
+ $ret;
};
- my $wcb = delete $env->{'qspawn.wcb'};
+ my $wcb = delete $env->{'qspawn.wcb'}; # or PSGI server supplies it
my $async = $env->{'pi-httpd.async'};
my $cb = sub {
my $r = $rd_hdr->() or return;
- $rd_hdr = undef;
+ $rd_hdr = undef; # done reading headers
my $filter = delete $env->{'qspawn.filter'};
if (scalar(@$r) == 3) { # error
if ($async) {
}
$wcb->($r);
} elsif ($async) {
+ # done reading headers, handoff to read body
$fh = $wcb->($r); # scalar @$r == 2
$fh = filter_fh($fh, $filter) if $filter;
$async->async_pass($env->{'psgix.io'}, $fh, \$buf);
$buf, $filter);
$wcb->($r);
}
+
+ # Workaround a leak under Perl 5.16.3 when combined with
+ # Plack::Middleware::Deflater:
+ $wcb = undef;
};
$limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
my $start_cb = sub { # may run later, much later...
if ($async) {
# PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
$async = $async->($rpipe, $cb, $end);
+ # $cb will call ->async_pass or ->close
} else { # generic PSGI
$cb->() while $rd_hdr;
}
}
}
-# captures everything into a buffer and executes a callback when done
-package PublicInbox::Qspawn::Qx;
-use strict;
-use warnings;
-
-sub new {
- my ($class) = @_;
- my $buf = '';
- bless \$buf, $class;
-}
-
-# called by PublicInbox::HTTPD::Async ($fh->write)
-sub write {
- ${$_[0]} .= $_[1];
- undef;
-}
-
1;