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         my ($barrier_wait, $barrier_note);
  19         pipe($barrier_wait, $barrier_note) or die "pipe failed: $!\n";
  20         binmode $_, ':raw' foreach ($r, $w, $barrier_wait, $barrier_note);
  21         my $pid = fork;
  22         defined $pid or die "fork failed: $!\n";
  23         if ($pid == 0) {
  24                 $v2writable->atfork_child;
  25                 $v2writable = undef;
  26                 close $w;
  27                 close $barrier_wait;
  28                 eval { skeleton_worker_loop($self, $r, $barrier_note) };
  29                 die "skeleton worker died: $@\n" if $@;
  30                 exit;
  31         }
  32         $self->{w} = $w;
  33         $self->{pid} = $pid;
  34         close $r;
  35         close $barrier_note;
  36         $self->{barrier_wait} = $barrier_wait;
  37
  38         $w->autoflush(1);
  39
  40         # lock on only exists in parent, not in worker
  41         my $l = $self->{lock_path} = $self->xdir . '/pi-v2-skeleton.lock';
  42         open my $fh, '>>', $l or die "failed to create $l: $!\n";
  43         $self;
  44 }
  45
  46 sub skeleton_worker_loop {
  47         my ($self, $r, $barrier_note) = @_;
  48         $barrier_note->autoflush(1);
  49         $0 = 'pi-v2-skeleton';
  50         my $xdb = $self->_xdb_acquire;
  51         $xdb->begin_transaction;
  52         my $txn = 1;
  53         my $barrier = undef;
  54         while (my $line = $r->getline) {
  55                 if ($line eq "commit\n") {
  56                         $xdb->commit_transaction if $txn;
  57                         $txn = undef;
  58                 } elsif ($line eq "close\n") {
  59                         $self->_xdb_release;
  60                         $xdb = $txn = undef;
  61                 } elsif ($line =~ /\Abarrier_init (\d+)\n\z/) {
  62                         my $n = $1 - 1;
  63                         die "barrier in-progress\n" if defined $barrier;
  64                         $barrier = { map { $_ => 1 } (0..$n) };
  65                 } elsif ($line =~ /\Abarrier (\d+)\n\z/) {
  66                         my $part = $1;
  67                         die "no barrier in-progress\n" unless defined $barrier;
  68                         delete $barrier->{$1} or die "unknown barrier: $part\n";
  69                         if ((scalar keys %$barrier) == 0) {
  70                                 $barrier = undef;
  71                                 $xdb->commit_transaction if $txn;
  72                                 $txn = undef;
  73                                 print $barrier_note "barrier_done\n" or die
  74                                         "print failed to barrier note: $!";
  75                         }
  76                 } elsif ($line =~ /\AD ([a-f0-9]{40,}) (.*)\n\z/s) {
  77                         my ($oid, $mid) = ($1, $2);
  78                         $xdb ||= $self->_xdb_acquire;
  79                         if (!$txn) {
  80                                 $xdb->begin_transaction;
  81                                 $txn = 1;
  82                         }
  83                         $self->remove_by_oid($oid, $mid);
  84                 } else {
  85                         my $len = int($line);
  86                         my $n = read($r, my $msg, $len) or die "read: $!\n";
  87                         $n == $len or die "short read: $n != $len\n";
  88                         $msg = thaw($msg); # should raise on error
  89                         defined $msg or die "failed to thaw buffer\n";
  90                         $xdb ||= $self->_xdb_acquire;
  91                         if (!$txn) {
  92                                 $xdb->begin_transaction;
  93                                 $txn = 1;
  94                         }
  95                         eval { index_skeleton_real($self, $msg) };
  96                         warn "failed to index message <$msg->[-1]>: $@\n" if $@;
  97                 }
  98         }
  99         die "xdb not released\n" if $xdb;
 100         die "in transaction\n" if $txn;
 101 }
 102
 103 # called by a partition worker
 104 sub index_skeleton {
 105         my ($self, $values) = @_;
 106         my $w = $self->{w};
 107         my $err;
 108         my $str = freeze($values);
 109         $str = length($str) . "\n" . $str;
 110
 111         # multiple processes write to the same pipe, so use flock
 112         # We can't avoid this lock for <=PIPE_BUF writes, either,
 113         # because those atomic writes can break up >PIPE_BUF ones
 114         $self->_lock_acquire;
 115         print $w $str or $err = $!;
 116         $self->_lock_release;
 117
 118         die "print failed: $err\n" if $err;
 119 }
 120
 121 sub remote_remove {
 122         my ($self, $oid, $mid) = @_;
 123         my $err;
 124         $self->_lock_acquire;
 125         eval { $self->SUPER::remote_remove($oid, $mid) };
 126         $err = $@;
 127         $self->_lock_release;
 128         die $err if $err;
 129 }
 130
 131 # values: [ TS, NUM, BYTES, LINES, MID, XPATH, doc_data ]
 132 sub index_skeleton_real ($$) {
 133         my ($self, $values) = @_;
 134         my $doc_data = pop @$values;
 135         my $xpath = pop @$values;
 136         my $mids = pop @$values;
 137         my $ts = $values->[PublicInbox::Search::TS];
 138         my $smsg = PublicInbox::SearchMsg->new(undef);
 139         my $doc = $smsg->{doc};
 140         PublicInbox::SearchIdx::add_values($doc, $values);
 141         $doc->set_data($doc_data);
 142         $smsg->{ts} = $ts;
 143         $smsg->load_from_data($doc_data);
 144         my $num = $values->[PublicInbox::Search::NUM];
 145         my @refs = ($smsg->references =~ /<([^>]+)>/g);
 146         $self->link_and_save($doc, $mids, \@refs, $num, $xpath);
 147 }
 148
 149 # write to the subprocess
 150 sub barrier_init {
 151         my ($self, $nparts) = @_;
 152         my $w = $self->{w};
 153         my $err;
 154         $self->_lock_acquire;
 155         print $w "barrier_init $nparts\n" or $err = "failed to write: $!\n";
 156         $self->_lock_release;
 157         die $err if $err;
 158 }
 159
 160 sub barrier_wait {
 161         my ($self) = @_;
 162         my $l = $self->{barrier_wait}->getline;
 163         $l eq "barrier_done\n" or die "bad response from barrier_wait: $l\n";
 164 }
 165
 166 1;