daemon: share and allow configuring Xapian helpers

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.

4 files changed, 38 insertions, 10 deletions

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);