about summary refs log tree commit homepage
path: root/lib/PublicInbox
diff options
context:
space:
mode:
authorEric Wong <e@80x24.org>2019-06-16 06:11:28 +0000
committerEric Wong <e@80x24.org>2019-06-16 06:36:46 +0000
commitc6e7de3579258b73aed9ee5f96c4cb25bb335c86 (patch)
tree2cf41bfd355431c74cdcc316840f9f2739ea215d /lib/PublicInbox
parent3c076ce7bb6ac638c49dc2fd533d2193b2a8dad2 (diff)
downloadpublic-inbox-c6e7de3579258b73aed9ee5f96c4cb25bb335c86.tar.gz
Having separate read/write callbacks in every class is too
confusing to my easily-confused mind.  Instead, give every class
an "event_step" callback which is easier to wrap my head around.

This will make future code to support IO::Socket::SSL-wrapped
sockets easier-to-digest, since SSL_write() can require waiting
on POLLIN events, and SSL_read() can require waiting on POLLOUT
events.
Diffstat (limited to 'lib/PublicInbox')
-rw-r--r--lib/PublicInbox/DS.pm47
-rw-r--r--lib/PublicInbox/EvCleanup.pm4
-rw-r--r--lib/PublicInbox/HTTP.pm22
-rw-r--r--lib/PublicInbox/HTTPD/Async.pm2
-rw-r--r--lib/PublicInbox/Listener.pm2
-rw-r--r--lib/PublicInbox/NNTP.pm21
-rw-r--r--lib/PublicInbox/ParentPipe.pm2
7 files changed, 35 insertions, 65 deletions
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index 2f028a36..2b04886a 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -239,21 +239,6 @@ sub RunTimers {
     return $timeout;
 }
 
