X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FLeiXSearch.pm;h=9f7f3885a3a6fce2a120bb23416d109495c324ad;hb=57fed2e4b78ed394;hp=b3fd79d005c9ad00954c6d6e9b5edfce4e653c3d;hpb=63e2751ace2dd35a661620a9ad30d6c92948a009;p=public-inbox.git

diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index b3fd79d0..9f7f3885 100644
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -9,7 +9,6 @@ use strict;
 use v5.10.1;
 use parent qw(PublicInbox::LeiSearch PublicInbox::IPC);
 use PublicInbox::DS qw(now);
-use PublicInbox::PktOp qw(pkt_do);
 use File::Temp 0.19 (); # 0.19 for ->newdir
 use File::Spec ();
 use PublicInbox::Search qw(xap_terms);
@@ -33,7 +32,7 @@ sub attach_external {
 	my ($self, $ibxish) = @_; # ibxish = ExtSearch or Inbox
 	my $desc = $ibxish->{inboxdir} // $ibxish->{topdir};
 	my $srch = $ibxish->search or
-		return warn("$desc not indexed for Xapian\n");
+		return warn("$desc not indexed for Xapian ($@ $!)\n");
 	my @shards = $srch->xdb_shards_flat or
 		return warn("$desc has no Xapian shards\n");
 
@@ -67,15 +66,21 @@ sub remotes { @{$_[0]->{remotes} // []} }
 # called by PublicInbox::Search::xdb (usually via ->mset)
 sub xdb_shards_flat { @{$_[0]->{shards_flat} // []} }
 
-sub mitem_kw ($$;$) {
-	my ($smsg, $mitem, $flagged) = @_;
-	my $kw = xap_terms('K', my $doc = $mitem->get_document);
+sub _mitem_kw { # retry_reopen callback
+	my ($srch, $smsg, $mitem, $flagged) = @_;
+	my $doc = $mitem->get_document;
+	my $kw = xap_terms('K', $doc);
 	$kw->{flagged} = 1 if $flagged;
+	my @L = xap_terms('L', $doc);
 	# we keep the empty {kw} array here to prevent expensive work in
 	# ->xsmsg_vmd, _unbless_smsg will clobber it iff it's empty
 	$smsg->{kw} = [ sort keys %$kw ];
-	my $L = xap_terms('L', $doc);
-	$smsg->{L} = [ sort keys %$L ] if scalar(keys %$L);
+	$smsg->{L} = \@L if scalar(@L);
+}
+
+sub mitem_kw ($$$;$) {
+	my ($srch, $smsg, $mitem, $flagged) = @_;
+	$srch->retry_reopen(\&_mitem_kw, $smsg, $mitem, $flagged);
 }
 
 # like over->get_art
@@ -90,7 +95,7 @@ sub smsg_for {
 	my $smsg = $ibx->over->get_art($num);
 	return if $smsg->{bytes} == 0; # external message
 	if ($ibx->can('msg_keywords')) {
-		mitem_kw($smsg, $mitem);
+		mitem_kw($self, $smsg, $mitem);
 	}
 	$smsg;
 }
@@ -136,14 +141,19 @@ sub mset_progress {
 	my $lei = shift;
 	return if $lei->{early_mua} || !$lei->{-progress};
 	if ($lei->{pkt_op_p}) {
-		pkt_do($lei->{pkt_op_p}, 'mset_progress', @_);
+		$lei->{pkt_op_p}->pkt_do('mset_progress', @_);
 	} else { # single lei-daemon consumer
 		my ($desc, $mset_size, $mset_total_est) = @_;
-		$lei->{-mset_total} += $mset_size;
+		$lei->{-mset_total} += $mset_size if $mset_total_est ne '?';
 		$lei->qerr("# $desc $mset_size/$mset_total_est");
 	}
 }
 
+sub l2m_progress {
+	my ($lei, $nr) = @_;
+	$lei->{-nr_write} += $nr;
+}
+
 sub query_one_mset { # for --threads and l2m w/o sort
 	my ($self, $ibxish) = @_;
 	local $0 = "$0 query_one_mset";
@@ -151,6 +161,7 @@ sub query_one_mset { # for --threads and l2m w/o sort
 	my ($srch, $over) = ($ibxish->search, $ibxish->over);
 	my $dir = $ibxish->{inboxdir} // $ibxish->{topdir};
 	return warn("$dir not indexed by Xapian\n") unless ($srch && $over);
+	bless $srch, 'PublicInbox::LeiSearch'; # for ->qparse_new
 	my $mo = { %{$lei->{mset_opt}} }; # copy
 	my $mset;
 	my $each_smsg = $lei->{ovv}->ovv_each_smsg_cb($lei);
@@ -189,7 +200,8 @@ sub query_one_mset { # for --threads and l2m w/o sort
 					my $mitem = delete $n2item{$n};
 					next if $smsg->{bytes} == 0;
 					if ($mitem && $can_kw) {
-						mitem_kw($smsg, $mitem, $fl);
+						mitem_kw($srch, $smsg, $mitem,
+							$fl);
 					} elsif ($mitem && $fl) {
 						# call ->xsmsg_vmd, later
 						$smsg->{lei_q_tt_flagged} = 1;
@@ -205,7 +217,7 @@ sub query_one_mset { # for --threads and l2m w/o sort
 				my $mitem = $items[$i++];
 				my $smsg = $over->get_art($n) or next;
 				next if $smsg->{bytes} == 0;
-				mitem_kw($smsg, $mitem, $fl) if $can_kw;
+				mitem_kw($srch, $smsg, $mitem, $fl) if $can_kw;
 				$each_smsg->($smsg, $mitem);
 			}
 		}
@@ -233,7 +245,7 @@ sub query_combined_mset { # non-parallel for non-"--threads" users
 	do {
 		$mset = $self->mset($mo->{qstr}, $mo);
 		mset_progress($lei, 'xsearch', $mset->size,
-				$mset->size, $mset->get_matches_estimated);
+				$mset->get_matches_estimated);
 		wait_startq($lei); # wait for keyword updates
 		for my $mitem ($mset->items) {
 			my $smsg = smsg_for($self, $mitem) or next;
@@ -244,19 +256,27 @@ sub query_combined_mset { # non-parallel for non-"--threads" users
 	$lei->{ovv}->ovv_atexit_child($lei);
 }
 
+sub _smsg_fill ($$) {
+	my ($smsg, $eml) = @_;
+	$smsg->populate($eml);
+	$smsg->parse_references($eml, mids($eml));
+	$smsg->{$_} //= '' for qw(from to cc ds subject references mid);
+	delete @$smsg{qw(From Subject -ds -ts)};
+}
+
 sub each_remote_eml { # callback for MboxReader->mboxrd
 	my ($eml, $self, $lei, $each_smsg) = @_;
 	my $xoids = $lei->{ale}->xoids_for($eml, 1);
+	my $smsg = bless {}, 'PublicInbox::Smsg';
 	if ($self->{import_sto} && !$xoids) {
-		$self->{import_sto}->ipc_do('add_eml', $eml);
+		my $res = $self->{import_sto}->ipc_do('add_eml', $eml);
+		if (ref($res) eq ref($smsg)) { # totally new message
+			$smsg = $res;
+			$smsg->{kw} = []; # short-circuit xsmsg_vmd
+		}
 	}
-	my $smsg = bless {}, 'PublicInbox::Smsg';
-	$smsg->{blob} = $xoids ? (keys(%$xoids))[0]
-				: git_sha(1, $eml)->hexdigest;
-	$smsg->populate($eml);
-	$smsg->parse_references($eml, mids($eml));
-	$smsg->{$_} //= '' for qw(from to cc ds subject references mid);
-	delete @$smsg{qw(From Subject -ds -ts)};
+	$smsg->{blob} //= $xoids ? (keys(%$xoids))[0] : $lei->git_blob_id($eml);
+	_smsg_fill($smsg, $eml);
 	wait_startq($lei);
 	if ($lei->{-progress}) {
 		++$lei->{-nr_remote_eml};
@@ -277,7 +297,9 @@ sub query_remote_mboxrd {
 	local $SIG{TERM} = sub { exit(0) }; # for DESTROY (File::Temp, $reap)
 	my $lei = $self->{lei};
 	my $opt = $lei->{opt};
-	my @qform = (q => $lei->{mset_opt}->{qstr}, x => 'm');
+	my $qstr = $lei->{mset_opt}->{qstr};
+	$qstr =~ s/[ \n\t]+/ /sg; # make URLs less ugly
+	my @qform = (q => $qstr, x => 'm');
 	push(@qform, t => 1) if $opt->{threads};
 	my $verbose = $opt->{verbose};
 	my ($reap_tail, $reap_curl);
@@ -352,11 +374,14 @@ sub query_done { # EOF callback for main daemon
 	if (my $lxs = delete $lei->{lxs}) {
 		$lxs->wq_wait_old(\&xsearch_done_wait, $lei);
 	}
-	my $wait = $lei->{sto} ? $lei->{sto}->ipc_do('done') : undef;
+	if ($lei->{opt}->{'mail-sync'} && !$lei->{sto}) {
+		warn "BUG: {sto} missing with --mail-sync";
+	}
+	$lei->sto_done_request if $lei->{sto};
+	my $wait = $lei->{v2w} ? $lei->{v2w}->ipc_do('done') : undef;
 	$lei->{ovv}->ovv_end($lei);
-	my @out;
+	my $start_mua;
 	if ($l2m) { # close() calls LeiToMail reap_compress
-		@out = (" in $lei->{ovv}->{dst}");
 		if (my $out = delete $lei->{old_1}) {
 			if (my $mbout = $lei->{1}) {
 				close($mbout) or return $lei->fail(<<"");
@@ -370,17 +395,28 @@ Error closing $lei->{ovv}->{dst}: $!
 			$lei->poke_mua;
 		} else { # mbox users
 			delete $l2m->{mbl}; # drop dotlock
-			$lei->start_mua;
+			$start_mua = 1;
 		}
 	}
-	$lei->{-progress} and
-		$lei->err('# ', $lei->{-mset_total} // 0, " matches", @out);
+	if ($lei->{-progress}) {
+		my $tot = $lei->{-mset_total} // 0;
+		my $nr = $lei->{-nr_write} // 0;
+		if ($l2m) {
+			my $m = "# $nr written to " .
+				"$lei->{ovv}->{dst} ($tot matches)";
+			$nr ? $lei->qfin($m) : $lei->qerr($m);
+		} else {
+			$lei->qerr("# $tot matches");
+		}
+	}
+	$lei->start_mua if $start_mua;
 	$lei->dclose;
 }
 
 sub do_post_augment {
 	my ($lei) = @_;
 	my $l2m = $lei->{l2m} or return; # client disconnected
+	$lei->fchdir or return;
 	my $err;
 	eval { $l2m->post_augment($lei) };
 	$err = $@;
@@ -431,6 +467,9 @@ sub start_query ($;$) { # always runs in main (lei-daemon) process
 	for my $uris (@$q) {
 		$self->wq_io_do('query_remote_mboxrd', [], $uris);
 	}
+	if ($self->{-do_lcat}) {
+		$self->wq_io_do('lcat_dump', []);
+	}
 	$self->wq_close(1); # lei_xsearch workers stop when done
 }
 
@@ -456,6 +495,7 @@ sub do_query {
 		'+' => [ \&incr_post_augment, $lei ],
 		'' => [ \&query_done, $lei ],
 		'mset_progress' => [ \&mset_progress, $lei ],
+		'l2m_progress' => [ \&l2m_progress, $lei ],
 		'x_it' => [ $lei->can('x_it'), $lei ],
 		'child_error' => [ $lei->can('child_error'), $lei ],
 		'incr_start_query' => [ \&incr_start_query, $self, $l2m ],
@@ -471,11 +511,22 @@ sub do_query {
 		if ($lei->{opt}->{augment} && delete $lei->{early_mua}) {
 			$lei->start_mua;
 		}
+		my $F_SETPIPE_SZ = $^O eq 'linux' ? 1031 : undef;
+		if ($l2m->{-wq_nr_workers} > 1 &&
+				$l2m->{base_type} =~ /\A(?:maildir|mbox)\z/) {
+			# setup two barriers to coordinate ->has_entries
+			# between l2m workers
+			pipe(my ($a_r, $a_w)) or die "pipe: $!";
+			fcntl($a_r, $F_SETPIPE_SZ, 4096) if $F_SETPIPE_SZ;
+			pipe(my ($b_r, $b_w)) or die "pipe: $!";
+			fcntl($b_r, $F_SETPIPE_SZ, 4096) if $F_SETPIPE_SZ;
+			$l2m->{au_peers} = [ $a_r, $a_w, $b_r, $b_w ];
+		}
 		$l2m->wq_workers_start('lei2mail', undef,
 					$lei->oldset, { lei => $lei });
 		pipe($lei->{startq}, $lei->{au_done}) or die "pipe: $!";
-		# 1031: F_SETPIPE_SZ
-		fcntl($lei->{startq}, 1031, 4096) if $^O eq 'linux';
+		fcntl($lei->{startq}, $F_SETPIPE_SZ, 4096) if $F_SETPIPE_SZ;
+		delete $l2m->{au_peers};
 	}
 	$self->wq_workers_start('lei_xsearch', undef,
 				$lei->oldset, { lei => $lei });
@@ -484,13 +535,14 @@ sub do_query {
 	@$end = ();
 	$self->{opt_threads} = $lei->{opt}->{threads};
 	$self->{opt_sort} = $lei->{opt}->{'sort'};
+	$self->{-do_lcat} = $lei->{lcat_blob} // $lei->{lcat_fid};
 	if ($l2m) {
-		$l2m->net_merge_complete unless $lei->{auth};
+		$l2m->net_merge_all_done unless $lei->{auth};
 	} else {
 		start_query($self);
 	}
 	$lei->event_step_init; # wait for shutdowns
-	$op_c->op_wait_event($ops);
+	$lei->wait_wq_events($op_c, $ops);
 }
 
 sub add_uri {
@@ -527,5 +579,48 @@ sub prepare_external {
 	push @{$self->{locals}}, $loc;
 }
 
+sub _lcat_i { # LeiMailSync->each_src iterator callback
+	my ($oidbin, $id, $each_smsg) = @_;
+	$each_smsg->({blob => unpack('H*', $oidbin), pct => 100});
+}
+
+sub _lcat2smsg { # git->cat_async callback
+	my ($bref, $oid, $type, $size, $smsg) = @_;
+	if ($bref) {
+		my $eml = PublicInbox::Eml->new($bref);
+		my $json_dump = delete $smsg->{-json_dump};
+		bless $smsg, 'PublicInbox::Smsg';
+		_smsg_fill($smsg, $eml);
+		$json_dump->($smsg, undef, $eml);
+	}
+}
+
+sub lcat_dump {
+	my ($self) = @_;
+	my $lei = $self->{lei};
+	my $each_smsg = $lei->{ovv}->ovv_each_smsg_cb($lei);
+	my $git = $lei->{ale}->git;
+	if (!$lei->{l2m}) {
+		my $json_dump = $each_smsg;
+		$each_smsg = sub {
+			my ($smsg) = @_;
+			use Data::Dumper;
+			$smsg->{-json_dump} = $json_dump;
+			$git->cat_async($smsg->{blob}, \&_lcat2smsg, $smsg);
+		};
+	}
+	for my $oid (@{$lei->{lcat_blob} // []}) {
+		$each_smsg->({ blob => $oid, pct => 100 });
+	}
+	if (my $fids = delete $lei->{lcat_fid}) {
+		my $lms = $lei->{lse}->lms;
+		for my $fid (@$fids) {
+			$lms->each_src({fid => $fid}, \&_lcat_i, $each_smsg);
+		}
+	}
+	$git->async_wait_all;
+	undef $each_smsg; # may commit
+	$lei->{ovv}->ovv_atexit_child($lei);
+}
 
 1;