[PATCH 3/3] git-http-backend: start async API for streaming

[Eric Wong <e@80x24.org>]

git-http-backend may take a while, ensure we can process other
requests while waiting on it.  We currently do this via
Danga::Socket in public-inbox-httpd; but avoid exposing this
internal implementation detail to the PSGI interface and
instead only expose a callback via: $env->{'pi-httpd.async'}
---
 lib/PublicInbox/GitHTTPBackend.pm | 26 +++++++++++++++-----------
 public-inbox-httpd                | 31 +++++++++++++++++++++++++++++++
 2 files changed, 46 insertions(+), 11 deletions(-)

diff --git a/lib/PublicInbox/GitHTTPBackend.pm b/lib/PublicInbox/GitHTTPBackend.pm
index 3cf7857..31b9cd8 100644
--- a/lib/PublicInbox/GitHTTPBackend.pm
+++ b/lib/PublicInbox/GitHTTPBackend.pm
@@ -179,23 +179,21 @@ sub serve_smart {
 		exec(@cmd) or die 'exec `' . join(' ', @cmd). "' failed: $!\n";
 	}
 	$wpipe = $in = undef;
-	$rpipe->blocking(0);
 	$buf = '';
-	my $vin;
-	vec($vin, fileno($rpipe), 1) = 1;
-	my ($fh, $res);
+	my ($vin, $fh, $res);
 	my $fail = sub {
 		my ($e) = @_;
 		if ($e eq 'EAGAIN') {
-			select($vin, undef, undef, undef);
+			select($vin, undef, undef, undef) if defined $vin;
+			# $vin is undef on async, so this is a noop on EAGAIN
 		} else {
 			$rpipe = undef;
 			$fh->close if $fh;
 			$err->print('git http-backend error: ', $e, "\n");
 		}
 	};
-	my $cb = sub {
-		my $r = sysread($rpipe, $buf, 8192, length($buf));
+	my $cb = sub { # read git-http-backend output and stream to client
+		my $r = $rpipe ? sysread($rpipe, $buf, 8192, length($buf)) : 0;
 		return $fail->($!{EAGAIN} ? 'EAGAIN' : $!) unless defined $r;
 		if ($r == 0) { # EOF
 			$rpipe = undef;
@@ -223,10 +221,16 @@ sub serve_smart {
 			$buf = '';
 		} # else { keep reading ... }
 	};
-	sub {
-		($res) = @_;
-		while ($rpipe) { $cb->() }
-	};
+	if (my $async = $env->{'pi-httpd.async'}) {
+		$async->($rpipe, $cb);
+		sub { ($res) = @_ } # let Danga::Socket handle the rest.
+	} else { # synchronous loop
+		vec($vin, fileno($rpipe), 1) = 1;
+		sub {
+			($res) = @_;
+			while ($rpipe) { $cb->() }
+		}
+	}
 }
 
 1;
diff --git a/public-inbox-httpd b/public-inbox-httpd
index 3635c9a..cf07628 100644
--- a/public-inbox-httpd
+++ b/public-inbox-httpd
@@ -53,11 +53,38 @@ daemon_run('0.0.0.0:8080', $refresh,
 	});
 
 1;
+
+package PublicInbox::HTTPD::Async;
+use strict;
+use warnings;
+use base qw(Danga::Socket);
+use fields qw(cb);
+
+sub new {
+	my ($class, $io, $cb) = @_;
+	my $self = fields::new($class);
+	$io->blocking(0);
+	$self->SUPER::new($io);
+	$self->{cb} = $cb;
+	$self->watch_read(1);
+	$self;
+}
+
+sub event_read { $_[0]->{cb}->() }
+sub event_hup { $_[0]->{cb}->() }
+
+1;
+
 package PublicInbox::HTTPD;
 use strict;
 use warnings;
 use Plack::Util;
 
+sub pi_httpd_async {
+	my ($io, $cb) = @_;
+	PublicInbox::HTTPD::Async->new($io, $cb);
+}
+
 sub new {
 	my ($class, $sock, $app) = @_;
 	my $n = getsockname($sock) or die "not a socket: $sock $!\n";
@@ -85,6 +112,10 @@ sub new {
 		'psgi.multiprocess' => Plack::Util::TRUE,
 		'psgix.harakiri'=> Plack::Util::FALSE,
 		'psgix.input.buffered' => Plack::Util::TRUE,
+		'pi-httpd.async' => do {
+			no warnings 'once';
+			*pi_httpd_async
+		},
 	);
 	bless {
 		err => \*STDERR,
-- 
EW