X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_handler.c;h=f244e61ed93fa879a61d13442596bf1ed215b182;hp=1d9f96c95d37d178c4349cb96a697e4461a8d407;hb=9849f6110680d6846e05e9787f1821b0f86b5461;hpb=93a6346f8b73f68cb5bc02a3c826ac0e5b4c236e diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c index 1d9f96c..f244e61 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 */ @@ -62,13 +63,11 @@ /* lu2dt_dev() */ #include #include -#include #include #include "mdt_internal.h" #include #include #include -#include #include mdl_mode_t mdt_mdl_lock_modes[] = { @@ -101,6 +100,12 @@ static const struct lu_object_operations mdt_obj_ops; /* Slab for MDT object allocation */ static struct kmem_cache *mdt_object_kmem; +/* For HSM restore handles */ +struct kmem_cache *mdt_hsm_cdt_kmem; + +/* For HSM request handles */ +struct kmem_cache *mdt_hsm_car_kmem; + static struct lu_kmem_descr mdt_caches[] = { { .ckd_cache = &mdt_object_kmem, @@ -108,6 +113,16 @@ static struct lu_kmem_descr mdt_caches[] = { .ckd_size = sizeof(struct mdt_object) }, { + .ckd_cache = &mdt_hsm_cdt_kmem, + .ckd_name = "mdt_cdt_restore_handle", + .ckd_size = sizeof(struct cdt_restore_handle) + }, + { + .ckd_cache = &mdt_hsm_car_kmem, + .ckd_name = "mdt_cdt_agent_req", + .ckd_size = sizeof(struct cdt_agent_req) + }, + { .ckd_cache = NULL } }; @@ -120,21 +135,25 @@ int mdt_get_disposition(struct ldlm_reply *rep, int flag) } void mdt_clear_disposition(struct mdt_thread_info *info, - struct ldlm_reply *rep, int flag) + struct ldlm_reply *rep, int flag) { - if (info) - info->mti_opdata &= ~flag; - if (rep) - rep->lock_policy_res1 &= ~flag; + if (info) { + info->mti_opdata &= ~flag; + tgt_opdata_clear(info->mti_env, flag); + } + if (rep) + rep->lock_policy_res1 &= ~flag; } void mdt_set_disposition(struct mdt_thread_info *info, - struct ldlm_reply *rep, int flag) + struct ldlm_reply *rep, int flag) { - if (info) - info->mti_opdata |= flag; - if (rep) - rep->lock_policy_res1 |= flag; + if (info) { + info->mti_opdata |= flag; + tgt_opdata_set(info->mti_env, flag); + } + if (rep) + rep->lock_policy_res1 |= flag; } void mdt_lock_reg_init(struct mdt_lock_handle *lh, ldlm_mode_t lm) @@ -145,27 +164,28 @@ void mdt_lock_reg_init(struct mdt_lock_handle *lh, ldlm_mode_t lm) lh->mlh_type = MDT_REG_LOCK; } -void mdt_lock_pdo_init(struct mdt_lock_handle *lh, ldlm_mode_t lm, - const char *name, int namelen) +void mdt_lock_pdo_init(struct mdt_lock_handle *lh, ldlm_mode_t lock_mode, + const struct lu_name *lname) { - lh->mlh_reg_mode = lm; - lh->mlh_rreg_mode = lm; - lh->mlh_type = MDT_PDO_LOCK; + lh->mlh_reg_mode = lock_mode; + lh->mlh_rreg_mode = lock_mode; + lh->mlh_type = MDT_PDO_LOCK; - if (name != NULL && (name[0] != '\0')) { - LASSERT(namelen > 0); - lh->mlh_pdo_hash = full_name_hash(name, namelen); + if (lu_name_is_valid(lname)) { + lh->mlh_pdo_hash = full_name_hash(lname->ln_name, + lname->ln_namelen); /* XXX Workaround for LU-2856 - * Zero is a valid return value of full_name_hash, but several - * users of mlh_pdo_hash assume a non-zero hash value. We - * therefore map zero onto an arbitrary, but consistent - * value (1) to avoid problems further down the road. */ - if (unlikely(!lh->mlh_pdo_hash)) + * + * Zero is a valid return value of full_name_hash, but + * several users of mlh_pdo_hash assume a non-zero + * hash value. We therefore map zero onto an + * arbitrary, but consistent value (1) to avoid + * problems further down the road. */ + if (unlikely(lh->mlh_pdo_hash == 0)) lh->mlh_pdo_hash = 1; - } else { - LASSERT(namelen == 0); - lh->mlh_pdo_hash = 0ull; - } + } else { + lh->mlh_pdo_hash = 0; + } } static void mdt_lock_pdo_mode(struct mdt_thread_info *info, struct mdt_object *o, @@ -243,32 +263,34 @@ static void mdt_lock_pdo_mode(struct mdt_thread_info *info, struct mdt_object *o EXIT; } -int mdt_getstatus(struct mdt_thread_info *info) +static int mdt_getstatus(struct tgt_session_info *tsi) { - struct mdt_device *mdt = info->mti_mdt; + struct mdt_thread_info *info = tsi2mdt_info(tsi); + struct mdt_device *mdt = info->mti_mdt; struct mdt_body *repbody; - int rc; + int rc; + ENTRY; - rc = mdt_check_ucred(info); - if (rc) - RETURN(err_serious(rc)); + rc = mdt_check_ucred(info); + if (rc) + GOTO(out, rc = err_serious(rc)); - if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETSTATUS_PACK)) - RETURN(err_serious(-ENOMEM)); + if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETSTATUS_PACK)) + 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 (mdt->mdt_opts.mo_mds_capa && + 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; + struct mdt_object *root; + struct lustre_capa *capa; - root = mdt_object_find(info->mti_env, mdt, &repbody->fid1); - if (IS_ERR(root)) - RETURN(PTR_ERR(root)); + 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); @@ -277,15 +299,18 @@ int mdt_getstatus(struct mdt_thread_info *info) 0); mdt_object_put(info->mti_env, root); if (rc == 0) - repbody->valid |= OBD_MD_FLMDSCAPA; + repbody->mbo_valid |= OBD_MD_FLMDSCAPA; } - - RETURN(rc); + EXIT; +out: + mdt_thread_info_fini(info); + return rc; } -int mdt_statfs(struct mdt_thread_info *info) +static int mdt_statfs(struct tgt_session_info *tsi) { - struct ptlrpc_request *req = mdt_info_req(info); + struct ptlrpc_request *req = tgt_ses_req(tsi); + struct mdt_thread_info *info = tsi2mdt_info(tsi); struct md_device *next = info->mti_mdt->mdt_child; struct ptlrpc_service_part *svcpt; struct obd_statfs *osfs; @@ -293,23 +318,23 @@ int mdt_statfs(struct mdt_thread_info *info) ENTRY; - svcpt = info->mti_pill->rc_req->rq_rqbd->rqbd_svcpt; + svcpt = req->rq_rqbd->rqbd_svcpt; /* This will trigger a watchdog timeout */ OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_STATFS_LCW_SLEEP, (MDT_SERVICE_WATCHDOG_FACTOR * at_get(&svcpt->scp_at_estimate)) + 1); - rc = mdt_check_ucred(info); - if (rc) - RETURN(err_serious(rc)); + rc = mdt_check_ucred(info); + if (rc) + GOTO(out, rc = err_serious(rc)); if (OBD_FAIL_CHECK(OBD_FAIL_MDS_STATFS_PACK)) - RETURN(err_serious(-ENOMEM)); + GOTO(out, rc = err_serious(-ENOMEM)); osfs = req_capsule_server_get(info->mti_pill, &RMF_OBD_STATFS); if (!osfs) - RETURN(-EPROTO); + GOTO(out, rc = -EPROTO); /** statfs information are cached in the mdt_device */ if (cfs_time_before_64(info->mti_mdt->mdt_osfs_age, @@ -317,7 +342,7 @@ int mdt_statfs(struct mdt_thread_info *info) /** statfs data is too old, get up-to-date one */ rc = next->md_ops->mdo_statfs(info->mti_env, next, osfs); if (rc) - RETURN(rc); + GOTO(out, rc); spin_lock(&info->mti_mdt->mdt_osfs_lock); info->mti_mdt->mdt_osfs = *osfs; info->mti_mdt->mdt_osfs_age = cfs_time_current_64(); @@ -331,7 +356,8 @@ int mdt_statfs(struct mdt_thread_info *info) if (rc == 0) mdt_counter_incr(req, LPROC_MDT_STATFS); - +out: + mdt_thread_info_fini(info); RETURN(rc); } @@ -357,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, @@ -369,60 +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; - } else if ((ma->ma_valid & MA_LOV) && ma->ma_lmm && - (ma->ma_lmm->lmm_pattern & LOV_PATTERN_F_RELEASED)) { + /* 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. */ - b->blocks = 0; - b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; + /* But return 1 block for released file, unless tools like tar + * will consider it fully sparse. (LU-3864) + */ + if (unlikely(b->mbo_size == 0)) + b->mbo_blocks = 0; + else + 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 (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) @@ -450,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, - 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; @@ -489,43 +555,83 @@ static int mdt_big_xattr_get(struct mdt_thread_info *info, struct mdt_object *o, RETURN(rc); } -int mdt_attr_get_lov(struct mdt_thread_info *info, - struct mdt_object *o, struct md_attr *ma) +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; int rc; - buf->lb_buf = ma->ma_lmm; - buf->lb_len = ma->ma_lmm_size; - rc = mo_xattr_get(info->mti_env, next, buf, XATTR_NAME_LOV); + if (strcmp(name, XATTR_NAME_LOV) == 0) { + buf->lb_buf = ma->ma_lmm; + buf->lb_len = ma->ma_lmm_size; + LASSERT(!(ma->ma_valid & MA_LOV)); + } else if (strcmp(name, XATTR_NAME_LMV) == 0) { + buf->lb_buf = ma->ma_lmv; + buf->lb_len = ma->ma_lmv_size; + LASSERT(!(ma->ma_valid & MA_LMV)); + } else if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) { + buf->lb_buf = ma->ma_lmv; + buf->lb_len = ma->ma_lmv_size; + LASSERT(!(ma->ma_valid & MA_LMV_DEF)); + } else { + return -EINVAL; + } + + rc = mo_xattr_get(info->mti_env, next, buf, name); if (rc > 0) { - ma->ma_lmm_size = rc; - ma->ma_valid |= MA_LOV; + +got: + if (strcmp(name, XATTR_NAME_LOV) == 0) { + 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) { + ma->ma_lmv_size = rc; + 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 */ rc = 0; } else if (rc == -ERANGE) { - rc = mdt_big_xattr_get(info, o, XATTR_NAME_LOV); + /* Default LMV has fixed size, so it must be able to fit + * in the original buffer */ + if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) + return rc; + rc = mdt_big_xattr_get(info, o, name); if (rc > 0) { info->mti_big_lmm_used = 1; - ma->ma_valid |= MA_LOV; - ma->ma_lmm = info->mti_big_lmm; - ma->ma_lmm_size = rc; - /* 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; @@ -577,13 +683,17 @@ int mdt_attr_get_complex(struct mdt_thread_info *info, const struct lu_env *env = info->mti_env; struct md_object *next = mdt_object_child(o); struct lu_buf *buf = &info->mti_buf; - u32 mode = lu_object_attr(&next->mo_lu); int need = ma->ma_need; int rc = 0, rc2; + u32 mode; ENTRY; ma->ma_valid = 0; + if (mdt_object_exists(o) == 0) + GOTO(out, rc = -ENOENT); + mode = lu_object_attr(&next->mo_lu); + if (need & MA_INODE) { ma->ma_need = MA_INODE; rc = mo_attr_get(env, next, ma); @@ -601,23 +711,21 @@ int mdt_attr_get_complex(struct mdt_thread_info *info, } if (need & MA_LOV && (S_ISREG(mode) || S_ISDIR(mode))) { - rc = mdt_attr_get_lov(info, o, ma); + rc = mdt_stripe_get(info, o, ma, XATTR_NAME_LOV); if (rc) GOTO(out, rc); } if (need & MA_LMV && S_ISDIR(mode)) { - buf->lb_buf = ma->ma_lmv; - buf->lb_len = ma->ma_lmv_size; - rc2 = mo_xattr_get(env, next, buf, XATTR_NAME_LMV); - if (rc2 > 0) { - ma->ma_lmv_size = rc2; - ma->ma_valid |= MA_LMV; - } else if (rc2 == -ENODATA) { - /* no LMV EA */ - ma->ma_lmv_size = 0; - } else - GOTO(out, rc = rc2); + rc = mdt_stripe_get(info, o, ma, XATTR_NAME_LMV); + if (rc != 0) + GOTO(out, rc); + } + + if (need & MA_LMV_DEF && S_ISDIR(mode)) { + rc = mdt_stripe_get(info, o, ma, XATTR_NAME_DEFAULT_LMV); + if (rc != 0) + GOTO(out, rc); } if (need & MA_SOM && S_ISREG(mode)) { @@ -693,36 +801,53 @@ 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); - else + 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), + req->rq_export->exp_client_uuid.uuid); + } - /* 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) { - /* 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_LMV | MA_INODE; - } else { - ma->ma_lmm = buffer->lb_buf; - ma->ma_lmm_size = buffer->lb_len; - ma->ma_need = MA_LOV | MA_INODE | MA_HSM; - } + /* If it is dir object and client require MEA, then we got MEA */ + if (S_ISDIR(lu_object_attr(&next->mo_lu)) && + (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->mbo_valid & OBD_MD_MEA) + ma->ma_need |= MA_LMV; + else if (reqbody->mbo_valid & OBD_MD_DEFAULT_MEA) + ma->ma_need |= MA_LMV_DEF; + } + } else { + ma->ma_lmm = buffer->lb_buf; + ma->ma_lmm_size = buffer->lb_len; + ma->ma_need = MA_INODE | MA_HSM; + if (ma->ma_lmm_size > 0) + ma->ma_need |= MA_LOV; + } 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; @@ -742,8 +867,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); @@ -751,7 +876,7 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, /* the Lustre protocol supposes to return default striping * on the user-visible root if explicitly requested */ if ((ma->ma_valid & MA_LOV) == 0 && S_ISDIR(la->la_mode) && - (ma->ma_need & MA_LOV_DEF && is_root) && (ma->ma_need & MA_LOV)) { + (ma->ma_need & MA_LOV_DEF && is_root) && ma->ma_need & MA_LOV) { struct lu_fid rootfid; struct mdt_object *root; struct mdt_device *mdt = info->mti_mdt; @@ -762,7 +887,7 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, root = mdt_object_find(env, mdt, &rootfid); if (IS_ERR(root)) RETURN(PTR_ERR(root)); - rc = mdt_attr_get_lov(info, root, ma); + rc = mdt_stripe_get(info, root, ma, XATTR_NAME_LOV); mdt_object_put(info->mti_env, root); if (unlikely(rc)) { CERROR("%s: getattr error for "DFID": rc = %d\n", @@ -780,24 +905,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; - } - if (ma->ma_valid & MA_LMV) { - LASSERT(S_ISDIR(la->la_mode)); - repbody->eadatasize = ma->ma_lmv_size; - repbody->valid |= (OBD_MD_FLDIREA|OBD_MD_MEA); + 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->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->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", @@ -809,16 +948,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; @@ -832,33 +971,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); @@ -867,8 +1006,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; @@ -879,16 +1018,16 @@ 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 && - info->mti_mdt->mdt_opts.mo_mds_capa && + 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; @@ -898,7 +1037,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: @@ -920,7 +1059,7 @@ static int mdt_renew_capa(struct mdt_thread_info *info) * return directly, client will find body->valid OBD_MD_FLOSSCAPA * flag not set. */ - if (!obj || !info->mti_mdt->mdt_opts.mo_oss_capa || + if (!obj || !info->mti_mdt->mdt_lut.lut_oss_capa || !(exp_connect_flags(info->mti_exp) & OBD_CONNECT_OSS_CAPA)) RETURN(0); @@ -936,24 +1075,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 mdt_thread_info *info) +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)); @@ -964,39 +1103,62 @@ int mdt_getattr(struct mdt_thread_info *info) 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); + } - /* 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); + if (rc < 0) + GOTO(out, rc = err_serious(rc)); - rc = req_capsule_server_pack(pill); - if (unlikely(rc != 0)) - RETURN(err_serious(rc)); + req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, rc); + + rc = req_capsule_server_pack(pill); + if (unlikely(rc != 0)) + GOTO(out, rc = err_serious(rc)); 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: @@ -1004,39 +1166,15 @@ out_shrink: rc2 = mdt_fix_reply(info); if (rc == 0) rc = rc2; - return rc; -} - -int mdt_is_subdir(struct mdt_thread_info *info) -{ - 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; - - RETURN(rc); +out: + mdt_thread_info_fini(info); + return rc; } -int mdt_swap_layouts(struct mdt_thread_info *info) +static int mdt_swap_layouts(struct tgt_session_info *tsi) { - struct ptlrpc_request *req = mdt_info_req(info); + struct mdt_thread_info *info; + struct ptlrpc_request *req = tgt_ses_req(tsi); struct obd_export *exp = req->rq_export; struct mdt_object *o1, *o2, *o; struct mdt_lock_handle *lh1, *lh2; @@ -1054,26 +1192,31 @@ int mdt_swap_layouts(struct mdt_thread_info *info) if (!exp_connect_layout(exp)) RETURN(-EOPNOTSUPP); + info = tsi2mdt_info(tsi); + + if (info->mti_dlm_req != NULL) + ldlm_request_cancel(req, info->mti_dlm_req, 0, LATF_SKIP); + 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); @@ -1101,13 +1244,13 @@ int mdt_swap_layouts(struct mdt_thread_info *info) lh2 = &info->mti_lh[MDT_LH_OLD]; mdt_lock_reg_init(lh2, LCK_EX); - rc = mdt_object_lock(info, o1, lh1, MDS_INODELOCK_LAYOUT, - MDT_LOCAL_LOCK); + rc = mdt_object_lock(info, o1, lh1, MDS_INODELOCK_LAYOUT | + MDS_INODELOCK_XATTR, MDT_LOCAL_LOCK); if (rc < 0) GOTO(put, rc); - rc = mdt_object_lock(info, o2, lh2, MDS_INODELOCK_LAYOUT, - MDT_LOCAL_LOCK); + rc = mdt_object_lock(info, o2, lh2, MDS_INODELOCK_LAYOUT | + MDS_INODELOCK_XATTR, MDT_LOCAL_LOCK); if (rc < 0) GOTO(unlock1, rc); @@ -1121,6 +1264,7 @@ unlock1: put: mdt_object_put(info->mti_env, o); out: + mdt_thread_info_fini(info); RETURN(rc); } @@ -1136,7 +1280,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); @@ -1158,8 +1302,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); } @@ -1179,126 +1323,121 @@ 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; - const char *name = NULL; - int namelen = 0; 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; + bool is_resent; + bool try_layout; + int ma_need = 0; + int rc; + ENTRY; - ENTRY; + is_resent = lustre_handle_is_used(&lhc->mlh_reg_lh); + LASSERT(ergo(is_resent, + lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)); - is_resent = lustre_handle_is_used(&lhc->mlh_reg_lh); - LASSERT(ergo(is_resent, - lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)); + if (parent == NULL) + RETURN(-ENOENT); - LASSERT(parent != NULL); - name = req_capsule_client_get(info->mti_pill, &RMF_NAME); - if (name == NULL) - RETURN(err_serious(-EFAULT)); + 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); - namelen = req_capsule_get_size(info->mti_pill, &RMF_NAME, - RCL_CLIENT) - 1; - if (!info->mti_cross_ref) { - /* - * XXX: Check for "namelen == 0" 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 (namelen == 0) { - 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)); - - name = NULL; - CDEBUG(D_INODE, "getattr with lock for "DFID"/"DFID", " - "ldlm_rep = %p\n", - PFID(mdt_object_fid(parent)), - PFID(&reqbody->fid2), ldlm_rep); - } else { - lname = mdt_name(info->mti_env, (char *)name, namelen); - CDEBUG(D_INODE, "getattr with lock for "DFID"/%s, " - "ldlm_rep = %p\n", PFID(mdt_object_fid(parent)), - name, ldlm_rep); - } + rc = mdt_check_resent_lock(info, child, lhc); + if (rc < 0) { + RETURN(rc); + } else if (rc > 0) { + mdt_lock_handle_init(lhc); + mdt_lock_reg_init(lhc, LCK_PR); + + /* + * Object's name is on another MDS, no lookup or layout + * lock is needed here but update lock is. + */ + child_bits &= ~(MDS_INODELOCK_LOOKUP | + MDS_INODELOCK_LAYOUT); + child_bits |= MDS_INODELOCK_PERM | MDS_INODELOCK_UPDATE; + + rc = mdt_object_lock(info, child, lhc, child_bits, + MDT_LOCAL_LOCK); + if (rc < 0) + RETURN(rc); + } + + /* 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); } - mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_EXECD); - if (unlikely(!mdt_object_exists(parent))) { + 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); - } else if (!info->mti_cross_ref) { - LASSERTF(!mdt_object_remote(parent), - "Parent "DFID" is on remote server\n", - PFID(mdt_object_fid(parent))); } - if (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_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 { - mdt_lock_handle_init(lhc); - mdt_lock_reg_init(lhc, LCK_PR); - /* - * Object's name is on another MDS, no lookup lock is - * needed here but update is. - */ - child_bits &= ~MDS_INODELOCK_LOOKUP; - child_bits |= MDS_INODELOCK_PERM | MDS_INODELOCK_UPDATE; + 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); + } - rc = mdt_object_lock(info, child, lhc, child_bits, - MDT_LOCAL_LOCK); - } - if (rc == 0) { - /* Finally, we can get attr for child. */ - 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); - } + 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 (lname) { - /* step 1: lock parent only if parent is a directory */ + /* 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]; - mdt_lock_pdo_init(lhp, LCK_PR, name, namelen); + 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); @@ -1308,60 +1447,45 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, /* 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); - 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); + } - /* - *step 3: find the child object by fid & lock it. - * regardless if it is local or remote. - */ - child = mdt_object_find(info->mti_env, info->mti_mdt, child_fid); + mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS); + + /* + *step 3: find the child object by fid & lock it. + * regardless if it is local or remote. + * + *Note: LU-3240 (commit 762f2114d282a98ebfa4dbbeea9298a8088ad24e) + * set parent dir fid the same as child fid in getattr by fid case + * we should not lu_object_find() the object again, could lead + * to hung if there is a concurrent unlink destroyed the object. + */ + if (lu_fid_eq(mdt_object_fid(parent), child_fid)) { + mdt_object_get(info->mti_env, parent); + child = parent; + } else { + child = mdt_object_find(info->mti_env, info->mti_mdt, + child_fid); + } 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, @@ -1380,11 +1504,11 @@ relock: if (unlikely(rc != 0)) GOTO(out_child, rc); - /* If the file has not been changed for some time, we - * return not only a LOOKUP lock, but also an UPDATE - * lock and this might save us RPC on later STAT. For - * directories, it also let negative dentry starts - * working for this dir. */ + /* If the file has not been changed for some time, we + * return not only a LOOKUP lock, but also an UPDATE + * lock and this might save us RPC on later STAT. For + * directories, it also let negative dentry cache start + * working for this dir. */ if (ma->ma_valid & MA_INODE && ma->ma_attr.la_valid & LA_CTIME && info->mti_mdt->mdt_namespace->ns_ctime_age_limit + @@ -1398,7 +1522,7 @@ relock: if (!OBD_FAIL_CHECK(OBD_FAIL_MDS_NO_LL_GETATTR) && exp_connect_layout(info->mti_exp) && S_ISREG(lu_object_attr(&child->mot_obj)) && - ldlm_rep != NULL) { + !mdt_object_remote(child) && ldlm_rep != NULL) { /* try to grant layout lock for regular file. */ try_layout = true; } @@ -1418,6 +1542,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); } @@ -1439,7 +1567,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), @@ -1462,8 +1589,9 @@ out_parent: } /* normal handler: should release the child lock */ -int mdt_getattr_name(struct mdt_thread_info *info) +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]; struct mdt_body *reqbody; struct mdt_body *repbody; @@ -1475,9 +1603,9 @@ int mdt_getattr_name(struct mdt_thread_info *info) 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)) @@ -1495,47 +1623,42 @@ out_shrink: rc2 = mdt_fix_reply(info); if (rc == 0) rc = rc2; - return rc; + mdt_thread_info_fini(info); + return rc; } static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len, void *karg, void *uarg); -int mdt_set_info(struct mdt_thread_info *info) +static int mdt_set_info(struct tgt_session_info *tsi) { - struct ptlrpc_request *req = mdt_info_req(info); - char *key; - void *val; - int keylen, vallen, rc = 0; - ENTRY; - - rc = req_capsule_server_pack(info->mti_pill); - if (rc) - RETURN(rc); + struct ptlrpc_request *req = tgt_ses_req(tsi); + char *key; + void *val; + int keylen, vallen, rc = 0; - key = req_capsule_client_get(info->mti_pill, &RMF_SETINFO_KEY); - if (key == NULL) { - DEBUG_REQ(D_HA, req, "no set_info key"); - RETURN(-EFAULT); - } + ENTRY; - keylen = req_capsule_get_size(info->mti_pill, &RMF_SETINFO_KEY, - RCL_CLIENT); + key = req_capsule_client_get(tsi->tsi_pill, &RMF_SETINFO_KEY); + if (key == NULL) { + DEBUG_REQ(D_HA, req, "no set_info key"); + RETURN(err_serious(-EFAULT)); + } - val = req_capsule_client_get(info->mti_pill, &RMF_SETINFO_VAL); - if (val == NULL) { - DEBUG_REQ(D_HA, req, "no set_info val"); - RETURN(-EFAULT); - } + keylen = req_capsule_get_size(tsi->tsi_pill, &RMF_SETINFO_KEY, + RCL_CLIENT); - vallen = req_capsule_get_size(info->mti_pill, &RMF_SETINFO_VAL, - RCL_CLIENT); + val = req_capsule_client_get(tsi->tsi_pill, &RMF_SETINFO_VAL); + if (val == NULL) { + DEBUG_REQ(D_HA, req, "no set_info val"); + RETURN(err_serious(-EFAULT)); + } - /* Swab any part of val you need to here */ - if (KEY_IS(KEY_READ_ONLY)) { - req->rq_status = 0; - lustre_msg_set_status(req->rq_repmsg, 0); + vallen = req_capsule_get_size(tsi->tsi_pill, &RMF_SETINFO_VAL, + RCL_CLIENT); + /* Swab any part of val you need to here */ + if (KEY_IS(KEY_READ_ONLY)) { spin_lock(&req->rq_export->exp_lock); if (*(__u32 *)val) *exp_connect_flags_ptr(req->rq_export) |= @@ -1544,130 +1667,44 @@ int mdt_set_info(struct mdt_thread_info *info) *exp_connect_flags_ptr(req->rq_export) &= ~OBD_CONNECT_RDONLY; spin_unlock(&req->rq_export->exp_lock); + } else if (KEY_IS(KEY_CHANGELOG_CLEAR)) { + struct changelog_setinfo *cs = val; - } else if (KEY_IS(KEY_CHANGELOG_CLEAR)) { - struct changelog_setinfo *cs = - (struct changelog_setinfo *)val; - if (vallen != sizeof(*cs)) { - CERROR("Bad changelog_clear setinfo size %d\n", vallen); - RETURN(-EINVAL); - } - if (ptlrpc_req_need_swab(req)) { - __swab64s(&cs->cs_recno); - __swab32s(&cs->cs_id); - } - - rc = mdt_iocontrol(OBD_IOC_CHANGELOG_CLEAR, info->mti_exp, - vallen, val, NULL); - lustre_msg_set_status(req->rq_repmsg, rc); - - } else { - RETURN(-EINVAL); - } - RETURN(0); -} - -/** - * Top-level handler for MDT connection requests. - */ -int mdt_connect(struct mdt_thread_info *info) -{ - int rc; - struct obd_connect_data *reply; - struct obd_export *exp; - struct ptlrpc_request *req = mdt_info_req(info); - - rc = target_handle_connect(req); - if (rc != 0) - return err_serious(rc); + if (vallen != sizeof(*cs)) { + CERROR("%s: bad changelog_clear setinfo size %d\n", + tgt_name(tsi->tsi_tgt), vallen); + RETURN(-EINVAL); + } + if (ptlrpc_req_need_swab(req)) { + __swab64s(&cs->cs_recno); + __swab32s(&cs->cs_id); + } - LASSERT(req->rq_export != NULL); - info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev); - rc = mdt_init_sec_level(info); - if (rc != 0) { - obd_disconnect(class_export_get(req->rq_export)); - return rc; + rc = mdt_iocontrol(OBD_IOC_CHANGELOG_CLEAR, req->rq_export, + vallen, val, NULL); + } else { + RETURN(-EINVAL); } - - /* To avoid exposing partially initialized connection flags, changes up - * to this point have been staged in reply->ocd_connect_flags. Now that - * connection handling has completed successfully, atomically update - * the connect flags in the shared export data structure. LU-1623 */ - reply = req_capsule_server_get(info->mti_pill, &RMF_CONNECT_DATA); - exp = req->rq_export; - spin_lock(&exp->exp_lock); - *exp_connect_flags_ptr(exp) = reply->ocd_connect_flags; - spin_unlock(&exp->exp_lock); - - rc = mdt_init_idmap(info); - if (rc != 0) - obd_disconnect(class_export_get(req->rq_export)); - - return rc; -} - -int mdt_disconnect(struct mdt_thread_info *info) -{ - int rc; - ENTRY; - - rc = target_handle_disconnect(mdt_info_req(info)); - if (rc) - rc = err_serious(rc); - RETURN(rc); + RETURN(rc); } -static int mdt_sendpage(struct mdt_thread_info *info, - struct lu_rdpg *rdpg, int nob) +static int mdt_readpage(struct tgt_session_info *tsi) { - struct ptlrpc_request *req = mdt_info_req(info); - struct obd_export *exp = req->rq_export; - struct ptlrpc_bulk_desc *desc; - struct l_wait_info *lwi = &info->mti_u.rdpg.mti_wait_info; - int tmpcount; - int tmpsize; - int i; - int rc; - ENTRY; - - desc = ptlrpc_prep_bulk_exp(req, rdpg->rp_npages, 1, BULK_PUT_SOURCE, - MDS_BULK_PORTAL); - if (desc == NULL) - RETURN(-ENOMEM); - - if (!(exp_connect_flags(exp) & OBD_CONNECT_BRW_SIZE)) - /* old client requires reply size in it's PAGE_SIZE, - * which is rdpg->rp_count */ - nob = rdpg->rp_count; - - for (i = 0, tmpcount = nob; i < rdpg->rp_npages && tmpcount > 0; - i++, tmpcount -= tmpsize) { - tmpsize = min_t(int, tmpcount, PAGE_CACHE_SIZE); - ptlrpc_prep_bulk_page_pin(desc, rdpg->rp_pages[i], 0, tmpsize); - } - - LASSERT(desc->bd_nob == nob); - rc = target_bulk_io(exp, desc, lwi); - ptlrpc_free_bulk_pin(desc); - RETURN(rc); -} + struct mdt_thread_info *info = mdt_th_info(tsi->tsi_env); + struct mdt_object *object = mdt_obj(tsi->tsi_corpus); + struct lu_rdpg *rdpg = &info->mti_u.rdpg.mti_rdpg; + const struct mdt_body *reqbody = tsi->tsi_mdt_body; + struct mdt_body *repbody; + int rc; + int i; -int mdt_readpage(struct mdt_thread_info *info) -{ - struct mdt_object *object = info->mti_object; - struct lu_rdpg *rdpg = &info->mti_u.rdpg.mti_rdpg; - struct mdt_body *reqbody; - struct mdt_body *repbody; - int rc; - int i; - ENTRY; + ENTRY; - if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READPAGE_PACK)) - RETURN(err_serious(-ENOMEM)); + if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READPAGE_PACK)) + RETURN(err_serious(-ENOMEM)); - reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY); - repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); - if (reqbody == NULL || repbody == NULL) + repbody = req_capsule_server_get(tsi->tsi_pill, &RMF_MDT_BODY); + if (repbody == NULL || reqbody == NULL) RETURN(err_serious(-EFAULT)); /* @@ -1675,18 +1712,18 @@ int mdt_readpage(struct mdt_thread_info *info) * 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; - if (exp_connect_flags(info->mti_exp) & OBD_CONNECT_64BITHASH) + 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, - exp_max_brw_size(info->mti_exp)); + 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; OBD_ALLOC(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]); @@ -1700,36 +1737,36 @@ int mdt_readpage(struct mdt_thread_info *info) } /* call lower layers to fill allocated pages with directory data */ - rc = mo_readpage(info->mti_env, mdt_object_child(object), rdpg); - if (rc < 0) - GOTO(free_rdpg, rc); + rc = mo_readpage(tsi->tsi_env, mdt_object_child(object), rdpg); + if (rc < 0) + GOTO(free_rdpg, rc); - /* send pages to client */ - rc = mdt_sendpage(info, rdpg, rc); + /* send pages to client */ + rc = tgt_sendpage(tsi, rdpg, rc); - EXIT; + EXIT; free_rdpg: - for (i = 0; i < rdpg->rp_npages; i++) - if (rdpg->rp_pages[i] != NULL) + for (i = 0; i < rdpg->rp_npages; i++) + if (rdpg->rp_pages[i] != NULL) __free_page(rdpg->rp_pages[i]); - OBD_FREE(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]); + OBD_FREE(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]); - if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE)) - RETURN(0); + if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE)) + RETURN(0); - return rc; + return rc; } static int mdt_reint_internal(struct mdt_thread_info *info, struct mdt_lock_handle *lhc, __u32 op) { - struct req_capsule *pill = info->mti_pill; - struct mdt_body *repbody; - int rc = 0, rc2; - ENTRY; + struct req_capsule *pill = info->mti_pill; + struct mdt_body *repbody; + int rc = 0, rc2; + ENTRY; rc = mdt_reint_unpack(info, op); if (rc != 0) { @@ -1740,7 +1777,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)) @@ -1755,8 +1792,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); @@ -1775,7 +1812,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); @@ -1790,23 +1828,24 @@ out_shrink: return rc; } -static long mdt_reint_opcode(struct mdt_thread_info *info, +static long mdt_reint_opcode(struct ptlrpc_request *req, const struct req_format **fmt) { - struct mdt_rec_reint *rec; - long opc; + struct mdt_device *mdt; + struct mdt_rec_reint *rec; + long opc; - rec = req_capsule_client_get(info->mti_pill, &RMF_REC_REINT); + rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT); if (rec != NULL) { opc = rec->rr_opcode; - DEBUG_REQ(D_INODE, mdt_info_req(info), "reint opt = %ld", opc); + DEBUG_REQ(D_INODE, req, "reint opt = %ld", opc); if (opc < REINT_MAX && fmt[opc] != NULL) - req_capsule_extend(info->mti_pill, fmt[opc]); + req_capsule_extend(&req->rq_pill, fmt[opc]); else { - CERROR("%s: Unsupported opcode '%ld' from client '%s': " - "rc = %d\n", mdt_obd_name(info->mti_mdt), opc, - info->mti_mdt->mdt_ldlm_client->cli_name, - -EFAULT); + mdt = mdt_exp2dev(req->rq_export); + CERROR("%s: Unsupported opcode '%ld' from client '%s':" + " rc = %d\n", req->rq_export->exp_obd->obd_name, + opc, mdt->mdt_ldlm_client->cli_name, -EFAULT); opc = err_serious(-EFAULT); } } else { @@ -1815,11 +1854,10 @@ static long mdt_reint_opcode(struct mdt_thread_info *info, return opc; } -int mdt_reint(struct mdt_thread_info *info) +static int mdt_reint(struct tgt_session_info *tsi) { long opc; int rc; - static const struct req_format *reint_fmts[REINT_MAX] = { [REINT_SETATTR] = &RQF_MDS_REINT_SETATTR, [REINT_CREATE] = &RQF_MDS_REINT_CREATE, @@ -1828,24 +1866,27 @@ int mdt_reint(struct mdt_thread_info *info) [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; + ENTRY; - opc = mdt_reint_opcode(info, reint_fmts); - if (opc >= 0) { - /* - * No lock possible here from client to pass it to reint code - * path. - */ - rc = mdt_reint_internal(info, NULL, opc); - } else { - rc = opc; - } + opc = mdt_reint_opcode(tgt_ses_req(tsi), reint_fmts); + if (opc >= 0) { + struct mdt_thread_info *info = tsi2mdt_info(tsi); + /* + * No lock possible here from client to pass it to reint code + * path. + */ + rc = mdt_reint_internal(info, NULL, opc); + mdt_thread_info_fini(info); + } else { + rc = opc; + } - info->mti_fail_id = OBD_FAIL_MDS_REINT_NET_REP; - RETURN(rc); + tsi->tsi_reply_fail_id = OBD_FAIL_MDS_REINT_NET_REP; + RETURN(rc); } /* this should sync the whole device */ @@ -1862,108 +1903,76 @@ 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; - - 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); - - RETURN(rc); -} - -int mdt_sync(struct mdt_thread_info *info) -{ - struct ptlrpc_request *req = mdt_info_req(info); - struct req_capsule *pill = info->mti_pill; - struct mdt_body *body; - int rc; - ENTRY; - - /* The fid may be zero, so we req_capsule_set manually */ - req_capsule_set(pill, &RQF_MDS_SYNC); - - body = req_capsule_client_get(pill, &RMF_MDT_BODY); - if (body == NULL) - RETURN(err_serious(-EINVAL)); - - if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SYNC_PACK)) - RETURN(err_serious(-ENOMEM)); + struct md_object *next; + int rc; + ENTRY; - if (fid_seq(&body->fid1) == 0) { - /* sync the whole device */ - rc = req_capsule_server_pack(pill); - if (rc == 0) - rc = mdt_device_sync(info->mti_env, info->mti_mdt); - else - rc = err_serious(rc); - } else { - /* sync an object */ - rc = mdt_unpack_req_pack_rep(info, HABEO_CORPUS|HABEO_REFERO); - if (rc == 0) { - rc = mdt_object_sync(info); - if (rc == 0) { - const struct lu_fid *fid; - struct lu_attr *la = &info->mti_attr.ma_attr; - - info->mti_attr.ma_need = MA_INODE; - info->mti_attr.ma_valid = 0; - rc = mdt_attr_get_complex(info, info->mti_object, - &info->mti_attr); - if (rc == 0) { - body = req_capsule_server_get(pill, - &RMF_MDT_BODY); - fid = mdt_object_fid(info->mti_object); - mdt_pack_attr2body(info, body, la, fid); - } - } - } else - rc = err_serious(rc); - } - if (rc == 0) - mdt_counter_incr(req, LPROC_MDT_SYNC); + 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); } -/* - * Quotacheck handler. - * in-kernel quotacheck isn't supported any more. - */ -int mdt_quotacheck(struct mdt_thread_info *info) +static int mdt_sync(struct tgt_session_info *tsi) { - struct obd_quotactl *oqctl; + struct ptlrpc_request *req = tgt_ses_req(tsi); + struct req_capsule *pill = tsi->tsi_pill; + struct mdt_body *body; int rc; + ENTRY; - oqctl = req_capsule_client_get(info->mti_pill, &RMF_OBD_QUOTACTL); - if (oqctl == NULL) - RETURN(err_serious(-EPROTO)); + if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SYNC_PACK)) + RETURN(err_serious(-ENOMEM)); - rc = req_capsule_server_pack(info->mti_pill); - if (rc) - RETURN(err_serious(rc)); + 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); + + /* sync an object */ + rc = mdt_object_sync(info); + if (rc == 0) { + const struct lu_fid *fid; + struct lu_attr *la = &info->mti_attr.ma_attr; + + info->mti_attr.ma_need = MA_INODE; + info->mti_attr.ma_valid = 0; + rc = mdt_attr_get_complex(info, info->mti_object, + &info->mti_attr); + if (rc == 0) { + body = req_capsule_server_get(pill, + &RMF_MDT_BODY); + fid = mdt_object_fid(info->mti_object); + mdt_pack_attr2body(info, body, la, fid); + } + } + mdt_thread_info_fini(info); + } + if (rc == 0) + mdt_counter_incr(req, LPROC_MDT_SYNC); - /* deprecated, not used any more */ - RETURN(-EOPNOTSUPP); + RETURN(rc); } /* * Handle quota control requests to consult current usage/limit, but also * to configure quota enforcement */ -int mdt_quotactl(struct mdt_thread_info *info) +static int mdt_quotactl(struct tgt_session_info *tsi) { - struct obd_export *exp = info->mti_exp; - struct req_capsule *pill = info->mti_pill; + struct obd_export *exp = tsi->tsi_exp; + struct req_capsule *pill = tsi->tsi_pill; struct obd_quotactl *oqctl, *repoqc; int id, rc; - struct lu_device *qmt = info->mti_mdt->mdt_qmt_dev; + struct mdt_device *mdt = mdt_exp2dev(exp); + struct lu_device *qmt = mdt->mdt_qmt_dev; ENTRY; oqctl = req_capsule_client_get(pill, &RMF_OBD_QUOTACTL); @@ -2004,7 +2013,7 @@ int mdt_quotactl(struct mdt_thread_info *info) if (exp_connect_rmtclient(exp)) { struct lustre_idmap_table *idmap; - idmap = mdt_req2med(mdt_info_req(info))->med_idmap; + idmap = exp->exp_mdt_data.med_idmap; if (unlikely(oqctl->qc_cmd != Q_GETQUOTA && oqctl->qc_cmd != Q_GETINFO)) @@ -2039,13 +2048,13 @@ int mdt_quotactl(struct mdt_thread_info *info) case Q_SETQUOTA: case Q_GETQUOTA: /* forward quotactl request to QMT */ - rc = qmt_hdls.qmth_quotactl(info->mti_env, qmt, oqctl); + rc = qmt_hdls.qmth_quotactl(tsi->tsi_env, qmt, oqctl); break; case Q_GETOINFO: case Q_GETOQUOTA: /* slave quotactl */ - rc = lquotactl_slv(info->mti_env, info->mti_mdt->mdt_bottom, + rc = lquotactl_slv(tsi->tsi_env, tsi->tsi_tgt->lut_bottom, oqctl); break; @@ -2061,120 +2070,6 @@ int mdt_quotactl(struct mdt_thread_info *info) RETURN(rc); } -/* - * OBD PING and other handlers. - */ -int mdt_obd_ping(struct mdt_thread_info *info) -{ - int rc; - ENTRY; - - req_capsule_set(info->mti_pill, &RQF_OBD_PING); - - rc = target_handle_ping(mdt_info_req(info)); - if (rc < 0) - rc = err_serious(rc); - RETURN(rc); -} - -/* - * OBD_IDX_READ handler - */ -int mdt_obd_idx_read(struct mdt_thread_info *info) -{ - struct mdt_device *mdt = info->mti_mdt; - struct lu_rdpg *rdpg = &info->mti_u.rdpg.mti_rdpg; - struct idx_info *req_ii, *rep_ii; - int rc, i; - ENTRY; - - memset(rdpg, 0, sizeof(*rdpg)); - req_capsule_set(info->mti_pill, &RQF_OBD_IDX_READ); - - /* extract idx_info buffer from request & reply */ - req_ii = req_capsule_client_get(info->mti_pill, &RMF_IDX_INFO); - if (req_ii == NULL || req_ii->ii_magic != IDX_INFO_MAGIC) - RETURN(err_serious(-EPROTO)); - - rc = req_capsule_server_pack(info->mti_pill); - if (rc) - RETURN(err_serious(rc)); - - rep_ii = req_capsule_server_get(info->mti_pill, &RMF_IDX_INFO); - if (rep_ii == NULL) - RETURN(err_serious(-EFAULT)); - rep_ii->ii_magic = IDX_INFO_MAGIC; - - /* extract hash to start with */ - rdpg->rp_hash = req_ii->ii_hash_start; - - /* extract requested attributes */ - rdpg->rp_attrs = req_ii->ii_attrs; - - /* check that fid packed in request is valid and supported */ - if (!fid_is_sane(&req_ii->ii_fid)) - RETURN(-EINVAL); - rep_ii->ii_fid = req_ii->ii_fid; - - /* copy flags */ - rep_ii->ii_flags = req_ii->ii_flags; - - /* compute number of pages to allocate, ii_count is the number of 4KB - * containers */ - if (req_ii->ii_count <= 0) - GOTO(out, rc = -EFAULT); - rdpg->rp_count = min_t(unsigned int, req_ii->ii_count << LU_PAGE_SHIFT, - exp_max_brw_size(info->mti_exp)); - rdpg->rp_npages = (rdpg->rp_count + PAGE_CACHE_SIZE - 1) >> - PAGE_CACHE_SHIFT; - - /* allocate pages to store the containers */ - OBD_ALLOC(rdpg->rp_pages, rdpg->rp_npages * sizeof(rdpg->rp_pages[0])); - if (rdpg->rp_pages == NULL) - GOTO(out, rc = -ENOMEM); - for (i = 0; i < rdpg->rp_npages; i++) { - rdpg->rp_pages[i] = alloc_page(GFP_IOFS); - if (rdpg->rp_pages[i] == NULL) - GOTO(out, rc = -ENOMEM); - } - - /* populate pages with key/record pairs */ - rc = dt_index_read(info->mti_env, mdt->mdt_bottom, rep_ii, rdpg); - if (rc < 0) - GOTO(out, rc); - - LASSERTF(rc <= rdpg->rp_count, "dt_index_read() returned more than " - "asked %d > %d\n", rc, rdpg->rp_count); - - /* send pages to client */ - rc = mdt_sendpage(info, rdpg, rc); - - GOTO(out, rc); -out: - if (rdpg->rp_pages) { - for (i = 0; i < rdpg->rp_npages; i++) - if (rdpg->rp_pages[i]) - __free_page(rdpg->rp_pages[i]); - OBD_FREE(rdpg->rp_pages, - rdpg->rp_npages * sizeof(rdpg->rp_pages[0])); - } - return rc; -} - -int mdt_obd_log_cancel(struct mdt_thread_info *info) -{ - return err_serious(-EOPNOTSUPP); -} - -int mdt_obd_qc_callback(struct mdt_thread_info *info) -{ - return err_serious(-EOPNOTSUPP); -} - -/* - * LLOG handlers. - */ - /** clone llog ctxt from child (mdd) * This allows remote llog (replicator) access. * We can either pass all llog RPCs (eg mdt_llog_create) on to child where the @@ -2219,150 +2114,45 @@ static int mdt_llog_ctxt_unclone(const struct lu_env *env, return 0; } -int mdt_llog_create(struct mdt_thread_info *info) -{ - int rc; - - req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_CREATE); - rc = llog_origin_handle_open(mdt_info_req(info)); - return (rc < 0 ? err_serious(rc) : rc); -} - -int mdt_llog_destroy(struct mdt_thread_info *info) -{ - int rc; - - req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_DESTROY); - rc = llog_origin_handle_destroy(mdt_info_req(info)); - return (rc < 0 ? err_serious(rc) : rc); -} - -int mdt_llog_read_header(struct mdt_thread_info *info) -{ - int rc; - - req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_READ_HEADER); - rc = llog_origin_handle_read_header(mdt_info_req(info)); - return (rc < 0 ? err_serious(rc) : rc); -} - -int mdt_llog_next_block(struct mdt_thread_info *info) -{ - int rc; - - req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_NEXT_BLOCK); - rc = llog_origin_handle_next_block(mdt_info_req(info)); - return (rc < 0 ? err_serious(rc) : rc); -} - -int mdt_llog_prev_block(struct mdt_thread_info *info) -{ - int rc; - - req_capsule_set(info->mti_pill, &RQF_LLOG_ORIGIN_HANDLE_PREV_BLOCK); - rc = llog_origin_handle_prev_block(mdt_info_req(info)); - return (rc < 0 ? err_serious(rc) : rc); -} - - -/* - * DLM handlers. - */ - -static struct ldlm_callback_suite cbs = { - .lcs_completion = ldlm_server_completion_ast, - .lcs_blocking = ldlm_server_blocking_ast, - .lcs_glimpse = ldlm_server_glimpse_ast -}; - -int mdt_enqueue(struct mdt_thread_info *info) -{ - struct ptlrpc_request *req; - int rc; - - /* - * info->mti_dlm_req already contains swapped and (if necessary) - * converted dlm request. - */ - LASSERT(info->mti_dlm_req != NULL); - - req = mdt_info_req(info); - rc = ldlm_handle_enqueue0(info->mti_mdt->mdt_namespace, - req, info->mti_dlm_req, &cbs); - info->mti_fail_id = OBD_FAIL_LDLM_REPLY; - return rc ? err_serious(rc) : req->rq_status; -} - -int mdt_convert(struct mdt_thread_info *info) -{ - int rc; - struct ptlrpc_request *req; - - LASSERT(info->mti_dlm_req); - req = mdt_info_req(info); - rc = ldlm_handle_convert0(req, info->mti_dlm_req); - return rc ? err_serious(rc) : req->rq_status; -} - -int mdt_bl_callback(struct mdt_thread_info *info) -{ - CERROR("bl callbacks should not happen on MDS\n"); - LBUG(); - return err_serious(-EOPNOTSUPP); -} - -int mdt_cp_callback(struct mdt_thread_info *info) -{ - CERROR("cp callbacks should not happen on MDS\n"); - LBUG(); - return err_serious(-EOPNOTSUPP); -} - /* * sec context handlers */ -int mdt_sec_ctx_handle(struct mdt_thread_info *info) +static int mdt_sec_ctx_handle(struct tgt_session_info *tsi) { - int rc; - - rc = mdt_handle_idmap(info); + int rc; - if (unlikely(rc)) { - struct ptlrpc_request *req = mdt_info_req(info); - __u32 opc; + rc = mdt_handle_idmap(tsi); + if (unlikely(rc)) { + struct ptlrpc_request *req = tgt_ses_req(tsi); + __u32 opc; - opc = lustre_msg_get_opc(req->rq_reqmsg); - if (opc == SEC_CTX_INIT || opc == SEC_CTX_INIT_CONT) - sptlrpc_svc_ctx_invalidate(req); - } + opc = lustre_msg_get_opc(req->rq_reqmsg); + if (opc == SEC_CTX_INIT || opc == SEC_CTX_INIT_CONT) + sptlrpc_svc_ctx_invalidate(req); + } - CFS_FAIL_TIMEOUT(OBD_FAIL_SEC_CTX_HDL_PAUSE, cfs_fail_val); + CFS_FAIL_TIMEOUT(OBD_FAIL_SEC_CTX_HDL_PAUSE, cfs_fail_val); - return rc; + return rc; } /* * quota request handlers */ -int mdt_quota_dqacq(struct mdt_thread_info *info) +static int mdt_quota_dqacq(struct tgt_session_info *tsi) { - struct lu_device *qmt = info->mti_mdt->mdt_qmt_dev; + struct mdt_device *mdt = mdt_exp2dev(tsi->tsi_exp); + struct lu_device *qmt = mdt->mdt_qmt_dev; int rc; ENTRY; if (qmt == NULL) RETURN(err_serious(-EOPNOTSUPP)); - rc = qmt_hdls.qmth_dqacq(info->mti_env, qmt, mdt_info_req(info)); + rc = qmt_hdls.qmth_dqacq(tsi->tsi_env, qmt, tgt_ses_req(tsi)); RETURN(rc); } -static struct mdt_object *mdt_obj(struct lu_object *o) -{ - LASSERT(lu_device_is_mdt(o->lo_dev)); - return container_of0(o, struct mdt_object, mot_obj); -} - struct mdt_object *mdt_object_new(const struct lu_env *env, struct mdt_device *d, const struct lu_fid *f) @@ -2410,12 +2200,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); } /** @@ -2490,10 +2281,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); @@ -2501,8 +2293,9 @@ int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, RETURN(rc); } -int mdt_md_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, - void *data, int flag) +/* Used for cross-MDT lock */ +int mdt_remote_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, + void *data, int flag) { struct lustre_handle lockh; int rc; @@ -2525,24 +2318,65 @@ int mdt_md_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); + 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"\n", + lhc->mlh_reg_lh.cookie); + RETURN(-ESTALE); + } + + 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; einfo->ei_mode = mode; - einfo->ei_cb_bl = mdt_md_blocking_ast; + 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; @@ -2552,9 +2386,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; @@ -2568,21 +2403,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); - ibits |= MDS_INODELOCK_LOOKUP; - } else { - LASSERTF(!(ibits & - (MDS_INODELOCK_UPDATE | MDS_INODELOCK_PERM)), - "%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 */ @@ -2617,11 +2440,12 @@ 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); - if (unlikely(rc)) - RETURN(rc); + 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 != 0)) + GOTO(out_unlock, rc); } /* @@ -2638,24 +2462,81 @@ 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); - if (rc) - mdt_object_unlock(info, o, lh, 1); - else if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_PDO_LOCK)) && - lh->mlh_pdo_hash != 0 && - (lh->mlh_reg_mode == LCK_PW || lh->mlh_reg_mode == LCK_EX)) { - OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PDO_LOCK, 15); - } + 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); +out_unlock: + if (rc != 0) + mdt_object_unlock(info, o, lh, 1); + else if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_PDO_LOCK)) && + lh->mlh_pdo_hash != 0 && + (lh->mlh_reg_mode == LCK_PW || lh->mlh_reg_mode == LCK_EX)) + OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PDO_LOCK, 15); - RETURN(rc); + 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, @@ -2664,7 +2545,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; @@ -2702,15 +2583,21 @@ void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h, LASSERTF(lock != NULL, "no lock for cookie "LPX64"\n", h->cookie); - 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; - ldlm_lock_downgrade(lock, LCK_COS); - mode = LCK_COS; - } - ptlrpc_save_lock(req, h, mode, no_ack); + /* 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; + ldlm_lock_downgrade(lock, LCK_COS); + mode = LCK_COS; + } + ptlrpc_save_lock(req, h, mode, no_ack); + } else { + ldlm_lock_decref(h, mode); + } if (mdt_is_lock_sync(lock)) { CDEBUG(D_HA, "found sync-lock," " async commit started\n"); @@ -2781,40 +2668,6 @@ void mdt_object_unlock_put(struct mdt_thread_info * info, mdt_object_put(info->mti_env, o); } -struct mdt_handler *mdt_handler_find(__u32 opc, struct mdt_opc_slice *supported) -{ - struct mdt_opc_slice *s; - struct mdt_handler *h; - - h = NULL; - for (s = supported; s->mos_hs != NULL; s++) { - if (s->mos_opc_start <= opc && opc < s->mos_opc_end) { - h = s->mos_hs + (opc - s->mos_opc_start); - if (likely(h->mh_opc != 0)) - LASSERTF(h->mh_opc == opc, - "opcode mismatch %d != %d\n", - h->mh_opc, opc); - else - h = NULL; /* unsupported opc */ - break; - } - } - return h; -} - -static int mdt_lock_resname_compat(struct mdt_device *m, - struct ldlm_request *req) -{ - /* XXX something... later. */ - return 0; -} - -static int mdt_lock_reply_compat(struct mdt_device *m, struct ldlm_reply *rep) -{ - /* XXX something... later. */ - return 0; -} - /* * Generic code handling requests that have struct mdt_body passed in: * @@ -2843,11 +2696,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); } @@ -2858,17 +2711,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; @@ -2897,7 +2750,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); @@ -2909,150 +2762,13 @@ static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags) static int mdt_init_capa_ctxt(const struct lu_env *env, struct mdt_device *m) { - struct md_device *next = m->mdt_child; + struct md_device *next = m->mdt_child; - return next->md_ops->mdo_init_capa_ctxt(env, next, - m->mdt_opts.mo_mds_capa, - m->mdt_capa_timeout, - m->mdt_capa_alg, - m->mdt_capa_keys); -} - -/* - * Invoke handler for this request opc. Also do necessary preprocessing - * (according to handler ->mh_flags), and post-processing (setting of - * ->last_{xid,committed}). - */ -static int mdt_req_handle(struct mdt_thread_info *info, - struct mdt_handler *h, struct ptlrpc_request *req) -{ - int rc, serious = 0; - __u32 flags; - - ENTRY; - - LASSERT(h->mh_act != NULL); - LASSERT(h->mh_opc == lustre_msg_get_opc(req->rq_reqmsg)); - LASSERT(current->journal_info == NULL); - - /* - * Checking for various OBD_FAIL_$PREF_$OPC_NET codes. _Do_ not try - * to put same checks into handlers like mdt_close(), mdt_reint(), - * etc., without talking to mdt authors first. Checking same thing - * there again is useless and returning 0 error without packing reply - * is buggy! Handlers either pack reply or return error. - * - * We return 0 here and do not send any reply in order to emulate - * network failure. Do not send any reply in case any of NET related - * fail_id has occured. - */ - if (OBD_FAIL_CHECK_ORSET(h->mh_fail_id, OBD_FAIL_ONCE)) - RETURN(0); - - rc = 0; - flags = h->mh_flags; - LASSERT(ergo(flags & (HABEO_CORPUS|HABEO_REFERO), h->mh_fmt != NULL)); - - if (h->mh_fmt != NULL) { - req_capsule_set(info->mti_pill, h->mh_fmt); - rc = mdt_unpack_req_pack_rep(info, flags); - } - - if (rc == 0 && flags & MUTABOR && - exp_connect_flags(req->rq_export) & OBD_CONNECT_RDONLY) - /* should it be rq_status? */ - rc = -EROFS; - - if (rc == 0 && flags & HABEO_CLAVIS) { - struct ldlm_request *dlm_req; - - LASSERT(h->mh_fmt != NULL); - - dlm_req = req_capsule_client_get(info->mti_pill, &RMF_DLM_REQ); - if (dlm_req != NULL) { - if (unlikely(dlm_req->lock_desc.l_resource.lr_type == - LDLM_IBITS && - dlm_req->lock_desc.l_policy_data.\ - l_inodebits.bits == 0)) { - /* - * Lock without inodebits makes no sense and - * will oops later in ldlm. If client miss to - * set such bits, do not trigger ASSERTION. - * - * For liblustre flock case, it maybe zero. - */ - rc = -EPROTO; - } else { - if (info->mti_mdt->mdt_opts.mo_compat_resname) - rc = mdt_lock_resname_compat( - info->mti_mdt, - dlm_req); - info->mti_dlm_req = dlm_req; - } - } else { - rc = -EFAULT; - } - } - - /* capability setting changed via /proc, needs reinitialize ctxt */ - if (info->mti_mdt && info->mti_mdt->mdt_capa_conf) { - mdt_init_capa_ctxt(info->mti_env, info->mti_mdt); - info->mti_mdt->mdt_capa_conf = 0; - } - - if (likely(rc == 0)) { - /* - * Process request, there can be two types of rc: - * 1) errors with msg unpack/pack, other failures outside the - * operation itself. This is counted as serious errors; - * 2) errors during fs operation, should be placed in rq_status - * only - */ - rc = h->mh_act(info); - if (rc == 0 && - !req->rq_no_reply && req->rq_reply_state == NULL) { - DEBUG_REQ(D_ERROR, req, "MDT \"handler\" %s did not " - "pack reply and returned 0 error\n", - h->mh_name); - LBUG(); - } - serious = is_serious(rc); - rc = clear_serious(rc); - } else - serious = 1; - - req->rq_status = rc; - - /* - * ELDLM_* codes which > 0 should be in rq_status only as well as - * all non-serious errors. - */ - if (rc > 0 || !serious) - rc = 0; - - LASSERT(current->journal_info == NULL); - - if (rc == 0 && (flags & HABEO_CLAVIS) && - info->mti_mdt->mdt_opts.mo_compat_resname) { - struct ldlm_reply *dlmrep; - - dlmrep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP); - if (dlmrep != NULL) - rc = mdt_lock_reply_compat(info->mti_mdt, dlmrep); - } - - /* If we're DISCONNECTing, the mdt_export_data is already freed */ - if (likely(rc == 0 && req->rq_export && h->mh_opc != MDS_DISCONNECT)) - target_committed_to_req(req); - - if (unlikely(req_is_replay(req) && - lustre_msg_get_transno(req->rq_reqmsg) == 0)) { - DEBUG_REQ(D_ERROR, req, "transno is 0 during REPLAY"); - LBUG(); - } - - target_send_reply(req, rc, info->mti_fail_id); - RETURN(0); + 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) @@ -3077,366 +2793,117 @@ void mdt_lock_handle_fini(struct mdt_lock_handle *lh) * uninitialized state, because it's too expensive to zero out whole * mdt_thread_info (> 1K) on each request arrival. */ -static void mdt_thread_info_init(struct ptlrpc_request *req, - struct mdt_thread_info *info) +void mdt_thread_info_init(struct ptlrpc_request *req, + struct mdt_thread_info *info) { int i; - req_capsule_init(&req->rq_pill, req, RCL_SERVER); info->mti_pill = &req->rq_pill; /* lock handle */ - for (i = 0; i < ARRAY_SIZE(info->mti_lh); i++) - mdt_lock_handle_init(&info->mti_lh[i]); - - /* mdt device: it can be NULL while CONNECT */ - if (req->rq_export) { - info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev); - info->mti_exp = req->rq_export; - } else - info->mti_mdt = NULL; - info->mti_env = req->rq_svc_thread->t_env; - info->mti_fail_id = OBD_FAIL_MDS_ALL_REPLY_NET; - info->mti_transno = lustre_msg_get_transno(req->rq_reqmsg); - info->mti_mos = NULL; - - memset(&info->mti_attr, 0, sizeof(info->mti_attr)); - info->mti_big_buf = LU_BUF_NULL; - info->mti_body = NULL; - info->mti_object = NULL; - info->mti_dlm_req = NULL; - info->mti_has_trans = 0; - info->mti_cross_ref = 0; - info->mti_opdata = 0; - info->mti_big_lmm_used = 0; - - /* To not check for split by default. */ - info->mti_spec.no_create = 0; - info->mti_spec.sp_rm_entry = 0; -} - -static void mdt_thread_info_fini(struct mdt_thread_info *info) -{ - int i; - - req_capsule_fini(info->mti_pill); - if (info->mti_object != NULL) { - mdt_object_put(info->mti_env, info->mti_object); - info->mti_object = NULL; - } - - for (i = 0; i < ARRAY_SIZE(info->mti_lh); i++) - mdt_lock_handle_fini(&info->mti_lh[i]); - info->mti_env = NULL; - - if (unlikely(info->mti_big_buf.lb_buf != NULL)) - lu_buf_free(&info->mti_big_buf); -} - -static int mdt_filter_recovery_request(struct ptlrpc_request *req, - struct obd_device *obd, int *process) -{ - switch (lustre_msg_get_opc(req->rq_reqmsg)) { - case MDS_CONNECT: /* This will never get here, but for completeness. */ - case OST_CONNECT: /* This will never get here, but for completeness. */ - case MDS_DISCONNECT: - case OST_DISCONNECT: - case OBD_IDX_READ: - *process = 1; - RETURN(0); - - case MDS_CLOSE: - case MDS_DONE_WRITING: - case MDS_SYNC: /* used in unmounting */ - case OBD_PING: - case MDS_REINT: - case UPDATE_OBJ: - case SEQ_QUERY: - case FLD_QUERY: - case LDLM_ENQUEUE: - *process = target_queue_recovery_request(req, obd); - RETURN(0); - - default: - DEBUG_REQ(D_ERROR, req, "not permitted during recovery"); - *process = -EAGAIN; - RETURN(0); - } -} - -/* - * Handle recovery. Return: - * +1: continue request processing; - * -ve: abort immediately with the given error code; - * 0: send reply with error code in req->rq_status; - */ -static int mdt_recovery(struct mdt_thread_info *info) -{ - struct ptlrpc_request *req = mdt_info_req(info); - struct obd_device *obd; - - ENTRY; - - switch (lustre_msg_get_opc(req->rq_reqmsg)) { - case MDS_CONNECT: - case SEC_CTX_INIT: - case SEC_CTX_INIT_CONT: - case SEC_CTX_FINI: - { -#if 0 - int rc; - - rc = mdt_handle_idmap(info); - if (rc) - RETURN(rc); - else -#endif - RETURN(+1); - } - } - - if (unlikely(!class_connected_export(req->rq_export))) { - CDEBUG(D_HA, "operation %d on unconnected MDS from %s\n", - lustre_msg_get_opc(req->rq_reqmsg), - libcfs_id2str(req->rq_peer)); - /* FIXME: For CMD cleanup, when mds_B stop, the req from - * mds_A will get -ENOTCONN(especially for ping req), - * which will cause that mds_A deactive timeout, then when - * mds_A cleanup, the cleanup process will be suspended since - * deactive timeout is not zero. - */ - req->rq_status = -ENOTCONN; - target_send_reply(req, -ENOTCONN, info->mti_fail_id); - RETURN(0); - } - - /* sanity check: if the xid matches, the request must be marked as a - * resent or replayed */ - if (req_xid_is_last(req)) { - if (!(lustre_msg_get_flags(req->rq_reqmsg) & - (MSG_RESENT | MSG_REPLAY))) { - DEBUG_REQ(D_WARNING, req, "rq_xid "LPU64" matches last_xid, " - "expected REPLAY or RESENT flag (%x)", req->rq_xid, - lustre_msg_get_flags(req->rq_reqmsg)); - LBUG(); - req->rq_status = -ENOTCONN; - RETURN(-ENOTCONN); - } - } - - /* else: note the opposite is not always true; a RESENT req after a - * failover will usually not match the last_xid, since it was likely - * never committed. A REPLAYed request will almost never match the - * last xid, however it could for a committed, but still retained, - * open. */ + for (i = 0; i < ARRAY_SIZE(info->mti_lh); i++) + mdt_lock_handle_init(&info->mti_lh[i]); - obd = req->rq_export->exp_obd; + /* mdt device: it can be NULL while CONNECT */ + if (req->rq_export) { + info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev); + info->mti_exp = req->rq_export; + } else + info->mti_mdt = NULL; + info->mti_env = req->rq_svc_thread->t_env; + info->mti_transno = lustre_msg_get_transno(req->rq_reqmsg); - /* Check for aborted recovery... */ - if (unlikely(obd->obd_recovering)) { - int rc; - int should_process; - DEBUG_REQ(D_INFO, req, "Got new replay"); - rc = mdt_filter_recovery_request(req, obd, &should_process); - if (rc != 0 || !should_process) - RETURN(rc); - else if (should_process < 0) { - req->rq_status = should_process; - rc = ptlrpc_error(req); - RETURN(rc); - } - } - RETURN(+1); -} + memset(&info->mti_attr, 0, sizeof(info->mti_attr)); + info->mti_big_buf = LU_BUF_NULL; + info->mti_body = NULL; + info->mti_object = NULL; + info->mti_dlm_req = NULL; + info->mti_has_trans = 0; + info->mti_cross_ref = 0; + info->mti_opdata = 0; + info->mti_big_lmm_used = 0; -static int mdt_msg_check_version(struct lustre_msg *msg) -{ - int rc; + info->mti_spec.no_create = 0; + info->mti_spec.sp_rm_entry = 0; - switch (lustre_msg_get_opc(msg)) { - case MDS_CONNECT: - case MDS_DISCONNECT: - case OBD_PING: - case SEC_CTX_INIT: - case SEC_CTX_INIT_CONT: - case SEC_CTX_FINI: - case OBD_IDX_READ: - rc = lustre_msg_check_version(msg, LUSTRE_OBD_VERSION); - if (rc) - CERROR("bad opc %u version %08x, expecting %08x\n", - lustre_msg_get_opc(msg), - lustre_msg_get_version(msg), - LUSTRE_OBD_VERSION); - break; - case MDS_GETSTATUS: - case MDS_GETATTR: - case MDS_GETATTR_NAME: - case MDS_STATFS: - case MDS_READPAGE: - case MDS_WRITEPAGE: - case MDS_IS_SUBDIR: - case MDS_REINT: - case MDS_CLOSE: - case MDS_DONE_WRITING: - case MDS_PIN: - case MDS_SYNC: - case MDS_GETXATTR: - case MDS_SETXATTR: - case MDS_SET_INFO: - case MDS_GET_INFO: - case MDS_HSM_PROGRESS: - case MDS_HSM_REQUEST: - case MDS_HSM_CT_REGISTER: - case MDS_HSM_CT_UNREGISTER: - case MDS_HSM_STATE_GET: - case MDS_HSM_STATE_SET: - case MDS_HSM_ACTION: - case MDS_QUOTACHECK: - case MDS_QUOTACTL: - case UPDATE_OBJ: - case MDS_SWAP_LAYOUTS: - case QUOTA_DQACQ: - case QUOTA_DQREL: - case SEQ_QUERY: - case FLD_QUERY: - rc = lustre_msg_check_version(msg, LUSTRE_MDS_VERSION); - if (rc) - CERROR("bad opc %u version %08x, expecting %08x\n", - lustre_msg_get_opc(msg), - lustre_msg_get_version(msg), - LUSTRE_MDS_VERSION); - break; - case LDLM_ENQUEUE: - case LDLM_CONVERT: - case LDLM_BL_CALLBACK: - case LDLM_CP_CALLBACK: - rc = lustre_msg_check_version(msg, LUSTRE_DLM_VERSION); - if (rc) - CERROR("bad opc %u version %08x, expecting %08x\n", - lustre_msg_get_opc(msg), - lustre_msg_get_version(msg), - LUSTRE_DLM_VERSION); - break; - case OBD_LOG_CANCEL: - case LLOG_ORIGIN_HANDLE_CREATE: - case LLOG_ORIGIN_HANDLE_NEXT_BLOCK: - case LLOG_ORIGIN_HANDLE_READ_HEADER: - case LLOG_ORIGIN_HANDLE_CLOSE: - case LLOG_ORIGIN_HANDLE_DESTROY: - case LLOG_ORIGIN_HANDLE_PREV_BLOCK: - case LLOG_CATINFO: - rc = lustre_msg_check_version(msg, LUSTRE_LOG_VERSION); - if (rc) - CERROR("bad opc %u version %08x, expecting %08x\n", - lustre_msg_get_opc(msg), - lustre_msg_get_version(msg), - LUSTRE_LOG_VERSION); - break; - default: - CERROR("MDS unknown opcode %d\n", lustre_msg_get_opc(msg)); - rc = -ENOTSUPP; - } - return rc; + info->mti_spec.u.sp_ea.eadata = NULL; + info->mti_spec.u.sp_ea.eadatalen = 0; } -static int mdt_handle0(struct ptlrpc_request *req, - struct mdt_thread_info *info, - struct mdt_opc_slice *supported) +void mdt_thread_info_fini(struct mdt_thread_info *info) { - struct mdt_handler *h; - struct lustre_msg *msg; - int rc; + int i; - ENTRY; + if (info->mti_object != NULL) { + mdt_object_put(info->mti_env, info->mti_object); + info->mti_object = NULL; + } - if (OBD_FAIL_CHECK_ORSET(OBD_FAIL_MDS_ALL_REQUEST_NET, OBD_FAIL_ONCE)) - RETURN(0); + for (i = 0; i < ARRAY_SIZE(info->mti_lh); i++) + mdt_lock_handle_fini(&info->mti_lh[i]); + info->mti_env = NULL; + info->mti_pill = NULL; + info->mti_exp = NULL; - LASSERT(current->journal_info == NULL); - - msg = req->rq_reqmsg; - rc = mdt_msg_check_version(msg); - if (likely(rc == 0)) { - rc = mdt_recovery(info); - if (likely(rc == +1)) { - h = mdt_handler_find(lustre_msg_get_opc(msg), - supported); - if (likely(h != NULL)) { - rc = mdt_req_handle(info, h, req); - } else { - CERROR("%s: opc unsupported: 0x%x\n", - mdt_obd_name(info->mti_mdt), - lustre_msg_get_opc(msg)); - req->rq_status = -ENOTSUPP; - rc = ptlrpc_error(req); - RETURN(rc); - } - } - } else { - CDEBUG(D_INFO, "%s: drops mal-formed request: rc = %d\n", - mdt_obd_name(info->mti_mdt), rc); - req->rq_status = rc; - rc = ptlrpc_error(req); - } - RETURN(rc); + if (unlikely(info->mti_big_buf.lb_buf != NULL)) + lu_buf_free(&info->mti_big_buf); } -/* - * MDT handler function called by ptlrpc service thread when request comes. - * - * XXX common "target" functionality should be factored into separate module - * shared by mdt, ost and stand-alone services like fld. - */ -int mdt_handle_common(struct ptlrpc_request *req, - struct mdt_opc_slice *supported) +struct mdt_thread_info *tsi2mdt_info(struct tgt_session_info *tsi) { - struct lu_env *env; - struct mdt_thread_info *info; - int rc; - ENTRY; + struct mdt_thread_info *mti; + struct lustre_capa *lc; - env = req->rq_svc_thread->t_env; - /* Refill(initilize) the context(mdt_thread_info), in case it is - * not initialized yet. Usually it happens during start up, after - * MDS(ptlrpc threads) is start up, it gets the first CONNECT request, - * before MDT_thread_info is initialized */ - lu_env_refill(env); - LASSERT(env != NULL); - LASSERT(env->le_ses != NULL); - LASSERT(env->le_ctx.lc_thread == req->rq_svc_thread); - info = lu_context_key_get(&env->le_ctx, &mdt_thread_key); - LASSERT(info != NULL); + mti = mdt_th_info(tsi->tsi_env); + LASSERT(mti != NULL); - mdt_thread_info_init(req, info); + mdt_thread_info_init(tgt_ses_req(tsi), mti); + if (tsi->tsi_corpus != NULL) { + struct req_capsule *pill = tsi->tsi_pill; - rc = mdt_handle0(req, info, supported); + mti->mti_object = mdt_obj(tsi->tsi_corpus); + lu_object_get(tsi->tsi_corpus); - mdt_thread_info_fini(info); - RETURN(rc); + /* + * 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; + + return mti; } -/* - * This is called from recovery code as handler of _all_ RPC types, FLD and SEQ - * as well. - */ -int mdt_recovery_handle(struct ptlrpc_request *req) +static int mdt_tgt_connect(struct tgt_session_info *tsi) { - int rc; - ENTRY; + struct ptlrpc_request *req = tgt_ses_req(tsi); + int rc; - switch (lustre_msg_get_opc(req->rq_reqmsg)) { - case FLD_QUERY: - rc = mdt_handle_common(req, mdt_fld_handlers); - break; - case SEQ_QUERY: - rc = mdt_handle_common(req, mdt_seq_handlers); - break; - default: - rc = mdt_handle_common(req, mdt_regular_handlers); - break; - } + ENTRY; - RETURN(rc); + 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); + + rc = mdt_init_idmap(tsi); + if (rc != 0) + GOTO(err, rc); + RETURN(0); +err: + obd_disconnect(class_export_get(req->rq_export)); + return rc; } enum mdt_it_code { @@ -3458,6 +2925,12 @@ static int mdt_intent_getattr(enum mdt_it_code opcode, struct mdt_thread_info *info, struct ldlm_lock **, __u64); + +static int mdt_intent_getxattr(enum mdt_it_code opcode, + struct mdt_thread_info *info, + struct ldlm_lock **lockp, + __u64 flags); + static int mdt_intent_layout(enum mdt_it_code opcode, struct mdt_thread_info *info, struct ldlm_lock **, @@ -3485,7 +2958,14 @@ static struct mdt_it_flavor { }, [MDT_IT_OCREAT] = { .it_fmt = &RQF_LDLM_INTENT, - .it_flags = MUTABOR, + /* + * OCREAT is not a MUTABOR request as if the file + * already exists. + * We do the extra check of OBD_CONNECT_RDONLY in + * mdt_reint_open() when we really need to create + * the object. + */ + .it_flags = 0, .it_act = mdt_intent_reint, .it_reint = REINT_OPEN }, @@ -3522,9 +3002,9 @@ static struct mdt_it_flavor { .it_act = NULL }, [MDT_IT_GETXATTR] = { - .it_fmt = NULL, - .it_flags = 0, - .it_act = NULL + .it_fmt = &RQF_LDLM_INTENT_GETXATTR, + .it_flags = HABEO_CORPUS, + .it_act = mdt_intent_getxattr }, [MDT_IT_LAYOUT] = { .it_fmt = &RQF_LDLM_INTENT_LAYOUT, @@ -3533,21 +3013,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; @@ -3579,6 +3056,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); } @@ -3619,43 +3098,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) + struct ldlm_lock *new_lock, + struct mdt_lock_handle *lh, + __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]; - - /* 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) { - 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); - return; - } - cfs_hash_put(exp->exp_lock_hash, &lock->l_exp_hash); - } + /* 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 @@ -3671,8 +3137,46 @@ 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, + struct mdt_thread_info *info, + struct ldlm_lock **lockp, + __u64 flags) +{ + struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_RMT]; + struct ldlm_reply *ldlm_rep = NULL; + int rc, grc; + + /* + * Initialize lhc->mlh_reg_lh either from a previously granted lock + * (for the resend case) or a new lock. Below we will use it to + * replace the original lock. + */ + 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, + MDS_INODELOCK_XATTR, + MDT_LOCAL_LOCK); + if (rc) + return rc; + } + + grc = mdt_getxattr(info); + + 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); + if (ldlm_rep == NULL) + RETURN(err_serious(-EFAULT)); + + ldlm_rep->lock_policy_res2 = grc; + + return rc; } static int mdt_intent_getattr(enum mdt_it_code opcode, @@ -3681,10 +3185,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; @@ -3696,9 +3198,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: @@ -3717,12 +3219,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); + /* Get lock from request for possible resent case. */ + 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); @@ -3735,7 +3236,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); @@ -3752,11 +3253,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) { @@ -3768,16 +3269,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; @@ -3786,17 +3291,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); @@ -3820,7 +3327,7 @@ static int mdt_intent_reint(enum mdt_it_code opcode, ENTRY; - opc = mdt_reint_opcode(info, intent_fmts); + opc = mdt_reint_opcode(mdt_info_req(info), intent_fmts); if (opc < 0) RETURN(opc); @@ -3830,8 +3337,8 @@ static int mdt_intent_reint(enum mdt_it_code opcode, RETURN(err_serious(-EPROTO)); } - /* Get lock from request for possible resent case. */ - mdt_intent_fixup_resent(info, *lockp, NULL, lhc); + /* Get lock from request for possible resent case. */ + mdt_intent_fixup_resent(info, *lockp, lhc, flags); rc = mdt_reint_internal(info, lhc, opc); @@ -3847,10 +3354,10 @@ 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 == -EREMOTE || mdt_get_disposition(rep, DISP_OPEN_LOCK)) { + if (rc == -MDT_EREMOTE_OPEN || 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); } @@ -3883,7 +3390,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; @@ -3942,17 +3449,16 @@ static int mdt_intent_code(long itcode) static int mdt_intent_opc(long 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; @@ -3968,52 +3474,54 @@ 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; /* 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 = -EOPNOTSUPP; - } - RETURN(rc); + } + + 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 mdt_thread_info *info; - struct ptlrpc_request *req = req_cookie; - struct ldlm_intent *it; - struct req_capsule *pill; - int rc; + struct tgt_session_info *tsi; + struct mdt_thread_info *info; + struct ptlrpc_request *req = req_cookie; + struct ldlm_intent *it; + struct req_capsule *pill; + int rc; - ENTRY; + ENTRY; + + LASSERT(req != NULL); - LASSERT(req != NULL); + tsi = tgt_ses_info(req->rq_svc_thread->t_env); - info = lu_context_key_get(&req->rq_svc_thread->t_env->le_ctx, - &mdt_thread_key); + info = tsi2mdt_info(tsi); LASSERT(info != NULL); pill = info->mti_pill; LASSERT(pill->rc_req == req); @@ -4045,100 +3553,203 @@ static int mdt_intent_policy(struct ldlm_namespace *ns, if (rc) rc = err_serious(rc); } - RETURN(rc); + mdt_thread_info_fini(info); + RETURN(rc); } -static int mdt_seq_fini(const struct lu_env *env, - struct mdt_device *m) +static void mdt_deregister_seq_exp(struct mdt_device *mdt) +{ + struct seq_server_site *ss = mdt_seq_site(mdt); + + if (ss->ss_node_id == 0) + return; + + if (ss->ss_client_seq != NULL) { + lustre_deregister_lwp_item(&ss->ss_client_seq->lcs_exp); + ss->ss_client_seq->lcs_exp = NULL; + } + + if (ss->ss_server_fld != NULL) { + lustre_deregister_lwp_item(&ss->ss_server_fld->lsf_control_exp); + ss->ss_server_fld->lsf_control_exp = NULL; + } +} + +static void mdt_seq_fini_cli(struct mdt_device *mdt) { - return seq_site_fini(env, mdt_seq_site(m)); + struct seq_server_site *ss = mdt_seq_site(mdt); + + if (ss == NULL) + return; + + if (ss->ss_server_seq != NULL) + seq_server_set_cli(NULL, ss->ss_server_seq, NULL); } -static int mdt_seq_init(const struct lu_env *env, - const char *uuid, - struct mdt_device *m) +static int mdt_seq_fini(const struct lu_env *env, struct mdt_device *mdt) { - struct seq_server_site *ss; - char *prefix; - int rc; + mdt_seq_fini_cli(mdt); + mdt_deregister_seq_exp(mdt); + + return seq_site_fini(env, mdt_seq_site(mdt)); +} + +/** + * It will retrieve its FLDB entries from MDT0, and it only happens + * when upgrading existent FS to 2.6 or when local FLDB is corrupted, + * and it needs to refresh FLDB from the MDT0. + **/ +static int mdt_register_lwp_callback(void *data) +{ + struct lu_env env; + struct mdt_device *mdt = data; + struct lu_server_fld *fld = mdt_seq_site(mdt)->ss_server_fld; + int rc; ENTRY; - ss = mdt_seq_site(m); + LASSERT(mdt_seq_site(mdt)->ss_node_id != 0); - /* - * This is sequence-controller node. Init seq-controller server on local - * MDT. - */ - if (ss->ss_node_id == 0) { - LASSERT(ss->ss_control_seq == NULL); + if (!likely(fld->lsf_new)) + RETURN(0); - OBD_ALLOC_PTR(ss->ss_control_seq); - if (ss->ss_control_seq == NULL) - RETURN(-ENOMEM); + rc = lu_env_init(&env, LCT_MD_THREAD); + if (rc) { + 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); + GOTO(out, rc); + } +out: + lu_env_fini(&env); + RETURN(rc); +} - rc = seq_server_init(ss->ss_control_seq, - m->mdt_bottom, uuid, - LUSTRE_SEQ_CONTROLLER, - ss, - env); +static int mdt_register_seq_exp(struct mdt_device *mdt) +{ + struct seq_server_site *ss = mdt_seq_site(mdt); + char *lwp_name = NULL; + int rc; - if (rc) - GOTO(out_seq_fini, rc); + if (ss->ss_node_id == 0) + return 0; - OBD_ALLOC_PTR(ss->ss_client_seq); - if (ss->ss_client_seq == NULL) - GOTO(out_seq_fini, rc = -ENOMEM); + OBD_ALLOC(lwp_name, MAX_OBD_NAME); + if (lwp_name == NULL) + GOTO(out_free, rc = -ENOMEM); - OBD_ALLOC(prefix, MAX_OBD_NAME + 5); - if (prefix == NULL) { - OBD_FREE_PTR(ss->ss_client_seq); - GOTO(out_seq_fini, rc = -ENOMEM); - } + rc = tgt_name2lwp_name(mdt_obd_name(mdt), lwp_name, MAX_OBD_NAME, 0); + if (rc != 0) + GOTO(out_free, rc); - snprintf(prefix, MAX_OBD_NAME + 5, "ctl-%s", - uuid); + rc = lustre_register_lwp_item(lwp_name, &ss->ss_client_seq->lcs_exp, + NULL, NULL); + if (rc != 0) + GOTO(out_free, rc); - /* - * Init seq-controller client after seq-controller server is - * ready. Pass ss->ss_control_seq to it for direct talking. - */ - rc = seq_client_init(ss->ss_client_seq, NULL, - LUSTRE_SEQ_METADATA, prefix, - ss->ss_control_seq); - OBD_FREE(prefix, MAX_OBD_NAME + 5); + rc = lustre_register_lwp_item(lwp_name, + &ss->ss_server_fld->lsf_control_exp, + mdt_register_lwp_callback, mdt); + if (rc != 0) { + lustre_deregister_lwp_item(&ss->ss_client_seq->lcs_exp); + ss->ss_client_seq->lcs_exp = NULL; + GOTO(out_free, rc); + } +out_free: + if (lwp_name != NULL) + OBD_FREE(lwp_name, MAX_OBD_NAME); + + return rc; +} + +/* + * Init client sequence manager which is used by local MDS to talk to sequence + * controller on remote node. + */ +static int mdt_seq_init_cli(const struct lu_env *env, struct mdt_device *mdt) +{ + struct seq_server_site *ss = mdt_seq_site(mdt); + int rc; + char *prefix; + ENTRY; + + /* check if this is adding the first MDC and controller is not yet + * initialized. */ + OBD_ALLOC_PTR(ss->ss_client_seq); + if (ss->ss_client_seq == NULL) + RETURN(-ENOMEM); + + OBD_ALLOC(prefix, MAX_OBD_NAME + 5); + if (prefix == NULL) { + OBD_FREE_PTR(ss->ss_client_seq); + ss->ss_client_seq = NULL; + RETURN(-ENOMEM); + } + + /* Note: seq_client_fini will be called in seq_site_fini */ + snprintf(prefix, MAX_OBD_NAME + 5, "ctl-%s", mdt_obd_name(mdt)); + rc = seq_client_init(ss->ss_client_seq, NULL, LUSTRE_SEQ_METADATA, + prefix, ss->ss_node_id == 0 ? ss->ss_control_seq : + NULL); + OBD_FREE(prefix, MAX_OBD_NAME + 5); + if (rc != 0) { + OBD_FREE_PTR(ss->ss_client_seq); + ss->ss_client_seq = NULL; + RETURN(rc); + } + rc = seq_server_set_cli(env, ss->ss_server_seq, ss->ss_client_seq); + + RETURN(rc); +} + +static int mdt_seq_init(const struct lu_env *env, struct mdt_device *mdt) +{ + struct seq_server_site *ss; + int rc; + ENTRY; + + ss = mdt_seq_site(mdt); + /* init sequence controller server(MDT0) */ + if (ss->ss_node_id == 0) { + OBD_ALLOC_PTR(ss->ss_control_seq); + if (ss->ss_control_seq == NULL) + RETURN(-ENOMEM); + + rc = seq_server_init(env, ss->ss_control_seq, mdt->mdt_bottom, + mdt_obd_name(mdt), LUSTRE_SEQ_CONTROLLER, + ss); if (rc) GOTO(out_seq_fini, rc); } - /* Init seq-server on local MDT */ - LASSERT(ss->ss_server_seq == NULL); - + /* Init normal sequence server */ OBD_ALLOC_PTR(ss->ss_server_seq); if (ss->ss_server_seq == NULL) GOTO(out_seq_fini, rc = -ENOMEM); - rc = seq_server_init(ss->ss_server_seq, - m->mdt_bottom, uuid, - LUSTRE_SEQ_SERVER, - ss, - env); + rc = seq_server_init(env, ss->ss_server_seq, mdt->mdt_bottom, + mdt_obd_name(mdt), LUSTRE_SEQ_SERVER, ss); if (rc) - GOTO(out_seq_fini, rc = -ENOMEM); + GOTO(out_seq_fini, rc); - /* Assign seq-controller client to local seq-server. */ - if (ss->ss_node_id == 0) { - LASSERT(ss->ss_client_seq != NULL); + /* init seq client for seq server to talk to seq controller(MDT0) */ + rc = mdt_seq_init_cli(env, mdt); + if (rc != 0) + GOTO(out_seq_fini, rc); - rc = seq_server_set_cli(ss->ss_server_seq, - ss->ss_client_seq, - env); - } + if (ss->ss_node_id != 0) + /* register controler export through lwp */ + rc = mdt_register_seq_exp(mdt); EXIT; out_seq_fini: if (rc) - mdt_seq_fini(env, m); + mdt_seq_fini(env, mdt); return rc; } @@ -4176,7 +3787,7 @@ static int mdt_fld_init(const struct lu_env *env, RETURN(rc = -ENOMEM); rc = fld_server_init(env, ss->ss_server_fld, m->mdt_bottom, uuid, - ss->ss_node_id, LU_SEQ_RANGE_MDT); + LU_SEQ_RANGE_MDT); if (rc) { OBD_FREE_PTR(ss->ss_server_fld); ss->ss_server_fld = NULL; @@ -4189,7 +3800,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; @@ -4204,7 +3814,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 @@ -4215,7 +3824,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); @@ -4359,7 +3968,7 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt, strcpy(name, dev); p = strstr(name, "-MDT"); if (p == NULL) - GOTO(cleanup_mem, rc = -ENOMEM); + GOTO(free_bufs, rc = -ENOMEM); p[3] = 'D'; snprintf(uuid, MAX_OBD_NAME, "%s_UUID", name); @@ -4368,7 +3977,7 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt, if (lprof == NULL || lprof->lp_dt == NULL) { CERROR("can't find the profile: %s\n", lustre_cfg_string(cfg, 0)); - GOTO(cleanup_mem, rc = -EINVAL); + GOTO(free_bufs, rc = -EINVAL); } lustre_cfg_bufs_reset(bufs, name); @@ -4388,7 +3997,7 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt, if (!obd) { CERROR("Can not find obd %s (%s in config)\n", MDD_OBD_NAME, lustre_cfg_string(cfg, 0)); - GOTO(class_detach, rc = -EINVAL); + GOTO(lcfg_cleanup, rc = -EINVAL); } lustre_cfg_free(lcfg); @@ -4407,7 +4016,7 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt, /* connect to MDD we just setup */ rc = mdt_connect_to_next(env, mdt, name, &mdt->mdt_child_exp); if (rc) - RETURN(rc); + GOTO(class_detach, rc); site = mdt->mdt_child_exp->exp_obd->obd_lu_dev->ld_site; LASSERT(site); @@ -4416,16 +4025,14 @@ 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); if (rc) - RETURN(rc); + GOTO(class_detach, rc); 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); @@ -4518,7 +4125,7 @@ static int mdt_quota_init(const struct lu_env *env, struct mdt_device *mdt, if (!obd) { CERROR("Can not find obd %s (%s in config)\n", qmtname, lustre_cfg_string(cfg, 0)); - GOTO(class_detach, rc = -EINVAL); + GOTO(lcfg_cleanup, rc = -EINVAL); } lustre_cfg_free(lcfg); @@ -4596,19 +4203,150 @@ static void mdt_quota_fini(const struct lu_env *env, struct mdt_device *mdt) EXIT; } -static void mdt_fini(const struct lu_env *env, struct mdt_device *m) +/* 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) { - struct md_device *next = m->mdt_child; - struct lu_device *d = &m->mdt_lu_dev; - struct obd_device *obd = mdt2obd_dev(m); - ENTRY; + struct mdt_thread_info *info = tsi2mdt_info(tsi); + int rc; + + rc = mdt_getxattr(info); + + mdt_thread_info_fini(info); + return rc; +} + +static struct tgt_handler mdt_tgt_handlers[] = { +TGT_RPC_HANDLER(MDS_FIRST_OPC, + 0, MDS_CONNECT, mdt_tgt_connect, + &RQF_CONNECT, LUSTRE_OBD_VERSION), +TGT_RPC_HANDLER(MDS_FIRST_OPC, + 0, MDS_DISCONNECT, tgt_disconnect, + &RQF_MDS_DISCONNECT, LUSTRE_OBD_VERSION), +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(HABEO_CORPUS, MDS_GETATTR, mdt_getattr), +TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_GETATTR_NAME, + mdt_getattr_name), +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), +TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO | MUTABOR, MDS_HSM_PROGRESS, + mdt_hsm_progress), +TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO | MUTABOR, MDS_HSM_CT_REGISTER, + mdt_hsm_ct_register), +TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO | MUTABOR, MDS_HSM_CT_UNREGISTER, + mdt_hsm_ct_unregister), +TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_HSM_STATE_GET, + mdt_hsm_state_get), +TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO | MUTABOR, MDS_HSM_STATE_SET, + mdt_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_CLAVIS | HABEO_CORPUS | HABEO_REFERO | MUTABOR, + MDS_SWAP_LAYOUTS, + mdt_swap_layouts), +}; + +static struct tgt_handler mdt_sec_ctx_ops[] = { +TGT_SEC_HDL_VAR(0, SEC_CTX_INIT, mdt_sec_ctx_handle), +TGT_SEC_HDL_VAR(0, SEC_CTX_INIT_CONT,mdt_sec_ctx_handle), +TGT_SEC_HDL_VAR(0, SEC_CTX_FINI, mdt_sec_ctx_handle) +}; + +static struct tgt_handler mdt_quota_ops[] = { +TGT_QUOTA_HDL(HABEO_REFERO, QUOTA_DQACQ, mdt_quota_dqacq), +}; + +static struct tgt_opc_slice mdt_common_slice[] = { + { + .tos_opc_start = MDS_FIRST_OPC, + .tos_opc_end = MDS_LAST_OPC, + .tos_hs = mdt_tgt_handlers + }, + { + .tos_opc_start = OBD_FIRST_OPC, + .tos_opc_end = OBD_LAST_OPC, + .tos_hs = tgt_obd_handlers + }, + { + .tos_opc_start = LDLM_FIRST_OPC, + .tos_opc_end = LDLM_LAST_OPC, + .tos_hs = tgt_dlm_handlers + }, + { + .tos_opc_start = SEC_FIRST_OPC, + .tos_opc_end = SEC_LAST_OPC, + .tos_hs = mdt_sec_ctx_ops + }, + { + .tos_opc_start = OUT_UPDATE_FIRST_OPC, + .tos_opc_end = OUT_UPDATE_LAST_OPC, + .tos_hs = tgt_out_handlers + }, + { + .tos_opc_start = FLD_FIRST_OPC, + .tos_opc_end = FLD_LAST_OPC, + .tos_hs = fld_handlers + }, + { + .tos_opc_start = SEQ_FIRST_OPC, + .tos_opc_end = SEQ_LAST_OPC, + .tos_hs = seq_handlers + }, + { + .tos_opc_start = QUOTA_DQACQ, + .tos_opc_end = QUOTA_LAST_OPC, + .tos_hs = mdt_quota_ops + }, + { + .tos_opc_start = LLOG_FIRST_OPC, + .tos_opc_end = LLOG_LAST_OPC, + .tos_hs = tgt_llog_handlers + }, + { + .tos_opc_start = LFSCK_FIRST_OPC, + .tos_opc_end = LFSCK_LAST_OPC, + .tos_hs = tgt_lfsck_handlers + }, + + { + .tos_hs = NULL + } +}; - target_recovery_fini(obd); +static void mdt_fini(const struct lu_env *env, struct mdt_device *m) +{ + struct md_device *next = m->mdt_child; + struct lu_device *d = &m->mdt_lu_dev; + struct obd_device *obd = mdt2obd_dev(m); + struct lfsck_stop stop; + ENTRY; - ping_evictor_stop(); + 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)); + if (m->mdt_opts.mo_coordinator) + mdt_hsm_cdt_stop(m); + + mdt_hsm_cdt_fini(m); + mdt_llog_ctxt_unclone(env, m, LLOG_AGENT_ORIG_CTXT); mdt_llog_ctxt_unclone(env, m, LLOG_CHANGELOG_ORIG_CTXT); obd_exports_barrier(obd); @@ -4629,18 +4367,10 @@ 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; - } - - next->md_ops->mdo_iocontrol(env, next, OBD_IOC_PAUSE_LFSCK, 0, NULL); + cfs_free_nidlist(&m->mdt_squash.rsi_nosquash_nids); mdt_seq_fini(env, m); mdt_fld_fini(env, m); - sptlrpc_rule_set_free(&m->mdt_sptlrpc_rset); next->md_ops->mdo_init_capa_ctxt(env, next, 0, 0, 0, NULL); cfs_timer_disarm(&m->mdt_ck_timer); @@ -4651,38 +4381,14 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) */ mdt_stack_fini(env, m, md2lu_dev(m->mdt_child)); - LASSERT(cfs_atomic_read(&d->ld_ref) == 0); + LASSERT(atomic_read(&d->ld_ref) == 0); - server_put_mount(mdt_obd_name(m), NULL); + server_put_mount(mdt_obd_name(m), true); EXIT; } -static int mdt_adapt_sptlrpc_conf(struct obd_device *obd, int initial) -{ - struct mdt_device *m = mdt_dev(obd->obd_lu_dev); - struct sptlrpc_rule_set tmp_rset; - int rc; - - sptlrpc_rule_set_init(&tmp_rset); - rc = sptlrpc_conf_target_get_rules(obd, &tmp_rset, initial); - if (rc) { - CERROR("mdt %s: failed get sptlrpc rules: %d\n", - mdt_obd_name(m), rc); - return rc; - } - - sptlrpc_target_update_exp_flavor(obd, &tmp_rset); - - write_lock(&m->mdt_sptlrpc_lock); - sptlrpc_rule_set_free(&m->mdt_sptlrpc_rset); - m->mdt_sptlrpc_rset = tmp_rset; - write_unlock(&m->mdt_sptlrpc_lock); - - return 0; -} - -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) @@ -4698,7 +4404,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, const char *identity_upcall = "NONE"; struct md_device *next; int rc; - int node_id; + long node_id; mntopt_t mntopts; ENTRY; @@ -4727,6 +4433,10 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, m->mdt_opts.mo_cos = MDT_COS_DEFAULT; + /* default is coordinator off, it is started through conf_param + * or /proc */ + m->mdt_opts.mo_coordinator = 0; + lmi = server_get_mount(dev); if (lmi == NULL) { CERROR("Cannot get mount info for %s!\n", dev); @@ -4737,24 +4447,16 @@ 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; - } - - rwlock_init(&m->mdt_sptlrpc_lock); - sptlrpc_rule_set_init(&m->mdt_sptlrpc_rset); + } spin_lock_init(&m->mdt_ioepoch_lock); - m->mdt_opts.mo_compat_resname = 0; - m->mdt_opts.mo_mds_capa = 1; - m->mdt_opts.mo_oss_capa = 1; 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; @@ -4801,7 +4503,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, if (rc) GOTO(err_fini_stack, rc); - rc = mdt_seq_init(env, mdt_obd_name(m), m); + rc = mdt_seq_init(env, m); if (rc) GOTO(err_fini_fld, rc); @@ -4823,11 +4525,18 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, 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", + mdt_obd_name(m), rc); + GOTO(err_free_ns, rc); + } + rc = mdt_ck_thread_start(m); if (rc) - GOTO(err_free_ns, rc); + GOTO(err_free_hsm, rc); - rc = tgt_init(env, &m->mdt_lut, obd, m->mdt_bottom, NULL, + 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) @@ -4837,7 +4546,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, if (rc) GOTO(err_tgt, rc); - mdt_adapt_sptlrpc_conf(obd, 1); + tgt_adapt_sptlrpc_conf(&m->mdt_lut, 1); next = m->mdt_child; rc = next->md_ops->mdo_iocontrol(env, next, OBD_IOC_GET_MNTOPT, 0, @@ -4850,6 +4559,10 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, else m->mdt_opts.mo_user_xattr = 0; + rc = next->md_ops->mdo_maxeasize_get(env, next, &m->mdt_max_ea_size); + if (rc) + GOTO(err_fs_cleanup, rc); + if (mntopts & MNTOPT_ACL) m->mdt_opts.mo_acl = 1; else @@ -4898,7 +4611,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, ldlm_timeout = MDS_LDLM_TIMEOUT_DEFAULT; RETURN(0); - err_procfs: mdt_procfs_fini(m); err_recovery: @@ -4912,6 +4624,8 @@ err_tgt: 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: ldlm_namespace_free(m->mdt_namespace, NULL, 0); obd->obd_namespace = m->mdt_namespace = NULL; @@ -4923,7 +4637,7 @@ err_fini_stack: mdt_stack_fini(env, m, md2lu_dev(m->mdt_child)); err_lmi: if (lmi) - server_put_mount(dev, lmi->lmi_mnt); + server_put_mount(dev, true); return(rc); } @@ -4951,7 +4665,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 */ @@ -4986,12 +4699,18 @@ 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, + rc = class_process_proc_param(PARAM_MDT, obd->obd_vars, cfg, obd); - if (rc > 0 || rc == -ENOSYS) - /* we don't understand; pass it on */ - rc = next->ld_ops->ldo_process_config(env, next, cfg); + if (rc > 0 || rc == -ENOSYS) { + /* is it an HSM var ? */ + rc = class_process_proc_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, + cfg); + } if (old_cfg != NULL) lustre_cfg_free(cfg); @@ -5007,24 +4726,24 @@ static int mdt_process_config(const struct lu_env *env, } static struct lu_object *mdt_object_alloc(const struct lu_env *env, - const struct lu_object_header *hdr, - struct lu_device *d) + const struct lu_object_header *hdr, + struct lu_device *d) { - struct mdt_object *mo; + struct mdt_object *mo; - ENTRY; + ENTRY; - OBD_SLAB_ALLOC_PTR_GFP(mo, mdt_object_kmem, __GFP_IO); - if (mo != NULL) { - struct lu_object *o; - struct lu_object_header *h; + OBD_SLAB_ALLOC_PTR_GFP(mo, mdt_object_kmem, GFP_NOFS); + if (mo != NULL) { + struct lu_object *o; + struct lu_object_header *h; o = &mo->mot_obj; - h = &mo->mot_header; - lu_object_header_init(h); - lu_object_init(o, h, d); - lu_object_add_top(h, o); - o->lo_ops = &mdt_obj_ops; + h = &mo->mot_header; + lu_object_header_init(h); + lu_object_init(o, h, d); + lu_object_add_top(h, o); + o->lo_ops = &mdt_obj_ops; mutex_init(&mo->mot_ioepoch_mutex); mutex_init(&mo->mot_lov_mutex); init_rwsem(&mo->mot_open_sem); @@ -5092,7 +4811,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; @@ -5111,16 +4829,10 @@ static int mdt_prepare(const struct lu_env *env, if (rc) RETURN(rc); - lsp.lsp_start = NULL; - lsp.lsp_namespace = mdt->mdt_namespace; - 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; - } + rc = lfsck_register_namespace(env, mdt->mdt_bottom, mdt->mdt_namespace); + /* The LFSCK instance is registered just now, so it must be there when + * register the namespace to such instance. */ + LASSERTF(rc == 0, "register namespace failed: rc = %d\n", rc); if (mdt->mdt_seq_site.ss_node_id == 0) { rc = mdt->mdt_child->md_ops->mdo_root_get(env, mdt->mdt_child, @@ -5130,7 +4842,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, mdt_recovery_handle); + target_recovery_init(&mdt->mdt_lut, tgt_request_handle); set_bit(MDT_FL_CFGLOG, &mdt->mdt_state); LASSERT(obd->obd_no_conn); spin_lock(&obd->obd_dev_lock); @@ -5161,18 +4873,16 @@ static int mdt_obd_set_info_async(const struct lu_env *env, __u32 vallen, void *val, struct ptlrpc_request_set *set) { - struct obd_device *obd = exp->exp_obd; - int rc; - ENTRY; + int rc; - LASSERT(obd); + ENTRY; - if (KEY_IS(KEY_SPTLRPC_CONF)) { - rc = mdt_adapt_sptlrpc_conf(obd, 0); - RETURN(rc); - } + if (KEY_IS(KEY_SPTLRPC_CONF)) { + rc = tgt_adapt_sptlrpc_conf(class_exp2tgt(exp), 0); + RETURN(rc); + } - RETURN(0); + RETURN(0); } /** @@ -5205,12 +4915,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; @@ -5251,8 +4963,6 @@ static int mdt_connect_internal(struct obd_export *exp, } data->ocd_version = LUSTRE_VERSION_CODE; - exp->exp_connect_data = *data; - exp->exp_mdt_data.med_ibits_known = data->ocd_ibits_known; if ((data->ocd_connect_flags & OBD_CONNECT_FID) == 0) { CWARN("%s: MDS requires FID support, but client not\n", @@ -5279,51 +4989,9 @@ static int mdt_connect_internal(struct obd_export *exp, } } - return 0; -} - -static int mdt_connect_check_sptlrpc(struct mdt_device *mdt, - struct obd_export *exp, - struct ptlrpc_request *req) -{ - struct sptlrpc_flavor flvr; - int rc = 0; - - if (exp->exp_flvr.sf_rpc == SPTLRPC_FLVR_INVALID) { - read_lock(&mdt->mdt_sptlrpc_lock); - sptlrpc_target_choose_flavor(&mdt->mdt_sptlrpc_rset, - req->rq_sp_from, - req->rq_peer.nid, - &flvr); - read_unlock(&mdt->mdt_sptlrpc_lock); - - spin_lock(&exp->exp_lock); - - exp->exp_sp_peer = req->rq_sp_from; - exp->exp_flvr = flvr; - - if (exp->exp_flvr.sf_rpc != SPTLRPC_FLVR_ANY && - exp->exp_flvr.sf_rpc != req->rq_flvr.sf_rpc) { - CERROR("unauthorized rpc flavor %x from %s, " - "expect %x\n", req->rq_flvr.sf_rpc, - libcfs_nid2str(req->rq_peer.nid), - exp->exp_flvr.sf_rpc); - rc = -EACCES; - } - - spin_unlock(&exp->exp_lock); - } else { - if (exp->exp_sp_peer != req->rq_sp_from) { - CERROR("RPC source %s doesn't match %s\n", - sptlrpc_part2name(req->rq_sp_from), - sptlrpc_part2name(exp->exp_sp_peer)); - rc = -EACCES; - } else { - rc = sptlrpc_target_export_check(exp, req); - } - } + data->ocd_max_easize = mdt->mdt_max_ea_size; - return rc; + return 0; } /* mds_connect copy */ @@ -5333,11 +5001,9 @@ static int mdt_obd_connect(const struct lu_env *env, struct obd_connect_data *data, void *localdata) { - struct mdt_thread_info *info; struct obd_export *lexp; struct lustre_handle conn = { 0 }; struct mdt_device *mdt; - struct ptlrpc_request *req; int rc; ENTRY; @@ -5345,18 +5011,18 @@ static int mdt_obd_connect(const struct lu_env *env, if (!exp || !obd || !cluuid) RETURN(-EINVAL); - info = lu_context_key_get(&env->le_ctx, &mdt_thread_key); - req = info->mti_pill->rc_req; - mdt = mdt_dev(obd->obd_lu_dev); + 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) && + if (!test_bit(MDT_FL_SYNCED, &mdt->mdt_state) && data != NULL && !(data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT)) { - rc = obd_health_check(env, mdt->mdt_child_exp->exp_obd); + 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); @@ -5369,26 +5035,22 @@ static int mdt_obd_connect(const struct lu_env *env, lexp = class_conn2export(&conn); LASSERT(lexp != NULL); - rc = mdt_connect_check_sptlrpc(mdt, lexp, req); - if (rc) - GOTO(out, rc); - - if (OBD_FAIL_CHECK(OBD_FAIL_TGT_RCVG_FLAG)) - lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECOVERING); - rc = mdt_connect_internal(lexp, mdt, data); if (rc == 0) { struct lsd_client_data *lcd = lexp->exp_target_data.ted_lcd; LASSERT(lcd); - info->mti_exp = lexp; 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. */ + spin_lock(&lexp->exp_lock); + lexp->exp_keep_sync = 1; + spin_unlock(&lexp->exp_lock); } -out: if (rc != 0) { class_disconnect(lexp); *exp = NULL; @@ -5405,23 +5067,12 @@ static int mdt_obd_reconnect(const struct lu_env *env, struct obd_connect_data *data, void *localdata) { - struct mdt_thread_info *info; - struct mdt_device *mdt; - struct ptlrpc_request *req; int rc; ENTRY; if (exp == NULL || obd == NULL || cluuid == NULL) RETURN(-EINVAL); - info = lu_context_key_get(&env->le_ctx, &mdt_thread_key); - req = info->mti_pill->rc_req; - mdt = mdt_dev(obd->obd_lu_dev); - - rc = mdt_connect_check_sptlrpc(mdt, exp, req); - if (rc) - RETURN(rc); - rc = mdt_connect_internal(exp, mdt_dev(obd->obd_lu_dev), data); if (rc == 0) mdt_export_stats_init(obd, exp, localdata); @@ -5429,27 +5080,53 @@ static int mdt_obd_reconnect(const struct lu_env *env, RETURN(rc); } +static int mdt_ctxt_add_dirty_flag(struct lu_env *env, + struct mdt_thread_info *info, + struct mdt_file_data *mfd) +{ + struct lu_context ses; + int rc; + ENTRY; + + rc = lu_context_init(&ses, LCT_SERVER_SESSION); + if (rc) + RETURN(rc); + + env->le_ses = &ses; + lu_context_enter(&ses); + + mdt_ucred(info)->uc_valid = UCRED_OLD; + rc = mdt_add_dirty_flag(info, mfd->mfd_object, &info->mti_attr); + + lu_context_exit(&ses); + lu_context_fini(&ses); + env->le_ses = NULL; + + RETURN(rc); +} + static int mdt_export_cleanup(struct obd_export *exp) { - struct mdt_export_data *med = &exp->exp_mdt_data; - struct obd_device *obd = exp->exp_obd; - struct mdt_device *mdt; - struct mdt_thread_info *info; - struct lu_env env; - CFS_LIST_HEAD(closing_list); - struct mdt_file_data *mfd, *n; - int rc = 0; - ENTRY; + struct list_head closing_list; + struct mdt_export_data *med = &exp->exp_mdt_data; + struct obd_device *obd = exp->exp_obd; + struct mdt_device *mdt; + struct mdt_thread_info *info; + struct lu_env env; + struct mdt_file_data *mfd, *n; + int rc = 0; + ENTRY; + INIT_LIST_HEAD(&closing_list); spin_lock(&med->med_open_lock); - while (!cfs_list_empty(&med->med_open_head)) { - cfs_list_t *tmp = med->med_open_head.next; - mfd = cfs_list_entry(tmp, struct mdt_file_data, mfd_list); + while (!list_empty(&med->med_open_head)) { + struct list_head *tmp = med->med_open_head.next; + mfd = list_entry(tmp, struct mdt_file_data, mfd_list); /* Remove mfd handle so it can't be found again. * We are consuming the mfd_list reference here. */ class_handle_unhash(&mfd->mfd_handle); - cfs_list_move_tail(&mfd->mfd_list, &closing_list); + list_move_tail(&mfd->mfd_list, &closing_list); } spin_unlock(&med->med_open_lock); mdt = mdt_dev(obd->obd_lu_dev); @@ -5466,13 +5143,32 @@ static int mdt_export_cleanup(struct obd_export *exp) info->mti_mdt = mdt; info->mti_exp = exp; - if (!cfs_list_empty(&closing_list)) { - struct md_attr *ma = &info->mti_attr; + if (!list_empty(&closing_list)) { + struct md_attr *ma = &info->mti_attr; - /* Close any open files (which may also cause orphan unlinking). */ - cfs_list_for_each_entry_safe(mfd, n, &closing_list, mfd_list) { - cfs_list_del_init(&mfd->mfd_list); + /* Close any open files (which may also cause orphan + * unlinking). */ + list_for_each_entry_safe(mfd, n, &closing_list, mfd_list) { + list_del_init(&mfd->mfd_list); ma->ma_need = ma->ma_valid = 0; + + /* This file is being closed due to an eviction, it + * could have been modified and now dirty regarding to + * HSM archive, check this! + * The logic here is to mark a file dirty if there's a + * chance it was dirtied before the client was evicted, + * so that we don't have to wait for a release attempt + * before finding out the file was actually dirty and + * fail the release. Aggressively marking it dirty here + * will cause the policy engine to attempt to + * re-archive it; when rearchiving, we can compare the + * current version to the HSM data_version and make the + * archive request into a noop if it's not actually + * dirty. + */ + if (mfd->mfd_mode & (FMODE_WRITE|MDS_FMODE_TRUNC)) + rc = mdt_ctxt_add_dirty_flag(&env, info, mfd); + /* Don't unlink orphan on failover umount, LU-184 */ if (exp->exp_flags & OBD_OPT_FAILOVER) { ma->ma_valid = MA_FLAGS; @@ -5499,23 +5195,23 @@ 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? */ static int mdt_init_export(struct obd_export *exp) { - struct mdt_export_data *med = &exp->exp_mdt_data; - int rc; - ENTRY; + struct mdt_export_data *med = &exp->exp_mdt_data; + int rc; + ENTRY; - CFS_INIT_LIST_HEAD(&med->med_open_head); + INIT_LIST_HEAD(&med->med_open_head); spin_lock_init(&med->med_open_lock); mutex_init(&med->med_idmap_mutex); med->med_idmap = NULL; @@ -5560,13 +5256,13 @@ static int mdt_destroy_export(struct obd_export *exp) &exp->exp_client_uuid))) RETURN(0); - ldlm_destroy_export(exp); - tgt_client_free(exp); + ldlm_destroy_export(exp); + tgt_client_free(exp); - LASSERT(cfs_list_empty(&exp->exp_outstanding_replies)); - LASSERT(cfs_list_empty(&exp->exp_mdt_data.med_open_head)); + LASSERT(list_empty(&exp->exp_outstanding_replies)); + LASSERT(list_empty(&exp->exp_mdt_data.med_open_head)); - RETURN(0); + RETURN(0); } /** The maximum depth that fid2path() will search. @@ -5588,8 +5284,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; @@ -5647,36 +5343,40 @@ 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 */ 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 (pli->pli_fidcount == 0 && + pli->pli_linkno < leh->leh_reccount) { int count; for (count = 0; count < pli->pli_linkno; count++) { lee = (struct link_ea_entry *) @@ -5689,6 +5389,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) @@ -5794,7 +5514,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); } @@ -5830,71 +5550,77 @@ static int mdt_rpc_fid2path(struct mdt_thread_info *info, void *key, RETURN(rc); } -int mdt_get_info(struct mdt_thread_info *info) +int mdt_get_info(struct tgt_session_info *tsi) { - struct ptlrpc_request *req = mdt_info_req(info); - char *key; - int keylen; - __u32 *vallen; - void *valout; - int rc; - ENTRY; + char *key; + int keylen; + __u32 *vallen; + void *valout; + int rc; - key = req_capsule_client_get(info->mti_pill, &RMF_GETINFO_KEY); - if (key == NULL) { - CDEBUG(D_IOCTL, "No GETINFO key"); - RETURN(-EFAULT); - } - keylen = req_capsule_get_size(info->mti_pill, &RMF_GETINFO_KEY, - RCL_CLIENT); + ENTRY; - vallen = req_capsule_client_get(info->mti_pill, &RMF_GETINFO_VALLEN); - if (vallen == NULL) { - CDEBUG(D_IOCTL, "Unable to get RMF_GETINFO_VALLEN buffer"); - RETURN(-EFAULT); - } + key = req_capsule_client_get(tsi->tsi_pill, &RMF_GETINFO_KEY); + if (key == NULL) { + CDEBUG(D_IOCTL, "No GETINFO key\n"); + RETURN(err_serious(-EFAULT)); + } + keylen = req_capsule_get_size(tsi->tsi_pill, &RMF_GETINFO_KEY, + RCL_CLIENT); - req_capsule_set_size(info->mti_pill, &RMF_GETINFO_VAL, RCL_SERVER, - *vallen); - rc = req_capsule_server_pack(info->mti_pill); - valout = req_capsule_server_get(info->mti_pill, &RMF_GETINFO_VAL); - if (valout == NULL) { - CDEBUG(D_IOCTL, "Unable to get get-info RPC out buffer"); - RETURN(-EFAULT); - } + vallen = req_capsule_client_get(tsi->tsi_pill, &RMF_GETINFO_VALLEN); + if (vallen == NULL) { + CDEBUG(D_IOCTL, "%s: cannot get RMF_GETINFO_VALLEN buffer\n", + tgt_name(tsi->tsi_tgt)); + RETURN(err_serious(-EFAULT)); + } - if (KEY_IS(KEY_FID2PATH)) - rc = mdt_rpc_fid2path(info, key, valout, *vallen); - else - rc = -EINVAL; + req_capsule_set_size(tsi->tsi_pill, &RMF_GETINFO_VAL, RCL_SERVER, + *vallen); + rc = req_capsule_server_pack(tsi->tsi_pill); + if (rc) + RETURN(err_serious(rc)); + + valout = req_capsule_server_get(tsi->tsi_pill, &RMF_GETINFO_VAL); + if (valout == NULL) { + CDEBUG(D_IOCTL, "%s: cannot get get-info RPC out buffer\n", + tgt_name(tsi->tsi_tgt)); + RETURN(err_serious(-EFAULT)); + } - lustre_msg_set_status(req->rq_repmsg, rc); + if (KEY_IS(KEY_FID2PATH)) { + struct mdt_thread_info *info = tsi2mdt_info(tsi); - RETURN(rc); + rc = mdt_rpc_fid2path(info, key, valout, *vallen); + mdt_thread_info_fini(info); + } else { + rc = -EINVAL; + } + RETURN(rc); } /* Pass the ioc down */ static int mdt_ioc_child(struct lu_env *env, struct mdt_device *mdt, - unsigned int cmd, int len, void *data) + unsigned int cmd, int len, void *data) { - struct lu_context ioctl_session; - struct md_device *next = mdt->mdt_child; - int rc; - ENTRY; + struct lu_context ioctl_session; + struct md_device *next = mdt->mdt_child; + int rc; + ENTRY; - rc = lu_context_init(&ioctl_session, LCT_SESSION); - if (rc) - RETURN(rc); - ioctl_session.lc_thread = (struct ptlrpc_thread *)cfs_current(); - lu_context_enter(&ioctl_session); - env->le_ses = &ioctl_session; + rc = lu_context_init(&ioctl_session, LCT_SERVER_SESSION); + if (rc) + RETURN(rc); + ioctl_session.lc_thread = (struct ptlrpc_thread *)current; + lu_context_enter(&ioctl_session); + env->le_ses = &ioctl_session; - LASSERT(next->md_ops->mdo_iocontrol); - rc = next->md_ops->mdo_iocontrol(env, next, cmd, len, data); + LASSERT(next->md_ops->mdo_iocontrol); + rc = next->md_ops->mdo_iocontrol(env, next, cmd, len, data); - lu_context_exit(&ioctl_session); - lu_context_fini(&ioctl_session); - RETURN(rc); + lu_context_exit(&ioctl_session); + lu_context_fini(&ioctl_session); + RETURN(rc); } static int mdt_ioc_version_get(struct mdt_thread_info *mti, void *karg) @@ -5988,14 +5714,24 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len, } lsp.lsp_start = (struct lfsck_start *)(data->ioc_inlbuf1); - lsp.lsp_namespace = mdt->mdt_namespace; + lsp.lsp_index_valid = 0; rc = next->md_ops->mdo_iocontrol(&env, next, cmd, 0, &lsp); break; } case OBD_IOC_STOP_LFSCK: { - struct md_device *next = mdt->mdt_child; + struct md_device *next = mdt->mdt_child; + struct obd_ioctl_data *data = karg; + struct lfsck_stop stop; + + stop.ls_status = LS_STOPPED; + /* Old lfsck utils may pass NULL @stop. */ + if (data->ioc_inlbuf1 == NULL) + stop.ls_flags = 0; + else + stop.ls_flags = + ((struct lfsck_stop *)(data->ioc_inlbuf1))->ls_flags; - rc = next->md_ops->mdo_iocontrol(&env, next, cmd, 0, NULL); + rc = next->md_ops->mdo_iocontrol(&env, next, cmd, 0, &stop); break; } case OBD_IOC_GET_OBJ_VERSION: { @@ -6009,6 +5745,15 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len, rc = mdt_ioc_version_get(mti, karg); break; } + case OBD_IOC_CATLOGLIST: { + struct mdt_thread_info *mti; + + mti = lu_context_key_get(&env.le_ctx, &mdt_thread_key); + lu_local_obj_fid(&mti->mti_tmp_fid1, LLOG_CATALOGS_OID); + rc = llog_catalog_list(&env, mdt->mdt_bottom, 0, karg, + &mti->mti_tmp_fid1); + break; + } default: rc = -EOPNOTSUPP; CERROR("%s: Not supported cmd = %d, rc = %d\n", @@ -6019,17 +5764,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; @@ -6143,13 +5898,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); } /** @@ -6179,7 +5934,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") == @@ -6194,10 +5948,8 @@ 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, - lvars.module_vars, LUSTRE_MDT_NAME, - &mdt_device_type); + rc = class_register_type(&mdt_obd_device_ops, NULL, true, NULL, + LUSTRE_MDT_NAME, &mdt_device_type); if (rc) GOTO(mds_fini, rc); lu_fini: