]> Sergey Matveev's repositories - public-inbox.git/blobdiff - lib/PublicInbox/DS.pm
ds: add missing label for systems w/o EPOLLEXCLUSIVE
[public-inbox.git] / lib / PublicInbox / DS.pm
index 4947192f8e4d604f1c1563595890326b8b31b174..d0caa5e73102f13ea27bf455e242b19cbb72220e 100644 (file)
@@ -3,50 +3,56 @@
 #
 # This license differs from the rest of public-inbox
 #
-# This is a fork of the (for now) unmaintained Danga::Socket 1.61.
-# Unused features will be removed, and updates will be made to take
-# advantage of newer kernels.
+# This is a fork of the unmaintained Danga::Socket (1.61) with
+# significant changes.  See Documentation/technical/ds.txt in our
+# source for details.
 #
-# API changes to diverge from Danga::Socket will happen to better
-# accomodate new features and improve scalability.  Do not expect
-# this to be a stable API like Danga::Socket.
-# Bugs encountered (and likely fixed) are reported to
-# bug-Danga-Socket@rt.cpan.org and visible at:
+# Do not expect this to be a stable API like Danga::Socket,
+# but it will evolve to suite our needs and to take advantage of
+# newer Linux and *BSD features.
+# Bugs encountered were reported to bug-Danga-Socket@rt.cpan.org,
+# fixed in Danga::Socket 1.62 and visible at:
 # https://rt.cpan.org/Public/Dist/Display.html?Name=Danga-Socket
+#
+# fields:
+# sock: underlying socket
+# rbuf: scalarref, usually undef
+# wbuf: arrayref of coderefs or tmpio (autovivified))
+#        (tmpio = [ GLOB, offset, [ length ] ])
 package PublicInbox::DS;
 use strict;
 use bytes;
-use POSIX ();
+use POSIX qw(WNOHANG);
 use IO::Handle qw();
-use Fcntl qw(SEEK_SET :DEFAULT);
+use Fcntl qw(SEEK_SET :DEFAULT O_APPEND);
 use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC);
 use parent qw(Exporter);
 our @EXPORT_OK = qw(now msg_more);
-use warnings;
 use 5.010_001;
-
+use Scalar::Util qw(blessed);
 use PublicInbox::Syscall qw(:epoll);
-
-use fields ('sock',              # underlying socket
-            'wbuf',              # arrayref of coderefs or GLOB refs
-            'wbuf_off',  # offset into first element of wbuf to start writing at
-            );
-
-use Errno  qw(EAGAIN EINVAL EEXIST);
-use Carp   qw(croak confess carp);
-require File::Spec;
-
+use PublicInbox::Tmpfile;
+use Errno qw(EAGAIN EINVAL);
+use Carp qw(confess carp);
+
+my $nextq; # queue for next_tick
+my $wait_pids; # list of [ pid, callback, callback_arg ]
+my $later_queue; # list of callbacks to run at some later interval
+my $EXPMAP; # fd -> idle_time
+our $EXPTIME = 180; # 3 minutes
+my ($later_timer, $reap_armed, $exp_timer);
+my $ToClose; # sockets to close when event loop is done
 our (
      %DescriptorMap,             # fd (num) -> PublicInbox::DS object
      $Epoll,                     # Global epoll fd (or DSKQXS ref)
      $_io,                       # IO::Handle for Epoll
-     @ToClose,                   # sockets to close when event loop is done
 
      $PostLoopCallback,          # subref to call at the end of each loop, if defined (global)
 
      $LoopTimeout,               # timeout of event loop in milliseconds
      $DoneInit,                  # if we've done the one-time module init yet
      @Timers,                    # timers
+     $in_loop,
      );
 
 Reset();
