X-Git-Url: http://www.git.stargrave.org/?p=public-inbox.git;a=blobdiff_plain;f=lib%2FPublicInbox%2FV2Writable.pm;h=877a45912fcea3fab814fd9814b3996526ceb0d9;hp=832384601625fcbb5bca8cfa374e47d05d72d46b;hb=fad9acd35e56a289ade90a62d056b2a6663d448c;hpb=79c5fdeacf75a2daeb57a7a5f00a170044e28634

diff --git a/lib/PublicInbox/V2Writable.pm b/lib/PublicInbox/V2Writable.pm
index 83238460..877a4591 100644
--- a/lib/PublicInbox/V2Writable.pm
+++ b/lib/PublicInbox/V2Writable.pm
@@ -69,9 +69,9 @@ sub new {
 		xpfx => $xpfx,
 		over => PublicInbox::OverIdxFork->new("$xpfx/over.sqlite3"),
 		lock_path => "$dir/inbox.lock",
-		# limit each repo to 1GB or so
+		# limit each git repo (epoch) to 1GB or so
 		rotate_bytes => int((1024 * 1024 * 1024) / $PACKING_FACTOR),
-		last_commit => [],
+		last_commit => [], # git repo -> commit
 	};
 	$self->{partitions} = count_partitions($self) || nproc();
 	bless $self, $class;
@@ -81,9 +81,9 @@ sub init_inbox {
 	my ($self, $parallel) = @_;
 	$self->{parallel} = $parallel;
 	$self->idx_init;
-	my $max_git = -1;
-	git_dir_latest($self, \$max_git);
-	$self->git_init($max_git >= 0 ? $max_git : 0);
+	my $epoch_max = -1;
+	git_dir_latest($self, \$epoch_max);
+	$self->git_init($epoch_max >= 0 ? $epoch_max : 0);
 	$self->done;
 }
 
