[PATCH 0/5] xap_helper stuff for public daemons

[PATCH 0/5] xap_helper stuff for public daemons

[Eric Wong]

1 and 2 are trivial fixes.

3 and 4 makes failures more graceful when dealing with resource
exhaustion.

5 allows configuring the Xapian helper processes from the
top-level daemon and reverts to disabling helpers by default.

Eric Wong (5):
  t/cindex: require DBD::SQLite for now
  www: mbox*: use Perl 5.12
  send_cmd4: make `tries' a per-call parameter
  search: async_mset: pass resource errors to callback
  daemon: share and allow configuring Xapian helpers

 Documentation/public-inbox-daemon.pod | 38 +++++++++++++++++++++++++--
 Makefile.PL                           |  6 +++++
 lib/PublicInbox/CmdIPC4.pm            | 12 ++++-----
 lib/PublicInbox/Daemon.pm             | 24 +++++++++++++++--
 lib/PublicInbox/Mbox.pm               |  2 +-
 lib/PublicInbox/MboxGz.pm             |  2 +-
 lib/PublicInbox/Search.pm             | 16 ++++++-----
 lib/PublicInbox/Spawn.pm              | 12 +++++----
 lib/PublicInbox/Syscall.pm            |  8 +++---
 lib/PublicInbox/TestCommon.pm         |  9 ++++++-
 lib/PublicInbox/XapClient.pm          |  7 ++---
 t/cindex.t                            |  2 +-
 12 files changed, 106 insertions(+), 32 deletions(-)

[PATCH 1/5] t/cindex: require DBD::SQLite for now

[Eric Wong]

Technically it's not required, but -compact blindly requires
DBD::SQLite at the moment since it was designed for inboxes in
mind.  Furthermore, cindex isn't useful at the moment without
inboxes to associate with, and inboxes can't be indexed without
SQLite.
---
 t/cindex.t | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/t/cindex.t b/t/cindex.t
index 90236287..0ae0b2b4 100644
--- a/t/cindex.t
+++ b/t/cindex.t
@@ -6,7 +6,7 @@ use PublicInbox::TestCommon;
 use Cwd qw(getcwd);
 use List::Util qw(sum);
 use autodie qw(close mkdir open rename);
-require_mods(qw(json Xapian +SCM_RIGHTS));
+require_mods(qw(json Xapian +SCM_RIGHTS DBD::SQLite));
 use_ok 'PublicInbox::CodeSearchIdx';
 use PublicInbox::Import;
 my ($tmp, $for_destroy) = tmpdir();

[PATCH 2/5] www: mbox*: use Perl 5.12

[Eric Wong]

We were already silently relying on v5.10 features (`//') and
all the regexps to work correctly with v5.12 unicode_strings.
---
 lib/PublicInbox/Mbox.pm   | 2 +-
 lib/PublicInbox/MboxGz.pm | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/PublicInbox/Mbox.pm b/lib/PublicInbox/Mbox.pm
index 82fba5c6..17893a09 100644
--- a/lib/PublicInbox/Mbox.pm
+++ b/lib/PublicInbox/Mbox.pm
@@ -4,7 +4,7 @@
 # Streaming interface for mboxrd HTTP responses
 # See PublicInbox::GzipFilter for details.
 package PublicInbox::Mbox;
-use strict;
+use v5.12;
 use parent 'PublicInbox::GzipFilter';
 use PublicInbox::MID qw/mid_escape/;
 use PublicInbox::Hval qw/to_filename/;
diff --git a/lib/PublicInbox/MboxGz.pm b/lib/PublicInbox/MboxGz.pm
index 864d701e..90e69c09 100644
--- a/lib/PublicInbox/MboxGz.pm
+++ b/lib/PublicInbox/MboxGz.pm
@@ -1,7 +1,7 @@
 # Copyright (C) all contributors <meta@public-inbox.org>
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 package PublicInbox::MboxGz;
-use strict;
+use v5.12;
 use parent 'PublicInbox::GzipFilter';
 use PublicInbox::Eml;
 use PublicInbox::Hval qw/to_filename/;

