lib/PublicInbox/DSKQXS.pm

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