X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FDaemon.pm;h=227ba5f979d09ef23eebaf892284b27137222165;hb=b6f480ed58abc5ae2a426ef4f792621b9d3cf283;hp=45c1563478a3c9adfa3ccd33de566d2bb0605c0b;hpb=1c0cfc8726c1c5e4085ad43bdf533b91af88e971;p=public-inbox.git diff --git a/lib/PublicInbox/Daemon.pm b/lib/PublicInbox/Daemon.pm index 45c15634..227ba5f9 100644 --- a/lib/PublicInbox/Daemon.pm +++ b/lib/PublicInbox/Daemon.pm @@ -1,19 +1,22 @@ -# Copyright (C) 2015 all contributors -# License: AGPLv3 or later (https://www.gnu.org/licenses/agpl-3.0.txt) -package PublicInbox::Daemon; # empty class :p - +# Copyright (C) 2015-2018 all contributors +# License: AGPL-3.0+ # 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. -package main; +package PublicInbox::Daemon; use strict; use warnings; use Getopt::Long qw/:config gnu_getopt no_ignore_case auto_abbrev/; use IO::Handle; +use IO::Socket; +use Cwd qw/abs_path/; +use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC); STDOUT->autoflush(1); STDERR->autoflush(1); -require Danga::Socket; +require PublicInbox::DS; +require PublicInbox::EvCleanup; require POSIX; require PublicInbox::Listener; +require PublicInbox::ParentPipe; my @CMD; my $set_user; my (@cfg_listen, $stdout, $stderr, $group, $user, $pid_file, $daemonize); @@ -54,30 +57,74 @@ sub daemon_prepare ($) { foreach my $l (@cfg_listen) { next if $listener_names{$l}; # already inherited - require IO::Socket::INET6; # works for IPv4, too - my %o = ( - LocalAddr => $l, - ReuseAddr => 1, - Proto => 'tcp', - ); - if (my $s = IO::Socket::INET6->new(%o)) { + my (%o, $sock_pkg); + if (index($l, '/') == 0) { + $sock_pkg = 'IO::Socket::UNIX'; + eval "use $sock_pkg"; + die $@ if $@; + %o = (Type => SOCK_STREAM, Peer => $l); + if (-S $l) { + my $c = $sock_pkg->new(%o); + if (!defined($c) && $!{ECONNREFUSED}) { + unlink $l or die +"failed to unlink stale socket=$l: $!\n"; + } # else: let the bind fail + } + $o{Local} = delete $o{Peer}; + } else { + # both work for IPv4, too + for (qw(IO::Socket::IP IO::Socket::INET6)) { + $sock_pkg = $_; + eval "use $sock_pkg"; + $@ or last; + } + die $@ if $@; + %o = (LocalAddr => $l, ReuseAddr => 1, Proto => 'tcp'); + } + $o{Listen} = 1024; + my $prev = umask 0000; + my $s = eval { $sock_pkg->new(%o) }; + warn "error binding $l: $! ($@)\n" unless $s; + umask $prev; + + if ($s) { $listener_names{sockname($s)} = $s; push @listeners, $s; - } else { - warn "error binding $l: $!\n"; } } die "No listeners bound\n" unless @listeners; } +sub check_absolute ($$) { + my ($var, $val) = @_; + if (defined $val && index($val, '/') != 0) { + die +"--$var must be an absolute path when using --daemonize: $val\n"; + } +} + sub daemonize () { - chdir '/' or die "chdir failed: $!"; - open(STDIN, '+<', '/dev/null') or die "redirect stdin failed: $!"; + if ($daemonize) { + foreach my $i (0..$#ARGV) { + my $arg = $ARGV[$i]; + next unless -e $arg; + $ARGV[$i] = abs_path($arg); + } + check_absolute('stdout', $stdout); + check_absolute('stderr', $stderr); + check_absolute('pid-file', $pid_file); + + chdir '/' or die "chdir failed: $!"; + } return unless (defined $pid_file || defined $group || defined $user || $daemonize); - require Net::Server::Daemonize; + eval { require Net::Server::Daemonize }; + if ($@) { + die +"Net::Server required for --pid-file, --group, --user, and --daemonize\n$@\n"; + } Net::Server::Daemonize::check_pid_file($pid_file) if defined $pid_file; $uid = Net::Server::Daemonize::get_uid($user) if defined $user; @@ -92,52 +139,72 @@ sub daemonize () { # The upgrade will create the ".oldbin" pid file in the # same directory as the given pid file. $uid and $set_user = sub { + $set_user = undef; Net::Server::Daemonize::set_user($uid, $gid); }; if ($daemonize) { - my ($pid, $err) = do_fork(); - die "could not fork: $err\n" unless defined $pid; + my $pid = fork; + die "could not fork: $!\n" unless defined $pid; exit if $pid; + open(STDIN, '+<', '/dev/null') or + die "redirect stdin failed: $!\n"; open STDOUT, '>&STDIN' or die "redirect stdout failed: $!\n"; open STDERR, '>&STDIN' or die "redirect stderr failed: $!\n"; POSIX::setsid(); - ($pid, $err) = do_fork(); - die "could not fork: $err\n" unless defined $pid; + $pid = fork; + 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); }; } } -sub worker_quit () { + +sub worker_quit { + my ($reason) = @_; # killing again terminates immediately: exit unless @listeners; + $_->close foreach @listeners; # call PublicInbox::DS::close @listeners = (); + $reason->close if ref($reason) eq 'PublicInbox::ParentPipe'; - # give slow clients 30s to finish reading/writing whatever - Danga::Socket->AddTimer(30, sub { exit }); - + my $proc_name; + my $warn = 0; # drop idle connections and try to quit gracefully - Danga::Socket->SetPostLoopCallback(sub { + PublicInbox::DS->SetPostLoopCallback(sub { my ($dmap, undef) = @_; my $n = 0; + my $now = clock_gettime(CLOCK_MONOTONIC); foreach my $s (values %$dmap) { - if ($s->can('busy') && $s->busy) { - $n = 1; + $s->can('busy') or next; + if ($s->busy($now)) { + ++$n; } else { # close as much as possible, early as possible $s->close; } } + if ($n) { + if (($warn + 5) < time) { + warn "$$ quitting, $n client(s) left\n"; + $warn = time; + } + unless (defined $proc_name) { + $proc_name = (split(/\s+/, $0))[0]; + $proc_name =~ s!\A.*?([^/]+)\z!$1!; + } + $0 = "$proc_name quitting, $n client(s) left"; + } $n; # true: loop continues, false: loop breaks }); } @@ -159,20 +226,52 @@ sub reopen_logs { sub sockname ($) { my ($s) = @_; - my $n = getsockname($s) or return; - my ($port, $addr); - if (length($n) >= 28) { - require Socket6; - ($port, $addr) = Socket6::unpack_sockaddr_in6($n); - } else { - ($port, $addr) = Socket::sockaddr_in($n); + my $addr = getsockname($s) or return; + my ($host, $port) = host_with_port($addr); + if ($port == 0 && $host eq '127.0.0.1') { + my ($path) = Socket::sockaddr_un($addr); + return $path; } - if (length($addr) == 4) { - $n = Socket::inet_ntoa($addr) - } else { - $n = '['.Socket6::inet_ntop(Socket6::AF_INET6(), $addr).']'; + "$host:$port"; +} + +sub unpack_ipv6 ($) { + my ($addr) = @_; + my ($port, $host); + + # Socket.pm in Perl 5.14+ supports IPv6: + eval { + ($port, $host) = Socket::unpack_sockaddr_in6($addr); + $host = Socket::inet_ntop(Socket::AF_INET6(), $host); + }; + + if ($@) { + # Perl 5.12 or earlier? SpamAssassin and Net::Server use + # Socket6, so it may be installed on our system, already + # (otherwise die here): + require Socket6; + + ($port, $host) = Socket6::unpack_sockaddr_in6($addr); + $host = Socket6::inet_ntop(Socket6::AF_INET6(), $host); } - $n .= ":$port"; + ($host, $port); +} + +sub host_with_port ($) { + my ($addr) = @_; + my ($port, $host); + + # this eval will die on Unix sockets: + eval { + if (length($addr) >= 28) { + ($host, $port) = unpack_ipv6($addr); + $host = "[$host]"; + } else { + ($port, $host) = Socket::sockaddr_in($addr); + $host = Socket::inet_ntoa($host); + } + }; + $@ ? ('127.0.0.1', 0) : ($host, $port); } sub inherit () { @@ -181,8 +280,7 @@ sub inherit () { my $end = $fds + 2; # LISTEN_FDS_START - 1 my @rv = (); foreach my $fd (3..$end) { - my $s = IO::Handle->new; - $s->fdopen($fd, 'r'); + my $s = IO::Handle->new_from_fd($fd, 'r'); if (my $k = sockname($s)) { $listener_names{$k} = $s; push @rv, $s; @@ -207,9 +305,9 @@ sub upgrade () { $pid_file .= '.oldbin'; write_pid($pid_file); } - my ($pid, $err) = do_fork(); + my $pid = fork; unless (defined $pid) { - warn "fork failed: $err\n"; + warn "fork failed: $!\n"; return; } if ($pid == 0) { @@ -234,17 +332,6 @@ sub kill_workers ($) { } } -sub do_fork () { - my $new = POSIX::SigSet->new; - $new->fillset; - my $old = POSIX::SigSet->new; - POSIX::sigprocmask(&POSIX::SIG_BLOCK, $new, $old) or die "SIG_BLOCK: $!"; - my $pid = fork; - my $err = $!; - POSIX::sigprocmask(&POSIX::SIG_SETMASK, $old) or die "SIG_SETMASK: $!"; - ($pid, $err); -} - sub upgrade_aborted ($) { my ($p) = @_; warn "reexec PID($p) died with: $?\n"; @@ -279,6 +366,7 @@ sub unlink_pid_file_safe_ish ($$) { return unless defined $unlink_pid && $unlink_pid == $$; open my $fh, '<', $file or return; + local $/ = "\n"; defined(my $read_pid = <$fh>) or return; chomp $read_pid; if ($read_pid == $unlink_pid) { @@ -289,6 +377,11 @@ sub unlink_pid_file_safe_ish ($$) { 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); my $set_workers = $worker_processes; my @caught; @@ -302,6 +395,7 @@ sub master_loop { } reopen_logs(); # main loop + my $quit = 0; while (1) { while (my $s = shift @caught) { if ($s eq 'USR1') { @@ -310,10 +404,16 @@ sub master_loop { } elsif ($s eq 'USR2') { upgrade(); } elsif ($s =~ /\A(?:QUIT|TERM|INT)\z/) { - # drops pipes and causes children to die - exit + exit if $quit++; + kill_workers($s); } elsif ($s eq 'WINCH') { - $worker_processes = 0; + 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); @@ -333,6 +433,11 @@ sub master_loop { } my $n = scalar keys %pids; + if ($quit) { + 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; @@ -340,9 +445,9 @@ sub master_loop { $n = $worker_processes; } foreach my $i ($n..($worker_processes - 1)) { - my ($pid, $err) = do_fork(); + my $pid = fork; if (!defined $pid) { - warn "failed to fork worker[$i]: $err\n"; + warn "failed to fork worker[$i]: $!\n"; } elsif ($pid == 0) { $set_user->() if $set_user; return $p0; # run normal work code @@ -359,16 +464,16 @@ sub master_loop { sub daemon_loop ($$) { my ($refresh, $post_accept) = @_; + PublicInbox::EvCleanup::enable(); # early for $refresh my $parent_pipe; if ($worker_processes > 0) { $refresh->(); # preload by default - $parent_pipe = master_loop(); # returns if in child process - my $fd = fileno($parent_pipe); - Danga::Socket->AddOtherFds($fd => sub { kill('TERM', $$) } ); + my $fh = master_loop(); # returns if in child process + $parent_pipe = PublicInbox::ParentPipe->new($fh, *worker_quit); } else { reopen_logs(); $set_user->() if $set_user; - $SIG{USR2} = sub { worker_quit() if upgrade() }; + $SIG{USR2} = sub { worker_quit('USR2') if upgrade() }; $refresh->(); } $uid = $gid = undef; @@ -376,14 +481,18 @@ sub daemon_loop ($$) { $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); # this calls epoll_create: - PublicInbox::Listener->new($_, $post_accept) for @listeners; - Danga::Socket->EventLoop; + @listeners = map { + PublicInbox::Listener->new($_, $post_accept) + } @listeners; + PublicInbox::DS->EventLoop; $parent_pipe = undef; } -sub daemon_run ($$$) { +sub run ($$$) { my ($default, $refresh, $post_accept) = @_; daemon_prepare($default); daemonize();