lib/PublicInbox/DSKQXS.pm

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