lib/PublicInbox/LeiXSearch.pm

   1 # Copyright (C) 2020-2021 all contributors <meta@public-inbox.org>
   2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
   3
   4 # Combine any combination of PublicInbox::Search,
   5 # PublicInbox::ExtSearch, and PublicInbox::LeiSearch objects
   6 # into one Xapian DB
   7 package PublicInbox::LeiXSearch;
   8 use strict;
   9 use v5.10.1;
  10 use parent qw(PublicInbox::LeiSearch PublicInbox::IPC);
  11 use PublicInbox::DS qw(dwaitpid);
  12 use PublicInbox::OpPipe;
  13 use PublicInbox::Import;
  14 use File::Temp 0.19 (); # 0.19 for ->newdir
  15 use File::Spec ();
  16
  17 sub new {
  18         my ($class) = @_;
  19         PublicInbox::Search::load_xapian();
  20         bless {
  21                 qp_flags => $PublicInbox::Search::QP_FLAGS |
  22                                 PublicInbox::Search::FLAG_PURE_NOT(),
  23         }, $class
  24 }
  25
  26 sub attach_external {
  27         my ($self, $ibxish) = @_; # ibxish = ExtSearch or Inbox
  28
  29         if (!$ibxish->can('over') || !$ibxish->over) {
  30                 return push(@{$self->{remotes}}, $ibxish)
  31         }
  32         my $desc = $ibxish->{inboxdir} // $ibxish->{topdir};
  33         my $srch = $ibxish->search or
  34                 return warn("$desc not indexed for Xapian\n");
  35         my @shards = $srch->xdb_shards_flat or
  36                 return warn("$desc has no Xapian shardsXapian\n");
  37
  38         if (delete $self->{xdb}) { # XXX: do we need this?
  39                 # clobber existing {xdb} if amending
  40                 my $expect = delete $self->{nshard};
  41                 my $shards = delete $self->{shards_flat};
  42                 scalar(@$shards) == $expect or die
  43                         "BUG: {nshard}$expect != shards=".scalar(@$shards);
  44
  45                 my $prev = {};
  46                 for my $old_ibxish (@{$self->{shard2ibx}}) {
  47                         next if $prev == $old_ibxish;
  48                         $prev = $old_ibxish;
  49                         my @shards = $old_ibxish->search->xdb_shards_flat;
  50                         push @{$self->{shards_flat}}, @shards;
  51                 }
  52                 my $nr = scalar(@{$self->{shards_flat}});
  53                 $nr == $expect or die
  54                         "BUG: reloaded $nr shards, expected $expect"
  55         }
  56         push @{$self->{shards_flat}}, @shards;
  57         push(@{$self->{shard2ibx}}, $ibxish) for (@shards);
  58 }
  59
  60 # returns a list of local inboxes (or count in scalar context)
  61 sub locals {
  62         my %uniq = map {; "$_" => $_ } @{$_[0]->{shard2ibx} // []};
  63         values %uniq;
  64 }
  65
  66 # called by PublicInbox::Search::xdb
  67 sub xdb_shards_flat { @{$_[0]->{shards_flat} // []} }
  68
  69 # like over->get_art
  70 sub smsg_for {
  71         my ($self, $mitem) = @_;
  72         # cf. https://trac.xapian.org/wiki/FAQ/MultiDatabaseDocumentID
  73         my $nshard = $self->{nshard};
  74         my $docid = $mitem->get_docid;
  75         my $shard = ($docid - 1) % $nshard;
  76         my $num = int(($docid - 1) / $nshard) + 1;
  77         my $smsg = $self->{shard2ibx}->[$shard]->over->get_art($num);
  78         $smsg->{docid} = $docid;
  79         $smsg;
  80 }
  81
  82 sub recent {
  83         my ($self, $qstr, $opt) = @_;
  84         $opt //= {};
  85         $opt->{relevance} //= -2;
  86         $self->mset($qstr //= 'bytes:1..', $opt);
  87 }
  88
  89 sub over {}
  90
  91 sub _mset_more ($$) {
  92         my ($mset, $mo) = @_;
  93         my $size = $mset->size;
  94         $size && (($mo->{offset} += $size) < ($mo->{limit} // 10000));
  95 }
  96
  97 # $startq will EOF when query_prepare is done augmenting and allow
  98 # query_mset and query_thread_mset to proceed.
  99 sub wait_startq ($) {
 100         my ($startq) = @_;
 101         $_[0] = undef;
 102         read($startq, my $query_prepare_done, 1);
 103 }
 104
 105 sub query_thread_mset { # for --thread
 106         my ($self, $lei, $ibxish) = @_;
 107         my $startq = delete $self->{5};
 108         my %sig = $lei->atfork_child_wq($self);
 109         local @SIG{keys %sig} = values %sig;
 110
 111         my ($srch, $over) = ($ibxish->search, $ibxish->over);
 112         unless ($srch && $over) {
 113                 my $desc = $ibxish->{inboxdir} // $ibxish->{topdir};
 114                 warn "$desc not indexed by Xapian\n";
 115                 return;
 116         }
 117         my $mo = { %{$lei->{mset_opt}} };
 118         my $mset;
 119         my $each_smsg = $lei->{ovv}->ovv_each_smsg_cb($lei, $ibxish);
 120         my $dedupe = $lei->{dedupe} // die 'BUG: {dedupe} missing';
 121         $dedupe->prepare_dedupe;
 122         do {
 123                 $mset = $srch->mset($mo->{qstr}, $mo);
 124                 my $ids = $srch->mset_to_artnums($mset, $mo);
 125                 my $ctx = { ids => $ids };
 126                 my $i = 0;
 127                 my %n2item = map { ($ids->[$i++], $_) } $mset->items;
 128                 while ($over->expand_thread($ctx)) {
 129                         for my $n (@{$ctx->{xids}}) {
 130                                 my $smsg = $over->get_art($n) or next;
 131                                 wait_startq($startq) if $startq;
 132                                 next if $dedupe->is_smsg_dup($smsg);
 133                                 my $mitem = delete $n2item{$smsg->{num}};
 134                                 $each_smsg->($smsg, $mitem);
 135                         }
 136                         @{$ctx->{xids}} = ();
 137                 }
 138         } while (_mset_more($mset, $mo));
 139         undef $each_smsg; # drops @io for l2m->{each_smsg_done}
 140         $lei->{ovv}->ovv_atexit_child($lei);
 141 }
 142
 143 sub query_mset { # non-parallel for non-"--thread" users
 144         my ($self, $lei, $srcs) = @_;
 145         my $startq = delete $self->{5};
 146         my %sig = $lei->atfork_child_wq($self);
 147         local @SIG{keys %sig} = values %sig;
 148         my $mo = { %{$lei->{mset_opt}} };
 149         my $mset;
 150         $self->attach_external($_) for @$srcs;
 151         my $each_smsg = $lei->{ovv}->ovv_each_smsg_cb($lei, $self);
 152         my $dedupe = $lei->{dedupe} // die 'BUG: {dedupe} missing';
 153         $dedupe->prepare_dedupe;
 154         do {
 155                 $mset = $self->mset($mo->{qstr}, $mo);
 156                 for my $it ($mset->items) {
 157                         my $smsg = smsg_for($self, $it) or next;
 158                         wait_startq($startq) if $startq;
 159                         next if $dedupe->is_smsg_dup($smsg);
 160                         $each_smsg->($smsg, $it);
 161                 }
 162         } while (_mset_more($mset, $mo));
 163         undef $each_smsg; # drops @io for l2m->{each_smsg_done}
 164         $lei->{ovv}->ovv_atexit_child($lei);
 165 }
 166
 167 sub git {
 168         my ($self) = @_;
 169         my (%seen, @dirs);
 170         my $tmp = File::Temp->newdir('lei_xsrch_git-XXXXXXXX', TMPDIR => 1);
 171         for my $ibx (@{$self->{shard2ibx} // []}) {
 172                 my $d = File::Spec->canonpath($ibx->git->{git_dir});
 173                 $seen{$d} //= push @dirs, "$d/objects\n"
 174         }
 175         my $git_dir = $tmp->dirname;
 176         PublicInbox::Import::init_bare($git_dir);
 177         my $f = "$git_dir/objects/info/alternates";
 178         open my $alt, '>', $f or die "open($f): $!";
 179         print $alt @dirs or die "print $f: $!";
 180         close $alt or die "close $f: $!";
 181         my $git = PublicInbox::Git->new($git_dir);
 182         $git->{-tmp} = $tmp;
 183         $git;
 184 }
 185
 186 sub query_done { # EOF callback
 187         my ($self, $lei) = @_;
 188         my $l2m = delete $lei->{l2m};
 189         if (my $pids = delete $self->{l2m_pids}) {
 190                 my $ipc_worker_reap = $self->can('ipc_worker_reap');
 191                 dwaitpid($_, $ipc_worker_reap, $l2m) for @$pids;
 192         }
 193         $lei->{ovv}->ovv_end($lei);
 194         $lei->start_mua if $l2m && !$l2m->lock_free;
 195         $lei->dclose;
 196 }
 197
 198 sub start_query { # always runs in main (lei-daemon) process
 199         my ($self, $io, $lei, $srcs) = @_;
 200         if (my $l2m = $lei->{l2m}) {
 201                 $lei->{1} = $io->[1];
 202                 $l2m->post_augment($lei);
 203                 $io->[1] = delete $lei->{1};
 204                 $lei->start_mua($io->[3]) if $l2m->lock_free;
 205         }
 206         my $remotes = $self->{remotes} // [];
 207         if ($lei->{opt}->{thread}) {
 208                 for my $ibxish (@$srcs) {
 209                         $self->wq_do('query_thread_mset', $io, $lei, $ibxish);
 210                 }
 211         } else {
 212                 $self->wq_do('query_mset', $io, $lei, $srcs);
 213         }
 214         # TODO
 215         for my $rmt (@$remotes) {
 216                 $self->wq_do('query_thread_mbox', $io, $lei, $rmt);
 217         }
 218         close $io->[0]; # qry_status_wr
 219         @$io = ();
 220 }
 221
 222 sub query_prepare { # called by wq_do
 223         my ($self, $lei) = @_;
 224         my %sig = $lei->atfork_child_wq($self);
 225         local @SIG{keys %sig} = values %sig;
 226         eval { $lei->{l2m}->do_augment($lei) };
 227         $lei->fail($@) if $@;
 228 }
 229
 230 sub sigpipe_handler {
 231         my ($self, $lei_orig, $pids) = @_;
 232         if ($pids) { # one-shot (no event loop)
 233                 kill 'TERM', @$pids;
 234                 kill 'PIPE', $$;
 235         } else {
 236                 $self->wq_kill;
 237                 $self->wq_close;
 238         }
 239         close(delete $lei_orig->{1}) if $lei_orig->{1};
 240 }
 241
 242 sub do_query {
 243         my ($self, $lei_orig, $srcs) = @_;
 244         my ($lei, @io) = $lei_orig->atfork_parent_wq($self);
 245         $io[0] = undef;
 246         pipe(my $done, $io[0]) or die "pipe $!";
 247
 248         $lei_orig->event_step_init; # wait for shutdowns
 249         my $done_op = {
 250                 '' => [ \&query_done, $self, $lei_orig ],
 251                 '!' => [ \&sigpipe_handler, $self, $lei_orig ]
 252         };
 253         my $in_loop = exists $lei_orig->{sock};
 254         $done = PublicInbox::OpPipe->new($done, $done_op, $in_loop);
 255         my $l2m = $lei->{l2m};
 256         if ($l2m) {
 257                 $l2m->pre_augment($lei_orig); # may redirect $lei->{1} for mbox
 258                 $io[1] = $lei_orig->{1};
 259                 my @l2m_io = (undef, @io[1..$#io]);
 260                 pipe(my $startq, $l2m_io[0]) or die "pipe: $!";
 261                 $self->wq_do('query_prepare', \@l2m_io, $lei);
 262                 $io[4] = *STDERR{GLOB}; # don't send l2m->{-wq_s1}
 263                 die "BUG: unexpected \$io[5]: $io[5]" if $io[5];
 264                 fcntl($startq, 1031, 4096) if $^O eq 'linux'; # F_SETPIPE_SZ
 265                 $io[5] = $startq;
 266         }
 267         start_query($self, \@io, $lei, $srcs);
 268         unless ($in_loop) {
 269                 my @pids = $self->wq_close;
 270                 # for the $lei->atfork_child_wq PIPE handler:
 271                 $done_op->{'!'}->[3] = \@pids;
 272                 $done->event_step;
 273                 my $ipc_worker_reap = $self->can('ipc_worker_reap');
 274                 if (my $l2m_pids = delete $self->{l2m_pids}) {
 275                         dwaitpid($_, $ipc_worker_reap, $l2m) for @$l2m_pids;
 276                 }
 277                 dwaitpid($_, $ipc_worker_reap, $self) for @pids;
 278         }
 279 }
 280
 281 sub ipc_atfork_prepare {
 282         my ($self) = @_;
 283         # (0: qry_status_wr, 1: stdout|mbox, 2: stderr,
 284         #  3: sock, 4: $l2m->{-wq_s1}, 5: $startq)
 285         $self->wq_set_recv_modes(qw[+<&= >&= >&= +<&= +<&= <&=]);
 286         $self->SUPER::ipc_atfork_prepare; # PublicInbox::IPC
 287 }
 288
 289 1;