LU-11158 mdt: grow lvb buffer to hold layout

[fs/lustre-release.git] / lustre / ldlm / ldlm_lockd.c

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index 6edef77..5789e8c 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -49,6 +49,11 @@ static int ldlm_num_threads;
 module_param(ldlm_num_threads, int, 0444);
 MODULE_PARM_DESC(ldlm_num_threads, "number of DLM service threads to start");
 
+static unsigned int ldlm_cpu_bind = 1;
+module_param(ldlm_cpu_bind, uint, 0444);
+MODULE_PARM_DESC(ldlm_cpu_bind,
+                "bind DLM service threads to particular CPU partitions");
+
 static char *ldlm_cpts;
 module_param(ldlm_cpts, charp, 0444);
 MODULE_PARM_DESC(ldlm_cpts, "CPU partitions ldlm threads should run on");
@@ -998,7 +1003,7 @@ int ldlm_server_completion_ast(struct ldlm_lock *lock, __u64 flags, void *data)
                if (req->rq_svc_thread)
                        env = req->rq_svc_thread->t_env;
 
-               lvb_len = ldlm_lvbo_fill(env, lock, lvb, lvb_len);
+               lvb_len = ldlm_lvbo_fill(env, lock, lvb, &lvb_len);
                if (lvb_len < 0) {
                        /* We still need to send the RPC to wake up the blocked
                         * enqueue thread on the client.
@@ -1404,43 +1409,59 @@ existing_lock:
                LDLM_DEBUG(lock, "server-side enqueue handler, sending reply"
                           "(err=%d, rc=%d)", err, rc);
 
-               if (rc == 0) {
-                       if (req_capsule_has_field(&req->rq_pill, &RMF_DLM_LVB,
-                                                 RCL_SERVER) &&
-                           ldlm_lvbo_size(lock) > 0) {
-                               void *buf;
-                               int buflen;
-
-                               buf = req_capsule_server_get(&req->rq_pill,
-                                                            &RMF_DLM_LVB);
-                               LASSERTF(buf != NULL, "req %p, lock %p\n",
-                                        req, lock);
-                               buflen = req_capsule_get_size(&req->rq_pill,
-                                               &RMF_DLM_LVB, RCL_SERVER);
-                               /* non-replayed lock, delayed lvb init may
-                                * need to be occur now */
-                               if ((buflen > 0) && !(flags & LDLM_FL_REPLAY)) {
-                                       buflen = ldlm_lvbo_fill(env, lock, buf,
-                                                               buflen);
-                                       if (buflen >= 0)
-                                               req_capsule_shrink(
+               if (rc == 0 &&
+                   req_capsule_has_field(&req->rq_pill, &RMF_DLM_LVB,
+                                         RCL_SERVER) &&
+                   ldlm_lvbo_size(lock) > 0) {
+                       void *buf;
+                       int buflen;
+
+retry:
+                       buf = req_capsule_server_get(&req->rq_pill,
+                                                    &RMF_DLM_LVB);
+                       LASSERTF(buf != NULL, "req %p, lock %p\n", req, lock);
+                       buflen = req_capsule_get_size(&req->rq_pill,
+                                       &RMF_DLM_LVB, RCL_SERVER);
+                       /* non-replayed lock, delayed lvb init may
+                        * need to be occur now
+                        */
+                       if ((buflen > 0) && !(flags & LDLM_FL_REPLAY)) {
+                               int rc2;
+
+                               rc2 = ldlm_lvbo_fill(env, lock, buf, &buflen);
+                               if (rc2 >= 0) {
+                                       req_capsule_shrink(&req->rq_pill,
+                                                          &RMF_DLM_LVB,
+                                                          rc2, RCL_SERVER);
+                               } else if (rc2 == -ERANGE) {
+                                       rc2 = req_capsule_server_grow(
                                                        &req->rq_pill,
-                                                       &RMF_DLM_LVB,
-                                                       buflen, RCL_SERVER);
-                                       else
-                                               rc = buflen;
-                               } else if (flags & LDLM_FL_REPLAY) {
-                                       /* no LVB resend upon replay */
-                                       if (buflen > 0)
+                                                       &RMF_DLM_LVB, buflen);
+                                       if (!rc2) {
+                                               goto retry;
+                                       } else {
+                                               /* if we can't grow the buffer,
+                                                * it's ok to return empty lvb
+                                                * to client.
+                                                */
                                                req_capsule_shrink(
                                                        &req->rq_pill,
-                                                       &RMF_DLM_LVB,
-                                                       0, RCL_SERVER);
-                                       else
-                                               rc = buflen;
+                                                       &RMF_DLM_LVB, 0,
+                                                       RCL_SERVER);
+                                       }
                                } else {
-                                       rc = buflen;
+                                       rc = rc2;
                                }
+                       } else if (flags & LDLM_FL_REPLAY) {
+                               /* no LVB resend upon replay */
+                               if (buflen > 0)
+                                       req_capsule_shrink(&req->rq_pill,
+                                                          &RMF_DLM_LVB,
+                                                          0, RCL_SERVER);
+                               else
+                                       rc = buflen;
+                       } else {
+                               rc = buflen;
                        }
                }
 
@@ -1785,6 +1806,21 @@ static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
        }
 
        lock_res_and_lock(lock);
