[PATCH 0/2] http: start migrating to pull-based I/O

[PATCH 0/2] http: start migrating to pull-based I/O

[Eric Wong]

There'll be more conversions coming, but this should decouple
some of our PSGI code from Danga::Socket internals while also
providing better backpressure handling inside PublicInbox::HTTP
to avoid excessive buffering.

Eric Wong (2):
      http: reduce over-buffering for getline responses
      mbox: switch generation over to pull model

 lib/PublicInbox/Git.pm  |   2 +
 lib/PublicInbox/HTTP.pm |  26 ++++++--
 lib/PublicInbox/Mbox.pm | 157 +++++++++++++++---------------------------------
 3 files changed, 74 insertions(+), 111 deletions(-)

[PATCH 1/2] http: reduce over-buffering for getline responses

[Eric Wong]

By switching to a "pull"-based I/O model for reading
application responses, we should be able to throttle
buffering to slow clients more effectively and avoid
wasting precious RAM.

This will also allow us to more Danga::Socket-specific
knowledge out of the PSGI application and keep it
confined to PublicInbox::HTTP.
---
 lib/PublicInbox/HTTP.pm | 26 ++++++++++++++++++++++----
 1 file changed, 22 insertions(+), 4 deletions(-)

diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index 1ef3fb3..f69056f 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -205,7 +205,7 @@ sub response_write {
 		if ($alive) {
 			$self->event_write; # watch for readability if done
 		} else {
-			$self->write(sub { $self->close });
+			Danga::Socket::write($self, sub { $self->close });
 		}
 		if (my $obj = $env->{'pi-httpd.inbox'}) {
 			# grace period for reaping resources
@@ -215,9 +215,27 @@ sub response_write {
 		$self->{env} = undef;
 	};
 
-	if (defined $res->[2]) {
-		Plack::Util::foreach($res->[2], $write);
-		$close->();
+	if (defined(my $body = $res->[2])) {
+		if (ref $body eq 'ARRAY') {
+			$write->($_) foreach @$body;
+			$close->();
+		} else {
+			my $pull;
+			$pull = sub {
+				local $/ = \8192;
+				while (defined(my $buf = $body->getline)) {
+					$write->($buf);
+					if ($self->{write_buf}) {
+						$self->write($pull);
+						return;
+					}
+				}
+				$pull = undef;
+				$body->close();
+				$close->();
+			};
+			$pull->();
+		}
 	} else {
 		# this is returned to the calling application:
 		Plack::Util::inline_object(write => $write, close => $close);
-- 
EW

[PATCH 2/2] mbox: switch generation over to pull model

[Eric Wong]

This allows us to easily provide gigantic inboxes
with proper backpressure handling for slow clients.

It also eliminates public-inbox-httpd and Danga::Socket-specific
knowledge from this class, making it easier to follow for
those used to generic PSGI applications.
---
 lib/PublicInbox/Git.pm  |   2 +
 lib/PublicInbox/Mbox.pm | 157 +++++++++++++++---------------------------------
 2 files changed, 52 insertions(+), 107 deletions(-)

diff --git a/lib/PublicInbox/Git.pm b/lib/PublicInbox/Git.pm
index d821182..473cdff 100644
--- a/lib/PublicInbox/Git.pm
+++ b/lib/PublicInbox/Git.pm
@@ -41,6 +41,7 @@ sub cat_file {
 	$self->{out}->print($obj, "\n") or fail($self, "write error: $!");
 
 	my $in = $self->{in};
+	local $/ = "\n";
 	my $head = $in->getline;
 	$head =~ / missing$/ and return undef;
 	$head =~ /^[0-9a-f]{40} \S+ (\d+)$/ or
@@ -90,6 +91,7 @@ sub check {
 	my ($self, $obj) = @_;
 	$self->_bidi_pipe(qw(--batch-check in_c out_c pid_c));
 	$self->{out_c}->print($obj, "\n") or fail($self, "write error: $!");
+	local $/ = "\n";
 	chomp(my $line = $self->{in_c}->getline);
 	my ($hex, $type, $size) = split(' ', $line);
 	return if $type eq 'missing';
diff --git a/lib/PublicInbox/Mbox.pm b/lib/PublicInbox/Mbox.pm
index 4c4b74f..40ca611 100644
--- a/lib/PublicInbox/Mbox.pm
+++ b/lib/PublicInbox/Mbox.pm
@@ -6,18 +6,11 @@
 package PublicInbox::Mbox;
 use strict;
 use warnings;
-use PublicInbox::MID qw/mid2path mid_clean/;
+use PublicInbox::MID qw/mid_clean/;
 use URI::Escape qw/uri_escape_utf8/;
+use Plack::Util;
 require Email::Simple;
 
-sub thread_mbox {
-	my ($ctx, $srch, $sfx) = @_;
-	sub {
-		my ($response) = @_; # Plack callback
-		emit_mbox($response, $ctx, $srch, $sfx);
-	}
-}
-
 sub emit1 {
 	my $simple = Email::Simple->new(pop);
 	my $ctx = pop;
@@ -84,104 +77,35 @@ sub emit_msg {
 	$fh->write($buf .= "\n");
 }
 
-sub emit_mbox {
-	my ($response, $ctx, $srch, $sfx) = @_;
-	my $type = 'mbox';
-	if ($sfx) {
-		eval { require IO::Compress::Gzip };
-		return need_gzip($response) if $@;
-		$type = 'gzip';
-	}
-
-	# http://www.iana.org/assignments/media-types/application/gzip
-	# http://www.iana.org/assignments/media-types/application/mbox
-	my $fh = $response->([200, ['Content-Type' => "application/$type"]]);
-	$fh = PublicInbox::MboxGz->new($fh) if $sfx;
-
-	require PublicInbox::Git;
-	my $mid = $ctx->{mid};
-	my $git = $ctx->{git} ||= PublicInbox::Git->new($ctx->{git_dir});
-	my %opts = (offset => 0, asc => 1);
-	my $nr;
-	do {
-		my $res = $srch->get_thread($mid, \%opts);
-		my $msgs = $res->{msgs};
-		$nr = scalar @$msgs;
-		while (defined(my $smsg = shift @$msgs)) {
-			my $msg = eval {
-				my $p = 'HEAD:'.mid2path($smsg->mid);
-				Email::Simple->new($git->cat_file($p));
-			};
-			emit_msg($ctx, $fh, $msg) if $msg;
-		}
+sub noop {}
 
-		$opts{offset} += $nr;
-	} while ($nr > 0);
+sub thread_mbox {
+	my ($ctx, $srch, $sfx) = @_;
+	eval { require IO::Compress::Gzip };
+	return sub { need_gzip(@_) } if $@;
 
-	$fh->close;
+	my $cb = sub { $srch->get_thread($ctx->{mid}, @_) };
+	# http://www.iana.org/assignments/media-types/application/gzip
+	[200, ['Content-Type' => 'application/gzip'],
+		PublicInbox::MboxGz->new($ctx, $cb) ];
 }
 
 sub emit_range {
 	my ($ctx, $range) = @_;
-	sub { _emit_range($_[0], $ctx, $range) };
-}
-
-sub _emit_range {
-	my ($res, $ctx, $range) = @_;
 
 	eval { require IO::Compress::Gzip };
-	return need_gzip($res) if $@;
+	return sub { need_gzip(@_) } if $@;
 	my $query;
 	if ($range eq 'all') { # TODO: YYYY[-MM]
 		$query = '';
 	} else {
-		$res->([404, [qw(Content-Type text/plain)], []]);
-		return;
+		return [404, [qw(Content-Type text/plain)], []];
 	}
+	my $cb = sub { $ctx->{srch}->query($query, @_) };
 
 	# http://www.iana.org/assignments/media-types/application/gzip
-	my $fh = $res->([200, [qw(Content-Type application/gzip)]]);
-	$fh = PublicInbox::MboxGz->new($fh);
-	my $env = $ctx->{cgi}->env;
-	my $srch = $ctx->{srch};
-	my $git = $ctx->{git};
-	my %opts = (offset => 0, asc => 1);
-	my $nr;
-	my $cb = sub {
-		my $res = $srch->query($query, \%opts);
-		my $msgs = $res->{msgs};
-		$nr = scalar @$msgs;
-		while (defined(my $smsg = shift @$msgs)) {
-			my $msg = eval {
-				my $p = 'HEAD:'.mid2path($smsg->mid);
-				Email::Simple->new($git->cat_file($p));
-			};
-			emit_msg($ctx, $fh, $msg) if $msg;
-		}
-
-		$opts{offset} += $nr;
-	};
-
-	$cb->(); # first part is free
-	return $fh->close if $nr == 0;
-
-	if ($env->{'pi-httpd.async'}) {
-		my $io = $env->{'psgix.io'} or die "no IO";
-		my $next;
-		$next = sub {
-			$cb->();
-			if ($nr > 0) {
-				$io->write($next);
-			} else {
-				$next = undef;
-				$fh->close;
-			}
-		};
-		$io->write($next); # Danga::Socket::write
-		return;
-	}
-	$cb->() while ($nr > 0);
-	$fh->close;
+	[200, [qw(Content-Type application/gzip)],
+		PublicInbox::MboxGz->new($ctx, $cb) ];
 }
 
 sub need_gzip {
@@ -198,40 +122,59 @@ EOF
 
 1;
 
-# fh may not be a proper IO, so we wrap the write and close methods
-# to prevent IO::Compress::Gzip from complaining
 package PublicInbox::MboxGz;
 use strict;
 use warnings;
+use PublicInbox::MID qw(mid2path);
 
 sub new {
-	my ($class, $fh) = @_;
+	my ($class, $ctx, $cb) = @_;
 	my $buf;
 	bless {
 		buf => \$buf,
 		gz => IO::Compress::Gzip->new(\$buf),
-		fh => $fh,
+		cb => $cb,
+		ctx => $ctx,
+		msgs => [],
+		opts => { asc => 1, offset => 0 },
 	}, $class;
 }
 
 sub _flush_buf {
 	my ($self) = @_;
-	if (defined ${$self->{buf}}) {
-		$self->{fh}->write(${$self->{buf}});
-		${$self->{buf}} = undef;
-	}
+	my $ret = $self->{buf};
+	$ret = $$ret;
+	${$self->{buf}} = undef;
+	$ret;
 }
 
-sub write {
-	$_[0]->{gz}->write($_[1]);
-	_flush_buf($_[0]);
-}
-
-sub close {
+# called by Plack::Util::foreach or similar
+sub getline {
 	my ($self) = @_;
+	my $res;
+	my $ctx = $self->{ctx};
+	my $git = $ctx->{git};
+	do {
+		while (defined(my $smsg = shift @{$self->{msgs}})) {
+			my $msg = eval {
+				my $p = 'HEAD:'.mid2path($smsg->mid);
+				Email::Simple->new($git->cat_file($p));
+			};
+			$msg or next;
+
+			PublicInbox::Mbox::emit_msg($ctx, $self->{gz}, $msg);
+			my $ret = _flush_buf($self);
+			return $ret if $ret;
+		}
+		$res = $self->{cb}->($self->{opts});
+		$self->{msgs} = $res->{msgs};
+		$res = scalar @{$self->{msgs}};
+		$self->{opts}->{offset} += $res;
+	} while ($res);
 	$self->{gz}->close;
 	_flush_buf($self);
-	$self->{fh}->close;
 }
 
+sub close {} # noop
+
 1;
-- 
EW

[PATCH 4/1] unsubscribe: prevent decrypt from showing random crap

[Eric Wong]

Wow, I don't know crypto at all.
---
 Hopefully this isn't exploitable somehow... Gah :x

 lib/PublicInbox/Unsubscribe.pm | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/PublicInbox/Unsubscribe.pm b/lib/PublicInbox/Unsubscribe.pm
index 4ccdb7e..97ff97f 100644
--- a/lib/PublicInbox/Unsubscribe.pm
+++ b/lib/PublicInbox/Unsubscribe.pm
@@ -77,7 +77,7 @@ sub _user_list_addr {
 			'Missing mailing list name in path component');
 	}
 	my $user = eval { $self->{cipher}->decrypt(decode_base64url($u)) };
-	if (!defined $user || $user eq '') {
+	if (!defined $user || index($user, '@') <= 1) {
 		my $err = quotemeta($@);
 		my $errors = $env->{'psgi.errors'};
 		$errors->print("error decrypting: $u\n");

[PATCH] localize $/ in more places to avoid potential problems

[Eric Wong]

This hopefully makes the intent of the code clearer, too.
The the HTTP use of the numeric reference for getline
caused problems in Git.pm, already.
---
 lib/PublicInbox/Config.pm    | 1 +
 lib/PublicInbox/Daemon.pm    | 1 +
 lib/PublicInbox/Feed.pm      | 2 ++
 lib/PublicInbox/Git.pm       | 1 +
 lib/PublicInbox/SearchIdx.pm | 2 ++
 t/httpd-corner.psgi          | 2 ++
 t/httpd-unix.t               | 1 +
 7 files changed, 10 insertions(+)

diff --git a/lib/PublicInbox/Config.pm b/lib/PublicInbox/Config.pm
index b5f0fcb..b38f444 100644
--- a/lib/PublicInbox/Config.pm
+++ b/lib/PublicInbox/Config.pm
@@ -80,6 +80,7 @@ sub git_config_dump {
 	my $pid = open(my $fh, '-|', @cmd);
 	defined $pid or die "$cmd failed: $!";
 	my %rv;
+	local $/ = "\n";
 	foreach my $line (<$fh>) {
 		chomp $line;
 		my ($k, $v) = split(/=/, $line, 2);
diff --git a/lib/PublicInbox/Daemon.pm b/lib/PublicInbox/Daemon.pm
index 8de7ff2..dc81010 100644
--- a/lib/PublicInbox/Daemon.pm
+++ b/lib/PublicInbox/Daemon.pm
@@ -350,6 +350,7 @@ sub unlink_pid_file_safe_ish ($$) {
 	return unless defined $unlink_pid && $unlink_pid == $$;
 
 	open my $fh, '<', $file or return;
+	local $/ = "\n";
 	defined(my $read_pid = <$fh>) or return;
 	chomp $read_pid;
 	if ($read_pid == $unlink_pid) {
diff --git a/lib/PublicInbox/Feed.pm b/lib/PublicInbox/Feed.pm
index e2df97b..6ed0085 100644
--- a/lib/PublicInbox/Feed.pm
+++ b/lib/PublicInbox/Feed.pm
@@ -224,6 +224,7 @@ sub each_recent_blob {
 	my $nr = 0;
 	my ($cur_commit, $first_commit, $last_commit);
 	my ($ts, $subj, $u);
+	local $/ = "\n";
 	while (defined(my $line = <$log>)) {
 		if ($line =~ /$addmsg/o) {
 			my $add = $1;
@@ -244,6 +245,7 @@ sub each_recent_blob {
 	}
 
 	if ($last) {
+		local $/ = "\n";
 		while (my $line = <$log>) {
 			if ($line =~ /^(${hex}{7,40})/o) {
 				$last_commit = $1;
diff --git a/lib/PublicInbox/Git.pm b/lib/PublicInbox/Git.pm
index 473cdff..bc0e506 100644
--- a/lib/PublicInbox/Git.pm
+++ b/lib/PublicInbox/Git.pm
@@ -122,6 +122,7 @@ sub popen {
 sub qx {
 	my ($self, @cmd) = @_;
 	my $fh = $self->popen(@cmd);
+	local $/ = "\n";
 	return <$fh> if wantarray;
 	local $/;
 	<$fh>
diff --git a/lib/PublicInbox/SearchIdx.pm b/lib/PublicInbox/SearchIdx.pm
index 9192bb0..4a4b2bd 100644
--- a/lib/PublicInbox/SearchIdx.pm
+++ b/lib/PublicInbox/SearchIdx.pm
@@ -338,6 +338,7 @@ sub rlog {
 				--raw -r --no-abbrev/, $range);
 	my $latest;
 	my $bytes;
+	local $/ = "\n";
 	while (defined(my $line = <$log>)) {
 		if ($line =~ /$addmsg/o) {
 			my $mime = do_cat_mail($git, $1, \$bytes) or next;
@@ -445,6 +446,7 @@ sub _read_git_config_perm {
 	my ($self) = @_;
 	my @cmd = qw(config core.sharedRepository);
 	my $fh = PublicInbox::Git->new($self->{git_dir})->popen(@cmd);
+	local $/ = "\n";
 	my $perm = <$fh>;
 	chomp $perm if defined $perm;
 	$perm;
diff --git a/t/httpd-corner.psgi b/t/httpd-corner.psgi
index 2f7be83..222b9e0 100644
--- a/t/httpd-corner.psgi
+++ b/t/httpd-corner.psgi
@@ -30,6 +30,7 @@ my $app = sub {
 			return sub {
 				open my $f, '<', $fifo or
 						die "open $fifo: $!\n";
+				local $/ = "\n";
 				my @r = <$f>;
 				$_[0]->([200, $h, \@r ]);
 			};
@@ -38,6 +39,7 @@ my $app = sub {
 				my $fh = $_[0]->([200, $h]);
 				open my $f, '<', $fifo or
 						die "open $fifo: $!\n";
+				local $/ = "\n";
 				while (defined(my $l = <$f>)) {
 					$fh->write($l);
 				}
diff --git a/t/httpd-unix.t b/t/httpd-unix.t
index 00adf13..16f7bdd 100644
--- a/t/httpd-unix.t
+++ b/t/httpd-unix.t
@@ -103,6 +103,7 @@ SKIP: {
 
 	ok(-f "$tmpdir/pid", 'pid file written');
 	open my $fh, '<', "$tmpdir/pid" or die "open failed: $!";
+	local $/ = "\n";
 	my $rpid = <$fh>;
 	chomp $rpid;
 	like($rpid, qr/\A\d+\z/s, 'pid file looks like a pid');