nntp: prevent event_read from firing twice in a row

When a client starts pipelining requests to us which trigger
long responses, we need to keep socket readiness checks disabled
and only enable them when our socket rbuf is drained.

Failure to do this caused aborted clients with
"BUG: nested long response" when Danga::Socket calls event_read
for read-readiness after our "next_tick" sub fires in the
same event loop iteration.

Reported-by: Jonathan Corbet <corbet@lwn.net>
cf. https://public-inbox.org/meta/20181013124658.23b9f9d2@lwn.net/

diff --git a/lib/PublicInbox/NNTP.pm b/lib/PublicInbox/NNTP.pm
index 022bb809585ca38edb1164d83807bd6063fc4db2..90a5a3a58e686b35eb1367e9accc95d1c9fc1b6f 100644 (file)
--- a/lib/PublicInbox/NNTP.pm
+++ b/lib/PublicInbox/NNTP.pm
@@ -51,8 +51,16 @@ sub next_tick () {
                # before finishing reading:
                if (my $long_cb = $nntp->{long_res}) {
                        $nntp->write($long_cb);
-               } elsif (&Danga::Socket::POLLIN & $nntp->{event_watch}) {
+               } else {
+                       # pipelined request, we bypassed socket-readiness
+                       # checks to get here:
                        event_read($nntp);
+
+                       # maybe there's more pipelined data, or we'll have
+                       # to register it for socket-readiness notifications
+                       if (!$nntp->{long_res} && !$nntp->{closed}) {
+                               check_read($nntp);
+                       }
                }
        }
 }
@@ -609,7 +617,7 @@ sub long_response ($$) {
                                           now() - $t0);
                        } else {
                                update_idle_time($self);
-                               $self->watch_read(1);
+                               check_read($self);
                        }
                } elsif ($more) { # $self->{write_buf_size}:
                        # no recursion, schedule another call ASAP
@@ -620,7 +628,7 @@ sub long_response ($$) {
                        $nextt ||= PublicInbox::EvCleanup::asap(*next_tick);
                } else { # all done!
                        $self->{long_res} = undef;
-                       $self->watch_read(1);
+                       check_read($self);
                        res($self, '.');
                        out($self, " deferred[$fd] done - %0.6f", now() - $t0);
                }
@@ -968,10 +976,9 @@ sub event_read {
        update_idle_time($self);
 }
 
-sub watch_read {
-       my ($self, $bool) = @_;
-       my $rv = $self->SUPER::watch_read($bool);
-       if ($bool && index($self->{rbuf}, "\n") >= 0) {
+sub check_read {
+       my ($self) = @_;
+       if (index($self->{rbuf}, "\n") >= 0) {
                # Force another read if there is a pipelined request.
                # We don't know if the socket has anything for us to read,
                # and we must double-check again by the time the timer fires
@@ -979,8 +986,11 @@ sub watch_read {
                # another long response.
                push @$nextq, $self;
                $nextt ||= PublicInbox::EvCleanup::asap(*next_tick);
+       } else {
+               # no pipelined requests available, let the kernel know
+               # to wake us up if there's more
+               $self->watch_read(1); # Danga::Socket::watch_read
        }
-       $rv;
 }
 
 sub not_idle_long ($$) {

diff --git a/t/nntpd.t b/t/nntpd.t
index 9c1d076259fb91866e55c2576cb41a22feecc0c7..ffed437631cde2926e0d7e7efa9ebe6aa1c9cc20 100644 (file)
--- a/t/nntpd.t
+++ b/t/nntpd.t
@@ -252,6 +252,27 @@ EOF
                ok($date <= $t1, 'valid date before stop');
        }
 
+       # pipelined requests:
+       {
+               my $nreq = 90;
+               syswrite($s, "GROUP $group\r\n");
+               my $res = <$s>;
+               my $rdr = fork;
+               if ($rdr == 0) {
+                       use POSIX qw(_exit);
+                       for (1..$nreq) {
+                               <$s> =~ /\A224 / or _exit(1);
+                               <$s> =~ /\A1/ or _exit(2);
+                               <$s> eq ".\r\n" or _exit(3);
+                       }
+                       _exit(0);
+               }
+               for (1..$nreq) {
+                       syswrite($s, "XOVER 1\r\n");
+               }
+               is($rdr, waitpid($rdr, 0), 'reader done');
+               is($? >> 8, 0, 'no errors');
+       }
        {
                setsockopt($s, IPPROTO_TCP, TCP_NODELAY, 1);
                syswrite($s, 'HDR List-id 1-');