X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FQspawn.pm;h=651fa39096d7144beb7d9bb6030fcb67c9873df0;hb=4d44a2ca68ba1b88ee5245df7b74492f9e41db6d;hp=294bf0a4dde2a8446a3c1e14cce849e0260bda51;hpb=aea0a8d8bf9bf26898c811bf28f023d968c39de3;p=public-inbox.git diff --git a/lib/PublicInbox/Qspawn.pm b/lib/PublicInbox/Qspawn.pm index 294bf0a4..651fa390 100644 --- a/lib/PublicInbox/Qspawn.pm +++ b/lib/PublicInbox/Qspawn.pm @@ -27,7 +27,6 @@ package PublicInbox::Qspawn; use strict; use warnings; use PublicInbox::Spawn qw(popen_rd); -use POSIX qw(WNOHANG); require Plack::Util; # n.b.: we get EAGAIN with public-inbox-httpd, and EINTR on other PSGI servers @@ -45,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; @@ -57,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 ($) { @@ -74,6 +78,11 @@ sub child_err ($) { $msg; } +sub log_err ($$) { + my ($env, $msg) = @_; + $env->{'psgi.errors'}->print($msg, "\n"); +} + # callback for dwaitpid sub waitpid_err ($$) { my ($self, $pid) = @_; @@ -85,33 +94,36 @@ sub waitpid_err ($$) { $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 = --$limiter->{running}; - # limiter->{max} may change dynamically - if (($running || $limiter->{running}) < $limiter->{max}) { - if (my $next = shift @{$limiter->{run_queue}}) { - _do_spawn(@$next); + if ($running < $limiter->{max}) { + if (my $next = shift(@{$limiter->{run_queue}})) { + _do_spawn(@$next, $limiter); } } - return unless $err; - $self->{err} = $err; - my $env = $self->{env} or return; - if (!$env->{'qspawn.quiet'}) { - $err = join(' ', @{$self->{args}->[0]}).": $err\n"; - $env->{'psgi.errors'}->print($err); + if ($err) { + $self->{err} = $err; + if ($env && !$env->{'qspawn.quiet'}) { + log_err($env, join(' ', @{$self->{args}}) . ": $err"); + } + } + if (my $fin_cb = delete $self->{fin_cb}) { + eval { $fin_cb->() } } } -sub do_waitpid ($;$) { - my ($self, $env) = @_; +sub do_waitpid ($;$$) { + my ($self, $env, $fin_cb) = @_; my $pid = $self->{pid}; - eval { # PublicInbox::DS may not be loaded - PublicInbox::DS::dwaitpid($pid, \&waitpid_err, $self); - $self->{env} = $env; - }; + $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: @@ -120,30 +132,21 @@ sub do_waitpid ($;$) { } } -sub finish ($;$) { - my ($self, $env) = @_; +sub finish ($;$$) { + my ($self, $env, $fin_cb) = @_; if (delete $self->{rpipe}) { - do_waitpid($self, $env); - } - - # limiter->{max} may change dynamically - my $limiter = $self->{limiter}; - if ($limiter->{running} < $limiter->{max}) { - if (my $next = shift @{$limiter->{run_queue}}) { - _do_spawn(@$next); - } + do_waitpid($self, $env, $fin_cb); + } elsif ($fin_cb) { + eval { $fin_cb->() }; } - $self->{err}; # may be meaningless if non-blocking } sub start { - my ($self, $limiter, $cb) = @_; - $self->{limiter} = $limiter; - + my ($self, $limiter, $start_cb) = @_; if ($limiter->{running} < $limiter->{max}) { - _do_spawn($self, $cb); + _do_spawn($self, $start_cb, $limiter); } else { - push @{$limiter->{run_queue}}, [ $self, $cb ]; + push @{$limiter->{run_queue}}, [ $self, $start_cb ]; } } @@ -156,29 +159,34 @@ sub psgi_qx { my $scalar = ''; open(my $qx, '+>', \$scalar) or die; # PerlIO::scalar my $end = sub { - finish($self, $env); - eval { $qx_cb->(\$scalar) }; - $qx = $scalar = undef; + 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, 65536); + 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; } @@ -224,6 +232,8 @@ sub psgi_return { my ($self, $env, $limiter, $parse_hdr) = @_; my ($fh, $rpipe); my $end = sub { + my $err = $_[0]; # $! + log_err($env, "psgi_return: $err") if defined($err); finish($self, $env); $fh->close if $fh; # async-only }; @@ -234,19 +244,24 @@ sub psgi_return { # we must loop until EAGAIN for EPOLLET in HTTPD/Async.pm # We also need to check EINTR for generic PSGI servers. my $ret; - my $n = 0; + my $total_rd = 0; do { my $r = sysread($rpipe, $buf, 4096, length($buf)); - return if !defined($r) && $! == EAGAIN || $! == EINTR; - - # $r may be undef, here: - $n += $r if $r; - $ret = $parse_hdr->($r ? $n : $r, \$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 { @@ -272,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... @@ -279,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; }