X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FLeiXSearch.pm;h=f64b2c62d307887fb451df5d97b6e114c6f13b33;hb=43c43f785aa53607a0dd050989da5d7fd0dcfff4;hp=10485220a359976ebe6f88fad2ab3ce91c69eff0;hpb=80ba871c557eb5ba79c73150519ce9ea5225d695;p=public-inbox.git diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm index 10485220..f64b2c62 100644 --- a/lib/PublicInbox/LeiXSearch.pm +++ b/lib/PublicInbox/LeiXSearch.pm @@ -18,6 +18,7 @@ use PublicInbox::MID qw(mids); use PublicInbox::Smsg; use PublicInbox::Eml; use Fcntl qw(SEEK_SET F_SETFL O_APPEND O_RDWR); +use PublicInbox::ContentHash qw(git_sha); sub new { my ($class) = @_; @@ -63,9 +64,20 @@ sub locals { @{$_[0]->{locals} // []} } sub remotes { @{$_[0]->{remotes} // []} } -# called by PublicInbox::Search::xdb +# 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); + $kw->{flagged} = 1 if $flagged; + # 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); +} + # like over->get_art sub smsg_for { my ($self, $mitem) = @_; @@ -76,11 +88,10 @@ sub smsg_for { my $num = int(($docid - 1) / $nshard) + 1; my $ibx = $self->{shard2ibx}->[$shard]; my $smsg = $ibx->over->get_art($num); - if (ref($ibx->can('msg_keywords'))) { - my $kw = xap_terms('K', $mitem->get_document); - $smsg->{kw} = [ sort keys %$kw ]; + return if $smsg->{bytes} == 0; # external message + if ($ibx->can('msg_keywords')) { + mitem_kw($smsg, $mitem); } - $smsg->{docid} = $docid; $smsg; } @@ -88,7 +99,7 @@ sub recent { my ($self, $qstr, $opt) = @_; $opt //= {}; $opt->{relevance} //= -2; - $self->mset($qstr //= 'bytes:1..', $opt); + $self->mset($qstr //= 'z:1..', $opt); } sub over {} @@ -99,21 +110,21 @@ sub _mset_more ($$) { $size >= $mo->{limit} && (($mo->{offset} += $size) < $mo->{limit}); } -# $startq will EOF when query_prepare is done augmenting and allow +# $startq will EOF when do_augment is done augmenting and allow # query_mset and query_thread_mset to proceed. sub wait_startq ($) { my ($lei) = @_; my $startq = delete $lei->{startq} or return; while (1) { - my $n = sysread($startq, my $query_prepare_done, 1); + my $n = sysread($startq, my $do_augment_done, 1); if (defined $n) { return if $n == 0; # no MUA - if ($query_prepare_done eq 'q') { + if ($do_augment_done eq 'q') { $lei->{opt}->{quiet} = 1; delete $lei->{opt}->{verbose}; delete $lei->{-progress}; } else { - $lei->fail("$$ WTF `$query_prepare_done'"); + $lei->fail("$$ WTF `$do_augment_done'"); } return; } @@ -143,10 +154,13 @@ sub query_thread_mset { # for --threads my $mo = { %{$lei->{mset_opt}} }; my $mset; my $each_smsg = $lei->{ovv}->ovv_each_smsg_cb($lei, $ibxish); + my $can_kw = !!$ibxish->can('msg_keywords'); + my $fl = $lei->{opt}->{threads} > 1 ? 1 : undef; do { $mset = $srch->mset($mo->{qstr}, $mo); mset_progress($lei, $desc, $mset->size, $mset->get_matches_estimated); + wait_startq($lei); # wait for keyword updates my $ids = $srch->mset_to_artnums($mset, $mo); my $ctx = { ids => $ids }; my $i = 0; @@ -154,8 +168,16 @@ sub query_thread_mset { # for --threads while ($over->expand_thread($ctx)) { for my $n (@{$ctx->{xids}}) { my $smsg = $over->get_art($n) or next; - wait_startq($lei); my $mitem = delete $n2item{$smsg->{num}}; + next if $smsg->{bytes} == 0; + if ($mitem) { + if ($can_kw) { + mitem_kw($smsg, $mitem, $fl); + } elsif ($fl) { + # call ->xsmsg_vmd, later + $smsg->{lei_q_tt_flagged} = 1; + } + } $each_smsg->($smsg, $mitem); } @{$ctx->{xids}} = (); @@ -179,9 +201,9 @@ sub query_mset { # non-parallel for non-"--threads" users $mset = $self->mset($mo->{qstr}, $mo); mset_progress($lei, 'xsearch', $mset->size, $mset->size, $mset->get_matches_estimated); + wait_startq($lei); # wait for keyword updates for my $mitem ($mset->items) { my $smsg = smsg_for($self, $mitem) or next; - wait_startq($lei); $each_smsg->($smsg, $mitem); } } while (_mset_more($mset, $mo)); @@ -189,9 +211,15 @@ sub query_mset { # non-parallel for non-"--threads" users $lei->{ovv}->ovv_atexit_child($lei); } -sub each_eml { # callback for MboxReader->mboxrd +sub each_remote_eml { # callback for MboxReader->mboxrd my ($eml, $self, $lei, $each_smsg) = @_; + my $xoids = $lei->{ale}->xoids_for($eml, 1); + if ($self->{import_sto} && !$xoids) { + $self->{import_sto}->ipc_do('add_eml', $eml); + } 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); @@ -215,7 +243,7 @@ sub query_remote_mboxrd { local $0 = "$0 query_remote_mboxrd"; local $SIG{TERM} = sub { exit(0) }; # for DESTROY (File::Temp, $reap) my $lei = $self->{lei}; - my ($opt, $env) = @$lei{qw(opt env)}; + my $opt = $lei->{opt}; my @qform = (q => $lei->{mset_opt}->{qstr}, x => 'm'); push(@qform, t => 1) if $opt->{threads}; my $verbose = $opt->{verbose}; @@ -235,23 +263,27 @@ sub query_remote_mboxrd { my $curl = PublicInbox::LeiCurl->new($lei, $self->{curl}) or return; push @$curl, '-s', '-d', ''; my $each_smsg = $lei->{ovv}->ovv_each_smsg_cb($lei); + $self->{import_sto} = $lei->{sto} if $lei->{opt}->{'import-remote'}; 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, $env, $rdr); + 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_eml, $self, + 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) { - my $nr = $lei->{-nr_remote_eml}; mset_progress($lei, $lei->{-current_url}, $nr, $nr); next; } @@ -272,27 +304,7 @@ sub query_remote_mboxrd { $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 git { $_[0]->{git} // die 'BUG: git uninitialized' } sub xsearch_done_wait { # dwaitpid callback my ($arg, $pid) = @_; @@ -321,6 +333,7 @@ Error closing $lei->{ovv}->{dst}: $! $l2m->poke_dst; $lei->poke_mua; } else { # mbox users + delete $l2m->{mbl}; # drop dotlock $lei->start_mua; } } @@ -331,18 +344,16 @@ Error closing $lei->{ovv}->{dst}: $! sub do_post_augment { my ($lei) = @_; - my $l2m = $lei->{l2m}; + my $l2m = $lei->{l2m} or return; # client disconnected my $err; - if ($l2m) { - 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"); + 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; @@ -350,6 +361,13 @@ sub do_post_augment { 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 return; # client disconnected + return if ++$lei->{nr_post_augment} != $l2m->{-wq_nr_workers}; + do_post_augment($lei); +} + my $MAX_PER_HOST = 4; sub concurrency { @@ -361,8 +379,8 @@ sub concurrency { } sub start_query { # always runs in main (lei-daemon) process - my ($self, $lei) = @_; - if ($lei->{opt}->{threads}) { + my ($self) = @_; + if ($self->{threads}) { for my $ibxish (locals($self)) { $self->wq_io_do('query_thread_mset', [], $ibxish); } @@ -377,65 +395,64 @@ sub start_query { # always runs in main (lei-daemon) process 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; + $self->{lei}->_lei_atfork_child; $SIG{__WARN__} = PublicInbox::Eml::warn_ignore_cb(); $self->SUPER::ipc_atfork_child; } -sub query_prepare { # called by wq_io_do - my ($self) = @_; - local $0 = "$0 query_prepare"; - my $lei = $self->{lei}; - eval { $lei->{l2m}->do_augment($lei) }; - $lei->fail($@) if $@; - pkt_do($lei->{pkt_op_p}, '.') == 1 or die "do_post_augment trigger: $!" -} - sub do_query { my ($self, $lei) = @_; + my $l2m = $lei->{l2m}; my $ops = { '|' => [ $lei->can('sigpipe_handler'), $lei ], '!' => [ $lei->can('fail_handler'), $lei ], '.' => [ \&do_post_augment, $lei ], + '+' => [ \&incr_post_augment, $lei ], '' => [ \&query_done, $lei ], 'mset_progress' => [ \&mset_progress, $lei ], 'x_it' => [ $lei->can('x_it'), $lei ], 'child_error' => [ $lei->can('child_error'), $lei ], + 'incr_start_query' => [ \&incr_start_query, $self, $l2m ], }; - ($lei->{pkt_op_c}, $lei->{pkt_op_p}) = PublicInbox::PktOp->pair($ops); + $lei->{auth}->op_merge($ops, $l2m) if $l2m && $lei->{auth}; + my $end = $lei->pkt_op_pair($ops); $lei->{1}->autoflush(1); $lei->start_pager if delete $lei->{need_pager}; $lei->{ovv}->ovv_begin($lei); - my $l2m = $lei->{l2m}; if ($l2m) { $l2m->pre_augment($lei); if ($lei->{opt}->{augment} && delete $lei->{early_mua}) { $lei->start_mua; } - $l2m->wq_workers_start('lei2mail', $l2m->{jobs}, + $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'; } - if (!$lei->{opt}->{threads} && locals($self)) { # for query_mset - # lei->{git_tmp} is set for wq_wait_old so we don't - # delete until all lei2mail + lei_xsearch workers are reaped - $lei->{git_tmp} = $self->{git_tmp} = git_tmp($self); - } - $self->wq_workers_start('lei_xsearch', $self->{jobs}, + $self->wq_workers_start('lei_xsearch', undef, $lei->oldset, { lei => $lei }); my $op = delete $lei->{pkt_op_c}; delete $lei->{pkt_op_p}; - $l2m->wq_close(1) if $l2m; + @$end = (); + $self->{threads} = $lei->{opt}->{threads}; + if ($l2m) { + $l2m->net_merge_complete unless $lei->{auth}; + } else { + start_query($self); + } $lei->event_step_init; # wait for shutdowns - $self->wq_io_do('query_prepare', []) if $l2m; # for augment/dedupe - start_query($self, $lei); - $self->wq_close(1); # lei_xsearch workers stop when done if ($lei->{oneshot}) { while ($op->{sock}) { $op->event_step } } @@ -462,8 +479,10 @@ sub prepare_external { return add_uri($self, URI->new($loc)); } elsif (-f "$loc/ei.lock") { require PublicInbox::ExtSearch; + die "`\\n' not allowed in `$loc'\n" if index($loc, "\n") >= 0; $loc = PublicInbox::ExtSearch->new($loc); } elsif (-f "$loc/inbox.lock" || -d "$loc/public-inbox") { + die "`\\n' not allowed in `$loc'\n" if index($loc, "\n") >= 0; require PublicInbox::Inbox; # v2, v1 $loc = bless { inboxdir => $loc }, 'PublicInbox::Inbox'; } else {