[PATCH 1/4] xap_helper: expire DB handles when FD table is near full

[PATCH 1/4] xap_helper: expire DB handles when FD table is near full

[Eric Wong]

For long-lived daemons across config reloads, we shouldn't keep
Xapian DBs open forever under FD pressure.  So estimate the
number of FDs we need per-shard and start clearing some out
if we have too many open.

While we're at it, hoist out our ulimit_n helper and share it
across extindex and the Perl XapHelper implementation.
---
 lib/PublicInbox/ExtSearchIdx.pm |  8 +----
 lib/PublicInbox/Search.pm       | 16 +++++++++
 lib/PublicInbox/XapHelper.pm    | 18 +++++++---
 lib/PublicInbox/XapHelperCxx.pm |  1 +
 lib/PublicInbox/xap_helper.h    | 58 +++++++++++++++++++++++++--------
 t/xap_helper.t                  | 23 +++++++++++++
 6 files changed, 100 insertions(+), 24 deletions(-)

diff --git a/lib/PublicInbox/ExtSearchIdx.pm b/lib/PublicInbox/ExtSearchIdx.pm
index 883dbea3..934197c0 100644
--- a/lib/PublicInbox/ExtSearchIdx.pm
+++ b/lib/PublicInbox/ExtSearchIdx.pm
@@ -543,13 +543,7 @@ sub _ibx_for ($$$) {
 sub _fd_constrained ($) {
 	my ($self) = @_;
 	$self->{-fd_constrained} //= do {
-		my $soft;
-		if (eval { require BSD::Resource; 1 }) {
-			my $NOFILE = BSD::Resource::RLIMIT_NOFILE();
-			($soft, undef) = BSD::Resource::getrlimit($NOFILE);
-		} else {
-			chomp($soft = `sh -c 'ulimit -n'`);
-		}
+		my $soft = PublicInbox::Search::ulimit_n;
 		if (defined($soft)) {
 			# $want is an estimate
 			my $want = scalar(@{$self->{ibx_active}}) + 64;
diff --git a/lib/PublicInbox/Search.pm b/lib/PublicInbox/Search.pm
index ae696735..cdf25172 100644
--- a/lib/PublicInbox/Search.pm
+++ b/lib/PublicInbox/Search.pm
@@ -54,6 +54,9 @@ use constant {
 	#
 	#      v1.6.0 adds BYTES, UID and THREADID values
 	SCHEMA_VERSION => 15,
+
+	# we may have up to 8 FDs per shard (depends on Xapian *shrug*)
+	SHARD_COST => 8,
 };
 
 use PublicInbox::Smsg;
@@ -748,4 +751,17 @@ sub get_doc ($$) {
 	}
 }
 
+# not sure where best to put this...
+sub ulimit_n () {
+	my $n;
+	if (eval { require BSD::Resource; 1 }) {
+		my $NOFILE = BSD::Resource::RLIMIT_NOFILE();
+		($n, undef) = BSD::Resource::getrlimit($NOFILE);
+	} else {
+		require PublicInbox::Spawn;
+		$n = PublicInbox::Spawn::run_qx([qw(/bin/sh -c), 'ulimit -n']);
+	}
+	$n;
+}
+
 1;
diff --git a/lib/PublicInbox/XapHelper.pm b/lib/PublicInbox/XapHelper.pm
index f1311bd4..db9e99ae 100644
--- a/lib/PublicInbox/XapHelper.pm
+++ b/lib/PublicInbox/XapHelper.pm
@@ -18,7 +18,7 @@ use POSIX qw(:signal_h);
 use Fcntl qw(LOCK_UN LOCK_EX);
 use Carp qw(croak);
 my $X = \%PublicInbox::Search::X;
-our (%SRCH, %WORKERS, $nworker, $workerset, $in);
+our (%SRCH, %WORKERS, $nworker, $workerset, $in, $SHARD_NFD, $MY_FD_MAX);
 our $stderr = \*STDERR;
 
 sub cmd_test_inspect {
@@ -193,8 +193,14 @@ sub dispatch {
 	my $key = "-d\0".join("\0-d\0", @$dirs);
 	$key .= "\0".join("\0", map { ('-Q', $_) } @{$req->{Q}}) if $req->{Q};
 	my $new;
-	$req->{srch} = $SRCH{$key} //= do {
+	$req->{srch} = $SRCH{$key} // do {
 		$new = { qp_flags => $PublicInbox::Search::QP_FLAGS };
+		my $nfd = scalar(@$dirs) * PublicInbox::Search::SHARD_COST;
+		$SHARD_NFD += $nfd;
+		if ($SHARD_NFD > $MY_FD_MAX) {
+			$SHARD_NFD = $nfd;
+			%SRCH = ();
+		}
 		my $first = shift @$dirs;
 		my $slow_phrase = -f "$first/iamchert";
 		$new->{xdb} = $X->{Database}->new($first);
@@ -207,7 +213,7 @@ sub dispatch {
 		bless $new, $req->{c} ? 'PublicInbox::CodeSearch' :
 					'PublicInbox::Search';
 		$new->{qp} = $new->qparse_new;
-		$new;
+		$SRCH{$key} = $new;
 	};
 	$req->{srch}->{xdb}->reopen unless $new;
 	$req->{Q} && !$req->{srch}->{qp_extra_done} and
@@ -305,7 +311,7 @@ sub start (@) {
 	my $c = getsockopt(local $in = \*STDIN, SOL_SOCKET, SO_TYPE);
 	unpack('i', $c) == SOCK_SEQPACKET or die 'stdin is not SOCK_SEQPACKET';
 
-	local (%SRCH, %WORKERS);
+	local (%SRCH, %WORKERS, $SHARD_NFD, $MY_FD_MAX);
 	PublicInbox::Search::load_xapian();
 	$GLP->getoptionsfromarray(\@argv, my $opt = { j => 1 }, 'j=i') or
 		die 'bad args';
@@ -314,6 +320,10 @@ sub start (@) {
 	for (@PublicInbox::DS::UNBLOCKABLE, POSIX::SIGUSR1) {
 		$workerset->delset($_) or die "delset($_): $!";
 	}
+	$MY_FD_MAX = PublicInbox::Search::ulimit_n //
+		die "E: unable to get RLIMIT_NOFILE: $!";
+	warn "W: RLIMIT_NOFILE=$MY_FD_MAX too low\n" if $MY_FD_MAX < 72;
+	$MY_FD_MAX -= 64;
 
 	local $nworker = $opt->{j};
 	return recv_loop() if $nworker == 0;
diff --git a/lib/PublicInbox/XapHelperCxx.pm b/lib/PublicInbox/XapHelperCxx.pm
index 74852ad1..922bd583 100644
--- a/lib/PublicInbox/XapHelperCxx.pm
+++ b/lib/PublicInbox/XapHelperCxx.pm
@@ -34,6 +34,7 @@ my $ldflags = '-Wl,-O1';
 $ldflags .= ' -Wl,--compress-debug-sections=zlib' if $^O ne 'openbsd';
 my $xflags = ($ENV{CXXFLAGS} // '-Wall -ggdb3 -pipe') . ' ' .
 	' -DTHREADID=' . PublicInbox::Search::THREADID .
+	' -DSHARD_COST=' . PublicInbox::Search::SHARD_COST .
 	' -DXH_SPEC="'.join('',
 		map { s/=.*/:/; $_ } @PublicInbox::Search::XH_SPEC) . '" ' .
 	($ENV{LDFLAGS} // $ldflags);
diff --git a/lib/PublicInbox/xap_helper.h b/lib/PublicInbox/xap_helper.h
index 44e0d63e..c71ac06d 100644
--- a/lib/PublicInbox/xap_helper.h
+++ b/lib/PublicInbox/xap_helper.h
@@ -140,6 +140,7 @@ static int srch_eq(const struct srch *a, const struct srch *b)
 KHASHL_CSET_INIT(KH_LOCAL, srch_set, srch_set, struct srch *,
 		srch_hash, srch_eq)
 static srch_set *srch_cache;
+static long my_fd_max, shard_nfd;
 // sock_fd is modified in signal handler, yes, it's SOCK_SEQPACKET
 static volatile int sock_fd = STDIN_FILENO;
 static sigset_t fullset, workerset;
@@ -552,6 +553,34 @@ static bool is_chert(const char *dir)
 	return false;
 }
 
+static void srch_free(struct srch *srch)
+{
+	delete srch->qp;
+	delete srch->db;
+	free(srch);
+}
+
+static void srch_cache_renew(struct srch *keep)
+{
+	khint_t k;
+
+	// can't delete while iterating, so just free each + clear
+	for (k = kh_begin(srch_cache); k != kh_end(srch_cache); k++) {
+		if (!kh_exist(srch_cache, k)) continue;
+		struct srch *cur = kh_key(srch_cache, k);
+
+		if (cur != keep)
+			srch_free(cur);
+	}
+	srch_set_cs_clear(srch_cache);
+	if (keep) {
+		int absent;
+		k = srch_set_put(srch_cache, keep, &absent);
+		assert(absent);
+		assert(k < kh_end(srch_cache));
+	}
+}
+
 static bool srch_init(struct req *req)
 {
 	int i;
@@ -563,6 +592,13 @@ static bool srch_init(struct req *req)
 			Xapian::QueryParser::FLAG_WILDCARD;
 	if (is_chert(req->dirv[0]))
 		srch->qp_flags &= ~FLAG_PHRASE;
+	long nfd = req->dirc * SHARD_COST;
+
+	shard_nfd += nfd;
+	if (shard_nfd > my_fd_max) {
+		srch_cache_renew(srch);
+		shard_nfd = nfd;
+	}
 	try {
 		srch->db = new Xapian::Database(req->dirv[0]);
 	} catch (...) {
@@ -629,13 +665,6 @@ static void srch_init_extra(struct req *req)
 	req->srch->qp_extra_done = true;
 }
 
-static void srch_free(struct srch *srch)
-{
-	delete srch->qp;
-	delete srch->db;
-	free(srch);
-}
-
 static void dispatch(struct req *req)
 {
 	int c;
@@ -900,12 +929,7 @@ static void cleanup_all(void)
 	cleanup_pids();
 	if (!srch_cache)
 		return;
-
-	khint_t k;
-	for (k = kh_begin(srch_cache); k != kh_end(srch_cache); k++) {
-		if (kh_exist(srch_cache, k))
-			srch_free(kh_key(srch_cache, k));
-	}
+	srch_cache_renew(NULL);
 	srch_set_destroy(srch_cache);
 	srch_cache = NULL;
 }
@@ -1041,12 +1065,20 @@ int main(int argc, char *argv[])
 	socklen_t slen = (socklen_t)sizeof(c);
 	stdout_path = getenv("STDOUT_PATH");
 	stderr_path = getenv("STDERR_PATH");
+	struct rlimit rl;
 
 	if (getsockopt(sock_fd, SOL_SOCKET, SO_TYPE, &c, &slen))
 		err(EXIT_FAILURE, "getsockopt");
 	if (c != SOCK_SEQPACKET)
 		errx(EXIT_FAILURE, "stdin is not SOCK_SEQPACKET");
 
+	if (getrlimit(RLIMIT_NOFILE, &rl))
+		err(EXIT_FAILURE, "getrlimit");
+	my_fd_max = rl.rlim_cur;
+	if (my_fd_max < 72)
+		warnx("W: RLIMIT_NOFILE=%ld too low\n", my_fd_max);
+	my_fd_max -= 64;
+
 	mail_nrp_init();
 	code_nrp_init();
 	srch_cache = srch_set_init();
diff --git a/t/xap_helper.t b/t/xap_helper.t
index 78be8539..b0fa75a2 100644
--- a/t/xap_helper.t
+++ b/t/xap_helper.t
@@ -284,4 +284,27 @@ for my $n (@NO_CXX) {
 	}
 }
 
+SKIP: {
+	my $nr = $ENV{TEST_XH_FDMAX} or
+		skip 'TEST_XH_FDMAX unset', 1;
+	my @xhc = map {
+		local $ENV{PI_NO_CXX} = $_;
+		PublicInbox::XapClient::start_helper('-j0');
+	} @NO_CXX;
+	my $n = 1;
+	my $exp;
+	for (0..(PublicInbox::Search::ulimit_n() * $nr)) {
+		for my $xhc (@xhc) {
+			my $r = $xhc->mkreq([], qw(mset -Q), "tst$n=XTST$n",
+					@ibx_shard_args, qw(rt:0..));
+			chomp(my @res = readline($r));
+			$exp //= $res[0];
+			$exp eq $res[0] or
+				is $exp, $res[0], "mset mismatch on n=$n";
+			++$n;
+		}
+	}
+	ok $exp, "got expected entries ($n)";
+}
+
 done_testing;

[PATCH 2/4] xap_helper: drop DB handles on EMFILE/ENFILE/etc...

[Eric Wong]

This allows the process to recover in case we get the SHARD_COST
calculation wrong in case Xapian uses more FDs than expected in
new versions.  We'll no longer attempt to recover from ENOMEM
and similar errors during Xapian DB initialization and instead
just tear down the process (as we do in other places).
---
 lib/PublicInbox/XapHelper.pm | 27 +++++++++----
 lib/PublicInbox/xap_helper.h | 76 +++++++++++++++++-------------------
 2 files changed, 56 insertions(+), 47 deletions(-)

diff --git a/lib/PublicInbox/XapHelper.pm b/lib/PublicInbox/XapHelper.pm
index db9e99ae..ba41b5d2 100644
--- a/lib/PublicInbox/XapHelper.pm
+++ b/lib/PublicInbox/XapHelper.pm
@@ -202,14 +202,27 @@ sub dispatch {
 			%SRCH = ();
 		}
 		my $first = shift @$dirs;
-		my $slow_phrase = -f "$first/iamchert";
-		$new->{xdb} = $X->{Database}->new($first);
-		for (@$dirs) {
-			$slow_phrase ||= -f "$_/iamchert";
-			$new->{xdb}->add_database($X->{Database}->new($_));
+		for my $retried (0, 1) {
+			my $slow_phrase = -f "$first/iamchert";
+			eval {
+				$new->{xdb} = $X->{Database}->new($first);
+				for (@$dirs) {
+					$slow_phrase ||= -f "$_/iamchert";
+					$new->{xdb}->add_database(
+							$X->{Database}->new($_))
+				}
+			};
+			last unless $@;
+			if ($retried) {
+				die "E: $@\n";
+			} else { # may be EMFILE/ENFILE/ENOMEM....
+				warn "W: $@, retrying...\n";
+				%SRCH = ();
+				$SHARD_NFD = $nfd;
+			}
+			$slow_phrase or $new->{qp_flags}
+				|= PublicInbox::Search::FLAG_PHRASE();
 		}
-		$slow_phrase or
-			$new->{qp_flags} |= PublicInbox::Search::FLAG_PHRASE();
 		bless $new, $req->{c} ? 'PublicInbox::CodeSearch' :
 					'PublicInbox::Search';
 		$new->{qp} = $new->qparse_new;
diff --git a/lib/PublicInbox/xap_helper.h b/lib/PublicInbox/xap_helper.h
index c71ac06d..831afdc6 100644
--- a/lib/PublicInbox/xap_helper.h
+++ b/lib/PublicInbox/xap_helper.h
@@ -581,17 +581,14 @@ static void srch_cache_renew(struct srch *keep)
 	}
 }
 
-static bool srch_init(struct req *req)
+static void srch_init(struct req *req)
 {
 	int i;
 	struct srch *srch = req->srch;
 	const unsigned FLAG_PHRASE = Xapian::QueryParser::FLAG_PHRASE;
-	srch->qp_flags = FLAG_PHRASE |
-			Xapian::QueryParser::FLAG_BOOLEAN |
+	srch->qp_flags = Xapian::QueryParser::FLAG_BOOLEAN |
 			Xapian::QueryParser::FLAG_LOVEHATE |
 			Xapian::QueryParser::FLAG_WILDCARD;
-	if (is_chert(req->dirv[0]))
-		srch->qp_flags &= ~FLAG_PHRASE;
 	long nfd = req->dirc * SHARD_COST;
 
 	shard_nfd += nfd;
@@ -599,37 +596,42 @@ static bool srch_init(struct req *req)
 		srch_cache_renew(srch);
 		shard_nfd = nfd;
 	}
-	try {
-		srch->db = new Xapian::Database(req->dirv[0]);
-	} catch (...) {
-		warn("E: Xapian::Database(%s)", req->dirv[0]);
-		return false;
-	}
-	try {
-		for (i = 1; i < req->dirc; i++) {
-			const char *dir = req->dirv[i];
-			if (srch->qp_flags & FLAG_PHRASE && is_chert(dir))
+	for (int retried = 0; retried < 2; retried++) {
+		srch->qp_flags |= FLAG_PHRASE;
+		i = 0;
+		try {
+			srch->db = new Xapian::Database(req->dirv[i]);
+			if (is_chert(req->dirv[0]))
 				srch->qp_flags &= ~FLAG_PHRASE;
-			srch->db->add_database(Xapian::Database(dir));
+			for (i = 1; i < req->dirc; i++) {
+				const char *dir = req->dirv[i];
+				if (srch->qp_flags & FLAG_PHRASE &&
+						is_chert(dir))
+					srch->qp_flags &= ~FLAG_PHRASE;
+				srch->db->add_database(Xapian::Database(dir));
+			}
+			break;
+		} catch (const Xapian::Error & e) {
+			warnx("E: Xapian::Error: %s (%s)",
+				e.get_description().c_str(), req->dirv[i]);
+		} catch (...) { // does this happen?
+			warn("E: add_database(%s)", req->dirv[i]);
+		}
+		if (retried) {
+			errx(EXIT_FAILURE, "E: can't open %s", req->dirv[i]);
+		} else {
+			warnx("retrying...");
+			if (srch->db)
+				delete srch->db;
+			srch->db = NULL;
+			srch_cache_renew(srch);
 		}
-	} catch (...) {
-		warn("E: add_database(%s)", req->dirv[i]);
-		return false;
-	}
-	try {
-		srch->qp = new Xapian::QueryParser;
-	} catch (...) {
-		perror("E: Xapian::QueryParser");
-		return false;
 	}
+	// these will raise and die on ENOMEM or other errors
+	srch->qp = new Xapian::QueryParser;
 	srch->qp->set_default_op(Xapian::Query::OP_AND);
 	srch->qp->set_database(*srch->db);
-	try {
-		srch->qp->set_stemmer(Xapian::Stem("english"));
-	} catch (...) {
-		perror("E: Xapian::Stem");
-		return false;
-	}
+	srch->qp->set_stemmer(Xapian::Stem("english"));
 	srch->qp->set_stemming_strategy(Xapian::QueryParser::STEM_SOME);
 	srch->qp->SET_MAX_EXPANSION(100);
 
@@ -637,7 +639,6 @@ static bool srch_init(struct req *req)
 		qp_init_code_search(srch->qp); // CodeSearch.pm
 	else
 		qp_init_mail_search(srch->qp); // Search.pm
-	return true;
 }
 
 // setup query parser for altid and arbitrary headers
@@ -761,15 +762,12 @@ static void dispatch(struct req *req)
 	khint_t ki = srch_set_put(srch_cache, kbuf.srch, &absent);
 	assert(ki < kh_end(srch_cache));
 	req->srch = kh_key(srch_cache, ki);
-	if (!absent) { // reuse existing
+	if (absent) {
+		srch_init(req);
+	} else {
 		assert(req->srch != kbuf.srch);
 		srch_free(kbuf.srch);
 		req->srch->db->reopen();
-	} else if (!srch_init(req)) {
-		int gone = srch_set_del(srch_cache, ki);
-		assert(gone);
-		srch_free(kbuf.srch);
-		goto cmd_err; // srch_init already warned
 	}
 	if (req->qpfxc && !req->srch->qp_extra_done)
 		srch_init_extra(req);
@@ -786,8 +784,6 @@ static void dispatch(struct req *req)
 	}
 	if (req->timeout_sec)
 		alarm(0);
-cmd_err:
-	return; // just be silent on errors, for now
 }
 
 static void cleanup_pids(void)

[PATCH 3/4] lei_saved_search: drop ->altid_map method

[Eric Wong]

->altid_map is not currently used by lei.  The only caller is
WwwText which should only see public inboxes, not lei storage.
---
 lib/PublicInbox/LeiSavedSearch.pm | 2 --
 1 file changed, 2 deletions(-)

diff --git a/lib/PublicInbox/LeiSavedSearch.pm b/lib/PublicInbox/LeiSavedSearch.pm
index 9ae9dcdb..1c8a1f73 100644
--- a/lib/PublicInbox/LeiSavedSearch.pm
+++ b/lib/PublicInbox/LeiSavedSearch.pm
@@ -263,8 +263,6 @@ sub reset_dedupe {
 
 sub mm { undef }
 
-sub altid_map { {} }
-
 sub cloneurl { [] }
 
 # find existing directory containing a `lei.saved-search' file based on

[PATCH 4/4] www_text: show indexheader contents as-is

[Eric Wong]

---
 lib/PublicInbox/WwwText.pm | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/PublicInbox/WwwText.pm b/lib/PublicInbox/WwwText.pm
index 84068f5a..1946a227 100644
--- a/lib/PublicInbox/WwwText.pm
+++ b/lib/PublicInbox/WwwText.pm
@@ -168,8 +168,8 @@ sub inbox_config ($$) {
 	url = https://example.com/$name/
 	url = http://example.onion/$name/
 EOS
-	for my $k (qw(address listid infourl watchheader)) {
-		defined(my $v = $ibx->{$k}) or next;
+	for my $k (qw(address listid infourl watchheader indexheader)) {
+		my $v = $ibx->{$k} // next;
 		$$txt .= "\t$k = $_\n" for @$v;
 	}
 	if (my $altid = $ibx->{altid}) {