input_pipe: do not loop in ->event_step for fairness

[public-inbox.git] / lib / PublicInbox / InputPipe.pm

diff --git a/lib/PublicInbox/InputPipe.pm b/lib/PublicInbox/InputPipe.pm
index a8bdf0318223c2a8993fc24bc414fc59f8b39146..00813a0701b172d4688997d3bf27116f98e4c15c 100644 (file)
--- 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