about summary refs log tree commit homepage
diff options
context:
space:
mode:
-rw-r--r--lib/PublicInbox/PktOp.pm53
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;