[PATCH] multi_accept redux

[Eric Wong <e@80x24.org>]

---
 Documentation/public-inbox-daemon.pod | 20 ++++++++++++
 lib/PublicInbox/Daemon.pm             |  7 ++--
 lib/PublicInbox/Listener.pm           | 46 +++++++++++++--------------
 3 files changed, 48 insertions(+), 25 deletions(-)

diff --git a/Documentation/public-inbox-daemon.pod b/Documentation/public-inbox-daemon.pod
index 81a79a10..6baef717 100644
--- a/Documentation/public-inbox-daemon.pod
+++ b/Documentation/public-inbox-daemon.pod
@@ -115,6 +115,26 @@ per-listener C<cert=> option.  The private key may be
 concatenated into the path used by the cert, in which case this
 option is not needed.
 
+=item --accept-nr INTEGER
+
+By default, each worker accepts one connection at-a-time to
+maximize fairness across multiple processes on a shared listen
+socket.  In deployments with one or few workers, using a
+positive value enables workers to accept the given number of
+client requests on each wakeup at the expense of fairness.
+
+Negative values allows workers to accept all available clients
+on each wakeup.  Thus C<-1> is similar to C<multi_accept yes> in
+nginx; while C<1> (the default) is similar to C<multi_accept no>
+in nginx.
+
+Using C<0> is synonmous with C<1>
+
+May be specifed on a per-listener basis via the C<accept-nr=>
+per-listener directive (e.g. C<-l 127.0.0.1?accept-nr=2>).
+
+Default: 1
+
 =back
 
 =head1 SIGNALS
diff --git a/lib/PublicInbox/Daemon.pm b/lib/PublicInbox/Daemon.pm
index 57435421..f24416e6 100644
--- a/lib/PublicInbox/Daemon.pm
+++ b/lib/PublicInbox/Daemon.pm
@@ -136,6 +136,7 @@ sub load_mod ($;$$) {
 	}
 	my $err = $tlsd->{err};
 	$tlsd->{warn_cb} = sub { print $err @_ }; # for local $SIG{__WARN__}
+	$xn{'accept-nr'} = $opt->{'accept-nr'}->[-1] if $opt->{'accept-nr'};
 	\%xn;
 }
 
@@ -167,6 +168,7 @@ EOF
 		'u|user=s' => \$user,
 		'g|group=s' => \$group,
 		'D|daemonize' => \$daemonize,
+		'accept-nr=i' => \$PublicInbox::Listener::ACCEPT_NR,
 		'cert=s' => \$default_cert,
 		'key=s' => \$default_key,
 		'help|h' => \(my $show_help),
@@ -251,7 +253,7 @@ EOF
 		$s->blocking(0);
 		my $sockname = sockname($s);
 		warn "# bound $scheme://$sockname\n";
-		$xnetd->{$sockname} //= load_mod($scheme);
+		$xnetd->{$sockname} //= load_mod($scheme, $opt);
 		$listener_names->{$sockname} = $s;
 		push @listeners, $s;
 	}
@@ -712,7 +714,8 @@ sub daemon_loop ($) {
 		defer_accept($_, $tls_cb ? 'dataready' : $xn->{af_default});
 
 		# this calls epoll_create:
-		PublicInbox::Listener->new($_, $tls_cb || $xn->{post_accept})
+		PublicInbox::Listener->new($_, $tls_cb || $xn->{post_accept},
+						$xn->{'accept-nr'})
 	} @listeners;
 	PublicInbox::DS::event_loop($sig, $oldset);
 }
diff --git a/lib/PublicInbox/Listener.pm b/lib/PublicInbox/Listener.pm
index 7cedc349..d171449f 100644
--- a/lib/PublicInbox/Listener.pm
+++ b/lib/PublicInbox/Listener.pm
@@ -1,14 +1,15 @@
-# Copyright (C) 2015-2021 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>
 #
 # Used by -nntpd for listen sockets
 package PublicInbox::Listener;
-use strict;
+use v5.12;
 use parent 'PublicInbox::DS';
 use Socket qw(SOL_SOCKET SO_KEEPALIVE IPPROTO_TCP TCP_NODELAY);
 use IO::Handle;
 use PublicInbox::Syscall qw(EPOLLIN EPOLLEXCLUSIVE);
 use Errno qw(EAGAIN ECONNABORTED);
+our $ACCEPT_NR = 1;
 
 # Warn on transient errors, mostly resource limitations.
 # EINTR would indicate the failure to set NonBlocking in systemd or similar
@@ -16,37 +17,36 @@ my %ERR_WARN = map {;
 	eval("Errno::$_()") => $_
 } qw(EMFILE ENFILE ENOBUFS ENOMEM EINTR);
 
-sub new ($$$) {
-	my ($class, $s, $cb) = @_;
+sub new {
+	my ($class, $s, $cb, $accept_nr) = @_;
 	setsockopt($s, SOL_SOCKET, SO_KEEPALIVE, 1);
 	setsockopt($s, IPPROTO_TCP, TCP_NODELAY, 1); # ignore errors on non-TCP
 	listen($s, 2**31 - 1); # kernel will clamp
 	my $self = bless { post_accept => $cb }, $class;
+	$self->{accept_nr} = $accept_nr //= $ACCEPT_NR;
+	$self->{accept_nr} = 1 if !$accept_nr;
 	$self->SUPER::new($s, EPOLLIN|EPOLLEXCLUSIVE);
 }
 
 sub event_step {
 	my ($self) = @_;
 	my $sock = $self->{sock} or return;
-
-	# no loop here, we want to fairly distribute clients
-	# between multiple processes sharing the same socket
-	# XXX our event loop needs better granularity for
-	# a single accept() here to be, umm..., acceptable
-	# on high-traffic sites.
-	if (my $addr = accept(my $c, $sock)) {
-		IO::Handle::blocking($c, 0); # no accept4 :<
-		eval { $self->{post_accept}->($c, $addr, $sock) };
-		warn "E: $@\n" if $@;
-	} elsif ($! == EAGAIN || $! == ECONNABORTED) {
-		# EAGAIN is common and likely
-		# ECONNABORTED is common with bad connections
-		return;
-	} elsif (my $sym = $ERR_WARN{int($!)}) {
-		warn "W: accept(): $! ($sym)\n";
-	} else {
-		warn "BUG?: accept(): $!\n";
-	}
+	my $n = $self->{accept_nr};
+	do {
+		if (my $addr = accept(my $c, $sock)) {
+			IO::Handle::blocking($c, 0); # no accept4 :<
+			eval { $self->{post_accept}->($c, $addr, $sock) };
+			warn "E: $@\n" if $@;
+		} elsif ($! == EAGAIN || $! == ECONNABORTED) {
+			# EAGAIN is common and likely
+			# ECONNABORTED is common with bad connections
+			return;
+		} elsif (my $sym = $ERR_WARN{int($!)}) {
+			warn "W: accept(): $! ($sym)\n";
+		} else {
+			warn "BUG?: accept(): $!\n";
+		}
+	} while ($n--);
 }
 
 1;