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