ds: handle deferred DS->close after timers

Our hacks in EvCleanup::next_tick and EvCleanup::asap were due
to the fact "closed" sockets were deferred and could not wake
up the event loop, causing certain actions to be delayed until
an event fired.

Instead, ensure we don't sleep if there are pending sockets to
close.

We can then remove most of the EvCleanup stuff

While we're at it, split out immediate timer handling into a
separate array so we don't need to deal with time calculations
for the event loop.

1 files changed, 9 insertions, 71 deletions

diff --git a/lib/PublicInbox/EvCleanup.pm b/lib/PublicInbox/EvCleanup.pm
index 33b54ebc..be6672ed 100644
--- a/lib/PublicInbox/EvCleanup.pm
+++ b/lib/PublicInbox/EvCleanup.pm
@@ -1,80 +1,23 @@
-# Copyright (C) 2016-2018 all contributors <meta@public-inbox.org>
+# Copyright (C) 2016-2019 all contributors <meta@public-inbox.org>
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 
-# event cleanups (currently for PublicInbox::DS)
+# event cleanups (for PublicInbox::DS)
 package PublicInbox::EvCleanup;
 use strict;
 use warnings;
-use base qw(PublicInbox::DS);
-use PublicInbox::Syscall qw(EPOLLOUT EPOLLONESHOT);
+require PublicInbox::DS;
 
+# this only runs under public-inbox-{httpd/nntpd}, not generic PSGI servers
 my $ENABLED;
 sub enabled { $ENABLED }
 sub enable { $ENABLED = 1 }
-my $singleton;
-my $asapq = [ [], undef ];
-my $nextq = [ [], undef ];
 my $laterq = [ [], undef ];
 
-sub once_init () {
-        my $self = fields::new('PublicInbox::EvCleanup');
-        my ($r, $w);
-
-        # This is a dummy pipe which is always writable so it can always
-        # fires in the next event loop iteration.
-        pipe($r, $w) or die "pipe: $!";
-        fcntl($w, 1031, 4096) if $^O eq 'linux'; # 1031: F_SETPIPE_SZ
-        $self->SUPER::new($w, 0);
-
-        # 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
-        $self->{wbuf} = $r;
-        $self;
-}
-
-sub _run_all ($) {
-        my ($q) = @_;
-
-        my $run = $q->[0];
-        $q->[0] = [];
-        $q->[1] = undef;
-        $_->() foreach @$run;
-}
-
-# ensure PublicInbox::DS::ToClose processing after timers fire
-sub _asap_close () { $asapq->[1] ||= _asap_timer() }
-
-# Called by PublicInbox::DS
-sub event_step { _run_all($asapq) }
-
-sub _run_next () {
-        _run_all($nextq);
-        _asap_close();
-}
-
 sub _run_later () {
-        _run_all($laterq);
-        _asap_close();
-}
-
-sub _asap_timer () {
-        $singleton ||= once_init();
-        $singleton->watch(EPOLLOUT|EPOLLONESHOT);
-        1;
-}
-
-sub asap ($) {
-        my ($cb) = @_;
-        push @{$asapq->[0]}, $cb;
-        $asapq->[1] ||= _asap_timer();
-}
-
-sub next_tick ($) {
-        my ($cb) = @_;
-        push @{$nextq->[0]}, $cb;
-        $nextq->[1] ||= PublicInbox::DS->AddTimer(0, *_run_next);
+        my $run = $laterq->[0];
+        $laterq->[0] = [];
+        $laterq->[1] = undef;
+        $_->() foreach @$run;
 }
 
 sub later ($) {
@@ -83,10 +26,5 @@ sub later ($) {
         $laterq->[1] ||= PublicInbox::DS->AddTimer(60, *_run_later);
 }
 
-END {
-        event_step();
-        _run_all($nextq);
-        _run_all($laterq);
-}
-
+END { _run_later() }
 1;