foo/ # "foo" is the name of the index
- ei.lock # lock file to protect global state
+ - open.lock # shared lock for readers (2.0+)
- ALL.git # empty, alternates for inboxes
- ei$SCHEMA_VERSION/$SHARD # per-shard Xapian DB
- ei$SCHEMA_VERSION/over.sqlite3 # overview DB for WWW, IMAP
- ei$SCHEMA_VERSION/misc # misc Xapian DB
-File and directory names are intentionally different from
+Most file and directory names are intentionally different from
analogous v2 names to ensure extindex and v2 inboxes can
easily be distinguished from each other.
=head1 LOCKING
-L<flock(2)> locking exclusively locks the empty ei.lock file
-for all non-atomic operations.
+L<flock(2)> locking exclusively locks the empty C<ei.lock> file
+for all non-atomic operations. In 2.0+, the L<flock(2)>
+C<open.lock> is a shared lock used by readers to protect them
+from L<public-inbox-compact(1)> and L<public-inbox-xcpdb(1)>.
=head1 THANKS
=head1 COPYRIGHT
-Copyright 2020-2021 all contributors L<mailto:meta@public-inbox.org>
+Copyright all contributors L<mailto:meta@public-inbox.org>
License: AGPL-3.0+ L<http://www.gnu.org/licenses/agpl-3.0.txt>
=head1 LOCKING
-L<flock(2)> locking exclusively locks the empty $GIT_DIR/ssoma.lock file
-for all non-atomic operations.
+L<flock(2)> locking exclusively locks the empty C<$GIT_DIR/ssoma.lock>
+file for all non-atomic operations.
=head1 EXAMPLE INPUT FLOW (SERVER-SIDE MDA)
$GIT_DIR/index) is not used at all as there is typically no working
tree.
+=item $GIT_DIR/public-inbox/open.lock
+
+An empty file for L<flock(2)> locking the C<xapian*> directory
+to protect the final phase of L<public-inbox-compact(1)>.
+Introduced in public-inbox 2.0+.
+
=back
Each client $GIT_DIR may have multiple mbox/maildir/command targets.
- xap$SCHEMA_VERSION/$SHARD # per-shard Xapian DB
- xap$SCHEMA_VERSION/over.sqlite3 # OVER-view DB for NNTP, threading
- msgmap.sqlite3 # same as the v1 msgmap
+ - open.lock # protect Xapian shard changes (2.0+)
For blob lookups, the reader only needs to open the "all.git"
repository with $GIT_DIR/objects/info/alternates which references
The SQLite msgmap DB is unchanged from v1, but it is now at the
top-level of the directory.
+=head2 over.lock
+
+An empty file for L<flock(2)> locking the C<xap*> directory to
+protect the final phase of L<public-inbox-compact(1)>.
+Introduced in public-inbox 2.0+.
+
=head1 OBJECT IDENTIFIERS
There are three distinct type of identifiers. content_hash is the
=head1 LOCKING
L<flock(2)> locking exclusively locks the empty inbox.lock file
-for all non-atomic operations.
+for all non-atomic operations. In 2.0+, the L<flock(2)>
+C<open.lock> is a shared lock used by readers to protect them
+from L<public-inbox-compact(1)> and L<public-inbox-xcpdb(1)>.
=head1 HEADERS
=head1 COPYRIGHT
-Copyright 2018-2021 all contributors L<mailto:meta@public-inbox.org>
+Copyright all contributors L<mailto:meta@public-inbox.org>
License: AGPL-3.0+ L<http://www.gnu.org/licenses/agpl-3.0.txt>
@recs = sort { $b->[0] <=> $a->[0] } @recs; # sort by commit time
}
+sub open_lock { "$_[0]->{topdir}/open.lock" }
+
1;
use PublicInbox::Over;
use PublicInbox::Inbox;
use PublicInbox::MiscSearch;
+use PublicInbox::Lock;
use DBI qw(:sql_types); # SQL_BLOB
# for ->reopen, ->mset, ->mset_to_artnums
sub misc {
my ($self) = @_;
- $self->{misc} //= PublicInbox::MiscSearch->new("$self->{xpfx}/misc");
+ $self->{misc} //= do {
+ my $lk = PublicInbox::Lock::may_sh "$self->{topdir}/open.lock";
+ PublicInbox::MiscSearch->new("$self->{xpfx}/misc");
+ };
}
# same as per-inbox ->over, for now...
sub thing_type { 'external index' }
+sub open_lock { "$_[0]->{topdir}/open.lock" }
+
no warnings 'once';
*base_url = \&PublicInbox::Inbox::base_url;
*smsg_eml = \&PublicInbox::Inbox::smsg_eml;
$self->{inboxdir}.($self->version >= 2 ? '/inbox.lock' : '/ssoma.lock')
}
+sub open_lock {
+ my ($self) = @_;
+ $self->{inboxdir}.(version($self) >= 2 ? '/open.lock'
+ : '/public-inbox/open.lock')
+}
+
1;
}
push @{$self->{shards_flat}}, @shards;
push @{$self->{shard_dirs}}, @dirs;
+ push @{$self->{open_locks}}, $srch->open_lock;
push(@{$self->{shard2ibx}}, $ibxish) for (@shards);
}
sub shard_dirs { @{$_[0]->{shard_dirs}} }
+sub xh_lock_args { map { ('-l', $_) } @{$_[0]->{open_locks}} }
+
# returns a list of local inboxes (or count in scalar context)
sub locals { @{$_[0]->{locals} // []} }
# only uses {lock_path} and {lockfh} fields
package PublicInbox::Lock;
use v5.12;
-use Fcntl qw(LOCK_UN LOCK_EX O_RDWR O_CREAT);
-use Carp qw(confess);
+use Fcntl qw(LOCK_UN LOCK_EX O_RDWR O_CREAT LOCK_SH);
+use Carp qw(carp confess);
use PublicInbox::OnDestroy;
-use Errno qw(EINTR);
+use Errno qw(EINTR ENOENT);
use autodie qw(close sysopen syswrite);
sub xflock ($$) {
on_destroy \&lock_release_fast, $self;
}
+sub lock_for_scope_shared ($) {
+ my ($self) = @_;
+ my $fn = $self->{lock_path};
+ if (open(my $fh, '<', $fn)) {
+ xflock($fh, LOCK_SH) or confess "LOCK_SH $fn: $!";
+ $self->{lockfh} = $fh;
+ on_destroy \&lock_release, $self;
+ } else {
+ carp "W: open($fn): $!" if $! != ENOENT;
+ undef;
+ }
+}
+
+sub may_sh ($) {
+ lock_for_scope_shared(new(__PACKAGE__, $_[0]));
+}
+
1;
use strict;
use v5.10.1;
use parent qw(Exporter);
+use PublicInbox::Lock;
our @EXPORT_OK = qw(retry_reopen int_val get_pct xap_terms);
use List::Util qw(max);
use POSIX qw(strftime);
'd=s@', # shard dirs
'g=s', # git dir (with -c)
'k=i', # sort column (like sort(1))
+ 'l=s', # open.lock path
'm=i', # maximum number of results
'o=i', # offset
'r', # 1=relevance then column
}
}
+sub open_lock ($) { "$_[0]->{xpfx}/../open.lock" }
+
# returns all shards as separate Xapian::Database objects w/o combining
sub xdb_shards_flat ($) {
my ($self) = @_;
load_xapian();
$self->{qp_flags} //= $QP_FLAGS;
my $slow_phrase;
+ my $lk = PublicInbox::Lock::may_sh open_lock($self);
my @xdb = map {
$slow_phrase ||= -f "$_/iamchert";
$X{Database}->new($_); # raises if missing
sub reopen {
my ($self) = @_;
if (my $xdb = $self->{xdb}) {
+ my $lk = defined($self->{xpfx}) ?
+ PublicInbox::Lock::may_sh(open_lock($self)) :
+ undef;
$xdb->reopen;
}
$self; # make chaining easier
$dedupe{$x->{prefix}.$sym.$x->{xprefix}} = undef;
}
# TODO: arbitrary header indexing goes here
- [ sort keys %dedupe ];
+ my @apfx = map { ('-Q', $_) } sort keys %dedupe;
+ \@apfx;
};
- ((map { ('-d', $_) } $self->shard_dirs), map { ('-Q', $_) } @$apfx);
+ ((map { ('-d', $_) } $self->shard_dirs), $self->xh_lock_args, @$apfx);
}
+sub xh_lock_args { ('-l', open_lock($_[0])) }
+
sub docids_by_postlist ($$) {
my ($self, $q) = @_;
my $cur = $self->xdb->postlist_begin($q);
use v5.10.1;
use parent qw(PublicInbox::Search PublicInbox::Lock PublicInbox::Umask
Exporter);
+use autodie qw(open);
use PublicInbox::Eml;
use PublicInbox::DS qw(now);
use PublicInbox::Search qw(xap_terms);
if ($version == 1) {
$self->{lock_path} = "$inboxdir/ssoma.lock";
$self->{oidx} = PublicInbox::OverIdx->new(
- $self->xdir.'/over.sqlite3', $creat_opt);
+ "$self->{xpfx}/over.sqlite3", $creat_opt);
} elsif ($version == 2) {
defined $shard or die "shard is required for v2\n";
# shard is a number
File::Path::mkpath($dir);
$self->{-opt}->{cow} or
PublicInbox::Syscall::nodatacow_dir($dir);
+ open my $fh, '>>', $owner->open_lock;
# owner == self for CodeSearchIdx
$self->{-set_has_threadid_once} = 1 if $owner != $self;
$flag |= $DB_DANGEROUS if $self->{-opt}->{dangerous};
$self->idx_acquire;
$self->set_metadata_once;
$self->idx_release;
- if ($v2w->{parallel}) {
- local $self->{-v2w_afc} = $v2w;
- $self->ipc_worker_spawn("shard[$shard]");
- # Increasing the pipe size for requests speeds V2 batch imports
- # across 8 cores by nearly 20%. Since many of our responses
- # are small, make the response pipe as small as possible
- if ($F_SETPIPE_SZ) {
- fcntl($self->{-ipc_req}, $F_SETPIPE_SZ, 1048576);
- fcntl($self->{-ipc_res}, $F_SETPIPE_SZ, 4096);
- }
- }
$self;
}
+sub start_v2w_worker {
+ my ($self, $v2w) = @_;
+ local $self->{-v2w_afc} = $v2w;
+ $self->ipc_worker_spawn("shard[$self->{shard}]");
+ # Increasing the pipe size for requests speeds V2 batch imports
+ # across 8 cores by nearly 20%. Since many of our responses
+ # are small, make the response pipe as small as possible
+ if ($F_SETPIPE_SZ) {
+ fcntl($self->{-ipc_req}, $F_SETPIPE_SZ, 1048576);
+ fcntl($self->{-ipc_res}, $F_SETPIPE_SZ, 4096);
+ }
+}
+
sub _worker_done { # OnDestroy cb
my ($self) = @_;
die "BUG: $$ $0 xdb active" if $self->need_xapian && $self->{xdb};
eval {
my $max = $self->{shards} - 1;
my $idx = $self->{idx_shards} = [];
- for (0..$max) {
+ my $lk = PublicInbox::Lock->new("$self->{xpfx}/../open.lock");
+ my $unlk = $lk->lock_for_scope;
+ for (0..$max) { # no forking, yet
push @$idx, PublicInbox::SearchIdxShard->new($self, $_)
}
+ undef $unlk; # releases lock
$self->{-need_xapian} = $idx->[0]->need_xapian;
+ if ($self->{parallel}) { # fork here:
+ $_->start_v2w_worker($self) for @$idx;
+ }
};
if ($@) {
delete $self->{idx_shards};
use autodie qw(open getsockopt);
use POSIX qw(:signal_h);
use Fcntl qw(LOCK_UN LOCK_EX);
+use PublicInbox::Lock;
use Carp qw(croak);
my $X = \%PublicInbox::Search::X;
our (%SRCH, %WORKERS, $nworker, $workerset, $in, $SHARD_NFD, $MY_FD_MAX);
sub iter_retry_check ($) {
if (ref($@) =~ /\bDatabaseModifiedError\b/) {
- $_[0]->{srch}->reopen;
+ my ($req) = @_;
+ my $lk = PublicInbox::Lock::may_sh $req->{l};
+ $req->{srch}->reopen;
undef; # retries
} elsif (ref($@) =~ /\bDocNotFoundError\b/) {
warn "doc not found: $@";
sub dump_roots_flush ($$) {
my ($req, $fh) = @_;
if ($req->{wbuf} ne '') {
- until (flock($fh, LOCK_EX)) { die "LOCK_EX: $!" if !$!{EINTR} }
+ PublicInbox::Lock::xflock($fh, LOCK_EX) or die "LOCK_EX: $!";
print { $req->{0} } $req->{wbuf} or die "print: $!";
- until (flock($fh, LOCK_UN)) { die "LOCK_UN: $!" if !$!{EINTR} }
+ PublicInbox::Lock::xflock($fh, LOCK_UN) or die "LOCK_UN: $!";
$req->{wbuf} = '';
}
}
for my $retried (0, 1) {
my $slow_phrase = -f "$first/iamchert";
eval {
+ my $lk = PublicInbox::Lock::may_sh $req->{l};
$new->{xdb} = $X->{Database}->new($first);
for (@$dirs) {
$slow_phrase ||= -f "$_/iamchert";
srch_init_extra $new, $req;
$SRCH{$key} = $new;
};
- $req->{srch}->{xdb}->reopen unless $new;
+ unless ($new) {
+ my $lk = PublicInbox::Lock::may_sh $req->{l};
+ $req->{srch}->{xdb}->reopen;
+ }
my $timeo = $req->{K};
alarm($timeo) if $timeo;
$fn->($req, @argv);
use autodie qw(chmod closedir open opendir rename syswrite);
use PublicInbox::Spawn qw(which popen_rd);
use PublicInbox::Syscall;
+use PublicInbox::Lock;
use PublicInbox::Admin qw(setup_signals);
use PublicInbox::Over;
use PublicInbox::Search qw(xap_terms);
# commands with a version number suffix (e.g. "xapian-compact-1.5")
our $XAPIAN_COMPACT = $ENV{XAPIAN_COMPACT} || 'xapian-compact';
our @COMPACT_OPT = qw(jobs|j=i quiet|q blocksize|b=s no-full|n fuller|F);
+my %SKIP = map { $_ => 1 } qw(. ..);
sub commit_changes ($$$$) {
my ($ibx, $im, $tmp, $opt) = @_;
my ($y) = ($b =~ m!/([0-9]+)/*\z!);
($y // -1) <=> ($x // -1) # we may have non-shards
} keys %$tmp;
- my ($xpfx, $mode);
+ my ($xpfx, $mode, $unlk);
if (@order) {
($xpfx) = ($order[0] =~ m!(.*/)[^/]+/*\z!);
+ my $lk = PublicInbox::Lock->new($ibx->open_lock);
+ $unlk = $lk->lock_for_scope;
$mode = (stat($xpfx))[2];
}
for my $old (@order) {
rename $new, $old;
push @old_shard, "$old/old" if $have_old;
}
+ undef $unlk; # unlock
# trigger ->check_inodes in read-only daemons
syswrite($im->{lockfh}, '.') if $over_chg && $im;
}
}
-sub cb_spawn {
- my ($cb, $args, $opt) = @_; # $cb = cpdb() or compact()
+sub cb_spawn ($$$$) {
+ my ($cb, $ibxish, $args, $opt) = @_; # $cb = cpdb() or compact()
my $pid = PublicInbox::DS::fork_persist;
return $pid if $pid > 0;
$SIG{__DIE__} = sub { warn @_; _exit(1) }; # don't jump up stack
- $cb->($args, $opt);
+ $cb->($ibxish, $args, $opt);
_exit(0);
}
kill($sig, keys %$pids); # pids may be empty
}
-sub process_queue {
- my ($queue, $task, $opt) = @_;
+sub process_queue ($$$$) {
+ my ($ibxish, $queue, $task, $opt) = @_;
my $max = $opt->{jobs} // scalar(@$queue);
my $cb = \&$task;
if ($max <= 1) {
while (defined(my $args = shift @$queue)) {
- $cb->($args, $opt);
+ $cb->($ibxish, $args, $opt);
}
return;
}
while (@$queue) {
while (scalar(keys(%pids)) < $max && scalar(@$queue)) {
my $args = shift @$queue;
- $pids{cb_spawn($cb, $args, $opt)} = $args;
+ $pids{cb_spawn($cb, $ibxish, $args, $opt)} = $args;
}
my $flags = 0;
while (defined(my $dn = readdir($dh))) {
if ($dn =~ /\A[0-9]+\z/) {
push(@old_shards, $dn + 0);
- } elsif ($dn eq '.' || $dn eq '..') {
+ } elsif ($SKIP{$dn}) {
} elsif ($dn =~ /\Aover\.sqlite3/) {
} elsif ($dn eq 'misc' && $misc_ok) {
} else {
my $wip_dn = $wip->dirname;
same_fs_or_die($old, $wip_dn);
my $cur = "$old/$dn";
- push @queue, [ $src // $cur , $wip ];
+ push @queue, [ $src // $cur, $wip ];
$opt->{cow} or
PublicInbox::Syscall::nodatacow_dir($wip_dn);
$tmp->{$cur} = $wip;
if ($task eq 'cpdb' && $opt->{reshard} && $ibx->can('cidx_run')) {
cidx_reshard($ibx, $queue, $opt);
} else {
- process_queue($queue, $task, $opt);
+ process_queue $ibx, $queue, $task, $opt;
}
($im // $ibx)->lock_acquire if !$opt->{-coarse_lock};
commit_changes($ibx, $im, $tmp, $opt);
}
# xapian-compact wrapper
-sub compact ($$) { # cb_spawn callback
- my ($args, $opt) = @_;
+sub compact ($$$) { # cb_spawn callback
+ my ($ibxish, $args, $opt) = @_;
my ($src, $newdir) = @$args;
my $dst = ref($newdir) ? $newdir->dirname : $newdir;
my $pfx = $opt->{-progress_pfx} ||= progress_pfx($src);
push @q, [ "$tmp", $wip ];
}
delete $opt->{-progress_pfx};
- process_queue(\@q, 'compact', $opt);
+ process_queue $cidx, \@q, 'compact', $opt;
}
# Like copydatabase(1), this is horribly slow; and it doesn't seem due
# to the overhead of Perl.
-sub cpdb ($$) { # cb_spawn callback
- my ($args, $opt) = @_;
+sub cpdb ($$$) { # cb_spawn callback
+ my ($ibxish, $args, $opt) = @_;
my ($old, $wip) = @$args;
my ($src, $cur_shard);
my $reshard;
my ($X, $flag) = xapian_write_prep($opt);
+ my $lk = PublicInbox::Lock::may_sh $ibxish->open_lock;
if (ref($old) eq 'ARRAY') {
my $new = $wip->dirname;
($cur_shard) = ($new =~ m!(?:xap|ei)[0-9]+/([0-9]+)\b!);
$src = $X->{Database}->new($_);
}
}
- } else {
+ } else { # 1:1 copy
$src = $X->{Database}->new($old);
}
# this is probably the best place to do xapian-compact
# since $dst isn't readable by HTTP or NNTP clients, yet:
- compact([ $tmp, $new ], $opt);
+ compact $ibxish, [ $tmp, $new ], $opt;
}
1;
{
#ifdef HAVE_REALLOCARRAY
void *ret = reallocarray(ptr, nmemb, size);
-#else // can't rely on __builtin_mul_overflow in gcc 4.x :<
+#else // everyone has g++ >=5 these days, right?
void *ret = NULL;
- if (nmemb && size > SIZE_MAX / nmemb)
+
+ if (__builtin_mul_overflow(nmemb, size, 0))
errno = ENOMEM;
else
ret = realloc(ptr, nmemb * size);
KHASHL_CSET_INIT(KH_LOCAL, srch_set, srch_set, struct srch *,
srch_hash, srch_eq)
static srch_set *srch_cache;
-static struct srch *cur_srch; // for ThreadFieldProcessor
+static struct req *cur_req; // for ThreadFieldProcessor
static long my_fd_max, shard_nfd;
// sock_fd is modified in signal handler, yes, it's SOCK_SEQPACKET
static volatile int sock_fd = STDIN_FILENO;
ITER_ABORT
};
-#define MY_ARG_MAX 256
+#define MY_ARG_MAX 256 // FIXME too small?
typedef bool (*cmd)(struct req *);
// only one request per-process since we have RLIMIT_CPU timeout
struct req { // argv and pfxv point into global rbuf
+ char *lockv[MY_ARG_MAX]; // open.lock files
char *argv[MY_ARG_MAX];
char *pfxv[MY_ARG_MAX]; // -A <prefix>
char *qpfxv[MY_ARG_MAX]; // -Q <user_prefix>[:=]<INTERNAL_PREFIX>
unsigned long timeout_sec;
size_t nr_out;
long sort_col; // value column, negative means BoolWeight
- int argc, pfxc, qpfxc, dirc;
+ int argc, pfxc, qpfxc, dirc, lockc;
FILE *fp[2]; // [0] response pipe or sock, [1] status/errors (optional)
bool has_input; // fp[0] is bidirectional
bool collapse_threads;
size_t len;
};
+struct open_locks {
+ struct req *req;
+ int lock_fd[]; // counted by req->lockc
+};
+
#define SPLIT2ARGV(dst,buf,len) split2argv(dst,buf,len,MY_ARRAY_SIZE(dst))
static size_t split2argv(char **dst, char *buf, size_t len, size_t limit)
{
return enq;
}
+static void xclose(int fd)
+{
+ if (close(fd) < 0 && errno != EINTR)
+ EABORT("BUG: close");
+}
+
+// NOT_UNUSED keeps clang happy (tested 14.0.5 on FreeBSD)
+#define NOT_UNUSED(v) (void)v
+#define AUTO_UNLOCK __attribute__((__cleanup__(unlock_ensure)))
+static void unlock_ensure(void *ptr)
+{
+ struct open_locks **lk_ = (struct open_locks **)ptr;
+ struct open_locks *lk = *lk_;
+
+ if (!lk)
+ return;
+ for (int i = 0; i < lk->req->lockc; i++)
+ if (lk->lock_fd[i] >= 0)
+ xclose(lk->lock_fd[i]); // implicit LOCK_UN
+}
+
+static struct open_locks *lock_shared_maybe(struct req *req)
+{
+ struct open_locks *lk = NULL;
+ size_t size;
+
+ assert(req->dirc);
+ if (!req->lockc) {
+ warn("W: %s has no -l (open.lock)", req->dirv[0]);
+ return NULL;
+ }
+ assert(req->lockc > 0);
+ assert(req->lockc < MY_ARG_MAX);
+ if (__builtin_mul_overflow(sizeof(int), (size_t)req->lockc, &size)) {
+ warnx("W: too many locks (%d)", req->lockc);
+ return NULL;
+ }
+ if (__builtin_add_overflow(sizeof(*lk), size, &size)) {
+ warnx("W: too many locks (%d)", req->lockc);
+ return NULL;
+ }
+ lk = (struct open_locks *)malloc(size);
+ if (!lk) EABORT("malloc(%zu)", size);
+ lk->req = req;
+ for (int i = 0; i < req->lockc; i++) {
+ lk->lock_fd[i] = open(req->lockv[i], O_RDONLY);
+
+ if (lk->lock_fd[i] < 0) {
+ if (errno != ENOENT)
+ warn("W: open(%s)", req->lockv[i]);
+ continue;
+ }
+ while (flock(lk->lock_fd[i], LOCK_SH) < 0) {
+ if (errno == EINTR)
+ continue;
+ warn("W: flock(%s, LOCK_SH)", req->lockv[i]);
+ break;
+ }
+ }
+ return lk;
+}
+
static Xapian::MSet enquire_mset(struct req *req, Xapian::Enquire *enq)
{
if (!req->max) {
Xapian::MSet mset = enq->get_mset(req->off, req->max);
return mset;
} catch (const Xapian::DatabaseModifiedError & e) {
- req->srch->db->reopen();
+ AUTO_UNLOCK struct open_locks *lk =
+ lock_shared_maybe(req);
+ req->srch->db->reopen(); // may throw
+ NOT_UNUSED(lk);
}
}
return enq->get_mset(req->off, req->max);
*qry = Xapian::Query(Xapian::Query::OP_FILTER, *qry, f);
return;
} catch (const Xapian::DatabaseModifiedError & e) {
+ AUTO_UNLOCK struct open_locks *lk =
+ lock_shared_maybe(req);
xdb->reopen();
+ NOT_UNUSED(lk);
}
}
}
e ? err(e, "ferror|fclose "#f) : perror("ferror|fclose "#f); \
} while (0)
-static void xclose(int fd)
-{
- if (close(fd) < 0 && errno != EINTR)
- EABORT("BUG: close");
-}
-
static size_t off2size(off_t n)
{
if (n < 0 || (uintmax_t)n > SIZE_MAX)
srch->qp_flags |= FLAG_PHRASE;
i = 0;
try {
+ AUTO_UNLOCK struct open_locks *lk =
+ lock_shared_maybe(req);
srch->db = new Xapian::Database(req->dirv[i]);
if (!lei && is_chert(req->dirv[0]))
srch->qp_flags &= ~FLAG_PHRASE;
srch->qp_flags &= ~FLAG_PHRASE;
srch->db->add_database(Xapian::Database(dir));
}
+ NOT_UNUSED(lk);
break;
} catch (const Xapian::Error & e) {
warnx("E: Xapian::Error: %s (%s)",
case LONG_MAX: case LONG_MIN: ABORT("-k %s", optarg);
}
break;
+ case 'l':
+ req->lockv[req->lockc++] = optarg;
+ if (MY_ARG_MAX == req->lockc) ABORT("too many -l");
+ break;
case 'm': OPT_U(m, req->max, strtoull, ULLONG_MAX); break;
case 'o': OPT_U(o, req->off, strtoull, ULLONG_MAX); break;
case 'r': req->relevance = true; break;
} else {
assert(req->srch != kbuf.srch);
srch_free(kbuf.srch);
+ AUTO_UNLOCK struct open_locks *lk = lock_shared_maybe(req);
req->srch->db->reopen();
+ NOT_UNUSED(lk);
}
if (req->timeout_sec)
alarm(req->timeout_sec > UINT_MAX ?
UINT_MAX : (unsigned)req->timeout_sec);
- cur_srch = req->srch; // set global for *FieldProcessor
+ cur_req = req; // set global for *FieldProcessor
try {
if (!req->fn(req))
warnx("`%s' failed", req->argv[0]);
{
struct req *req = (struct req *)ptr;
free(req->lenv);
- cur_srch = NULL;
+ cur_req = NULL;
}
static void reopen_logs(void)
for (int p = 0; p < req->pfxc; p++)
dump_ibx_term(req, p, &doc, ibx_id);
} catch (const Xapian::DatabaseModifiedError & e) {
+ AUTO_UNLOCK struct open_locks *lk = lock_shared_maybe(req);
req->srch->db->reopen();
+ NOT_UNUSED(lk);
return ITER_RETRY;
} catch (const Xapian::DocNotFoundError & e) { // oh well...
warnx("doc not found: %s", e.get_description().c_str());
for (int p = 0; p < req->pfxc; p++)
dump_roots_term(req, p, drt, &root_offs, &doc);
} catch (const Xapian::DatabaseModifiedError & e) {
+ AUTO_UNLOCK struct open_locks *lk = lock_shared_maybe(req);
req->srch->db->reopen();
+ NOT_UNUSED(lk);
return ITER_RETRY;
} catch (const Xapian::DocNotFoundError & e) { // oh well...
warnx("doc not found: %s", e.get_description().c_str());
*xqry |= Xapian::Query(Xapian::Query::OP_VALUE_RANGE,
column, val, val);
} catch (const Xapian::DatabaseModifiedError &e) {
- cur_srch->db->reopen();
+ AUTO_UNLOCK struct open_locks *lk = lock_shared_maybe(cur_req);
+ cur_req->srch->db->reopen();
+ NOT_UNUSED(lk);
return ITER_RETRY;
} catch (const Xapian::DocNotFoundError &e) { // oh well...
warnx("doc not found: %s", e.get_description().c_str());
static Xapian::Query qry_xpand_col(Xapian::Query qry, unsigned column)
{
Xapian::Query xqry = Xapian::Query::MatchNothing;
- Xapian::Enquire enq(*cur_srch->db);
+ Xapian::Enquire enq(*cur_req->srch->db);
enq.set_weighting_scheme(Xapian::BoolWeight());
enq.set_query(qry);
enq.set_collapse_key(column);
- Xapian::MSet mset = enq.get_mset(0, cur_srch->db->get_doccount());
+ Xapian::MSet mset = enq.get_mset(0, cur_req->srch->db->get_doccount());
for (Xapian::MSetIterator i = mset.begin(); i != mset.end(); i++) {
for (int t = 10; t > 0; --t)
throw Xapian::QueryParserError("missing } in '" + str + "'");
} else { // thread:"{hello world}"
std::string qstr = str.substr(1, str.size() - 2);
- qry = cur_srch->qp->parse_query(qstr, cur_srch->qp_flags);
+ qry = cur_req->srch->qp->parse_query(qstr,
+ cur_req->srch->qp_flags);
}
return qry_xpand_col(qry, THREADID);
}
SKIP: {
have_xapian_compact 2;
- ok(run_script([qw(-compact -q), "$tmp/ext"]), 'compact on full');
- ok(run_script([qw(-compact -q), "$tmp/med"]), 'compact on medium');
+ # ok(run_script([qw(-compact -q), "$tmp/ext"]), 'compact on full');
+ # ok(run_script([qw(-compact -q), "$tmp/med"]), 'compact on medium');
}
my $no_metadata_set = sub {
my $rdr = { 1 => \$out, 2 => \$err };
my $cmd = [ '-compact', $ibx->{inboxdir} ];
-ok(run_script($cmd, undef, $rdr), 'v1 compact works');
+ok(run_script($cmd, undef, $rdr), 'v1 compact works') or diag $err;
@xdir = glob("$ibx->{inboxdir}/public-inbox/xap*");
is(scalar(@xdir), 1, 'got one xapian directory after compact');
};
my @ibx_idx = glob("$v2->{inboxdir}/xap*/?");
-my @ibx_shard_args = map { ('-d', $_) } @ibx_idx;
+my @v2ol = ('-l', "$v2->{inboxdir}/open.lock");
+my @ibx_shard_args = (@v2ol, map { ('-d', $_) } @ibx_idx);
my (@int) = glob("$crepo/public-inbox-cindex/cidx*/?");
my (@ext) = glob("$crepo/cidx-ext/cidx*/?");
is(scalar(@ext), 2, 'have 2 external shards') or diag explain(\@ext);
is(scalar(@int), 1, 'have 1 internal shard') or diag explain(\@int);
+my @ciol = ('-l', "$crepo/public-inbox-cindex/open.lock");
+my @cidx_int_shard_args = (@ciol, map { ('-d', $_) } @int);
my $doreq = sub {
my ($s, @arg) = @_;
my $ar = PublicInbox::AutoReap->new($pid);
diag "$cmd[-1] running pid=$pid";
close $y;
- my $r = $doreq->($s, qw(test_inspect -d), $ibx_idx[0]);
+ my $r = $doreq->($s, qw(test_inspect -d), $ibx_idx[0], @v2ol);
my %info = map { split(/=/, $_, 2) } split(/ /, do { local $/; <$r> });
is($info{has_threadid}, '1', 'has_threadid true for inbox');
like($info{pid}, qr/\A\d+\z/, 'got PID from inbox inspect');
- $r = $doreq->($s, qw(test_inspect -d), $int[0]);
+ $r = $doreq->($s, qw(test_inspect -d), $int[0], @ciol);
my %cinfo = map { split(/=/, $_, 2) } split(/ /, do { local $/; <$r> });
is($cinfo{has_threadid}, '0', 'has_threadid false for cindex');
is($cinfo{pid}, $info{pid}, 'PID unchanged for cindex');
kill('TERM', $cinfo{pid});
my $tries = 0;
do {
- $r = $doreq->($s, qw(test_inspect -d), $ibx_idx[0]);
+ $r = $doreq->($s, qw(test_inspect -d), $ibx_idx[0], @v2ol);
%info = map { split(/=/, $_, 2) }
split(/ /, do { local $/; <$r> });
} while ($info{pid} == $cinfo{pid} && ++$tries < 10);
my %pids;
$tries = 0;
- my @ins = ($s, qw(test_inspect -d), $ibx_idx[0]);
+ my @ins = ($s, qw(test_inspect -d), $ibx_idx[0], @v2ol);
kill('TTIN', $pid);
until (scalar(keys %pids) >= 2 || ++$tries > 100) {
tick;
pipe my $r, my $w;
$xhc->mkreq([ $w, $err_w ], qw(dump_ibx -A XDFID -A Q),
- (map { ('-d', $_) } @ibx_idx),
- 9, "mid:$mid");
+ @ibx_shard_args, 9, "mid:$mid");
close $err_w;
close $w;
my $res = do { local $/; <$r> };
pipe($err_r, $err_w);
pipe $r, $w;
$xhc->mkreq([ $w, $err_w ], qw(dump_roots -c -A XDFID),
- (map { ('-d', $_) } @int),
+ @cidx_int_shard_args,
$root2id_file, 'dt:19700101'.'000000..');
close $err_w;
close $w;
pipe($err_r, $err_w);
pipe($r, $w);
$xhc->mkreq([ $w, $err_w ], qw(dump_roots -c -A),
- "XDFPOST$i", (map { ('-d', $_) } @int),
+ "XDFPOST$i", @cidx_int_shard_args,
$root2id_file, 'dt:19700101'.'000000..');
close $err_w;
close $w;
require PublicInbox::XhcMset;
my $over = $thr->over;
my @thr_idx = glob("$thr->{inboxdir}/xap*/?");
- my @thr_shard_args = map { ('-d', $_) } @thr_idx;
+ my @thr_shard_args = ('-l', "$thr->{inboxdir}/open.lock",
+ map { ('-d', $_) } @thr_idx);
+
my (@art, $mset, $err);
my $capture = sub { ($mset, $err) = @_ };
my $retrieve = sub {
my $t0 = now;
pipe $r, $w;
$xhc->mkreq([ $w ], qw(test_sleep -K 1 -d),
- $ibx_idx[0]);
+ $ibx_idx[0], @v2ol);
close $w;
is readline($r), undef, 'got EOF';
my $diff = now - $t0;
projects / thirdparty / public-inbox.git / commitdiff
