LU-15472 ldlm: optimize flock reprocess

Resource reprocess on flock unlock can be done once
after all pending unlock requests.
It allows to reduce spinlock contention.

Lustre-change: https://review.whamcloud.com/46257
Lustre-commit: 42f377db4a24cefa7a041fcd3106dd58771eb319

Change-Id: I2809070f27fe3af7e1fc34e2b4b22603931f3dff
HPE-bug-id: LUS-10471, LUS-10909
Signed-off-by: Andriy Skulysh <c17819@cray.com>
Reviewed-on: https://review.whamcloud.com/c/ex/lustre-release/+/48818
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>

diff --git a/lustre/include/lustre_dlm.h b/lustre/include/lustre_dlm.h
index 4bd273e..3fee50f 100644 (file)
--- a/lustre/include/lustre_dlm.h
+++ b/lustre/include/lustre_dlm.h
@@ -715,6 +715,11 @@ struct ldlm_ibits_node {
        struct ldlm_lock        *lock;
 };
 
+struct ldlm_flock_node {
+       atomic_t                lfn_unlock_pending;
+       bool                    lfn_needs_reprocess;
+};
+
 /** Whether to track references to exports by LDLM locks. */
 #define LUSTRE_TRACKS_LOCK_EXP_REFS (0)
 
@@ -1092,6 +1097,7 @@ struct ldlm_resource {
                 */
                struct ldlm_interval_tree *lr_itree;
                struct ldlm_ibits_queues *lr_ibits_queues;
+               struct ldlm_flock_node lr_flock_node;
        };
 
        union {

diff --git a/lustre/ldlm/ldlm_flock.c b/lustre/ldlm/ldlm_flock.c
index aefab40..fe77edb 100644 (file)
--- a/lustre/ldlm/ldlm_flock.c
+++ b/lustre/ldlm/ldlm_flock.c
@@ -296,7 +296,6 @@ ldlm_process_flock_lock(struct ldlm_lock *req, __u64 *flags,
        enum ldlm_mode mode = req->l_req_mode;
        int local = ns_is_client(ns);
        int added = (mode == LCK_NL);
-       int overlaps = 0;
        int splitted = 0;
        const struct ldlm_callback_suite null_cbs = { NULL };
 #ifdef HAVE_SERVER_SUPPORT
@@ -498,7 +497,7 @@ reprocess:
                    lock->l_policy_data.l_flock.start)
                        break;
 
-               ++overlaps;
+               res->lr_flock_node.lfn_needs_reprocess = true;
 
                if (new->l_policy_data.l_flock.start <=
                    lock->l_policy_data.l_flock.start) {
@@ -607,21 +606,25 @@ reprocess:
                         * but only once because 'intention' won't be
                         * LDLM_PROCESS_ENQUEUE from ldlm_reprocess_queue.
                         */
-                       if ((mode == LCK_NL) && overlaps) {
+                       struct ldlm_flock_node *fn = &res->lr_flock_node;
+restart:
+                       if (mode == LCK_NL && fn->lfn_needs_reprocess &&
+                           atomic_read(&fn->lfn_unlock_pending) == 0) {
                                LIST_HEAD(rpc_list);
                                int rc;
 
-restart:
                                ldlm_reprocess_queue(res, &res->lr_waiting,
                                                     &rpc_list,
                                                     LDLM_PROCESS_RESCAN, NULL);
-
+                               fn->lfn_needs_reprocess = false;
                                unlock_res_and_lock(req);
                                rc = ldlm_run_ast_work(ns, &rpc_list,
                                                       LDLM_WORK_CP_AST);
                                lock_res_and_lock(req);
-                               if (rc == -ERESTART)
+                               if (rc == -ERESTART) {
+                                       fn->lfn_needs_reprocess = true;
                                        GOTO(restart, rc);
+                               }
                        }
                } else {
                        LASSERT(req->l_completion_ast);

diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c
index dcb6448..06626ae 100644 (file)
--- a/lustre/ldlm/ldlm_lock.c
+++ b/lustre/ldlm/ldlm_lock.c
@@ -1840,8 +1840,27 @@ enum ldlm_error ldlm_lock_enqueue(const struct lu_env *env,
                        RETURN(rc);
                }
        }
+
+       if (!local && lock->l_resource->lr_type == LDLM_FLOCK) {
+               struct ldlm_flock_node *fn = &lock->l_resource->lr_flock_node;
+               if (lock->l_req_mode == LCK_NL) {
+                       atomic_inc(&fn->lfn_unlock_pending);
+                       res = lock_res_and_lock(lock);
+                       atomic_dec(&fn->lfn_unlock_pending);
+               } else {
+                       res  = lock_res_and_lock(lock);
+
+                       while (atomic_read(&fn->lfn_unlock_pending)) {
+                               unlock_res_and_lock(lock);
+                               cond_resched();
+                               lock_res_and_lock(lock);
+                       }
+               }
+       } else
 #endif
-       res = lock_res_and_lock(lock);
+       {
+               res = lock_res_and_lock(lock);
+       }
        if (local && ldlm_is_granted(lock)) {
                 /* The server returned a blocked lock, but it was granted
                  * before we got a chance to actually enqueue it.  We don't

diff --git a/lustre/ldlm/ldlm_resource.c b/lustre/ldlm/ldlm_resource.c
index 60271df..0be4b1d 100644 (file)
--- a/lustre/ldlm/ldlm_resource.c
+++ b/lustre/ldlm/ldlm_resource.c
@@ -1409,6 +1409,14 @@ static bool ldlm_resource_inodebits_new(struct ldlm_resource *res)
        return true;
 }
 
+static bool ldlm_resource_flock_new(struct ldlm_resource *res)
+{
+       res->lr_flock_node.lfn_needs_reprocess = false;
+       atomic_set(&res->lr_flock_node.lfn_unlock_pending, 0);
+
+       return true;
+}
+
 /** Create and initialize new resource. */
 static struct ldlm_resource *ldlm_resource_new(enum ldlm_type ldlm_type)
 {
@@ -1426,6 +1434,9 @@ static struct ldlm_resource *ldlm_resource_new(enum ldlm_type ldlm_type)
        case LDLM_IBITS:
                rc = ldlm_resource_inodebits_new(res);
                break;
+       case LDLM_FLOCK:
+               rc = ldlm_resource_flock_new(res);
+               break;
        default:
                rc = true;
                break;