package PublicInbox::SearchIdx;
use strict;
use warnings;
-use Fcntl qw(:flock :DEFAULT);
+use base qw(PublicInbox::Search PublicInbox::Lock);
use PublicInbox::MIME;
-use base qw(PublicInbox::Search);
use PublicInbox::MID qw/mid_clean id_compress mid_mime mids references/;
use PublicInbox::MsgIter;
use Carp qw(croak);
my ($self) = @_;
my $xdb = delete $self->{xdb} or croak 'not acquired';
$xdb->close;
- _lock_release($self) if $self->{creat};
+ $self->lock_release if $self->{creat};
undef;
}
my $flag = Search::Xapian::DB_OPEN;
if ($self->{creat}) {
require File::Path;
- _lock_acquire($self);
+ $self->lock_acquire;
File::Path::mkpath($dir);
$flag = Search::Xapian::DB_CREATE_OR_OPEN;
}
$self->{xdb} = Search::Xapian::WritableDatabase->new($dir, $flag);
}
-# we only acquire the flock if creating or reindexing;
-# PublicInbox::Import already has the lock on its own.
-sub _lock_acquire {
- my ($self) = @_;
- croak 'already locked' if $self->{lockfh};
- my $lock_path = $self->{lock_path} or return;
- sysopen(my $lockfh, $lock_path, O_WRONLY|O_CREAT) or
- die "failed to open lock $lock_path: $!\n";
- flock($lockfh, LOCK_EX) or die "lock failed: $!\n";
- $self->{lockfh} = $lockfh;
-}
-
-sub _lock_release {
- my ($self) = @_;
- return unless $self->{lock_path};
- my $lockfh = delete $self->{lockfh} or croak 'not locked';
- flock($lockfh, LOCK_UN) or die "unlock failed: $!\n";
- close $lockfh or die "close failed: $!\n";
-}
-
sub add_val ($$$) {
my ($doc, $col, $num) = @_;
$num = Search::Xapian::sortable_serialise($num);
$doc->add_boolean_term('Q' . $_) foreach @$mids;
my $vivified = 0;
+ $self->{skel} and die "Should not have read-only skel here\n";;
foreach my $mid (@$mids) {
$self->each_smsg_by_mid($mid, sub {
my ($cur) = @_;
# remote_* subs are only used by SearchIdxPart and SearchIdxSkeleton
sub remote_commit {
my ($self) = @_;
- print { $self->{w} } "commit\n" or die "failed to write commit: $!";
+ if (my $w = $self->{w}) {
+ print $w "commit\n" or die "failed to write commit: $!";
+ } else {
+ $self->commit_txn_lazy;
+ if (my $skel = $self->{skeleton}) {
+ $skel->commit_txn_lazy;
+ }
+ }
}
sub remote_close {
my ($self) = @_;
- my $pid = delete $self->{pid} or die "no process to wait on\n";
- my $w = delete $self->{w} or die "no pipe to write to\n";
- print $w "close\n" or die "failed to write to pid:$pid: $!\n";
- close $w or die "failed to close pipe for pid:$pid: $!\n";
- waitpid($pid, 0) == $pid or die "remote process did not finish";
- $? == 0 or die ref($self)." pid:$pid exited with: $?";
+ if (my $w = delete $self->{w}) {
+ my $pid = delete $self->{pid} or die "no process to wait on\n";
+ print $w "close\n" or die "failed to write to pid:$pid: $!\n";
+ close $w or die "failed to close pipe for pid:$pid: $!\n";
+ waitpid($pid, 0) == $pid or die "remote process did not finish";
+ $? == 0 or die ref($self)." pid:$pid exited with: $?";
+ } else {
+ die "transaction in progress $self\n" if $self->{txn};
+ $self->_xdb_release if $self->{xdb};
+ }
}
-# triggers remove_by_oid in partition or skeleton
sub remote_remove {
my ($self, $oid, $mid) = @_;
- print { $self->{w} } "D $oid $mid\n" or
- die "failed to write remove $!";
+ if (my $w = $self->{w}) {
+ # triggers remove_by_oid in partition or skeleton
+ print $w "D $oid $mid\n" or die "failed to write remove $!";
+ } else {
+ $self->begin_txn_lazy;
+ $self->remove_by_oid($oid, $mid);
+ }
+}
+
+sub begin_txn_lazy {
+ my ($self) = @_;
+ return if $self->{txn};
+ my $xdb = $self->{xdb} || $self->_xdb_acquire;
+ $xdb->begin_transaction;
+ $self->{txn} = 1;
+}
+
+sub commit_txn_lazy {
+ my ($self) = @_;
+ delete $self->{txn} or return;
+ $self->{xdb}->commit_transaction;
+}
+
+sub worker_done {
+ my ($self) = @_;
+ die "$$ $0 xdb not released\n" if $self->{xdb};
+ die "$$ $0 still in transaction\n" if $self->{txn};
}
1;