v2writable: support "barrier" operation to avoid reforking

[public-inbox.git] / lib / PublicInbox / SearchIdxSkeleton.pm

diff --git a/lib/PublicInbox/SearchIdxSkeleton.pm b/lib/PublicInbox/SearchIdxSkeleton.pm
index aa2713f12743b58944badaa003e67b7215430c68..4cb10f59136cf0fe3f1365f642060a6f2a339f30 100644 (file)
--- a/lib/PublicInbox/SearchIdxSkeleton.pm
+++ b/lib/PublicInbox/SearchIdxSkeleton.pm
@@ -15,21 +15,25 @@ sub new {
 
        my ($r, $w);
        pipe($r, $w) or die "pipe failed: $!\n";
-       binmode $r, ':raw';
-       binmode $w, ':raw';
+       my ($barrier_wait, $barrier_note);
+       pipe($barrier_wait, $barrier_note) or die "pipe failed: $!\n";
+       binmode $_, ':raw' foreach ($r, $w, $barrier_wait, $barrier_note);
        my $pid = fork;
        defined $pid or die "fork failed: $!\n";
        if ($pid == 0) {
                $v2writable->atfork_child;
                $v2writable = undef;
                close $w;
-               eval { skeleton_worker_loop($self, $r) };
+               close $barrier_wait;
+               eval { skeleton_worker_loop($self, $r, $barrier_note) };
                die "skeleton worker died: $@\n" if $@;
                exit;
        }
        $self->{w} = $w;
        $self->{pid} = $pid;
        close $r;
+       close $barrier_note;
+       $self->{barrier_wait} = $barrier_wait;
 
        $w->autoflush(1);
 
@@ -40,11 +44,13 @@ sub new {
 }
 
 sub skeleton_worker_loop {
-       my ($self, $r) = @_;
+       my ($self, $r, $barrier_note) = @_;
+       $barrier_note->autoflush(1);
        $0 = 'pi-v2-skeleton';
        my $xdb = $self->_xdb_acquire;
        $xdb->begin_transaction;
        my $txn = 1;
+       my $barrier = undef;
        while (my $line = $r->getline) {
                if ($line eq "commit\n") {
                        $xdb->commit_transaction if $txn;
@@ -52,6 +58,21 @@ sub skeleton_worker_loop {
                } elsif ($line eq "close\n") {
                        $self->_xdb_release;
                        $xdb = $txn = undef;
+               } elsif ($line =~ /\Abarrier_init (\d+)\n\z/) {
+                       my $n = $1 - 1;
+                       die "barrier in-progress\n" if defined $barrier;
+                       $barrier = { map { $_ => 1 } (0..$n) };
+               } elsif ($line =~ /\Abarrier (\d+)\n\z/) {
+                       my $part = $1;
+                       die "no barrier in-progress\n" unless defined $barrier;
+                       delete $barrier->{$1} or die "unknown barrier: $part\n";
+                       if ((scalar keys %$barrier) == 0) {
+                               $barrier = undef;
+                               $xdb->commit_transaction if $txn;
+                               $txn = undef;
+                               print $barrier_note "barrier_done\n" or die
+                                       "print failed to barrier note: $!";
+                       }
                } else {
                        my $len = int($line);
                        my $n = read($r, my $msg, $len) or die "read: $!\n";
@@ -80,6 +101,8 @@ sub index_skeleton {
        $str = length($str) . "\n" . $str;
 
        # multiple processes write to the same pipe, so use flock
+       # We can't avoid this lock for <=PIPE_BUF writes, either,
+       # because those atomic writes can break up >PIPE_BUF ones
        $self->_lock_acquire;
        print $w $str or $err = $!;
        $self->_lock_release;
@@ -92,34 +115,31 @@ sub index_skeleton_real ($$) {
        my ($self, $values) = @_;
        my $doc_data = pop @$values;
        my $xpath = pop @$values;
-       my $mid = pop @$values;
+       my $mids = pop @$values;
        my $ts = $values->[PublicInbox::Search::TS];
-       my $smsg = $self->lookup_message($mid);
-       my ($old_tid, $doc_id);
-       if ($smsg) {
-               # convert a ghost to a regular message
-               # it will also clobber any existing regular message
-               $doc_id = $smsg->{doc_id};
-               $old_tid = $smsg->thread_id;
-       } else {
-               $smsg = PublicInbox::SearchMsg->new(undef);
-               $smsg->{mid} = $mid;
-       }
+       my $smsg = PublicInbox::SearchMsg->new(undef);
        my $doc = $smsg->{doc};
-       $doc->add_term('XPATH' . $xpath) if defined $xpath;
-       $doc->add_term('XMID' . $mid);
        PublicInbox::SearchIdx::add_values($doc, $values);
        $doc->set_data($doc_data);
        $smsg->{ts} = $ts;
        $smsg->load_from_data($doc_data);
+       my $num = $values->[PublicInbox::Search::NUM];
        my @refs = ($smsg->references =~ /<([^>]+)>/g);
-       $self->link_message($smsg, \@refs, $old_tid);
-       my $db = $self->{xdb};
-       if (defined $doc_id) {
-               $db->replace_document($doc_id, $doc);
-       } else {
-               $doc_id = $db->add_document($doc);
-       }
+       $self->link_and_save($doc, $mids, \@refs, $num, $xpath);
+}
+
+# write to the subprocess
+sub barrier_init {
+       my ($self, $nparts) = @_;
+       my $w = $_[0]->{w};
+       print $w "barrier_init $nparts\n" or die "failed to write: $!";
+       $w->flush or die "failed to flush: $!";
+}
+
+sub barrier_wait {
+       my ($self) = @_;
+       my $l = $self->{barrier_wait}->getline;
+       $l eq "barrier_done\n" or die "bad response from barrier_wait: $l\n";
 }
 
 1;