X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FV2Writable.pm;h=f599e0a03d80af5ec2bd64800c3ca2756ca6d80e;hb=46742d95647c7a80cb2f60d5c134717dd91e22e2;hp=72d8d5af6641aac8e0080132a92d26a11642c173;hpb=45e2c97624268da5e030ac3e4c033e6022b8c0de;p=public-inbox.git diff --git a/lib/PublicInbox/V2Writable.pm b/lib/PublicInbox/V2Writable.pm index 72d8d5af..f599e0a0 100644 --- a/lib/PublicInbox/V2Writable.pm +++ b/lib/PublicInbox/V2Writable.pm @@ -1,4 +1,4 @@ -# Copyright (C) 2018-2019 all contributors +# Copyright (C) 2018-2020 all contributors # License: AGPL-3.0+ # This interface wraps and mimics PublicInbox::Import @@ -7,8 +7,9 @@ package PublicInbox::V2Writable; use strict; use warnings; use base qw(PublicInbox::Lock); +use 5.010_001; use PublicInbox::SearchIdxShard; -use PublicInbox::MIME; +use PublicInbox::Eml; use PublicInbox::Git; use PublicInbox::Import; use PublicInbox::MID qw(mids references); @@ -18,6 +19,8 @@ use PublicInbox::OverIdx; use PublicInbox::Msgmap; use PublicInbox::Spawn qw(spawn popen_rd); use PublicInbox::SearchIdx; +use PublicInbox::MsgTime qw(msg_timestamp msg_datestamp); +use PublicInbox::MultiMidQueue; use IO::Handle; # ->autoflush use File::Temp qw(tempfile); @@ -32,14 +35,29 @@ my $PACKING_FACTOR = 0.4; # to increase Xapian shards our $NPROC_MAX_DEFAULT = 4; +sub detect_nproc () { + for my $nproc (qw(nproc gnproc)) { # GNU coreutils nproc + `$nproc 2>/dev/null` =~ /^(\d+)$/ and return $1; + } + + # getconf(1) is POSIX, but *NPROCESSORS* vars are not + for (qw(_NPROCESSORS_ONLN NPROCESSORS_ONLN)) { + `getconf $_ 2>/dev/null` =~ /^(\d+)$/ and return $1; + } + + # should we bother with `sysctl hw.ncpu`? Those only give + # us total processor count, not online processor count. + undef +} + sub nproc_shards ($) { my ($creat_opt) = @_; my $n = $creat_opt->{nproc} if ref($creat_opt) eq 'HASH'; $n //= $ENV{NPROC}; if (!$n) { - chomp($n = `nproc 2>/dev/null`); - # assume 2 cores if GNU nproc(1) is not available - $n = 2 if !$n; + # assume 2 cores if not detectable or zero + state $NPROC_DETECTED = detect_nproc() || 2; + $n = $NPROC_DETECTED; $n = $NPROC_MAX_DEFAULT if $n > $NPROC_MAX_DEFAULT; } @@ -102,6 +120,7 @@ sub new { last_commit => [], # git repo -> commit }; $self->{shards} = count_shards($self) || nproc_shards($creat); + $self->{index_max_size} = $v2ibx->{index_max_size}; bless $self, $class; } @@ -132,12 +151,14 @@ sub add { } # indexes a message, returns true if checkpointing is needed -sub do_idx ($$$$$$$) { - my ($self, $msgref, $mime, $len, $num, $oid, $mid0) = @_; - $self->{over}->add_overview($mime, $len, $num, $oid, $mid0); - my $idx = idx_shard($self, $num % $self->{shards}); - $idx->index_raw($len, $msgref, $num, $oid, $mid0, $mime); - my $n = $self->{transact_bytes} += $len; +sub do_idx ($$$$) { + my ($self, $msgref, $mime, $smsg) = @_; + $smsg->{ds} //= msg_datestamp($mime->header_obj, $self->{autime}); + $smsg->{ts} //= msg_timestamp($mime->header_obj, $self->{cotime}); + $self->{over}->add_overview($mime, $smsg); + my $idx = idx_shard($self, $smsg->{num} % $self->{shards}); + $idx->index_raw($msgref, $mime, $smsg); + my $n = $self->{transact_bytes} += $smsg->{bytes}; $n >= (PublicInbox::SearchIdx::BATCH_BYTES * $self->{shards}); } @@ -160,12 +181,13 @@ sub _add { defined $num or return; # duplicate defined $mid0 or die "BUG: $mid0 undefined\n"; my $im = $self->importer; - my $cmt = $im->add($mime); + my $smsg = bless { mid => $mid0, num => $num }, 'PublicInbox::Smsg'; + my $cmt = $im->add($mime, undef, $smsg); # sets $smsg->{ds|ts|blob} $cmt = $im->get_mark($cmt); $self->{last_commit}->[$self->{epoch_max}] = $cmt; - my ($oid, $len, $msgref) = @{$im->{last_object}}; - if (do_idx($self, $msgref, $mime, $len, $num, $oid, $mid0)) { + my $msgref = delete $smsg->{-raw_email}; + if (do_idx($self, $msgref, $mime, $smsg)) { $self->checkpoint; } @@ -185,11 +207,10 @@ sub v2_num_for { # crap, Message-ID is already known, hope somebody just resent: foreach my $m (@$mids) { # read-only lookup now safe to do after above barrier - my $existing = lookup_content($self, $mime, $m); # easy, don't store duplicates # note: do not add more diagnostic info here since # it gets noisy on public-inbox-watch restarts - return () if $existing; + return () if content_exists($self, $mime, $m); } # AltId may pre-populate article numbers (e.g. X-Mail-Count @@ -336,9 +357,10 @@ sub content_ids ($) { my ($mime) = @_; my @cids = ( content_id($mime) ); + # We still support Email::MIME, here, and # Email::MIME->as_string doesn't always round-trip, so we may # use a second content_id - my $rt = content_id(PublicInbox::MIME->new(\($mime->as_string))); + my $rt = content_id(PublicInbox::Eml->new(\($mime->as_string))); push @cids, $rt if $cids[0] ne $rt; \@cids; } @@ -365,7 +387,7 @@ sub rewrite_internal ($$;$$$) { } my $over = $self->{over}; my $cids = content_ids($old_mime); - my $removed; + my @removed; my $mids = mids($old_mime->header_obj); # We avoid introducing new blobs into git since the raw content @@ -375,7 +397,7 @@ sub rewrite_internal ($$;$$$) { my $mark; foreach my $mid (@$mids) { - my %gone; # num => [ smsg, raw ] + my %gone; # num => [ smsg, $mime, raw ] my ($id, $prev); while (my $smsg = $over->next_by_mid($mid, \$id, \$prev)) { my $msg = get_blob($self, $smsg); @@ -384,10 +406,9 @@ sub rewrite_internal ($$;$$$) { next; # continue } my $orig = $$msg; - my $cur = PublicInbox::MIME->new($msg); + my $cur = PublicInbox::Eml->new($msg); if (content_matches($cids, $cur)) { - $smsg->{mime} = $cur; - $gone{$smsg->{num}} = [ $smsg, \$orig ]; + $gone{$smsg->{num}} = [ $smsg, $cur, \$orig ]; } } my $n = scalar keys %gone; @@ -397,15 +418,16 @@ sub rewrite_internal ($$;$$$) { join(',', sort keys %gone), "\n"; } foreach my $num (keys %gone) { - my ($smsg, $orig) = @{$gone{$num}}; - # $removed should only be set once assuming + my ($smsg, $mime, $orig) = @{$gone{$num}}; + # @removed should only be set once assuming # no bugs in our deduplication code: - $removed = $smsg; + @removed = (undef, $mime, $smsg); my $oid = $smsg->{blob}; if ($replace_map) { $replace_map->{$oid} = $sref; } else { ($mark, undef) = $im->remove($orig, $cmt_msg); + $removed[0] = $mark; } $orig = undef; if ($need_reindex) { # ->replace @@ -425,15 +447,18 @@ sub rewrite_internal ($$;$$$) { my $rewrites = _replace_oids($self, $new_mime, $replace_map); return { rewrites => $rewrites, need_reindex => $need_reindex }; } - $removed; + defined($mark) ? @removed : undef; } -# public +# public (see PublicInbox::Import->remove), but note the 3rd element +# (retval[2]) is not part of the stable API shared with Import->remove sub remove { my ($self, $mime, $cmt_msg) = @_; + my @ret; $self->{-inbox}->with_umask(sub { - rewrite_internal($self, $mime, $cmt_msg); + @ret = rewrite_internal($self, $mime, $cmt_msg); }); + defined($ret[0]) ? @ret : undef; } sub _replace ($$;$$) { @@ -533,17 +558,21 @@ W: $list } # make sure we really got the OID: - my ($oid, $type, $len) = $self->{-inbox}->git->check($expect_oid); - $oid eq $expect_oid or die "BUG: $expect_oid not found after replace"; + my ($blob, $type, $bytes) = $self->{-inbox}->git->check($expect_oid); + $blob eq $expect_oid or die "BUG: $expect_oid not found after replace"; # don't leak FDs to Xapian: $self->{-inbox}->git->cleanup; # reindex modified messages: for my $smsg (@$need_reindex) { - my $num = $smsg->{num}; - my $mid0 = $smsg->{mid}; - do_idx($self, \$raw, $new_mime, $len, $num, $oid, $mid0); + my $new_smsg = bless { + blob => $blob, + bytes => $bytes, + num => $smsg->{num}, + mid => $smsg->{mid}, + }, 'PublicInbox::Smsg'; + do_idx($self, \$raw, $new_mime, $new_smsg); } $rewritten->{rewrites}; } @@ -703,9 +732,8 @@ sub fill_alternates ($$) { sub git_init { my ($self, $epoch) = @_; my $git_dir = "$self->{-inbox}->{inboxdir}/git/$epoch.git"; - my @cmd = (qw(git init --bare -q), $git_dir); - PublicInbox::Import::run_die(\@cmd); - @cmd = (qw/git config/, "--file=$git_dir/config", + PublicInbox::Import::init_bare($git_dir); + my @cmd = (qw/git config/, "--file=$git_dir/config", 'include.path', '../../all.git/config'); PublicInbox::Import::run_die(\@cmd); fill_alternates($self, $epoch); @@ -769,7 +797,6 @@ sub import_init { my ($self, $git, $packed_bytes, $tmp) = @_; my $im = PublicInbox::Import->new($git, undef, undef, $self->{-inbox}); $im->{bytes_added} = int($packed_bytes / $PACKING_FACTOR); - $im->{want_object_info} = 1; $im->{lock_path} = undef; $im->{path_type} = 'v2'; $self->{im} = $im unless $tmp; @@ -805,7 +832,7 @@ sub get_blob ($$) { $ibx->msg_by_smsg($smsg); } -sub lookup_content ($$$) { +sub content_exists ($$$) { my ($self, $mime, $mid) = @_; my $over = $self->{over}; my $cids = content_ids($mime); @@ -816,12 +843,8 @@ sub lookup_content ($$$) { warn "broken smsg for $mid\n"; next; } - my $cur = PublicInbox::MIME->new($msg); - if (content_matches($cids, $cur)) { - $smsg->{mime} = $cur; - return $smsg; - } - + my $cur = PublicInbox::Eml->new($msg); + return 1 if content_matches($cids, $cur); # XXX DEBUG_DIFF is experimental and may be removed diff($mid, $cur, $mime) if $ENV{DEBUG_DIFF}; @@ -846,8 +869,9 @@ sub atfork_child { sub mark_deleted ($$$$) { my ($self, $sync, $git, $oid) = @_; + return if PublicInbox::SearchIdx::too_big($self, $git, $oid); my $msgref = $git->cat_file($oid); - my $mime = PublicInbox::MIME->new($$msgref); + my $mime = PublicInbox::Eml->new($$msgref); my $mids = mids($mime->header_obj); my $cid = content_id($mime); foreach my $mid (@$mids) { @@ -878,7 +902,7 @@ sub reindex_oid_m ($$$$;$) { $self->{current_info} = "multi_mid $oid"; my ($num, $mid0, $len); my $msgref = $git->cat_file($oid, \$len); - my $mime = PublicInbox::MIME->new($$msgref); + my $mime = PublicInbox::Eml->new($$msgref); my $mids = mids($mime->header_obj); my $cid = content_id($mime); die "BUG: reindex_oid_m called for <=1 mids" if scalar(@$mids) <= 1; @@ -935,7 +959,13 @@ sub reindex_oid_m ($$$$;$) { } } $sync->{nr}++; - if (do_idx($self, $msgref, $mime, $len, $num, $oid, $mid0)) { + my $smsg = bless { + bytes => $len, + num => $num, + blob => $oid, + mid => $mid0, + }, 'PublicInbox::Smsg'; + if (do_idx($self, $msgref, $mime, $smsg)) { reindex_checkpoint($self, $sync, $git); } } @@ -952,36 +982,25 @@ sub check_unindexed ($$$) { } } -# reuse Msgmap to store num => oid mapping (rather than num => mid) -sub multi_mid_q_new () { - my ($fh, $fn) = tempfile('multi_mid-XXXXXXX', EXLOCK => 0, TMPDIR => 1); - my $multi_mid = PublicInbox::Msgmap->new_file($fn, 1); - $multi_mid->{dbh}->do('PRAGMA synchronous = OFF'); - # for Msgmap->DESTROY: - $multi_mid->{tmp_name} = $fn; - $multi_mid->{pid} = $$; - close $fh or die "failed to close $fn: $!"; - $multi_mid -} - -sub multi_mid_q_push ($$) { - my ($sync, $oid) = @_; - my $multi_mid = $sync->{multi_mid} //= multi_mid_q_new(); +sub multi_mid_q_push ($$$) { + my ($self, $sync, $oid) = @_; + my $multi_mid = $sync->{multi_mid} //= PublicInbox::MultiMidQueue->new; if ($sync->{reindex}) { # no regen on reindex - $multi_mid->mid_insert($oid); + $multi_mid->push_oid($oid, $self); } else { my $num = $sync->{regen}--; die "BUG: ran out of article numbers" if $num <= 0; - $multi_mid->mid_set($num, $oid); + $multi_mid->set_oid($num, $oid, $self); } } sub reindex_oid ($$$$) { my ($self, $sync, $git, $oid) = @_; + return if PublicInbox::SearchIdx::too_big($self, $git, $oid); my ($num, $mid0, $len); my $msgref = $git->cat_file($oid, \$len); return if $len == 0; # purged - my $mime = PublicInbox::MIME->new($$msgref); + my $mime = PublicInbox::Eml->new($$msgref); my $mids = mids($mime->header_obj); my $cid = content_id($mime); @@ -1024,14 +1043,20 @@ sub reindex_oid ($$$$) { # do not delete from {mm_tmp}, since another # single-MID message may use it. } else { # handle them at the end: - multi_mid_q_push($sync, $oid); + multi_mid_q_push($self, $sync, $oid); } return; } $sync->{mm_tmp}->mid_delete($mid0) or die "failed to delete <$mid0> for article #$num\n"; $sync->{nr}++; - if (do_idx($self, $msgref, $mime, $len, $num, $oid, $mid0)) { + my $smsg = bless { + bytes => $len, + num => $num, + blob => $oid, + mid => $mid0, + }, 'PublicInbox::Smsg'; + if (do_idx($self, $msgref, $mime, $smsg)) { reindex_checkpoint($self, $sync, $git); } } @@ -1169,7 +1194,7 @@ sub unindex_oid ($$$;$) { my ($self, $git, $oid, $unindexed) = @_; my $mm = $self->{mm}; my $msgref = $git->cat_file($oid); - my $mime = PublicInbox::MIME->new($msgref); + my $mime = PublicInbox::Eml->new($msgref); my $mids = mids($mime->header_obj); $mime = $msgref = undef; my $over = $self->{over}; @@ -1252,15 +1277,17 @@ sub index_epoch ($$$) { $pr->("$i.git indexing $range\n"); } - my @cmd = qw(log --raw -r --pretty=tformat:%H + my @cmd = qw(log --raw -r --pretty=tformat:%H.%at.%ct --no-notes --no-color --no-abbrev --no-renames); my $fh = $self->{reindex_pipe} = $git->popen(@cmd, $range); my $cmt; while (<$fh>) { chomp; $self->{current_info} = "$i.git $_"; - if (/\A$x40$/o && !defined($cmt)) { - $cmt = $_; + if (/\A($x40)\.([0-9]+)\.([0-9]+)$/o) { + $cmt //= $1; + $self->{autime} = $2; + $self->{cotime} = $3; } elsif (/\A:\d{6} 100644 $x40 ($x40) [AM]\tm$/o) { reindex_oid($self, $sync, $git, $1); } elsif (/\A:\d{6} 100644 $x40 ($x40) [AM]\td$/o) { @@ -1317,19 +1344,21 @@ sub index_sync { } if (my $multi_mid = delete $sync->{multi_mid}) { $git //= $self->{-inbox}->git; - my ($min, $max) = $multi_mid->minmax; + my $min = $multi_mid->{min}; + my $max = $multi_mid->{max}; if ($sync->{reindex}) { # we may need to create new Message-IDs if mirrors # were initially indexed with old versions for (my $i = $max; $i >= $min; $i--) { - my $oid = $multi_mid->mid_for($i); + my $oid; + $oid = $multi_mid->get_oid($i, $self) or next; next unless defined $oid; reindex_oid_m($self, $sync, $git, $oid); } } else { # regen on initial index for my $num ($min..$max) { - my $oid = $multi_mid->mid_for($num); - next unless defined $oid; + my $oid; + $oid = $multi_mid->get_oid($num, $self) or next; reindex_oid_m($self, $sync, $git, $oid, $num); } }