lib/PublicInbox/SearchIdxThread.pm

   1 # Copyright (C) 2018 all contributors <meta@public-inbox.org>
   2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
   3 package PublicInbox::SearchIdxThread;
   4 use strict;
   5 use warnings;
   6 use base qw(PublicInbox::SearchIdx);
   7 use Storable qw(freeze thaw);
   8
   9 sub new {
  10         my ($class, $v2ibx) = @_;
  11         my $self = $class->SUPER::new($v2ibx, 1, 'all');
  12         # create the DB:
  13         $self->_xdb_acquire;
  14         $self->_xdb_release;
  15
  16         my ($r, $w);
  17         pipe($r, $w) or die "pipe failed: $!\n";
  18         binmode $r, ':raw';
  19         binmode $w, ':raw';
  20         my $pid = fork;
  21         defined $pid or die "fork failed: $!\n";
  22         if ($pid == 0) {
  23                 close $w;
  24                 eval { thread_worker_loop($self, $r) };
  25                 die "thread worker died: $@\n" if $@;
  26                 exit;
  27         }
  28         $self->{w} = $w;
  29         $self->{pid} = $pid;
  30         close $r;
  31
  32         $w->autoflush(1);
  33
  34         # lock on only exists in parent, not in worker
  35         my $l = $self->{lock_path} = $self->xdir . '/thread.lock';
  36         open my $fh, '>>', $l or die "failed to create $l: $!\n";
  37         $self;
  38 }
  39
  40 sub thread_worker_loop {
  41         my ($self, $r) = @_;
  42         my $msg;
  43         my $xdb = $self->_xdb_acquire;
  44         $xdb->begin_transaction;
  45         my $txn = 1;
  46         while (my $line = $r->getline) {
  47                 if ($line eq "commit\n") {
  48                         $xdb->commit_transaction if $txn;
  49                         $txn = undef;
  50                 } elsif ($line eq "close\n") {
  51                         $self->_xdb_release;
  52                         $xdb = $txn = undef;
  53                 } else {
  54                         read($r, $msg, $line) or die "read failed: $!\n";
  55                         $msg = thaw($msg); # should raise on error
  56                         defined $msg or die "failed to thaw buffer\n";
  57                         if (!$txn) {
  58                                 $xdb->begin_transaction;
  59                                 $txn = 1;
  60                         }
  61                         eval { $self->thread_msg_real(@$msg) };
  62                         warn "failed to index message <$msg->[0]>: $@\n" if $@;
  63                 }
  64         }
  65 }
  66
  67 # called by a partition worker
  68 sub thread_msg {
  69         my ($self, $mid, $ts, $xpath, $doc_data) = @_;
  70         my $w = $self->{w};
  71         my $err;
  72         my $str = freeze([ $mid, $ts, $xpath, $doc_data ]);
  73         my $len = length($str) . "\n";
  74
  75         # multiple processes write to the same pipe, so use flock
  76         $self->_lock_acquire;
  77         print $w $len, $str or $err = $!;
  78         $self->_lock_release;
  79
  80         die "print failed: $err\n" if $err;
  81 }
  82
  83 sub thread_msg_real {
  84         my ($self, $mid, $ts, $xpath, $doc_data) = @_;
  85         my $smsg = $self->lookup_message($mid);
  86         my ($old_tid, $doc_id);
  87         if ($smsg) {
  88                 # convert a ghost to a regular message
  89                 # it will also clobber any existing regular message
  90                 $doc_id = $smsg->{doc_id};
  91                 $old_tid = $smsg->thread_id;
  92         } else {
  93                 $smsg = PublicInbox::SearchMsg->new(undef);
  94                 $smsg->{mid} = $mid;
  95         }
  96         my $doc = $smsg->{doc};
  97         $doc->add_term('XPATH' . $xpath) if defined $xpath;
  98         $doc->add_term('XMID' . $mid);
  99         $doc->set_data($doc_data);
 100         $smsg->{ts} = $ts;
 101         my @refs = ($smsg->references =~ /<([^>]+)>/g);
 102         $self->link_message($smsg, \@refs, $old_tid);
 103         my $db = $self->{xdb};
 104         if (defined $doc_id) {
 105                 $db->replace_document($doc_id, $doc);
 106         } else {
 107                 $doc_id = $db->add_document($doc);
 108         }
 109 }
 110
 111 1;