-# Copyright (C) 2015-2018 all contributors <meta@public-inbox.org>
+# Copyright (C) 2015-2019 all contributors <meta@public-inbox.org>
# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
# contains common daemon code for the nntpd and httpd servers.
# This may be used for read-only IMAP server if we decide to implement it.
use strict;
use warnings;
use Getopt::Long qw/:config gnu_getopt no_ignore_case auto_abbrev/;
-use IO::Handle;
+use IO::Handle; # ->autoflush
use IO::Socket;
+use POSIX qw(WNOHANG :signal_h);
use Socket qw(IPPROTO_TCP SOL_SOCKET);
sub SO_ACCEPTFILTER () { 0x1000 }
use Cwd qw/abs_path/;
STDOUT->autoflush(1);
STDERR->autoflush(1);
use PublicInbox::DS qw(now);
-require PublicInbox::EvCleanup;
-require POSIX;
+use PublicInbox::Syscall qw(SFD_NONBLOCK);
require PublicInbox::Listener;
require PublicInbox::ParentPipe;
+require PublicInbox::Sigfd;
my @CMD;
-my $set_user;
+my ($set_user, $oldset, $newset);
my (@cfg_listen, $stdout, $stderr, $group, $user, $pid_file, $daemonize);
my $worker_processes = 1;
my @listeners;
my %pids;
-my %listener_names; # sockname => IO::Handle
my %tls_opt; # scheme://sockname => args for IO::Socket::SSL->start_SSL
my $reexec_pid;
-my $cleanup;
my ($uid, $gid);
my ($default_cert, $default_key);
-END { $cleanup->() if $cleanup };
my %KNOWN_TLS = ( 443 => 'https', 563 => 'nntps' );
my %KNOWN_STARTTLS = ( 119 => 'nntp' );
{ SSL_server => 1, SSL_startHandshake => 0, SSL_reuse_ctx => $ctx };
}
+sub sig_setmask { sigprocmask(SIG_SETMASK, @_) or die "sigprocmask: $!" }
+
sub daemon_prepare ($) {
my ($default_listen) = @_;
+ my $listener_names = {}; # sockname => IO::Handle
+ $oldset = POSIX::SigSet->new();
+ $newset = POSIX::SigSet->new();
+ $newset->fillset or die "fillset: $!";
+ sig_setmask($newset, $oldset);
@CMD = ($0, @ARGV);
- $SIG{HUP} = $SIG{USR1} = $SIG{USR2} = $SIG{PIPE} =
- $SIG{TTIN} = $SIG{TTOU} = $SIG{WINCH} = 'IGNORE';
my %opts = (
'l|listen=s' => \@cfg_listen,
'1|stdout=s' => \$stdout,
if (defined $pid_file && $pid_file =~ /\.oldbin\z/) {
die "--pid-file cannot end with '.oldbin'\n";
}
- @listeners = inherit();
+ @listeners = inherit($listener_names);
# allow socket-activation users to set certs once and not
# have to configure each socket:
- my @inherited_names = keys(%listener_names) if defined($default_cert);
+ my @inherited_names = keys(%$listener_names) if defined($default_cert);
# ignore daemonize when inheriting
$daemonize = undef if scalar @listeners;
}
# TODO: use scheme to load either NNTP.pm or HTTP.pm
- next if $listener_names{$l}; # already inherited
+ next if $listener_names->{$l}; # already inherited
my (%o, $sock_pkg);
if (index($l, '/') == 0) {
$sock_pkg = 'IO::Socket::UNIX';
my $s = eval { $sock_pkg->new(%o) };
warn "error binding $l: $! ($@)\n" unless $s;
umask $prev;
-
if ($s) {
- $listener_names{sockname($s)} = $s;
+ $listener_names->{sockname($s)} = $s;
+ $s->blocking(0);
push @listeners, $s;
}
}
die "could not fork: $!\n" unless defined $pid;
exit if $pid;
}
- if (defined $pid_file) {
- write_pid($pid_file);
- my $unlink_pid = $$;
- $cleanup = sub {
- $cleanup = undef; # avoid cyclic reference
- unlink_pid_file_safe_ish($unlink_pid, $pid_file);
- };
- }
-}
+ return unless defined $pid_file;
+ write_pid($pid_file);
+ # for ->DESTROY:
+ bless { pid => $$, pid_file => $pid_file }, __PACKAGE__;
+}
-sub worker_quit {
+sub worker_quit { # $_[0] = signal name or number (unused)
# killing again terminates immediately:
exit unless @listeners;
$_->close foreach @listeners; # call PublicInbox::DS::close
@listeners = ();
-
my $proc_name;
my $warn = 0;
# drop idle connections and try to quit gracefully
$@ ? ('127.0.0.1', 0) : ($host, $port);
}
-sub inherit () {
+sub inherit ($) {
+ my ($listener_names) = @_;
return () if ($ENV{LISTEN_PID} || 0) != $$;
my $fds = $ENV{LISTEN_FDS} or return ();
my $end = $fds + 2; # LISTEN_FDS_START - 1
foreach my $fd (3..$end) {
my $s = IO::Handle->new_from_fd($fd, 'r');
if (my $k = sockname($s)) {
- $listener_names{$k} = $s;
+ if ($s->blocking) {
+ $s->blocking(0);
+ warn <<"";
+Inherited socket (fd=$fd) is blocking, making it non-blocking.
+Set 'NonBlocking = true' in the systemd.service unit to avoid stalled
+processes when multiple service instances start.
+
+ }
+ $listener_names->{$k} = $s;
push @rv, $s;
} else {
warn "failed to inherit fd=$fd (LISTEN_FDS=$fds)";
@rv
}
-sub upgrade () {
+sub upgrade { # $_[0] = signal name or number (unused)
if ($reexec_pid) {
warn "upgrade in-progress: $reexec_pid\n";
return;
warn $@, "\n" if $@;
}
-sub reap_children () {
+sub reap_children { # $_[0] = 'CHLD' or POSIX::SIGCHLD()
while (1) {
- my $p = waitpid(-1, &POSIX::WNOHANG) or return;
+ my $p = waitpid(-1, WNOHANG) or return;
if (defined $reexec_pid && $p == $reexec_pid) {
upgrade_aborted($p);
} elsif (defined(my $id = delete $pids{$p})) {
}
}
+sub master_quit ($) {
+ exit unless @listeners;
+ @listeners = ();
+ kill_workers($_[0]);
+}
+
sub master_loop {
pipe(my ($p0, $p1)) or die "failed to create parent-pipe: $!";
- pipe(my ($r, $w)) or die "failed to create self-pipe: $!";
-
- if ($^O eq 'linux') { # 1031: F_SETPIPE_SZ = 1031
- fcntl($_, 1031, 4096) for ($w, $p1);
- }
-
- IO::Handle::blocking($w, 0);
+ # 1031: F_SETPIPE_SZ, 4096: page size
+ fcntl($p1, 1031, 4096) if $^O eq 'linux';
my $set_workers = $worker_processes;
- my @caught;
- my $master_pid = $$;
- foreach my $s (qw(HUP CHLD QUIT INT TERM USR1 USR2 TTIN TTOU WINCH)) {
- $SIG{$s} = sub {
- return if $$ != $master_pid;
- push @caught, $s;
- syswrite($w, '.');
- };
- }
reopen_logs();
- # main loop
- my $quit = 0;
- while (1) {
- while (my $s = shift @caught) {
- if ($s eq 'USR1') {
- reopen_logs();
- kill_workers($s);
- } elsif ($s eq 'USR2') {
- upgrade();
- } elsif ($s =~ /\A(?:QUIT|TERM|INT)\z/) {
- exit if $quit++;
- kill_workers($s);
- } elsif ($s eq 'WINCH') {
- if (-t STDIN || -t STDOUT || -t STDERR) {
- warn
-"ignoring SIGWINCH since we are not daemonized\n";
- $SIG{WINCH} = 'IGNORE';
- } else {
- $worker_processes = 0;
- }
- } elsif ($s eq 'HUP') {
- $worker_processes = $set_workers;
- kill_workers($s);
- } elsif ($s eq 'TTIN') {
- if ($set_workers > $worker_processes) {
- ++$worker_processes;
- } else {
- $worker_processes = ++$set_workers;
- }
- } elsif ($s eq 'TTOU') {
- if ($set_workers > 0) {
- $worker_processes = --$set_workers;
- }
- } elsif ($s eq 'CHLD') {
- reap_children();
+ my $ignore_winch;
+ my $sig = {
+ 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;
}
- }
-
+ },
+ HUP => sub {
+ return unless @listeners;
+ $worker_processes = $set_workers;
+ kill_workers($_[0]);
+ },
+ TTIN => sub {
+ return unless @listeners;
+ if ($set_workers > $worker_processes) {
+ ++$worker_processes;
+ } else {
+ $worker_processes = ++$set_workers;
+ }
+ },
+ TTOU => sub {
+ $worker_processes = --$set_workers if $set_workers > 0;
+ },
+ CHLD => \&reap_children,
+ };
+ my $sigfd = PublicInbox::Sigfd->new($sig, 0);
+ local %SIG = (%SIG, %$sig) if !$sigfd;
+ sig_setmask($oldset) if !$sigfd;
+ while (1) { # main loop
my $n = scalar keys %pids;
- if ($quit) {
+ unless (@listeners) {
exit if $n == 0;
$set_workers = $worker_processes = $n = 0;
}
}
$n = $worker_processes;
}
- foreach my $i ($n..($worker_processes - 1)) {
- my $pid = fork;
- if (!defined $pid) {
- warn "failed to fork worker[$i]: $!\n";
- } elsif ($pid == 0) {
- $set_user->() if $set_user;
- return $p0; # run normal work code
- } else {
- warn "PID=$pid is worker[$i]\n";
- $pids{$pid} = $i;
+ my $want = $worker_processes - 1;
+ if ($n <= $want) {
+ sig_setmask($newset) if !$sigfd;
+ for my $i ($n..$want) {
+ my $pid = fork;
+ if (!defined $pid) {
+ warn "failed to fork worker[$i]: $!\n";
+ } elsif ($pid == 0) {
+ $set_user->() if $set_user;
+ return $p0; # run normal work code
+ } else {
+ warn "PID=$pid is worker[$i]\n";
+ $pids{$pid} = $i;
+ }
}
+ sig_setmask($oldset) if !$sigfd;
+ }
+
+ if ($sigfd) { # Linux and IO::KQueue users:
+ $sigfd->wait_once;
+ } else { # wake up every second
+ sleep(1);
}
- # just wait on signal events here:
- sysread($r, my $buf, 8);
}
exit # never gets here, just for documentation
}
sub daemon_loop ($$$$) {
my ($refresh, $post_accept, $nntpd, $af_default) = @_;
- PublicInbox::EvCleanup::enable(); # early for $refresh
my %post_accept;
while (my ($k, $v) = each %tls_opt) {
if ($k =~ s!\A(?:nntps|https)://!!) {
$nntpd->{accept_tls} = $v;
}
}
+ 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,
+ };
my $parent_pipe;
if ($worker_processes > 0) {
$refresh->(); # preload by default
} else {
reopen_logs();
$set_user->() if $set_user;
- $SIG{USR2} = sub { worker_quit() if upgrade() };
+ $sig->{USR2} = sub { worker_quit() if upgrade() };
$refresh->();
}
$uid = $gid = undef;
reopen_logs();
- $SIG{QUIT} = $SIG{INT} = $SIG{TERM} = *worker_quit;
- $SIG{USR1} = *reopen_logs;
- $SIG{HUP} = $refresh;
- $SIG{CHLD} = 'DEFAULT';
- $SIG{$_} = 'IGNORE' for qw(USR2 TTIN TTOU WINCH);
@listeners = map {;
my $tls_cb = $post_accept{sockname($_)};
# this calls epoll_create:
PublicInbox::Listener->new($_, $tls_cb || $post_accept)
} @listeners;
+ my $sigfd = PublicInbox::Sigfd->new($sig, SFD_NONBLOCK);
+ local %SIG = (%SIG, %$sig) if !$sigfd;
+ if (!$sigfd) {
+ # wake up every second to accept signals if we don't
+ # have signalfd or IO::KQueue:
+ sig_setmask($oldset);
+ PublicInbox::DS->SetLoopTimeout(1000);
+ }
PublicInbox::DS->EventLoop;
- $parent_pipe = undef;
}
-
sub run ($$$;$) {
my ($default, $refresh, $post_accept, $nntpd) = @_;
daemon_prepare($default);
my $af_default = $default =~ /:8080\z/ ? 'httpready' : undef;
- daemonize();
+ my $for_destroy = daemonize();
daemon_loop($refresh, $post_accept, $nntpd, $af_default);
+ PublicInbox::DS->Reset;
+ # ->DESTROY runs when $for_destroy goes out-of-scope
}
sub do_chown ($) {
do_chown($path);
}
+sub DESTROY {
+ unlink_pid_file_safe_ish($_[0]->{pid}, $_[0]->{pid_file});
+}
+
1;