This avoids forking new shard processes for each repo we scan,
but we can't avoid many excessive commits since we need to
ensure the `seen()' sub can avoid excessive work.
---
lib/PublicInbox/CodeSearchIdx.pm | 374 ++++++++++++++++++++-----------
1 file changed, 240 insertions(+), 134 deletions(-)
diff --git a/lib/PublicInbox/CodeSearchIdx.pm b/lib/PublicInbox/CodeSearchIdx.pm
index 02c9ed84..13fe1c28 100644
--- a/lib/PublicInbox/CodeSearchIdx.pm
+++ b/lib/PublicInbox/CodeSearchIdx.pm
@@ -14,9 +14,11 @@
# See PublicInbox::CodeSearch (read-only API) for more
package PublicInbox::CodeSearchIdx;
use v5.12;
-use parent qw(PublicInbox::Lock PublicInbox::CodeSearch
PublicInbox::SearchIdx);
+# parent order matters, we want ->DESTROY from IPC, not SearchIdx
+use parent qw(PublicInbox::CodeSearch PublicInbox::IPC PublicInbox::SearchIdx);
use PublicInbox::Eml;
-use PublicInbox::DS ();
+use PublicInbox::DS qw(awaitpid);
+use PublicInbox::PktOp;
use PublicInbox::IPC qw(nproc_shards);
use PublicInbox::Admin;
use POSIX qw(WNOHANG SEEK_SET);
@@ -26,11 +28,19 @@ use PublicInbox::SHA qw(sha256_hex);
use PublicInbox::Search qw(xap_terms);
use PublicInbox::SearchIdx qw(add_val);
use PublicInbox::Config;
-use PublicInbox::Spawn qw(spawn);
+use PublicInbox::Spawn qw(spawn popen_rd);
use PublicInbox::OnDestroy;
-our $LIVE; # pid => callback
-our $LIVE_JOBS;
-our @XDB_SHARDS_FLAT;
+use Socket qw(MSG_EOR);
+use Carp ();
+our (
+ $LIVE, # pid => cmd
+ $DEFER, # [ [ cb, @args ], ... ]
+ $LIVE_JOBS, # integer
+ $MY_SIG, # like %SIG
+ $SIGSET,
+ @RDONLY_SHARDS, # Xapian::Database
+ @IDX_SHARDS # clones of self
+);
# stop walking history if we see >$SEEN_MAX existing commits, this assumes
# branches don't diverge by more than this number of commits...
@@ -110,14 +120,14 @@ sub progress {
$pr->($self->{git} ? ("$self->{git}->{git_dir}: ") : (), @msg, "\n");
}
-sub store_repo ($$$) {
- my ($self, $git, $repo) = @_;
- my $xdb = delete($repo->{shard})->idx_acquire;
- $xdb->begin_transaction;
+sub store_repo { # wq_do - returns docid
+ my ($self, $repo) = @_;
+ $self->begin_txn_lazy;
+ my $xdb = $self->{xdb};
for (@{$repo->{to_delete}}) { $xdb->delete_document($_) } # XXX needed?
- if (defined $repo->{id}) {
- my $doc = $xdb->get_document($repo->{id}) //
- die "$git->{git_dir} doc #$repo->{id} gone";
+ if (defined $repo->{docid}) {
+ my $doc = $xdb->get_document($repo->{docid}) //
+ die "$repo->{git_dir} doc #$repo->{docid} gone";
add_val($doc, PublicInbox::CodeSearch::CT, $repo->{ct});
my %new = map { $_ => undef } @{$repo->{roots}};
my $old = xap_terms('G', $doc);
@@ -126,34 +136,38 @@ sub store_repo ($$$) {
delete @$old{@{$repo->{roots}}};
$doc->remove_term('G'.$_) for keys %$old;
$doc->set_data($repo->{fp});
- $xdb->replace_document($repo->{id}, $doc);
+ $xdb->replace_document($repo->{docid}, $doc);
+ $repo->{docid}
} else {
my $new = $PublicInbox::Search::X{Document}->new;
add_val($new, PublicInbox::CodeSearch::CT, $repo->{ct});
- $new->add_boolean_term("P$git->{git_dir}");
+ $new->add_boolean_term("P$repo->{git_dir}");
$new->add_boolean_term('T'.'r');
$new->add_boolean_term('G'.$_) for @{$repo->{roots}};
$new->set_data($repo->{fp}); # \n delimited
$xdb->add_document($new);
}
- $xdb->commit_transaction;
}
# sharded reader for `git log --pretty=format: --stdin'
-sub shard_worker ($$$) {
- my ($self, $r, $sigset) = @_;
+sub shard_index { # via wq_io_do
+ my ($self, $git, $n, $roots) = @_;
+ local $self->{current_info} = "$git->{git_dir} [$n]";
my ($quit, $cmt);
+ 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;
my $max = $batch_bytes;
- $SIG{USR1} = sub { $max = -1 }; # similar to `git fast-import'
- $SIG{QUIT} = $SIG{TERM} = $SIG{INT} = sub { $quit = shift };
- PublicInbox::DS::sig_setmask($sigset);
-
- # the parent process of this shard process writes directly to
- # the stdin of `git log', we consume git log's stdout:
- my $rd = $self->{git}->popen(@LOG_STDIN, undef, { 0 => $r });
- close $r or die "close: $!";
+ my $set_quit = sub { $quit = shift };
+ local $SIG{USR1} = sub { $max = -1 }; # similar to `git fast-import'
+ local $SIG{QUIT} = $set_quit;
+ local $SIG{TERM} = $set_quit;
+ local $SIG{INT} = $set_quit;
+ local $self->{git} = $git; # for patchid
+ my $rd = $git->popen(@LOG_STDIN, undef, { 0 => $in });
+ close $in or die "close: $!";
my $nr = 0;
# a patch may have \0, see c4201214cbf10636e2c1ab9131573f735b42c8d4
@@ -162,8 +176,7 @@ sub shard_worker ($$$) {
local $/ = $FS;
my $buf = <$rd> // return; # leading $FS
$buf eq $FS or die "BUG: not LF-NUL: $buf\n";
- my $xdb = $self->idx_acquire;
- $xdb->begin_transaction;
+ $self->begin_txn_lazy;
while (defined($buf = <$rd>)) {
chomp($buf);
$max -= length($buf);
@@ -174,24 +187,40 @@ sub shard_worker ($$$) {
++$nr;
if ($max <= 0 && !$PublicInbox::Search::X{CLOEXEC_UNSET}) {
progress($self, $nr);
- $xdb->commit_transaction;
+ $self->{xdb}->commit_transaction;
$max = $batch_bytes;
- $xdb->begin_transaction;
+ $self->{xdb}->begin_transaction;
}
$/ = $FS;
}
close($rd);
if (!$? || ($quit && ($? & 127) == POSIX::SIGPIPE)) {
- $xdb->commit_transaction;
+ send($op_p, "shard_done $n", MSG_EOR);
} else {
warn "E: git @LOG_STDIN: \$?=$?\n";
- $xdb->cancel_transaction;
+ $self->{xdb}->cancel_transaction;
}
}
+sub shard_done { # called via PktOp on shard_index completion
+ my ($self, $n) = @_;
+ $self->{-shard_ok}->{$n} = 1 if defined($self->{-shard_ok});
+}
+
sub seen ($$) {
my ($xdb, $q) = @_; # $q = "Q$COMMIT_HASH"
- $xdb->postlist_begin($q) != $xdb->postlist_end($q)
+ for (1..100) {
+ my $ret = eval {
+ $xdb->postlist_begin($q) != $xdb->postlist_end($q);
+ };
+ return $ret unless $@;
+ if (ref($@) =~ /\bDatabaseModifiedError\b/) {
+ $xdb->reopen;
+ } else {
+ Carp::croak($@);
+ }
+ }
+ Carp::croak('too many Xapian DB modifications in progress');
}
# used to select the shard for a GIT_DIR
@@ -206,18 +235,42 @@ sub docids_by_postlist ($$) { # consider moving to
PublicInbox::Search
@ids;
}
+sub run_todo ($) {
+ my ($self) = @_;
+ my $n;
+ while (defined(my $x = shift(@{$self->{todo} // []}))) {
+ my $cb = shift @$x;
+ $cb->(@$x);
+ ++$n;
+ }
+ $n;
+}
+
sub cidx_reap ($$) {
my ($self, $jobs) = @_;
- while (keys(%$LIVE) >= $jobs) {
- my $pid = waitpid(-1, 0) // die "waitpid(-1): $!";
- last if $pid < 0;
- if (my $x = delete $LIVE->{$pid}) {
- my $cb = shift @$x;
- $cb->(@$x) if $cb;
- } else {
- warn "reaped unknown PID=$pid ($?)\n";
- }
+ while (run_todo($self)) {}
+ my $cb = sub { keys(%$LIVE) > $jobs };
+ PublicInbox::DS->SetPostLoopCallback($cb);
+ PublicInbox::DS::event_loop($MY_SIG, $SIGSET) while $cb->();
+ while (!$jobs && run_todo($self)) {}
+}
+
+sub cidx_await_cb { # awaitpid cb
+ my ($pid, $cb, $self, $git, @args) = @_;
+ return if !$LIVE; # premature shutdown
+ my $cmd = delete $LIVE->{$pid} // die 'BUG: no $cmd';
+ PublicInbox::DS::enqueue_reap() if !keys(%$LIVE); # once more for PLC
+ if ($?) {
+ $git->{-cidx_err} = 1;
+ return warn("@$cmd error: \$?=$?\n");
}
+ push(@$DEFER, [ $cb, $self, $git, @args ]) if $DEFER;
+}
+
+sub cidx_await ($$$$$@) {
+ my ($pid, $cmd, $cb, $self, $git, @args) = @_;
+ $LIVE->{$pid} = $cmd;
+ awaitpid($pid, \&cidx_await_cb, $cb, $self, $git, @args);
}
# this is different from the grokmirror-compatible fingerprint since we
@@ -227,13 +280,14 @@ sub fp_start ($$$) {
return if !$LIVE; # premature exit
cidx_reap($self, $LIVE_JOBS);
open my $refs, '+>', undef or die "open: $!";
- my $pid = spawn(['git', "--git-dir=$git->{git_dir}",
- qw(show-ref --heads --tags --hash)], undef, { 1 => $refs });
+ my $cmd = ['git', "--git-dir=$git->{git_dir}",
+ qw(show-ref --heads --tags --hash)];
+ my $pid = spawn($cmd, undef, { 1 => $refs });
$git->{-repo}->{refs} = $refs;
- $LIVE->{$pid} = [ \&fp_fini, $self, $git, $prep_repo ];
+ cidx_await($pid, $cmd, \&fp_fini, $self, $git, $prep_repo);
}
-sub fp_fini {
+sub fp_fini { # cidx_await cb
my ($self, $git, $prep_repo) = @_;
my $refs = $git->{-repo}->{refs} // die 'BUG: no {-repo}->{refs}';
seek($refs, 0, SEEK_SET) or die "seek: $!";
@@ -247,13 +301,15 @@ sub ct_start ($$$) {
my ($self, $git, $prep_repo) = @_;
return if !$LIVE; # premature exit
cidx_reap($self, $LIVE_JOBS);
- my ($rd, $pid) = $git->popen([qw[for-each-ref --sort=-committerdate
+ my $cmd = [ 'git', "--git-dir=$git->{git_dir}",
+ qw[for-each-ref --sort=-committerdate
--format=%(committerdate:raw) --count=1
- refs/heads/ refs/tags/]]);
- $LIVE->{$pid} = [ \&ct_fini, $self, $git, $rd, $prep_repo ];
+ refs/heads/ refs/tags/] ];
+ my ($rd, $pid) = popen_rd($cmd);
+ cidx_await($pid, $cmd, \&ct_fini, $self, $git, $rd, $prep_repo);
}
-sub ct_fini {
+sub ct_fini { # cidx_await cb
my ($self, $git, $rd, $prep_repo) = @_;
defined(my $ct = <$rd>) or return;
$ct =~ s/\s+.*\z//s; # drop TZ + LF
@@ -263,34 +319,38 @@ sub ct_fini {
# TODO: also index gitweb.owner and the full fingerprint for grokmirror?
sub prep_repo ($$) {
my ($self, $git) = @_;
- return if !$LIVE; # premature exit
+ return if !$LIVE || $git->{-cidx_err}; # premature exit
my $repo = $git->{-repo} // die 'BUG: no {-repo}';
- my $git_dir = $git->{git_dir};
if (!defined($repo->{ct})) {
- warn "W: $git_dir has no commits, skipping\n";
+ warn "W: $git->{git_dir} has no commits, skipping\n";
delete $git->{-repo};
return;
}
- my $n = git_dir_hash($git_dir) % $self->{nshard};
- my $shard = $repo->{shard} = bless { %$self, shard => $n }, ref($self);
+ my $n = git_dir_hash($git->{git_dir}) % $self->{nshard};
+ my $shard = bless { %$self, shard => $n }, ref($self);
+ $repo->{shard_n} = $n;
delete @$shard{qw(lockfh lock_path)};
- my $xdb = $XDB_SHARDS_FLAT[$n] // die "BUG: shard[$n] undef";
- $xdb->reopen;
- my @docids = docids_by_postlist({ xdb => $xdb }, 'P'.$git_dir);
+ local $shard->{xdb} = $RDONLY_SHARDS[$n] // die "BUG: shard[$n] undef";
+ $shard->retry_reopen(\&check_existing, $self, $git);
+}
+
+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);
- if (@docids) {
- warn "BUG: $git_dir indexed multiple times, culling\n";
- $repo->{to_delete} = \@docids; # XXX needed?
- }
- my $doc = $xdb->get_document($docid) //
- die "BUG: no #$docid ($git_dir)";
+ my $doc = $shard->{xdb}->get_document($docid) //
+ die "BUG: no #$docid ($git->{git_dir})";
my $old_fp = $doc->get_data;
- if ($old_fp eq $repo->{fp}) { # no change
- progress($self, "$git_dir unchanged");
+ if ($old_fp eq $git->{-repo}->{fp}) { # no change
+ progress($self, "$git->{git_dir} unchanged");
delete $git->{-repo};
return;
}
- $repo->{id} = $docid;
+ $git->{-repo}->{docid} = $docid;
+ if (@docids) {
+ warn "BUG: $git->{git_dir} indexed multiple times, culling\n";
+ $git->{-repo}->{to_delete} = \@docids; # XXX needed?
+ }
get_roots($self, $git);
}
@@ -304,12 +364,12 @@ sub partition_refs ($$$) {
$_->reopen;
open my $fh, '+>', undef or die "open: $!";
$fh;
- } @XDB_SHARDS_FLAT;
+ } @RDONLY_SHARDS;
while (defined(my $cmt = <$fh>)) {
chomp $cmt;
- my $n = hex(substr($cmt, 0, 8)) % scalar(@XDB_SHARDS_FLAT);
- if (seen($XDB_SHARDS_FLAT[$n], 'Q'.$cmt)) {
+ my $n = hex(substr($cmt, 0, 8)) % scalar(@RDONLY_SHARDS);
+ if (seen($RDONLY_SHARDS[$n], 'Q'.$cmt)) {
last if ++$seen > $SEEN_MAX;
} else {
say { $shard_in[$n] } $cmt or die "say: $!";
@@ -330,9 +390,33 @@ sub partition_refs ($$$) {
die "git --git-dir=$git->{git_dir} rev-list: \$?=$?\n";
}
-sub index_repo {
+sub shard_commit { # via wq_io_do
+ my ($self, $n) = @_;
+ my $op_p = delete($self->{0}) // die 'BUG: no {0} op_p';
+ $self->commit_txn_lazy;
+ send($op_p, "shard_done $n", MSG_EOR);
+}
+
+sub commit_used_shards ($$$) {
+ my ($self, $git, $consumers) = @_;
+ local $self->{-shard_ok} = {};
+ for my $n (keys %$consumers) {
+ my ($c, $p) = PublicInbox::PktOp->pair;
+ $c->{ops}->{shard_done} = [ $self ];
+ $IDX_SHARDS[$n]->wq_io_do('shard_commit', [ $p->{op_p} ], $n);
+ $consumers->{$n} = $c;
+ }
+ PublicInbox::DS->SetPostLoopCallback(sub {
+ scalar(grep { $_->{sock} } values %$consumers);
+ });
+ PublicInbox::DS::event_loop($MY_SIG, $SIGSET);
+ my $n = grep { ! $self->{-shard_ok}->{$_} } keys %$consumers;
+ die "E: $git->{git_dir} $n shards failed" if $n;
+}
+
+sub index_repo { # cidx_await cb
my ($self, $git, $roots) = @_;
- return if !$LIVE; # premature exit
+ return if $git->{-cidx_err};
my $repo = delete $git->{-repo} or return;
seek($roots, 0, SEEK_SET) or die "seek: $!";
chomp(my @roots = <$roots>);
@@ -341,73 +425,45 @@ sub index_repo {
$repo->{roots} = \@roots;
local $self->{current_info} = $git->{git_dir};
my @shard_in = partition_refs($self, $git, delete($repo->{refs}));
- my %pids;
- my $fwd_kill = sub {
- my ($sig) = @_;
- kill($sig, $_) for keys %pids;
- };
- local $SIG{USR1} = $fwd_kill;
- local $SIG{QUIT} = $fwd_kill;
- local $SIG{INT} = $fwd_kill;
- local $SIG{TERM} = $fwd_kill;
- my $sigset = PublicInbox::DS::block_signals();
- for (my $n = 0; $n <= $#shard_in; $n++) {
+ local $self->{-shard_ok} = {}; # [0..$#shard_in] => 1
+ my %CONSUMERS;
+ for my $n (0..$#shard_in) {
-s $shard_in[$n] or next;
- my $pid = fork // die "fork: $!";
- if ($pid == 0) { # no RNG use, here
- $0 = "code index [$n]";
- $self->{git} = $git;
- $self->{shard} = $n;
- $self->{current_info} = "$self->{current_info} [$n]";
- delete @$self{qw(lockfh lock_path)};
- my $in = $shard_in[$n];
- @shard_in = ();
- $self->{roots} = \@roots;
- undef $repo;
- eval { shard_worker($self, $in, $sigset) };
- warn "E: $@" if $@;
- POSIX::_exit($@ ? 1 : 0);
- } else {
- $pids{$pid} = "code index [$n]";
- }
+ my ($c, $p) = PublicInbox::PktOp->pair;
+ $c->{ops}->{shard_done} = [ $self ];
+ $IDX_SHARDS[$n]->wq_io_do('shard_index',
+ [ $shard_in[$n], $p->{op_p} ],
+ $git, $n, \@roots);
+ $CONSUMERS{$n} = $c;
}
- PublicInbox::DS::sig_setmask($sigset);
@shard_in = ();
- my ($err, @todo);
- while (keys %pids) {
- my $pid = waitpid(-1, 0) // die "waitpid: $!";
- if (my $j = delete $pids{$pid}) {
- next if $? == 0;
- warn "PID:$pid $j exited with \$?=$?\n";
- $err = 1;
- } elsif (my $todo = delete $LIVE->{$pid}) {
- warn "PID:$pid exited with \$?=$?\n" if $?;
- push @todo, $todo;
- } else {
- warn "reaped unknown PID=$pid ($?)\n";
- }
- }
- die "subprocess(es) failed\n" if $err;
- store_repo($self, $git, $repo);
- progress($self, "$git->{git_dir}: done");
- # TODO: check fp afterwards?
- while (my $x = shift @todo) {
- my $cb = shift @$x;
- $cb->(@$x) if $cb;
+ PublicInbox::DS->SetPostLoopCallback(sub {
+ scalar(grep { $_->{sock} } values %CONSUMERS);
+ });
+ PublicInbox::DS::event_loop($MY_SIG, $SIGSET);
+ my $n = grep { ! $self->{-shard_ok}->{$_} } keys %CONSUMERS;
+ die "E: $git->{git_dir} $n shards failed" if $n;
+ $repo->{git_dir} = $git->{git_dir};
+ my $id = $IDX_SHARDS[$repo->{shard_n}]->wq_do('store_repo', $repo);
+ if ($id > 0) {
+ $CONSUMERS{$repo->{shard_n}} = undef;
+ commit_used_shards($self, $git, \%CONSUMERS);
+ progress($self, "$git->{git_dir}: done");
+ return run_todo($self);
}
+ die "E: store_repo $git->{git_dir}: id=$id";
}
sub get_roots ($$) {
my ($self, $git) = @_;
return if !$LIVE; # premature exit
- cidx_reap($self, $LIVE_JOBS);
my $refs = $git->{-repo}->{refs} // die 'BUG: no {-repo}->{refs}';
sysseek($refs, 0, SEEK_SET) or die "seek: $!";
open my $roots, '+>', undef or die "open: $!";
- my $pid = spawn(['git', "--git-dir=$git->{git_dir}",
- qw(rev-list --stdin --max-parents=0)],
- undef, { 0 => $refs, 1 => $roots });
- $LIVE->{$pid} = [ \&index_repo, $self, $git, $roots ];
+ my $cmd = [ 'git', "--git-dir=$git->{git_dir}",
+ qw(rev-list --stdin --max-parents=0) ];
+ my $pid = spawn($cmd, undef, { 0 => $refs, 1 => $roots });
+ cidx_await($pid, $cmd, \&index_repo, $self, $git, $roots);
}
# for PublicInbox::SearchIdx::patch_id and with_umask
@@ -434,9 +490,17 @@ sub cidx_init ($) {
warn "# creating $dir\n" if !$self->{-opt}->{quiet};
File::Path::mkpath($dir);
}
+ $self->lock_acquire;
+ my @shards;
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, undef, {
+ siblings => \@shards, # for ipc_atfork_child
+ }, \&shard_done_wait, $self);
+ push @shards, $shard;
}
# this warning needs to happen after idx_acquire
state $once;
@@ -444,14 +508,11 @@ sub cidx_init ($) {
W: Xapian v1.2.21..v1.2.24 were missing close-on-exec on OFD locks,
W: memory usage may be high for large indexing runs
EOM
+ @shards;
}
sub scan_git_dirs ($) {
my ($self) = @_;
- local $LIVE_JOBS = $self->{-opt}->{jobs} //
- PublicInbox::IPC::detect_nproc() // 2;
- local $LIVE = {};
- local @XDB_SHARDS_FLAT = $self->xdb_shards_flat;
for (@{$self->{git_dirs}}) {
my $git = PublicInbox::Git->new($_);
my $prep_repo = PublicInbox::OnDestroy->new($$, \&prep_repo,
@@ -462,18 +523,31 @@ sub scan_git_dirs ($) {
cidx_reap($self, 0);
}
-sub cidx_run {
+sub shards_active { # PostLoopCallback
+ scalar(grep { $_->{-cidx_quit} } @IDX_SHARDS);
+}
+
+sub cidx_run { # main entry point
my ($self) = @_;
- cidx_init($self);
+ local $self->{todo} = [];
+ local $DEFER = $self->{todo};
+ local $SIGSET = PublicInbox::DS::block_signals();
+ my $restore = PublicInbox::OnDestroy->new($$,
+ \&PublicInbox::DS::sig_setmask, $SIGSET);
+ local $LIVE = {};
+ local @IDX_SHARDS = cidx_init($self);
local $self->{current_info} = '';
my $cb = $SIG{__WARN__} || \&CORE::warn;
+ local $MY_SIG = {
+ CHLD => \&PublicInbox::DS::enqueue_reap,
+ INT => sub { exit },
+ };
local $SIG{__WARN__} = sub {
my $m = shift @_;
$self->{current_info} eq '' or
$m =~ s/\A(#?\s*)/$1$self->{current_info}: /;
$cb->($m, @_);
};
- $self->lock_acquire;
load_existing($self);
my @nc = grep { File::Spec->canonpath($_) ne $_ } @{$self->{git_dirs}};
if (@nc) {
@@ -486,9 +560,41 @@ sub cidx_run {
warn "E: canonicalized and attempting to continue\n";
}
local $self->{nchange} = 0;
+ local $LIVE_JOBS = $self->{-opt}->{jobs} ||
+ PublicInbox::IPC::detect_nproc() || 2;
+ local @RDONLY_SHARDS = $self->xdb_shards_flat;
+
# do_prune($self) if $self->{-opt}->{prune}; TODO
scan_git_dirs($self) if $self->{-opt}->{scan} // 1;
+
+ for my $s (@IDX_SHARDS) {
+ $s->{-cidx_quit} = 1;
+ $s->wq_close;
+ }
+
+ PublicInbox::DS->SetPostLoopCallback(\&shards_active);
+ PublicInbox::DS::event_loop($MY_SIG, $SIGSET) if shards_active();
$self->lock_release(!!$self->{nchange});
}
+sub ipc_atfork_child {
+ my ($self) = @_;
+ $self->SUPER::ipc_atfork_child;
+ my $x = delete $self->{siblings} // die 'BUG: no {siblings}';
+ $_->wq_close for @$x;
+}
+
+sub shard_done_wait { # awaitpid cb via ipc_worker_reap
+ my ($pid, $shard, $self) = @_;
+ delete($shard->{-cidx_quit}) // warn 'BUG: {-cidx_quit} unset';
+ return unless $?;
+ warn "PID:$pid $shard->{shard} exited with \$?=$?\n";
+ ++$self->{shard_err} if defined($self->{shard_err});
+}
+
+sub with_umask { # TODO
+ my ($self, $cb, @arg) = @_;
+ $cb->(@arg);
+}
+
1;
