X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fquota%2Fqmt_lock.c;h=50fa1b32fe2d2e439ea1ed591a568f2bcb099460;hb=fa3598d0beca2bc8c5eb4c570210f5298a8613d0;hp=e37d041bbc6a0901516a0e91482ceeabd0660146;hpb=2d61b338d9990f2098ca646ecfe818668579ca76;p=fs%2Flustre-release.git diff --git a/lustre/quota/qmt_lock.c b/lustre/quota/qmt_lock.c index e37d041..50fa1b3 100644 --- a/lustre/quota/qmt_lock.c +++ b/lustre/quota/qmt_lock.c @@ -21,20 +21,19 @@ * GPL HEADER END */ /* - * Copyright (c) 2012 Intel, Inc. + * Copyright (c) 2012, 2016, 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 + #include +#include #include #include "qmt_internal.h" @@ -51,10 +50,14 @@ int qmt_intent_policy(const struct lu_env *env, struct lu_device *ld, struct quota_body *repbody; struct obd_uuid *uuid; struct lquota_lvb *lvb; - int rc; + struct ldlm_resource *res = (*lockp)->l_resource; + struct ldlm_reply *ldlm_rep; + int rc, lvb_len; ENTRY; req_capsule_extend(&req->rq_pill, &RQF_LDLM_INTENT_QUOTA); + req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER, + ldlm_lvbo_size(*lockp)); /* extract quota body and intent opc */ it = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT); @@ -76,16 +79,51 @@ int qmt_intent_policy(const struct lu_env *env, struct lu_device *ld, if (repbody == NULL) RETURN(err_serious(-EFAULT)); + ldlm_rep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP); + if (ldlm_rep == NULL) + RETURN(err_serious(-EFAULT)); + uuid = &(*lockp)->l_export->exp_client_uuid; switch (it->opc) { - case IT_QUOTA_DQACQ: - /* XXX: to be added in a next patch */ - GOTO(out, -EOPNOTSUPP); + case IT_QUOTA_DQACQ: { + struct lquota_entry *lqe; + struct ldlm_lock *lock; + + if (res->lr_name.name[LUSTRE_RES_ID_QUOTA_SEQ_OFF] == 0) + /* acquire on global lock? something is wrong ... */ + GOTO(out, rc = -EPROTO); + + /* verify global lock isn't stale */ + if (!lustre_handle_is_used(&reqbody->qb_glb_lockh)) + GOTO(out, rc = -ENOLCK); + + lock = ldlm_handle2lock(&reqbody->qb_glb_lockh); + if (lock == NULL) + GOTO(out, rc = -ENOLCK); + LDLM_LOCK_PUT(lock); + + lqe = res->lr_lvb_data; + LASSERT(lqe != NULL); + lqe_getref(lqe); + + /* acquire quota space */ + rc = qmt_dqacq0(env, lqe, qmt, uuid, reqbody->qb_flags, + reqbody->qb_count, reqbody->qb_usage, + repbody); + lqe_putref(lqe); + if (rc) + GOTO(out, rc); break; + } case IT_QUOTA_CONN: /* new connection from slave */ + + if (res->lr_name.name[LUSTRE_RES_ID_QUOTA_SEQ_OFF] != 0) + /* connection on per-ID lock? something is wrong ... */ + GOTO(out, rc = -EPROTO); + rc = qmt_pool_new_conn(env, qmt, &reqbody->qb_fid, &repbody->qb_slv_fid, &repbody->qb_slv_ver, uuid); @@ -94,19 +132,23 @@ int qmt_intent_policy(const struct lu_env *env, struct lu_device *ld, break; default: - CERROR("%s: invalid intent opcode: "LPU64"\n", qmt->qmt_svname, + CERROR("%s: invalid intent opcode: %llu\n", qmt->qmt_svname, it->opc); - GOTO(out, rc = err_serious(-EINVAL)); + GOTO(out, rc = -EINVAL); } /* on success, pack lvb in reply */ - req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER, - ldlm_lvbo_size(*lockp)); lvb = req_capsule_server_get(&req->rq_pill, &RMF_DLM_LVB); - ldlm_lvbo_fill(*lockp, lvb, ldlm_lvbo_size(*lockp)); - EXIT; + lvb_len = ldlm_lvbo_size(*lockp); + lvb_len = ldlm_lvbo_fill(*lockp, lvb, lvb_len); + if (lvb_len < 0) + GOTO(out, rc = lvb_len); + + req_capsule_shrink(&req->rq_pill, &RMF_DLM_LVB, lvb_len, RCL_SERVER); out: - return rc; + ldlm_rep->lock_policy_res2 = clear_serious(rc); + EXIT; + return ELDLM_OK; } /* @@ -137,14 +179,12 @@ int qmt_lvbo_init(struct lu_device *ld, struct ldlm_resource *res) /* initialize environment */ rc = lu_env_init(env, LCT_MD_THREAD); - if (rc) { - OBD_FREE_PTR(env); - RETURN(rc); - } + if (rc != 0) + GOTO(out_free, rc); qti = qmt_info(env); /* extract global index FID and quota identifier */ - fid_extract_quota_resid(&res->lr_name, &qti->qti_fid, &qti->qti_id); + fid_extract_from_quota_res(&qti->qti_fid, &qti->qti_id, &res->lr_name); /* sanity check the global index FID */ rc = lquota_extract_fid(&qti->qti_fid, &pool_id, &pool_type, &qtype); @@ -176,7 +216,7 @@ int qmt_lvbo_init(struct lu_device *ld, struct ldlm_resource *res) if (IS_ERR(obj)) GOTO(out, rc = PTR_ERR(obj)); if (!dt_object_exists(obj)) { - lu_object_put(env, &obj->do_lu); + dt_object_put(env, obj); GOTO(out, rc = -ENOENT); } @@ -185,10 +225,11 @@ int qmt_lvbo_init(struct lu_device *ld, struct ldlm_resource *res) CDEBUG(D_QUOTA, DFID" initialized lvb\n", PFID(&qti->qti_fid)); } - res->lr_lvb_len = sizeof(struct lquota_lvb); + res->lr_lvb_len = sizeof(struct lquota_lvb); EXIT; out: lu_env_fini(env); +out_free: OBD_FREE_PTR(env); return rc; } @@ -206,6 +247,8 @@ int qmt_lvbo_update(struct lu_device *ld, struct ldlm_resource *res, struct qmt_device *qmt = lu2qmt_dev(ld); struct lquota_entry *lqe; struct lquota_lvb *lvb; + struct ldlm_lock *lock; + struct obd_export *exp; int rc = 0; ENTRY; @@ -218,35 +261,78 @@ int qmt_lvbo_update(struct lu_device *ld, struct ldlm_resource *res, /* no need to update lvb for global quota locks */ RETURN(0); + lvb = req_capsule_server_swab_get(&req->rq_pill, &RMF_DLM_LVB, + lustre_swab_lquota_lvb); + if (lvb == NULL) { + CERROR("%s: failed to extract lvb from request\n", + qmt->qmt_svname); + RETURN(-EFAULT); + } + lqe = res->lr_lvb_data; LASSERT(lqe != NULL); + lqe_getref(lqe); + + LQUOTA_DEBUG(lqe, "releasing:%llu may release:%llu", + lvb->lvb_id_rel, lvb->lvb_id_may_rel); + + if (lvb->lvb_id_rel == 0) { + /* nothing to release */ + if (lvb->lvb_id_may_rel != 0) + /* but might still release later ... */ + lqe->lqe_may_rel += lvb->lvb_id_may_rel; + GOTO(out_lqe, rc = 0); + } /* allocate environement */ OBD_ALLOC_PTR(env); if (env == NULL) - RETURN(-ENOMEM); + GOTO(out_lqe, rc = -ENOMEM); /* initialize environment */ rc = lu_env_init(env, LCT_MD_THREAD); - if (rc) { - OBD_FREE_PTR(env); - RETURN(rc); - } + if (rc) + GOTO(out_env, rc); qti = qmt_info(env); - lvb = req_capsule_server_get(&req->rq_pill, &RMF_DLM_LVB); - if (lvb == NULL) { - CERROR("%s: failed to extract lvb from request\n", + /* The request is a glimpse callback which was sent via the + * reverse import to the slave. What we care about here is the + * export associated with the slave and req->rq_export is + * definitely not what we are looking for (it is actually set to + * NULL here). + * Therefore we extract the lock from the request argument + * and use lock->l_export. */ + lock = ldlm_request_lock(req); + if (IS_ERR(lock)) { + CERROR("%s: failed to get lock from request!\n", qmt->qmt_svname); - GOTO(out, rc); + GOTO(out_env_init, rc = PTR_ERR(lock)); } - /* XXX: Space release handling to be added in a next patch */ + exp = class_export_get(lock->l_export); + if (exp == NULL) { + CERROR("%s: failed to get export from lock!\n", + qmt->qmt_svname); + GOTO(out_env_init, rc = -EFAULT); + } + /* release quota space */ + rc = qmt_dqacq0(env, lqe, qmt, &exp->exp_client_uuid, + QUOTA_DQACQ_FL_REL, lvb->lvb_id_rel, 0, &qti->qti_body); + if (rc || qti->qti_body.qb_count != lvb->lvb_id_rel) + LQUOTA_ERROR(lqe, "failed to release quota space on glimpse " + "%llu!=%llu : rc = %d\n", qti->qti_body.qb_count, + lvb->lvb_id_rel, rc); + class_export_put(exp); + if (rc) + GOTO(out_env_init, rc); EXIT; -out: +out_env_init: lu_env_fini(env); +out_env: OBD_FREE_PTR(env); +out_lqe: + lqe_putref(lqe); return rc; } @@ -349,7 +435,7 @@ int qmt_lvbo_free(struct lu_device *ld, struct ldlm_resource *res) } /* release object reference */ - lu_object_put(env, &obj->do_lu); + dt_object_put(env, obj); lu_env_fini(env); OBD_FREE_PTR(env); } @@ -360,9 +446,92 @@ int qmt_lvbo_free(struct lu_device *ld, struct ldlm_resource *res) RETURN(0); } -typedef int (*qmt_glimpse_cb_t)(const struct lu_env *, struct qmt_device *, - struct obd_uuid *, union ldlm_gl_desc *, - void *); +typedef int (*qmt_glimpse_cb_t)(struct ldlm_lock *, void *); + +struct qmt_gl_lock_array { + unsigned long q_max; + unsigned long q_cnt; + struct ldlm_lock **q_locks; +}; + +static void qmt_free_lock_array(struct qmt_gl_lock_array *array) +{ + int i; + + if (array->q_max == 0) { + LASSERT(array->q_locks == NULL); + return; + } + + for (i = 0; i < array->q_cnt; i++) { + LASSERT(array->q_locks[i]); + LDLM_LOCK_RELEASE(array->q_locks[i]); + array->q_locks[i] = NULL; + } + array->q_cnt = 0; + OBD_FREE(array->q_locks, array->q_max * sizeof(*array->q_locks)); + array->q_locks = NULL; + array->q_max = 0; +} + +static int qmt_alloc_lock_array(struct ldlm_resource *res, + struct qmt_gl_lock_array *array, + qmt_glimpse_cb_t cb, void *arg) +{ + struct list_head *pos; + unsigned long count = 0; + int fail_cnt = 0; + ENTRY; + + LASSERT(!array->q_max && !array->q_cnt && !array->q_locks); +again: + lock_res(res); + /* scan list of granted locks */ + list_for_each(pos, &res->lr_granted) { + struct ldlm_lock *lock; + int rc; + + lock = list_entry(pos, struct ldlm_lock, l_res_link); + LASSERT(lock->l_export); + + if (cb != NULL) { + rc = cb(lock, arg); + /* slave should not be notified */ + if (rc == 0) + continue; + } + + count++; + if (array->q_max != 0 && array->q_cnt < array->q_max) { + array->q_locks[array->q_cnt] = LDLM_LOCK_GET(lock); + array->q_cnt++; + } + } + unlock_res(res); + + if (count > array->q_max) { + qmt_free_lock_array(array); + if (++fail_cnt > 5) + RETURN(-EAGAIN); + /* + * allocate more slots in case of more qualified locks are + * found during next loop + */ + array->q_max = count + count / 2 + 10; + count = 0; + LASSERT(array->q_locks == NULL && array->q_cnt == 0); + OBD_ALLOC(array->q_locks, + sizeof(*array->q_locks) * array->q_max); + if (array->q_locks == NULL) { + array->q_max = 0; + RETURN(-ENOMEM); + } + + goto again; + } + RETURN(0); +} + /* * Send glimpse callback to slaves holding a lock on resource \res. * This is used to notify slaves of new quota settings or to claim quota space @@ -380,51 +549,43 @@ static int qmt_glimpse_lock(const struct lu_env *env, struct qmt_device *qmt, struct ldlm_resource *res, union ldlm_gl_desc *desc, qmt_glimpse_cb_t cb, void *arg) { - cfs_list_t *tmp, *pos; - CFS_LIST_HEAD(gl_list); - int rc = 0; + struct list_head *tmp, *pos; + struct list_head gl_list = LIST_HEAD_INIT(gl_list); + struct qmt_gl_lock_array locks; + unsigned long i; + int rc = 0; ENTRY; - lock_res(res); - /* scan list of granted locks */ - cfs_list_for_each(pos, &res->lr_granted) { - struct ldlm_glimpse_work *work; - struct ldlm_lock *lock; - struct obd_uuid *uuid; - - lock = cfs_list_entry(pos, struct ldlm_lock, l_res_link); - LASSERT(lock->l_export); - uuid = &lock->l_export->exp_client_uuid; + memset(&locks, 0, sizeof(locks)); + rc = qmt_alloc_lock_array(res, &locks, cb, arg); + if (rc) { + CERROR("%s: failed to allocate glimpse lock array (%d)\n", + qmt->qmt_svname, rc); + RETURN(rc); + } - if (cb != NULL) { - rc = cb(env, qmt, uuid, desc, arg); - if (rc == 0) - /* slave should not be notified */ - continue; - if (rc < 0) - /* something wrong happened, we still notify */ - CERROR("%s: callback function failed to " - "determine whether slave %s should be " - "notified (%d)\n", qmt->qmt_svname, - obd_uuid2str(uuid), rc); - } + for (i = locks.q_cnt; i > 0; i--) { + struct ldlm_glimpse_work *work; OBD_ALLOC_PTR(work); if (work == NULL) { - CERROR("%s: failed to notify %s\n", qmt->qmt_svname, - obd_uuid2str(uuid)); + CERROR("%s: failed to notify a lock.\n", + qmt->qmt_svname); continue; } - cfs_list_add_tail(&work->gl_list, &gl_list); - work->gl_lock = LDLM_LOCK_GET(lock); + list_add_tail(&work->gl_list, &gl_list); + work->gl_lock = locks.q_locks[i - 1]; work->gl_flags = 0; work->gl_desc = desc; + locks.q_locks[i - 1] = NULL; + locks.q_cnt--; } - unlock_res(res); - if (cfs_list_empty(&gl_list)) { + qmt_free_lock_array(&locks); + + if (list_empty(&gl_list)) { CDEBUG(D_QUOTA, "%s: nobody to notify\n", qmt->qmt_svname); RETURN(0); } @@ -432,12 +593,12 @@ static int qmt_glimpse_lock(const struct lu_env *env, struct qmt_device *qmt, /* issue glimpse callbacks to all connected slaves */ rc = ldlm_glimpse_locks(res, &gl_list); - cfs_list_for_each_safe(pos, tmp, &gl_list) { + list_for_each_safe(pos, tmp, &gl_list) { struct ldlm_glimpse_work *work; - work = cfs_list_entry(pos, struct ldlm_glimpse_work, gl_list); + work = list_entry(pos, struct ldlm_glimpse_work, gl_list); - cfs_list_del(&work->gl_list); + list_del(&work->gl_list); CERROR("%s: failed to notify %s of new quota settings\n", qmt->qmt_svname, obd_uuid2str(&work->gl_lock->l_export->exp_client_uuid)); @@ -462,7 +623,6 @@ void qmt_glb_lock_notify(const struct lu_env *env, struct lquota_entry *lqe, struct qmt_thread_info *qti = qmt_info(env); struct qmt_pool_info *pool = lqe2qpi(lqe); struct ldlm_resource *res = NULL; - int rc; ENTRY; lquota_generate_fid(&qti->qti_fid, pool->qpi_key & 0x0000ffff, @@ -473,13 +633,14 @@ void qmt_glb_lock_notify(const struct lu_env *env, struct lquota_entry *lqe, qti->qti_gl_desc.lquota_desc.gl_flags = 0; qti->qti_gl_desc.lquota_desc.gl_hardlimit = lqe->lqe_hardlimit; qti->qti_gl_desc.lquota_desc.gl_softlimit = lqe->lqe_softlimit; + qti->qti_gl_desc.lquota_desc.gl_time = lqe->lqe_gracetime; qti->qti_gl_desc.lquota_desc.gl_ver = ver; /* look up ldlm resource associated with global index */ fid_build_reg_res_name(&qti->qti_fid, &qti->qti_resid); res = ldlm_resource_get(pool->qpi_qmt->qmt_ns, NULL, &qti->qti_resid, LDLM_PLAIN, 0); - if (res == NULL) { + if (IS_ERR(res)) { /* this might happen if no slaves have enqueued global quota * locks yet */ LQUOTA_DEBUG(lqe, "failed to lookup ldlm resource associated " @@ -487,19 +648,18 @@ void qmt_glb_lock_notify(const struct lu_env *env, struct lquota_entry *lqe, RETURN_EXIT; } - rc = qmt_glimpse_lock(env, pool->qpi_qmt, res, &qti->qti_gl_desc, - NULL, NULL); + qmt_glimpse_lock(env, pool->qpi_qmt, res, &qti->qti_gl_desc, + NULL, NULL); ldlm_resource_putref(res); EXIT; } /* Callback function used to select locks that should be glimpsed when * broadcasting the new qunit value */ -static int qmt_id_lock_cb(const struct lu_env *env, struct qmt_device *qmt, - struct obd_uuid *uuid, union ldlm_gl_desc *desc, - void *arg) +static int qmt_id_lock_cb(struct ldlm_lock *lock, void *arg) { struct obd_uuid *slv_uuid = arg; + struct obd_uuid *uuid = &lock->l_export->exp_client_uuid; if (slv_uuid != NULL && obd_uuid_equals(uuid, slv_uuid)) RETURN(0); @@ -521,7 +681,6 @@ static void qmt_id_lock_glimpse(const struct lu_env *env, struct qmt_thread_info *qti = qmt_info(env); struct qmt_pool_info *pool = lqe2qpi(lqe); struct ldlm_resource *res = NULL; - int rc; ENTRY; if (!lqe->lqe_enforced) @@ -529,18 +688,23 @@ static void qmt_id_lock_glimpse(const struct lu_env *env, lquota_generate_fid(&qti->qti_fid, pool->qpi_key & 0x0000ffff, pool->qpi_key >> 16, lqe->lqe_site->lqs_qtype); - fid_build_quota_resid(&qti->qti_fid, &lqe->lqe_id, &qti->qti_resid); + fid_build_quota_res_name(&qti->qti_fid, &lqe->lqe_id, &qti->qti_resid); res = ldlm_resource_get(qmt->qmt_ns, NULL, &qti->qti_resid, LDLM_PLAIN, 0); - if (res == NULL) { + if (IS_ERR(res)) { /* this might legitimately happens if slaves haven't had the * opportunity to enqueue quota lock yet. */ LQUOTA_DEBUG(lqe, "failed to lookup ldlm resource for per-ID " "lock "DFID, PFID(&qti->qti_fid)); + lqe_write_lock(lqe); + if (lqe->lqe_revoke_time == 0 && + lqe->lqe_qunit == pool->qpi_least_qunit) + lqe->lqe_revoke_time = ktime_get_seconds(); + lqe_write_unlock(lqe); RETURN_EXIT; } - lqe_read_lock(lqe); + lqe_write_lock(lqe); /* The purpose of glimpse callback on per-ID lock is twofold: * - notify slaves of new qunit value and hope they will release some * spare quota space in return @@ -553,18 +717,32 @@ static void qmt_id_lock_glimpse(const struct lu_env *env, else qti->qti_gl_desc.lquota_desc.gl_flags = 0; qti->qti_gl_desc.lquota_desc.gl_qunit = lqe->lqe_qunit; - lqe_read_unlock(lqe); + + if (lqe->lqe_revoke_time == 0 && + qti->qti_gl_desc.lquota_desc.gl_qunit == pool->qpi_least_qunit) + /* reset lqe_may_rel, it will be updated on glimpse callback + * replies if needed */ + lqe->lqe_may_rel = 0; /* The rebalance thread is the only thread which can issue glimpses */ LASSERT(!lqe->lqe_gl); lqe->lqe_gl = true; + lqe_write_unlock(lqe); /* issue glimpse callback to slaves */ - rc = qmt_glimpse_lock(env, qmt, res, &qti->qti_gl_desc, - uuid ? qmt_id_lock_cb : NULL, (void *)uuid); - + qmt_glimpse_lock(env, qmt, res, &qti->qti_gl_desc, + uuid ? qmt_id_lock_cb : NULL, (void *)uuid); + + lqe_write_lock(lqe); + if (lqe->lqe_revoke_time == 0 && + qti->qti_gl_desc.lquota_desc.gl_qunit == pool->qpi_least_qunit && + lqe->lqe_qunit == pool->qpi_least_qunit) { + lqe->lqe_revoke_time = ktime_get_seconds(); + qmt_adjust_edquot(lqe, ktime_get_real_seconds()); + } LASSERT(lqe->lqe_gl); lqe->lqe_gl = false; + lqe_write_unlock(lqe); ldlm_resource_putref(res); EXIT; @@ -583,15 +761,15 @@ void qmt_id_lock_notify(struct qmt_device *qmt, struct lquota_entry *lqe) ENTRY; lqe_getref(lqe); - cfs_spin_lock(&qmt->qmt_reba_lock); - if (!qmt->qmt_stopping && cfs_list_empty(&lqe->lqe_link)) { - cfs_list_add_tail(&lqe->lqe_link, &qmt->qmt_reba_list); + spin_lock(&qmt->qmt_reba_lock); + if (!qmt->qmt_stopping && list_empty(&lqe->lqe_link)) { + list_add_tail(&lqe->lqe_link, &qmt->qmt_reba_list); added = true; } - cfs_spin_unlock(&qmt->qmt_reba_lock); + spin_unlock(&qmt->qmt_reba_lock); if (added) - cfs_waitq_signal(&qmt->qmt_reba_thread.t_ctl_waitq); + wake_up(&qmt->qmt_reba_thread.t_ctl_waitq); else lqe_putref(lqe); EXIT; @@ -614,7 +792,6 @@ static int qmt_reba_thread(void *arg) struct l_wait_info lwi = { 0 }; struct lu_env *env; struct lquota_entry *lqe, *tmp; - char pname[MTI_NAME_MAXLEN]; int rc; ENTRY; @@ -629,30 +806,27 @@ static int qmt_reba_thread(void *arg) RETURN(rc); } - snprintf(pname, MTI_NAME_MAXLEN, "qmt_reba_%s", qmt->qmt_svname); - cfs_daemonize(pname); - thread_set_flags(thread, SVC_RUNNING); - cfs_waitq_signal(&thread->t_ctl_waitq); + wake_up(&thread->t_ctl_waitq); while (1) { l_wait_event(thread->t_ctl_waitq, - !cfs_list_empty(&qmt->qmt_reba_list) || + !list_empty(&qmt->qmt_reba_list) || !thread_is_running(thread), &lwi); - cfs_spin_lock(&qmt->qmt_reba_lock); - cfs_list_for_each_entry_safe(lqe, tmp, &qmt->qmt_reba_list, - lqe_link) { - cfs_list_del_init(&lqe->lqe_link); - cfs_spin_unlock(&qmt->qmt_reba_lock); + spin_lock(&qmt->qmt_reba_lock); + list_for_each_entry_safe(lqe, tmp, &qmt->qmt_reba_list, + lqe_link) { + list_del_init(&lqe->lqe_link); + spin_unlock(&qmt->qmt_reba_lock); if (thread_is_running(thread)) qmt_id_lock_glimpse(env, qmt, lqe, NULL); lqe_putref(lqe); - cfs_spin_lock(&qmt->qmt_reba_lock); + spin_lock(&qmt->qmt_reba_lock); } - cfs_spin_unlock(&qmt->qmt_reba_lock); + spin_unlock(&qmt->qmt_reba_lock); if (!thread_is_running(thread)) break; @@ -660,7 +834,7 @@ static int qmt_reba_thread(void *arg) lu_env_fini(env); OBD_FREE_PTR(env); thread_set_flags(thread, SVC_STOPPED); - cfs_waitq_signal(&thread->t_ctl_waitq); + wake_up(&thread->t_ctl_waitq); RETURN(rc); } @@ -671,15 +845,16 @@ int qmt_start_reba_thread(struct qmt_device *qmt) { struct ptlrpc_thread *thread = &qmt->qmt_reba_thread; struct l_wait_info lwi = { 0 }; - int rc; + struct task_struct *task; ENTRY; - rc = cfs_create_thread(qmt_reba_thread, (void *)qmt, 0); - if (rc < 0) { - CERROR("%s: failed to start rebalance thread (%d)\n", - qmt->qmt_svname, rc); + task = kthread_run(qmt_reba_thread, (void *)qmt, + "qmt_reba_%s", qmt->qmt_svname); + if (IS_ERR(task)) { + CERROR("%s: failed to start rebalance thread (%ld)\n", + qmt->qmt_svname, PTR_ERR(task)); thread_set_flags(thread, SVC_STOPPED); - RETURN(rc); + RETURN(PTR_ERR(task)); } l_wait_event(thread->t_ctl_waitq, @@ -700,10 +875,10 @@ void qmt_stop_reba_thread(struct qmt_device *qmt) struct l_wait_info lwi = { 0 }; thread_set_flags(thread, SVC_STOPPING); - cfs_waitq_signal(&thread->t_ctl_waitq); + wake_up(&thread->t_ctl_waitq); l_wait_event(thread->t_ctl_waitq, thread_is_stopped(thread), &lwi); } - LASSERT(cfs_list_empty(&qmt->qmt_reba_list)); + LASSERT(list_empty(&qmt->qmt_reba_list)); }