+ undef $each_smsg; # drops @io for l2m->{each_smsg_done}
+ $lei->{ovv}->ovv_atexit_child($lei);
+}
+
+sub each_remote_eml { # callback for MboxReader->mboxrd
+ my ($eml, $self, $lei, $each_smsg) = @_;
+ $lei->{sto}->ipc_do('set_eml', $eml) if $lei->{sto}; # --import-remote
+ my $smsg = bless {}, 'PublicInbox::Smsg';
+ $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)};
+ wait_startq($lei);
+ if ($lei->{-progress}) {
+ ++$lei->{-nr_remote_eml};
+ my $now = now();
+ my $next = $lei->{-next_progress} //= ($now + 1);
+ if ($now > $next) {
+ $lei->{-next_progress} = $now + 1;
+ my $nr = $lei->{-nr_remote_eml};
+ $lei->err("# $lei->{-current_url} $nr/?");
+ }
+ }
+ $each_smsg->($smsg, undef, $eml);
+}
+
+sub query_remote_mboxrd {
+ my ($self, $uris) = @_;
+ local $0 = "$0 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');
+ push(@qform, t => 1) if $opt->{threads};
+ my $verbose = $opt->{verbose};
+ my ($reap_tail, $reap_curl);
+ my $cerr = File::Temp->new(TEMPLATE => 'curl.err-XXXX', TMPDIR => 1);
+ fcntl($cerr, F_SETFL, O_APPEND|O_RDWR) or warn "set O_APPEND: $!";
+ my $rdr = { 2 => $cerr, pgid => 0 };
+ my $sigint_reap = $lei->can('sigint_reap');
+ if ($verbose) {
+ # spawn a process to force line-buffering, otherwise curl
+ # will write 1 character at-a-time and parallel outputs
+ # mmmaaayyy llloookkk llliiikkkeee ttthhhiiisss
+ my $o = { 1 => $lei->{2}, 2 => $lei->{2}, pgid => 0 };
+ my $pid = spawn(['tail', '-f', $cerr->filename], undef, $o);
+ $reap_tail = PublicInbox::OnDestroy->new($sigint_reap, $pid);
+ }
+ my $curl = PublicInbox::LeiCurl->new($lei, $self->{curl}) or return;
+ push @$curl, '-s', '-d', '';
+ my $each_smsg = $lei->{ovv}->ovv_each_smsg_cb($lei);
+ for my $uri (@$uris) {
+ $lei->{-current_url} = $uri->as_string;
+ $lei->{-nr_remote_eml} = 0;
+ $uri->query_form(@qform);
+ my $cmd = $curl->for_uri($lei, $uri);
+ $lei->qerr("# $cmd");
+ my ($fh, $pid) = popen_rd($cmd, undef, $rdr);
+ $reap_curl = PublicInbox::OnDestroy->new($sigint_reap, $pid);
+ $fh = IO::Uncompress::Gunzip->new($fh);
+ PublicInbox::MboxReader->mboxrd($fh, \&each_remote_eml, $self,
+ $lei, $each_smsg);
+ my $err = waitpid($pid, 0) == $pid ? undef
+ : "BUG: waitpid($cmd): $!";
+ @$reap_curl = (); # cancel OnDestroy
+ die $err if $err;
+ my $nr = $lei->{-nr_remote_eml};
+ if ($nr && $lei->{sto}) {
+ my $wait = $lei->{sto}->ipc_do('done');
+ }
+ if ($? == 0) {
+ mset_progress($lei, $lei->{-current_url}, $nr, $nr);
+ next;
+ }
+ $err = '';
+ if (-s $cerr) {
+ seek($cerr, 0, SEEK_SET) or
+ $lei->err("seek($cmd stderr): $!");
+ $err = do { local $/; <$cerr> } //
+ "read($cmd stderr): $!";
+ truncate($cerr, 0) or
+ $lei->err("truncate($cmd stderr): $!");
+ }
+ next if (($? >> 8) == 22 && $err =~ /\b404\b/);
+ $uri->query_form(q => $lei->{mset_opt}->{qstr});
+ $lei->child_error($?, "E: <$uri> $err");
+ }
+ undef $each_smsg;
+ $lei->{ovv}->ovv_atexit_child($lei);
+}
+
+# called by LeiOverview::each_smsg_cb
+sub git { $_[0]->{git_tmp} // die 'BUG: caller did not set {git_tmp}' }
+
+sub git_tmp ($) {
+ my ($self) = @_;
+ my (%seen, @dirs);
+ my $tmp = File::Temp->newdir("lei_xsearch_git.$$-XXXX", TMPDIR => 1);
+ for my $ibxish (locals($self)) {
+ my $d = File::Spec->canonpath($ibxish->git->{git_dir});
+ $seen{$d} //= push @dirs, "$d/objects\n"
+ }
+ my $git_dir = $tmp->dirname;
+ PublicInbox::Import::init_bare($git_dir);
+ my $f = "$git_dir/objects/info/alternates";
+ open my $alt, '>', $f or die "open($f): $!";
+ print $alt @dirs or die "print $f: $!";
+ close $alt or die "close $f: $!";
+ my $git = PublicInbox::Git->new($git_dir);
+ $git->{-tmp} = $tmp;
+ $git;
+}
+
+sub xsearch_done_wait { # dwaitpid callback
+ my ($arg, $pid) = @_;
+ my ($wq, $lei) = @$arg;
+ $lei->child_error($?, 'non-fatal error from '.ref($wq)) if $?;
+}
+
+sub query_done { # EOF callback for main daemon
+ my ($lei) = @_;
+ my $l2m = delete $lei->{l2m};
+ $l2m->wq_wait_old(\&xsearch_done_wait, $lei) if $l2m;
+ if (my $lxs = delete $lei->{lxs}) {
+ $lxs->wq_wait_old(\&xsearch_done_wait, $lei);
+ }
+ $lei->{ovv}->ovv_end($lei);
+ if ($l2m) { # close() calls LeiToMail reap_compress
+ if (my $out = delete $lei->{old_1}) {
+ if (my $mbout = $lei->{1}) {
+ close($mbout) or return $lei->fail(<<"");
+Error closing $lei->{ovv}->{dst}: $!
+
+ }
+ $lei->{1} = $out;
+ }
+ if ($l2m->lock_free) {
+ $l2m->poke_dst;
+ $lei->poke_mua;
+ } else { # mbox users
+ $lei->start_mua;
+ }
+ }
+ $lei->{-progress} and
+ $lei->err('# ', $lei->{-mset_total} // 0, " matches");
+ $lei->dclose;
+}
+
+sub do_post_augment {
+ my ($lei) = @_;
+ my $l2m = $lei->{l2m} or die 'BUG: unexpected do_post_augment';
+ my $err;
+ eval { $l2m->post_augment($lei) };
+ $err = $@;
+ if ($err) {
+ if (my $lxs = delete $lei->{lxs}) {
+ $lxs->wq_kill;
+ $lxs->wq_close(0, undef, $lei);
+ }
+ $lei->fail("$err");
+ }
+ if (!$err && delete $lei->{early_mua}) { # non-augment case
+ $lei->start_mua;
+ }
+ close(delete $lei->{au_done}); # triggers wait_startq in lei_xsearch
+}
+
+sub incr_post_augment { # called whenever an l2m shard finishes augment
+ my ($lei) = @_;
+ my $l2m = $lei->{l2m} or die 'BUG: unexpected incr_post_augment';
+ return if ++$lei->{nr_post_augment} != $l2m->{-wq_nr_workers};
+ do_post_augment($lei);
+}
+
+my $MAX_PER_HOST = 4;
+
+sub concurrency {
+ my ($self, $opt) = @_;
+ my $nl = $opt->{threads} ? locals($self) : 1;
+ my $nr = remotes($self);
+ $nr = $MAX_PER_HOST if $nr > $MAX_PER_HOST;
+ $nl + $nr;
+}
+
+sub start_query { # always runs in main (lei-daemon) process
+ my ($self) = @_;
+ if ($self->{threads}) {
+ for my $ibxish (locals($self)) {
+ $self->wq_io_do('query_thread_mset', [], $ibxish);
+ }
+ } elsif (locals($self)) {
+ $self->wq_io_do('query_mset', []);
+ }
+ my $i = 0;
+ my $q = [];
+ for my $uri (remotes($self)) {
+ push @{$q->[$i++ % $MAX_PER_HOST]}, $uri;
+ }
+ for my $uris (@$q) {
+ $self->wq_io_do('query_remote_mboxrd', [], $uris);
+ }
+ $self->wq_close(1); # lei_xsearch workers stop when done
+}
+
+sub incr_start_query { # called whenever an l2m shard starts do_post_auth
+ my ($self, $l2m) = @_;
+ return if ++$self->{nr_start_query} != $l2m->{-wq_nr_workers};
+ start_query($self);
+}
+
+sub ipc_atfork_child {
+ my ($self) = @_;
+ $self->{lei}->lei_atfork_child;
+ $SIG{__WARN__} = PublicInbox::Eml::warn_ignore_cb();
+ $self->SUPER::ipc_atfork_child;