[PATCH 0/4] -httpd static file improvements

[PATCH 0/4] -httpd static file improvements

[Eric Wong]

Serving large static files to slow clients could lead to
public-inbox-httpd buffering data already in static files again
into a temporary file.

This was inefficient, and solving it in a generic way could
actually break other PSGI servers since their sendfile
optimizations don't handle 206 (partial content) responses
correctly.

So we make minor changes to the way PublicInbox::DS handles
write buffers and inject static files, offsets, and length
limits directly into the {wbuf} queue.

Eric Wong (4):
  http: eliminate short-lived cyclic ref for psgix.io
  wwwstatic: offload error handling to PSGI server
  ds: tmpio: store offsets per-buffer
  wwwstatic: wire up buffer bypass for -httpd

 lib/PublicInbox/DS.pm        | 36 ++++++++++----------
 lib/PublicInbox/HTTP.pm      |  7 ++--
 lib/PublicInbox/WwwStatic.pm | 65 +++++++++++++++++++++---------------
 3 files changed, 61 insertions(+), 47 deletions(-)

[PATCH 1/4] http: eliminate short-lived cyclic ref for psgix.io

[Eric Wong]

While there is no known actual leak due to reference cycles,
here, eliminating a potential source of leaks is helpful.
---
 Just a minor cleanup I noticed along the way.

 lib/PublicInbox/HTTP.pm | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index 32537153..b19a15d6 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -80,7 +80,7 @@ sub event_step { # called by PublicInbox::DS
 	# only read more requests if we've drained the write buffer,
 	# otherwise we can be buffering infinitely w/o backpressure
 
-	return read_input($self) if defined $self->{env};
+	return read_input($self) if ref($self->{env});
 	my $rbuf = $self->{rbuf} // (\(my $x = ''));
 	$self->do_read($rbuf, 8192, bytes::length($$rbuf)) or return;
 	rbuf_process($self, $rbuf);
@@ -124,7 +124,6 @@ sub read_input ($;$) {
 	my ($self, $rbuf) = @_;
 	$rbuf //= $self->{rbuf} // (\(my $x = ''));
 	my $env = $self->{env};
-	return if $env->{REMOTE_ADDR}; # in app dispatch
 	return read_input_chunked($self, $rbuf) if env_chunked($env);
 
 	# env->{CONTENT_LENGTH} (identity)
@@ -153,6 +152,7 @@ sub app_dispatch {
 	my ($self, $input, $rbuf) = @_;
 	$self->rbuf_idle($rbuf);
 	my $env = $self->{env};
+	$self->{env} = undef; # for exists() check in ->busy
 	$env->{REMOTE_ADDR} = $self->{remote_addr};
 	$env->{REMOTE_PORT} = $self->{remote_port};
 	if (defined(my $host = $env->{HTTP_HOST})) {
@@ -455,7 +455,6 @@ sub quit {
 
 sub close {
 	my $self = $_[0];
-	delete $self->{env}; # prevent circular references
 	if (my $forward = delete $self->{forward}) {
 		eval { $forward->close };
 		err($self, "forward ->close error: $@") if $@;
@@ -466,7 +465,7 @@ sub close {
 # for graceful shutdown in PublicInbox::Daemon:
 sub busy () {
 	my ($self) = @_;
-	($self->{rbuf} || $self->{env} || $self->{wbuf});
+	($self->{rbuf} || exists($self->{env}) || $self->{wbuf});
 }
 
 # Chunked and Identity packages are used for writing responses.

[PATCH 2/4] wwwstatic: offload error handling to PSGI server

[Eric Wong]

The PSGI server needs to account for ->getline failing
due to disk failures or truncated files, anyways.  So
just die() ourselves and let the PSGI server log and
drop the client.
---
 lib/PublicInbox/WwwStatic.pm | 11 +----------
 1 file changed, 1 insertion(+), 10 deletions(-)

diff --git a/lib/PublicInbox/WwwStatic.pm b/lib/PublicInbox/WwwStatic.pm
index 917049bb..e1f536f3 100644
--- a/lib/PublicInbox/WwwStatic.pm
+++ b/lib/PublicInbox/WwwStatic.pm
@@ -169,16 +169,7 @@ sub getline {
 		return $buf;
 	}
 	my $m = defined $r ? "EOF with $len bytes left" : "read error: $!";
-	my $env = $self->{env};
-	$env->{'psgi.errors'}->print("$self->{path} $m\n");
-
-	# drop the client on error
-	if (my $io = $env->{'psgix.io'}) {
-		$io->close; # this is likely PublicInbox::DS::close
-	} else { # for some PSGI servers w/o psgix.io
-		die "dropping client socket\n";
-	}
-	undef;
+	die "$self->{path} $m, dropping client socket\n";
 }
 
 sub close {} # noop, called by PSGI server, just let everything go out-of-scope

[PATCH 3/4] ds: tmpio: store offsets per-buffer

[Eric Wong]

We want to be able to inject existing file handles + offsets and
even lengths into this in the future, without going through the
->getline interface[1]

We also switch to using a 64K buffer size since we can safely
discard whatever got truncated on write and full writes can help
negotiate a larger TCP window for high-latency, high-bandwidth
links.

While we're at it, make it obvious that we're using O_APPEND for
our tmpfile() interface so we can seek freely for reading while
the writer always prints to the end of the file.

[1] the getline interface for serving static files may result
    in us buffering on-FS data into another temporary file,
    which is a waste.
---
 lib/PublicInbox/DS.pm | 36 +++++++++++++++++++-----------------
 1 file changed, 19 insertions(+), 17 deletions(-)

diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index 970061fd..a9ac7fcd 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -18,7 +18,7 @@ use strict;
 use bytes;
 use POSIX qw(WNOHANG);
 use IO::Handle qw();
-use Fcntl qw(SEEK_SET :DEFAULT);
+use Fcntl qw(SEEK_SET :DEFAULT O_APPEND);
 use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC);
 use parent qw(Exporter);
 our @EXPORT_OK = qw(now msg_more);
@@ -31,8 +31,8 @@ use PublicInbox::Tmpfile;
 
 use fields ('sock',              # underlying socket
             'rbuf',              # scalarref, usually undef
-            'wbuf', # arrayref of coderefs or GLOB refs (autovivified)
-            'wbuf_off',  # offset into first element of wbuf to start writing at
+            'wbuf', # arrayref of coderefs or tmpio (autovivified))
+                    # (tmpio = [ GLOB, offset, [ length ] ])
             );
 
 use Errno qw(EAGAIN EINVAL);
@@ -392,11 +392,13 @@ sub close {
 }
 
 # portable, non-thread-safe sendfile emulation (no pread, yet)
-sub psendfile ($$$) {
-    my ($sock, $fh, $off) = @_;
+sub send_tmpio ($$) {
+    my ($sock, $tmpio) = @_;
 
-    seek($fh, $$off, SEEK_SET) or return;
-    defined(my $to_write = read($fh, my $buf, 16384)) or return;
+    seek($tmpio->[0], $tmpio->[1], SEEK_SET) or return;
+    my $n = $tmpio->[2] // 65536;
+    $n = 65536 if $n > 65536;
+    defined(my $to_write = read($tmpio->[0], my $buf, $n)) or return;
     my $written = 0;
     while ($to_write > 0) {
         if (defined(my $w = syswrite($sock, $buf, $to_write, $written))) {
@@ -407,7 +409,8 @@ sub psendfile ($$$) {
             last;
         }
     }
-    $$off += $written;
+    $tmpio->[1] += $written; # offset
+    $tmpio->[2] -= $written if defined($tmpio->[2]); # length
     $written;
 }
 
@@ -424,9 +427,8 @@ sub flush_write ($) {
 next_buf:
     while (my $bref = $wbuf->[0]) {
         if (ref($bref) ne 'CODE') {
-            my $off = delete($self->{wbuf_off}) // 0;
             while ($sock) {
-                my $w = psendfile($sock, $bref, \$off);
+                my $w = send_tmpio($sock, $bref); # bref is tmpio
                 if (defined $w) {
                     if ($w == 0) {
                         shift @$wbuf;
@@ -434,13 +436,12 @@ next_buf:
                     }
                 } elsif ($! == EAGAIN) {
                     epwait($sock, epbit($sock, EPOLLOUT) | EPOLLONESHOT);
-                    $self->{wbuf_off} = $off;
                     return 0;
                 } else {
                     return $self->close;
                 }
             }
-        } else { #($ref eq 'CODE') {
+        } else { #(ref($bref) eq 'CODE') {
             shift @$wbuf;
             my $before = scalar(@$wbuf);
             $bref->($self);
@@ -490,12 +491,13 @@ sub drop {
 # PerlIO::mmap or PerlIO::scalar if needed
 sub tmpio ($$$) {
     my ($self, $bref, $off) = @_;
-    my $fh = tmpfile('wbuf', $self->{sock}, 1) or
+    my $fh = tmpfile('wbuf', $self->{sock}, O_APPEND) or
         return drop($self, "tmpfile $!");
     $fh->autoflush(1);
+    binmode $fh, ':unix'; # reduce syscalls for read() >8192 bytes
     my $len = bytes::length($$bref) - $off;
     $fh->write($$bref, $len, $off) or return drop($self, "write ($len): $!");
-    $fh
+    [ $fh, 0 ] # [1] = offset, [2] = length, not set by us
 }
 
 =head2 C<< $obj->write( $data ) >>
@@ -524,9 +526,9 @@ sub write {
         if ($ref eq 'CODE') {
             push @$wbuf, $bref;
         } else {
-            my $last = $wbuf->[-1];
-            if (ref($last) eq 'GLOB') { # append to tmp file buffer
-                $last->print($$bref) or return drop($self, "print: $!");
+            my $tmpio = $wbuf->[-1];
+            if ($tmpio && !defined($tmpio->[2])) { # append to tmp file buffer
+                $tmpio->[0]->print($$bref) or return drop($self, "print: $!");
             } else {
                 my $tmpio = tmpio($self, $bref, 0) or return 0;
                 push @$wbuf, $tmpio;

Re: [PATCH 3/4] ds: tmpio: store offsets per-buffer

[Eric Wong]

Eric Wong <e@yhbt.net> wrote:
> diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
> index 970061fd..a9ac7fcd 100644
> --- a/lib/PublicInbox/DS.pm
> +++ b/lib/PublicInbox/DS.pm
> @@ -490,12 +491,13 @@ sub drop {
>  # PerlIO::mmap or PerlIO::scalar if needed
>  sub tmpio ($$$) {
>      my ($self, $bref, $off) = @_;
> -    my $fh = tmpfile('wbuf', $self->{sock}, 1) or
> +    my $fh = tmpfile('wbuf', $self->{sock}, O_APPEND) or
>          return drop($self, "tmpfile $!");
>      $fh->autoflush(1);
> +    binmode $fh, ':unix'; # reduce syscalls for read() >8192 bytes

That binmode :unix call triggers a leak in Perl[1], going to
have to squash the patch below to workaround it:

diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index a9ac7fcd..c76a5038 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -494,7 +494,6 @@ sub tmpio ($$$) {
     my $fh = tmpfile('wbuf', $self->{sock}, O_APPEND) or
         return drop($self, "tmpfile $!");
     $fh->autoflush(1);
-    binmode $fh, ':unix'; # reduce syscalls for read() >8192 bytes
     my $len = bytes::length($$bref) - $off;
     $fh->write($$bref, $len, $off) or return drop($self, "write ($len): $!");
     [ $fh, 0 ] # [1] = offset, [2] = length, not set by us


... And perhaps just use sysread() instead of read(), since
I'm not seeing a good reason to keep compatibility with
PerlIO::scalar for buffering, after all.

[1] http://nntp.perl.org/group/perl.perl5.porters/256918

[PATCH 4/4] wwwstatic: wire up buffer bypass for -httpd

[Eric Wong]

This prevents public-inbox-httpd from buffering ->getline
results from a static file into another temporary file when
writing to slow clients.  Instead we inject the static file
ref with offsets and length directly into the {wbuf} queue.

It took me a while to decide to go this route, some
rejected ideas:

1. Using Plack::Util::set_io_path and having PublicInbox::HTTP
   serve the result directly.  This is compatible with what
   some other PSGI servers do using sendfile.  However, neither
   Starman or Twiggy currently use sendfile for partial responses.

2. Parsing the Content-Range response header for offsets and
   lengths to use with set_io_path for partial responses.

These rejected ideas required increasing the complexity of HTTP
response writing in PublicInbox::HTTP in the common, non-static
file cases.  Instead, we made minor changes to the colder write
buffering path of PublicInbox::DS and leave the hot paths
untouched.

We still support generic PSGI servers via ->getline.  However,
since we don't know the characteristics of other PSGI servers,
we no longer do a 64K initial read in an attempt to negotiate a
larger TCP window.
---
 lib/PublicInbox/WwwStatic.pm | 54 +++++++++++++++++++++++++-----------
 1 file changed, 38 insertions(+), 16 deletions(-)

diff --git a/lib/PublicInbox/WwwStatic.pm b/lib/PublicInbox/WwwStatic.pm
index e1f536f3..174f4752 100644
--- a/lib/PublicInbox/WwwStatic.pm
+++ b/lib/PublicInbox/WwwStatic.pm
@@ -51,7 +51,24 @@ sub r ($;$) {
 	  [ $msg ] ]
 }
 
-sub prepare_range {
+sub getline_response ($$$$$) {
+	my ($env, $in, $off, $len, $path) = @_;
+	my $r = bless {}, __PACKAGE__;
+	if ($env->{'pi-httpd.async'}) { # public-inbox-httpd-only mode
+		$env->{'psgix.no-compress'} = 1; # do not chunk response
+		%$r = ( bypass => [$in, $off, $len, $env->{'psgix.io'}] );
+	} else {
+		%$r = ( in => $in, off => $off, len => $len, path => $path );
+	}
+	$r;
+}
+
+sub getline_generic ($$$$) {
+	my ($in, $off, $len, $p) = @_;
+	bless { in => $in, off => $off, len => $len, path => $p }, __PACKAGE__;
+}
+
+sub setup_range {
 	my ($env, $in, $h, $beg, $end, $size) = @_;
 	my $code = 200;
 	my $len = $size;
@@ -81,9 +98,6 @@ sub prepare_range {
 		if ($len <= 0) {
 			$code = 416;
 		} else {
-			if ($in) {
-				sysseek($in, $beg, SEEK_SET) or return r(500);
-			}
 			push @$h, qw(Accept-Ranges bytes Content-Range);
 			push @$h, "bytes $beg-$end/$size";
 
@@ -95,7 +109,7 @@ sub prepare_range {
 		push @$h, 'Content-Range', "bytes */$size";
 		return [ 416, $h, [] ];
 	}
-	($code, $len);
+	($code, $beg, $len);
 }
 
 # returns a PSGI arrayref response iff .gz and non-.gz mtimes match
@@ -131,6 +145,7 @@ sub response ($$$;$) {
 			return r(500);
 		}
 		return r(404) unless -f $in;
+		binmode $in, ':unix'; # reduce syscalls for read() >8192 bytes
 	}
 	my $size = -s _; # bare "_" reuses "struct stat" from "-f" above
 	my $mtime = time2str((stat(_))[9]);
@@ -142,30 +157,37 @@ sub response ($$$;$) {
 	my $len = $size;
 	my $code = 200;
 	push @$h, 'Content-Type', $type;
+	my $off = 0;
 	if (($env->{HTTP_RANGE} || '') =~ /\bbytes=([0-9]*)-([0-9]*)\z/) {
-		($code, $len) = prepare_range($env, $in, $h, $1, $2, $size);
+		($code, $off, $len) = setup_range($env, $in, $h, $1, $2, $size);
 		return $code if ref($code);
 	}
 	push @$h, 'Content-Length', $len, 'Last-Modified', $mtime;
-	my $body = $in ? bless {
-		initial_rd => 65536,
-		len => $len,
-		in => $in,
-		path => $path,
-		env => $env,
-	}, __PACKAGE__ : [];
-	[ $code, $h, $body ];
+	[ $code, $h, $in ? getline_response($env, $in, $off, $len, $path) : [] ]
 }
 
 # called by PSGI servers on each response chunk:
 sub getline {
 	my ($self) = @_;
+
+	# avoid buffering, by becoming the buffer! (public-inbox-httpd)
+	if (my $tmpio = delete $self->{bypass}) {
+		my $http = pop @$tmpio; # PublicInbox::HTTP
+		push @{$http->{wbuf}}, $tmpio; # [ $in, $off, $len ]
+		$http->flush_write;
+		return; # undef, EOF
+	}
+
+	# generic PSGI runs this:
 	my $len = $self->{len} or return; # undef, tells server we're done
-	my $n = delete($self->{initial_rd}) // 8192;
+	my $n = 8192;
 	$n = $len if $len < $n;
-	my $r = sysread($self->{in}, my $buf, $n);
+	seek($self->{in}, $self->{off}, SEEK_SET) or
+			die "seek ($self->{path}): $!";
+	my $r = read($self->{in}, my $buf, $n);
 	if (defined $r && $r > 0) { # success!
 		$self->{len} = $len - $r;
+		$self->{off} += $r;
 		return $buf;
 	}
 	my $m = defined $r ? "EOF with $len bytes left" : "read error: $!";

Re: [PATCH 4/4] wwwstatic: wire up buffer bypass for -httpd

[Eric Wong]

Eric Wong <e@yhbt.net> wrote:
> diff --git a/lib/PublicInbox/WwwStatic.pm b/lib/PublicInbox/WwwStatic.pm
> index e1f536f3..174f4752 100644
> --- a/lib/PublicInbox/WwwStatic.pm
> +++ b/lib/PublicInbox/WwwStatic.pm

> @@ -131,6 +145,7 @@ sub response ($$$;$) {
>  			return r(500);
>  		}
>  		return r(404) unless -f $in;
> +		binmode $in, ':unix'; # reduce syscalls for read() >8192 bytes
>  	}
>  	my $size = -s _; # bare "_" reuses "struct stat" from "-f" above
>  	my $mtime = time2str((stat(_))[9]);

That binmode would also lead to leaks.  Will squash to workaround
current/old Perls: http://nntp.perl.org/group/perl.perl5.porters/256935

diff --git a/lib/PublicInbox/WwwStatic.pm b/lib/PublicInbox/WwwStatic.pm
index 174f4752..d75c0076 100644
--- a/lib/PublicInbox/WwwStatic.pm
+++ b/lib/PublicInbox/WwwStatic.pm
@@ -145,7 +145,6 @@ sub response ($$$;$) {
 			return r(500);
 		}
 		return r(404) unless -f $in;
-		binmode $in, ':unix'; # reduce syscalls for read() >8192 bytes
 	}
 	my $size = -s _; # bare "_" reuses "struct stat" from "-f" above
 	my $mtime = time2str((stat(_))[9]);

Re: [PATCH 4/4] wwwstatic: wire up buffer bypass for -httpd

[Eric Wong]

Eric Wong <e@yhbt.net> wrote:
> diff --git a/lib/PublicInbox/WwwStatic.pm b/lib/PublicInbox/WwwStatic.pm
> index e1f536f3..174f4752 100644
> --- a/lib/PublicInbox/WwwStatic.pm
> +++ b/lib/PublicInbox/WwwStatic.pm
> @@ -51,7 +51,24 @@ sub r ($;$) {
>  	  [ $msg ] ]
>  }
>  
> -sub prepare_range {
> +sub getline_response ($$$$$) {
> +	my ($env, $in, $off, $len, $path) = @_;
> +	my $r = bless {}, __PACKAGE__;
> +	if ($env->{'pi-httpd.async'}) { # public-inbox-httpd-only mode
> +		$env->{'psgix.no-compress'} = 1; # do not chunk response
> +		%$r = ( bypass => [$in, $off, $len, $env->{'psgix.io'}] );
> +	} else {
> +		%$r = ( in => $in, off => $off, len => $len, path => $path );
> +	}
> +	$r;
> +}
> +
> +sub getline_generic ($$$$) {
> +	my ($in, $off, $len, $p) = @_;
> +	bless { in => $in, off => $off, len => $len, path => $p }, __PACKAGE__;
> +}

Oops, getline_generic isn't used anywhere, since I inlined it
into getline_response.  Going to squash the following in:

diff --git a/lib/PublicInbox/WwwStatic.pm b/lib/PublicInbox/WwwStatic.pm
index d75c0076..60a71d8d 100644
--- a/lib/PublicInbox/WwwStatic.pm
+++ b/lib/PublicInbox/WwwStatic.pm
@@ -63,11 +63,6 @@ sub getline_response ($$$$$) {
 	$r;
 }
 
-sub getline_generic ($$$$) {
-	my ($in, $off, $len, $p) = @_;
-	bless { in => $in, off => $off, len => $len, path => $p }, __PACKAGE__;
-}
-
 sub setup_range {
 	my ($env, $in, $h, $beg, $end, $size) = @_;
 	my $code = 200;