[PATCH 0/5] a few more HTTP-async-related simplifications

[PATCH 0/5] a few more HTTP-async-related simplifications

[Eric Wong]

Since we'll be doing more expensive processes in repobrowse,
it's prudent to refactor and commonalize some of this code.
Here's a start...

Eric Wong (5):
      t/spawn.t: additional tests for popen_rd
      git-http-backend: remove process limit
      git-http-backend: simplify dumb serving
      http: rework async_pass support
      git-http-backend: switch to async_pass

 lib/PublicInbox/GitHTTPBackend.pm | 251 ++++++++++++++++----------------------
 lib/PublicInbox/HTTP.pm           |  24 +---
 lib/PublicInbox/HTTPD/Async.pm    |  45 ++++---
 t/spawn.t                         |  10 ++
 4 files changed, 143 insertions(+), 187 deletions(-)

[PATCH 1/5] t/spawn.t: additional tests for popen_rd

[Eric Wong]

We need to ensure $? is set properly for users.
---
 t/spawn.t | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/t/spawn.t b/t/spawn.t
index d52b646..9e58f67 100644
--- a/t/spawn.t
+++ b/t/spawn.t
@@ -70,6 +70,15 @@ use PublicInbox::Spawn qw(which spawn popen_rd);
 	is(sysread($fh, $buf, 6), 6, 'sysread got 6 bytes');
 	is($buf, "hello\n", 'tied gets works');
 	is(sysread($fh, $buf, 6), 0, 'sysread got EOF');
+	$? = 1;
+	close $fh;
+	is($?, 0, '$? set properly');
+}
+
+{
+	my $fh = popen_rd([qw(false)]);
+	close $fh;
+	isnt($?, 0, '$? set properly: '.$?);
 }
 
 {
@@ -80,6 +89,7 @@ use PublicInbox::Spawn qw(which spawn popen_rd);
 	   'sysread returned quickly with EAGAIN');
 	is(kill(15, $pid), 1, 'child process killed early');
 	is(waitpid($pid, 0), $pid, 'child process reapable');
+	isnt($?, 0, '$? set properly: '.$?);
 }
 
 done_testing();

[PATCH 2/5] git-http-backend: remove process limit

[Eric Wong]

We will figure out a different way to avoid overloading...
---
 lib/PublicInbox/GitHTTPBackend.pm | 11 -----------
 1 file changed, 11 deletions(-)

diff --git a/lib/PublicInbox/GitHTTPBackend.pm b/lib/PublicInbox/GitHTTPBackend.pm
index b58cc30..35c3383 100644
--- a/lib/PublicInbox/GitHTTPBackend.pm
+++ b/lib/PublicInbox/GitHTTPBackend.pm
@@ -11,14 +11,6 @@ use IO::File;
 use PublicInbox::Spawn qw(spawn);
 use HTTP::Date qw(time2str);
 
