[PATCH 0/2] send_cmd IPC improvements for busy sytems

[PATCH 0/2] send_cmd IPC improvements for busy sytems

[Eric Wong]

Making lei and admin tools fail less on overloaded systems...

Eric Wong (2):
  send_cmd: use (practically) infinite retries for writers
  send_cmd: throttle `sleeping on sendmsg' messages

 lib/PublicInbox/CmdIPC4.pm   |  7 ++++---
 lib/PublicInbox/IPC.pm       | 12 +++++++-----
 lib/PublicInbox/Spawn.pm     | 12 +++++++-----
 lib/PublicInbox/Syscall.pm   |  2 +-
 lib/PublicInbox/WQBlocked.pm | 11 +++++++++--
 lib/PublicInbox/XapClient.pm |  2 +-
 6 files changed, 29 insertions(+), 17 deletions(-)

[PATCH 1/2] send_cmd: use (practically) infinite retries for writers

[Eric Wong]

Write tools (-*index, -watch, -mda, lei) should never croak due
to the system being busy.  So make the retry infinite to benefit
users who run several parallel imports at once on a slower
system.  The previous 5s timeout was too close to failing in
my own experience using `lei import' on an old, busy machine.

For lei (inotify || EVFILT_VNODE) watches, we now retry on busy
sockets to avoid loss of FS change notifications.

On the contrary, public-facing read-only interfaces have always
been assumed to constantly be under attack.  Thus continuing to
drop requests due to a lack of kernel memory/buffers is probably
prudent.
---
 lib/PublicInbox/CmdIPC4.pm   |  4 ++--
 lib/PublicInbox/IPC.pm       | 12 +++++++-----
 lib/PublicInbox/Spawn.pm     |  8 ++++----
 lib/PublicInbox/Syscall.pm   |  2 +-
 lib/PublicInbox/WQBlocked.pm | 11 +++++++++--
 lib/PublicInbox/XapClient.pm |  2 +-
 6 files changed, 24 insertions(+), 15 deletions(-)