@@ -115,7 +115,7 @@ sub add {
 
 	my $nparts = $self->{partitions};
 	my $part = $num % $nparts;
-	$self->{last_commit}->[$self->{max_git}] = $cmt;
+	$self->{last_commit}->[$self->{epoch_max}] = $cmt;
 	my $idx = $self->idx_part($part);
 	$idx->index_raw($len, $msgref, $num, $oid, $mid0, $mime);
 	my $n = $self->{transact_bytes} += $len;
@@ -242,7 +242,7 @@ sub purge_oids {
 	$self->done;
 	my $pfx = "$self->{-inbox}->{mainrepo}/git";
 	my $purges = [];
-	foreach my $i (0..$self->{max_git}) {
+	foreach my $i (0..$self->{epoch_max}) {
 		my $git = PublicInbox::Git->new("$pfx/$i.git");
 		my $im = $self->import_init($git, 0, 1);
 		$purges->[$i] = $im->purge_oids($purge);
@@ -271,22 +271,20 @@ sub remove_internal {
 
 	foreach my $mid (@$mids) {
 		my %gone;
-		$srch->reopen->each_smsg_by_mid($mid, sub {
-			my ($smsg) = @_;
-			$smsg->load_expand;
+		my ($id, $prev);
+		while (my $smsg = $srch->next_by_mid($mid, \$id, \$prev)) {
 			my $msg = $ibx->msg_by_smsg($smsg);
 			if (!defined($msg)) {
 				warn "broken smsg for $mid\n";
-				return 1; # continue
+				next; # continue
 			}
 			my $orig = $$msg;
 			my $cur = PublicInbox::MIME->new($msg);
 			if (content_id($cur) eq $cid) {
 				$smsg->{mime} = $cur;
-				$gone{$smsg->num} = [ $smsg, \$orig ];
+				$gone{$smsg->{num}} = [ $smsg, \$orig ];
 			}
-			1; # continue
-		});
+		}
 		my $n = scalar keys %gone;
 		next unless $n;
 		if ($n > 1) {
@@ -315,7 +313,7 @@ sub remove_internal {
 
 	if (defined $mark) {
 		my $cmt = $im->get_mark($mark);
-		$self->{last_commit}->[$self->{max_git}] = $cmt;
+		$self->{last_commit}->[$self->{epoch_max}] = $cmt;
 	}
 	if ($purge && scalar keys %$purge) {
 		return purge_oids($self, $purge);
@@ -339,15 +337,20 @@ sub purge {
 	$purges;
 }
 
+sub last_commit_part ($$;$) {
+	my ($self, $i, $cmt) = @_;
+	my $v = PublicInbox::Search::SCHEMA_VERSION();
+	$self->{mm}->last_commit_xap($v, $i, $cmt);
+}
+
 sub set_last_commits ($) {
 	my ($self) = @_;
-	defined(my $max_git = $self->{max_git}) or return;
-	my $mm = $self->{mm};
+	defined(my $epoch_max = $self->{epoch_max}) or return;
 	my $last_commit = $self->{last_commit};
-	foreach my $i (0..$max_git) {
+	foreach my $i (0..$epoch_max) {
 		defined(my $cmt = $last_commit->[$i]) or next;
 		$last_commit->[$i] = undef;
-		$mm->last_commit_n($i, $cmt);
+		last_commit_part($self, $i, $cmt);
 	}
 }
 
@@ -392,7 +395,7 @@ sub checkpoint {
 # issue a write barrier to ensure all data is visible to other processes
 # and read-only ops.  Order of data importance is: git > SQLite > Xapian
 sub barrier {
-	my ($self, $fsync) = @_;
+	my ($self) = @_;
 
 	if (my $im = $self->{im}) {
 		$im->barrier;
@@ -411,7 +414,6 @@ sub barrier {
 		$_->remote_barrier foreach @$parts;
 
 		$over->barrier_wait; # wait for each Xapian partition
-		$over->commit_fsync if $fsync;
 
 		# last_commit is special, don't commit these until
 		# remote partitions are done:
@@ -425,9 +427,9 @@ sub barrier {
 }
 
 sub git_init {
-	my ($self, $new) = @_;
+	my ($self, $epoch) = @_;
 	my $pfx = "$self->{-inbox}->{mainrepo}/git";
-	my $git_dir = "$pfx/$new.git";
+	my $git_dir = "$pfx/$epoch.git";
 	my @cmd = (qw(git init --bare -q), $git_dir);
 	PublicInbox::Import::run_die(\@cmd);
 
@@ -445,7 +447,7 @@ sub git_init {
 	PublicInbox::Import::run_die(\@cmd);
 
 	my $alt = "$all/objects/info/alternates";
-	my $new_obj_dir = "../../git/$new.git/objects";
+	my $new_obj_dir = "../../git/$epoch.git/objects";
 	my %alts;
 	if (-e $alt) {
 		open(my $fh, '<', $alt) or die "open < $alt: $!\n";
@@ -486,26 +488,26 @@ sub importer {
 			$im->done;
 			$self->barrier(1);
 			$im = undef;
-			my $git_dir = $self->git_init(++$self->{max_git});
+			my $git_dir = $self->git_init(++$self->{epoch_max});
 			my $git = PublicInbox::Git->new($git_dir);
 			return $self->import_init($git, 0);
 		}
 	}
-	my $new = 0;
+	my $epoch = 0;
 	my $max;
 	my $latest = git_dir_latest($self, \$max);
 	if (defined $latest) {
 		my $git = PublicInbox::Git->new($latest);
 		my $packed_bytes = $git->packed_bytes;
 		if ($packed_bytes >= $self->{rotate_bytes}) {
-			$new = $max + 1;
+			$epoch = $max + 1;
 		} else {
-			$self->{max_git} = $max;
+			$self->{epoch_max} = $max;
 			return $self->import_init($git, $packed_bytes);
 		}
 	}
-	$self->{max_git} = $new;
-	$latest = $self->git_init($new);
+	$self->{epoch_max} = $epoch;
+	$latest = $self->git_init($epoch);
 	$self->import_init(PublicInbox::Git->new($latest), 0);
 }
 
@@ -548,26 +550,23 @@ sub lookup_content {
 	my $srch = $ibx->search->reopen;
 	my $cid = content_id($mime);
 	my $found;
-	$srch->each_smsg_by_mid($mid, sub {
-		my ($smsg) = @_;
-		$smsg->load_expand;
+	my ($id, $prev);
+	while (my $smsg = $srch->next_by_mid($mid, \$id, \$prev)) {
 		my $msg = $ibx->msg_by_smsg($smsg);
 		if (!defined($msg)) {
 			warn "broken smsg for $mid\n";
-			return 1; # continue
+			next;
 		}
 		my $cur = PublicInbox::MIME->new($msg);
 		if (content_id($cur) eq $cid) {
 			$smsg->{mime} = $cur;
 			$found = $smsg;
-			return 0; # break out of loop
+			last;
 		}
 
 		# XXX DEBUG_DIFF is experimental and may be removed
 		diff($mid, $cur, $mime) if $ENV{DEBUG_DIFF};
-
-		1; # continue
-	});
+	}
 	$found;
 }
 
@@ -673,22 +672,22 @@ sub reindex_oid {
 # only update last_commit for $i on reindex iff newer than current
 sub update_last_commit {
 	my ($self, $git, $i, $cmt) = @_;
-	my $last = $self->{mm}->last_commit_n($i);
+	my $last = last_commit_part($self, $i);
 	if (defined $last && is_ancestor($git, $last, $cmt)) {
 		my @cmd = (qw(rev-list --count), "$last..$cmt");
 		chomp(my $n = $git->qx(@cmd));
 		return if $n ne '' && $n == 0;
 	}
-	$self->{mm}->last_commit_n($i, $cmt);
+	last_commit_part($self, $i, $cmt);
 }
 
 sub git_dir_n ($$) { "$_[0]->{-inbox}->{mainrepo}/git/$_[1].git" }
 
 sub last_commits {
-	my ($self, $max_git) = @_;
+	my ($self, $epoch_max) = @_;
 	my $heads = [];
-	for (my $i = $max_git; $i >= 0; $i--) {
-		$heads->[$i] = $self->{mm}->last_commit_n($i);
+	for (my $i = $epoch_max; $i >= 0; $i--) {
+		$heads->[$i] = last_commit_part($self, $i);
 	}
 	$heads;
 }
@@ -705,10 +704,10 @@ sub is_ancestor ($$$) {
 }
 
 sub index_prepare {
-	my ($self, $opts, $max_git, $ranges) = @_;
+	my ($self, $opts, $epoch_max, $ranges) = @_;
 	my $regen_max = 0;
 	my $head = $self->{-inbox}->{ref_head} || 'refs/heads/master';
-	for (my $i = $max_git; $i >= 0; $i--) {
+	for (my $i = $epoch_max; $i >= 0; $i--) {
 		die "already indexing!\n" if $self->{index_pipe};
 		my $git_dir = git_dir_n($self, $i);
 		-d $git_dir or next; # missing parts are fine
@@ -766,15 +765,14 @@ sub unindex_oid {
 	my $mime = PublicInbox::MIME->new($msgref);
 	my $mids = mids($mime->header_obj);
 	$mime = $msgref = undef;
-
+	my $srch = $self->{-inbox}->search;
 	foreach my $mid (@$mids) {
 		my %gone;
-		$self->{-inbox}->search->reopen->each_smsg_by_mid($mid, sub {
-			my ($smsg) = @_;
-			$smsg->load_expand;
+		my ($id, $prev);
+		while (my $smsg = $srch->next_by_mid($mid, \$id, \$prev)) {
 			$gone{$smsg->num} = 1 if $oid eq $smsg->{blob};
 			1; # continue
-		});
+		}
 		my $n = scalar keys %gone;
 		next unless $n;
 		if ($n > 1) {
@@ -817,15 +815,15 @@ sub index_sync {
 	my ($self, $opts) = @_;
 	$opts ||= {};
 	my $ibx = $self->{-inbox};
-	my $max_git;
-	my $latest = git_dir_latest($self, \$max_git);
+	my $epoch_max;
+	my $latest = git_dir_latest($self, \$epoch_max);
 	return unless defined $latest;
 	$self->idx_init; # acquire lock
 	my $mm_tmp = $self->{mm}->tmp_clone;
-	my $ranges = $opts->{reindex} ? [] : $self->last_commits($max_git);
+	my $ranges = $opts->{reindex} ? [] : $self->last_commits($epoch_max);
 
 	my ($min, $max) = $mm_tmp->minmax;
-	my $regen = $self->index_prepare($opts, $max_git, $ranges);
+	my $regen = $self->index_prepare($opts, $epoch_max, $ranges);
 	$$regen += $max if $max;
 	my $D = {};
 	my @cmd = qw(log --raw -r --pretty=tformat:%h
@@ -833,7 +831,7 @@ sub index_sync {
 
 	# work backwards through history
 	my $last_commit = [];
-	for (my $i = $max_git; $i >= 0; $i--) {
+	for (my $i = $epoch_max; $i >= 0; $i--) {
 		my $git_dir = git_dir_n($self, $i);
 		die "already reindexing!\n" if delete $self->{reindex_pipe};
 		-d $git_dir or next; # missing parts are fine