topdir => $dir,
creat => $opt->{creat},
ibx_map => {}, # (newsgroup//inboxdir) => $ibx
- ibx_list => [],
+ ibx_active => [], # by config section order
+ ibx_known => [], # by config section order
indexlevel => $l,
transact_bytes => 0,
total_bytes => 0,
parallel => 1,
lock_path => "$dir/ei.lock",
}, __PACKAGE__;
- $self->{shards} = $self->count_shards || nproc_shards($opt->{creat});
+ $self->{shards} = $self->count_shards ||
+ nproc_shards({ nproc => $opt->{jobs} });
my $oidx = PublicInbox::OverIdx->new("$self->{xpfx}/over.sqlite3");
$self->{-no_fsync} = $oidx->{-no_fsync} = 1 if !$opt->{fsync};
$self->{oidx} = $oidx;
}
sub attach_inbox {
- my ($self, $ibx) = @_;
+ my ($self, $ibx, $types) = @_;
$self->{ibx_map}->{$ibx->eidx_key} //= do {
- push @{$self->{ibx_list}}, $ibx;
+ delete $self->{-ibx_ary_known}; # invalidate cache
+ delete $self->{-ibx_ary_active}; # invalidate cache
+ $types //= [ qw(active known) ];
+ for my $t (@$types) {
+ push @{$self->{"ibx_$t"}}, $ibx;
+ }
$ibx;
}
}
sub _ibx_attach { # each_inbox callback
- my ($ibx, $self) = @_;
- attach_inbox($self, $ibx);
+ my ($ibx, $self, $types) = @_;
+ attach_inbox($self, $ibx, $types);
}
sub attach_config {
- my ($self, $cfg) = @_;
+ my ($self, $cfg, $ibxs) = @_;
$self->{cfg} = $cfg;
- $cfg->each_inbox(\&_ibx_attach, $self);
+ my $types;
+ if ($ibxs) {
+ for my $ibx (@$ibxs) {
+ $self->{ibx_map}->{$ibx->eidx_key} //= do {
+ push @{$self->{ibx_active}}, $ibx;
+ push @{$self->{ibx_known}}, $ibx;
+ }
+ }
+ # invalidate cache
+ delete $self->{-ibx_ary_known};
+ delete $self->{-ibx_ary_active};
+ $types = [ 'known' ];
+ }
+ $types //= [ qw(known active) ];
+ $cfg->each_inbox(\&_ibx_attach, $self, $types);
}
sub check_batch_limit ($) {
${$req->{need_checkpoint}} = 1 if $n >= $self->{batch_bytes};
}
+sub apply_boost ($$) {
+ my ($req, $smsg) = @_;
+ my $id2pos = $req->{id2pos}; # index in ibx_sorted
+ my $xr3 = $req->{self}->{oidx}->get_xref3($smsg->{num}, 1);
+ @$xr3 = sort {
+ $id2pos->{$a->[0]} <=> $id2pos->{$b->[0]}
+ ||
+ $a->[1] <=> $b->[1] # break ties with {xnum}
+ } @$xr3;
+ my $top_blob = unpack('H*', $xr3->[0]->[2]);
+ my $new_smsg = $req->{new_smsg};
+ return if $top_blob ne $new_smsg->{blob}; # loser
+
+ # replace the old smsg with the more boosted one
+ $new_smsg->{num} = $smsg->{num};
+ $new_smsg->populate($req->{eml}, $req);
+ $req->{self}->{oidx}->add_overview($req->{eml}, $new_smsg);
+}
+
sub do_xpost ($$) {
my ($req, $smsg) = @_;
my $self = $req->{self};
my $xnum = $req->{xnum};
$self->{oidx}->add_xref3($docid, $xnum, $oid, $eidx_key);
$idx->ipc_do('add_eidx_info', $docid, $eidx_key, $eml);
- check_batch_limit($req);
+ apply_boost($req, $smsg) if $req->{boost_in_use};
} else { # 'd'
my $rm_eidx_info;
my $nr = $self->{oidx}->remove_xref3($docid, $oid, $eidx_key,
my ($self, $sync, $smsg) = @_;
my $ibx_id = delete($smsg->{ibx_id}) // die '{ibx_id} unset';
my $pos = $sync->{id2pos}->{$ibx_id} // die "$ibx_id no pos";
- $self->{ibx_list}->[$pos] // die "BUG: ibx for $smsg->{blob} not mapped"
+ $self->{-ibx_ary_known}->[$pos] //
+ die "BUG: ibx for $smsg->{blob} not mapped"
}
sub _fd_constrained ($) {
chomp($soft = `sh -c 'ulimit -n'`);
}
if (defined($soft)) {
- my $want = scalar(@{$self->{ibx_list}}) + 64; # estimate
+ # $want is an estimate
+ my $want = scalar(@{$self->{ibx_active}}) + 64;
my $ret = $want > $soft;
if ($ret) {
warn <<EOF;
return;
}
- # we sort {xr3r} in the reverse order of {ibx_list} so we can
+ # we sort {xr3r} in the reverse order of ibx_sorted so we can
# hit the common case in _reindex_finalize without rereading
# from git (or holding multiple messages in memory).
- my $id2pos = $sync->{id2pos}; # index in {ibx_list}
+ my $id2pos = $sync->{id2pos}; # index in ibx_sorted
@$xr3 = sort {
$id2pos->{$b->[0]} <=> $id2pos->{$a->[0]}
||
undef;
}
+sub ibx_sorted ($$) {
+ my ($self, $type) = @_;
+ $self->{"-ibx_ary_$type"} //= do {
+ # highest boost first, stable for config-ordering tiebreaker
+ use sort 'stable';
+ [ sort {
+ ($b->{boost} // 0) <=> ($a->{boost} // 0)
+ } @{$self->{'ibx_'.$type} // die "BUG: $type unknown"} ];
+ }
+}
+
+sub prep_id2pos ($) {
+ my ($self) = @_;
+ my %id2pos;
+ my $pos = 0;
+ $id2pos{$_->{-ibx_id}} = $pos++ for (@{ibx_sorted($self, 'known')});
+ \%id2pos;
+}
+
sub eidxq_process ($$) { # for reindexing
my ($self, $sync) = @_;
my $max = $dbh->selectrow_array('SELECT MAX(docid) FROM eidxq');
$pr->("Xapian indexing $min..$max (total=$tot)\n");
}
- $sync->{id2pos} //= do {
- my %id2pos;
- my $pos = 0;
- $id2pos{$_->{-ibx_id}} = $pos++ for @{$self->{ibx_list}};
- \%id2pos;
- };
+ $sync->{id2pos} //= prep_id2pos($self);
my ($del, $iter);
restart:
$del = $dbh->prepare('DELETE FROM eidxq WHERE docid = ?');
$fetching = $min = $docid;
my $smsg = $ibx->over->get_art($xnum);
- my $oidhex = unpack('H*', $oidbin);
my $err;
if (!$smsg) {
$err = 'stale';
- } elsif ($smsg->{blob} ne $oidhex) {
+ } elsif (pack('H*', $smsg->{blob}) ne $oidbin) {
$err = "mismatch (!= $smsg->{blob})";
} else {
next; # likely, all good
}
# current_info already has eidx_key
+ my $oidhex = unpack('H*', $oidbin);
warn "$xnum:$oidhex (#$docid): $err\n";
my $del = $self->{oidx}->dbh->prepare_cached(<<'');
DELETE FROM xref3 WHERE ibx_id = ? AND xnum = ? AND oidbin = ?
warn "E: aborting --reindex\n";
return;
}
- for my $ibx (@{$self->{ibx_list}}) {
+ for my $ibx (@{ibx_sorted($self, 'active')}) {
_reindex_inbox($self, $sync, $ibx);
last if $sync->{quit};
}
warn $err, "\n" if defined($err);
}
+sub dd_smsg { # git->cat_async callback
+ my ($bref, $oid, $type, $size, $dd) = @_;
+ my $smsg = $dd->{smsg} // die 'BUG: dd->{smsg} missing';
+ my $self = $dd->{self} // die 'BUG: {self} missing';
+ my $per_mid = $dd->{per_mid} // die 'BUG: {per_mid} missing';
+ if ($type eq 'missing') {
+ _blob_missing($dd, $smsg);
+ } elsif (!is_bad_blob($oid, $type, $size, $smsg->{blob})) {
+ local $self->{current_info} = "$self->{current_info} $oid";
+ my $chash = content_hash(PublicInbox::Eml->new($bref));
+ push(@{$per_mid->{dd_chash}->{$chash}}, $smsg);
+ }
+ return if $per_mid->{last_smsg} != $smsg;
+ while (my ($chash, $ary) = each %{$per_mid->{dd_chash}}) {
+ my $keep = shift @$ary;
+ next if !scalar(@$ary);
+ $per_mid->{sync}->{dedupe_cull} += scalar(@$ary);
+ print STDERR
+ "# <$keep->{mid}> keeping #$keep->{num}, dropping ",
+ join(', ', map { "#$_->{num}" } @$ary),"\n";
+ next if $per_mid->{sync}->{-opt}->{'dry-run'};
+ my $oidx = $self->{oidx};
+ for my $smsg (@$ary) {
+ my $gone = $smsg->{num};
+ $oidx->merge_xref3($keep->{num}, $gone, $smsg->{blob});
+ $self->idx_shard($gone)->ipc_do('xdb_remove', $gone);
+ $oidx->delete_by_num($gone);
+ }
+ }
+}
+
+sub eidx_dedupe ($$$) {
+ my ($self, $sync, $msgids) = @_;
+ $sync->{dedupe_cull} = 0;
+ my $candidates = 0;
+ my $nr_mid = 0;
+ return unless eidxq_lock_acquire($self);
+ my ($iter, $cur_mid);
+ my $min_id = 0;
+ my $idx = 0;
+ my ($max_id) = $self->{oidx}->dbh->selectrow_array(<<EOS);
+SELECT MAX(id) FROM msgid
+EOS
+ local $sync->{-regen_fmt} = "dedupe %u/$max_id\n";
+
+ # note: we could write this query more intelligently,
+ # but that causes lock contention with read-only processes
+dedupe_restart:
+ $cur_mid = $msgids->[$idx];
+ if ($cur_mid eq '') { # all Message-IDs
+ $iter = $self->{oidx}->dbh->prepare(<<EOS);
+SELECT mid,id FROM msgid WHERE id > ? ORDER BY id ASC
+EOS
+ $iter->execute($min_id);
+ } else {
+ $iter = $self->{oidx}->dbh->prepare(<<EOS);
+SELECT mid,id FROM msgid WHERE mid = ? AND id > ? ORDER BY id ASC
+EOS
+ $iter->execute($cur_mid, $min_id);
+ }
+ while (my ($mid, $id) = $iter->fetchrow_array) {
+ last if $sync->{quit};
+ $self->{current_info} = "dedupe $mid";
+ ${$sync->{nr}} = $min_id = $id;
+ my ($prv, @smsg);
+ while (my $x = $self->{oidx}->next_by_mid($mid, \$id, \$prv)) {
+ push @smsg, $x;
+ }
+ next if scalar(@smsg) < 2;
+ my $per_mid = {
+ dd_chash => {}, # chash => [ary of smsgs]
+ last_smsg => $smsg[-1],
+ sync => $sync
+ };
+ $nr_mid++;
+ $candidates += scalar(@smsg) - 1;
+ for my $smsg (@smsg) {
+ my $dd = {
+ per_mid => $per_mid,
+ smsg => $smsg,
+ self => $self,
+ };
+ $self->git->cat_async($smsg->{blob}, \&dd_smsg, $dd);
+ }
+ # need to wait on every single one @smsg contents can get
+ # invalidated inside dd_smsg for messages with multiple
+ # Message-IDs.
+ $self->git->async_wait_all;
+
+ if (checkpoint_due($sync)) {
+ undef $iter;
+ reindex_checkpoint($self, $sync);
+ goto dedupe_restart;
+ }
+ }
+ goto dedupe_restart if defined($msgids->[++$idx]);
+
+ my $n = delete $sync->{dedupe_cull};
+ if (my $pr = $sync->{-opt}->{-progress}) {
+ $pr->("culled $n/$candidates candidates ($nr_mid msgids)\n");
+ }
+ ${$sync->{nr}} = 0;
+}
+
sub eidx_sync { # main entry point
my ($self, $opt) = @_;
my $warn_cb = $SIG{__WARN__} || \&CORE::warn;
local $self->{current_info} = '';
local $SIG{__WARN__} = sub {
+ return if PublicInbox::Eml::warn_ignore(@_);
$warn_cb->($self->{current_info}, ': ', @_);
};
$self->idx_init($opt); # acquire lock via V2Writable::_idx_init
local $SIG{QUIT} = $quit;
local $SIG{INT} = $quit;
local $SIG{TERM} = $quit;
- for my $ibx (@{$self->{ibx_list}}) {
+ for my $ibx (@{ibx_sorted($self, 'known')}) {
$ibx->{-ibx_id} //= $self->{oidx}->ibx_id($ibx->eidx_key);
}
+
+ if (scalar(grep { defined($_->{boost}) } @{$self->{ibx_known}})) {
+ $sync->{id2pos} //= prep_id2pos($self);
+ $sync->{boost_in_use} = 1;
+ }
+
+ if (my $msgids = delete($opt->{dedupe})) {
+ local $sync->{checkpoint_unlocks} = 1;
+ eidx_dedupe($self, $sync, $msgids);
+ }
if (delete($opt->{reindex})) {
local $sync->{checkpoint_unlocks} = 1;
eidx_reindex($self, $sync);
# don't use $_ here, it'll get clobbered by reindex_checkpoint
if ($opt->{scan} // 1) {
- for my $ibx (@{$self->{ibx_list}}) {
+ for my $ibx (@{ibx_sorted($self, 'active')}) {
last if $sync->{quit};
sync_inbox($self, $sync, $ibx);
}
}
undef $dh;
}
- for my $ibx (@{$self->{ibx_list}}) {
+ for my $ibx (@{ibx_sorted($self, 'active')}) {
# create symlinks for multi-pack-index
$git_midx += symlink_packs($ibx, $pd);
# add new lines to our alternates file
my $pr = $self->{-watch_sync}->{-opt}->{-progress};
$pr->('reloading ...') if $pr;
delete $self->{-resync_queue};
- @{$self->{ibx_list}} = ();
+ delete $self->{-ibx_ary_known};
+ delete $self->{-ibx_ary_active};
+ $self->{ibx_known} = [];
+ $self->{ibx_active} = [];
%{$self->{ibx_map}} = ();
delete $self->{-watch_sync}->{id2pos};
my $cfg = PublicInbox::Config->new;
sub eidx_resync_start ($) { # -extindex --watch SIGUSR1 handler
my ($self) = @_;
- $self->{-resync_queue} //= [ @{$self->{ibx_list}} ];
+ $self->{-resync_queue} //= [ @{ibx_sorted($self, 'active')} ];
PublicInbox::DS::requeue($self); # trigger our ->event_step
}
require PublicInbox::Sigfd;
my $idler = PublicInbox::InboxIdle->new($self->{cfg});
if (!$self->{cfg}) {
- $idler->watch_inbox($_) for @{$self->{ibx_list}};
+ $idler->watch_inbox($_) for (@{ibx_sorted($self, 'active')});
+ }
+ for my $ibx (@{ibx_sorted($self, 'active')}) {
+ $ibx->subscribe_unlock(__PACKAGE__, $self)
}
- $_->subscribe_unlock(__PACKAGE__, $self) for @{$self->{ibx_list}};
my $pr = $opt->{-progress};
$pr->("performing initial scan ...\n") if $pr;
my $sync = eidx_sync($self, $opt); # initial sync
my $oldset = PublicInbox::DS::block_signals();
local $self->{current_info} = '';
my $cb = $SIG{__WARN__} || \&CORE::warn;
- local $SIG{__WARN__} = sub { $cb->($self->{current_info}, ': ', @_) };
+ local $SIG{__WARN__} = sub {
+ return if PublicInbox::Eml::warn_ignore(@_);
+ $cb->($self->{current_info}, ': ', @_);
+ };
my $sig = {
HUP => sub { eidx_reload($self, $idler) },
USR1 => sub { eidx_resync_start($self) },