lib/PublicInbox/DSKQXS.pm

   1 # Copyright (C) 2019-2020 all contributors <meta@public-inbox.org>
   2 # Licensed the same as Danga::Socket (and Perl5)
   3 # License: GPL-1.0+ or Artistic-1.0-Perl
   4 #  <https://www.gnu.org/licenses/gpl-1.0.txt>
   5 #  <https://dev.perl.org/licenses/artistic.html>
   6 #
   7 # kqueue support via IO::KQueue XS module.  This makes kqueue look
   8 # like epoll to simplify the code in DS.pm.  This is NOT meant to be
   9 # an all encompassing emulation of epoll via IO::KQueue, but just to
  10 # support cases public-inbox-nntpd/httpd care about.
  11 #
  12 # It also implements signalfd(2) emulation via "tie".
  13 #
  14 # A pure-Perl version using syscall() is planned.
  15 package PublicInbox::DSKQXS;
  16 use strict;
  17 use warnings;
  18 use parent qw(Exporter);
  19 use Symbol qw(gensym);
  20 use IO::KQueue;
  21 use Errno qw(EAGAIN);
  22 use PublicInbox::Syscall qw(EPOLLONESHOT EPOLLIN EPOLLOUT EPOLLET
  23         EPOLL_CTL_ADD EPOLL_CTL_MOD EPOLL_CTL_DEL SFD_NONBLOCK);
  24 our @EXPORT_OK = qw(epoll_ctl epoll_wait);
  25
  26 sub EV_DISPATCH () { 0x0080 }
  27
  28 # map EPOLL* bits to kqueue EV_* flags for EV_SET
  29 sub kq_flag ($$) {
  30         my ($bit, $ev) = @_;
  31         if ($ev & $bit) {
  32                 my $fl = EV_ENABLE;
  33                 $fl |= EV_CLEAR if $fl & EPOLLET;
  34
  35                 # EV_DISPATCH matches EPOLLONESHOT semantics more closely
  36                 # than EV_ONESHOT, in that EV_ADD is not required to
  37                 # re-enable a disabled watch.
  38                 ($ev & EPOLLONESHOT) ? ($fl | EV_DISPATCH) : $fl;
  39         } else {
  40                 EV_DISABLE;
  41         }
  42 }
  43
  44 sub new {
  45         my ($class) = @_;
  46         bless { kq => IO::KQueue->new, owner_pid => $$ }, $class;
  47 }
  48
  49 # returns a new instance which behaves like signalfd on Linux.
  50 # It's wasteful in that it uses another FD, but it simplifies
  51 # our epoll-oriented code.
  52 sub signalfd {
  53         my ($class, $signo, $flags) = @_;
  54         my $sym = gensym;
  55         tie *$sym, $class, $signo, $flags; # calls TIEHANDLE
  56         $sym
  57 }
  58
  59 sub TIEHANDLE { # similar to signalfd()
  60         my ($class, $signo, $flags) = @_;
  61         my $self = $class->new;
  62         $self->{timeout} = ($flags & SFD_NONBLOCK) ? 0 : -1;
  63         my $kq = $self->{kq};
  64         $kq->EV_SET($_, EVFILT_SIGNAL, EV_ADD) for @$signo;
  65         $self;
  66 }
  67
  68 sub READ { # called by sysread() for signalfd compatibility
  69         my ($self, undef, $len, $off) = @_; # $_[1] = buf
  70         die "bad args for signalfd read" if ($len % 128) // defined($off);
  71         my $timeout = $self->{timeout};
  72         my $sigbuf = $self->{sigbuf} //= [];
  73         my $nr = $len / 128;
  74         my $r = 0;
  75         $_[1] = '';
  76         do {
  77                 while ($nr--) {
  78                         my $signo = shift(@$sigbuf) or last;
  79                         # caller only cares about signalfd_siginfo.ssi_signo:
  80                         $_[1] .= pack('L', $signo) . ("\0" x 124);
  81                         $r += 128;
  82                 }
  83                 return $r if $r;
  84                 my @events = eval { $self->{kq}->kevent($timeout) };
  85                 # workaround https://rt.cpan.org/Ticket/Display.html?id=116615
  86                 if ($@) {
  87                         next if $@ =~ /Interrupted system call/;
  88                         die;
  89                 }
  90                 if (!scalar(@events) && $timeout == 0) {
  91                         $! = EAGAIN;
  92                         return;
  93                 }
  94
  95                 # Grab the kevent.ident (signal number).  The kevent.data
  96                 # field shows coalesced signals, and maybe we'll use it
  97                 # in the future...
  98                 @$sigbuf = map { $_->[0] } @events;
  99         } while (1);
 100 }
 101
 102 # for fileno() calls in PublicInbox::DS
 103 sub FILENO { ${$_[0]->{kq}} }
 104
 105 sub epoll_ctl {
 106         my ($self, $op, $fd, $ev) = @_;
 107         my $kq = $self->{kq};
 108         if ($op == EPOLL_CTL_MOD) {
 109                 $kq->EV_SET($fd, EVFILT_READ, kq_flag(EPOLLIN, $ev));
 110                 $kq->EV_SET($fd, EVFILT_WRITE, kq_flag(EPOLLOUT, $ev));
 111         } elsif ($op == EPOLL_CTL_DEL) {
 112                 $kq->EV_SET($fd, EVFILT_READ, EV_DISABLE);
 113                 $kq->EV_SET($fd, EVFILT_WRITE, EV_DISABLE);
 114         } else { # EPOLL_CTL_ADD
 115                 $kq->EV_SET($fd, EVFILT_READ, EV_ADD|kq_flag(EPOLLIN, $ev));
 116
 117                 # we call this blindly for read-only FDs such as tied
 118                 # DSKQXS (signalfd emulation) and Listeners
 119                 eval {
 120                         $kq->EV_SET($fd, EVFILT_WRITE, EV_ADD |
 121                                                         kq_flag(EPOLLOUT, $ev));
 122                 };
 123         }
 124         0;
 125 }
 126
 127 sub epoll_wait {
 128         my ($self, $maxevents, $timeout_msec, $events) = @_;
 129         @$events = eval { $self->{kq}->kevent($timeout_msec) };
 130         if (my $err = $@) {
 131                 # workaround https://rt.cpan.org/Ticket/Display.html?id=116615
 132                 if ($err =~ /Interrupted system call/) {
 133                         @$events = ();
 134                 } else {
 135                         die $err;
 136                 }
 137         }
 138         # caller only cares for $events[$i]->[0]
 139         scalar(@$events);
 140 }
 141
 142 # kqueue is close-on-fork (not exec), so we must not close it
 143 # in forked processes:
 144 sub DESTROY {
 145         my ($self) = @_;
 146         my $kq = delete $self->{kq} or return;
 147         if (delete($self->{owner_pid}) == $$) {
 148                 POSIX::close($$kq);
 149         }
 150 }
 151
 152 1;