]> Sergey Matveev's repositories - public-inbox.git/blobdiff - lib/PublicInbox/Daemon.pm
public-inbox-netd: a multi-protocol superserver
[public-inbox.git] / lib / PublicInbox / Daemon.pm
index 1520f8f20e563850fad496df741bdd64dc797cff..13cce1ec51abcf3499363e388b359db6bd7f4a1a 100644 (file)
@@ -1,25 +1,27 @@
-# Copyright (C) 2015-2020 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>
-# contains common daemon code for the httpd, imapd, and nntpd servers.
-# This may be used for read-only IMAP server if we decide to implement it.
+#
+# Contains common daemon code for the httpd, imapd, and nntpd servers
+# and designed for handling thousands of untrusted clients over slow
+# and/or lossy connections.
 package PublicInbox::Daemon;
 use strict;
-use warnings;
-use Getopt::Long qw/:config gnu_getopt no_ignore_case auto_abbrev/;
+use v5.10.1;
+use Getopt::Long qw(:config gnu_getopt no_ignore_case auto_abbrev);
 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);
-use PublicInbox::Syscall qw($SFD_NONBLOCK);
-require PublicInbox::Listener;
+use PublicInbox::Listener;
 use PublicInbox::EOFpipe;
 use PublicInbox::Sigfd;
+use PublicInbox::Git;
 use PublicInbox::GitAsyncCat;
+use PublicInbox::Eml;
+our $SO_ACCEPTFILTER = 0x1000;
 my @CMD;
 my ($set_user, $oldset);
 my (@cfg_listen, $stdout, $stderr, $group, $user, $pid_file, $daemonize);
@@ -73,18 +75,34 @@ sub accept_tls_opt ($) {
        { SSL_server => 1, SSL_startHandshake => 0, SSL_reuse_ctx => $ctx };
 }
 
