Lock: new base class for writable lockers

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

diff --git a/lib/PublicInbox/SearchIdxSkeleton.pm b/lib/PublicInbox/SearchIdxSkeleton.pm
index 4066b5915470872b2fc766e45650534b62560469..54a59ab02a61b5985a1a859c898ae86ca1d0af2a 100644 (file)
--- a/lib/PublicInbox/SearchIdxSkeleton.pm
+++ b/lib/PublicInbox/SearchIdxSkeleton.pm
@@ -15,36 +15,41 @@ 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);
 
        # lock on only exists in parent, not in worker
-       my $l = $self->{lock_path} = $self->xdir . '/pi-v2-skeleton.lock';
-       open my $fh, '>>', $l or die "failed to create $l: $!\n";
+       $self->{lock_path} = $self->xdir . '/pi-v2-skeleton.lock';
        $self;
 }
 
 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 +57,29 @@ 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: $!";
+                       }
+               } elsif ($line =~ /\AD ([a-f0-9]{40,}) (.*)\n\z/s) {
+                       my ($oid, $mid) = ($1, $2);
+                       $xdb ||= $self->_xdb_acquire;
+                       if (!$txn) {
+                               $xdb->begin_transaction;
+                               $txn = 1;
+                       }
+                       $self->remove_by_oid($oid, $mid);
                } else {
                        my $len = int($line);
                        my $n = read($r, my $msg, $len) or die "read: $!\n";
@@ -82,13 +110,23 @@ sub index_skeleton {
        # 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;
+       $self->lock_acquire;
        print $w $str or $err = $!;
-       $self->_lock_release;
+       $self->lock_release;
 
        die "print failed: $err\n" if $err;
 }
 
+sub remote_remove {
+       my ($self, $oid, $mid) = @_;
+       my $err;
+       $self->lock_acquire;
+       eval { $self->SUPER::remote_remove($oid, $mid) };
+       $err = $@;
+       $self->lock_release;
+       die $err if $err;
+}
+
 # values: [ TS, NUM, BYTES, LINES, MID, XPATH, doc_data ]
 sub index_skeleton_real ($$) {
        my ($self, $values) = @_;
@@ -98,9 +136,6 @@ sub index_skeleton_real ($$) {
        my $ts = $values->[PublicInbox::Search::TS];
        my $smsg = PublicInbox::SearchMsg->new(undef);
        my $doc = $smsg->{doc};
-       foreach my $mid (@$mids) {
-               $doc->add_term('Q' . $mid);
-       }
        PublicInbox::SearchIdx::add_values($doc, $values);
        $doc->set_data($doc_data);
        $smsg->{ts} = $ts;
@@ -110,4 +145,21 @@ sub index_skeleton_real ($$) {
        $self->link_and_save($doc, $mids, \@refs, $num, $xpath);
 }
 
+# write to the subprocess
+sub barrier_init {
+       my ($self, $nparts) = @_;
+       my $w = $self->{w};
+       my $err;
+       $self->lock_acquire;
+       print $w "barrier_init $nparts\n" or $err = "failed to write: $!\n";
+       $self->lock_release;
+       die $err if $err;
+}
+
+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;