about summary refs log tree commit homepage
path: root/lib/PublicInbox/DS.pm
diff options
context:
space:
mode:
Diffstat (limited to 'lib/PublicInbox/DS.pm')
-rw-r--r--lib/PublicInbox/DS.pm433
1 files changed, 205 insertions, 228 deletions
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index 77e2e5e9..8bc8cfb7 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -24,29 +24,30 @@ use strict;
 use v5.10.1;
 use parent qw(Exporter);
 use bytes qw(length substr); # FIXME(?): needed for PublicInbox::NNTP
-use POSIX qw(WNOHANG sigprocmask SIG_SETMASK);
+use POSIX qw(WNOHANG sigprocmask SIG_SETMASK SIG_UNBLOCK);
 use Fcntl qw(SEEK_SET :DEFAULT O_APPEND);
 use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC);
 use Scalar::Util qw(blessed);
-use PublicInbox::Syscall qw(:epoll);
+use PublicInbox::Syscall qw(%SIGNUM
+        EPOLLIN EPOLLOUT EPOLLONESHOT EPOLLEXCLUSIVE);
 use PublicInbox::Tmpfile;
-use Errno qw(EAGAIN EINVAL);
+use PublicInbox::Select;
+use Errno qw(EAGAIN EINVAL ECHILD);
 use Carp qw(carp croak);
-our @EXPORT_OK = qw(now msg_more dwaitpid add_timer add_uniq_timer);
+use autodie qw(fork);
+our @EXPORT_OK = qw(now msg_more awaitpid add_timer add_uniq_timer);
 
-my %Stack;
 my $nextq; # queue for next_tick
-my $wait_pids; # list of [ pid, callback, callback_arg ]
 my $reap_armed;
