X-Git-Url: http://git.ozlabs.org/?p=ccan;a=blobdiff_plain;f=ccan%2Ftdb2%2Flock.c;h=5b04c0d243d9b29f2a5104af3f7836a5a47e8f40;hp=82fcdf17d1e6dfd0126d17232826cc53391329ee;hb=4bde5a87c0df66d2000d972a2a0a9992a9d9396d;hpb=2960e90f2d036935273d163593839d3777be7980 diff --git a/ccan/tdb2/lock.c b/ccan/tdb2/lock.c index 82fcdf17..5b04c0d2 100644 --- a/ccan/tdb2/lock.c +++ b/ccan/tdb2/lock.c @@ -30,13 +30,33 @@ #include /* 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; } @@ -51,6 +71,11 @@ static int fcntl_lock(struct tdb_context *tdb, 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); @@ -193,7 +218,8 @@ static enum TDB_ERROR tdb_brunlock(struct tdb_context *tdb, 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", @@ -213,6 +239,9 @@ enum TDB_ERROR tdb_allrecord_upgrade(struct tdb_context *tdb) { 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:" @@ -226,6 +255,10 @@ enum TDB_ERROR tdb_allrecord_upgrade(struct tdb_context *tdb) " 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, @@ -266,6 +299,9 @@ enum TDB_ERROR tdb_lock_and_recover(struct tdb_context *tdb) { 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) { @@ -302,12 +338,17 @@ static enum TDB_ERROR tdb_nest_lock(struct tdb_context *tdb, 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, @@ -470,14 +511,24 @@ enum TDB_ERROR tdb_allrecord_lock(struct tdb_context *tdb, int ltype, 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", @@ -582,6 +633,9 @@ void tdb_unlock_expand(struct tdb_context *tdb, int ltype) /* 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!"); @@ -658,12 +712,17 @@ enum TDB_ERROR tdb_lock_hashes(struct tdb_context *tdb, + (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", @@ -729,6 +788,9 @@ enum TDB_ERROR tdb_lock_free_bucket(struct tdb_context *tdb, tdb_off_t b_off, /* 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, @@ -755,7 +817,27 @@ void tdb_unlock_free_bucket(struct tdb_context *tdb, tdb_off_t b_off) 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;