[PATCH 3/5] send_cmd4: make `tries' a per-call parameter

[Eric Wong]

While existing callers are private (lei, *-index, -watch) are
private, we should not be blocking the event loop in
public-facing servers when we hit ETOOMANYREFS, ENOMEM, or
ENOBUFS.
---
 lib/PublicInbox/CmdIPC4.pm | 12 ++++++------
 lib/PublicInbox/Spawn.pm   | 12 +++++++-----
 lib/PublicInbox/Syscall.pm |  8 ++++----
 3 files changed, 17 insertions(+), 15 deletions(-)

diff --git a/lib/PublicInbox/CmdIPC4.pm b/lib/PublicInbox/CmdIPC4.pm
index 2f102ec6..fc77bd03 100644
--- a/lib/PublicInbox/CmdIPC4.pm
+++ b/lib/PublicInbox/CmdIPC4.pm
@@ -11,8 +11,8 @@ use Socket qw(SOL_SOCKET SCM_RIGHTS);
 sub sendmsg_retry ($) {
 	return 1 if $!{EINTR};
 	return unless ($!{ENOMEM} || $!{ENOBUFS} || $!{ETOOMANYREFS});
-	return if ++$_[0] >= 50;
-	warn "# sleeping on sendmsg: $! (#$_[0])\n";
+	return if --$_[0] < 0;
+	warn "# sleeping on sendmsg: $! ($_[0] tries left)\n";
 	select(undef, undef, undef, 0.1);
 	1;
 }
@@ -22,15 +22,15 @@ require Socket::MsgHdr; # XS
 no warnings 'once';
 
 # any number of FDs per-sendmsg(2) + buffer
-*send_cmd4 = sub ($$$$) { # (sock, fds, buf, flags) = @_;
-	my ($sock, $fds, undef, $flags) = @_;
+*send_cmd4 = sub ($$$$;$) { # (sock, fds, buf, flags) = @_;
+	my ($sock, $fds, undef, $flags, $tries) = @_;
+	$tries //= 50;
 	my $mh = Socket::MsgHdr->new(buf => $_[2]);
 	$mh->cmsghdr(SOL_SOCKET, SCM_RIGHTS, pack('i' x scalar(@$fds), @$fds));
 	my $s;
-	my $try = 0;
 	do {
 		$s = Socket::MsgHdr::sendmsg($sock, $mh, $flags);
-	} while (!defined($s) && sendmsg_retry($try));
+	} while (!defined($s) && sendmsg_retry($tries));
 	$s;
 };
 
diff --git a/lib/PublicInbox/Spawn.pm b/lib/PublicInbox/Spawn.pm
index e36659ce..e9e81e88 100644
--- a/lib/PublicInbox/Spawn.pm
+++ b/lib/PublicInbox/Spawn.pm
@@ -176,15 +176,15 @@ out:
 	return (int)pid;
 }
 
-static int sendmsg_retry(unsigned *tries)
+static int sendmsg_retry(int *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 >= 50) return 0;
-		fprintf(stderr, "# sleeping on sendmsg: %s (#%u)\n",
+		if (--*tries < 0) return 0;
+		fprintf(stderr, "# sleeping on sendmsg: %s (%d 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)
+SV *send_cmd4_(PerlIO *s, SV *svfds, SV *data, int flags, int tries)
 {
 	struct msghdr msg = { 0 };
 	union my_cmsg cmsg = { 0 };
@@ -211,7 +211,6 @@ SV *send_cmd4(PerlIO *s, SV *svfds, SV *data, int flags)
 	AV *fds = (AV *)SvRV(svfds);
 	I32 i, nfds = av_len(fds) + 1;
 	int *fdp;
-	unsigned tries = 0;
 
 	if (SvOK(data)) {
 		iov.iov_base = SvPV(data, dlen);
@@ -332,6 +331,9 @@ EOM
 	if (defined $all_libc) { # set for Gcf2
 		$ENV{PERL_INLINE_DIRECTORY} = $inline_dir;
 		%RLIMITS = rlimit_map();
+		*send_cmd4 = sub ($$$$;$) {
+			send_cmd4_($_[0], $_[1], $_[2], $_[3], 50);
+		}
 	} else {
 		require PublicInbox::SpawnPP;
 		*pi_fork_exec = \&PublicInbox::SpawnPP::pi_fork_exec
diff --git a/lib/PublicInbox/Syscall.pm b/lib/PublicInbox/Syscall.pm
index 99af5bf5..4cbe9623 100644
--- a/lib/PublicInbox/Syscall.pm
+++ b/lib/PublicInbox/Syscall.pm
@@ -467,8 +467,8 @@ if (defined($SYS_sendmsg) && defined($SYS_recvmsg)) {
 no warnings 'once';
 require PublicInbox::CmdIPC4;
 
-*send_cmd4 = sub ($$$$) {
-	my ($sock, $fds, undef, $flags) = @_;
+*send_cmd4 = sub ($$$$;$) {
+	my ($sock, $fds, undef, $flags, $tries) = @_;
 	my $iov = pack('P'.TMPL_size_t,
 			$_[2] // NUL, length($_[2] // NUL) || 1);
 	my $fd_space = scalar(@$fds) * SIZEOF_int;
@@ -487,10 +487,10 @@ require PublicInbox::CmdIPC4;
 			$msg_controllen,
 			0); # msg_flags
 	my $s;
-	my $try = 0;
+	$tries //= 50;
 	do {
 		$s = syscall($SYS_sendmsg, fileno($sock), $mh, $flags);
-	} while ($s < 0 && PublicInbox::CmdIPC4::sendmsg_retry($try));
+	} while ($s < 0 && PublicInbox::CmdIPC4::sendmsg_retry($tries));
 	$s >= 0 ? $s : undef;
 };
 

[PATCH 4/5] search: async_mset: pass resource errors to callback

[Eric Wong]

We need to be able to handle resource limitation errors in
public-facing daemons.
---
 lib/PublicInbox/Search.pm | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/lib/PublicInbox/Search.pm b/lib/PublicInbox/Search.pm
index 60d12dbf..b7732ae5 100644
--- a/lib/PublicInbox/Search.pm
+++ b/lib/PublicInbox/Search.pm
@@ -469,8 +469,12 @@ sub async_mset {
 	if ($XHC) { # unconditionally retrieving pct + rank for now
 		xdb($self); # populate {nshards}
 		my @margs = ($self->xh_args, xh_opt($opt));
-		my $rd = $XHC->mkreq(undef, 'mset', @margs, $qry_str);
-		PublicInbox::XhcMset->maybe_new($rd, $self, $cb, @args);
+		my $ret = eval {
+			my $rd = $XHC->mkreq(undef, 'mset', @margs, $qry_str);
+			PublicInbox::XhcMset->maybe_new($rd, $self, $cb, @args);
+		};
+		$cb->(@args, undef, $@) if $@;
+		$ret;
 	} else { # synchronous
 		my $mset = $self->mset($qry_str, $opt);
 		$cb->(@args, $mset);

[PATCH 5/5] daemon: share and allow configuring Xapian helpers

[Eric Wong]

Xapian helper processes are disabled by default once again.
However, they can be enabled via the new `-X INTEGER' parameter.
One big positive is the Xapian helpers being spawned by the
top-level daemon means they can be shared freely across all
workers for improved load balancing and memory reduction.
---
 Documentation/public-inbox-daemon.pod | 38 +++++++++++++++++++++++++--
 Makefile.PL                           |  6 +++++
 lib/PublicInbox/Daemon.pm             | 24 +++++++++++++++--
 lib/PublicInbox/Search.pm             |  8 +++---
 lib/PublicInbox/TestCommon.pm         |  9 ++++++-
 lib/PublicInbox/XapClient.pm          |  7 ++---
 6 files changed, 80 insertions(+), 12 deletions(-)

