X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fquota%2Fqsd_lock.c;h=47b0ac1967ef15305de98225232595db73da16a5;hb=refs%2Fchanges%2F77%2F4777%2F8;hp=679cbb7b6c580c606c86f2a9864e5a0185b31ac6;hpb=318fd8d197ff607a032dac6ed9cb15922e44a53f;p=fs%2Flustre-release.git diff --git a/lustre/quota/qsd_lock.c b/lustre/quota/qsd_lock.c index 679cbb7..47b0ac1 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, 2013, Intel Corporation. * Use is subject to license terms. * - * Author: Johann Lombardi - * Author: Niu Yawei + * 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); } @@ -126,6 +149,8 @@ static int qsd_common_glimpse_ast(struct ptlrpc_request *req, RETURN(-EFAULT); /* prepare reply */ + req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER, + sizeof(struct lquota_lvb)); rc = req_capsule_server_pack(&req->rq_pill); if (rc != 0) { CERROR("Can't pack response, rc %d\n", rc); @@ -161,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: { @@ -175,12 +200,12 @@ static int qsd_glb_blocking_ast(struct ldlm_lock *lock, /* we are losing the global index lock, so let's mark the * global & slave indexes as not up-to-date any more */ - cfs_write_lock(&qqi->qqi_qsd->qsd_lock); + write_lock(&qqi->qqi_qsd->qsd_lock); qqi->qqi_glb_uptodate = false; qqi->qqi_slv_uptodate = false; if (lock->l_handle.h_cookie == qqi->qqi_lockh.cookie) memset(&qqi->qqi_lockh, 0, sizeof(qqi->qqi_lockh)); - cfs_write_unlock(&qqi->qqi_qsd->qsd_lock); + write_unlock(&qqi->qqi_qsd->qsd_lock); CDEBUG(D_QUOTA, "%s: losing global index lock for %s type\n", qqi->qqi_qsd->qsd_svname, QTYPE_NAME((qqi->qqi_qtype))); @@ -227,10 +252,10 @@ static int qsd_glb_glimpse_ast(struct ldlm_lock *lock, void *data) /* valid race */ GOTO(out, rc = -ELDLM_NO_LOCK_DATA); - LCONSOLE_INFO("%s: glimpse on glb quota locks, id:"LPU64" ver:"LPU64 - " hard:" LPU64" soft:"LPU64"\n", qqi->qqi_qsd->qsd_svname, - desc->gl_id.qid_uid, desc->gl_ver, desc->gl_hardlimit, - desc->gl_softlimit); + CDEBUG(D_QUOTA, "%s: glimpse on glb quota locks, id:"LPU64" ver:"LPU64 + " hard:" LPU64" soft:"LPU64"\n", qqi->qqi_qsd->qsd_svname, + desc->gl_id.qid_uid, desc->gl_ver, desc->gl_hardlimit, + desc->gl_softlimit); if (desc->gl_ver == 0) { CERROR("%s: invalid global index version "LPU64"\n", @@ -241,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 @@ -257,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. @@ -285,7 +304,7 @@ 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: { @@ -339,7 +358,7 @@ static int qsd_id_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *de * which means there could be a short window that slave is * holding spare grant wihtout per-ID lock. */ if (rel) - rc = qsd_dqacq(env, lqe, QSD_REL); + rc = qsd_adjust(env, lqe); /* release lqe reference grabbed by qsd_id_ast_data_get() */ lqe_putref(lqe); @@ -380,8 +399,8 @@ static int qsd_id_glimpse_ast(struct ldlm_lock *lock, void *data) /* valid race */ GOTO(out, rc = -ELDLM_NO_LOCK_DATA); - LQUOTA_CONSOLE(lqe, "glimpse on quota locks, new qunit:"LPU64, - desc->gl_qunit); + LQUOTA_DEBUG(lqe, "glimpse on quota locks, new qunit:"LPU64, + desc->gl_qunit); qsd = lqe2qqi(lqe)->qqi_qsd; @@ -400,14 +419,14 @@ static int qsd_id_glimpse_ast(struct ldlm_lock *lock, void *data) if (space > 0) { if (lqe->lqe_pending_req > 0) { - LQUOTA_ERROR(lqe, "request in flight, postpone " + LQUOTA_DEBUG(lqe, "request in flight, postpone " "release of "LPD64, space); lvb->lvb_id_may_rel = space; } else { lqe->lqe_pending_req++; /* release quota space in glimpse reply */ - LQUOTA_ERROR(lqe, "releasing "LPD64, space); + LQUOTA_DEBUG(lqe, "releasing "LPD64, space); lqe->lqe_granted -= space; lvb->lvb_id_rel = space; @@ -435,14 +454,7 @@ out: 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. @@ -470,6 +482,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