X-Git-Url: http://www.git.stargrave.org/?p=public-inbox.git;a=blobdiff_plain;f=lib%2FPublicInbox%2FV2Writable.pm;h=34dd139b8ab82495296ddb9abd76847dee54b5fc;hp=9073d9ef89a8db95dd3db2d7c1b76cc26b758194;hb=1218a4126807951a0f47286338dc04d7f197bb78;hpb=c5c709dcef1b5904467ff2aea3943f00a891dade diff --git a/lib/PublicInbox/V2Writable.pm b/lib/PublicInbox/V2Writable.pm index 9073d9ef..34dd139b 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,6 +7,7 @@ package PublicInbox::V2Writable; use strict; use warnings; use base qw(PublicInbox::Lock); +use 5.010_001; use PublicInbox::SearchIdxShard; use PublicInbox::MIME; use PublicInbox::Git; @@ -32,19 +33,29 @@ my $PACKING_FACTOR = 0.4; # to increase Xapian shards our $NPROC_MAX_DEFAULT = 4; -sub nproc_shards ($) { - my ($creat_opt) = @_; - if (ref($creat_opt) eq 'HASH') { - if (defined(my $n = $creat_opt->{nproc})) { - return $n - } +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; } - my $n = $ENV{NPROC}; + # 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; } @@ -112,8 +123,11 @@ sub new { # public (for now?) sub init_inbox { - my ($self, $parallel, $skip_epoch) = @_; - $self->{parallel} = $parallel; + my ($self, $shards, $skip_epoch) = @_; + if (defined $shards) { + $self->{parallel} = 0 if $shards == 0; + $self->{shards} = $shards if $shards > 0; + } $self->idx_init; my $epoch_max = -1; git_dir_latest($self, \$epoch_max); @@ -136,9 +150,15 @@ 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 $smsg = bless { + bytes => $len, + num => $num, + blob => $oid, + mid => $mid0, + }, 'PublicInbox::Smsg'; + $self->{over}->add_overview($mime, $smsg, $self); my $idx = idx_shard($self, $num % $self->{shards}); - $idx->index_raw($len, $msgref, $num, $oid, $mid0, $mime); + $idx->index_raw($len, $msgref, $num, $oid, $mid0, $mime, $self); my $n = $self->{transact_bytes} += $len; $n >= (PublicInbox::SearchIdx::BATCH_BYTES * $self->{shards}); } @@ -162,7 +182,7 @@ 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 $cmt = $im->add($mime, undef, $self); # sets $self->{(au|co)time} $cmt = $im->get_mark($cmt); $self->{last_commit}->[$self->{epoch_max}] = $cmt; @@ -187,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 @@ -367,7 +386,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 @@ -377,7 +396,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); @@ -388,8 +407,7 @@ sub rewrite_internal ($$;$$$) { my $orig = $$msg; my $cur = PublicInbox::MIME->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; @@ -399,15 +417,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 @@ -427,15 +446,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 ($$;$$) { @@ -663,23 +685,43 @@ sub fill_alternates ($$) { unless (-d $all) { PublicInbox::Import::init_bare($all); } - my $alt = "$all/objects/info/alternates"; - my %alts; - my @add; + my $info_dir = "$all/objects/info"; + my $alt = "$info_dir/alternates"; + my (%alt, $new); + my $mode = 0644; if (-e $alt) { open(my $fh, '<', $alt) or die "open < $alt: $!\n"; - %alts = map { chomp; $_ => 1 } (<$fh>); + $mode = (stat($fh))[2] & 07777; + + # we assign a sort score to every alternate and favor + # the newest (highest numbered) one when we + my $score; + my $other = 0; # in case admin adds non-epoch repos + %alt = map {; + if (m!\A\Q../../\E([0-9]+)\.git/objects\z!) { + $score = $1 + 0; + } else { + $score = --$other; + } + $_ => $score; + } split(/\n+/, do { local $/; <$fh> }); } + foreach my $i (0..$epoch) { my $dir = "../../git/$i.git/objects"; - push @add, $dir if !$alts{$dir} && -d "$pfx/$i.git"; - } - return unless @add; - open my $fh, '>>', $alt or die "open >> $alt: $!\n"; - foreach my $dir (@add) { - print $fh "$dir\n" or die "print >> $alt: $!\n"; + if (!exists($alt{$dir}) && -d "$pfx/$i.git") { + $alt{$dir} = $i; + $new = 1; + } } - close $fh or die "close $alt: $!\n"; + return unless $new; + + my ($fh, $tmp) = tempfile('alt-XXXXXXXX', DIR => $info_dir); + print $fh join("\n", sort { $alt{$b} <=> $alt{$a} } keys %alt), "\n" + or die "print $tmp: $!\n"; + chmod($mode, $fh) or die "fchmod $tmp: $!\n"; + close $fh or die "close $tmp $!\n"; + rename($tmp, $alt) or die "rename $tmp => $alt: $!\n"; } sub git_init { @@ -787,7 +829,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); @@ -799,11 +841,7 @@ sub lookup_content ($$$) { next; } my $cur = PublicInbox::MIME->new($msg); - if (content_matches($cids, $cur)) { - $smsg->{mime} = $cur; - return $smsg; - } - + return 1 if content_matches($cids, $cur); # XXX DEBUG_DIFF is experimental and may be removed diff($mid, $cur, $mime) if $ENV{DEBUG_DIFF}; @@ -1234,15 +1272,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) {