* [PATCH 1/7] tests: map CLOFORK->FD_CLOEXEC temporarily for `tail -f'
2023-09-11 9:41 [PATCH 0/7] system-related updates and cleanups Eric Wong
@ 2023-09-11 9:41 ` Eric Wong
2023-09-11 9:41 ` [PATCH 2/7] daemon: depend on DS event_loop in master process, too Eric Wong
` (5 subsequent siblings)
6 siblings, 0 replies; 10+ messages in thread
From: Eric Wong @ 2023-09-11 9:41 UTC (permalink / raw)
To: meta
This fixes `TAIL="tail -F" prove -bvw t/lei-refresh-mail-sync.t'
since that test relies on lacking FD_CLOEXEC to detect dead
lei-daemons, but we still want FD_CLOEXEC when when relying
on tail(1) to check -imapd output.
---
lib/PublicInbox/TestCommon.pm | 22 +++++++++++++++++-----
1 file changed, 17 insertions(+), 5 deletions(-)
diff --git a/lib/PublicInbox/TestCommon.pm b/lib/PublicInbox/TestCommon.pm
index ec300b3f..b7f1eb57 100644
--- a/lib/PublicInbox/TestCommon.pm
+++ b/lib/PublicInbox/TestCommon.pm
@@ -6,7 +6,7 @@ package PublicInbox::TestCommon;
use strict;
use parent qw(Exporter);
use v5.10.1;
-use Fcntl qw(F_SETFD :seek);
+use Fcntl qw(F_SETFD F_GETFD FD_CLOEXEC :seek);
use POSIX qw(dup2);
use IO::Socket::INET;
use File::Spec;
@@ -326,7 +326,7 @@ sub run_script ($;$$) {
die "unable to deal with $ref $redir";
}
}
- my $tail = @tail_paths ? tail_f(@tail_paths) : undef;
+ my $tail = @tail_paths ? tail_f(@tail_paths, $opt) : undef;
if ($key =~ /-(index|cindex|extindex|convert|xcpdb)\z/) {
unshift @argv, '--no-fsync';
}
@@ -442,11 +442,23 @@ sub xqx {
}
sub tail_f (@) {
+ my @f = grep(defined, @_);
$tail_cmd or return; # "tail -F" or "tail -f"
- for (@_) { open(my $fh, '>>', $_) or die $! };
- my $cmd = [ split(/ /, $tail_cmd), @_ ];
+ my $opt = (ref($f[-1]) eq 'HASH') ? pop(@f) : {};
+ my $clofork = $opt->{-CLOFORK} // [];
+ use autodie qw(fcntl open);
+ my @cfmap = map {
+ my $fl = fcntl($_, F_GETFD, 0);
+ fcntl($_, F_SETFD, $fl | FD_CLOEXEC) unless $fl & FD_CLOEXEC;
+ ($_, $fl);
+ } @$clofork;
+ for (@f) { open(my $fh, '>>', $_) };
+ my $cmd = [ split(/ /, $tail_cmd), @f ];
require PublicInbox::Spawn;
my $pid = PublicInbox::Spawn::spawn($cmd, undef, { 1 => 2 });
+ while (my ($io, $fl) = splice(@cfmap, 0, 2)) {
+ fcntl($io, F_SETFD, $fl);
+ }
wait_for_tail($pid, scalar @_);
require PublicInbox::AutoReap;
PublicInbox::AutoReap->new($pid, \&wait_for_tail);
@@ -476,7 +488,7 @@ sub start_script {
}
}
}
- $tail = tail_f(@paths);
+ $tail = tail_f(@paths, $opt);
}
my $oset = PublicInbox::DS::block_signals();
require PublicInbox::OnDestroy;
^ permalink raw reply related [flat|nested] 10+ messages in thread* [PATCH 2/7] daemon: depend on DS event_loop in master process, too
2023-09-11 9:41 [PATCH 0/7] system-related updates and cleanups Eric Wong
2023-09-11 9:41 ` [PATCH 1/7] tests: map CLOFORK->FD_CLOEXEC temporarily for `tail -f' Eric Wong
@ 2023-09-11 9:41 ` Eric Wong
2023-09-11 9:41 ` [PATCH 3/7] ds: use object-oriented API for epoll Eric Wong
` (4 subsequent siblings)
6 siblings, 0 replies; 10+ messages in thread
From: Eric Wong @ 2023-09-11 9:41 UTC (permalink / raw)
To: meta
The awaitpid API turns out to be quite handy for managing
long-lived worker processes. This allows us to ensure all our
uses of signalfd (and kevent emulation) are non-blocking.
---
lib/PublicInbox/DS.pm | 2 +-
lib/PublicInbox/DSKQXS.pm | 12 +-
lib/PublicInbox/Daemon.pm | 252 +++++++++++++++++--------------------
lib/PublicInbox/Sigfd.pm | 12 +-
lib/PublicInbox/Syscall.pm | 6 +-
t/httpd-unix.t | 20 ++-
t/sigfd.t | 4 +-
7 files changed, 146 insertions(+), 162 deletions(-)
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index ff10c9c0..d6e3d10e 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -280,7 +280,7 @@ sub event_loop (;$$) {
my ($sig, $oldset) = @_;
$Epoll //= _InitPoller();
require PublicInbox::Sigfd if $sig;
- my $sigfd = $sig ? PublicInbox::Sigfd->new($sig, 1) : undef;
+ my $sigfd = $sig ? PublicInbox::Sigfd->new($sig) : undef;
if ($sigfd && $sigfd->{is_kq}) {
my $tmp = allowset($sig);
local @SIG{keys %$sig} = values(%$sig);
diff --git a/lib/PublicInbox/DSKQXS.pm b/lib/PublicInbox/DSKQXS.pm
index 3fcb4e40..b6e5c4e9 100644
--- a/lib/PublicInbox/DSKQXS.pm
+++ b/lib/PublicInbox/DSKQXS.pm
@@ -47,16 +47,15 @@ sub new {
# It's wasteful in that it uses another FD, but it simplifies
# our epoll-oriented code.
sub signalfd {
- my ($class, $signo, $nonblock) = @_;
+ my ($class, $signo) = @_;
my $sym = gensym;
- tie *$sym, $class, $signo, $nonblock; # calls TIEHANDLE
+ tie *$sym, $class, $signo; # calls TIEHANDLE
$sym
}
sub TIEHANDLE { # similar to signalfd()
- my ($class, $signo, $nonblock) = @_;
+ my ($class, $signo) = @_;
my $self = $class->new;
- $self->{timeout} = $nonblock ? 0 : -1;
my $kq = $self->{kq};
$kq->EV_SET($_, EVFILT_SIGNAL, EV_ADD) for @$signo;
$self;
@@ -65,7 +64,6 @@ sub TIEHANDLE { # similar to signalfd()
sub READ { # called by sysread() for signalfd compatibility
my ($self, undef, $len, $off) = @_; # $_[1] = buf
die "bad args for signalfd read" if ($len % 128) // defined($off);
- my $timeout = $self->{timeout};
my $sigbuf = $self->{sigbuf} //= [];
my $nr = $len / 128;
my $r = 0;
@@ -78,13 +76,13 @@ sub READ { # called by sysread() for signalfd compatibility
$r += 128;
}
return $r if $r;
- my @events = eval { $self->{kq}->kevent($timeout) };
+ my @events = eval { $self->{kq}->kevent(0) };
# workaround https://rt.cpan.org/Ticket/Display.html?id=116615
if ($@) {
next if $@ =~ /Interrupted system call/;
die;
}
- if (!scalar(@events) && $timeout == 0) {
+ if (!scalar(@events)) {
$! = EAGAIN;
return;
}
diff --git a/lib/PublicInbox/Daemon.pm b/lib/PublicInbox/Daemon.pm
index 88b0fa45..222093bc 100644
--- a/lib/PublicInbox/Daemon.pm
+++ b/lib/PublicInbox/Daemon.pm
@@ -5,8 +5,7 @@
# and designed for handling thousands of untrusted clients over slow
# and/or lossy connections.
package PublicInbox::Daemon;
-use strict;
-use v5.10.1;
+use v5.12;
use Getopt::Long qw(:config gnu_getopt no_ignore_case auto_abbrev);
use IO::Handle; # ->autoflush
use IO::Socket;
@@ -15,10 +14,9 @@ use POSIX qw(WNOHANG :signal_h F_SETFD);
use Socket qw(IPPROTO_TCP SOL_SOCKET);
STDOUT->autoflush(1);
STDERR->autoflush(1);
-use PublicInbox::DS qw(now);
+use PublicInbox::DS qw(now awaitpid);
use PublicInbox::Listener;
use PublicInbox::EOFpipe;
-use PublicInbox::Sigfd;
use PublicInbox::Git;
use PublicInbox::GitAsyncCat;
use PublicInbox::Eml;
@@ -27,9 +25,7 @@ our $SO_ACCEPTFILTER = 0x1000;
my @CMD;
my ($set_user, $oldset);
my (@cfg_listen, $stdout, $stderr, $group, $user, $pid_file, $daemonize);
-my $worker_processes = 1;
-my @listeners;
-my (%pids, %logs);
+my ($nworker, @listeners, %WORKERS, %logs);
my %tls_opt; # scheme://sockname => args for IO::Socket::SSL::SSL_Context->new
my $reexec_pid;
my ($uid, $gid);
@@ -40,6 +36,19 @@ my %SCHEME2PORT = map { $KNOWN_TLS{$_} => $_ + 0 } keys %KNOWN_TLS;
for (keys %KNOWN_STARTTLS) { $SCHEME2PORT{$KNOWN_STARTTLS{$_}} = $_ + 0 }
$SCHEME2PORT{http} = 80;
+our ($parent_pipe, %POST_ACCEPT, %XNETD);
+our %WORKER_SIG = (
+ INT => \&worker_quit,
+ QUIT => \&worker_quit,
+ TERM => \&worker_quit,
+ TTIN => 'IGNORE',
+ TTOU => 'IGNORE',
+ USR1 => \&reopen_logs,
+ USR2 => 'IGNORE',
+ WINCH => 'IGNORE',
+ CHLD => \&PublicInbox::DS::enqueue_reap,
+);
+
sub listener_opt ($) {
my ($str) = @_; # opt1=val1,opt2=val2 (opt may repeat for multi-value)
my $o = {};
@@ -141,8 +150,8 @@ sub load_mod ($;$$) {
\%xn;
}
-sub daemon_prepare ($$) {
- my ($default_listen, $xnetd) = @_;
+sub daemon_prepare ($) {
+ my ($default_listen) = @_;
my $listener_names = {}; # sockname => IO::Handle
$oldset = PublicInbox::DS::block_signals();
@CMD = ($0, @ARGV);
@@ -164,7 +173,7 @@ EOF
'l|listen=s' => \@cfg_listen,
'1|stdout=s' => \$stdout,
'2|stderr=s' => \$stderr,
- 'W|worker-processes=i' => \$worker_processes,
+ 'W|worker-processes=i' => \$nworker,
'P|pid-file=s' => \$pid_file,
'u|user=s' => \$user,
'g|group=s' => \$group,
@@ -218,7 +227,7 @@ EOF
die "$orig specified w/o cert=\n";
}
if ($listener_names->{$l}) { # already inherited
- $xnetd->{$l} = load_mod($scheme, $opt, $l);
+ $XNETD{$l} = load_mod($scheme, $opt, $l);
next;
}
my (%o, $sock_pkg);
@@ -254,7 +263,7 @@ EOF
$s->blocking(0);
my $sockname = sockname($s);
warn "# bound $scheme://$sockname\n";
- $xnetd->{$sockname} //= load_mod($scheme, $opt);
+ $XNETD{$sockname} //= load_mod($scheme, $opt);
$listener_names->{$sockname} = $s;
push @listeners, $s;
}
@@ -268,10 +277,10 @@ EOF
for my $x (@inherited_names) {
$x =~ /:([0-9]+)\z/ or next; # no TLS for AF_UNIX
if (my $scheme = $KNOWN_TLS{$1}) {
- $xnetd->{$x} //= load_mod($scheme);
+ $XNETD{$x} //= load_mod($scheme);
$tls_opt{"$scheme://$x"} ||= accept_tls_opt('');
} elsif (($scheme = $KNOWN_STARTTLS{$1})) {
- $xnetd->{$x} //= load_mod($scheme);
+ $XNETD{$x} //= load_mod($scheme);
$tls_opt{"$scheme://$x"} ||= accept_tls_opt('');
} elsif (defined $stls) {
$tls_opt{"$stls://$x"} ||= accept_tls_opt('');
@@ -280,7 +289,7 @@ EOF
}
if (defined $default_scheme) {
for my $x (@inherited_names) {
- $xnetd->{$x} //= load_mod($default_scheme);
+ $XNETD{$x} //= load_mod($default_scheme);
}
}
die "No listeners bound\n" unless @listeners;
@@ -476,11 +485,9 @@ sub upgrade { # $_[0] = signal name or number (unused)
write_pid($pid_file);
}
my $pid = fork;
- unless (defined $pid) {
+ if (!defined($pid)) {
warn "fork failed: $!\n";
- return;
- }
- if ($pid == 0) {
+ } elsif ($pid == 0) {
$ENV{LISTEN_FDS} = scalar @listeners;
$ENV{LISTEN_PID} = $$;
foreach my $s (@listeners) {
@@ -490,18 +497,17 @@ sub upgrade { # $_[0] = signal name or number (unused)
}
exec @CMD;
die "Failed to exec: $!\n";
+ } else {
+ awaitpid($pid, \&upgrade_aborted);
+ $reexec_pid = $pid;
}
- $reexec_pid = $pid;
}
-sub kill_workers ($) {
- my ($sig) = @_;
- kill $sig, keys(%pids);
-}
+sub kill_workers ($) { kill $_[0], values(%WORKERS) }
-sub upgrade_aborted ($) {
- my ($p) = @_;
- warn "reexec PID($p) died with: $?\n";
+sub upgrade_aborted {
+ my ($pid) = @_;
+ warn "reexec PID($pid) died with: $?\n";
$reexec_pid = undef;
return unless $pid_file;
@@ -513,21 +519,6 @@ sub upgrade_aborted ($) {
warn $@, "\n" if $@;
}
-sub reap_children { # $_[0] = 'CHLD'
- while (1) {
- my $p = waitpid(-1, WNOHANG) or return;
- if (defined $reexec_pid && $p == $reexec_pid) {
- upgrade_aborted($p);
- } elsif (defined(my $id = delete $pids{$p})) {
- warn "worker[$id] PID($p) died with: $?\n";
- } elsif ($p > 0) {
- warn "unknown PID($p) reaped: $?\n";
- } else {
- return;
- }
- }
-}
-
sub unlink_pid_file_safe_ish ($$) {
my ($unlink_pid, $file) = @_;
return unless defined $unlink_pid && $unlink_pid == $$;
@@ -544,92 +535,90 @@ sub unlink_pid_file_safe_ish ($$) {
sub master_quit ($) {
exit unless @listeners;
@listeners = ();
- kill_workers($_[0]);
+ exit unless kill_workers($_[0]);
+}
+
+sub reap_worker { # awaitpid CB
+ my ($pid, $nr) = @_;
+ warn "worker[$nr] died \$?=$?\n" if $?;
+ delete $WORKERS{$nr};
+ exit if !@listeners && !keys(%WORKERS);
+ PublicInbox::DS::requeue(\&start_workers);
+}
+
+sub start_worker ($) {
+ my ($nr) = @_;
+ my $seed = rand(0xffffffff);
+ return unless @listeners;
+ my $pid = fork;
+ if (!defined($pid)) {
+ warn "fork: $!";
+ } elsif ($pid == 0) {
+ undef %WORKERS;
+ PublicInbox::DS::Reset();
+ srand($seed);
+ eval { Net::SSLeay::randomize() };
+ $set_user->() if $set_user;
+ PublicInbox::EOFpipe->new($parent_pipe, \&worker_quit);
+ worker_loop();
+ exit 0;
+ } else {
+ $WORKERS{$nr} = $pid;
+ awaitpid($pid, \&reap_worker, $nr);
+ }
+}
+
+sub start_workers {
+ for my $nr (grep { !defined($WORKERS{$_}) } (0..($nworker - 1))) {
+ start_worker($nr);
+ }
+}
+
+sub trim_workers {
+ my @nr = grep { $_ >= $nworker } keys %WORKERS;
+ kill('TERM', @WORKERS{@nr});
}
sub master_loop {
- pipe(my ($p0, $p1)) or die "failed to create parent-pipe: $!";
- my $set_workers = $worker_processes;
+ local $parent_pipe;
+ pipe($parent_pipe, my $p1) or die "failed to create parent-pipe: $!";
+ my $set_workers = $nworker; # for SIGWINCH
reopen_logs();
- my $ignore_winch;
- my $sig = {
+ my $msig = {
USR1 => sub { reopen_logs(); kill_workers($_[0]); },
USR2 => \&upgrade,
QUIT => \&master_quit,
INT => \&master_quit,
TERM => \&master_quit,
WINCH => sub {
- return if $ignore_winch || !@listeners;
- if (-t STDIN || -t STDOUT || -t STDERR) {
- $ignore_winch = 1;
- warn <<EOF;
-ignoring SIGWINCH since we are not daemonized
-EOF
- } else {
- $worker_processes = 0;
- }
+ $nworker = 0;
+ trim_workers();
},
HUP => sub {
- return unless @listeners;
- $worker_processes = $set_workers;
+ $nworker = $set_workers; # undo WINCH
kill_workers($_[0]);
+ PublicInbox::DS::requeue(\&start_workers)
},
TTIN => sub {
- return unless @listeners;
- if ($set_workers > $worker_processes) {
- ++$worker_processes;
+ if ($set_workers > $nworker) {
+ ++$nworker;
} else {
- $worker_processes = ++$set_workers;
+ $nworker = ++$set_workers;
}
+ PublicInbox::DS::requeue(\&start_workers);
},
TTOU => sub {
- $worker_processes = --$set_workers if $set_workers > 0;
+ return if $nworker <= 0;
+ --$nworker;
+ trim_workers();
},
- CHLD => \&reap_children,
+ CHLD => \&PublicInbox::DS::enqueue_reap,
};
- my $sigfd = PublicInbox::Sigfd->new($sig);
- local @SIG{keys %$sig} = values(%$sig) unless $sigfd;
- PublicInbox::DS::sig_setmask($oldset) if !$sigfd;
- while (1) { # main loop
- my $n = scalar keys %pids;
- unless (@listeners) {
- exit if $n == 0;
- $set_workers = $worker_processes = $n = 0;
- }
-
- if ($n > $worker_processes) {
- while (my ($k, $v) = each %pids) {
- kill('TERM', $k) if $v >= $worker_processes;
- }
- $n = $worker_processes;
- }
- my $want = $worker_processes - 1;
- if ($n <= $want) {
- PublicInbox::DS::block_signals() if !$sigfd;
- for my $i ($n..$want) {
- my $seed = rand(0xffffffff);
- my $pid = fork;
- if (!defined $pid) {
- warn "failed to fork worker[$i]: $!\n";
- } elsif ($pid == 0) {
- srand($seed);
- eval { Net::SSLeay::randomize() };
- $set_user->() if $set_user;
- return $p0; # run normal work code
- } else {
- warn "PID=$pid is worker[$i]\n";
- $pids{$pid} = $i;
- }
- }
- PublicInbox::DS::sig_setmask($oldset) if !$sigfd;
- }
-
- if ($sigfd) { # Linux and IO::KQueue users:
- $sigfd->wait_once;
- } else { # wake up every second
- sleep(1);
- }
- }
+ $msig->{WINCH} = sub {
+ warn "ignoring SIGWINCH since we are not daemonized\n";
+ } if -t STDIN || -t STDOUT || -t STDERR;
+ start_workers();
+ PublicInbox::DS::event_loop($msig, $oldset);
exit # never gets here, just for documentation
}
@@ -659,56 +648,45 @@ sub defer_accept ($$) {
}
}
-sub daemon_loop ($) {
- my ($xnetd) = @_;
+sub daemon_loop () {
local $PublicInbox::Config::DEDUPE = {}; # enable dedupe cache
- my $refresh = sub {
+ my $refresh = $WORKER_SIG{HUP} = sub {
my ($sig) = @_;
%$PublicInbox::Config::DEDUPE = (); # clear cache
- for my $xn (values %$xnetd) {
+ for my $xn (values %XNETD) {
delete $xn->{tlsd}->{ssl_ctx}; # PublicInbox::TLS::start
eval { $xn->{refresh}->($sig) };
warn "refresh $@\n" if $@;
}
};
- my %post_accept;
while (my ($k, $ctx_opt) = each %tls_opt) {
$ctx_opt // next;
my ($scheme, $l) = split(m!://!, $k, 2);
- my $xn = $xnetd->{$l} // die "BUG: no xnetd for $k";
+ my $xn = $XNETD{$l} // die "BUG: no xnetd for $k";
$xn->{tlsd}->{ssl_ctx_opt} //= $ctx_opt;
$scheme =~ m!\A(?:https|imaps|nntps|pop3s)! and
- $post_accept{$l} = tls_cb(@$xn{qw(post_accept tlsd)});
+ $POST_ACCEPT{$l} = tls_cb(@$xn{qw(post_accept tlsd)});
}
undef %tls_opt;
- my $sig = {
- HUP => $refresh,
- INT => \&worker_quit,
- QUIT => \&worker_quit,
- TERM => \&worker_quit,
- TTIN => 'IGNORE',
- TTOU => 'IGNORE',
- USR1 => \&reopen_logs,
- USR2 => 'IGNORE',
- WINCH => 'IGNORE',
- CHLD => \&PublicInbox::DS::enqueue_reap,
- };
- if ($worker_processes > 0) {
+ if ($nworker > 0) {
$refresh->(); # preload by default
- my $fh = master_loop(); # returns if in child process
- PublicInbox::EOFpipe->new($fh, \&worker_quit);
+ return master_loop();
} else {
reopen_logs();
$set_user->() if $set_user;
- $sig->{USR2} = sub { worker_quit() if upgrade() };
+ $WORKER_SIG{USR2} = sub { worker_quit() if upgrade() };
$refresh->();
}
+ worker_loop();
+}
+
+sub worker_loop {
$uid = $gid = undef;
reopen_logs();
@listeners = map {;
my $l = sockname($_);
- my $tls_cb = $post_accept{$l};
- my $xn = $xnetd->{$l} // die "BUG: no xnetd for $l";
+ my $tls_cb = $POST_ACCEPT{$l};
+ my $xn = $XNETD{$l} // die "BUG: no xnetd for $l";
# NNTPS, HTTPS, HTTP, IMAPS and POP3S are client-first traffic
# IMAP, NNTP and POP3 are server-first
@@ -718,20 +696,24 @@ sub daemon_loop ($) {
PublicInbox::Listener->new($_, $tls_cb || $xn->{post_accept},
$xn->{'multi-accept'})
} @listeners;
- PublicInbox::DS::event_loop($sig, $oldset);
+ PublicInbox::DS::event_loop(\%WORKER_SIG, $oldset);
}
sub run {
my ($default_listen) = @_;
- daemon_prepare($default_listen, my $xnetd = {});
+ $nworker = 1;
+ local (%XNETD, %POST_ACCEPT);
+ daemon_prepare($default_listen);
my $for_destroy = daemonize();
# localize GCF2C for tests:
local $PublicInbox::GitAsyncCat::GCF2C;
local $PublicInbox::Git::async_warn = 1;
local $SIG{__WARN__} = PublicInbox::Eml::warn_ignore_cb();
+ local %WORKER_SIG = %WORKER_SIG;
+ local %POST_ACCEPT;
- daemon_loop($xnetd);
+ daemon_loop();
PublicInbox::DS->Reset;
# ->DESTROY runs when $for_destroy goes out-of-scope
}
diff --git a/lib/PublicInbox/Sigfd.pm b/lib/PublicInbox/Sigfd.pm
index 5656baeb..b8a1ddfb 100644
--- a/lib/PublicInbox/Sigfd.pm
+++ b/lib/PublicInbox/Sigfd.pm
@@ -12,26 +12,22 @@ use POSIX ();
# returns a coderef to unblock signals if neither signalfd or kqueue
# are available.
sub new {
- my ($class, $sig, $nonblock) = @_;
+ my ($class, $sig) = @_;
my %signo = map {;
# $num => [ $cb, $signame ];
($SIGNUM{$_} // POSIX->can("SIG$_")->()) => [ $sig->{$_}, $_ ]
} keys %$sig;
my $self = bless { sig => \%signo }, $class;
my $io;
- my $fd = signalfd([keys %signo], $nonblock);
+ my $fd = signalfd([keys %signo]);
if (defined $fd && $fd >= 0) {
open($io, '+<&=', $fd) or die "open: $!";
} elsif (eval { require PublicInbox::DSKQXS }) {
- $io = PublicInbox::DSKQXS->signalfd([keys %signo], $nonblock);
+ $io = PublicInbox::DSKQXS->signalfd([keys %signo]);
} else {
return; # wake up every second to check for signals
}
- if ($nonblock) { # it can go into the event loop
- $self->SUPER::new($io, EPOLLIN | EPOLLET);
- } else { # master main loop
- $self->{sock} = $io;
- }
+ $self->SUPER::new($io, EPOLLIN | EPOLLET);
$self->{is_kq} = 1 if tied(*$io);
$self;
}
diff --git a/lib/PublicInbox/Syscall.pm b/lib/PublicInbox/Syscall.pm
index 4609b32d..14cd1720 100644
--- a/lib/PublicInbox/Syscall.pm
+++ b/lib/PublicInbox/Syscall.pm
@@ -327,15 +327,15 @@ sub epoll_wait_mod8 {
}
}
-sub signalfd ($$) {
- my ($signos, $nonblock) = @_;
+sub signalfd ($) {
+ my ($signos) = @_;
if ($SYS_signalfd4) {
my $set = POSIX::SigSet->new(@$signos);
syscall($SYS_signalfd4, -1, "$$set",
# $Config{sig_count} is NSIG, so this is NSIG/8:
int($Config{sig_count}/8),
# SFD_NONBLOCK == O_NONBLOCK for every architecture
- ($nonblock ? O_NONBLOCK : 0) |$SFD_CLOEXEC);
+ O_NONBLOCK|$SFD_CLOEXEC);
} else {
$! = ENOSYS;
undef;
diff --git a/t/httpd-unix.t b/t/httpd-unix.t
index 414ca0c8..d90c6c3e 100644
--- a/t/httpd-unix.t
+++ b/t/httpd-unix.t
@@ -2,8 +2,7 @@
# Copyright (C) all contributors <meta@public-inbox.org>
# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
# Tests for binding Unix domain sockets
-use strict;
-use Test::More;
+use v5.12;
use PublicInbox::TestCommon;
use Errno qw(EADDRINUSE);
use Cwd qw(abs_path);
@@ -12,6 +11,7 @@ use Fcntl qw(FD_CLOEXEC F_SETFD);
require_mods(qw(Plack::Util Plack::Builder HTTP::Date HTTP::Status));
use IO::Socket::UNIX;
use POSIX qw(mkfifo);
+require PublicInbox::Sigfd;
my ($tmpdir, $for_destroy) = tmpdir();
my $unix = "$tmpdir/unix.sock";
my $psgi = './t/httpd-corner.psgi';
@@ -99,16 +99,17 @@ check_sock($unix);
# portable Perl can delay or miss signal dispatches due to races,
# so disable some tests on systems lacking signalfd(2) or EVFILT_SIGNAL
-my $has_sigfd = PublicInbox::Sigfd->new({}, 0) ? 1 : $ENV{TEST_UNRELIABLE};
+my $has_sigfd = PublicInbox::Sigfd->new({}) ? 1 : $ENV{TEST_UNRELIABLE};
+PublicInbox::DS::Reset() if $has_sigfd;
sub delay_until {
- my $cond = shift;
+ my ($cond, $msg) = @_;
my $end = time + 30;
do {
return if $cond->();
tick(0.012);
} until (time > $end);
- Carp::confess('condition failed');
+ Carp::confess($msg // 'condition failed');
}
SKIP: {
@@ -140,6 +141,8 @@ SKIP: {
is(select($rvec, undef, undef, 1), 1, 'timeout for pipe HUP');
is(my $undef = <$p0>, undef, 'process closed pipe writer at exit');
ok(!-e $pid_file, "$w pid file unlinked at exit");
+ delay_until(sub { !kill(0, $pid) },
+ "daemonized $w really not running");
}
my $httpd = abs_path('blib/script/public-inbox-httpd');
@@ -181,6 +184,9 @@ SKIP: {
delay_until(sub {
$pid == (eval { $read_pid->($pid_file) } // 0)
});
+
+ delay_until(sub { !kill(0, $new_pid) }, 'new PID really died');
+
is($read_pid->($pid_file), $pid, 'old PID file restored');
ok(!-f "$pid_file.oldbin", '.oldbin PID file gone');
@@ -196,7 +202,7 @@ SKIP: {
# drop the old parent
kill('QUIT', $old_pid) or die "QUIT failed: $!";
- delay_until(sub { !kill(0, $old_pid) }); # UGH
+ delay_until(sub { !kill(0, $old_pid) }, 'old PID really died');
ok(!-f "$pid_file.oldbin", '.oldbin PID file gone');
@@ -209,6 +215,7 @@ SKIP: {
is(my $u = <$p0>, undef, 'process closed pipe writer at exit');
ok(!-f $pid_file, 'PID file is gone');
+ delay_until(sub { !kill(0, $new_pid) }, 'new PID really died');
}
if ('try USR2 without workers (-W0)') {
@@ -234,6 +241,7 @@ SKIP: {
is(select($rvec, undef, undef, 1), 1, 'timeout for pipe HUP');
is(my $u = <$p0>, undef, 'process closed pipe writer at exit');
ok(!-f $pid_file, 'PID file is gone');
+ delay_until(sub { !kill(0, $pid) }, '-W0 daemon is gone');
}
}
diff --git a/t/sigfd.t b/t/sigfd.t
index f6449dab..9a7b947d 100644
--- a/t/sigfd.t
+++ b/t/sigfd.t
@@ -29,7 +29,7 @@ SKIP: {
ok(!defined($hit->{USR2}), 'no USR2 yet') or diag explain($hit);
PublicInbox::DS->Reset;
ok($PublicInbox::Syscall::SIGNUM{WINCH}, 'SIGWINCH number defined');
- my $sigfd = PublicInbox::Sigfd->new($sig, 0);
+ my $sigfd = PublicInbox::Sigfd->new($sig);
if ($sigfd) {
$linux_sigfd = 1 if $^O eq 'linux';
$has_sigfd = 1;
@@ -57,7 +57,7 @@ SKIP: {
PublicInbox::DS->Reset;
$sigfd = undef;
- my $nbsig = PublicInbox::Sigfd->new($sig, 1);
+ my $nbsig = PublicInbox::Sigfd->new($sig);
ok($nbsig, 'Sigfd->new SFD_NONBLOCK works');
is($nbsig->wait_once, undef, 'nonblocking ->wait_once');
ok($! == Errno::EAGAIN, 'got EAGAIN');
^ permalink raw reply related [flat|nested] 10+ messages in thread* [PATCH 3/7] ds: use object-oriented API for epoll
2023-09-11 9:41 [PATCH 0/7] system-related updates and cleanups Eric Wong
2023-09-11 9:41 ` [PATCH 1/7] tests: map CLOFORK->FD_CLOEXEC temporarily for `tail -f' Eric Wong
2023-09-11 9:41 ` [PATCH 2/7] daemon: depend on DS event_loop in master process, too Eric Wong
@ 2023-09-11 9:41 ` Eric Wong
2023-09-11 9:41 ` [PATCH 4/7] favor poll(2) for most daemons Eric Wong
` (3 subsequent siblings)
6 siblings, 0 replies; 10+ messages in thread
From: Eric Wong @ 2023-09-11 9:41 UTC (permalink / raw)
To: meta
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.
---
MANIFEST | 1 +
lib/PublicInbox/DS.pm | 40 ++++++++++++---------------------
lib/PublicInbox/DSKQXS.pm | 46 +++++++++++++++++---------------------
lib/PublicInbox/DSPoll.pm | 31 +++++++++----------------
lib/PublicInbox/Epoll.pm | 23 +++++++++++++++++++
lib/PublicInbox/Syscall.pm | 6 -----
t/ds-kqxs.t | 4 ++--
t/ds-poll.t | 29 +++++++++++-------------
t/epoll.t | 23 +++++++++----------
9 files changed, 95 insertions(+), 108 deletions(-)
create mode 100644 lib/PublicInbox/Epoll.pm
diff --git a/MANIFEST b/MANIFEST
index 1fe1c7f7..d7a670b8 100644
--- a/MANIFEST
+++ b/MANIFEST
@@ -184,6 +184,7 @@ lib/PublicInbox/EOFpipe.pm
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
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index d6e3d10e..9300ac77 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -28,7 +28,8 @@ use POSIX qw(WNOHANG sigprocmask SIG_SETMASK SIG_UNBLOCK);
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);
@@ -41,8 +42,7 @@ my $reap_armed;
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
@@ -75,7 +75,6 @@ sub Reset {
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) ||
@@ -126,21 +125,13 @@ sub add_uniq_timer { # ($name, $secs, $coderef, @args) = @_;
# 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) }
@@ -307,7 +298,7 @@ sub event_loop (;$$) {
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
@@ -345,7 +336,7 @@ sub new {
$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;
@@ -399,9 +390,7 @@ sub close {
# 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
@@ -619,9 +608,8 @@ sub msg_more ($$) {
}
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)
diff --git a/lib/PublicInbox/DSKQXS.pm b/lib/PublicInbox/DSKQXS.pm
index b6e5c4e9..8ef8ffb6 100644
--- a/lib/PublicInbox/DSKQXS.pm
+++ b/lib/PublicInbox/DSKQXS.pm
@@ -12,13 +12,10 @@
# 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 }
@@ -97,30 +94,29 @@ sub READ { # called by sysread() for signalfd compatibility
# 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 = $@) {
diff --git a/lib/PublicInbox/DSPoll.pm b/lib/PublicInbox/DSPoll.pm
index 56a400c2..fc282de0 100644
--- a/lib/PublicInbox/DSPoll.pm
+++ b/lib/PublicInbox/DSPoll.pm
@@ -1,4 +1,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>
@@ -9,28 +9,13 @@
# 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) {
@@ -54,4 +39,10 @@ sub epoll_wait {
}
}
+sub ep_del { delete($_[0]->{fileno($_[1])}); 0 }
+sub ep_add { $_[0]->{fileno($_[1])} = $_[2]; 0 }
+
+no warnings 'once';
+*ep_mod = \&ep_add;
+
1;
diff --git a/lib/PublicInbox/Epoll.pm b/lib/PublicInbox/Epoll.pm
new file mode 100644
index 00000000..d55c8535
--- /dev/null
+++ b/lib/PublicInbox/Epoll.pm
@@ -0,0 +1,23 @@
+# 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;
diff --git a/lib/PublicInbox/Syscall.pm b/lib/PublicInbox/Syscall.pm
index 14cd1720..0a0912fb 100644
--- a/lib/PublicInbox/Syscall.pm
+++ b/lib/PublicInbox/Syscall.pm
@@ -29,12 +29,6 @@ our @EXPORT_OK = qw(epoll_ctl epoll_create epoll_wait
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,
diff --git a/t/ds-kqxs.t b/t/ds-kqxs.t
index 43c71fed..57acb53f 100644
--- a/t/ds-kqxs.t
+++ b/t/ds-kqxs.t
@@ -1,9 +1,9 @@
-# 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'
diff --git a/t/ds-poll.t b/t/ds-poll.t
index d8861369..57fac3ef 100644
--- a/t/ds-poll.t
+++ b/t/ds-poll.t
@@ -1,12 +1,11 @@
-# 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;
@@ -14,37 +13,35 @@ 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;
diff --git a/t/epoll.t b/t/epoll.t
index f346b387..54dc6f47 100644
--- a/t/epoll.t
+++ b/t/epoll.t
@@ -1,25 +1,22 @@
#!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;
^ permalink raw reply related [flat|nested] 10+ messages in thread* [PATCH 4/7] favor poll(2) for most daemons
2023-09-11 9:41 [PATCH 0/7] system-related updates and cleanups Eric Wong
` (2 preceding siblings ...)
2023-09-11 9:41 ` [PATCH 3/7] ds: use object-oriented API for epoll Eric Wong
@ 2023-09-11 9:41 ` Eric Wong
2023-09-11 9:41 ` [PATCH 5/7] dspoll: switch to the documented IO::Poll API Eric Wong
` (2 subsequent siblings)
6 siblings, 0 replies; 10+ messages in thread
From: Eric Wong @ 2023-09-11 9:41 UTC (permalink / raw)
To: meta
public-inbox-watch, lei-daemon, the master process of
public-inbox-(netd|httpd|imapd|nntpd|pop3d),
and the (mostly) Perl implementation of XapHelper do not
have many FDs to watch so epoll|kqueue end up being overkill.
Of course, *BSDs already have separate kqueue FDs emulating
signalfd and/or inotify, even.
In other words, only the worker processes of
public-inbox-(netd|httpd|imapd|nntpd|pop3d) are expected
to see C10K (or C100K) types of traffic where epoll|kqueue
shine.
Perhaps lei could benefit from epoll/kqueue on some virtual users
IMAP/JMAP system one day; as could -watch with many IMAP IDLE
folders; but we'll probably add a knob if/when it comes to that.
---
lib/PublicInbox/DS.pm | 20 +++++++++++---------
lib/PublicInbox/Daemon.pm | 2 ++
lib/PublicInbox/TestCommon.pm | 21 +++++++++++++++++++--
t/httpd-unix.t | 1 +
t/lei-daemon.t | 1 +
t/watch_maildir.t | 1 +
t/xap_helper.t | 7 ++++---
7 files changed, 39 insertions(+), 14 deletions(-)
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index 9300ac77..f2b14799 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -31,6 +31,7 @@ use Scalar::Util qw(blessed);
use PublicInbox::Syscall qw(%SIGNUM
EPOLLIN EPOLLOUT EPOLLONESHOT EPOLLEXCLUSIVE);
use PublicInbox::Tmpfile;
+use PublicInbox::DSPoll;
use Errno qw(EAGAIN EINVAL ECHILD EINTR);
use Carp qw(carp croak);
our @EXPORT_OK = qw(now msg_more awaitpid add_timer add_uniq_timer);
@@ -42,7 +43,7 @@ my $reap_armed;
my $ToClose; # sockets to close when event loop is done
our (
%DescriptorMap, # fd (num) -> PublicInbox::DS object
- $Epoll, # global Epoll, DSPoll, or DSKQXS ref
+ $Poller, # global Epoll, DSPoll, or DSKQXS ref
@post_loop_do, # subref + args to call at the end of each loop
@@ -75,13 +76,14 @@ sub Reset {
my @q = delete @Stack{keys %Stack};
for my $q (@q) { @$q = () }
$AWAIT_PIDS = $nextq = $ToClose = undef;
- $Epoll = undef; # may call DSKQXS::DESTROY
+ $Poller = undef; # may call DSKQXS::DESTROY
} while (@Timers || keys(%Stack) || $nextq || $AWAIT_PIDS ||
$ToClose || keys(%DescriptorMap) ||
@post_loop_do || keys(%UniqTimer));
$reap_armed = undef;
$LoopTimeout = -1; # no timeout by default
+ $Poller = PublicInbox::DSPoll->new;
}
=head2 C<< CLASS->SetLoopTimeout( $timeout ) >>
@@ -123,7 +125,7 @@ sub add_uniq_timer { # ($name, $secs, $coderef, @args) = @_;
$UniqTimer{$_[0]} //= _add_named_timer(@_);
}
-# caller sets return value to $Epoll
+# caller sets return value to $Poller
sub _InitPoller () {
my @try = ($^O eq 'linux' ? 'Epoll' : 'DSKQXS');
my $cls;
@@ -269,7 +271,7 @@ sub unblockset ($) { sigset_prep $_[0], 'emptyset', 'addset' }
# C<post_loop_do> for how to exit the loop.
sub event_loop (;$$) {
my ($sig, $oldset) = @_;
- $Epoll //= _InitPoller();
+ $Poller //= _InitPoller();
require PublicInbox::Sigfd if $sig;
my $sigfd = $sig ? PublicInbox::Sigfd->new($sig) : undef;
if ($sigfd && $sigfd->{is_kq}) {
@@ -298,7 +300,7 @@ sub event_loop (;$$) {
my $timeout = RunTimers();
# get up to 1000 events
- $Epoll->ep_wait(1000, $timeout, \@events);
+ $Poller->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
@@ -334,9 +336,9 @@ sub new {
$self->{sock} = $sock;
my $fd = fileno($sock);
- $Epoll //= _InitPoller();
+ $Poller //= _InitPoller();
retry:
- if ($Epoll->ep_add($sock, $ev)) {
+ if ($Poller->ep_add($sock, $ev)) {
if ($! == EINVAL && ($ev & EPOLLEXCLUSIVE)) {
$ev &= ~EPOLLEXCLUSIVE;
goto retry;
@@ -390,7 +392,7 @@ sub close {
# if we're using epoll, we have to remove this from our epoll fd so we stop getting
# notifications about it
- $Epoll->ep_del($sock) and croak("EPOLL_CTL_DEL($self/$sock): $!");
+ $Poller->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
@@ -609,7 +611,7 @@ sub msg_more ($$) {
sub epwait ($$) {
my ($io, $ev) = @_;
- $Epoll->ep_mod($io, $ev) and croak("EPOLL_CTL_MOD($io): $!");
+ $Poller->ep_mod($io, $ev) and croak("EPOLL_CTL_MOD($io): $!");
}
# return true if complete, false if incomplete (or failure)
diff --git a/lib/PublicInbox/Daemon.pm b/lib/PublicInbox/Daemon.pm
index 222093bc..07883153 100644
--- a/lib/PublicInbox/Daemon.pm
+++ b/lib/PublicInbox/Daemon.pm
@@ -556,6 +556,7 @@ sub start_worker ($) {
} elsif ($pid == 0) {
undef %WORKERS;
PublicInbox::DS::Reset();
+ local $PublicInbox::DS::Poller; # allow epoll/kqueue
srand($seed);
eval { Net::SSLeay::randomize() };
$set_user->() if $set_user;
@@ -677,6 +678,7 @@ sub daemon_loop () {
$WORKER_SIG{USR2} = sub { worker_quit() if upgrade() };
$refresh->();
}
+ local $PublicInbox::DS::Poller; # allow epoll/kqueue
worker_loop();
}
diff --git a/lib/PublicInbox/TestCommon.pm b/lib/PublicInbox/TestCommon.pm
index b7f1eb57..17057e18 100644
--- a/lib/PublicInbox/TestCommon.pm
+++ b/lib/PublicInbox/TestCommon.pm
@@ -25,7 +25,7 @@ BEGIN {
create_coderepo no_scm_rights
tcp_host_port test_lei lei lei_ok $lei_out $lei_err $lei_opt
test_httpd xbail require_cmd is_xdeeply tail_f
- ignore_inline_c_missing);
+ ignore_inline_c_missing no_pollerfd);
require Test::More;
my @methods = grep(!/\W/, @Test::More::EXPORT);
eval(join('', map { "*$_=\\&Test::More::$_;" } @methods));
@@ -843,7 +843,24 @@ sub test_httpd ($$;$$) {
}
};
-
+sub no_pollerfd ($) {
+ my ($pid) = @_;
+ my ($re, @cmd);
+ $^O eq 'linux' and
+ ($re, @cmd) = (qr/\Q[eventpoll]\E/, qw(lsof -p), $pid);
+ # n.b. *BSDs uses kqueue to emulate signalfd and/or inotify,
+ # and we can't distinguish which is which easily.
+ SKIP: {
+ (@cmd && $re) or
+ skip 'open poller test is Linux-only', 1;
+ my $bin = require_cmd($cmd[0], 1) or skip "$cmd[0] missing", 1;
+ $cmd[0] = $bin;
+ my @of = xqx(\@cmd, {}, {2 => \(my $e)});
+ my $err = $?;
+ skip "$bin broken? (\$?=$err) ($e)", 1 if $err;
+ is(grep(/$re/, @of), 0, "no $re FDs") or diag explain(\@of);
+ }
+}
package PublicInbox::TestCommon::InboxWakeup;
use strict;
sub on_inbox_unlock { ${$_[0]}->($_[1]) }
diff --git a/t/httpd-unix.t b/t/httpd-unix.t
index d90c6c3e..95f589ad 100644
--- a/t/httpd-unix.t
+++ b/t/httpd-unix.t
@@ -135,6 +135,7 @@ SKIP: {
check_sock($unix);
ok(-s $pid_file, "$w pid file written");
my $pid = $read_pid->($pid_file);
+ no_pollerfd($pid) if $w eq '-W1';
is(kill('TERM', $pid), 1, "signaled daemonized $w process");
vec(my $rvec = '', fileno($p0), 1) = 1;
delete $td->{-extra}; # drop tail(1) process
diff --git a/t/lei-daemon.t b/t/lei-daemon.t
index 78ed265e..2be967be 100644
--- a/t/lei-daemon.t
+++ b/t/lei-daemon.t
@@ -21,6 +21,7 @@ test_lei({ daemon_only => 1 }, sub {
is($lei_err, '', 'no error from daemon-pid');
like($lei_out, qr/\A[0-9]+\n\z/s, 'pid returned') or BAIL_OUT;
chomp(my $pid = $lei_out);
+ no_pollerfd($pid);
ok(kill(0, $pid), 'pid is valid');
ok(-S $sock, 'sock created');
is(-s $err_log, 0, 'nothing in errors.log');
diff --git a/t/watch_maildir.t b/t/watch_maildir.t
index 6836a3d9..d0df1c1e 100644
--- a/t/watch_maildir.t
+++ b/t/watch_maildir.t
@@ -151,6 +151,7 @@ More majordomo info at http://vger.kernel.org/majordomo-info.html\n);
# n.b. --no-scan is only intended for testing atm
my $wm = start_script([qw(-watch --no-scan)], $env);
+ no_pollerfd($wm->{pid});
my $eml = eml_load('t/data/0001.patch');
$eml->header_set('Cc', $addr);
my $em = PublicInbox::Emergency->new($maildir);
diff --git a/t/xap_helper.t b/t/xap_helper.t
index fe5d2d14..54bef191 100644
--- a/t/xap_helper.t
+++ b/t/xap_helper.t
@@ -93,7 +93,7 @@ my $test = sub {
my $stats = do { local $/; <$err_rd> };
is($stats, "mset.size=6 nr_out=6\n", 'mset.size reported');
- return $ar if $cinfo{pid} == $pid;
+ return wantarray ? ($ar, $s) : $ar if $cinfo{pid} == $pid;
# test worker management:
kill('TERM', $cinfo{pid});
@@ -136,15 +136,16 @@ my $test = sub {
is(scalar keys %pids, 1, 'have one pid') or diag explain(\%pids);
is($info{pid}, (keys %pids)[0], 'kept oldest PID after TTOU');
- $ar;
+ wantarray ? ($ar, $s) : $ar;
};
my @NO_CXX = (1);
unless ($ENV{TEST_XH_CXX_ONLY}) {
my $ar = $test->(qw[-MPublicInbox::XapHelper -e
PublicInbox::XapHelper::start('-j0')]);
- $ar = $test->(qw[-MPublicInbox::XapHelper -e
+ ($ar, my $s) = $test->(qw[-MPublicInbox::XapHelper -e
PublicInbox::XapHelper::start('-j1')]);
+ no_pollerfd($ar->{pid});
}
SKIP: {
eval {
^ permalink raw reply related [flat|nested] 10+ messages in thread* [PATCH 5/7] dspoll: switch to the documented IO::Poll API
2023-09-11 9:41 [PATCH 0/7] system-related updates and cleanups Eric Wong
` (3 preceding siblings ...)
2023-09-11 9:41 ` [PATCH 4/7] favor poll(2) for most daemons Eric Wong
@ 2023-09-11 9:41 ` Eric Wong
2023-09-12 2:34 ` Eric Wong
2023-09-11 9:41 ` [PATCH 6/7] ds: use constants for @UNBLOCKABLE list Eric Wong
2023-09-11 9:41 ` [PATCH 7/7] spawn: do not block ABRT/BUS/ILL/SEGV signals Eric Wong
6 siblings, 1 reply; 10+ messages in thread
From: Eric Wong @ 2023-09-11 9:41 UTC (permalink / raw)
To: meta
IO::Poll::_poll has always been an undocumented API. While it's
remained working so far (since the early 2000s with Danga::Socket),
I'm uncomfortable continuing with it moving forward since it's
not documented (the leading underscore typically means it's
not meant to be used by 3rd-parties).
So switch to the documented API and just learn to live with some
redundant object references and awkwardness in the API.
---
lib/PublicInbox/DSPoll.pm | 43 +++++++++++++++++++--------------------
1 file changed, 21 insertions(+), 22 deletions(-)
diff --git a/lib/PublicInbox/DSPoll.pm b/lib/PublicInbox/DSPoll.pm
index fc282de0..8ab5d19f 100644
--- a/lib/PublicInbox/DSPoll.pm
+++ b/lib/PublicInbox/DSPoll.pm
@@ -13,34 +13,33 @@ use v5.12;
use IO::Poll;
use PublicInbox::Syscall qw(EPOLLONESHOT EPOLLIN EPOLLOUT);
-sub new { bless {}, __PACKAGE__ } # fd => events
+sub new { bless { poll => IO::Poll->new }, __PACKAGE__ } # fd => events
sub ep_wait {
my ($self, $maxevents, $timeout_msec, $events) = @_;
- my @pset;
- while (my ($fd, $events) = each %$self) {
- my $pevents = $events & EPOLLIN ? POLLIN : 0;
- $pevents |= $events & EPOLLOUT ? POLLOUT : 0;
- push(@pset, $fd, $pevents);
- }
- @$events = ();
- my $n = IO::Poll::_poll($timeout_msec, @pset);
- if ($n >= 0) {
- for (my $i = 0; $i < @pset; ) {
- my $fd = $pset[$i++];
- my $revents = $pset[$i++] or next;
- delete($self->{$fd}) if $self->{$fd} & EPOLLONESHOT;
- push @$events, $fd;
- }
- my $nevents = scalar @$events;
- if ($n != $nevents) {
- warn "BUG? poll() returned $n, but got $nevents";
- }
+ $self->{poll}->poll($timeout_msec/1000) > 0 or return (@$events = ());
+ my @io = $self->{poll}->handles(POLLIN|POLLOUT);
+ @$events = map { fileno($_) } @io;
+ for (@$events) {
+ my $io = shift @io;
+ $self->{poll}->remove($io) if delete($self->{oneshot}->{$_});
}
}
-sub ep_del { delete($_[0]->{fileno($_[1])}); 0 }
-sub ep_add { $_[0]->{fileno($_[1])} = $_[2]; 0 }
+sub ep_del {
+ my ($self, $io) = @_;
+ delete $self->{oneshot}->{fileno($io)};
+ $self->{poll}->remove($io);
+ 0;
+}
+
+sub ep_add {
+ my ($self, $io, $ev) = @_;
+ $self->{oneshot}->{fileno($io)} = 1 if $ev & EPOLLONESHOT;
+ $self->{poll}->mask($io, ($ev & EPOLLIN ? POLLIN : 0) |
+ ($ev & EPOLLOUT ? POLLOUT : 0));
+ 0;
+}
no warnings 'once';
*ep_mod = \&ep_add;
^ permalink raw reply related [flat|nested] 10+ messages in thread* Re: [PATCH 5/7] dspoll: switch to the documented IO::Poll API
2023-09-11 9:41 ` [PATCH 5/7] dspoll: switch to the documented IO::Poll API Eric Wong
@ 2023-09-12 2:34 ` Eric Wong
2023-09-12 6:13 ` [PATCH 8/7] provide select(2) backend for PublicInbox::DS Eric Wong
0 siblings, 1 reply; 10+ messages in thread
From: Eric Wong @ 2023-09-12 2:34 UTC (permalink / raw)
To: meta
Eric Wong <e@80x24.org> wrote:
> So switch to the documented API and just learn to live with some
> redundant object references and awkwardness in the API.
Unfortunately, this introduces unintended consequences with
object destruction order and some explicit calls to DS->Reset
(and modifications to destruction order) becomes necessary.
I noticed this testing t/gcf2_client.t (and also realized
libgit2 isn't in the CI deps, gotta add that along with
a few other tweaks brewing...)
So I guess we'll just have to keep an eye on IO::Poll::_poll
changes or replace it with select(2) since it should all
be low-numbered FDs anyways.
^ permalink raw reply [flat|nested] 10+ messages in thread
* [PATCH 8/7] provide select(2) backend for PublicInbox::DS
2023-09-12 2:34 ` Eric Wong
@ 2023-09-12 6:13 ` Eric Wong
0 siblings, 0 replies; 10+ messages in thread
From: Eric Wong @ 2023-09-12 6:13 UTC (permalink / raw)
To: meta
This is safer than relying on an internal API of IO::Poll
and doesn't create extra references to IO globs like the
public one.
---
Not sure if this is redundant or not, or if perhaps dropping
DSPoll would be preferable, even...
MANIFEST | 2 ++
lib/PublicInbox/DS.pm | 6 +++---
lib/PublicInbox/Select.pm | 40 +++++++++++++++++++++++++++++++++++++++
t/select.t | 4 ++++
4 files changed, 49 insertions(+), 3 deletions(-)
create mode 100644 lib/PublicInbox/Select.pm
create mode 100644 t/select.t
diff --git a/MANIFEST b/MANIFEST
index d7a670b8..63287bad 100644
--- a/MANIFEST
+++ b/MANIFEST
@@ -331,6 +331,7 @@ lib/PublicInbox/SearchIdxShard.pm
lib/PublicInbox/SearchQuery.pm
lib/PublicInbox/SearchThread.pm
lib/PublicInbox/SearchView.pm
+lib/PublicInbox/Select.pm
lib/PublicInbox/SharedKV.pm
lib/PublicInbox/Sigfd.pm
lib/PublicInbox/Smsg.pm
@@ -578,6 +579,7 @@ t/run.perl
t/search-amsg.eml
t/search-thr-index.t
t/search.t
+t/select.t
t/sha.t
t/shared_kv.t
t/sigfd.t
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index b3edc094..d47df491 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -31,7 +31,7 @@ use Scalar::Util qw(blessed);
use PublicInbox::Syscall qw(%SIGNUM
EPOLLIN EPOLLOUT EPOLLONESHOT EPOLLEXCLUSIVE);
use PublicInbox::Tmpfile;
-use PublicInbox::DSPoll;
+use PublicInbox::Select;
use Errno qw(EAGAIN EINVAL ECHILD EINTR);
use Carp qw(carp croak);
our @EXPORT_OK = qw(now msg_more awaitpid add_timer add_uniq_timer);
@@ -43,7 +43,7 @@ my $reap_armed;
my $ToClose; # sockets to close when event loop is done
our (
%DescriptorMap, # fd (num) -> PublicInbox::DS object
- $Poller, # global Epoll, DSPoll, or DSKQXS ref
+ $Poller, # global Select, Epoll, DSPoll, or DSKQXS ref
@post_loop_do, # subref + args to call at the end of each loop
@@ -83,7 +83,7 @@ sub Reset {
$reap_armed = undef;
$LoopTimeout = -1; # no timeout by default
- $Poller = PublicInbox::DSPoll->new;
+ $Poller = PublicInbox::Select->new;
}
=head2 C<< CLASS->SetLoopTimeout( $timeout ) >>
diff --git a/lib/PublicInbox/Select.pm b/lib/PublicInbox/Select.pm
new file mode 100644
index 00000000..9df3a6bd
--- /dev/null
+++ b/lib/PublicInbox/Select.pm
@@ -0,0 +1,40 @@
+# Copyright (C) all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+#
+# This makes select(2) look like epoll to simplify the code in DS.pm.
+# Unlike IO::Select, it does NOT hold references to IO handles.
+# This is NOT meant to be an all encompassing emulation of epoll
+# via select, but only to support cases we care about.
+package PublicInbox::Select;
+use v5.12;
+use PublicInbox::Syscall qw(EPOLLONESHOT EPOLLIN EPOLLOUT);
+
+sub new { bless {}, __PACKAGE__ } # fd => events
+
+sub ep_wait {
+ my ($self, $maxevents, $msec, $events) = @_;
+ my ($rvec, $wvec) = ('', ''); # we don't use EPOLLERR
+ while (my ($fd, $ev) = each %$self) {
+ vec($rvec, $fd, 1) = 1 if $ev & EPOLLIN;
+ vec($wvec, $fd, 1) = 1 if $ev & EPOLLOUT;
+ }
+ @$events = ();
+ my $n = select($rvec, $wvec, undef, $msec < 0 ? undef : ($msec/1000));
+ return if $n <= 0;
+ while (my ($fd, $ev) = each %$self) {
+ if (vec($rvec, $fd, 1) || vec($wvec, $fd, 1)) {
+ delete($self->{$fd}) if $ev & EPOLLONESHOT;
+ push @$events, $fd;
+ }
+ }
+ $n == scalar(@$events) or
+ warn "BUG? select() returned $n, but got ".scalar(@$events);
+}
+
+sub ep_del { delete($_[0]->{fileno($_[1])}); 0 }
+sub ep_add { $_[0]->{fileno($_[1])} = $_[2]; 0 }
+
+no warnings 'once';
+*ep_mod = \&ep_add;
+
+1;
diff --git a/t/select.t b/t/select.t
new file mode 100644
index 00000000..e8032c5a
--- /dev/null
+++ b/t/select.t
@@ -0,0 +1,4 @@
+# Copyright (C) all contributors <meta@public-inbox.org>
+use v5.12;
+local $ENV{TEST_IOPOLLER} = 'PublicInbox::Select';
+require './t/ds-poll.t';
^ permalink raw reply related [flat|nested] 10+ messages in thread
* [PATCH 6/7] ds: use constants for @UNBLOCKABLE list
2023-09-11 9:41 [PATCH 0/7] system-related updates and cleanups Eric Wong
` (4 preceding siblings ...)
2023-09-11 9:41 ` [PATCH 5/7] dspoll: switch to the documented IO::Poll API Eric Wong
@ 2023-09-11 9:41 ` Eric Wong
2023-09-11 9:41 ` [PATCH 7/7] spawn: do not block ABRT/BUS/ILL/SEGV signals Eric Wong
6 siblings, 0 replies; 10+ messages in thread
From: Eric Wong @ 2023-09-11 9:41 UTC (permalink / raw)
To: meta
There's no need for for a complicated map {} block here. All
these unblockable signals are POSIX since 2001 at the latest, so
there's no reason any platform would lack them.
---
lib/PublicInbox/DS.pm | 8 +++-----
1 file changed, 3 insertions(+), 5 deletions(-)
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index f2b14799..b3edc094 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -186,11 +186,9 @@ sub RunTimers {
sub sig_setmask { sigprocmask(SIG_SETMASK, @_) or die "sigprocmask: $!" }
-our @UNBLOCKABLE = map { # ensure we detect bugs, HW problems and user rlimits
- my $cb = POSIX->can("SIG$_");
- my $num = $cb ? $cb->() : undef;
- $num ? ($num) : ();
-} qw(ABRT BUS FPE ILL SEGV XCPU XFSZ);
+# ensure we detect bugs, HW problems and user rlimits
+our @UNBLOCKABLE = (POSIX::SIGABRT, POSIX::SIGBUS, POSIX::SIGFPE,
+ POSIX::SIGILL, POSIX::SIGSEGV, POSIX::SIGXCPU, POSIX::SIGXFSZ);
sub block_signals { # anything in @_ stays unblocked
my $newset = POSIX::SigSet->new;
^ permalink raw reply related [flat|nested] 10+ messages in thread* [PATCH 7/7] spawn: do not block ABRT/BUS/ILL/SEGV signals
2023-09-11 9:41 [PATCH 0/7] system-related updates and cleanups Eric Wong
` (5 preceding siblings ...)
2023-09-11 9:41 ` [PATCH 6/7] ds: use constants for @UNBLOCKABLE list Eric Wong
@ 2023-09-11 9:41 ` Eric Wong
6 siblings, 0 replies; 10+ messages in thread
From: Eric Wong @ 2023-09-11 9:41 UTC (permalink / raw)
To: meta
SIGABRT, SIGBUS, SIGILL, and SIGSEGV may all happen if we
introduce bugs in the section where signals are blocked.
We can delay handling of SIGFPE, SIGXCPU and SIGXFSZ since
there's no floating point operations; while SIGXCPU and
SIGXFSZ are safe to delay, especially in the absence of
threads in our current code paths.
---
lib/PublicInbox/Spawn.pm | 11 ++++++++---
lib/PublicInbox/SpawnPP.pm | 4 ++++
2 files changed, 12 insertions(+), 3 deletions(-)
diff --git a/lib/PublicInbox/Spawn.pm b/lib/PublicInbox/Spawn.pm
index 17d87f57..ed698afc 100644
--- a/lib/PublicInbox/Spawn.pm
+++ b/lib/PublicInbox/Spawn.pm
@@ -92,18 +92,23 @@ int pi_fork_exec(SV *redirref, SV *file, SV *cmdref, SV *envref, SV *rlimref,
sigset_t set, old;
int ret, perrnum;
volatile int cerrnum = 0; /* shared due to vfork */
- int chld_is_member;
+ int chld_is_member; /* needed due to shared memory w/ vfork */
I32 max_fd = av_len(redir);
AV2C_COPY(argv, cmd);
AV2C_COPY(envp, env);
if (sigfillset(&set)) return -1;
+ if (sigdelset(&set, SIGABRT)) return -1;
+ if (sigdelset(&set, SIGBUS)) return -1;
+ if (sigdelset(&set, SIGFPE)) return -1;
+ if (sigdelset(&set, SIGILL)) return -1;
+ if (sigdelset(&set, SIGSEGV)) return -1;
+ /* no XCPU/XFSZ here */
if (sigprocmask(SIG_SETMASK, &set, &old)) return -1;
chld_is_member = sigismember(&old, SIGCHLD);
if (chld_is_member < 0) return -1;
- if (chld_is_member > 0)
- sigdelset(&old, SIGCHLD);
+ if (chld_is_member > 0 && sigdelset(&old, SIGCHLD)) return -1;
pid = vfork();
if (pid == 0) {
diff --git a/lib/PublicInbox/SpawnPP.pm b/lib/PublicInbox/SpawnPP.pm
index d6c863f8..e7174d6f 100644
--- a/lib/PublicInbox/SpawnPP.pm
+++ b/lib/PublicInbox/SpawnPP.pm
@@ -15,6 +15,10 @@ sub pi_fork_exec ($$$$$$$) {
my $old = POSIX::SigSet->new();
my $set = POSIX::SigSet->new();
$set->fillset or die "sigfillset: $!";
+ for (POSIX::SIGABRT, POSIX::SIGBUS, POSIX::SIGFPE,
+ POSIX::SIGILL, POSIX::SIGSEGV) {
+ $set->delset($_) or die "delset($_): $!";
+ }
sigprocmask(SIG_SETMASK, $set, $old) or die "SIG_SETMASK(set): $!";
my $syserr;
pipe(my ($r, $w)) or die "pipe: $!";
^ permalink raw reply related [flat|nested] 10+ messages in thread