@@ -62,7 +68,9 @@ Reset all state
 =cut
 sub Reset {
     %DescriptorMap = ();
-    @ToClose = ();
+    $in_loop = $wait_pids = $later_queue = $reap_armed = undef;
+    $EXPMAP = {};
+    $nextq = $ToClose = $later_timer = $exp_timer = undef;
     $LoopTimeout = -1;  # no timeout by default
     @Timers = ();
 
@@ -87,26 +95,18 @@ sub SetLoopTimeout {
     return $LoopTimeout = $_[1] + 0;
 }
 
-=head2 C<< CLASS->AddTimer( $seconds, $coderef ) >>
+=head2 C<< PublicInbox::DS::add_timer( $seconds, $coderef, $arg) >>
 
 Add a timer to occur $seconds from now. $seconds may be fractional, but timers
 are not guaranteed to fire at the exact time you ask for.
 
-Returns a timer object which you can call C<< $timer->cancel >> on if you need to.
-
 =cut
-sub AddTimer {
-    my ($class, $secs, $coderef) = @_;
-
-    if (!$secs) {
-        my $timer = bless([0, $coderef], 'PublicInbox::DS::Timer');
-        unshift(@Timers, $timer);
-        return $timer;
-    }
+sub add_timer ($$;$) {
+    my ($secs, $coderef, $arg) = @_;
 
     my $fire_time = now() + $secs;
 
-    my $timer = bless [$fire_time, $coderef], "PublicInbox::DS::Timer";
+    my $timer = [$fire_time, $coderef, $arg];
 
     if (!@Timers || $fire_time >= $Timers[-1][0]) {
         push @Timers, $timer;
@@ -151,7 +151,7 @@ sub _InitPoller
             $cls = "PublicInbox::$_";
             last if eval "require $cls";
         }
-        $cls->import;
+        $cls->import(qw(epoll_ctl epoll_wait));
         $Epoll = $cls->new;
     }
     *EventLoop = *EpollEventLoop;
@@ -173,18 +173,37 @@ sub FirstTimeEventLoop {
 
 sub now () { clock_gettime(CLOCK_MONOTONIC) }
 
+sub next_tick () {
+    my $q = $nextq or return;
+    $nextq = undef;
+    for (@$q) {
+        # we 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($_)) {
+            $_->event_step;
+        } else {
+            $_->();
+        }
+    }
+}
+
 # runs timers and returns milliseconds for next one, or next event loop
 sub RunTimers {
-    return $LoopTimeout unless @Timers;
+    next_tick();
+
+    return (($nextq || $ToClose) ? 0 : $LoopTimeout) unless @Timers;
 
     my $now = now();
 
     # Run expired timers
     while (@Timers && $Timers[0][0] <= $now) {
         my $to_run = shift(@Timers);
-        $to_run->[1]->($now) if $to_run->[1];
+        $to_run->[1]->($to_run->[2]);
     }
 
+    # timers may enqueue into nextq:
+    return 0 if ($nextq || $ToClose);
+
     return $LoopTimeout unless @Timers;
 
     # convert time to an even number of milliseconds, adding 1
@@ -202,8 +221,59 @@ sub RunTimers {
     return $timeout;
 }
 
+# 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 reap_pids {
+       $reap_armed = undef;
+       my $tmp = $wait_pids or return;
+       $wait_pids = undef;
+       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 ($cb) {
+                       eval { $cb->($arg, $pid) };
+               }
+       }
+       # we may not be done, yet, and could've missed/masked a SIGCHLD:
+       $reap_armed //= requeue(\&reap_pids) if $wait_pids;
+}
+
+# reentrant SIGCHLD handler (since reap_pids is not reentrant)
+sub enqueue_reap () { $reap_armed //= requeue(\&reap_pids) }
+
+sub in_loop () { $in_loop }
+
+# 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;
+
+               # order matters, destroy expiry times, first:
+               delete @$EXPMAP{@$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;
+}
+
 sub EpollEventLoop {
-    while (1) {
+    local $in_loop = 1;
+    do {
         my @events;
         my $i;
         my $timeout = RunTimers();
@@ -217,8 +287,8 @@ sub EpollEventLoop {
             # in that event.
             $DescriptorMap{$events[$i]->[0]}->event_step;
         }
-        return unless PostEventLoop();
-    }
+    } while (PostEventLoop());
+    _run_later();
 }
 
 =head2 C<< CLASS->SetPostLoopCallback( CODEREF ) >>
@@ -239,37 +309,6 @@ sub SetPostLoopCallback {
     $PostLoopCallback = (defined $ref && ref $ref eq 'CODE') ? $ref : undef;
 }
 
-# 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)
-    while (my $sock = shift @ToClose) {
-        my $fd = fileno($sock);
-
-        # close the socket. (not a PublicInbox::DS close)
-        CORE::close($sock);
-
-        # and now we can finally remove the fd from the map.  see
-        # comment above in ->close.
-        delete $DescriptorMap{$fd};
-    }
-
-
-    # by default we keep running, unless a postloop callback (either per-object
-    # or global) cancels it
-    my $keep_running = 1;
-
-    # now we're at the very end, call callback if defined
-    if (defined $PostLoopCallback) {
-        $keep_running &&= $PostLoopCallback->(\%DescriptorMap);
-    }
-
-    return $keep_running;
-}
-
 #####################################################################
 ### PublicInbox::DS-the-object code
 #####################################################################
@@ -288,16 +327,12 @@ This is normally (always?) called from your subclass via:
 =cut
 sub new {
     my ($self, $sock, $ev) = @_;
-    $self = fields::new($self) unless ref $self;
-
     $self->{sock} = $sock;
     my $fd = fileno($sock);
 
-    Carp::cluck("undef sock and/or fd in PublicInbox::DS->new.  sock=" . ($sock || "") . ", fd=" . ($fd || ""))
-        unless $sock && $fd;
-
     _InitPoller();
 
+retry:
     if (epoll_ctl($Epoll, EPOLL_CTL_ADD, $fd, $ev)) {
         if ($! == EINVAL && ($ev & EPOLLEXCLUSIVE)) {
             $ev &= ~EPOLLEXCLUSIVE;
@@ -305,11 +340,10 @@ sub new {
         }
         die "couldn't add epoll watch for $fd: $!\n";
     }
-    Carp::cluck("PublicInbox::DS::new blowing away existing descriptor map for fd=$fd ($DescriptorMap{$fd})")
-        if $DescriptorMap{$fd};
+    confess("DescriptorMap{$fd} defined ($DescriptorMap{$fd})")
+        if defined($DescriptorMap{$fd});
 
     $DescriptorMap{$fd} = $self;
-    return $self;
 }
 
 
@@ -317,6 +351,8 @@ sub new {
 ### 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.
@@ -348,17 +384,19 @@ sub close {
 
     # defer closing the actual socket until the event loop is done
     # processing this round of events.  (otherwise we might reuse fds)
-    push @ToClose, $sock;
+    push @$ToClose, $sock; # autovivifies $ToClose
 
     return 0;
 }
 
 # portable, non-thread-safe sendfile emulation (no pread, yet)
-sub psendfile ($$$) {
-    my ($sock, $fh, $off) = @_;
+sub send_tmpio ($$) {
+    my ($sock, $tmpio) = @_;
 
-    seek($fh, $$off, SEEK_SET) or return;
-    defined(my $to_write = read($fh, my $buf, 16384)) or return;
+    sysseek($tmpio->[0], $tmpio->[1], SEEK_SET) or return;
+    my $n = $tmpio->[2] // 65536;
+    $n = 65536 if $n > 65536;
+    defined(my $to_write = sysread($tmpio->[0], my $buf, $n)) or return;
     my $written = 0;
     while ($to_write > 0) {
         if (defined(my $w = syswrite($sock, $buf, $to_write, $written))) {
@@ -369,36 +407,39 @@ sub psendfile ($$$) {
             last;
         }
     }
-    $$off += $written;
+    $tmpio->[1] += $written; # offset
+    $tmpio->[2] -= $written if defined($tmpio->[2]); # length
     $written;
 }
 
+sub epbit ($$) { # (sock, default)
+       $_[0]->can('stop_SSL') ? PublicInbox::TLS::epollbit() : $_[1];
+}
+
 # returns 1 if done, 0 if incomplete
 sub flush_write ($) {
     my ($self) = @_;
+    my $sock = $self->{sock} or return;
     my $wbuf = $self->{wbuf} or return 1;
-    my $sock = $self->{sock};
 
 next_buf:
     while (my $bref = $wbuf->[0]) {
         if (ref($bref) ne 'CODE') {
-            my $off = delete($self->{wbuf_off}) // 0;
             while ($sock) {
-                my $w = psendfile($sock, $bref, \$off);
+                my $w = send_tmpio($sock, $bref); # bref is tmpio
                 if (defined $w) {
                     if ($w == 0) {
                         shift @$wbuf;
                         goto next_buf;
                     }
                 } elsif ($! == EAGAIN) {
-                    $self->{wbuf_off} = $off;
-                    watch($self, EPOLLOUT|EPOLLONESHOT);
+                    epwait($sock, epbit($sock, EPOLLOUT) | EPOLLONESHOT);
                     return 0;
                 } else {
                     return $self->close;
                 }
             }
-        } else { #($ref eq 'CODE') {
+        } else { #(ref($bref) eq 'CODE') {
             shift @$wbuf;
             my $before = scalar(@$wbuf);
             $bref->($self);
@@ -412,17 +453,25 @@ next_buf:
     1; # all done
 }
 
-sub do_read ($$$$) {
+sub rbuf_idle ($$) {
+    my ($self, $rbuf) = @_;
+    if ($$rbuf eq '') { # who knows how long till we can read again
+        delete $self->{rbuf};
+    } else {
+        $self->{rbuf} = $rbuf;
+    }
+}
+
+sub do_read ($$$;$) {
     my ($self, $rbuf, $len, $off) = @_;
-    my $r = sysread($self->{sock}, $$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 (ref($self) eq 'IO::Socket::SSL') {
-        my $ev = PublicInbox::TLS::epollbit() or return $self->close;
-        watch($self, $ev | EPOLLONESHOT);
-    } elsif ($! == EAGAIN) {
-        watch($self, EPOLLIN | EPOLLONESHOT);
+    if ($! == EAGAIN) {
+        epwait($sock, epbit($sock, EPOLLIN) | EPOLLONESHOT);
+        rbuf_idle($self, $rbuf);
+        0;
     } else {
         $self->close;
     }
@@ -440,19 +489,12 @@ sub drop {
 # PerlIO::mmap or PerlIO::scalar if needed
 sub tmpio ($$$) {
     my ($self, $bref, $off) = @_;
-    my $fh; # open(my $fh, '+>>', undef) doesn't set O_APPEND
-    do {
-        my $fn = File::Spec->tmpdir . '/wbuf-' . rand;
-        if (sysopen($fh, $fn, O_RDWR|O_CREAT|O_EXCL|O_APPEND, 0600)) { # likely
-            unlink($fn) or return drop($self, "unlink($fn) $!");
-        } elsif ($! != EEXIST) { # EMFILE/ENFILE/ENOSPC/ENOMEM
-            return drop($self, "open: $!");
-        }
-    } until (defined $fh);
+    my $fh = tmpfile('wbuf', $self->{sock}, O_APPEND) or
+        return drop($self, "tmpfile $!");
     $fh->autoflush(1);
     my $len = bytes::length($$bref) - $off;
     $fh->write($$bref, $len, $off) or return drop($self, "write ($len): $!");
-    $fh
+    [ $fh, 0 ] # [1] = offset, [2] = length, not set by us
 }
 
 =head2 C<< $obj->write( $data ) >>
@@ -481,9 +523,9 @@ sub write {
         if ($ref eq 'CODE') {
             push @$wbuf, $bref;
         } else {
-            my $last = $wbuf->[-1];
-            if (ref($last) eq 'GLOB') { # append to tmp file buffer
-                $last->print($$bref) or return drop($self, "print: $!");
+            my $tmpio = $wbuf->[-1];
+            if ($tmpio && !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;
                 push @$wbuf, $tmpio;
@@ -499,14 +541,20 @@ sub write {
 
         if (defined $written) {
             return 1 if $written == $to_write;
+            requeue($self); # runs: event_step -> flush_write
         } elsif ($! == EAGAIN) {
+            epwait($sock, epbit($sock, EPOLLOUT) | EPOLLONESHOT);
             $written = 0;
         } else {
             return $self->close;
         }
+
+        # deal with EAGAIN or partial write:
         my $tmpio = tmpio($self, $bref, $written) or return 0;
-        $self->{wbuf} = [ $tmpio ];
-        watch($self, EPOLLOUT|EPOLLONESHOT);
+
+        # wbuf may be an empty array if we're being called inside
+        # ->flush_write via CODE bref:
+        push @{$self->{wbuf}}, $tmpio; # autovivifies
         return 0;
     }
 }
@@ -516,73 +564,115 @@ use constant MSG_MORE => ($^O eq 'linux') ? 0x8000 : 0;
 sub msg_more ($$) {
     my $self = $_[0];
     my $sock = $self->{sock} or return 1;
+    my $wbuf = $self->{wbuf};
 
-    if (MSG_MORE && !$self->{wbuf} && ref($sock) ne 'IO::Socket::SSL') {
+    if (MSG_MORE && (!defined($wbuf) || !scalar(@$wbuf)) &&
+               !$sock->can('stop_SSL')) {
         my $n = send($sock, $_[1], MSG_MORE);
         if (defined $n) {
             my $nlen = bytes::length($_[1]) - $n;
             return 1 if $nlen == 0; # all done!
             # queue up the unwritten substring:
             my $tmpio = tmpio($self, \($_[1]), $n) or return 0;
-            $self->{wbuf} = [ $tmpio ];
-            watch($self, EPOLLOUT|EPOLLONESHOT);
+            push @{$self->{wbuf}}, $tmpio; # autovivifies
+            epwait($sock, EPOLLOUT|EPOLLONESHOT);
             return 0;
         }
     }
-    $self->write(\($_[1]));
+
+    # don't redispatch into NNTPdeflate::write
+    PublicInbox::DS::write($self, \($_[1]));
 }
 
-sub watch ($$) {
-    my ($self, $ev) = @_;
-    my $sock = $self->{sock} or return;
+sub epwait ($$) {
+    my ($sock, $ev) = @_;
     epoll_ctl($Epoll, EPOLL_CTL_MOD, fileno($sock), $ev) and
         confess("EPOLL_CTL_MOD $!");
-    0;
 }
 
-sub watch_in1 ($) { watch($_[0], EPOLLIN | EPOLLONESHOT) }
-
 # return true if complete, false if incomplete (or failure)
 sub accept_tls_step ($) {
     my ($self) = @_;
     my $sock = $self->{sock} or return;
     return 1 if $sock->accept_SSL;
     return $self->close if $! != EAGAIN;
-    if (my $ev = PublicInbox::TLS::epollbit()) {
-        unshift @{$self->{wbuf} ||= []}, \&accept_tls_step;
-        return watch($self, $ev | EPOLLONESHOT);
-    }
-    drop($self, 'BUG? EAGAIN but '.PublicInbox::TLS::err());
+    epwait($sock, PublicInbox::TLS::epollbit() | EPOLLONESHOT);
+    unshift(@{$self->{wbuf}}, \&accept_tls_step); # autovivifies
+    0;
 }
 
+# return true if complete, false if incomplete (or failure)
 sub shutdn_tls_step ($) {
     my ($self) = @_;
     my $sock = $self->{sock} or return;
     return $self->close if $sock->stop_SSL(SSL_fast_shutdown => 1);
     return $self->close if $! != EAGAIN;
-    if (my $ev = PublicInbox::TLS::epollbit()) {
-        unshift @{$self->{wbuf} ||= []}, \&shutdn_tls_step;
-        return watch($self, $ev | EPOLLONESHOT);
-    }
-    drop($self, 'BUG? EAGAIN but '.PublicInbox::TLS::err());
+    epwait($sock, PublicInbox::TLS::epollbit() | 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 inadvertant socket sharing
+# or fork w/o exec, so no inadvertent socket sharing
 sub shutdn ($) {
     my ($self) = @_;
     my $sock = $self->{sock} or return;
-    if (ref($sock) eq 'IO::Socket::SSL') {
+    if ($sock->can('stop_SSL')) {
         shutdn_tls_step($self);
     } else {
        $self->close;
     }
 }
 
-package PublicInbox::DS::Timer;
-# [$abs_float_firetime, $coderef];
-sub cancel {
-    $_[0][1] = undef;
+# must be called with eval, PublicInbox::DS may not be loaded (see t/qspawn.t)
+sub dwaitpid ($$$) {
+       die "Not in EventLoop\n" unless $in_loop;
+       push @$wait_pids, [ @_ ]; # [ $pid, $cb, $arg ]
+
+       # We could've just missed our SIGCHLD, cover it, here:
+       enqueue_reap();
+}
+
+sub _run_later () {
+       my $run = $later_queue or return;
+       $later_timer = $later_queue = undef;
+       $_->() for @$run;
+}
+
+sub later ($) {
+       push @$later_queue, $_[0]; # autovivifies @$later_queue
+       $later_timer //= add_timer(60, \&_run_later);
+}
+
+sub expire_old () {
+       my $now = now();
+       my $exp = $EXPTIME;
+       my $old = $now - $exp;
+       my %new;
+       while (my ($fd, $idle_at) = each %$EXPMAP) {
+               if ($idle_at < $old) {
+                       my $ds_obj = $DescriptorMap{$fd};
+                       $new{$fd} = $idle_at if !$ds_obj->shutdn;
+               } else {
+                       $new{$fd} = $idle_at;
+               }
+       }
+       $EXPMAP = \%new;
+       $exp_timer = scalar(keys %new) ? later(\&expire_old) : undef;
+}
+
+sub update_idle_time {
+       my ($self) = @_;
+       my $sock = $self->{sock} or return;
+       $EXPMAP->{fileno($sock)} = now();
+       $exp_timer //= later(\&expire_old);
+}
+
+sub not_idle_long {
+       my ($self, $now) = @_;
+       my $sock = $self->{sock} or return;
+       my $idle_at = $EXPMAP->{fileno($sock)} or return;
+       ($idle_at + $EXPTIME) > $now;
 }
 
 1;