]> Sergey Matveev's repositories - public-inbox.git/blobdiff - lib/PublicInbox/Import.pm
import: implement barrier operation for v1 repos
[public-inbox.git] / lib / PublicInbox / Import.pm
index 845fbb69852e57f465a47e73de844330215ff888..12df7d598fc71d38088d4aeaebd0ed6d8ed3e37d 100644 (file)
@@ -11,9 +11,7 @@ use Fcntl qw(:flock :DEFAULT);
 use PublicInbox::Spawn qw(spawn);
 use PublicInbox::MID qw(mid_mime mid2path);
 use PublicInbox::Address;
-use PublicInbox::ContentId qw(content_id);
-use Date::Parse qw(str2time);
-use Time::Zone qw(tz_offset);
+use PublicInbox::MsgTime qw(msg_timestamp);
 
 sub new {
        my ($class, $git, $name, $email, $ibx) = @_;
@@ -31,6 +29,7 @@ sub new {
                inbox => $ibx,
                path_type => '2/38', # or 'v2'
                ssoma_lock => 1, # disable for v2
+               bytes_added => 0,
        }, $class
 }
 
@@ -122,7 +121,7 @@ sub check_remove_v1 {
        $n = read($r, my $lf, 1);
        defined($n) or die "read final byte of cat-blob failed: $!";
        die "bad read on final byte: <$lf>" if $lf ne "\n";
-       my $cur = PublicInbox::MIME->new($buf);
+       my $cur = PublicInbox::MIME->new(\$buf);
        my $cur_s = $cur->header('Subject');
        $cur_s = '' unless defined $cur_s;
        my $cur_m = $mime->header('Subject');
@@ -133,7 +132,6 @@ sub check_remove_v1 {
        (undef, $cur);
 }
 
-# used for v2 (maybe)
 sub checkpoint {
        my ($self) = @_;
        return unless $self->{pid};
@@ -141,6 +139,51 @@ sub checkpoint {
        undef;
 }
 
+sub progress {
+       my ($self, $msg) = @_;
+       return unless $self->{pid};
+       print { $self->{out} } "progress $msg\n" or wfail;
+       $self->{in}->getline eq "progress $msg\n" or die
+               "progress $msg not received\n";
+       undef;
+}
+
+sub _update_git_info ($$) {
+       my ($self, $do_gc) = @_;
+       # for compatibility with existing ssoma installations
+       # we can probably remove this entirely by 2020
+       my $git_dir = $self->{git}->{git_dir};
+       my @cmd = ('git', "--git-dir=$git_dir");
+       my $index = "$git_dir/ssoma.index";
+       if (-e $index && !$ENV{FAST}) {
+               my $env = { GIT_INDEX_FILE => $index };
+               run_die([@cmd, qw(read-tree -m -v -i), $self->{ref}], $env);
+       }
+       run_die([@cmd, 'update-server-info'], undef);
+       ($self->{path_type} eq '2/38') and eval {
+               require PublicInbox::SearchIdx;
+               my $inbox = $self->{inbox} || $git_dir;
+               my $s = PublicInbox::SearchIdx->new($inbox);
+               $s->index_sync({ ref => $self->{ref} });
+       };
+       eval { run_die([@cmd, qw(gc --auto)], undef) } if $do_gc;
+}
+
+sub barrier {
+       my ($self) = @_;
+
+       # For safety, we ensure git checkpoint is complete before because
+       # the data in git is still more important than what is in Xapian
+       # in v2.  Performance may be gained by delaying the ->progress
+       # call but we lose safety
+       if ($self->{nchg}) {
+               $self->checkpoint;
+               $self->progress('checkpoint');
+               _update_git_info($self, 0);
+               $self->{nchg} = 0;
+       }
+}
+
 # used for v2
 sub get_mark {
        my ($self, $mark) = @_;
@@ -155,7 +198,6 @@ sub get_mark {
 # ('MISMATCH', Email::MIME) on mismatch
 # (:MARK, Email::MIME) on success
 #
-# For v2 inboxes, the content_id is returned instead of the msg
 # v2 callers should check with Xapian before calling this as
 # it is not idempotent.
 sub remove {
@@ -171,10 +213,17 @@ sub remove {
                ($err, $cur) = check_remove_v1($r, $w, $tip, $path, $mime);
                return ($err, $cur) if $err;
        } else {
-               $cur = content_id($mime);
-               my $len = length($cur);
+               my $sref;
+               if (ref($mime) eq 'SCALAR') { # optimization used by V2Writable
+                       $sref = $mime;
+               } else { # XXX should not be necessary:
+                       my $str = $mime->as_string;
+                       $sref = \$str;
+               }
+               my $len = length($$sref);
                $blob = $self->{mark}++;
-               print $w "blob\nmark :$blob\ndata $len\n$cur\n" or wfail;
+               print $w "blob\nmark :$blob\ndata $len\n",
+                       $$sref, "\n" or wfail;
        }
 
        my $ref = $self->{ref};
@@ -195,7 +244,7 @@ sub remove {
        if (defined $path) {
                print $w "D $path\n\n" or wfail;
        } else {
-               print $w "M 100644 :$blob d\n\n" or wfail;
+               print $w "M 100644 :$blob _/D\n\n" or wfail;
        }
        $self->{nchg}++;
        (($self->{tip} = ":$commit"), $cur);
@@ -203,55 +252,55 @@ sub remove {
 
 sub parse_date ($) {
        my ($mime) = @_;
-       my $hdr = $mime->header_obj;
-       my $date = $hdr->header_raw('Date');
-       my ($ts, $zone);
-       my $mid = $hdr->header_raw('Message-ID');
-       if ($date) {
-               $ts = eval { str2time($date) };
-               if ($@) {
-                       warn "bad Date: $date in $mid: $@\n";
-               } elsif ($date =~ /\s+([\+\-]\d+)\s*\z/) {
-                       $zone = $1;
-               }
-       }
-       unless ($ts) {
-               my @recvd = $hdr->header_raw('Received');
-               foreach my $r (@recvd) {
-                       $zone = undef;
-                       $r =~ /\s*(\d+\s+[[:alpha:]]+\s+\d{2,4}\s+
-                               \d+\D\d+(?:\D\d+)\s+([\+\-]\d+))/osx or next;
-                       $zone = $2;
-                       $ts = eval { str2time($1) } and last;
-                       warn "no date in Received: $r\n";
-               }
-       }
-       $zone ||= '+0000';
-       # "-1200" is the furthest westermost zone offset,
-       # but git fast-import is liberal so we use "-1400"
-       if ($zone >= 1400 || $zone <= -1400) {
-               warn "bogus TZ offset: $zone, ignoring and assuming +0000\n";
-               $zone = '+0000';
-       }
-       $ts ||= time;
+       my ($ts, $zone) = msg_timestamp($mime->header_obj);
        $ts = 0 if $ts < 0; # git uses unsigned times
        "$ts $zone";
 }
 
-# returns undef on duplicate
-# returns the :MARK of the most recent commit
-sub add {
-       my ($self, $mime, $check_cb) = @_; # mime = Email::MIME
+sub extract_author_info ($) {
+       my ($mime) = @_;
 
+       my $sender = '';
        my $from = $mime->header('From');
        my ($email) = PublicInbox::Address::emails($from);
        my ($name) = PublicInbox::Address::names($from);
+       if (!defined($name) || !defined($email)) {
+               $sender = $mime->header('Sender');
+               if (!defined($name)) {
+                       ($name) = PublicInbox::Address::names($sender);
+               }
+               if (!defined($email)) {
+                       ($email) = PublicInbox::Address::emails($sender);
+               }
+       }
+       if (defined $email) {
+               # quiet down wide character warnings with utf8::encode
+               utf8::encode($email);
+       } else {
+               $email = '';
+               warn "no email in From: $from or Sender: $sender\n";
+       }
+
        # git gets confused with:
        #  "'A U Thor <u@example.com>' via foo" <foo@example.com>
        # ref:
        # <CAD0k6qSUYANxbjjbE4jTW4EeVwOYgBD=bXkSu=akiYC_CB7Ffw@mail.gmail.com>
-       $name =~ tr/<>//d;
+       if (defined $name) {
+               $name =~ tr/<>//d;
+               utf8::encode($name);
+       } else {
+               $name = '';
+               warn "no name in From: $from or Sender: $sender\n";
+       }
+       ($name, $email);
+}
+
+# returns undef on duplicate
+# returns the :MARK of the most recent commit
+sub add {
+       my ($self, $mime, $check_cb) = @_; # mime = Email::MIME
 
+       my ($name, $email) = extract_author_info($mime);
        my $date_raw = parse_date($mime);
        my $subject = $mime->header('Subject');
        $subject = '(no subject)' unless defined $subject;
@@ -280,15 +329,16 @@ sub add {
 
        my $blob = $self->{mark}++;
        my $str = $mime->as_string;
-       print $w "blob\nmark :$blob\ndata ", length($str), "\n" or wfail;
+       my $n = length($str);
+       $self->{bytes_added} += $n;
+       print $w "blob\nmark :$blob\ndata ", $n, "\n" or wfail;
        print $w $str, "\n" or wfail;
-       $str = undef;
 
        # v2: we need this for Xapian
-       if ($self->{want_object_id}) {
-               chomp($self->{last_object_id} = $self->get_mark(":$blob"));
+       if ($self->{want_object_info}) {
+               chomp(my $oid = $self->get_mark(":$blob"));
+               $self->{last_object} = [ $oid, $n, \$str ];
        }
-
        my $ref = $self->{ref};
        my $commit = $self->{mark}++;
        my $parent = $tip =~ /\A:/ ? $tip : undef;
@@ -297,10 +347,7 @@ sub add {
                print $w "reset $ref\n" or wfail;
        }
 
-       utf8::encode($email);
-       utf8::encode($name);
        utf8::encode($subject);
-       # quiet down wide character warnings:
        print $w "commit $ref\nmark :$commit\n",
                "author $name <$email> $date_raw\n",
                "committer $self->{ident} ", now_raw(), "\n" or wfail;
@@ -314,7 +361,7 @@ sub add {
        $self->{tip} = ":$commit";
 }
 
-sub run_die ($$) {
+sub run_die ($;$) {
        my ($cmd, $env) = @_;
        my $pid = spawn($cmd, $env, undef);
        defined $pid or die "spawning ".join(' ', @$cmd)." failed: $!";
@@ -330,28 +377,8 @@ sub done {
        my $pid = delete $self->{pid} or die 'BUG: missing {pid} when done';
        waitpid($pid, 0) == $pid or die 'fast-import did not finish';
        $? == 0 or die "fast-import failed: $?";
-       my $nchg = delete $self->{nchg};
 
-       # for compatibility with existing ssoma installations
-       # we can probably remove this entirely by 2020
-       my $git_dir = $self->{git}->{git_dir};
-       my @cmd = ('git', "--git-dir=$git_dir");
-       my $index = "$git_dir/ssoma.index";
-       if ($nchg && -e $index && !$ENV{FAST}) {
-               my $env = { GIT_INDEX_FILE => $index };
-               run_die([@cmd, qw(read-tree -m -v -i), $self->{ref}], $env);
-       }
-       if ($nchg) {
-               run_die([@cmd, 'update-server-info'], undef);
-               eval {
-                       require PublicInbox::SearchIdx;
-                       my $inbox = $self->{inbox} || $git_dir;
-                       my $s = PublicInbox::SearchIdx->new($inbox);
-                       $s->index_sync({ ref => $self->{ref} });
-               };
-
-               eval { run_die([@cmd, qw(gc --auto)], undef) };
-       }
+       _update_git_info($self, 1) if delete $self->{nchg};
 
        $self->{ssoma_lock} or return;
        my $lockfh = delete $self->{lockfh} or die "BUG: not locked: $!";
@@ -359,6 +386,13 @@ sub done {
        close $lockfh or die "close lock failed: $!";
 }
 
+sub atfork_child {
+       my ($self) = @_;
+       foreach my $f (qw(in out)) {
+               close $self->{$f} or die "failed to close import[$f]: $!\n";
+       }
+}
+
 1;
 __END__
 =pod