X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_handler.c;h=66b0eb4f4fbab9bc06654a0803943215f308077c;hp=17b76054f7f7042a3673610262c77d1b04935fcc;hb=a1ef13bca025e09e10ce2d6e007013d0e187e6eb;hpb=77eea1985bb1655e58c8b7df00703b4f08b58ec7 diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c index 17b7605..66b0eb4 100644 --- a/lustre/mdt/mdt_handler.c +++ b/lustre/mdt/mdt_handler.c @@ -53,6 +53,7 @@ * struct OBD_{ALLOC,FREE}*() */ #include +#include /* struct ptlrpc_request */ #include /* struct obd_export */ @@ -67,7 +68,6 @@ #include #include #include -#include #include mdl_mode_t mdt_mdl_lock_modes[] = { @@ -263,7 +263,7 @@ static void mdt_lock_pdo_mode(struct mdt_thread_info *info, struct mdt_object *o EXIT; } -int mdt_getstatus(struct tgt_session_info *tsi) +static int mdt_getstatus(struct tgt_session_info *tsi) { struct mdt_thread_info *info = tsi2mdt_info(tsi); struct mdt_device *mdt = info->mti_mdt; @@ -280,15 +280,15 @@ int mdt_getstatus(struct tgt_session_info *tsi) GOTO(out, rc = err_serious(-ENOMEM)); repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); - repbody->fid1 = mdt->mdt_md_root_fid; - repbody->valid |= OBD_MD_FLID; + repbody->mbo_fid1 = mdt->mdt_md_root_fid; + repbody->mbo_valid |= OBD_MD_FLID; 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; - root = mdt_object_find(info->mti_env, mdt, &repbody->fid1); + root = mdt_object_find(info->mti_env, mdt, &repbody->mbo_fid1); if (IS_ERR(root)) GOTO(out, rc = PTR_ERR(root)); @@ -299,7 +299,7 @@ int mdt_getstatus(struct tgt_session_info *tsi) 0); mdt_object_put(info->mti_env, root); if (rc == 0) - repbody->valid |= OBD_MD_FLMDSCAPA; + repbody->mbo_valid |= OBD_MD_FLMDSCAPA; } EXIT; out: @@ -307,7 +307,7 @@ out: return rc; } -int mdt_statfs(struct tgt_session_info *tsi) +static int mdt_statfs(struct tgt_session_info *tsi) { struct ptlrpc_request *req = tgt_ses_req(tsi); struct mdt_thread_info *info = tsi2mdt_info(tsi); @@ -383,9 +383,9 @@ static void mdt_pack_size2body(struct mdt_thread_info *info, !(ma->ma_valid & MA_SOM)) return; - b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; - b->size = ma->ma_som->msd_size; - b->blocks = ma->ma_som->msd_blocks; + 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; } void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b, @@ -395,66 +395,68 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b, LASSERT(ma->ma_valid & MA_INODE); - b->atime = attr->la_atime; - b->mtime = attr->la_mtime; - b->ctime = attr->la_ctime; - b->mode = attr->la_mode; - b->size = attr->la_size; - b->blocks = attr->la_blocks; - b->uid = attr->la_uid; - b->gid = attr->la_gid; - b->flags = attr->la_flags; - b->nlink = attr->la_nlink; - b->rdev = attr->la_rdev; - - /*XXX should pack the reply body according to lu_valid*/ - b->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 (!S_ISREG(attr->la_mode)) { - b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS | OBD_MD_FLRDEV; + 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; + + /* 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 (!S_ISREG(attr->la_mode)) { + b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS | OBD_MD_FLRDEV; } else if (ma->ma_need & MA_LOV && !(ma->ma_valid & MA_LOV)) { - /* means no objects are allocated on osts. */ - LASSERT(!(ma->ma_valid & MA_LOV)); - /* just ignore blocks occupied by extend attributes on MDS */ - b->blocks = 0; - /* if no object is allocated on osts, the size on mds is valid. b=22272 */ - b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; + /* means no objects are allocated on osts. */ + LASSERT(!(ma->ma_valid & MA_LOV)); + /* just ignore blocks occupied by extend attributes on MDS */ + b->mbo_blocks = 0; + /* if no object is allocated on osts, the size on mds is valid. + * b=22272 */ + b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; } else if ((ma->ma_valid & MA_LOV) && ma->ma_lmm != NULL && ma->ma_lmm->lmm_pattern & LOV_PATTERN_F_RELEASED) { /* A released file stores its size on MDS. */ /* But return 1 block for released file, unless tools like tar * will consider it fully sparse. (LU-3864) */ - if (unlikely(b->size == 0)) - b->blocks = 0; + if (unlikely(b->mbo_size == 0)) + b->mbo_blocks = 0; else - b->blocks = 1; - b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; + b->mbo_blocks = 1; + b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; } - if (fid) { - b->fid1 = *fid; - b->valid |= OBD_MD_FLID; - CDEBUG(D_INODE, DFID": nlink=%d, mode=%o, size="LPU64"\n", - PFID(fid), b->nlink, b->mode, b->size); - } + 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 (info) + mdt_body_reverse_idmap(info, b); - if (fid != NULL && (b->valid & OBD_MD_FLSIZE)) - CDEBUG(D_VFSTRACE, DFID": returning size %llu\n", - PFID(fid), (unsigned long long)b->size); + 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); } static inline int mdt_body_has_lov(const struct lu_attr *la, - const struct mdt_body *body) + const struct mdt_body *body) { - return ((S_ISREG(la->la_mode) && (body->valid & OBD_MD_FLEASIZE)) || - (S_ISDIR(la->la_mode) && (body->valid & OBD_MD_FLDIREA )) ); + return (S_ISREG(la->la_mode) && (body->mbo_valid & OBD_MD_FLEASIZE)) || + (S_ISDIR(la->la_mode) && (body->mbo_valid & OBD_MD_FLDIREA)); } void mdt_client_compatibility(struct mdt_thread_info *info) @@ -482,8 +484,40 @@ void mdt_client_compatibility(struct mdt_thread_info *info) EXIT; } -static int mdt_big_xattr_get(struct mdt_thread_info *info, struct mdt_object *o, - const char *name) +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; + + rc = mo_xattr_get(env, mdt_object_child(o), &LU_BUF_NULL, + XATTR_NAME_LOV); + + if (rc == -ENODATA) + rc = 0; + + if (rc < 0) + goto out; + + /* Is it a directory? Let's check for the LMV as well */ + if (S_ISDIR(lu_object_attr(&mdt_object_child(o)->mo_lu))) { + rc2 = mo_xattr_get(env, mdt_object_child(o), &LU_BUF_NULL, + XATTR_NAME_LMV); + + if (rc2 == -ENODATA) + rc2 = mo_xattr_get(env, mdt_object_child(o), + &LU_BUF_NULL, + XATTR_NAME_DEFAULT_LMV); + + if ((rc2 < 0 && rc2 != -ENODATA) || (rc2 > rc)) + rc = rc2; + } + +out: + return rc; +} + +int mdt_big_xattr_get(struct mdt_thread_info *info, struct mdt_object *o, + const char *name) { const struct lu_env *env = info->mti_env; int rc; @@ -521,8 +555,8 @@ static int mdt_big_xattr_get(struct mdt_thread_info *info, struct mdt_object *o, RETURN(rc); } -static int mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o, - struct md_attr *ma, const char *name) +int mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o, + struct md_attr *ma, const char *name) { struct md_object *next = mdt_object_child(o); struct lu_buf *buf = &info->mti_buf; @@ -546,10 +580,26 @@ static int mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o, rc = mo_xattr_get(info->mti_env, next, buf, name); if (rc > 0) { + +got: if (strcmp(name, XATTR_NAME_LOV) == 0) { - ma->ma_lmm_size = rc; - ma->ma_valid |= MA_LOV; + if (info->mti_big_lmm_used) + ma->ma_lmm = info->mti_big_lmm; + + /* NOT return LOV EA with hole to old client. */ + if (unlikely(le32_to_cpu(ma->ma_lmm->lmm_pattern) & + LOV_PATTERN_F_HOLE) && + !(exp_connect_flags(info->mti_exp) & + OBD_CONNECT_LFSCK)) { + return -EIO; + } else { + ma->ma_lmm_size = rc; + ma->ma_valid |= MA_LOV; + } } else if (strcmp(name, XATTR_NAME_LMV) == 0) { + if (info->mti_big_lmm_used) + ma->ma_lmv = info->mti_big_lmm; + ma->ma_lmv_size = rc; ma->ma_valid |= MA_LMV; } else if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) { @@ -557,6 +607,10 @@ static int mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o, ma->ma_valid |= MA_LMV_DEF; } + /* Update mdt_max_mdsize so all clients will be aware that */ + if (info->mti_mdt->mdt_max_mdsize < rc) + info->mti_mdt->mdt_max_mdsize = rc; + rc = 0; } else if (rc == -ENODATA) { /* no LOV EA */ @@ -569,31 +623,15 @@ static int mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o, rc = mdt_big_xattr_get(info, o, name); if (rc > 0) { info->mti_big_lmm_used = 1; - if (!strcmp(name, XATTR_NAME_LOV)) { - ma->ma_valid |= MA_LOV; - ma->ma_lmm = info->mti_big_lmm; - ma->ma_lmm_size = rc; - } else if (!strcmp(name, XATTR_NAME_LMV)) { - ma->ma_valid |= MA_LMV; - ma->ma_lmv = info->mti_big_lmm; - ma->ma_lmv_size = rc; - } else { - return -EINVAL; - } - - /* update mdt_max_mdsize so all clients - * will be aware about that */ - if (info->mti_mdt->mdt_max_mdsize < rc) - info->mti_mdt->mdt_max_mdsize = rc; - rc = 0; + goto got; } } return rc; } -int mdt_attr_get_pfid(struct mdt_thread_info *info, - struct mdt_object *o, struct lu_fid *pfid) +static int mdt_attr_get_pfid(struct mdt_thread_info *info, + struct mdt_object *o, struct lu_fid *pfid) { struct lu_buf *buf = &info->mti_buf; struct link_ea_header *leh; @@ -759,22 +797,24 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, if (mdt_object_remote(o)) { /* This object is located on remote node.*/ - /* Return -EIO for old client */ + /* Return -ENOTSUPP for old client */ if (!mdt_is_dne_client(req->rq_export)) - GOTO(out, rc = -EIO); + GOTO(out, rc = -ENOTSUPP); - repbody->fid1 = *mdt_object_fid(o); - repbody->valid = OBD_MD_FLID | OBD_MD_MDS; + repbody->mbo_fid1 = *mdt_object_fid(o); + repbody->mbo_valid = OBD_MD_FLID | OBD_MD_MDS; GOTO(out, rc = 0); } - buffer->lb_len = reqbody->eadatasize; - if (buffer->lb_len > 0) { + if (reqbody->mbo_eadatasize > 0) { buffer->lb_buf = req_capsule_server_get(pill, &RMF_MDT_MD); if (buffer->lb_buf == NULL) GOTO(out, rc = -EPROTO); + buffer->lb_len = req_capsule_get_size(pill, &RMF_MDT_MD, + RCL_SERVER); } else { buffer->lb_buf = NULL; + buffer->lb_len = 0; ma_need &= ~(MA_LOV | MA_LMV); CDEBUG(D_INFO, "%s: RPC from %s: does not need LOVEA.\n", mdt_obd_name(info->mti_mdt), @@ -783,15 +823,15 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, /* If it is dir object and client require MEA, then we got MEA */ if (S_ISDIR(lu_object_attr(&next->mo_lu)) && - (reqbody->valid & (OBD_MD_MEA | OBD_MD_DEFAULT_MEA))) { + (reqbody->mbo_valid & (OBD_MD_MEA | OBD_MD_DEFAULT_MEA))) { /* Assumption: MDT_MD size is enough for lmv size. */ ma->ma_lmv = buffer->lb_buf; ma->ma_lmv_size = buffer->lb_len; ma->ma_need = MA_INODE; if (ma->ma_lmv_size > 0) { - if (reqbody->valid & OBD_MD_MEA) + if (reqbody->mbo_valid & OBD_MD_MEA) ma->ma_need |= MA_LMV; - else if (reqbody->valid & OBD_MD_DEFAULT_MEA) + else if (reqbody->mbo_valid & OBD_MD_DEFAULT_MEA) ma->ma_need |= MA_LMV_DEF; } } else { @@ -803,7 +843,7 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, } if (S_ISDIR(lu_object_attr(&next->mo_lu)) && - reqbody->valid & OBD_MD_FLDIREA && + reqbody->mbo_valid & OBD_MD_FLDIREA && lustre_msg_get_opc(req->rq_reqmsg) == MDS_GETATTR) { /* get default stripe info for this dir. */ ma->ma_need |= MA_LOV_DEF; @@ -823,8 +863,8 @@ 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->t_state = MS_RESTORE; - repbody->valid |= OBD_MD_TSTATE; + repbody->mbo_t_state = MS_RESTORE; + repbody->mbo_valid |= OBD_MD_TSTATE; } is_root = lu_fid_eq(mdt_object_fid(o), &info->mti_mdt->mdt_md_root_fid); @@ -861,30 +901,38 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, if (mdt_body_has_lov(la, reqbody)) { if (ma->ma_valid & MA_LOV) { LASSERT(ma->ma_lmm_size); - mdt_dump_lmm(D_INFO, ma->ma_lmm); - repbody->eadatasize = ma->ma_lmm_size; - if (S_ISDIR(la->la_mode)) - repbody->valid |= OBD_MD_FLDIREA; - else - repbody->valid |= OBD_MD_FLEASIZE; + repbody->mbo_eadatasize = ma->ma_lmm_size; + if (S_ISDIR(la->la_mode)) + repbody->mbo_valid |= OBD_MD_FLDIREA; + else + repbody->mbo_valid |= OBD_MD_FLEASIZE; + mdt_dump_lmm(D_INFO, ma->ma_lmm, repbody->mbo_valid); } if (ma->ma_valid & MA_LMV) { + /* Return -ENOTSUPP for old client */ + if (!mdt_is_striped_client(req->rq_export)) + RETURN(-ENOTSUPP); + LASSERT(S_ISDIR(la->la_mode)); mdt_dump_lmv(D_INFO, ma->ma_lmv); - repbody->eadatasize = ma->ma_lmv_size; - repbody->valid |= (OBD_MD_FLDIREA|OBD_MD_MEA); + repbody->mbo_eadatasize = ma->ma_lmv_size; + repbody->mbo_valid |= (OBD_MD_FLDIREA|OBD_MD_MEA); } if (ma->ma_valid & MA_LMV_DEF) { + /* Return -ENOTSUPP for old client */ + if (!mdt_is_striped_client(req->rq_export)) + RETURN(-ENOTSUPP); LASSERT(S_ISDIR(la->la_mode)); - repbody->eadatasize = ma->ma_lmv_size; - repbody->valid |= (OBD_MD_FLDIREA|OBD_MD_DEFAULT_MEA); + repbody->mbo_eadatasize = ma->ma_lmv_size; + repbody->mbo_valid |= (OBD_MD_FLDIREA | + OBD_MD_DEFAULT_MEA); } } else if (S_ISLNK(la->la_mode) && - reqbody->valid & OBD_MD_LINKNAME) { + reqbody->mbo_valid & OBD_MD_LINKNAME) { buffer->lb_buf = ma->ma_lmm; /* eadatasize from client includes NULL-terminator, so * there is no need to read it */ - buffer->lb_len = reqbody->eadatasize - 1; + buffer->lb_len = reqbody->mbo_eadatasize - 1; rc = mo_readlink(env, next, buffer); if (unlikely(rc <= 0)) { CERROR("%s: readlink failed for "DFID": rc = %d\n", @@ -896,16 +944,16 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READLINK_EPROTO)) rc -= 2; - repbody->valid |= OBD_MD_LINKNAME; + repbody->mbo_valid |= OBD_MD_LINKNAME; /* we need to report back size with NULL-terminator * because client expects that */ - repbody->eadatasize = rc + 1; - if (repbody->eadatasize != reqbody->eadatasize) + repbody->mbo_eadatasize = rc + 1; + if (repbody->mbo_eadatasize != reqbody->mbo_eadatasize) CDEBUG(D_INODE, "%s: Read shorter symlink %d " "on "DFID ", expected %d\n", mdt_obd_name(info->mti_mdt), rc, PFID(mdt_object_fid(o)), - reqbody->eadatasize - 1); + reqbody->mbo_eadatasize - 1); /* NULL terminate */ ((char *)ma->ma_lmm)[rc] = 0; @@ -919,33 +967,33 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, } } - if (reqbody->valid & OBD_MD_FLMODEASIZE) { - repbody->max_cookiesize = 0; - repbody->max_mdsize = info->mti_mdt->mdt_max_mdsize; - repbody->valid |= OBD_MD_FLMODEASIZE; - CDEBUG(D_INODE, "I am going to change the MAX_MD_SIZE & " - "MAX_COOKIE to : %d:%d\n", repbody->max_mdsize, - repbody->max_cookiesize); - } + 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); + } - if (exp_connect_rmtclient(info->mti_exp) && - reqbody->valid & OBD_MD_FLRMTPERM) { - void *buf = req_capsule_server_get(pill, &RMF_ACL); + if (exp_connect_rmtclient(info->mti_exp) && + reqbody->mbo_valid & OBD_MD_FLRMTPERM) { + void *buf = req_capsule_server_get(pill, &RMF_ACL); - /* mdt_getattr_lock only */ - rc = mdt_pack_remote_perm(info, o, buf); - if (rc) { - repbody->valid &= ~OBD_MD_FLRMTPERM; - repbody->aclsize = 0; - RETURN(rc); - } else { - repbody->valid |= OBD_MD_FLRMTPERM; - repbody->aclsize = sizeof(struct mdt_remote_perm); - } - } + /* mdt_getattr_lock only */ + rc = mdt_pack_remote_perm(info, o, buf); + if (rc) { + repbody->mbo_valid &= ~OBD_MD_FLRMTPERM; + repbody->mbo_aclsize = 0; + RETURN(rc); + } else { + repbody->mbo_valid |= OBD_MD_FLRMTPERM; + repbody->mbo_aclsize = sizeof(struct mdt_remote_perm); + } + } #ifdef CONFIG_FS_POSIX_ACL else if ((exp_connect_flags(req->rq_export) & OBD_CONNECT_ACL) && - (reqbody->valid & OBD_MD_FLACL)) { + (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); @@ -954,8 +1002,8 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, XATTR_NAME_ACL_ACCESS); if (rc < 0) { if (rc == -ENODATA) { - repbody->aclsize = 0; - repbody->valid |= OBD_MD_FLACL; + repbody->mbo_aclsize = 0; + repbody->mbo_valid |= OBD_MD_FLACL; rc = 0; } else if (rc == -EOPNOTSUPP) { rc = 0; @@ -966,15 +1014,15 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, PFID(mdt_object_fid(o)), rc); } } else { - repbody->aclsize = rc; - repbody->valid |= OBD_MD_FLACL; - rc = 0; - } - } - } + repbody->mbo_aclsize = rc; + repbody->mbo_valid |= OBD_MD_FLACL; + rc = 0; + } + } + } #endif - if (reqbody->valid & OBD_MD_FLMDSCAPA && + 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; @@ -985,7 +1033,7 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, rc = mo_capa_get(env, next, capa, 0); if (rc) RETURN(rc); - repbody->valid |= OBD_MD_FLMDSCAPA; + repbody->mbo_valid |= OBD_MD_FLMDSCAPA; } out: @@ -1023,25 +1071,24 @@ static int mdt_renew_capa(struct mdt_thread_info *info) *capa = *c; rc = mo_capa_get(info->mti_env, mdt_object_child(obj), capa, 1); if (rc == 0) - body->valid |= OBD_MD_FLOSSCAPA; + body->mbo_valid |= OBD_MD_FLOSSCAPA; RETURN(rc); } -int mdt_getattr(struct tgt_session_info *tsi) +static int mdt_getattr(struct tgt_session_info *tsi) { struct mdt_thread_info *info = tsi2mdt_info(tsi); struct mdt_object *obj = info->mti_object; struct req_capsule *pill = info->mti_pill; struct mdt_body *reqbody; struct mdt_body *repbody; - mode_t mode; int rc, rc2; ENTRY; reqbody = req_capsule_client_get(pill, &RMF_MDT_BODY); LASSERT(reqbody); - if (reqbody->valid & OBD_MD_FLOSSCAPA) { + if (reqbody->mbo_valid & OBD_MD_FLOSSCAPA) { rc = req_capsule_server_pack(pill); if (unlikely(rc)) RETURN(err_serious(rc)); @@ -1052,13 +1099,36 @@ int mdt_getattr(struct tgt_session_info *tsi) LASSERT(obj != NULL); LASSERT(lu_object_assert_exists(&obj->mot_obj)); - mode = lu_object_attr(&obj->mot_obj); + /* Unlike intent case where we need to pre-fill out buffers early on + * in intent policy for ldlm reasons, here we can have a much better + * guess at EA size by just reading it from disk. + * Exceptions are readdir and (missing) directory striping */ + /* Readlink */ + if (reqbody->mbo_valid & OBD_MD_LINKNAME) { + /* No easy way to know how long is the symlink, but it cannot + * be more than PATH_MAX, so we allocate +1 */ + rc = PATH_MAX + 1; + + /* A special case for fs ROOT: getattr there might fetch + * default EA for entire fs, not just for this dir! + */ + } else if (lu_fid_eq(mdt_object_fid(obj), + &info->mti_mdt->mdt_md_root_fid) && + (reqbody->mbo_valid & OBD_MD_FLDIREA) && + (lustre_msg_get_opc(mdt_info_req(info)->rq_reqmsg) == + MDS_GETATTR)) { + /* Should the default strping be bigger, mdt_fix_reply + * will reallocate */ + rc = DEF_REP_MD_SIZE; + } else { + /* Read the actual EA size from disk */ + rc = mdt_attr_get_eabuf_size(info, obj); + } + + if (rc < 0) + GOTO(out_shrink, rc); - /* old clients may not report needed easize, use max value then */ - req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, - reqbody->eadatasize == 0 ? - info->mti_mdt->mdt_max_mdsize : - reqbody->eadatasize); + req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, rc); rc = req_capsule_server_pack(pill); if (unlikely(rc != 0)) @@ -1066,25 +1136,25 @@ int mdt_getattr(struct tgt_session_info *tsi) repbody = req_capsule_server_get(pill, &RMF_MDT_BODY); LASSERT(repbody != NULL); - repbody->eadatasize = 0; - repbody->aclsize = 0; + repbody->mbo_eadatasize = 0; + repbody->mbo_aclsize = 0; - if (reqbody->valid & OBD_MD_FLRMTPERM) - rc = mdt_init_ucred(info, reqbody); - else - rc = mdt_check_ucred(info); - if (unlikely(rc)) - GOTO(out_shrink, rc); + if (reqbody->mbo_valid & OBD_MD_FLRMTPERM) + rc = mdt_init_ucred(info, reqbody); + else + rc = mdt_check_ucred(info); + if (unlikely(rc)) + GOTO(out_shrink, rc); - info->mti_cross_ref = !!(reqbody->valid & OBD_MD_FLCROSSREF); + 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->fid1, BYPASS_CAPA); - rc = mdt_getattr_internal(info, obj, 0); - if (reqbody->valid & OBD_MD_FLRMTPERM) + /* + * 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); EXIT; out_shrink: @@ -1097,36 +1167,7 @@ out: return rc; } -int mdt_is_subdir(struct tgt_session_info *tsi) -{ - struct mdt_thread_info *info = tsi2mdt_info(tsi); - struct mdt_object *o = info->mti_object; - struct req_capsule *pill = info->mti_pill; - const struct mdt_body *body = info->mti_body; - struct mdt_body *repbody; - int rc; - ENTRY; - - LASSERT(o != NULL); - - repbody = req_capsule_server_get(pill, &RMF_MDT_BODY); - - /* - * We save last checked parent fid to @repbody->fid1 for remote - * directory case. - */ - LASSERT(fid_is_sane(&body->fid2)); - LASSERT(mdt_object_exists(o) && !mdt_object_remote(o)); - rc = mdo_is_subdir(info->mti_env, mdt_object_child(o), - &body->fid2, &repbody->fid1); - if (rc == 0 || rc == -EREMOTE) - repbody->valid |= OBD_MD_FLID; - - mdt_thread_info_fini(info); - RETURN(rc); -} - -int mdt_swap_layouts(struct tgt_session_info *tsi) +static int mdt_swap_layouts(struct tgt_session_info *tsi) { struct mdt_thread_info *info; struct ptlrpc_request *req = tgt_ses_req(tsi); @@ -1148,26 +1189,30 @@ int mdt_swap_layouts(struct tgt_session_info *tsi) RETURN(-EOPNOTSUPP); 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->fid1, + 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->fid2, + mdt_set_capainfo(info, 1, &info->mti_body->mbo_fid2, req_capsule_client_get(info->mti_pill, &RMF_CAPA2)); o1 = info->mti_object; o = o2 = mdt_object_find(info->mti_env, info->mti_mdt, - &info->mti_body->fid2); + &info->mti_body->mbo_fid2); if (IS_ERR(o)) GOTO(out, rc = PTR_ERR(o)); if (mdt_object_remote(o) || !mdt_object_exists(o)) /* remote object */ GOTO(put, rc = -ENOENT); - rc = lu_fid_cmp(&info->mti_body->fid1, &info->mti_body->fid2); + rc = lu_fid_cmp(&info->mti_body->mbo_fid1, &info->mti_body->mbo_fid2); if (unlikely(rc == 0)) /* same file, you kidding me? no-op. */ GOTO(put, rc); @@ -1231,7 +1276,7 @@ static int mdt_raw_lookup(struct mdt_thread_info *info, int rc; ENTRY; - if (reqbody->valid != OBD_MD_FLID) + if (reqbody->mbo_valid != OBD_MD_FLID) RETURN(0); LASSERT(!info->mti_cross_ref); @@ -1253,8 +1298,8 @@ static int mdt_raw_lookup(struct mdt_thread_info *info, #endif if (rc == 0) { repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); - repbody->fid1 = *child_fid; - repbody->valid = OBD_MD_FLID; + repbody->mbo_fid1 = *child_fid; + repbody->mbo_valid = OBD_MD_FLID; } RETURN(1); } @@ -1279,88 +1324,31 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, struct lu_name *lname = NULL; struct mdt_lock_handle *lhp = NULL; struct ldlm_lock *lock; - struct ldlm_res_id *res_id; - int is_resent; - int ma_need = 0; - int rc; - - ENTRY; - - is_resent = lustre_handle_is_used(&lhc->mlh_reg_lh); - LASSERT(ergo(is_resent, - lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)); - - LASSERT(parent != NULL); - - lname = &info->mti_name; - mdt_name_unpack(info->mti_pill, &RMF_NAME, lname, MNF_FIX_ANON); - - if (!info->mti_cross_ref) { - /* - * XXX: Check for anonymous name is for getattr by fid - * (OBD_CONNECT_ATTRFID), otherwise do not allow empty name, - * that is the name must contain at least one character and - * the terminating '\0'. - */ - if (!lu_name_is_valid(lname)) { - reqbody = req_capsule_client_get(info->mti_pill, - &RMF_MDT_BODY); - if (unlikely(reqbody == NULL)) - RETURN(err_serious(-EFAULT)); - - if (unlikely(!fid_is_sane(&reqbody->fid2))) - RETURN(err_serious(-EINVAL)); - - CDEBUG(D_INODE, "getattr with lock for "DFID"/"DFID", " - "ldlm_rep = %p\n", - PFID(mdt_object_fid(parent)), - PFID(&reqbody->fid2), ldlm_rep); - } else { - CDEBUG(D_INODE, "getattr with lock for "DFID"/"DNAME", " - "ldlm_rep = %p\n", PFID(mdt_object_fid(parent)), - PNAME(lname), ldlm_rep); - } - } + bool is_resent; + bool try_layout; + int ma_need = 0; + int rc; + ENTRY; - mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_EXECD); + is_resent = lustre_handle_is_used(&lhc->mlh_reg_lh); + LASSERT(ergo(is_resent, + lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)); - 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"); - RETURN(-ESTALE); - } else if (!info->mti_cross_ref) { - LASSERTF(!mdt_object_remote(parent), - "Parent "DFID" is on remote server\n", - PFID(mdt_object_fid(parent))); - } + LASSERT(parent != NULL); - if (lu_name_is_valid(lname)) { - rc = mdt_raw_lookup(info, parent, lname, ldlm_rep); - if (rc != 0) { - if (rc > 0) - rc = 0; - RETURN(rc); - } - } + if (info->mti_cross_ref) { + /* Only getattr on the child. Parent is on another node. */ + mdt_set_disposition(info, ldlm_rep, + DISP_LOOKUP_EXECD | DISP_LOOKUP_POS); + child = parent; + CDEBUG(D_INODE, "partial getattr_name child_fid = "DFID", " + "ldlm_rep = %p\n", + PFID(mdt_object_fid(child)), ldlm_rep); - if (info->mti_cross_ref) { - /* Only getattr on the child. Parent is on another node. */ - mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS); - child = parent; - CDEBUG(D_INODE, "partial getattr_name child_fid = "DFID", " - "ldlm_rep=%p\n", PFID(mdt_object_fid(child)), ldlm_rep); - - if (is_resent) { - /* Do not take lock for resent case. */ - lock = ldlm_handle2lock(&lhc->mlh_reg_lh); - LASSERTF(lock != NULL, "Invalid lock handle "LPX64"\n", - lhc->mlh_reg_lh.cookie); - LASSERT(fid_res_name_eq(mdt_object_fid(child), - &lock->l_resource->lr_name)); - LDLM_LOCK_PUT(lock); - rc = 0; - } else { + rc = mdt_check_resent_lock(info, child, lhc); + if (rc < 0) { + RETURN(-EPROTO); + } else if (rc > 0) { mdt_lock_handle_init(lhc); mdt_lock_reg_init(lhc, LCK_PR); @@ -1374,27 +1362,74 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, rc = mdt_object_lock(info, child, lhc, child_bits, MDT_LOCAL_LOCK); + if (rc < 0) + RETURN(rc); } - if (rc == 0) { - /* Finally, we can get attr for child. */ - if (!mdt_object_exists(child)) { - LU_OBJECT_DEBUG(D_INFO, info->mti_env, - &child->mot_obj, - "remote object doesn't exist.\n"); - 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); - } + /* Finally, we can get attr for child. */ + if (!mdt_object_exists(child)) { + LU_OBJECT_DEBUG(D_INFO, info->mti_env, + &child->mot_obj, + "remote object doesn't exist.\n"); + 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); + RETURN(rc); } + lname = &info->mti_name; + mdt_name_unpack(info->mti_pill, &RMF_NAME, lname, MNF_FIX_ANON); + + if (lu_name_is_valid(lname)) { + CDEBUG(D_INODE, "getattr with lock for "DFID"/"DNAME", " + "ldlm_rep = %p\n", PFID(mdt_object_fid(parent)), + PNAME(lname), ldlm_rep); + } else { + reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY); + if (unlikely(reqbody == NULL)) + RETURN(err_serious(-EPROTO)); + + *child_fid = reqbody->mbo_fid2; + + if (unlikely(!fid_is_sane(child_fid))) + RETURN(err_serious(-EINVAL)); + + CDEBUG(D_INODE, "getattr with lock for "DFID"/"DFID", " + "ldlm_rep = %p\n", + PFID(mdt_object_fid(parent)), + PFID(&reqbody->mbo_fid2), ldlm_rep); + } + + mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_EXECD); + + 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"); + RETURN(-ESTALE); + } + + if (mdt_object_remote(parent)) { + CERROR("%s: parent "DFID" is on remote target\n", + mdt_obd_name(info->mti_mdt), + PFID(mdt_object_fid(parent))); + RETURN(-EIO); + } + if (lu_name_is_valid(lname)) { + rc = mdt_raw_lookup(info, parent, lname, ldlm_rep); + if (rc != 0) { + if (rc > 0) + rc = 0; + RETURN(rc); + } + /* step 1: lock parent only if parent is a directory */ if (S_ISDIR(lu_object_attr(&parent->mot_obj))) { lhp = &info->mti_lh[MDT_LH_PARENT]; @@ -1410,18 +1445,14 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, fid_zero(child_fid); rc = mdo_lookup(info->mti_env, next, lname, child_fid, &info->mti_spec); + if (rc == -ENOENT) + mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG); - if (rc != 0) { - if (rc == -ENOENT) - mdt_set_disposition(info, ldlm_rep, - DISP_LOOKUP_NEG); - GOTO(out_parent, rc); - } else - mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS); - } else { - *child_fid = reqbody->fid2; - mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS); - } + if (rc != 0) + GOTO(out_parent, rc); + } + + mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS); /* *step 3: find the child object by fid & lock it. @@ -1442,37 +1473,15 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, if (unlikely(IS_ERR(child))) GOTO(out_parent, rc = PTR_ERR(child)); - if (is_resent) { - /* Do not take lock for resent case. */ - lock = ldlm_handle2lock(&lhc->mlh_reg_lh); - LASSERTF(lock != NULL, "Invalid lock handle "LPX64"\n", - lhc->mlh_reg_lh.cookie); - res_id = &lock->l_resource->lr_name; - if (!fid_res_name_eq(mdt_object_fid(child), - &lock->l_resource->lr_name)) { - LASSERTF(fid_res_name_eq(mdt_object_fid(parent), - &lock->l_resource->lr_name), - "Lock res_id: "DLDLMRES", fid: "DFID"\n", - PLDLMRES(lock->l_resource), - PFID(mdt_object_fid(parent))); - CWARN("Although resent, but still not get child lock" - "parent:"DFID" child:"DFID"\n", - PFID(mdt_object_fid(parent)), - PFID(mdt_object_fid(child))); - lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT); - LDLM_LOCK_PUT(lock); - GOTO(relock, 0); - } - LDLM_LOCK_PUT(lock); - rc = 0; - } else { - bool try_layout = false; - -relock: + 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, @@ -1529,6 +1538,10 @@ relock: ma_need |= MA_LOV; } } else { + /* Do not enqueue the UPDATE lock from MDT(cross-MDT), + * 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); } @@ -1550,7 +1563,6 @@ relock: mdt_object_unlock(info, child, lhc, 1); } else if (lock) { /* Debugging code. */ - res_id = &lock->l_resource->lr_name; LDLM_DEBUG(lock, "Returning lock to client"); LASSERTF(fid_res_name_eq(mdt_object_fid(child), &lock->l_resource->lr_name), @@ -1573,7 +1585,7 @@ out_parent: } /* normal handler: should release the child lock */ -int mdt_getattr_name(struct tgt_session_info *tsi) +static int mdt_getattr_name(struct tgt_session_info *tsi) { struct mdt_thread_info *info = tsi2mdt_info(tsi); struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_CHILD]; @@ -1587,9 +1599,9 @@ int mdt_getattr_name(struct tgt_session_info *tsi) repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); LASSERT(repbody != NULL); - info->mti_cross_ref = !!(reqbody->valid & OBD_MD_FLCROSSREF); - repbody->eadatasize = 0; - repbody->aclsize = 0; + info->mti_cross_ref = !!(reqbody->mbo_valid & OBD_MD_FLCROSSREF); + repbody->mbo_eadatasize = 0; + repbody->mbo_aclsize = 0; rc = mdt_init_ucred(info, reqbody); if (unlikely(rc)) @@ -1614,7 +1626,7 @@ out_shrink: static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len, void *karg, void *uarg); -int mdt_set_info(struct tgt_session_info *tsi) +static int mdt_set_info(struct tgt_session_info *tsi) { struct ptlrpc_request *req = tgt_ses_req(tsi); char *key; @@ -1672,7 +1684,7 @@ int mdt_set_info(struct tgt_session_info *tsi) RETURN(rc); } -int mdt_readpage(struct tgt_session_info *tsi) +static int mdt_readpage(struct tgt_session_info *tsi) { struct mdt_thread_info *info = mdt_th_info(tsi->tsi_env); struct mdt_object *object = mdt_obj(tsi->tsi_corpus); @@ -1696,17 +1708,17 @@ int mdt_readpage(struct tgt_session_info *tsi) * reqbody->size contains offset of where to start to read and * reqbody->nlink contains number bytes to read. */ - rdpg->rp_hash = reqbody->size; - if (rdpg->rp_hash != reqbody->size) { - CERROR("Invalid hash: "LPX64" != "LPX64"\n", - rdpg->rp_hash, reqbody->size); - RETURN(-EFAULT); - } + rdpg->rp_hash = reqbody->mbo_size; + if (rdpg->rp_hash != reqbody->mbo_size) { + CERROR("Invalid hash: "LPX64" != "LPX64"\n", + rdpg->rp_hash, reqbody->mbo_size); + RETURN(-EFAULT); + } - rdpg->rp_attrs = reqbody->mode; + rdpg->rp_attrs = reqbody->mbo_mode; if (exp_connect_flags(tsi->tsi_exp) & OBD_CONNECT_64BITHASH) rdpg->rp_attrs |= LUDA_64BITHASH; - rdpg->rp_count = min_t(unsigned int, reqbody->nlink, + rdpg->rp_count = min_t(unsigned int, reqbody->mbo_nlink, exp_max_brw_size(tsi->tsi_exp)); rdpg->rp_npages = (rdpg->rp_count + PAGE_CACHE_SIZE - 1) >> PAGE_CACHE_SHIFT; @@ -1761,7 +1773,7 @@ static int mdt_reint_internal(struct mdt_thread_info *info, /* for replay (no_create) lmm is not needed, client has it already */ if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER)) req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, - info->mti_rr.rr_eadatalen); + DEF_REP_MD_SIZE); /* llog cookies are always 0, the field is kept for compatibility */ if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER)) @@ -1776,8 +1788,8 @@ static int mdt_reint_internal(struct mdt_thread_info *info, if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_SERVER)) { repbody = req_capsule_server_get(pill, &RMF_MDT_BODY); LASSERT(repbody); - repbody->eadatasize = 0; - repbody->aclsize = 0; + repbody->mbo_eadatasize = 0; + repbody->mbo_aclsize = 0; } OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_REINT_DELAY, 10); @@ -1796,7 +1808,8 @@ static int mdt_reint_internal(struct mdt_thread_info *info, GOTO(out_ucred, rc = err_serious(rc)); if (mdt_check_resent(info, mdt_reconstruct, lhc)) { - rc = lustre_msg_get_status(mdt_info_req(info)->rq_repmsg); + 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); } rc = mdt_reint_rec(info, lhc); @@ -1837,7 +1850,7 @@ static long mdt_reint_opcode(struct ptlrpc_request *req, return opc; } -int mdt_reint(struct tgt_session_info *tsi) +static int mdt_reint(struct tgt_session_info *tsi) { long opc; int rc; @@ -1849,7 +1862,8 @@ int mdt_reint(struct tgt_session_info *tsi) [REINT_RENAME] = &RQF_MDS_REINT_RENAME, [REINT_OPEN] = &RQF_MDS_REINT_OPEN, [REINT_SETXATTR] = &RQF_MDS_REINT_SETXATTR, - [REINT_RMENTRY] = &RQF_MDS_REINT_UNLINK + [REINT_RMENTRY] = &RQF_MDS_REINT_UNLINK, + [REINT_MIGRATE] = &RQF_MDS_REINT_RENAME }; ENTRY; @@ -1885,22 +1899,23 @@ static int mdt_device_sync(const struct lu_env *env, struct mdt_device *mdt) /* this should sync this object */ static int mdt_object_sync(struct mdt_thread_info *info) { - struct md_object *next; - int rc; - ENTRY; + struct md_object *next; + int rc; + ENTRY; - if (!mdt_object_exists(info->mti_object)) { - CWARN("Non existing object "DFID"!\n", - PFID(mdt_object_fid(info->mti_object))); - RETURN(-ESTALE); - } - next = mdt_object_child(info->mti_object); - rc = mo_object_sync(info->mti_env, next); + if (!mdt_object_exists(info->mti_object)) { + CWARN("%s: non existing object "DFID": rc = %d\n", + mdt_obd_name(info->mti_mdt), + PFID(mdt_object_fid(info->mti_object)), -ESTALE); + RETURN(-ESTALE); + } + next = mdt_object_child(info->mti_object); + rc = mo_object_sync(info->mti_env, next); - RETURN(rc); + RETURN(rc); } -int mdt_sync(struct tgt_session_info *tsi) +static int mdt_sync(struct tgt_session_info *tsi) { struct ptlrpc_request *req = tgt_ses_req(tsi); struct req_capsule *pill = tsi->tsi_pill; @@ -1912,7 +1927,7 @@ int mdt_sync(struct tgt_session_info *tsi) if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SYNC_PACK)) RETURN(err_serious(-ENOMEM)); - if (fid_seq(&tsi->tsi_mdt_body->fid1) == 0) { + if (fid_seq(&tsi->tsi_mdt_body->mbo_fid1) == 0) { rc = mdt_device_sync(tsi->tsi_env, mdt_exp2dev(tsi->tsi_exp)); } else { struct mdt_thread_info *info = tsi2mdt_info(tsi); @@ -1946,7 +1961,7 @@ int mdt_sync(struct tgt_session_info *tsi) * Handle quota control requests to consult current usage/limit, but also * to configure quota enforcement */ -int mdt_quotactl(struct tgt_session_info *tsi) +static int mdt_quotactl(struct tgt_session_info *tsi) { struct obd_export *exp = tsi->tsi_exp; struct req_capsule *pill = tsi->tsi_pill; @@ -2098,7 +2113,7 @@ static int mdt_llog_ctxt_unclone(const struct lu_env *env, /* * sec context handlers */ -int mdt_sec_ctx_handle(struct tgt_session_info *tsi) +static int mdt_sec_ctx_handle(struct tgt_session_info *tsi) { int rc; @@ -2120,7 +2135,7 @@ int mdt_sec_ctx_handle(struct tgt_session_info *tsi) /* * quota request handlers */ -int mdt_quota_dqacq(struct tgt_session_info *tsi) +static int mdt_quota_dqacq(struct tgt_session_info *tsi) { struct mdt_device *mdt = mdt_exp2dev(tsi->tsi_exp); struct lu_device *qmt = mdt->mdt_qmt_dev; @@ -2181,12 +2196,13 @@ struct mdt_object *mdt_object_find(const struct lu_env *env, static void mdt_device_commit_async(const struct lu_env *env, struct mdt_device *mdt) { - struct dt_device *dt = mdt->mdt_bottom; - int rc; + struct dt_device *dt = mdt->mdt_bottom; + int rc; - rc = dt->dd_ops->dt_commit_async(env, dt); - if (unlikely(rc != 0)) - CWARN("async commit start failed with rc = %d", rc); + 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); } /** @@ -2261,10 +2277,11 @@ int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, if (lock->l_req_mode == LCK_COS && lock->l_blocking_lock != NULL) { struct lu_env env; - rc = lu_env_init(&env, LCT_LOCAL); - if (unlikely(rc != 0)) - CWARN("lu_env initialization failed with rc = %d," - "cannot start asynchronous commit\n", rc); + rc = lu_env_init(&env, LCT_LOCAL); + if (unlikely(rc != 0)) + CWARN("%s: lu_env initialization failed, cannot " + "start asynchronous commit: rc = %d\n", + obd->obd_name, rc); else mdt_device_commit_async(&env, mdt); lu_env_fini(&env); @@ -2297,18 +2314,51 @@ int mdt_remote_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, RETURN(0); } +int mdt_check_resent_lock(struct mdt_thread_info *info, + struct mdt_object *mo, + struct mdt_lock_handle *lhc) +{ + /* the lock might already be gotten in ldlm_handle_enqueue() */ + if (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 (!fid_res_name_eq(mdt_object_fid(mo), + &lock->l_resource->lr_name)) { + CWARN("%s: Although resent, but still not " + "get child lock:"DFID"\n", + info->mti_exp->exp_obd->obd_name, + PFID(mdt_object_fid(mo))); + LDLM_LOCK_PUT(lock); + RETURN(-EPROTO); + } + LDLM_LOCK_PUT(lock); + return 0; + } + return 1; +} + int mdt_remote_object_lock(struct mdt_thread_info *mti, - struct mdt_object *o, struct lustre_handle *lh, - ldlm_mode_t mode, __u64 ibits) + struct mdt_object *o, const struct lu_fid *fid, + struct lustre_handle *lh, ldlm_mode_t mode, + __u64 ibits) { struct ldlm_enqueue_info *einfo = &mti->mti_einfo; ldlm_policy_data_t *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); + LASSERT(ibits == MDS_INODELOCK_UPDATE); + + fid_build_reg_res_name(fid, res_id); memset(einfo, 0, sizeof(*einfo)); einfo->ei_type = LDLM_IBITS; @@ -2316,6 +2366,7 @@ int mdt_remote_object_lock(struct mdt_thread_info *mti, einfo->ei_cb_bl = mdt_remote_blocking_ast; einfo->ei_cb_cp = ldlm_completion_ast; einfo->ei_enq_slave = 0; + einfo->ei_res_id = res_id; memset(policy, 0, sizeof(*policy)); policy->l_inodebits.bits = ibits; @@ -2325,9 +2376,10 @@ int mdt_remote_object_lock(struct mdt_thread_info *mti, RETURN(rc); } -static int mdt_object_lock0(struct mdt_thread_info *info, struct mdt_object *o, - struct mdt_lock_handle *lh, __u64 ibits, - bool nonblock, int locality) +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) { struct ldlm_namespace *ns = info->mti_mdt->mdt_namespace; ldlm_policy_data_t *policy = &info->mti_policy; @@ -2341,23 +2393,9 @@ static int mdt_object_lock0(struct mdt_thread_info *info, struct mdt_object *o, LASSERT(lh->mlh_reg_mode != LCK_MINMODE); LASSERT(lh->mlh_type != MDT_NUL_LOCK); - if (mdt_object_remote(o)) { - if (locality == MDT_CROSS_LOCK) { - ibits &= ~(MDS_INODELOCK_UPDATE | MDS_INODELOCK_PERM | - MDS_INODELOCK_LAYOUT); - ibits |= MDS_INODELOCK_LOOKUP; - } else { - LASSERTF(!(ibits & - (MDS_INODELOCK_UPDATE | MDS_INODELOCK_PERM | - MDS_INODELOCK_LAYOUT)), - "%s: wrong bit "LPX64" for remote obj "DFID"\n", - mdt_obd_name(info->mti_mdt), ibits, - PFID(mdt_object_fid(o))); - LASSERT(ibits & MDS_INODELOCK_LOOKUP); - } - /* No PDO lock on remote object */ - LASSERT(lh->mlh_type != MDT_PDO_LOCK); - } + /* Only enqueue LOOKUP lock for remote object */ + if (mdt_object_remote(o)) + LASSERT(ibits == MDS_INODELOCK_LOOKUP); if (lh->mlh_type == MDT_PDO_LOCK) { /* check for exists after object is locked */ @@ -2392,9 +2430,10 @@ static int mdt_object_lock0(struct mdt_thread_info *info, struct mdt_object *o, * want it slowed down due to possible cancels. */ policy->l_inodebits.bits = MDS_INODELOCK_UPDATE; - rc = mdt_fid_lock(ns, &lh->mlh_pdo_lh, lh->mlh_pdo_mode, - policy, res_id, dlmflags, - &info->mti_exp->exp_handle.h_cookie); + rc = mdt_fid_lock(ns, &lh->mlh_pdo_lh, lh->mlh_pdo_mode, + policy, res_id, dlmflags, + info->mti_exp == NULL ? NULL : + &info->mti_exp->exp_handle.h_cookie); if (unlikely(rc)) RETURN(rc); } @@ -2413,9 +2452,10 @@ static int mdt_object_lock0(struct mdt_thread_info *info, struct mdt_object *o, * going to be sent to client. If it is - mdt_intent_policy() path will * fix it up and turn FL_LOCAL flag off. */ - rc = mdt_fid_lock(ns, &lh->mlh_reg_lh, lh->mlh_reg_mode, policy, - res_id, LDLM_FL_LOCAL_ONLY | dlmflags, - &info->mti_exp->exp_handle.h_cookie); + rc = mdt_fid_lock(ns, &lh->mlh_reg_lh, lh->mlh_reg_mode, policy, + res_id, LDLM_FL_LOCAL_ONLY | dlmflags, + info->mti_exp == NULL ? NULL : + &info->mti_exp->exp_handle.h_cookie); if (rc) mdt_object_unlock(info, o, lh, 1); else if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_PDO_LOCK)) && @@ -2427,10 +2467,66 @@ static int mdt_object_lock0(struct mdt_thread_info *info, struct mdt_object *o, RETURN(rc); } +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) +{ + 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); + } + + /* XXX do not support PERM/LAYOUT/XATTR lock for remote object yet */ + ibits &= ~(MDS_INODELOCK_PERM | MDS_INODELOCK_LAYOUT | + MDS_INODELOCK_XATTR); + 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 + * RPC, so it will instead enqueue EX lock for remote + * object anyway XXX*/ + if (lh->mlh_type == MDT_PDO_LOCK && + lh->mlh_pdo_hash != 0) { + CDEBUG(D_INFO, "%s: "DFID" convert PDO lock to" + "EX lock.\n", mdt_obd_name(info->mti_mdt), + PFID(mdt_object_fid(o))); + lh->mlh_pdo_hash = 0; + lh->mlh_rreg_mode = LCK_EX; + lh->mlh_type = MDT_REG_LOCK; + } + 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) + 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) { - return mdt_object_lock0(info, o, lh, ibits, false, locality); + return mdt_object_lock_internal(info, o, lh, ibits, false, locality); } int mdt_object_lock_try(struct mdt_thread_info *info, struct mdt_object *o, @@ -2439,7 +2535,7 @@ int mdt_object_lock_try(struct mdt_thread_info *info, struct mdt_object *o, struct mdt_lock_handle tmp = *lh; int rc; - rc = mdt_object_lock0(info, o, &tmp, ibits, true, locality); + rc = mdt_object_lock_internal(info, o, &tmp, ibits, true, locality); if (rc == 0) *lh = tmp; @@ -2590,11 +2686,11 @@ static int mdt_body_unpack(struct mdt_thread_info *info, __u32 flags) if (body == NULL) RETURN(-EFAULT); - if (!(body->valid & OBD_MD_FLID)) - RETURN(0); + if (!(body->mbo_valid & OBD_MD_FLID)) + RETURN(0); - if (!fid_is_sane(&body->fid1)) { - CERROR("Invalid fid: "DFID"\n", PFID(&body->fid1)); + if (!fid_is_sane(&body->mbo_fid1)) { + CERROR("Invalid fid: "DFID"\n", PFID(&body->mbo_fid1)); RETURN(-EINVAL); } @@ -2605,17 +2701,17 @@ static int mdt_body_unpack(struct mdt_thread_info *info, __u32 flags) */ 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->fid1, - req_capsule_client_get(pill, &RMF_CAPA1)); - - obj = mdt_object_find(env, info->mti_mdt, &body->fid1); - if (!IS_ERR(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->valid & OBD_MD_FLOSSCAPA) + 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)) { + 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; @@ -2644,7 +2740,7 @@ static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags) /* Pack reply. */ if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER)) req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, - info->mti_body->eadatasize); + DEF_REP_MD_SIZE); if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER)) req_capsule_set_size(pill, &RMF_LOGCOOKIES, RCL_SERVER, 0); @@ -2719,6 +2815,9 @@ 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.u.sp_ea.eadata = NULL; + info->mti_spec.u.sp_ea.eadatalen = 0; } void mdt_thread_info_fini(struct mdt_thread_info *info) @@ -2762,7 +2861,8 @@ struct mdt_thread_info *tsi2mdt_info(struct tgt_session_info *tsi) 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->fid1, lc); + mdt_set_capainfo(mti, 0, &tsi->tsi_mdt_body->mbo_fid1, + lc); } } mti->mti_body = tsi->tsi_mdt_body; @@ -2771,13 +2871,18 @@ struct mdt_thread_info *tsi2mdt_info(struct tgt_session_info *tsi) return mti; } -int mdt_tgt_connect(struct tgt_session_info *tsi) +static int mdt_tgt_connect(struct tgt_session_info *tsi) { struct ptlrpc_request *req = tgt_ses_req(tsi); int rc; ENTRY; + 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); + rc = tgt_connect(tsi); if (rc != 0) RETURN(rc); @@ -2898,21 +3003,18 @@ static struct mdt_it_flavor { } }; -int mdt_intent_lock_replace(struct mdt_thread_info *info, - struct ldlm_lock **lockp, - struct ldlm_lock *new_lock, - struct mdt_lock_handle *lh, - __u64 flags) +static int +mdt_intent_lock_replace(struct mdt_thread_info *info, + struct ldlm_lock **lockp, + struct mdt_lock_handle *lh, + __u64 flags) { struct ptlrpc_request *req = mdt_info_req(info); struct ldlm_lock *lock = *lockp; + struct ldlm_lock *new_lock; - /* - * Get new lock only for cases when possible resent did not find any - * lock. - */ - if (new_lock == NULL) - new_lock = ldlm_handle2lock_long(&lh->mlh_reg_lh, 0); + /* If possible resent found a lock, @lh is set to its handle */ + new_lock = ldlm_handle2lock_long(&lh->mlh_reg_lh, 0); if (new_lock == NULL && (flags & LDLM_FL_INTENT_ONLY)) { lh->mlh_reg_lh.cookie = 0; @@ -2944,6 +3046,8 @@ int mdt_intent_lock_replace(struct mdt_thread_info *info, */ LASSERT(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT); + + LDLM_LOCK_RELEASE(new_lock); lh->mlh_reg_lh.cookie = 0; RETURN(ELDLM_LOCK_REPLACED); } @@ -2985,52 +3089,30 @@ int mdt_intent_lock_replace(struct mdt_thread_info *info, static void mdt_intent_fixup_resent(struct mdt_thread_info *info, struct ldlm_lock *new_lock, - struct ldlm_lock **old_lock, struct mdt_lock_handle *lh, - enum mdt_it_code opcode) + __u64 flags) { struct ptlrpc_request *req = mdt_info_req(info); - struct obd_export *exp = req->rq_export; - struct lustre_handle remote_hdl; struct ldlm_request *dlmreq; - struct ldlm_lock *lock; if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)) return; dlmreq = req_capsule_client_get(info->mti_pill, &RMF_DLM_REQ); - remote_hdl = dlmreq->lock_handle[0]; - /* If the client does not require open lock, it does not need to - * search lock in exp_lock_hash, since the server thread will - * make sure the lock will be released, and the resend request - * can always re-enqueue the lock */ - if ((opcode != MDT_IT_OPEN) || (opcode == MDT_IT_OPEN && - info->mti_spec.sp_cr_flags & MDS_OPEN_LOCK)) { - /* In the function below, .hs_keycmp resolves to - * ldlm_export_lock_keycmp() */ - /* coverity[overrun-buffer-val] */ - lock = cfs_hash_lookup(exp->exp_lock_hash, &remote_hdl); - if (lock) { - lock_res_and_lock(lock); - if (lock != new_lock) { - lh->mlh_reg_lh.cookie = lock->l_handle.h_cookie; - lh->mlh_reg_mode = lock->l_granted_mode; - - LDLM_DEBUG(lock, "Restoring lock cookie"); - DEBUG_REQ(D_DLMTRACE, req, - "restoring lock cookie "LPX64, - lh->mlh_reg_lh.cookie); - if (old_lock) - *old_lock = LDLM_LOCK_GET(lock); - cfs_hash_put(exp->exp_lock_hash, - &lock->l_exp_hash); - unlock_res_and_lock(lock); - return; - } - cfs_hash_put(exp->exp_lock_hash, &lock->l_exp_hash); - unlock_res_and_lock(lock); - } + + /* Check if this is a resend case (MSG_RESENT is set on RPC) and a + * lock was found by ldlm_handle_enqueue(); if so @lh must be + * initialized. */ + if (flags & LDLM_FL_RESENT) { + lh->mlh_reg_lh.cookie = new_lock->l_handle.h_cookie; + lh->mlh_reg_mode = new_lock->l_granted_mode; + + LDLM_DEBUG(new_lock, "Restoring lock cookie"); + DEBUG_REQ(D_DLMTRACE, req, "restoring lock cookie "LPX64, + lh->mlh_reg_lh.cookie); + 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. @@ -3045,8 +3127,8 @@ static void mdt_intent_fixup_resent(struct mdt_thread_info *info, */ lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT); - DEBUG_REQ(D_DLMTRACE, req, "no existing lock with rhandle "LPX64, - remote_hdl.cookie); + DEBUG_REQ(D_DLMTRACE, req, "no existing lock with rhandle "LPX64, + dlmreq->lock_handle[0].cookie); } static int mdt_intent_getxattr(enum mdt_it_code opcode, @@ -3063,7 +3145,7 @@ static int mdt_intent_getxattr(enum mdt_it_code opcode, * (for the resend case) or a new lock. Below we will use it to * replace the original lock. */ - mdt_intent_fixup_resent(info, *lockp, NULL, lhc, opcode); + mdt_intent_fixup_resent(info, *lockp, lhc, flags); 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, @@ -3075,7 +3157,7 @@ static int mdt_intent_getxattr(enum mdt_it_code opcode, grc = mdt_getxattr(info); - rc = mdt_intent_lock_replace(info, lockp, NULL, lhc, flags); + rc = mdt_intent_lock_replace(info, lockp, lhc, flags); if (mdt_info_req(info)->rq_repmsg != NULL) ldlm_rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP); @@ -3093,10 +3175,8 @@ static int mdt_intent_getattr(enum mdt_it_code opcode, __u64 flags) { struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_RMT]; - struct ldlm_lock *new_lock = NULL; __u64 child_bits; struct ldlm_reply *ldlm_rep; - struct ptlrpc_request *req; struct mdt_body *reqbody; struct mdt_body *repbody; int rc, rc2; @@ -3108,9 +3188,9 @@ static int mdt_intent_getattr(enum mdt_it_code opcode, repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); LASSERT(repbody); - info->mti_cross_ref = !!(reqbody->valid & OBD_MD_FLCROSSREF); - repbody->eadatasize = 0; - repbody->aclsize = 0; + info->mti_cross_ref = !!(reqbody->mbo_valid & OBD_MD_FLCROSSREF); + repbody->mbo_eadatasize = 0; + repbody->mbo_aclsize = 0; switch (opcode) { case MDT_IT_LOOKUP: @@ -3129,12 +3209,11 @@ static int mdt_intent_getattr(enum mdt_it_code opcode, if (rc) GOTO(out_shrink, rc); - req = info->mti_pill->rc_req; ldlm_rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP); mdt_set_disposition(info, ldlm_rep, DISP_IT_EXECD); /* Get lock from request for possible resent case. */ - mdt_intent_fixup_resent(info, *lockp, &new_lock, lhc, opcode); + mdt_intent_fixup_resent(info, *lockp, lhc, flags); rc = mdt_getattr_name_lock(info, lhc, child_bits, ldlm_rep); ldlm_rep->lock_policy_res2 = clear_serious(rc); @@ -3147,7 +3226,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, new_lock, lhc, flags); + rc = mdt_intent_lock_replace(info, lockp, lhc, flags); EXIT; out_ucred: mdt_exit_ucred(info); @@ -3164,11 +3243,11 @@ static int mdt_intent_layout(enum mdt_it_code opcode, struct ldlm_lock **lockp, __u64 flags) { + struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_LAYOUT]; struct layout_intent *layout; struct lu_fid *fid; struct mdt_object *obj = NULL; - struct md_object *child = NULL; - int rc; + int rc = 0; ENTRY; if (opcode != MDT_IT_LAYOUT) { @@ -3180,16 +3259,20 @@ static int mdt_intent_layout(enum mdt_it_code opcode, fid = &info->mti_tmp_fid2; fid_extract_from_res_name(fid, &(*lockp)->l_resource->lr_name); + /* Get lock from request for possible resent case. */ + mdt_intent_fixup_resent(info, *lockp, lhc, flags); + obj = mdt_object_find(info->mti_env, info->mti_mdt, fid); if (IS_ERR(obj)) RETURN(PTR_ERR(obj)); if (mdt_object_exists(obj) && !mdt_object_remote(obj)) { - child = mdt_object_child(obj); - /* get the length of lsm */ - rc = mo_xattr_get(info->mti_env, child, &LU_BUF_NULL, - XATTR_NAME_LOV); + rc = mdt_attr_get_eabuf_size(info, obj); + if (rc < 0) { + mdt_object_put(info->mti_env, obj); + RETURN(rc); + } if (rc > info->mti_mdt->mdt_max_mdsize) info->mti_mdt->mdt_max_mdsize = rc; @@ -3198,17 +3281,19 @@ static int mdt_intent_layout(enum mdt_it_code opcode, 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, - ldlm_lvbo_size(*lockp)); + req_capsule_set_size(info->mti_pill, &RMF_DLM_LVB, RCL_SERVER, rc); rc = req_capsule_server_pack(info->mti_pill); if (rc != 0) RETURN(-EINVAL); + if (lustre_handle_is_used(&lhc->mlh_reg_lh)) + rc = mdt_intent_lock_replace(info, lockp, lhc, flags); + layout = req_capsule_client_get(info->mti_pill, &RMF_LAYOUT_INTENT); LASSERT(layout != NULL); if (layout->li_opc == LAYOUT_INTENT_ACCESS) - /* return to normal ldlm handling */ - RETURN(0); + /* return to normal/resent ldlm handling */ + RETURN(rc); CERROR("%s: Unsupported layout intent (%d)\n", mdt_obd_name(info->mti_mdt), layout->li_opc); @@ -3243,7 +3328,7 @@ static int mdt_intent_reint(enum mdt_it_code opcode, } /* Get lock from request for possible resent case. */ - mdt_intent_fixup_resent(info, *lockp, NULL, lhc, opcode); + mdt_intent_fixup_resent(info, *lockp, lhc, flags); rc = mdt_reint_internal(info, lhc, opc); @@ -3262,7 +3347,7 @@ static int mdt_intent_reint(enum mdt_it_code opcode, if (rc == -EREMOTE || mdt_get_disposition(rep, DISP_OPEN_LOCK)) { LASSERT(lustre_handle_is_used(&lhc->mlh_reg_lh)); rep->lock_policy_res2 = 0; - rc = mdt_intent_lock_replace(info, lockp, NULL, lhc, flags); + rc = mdt_intent_lock_replace(info, lockp, lhc, flags); RETURN(rc); } @@ -3295,7 +3380,7 @@ static int mdt_intent_reint(enum mdt_it_code opcode, 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, NULL, lhc, flags); + rc = mdt_intent_lock_replace(info, lockp, lhc, flags); RETURN(rc); } else { lhc->mlh_reg_lh.cookie = 0ull; @@ -3488,10 +3573,8 @@ static void mdt_seq_fini_cli(struct mdt_device *mdt) if (ss == NULL) return; - if (ss->ss_server_seq == NULL) + if (ss->ss_server_seq != NULL) seq_server_set_cli(NULL, ss->ss_server_seq, NULL); - - return; } static int mdt_seq_fini(const struct lu_env *env, struct mdt_device *mdt) @@ -3708,7 +3791,6 @@ static int mdt_fld_init(const struct lu_env *env, static void mdt_stack_pre_fini(const struct lu_env *env, struct mdt_device *m, struct lu_device *top) { - struct obd_device *obd; struct lustre_cfg_bufs *bufs; struct lustre_cfg *lcfg; struct mdt_thread_info *info; @@ -3723,7 +3805,6 @@ static void mdt_stack_pre_fini(const struct lu_env *env, LASSERT(m->mdt_child_exp); LASSERT(m->mdt_child_exp->exp_obd); - obd = m->mdt_child_exp->exp_obd; /* process cleanup, pass mdt obd name to get obd umount flags */ /* XXX: this is needed because all layers are referenced by @@ -3734,7 +3815,7 @@ static void mdt_stack_pre_fini(const struct lu_env *env, 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)); + CERROR("%s: cannot alloc lcfg\n", mdt_obd_name(m)); return; } top->ld_ops->ldo_process_config(env, top, lcfg); @@ -3935,7 +4016,6 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt, site->ls_top_dev = &mdt->mdt_lu_dev; mdt->mdt_child = lu2md_dev(mdt->mdt_child_exp->exp_obd->obd_lu_dev); - /* now connect to bottom OSD */ snprintf(name, MAX_OBD_NAME, "%s-osd", dev); rc = mdt_connect_to_next(env, mdt, name, &mdt->mdt_bottom_exp); @@ -3944,7 +4024,6 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt, mdt->mdt_bottom = lu2dt_dev(mdt->mdt_bottom_exp->exp_obd->obd_lu_dev); - rc = lu_env_refill((struct lu_env *)env); if (rc != 0) CERROR("Failure to refill session: '%d'\n", rc); @@ -4152,7 +4231,6 @@ 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(HABEO_CORPUS| HABEO_REFERO, MDS_IS_SUBDIR, mdt_is_subdir), TGT_MDT_HDL(0, MDS_QUOTACTL, mdt_quotactl), TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO | MUTABOR, MDS_HSM_PROGRESS, mdt_hsm_progress), @@ -4167,8 +4245,9 @@ TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO | MUTABOR, MDS_HSM_STATE_SET, TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_HSM_ACTION, mdt_hsm_action), TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_HSM_REQUEST, mdt_hsm_request), -TGT_MDT_HDL(HABEO_CORPUS|HABEO_REFERO | MUTABOR, MDS_SWAP_LAYOUTS, - mdt_swap_layouts) +TGT_MDT_HDL(HABEO_CLAVIS | HABEO_CORPUS | HABEO_REFERO | MUTABOR, + MDS_SWAP_LAYOUTS, + mdt_swap_layouts), }; static struct tgt_handler mdt_sec_ctx_ops[] = { @@ -4203,8 +4282,8 @@ static struct tgt_opc_slice mdt_common_slice[] = { .tos_hs = mdt_sec_ctx_ops }, { - .tos_opc_start = UPDATE_OBJ, - .tos_opc_end = UPDATE_LAST_OPC, + .tos_opc_start = OUT_UPDATE_FIRST_OPC, + .tos_opc_end = OUT_UPDATE_LAST_OPC, .tos_hs = tgt_out_handlers }, { @@ -4246,6 +4325,10 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) struct lfsck_stop stop; ENTRY; + stop.ls_status = LS_PAUSED; + stop.ls_flags = 0; + next->md_ops->mdo_iocontrol(env, next, OBD_IOC_STOP_LFSCK, 0, &stop); + target_recovery_fini(obd); ping_evictor_stop(); mdt_stack_pre_fini(env, m, md2lu_dev(m->mdt_child)); @@ -4275,16 +4358,7 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) mdt_quota_fini(env, m); - cfs_free_nidlist(&m->mdt_nosquash_nids); - if (m->mdt_nosquash_str) { - OBD_FREE(m->mdt_nosquash_str, m->mdt_nosquash_strlen); - m->mdt_nosquash_str = NULL; - m->mdt_nosquash_strlen = 0; - } - - stop.ls_status = LS_PAUSED; - stop.ls_flags = 0; - next->md_ops->mdo_iocontrol(env, next, OBD_IOC_STOP_LFSCK, 0, &stop); + cfs_free_nidlist(&m->mdt_squash.rsi_nosquash_nids); mdt_seq_fini(env, m); mdt_fld_fini(env, m); @@ -4300,12 +4374,12 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) LASSERT(atomic_read(&d->ld_ref) == 0); - server_put_mount(mdt_obd_name(m)); + server_put_mount(mdt_obd_name(m), true); EXIT; } -int mdt_postrecov(const struct lu_env *, struct mdt_device *); +static int mdt_postrecov(const struct lu_env *, struct mdt_device *); static int mdt_init0(const struct lu_env *env, struct mdt_device *m, struct lu_device_type *ldt, struct lustre_cfg *cfg) @@ -4370,12 +4444,10 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, m->mdt_capa_timeout = CAPA_TIMEOUT; m->mdt_capa_alg = CAPA_HMAC_ALG_SHA1; m->mdt_ck_timeout = CAPA_KEY_TIMEOUT; - m->mdt_squash_uid = 0; - m->mdt_squash_gid = 0; - CFS_INIT_LIST_HEAD(&m->mdt_nosquash_nids); - m->mdt_nosquash_str = NULL; - m->mdt_nosquash_strlen = 0; - init_rwsem(&m->mdt_squash_sem); + m->mdt_squash.rsi_uid = 0; + m->mdt_squash.rsi_gid = 0; + INIT_LIST_HEAD(&m->mdt_squash.rsi_nosquash_nids); + init_rwsem(&m->mdt_squash.rsi_sem); spin_lock_init(&m->mdt_osfs_lock); m->mdt_osfs_age = cfs_time_shift_64(-1000); m->mdt_enable_remote_dir = 0; @@ -4556,7 +4628,7 @@ err_fini_stack: mdt_stack_fini(env, m, md2lu_dev(m->mdt_child)); err_lmi: if (lmi) - server_put_mount(dev); + server_put_mount(dev, true); return(rc); } @@ -4584,7 +4656,6 @@ static int mdt_process_config(const struct lu_env *env, switch (cfg->lcfg_command) { case LCFG_PARAM: { - struct lprocfs_static_vars lvars; struct obd_device *obd = d->ld_obd; /* For interoperability */ @@ -4619,14 +4690,13 @@ static int mdt_process_config(const struct lu_env *env, } } - lprocfs_mdt_init_vars(&lvars); - rc = class_process_proc_param(PARAM_MDT, lvars.obd_vars, - cfg, obd); + rc = class_process_proc_seq_param(PARAM_MDT, obd->obd_vars, + cfg, obd); if (rc > 0 || rc == -ENOSYS) { /* is it an HSM var ? */ - rc = class_process_proc_param(PARAM_HSM, - hsm_cdt_get_proc_vars(), - cfg, obd); + rc = class_process_proc_seq_param(PARAM_HSM, + hsm_cdt_get_proc_vars(), + cfg, obd); if (rc > 0 || rc == -ENOSYS) /* we don't understand; pass it on */ rc = next->ld_ops->ldo_process_config(env, next, @@ -4732,7 +4802,6 @@ static int mdt_prepare(const struct lu_env *env, struct mdt_device *mdt = mdt_dev(cdev); struct lu_device *next = &mdt->mdt_child->md_lu_dev; struct obd_device *obd = cdev->ld_obd; - struct lfsck_start_param lsp; int rc; ENTRY; @@ -4756,17 +4825,6 @@ static int mdt_prepare(const struct lu_env *env, * register the namespace to such instance. */ LASSERTF(rc == 0, "register namespace failed: rc = %d\n", rc); - 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) { - CWARN("%s: auto trigger paused LFSCK failed: rc = %d\n", - mdt_obd_name(mdt), rc); - rc = 0; - } - if (mdt->mdt_seq_site.ss_node_id == 0) { rc = mdt->mdt_child->md_ops->mdo_root_get(env, mdt->mdt_child, &mdt->mdt_md_root_fid); @@ -4848,12 +4906,14 @@ static int mdt_connect_internal(struct obd_export *exp, data->ocd_connect_flags &= MDT_CONNECT_SUPPORTED; data->ocd_ibits_known &= MDS_INODELOCK_FULL; - /* If no known bits (which should not happen, probably, - as everybody should support LOOKUP and UPDATE bits at least) - revert to compat mode with plain locks. */ - if (!data->ocd_ibits_known && - data->ocd_connect_flags & OBD_CONNECT_IBITS) - data->ocd_connect_flags &= ~OBD_CONNECT_IBITS; + if (!(data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) && + !(data->ocd_connect_flags & OBD_CONNECT_IBITS)) { + CWARN("%s: client %s does not support ibits lock, either " + "very old or an invalid client: flags "LPX64"\n", + mdt_obd_name(mdt), exp->exp_client_uuid.uuid, + data->ocd_connect_flags); + return -EBADE; + } if (!mdt->mdt_opts.mo_acl) data->ocd_connect_flags &= ~OBD_CONNECT_ACL; @@ -5122,13 +5182,13 @@ static int mdt_obd_disconnect(struct obd_export *exp) LASSERT(exp); class_export_get(exp); - rc = server_disconnect_export(exp); - if (rc != 0) - CDEBUG(D_IOCTL, "server disconnect error: %d\n", rc); + rc = server_disconnect_export(exp); + if (rc != 0) + CDEBUG(D_IOCTL, "server disconnect error: rc = %d\n", rc); - rc = mdt_export_cleanup(exp); - class_export_put(exp); - RETURN(rc); + rc = mdt_export_cleanup(exp); + class_export_put(exp); + RETURN(rc); } /* FIXME: Can we avoid using these two interfaces? */ @@ -5211,8 +5271,8 @@ struct path_lookup_info { int pli_fidcount; /**< number of \a pli_fids */ }; -static int mdt_links_read(struct mdt_thread_info *info, - struct mdt_object *mdt_obj, struct linkea_data *ldata) +int mdt_links_read(struct mdt_thread_info *info, struct mdt_object *mdt_obj, + struct linkea_data *ldata) { int rc; @@ -5270,28 +5330,32 @@ static int mdt_path_current(struct mdt_thread_info *info, --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])) { - mdt_obj = mdt_object_find(info->mti_env, mdt, - &pli->pli_fids[pli->pli_fidcount]); + struct lu_buf lmv_buf; + + mdt_obj = mdt_object_find(info->mti_env, mdt, tmpfid); if (IS_ERR(mdt_obj)) GOTO(out, rc = PTR_ERR(mdt_obj)); - if (mdt_object_remote(mdt_obj)) { - mdt_object_put(info->mti_env, mdt_obj); - GOTO(remote_out, rc = -EREMOTE); - } + if (!mdt_object_exists(mdt_obj)) { mdt_object_put(info->mti_env, mdt_obj); GOTO(out, rc = -ENOENT); } + if (mdt_object_remote(mdt_obj)) { + mdt_object_put(info->mti_env, mdt_obj); + GOTO(remote_out, rc = -EREMOTE); + } + rc = mdt_links_read(info, mdt_obj, &ldata); - mdt_object_put(info->mti_env, mdt_obj); - if (rc != 0) + if (rc != 0) { + mdt_object_put(info->mti_env, mdt_obj); GOTO(out, rc); + } leh = buf->lb_buf; lee = (struct link_ea_entry *)(leh + 1); /* link #0 */ @@ -5312,6 +5376,26 @@ static int mdt_path_current(struct mdt_thread_info *info, pli->pli_linkno++; } + lmv_buf.lb_buf = info->mti_xattr_buf; + lmv_buf.lb_len = sizeof(info->mti_xattr_buf); + /* Check if it is slave stripes */ + rc = mo_xattr_get(info->mti_env, mdt_object_child(mdt_obj), + &lmv_buf, XATTR_NAME_LMV); + mdt_object_put(info->mti_env, mdt_obj); + if (rc > 0) { + union lmv_mds_md *lmm = lmv_buf.lb_buf; + + /* For slave stripes, get its master */ + if (le32_to_cpu(lmm->lmv_magic) == LMV_MAGIC_STRIPE) { + pli->pli_fids[pli->pli_fidcount] = *tmpfid; + continue; + } + } else if (rc < 0 && rc != -ENODATA) { + GOTO(out, rc); + } + + rc = 0; + /* Pack the name in the end of the buffer */ ptr -= tmpname->ln_namelen; if (ptr - 1 <= pli->pli_path) @@ -5417,7 +5501,7 @@ static int mdt_fid2path(struct mdt_thread_info *info, if (rc < 0) { mdt_object_put(info->mti_env, obj); - CDEBUG(D_IOCTL, "nonlocal object "DFID": %d\n", + CDEBUG(D_IOCTL, "nonlocal object "DFID": rc = %d\n", PFID(&fp->gf_fid), rc); RETURN(rc); } @@ -5465,7 +5549,7 @@ int mdt_get_info(struct tgt_session_info *tsi) key = req_capsule_client_get(tsi->tsi_pill, &RMF_GETINFO_KEY); if (key == NULL) { - CDEBUG(D_IOCTL, "No GETINFO key"); + CDEBUG(D_IOCTL, "No GETINFO key\n"); RETURN(err_serious(-EFAULT)); } keylen = req_capsule_get_size(tsi->tsi_pill, &RMF_GETINFO_KEY, @@ -5473,7 +5557,8 @@ int mdt_get_info(struct tgt_session_info *tsi) vallen = req_capsule_client_get(tsi->tsi_pill, &RMF_GETINFO_VALLEN); if (vallen == NULL) { - CDEBUG(D_IOCTL, "Unable to get RMF_GETINFO_VALLEN buffer"); + CDEBUG(D_IOCTL, "%s: cannot get RMF_GETINFO_VALLEN buffer\n", + tgt_name(tsi->tsi_tgt)); RETURN(err_serious(-EFAULT)); } @@ -5485,7 +5570,8 @@ int mdt_get_info(struct tgt_session_info *tsi) valout = req_capsule_server_get(tsi->tsi_pill, &RMF_GETINFO_VAL); if (valout == NULL) { - CDEBUG(D_IOCTL, "Unable to get get-info RPC out buffer"); + CDEBUG(D_IOCTL, "%s: cannot get get-info RPC out buffer\n", + tgt_name(tsi->tsi_tgt)); RETURN(err_serious(-EFAULT)); } @@ -5665,17 +5751,27 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len, RETURN(rc); } -int mdt_postrecov(const struct lu_env *env, struct mdt_device *mdt) +static int mdt_postrecov(const struct lu_env *env, struct mdt_device *mdt) { - struct lu_device *ld = md2lu_dev(mdt->mdt_child); - int rc; - ENTRY; + struct lu_device *ld = md2lu_dev(mdt->mdt_child); + struct lfsck_start_param lsp; + int rc; + ENTRY; - rc = ld->ld_ops->ldo_recovery_complete(env, ld); - RETURN(rc); + 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); } -int mdt_obd_postrecov(struct obd_device *obd) +static int mdt_obd_postrecov(struct obd_device *obd) { struct lu_env env; int rc; @@ -5789,13 +5885,13 @@ void mdt_enable_cos(struct mdt_device *mdt, int val) mdt->mdt_opts.mo_cos = !!val; rc = lu_env_init(&env, LCT_LOCAL); - if (unlikely(rc != 0)) { - CWARN("lu_env initialization failed with rc = %d," - "cannot sync\n", rc); - return; - } - mdt_device_sync(&env, mdt); - lu_env_fini(&env); + if (unlikely(rc != 0)) { + CWARN("%s: lu_env initialization failed, cannot " + "sync: rc = %d\n", mdt_obd_name(mdt), rc); + return; + } + mdt_device_sync(&env, mdt); + lu_env_fini(&env); } /** @@ -5825,7 +5921,6 @@ static struct lu_device_type mdt_device_type = { static int __init mdt_mod_init(void) { - struct lprocfs_static_vars lvars; int rc; CLASSERT(sizeof("0x0123456789ABCDEF:0x01234567:0x01234567") == @@ -5840,12 +5935,11 @@ static int __init mdt_mod_init(void) if (rc) GOTO(lu_fini, rc); - lprocfs_mdt_init_vars(&lvars); - rc = class_register_type(&mdt_obd_device_ops, NULL, NULL, + rc = class_register_type(&mdt_obd_device_ops, NULL, true, NULL, #ifndef HAVE_ONLY_PROCFS_SEQ - lvars.module_vars, + NULL, #endif - LUSTRE_MDT_NAME, &mdt_device_type); + LUSTRE_MDT_NAME, &mdt_device_type); if (rc) GOTO(mds_fini, rc); lu_fini: