X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FQspawn.pm;h=63ec3648c933bdb7775a51a2fe90e3cfc235231c;hb=95bdac7f09c69036efed537a4d03d5bdd2ae4eb6;hp=33e20147bbb9bad6acc18c621bc42a0e91c69d54;hpb=574cf9098d09c7c2e10c28cf577321161ddecd61;p=public-inbox.git diff --git a/lib/PublicInbox/Qspawn.pm b/lib/PublicInbox/Qspawn.pm index 33e20147..63ec3648 100644 --- a/lib/PublicInbox/Qspawn.pm +++ b/lib/PublicInbox/Qspawn.pm @@ -1,4 +1,4 @@ -# Copyright (C) 2016-2019 all contributors +# Copyright (C) 2016-2020 all contributors # License: AGPL-3.0+ # Like most Perl modules in public-inbox, this is internal and @@ -27,7 +27,6 @@ package PublicInbox::Qspawn; use strict; 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); @@ -46,27 +45,28 @@ sub new ($$$;) { sub _do_spawn { my ($self, $start_cb, $limiter) = @_; my $err; - my ($cmd, $cmd_env, $opts) = @{$self->{args}}; - my %opts = %{$opts || {}}; + my ($cmd, $cmd_env, $opt) = @{$self->{args}}; + my %o = %{$opt || {}}; $self->{limiter} = $limiter; foreach my $k (PublicInbox::Spawn::RLIMITS()) { if (defined(my $rlimit = $limiter->{$k})) { - $opts{$k} = $rlimit; + $o{$k} = $rlimit; } } + eval { + # popen_rd may die on EMFILE, ENFILE + ($self->{rpipe}, $self->{pid}) = popen_rd($cmd, $cmd_env, \%o); + $self->{args} = $o{quiet} ? undef : $cmd; - ($self->{rpipe}, $self->{pid}) = popen_rd($cmd, $cmd_env, \%opts); + die "E: $!" unless defined($self->{pid}); - # 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} = $!; + $start_cb->($self); # EPOLL_CTL_ADD may ENOSPC/ENOMEM + }; + if ($@) { + $self->{err} = $@; + finish($self); } - $start_cb->($self); } sub child_err ($) { @@ -108,8 +108,12 @@ sub waitpid_err ($$) { } if ($err) { - $self->{err} = $err; - if ($env && !$env->{'qspawn.quiet'}) { + if ($self->{err}) { + $self->{err} .= "; $err"; + } else { + $self->{err} = $err; + } + if ($env && $self->{args}) { log_err($env, join(' ', @{$self->{args}}) . ": $err"); } } @@ -149,6 +153,37 @@ sub start ($$$) { } } +sub psgi_qx_init_cb { + my ($self) = @_; + my $async = delete $self->{async}; + my ($r, $buf); + my $qx_fh = $self->{qx_fh}; +reread: + $r = sysread($self->{rpipe}, $buf, 65536); + if ($async) { + $async->async_pass($self->{psgi_env}->{'psgix.io'}, + $qx_fh, \$buf); + } elsif (defined $r) { + $r ? (print $qx_fh $buf) : event_step($self, undef); + } else { + return if $! == EAGAIN; # try again when notified + goto reread if $! == EINTR; + event_step($self, $!); + } +} + +sub psgi_qx_start { + my ($self) = @_; + if (my $async = $self->{psgi_env}->{'pi-httpd.async'}) { + # PublicInbox::HTTPD::Async->new(rpipe, $cb, cb_arg, $end_obj) + $self->{async} = $async->($self->{rpipe}, + \&psgi_qx_init_cb, $self, $self); + # init_cb will call ->async_pass or ->close + } else { # generic PSGI + psgi_qx_init_cb($self) while $self->{qx_fh}; + } +} + # 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 @@ -162,44 +197,8 @@ sub psgi_qx { $self->{qx_arg} = $qx_arg; $self->{qx_fh} = $qx_fh; $self->{qx_buf} = \$qx_buf; - my $async = $env->{'pi-httpd.async'}; - my $cb = sub { - my ($r, $buf); -reread: - $r = sysread($self->{rpipe}, $buf, 65536); - if ($async) { - $async->async_pass($env->{'psgix.io'}, $qx_fh, \$buf); - } elsif (defined $r) { - $r ? $qx_fh->write($buf) : event_step($self, undef); - } else { - return if $! == EAGAIN; # try again when notified - goto reread if $! == EINTR; - event_step($self, $!); - } - }; $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32); - $self->start($limiter, sub { # start_cb, may run later, much later... - if ($async) { - # PublicInbox::HTTPD::Async->new(rpipe, $cb, $end_obj) - $async = $async->($self->{rpipe}, $cb, undef, $self); - # $cb will call ->async_pass or ->close - } else { # generic PSGI - $cb->() while $self->{qx_fh}; - } - }); -} - -# create a filter for "push"-based streaming PSGI writes used by HTTPD::Async -sub filter_fh ($$) { - my ($fh, $filter) = @_; - Plack::Util::inline_object( - close => sub { - $fh->write($filter->(undef)); - $fh->close; - }, - write => sub { - $fh->write($filter->($_[0])); - }); + start($self, $limiter, \&psgi_qx_start); } # this is called on pipe EOF to reap the process, may be called @@ -221,12 +220,13 @@ sub rd_hdr ($) { my $ret; my $total_rd = 0; my $hdr_buf = $self->{hdr_buf}; + my ($ph_cb, $ph_arg) = @{$self->{parse_hdr}}; do { my $r = sysread($self->{rpipe}, $$hdr_buf, 4096, length($$hdr_buf)); if (defined($r)) { $total_rd += $r; - $ret = $self->{parse_hdr}->($total_rd, $hdr_buf); + $ret = $ph_cb->($total_rd, $hdr_buf, $ph_arg); } else { # caller should notify us when it's ready: return if $! == EAGAIN; @@ -243,7 +243,6 @@ sub psgi_return_init_cb { my ($self) = @_; my $r = rd_hdr($self) or return; my $env = $self->{psgi_env}; - my $filter = delete $env->{'qspawn.filter'}; my $wcb = delete $env->{'qspawn.wcb'}; my $async = delete $self->{async}; if (scalar(@$r) == 3) { # error @@ -258,7 +257,6 @@ sub psgi_return_init_cb { } elsif ($async) { # done reading headers, handoff to read body my $fh = $wcb->($r); # scalar @$r == 2 - $fh = filter_fh($fh, $filter) if $filter; $self->{fh} = $fh; $async->async_pass($env->{'psgix.io'}, $fh, delete($self->{hdr_buf})); @@ -266,7 +264,7 @@ sub psgi_return_init_cb { require PublicInbox::GetlineBody; $r->[2] = PublicInbox::GetlineBody->new($self->{rpipe}, \&event_step, $self, - ${$self->{hdr_buf}}, $filter); + ${$self->{hdr_buf}}); $wcb->($r); } @@ -296,9 +294,6 @@ sub psgi_return_start { # may run later, much later... # psgi_return will return an anonymous # sub for the PSGI server to call # -# $env->{'qspawn.filter'} - filter callback, receives a string as input, -# undef on EOF -# # $limiter - the Limiter object to use (uses the def_limiter if not given) # # $parse_hdr - Initial read function; often for parsing CGI header output. @@ -309,10 +304,10 @@ sub psgi_return_start { # may run later, much later... # psgix.io. 3-element arrays means the body is available # immediately (or streamed via ->getline (pull-based)). sub psgi_return { - my ($self, $env, $limiter, $parse_hdr) = @_; + my ($self, $env, $limiter, $parse_hdr, $hdr_arg) = @_; $self->{psgi_env} = $env; $self->{hdr_buf} = \(my $hdr_buf = ''); - $self->{parse_hdr} = $parse_hdr; + $self->{parse_hdr} = [ $parse_hdr, $hdr_arg ]; $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32); # the caller already captured the PSGI write callback from