X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_handler.c;h=5118c445e1692707838e1dbf7b5d8aa4a3f32081;hp=f64795fa8514cda4a5115ca4523442d459219265;hb=ad1810a2dbea1eed5e8b5feb55bdf915a545feb3;hpb=de2d5808bd2987f76d2486272e1a9c192ba277d4 diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c index f64795f..5118c44 100644 --- a/lustre/mdt/mdt_handler.c +++ b/lustre/mdt/mdt_handler.c @@ -27,7 +27,7 @@ * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2010, 2013, Intel Corporation. + * Copyright (c) 2010, 2014, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -69,6 +69,7 @@ #include #include #include +#include mdl_mode_t mdt_mdl_lock_modes[] = { [LCK_MINMODE] = MDL_MINMODE, @@ -388,12 +389,70 @@ static void mdt_pack_size2body(struct mdt_thread_info *info, b->mbo_blocks = ma->ma_som->msd_blocks; } +#ifdef CONFIG_FS_POSIX_ACL +/* + * Pack ACL data into the reply. UIDs/GIDs are mapped and filtered by nodemap. + * + * \param info thread info object + * \param repbody reply to pack ACLs into + * \param o mdt object of file to examine + * \param nodemap nodemap of client to reply to + * \retval 0 success + * \retval -errno error getting or parsing ACL from disk + */ +int mdt_pack_acl2body(struct mdt_thread_info *info, struct mdt_body *repbody, + struct mdt_object *o, struct lu_nodemap *nodemap) +{ + const struct lu_env *env = info->mti_env; + struct md_object *next = mdt_object_child(o); + struct lu_buf *buf = &info->mti_buf; + int rc; + + buf->lb_buf = req_capsule_server_get(info->mti_pill, &RMF_ACL); + buf->lb_len = req_capsule_get_size(info->mti_pill, &RMF_ACL, + RCL_SERVER); + if (buf->lb_len == 0) + return 0; + + rc = mo_xattr_get(env, next, buf, XATTR_NAME_ACL_ACCESS); + if (rc < 0) { + if (rc == -ENODATA) { + repbody->mbo_aclsize = 0; + repbody->mbo_valid |= OBD_MD_FLACL; + rc = 0; + } else if (rc == -EOPNOTSUPP) { + rc = 0; + } else { + CERROR("%s: unable to read "DFID" ACL: rc = %d\n", + mdt_obd_name(info->mti_mdt), + PFID(mdt_object_fid(o)), rc); + } + } else { + rc = nodemap_map_acl(nodemap, buf->lb_buf, + rc, NODEMAP_FS_TO_CLIENT); + /* if all ACLs mapped out, rc is still >= 0 */ + if (rc < 0) { + CERROR("%s: nodemap_map_acl unable to parse "DFID + " ACL: rc = %d\n", mdt_obd_name(info->mti_mdt), + PFID(mdt_object_fid(o)), rc); + } else { + repbody->mbo_aclsize = rc; + repbody->mbo_valid |= OBD_MD_FLACL; + rc = 0; + } + } + return rc; +} +#endif + void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b, const struct lu_attr *attr, const struct lu_fid *fid) { - struct md_attr *ma = &info->mti_attr; + struct md_attr *ma = &info->mti_attr; + struct obd_export *exp = info->mti_exp; + struct lu_nodemap *nodemap = exp->exp_target_data.ted_nodemap; - LASSERT(ma->ma_valid & MA_INODE); + LASSERT(ma->ma_valid & MA_INODE); b->mbo_atime = attr->la_atime; b->mbo_mtime = attr->la_mtime; @@ -401,8 +460,12 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b, 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_uid = nodemap_map_id(nodemap, NODEMAP_UID, + NODEMAP_FS_TO_CLIENT, + attr->la_uid); + b->mbo_gid = nodemap_map_id(nodemap, NODEMAP_GID, + NODEMAP_FS_TO_CLIENT, + attr->la_gid); b->mbo_flags = attr->la_flags; b->mbo_nlink = attr->la_nlink; b->mbo_rdev = attr->la_rdev; @@ -484,7 +547,8 @@ void mdt_client_compatibility(struct mdt_thread_info *info) EXIT; } -int mdt_attr_get_eabuf_size(struct mdt_thread_info *info, struct mdt_object *o) +static int mdt_attr_get_eabuf_size(struct mdt_thread_info *info, + struct mdt_object *o) { const struct lu_env *env = info->mti_env; int rc, rc2; @@ -683,13 +747,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); @@ -775,25 +843,27 @@ out: static int mdt_getattr_internal(struct mdt_thread_info *info, struct mdt_object *o, int ma_need) { - struct md_object *next = mdt_object_child(o); - const struct mdt_body *reqbody = info->mti_body; - struct ptlrpc_request *req = mdt_info_req(info); - struct md_attr *ma = &info->mti_attr; - struct lu_attr *la = &ma->ma_attr; - struct req_capsule *pill = info->mti_pill; - const struct lu_env *env = info->mti_env; - struct mdt_body *repbody; - struct lu_buf *buffer = &info->mti_buf; - int rc; - int is_root; - ENTRY; + struct md_object *next = mdt_object_child(o); + const struct mdt_body *reqbody = info->mti_body; + struct ptlrpc_request *req = mdt_info_req(info); + struct md_attr *ma = &info->mti_attr; + struct lu_attr *la = &ma->ma_attr; + struct req_capsule *pill = info->mti_pill; + const struct lu_env *env = info->mti_env; + struct mdt_body *repbody; + struct lu_buf *buffer = &info->mti_buf; + struct obd_export *exp = info->mti_exp; + struct lu_nodemap *nodemap = exp->exp_target_data.ted_nodemap; + int rc; + int is_root; + ENTRY; - if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) - RETURN(err_serious(-ENOMEM)); + if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) + RETURN(err_serious(-ENOMEM)); - repbody = req_capsule_server_get(pill, &RMF_MDT_BODY); + repbody = req_capsule_server_get(pill, &RMF_MDT_BODY); - ma->ma_valid = 0; + ma->ma_valid = 0; if (mdt_object_remote(o)) { /* This object is located on remote node.*/ @@ -993,33 +1063,8 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, } #ifdef CONFIG_FS_POSIX_ACL else if ((exp_connect_flags(req->rq_export) & OBD_CONNECT_ACL) && - (reqbody->mbo_valid & OBD_MD_FLACL)) { - buffer->lb_buf = req_capsule_server_get(pill, &RMF_ACL); - buffer->lb_len = req_capsule_get_size(pill, - &RMF_ACL, RCL_SERVER); - if (buffer->lb_len > 0) { - rc = mo_xattr_get(env, next, buffer, - XATTR_NAME_ACL_ACCESS); - if (rc < 0) { - if (rc == -ENODATA) { - repbody->mbo_aclsize = 0; - repbody->mbo_valid |= OBD_MD_FLACL; - rc = 0; - } else if (rc == -EOPNOTSUPP) { - rc = 0; - } else { - CERROR("%s: unable to read "DFID - " ACL: rc = %d\n", - mdt_obd_name(info->mti_mdt), - PFID(mdt_object_fid(o)), rc); - } - } else { - repbody->mbo_aclsize = rc; - repbody->mbo_valid |= OBD_MD_FLACL; - rc = 0; - } - } - } + (reqbody->mbo_valid & OBD_MD_FLACL)) + rc = mdt_pack_acl2body(info, repbody, o, nodemap); #endif if (reqbody->mbo_valid & OBD_MD_FLMDSCAPA && @@ -1126,7 +1171,7 @@ static int mdt_getattr(struct tgt_session_info *tsi) } if (rc < 0) - GOTO(out_shrink, rc); + GOTO(out, rc = err_serious(rc)); req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, rc); @@ -1191,7 +1236,7 @@ static int mdt_swap_layouts(struct tgt_session_info *tsi) info = tsi2mdt_info(tsi); if (info->mti_dlm_req != NULL) - ldlm_request_cancel(req, info->mti_dlm_req, 0); + 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->mbo_fid1, @@ -1265,43 +1310,32 @@ out: } static int mdt_raw_lookup(struct mdt_thread_info *info, - struct mdt_object *parent, - const struct lu_name *lname, - struct ldlm_reply *ldlm_rep) + struct mdt_object *parent, + const struct lu_name *lname, + struct ldlm_reply *ldlm_rep) { - struct md_object *next = mdt_object_child(info->mti_object); - const struct mdt_body *reqbody = info->mti_body; - struct lu_fid *child_fid = &info->mti_tmp_fid1; - struct mdt_body *repbody; - int rc; - ENTRY; - - if (reqbody->mbo_valid != OBD_MD_FLID) - RETURN(0); + struct lu_fid *child_fid = &info->mti_tmp_fid1; + int rc; + ENTRY; - LASSERT(!info->mti_cross_ref); + LASSERT(!info->mti_cross_ref); - /* Only got the fid of this obj by name */ - fid_zero(child_fid); - rc = mdo_lookup(info->mti_env, next, lname, child_fid, - &info->mti_spec); -#if 0 - /* XXX is raw_lookup possible as intent operation? */ - if (rc != 0) { - if (rc == -ENOENT) - mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG); - RETURN(rc); - } else - mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS); + /* Only got the fid of this obj by name */ + fid_zero(child_fid); + rc = mdo_lookup(info->mti_env, mdt_object_child(info->mti_object), + lname, child_fid, &info->mti_spec); + if (rc == 0) { + struct mdt_body *repbody; - repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); -#endif - if (rc == 0) { - repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); + repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); repbody->mbo_fid1 = *child_fid; repbody->mbo_valid = OBD_MD_FLID; - } - RETURN(1); + mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS); + } else if (rc == -ENOENT) { + mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG); + } + + RETURN(rc); } /* @@ -1319,7 +1353,6 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, struct mdt_body *reqbody = NULL; struct mdt_object *parent = info->mti_object; struct mdt_object *child; - struct md_object *next = mdt_object_child(parent); struct lu_fid *child_fid = &info->mti_tmp_fid1; struct lu_name *lname = NULL; struct mdt_lock_handle *lhp = NULL; @@ -1334,7 +1367,8 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, LASSERT(ergo(is_resent, lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT)); - LASSERT(parent != NULL); + if (parent == NULL) + RETURN(-ENOENT); if (info->mti_cross_ref) { /* Only getattr on the child. Parent is on another node. */ @@ -1347,7 +1381,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, rc = mdt_check_resent_lock(info, child, lhc); if (rc < 0) { - RETURN(-EPROTO); + RETURN(rc); } else if (rc > 0) { mdt_lock_handle_init(lhc); mdt_lock_reg_init(lhc, LCK_PR); @@ -1423,10 +1457,15 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, } if (lu_name_is_valid(lname)) { - rc = mdt_raw_lookup(info, parent, lname, ldlm_rep); - if (rc != 0) { - if (rc > 0) - rc = 0; + /* Always allow to lookup ".." */ + if (unlikely(lname->ln_namelen == 2 && + lname->ln_name[0] == '.' && + lname->ln_name[1] == '.')) + info->mti_spec.sp_permitted = 1; + + if (info->mti_body->mbo_valid == OBD_MD_FLID) { + rc = mdt_raw_lookup(info, parent, lname, ldlm_rep); + RETURN(rc); } @@ -1443,8 +1482,8 @@ 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); @@ -1899,19 +1938,20 @@ static int mdt_device_sync(const struct lu_env *env, struct mdt_device *mdt) /* this should sync this object */ static int mdt_object_sync(struct mdt_thread_info *info) { - struct md_object *next; - int rc; - ENTRY; + struct md_object *next; + int rc; + ENTRY; - if (!mdt_object_exists(info->mti_object)) { - CWARN("Non existing object "DFID"!\n", - PFID(mdt_object_fid(info->mti_object))); - RETURN(-ESTALE); - } - next = mdt_object_child(info->mti_object); - rc = mo_object_sync(info->mti_env, next); + if (!mdt_object_exists(info->mti_object)) { + CWARN("%s: non existing object "DFID": rc = %d\n", + mdt_obd_name(info->mti_mdt), + PFID(mdt_object_fid(info->mti_object)), -ESTALE); + RETURN(-ESTALE); + } + next = mdt_object_child(info->mti_object); + rc = mo_object_sync(info->mti_env, next); - RETURN(rc); + RETURN(rc); } static int mdt_sync(struct tgt_session_info *tsi) @@ -1968,6 +2008,7 @@ static int mdt_quotactl(struct tgt_session_info *tsi) int id, rc; struct mdt_device *mdt = mdt_exp2dev(exp); struct lu_device *qmt = mdt->mdt_qmt_dev; + struct lu_nodemap *nodemap = exp->exp_target_data.ted_nodemap; ENTRY; oqctl = req_capsule_client_get(pill, &RMF_OBD_QUOTACTL); @@ -1988,9 +2029,11 @@ static int mdt_quotactl(struct tgt_session_info *tsi) /* deprecated, not used any more */ RETURN(-EOPNOTSUPP); /* master quotactl */ - case Q_GETINFO: case Q_SETINFO: case Q_SETQUOTA: + if (!nodemap_can_setquota(nodemap)) + RETURN(-EPERM); + case Q_GETINFO: case Q_GETQUOTA: if (qmt == NULL) RETURN(-EOPNOTSUPP); @@ -2029,6 +2072,13 @@ static int mdt_quotactl(struct tgt_session_info *tsi) } } + if (oqctl->qc_type == USRQUOTA) + id = nodemap_map_id(nodemap, NODEMAP_UID, + NODEMAP_CLIENT_TO_FS, id); + else if (oqctl->qc_type == GRPQUOTA) + id = nodemap_map_id(nodemap, NODEMAP_UID, + NODEMAP_CLIENT_TO_FS, id); + repoqc = req_capsule_server_get(pill, &RMF_OBD_QUOTACTL); if (repoqc == NULL) RETURN(err_serious(-EFAULT)); @@ -2195,12 +2245,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); } /** @@ -2275,10 +2326,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); @@ -2316,14 +2368,20 @@ int mdt_check_resent_lock(struct mdt_thread_info *info, struct mdt_lock_handle *lhc) { /* the lock might already be gotten in ldlm_handle_enqueue() */ - if (lustre_handle_is_used(&lhc->mlh_reg_lh)) { + if (unlikely(lustre_handle_is_used(&lhc->mlh_reg_lh))) { struct ptlrpc_request *req = mdt_info_req(info); struct ldlm_lock *lock; lock = ldlm_handle2lock(&lhc->mlh_reg_lh); LASSERT(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT); - LASSERTF(lock != NULL, "Invalid lock handle "LPX64"\n", - lhc->mlh_reg_lh.cookie); + if (lock == NULL) { + /* Lock is pinned by ldlm_handle_enqueue0() as it is + * a resend case, however, it could be already destroyed + * due to client eviction or a raced cancel RPC. */ + LDLM_DEBUG_NOLOCK("Invalid lock handle "LPX64"\n", + lhc->mlh_reg_lh.cookie); + RETURN(-ESTALE); + } if (!fid_res_name_eq(mdt_object_fid(mo), &lock->l_resource->lr_name)) { @@ -2431,8 +2489,8 @@ static int mdt_object_local_lock(struct mdt_thread_info *info, policy, res_id, dlmflags, info->mti_exp == NULL ? NULL : &info->mti_exp->exp_handle.h_cookie); - if (unlikely(rc)) - RETURN(rc); + if (unlikely(rc != 0)) + GOTO(out_unlock, rc); } /* @@ -2453,15 +2511,15 @@ static int mdt_object_local_lock(struct mdt_thread_info *info, res_id, LDLM_FL_LOCAL_ONLY | dlmflags, info->mti_exp == NULL ? NULL : &info->mti_exp->exp_handle.h_cookie); - if (rc) - mdt_object_unlock(info, o, lh, 1); - else if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_PDO_LOCK)) && - 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); - } +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 @@ -2812,6 +2870,7 @@ void mdt_thread_info_init(struct ptlrpc_request *req, info->mti_spec.no_create = 0; info->mti_spec.sp_rm_entry = 0; + info->mti_spec.sp_permitted = 0; info->mti_spec.u.sp_ea.eadata = NULL; info->mti_spec.u.sp_ea.eadatalen = 0; @@ -2875,6 +2934,12 @@ static int mdt_tgt_connect(struct tgt_session_info *tsi) ENTRY; + if (OBD_FAIL_CHECK(OBD_FAIL_TGT_DELAY_CONDITIONAL) && + cfs_fail_val == + tsi2mdt_info(tsi)->mti_mdt->mdt_seq_site.ss_node_id) + schedule_timeout_and_set_state(TASK_UNINTERRUPTIBLE, + msecs_to_jiffies(3 * MSEC_PER_SEC)); + rc = tgt_connect(tsi); if (rc != 0) RETURN(rc); @@ -3336,8 +3401,8 @@ 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)) { - LASSERT(lustre_handle_is_used(&lhc->mlh_reg_lh)); + if (lustre_handle_is_used(&lhc->mlh_reg_lh) && + (rc == 0 || rc == -MDT_EREMOTE_OPEN)) { rep->lock_policy_res2 = 0; rc = mdt_intent_lock_replace(info, lockp, lhc, flags); RETURN(rc); @@ -3349,6 +3414,7 @@ static int mdt_intent_reint(enum mdt_it_code opcode, mdt_get_disposition(rep, DISP_LOOKUP_NEG)) rep->lock_policy_res2 = 0; + lhc->mlh_reg_lh.cookie = 0ull; if (rc == -ENOTCONN || rc == -ENODEV || rc == -EOVERFLOW) { /**< if VBR failure then return error */ /* @@ -3357,28 +3423,13 @@ static int mdt_intent_reint(enum mdt_it_code opcode, * will detect this, then disconnect, reconnect the import * immediately, instead of impacting the following the rpc. */ - lhc->mlh_reg_lh.cookie = 0ull; RETURN(rc); - } else { - /* - * For other cases, the error will be returned by intent. - * and client will retrieve the result from intent. - */ - /* - * FIXME: when open lock is finished, that should be - * checked here. - */ - if (lustre_handle_is_used(&lhc->mlh_reg_lh)) { - LASSERTF(rc == 0, "Error occurred but lock handle " - "is still in use, rc = %d\n", rc); - rep->lock_policy_res2 = 0; - rc = mdt_intent_lock_replace(info, lockp, lhc, flags); - RETURN(rc); - } else { - lhc->mlh_reg_lh.cookie = 0ull; - RETURN(ELDLM_LOCK_ABORTED); - } } + /* + * For other cases, the error will be returned by intent, and client + * will retrieve the result from intent. + */ + RETURN(ELDLM_LOCK_ABORTED); } static int mdt_intent_code(long itcode) @@ -3431,17 +3482,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; @@ -3457,33 +3507,33 @@ 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 = -EPROTO; } + RETURN(rc); } @@ -3806,10 +3856,9 @@ static void mdt_stack_pre_fini(const struct lu_env *env, lustre_cfg_bufs_reset(bufs, mdt_obd_name(m)); lustre_cfg_bufs_set_string(bufs, 1, NULL); lcfg = lustre_cfg_new(LCFG_PRE_CLEANUP, bufs); - if (!lcfg) { - CERROR("%s:Cannot alloc lcfg!\n", mdt_obd_name(m)); - return; - } + if (lcfg == NULL) + RETURN_EXIT; + top->ld_ops->ldo_process_config(env, top, lcfg); lustre_cfg_free(lcfg); EXIT; @@ -3842,10 +3891,9 @@ static void mdt_stack_fini(const struct lu_env *env, strcat(flags, "A"); lustre_cfg_bufs_set_string(bufs, 1, flags); lcfg = lustre_cfg_new(LCFG_CLEANUP, bufs); - if (!lcfg) { - CERROR("Cannot alloc lcfg!\n"); - return; - } + if (lcfg == NULL) + RETURN_EXIT; + LASSERT(top); top->ld_ops->ldo_process_config(env, top, lcfg); lustre_cfg_free(lcfg); @@ -3969,7 +4017,7 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt, lustre_cfg_bufs_set_string(bufs, 3, lprof->lp_dt); lcfg = lustre_cfg_new(LCFG_ATTACH, bufs); - if (!lcfg) + if (lcfg == NULL) GOTO(free_bufs, rc = -ENOMEM); rc = class_attach(lcfg); @@ -3980,7 +4028,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); @@ -3991,6 +4039,8 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt, lustre_cfg_bufs_set_string(bufs, 3, lprof->lp_dt); lcfg = lustre_cfg_new(LCFG_SETUP, bufs); + if (lcfg == NULL) + GOTO(class_detach, rc = -ENOMEM); rc = class_setup(obd, lcfg); if (rc) @@ -4097,7 +4147,7 @@ static int mdt_quota_init(const struct lu_env *env, struct mdt_device *mdt, lustre_cfg_bufs_set_string(bufs, 3, lprof->lp_dt); lcfg = lustre_cfg_new(LCFG_ATTACH, bufs); - if (!lcfg) + if (lcfg == NULL) GOTO(cleanup_mem, rc = -ENOMEM); rc = class_attach(lcfg); @@ -4108,7 +4158,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); @@ -4122,6 +4172,8 @@ static int mdt_quota_init(const struct lu_env *env, struct mdt_device *mdt, mdt->mdt_bottom->dd_lu_dev.ld_obd->obd_name); lcfg = lustre_cfg_new(LCFG_SETUP, bufs); + if (lcfg == NULL) + GOTO(class_detach, rc = -ENOMEM); rc = class_setup(obd, lcfg); if (rc) @@ -4189,7 +4241,7 @@ static void mdt_quota_fini(const struct lu_env *env, struct mdt_device *mdt) /* mdt_getxattr() is used from mdt_intent_getxattr(), use this wrapper * for now. This will be removed along with converting rest of MDT code * to use tgt_session_info */ -int mdt_tgt_getxattr(struct tgt_session_info *tsi) +static int mdt_tgt_getxattr(struct tgt_session_info *tsi) { struct mdt_thread_info *info = tsi2mdt_info(tsi); int rc; @@ -4682,13 +4734,13 @@ static int mdt_process_config(const struct lu_env *env, } } - rc = class_process_proc_seq_param(PARAM_MDT, obd->obd_vars, - cfg, obd); + rc = class_process_proc_param(PARAM_MDT, obd->obd_vars, + cfg, obd); if (rc > 0 || rc == -ENOSYS) { /* is it an HSM var ? */ - rc = class_process_proc_seq_param(PARAM_HSM, - hsm_cdt_get_proc_vars(), - cfg, obd); + 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, @@ -4898,12 +4950,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; @@ -4975,90 +5029,6 @@ static int mdt_connect_internal(struct obd_export *exp, return 0; } -/* mds_connect copy */ -static int mdt_obd_connect(const struct lu_env *env, - struct obd_export **exp, struct obd_device *obd, - struct obd_uuid *cluuid, - struct obd_connect_data *data, - void *localdata) -{ - struct obd_export *lexp; - struct lustre_handle conn = { 0 }; - struct mdt_device *mdt; - int rc; - ENTRY; - - LASSERT(env != NULL); - if (!exp || !obd || !cluuid) - RETURN(-EINVAL); - - mdt = mdt_dev(obd->obd_lu_dev); - - /* - * first, check whether the stack is ready to handle requests - * XXX: probably not very appropriate method is used now - * at some point we should find a better one - */ - if (!test_bit(MDT_FL_SYNCED, &mdt->mdt_state) && data != NULL && - !(data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT)) { - rc = obd_get_info(env, mdt->mdt_child_exp, - sizeof(KEY_OSP_CONNECTED), - KEY_OSP_CONNECTED, NULL, NULL, NULL); - if (rc) - RETURN(-EAGAIN); - set_bit(MDT_FL_SYNCED, &mdt->mdt_state); - } - - rc = class_connect(&conn, obd, cluuid); - if (rc) - RETURN(rc); - - lexp = class_conn2export(&conn); - LASSERT(lexp != NULL); - - rc = mdt_connect_internal(lexp, mdt, data); - if (rc == 0) { - struct lsd_client_data *lcd = lexp->exp_target_data.ted_lcd; - - LASSERT(lcd); - memcpy(lcd->lcd_uuid, cluuid, sizeof lcd->lcd_uuid); - rc = tgt_client_new(env, lexp); - if (rc == 0) - mdt_export_stats_init(obd, lexp, localdata); - - /* For phase I, sync for cross-ref operation. */ - lexp->exp_keep_sync = 1; - } - - if (rc != 0) { - class_disconnect(lexp); - *exp = NULL; - } else { - *exp = lexp; - } - - RETURN(rc); -} - -static int mdt_obd_reconnect(const struct lu_env *env, - struct obd_export *exp, struct obd_device *obd, - struct obd_uuid *cluuid, - struct obd_connect_data *data, - void *localdata) -{ - int rc; - ENTRY; - - if (exp == NULL || obd == NULL || cluuid == NULL) - RETURN(-EINVAL); - - rc = mdt_connect_internal(exp, mdt_dev(obd->obd_lu_dev), data); - if (rc == 0) - mdt_export_stats_init(obd, exp, localdata); - - RETURN(rc); -} - static int mdt_ctxt_add_dirty_flag(struct lu_env *env, struct mdt_thread_info *info, struct mdt_file_data *mfd) @@ -5086,25 +5056,26 @@ static int mdt_ctxt_add_dirty_flag(struct lu_env *env, 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); @@ -5121,12 +5092,13 @@ 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 @@ -5172,23 +5144,120 @@ 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); + nodemap_del_member(exp); + 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); +} + +/* mds_connect copy */ +static int mdt_obd_connect(const struct lu_env *env, + struct obd_export **exp, struct obd_device *obd, + struct obd_uuid *cluuid, + struct obd_connect_data *data, + void *localdata) +{ + struct obd_export *lexp; + struct lustre_handle conn = { 0 }; + struct mdt_device *mdt; + int rc; + lnet_nid_t *client_nid = localdata; + ENTRY; + + LASSERT(env != NULL); + if (!exp || !obd || !cluuid) + RETURN(-EINVAL); + + mdt = mdt_dev(obd->obd_lu_dev); + + /* + * first, check whether the stack is ready to handle requests + * XXX: probably not very appropriate method is used now + * at some point we should find a better one + */ + if (!test_bit(MDT_FL_SYNCED, &mdt->mdt_state) && data != NULL && + !(data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT)) { + rc = obd_get_info(env, mdt->mdt_child_exp, + sizeof(KEY_OSP_CONNECTED), + KEY_OSP_CONNECTED, NULL, NULL, NULL); + if (rc) + RETURN(-EAGAIN); + set_bit(MDT_FL_SYNCED, &mdt->mdt_state); + } + + rc = class_connect(&conn, obd, cluuid); + if (rc) + RETURN(rc); + + lexp = class_conn2export(&conn); + LASSERT(lexp != NULL); + + rc = mdt_connect_internal(lexp, mdt, data); + if (rc == 0) { + struct lsd_client_data *lcd = lexp->exp_target_data.ted_lcd; + + LASSERT(lcd); + memcpy(lcd->lcd_uuid, cluuid, sizeof lcd->lcd_uuid); + rc = tgt_client_new(env, lexp); + if (rc == 0) { + rc = nodemap_add_member(*client_nid, lexp); + if (rc != 0 && rc != -EEXIST) + goto out; + + 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; + } else { + *exp = lexp; + } + + RETURN(rc); +} + +static int mdt_obd_reconnect(const struct lu_env *env, + struct obd_export *exp, struct obd_device *obd, + struct obd_uuid *cluuid, + struct obd_connect_data *data, + void *localdata) +{ + lnet_nid_t *client_nid = localdata; + int rc; + ENTRY; + + if (exp == NULL || obd == NULL || cluuid == NULL) + RETURN(-EINVAL); + + rc = mdt_connect_internal(exp, mdt_dev(obd->obd_lu_dev), data); + if (rc == 0) { + rc = nodemap_add_member(*client_nid, exp); + if (rc == 0 || rc == -EEXIST) + mdt_export_stats_init(obd, exp, localdata); + } + + 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; @@ -5233,34 +5302,15 @@ 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. - * This is limited only because we want to store the fids for - * historical path lookup purposes. - */ -#define MAX_PATH_DEPTH 100 - -/** mdt_path() lookup structure. */ -struct path_lookup_info { - __u64 pli_recno; /**< history point */ - __u64 pli_currec; /**< current record */ - struct lu_fid pli_fid; - struct lu_fid pli_fids[MAX_PATH_DEPTH]; /**< path, in fids */ - struct mdt_object *pli_mdt_obj; - char *pli_path; /**< full path */ - int pli_pathlen; - int pli_linkno; /**< which hardlink to follow */ - int pli_fidcount; /**< number of \a pli_fids */ -}; - int mdt_links_read(struct mdt_thread_info *info, struct mdt_object *mdt_obj, struct linkea_data *ldata) { @@ -5292,8 +5342,21 @@ int mdt_links_read(struct mdt_thread_info *info, struct mdt_object *mdt_obj, return linkea_init(ldata); } +/** + * Given an MDT object, try to look up the full path to the object. + * Part of the MDT layer implementation of lfs fid2path. + * + * \param[in] info Per-thread common data shared by MDT level handlers. + * \param[in] obj Object to do path lookup of + * \param[in,out] fp User-provided struct to store path information + * + * \retval 0 Lookup successful, path information stored in fp + * \retval -EAGAIN Lookup failed, usually because object is being moved + * \retval negative errno if there was a problem + */ static int mdt_path_current(struct mdt_thread_info *info, - struct path_lookup_info *pli) + struct mdt_object *obj, + struct getinfo_fid2path *fp) { struct mdt_device *mdt = info->mti_mdt; struct mdt_object *mdt_obj; @@ -5306,6 +5369,7 @@ static int mdt_path_current(struct mdt_thread_info *info, int reclen; struct linkea_data ldata = { 0 }; int rc = 0; + bool first = true; ENTRY; /* temp buffer for path element, the buffer will be finally freed @@ -5315,16 +5379,14 @@ static int mdt_path_current(struct mdt_thread_info *info, RETURN(-ENOMEM); ldata.ld_buf = buf; - ptr = pli->pli_path + pli->pli_pathlen - 1; + ptr = fp->gf_path + fp->gf_pathlen - 1; *ptr = 0; --ptr; - pli->pli_fidcount = 0; - pli->pli_fids[0] = *(struct lu_fid *)mdt_object_fid(pli->pli_mdt_obj); - *tmpfid = pli->pli_fids[0]; + *tmpfid = fp->gf_fid = *mdt_object_fid(obj); + /* root FID only exists on MDT0, and fid2path should also ends at MDT0, * so checking root_fid can only happen on MDT0. */ - while (!lu_fid_eq(&mdt->mdt_md_root_fid, - &pli->pli_fids[pli->pli_fidcount])) { + while (!lu_fid_eq(&mdt->mdt_md_root_fid, &fp->gf_fid)) { struct lu_buf lmv_buf; mdt_obj = mdt_object_find(info->mti_env, mdt, tmpfid); @@ -5352,18 +5414,17 @@ static int mdt_path_current(struct mdt_thread_info *info, linkea_entry_unpack(lee, &reclen, tmpname, tmpfid); /* If set, use link #linkno for path lookup, otherwise use link #0. Only do this for the final path element. */ - if (pli->pli_fidcount == 0 && - pli->pli_linkno < leh->leh_reccount) { + if (first && fp->gf_linkno < leh->leh_reccount) { int count; - for (count = 0; count < pli->pli_linkno; count++) { + for (count = 0; count < fp->gf_linkno; count++) { lee = (struct link_ea_entry *) ((char *)lee + reclen); linkea_entry_unpack(lee, &reclen, tmpname, tmpfid); } - if (pli->pli_linkno < leh->leh_reccount - 1) + if (fp->gf_linkno < leh->leh_reccount - 1) /* indicate to user there are more links */ - pli->pli_linkno++; + fp->gf_linkno++; } lmv_buf.lb_buf = info->mti_xattr_buf; @@ -5377,7 +5438,7 @@ static int mdt_path_current(struct mdt_thread_info *info, /* For slave stripes, get its master */ if (le32_to_cpu(lmm->lmv_magic) == LMV_MAGIC_STRIPE) { - pli->pli_fids[pli->pli_fidcount] = *tmpfid; + fp->gf_fid = *tmpfid; continue; } } else if (rc < 0 && rc != -ENODATA) { @@ -5388,72 +5449,81 @@ static int mdt_path_current(struct mdt_thread_info *info, /* Pack the name in the end of the buffer */ ptr -= tmpname->ln_namelen; - if (ptr - 1 <= pli->pli_path) + if (ptr - 1 <= fp->gf_path) GOTO(out, rc = -EOVERFLOW); strncpy(ptr, tmpname->ln_name, tmpname->ln_namelen); *(--ptr) = '/'; - /* Store the parent fid for historic lookup */ - if (++pli->pli_fidcount >= MAX_PATH_DEPTH) - GOTO(out, rc = -EOVERFLOW); - pli->pli_fids[pli->pli_fidcount] = *tmpfid; + /* keep the last resolved fid to the client, so the + * client will build the left path on another MDT for + * remote object */ + fp->gf_fid = *tmpfid; + + first = false; } remote_out: ptr++; /* skip leading / */ - memmove(pli->pli_path, ptr, pli->pli_path + pli->pli_pathlen - ptr); + memmove(fp->gf_path, ptr, fp->gf_path + fp->gf_pathlen - ptr); - EXIT; out: - return rc; + RETURN(rc); } -/* Returns the full path to this fid, as of changelog record recno. */ +/** + * Given an MDT object, use mdt_path_current to get the path. + * Essentially a wrapper to retry mdt_path_current a set number of times + * if -EAGAIN is returned (usually because an object is being moved). + * + * Part of the MDT layer implementation of lfs fid2path. + * + * \param[in] info Per-thread common data shared by mdt level handlers. + * \param[in] obj Object to do path lookup of + * \param[in,out] fp User-provided struct for arguments and to store path + * information + * + * \retval 0 Lookup successful, path information stored in fp + * \retval negative errno if there was a problem + */ static int mdt_path(struct mdt_thread_info *info, struct mdt_object *obj, - char *path, int pathlen, __u64 *recno, int *linkno, - struct lu_fid *fid) + struct getinfo_fid2path *fp) { struct mdt_device *mdt = info->mti_mdt; - struct path_lookup_info *pli; int tries = 3; int rc = -EAGAIN; ENTRY; - if (pathlen < 3) + if (fp->gf_pathlen < 3) RETURN(-EOVERFLOW); if (lu_fid_eq(&mdt->mdt_md_root_fid, mdt_object_fid(obj))) { - path[0] = '\0'; + fp->gf_path[0] = '\0'; RETURN(0); } - OBD_ALLOC_PTR(pli); - if (pli == NULL) - RETURN(-ENOMEM); - - pli->pli_mdt_obj = obj; - pli->pli_recno = *recno; - pli->pli_path = path; - pli->pli_pathlen = pathlen; - pli->pli_linkno = *linkno; - /* Retry multiple times in case file is being moved */ while (tries-- && rc == -EAGAIN) - rc = mdt_path_current(info, pli); - - /* return the last resolved fids to the client, so the client will - * build the left path on another MDT for remote object */ - *fid = pli->pli_fids[pli->pli_fidcount]; - - *recno = pli->pli_currec; - /* Return next link index to caller */ - *linkno = pli->pli_linkno; - - OBD_FREE_PTR(pli); + rc = mdt_path_current(info, obj, fp); RETURN(rc); } +/** + * Get the full path of the provided FID, as of changelog record recno. + * + * This checks sanity and looks up object for user provided FID + * before calling the actual path lookup code. + * + * Part of the MDT layer implementation of lfs fid2path. + * + * \param[in] info Per-thread common data shared by mdt level handlers. + * \param[in,out] fp User-provided struct for arguments and to store path + * information + * + * \retval 0 Lookup successful, path information and recno stored in fp + * \retval -ENOENT, object does not exist + * \retval negative errno if there was a problem + */ static int mdt_fid2path(struct mdt_thread_info *info, struct getinfo_fid2path *fp) { @@ -5491,13 +5561,12 @@ 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); } - rc = mdt_path(info, obj, fp->gf_path, fp->gf_pathlen, &fp->gf_recno, - &fp->gf_linkno, &fp->gf_fid); + rc = mdt_path(info, obj, fp); CDEBUG(D_INFO, "fid "DFID", path %s recno "LPX64" linkno %u\n", PFID(&fp->gf_fid), fp->gf_path, fp->gf_recno, fp->gf_linkno); @@ -5539,7 +5608,7 @@ int mdt_get_info(struct tgt_session_info *tsi) key = req_capsule_client_get(tsi->tsi_pill, &RMF_GETINFO_KEY); if (key == NULL) { - CDEBUG(D_IOCTL, "No GETINFO key"); + CDEBUG(D_IOCTL, "No GETINFO key\n"); RETURN(err_serious(-EFAULT)); } keylen = req_capsule_get_size(tsi->tsi_pill, &RMF_GETINFO_KEY, @@ -5547,7 +5616,8 @@ int mdt_get_info(struct tgt_session_info *tsi) vallen = req_capsule_client_get(tsi->tsi_pill, &RMF_GETINFO_VALLEN); if (vallen == NULL) { - CDEBUG(D_IOCTL, "Unable to get RMF_GETINFO_VALLEN buffer"); + CDEBUG(D_IOCTL, "%s: cannot get RMF_GETINFO_VALLEN buffer\n", + tgt_name(tsi->tsi_tgt)); RETURN(err_serious(-EFAULT)); } @@ -5559,7 +5629,8 @@ int mdt_get_info(struct tgt_session_info *tsi) valout = req_capsule_server_get(tsi->tsi_pill, &RMF_GETINFO_VAL); if (valout == NULL) { - CDEBUG(D_IOCTL, "Unable to get get-info RPC out buffer"); + CDEBUG(D_IOCTL, "%s: cannot get get-info RPC out buffer\n", + tgt_name(tsi->tsi_tgt)); RETURN(err_serious(-EFAULT)); } @@ -5873,13 +5944,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); } /** @@ -5924,9 +5995,6 @@ static int __init mdt_mod_init(void) GOTO(lu_fini, rc); rc = class_register_type(&mdt_obd_device_ops, NULL, true, NULL, -#ifndef HAVE_ONLY_PROCFS_SEQ - NULL, -#endif LUSTRE_MDT_NAME, &mdt_device_type); if (rc) GOTO(mds_fini, rc);