use PublicInbox::OverIdx;
use PublicInbox::Msgmap;
use PublicInbox::Spawn qw(spawn popen_rd);
-use PublicInbox::SearchIdx qw(log2stack crlf_adjust is_ancestor check_size);
+use PublicInbox::SearchIdx qw(log2stack crlf_adjust is_ancestor check_size
+ is_bad_blob);
use IO::Handle; # ->autoflush
use File::Temp ();
sub count_shards ($) {
my ($self) = @_;
- $self->{ibx} ? do {
+ if (my $ibx = $self->{ibx}) {
# always load existing shards in case core count changes:
# Also, shard count may change while -watch is running
- my $srch = $self->{ibx}->search or return 0;
- delete $self->{ibx}->{search};
+ my $srch = $ibx->search or return 0;
+ delete $ibx->{search};
$srch->{nshard} // 0
- } : $self->{nshard}; # self->{nshard} is for ExtSearchIdx
+ } else { # ExtSearchIdx
+ $self->{nshard} // do {
+ if ($self->xdb_sharded) {
+ $self->{nshard} // die 'BUG: {nshard} unset';
+ } else {
+ 0;
+ }
+ }
+ }
}
sub new {
}
$self->idx_init;
$self->{mm}->skip_artnum($skip_artnum) if defined $skip_artnum;
- my $epoch_max = -1;
- $self->{ibx}->git_dir_latest(\$epoch_max);
- if (defined $skip_epoch && $epoch_max == -1) {
- $epoch_max = $skip_epoch;
- }
- $self->git_init($epoch_max >= 0 ? $epoch_max : 0);
+ my $max = $self->{ibx}->max_git_epoch;
+ $max = $skip_epoch if (defined($skip_epoch) && !defined($max));
+ $self->git_init($max // 0);
$self->done;
}
$self->{shards} = $nshards if $nshards && $nshards != $self->{shards};
$self->{batch_bytes} = $opt->{batch_size} //
$PublicInbox::SearchIdx::BATCH_BYTES;
- $self->{batch_bytes} *= $self->{shards} if $self->{parallel};
# need to create all shards before initializing msgmap FD
# idx_shards must be visible to all forked processes
my $max = $self->{shards} - 1;
my $idx = $self->{idx_shards} = [];
push @$idx, PublicInbox::SearchIdxShard->new($self, $_) for (0..$max);
+
+ # SearchIdxShard may do their own flushing, so don't scale
+ # until after forking
+ $self->{batch_bytes} *= $self->{shards} if $self->{parallel};
+
my $ibx = $self->{ibx} or return; # ExtIdxSearch
# Now that all subprocesses are up, we can open the FDs
my $ibx = $self->{ibx};
my $pfx = "$ibx->{inboxdir}/git";
my $rewrites = []; # epoch => commit
- my $max = $self->{epoch_max};
-
- unless (defined($max)) {
- defined(my $latest = $ibx->git_dir_latest(\$max)) or return;
- $self->{epoch_max} = $max;
- }
+ my $max = $self->{epoch_max} //= $ibx->max_git_epoch // return;
foreach my $i (0..$max) {
my $git_dir = "$pfx/$i.git";
my $mm_tmp = $sync->{mm_tmp};
$mm_tmp->atfork_prepare if $mm_tmp;
die 'BUG: {im} during reindex' if $self->{im};
- if ($self->{ibx_map}) {
+ if ($self->{ibx_map} && !$sync->{checkpoint_unlocks}) {
checkpoint($self, 1); # no need to release lock on pure index
} else {
$self->done; # release lock
# allow -watch or -mda to write...
$self->idx_init($sync->{-opt}); # reacquire lock
+ if (my $intvl = $sync->{check_intvl}) { # eidx
+ $sync->{next_check} = PublicInbox::DS::now() + $intvl;
+ }
$mm_tmp->atfork_parent if $mm_tmp;
}
sub index_oid { # cat_async callback
my ($bref, $oid, $type, $size, $arg) = @_;
+ return if is_bad_blob($oid, $type, $size, $arg->{oid});
my $self = $arg->{self};
local $self->{current_info} = "$self->{current_info} $oid";
return if $size == 0; # purged
$pfx //= $sync->{ibx}->{inboxdir};
}
- # reindex stops at the current heads and we later rerun index_sync
- # without {reindex}
- my $reindex_heads = $self->last_commits($sync) if $sync->{reindex};
-
+ my $reindex_heads;
+ if ($self->{ibx_map}) {
+ # ExtSearchIdx won't index messages unless they're in
+ # over.sqlite3 for a given inbox, so don't read beyond
+ # what's in the per-inbox index.
+ $reindex_heads = [];
+ my $v = PublicInbox::Search::SCHEMA_VERSION;
+ my $mm = $sync->{ibx}->mm;
+ for my $i (0..$sync->{epoch_max}) {
+ $reindex_heads->[$i] = $mm->last_commit_xap($v, $i);
+ }
+ } elsif ($sync->{reindex}) { # V2 inbox
+ # reindex stops at the current heads and we later
+ # rerun index_sync without {reindex}
+ $reindex_heads = $self->last_commits($sync);
+ }
if ($sync->{max_size} = $sync->{-opt}->{max_size}) {
$sync->{index_oid} = $self->can('index_oid');
}
+ my $git_pfx = "$sync->{ibx}->{inboxdir}/git";
for (my $i = $sync->{epoch_max}; $i >= 0; $i--) {
- my $git_dir = $sync->{ibx}->git_dir_n($i);
+ my $git_dir = "$git_pfx/$i.git";
-d $git_dir or next; # missing epochs are fine
my $git = PublicInbox::Git->new($git_dir);
my $unit = { git => $git, epoch => $i };
local $self->{current_info} = 'leftover ';
my $unindex_oid = $self->can('unindex_oid');
for my $oid (@leftovers) {
+ last if $sync->{quit};
$oid = unpack('H*', $oid);
my $req = { %$sync, oid => $oid };
$self->git->cat_async($oid, $unindex_oid, $req);
- last if $sync->{quit};
}
$self->git->cat_async_wait;
}
my @removed = $self->{oidx}->remove_oid($oid, $mid);
for my $num (@removed) {
my $idx = idx_shard($self, $num);
- $idx->shard_remove($oid, $num);
+ $idx->shard_remove($num);
}
}
sub unindex_oid ($$;$) { # git->cat_async callback
my ($bref, $oid, $type, $size, $sync) = @_;
+ return if is_bad_blob($oid, $type, $size, $sync->{oid});
my $self = $sync->{self};
local $self->{current_info} = "$self->{current_info} $oid";
my $unindexed = $sync->{in_unindex} ? $sync->{unindexed} : undef;
"$beg..$end (% $step)\n");
}
for (my $num = $beg; $num <= $end; $num += $step) {
+ last if $sync->{quit};
my $smsg = $ibx->over->get_art($num) or next;
$smsg->{self} = $self;
$ibx->git->cat_async($smsg->{blob}, \&index_xap_only, $smsg);
local $sync->{latest_cmt} = \(my $latest_cmt);
local $sync->{unit} = $unit;
while (my ($f, $at, $ct, $oid, $cmt) = $stk->pop_rec) {
+ if ($sync->{quit}) {
+ warn "waiting to quit...\n";
+ $all->async_wait_all;
+ $self->update_last_commit($sync);
+ return;
+ }
my $req = {
%$sync,
autime => $at,
} elsif ($f eq 'd') {
$all->cat_async($oid, $unindex_oid, $req);
}
- if ($sync->{quit}) {
- warn "waiting to quit...\n";
- $all->async_wait_all;
- $self->update_last_commit($sync);
- return;
- }
if (${$sync->{need_checkpoint}}) {
reindex_checkpoint($self, $sync);
}
if ($seq || !$self->{parallel}) {
my $shard_end = $self->{shards} - 1;
for my $i (0..$shard_end) {
+ last if $sync->{quit};
index_xap_step($self, $sync, $art_beg + $i);
if ($i != $shard_end) {
reindex_checkpoint($self, $sync);
ibx => $self->{ibx},
epoch_max => $epoch_max,
};
- my $quit = sub { $sync->{quit} = 1 };
+ my $quit = PublicInbox::SearchIdx::quit_cb($sync);
local $SIG{QUIT} = $quit;
local $SIG{INT} = $quit;
local $SIG{TERM} = $quit;
$pr->('all.git '.sprintf($sync->{-regen_fmt}, $$nr)) if $pr;
}
+ my $quit_warn;
# deal with Xapian shards sequentially
if ($seq && delete($sync->{mm_tmp})) {
- $self->{ibx}->{indexlevel} = $idxlevel;
- xapian_only($self, $opt, $sync, $art_beg);
+ if ($sync->{quit}) {
+ $quit_warn = 1;
+ } else {
+ $self->{ibx}->{indexlevel} = $idxlevel;
+ xapian_only($self, $opt, $sync, $art_beg);
+ $quit_warn = 1 if $sync->{quit};
+ }
}
# --reindex on the command-line
- if ($opt->{reindex} && !ref($opt->{reindex}) && $idxlevel ne 'basic') {
+ if (!$sync->{quit} && $opt->{reindex} &&
+ !ref($opt->{reindex}) && $idxlevel ne 'basic') {
$self->lock_acquire;
my $s0 = PublicInbox::SearchIdx->new($self->{ibx}, 0, 0);
if (my $xdb = $s0->idx_acquire) {
}
# reindex does not pick up new changes, so we rerun w/o it:
- if ($opt->{reindex}) {
+ if ($opt->{reindex} && !$sync->{quit}) {
my %again = %$opt;
$sync = undef;
delete @again{qw(rethread reindex -skip_lock)};
index_sync($self, \%again);
+ $opt->{quit} = $again{quit}; # propagate to caller
}
+ warn <<EOF if $quit_warn;
+W: interrupted, --xapian-only --reindex required upon restart
+EOF
}
1;