lib/PublicInbox/DS.pm

   1 # This library is free software; you can redistribute it and/or modify
   2 # it under the same terms as Perl itself.
   3 #
   4 # This license differs from the rest of public-inbox
   5 #
   6 # This is a fork of the (for now) unmaintained Danga::Socket 1.61.
   7 # Unused features will be removed, and updates will be made to take
   8 # advantage of newer kernels.
   9 #
  10 # API changes to diverge from Danga::Socket will happen to better
  11 # accomodate new features and improve scalability.  Do not expect
  12 # this to be a stable API like Danga::Socket.
  13 # Bugs encountered (and likely fixed) are reported to
  14 # bug-Danga-Socket@rt.cpan.org and visible at:
  15 # https://rt.cpan.org/Public/Dist/Display.html?Name=Danga-Socket
  16 package PublicInbox::DS;
  17 use strict;
  18 use bytes;
  19 use POSIX ();
  20 use IO::Handle qw();
  21 use Fcntl qw(FD_CLOEXEC F_SETFD F_GETFD);
  22 use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC);
  23 use parent qw(Exporter);
  24 our @EXPORT_OK = qw(now);
  25 use warnings;
  26
  27 use PublicInbox::Syscall qw(:epoll);
  28
  29 use fields ('sock',              # underlying socket
  30             'wbuf',              # arrayref of scalars, scalarrefs, or coderefs to write
  31             'wbuf_off',  # offset into first element of wbuf to start writing at
  32             'event_watch',       # bitmask of events the client is interested in (POLLIN,OUT,etc.)
  33             );
  34
  35 use Errno  qw(EAGAIN EINVAL);
  36 use Carp   qw(croak confess);
  37
  38 use constant POLLIN        => 1;
  39 use constant POLLOUT       => 4;
  40 use constant POLLERR       => 8;
  41 use constant POLLHUP       => 16;
  42 use constant POLLNVAL      => 32;
  43
  44 our $HAVE_KQUEUE = eval { require IO::KQueue; 1 };
  45
  46 our (
  47      $HaveEpoll,                 # Flag -- is epoll available?  initially undefined.
  48      $HaveKQueue,
  49      %DescriptorMap,             # fd (num) -> PublicInbox::DS object
  50      $Epoll,                     # Global epoll fd (for epoll mode only)
  51      $KQueue,                    # Global kqueue fd ref (for kqueue mode only)
  52      $_io,                       # IO::Handle for Epoll
  53      @ToClose,                   # sockets to close when event loop is done
  54
  55      $PostLoopCallback,          # subref to call at the end of each loop, if defined (global)
  56
  57      $LoopTimeout,               # timeout of event loop in milliseconds
  58      $DoneInit,                  # if we've done the one-time module init yet
  59      @Timers,                    # timers
  60      );
  61
  62 # this may be set to zero with old kernels
  63 our $EPOLLEXCLUSIVE = EPOLLEXCLUSIVE;
  64 Reset();
  65
  66 #####################################################################
  67 ### C L A S S   M E T H O D S
  68 #####################################################################
  69
  70 =head2 C<< CLASS->Reset() >>
  71
  72 Reset all state
  73
  74 =cut
  75 sub Reset {
  76     %DescriptorMap = ();
  77     @ToClose = ();
  78     $LoopTimeout = -1;  # no timeout by default
  79     @Timers = ();
  80
  81     $PostLoopCallback = undef;
  82     $DoneInit = 0;
  83
  84     # NOTE kqueue is close-on-fork, and we don't account for it, yet
  85     # OTOH, we (public-inbox) don't need this sub outside of tests...
  86     POSIX::close($$KQueue) if !$_io && $KQueue && $$KQueue >= 0;
  87     $KQueue = undef;
  88
  89     $_io = undef; # close $Epoll
  90     $Epoll = undef;
  91
  92     *EventLoop = *FirstTimeEventLoop;
  93 }
  94
  95 =head2 C<< CLASS->SetLoopTimeout( $timeout ) >>
  96
  97 Set the loop timeout for the event loop to some value in milliseconds.
  98
  99 A timeout of 0 (zero) means poll forever. A timeout of -1 means poll and return
 100 immediately.
 101
 102 =cut
 103 sub SetLoopTimeout {
 104     return $LoopTimeout = $_[1] + 0;
 105 }
 106
 107 =head2 C<< CLASS->AddTimer( $seconds, $coderef ) >>
 108
 109 Add a timer to occur $seconds from now. $seconds may be fractional, but timers
 110 are not guaranteed to fire at the exact time you ask for.
 111
 112 Returns a timer object which you can call C<< $timer->cancel >> on if you need to.
 113
 114 =cut
 115 sub AddTimer {
 116     my $class = shift;
 117     my ($secs, $coderef) = @_;
 118
 119     my $fire_time = now() + $secs;
 120
 121     my $timer = bless [$fire_time, $coderef], "PublicInbox::DS::Timer";
 122
 123     if (!@Timers || $fire_time >= $Timers[-1][0]) {
 124         push @Timers, $timer;
 125         return $timer;
 126     }
 127
 128     # Now, where do we insert?  (NOTE: this appears slow, algorithm-wise,
 129     # but it was compared against calendar queues, heaps, naive push/sort,
 130     # and a bunch of other versions, and found to be fastest with a large
 131     # variety of datasets.)
 132     for (my $i = 0; $i < @Timers; $i++) {
 133         if ($Timers[$i][0] > $fire_time) {
 134             splice(@Timers, $i, 0, $timer);
 135             return $timer;
 136         }
 137     }
 138
 139     die "Shouldn't get here.";
 140 }
 141
 142 # keeping this around in case we support other FD types for now,
 143 # epoll_create1(EPOLL_CLOEXEC) requires Linux 2.6.27+...
 144 sub set_cloexec ($) {
 145     my ($fd) = @_;
 146
 147     $_io = IO::Handle->new_from_fd($fd, 'r+') or return;
 148     defined(my $fl = fcntl($_io, F_GETFD, 0)) or return;
 149     fcntl($_io, F_SETFD, $fl | FD_CLOEXEC);
 150 }
 151
 152 sub _InitPoller
 153 {
 154     return if $DoneInit;
 155     $DoneInit = 1;
 156
 157     if ($HAVE_KQUEUE) {
 158         $KQueue = IO::KQueue->new();
 159         $HaveKQueue = defined $KQueue;
 160         if ($HaveKQueue) {
 161             *EventLoop = *KQueueEventLoop;
 162         }
 163     }
 164     elsif (PublicInbox::Syscall::epoll_defined()) {
 165         $Epoll = eval { epoll_create(1024); };
 166         $HaveEpoll = defined $Epoll && $Epoll >= 0;
 167         if ($HaveEpoll) {
 168             set_cloexec($Epoll);
 169             *EventLoop = *EpollEventLoop;
 170         }
 171     }
 172
 173     if (!$HaveEpoll && !$HaveKQueue) {
 174         require IO::Poll;
 175         *EventLoop = *PollEventLoop;
 176     }
 177 }
 178
 179 =head2 C<< CLASS->EventLoop() >>
 180
 181 Start processing IO events. In most daemon programs this never exits. See
 182 C<PostLoopCallback> below for how to exit the loop.
 183
 184 =cut
 185 sub FirstTimeEventLoop {
 186     my $class = shift;
 187
 188     _InitPoller();
 189
 190     if ($HaveEpoll) {
 191         EpollEventLoop($class);
 192     } elsif ($HaveKQueue) {
 193         KQueueEventLoop($class);
 194     } else {
 195         PollEventLoop($class);
 196     }
 197 }
 198
 199 sub now () { clock_gettime(CLOCK_MONOTONIC) }
 200
 201 # runs timers and returns milliseconds for next one, or next event loop
 202 sub RunTimers {
 203     return $LoopTimeout unless @Timers;
 204
 205     my $now = now();
 206
 207     # Run expired timers
 208     while (@Timers && $Timers[0][0] <= $now) {
 209         my $to_run = shift(@Timers);
 210         $to_run->[1]->($now) if $to_run->[1];
 211     }
 212
 213     return $LoopTimeout unless @Timers;
 214
 215     # convert time to an even number of milliseconds, adding 1
 216     # extra, otherwise floating point fun can occur and we'll
 217     # call RunTimers like 20-30 times, each returning a timeout
 218     # of 0.0000212 seconds
 219     my $timeout = int(($Timers[0][0] - $now) * 1000) + 1;
 220
 221     # -1 is an infinite timeout, so prefer a real timeout
 222     return $timeout     if $LoopTimeout == -1;
 223
 224     # otherwise pick the lower of our regular timeout and time until
 225     # the next timer
 226     return $LoopTimeout if $LoopTimeout < $timeout;
 227     return $timeout;
 228 }
 229
 230 ### The epoll-based event loop. Gets installed as EventLoop if IO::Epoll loads
 231 ### okay.
 232 sub EpollEventLoop {
 233     my $class = shift;
 234
 235     while (1) {
 236         my @events;
 237         my $i;
 238         my $timeout = RunTimers();
 239
 240         # get up to 1000 events
 241         my $evcount = epoll_wait($Epoll, 1000, $timeout, \@events);
 242         for ($i=0; $i<$evcount; $i++) {
 243             # it's possible epoll_wait returned many events, including some at the end
 244             # that ones in the front triggered unregister-interest actions.  if we
 245             # can't find the %sock entry, it's because we're no longer interested
 246             # in that event.
 247             $DescriptorMap{$events[$i]->[0]}->event_step;
 248         }
 249         return unless PostEventLoop();
 250     }
 251     exit 0;
 252 }
 253
 254 ### The fallback IO::Poll-based event loop. Gets installed as EventLoop if
 255 ### IO::Epoll fails to load.
 256 sub PollEventLoop {
 257     my $class = shift;
 258
 259     my PublicInbox::DS $pob;
 260
 261     while (1) {
 262         my $timeout = RunTimers();
 263
 264         # the following sets up @poll as a series of ($poll,$event_mask)
 265         # items, then uses IO::Poll::_poll, implemented in XS, which
 266         # modifies the array in place with the even elements being
 267         # replaced with the event masks that occured.
 268         my @poll;
 269         while ( my ($fd, $sock) = each %DescriptorMap ) {
 270             push @poll, $fd, $sock->{event_watch};
 271         }
 272
 273         # if nothing to poll, either end immediately (if no timeout)
 274         # or just keep calling the callback
 275         unless (@poll) {
 276             select undef, undef, undef, ($timeout / 1000);
 277             return unless PostEventLoop();
 278             next;
 279         }
 280
 281         my $count = IO::Poll::_poll($timeout, @poll);
 282         unless ($count >= 0) {
 283             return unless PostEventLoop();
 284             next;
 285         }
 286
 287         # Fetch handles with read events
 288         while (@poll) {
 289             my ($fd, $state) = splice(@poll, 0, 2);
 290             $DescriptorMap{$fd}->event_step if $state;
 291         }
 292
 293         return unless PostEventLoop();
 294     }
 295
 296     exit 0;
 297 }
 298
 299 ### The kqueue-based event loop. Gets installed as EventLoop if IO::KQueue works
 300 ### okay.
 301 sub KQueueEventLoop {
 302     my $class = shift;
 303
 304     while (1) {
 305         my $timeout = RunTimers();
 306         my @ret = eval { $KQueue->kevent($timeout) };
 307         if (my $err = $@) {
 308             # workaround https://rt.cpan.org/Ticket/Display.html?id=116615
 309             if ($err =~ /Interrupted system call/) {
 310                 @ret = ();
 311             } else {
 312                 die $err;
 313             }
 314         }
 315
 316         foreach my $kev (@ret) {
 317             $DescriptorMap{$kev->[0]}->event_step;
 318         }
 319         return unless PostEventLoop();
 320     }
 321
 322     exit(0);
 323 }
 324
 325 =head2 C<< CLASS->SetPostLoopCallback( CODEREF ) >>
 326
 327 Sets post loop callback function.  Pass a subref and it will be
 328 called every time the event loop finishes.
 329
 330 Return 1 (or any true value) from the sub to make the loop continue, 0 or false
 331 and it will exit.
 332
 333 The callback function will be passed two parameters: \%DescriptorMap
 334
 335 =cut
 336 sub SetPostLoopCallback {
 337     my ($class, $ref) = @_;
 338
 339     # global callback
 340     $PostLoopCallback = (defined $ref && ref $ref eq 'CODE') ? $ref : undef;
 341 }
 342
 343 # Internal function: run the post-event callback, send read events
 344 # for pushed-back data, and close pending connections.  returns 1
 345 # if event loop should continue, or 0 to shut it all down.
 346 sub PostEventLoop {
 347     # now we can close sockets that wanted to close during our event processing.
 348     # (we didn't want to close them during the loop, as we didn't want fd numbers
 349     #  being reused and confused during the event loop)
 350     while (my $sock = shift @ToClose) {
 351         my $fd = fileno($sock);
 352
 353         # close the socket.  (not a PublicInbox::DS close)
 354         $sock->close;
 355
 356         # and now we can finally remove the fd from the map.  see
 357         # comment above in ->close.
 358         delete $DescriptorMap{$fd};
 359     }
 360
 361
 362     # by default we keep running, unless a postloop callback (either per-object
 363     # or global) cancels it
 364     my $keep_running = 1;
 365
 366     # now we're at the very end, call callback if defined
 367     if (defined $PostLoopCallback) {
 368         $keep_running &&= $PostLoopCallback->(\%DescriptorMap);
 369     }
 370
 371     return $keep_running;
 372 }
 373
 374 #####################################################################
 375 ### PublicInbox::DS-the-object code
 376 #####################################################################
 377
 378 =head2 OBJECT METHODS
 379
 380 =head2 C<< CLASS->new( $socket ) >>
 381
 382 Create a new PublicInbox::DS subclass object for the given I<socket> which will
 383 react to events on it during the C<EventLoop>.
 384
 385 This is normally (always?) called from your subclass via:
 386
 387   $class->SUPER::new($socket);
 388
 389 =cut
 390 sub new {
 391     my ($self, $sock, $exclusive) = @_;
 392     $self = fields::new($self) unless ref $self;
 393
 394     $self->{sock} = $sock;
 395     my $fd = fileno($sock);
 396
 397     Carp::cluck("undef sock and/or fd in PublicInbox::DS->new.  sock=" . ($sock || "") . ", fd=" . ($fd || ""))
 398         unless $sock && $fd;
 399
 400     $self->{wbuf} = [];
 401     $self->{wbuf_off} = 0;
 402
 403     my $ev = $self->{event_watch} = POLLERR|POLLHUP|POLLNVAL;
 404
 405     _InitPoller();
 406
 407     if ($HaveEpoll) {
 408         if ($exclusive) {
 409             $ev = $self->{event_watch} = EPOLLIN|EPOLLERR|EPOLLHUP|$EPOLLEXCLUSIVE;
 410         }
 411 retry:
 412         if (epoll_ctl($Epoll, EPOLL_CTL_ADD, $fd, $ev)) {
 413             if ($! == EINVAL && ($ev & $EPOLLEXCLUSIVE)) {
 414                 $EPOLLEXCLUSIVE = 0; # old kernel
 415                 $ev = $self->{event_watch} = EPOLLIN|EPOLLERR|EPOLLHUP;
 416                 goto retry;
 417             }
 418             die "couldn't add epoll watch for $fd: $!\n";
 419         }
 420     }
 421     elsif ($HaveKQueue) {
 422         # Add them to the queue but disabled for now
 423         $KQueue->EV_SET($fd, IO::KQueue::EVFILT_READ(),
 424                         IO::KQueue::EV_ADD() | IO::KQueue::EV_DISABLE());
 425         $KQueue->EV_SET($fd, IO::KQueue::EVFILT_WRITE(),
 426                         IO::KQueue::EV_ADD() | IO::KQueue::EV_DISABLE());
 427     }
 428
 429     Carp::cluck("PublicInbox::DS::new blowing away existing descriptor map for fd=$fd ($DescriptorMap{$fd})")
 430         if $DescriptorMap{$fd};
 431
 432     $DescriptorMap{$fd} = $self;
 433     return $self;
 434 }
 435
 436
 437 #####################################################################
 438 ### I N S T A N C E   M E T H O D S
 439 #####################################################################
 440
 441 =head2 C<< $obj->close >>
 442
 443 Close the socket.
 444
 445 =cut
 446 sub close {
 447     my ($self) = @_;
 448     my $sock = delete $self->{sock} or return;
 449
 450     # we need to flush our write buffer, as there may
 451     # be self-referential closures (sub { $client->close })
 452     # preventing the object from being destroyed
 453     @{$self->{wbuf}} = ();
 454
 455     # if we're using epoll, we have to remove this from our epoll fd so we stop getting
 456     # notifications about it
 457     if ($HaveEpoll) {
 458         my $fd = fileno($sock);
 459         epoll_ctl($Epoll, EPOLL_CTL_DEL, $fd, $self->{event_watch}) and
 460             confess("EPOLL_CTL_DEL: $!");
 461     }
 462
 463     # we explicitly don't delete from DescriptorMap here until we
 464     # actually close the socket, as we might be in the middle of
 465     # processing an epoll_wait/etc that returned hundreds of fds, one
 466     # of which is not yet processed and is what we're closing.  if we
 467     # keep it in DescriptorMap, then the event harnesses can just
 468     # looked at $pob->{sock} == undef and ignore it.  but if it's an
 469     # un-accounted for fd, then it (understandably) freak out a bit
 470     # and emit warnings, thinking their state got off.
 471
 472     # defer closing the actual socket until the event loop is done
 473     # processing this round of events.  (otherwise we might reuse fds)
 474     push @ToClose, $sock;
 475
 476     return 0;
 477 }
 478
 479 =head2 C<< $obj->write( $data ) >>
 480
 481 Write the specified data to the underlying handle.  I<data> may be scalar,
 482 scalar ref, code ref (to run when there), or undef just to kick-start.
 483 Returns 1 if writes all went through, or 0 if there are writes in queue. If
 484 it returns 1, caller should stop waiting for 'writable' events)
 485
 486 =cut
 487 sub write {
 488     my PublicInbox::DS $self;
 489     my $data;
 490     ($self, $data) = @_;
 491
 492     # nobody should be writing to closed sockets, but caller code can
 493     # do two writes within an event, have the first fail and
 494     # disconnect the other side (whose destructor then closes the
 495     # calling object, but it's still in a method), and then the
 496     # now-dead object does its second write.  that is this case.  we
 497     # just lie and say it worked.  it'll be dead soon and won't be
 498     # hurt by this lie.
 499     return 1 unless $self->{sock};
 500
 501     my $bref;
 502
 503     # just queue data if there's already a wait
 504     my $need_queue;
 505     my $wbuf = $self->{wbuf};
 506
 507     if (defined $data) {
 508         $bref = ref $data ? $data : \$data;
 509         if (scalar @$wbuf) {
 510             push @$wbuf, $bref;
 511             return 0;
 512         }
 513
 514         # this flag says we're bypassing the queue system, knowing we're the
 515         # only outstanding write, and hoping we don't ever need to use it.
 516         # if so later, though, we'll need to queue
 517         $need_queue = 1;
 518     }
 519
 520   WRITE:
 521     while (1) {
 522         return 1 unless $bref ||= $wbuf->[0];
 523
 524         my $len;
 525         eval {
 526             $len = length($$bref); # this will die if $bref is a code ref, caught below
 527         };
 528         if ($@) {
 529             if (UNIVERSAL::isa($bref, "CODE")) {
 530                 unless ($need_queue) {
 531                     shift @$wbuf;
 532                 }
 533                 $bref->();
 534
 535                 # code refs are just run and never get reenqueued
 536                 # (they're one-shot), so turn off the flag indicating the
 537                 # outstanding data needs queueing.
 538                 $need_queue = 0;
 539
 540                 undef $bref;
 541                 next WRITE;
 542             }
 543             die "Write error: $@ <$bref>";
 544         }
 545
 546         my $to_write = $len - $self->{wbuf_off};
 547         my $written = syswrite($self->{sock}, $$bref, $to_write,
 548                                $self->{wbuf_off});
 549
 550         if (! defined $written) {
 551             if ($! == EAGAIN) {
 552                 # since connection has stuff to write, it should now be
 553                 # interested in pending writes:
 554                 if ($need_queue) {
 555                     push @$wbuf, $bref;
 556                 }
 557                 $self->watch_write(1);
 558                 return 0;
 559             }
 560
 561             return $self->close;
 562         } elsif ($written != $to_write) {
 563             if ($need_queue) {
 564                 push @$wbuf, $bref;
 565             }
 566             # since connection has stuff to write, it should now be
 567             # interested in pending writes:
 568             $self->{wbuf_off} += $written;
 569             $self->on_incomplete_write;
 570             return 0;
 571         } elsif ($written == $to_write) {
 572             $self->{wbuf_off} = 0;
 573             $self->watch_write(0);
 574
 575             # this was our only write, so we can return immediately
 576             # since we avoided incrementing the buffer size or
 577             # putting it in the buffer.  we also know there
 578             # can't be anything else to write.
 579             return 1 if $need_queue;
 580
 581             shift @$wbuf;
 582             undef $bref;
 583             next WRITE;
 584         }
 585     }
 586 }
 587
 588 sub on_incomplete_write {
 589     my PublicInbox::DS $self = shift;
 590     $self->watch_write(1);
 591 }
 592
 593 =head2 C<< $obj->watch_read( $boolean ) >>
 594
 595 Turn 'readable' event notification on or off.
 596
 597 =cut
 598 sub watch_read {
 599     my PublicInbox::DS $self = shift;
 600     my $sock = $self->{sock} or return;
 601
 602     my $val = shift;
 603     my $event = $self->{event_watch};
 604
 605     $event &= ~POLLIN if ! $val;
 606     $event |=  POLLIN if   $val;
 607
 608     my $fd = fileno($sock);
 609     # If it changed, set it
 610     if ($event != $self->{event_watch}) {
 611         if ($HaveKQueue) {
 612             $KQueue->EV_SET($fd, IO::KQueue::EVFILT_READ(),
 613                             $val ? IO::KQueue::EV_ENABLE() : IO::KQueue::EV_DISABLE());
 614         }
 615         elsif ($HaveEpoll) {
 616             epoll_ctl($Epoll, EPOLL_CTL_MOD, $fd, $event) and
 617                 confess("EPOLL_CTL_MOD: $!");
 618         }
 619         $self->{event_watch} = $event;
 620     }
 621 }
 622
 623 =head2 C<< $obj->watch_write( $boolean ) >>
 624
 625 Turn 'writable' event notification on or off.
 626
 627 =cut
 628 sub watch_write {
 629     my PublicInbox::DS $self = shift;
 630     my $sock = $self->{sock} or return;
 631
 632     my $val = shift;
 633     my $event = $self->{event_watch};
 634
 635     $event &= ~POLLOUT if ! $val;
 636     $event |=  POLLOUT if   $val;
 637     my $fd = fileno($sock);
 638
 639     # If it changed, set it
 640     if ($event != $self->{event_watch}) {
 641         if ($HaveKQueue) {
 642             $KQueue->EV_SET($fd, IO::KQueue::EVFILT_WRITE(),
 643                             $val ? IO::KQueue::EV_ENABLE() : IO::KQueue::EV_DISABLE());
 644         }
 645         elsif ($HaveEpoll) {
 646             epoll_ctl($Epoll, EPOLL_CTL_MOD, $fd, $event) and
 647                     confess "EPOLL_CTL_MOD: $!";
 648         }
 649         $self->{event_watch} = $event;
 650     }
 651 }
 652
 653 package PublicInbox::DS::Timer;
 654 # [$abs_float_firetime, $coderef];
 655 sub cancel {
 656     $_[0][1] = undef;
 657 }
 658
 659 1;
 660
 661 =head1 AUTHORS (Danga::Socket)
 662
 663 Brad Fitzpatrick <brad@danga.com> - author
 664
 665 Michael Granger <ged@danga.com> - docs, testing
 666
 667 Mark Smith <junior@danga.com> - contributor, heavy user, testing
 668
 669 Matt Sergeant <matt@sergeant.org> - kqueue support, docs, timers, other bits