X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FDaemon.pm;h=b3743f5c3ddf80d608c7becf4bc48d242f04eb88;hb=9bd675d33ad1e49bd2ebe12a1d216216e61380de;hp=c101ecb7ad538313324879c4986587ca30fee0d8;hpb=b38de6f02fa04e36b881d2aad9c7f792beb0b6a1;p=public-inbox.git

diff --git a/lib/PublicInbox/Daemon.pm b/lib/PublicInbox/Daemon.pm
index c101ecb7..b3743f5c 100644
--- a/lib/PublicInbox/Daemon.pm
+++ b/lib/PublicInbox/Daemon.pm
@@ -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-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.
 package PublicInbox::Daemon;
@@ -7,22 +7,72 @@ use strict;
 use warnings;
 use Getopt::Long qw/:config gnu_getopt no_ignore_case auto_abbrev/;
 use IO::Handle;
+use IO::Socket;
+use POSIX qw(WNOHANG);
+use Socket qw(IPPROTO_TCP SOL_SOCKET);
+sub SO_ACCEPTFILTER () { 0x1000 }
+use Cwd qw/abs_path/;
 STDOUT->autoflush(1);
 STDERR->autoflush(1);
-require Danga::Socket;
-require POSIX;
+use PublicInbox::DS qw(now);
+require PublicInbox::EvCleanup;
 require PublicInbox::Listener;
+require PublicInbox::ParentPipe;
 my @CMD;
 my $set_user;
 my (@cfg_listen, $stdout, $stderr, $group, $user, $pid_file, $daemonize);
 my $worker_processes = 1;
 my @listeners;
 my %pids;
-my %listener_names;
+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' );
+
+sub accept_tls_opt ($) {
+	my ($opt_str) = @_;
+	# opt_str: opt1=val1,opt2=val2 (opt may repeat for multi-value)
+	require PublicInbox::TLS;
+	my $o = {};
+	# allow ',' as delimiter since '&' is shell-unfriendly
+	foreach (split(/[,&]/, $opt_str)) {
+		my ($k, $v) = split(/=/, $_, 2);
+		push @{$o->{$k} ||= []}, $v;
+	}
+
+	# key may be a part of cert.  At least
+	# p5-io-socket-ssl/example/ssl_server.pl has this fallback:
+	$o->{cert} //= [ $default_cert ];
+	$o->{key} //= defined($default_key) ? [ $default_key ] : $o->{cert};
+	my %ctx_opt = (SSL_server => 1);
+	# parse out hostname:/path/to/ mappings:
+	foreach my $k (qw(cert key)) {
+		my $x = $ctx_opt{'SSL_'.$k.'_file'} = {};
+		foreach my $path (@{$o->{$k}}) {
+			my $host = '';
+			$path =~ s/\A([^:]+):// and $host = $1;
+			$x->{$host} = $path;
+		}
+	}
+	my $ctx = IO::Socket::SSL::SSL_Context->new(%ctx_opt) or
+		die 'SSL_Context->new: '.PublicInbox::TLS::err();
+
+	# save ~34K per idle connection (cf. SSL_CTX_set_mode(3ssl))
+	# RSS goes from 346MB to 171MB with 10K idle NNTPS clients on amd64
+	# cf. https://rt.cpan.org/Ticket/Display.html?id=129463
+	my $mode = eval { Net::SSLeay::MODE_RELEASE_BUFFERS() };
+	if ($mode && $ctx->{context}) {
+		eval { Net::SSLeay::CTX_set_mode($ctx->{context}, $mode) };
+		warn "W: $@ (setting SSL_MODE_RELEASE_BUFFERS)\n" if $@;
+	}
+
+	{ SSL_server => 1, SSL_startHandshake => 0, SSL_reuse_ctx => $ctx };
+}
 
 sub daemon_prepare ($) {
 	my ($default_listen) = @_;
@@ -38,6 +88,8 @@ sub daemon_prepare ($) {
 		'u|user=s' => \$user,
 		'g|group=s' => \$group,
 		'D|daemonize' => \$daemonize,
+		'cert=s' => \$default_cert,
+		'key=s' => \$default_key,
 	);
 	GetOptions(%opts) or die "bad command-line args\n";
 
@@ -45,37 +97,117 @@ sub daemon_prepare ($) {
 		die "--pid-file cannot end with '.oldbin'\n";
 	}
 	@listeners = inherit();
+
+	# 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);
+
 	# ignore daemonize when inheriting
 	$daemonize = undef if scalar @listeners;
 
 	push @cfg_listen, $default_listen unless (@listeners || @cfg_listen);
 
 	foreach my $l (@cfg_listen) {
+		my $orig = $l;
+		my $scheme = '';
+		if ($l =~ s!\A([^:]+)://!!) {
+			$scheme = $1;
+		} elsif ($l =~ /\A(?:\[[^\]]+\]|[^:]+):([0-9])+/) {
+			my $s = $KNOWN_TLS{$1} // $KNOWN_STARTTLS{$1};
+			$scheme = $s if defined $s;
+		}
+		if ($l =~ s!/?\?(.+)\z!!) {
+			$tls_opt{"$scheme://$l"} = accept_tls_opt($1);
+		} elsif (defined($default_cert)) {
+			$tls_opt{"$scheme://$l"} = accept_tls_opt('');
+		} elsif ($scheme =~ /\A(?:nntps|https)\z/) {
+			die "$orig specified w/o cert=\n";
+		}
+		# TODO: use scheme to load either NNTP.pm or HTTP.pm
+
 		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;
+			$s->blocking(0);
 			push @listeners, $s;
-		} else {
-			warn "error binding $l: $!\n";
 		}
 	}
