X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FQspawn.pm;h=5e4fd5cb8541a2a8202d188ae1fd73c88fe64188;hb=HEAD;hp=ab0ef25c5611f3e645a4f4e593ae143ee94291a6;hpb=a30dd4aa9455f4166c6bc18e043fccd645b4231c;p=public-inbox.git

diff --git a/lib/PublicInbox/Qspawn.pm b/lib/PublicInbox/Qspawn.pm
index ab0ef25c..5e4fd5cb 100644
--- a/lib/PublicInbox/Qspawn.pm
+++ b/lib/PublicInbox/Qspawn.pm
@@ -25,10 +25,10 @@
 # processes such as git-apply(1).
 
 package PublicInbox::Qspawn;
-use strict;
-use v5.10.1;
+use v5.12;
 use PublicInbox::Spawn qw(popen_rd);
 use PublicInbox::GzipFilter;
+use Scalar::Util qw(blessed);
 
 # n.b.: we get EAGAIN with public-inbox-httpd, and EINTR on other PSGI servers
 use Errno qw(EAGAIN EINTR);
@@ -56,36 +56,23 @@ sub _do_spawn {
 			$o{$k} = $rlimit;
 		}
 	}
-	$self->{cmd} = $o{quiet} ? undef : $cmd;
+	$self->{cmd} = $cmd;
+	$self->{-quiet} = 1 if $o{quiet};
+	$o{cb_arg} = [ \&waitpid_err, $self ];
 	eval {
 		# popen_rd may die on EMFILE, ENFILE
-		$self->{rpipe} = popen_rd($cmd, $cmd_env, \%o);
-
-		die "E: $!" unless defined($self->{rpipe});
-
+		$self->{rpipe} = popen_rd($cmd, $cmd_env, \%o) // die "E: $!";
 		$limiter->{running}++;
 		$start_cb->($self); # EPOLL_CTL_ADD may ENOSPC/ENOMEM
 	};
 	finish($self, $@) if $@;
 }
 
