tdb: tdb_allrecord_lock/tdb_allrecord_unlock/tdb_allrecord_upgrade

Centralize locking of all chains of the tdb; rename _tdb_lockall to
tdb_allrecord_lock and _tdb_unlockall to tdb_allrecord_unlock, and
tdb_brlock_upgrade to tdb_allrecord_upgrade.

Then we use this in the transaction code.  Unfortunately, if the transaction
code records that it has grabbed the allrecord lock read-only, write locks
will fail, so we treat this upgradable lock as a write lock, and mark it
as upgradable using the otherwise-unused offset field.

One subtlety: now the transaction code is using the allrecord_lock, the
tdb_release_extra_locks() function drops it for us, so we no longer need
to do it manually in _tdb_transaction_cancel.

Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>

diff --git a/ccan/tdb/lock.c b/ccan/tdb/lock.c
index b1153b2e6f72b20f81e65f135d1d56f051036727..a59ba365213e473630617ee69ff1f238585d6782 100644 (file)
--- a/ccan/tdb/lock.c
+++ b/ccan/tdb/lock.c
@@ -197,13 +197,29 @@ int 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.  
 */
   deadlock detection and claim a deadlock when progress can be
   made. For those OSes we may loop for a while.  
 */

-int tdb_brlock_upgrade(struct tdb_context *tdb, tdb_off_t offset, size_t len)
+int tdb_allrecord_upgrade(struct tdb_context *tdb)

 {
        int count = 1000;
 {
        int count = 1000;

+
+       if (tdb->allrecord_lock.count != 1) {
+               TDB_LOG((tdb, TDB_DEBUG_ERROR,
+                        "tdb_allrecord_upgrade failed: count %u too high\n",
+                        tdb->allrecord_lock.count));
+               return -1;
+       }
+
+       if (tdb->allrecord_lock.off != 1) {
+               TDB_LOG((tdb, TDB_DEBUG_ERROR,
+                        "tdb_allrecord_upgrade failed: already upgraded?\n"));
+               return -1;
+       }
+

        while (count--) {
                struct timeval tv;
        while (count--) {
                struct timeval tv;

-               if (tdb_brlock(tdb, F_WRLCK, offset, len,
+               if (tdb_brlock(tdb, F_WRLCK, FREELIST_TOP, 0,

                               TDB_LOCK_WAIT|TDB_LOCK_PROBE) == 0) {
                               TDB_LOCK_WAIT|TDB_LOCK_PROBE) == 0) {

+                       tdb->allrecord_lock.ltype = F_WRLCK;
+                       tdb->allrecord_lock.off = 0;

                        return 0;
                }
                if (errno != EDEADLK) {
                        return 0;
                }
                if (errno != EDEADLK) {


@@ -214,7 +230,7 @@ int tdb_brlock_upgrade(struct tdb_context *tdb, tdb_off_t offset, size_t len)
                tv.tv_usec = 1;
                select(0, NULL, NULL, NULL, &tv);
        }
                tv.tv_usec = 1;
                select(0, NULL, NULL, NULL, &tv);
        }

-       TDB_LOG((tdb, TDB_DEBUG_TRACE,"tdb_brlock_upgrade failed at offset %d\n", offset));
+       TDB_LOG((tdb, TDB_DEBUG_TRACE,"tdb_allrecord_upgrade failed\n"));

        return -1;
 }
 
        return -1;
 }
 


@@ -420,11 +436,10 @@ int tdb_transaction_unlock(struct tdb_context *tdb, int ltype)
 }
 
 
 }
 
 