diff --git a/Documentation/public-inbox-daemon.pod b/Documentation/public-inbox-daemon.pod
index 6f1e3b53..092be667 100644
--- a/Documentation/public-inbox-daemon.pod
+++ b/Documentation/public-inbox-daemon.pod
@@ -79,9 +79,9 @@ C<err=> may also be specified on a per-listener basis.
 
 Default: /dev/null with C<--daemonize>, inherited otherwise
 
-=item -W
+=item -W INTEGER
 
-=item --worker-processes
+=item --worker-processes INTEGER
 
 Set the number of worker processes.
 
@@ -96,6 +96,40 @@ the master on crashes.
 
 Default: 1
 
+=item -X INTEGER
+
+=item --xapian-helpers INTEGER
+
+Enables the use of Xapian helper processes to handle expensive,
+non-deterministic Xapian search queries asynchronously without
+blocking simple requests.
+
+With positive values, there is an additional manager process
+that can be signaled to control the number of Xapian helper workers.
+
+* C<-X0> one worker, no manager process
+* C<-X1> one worker, one manager process
+...
+* C<-X8> eight workers, one manager process
+
+As with the public-facing public-inbox-* daemons, sending C<SIGTTIN>
+or C<SIGTTOU> to the Xapian helper manager process will increment or
+decrement the number of workers.
+
+Both Xapian helper workers and managers automatically respawn if they
+crash or are explicitly killed, even with C<-X0>.
+
+A C++ compiler, L<pkg-config(1)>, and Xapian development files (e.g.
+C<libxapian-dev> or C<xapian*-core-dev*>) are required to gain access to
+some expensive queries and significant memory savings.
+
+Xapian helper workers are shared by all C<--worker-processes> of the
+Perl daemon for additional memory savings.
+
+New in public-inbox 2.0.0.
+
+Default: undefined, search queries are handled synchronously
+
 =item --cert /path/to/cert
 
 The default TLS certificate for HTTPS, IMAPS, NNTPS, POP3S and/or STARTTLS
