[RFC 2/2] index+extindex: support indexSequentialCommit

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

This is intended as a middle ground between the default
fully-parallel commit and indexSequentialShard.

Fwiw, my gut is telling me this should be the default
behavior going forward since the worst case with sequential
commits enabled is less likely to be pathologically slow.
---
 Documentation/public-inbox-index.pod | 22 ++++++++++++++++++++++
 lib/PublicInbox/Admin.pm             |  2 +-
 lib/PublicInbox/Config.pm            |  2 +-
 lib/PublicInbox/ExtSearchIdx.pm      |  1 +
 lib/PublicInbox/V2Writable.pm        | 19 ++++++++++++++-----
 script/public-inbox-extindex         |  1 +
 script/public-inbox-index            | 11 +++++++----
 t/v2mirror.t                         |  8 ++++----
 8 files changed, 51 insertions(+), 15 deletions(-)

diff --git a/Documentation/public-inbox-index.pod b/Documentation/public-inbox-index.pod
index 57fedb69..ba4e7d64 100644
--- a/Documentation/public-inbox-index.pod
+++ b/Documentation/public-inbox-index.pod
@@ -158,6 +158,14 @@ below.
 
 Available in public-inbox 1.6.0+.
 
+=item --sequential-commit
+
+Sets or overrides L</publicinbox.indexSequentialCommit> on a
+per-invocation basis.  See L</publicinbox.indexSequentialCommit>
+below.
+
+Available in public-inbox 1.7.0+.
+
 =item --skip-docdata
 
 Stop storing document data in Xapian on an existing inbox.
@@ -266,11 +274,25 @@ This is ignored on L<public-inbox-v1-format(5)> inboxes.
 
 Default: false, shards are indexed in parallel
 
+=item publicinbox.indexSequentialCommit
+
+This allows parallel indexing but commit to Xapian DBs
+synchronously to reduce I/O write contention.  This is intended
+to be a middle ground between the fully-parallel (default)
+behavior and fully-synchronous behavior of indexSequentialShard.
+
+Default: false, shards are indexed and flushed in parallel
+
 =item publicinbox.<name>.indexSequentialShard
 
 Identical to L</publicinbox.indexSequentialShard>,
 but only affect the inbox matching E<lt>nameE<gt>.
 
+=item publicinbox.<name>.indexSequentialCommit
+
+Identical to L</publicinbox.indexSequentialCommit>,
+but only affect the inbox matching E<lt>nameE<gt>.
+
 =back
 
 =head1 ENVIRONMENT
