Viewing: qsd_lock.c
// SPDX-License-Identifier: GPL-2.0
/*
* Copyright (c) 2012, 2017, Intel Corporation.
* Use is subject to license terms.
*/
/*
* Author: Johann Lombardi <johann.lombardi@intel.com>
* Author: Niu Yawei <yawei.niu@intel.com>
*/
#define DEBUG_SUBSYSTEM S_LQUOTA
#include <lustre_dlm.h>
#include <obd_class.h>
#include <lustre_swab.h>
#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
* @lock: - is the global lock from which we should extract the qqi
* @reset: - whether lock->l_ast_data should be cleared
*/
static struct qsd_qtype_info *qsd_glb_ast_data_get(struct ldlm_lock *lock,
bool reset)
{
struct qsd_qtype_info *qqi;
ENTRY;
lock_res_and_lock(lock);
qqi = lock->l_ast_data;
if (qqi) {
qqi_getref(qqi);
if (reset)
lock->l_ast_data = NULL;
}
unlock_res_and_lock(lock);
if (reset && qqi) {
qqi_putref(qqi);
}
RETURN(qqi);
}
/*
* Return lquota entry structure associated with a per-ID lock
* @lock: is the per-ID lock from which we should extract the lquota entry
* @reset: whether lock->l_ast_data should be cleared
*/
static struct lquota_entry *qsd_id_ast_data_get(struct ldlm_lock *lock,
bool reset)
{
struct lquota_entry *lqe;
ENTRY;
lock_res_and_lock(lock);
lqe = lock->l_ast_data;
if (lqe) {
lqe_getref(lqe);
if (reset)
lock->l_ast_data = NULL;
}
unlock_res_and_lock(lock);
if (reset && lqe)
/* release lqe reference hold for the lock */
lqe_putref(lqe);
RETURN(lqe);
}
/*
* Glimpse callback handler for all quota locks. This function extracts
* information from the glimpse request.
* @lock: is the lock targeted by the glimpse
* @data: is a pointer to the glimpse ptlrpc request
* @req: is the glimpse request
* @desc: glimpse descriptor describing the purpose of the glimpse request.
* @lvb: is the pointer to the lvb in the reply buffer
*
* Return 0 on success and @desc, @lvb & @arg point to a valid structures,
* appropriate error on failure
*/
static int qsd_common_glimpse_ast(struct ptlrpc_request *req,
struct ldlm_gl_lquota_desc **desc, void **lvb)
{
int rc;
ENTRY;
LASSERT(lustre_msg_get_opc(req->rq_reqmsg) == LDLM_GL_CALLBACK);
/* glimpse on quota locks always packs a glimpse descriptor */
req_capsule_extend(&req->rq_pill, &RQF_LDLM_GL_CALLBACK_DESC);
/* extract glimpse descriptor */
*desc = req_capsule_client_get(&req->rq_pill, &RMF_DLM_GL_DESC);
if (!*desc)
RETURN(-EFAULT);
if (req_capsule_req_need_swab(&req->rq_pill))
lustre_swab_gl_lquota_desc(*desc);
/* 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);
RETURN(rc);
}
/* extract lvb */
*lvb = req_capsule_server_get(&req->rq_pill, &RMF_DLM_LVB);
RETURN(0);
}
/*
* Blocking callback handler for global index lock
* @lock: is the lock for which ast occurred.
* @desc: is the description of a conflicting lock in case of blocking ast.
* @data: is the value of lock->l_ast_data
* @flag: LDLM_CB_BLOCKING or LDLM_CB_CANCELING. Used to distinguish
* cancellation and blocking ast's.
*/
static int qsd_glb_blocking_ast(struct ldlm_lock *lock,
struct ldlm_lock_desc *desc, void *data,
int flag)
{
int rc = 0;
ENTRY;
switch (flag) {
case LDLM_CB_BLOCKING: {
struct lustre_handle lockh;
LDLM_DEBUG(lock, "blocking AST on global quota lock");
ldlm_lock2handle(lock, &lockh);
rc = ldlm_cli_cancel(&lockh, LCF_ASYNC);
break;
}
case LDLM_CB_CANCELING: {
struct qsd_qtype_info *qqi;
LDLM_DEBUG(lock, "canceling global quota lock");
qqi = qsd_glb_ast_data_get(lock, true);
if (!qqi)
break;
/*
* we are losing the global index lock, so let's mark the
* global & slave indexes as not up-to-date any more
*/
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));
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)));
/*
* 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))
qsd_start_reint_thread(qqi);
qqi_putref(qqi);
break;
}
default:
LASSERTF(0, "invalid flags for blocking ast %d\n", flag);
}
RETURN(rc);
}
static int qsd_entry_def_iter_cb(struct cfs_hash *hs, struct cfs_hash_bd *bd,
struct hlist_node *hnode, void *data)
{
struct qsd_qtype_info *qqi = (struct qsd_qtype_info *)data;
struct lquota_entry *lqe;
lqe = hlist_entry(hnode, struct lquota_entry, lqe_hash);
LASSERT(kref_read(&lqe->lqe_ref) > 0);
if (lqe->lqe_id.qid_uid == 0 || !lqe->lqe_is_default)
return 0;
lqe_write_lock(lqe);
if (qqi->qqi_default_hardlimit == 0 && qqi->qqi_default_softlimit == 0)
lqe->lqe_enforced = false;
else
lqe->lqe_enforced = true;
lqe_write_unlock(lqe);
return 0;
}
/**
* qsd_update_default_quota() - Update the quota entries after receiving default
* quota update
* @qqi: is the qsd_qtype_info associated with the quota entries
* @hardlimit: new hardlimit of default quota
* @softlimit: new softlimit of default quota
* @gracetime: new gracetime of default quota
*/
void qsd_update_default_quota(struct qsd_qtype_info *qqi, __u64 hardlimit,
__u64 softlimit, __u64 gracetime)
{
CDEBUG(D_QUOTA, "%s: update default quota setting from QMT.\n",
qqi->qqi_qsd->qsd_svname);
qqi->qqi_default_hardlimit = hardlimit;
qqi->qqi_default_softlimit = softlimit;
qqi->qqi_default_gracetime = gracetime;
cfs_hash_for_each_safe(qqi->qqi_site->lqs_hash,
qsd_entry_def_iter_cb, qqi);
}
/*
* Glimpse callback handler for global quota lock.
* @lock: - is the lock targeted by the glimpse
* @data: - is a pointer to the glimpse ptlrpc request
*/
static int qsd_glb_glimpse_ast(struct ldlm_lock *lock, void *data)
{
struct ptlrpc_request *req = data;
struct qsd_qtype_info *qqi;
struct ldlm_gl_lquota_desc *desc;
struct lquota_lvb *lvb;
struct lquota_glb_rec rec;
int rc;
ENTRY;
rc = qsd_common_glimpse_ast(req, &desc, (void **)&lvb);
if (rc)
GOTO(out, rc);
qqi = qsd_glb_ast_data_get(lock, false);
if (!qqi)
/* valid race */
GOTO(out, rc = -ELDLM_NO_LOCK_DATA);
CDEBUG(D_QUOTA,
"%s: glimpse on glb quota locks, id:%llu ver:%llu hard:%llu soft:%llu\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 %llu\n",
qqi->qqi_qsd->qsd_svname, desc->gl_ver);
GOTO(out_qqi, rc = -EINVAL);
}
/* extract new hard & soft limits from the glimpse descriptor */
rec.qbr_hardlimit = desc->gl_hardlimit;
rec.qbr_softlimit = desc->gl_softlimit;
rec.qbr_time = desc->gl_time;
rec.qbr_granted = 0;
if (desc->gl_id.qid_uid == 0)
qsd_update_default_quota(qqi, desc->gl_hardlimit,
desc->gl_softlimit, desc->gl_time);
/*
* We can't afford disk io in the context of glimpse callback handling
* thread, so the on-disk global limits update has to be deferred.
*/
if (!CFS_FAIL_CHECK(OBD_FAIL_QUOTA_DROP_VER_UPDATE))
qsd_upd_schedule(qqi, NULL, &desc->gl_id,
(union lquota_rec *)&rec, desc->gl_ver, true);
EXIT;
out_qqi:
qqi_putref(qqi);
out:
req->rq_status = rc;
return rc;
}
/**
* qsd_id_blocking_ast() - Blocking callback handler for per-ID lock
* @lock: is the lock for which ast occurred.
* @desc: is the description of a conflicting lock in case of blocking ast.
* @data: is the value of lock->l_ast_data
* @flag: Used to distinguish cancellation and blocking ast's.
* (LDLM_CB_BLOCKING or LDLM_CB_CANCELING.)
*
* Return %0 on success, appropriate errors on failure
*/
static int qsd_id_blocking_ast(struct ldlm_lock *lock,
struct ldlm_lock_desc *desc,
void *data, int flag)
{
struct lustre_handle lockh;
int rc = 0;
ENTRY;
switch (flag) {
case LDLM_CB_BLOCKING: {
LDLM_DEBUG(lock, "blocking AST on ID quota lock");
ldlm_lock2handle(lock, &lockh);
rc = ldlm_cli_cancel(&lockh, LCF_ASYNC);
break;
}
case LDLM_CB_CANCELING: {
struct lu_env *env;
struct lquota_entry *lqe;
LDLM_DEBUG(lock, "canceling ID quota lock");
lqe = qsd_id_ast_data_get(lock, true);
if (!lqe)
break;
LQUOTA_DEBUG(lqe, "losing ID lock");
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));
qsd_set_edquot(lqe, false);
}
lqe_write_unlock(lqe);
/*
* 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.
*/
/*
* don't release quota space for local cancel (stack clean
* up or eviction)
*/
if (!(lock->l_flags & LDLM_FL_LOCAL_ONLY)) {
env = lu_env_find();
if (env)
rc = qsd_adjust(env, lqe);
else
rc = -ENOMEM;
}
/* release lqe reference grabbed by qsd_id_ast_data_get() */
lqe_putref(lqe);
break;
}
default:
LASSERTF(0, "invalid flags for blocking ast %d\n", flag);
}
RETURN(rc);
}
/*
* Glimpse callback handler for per-ID quota locks.
* @lock: - is the lock targeted by the glimpse
* @data: - is a pointer to the glimpse ptlrpc request
*/
static int qsd_id_glimpse_ast(struct ldlm_lock *lock, void *data)
{
struct ptlrpc_request *req = data;
struct lquota_entry *lqe;
struct ldlm_gl_lquota_desc *desc;
struct lquota_lvb *lvb;
int rc;
bool wakeup = false;
ENTRY;
rc = qsd_common_glimpse_ast(req, &desc, (void **)&lvb);
if (rc)
GOTO(out, rc);
lqe = qsd_id_ast_data_get(lock, false);
if (!lqe)
/* valid race */
GOTO(out, rc = -ELDLM_NO_LOCK_DATA);
LQUOTA_DEBUG(lqe, "glimpse on quota locks, new qunit:%llu, edquot:%d",
desc->gl_qunit, !!(desc->gl_flags & LQUOTA_FL_EDQUOT));
lqe_write_lock(lqe);
lvb->lvb_id_rel = 0;
if (desc->gl_qunit != 0 && desc->gl_qunit != lqe->lqe_qunit) {
struct lu_env *env;
long long space;
/* extract new qunit from glimpse request */
qsd_set_qunit(lqe, desc->gl_qunit);
lqe_write_unlock(lqe);
env = lu_env_find();
if (env)
qsd_refresh_usage(env, lqe);
lqe_write_lock(lqe);
lqe2qqi(lqe)->qqi_qsd->qsd_glimpse_refresh++;
space = lqe->lqe_granted - lqe->lqe_pending_rel;
space -= lqe->lqe_usage;
space -= lqe->lqe_pending_write + lqe->lqe_waiting_write;
space -= lqe->lqe_qunit;
if (space > 0) {
if (lqe->lqe_pending_req > 0) {
LQUOTA_DEBUG(lqe,
"request in flight, postpone release of %lld",
space);
lvb->lvb_id_may_rel = space;
} else {
lqe->lqe_pending_req++;
/* release quota space in glimpse reply */
LQUOTA_DEBUG(lqe, "releasing %lld", space);
lqe->lqe_granted -= space;
lvb->lvb_id_rel = space;
lqe_write_unlock(lqe);
/* change the lqe_granted */
qsd_upd_schedule(lqe2qqi(lqe), lqe,
&lqe->lqe_id,
(union lquota_rec *)
&lqe->lqe_granted, 0, false);
lqe_write_lock(lqe);
lqe->lqe_pending_req--;
wakeup = true;
}
}
}
qsd_set_edquot(lqe, !!(desc->gl_flags & LQUOTA_FL_EDQUOT));
lqe_write_unlock(lqe);
if (!!(desc->gl_flags & LQUOTA_FL_EDQUOT))
qsd_adjust_schedule(lqe, false, false);
if (wakeup)
wake_up(&lqe->lqe_waiters);
lqe_putref(lqe);
out:
req->rq_status = rc;
RETURN(rc);
}
/**
* qsd_id_lock_match() - Check whether a slave already own a ldlm lock for the
* quota identifier(QID)
* @lockh: local lock handle from lquota entry.
* @rlockh: remote lock handle of the matched lock, if any.
*
* Return:
* * %0 on successful look up and \lockh contains the lock handle.
* * %-ENOENT no lock found
*/
int qsd_id_lock_match(struct lustre_handle *lockh, struct lustre_handle *rlockh)
{
struct ldlm_lock *lock;
int rc;
ENTRY;
LASSERT(lockh);
if (!lustre_handle_is_used(lockh))
RETURN(-ENOENT);
rc = ldlm_lock_addref_try(lockh, qsd_id_einfo.ei_mode);
if (rc)
RETURN(-ENOENT);
LASSERT(lustre_handle_is_used(lockh));
ldlm_lock_dump_handle(D_QUOTA, lockh);
if (!rlockh)
/* caller not interested in remote handle */
RETURN(0);
/*
* look up lock associated with local handle and extract remote handle
* to be packed in quota request
*/
lock = ldlm_handle2lock(lockh);
LASSERT(lock != NULL);
lustre_handle_copy(rlockh, &lock->l_remote_handle);
ldlm_lock_put(lock);
RETURN(0);
}
int qsd_id_lock_cancel(const struct lu_env *env, struct lquota_entry *lqe)
{
struct qsd_thread_info *qti = qsd_info(env);
int rc;
ENTRY;
lqe_write_lock(lqe);
if (lqe->lqe_pending_write || lqe->lqe_waiting_write ||
lqe->lqe_usage || lqe->lqe_granted) {
lqe_write_unlock(lqe);
RETURN(0);
}
lustre_handle_copy(&qti->qti_lockh, &lqe->lqe_lockh);
if (lustre_handle_is_used(&qti->qti_lockh)) {
memset(&lqe->lqe_lockh, 0, sizeof(lqe->lqe_lockh));
qsd_set_qunit(lqe, 0);
qsd_set_edquot(lqe, false);
}
lqe_write_unlock(lqe);
rc = qsd_id_lock_match(&qti->qti_lockh, NULL);
if (rc)
RETURN(rc);
ldlm_lock_decref_and_cancel(&qti->qti_lockh, qsd_id_einfo.ei_mode);
RETURN(0);
}
