# operate in. This can be useful to ensure smaller inboxes can
# be cloned while cloning of large inboxes is maxed out.
#
-# This does not depend on Danga::Socket or any other external
+# This does not depend on PublicInbox::DS or any other external
# scheduling mechanism, you just need to call start() and finish()
-# appropriately. However, public-inbox-httpd (which uses Danga::Socket)
+# appropriately. However, public-inbox-httpd (which uses PublicInbox::DS)
# will be able to schedule this based on readability of stdout from
# the spawned process. See GitHTTPBackend.pm and SolverGit.pm for
# usage examples. It does not depend on any form of threading.
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
+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 $err;
+ my ($cmd, $env, $opts) = @{$self->{args}};
+ my %opts = %{$opts || {}};
+ my $limiter = $self->{limiter};
+ foreach my $k (PublicInbox::Spawn::RLIMITS()) {
+ if (defined(my $rlimit = $limiter->{$k})) {
+ $opts{$k} = $rlimit;
+ }
+ }
- ($self->{rpipe}, $self->{pid}) = popen_rd(@{$self->{args}});
+ ($self->{rpipe}, $self->{pid}) = popen_rd($cmd, $env, \%opts);
if (defined $self->{pid}) {
- $self->{limiter}->{running}++;
+ $limiter->{running}++;
} else {
$self->{err} = $!;
}
$msg;
}
-sub finish ($) {
- my ($self) = @_;
+# 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
+
+ # done, spawn whatever's in the queue
my $limiter = $self->{limiter};
- my $running;
+ 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);
+ }
+ }
+
+ 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);
+ }
+}
+
+sub do_waitpid ($;$) {
+ my ($self, $env) = @_;
+ my $pid = $self->{pid};
+ eval { # PublicInbox::DS may not be loaded
+ PublicInbox::DS::dwaitpid($pid, \&waitpid_err, $self);
+ $self->{env} = $env;
+ };
+ # 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);
+ }
+}
+
+sub finish ($;$) {
+ my ($self, $env) = @_;
if (delete $self->{rpipe}) {
- my $pid = delete $self->{pid};
- $self->{err} = $pid == waitpid($pid, 0) ? child_err($?) :
- "PID:$pid still running?";
- $running = --$limiter->{running};
+ do_waitpid($self, $env);
}
# limiter->{max} may change dynamically
- if (($running || $limiter->{running}) < $limiter->{max}) {
+ my $limiter = $self->{limiter};
+ if ($limiter->{running} < $limiter->{max}) {
if (my $next = shift @{$limiter->{run_queue}}) {
_do_spawn(@$next);
}
}
- $self->{err};
+ $self->{err}; # may be meaningless if non-blocking
}
sub start {
}
}
-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);
- }
-}
-
# Similar to `backtick` or "qx" ("perldoc -f qx"), it calls $qx_cb with
# the stdout of the given command when done; but respects the given limiter
# $env is the PSGI env. As with ``/qx; only use this when output is small
my ($self, $env, $limiter, $qx_cb) = @_;
my $qx = PublicInbox::Qspawn::Qx->new;
my $end = sub {
- _psgi_finish($self, $env);
+ finish($self, $env);
eval { $qx_cb->($qx) };
$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 = sysread($rpipe, my $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
+ return if $! == EAGAIN || $! == EINTR; # loop again
$end->();
}
};
$limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
$self->start($limiter, sub { # may run later, much later...
- ($rpipe) = @_;
+ ($rpipe) = @_; # popen_rd result
if ($async) {
# PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
$async = $async->($rpipe, $cb, $end);
my ($self, $env, $limiter, $parse_hdr) = @_;
my ($fh, $rpipe);
my $end = sub {
- _psgi_finish($self, $env);
+ 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);
+ # 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;
+ 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);
+ } until (defined $ret);
+ $ret;
};
my $wcb = delete $env->{'qspawn.wcb'};
max => $max || 32,
running => 0,
run_queue => [],
+ # RLIMIT_CPU => undef,
+ # RLIMIT_DATA => undef,
+ # RLIMIT_CORE => undef,
}, $class;
}
+sub setup_rlimit {
+ my ($self, $name, $config) = @_;
+ foreach my $rlim (PublicInbox::Spawn::RLIMITS()) {
+ my $k = lc($rlim);
+ $k =~ tr/_//d;
+ $k = "publicinboxlimiter.$name.$k";
+ defined(my $v = $config->{$k}) or next;
+ my @rlimit = split(/\s*,\s*/, $v);
+ if (scalar(@rlimit) == 1) {
+ push @rlimit, $rlimit[0];
+ } elsif (scalar(@rlimit) != 2) {
+ warn "could not parse $k: $v\n";
+ }
+ eval { require BSD::Resource };
+ if ($@) {
+ warn "BSD::Resource missing for $rlim";
+ next;
+ }
+ foreach my $i (0..$#rlimit) {
+ next if $rlimit[$i] ne 'INFINITY';
+ $rlimit[$i] = BSD::Resource::RLIM_INFINITY();
+ }
+ $self->{$rlim} = \@rlimit;
+ }
+}
+
# captures everything into a buffer and executes a callback when done
package PublicInbox::Qspawn::Qx;
use strict;