[PATCH 1/4] http: fix typo: write_buf => write_buf_size

[PATCH 1/4] http: fix typo: write_buf => write_buf_size

[Eric Wong]

Otherwise, we get deep recursion as we keep calling
recursively on giant responses
---
 lib/PublicInbox/HTTP.pm | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index d523bd4..6576bf6 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -228,7 +228,7 @@ sub response_write {
 				my $buf = $body->getline;
 				if (defined $buf) {
 					$write->($buf);
-					if ($self->{write_buf}) {
+					if ($self->{write_buf_size}) {
 						$body->watch_read(0);
 						$self->write($restart_read);
 					}
@@ -243,7 +243,7 @@ sub response_write {
 				local $/ = \8192;
 				while (defined(my $buf = $body->getline)) {
 					$write->($buf);
-					if ($self->{write_buf}) {
+					if ($self->{write_buf_size}) {
 						$self->write($pull);
 						return;
 					}
-- 
2.8.0.rc2.38.gfdf5e7f

[PATCH 2/4] http: pass reference to Danga::Socket::write

[Eric Wong]

This can avoid an expensive copy.
---
 lib/PublicInbox/HTTP.pm | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index 6576bf6..4eb1448 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -200,7 +200,7 @@ sub response_write {
 	my $alive = response_header_write($self, $env, $res);
 
 	# middlewares such as Deflater may write empty strings
-	my $write = sub { $self->write($_[0]) if $_[0] ne '' };
+	my $write = sub { $self->write(\($_[0])) if $_[0] ne '' };
 	my $close = sub {
 		if ($alive) {
 			$self->event_write; # watch for readability if done
-- 
2.8.0.rc2.38.gfdf5e7f

[PATCH 3/4] http: rework async_pass support

[Eric Wong]

Unfortunately, the original design did not work because
middleware to wrap the response body.
---
 lib/PublicInbox/HTTP.pm        | 18 ------------------
 lib/PublicInbox/HTTPD/Async.pm | 36 ++++++++++++++++++------------------
 2 files changed, 18 insertions(+), 36 deletions(-)

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index 4eb1448..0489298 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -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 {
diff --git a/lib/PublicInbox/HTTPD/Async.pm b/lib/PublicInbox/HTTPD/Async.pm
index 8f3a6a0..b1b5ee2 100644
--- a/lib/PublicInbox/HTTPD/Async.pm
+++ b/lib/PublicInbox/HTTPD/Async.pm
@@ -21,29 +21,29 @@ sub new {
 	$self;
 }
 
-sub async_pass { $_[0]->{cb} = $_[1] }
+sub async_pass {
+	my ($self, $io, $fh) = @_;
+	my $restart_read = sub { $self->watch_read(1) };
+	$self->{cb} = sub {
+		my $bref = $self->read(8192);
+		if (defined $bref) {
+			$fh->write($$bref);
+			if ($io->{write_buf_size}) {
+				$self->watch_read(0);
+				$io->write($restart_read);
+			}
+			return; # stay in watch_read
+		}
+		$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 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;
-
-		return unless $!{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;
-	}
-}
-
 sub close {
 	my $self = shift;
 	$self->{cb} = undef;
-- 
2.8.0.rc2.38.gfdf5e7f

[PATCH 4/4] githttpbackend: switch to async_pass

[Eric Wong]

From: Eric Wong <normalperson@yhbt.net>

---
 lib/PublicInbox/GitHTTPBackend.pm | 259 +++++++++++++-------------------------
 1 file changed, 89 insertions(+), 170 deletions(-)

diff --git a/lib/PublicInbox/GitHTTPBackend.pm b/lib/PublicInbox/GitHTTPBackend.pm
index b58cc30..a05aa66 100644
--- a/lib/PublicInbox/GitHTTPBackend.pm
+++ b/lib/PublicInbox/GitHTTPBackend.pm
@@ -8,16 +8,9 @@ use strict;
 use warnings;
 use Fcntl qw(:seek);
 use IO::File;
-use PublicInbox::Spawn qw(spawn);
+use PublicInbox::Spawn qw(popen_rd);
 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;
+use HTTP::Status qw(status_message);
 
 # n.b. serving "description" and "cloneurl" should be innocuous enough to
 # not cause problems.  serving "config" might...
@@ -38,23 +31,16 @@ my @no_cache = ('Expires', 'Fri, 01 Jan 1980 00:00:00 GMT',
 		'Pragma', 'no-cache',
 		'Cache-Control', 'no-cache, max-age=0, must-revalidate');
 
-my $nextq;
-sub do_next () {
-	my $q = $nextq;
-	$nextq = undef;
-	while (my $cb = shift @$q) {
-		$cb->(); # this may redefine nextq
-	}
-}
-
-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 {
 	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/) {
@@ -101,43 +87,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";
@@ -145,18 +97,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 {
@@ -204,20 +162,10 @@ 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 {
-		$in = input_to_file($env) or return r(500);
-	}
-	my ($rpipe, $wpipe);
-	unless (pipe($rpipe, $wpipe)) {
-		err($env, "error creating pipe: $! - going static");
-		return;
+	unless (defined $fd && $fd >= 0) {
+		$input = input_to_file($env) or return r(500);
 	}
 	my %env = %ENV;
 	# GIT_COMMITTER_NAME, GIT_COMMITTER_EMAIL
@@ -231,103 +179,57 @@ sub serve_smart {
 		my $val = $env->{$name};
 		$env{$name} = $val if defined $val;
 	}
-	my $git_dir = $git->{git_dir};
 	$env{GIT_HTTP_EXPORT_ALL} = '1';
-	$env{PATH_TRANSLATED} = "$git_dir/$path";
-	my %rdr = ( 0 => fileno($in), 1 => fileno($wpipe) );
-	my $pid = spawn([qw(git http-backend)], \%env, \%rdr);
-	unless (defined $pid) {
+	$env{PATH_TRANSLATED} = "$git->{git_dir}/$path";
+	my %rdr = ( 0 => fileno($input) );
+	my $rpipe = popen_rd([qw(git http-backend)], \%env, \%rdr);
+	unless (defined $rpipe) {
 		err($env, "error spawning: $! - going static");
 		return;
 	}
-	$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:
-	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;
-			$nr_running--;
-		}
-		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;
-		}
-		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 (my $async = $env->{'pi-httpd.async'}) {
+		my $res;
+		$async = $async->($rpipe, sub {
+			my $bref = $async->read(8192); # Danga::Socket::read
+			if (!$bref || $$bref eq '') {
+				$async->close;
+				return r(500, 'http-backend error');
 			}
-			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);
+			my $r = parse_cgi_headers($bref);
+			unless ($r) {
+				# something is wrong if http-backend trickles
+				$async->close;
+				return r(500, 'http-backend trickled!');
 			}
-			$buf = '';
-		} # else { keep reading ... }
-	};
-	if (my $async = $env->{'pi-httpd.async'}) {
-		# $async is PublicInbox::HTTPD::Async->new($rpipe, $cb)
-		$rpipe = $async->($rpipe, $cb);
+			if ($r->[0] == 403) {
+				$async->close;
+				return $res->(serve_dumb($cgi, $git, $path));
+			}
+			$async->push_back_read($bref);
+			my $fh = $res->($r);
+			$async->async_pass($env->{'psgix.io'}, $fh);
+		});
 		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->() }
-		}
+	} else { # normal PSGI servers:
+		require Plack::Util;
+		my $n = $rpipe->read(my $ibuf, 8192);
+		return r(500, 'snapshot error') unless $n;
+		my $r = parse_cgi_headers(\$ibuf) or
+				# something is wrong if http-backend trickles
+				return r(500, 'http-backend trickled!');
+		return serve_dumb($cgi, $git, $path) if $r->[0] == 403;
+		$r->[2] = Plack::Util::inline_object(
+			close => sub {
+				close $rpipe;
+				drop_client($env) if $?;
+			},
+			getline => sub {
+				my $ret = $ibuf;
+				$ibuf = undef;
+				defined $ret ? $ret : $rpipe->getline;
+			});
+		$r;
 	}
 }
 
@@ -350,4 +252,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;
-- 
2.8.0.rc2.38.gfdf5e7f