X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_handler.c;h=001ef1de171b4b035dee7d55ce75e102117e89a4;hp=3583a2befde7aec7bcc040b8712faea179e8bee1;hb=a88ec6990b4bbdcf6c8481374ef8ecb98cb93cf0;hpb=f95393b0d0a59cf3dc2f29cffc35dcc4cc9d7728 diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c index 3583a2b..001ef1d 100644 --- a/lustre/mdt/mdt_handler.c +++ b/lustre/mdt/mdt_handler.c @@ -30,6 +30,9 @@ * Use is subject to license terms. */ /* + * Copyright (c) 2011 Whamcloud, Inc. + */ +/* * This file is part of Lustre, http://www.lustre.org/ * Lustre is a trademark of Sun Microsystems, Inc. * @@ -341,6 +344,7 @@ static int mdt_getstatus(struct mdt_thread_info *info) static int mdt_statfs(struct mdt_thread_info *info) { + struct ptlrpc_request *req = mdt_info_req(info); struct md_device *next = info->mti_mdt->mdt_child; struct ptlrpc_service *svc; struct obd_statfs *osfs; @@ -367,6 +371,10 @@ static int mdt_statfs(struct mdt_thread_info *info) &info->mti_u.ksfs); statfs_pack(osfs, &info->mti_u.ksfs); } + + if (rc == 0) + mdt_counter_incr(req->rq_export, LPROC_MDT_STATFS); + RETURN(rc); } @@ -448,6 +456,10 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *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); } static inline int mdt_body_has_lov(const struct lu_attr *la, @@ -663,6 +675,7 @@ static int mdt_renew_capa(struct mdt_thread_info *info) static int mdt_getattr(struct mdt_thread_info *info) { + struct ptlrpc_request *req = mdt_info_req(info); struct mdt_object *obj = info->mti_object; struct req_capsule *pill = info->mti_pill; struct mdt_body *reqbody; @@ -724,6 +737,9 @@ static int mdt_getattr(struct mdt_thread_info *info) mdt_exit_ucred(info); EXIT; out_shrink: + if (rc == 0) + mdt_counter_incr(req->rq_export, LPROC_MDT_GETATTR); + mdt_shrink_reply(info); return rc; } @@ -814,7 +830,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, struct lu_name *lname = NULL; const char *name = NULL; int namelen = 0; - struct mdt_lock_handle *lhp; + struct mdt_lock_handle *lhp = NULL; struct ldlm_lock *lock; struct ldlm_res_id *res_id; int is_resent; @@ -844,9 +860,13 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, if (namelen == 0) { reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY); - LASSERT(fid_is_sane(&reqbody->fid2)); - name = NULL; + 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), @@ -920,16 +940,15 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, RETURN(rc); } - /* step 1: lock parent */ - lhp = &info->mti_lh[MDT_LH_PARENT]; - mdt_lock_pdo_init(lhp, LCK_PR, name, namelen); - rc = mdt_object_lock(info, parent, lhp, MDS_INODELOCK_UPDATE, - MDT_LOCAL_LOCK); - - if (unlikely(rc != 0)) - RETURN(rc); - if (lname) { + /* step 1: lock parent */ + lhp = &info->mti_lh[MDT_LH_PARENT]; + mdt_lock_pdo_init(lhp, LCK_PR, name, namelen); + rc = mdt_object_lock(info, parent, lhp, MDS_INODELOCK_UPDATE, + MDT_LOCAL_LOCK); + if (unlikely(rc != 0)) + RETURN(rc); + /* step 2: lookup child's fid by name */ rc = mdo_lookup(info->mti_env, next, lname, child_fid, &info->mti_spec); @@ -980,10 +999,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, LDLM_LOCK_PUT(lock); rc = 0; } else { - struct md_attr *ma; relock: - ma = &info->mti_attr; - OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RESEND, obd_timeout*2); mdt_lock_handle_init(lhc); mdt_lock_reg_init(lhc, LCK_PR); @@ -992,24 +1008,30 @@ relock: LU_OBJECT_DEBUG(D_WARNING, info->mti_env, &child->mot_obj.mo_lu, "Object doesn't exist!\n"); - GOTO(out_child, rc = -ESTALE); + GOTO(out_child, rc = -ENOENT); } - ma->ma_valid = 0; - ma->ma_need = MA_INODE; - rc = mo_attr_get(info->mti_env, next, ma); - if (unlikely(rc != 0)) - GOTO(out_child, rc); + if (!(child_bits & MDS_INODELOCK_UPDATE)) { + struct md_attr *ma = &info->mti_attr; - /* 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 (ma->ma_valid & MA_INODE && - ma->ma_attr.la_valid & LA_CTIME && - info->mti_mdt->mdt_namespace->ns_ctime_age_limit + - ma->ma_attr.la_ctime < cfs_time_current_sec()) - child_bits |= MDS_INODELOCK_UPDATE; + ma->ma_valid = 0; + ma->ma_need = MA_INODE; + rc = mo_attr_get(info->mti_env, + mdt_object_child(child), ma); + 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 (ma->ma_valid & MA_INODE && + ma->ma_attr.la_valid & LA_CTIME && + info->mti_mdt->mdt_namespace->ns_ctime_age_limit + + ma->ma_attr.la_ctime < cfs_time_current_sec()) + child_bits |= MDS_INODELOCK_UPDATE; + } rc = mdt_object_lock(info, child, lhc, child_bits, MDT_CROSS_LOCK); @@ -1050,7 +1072,8 @@ relock: out_child: mdt_object_put(info->mti_env, child); out_parent: - mdt_object_unlock(info, parent, lhp, 1); + if (lhp) + mdt_object_unlock(info, parent, lhp, 1); return rc; } @@ -1196,7 +1219,7 @@ static int mdt_disconnect(struct mdt_thread_info *info) } static int mdt_sendpage(struct mdt_thread_info *info, - struct lu_rdpg *rdpg) + struct lu_rdpg *rdpg, int nob) { struct ptlrpc_request *req = mdt_info_req(info); struct obd_export *exp = req->rq_export; @@ -1204,7 +1227,6 @@ static int mdt_sendpage(struct mdt_thread_info *info, struct l_wait_info *lwi = &info->mti_u.rdpg.mti_wait_info; int tmpcount; int tmpsize; - int timeout; int i; int rc; ENTRY; @@ -1214,58 +1236,16 @@ static int mdt_sendpage(struct mdt_thread_info *info, if (desc == NULL) RETURN(-ENOMEM); - for (i = 0, tmpcount = rdpg->rp_count; - i < rdpg->rp_npages; i++, tmpcount -= tmpsize) { + for (i = 0, tmpcount = nob; + i < rdpg->rp_npages && tmpcount > 0; i++, tmpcount -= tmpsize) { tmpsize = min_t(int, tmpcount, CFS_PAGE_SIZE); ptlrpc_prep_bulk_page(desc, rdpg->rp_pages[i], 0, tmpsize); } - LASSERT(desc->bd_nob == rdpg->rp_count); - rc = sptlrpc_svc_wrap_bulk(req, desc); - if (rc) - GOTO(free_desc, rc); - - rc = ptlrpc_start_bulk_transfer(desc); - if (rc) - GOTO(free_desc, rc); - - if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE)) - GOTO(abort_bulk, rc = 0); - - timeout = (int) req->rq_deadline - cfs_time_current_sec(); - if (timeout < 0) - CERROR("Req deadline already passed %lu (now: %lu)\n", - req->rq_deadline, cfs_time_current_sec()); - *lwi = LWI_TIMEOUT_INTERVAL(cfs_time_seconds(max(timeout, 1)), - cfs_time_seconds(1), NULL, NULL); - rc = l_wait_event(desc->bd_waitq, !ptlrpc_server_bulk_active(desc) || - exp->exp_failed || exp->exp_abort_active_req, lwi); - LASSERT (rc == 0 || rc == -ETIMEDOUT); - - if (rc == 0) { - if (desc->bd_success && - desc->bd_nob_transferred == rdpg->rp_count) - GOTO(free_desc, rc); - - rc = -ETIMEDOUT; - if (exp->exp_abort_active_req || exp->exp_failed) - GOTO(abort_bulk, rc); - } - - DEBUG_REQ(D_ERROR, req, "bulk failed: %s %d(%d), evicting %s@%s", - (rc == -ETIMEDOUT) ? "timeout" : "network error", - desc->bd_nob_transferred, rdpg->rp_count, - exp->exp_client_uuid.uuid, - exp->exp_connection->c_remote_uuid.uuid); - - class_fail_export(exp); - - EXIT; -abort_bulk: - ptlrpc_abort_bulk(desc); -free_desc: + LASSERT(desc->bd_nob == nob); + rc = target_bulk_io(exp, desc, lwi); ptlrpc_free_bulk(desc); - return rc; + RETURN(rc); } #ifdef HAVE_SPLIT_SUPPORT @@ -1473,8 +1453,12 @@ static int mdt_readpage(struct mdt_thread_info *info) } rdpg->rp_attrs = reqbody->mode; - rdpg->rp_count = reqbody->nlink; - rdpg->rp_npages = (rdpg->rp_count + CFS_PAGE_SIZE - 1)>>CFS_PAGE_SHIFT; + if (info->mti_exp->exp_connect_flags & OBD_CONNECT_64BITHASH) + rdpg->rp_attrs |= LUDA_64BITHASH; + rdpg->rp_count = min_t(unsigned int, reqbody->nlink, + PTLRPC_MAX_BRW_SIZE); + rdpg->rp_npages = (rdpg->rp_count + CFS_PAGE_SIZE - 1) >> + CFS_PAGE_SHIFT; OBD_ALLOC(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]); if (rdpg->rp_pages == NULL) RETURN(-ENOMEM); @@ -1487,11 +1471,11 @@ static 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) + if (rc < 0) GOTO(free_rdpg, rc); /* send pages to client */ - rc = mdt_sendpage(info, rdpg); + rc = mdt_sendpage(info, rdpg, rc); EXIT; free_rdpg: @@ -1666,6 +1650,7 @@ static int mdt_object_sync(struct mdt_thread_info *info) static 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; @@ -1713,6 +1698,9 @@ static int mdt_sync(struct mdt_thread_info *info) } else rc = err_serious(rc); } + if (rc == 0) + mdt_counter_incr(req->rq_export, LPROC_MDT_SYNC); + RETURN(rc); } @@ -2036,7 +2024,7 @@ static int mdt_sec_ctx_handle(struct mdt_thread_info *info) sptlrpc_svc_ctx_invalidate(req); } - OBD_FAIL_TIMEOUT(OBD_FAIL_SEC_CTX_HDL_PAUSE, obd_fail_val); + CFS_FAIL_TIMEOUT(OBD_FAIL_SEC_CTX_HDL_PAUSE, cfs_fail_val); return rc; } @@ -2129,7 +2117,7 @@ static inline int mdt_is_lock_sync(struct ldlm_lock *lock) int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, void *data, int flag) { - struct obd_device *obd = lock->l_resource->lr_namespace->ns_obd; + struct obd_device *obd = ldlm_lock_to_ns(lock)->ns_obd; struct mdt_device *mdt = mdt_dev(obd->obd_lu_dev); int rc; ENTRY; @@ -2200,7 +2188,8 @@ int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o, RETURN(-ESTALE); } else { /* Non-dir object shouldn't have PDO lock */ - LASSERT(S_ISDIR(lu_object_attr(&o->mot_obj.mo_lu))); + if (!S_ISDIR(lu_object_attr(&o->mot_obj.mo_lu))) + RETURN(-ENOTDIR); } } @@ -2250,7 +2239,7 @@ int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o, 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, 10); + OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PDO_LOCK, 15); } RETURN(rc); @@ -3391,6 +3380,10 @@ static int mdt_intent_reint(enum mdt_it_code opcode, } rep->lock_policy_res2 = clear_serious(rc); + if (rep->lock_policy_res2 == -ENOENT && + mdt_get_disposition(rep, DISP_LOOKUP_NEG)) + rep->lock_policy_res2 = 0; + if (rc == -ENOTCONN || rc == -ENODEV || rc == -EOVERFLOW) { /**< if VBR failure then return error */ /* @@ -3411,6 +3404,8 @@ static int mdt_intent_reint(enum mdt_it_code opcode, * checked here. */ if (lustre_handle_is_used(&lhc->mlh_reg_lh)) { + LASSERTF(rc == 0, "Error occurred but lock handle " + "is still in use\n"); rep->lock_policy_res2 = 0; rc = mdt_intent_lock_replace(info, lockp, NULL, lhc, flags); RETURN(rc); @@ -3883,7 +3878,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) if (m->mdt_regular_service == NULL) RETURN(-ENOMEM); - rc = ptlrpc_start_threads(NULL, m->mdt_regular_service); + rc = ptlrpc_start_threads(m->mdt_regular_service); if (rc) GOTO(err_mdt_svc, rc); @@ -3913,7 +3908,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_readpage_service); + rc = ptlrpc_start_threads(m->mdt_readpage_service); /* * setattr service configuration. @@ -3941,7 +3936,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_setattr_service); + rc = ptlrpc_start_threads(m->mdt_setattr_service); if (rc) GOTO(err_mdt_svc, rc); @@ -3970,7 +3965,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_mdsc_service); + rc = ptlrpc_start_threads(m->mdt_mdsc_service); if (rc) GOTO(err_mdt_svc, rc); @@ -3999,7 +3994,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_mdss_service); + rc = ptlrpc_start_threads(m->mdt_mdss_service); if (rc) GOTO(err_mdt_svc, rc); @@ -4031,7 +4026,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_dtss_service); + rc = ptlrpc_start_threads(m->mdt_dtss_service); if (rc) GOTO(err_mdt_svc, rc); @@ -4058,7 +4053,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_fld_service); + rc = ptlrpc_start_threads(m->mdt_fld_service); if (rc) GOTO(err_mdt_svc, rc); @@ -4088,7 +4083,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_xmds_service); + rc = ptlrpc_start_threads(m->mdt_xmds_service); if (rc) GOTO(err_mdt_svc, rc); @@ -4400,49 +4395,6 @@ static int mdt_adapt_sptlrpc_conf(struct obd_device *obd, int initial) return 0; } -static void fsoptions_to_mdt_flags(struct mdt_device *m, char *options) -{ - char *p = options; - - m->mdt_opts.mo_mds_capa = 1; - m->mdt_opts.mo_oss_capa = 1; -#ifdef CONFIG_FS_POSIX_ACL - /* ACLs should be enabled by default (b=13829) */ - m->mdt_opts.mo_acl = 1; - LCONSOLE_INFO("Enabling ACL\n"); -#else - m->mdt_opts.mo_acl = 0; - LCONSOLE_INFO("Disabling ACL\n"); -#endif - - if (!options) - return; - - while (*options) { - int len; - - while (*p && *p != ',') - p++; - - len = p - options; - if ((len == sizeof("user_xattr") - 1) && - (memcmp(options, "user_xattr", len) == 0)) { - m->mdt_opts.mo_user_xattr = 1; - LCONSOLE_INFO("Enabling user_xattr\n"); - } else if ((len == sizeof("nouser_xattr") - 1) && - (memcmp(options, "nouser_xattr", len) == 0)) { - m->mdt_opts.mo_user_xattr = 0; - LCONSOLE_INFO("Disabling user_xattr\n"); - } else if ((len == sizeof("noacl") - 1) && - (memcmp(options, "noacl", len) == 0)) { - m->mdt_opts.mo_acl = 0; - LCONSOLE_INFO("Disabling ACL\n"); - } - - options = ++p; - } -} - int mdt_postrecov(const struct lu_env *, struct mdt_device *); static int mdt_init0(const struct lu_env *env, struct mdt_device *m, @@ -4458,11 +4410,10 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, struct lu_site *s; struct md_site *mite; const char *identity_upcall = "NONE"; -#ifdef HAVE_QUOTA_SUPPORT struct md_device *next; -#endif int rc; int node_id; + mntopt_t mntopts; ENTRY; md_device_init(&m->mdt_md_dev, ldt); @@ -4488,8 +4439,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, m->mdt_max_cookiesize = sizeof(struct llog_cookie); m->mdt_som_conf = 0; - m->mdt_opts.mo_user_xattr = 0; - m->mdt_opts.mo_acl = 0; m->mdt_opts.mo_cos = MDT_COS_DEFAULT; lmi = server_get_mount_2(dev); if (lmi == NULL) { @@ -4497,7 +4446,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, RETURN(-EFAULT); } else { lsi = s2lsi(lmi->lmi_sb); - fsoptions_to_mdt_flags(m, lsi->lsi_lmd->lmd_opts); /* CMD is supported only in IAM mode */ ldd = lsi->lsi_ldd; LASSERT(num); @@ -4506,14 +4454,8 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, CERROR("CMD Operation not allowed in IOP mode\n"); GOTO(err_lmi, rc = -EINVAL); } - /* Read recovery timeouts */ - if (lsi->lsi_lmd && lsi->lsi_lmd->lmd_recovery_time_soft) - obd->obd_recovery_timeout = - lsi->lsi_lmd->lmd_recovery_time_soft; - if (lsi->lsi_lmd && lsi->lsi_lmd->lmd_recovery_time_hard) - obd->obd_recovery_time_hard = - lsi->lsi_lmd->lmd_recovery_time_hard; + obd->u.obt.obt_magic = OBT_MAGIC; } cfs_rwlock_init(&m->mdt_sptlrpc_lock); @@ -4521,6 +4463,8 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, cfs_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; @@ -4596,7 +4540,8 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, LUSTRE_MDT_NAME"-%p", m); m->mdt_namespace = ldlm_namespace_new(obd, info->mti_u.ns_name, LDLM_NAMESPACE_SERVER, - LDLM_NAMESPACE_GREEDY); + LDLM_NAMESPACE_GREEDY, + LDLM_NS_TYPE_MDT); if (m->mdt_namespace == NULL) GOTO(err_fini_seq, rc = -ENOMEM); @@ -4604,19 +4549,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, /* set obd_namespace for compatibility with old code */ obd->obd_namespace = m->mdt_namespace; - /* XXX: to support suppgid for ACL, we enable identity_upcall - * by default, otherwise, maybe got unexpected -EACCESS. */ - if (m->mdt_opts.mo_acl) - identity_upcall = MDT_IDENTITY_UPCALL_PATH; - - m->mdt_identity_cache = upcall_cache_init(obd->obd_name, identity_upcall, - &mdt_identity_upcall_cache_ops); - if (IS_ERR(m->mdt_identity_cache)) { - rc = PTR_ERR(m->mdt_identity_cache); - m->mdt_identity_cache = NULL; - GOTO(err_free_ns, rc); - } - cfs_timer_init(&m->mdt_ck_timer, mdt_ck_timer_callback, m); rc = mdt_ck_thread_start(m); @@ -4637,8 +4569,8 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, mdt_adapt_sptlrpc_conf(obd, 1); -#ifdef HAVE_QUOTA_SUPPORT next = m->mdt_child; +#ifdef HAVE_QUOTA_SUPPORT rc = next->md_ops->mdo_quota.mqo_setup(env, next, lmi->lmi_mnt); if (rc) GOTO(err_llog_cleanup, rc); @@ -4647,6 +4579,34 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, server_put_mount_2(dev, lmi->lmi_mnt); lmi = NULL; + rc = next->md_ops->mdo_iocontrol(env, next, OBD_IOC_GET_MNTOPT, 0, + &mntopts); + if (rc) + GOTO(err_quota, rc); + + if (mntopts & MNTOPT_USERXATTR) + m->mdt_opts.mo_user_xattr = 1; + else + m->mdt_opts.mo_user_xattr = 0; + + if (mntopts & MNTOPT_ACL) + m->mdt_opts.mo_acl = 1; + else + m->mdt_opts.mo_acl = 0; + + /* XXX: to support suppgid for ACL, we enable identity_upcall + * by default, otherwise, maybe got unexpected -EACCESS. */ + if (m->mdt_opts.mo_acl) + identity_upcall = MDT_IDENTITY_UPCALL_PATH; + + m->mdt_identity_cache = upcall_cache_init(obd->obd_name,identity_upcall, + &mdt_identity_upcall_cache_ops); + if (IS_ERR(m->mdt_identity_cache)) { + rc = PTR_ERR(m->mdt_identity_cache); + m->mdt_identity_cache = NULL; + GOTO(err_quota, rc); + } + target_recovery_init(&m->mdt_lut, mdt_recovery_handle); rc = mdt_start_ptlrpc_service(m); @@ -4677,6 +4637,9 @@ err_stop_service: mdt_stop_ptlrpc_service(m); err_recovery: target_recovery_fini(obd); + upcall_cache_cleanup(m->mdt_identity_cache); + m->mdt_identity_cache = NULL; +err_quota: #ifdef HAVE_QUOTA_SUPPORT next->md_ops->mdo_quota.mqo_cleanup(env, next); #endif @@ -4689,8 +4652,6 @@ err_capa: cfs_timer_disarm(&m->mdt_ck_timer); mdt_ck_thread_stop(m); err_free_ns: - upcall_cache_cleanup(m->mdt_identity_cache); - m->mdt_identity_cache = NULL; ldlm_namespace_free(m->mdt_namespace, NULL, 0); obd->obd_namespace = m->mdt_namespace = NULL; err_fini_seq: @@ -4729,15 +4690,21 @@ static int mdt_process_config(const struct lu_env *env, /* * For interoperability between 1.8 and 2.0, - * skip old "mdt.group_upcall" param. */ { + /* Skip old "mdt.group_upcall" param. */ char *param = lustre_cfg_string(cfg, 1); if (param && !strncmp("mdt.group_upcall", param, 16)) { CWARN("For 1.8 interoperability, skip this" " mdt.group_upcall. It is obsolete\n"); break; } + /* Rename old "mdt.quota_type" to "mdd.quota_type. */ + if (param && !strncmp("mdt.quota_type", param, 14)) { + CWARN("Found old param mdt.quota_type, changed" + " it to mdd.quota_type.\n"); + param[2] = 'd'; + } } lprocfs_mdt_init_vars(&lvars); @@ -4787,6 +4754,7 @@ static struct lu_object *mdt_object_alloc(const struct lu_env *env, lu_object_add_top(h, o); o->lo_ops = &mdt_obj_ops; cfs_sema_init(&mo->mot_ioepoch_sem, 1); + cfs_sema_init(&mo->mot_lov_sem, 1); RETURN(o); } else RETURN(NULL); @@ -4895,6 +4863,24 @@ static int mdt_connect_internal(struct obd_export *exp, if (!mdt->mdt_som_conf) data->ocd_connect_flags &= ~OBD_CONNECT_SOM; + if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) { + data->ocd_brw_size = min(data->ocd_brw_size, + (__u32)(PTLRPC_MAX_BRW_PAGES << CFS_PAGE_SHIFT)); + if (data->ocd_brw_size == 0) { + CERROR("%s: cli %s/%p ocd_connect_flags: "LPX64 + " ocd_version: %x ocd_grant: %d " + "ocd_index: %u ocd_brw_size is " + "unexpectedly zero, network data " + "corruption? Refusing connection of this" + " client\n", + exp->exp_obd->obd_name, + exp->exp_client_uuid.uuid, + exp, data->ocd_connect_flags, data->ocd_version, + data->ocd_grant, data->ocd_index); + return -EPROTO; + } + } + cfs_spin_lock(&exp->exp_lock); exp->exp_connect_flags = data->ocd_connect_flags; cfs_spin_unlock(&exp->exp_lock); @@ -5005,6 +4991,9 @@ static int mdt_obd_connect(const struct lu_env *env, 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 mdt_thread_info *mti; @@ -5102,12 +5091,12 @@ static int mdt_export_cleanup(struct obd_export *exp) int cookie_size; lmm_size = mdt->mdt_max_mdsize; - OBD_ALLOC(ma->ma_lmm, lmm_size); + OBD_ALLOC_LARGE(ma->ma_lmm, lmm_size); if (ma->ma_lmm == NULL) GOTO(out_lmm, rc = -ENOMEM); cookie_size = mdt->mdt_max_cookiesize; - OBD_ALLOC(ma->ma_cookie, cookie_size); + OBD_ALLOC_LARGE(ma->ma_cookie, cookie_size); if (ma->ma_cookie == NULL) GOTO(out_cookie, rc = -ENOMEM); @@ -5120,21 +5109,24 @@ static int mdt_export_cleanup(struct obd_export *exp) ma->ma_need = 0; /* It is not for setattr, just tell MDD to send * DESTROY RPC to OSS if needed */ - ma->ma_attr_flags = MDS_CLOSE_CLEANUP; ma->ma_valid = MA_FLAGS; + ma->ma_attr_flags = MDS_CLOSE_CLEANUP; + /* Don't unlink orphan on failover umount, LU-184 */ + if (exp->exp_flags & OBD_OPT_FAILOVER) + ma->ma_attr_flags |= MDS_KEEP_ORPHAN; mdt_mfd_close(info, mfd); } - OBD_FREE(ma->ma_cookie, cookie_size); + OBD_FREE_LARGE(ma->ma_cookie, cookie_size); ma->ma_cookie = NULL; out_cookie: - OBD_FREE(ma->ma_lmm, lmm_size); + OBD_FREE_LARGE(ma->ma_lmm, lmm_size); ma->ma_lmm = NULL; } out_lmm: info->mti_mdt = NULL; /* cleanup client slot early */ /* Do not erase record for recoverable client. */ - if (!obd->obd_fail || exp->exp_failed) + if (!(exp->exp_flags & OBD_OPT_FAILOVER) || exp->exp_failed) mdt_client_del(&env, mdt); lu_env_fini(&env); @@ -5172,35 +5164,42 @@ static int mdt_init_export(struct obd_export *exp) cfs_spin_lock(&exp->exp_lock); exp->exp_connecting = 1; cfs_spin_unlock(&exp->exp_lock); + + /* self-export doesn't need client data and ldlm initialization */ + if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid, + &exp->exp_client_uuid))) + RETURN(0); + rc = lut_client_alloc(exp); if (rc == 0) rc = ldlm_init_export(exp); - if (rc) - CERROR("Error %d while initializing export\n", rc); + CERROR("%s: Error %d while initializing export\n", + exp->exp_obd->obd_name, rc); RETURN(rc); } static int mdt_destroy_export(struct obd_export *exp) { - struct mdt_export_data *med; - int rc = 0; ENTRY; - med = &exp->exp_mdt_data; if (exp_connect_rmtclient(exp)) mdt_cleanup_idmap(&exp->exp_mdt_data); target_destroy_export(exp); + /* destroy can be called from failed obd_setup, so + * checking uuid is safer than obd_self_export */ + if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid, + &exp->exp_client_uuid))) + RETURN(0); + ldlm_destroy_export(exp); lut_client_free(exp); LASSERT(cfs_list_empty(&exp->exp_outstanding_replies)); LASSERT(cfs_list_empty(&exp->exp_mdt_data.med_open_head)); - if (obd_uuid_equals(&exp->exp_client_uuid, &exp->exp_obd->obd_uuid)) - RETURN(0); - RETURN(rc); + RETURN(0); } static void mdt_allow_cli(struct mdt_device *m, unsigned int flag) @@ -5215,12 +5214,12 @@ static void mdt_allow_cli(struct mdt_device *m, unsigned int flag) if (cfs_test_bit(MDT_FL_CFGLOG, &m->mdt_state) && cfs_test_bit(MDT_FL_SYNCED, &m->mdt_state)) { struct obd_device *obd = m->mdt_md_dev.md_lu_dev.ld_obd; - + /* Open for clients */ if (obd->obd_no_conn) { - cfs_spin_lock_bh(&obd->obd_processing_task_lock); + cfs_spin_lock(&obd->obd_dev_lock); obd->obd_no_conn = 0; - cfs_spin_unlock_bh(&obd->obd_processing_task_lock); + cfs_spin_unlock(&obd->obd_dev_lock); } } } @@ -5243,7 +5242,8 @@ static int mdt_upcall(const struct lu_env *env, struct md_device *md, m->mdt_max_mdsize, m->mdt_max_cookiesize); mdt_allow_cli(m, CONFIG_SYNC); if (data) - (*(__u64 *)data) = m->mdt_lut.lut_mount_count; + (*(__u64 *)data) = + m->mdt_lut.lut_obd->u.obt.obt_mount_count; break; case MD_NO_TRANS: mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key); @@ -5269,11 +5269,11 @@ static int mdt_upcall(const struct lu_env *env, struct md_device *md, RETURN(rc); } -static int mdt_obd_notify(struct obd_device *host, +static int mdt_obd_notify(struct obd_device *obd, struct obd_device *watched, enum obd_notify_event ev, void *data) { - struct mdt_device *mdt = mdt_dev(host->obd_lu_dev); + struct mdt_device *mdt = mdt_dev(obd->obd_lu_dev); #ifdef HAVE_QUOTA_SUPPORT struct md_device *next = mdt->mdt_child; #endif @@ -5431,6 +5431,7 @@ static int mdt_ioc_version_get(struct mdt_thread_info *mti, void *karg) struct mdt_lock_handle *lh; int rc; ENTRY; + CDEBUG(D_IOCTL, "getting version for "DFID"\n", PFID(fid)); if (!fid_is_sane(fid)) RETURN(-EINVAL); @@ -5450,6 +5451,9 @@ static int mdt_ioc_version_get(struct mdt_thread_info *mti, void *karg) * fid, this is error to find remote object here */ CERROR("nonlocal object "DFID"\n", PFID(fid)); + } else if (rc == 0) { + *(__u64 *)data->ioc_inlbuf2 = ENOENT_VERSION; + rc = -ENOENT; } else { version = mo_version_get(mti->mti_env, mdt_object_child(obj)); *(__u64 *)data->ioc_inlbuf2 = version; @@ -5516,8 +5520,8 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len, int mdt_postrecov(const struct lu_env *env, struct mdt_device *mdt) { struct lu_device *ld = md2lu_dev(mdt->mdt_child); - struct obd_device *obd = mdt2obd_dev(mdt); #ifdef HAVE_QUOTA_SUPPORT + struct obd_device *obd = mdt2obd_dev(mdt); struct md_device *next = mdt->mdt_child; #endif int rc; @@ -5665,15 +5669,6 @@ LU_KEY_INIT_FINI(mdt, struct mdt_thread_info); /* context key: mdt_thread_key */ LU_CONTEXT_KEY_DEFINE(mdt, LCT_MD_THREAD); -/* context key constructor/destructor: mdt_txn_key_init, mdt_txn_key_fini */ -LU_KEY_INIT_FINI(mdt_txn, struct mdt_txn_info); - -struct lu_context_key mdt_txn_key = { - .lct_tags = LCT_TX_HANDLE, - .lct_init = mdt_txn_key_init, - .lct_fini = mdt_txn_key_fini -}; - struct md_ucred *mdt_ucred(const struct mdt_thread_info *info) { return md_ucred(info->mti_env); @@ -5716,7 +5711,7 @@ int mdt_cos_is_enabled(struct mdt_device *mdt) } /* type constructor/destructor: mdt_type_init, mdt_type_fini */ -LU_TYPE_INIT_FINI(mdt, &mdt_thread_key, &mdt_txn_key); +LU_TYPE_INIT_FINI(mdt, &mdt_thread_key); static struct lu_device_type_operations mdt_device_type_ops = { .ldto_init = mdt_type_init, @@ -5759,8 +5754,6 @@ static int __init mdt_mod_init(void) } else { mdt_max_threads = MDT_MAX_THREADS; mdt_min_threads = MDT_MIN_THREADS; - if (mdt_min_threads < MDT_NUM_THREADS) - mdt_min_threads = MDT_NUM_THREADS; } lprocfs_mdt_init_vars(&lvars);