diff --git a/lib/PublicInbox/CmdIPC4.pm b/lib/PublicInbox/CmdIPC4.pm
index fc77bd03..bdf17a3f 100644
--- a/lib/PublicInbox/CmdIPC4.pm
+++ b/lib/PublicInbox/CmdIPC4.pm
@@ -11,7 +11,7 @@ use Socket qw(SOL_SOCKET SCM_RIGHTS);
 sub sendmsg_retry ($) {
 	return 1 if $!{EINTR};
 	return unless ($!{ENOMEM} || $!{ENOBUFS} || $!{ETOOMANYREFS});
-	return if --$_[0] < 0;
+	return if $_[0]-- == 0;
 	warn "# sleeping on sendmsg: $! ($_[0] tries left)\n";
 	select(undef, undef, undef, 0.1);
 	1;
@@ -24,7 +24,7 @@ no warnings 'once';
 # any number of FDs per-sendmsg(2) + buffer
 *send_cmd4 = sub ($$$$;$) { # (sock, fds, buf, flags) = @_;
 	my ($sock, $fds, undef, $flags, $tries) = @_;
-	$tries //= 50;
+	$tries //= -1; # infinite
 	my $mh = Socket::MsgHdr->new(buf => $_[2]);
 	$mh->cmsghdr(SOL_SOCKET, SCM_RIGHTS, pack('i' x scalar(@$fds), @$fds));
 	my $s;
diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
index 13a897be..806653dc 100644
--- a/lib/PublicInbox/IPC.pm
+++ b/lib/PublicInbox/IPC.pm
@@ -331,11 +331,13 @@ sub wq_nonblock_do { # always async
 	my $buf = ipc_freeze([$sub, @args]);
 	if ($self->{wqb}) { # saturated once, assume saturated forever
 		$self->{wqb}->flush_send($buf);
-	} else {
-		$send_cmd->($self->{-wq_s1}, [], $buf, 0) //
-			($!{EAGAIN} ? PublicInbox::WQBlocked->new($self, $buf)
-					: croak("sendmsg: $!"));
-	}
+	} elsif (!defined $send_cmd->($self->{-wq_s1}, [], $buf, 0)) {
+		if ($!{EAGAIN} || $!{ENOBUFS} || $!{ENOMEM}) {
+			PublicInbox::WQBlocked->new($self, $buf);
+		} else {
+			croak "sendmsg: $!";
+		}
+	} # else success
 }
 
 sub _wq_worker_start {
diff --git a/lib/PublicInbox/Spawn.pm b/lib/PublicInbox/Spawn.pm
index 19bedb10..1c6cc5be 100644
--- a/lib/PublicInbox/Spawn.pm
+++ b/lib/PublicInbox/Spawn.pm
@@ -176,15 +176,15 @@ out:
 	return (int)pid;
 }
 
-static int sendmsg_retry(int *tries)
+static int sendmsg_retry(long *tries)
 {
 	const struct timespec req = { 0, 100000000 }; /* 100ms */
 	int err = errno;
 	switch (err) {
 	case EINTR: PERL_ASYNC_CHECK(); return 1;
 	case ENOBUFS: case ENOMEM: case ETOOMANYREFS:
-		if (--*tries < 0) return 0;
-		fprintf(stderr, "# sleeping on sendmsg: %s (%d tries left)\n",
+		if (*tries-- == 0) return 0;
+		fprintf(stderr, "# sleeping on sendmsg: %s (%ld tries left)\n",
 			strerror(err), *tries);
 		nanosleep(&req, NULL);
 		PERL_ASYNC_CHECK();
@@ -201,7 +201,7 @@ union my_cmsg {
 	char pad[sizeof(struct cmsghdr) + 16 + SEND_FD_SPACE];
 };
 
-SV *send_cmd4_(PerlIO *s, SV *svfds, SV *data, int flags, int tries)
+SV *send_cmd4_(PerlIO *s, SV *svfds, SV *data, int flags, long tries)
 {
 	struct msghdr msg = { 0 };
 	union my_cmsg cmsg = { 0 };
diff --git a/lib/PublicInbox/Syscall.pm b/lib/PublicInbox/Syscall.pm
index ebcedb89..e5e3340f 100644
--- a/lib/PublicInbox/Syscall.pm
+++ b/lib/PublicInbox/Syscall.pm
@@ -502,7 +502,7 @@ require PublicInbox::CmdIPC4;
 			$msg_controllen,
 			0); # msg_flags
 	my $s;
-	$tries //= 50;
+	$tries //= -1;
 	do {
 		$s = syscall($SYS_sendmsg, fileno($sock), $mh, $flags);
 	} while ($s < 0 && PublicInbox::CmdIPC4::sendmsg_retry($tries));
diff --git a/lib/PublicInbox/WQBlocked.pm b/lib/PublicInbox/WQBlocked.pm
index 8d931fa9..b2463898 100644
--- a/lib/PublicInbox/WQBlocked.pm
+++ b/lib/PublicInbox/WQBlocked.pm
@@ -26,8 +26,15 @@ sub flush_send {
 			my $n = $PublicInbox::IPC::send_cmd->($wq_s1, [], $buf,
 								0);
 			next if defined($n);
-			Carp::croak("sendmsg: $!") unless $!{EAGAIN};
-			PublicInbox::DS::epwait($wq_s1, EPOLLOUT|EPOLLONESHOT);
+			if ($!{EAGAIN}) {
+				PublicInbox::DS::epwait($wq_s1,
+							EPOLLOUT|EPOLLONESHOT);
+			} elsif ($!{ENOBUFS} || $!{ENOMEM}) {
+				PublicInbox::DS::add_timer(0.1, \&flush_send,
+							$self);
+			} else {
+				Carp::croak("sendmsg: $!");
+			}
 			unshift @{$self->{msgq}}, $buf;
 			last; # wait for ->event_step
 		}
diff --git a/lib/PublicInbox/XapClient.pm b/lib/PublicInbox/XapClient.pm
index 24b3f45e..786982d8 100644
--- a/lib/PublicInbox/XapClient.pm
+++ b/lib/PublicInbox/XapClient.pm
@@ -12,7 +12,7 @@ use PublicInbox::Spawn qw(spawn);
 use Socket qw(AF_UNIX SOCK_SEQPACKET);
 use PublicInbox::IPC;
 use autodie qw(pipe socketpair);
-our $tries = 50;
+our $tries = -1; # set to zero by read-only daemon
 
 sub mkreq {
 	my ($self, $ios, @arg) = @_;

[PATCH 2/2] send_cmd: throttle `sleeping on sendmsg' messages

[Eric Wong]

Emitting a message every 100ms was too much for busy machines.
Throttle it to 1.6s to avoid flooding user terminals by emitting
every 16 sleeps.  A power-of-two (16) was chosen for its
optimization potential via bitwise AND.  Since perl(1) doesn't
do this optimization, we open code the bitwise AND.  I don't
assume an optimizing compiler for Inline::C, either, since I
find working in C far more enjoyable with optimizations off.
---
 lib/PublicInbox/CmdIPC4.pm | 3 ++-
 lib/PublicInbox/Spawn.pm   | 6 ++++--
 2 files changed, 6 insertions(+), 3 deletions(-)

diff --git a/lib/PublicInbox/CmdIPC4.pm b/lib/PublicInbox/CmdIPC4.pm
index bdf17a3f..af49f36d 100644
--- a/lib/PublicInbox/CmdIPC4.pm
+++ b/lib/PublicInbox/CmdIPC4.pm
@@ -12,7 +12,8 @@ sub sendmsg_retry ($) {
 	return 1 if $!{EINTR};
 	return unless ($!{ENOMEM} || $!{ENOBUFS} || $!{ETOOMANYREFS});
 	return if $_[0]-- == 0;
-	warn "# sleeping on sendmsg: $! ($_[0] tries left)\n";
+	# n.b. `N & (power-of-two - 1)' is a faster `N % power-of-two'
+	warn "# sleeping on sendmsg: $! ($_[0] tries left)\n" if !($_[0] & 15);
 	select(undef, undef, undef, 0.1);
 	1;
 }
diff --git a/lib/PublicInbox/Spawn.pm b/lib/PublicInbox/Spawn.pm
index 1c6cc5be..d818af0f 100644
--- a/lib/PublicInbox/Spawn.pm
+++ b/lib/PublicInbox/Spawn.pm
@@ -184,8 +184,10 @@ static int sendmsg_retry(long *tries)
 	case EINTR: PERL_ASYNC_CHECK(); return 1;
 	case ENOBUFS: case ENOMEM: case ETOOMANYREFS:
 		if (*tries-- == 0) return 0;
-		fprintf(stderr, "# sleeping on sendmsg: %s (%ld tries left)\n",
-			strerror(err), *tries);
+		if (!(*tries & 15))
+			fprintf(stderr,
+				"# sleeping on sendmsg: %s (%ld tries left)\n",
+				strerror(err), *tries);
 		nanosleep(&req, NULL);
 		PERL_ASYNC_CHECK();
 		return 1;

[PATCH 3/2] wqblocked: use per-instance unique timer

[Eric Wong]

It was possible to end up with multiple timers being armed on a
busy system from LeiNoteEvent due to the often rapid nature of
renames.  Since ->flush_send will flush all pending messages for
the WQ, it's only necessary to arm a single timer for a given WQ
socket.  This saves some memory and unnecessary wakeups by using
the handy unique timer mechanism in DS.
---
 lib/PublicInbox/WQBlocked.pm | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/PublicInbox/WQBlocked.pm b/lib/PublicInbox/WQBlocked.pm
index b2463898..3e4d67f2 100644
--- a/lib/PublicInbox/WQBlocked.pm
+++ b/lib/PublicInbox/WQBlocked.pm
@@ -30,8 +30,8 @@ sub flush_send {
 				PublicInbox::DS::epwait($wq_s1,
 							EPOLLOUT|EPOLLONESHOT);
 			} elsif ($!{ENOBUFS} || $!{ENOMEM}) {
-				PublicInbox::DS::add_timer(0.1, \&flush_send,
-							$self);
+				PublicInbox::DS::add_uniq_timer($self + 0,
+						0.1, \&flush_send, $self);
 			} else {
 				Carp::croak("sendmsg: $!");
 			}