[PATCH 0/3] http: another round EPIPE fixes

[PATCH 0/3] http: another round EPIPE fixes

[Eric Wong]

Hopefully this is end of resource leaks on prematurely aborted
client connections.

Eric Wong (3):
      httpd/async: prevent circular reference
      http: avoid circular reference for getline responses
      git-http-backend: fix aborts for generic PSGI clone

 lib/PublicInbox/GetlineBody.pm    | 31 +++++++++++++++++++++++++++
 lib/PublicInbox/GitHTTPBackend.pm | 13 ++++-------
 lib/PublicInbox/HTTP.pm           | 45 ++++++++++++++++++++++++---------------
 lib/PublicInbox/HTTPD/Async.pm    |  4 +++-
 4 files changed, 66 insertions(+), 27 deletions(-)

[PATCH 1/3] httpd/async: prevent circular reference

[Eric Wong]

We must avoid circular references which can cause leaks in
long-running processes.  This callback is dangerous since
it may never be called to properly terminate everything.
---
 lib/PublicInbox/HTTPD/Async.pm | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/lib/PublicInbox/HTTPD/Async.pm b/lib/PublicInbox/HTTPD/Async.pm
index 47ba27d..b00e637 100644
--- a/lib/PublicInbox/HTTPD/Async.pm
+++ b/lib/PublicInbox/HTTPD/Async.pm
@@ -10,6 +10,7 @@ use strict;
 use warnings;
 use base qw(Danga::Socket);
 use fields qw(cb cleanup);
