[PATCH 0/3] lei: stdin handling improvements

[PATCH 0/3] lei: stdin handling improvements

[Eric Wong]

I'm not sure about 3/3 since termios(3) stuff is mostly alien to
me.  I'm very slowly expanding my horizons...

Eric Wong (3):
  lei: consolidate stdin slurp, fix warnings
  input_pipe: improve error handling
  input_pipe: handle noncanonical TTY

 lib/PublicInbox/InputPipe.pm  | 79 ++++++++++++++++++++++++++++-------
 lib/PublicInbox/LEI.pm        | 13 ++++++
 lib/PublicInbox/LeiInspect.pm | 12 +-----
 lib/PublicInbox/LeiLcat.pm    | 13 +-----
 lib/PublicInbox/LeiQuery.pm   | 14 ++-----
 t/lei.t                       |  5 +++
 6 files changed, 88 insertions(+), 48 deletions(-)

[PATCH 1/3] lei: consolidate stdin slurp, fix warnings

[Eric Wong]

We can share more code amongst stdin slurper (not streaming)
commands.  This also fixes uninitialized variable warnings when
feeding an empty stdin to these commands.
---
 lib/PublicInbox/LEI.pm        | 13 +++++++++++++
 lib/PublicInbox/LeiInspect.pm | 12 ++----------
 lib/PublicInbox/LeiLcat.pm    | 13 ++-----------
 lib/PublicInbox/LeiQuery.pm   | 14 +++-----------
 t/lei.t                       |  5 +++++
 5 files changed, 25 insertions(+), 32 deletions(-)

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index b00be1a1..1ff6d67f 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -1573,4 +1573,17 @@ sub request_umask {
 	$u eq 'u' or warn "E: recv $v has no umask";
 }
 
+sub _stdin_cb { # PublicInbox::InputPipe::consume callback for --stdin
+	my ($lei, $cb) = @_; # $_[-1] = $rbuf
+	$_[1] // return $lei->fail("error reading stdin: $!");
+	$lei->{stdin_buf} .= $_[-1];
+	do_env($lei, $cb) if $_[-1] eq '';
+}
+
+sub slurp_stdin {
+	my ($lei, $cb) = @_;
+	require PublicInbox::InputPipe;
+	PublicInbox::InputPipe::consume($lei->{0}, \&_stdin_cb, $lei, $cb);
+}
+
 1;
