#include <obd.h>
#include <obd_support.h>
+#include <llog_swab.h>
+
#include "mdt_internal.h"
static unsigned int max_mod_rpcs_per_client = 8;
-CFS_MODULE_PARM(max_mod_rpcs_per_client, "i", uint, 0644,
- "maximum number of modify RPCs in flight allowed per client");
+module_param(max_mod_rpcs_per_client, uint, 0644);
+MODULE_PARM_DESC(max_mod_rpcs_per_client, "maximum number of modify RPCs in flight allowed per client");
mdl_mode_t mdt_mdl_lock_modes[] = {
[LCK_MINMODE] = MDL_MINMODE,
const struct lu_name *lname)
{
lh->mlh_reg_mode = lock_mode;
+ lh->mlh_pdo_mode = LCK_MINMODE;
lh->mlh_rreg_mode = lock_mode;
lh->mlh_type = MDT_PDO_LOCK;
EXIT;
}
-static int mdt_getstatus(struct tgt_session_info *tsi)
+static int mdt_lookup_fileset(struct mdt_thread_info *info, const char *fileset,
+ struct lu_fid *fid)
+{
+ struct mdt_device *mdt = info->mti_mdt;
+ struct lu_name *lname = &info->mti_name;
+ char *name = NULL;
+ struct mdt_object *parent;
+ u32 mode;
+ int rc = 0;
+
+ LASSERT(!info->mti_cross_ref);
+
+ OBD_ALLOC(name, NAME_MAX + 1);
+ if (name == NULL)
+ return -ENOMEM;
+ lname->ln_name = name;
+
+ /*
+ * We may want to allow this to mount a completely separate
+ * fileset from the MDT in the future, but keeping it to
+ * ROOT/ only for now avoid potential security issues.
+ */
+ *fid = mdt->mdt_md_root_fid;
+
+ while (rc == 0 && fileset != NULL && *fileset != '\0') {
+ const char *s1 = fileset;
+ const char *s2;
+
+ while (*++s1 == '/')
+ ;
+ s2 = s1;
+ while (*s2 != '/' && *s2 != '\0')
+ s2++;
+
+ if (s2 == s1)
+ break;
+
+ fileset = s2;
+
+ lname->ln_namelen = s2 - s1;
+ if (lname->ln_namelen > NAME_MAX) {
+ rc = -EINVAL;
+ break;
+ }
+
+ /* reject .. as a path component */
+ if (lname->ln_namelen == 2 &&
+ strncmp(s1, "..", 2) == 0) {
+ rc = -EINVAL;
+ break;
+ }
+
+ strncpy(name, s1, lname->ln_namelen);
+ name[lname->ln_namelen] = '\0';
+
+ parent = mdt_object_find(info->mti_env, mdt, fid);
+ if (IS_ERR(parent)) {
+ rc = PTR_ERR(parent);
+ break;
+ }
+ /* Only got the fid of this obj by name */
+ fid_zero(fid);
+ rc = mdo_lookup(info->mti_env, mdt_object_child(parent), lname,
+ fid, &info->mti_spec);
+ mdt_object_put(info->mti_env, parent);
+ }
+ if (!rc) {
+ parent = mdt_object_find(info->mti_env, mdt, fid);
+ if (IS_ERR(parent))
+ rc = PTR_ERR(parent);
+ else {
+ mode = lu_object_attr(&parent->mot_obj);
+ mdt_object_put(info->mti_env, parent);
+ if (!S_ISDIR(mode))
+ rc = -ENOTDIR;
+ }
+ }
+
+ OBD_FREE(name, NAME_MAX + 1);
+
+ return rc;
+}
+
+static int mdt_get_root(struct tgt_session_info *tsi)
{
struct mdt_thread_info *info = tsi2mdt_info(tsi);
struct mdt_device *mdt = info->mti_mdt;
struct mdt_body *repbody;
+ char *fileset = NULL, *buffer = NULL;
int rc;
+ struct obd_export *exp = info->mti_exp;
+ char *nodemap_fileset;
ENTRY;
if (rc)
GOTO(out, rc = err_serious(rc));
- if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETSTATUS_PACK))
+ if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GET_ROOT_PACK))
GOTO(out, rc = err_serious(-ENOMEM));
repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
- repbody->mbo_fid1 = mdt->mdt_md_root_fid;
+ if (req_capsule_get_size(info->mti_pill, &RMF_NAME, RCL_CLIENT) > 0) {
+ fileset = req_capsule_client_get(info->mti_pill, &RMF_NAME);
+ if (fileset == NULL)
+ GOTO(out, rc = err_serious(-EFAULT));
+ }
+
+ nodemap_fileset = nodemap_get_fileset(exp->exp_target_data.ted_nodemap);
+ if (nodemap_fileset && nodemap_fileset[0]) {
+ if (fileset) {
+ /* consider fileset from client as a sub-fileset
+ * of the nodemap one */
+ OBD_ALLOC(buffer, PATH_MAX + 1);
+ if (buffer == NULL)
+ GOTO(out, rc = err_serious(-ENOMEM));
+ if (snprintf(buffer, PATH_MAX + 1, "%s/%s",
+ nodemap_fileset, fileset) >= PATH_MAX + 1)
+ GOTO(out, rc = err_serious(-EINVAL));
+ fileset = buffer;
+ } else {
+ /* enforce fileset as specified in the nodemap */
+ fileset = nodemap_fileset;
+ }
+ }
+
+ if (fileset) {
+ rc = mdt_lookup_fileset(info, fileset, &repbody->mbo_fid1);
+ if (rc < 0)
+ GOTO(out, rc = err_serious(rc));
+ } else {
+ repbody->mbo_fid1 = mdt->mdt_md_root_fid;
+ }
repbody->mbo_valid |= OBD_MD_FLID;
EXIT;
out:
mdt_thread_info_fini(info);
+ if (buffer)
+ OBD_FREE(buffer, PATH_MAX+1);
return rc;
}
void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b,
const struct lu_attr *attr, const struct lu_fid *fid)
{
- struct md_attr *ma = &info->mti_attr;
- struct obd_export *exp = info->mti_exp;
- struct lu_nodemap *nodemap = exp->exp_target_data.ted_nodemap;
+ struct md_attr *ma = &info->mti_attr;
+ struct obd_export *exp = info->mti_exp;
+ struct lu_nodemap *nodemap = NULL;
LASSERT(ma->ma_valid & MA_INODE);
b->mbo_nlink = attr->la_nlink;
b->mbo_valid |= OBD_MD_FLNLINK;
}
+ if (attr->la_valid & (LA_UID|LA_GID)) {
+ nodemap = nodemap_get_from_exp(exp);
+ if (IS_ERR(nodemap))
+ goto out;
+ }
if (attr->la_valid & LA_UID) {
b->mbo_uid = nodemap_map_id(nodemap, NODEMAP_UID,
NODEMAP_FS_TO_CLIENT,
attr->la_gid);
b->mbo_valid |= OBD_MD_FLGID;
}
+
b->mbo_mode = attr->la_mode;
if (attr->la_valid & LA_MODE)
b->mbo_valid |= OBD_MD_FLMODE;
if (fid != NULL && (b->mbo_valid & OBD_MD_FLSIZE))
CDEBUG(D_VFSTRACE, DFID": returning size %llu\n",
PFID(fid), (unsigned long long)b->mbo_size);
+
+out:
+ if (!IS_ERR_OR_NULL(nodemap))
+ nodemap_putref(nodemap);
}
static inline int mdt_body_has_lov(const struct lu_attr *la,
struct mdt_body *repbody;
struct lu_buf *buffer = &info->mti_buf;
struct obd_export *exp = info->mti_exp;
- struct lu_nodemap *nodemap = exp->exp_target_data.ted_nodemap;
int rc;
int is_root;
ENTRY;
}
#ifdef CONFIG_FS_POSIX_ACL
else if ((exp_connect_flags(req->rq_export) & OBD_CONNECT_ACL) &&
- (reqbody->mbo_valid & OBD_MD_FLACL))
+ (reqbody->mbo_valid & OBD_MD_FLACL)) {
+ struct lu_nodemap *nodemap = nodemap_get_from_exp(exp);
+ if (IS_ERR(nodemap))
+ RETURN(PTR_ERR(nodemap));
+
rc = mdt_pack_acl2body(info, repbody, o, nodemap);
+ nodemap_putref(nodemap);
+ }
#endif
out:
if (!mdt_object_exists(child)) {
LU_OBJECT_DEBUG(D_INFO, info->mti_env,
&child->mot_obj,
- "remote object doesn't exist.\n");
+ "remote object doesn't exist.");
mdt_object_unlock(info, child, lhc, 1);
RETURN(-ENOENT);
}
if (unlikely(!mdt_object_exists(parent)) && lu_name_is_valid(lname)) {
LU_OBJECT_DEBUG(D_INODE, info->mti_env,
&parent->mot_obj,
- "Parent doesn't exist!\n");
+ "Parent doesn't exist!");
RETURN(-ESTALE);
}
if (!mdt_object_exists(child)) {
LU_OBJECT_DEBUG(D_INODE, info->mti_env,
&child->mot_obj,
- "Object doesn't exist!\n");
+ "Object doesn't exist!");
GOTO(out_child, rc = -ENOENT);
}
}
static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
- void *karg, void *uarg);
+ void *karg, void __user *uarg);
static int mdt_set_info(struct tgt_session_info *tsi)
{
int id, rc;
struct mdt_device *mdt = mdt_exp2dev(exp);
struct lu_device *qmt = mdt->mdt_qmt_dev;
- struct lu_nodemap *nodemap = exp->exp_target_data.ted_nodemap;
+ struct lu_nodemap *nodemap;
ENTRY;
oqctl = req_capsule_client_get(pill, &RMF_OBD_QUOTACTL);
if (rc)
RETURN(err_serious(rc));
+ nodemap = nodemap_get_from_exp(exp);
+ if (IS_ERR(nodemap))
+ RETURN(PTR_ERR(nodemap));
+
switch (oqctl->qc_cmd) {
/* master quotactl */
case Q_SETINFO:
case Q_SETQUOTA:
if (!nodemap_can_setquota(nodemap))
- RETURN(-EPERM);
+ GOTO(out_nodemap, rc = -EPERM);
case Q_GETINFO:
case Q_GETQUOTA:
if (qmt == NULL)
- RETURN(-EOPNOTSUPP);
+ GOTO(out_nodemap, rc = -EOPNOTSUPP);
/* slave quotactl */
case Q_GETOINFO:
case Q_GETOQUOTA:
break;
default:
CERROR("Unsupported quotactl command: %d\n", oqctl->qc_cmd);
- RETURN(-EFAULT);
+ GOTO(out_nodemap, rc = -EFAULT);
}
/* map uid/gid for remote client */
if (unlikely(oqctl->qc_cmd != Q_GETQUOTA &&
oqctl->qc_cmd != Q_GETINFO))
- RETURN(-EPERM);
+ GOTO(out_nodemap, rc = -EPERM);
if (oqctl->qc_type == USRQUOTA)
id = lustre_idmap_lookup_uid(NULL, idmap, 0,
id = lustre_idmap_lookup_gid(NULL, idmap, 0,
oqctl->qc_id);
else
- RETURN(-EINVAL);
+ GOTO(out_nodemap, rc = -EINVAL);
if (id == CFS_IDMAP_NOTFOUND) {
CDEBUG(D_QUOTA, "no mapping for id %u\n", oqctl->qc_id);
- RETURN(-EACCES);
+ GOTO(out_nodemap, rc = -EACCES);
}
}
repoqc = req_capsule_server_get(pill, &RMF_OBD_QUOTACTL);
if (repoqc == NULL)
- RETURN(err_serious(-EFAULT));
+ GOTO(out_nodemap, rc = err_serious(-EFAULT));
if (oqctl->qc_id != id)
swap(oqctl->qc_id, id);
default:
CERROR("Unsupported quotactl command: %d\n", oqctl->qc_cmd);
- RETURN(-EFAULT);
+ GOTO(out_nodemap, rc = -EFAULT);
}
if (oqctl->qc_id != id)
swap(oqctl->qc_id, id);
*repoqc = *oqctl;
- RETURN(rc);
+
+ EXIT;
+
+out_nodemap:
+ nodemap_putref(nodemap);
+
+ return rc;
}
/** clone llog ctxt from child (mdd)
{
struct dt_device *dt = mdt->mdt_bottom;
int rc;
+ ENTRY;
rc = dt->dd_ops->dt_commit_async(env, dt);
if (unlikely(rc != 0))
CWARN("%s: async commit start failed: rc = %d\n",
mdt_obd_name(mdt), rc);
+ atomic_inc(&mdt->mdt_async_commit_count);
+ EXIT;
}
/**
if (flag == LDLM_CB_CANCELING)
RETURN(0);
+
lock_res_and_lock(lock);
if (lock->l_blocking_ast != mdt_blocking_ast) {
unlock_res_and_lock(lock);
RETURN(0);
}
- if (mdt_cos_is_enabled(mdt) &&
- lock->l_req_mode & (LCK_PW | LCK_EX) &&
- lock->l_blocking_lock != NULL &&
- lock->l_client_cookie != lock->l_blocking_lock->l_client_cookie) {
- mdt_set_lock_sync(lock);
- }
+ if (lock->l_req_mode & (LCK_PW | LCK_EX) &&
+ lock->l_blocking_lock != NULL) {
+ if (mdt_cos_is_enabled(mdt) &&
+ lock->l_client_cookie !=
+ lock->l_blocking_lock->l_client_cookie)
+ mdt_set_lock_sync(lock);
+ else if (mdt_slc_is_enabled(mdt) &&
+ ldlm_is_cos_incompat(lock->l_blocking_lock))
+ mdt_set_lock_sync(lock);
+ }
rc = ldlm_blocking_ast_nocheck(lock);
/* There is no lock conflict if l_blocking_lock == NULL,
RETURN(rc);
}
-/* Used for cross-MDT lock */
+/*
+ * Blocking AST for cross-MDT lock
+ *
+ * Discard lock from uncommitted_slc_locks and cancel it.
+ *
+ * \param lock the lock which blocks a request or cancelling lock
+ * \param desc unused
+ * \param data unused
+ * \param flag indicates whether this cancelling or blocking callback
+ * \retval 0 on success
+ * \retval negative number on error
+ */
int mdt_remote_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
void *data, int flag)
{
struct lustre_handle lockh;
int rc;
+ ENTRY;
switch (flag) {
case LDLM_CB_BLOCKING:
}
break;
case LDLM_CB_CANCELING:
- LDLM_DEBUG(lock, "Revoke remote lock\n");
+ LDLM_DEBUG(lock, "Revoke remote lock");
+ /* discard slc lock here so that it can be cleaned anytime,
+ * especially for cleanup_resource() */
+ tgt_discard_slc_lock(lock);
break;
default:
LBUG();
}
+
RETURN(0);
}
/* Lock is pinned by ldlm_handle_enqueue0() as it is
* a resend case, however, it could be already destroyed
* due to client eviction or a raced cancel RPC. */
- LDLM_DEBUG_NOLOCK("Invalid lock handle "LPX64"\n",
+ LDLM_DEBUG_NOLOCK("Invalid lock handle "LPX64,
lhc->mlh_reg_lh.cookie);
RETURN(-ESTALE);
}
static int mdt_object_local_lock(struct mdt_thread_info *info,
struct mdt_object *o,
struct mdt_lock_handle *lh, __u64 ibits,
- bool nonblock)
+ bool nonblock, bool cos_incompat)
{
struct ldlm_namespace *ns = info->mti_mdt->mdt_namespace;
union ldlm_policy_data *policy = &info->mti_policy;
struct ldlm_res_id *res_id = &info->mti_res_id;
- __u64 dlmflags;
+ __u64 dlmflags = 0;
int rc;
ENTRY;
LASSERT(lh->mlh_reg_mode != LCK_MINMODE);
LASSERT(lh->mlh_type != MDT_NUL_LOCK);
+ if (cos_incompat) {
+ LASSERT(lh->mlh_reg_mode == LCK_PW ||
+ lh->mlh_reg_mode == LCK_EX);
+ dlmflags |= LDLM_FL_COS_INCOMPAT;
+ } else if (mdt_cos_is_enabled(info->mti_mdt)) {
+ dlmflags |= LDLM_FL_COS_ENABLED;
+ }
+
/* Only enqueue LOOKUP lock for remote object */
if (mdt_object_remote(o))
LASSERT(ibits == MDS_INODELOCK_LOOKUP);
memset(policy, 0, sizeof(*policy));
fid_build_reg_res_name(mdt_object_fid(o), res_id);
- dlmflags = LDLM_FL_ATOMIC_CB;
+ dlmflags |= LDLM_FL_ATOMIC_CB;
if (nonblock)
dlmflags |= LDLM_FL_BLOCK_NOWAIT;
static int
mdt_object_lock_internal(struct mdt_thread_info *info, struct mdt_object *o,
- struct mdt_lock_handle *lh, __u64 ibits,
- bool nonblock)
+ struct mdt_lock_handle *lh, __u64 ibits, bool nonblock,
+ bool cos_incompat)
{
struct mdt_lock_handle *local_lh = NULL;
int rc;
ENTRY;
- if (!mdt_object_remote(o))
- return mdt_object_local_lock(info, o, lh, ibits, nonblock);
+ if (!mdt_object_remote(o)) {
+ rc = mdt_object_local_lock(info, o, lh, ibits, nonblock,
+ cos_incompat);
+ RETURN(rc);
+ }
/* XXX do not support PERM/LAYOUT/XATTR lock for remote object yet */
ibits &= ~(MDS_INODELOCK_PERM | MDS_INODELOCK_LAYOUT |
/* Only enqueue LOOKUP lock for remote object */
if (ibits & MDS_INODELOCK_LOOKUP) {
- rc = mdt_object_local_lock(info, o, lh,
- MDS_INODELOCK_LOOKUP,
- nonblock);
+ rc = mdt_object_local_lock(info, o, lh, MDS_INODELOCK_LOOKUP,
+ nonblock, cos_incompat);
if (rc != ELDLM_OK)
RETURN(rc);
int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
struct mdt_lock_handle *lh, __u64 ibits)
{
- return mdt_object_lock_internal(info, o, lh, ibits, false);
+ return mdt_object_lock_internal(info, o, lh, ibits, false, false);
+}
+
+int mdt_reint_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
+ struct mdt_lock_handle *lh, __u64 ibits,
+ bool cos_incompat)
+{
+ LASSERT(lh->mlh_reg_mode == LCK_PW || lh->mlh_reg_mode == LCK_EX);
+ return mdt_object_lock_internal(info, o, lh, ibits, false,
+ cos_incompat);
}
int mdt_object_lock_try(struct mdt_thread_info *info, struct mdt_object *o,
struct mdt_lock_handle tmp = *lh;
int rc;
- rc = mdt_object_lock_internal(info, o, &tmp, ibits, true);
+ rc = mdt_object_lock_internal(info, o, &tmp, ibits, true, false);
+ if (rc == 0)
+ *lh = tmp;
+
+ return rc == 0;
+}
+
+int mdt_reint_object_lock_try(struct mdt_thread_info *info,
+ struct mdt_object *o, struct mdt_lock_handle *lh,
+ __u64 ibits, bool cos_incompat)
+{
+ struct mdt_lock_handle tmp = *lh;
+ int rc;
+
+ LASSERT(lh->mlh_reg_mode == LCK_PW || lh->mlh_reg_mode == LCK_EX);
+ rc = mdt_object_lock_internal(info, o, &tmp, ibits, true, cos_incompat);
if (rc == 0)
*lh = tmp;
struct mdt_device *mdt = info->mti_mdt;
struct ldlm_lock *lock = ldlm_handle2lock(h);
struct ptlrpc_request *req = mdt_info_req(info);
- int no_ack = 0;
+ int cos;
+
+ cos = (mdt_cos_is_enabled(mdt) ||
+ mdt_slc_is_enabled(mdt));
LASSERTF(lock != NULL, "no lock for cookie "LPX64"\n",
h->cookie);
+
/* there is no request if mdt_object_unlock() is called
* from mdt_export_cleanup()->mdt_add_dirty_flag() */
if (likely(req != NULL)) {
CDEBUG(D_HA, "request = %p reply state = %p"
" transno = "LPD64"\n", req,
req->rq_reply_state, req->rq_transno);
- if (mdt_cos_is_enabled(mdt)) {
- no_ack = 1;
+ if (cos) {
ldlm_lock_downgrade(lock, LCK_COS);
mode = LCK_COS;
}
- ptlrpc_save_lock(req, h, mode, no_ack);
+ ptlrpc_save_lock(req, h, mode, cos);
} else {
- ldlm_lock_decref(h, mode);
+ mdt_fid_unlock(h, mode);
}
if (mdt_is_lock_sync(lock)) {
CDEBUG(D_HA, "found sync-lock,"
}
/**
+ * Save cross-MDT lock in uncommitted_slc_locks
+ *
+ * Keep the lock referenced until transaction commit happens or release the lock
+ * immediately depending on input parameters.
+ *
+ * \param info thead info object
+ * \param h lock handle
+ * \param mode lock mode
+ * \param decref force immediate lock releasing
+ */
+static void mdt_save_remote_lock(struct mdt_thread_info *info,
+ struct lustre_handle *h, enum ldlm_mode mode,
+ int decref)
+{
+ ENTRY;
+
+ if (lustre_handle_is_used(h)) {
+ if (decref || !info->mti_has_trans ||
+ !(mode & (LCK_PW | LCK_EX))) {
+ ldlm_lock_decref_and_cancel(h, mode);
+ } else {
+ struct ldlm_lock *lock = ldlm_handle2lock(h);
+ struct ptlrpc_request *req = mdt_info_req(info);
+
+ LASSERT(req != NULL);
+ tgt_save_slc_lock(lock, req->rq_transno);
+ ldlm_lock_decref(h, mode);
+ }
+ h->cookie = 0ull;
+ }
+
+ EXIT;
+}
+
+/**
* Unlock mdt object.
*
* Immeditely release the regular lock and the PDO lock or save the
* \param decref force immediate lock releasing
*/
void mdt_object_unlock(struct mdt_thread_info *info, struct mdt_object *o,
- struct mdt_lock_handle *lh, int decref)
+ struct mdt_lock_handle *lh, int decref)
{
- ENTRY;
-
- mdt_save_lock(info, &lh->mlh_pdo_lh, lh->mlh_pdo_mode, decref);
- mdt_save_lock(info, &lh->mlh_reg_lh, lh->mlh_reg_mode, decref);
+ ENTRY;
- if (lustre_handle_is_used(&lh->mlh_rreg_lh))
- ldlm_lock_decref(&lh->mlh_rreg_lh, lh->mlh_rreg_mode);
+ mdt_save_lock(info, &lh->mlh_pdo_lh, lh->mlh_pdo_mode, decref);
+ mdt_save_lock(info, &lh->mlh_reg_lh, lh->mlh_reg_mode, decref);
+ mdt_save_remote_lock(info, &lh->mlh_rreg_lh, lh->mlh_rreg_mode, decref);
- EXIT;
+ EXIT;
}
struct mdt_object *mdt_object_find_lock(struct mdt_thread_info *info,
mdt_intent_lock_replace(struct mdt_thread_info *info,
struct ldlm_lock **lockp,
struct mdt_lock_handle *lh,
- __u64 flags)
+ __u64 flags, int result)
{
struct ptlrpc_request *req = mdt_info_req(info);
struct ldlm_lock *lock = *lockp;
RETURN(0);
}
- LASSERTF(new_lock != NULL,
- "lockh "LPX64"\n", lh->mlh_reg_lh.cookie);
+ if (new_lock == NULL && (flags & LDLM_FL_RESENT)) {
+ /* Lock is pinned by ldlm_handle_enqueue0() as it is
+ * a resend case, however, it could be already destroyed
+ * due to client eviction or a raced cancel RPC. */
+ LDLM_DEBUG_NOLOCK("Invalid lock handle "LPX64"\n",
+ lh->mlh_reg_lh.cookie);
+ lh->mlh_reg_lh.cookie = 0;
+ RETURN(-ESTALE);
+ }
+
+ LASSERTF(new_lock != NULL,
+ "lockh "LPX64" flags "LPX64" rc %d\n",
+ lh->mlh_reg_lh.cookie, flags, result);
/*
* If we've already given this lock to a client once, then we should
grc = mdt_getxattr(info);
- rc = mdt_intent_lock_replace(info, lockp, lhc, flags);
+ rc = mdt_intent_lock_replace(info, lockp, lhc, flags, 0);
if (mdt_info_req(info)->rq_repmsg != NULL)
ldlm_rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP);
GOTO(out_ucred, rc = ELDLM_LOCK_ABORTED);
}
- rc = mdt_intent_lock_replace(info, lockp, lhc, flags);
+ rc = mdt_intent_lock_replace(info, lockp, lhc, flags, rc);
EXIT;
out_ucred:
mdt_exit_ucred(info);
mdt_object_put(info->mti_env, obj);
if (rc == 0 && lustre_handle_is_used(&lhc->mlh_reg_lh))
- rc = mdt_intent_lock_replace(info, lockp, lhc, flags);
+ rc = mdt_intent_lock_replace(info, lockp, lhc, flags, rc);
out:
lhc->mlh_reg_lh.cookie = 0;
if (lustre_handle_is_used(&lhc->mlh_reg_lh) &&
(rc == 0 || rc == -MDT_EREMOTE_OPEN)) {
rep->lock_policy_res2 = 0;
- rc = mdt_intent_lock_replace(info, lockp, lhc, flags);
+ rc = mdt_intent_lock_replace(info, lockp, lhc, flags, rc);
RETURN(rc);
}
HABEO_REFERO, MDS_SET_INFO, mdt_set_info,
&RQF_OBD_SET_INFO, LUSTRE_MDS_VERSION),
TGT_MDT_HDL(0, MDS_GET_INFO, mdt_get_info),
-TGT_MDT_HDL(0 | HABEO_REFERO, MDS_GETSTATUS, mdt_getstatus),
+TGT_MDT_HDL(0 | HABEO_REFERO, MDS_GET_ROOT, mdt_get_root),
TGT_MDT_HDL(HABEO_CORPUS, MDS_GETATTR, mdt_getattr),
TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_GETATTR_NAME,
mdt_getattr_name),
m->mdt_enable_remote_dir = 0;
m->mdt_enable_remote_dir_gid = 0;
+ atomic_set(&m->mdt_mds_mds_conns, 0);
+ atomic_set(&m->mdt_async_commit_count, 0);
+
m->mdt_lu_dev.ld_ops = &mdt_lu_ops;
m->mdt_lu_dev.ld_obd = obd;
/* Set this lu_device to obd for error handling purposes. */
{ "mdt.group_upcall", NULL },
{ "mdt.quota_type", NULL },
{ "mdd.quota_type", NULL },
+ { "mdt.som", NULL },
{ "mdt.rootsquash", "mdt.root_squash" },
{ "mdt.nosquash_nid", "mdt.nosquash_nids" },
{ NULL }
RETURN(rc);
}
+static inline void mdt_enable_slc(struct mdt_device *mdt)
+{
+ if (mdt->mdt_lut.lut_sync_lock_cancel == NEVER_SYNC_ON_CANCEL)
+ mdt->mdt_lut.lut_sync_lock_cancel = BLOCKING_SYNC_ON_CANCEL;
+}
+
+static inline void mdt_disable_slc(struct mdt_device *mdt)
+{
+ if (mdt->mdt_lut.lut_sync_lock_cancel == BLOCKING_SYNC_ON_CANCEL)
+ mdt->mdt_lut.lut_sync_lock_cancel = NEVER_SYNC_ON_CANCEL;
+}
+
static int mdt_obd_disconnect(struct obd_export *exp)
{
int rc;
LASSERT(exp);
class_export_get(exp);
+ if ((exp_connect_flags(exp) & OBD_CONNECT_MDS_MDS) &&
+ !(exp_connect_flags(exp) & OBD_CONNECT_LIGHTWEIGHT)) {
+ struct mdt_device *mdt = mdt_dev(exp->exp_obd->obd_lu_dev);
+
+ if (atomic_dec_and_test(&mdt->mdt_mds_mds_conns))
+ mdt_disable_slc(mdt);
+ }
+
rc = server_disconnect_export(exp);
if (rc != 0)
CDEBUG(D_IOCTL, "server disconnect error: rc = %d\n", rc);
mdt = mdt_dev(obd->obd_lu_dev);
+ if ((data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) &&
+ !(data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT)) {
+ atomic_inc(&mdt->mdt_mds_mds_conns);
+ mdt_enable_slc(mdt);
+ }
+
/*
* first, check whether the stack is ready to handle requests
* XXX: probably not very appropriate method is used now
*exp = NULL;
} else {
*exp = lexp;
+ /* Because we do not want this export to be evicted by pinger,
+ * let's not add this export to the timed chain list. */
+ if (data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) {
+ spin_lock(&lexp->exp_obd->obd_dev_lock);
+ list_del_init(&lexp->exp_obd_chain_timed);
+ spin_unlock(&lexp->exp_obd->obd_dev_lock);
+ }
}
RETURN(rc);
* \param[in] info Per-thread common data shared by MDT level handlers.
* \param[in] obj Object to do path lookup of
* \param[in,out] fp User-provided struct to store path information
+ * \param[in] root_fid Root FID of current path should reach
*
* \retval 0 Lookup successful, path information stored in fp
* \retval -EAGAIN Lookup failed, usually because object is being moved
*/
static int mdt_path_current(struct mdt_thread_info *info,
struct mdt_object *obj,
- struct getinfo_fid2path *fp)
+ struct getinfo_fid2path *fp,
+ struct lu_fid *root_fid)
{
struct mdt_device *mdt = info->mti_mdt;
struct mdt_object *mdt_obj;
RETURN(-ENOMEM);
ldata.ld_buf = buf;
- ptr = fp->gf_path + fp->gf_pathlen - 1;
+ ptr = fp->gf_u.gf_path + fp->gf_pathlen - 1;
*ptr = 0;
--ptr;
*tmpfid = fp->gf_fid = *mdt_object_fid(obj);
- /* root FID only exists on MDT0, and fid2path should also ends at MDT0,
- * so checking root_fid can only happen on MDT0. */
- while (!lu_fid_eq(&mdt->mdt_md_root_fid, &fp->gf_fid)) {
+ while (!lu_fid_eq(root_fid, &fp->gf_fid)) {
struct lu_buf lmv_buf;
+ if (!lu_fid_eq(root_fid, &mdt->mdt_md_root_fid) &&
+ lu_fid_eq(&mdt->mdt_md_root_fid, &fp->gf_fid))
+ GOTO(out, rc = -ENOENT);
+
mdt_obj = mdt_object_find(info->mti_env, mdt, tmpfid);
if (IS_ERR(mdt_obj))
GOTO(out, rc = PTR_ERR(mdt_obj));
/* Pack the name in the end of the buffer */
ptr -= tmpname->ln_namelen;
- if (ptr - 1 <= fp->gf_path)
+ if (ptr - 1 <= fp->gf_u.gf_path)
GOTO(out, rc = -EOVERFLOW);
strncpy(ptr, tmpname->ln_name, tmpname->ln_namelen);
*(--ptr) = '/';
remote_out:
ptr++; /* skip leading / */
- memmove(fp->gf_path, ptr, fp->gf_path + fp->gf_pathlen - ptr);
+ memmove(fp->gf_u.gf_path, ptr,
+ fp->gf_u.gf_path + fp->gf_pathlen - ptr);
out:
RETURN(rc);
* \retval negative errno if there was a problem
*/
static int mdt_path(struct mdt_thread_info *info, struct mdt_object *obj,
- struct getinfo_fid2path *fp)
+ struct getinfo_fid2path *fp, struct lu_fid *root_fid)
{
struct mdt_device *mdt = info->mti_mdt;
int tries = 3;
if (fp->gf_pathlen < 3)
RETURN(-EOVERFLOW);
- if (lu_fid_eq(&mdt->mdt_md_root_fid, mdt_object_fid(obj))) {
- fp->gf_path[0] = '\0';
+ if (root_fid == NULL)
+ root_fid = &mdt->mdt_md_root_fid;
+
+ if (lu_fid_eq(root_fid, mdt_object_fid(obj))) {
+ fp->gf_u.gf_path[0] = '\0';
RETURN(0);
}
/* Retry multiple times in case file is being moved */
while (tries-- && rc == -EAGAIN)
- rc = mdt_path_current(info, obj, fp);
+ rc = mdt_path_current(info, obj, fp, root_fid);
RETURN(rc);
}
* \retval negative errno if there was a problem
*/
static int mdt_fid2path(struct mdt_thread_info *info,
+ struct lu_fid *root_fid,
struct getinfo_fid2path *fp)
{
struct mdt_device *mdt = info->mti_mdt;
RETURN(rc);
}
- rc = mdt_path(info, obj, fp);
+ rc = mdt_path(info, obj, fp, root_fid);
CDEBUG(D_INFO, "fid "DFID", path %s recno "LPX64" linkno %u\n",
- PFID(&fp->gf_fid), fp->gf_path, fp->gf_recno, fp->gf_linkno);
+ PFID(&fp->gf_fid), fp->gf_u.gf_path,
+ fp->gf_recno, fp->gf_linkno);
mdt_object_put(info->mti_env, obj);
RETURN(rc);
}
-static int mdt_rpc_fid2path(struct mdt_thread_info *info, void *key,
+static int mdt_rpc_fid2path(struct mdt_thread_info *info, void *key, int keylen,
void *val, int vallen)
{
struct getinfo_fid2path *fpout, *fpin;
+ struct lu_fid *root_fid = NULL;
int rc = 0;
fpin = key + cfs_size_round(sizeof(KEY_FID2PATH));
if (fpout->gf_pathlen != vallen - sizeof(*fpin))
RETURN(-EINVAL);
- rc = mdt_fid2path(info, fpout);
+ if (keylen >= cfs_size_round(sizeof(KEY_FID2PATH)) + sizeof(*fpin) +
+ sizeof(struct lu_fid)) {
+ /* client sent its root FID, which is normally fileset FID */
+ root_fid = fpin->gf_u.gf_root_fid;
+ if (ptlrpc_req_need_swab(info->mti_pill->rc_req))
+ lustre_swab_lu_fid(root_fid);
+
+ if (root_fid != NULL && !fid_is_sane(root_fid))
+ RETURN(-EINVAL);
+ }
+
+ rc = mdt_fid2path(info, root_fid, fpout);
RETURN(rc);
}
if (KEY_IS(KEY_FID2PATH)) {
struct mdt_thread_info *info = tsi2mdt_info(tsi);
- rc = mdt_rpc_fid2path(info, key, valout, *vallen);
+ rc = mdt_rpc_fid2path(info, key, keylen, valout, *vallen);
mdt_thread_info_fini(info);
} else {
rc = -EINVAL;
/* ioctls on obd dev */
static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
- void *karg, void *uarg)
+ void *karg, void __user *uarg)
{
struct lu_env env;
struct obd_device *obd = exp->exp_obd;
if (rc)
RETURN(rc);
- switch (cmd) {
- case OBD_IOC_SYNC:
- rc = mdt_device_sync(&env, mdt);
- break;
- case OBD_IOC_SET_READONLY:
- rc = dt->dd_ops->dt_ro(&env, dt);
- break;
+ switch (cmd) {
+ case OBD_IOC_SYNC:
+ rc = mdt_device_sync(&env, mdt);
+ break;
+ case OBD_IOC_SET_READONLY:
+ rc = dt_sync(&env, dt);
+ if (rc == 0)
+ rc = dt_ro(&env, dt);
+ break;
case OBD_IOC_ABORT_RECOVERY:
CERROR("%s: Aborting recovery for device\n", mdt_obd_name(mdt));
obd->obd_abort_recovery = 1;
rc = next->md_ops->mdo_iocontrol(&env, next, cmd, 0, &stop);
break;
}
+ case OBD_IOC_QUERY_LFSCK: {
+ struct md_device *next = mdt->mdt_child;
+ struct obd_ioctl_data *data = karg;
+
+ rc = next->md_ops->mdo_iocontrol(&env, next, cmd, 0,
+ data->ioc_inlbuf1);
+ break;
+ }
case OBD_IOC_GET_OBJ_VERSION: {
struct mdt_thread_info *mti;
mti = lu_context_key_get(&env.le_ctx, &mdt_thread_key);
git://git.whamcloud.com / fs / lustre-release.git / blobdiff