diff options
Diffstat (limited to 'lib/PublicInbox/Xapcmd.pm')
| -rw-r--r-- | lib/PublicInbox/Xapcmd.pm | 223 |
1 files changed, 148 insertions, 75 deletions
diff --git a/lib/PublicInbox/Xapcmd.pm b/lib/PublicInbox/Xapcmd.pm index 44e0f8e5..9a148ae4 100644 --- a/lib/PublicInbox/Xapcmd.pm +++ b/lib/PublicInbox/Xapcmd.pm @@ -1,14 +1,17 @@ -# Copyright (C) 2018-2021 all contributors <meta@public-inbox.org> +# Copyright (C) all contributors <meta@public-inbox.org> # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt> package PublicInbox::Xapcmd; -use strict; -use PublicInbox::Spawn qw(which popen_rd nodatacow_dir); +use v5.12; +use PublicInbox::Spawn qw(which popen_rd); +use PublicInbox::Syscall; use PublicInbox::Admin qw(setup_signals); use PublicInbox::Over; +use PublicInbox::Search qw(xap_terms); use PublicInbox::SearchIdx; use File::Temp 0.19 (); # ->newdir use File::Path qw(remove_tree); use POSIX qw(WNOHANG _exit); +use PublicInbox::DS; # support testing with dev versions of Xapian which installs # commands with a version number suffix (e.g. "xapian-compact-1.5") @@ -74,15 +77,15 @@ sub commit_changes ($$$$) { $tmp = undef; if (!$opt->{-coarse_lock}) { $opt->{-skip_lock} = 1; - $im //= $ibx if $ibx->can('eidx_sync'); - if ($im->can('count_shards')) { # v2w or eidx + $im //= $ibx if $ibx->can('eidx_sync') || $ibx->can('cidx_run'); + if ($im->can('count_shards')) { # v2w, eidx, cidx my $pr = $opt->{-progress}; my $n = $im->count_shards; if (defined $reshard && $n != $reshard) { die "BUG: counted $n shards after resharding to $reshard"; } - my $prev = $im->{shards}; + my $prev = $im->{shards} // $ibx->{nshard}; if ($pr && $prev != $n) { $pr->("shard count changed: $prev => $n\n"); $im->{shards} = $n; @@ -92,7 +95,7 @@ sub commit_changes ($$$$) { local %ENV = (%ENV, %$env) if $env; if ($ibx->can('eidx_sync')) { $ibx->eidx_sync($opt); - } else { + } elsif (!$ibx->can('cidx_run')) { PublicInbox::Admin::index_inbox($ibx, $im, $opt); } } @@ -100,10 +103,8 @@ sub commit_changes ($$$$) { sub cb_spawn { my ($cb, $args, $opt) = @_; # $cb = cpdb() or compact() - my $seed = rand(0xffffffff); - my $pid = fork // die "fork: $!"; + my $pid = PublicInbox::DS::fork_persist; return $pid if $pid > 0; - srand($seed); $SIG{__DIE__} = sub { warn @_; _exit(1) }; # don't jump up stack $cb->($args, $opt); _exit(0); @@ -116,7 +117,8 @@ sub runnable_or_die ($) { sub prepare_reindex ($$) { my ($ibx, $opt) = @_; - if ($ibx->can('eidx_sync')) { # no prep needed for ExtSearchIdx + if ($ibx->can('eidx_sync') || $ibx->can('cidx_run')) { + # no prep needed for ExtSearchIdx nor CodeSearchIdx } elsif ($ibx->version == 1) { my $dir = $ibx->search->xdir(1); my $xdb = $PublicInbox::Search::X{Database}->new($dir); @@ -146,8 +148,9 @@ sub kill_pids { } sub process_queue { - my ($queue, $cb, $opt) = @_; + my ($queue, $task, $opt) = @_; my $max = $opt->{jobs} // scalar(@$queue); + my $cb = \&$task; if ($max <= 1) { while (defined(my $args = shift @$queue)) { $cb->($args, $opt); @@ -185,7 +188,9 @@ sub prepare_run { my $tmp = {}; # old shard dir => File::Temp->newdir object or undef my @queue; # ([old//src,newdir]) - list of args for cpdb() or compact() my ($old, $misc_ok); - if ($ibx->can('eidx_sync')) { + if ($ibx->can('cidx_run')) { + $old = $ibx->xdir(1); + } elsif ($ibx->can('eidx_sync')) { $misc_ok = 1; $old = $ibx->xdir(1); } elsif (my $srch = $ibx->search) { @@ -211,14 +216,14 @@ sub prepare_run { my $v = PublicInbox::Search::SCHEMA_VERSION(); my $wip = File::Temp->newdir("xapian$v-XXXX", DIR => $dir); $tmp->{$old} = $wip; - nodatacow_dir($wip->dirname); + PublicInbox::Syscall::nodatacow_dir($wip->dirname); push @queue, [ $old, $wip ]; } elsif (defined $old) { opendir my $dh, $old or die "Failed to opendir $old: $!\n"; my @old_shards; while (defined(my $dn = readdir($dh))) { if ($dn =~ /\A[0-9]+\z/) { - push @old_shards, $dn; + push(@old_shards, $dn + 0); } elsif ($dn eq '.' || $dn eq '..') { } elsif ($dn =~ /\Aover\.sqlite3/) { } elsif ($dn eq 'misc' && $misc_ok) { @@ -227,7 +232,7 @@ sub prepare_run { } } die "No Xapian shards found in $old\n" unless @old_shards; - + @old_shards = sort { $a <=> $b } @old_shards; my ($src, $max_shard); if (!defined($reshard) || $reshard == scalar(@old_shards)) { # 1:1 copy @@ -242,7 +247,7 @@ sub prepare_run { same_fs_or_die($old, $wip->dirname); my $cur = "$old/$dn"; push @queue, [ $src // $cur , $wip ]; - nodatacow_dir($wip->dirname); + PublicInbox::Syscall::nodatacow_dir($wip->dirname); $tmp->{$cur} = $wip; } # mark old shards to be unlinked @@ -255,38 +260,21 @@ sub prepare_run { sub check_compact () { runnable_or_die($XAPIAN_COMPACT) } -sub _run { # with_umask callback - my ($ibx, $cb, $opt) = @_; - my $im = $ibx->can('importer') ? $ibx->importer(0) : undef; - ($im // $ibx)->lock_acquire; - my ($tmp, $queue) = prepare_run($ibx, $opt); - - # fine-grained locking if we prepare for reindex - if (!$opt->{-coarse_lock}) { - prepare_reindex($ibx, $opt); - ($im // $ibx)->lock_release; - } - - $ibx->cleanup if $ibx->can('cleanup'); - process_queue($queue, $cb, $opt); - ($im // $ibx)->lock_acquire if !$opt->{-coarse_lock}; - commit_changes($ibx, $im, $tmp, $opt); -} - sub run { my ($ibx, $task, $opt) = @_; # task = 'cpdb' or 'compact' - my $cb = \&$task; PublicInbox::Admin::progress_prepare($opt ||= {}); my $dir; - for my $fld (qw(inboxdir topdir)) { + for my $fld (qw(inboxdir topdir cidx_dir)) { my $d = $ibx->{$fld} // next; -d $d or die "$fld=$d does not exist\n"; $dir = $d; last; } - check_compact() if $opt->{compact} && $ibx->search; + check_compact() if $opt->{compact} && + ($ibx->can('cidx_run') || $ibx->search); - if (!$ibx->can('eidx_sync') && !$opt->{-coarse_lock}) { + if (!$ibx->can('eidx_sync') && $ibx->can('version') && + !$opt->{-coarse_lock}) { # per-epoch ranges for v2 # v1:{ from => $OID }, v2:{ from => [ $OID, $OID, $OID ] } } $opt->{reindex} = { from => $ibx->version == 1 ? '' : [] }; @@ -295,7 +283,26 @@ sub run { local @SIG{keys %SIG} = values %SIG; setup_signals(); - $ibx->with_umask(\&_run, $ibx, $cb, $opt); + my $restore = $ibx->with_umask; + + my $im = $ibx->can('importer') ? $ibx->importer(0) : undef; + ($im // $ibx)->lock_acquire; + my ($tmp, $queue) = prepare_run($ibx, $opt); + + # fine-grained locking if we prepare for reindex + if (!$opt->{-coarse_lock}) { + prepare_reindex($ibx, $opt); + ($im // $ibx)->lock_release; + } + + $ibx->cleanup if $ibx->can('cleanup'); + if ($task eq 'cpdb' && $opt->{reshard} && $ibx->can('cidx_run')) { + cidx_reshard($ibx, $queue, $opt); + } else { + process_queue($queue, $task, $opt); + } + ($im // $ibx)->lock_acquire if !$opt->{-coarse_lock}; + commit_changes($ibx, $im, $tmp, $opt); } sub cpdb_retryable ($$) { @@ -314,15 +321,16 @@ sub cpdb_retryable ($$) { sub progress_pfx ($) { my ($wip) = @_; # tempdir v2: ([0-9])+-XXXX - my @p = split('/', $wip); + my @p = split(m'/', $wip); - # return "xap15/0" for v2, or "xapian15" for v1: - ($p[-1] =~ /\A([0-9]+)/) ? "$p[-2]/$1" : $p[-1]; + # "basename(inboxdir)/xap15/0" for v2, + # "basename(inboxdir)/xapian15" for v1: + ($p[-1] =~ /\A([0-9]+)/) ? "$p[-3]/$p[-2]/$1" : "$p[-2]/$p[-1]"; } sub kill_compact { # setup_signals callback - my ($sig, $pidref) = @_; - kill($sig, $$pidref) if defined($$pidref); + my ($sig, $ioref) = @_; + kill($sig, $$ioref->attached_pid // return) if defined($$ioref); } # xapian-compact wrapper @@ -350,18 +358,16 @@ sub compact ($$) { # cb_spawn callback } $pr->("$pfx `".join(' ', @$cmd)."'\n") if $pr; push @$cmd, $src, $dst; - my ($rd, $pid); local @SIG{keys %SIG} = values %SIG; - setup_signals(\&kill_compact, \$pid); - ($rd, $pid) = popen_rd($cmd, undef, $rdr); + setup_signals(\&kill_compact, \my $rd); + $rd = popen_rd($cmd, undef, $rdr); while (<$rd>) { if ($pr) { s/\r/\r$pfx /g; $pr->("$pfx $_"); } } - waitpid($pid, 0); - die "@$cmd failed: \$?=$?\n" if $?; + $rd->close or die "@$cmd failed: \$?=$?\n"; } sub cpdb_loop ($$$;$$) { @@ -405,17 +411,95 @@ sub cpdb_loop ($$$;$$) { } while (cpdb_retryable($src, $pfx)); } +sub xapian_write_prep ($) { + my ($opt) = @_; + PublicInbox::SearchIdx::load_xapian_writable(); + my $flag = eval($PublicInbox::Search::Xap.'::DB_CREATE()'); + die if $@; + $flag |= $PublicInbox::SearchIdx::DB_NO_SYNC if !$opt->{fsync}; + (\%PublicInbox::Search::X, $flag); +} + +sub compact_tmp_shard ($) { + my ($wip) = @_; + my $new = $wip->dirname; + my ($dir) = ($new =~ m!(.*?/)[^/]+/*\z!); + same_fs_or_die($dir, $new); + my $ft = File::Temp->newdir("$new.compact-XXXX", DIR => $dir); + PublicInbox::Syscall::nodatacow_dir($ft->dirname); + $ft; +} + +sub cidx_reshard { # not docid based + my ($cidx, $queue, $opt) = @_; + my ($X, $flag) = xapian_write_prep($opt); + my $src = $cidx->xdb; + delete($cidx->{xdb}) == $src or die "BUG: xdb != $src"; + my $pfx = $opt->{-progress_pfx} = progress_pfx($cidx->xdir.'/0'); + my $pr = $opt->{-progress}; + my $pr_data = { pr => $pr, pfx => $pfx, nr => 0 } if $pr; + local @SIG{keys %SIG} = values %SIG; + + # like copydatabase(1), be sure we don't overwrite anything in case + # of other bugs: + setup_signals() if $opt->{compact}; + my @tmp; + my @dst = map { + my $wip = $_->[1]; + my $tmp = $opt->{compact} ? compact_tmp_shard($wip) : $wip; + push @tmp, $tmp; + $X->{WritableDatabase}->new($tmp->dirname, $flag); + } @$queue; + my $l = $src->get_metadata('indexlevel'); + $dst[0]->set_metadata('indexlevel', $l) if $l eq 'medium'; + my $fmt; + if ($pr_data) { + my $tot = $src->get_doccount; + $fmt = "$pfx % ".length($tot)."u/$tot\n"; + $pr->("$pfx copying $tot documents\n"); + } + my $cur = $src->postlist_begin(''); + my $end = $src->postlist_end(''); + my $git_dir_hash = $cidx->can('git_dir_hash'); + my ($n, $nr); + for (; $cur != $end; $cur++) { + my $doc = $src->get_document($cur->get_docid); + if (my @cmt = xap_terms('Q', $doc)) { + $n = hex(substr($cmt[0], 0, 8)) % scalar(@dst); + warn "W: multi-commit: @cmt" if scalar(@cmt) != 1; + } elsif (my @P = xap_terms('P', $doc)) { + $n = $git_dir_hash->($P[0]) % scalar(@dst); + warn "W: multi-path @P " if scalar(@P) != 1; + } else { + warn "W: skipped, no terms in ".$cur->get_docid; + next; + } + $dst[$n]->add_document($doc); + $pr->(sprintf($fmt, $nr)) if $pr_data && !(++$nr & 1023); + } + return if !$opt->{compact}; + $src = undef; + @dst = (); # flushes and closes + my @q; + for my $tmp (@tmp) { + my $arg = shift @$queue // die 'BUG: $queue empty'; + my $wip = $arg->[1] // die 'BUG: no $wip'; + push @q, [ "$tmp", $wip ]; + } + delete $opt->{-progress_pfx}; + process_queue(\@q, 'compact', $opt); +} + # Like copydatabase(1), this is horribly slow; and it doesn't seem due # to the overhead of Perl. sub cpdb ($$) { # cb_spawn callback my ($args, $opt) = @_; - my ($old, $newdir) = @$args; - my $new = $newdir->dirname; + my ($old, $wip) = @$args; my ($src, $cur_shard); my $reshard; - PublicInbox::SearchIdx::load_xapian_writable(); - my $XapianDatabase = $PublicInbox::Search::X{Database}; + my ($X, $flag) = xapian_write_prep($opt); if (ref($old) eq 'ARRAY') { + my $new = $wip->dirname; ($cur_shard) = ($new =~ m!(?:xap|ei)[0-9]+/([0-9]+)\b!); defined $cur_shard or die "BUG: could not extract shard # from $new"; @@ -425,36 +509,27 @@ sub cpdb ($$) { # cb_spawn callback # resharding, M:N copy means have full read access foreach (@$old) { if ($src) { - my $sub = $XapianDatabase->new($_); + my $sub = $X->{Database}->new($_); $src->add_database($sub); } else { - $src = $XapianDatabase->new($_); + $src = $X->{Database}->new($_); } } } else { - $src = $XapianDatabase->new($old); + $src = $X->{Database}->new($old); } - my ($tmp, $ft); + my $tmp = $wip; local @SIG{keys %SIG} = values %SIG; if ($opt->{compact}) { - my ($dir) = ($new =~ m!(.*?/)[^/]+/*\z!); - same_fs_or_die($dir, $new); - $ft = File::Temp->newdir("$new.compact-XXXX", DIR => $dir); + $tmp = compact_tmp_shard($wip); setup_signals(); - $tmp = $ft->dirname; - nodatacow_dir($tmp); - } else { - $tmp = $new; } # like copydatabase(1), be sure we don't overwrite anything in case # of other bugs: - my $flag = eval($PublicInbox::Search::Xap.'::DB_CREATE()'); - die if $@; - my $XapianWritableDatabase = $PublicInbox::Search::X{WritableDatabase}; - $flag |= $PublicInbox::SearchIdx::DB_NO_SYNC if !$opt->{fsync}; - my $dst = $XapianWritableDatabase->new($tmp, $flag); + my $new = $wip->dirname; + my $dst = $X->{WritableDatabase}->new($tmp->dirname, $flag); my $pr = $opt->{-progress}; my $pfx = $opt->{-progress_pfx} = progress_pfx($new); my $pr_data = { pr => $pr, pfx => $pfx, nr => 0 } if $pr; @@ -466,11 +541,10 @@ sub cpdb ($$) { # cb_spawn callback $dst->set_metadata('last_commit', $lc) if $lc; # only the first xapian shard (0) gets 'indexlevel' - if ($new =~ m!(?:xapian[0-9]+|xap[0-9]+/0)\b!) { + if ($new =~ m!/(?:xapian[0-9]+|(?:ei|xap)[0-9]+/0)\b!) { my $l = $src->get_metadata('indexlevel'); - if ($l eq 'medium') { + $l eq 'medium' and $dst->set_metadata('indexlevel', $l); - } } if ($pr_data) { my $tot = $src->get_doccount; @@ -497,7 +571,7 @@ sub cpdb ($$) { # cb_spawn callback # individually. $src = undef; foreach (@$old) { - my $old = $XapianDatabase->new($_); + my $old = $X->{Database}->new($_); cpdb_loop($old, $dst, $pr_data, $cur_shard, $reshard); } } else { @@ -512,7 +586,6 @@ sub cpdb ($$) { # cb_spawn callback # this is probably the best place to do xapian-compact # since $dst isn't readable by HTTP or NNTP clients, yet: compact([ $tmp, $new ], $opt); - remove_tree($tmp) or die "failed to remove $tmp: $!\n"; } 1; |