X-Git-Url: http://git.ozlabs.org/?p=ccan;a=blobdiff_plain;f=ccan%2Ftdb2%2Flock.c;h=bf62d9719e90c60f795ecd8acbce34c99df212b4;hp=8bd9c4f0f4bca5c9a5079a74120919a1ae73c3d5;hb=380372e733416c2b348d5307f536d0a0807e95df;hpb=0693529224e36235fc615fa323e57d5b07879e3f diff --git a/ccan/tdb2/lock.c b/ccan/tdb2/lock.c index 8bd9c4f0..bf62d971 100644 --- a/ccan/tdb2/lock.c +++ b/ccan/tdb2/lock.c @@ -30,7 +30,7 @@ #include /* If we were threaded, we could wait for unlock, but we're not, so fail. */ -static enum TDB_ERROR owner_conflict(struct tdb_context *tdb, const char *call) +enum TDB_ERROR owner_conflict(struct tdb_context *tdb, const char *call) { return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_USE_ERROR, "%s: lock owned by another tdb in this process.", @@ -38,8 +38,7 @@ static enum TDB_ERROR owner_conflict(struct tdb_context *tdb, const char *call) } /* If we fork, we no longer really own locks. */ -static bool check_lock_pid(struct tdb_context *tdb, - const char *call, bool log) +bool check_lock_pid(struct tdb_context *tdb, const char *call, bool log) { /* No locks? No problem! */ if (tdb->file->allrecord_lock.count == 0 @@ -60,34 +59,64 @@ static bool check_lock_pid(struct tdb_context *tdb, return false; } -static int fcntl_lock(struct tdb_context *tdb, - int rw, off_t off, off_t len, bool waitflag) +int tdb_fcntl_lock(int fd, int rw, off_t off, off_t len, bool waitflag, + void *unused) { struct flock fl; + int ret; + + do { + fl.l_type = rw; + fl.l_whence = SEEK_SET; + fl.l_start = off; + fl.l_len = len; + + if (waitflag) + ret = fcntl(fd, F_SETLKW, &fl); + else + ret = fcntl(fd, F_SETLK, &fl); + } while (ret != 0 && errno == EINTR); + return ret; +} + +int tdb_fcntl_unlock(int fd, int rw, off_t off, off_t len, void *unused) +{ + struct flock fl; + int ret; - fl.l_type = rw; - fl.l_whence = SEEK_SET; - fl.l_start = off; - fl.l_len = len; - fl.l_pid = 0; + do { + fl.l_type = F_UNLCK; + fl.l_whence = SEEK_SET; + fl.l_start = off; + fl.l_len = len; + + ret = fcntl(fd, F_SETLKW, &fl); + } while (ret != 0 && errno == EINTR); + return ret; +} +static int lock(struct tdb_context *tdb, + int rw, off_t off, off_t len, bool waitflag) +{ + int ret; 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); - else { - add_stat(tdb, lock_nonblock, 1); - return fcntl(tdb->file->fd, F_SETLK, &fl); + tdb->stats.lock_lowlevel++; + ret = tdb->lock_fn(tdb->file->fd, rw, off, len, waitflag, + tdb->lock_data); + if (!waitflag) { + tdb->stats.lock_nonblock++; + if (ret != 0) + tdb->stats.lock_nonblock_fail++; } + return ret; } -static int fcntl_unlock(struct tdb_context *tdb, int rw, off_t off, off_t len) +static int unlock(struct tdb_context *tdb, int rw, off_t off, off_t len) { - struct flock fl; #if 0 /* Check they matched up locks and unlocks correctly. */ char line[80]; FILE *locks; @@ -146,13 +175,7 @@ static int fcntl_unlock(struct tdb_context *tdb, int rw, off_t off, off_t len) fclose(locks); #endif - fl.l_type = F_UNLCK; - fl.l_whence = SEEK_SET; - fl.l_start = off; - fl.l_len = len; - fl.l_pid = 0; - - return fcntl(tdb->file->fd, F_SETLKW, &fl); + return tdb->unlock_fn(tdb->file->fd, rw, off, len, tdb->lock_data); } /* a byte range locking function - return 0 on success @@ -160,9 +183,9 @@ static int fcntl_unlock(struct tdb_context *tdb, int rw, off_t off, off_t len) note that a len of zero means lock to end of file */ -static enum TDB_ERROR tdb_brlock(struct tdb_context *tdb, - int rw_type, tdb_off_t offset, tdb_off_t len, - enum tdb_lock_flags flags) +enum TDB_ERROR tdb_brlock(struct tdb_context *tdb, + int rw_type, tdb_off_t offset, tdb_off_t len, + enum tdb_lock_flags flags) { int ret; @@ -170,7 +193,7 @@ static enum TDB_ERROR tdb_brlock(struct tdb_context *tdb, return TDB_SUCCESS; } - if (rw_type == F_WRLCK && tdb->read_only) { + if (rw_type == F_WRLCK && (tdb->flags & TDB_RDONLY)) { return tdb_logerr(tdb, TDB_ERR_RDONLY, TDB_LOG_USE_ERROR, "Write lock attempted on read-only database"); } @@ -183,16 +206,13 @@ static enum TDB_ERROR tdb_brlock(struct tdb_context *tdb, (long long)(offset + len)); } - do { - ret = fcntl_lock(tdb, rw_type, offset, len, - flags & TDB_LOCK_WAIT); - } while (ret == -1 && errno == EINTR); - - if (ret == -1) { + ret = lock(tdb, rw_type, offset, len, flags & TDB_LOCK_WAIT); + if (ret != 0) { /* Generic lock error. errno set by fcntl. * EAGAIN is an expected return from non-blocking * locks. */ - if (!(flags & TDB_LOCK_PROBE) && errno != EAGAIN) { + if (!(flags & TDB_LOCK_PROBE) + && (errno != EAGAIN && errno != EINTR)) { tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_ERROR, "tdb_brlock failed (fd=%d) at" " offset %zu rw_type=%d flags=%d len=%zu:" @@ -205,26 +225,22 @@ static enum TDB_ERROR tdb_brlock(struct tdb_context *tdb, return TDB_SUCCESS; } -static enum TDB_ERROR tdb_brunlock(struct tdb_context *tdb, - int rw_type, tdb_off_t offset, size_t len) +enum TDB_ERROR tdb_brunlock(struct tdb_context *tdb, + int rw_type, tdb_off_t offset, size_t len) { - int ret; - if (tdb->flags & TDB_NOLOCK) { return TDB_SUCCESS; } - do { - ret = fcntl_unlock(tdb, rw_type, offset, len); - } while (ret == -1 && errno == EINTR); + if (!check_lock_pid(tdb, "tdb_brunlock", true)) + return TDB_ERR_LOCK; - /* If we fail, *then* we verify that we owned the lock. If not, ok. */ - if (ret == -1 && check_lock_pid(tdb, "tdb_brunlock", false)) { + if (unlock(tdb, rw_type, offset, len) == -1) { return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_ERROR, "tdb_brunlock failed (fd=%d) at offset %zu" - " rw_type=%d len=%zu", + " rw_type=%d len=%zu: %s", tdb->file->fd, (size_t)offset, rw_type, - (size_t)len); + (size_t)len, strerror(errno)); } return TDB_SUCCESS; } @@ -235,7 +251,7 @@ static enum TDB_ERROR tdb_brunlock(struct tdb_context *tdb, deadlock detection and claim a deadlock when progress can be made. For those OSes we may loop for a while. */ -enum TDB_ERROR tdb_allrecord_upgrade(struct tdb_context *tdb) +enum TDB_ERROR tdb_allrecord_upgrade(struct tdb_context *tdb, off_t start) { int count = 1000; @@ -261,8 +277,7 @@ enum TDB_ERROR tdb_allrecord_upgrade(struct tdb_context *tdb) while (count--) { struct timeval tv; - if (tdb_brlock(tdb, F_WRLCK, - TDB_HASH_LOCK_START, 0, + if (tdb_brlock(tdb, F_WRLCK, start, 0, TDB_LOCK_WAIT|TDB_LOCK_PROBE) == TDB_SUCCESS) { tdb->file->allrecord_lock.ltype = F_WRLCK; tdb->file->allrecord_lock.off = 0; @@ -276,8 +291,11 @@ enum TDB_ERROR tdb_allrecord_upgrade(struct tdb_context *tdb) tv.tv_usec = 1; select(0, NULL, NULL, NULL, &tv); } - return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_ERROR, - "tdb_allrecord_upgrade failed"); + + if (errno != EAGAIN && errno != EINTR) + tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_ERROR, + "tdb_allrecord_upgrade failed"); + return TDB_ERR_LOCK; } static struct tdb_lock *find_nestlock(struct tdb_context *tdb, tdb_off_t offset, @@ -308,28 +326,29 @@ enum TDB_ERROR tdb_lock_and_recover(struct tdb_context *tdb) return ecode; } - ecode = tdb_lock_open(tdb, TDB_LOCK_WAIT|TDB_LOCK_NOCHECK); + ecode = tdb_lock_open(tdb, F_WRLCK, TDB_LOCK_WAIT|TDB_LOCK_NOCHECK); if (ecode != TDB_SUCCESS) { tdb_allrecord_unlock(tdb, F_WRLCK); return ecode; } ecode = tdb_transaction_recover(tdb); - tdb_unlock_open(tdb); + tdb_unlock_open(tdb, F_WRLCK); tdb_allrecord_unlock(tdb, F_WRLCK); return ecode; } /* lock an offset in the database. */ -static enum TDB_ERROR tdb_nest_lock(struct tdb_context *tdb, - tdb_off_t offset, int ltype, - enum tdb_lock_flags flags) +enum TDB_ERROR tdb_nest_lock(struct tdb_context *tdb, + tdb_off_t offset, int ltype, + enum tdb_lock_flags flags) { struct tdb_lock *new_lck; enum TDB_ERROR ecode; - if (offset > (TDB_HASH_LOCK_START + TDB_HASH_LOCK_RANGE - + tdb->file->map_size / 8)) { + if (!(tdb->flags & TDB_VERSION1) + && offset > (TDB_HASH_LOCK_START + TDB_HASH_LOCK_RANGE + + tdb->file->map_size / 8)) { return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_ERROR, "tdb_nest_lock: invalid offset %zu ltype=%d", (size_t)offset, ltype); @@ -342,7 +361,7 @@ static enum TDB_ERROR tdb_nest_lock(struct tdb_context *tdb, return TDB_ERR_LOCK; } - add_stat(tdb, locks, 1); + tdb->stats.locks++; new_lck = find_nestlock(tdb, offset, NULL); if (new_lck) { @@ -361,12 +380,14 @@ static enum TDB_ERROR tdb_nest_lock(struct tdb_context *tdb, return TDB_SUCCESS; } +#if 0 if (tdb->file->num_lockrecs && offset >= TDB_HASH_LOCK_START && offset < TDB_HASH_LOCK_START + TDB_HASH_LOCK_RANGE) { return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_ERROR, "tdb_nest_lock: already have a hash lock?"); } +#endif new_lck = (struct tdb_lock *)realloc( tdb->file->lockrecs, @@ -415,8 +436,8 @@ static enum TDB_ERROR tdb_nest_lock(struct tdb_context *tdb, return TDB_SUCCESS; } -static enum TDB_ERROR tdb_nest_unlock(struct tdb_context *tdb, - tdb_off_t off, int ltype) +enum TDB_ERROR tdb_nest_unlock(struct tdb_context *tdb, + tdb_off_t off, int ltype) { struct tdb_lock *lck; enum TDB_ERROR ecode; @@ -471,9 +492,9 @@ void tdb_transaction_unlock(struct tdb_context *tdb, int ltype) /* We only need to lock individual bytes, but Linux merges consecutive locks * so we lock in contiguous ranges. */ -static enum TDB_ERROR tdb_lock_gradual(struct tdb_context *tdb, - int ltype, enum tdb_lock_flags flags, - tdb_off_t off, tdb_off_t len) +enum TDB_ERROR tdb_lock_gradual(struct tdb_context *tdb, + int ltype, enum tdb_lock_flags flags, + tdb_off_t off, tdb_off_t len) { enum TDB_ERROR ecode; enum tdb_lock_flags nb_flags = (flags & ~TDB_LOCK_WAIT); @@ -486,8 +507,9 @@ static enum TDB_ERROR tdb_lock_gradual(struct tdb_context *tdb, } /* First we try non-blocking. */ - if (tdb_brlock(tdb, ltype, off, len, nb_flags) == TDB_SUCCESS) { - return TDB_SUCCESS; + ecode = tdb_brlock(tdb, ltype, off, len, nb_flags); + if (ecode != TDB_ERR_LOCK) { + return ecode; } /* Try locking first half, then second. */ @@ -511,6 +533,12 @@ enum TDB_ERROR tdb_allrecord_lock(struct tdb_context *tdb, int ltype, enum TDB_ERROR ecode; tdb_bool_err berr; + if (tdb->flags & TDB_VERSION1) { + if (tdb1_allrecord_lock(tdb, ltype, flags, upgradable) == -1) + return tdb->last_error; + return TDB_SUCCESS; + } + if (tdb->flags & TDB_NOLOCK) return TDB_SUCCESS; @@ -550,7 +578,7 @@ enum TDB_ERROR tdb_allrecord_lock(struct tdb_context *tdb, int ltype, " can't upgrade a write lock"); } - add_stat(tdb, locks, 1); + tdb->stats.locks++; again: /* Lock hashes, gradually. */ ecode = tdb_lock_gradual(tdb, ltype, flags, TDB_HASH_LOCK_START, @@ -593,14 +621,15 @@ again: goto again; } -enum TDB_ERROR tdb_lock_open(struct tdb_context *tdb, enum tdb_lock_flags flags) +enum TDB_ERROR tdb_lock_open(struct tdb_context *tdb, + int ltype, enum tdb_lock_flags flags) { - return tdb_nest_lock(tdb, TDB_OPEN_LOCK, F_WRLCK, flags); + return tdb_nest_lock(tdb, TDB_OPEN_LOCK, ltype, flags); } -void tdb_unlock_open(struct tdb_context *tdb) +void tdb_unlock_open(struct tdb_context *tdb, int ltype) { - tdb_nest_unlock(tdb, TDB_OPEN_LOCK, F_WRLCK); + tdb_nest_unlock(tdb, TDB_OPEN_LOCK, ltype); } bool tdb_has_open_lock(struct tdb_context *tdb) @@ -624,6 +653,11 @@ 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_VERSION1) { + tdb1_allrecord_unlock(tdb, ltype); + return; + } + if (tdb->flags & TDB_NOLOCK) return; @@ -699,7 +733,7 @@ enum TDB_ERROR tdb_lock_hashes(struct tdb_context *tdb, int ltype, enum tdb_lock_flags waitflag) { /* FIXME: Do this properly, using hlock_range */ - unsigned lock = TDB_HASH_LOCK_START + unsigned l = TDB_HASH_LOCK_START + (hash_lock >> (64 - TDB_HASH_LOCK_RANGE_BITS)); /* a allrecord lock allows us to avoid per chain locks */ @@ -732,14 +766,14 @@ enum TDB_ERROR tdb_lock_hashes(struct tdb_context *tdb, " already have expansion lock"); } - return tdb_nest_lock(tdb, lock, ltype, waitflag); + return tdb_nest_lock(tdb, l, ltype, waitflag); } enum TDB_ERROR tdb_unlock_hashes(struct tdb_context *tdb, tdb_off_t hash_lock, tdb_len_t hash_range, int ltype) { - unsigned lock = TDB_HASH_LOCK_START + unsigned l = TDB_HASH_LOCK_START + (hash_lock >> (64 - TDB_HASH_LOCK_RANGE_BITS)); if (tdb->flags & TDB_NOLOCK) @@ -752,10 +786,15 @@ enum TDB_ERROR tdb_unlock_hashes(struct tdb_context *tdb, return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_ERROR, "tdb_unlock_hashes RO allrecord!"); } + if (tdb->file->allrecord_lock.owner != tdb) { + return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_USE_ERROR, + "tdb_unlock_hashes:" + " not locked by us!"); + } return TDB_SUCCESS; } - return tdb_nest_unlock(tdb, lock, ltype); + return tdb_nest_unlock(tdb, l, ltype); } /* Hash locks use TDB_HASH_LOCK_START + the next 30 bits. @@ -782,6 +821,10 @@ enum TDB_ERROR tdb_lock_free_bucket(struct tdb_context *tdb, tdb_off_t b_off, if (!check_lock_pid(tdb, "tdb_lock_free_bucket", true)) return TDB_ERR_LOCK; + if (tdb->file->allrecord_lock.owner != tdb) { + return owner_conflict(tdb, "tdb_lock_free_bucket"); + } + if (tdb->file->allrecord_lock.ltype == F_WRLCK) return 0; return tdb_logerr(tdb, TDB_ERR_LOCK, TDB_LOG_ERROR, @@ -832,6 +875,10 @@ void tdb_lock_cleanup(struct tdb_context *tdb) { unsigned int i; + /* We don't want to warn: they're allowed to close tdb after fork. */ + if (!check_lock_pid(tdb, "tdb_close", false)) + return; + while (tdb->file->allrecord_lock.count && tdb->file->allrecord_lock.owner == tdb) { tdb_allrecord_unlock(tdb, tdb->file->allrecord_lock.ltype);