X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_handler.c;h=be9797b99f111552c19c341374fcb51b444ea6fa;hp=0864c6d5fe0a1afa595ae72adad8bd9ba5fa49a6;hb=4115148e4ac5ccb3620a9f8d3e48da0a712943c5;hpb=7d33c7714a432b451d790031e31e1017ab405311 diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c index 0864c6d..be9797b 100644 --- a/lustre/mdt/mdt_handler.c +++ b/lustre/mdt/mdt_handler.c @@ -23,7 +23,7 @@ * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2010, 2016, Intel Corporation. + * Copyright (c) 2010, 2017, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -58,6 +58,7 @@ #include #include #include +#include #include #include #include @@ -66,7 +67,6 @@ #include "mdt_internal.h" - static unsigned int max_mod_rpcs_per_client = 8; module_param(max_mod_rpcs_per_client, uint, 0644); MODULE_PARM_DESC(max_mod_rpcs_per_client, "maximum number of modify RPCs in flight allowed per client"); @@ -94,7 +94,6 @@ enum ldlm_mode mdt_dlm_lock_modes[] = { }; static struct mdt_device *mdt_dev(struct lu_device *d); -static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags); static const struct lu_object_operations mdt_obj_ops; @@ -270,18 +269,13 @@ static int mdt_lookup_fileset(struct mdt_thread_info *info, const char *fileset, { struct mdt_device *mdt = info->mti_mdt; struct lu_name *lname = &info->mti_name; - char *name = NULL; + char *filename = info->mti_filename; struct mdt_object *parent; u32 mode; int rc = 0; LASSERT(!info->mti_cross_ref); - OBD_ALLOC(name, NAME_MAX + 1); - if (name == NULL) - return -ENOMEM; - lname->ln_name = name; - /* * We may want to allow this to mount a completely separate * fileset from the MDT in the future, but keeping it to @@ -317,8 +311,9 @@ static int mdt_lookup_fileset(struct mdt_thread_info *info, const char *fileset, break; } - strncpy(name, s1, lname->ln_namelen); - name[lname->ln_namelen] = '\0'; + strncpy(filename, s1, lname->ln_namelen); + filename[lname->ln_namelen] = '\0'; + lname->ln_name = filename; parent = mdt_object_find(info->mti_env, mdt, fid); if (IS_ERR(parent)) { @@ -343,8 +338,6 @@ static int mdt_lookup_fileset(struct mdt_thread_info *info, const char *fileset, } } - OBD_FREE(name, NAME_MAX + 1); - return rc; } @@ -413,13 +406,16 @@ out: static int mdt_statfs(struct tgt_session_info *tsi) { - struct ptlrpc_request *req = tgt_ses_req(tsi); - struct mdt_thread_info *info = tsi2mdt_info(tsi); - struct mdt_device *mdt = info->mti_mdt; - struct tg_grants_data *tgd = &mdt->mdt_lut.lut_tgd; - struct ptlrpc_service_part *svcpt; - struct obd_statfs *osfs; - int rc; + struct ptlrpc_request *req = tgt_ses_req(tsi); + struct mdt_thread_info *info = tsi2mdt_info(tsi); + struct mdt_device *mdt = info->mti_mdt; + struct tg_grants_data *tgd = &mdt->mdt_lut.lut_tgd; + struct md_device *next = mdt->mdt_child; + struct ptlrpc_service_part *svcpt; + struct obd_statfs *osfs; + struct mdt_body *reqbody = NULL; + struct mdt_statfs_cache *msf; + int rc; ENTRY; @@ -441,11 +437,39 @@ static int mdt_statfs(struct tgt_session_info *tsi) if (!osfs) GOTO(out, rc = -EPROTO); - rc = tgt_statfs_internal(tsi->tsi_env, &mdt->mdt_lut, osfs, - cfs_time_shift_64(-OBD_STATFS_CACHE_SECONDS), - NULL); - if (unlikely(rc)) - GOTO(out, rc); + if (mdt_is_sum_statfs_client(req->rq_export) && + lustre_packed_msg_size(req->rq_reqmsg) == + req_capsule_fmt_size(req->rq_reqmsg->lm_magic, + &RQF_MDS_STATFS_NEW, RCL_CLIENT)) { + req_capsule_extend(info->mti_pill, &RQF_MDS_STATFS_NEW); + reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY); + } + + if (reqbody && reqbody->mbo_valid & OBD_MD_FLAGSTATFS) + msf = &mdt->mdt_sum_osfs; + else + msf = &mdt->mdt_osfs; + + if (msf->msf_age + OBD_STATFS_CACHE_SECONDS <= ktime_get_seconds()) { + /** statfs data is too old, get up-to-date one */ + if (reqbody && reqbody->mbo_valid & OBD_MD_FLAGSTATFS) + rc = next->md_ops->mdo_statfs(info->mti_env, + next, osfs); + else + rc = dt_statfs(info->mti_env, mdt->mdt_bottom, + osfs); + if (rc) + GOTO(out, rc); + spin_lock(&mdt->mdt_lock); + msf->msf_osfs = *osfs; + msf->msf_age = ktime_get_seconds(); + spin_unlock(&mdt->mdt_lock); + } else { + /** use cached statfs data */ + spin_lock(&mdt->mdt_lock); + *osfs = msf->msf_osfs; + spin_unlock(&mdt->mdt_lock); + } /* at least try to account for cached pages. its still racy and * might be under-reporting if clients haven't announced their @@ -486,15 +510,68 @@ out: RETURN(rc); } +__u32 mdt_lmm_dom_entry_check(struct lov_mds_md *lmm, int *is_dom_only) +{ + struct lov_comp_md_v1 *comp_v1; + struct lov_mds_md *v1; + __u32 off; + __u32 dom_stripesize = 0; + int i; + bool has_ost_stripes = false; + + ENTRY; + + if (is_dom_only) + *is_dom_only = 0; + + if (le32_to_cpu(lmm->lmm_magic) != LOV_MAGIC_COMP_V1) + RETURN(0); + + comp_v1 = (struct lov_comp_md_v1 *)lmm; + off = le32_to_cpu(comp_v1->lcm_entries[0].lcme_offset); + v1 = (struct lov_mds_md *)((char *)comp_v1 + off); + + /* Fast check for DoM entry with no mirroring, should be the first */ + if (le16_to_cpu(comp_v1->lcm_mirror_count) == 0 && + lov_pattern(le32_to_cpu(v1->lmm_pattern)) != LOV_PATTERN_MDT) + RETURN(0); + + /* check all entries otherwise */ + for (i = 0; i < le16_to_cpu(comp_v1->lcm_entry_count); i++) { + struct lov_comp_md_entry_v1 *lcme; + + lcme = &comp_v1->lcm_entries[i]; + if (!(le32_to_cpu(lcme->lcme_flags) & LCME_FL_INIT)) + continue; + + off = le32_to_cpu(lcme->lcme_offset); + v1 = (struct lov_mds_md *)((char *)comp_v1 + off); + + if (lov_pattern(le32_to_cpu(v1->lmm_pattern)) == + LOV_PATTERN_MDT) + dom_stripesize = le32_to_cpu(v1->lmm_stripe_size); + else + has_ost_stripes = true; + + if (dom_stripesize && has_ost_stripes) + RETURN(dom_stripesize); + } + /* DoM-only case exits here */ + if (is_dom_only && dom_stripesize) + *is_dom_only = 1; + RETURN(dom_stripesize); +} + /** * Pack size attributes into the reply. */ int mdt_pack_size2body(struct mdt_thread_info *info, - const struct lu_fid *fid, bool dom_lock) + const struct lu_fid *fid, struct lustre_handle *lh) { struct mdt_body *b; struct md_attr *ma = &info->mti_attr; - int dom_stripe; + __u32 dom_stripe; + bool dom_lock = false; ENTRY; @@ -504,11 +581,21 @@ int mdt_pack_size2body(struct mdt_thread_info *info, !(ma->ma_valid & MA_LOV && ma->ma_lmm != NULL)) RETURN(-ENODATA); - dom_stripe = mdt_lmm_dom_entry(ma->ma_lmm); + dom_stripe = mdt_lmm_dom_stripesize(ma->ma_lmm); /* no DoM stripe, no size in reply */ - if (dom_stripe == LMM_NO_DOM) + if (!dom_stripe) RETURN(-ENOENT); + if (lustre_handle_is_used(lh)) { + struct ldlm_lock *lock; + + lock = ldlm_handle2lock(lh); + if (lock != NULL) { + dom_lock = ldlm_has_dom(lock); + LDLM_LOCK_PUT(lock); + } + } + /* no DoM lock, no size in reply */ if (!dom_lock) RETURN(0); @@ -521,7 +608,7 @@ int mdt_pack_size2body(struct mdt_thread_info *info, RETURN(0); } -#ifdef CONFIG_FS_POSIX_ACL +#ifdef CONFIG_LUSTRE_FS_POSIX_ACL /* * Pack ACL data into the reply. UIDs/GIDs are mapped and filtered by nodemap. * @@ -539,13 +626,13 @@ int mdt_pack_acl2body(struct mdt_thread_info *info, struct mdt_body *repbody, struct md_object *next = mdt_object_child(o); struct lu_buf *buf = &info->mti_buf; struct mdt_device *mdt = info->mti_mdt; + struct req_capsule *pill = info->mti_pill; int rc; ENTRY; - 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); + buf->lb_buf = req_capsule_server_get(pill, &RMF_ACL); + buf->lb_len = req_capsule_get_size(pill, &RMF_ACL, RCL_SERVER); if (buf->lb_len == 0) RETURN(0); @@ -563,25 +650,27 @@ again: exp_connect_large_acl(info->mti_exp) && buf->lb_buf != info->mti_big_acl) { if (info->mti_big_acl == NULL) { + info->mti_big_aclsize = + min_t(unsigned int, + mdt->mdt_max_ea_size, + XATTR_SIZE_MAX); OBD_ALLOC_LARGE(info->mti_big_acl, - mdt->mdt_max_ea_size); + info->mti_big_aclsize); if (info->mti_big_acl == NULL) { + info->mti_big_aclsize = 0; CERROR("%s: unable to grow " DFID" ACL buffer\n", mdt_obd_name(mdt), PFID(mdt_object_fid(o))); RETURN(-ENOMEM); } - - info->mti_big_aclsize = - mdt->mdt_max_ea_size; } CDEBUG(D_INODE, "%s: grow the "DFID " ACL buffer to size %d\n", mdt_obd_name(mdt), PFID(mdt_object_fid(o)), - mdt->mdt_max_ea_size); + info->mti_big_aclsize); buf->lb_buf = info->mti_big_acl; buf->lb_len = info->mti_big_aclsize; @@ -593,6 +682,36 @@ again: mdt_obd_name(mdt), PFID(mdt_object_fid(o)), rc); } } else { + int client; + int server; + int acl_buflen; + int lmm_buflen = 0; + int lmmsize = 0; + + acl_buflen = req_capsule_get_size(pill, &RMF_ACL, RCL_SERVER); + if (acl_buflen >= rc) + goto map; + + /* If LOV/LMA EA is small, we can reuse part of their buffer */ + client = ptlrpc_req_get_repsize(pill->rc_req); + server = lustre_packed_msg_size(pill->rc_req->rq_repmsg); + if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER)) { + lmm_buflen = req_capsule_get_size(pill, &RMF_MDT_MD, + RCL_SERVER); + lmmsize = repbody->mbo_eadatasize; + } + + if (client < server - acl_buflen - lmm_buflen + rc + lmmsize) { + CDEBUG(D_INODE, "%s: client prepared buffer size %d " + "is not big enough with the ACL size %d (%d)\n", + mdt_obd_name(mdt), client, rc, + server - acl_buflen - lmm_buflen + rc + lmmsize); + repbody->mbo_aclsize = 0; + repbody->mbo_valid &= ~OBD_MD_FLACL; + RETURN(-ERANGE); + } + +map: if (buf->lb_buf == info->mti_big_acl) info->mti_big_acl_used = 1; @@ -603,6 +722,8 @@ again: CERROR("%s: nodemap_map_acl unable to parse "DFID " ACL: rc = %d\n", mdt_obd_name(mdt), PFID(mdt_object_fid(o)), rc); + repbody->mbo_aclsize = 0; + repbody->mbo_valid &= ~OBD_MD_FLACL; } else { repbody->mbo_aclsize = rc; repbody->mbo_valid |= OBD_MD_FLACL; @@ -659,6 +780,10 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b, b->mbo_ctime = attr->la_ctime; b->mbo_valid |= OBD_MD_FLCTIME; } + if (attr->la_valid & LA_BTIME) { + b->mbo_btime = attr->la_btime; + b->mbo_valid |= OBD_MD_FLBTIME; + } if (attr->la_valid & LA_FLAGS) { b->mbo_flags = attr->la_flags; b->mbo_valid |= OBD_MD_FLFLAGS; @@ -734,10 +859,15 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b, b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; } else if (info->mti_som_valid) { /* som is valid */ b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; + } else if (ma->ma_valid & MA_SOM) { /* lsom is valid */ + b->mbo_valid |= OBD_MD_FLLAZYSIZE | OBD_MD_FLLAZYBLOCKS; + b->mbo_size = ma->ma_som.ms_size; + b->mbo_blocks = ma->ma_som.ms_blocks; } } - if (fid != NULL && (b->mbo_valid & OBD_MD_FLSIZE)) + if (fid != NULL && (b->mbo_valid & OBD_MD_FLSIZE || + b->mbo_valid & OBD_MD_FLLAZYSIZE)) CDEBUG(D_VFSTRACE, DFID": returning size %llu\n", PFID(fid), (unsigned long long)b->mbo_size); @@ -850,8 +980,8 @@ int mdt_big_xattr_get(struct mdt_thread_info *info, struct mdt_object *o, RETURN(rc); } -int mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o, - struct md_attr *ma, const char *name) +int __mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o, + struct md_attr *ma, const char *name) { struct md_object *next = mdt_object_child(o); struct lu_buf *buf = &info->mti_buf; @@ -866,13 +996,15 @@ int mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o, buf->lb_len = ma->ma_lmv_size; LASSERT(!(ma->ma_valid & MA_LMV)); } else if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) { - buf->lb_buf = ma->ma_lmv; - buf->lb_len = ma->ma_lmv_size; + buf->lb_buf = ma->ma_default_lmv; + buf->lb_len = ma->ma_default_lmv_size; LASSERT(!(ma->ma_valid & MA_LMV_DEF)); } else { return -EINVAL; } + LASSERT(buf->lb_buf); + rc = mo_xattr_get(info->mti_env, next, buf, name); if (rc > 0) { @@ -898,7 +1030,7 @@ got: ma->ma_lmv_size = rc; ma->ma_valid |= MA_LMV; } else if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) { - ma->ma_lmv_size = rc; + ma->ma_default_lmv_size = rc; ma->ma_valid |= MA_LMV_DEF; } @@ -925,8 +1057,42 @@ got: return rc; } -static int mdt_attr_get_pfid(struct mdt_thread_info *info, - struct mdt_object *o, struct lu_fid *pfid) +int mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o, + struct md_attr *ma, const char *name) +{ + int rc; + + if (!info->mti_big_lmm) { + OBD_ALLOC(info->mti_big_lmm, PAGE_SIZE); + if (!info->mti_big_lmm) + return -ENOMEM; + info->mti_big_lmmsize = PAGE_SIZE; + } + + if (strcmp(name, XATTR_NAME_LOV) == 0) { + ma->ma_lmm = info->mti_big_lmm; + ma->ma_lmm_size = info->mti_big_lmmsize; + ma->ma_valid &= ~MA_LOV; + } else if (strcmp(name, XATTR_NAME_LMV) == 0) { + ma->ma_lmv = info->mti_big_lmm; + ma->ma_lmv_size = info->mti_big_lmmsize; + ma->ma_valid &= ~MA_LMV; + } else { + LBUG(); + } + + LASSERT(!info->mti_big_lmm_used); + rc = __mdt_stripe_get(info, o, ma, name); + /* since big_lmm is always used here, clear 'used' flag to avoid + * assertion in mdt_big_xattr_get(). + */ + info->mti_big_lmm_used = 0; + + return rc; +} + +int mdt_attr_get_pfid(struct mdt_thread_info *info, struct mdt_object *o, + struct lu_fid *pfid) { struct lu_buf *buf = &info->mti_buf; struct link_ea_header *leh; @@ -972,6 +1138,51 @@ static int mdt_attr_get_pfid(struct mdt_thread_info *info, RETURN(0); } +int mdt_attr_get_pfid_name(struct mdt_thread_info *info, struct mdt_object *o, + struct lu_fid *pfid, struct lu_name *lname) +{ + struct lu_buf *buf = &info->mti_buf; + struct link_ea_header *leh; + struct link_ea_entry *lee; + int reclen; + int rc; + + buf->lb_buf = info->mti_xattr_buf; + buf->lb_len = sizeof(info->mti_xattr_buf); + rc = mo_xattr_get(info->mti_env, mdt_object_child(o), buf, + XATTR_NAME_LINK); + if (rc == -ERANGE) { + rc = mdt_big_xattr_get(info, o, XATTR_NAME_LINK); + buf->lb_buf = info->mti_big_lmm; + buf->lb_len = info->mti_big_lmmsize; + } + if (rc < 0) + return rc; + + if (rc < sizeof(*leh)) { + CERROR("short LinkEA on "DFID": rc = %d\n", + PFID(mdt_object_fid(o)), rc); + return -ENODATA; + } + + leh = (struct link_ea_header *)buf->lb_buf; + lee = (struct link_ea_entry *)(leh + 1); + if (leh->leh_magic == __swab32(LINK_EA_MAGIC)) { + leh->leh_magic = LINK_EA_MAGIC; + leh->leh_reccount = __swab32(leh->leh_reccount); + leh->leh_len = __swab64(leh->leh_len); + } + if (leh->leh_magic != LINK_EA_MAGIC) + return -EINVAL; + + if (leh->leh_reccount == 0) + return -ENODATA; + + linkea_entry_unpack(lee, &reclen, lname, pfid); + + return 0; +} + int mdt_attr_get_complex(struct mdt_thread_info *info, struct mdt_object *o, struct md_attr *ma) { @@ -996,7 +1207,7 @@ int mdt_attr_get_complex(struct mdt_thread_info *info, GOTO(out, rc); if (S_ISREG(mode)) - (void) mdt_get_som(info, o, &ma->ma_attr); + (void) mdt_get_som(info, o, ma); ma->ma_valid |= MA_INODE; } @@ -1009,19 +1220,28 @@ int mdt_attr_get_complex(struct mdt_thread_info *info, } if (need & MA_LOV && (S_ISREG(mode) || S_ISDIR(mode))) { - rc = mdt_stripe_get(info, o, ma, XATTR_NAME_LOV); + rc = __mdt_stripe_get(info, o, ma, XATTR_NAME_LOV); if (rc) GOTO(out, rc); } if (need & MA_LMV && S_ISDIR(mode)) { - rc = mdt_stripe_get(info, o, ma, XATTR_NAME_LMV); + rc = __mdt_stripe_get(info, o, ma, XATTR_NAME_LMV); if (rc != 0) GOTO(out, rc); } if (need & MA_LMV_DEF && S_ISDIR(mode)) { - rc = mdt_stripe_get(info, o, ma, XATTR_NAME_DEFAULT_LMV); + rc = __mdt_stripe_get(info, o, ma, XATTR_NAME_DEFAULT_LMV); + if (rc != 0) + GOTO(out, rc); + } + + /* + * In the handle of MA_INODE, we may already get the SOM attr. + */ + if (need & MA_SOM && S_ISREG(mode) && !(ma->ma_valid & MA_SOM)) { + rc = mdt_get_som(info, o, ma); if (rc != 0) GOTO(out, rc); } @@ -1029,8 +1249,8 @@ int mdt_attr_get_complex(struct mdt_thread_info *info, if (need & MA_HSM && S_ISREG(mode)) { buf->lb_buf = info->mti_xattr_buf; buf->lb_len = sizeof(info->mti_xattr_buf); - CLASSERT(sizeof(struct hsm_attrs) <= - sizeof(info->mti_xattr_buf)); + BUILD_BUG_ON(sizeof(struct hsm_attrs) > + sizeof(info->mti_xattr_buf)); rc2 = mo_xattr_get(info->mti_env, next, buf, XATTR_NAME_HSM); rc2 = lustre_buf2hsm(info->mti_xattr_buf, rc2, &ma->ma_hsm); if (rc2 == 0) @@ -1039,7 +1259,7 @@ int mdt_attr_get_complex(struct mdt_thread_info *info, GOTO(out, rc = rc2); } -#ifdef CONFIG_FS_POSIX_ACL +#ifdef CONFIG_LUSTRE_FS_POSIX_ACL if (need & MA_ACL_DEF && S_ISDIR(mode)) { buf->lb_buf = ma->ma_acl; buf->lb_len = ma->ma_acl_size; @@ -1062,19 +1282,21 @@ out: } static int mdt_getattr_internal(struct mdt_thread_info *info, - struct mdt_object *o, int ma_need) + 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; - struct obd_export *exp = info->mti_exp; - int rc; + struct mdt_device *mdt = info->mti_mdt; + 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; + int rc; + ENTRY; if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) @@ -1110,34 +1332,64 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, req->rq_export->exp_client_uuid.uuid); } - /* If it is dir object and client require MEA, then we got MEA */ + /* from 2.12.58 intent_getattr pack default LMV in reply */ if (S_ISDIR(lu_object_attr(&next->mo_lu)) && - (reqbody->mbo_valid & (OBD_MD_MEA | OBD_MD_DEFAULT_MEA))) { + ((reqbody->mbo_valid & (OBD_MD_MEA | OBD_MD_DEFAULT_MEA)) == + (OBD_MD_MEA | OBD_MD_DEFAULT_MEA)) && + req_capsule_has_field(&req->rq_pill, &RMF_DEFAULT_MDT_MD, + RCL_SERVER)) { + ma->ma_lmv = buffer->lb_buf; + ma->ma_lmv_size = buffer->lb_len; + ma->ma_default_lmv = req_capsule_server_get(pill, + &RMF_DEFAULT_MDT_MD); + ma->ma_default_lmv_size = req_capsule_get_size(pill, + &RMF_DEFAULT_MDT_MD, + RCL_SERVER); + ma->ma_need = MA_INODE; + if (ma->ma_lmv_size > 0) + ma->ma_need |= MA_LMV; + if (ma->ma_default_lmv_size > 0) + ma->ma_need |= MA_LMV_DEF; + } else if (S_ISDIR(lu_object_attr(&next->mo_lu)) && + (reqbody->mbo_valid & (OBD_MD_MEA | OBD_MD_DEFAULT_MEA))) { + /* If it is dir and client require MEA, then we got MEA */ /* Assumption: MDT_MD size is enough for lmv size. */ ma->ma_lmv = buffer->lb_buf; ma->ma_lmv_size = buffer->lb_len; ma->ma_need = MA_INODE; if (ma->ma_lmv_size > 0) { - if (reqbody->mbo_valid & OBD_MD_MEA) + if (reqbody->mbo_valid & OBD_MD_MEA) { ma->ma_need |= MA_LMV; - else if (reqbody->mbo_valid & OBD_MD_DEFAULT_MEA) + } else if (reqbody->mbo_valid & OBD_MD_DEFAULT_MEA) { ma->ma_need |= MA_LMV_DEF; + ma->ma_default_lmv = buffer->lb_buf; + ma->ma_lmv = NULL; + ma->ma_default_lmv_size = buffer->lb_len; + ma->ma_lmv_size = 0; + } } } else { ma->ma_lmm = buffer->lb_buf; ma->ma_lmm_size = buffer->lb_len; ma->ma_need = MA_INODE | MA_HSM; - if (ma->ma_lmm_size > 0) + if (ma->ma_lmm_size > 0) { ma->ma_need |= MA_LOV; + /* Older clients may crash if they getattr overstriped + * files + */ + if (!exp_connect_overstriping(exp) && + mdt_lmm_is_overstriping(ma->ma_lmm)) + RETURN(-EOPNOTSUPP); + } } - if (S_ISDIR(lu_object_attr(&next->mo_lu)) && + if (S_ISDIR(lu_object_attr(&next->mo_lu)) && reqbody->mbo_valid & OBD_MD_FLDIREA && - lustre_msg_get_opc(req->rq_reqmsg) == MDS_GETATTR) { - /* get default stripe info for this dir. */ - ma->ma_need |= MA_LOV_DEF; - } - ma->ma_need |= ma_need; + lustre_msg_get_opc(req->rq_reqmsg) == MDS_GETATTR) { + /* get default stripe info for this dir. */ + ma->ma_need |= MA_LOV_DEF; + } + ma->ma_need |= ma_need; rc = mdt_attr_get_complex(info, o, ma); if (unlikely(rc)) { @@ -1156,22 +1408,27 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, repbody->mbo_t_state = MS_RESTORE; } - if (likely(ma->ma_valid & MA_INODE)) - mdt_pack_attr2body(info, repbody, la, mdt_object_fid(o)); - else - RETURN(-EFAULT); + if (unlikely(!(ma->ma_valid & MA_INODE))) + RETURN(-EFAULT); + + mdt_pack_attr2body(info, repbody, la, mdt_object_fid(o)); + + if (mdt_body_has_lov(la, reqbody)) { + u32 stripe_count = 1; - if (mdt_body_has_lov(la, reqbody)) { - if (ma->ma_valid & MA_LOV) { - LASSERT(ma->ma_lmm_size); + if (ma->ma_valid & MA_LOV) { + LASSERT(ma->ma_lmm_size); repbody->mbo_eadatasize = ma->ma_lmm_size; if (S_ISDIR(la->la_mode)) repbody->mbo_valid |= OBD_MD_FLDIREA; else repbody->mbo_valid |= OBD_MD_FLEASIZE; mdt_dump_lmm(D_INFO, ma->ma_lmm, repbody->mbo_valid); - } + } if (ma->ma_valid & MA_LMV) { + struct lmv_mds_md_v1 *lmv = &ma->ma_lmv->lmv_md_v1; + u32 magic = le32_to_cpu(lmv->lmv_magic); + /* Return -ENOTSUPP for old client */ if (!mdt_is_striped_client(req->rq_export)) RETURN(-ENOTSUPP); @@ -1180,17 +1437,41 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, mdt_dump_lmv(D_INFO, ma->ma_lmv); repbody->mbo_eadatasize = ma->ma_lmv_size; repbody->mbo_valid |= (OBD_MD_FLDIREA|OBD_MD_MEA); + + stripe_count = le32_to_cpu(lmv->lmv_stripe_count); + if (magic == LMV_MAGIC_STRIPE && lmv_is_restriping(lmv)) + mdt_restripe_migrate_add(info, o); + else if (magic == LMV_MAGIC_V1 && + lmv_is_restriping(lmv)) + mdt_restripe_update_add(info, o); } if (ma->ma_valid & MA_LMV_DEF) { /* Return -ENOTSUPP for old client */ if (!mdt_is_striped_client(req->rq_export)) RETURN(-ENOTSUPP); LASSERT(S_ISDIR(la->la_mode)); - mdt_dump_lmv(D_INFO, ma->ma_lmv); - repbody->mbo_eadatasize = ma->ma_lmv_size; + /* + * when ll_dir_getstripe() gets default LMV, it + * checks mbo_eadatasize. + */ + if (!(ma->ma_valid & MA_LMV)) + repbody->mbo_eadatasize = + ma->ma_default_lmv_size; repbody->mbo_valid |= (OBD_MD_FLDIREA | OBD_MD_DEFAULT_MEA); } + CDEBUG(D_VFSTRACE, + "dirent count %llu stripe count %u MDT count %d\n", + ma->ma_attr.la_dirent_count, stripe_count, + atomic_read(&mdt->mdt_mds_mds_conns) + 1); + if (ma->ma_attr.la_dirent_count != LU_DIRENT_COUNT_UNSET && + ma->ma_attr.la_dirent_count > + mdt->mdt_restriper.mdr_dir_split_count && + !fid_is_root(mdt_object_fid(o)) && + mdt->mdt_enable_dir_auto_split && + !o->mot_restriping && + stripe_count < atomic_read(&mdt->mdt_mds_mds_conns) + 1) + mdt_auto_split_add(info, o); } else if (S_ISLNK(la->la_mode) && reqbody->mbo_valid & OBD_MD_LINKNAME) { buffer->lb_buf = ma->ma_lmm; @@ -1228,8 +1509,8 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, print_limit < rc ? "..." : "", print_limit, (char *)ma->ma_lmm + rc - print_limit, rc); rc = 0; - } - } + } + } if (reqbody->mbo_valid & OBD_MD_FLMODEASIZE) { repbody->mbo_max_mdsize = info->mti_mdt->mdt_max_mdsize; @@ -1238,7 +1519,7 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, repbody->mbo_max_mdsize); } -#ifdef CONFIG_FS_POSIX_ACL +#ifdef CONFIG_LUSTRE_FS_POSIX_ACL if ((exp_connect_flags(req->rq_export) & OBD_CONNECT_ACL) && (reqbody->mbo_valid & OBD_MD_FLACL)) { struct lu_nodemap *nodemap = nodemap_get_from_exp(exp); @@ -1251,10 +1532,10 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, #endif out: - if (rc == 0) + if (rc == 0) mdt_counter_incr(req, LPROC_MDT_GETATTR); - RETURN(rc); + RETURN(rc); } static int mdt_getattr(struct tgt_session_info *tsi) @@ -1267,11 +1548,19 @@ static int mdt_getattr(struct tgt_session_info *tsi) int rc, rc2; ENTRY; - reqbody = req_capsule_client_get(pill, &RMF_MDT_BODY); - LASSERT(reqbody); - LASSERT(obj != NULL); + if (unlikely(info->mti_object == NULL)) + RETURN(-EPROTO); + + reqbody = req_capsule_client_get(pill, &RMF_MDT_BODY); + LASSERT(reqbody); LASSERT(lu_object_assert_exists(&obj->mot_obj)); + /* Special case for Data-on-MDT files to get data version */ + if (unlikely(reqbody->mbo_valid & OBD_MD_FLDATAVERSION)) { + rc = mdt_data_version_get(tsi); + GOTO(out, rc); + } + /* Unlike intent case where we need to pre-fill out buffers early on * in intent policy for ldlm reasons, here we can have a much better * guess at EA size by just reading it from disk. @@ -1281,7 +1570,6 @@ static int mdt_getattr(struct tgt_session_info *tsi) /* No easy way to know how long is the symlink, but it cannot * be more than PATH_MAX, so we allocate +1 */ rc = PATH_MAX + 1; - /* A special case for fs ROOT: getattr there might fetch * default EA for entire fs, not just for this dir! */ @@ -1325,12 +1613,12 @@ static int mdt_getattr(struct tgt_session_info *tsi) info->mti_cross_ref = !!(reqbody->mbo_valid & OBD_MD_FLCROSSREF); rc = mdt_getattr_internal(info, obj, 0); - EXIT; + EXIT; out_shrink: - mdt_client_compatibility(info); - rc2 = mdt_fix_reply(info); - if (rc == 0) - rc = rc2; + mdt_client_compatibility(info); + rc2 = mdt_fix_reply(info); + if (rc == 0) + rc = rc2; out: mdt_thread_info_fini(info); return rc; @@ -1339,41 +1627,59 @@ out: /** * Handler of layout intent RPC requiring the layout modification * - * \param[in] info thread environment - * \param[in] obj object - * \param[in] layout layout change descriptor + * \param[in] info thread environment + * \param[in] obj object + * \param[out] lhc object ldlm lock handle + * \param[in] layout layout change descriptor * * \retval 0 on success * \retval < 0 error code */ int mdt_layout_change(struct mdt_thread_info *info, struct mdt_object *obj, + struct mdt_lock_handle *lhc, struct md_layout_change *layout) { - struct mdt_lock_handle *lh = &info->mti_lh[MDT_LH_LOCAL]; int rc; + ENTRY; if (!mdt_object_exists(obj)) - GOTO(out, rc = -ENOENT); + RETURN(-ENOENT); if (!S_ISREG(lu_object_attr(&obj->mot_obj))) - GOTO(out, rc = -EINVAL); + RETURN(-EINVAL); rc = mo_permission(info->mti_env, NULL, mdt_object_child(obj), NULL, MAY_WRITE); if (rc) - GOTO(out, rc); + RETURN(rc); - /* take layout lock to prepare layout change */ - mdt_lock_reg_init(lh, LCK_EX); - rc = mdt_object_lock(info, obj, lh, MDS_INODELOCK_LAYOUT); - if (rc) - GOTO(out, rc); + rc = mdt_check_resent_lock(info, obj, lhc); + if (rc < 0) + RETURN(rc); + + if (rc > 0) { + /* not resent */ + __u64 lockpart = MDS_INODELOCK_LAYOUT; + + /* take layout lock to prepare layout change */ + if (layout->mlc_opc == MD_LAYOUT_WRITE) + lockpart |= MDS_INODELOCK_UPDATE; + mdt_lock_handle_init(lhc); + mdt_lock_reg_init(lhc, LCK_EX); + rc = mdt_reint_object_lock(info, obj, lhc, lockpart, false); + if (rc) + RETURN(rc); + } + + mutex_lock(&obj->mot_som_mutex); rc = mo_layout_change(info->mti_env, mdt_object_child(obj), layout); + mutex_unlock(&obj->mot_som_mutex); + + if (rc) + mdt_object_unlock(info, obj, lhc, 1); - mdt_object_unlock(info, obj, lh, 1); -out: RETURN(rc); } @@ -1421,6 +1727,8 @@ static int mdt_swap_layouts(struct tgt_session_info *tsi) RETURN(-EOPNOTSUPP); info = tsi2mdt_info(tsi); + if (unlikely(info->mti_object == NULL)) + RETURN(-EPROTO); if (info->mti_dlm_req != NULL) ldlm_request_cancel(req, info->mti_dlm_req, 0, LATF_SKIP); @@ -1444,12 +1752,12 @@ static int mdt_swap_layouts(struct tgt_session_info *tsi) /* permission check. Make sure the calling process having permission * to write both files. */ rc = mo_permission(info->mti_env, NULL, mdt_object_child(o1), NULL, - MAY_WRITE); + MAY_WRITE); if (rc < 0) GOTO(put, rc); rc = mo_permission(info->mti_env, NULL, mdt_object_child(o2), NULL, - MAY_WRITE); + MAY_WRITE); if (rc < 0) GOTO(put, rc); @@ -1538,6 +1846,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, struct lu_name *lname = NULL; struct mdt_lock_handle *lhp = NULL; struct ldlm_lock *lock; + struct req_capsule *pill = info->mti_pill; __u64 try_bits = 0; bool is_resent; int ma_need = 0; @@ -1569,12 +1878,15 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, mdt_lock_reg_init(lhc, LCK_PR); /* - * Object's name is on another MDS, no lookup or layout - * lock is needed here but update lock is. + * Object's name entry is on another MDS, it will + * request PERM lock only because LOOKUP lock is owned + * by the MDS where name entry resides. + * + * TODO: it should try layout lock too. - Jinshan */ child_bits &= ~(MDS_INODELOCK_LOOKUP | MDS_INODELOCK_LAYOUT); - child_bits |= MDS_INODELOCK_PERM | MDS_INODELOCK_UPDATE; + child_bits |= MDS_INODELOCK_PERM; rc = mdt_object_lock(info, child, lhc, child_bits); if (rc < 0) @@ -1591,21 +1903,26 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, } rc = mdt_getattr_internal(info, child, 0); - if (unlikely(rc != 0)) + if (unlikely(rc != 0)) { mdt_object_unlock(info, child, lhc, 1); + RETURN(rc); + } - RETURN(rc); - } + rc = mdt_pack_secctx_in_reply(info, child); + if (unlikely(rc)) + mdt_object_unlock(info, child, lhc, 1); + RETURN(rc); + } lname = &info->mti_name; - mdt_name_unpack(info->mti_pill, &RMF_NAME, lname, MNF_FIX_ANON); + mdt_name_unpack(pill, &RMF_NAME, lname, MNF_FIX_ANON); if (lu_name_is_valid(lname)) { CDEBUG(D_INODE, "getattr with lock for "DFID"/"DNAME", " "ldlm_rep = %p\n", PFID(mdt_object_fid(parent)), PNAME(lname), ldlm_rep); } else { - reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY); + reqbody = req_capsule_client_get(pill, &RMF_MDT_BODY); if (unlikely(reqbody == NULL)) RETURN(err_serious(-EPROTO)); @@ -1708,7 +2025,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, mdt_lock_reg_init(lhc, LCK_PR); if (!(child_bits & MDS_INODELOCK_UPDATE) && - mdt_object_exists(child) && !mdt_object_remote(child)) { + !mdt_object_remote(child)) { struct md_attr *ma = &info->mti_attr; ma->ma_valid = 0; @@ -1722,12 +2039,12 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, * lock and this might save us RPC on later STAT. For * directories, it also let negative dentry cache start * working for this dir. */ - if (ma->ma_valid & MA_INODE && - ma->ma_attr.la_valid & LA_CTIME && - info->mti_mdt->mdt_namespace->ns_ctime_age_limit + - ma->ma_attr.la_ctime < ktime_get_real_seconds()) - child_bits |= MDS_INODELOCK_UPDATE; - } + 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 < ktime_get_real_seconds()) + child_bits |= MDS_INODELOCK_UPDATE; + } /* layout lock must be granted in a best-effort way * for IT operations */ @@ -1758,17 +2075,25 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, child_bits &= ~MDS_INODELOCK_UPDATE; rc = mdt_object_lock(info, child, lhc, child_bits); } - if (unlikely(rc != 0)) - GOTO(out_child, rc); - } + if (unlikely(rc != 0)) + GOTO(out_child, rc); + } - lock = ldlm_handle2lock(&lhc->mlh_reg_lh); + /* finally, we can get attr for child. */ + rc = mdt_getattr_internal(info, child, ma_need); + if (unlikely(rc != 0)) { + mdt_object_unlock(info, child, lhc, 1); + GOTO(out_child, rc); + } + + rc = mdt_pack_secctx_in_reply(info, child); + if (unlikely(rc)) { + mdt_object_unlock(info, child, lhc, 1); + GOTO(out_child, rc); + } - /* finally, we can get attr for child. */ - rc = mdt_getattr_internal(info, child, ma_need); - if (unlikely(rc != 0)) { - mdt_object_unlock(info, child, lhc, 1); - } else if (lock) { + lock = ldlm_handle2lock(&lhc->mlh_reg_lh); + if (lock) { /* Debugging code. */ LDLM_DEBUG(lock, "Returning lock to client"); LASSERTF(fid_res_name_eq(mdt_object_fid(child), @@ -1778,36 +2103,31 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, PFID(mdt_object_fid(child))); if (S_ISREG(lu_object_attr(&child->mot_obj)) && - mdt_object_exists(child) && !mdt_object_remote(child) && - child != parent) { - LDLM_LOCK_PUT(lock); + !mdt_object_remote(child) && child != parent) { mdt_object_put(info->mti_env, child); - /* NB: call the mdt_pack_size2body always after - * mdt_object_put(), that is why this speacial - * exit path is used. */ rc = mdt_pack_size2body(info, child_fid, - child_bits & MDS_INODELOCK_DOM); + &lhc->mlh_reg_lh); if (rc != 0 && child_bits & MDS_INODELOCK_DOM) { /* DOM lock was taken in advance but this is - * not DoM file. Drop the lock. */ + * not DoM file. Drop the lock. + */ lock_res_and_lock(lock); ldlm_inodebits_drop(lock, MDS_INODELOCK_DOM); unlock_res_and_lock(lock); } - + LDLM_LOCK_PUT(lock); GOTO(out_parent, rc = 0); } - } - if (lock) - LDLM_LOCK_PUT(lock); + LDLM_LOCK_PUT(lock); + } - EXIT; + EXIT; out_child: - mdt_object_put(info->mti_env, child); + mdt_object_put(info->mti_env, child); out_parent: - if (lhp) - mdt_object_unlock(info, parent, lhp, 1); - return rc; + if (lhp) + mdt_object_unlock(info, parent, lhp, 1); + return rc; } /* normal handler: should release the child lock */ @@ -1849,6 +2169,219 @@ out_shrink: return rc; } +static int mdt_rmfid_unlink(struct mdt_thread_info *info, + const struct lu_fid *pfid, + const struct lu_name *name, + struct mdt_object *obj, s64 ctime) +{ + struct lu_fid *child_fid = &info->mti_tmp_fid1; + struct ldlm_enqueue_info *einfo = &info->mti_einfo[0]; + struct mdt_device *mdt = info->mti_mdt; + struct md_attr *ma = &info->mti_attr; + struct mdt_lock_handle *parent_lh; + struct mdt_lock_handle *child_lh; + struct mdt_object *pobj; + bool cos_incompat = false; + int rc; + ENTRY; + + pobj = mdt_object_find(info->mti_env, mdt, pfid); + if (IS_ERR(pobj)) + GOTO(out, rc = PTR_ERR(pobj)); + + parent_lh = &info->mti_lh[MDT_LH_PARENT]; + mdt_lock_pdo_init(parent_lh, LCK_PW, name); + rc = mdt_object_lock(info, pobj, parent_lh, MDS_INODELOCK_UPDATE); + if (rc != 0) + GOTO(put_parent, rc); + + if (mdt_object_remote(pobj)) + cos_incompat = true; + + rc = mdo_lookup(info->mti_env, mdt_object_child(pobj), + name, child_fid, &info->mti_spec); + if (rc != 0) + GOTO(unlock_parent, rc); + + if (!lu_fid_eq(child_fid, mdt_object_fid(obj))) + GOTO(unlock_parent, rc = -EREMCHG); + + child_lh = &info->mti_lh[MDT_LH_CHILD]; + mdt_lock_reg_init(child_lh, LCK_EX); + rc = mdt_reint_striped_lock(info, obj, child_lh, + MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE, + einfo, cos_incompat); + if (rc != 0) + GOTO(unlock_parent, rc); + + if (atomic_read(&obj->mot_open_count)) { + CDEBUG(D_OTHER, "object "DFID" open, skip\n", + PFID(mdt_object_fid(obj))); + GOTO(unlock_child, rc = -EBUSY); + } + + ma->ma_need = 0; + ma->ma_valid = MA_INODE; + ma->ma_attr.la_valid = LA_CTIME; + ma->ma_attr.la_ctime = ctime; + + mutex_lock(&obj->mot_lov_mutex); + + rc = mdo_unlink(info->mti_env, mdt_object_child(pobj), + mdt_object_child(obj), name, ma, 0); + + mutex_unlock(&obj->mot_lov_mutex); + +unlock_child: + mdt_reint_striped_unlock(info, obj, child_lh, einfo, 1); +unlock_parent: + mdt_object_unlock(info, pobj, parent_lh, 1); +put_parent: + mdt_object_put(info->mti_env, pobj); +out: + RETURN(rc); +} + +static int mdt_rmfid_check_permission(struct mdt_thread_info *info, + struct mdt_object *obj) +{ + struct lu_ucred *uc = lu_ucred(info->mti_env); + struct md_attr *ma = &info->mti_attr; + struct lu_attr *la = &ma->ma_attr; + int rc = 0; + ENTRY; + + ma->ma_need = MA_INODE; + rc = mo_attr_get(info->mti_env, mdt_object_child(obj), ma); + if (rc) + GOTO(out, rc); + + if (la->la_flags & LUSTRE_IMMUTABLE_FL) + rc = -EACCES; + + if (md_capable(uc, CFS_CAP_DAC_OVERRIDE)) + RETURN(0); + if (uc->uc_fsuid == la->la_uid) { + if ((la->la_mode & S_IWUSR) == 0) + rc = -EACCES; + } else if (uc->uc_fsgid == la->la_gid) { + if ((la->la_mode & S_IWGRP) == 0) + rc = -EACCES; + } else if ((la->la_mode & S_IWOTH) == 0) { + rc = -EACCES; + } + +out: + RETURN(rc); +} + +static int mdt_rmfid_one(struct mdt_thread_info *info, struct lu_fid *fid, + s64 ctime) +{ + struct mdt_device *mdt = info->mti_mdt; + struct mdt_object *obj = NULL; + struct linkea_data ldata = { NULL }; + struct lu_buf *buf = &info->mti_big_buf; + struct lu_name *name = &info->mti_name; + struct lu_fid *pfid = &info->mti_tmp_fid1; + struct link_ea_header *leh; + struct link_ea_entry *lee; + int reclen, count, rc = 0; + ENTRY; + + if (!fid_is_sane(fid)) + GOTO(out, rc = -EINVAL); + + if (!fid_is_namespace_visible(fid)) + GOTO(out, rc = -EINVAL); + + obj = mdt_object_find(info->mti_env, mdt, fid); + if (IS_ERR(obj)) + GOTO(out, rc = PTR_ERR(obj)); + + if (mdt_object_remote(obj)) + GOTO(out, rc = -EREMOTE); + if (!mdt_object_exists(obj) || lu_object_is_dying(&obj->mot_header)) + GOTO(out, rc = -ENOENT); + + rc = mdt_rmfid_check_permission(info, obj); + if (rc) + GOTO(out, rc); + + /* take LinkEA */ + buf = lu_buf_check_and_alloc(buf, PATH_MAX); + if (!buf->lb_buf) + GOTO(out, rc = -ENOMEM); + + ldata.ld_buf = buf; + rc = mdt_links_read(info, obj, &ldata); + if (rc) + GOTO(out, rc); + + leh = buf->lb_buf; + lee = (struct link_ea_entry *)(leh + 1); + for (count = 0; count < leh->leh_reccount; count++) { + /* remove every hardlink */ + linkea_entry_unpack(lee, &reclen, name, pfid); + lee = (struct link_ea_entry *) ((char *)lee + reclen); + rc = mdt_rmfid_unlink(info, pfid, name, obj, ctime); + if (rc) + break; + } + +out: + if (obj && !IS_ERR(obj)) + mdt_object_put(info->mti_env, obj); + if (info->mti_big_buf.lb_buf) + lu_buf_free(&info->mti_big_buf); + + RETURN(rc); +} + +static int mdt_rmfid(struct tgt_session_info *tsi) +{ + struct mdt_thread_info *mti = tsi2mdt_info(tsi); + struct mdt_body *reqbody; + struct lu_fid *fids, *rfids; + int bufsize, rc; + __u32 *rcs; + int i, nr; + ENTRY; + + reqbody = req_capsule_client_get(tsi->tsi_pill, &RMF_MDT_BODY); + if (reqbody == NULL) + RETURN(-EPROTO); + bufsize = req_capsule_get_size(tsi->tsi_pill, &RMF_FID_ARRAY, + RCL_CLIENT); + nr = bufsize / sizeof(struct lu_fid); + if (nr * sizeof(struct lu_fid) != bufsize) + RETURN(-EINVAL); + req_capsule_set_size(tsi->tsi_pill, &RMF_RCS, + RCL_SERVER, nr * sizeof(__u32)); + req_capsule_set_size(tsi->tsi_pill, &RMF_FID_ARRAY, + RCL_SERVER, nr * sizeof(struct lu_fid)); + rc = req_capsule_server_pack(tsi->tsi_pill); + if (rc) + GOTO(out, rc = err_serious(rc)); + fids = req_capsule_client_get(tsi->tsi_pill, &RMF_FID_ARRAY); + if (fids == NULL) + RETURN(-EPROTO); + rcs = req_capsule_server_get(tsi->tsi_pill, &RMF_RCS); + LASSERT(rcs); + rfids = req_capsule_server_get(tsi->tsi_pill, &RMF_FID_ARRAY); + LASSERT(rfids); + + mdt_init_ucred(mti, reqbody); + for (i = 0; i < nr; i++) { + rfids[i] = fids[i]; + rcs[i] = mdt_rmfid_one(mti, fids + i, reqbody->mbo_ctime); + } + mdt_exit_ucred(mti); + +out: + RETURN(rc); +} + static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len, void *karg, void __user *uarg); @@ -1902,6 +2435,8 @@ static int mdt_set_info(struct tgt_session_info *tsi) __swab32s(&cs->cs_id); } + if (!mdt_is_rootadmin(tsi2mdt_info(tsi))) + RETURN(-EACCES); rc = mdt_iocontrol(OBD_IOC_CHANGELOG_CLEAR, req->rq_export, vallen, val, NULL); } else if (KEY_IS(KEY_EVICT_BY_NID)) { @@ -1951,17 +2486,17 @@ static int mdt_readpage(struct tgt_session_info *tsi) exp_max_brw_size(tsi->tsi_exp)); rdpg->rp_npages = (rdpg->rp_count + PAGE_SIZE - 1) >> PAGE_SHIFT; - OBD_ALLOC(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]); - if (rdpg->rp_pages == NULL) - RETURN(-ENOMEM); + OBD_ALLOC_PTR_ARRAY(rdpg->rp_pages, rdpg->rp_npages); + if (rdpg->rp_pages == NULL) + RETURN(-ENOMEM); - for (i = 0; i < rdpg->rp_npages; ++i) { + for (i = 0; i < rdpg->rp_npages; ++i) { rdpg->rp_pages[i] = alloc_page(GFP_NOFS); - if (rdpg->rp_pages[i] == NULL) - GOTO(free_rdpg, rc = -ENOMEM); - } + if (rdpg->rp_pages[i] == NULL) + GOTO(free_rdpg, rc = -ENOMEM); + } - /* call lower layers to fill allocated pages with directory data */ + /* call lower layers to fill allocated pages with directory data */ rc = mo_readpage(tsi->tsi_env, mdt_object_child(object), rdpg); if (rc < 0) GOTO(free_rdpg, rc); @@ -1975,7 +2510,7 @@ free_rdpg: for (i = 0; i < rdpg->rp_npages; i++) if (rdpg->rp_pages[i] != NULL) __free_page(rdpg->rp_pages[i]); - OBD_FREE(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]); + OBD_FREE_PTR_ARRAY(rdpg->rp_pages, rdpg->rp_npages); if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE)) RETURN(0); @@ -2003,6 +2538,34 @@ static int mdt_fix_attr_ucred(struct mdt_thread_info *info, __u32 op) return 0; } +static inline bool mdt_is_readonly_open(struct mdt_thread_info *info, __u32 op) +{ + return op == REINT_OPEN && + !(info->mti_spec.sp_cr_flags & (MDS_FMODE_WRITE | MDS_OPEN_CREAT)); +} + +static void mdt_preset_secctx_size(struct mdt_thread_info *info) +{ + struct req_capsule *pill = info->mti_pill; + + if (req_capsule_has_field(pill, &RMF_FILE_SECCTX, + RCL_SERVER) && + req_capsule_has_field(pill, &RMF_FILE_SECCTX_NAME, + RCL_CLIENT)) { + if (req_capsule_get_size(pill, &RMF_FILE_SECCTX_NAME, + RCL_CLIENT) != 0) { + /* pre-set size in server part with max size */ + req_capsule_set_size(pill, &RMF_FILE_SECCTX, + RCL_SERVER, + info->mti_mdt->mdt_max_ea_size); + } else { + req_capsule_set_size(pill, &RMF_FILE_SECCTX, + RCL_SERVER, 0); + } + } + +} + static int mdt_reint_internal(struct mdt_thread_info *info, struct mdt_lock_handle *lhc, __u32 op) @@ -2013,19 +2576,25 @@ static int mdt_reint_internal(struct mdt_thread_info *info, ENTRY; - rc = mdt_reint_unpack(info, op); - if (rc != 0) { - CERROR("Can't unpack reint, rc %d\n", rc); - RETURN(err_serious(rc)); - } + rc = mdt_reint_unpack(info, op); + if (rc != 0) { + CERROR("Can't unpack reint, rc %d\n", rc); + RETURN(err_serious(rc)); + } - /* for replay (no_create) lmm is not needed, client has it already */ - if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER)) - req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, + + /* check if the file system is set to readonly. O_RDONLY open + * is still allowed even the file system is set to readonly mode */ + if (mdt_rdonly(info->mti_exp) && !mdt_is_readonly_open(info, op)) + RETURN(err_serious(-EROFS)); + + /* for replay (no_create) lmm is not needed, client has it already */ + if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER)) + req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, DEF_REP_MD_SIZE); /* llog cookies are always 0, the field is kept for compatibility */ - if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER)) + if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER)) req_capsule_set_size(pill, &RMF_LOGCOOKIES, RCL_SERVER, 0); /* Set ACL reply buffer size as LUSTRE_POSIX_ACL_MAX_SIZE_OLD @@ -2035,52 +2604,64 @@ static int mdt_reint_internal(struct mdt_thread_info *info, req_capsule_set_size(pill, &RMF_ACL, RCL_SERVER, LUSTRE_POSIX_ACL_MAX_SIZE_OLD); - rc = req_capsule_server_pack(pill); - if (rc != 0) { - CERROR("Can't pack response, rc %d\n", rc); - RETURN(err_serious(rc)); - } + mdt_preset_secctx_size(info); + + rc = req_capsule_server_pack(pill); + if (rc != 0) { + CERROR("Can't pack response, rc %d\n", rc); + RETURN(err_serious(rc)); + } - if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_SERVER)) { - repbody = req_capsule_server_get(pill, &RMF_MDT_BODY); - LASSERT(repbody); + if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_SERVER)) { + repbody = req_capsule_server_get(pill, &RMF_MDT_BODY); + LASSERT(repbody); repbody->mbo_eadatasize = 0; repbody->mbo_aclsize = 0; - } + } - OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_REINT_DELAY, 10); + OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_REINT_DELAY, 10); - /* for replay no cookkie / lmm need, because client have this already */ - if (info->mti_spec.no_create) - if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER)) - req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, 0); + /* for replay no cookkie / lmm need, because client have this already */ + if (info->mti_spec.no_create) + if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER)) + req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, 0); - rc = mdt_init_ucred_reint(info); - if (rc) - GOTO(out_shrink, rc); + rc = mdt_init_ucred_reint(info); + if (rc) + GOTO(out_shrink, rc); - rc = mdt_fix_attr_ucred(info, op); - if (rc != 0) - GOTO(out_ucred, rc = err_serious(rc)); + rc = mdt_fix_attr_ucred(info, op); + if (rc != 0) + GOTO(out_ucred, rc = err_serious(rc)); rc = mdt_check_resent(info, mdt_reconstruct, lhc); if (rc < 0) { GOTO(out_ucred, rc); } else if (rc == 1) { - DEBUG_REQ(D_INODE, mdt_info_req(info), "resent opt."); + DEBUG_REQ(D_INODE, mdt_info_req(info), "resent opt"); rc = lustre_msg_get_status(mdt_info_req(info)->rq_repmsg); - GOTO(out_ucred, rc); - } - rc = mdt_reint_rec(info, lhc); - EXIT; + GOTO(out_ucred, rc); + } + rc = mdt_reint_rec(info, lhc); + EXIT; out_ucred: - mdt_exit_ucred(info); + mdt_exit_ucred(info); out_shrink: - mdt_client_compatibility(info); - rc2 = mdt_fix_reply(info); - if (rc == 0) - rc = rc2; - return rc; + mdt_client_compatibility(info); + + rc2 = mdt_fix_reply(info); + if (rc == 0) + rc = rc2; + + /* + * Data-on-MDT optimization - read data along with OPEN and return it + * in reply when possible. + */ + if (rc == 0 && op == REINT_OPEN && !req_is_replay(pill->rc_req)) + rc = mdt_dom_read_on_open(info, info->mti_mdt, + &lhc->mlh_reg_lh); + + return rc; } static long mdt_reint_opcode(struct ptlrpc_request *req, @@ -2122,7 +2703,7 @@ static int mdt_reint(struct tgt_session_info *tsi) [REINT_OPEN] = &RQF_MDS_REINT_OPEN, [REINT_SETXATTR] = &RQF_MDS_REINT_SETXATTR, [REINT_RMENTRY] = &RQF_MDS_REINT_UNLINK, - [REINT_MIGRATE] = &RQF_MDS_REINT_RENAME, + [REINT_MIGRATE] = &RQF_MDS_REINT_MIGRATE, [REINT_RESYNC] = &RQF_MDS_REINT_RESYNC, }; @@ -2146,7 +2727,7 @@ static int mdt_reint(struct tgt_session_info *tsi) } /* this should sync the whole device */ -static int mdt_device_sync(const struct lu_env *env, struct mdt_device *mdt) +int mdt_device_sync(const struct lu_env *env, struct mdt_device *mdt) { struct dt_device *dt = mdt->mdt_bottom; int rc; @@ -2160,7 +2741,7 @@ static int mdt_device_sync(const struct lu_env *env, struct mdt_device *mdt) static int mdt_object_sync(const struct lu_env *env, struct obd_export *exp, struct mdt_object *mo) { - int rc; + int rc = 0; ENTRY; @@ -2171,7 +2752,16 @@ static int mdt_object_sync(const struct lu_env *env, struct obd_export *exp, RETURN(-ESTALE); } - rc = mo_object_sync(env, mdt_object_child(mo)); + if (S_ISREG(lu_object_attr(&mo->mot_obj))) { + struct lu_target *tgt = tgt_ses_info(env)->tsi_tgt; + dt_obj_version_t version; + + version = dt_version_get(env, mdt_obj2dt(mo)); + if (version > tgt->lut_obd->obd_last_committed) + rc = mo_object_sync(env, mdt_object_child(mo)); + } else { + rc = mo_object_sync(env, mdt_object_child(mo)); + } RETURN(rc); } @@ -2193,6 +2783,9 @@ static int mdt_sync(struct tgt_session_info *tsi) } else { struct mdt_thread_info *info = tsi2mdt_info(tsi); + if (unlikely(info->mti_object == NULL)) + RETURN(-EPROTO); + /* sync an object */ rc = mdt_object_sync(tsi->tsi_env, tsi->tsi_exp, info->mti_object); @@ -2273,17 +2866,17 @@ put: */ static int mdt_quotactl(struct tgt_session_info *tsi) { - struct obd_export *exp = tsi->tsi_exp; - struct req_capsule *pill = tsi->tsi_pill; - struct obd_quotactl *oqctl, *repoqc; - int id, rc; - struct mdt_device *mdt = mdt_exp2dev(exp); - struct lu_device *qmt = mdt->mdt_qmt_dev; - struct lu_nodemap *nodemap; + struct obd_export *exp = tsi->tsi_exp; + struct req_capsule *pill = tsi->tsi_pill; + struct obd_quotactl *oqctl, *repoqc; + int id, rc; + struct mdt_device *mdt = mdt_exp2dev(exp); + struct lu_device *qmt = mdt->mdt_qmt_dev; + struct lu_nodemap *nodemap; ENTRY; oqctl = req_capsule_client_get(pill, &RMF_OBD_QUOTACTL); - if (oqctl == NULL) + if (!oqctl) RETURN(err_serious(-EPROTO)); rc = req_capsule_server_pack(pill); @@ -2298,19 +2891,29 @@ static int mdt_quotactl(struct tgt_session_info *tsi) /* master quotactl */ case Q_SETINFO: case Q_SETQUOTA: + case LUSTRE_Q_SETDEFAULT: + case LUSTRE_Q_SETQUOTAPOOL: + case LUSTRE_Q_SETINFOPOOL: if (!nodemap_can_setquota(nodemap)) GOTO(out_nodemap, rc = -EPERM); + /* fallthrough */ case Q_GETINFO: case Q_GETQUOTA: + case LUSTRE_Q_GETDEFAULT: + case LUSTRE_Q_GETQUOTAPOOL: + case LUSTRE_Q_GETINFOPOOL: if (qmt == NULL) GOTO(out_nodemap, rc = -EOPNOTSUPP); /* slave quotactl */ + /* fallthrough */ case Q_GETOINFO: case Q_GETOQUOTA: break; default: - CERROR("Unsupported quotactl command: %d\n", oqctl->qc_cmd); - GOTO(out_nodemap, rc = -EFAULT); + rc = -EFAULT; + CERROR("%s: unsupported quotactl command %d: rc = %d\n", + mdt_obd_name(mdt), oqctl->qc_cmd, rc); + GOTO(out_nodemap, rc); } id = oqctl->qc_id; @@ -2351,6 +2954,12 @@ static int mdt_quotactl(struct tgt_session_info *tsi) case Q_SETINFO: case Q_SETQUOTA: case Q_GETQUOTA: + case LUSTRE_Q_SETDEFAULT: + case LUSTRE_Q_GETDEFAULT: + case LUSTRE_Q_SETQUOTAPOOL: + case LUSTRE_Q_GETQUOTAPOOL: + case LUSTRE_Q_SETINFOPOOL: + case LUSTRE_Q_GETINFOPOOL: /* forward quotactl request to QMT */ rc = qmt_hdls.qmth_quotactl(tsi->tsi_env, qmt, oqctl); break; @@ -2370,8 +2979,7 @@ static int mdt_quotactl(struct tgt_session_info *tsi) if (oqctl->qc_id != id) swap(oqctl->qc_id, id); - *repoqc = *oqctl; - + QCTL_COPY(repoqc, oqctl); EXIT; out_nodemap: @@ -2558,6 +3166,7 @@ int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, { struct obd_device *obd = ldlm_lock_to_ns(lock)->ns_obd; struct mdt_device *mdt = mdt_dev(obd->obd_lu_dev); + struct ldlm_cb_set_arg *arg = data; bool commit_async = false; int rc; ENTRY; @@ -2570,17 +3179,22 @@ int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, unlock_res_and_lock(lock); RETURN(0); } - /* There is no lock conflict if l_blocking_lock == NULL, - * it indicates a blocking ast sent from ldlm_lock_decref_internal - * when the last reference to a local lock was released */ - if (lock->l_req_mode & (LCK_PW | LCK_EX) && - lock->l_blocking_lock != NULL) { + + /* A blocking ast may be sent from ldlm_lock_decref_internal + * when the last reference to a local lock was released and + * during blocking event from ldlm_work_bl_ast_lock(). + * The 'data' parameter is l_ast_data in the first case and + * callback arguments in the second one. Distinguish them by that. + */ + if (!data || data == lock->l_ast_data || !arg->bl_desc) + goto skip_cos_checks; + + if (lock->l_req_mode & (LCK_PW | LCK_EX)) { if (mdt_cos_is_enabled(mdt)) { - if (lock->l_client_cookie != - lock->l_blocking_lock->l_client_cookie) + if (!arg->bl_desc->bl_same_client) mdt_set_lock_sync(lock); } else if (mdt_slc_is_enabled(mdt) && - ldlm_is_cos_incompat(lock->l_blocking_lock)) { + arg->bl_desc->bl_cos_incompat) { mdt_set_lock_sync(lock); /* * we may do extra commit here, but there is a small @@ -2594,11 +3208,11 @@ int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, */ commit_async = true; } - } else if (lock->l_req_mode == LCK_COS && - lock->l_blocking_lock != NULL) { + } else if (lock->l_req_mode == LCK_COS) { commit_async = true; } +skip_cos_checks: rc = ldlm_blocking_ast_nocheck(lock); if (commit_async) { @@ -2665,10 +3279,9 @@ int mdt_remote_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, rc = lu_env_init(&env, LCT_MD_THREAD); if (unlikely(rc != 0)) { - CWARN("%s: lu_env initialization failed, object" - "%p "DFID" is leaked!\n", + CWARN("%s: lu_env initialization failed, object %p "DFID" is leaked!: rc = %d\n", obd->obd_name, mo, - PFID(mdt_object_fid(mo))); + PFID(mdt_object_fid(mo)), rc); RETURN(rc); } @@ -2724,12 +3337,17 @@ int mdt_check_resent_lock(struct mdt_thread_info *info, return 1; } +static void mdt_remote_object_lock_created_cb(struct ldlm_lock *lock) +{ + mdt_object_get(NULL, lock->l_ast_data); +} + int mdt_remote_object_lock_try(struct mdt_thread_info *mti, struct mdt_object *o, const struct lu_fid *fid, struct lustre_handle *lh, enum ldlm_mode mode, __u64 *ibits, __u64 trybits, bool cache) { - struct ldlm_enqueue_info *einfo = &mti->mti_einfo; + struct ldlm_enqueue_info *einfo = &mti->mti_remote_einfo; union ldlm_policy_data *policy = &mti->mti_policy; struct ldlm_res_id *res_id = &mti->mti_res_id; int rc = 0; @@ -2752,21 +3370,16 @@ int mdt_remote_object_lock_try(struct mdt_thread_info *mti, * if we cache lock, couple lock with mdt_object, so that object * can be easily found in lock ASTs. */ - mdt_object_get(mti->mti_env, o); einfo->ei_cbdata = o; + einfo->ei_cb_created = mdt_remote_object_lock_created_cb; } - memset(policy, 0, sizeof(*policy)); policy->l_inodebits.bits = *ibits; policy->l_inodebits.try_bits = trybits; rc = mo_object_lock(mti->mti_env, mdt_object_child(o), lh, einfo, policy); - if (rc < 0 && cache) { - mdt_object_put(mti->mti_env, o); - einfo->ei_cbdata = NULL; - } /* Return successfully acquired bits to a caller */ if (rc == 0) { @@ -2795,7 +3408,7 @@ static int mdt_object_local_lock(struct mdt_thread_info *info, struct ldlm_namespace *ns = info->mti_mdt->mdt_namespace; union ldlm_policy_data *policy = &info->mti_policy; struct ldlm_res_id *res_id = &info->mti_res_id; - __u64 dlmflags = 0; + __u64 dlmflags = 0, *cookie = NULL; int rc; ENTRY; @@ -2813,9 +3426,12 @@ static int mdt_object_local_lock(struct mdt_thread_info *info, } /* Only enqueue LOOKUP lock for remote object */ - if (mdt_object_remote(o)) { - LASSERT(*ibits == MDS_INODELOCK_LOOKUP); - } + LASSERT(ergo(mdt_object_remote(o), *ibits == MDS_INODELOCK_LOOKUP)); + + /* Lease lock are granted with LDLM_FL_CANCEL_ON_BLOCK */ + if (lh->mlh_type == MDT_REG_LOCK && lh->mlh_reg_mode == LCK_EX && + *ibits == MDS_INODELOCK_OPEN) + dlmflags |= LDLM_FL_CANCEL_ON_BLOCK; if (lh->mlh_type == MDT_PDO_LOCK) { /* check for exists after object is locked */ @@ -2829,10 +3445,12 @@ static int mdt_object_local_lock(struct mdt_thread_info *info, } } - fid_build_reg_res_name(mdt_object_fid(o), res_id); dlmflags |= LDLM_FL_ATOMIC_CB; + if (info->mti_exp) + cookie = &info->mti_exp->exp_handle.h_cookie; + /* * Take PDO lock on whole directory and build correct @res_id for lock * on part of directory. @@ -2846,12 +3464,16 @@ static int mdt_object_local_lock(struct mdt_thread_info *info, * is never going to be sent to client and we do not * want it slowed down due to possible cancels. */ - policy->l_inodebits.bits = MDS_INODELOCK_UPDATE; - policy->l_inodebits.try_bits = 0; - rc = mdt_fid_lock(ns, &lh->mlh_pdo_lh, lh->mlh_pdo_mode, - policy, res_id, dlmflags, - info->mti_exp == NULL ? NULL : - &info->mti_exp->exp_handle.h_cookie); + policy->l_inodebits.bits = + *ibits & MDS_INODELOCK_UPDATE; + policy->l_inodebits.try_bits = + trybits & MDS_INODELOCK_UPDATE; + /* at least one of them should be set */ + LASSERT(policy->l_inodebits.bits | + policy->l_inodebits.try_bits); + rc = mdt_fid_lock(info->mti_env, ns, &lh->mlh_pdo_lh, + lh->mlh_pdo_mode, policy, res_id, + dlmflags, cookie); if (unlikely(rc != 0)) GOTO(out_unlock, rc); } @@ -2871,10 +3493,9 @@ static int mdt_object_local_lock(struct mdt_thread_info *info, * going to be sent to client. If it is - mdt_intent_policy() path will * fix it up and turn FL_LOCAL flag off. */ - rc = mdt_fid_lock(ns, &lh->mlh_reg_lh, lh->mlh_reg_mode, policy, - res_id, LDLM_FL_LOCAL_ONLY | dlmflags, - info->mti_exp == NULL ? NULL : - &info->mti_exp->exp_handle.h_cookie); + rc = mdt_fid_lock(info->mti_env, ns, &lh->mlh_reg_lh, lh->mlh_reg_mode, + policy, res_id, LDLM_FL_LOCAL_ONLY | dlmflags, + cookie); out_unlock: if (rc != 0) mdt_object_unlock(info, o, lh, 1); @@ -2932,8 +3553,9 @@ mdt_object_lock_internal(struct mdt_thread_info *info, struct mdt_object *o, * object anyway XXX*/ if (lh->mlh_type == MDT_PDO_LOCK && lh->mlh_pdo_hash != 0) { - CDEBUG(D_INFO, "%s: "DFID" convert PDO lock to" - "EX lock.\n", mdt_obd_name(info->mti_mdt), + CDEBUG(D_INFO, + "%s: "DFID" convert PDO lock to EX lock.\n", + mdt_obd_name(info->mti_mdt), PFID(mdt_object_fid(o))); lh->mlh_pdo_hash = 0; lh->mlh_rreg_mode = LCK_EX; @@ -2951,6 +3573,10 @@ mdt_object_lock_internal(struct mdt_thread_info *info, struct mdt_object *o, } } + /* other components like LFSCK can use lockless access + * and populate cache, so we better invalidate it */ + mo_invalidate(info->mti_env, mdt_object_child(o)); + RETURN(0); } @@ -2973,8 +3599,17 @@ int mdt_object_lock_try(struct mdt_thread_info *info, struct mdt_object *o, struct mdt_lock_handle *lh, __u64 *ibits, __u64 trybits, bool cos_incompat) { - return mdt_object_lock_internal(info, o, lh, ibits, trybits, - cos_incompat); + bool trylock_only = *ibits == 0; + int rc; + + LASSERT(!(*ibits & trybits)); + rc = mdt_object_lock_internal(info, o, lh, ibits, trybits, + cos_incompat); + if (rc && trylock_only) { /* clear error for try ibits lock only */ + LASSERT(*ibits == 0); + rc = 0; + } + return rc; } /** @@ -3016,7 +3651,7 @@ void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h, "state %p transno %lld\n", req, req->rq_reply_state, req->rq_transno); if (cos) { - ldlm_lock_downgrade(lock, LCK_COS); + ldlm_lock_mode_downgrade(lock, LCK_COS); mode = LCK_COS; } if (req->rq_export->exp_disconnected) @@ -3060,20 +3695,18 @@ static void mdt_save_remote_lock(struct mdt_thread_info *info, if (lustre_handle_is_used(h)) { struct ldlm_lock *lock = ldlm_handle2lock(h); + struct ptlrpc_request *req = mdt_info_req(info); if (o != NULL && (lock->l_policy_data.l_inodebits.bits & (MDS_INODELOCK_XATTR | MDS_INODELOCK_UPDATE))) mo_invalidate(info->mti_env, mdt_object_child(o)); - if (decref || !info->mti_has_trans || + if (decref || !info->mti_has_trans || !req || !(mode & (LCK_PW | LCK_EX))) { ldlm_lock_decref_and_cancel(h, mode); LDLM_LOCK_PUT(lock); } else { - struct ptlrpc_request *req = mdt_info_req(info); - - LASSERT(req != NULL); tgt_save_slc_lock(&info->mti_mdt->mdt_lut, lock, req->rq_transno); ldlm_lock_decref(h, mode); @@ -3148,11 +3781,12 @@ void mdt_object_unlock_put(struct mdt_thread_info * info, * - create lu_object, corresponding to the fid in mdt_body, and save it in * @info; * - * - if HABEO_CORPUS flag is set for this request type check whether object + * - if HAS_BODY flag is set for this request type check whether object * actually exists on storage (lu_object_exists()). * */ -static int mdt_body_unpack(struct mdt_thread_info *info, __u32 flags) +static int mdt_body_unpack(struct mdt_thread_info *info, + enum tgt_handler_flags flags) { const struct mdt_body *body; struct mdt_object *obj; @@ -3178,7 +3812,7 @@ static int mdt_body_unpack(struct mdt_thread_info *info, __u32 flags) obj = mdt_object_find(env, info->mti_mdt, &body->mbo_fid1); if (!IS_ERR(obj)) { - if ((flags & HABEO_CORPUS) && !mdt_object_exists(obj)) { + if ((flags & HAS_BODY) && !mdt_object_exists(obj)) { mdt_object_put(env, obj); rc = -ENOENT; } else { @@ -3191,23 +3825,25 @@ static int mdt_body_unpack(struct mdt_thread_info *info, __u32 flags) RETURN(rc); } -static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags) +static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, + enum tgt_handler_flags flags) { - struct req_capsule *pill = info->mti_pill; - int rc; - ENTRY; + struct req_capsule *pill = info->mti_pill; + int rc; + + ENTRY; - if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_CLIENT)) - rc = mdt_body_unpack(info, flags); - else - rc = 0; + if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_CLIENT)) + rc = mdt_body_unpack(info, flags); + else + rc = 0; - if (rc == 0 && (flags & HABEO_REFERO)) { - /* Pack reply. */ - if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER)) - req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, + if (rc == 0 && (flags & HAS_REPLY)) { + /* Pack reply. */ + if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER)) + req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, DEF_REP_MD_SIZE); - if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER)) + if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER)) req_capsule_set_size(pill, &RMF_LOGCOOKIES, RCL_SERVER, 0); @@ -3218,9 +3854,14 @@ static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags) req_capsule_set_size(pill, &RMF_ACL, RCL_SERVER, LUSTRE_POSIX_ACL_MAX_SIZE_OLD); - rc = req_capsule_server_pack(pill); - } - RETURN(rc); + mdt_preset_secctx_size(info); + + rc = req_capsule_server_pack(pill); + if (rc) + CWARN("%s: cannot pack response: rc = %d\n", + mdt_obd_name(info->mti_mdt), rc); + } + RETURN(rc); } void mdt_lock_handle_init(struct mdt_lock_handle *lh) @@ -3280,7 +3921,6 @@ 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.sp_migrate_close = 0; info->mti_spec.u.sp_ea.eadata = NULL; info->mti_spec.u.sp_ea.eadatalen = 0; @@ -3300,6 +3940,7 @@ void mdt_thread_info_fini(struct mdt_thread_info *info) info->mti_env = NULL; info->mti_pill = NULL; info->mti_exp = NULL; + info->mti_mdt = NULL; if (unlikely(info->mti_big_buf.lb_buf != NULL)) lu_buf_free(&info->mti_big_buf); @@ -3327,56 +3968,20 @@ static int mdt_tgt_connect(struct tgt_session_info *tsi) { if (OBD_FAIL_CHECK(OBD_FAIL_TGT_DELAY_CONDITIONAL) && cfs_fail_val == - tsi2mdt_info(tsi)->mti_mdt->mdt_seq_site.ss_node_id) { - set_current_state(TASK_UNINTERRUPTIBLE); - schedule_timeout(msecs_to_jiffies(3 * MSEC_PER_SEC)); - } + tsi2mdt_info(tsi)->mti_mdt->mdt_seq_site.ss_node_id) + schedule_timeout_uninterruptible(cfs_time_seconds(3)); return tgt_connect(tsi); } -enum mdt_it_code { - MDT_IT_OPEN, - MDT_IT_OCREAT, - MDT_IT_CREATE, - MDT_IT_GETATTR, - MDT_IT_READDIR, - MDT_IT_LOOKUP, - MDT_IT_UNLINK, - MDT_IT_TRUNC, - MDT_IT_GETXATTR, - MDT_IT_LAYOUT, - MDT_IT_QUOTA, - MDT_IT_GLIMPSE, - MDT_IT_BRW, - MDT_IT_NR -}; - -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 **, - __u64); -static int mdt_intent_reint(enum mdt_it_code opcode, - struct mdt_thread_info *info, - struct ldlm_lock **, - __u64); -static int mdt_intent_glimpse(enum mdt_it_code opcode, +static int mdt_intent_glimpse(enum ldlm_intent_flags it_opc, struct mdt_thread_info *info, struct ldlm_lock **lockp, __u64 flags) { return mdt_glimpse_enqueue(info, info->mti_mdt->mdt_namespace, lockp, flags); } -static int mdt_intent_brw(enum mdt_it_code opcode, +static int mdt_intent_brw(enum ldlm_intent_flags it_opc, struct mdt_thread_info *info, struct ldlm_lock **lockp, __u64 flags) { @@ -3384,90 +3989,6 @@ static int mdt_intent_brw(enum mdt_it_code opcode, lockp, flags); } -static struct mdt_it_flavor { - const struct req_format *it_fmt; - __u32 it_flags; - int (*it_act)(enum mdt_it_code , - struct mdt_thread_info *, - struct ldlm_lock **, - __u64); - long it_reint; -} mdt_it_flavor[] = { - [MDT_IT_OPEN] = { - .it_fmt = &RQF_LDLM_INTENT, - /*.it_flags = HABEO_REFERO,*/ - .it_flags = 0, - .it_act = mdt_intent_reint, - .it_reint = REINT_OPEN - }, - [MDT_IT_OCREAT] = { - .it_fmt = &RQF_LDLM_INTENT, - /* - * 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 - }, - [MDT_IT_CREATE] = { - .it_fmt = &RQF_LDLM_INTENT, - .it_flags = MUTABOR, - .it_act = mdt_intent_reint, - .it_reint = REINT_CREATE - }, - [MDT_IT_GETATTR] = { - .it_fmt = &RQF_LDLM_INTENT_GETATTR, - .it_flags = HABEO_REFERO, - .it_act = mdt_intent_getattr - }, - [MDT_IT_READDIR] = { - .it_fmt = NULL, - .it_flags = 0, - .it_act = NULL - }, - [MDT_IT_LOOKUP] = { - .it_fmt = &RQF_LDLM_INTENT_GETATTR, - .it_flags = HABEO_REFERO, - .it_act = mdt_intent_getattr - }, - [MDT_IT_UNLINK] = { - .it_fmt = &RQF_LDLM_INTENT_UNLINK, - .it_flags = MUTABOR, - .it_act = NULL, - .it_reint = REINT_UNLINK - }, - [MDT_IT_TRUNC] = { - .it_fmt = NULL, - .it_flags = MUTABOR, - .it_act = NULL - }, - [MDT_IT_GETXATTR] = { - .it_fmt = &RQF_LDLM_INTENT_GETXATTR, - .it_flags = HABEO_CORPUS, - .it_act = mdt_intent_getxattr - }, - [MDT_IT_LAYOUT] = { - .it_fmt = &RQF_LDLM_INTENT_LAYOUT, - .it_flags = 0, - .it_act = mdt_intent_layout - }, - [MDT_IT_GLIMPSE] = { - .it_fmt = &RQF_LDLM_INTENT, - .it_flags = 0, - .it_act = mdt_intent_glimpse, - }, - [MDT_IT_BRW] = { - .it_fmt = &RQF_LDLM_INTENT, - .it_flags = 0, - .it_act = mdt_intent_brw, - }, - -}; - int mdt_intent_lock_replace(struct mdt_thread_info *info, struct ldlm_lock **lockp, struct mdt_lock_handle *lh, @@ -3593,7 +4114,7 @@ void mdt_intent_fixup_resent(struct mdt_thread_info *info, * If the xid matches, then we know this is a resent request, and allow * it. (It's probably an OPEN, for which we don't send a lock. */ - if (req_can_reconstruct(req, NULL)) + if (req_can_reconstruct(req, NULL) != 0) return; /* @@ -3607,10 +4128,10 @@ void mdt_intent_fixup_resent(struct mdt_thread_info *info, dlmreq->lock_handle[0].cookie); } -static int mdt_intent_getxattr(enum mdt_it_code opcode, - struct mdt_thread_info *info, - struct ldlm_lock **lockp, - __u64 flags) +static int mdt_intent_getxattr(enum ldlm_intent_flags it_opc, + 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; @@ -3639,7 +4160,10 @@ static int mdt_intent_getxattr(enum mdt_it_code opcode, if (ldlm_rep == NULL || OBD_FAIL_CHECK(OBD_FAIL_MDS_XATTR_REP)) { mdt_object_unlock(info, info->mti_object, lhc, 1); - RETURN(err_serious(-EFAULT)); + if (is_serious(rc)) + RETURN(rc); + else + RETURN(err_serious(-EFAULT)); } ldlm_rep->lock_policy_res2 = clear_serious(rc); @@ -3657,7 +4181,7 @@ static int mdt_intent_getxattr(enum mdt_it_code opcode, RETURN(rc); } -static int mdt_intent_getattr(enum mdt_it_code opcode, +static int mdt_intent_getattr(enum ldlm_intent_flags it_opc, struct mdt_thread_info *info, struct ldlm_lock **lockp, __u64 flags) @@ -3680,18 +4204,19 @@ static int mdt_intent_getattr(enum mdt_it_code opcode, repbody->mbo_eadatasize = 0; repbody->mbo_aclsize = 0; - switch (opcode) { - case MDT_IT_LOOKUP: + switch (it_opc) { + case IT_LOOKUP: child_bits = MDS_INODELOCK_LOOKUP | MDS_INODELOCK_PERM; - break; - case MDT_IT_GETATTR: + break; + case IT_GETATTR: child_bits = MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE | MDS_INODELOCK_PERM; - break; - default: - CERROR("Unsupported intent (%d)\n", opcode); - GOTO(out_shrink, rc = -EINVAL); - } + break; + default: + CERROR("%s: unsupported intent %#x\n", + mdt_obd_name(info->mti_mdt), (unsigned int)it_opc); + GOTO(out_shrink, rc = -EINVAL); + } rc = mdt_init_ucred_intent_getattr(info, reqbody); if (rc) @@ -3726,25 +4251,22 @@ out_shrink: return rc; } -static int mdt_intent_layout(enum mdt_it_code opcode, +static int mdt_intent_layout(enum ldlm_intent_flags it_opc, struct mdt_thread_info *info, struct ldlm_lock **lockp, __u64 flags) { - struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_LAYOUT]; + struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_RMT]; struct md_layout_change layout = { .mlc_opc = MD_LAYOUT_NOP }; struct layout_intent *intent; + struct ldlm_reply *ldlm_rep; struct lu_fid *fid = &info->mti_tmp_fid2; struct mdt_object *obj = NULL; int layout_size = 0; + struct lu_buf *buf = &layout.mlc_buf; int rc = 0; - ENTRY; - if (opcode != MDT_IT_LAYOUT) { - CERROR("%s: Unknown intent (%d)\n", mdt_obd_name(info->mti_mdt), - opcode); - RETURN(-EINVAL); - } + ENTRY; fid_extract_from_res_name(fid, &(*lockp)->l_resource->lr_name); @@ -3771,110 +4293,108 @@ static int mdt_intent_layout(enum mdt_it_code opcode, case LAYOUT_INTENT_RESTORE: CERROR("%s: Unsupported layout intent opc %d\n", mdt_obd_name(info->mti_mdt), intent->li_opc); - rc = -ENOTSUPP; - break; + RETURN(-ENOTSUPP); default: CERROR("%s: Unknown layout intent opc %d\n", mdt_obd_name(info->mti_mdt), intent->li_opc); - rc = -EINVAL; - break; + RETURN(-EINVAL); } - if (rc < 0) - RETURN(rc); - - /* Get lock from request for possible resent case. */ - mdt_intent_fixup_resent(info, *lockp, lhc, flags); obj = mdt_object_find(info->mti_env, info->mti_mdt, fid); if (IS_ERR(obj)) - GOTO(out, rc = PTR_ERR(obj)); + RETURN(PTR_ERR(obj)); if (mdt_object_exists(obj) && !mdt_object_remote(obj)) { - layout_size = mdt_attr_get_eabuf_size(info, obj); - if (layout_size < 0) - GOTO(out_obj, rc = layout_size); + /* if layout is going to be changed don't use the current EA + * size but the maximum one. That buffer will be shrinked + * to the actual size in req_capsule_shrink() before reply. + */ + if (layout.mlc_opc == MD_LAYOUT_WRITE) { + layout_size = info->mti_mdt->mdt_max_mdsize; + } else { + layout_size = mdt_attr_get_eabuf_size(info, obj); + if (layout_size < 0) + GOTO(out, rc = layout_size); - if (layout_size > info->mti_mdt->mdt_max_mdsize) - info->mti_mdt->mdt_max_mdsize = layout_size; + if (layout_size > info->mti_mdt->mdt_max_mdsize) + info->mti_mdt->mdt_max_mdsize = layout_size; + } + CDEBUG(D_INFO, "%s: layout_size %d\n", + mdt_obd_name(info->mti_mdt), layout_size); } /* * set reply buffer size, so that ldlm_handle_enqueue0()-> * ldlm_lvbo_fill() will fill the reply buffer with lovea. */ - (*lockp)->l_lvb_type = LVB_T_LAYOUT; req_capsule_set_size(info->mti_pill, &RMF_DLM_LVB, RCL_SERVER, layout_size); rc = req_capsule_server_pack(info->mti_pill); if (rc) - GOTO(out_obj, rc); + GOTO(out, rc); + ldlm_rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP); + if (!ldlm_rep) + GOTO(out, rc = -EPROTO); - if (layout.mlc_opc != MD_LAYOUT_NOP) { - struct lu_buf *buf = &layout.mlc_buf; + mdt_set_disposition(info, ldlm_rep, DISP_IT_EXECD); - /** - * mdt_layout_change is a reint operation, when the request - * is resent, layout write shouldn't reprocess it again. - */ - rc = mdt_check_resent(info, mdt_reconstruct_generic, lhc); - if (rc) - GOTO(out_obj, rc = rc < 0 ? rc : 0); + /* take lock in ldlm_lock_enqueue() for LAYOUT_INTENT_ACCESS */ + if (layout.mlc_opc == MD_LAYOUT_NOP) + GOTO(out, rc = 0); - /** - * There is another resent case: the client's job has been - * done by another client, referring lod_declare_layout_change - * -EALREADY case, and it became a operation w/o transaction, - * so we should not do the layout change, otherwise - * mdt_layout_change() will try to cancel the granted server - * CR lock whose remote counterpart is still in hold on the - * client, and a deadlock ensues. - */ - rc = mdt_check_resent_lock(info, obj, lhc); - if (rc <= 0) - GOTO(out_obj, rc); - - buf->lb_buf = NULL; - buf->lb_len = 0; - if (unlikely(req_is_replay(mdt_info_req(info)))) { - buf->lb_buf = req_capsule_client_get(info->mti_pill, - &RMF_EADATA); - buf->lb_len = req_capsule_get_size(info->mti_pill, - &RMF_EADATA, RCL_CLIENT); - /* - * If it's a replay of layout write intent RPC, the - * client has saved the extended lovea when - * it get reply then. - */ - if (buf->lb_len > 0) - mdt_fix_lov_magic(info, buf->lb_buf); - } + rc = mdt_check_resent(info, mdt_reconstruct_generic, lhc); + if (rc < 0) + GOTO(out, rc); + if (rc == 1) { + DEBUG_REQ(D_INODE, mdt_info_req(info), "resent opt."); + rc = lustre_msg_get_status(mdt_info_req(info)->rq_repmsg); + GOTO(out, rc); + } + buf->lb_buf = NULL; + buf->lb_len = 0; + if (unlikely(req_is_replay(mdt_info_req(info)))) { + buf->lb_buf = req_capsule_client_get(info->mti_pill, + &RMF_EADATA); + buf->lb_len = req_capsule_get_size(info->mti_pill, + &RMF_EADATA, RCL_CLIENT); /* - * Instantiate some layout components, if @buf contains - * lovea, then it's a replay of the layout intent write - * RPC. + * If it's a replay of layout write intent RPC, the client has + * saved the extended lovea when it get reply then. */ - rc = mdt_layout_change(info, obj, &layout); - if (rc) - GOTO(out_obj, rc); + if (buf->lb_len > 0) + mdt_fix_lov_magic(info, buf->lb_buf); } -out_obj: - mdt_object_put(info->mti_env, obj); - if (rc == 0 && lustre_handle_is_used(&lhc->mlh_reg_lh)) + /* Get lock from request for possible resent case. */ + mdt_intent_fixup_resent(info, *lockp, lhc, flags); + (*lockp)->l_lvb_type = LVB_T_LAYOUT; + + /* + * Instantiate some layout components, if @buf contains lovea, then it's + * a replay of the layout intent write RPC. + */ + rc = mdt_layout_change(info, obj, lhc, &layout); + ldlm_rep->lock_policy_res2 = clear_serious(rc); + + if (lustre_handle_is_used(&lhc->mlh_reg_lh)) { rc = mdt_intent_lock_replace(info, lockp, lhc, flags, rc); + if (rc == ELDLM_LOCK_REPLACED && + (*lockp)->l_granted_mode == LCK_EX) + ldlm_lock_mode_downgrade(*lockp, LCK_CR); + } + EXIT; out: - lhc->mlh_reg_lh.cookie = 0; - + mdt_object_put(info->mti_env, obj); return rc; } -static int mdt_intent_reint(enum mdt_it_code opcode, - struct mdt_thread_info *info, - struct ldlm_lock **lockp, - __u64 flags) +static int mdt_intent_open(enum ldlm_intent_flags it_opc, + 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 *rep = NULL; @@ -3892,22 +4412,20 @@ static int mdt_intent_reint(enum mdt_it_code opcode, if (opc < 0) RETURN(opc); - if (mdt_it_flavor[opcode].it_reint != opc) { - CERROR("Reint code %ld doesn't match intent: %d\n", - opc, opcode); - RETURN(err_serious(-EPROTO)); - } - /* Get lock from request for possible resent case. */ mdt_intent_fixup_resent(info, *lockp, lhc, flags); rc = mdt_reint_internal(info, lhc, opc); - /* Check whether the reply has been packed successfully. */ - if (mdt_info_req(info)->rq_repmsg != NULL) - rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP); - if (rep == NULL) - RETURN(err_serious(-EFAULT)); + /* Check whether the reply has been packed successfully. */ + if (mdt_info_req(info)->rq_repmsg != NULL) + rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP); + if (rep == NULL) { + if (is_serious(rc)) + RETURN(rc); + else + RETURN(err_serious(-EFAULT)); + } /* MDC expects this in any case */ if (rc != 0) @@ -3947,75 +4465,64 @@ static int mdt_intent_reint(enum mdt_it_code opcode, RETURN(ELDLM_LOCK_ABORTED); } -static int mdt_intent_code(enum ldlm_intent_flags itcode) -{ +static int mdt_intent_opc(enum ldlm_intent_flags it_opc, + struct mdt_thread_info *info, + struct ldlm_lock **lockp, + u64 flags /* LDLM_FL_* */) +{ + struct req_capsule *pill = info->mti_pill; + struct ptlrpc_request *req = mdt_info_req(info); + const struct req_format *it_format; + int (*it_handler)(enum ldlm_intent_flags, + struct mdt_thread_info *, + struct ldlm_lock **, + u64); + enum tgt_handler_flags it_handler_flags = 0; + struct ldlm_reply *rep; + bool check_mdt_object = false; int rc; + ENTRY; - switch (itcode) { + switch (it_opc) { case IT_OPEN: - rc = MDT_IT_OPEN; - break; case IT_OPEN|IT_CREAT: - rc = MDT_IT_OCREAT; - break; - case IT_CREAT: - rc = MDT_IT_CREATE; - break; - case IT_READDIR: - rc = MDT_IT_READDIR; + /* + * OCREAT is not a IS_MUTABLE request since the file may + * already exist. We do the extra check of + * OBD_CONNECT_RDONLY in mdt_reint_open() when we + * really need to create the object. + */ + it_format = &RQF_LDLM_INTENT; + it_handler = &mdt_intent_open; break; case IT_GETATTR: - rc = MDT_IT_GETATTR; - break; + check_mdt_object = true; + /* fallthrough */ case IT_LOOKUP: - rc = MDT_IT_LOOKUP; - break; - case IT_UNLINK: - rc = MDT_IT_UNLINK; - break; - case IT_TRUNC: - rc = MDT_IT_TRUNC; + it_format = &RQF_LDLM_INTENT_GETATTR; + it_handler = &mdt_intent_getattr; + it_handler_flags = HAS_REPLY; break; case IT_GETXATTR: - rc = MDT_IT_GETXATTR; + check_mdt_object = true; + it_format = &RQF_LDLM_INTENT_GETXATTR; + it_handler = &mdt_intent_getxattr; + it_handler_flags = HAS_BODY; break; case IT_LAYOUT: - rc = MDT_IT_LAYOUT; - break; - case IT_QUOTA_DQACQ: - case IT_QUOTA_CONN: - rc = MDT_IT_QUOTA; + it_format = &RQF_LDLM_INTENT_LAYOUT; + it_handler = &mdt_intent_layout; break; case IT_GLIMPSE: - rc = MDT_IT_GLIMPSE; + it_format = &RQF_LDLM_INTENT; + it_handler = &mdt_intent_glimpse; break; case IT_BRW: - rc = MDT_IT_BRW; - break; - default: - CERROR("Unknown intent opcode: 0x%08x\n", itcode); - rc = -EINVAL; + it_format = &RQF_LDLM_INTENT; + it_handler = &mdt_intent_brw; break; - } - return rc; -} - -static int mdt_intent_opc(enum ldlm_intent_flags itopc, - struct mdt_thread_info *info, - struct ldlm_lock **lockp, __u64 flags) -{ - 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; - - opc = mdt_intent_code(itopc); - if (opc < 0) - RETURN(-EINVAL); - - if (opc == MDT_IT_QUOTA) { + case IT_QUOTA_DQACQ: + case IT_QUOTA_CONN: { struct lu_device *qmt = info->mti_mdt->mdt_qmt_dev; if (qmt == NULL) @@ -4031,33 +4538,34 @@ static int mdt_intent_opc(enum ldlm_intent_flags itopc, flags); RETURN(rc); } + default: + CERROR("%s: unknown intent code %#x\n", + mdt_obd_name(info->mti_mdt), it_opc); + RETURN(-EPROTO); + } - flv = &mdt_it_flavor[opc]; - if (flv->it_fmt != NULL) - req_capsule_extend(pill, flv->it_fmt); + req_capsule_extend(pill, it_format); - rc = mdt_unpack_req_pack_rep(info, flv->it_flags); + rc = mdt_unpack_req_pack_rep(info, it_handler_flags); if (rc < 0) RETURN(rc); - if (flv->it_flags & MUTABOR && mdt_rdonly(req->rq_export)) - RETURN(-EROFS); + if (unlikely(info->mti_object == NULL && check_mdt_object)) + RETURN(-EPROTO); - if (flv->it_act != NULL) { - struct ldlm_reply *rep; + if (it_handler_flags & IS_MUTABLE && mdt_rdonly(req->rq_export)) + RETURN(-EROFS); - OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_INTENT_DELAY, 10); + OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_INTENT_DELAY, 10); - /* execute policy */ - rc = flv->it_act(opc, info, lockp, flags); + /* execute policy */ + rc = (*it_handler)(it_opc, info, lockp, flags); - /* Check whether the reply has been packed successfully. */ - 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); - } + /* Check whether the reply has been packed successfully. */ + 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); } RETURN(rc); @@ -4075,9 +4583,12 @@ static void mdt_ptlrpc_stats_update(struct ptlrpc_request *req, LDLM_GLIMPSE_ENQUEUE : LDLM_IBITS_ENQUEUE)); } -static int mdt_intent_policy(struct ldlm_namespace *ns, - struct ldlm_lock **lockp, void *req_cookie, - enum ldlm_mode mode, __u64 flags, void *data) +static int mdt_intent_policy(const struct lu_env *env, + struct ldlm_namespace *ns, + struct ldlm_lock **lockp, + void *req_cookie, + enum ldlm_mode mode, + __u64 flags, void *data) { struct tgt_session_info *tsi; struct mdt_thread_info *info; @@ -4091,7 +4602,7 @@ static int mdt_intent_policy(struct ldlm_namespace *ns, LASSERT(req != NULL); - tsi = tgt_ses_info(req->rq_svc_thread->t_env); + tsi = tgt_ses_info(env); info = tsi2mdt_info(tsi); LASSERT(info != NULL); @@ -4250,9 +4761,8 @@ out_free: */ static int mdt_seq_init_cli(const struct lu_env *env, struct mdt_device *mdt) { - struct seq_server_site *ss = mdt_seq_site(mdt); - int rc; - char *prefix; + struct seq_server_site *ss = mdt_seq_site(mdt); + char *prefix; ENTRY; /* check if this is adding the first MDC and controller is not yet @@ -4270,19 +4780,12 @@ static int mdt_seq_init_cli(const struct lu_env *env, struct mdt_device *mdt) /* Note: seq_client_fini will be called in seq_site_fini */ snprintf(prefix, MAX_OBD_NAME + 5, "ctl-%s", mdt_obd_name(mdt)); - rc = seq_client_init(ss->ss_client_seq, NULL, LUSTRE_SEQ_METADATA, - prefix, ss->ss_node_id == 0 ? ss->ss_control_seq : + seq_client_init(ss->ss_client_seq, NULL, LUSTRE_SEQ_METADATA, + prefix, ss->ss_node_id == 0 ? ss->ss_control_seq : NULL); OBD_FREE(prefix, MAX_OBD_NAME + 5); - if (rc != 0) { - OBD_FREE_PTR(ss->ss_client_seq); - ss->ss_client_seq = NULL; - RETURN(rc); - } - - rc = seq_server_set_cli(env, ss->ss_server_seq, ss->ss_client_seq); - RETURN(rc); + RETURN(seq_server_set_cli(env, ss->ss_server_seq, ss->ss_client_seq)); } static int mdt_seq_init(const struct lu_env *env, struct mdt_device *mdt) @@ -4735,9 +5238,12 @@ static int mdt_quota_init(const struct lu_env *env, struct mdt_device *mdt, mdt->mdt_qmt_dev = obd->obd_lu_dev; /* configure local quota objects */ - rc = mdt->mdt_qmt_dev->ld_ops->ldo_prepare(env, - &mdt->mdt_lu_dev, - mdt->mdt_qmt_dev); + if (OBD_FAIL_CHECK(OBD_FAIL_QUOTA_INIT)) + rc = -EBADF; + else + rc = mdt->mdt_qmt_dev->ld_ops->ldo_prepare(env, + &mdt->mdt_lu_dev, + mdt->mdt_qmt_dev); if (rc) GOTO(class_cleanup, rc); @@ -4757,6 +5263,7 @@ class_cleanup: if (rc) { class_manual_cleanup(obd); mdt->mdt_qmt_dev = NULL; + GOTO(lcfg_cleanup, rc); } class_detach: if (rc) @@ -4801,12 +5308,25 @@ static int mdt_tgt_getxattr(struct tgt_session_info *tsi) struct mdt_thread_info *info = tsi2mdt_info(tsi); int rc; + if (unlikely(info->mti_object == NULL)) + return -EPROTO; + rc = mdt_getxattr(info); mdt_thread_info_fini(info); return rc; } +static int mdt_llog_open(struct tgt_session_info *tsi) +{ + ENTRY; + + if (!mdt_is_rootadmin(tsi2mdt_info(tsi))) + RETURN(err_serious(-EACCES)); + + RETURN(tgt_llog_open(tsi)); +} + #define OBD_FAIL_OST_READ_NET OBD_FAIL_OST_BRW_NET #define OBD_FAIL_OST_WRITE_NET OBD_FAIL_OST_BRW_NET #define OST_BRW_READ OST_READ @@ -4820,44 +5340,48 @@ TGT_RPC_HANDLER(MDS_FIRST_OPC, 0, MDS_DISCONNECT, tgt_disconnect, &RQF_MDS_DISCONNECT, LUSTRE_OBD_VERSION), TGT_RPC_HANDLER(MDS_FIRST_OPC, - HABEO_REFERO, MDS_SET_INFO, mdt_set_info, - &RQF_OBD_SET_INFO, LUSTRE_MDS_VERSION), + HAS_REPLY, MDS_SET_INFO, mdt_set_info, + &RQF_MDT_SET_INFO, LUSTRE_MDS_VERSION), TGT_MDT_HDL(0, MDS_GET_INFO, mdt_get_info), -TGT_MDT_HDL(0 | HABEO_REFERO, MDS_GET_ROOT, mdt_get_root), -TGT_MDT_HDL(HABEO_CORPUS, MDS_GETATTR, mdt_getattr), -TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_GETATTR_NAME, +TGT_MDT_HDL(HAS_REPLY, MDS_GET_ROOT, mdt_get_root), +TGT_MDT_HDL(HAS_BODY, MDS_GETATTR, mdt_getattr), +TGT_MDT_HDL(HAS_BODY | HAS_REPLY, MDS_GETATTR_NAME, mdt_getattr_name), -TGT_MDT_HDL(HABEO_CORPUS, MDS_GETXATTR, mdt_tgt_getxattr), -TGT_MDT_HDL(0 | HABEO_REFERO, MDS_STATFS, mdt_statfs), -TGT_MDT_HDL(0 | MUTABOR, MDS_REINT, mdt_reint), -TGT_MDT_HDL(HABEO_CORPUS, MDS_CLOSE, mdt_close), -TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_READPAGE, mdt_readpage), -TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_SYNC, mdt_sync), +TGT_MDT_HDL(HAS_BODY, MDS_GETXATTR, mdt_tgt_getxattr), +TGT_MDT_HDL(HAS_REPLY, MDS_STATFS, mdt_statfs), +TGT_MDT_HDL(IS_MUTABLE, MDS_REINT, mdt_reint), +TGT_MDT_HDL(HAS_BODY, MDS_CLOSE, mdt_close), +TGT_MDT_HDL(HAS_BODY | HAS_REPLY, MDS_READPAGE, mdt_readpage), +TGT_MDT_HDL(HAS_BODY | HAS_REPLY, MDS_SYNC, mdt_sync), TGT_MDT_HDL(0, MDS_QUOTACTL, mdt_quotactl), -TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO | MUTABOR, MDS_HSM_PROGRESS, +TGT_MDT_HDL(HAS_BODY | HAS_REPLY | IS_MUTABLE, MDS_HSM_PROGRESS, mdt_hsm_progress), -TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO | MUTABOR, MDS_HSM_CT_REGISTER, +TGT_MDT_HDL(HAS_BODY | HAS_REPLY | IS_MUTABLE, MDS_HSM_CT_REGISTER, mdt_hsm_ct_register), -TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO | MUTABOR, MDS_HSM_CT_UNREGISTER, +TGT_MDT_HDL(HAS_BODY | HAS_REPLY | IS_MUTABLE, MDS_HSM_CT_UNREGISTER, mdt_hsm_ct_unregister), -TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_HSM_STATE_GET, +TGT_MDT_HDL(HAS_BODY | HAS_REPLY, MDS_HSM_STATE_GET, mdt_hsm_state_get), -TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO | MUTABOR, MDS_HSM_STATE_SET, +TGT_MDT_HDL(HAS_BODY | HAS_REPLY | IS_MUTABLE, MDS_HSM_STATE_SET, mdt_hsm_state_set), -TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_HSM_ACTION, mdt_hsm_action), -TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO, MDS_HSM_REQUEST, +TGT_MDT_HDL(HAS_BODY | HAS_REPLY, MDS_HSM_ACTION, mdt_hsm_action), +TGT_MDT_HDL(HAS_BODY | HAS_REPLY, MDS_HSM_REQUEST, mdt_hsm_request), -TGT_MDT_HDL(HABEO_CLAVIS | HABEO_CORPUS | HABEO_REFERO | MUTABOR, +TGT_MDT_HDL(HAS_KEY | HAS_BODY | HAS_REPLY | IS_MUTABLE, MDS_SWAP_LAYOUTS, mdt_swap_layouts), +TGT_MDT_HDL(IS_MUTABLE, MDS_RMFID, mdt_rmfid), }; static struct tgt_handler mdt_io_ops[] = { -TGT_OST_HDL(HABEO_CORPUS | HABEO_REFERO, OST_BRW_READ, tgt_brw_read), -TGT_OST_HDL(HABEO_CORPUS | MUTABOR, OST_BRW_WRITE, tgt_brw_write), -TGT_OST_HDL(HABEO_CORPUS | HABEO_REFERO | MUTABOR, - OST_PUNCH, mdt_punch_hdl), -TGT_OST_HDL(HABEO_CORPUS | HABEO_REFERO, OST_SYNC, mdt_data_sync), +TGT_OST_HDL_HP(HAS_BODY | HAS_REPLY, OST_BRW_READ, tgt_brw_read, + mdt_hp_brw), +TGT_OST_HDL_HP(HAS_BODY | IS_MUTABLE, OST_BRW_WRITE, tgt_brw_write, + mdt_hp_brw), +TGT_OST_HDL_HP(HAS_BODY | HAS_REPLY | IS_MUTABLE, + OST_PUNCH, mdt_punch_hdl, + mdt_hp_punch), +TGT_OST_HDL(HAS_BODY | HAS_REPLY, OST_SYNC, mdt_data_sync), }; static struct tgt_handler mdt_sec_ctx_ops[] = { @@ -4867,7 +5391,14 @@ TGT_SEC_HDL_VAR(0, SEC_CTX_FINI, mdt_sec_ctx_handle) }; static struct tgt_handler mdt_quota_ops[] = { -TGT_QUOTA_HDL(HABEO_REFERO, QUOTA_DQACQ, mdt_quota_dqacq), +TGT_QUOTA_HDL(HAS_REPLY, QUOTA_DQACQ, mdt_quota_dqacq), +}; + +static struct tgt_handler mdt_llog_handlers[] = { + TGT_LLOG_HDL(0, LLOG_ORIGIN_HANDLE_CREATE, mdt_llog_open), + TGT_LLOG_HDL(0, LLOG_ORIGIN_HANDLE_NEXT_BLOCK, tgt_llog_next_block), + TGT_LLOG_HDL(0, LLOG_ORIGIN_HANDLE_READ_HEADER, tgt_llog_read_header), + TGT_LLOG_HDL(0, LLOG_ORIGIN_HANDLE_PREV_BLOCK, tgt_llog_prev_block), }; static struct tgt_opc_slice mdt_common_slice[] = { @@ -4914,7 +5445,7 @@ static struct tgt_opc_slice mdt_common_slice[] = { { .tos_opc_start = LLOG_FIRST_OPC, .tos_opc_end = LLOG_LAST_OPC, - .tos_hs = tgt_llog_handlers + .tos_hs = mdt_llog_handlers }, { .tos_opc_start = LFSCK_FIRST_OPC, @@ -4944,11 +5475,13 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) next->md_ops->mdo_iocontrol(env, next, OBD_IOC_STOP_LFSCK, 0, &stop); mdt_stack_pre_fini(env, m, md2lu_dev(m->mdt_child)); + + mdt_restriper_stop(m); ping_evictor_stop(); /* Remove the HSM /proc entry so the coordinator cannot be - * restarted by a user while it's shutting down. */ - hsm_cdt_procfs_fini(m); + * restarted by a user while it's shutting down. + */ mdt_hsm_cdt_stop(m); mdt_llog_ctxt_unclone(env, m, LLOG_AGENT_ORIG_CTXT); @@ -4968,7 +5501,7 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) /* Calling the cleanup functions in the same order as in the mdt_init0 * error path */ - mdt_procfs_fini(m); + mdt_tunables_fini(m); target_recovery_fini(obd); upcall_cache_cleanup(m->mdt_identity_cache); @@ -4980,6 +5513,11 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) mdt_hsm_cdt_fini(m); + if (m->mdt_los != NULL) { + local_oid_storage_fini(env, m->mdt_los); + m->mdt_los = NULL; + } + if (m->mdt_namespace != NULL) { ldlm_namespace_free_post(m->mdt_namespace); d->ld_obd->obd_namespace = m->mdt_namespace = NULL; @@ -5023,6 +5561,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, struct seq_server_site *ss_site; const char *identity_upcall = "NONE"; struct md_device *next; + struct lu_fid fid; int rc; long node_id; mntopt_t mntopts; @@ -5047,7 +5586,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, obd = class_name2obd(dev); LASSERT(obd != NULL); - m->mdt_max_mdsize = MAX_MD_SIZE; /* 4 stripes */ + m->mdt_max_mdsize = MAX_MD_SIZE_OLD; m->mdt_opts.mo_evict_tgt_nids = 1; m->mdt_opts.mo_cos = MDT_COS_DEFAULT; @@ -5057,25 +5596,37 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, RETURN(-EFAULT); } else { lsi = s2lsi(lmi->lmi_sb); + LASSERT(lsi->lsi_lmd); /* CMD is supported only in IAM mode */ LASSERT(num); - node_id = simple_strtol(num, NULL, 10); + rc = kstrtol(num, 10, &node_id); + if (rc) + RETURN(rc); + obd->u.obt.obt_magic = OBT_MAGIC; - if (lsi->lsi_lmd != NULL && - lsi->lsi_lmd->lmd_flags & LMD_FLG_SKIP_LFSCK) + if (lsi->lsi_lmd->lmd_flags & LMD_FLG_SKIP_LFSCK) m->mdt_skip_lfsck = 1; } - /* DoM files get IO lock at open by default */ - m->mdt_opts.mo_dom_lock = 1; + /* DoM files get IO lock at open optionally by default */ + m->mdt_opts.mo_dom_lock = ALWAYS_DOM_LOCK_ON_OPEN; + /* DoM files are read at open and data is packed in the reply */ + m->mdt_opts.mo_dom_read_open = 1; m->mdt_squash.rsi_uid = 0; m->mdt_squash.rsi_gid = 0; INIT_LIST_HEAD(&m->mdt_squash.rsi_nosquash_nids); - init_rwsem(&m->mdt_squash.rsi_sem); + spin_lock_init(&m->mdt_squash.rsi_lock); spin_lock_init(&m->mdt_lock); - m->mdt_enable_remote_dir = 0; + m->mdt_enable_remote_dir = 1; + m->mdt_enable_striped_dir = 1; + m->mdt_enable_dir_migration = 1; + m->mdt_enable_dir_restripe = 0; + m->mdt_enable_dir_auto_split = 0; m->mdt_enable_remote_dir_gid = 0; + m->mdt_enable_chprojid_gid = 0; + m->mdt_enable_remote_rename = 1; + m->mdt_dir_restripe_nsonly = 1; atomic_set(&m->mdt_mds_mds_conns, 0); atomic_set(&m->mdt_async_commit_count, 0); @@ -5141,22 +5692,15 @@ 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; - 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 = 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_free_hsm, rc); + GOTO(err_free_ns, rc); /* Amount of available space excluded from granting and reserved - * for metadata. It is in percentage and 50% is default value. */ - tgd->tgd_reserved_pcnt = 50; + * for metadata. It is a percentage of the total MDT size. */ + tgd->tgd_reserved_pcnt = 10; if (ONE_MB_BRW_SIZE < (1U << tgd->tgd_blockbits)) m->mdt_brw_size = 1U << tgd->tgd_blockbits; @@ -5167,6 +5711,20 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, if (rc) GOTO(err_tgt, rc); + fid.f_seq = FID_SEQ_LOCAL_NAME; + fid.f_oid = 1; + fid.f_ver = 0; + rc = local_oid_storage_init(env, m->mdt_bottom, &fid, &m->mdt_los); + if (rc != 0) + GOTO(err_fs_cleanup, 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_los_fini, rc); + } + tgt_adapt_sptlrpc_conf(&m->mdt_lut); next = m->mdt_child; @@ -5197,10 +5755,10 @@ 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_fs_cleanup, rc); + GOTO(err_free_hsm, rc); } - rc = mdt_procfs_init(m, dev); + rc = mdt_tunables_init(m, dev); if (rc) { CERROR("Can't init MDT lprocfs, rc %d\n", rc); GOTO(err_recovery, rc); @@ -5226,19 +5784,36 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, if (ldlm_timeout == LDLM_TIMEOUT_DEFAULT) ldlm_timeout = MDS_LDLM_TIMEOUT_DEFAULT; + if ((lsi->lsi_lmd->lmd_flags & LMD_FLG_LOCAL_RECOV)) + m->mdt_lut.lut_local_recovery = 1; + + rc = mdt_restriper_start(m); + if (rc) + GOTO(err_ping_evictor, rc); + RETURN(0); + +err_ping_evictor: + ping_evictor_stop(); err_procfs: - mdt_procfs_fini(m); + mdt_tunables_fini(m); err_recovery: - target_recovery_fini(obd); upcall_cache_cleanup(m->mdt_identity_cache); m->mdt_identity_cache = NULL; +err_free_hsm: + mdt_hsm_cdt_fini(m); +err_los_fini: + local_oid_storage_fini(env, m->mdt_los); + m->mdt_los = NULL; err_fs_cleanup: mdt_fs_cleanup(env, m); err_tgt: + /* keep recoverable clients */ + obd->obd_fail = 1; + target_recovery_fini(obd); + obd_exports_barrier(obd); + obd_zombie_barrier(); tgt_fini(env, &m->mdt_lut); -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; @@ -5279,12 +5854,12 @@ static int mdt_process_config(const struct lu_env *env, switch (cfg->lcfg_command) { case LCFG_PARAM: { - struct obd_device *obd = d->ld_obd; - + struct obd_device *obd = d->ld_obd; /* For interoperability */ - struct cfg_interop_param *ptr = NULL; - struct lustre_cfg *old_cfg = NULL; - char *param = NULL; + struct cfg_interop_param *ptr = NULL; + struct lustre_cfg *old_cfg = NULL; + char *param = NULL; + ssize_t count; param = lustre_cfg_string(cfg, 1); if (param == NULL) { @@ -5313,17 +5888,22 @@ static int mdt_process_config(const struct lu_env *env, } } - rc = class_process_proc_param(PARAM_MDT, obd->obd_vars, - cfg, obd); - if (rc > 0 || rc == -ENOSYS) { + count = class_modify_config(cfg, PARAM_MDT, + &obd->obd_kset.kobj); + if (count < 0) { + struct coordinator *cdt = &m->mdt_coordinator; + /* is it an HSM var ? */ - rc = class_process_proc_param(PARAM_HSM, - hsm_cdt_get_proc_vars(), - cfg, obd); - if (rc > 0 || rc == -ENOSYS) + count = class_modify_config(cfg, PARAM_HSM, + &cdt->cdt_hsm_kobj); + if (count < 0) /* we don't understand; pass it on */ rc = next->ld_ops->ldo_process_config(env, next, cfg); + else + rc = count > 0 ? 0 : count; + } else { + rc = count > 0 ? 0 : count; } if (old_cfg) @@ -5359,10 +5939,13 @@ static struct lu_object *mdt_object_alloc(const struct lu_env *env, lu_object_add_top(h, o); o->lo_ops = &mdt_obj_ops; spin_lock_init(&mo->mot_write_lock); + mutex_init(&mo->mot_som_mutex); mutex_init(&mo->mot_lov_mutex); init_rwsem(&mo->mot_dom_sem); init_rwsem(&mo->mot_open_sem); atomic_set(&mo->mot_open_count, 0); + mo->mot_restripe_offset = 0; + INIT_LIST_HEAD(&mo->mot_restripe_linkage); RETURN(o); } RETURN(NULL); @@ -5390,22 +5973,31 @@ static int mdt_object_init(const struct lu_env *env, struct lu_object *o, RETURN(rc); } +static void mdt_object_free_rcu(struct rcu_head *head) +{ + struct mdt_object *mo = container_of(head, struct mdt_object, + mot_header.loh_rcu); + + kmem_cache_free(mdt_object_kmem, mo); +} + static void mdt_object_free(const struct lu_env *env, struct lu_object *o) { - struct mdt_object *mo = mdt_obj(o); - struct lu_object_header *h; - ENTRY; + struct mdt_object *mo = mdt_obj(o); + struct lu_object_header *h; + ENTRY; - h = o->lo_header; - CDEBUG(D_INFO, "object free, fid = "DFID"\n", - PFID(lu_object_fid(o))); + h = o->lo_header; + CDEBUG(D_INFO, "object free, fid = "DFID"\n", + PFID(lu_object_fid(o))); 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); + OBD_FREE_PRE(mo, sizeof(*mo), "slab-freed"); + call_rcu(&mo->mot_header.loh_rcu, mdt_object_free_rcu); EXIT; } @@ -5504,6 +6096,18 @@ static int mdt_obd_set_info_async(const struct lu_env *env, RETURN(0); } +static inline void mdt_enable_slc(struct mdt_device *mdt) +{ + if (mdt->mdt_lut.lut_sync_lock_cancel == SYNC_LOCK_CANCEL_NEVER) + mdt->mdt_lut.lut_sync_lock_cancel = SYNC_LOCK_CANCEL_BLOCKING; +} + +static inline void mdt_disable_slc(struct mdt_device *mdt) +{ + if (mdt->mdt_lut.lut_sync_lock_cancel == SYNC_LOCK_CANCEL_BLOCKING) + mdt->mdt_lut.lut_sync_lock_cancel = SYNC_LOCK_CANCEL_NEVER; +} + /** * Match client and server connection feature flags. * @@ -5534,6 +6138,7 @@ static int mdt_connect_internal(const struct lu_env *env, struct mdt_device *mdt, struct obd_connect_data *data, bool reconnect) { + const char *obd_name = mdt_obd_name(mdt); LASSERT(data != NULL); data->ocd_connect_flags &= MDT_CONNECT_SUPPORTED; @@ -5562,8 +6167,7 @@ static int mdt_connect_internal(const struct lu_env *env, "ocd_version: %x ocd_grant: %d ocd_index: %u " "ocd_brw_size unexpectedly zero, network data " "corruption? Refusing to connect this client\n", - mdt_obd_name(mdt), - exp->exp_client_uuid.uuid, + obd_name, exp->exp_client_uuid.uuid, exp, data->ocd_connect_flags, data->ocd_version, data->ocd_grant, data->ocd_index); return -EPROTO; @@ -5584,12 +6188,12 @@ static int mdt_connect_internal(const struct lu_env *env, data->ocd_grant_max_blks = ddp->ddp_max_extent_blks; } - if (OCD_HAS_FLAG(data, GRANT)) { - /* Save connect_data we have so far because tgt_grant_connect() - * uses it to calculate grant. */ - exp->exp_connect_data = *data; + /* Save connect_data we have so far because tgt_grant_connect() + * uses it to calculate grant, and we want to save the client + * version before it is overwritten by LUSTRE_VERSION_CODE. */ + exp->exp_connect_data = *data; + if (OCD_HAS_FLAG(data, GRANT)) tgt_grant_connect(env, exp, data, !reconnect); - } if (OCD_HAS_FLAG(data, MAXBYTES)) data->ocd_maxbytes = mdt->mdt_lut.lut_dt_conf.ddp_maxbytes; @@ -5609,7 +6213,7 @@ static int mdt_connect_internal(const struct lu_env *env, if ((data->ocd_connect_flags & OBD_CONNECT_FID) == 0) { CWARN("%s: MDS requires FID support, but client not\n", - mdt_obd_name(mdt)); + obd_name); return -EBADE; } @@ -5643,7 +6247,8 @@ static int mdt_connect_internal(const struct lu_env *env, /* The client set in ocd_cksum_types the checksum types it * supports. We have to mask off the algorithms that we don't * support */ - data->ocd_cksum_types &= cksum_types_supported_server(); + data->ocd_cksum_types &= + obd_cksum_types_supported_server(obd_name); if (unlikely(data->ocd_cksum_types == 0)) { CERROR("%s: Connect with checksum support but no " @@ -5663,6 +6268,12 @@ static int mdt_connect_internal(const struct lu_env *env, exp->exp_obd->obd_name, obd_export_nid2str(exp)); } + if ((data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) && + !(data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT)) { + atomic_inc(&mdt->mdt_mds_mds_conns); + mdt_enable_slc(mdt); + } + return 0; } @@ -5693,7 +6304,7 @@ static int mdt_ctxt_add_dirty_flag(struct lu_env *env, static int mdt_export_cleanup(struct obd_export *exp) { - struct list_head closing_list; + LIST_HEAD(closing_list); struct mdt_export_data *med = &exp->exp_mdt_data; struct obd_device *obd = exp->exp_obd; struct mdt_device *mdt; @@ -5703,7 +6314,6 @@ static int mdt_export_cleanup(struct obd_export *exp) int rc = 0; ENTRY; - INIT_LIST_HEAD(&closing_list); spin_lock(&med->med_open_lock); while (!list_empty(&med->med_open_head)) { struct list_head *tmp = med->med_open_head.next; @@ -5711,7 +6321,7 @@ static int mdt_export_cleanup(struct obd_export *exp) /* Remove mfd handle so it can't be found again. * We are consuming the mfd_list reference here. */ - class_handle_unhash(&mfd->mfd_handle); + class_handle_unhash(&mfd->mfd_open_handle); list_move_tail(&mfd->mfd_list, &closing_list); } spin_unlock(&med->med_open_lock); @@ -5752,7 +6362,7 @@ static int mdt_export_cleanup(struct obd_export *exp) * archive request into a noop if it's not actually * dirty. */ - if (mfd->mfd_mode & FMODE_WRITE) + if (mfd->mfd_open_flags & MDS_FMODE_WRITE) rc = mdt_ctxt_add_dirty_flag(&env, info, mfd); /* Don't unlink orphan on failover umount, LU-184 */ @@ -5760,31 +6370,20 @@ static int mdt_export_cleanup(struct obd_export *exp) ma->ma_valid = MA_FLAGS; ma->ma_attr_flags |= MDS_KEEP_ORPHAN; } - mdt_mfd_close(info, mfd); - } - } - info->mti_mdt = NULL; - /* cleanup client slot early */ - /* Do not erase record for recoverable client. */ - if (!(exp->exp_flags & OBD_OPT_FAILOVER) || exp->exp_failed) + ma->ma_valid |= MA_FORCE_LOG; + mdt_mfd_close(info, mfd); + } + } + info->mti_mdt = NULL; + /* cleanup client slot early */ + /* Do not erase record for recoverable client. */ + if (!(exp->exp_flags & OBD_OPT_FAILOVER) || exp->exp_failed) tgt_client_del(&env, exp); lu_env_fini(&env); RETURN(rc); } -static inline void mdt_enable_slc(struct mdt_device *mdt) -{ - if (mdt->mdt_lut.lut_sync_lock_cancel == NEVER_SYNC_ON_CANCEL) - mdt->mdt_lut.lut_sync_lock_cancel = BLOCKING_SYNC_ON_CANCEL; -} - -static inline void mdt_disable_slc(struct mdt_device *mdt) -{ - if (mdt->mdt_lut.lut_sync_lock_cancel == BLOCKING_SYNC_ON_CANCEL) - mdt->mdt_lut.lut_sync_lock_cancel = NEVER_SYNC_ON_CANCEL; -} - static int mdt_obd_disconnect(struct obd_export *exp) { int rc; @@ -5839,12 +6438,6 @@ static int mdt_obd_connect(const struct lu_env *env, mdt = mdt_dev(obd->obd_lu_dev); - if ((data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) && - !(data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT)) { - atomic_inc(&mdt->mdt_mds_mds_conns); - mdt_enable_slc(mdt); - } - /* * first, check whether the stack is ready to handle requests * XXX: probably not very appropriate method is used now @@ -5941,6 +6534,11 @@ static int mdt_init_export(struct obd_export *exp) exp->exp_connecting = 1; spin_unlock(&exp->exp_lock); + OBD_ALLOC(exp->exp_used_slots, + BITS_TO_LONGS(OBD_MAX_RIF_MAX) * sizeof(long)); + if (exp->exp_used_slots == NULL) + RETURN(-ENOMEM); + /* self-export doesn't need client data and ldlm initialization */ if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid, &exp->exp_client_uuid))) @@ -5959,6 +6557,10 @@ static int mdt_init_export(struct obd_export *exp) err_free: tgt_client_free(exp); err: + OBD_FREE(exp->exp_used_slots, + BITS_TO_LONGS(OBD_MAX_RIF_MAX) * sizeof(long)); + exp->exp_used_slots = NULL; + CERROR("%s: Failed to initialize export: rc = %d\n", exp->exp_obd->obd_name, rc); return rc; @@ -5969,6 +6571,10 @@ static int mdt_destroy_export(struct obd_export *exp) ENTRY; target_destroy_export(exp); + if (exp->exp_used_slots) + OBD_FREE(exp->exp_used_slots, + BITS_TO_LONGS(OBD_MAX_RIF_MAX) * sizeof(long)); + /* 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, @@ -6547,7 +7153,7 @@ static int mdt_obd_postrecov(struct obd_device *obd) return rc; } -static struct obd_ops mdt_obd_device_ops = { +static const struct obd_ops mdt_obd_device_ops = { .o_owner = THIS_MODULE, .o_set_info_async = mdt_obd_set_info_async, .o_connect = mdt_obd_connect, @@ -6651,12 +7257,12 @@ struct lu_ucred *mdt_ucred_check(const struct mdt_thread_info *info) * \param mdt mdt device * \param val 0 disables COS, other values enable COS */ -void mdt_enable_cos(struct mdt_device *mdt, int val) +void mdt_enable_cos(struct mdt_device *mdt, bool val) { struct lu_env env; int rc; - mdt->mdt_opts.mo_cos = !!val; + mdt->mdt_opts.mo_cos = val; rc = lu_env_init(&env, LCT_LOCAL); if (unlikely(rc != 0)) { CWARN("%s: lu_env initialization failed, cannot " @@ -6696,10 +7302,10 @@ static int __init mdt_init(void) { int rc; - CLASSERT(sizeof("0x0123456789ABCDEF:0x01234567:0x01234567") == - FID_NOBRACE_LEN + 1); - CLASSERT(sizeof("[0x0123456789ABCDEF:0x01234567:0x01234567]") == - FID_LEN + 1); + BUILD_BUG_ON(sizeof("0x0123456789ABCDEF:0x01234567:0x01234567") != + FID_NOBRACE_LEN + 1); + BUILD_BUG_ON(sizeof("[0x0123456789ABCDEF:0x01234567:0x01234567]") != + FID_LEN + 1); rc = lu_kmem_init(mdt_caches); if (rc) return rc;