lib/PublicInbox/SearchIdxSkeleton.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::SearchIdxSkeleton;
   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, $v2writable) = @_;
  11         my $self = $class->SUPER::new($v2writable->{-inbox}, 1, 'skel');
  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                 $v2writable->atfork_child;
  24                 $v2writable = undef;
  25                 close $w;
  26                 eval { skeleton_worker_loop($self, $r) };
  27                 die "skeleton worker died: $@\n" if $@;
  28                 exit;
  29         }
  30         $self->{w} = $w;
  31         $self->{pid} = $pid;
  32         close $r;
  33
  34         $w->autoflush(1);
  35
  36         # lock on only exists in parent, not in worker
  37         my $l = $self->{lock_path} = $self->xdir . '/pi-v2-skeleton.lock';
  38         open my $fh, '>>', $l or die "failed to create $l: $!\n";
  39         $self;
  40 }
  41
  42 sub skeleton_worker_loop {
  43         my ($self, $r) = @_;
  44         $0 = 'pi-v2-skeleton';
  45         my $xdb = $self->_xdb_acquire;
  46         $xdb->begin_transaction;
  47         my $txn = 1;
  48         while (my $line = $r->getline) {
  49                 if ($line eq "commit\n") {
  50                         $xdb->commit_transaction if $txn;
  51                         $txn = undef;
  52                 } elsif ($line eq "close\n") {
  53                         $self->_xdb_release;
  54                         $xdb = $txn = undef;
  55                 } else {
  56                         my $len = int($line);
  57                         my $n = read($r, my $msg, $len) or die "read: $!\n";
  58                         $n == $len or die "short read: $n != $len\n";
  59                         $msg = thaw($msg); # should raise on error
  60                         defined $msg or die "failed to thaw buffer\n";
  61                         $xdb ||= $self->_xdb_acquire;
  62                         if (!$txn) {
  63                                 $xdb->begin_transaction;
  64                                 $txn = 1;
  65                         }
  66                         eval { index_skeleton_real($self, $msg) };
  67                         warn "failed to index message <$msg->[-1]>: $@\n" if $@;
  68                 }
  69         }
  70         die "xdb not released\n" if $xdb;
  71         die "in transaction\n" if $txn;
  72 }
  73
  74 # called by a partition worker
  75 sub index_skeleton {
  76         my ($self, $values) = @_;
  77         my $w = $self->{w};
  78         my $err;
  79         my $str = freeze($values);
  80         $str = length($str) . "\n" . $str;
  81
  82         # multiple processes write to the same pipe, so use flock
  83         $self->_lock_acquire;
  84         print $w $str or $err = $!;
  85         $self->_lock_release;
  86
  87         die "print failed: $err\n" if $err;
  88 }
  89
  90 # values: [ TS, NUM, BYTES, LINES, MID, XPATH, doc_data ]
  91 sub index_skeleton_real ($$) {
  92         my ($self, $values) = @_;
  93         my $doc_data = pop @$values;
  94         my $xpath = pop @$values;
  95         my $mid = pop @$values;
  96         my $ts = $values->[PublicInbox::Search::TS];
  97         my $smsg = $self->lookup_message($mid);
  98         my ($old_tid, $doc_id);
  99         if ($smsg) {
 100                 # convert a ghost to a regular message
 101                 # it will also clobber any existing regular message
 102                 $doc_id = $smsg->{doc_id};
 103                 $old_tid = $smsg->thread_id;
 104         } else {
 105                 $smsg = PublicInbox::SearchMsg->new(undef);
 106                 $smsg->{mid} = $mid;
 107         }
 108         my $doc = $smsg->{doc};
 109         $doc->add_term('XPATH' . $xpath) if defined $xpath;
 110         $doc->add_term('XMID' . $mid);
 111         PublicInbox::SearchIdx::add_values($doc, $values);
 112         $doc->set_data($doc_data);
 113         $smsg->{ts} = $ts;
 114         $smsg->load_from_data($doc_data);
 115         my @refs = ($smsg->references =~ /<([^>]+)>/g);
 116         $self->link_message($smsg, \@refs, $old_tid);
 117         my $db = $self->{xdb};
 118         if (defined $doc_id) {
 119                 $db->replace_document($doc_id, $doc);
 120         } else {
 121                 $doc_id = $db->add_document($doc);
 122         }
 123 }
 124
 125 1;