lib/PublicInbox/Inbox.pm
lib/PublicInbox/Linkify.pm
lib/PublicInbox/Listener.pm
+lib/PublicInbox/Lock.pm
lib/PublicInbox/MDA.pm
lib/PublicInbox/MID.pm
lib/PublicInbox/MIME.pm
package PublicInbox::Import;
use strict;
use warnings;
-use Fcntl qw(:flock :DEFAULT);
+use base qw(PublicInbox::Lock);
use PublicInbox::Spawn qw(spawn);
use PublicInbox::MID qw(mids mid_mime mid2path);
use PublicInbox::Address;
pipe($in_r, $in_w) or die "pipe failed: $!";
pipe($out_r, $out_w) or die "pipe failed: $!";
my $git = $self->{git};
- my $git_dir = $git->{git_dir};
- if (my $lock_path = $self->{lock_path}) {
- sysopen(my $lockfh, $lock_path, O_WRONLY|O_CREAT) or
- die "failed to open lock $lock_path: $!";
- # wait for other processes to be done
- flock($lockfh, LOCK_EX) or die "lock failed: $!\n";
- $self->{lockfh} = $lockfh;
- }
+ $self->lock_acquire;
local $/ = "\n";
chomp($self->{tip} = $git->qx(qw(rev-parse --revs-only), $self->{ref}));
+ my $git_dir = $git->{git_dir};
my @cmd = ('git', "--git-dir=$git_dir", qw(fast-import
--quiet --done --date-format=raw));
my $rdr = { 0 => fileno($out_r), 1 => fileno($in_w) };
_update_git_info($self, 1) if delete $self->{nchg};
- $self->{lock_path} or return;
- my $lockfh = delete $self->{lockfh} or die "BUG: not locked: $!";
- flock($lockfh, LOCK_UN) or die "unlock failed: $!";
- close $lockfh or die "close lock failed: $!";
+ $self->lock_release;
}
sub atfork_child {
--- /dev/null
+# Copyright (C) 2018 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# Base class for per-inbox locking
+package PublicInbox::Lock;
+use strict;
+use warnings;
+use Fcntl qw(:flock :DEFAULT);
+use Carp qw(croak);
+
+# 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";
+}
+
+1;
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);
$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;
}
# 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;
+ $self->lock_acquire;
eval { $self->SUPER::remote_remove($oid, $mid) };
$err = $@;
- $self->_lock_release;
+ $self->lock_release;
die $err if $err;
}
my ($self, $nparts) = @_;
my $w = $self->{w};
my $err;
- $self->_lock_acquire;
+ $self->lock_acquire;
print $w "barrier_init $nparts\n" or $err = "failed to write: $!\n";
- $self->_lock_release;
+ $self->lock_release;
die $err if $err;
}
package PublicInbox::V2Writable;
use strict;
use warnings;
-use Fcntl qw(:flock :DEFAULT);
+use base qw(PublicInbox::Lock);
use PublicInbox::SearchIdxPart;
use PublicInbox::SearchIdxSkeleton;
use PublicInbox::MIME;
sub new {
my ($class, $v2ibx, $creat) = @_;
my $dir = $v2ibx->{mainrepo} or die "no mainrepo in inbox\n";
- my $lock_path = "$dir/inbox.lock";
unless (-d $dir) {
if ($creat) {
require File::Path;
File::Path::mkpath($dir);
- open my $fh, '>>', $lock_path or
- die "failed to open $lock_path: $!\n";
} else {
die "$dir does not exist\n";
}
# limit each repo to 1GB or so
rotate_bytes => int((1024 * 1024 * 1024) / $PACKING_FACTOR),
};
- bless $self, $class
+ bless $self, $class;
}
# returns undef on duplicate or spam
# frequently activated.
delete $ibx->{$_} foreach (qw(git mm search));
+ $self->lock_acquire;
+
# first time initialization, first we create the skeleton pipe:
my $skel = $self->{skel} = PublicInbox::SearchIdxSkeleton->new($self);
my $im = delete $self->{im};
$im->done if $im; # PublicInbox::Import::done
$self->searchidx_checkpoint(0);
+ $self->lock_release;
}
sub checkpoint {
my $im = PublicInbox::Import->new($git, undef, undef, $self->{-inbox});
$im->{bytes_added} = int($packed_bytes / $PACKING_FACTOR);
$im->{want_object_info} = 1;
- $im->{lock_path} = $self->{lock_path};
+ $im->{lock_path} = undef;
$im->{path_type} = 'v2';
$self->{im} = $im;
}