+
+	# cert/key options in @cfg_listen takes precedence when inheriting,
+	# but map well-known inherited ports if --listen isn't specified
+	# at all
+	for my $sockname (@inherited_names) {
+		$sockname =~ /:([0-9]+)\z/ or next;
+		if (my $scheme = $KNOWN_TLS{$1}) {
+			$tls_opt{"$scheme://$sockname"} ||= accept_tls_opt('');
+		} elsif (($scheme = $KNOWN_STARTTLS{$1})) {
+			next if $tls_opt{"$scheme://$sockname"};
+			$tls_opt{''} ||= accept_tls_opt('');
+		}
+	}
+
 	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;
@@ -90,52 +222,70 @@ 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 {
 	# killing again terminates immediately:
 	exit unless @listeners;
 
+	$_->close foreach @listeners; # call PublicInbox::DS::close
 	@listeners = ();
 
-	# 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 = now();
 
 		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) < now()) {
+				warn "$$ quitting, $n client(s) left\n";
+				$warn = now();
+			}
+			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,32 +309,68 @@ 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 host_with_port ($) {
+sub unpack_ipv6 ($) {
 	my ($addr) = @_;
 	my ($port, $host);
-	if (length($addr) >= 28) {
+
+	# 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).']';
-	} else {
-		($port, $host) = Socket::sockaddr_in($addr);
-		$host = Socket::inet_ntoa($host);
+		$host = Socket6::inet_ntop(Socket6::AF_INET6(), $host);
 	}
 	($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 () {
 	return () if ($ENV{LISTEN_PID} || 0) != $$;
 	my $fds = $ENV{LISTEN_FDS} or return ();
 	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)) {
+			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 {
@@ -208,9 +394,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) {
@@ -235,17 +421,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";
@@ -262,7 +437,7 @@ sub upgrade_aborted ($) {
 
 sub reap_children () {
 	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})) {
@@ -280,6 +455,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) {
@@ -290,6 +466,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;
@@ -303,6 +484,7 @@ sub master_loop {
 	}
 	reopen_logs();
 	# main loop
+	my $quit = 0;
 	while (1) {
 		while (my $s = shift @caught) {
 			if ($s eq 'USR1') {
@@ -311,10 +493,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);
@@ -334,6 +522,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;
@@ -341,9 +534,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
@@ -358,14 +551,48 @@ sub master_loop {
 	exit # never gets here, just for documentation
 }
 
-sub daemon_loop ($$) {
-	my ($refresh, $post_accept) = @_;
+sub tls_start_cb ($$) {
+	my ($opt, $orig_post_accept) = @_;
+	sub {
+		my ($io, $addr, $srv) = @_;
+		my $ssl = IO::Socket::SSL->start_SSL($io, %$opt);
+		$orig_post_accept->($ssl, $addr, $srv);
+	}
+}
+
+sub defer_accept ($$) {
+	my ($s, $af_name) = @_;
+	return unless defined $af_name;
+	if ($^O eq 'linux') {
+		my $x = getsockopt($s, IPPROTO_TCP, Socket::TCP_DEFER_ACCEPT());
+		return unless defined $x; # may be Unix socket
+		my $sec = unpack('i', $x);
+		return if $sec > 0; # systemd users may set a higher value
+		setsockopt($s, IPPROTO_TCP, Socket::TCP_DEFER_ACCEPT(), 1);
+	} elsif ($^O eq 'freebsd') {
+		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);
+	}
+}
+
+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)://!!) {
+			$post_accept{$k} = tls_start_cb($v, $post_accept);
+		} elsif ($nntpd) { # STARTTLS, $k eq '' is OK
+			$nntpd->{accept_tls} = $v;
+		}
+	}
 	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;
@@ -377,18 +604,29 @@ sub daemon_loop ($$) {
 	$SIG{QUIT} = $SIG{INT} = $SIG{TERM} = *worker_quit;
 	$SIG{USR1} = *reopen_logs;
 	$SIG{HUP} = $refresh;
-	# this calls epoll_create:
-	PublicInbox::Listener->new($_, $post_accept) for @listeners;
-	Danga::Socket->EventLoop;
+	$SIG{CHLD} = 'DEFAULT';
+	$SIG{$_} = 'IGNORE' for qw(USR2 TTIN TTOU WINCH);
+	@listeners = map {;
+		my $tls_cb = $post_accept{sockname($_)};
+
+		# NNTPS, HTTPS, HTTP, and POP3S are client-first traffic
+		# NNTP and POP3 are server-first
+		defer_accept($_, $tls_cb ? 'dataready' : $af_default);
+
+		# this calls epoll_create:
+		PublicInbox::Listener->new($_, $tls_cb || $post_accept)
+	} @listeners;
+	PublicInbox::DS->EventLoop;
 	$parent_pipe = undef;
 }
 
 
-sub run ($$$) {
-	my ($default, $refresh, $post_accept) = @_;
+sub run ($$$;$) {
+	my ($default, $refresh, $post_accept, $nntpd) = @_;
 	daemon_prepare($default);
+	my $af_default = $default =~ /:8080\z/ ? 'httpready' : undef;
 	daemonize();
-	daemon_loop($refresh, $post_accept);
+	daemon_loop($refresh, $post_accept, $nntpd, $af_default);
 }
 
 sub do_chown ($) {