This allows us to cut down on imports and reduce code.
This also makes it easier (in the next commit) to provide an option
to disable epoll/kqueue when saving an FD is valued over scalability.
lib/PublicInbox/Emergency.pm
lib/PublicInbox/Eml.pm
lib/PublicInbox/EmlContentFoo.pm
+lib/PublicInbox/Epoll.pm
lib/PublicInbox/ExtMsg.pm
lib/PublicInbox/ExtSearch.pm
lib/PublicInbox/ExtSearchIdx.pm
use Fcntl qw(SEEK_SET :DEFAULT O_APPEND);
use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC);
use Scalar::Util qw(blessed);
-use PublicInbox::Syscall qw(:epoll %SIGNUM);
+use PublicInbox::Syscall qw(%SIGNUM
+ EPOLLIN EPOLLOUT EPOLLONESHOT EPOLLEXCLUSIVE);
use PublicInbox::Tmpfile;
use Errno qw(EAGAIN EINVAL ECHILD EINTR);
use Carp qw(carp croak);
my $ToClose; # sockets to close when event loop is done
our (
%DescriptorMap, # fd (num) -> PublicInbox::DS object
- $Epoll, # Global epoll fd (or DSKQXS ref)
- $ep_io, # IO::Handle for Epoll
+ $Epoll, # global Epoll, DSPoll, or DSKQXS ref
@post_loop_do, # subref + args to call at the end of each loop
my @q = delete @Stack{keys %Stack};
for my $q (@q) { @$q = () }
$AWAIT_PIDS = $nextq = $ToClose = undef;
- $ep_io = undef; # closes real $Epoll FD
$Epoll = undef; # may call DSKQXS::DESTROY
} while (@Timers || keys(%Stack) || $nextq || $AWAIT_PIDS ||
$ToClose || keys(%DescriptorMap) ||
# caller sets return value to $Epoll
sub _InitPoller () {
- if (defined $PublicInbox::Syscall::SYS_epoll_create) {
- my $fd = epoll_create();
- die "epoll_create: $!" if $fd < 0;
- open($ep_io, '+<&=', $fd) or return;
- fcntl($ep_io, F_SETFD, FD_CLOEXEC);
- $fd;
- } else {
- my $cls;
- for (qw(DSKQXS DSPoll)) {
- $cls = "PublicInbox::$_";
- last if eval "require $cls";
- }
- $cls->import(qw(epoll_ctl epoll_wait));
- $cls->new;
+ my @try = ($^O eq 'linux' ? 'Epoll' : 'DSKQXS');
+ my $cls;
+ for (@try, 'DSPoll') {
+ $cls = "PublicInbox::$_";
+ last if eval "require $cls";
}
+ $cls->new;
}
sub now () { clock_gettime(CLOCK_MONOTONIC) }
my $timeout = RunTimers();
# get up to 1000 events
- epoll_wait($Epoll, 1000, $timeout, \@events);
+ $Epoll->ep_wait(1000, $timeout, \@events);
for my $fd (@events) {
# it's possible epoll_wait returned many events,
# including some at the end that ones in the front
$Epoll //= _InitPoller();
retry:
- if (epoll_ctl($Epoll, EPOLL_CTL_ADD, $fd, $ev)) {
+ if ($Epoll->ep_add($sock, $ev)) {
if ($! == EINVAL && ($ev & EPOLLEXCLUSIVE)) {
$ev &= ~EPOLLEXCLUSIVE;
goto retry;
# if we're using epoll, we have to remove this from our epoll fd so we stop getting
# notifications about it
- my $fd = fileno($sock);
- epoll_ctl($Epoll, EPOLL_CTL_DEL, $fd, 0) and
- croak("EPOLL_CTL_DEL($self/$sock): $!");
+ $Epoll->ep_del($sock) and croak("EPOLL_CTL_DEL($self/$sock): $!");
# we explicitly don't delete from DescriptorMap here until we
# actually close the socket, as we might be in the middle of
}
sub epwait ($$) {
- my ($sock, $ev) = @_;
- epoll_ctl($Epoll, EPOLL_CTL_MOD, fileno($sock), $ev) and
- croak("EPOLL_CTL_MOD($sock): $!");
+ my ($io, $ev) = @_;
+ $Epoll->ep_mod($io, $ev) and croak("EPOLL_CTL_MOD($io): $!");
}
# return true if complete, false if incomplete (or failure)
# It also implements signalfd(2) emulation via "tie".
package PublicInbox::DSKQXS;
use v5.12;
-use parent qw(Exporter);
use Symbol qw(gensym);
use IO::KQueue;
use Errno qw(EAGAIN);
-use PublicInbox::Syscall qw(EPOLLONESHOT EPOLLIN EPOLLOUT EPOLLET
- EPOLL_CTL_ADD EPOLL_CTL_MOD EPOLL_CTL_DEL);
-our @EXPORT_OK = qw(epoll_ctl epoll_wait);
+use PublicInbox::Syscall qw(EPOLLONESHOT EPOLLIN EPOLLOUT EPOLLET);
sub EV_DISPATCH () { 0x0080 }
# for fileno() calls in PublicInbox::DS
sub FILENO { ${$_[0]->{kq}} }
-sub epoll_ctl {
- my ($self, $op, $fd, $ev) = @_;
- my $kq = $self->{kq};
- if ($op == EPOLL_CTL_MOD) {
- $kq->EV_SET($fd, EVFILT_READ, kq_flag(EPOLLIN, $ev));
- eval { $kq->EV_SET($fd, EVFILT_WRITE, kq_flag(EPOLLOUT, $ev)) };
- } elsif ($op == EPOLL_CTL_DEL) {
- $kq // return; # called in cleanup
- $kq->EV_SET($fd, EVFILT_READ, EV_DISABLE);
- eval { $kq->EV_SET($fd, EVFILT_WRITE, EV_DISABLE) };
- } else { # EPOLL_CTL_ADD
- $kq->EV_SET($fd, EVFILT_READ, EV_ADD|kq_flag(EPOLLIN, $ev));
-
- # we call this blindly for read-only FDs such as tied
- # DSKQXS (signalfd emulation) and Listeners
- eval {
- $kq->EV_SET($fd, EVFILT_WRITE, EV_ADD |
- kq_flag(EPOLLOUT, $ev));
- };
- }
+sub _ep_mod_add ($$$$) {
+ my ($kq, $fd, $ev, $add) = @_;
+ $kq->EV_SET($fd, EVFILT_READ, $add|kq_flag(EPOLLIN, $ev));
+
+ # we call this blindly for read-only FDs such as tied
+ # DSKQXS (signalfd emulation) and Listeners
+ eval { $kq->EV_SET($fd, EVFILT_WRITE, $add|kq_flag(EPOLLOUT, $ev)) };
+ 0;
+}
+
+sub ep_add { _ep_mod_add($_[0]->{kq}, fileno($_[1]), $_[2], EV_ADD) };
+sub ep_mod { _ep_mod_add($_[0]->{kq}, fileno($_[1]), $_[2], 0) };
+
+sub ep_del {
+ my ($self, $io, $ev) = @_;
+ my $kq = $_[0]->{kq} // return; # called in cleanup
+ my $fd = fileno($io);
+ $kq->EV_SET($fd, EVFILT_READ, EV_DISABLE);
+ eval { $kq->EV_SET($fd, EVFILT_WRITE, EV_DISABLE) };
0;
}
-sub epoll_wait {
+sub ep_wait {
my ($self, $maxevents, $timeout_msec, $events) = @_;
@$events = eval { $self->{kq}->kevent($timeout_msec) };
if (my $err = $@) {
-# Copyright (C) 2019-2021 all contributors <meta@public-inbox.org>
+# Copyright (C) all contributors <meta@public-inbox.org>
# Licensed the same as Danga::Socket (and Perl5)
# License: GPL-1.0+ or Artistic-1.0-Perl
# <https://www.gnu.org/licenses/gpl-1.0.txt>
# an all encompassing emulation of epoll via IO::Poll, but just to
# support cases public-inbox-nntpd/httpd care about.
package PublicInbox::DSPoll;
-use strict;
-use warnings;
-use parent qw(Exporter);
+use v5.12;
use IO::Poll;
-use PublicInbox::Syscall qw(EPOLLONESHOT EPOLLIN EPOLLOUT EPOLL_CTL_DEL);
-our @EXPORT_OK = qw(epoll_ctl epoll_wait);
+use PublicInbox::Syscall qw(EPOLLONESHOT EPOLLIN EPOLLOUT);
-sub new { bless {}, $_[0] } # fd => events
+sub new { bless {}, __PACKAGE__ } # fd => events
-sub epoll_ctl {
- my ($self, $op, $fd, $ev) = @_;
-
- # not wasting time on error checking
- if ($op != EPOLL_CTL_DEL) {
- $self->{$fd} = $ev;
- } else {
- delete $self->{$fd};
- }
- 0;
-}
-
-sub epoll_wait {
+sub ep_wait {
my ($self, $maxevents, $timeout_msec, $events) = @_;
my @pset;
while (my ($fd, $events) = each %$self) {
}
}
+sub ep_del { delete($_[0]->{fileno($_[1])}); 0 }
+sub ep_add { $_[0]->{fileno($_[1])} = $_[2]; 0 }
+
+no warnings 'once';
+*ep_mod = \&ep_add;
+
1;
--- /dev/null
+# Copyright (C) all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# OO API for epoll
+package PublicInbox::Epoll;
+use v5.12;
+use PublicInbox::Syscall qw(epoll_create epoll_ctl epoll_wait
+ EPOLL_CTL_ADD EPOLL_CTL_MOD EPOLL_CTL_DEL);
+use Fcntl qw(F_SETFD FD_CLOEXEC);
+use autodie qw(open fcntl);
+
+sub new {
+ open(my $fh, '+<&=', epoll_create());
+ fcntl($fh, F_SETFD, FD_CLOEXEC);
+ bless \$fh, __PACKAGE__;
+}
+
+sub ep_add { epoll_ctl(fileno(${$_[0]}), EPOLL_CTL_ADD, fileno($_[1]), $_[2]) }
+sub ep_mod { epoll_ctl(fileno(${$_[0]}), EPOLL_CTL_MOD, fileno($_[1]), $_[2]) }
+sub ep_del { epoll_ctl(fileno(${$_[0]}), EPOLL_CTL_DEL, fileno($_[1]), 0) }
+sub ep_wait { epoll_wait(fileno(${$_[0]}), @_[1, 2, 3]) }
+
+1;
EPOLL_CTL_ADD EPOLL_CTL_DEL EPOLL_CTL_MOD
EPOLLONESHOT EPOLLEXCLUSIVE
signalfd rename_noreplace %SIGNUM);
-our %EXPORT_TAGS = (epoll => [qw(epoll_ctl epoll_create epoll_wait
- EPOLLIN EPOLLOUT
- EPOLL_CTL_ADD EPOLL_CTL_DEL EPOLL_CTL_MOD
- EPOLLONESHOT EPOLLEXCLUSIVE)],
- );
-
use constant {
EPOLLIN => 1,
EPOLLOUT => 4,
-# Copyright (C) 2019-2021 all contributors <meta@public-inbox.org>
+# Copyright (C) all contributors <meta@public-inbox.org>
# Licensed the same as Danga::Socket (and Perl5)
# License: GPL-1.0+ or Artistic-1.0-Perl
# <https://www.gnu.org/licenses/gpl-1.0.txt>
# <https://dev.perl.org/licenses/artistic.html>
-use strict;
+use v5.12;
use Test::More;
unless (eval { require IO::KQueue }) {
my $m = $^O !~ /bsd/ ? 'DSKQXS is only for *BSD systems'
-# Copyright (C) 2019-2021 all contributors <meta@public-inbox.org>
+# Copyright (C) all contributors <meta@public-inbox.org>
# Licensed the same as Danga::Socket (and Perl5)
# License: GPL-1.0+ or Artistic-1.0-Perl
# <https://www.gnu.org/licenses/gpl-1.0.txt>
# <https://dev.perl.org/licenses/artistic.html>
-use strict;
-use warnings;
+use v5.12;
use Test::More;
-use PublicInbox::Syscall qw(:epoll);
+use PublicInbox::Syscall qw(EPOLLIN EPOLLOUT EPOLLONESHOT);
my $cls = $ENV{TEST_IOPOLLER} // 'PublicInbox::DSPoll';
use_ok $cls;
my $p = $cls->new;
my ($r, $w, $x, $y);
pipe($r, $w) or die;
pipe($x, $y) or die;
-is($p->epoll_ctl(EPOLL_CTL_ADD, fileno($r), EPOLLIN), 0, 'add EPOLLIN');
+is($p->ep_add($r, EPOLLIN), 0, 'add EPOLLIN');
my $events = [];
-$p->epoll_wait(9, 0, $events);
+$p->ep_wait(9, 0, $events);
is_deeply($events, [], 'no events set');
-is($p->epoll_ctl(EPOLL_CTL_ADD, fileno($w), EPOLLOUT|EPOLLONESHOT), 0,
- 'add EPOLLOUT|EPOLLONESHOT');
-$p->epoll_wait(9, -1, $events);
+is($p->ep_add($w, EPOLLOUT|EPOLLONESHOT), 0, 'add EPOLLOUT|EPOLLONESHOT');
+$p->ep_wait(9, -1, $events);
is(scalar(@$events), 1, 'got POLLOUT event');
is($events->[0], fileno($w), '$w ready');
-$p->epoll_wait(9, 0, $events);
+$p->ep_wait(9, 0, $events);
is(scalar(@$events), 0, 'nothing ready after oneshot');
is_deeply($events, [], 'no events set after oneshot');
syswrite($w, '1') == 1 or die;
for my $t (0..1) {
- $p->epoll_wait(9, $t, $events);
+ $p->ep_wait(9, $t, $events);
is($events->[0], fileno($r), "level-trigger POLLIN ready #$t");
is(scalar(@$events), 1, "only event ready #$t");
}
syswrite($y, '1') == 1 or die;
-is($p->epoll_ctl(EPOLL_CTL_ADD, fileno($x), EPOLLIN|EPOLLONESHOT), 0,
- 'EPOLLIN|EPOLLONESHOT add');
-$p->epoll_wait(9, -1, $events);
+is($p->ep_add($x, EPOLLIN|EPOLLONESHOT), 0, 'EPOLLIN|EPOLLONESHOT add');
+$p->ep_wait(9, -1, $events);
is(scalar @$events, 2, 'epoll_wait has 2 ready');
my @fds = sort @$events;
my @exp = sort((fileno($r), fileno($x)));
is_deeply(\@fds, \@exp, 'got both ready FDs');
-is($p->epoll_ctl(EPOLL_CTL_DEL, fileno($r), 0), 0, 'EPOLL_CTL_DEL OK');
-$p->epoll_wait(9, 0, $events);
+is($p->ep_del($r, 0), 0, 'EPOLL_CTL_DEL OK');
+$p->ep_wait(9, 0, $events);
is(scalar @$events, 0, 'nothing ready after EPOLL_CTL_DEL');
done_testing;
#!perl -w
-# Copyright (C) 2020-2021 all contributors <meta@public-inbox.org>
+# Copyright (C) all contributors <meta@public-inbox.org>
# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
-use strict;
-use v5.10.1;
+use v5.12;
use Test::More;
-use PublicInbox::Syscall qw(:epoll);
+use autodie;
+use PublicInbox::Syscall qw(EPOLLOUT);
plan skip_all => 'not Linux' if $^O ne 'linux';
-my $epfd = epoll_create();
-ok($epfd >= 0, 'epoll_create');
-open(my $hnd, '+<&=', $epfd); # for autoclose
-
-pipe(my ($r, $w)) or die "pipe: $!";
-is(epoll_ctl($epfd, EPOLL_CTL_ADD, fileno($w), EPOLLOUT), 0,
- 'epoll_ctl socket EPOLLOUT');
+require_ok 'PublicInbox::Epoll';
+my $ep = PublicInbox::Epoll->new;
+pipe(my $r, my $w);
+is($ep->ep_add($w, EPOLLOUT), 0, 'epoll_ctl pipe EPOLLOUT');
my @events;
-epoll_wait($epfd, 100, 10000, \@events);
+$ep->ep_wait(100, 10000, \@events);
is(scalar(@events), 1, 'got one event');
is($events[0], fileno($w), 'got expected FD');
close $w;
-epoll_wait($epfd, 100, 0, \@events);
+$ep->ep_wait(100, 0, \@events);
is(scalar(@events), 0, 'epoll_wait timeout');
done_testing;
projects / thirdparty / public-inbox.git / commitdiff