-
-
-/* lock/unlock entire database */
-static int _tdb_lockall(struct tdb_context *tdb, int ltype,
-                       enum tdb_lock_flags flags)
+/* lock/unlock entire database.  It can only be upgradable if you have some
+ * other way of guaranteeing exclusivity (ie. transaction write lock). */
+int tdb_allrecord_lock(struct tdb_context *tdb, int ltype,
+                      enum tdb_lock_flags flags, bool upgradable)

 {
        /* There are no locks on read-only dbs */
        if (tdb->read_only || tdb->traverse_read) {
 {
        /* There are no locks on read-only dbs */
        if (tdb->read_only || tdb->traverse_read) {


@@ -449,6 +464,12 @@ static int _tdb_lockall(struct tdb_context *tdb, int ltype,
                return -1;
        }
 
                return -1;
        }
 

+       if (upgradable && ltype != F_RDLCK) {
+               /* tdb error: you can't upgrade a write lock! */
+               tdb->ecode = TDB_ERR_LOCK;
+               return -1;
+       }
+

        if (tdb->methods->brlock(tdb, ltype, FREELIST_TOP, 0, flags)) {
                if (flags & TDB_LOCK_WAIT) {
                        TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_lockall failed (%s)\n", strerror(errno)));
        if (tdb->methods->brlock(tdb, ltype, FREELIST_TOP, 0, flags)) {
                if (flags & TDB_LOCK_WAIT) {
                        TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_lockall failed (%s)\n", strerror(errno)));


@@ -457,7 +478,10 @@ static int _tdb_lockall(struct tdb_context *tdb, int ltype,
        }
 
        tdb->allrecord_lock.count = 1;
        }
 
        tdb->allrecord_lock.count = 1;

-       tdb->allrecord_lock.ltype = ltype;
+       /* If it's upgradable, it's actually exclusive so we can treat
+        * it as a write lock. */
+       tdb->allrecord_lock.ltype = upgradable ? F_WRLCK : ltype;
+       tdb->allrecord_lock.off = upgradable;

 
        return 0;
 }
 
        return 0;
 }


@@ -465,7 +489,7 @@ static int _tdb_lockall(struct tdb_context *tdb, int ltype,
 
 
 /* unlock entire db */
 
 
 /* unlock entire db */

-static int _tdb_unlockall(struct tdb_context *tdb, int ltype, bool mark_lock)
+int tdb_allrecord_unlock(struct tdb_context *tdb, int ltype, bool mark_lock)

 {
        /* There are no locks on read-only dbs */
        if (tdb->read_only || tdb->traverse_read) {
 {
        /* There are no locks on read-only dbs */
        if (tdb->read_only || tdb->traverse_read) {


@@ -473,7 +497,14 @@ static int _tdb_unlockall(struct tdb_context *tdb, int ltype, bool mark_lock)
                return -1;
        }
 
                return -1;
        }
 

-       if (tdb->allrecord_lock.ltype != ltype || tdb->allrecord_lock.count == 0) {
+       if (tdb->allrecord_lock.count == 0) {
+               tdb->ecode = TDB_ERR_LOCK;
+               return -1;
+       }
+
+       /* Upgradable locks are marked as write locks. */
+       if (tdb->allrecord_lock.ltype != ltype
+           && (!tdb->allrecord_lock.off || ltype != F_RDLCK)) {

                tdb->ecode = TDB_ERR_LOCK;
                return -1;
        }
                tdb->ecode = TDB_ERR_LOCK;
                return -1;
        }


@@ -499,27 +530,27 @@ static int _tdb_unlockall(struct tdb_context *tdb, int ltype, bool mark_lock)
 int tdb_lockall(struct tdb_context *tdb)
 {
        tdb_trace(tdb, "tdb_lockall");
 int tdb_lockall(struct tdb_context *tdb)
 {
        tdb_trace(tdb, "tdb_lockall");

-       return _tdb_lockall(tdb, F_WRLCK, TDB_LOCK_WAIT);
+       return tdb_allrecord_lock(tdb, F_WRLCK, TDB_LOCK_WAIT, false);

 }
 
 /* lock entire database with write lock - mark only */
 int tdb_lockall_mark(struct tdb_context *tdb)
 {
        tdb_trace(tdb, "tdb_lockall_mark");
 }
 
 /* lock entire database with write lock - mark only */
 int tdb_lockall_mark(struct tdb_context *tdb)
 {
        tdb_trace(tdb, "tdb_lockall_mark");

-       return _tdb_lockall(tdb, F_WRLCK, TDB_LOCK_MARK_ONLY);
+       return tdb_allrecord_lock(tdb, F_WRLCK, TDB_LOCK_MARK_ONLY, false);

 }
 
 /* unlock entire database with write lock - unmark only */
 int tdb_lockall_unmark(struct tdb_context *tdb)
 {
        tdb_trace(tdb, "tdb_lockall_unmark");
 }
 
 /* unlock entire database with write lock - unmark only */
 int tdb_lockall_unmark(struct tdb_context *tdb)
 {
        tdb_trace(tdb, "tdb_lockall_unmark");

-       return _tdb_unlockall(tdb, F_WRLCK, true);
+       return tdb_allrecord_unlock(tdb, F_WRLCK, true);

 }
 
 /* lock entire database with write lock - nonblocking varient */
 int tdb_lockall_nonblock(struct tdb_context *tdb)
 {
 }
 
 /* lock entire database with write lock - nonblocking varient */
 int tdb_lockall_nonblock(struct tdb_context *tdb)
 {

-       int ret = _tdb_lockall(tdb, F_WRLCK, TDB_LOCK_NOWAIT);
+       int ret = tdb_allrecord_lock(tdb, F_WRLCK, TDB_LOCK_NOWAIT, false);

        tdb_trace_ret(tdb, "tdb_lockall_nonblock", ret);
        return ret;
 }
        tdb_trace_ret(tdb, "tdb_lockall_nonblock", ret);
        return ret;
 }


@@ -528,20 +559,20 @@ int tdb_lockall_nonblock(struct tdb_context *tdb)
 int tdb_unlockall(struct tdb_context *tdb)
 {
        tdb_trace(tdb, "tdb_unlockall");
 int tdb_unlockall(struct tdb_context *tdb)
 {
        tdb_trace(tdb, "tdb_unlockall");

-       return _tdb_unlockall(tdb, F_WRLCK, false);
+       return tdb_allrecord_unlock(tdb, F_WRLCK, false);

 }
 
 /* lock entire database with read lock */
 int tdb_lockall_read(struct tdb_context *tdb)
 {
        tdb_trace(tdb, "tdb_lockall_read");
 }
 
 /* lock entire database with read lock */
 int tdb_lockall_read(struct tdb_context *tdb)
 {
        tdb_trace(tdb, "tdb_lockall_read");

-       return _tdb_lockall(tdb, F_RDLCK, TDB_LOCK_WAIT);
+       return tdb_allrecord_lock(tdb, F_RDLCK, TDB_LOCK_WAIT, false);

 }
 
 /* lock entire database with read lock - nonblock varient */
 int tdb_lockall_read_nonblock(struct tdb_context *tdb)
 {
 }
 
 /* lock entire database with read lock - nonblock varient */
 int tdb_lockall_read_nonblock(struct tdb_context *tdb)
 {

-       int ret = _tdb_lockall(tdb, F_RDLCK, TDB_LOCK_NOWAIT);
+       int ret = tdb_allrecord_lock(tdb, F_RDLCK, TDB_LOCK_NOWAIT, false);

        tdb_trace_ret(tdb, "tdb_lockall_read_nonblock", ret);
        return ret;
 }
        tdb_trace_ret(tdb, "tdb_lockall_read_nonblock", ret);
        return ret;
 }


@@ -550,7 +581,7 @@ int tdb_lockall_read_nonblock(struct tdb_context *tdb)
 int tdb_unlockall_read(struct tdb_context *tdb)
 {
        tdb_trace(tdb, "tdb_unlockall_read");
 int tdb_unlockall_read(struct tdb_context *tdb)
 {
        tdb_trace(tdb, "tdb_unlockall_read");

-       return _tdb_unlockall(tdb, F_RDLCK, false);
+       return tdb_allrecord_unlock(tdb, F_RDLCK, false);

 }
 
 /* lock/unlock one hash chain. This is meant to be used to reduce
 }
 
 /* lock/unlock one hash chain. This is meant to be used to reduce


@@ -661,7 +692,8 @@ bool tdb_have_extra_locks(struct tdb_context *tdb)
 {
        unsigned int extra = tdb->num_lockrecs;
 
 {
        unsigned int extra = tdb->num_lockrecs;
 

-       if (tdb->allrecord_lock.count) {
+       /* A transaction holds the lock for all records. */
+       if (!tdb->transaction && tdb->allrecord_lock.count) {

                return true;
        }
 
                return true;
        }
 

diff --git a/ccan/tdb/tdb_private.h b/ccan/tdb/tdb_private.h
index 25ecca95248b999351924a58b2dcc1b4bd97a91b..52383287a067605f790200562d72cf151b34cef9 100644 (file)
--- a/ccan/tdb/tdb_private.h
+++ b/ccan/tdb/tdb_private.h
@@ -223,7 +223,7 @@ struct tdb_context {
        int read_only; /* opened read-only */
        int traverse_read; /* read-only traversal */
        int traverse_write; /* read-write traversal */
        int read_only; /* opened read-only */
        int traverse_read; /* read-only traversal */
        int traverse_write; /* read-write traversal */

-       struct tdb_lock_type allrecord_lock;
+       struct tdb_lock_type allrecord_lock; /* .offset == upgradable */

        int num_lockrecs;
        struct tdb_lock_type *lockrecs; /* only real locks, all with count>0 */
        enum TDB_ERROR ecode; /* error code for last tdb error */
        int num_lockrecs;
        struct tdb_lock_type *lockrecs; /* only real locks, all with count>0 */
        enum TDB_ERROR ecode; /* error code for last tdb error */


@@ -269,7 +269,10 @@ bool tdb_have_extra_locks(struct tdb_context *tdb);
 void tdb_release_extra_locks(struct tdb_context *tdb);
 int tdb_transaction_lock(struct tdb_context *tdb, int ltype);
 int tdb_transaction_unlock(struct tdb_context *tdb, int ltype);
 void tdb_release_extra_locks(struct tdb_context *tdb);
 int tdb_transaction_lock(struct tdb_context *tdb, int ltype);
 int tdb_transaction_unlock(struct tdb_context *tdb, int ltype);

-int tdb_brlock_upgrade(struct tdb_context *tdb, tdb_off_t offset, size_t len);
+int tdb_allrecord_lock(struct tdb_context *tdb, int ltype,
+                      enum tdb_lock_flags flags, bool upgradable);
+int tdb_allrecord_unlock(struct tdb_context *tdb, int ltype, bool mark_lock);
+int tdb_allrecord_upgrade(struct tdb_context *tdb);

 int tdb_write_lock_record(struct tdb_context *tdb, tdb_off_t off);
 int tdb_write_unlock_record(struct tdb_context *tdb, tdb_off_t off);
 int tdb_ofs_read(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d);
 int tdb_write_lock_record(struct tdb_context *tdb, tdb_off_t off);
 int tdb_write_unlock_record(struct tdb_context *tdb, tdb_off_t off);
 int tdb_ofs_read(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d);

diff --git a/ccan/tdb/transaction.c b/ccan/tdb/transaction.c
index ded3fa836fc721a07932c54a6f741b89c8329066..a6cffb35d5e7e83943c027b70070b05cf2ae1802 100644 (file)
--- a/ccan/tdb/transaction.c
+++ b/ccan/tdb/transaction.c
@@ -510,7 +510,6 @@ int _tdb_transaction_cancel(struct tdb_context *tdb, int ltype)
        /* restore the normal io methods */
        tdb->methods = tdb->transaction->io_methods;
 
        /* restore the normal io methods */
        tdb->methods = tdb->transaction->io_methods;
 

-       tdb_brunlock(tdb, ltype, FREELIST_TOP, 0);

        tdb_transaction_unlock(tdb, F_WRLCK);
        SAFE_FREE(tdb->transaction->hash_heads);
        SAFE_FREE(tdb->transaction);
        tdb_transaction_unlock(tdb, F_WRLCK);
        SAFE_FREE(tdb->transaction->hash_heads);
        SAFE_FREE(tdb->transaction);


@@ -583,10 +582,9 @@ int tdb_transaction_start(struct tdb_context *tdb)
        
        /* get a read lock from the freelist to the end of file. This
           is upgraded to a write lock during the commit */
        
        /* get a read lock from the freelist to the end of file. This
           is upgraded to a write lock during the commit */

-       if (tdb_brlock(tdb, F_RDLCK, FREELIST_TOP, 0, TDB_LOCK_WAIT) == -1) {
+       if (tdb_allrecord_lock(tdb, F_RDLCK, TDB_LOCK_WAIT, true) == -1) {

                TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: failed to get hash locks\n"));
                TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: failed to get hash locks\n"));

-               tdb->ecode = TDB_ERR_LOCK;
-               goto fail;
+               goto fail_allrecord_lock;

        }
 
        /* setup a copy of the hash table heads so the hash scan in
        }
 
        /* setup a copy of the hash table heads so the hash scan in


@@ -619,7 +617,8 @@ int tdb_transaction_start(struct tdb_context *tdb)
        return 0;
        
 fail:
        return 0;
        
 fail:

-       tdb_brunlock(tdb, F_RDLCK, FREELIST_TOP, 0);
+       tdb_allrecord_unlock(tdb, F_RDLCK, false);
+fail_allrecord_lock:

        tdb_transaction_unlock(tdb, F_WRLCK);
        SAFE_FREE(tdb->transaction->blocks);
        SAFE_FREE(tdb->transaction->hash_heads);
        tdb_transaction_unlock(tdb, F_WRLCK);
        SAFE_FREE(tdb->transaction->blocks);
        SAFE_FREE(tdb->transaction->hash_heads);


@@ -932,9 +931,8 @@ static int _tdb_transaction_prepare_commit(struct tdb_context *tdb)
        }
 
        /* upgrade the main transaction lock region to a write lock */
        }
 
        /* upgrade the main transaction lock region to a write lock */

-       if (tdb_brlock_upgrade(tdb, FREELIST_TOP, 0) == -1) {
+       if (tdb_allrecord_upgrade(tdb) == -1) {

                TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_prepare_commit: failed to upgrade hash locks\n"));
                TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_prepare_commit: failed to upgrade hash locks\n"));

-               tdb->ecode = TDB_ERR_LOCK;

                _tdb_transaction_cancel(tdb, F_RDLCK);
                return -1;
        }
                _tdb_transaction_cancel(tdb, F_RDLCK);
                return -1;
        }