X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_handler.c;h=f786f008bb777e335d8df482d028786f6c6609ef;hp=409f556a10ccc44dd9145a40975f7222639efd3d;hb=06072de19cf490f52b9f20dd5a8d9dc8509ceb3f;hpb=3e2476ece46c3b92e9a5de76f877f607647873ae diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c index 409f556..f786f00 100644 --- a/lustre/mdt/mdt_handler.c +++ b/lustre/mdt/mdt_handler.c @@ -27,7 +27,7 @@ * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2010, 2013, Intel Corporation. + * Copyright (c) 2010, 2015, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -49,26 +49,30 @@ #define DEBUG_SUBSYSTEM S_MDS #include -/* - * struct OBD_{ALLOC,FREE}*() - */ -#include + +#include +#include +#include #include -/* struct ptlrpc_request */ +#include +#include #include -/* struct obd_export */ -#include -/* struct obd_device */ -#include -/* lu2dt_dev() */ -#include +#include #include -#include -#include "mdt_internal.h" -#include #include #include -#include +#include +#include +#include + +#include + +#include "mdt_internal.h" + + +static unsigned int max_mod_rpcs_per_client = 8; +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, @@ -81,15 +85,15 @@ mdl_mode_t mdt_mdl_lock_modes[] = { [LCK_GROUP] = MDL_GROUP }; -ldlm_mode_t mdt_dlm_lock_modes[] = { - [MDL_MINMODE] = LCK_MINMODE, - [MDL_EX] = LCK_EX, - [MDL_PW] = LCK_PW, - [MDL_PR] = LCK_PR, - [MDL_CW] = LCK_CW, - [MDL_CR] = LCK_CR, - [MDL_NL] = LCK_NL, - [MDL_GROUP] = LCK_GROUP +enum ldlm_mode mdt_dlm_lock_modes[] = { + [MDL_MINMODE] = LCK_MINMODE, + [MDL_EX] = LCK_EX, + [MDL_PW] = LCK_PW, + [MDL_PR] = LCK_PR, + [MDL_CW] = LCK_CW, + [MDL_CR] = LCK_CR, + [MDL_NL] = LCK_NL, + [MDL_GROUP] = LCK_GROUP }; static struct mdt_device *mdt_dev(struct lu_device *d); @@ -127,47 +131,48 @@ static struct lu_kmem_descr mdt_caches[] = { } }; -int mdt_get_disposition(struct ldlm_reply *rep, int flag) +__u64 mdt_get_disposition(struct ldlm_reply *rep, __u64 op_flag) { - if (!rep) - return 0; - return (rep->lock_policy_res1 & flag); + if (!rep) + return 0; + return rep->lock_policy_res1 & op_flag; } void mdt_clear_disposition(struct mdt_thread_info *info, - struct ldlm_reply *rep, int flag) + struct ldlm_reply *rep, __u64 op_flag) { if (info) { - info->mti_opdata &= ~flag; - tgt_opdata_clear(info->mti_env, flag); + info->mti_opdata &= ~op_flag; + tgt_opdata_clear(info->mti_env, op_flag); } if (rep) - rep->lock_policy_res1 &= ~flag; + rep->lock_policy_res1 &= ~op_flag; } void mdt_set_disposition(struct mdt_thread_info *info, - struct ldlm_reply *rep, int flag) + struct ldlm_reply *rep, __u64 op_flag) { if (info) { - info->mti_opdata |= flag; - tgt_opdata_set(info->mti_env, flag); + info->mti_opdata |= op_flag; + tgt_opdata_set(info->mti_env, op_flag); } if (rep) - rep->lock_policy_res1 |= flag; + rep->lock_policy_res1 |= op_flag; } -void mdt_lock_reg_init(struct mdt_lock_handle *lh, ldlm_mode_t lm) +void mdt_lock_reg_init(struct mdt_lock_handle *lh, enum ldlm_mode lm) { - lh->mlh_pdo_hash = 0; - lh->mlh_reg_mode = lm; + lh->mlh_pdo_hash = 0; + lh->mlh_reg_mode = lm; lh->mlh_rreg_mode = lm; - lh->mlh_type = MDT_REG_LOCK; + lh->mlh_type = MDT_REG_LOCK; } -void mdt_lock_pdo_init(struct mdt_lock_handle *lh, ldlm_mode_t lock_mode, +void mdt_lock_pdo_init(struct mdt_lock_handle *lh, enum ldlm_mode lock_mode, 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; @@ -263,12 +268,98 @@ static void mdt_lock_pdo_mode(struct mdt_thread_info *info, struct mdt_object *o 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; @@ -276,34 +367,48 @@ static int mdt_getstatus(struct tgt_session_info *tsi) 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; - repbody->mbo_valid |= OBD_MD_FLID; + 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)); + } - if (tsi->tsi_tgt->lut_mds_capa && - exp_connect_flags(info->mti_exp) & OBD_CONNECT_MDS_CAPA) { - struct mdt_object *root; - struct lustre_capa *capa; + 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; - root = mdt_object_find(info->mti_env, mdt, &repbody->mbo_fid1); - if (IS_ERR(root)) - GOTO(out, rc = PTR_ERR(root)); - - capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA1); - LASSERT(capa); - capa->lc_opc = CAPA_OPC_MDS_DEFAULT; - rc = mo_capa_get(info->mti_env, mdt_object_child(root), capa, - 0); - mdt_object_put(info->mti_env, root); - if (rc == 0) - repbody->mbo_valid |= OBD_MD_FLMDSCAPA; - } EXIT; out: mdt_thread_info_fini(info); + if (buffer) + OBD_FREE(buffer, PATH_MAX+1); return rc; } @@ -361,57 +466,131 @@ out: RETURN(rc); } -/** - * Pack SOM attributes into the reply. - * Call under a DLM UPDATE lock. +#ifdef CONFIG_FS_POSIX_ACL +/* + * Pack ACL data into the reply. UIDs/GIDs are mapped and filtered by nodemap. + * + * \param info thread info object + * \param repbody reply to pack ACLs into + * \param o mdt object of file to examine + * \param nodemap nodemap of client to reply to + * \retval 0 success + * \retval -errno error getting or parsing ACL from disk */ -static void mdt_pack_size2body(struct mdt_thread_info *info, - struct mdt_object *mo) -{ - struct mdt_body *b; - struct md_attr *ma = &info->mti_attr; - - LASSERT(ma->ma_attr.la_valid & LA_MODE); - b = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); - - /* Check if Size-on-MDS is supported, if this is a regular file, - * if SOM is enabled on the object and if SOM cache exists and valid. - * Otherwise do not pack Size-on-MDS attributes to the reply. */ - if (!(mdt_conn_flags(info) & OBD_CONNECT_SOM) || - !S_ISREG(ma->ma_attr.la_mode) || - !mdt_object_is_som_enabled(mo) || - !(ma->ma_valid & MA_SOM)) - return; +int mdt_pack_acl2body(struct mdt_thread_info *info, struct mdt_body *repbody, + struct mdt_object *o, struct lu_nodemap *nodemap) +{ + const struct lu_env *env = info->mti_env; + struct md_object *next = mdt_object_child(o); + struct lu_buf *buf = &info->mti_buf; + int rc; + + buf->lb_buf = req_capsule_server_get(info->mti_pill, &RMF_ACL); + buf->lb_len = req_capsule_get_size(info->mti_pill, &RMF_ACL, + RCL_SERVER); + if (buf->lb_len == 0) + return 0; - b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; - b->mbo_size = ma->ma_som->msd_size; - b->mbo_blocks = ma->ma_som->msd_blocks; + rc = mo_xattr_get(env, next, buf, XATTR_NAME_ACL_ACCESS); + if (rc < 0) { + if (rc == -ENODATA) { + repbody->mbo_aclsize = 0; + repbody->mbo_valid |= OBD_MD_FLACL; + rc = 0; + } else if (rc == -EOPNOTSUPP) { + rc = 0; + } else { + CERROR("%s: unable to read "DFID" ACL: rc = %d\n", + mdt_obd_name(info->mti_mdt), + PFID(mdt_object_fid(o)), rc); + } + } else { + rc = nodemap_map_acl(nodemap, buf->lb_buf, + rc, NODEMAP_FS_TO_CLIENT); + /* if all ACLs mapped out, rc is still >= 0 */ + if (rc < 0) { + CERROR("%s: nodemap_map_acl unable to parse "DFID + " ACL: rc = %d\n", mdt_obd_name(info->mti_mdt), + PFID(mdt_object_fid(o)), rc); + } else { + repbody->mbo_aclsize = rc; + repbody->mbo_valid |= OBD_MD_FLACL; + rc = 0; + } + } + return rc; } +#endif 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 md_attr *ma = &info->mti_attr; + struct obd_export *exp = info->mti_exp; + struct lu_nodemap *nodemap = NULL; - LASSERT(ma->ma_valid & MA_INODE); + LASSERT(ma->ma_valid & MA_INODE); - b->mbo_atime = attr->la_atime; - b->mbo_mtime = attr->la_mtime; - b->mbo_ctime = attr->la_ctime; - b->mbo_mode = attr->la_mode; - b->mbo_size = attr->la_size; - b->mbo_blocks = attr->la_blocks; - b->mbo_uid = attr->la_uid; - b->mbo_gid = attr->la_gid; - b->mbo_flags = attr->la_flags; - b->mbo_nlink = attr->la_nlink; - b->mbo_rdev = attr->la_rdev; + if (attr->la_valid & LA_ATIME) { + b->mbo_atime = attr->la_atime; + b->mbo_valid |= OBD_MD_FLATIME; + } + if (attr->la_valid & LA_MTIME) { + b->mbo_mtime = attr->la_mtime; + b->mbo_valid |= OBD_MD_FLMTIME; + } + if (attr->la_valid & LA_CTIME) { + b->mbo_ctime = attr->la_ctime; + b->mbo_valid |= OBD_MD_FLCTIME; + } + if (attr->la_valid & LA_FLAGS) { + b->mbo_flags = attr->la_flags; + b->mbo_valid |= OBD_MD_FLFLAGS; + } + if (attr->la_valid & LA_NLINK) { + 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_uid); + b->mbo_valid |= OBD_MD_FLUID; + } + if (attr->la_valid & LA_GID) { + b->mbo_gid = nodemap_map_id(nodemap, NODEMAP_GID, + 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 (attr->la_valid & LA_TYPE) + b->mbo_valid |= OBD_MD_FLTYPE; - /* XXX should pack the reply body according to lu_valid */ - b->mbo_valid |= OBD_MD_FLCTIME | OBD_MD_FLUID | - OBD_MD_FLGID | OBD_MD_FLTYPE | - OBD_MD_FLMODE | OBD_MD_FLNLINK | OBD_MD_FLFLAGS | - OBD_MD_FLATIME | OBD_MD_FLMTIME ; + if (fid != NULL) { + b->mbo_fid1 = *fid; + b->mbo_valid |= OBD_MD_FLID; + CDEBUG(D_INODE, DFID": nlink=%d, mode=%o, valid="LPX64"\n", + PFID(fid), b->mbo_nlink, b->mbo_mode, b->mbo_valid); + } + + if (info != NULL) + mdt_body_reverse_idmap(info, b); + + if (!(attr->la_valid & LA_TYPE)) + return; + + b->mbo_rdev = attr->la_rdev; + b->mbo_size = attr->la_size; + b->mbo_blocks = attr->la_blocks; if (!S_ISREG(attr->la_mode)) { b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS | OBD_MD_FLRDEV; @@ -436,20 +615,13 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b, b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; } - if (fid) { - b->mbo_fid1 = *fid; - b->mbo_valid |= OBD_MD_FLID; - CDEBUG(D_INODE, DFID": nlink=%d, mode=%o, size="LPU64"\n", - PFID(fid), b->mbo_nlink, - b->mbo_mode, b->mbo_size); - } - - if (info) - mdt_body_reverse_idmap(info, b); - 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, @@ -484,7 +656,8 @@ void mdt_client_compatibility(struct mdt_thread_info *info) EXIT; } -int mdt_attr_get_eabuf_size(struct mdt_thread_info *info, struct mdt_object *o) +static int mdt_attr_get_eabuf_size(struct mdt_thread_info *info, + struct mdt_object *o) { const struct lu_env *env = info->mti_env; int rc, rc2; @@ -728,19 +901,6 @@ int mdt_attr_get_complex(struct mdt_thread_info *info, GOTO(out, rc); } - if (need & MA_SOM && S_ISREG(mode)) { - buf->lb_buf = info->mti_xattr_buf; - buf->lb_len = sizeof(info->mti_xattr_buf); - CLASSERT(sizeof(struct som_attrs) <= - sizeof(info->mti_xattr_buf)); - rc2 = mo_xattr_get(info->mti_env, next, buf, XATTR_NAME_SOM); - rc2 = lustre_buf2som(info->mti_xattr_buf, rc2, ma->ma_som); - if (rc2 == 0) - ma->ma_valid |= MA_SOM; - else if (rc2 < 0 && rc2 != -ENODATA) - GOTO(out, rc = rc2); - } - if (need & MA_HSM && S_ISREG(mode)) { buf->lb_buf = info->mti_xattr_buf; buf->lb_len = sizeof(info->mti_xattr_buf); @@ -779,25 +939,26 @@ out: static int mdt_getattr_internal(struct mdt_thread_info *info, struct mdt_object *o, int ma_need) { - struct md_object *next = mdt_object_child(o); - const struct mdt_body *reqbody = info->mti_body; - struct ptlrpc_request *req = mdt_info_req(info); - struct md_attr *ma = &info->mti_attr; - struct lu_attr *la = &ma->ma_attr; - struct req_capsule *pill = info->mti_pill; - const struct lu_env *env = info->mti_env; - struct mdt_body *repbody; - struct lu_buf *buffer = &info->mti_buf; - int rc; - int is_root; - ENTRY; + struct md_object *next = mdt_object_child(o); + const struct mdt_body *reqbody = info->mti_body; + struct ptlrpc_request *req = mdt_info_req(info); + struct md_attr *ma = &info->mti_attr; + struct lu_attr *la = &ma->ma_attr; + struct req_capsule *pill = info->mti_pill; + const struct lu_env *env = info->mti_env; + struct mdt_body *repbody; + struct lu_buf *buffer = &info->mti_buf; + struct obd_export *exp = info->mti_exp; + int rc; + int is_root; + ENTRY; - if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) - RETURN(err_serious(-ENOMEM)); + if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) + RETURN(err_serious(-ENOMEM)); - repbody = req_capsule_server_get(pill, &RMF_MDT_BODY); + repbody = req_capsule_server_get(pill, &RMF_MDT_BODY); - ma->ma_valid = 0; + ma->ma_valid = 0; if (mdt_object_remote(o)) { /* This object is located on remote node.*/ @@ -853,8 +1014,6 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, ma->ma_need |= MA_LOV_DEF; } ma->ma_need |= ma_need; - if (ma->ma_need & MA_SOM) - ma->ma_som = &info->mti_u.som.data; rc = mdt_attr_get_complex(info, o, ma); if (unlikely(rc)) { @@ -865,10 +1024,11 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, } /* if file is released, check if a restore is running */ - if ((ma->ma_valid & MA_HSM) && (ma->ma_hsm.mh_flags & HS_RELEASED) && - mdt_hsm_restore_is_running(info, mdt_object_fid(o))) { - repbody->mbo_t_state = MS_RESTORE; + if (ma->ma_valid & MA_HSM) { repbody->mbo_valid |= OBD_MD_TSTATE; + if ((ma->ma_hsm.mh_flags & HS_RELEASED) && + mdt_hsm_restore_is_running(info, mdt_object_fid(o))) + repbody->mbo_t_state = MS_RESTORE; } is_root = lu_fid_eq(mdt_object_fid(o), &info->mti_mdt->mdt_md_root_fid); @@ -972,12 +1132,10 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, } if (reqbody->mbo_valid & OBD_MD_FLMODEASIZE) { - repbody->mbo_max_cookiesize = 0; repbody->mbo_max_mdsize = info->mti_mdt->mdt_max_mdsize; repbody->mbo_valid |= OBD_MD_FLMODEASIZE; - CDEBUG(D_INODE, "I am going to change the MAX_MD_SIZE & " - "MAX_COOKIE to : %d:%d\n", repbody->mbo_max_mdsize, - repbody->mbo_max_cookiesize); + CDEBUG(D_INODE, "changing the max MD size to %u\n", + repbody->mbo_max_mdsize); } if (exp_connect_rmtclient(info->mti_exp) && @@ -998,48 +1156,15 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, #ifdef CONFIG_FS_POSIX_ACL else if ((exp_connect_flags(req->rq_export) & OBD_CONNECT_ACL) && (reqbody->mbo_valid & OBD_MD_FLACL)) { - buffer->lb_buf = req_capsule_server_get(pill, &RMF_ACL); - buffer->lb_len = req_capsule_get_size(pill, - &RMF_ACL, RCL_SERVER); - if (buffer->lb_len > 0) { - rc = mo_xattr_get(env, next, buffer, - XATTR_NAME_ACL_ACCESS); - if (rc < 0) { - if (rc == -ENODATA) { - repbody->mbo_aclsize = 0; - repbody->mbo_valid |= OBD_MD_FLACL; - rc = 0; - } else if (rc == -EOPNOTSUPP) { - rc = 0; - } else { - CERROR("%s: unable to read "DFID - " ACL: rc = %d\n", - mdt_obd_name(info->mti_mdt), - PFID(mdt_object_fid(o)), rc); - } - } else { - repbody->mbo_aclsize = rc; - repbody->mbo_valid |= OBD_MD_FLACL; - rc = 0; - } - } + 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 - if (reqbody->mbo_valid & OBD_MD_FLMDSCAPA && - info->mti_mdt->mdt_lut.lut_mds_capa && - exp_connect_flags(info->mti_exp) & OBD_CONNECT_MDS_CAPA) { - struct lustre_capa *capa; - - capa = req_capsule_server_get(pill, &RMF_CAPA1); - LASSERT(capa); - capa->lc_opc = CAPA_OPC_MDS_DEFAULT; - rc = mo_capa_get(env, next, capa, 0); - if (rc) - RETURN(rc); - repbody->mbo_valid |= OBD_MD_FLMDSCAPA; - } - out: if (rc == 0) mdt_counter_incr(req, LPROC_MDT_GETATTR); @@ -1047,38 +1172,6 @@ out: RETURN(rc); } -static int mdt_renew_capa(struct mdt_thread_info *info) -{ - struct mdt_object *obj = info->mti_object; - struct mdt_body *body; - struct lustre_capa *capa, *c; - int rc; - ENTRY; - - /* if object doesn't exist, or server has disabled capability, - * return directly, client will find body->valid OBD_MD_FLOSSCAPA - * flag not set. - */ - if (!obj || !info->mti_mdt->mdt_lut.lut_oss_capa || - !(exp_connect_flags(info->mti_exp) & OBD_CONNECT_OSS_CAPA)) - RETURN(0); - - body = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); - LASSERT(body != NULL); - - c = req_capsule_client_get(info->mti_pill, &RMF_CAPA1); - LASSERT(c); - - capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA2); - LASSERT(capa); - - *capa = *c; - rc = mo_capa_get(info->mti_env, mdt_object_child(obj), capa, 1); - if (rc == 0) - body->mbo_valid |= OBD_MD_FLOSSCAPA; - RETURN(rc); -} - static int mdt_getattr(struct tgt_session_info *tsi) { struct mdt_thread_info *info = tsi2mdt_info(tsi); @@ -1091,15 +1184,6 @@ static int mdt_getattr(struct tgt_session_info *tsi) reqbody = req_capsule_client_get(pill, &RMF_MDT_BODY); LASSERT(reqbody); - - if (reqbody->mbo_valid & OBD_MD_FLOSSCAPA) { - rc = req_capsule_server_pack(pill); - if (unlikely(rc)) - RETURN(err_serious(rc)); - rc = mdt_renew_capa(info); - GOTO(out_shrink, rc); - } - LASSERT(obj != NULL); LASSERT(lu_object_assert_exists(&obj->mot_obj)); @@ -1152,11 +1236,6 @@ static int mdt_getattr(struct tgt_session_info *tsi) info->mti_cross_ref = !!(reqbody->mbo_valid & OBD_MD_FLCROSSREF); - /* - * Don't check capability at all, because rename might getattr for - * remote obj, and at that time no capability is available. - */ - mdt_set_capainfo(info, 1, &reqbody->mbo_fid1, BYPASS_CAPA); rc = mdt_getattr_internal(info, obj, 0); if (reqbody->mbo_valid & OBD_MD_FLRMTPERM) mdt_exit_ucred(info); @@ -1171,6 +1250,32 @@ out: return rc; } +/** + * Exchange MOF_LOV_CREATED flags between two objects after a + * layout swap. No assumption is made on whether o1 or o2 have + * created objects or not. + * + * \param[in,out] o1 First swap layout object + * \param[in,out] o2 Second swap layout object + */ +static void mdt_swap_lov_flag(struct mdt_object *o1, struct mdt_object *o2) +{ + __u64 o1_flags; + + mutex_lock(&o1->mot_lov_mutex); + mutex_lock(&o2->mot_lov_mutex); + + o1_flags = o1->mot_flags; + o1->mot_flags = (o1->mot_flags & ~MOF_LOV_CREATED) | + (o2->mot_flags & MOF_LOV_CREATED); + + o2->mot_flags = (o2->mot_flags & ~MOF_LOV_CREATED) | + (o1_flags & MOF_LOV_CREATED); + + mutex_unlock(&o2->mot_lov_mutex); + mutex_unlock(&o1->mot_lov_mutex); +} + static int mdt_swap_layouts(struct tgt_session_info *tsi) { struct mdt_thread_info *info; @@ -1195,17 +1300,7 @@ static int mdt_swap_layouts(struct tgt_session_info *tsi) info = tsi2mdt_info(tsi); if (info->mti_dlm_req != NULL) - ldlm_request_cancel(req, info->mti_dlm_req, 0); - - if (req_capsule_get_size(info->mti_pill, &RMF_CAPA1, RCL_CLIENT)) - mdt_set_capainfo(info, 0, &info->mti_body->mbo_fid1, - req_capsule_client_get(info->mti_pill, - &RMF_CAPA1)); - - if (req_capsule_get_size(info->mti_pill, &RMF_CAPA2, RCL_CLIENT)) - mdt_set_capainfo(info, 1, &info->mti_body->mbo_fid2, - req_capsule_client_get(info->mti_pill, - &RMF_CAPA2)); + ldlm_request_cancel(req, info->mti_dlm_req, 0, LATF_SKIP); o1 = info->mti_object; o = o2 = mdt_object_find(info->mti_env, info->mti_mdt, @@ -1245,18 +1340,22 @@ static int mdt_swap_layouts(struct tgt_session_info *tsi) mdt_lock_reg_init(lh2, LCK_EX); rc = mdt_object_lock(info, o1, lh1, MDS_INODELOCK_LAYOUT | - MDS_INODELOCK_XATTR, MDT_LOCAL_LOCK); + MDS_INODELOCK_XATTR); if (rc < 0) GOTO(put, rc); rc = mdt_object_lock(info, o2, lh2, MDS_INODELOCK_LAYOUT | - MDS_INODELOCK_XATTR, MDT_LOCAL_LOCK); + MDS_INODELOCK_XATTR); if (rc < 0) GOTO(unlock1, rc); rc = mo_swap_layouts(info->mti_env, mdt_object_child(o1), mdt_object_child(o2), msl->msl_flags); - GOTO(unlock2, rc); + if (rc < 0) + GOTO(unlock2, rc); + + mdt_swap_lov_flag(o1, o2); + unlock2: mdt_object_unlock(info, o2, lh2, rc); unlock1: @@ -1269,47 +1368,36 @@ out: } static int mdt_raw_lookup(struct mdt_thread_info *info, - struct mdt_object *parent, - const struct lu_name *lname, - struct ldlm_reply *ldlm_rep) -{ - struct md_object *next = mdt_object_child(info->mti_object); - const struct mdt_body *reqbody = info->mti_body; - struct lu_fid *child_fid = &info->mti_tmp_fid1; - struct mdt_body *repbody; - int rc; - ENTRY; + struct mdt_object *parent, + const struct lu_name *lname, + struct ldlm_reply *ldlm_rep) +{ + struct lu_fid *child_fid = &info->mti_tmp_fid1; + int rc; + ENTRY; - if (reqbody->mbo_valid != OBD_MD_FLID) - RETURN(0); + LASSERT(!info->mti_cross_ref); - LASSERT(!info->mti_cross_ref); + /* Only got the fid of this obj by name */ + fid_zero(child_fid); + rc = mdo_lookup(info->mti_env, mdt_object_child(info->mti_object), + lname, child_fid, &info->mti_spec); + if (rc == 0) { + struct mdt_body *repbody; - /* Only got the fid of this obj by name */ - fid_zero(child_fid); - rc = mdo_lookup(info->mti_env, next, lname, child_fid, - &info->mti_spec); -#if 0 - /* XXX is raw_lookup possible as intent operation? */ - if (rc != 0) { - if (rc == -ENOENT) - mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG); - RETURN(rc); - } else - mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS); - - repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); -#endif - if (rc == 0) { - repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); + repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); repbody->mbo_fid1 = *child_fid; repbody->mbo_valid = OBD_MD_FLID; - } - RETURN(1); + mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS); + } else if (rc == -ENOENT) { + mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG); + } + + RETURN(rc); } /* - * UPDATE lock should be taken against parent, and be release before exit; + * UPDATE lock should be taken against parent, and be released before exit; * child_bits lock should be taken against child, and be returned back: * (1)normal request should release the child lock; * (2)intent request will grant the lock to client. @@ -1323,7 +1411,6 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, struct mdt_body *reqbody = NULL; struct mdt_object *parent = info->mti_object; struct mdt_object *child; - struct md_object *next = mdt_object_child(parent); struct lu_fid *child_fid = &info->mti_tmp_fid1; struct lu_name *lname = NULL; struct mdt_lock_handle *lhp = NULL; @@ -1338,7 +1425,8 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, LASSERT(ergo(is_resent, lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)); - LASSERT(parent != NULL); + if (parent == NULL) + RETURN(-ENOENT); if (info->mti_cross_ref) { /* Only getattr on the child. Parent is on another node. */ @@ -1351,7 +1439,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, rc = mdt_check_resent_lock(info, child, lhc); if (rc < 0) { - RETURN(-EPROTO); + RETURN(rc); } else if (rc > 0) { mdt_lock_handle_init(lhc); mdt_lock_reg_init(lhc, LCK_PR); @@ -1364,8 +1452,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, MDS_INODELOCK_LAYOUT); child_bits |= MDS_INODELOCK_PERM | MDS_INODELOCK_UPDATE; - rc = mdt_object_lock(info, child, lhc, child_bits, - MDT_LOCAL_LOCK); + rc = mdt_object_lock(info, child, lhc, child_bits); if (rc < 0) RETURN(rc); } @@ -1374,12 +1461,11 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, 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); } - mdt_set_capainfo(info, 0, mdt_object_fid(child), BYPASS_CAPA); rc = mdt_getattr_internal(info, child, 0); if (unlikely(rc != 0)) mdt_object_unlock(info, child, lhc, 1); @@ -1415,7 +1501,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, 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); } @@ -1427,10 +1513,15 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, } if (lu_name_is_valid(lname)) { - rc = mdt_raw_lookup(info, parent, lname, ldlm_rep); - if (rc != 0) { - if (rc > 0) - rc = 0; + /* Always allow to lookup ".." */ + if (unlikely(lname->ln_namelen == 2 && + lname->ln_name[0] == '.' && + lname->ln_name[1] == '.')) + info->mti_spec.sp_permitted = 1; + + if (info->mti_body->mbo_valid == OBD_MD_FLID) { + rc = mdt_raw_lookup(info, parent, lname, ldlm_rep); + RETURN(rc); } @@ -1438,17 +1529,16 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, if (S_ISDIR(lu_object_attr(&parent->mot_obj))) { lhp = &info->mti_lh[MDT_LH_PARENT]; mdt_lock_pdo_init(lhp, LCK_PR, lname); - rc = mdt_object_lock(info, parent, lhp, - MDS_INODELOCK_UPDATE, - MDT_LOCAL_LOCK); - if (unlikely(rc != 0)) - RETURN(rc); - } + rc = mdt_object_lock(info, parent, lhp, + MDS_INODELOCK_UPDATE); + if (unlikely(rc != 0)) + RETURN(rc); + } /* step 2: lookup child's fid by name */ fid_zero(child_fid); - rc = mdo_lookup(info->mti_env, next, lname, child_fid, - &info->mti_spec); + rc = mdo_lookup(info->mti_env, mdt_object_child(parent), lname, + child_fid, &info->mti_spec); if (rc == -ENOENT) mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG); @@ -1478,22 +1568,22 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, if (unlikely(IS_ERR(child))) GOTO(out_parent, rc = PTR_ERR(child)); + OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RESEND, obd_timeout * 2); + if (!mdt_object_exists(child)) { + LU_OBJECT_DEBUG(D_INODE, info->mti_env, + &child->mot_obj, + "Object doesn't exist!"); + GOTO(out_child, rc = -ENOENT); + } + rc = mdt_check_resent_lock(info, child, lhc); if (rc < 0) { GOTO(out_child, rc); } else if (rc > 0) { - OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RESEND, obd_timeout*2); mdt_lock_handle_init(lhc); mdt_lock_reg_init(lhc, LCK_PR); try_layout = false; - if (!mdt_object_exists(child)) { - LU_OBJECT_DEBUG(D_INODE, info->mti_env, - &child->mot_obj, - "Object doesn't exist!\n"); - GOTO(out_child, rc = -ENOENT); - } - if (!(child_bits & MDS_INODELOCK_UPDATE) && mdt_object_exists(child) && !mdt_object_remote(child)) { struct md_attr *ma = &info->mti_attr; @@ -1532,12 +1622,12 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, child_bits |= MDS_INODELOCK_LAYOUT; /* try layout lock, it may fail to be granted due to * contention at LOOKUP or UPDATE */ - if (!mdt_object_lock_try(info, child, lhc, child_bits, - MDT_CROSS_LOCK)) { + if (!mdt_object_lock_try(info, child, lhc, + child_bits)) { child_bits &= ~MDS_INODELOCK_LAYOUT; LASSERT(child_bits != 0); rc = mdt_object_lock(info, child, lhc, - child_bits, MDT_CROSS_LOCK); + child_bits); } else { ma_need |= MA_LOV; } @@ -1546,22 +1636,15 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, * client will enqueue the lock to the remote MDT */ if (mdt_object_remote(child)) child_bits &= ~MDS_INODELOCK_UPDATE; - rc = mdt_object_lock(info, child, lhc, child_bits, - MDT_CROSS_LOCK); + rc = mdt_object_lock(info, child, lhc, child_bits); } if (unlikely(rc != 0)) GOTO(out_child, rc); } lock = ldlm_handle2lock(&lhc->mlh_reg_lh); - /* Get MA_SOM attributes if update lock is given. */ - if (lock && - lock->l_policy_data.l_inodebits.bits & MDS_INODELOCK_UPDATE && - S_ISREG(lu_object_attr(&mdt_object_child(child)->mo_lu))) - ma_need |= MA_SOM; /* finally, we can get attr for child. */ - mdt_set_capainfo(info, 1, child_fid, BYPASS_CAPA); rc = mdt_getattr_internal(info, child, ma_need); if (unlikely(rc != 0)) { mdt_object_unlock(info, child, lhc, 1); @@ -1573,8 +1656,6 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, "Lock res_id: "DLDLMRES", fid: "DFID"\n", PLDLMRES(lock->l_resource), PFID(mdt_object_fid(child))); - if (mdt_object_exists(child) && !mdt_object_remote(child)) - mdt_pack_size2body(info, child); } if (lock) LDLM_LOCK_PUT(lock); @@ -1607,7 +1688,7 @@ static int mdt_getattr_name(struct tgt_session_info *tsi) repbody->mbo_eadatasize = 0; repbody->mbo_aclsize = 0; - rc = mdt_init_ucred(info, reqbody); + rc = mdt_init_ucred_intent_getattr(info, reqbody); if (unlikely(rc)) GOTO(out_shrink, rc); @@ -1628,7 +1709,7 @@ out_shrink: } 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) { @@ -1682,6 +1763,9 @@ static int mdt_set_info(struct tgt_session_info *tsi) rc = mdt_iocontrol(OBD_IOC_CHANGELOG_CLEAR, req->rq_export, vallen, val, NULL); + } else if (KEY_IS(KEY_EVICT_BY_NID)) { + if (vallen > 0) + obd_export_evict_by_nid(req->rq_export->exp_obd, val); } else { RETURN(-EINVAL); } @@ -1811,7 +1895,10 @@ static int mdt_reint_internal(struct mdt_thread_info *info, if (rc != 0) GOTO(out_ucred, rc = err_serious(rc)); - if (mdt_check_resent(info, mdt_reconstruct, lhc)) { + rc = mdt_check_resent(info, mdt_reconstruct, lhc); + if (rc < 0) { + GOTO(out_ucred, rc); + } else if (rc == 1) { DEBUG_REQ(D_INODE, mdt_info_req(info), "resent opt."); rc = lustre_msg_get_status(mdt_info_req(info)->rq_repmsg); GOTO(out_ucred, rc); @@ -1973,6 +2060,7 @@ static int mdt_quotactl(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; ENTRY; oqctl = req_capsule_client_get(pill, &RMF_OBD_QUOTACTL); @@ -1983,29 +2071,27 @@ static int mdt_quotactl(struct tgt_session_info *tsi) if (rc) RETURN(err_serious(rc)); + nodemap = nodemap_get_from_exp(exp); + if (IS_ERR(nodemap)) + RETURN(PTR_ERR(nodemap)); + switch (oqctl->qc_cmd) { - case Q_QUOTACHECK: - case LUSTRE_Q_INVALIDATE: - case LUSTRE_Q_FINVALIDATE: - case Q_QUOTAON: - case Q_QUOTAOFF: - case Q_INITQUOTA: - /* deprecated, not used any more */ - RETURN(-EOPNOTSUPP); /* master quotactl */ - case Q_GETINFO: case Q_SETINFO: case Q_SETQUOTA: + if (!nodemap_can_setquota(nodemap)) + 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 */ @@ -2017,7 +2103,7 @@ static int mdt_quotactl(struct tgt_session_info *tsi) 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, @@ -2026,17 +2112,24 @@ static int mdt_quotactl(struct tgt_session_info *tsi) 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); } } + if (oqctl->qc_type == USRQUOTA) + id = nodemap_map_id(nodemap, NODEMAP_UID, + NODEMAP_CLIENT_TO_FS, id); + else if (oqctl->qc_type == GRPQUOTA) + id = nodemap_map_id(nodemap, NODEMAP_UID, + NODEMAP_CLIENT_TO_FS, id); + 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); @@ -2060,14 +2153,20 @@ static int mdt_quotactl(struct tgt_session_info *tsi) 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) @@ -2202,11 +2301,14 @@ static void mdt_device_commit_async(const struct lu_env *env, { 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; } /** @@ -2262,17 +2364,22 @@ int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, 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, @@ -2293,28 +2400,45 @@ int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, 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: ldlm_lock2handle(lock, &lockh); - rc = ldlm_cli_cancel(&lockh, LCF_ASYNC); + rc = ldlm_cli_cancel(&lockh, + ldlm_is_atomic_cb(lock) ? 0 : LCF_ASYNC); if (rc < 0) { CDEBUG(D_INODE, "ldlm_cli_cancel: %d\n", rc); RETURN(rc); } 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); } @@ -2323,14 +2447,20 @@ int mdt_check_resent_lock(struct mdt_thread_info *info, struct mdt_lock_handle *lhc) { /* the lock might already be gotten in ldlm_handle_enqueue() */ - if (lustre_handle_is_used(&lhc->mlh_reg_lh)) { + if (unlikely(lustre_handle_is_used(&lhc->mlh_reg_lh))) { struct ptlrpc_request *req = mdt_info_req(info); struct ldlm_lock *lock; lock = ldlm_handle2lock(&lhc->mlh_reg_lh); LASSERT(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT); - LASSERTF(lock != NULL, "Invalid lock handle "LPX64"\n", - lhc->mlh_reg_lh.cookie); + if (lock == NULL) { + /* 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, + lhc->mlh_reg_lh.cookie); + RETURN(-ESTALE); + } if (!fid_res_name_eq(mdt_object_fid(mo), &lock->l_resource->lr_name)) { @@ -2347,21 +2477,18 @@ int mdt_check_resent_lock(struct mdt_thread_info *info, return 1; } -int mdt_remote_object_lock(struct mdt_thread_info *mti, - struct mdt_object *o, const struct lu_fid *fid, - struct lustre_handle *lh, ldlm_mode_t mode, - __u64 ibits) +int mdt_remote_object_lock(struct mdt_thread_info *mti, struct mdt_object *o, + const struct lu_fid *fid, struct lustre_handle *lh, + enum ldlm_mode mode, __u64 ibits, bool nonblock) { struct ldlm_enqueue_info *einfo = &mti->mti_einfo; - ldlm_policy_data_t *policy = &mti->mti_policy; + union ldlm_policy_data *policy = &mti->mti_policy; struct ldlm_res_id *res_id = &mti->mti_res_id; int rc = 0; ENTRY; LASSERT(mdt_object_remote(o)); - LASSERT(ibits == MDS_INODELOCK_UPDATE); - fid_build_reg_res_name(fid, res_id); memset(einfo, 0, sizeof(*einfo)); @@ -2371,6 +2498,8 @@ int mdt_remote_object_lock(struct mdt_thread_info *mti, einfo->ei_cb_cp = ldlm_completion_ast; einfo->ei_enq_slave = 0; einfo->ei_res_id = res_id; + if (nonblock) + einfo->ei_nonblock = 1; memset(policy, 0, sizeof(*policy)); policy->l_inodebits.bits = ibits; @@ -2383,20 +2512,28 @@ int mdt_remote_object_lock(struct mdt_thread_info *mti, static int mdt_object_local_lock(struct mdt_thread_info *info, struct mdt_object *o, struct mdt_lock_handle *lh, __u64 ibits, - bool nonblock, int locality) + bool nonblock, bool cos_incompat) { - struct ldlm_namespace *ns = info->mti_mdt->mdt_namespace; - ldlm_policy_data_t *policy = &info->mti_policy; - struct ldlm_res_id *res_id = &info->mti_res_id; - __u64 dlmflags; - int rc; - ENTRY; + 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 = 0; + int rc; + ENTRY; LASSERT(!lustre_handle_is_used(&lh->mlh_reg_lh)); LASSERT(!lustre_handle_is_used(&lh->mlh_pdo_lh)); 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); @@ -2416,7 +2553,7 @@ static int mdt_object_local_lock(struct mdt_thread_info *info, 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; @@ -2473,26 +2610,33 @@ out_unlock: static int mdt_object_lock_internal(struct mdt_thread_info *info, struct mdt_object *o, - struct mdt_lock_handle *lh, __u64 ibits, - bool nonblock, int locality) + 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, - locality); - - if (locality == MDT_LOCAL_LOCK) { - CERROR("%s: try to get local lock for remote object" - DFID".\n", mdt_obd_name(info->mti_mdt), - PFID(mdt_object_fid(o))); - RETURN(-EPROTO); + 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 | MDS_INODELOCK_XATTR); + + /* Only enqueue LOOKUP lock for remote object */ + if (ibits & MDS_INODELOCK_LOOKUP) { + rc = mdt_object_local_lock(info, o, lh, MDS_INODELOCK_LOOKUP, + nonblock, cos_incompat); + if (rc != ELDLM_OK) + RETURN(rc); + + local_lh = lh; + } + if (ibits & MDS_INODELOCK_UPDATE) { /* Sigh, PDO needs to enqueue 2 locks right now, but * enqueue RPC can only request 1 lock, to avoid extra @@ -2510,36 +2654,54 @@ mdt_object_lock_internal(struct mdt_thread_info *info, struct mdt_object *o, rc = mdt_remote_object_lock(info, o, mdt_object_fid(o), &lh->mlh_rreg_lh, lh->mlh_rreg_mode, - MDS_INODELOCK_UPDATE); - if (rc != ELDLM_OK) - RETURN(rc); - } - - /* Only enqueue LOOKUP lock for remote object */ - if (ibits & MDS_INODELOCK_LOOKUP) { - rc = mdt_object_local_lock(info, o, lh, - MDS_INODELOCK_LOOKUP, - nonblock, locality); - if (rc != ELDLM_OK) + MDS_INODELOCK_UPDATE, nonblock); + if (rc != ELDLM_OK) { + if (local_lh != NULL) + mdt_object_unlock(info, o, local_lh, rc); RETURN(rc); + } } RETURN(0); } int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o, - struct mdt_lock_handle *lh, __u64 ibits, int locality) + struct mdt_lock_handle *lh, __u64 ibits) +{ + 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) { - return mdt_object_lock_internal(info, o, lh, ibits, false, locality); + 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 *lh, __u64 ibits, int locality) + struct mdt_lock_handle *lh, __u64 ibits) +{ + struct mdt_lock_handle tmp = *lh; + int rc; + + 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; - rc = mdt_object_lock_internal(info, o, &tmp, ibits, true, locality); + 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; @@ -2559,38 +2721,40 @@ int mdt_object_lock_try(struct mdt_thread_info *info, struct mdt_object *o, * \param mode lock mode * \param decref force immediate lock releasing */ -static -void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h, - ldlm_mode_t mode, int decref) +static void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h, + enum ldlm_mode mode, int decref) { - ENTRY; + ENTRY; - if (lustre_handle_is_used(h)) { - if (decref || !info->mti_has_trans || - !(mode & (LCK_PW | LCK_EX))){ - mdt_fid_unlock(h, mode); - } else { - 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; + if (lustre_handle_is_used(h)) { + if (decref || !info->mti_has_trans || + !(mode & (LCK_PW | LCK_EX))) { + mdt_fid_unlock(h, mode); + } else { + struct mdt_device *mdt = info->mti_mdt; + struct ldlm_lock *lock = ldlm_handle2lock(h); + struct ptlrpc_request *req = mdt_info_req(info); + int cos; + + cos = (mdt_cos_is_enabled(mdt) || + mdt_slc_is_enabled(mdt)); + + LASSERTF(lock != NULL, "no lock for cookie "LPX64"\n", + h->cookie); - 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," @@ -2607,10 +2771,45 @@ void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h, } /** + * 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 - * lock in reqeuest and keep them referenced until client ACK or + * lock in request and keep them referenced until client ACK or * transaction commit. * * \param info thread info object @@ -2619,17 +2818,15 @@ void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h, * \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, @@ -2643,8 +2840,7 @@ struct mdt_object *mdt_object_find_lock(struct mdt_thread_info *info, if (!IS_ERR(o)) { int rc; - rc = mdt_object_lock(info, o, lh, ibits, - MDT_LOCAL_LOCK); + rc = mdt_object_lock(info, o, lh, ibits); if (rc != 0) { mdt_object_put(info->mti_env, o); o = ERR_PTR(rc); @@ -2698,27 +2894,11 @@ static int mdt_body_unpack(struct mdt_thread_info *info, __u32 flags) RETURN(-EINVAL); } - /* - * Do not get size or any capa fields before we check that request - * contains capa actually. There are some requests which do not, for - * instance MDS_IS_SUBDIR. - */ - if (req_capsule_has_field(pill, &RMF_CAPA1, RCL_CLIENT) && - req_capsule_get_size(pill, &RMF_CAPA1, RCL_CLIENT)) - mdt_set_capainfo(info, 0, &body->mbo_fid1, - req_capsule_client_get(pill, &RMF_CAPA1)); - obj = mdt_object_find(env, info->mti_mdt, &body->mbo_fid1); if (!IS_ERR(obj)) { - if ((flags & HABEO_CORPUS) && - !mdt_object_exists(obj)) { + if ((flags & HABEO_CORPUS) && !mdt_object_exists(obj)) { mdt_object_put(env, obj); - /* for capability renew ENOENT will be handled in - * mdt_renew_capa */ - if (body->mbo_valid & OBD_MD_FLOSSCAPA) - rc = 0; - else - rc = -ENOENT; + rc = -ENOENT; } else { info->mti_object = obj; rc = 0; @@ -2754,17 +2934,6 @@ static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags) RETURN(rc); } -static int mdt_init_capa_ctxt(const struct lu_env *env, struct mdt_device *m) -{ - struct md_device *next = m->mdt_child; - - return next->md_ops->mdo_init_capa_ctxt(env, next, - m->mdt_lut.lut_mds_capa, - m->mdt_capa_timeout, - m->mdt_capa_alg, - m->mdt_capa_keys); -} - void mdt_lock_handle_init(struct mdt_lock_handle *lh) { lh->mlh_type = MDT_NUL_LOCK; @@ -2819,6 +2988,8 @@ void mdt_thread_info_init(struct ptlrpc_request *req, info->mti_spec.no_create = 0; info->mti_spec.sp_rm_entry = 0; + info->mti_spec.sp_permitted = 0; + info->mti_spec.sp_migrate_close = 0; info->mti_spec.u.sp_ea.eadata = NULL; info->mti_spec.u.sp_ea.eadatalen = 0; @@ -2846,28 +3017,14 @@ void mdt_thread_info_fini(struct mdt_thread_info *info) struct mdt_thread_info *tsi2mdt_info(struct tgt_session_info *tsi) { struct mdt_thread_info *mti; - struct lustre_capa *lc; mti = mdt_th_info(tsi->tsi_env); LASSERT(mti != NULL); mdt_thread_info_init(tgt_ses_req(tsi), mti); if (tsi->tsi_corpus != NULL) { - struct req_capsule *pill = tsi->tsi_pill; - mti->mti_object = mdt_obj(tsi->tsi_corpus); lu_object_get(tsi->tsi_corpus); - - /* - * XXX: must be part of tgt_mdt_body_unpack but moved here - * due to mdt_set_capainfo(). - */ - if (req_capsule_has_field(pill, &RMF_CAPA1, RCL_CLIENT) && - req_capsule_get_size(pill, &RMF_CAPA1, RCL_CLIENT) > 0) { - lc = req_capsule_client_get(pill, &RMF_CAPA1); - mdt_set_capainfo(mti, 0, &tsi->tsi_mdt_body->mbo_fid1, - lc); - } } mti->mti_body = tsi->tsi_mdt_body; mti->mti_dlm_req = tsi->tsi_dlm_req; @@ -2884,8 +3041,10 @@ static int mdt_tgt_connect(struct tgt_session_info *tsi) if (OBD_FAIL_CHECK(OBD_FAIL_TGT_DELAY_CONDITIONAL) && cfs_fail_val == - tsi2mdt_info(tsi)->mti_mdt->mdt_seq_site.ss_node_id) - schedule_timeout_and_set_state(TASK_UNINTERRUPTIBLE, HZ * 3); + tsi2mdt_info(tsi)->mti_mdt->mdt_seq_site.ss_node_id) { + set_current_state(TASK_UNINTERRUPTIBLE); + schedule_timeout(msecs_to_jiffies(3 * MSEC_PER_SEC)); + } rc = tgt_connect(tsi); if (rc != 0) @@ -3011,7 +3170,7 @@ static int 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; @@ -3025,8 +3184,19 @@ mdt_intent_lock_replace(struct mdt_thread_info *info, 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 @@ -3117,12 +3287,12 @@ static void mdt_intent_fixup_resent(struct mdt_thread_info *info, return; } - /* - * If the xid matches, then we know this is a resent request, and allow - * it. (It's probably an OPEN, for which we don't send a lock. - */ - if (req_xid_is_last(req)) - return; + /* + * If the xid matches, then we know this is a resent request, and allow + * it. (It's probably an OPEN, for which we don't send a lock. + */ + if (req_can_reconstruct(req, NULL)) + return; /* * This remote handle isn't enqueued, so we never received or processed @@ -3153,15 +3323,14 @@ static int mdt_intent_getxattr(enum mdt_it_code opcode, if (!lustre_handle_is_used(&lhc->mlh_reg_lh)) { mdt_lock_reg_init(lhc, (*lockp)->l_req_mode); rc = mdt_object_lock(info, info->mti_object, lhc, - MDS_INODELOCK_XATTR, - MDT_LOCAL_LOCK); + MDS_INODELOCK_XATTR); if (rc) return rc; } 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); @@ -3209,9 +3378,9 @@ static int mdt_intent_getattr(enum mdt_it_code opcode, GOTO(out_shrink, rc = -EINVAL); } - rc = mdt_init_ucred(info, reqbody); - if (rc) - GOTO(out_shrink, rc); + rc = mdt_init_ucred_intent_getattr(info, reqbody); + if (rc) + GOTO(out_shrink, rc); ldlm_rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP); mdt_set_disposition(info, ldlm_rep, DISP_IT_EXECD); @@ -3230,7 +3399,7 @@ static int mdt_intent_getattr(enum mdt_it_code opcode, 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); @@ -3251,6 +3420,7 @@ static int mdt_intent_layout(enum mdt_it_code opcode, struct layout_intent *layout; struct lu_fid *fid; struct mdt_object *obj = NULL; + int layout_size = 0; int rc = 0; ENTRY; @@ -3260,6 +3430,16 @@ static int mdt_intent_layout(enum mdt_it_code opcode, RETURN(-EINVAL); } + layout = req_capsule_client_get(info->mti_pill, &RMF_LAYOUT_INTENT); + if (layout == NULL) + RETURN(-EPROTO); + + if (layout->li_opc != LAYOUT_INTENT_ACCESS) { + CERROR("%s: Unsupported layout intent opc %d\n", + mdt_obd_name(info->mti_mdt), layout->li_opc); + RETURN(-EINVAL); + } + fid = &info->mti_tmp_fid2; fid_extract_from_res_name(fid, &(*lockp)->l_resource->lr_name); @@ -3268,40 +3448,33 @@ static int mdt_intent_layout(enum mdt_it_code opcode, obj = mdt_object_find(info->mti_env, info->mti_mdt, fid); if (IS_ERR(obj)) - RETURN(PTR_ERR(obj)); + GOTO(out, rc = PTR_ERR(obj)); if (mdt_object_exists(obj) && !mdt_object_remote(obj)) { - /* get the length of lsm */ - rc = mdt_attr_get_eabuf_size(info, obj); - if (rc < 0) { - mdt_object_put(info->mti_env, obj); - RETURN(rc); - } + layout_size = mdt_attr_get_eabuf_size(info, obj); + if (layout_size < 0) + GOTO(out_obj, rc = layout_size); - if (rc > info->mti_mdt->mdt_max_mdsize) - info->mti_mdt->mdt_max_mdsize = rc; + if (layout_size > info->mti_mdt->mdt_max_mdsize) + info->mti_mdt->mdt_max_mdsize = layout_size; } - mdt_object_put(info->mti_env, obj); - (*lockp)->l_lvb_type = LVB_T_LAYOUT; - req_capsule_set_size(info->mti_pill, &RMF_DLM_LVB, RCL_SERVER, rc); + req_capsule_set_size(info->mti_pill, &RMF_DLM_LVB, RCL_SERVER, + layout_size); rc = req_capsule_server_pack(info->mti_pill); - if (rc != 0) - RETURN(-EINVAL); + GOTO(out_obj, rc); - if (lustre_handle_is_used(&lhc->mlh_reg_lh)) - rc = mdt_intent_lock_replace(info, lockp, lhc, flags); +out_obj: + mdt_object_put(info->mti_env, obj); - layout = req_capsule_client_get(info->mti_pill, &RMF_LAYOUT_INTENT); - LASSERT(layout != NULL); - if (layout->li_opc == LAYOUT_INTENT_ACCESS) - /* return to normal/resent ldlm handling */ - RETURN(rc); + if (rc == 0 && lustre_handle_is_used(&lhc->mlh_reg_lh)) + rc = mdt_intent_lock_replace(info, lockp, lhc, flags, rc); - CERROR("%s: Unsupported layout intent (%d)\n", - mdt_obd_name(info->mti_mdt), layout->li_opc); - RETURN(-EINVAL); +out: + lhc->mlh_reg_lh.cookie = 0; + + return rc; } static int mdt_intent_reint(enum mdt_it_code opcode, @@ -3348,19 +3521,21 @@ static int mdt_intent_reint(enum mdt_it_code opcode, /* the open lock or the lock for cross-ref object should be * returned to the client */ - if (rc == -MDT_EREMOTE_OPEN || mdt_get_disposition(rep, DISP_OPEN_LOCK)) { - LASSERT(lustre_handle_is_used(&lhc->mlh_reg_lh)); + 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); } rep->lock_policy_res2 = clear_serious(rc); if (rep->lock_policy_res2 == -ENOENT && - mdt_get_disposition(rep, DISP_LOOKUP_NEG)) - rep->lock_policy_res2 = 0; + mdt_get_disposition(rep, DISP_LOOKUP_NEG) && + !mdt_get_disposition(rep, DISP_OPEN_CREATE)) + rep->lock_policy_res2 = 0; + lhc->mlh_reg_lh.cookie = 0ull; if (rc == -ENOTCONN || rc == -ENODEV || rc == -EOVERFLOW) { /**< if VBR failure then return error */ /* @@ -3369,91 +3544,76 @@ static int mdt_intent_reint(enum mdt_it_code opcode, * will detect this, then disconnect, reconnect the import * immediately, instead of impacting the following the rpc. */ - lhc->mlh_reg_lh.cookie = 0ull; RETURN(rc); - } else { - /* - * For other cases, the error will be returned by intent. - * and client will retrieve the result from intent. - */ - /* - * FIXME: when open lock is finished, that should be - * checked here. - */ - if (lustre_handle_is_used(&lhc->mlh_reg_lh)) { - LASSERTF(rc == 0, "Error occurred but lock handle " - "is still in use, rc = %d\n", rc); - rep->lock_policy_res2 = 0; - rc = mdt_intent_lock_replace(info, lockp, lhc, flags); - RETURN(rc); - } else { - lhc->mlh_reg_lh.cookie = 0ull; - RETURN(ELDLM_LOCK_ABORTED); - } } + /* + * For other cases, the error will be returned by intent, and client + * will retrieve the result from intent. + */ + RETURN(ELDLM_LOCK_ABORTED); } -static int mdt_intent_code(long itcode) +static int mdt_intent_code(enum ldlm_intent_flags itcode) { - int rc; + int rc; - switch(itcode) { - case IT_OPEN: - rc = MDT_IT_OPEN; - break; - case IT_OPEN|IT_CREAT: - rc = MDT_IT_OCREAT; - break; - case IT_CREAT: - rc = MDT_IT_CREATE; - break; - case IT_READDIR: - rc = MDT_IT_READDIR; - break; - case IT_GETATTR: - rc = MDT_IT_GETATTR; - break; - case IT_LOOKUP: - rc = MDT_IT_LOOKUP; - break; - case IT_UNLINK: - rc = MDT_IT_UNLINK; - break; - case IT_TRUNC: - rc = MDT_IT_TRUNC; - break; - case IT_GETXATTR: - rc = MDT_IT_GETXATTR; - break; - case IT_LAYOUT: - rc = MDT_IT_LAYOUT; - break; + switch (itcode) { + case IT_OPEN: + rc = MDT_IT_OPEN; + break; + case IT_OPEN|IT_CREAT: + rc = MDT_IT_OCREAT; + break; + case IT_CREAT: + rc = MDT_IT_CREATE; + break; + case IT_READDIR: + rc = MDT_IT_READDIR; + break; + case IT_GETATTR: + rc = MDT_IT_GETATTR; + break; + case IT_LOOKUP: + rc = MDT_IT_LOOKUP; + break; + case IT_UNLINK: + rc = MDT_IT_UNLINK; + break; + case IT_TRUNC: + rc = MDT_IT_TRUNC; + break; + case IT_GETXATTR: + rc = MDT_IT_GETXATTR; + break; + case IT_LAYOUT: + rc = MDT_IT_LAYOUT; + break; case IT_QUOTA_DQACQ: case IT_QUOTA_CONN: rc = MDT_IT_QUOTA; break; - default: - CERROR("Unknown intent opcode: %ld\n", itcode); - rc = -EINVAL; - break; - } - return rc; + default: + CERROR("Unknown intent opcode: 0x%08x\n", itcode); + rc = -EINVAL; + break; + } + return rc; } -static int mdt_intent_opc(long itopc, struct mdt_thread_info *info, +static int mdt_intent_opc(enum ldlm_intent_flags itopc, + struct mdt_thread_info *info, struct ldlm_lock **lockp, __u64 flags) { - struct req_capsule *pill; - struct mdt_it_flavor *flv; - int opc; - int rc; - ENTRY; - - opc = mdt_intent_code(itopc); - if (opc < 0) - RETURN(-EINVAL); + struct req_capsule *pill = info->mti_pill; + struct ptlrpc_request *req = mdt_info_req(info); + struct mdt_it_flavor *flv; + int opc; + int rc; + ENTRY; - pill = info->mti_pill; + opc = mdt_intent_code(itopc); + if (opc < 0) + RETURN(-EINVAL); if (opc == MDT_IT_QUOTA) { struct lu_device *qmt = info->mti_mdt->mdt_qmt_dev; @@ -3469,39 +3629,41 @@ static int mdt_intent_opc(long itopc, struct mdt_thread_info *info, RETURN(rc); } - flv = &mdt_it_flavor[opc]; - if (flv->it_fmt != NULL) - req_capsule_extend(pill, flv->it_fmt); + flv = &mdt_it_flavor[opc]; + if (flv->it_fmt != NULL) + req_capsule_extend(pill, flv->it_fmt); - rc = mdt_unpack_req_pack_rep(info, flv->it_flags); - if (rc == 0) { - struct ptlrpc_request *req = mdt_info_req(info); - if (flv->it_flags & MUTABOR && - exp_connect_flags(req->rq_export) & OBD_CONNECT_RDONLY) - RETURN(-EROFS); - } - if (rc == 0 && flv->it_act != NULL) { + rc = mdt_unpack_req_pack_rep(info, flv->it_flags); + if (rc < 0) + RETURN(rc); + + if (flv->it_flags & MUTABOR && + exp_connect_flags(req->rq_export) & OBD_CONNECT_RDONLY) + RETURN(-EROFS); + + if (flv->it_act != NULL) { struct ldlm_reply *rep; + OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_INTENT_DELAY, 10); + /* execute policy */ rc = flv->it_act(opc, info, lockp, flags); /* Check whether the reply has been packed successfully. */ - if (mdt_info_req(info)->rq_repmsg != NULL) { + if (req->rq_repmsg != NULL) { rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP); rep->lock_policy_res2 = ptlrpc_status_hton(rep->lock_policy_res2); } - } else { - rc = -EPROTO; } + RETURN(rc); } static int mdt_intent_policy(struct ldlm_namespace *ns, - struct ldlm_lock **lockp, void *req_cookie, - ldlm_mode_t mode, __u64 flags, void *data) + struct ldlm_lock **lockp, void *req_cookie, + enum ldlm_mode mode, __u64 flags, void *data) { struct tgt_session_info *tsi; struct mdt_thread_info *info; @@ -3604,20 +3766,26 @@ static int mdt_register_lwp_callback(void *data) LASSERT(mdt_seq_site(mdt)->ss_node_id != 0); - if (!likely(fld->lsf_new)) - RETURN(0); - rc = lu_env_init(&env, LCT_MD_THREAD); - if (rc) { + if (rc < 0) { CERROR("%s: cannot init env: rc = %d\n", mdt_obd_name(mdt), rc); RETURN(rc); } - rc = fld_update_from_controller(&env, fld); - if (rc != 0) { - CERROR("%s: cannot update controller: rc = %d\n", - mdt_obd_name(mdt), rc); + /* Allocate new sequence now to avoid creating local transaction + * in the normal transaction process */ + rc = seq_server_check_and_alloc_super(&env, + mdt_seq_site(mdt)->ss_server_seq); + if (rc < 0) GOTO(out, rc); + + if (fld->lsf_new) { + rc = fld_update_from_controller(&env, fld); + if (rc != 0) { + CERROR("%s: cannot update controller: rc = %d\n", + mdt_obd_name(mdt), rc); + GOTO(out, rc); + } } out: lu_env_fini(&env); @@ -3738,7 +3906,7 @@ static int mdt_seq_init(const struct lu_env *env, struct mdt_device *mdt) GOTO(out_seq_fini, rc); if (ss->ss_node_id != 0) - /* register controler export through lwp */ + /* register controller export through lwp */ rc = mdt_register_seq_exp(mdt); EXIT; @@ -3818,10 +3986,9 @@ static void mdt_stack_pre_fini(const struct lu_env *env, lustre_cfg_bufs_reset(bufs, mdt_obd_name(m)); lustre_cfg_bufs_set_string(bufs, 1, NULL); lcfg = lustre_cfg_new(LCFG_PRE_CLEANUP, bufs); - if (!lcfg) { - CERROR("%s: cannot alloc lcfg\n", mdt_obd_name(m)); - return; - } + if (lcfg == NULL) + RETURN_EXIT; + top->ld_ops->ldo_process_config(env, top, lcfg); lustre_cfg_free(lcfg); EXIT; @@ -3854,10 +4021,9 @@ static void mdt_stack_fini(const struct lu_env *env, strcat(flags, "A"); lustre_cfg_bufs_set_string(bufs, 1, flags); lcfg = lustre_cfg_new(LCFG_CLEANUP, bufs); - if (!lcfg) { - CERROR("Cannot alloc lcfg!\n"); - return; - } + if (lcfg == NULL) + RETURN_EXIT; + LASSERT(top); top->ld_ops->ldo_process_config(env, top, lcfg); lustre_cfg_free(lcfg); @@ -3981,8 +4147,8 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt, lustre_cfg_bufs_set_string(bufs, 3, lprof->lp_dt); lcfg = lustre_cfg_new(LCFG_ATTACH, bufs); - if (!lcfg) - GOTO(free_bufs, rc = -ENOMEM); + if (lcfg == NULL) + GOTO(put_profile, rc = -ENOMEM); rc = class_attach(lcfg); if (rc) @@ -4003,6 +4169,8 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt, lustre_cfg_bufs_set_string(bufs, 3, lprof->lp_dt); lcfg = lustre_cfg_new(LCFG_SETUP, bufs); + if (lcfg == NULL) + GOTO(class_detach, rc = -ENOMEM); rc = class_setup(obd, lcfg); if (rc) @@ -4040,6 +4208,8 @@ class_detach: class_detach(obd, lcfg); lcfg_cleanup: lustre_cfg_free(lcfg); +put_profile: + class_put_profile(lprof); free_bufs: OBD_FREE_PTR(bufs); cleanup_mem: @@ -4109,8 +4279,8 @@ static int mdt_quota_init(const struct lu_env *env, struct mdt_device *mdt, lustre_cfg_bufs_set_string(bufs, 3, lprof->lp_dt); lcfg = lustre_cfg_new(LCFG_ATTACH, bufs); - if (!lcfg) - GOTO(cleanup_mem, rc = -ENOMEM); + if (lcfg == NULL) + GOTO(put_profile, rc = -ENOMEM); rc = class_attach(lcfg); if (rc) @@ -4134,6 +4304,8 @@ static int mdt_quota_init(const struct lu_env *env, struct mdt_device *mdt, mdt->mdt_bottom->dd_lu_dev.ld_obd->obd_name); lcfg = lustre_cfg_new(LCFG_SETUP, bufs); + if (lcfg == NULL) + GOTO(class_detach, rc = -ENOMEM); rc = class_setup(obd, lcfg); if (rc) @@ -4170,6 +4342,8 @@ class_detach: class_detach(obd, lcfg); lcfg_cleanup: lustre_cfg_free(lcfg); +put_profile: + class_put_profile(lprof); cleanup_mem: if (bufs) OBD_FREE_PTR(bufs); @@ -4201,7 +4375,7 @@ static void mdt_quota_fini(const struct lu_env *env, struct mdt_device *mdt) /* mdt_getxattr() is used from mdt_intent_getxattr(), use this wrapper * for now. This will be removed along with converting rest of MDT code * to use tgt_session_info */ -int mdt_tgt_getxattr(struct tgt_session_info *tsi) +static int mdt_tgt_getxattr(struct tgt_session_info *tsi) { struct mdt_thread_info *info = tsi2mdt_info(tsi); int rc; @@ -4223,7 +4397,7 @@ TGT_RPC_HANDLER(MDS_FIRST_OPC, 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), @@ -4231,8 +4405,6 @@ TGT_MDT_HDL(HABEO_CORPUS, MDS_GETXATTR, mdt_tgt_getxattr), TGT_MDT_HDL(0 | HABEO_REFERO, MDS_STATFS, mdt_statfs), TGT_MDT_HDL(0 | MUTABOR, MDS_REINT, mdt_reint), TGT_MDT_HDL(HABEO_CORPUS, MDS_CLOSE, mdt_close), -TGT_MDT_HDL(HABEO_CORPUS, MDS_DONE_WRITING, - mdt_done_writing), TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_READPAGE, mdt_readpage), TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_SYNC, mdt_sync), TGT_MDT_HDL(0, MDS_QUOTACTL, mdt_quotactl), @@ -4333,9 +4505,9 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) stop.ls_flags = 0; next->md_ops->mdo_iocontrol(env, next, OBD_IOC_STOP_LFSCK, 0, &stop); + mdt_stack_pre_fini(env, m, md2lu_dev(m->mdt_child)); target_recovery_fini(obd); ping_evictor_stop(); - mdt_stack_pre_fini(env, m, md2lu_dev(m->mdt_child)); if (m->mdt_opts.mo_coordinator) mdt_hsm_cdt_stop(m); @@ -4344,6 +4516,11 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) mdt_llog_ctxt_unclone(env, m, LLOG_AGENT_ORIG_CTXT); mdt_llog_ctxt_unclone(env, m, LLOG_CHANGELOG_ORIG_CTXT); + + if (m->mdt_namespace != NULL) + ldlm_namespace_free_prior(m->mdt_namespace, NULL, + d->ld_obd->obd_force); + obd_exports_barrier(obd); obd_zombie_barrier(); @@ -4354,11 +4531,10 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) upcall_cache_cleanup(m->mdt_identity_cache); m->mdt_identity_cache = NULL; - if (m->mdt_namespace != NULL) { - ldlm_namespace_free(m->mdt_namespace, NULL, - d->ld_obd->obd_force); - d->ld_obd->obd_namespace = m->mdt_namespace = NULL; - } + if (m->mdt_namespace != NULL) { + ldlm_namespace_free_post(m->mdt_namespace); + d->ld_obd->obd_namespace = m->mdt_namespace = NULL; + } mdt_quota_fini(env, m); @@ -4367,10 +4543,6 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) mdt_seq_fini(env, m); mdt_fld_fini(env, m); - next->md_ops->mdo_init_capa_ctxt(env, next, 0, 0, 0, NULL); - cfs_timer_disarm(&m->mdt_ck_timer); - mdt_ck_thread_stop(m); - /* * Finish the stack */ @@ -4423,9 +4595,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, LASSERT(obd != NULL); m->mdt_max_mdsize = MAX_MD_SIZE; /* 4 stripes */ - - m->mdt_som_conf = 0; - + m->mdt_opts.mo_evict_tgt_nids = 1; m->mdt_opts.mo_cos = MDT_COS_DEFAULT; /* default is coordinator off, it is started through conf_param @@ -4442,12 +4612,11 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, LASSERT(num); node_id = simple_strtol(num, NULL, 10); obd->u.obt.obt_magic = OBT_MAGIC; + if (lsi->lsi_lmd != NULL && + lsi->lsi_lmd->lmd_flags & LMD_FLG_SKIP_LFSCK) + m->mdt_skip_lfsck = 1; } - spin_lock_init(&m->mdt_ioepoch_lock); - m->mdt_capa_timeout = CAPA_TIMEOUT; - m->mdt_capa_alg = CAPA_HMAC_ALG_SHA1; - m->mdt_ck_timeout = CAPA_KEY_TIMEOUT; m->mdt_squash.rsi_uid = 0; m->mdt_squash.rsi_gid = 0; INIT_LIST_HEAD(&m->mdt_squash.rsi_nosquash_nids); @@ -4457,6 +4626,9 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, 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. */ @@ -4518,8 +4690,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, /* set obd_namespace for compatibility with old code */ obd->obd_namespace = m->mdt_namespace; - cfs_timer_init(&m->mdt_ck_timer, mdt_ck_timer_callback, m); - rc = mdt_hsm_cdt_init(m); if (rc != 0) { CERROR("%s: error initializing coordinator, rc %d\n", @@ -4527,15 +4697,11 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, GOTO(err_free_ns, rc); } - rc = mdt_ck_thread_start(m); - if (rc) - GOTO(err_free_hsm, rc); - rc = tgt_init(env, &m->mdt_lut, obd, m->mdt_bottom, mdt_common_slice, OBD_FAIL_MDS_ALL_REQUEST_NET, OBD_FAIL_MDS_ALL_REPLY_NET); if (rc) - GOTO(err_capa, rc); + GOTO(err_free_hsm, rc); rc = mdt_fs_setup(env, m, obd, lsi); if (rc) @@ -4597,8 +4763,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, * when the whole stack is complete and ready * to serve the requests */ - mdt_init_capa_ctxt(env, m); - /* Reduce the initial timeout on an MDS because it doesn't need such * a long timeout as an OST does. Adaptive timeouts will adjust this * value appropriately. */ @@ -4616,9 +4780,6 @@ err_fs_cleanup: mdt_fs_cleanup(env, m); err_tgt: tgt_fini(env, &m->mdt_lut); -err_capa: - cfs_timer_disarm(&m->mdt_ck_timer); - mdt_ck_thread_stop(m); err_free_hsm: mdt_hsm_cdt_fini(m); err_free_ns: @@ -4643,6 +4804,7 @@ static struct cfg_interop_param mdt_interop_param[] = { { "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 } @@ -4739,7 +4901,7 @@ static struct lu_object *mdt_object_alloc(const struct lu_env *env, lu_object_init(o, h, d); lu_object_add_top(h, o); o->lo_ops = &mdt_obj_ops; - mutex_init(&mo->mot_ioepoch_mutex); + spin_lock_init(&mo->mot_write_lock); mutex_init(&mo->mot_lov_mutex); init_rwsem(&mo->mot_open_sem); RETURN(o); @@ -4790,13 +4952,13 @@ static void mdt_object_free(const struct lu_env *env, struct lu_object *o) } static int mdt_object_print(const struct lu_env *env, void *cookie, - lu_printer_t p, const struct lu_object *o) + lu_printer_t p, const struct lu_object *o) { - struct mdt_object *mdto = mdt_obj((struct lu_object *)o); - return (*p)(env, cookie, LUSTRE_MDT_NAME"-object@%p(ioepoch="LPU64" " - "flags="LPX64", epochcount=%d, writecount=%d)", - mdto, mdto->mot_ioepoch, mdto->mot_flags, - mdto->mot_ioepoch_count, mdto->mot_writecount); + struct mdt_object *mdto = mdt_obj((struct lu_object *)o); + + return (*p)(env, cookie, + LUSTRE_MDT_NAME"-object@%p(flags=%d, writecount=%d)", + mdto, mdto->mot_flags, mdto->mot_write_count); } static int mdt_prepare(const struct lu_env *env, @@ -4837,6 +4999,7 @@ static int mdt_prepare(const struct lu_env *env, } LASSERT(!test_bit(MDT_FL_CFGLOG, &mdt->mdt_state)); + target_recovery_init(&mdt->mdt_lut, tgt_request_handle); set_bit(MDT_FL_CFGLOG, &mdt->mdt_state); LASSERT(obd->obd_no_conn); @@ -4908,6 +5071,7 @@ static int mdt_connect_internal(struct obd_export *exp, LASSERT(data != NULL); data->ocd_connect_flags &= MDT_CONNECT_SUPPORTED; + data->ocd_connect_flags2 &= MDT_CONNECT_SUPPORTED2; data->ocd_ibits_known &= MDS_INODELOCK_FULL; if (!(data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) && @@ -4925,9 +5089,6 @@ static int mdt_connect_internal(struct obd_export *exp, if (!mdt->mdt_opts.mo_user_xattr) data->ocd_connect_flags &= ~OBD_CONNECT_XATTR; - if (!mdt->mdt_som_conf) - data->ocd_connect_flags &= ~OBD_CONNECT_SOM; - if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) { data->ocd_brw_size = min(data->ocd_brw_size, (__u32)MD_MAX_BRW_SIZE); @@ -4965,15 +5126,6 @@ static int mdt_connect_internal(struct obd_export *exp, return -EBADE; } - if (mdt->mdt_som_conf && - !(data->ocd_connect_flags & (OBD_CONNECT_LIGHTWEIGHT | - OBD_CONNECT_MDS_MDS | - OBD_CONNECT_SOM))) { - CWARN("%s: MDS has SOM enabled, but client does not support " - "it\n", mdt_obd_name(mdt)); - return -EBADE; - } - if (OCD_HAS_FLAG(data, PINGLESS)) { if (ptlrpc_pinger_suppress_pings()) { spin_lock(&exp->exp_obd->obd_dev_lock); @@ -4986,91 +5138,19 @@ static int mdt_connect_internal(struct obd_export *exp, data->ocd_max_easize = mdt->mdt_max_ea_size; - return 0; -} - -/* mds_connect copy */ -static int mdt_obd_connect(const struct lu_env *env, - struct obd_export **exp, struct obd_device *obd, - struct obd_uuid *cluuid, - struct obd_connect_data *data, - void *localdata) -{ - struct obd_export *lexp; - struct lustre_handle conn = { 0 }; - struct mdt_device *mdt; - int rc; - ENTRY; - - LASSERT(env != NULL); - if (!exp || !obd || !cluuid) - RETURN(-EINVAL); - - mdt = mdt_dev(obd->obd_lu_dev); - - /* - * first, check whether the stack is ready to handle requests - * XXX: probably not very appropriate method is used now - * at some point we should find a better one - */ - if (!test_bit(MDT_FL_SYNCED, &mdt->mdt_state) && data != NULL && - !(data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT)) { - rc = obd_get_info(env, mdt->mdt_child_exp, - sizeof(KEY_OSP_CONNECTED), - KEY_OSP_CONNECTED, NULL, NULL, NULL); - if (rc) - RETURN(-EAGAIN); - set_bit(MDT_FL_SYNCED, &mdt->mdt_state); + /* NB: Disregard the rule against updating + * exp_connect_data.ocd_connect_flags in this case, since + * tgt_client_new() needs to know if this is client supports + * multiple modify RPCs, and it is safe to expose this flag before + * connection processing completes. */ + if (data->ocd_connect_flags & OBD_CONNECT_MULTIMODRPCS) { + data->ocd_maxmodrpcs = max_mod_rpcs_per_client; + spin_lock(&exp->exp_lock); + *exp_connect_flags_ptr(exp) |= OBD_CONNECT_MULTIMODRPCS; + spin_unlock(&exp->exp_lock); } - rc = class_connect(&conn, obd, cluuid); - if (rc) - RETURN(rc); - - lexp = class_conn2export(&conn); - LASSERT(lexp != NULL); - - rc = mdt_connect_internal(lexp, mdt, data); - if (rc == 0) { - struct lsd_client_data *lcd = lexp->exp_target_data.ted_lcd; - - LASSERT(lcd); - memcpy(lcd->lcd_uuid, cluuid, sizeof lcd->lcd_uuid); - rc = tgt_client_new(env, lexp); - if (rc == 0) - mdt_export_stats_init(obd, lexp, localdata); - - /* For phase I, sync for cross-ref operation. */ - lexp->exp_keep_sync = 1; - } - - if (rc != 0) { - class_disconnect(lexp); - *exp = NULL; - } else { - *exp = lexp; - } - - RETURN(rc); -} - -static int mdt_obd_reconnect(const struct lu_env *env, - struct obd_export *exp, struct obd_device *obd, - struct obd_uuid *cluuid, - struct obd_connect_data *data, - void *localdata) -{ - int rc; - ENTRY; - - if (exp == NULL || obd == NULL || cluuid == NULL) - RETURN(-EINVAL); - - rc = mdt_connect_internal(exp, mdt_dev(obd->obd_lu_dev), data); - if (rc == 0) - mdt_export_stats_init(obd, exp, localdata); - - RETURN(rc); + return 0; } static int mdt_ctxt_add_dirty_flag(struct lu_env *env, @@ -5159,7 +5239,7 @@ static int mdt_export_cleanup(struct obd_export *exp) * archive request into a noop if it's not actually * dirty. */ - if (mfd->mfd_mode & (FMODE_WRITE|MDS_FMODE_TRUNC)) + if (mfd->mfd_mode & FMODE_WRITE) rc = mdt_ctxt_add_dirty_flag(&env, info, mfd); /* Don't unlink orphan on failover umount, LU-184 */ @@ -5180,6 +5260,18 @@ static int mdt_export_cleanup(struct obd_export *exp) 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; @@ -5188,15 +5280,134 @@ static int mdt_obd_disconnect(struct obd_export *exp) 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); rc = mdt_export_cleanup(exp); + nodemap_del_member(exp); class_export_put(exp); RETURN(rc); } +/* mds_connect copy */ +static int mdt_obd_connect(const struct lu_env *env, + struct obd_export **exp, struct obd_device *obd, + struct obd_uuid *cluuid, + struct obd_connect_data *data, + void *localdata) +{ + struct obd_export *lexp; + struct lustre_handle conn = { 0 }; + struct mdt_device *mdt; + int rc; + lnet_nid_t *client_nid = localdata; + ENTRY; + + LASSERT(env != NULL); + LASSERT(data != NULL); + + if (!exp || !obd || !cluuid) + RETURN(-EINVAL); + + 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 + * at some point we should find a better one + */ + if (!test_bit(MDT_FL_SYNCED, &mdt->mdt_state) && + !(data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT) && + !(data->ocd_connect_flags & OBD_CONNECT_MDS_MDS)) { + rc = obd_get_info(env, mdt->mdt_child_exp, + sizeof(KEY_OSP_CONNECTED), + KEY_OSP_CONNECTED, NULL, NULL); + if (rc) + RETURN(-EAGAIN); + set_bit(MDT_FL_SYNCED, &mdt->mdt_state); + } + + rc = class_connect(&conn, obd, cluuid); + if (rc) + RETURN(rc); + + lexp = class_conn2export(&conn); + LASSERT(lexp != NULL); + + rc = nodemap_add_member(*client_nid, lexp); + if (rc != 0 && rc != -EEXIST) + GOTO(out, rc); + + rc = mdt_connect_internal(lexp, mdt, data); + if (rc == 0) { + struct lsd_client_data *lcd = lexp->exp_target_data.ted_lcd; + + LASSERT(lcd); + memcpy(lcd->lcd_uuid, cluuid, sizeof lcd->lcd_uuid); + rc = tgt_client_new(env, lexp); + if (rc == 0) + mdt_export_stats_init(obd, lexp, localdata); + } +out: + if (rc != 0) { + class_disconnect(lexp); + nodemap_del_member(lexp); + *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); +} + +static int mdt_obd_reconnect(const struct lu_env *env, + struct obd_export *exp, struct obd_device *obd, + struct obd_uuid *cluuid, + struct obd_connect_data *data, + void *localdata) +{ + lnet_nid_t *client_nid = localdata; + int rc; + ENTRY; + + if (exp == NULL || obd == NULL || cluuid == NULL) + RETURN(-EINVAL); + + rc = nodemap_add_member(*client_nid, exp); + if (rc != 0 && rc != -EEXIST) + RETURN(rc); + + rc = mdt_connect_internal(exp, mdt_dev(obd->obd_lu_dev), data); + if (rc == 0) + mdt_export_stats_init(obd, exp, localdata); + else + nodemap_del_member(exp); + + RETURN(rc); +} + /* FIXME: Can we avoid using these two interfaces? */ static int mdt_init_export(struct obd_export *exp) { @@ -5258,25 +5469,6 @@ static int mdt_destroy_export(struct obd_export *exp) RETURN(0); } -/** The maximum depth that fid2path() will search. - * This is limited only because we want to store the fids for - * historical path lookup purposes. - */ -#define MAX_PATH_DEPTH 100 - -/** mdt_path() lookup structure. */ -struct path_lookup_info { - __u64 pli_recno; /**< history point */ - __u64 pli_currec; /**< current record */ - struct lu_fid pli_fid; - struct lu_fid pli_fids[MAX_PATH_DEPTH]; /**< path, in fids */ - struct mdt_object *pli_mdt_obj; - char *pli_path; /**< full path */ - int pli_pathlen; - int pli_linkno; /**< which hardlink to follow */ - int pli_fidcount; /**< number of \a pli_fids */ -}; - int mdt_links_read(struct mdt_thread_info *info, struct mdt_object *mdt_obj, struct linkea_data *ldata) { @@ -5308,8 +5500,23 @@ int mdt_links_read(struct mdt_thread_info *info, struct mdt_object *mdt_obj, return linkea_init(ldata); } +/** + * Given an MDT object, try to look up the full path to the object. + * Part of the MDT layer implementation of lfs fid2path. + * + * \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 + * \retval negative errno if there was a problem + */ static int mdt_path_current(struct mdt_thread_info *info, - struct path_lookup_info *pli) + struct mdt_object *obj, + struct getinfo_fid2path *fp, + struct lu_fid *root_fid) { struct mdt_device *mdt = info->mti_mdt; struct mdt_object *mdt_obj; @@ -5320,8 +5527,9 @@ static int mdt_path_current(struct mdt_thread_info *info, struct lu_buf *buf = &info->mti_big_buf; char *ptr; int reclen; - struct linkea_data ldata = { 0 }; + struct linkea_data ldata = { NULL }; int rc = 0; + bool first = true; ENTRY; /* temp buffer for path element, the buffer will be finally freed @@ -5331,18 +5539,18 @@ static int mdt_path_current(struct mdt_thread_info *info, RETURN(-ENOMEM); ldata.ld_buf = buf; - ptr = pli->pli_path + pli->pli_pathlen - 1; + ptr = fp->gf_u.gf_path + fp->gf_pathlen - 1; *ptr = 0; --ptr; - pli->pli_fidcount = 0; - pli->pli_fids[0] = *(struct lu_fid *)mdt_object_fid(pli->pli_mdt_obj); - *tmpfid = pli->pli_fids[0]; - /* 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, - &pli->pli_fids[pli->pli_fidcount])) { + *tmpfid = fp->gf_fid = *mdt_object_fid(obj); + + 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)); @@ -5368,18 +5576,17 @@ static int mdt_path_current(struct mdt_thread_info *info, linkea_entry_unpack(lee, &reclen, tmpname, tmpfid); /* If set, use link #linkno for path lookup, otherwise use link #0. Only do this for the final path element. */ - if (pli->pli_fidcount == 0 && - pli->pli_linkno < leh->leh_reccount) { + if (first && fp->gf_linkno < leh->leh_reccount) { int count; - for (count = 0; count < pli->pli_linkno; count++) { + for (count = 0; count < fp->gf_linkno; count++) { lee = (struct link_ea_entry *) ((char *)lee + reclen); linkea_entry_unpack(lee, &reclen, tmpname, tmpfid); } - if (pli->pli_linkno < leh->leh_reccount - 1) + if (fp->gf_linkno < leh->leh_reccount - 1) /* indicate to user there are more links */ - pli->pli_linkno++; + fp->gf_linkno++; } lmv_buf.lb_buf = info->mti_xattr_buf; @@ -5393,7 +5600,7 @@ static int mdt_path_current(struct mdt_thread_info *info, /* For slave stripes, get its master */ if (le32_to_cpu(lmm->lmv_magic) == LMV_MAGIC_STRIPE) { - pli->pli_fids[pli->pli_fidcount] = *tmpfid; + fp->gf_fid = *tmpfid; continue; } } else if (rc < 0 && rc != -ENODATA) { @@ -5404,73 +5611,87 @@ static int mdt_path_current(struct mdt_thread_info *info, /* Pack the name in the end of the buffer */ ptr -= tmpname->ln_namelen; - if (ptr - 1 <= pli->pli_path) + if (ptr - 1 <= fp->gf_u.gf_path) GOTO(out, rc = -EOVERFLOW); strncpy(ptr, tmpname->ln_name, tmpname->ln_namelen); *(--ptr) = '/'; - /* Store the parent fid for historic lookup */ - if (++pli->pli_fidcount >= MAX_PATH_DEPTH) - GOTO(out, rc = -EOVERFLOW); - pli->pli_fids[pli->pli_fidcount] = *tmpfid; + /* keep the last resolved fid to the client, so the + * client will build the left path on another MDT for + * remote object */ + fp->gf_fid = *tmpfid; + + first = false; } remote_out: ptr++; /* skip leading / */ - memmove(pli->pli_path, ptr, pli->pli_path + pli->pli_pathlen - ptr); + memmove(fp->gf_u.gf_path, ptr, + fp->gf_u.gf_path + fp->gf_pathlen - ptr); - EXIT; out: - return rc; + RETURN(rc); } -/* Returns the full path to this fid, as of changelog record recno. */ +/** + * Given an MDT object, use mdt_path_current to get the path. + * Essentially a wrapper to retry mdt_path_current a set number of times + * if -EAGAIN is returned (usually because an object is being moved). + * + * Part of the MDT layer implementation of lfs fid2path. + * + * \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 for arguments and to store path + * information + * + * \retval 0 Lookup successful, path information stored in fp + * \retval negative errno if there was a problem + */ static int mdt_path(struct mdt_thread_info *info, struct mdt_object *obj, - char *path, int pathlen, __u64 *recno, int *linkno, - struct lu_fid *fid) + struct getinfo_fid2path *fp, struct lu_fid *root_fid) { struct mdt_device *mdt = info->mti_mdt; - struct path_lookup_info *pli; int tries = 3; int rc = -EAGAIN; ENTRY; - if (pathlen < 3) + if (fp->gf_pathlen < 3) RETURN(-EOVERFLOW); - if (lu_fid_eq(&mdt->mdt_md_root_fid, mdt_object_fid(obj))) { - 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); } - OBD_ALLOC_PTR(pli); - if (pli == NULL) - RETURN(-ENOMEM); - - pli->pli_mdt_obj = obj; - pli->pli_recno = *recno; - pli->pli_path = path; - pli->pli_pathlen = pathlen; - pli->pli_linkno = *linkno; - /* Retry multiple times in case file is being moved */ while (tries-- && rc == -EAGAIN) - rc = mdt_path_current(info, pli); - - /* return the last resolved fids to the client, so the client will - * build the left path on another MDT for remote object */ - *fid = pli->pli_fids[pli->pli_fidcount]; - - *recno = pli->pli_currec; - /* Return next link index to caller */ - *linkno = pli->pli_linkno; - - OBD_FREE_PTR(pli); + rc = mdt_path_current(info, obj, fp, root_fid); RETURN(rc); } +/** + * Get the full path of the provided FID, as of changelog record recno. + * + * This checks sanity and looks up object for user provided FID + * before calling the actual path lookup code. + * + * Part of the MDT layer implementation of lfs fid2path. + * + * \param[in] info Per-thread common data shared by mdt level handlers. + * \param[in,out] fp User-provided struct for arguments and to store path + * information + * + * \retval 0 Lookup successful, path information and recno stored in fp + * \retval -ENOENT, object does not exist + * \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; @@ -5492,10 +5713,11 @@ static int mdt_fid2path(struct mdt_thread_info *info, } obj = mdt_object_find(info->mti_env, mdt, &fp->gf_fid); - if (obj == NULL || IS_ERR(obj)) { - CDEBUG(D_IOCTL, "no object "DFID": %ld\n", PFID(&fp->gf_fid), - PTR_ERR(obj)); - RETURN(-EINVAL); + if (IS_ERR(obj)) { + rc = PTR_ERR(obj); + CDEBUG(D_IOCTL, "cannot find "DFID": rc = %d\n", + PFID(&fp->gf_fid), rc); + RETURN(rc); } if (mdt_object_remote(obj)) @@ -5512,21 +5734,22 @@ static int mdt_fid2path(struct mdt_thread_info *info, RETURN(rc); } - rc = mdt_path(info, obj, fp->gf_path, fp->gf_pathlen, &fp->gf_recno, - &fp->gf_linkno, &fp->gf_fid); + 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)); @@ -5539,7 +5762,18 @@ static int mdt_rpc_fid2path(struct mdt_thread_info *info, void *key, 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); } @@ -5584,7 +5818,7 @@ int mdt_get_info(struct tgt_session_info *tsi) 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; @@ -5665,7 +5899,7 @@ static int mdt_ioc_version_get(struct mdt_thread_info *mti, void *karg) /* 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; @@ -5679,15 +5913,18 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len, 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; target_stop_recovery_thread(obd); rc = 0; break; @@ -5727,6 +5964,14 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len, 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); @@ -5760,18 +6005,21 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len, static int mdt_postrecov(const struct lu_env *env, struct mdt_device *mdt) { struct lu_device *ld = md2lu_dev(mdt->mdt_child); - struct lfsck_start_param lsp; int rc; ENTRY; - lsp.lsp_start = NULL; - lsp.lsp_index_valid = 0; - rc = mdt->mdt_child->md_ops->mdo_iocontrol(env, mdt->mdt_child, - OBD_IOC_START_LFSCK, - 0, &lsp); - if (rc != 0 && rc != -EALREADY) - CWARN("%s: auto trigger paused LFSCK failed: rc = %d\n", - mdt_obd_name(mdt), rc); + if (!mdt->mdt_skip_lfsck) { + struct lfsck_start_param lsp; + + lsp.lsp_start = NULL; + lsp.lsp_index_valid = 0; + rc = mdt->mdt_child->md_ops->mdo_iocontrol(env, mdt->mdt_child, + OBD_IOC_START_LFSCK, + 0, &lsp); + if (rc != 0 && rc != -EALREADY) + CWARN("%s: auto trigger paused LFSCK failed: rc = %d\n", + mdt_obd_name(mdt), rc); + } rc = ld->ld_ops->ldo_recovery_complete(env, ld); RETURN(rc); @@ -5925,7 +6173,7 @@ static struct lu_device_type mdt_device_type = { .ldt_ctx_tags = LCT_MD_THREAD }; -static int __init mdt_mod_init(void) +static int __init mdt_init(void) { int rc; @@ -5954,15 +6202,17 @@ mds_fini: return rc; } -static void __exit mdt_mod_exit(void) +static void __exit mdt_exit(void) { class_unregister_type(LUSTRE_MDT_NAME); mds_mod_exit(); lu_kmem_fini(mdt_caches); } -MODULE_AUTHOR("Sun Microsystems, Inc. "); +MODULE_AUTHOR("OpenSFS, Inc. "); MODULE_DESCRIPTION("Lustre Metadata Target ("LUSTRE_MDT_NAME")"); +MODULE_VERSION(LUSTRE_VERSION_STRING); MODULE_LICENSE("GPL"); -cfs_module(mdt, LUSTRE_VERSION_STRING, mdt_mod_init, mdt_mod_exit); +module_init(mdt_init); +module_exit(mdt_exit);