-sub child_err ($) {
-	my ($child_error) = @_; # typically $?
-	my $exitstatus = ($child_error >> 8) or return;
-	my $sig = $child_error & 127;
-	my $msg = "exit status=$exitstatus";
-	$msg .= " signal=$sig" if $sig;
-	$msg;
-}
-
-sub finalize ($$) {
-	my ($self, $err) = @_;
-
-	my ($env, $qx_cb, $qx_arg, $qx_buf) =
-		delete @$self{qw(psgi_env qx_cb qx_arg qx_buf)};
+sub finalize ($) {
+	my ($self) = @_;
 
-	# done, spawn whatever's in the queue
-	my $limiter = $self->{limiter};
+	# process is done, spawn whatever's in the queue
+	my $limiter = delete $self->{limiter} or return;
 	my $running = --$limiter->{running};
 
 	if ($running < $limiter->{max}) {
@@ -93,14 +80,16 @@ sub finalize ($$) {
 			_do_spawn(@$next, $limiter);
 		}
 	}
-
-	if ($err) {
+	if (my $err = $self->{_err}) { # set by finish or waitpid_err
 		utf8::decode($err);
 		if (my $dst = $self->{qsp_err}) {
 			$$dst .= $$dst ? " $err" : "; $err";
 		}
-		warn "@{$self->{cmd}}: $err" if $self->{cmd};
+		warn "@{$self->{cmd}}: $err\n" if !$self->{-quiet};
 	}
+
+	my ($env, $qx_cb, $qx_arg, $qx_buf) =
+		delete @$self{qw(psgi_env qx_cb qx_arg qx_buf)};
 	if ($qx_cb) {
 		eval { $qx_cb->($qx_buf, $qx_arg) };
 		return unless $@;
@@ -109,19 +98,36 @@ sub finalize ($$) {
 	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));
 	}
 }
 
-# callback for dwaitpid or ProcessPipe
-sub waitpid_err { finalize($_[0], child_err($?)) }
+sub DESTROY { finalize($_[0]) } # ->finalize is idempotent
+
+sub waitpid_err { # callback for awaitpid
+	my (undef, $self) = @_; # $_[0]: pid
+	$self->{_err} = ''; # for defined check in ->finish
+	if ($?) {
+		my $status = $? >> 8;
+		my $sig = $? & 127;
+		$self->{_err} .= "exit status=$status";
+		$self->{_err} .= " signal=$sig" if $sig;
+	}
+	finalize($self) if !$self->{rpipe};
+}
 
 sub finish ($;$) {
 	my ($self, $err) = @_;
-	my $tied_pp = delete($self->{rpipe}) or return finalize($self, $err);
-	my PublicInbox::ProcessPipe $pp = tied *$tied_pp;
-	@$pp{qw(cb arg)} = (\&waitpid_err, $self); # for ->DESTROY
+	$self->{_err} //= $err; # only for $@
+
+	# we can safely finalize if pipe was closed before, or if
+	# {_err} is defined by waitpid_err.  Deleting {rpipe} will
+	# trigger PublicInbox::ProcessPipe::DESTROY -> waitpid_err,
+	# but it may not fire right away if inside the event loop.
+	my $closed_before = !delete($self->{rpipe});
+	finalize($self) if $closed_before || defined($self->{_err});
 }
 
 sub start ($$$) {
@@ -133,22 +139,22 @@ sub start ($$$) {
 	}
 }
 
-sub psgi_qx_init_cb {
+sub psgi_qx_init_cb { # this may be PublicInbox::HTTPD::Async {cb}
 	my ($self) = @_;
-	my $async = delete $self->{async}; # PublicInbox::HTTPD::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 {
+	if (!defined($r)) {
 		return if $! == EAGAIN; # try again when notified
 		goto reread if $! == EINTR;
 		event_step($self, $!);
+	} elsif (my $as = delete $self->{async}) { # PublicInbox::HTTPD::Async
+		$as->async_pass($self->{psgi_env}->{'psgix.io'},
+				$self->{qx_fh}, \$buf);
+	} elsif ($r) { # generic PSGI:
+		print { $self->{qx_fh} } $buf;
+	} else { # EOF
+		event_step($self, undef);
 	}
 }
 
@@ -200,7 +206,7 @@ sub rd_hdr ($) {
 	my $total_rd = 0;
 	my $hdr_buf = $self->{hdr_buf};
 	my ($ph_cb, $ph_arg) = @{$self->{parse_hdr}};
-	do {
+	until (defined($ret)) {
 		my $r = sysread($self->{rpipe}, $$hdr_buf, 4096,
 				length($$hdr_buf));
 		if (defined($r)) {
@@ -209,6 +215,11 @@ sub rd_hdr ($) {
 			if ($@) {
 				warn "parse_hdr: $@";
 				$ret = [ 500, [], [ "Internal error\n" ] ];
+			} elsif (!defined($ret) && !$r) {
+				warn <<EOM;
+EOF parsing headers from @{$self->{cmd}} ($self->{psgi_env}->{REQUEST_URI})
+EOM
+				$ret = [ 500, [], [ "Internal error\n" ] ];
 			}
 		} else {
 			# caller should notify us when it's ready:
@@ -217,18 +228,19 @@ sub rd_hdr ($) {
 			warn "error reading header: $!";
 			$ret = [ 500, [], [ "Internal error\n" ] ];
 		}
-	} until (defined $ret);
+	}
 	delete $self->{parse_hdr}; # done parsing headers
 	$ret;
 }
 
-sub psgi_return_init_cb {
+sub psgi_return_init_cb { # this may be PublicInbox::HTTPD::Async {cb}
 	my ($self) = @_;
 	my $r = rd_hdr($self) or return;
 	my $env = $self->{psgi_env};
 	my $filter;
-	if (ref($r) eq 'ARRAY' && Scalar::Util::blessed($r->[2]) &&
-			$r->[2]->can('attach')) {
+
+	# this is for RepoAtom since that can fire after parse_cgi_headers
+	if (ref($r) eq 'ARRAY' && blessed($r->[2]) && $r->[2]->can('attach')) {
 		$filter = pop @$r;
 	}
 	$filter //= delete($env->{'qspawn.filter'}) // (ref($r) eq 'ARRAY' ?
@@ -239,10 +251,9 @@ sub psgi_return_init_cb {
 	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:
+		} else { # generic PSGI, use PublicInbox::ProcessPipe::CLOSE
 			delete($self->{rpipe})->close;
 			event_step($self);
-			waitpid_err($self);
 		}
 		if (ref($r) eq 'ARRAY') { # error
 			$wcb->($r)
@@ -324,8 +335,7 @@ sub psgi_return {
 }
 
 package PublicInbox::Qspawn::Limiter;
-use strict;
-use warnings;
+use v5.12;
 
 sub new {
 	my ($class, $max) = @_;