X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FQspawn.pm;h=f8cbffcb359b74629a888119bcbba4237335d0ad;hb=81a906c55b9510b38ea035dca13475fdd9d27083;hp=7e50a59ae49e1e65a67b16bf34109288a17694c8;hpb=22e51bd9da476fa9f4d72d4ec25ed0ec4a1f636f;p=public-inbox.git diff --git a/lib/PublicInbox/Qspawn.pm b/lib/PublicInbox/Qspawn.pm index 7e50a59a..f8cbffcb 100644 --- a/lib/PublicInbox/Qspawn.pm +++ b/lib/PublicInbox/Qspawn.pm @@ -1,4 +1,4 @@ -# Copyright (C) 2016-2021 all contributors +# Copyright (C) all contributors # License: AGPL-3.0+ # Like most Perl modules in public-inbox, this is internal and @@ -12,7 +12,7 @@ # 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 the PublicInbox::DS->EventLoop or any +# This does not depend on the PublicInbox::DS::event_loop or any # other external scheduling mechanism, you just need to call # start() and finish() appropriately. However, public-inbox-httpd # (which uses PublicInbox::DS) will be able to schedule this @@ -26,6 +26,7 @@ package PublicInbox::Qspawn; use strict; +use v5.10.1; use PublicInbox::Spawn qw(popen_rd); use PublicInbox::GzipFilter; @@ -38,7 +39,8 @@ my $def_limiter; # $cmd is the command to spawn # $cmd_env is the environ for the child process (not PSGI env) # $opt can include redirects and perhaps other process spawning options -sub new ($$$;) { +# {qsp_err} is an optional error buffer callers may access themselves +sub new { my ($class, $cmd, $cmd_env, $opt) = @_; bless { args => [ $cmd, $cmd_env, $opt ] }, $class; } @@ -76,11 +78,6 @@ sub child_err ($) { $msg; } -sub log_err ($$) { - my ($env, $msg) = @_; - $env->{'psgi.errors'}->print($msg, "\n"); -} - sub finalize ($$) { my ($self, $err) = @_; @@ -98,21 +95,23 @@ sub finalize ($$) { } if ($err) { - if (defined $self->{err}) { - $self->{err} .= "; $err"; - } else { - $self->{err} = $err; - } - if ($env && $self->{cmd}) { - log_err($env, join(' ', @{$self->{cmd}}) . ": $err"); + utf8::decode($err); + if (my $dst = $self->{qsp_err}) { + $$dst .= $$dst ? " $err" : "; $err"; } + warn "@{$self->{cmd}}: $err" if $self->{cmd}; } if ($qx_cb) { eval { $qx_cb->($qx_buf, $qx_arg) }; - } elsif (my $wcb = delete $env->{'qspawn.wcb'}) { + return unless $@; + warn "E: $@"; # hope qspawn.wcb can handle it + } + return if $self->{passed}; # another command chained it + if (my $wcb = delete $env->{'qspawn.wcb'}) { # have we started writing, yet? + my $code = delete $env->{'qspawn.fallback'} // 500; require PublicInbox::WwwStatic; - $wcb->(PublicInbox::WwwStatic::r(500)); + $wcb->(PublicInbox::WwwStatic::r($code)); } } @@ -137,7 +136,7 @@ sub start ($$$) { sub psgi_qx_init_cb { my ($self) = @_; - my $async = delete $self->{async}; + my $async = delete $self->{async}; # PublicInbox::HTTPD::Async my ($r, $buf); my $qx_fh = $self->{qx_fh}; reread: @@ -188,16 +187,15 @@ sub psgi_qx { # PSGI servers. sub event_step { my ($self, $err) = @_; # $err: $! - log_err($self->{psgi_env}, "psgi_{return,qx} $err") if defined($err); + warn "psgi_{return,qx} $err" if defined($err); finish($self); - my ($fh, $qx_fh) = delete(@$self{qw(fh qx_fh)}); + my ($fh, $qx_fh) = delete(@$self{qw(qfh qx_fh)}); $fh->close if $fh; # async-only (psgi_return) } sub rd_hdr ($) { my ($self) = @_; # 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; @@ -210,14 +208,14 @@ sub rd_hdr ($) { $total_rd += $r; eval { $ret = $ph_cb->($total_rd, $hdr_buf, $ph_arg) }; if ($@) { - log_err($self->{psgi_env}, "parse_hdr: $@"); + warn "parse_hdr: $@"; $ret = [ 500, [], [ "Internal error\n" ] ]; } } else { # caller should notify us when it's ready: return if $! == EAGAIN; next if $! == EINTR; # immediate retry - log_err($self->{psgi_env}, "error reading header: $!"); + warn "error reading header: $!"; $ret = [ 500, [], [ "Internal error\n" ] ]; } } until (defined $ret); @@ -229,41 +227,52 @@ 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'} // - PublicInbox::GzipFilter::qsp_maybe($r->[1], $env); + my $filter; + if (ref($r) eq 'ARRAY' && Scalar::Util::blessed($r->[2]) && + $r->[2]->can('attach')) { + $filter = pop @$r; + } + $filter //= delete($env->{'qspawn.filter'}) // (ref($r) eq 'ARRAY' ? + PublicInbox::GzipFilter::qsp_maybe($r->[1], $env) : undef); my $wcb = delete $env->{'qspawn.wcb'}; - my $async = delete $self->{async}; - if (scalar(@$r) == 3) { # error - if ($async) { - # calls rpipe->close && ->event_step - $async->close; - } else { - $self->{rpipe}->close; + my $async = delete $self->{async}; # PublicInbox::HTTPD::Async + if (ref($r) ne 'ARRAY' || scalar(@$r) == 3) { # error + if ($async) { # calls rpipe->close && ->event_step + $async->close; # PublicInbox::HTTPD::Async::close + } else { # generic PSGI: + delete($self->{rpipe})->close; event_step($self); + waitpid_err($self); } - $wcb->($r); + if (ref($r) eq 'ARRAY') { # error + $wcb->($r) + } elsif (ref($r) eq 'CODE') { # chain another command + $r->($wcb); + $self->{passed} = 1; + } + # else do nothing } elsif ($async) { # done reading headers, handoff to read body my $fh = $wcb->($r); # scalar @$r == 2 $fh = $filter->attach($fh) if $filter; - $self->{fh} = $fh; + $self->{qfh} = $fh; $async->async_pass($env->{'psgix.io'}, $fh, delete($self->{hdr_buf})); } else { # for synchronous PSGI servers require PublicInbox::GetlineBody; + my $buf = delete $self->{hdr_buf}; $r->[2] = PublicInbox::GetlineBody->new($self->{rpipe}, - \&event_step, $self, - ${$self->{hdr_buf}}, $filter); + \&event_step, $self, $$buf, $filter); $wcb->($r); } } sub psgi_return_start { # may run later, much later... my ($self) = @_; - if (my $async = $self->{psgi_env}->{'pi-httpd.async'}) { + if (my $cb = $self->{psgi_env}->{'pi-httpd.async'}) { # PublicInbox::HTTPD::Async->new(rpipe, $cb, $cb_arg, $end_obj) - $self->{async} = $async->($self->{rpipe}, + $self->{async} = $cb->($self->{rpipe}, \&psgi_return_init_cb, $self, $self); } else { # generic PSGI psgi_return_init_cb($self) while $self->{parse_hdr};