diff --git a/Makefile.PL b/Makefile.PL
index 2b2e6b18..27fe02ff 100644
--- a/Makefile.PL
+++ b/Makefile.PL
@@ -255,6 +255,12 @@ check-run : check-man
 # GNU and *BSD both allow it.
 check-run_T_ARGS = -j\$(N)
 
+check-xh0 :
+	\$(MAKE) check-run TEST_DAEMON_XH='-X0'
+
+check-xh1 :
+	\$(MAKE) check-run TEST_DAEMON_XH='-X1'
+
 check-debris check-run : pure_all
 	\$(EATMYDATA) \$(PROVE) -bvw xt/\$@.t :: \$(\$\@_T_ARGS)
 	-@\$(check_manifest)
diff --git a/lib/PublicInbox/Daemon.pm b/lib/PublicInbox/Daemon.pm
index ec76d6b8..e08102e9 100644
--- a/lib/PublicInbox/Daemon.pm
+++ b/lib/PublicInbox/Daemon.pm
@@ -22,9 +22,11 @@ use PublicInbox::GitAsyncCat;
 use PublicInbox::Eml;
 use PublicInbox::Config;
 use PublicInbox::OnDestroy;
+use PublicInbox::Search;
+use PublicInbox::XapClient;
 our $SO_ACCEPTFILTER = 0x1000;
 my @CMD;
-my ($set_user, $oldset);
+my ($set_user, $oldset, $xh_workers);
 my (@cfg_listen, $stdout, $stderr, $group, $user, $pid_file, $daemonize);
 my ($nworker, @listeners, %WORKERS, %logs);
 my %tls_opt; # scheme://sockname => args for IO::Socket::SSL::SSL_Context->new
@@ -170,6 +172,7 @@ options:
   --cert=FILE   default SSL/TLS certificate
   --key=FILE    default SSL/TLS certificate key
   -W WORKERS    number of worker processes to spawn (default: 1)
+  -X XWORKERS   number of Xapian helper processes (default: undefined)
 
 See public-inbox-daemon(8) and $prog(1) man pages for more.
 EOF
@@ -185,6 +188,7 @@ EOF
 		'multi-accept=i' => \$PublicInbox::Listener::MULTI_ACCEPT,
 		'cert=s' => \$default_cert,
 		'key=s' => \$default_key,
+		'X|xapian-helpers=i' => \$xh_workers,
 		'help|h' => \(my $show_help),
 	);
 	GetOptions(%opt) or die $help;
@@ -687,6 +691,14 @@ sub worker_loop {
 	PublicInbox::DS::event_loop(\%WORKER_SIG, $oldset);
 }
 
