X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FV2Writable.pm;h=a029fe4c4075029115f330d1a692014c60b067b3;hb=8b1950055d51d4360e596446e5ac0f41008e357d;hp=e1c9a393a36f0fa12b3a04e804984f6981c03223;hpb=0faddbbfecaa784c584d3a625628c288fe9316c7;p=public-inbox.git diff --git a/lib/PublicInbox/V2Writable.pm b/lib/PublicInbox/V2Writable.pm index e1c9a393..a029fe4c 100644 --- a/lib/PublicInbox/V2Writable.pm +++ b/lib/PublicInbox/V2Writable.pm @@ -122,7 +122,7 @@ sub new { rotate_bytes => int((1024 * 1024 * 1024) / $PACKING_FACTOR), last_commit => [], # git epoch -> commit }; - $self->{over}->{-no_sync} = 1 if $v2ibx->{-no_sync}; + $self->{over}->{-no_fsync} = 1 if $v2ibx->{-no_fsync}; $self->{shards} = count_shards($self) || nproc_shards($creat); bless $self, $class; } @@ -197,7 +197,7 @@ sub _add { sub v2_num_for { my ($self, $mime) = @_; - my $mids = mids($mime->header_obj); + my $mids = mids($mime); if (@$mids) { my $mid = $mids->[0]; my $num = $self->{mm}->mid_insert($mid); @@ -244,28 +244,27 @@ sub v2_num_for { } sub v2_num_for_harder { - my ($self, $mime) = @_; + my ($self, $eml) = @_; - my $hdr = $mime->header_obj; - my $dig = content_digest($mime); - my $mid0 = PublicInbox::Import::digest2mid($dig, $hdr); + my $dig = content_digest($eml); + my $mid0 = PublicInbox::Import::digest2mid($dig, $eml); my $num = $self->{mm}->mid_insert($mid0); unless (defined $num) { # it's hard to spoof the last Received: header - my @recvd = $hdr->header_raw('Received'); + my @recvd = $eml->header_raw('Received'); $dig->add("Received: $_") foreach (@recvd); - $mid0 = PublicInbox::Import::digest2mid($dig, $hdr); + $mid0 = PublicInbox::Import::digest2mid($dig, $eml); $num = $self->{mm}->mid_insert($mid0); # fall back to a random Message-ID and give up determinism: until (defined($num)) { $dig->add(rand); - $mid0 = PublicInbox::Import::digest2mid($dig, $hdr); + $mid0 = PublicInbox::Import::digest2mid($dig, $eml); warn "using random Message-ID <$mid0> as fallback\n"; $num = $self->{mm}->mid_insert($mid0); } } - PublicInbox::Import::append_mid($hdr, $mid0); + PublicInbox::Import::append_mid($eml, $mid0); ($num, $mid0); } @@ -293,7 +292,7 @@ sub _idx_init { # with_umask callback # for SQLite: my $mm = $self->{mm} = PublicInbox::Msgmap->new_file( "$self->{ibx}->{inboxdir}/msgmap.sqlite3", - $self->{ibx}->{-no_sync} ? 2 : 1); + $self->{ibx}->{-no_fsync} ? 2 : 1); $mm->{dbh}->begin_work; } @@ -384,7 +383,7 @@ sub rewrite_internal ($$;$$$) { my $over = $self->{over}; my $chashes = content_hashes($old_eml); my $removed = []; - my $mids = mids($old_eml->header_obj); + my $mids = mids($old_eml); # We avoid introducing new blobs into git since the raw content # can be slightly different, so we do not need the user-supplied @@ -514,9 +513,7 @@ sub _check_mids_match ($$$) { # Message-IDs are pretty complex and rethreading hasn't been fully # implemented, yet. sub check_mids_match ($$) { - my ($old_mime, $new_mime) = @_; - my $old = $old_mime->header_obj; - my $new = $new_mime->header_obj; + my ($old, $new) = @_; _check_mids_match(mids($old), mids($new), 'Message-ID(s)'); _check_mids_match(references($old), references($new), 'References/In-Reply-To'); @@ -878,7 +875,8 @@ sub reindex_checkpoint ($$) { $self->{ibx}->git->cleanup; # *async_wait ${$sync->{need_checkpoint}} = 0; - $sync->{mm_tmp}->atfork_prepare; + my $mm_tmp = $sync->{mm_tmp}; + $mm_tmp->atfork_prepare if $mm_tmp; $self->done; # release lock if (my $pr = $sync->{-opt}->{-progress}) { @@ -887,16 +885,16 @@ sub reindex_checkpoint ($$) { # allow -watch or -mda to write... $self->idx_init; # reacquire lock - $sync->{mm_tmp}->atfork_parent; + $mm_tmp->atfork_parent if $mm_tmp; } sub index_oid { # cat_async callback my ($bref, $oid, $type, $size, $arg) = @_; return if $size == 0; # purged my ($num, $mid0); - my $mime = PublicInbox::Eml->new($$bref); - my $mids = mids($mime->header_obj); - my $chash = content_hash($mime); + my $eml = PublicInbox::Eml->new($$bref); + my $mids = mids($eml); + my $chash = content_hash($eml); my $self = $arg->{v2w}; if (scalar(@$mids) == 0) { @@ -960,8 +958,8 @@ sub index_oid { # cat_async callback blob => $oid, mid => $mid0, }, 'PublicInbox::Smsg'; - $smsg->populate($mime, $arg); - if (do_idx($self, $bref, $mime, $smsg)) { + $smsg->populate($eml, $arg); + if (do_idx($self, $bref, $eml, $smsg)) { ${$arg->{need_checkpoint}} = 1; } } @@ -1088,7 +1086,10 @@ sub sync_prepare ($$$) { } $all->cat_async_wait; } - return 0 if (!$regen_max && !keys(%{$self->{unindex_range}})); + if (!$regen_max && !keys(%{$self->{unindex_range}})) { + $sync->{-regen_fmt} = "%u/?\n"; + return 0; + } # reindex should NOT see new commits anymore, if we do, # it's a problem and we need to notice it via die() @@ -1113,7 +1114,7 @@ sub unindex_oid ($$;$) { # git->cat_async callback my $self = $sync->{v2w}; my $unindexed = $sync->{in_unindex} ? $sync->{unindexed} : undef; my $mm = $self->{mm}; - my $mids = mids(PublicInbox::Eml->new($bref)->header_obj); + my $mids = mids(PublicInbox::Eml->new($bref)); undef $$bref; my $over = $self->{over}; foreach my $mid (@$mids) { @@ -1180,6 +1181,38 @@ sub sync_ranges ($$$) { $ranges; } +sub index_xap_only { # git->cat_async callback + my ($bref, $oid, $type, $size, $smsg) = @_; + my $self = $smsg->{v2w}; + my $idx = idx_shard($self, $smsg->{num} % $self->{shards}); + $smsg->{raw_bytes} = $size; + $idx->index_raw($bref, undef, $smsg); + $self->{transact_bytes} += $size; +} + +sub index_xap_step ($$$;$) { + my ($self, $sync, $beg, $step) = @_; + my $ibx = $self->{ibx}; + my $all = $ibx->git; + my $over = $ibx->over; + my $batch_bytes = $PublicInbox::SearchIdx::BATCH_BYTES; + $step //= $self->{shards}; + my $end = $sync->{art_end}; + if (my $pr = $sync->{-opt}->{-progress}) { + $pr->("Xapian indexlevel=$ibx->{indexlevel} ". + "$beg..$end (% $step)\n"); + } + for (my $num = $beg; $num <= $end; $num += $step) { + my $smsg = $over->get_art($num) or next; + $smsg->{v2w} = $self; + $all->cat_async($smsg->{blob}, \&index_xap_only, $smsg); + if ($self->{transact_bytes} >= $batch_bytes) { + ${$sync->{nr}} = $num; + reindex_checkpoint($self, $sync); + } + } +} + sub index_epoch ($$$) { my ($self, $sync, $i) = @_; @@ -1213,14 +1246,46 @@ sub index_epoch ($$$) { update_last_commit($self, $git, $i, $stk->{latest_cmt}); } +sub xapian_only { + my ($self, $opt, $sync) = @_; + my $seq = $opt->{sequentialshard}; + local $self->{parallel} = 0 if $seq; + $self->idx_init($opt); # acquire lock + if (my $art_end = $self->{ibx}->mm->max) { + $sync //= { + need_checkpoint => \(my $bool = 0), + -opt => $opt, + v2w => $self, + nr => \(my $nr = 0), + -regen_fmt => "%u/?\n", + }; + $sync->{art_end} = $art_end; + if ($seq || !$self->{parallel}) { + my $shard_end = $self->{shards} - 1; + index_xap_step($self, $sync, $_) for (0..$shard_end); + } else { # parallel (maybe) + index_xap_step($self, $sync, 0, 1); + } + } + $self->{ibx}->git->cat_async_wait; + $self->done; +} + # public, called by public-inbox-index sub index_sync { my ($self, $opt) = @_; - $opt ||= {}; + $opt //= $_[1] //= {}; + goto \&xapian_only if $opt->{xapianonly}; + my $pr = $opt->{-progress}; my $epoch_max; my $latest = git_dir_latest($self, \$epoch_max); return unless defined $latest; + + my $seq = $opt->{sequentialshard}; + my $idxlevel = $self->{ibx}->{indexlevel}; + local $self->{ibx}->{indexlevel} = 'basic' if $seq; + $self->idx_init($opt); # acquire lock fill_alternates($self, $epoch_max); $self->{over}->rethread_prepare($opt); @@ -1247,13 +1312,19 @@ sub index_sync { } # work forwards through history index_epoch($self, $sync, $_) for (0..$epoch_max); + $self->{over}->rethread_done($opt); $self->done; if (my $nr = $sync->{nr}) { my $pr = $sync->{-opt}->{-progress}; $pr->('all.git '.sprintf($sync->{-regen_fmt}, $$nr)) if $pr; } - $self->{over}->rethread_done($opt); + + if ($seq) { # deal with Xapian shards sequentially + $self->{ibx}->{indexlevel} = $idxlevel; + delete $sync->{mm_tmp}; + xapian_only($self, $opt, $sync); + } # reindex does not pick up new changes, so we rerun w/o it: if ($opt->{reindex}) {