+use Scalar::Util qw(weaken);
 require PublicInbox::EvCleanup;
 
 sub new {
@@ -25,11 +26,12 @@ sub new {
 
 sub async_pass {
 	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);
+	weaken($self);
+	my $restart_read = sub { $self->watch_read(1) };
 	$self->{cb} = sub {
 		my $r = sysread($self->{sock}, $$bref, 8192);
 		if ($r) {

[PATCH 2/3] http: avoid circular reference for getline responses

[Eric Wong]

Lightly tested, this seems to work when mass-aborting
responses.  Will still need to automate the testing...
---
 lib/PublicInbox/HTTP.pm | 45 ++++++++++++++++++++++++++++-----------------
 1 file changed, 28 insertions(+), 17 deletions(-)

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index 6aae5c8..0454f60 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -11,11 +11,12 @@ package PublicInbox::HTTP;
 use strict;
 use warnings;
 use base qw(Danga::Socket);
-use fields qw(httpd env rbuf input_left remote_addr remote_port forward);
+use fields qw(httpd env rbuf input_left remote_addr remote_port forward pull);
 use Fcntl qw(:seek);
 use Plack::HTTPParser qw(parse_http_request); # XS or pure Perl
 use HTTP::Status qw(status_message);
 use HTTP::Date qw(time2str);
+use Scalar::Util qw(weaken);
 use IO::File;
 use constant {
 	CHUNK_START => -1,   # [a-f0-9]+\r\n
@@ -255,6 +256,28 @@ sub response_done ($$) {
 	}
 }
 
+sub getline_response {
+	my ($self, $body, $write, $close) = @_;
+	$self->{forward} = $body;
+	weaken($self);
+	my $pull = $self->{pull} = sub {
+		local $/ = \8192;
+		my $forward = $self->{forward};
+		while ($forward && defined(my $buf = $forward->getline)) {
+			$write->($buf);
+			last if $self->{closed};
+			if ($self->{write_buf_size}) {
+				$self->write($self->{pull});
+				return;
+			}
+		}
+		$self->{forward} = $self->{pull} = undef;
+		$forward->close if $forward; # avoid recursion
+		$close->();
+	};
+	$pull->();
+}
+
 sub response_write {
 	my ($self, $env, $res) = @_;
 	my $alive = response_header_write($self, $env, $res);
@@ -266,21 +289,7 @@ sub response_write {
 			$write->($_) foreach @$body;
 			$close->();
 		} else {
-			my $pull;
-			$pull = sub {
-				local $/ = \8192;
-				while (defined(my $buf = $body->getline)) {
-					$write->($buf);
-					if ($self->{write_buf_size}) {
-						$self->write($pull);
-						return;
-					}
-				}
-				$pull = undef;
-				$body->close();
-				$close->();
-			};
-			$pull->();
+			getline_response($self, $body, $write, $close);
 		}
 	} else {
 		# this is returned to the calling application:
@@ -445,8 +454,10 @@ sub event_err { $_[0]->close }
 sub close {
 	my $self = shift;
 	my $forward = $self->{forward};
+	my $env = $self->{env};
+	delete $env->{'psgix.io'} if $env; # prevent circular referernces
+	$self->{pull} = $self->{forward} = $self->{env} = undef;
 	$forward->close if $forward;
-	$self->{forward} = $self->{env} = undef;
 	$self->SUPER::close(@_);
 }
 

[PATCH 3/3] git-http-backend: fix aborts for generic PSGI clone

[Eric Wong]

We need to avoid circular references in the generic PSGI layer,
do it by abusing DESTROY.
---
 lib/PublicInbox/GetlineBody.pm    | 31 +++++++++++++++++++++++++++++++
 lib/PublicInbox/GitHTTPBackend.pm | 13 ++++---------
 2 files changed, 35 insertions(+), 9 deletions(-)
 create mode 100644 lib/PublicInbox/GetlineBody.pm

diff --git a/lib/PublicInbox/GetlineBody.pm b/lib/PublicInbox/GetlineBody.pm
new file mode 100644
index 0000000..4f8765b
--- /dev/null
+++ b/lib/PublicInbox/GetlineBody.pm
@@ -0,0 +1,31 @@
+# Copyright (C) 2016 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# Wrap a pipe or file for PSGI streaming response bodies and calls the
+# end callback when the object goes out-of-scope.
+# This depends on rpipe being _blocking_ on getline.
+package PublicInbox::GetlineBody;
+use strict;
+use warnings;
+
+sub new {
+	my ($class, $rpipe, $end, $buf) = @_;
+	bless { rpipe => $rpipe, end => $end, buf => $buf }, $class;
+}
+
+sub DESTROY { $_[0]->close }
+
+sub getline {
+	my ($self) = @_;
+	my $buf = delete $self->{buf};
+	defined $buf ? $buf : $self->{rpipe}->getline;
+}
+
+sub close {
+	my ($self) = @_;
+	delete $self->{rpipe};
+	my $end = delete $self->{end} or return;
+	$end->();
+}
+
+1;
diff --git a/lib/PublicInbox/GitHTTPBackend.pm b/lib/PublicInbox/GitHTTPBackend.pm
index 9464cb4..fd7afbc 100644
--- a/lib/PublicInbox/GitHTTPBackend.pm
+++ b/lib/PublicInbox/GitHTTPBackend.pm
@@ -186,7 +186,7 @@ sub serve_smart {
 	my $x = PublicInbox::Qspawn->new([qw(git http-backend)], \%env, \%rdr);
 	my ($fh, $rpipe);
 	my $end = sub {
-		$rpipe = undef;
+		close $rpipe if $rpipe && !$fh; # generic PSGI
 		if (my $err = $x->finish) {
 			err($env, "git http-backend ($git_dir): $err");
 			drop_client($env);
@@ -201,7 +201,7 @@ sub serve_smart {
 		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 = parse_cgi_headers(\$buf) or return; # incomplete headers
 		$r->[0] == 403 ? serve_dumb($cgi, $git, $path) : $r;
 	};
 	my $res;
@@ -220,13 +220,8 @@ sub serve_smart {
 		}
 
 		# for synchronous PSGI servers
-		$r->[2] = Plack::Util::inline_object(
-			close => $end,
-			getline => sub {
-				my $ret = $buf;
-				$buf = undef;
-				defined $ret ? $ret : $rpipe->getline;
-			});
+		require PublicInbox::GetlineBody;
+		$r->[2] = PublicInbox::GetlineBody->new($rpipe, $end, $buf);
 		$res->($r);
 	};
 	sub {

[PATCH 4/3] httpd/async: do not needlessly weaken

[Eric Wong]

The restart_read callback has no chance of circular reference,
and weakening $self before we create it can cause $self to
be undefined inside the callback (seen during stress testing).

Fixes: 395406118cb2 ("httpd/async: prevent circular reference")
---
 lib/PublicInbox/HTTPD/Async.pm | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/lib/PublicInbox/HTTPD/Async.pm b/lib/PublicInbox/HTTPD/Async.pm
index b00e637..add07ce 100644
--- a/lib/PublicInbox/HTTPD/Async.pm
+++ b/lib/PublicInbox/HTTPD/Async.pm
@@ -24,14 +24,19 @@ sub new {
 	$self;
 }
 
+sub restart_read_cb ($) {
+	my ($self) = @_;
+	sub { $self->watch_read(1) }
+}
+
 sub async_pass {
 	my ($self, $io, $fh, $bref) = @_;
 	# In case the client HTTP connection ($io) dies, it
 	# will automatically close this ($self) object.
 	$io->{forward} = $self;
 	$fh->write($$bref);
+	my $restart_read = restart_read_cb($self);
 	weaken($self);
-	my $restart_read = sub { $self->watch_read(1) };
 	$self->{cb} = sub {
 		my $r = sysread($self->{sock}, $$bref, 8192);
 		if ($r) {