[PATCH 0/4] lei: prioritize signals

[PATCH 0/4] lei: prioritize signals

[Eric Wong]

We drop Edge-Triggered kevent/epoll in nearly all places
to give signalfd / EVFILT_SIGNAL priority.

Eric Wong (4):
  wqworker: favor level-triggered epoll for fairness
  pkt_op: favor level-triggered epoll for fairness
  input_pipe: do not loop in ->event_step for fairness
  lei sockets: favor level-triggered epoll for fairness

 lib/PublicInbox/InputPipe.pm     | 17 +++++-----
 lib/PublicInbox/LEI.pm           | 20 +++++-------
 lib/PublicInbox/LeiSelfSocket.pm | 27 +++++++---------
 lib/PublicInbox/PktOp.pm         | 53 +++++++++++++++-----------------
 lib/PublicInbox/WQWorker.pm      | 18 +++++------
 5 files changed, 60 insertions(+), 75 deletions(-)

[PATCH 1/4] wqworker: favor level-triggered epoll for fairness

[Eric Wong]

Sigfd->event_step needs priority over WQWorkers (and everything
else).  Do that by running once per event_loop iteration rather
than looping inside event_step.  This lowers throughput since it
requires more syscalls, but that's the price of fairness.
---
 lib/PublicInbox/WQWorker.pm | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/lib/PublicInbox/WQWorker.pm b/lib/PublicInbox/WQWorker.pm
index 48b901bb139f..950bd17052a5 100644
--- a/lib/PublicInbox/WQWorker.pm
+++ b/lib/PublicInbox/WQWorker.pm
@@ -6,7 +6,7 @@ package PublicInbox::WQWorker;
 use strict;
 use v5.10.1;
 use parent qw(PublicInbox::DS);
-use PublicInbox::Syscall qw(EPOLLIN EPOLLEXCLUSIVE EPOLLET);
+use PublicInbox::Syscall qw(EPOLLIN EPOLLEXCLUSIVE);
 use Errno qw(EAGAIN ECONNRESET);
 use IO::Handle (); # blocking
 
@@ -14,19 +14,19 @@ sub new {
 	my ($cls, $wq, $sock) = @_;
 	$sock->blocking(0);
 	my $self = bless { sock => $sock, wq => $wq }, $cls;
-	$self->SUPER::new($sock, EPOLLEXCLUSIVE|EPOLLIN|EPOLLET);
+	$self->SUPER::new($sock, EPOLLEXCLUSIVE|EPOLLIN);
 	$self;
 }
 
 sub event_step {
 	my ($self) = @_;
-	my $n;
-	do {
-		$n = $self->{wq}->recv_and_run($self->{sock});
-	} while ($n);
-	return if !defined($n) && $! == EAGAIN; # likely
-	warn "wq worker error: $!\n" if !defined($n) && $! != ECONNRESET;
-	$self->{wq}->wq_atexit_child if $self->{sock} == $self->{wq}->{-wq_s2};
+	my $n = $self->{wq}->recv_and_run($self->{sock}) and return;
+	unless (defined $n) {
+		return if $! == EAGAIN;
+		warn "recvmsg: $!" if $! != ECONNRESET;
+	}
+	$self->{sock} == $self->{wq}->{-wq_s2} and
+		$self->{wq}->wq_atexit_child;
 	$self->close; # PublicInbox::DS::close
 }
 

[PATCH 2/4] pkt_op: favor level-triggered epoll for fairness

[Eric Wong]

Sigfd->event_step needs priority over PktOp (and everything else).
We'll also add ECONNRESET checking, here, since it could see
bidirectional use in the future.

This is unlikely to have any sort of performance difference
since this is only for small, occasional packets, but the code
reduction is nice.
---
 lib/PublicInbox/PktOp.pm | 53 ++++++++++++++++++----------------------
 1 file changed, 24 insertions(+), 29 deletions(-)

diff --git a/lib/PublicInbox/PktOp.pm b/lib/PublicInbox/PktOp.pm
index fd2569badd74..4c434566d31f 100644
--- a/lib/PublicInbox/PktOp.pm
+++ b/lib/PublicInbox/PktOp.pm
@@ -9,8 +9,8 @@ package PublicInbox::PktOp;
 use strict;
 use v5.10.1;
 use parent qw(PublicInbox::DS);
-use Errno qw(EAGAIN EINTR);
-use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
+use Errno qw(EAGAIN ECONNRESET);
+use PublicInbox::Syscall qw(EPOLLIN);
 use Socket qw(AF_UNIX MSG_EOR SOCK_SEQPACKET);
 use PublicInbox::IPC qw(ipc_freeze ipc_thaw);
 use Scalar::Util qw(blessed);
