[PATCH 2/7] daemon: depend on DS event_loop in master process, too

[Eric Wong <e@80x24.org>]

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');