lib/PublicInbox/SearchIdxPart.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::SearchIdxPart;
   4 use strict;
   5 use warnings;
   6 use base qw(PublicInbox::SearchIdx);
   7
   8 sub new {
   9         my ($class, $v2writable, $part) = @_;
  10         my $self = $class->SUPER::new($v2writable->{-inbox}, 1, $part);
  11         # create the DB before forking:
  12         $self->_xdb_acquire;
  13         $self->_xdb_release;
  14         $self->spawn_worker($v2writable, $part) if $v2writable->{parallel};
  15         $self;
  16 }
  17
  18 sub spawn_worker {
  19         my ($self, $v2writable, $part) = @_;
  20         my ($r, $w);
  21         pipe($r, $w) or die "pipe failed: $!\n";
  22         binmode $r, ':raw';
  23         binmode $w, ':raw';
  24         my $pid = fork;
  25         defined $pid or die "fork failed: $!\n";
  26         if ($pid == 0) {
  27                 my $bnote = $v2writable->atfork_child;
  28                 $v2writable = undef;
  29                 close $w or die "failed to close: $!";
  30
  31                 # F_SETPIPE_SZ = 1031 on Linux; increasing the pipe size here
  32                 # speeds V2Writable batch imports across 8 cores by nearly 20%
  33                 fcntl($r, 1031, 1048576) if $^O eq 'linux';
  34
  35                 eval { partition_worker_loop($self, $r, $part, $bnote) };
  36                 die "worker $part died: $@\n" if $@;
  37                 die "unexpected MM $self->{mm}" if $self->{mm};
  38                 exit;
  39         }
  40         $self->{pid} = $pid;
  41         $self->{w} = $w;
  42         close $r or die "failed to close: $!";
  43 }
  44
  45 sub partition_worker_loop ($$$$) {
  46         my ($self, $r, $part, $bnote) = @_;
  47         $0 = "pi-v2-partition[$part]";
  48         $self->begin_txn_lazy;
  49         while (my $line = $r->getline) {
  50                 if ($line eq "commit\n") {
  51                         $self->commit_txn_lazy;
  52                 } elsif ($line eq "close\n") {
  53                         $self->_xdb_release;
  54                 } elsif ($line eq "barrier\n") {
  55                         $self->commit_txn_lazy;
  56                         # no need to lock < 512 bytes is atomic under POSIX
  57                         print $bnote "barrier $part\n" or
  58                                         die "write failed for barrier $!\n";
  59                 } elsif ($line =~ /\AD ([a-f0-9]{40,}) (.+)\n\z/s) {
  60                         my ($oid, $mid) = ($1, $2);
  61                         $self->begin_txn_lazy;
  62                         $self->remove_by_oid($oid, $mid);
  63                 } else {
  64                         chomp $line;
  65                         my ($len, $artnum, $oid, $mid0) = split(/ /, $line);
  66                         $self->begin_txn_lazy;
  67                         my $n = read($r, my $msg, $len) or die "read: $!\n";
  68                         $n == $len or die "short read: $n != $len\n";
  69                         my $mime = PublicInbox::MIME->new(\$msg);
  70                         $artnum = int($artnum);
  71                         $self->add_message($mime, $n, $artnum, $oid, $mid0);
  72                 }
  73         }
  74         $self->worker_done;
  75 }
  76
  77 # called by V2Writable
  78 sub index_raw {
  79         my ($self, $bytes, $msgref, $artnum, $oid, $mid0, $mime) = @_;
  80         if (my $w = $self->{w}) {
  81                 print $w "$bytes $artnum $oid $mid0\n", $$msgref or die
  82                         "failed to write partition $!\n";
  83                 $w->flush or die "failed to flush: $!\n";
  84         } else {
  85                 $$msgref = undef;
  86                 $self->begin_txn_lazy;
  87                 $self->add_message($mime, $bytes, $artnum, $oid, $mid0);
  88         }
  89 }
  90
  91 sub atfork_child {
  92         close $_[0]->{w} or die "failed to close write pipe: $!\n";
  93 }
  94
  95 # called by V2Writable:
  96 sub remote_barrier {
  97         my ($self) = @_;
  98         if (my $w = $self->{w}) {
  99                 print $w "barrier\n" or die "failed to print: $!";
 100                 $w->flush or die "failed to flush: $!";
 101         } else {
 102                 $self->commit_txn_lazy;
 103         }
 104 }
 105
 106 1;