diff options
Diffstat (limited to 'lib/PublicInbox/Xapcmd.pm')
| -rw-r--r-- | lib/PublicInbox/Xapcmd.pm | 307 |
1 files changed, 201 insertions, 106 deletions
diff --git a/lib/PublicInbox/Xapcmd.pm b/lib/PublicInbox/Xapcmd.pm index e2d67f6a..9a148ae4 100644 --- a/lib/PublicInbox/Xapcmd.pm +++ b/lib/PublicInbox/Xapcmd.pm @@ -1,15 +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 File::Basename qw(dirname); 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") @@ -21,13 +23,25 @@ sub commit_changes ($$$$) { my $reshard = $opt->{reshard}; $SIG{INT} or die 'BUG: $SIG{INT} not handled'; - my @old_shard; - my $over_chg; - - while (my ($old, $newdir) = each %$tmp) { + my (@old_shard, $over_chg); + + # Sort shards highest-to-lowest, since ->xdb_shards_flat + # determines the number of shards to load based on the max; + # and we'd rather xdb_shards_flat to momentarily fail rather + # than load out-of-date shards + my @order = sort { + my ($x) = ($a =~ m!/([0-9]+)/*\z!); + my ($y) = ($b =~ m!/([0-9]+)/*\z!); + ($y // -1) <=> ($x // -1) # we may have non-shards + } keys %$tmp; + + my ($dname) = ($order[0] =~ m!(.*/)[^/]+/*\z!); + my $mode = (stat($dname))[2]; + for my $old (@order) { next if $old eq ''; # no invalid paths - my @st = stat($old); - if (!@st && !defined($opt->{reshard})) { + my $newdir = $tmp->{$old}; + my $have_old = -e $old; + if (!$have_old && !defined($opt->{reshard})) { die "failed to stat($old): $!"; } @@ -47,35 +61,31 @@ sub commit_changes ($$$$) { next; } - if (@st) { - chmod($st[2] & 07777, $new) or die "chmod $old: $!\n"; + chmod($mode & 07777, $new) or die "chmod($new): $!\n"; + if ($have_old) { rename($old, "$new/old") or die "rename $old => $new/old: $!\n"; } rename($new, $old) or die "rename $new => $old: $!\n"; - if (@st) { - my $prev = "$old/old"; - remove_tree($prev) or - die "failed to remove $prev: $!\n"; - } + push @old_shard, "$old/old" if $have_old; } # trigger ->check_inodes in read-only daemons - syswrite($im->{lockfh}, '.') if $over_chg; + syswrite($im->{lockfh}, '.') if $over_chg && $im; remove_tree(@old_shard); $tmp = undef; if (!$opt->{-coarse_lock}) { $opt->{-skip_lock} = 1; - - if ($im->can('count_shards')) { + $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; @@ -83,16 +93,18 @@ sub commit_changes ($$$$) { } my $env = $opt->{-idx_env}; local %ENV = (%ENV, %$env) if $env; - PublicInbox::Admin::index_inbox($ibx, $im, $opt); + if ($ibx->can('eidx_sync')) { + $ibx->eidx_sync($opt); + } elsif (!$ibx->can('cidx_run')) { + PublicInbox::Admin::index_inbox($ibx, $im, $opt); + } } } 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); @@ -103,9 +115,11 @@ sub runnable_or_die ($) { which($exe) or die "$exe not found in PATH\n"; } -sub prepare_reindex ($$$) { - my ($ibx, $im, $opt) = @_; - if ($ibx->version == 1) { +sub prepare_reindex ($$) { + my ($ibx, $opt) = @_; + 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); if (my $lc = $xdb->get_metadata('last_commit')) { @@ -134,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); @@ -145,7 +160,7 @@ sub process_queue { # run in parallel: my %pids; - local %SIG = %SIG; + local @SIG{keys %SIG} = values %SIG; setup_signals(\&kill_pids, \%pids); while (@$queue) { while (scalar(keys(%pids)) < $max && scalar(@$queue)) { @@ -172,9 +187,16 @@ sub prepare_run { my ($ibx, $opt) = @_; 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; - if (my $srch = $ibx->search) { + my ($old, $misc_ok); + 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) { $old = $srch->xdir(1); + } + if (defined $old) { -d $old or die "$old does not exist\n"; } my $reshard = $opt->{reshard}; @@ -184,32 +206,33 @@ sub prepare_run { # we want temporary directories to be as deep as possible, # so v2 shards can keep "xap$SCHEMA_VERSION" on a separate FS. - if ($old && $ibx->version == 1) { + if (defined($old) && $ibx->can('version') && $ibx->version == 1) { if (defined $reshard) { warn "--reshard=$reshard ignored for v1 $ibx->{inboxdir}\n"; } - my $dir = dirname($old); + my ($dir) = ($old =~ m!(.*?/)[^/]+/*\z!); same_fs_or_die($dir, $old); my $v = PublicInbox::Search::SCHEMA_VERSION(); - my $wip = File::Temp->newdir("xapian$v-XXXXXXXX", DIR => $dir); + 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 ($old) { + } 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) { } else { warn "W: skipping unknown dir: $old/$dn\n" } } 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 @@ -220,12 +243,11 @@ sub prepare_run { $src = [ map { "$old/$_" } @old_shards ]; } foreach my $dn (0..$max_shard) { - my $tmpl = "$dn-XXXXXXXX"; - my $wip = File::Temp->newdir($tmpl, DIR => $old); + my $wip = File::Temp->newdir("$dn-XXXX", DIR => $old); 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 @@ -238,42 +260,49 @@ sub prepare_run { sub check_compact () { runnable_or_die($XAPIAN_COMPACT) } -sub _run { # with_umask callback - my ($ibx, $cb, $opt) = @_; - my $im = $ibx->importer(0); - $im->lock_acquire; - my ($tmp, $queue) = prepare_run($ibx, $opt); - - # fine-grained locking if we prepare for reindex - if (!$opt->{-coarse_lock}) { - prepare_reindex($ibx, $im, $opt); - $im->lock_release; - } - - $ibx->cleanup; - process_queue($queue, $cb, $opt); - $im->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 ||= {}); - defined(my $dir = $ibx->{inboxdir}) or die "no inboxdir defined\n"; - -d $dir or die "inboxdir=$dir does not exist\n"; - check_compact() if $opt->{compact} && $ibx->search; + my $dir; + 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->can('cidx_run') || $ibx->search); - if (!$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 ? '' : [] }; PublicInbox::SearchIdx::load_xapian_writable(); } - local %SIG = %SIG; + 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 ($$) { @@ -291,16 +320,17 @@ sub cpdb_retryable ($$) { } sub progress_pfx ($) { - my ($wip) = @_; # tempdir v2: ([0-9])+-XXXXXXXX - my @p = split('/', $wip); + my ($wip) = @_; # tempdir v2: ([0-9])+-XXXX + 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 @@ -328,18 +358,16 @@ sub compact ($$) { # cb_spawn callback } $pr->("$pfx `".join(' ', @$cmd)."'\n") if $pr; push @$cmd, $src, $dst; - my ($rd, $pid); - local %SIG = %SIG; - setup_signals(\&kill_compact, \$pid); - ($rd, $pid) = popen_rd($cmd, undef, $rdr); + local @SIG{keys %SIG} = values %SIG; + 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 ($$$;$$) { @@ -383,18 +411,96 @@ 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() or die; - my $XapianDatabase = $PublicInbox::Search::X{Database}; + my ($X, $flag) = xapian_write_prep($opt); if (ref($old) eq 'ARRAY') { - ($cur_shard) = ($new =~ m!xap[0-9]+/([0-9]+)\b!); + 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"; $reshard = $opt->{reshard}; @@ -403,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); - local %SIG = %SIG; + my $tmp = $wip; + local @SIG{keys %SIG} = values %SIG; if ($opt->{compact}) { - my $dir = dirname($new); - same_fs_or_die($dir, $new); - $ft = File::Temp->newdir("$new.compact-XXXXXX", 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; @@ -444,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; @@ -475,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 { @@ -490,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; |