+sub respawn_xh { # awaitpid cb
+	my ($pid) = @_;
+	return unless @listeners;
+	warn "W: xap_helper PID:$pid died: \$?=$?, respawning...\n";
+	$PublicInbox::Search::XHC =
+		PublicInbox::XapClient::start_helper('-j', $xh_workers);
+}
+
 sub run {
 	my ($default_listen) = @_;
 	$nworker = 1;
@@ -699,7 +711,15 @@ sub run {
 	local $PublicInbox::Git::async_warn = 1;
 	local $SIG{__WARN__} = PublicInbox::Eml::warn_ignore_cb();
 	local %WORKER_SIG = %WORKER_SIG;
-	local %POST_ACCEPT;
+	local $PublicInbox::XapClient::tries = 0;
+
+	local $PublicInbox::Search::XHC = PublicInbox::XapClient::start_helper(
+			'-j', $xh_workers) if defined($xh_workers);
+	if ($PublicInbox::Search::XHC) {
+		require PublicInbox::XhcMset;
+		awaitpid($PublicInbox::Search::XHC->{io}->attached_pid,
+			\&respawn_xh);
+	}
 
 	daemon_loop();
 	# $unlink_on_leave runs
diff --git a/lib/PublicInbox/Search.pm b/lib/PublicInbox/Search.pm
index b7732ae5..4adef366 100644
--- a/lib/PublicInbox/Search.pm
+++ b/lib/PublicInbox/Search.pm
@@ -11,7 +11,7 @@ our @EXPORT_OK = qw(retry_reopen int_val get_pct xap_terms);
 use List::Util qw(max);
 use POSIX qw(strftime);
 use Carp ();
-our $XHC;
+our $XHC = 0; # defined but false
 
 # values for searching, changing the numeric value breaks
 # compatibility with old indices (so don't change them it)
@@ -57,7 +57,7 @@ use constant {
 };
 
 use PublicInbox::Smsg;
-use PublicInbox::Over;
+eval { require PublicInbox::Over };
 our $QP_FLAGS;
 our %X = map { $_ => 0 } qw(BoolWeight Database Enquire QueryParser Stem Query);
 our $Xap; # 'Xapian' or 'Search::Xapian'
@@ -428,9 +428,9 @@ sub mset {
 	do_enquire($self, $qry, $opt, TS);
 }
 
-sub xhc_start_maybe () {
+sub xhc_start_maybe (@) {
 	require PublicInbox::XapClient;
-	my $xhc = PublicInbox::XapClient::start_helper();
+	my $xhc = PublicInbox::XapClient::start_helper(@_);
 	require PublicInbox::XhcMset if $xhc;
 	$xhc;
 }
diff --git a/lib/PublicInbox/TestCommon.pm b/lib/PublicInbox/TestCommon.pm
index a7ec9b5b..b8b7b827 100644
--- a/lib/PublicInbox/TestCommon.pm
+++ b/lib/PublicInbox/TestCommon.pm
@@ -17,6 +17,7 @@ my $lei_loud = $ENV{TEST_LEI_ERR_LOUD};
 our $tail_cmd = $ENV{TAIL};
 our ($lei_opt, $lei_out, $lei_err);
 use autodie qw(chdir close fcntl mkdir open opendir seek unlink);
+$ENV{XDG_CACHE_HOME} //= "$ENV{HOME}/.cache"; # reuse C++ xap_helper builds
 
 $_ = File::Spec->rel2abs($_) for (grep(!m!^/!, @INC));
 
@@ -565,6 +566,9 @@ sub start_script {
 	my $run_mode = $ENV{TEST_RUN_MODE} // $opt->{run_mode} // 2;
 	my $sub = $run_mode == 0 ? undef : key2sub($key);
 	my $tail;
+	my $xh = $ENV{TEST_DAEMON_XH};
+	$xh && $key =~ /-(?:imapd|netd|httpd|pop3d|nntpd)\z/ and
+		push @argv, split(/\s+/, $xh);
 	if ($tail_cmd) {
 		my @paths;
 		for (@argv) {
@@ -720,7 +724,10 @@ SKIP: {
 	require PublicInbox::Spawn;
 	require PublicInbox::Config;
 	require File::Path;
-
+	eval { # use XDG_CACHE_HOME, first:
+		require PublicInbox::XapHelperCxx;
+		PublicInbox::XapHelperCxx::build();
+	};
 	local %ENV = %ENV;
 	delete $ENV{XDG_DATA_HOME};
 	delete $ENV{XDG_CONFIG_HOME};
diff --git a/lib/PublicInbox/XapClient.pm b/lib/PublicInbox/XapClient.pm
index f0270091..24b3f45e 100644
--- a/lib/PublicInbox/XapClient.pm
+++ b/lib/PublicInbox/XapClient.pm
@@ -12,6 +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;
 
 sub mkreq {
 	my ($self, $ios, @arg) = @_;
@@ -19,13 +20,13 @@ sub mkreq {
 	pipe($r, $ios->[0]) if !defined($ios->[0]);
 	my @fds = map fileno($_), @$ios;
 	my $buf = join("\0", @arg, '');
-	$n = $PublicInbox::IPC::send_cmd->($self->{io}, \@fds, $buf, 0) //
-		die "send_cmd: $!";
+	$n = $PublicInbox::IPC::send_cmd->($self->{io}, \@fds, $buf, 0, $tries)
+		// die "send_cmd: $!";
 	$n == length($buf) or die "send_cmd: $n != ".length($buf);
 	$r;
 }
 
-sub start_helper {
+sub start_helper (@) {
 	$PublicInbox::IPC::send_cmd or return; # can't work w/o SCM_RIGHTS
 	my @argv = @_;
 	socketpair(my $sock, my $in, AF_UNIX, SOCK_SEQPACKET, 0);