-# TODO: make configurable, but keep in mind it's better to have
-# multiple -httpd worker processes which are already scaled to
-# the proper number of CPUs and memory.  git-pack-objects(1) may
-# also use threads and bust memory limits, too, so I recommend
-# limiting threads to 1 (via `pack.threads` knob in git) for serving.
-my $LIMIT = 1;
-my $nr_running = 0;
-
 # n.b. serving "description" and "cloneurl" should be innocuous enough to
 # not cause problems.  serving "config" might...
 my @text = qw[HEAD info/refs
@@ -54,7 +46,6 @@ sub r ($) {
 
 sub serve {
 	my ($cgi, $git, $path) = @_;
-	return serve_dumb($cgi, $git, $path) if $nr_running >= $LIMIT;
 
 	my $service = $cgi->param('service') || '';
 	if ($service =~ /\Agit-\w+-pack\z/ || $path =~ /\Agit-\w+-pack\z/) {
@@ -243,7 +234,6 @@ sub serve_smart {
 	$wpipe = $in = undef;
 	$buf = '';
 	my ($vin, $fh, $res);
-	$nr_running++;
 
 	# Danga::Socket users, we queue up the read_enable callback to
 	# fire after pending writes are complete:
@@ -264,7 +254,6 @@ sub serve_smart {
 			# PublicInbox::HTTPD::Async::close:
 			$rpipe->close;
 			$rpipe = undef;
-			$nr_running--;
 		}
 		if (defined $pid) {
 			my $e = $pid == waitpid($pid, 0) ?

[PATCH 3/5] git-http-backend: simplify dumb serving

[Eric Wong]

We can rely entirely on getline + close callbacks
and be compatible with 100% of PSGI servers.
---
 lib/PublicInbox/GitHTTPBackend.pm | 66 +++++++++++----------------------------
 1 file changed, 19 insertions(+), 47 deletions(-)

diff --git a/lib/PublicInbox/GitHTTPBackend.pm b/lib/PublicInbox/GitHTTPBackend.pm
index 35c3383..97d96d5 100644
--- a/lib/PublicInbox/GitHTTPBackend.pm
+++ b/lib/PublicInbox/GitHTTPBackend.pm
@@ -92,43 +92,9 @@ sub serve_dumb {
 	# TODO: If-Modified-Since and Last-Modified?
 	open my $in, '<', $f or return r(404);
 	my $len = $size;
-	my $n = 65536; # try to negotiate a big TCP window, first
-	my ($next, $fh);
-	my $cb = sub {
-		$n = $len if $len < $n;
-		my $r = sysread($in, my $buf, $n);
-		if (!defined $r) {
-			err($env, "$f read error: $!");
-			drop_client($env);
-		} elsif ($r <= 0) {
-			err($env, "$f EOF with $len bytes left");
-			drop_client($env);
-		} else {
-			$len -= $r;
-			$fh->write($buf);
-			if ($len == 0) {
-				$fh->close;
-			} elsif ($next) {
-				# avoid recursion in Danga::Socket::write
-				unless ($nextq) {
-					$nextq = [];
-					Danga::Socket->AddTimer(0, *do_next);
-				}
-				# avoid buffering too much in case we have
-				# slow clients:
-				$n = 8192;
-				push @$nextq, $next;
-				return;
-			}
-		}
-		# all done, cleanup references:
-		$fh = $next = undef;
-	};
-
 	my $code = 200;
 	push @h, 'Content-Type', $type;
-	my $range = $env->{HTTP_RANGE};
-	if (defined $range && $range =~ /\bbytes=(\d*)-(\d*)\z/) {
+	if (($env->{HTTP_RANGE} || '') =~ /\bbytes=(\d*)-(\d*)\z/) {
 		($code, $len) = prepare_range($cgi, $in, \@h, $1, $2, $size);
 		if ($code == 416) {
 			push @h, 'Content-Range', "bytes */$size";
@@ -136,18 +102,24 @@ sub serve_dumb {
 		}
 	}
 	push @h, 'Content-Length', $len;
-
-	sub {
-		my ($res) = @_; # Plack callback
-		$fh = $res->([ $code, \@h ]);
-		if (defined $env->{'pi-httpd.async'}) {
-			my $pi_http = $env->{'psgix.io'};
-			$next = sub { $pi_http->write($cb) };
-			$cb->(); # start it off!
-		} else {
-			$cb->() while $fh;
-		}
-	}
+	my $n = 65536;
+	[ $code, \@h, Plack::Util::inline_object(close => sub { close $in },
+		getline => sub {
+			return if $len == 0;
+			$n = $len if $len < $n;
+			my $r = sysread($in, my $buf, $n);
+			if (!defined $r) {
+				err($env, "$f read error: $!");
+			} elsif ($r <= 0) {
+				err($env, "$f EOF with $len bytes left");
+			} else {
+				$len -= $r;
+				$n = 8192;
+				return $buf;
+			}
+			drop_client($env);
+			return;
+		})]
 }
 
 sub prepare_range {

[PATCH 4/5] http: rework async_pass support

[Eric Wong]

Unfortunately, the original design did not work because
middleware can wrap the response body and make `async_pass'
invisible to HTTP.pm
---
 lib/PublicInbox/HTTP.pm        | 24 ++++------------------
 lib/PublicInbox/HTTPD/Async.pm | 45 +++++++++++++++++++++++++-----------------
 2 files changed, 31 insertions(+), 38 deletions(-)

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index 4eb1448..480800b 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -11,7 +11,7 @@ package PublicInbox::HTTP;
 use strict;
 use warnings;
 use base qw(Danga::Socket);
-use fields qw(httpd env rbuf input_left remote_addr remote_port);
+use fields qw(httpd env rbuf input_left remote_addr remote_port forward);
 use Fcntl qw(:seek);
 use Plack::HTTPParser qw(parse_http_request); # XS or pure Perl
 use HTTP::Status qw(status_message);
@@ -219,24 +219,6 @@ sub response_write {
 		if (ref $body eq 'ARRAY') {
 			$write->($_) foreach @$body;
 			$close->();
-		} elsif ($body->can('async_pass')) { # HTTPD::Async
-			# prevent us from reading the body faster than we
-			# can write to the client
-			my $restart_read = sub { $body->watch_read(1) };
-			$body->async_pass(sub {
-				local $/ = \8192;
-				my $buf = $body->getline;
-				if (defined $buf) {
-					$write->($buf);
-					if ($self->{write_buf_size}) {
-						$body->watch_read(0);
-						$self->write($restart_read);
-					}
-					return; # continue waiting
-				}
-				$body->close;
-				$close->();
-			});
 		} else {
 			my $pull;
 			$pull = sub {
@@ -438,7 +420,9 @@ sub event_err { $_[0]->close }
 
 sub close {
 	my $self = shift;
-	$self->{env} = undef;
+	my $forward = $self->{forward};
+	$forward->close if $forward;
+	$self->{forward} = $self->{env} = undef;
 	$self->SUPER::close(@_);
 }
 
diff --git a/lib/PublicInbox/HTTPD/Async.pm b/lib/PublicInbox/HTTPD/Async.pm
index 8f3a6a0..8efa7a6 100644
--- a/lib/PublicInbox/HTTPD/Async.pm
+++ b/lib/PublicInbox/HTTPD/Async.pm
@@ -21,29 +21,38 @@ sub new {
 	$self;
 }
 
-sub async_pass { $_[0]->{cb} = $_[1] }
-sub event_read { $_[0]->{cb}->() }
-sub event_hup { $_[0]->{cb}->() }
-sub event_err { $_[0]->{cb}->() }
-sub sysread { shift->{sock}->sysread(@_) }
-
-sub getline {
-	my ($self) = @_;
-	die 'getline called without $/ ref' unless ref $/;
-	while (1) {
-		my $ret = $self->read(8192); # Danga::Socket::read
-		return $$ret if defined $ret;
+sub async_pass {
+	my ($self, $io, $fh) = @_;
+	my $restart_read = sub { $self->watch_read(1) };
 
-		return unless $!{EAGAIN} || $!{EINTR};
+	# In case the client HTTP connection ($io) dies, it
+	# will automatically close this ($self) object.
+	$io->{forward} = $self;
+	$self->{cb} = sub {
+		my $r = sysread($self->{sock}, my $buf, 8192);
+		if ($r) {
+			$fh->write($buf);
+			if ($io->{write_buf_size}) {
+				$self->watch_read(0);
+				$io->write($restart_read);
+			}
+			return; # stay in watch_read
+		} elsif (!defined $r) {
+			return if $!{EAGAIN} || $!{EINTR};
+		}
 
-		# in case of spurious wakeup, hopefully we never hit this
-		my $vin = '';
-		vec($vin, $self->{fd}, 1) = 1;
-		my $n;
-		do { $n = select($vin, undef, undef, undef) } until $n;
+		# Done! Error handling will happen in $fh->close
+		$io->{forward} = undef;
+		$self->close;
+		$fh->close;
 	}
 }
 
+sub event_read { $_[0]->{cb}->() }
+sub event_hup { $_[0]->{cb}->() }
+sub event_err { $_[0]->{cb}->() }
+sub sysread { shift->{sock}->sysread(@_) }
+
 sub close {
 	my $self = shift;
 	$self->{cb} = undef;

[PATCH 5/5] git-http-backend: switch to async_pass

[Eric Wong]

This simplifies the code somewhat; but it could probably
still be made simpler.  It will need to support command
queueing for expensive commands so expensive processes
can be queued up.
---
 lib/PublicInbox/GitHTTPBackend.pm | 174 ++++++++++++++++++--------------------
 1 file changed, 83 insertions(+), 91 deletions(-)

diff --git a/lib/PublicInbox/GitHTTPBackend.pm b/lib/PublicInbox/GitHTTPBackend.pm
index 97d96d5..cca8a6d 100644
--- a/lib/PublicInbox/GitHTTPBackend.pm
+++ b/lib/PublicInbox/GitHTTPBackend.pm
@@ -10,6 +10,7 @@ use Fcntl qw(:seek);
 use IO::File;
 use PublicInbox::Spawn qw(spawn);
 use HTTP::Date qw(time2str);
+use HTTP::Status qw(status_message);
 
 # n.b. serving "description" and "cloneurl" should be innocuous enough to
 # not cause problems.  serving "config" might...
@@ -39,9 +40,12 @@ sub do_next () {
 	}
 }
 
-sub r ($) {
-	my ($s) = @_;
-	[ $s, [qw(Content-Type text/plain Content-Length 0), @no_cache ], [] ]
+sub r ($;$) {
+	my ($code, $msg) = @_;
+	$msg ||= status_message($code);
+	my $len = length($msg);
+	[ $code, [qw(Content-Type text/plain Content-Length), $len, @no_cache],
+		[$msg] ]
 }
 
 sub serve {
@@ -167,14 +171,9 @@ sub prepare_range {
 sub serve_smart {
 	my ($cgi, $git, $path) = @_;
 	my $env = $cgi->{env};
-
-	my $input = $env->{'psgi.input'};
-	my $buf;
-	my $in;
-	my $fd = eval { fileno($input) };
-	if (defined $fd && $fd >= 0) {
-		$in = $input;
-	} else {
+	my $in = $env->{'psgi.input'};
+	my $fd = eval { fileno($in) };
+	unless (defined $fd && $fd >= 0) {
 		$in = input_to_file($env) or return r(500);
 	}
 	my ($rpipe, $wpipe);
@@ -204,91 +203,67 @@ sub serve_smart {
 		return;
 	}
 	$wpipe = $in = undef;
-	$buf = '';
-	my ($vin, $fh, $res);
-
-	# Danga::Socket users, we queue up the read_enable callback to
-	# fire after pending writes are complete:
-	my $pi_http = $env->{'psgix.io'};
-	my $read_enable = sub { $rpipe->watch_read(1) };
-	my $read_disable = sub {
-		$rpipe->watch_read(0);
-		$pi_http->write($read_enable);
-	};
-
 	my $end = sub {
-		if ($fh) {
-			$fh->close;
-			$fh = undef;
-		}
-		if ($rpipe) {
-			# _may_ be Danga::Socket::close via
-			# PublicInbox::HTTPD::Async::close:
-			$rpipe->close;
-			$rpipe = undef;
-		}
-		if (defined $pid) {
-			my $e = $pid == waitpid($pid, 0) ?
-				$? : "PID:$pid still running?";
-			err($env, "git http-backend ($git_dir): $e") if $e;
-		}
-		return unless $res;
-		my $dumb = serve_dumb($cgi, $git, $path);
-		ref($dumb) eq 'ARRAY' ? $res->($dumb) : $dumb->($res);
-	};
-	my $fail = sub {
-		if ($!{EAGAIN} || $!{EINTR}) {
-			select($vin, undef, undef, undef) if defined $vin;
-			# $vin is undef on async, so this is a noop on EAGAIN
-			return;
+		$rpipe = undef;
+		my $e = $pid == waitpid($pid, 0) ?
+			$? : "PID:$pid still running?";
+		if ($e) {
+			err($env, "git http-backend ($git_dir): $e");
+			drop_client($env);
 		}
-		my $e = $!;
-		$end->();
-		err($env, "git http-backend ($git_dir): $e\n");
-	};
-	my $cb = sub { # read git-http-backend output and stream to client
-		my $r = $rpipe ? $rpipe->sysread($buf, 8192, length($buf)) : 0;
-		return $fail->() unless defined $r;
-		return $end->() if $r == 0; # EOF
-		if ($fh) { # stream body from git-http-backend to HTTP client
-			$fh->write($buf);
-			$buf = '';
-			$read_disable->() if $read_disable;
-		} elsif ($buf =~ s/\A(.*?)\r\n\r\n//s) { # parse headers
-			my $h = $1;
-			my $code = 200;
-			my @h;
-			foreach my $l (split(/\r\n/, $h)) {
-				my ($k, $v) = split(/:\s*/, $l, 2);
-				if ($k =~ /\AStatus\z/i) {
-					($code) = ($v =~ /\b(\d+)\b/);
-				} else {
-					push @h, $k, $v;
-				}
-			}
-			if ($code == 403) {
-				# smart cloning disabled, serve dumbly
-				# in $end since we never undef $res in here
-			} else { # write response header:
-				$fh = $res->([ $code, \@h ]);
-				$res = undef;
-				$fh->write($buf);
-			}
-			$buf = '';
-		} # else { keep reading ... }
 	};
+
+	# Danga::Socket users, we queue up the read_enable callback to
+	# fire after pending writes are complete:
+	my $buf = '';
 	if (my $async = $env->{'pi-httpd.async'}) {
+		my $res;
+		my $q = sub {
+			$async->close;
+			$end->();
+			$res->(@_);
+		};
 		# $async is PublicInbox::HTTPD::Async->new($rpipe, $cb)
-		$rpipe = $async->($rpipe, $cb);
-		sub { ($res) = @_ } # let Danga::Socket handle the rest.
-	} else { # synchronous loop for other PSGI servers
-		$read_enable = $read_disable = undef;
-		$vin = '';
-		vec($vin, fileno($rpipe), 1) = 1;
-		sub {
-			($res) = @_;
-			while ($rpipe) { $cb->() }
-		}
+		$async = $async->($rpipe, sub {
+			my $r = sysread($rpipe, $buf, 1024, length($buf));
+			if (!defined $r || $r == 0) {
+				return $q->(r(500, 'http-backend error'));
+			}
+			$r = parse_cgi_headers(\$buf) or return;
+			if ($r->[0] == 403) {
+				return $q->(serve_dumb($cgi, $git, $path));
+			}
+			my $fh = $res->($r);
+			$fh->write($buf);
+			$buf = undef;
+			my $dst = Plack::Util::inline_object(
+				write => sub { $fh->write(@_) },
+				close => sub {
+					$end->();
+					$fh->close;
+				});
+			$async->async_pass($env->{'psgix.io'}, $dst);
+		});
+		sub { ($res) = @_ }; # let Danga::Socket handle the rest.
+	} else { # getline + close for other PSGI servers
+		my $r;
+		do {
+			$r = read($rpipe, $buf, 1024, length($buf));
+			if (!defined $r || $r == 0) {
+				return r(500, 'http-backend error');
+			}
+			$r = parse_cgi_headers(\$buf);
+		} until ($r);
+		return serve_dumb($cgi, $git, $path) if $r->[0] == 403;
+		$r->[2] = Plack::Util::inline_object(
+			close => sub { $end->() },
+			getline => sub {
+				my $ret = $buf;
+				$buf = undef;
+				defined $ret ? $ret : $rpipe->getline;
+			});
+		$r;
+
 	}
 }
 
@@ -311,4 +286,21 @@ sub input_to_file {
 	return $in;
 }
 
+sub parse_cgi_headers {
+	my ($bref) = @_;
+	$$bref =~ s/\A(.*?)\r\n\r\n//s or return;
+	my $h = $1;
+	my $code = 200;
+	my @h;
+	foreach my $l (split(/\r\n/, $h)) {
+		my ($k, $v) = split(/:\s*/, $l, 2);
+		if ($k =~ /\AStatus\z/i) {
+			($code) = ($v =~ /\b(\d+)\b/);
+		} else {
+			push @h, $k, $v;
+		}
+	}
+	[ $code, \@h ]
+}
+
 1;

[PATCH 0/2] more git-http-backend cleanups

[Eric Wong]

git-http-backend (and all of our long-running processes) is
likely to encounter dropped connections.  Ensure we release
resources properly when a client hits Ctrl-C or similar while
waiting on a slow git clone.

Eric Wong (2):
      git-http-backend: avoid Plack::Request parsing body
      git-http-backend: refactor to support cleanup

 lib/PublicInbox/GitHTTPBackend.pm | 84 +++++++++++++++++++--------------------
 lib/PublicInbox/HTTPD.pm          |  5 +--
 lib/PublicInbox/HTTPD/Async.pm    | 23 ++++++-----
 3 files changed, 56 insertions(+), 56 deletions(-)

[PATCH 1/2] git-http-backend: avoid Plack::Request parsing body

[Eric Wong]

Only check query parameters since there's no useful body
in there.
---
 lib/PublicInbox/GitHTTPBackend.pm | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/PublicInbox/GitHTTPBackend.pm b/lib/PublicInbox/GitHTTPBackend.pm
index 97834de..70990eb 100644
--- a/lib/PublicInbox/GitHTTPBackend.pm
+++ b/lib/PublicInbox/GitHTTPBackend.pm
@@ -42,7 +42,7 @@ sub r ($;$) {
 sub serve {
 	my ($cgi, $git, $path) = @_;
 
-	my $service = $cgi->param('service') || '';
+	my $service = $cgi->query_parameters->get('service') || '';
 	if ($service =~ /\Agit-\w+-pack\z/ || $path =~ /\Agit-\w+-pack\z/) {
 		my $ok = serve_smart($cgi, $git, $path);
 		return $ok if $ok;

[PATCH 2/2] git-http-backend: refactor to support cleanup

[Eric Wong]

We will have clients dropping connections during long clone
and fetch operations; so do not retain references holding
backend processes once we detect a client has dropped.
---
 lib/PublicInbox/GitHTTPBackend.pm | 82 +++++++++++++++++++--------------------
 lib/PublicInbox/HTTPD.pm          |  5 +--
 lib/PublicInbox/HTTPD/Async.pm    | 23 ++++++-----
 3 files changed, 55 insertions(+), 55 deletions(-)

diff --git a/lib/PublicInbox/GitHTTPBackend.pm b/lib/PublicInbox/GitHTTPBackend.pm
index 70990eb..ded56b3 100644
--- a/lib/PublicInbox/GitHTTPBackend.pm
+++ b/lib/PublicInbox/GitHTTPBackend.pm
@@ -194,6 +194,7 @@ sub serve_smart {
 		return;
 	}
 	$wpipe = $in = undef;
+	my $fh;
 	my $end = sub {
 		$rpipe = undef;
 		my $e = $pid == waitpid($pid, 0) ?
@@ -202,60 +203,57 @@ sub serve_smart {
 			err($env, "git http-backend ($git_dir): $e");
 			drop_client($env);
 		}
+		$fh->close if $fh; # async-only
 	};
 
 	# Danga::Socket users, we queue up the read_enable callback to
 	# fire after pending writes are complete:
 	my $buf = '';
-	if (my $async = $env->{'pi-httpd.async'}) {
-		my $res;
-		my $q = sub {
-			$async->close;
-			$end->();
-			$res->(@_);
-		};
-		# $async is PublicInbox::HTTPD::Async->new($rpipe, $cb)
-		$async = $async->($rpipe, sub {
-			my $r = sysread($rpipe, $buf, 1024, length($buf));
-			if (!defined $r || $r == 0) {
-				return $q->(r(500, 'http-backend error'));
-			}
-			$r = parse_cgi_headers(\$buf) or return;
-			if ($r->[0] == 403) {
-				return $q->(serve_dumb($cgi, $git, $path));
-			}
-			my $fh = $res->($r);
-			$fh->write($buf);
-			$buf = undef;
-			my $dst = Plack::Util::inline_object(
-				write => sub { $fh->write(@_) },
-				close => sub {
-					$end->();
-					$fh->close;
-				});
-			$async->async_pass($env->{'psgix.io'}, $dst);
-		});
-		sub { ($res) = @_ }; # let Danga::Socket handle the rest.
-	} else { # getline + close for other PSGI servers
-		my $r;
-		do {
-			$r = read($rpipe, $buf, 1024, length($buf));
-			if (!defined $r || $r == 0) {
-				return r(500, 'http-backend error');
-			}
-			$r = parse_cgi_headers(\$buf);
-		} until ($r);
-		return serve_dumb($cgi, $git, $path) if $r->[0] == 403;
+	my $rd_hdr = sub {
+		my $r = sysread($rpipe, $buf, 1024, length($buf));
+		return if !defined($r) && ($!{EINTR} || $!{EAGAIN});
+		return r(500, 'http-backend error') unless $r;
+		$r = parse_cgi_headers(\$buf) or return;
+		$r->[0] == 403 ? serve_dumb($cgi, $git, $path) : $r;
+	};
+	my $res;
+	my $async = $env->{'pi-httpd.async'};
+	my $io = $env->{'psgix.io'};
+	my $cb = sub {
+		my $r = $rd_hdr->() or return;
+		$rd_hdr = undef;
+		if (scalar(@$r) == 3) { # error:
+			$async->close if $async;
+			return $res->($r);
+		}
+		if ($async) {
+			$fh = $res->($r);
+			return $async->async_pass($io, $fh, \$buf);
+		}
+
+		# for synchronous PSGI servers
 		$r->[2] = Plack::Util::inline_object(
-			close => sub { $end->() },
+			close => $end,
 			getline => sub {
 				my $ret = $buf;
 				$buf = undef;
 				defined $ret ? $ret : $rpipe->getline;
 			});
-		$r;
+		$res->($r);
+	};
+	sub {
+		($res) = @_;
 
-	}
+		# hopefully this doesn't break any middlewares,
+		# holding the input here is a waste of FDs and memory
+		$env->{'psgi.input'} = undef;
+
+		if ($async) {
+			$async = $async->($rpipe, $cb, $end);
+		} else { # generic PSGI
+			$cb->() while $rd_hdr;
+		}
+	};
 }
 
 sub input_to_file {
diff --git a/lib/PublicInbox/HTTPD.pm b/lib/PublicInbox/HTTPD.pm
index 78efaa5..433d6da 100644
--- a/lib/PublicInbox/HTTPD.pm
+++ b/lib/PublicInbox/HTTPD.pm
@@ -8,10 +8,7 @@ use Plack::Util;
 require PublicInbox::HTTPD::Async;
 require PublicInbox::Daemon;
 
-sub pi_httpd_async {
-	my ($io, $cb) = @_;
-	PublicInbox::HTTPD::Async->new($io, $cb);
-}
+sub pi_httpd_async { PublicInbox::HTTPD::Async->new(@_) }
 
 sub new {
 	my ($class, $sock, $app) = @_;
diff --git a/lib/PublicInbox/HTTPD/Async.pm b/lib/PublicInbox/HTTPD/Async.pm
index 8efa7a6..bd2eacb 100644
--- a/lib/PublicInbox/HTTPD/Async.pm
+++ b/lib/PublicInbox/HTTPD/Async.pm
@@ -9,32 +9,33 @@ package PublicInbox::HTTPD::Async;
 use strict;
 use warnings;
 use base qw(Danga::Socket);
-use fields qw(cb);
+use fields qw(cb cleanup);
 
 sub new {
-	my ($class, $io, $cb) = @_;
+	my ($class, $io, $cb, $cleanup) = @_;
 	my $self = fields::new($class);
 	IO::Handle::blocking($io, 0);
 	$self->SUPER::new($io);
 	$self->{cb} = $cb;
+	$self->{cleanup} = $cleanup;
 	$self->watch_read(1);
 	$self;
 }
 
 sub async_pass {
-	my ($self, $io, $fh) = @_;
+	my ($self, $io, $fh, $bref) = @_;
 	my $restart_read = sub { $self->watch_read(1) };
-
 	# In case the client HTTP connection ($io) dies, it
 	# will automatically close this ($self) object.
 	$io->{forward} = $self;
+	$fh->write($$bref);
 	$self->{cb} = sub {
-		my $r = sysread($self->{sock}, my $buf, 8192);
+		my $r = sysread($self->{sock}, $$bref, 8192);
 		if ($r) {
-			$fh->write($buf);
+			$fh->write($$bref);
 			if ($io->{write_buf_size}) {
 				$self->watch_read(0);
-				$io->write($restart_read);
+				$io->write($restart_read); # D::S::write
 			}
 			return; # stay in watch_read
 		} elsif (!defined $r) {
@@ -42,9 +43,9 @@ sub async_pass {
 		}
 
 		# Done! Error handling will happen in $fh->close
+		# called by the {cleanup} handler
 		$io->{forward} = undef;
 		$self->close;
-		$fh->close;
 	}
 }
 
@@ -55,8 +56,12 @@ sub sysread { shift->{sock}->sysread(@_) }
 
 sub close {
 	my $self = shift;
-	$self->{cb} = undef;
+	my $cleanup = $self->{cleanup};
+	$self->{cleanup} = $self->{cb} = undef;
 	$self->SUPER::close(@_);
+
+	# we defer this to the next timer loop since close is deferred
+	Danga::Socket->AddTimer(0, $cleanup) if $cleanup;
 }
 
 # do not let ourselves be closed during graceful termination