[PATCH] http: support graceful shutdown like nntp

[PATCH] http: support graceful shutdown like nntp

[Eric Wong]

HTTP responses may be long-running or requests may be slow or
pipelined.  Ensure we don't kill them off prematurely.
---
 lib/PublicInbox/HTTP.pm |  8 ++++-
 lib/PublicInbox/NNTP.pm |  1 +
 t/httpd-corner.psgi     | 20 +++++++++++
 t/httpd-corner.t        | 95 +++++++++++++++++++++++++++++++++++++++++++++----
 4 files changed, 116 insertions(+), 8 deletions(-)

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index f1016d2..928c0f2 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -103,7 +103,6 @@ sub app_dispatch ($) {
 	my ($self) = @_;
 	$self->watch_read(0);
 	my $env = $self->{env};
-	$self->{env} = undef;
 	$env->{REMOTE_ADDR} = $self->peer_ip_string; # Danga::Socket
 	$env->{REMOTE_PORT} = $self->{peer_port}; # set by peer_ip_string
 	if (my $host = $env->{HTTP_HOST}) {
@@ -169,6 +168,7 @@ sub response_write {
 		} else {
 			$self->write(sub { $self->close });
 		}
+		$self->{env} = undef;
 	};
 
 	if (defined $res->[2]) {
@@ -336,4 +336,10 @@ sub quit {
 sub event_hup { $_[0]->close }
 sub event_err { $_[0]->close }
 
+# for graceful shutdown in PublicInbox::Daemon:
+sub busy () {
+	my ($self) = @_;
+	($self->{rbuf} ne '' || $self->{env} || $self->{write_buf_size});
+}
+
 1;
diff --git a/lib/PublicInbox/NNTP.pm b/lib/PublicInbox/NNTP.pm
index 097c57e..bcce770 100644
--- a/lib/PublicInbox/NNTP.pm
+++ b/lib/PublicInbox/NNTP.pm
@@ -954,6 +954,7 @@ sub watch_read {
 	$rv;
 }
 
+# for graceful shutdown in PublicInbox::Daemon:
 sub busy () {
 	my ($self) = @_;
 	($self->{rbuf} ne '' || $self->{long_res} || $self->{write_buf_size});
diff --git a/t/httpd-corner.psgi b/t/httpd-corner.psgi
index 1947f37..0e0e21a 100644
--- a/t/httpd-corner.psgi
+++ b/t/httpd-corner.psgi
@@ -26,7 +26,27 @@ my $app = sub {
 		}
 		$code = 200;
 		push @$body, $sha1->hexdigest;
+	} elsif (my $fifo = $env->{HTTP_X_CHECK_FIFO}) {
+		if ($path eq '/slow-header') {
+			return sub {
+				open my $f, '<', $fifo or
+						die "open $fifo: $!\n";
+				my @r = <$f>;
+				$_[0]->([200, $h, \@r ]);
+			};
+		} elsif ($path eq '/slow-body') {
+			return sub {
+				my $fh = $_[0]->([200, $h]);
+				open my $f, '<', $fifo or
+						die "open $fifo: $!\n";
+				while (defined(my $l = <$f>)) {
+					$fh->write($l);
+				}
+				$fh->close;
+			};
+		}
 	}
+
 	[ $code, $h, $body ]
 };
 
diff --git a/t/httpd-corner.t b/t/httpd-corner.t
index 366e56c..4069208 100644
--- a/t/httpd-corner.t
+++ b/t/httpd-corner.t
@@ -18,7 +18,10 @@ use Cwd qw/getcwd/;
 use IO::Socket;
 use Fcntl qw(FD_CLOEXEC F_SETFD F_GETFD :seek);
 use Socket qw(SO_KEEPALIVE IPPROTO_TCP TCP_NODELAY);
+use POSIX qw(dup2 mkfifo :sys_wait_h);
 my $tmpdir = tempdir(CLEANUP => 1);
+my $fifo = "$tmpdir/fifo";
+ok(defined mkfifo($fifo, 0777), 'created FIFO');
 my $err = "$tmpdir/stderr.log";
 my $out = "$tmpdir/stdout.log";
 my $httpd = 'blib/script/public-inbox-httpd';
@@ -33,27 +36,31 @@ my %opts = (
 my $sock = IO::Socket::INET->new(%opts);
 my $pid;
 END { kill 'TERM', $pid if defined $pid };
-{
-	ok($sock, 'sock created');
-	$! = 0;
+my $spawn_httpd = sub {
+	my (@args) = @_;
 	my $fl = fcntl($sock, F_GETFD, 0);
 	ok(! $!, 'no error from fcntl(F_GETFD)');
 	is($fl, FD_CLOEXEC, 'cloexec set by default (Perl behavior)');
 	$pid = fork;
 	if ($pid == 0) {
-		use POSIX qw(dup2);
 		# pretend to be systemd
 		fcntl($sock, F_SETFD, $fl &= ~FD_CLOEXEC);
 		dup2(fileno($sock), 3) or die "dup2 failed: $!\n";
 		$ENV{LISTEN_PID} = $$;
 		$ENV{LISTEN_FDS} = 1;
-		exec $httpd, '-W0', "--stdout=$out", "--stderr=$err", $psgi;
+		exec $httpd, @args, "--stdout=$out", "--stderr=$err", $psgi;
 		die "FAIL: $!\n";
 	}
 	ok(defined $pid, 'forked httpd process successfully');
+};
+
+{
+	ok($sock, 'sock created');
 	$! = 0;
-	fcntl($sock, F_SETFD, $fl |= FD_CLOEXEC);
-	ok(! $!, 'no error from fcntl(F_SETFD)');
+	my $fl = fcntl($sock, F_GETFD, 0);
+	ok(! $!, 'no error from fcntl(F_GETFD)');
+	is($fl, FD_CLOEXEC, 'cloexec set by default (Perl behavior)');
+	$spawn_httpd->('-W0');
 }
 
 sub conn_for {
@@ -69,6 +76,58 @@ sub conn_for {
 	return $conn;
 }
 
+# graceful termination
+{
+	my $conn = conn_for($sock, 'graceful termination via slow header');
+	$conn->write("GET /slow-header HTTP/1.0\r\n" .
+			"X-Check-Fifo: $fifo\r\n\r\n");
+	open my $f, '>', $fifo or die "open $fifo: $!\n";
+	$f->autoflush(1);
+	ok(print($f "hello\n"), 'wrote something to fifo');
+	my $kpid = $pid;
+	$pid = undef;
+	is(kill('TERM', $kpid), 1, 'started graceful shutdown');
+	ok(print($f "world\n"), 'wrote else to fifo');
+	close $f or die "close fifo: $!\n";
+	$conn->read(my $buf, 8192);
+	my ($head, $body) = split(/\r\n\r\n/, $buf, 2);
+	like($head, qr!\AHTTP/1\.[01] 200 OK!, 'got 200 for slow-header');
+	is($body, "hello\nworld\n", 'read expected body');
+	is(waitpid($kpid, 0), $kpid, 'reaped httpd');
+	is($?, 0, 'no error');
+	$spawn_httpd->('-W0');
+}
+
+{
+	my $conn = conn_for($sock, 'graceful termination via slow-body');
+	$conn->write("GET /slow-body HTTP/1.0\r\n" .
+			"X-Check-Fifo: $fifo\r\n\r\n");
+	open my $f, '>', $fifo or die "open $fifo: $!\n";
+	$f->autoflush(1);
+	my $buf;
+	$conn->sysread($buf, 8192);
+	like($buf, qr!\AHTTP/1\.[01] 200 OK!, 'got 200 for slow-body');
+	like($buf, qr!\r\n\r\n!, 'finished HTTP response header');
+
+	foreach my $c ('a'..'c') {
+		$c .= "\n";
+		ok(print($f $c), 'wrote line to fifo');
+		$conn->sysread($buf, 8192);
+		is($buf, $c, 'got trickle for reading');
+	}
+	my $kpid = $pid;
+	$pid = undef;
+	is(kill('TERM', $kpid), 1, 'started graceful shutdown');
+	ok(print($f "world\n"), 'wrote else to fifo');
+	close $f or die "close fifo: $!\n";
+	$conn->sysread($buf, 8192);
+	is($buf, "world\n", 'read expected body');
+	is($conn->sysread($buf, 8192), 0, 'got EOF from server');
+	is(waitpid($kpid, 0), $kpid, 'reaped httpd');
+	is($?, 0, 'no error');
+	$spawn_httpd->('-W0');
+}
+
 sub delay { select(undef, undef, undef, shift || rand(0.02)) }
 
 my $str = 'abcdefghijklmnopqrstuvwxyz';
@@ -140,6 +199,28 @@ SKIP: {
 	}
 }
 
+{
+	my $conn = conn_for($sock, 'graceful termination during slow request');
+	$conn->write("PUT /sha1 HTTP/1.0\r\n");
+	delay();
+	$conn->write("Content-Length: $len\r\n");
+	delay();
+	$conn->write("\r\n");
+	my $kpid = $pid;
+	$pid = undef;
+	is(kill('TERM', $kpid), 1, 'started graceful shutdown');
+	delay();
+	my $n = 0;
+	foreach my $c ('a'..'z') {
+		$n += $conn->write($c);
+	}
+	is($n, $len, 'wrote alphabet');
+	$check_self->($conn);
+	is(waitpid($kpid, 0), $kpid, 'reaped httpd');
+	is($?, 0, 'no error');
+	$spawn_httpd->('-W0');
+}
+
 # various DoS attacks against the chunk parser:
 {
 	local $SIG{PIPE} = 'IGNORE';
-- 
EW