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 use PublicInbox::Search qw(xap_terms);
  17
  18 sub new {
  19         my ($class) = @_;
  20         PublicInbox::Search::load_xapian();
  21         bless {
  22                 qp_flags => $PublicInbox::Search::QP_FLAGS |
  23                                 PublicInbox::Search::FLAG_PURE_NOT(),
  24         }, $class
  25 }
  26
  27 sub attach_external {
  28         my ($self, $ibxish) = @_; # ibxish = ExtSearch or Inbox
  29         my $desc = $ibxish->{inboxdir} // $ibxish->{topdir};
  30         my $srch = $ibxish->search or
  31                 return warn("$desc not indexed for Xapian\n");
  32         my @shards = $srch->xdb_shards_flat or
  33                 return warn("$desc has no Xapian shardsXapian\n");
  34
  35         if (delete $self->{xdb}) { # XXX: do we need this?
  36                 # clobber existing {xdb} if amending
  37                 my $expect = delete $self->{nshard};
  38                 my $shards = delete $self->{shards_flat};
  39                 scalar(@$shards) == $expect or die
  40                         "BUG: {nshard}$expect != shards=".scalar(@$shards);
  41
  42                 my $prev = {};
  43                 for my $old_ibxish (@{$self->{shard2ibx}}) {
  44                         next if $prev == $old_ibxish;
  45                         $prev = $old_ibxish;
  46                         my @shards = $old_ibxish->search->xdb_shards_flat;
  47                         push @{$self->{shards_flat}}, @shards;
  48                 }
  49                 my $nr = scalar(@{$self->{shards_flat}});
  50                 $nr == $expect or die
  51                         "BUG: reloaded $nr shards, expected $expect"
  52         }
  53         push @{$self->{shards_flat}}, @shards;
  54         push(@{$self->{shard2ibx}}, $ibxish) for (@shards);
  55 }
  56
  57 # returns a list of local inboxes (or count in scalar context)
  58 sub locals { @{$_[0]->{locals} // []} }
  59
  60 sub remotes { @{$_[0]->{remotes} // []} }
  61
  62 # called by PublicInbox::Search::xdb
  63 sub xdb_shards_flat { @{$_[0]->{shards_flat} // []} }
  64
  65 # like over->get_art
  66 sub smsg_for {
  67         my ($self, $mitem) = @_;
  68         # cf. https://trac.xapian.org/wiki/FAQ/MultiDatabaseDocumentID
  69         my $nshard = $self->{nshard};
  70         my $docid = $mitem->get_docid;
  71         my $shard = ($docid - 1) % $nshard;
  72         my $num = int(($docid - 1) / $nshard) + 1;
  73         my $ibx = $self->{shard2ibx}->[$shard];
  74         my $smsg = $ibx->over->get_art($num);
  75         if (ref($ibx->can('msg_keywords'))) {
  76                 my $kw = xap_terms('K', $mitem->get_document);
  77                 $smsg->{kw} = [ sort keys %$kw ];
  78         }
  79         $smsg->{docid} = $docid;
  80         $smsg;
  81 }
  82
  83 sub recent {
  84         my ($self, $qstr, $opt) = @_;
  85         $opt //= {};
  86         $opt->{relevance} //= -2;
  87         $self->mset($qstr //= 'bytes:1..', $opt);
  88 }
  89
  90 sub over {}
  91
  92 sub _mset_more ($$) {
  93         my ($mset, $mo) = @_;
  94         my $size = $mset->size;
  95         $size && (($mo->{offset} += $size) < ($mo->{limit} // 10000));
  96 }
  97
  98 # $startq will EOF when query_prepare is done augmenting and allow
  99 # query_mset and query_thread_mset to proceed.
 100 sub wait_startq ($) {
 101         my ($startq) = @_;
 102         $_[0] = undef;
 103         read($startq, my $query_prepare_done, 1);
 104 }
 105
 106 sub query_thread_mset { # for --thread
 107         my ($self, $lei, $ibxish) = @_;
 108         local $0 = "$0 query_thread_mset";
 109         my $startq = delete $self->{5};
 110         my %sig = $lei->atfork_child_wq($self);
 111         local @SIG{keys %sig} = values %sig;
 112
 113         my ($srch, $over) = ($ibxish->search, $ibxish->over);
 114         unless ($srch && $over) {
 115                 my $desc = $ibxish->{inboxdir} // $ibxish->{topdir};
 116                 warn "$desc not indexed by Xapian\n";
 117                 return;
 118         }
 119         my $mo = { %{$lei->{mset_opt}} };
 120         my $mset;
 121         my $each_smsg = $lei->{ovv}->ovv_each_smsg_cb($lei, $ibxish);
 122         my $dedupe = $lei->{dedupe} // die 'BUG: {dedupe} missing';
 123         $dedupe->prepare_dedupe;
 124         do {
 125                 $mset = $srch->mset($mo->{qstr}, $mo);
 126                 my $ids = $srch->mset_to_artnums($mset, $mo);
 127                 my $ctx = { ids => $ids };
 128                 my $i = 0;
 129                 my %n2item = map { ($ids->[$i++], $_) } $mset->items;
 130                 while ($over->expand_thread($ctx)) {
 131                         for my $n (@{$ctx->{xids}}) {
 132                                 my $smsg = $over->get_art($n) or next;
 133                                 wait_startq($startq) if $startq;
 134                                 next if $dedupe->is_smsg_dup($smsg);
 135                                 my $mitem = delete $n2item{$smsg->{num}};
 136                                 $each_smsg->($smsg, $mitem);
 137                         }
 138                         @{$ctx->{xids}} = ();
 139                 }
 140         } while (_mset_more($mset, $mo));
 141         undef $each_smsg; # drops @io for l2m->{each_smsg_done}
 142         $lei->{ovv}->ovv_atexit_child($lei);
 143 }
 144
 145 sub query_mset { # non-parallel for non-"--thread" users
 146         my ($self, $lei) = @_;
 147         local $0 = "$0 query_mset";
 148         my $startq = delete $self->{5};
 149         my %sig = $lei->atfork_child_wq($self);
 150         local @SIG{keys %sig} = values %sig;
 151         my $mo = { %{$lei->{mset_opt}} };
 152         my $mset;
 153         for my $loc (locals($self)) {
 154                 attach_external($self, $loc);
 155         }
 156         my $each_smsg = $lei->{ovv}->ovv_each_smsg_cb($lei, $self);
 157         my $dedupe = $lei->{dedupe} // die 'BUG: {dedupe} missing';
 158         $dedupe->prepare_dedupe;
 159         do {
 160                 $mset = $self->mset($mo->{qstr}, $mo);
 161                 for my $mitem ($mset->items) {
 162                         my $smsg = smsg_for($self, $mitem) or next;
 163                         wait_startq($startq) if $startq;
 164                         next if $dedupe->is_smsg_dup($smsg);
 165                         $each_smsg->($smsg, $mitem);
 166                 }
 167         } while (_mset_more($mset, $mo));
 168         undef $each_smsg; # drops @io for l2m->{each_smsg_done}
 169         $lei->{ovv}->ovv_atexit_child($lei);
 170 }
 171
 172 sub query_remote_mboxrd {
 173         my ($self, $lei, $uri) = @_;
 174 }
 175
 176 sub git {
 177         my ($self) = @_;
 178         my (%seen, @dirs);
 179         my $tmp = File::Temp->newdir('lei_xsrch_git-XXXXXXXX', TMPDIR => 1);
 180         for my $ibx (@{$self->{shard2ibx} // []}) {
 181                 my $d = File::Spec->canonpath($ibx->git->{git_dir});
 182                 $seen{$d} //= push @dirs, "$d/objects\n"
 183         }
 184         my $git_dir = $tmp->dirname;
 185         PublicInbox::Import::init_bare($git_dir);
 186         my $f = "$git_dir/objects/info/alternates";
 187         open my $alt, '>', $f or die "open($f): $!";
 188         print $alt @dirs or die "print $f: $!";
 189         close $alt or die "close $f: $!";
 190         my $git = PublicInbox::Git->new($git_dir);
 191         $git->{-tmp} = $tmp;
 192         $git;
 193 }
 194
 195 sub query_done { # EOF callback
 196         my ($lei) = @_;
 197         my $has_l2m = exists $lei->{l2m};
 198         for my $f (qw(lxs l2m)) {
 199                 my $wq = delete $lei->{$f} or next;
 200                 $wq->wq_wait_old;
 201         }
 202         $lei->{ovv}->ovv_end($lei);
 203         if ($has_l2m) { # close() calls LeiToMail reap_compress
 204                 close(delete($lei->{1})) if $lei->{1};
 205                 $lei->start_mua;
 206         }
 207         $lei->dclose;
 208 }
 209
 210 sub do_post_augment {
 211         my ($lei, $zpipe, $au_done) = @_;
 212         my $l2m = $lei->{l2m} or die 'BUG: no {l2m}';
 213         eval { $l2m->post_augment($lei, $zpipe) };
 214         if (my $err = $@) {
 215                 if (my $lxs = delete $lei->{lxs}) {
 216                         $lxs->wq_kill;
 217                         $lxs->wq_close;
 218                 }
 219                 $lei->fail("$err");
 220         }
 221         close $au_done; # triggers wait_startq
 222 }
 223
 224 sub start_query { # always runs in main (lei-daemon) process
 225         my ($self, $io, $lei) = @_;
 226         if ($lei->{opt}->{thread}) {
 227                 for my $ibxish (locals($self)) {
 228                         $self->wq_do('query_thread_mset', $io, $lei, $ibxish);
 229                 }
 230         } else {
 231                 $self->wq_do('query_mset', $io, $lei);
 232         }
 233         # TODO
 234         for my $uri (remotes($self)) {
 235                 $self->wq_do('query_remote_mboxrd', $io, $lei, $uri);
 236         }
 237         @$io = ();
 238 }
 239
 240 sub query_prepare { # called by wq_do
 241         my ($self, $lei) = @_;
 242         local $0 = "$0 query_prepare";
 243         my %sig = $lei->atfork_child_wq($self);
 244         -p $lei->{0} or die "BUG: \$done pipe expected";
 245         local @SIG{keys %sig} = values %sig;
 246         eval { $lei->{l2m}->do_augment($lei) };
 247         $lei->fail($@) if $@;
 248         syswrite($lei->{0}, '.') == 1 or die "do_post_augment trigger: $!";
 249 }
 250
 251 sub sigpipe_handler { # handles SIGPIPE from l2m/lxs workers
 252         my ($lei) = @_;
 253         my $lxs = delete $lei->{lxs};
 254         if ($lxs && $lxs->wq_kill_old) {
 255                 kill 'PIPE', $$;
 256                 $lxs->wq_wait_old;
 257         }
 258         close(delete $lei->{1}) if $lei->{1};
 259 }
 260
 261 sub do_query {
 262         my ($self, $lei_orig) = @_;
 263         my ($lei, @io) = $lei_orig->atfork_parent_wq($self);
 264         $io[0] = undef;
 265         pipe(my $done, $io[0]) or die "pipe $!";
 266
 267         $lei_orig->event_step_init; # wait for shutdowns
 268         my $done_op = {
 269                 '' => [ \&query_done, $lei_orig ],
 270                 '!' => [ \&sigpipe_handler, $lei_orig ]
 271         };
 272         my $in_loop = exists $lei_orig->{sock};
 273         $done = PublicInbox::OpPipe->new($done, $done_op, $in_loop);
 274         my $l2m = $lei->{l2m};
 275         if ($l2m) {
 276                 # may redirect $lei->{1} for mbox
 277                 my $zpipe = $l2m->pre_augment($lei_orig);
 278                 $io[1] = $lei_orig->{1};
 279                 pipe(my ($startq, $au_done)) or die "pipe: $!";
 280                 $done_op->{'.'} = [ \&do_post_augment, $lei_orig,
 281                                         $zpipe, $au_done ];
 282                 local $io[4] = *STDERR{GLOB}; # don't send l2m->{-wq_s1}
 283                 die "BUG: unexpected \$io[5]: $io[5]" if $io[5];
 284                 $self->wq_do('query_prepare', \@io, $lei);
 285                 fcntl($startq, 1031, 4096) if $^O eq 'linux'; # F_SETPIPE_SZ
 286                 $io[5] = $startq;
 287                 $io[1] = $zpipe->[1] if $zpipe;
 288         }
 289         start_query($self, \@io, $lei);
 290         $self->wq_close(1);
 291         unless ($in_loop) {
 292                 # for the $lei->atfork_child_wq PIPE handler:
 293                 while ($done->{sock}) { $done->event_step }
 294         }
 295 }
 296
 297 sub ipc_atfork_prepare {
 298         my ($self) = @_;
 299         # (0: done_wr, 1: stdout|mbox, 2: stderr,
 300         #  3: sock, 4: $l2m->{-wq_s1}, 5: $startq)
 301         $self->wq_set_recv_modes(qw[+<&= >&= >&= +<&= +<&= <&=]);
 302         $self->SUPER::ipc_atfork_prepare; # PublicInbox::IPC
 303 }
 304
 305 sub prepare_external {
 306         my ($self, $loc, $boost) = @_; # n.b. already ordered by boost
 307         if (ref $loc) { # already a URI, or PublicInbox::Inbox-like object
 308                 return push(@{$self->{remotes}}, $loc) if $loc->can('scheme');
 309         } elsif ($loc =~ m!\Ahttps?://!) {
 310                 require URI;
 311                 return push(@{$self->{remotes}}, URI->new($loc));
 312         } elsif (-f "$loc/ei.lock") {
 313                 require PublicInbox::ExtSearch;
 314                 $loc = PublicInbox::ExtSearch->new($loc);
 315         } elsif (-f "$loc/inbox.lock" || -d "$loc/public-inbox") {
 316                 require PublicInbox::Inbox; # v2, v1
 317                 $loc = bless { inboxdir => $loc }, 'PublicInbox::Inbox';
 318         } else {
 319                 warn "W: ignoring $loc, unable to determine type\n";
 320                 return;
 321         }
 322         push @{$self->{locals}}, $loc;
 323 }
 324
 325
 326 1;