[PATCH 6/6] cidx interleaved-prune

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

---
 lib/PublicInbox/CodeSearchIdx.pm | 275 +++++++++++++++++++++----------
 lib/PublicInbox/SearchIdx.pm     |  12 +-
 2 files changed, 196 insertions(+), 91 deletions(-)

diff --git a/lib/PublicInbox/CodeSearchIdx.pm b/lib/PublicInbox/CodeSearchIdx.pm
index 6907570d..e3a021d5 100644
--- a/lib/PublicInbox/CodeSearchIdx.pm
+++ b/lib/PublicInbox/CodeSearchIdx.pm
@@ -32,6 +32,7 @@ use PublicInbox::Spawn qw(spawn popen_rd);
 use PublicInbox::OnDestroy;
 use Socket qw(MSG_EOR);
 use Carp ();
+use List::Util qw(max);
 our (
 	$LIVE, # pid => cmd
 	$DEFER, # [ [ cb, @args ], ... ]
@@ -39,12 +40,22 @@ our (
 	$MY_SIG, # like %SIG
 	$SIGSET,
 	$TXN_BYTES, # number of bytes in current shard transaction
+	$BATCH_BYTES,
 	$DO_QUIT, # signal number
-	@RDONLY_SHARDS, # Xapian::Database
+	@RDONLY_XDB, # Xapian::Database
+	@WORKER_SHARDS, # read-only clones of self
 	@IDX_SHARDS, # clones of self
 	$MAX_SIZE,
 	$TMP_GIT, # PublicInbox::Git object for --prune
 	$REINDEX, # PublicInbox::SharedKV
+	@GIT_DIR_GONE, # [ git_dir1, git_dir2 ]
+	%TO_PRUNE, # (docid => docid) mapping (hash in case of retry_reopen)
+	$PRUNE_CUR, # per-shard document ID
+	$PRUNE_FH, # tracks PRUNE_CUR in the event of C++ exception
+	$PRUNE_SENT, # total number pruned by one of WORKER_SHARDS
+	$PRUNE_RECV, # pruned by one of IDX_SHARDS
+	$NCHANGE, # current number of changes
+	%ACTIVE_GIT_DIR, # GIT_DIR => undef mapping for prune
 );
 
 # stop walking history if we see >$SEEN_MAX existing commits, this assumes
@@ -137,7 +148,7 @@ sub store_repo { # wq_do - returns docid
 	my $xdb = $self->{xdb};
 	for (@{$repo->{to_delete}}) { $xdb->delete_document($_) } # XXX needed?
 	if (defined $repo->{docid}) {
-		my $doc = $xdb->get_document($repo->{docid}) //
+		my $doc = $self->get_doc($repo->{docid}) //
 			die "$repo->{git_dir} doc #$repo->{docid} gone";
 		add_val($doc, PublicInbox::CodeSearch::CT, $repo->{ct});
 		my %new = map { $_ => undef } @{$repo->{roots}};
@@ -198,17 +209,15 @@ EOM
 }
 
 # sharded reader for `git log --pretty=format: --stdin'
-sub shard_index { # via wq_io_do
+sub shard_index { # via wq_io_do in IDX_SHARDS
 	my ($self, $git, $n, $roots) = @_;
 	local $self->{current_info} = "$git->{git_dir} [$n]";
 	local $self->{roots} = $roots;
 	my $in = delete($self->{0}) // die 'BUG: no {0} input';
 	my $op_p = delete($self->{1}) // die 'BUG: no {1} op_p';
-	my $batch_bytes = $self->{-opt}->{batch_size} //
-				$PublicInbox::SearchIdx::BATCH_BYTES;
 	local $MAX_SIZE = $self->{-opt}->{max_size};
 	# local-ized in parent before fork
-	$TXN_BYTES = $batch_bytes;
+	$TXN_BYTES = $BATCH_BYTES;
 	local $self->{git} = $git; # for patchid
 	return if $DO_QUIT;
 	my $rd = $git->popen(@LOG_STDIN, undef, { 0 => $in });
@@ -236,13 +245,13 @@ sub shard_index { # via wq_io_do
 		$TXN_BYTES -= $len;
 		if ($TXN_BYTES <= 0) {
 			cidx_ckpoint($self, "[$n] $nr");
-			$TXN_BYTES = $batch_bytes - $len;
+			$TXN_BYTES = $BATCH_BYTES - $len;
 		}
 		update_commit($self, $cmt);
 		++$nr;
 		if ($TXN_BYTES <= 0) {
 			cidx_ckpoint($self, "[$n] $nr");
-			$TXN_BYTES = $batch_bytes;
+			$TXN_BYTES = $BATCH_BYTES;
 		}
 		$/ = $FS;
 	}
@@ -390,7 +399,7 @@ sub prep_repo ($$) {
 	my $shard = bless { %$self, shard => $n }, ref($self);
 	$repo->{shard_n} = $n;
 	delete @$shard{qw(lockfh lock_path)};
-	local $shard->{xdb} = $RDONLY_SHARDS[$n] // die "BUG: shard[$n] undef";
+	local $shard->{xdb} = $RDONLY_XDB[$n] // die "BUG: shard[$n] undef";
 	$shard->retry_reopen(\&check_existing, $self, $git);
 }
 
@@ -398,7 +407,7 @@ sub check_existing { # retry_reopen callback
 	my ($shard, $self, $git) = @_;
 	my @docids = docids_by_postlist($shard, 'P'.$git->{git_dir});
 	my $docid = shift(@docids) // return get_roots($self, $git);
-	my $doc = $shard->{xdb}->get_document($docid) //
+	my $doc = $shard->get_doc($docid) //
 			die "BUG: no #$docid ($git->{git_dir})";
 	my $old_fp = $REINDEX ? "\0invalid" : $doc->get_data;
 	if ($old_fp eq $git->{-repo}->{fp}) { # no change
@@ -418,24 +427,24 @@ sub partition_refs ($$$) {
 	sysseek($refs, 0, SEEK_SET) or die "seek: $!"; # for rev-list --stdin
 	my $rfh = $git->popen(qw(rev-list --stdin), undef, { 0 => $refs });
 	close $refs or die "close: $!";
-	my ($seen, $nchange) = (0, 0);
+	my $seen = 0;
 	my @shard_in = map {
 		$_->reopen;
 		open my $fh, '+>', undef or die "open: $!";
 		$fh;
-	} @RDONLY_SHARDS;
+	} @RDONLY_XDB;
 
 	while (defined(my $cmt = <$rfh>)) {
 		chomp $cmt;
-		my $n = hex(substr($cmt, 0, 8)) % scalar(@RDONLY_SHARDS);
+		my $n = hex(substr($cmt, 0, 8)) % scalar(@RDONLY_XDB);
 		if ($REINDEX && $REINDEX->set_maybe(pack('H*', $cmt), '')) {
 			say { $shard_in[$n] } $cmt or die "say: $!";
-			++$nchange;
-		} elsif (seen($RDONLY_SHARDS[$n], 'Q'.$cmt)) {
+			++$NCHANGE;
+		} elsif (seen($RDONLY_XDB[$n], 'Q'.$cmt)) {
 			last if ++$seen > $SEEN_MAX;
 		} else {
 			say { $shard_in[$n] } $cmt or die "say: $!";
-			++$nchange;
+			++$NCHANGE;
 			$seen = 0;
 		}
 		if ($DO_QUIT) {
@@ -446,8 +455,7 @@ sub partition_refs ($$$) {
 	close($rfh);
 	return () if $DO_QUIT;
 	if (!$? || (($? & 127) == POSIX::SIGPIPE && $seen > $SEEN_MAX)) {
-		$self->{nchange} += $nchange;
-		progress($self, "$git->{git_dir}: $nchange commits");
+		progress($self, "$git->{git_dir}: $NCHANGE commits");
 		for my $fh (@shard_in) {
 			$fh->flush or die "flush: $!";
 			sysseek($fh, 0, SEEK_SET) or die "seek: $!";
@@ -548,25 +556,25 @@ sub git { $_[0]->{git} }
 
 sub load_existing ($) { # for -u/--update
 	my ($self) = @_;
-	my $dirs = $self->{git_dirs} // [];
+	my $dirs = $self->{git_dirs} //= [];
 	if ($self->{-opt}->{update} || $self->{-opt}->{prune}) {
 		local $self->{xdb};
 		$self->xdb or
 			die "E: $self->{cidx_dir} non-existent for --update\n";
-		my @missing;
 		my @cur = grep {
 			if (-e $_) {
 				1;
 			} else {
-				push @missing, $_;
+				push @GIT_DIR_GONE, $_;
 				undef;
 			}
 		} $self->all_terms('P');
-		@missing = () if $self->{-opt}->{prune};
-		@missing and warn "W: the following repos no longer exist:\n",
-				(map { "W:\t$_\n" } @missing),
+		if (@GIT_DIR_GONE && !$self->{-opt}->{prune}) {
+			warn "W: the following repos no longer exist:\n",
+				(map { "W:\t$_\n" } @GIT_DIR_GONE),
 				"W: use --prune to remove them from ",
 				$self->{cidx_dir}, "\n";
+		}
 		push @$dirs, @cur;
 	}
 	my %uniq; # List::Util::uniq requires Perl 5.26+
@@ -586,13 +594,12 @@ sub cidx_init ($) {
 	}
 	$self->lock_acquire;
 	my @shards;
-	local $TXN_BYTES;
 	for my $n (0..($self->{nshard} - 1)) {
 		my $shard = bless { %$self, shard => $n }, ref($self);
 		delete @$shard{qw(lockfh lock_path)};
 		$shard->idx_acquire;
 		$shard->idx_release;
-		$shard->wq_workers_start("shard[$n]", 1, $SIGSET, {
+		$shard->wq_workers_start("cidx shard[$n]", 1, $SIGSET, {
 			siblings => \@shards, # for ipc_atfork_child
 		}, \&shard_done_wait, $self);
 		push @shards, $shard;
@@ -619,82 +626,166 @@ sub scan_git_dirs ($) {
 	cidx_reap($self, 0);
 }
 
+sub prune_recv { # via wq_do in IDX_SHARDS, called from WORKER_SHARDS
+	my ($self, $to_prune) = @_;
+	return if $DO_QUIT;
+	$self->begin_txn_lazy;
+	$PRUNE_RECV += scalar @$to_prune;
+	my $xdb = $self->{xdb};
+	$xdb->delete_document($_) for @$to_prune;
+	$self->commit_txn_lazy;
+	progress($self, "< prune [$self->{shard}] $PRUNE_RECV");
+}
+
+sub prune_ckpoint ($) {
+	my ($self) = @_;
+	$TXN_BYTES = $BATCH_BYTES;
+	return if $DO_QUIT;
+	my @to_prune = values(%TO_PRUNE) or return;
+	%TO_PRUNE = ();
+	$PRUNE_SENT += scalar(@to_prune);
+	progress($self, "> prune [$self->{shard}] $PRUNE_SENT");
+	$IDX_SHARDS[$self->{shard}]->wq_do('prune_recv', \@to_prune);
+	seek($PRUNE_FH, 0, SEEK_SET) or die "seek: $!";
+	print $PRUNE_FH pack('J',max(@to_prune)) or die "print: $!";
+}
+
+sub get_doclen { # retry_reopen callback
+	my ($self, $id) = @_;
+	$self->{xdb}->get_doclength($id);
+}
+
 sub prune_cb { # git->check_async callback
 	my ($hex, $type, undef, $self_id) = @_;
-	return if $type eq 'commit';
 	my ($self, $id) = @$self_id;
-	my $len = $self->{xdb}->get_doclength($id);
-	progress($self, "$hex $type (doclength=$len)");
-	++$self->{pruned};
-	$self->{xdb}->delete_document($id);
+	return ($PRUNE_CUR = $id) if $type eq 'commit';
+	progress($self, "$hex $type #$id") if ($self->{-opt}->{verbose}//0) > 1;
+	my $len = $self->retry_reopen(\&get_doclen, $id);
+	$TO_PRUNE{$id} = $PRUNE_CUR = $id;
 
-	# all math around batch_bytes calculation is pretty fuzzy,
+	# all math around TXN_BYTES calculation is pretty fuzzy,
 	# but need a way to regularly flush output to avoid OOM,
 	# so assume the average term + position overhead is the
 	# answer to everything: 42
-	return if ($self->{batch_bytes} -= ($len * 42)) > 0;
-	cidx_ckpoint($self, "[$self->{shard}] $self->{pruned}");
-	$self->{batch_bytes} = $self->{-opt}->{batch_size} //
-			$PublicInbox::SearchIdx::BATCH_BYTES;
+	return if ($TXN_BYTES -= ($len * 42)) > 0;
+	prune_ckpoint($self);
+}
+
+sub prune_git_dir ($$$) {
+	my ($self, $id, $doc) = @_;
+	my @P = xap_terms('P', $doc);
+	scalar(@P) == 1 or warn
+"BUG? shard[$self->{shard}] #$id has zero or multiple paths: @P";
+	for my $P (@P) {
+		if (exists($ACTIVE_GIT_DIR{$P}) && -d $P) {
+			$PRUNE_CUR = $id;
+		} else {
+			$TO_PRUNE{$id} = $id;
+			progress($self, "$P gone #$id");
+			my $len = $self->{xdb}->get_doclength($id);
+			$PRUNE_CUR = $id;
+			return if ($TXN_BYTES -= ($len * 42)) > 0;
+			prune_ckpoint($self);
+		}
+	}
 }
 
-sub shard_prune { # via wq_io_do
-	my ($self, $n, $git_dir) = @_;
-	my $op_p = delete($self->{0}) // die 'BUG: no {0} op_p';
-	my $git = PublicInbox::Git->new($git_dir); # TMP_GIT copy
-	$self->begin_txn_lazy;
-	my $xdb = $self->{xdb};
-	my $cur = $xdb->postlist_begin('Tc');
-	my $end = $xdb->postlist_end('Tc');
-	my ($id, @cmt, $oid);
-	local $self->{batch_bytes} = $self->{-opt}->{batch_size} //
-				$PublicInbox::SearchIdx::BATCH_BYTES;
-	local $self->{pruned} = 0;
-	for (; $cur != $end && !$DO_QUIT; $cur++) {
-		@cmt = xap_terms('Q', $xdb, $id = $cur->get_docid);
-		scalar(@cmt) == 1 or
-			warn "BUG? shard[$n] #$id has multiple commits: @cmt";
-		for $oid (@cmt) {
-			$git->check_async($oid, \&prune_cb, [ $self, $id ]);
+sub prune_all { # retry_reopen cb
+	my ($self, $git) = @_;
+	my $max = $self->{xdb}->get_lastdocid;
+	my $cur = $self->{xdb}->postlist_begin('Tc');
+	my $end = $self->{xdb}->postlist_end('Tc');
+	my $id = $PRUNE_CUR;
+	local %ACTIVE_GIT_DIR = map {
+		$_ => undef
+	} (@{$self->{git_dirs}}, @GIT_DIR_GONE);
+
+	for (; $id <= $max && !$DO_QUIT; $id++) {
+		my $doc = $self->get_doc($id) // next;
+		my @cmt = xap_terms('Q', $doc);
+		if (scalar(@cmt) == 0) {
+			prune_git_dir($self, $id, $doc);
+		} else {
+			scalar(@cmt) == 1 or warn
+"BUG? shard[$self->{shard}] #$id has multiple commits: @cmt";
+			for my $o (@cmt) {
+				$git->check_async($o, \&prune_cb, [$self, $id])
+			}
 		}
 	}
-	$git->async_wait_all;
-	for my $d ($self->all_terms('P')) { # GIT_DIR paths
-		last if $DO_QUIT;
-		next if -d $d;
-		for $id (docids_by_postlist($self, 'P'.$d)) {
-			progress($self, "$d gone #$id");
-			$xdb->delete_document($id);
-		}
+}
+
+sub prune_send { # via wq_io_do in WORKER_SHARDS
+	my ($self, $git_dir) = @_;
+	my $n = $self->{shard} // die 'BUG: no {shard}';
+	my ($pid, $wpid);
+	my $cur = 1;
+	open my $fh, '+>', undef or die "open: $!";
+	$fh->autoflush(1);
+	# fork off into a child to deal with unwrapped
+	# Xapian::DatabaseModifiedError C++ exceptions from Search::Xapian
+	# retry_reopen catches most of the properly wrapped-into-Perl
+	# exceptions, but we can't catch (non-Perl) C++ exceptions
+again:
+	$pid = fork // die "fork: $!";
+	if ($pid == 0) {
+		my $end = PublicInbox::OnDestroy->new($$, \&CORE::exit, 1);
+		my $git = PublicInbox::Git->new($git_dir); # TMP_GIT copy
+		local $TXN_BYTES = $BATCH_BYTES;
+		local $PRUNE_SENT = 0;
+		local $PRUNE_FH = $fh;
+		local $PRUNE_CUR = $cur;
+		local %TO_PRUNE;
+		$self->{xdb}->reopen;
+		$self->retry_reopen(\&prune_all, $git);
+		$git->async_wait_all;
+		$git->cleanup;
+		prune_ckpoint($self);
+		$PRUNE_SENT and
+			progress($self, "[$n] pruned $PRUNE_SENT items");
+		$end->cancel;
+		exit 0;
+	}
+	$wpid = waitpid($pid, 0) // die "waitpid($pid) $!";
+	die("W: waitpid($pid) => $wpid ($!)") if $wpid != $pid;
+	if ($? == 134) {
+		seek($fh, 0, SEEK_SET) or die "seek: $!";
+		read($fh, $cur, -s $fh) // die "read: $!";
+		$cur = unpack('J', $cur);
+		warn "W: retrying $$ on uncaught exception from #$cur..\n";
+		goto again;
+	} elsif ($?) {
+		warn "W: prune_send worker exited with \$?=$?\n";
 	}
-	$self->commit_txn_lazy;
-	$self->{pruned} and
-		progress($self, "[$n] pruned $self->{pruned} commits");
-	send($op_p, "shard_done $n", MSG_EOR);
 }
 
-sub do_prune ($) {
+sub ro_workers_start ($) {
 	my ($self) = @_;
-	my $consumers = {};
-	my $git_dir = $TMP_GIT->{git_dir};
-	my $n = 0;
-	local $self->{-shard_ok} = {};
-	for my $s (@IDX_SHARDS) {
-		my ($c, $p) = PublicInbox::PktOp->pair;
-		$c->{ops}->{shard_done} = [ $self ];
-		$s->wq_io_do('shard_prune', [ $p->{op_p} ], $n, $git_dir);
-		$consumers->{$n++} = $c;
+	return () unless $self->{-opt}->{prune}; # TODO fsck?
+	init_tmp_git_dir();
+	my @shards;
+	for my $n (0..($self->{nshard} - 1)) {
+		my $shard = bless { %$self, shard => $n }, ref($self);
+		$shard->wq_workers_start("cidx-ro shard[$n]", 1, $SIGSET, {
+			siblings => \@shards, # for ipc_atfork_child
+			-cidx_ro => 1,
+		}, \&shard_done_wait, $self);
+		push @shards, $shard;
 	}
-	wait_consumers($self, $TMP_GIT, $consumers);
+	@shards;
+}
+
+sub start_prune () {
+	$_->wq_do('prune_send', $TMP_GIT->{git_dir}) for @WORKER_SHARDS;
 }
 
 sub shards_active { # post_loop_do
 	return if $DO_QUIT;
-	scalar(grep { $_->{-cidx_quit} } @IDX_SHARDS);
+	scalar(grep { $_->{-cidx_quit} } (@WORKER_SHARDS, @IDX_SHARDS));
 }
 
 # signal handlers
-sub kill_shards { $_->wq_kill(@_) for @IDX_SHARDS }
+sub kill_shards { $_->wq_kill(@_) for (@WORKER_SHARDS, @IDX_SHARDS) }
 
 sub parent_quit {
 	$DO_QUIT = POSIX->can("SIG$_[0]")->();
@@ -704,7 +795,6 @@ sub parent_quit {
 
 sub init_tmp_git_dir ($) {
 	my ($self) = @_;
-	return unless $self->{-opt}->{prune};
 	require File::Temp;
 	require PublicInbox::Import;
 	my $tmp = File::Temp->newdir('cidx-all-git-XXXX', TMPDIR => 1);
@@ -755,7 +845,10 @@ sub cidx_run { # main entry point
 	my $restore = PublicInbox::OnDestroy->new($$,
 		\&PublicInbox::DS::sig_setmask, $SIGSET);
 	local $LIVE = {};
-	local ($DO_QUIT, $TMP_GIT, $REINDEX);
+	local ($DO_QUIT, $TMP_GIT, $REINDEX, $TXN_BYTES, @GIT_DIR_GONE);
+	local $PRUNE_RECV; # for IDX_SHARDS
+	local $BATCH_BYTES = $self->{-opt}->{batch_size} //
+				$PublicInbox::SearchIdx::BATCH_BYTES;
 	local @IDX_SHARDS = cidx_init($self);
 	local $self->{current_info} = '';
 	local $MY_SIG = {
@@ -795,31 +888,37 @@ sub cidx_run { # main entry point
 			$_ =~ /$re/ ? (warn("# excluding $_\n"), 0) : 1;
 		} @{$self->{git_dirs}};
 	}
-	local $self->{nchange} = 0;
+	local $NCHANGE = 0;
 	local $LIVE_JOBS = $self->{-opt}->{jobs} ||
 			PublicInbox::IPC::detect_nproc() || 2;
-	local @RDONLY_SHARDS = $self->xdb_shards_flat;
-	init_tmp_git_dir($self);
-	do_prune($self) if $self->{-opt}->{prune};
+	local @RDONLY_XDB = $self->xdb_shards_flat;
+	local @WORKER_SHARDS = ro_workers_start($self);
+	start_prune() if $self->{-opt}->{prune};
 	scan_git_dirs($self) if $self->{-opt}->{scan} // 1;
 
-	for my $s (@IDX_SHARDS) {
+	for my $s (@WORKER_SHARDS, @IDX_SHARDS) {
 		$s->{-cidx_quit} = 1;
 		$s->wq_close;
 	}
 
 	local @PublicInbox::DS::post_loop_do = (\&shards_active);
 	PublicInbox::DS::event_loop($MY_SIG, $SIGSET) if shards_active();
-	$self->lock_release(!!$self->{nchange});
+	$self->lock_release(!!$NCHANGE);
 }
 
-sub ipc_atfork_child {
+sub ipc_atfork_child { # both @WORKER_SHARDS and @IDX_SHARDS
 	my ($self) = @_;
 	$self->SUPER::ipc_atfork_child;
 	$SIG{USR1} = \&shard_usr1;
 	$SIG{$_} = \&shard_quit for qw(INT TERM QUIT);
 	my $x = delete $self->{siblings} // die 'BUG: no {siblings}';
 	$_->wq_close for @$x;
+	my $n = $self->{shard} // die 'BUG: no {shard}';
+	if ($self->{-cidx_ro}) {
+		$self->{xdb} = $RDONLY_XDB[$n] // die "BUG: no RDONLY_XDB[$n]";
+		$self->{xdb}->reopen;
+		@RDONLY_XDB = ();
+	}
 	undef;
 }
 
diff --git a/lib/PublicInbox/SearchIdx.pm b/lib/PublicInbox/SearchIdx.pm
index 3baeaa9c..b907772e 100644
--- a/lib/PublicInbox/SearchIdx.pm
+++ b/lib/PublicInbox/SearchIdx.pm
@@ -552,11 +552,17 @@ sub add_message {
 	$smsg->{num};
 }
 
+sub get_doc ($$) {
+	my ($self, $docid) = @_;
+	eval { $self->{xdb}->get_document($docid) } // do {
+		die $@ if $@ && ref($@) !~ /\bDocNotFoundError\b/;
+		undef;
+	}
+}
+
 sub _get_doc ($$) {
 	my ($self, $docid) = @_;
-	my $doc = eval { $self->{xdb}->get_document($docid) };
-	$doc // do {
-		warn "E: $@\n" if $@;
+	get_doc($self, $docid) // do {
 		warn "E: #$docid missing in Xapian\n";
 		undef;
 	}