LU-10894 dom: per-resource ELC for WRITE lock enqueue

Improve client write lock enqueue by doing ELC for any
read lock on the same resource. This helps with read/write
access, e.g. compilebench shows ~10% better results with
about 45% less ldlm cancel RPCs.

In mdc_enqueue_send() collect resource unused read locks
and pack them into enqueue request.

The ldlm_cancel_resource_local() is changed also to don't
skip DOM lock if it is set in policy explicitly

Signed-off-by: Mikhail Pershin <mpershin@whamcloud.com>
Change-Id: I06ece95d837495e2e970ce670db61ba0aa4e1ab4
Reviewed-on: https://review.whamcloud.com/34736
Tested-by: Jenkins
Reviewed-by: Patrick Farrell <pfarrell@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Alexey Lyashkov <c17817@cray.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/ldlm/ldlm_request.c b/lustre/ldlm/ldlm_request.c
index ac63e69..4d56e97 100644 (file)
--- a/lustre/ldlm/ldlm_request.c
+++ b/lustre/ldlm/ldlm_request.c
@@ -2192,12 +2192,19 @@ int ldlm_cancel_resource_local(struct ldlm_resource *res,
                /*
                 * If policy is given and this is IBITS lock, add to list only
                 * those locks that match by policy.
-                * Skip locks with DoM bit always to don't flush data.
                 */
-               if (policy && (lock->l_resource->lr_type == LDLM_IBITS) &&
-                   (!(lock->l_policy_data.l_inodebits.bits &
-                     policy->l_inodebits.bits) || ldlm_has_dom(lock)))
-                       continue;
+               if (policy && (lock->l_resource->lr_type == LDLM_IBITS)) {
+                       if (!(lock->l_policy_data.l_inodebits.bits &
+                             policy->l_inodebits.bits))
+                               continue;
+                       /* Skip locks with DoM bit if it is not set in policy
+                        * to don't flush data by side-bits. Lock convert will
+                        * drop those bits separately.
+                        */
+                       if (ldlm_has_dom(lock) &&
+                           !(policy->l_inodebits.bits & MDS_INODELOCK_DOM))
+                               continue;
+               }
 
                /* See CBPENDING comment in ldlm_cancel_lru */
                lock->l_flags |= LDLM_FL_CBPENDING | LDLM_FL_CANCELING |

diff --git a/lustre/mdc/mdc_dev.c b/lustre/mdc/mdc_dev.c
index 1adc457..a59b1a8 100644 (file)
--- a/lustre/mdc/mdc_dev.c
+++ b/lustre/mdc/mdc_dev.c
@@ -686,7 +686,8 @@ int mdc_enqueue_send(const struct lu_env *env, struct obd_export *exp,
        enum ldlm_mode mode;
        bool glimpse = *flags & LDLM_FL_HAS_INTENT;
        __u64 match_flags = *flags;
-       int rc;
+       struct list_head cancels = LIST_HEAD_INIT(cancels);
+       int rc, count;
 
        ENTRY;
 
@@ -744,7 +745,15 @@ no_match:
        if (req == NULL)
                RETURN(-ENOMEM);
 
-       rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
+       /* For WRITE lock cancel other locks on resource early if any */
+       if (einfo->ei_mode & LCK_PW)
+               count = mdc_resource_get_unused_res(exp, res_id, &cancels,
+                                                   einfo->ei_mode,
+                                                   MDS_INODELOCK_DOM);
+       else
+               count = 0;
+
+       rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
        if (rc < 0) {
                ptlrpc_request_free(req);
                RETURN(rc);

diff --git a/lustre/mdc/mdc_internal.h b/lustre/mdc/mdc_internal.h
index 81594bc..0de0ebd 100644 (file)
--- a/lustre/mdc/mdc_internal.h
+++ b/lustre/mdc/mdc_internal.h
@@ -86,7 +86,10 @@ int mdc_enqueue(struct obd_export *exp, struct ldlm_enqueue_info *einfo,
                const union ldlm_policy_data *policy,
                struct md_op_data *op_data,
                struct lustre_handle *lockh, __u64 extra_lock_flags);
-
+int mdc_resource_get_unused_res(struct obd_export *exp,
+                               struct ldlm_res_id *res_id,
+                               struct list_head *cancels,
+                               enum ldlm_mode mode, __u64 bits);
 int mdc_resource_get_unused(struct obd_export *exp, const struct lu_fid *fid,
                            struct list_head *cancels, enum ldlm_mode mode,
                             __u64 bits);

diff --git a/lustre/mdc/mdc_reint.c b/lustre/mdc/mdc_reint.c
index b8331e6..07c3e05 100644 (file)
--- a/lustre/mdc/mdc_reint.c
+++ b/lustre/mdc/mdc_reint.c
@@ -60,15 +60,16 @@ static int mdc_reint(struct ptlrpc_request *request, int level)
 /* Find and cancel locally locks matched by inode @bits & @mode in the resource
  * found by @fid. Found locks are added into @cancel list. Returns the amount of
  * locks added to @cancels list. */
-int mdc_resource_get_unused(struct obd_export *exp, const struct lu_fid *fid,
-                           struct list_head *cancels, enum ldlm_mode mode,
-                           __u64 bits)
+int mdc_resource_get_unused_res(struct obd_export *exp,
+                               struct ldlm_res_id *res_id,
+                               struct list_head *cancels,
+                               enum ldlm_mode mode, __u64 bits)
 {
        struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
-       union ldlm_policy_data policy = { {0} };
-       struct ldlm_res_id res_id;
+       union ldlm_policy_data policy = { { 0 } };
        struct ldlm_resource *res;
        int count;
+
        ENTRY;
 
        /* Return, i.e. cancel nothing, only if ELC is supported (flag in
@@ -80,21 +81,29 @@ int mdc_resource_get_unused(struct obd_export *exp, const struct lu_fid *fid,
        if (exp_connect_cancelset(exp) && !ns_connect_cancelset(ns))
                RETURN(0);
 
-       fid_build_reg_res_name(fid, &res_id);
-       res = ldlm_resource_get(exp->exp_obd->obd_namespace,
-                               NULL, &res_id, 0, 0);
+       res = ldlm_resource_get(ns, NULL, res_id, 0, 0);
        if (IS_ERR(res))
                RETURN(0);
        LDLM_RESOURCE_ADDREF(res);
        /* Initialize ibits lock policy. */
        policy.l_inodebits.bits = bits;
-       count = ldlm_cancel_resource_local(res, cancels, &policy,
-                                          mode, 0, 0, NULL);
+       count = ldlm_cancel_resource_local(res, cancels, &policy, mode, 0, 0,
+                                          NULL);
        LDLM_RESOURCE_DELREF(res);
        ldlm_resource_putref(res);
        RETURN(count);
 }
 
+int mdc_resource_get_unused(struct obd_export *exp, const struct lu_fid *fid,
+                           struct list_head *cancels, enum ldlm_mode mode,
+                           __u64 bits)
+{
+       struct ldlm_res_id res_id;
+
+       fid_build_reg_res_name(fid, &res_id);
+       return mdc_resource_get_unused_res(exp, &res_id, cancels, mode, bits);
+}
+
 int mdc_setattr(struct obd_export *exp, struct md_op_data *op_data,
                void *ea, size_t ealen, struct ptlrpc_request **request)
 {