lib/PublicInbox/OverIdxFork.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::OverIdxFork;
   4 use strict;
   5 use warnings;
   6 use base qw(PublicInbox::OverIdx PublicInbox::Lock);
   7 use Storable qw(freeze thaw);
   8 use IO::Handle;
   9
  10 sub create {
  11         my ($self, $v2writable) = @_;
  12         $self->SUPER::create();
  13         $self->spawn_worker($v2writable) if $v2writable->{parallel};
  14 }
  15
  16 sub spawn_worker {
  17         my ($self, $v2writable) = @_;
  18         my ($r, $w);
  19         pipe($r, $w) or die "pipe failed: $!\n";
  20         my ($barrier_wait, $barrier_note);
  21         pipe($barrier_wait, $barrier_note) or die "pipe failed: $!\n";
  22         binmode $_, ':raw' foreach ($r, $w, $barrier_wait, $barrier_note);
  23         my $pid = fork;
  24         defined $pid or die "fork failed: $!\n";
  25         if ($pid == 0) {
  26                 $v2writable->atfork_child;
  27                 $v2writable = undef;
  28                 close $w;
  29                 close $barrier_wait;
  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 { over_worker_loop($self, $r, $barrier_note) };
  36                 die "over worker died: $@\n" if $@;
  37                 exit;
  38         }
  39         $self->{w} = $w;
  40         $self->{pid} = $pid;
  41         $self->{lock_path} = "$self->{filename}.pipe.lock";
  42         close $r;
  43         close $barrier_note;
  44         $self->{barrier_wait} = $barrier_wait;
  45         $w->autoflush(1);
  46 }
  47
  48 sub over_worker_loop {
  49         my ($self, $r, $barrier_note) = @_;
  50         $barrier_note->autoflush(1);
  51         $0 = 'pi-v2-overview';
  52         $self->begin_lazy;
  53         my $barrier = undef;
  54         while (my $line = $r->getline) {
  55                 if ($line eq "commit\n") {
  56                         $self->commit_lazy;
  57                 } elsif ($line eq "close\n") {
  58                         $self->disconnect;
  59                 } elsif ($line =~ /\Abarrier_init (\d+)\n\z/) {
  60                         my $n = $1 - 1;
  61                         die "barrier in-progress\n" if defined $barrier;
  62                         $barrier = { map { $_ => 1 } (0..$n) };
  63                 } elsif ($line =~ /\Abarrier (\d+)\n\z/) {
  64                         my $part = $1;
  65                         die "no barrier in-progress\n" unless defined $barrier;
  66                         delete $barrier->{$1} or die "unknown barrier: $part\n";
  67                         if ((scalar keys %$barrier) == 0) {
  68                                 $barrier = undef;
  69                                 $self->commit_lazy;
  70                                 print $barrier_note "barrier_done\n" or die
  71                                         "print failed to barrier note: $!";
  72                         }
  73                 } elsif ($line =~ /\AD ([a-f0-9]{40,}) (.*)\n\z/s) {
  74                         my ($oid, $mid) = ($1, $2);
  75                         $self->remove_oid($oid, $mid);
  76                 } else {
  77                         my $len = int($line);
  78                         my $n = read($r, my $msg, $len) or die "read: $!\n";
  79                         $n == $len or die "short read: $n != $len\n";
  80                         $msg = thaw($msg); # should raise on error
  81                         defined $msg or die "failed to thaw buffer\n";
  82                         eval { add_over($self, $msg) };
  83                         warn "failed to index message <$msg->[-1]>: $@\n" if $@;
  84                 }
  85         }
  86         die "$$ $0 dbh not released\n" if $self->{dbh};
  87         die "$$ $0 still in transaction\n" if $self->{txn};
  88 }
  89
  90 # called by a partition worker
  91 # values: [ DS, NUM, BYTES, LINES, TS, MIDS, XPATH, doc_data ]
  92 sub add_over {
  93         my ($self, $values) = @_;
  94         if (my $w = $self->{w}) {
  95                 my $err;
  96                 my $str = freeze($values);
  97                 $str = length($str) . "\n" . $str;
  98
  99                 # multiple processes write to the same pipe, so use flock
 100                 # We can't avoid this lock for <=PIPE_BUF writes, either,
 101                 # because those atomic writes can break up >PIPE_BUF ones
 102                 $self->lock_acquire;
 103                 print $w $str or $err = $!;
 104                 $self->lock_release;
 105
 106                 die "print failed: $err\n" if $err;
 107         } else {
 108                 $self->SUPER::add_over($values);
 109         }
 110 }
 111
 112 sub remove_oid {
 113         my ($self, $oid, $mid) = @_;
 114         if (my $w = $self->{w}) {
 115                 my $err;
 116                 $self->lock_acquire;
 117                 print $w "D $oid $mid\n" or $err = $!;
 118                 $self->lock_release;
 119                 die $err if $err;
 120         } else {
 121                 $self->SUPER::remove_oid($oid, $mid); # OverIdx
 122         }
 123 }
 124
 125 # write to the subprocess
 126 sub barrier_init {
 127         my ($self, $nparts) = @_;
 128         my $w = $self->{w} or return;
 129         my $err;
 130         $self->lock_acquire;
 131         print $w "barrier_init $nparts\n" or $err = $!;
 132         $self->lock_release;
 133         die $err if $err;
 134 }
 135
 136 sub barrier_wait {
 137         my ($self) = @_;
 138         my $bw = $self->{barrier_wait} or return;
 139         my $l = $bw->getline;
 140         $l eq "barrier_done\n" or die "bad response from barrier_wait: $l\n";
 141 }
 142
 143 sub remote_commit {
 144         my ($self) = @_;
 145         if (my $w = $self->{w}) {
 146                 my $err;
 147                 $self->lock_acquire;
 148                 print $w "commit\n" or $err = $!;
 149                 $self->lock_release;
 150                 die $err if $err;
 151         } else {
 152                 $self->commit_lazy;
 153         }
 154 }
 155
 156 # prevent connections when using forked subprocesses
 157 sub connect {
 158         my ($self) = @_;
 159         return if $self->{w};
 160         $self->SUPER::connect;
 161 }
 162
 163 sub remote_close {
 164         my ($self) = @_;
 165         if (my $w = delete $self->{w}) {
 166                 my $pid = delete $self->{pid} or die "no process to wait on\n";
 167                 print $w "close\n" or die "failed to write to pid:$pid: $!\n";
 168                 close $w or die "failed to close pipe for pid:$pid: $!\n";
 169                 waitpid($pid, 0) == $pid or die "remote process did not finish";
 170                 $? == 0 or die ref($self)." pid:$pid exited with: $?";
 171         } else {
 172                 die "transaction in progress $self\n" if $self->{txn};
 173                 $self->disconnect;
 174         }
 175 }
 176
 177 sub commit_fsync {
 178         my ($self) = @_;
 179         return if $self->{w}; # don't bother; main parent can also call this
 180         $self->SUPER::commit_fsync;
 181 }
 182
 183 1;