X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fquota%2Fqsd_lock.c;h=29c70cd9f9fd49313ef97d8aacd3894133f9338c;hb=473bc6473eb8d6d78069a561dbe035e795d8b7e2;hp=68c835288f59b76c63f2b132069992c463d9c91c;hpb=929ec628e6fef5609e55d519a1eb9e2cbbf1f1e8;p=fs%2Flustre-release.git diff --git a/lustre/quota/qsd_lock.c b/lustre/quota/qsd_lock.c index 68c8352..29c70cd 100644 --- a/lustre/quota/qsd_lock.c +++ b/lustre/quota/qsd_lock.c @@ -21,17 +21,13 @@ * GPL HEADER END */ /* - * Copyright (c) 2011, 2012, Intel, Inc. + * Copyright (c) 2012, 2014, Intel Corporation. * Use is subject to license terms. * * Author: Johann Lombardi * Author: Niu Yawei */ -#ifndef EXPORT_SYMTAB -# define EXPORT_SYMTAB -#endif - #define DEBUG_SUBSYSTEM S_LQUOTA #include @@ -39,6 +35,30 @@ #include "qsd_internal.h" +typedef int (enqi_bl_cb_t)(struct ldlm_lock *lock, + struct ldlm_lock_desc *desc, void *data, + int flag); +static enqi_bl_cb_t qsd_glb_blocking_ast, qsd_id_blocking_ast; + +typedef int (enqi_gl_cb_t)(struct ldlm_lock *lock, void *data); +static enqi_gl_cb_t qsd_glb_glimpse_ast, qsd_id_glimpse_ast; + +struct ldlm_enqueue_info qsd_glb_einfo = { + .ei_type = LDLM_PLAIN, + .ei_mode = LCK_CR, + .ei_cb_bl = qsd_glb_blocking_ast, + .ei_cb_cp = ldlm_completion_ast, + .ei_cb_gl = qsd_glb_glimpse_ast, +}; + +struct ldlm_enqueue_info qsd_id_einfo = { + .ei_type = LDLM_PLAIN, + .ei_mode = LCK_CR, + .ei_cb_bl = qsd_id_blocking_ast, + .ei_cb_cp = ldlm_completion_ast, + .ei_cb_gl = qsd_id_glimpse_ast, +}; + /* * Return qsd_qtype_info structure associated with a global lock * @@ -54,16 +74,19 @@ static struct qsd_qtype_info *qsd_glb_ast_data_get(struct ldlm_lock *lock, qqi = lock->l_ast_data; if (qqi != NULL) { qqi_getref(qqi); - lu_ref_add(&qqi->qqi_reference, "ast_data_get", lock); if (reset) lock->l_ast_data = NULL; } unlock_res_and_lock(lock); + if (qqi != NULL) + /* it is not safe to call lu_ref_add() under spinlock */ + lu_ref_add(&qqi->qqi_reference, "ast_data_get", lock); + if (reset && qqi != NULL) { /* release qqi reference hold for the lock */ - qqi_putref(qqi); lu_ref_del(&qqi->qqi_reference, "glb_lock", lock); + qqi_putref(qqi); } RETURN(qqi); } @@ -163,7 +186,7 @@ static int qsd_glb_blocking_ast(struct ldlm_lock *lock, LDLM_DEBUG(lock, "blocking AST on global quota lock"); ldlm_lock2handle(lock, &lockh); - rc = ldlm_cli_cancel(&lockh); + rc = ldlm_cli_cancel(&lockh, LCF_ASYNC); break; } case LDLM_CB_CANCELING: { @@ -190,7 +213,7 @@ static int qsd_glb_blocking_ast(struct ldlm_lock *lock, /* kick off reintegration thread if not running already, if * it's just local cancel (for stack clean up or eviction), * don't re-trigger the reintegration. */ - if ((lock->l_flags & LDLM_FL_LOCAL_ONLY) == 0) + if (!ldlm_is_local_only(lock)) qsd_start_reint_thread(qqi); lu_ref_del(&qqi->qqi_reference, "ast_data_get", lock); @@ -198,7 +221,7 @@ static int qsd_glb_blocking_ast(struct ldlm_lock *lock, break; } default: - LASSERTF(0, "invalid flags for blocking ast %d", flag); + LASSERTF(0, "invalid flags for blocking ast %d\n", flag); } RETURN(rc); @@ -243,7 +266,7 @@ static int qsd_glb_glimpse_ast(struct ldlm_lock *lock, void *data) /* extract new hard & soft limits from the glimpse descriptor */ rec.qbr_hardlimit = desc->gl_hardlimit; rec.qbr_softlimit = desc->gl_softlimit; - rec.qbr_time = 0; + rec.qbr_time = desc->gl_time; rec.qbr_granted = 0; /* We can't afford disk io in the context of glimpse callback handling @@ -259,13 +282,7 @@ out: return rc; } -struct ldlm_enqueue_info qsd_glb_einfo = { LDLM_PLAIN, - LCK_CR, - qsd_glb_blocking_ast, - ldlm_completion_ast, - qsd_glb_glimpse_ast, - NULL, NULL }; -/* +/** * Blocking callback handler for per-ID lock * * \param lock - is the lock for which ast occurred. @@ -287,70 +304,66 @@ static int qsd_id_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *de LDLM_DEBUG(lock, "blocking AST on ID quota lock"); ldlm_lock2handle(lock, &lockh); - rc = ldlm_cli_cancel(&lockh); + rc = ldlm_cli_cancel(&lockh, LCF_ASYNC); break; } case LDLM_CB_CANCELING: { struct lu_env *env; struct lquota_entry *lqe; - bool rel = false; - LDLM_DEBUG(lock, "canceling global quota lock"); + LDLM_DEBUG(lock, "canceling ID quota lock"); lqe = qsd_id_ast_data_get(lock, true); if (lqe == NULL) break; LQUOTA_DEBUG(lqe, "losing ID lock"); - /* just local cancel (for stack clean up or eviction), don't - * release quota space in this case */ - if ((lock->l_flags & LDLM_FL_LOCAL_ONLY) != 0) { - lqe_putref(lqe); - break; - } - - /* allocate environment */ - OBD_ALLOC_PTR(env); - if (env == NULL) { - lqe_putref(lqe); - rc = -ENOMEM; - break; - } - - /* initialize environment */ - rc = lu_env_init(env, LCT_DT_THREAD); - if (rc) { - OBD_FREE_PTR(env); - lqe_putref(lqe); - break; - } - ldlm_lock2handle(lock, &lockh); lqe_write_lock(lqe); if (lustre_handle_equal(&lockh, &lqe->lqe_lockh)) { /* Clear lqe_lockh & reset qunit to 0 */ qsd_set_qunit(lqe, 0); memset(&lqe->lqe_lockh, 0, sizeof(lqe->lqe_lockh)); - lqe->lqe_edquot = false; - rel = true; + qsd_set_edquot(lqe, false); } lqe_write_unlock(lqe); - /* If there is qqacq inflight, the release will be skipped + /* If there is dqacq inflight, the release will be skipped * at this time, and triggered on dqacq completion later, * which means there could be a short window that slave is * holding spare grant wihtout per-ID lock. */ - if (rel) + + /* don't release quota space for local cancel (stack clean + * up or eviction) */ + if (!ldlm_is_local_only(lock)) { + /* allocate environment */ + OBD_ALLOC_PTR(env); + if (env == NULL) { + lqe_putref(lqe); + rc = -ENOMEM; + break; + } + + /* initialize environment */ + rc = lu_env_init(env, LCT_DT_THREAD); + if (rc) { + OBD_FREE_PTR(env); + lqe_putref(lqe); + break; + } + rc = qsd_adjust(env, lqe); + lu_env_fini(env); + OBD_FREE_PTR(env); + } + /* release lqe reference grabbed by qsd_id_ast_data_get() */ lqe_putref(lqe); - lu_env_fini(env); - OBD_FREE_PTR(env); break; } default: - LASSERTF(0, "invalid flags for blocking ast %d", flag); + LASSERTF(0, "invalid flags for blocking ast %d\n", flag); } RETURN(rc); @@ -426,25 +439,18 @@ static int qsd_id_glimpse_ast(struct ldlm_lock *lock, void *data) } } - lqe->lqe_edquot = !!(desc->gl_flags & LQUOTA_FL_EDQUOT); + qsd_set_edquot(lqe, !!(desc->gl_flags & LQUOTA_FL_EDQUOT)); lqe_write_unlock(lqe); if (wakeup) - cfs_waitq_broadcast(&lqe->lqe_waiters); + wake_up_all(&lqe->lqe_waiters); lqe_putref(lqe); out: req->rq_status = rc; RETURN(rc); } -struct ldlm_enqueue_info qsd_id_einfo = { LDLM_PLAIN, - LCK_CR, - qsd_id_blocking_ast, - ldlm_completion_ast, - qsd_id_glimpse_ast, - NULL, NULL }; - -/* +/** * Check whether a slave already own a ldlm lock for the quota identifier \qid. * * \param lockh - is the local lock handle from lquota entry. @@ -472,6 +478,7 @@ int qsd_id_lock_match(struct lustre_handle *lockh, struct lustre_handle *rlockh) ldlm_lock_dump_handle(D_QUOTA, lockh); if (rlockh == NULL) + /* caller not interested in remote handle */ RETURN(0); /* look up lock associated with local handle and extract remote handle @@ -501,7 +508,7 @@ int qsd_id_lock_cancel(const struct lu_env *env, struct lquota_entry *lqe) if (lustre_handle_is_used(&qti->qti_lockh)) { memset(&lqe->lqe_lockh, 0, sizeof(lqe->lqe_lockh)); qsd_set_qunit(lqe, 0); - lqe->lqe_edquot = false; + qsd_set_edquot(lqe, false); } lqe_write_unlock(lqe);