locking: Implement active lock requests limit per database

This limit was currently a global limit and not per database.  This
prevents any database freeze lock requests from getting scheduled if
the global limit was reached.

Only individual record requests should be limited and database freeze
requests should always get scheduled.

Signed-off-by: Amitay Isaacs <amitay@gmail.com>

diff --git a/include/ctdb_private.h b/include/ctdb_private.h
index 279fa2f44ada48c8001f43e8607a7fe39f63e8d0..8135112f62c78567523f62e999aefc3bf22dd722 100644 (file)
--- a/include/ctdb_private.h
+++ b/include/ctdb_private.h
@@ -556,7 +556,6 @@ struct ctdb_context {
        struct trbt_tree *child_processes; 
 
        /* Used for locking record/db/alldb */
-       int lock_num_current;
        int lock_num_pending;
        struct lock_context *lock_current;
        struct lock_context *lock_pending;
@@ -596,6 +595,8 @@ struct ctdb_db_context {
        struct trbt_tree *deferred_fetch;
 
        struct ctdb_db_statistics statistics;
+
+       int lock_num_current;
 };
 
 

diff --git a/server/ctdb_lock.c b/server/ctdb_lock.c
index bb66f9496320665ade63b5ee2f477b2823477a10..23be5f52e4cbb93b0197058f4f446b6210bda689 100644 (file)
--- a/server/ctdb_lock.c
+++ b/server/ctdb_lock.c
@@ -279,7 +279,9 @@ static int ctdb_lock_context_destructor(struct lock_context *lock_ctx)
        if (lock_ctx->child > 0) {
                ctdb_kill(lock_ctx->ctdb, lock_ctx->child, SIGKILL);
                DLIST_REMOVE(lock_ctx->ctdb->lock_current, lock_ctx);
-               lock_ctx->ctdb->lock_num_current--;
+               if (lock_ctx->ctdb_db) {
+                       lock_ctx->ctdb_db->lock_num_current--;
+               }
                CTDB_DECREMENT_STAT(lock_ctx->ctdb, locks.num_current);
                if (lock_ctx->type == LOCK_RECORD || lock_ctx->type == LOCK_DB) {
                        CTDB_DECREMENT_DB_STAT(lock_ctx->ctdb_db, locks.num_current);
@@ -741,10 +743,6 @@ static void ctdb_lock_schedule(struct ctdb_context *ctdb)
                CTDB_NO_MEMORY_VOID(ctdb, prog);
        }
 
-       if (ctdb->lock_num_current >= MAX_LOCK_PROCESSES_PER_DB) {
-               return;
-       }
-
        if (ctdb->lock_pending == NULL) {
                return;
        }
@@ -767,8 +765,11 @@ static void ctdb_lock_schedule(struct ctdb_context *ctdb)
                                                       lock_ctx->key, lock_ctx->priority,
                                                       lock_ctx->type);
                        if (active_ctx == NULL) {
-                               /* Found a lock context with lock requests */
-                               break;
+                               if (lock_ctx->ctdb_db == NULL ||
+                                   lock_ctx->ctdb_db->lock_num_current < MAX_LOCK_PROCESSES_PER_DB) {
+                                       /* Found a lock context with lock requests */
+                                       break;
+                               }
                        }
 
                        /* There is already a child waiting for the
@@ -874,7 +875,9 @@ static void ctdb_lock_schedule(struct ctdb_context *ctdb)
        DLIST_REMOVE(ctdb->lock_pending, lock_ctx);
        ctdb->lock_num_pending--;
        DLIST_ADD_END(ctdb->lock_current, lock_ctx, NULL);
-       ctdb->lock_num_current++;
+       if (lock_ctx->ctdb_db) {
+               lock_ctx->ctdb_db->lock_num_current++;
+       }
 }