Readrecordlock changes:

Make the use of ctdb_release_lock() mandatory from the callback.

Split ctdb_release_lock() in two, release the tdb lock in the
ctdb_release_lock() function and move the freeing of the lock structure to ctdb_free_lock() which is private to libctdb.

When the callback returns, verify that the callback has actually released the lock and warn (FIXME) if not.

Update ctdb_writerecord to warn and fail (FIXME) if writing while the lock is not held.

diff --git a/libctdb/ctdb.c b/libctdb/ctdb.c
index 76e8d50a28168f9ed978468fe45dbc4e2f58a2aa..52cd38e4a192b6b173554ee3e23db395c2c9c009 100644 (file)
--- a/libctdb/ctdb.c
+++ b/libctdb/ctdb.c
@@ -536,8 +536,18 @@ void ctdb_release_lock(struct ctdb_lock *lock)
 {
        if (lock->held) {
                tdb_chainunlock(lock->ctdb_db->tdb, lock->key);
+               lock->held = false;
        }
-       free(lock->hdr); /* Also frees data */
+}
+
+static void ctdb_free_lock(struct ctdb_lock *lock)
+{
+       if (lock->held) {
+               /* FIXME: report error. Callback never released the lock */
+               ctdb_release_lock(lock);
+       }
+
+       free(lock->hdr);
        free(lock);
 }
 
@@ -566,6 +576,7 @@ static bool try_readrecordlock(struct ctdb_lock *lock, TDB_DATA *data)
 static void destroy_lock(struct ctdb_request *req)
 {
        ctdb_release_lock(req->extra);
+       ctdb_free_lock(req->extra);
 }
 
 static void readrecordlock_retry(struct ctdb_connection *ctdb,
@@ -580,6 +591,7 @@ static void readrecordlock_retry(struct ctdb_connection *ctdb,
        if (!reply || reply->status != 0) {
                lock->callback(lock->ctdb_db, NULL, tdb_null, private);
                ctdb_request_free(req); /* Also frees lock. */
+               ctdb_free_lock(lock);
                return;
        }
 
@@ -588,6 +600,7 @@ static void readrecordlock_retry(struct ctdb_connection *ctdb,
                /* Now it's their responsibility to free lock & request! */
                req->extra_destructor = NULL;
                lock->callback(lock->ctdb_db, lock, data, private);
+               ctdb_free_lock(lock);
                return;
        }
 
@@ -619,6 +632,7 @@ ctdb_readrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
        /* Fast path. */
        if (try_readrecordlock(lock, &data)) {
                callback(ctdb_db, lock, data, cbdata);
+               ctdb_free_lock(lock);
                return true;
        }
 
@@ -627,6 +641,7 @@ ctdb_readrecordlock_async(struct ctdb_db *ctdb_db, TDB_DATA key,
                               + key.dsize, readrecordlock_retry, cbdata);
        if (!req) {
                ctdb_release_lock(lock);
+               ctdb_free_lock(lock);
                return NULL;
        }
        req->extra = lock;
@@ -655,6 +670,11 @@ int ctdb_writerecord(struct ctdb_lock *lock, TDB_DATA data)
                return -1;
        }
 
+       if (!lock->held) {
+               /* FIXME: Report error. */
+               return -1;
+       }
+
        return ctdb_local_store(lock->ctdb_db->tdb, lock->key, lock->hdr,
                                data);
 }