* [PATCH] *index: more consistent graceful shutdown checks
@ 2020-11-27 21:33 Eric Wong
0 siblings, 0 replies; only message in thread
From: Eric Wong @ 2020-11-27 21:33 UTC (permalink / raw)
To: meta
v1 and v2 inbox indexing now supports graceful shutdown checks
just like ExtSearchIdx. Additionally, we'll consistently
perform quit checks at the top of loops for consistency.
Interaction with the --xapian-only and --sequential-shard
options are a bit lacking, and will warn the user to use
"--reindex --xapian-only" to fix.
---
lib/PublicInbox/ExtSearchIdx.pm | 12 +++++------
lib/PublicInbox/SearchIdx.pm | 25 ++++++++++++++++++----
lib/PublicInbox/V2Writable.pm | 37 ++++++++++++++++++++++-----------
script/public-inbox-index | 1 +
4 files changed, 53 insertions(+), 22 deletions(-)
diff --git a/lib/PublicInbox/ExtSearchIdx.pm b/lib/PublicInbox/ExtSearchIdx.pm
index 7ab0c4af..cf90c562 100644
--- a/lib/PublicInbox/ExtSearchIdx.pm
+++ b/lib/PublicInbox/ExtSearchIdx.pm
@@ -309,10 +309,11 @@ sub _sync_inbox ($$$) {
warn "E: $ekey unsupported inbox version (v$v)\n";
return;
}
- unless ($sync->{quit}) {
- index_todo($self, $sync, $_) for @{delete($sync->{todo}) // []};
- $self->{midx}->index_ibx($ibx) unless $sync->{quit};
+ for my $unit (@{delete($sync->{todo}) // []}) {
+ last if $sync->{quit};
+ index_todo($self, $sync, $unit);
}
+ $self->{midx}->index_ibx($ibx) unless $sync->{quit};
$ibx->git->cleanup; # done with this inbox, now
}
@@ -334,17 +335,16 @@ sub eidx_sync { # main entry point
-regen_fmt => "%u/?\n",
};
local $SIG{USR1} = sub { $need_checkpoint = 1 };
- my $quit = sub { $sync->{quit} = 1; warn "gracefully quitting\n"; };
+ my $quit = PublicInbox::SearchIdx::quit_cb($sync);
local $SIG{QUIT} = $quit;
local $SIG{INT} = $quit;
local $SIG{TERM} = $quit;
# don't use $_ here, it'll get clobbered by reindex_checkpoint
for my $ibx (@{$self->{ibx_list}}) {
- _sync_inbox($self, $sync, $ibx);
last if $sync->{quit};
+ _sync_inbox($self, $sync, $ibx);
}
-
$self->{oidx}->rethread_done($opt) unless $sync->{quit};
PublicInbox::V2Writable::done($self);
diff --git a/lib/PublicInbox/SearchIdx.pm b/lib/PublicInbox/SearchIdx.pm
index 18390602..d06c159b 100644
--- a/lib/PublicInbox/SearchIdx.pm
+++ b/lib/PublicInbox/SearchIdx.pm
@@ -632,10 +632,11 @@ sub index_sync {
my ($self, $opt) = @_;
delete $self->{lock_path} if $opt->{-skip_lock};
$self->with_umask(\&_index_sync, $self, $opt);
- if ($opt->{reindex}) {
+ if ($opt->{reindex} && !$opt->{quit}) {
my %again = %$opt;
delete @again{qw(rethread reindex)};
index_sync($self, \%again);
+ $opt->{quit} = $again{quit}; # propagate to caller
}
}
@@ -688,7 +689,7 @@ sub v1_checkpoint ($$;$) {
if (my $pr = $sync->{-opt}->{-progress}) {
$pr->("indexed $nr/$sync->{ntodo}\n") if $nr;
}
- if (!$stk) { # more to come
+ if (!$stk && !$sync->{quit}) { # more to come
begin_txn_lazy($self);
$self->{mm}->{dbh}->begin_work;
}
@@ -709,6 +710,7 @@ sub process_stack {
if (my @leftovers = keys %{delete($sync->{D}) // {}}) {
warn('W: unindexing '.scalar(@leftovers)." leftovers\n");
for my $oid (@leftovers) {
+ last if $sync->{quit};
$oid = unpack('H*', $oid);
$git->cat_async($oid, \&unindex_both, $sync);
}
@@ -718,6 +720,7 @@ sub process_stack {
}
while (my ($f, $at, $ct, $oid, $cur_cmt) = $stk->pop_rec) {
my $arg = { %$sync, cur_cmt => $cur_cmt };
+ last if $sync->{quit};
if ($f eq 'm') {
$arg->{autime} = $at;
$arg->{cotime} = $ct;
@@ -731,7 +734,7 @@ sub process_stack {
$git->cat_async($oid, \&unindex_both, $arg);
}
}
- v1_checkpoint($self, $sync, $stk);
+ v1_checkpoint($self, $sync, $sync->{quit} ? undef : $stk);
}
sub log2stack ($$$) {
@@ -841,6 +844,16 @@ sub reindex_from ($$) {
ref($reindex) eq 'HASH' ? $reindex->{from} : '';
}
+sub quit_cb ($) {
+ my ($sync) = @_;
+ sub {
+ # we set {-opt}->{quit} too, so ->index_sync callers
+ # can abort multi-inbox loops this way
+ $sync->{quit} = $sync->{-opt}->{quit} = 1;
+ warn "gracefully quitting\n";
+ }
+}
+
# indexes all unindexed messages (v1 only)
sub _index_sync {
my ($self, $opt) = @_;
@@ -850,6 +863,10 @@ sub _index_sync {
$ibx->git->batch_prepare;
my $pr = $opt->{-progress};
my $sync = { reindex => $opt->{reindex}, -opt => $opt, ibx => $ibx };
+ my $quit = quit_cb($sync);
+ local $SIG{QUIT} = $quit;
+ local $SIG{INT} = $quit;
+ local $SIG{TERM} = $quit;
my $xdb = $self->begin_txn_lazy;
$self->{oidx}->rethread_prepare($opt);
my $mm = _msgmap_init($self);
@@ -870,7 +887,7 @@ sub _index_sync {
my $stk = prepare_stack($sync, $range);
$sync->{ntodo} = $stk ? $stk->num_records : 0;
$pr->("$sync->{ntodo}\n") if $pr; # continue previous line
- process_stack($self, $sync, $stk);
+ process_stack($self, $sync, $stk) if !$sync->{quit};
}
sub DESTROY {
diff --git a/lib/PublicInbox/V2Writable.pm b/lib/PublicInbox/V2Writable.pm
index afba0220..c94623e1 100644
--- a/lib/PublicInbox/V2Writable.pm
+++ b/lib/PublicInbox/V2Writable.pm
@@ -1106,10 +1106,10 @@ sub sync_prepare ($$) {
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;
}
@@ -1233,6 +1233,7 @@ sub index_xap_step ($$$;$) {
"$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);
@@ -1262,6 +1263,12 @@ sub index_todo ($$$) {
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,
@@ -1278,12 +1285,6 @@ sub index_todo ($$$) {
} 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);
}
@@ -1310,6 +1311,7 @@ sub xapian_only {
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);
@@ -1350,7 +1352,7 @@ sub index_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;
@@ -1381,14 +1383,21 @@ sub index_sync {
$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) {
@@ -1400,12 +1409,16 @@ sub index_sync {
}
# 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;
diff --git a/script/public-inbox-index b/script/public-inbox-index
index 55e4f641..5909801e 100755
--- a/script/public-inbox-index
+++ b/script/public-inbox-index
@@ -90,6 +90,7 @@ EOL
$ibx_opt = { %$opt, sequential_shard => $v };
}
PublicInbox::Admin::index_inbox($ibx, undef, $ibx_opt);
+ last if $ibx_opt->{quit};
if (my $copt = $opt->{compact_opt}) {
local $copt->{jobs} = 0 if $ibx_opt->{sequential_shard};
PublicInbox::Xapcmd::run($ibx, 'compact', $copt);
^ permalink raw reply related [flat|nested] only message in thread
only message in thread, other threads:[~2020-11-27 21:33 UTC | newest]
Thread overview: (only message) (download: mbox.gz / follow: Atom feed)
-- links below jump to the message on this page --
2020-11-27 21:33 [PATCH] *index: more consistent graceful shutdown checks Eric Wong
Code repositories for project(s) associated with this public inbox
https://80x24.org/public-inbox.git
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox;
as well as URLs for read-only IMAP folder(s) and NNTP newsgroup(s).