X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_handler.c;h=0de83de0b06d8c8634376d0773ba499e41a116f7;hb=9a2bdd6ca1e441e3b019725c30dc460be0853216;hp=d13acbb83887582d65e89393db1fd7f7cdd914d3;hpb=bca113c04840b05aa1261b52769836add3787f8c;p=fs%2Flustre-release.git diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c index d13acbb..0de83de 100644 --- a/lustre/mdt/mdt_handler.c +++ b/lustre/mdt/mdt_handler.c @@ -62,7 +62,6 @@ /* lu2dt_dev() */ #include #include -#include #include #include "mdt_internal.h" #include @@ -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 } }; @@ -396,7 +411,12 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b, 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 != 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; + } if (fid) { b->fid1 = *fid; @@ -734,12 +754,19 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, RETURN(rc); } + /* 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; + } + is_root = lu_fid_eq(mdt_object_fid(o), &info->mti_mdt->mdt_md_root_fid); /* 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; @@ -1317,35 +1344,33 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, */ 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: %lu/%lu/%lu, Fid: "DFID".\n", - (unsigned long)res_id->name[0], - (unsigned long)res_id->name[1], - (unsigned long)res_id->name[2], - 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 { + 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: @@ -1427,17 +1452,15 @@ relock: rc = mdt_getattr_internal(info, child, ma_need); if (unlikely(rc != 0)) { 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), - "Lock res_id: %lu/%lu/%lu, Fid: "DFID".\n", - (unsigned long)res_id->name[0], - (unsigned long)res_id->name[1], - (unsigned long)res_id->name[2], - PFID(mdt_object_fid(child))); + } 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), + "Lock res_id: "DLDLMRES", fid: "DFID"\n", + PLDLMRES(lock->l_resource), + PFID(mdt_object_fid(child))); if (mdt_object_exists(child) && !mdt_object_remote(child)) mdt_pack_size2body(info, child); } @@ -1559,6 +1582,9 @@ int mdt_set_info(struct mdt_thread_info *info) RETURN(0); } +int mdt_connect_check_sptlrpc(struct mdt_device *mdt, struct obd_export *exp, + struct ptlrpc_request *req); + /** * Top-level handler for MDT connection requests. */ @@ -1569,32 +1595,39 @@ int mdt_connect(struct mdt_thread_info *info) struct obd_export *exp; struct ptlrpc_request *req = mdt_info_req(info); + ENTRY; + rc = target_handle_connect(req); if (rc != 0) - return err_serious(rc); + RETURN(err_serious(rc)); LASSERT(req->rq_export != NULL); - info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev); + exp = req->rq_export; + info->mti_exp = exp; + info->mti_mdt = mdt_dev(exp->exp_obd->obd_lu_dev); rc = mdt_init_sec_level(info); - if (rc != 0) { - obd_disconnect(class_export_get(req->rq_export)); - return rc; - } + if (rc != 0) + GOTO(err, rc); + + rc = mdt_connect_check_sptlrpc(info->mti_mdt, exp, req); + if (rc) + GOTO(err, rc); /* 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)); - + GOTO(err, rc); + RETURN(0); +err: + obd_disconnect(class_export_get(req->rq_export)); return rc; } @@ -2528,7 +2561,7 @@ int mdt_remote_object_lock(struct mdt_thread_info *mti, LASSERT(mdt_object_remote(o)); - LASSERT((ibits & MDS_INODELOCK_UPDATE)); + LASSERT(ibits & MDS_INODELOCK_UPDATE); memset(einfo, 0, sizeof(*einfo)); einfo->ei_type = LDLM_IBITS; @@ -3069,12 +3102,11 @@ 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 */ @@ -3107,11 +3139,10 @@ static void mdt_thread_info_init(struct ptlrpc_request *req, info->mti_spec.sp_rm_entry = 0; } -static void mdt_thread_info_fini(struct mdt_thread_info *info) +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; @@ -3120,11 +3151,41 @@ static void mdt_thread_info_fini(struct mdt_thread_info *info) 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; if (unlikely(info->mti_big_buf.lb_buf != NULL)) lu_buf_free(&info->mti_big_buf); } +int mdt_tgt_connect(struct tgt_session_info *tsi) +{ + struct ptlrpc_request *req = tgt_ses_req(tsi); + struct mdt_thread_info *mti; + int rc; + + ENTRY; + + rc = tgt_connect(tsi); + if (rc != 0) + RETURN(rc); + + /* XXX: switch mdt_init_idmap() to use tgt_session_info */ + lu_env_refill((void *)tsi->tsi_env); + mti = lu_context_key_get(&tsi->tsi_env->le_ctx, &mdt_thread_key); + LASSERT(mti != NULL); + + mdt_thread_info_init(req, mti); + rc = mdt_init_idmap(mti); + mdt_thread_info_fini(mti); + if (rc != 0) + GOTO(err, rc); + RETURN(0); +err: + obd_disconnect(class_export_get(req->rq_export)); + return rc; +} + static int mdt_filter_recovery_request(struct ptlrpc_request *req, struct obd_device *obd, int *process) { @@ -3399,11 +3460,13 @@ int mdt_handle_common(struct ptlrpc_request *req, info = lu_context_key_get(&env->le_ctx, &mdt_thread_key); LASSERT(info != NULL); + req_capsule_init(&req->rq_pill, req, RCL_SERVER); mdt_thread_info_init(req, info); rc = mdt_handle0(req, info, supported); mdt_thread_info_fini(info); + req_capsule_fini(&req->rq_pill); RETURN(rc); } @@ -3413,22 +3476,13 @@ int mdt_handle_common(struct ptlrpc_request *req, */ int mdt_recovery_handle(struct ptlrpc_request *req) { - int rc; - ENTRY; + 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); + rc = mdt_handle_common(req, mdt_regular_handlers); + + RETURN(rc); } enum mdt_it_code { @@ -3450,6 +3504,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 **, @@ -3477,7 +3537,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 }, @@ -3514,9 +3581,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, @@ -3667,6 +3734,44 @@ static void mdt_intent_fixup_resent(struct mdt_thread_info *info, remote_hdl.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, NULL, lhc); + 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, NULL, 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, struct mdt_thread_info *info, struct ldlm_lock **lockp, @@ -4588,6 +4693,56 @@ static void mdt_quota_fini(const struct lu_env *env, struct mdt_device *mdt) EXIT; } +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), +}; + +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 = tgt_sec_ctx_handlers + }, + { + .tos_opc_start = UPDATE_OBJ, + .tos_opc_end = 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_hs = NULL + } +}; + static void mdt_fini(const struct lu_env *env, struct mdt_device *m) { struct md_device *next = m->mdt_child; @@ -4599,8 +4754,14 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) 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); obd_zombie_barrier(); @@ -4689,7 +4850,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; @@ -4717,6 +4878,11 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, m->mdt_som_conf = 0; 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); @@ -4730,7 +4896,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, } rwlock_init(&m->mdt_sptlrpc_lock); - sptlrpc_rule_set_init(&m->mdt_sptlrpc_rset); + sptlrpc_rule_set_init(&m->mdt_sptlrpc_rset); spin_lock_init(&m->mdt_ioepoch_lock); m->mdt_opts.mo_compat_resname = 0; @@ -4787,13 +4953,9 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, } } - rc = tgt_init(env, &m->mdt_lut, obd, m->mdt_bottom); - if (rc) - GOTO(err_fini_stack, rc); - rc = mdt_fld_init(env, mdt_obd_name(m), m); if (rc) - GOTO(err_lut, rc); + GOTO(err_fini_stack, rc); rc = mdt_seq_init(env, mdt_obd_name(m), m); if (rc) @@ -4817,13 +4979,26 @@ 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_ck_thread_start(m); - if (rc) + 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_fs_setup(env, m, obd, lsi); + rc = mdt_ck_thread_start(m); if (rc) - GOTO(err_capa, rc); + GOTO(err_free_hsm, rc); + + rc = tgt_init(env, &m->mdt_lut, obd, m->mdt_bottom, mdt_common_slice, + OBD_FAIL_MDS_ALL_REQUEST_NET, + OBD_FAIL_MDS_ALL_REPLY_NET); + if (rc) + GOTO(err_capa, rc); + + rc = mdt_fs_setup(env, m, obd, lsi); + if (rc) + GOTO(err_tgt, rc); mdt_adapt_sptlrpc_conf(obd, 1); @@ -4831,13 +5006,17 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, rc = next->md_ops->mdo_iocontrol(env, next, OBD_IOC_GET_MNTOPT, 0, &mntopts); if (rc) - GOTO(err_llog_cleanup, rc); + GOTO(err_fs_cleanup, rc); if (mntopts & MNTOPT_USERXATTR) m->mdt_opts.mo_user_xattr = 1; 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 @@ -4854,7 +5033,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, if (IS_ERR(m->mdt_identity_cache)) { rc = PTR_ERR(m->mdt_identity_cache); m->mdt_identity_cache = NULL; - GOTO(err_llog_cleanup, rc); + GOTO(err_fs_cleanup, rc); } rc = mdt_procfs_init(m, dev); @@ -4871,7 +5050,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL, "mdt_ldlm_client", m->mdt_ldlm_client); - ping_evictor_start(); + ping_evictor_start(); /* recovery will be started upon mdt_prepare() * when the whole stack is complete and ready @@ -4886,34 +5065,34 @@ 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); + mdt_procfs_fini(m); err_recovery: - target_recovery_fini(obd); - upcall_cache_cleanup(m->mdt_identity_cache); - m->mdt_identity_cache = NULL; -err_llog_cleanup: - mdt_llog_ctxt_unclone(env, m, LLOG_CHANGELOG_ORIG_CTXT); - mdt_fs_cleanup(env, m); + target_recovery_fini(obd); + upcall_cache_cleanup(m->mdt_identity_cache); + m->mdt_identity_cache = NULL; +err_fs_cleanup: + mdt_fs_cleanup(env, m); +err_tgt: + tgt_fini(env, &m->mdt_lut); err_capa: - cfs_timer_disarm(&m->mdt_ck_timer); - mdt_ck_thread_stop(m); + 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; + ldlm_namespace_free(m->mdt_namespace, NULL, 0); + obd->obd_namespace = m->mdt_namespace = NULL; err_fini_seq: mdt_seq_fini(env, m); err_fini_fld: mdt_fld_fini(env, m); -err_lut: - tgt_fini(env, &m->mdt_lut); err_fini_stack: - mdt_stack_fini(env, m, md2lu_dev(m->mdt_child)); + mdt_stack_fini(env, m, md2lu_dev(m->mdt_child)); err_lmi: if (lmi) server_put_mount(dev, lmi->lmi_mnt); - return (rc); + return(rc); } /* For interoperability, the left element is old parameter, the right one @@ -5016,9 +5195,11 @@ static struct lu_object *mdt_object_alloc(const struct lu_env *env, o->lo_ops = &mdt_obj_ops; mutex_init(&mo->mot_ioepoch_mutex); mutex_init(&mo->mot_lov_mutex); - RETURN(o); - } else - RETURN(NULL); + init_rwsem(&mo->mot_open_sem); + init_rwsem(&mo->mot_xattr_sem); + RETURN(o); + } + RETURN(NULL); } static int mdt_object_init(const struct lu_env *env, struct lu_object *o, @@ -5053,11 +5234,14 @@ static void mdt_object_free(const struct lu_env *env, struct lu_object *o) CDEBUG(D_INFO, "object free, fid = "DFID"\n", PFID(lu_object_fid(o))); - lu_object_fini(o); - lu_object_header_fini(h); + LASSERT(atomic_read(&mo->mot_open_count) == 0); + LASSERT(atomic_read(&mo->mot_lease_count) == 0); + + lu_object_fini(o); + lu_object_header_fini(h); OBD_SLAB_FREE_PTR(mo, mdt_object_kmem); - EXIT; + EXIT; } static int mdt_object_print(const struct lu_env *env, void *cookie, @@ -5092,6 +5276,10 @@ static int mdt_prepare(const struct lu_env *env, if (rc) RETURN(rc); + rc = mdt_llog_ctxt_clone(env, mdt, LLOG_AGENT_ORIG_CTXT); + 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, @@ -5109,6 +5297,7 @@ static int mdt_prepare(const struct lu_env *env, if (rc) RETURN(rc); } + LASSERT(!test_bit(MDT_FL_CFGLOG, &mdt->mdt_state)); target_recovery_init(&mdt->mdt_lut, mdt_recovery_handle); set_bit(MDT_FL_CFGLOG, &mdt->mdt_state); @@ -5259,12 +5448,13 @@ static int mdt_connect_internal(struct obd_export *exp, } } + data->ocd_max_easize = mdt->mdt_max_ea_size; + return 0; } -static int mdt_connect_check_sptlrpc(struct mdt_device *mdt, - struct obd_export *exp, - struct ptlrpc_request *req) +int mdt_connect_check_sptlrpc(struct mdt_device *mdt, struct obd_export *exp, + struct ptlrpc_request *req) { struct sptlrpc_flavor flvr; int rc = 0; @@ -5313,11 +5503,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; @@ -5325,9 +5513,7 @@ 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 @@ -5349,26 +5535,17 @@ 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); } -out: if (rc != 0) { class_disconnect(lexp); *exp = NULL; @@ -5385,23 +5562,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); @@ -5409,6 +5575,31 @@ 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_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; @@ -5453,6 +5644,24 @@ static int mdt_export_cleanup(struct obd_export *exp) cfs_list_for_each_entry_safe(mfd, n, &closing_list, mfd_list) { cfs_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; @@ -5655,8 +5864,8 @@ 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 (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 *)