@@ -19,7 +19,7 @@ sub new {
 	my ($cls, $r) = @_;
 	my $self = bless { sock => $r }, $cls;
 	$r->blocking(0);
-	$self->SUPER::new($r, EPOLLIN|EPOLLET);
+	$self->SUPER::new($r, EPOLLIN);
 }
 
 # returns a blessed objects as the consumer and producer
@@ -38,33 +38,28 @@ sub pkt_do { # for the producer to trigger event_step in consumer
 sub event_step {
 	my ($self) = @_;
 	my $c = $self->{sock};
-	my $msg;
-	while (1) {
-		my $n = recv($c, $msg, 4096, 0);
-		unless (defined $n) {
-			return if $! == EAGAIN;
-			next if $! == EINTR;
-			$self->close;
-			die "recv: $!";
-		}
-		my ($cmd, @pargs);
-		if (index($msg, "\0") > 0) {
-			($cmd, my $pargs) = split(/\0/, $msg, 2);
-			@pargs = @{ipc_thaw($pargs)};
-		} else {
-			# for compatibility with the script/lei in client mode,
-			# it doesn't load Sereal||Storable for startup speed
-			($cmd, @pargs) = split(/ /, $msg);
-		}
-		my $op = $self->{ops}->{$cmd //= $msg};
-		if ($op) {
-			my ($obj, @args) = (@$op, @pargs);
-			blessed($obj) ? $obj->$cmd(@args) : $obj->(@args);
-		} elsif ($msg ne '') {
-			die "BUG: unknown message: `$cmd'";
-		}
-		return $self->close if $msg eq ''; # close on EOF
+	my $n = recv($c, my $msg, 4096, 0);
+	unless (defined $n) {
+		return if $! == EAGAIN;
+		die "recv: $!" if $! != ECONNRESET; # we may be bidirectional
 	}
+	my ($cmd, @pargs);
+	if (index($msg, "\0") > 0) {
+		($cmd, my $pargs) = split(/\0/, $msg, 2);
+		@pargs = @{ipc_thaw($pargs)};
+	} else {
+		# for compatibility with the script/lei in client mode,
+		# it doesn't load Sereal||Storable for startup speed
+		($cmd, @pargs) = split(/ /, $msg);
+	}
+	my $op = $self->{ops}->{$cmd //= $msg};
+	if ($op) {
+		my ($obj, @args) = (@$op, @pargs);
+		blessed($obj) ? $obj->$cmd(@args) : $obj->(@args);
+	} elsif ($msg ne '') {
+		die "BUG: unknown message: `$cmd'";
+	}
+	$self->close if $msg eq ''; # close on EOF
 }
 
 1;

[PATCH 3/4] input_pipe: do not loop in ->event_step for fairness

[Eric Wong]

Sigfd->event_step needs priority over InputPipe (and everything
else).  We keep Edge Triggering here but use ->requeue instead
of looping inside event_step.  This was necessary because
InputPipe can be used with regular files which can't be
monitored with epoll.

We'll also rid of the vestigial lei-oneshot support while we're
at it.
---
 lib/PublicInbox/InputPipe.pm | 17 ++++++++---------
 1 file changed, 8 insertions(+), 9 deletions(-)

diff --git a/lib/PublicInbox/InputPipe.pm b/lib/PublicInbox/InputPipe.pm
index a8bdf0318223..00813a0701b1 100644
--- a/lib/PublicInbox/InputPipe.pm
+++ b/lib/PublicInbox/InputPipe.pm
@@ -10,25 +10,24 @@ use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
 
 sub consume {
 	my ($in, $cb, @args) = @_;
-	my $self = bless { cb => $cb, sock => $in, args => \@args },__PACKAGE__;
-	if ($PublicInbox::DS::in_loop) {
-		eval { $self->SUPER::new($in, EPOLLIN|EPOLLET) };
-		return $in->blocking(0) unless $@; # regular file sets $@
-	}
-	event_step($self) while $self->{sock};
+	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
 }
 
 sub event_step {
 	my ($self) = @_;
-	my ($r, $rbuf);
-	while (($r = sysread($self->{sock}, $rbuf, 65536))) {
+	my $r = sysread($self->{sock}, 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 {
+	} else { # another error
 		$self->{cb}->(@{$self->{args} // []}, undef)
 	}
 	$self->{sock}->blocking ? delete($self->{sock}) : $self->close

[PATCH 4/4] lei sockets: favor level-triggered epoll for fairness

[Eric Wong]

Sigfd->event_step needs priority over script/lei clients,
LeiSelfSocket, and everything else.
---
 lib/PublicInbox/LEI.pm           | 20 ++++++++------------
 lib/PublicInbox/LeiSelfSocket.pm | 27 +++++++++++----------------
 2 files changed, 19 insertions(+), 28 deletions(-)

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 876598f9530e..6b989b33647e 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -12,13 +12,13 @@ use parent qw(PublicInbox::DS PublicInbox::LeiExternal
 	PublicInbox::LeiQuery);
 use Getopt::Long ();
 use Socket qw(AF_UNIX SOCK_SEQPACKET MSG_EOR pack_sockaddr_un);
-use Errno qw(EPIPE EAGAIN EINTR ECONNREFUSED ENOENT ECONNRESET);
+use Errno qw(EPIPE EAGAIN ECONNREFUSED ENOENT ECONNRESET);
 use Cwd qw(getcwd);
 use POSIX qw(strftime);
 use IO::Handle ();
 use Fcntl qw(SEEK_SET);
 use PublicInbox::Config;
-use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
+use PublicInbox::Syscall qw(EPOLLIN);
 use PublicInbox::DS qw(now dwaitpid);
 use PublicInbox::Spawn qw(spawn popen_rd);
 use PublicInbox::Lock;
@@ -1125,16 +1125,12 @@ sub event_step {
 	local %ENV = %{$self->{env}};
 	local $current_lei = $self;
 	eval {
-		my $buf;
-		while (my @fds = $recv_cmd->($self->{sock}, $buf, 4096)) {
-			if (scalar(@fds) == 1 && !defined($fds[0])) {
-				return if $! == EAGAIN;
-				next if $! == EINTR;
-				last if $! == ECONNRESET;
-				die "recvmsg: $!";
-			}
-			for (@fds) { open my $rfh, '+<&=', $_ }
+		my @fds = $recv_cmd->($self->{sock}, my $buf, 4096);
+		if (scalar(@fds) == 1 && !defined($fds[0])) {
+			return if $! == EAGAIN;
+			die "recvmsg: $!" if $! != ECONNRESET;
 		}
+		for (@fds) { open my $rfh, '+<&=', $_ }
 		if ($buf eq '') {
 			_drop_wq($self); # EOF, client disconnected
 			dclose($self);
@@ -1162,7 +1158,7 @@ sub event_step_init {
 	my $sock = $self->{sock} or return;
 	$self->{-event_init_done} //= do { # persist til $ops done
 		$sock->blocking(0);
-		$self->SUPER::new($sock, EPOLLIN|EPOLLET);
+		$self->SUPER::new($sock, EPOLLIN);
 		$sock;
 	};
 }
diff --git a/lib/PublicInbox/LeiSelfSocket.pm b/lib/PublicInbox/LeiSelfSocket.pm
index 3d847649e3b1..dd64b6cfd491 100644
--- a/lib/PublicInbox/LeiSelfSocket.pm
+++ b/lib/PublicInbox/LeiSelfSocket.pm
@@ -10,7 +10,7 @@ use v5.10.1;
 use parent qw(PublicInbox::DS);
 use Data::Dumper;
 $Data::Dumper::Useqq = 1; # should've been the Perl default :P
-use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
+use PublicInbox::Syscall qw(EPOLLIN);
 use PublicInbox::Spawn;
 my $recv_cmd;
 
@@ -20,26 +20,21 @@ sub new {
 	$r->blocking(0);
 	no warnings 'once';
 	$recv_cmd = $PublicInbox::LEI::recv_cmd;
-	$self->SUPER::new($r, EPOLLIN|EPOLLET);
+	$self->SUPER::new($r, EPOLLIN);
 }
 
 sub event_step {
 	my ($self) = @_;
-	while (1) {
-		my (@fds) = $recv_cmd->($self->{sock}, my $buf, 4096 * 33);
-		if (scalar(@fds) == 1 && !defined($fds[0])) {
-			return if $!{EAGAIN};
-			next if $!{EINTR};
-			die "recvmsg: $!";
-		}
-		# open so perl can auto-close them:
-		for my $fd (@fds) {
-			open(my $newfh, '+<&=', $fd) or die "open +<&=$fd: $!";
-		}
-		return $self->close if $buf eq '';
-		warn Dumper({ 'unexpected self msg' => $buf, fds => \@fds });
-		# TODO: figure out what to do with these messages...
+	my (@fds) = $recv_cmd->($self->{sock}, my $buf, 4096 * 33);
+	if (scalar(@fds) == 1 && !defined($fds[0])) {
+		return if $!{EAGAIN};
+		die "recvmsg: $!" unless $!{ECONNRESET};
+	} else { # just in case open so perl can auto-close them:
+		for (@fds) { open my $fh, '+<&=', $_ };
 	}
+	return $self->close if $buf eq '';
+	warn Dumper({ 'unexpected self msg' => $buf, fds => \@fds });
+	# TODO: figure out what to do with these messages...
 }
 
 1;