pkt_op: favor level-triggered epoll for fairness

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.

1 files changed, 24 insertions, 29 deletions

diff --git a/lib/PublicInbox/PktOp.pm b/lib/PublicInbox/PktOp.pm
index fd2569ba..4c434566 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;