-sub daemon_prepare ($) {
-       my ($default_listen) = @_;
+sub load_mod ($) {
+       my ($scheme) = @_;
+       my $modc = "PublicInbox::\U$1";
+       my $mod = $modc.'D';
+       eval "require $mod"; # IMAPD|HTTPD|NNTPD|POP3D
+       die $@ if $@;
+       my %xn = map { $_ => $mod->can($_) } qw(refresh post_accept);
+       $xn{tlsd} = $mod->new if $mod->can('refresh_groups'); #!HTTPD
+       my $tlsd = $xn{tlsd};
+       $xn{refresh} //= sub { $tlsd->refresh_groups(@_) };
+       $xn{post_accept} //= sub { $modc->new($_[0], $tlsd) };
+       $xn{af_default} = 'httpready' if $modc eq 'PublicInbox::HTTP';
+       \%xn;
+}
+
+sub daemon_prepare ($$) {
+       my ($default_listen, $xnetd) = @_;
        my $listener_names = {}; # sockname => IO::Handle
-       $oldset = PublicInbox::Sigfd::block_signals();
+       $oldset = PublicInbox::DS::block_signals();
        @CMD = ($0, @ARGV);
        my ($prog) = ($CMD[0] =~ m!([^/]+)\z!g);
+       my $dh = defined($default_listen) ? " (default: $default_listen)" : '';
        my $help = <<EOF;
 usage: $prog [-l ADDRESS] [--cert=FILE] [--key=FILE]
 
 options:
 
-  -l ADDRESS    address to listen on (default: $default_listen)
+  -l ADDRESS    address to listen on$dh
   --cert=FILE   default SSL/TLS certificate
   --key=FILE    default SSL/TLS certificate
   -W WORKERS    number of worker processes to spawn (default: 1)
@@ -119,7 +137,10 @@ EOF
        # ignore daemonize when inheriting
        $daemonize = undef if scalar @listeners;
 
-       push @cfg_listen, $default_listen unless (@listeners || @cfg_listen);
+       unless (@listeners || @cfg_listen) {
+               $default_listen // die "no listeners specified\n";
+               push @cfg_listen, $default_listen
+       }
 
        foreach my $l (@cfg_listen) {
                my $orig = $l;
@@ -137,7 +158,8 @@ EOF
                } elsif ($scheme =~ /\A(?:https|imaps|imaps)\z/) {
                        die "$orig specified w/o cert=\n";
                }
-               # TODO: use scheme to load either NNTP.pm or HTTP.pm
+               $scheme =~ /\A(http|imap|nntp|pop3)/ and
+                       $xnetd->{$l} = load_mod($1);
 
                next if $listener_names->{$l}; # already inherited
                my (%o, $sock_pkg);
@@ -202,10 +224,11 @@ sub check_absolute ($$) {
 
 sub daemonize () {
        if ($daemonize) {
+               require Cwd;
                foreach my $i (0..$#ARGV) {
                        my $arg = $ARGV[$i];
                        next unless -e $arg;
-                       $ARGV[$i] = abs_path($arg);
+                       $ARGV[$i] = Cwd::abs_path($arg);
                }
                check_absolute('stdout', $stdout);
                check_absolute('stderr', $stderr);
@@ -213,16 +236,12 @@ sub daemonize () {
 
                chdir '/' or die "chdir failed: $!";
        }
-
-       return unless (defined $pid_file || defined $group || defined $user
-                       || $daemonize);
-
-       eval { require Net::Server::Daemonize };
-       if ($@) {
-               die
-"Net::Server required for --pid-file, --group, --user, and --daemonize\n$@\n";
+       if (defined($pid_file) || defined($group) || defined($user)) {
+               eval { require Net::Server::Daemonize; 1 } // die <<EOF;
+Net::Server required for --pid-file, --group, --user
+$@
+EOF
        }
-
        Net::Server::Daemonize::check_pid_file($pid_file) if defined $pid_file;
        $uid = Net::Server::Daemonize::get_uid($user) if defined $user;
        if (defined $group) {
@@ -241,8 +260,7 @@ sub daemonize () {
        };
 
        if ($daemonize) {
-               my $pid = fork;
-               die "could not fork: $!\n" unless defined $pid;
+               my $pid = fork // die "fork: $!";
                exit if $pid;
 
                open(STDIN, '+<', '/dev/null') or
@@ -250,8 +268,7 @@ sub daemonize () {
                open STDOUT, '>&STDIN' or die "redirect stdout failed: $!\n";
                open STDERR, '>&STDIN' or die "redirect stderr failed: $!\n";
                POSIX::setsid();
-               $pid = fork;
-               die "could not fork: $!\n" unless defined $pid;
+               $pid = fork // die "fork: $!";
                exit if $pid;
        }
        return unless defined $pid_file;
@@ -274,13 +291,11 @@ sub worker_quit { # $_[0] = signal name or number (unused)
                my ($dmap, undef) = @_;
                my $n = 0;
                my $now = now();
-
-               foreach my $s (values %$dmap) {
+               for my $s (values %$dmap) {
                        $s->can('busy') or next;
-                       if ($s->busy($now)) {
+                       if ($s->busy) {
                                ++$n;
-                       } else {
-                               # close as much as possible, early as possible
+                       } else { # close as much as possible, early as possible
                                $s->close;
                        }
                }
@@ -371,16 +386,14 @@ sub inherit ($) {
        my $end = $fds + 2; # LISTEN_FDS_START - 1
        my @rv = ();
        foreach my $fd (3..$end) {
-               my $s = IO::Handle->new_from_fd($fd, 'r');
+               open(my $s, '<&=', $fd) or warn "fdopen fd=$fd: $!";
                if (my $k = sockname($s)) {
-                       if ($s->blocking) {
-                               $s->blocking(0);
-                               warn <<"";
+                       my $prev_was_blocking = $s->blocking(0);
+                       warn <<"" if $prev_was_blocking;
 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 {
@@ -427,11 +440,8 @@ sub upgrade { # $_[0] = signal name or number (unused)
 }
 
 sub kill_workers ($) {
-       my ($s) = @_;
-
-       while (my ($pid, $id) = each %pids) {
-               kill $s, $pid;
-       }
+       my ($sig) = @_;
+       kill $sig, keys(%pids);
 }
 
 sub upgrade_aborted ($) {
@@ -522,9 +532,9 @@ EOF
                },
                CHLD => \&reap_children,
        };
-       my $sigfd = PublicInbox::Sigfd->new($sig, 0);
-       local %SIG = (%SIG, %$sig) if !$sigfd;
-       PublicInbox::Sigfd::sig_setmask($oldset) if !$sigfd;
+       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) {
@@ -540,12 +550,15 @@ EOF
                }
                my $want = $worker_processes - 1;
                if ($n <= $want) {
-                       PublicInbox::Sigfd::block_signals() if !$sigfd;
+                       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 {
@@ -553,7 +566,7 @@ EOF
                                        $pids{$pid} = $i;
                                }
                        }
-                       PublicInbox::Sigfd::sig_setmask($oldset) if !$sigfd;
+                       PublicInbox::DS::sig_setmask($oldset) if !$sigfd;
                }
 
                if ($sigfd) { # Linux and IO::KQueue users:
@@ -585,21 +598,32 @@ sub defer_accept ($$) {
                return if $sec > 0; # systemd users may set a higher value
                setsockopt($s, IPPROTO_TCP, $TCP_DEFER_ACCEPT, 1);
        } elsif ($^O eq 'freebsd') {
-               my $x = getsockopt($s, SOL_SOCKET, SO_ACCEPTFILTER);
+               my $x = getsockopt($s, SOL_SOCKET, $SO_ACCEPTFILTER);
                return if defined $x; # don't change if set
                my $accf_arg = pack('a16a240', $af_name, '');
-               setsockopt($s, SOL_SOCKET, SO_ACCEPTFILTER, $accf_arg);
+               setsockopt($s, SOL_SOCKET, $SO_ACCEPTFILTER, $accf_arg);
        }
 }
 
-sub daemon_loop ($$$$) {
-       my ($refresh, $post_accept, $tlsd, $af_default) = @_;
+sub daemon_loop ($) {
+       my ($xnetd) = @_;
+       my $refresh = sub {
+               my ($sig) = @_;
+               for my $xn (values %$xnetd) {
+                       eval { $xn->{refresh}->($sig) };
+                       warn "refresh $@\n" if $@;
+               }
+       };
        my %post_accept;
        while (my ($k, $v) = each %tls_opt) {
-               if ($k =~ s!\A(?:https|imaps|nntps)://!!) {
-                       $post_accept{$k} = tls_start_cb($v, $post_accept);
-               } elsif ($tlsd) { # STARTTLS, $k eq '' is OK
-                       $tlsd->{accept_tls} = $v;
+               my $l = $k;
+               $l =~ s!\A([^:]+)://!!;
+               my $scheme = $1;
+               my $xn = $xnetd->{$l} // $xnetd->{''};
+               if ($scheme =~ s!\A(?:https|imaps|nntps)!!) {
+                       $post_accept{$l} = tls_start_cb($v, $xn->{post_accept});
+               } elsif ($xn->{tlsd}) { # STARTTLS, $k eq '' is OK
+                       $xn->{tlsd}->{accept_tls} = $v;
                }
        }
        my $sig = {
@@ -627,43 +651,37 @@ sub daemon_loop ($$$$) {
        $uid = $gid = undef;
        reopen_logs();
        @listeners = map {;
-               my $tls_cb = $post_accept{sockname($_)};
+               my $l = sockname($_);
+               my $tls_cb = $post_accept{$l};
+               my $xn = $xnetd->{$l} // $xnetd->{''};
 
                # NNTPS, HTTPS, HTTP, IMAPS and POP3S are client-first traffic
                # IMAP, NNTP and POP3 are server-first
-               defer_accept($_, $tls_cb ? 'dataready' : $af_default);
+               defer_accept($_, $tls_cb ? 'dataready' : $xn->{af_default});
 
                # this calls epoll_create:
-               PublicInbox::Listener->new($_, $tls_cb || $post_accept)
+               PublicInbox::Listener->new($_, $tls_cb || $xn->{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:
-               PublicInbox::Sigfd::sig_setmask($oldset);
-               PublicInbox::DS->SetLoopTimeout(1000);
-       }
-       PublicInbox::DS->EventLoop;
+       PublicInbox::DS::event_loop($sig, $oldset);
 }
 
-sub run ($$$;$) {
-       my ($default, $refresh, $post_accept, $tlsd) = @_;
-       local $SIG{PIPE} = 'IGNORE';
-       daemon_prepare($default);
-       my $af_default = $default =~ /:8080\z/ ? 'httpready' : undef;
+sub run {
+       my ($default_listen) = @_;
+       my $xnetd = {};
+       if ($default_listen) {
+               $default_listen =~ /\A(http|imap|nntp|pop3)/ or
+                       die "BUG: $default_listen";
+               $xnetd->{''} = load_mod($1);
+       }
+       daemon_prepare($default_listen, $xnetd);
        my $for_destroy = daemonize();
 
-       # this wastes a bit of memory for non-PublicInbox::WWW -httpd users
-       # oh well...
-       eval {
-               require PublicInbox::Gcf2;
-               require PublicInbox::Gcf2Client;
-       };
-       local $PublicInbox::GitAsyncCat::GCF2C =
-                               PublicInbox::Gcf2Client::new() if !$@;
+       # localize GCF2C for tests:
+       local $PublicInbox::GitAsyncCat::GCF2C;
+       local $PublicInbox::Git::async_warn = 1;
+       local $SIG{__WARN__} = PublicInbox::Eml::warn_ignore_cb();
 
-       daemon_loop($refresh, $post_accept, $tlsd, $af_default);
+       daemon_loop($xnetd);
        PublicInbox::DS->Reset;
        # ->DESTROY runs when $for_destroy goes out-of-scope
 }