-sub event_step ($) {
-    my ($self) = @_;
-    return if $self->{closed};
-
-    my $wbuf = $self->{wbuf};
-    if (@$wbuf) {
-        $self->event_write;
-        return if $self->{closed} || scalar(@$wbuf);
-    }
-
-    # only read more requests if we've drained the write buffer,
-    # otherwise we can be buffering infinitely w/o backpressure
-    $self->event_read;
-}
-
 ### The epoll-based event loop. Gets installed as EventLoop if IO::Epoll loads
 ### okay.
 sub EpollEventLoop {
@@ -267,13 +252,11 @@ sub EpollEventLoop {
         # get up to 1000 events
         my $evcount = epoll_wait($Epoll, 1000, $timeout, \@events);
         for ($i=0; $i<$evcount; $i++) {
-            my $ev = $events[$i];
-
             # it's possible epoll_wait returned many events, including some at the end
             # that ones in the front triggered unregister-interest actions.  if we
             # can't find the %sock entry, it's because we're no longer interested
             # in that event.
-            event_step($DescriptorMap{$ev->[0]});
+            $DescriptorMap{$events[$i]->[0]}->event_step;
         }
         return unless PostEventLoop();
     }
@@ -316,9 +299,7 @@ sub PollEventLoop {
         # Fetch handles with read events
         while (@poll) {
             my ($fd, $state) = splice(@poll, 0, 2);
-            next unless $state;
-
-            event_step($DescriptorMap{$fd});
+            $DescriptorMap{$fd}->event_step if $state;
         }
 
         return unless PostEventLoop();
@@ -345,8 +326,7 @@ sub KQueueEventLoop {
         }
 
         foreach my $kev (@ret) {
-            my ($fd, $filter, $flags, $fflags) = @$kev;
-            event_step($DescriptorMap{$fd});
+            $DescriptorMap{$kev->[0]}->event_step;
         }
         return unless PostEventLoop();
     }
@@ -647,27 +627,6 @@ sub on_incomplete_write {
     $self->watch_write(1);
 }
 
-=head2 (VIRTUAL) C<< $obj->event_read() >>
-
-Readable event handler. Concrete deriviatives of PublicInbox::DS should
-provide an implementation of this. The default implementation is a noop
-if called.
-
-=cut
-sub event_read {} # noop
-
-=head2 C<< $obj->event_write() >>
-
-Writable event handler. Concrete deriviatives of PublicInbox::DS may wish to
-provide an implementation of this. The default implementation calls
-C<write()> with an C<undef>.
-
-=cut
-sub event_write {
-    my $self = shift;
-    $self->write(undef);
-}
-
 =head2 C<< $obj->watch_read( $boolean ) >>
 
 Turn 'readable' event notification on or off.
diff --git a/lib/PublicInbox/EvCleanup.pm b/lib/PublicInbox/EvCleanup.pm
index f76fb681..c64e2388 100644
--- a/lib/PublicInbox/EvCleanup.pm
+++ b/lib/PublicInbox/EvCleanup.pm
@@ -25,7 +25,7 @@ sub once_init () {
         fcntl($w, 1031, 4096) if $^O eq 'linux'; # 1031: F_SETPIPE_SZ
         $self->SUPER::new($w);
 
-        # always writable, since PublicInbox::EvCleanup::event_write
+        # always writable, since PublicInbox::EvCleanup::event_step
         # never drains wbuf.  We can avoid wasting a hash slot by
         # stuffing the read-end of the pipe into the never-to-be-touched
         # wbuf
@@ -57,7 +57,7 @@ sub _run_later () {
 }
 
 # Called by PublicInbox::DS
-sub event_write {
+sub event_step {
         my ($self) = @_;
         $self->watch_write(0);
         _run_asap();
diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index 4fbc34ee..45bf23ec 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -64,10 +64,18 @@ sub new ($$$) {
         $self;
 }
 
-sub event_read { # called by PublicInbox::DS
+sub event_step { # called by PublicInbox::DS
         my ($self) = @_;
 
-        return event_read_input($self) if defined $self->{env};
+        my $wbuf = $self->{wbuf};
+        if (@$wbuf) {
+                $self->write(undef);
+                return if $self->{closed} || scalar(@$wbuf);
+        }
+        # only read more requests if we've drained the write buffer,
+        # otherwise we can be buffering infinitely w/o backpressure
+
+        return read_input($self) if defined $self->{env};
 
         my $off = length($self->{rbuf});
         my $r = sysread($self->{sock}, $self->{rbuf}, 8192, $off);
@@ -101,13 +109,14 @@ sub rbuf_process {
         my $len = input_prepare($self, \%env);
         defined $len or return write_err($self, undef); # EMFILE/ENFILE
 
-        $len ? event_read_input($self) : app_dispatch($self);
+        $len ? read_input($self) : app_dispatch($self);
 }
 
-sub event_read_input ($) {
+sub read_input ($) {
         my ($self) = @_;
         my $env = $self->{env};
-        return event_read_input_chunked($self) if env_chunked($env);
+        return if $env->{REMOTE_ADDR}; # in app dispatch
+        return read_input_chunked($self) if env_chunked($env);
 
         # env->{CONTENT_LENGTH} (identity)
         my $sock = $self->{sock};
@@ -229,7 +238,6 @@ sub identity_wcb ($) {
 
 sub next_request ($) {
         my ($self) = @_;
-        $self->watch_write(0);
         if ($self->{rbuf} eq '') { # wait for next request
                 $self->watch_read(1);
         } else { # avoid recursion for pipelined requests
@@ -392,7 +400,7 @@ sub write_in_full {
         $rv
 }
 
-sub event_read_input_chunked { # unlikely...
+sub read_input_chunked { # unlikely...
         my ($self) = @_;
         my $input = $self->{env}->{'psgi.input'};
         my $sock = $self->{sock};
diff --git a/lib/PublicInbox/HTTPD/Async.pm b/lib/PublicInbox/HTTPD/Async.pm
index 4d0c8d5b..604627ab 100644
--- a/lib/PublicInbox/HTTPD/Async.pm
+++ b/lib/PublicInbox/HTTPD/Async.pm
@@ -75,7 +75,7 @@ sub async_pass {
         $self->{cb} = main_cb($http, $fh, $bref);
 }
 
-sub event_read { $_[0]->{cb}->(@_) }
+sub event_step { $_[0]->{cb}->(@_) }
 
 sub close {
         my $self = shift;
diff --git a/lib/PublicInbox/Listener.pm b/lib/PublicInbox/Listener.pm
index a75a6fd9..6ee3abb1 100644
--- a/lib/PublicInbox/Listener.pm
+++ b/lib/PublicInbox/Listener.pm
@@ -23,7 +23,7 @@ sub new ($$$) {
         $self
 }
 
-sub event_read {
+sub event_step {
         my ($self) = @_;
         my $sock = $self->{sock};
 
diff --git a/lib/PublicInbox/NNTP.pm b/lib/PublicInbox/NNTP.pm
index fa412f8c..796ac74d 100644
--- a/lib/PublicInbox/NNTP.pm
+++ b/lib/PublicInbox/NNTP.pm
@@ -53,7 +53,7 @@ sub next_tick () {
                 } else {
                         # pipelined request, we bypassed socket-readiness
                         # checks to get here:
-                        event_read($nntp);
+                        event_step($nntp);
 
                         # maybe there's more pipelined data, or we'll have
                         # to register it for socket-readiness notifications
@@ -964,17 +964,20 @@ sub do_more ($$) {
         do_write($self, $data);
 }
 
-sub event_write {
+sub event_step {
         my ($self) = @_;
-        update_idle_time($self);
-        # only continue watching for readability when we are done writing:
-        if ($self->write(undef) == 1 && !$self->{long_res}) {
-                $self->watch_read(1);
+        return if $self->{closed};
+
+        my $wbuf = $self->{wbuf};
+        if (@$wbuf) {
+                update_idle_time($self);
+                $self->write(undef);
+                return if $self->{closed} || scalar(@$wbuf);
         }
-}
+        return if $self->{long_res};
+        # only read more requests if we've drained the write buffer,
+        # otherwise we can be buffering infinitely w/o backpressure
 
-sub event_read {
-        my ($self) = @_;
         use constant LINE_MAX => 512; # RFC 977 section 2.3
         my $rbuf = \($self->{rbuf});
         my $r;
diff --git a/lib/PublicInbox/ParentPipe.pm b/lib/PublicInbox/ParentPipe.pm
index 25f13a82..a9f05fc1 100644
--- a/lib/PublicInbox/ParentPipe.pm
+++ b/lib/PublicInbox/ParentPipe.pm
@@ -16,6 +16,6 @@ sub new ($$$) {
         $self;
 }
 
-sub event_read { $_[0]->{cb}->($_[0]) }
+sub event_step { $_[0]->{cb}->($_[0]) }
 
 1;