-my $ToClose; # sockets to close when event loop is done
-our (
-     %DescriptorMap,             # fd (num) -> PublicInbox::DS object
-     $Epoll,                     # Global epoll fd (or DSKQXS ref)
-     $ep_io,                     # IO::Handle for Epoll
+my @active; # FDs (or objs) returned by epoll/kqueue
+our (%AWAIT_PIDS, # pid => [ $callback, @args ]
+        $cur_runq, # only set inside next_tick
+     @FD_MAP, # fd (num) -> PublicInbox::DS object
+     $Poller, # global Select, Epoll, DSPoll, or DSKQXS ref
 
-     $PostLoopCallback,          # subref to call at the end of each loop, if defined (global)
+     @post_loop_do,              # subref + args to call at the end of each loop
 
-     $LoopTimeout,               # timeout of event loop in milliseconds
+     $loop_timeout,               # timeout of event loop in milliseconds
      @Timers,                    # timers
      %UniqTimer,
      $in_loop,
@@ -54,6 +55,9 @@ our (
 
 Reset();
 
+# clobber everything explicitly to avoid DESTROY ordering problems w/ DBI
+END { Reset() }
+
 #####################################################################
 ### C L A S S   M E T H O D S
 #####################################################################
@@ -64,37 +68,32 @@ Reset all state
 
 =cut
 sub Reset {
+        $Poller = bless [], 'PublicInbox::DummyPoller';
         do {
                 $in_loop = undef; # first in case DESTROY callbacks use this
-                %DescriptorMap = ();
+                # clobbering $Poller may call DSKQXS::DESTROY,
+                # we must always have this set to something to avoid
+                # needing branches before ep_del/ep_mod calls (via ->close).
+                @FD_MAP = ();
                 @Timers = ();
                 %UniqTimer = ();
-                $PostLoopCallback = undef;
-
-                # we may be iterating inside one of these on our stack
-                my @q = delete @Stack{keys %Stack};
-                for my $q (@q) { @$q = () }
-                $wait_pids = $nextq = $ToClose = undef;
-                $ep_io = undef; # closes real $Epoll FD
-                $Epoll = undef; # may call DSKQXS::DESTROY
-        } while (@Timers || keys(%Stack) || $nextq || $wait_pids ||
-                $ToClose || keys(%DescriptorMap) ||
-                $PostLoopCallback || keys(%UniqTimer));
+                @post_loop_do = ();
+
+                # we may be called from an *atfork_child inside next_tick:
+                @$cur_runq = () if $cur_runq;
+                @active = ();
+                $nextq = undef; # may call ep_del
+                %AWAIT_PIDS = ();
+        } while (@Timers || $nextq || keys(%AWAIT_PIDS) ||
+                @active || @FD_MAP ||
+                @post_loop_do || keys(%UniqTimer) ||
+                scalar(@{$cur_runq // []})); # do not vivify cur_runq
 
         $reap_armed = undef;
-        $LoopTimeout = -1;  # no timeout by default
+        $loop_timeout = -1;  # no timeout by default
+        $Poller = PublicInbox::Select->new;
 }
 
-=head2 C<< CLASS->SetLoopTimeout( $timeout ) >>
-
-Set the loop timeout for the event loop to some value in milliseconds.
-
-A timeout of 0 (zero) means poll forever. A timeout of -1 means poll and return
-immediately.
-
-=cut
-sub SetLoopTimeout { $LoopTimeout = $_[1] + 0 }
-
 sub _add_named_timer {
         my ($name, $secs, $coderef, @args) = @_;
         my $fire_time = now() + $secs;
@@ -124,49 +123,35 @@ sub add_uniq_timer { # ($name, $secs, $coderef, @args) = @_;
         $UniqTimer{$_[0]} //= _add_named_timer(@_);
 }
 
-# caller sets return value to $Epoll
+# caller sets return value to $Poller
 sub _InitPoller () {
-        if (PublicInbox::Syscall::epoll_defined())  {
-                my $fd = epoll_create();
-                die "epoll_create: $!" if $fd < 0;
-                open($ep_io, '+<&=', $fd) or return;
-                my $fl = fcntl($ep_io, F_GETFD, 0);
-                fcntl($ep_io, F_SETFD, $fl | FD_CLOEXEC);
-                $fd;
-        } else {
-                my $cls;
-                for (qw(DSKQXS DSPoll)) {
-                        $cls = "PublicInbox::$_";
-                        last if eval "require $cls";
-                }
-                $cls->import(qw(epoll_ctl epoll_wait));
-                $cls->new;
+        my @try = ($^O eq 'linux' ? 'Epoll' : 'DSKQXS');
+        my $cls;
+        for (@try, 'DSPoll') {
+                $cls = "PublicInbox::$_";
+                last if eval "require $cls";
         }
+        $cls->new;
 }
 
 sub now () { clock_gettime(CLOCK_MONOTONIC) }
 
 sub next_tick () {
-        my $q = $nextq or return;
+        $cur_runq = $nextq or return;
         $nextq = undef;
-        $Stack{cur_runq} = $q;
-        for my $obj (@$q) {
+        while (my $obj = shift @$cur_runq) {
                 # avoid "ref" on blessed refs to workaround a Perl 5.16.3 leak:
                 # https://rt.perl.org/Public/Bug/Display.html?id=114340
-                if (blessed($obj)) {
-                        $obj->event_step;
-                } else {
-                        $obj->();
-                }
+                blessed($obj) ? $obj->event_step : $obj->();
         }
-        delete $Stack{cur_runq};
+        1;
 }
 
 # runs timers and returns milliseconds for next one, or next event loop
 sub RunTimers {
-        next_tick();
+        my $ran = next_tick();
 
-        return (($nextq || $ToClose) ? 0 : $LoopTimeout) unless @Timers;
+        return ($nextq || $ran ? 0 : $loop_timeout) unless @Timers;
 
         my $now = now();
 
@@ -175,60 +160,64 @@ sub RunTimers {
                 my $to_run = shift(@Timers);
                 delete $UniqTimer{$to_run->[1] // ''};
                 $to_run->[2]->(@$to_run[3..$#$to_run]);
+                $ran = 1;
         }
 
         # timers may enqueue into nextq:
-        return 0 if ($nextq || $ToClose);
+        return 0 if $nextq || $ran;
 
-        return $LoopTimeout unless @Timers;
+        return $loop_timeout unless @Timers;
 
         # convert time to an even number of milliseconds, adding 1
         # extra, otherwise floating point fun can occur and we'll
         # call RunTimers like 20-30 times, each returning a timeout
         # of 0.0000212 seconds
-        my $timeout = int(($Timers[0][0] - $now) * 1000) + 1;
+        my $t = int(($Timers[0][0] - $now) * 1000) + 1;
 
         # -1 is an infinite timeout, so prefer a real timeout
-        ($LoopTimeout < 0 || $LoopTimeout >= $timeout) ? $timeout : $LoopTimeout
+        ($loop_timeout < 0 || $loop_timeout >= $t) ? $t : $loop_timeout
 }
 
 sub sig_setmask { sigprocmask(SIG_SETMASK, @_) or die "sigprocmask: $!" }
 
-sub block_signals () {
-        my $oldset = POSIX::SigSet->new;
+# ensure we detect bugs, HW problems and user rlimits
+our @UNBLOCKABLE = (POSIX::SIGABRT, POSIX::SIGBUS, POSIX::SIGFPE,
+        POSIX::SIGILL, POSIX::SIGSEGV, POSIX::SIGXCPU, POSIX::SIGXFSZ);
+
+sub block_signals { # anything in @_ stays unblocked
         my $newset = POSIX::SigSet->new;
         $newset->fillset or die "fillset: $!";
+        for (@_, @UNBLOCKABLE) { $newset->delset($_) or die "delset($_): $!" }
+        my $oldset = POSIX::SigSet->new;
         sig_setmask($newset, $oldset);
         $oldset;
 }
 
-# We can't use waitpid(-1) safely here since it can hit ``, system(),
-# and other things.  So we scan the $wait_pids list, which is hopefully
-# not too big.  We keep $wait_pids small by not calling dwaitpid()
-# until we've hit EOF when reading the stdout of the child.
+sub await_cb ($;@) {
+        my ($pid, @cb_args) = @_;
+        my $cb = shift @cb_args or return;
+        eval { $cb->($pid, @cb_args) };
+        warn "E: awaitpid($pid): $@" if $@;
+}
 
+# This relies on our Perl process being single-threaded, or at least
+# no threads spawning and waiting on processes (``, system(), etc...)
+# Threads are officially discouraged by the Perl5 team, and I expect
+# that to remain the case.
 sub reap_pids {
         $reap_armed = undef;
-        my $tmp = $wait_pids or return;
-        $wait_pids = undef;
-        $Stack{reap_runq} = $tmp;
-        my $oldset = block_signals();
-        foreach my $ary (@$tmp) {
-                my ($pid, $cb, $arg) = @$ary;
-                my $ret = waitpid($pid, WNOHANG);
-                if ($ret == 0) {
-                        push @$wait_pids, $ary; # autovivifies @$wait_pids
-                } elsif ($ret == $pid) {
-                        if ($cb) {
-                                eval { $cb->($arg, $pid) };
-                                warn "E: dwaitpid($pid) in_loop: $@" if $@;
-                        }
-                } else {
-                        warn "waitpid($pid, WNOHANG) = $ret, \$!=$!, \$?=$?";
+        while (1) {
+                my $pid = waitpid(-1, WNOHANG) or return;
+                if (defined(my $cb_args = delete $AWAIT_PIDS{$pid})) {
+                        await_cb($pid, @$cb_args) if $cb_args;
+                } elsif ($pid == -1 && $! == ECHILD) {
+                        return requeue(\&dflush); # force @post_loop_do to run
+                } elsif ($pid > 0) {
+                        warn "W: reaped unknown PID=$pid: \$?=$?\n";
+                } else { # does this happen?
+                        return warn("W: waitpid(-1, WNOHANG) => $pid ($!)");
                 }
         }
-        sig_setmask($oldset);
-        delete $Stack{reap_runq};
 }
 
 # reentrant SIGCHLD handler (since reap_pids is not reentrant)
@@ -236,81 +225,82 @@ sub enqueue_reap () { $reap_armed //= requeue(\&reap_pids) }
 
 sub in_loop () { $in_loop }
 
+# use inside @post_loop_do, returns number of busy clients
+sub close_non_busy () {
+        my $n = 0;
+        for my $s (grep defined, @FD_MAP) {
+                # close as much as possible, early as possible
+                ($s->busy ? ++$n : $s->close) if $s->can('busy');
+        }
+        $n;
+}
+
 # Internal function: run the post-event callback, send read events
 # for pushed-back data, and close pending connections.  returns 1
 # if event loop should continue, or 0 to shut it all down.
 sub PostEventLoop () {
-        # now we can close sockets that wanted to close during our event
-        # processing.  (we didn't want to close them during the loop, as we
-        # didn't want fd numbers being reused and confused during the event
-        # loop)
-        if (my $close_now = $ToClose) {
-                $ToClose = undef; # will be autovivified on push
-                @$close_now = map { fileno($_) } @$close_now;
-
-                # ->DESTROY methods may populate ToClose
-                delete @DescriptorMap{@$close_now};
-        }
-
         # by default we keep running, unless a postloop callback cancels it
-        $PostLoopCallback ? $PostLoopCallback->(\%DescriptorMap) : 1;
+        @post_loop_do ? $post_loop_do[0]->(@post_loop_do[1..$#post_loop_do]) : 1
+}
+
+sub sigset_prep ($$$) {
+        my ($sig, $init, $each) = @_; # $sig: { signame => whatever }
+        my $ret = POSIX::SigSet->new;
+        $ret->$init or die "$init: $!";
+        for my $s (keys %$sig) {
+                my $num = $SIGNUM{$s} // POSIX->can("SIG$s")->();
+                $ret->$each($num) or die "$each ($s => $num): $!";
+        }
+        for (@UNBLOCKABLE) { $ret->$each($_) or die "$each ($_): $!" }
+        $ret;
 }
 
+sub allowset ($) { sigset_prep $_[0], 'fillset', 'delset' }
+sub unblockset ($) { sigset_prep $_[0], 'emptyset', 'addset' }
+
 # Start processing IO events. In most daemon programs this never exits. See
-# C<PostLoopCallback> for how to exit the loop.
+# C<post_loop_do> for how to exit the loop.
 sub event_loop (;$$) {
         my ($sig, $oldset) = @_;
-        $Epoll //= _InitPoller();
+        $Poller //= _InitPoller();
         require PublicInbox::Sigfd if $sig;
-        my $sigfd = PublicInbox::Sigfd->new($sig, 1) if $sig;
+        my $sigfd = $sig ? PublicInbox::Sigfd->new($sig) : undef;
+        if ($sigfd && $sigfd->{is_kq}) {
+                my $tmp = allowset($sig);
+                local @SIG{keys %$sig} = values(%$sig);
+                sig_setmask($tmp, my $old = POSIX::SigSet->new);
+                # Unlike Linux signalfd, EVFILT_SIGNAL can't handle
+                # signals received before the filter is created,
+                # so we peek at signals here.
+                sig_setmask($old);
+        }
         local @SIG{keys %$sig} = values(%$sig) if $sig && !$sigfd;
         local $SIG{PIPE} = 'IGNORE';
         if (!$sigfd && $sig) {
                 # wake up every second to accept signals if we don't
                 # have signalfd or IO::KQueue:
-                sig_setmask($oldset);
-                PublicInbox::DS->SetLoopTimeout(1000);
+                sig_setmask($oldset) if $oldset;
+                sigprocmask(SIG_UNBLOCK, unblockset($sig)) or
+                        die "SIG_UNBLOCK: $!";
+                $loop_timeout = 1000;
         }
         $_[0] = $sigfd = $sig = undef; # $_[0] == sig
         local $in_loop = 1;
-        my @events;
         do {
                 my $timeout = RunTimers();
 
-                # get up to 1000 events
-                epoll_wait($Epoll, 1000, $timeout, \@events);
-                for my $fd (@events) {
-                        # 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.
-
-                        # guard stack-not-refcounted w/ Carp + @DB::args
-                        my $obj = $DescriptorMap{$fd};
+                # grab whatever FDs are ready
+                $Poller->ep_wait($timeout, \@active);
+
+                # map all FDs to their associated Perl object
+                @active = @FD_MAP[@active];
+
+                while (my $obj = shift @active) {
                         $obj->event_step;
                 }
         } while (PostEventLoop());
 }
 
-=head2 C<< CLASS->SetPostLoopCallback( CODEREF ) >>
-
-Sets post loop callback function.  Pass a subref and it will be
-called every time the event loop finishes.
-
-Return 1 (or any true value) from the sub to make the loop continue, 0 or false
-and it will exit.
-
-The callback function will be passed two parameters: \%DescriptorMap
-
-=cut
-sub SetPostLoopCallback {
-    my ($class, $ref) = @_;
-
-    # global callback
-    $PostLoopCallback = (defined $ref && ref $ref eq 'CODE') ? $ref : undef;
-}
-
 #####################################################################
 ### PublicInbox::DS-the-object code
 #####################################################################
@@ -332,19 +322,19 @@ sub new {
     $self->{sock} = $sock;
     my $fd = fileno($sock);
 
-    $Epoll //= _InitPoller();
+    $Poller //= _InitPoller();
 retry:
-    if (epoll_ctl($Epoll, EPOLL_CTL_ADD, $fd, $ev)) {
+    if ($Poller->ep_add($sock, $ev)) {
         if ($! == EINVAL && ($ev & EPOLLEXCLUSIVE)) {
             $ev &= ~EPOLLEXCLUSIVE;
             goto retry;
         }
         die "EPOLL_CTL_ADD $self/$sock/$fd: $!";
     }
-    croak("FD:$fd in use by $DescriptorMap{$fd} (for $self/$sock)")
-        if defined($DescriptorMap{$fd});
+    defined($FD_MAP[$fd]) and
+                croak("BUG: FD:$fd in use by $FD_MAP[$fd] (for $self/$sock)");
 
-    $DescriptorMap{$fd} = $self;
+    $FD_MAP[$fd] = $self;
 }
 
 # for IMAP, NNTP, and POP3 which greet clients upon connect
@@ -353,8 +343,7 @@ sub greet {
         my $ev = EPOLLIN;
         my $wbuf;
         if ($sock->can('accept_SSL') && !$sock->accept_SSL) {
-                return CORE::close($sock) if $! != EAGAIN;
-                $ev = PublicInbox::TLS::epollbit() or return CORE::close($sock);
+                return if $! != EAGAIN || !($ev = PublicInbox::TLS::epollbit());
                 $wbuf = [ \&accept_tls_step, $self->can('do_greet')];
         }
         new($self, $sock, $ev | EPOLLONESHOT);
@@ -366,46 +355,21 @@ sub greet {
         $self;
 }
 
-#####################################################################
-### I N S T A N C E   M E T H O D S
-#####################################################################
-
 sub requeue ($) { push @$nextq, $_[0] } # autovivifies
 
-=head2 C<< $obj->close >>
-
-Close the socket.
-
-=cut
+# drop the IO::Handle ref, true if successful, false if not (or already dropped)
+# (this is closer to CORE::close than Danga::Socket::close)
 sub close {
-    my ($self) = @_;
-    my $sock = delete $self->{sock} or return;
-
-    # we need to flush our write buffer, as there may
-    # be self-referential closures (sub { $client->close })
-    # preventing the object from being destroyed
-    delete $self->{wbuf};
-
-    # if we're using epoll, we have to remove this from our epoll fd so we stop getting
-    # notifications about it
-    my $fd = fileno($sock);
-    epoll_ctl($Epoll, EPOLL_CTL_DEL, $fd, 0) and
-        croak("EPOLL_CTL_DEL($self/$sock): $!");
-
-    # we explicitly don't delete from DescriptorMap here until we
-    # actually close the socket, as we might be in the middle of
-    # processing an epoll_wait/etc that returned hundreds of fds, one
-    # of which is not yet processed and is what we're closing.  if we
-    # keep it in DescriptorMap, then the event harnesses can just
-    # looked at $pob->{sock} == undef and ignore it.  but if it's an
-    # un-accounted for fd, then it (understandably) freak out a bit
-    # and emit warnings, thinking their state got off.
+        my ($self) = @_;
+        my $sock = delete $self->{sock} or return;
 
-    # defer closing the actual socket until the event loop is done
-    # processing this round of events.  (otherwise we might reuse fds)
-    push @$ToClose, $sock; # autovivifies $ToClose
+        # we need to clear our write buffer, as there may
+        # be self-referential closures (sub { $client->close })
+        # preventing the object from being destroyed
+        delete $self->{wbuf};
+        $FD_MAP[fileno($sock)] = undef;
 
-    return 0;
+        !$Poller->ep_del($sock); # stop getting notifications
 }
 
 # portable, non-thread-safe sendfile emulation (no pread, yet)
@@ -451,8 +415,7 @@ next_buf:
                         shift @$wbuf;
                         goto next_buf;
                     }
-                } elsif ($! == EAGAIN) {
-                    my $ev = epbit($sock, EPOLLOUT) or return $self->close;
+                } elsif ($! == EAGAIN && (my $ev = epbit($sock, EPOLLOUT))) {
                     epwait($sock, $ev | EPOLLONESHOT);
                     return 0;
                 } else {
@@ -482,28 +445,28 @@ sub rbuf_idle ($$) {
     }
 }
 
+# returns true if bytes are read, false otherwise
 sub do_read ($$$;$) {
-    my ($self, $rbuf, $len, $off) = @_;
-    my $r = sysread(my $sock = $self->{sock}, $$rbuf, $len, $off // 0);
-    return ($r == 0 ? $self->close : $r) if defined $r;
-    # common for clients to break connections without warning,
-    # would be too noisy to log here:
-    if ($! == EAGAIN) {
-        my $ev = epbit($sock, EPOLLIN) or return $self->close;
-        epwait($sock, $ev | EPOLLONESHOT);
-        rbuf_idle($self, $rbuf);
-        0;
-    } else {
-        $self->close;
-    }
+        my ($self, $rbuf, $len, $off) = @_;
+        my ($ev, $r, $s);
+        $r = sysread($s = $self->{sock}, $$rbuf, $len, $off // 0) and return $r;
+
+        if (!defined($r) && $! == EAGAIN && ($ev = epbit($s, EPOLLIN))) {
+                epwait($s, $ev | EPOLLONESHOT);
+                rbuf_idle($self, $rbuf);
+        } else {
+                $self->close;
+        }
+        undef;
 }
 
 # drop the socket if we hit unrecoverable errors on our system which
 # require BOFH attention: ENOSPC, EFBIG, EIO, EMFILE, ENFILE...
 sub drop {
-    my $self = shift;
-    carp(@_);
-    $self->close;
+        my $self = shift;
+        carp(@_);
+        $self->close;
+        undef;
 }
 
 sub tmpio ($$$) {
@@ -545,7 +508,8 @@ sub write {
             push @$wbuf, $bref;
         } else {
             my $tmpio = $wbuf->[-1];
-            if ($tmpio && !defined($tmpio->[2])) { # append to tmp file buffer
+            if (ref($tmpio) eq 'ARRAY' && !defined($tmpio->[2])) {
+                # append to tmp file buffer
                 $tmpio->[0]->print($$bref) or return drop($self, "print: $!");
             } else {
                 my $tmpio = tmpio($self, $bref, 0) or return 0;
@@ -607,9 +571,8 @@ sub msg_more ($$) {
 }
 
 sub epwait ($$) {
-    my ($sock, $ev) = @_;
-    epoll_ctl($Epoll, EPOLL_CTL_MOD, fileno($sock), $ev) and
-        croak("EPOLL_CTL_MOD($sock): $!");
+        my ($io, $ev) = @_;
+        $Poller->ep_mod($io, $ev) and croak("EPOLL_CTL_MOD($io): $!");
 }
 
 # return true if complete, false if incomplete (or failure)
@@ -624,7 +587,7 @@ sub accept_tls_step ($) {
     0;
 }
 
-# return true if complete, false if incomplete (or failure)
+# return value irrelevant
 sub shutdn_tls_step ($) {
     my ($self) = @_;
     my $sock = $self->{sock} or return;
@@ -633,19 +596,14 @@ sub shutdn_tls_step ($) {
     my $ev = PublicInbox::TLS::epollbit() or return $self->close;
     epwait($sock, $ev | EPOLLONESHOT);
     unshift(@{$self->{wbuf}}, \&shutdn_tls_step); # autovivifies
-    0;
 }
 
 # don't bother with shutdown($sock, 2), we don't fork+exec w/o CLOEXEC
 # or fork w/o exec, so no inadvertent socket sharing
 sub shutdn ($) {
-    my ($self) = @_;
-    my $sock = $self->{sock} or return;
-    if ($sock->can('stop_SSL')) {
-        shutdn_tls_step($self);
-    } else {
-        $self->close;
-    }
+        my ($self) = @_;
+        my $sock = $self->{sock} or return;
+        $sock->can('stop_SSL') ? shutdn_tls_step($self) : $self->close;
 }
 
 sub dflush {} # overridden by DSdeflate
@@ -660,8 +618,8 @@ sub long_step {
         if ($@ || !$self->{sock}) { # something bad happened...
                 delete $self->{long_cb};
                 my $elapsed = now() - $t0;
-                $@ and $self->err("%s during long response[$fd] - %0.6f",
-                                    $@, $elapsed);
+                $@ and warn("$@ during long response[$fd] - ",
+                                sprintf('%0.6f', $elapsed),"\n");
                 $self->out(" deferred[$fd] aborted - %0.6f", $elapsed);
                 $self->close;
         } elsif ($more) { # $self->{wbuf}:
@@ -671,7 +629,6 @@ sub long_step {
                 delete $self->{long_cb};
                 $self->long_response_done;
                 my $elapsed = now() - $t0;
-                my $fd = fileno($self->{sock});
                 $self->out(" deferred[$fd] done - %0.6f", $elapsed);
                 my $wbuf = $self->{wbuf}; # do NOT autovivify
                 requeue($self) unless $wbuf && @$wbuf;
@@ -688,7 +645,7 @@ sub requeue_once {
         # but only after all pending writes are done.
         # autovivify wbuf.  wbuf may be populated by $cb,
         # no need to rearm if so: (push returns new size of array)
-        requeue($self) if push(@{$self->{wbuf}}, \&long_step) == 1;
+        $self->requeue if push(@{$self->{wbuf}}, \&long_step) == 1;
 }
 
 sub long_response ($$;@) {
@@ -702,25 +659,45 @@ sub long_response ($$;@) {
         undef;
 }
 
-sub dwaitpid ($;$$) {
-        my ($pid, $cb, $arg) = @_;
-        if ($in_loop) {
-                push @$wait_pids, [ $pid, $cb, $arg ];
-                # We could've just missed our SIGCHLD, cover it, here:
-                enqueue_reap();
-        } else {
+sub awaitpid {
+        my ($pid, @cb_args) = @_; # @cb_args = ($cb, @args), $cb may be undef
+        $AWAIT_PIDS{$pid} = \@cb_args if @cb_args;
+        # provide synchronous API
+        if (defined(wantarray) || (!$in_loop && !@cb_args)) {
                 my $ret = waitpid($pid, 0);
                 if ($ret == $pid) {
-                        if ($cb) {
-                                eval { $cb->($arg, $pid) };
-                                carp "E: dwaitpid($pid) !in_loop: $@" if $@;
-                        }
+                        my $cb_args = delete $AWAIT_PIDS{$pid};
+                        @cb_args = @$cb_args if !@cb_args && $cb_args;
+                        await_cb($pid, @cb_args);
                 } else {
-                        carp "waitpid($pid, 0) = $ret, \$!=$!, \$?=$?";
+                        carp "waitpid($pid) => $ret ($!)";
+                        delete $AWAIT_PIDS{$pid};
                 }
+                return $ret;
+        } elsif ($in_loop) { # We could've just missed our SIGCHLD, cover it, here:
+                enqueue_reap();
         }
 }
 
+sub do_fork () {
+        my $seed = rand(0xffffffff);
+        my $pid = fork;
+        if ($pid == 0) {
+                srand($seed);
+                eval { Net::SSLeay::randomize() };
+                Reset();
+        }
+        $pid;
+}
+
+package PublicInbox::DummyPoller; # only used during Reset
+use v5.12;
+
+sub ep_del {}
+no warnings 'once';
+*ep_add = \&ep_del;
+*ep_mod = \&ep_del;
+
 1;
 
 =head1 AUTHORS (Danga::Socket)