+
+       if (!ldlm_res_eq(&dlm_req->lock_desc.l_resource.lr_name,
+                        &lock->l_resource->lr_name)) {
+               ldlm_resource_unlink_lock(lock);
+               unlock_res_and_lock(lock);
+               rc = ldlm_lock_change_resource(ns, lock,
+                               &dlm_req->lock_desc.l_resource.lr_name);
+               if (rc < 0) {
+                       LDLM_ERROR(lock, "Failed to allocate resource");
+                       GOTO(out, rc);
+               }
+               LDLM_DEBUG(lock, "completion AST, new resource");
+               lock_res_and_lock(lock);
+       }
+
        if (ldlm_is_destroyed(lock) ||
            lock->l_granted_mode == lock->l_req_mode) {
                /* bug 11300: the lock has already been granted */
@@ -1808,21 +1844,7 @@ static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
                LDLM_DEBUG(lock, "completion AST, new policy data");
        }
 
-        ldlm_resource_unlink_lock(lock);
-        if (memcmp(&dlm_req->lock_desc.l_resource.lr_name,
-                   &lock->l_resource->lr_name,
-                   sizeof(lock->l_resource->lr_name)) != 0) {
-                unlock_res_and_lock(lock);
-               rc = ldlm_lock_change_resource(ns, lock,
-                               &dlm_req->lock_desc.l_resource.lr_name);
-               if (rc < 0) {
-                       LDLM_ERROR(lock, "Failed to allocate resource");
-                       GOTO(out, rc);
-               }
-                LDLM_DEBUG(lock, "completion AST, new resource");
-                CERROR("change resource!\n");
-                lock_res_and_lock(lock);
-        }
+       ldlm_resource_unlink_lock(lock);
 
         if (dlm_req->lock_flags & LDLM_FL_AST_SENT) {
                /* BL_AST locks are not needed in LRU.
@@ -2954,11 +2976,12 @@ static int ldlm_setup(void)
                        .tc_nthrs_base          = LDLM_NTHRS_BASE,
                        .tc_nthrs_max           = LDLM_NTHRS_MAX,
                        .tc_nthrs_user          = ldlm_num_threads,
-                       .tc_cpu_affinity        = 1,
+                       .tc_cpu_bind            = ldlm_cpu_bind,
                        .tc_ctx_tags            = LCT_MD_THREAD | LCT_DT_THREAD,
                },
                .psc_cpt                = {
                        .cc_pattern             = ldlm_cpts,
+                       .cc_affinity            = true,
                },
                .psc_ops                = {
                        .so_req_handler         = ldlm_callback_handler,
@@ -2995,13 +3018,14 @@ static int ldlm_setup(void)
                        .tc_nthrs_base          = LDLM_NTHRS_BASE,
                        .tc_nthrs_max           = LDLM_NTHRS_MAX,
                        .tc_nthrs_user          = ldlm_num_threads,
-                       .tc_cpu_affinity        = 1,
+                       .tc_cpu_bind            = ldlm_cpu_bind,
                        .tc_ctx_tags            = LCT_MD_THREAD | \
                                                  LCT_DT_THREAD | \
                                                  LCT_CL_THREAD,
                },
                .psc_cpt                = {
                        .cc_pattern             = ldlm_cpts,
+                       .cc_affinity            = true,
                },
                .psc_ops                = {
                        .so_req_handler         = ldlm_cancel_handler,