#include <ccan/build_assert/build_assert.h>
/* If we were threaded, we could wait for unlock, but we're not, so fail. */
-static bool owner_conflict(struct tdb_context *tdb, struct tdb_lock *lock)
+static enum TDB_ERROR owner_conflict(struct tdb_context *tdb, const char *call)
{
- if (lock->owner != tdb) {
- tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_USE_ERROR,
- "Lock already owned by another opener");
+ return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_USE_ERROR,
+ "%s: lock owned by another tdb in this process.",
+ call);
+}
+
+/* If we fork, we no longer really own locks. */
+static bool check_lock_pid(struct tdb_context *tdb,
+ const char *call, bool log)
+{
+ /* No locks? No problem! */
+ if (tdb->file->allrecord_lock.count == 0
+ && tdb->file->num_lockrecs == 0) {
+ return true;
+ }
+
+ /* No fork? No problem! */
+ if (tdb->file->locker == getpid()) {
return true;
}
+
+ if (log) {
+ tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_USE_ERROR,
+ "%s: fork() detected after lock acquisition!"
+ " (%u vs %u)", call, tdb->file->locker, getpid());
+ }
return false;
}
fl.l_len = len;
fl.l_pid = 0;
+ if (tdb->file->allrecord_lock.count == 0
+ && tdb->file->num_lockrecs == 0) {
+ tdb->file->locker = getpid();
+ }
+
add_stat(tdb, lock_lowlevel, 1);
if (waitflag)
return fcntl(tdb->file->fd, F_SETLKW, &fl);
ret = fcntl_unlock(tdb, rw_type, offset, len);
} while (ret == -1 && errno == EINTR);
- if (ret == -1) {
+ /* If we fail, *then* we verify that we owned the lock. If not, ok. */
+ if (ret == -1 && check_lock_pid(tdb, "tdb_brunlock", false)) {
return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_ERROR,
"tdb_brunlock failed (fd=%d) at offset %zu"
" rw_type=%d len=%zu",
{
int count = 1000;
+ if (!check_lock_pid(tdb, "tdb_transaction_prepare_commit", true))
+ return TDB_ERR_LOCK;
+
if (tdb->file->allrecord_lock.count != 1) {
return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_ERROR,
"tdb_allrecord_upgrade failed:"
" already upgraded?");
}
+ if (tdb->file->allrecord_lock.owner != tdb) {
+ return owner_conflict(tdb, "tdb_allrecord_upgrade");
+ }
+
while (count--) {
struct timeval tv;
if (tdb_brlock(tdb, F_WRLCK,
{
enum TDB_ERROR ecode;
+ if (!check_lock_pid(tdb, "tdb_transaction_prepare_commit", true))
+ return TDB_ERR_LOCK;
+
ecode = tdb_allrecord_lock(tdb, F_WRLCK, TDB_LOCK_WAIT|TDB_LOCK_NOCHECK,
false);
if (ecode != TDB_SUCCESS) {
if (tdb->flags & TDB_NOLOCK)
return TDB_SUCCESS;
+ if (!check_lock_pid(tdb, "tdb_nest_lock", true)) {
+ return TDB_ERR_LOCK;
+ }
+
add_stat(tdb, locks, 1);
new_lck = find_nestlock(tdb, offset, NULL);
if (new_lck) {
- if (owner_conflict(tdb, new_lck))
- return -1;
+ if (new_lck->owner != tdb) {
+ return owner_conflict(tdb, "tdb_nest_lock");
+ }
if (new_lck->ltype == F_RDLCK && ltype == F_WRLCK) {
return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_ERROR,
enum TDB_ERROR ecode;
tdb_bool_err berr;
- if (tdb->file->allrecord_lock.count
- && (ltype == F_RDLCK
- || tdb->file->allrecord_lock.ltype == F_WRLCK)) {
- tdb->file->allrecord_lock.count++;
+ if (tdb->flags & TDB_NOLOCK)
return TDB_SUCCESS;
+
+ if (!check_lock_pid(tdb, "tdb_allrecord_lock", true)) {
+ return TDB_ERR_LOCK;
}
if (tdb->file->allrecord_lock.count) {
+ if (tdb->file->allrecord_lock.owner != tdb) {
+ return owner_conflict(tdb, "tdb_allrecord_lock");
+ }
+
+ if (ltype == F_RDLCK
+ || tdb->file->allrecord_lock.ltype == F_WRLCK) {
+ tdb->file->allrecord_lock.count++;
+ return TDB_SUCCESS;
+ }
+
/* a global lock of a different type exists */
return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_USE_ERROR,
"tdb_allrecord_lock: already have %s lock",
/* unlock entire db */
void tdb_allrecord_unlock(struct tdb_context *tdb, int ltype)
{
+ if (tdb->flags & TDB_NOLOCK)
+ return;
+
if (tdb->file->allrecord_lock.count == 0) {
tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_USE_ERROR,
"tdb_allrecord_unlock: not locked!");
+ (hash_lock >> (64 - TDB_HASH_LOCK_RANGE_BITS));
/* a allrecord lock allows us to avoid per chain locks */
- if (tdb->file->allrecord_lock.count &&
- (ltype == tdb->file->allrecord_lock.ltype || ltype == F_RDLCK)) {
- return TDB_SUCCESS;
- }
-
if (tdb->file->allrecord_lock.count) {
+ if (!check_lock_pid(tdb, "tdb_lock_hashes", true))
+ return TDB_ERR_LOCK;
+
+ if (tdb->file->allrecord_lock.owner != tdb)
+ return owner_conflict(tdb, "tdb_lock_hashes");
+ if (ltype == tdb->file->allrecord_lock.ltype
+ || ltype == F_RDLCK) {
+ return TDB_SUCCESS;
+ }
+
return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_USE_ERROR,
"tdb_lock_hashes:"
" already have %s allrecordlock",
/* a allrecord lock allows us to avoid per chain locks */
if (tdb->file->allrecord_lock.count) {
+ if (!check_lock_pid(tdb, "tdb_lock_free_bucket", true))
+ return TDB_ERR_LOCK;
+
if (tdb->file->allrecord_lock.ltype == F_WRLCK)
return 0;
return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_ERROR,
tdb_nest_unlock(tdb, free_lock_off(b_off), F_WRLCK);
}
-void tdb_unlock_all(struct tdb_context *tdb)
+enum TDB_ERROR tdb_lockall(struct tdb_context *tdb)
+{
+ return tdb_allrecord_lock(tdb, F_WRLCK, TDB_LOCK_WAIT, false);
+}
+
+void tdb_unlockall(struct tdb_context *tdb)
+{
+ tdb_allrecord_unlock(tdb, F_WRLCK);
+}
+
+enum TDB_ERROR tdb_lockall_read(struct tdb_context *tdb)
+{
+ return tdb_allrecord_lock(tdb, F_RDLCK, TDB_LOCK_WAIT, false);
+}
+
+void tdb_unlockall_read(struct tdb_context *tdb)
+{
+ tdb_allrecord_unlock(tdb, F_RDLCK);
+}
+
+void tdb_lock_cleanup(struct tdb_context *tdb)
{
unsigned int i;