diff --git a/lib/PublicInbox/Admin.pm b/lib/PublicInbox/Admin.pm
index eb38dd8f..9eaed9f1 100644
--- a/lib/PublicInbox/Admin.pm
+++ b/lib/PublicInbox/Admin.pm
@@ -327,7 +327,7 @@ sub index_prepare ($$) {
 	$opt->{batch_size} and
 		$env = { XAPIAN_FLUSH_THRESHOLD => '4294967295' };
 
-	for my $k (qw(sequential-shard)) {
+	for my $k (qw(sequential-shard sequential-commit)) {
 		my $git_key = "publicInbox.index".ucfirst($k);
 		$git_key =~ s/-([a-z])/\U$1/g;
 		defined(my $s = $opt->{$k} // $cfg->{lc($git_key)}) or next;
diff --git a/lib/PublicInbox/Config.pm b/lib/PublicInbox/Config.pm
index 7aa1f6c8..33fa4faa 100644
--- a/lib/PublicInbox/Config.pm
+++ b/lib/PublicInbox/Config.pm
@@ -521,7 +521,7 @@ sub _fill_ei ($$) {
 		return;
 	}
 	my $es = PublicInbox::ExtSearch->new($d);
-	for my $k (qw(indexlevel indexsequentialshard)) {
+	for my $k (qw(indexlevel indexsequentialshard indexsequentialcommit)) {
 		my $v = get_1($self, $pfx, $k) // next;
 		$es->{$k} = $v;
 	}
diff --git a/lib/PublicInbox/ExtSearchIdx.pm b/lib/PublicInbox/ExtSearchIdx.pm
index 22edbb4b..72e6ac65 100644
--- a/lib/PublicInbox/ExtSearchIdx.pm
+++ b/lib/PublicInbox/ExtSearchIdx.pm
@@ -52,6 +52,7 @@ sub new {
 		parallel => 1,
 		lock_path => "$dir/ei.lock",
 	}, __PACKAGE__;
+	$self->{'sequential-commit'} = 1 if $opt->{'sequential-commit'};
 	$self->{shards} = $self->count_shards ||
 		nproc_shards({ nproc => $opt->{jobs} });
 	my $oidx = PublicInbox::OverIdx->new("$self->{xpfx}/over.sqlite3");
diff --git a/lib/PublicInbox/V2Writable.pm b/lib/PublicInbox/V2Writable.pm
index 025487d2..991fa371 100644
--- a/lib/PublicInbox/V2Writable.pm
+++ b/lib/PublicInbox/V2Writable.pm
@@ -559,11 +559,19 @@ sub checkpoint ($;$) {
 		# SQLite overview is third
 		$self->{oidx}->commit_lazy;
 
-		# Now deal with Xapian
-
-		# start commit_txn_lazy asynchronously on all parallel shards
-		# (non-parallel waits here)
-		$_->ipc_do('commit_txn_lazy') for @$shards;
+		# Now deal with Xapian shards
+		if ($self->{'sequential-commit'}) {
+			for my $shard (@$shards) {
+				my @wait = $shard->ipc_do('commit_txn_lazy');
+			}
+			# already waited due to return value, so no need
+			# to `echo' below:
+			$wait = undef;
+		} else {
+			# start commit_txn_lazy asynchronously on all
+			# parallel shards (non-parallel waits here)
+			$_->ipc_do('commit_txn_lazy') for @$shards;
+		}
 
 		# transactions started on parallel shards,
 		# wait for them by issuing an echo command (echo can only
@@ -1314,6 +1322,7 @@ sub xapian_only {
 sub index_sync {
 	my ($self, $opt) = @_;
 	$opt //= {};
+	$self->{'sequential-commit'} = 1 if $opt->{'sequential-commit'};
 	return xapian_only($self, $opt) if $opt->{xapian_only};
 
 	my $epoch_max;
diff --git a/script/public-inbox-extindex b/script/public-inbox-extindex
index addd5ac6..fd6efad6 100755
--- a/script/public-inbox-extindex
+++ b/script/public-inbox-extindex
@@ -29,6 +29,7 @@ GetOptions($opt, qw(verbose|v+ reindex rethread compact|c+ jobs|j=i
 		fsync|sync!
 		indexlevel|index-level|L=s max_size|max-size=s
 		batch_size|batch-size=s
+		sequential-commit|seq-commit
 		dedupe:s@ gc commit-interval=i watch scan! dry-run|n
 		all help|h))
 	or die $help;
diff --git a/script/public-inbox-index b/script/public-inbox-index
index 0034d44c..2ac9cfb6 100755
--- a/script/public-inbox-index
+++ b/script/public-inbox-index
@@ -21,6 +21,7 @@ options:
   --all               index all configured inboxes
   --compact | -c      run public-inbox-compact(1) after indexing
   --sequential-shard  index Xapian shards sequentially for slow storage
+  --sequential-commit commit Xapian shards sequentially for mediocre storage
   --jobs=NUM          set or disable parallelization (NUM=0)
   --batch-size=BYTES  flush changes to OS after a given number of bytes
   --max-size=BYTES    do not index messages larger than the given size
@@ -40,7 +41,7 @@ GetOptions($opt, qw(verbose|v+ reindex rethread compact|c+ jobs|j=i prune
 		fsync|sync! xapian_only|xapian-only
 		indexlevel|index-level|L=s max_size|max-size=s
 		batch_size|batch-size=s
-		sequential-shard|seq-shard
+		sequential-commit|seq-commit sequential-shard|seq-shard
 		no-update-extindex update-extindex|E=s@
 		fast-noop|F skip-docdata all help|h))
 	or die $help;
@@ -125,11 +126,13 @@ for my $ibx (@ibxs) {
 	$ibx->{-skip_docdata} //= $opt->{'skip-docdata'};
 
 	my $ibx_opt = $opt;
-	if (defined(my $s = $ibx->{lc('indexSequentialShard')})) {
+	for my $k (qw(Shard Commit)) {
+		# indexSequentialShard and indexSequentialCommit
+		my $s = $ibx->{"indexsequential\l$k"} // next;
 		defined(my $v = $cfg->git_bool($s)) or die <<EOL;
-publicInbox.$ibx->{name}.indexSequentialShard not boolean
+publicInbox.$ibx->{name}.indexSequential$k not boolean
 EOL
-		$ibx_opt = { %$opt, 'sequential-shard' => $v };
+		$ibx_opt = { %$ibx_opt, "sequential-\l$k" => $v };
 	}
 	my $nidx = PublicInbox::Admin::index_inbox($ibx, undef, $ibx_opt);
 	last if $ibx_opt->{quit};
diff --git a/t/v2mirror.t b/t/v2mirror.t
index 012e5bd2..3ab358b8 100644
--- a/t/v2mirror.t
+++ b/t/v2mirror.t
@@ -190,17 +190,17 @@ is($mibx->git->check($to_purge), undef, 'unindex+prune successful in mirror');
 	is(scalar($mset->items), 0, '1@example.com no longer visible in mirror');
 }
 
-if ('sequential-shard') {
+for my $sw (qw(--sequential-shard --sequential-commit)) {
 	$mset = $mibx->search->mset('m:15@example.com');
 	is(scalar($mset->items), 1, 'large message not indexed');
 	remove_tree(glob("$tmpdir/m/xap*"), glob("$tmpdir/m/msgmap.*"));
-	my $cmd = [ qw(-index -j9 --sequential-shard), "$tmpdir/m" ];
-	ok(run_script($cmd), '--sequential-shard works');
+	my $cmd = [ qw(-index -j9), $sw, "$tmpdir/m" ];
+	ok(run_script($cmd), "$sw works");
 	my @shards = glob("$tmpdir/m/xap*/?");
 	is(scalar(@shards), 8, 'got expected shard count');
 	PublicInbox::InboxWritable::cleanup($mibx);
 	$mset = $mibx->search->mset('m:15@example.com');
-	is(scalar($mset->items), 1, 'search works after --sequential-shard');
+	is(scalar($mset->items), 1, "search works after $sw");
 }
 
 if ('max size') {