diff --git a/lib/PublicInbox/LeiInspect.pm b/lib/PublicInbox/LeiInspect.pm
index 65c64cf2..d4ad03eb 100644
--- a/lib/PublicInbox/LeiInspect.pm
+++ b/lib/PublicInbox/LeiInspect.pm
@@ -253,20 +253,13 @@ sub inspect_start ($$) {
 
 sub do_inspect { # lei->do_env cb
 	my ($lei) = @_;
-	my $str = delete $lei->{istr};
+	my $str = delete $lei->{stdin_buf};
 	PublicInbox::Eml::strip_from($str);
 	my $eml = PublicInbox::Eml->new(\$str);
 	inspect_start($lei, [ 'blob:'.$lei->git_oid($eml)->hexdigest,
 			map { "mid:$_" } @{mids($eml)} ]);
 }
 
-sub ins_add { # InputPipe->consume callback
-	my ($lei) = @_; # $_[1] = $rbuf
-	$_[1] // return $lei->fail("error reading stdin: $!");
-	return $lei->{istr} .= $_[1] if $_[1] ne '';
-	$lei->do_env(\&do_inspect);
-}
-
 sub lei_inspect {
 	my ($lei, @argv) = @_;
 	$lei->{json} = ref(PublicInbox::Config::json())->new->utf8->canonical;
@@ -281,8 +274,7 @@ sub lei_inspect {
 		return $lei->fail(<<'') if @argv;
 no args allowed on command-line with --stdin
 
-		require PublicInbox::InputPipe;
-		PublicInbox::InputPipe::consume($lei->{0}, \&ins_add, $lei);
+		$lei->slurp_stdin(\&do_inspect);
 	} else {
 		inspect_start($lei, \@argv);
 	}
diff --git a/lib/PublicInbox/LeiLcat.pm b/lib/PublicInbox/LeiLcat.pm
index 72875dc6..274a9605 100644
--- a/lib/PublicInbox/LeiLcat.pm
+++ b/lib/PublicInbox/LeiLcat.pm
@@ -124,18 +124,11 @@ could not extract Message-ID from $x
 
 sub do_lcat { # lei->do_env cb
 	my ($lei) = @_;
-	my @argv = split(/\s+/, $lei->{mset_opt}->{qstr});
+	my @argv = split(/\s+/, delete($lei->{stdin_buf}));
 	$lei->{mset_opt}->{qstr} = extract_all($lei, @argv) or return;
 	$lei->_start_query;
 }
 
-sub _stdin { # PublicInbox::InputPipe::consume callback for --stdin
-	my ($lei) = @_; # $_[1] = $rbuf
-	$_[1] // return $lei->fail("error reading stdin: $!");
-	return $lei->{mset_opt}->{qstr} .= $_[1] if $_[1] ne '';
-	$lei->do_env(\&do_lcat);
-}
-
 sub lei_lcat {
 	my ($lei, @argv) = @_;
 	my $lxs = $lei->lxs_prepare or return;
@@ -152,9 +145,7 @@ sub lei_lcat {
 		return $lei->fail(<<'') if @argv;
 no args allowed on command-line with --stdin
 
-		require PublicInbox::InputPipe;
-		PublicInbox::InputPipe::consume($lei->{0}, \&_stdin, $lei);
-		return;
+		return $lei->slurp_stdin(\&do_lcat);
 	}
 	$lei->{mset_opt}->{qstr} = extract_all($lei, @argv) or return;
 	$lei->_start_query;
diff --git a/lib/PublicInbox/LeiQuery.pm b/lib/PublicInbox/LeiQuery.pm
index e2d8a096..eadf811f 100644
--- a/lib/PublicInbox/LeiQuery.pm
+++ b/lib/PublicInbox/LeiQuery.pm
@@ -61,19 +61,13 @@ sub _start_query { # used by "lei q" and "lei up"
 
 sub do_qry { # do_env cb
 	my ($lei) = @_;
-	$lei->{mset_opt}->{q_raw} = $lei->{mset_opt}->{qstr};
+	$lei->{mset_opt}->{q_raw} = $lei->{mset_opt}->{qstr}
+						= delete $lei->{stdin_buf};
 	$lei->{lse}->query_approxidate($lei->{lse}->git,
 					$lei->{mset_opt}->{qstr});
 	_start_query($lei);
 }
 
-sub qstr_add { # PublicInbox::InputPipe::consume callback for --stdin
-	my ($lei) = @_; # $_[1] = $rbuf
-	$_[1] // $lei->fail("error reading stdin: $!");
-	return $lei->{mset_opt}->{qstr} .= $_[1] if $_[1] ne '';
-	$lei->do_env(\&do_qry);
-}
-
 # make the URI||PublicInbox::{Inbox,ExtSearch} a config-file friendly string
 sub cfg_ext ($) {
 	my ($x) = @_;
@@ -159,9 +153,7 @@ sub lei_q {
 		return $self->fail(<<'') if @argv;
 no query allowed on command-line with --stdin
 
-		require PublicInbox::InputPipe;
-		PublicInbox::InputPipe::consume($self->{0}, \&qstr_add, $self);
-		return;
+		return $self->slurp_stdin(\&do_qry);
 	}
 	chomp(@argv) and $self->qerr("# trailing `\\n' removed");
 	$mset_opt{q_raw} = [ @argv ]; # copy
diff --git a/t/lei.t b/t/lei.t
index 3ac804a8..1dbc9d4c 100644
--- a/t/lei.t
+++ b/t/lei.t
@@ -182,6 +182,11 @@ my $test_fail = sub {
 	}
 	lei_ok('sucks', \'yes, but hopefully less every day');
 	like($lei_out, qr/loaded features/, 'loaded features shown');
+
+	lei_ok([qw(q --stdin -f text)], undef, { 0 => \'', %$lei_opt });
+	is($lei_err, '', 'no errors on empty stdin');
+	is($lei_out, '', 'no output on empty query');
+
 SKIP: {
 	skip 'no curl', 3 unless require_cmd('curl', 1);
 	lei(qw(q --only http://127.0.0.1:99999/bogus/ t:m));

[PATCH 2/3] input_pipe: improve error handling

[Eric Wong]

Ensure the callback is always guarded by `eval' to catch
exceptions and to force a ->close (EPOLL_CTL_DEL).

We also don't want to blindly set O_NONBLOCK on TTYs since their
O_NONBLOCK semantics aren't well-defined by POSIX.  We can also
drop EPOLLET (edge-triggered) use to reduce the need to make
->requeue calls on our end.
---
 lib/PublicInbox/InputPipe.pm | 48 ++++++++++++++++++++++++------------
 1 file changed, 32 insertions(+), 16 deletions(-)

diff --git a/lib/PublicInbox/InputPipe.pm b/lib/PublicInbox/InputPipe.pm
index 60a9f01f..39aefab2 100644
--- a/lib/PublicInbox/InputPipe.pm
+++ b/lib/PublicInbox/InputPipe.pm
@@ -1,35 +1,51 @@
 # Copyright (C) all contributors <meta@public-inbox.org>
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 
-# for reading pipes and sockets off the DS event loop
+# for reading pipes, sockets, and TTYs off the DS event loop
 package PublicInbox::InputPipe;
 use v5.12;
 use parent qw(PublicInbox::DS);
-use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
+use PublicInbox::Syscall qw(EPOLLIN);
 
 sub consume {
 	my ($in, $cb, @args) = @_;
 	my $self = bless { cb => $cb, args => \@args }, __PACKAGE__;
-	eval { $self->SUPER::new($in, EPOLLIN|EPOLLET) };
-	return $self->requeue if $@; # regular file
-	$in->blocking(0); # pipe or socket
+	eval { $self->SUPER::new($in, EPOLLIN) };
+	if ($@) { # regular file (but not w/ select|IO::Poll backends)
+		$self->{-need_rq} = 1;
+		$self->requeue;
+	} elsif (-p $in || -s _) { # O_NONBLOCK for sockets and pipes
+		$in->blocking(0);
+	} # TODO: tty
+}
+
+sub close {
+	my ($self) = @_;
+	$self->{-need_rq} ? delete($self->{sock}) : $self->SUPER::close
 }
 
 sub event_step {
 	my ($self) = @_;
 	my $r = sysread($self->{sock} // return, my $rbuf, 65536);
-	if ($r) {
-		$self->{cb}->(@{$self->{args}}, $rbuf);
-		return $self->requeue; # may be regular file or pipe
-	}
-	if (defined($r)) { # EOF
-		$self->{cb}->(@{$self->{args}}, '');
-	} elsif ($!{EAGAIN}) {
-		return;
-	} else { # another error
-		$self->{cb}->(@{$self->{args}}, undef)
+	eval {
+		if ($r) {
+			$self->{cb}->(@{$self->{args}}, $rbuf);
+			$self->requeue if $self->{-need_rq};
+		} elsif (defined($r)) { # EOF
+			$self->{cb}->(@{$self->{args}}, '');
+			$self->close
+		} elsif ($!{EAGAIN}) { # rely on EPOLLIN
+		} elsif ($!{EINTR}) { # rely on EPOLLIN for sockets/pipes/tty
+			$self->requeue if $self->{-need_rq};
+		} else { # another error
+			$self->{cb}->(@{$self->{args}}, undef);
+			$self->close;
+		}
+	};
+	if ($@) {
+		warn "E: $@";
+		$self->close;
 	}
-	$self->{sock}->blocking ? delete($self->{sock}) : $self->close
 }
 
 1;

[RFC 3/3] input_pipe: handle noncanonical TTY

[Eric Wong]

lei could get a TTY in noncanonical mode for stdin, so rely on
VMIN+VTIME to get the desired non-blocking semantics we'd expect
from a pipe or socket.  This ought to prevent read(2) (Perl sysread)
from returning zero when we really want to hit EAGAIN.
---
 lib/PublicInbox/InputPipe.pm | 33 ++++++++++++++++++++++++++++++++-
 1 file changed, 32 insertions(+), 1 deletion(-)

diff --git a/lib/PublicInbox/InputPipe.pm b/lib/PublicInbox/InputPipe.pm
index 39aefab2..8358ddd6 100644
--- a/lib/PublicInbox/InputPipe.pm
+++ b/lib/PublicInbox/InputPipe.pm
@@ -6,6 +6,31 @@ package PublicInbox::InputPipe;
 use v5.12;
 use parent qw(PublicInbox::DS);
 use PublicInbox::Syscall qw(EPOLLIN);
+use POSIX ();
+use Carp qw(croak carp);
+
+# I'm not sure what I'm doing w.r.t terminals.
+# FIXME needs non-interactive tests
+sub unblock_tty ($) {
+	my ($self) = @_;
+	my $fd = fileno($self->{sock});
+	my $t = POSIX::Termios->new;
+	$t->getattr($fd) or croak("tcgetattr($fd): $!");
+	return if $t->getlflag & POSIX::ICANON; # line-oriented, good
+
+	# make noncanonical mode TTYs behave like a O_NONBLOCK pipe.
+	# O_NONBLOCK itself isn't well-defined, here, so rely on VMIN + VTIME
+	my ($vmin, $vtime) = ($t->getcc(POSIX::VMIN), $t->getcc(POSIX::VTIME));
+	return if $vmin == 1 && $vtime == 0;
+
+	$t->setcc(POSIX::VMIN, 1); # 1 byte minimum
+	$t->setcc(POSIX::VTIME, 0); # no timeout
+	$t->setattr($fd, POSIX::TCSANOW) or croak("tcsetattr($fd): $!");
+
+	$t->setcc(POSIX::VMIN, $vmin);
+	$t->setcc(POSIX::VTIME, $vtime);
+	$self->{restore_termios} = $t;
+}
 
 sub consume {
 	my ($in, $cb, @args) = @_;
@@ -16,11 +41,17 @@ sub consume {
 		$self->requeue;
 	} elsif (-p $in || -s _) { # O_NONBLOCK for sockets and pipes
 		$in->blocking(0);
-	} # TODO: tty
+	} elsif (-t $in) { # isatty(3) can't use `_' stat cache
+		unblock_tty($self);
+	}
 }
 
 sub close {
 	my ($self) = @_;
+	if (my $t = delete($self->{restore_termios})) {
+		my $fd = fileno($self->{sock} // return);
+		$t->setattr($fd, POSIX::TCSANOW) or carp("tcsetattr($fd): $!")
+	}
 	$self->{-need_rq} ? delete($self->{sock}) : $self->SUPER::close
 }