]> Sergey Matveev's repositories - public-inbox.git/blobdiff - lib/PublicInbox/Daemon.pm
remove hard Devel::Peek dependency and lazy load for daemons
[public-inbox.git] / lib / PublicInbox / Daemon.pm
index c3199cd24f0a9859261408fe479edf535b916544..227ba5f979d09ef23eebaf892284b27137222165 100644 (file)
@@ -1,5 +1,5 @@
-# Copyright (C) 2015 all contributors <meta@public-inbox.org>
-# License: AGPLv3 or later (https://www.gnu.org/licenses/agpl-3.0.txt)
+# Copyright (C) 2015-2018 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.
 package PublicInbox::Daemon;
@@ -8,11 +8,15 @@ 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);
@@ -68,15 +72,19 @@ sub daemon_prepare ($) {
                        }
                        $o{Local} = delete $o{Peer};
                } else {
-                       $sock_pkg = 'IO::Socket::INET6'; # works for IPv4, too
-                       eval "use $sock_pkg";
+                       # 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;
+               warn "error binding $l: $! ($@)\n" unless $s;
                umask $prev;
 
                if ($s) {
@@ -87,14 +95,36 @@ sub daemon_prepare ($) {
        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;
@@ -109,53 +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 Danga::Socket::close
+       $_->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
        });
 }
@@ -179,9 +228,35 @@ sub sockname ($) {
        my ($s) = @_;
        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;
+       }
        "$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);
+       }
+       ($host, $port);
+}
+
 sub host_with_port ($) {
        my ($addr) = @_;
        my ($port, $host);
@@ -189,9 +264,7 @@ sub host_with_port ($) {
        # this eval will die on Unix sockets:
        eval {
                if (length($addr) >= 28) {
-                       require Socket6;
-                       ($port, $host) = Socket6::unpack_sockaddr_in6($addr);
-                       $host = Socket6::inet_ntop(Socket6::AF_INET6(), $host);
+                       ($host, $port) = unpack_ipv6($addr);
                        $host = "[$host]";
                } else {
                        ($port, $host) = Socket::sockaddr_in($addr);
@@ -232,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) {
@@ -259,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";
@@ -304,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) {
@@ -314,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;
@@ -327,6 +395,7 @@ sub master_loop {
        }
        reopen_logs();
        # main loop
+       my $quit = 0;
        while (1) {
                while (my $s = shift @caught) {
                        if ($s eq 'USR1') {
@@ -335,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);
@@ -358,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;
@@ -365,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
@@ -384,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;
@@ -401,11 +481,13 @@ 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:
        @listeners = map {
                PublicInbox::Listener->new($_, $post_accept)
        } @listeners;
-       Danga::Socket->EventLoop;
+       PublicInbox::DS->EventLoop;
        $parent_pipe = undef;
 }