X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FQspawn.pm;h=651fa39096d7144beb7d9bb6030fcb67c9873df0;hb=4d44a2ca68ba1b88ee5245df7b74492f9e41db6d;hp=9aede103521b47d3684032ca29ac6ace7f0a1747;hpb=fe86054ee1e2d93b19762a34ed1aaebe43635ff0;p=public-inbox.git diff --git a/lib/PublicInbox/Qspawn.pm b/lib/PublicInbox/Qspawn.pm index 9aede103..651fa390 100644 --- a/lib/PublicInbox/Qspawn.pm +++ b/lib/PublicInbox/Qspawn.pm @@ -29,6 +29,9 @@ use warnings; 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). @@ -41,11 +44,11 @@ sub new ($$$;) { } 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; @@ -53,12 +56,17 @@ sub _do_spawn { } ($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 ($) { @@ -70,43 +78,75 @@ 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 ]; } } @@ -116,31 +156,37 @@ sub _psgi_finish ($$) { # 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; + 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... - ($rpipe) = @_; + $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; } @@ -186,23 +232,41 @@ sub psgi_return { 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) { @@ -213,6 +277,7 @@ sub psgi_return { } $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); @@ -222,6 +287,10 @@ sub psgi_return { $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... @@ -229,6 +298,7 @@ sub psgi_return { 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; } @@ -291,21 +361,4 @@ sub setup_rlimit { } } -# 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;