X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_handler.c;h=e94ade4c17fe996bccb65efd844ef835d0e78b98;hp=e6e39bd4756e5ba77a678f57852b3ebe1522c6c0;hb=8da33c6cc5192303fcd18f45892e1f115004e662;hpb=35d0dbb7cf7184bb2f455b9b3a090b6c9c46d175 diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c index e6e39bd..e94ade4 100644 --- a/lustre/mdt/mdt_handler.c +++ b/lustre/mdt/mdt_handler.c @@ -26,10 +26,13 @@ * GPL HEADER END */ /* - * Copyright 2008 Sun Microsystems, Inc. All rights reserved + * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. */ /* + * Copyright (c) 2011 Whamcloud, Inc. + */ +/* * This file is part of Lustre, http://www.lustre.org/ * Lustre is a trademark of Sun Microsystems, Inc. * @@ -99,7 +102,9 @@ ldlm_mode_t mdt_dlm_lock_modes[] = { /* * Initialized in mdt_mod_init(). */ -unsigned long mdt_num_threads; +static unsigned long mdt_num_threads; +static unsigned long mdt_min_threads; +static unsigned long mdt_max_threads; /* ptlrpc request handler for MDT. All handlers are * grouped into several slices - struct mdt_opc_slice, @@ -339,6 +344,7 @@ static int mdt_getstatus(struct mdt_thread_info *info) static int mdt_statfs(struct mdt_thread_info *info) { + struct ptlrpc_request *req = mdt_info_req(info); struct md_device *next = info->mti_mdt->mdt_child; struct ptlrpc_service *svc; struct obd_statfs *osfs; @@ -365,36 +371,46 @@ static int mdt_statfs(struct mdt_thread_info *info) &info->mti_u.ksfs); statfs_pack(osfs, &info->mti_u.ksfs); } + + if (rc == 0) + mdt_counter_incr(req->rq_export, LPROC_MDT_STATFS); + RETURN(rc); } -void mdt_pack_size2body(struct mdt_thread_info *info, struct mdt_object *o) +/** + * Pack SOM attributes into the reply. + * Call under a DLM UPDATE lock. + */ +static void mdt_pack_size2body(struct mdt_thread_info *info, + struct mdt_object *mo) { struct mdt_body *b; - struct lu_attr *attr = &info->mti_attr.ma_attr; + struct md_attr *ma = &info->mti_attr; + LASSERT(ma->ma_attr.la_valid & LA_MODE); b = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); - /* Check if Size-on-MDS is enabled. */ - if ((mdt_conn_flags(info) & OBD_CONNECT_SOM) && - S_ISREG(attr->la_mode) && mdt_sizeonmds_enabled(o)) { - b->valid |= (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS); - b->size = attr->la_size; - b->blocks = attr->la_blocks; - } + /* Check if Size-on-MDS is supported, if this is a regular file, + * if SOM is enabled on the object and if SOM cache exists and valid. + * Otherwise do not pack Size-on-MDS attributes to the reply. */ + if (!(mdt_conn_flags(info) & OBD_CONNECT_SOM) || + !S_ISREG(ma->ma_attr.la_mode) || + !mdt_object_is_som_enabled(mo) || + !(ma->ma_valid & MA_SOM)) + return; + + b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; + b->size = ma->ma_som->msd_size; + b->blocks = ma->ma_som->msd_blocks; } void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b, const struct lu_attr *attr, const struct lu_fid *fid) { - /*XXX should pack the reply body according to lu_valid*/ - b->valid |= OBD_MD_FLCTIME | OBD_MD_FLUID | - OBD_MD_FLGID | OBD_MD_FLTYPE | - OBD_MD_FLMODE | OBD_MD_FLNLINK | OBD_MD_FLFLAGS | - OBD_MD_FLATIME | OBD_MD_FLMTIME ; + struct md_attr *ma = &info->mti_attr; - if (!S_ISREG(attr->la_mode)) - b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS | OBD_MD_FLRDEV; + LASSERT(ma->ma_valid & MA_INODE); b->atime = attr->la_atime; b->mtime = attr->la_mtime; @@ -408,6 +424,23 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b, b->nlink = attr->la_nlink; b->rdev = attr->la_rdev; + /*XXX should pack the reply body according to lu_valid*/ + b->valid |= OBD_MD_FLCTIME | OBD_MD_FLUID | + OBD_MD_FLGID | OBD_MD_FLTYPE | + OBD_MD_FLMODE | OBD_MD_FLNLINK | OBD_MD_FLFLAGS | + OBD_MD_FLATIME | OBD_MD_FLMTIME ; + + if (!S_ISREG(attr->la_mode)) { + b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS | OBD_MD_FLRDEV; + } else if (ma->ma_need & MA_LOV && ma->ma_lmm_size == 0) { + /* means no objects are allocated on osts. */ + LASSERT(!(ma->ma_valid & MA_LOV)); + /* just ignore blocks occupied by extend attributes on MDS */ + b->blocks = 0; + /* if no object is allocated on osts, the size on mds is valid. b=22272 */ + b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; + } + if (fid) { b->fid1 = *fid; b->valid |= OBD_MD_FLID; @@ -423,6 +456,10 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b, if (info) mdt_body_reverse_idmap(info, b); + + if (b->valid & OBD_MD_FLSIZE) + CDEBUG(D_VFSTRACE, DFID": returning size %llu\n", + PFID(fid), (unsigned long long)b->size); } static inline int mdt_body_has_lov(const struct lu_attr *la, @@ -433,7 +470,7 @@ static inline int mdt_body_has_lov(const struct lu_attr *la, } static int mdt_getattr_internal(struct mdt_thread_info *info, - struct mdt_object *o) + struct mdt_object *o, int ma_need) { struct md_object *next = mdt_object_child(o); const struct mdt_body *reqbody = info->mti_body; @@ -484,6 +521,10 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, /* get default stripe info for this dir. */ ma->ma_need |= MA_LOV_DEF; } + ma->ma_need |= ma_need; + if (ma->ma_need & MA_SOM) + ma->ma_som = &info->mti_u.som.data; + rc = mo_attr_get(env, next, ma); if (unlikely(rc)) { CERROR("getattr error for "DFID": %d\n", @@ -511,9 +552,6 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, repbody->eadatasize = ma->ma_lmv_size; repbody->valid |= (OBD_MD_FLDIREA|OBD_MD_MEA); } - if (!(ma->ma_valid & MA_LOV) && !(ma->ma_valid & MA_LMV)) { - repbody->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS; - } } else if (S_ISLNK(la->la_mode) && reqbody->valid & OBD_MD_LINKNAME) { buffer->lb_buf = ma->ma_lmm; @@ -637,6 +675,7 @@ static int mdt_renew_capa(struct mdt_thread_info *info) static int mdt_getattr(struct mdt_thread_info *info) { + struct ptlrpc_request *req = mdt_info_req(info); struct mdt_object *obj = info->mti_object; struct req_capsule *pill = info->mti_pill; struct mdt_body *reqbody; @@ -693,11 +732,14 @@ static int mdt_getattr(struct mdt_thread_info *info) * remote obj, and at that time no capability is available. */ mdt_set_capainfo(info, 1, &reqbody->fid1, BYPASS_CAPA); - rc = mdt_getattr_internal(info, obj); + rc = mdt_getattr_internal(info, obj, 0); if (reqbody->valid & OBD_MD_FLRMTPERM) mdt_exit_ucred(info); EXIT; out_shrink: + if (rc == 0) + mdt_counter_incr(req->rq_export, LPROC_MDT_GETATTR); + mdt_shrink_reply(info); return rc; } @@ -788,10 +830,11 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, struct lu_name *lname = NULL; const char *name = NULL; int namelen = 0; - struct mdt_lock_handle *lhp; + struct mdt_lock_handle *lhp = NULL; struct ldlm_lock *lock; struct ldlm_res_id *res_id; int is_resent; + int ma_need = 0; int rc; ENTRY; @@ -817,9 +860,13 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, if (namelen == 0) { reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY); - LASSERT(fid_is_sane(&reqbody->fid2)); - name = NULL; + if (unlikely(reqbody == NULL)) + RETURN(err_serious(-EFAULT)); + if (unlikely(!fid_is_sane(&reqbody->fid2))) + RETURN(err_serious(-EINVAL)); + + name = NULL; CDEBUG(D_INODE, "getattr with lock for "DFID"/"DFID", " "ldlm_rep = %p\n", PFID(mdt_object_fid(parent)), PFID(&reqbody->fid2), @@ -886,23 +933,22 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, /* Finally, we can get attr for child. */ mdt_set_capainfo(info, 0, mdt_object_fid(child), BYPASS_CAPA); - rc = mdt_getattr_internal(info, child); + rc = mdt_getattr_internal(info, child, 0); if (unlikely(rc != 0)) mdt_object_unlock(info, child, lhc, 1); } RETURN(rc); } - /* step 1: lock parent */ - lhp = &info->mti_lh[MDT_LH_PARENT]; - mdt_lock_pdo_init(lhp, LCK_PR, name, namelen); - rc = mdt_object_lock(info, parent, lhp, MDS_INODELOCK_UPDATE, - MDT_LOCAL_LOCK); - - if (unlikely(rc != 0)) - RETURN(rc); - if (lname) { + /* step 1: lock parent */ + lhp = &info->mti_lh[MDT_LH_PARENT]; + mdt_lock_pdo_init(lhp, LCK_PR, name, namelen); + rc = mdt_object_lock(info, parent, lhp, MDS_INODELOCK_UPDATE, + MDT_LOCAL_LOCK); + if (unlikely(rc != 0)) + RETURN(rc); + /* step 2: lookup child's fid by name */ rc = mdo_lookup(info->mti_env, next, lname, child_fid, &info->mti_spec); @@ -953,10 +999,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, LDLM_LOCK_PUT(lock); rc = 0; } else { - struct md_attr *ma; relock: - ma = &info->mti_attr; - OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RESEND, obd_timeout*2); mdt_lock_handle_init(lhc); mdt_lock_reg_init(lhc, LCK_PR); @@ -965,24 +1008,30 @@ relock: LU_OBJECT_DEBUG(D_WARNING, info->mti_env, &child->mot_obj.mo_lu, "Object doesn't exist!\n"); - GOTO(out_child, rc = -ESTALE); + GOTO(out_child, rc = -ENOENT); } - ma->ma_valid = 0; - ma->ma_need = MA_INODE; - rc = mo_attr_get(info->mti_env, next, ma); - if (unlikely(rc != 0)) - GOTO(out_child, rc); + if (!(child_bits & MDS_INODELOCK_UPDATE)) { + struct md_attr *ma = &info->mti_attr; - /* If the file has not been changed for some time, we return - * not only a LOOKUP lock, but also an UPDATE lock and this - * might save us RPC on later STAT. For directories, it also - * let negative dentry starts working for this dir. */ - if (ma->ma_valid & MA_INODE && - ma->ma_attr.la_valid & LA_CTIME && - info->mti_mdt->mdt_namespace->ns_ctime_age_limit + - ma->ma_attr.la_ctime < cfs_time_current_sec()) - child_bits |= MDS_INODELOCK_UPDATE; + ma->ma_valid = 0; + ma->ma_need = MA_INODE; + rc = mo_attr_get(info->mti_env, + mdt_object_child(child), ma); + if (unlikely(rc != 0)) + GOTO(out_child, rc); + + /* If the file has not been changed for some time, we + * return not only a LOOKUP lock, but also an UPDATE + * lock and this might save us RPC on later STAT. For + * directories, it also let negative dentry starts + * working for this dir. */ + if (ma->ma_valid & MA_INODE && + ma->ma_attr.la_valid & LA_CTIME && + info->mti_mdt->mdt_namespace->ns_ctime_age_limit + + ma->ma_attr.la_ctime < cfs_time_current_sec()) + child_bits |= MDS_INODELOCK_UPDATE; + } rc = mdt_object_lock(info, child, lhc, child_bits, MDT_CROSS_LOCK); @@ -991,43 +1040,40 @@ relock: GOTO(out_child, rc); } + lock = ldlm_handle2lock(&lhc->mlh_reg_lh); + /* Get MA_SOM attributes if update lock is given. */ + if (lock && + lock->l_policy_data.l_inodebits.bits & MDS_INODELOCK_UPDATE && + S_ISREG(lu_object_attr(&mdt_object_child(child)->mo_lu))) + ma_need = MA_SOM; + /* finally, we can get attr for child. */ mdt_set_capainfo(info, 1, child_fid, BYPASS_CAPA); - rc = mdt_getattr_internal(info, child); + rc = mdt_getattr_internal(info, child, ma_need); if (unlikely(rc != 0)) { mdt_object_unlock(info, child, lhc, 1); - } else { - lock = ldlm_handle2lock(&lhc->mlh_reg_lh); - if (lock) { - struct mdt_body *repbody; + } else if (lock) { + /* Debugging code. */ + res_id = &lock->l_resource->lr_name; + LDLM_DEBUG(lock, "Returning lock to client"); + LASSERTF(fid_res_name_eq(mdt_object_fid(child), + &lock->l_resource->lr_name), + "Lock res_id: %lu/%lu/%lu, Fid: "DFID".\n", + (unsigned long)res_id->name[0], + (unsigned long)res_id->name[1], + (unsigned long)res_id->name[2], + PFID(mdt_object_fid(child))); + mdt_pack_size2body(info, child); + } + if (lock) + LDLM_LOCK_PUT(lock); - /* Debugging code. */ - res_id = &lock->l_resource->lr_name; - LDLM_DEBUG(lock, "Returning lock to client"); - LASSERTF(fid_res_name_eq(mdt_object_fid(child), - &lock->l_resource->lr_name), - "Lock res_id: %lu/%lu/%lu, Fid: "DFID".\n", - (unsigned long)res_id->name[0], - (unsigned long)res_id->name[1], - (unsigned long)res_id->name[2], - PFID(mdt_object_fid(child))); - /* - * Pack Size-on-MDS inode attributes to the body if - * update lock is given. - */ - repbody = req_capsule_server_get(info->mti_pill, - &RMF_MDT_BODY); - if (lock->l_policy_data.l_inodebits.bits & - MDS_INODELOCK_UPDATE) - mdt_pack_size2body(info, child); - LDLM_LOCK_PUT(lock); - } - } EXIT; out_child: mdt_object_put(info->mti_env, child); out_parent: - mdt_object_unlock(info, parent, lhp, 1); + if (lhp) + mdt_object_unlock(info, parent, lhp, 1); return rc; } @@ -1111,12 +1157,12 @@ static int mdt_set_info(struct mdt_thread_info *info) req->rq_status = 0; lustre_msg_set_status(req->rq_repmsg, 0); - spin_lock(&req->rq_export->exp_lock); + cfs_spin_lock(&req->rq_export->exp_lock); if (*(__u32 *)val) req->rq_export->exp_connect_flags |= OBD_CONNECT_RDONLY; else req->rq_export->exp_connect_flags &=~OBD_CONNECT_RDONLY; - spin_unlock(&req->rq_export->exp_lock); + cfs_spin_unlock(&req->rq_export->exp_lock); } else if (KEY_IS(KEY_CHANGELOG_CLEAR)) { struct changelog_setinfo *cs = @@ -1173,7 +1219,7 @@ static int mdt_disconnect(struct mdt_thread_info *info) } static int mdt_sendpage(struct mdt_thread_info *info, - struct lu_rdpg *rdpg) + struct lu_rdpg *rdpg, int nob) { struct ptlrpc_request *req = mdt_info_req(info); struct obd_export *exp = req->rq_export; @@ -1181,7 +1227,6 @@ static int mdt_sendpage(struct mdt_thread_info *info, struct l_wait_info *lwi = &info->mti_u.rdpg.mti_wait_info; int tmpcount; int tmpsize; - int timeout; int i; int rc; ENTRY; @@ -1191,58 +1236,16 @@ static int mdt_sendpage(struct mdt_thread_info *info, if (desc == NULL) RETURN(-ENOMEM); - for (i = 0, tmpcount = rdpg->rp_count; - i < rdpg->rp_npages; i++, tmpcount -= tmpsize) { + for (i = 0, tmpcount = nob; + i < rdpg->rp_npages && tmpcount > 0; i++, tmpcount -= tmpsize) { tmpsize = min_t(int, tmpcount, CFS_PAGE_SIZE); ptlrpc_prep_bulk_page(desc, rdpg->rp_pages[i], 0, tmpsize); } - LASSERT(desc->bd_nob == rdpg->rp_count); - rc = sptlrpc_svc_wrap_bulk(req, desc); - if (rc) - GOTO(free_desc, rc); - - rc = ptlrpc_start_bulk_transfer(desc); - if (rc) - GOTO(free_desc, rc); - - if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SENDPAGE)) - GOTO(abort_bulk, rc = 0); - - timeout = (int) req->rq_deadline - cfs_time_current_sec(); - if (timeout < 0) - CERROR("Req deadline already passed %lu (now: %lu)\n", - req->rq_deadline, cfs_time_current_sec()); - *lwi = LWI_TIMEOUT_INTERVAL(cfs_time_seconds(max(timeout, 1)), - cfs_time_seconds(1), NULL, NULL); - rc = l_wait_event(desc->bd_waitq, !ptlrpc_server_bulk_active(desc) || - exp->exp_failed || exp->exp_abort_active_req, lwi); - LASSERT (rc == 0 || rc == -ETIMEDOUT); - - if (rc == 0) { - if (desc->bd_success && - desc->bd_nob_transferred == rdpg->rp_count) - GOTO(free_desc, rc); - - rc = -ETIMEDOUT; - if (exp->exp_abort_active_req || exp->exp_failed) - GOTO(abort_bulk, rc); - } - - DEBUG_REQ(D_ERROR, req, "bulk failed: %s %d(%d), evicting %s@%s", - (rc == -ETIMEDOUT) ? "timeout" : "network error", - desc->bd_nob_transferred, rdpg->rp_count, - exp->exp_client_uuid.uuid, - exp->exp_connection->c_remote_uuid.uuid); - - class_fail_export(exp); - - EXIT; -abort_bulk: - ptlrpc_abort_bulk(desc); -free_desc: + LASSERT(desc->bd_nob == nob); + rc = target_bulk_io(exp, desc, lwi); ptlrpc_free_bulk(desc); - return rc; + RETURN(rc); } #ifdef HAVE_SPLIT_SUPPORT @@ -1381,7 +1384,7 @@ static int mdt_writepage(struct mdt_thread_info *info) else rc = ptlrpc_start_bulk_transfer (desc); if (rc == 0) { - *lwi = LWI_TIMEOUT_INTERVAL(obd_timeout * HZ / 4, HZ, + *lwi = LWI_TIMEOUT_INTERVAL(obd_timeout * CFS_HZ / 4, CFS_HZ, mdt_bulk_timeout, desc); rc = l_wait_event(desc->bd_waitq, !ptlrpc_bulk_active(desc) || desc->bd_export->exp_failed, lwi); @@ -1450,8 +1453,12 @@ static int mdt_readpage(struct mdt_thread_info *info) } rdpg->rp_attrs = reqbody->mode; - rdpg->rp_count = reqbody->nlink; - rdpg->rp_npages = (rdpg->rp_count + CFS_PAGE_SIZE - 1)>>CFS_PAGE_SHIFT; + if (info->mti_exp->exp_connect_flags & OBD_CONNECT_64BITHASH) + rdpg->rp_attrs |= LUDA_64BITHASH; + rdpg->rp_count = min_t(unsigned int, reqbody->nlink, + PTLRPC_MAX_BRW_SIZE); + rdpg->rp_npages = (rdpg->rp_count + CFS_PAGE_SIZE - 1) >> + CFS_PAGE_SHIFT; OBD_ALLOC(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]); if (rdpg->rp_pages == NULL) RETURN(-ENOMEM); @@ -1464,11 +1471,11 @@ static int mdt_readpage(struct mdt_thread_info *info) /* call lower layers to fill allocated pages with directory data */ rc = mo_readpage(info->mti_env, mdt_object_child(object), rdpg); - if (rc) + if (rc < 0) GOTO(free_rdpg, rc); /* send pages to client */ - rc = mdt_sendpage(info, rdpg); + rc = mdt_sendpage(info, rdpg, rc); EXIT; free_rdpg: @@ -1643,6 +1650,7 @@ static int mdt_object_sync(struct mdt_thread_info *info) static int mdt_sync(struct mdt_thread_info *info) { + struct ptlrpc_request *req = mdt_info_req(info); struct req_capsule *pill = info->mti_pill; struct mdt_body *body; int rc; @@ -1690,6 +1698,9 @@ static int mdt_sync(struct mdt_thread_info *info) } else rc = err_serious(rc); } + if (rc == 0) + mdt_counter_incr(req->rq_export, LPROC_MDT_SYNC); + RETURN(rc); } @@ -1955,7 +1966,6 @@ static struct ldlm_callback_suite cbs = { static int mdt_enqueue(struct mdt_thread_info *info) { struct ptlrpc_request *req; - __u64 req_bits; int rc; /* @@ -1965,17 +1975,6 @@ static int mdt_enqueue(struct mdt_thread_info *info) LASSERT(info->mti_dlm_req != NULL); req = mdt_info_req(info); - - /* - * Lock without inodebits makes no sense and will oops later in - * ldlm. Let's check it now to see if we have wrong lock from client or - * bits get corrupted somewhere in mdt_intent_policy(). - */ - req_bits = info->mti_dlm_req->lock_desc.l_policy_data.l_inodebits.bits; - /* This is disabled because we need to support liblustre flock. - * LASSERT(req_bits != 0); - */ - rc = ldlm_handle_enqueue0(info->mti_mdt->mdt_namespace, req, info->mti_dlm_req, &cbs); info->mti_fail_id = OBD_FAIL_LDLM_REPLY; @@ -2025,7 +2024,7 @@ static int mdt_sec_ctx_handle(struct mdt_thread_info *info) sptlrpc_svc_ctx_invalidate(req); } - OBD_FAIL_TIMEOUT(OBD_FAIL_SEC_CTX_HDL_PAUSE, obd_fail_val); + CFS_FAIL_TIMEOUT(OBD_FAIL_SEC_CTX_HDL_PAUSE, cfs_fail_val); return rc; } @@ -2118,7 +2117,7 @@ static inline int mdt_is_lock_sync(struct ldlm_lock *lock) int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, void *data, int flag) { - struct obd_device *obd = lock->l_resource->lr_namespace->ns_obd; + struct obd_device *obd = ldlm_lock_to_ns(lock)->ns_obd; struct mdt_device *mdt = mdt_dev(obd->obd_lu_dev); int rc; ENTRY; @@ -2189,7 +2188,8 @@ int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o, RETURN(-ESTALE); } else { /* Non-dir object shouldn't have PDO lock */ - LASSERT(S_ISDIR(lu_object_attr(&o->mot_obj.mo_lu))); + if (!S_ISDIR(lu_object_attr(&o->mot_obj.mo_lu))) + RETURN(-ENOTDIR); } } @@ -2235,12 +2235,12 @@ int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o, res_id, LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB, &info->mti_exp->exp_handle.h_cookie); if (rc) - GOTO(out, rc); - -out: - if (rc) mdt_object_unlock(info, o, lh, 1); - + else if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_PDO_LOCK)) && + lh->mlh_pdo_hash != 0 && + (lh->mlh_reg_mode == LCK_PW || lh->mlh_reg_mode == LCK_EX)) { + OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PDO_LOCK, 10); + } RETURN(rc); } @@ -2545,10 +2545,25 @@ static int mdt_req_handle(struct mdt_thread_info *info, dlm_req = req_capsule_client_get(info->mti_pill, &RMF_DLM_REQ); if (dlm_req != NULL) { - if (info->mti_mdt->mdt_opts.mo_compat_resname) - rc = mdt_lock_resname_compat(info->mti_mdt, - dlm_req); - info->mti_dlm_req = dlm_req; + if (unlikely(dlm_req->lock_desc.l_resource.lr_type == + LDLM_IBITS && + dlm_req->lock_desc.l_policy_data.\ + l_inodebits.bits == 0)) { + /* + * Lock without inodebits makes no sense and + * will oops later in ldlm. If client miss to + * set such bits, do not trigger ASSERTION. + * + * For liblustre flock case, it maybe zero. + */ + rc = -EPROTO; + } else { + if (info->mti_mdt->mdt_opts.mo_compat_resname) + rc = mdt_lock_resname_compat( + info->mti_mdt, + dlm_req); + info->mti_dlm_req = dlm_req; + } } else { rc = -EFAULT; } @@ -2605,7 +2620,7 @@ static int mdt_req_handle(struct mdt_thread_info *info, if (likely(rc == 0 && req->rq_export && h->mh_opc != MDS_DISCONNECT)) target_committed_to_req(req); - if (unlikely((lustre_msg_get_flags(req->rq_reqmsg) & MSG_REPLAY) && + if (unlikely(req_is_replay(req) && lustre_msg_get_transno(req->rq_reqmsg) == 0)) { DEBUG_REQ(D_ERROR, req, "transno is 0 during REPLAY"); LBUG(); @@ -2669,10 +2684,7 @@ static void mdt_thread_info_init(struct ptlrpc_request *req, info->mti_fail_id = OBD_FAIL_MDS_ALL_REPLY_NET; info->mti_transno = lustre_msg_get_transno(req->rq_reqmsg); - info->mti_mos[0] = NULL; - info->mti_mos[1] = NULL; - info->mti_mos[2] = NULL; - info->mti_mos[3] = NULL; + info->mti_mos = NULL; memset(&info->mti_attr, 0, sizeof(info->mti_attr)); info->mti_body = NULL; @@ -2745,7 +2757,6 @@ static int mdt_filter_recovery_request(struct ptlrpc_request *req, static int mdt_recovery(struct mdt_thread_info *info) { struct ptlrpc_request *req = mdt_info_req(info); - int recovering; struct obd_device *obd; ENTRY; @@ -2768,7 +2779,7 @@ static int mdt_recovery(struct mdt_thread_info *info) } } - if (unlikely(req->rq_export == NULL)) { + if (unlikely(!class_connected_export(req->rq_export))) { CERROR("operation %d on unconnected MDS from %s\n", lustre_msg_get_opc(req->rq_reqmsg), libcfs_id2str(req->rq_peer)); @@ -2806,10 +2817,7 @@ static int mdt_recovery(struct mdt_thread_info *info) obd = req->rq_export->exp_obd; /* Check for aborted recovery... */ - spin_lock_bh(&obd->obd_processing_task_lock); - recovering = obd->obd_recovering; - spin_unlock_bh(&obd->obd_processing_task_lock); - if (unlikely(recovering)) { + if (unlikely(obd->obd_recovering)) { int rc; int should_process; DEBUG_REQ(D_INFO, req, "Got new replay"); @@ -2832,7 +2840,6 @@ static int mdt_msg_check_version(struct lustre_msg *msg) switch (lustre_msg_get_opc(msg)) { case MDS_CONNECT: case MDS_DISCONNECT: - case MDS_SET_INFO: case OBD_PING: case SEC_CTX_INIT: case SEC_CTX_INIT_CONT: @@ -2858,6 +2865,7 @@ static int mdt_msg_check_version(struct lustre_msg *msg) case MDS_SYNC: case MDS_GETXATTR: case MDS_SETXATTR: + case MDS_SET_INFO: case MDS_GET_INFO: case MDS_QUOTACHECK: case MDS_QUOTACTL: @@ -3180,7 +3188,7 @@ int mdt_intent_lock_replace(struct mdt_thread_info *info, new_lock->l_writers--; } - new_lock->l_export = class_export_lock_get(req->rq_export); + new_lock->l_export = class_export_lock_get(req->rq_export, new_lock); new_lock->l_blocking_ast = lock->l_blocking_ast; new_lock->l_completion_ast = lock->l_completion_ast; new_lock->l_remote_handle = lock->l_remote_handle; @@ -3188,9 +3196,9 @@ int mdt_intent_lock_replace(struct mdt_thread_info *info, unlock_res_and_lock(new_lock); - lustre_hash_add(new_lock->l_export->exp_lock_hash, - &new_lock->l_remote_handle, - &new_lock->l_exp_hash); + cfs_hash_add(new_lock->l_export->exp_lock_hash, + &new_lock->l_remote_handle, + &new_lock->l_exp_hash); LDLM_LOCK_RELEASE(new_lock); lh->mlh_reg_lh.cookie = 0; @@ -3215,7 +3223,7 @@ static void mdt_intent_fixup_resent(struct mdt_thread_info *info, dlmreq = req_capsule_client_get(info->mti_pill, &RMF_DLM_REQ); remote_hdl = dlmreq->lock_handle[0]; - lock = lustre_hash_lookup(exp->exp_lock_hash, &remote_hdl); + lock = cfs_hash_lookup(exp->exp_lock_hash, &remote_hdl); if (lock) { if (lock != new_lock) { lh->mlh_reg_lh.cookie = lock->l_handle.h_cookie; @@ -3227,11 +3235,11 @@ static void mdt_intent_fixup_resent(struct mdt_thread_info *info, lh->mlh_reg_lh.cookie); if (old_lock) *old_lock = LDLM_LOCK_GET(lock); - lh_put(exp->exp_lock_hash, &lock->l_exp_hash); + cfs_hash_put(exp->exp_lock_hash, &lock->l_exp_hash); return; } - lh_put(exp->exp_lock_hash, &lock->l_exp_hash); + cfs_hash_put(exp->exp_lock_hash, &lock->l_exp_hash); } /* @@ -3372,6 +3380,10 @@ static int mdt_intent_reint(enum mdt_it_code opcode, } rep->lock_policy_res2 = clear_serious(rc); + if (rep->lock_policy_res2 == -ENOENT && + mdt_get_disposition(rep, DISP_LOOKUP_NEG)) + rep->lock_policy_res2 = 0; + if (rc == -ENOTCONN || rc == -ENODEV || rc == -EOVERFLOW) { /**< if VBR failure then return error */ /* @@ -3392,6 +3404,8 @@ static int mdt_intent_reint(enum mdt_it_code opcode, * checked here. */ if (lustre_handle_is_used(&lhc->mlh_reg_lh)) { + LASSERTF(rc == 0, "Error occurred but lock handle " + "is still in use\n"); rep->lock_policy_res2 = 0; rc = mdt_intent_lock_replace(info, lockp, NULL, lhc, flags); RETURN(rc); @@ -3501,23 +3515,19 @@ static int mdt_intent_policy(struct ldlm_namespace *ns, req_capsule_extend(pill, &RQF_LDLM_INTENT); it = req_capsule_client_get(pill, &RMF_LDLM_INTENT); if (it != NULL) { - const struct ldlm_request *dlmreq; - __u64 req_bits; - rc = mdt_intent_opc(it->opc, info, lockp, flags); if (rc == 0) rc = ELDLM_OK; - /* - * Lock without inodebits makes no sense and will oops + /* Lock without inodebits makes no sense and will oops * later in ldlm. Let's check it now to see if we have - * wrong lock from client or bits get corrupted - * somewhere in mdt_intent_opc(). - */ - dlmreq = info->mti_dlm_req; - req_bits = dlmreq->lock_desc.l_policy_data.l_inodebits.bits; - LASSERT(req_bits != 0); - + * ibits corrupted somewhere in mdt_intent_opc(). + * The case for client miss to set ibits has been + * processed by others. */ + LASSERT(ergo(info->mti_dlm_req->lock_desc.l_resource.\ + lr_type == LDLM_IBITS, + info->mti_dlm_req->lock_desc.\ + l_policy_data.l_inodebits.bits != 0)); } else rc = err_serious(-EFAULT); } else { @@ -3530,25 +3540,6 @@ static int mdt_intent_policy(struct ldlm_namespace *ns, RETURN(rc); } -/* - * Seq wrappers - */ -static void mdt_seq_adjust(const struct lu_env *env, - struct mdt_device *m, int lost) -{ - struct md_site *ms = mdt_md_site(m); - struct lu_seq_range out; - ENTRY; - - LASSERT(ms && ms->ms_server_seq); - LASSERT(lost >= 0); - /* get extra seq from seq_server, moving it's range up */ - while (lost-- > 0) { - seq_server_alloc_meta(ms->ms_server_seq, NULL, &out, env); - } - EXIT; -} - static int mdt_seq_fini(const struct lu_env *env, struct mdt_device *m) { @@ -3871,9 +3862,8 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) * We'd like to have a mechanism to set this on a per-device * basis, but alas... */ - .psc_min_threads = min(max(mdt_num_threads, MDT_MIN_THREADS), - MDT_MAX_THREADS), - .psc_max_threads = MDT_MAX_THREADS, + .psc_min_threads = mdt_min_threads, + .psc_max_threads = mdt_max_threads, .psc_ctx_tags = LCT_MD_THREAD }; @@ -3888,7 +3878,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) if (m->mdt_regular_service == NULL) RETURN(-ENOMEM); - rc = ptlrpc_start_threads(NULL, m->mdt_regular_service); + rc = ptlrpc_start_threads(m->mdt_regular_service); if (rc) GOTO(err_mdt_svc, rc); @@ -3904,9 +3894,8 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) .psc_req_portal = MDS_READPAGE_PORTAL, .psc_rep_portal = MDC_REPLY_PORTAL, .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = min(max(mdt_num_threads, MDT_MIN_THREADS), - MDT_MAX_THREADS), - .psc_max_threads = MDT_MAX_THREADS, + .psc_min_threads = mdt_min_threads, + .psc_max_threads = mdt_max_threads, .psc_ctx_tags = LCT_MD_THREAD }; m->mdt_readpage_service = @@ -3919,7 +3908,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_readpage_service); + rc = ptlrpc_start_threads(m->mdt_readpage_service); /* * setattr service configuration. @@ -3932,9 +3921,8 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) .psc_req_portal = MDS_SETATTR_PORTAL, .psc_rep_portal = MDC_REPLY_PORTAL, .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = min(max(mdt_num_threads, MDT_MIN_THREADS), - MDT_MAX_THREADS), - .psc_max_threads = MDT_MAX_THREADS, + .psc_min_threads = mdt_min_threads, + .psc_max_threads = mdt_max_threads, .psc_ctx_tags = LCT_MD_THREAD }; @@ -3948,7 +3936,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_setattr_service); + rc = ptlrpc_start_threads(m->mdt_setattr_service); if (rc) GOTO(err_mdt_svc, rc); @@ -3963,8 +3951,8 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) .psc_req_portal = SEQ_CONTROLLER_PORTAL, .psc_rep_portal = MDC_REPLY_PORTAL, .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = SEQ_NUM_THREADS, - .psc_max_threads = SEQ_NUM_THREADS, + .psc_min_threads = mdt_min_threads, + .psc_max_threads = mdt_max_threads, .psc_ctx_tags = LCT_MD_THREAD|LCT_DT_THREAD }; @@ -3977,7 +3965,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_mdsc_service); + rc = ptlrpc_start_threads(m->mdt_mdsc_service); if (rc) GOTO(err_mdt_svc, rc); @@ -3992,8 +3980,8 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) .psc_req_portal = SEQ_METADATA_PORTAL, .psc_rep_portal = MDC_REPLY_PORTAL, .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = SEQ_NUM_THREADS, - .psc_max_threads = SEQ_NUM_THREADS, + .psc_min_threads = mdt_min_threads, + .psc_max_threads = mdt_max_threads, .psc_ctx_tags = LCT_MD_THREAD|LCT_DT_THREAD }; @@ -4006,7 +3994,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_mdss_service); + rc = ptlrpc_start_threads(m->mdt_mdss_service); if (rc) GOTO(err_mdt_svc, rc); @@ -4024,8 +4012,8 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) .psc_req_portal = SEQ_DATA_PORTAL, .psc_rep_portal = OSC_REPLY_PORTAL, .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = SEQ_NUM_THREADS, - .psc_max_threads = SEQ_NUM_THREADS, + .psc_min_threads = mdt_min_threads, + .psc_max_threads = mdt_max_threads, .psc_ctx_tags = LCT_MD_THREAD|LCT_DT_THREAD }; @@ -4038,7 +4026,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_dtss_service); + rc = ptlrpc_start_threads(m->mdt_dtss_service); if (rc) GOTO(err_mdt_svc, rc); @@ -4051,8 +4039,8 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) .psc_req_portal = FLD_REQUEST_PORTAL, .psc_rep_portal = MDC_REPLY_PORTAL, .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = FLD_NUM_THREADS, - .psc_max_threads = FLD_NUM_THREADS, + .psc_min_threads = mdt_min_threads, + .psc_max_threads = mdt_max_threads, .psc_ctx_tags = LCT_DT_THREAD|LCT_MD_THREAD }; @@ -4065,7 +4053,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_fld_service); + rc = ptlrpc_start_threads(m->mdt_fld_service); if (rc) GOTO(err_mdt_svc, rc); @@ -4081,9 +4069,8 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) .psc_req_portal = MDS_MDS_PORTAL, .psc_rep_portal = MDC_REPLY_PORTAL, .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = min(max(mdt_num_threads, MDT_MIN_THREADS), - MDT_MAX_THREADS), - .psc_max_threads = MDT_MAX_THREADS, + .psc_min_threads = mdt_min_threads, + .psc_max_threads = mdt_max_threads, .psc_ctx_tags = LCT_MD_THREAD }; m->mdt_xmds_service = @@ -4096,7 +4083,7 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m) GOTO(err_mdt_svc, rc = -ENOMEM); } - rc = ptlrpc_start_threads(NULL, m->mdt_xmds_service); + rc = ptlrpc_start_threads(m->mdt_xmds_service); if (rc) GOTO(err_mdt_svc, rc); @@ -4367,15 +4354,9 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) */ mdt_stack_fini(env, m, md2lu_dev(m->mdt_child)); - mdt_procfs_fini(m); - if (obd->obd_proc_exports_entry) { - lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry); - obd->obd_proc_exports_entry = NULL; - } lprocfs_free_per_client_stats(obd); lprocfs_free_obd_stats(obd); - ptlrpc_lprocfs_unregister_obd(obd); - lprocfs_obd_cleanup(obd); + mdt_procfs_fini(m); if (ls) { struct md_site *mite; @@ -4385,7 +4366,7 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) OBD_FREE_PTR(mite); d->ld_site = NULL; } - LASSERT(atomic_read(&d->ld_ref) == 0); + LASSERT(cfs_atomic_read(&d->ld_ref) == 0); EXIT; } @@ -4406,10 +4387,10 @@ static int mdt_adapt_sptlrpc_conf(struct obd_device *obd, int initial) sptlrpc_target_update_exp_flavor(obd, &tmp_rset); - write_lock(&m->mdt_sptlrpc_lock); + cfs_write_lock(&m->mdt_sptlrpc_lock); sptlrpc_rule_set_free(&m->mdt_sptlrpc_rset); m->mdt_sptlrpc_rset = tmp_rset; - write_unlock(&m->mdt_sptlrpc_lock); + cfs_write_unlock(&m->mdt_sptlrpc_lock); return 0; } @@ -4453,6 +4434,9 @@ static void fsoptions_to_mdt_flags(struct mdt_device *m, char *options) LCONSOLE_INFO("Disabling ACL\n"); } + if (!*p) + break; + options = ++p; } } @@ -4462,7 +4446,6 @@ int mdt_postrecov(const struct lu_env *, struct mdt_device *); static int mdt_init0(const struct lu_env *env, struct mdt_device *m, struct lu_device_type *ldt, struct lustre_cfg *cfg) { - struct lprocfs_static_vars lvars; struct mdt_thread_info *info; struct obd_device *obd; const char *dev = lustre_cfg_string(cfg, 0); @@ -4499,8 +4482,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, obd = class_name2obd(dev); LASSERT(obd != NULL); - spin_lock_init(&m->mdt_transno_lock); - m->mdt_max_mdsize = MAX_MD_SIZE; m->mdt_max_cookiesize = sizeof(struct llog_cookie); m->mdt_som_conf = 0; @@ -4523,12 +4504,14 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, CERROR("CMD Operation not allowed in IOP mode\n"); GOTO(err_lmi, rc = -EINVAL); } + + obd->u.obt.obt_magic = OBT_MAGIC; } - rwlock_init(&m->mdt_sptlrpc_lock); + cfs_rwlock_init(&m->mdt_sptlrpc_lock); sptlrpc_rule_set_init(&m->mdt_sptlrpc_rset); - spin_lock_init(&m->mdt_ioepoch_lock); + cfs_spin_lock_init(&m->mdt_ioepoch_lock); m->mdt_opts.mo_compat_resname = 0; m->mdt_capa_timeout = CAPA_TIMEOUT; m->mdt_capa_alg = CAPA_HMAC_ALG_SHA1; @@ -4538,9 +4521,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, CFS_INIT_LIST_HEAD(&m->mdt_nosquash_nids); m->mdt_nosquash_str = NULL; m->mdt_nosquash_strlen = 0; - init_rwsem(&m->mdt_squash_sem); - - spin_lock_init(&m->mdt_client_bitmap_lock); + cfs_init_rwsem(&m->mdt_squash_sem); OBD_ALLOC_PTR(mite); if (mite == NULL) @@ -4559,27 +4540,12 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, GOTO(err_free_site, rc); } - lprocfs_mdt_init_vars(&lvars); - rc = lprocfs_obd_setup(obd, lvars.obd_vars); - if (rc) { - CERROR("Can't init lprocfs, rc %d\n", rc); - GOTO(err_fini_site, rc); - } - ptlrpc_lprocfs_register_obd(obd); - rc = mdt_procfs_init(m, dev); if (rc) { CERROR("Can't init MDT lprocfs, rc %d\n", rc); GOTO(err_fini_proc, rc); } - obd->obd_proc_exports_entry = proc_mkdir("exports", - obd->obd_proc_entry); - if (obd->obd_proc_exports_entry) - lprocfs_add_simple(obd->obd_proc_exports_entry, - "clear", lprocfs_nid_stats_clear_read, - lprocfs_nid_stats_clear_write, obd, NULL); - /* set server index */ lu_site2md(s)->ms_node_id = node_id; @@ -4622,7 +4588,8 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, LUSTRE_MDT_NAME"-%p", m); m->mdt_namespace = ldlm_namespace_new(obd, info->mti_u.ns_name, LDLM_NAMESPACE_SERVER, - LDLM_NAMESPACE_GREEDY); + LDLM_NAMESPACE_GREEDY, + LDLM_NS_TYPE_MDT); if (m->mdt_namespace == NULL) GOTO(err_fini_seq, rc = -ENOMEM); @@ -4729,13 +4696,6 @@ err_fini_stack: mdt_stack_fini(env, m, md2lu_dev(m->mdt_child)); err_fini_proc: mdt_procfs_fini(m); - if (obd->obd_proc_exports_entry) { - lprocfs_remove_proc_entry("clear", obd->obd_proc_exports_entry); - obd->obd_proc_exports_entry = NULL; - } - ptlrpc_lprocfs_unregister_obd(obd); - lprocfs_obd_cleanup(obd); -err_fini_site: lu_site_fini(s); err_free_site: OBD_FREE_PTR(mite); @@ -4762,15 +4722,21 @@ static int mdt_process_config(const struct lu_env *env, /* * For interoperability between 1.8 and 2.0, - * skip old "mdt.group_upcall" param. */ { + /* Skip old "mdt.group_upcall" param. */ char *param = lustre_cfg_string(cfg, 1); if (param && !strncmp("mdt.group_upcall", param, 16)) { CWARN("For 1.8 interoperability, skip this" " mdt.group_upcall. It is obsolete\n"); break; } + /* Rename old "mdt.quota_type" to "mdd.quota_type. */ + if (param && !strncmp("mdt.quota_type", param, 14)) { + CWARN("Found old param mdt.quota_type, changed" + " it to mdd.quota_type.\n"); + param[2] = 'd'; + } } lprocfs_mdt_init_vars(&lvars); @@ -4819,6 +4785,8 @@ static struct lu_object *mdt_object_alloc(const struct lu_env *env, lu_object_init(o, h, d); lu_object_add_top(h, o); o->lo_ops = &mdt_obj_ops; + cfs_sema_init(&mo->mot_ioepoch_sem, 1); + cfs_sema_init(&mo->mot_lov_sem, 1); RETURN(o); } else RETURN(NULL); @@ -4866,10 +4834,10 @@ static int mdt_object_print(const struct lu_env *env, void *cookie, lu_printer_t p, const struct lu_object *o) { struct mdt_object *mdto = mdt_obj((struct lu_object *)o); - return (*p)(env, cookie, LUSTRE_MDT_NAME"-object@%p(ioepoch=%llu " - "flags=%llx, epochcount=%d, writecount=%d)", + return (*p)(env, cookie, LUSTRE_MDT_NAME"-object@%p(ioepoch="LPU64" " + "flags="LPX64", epochcount=%d, writecount=%d)", mdto, mdto->mot_ioepoch, mdto->mot_flags, - mdto->mot_epochcount, mdto->mot_writecount); + mdto->mot_ioepoch_count, mdto->mot_writecount); } static const struct lu_device_operations mdt_lu_ops = { @@ -4927,9 +4895,27 @@ static int mdt_connect_internal(struct obd_export *exp, if (!mdt->mdt_som_conf) data->ocd_connect_flags &= ~OBD_CONNECT_SOM; - spin_lock(&exp->exp_lock); + if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) { + data->ocd_brw_size = min(data->ocd_brw_size, + (__u32)(PTLRPC_MAX_BRW_PAGES << CFS_PAGE_SHIFT)); + if (data->ocd_brw_size == 0) { + CERROR("%s: cli %s/%p ocd_connect_flags: "LPX64 + " ocd_version: %x ocd_grant: %d " + "ocd_index: %u ocd_brw_size is " + "unexpectedly zero, network data " + "corruption? Refusing connection of this" + " client\n", + exp->exp_obd->obd_name, + exp->exp_client_uuid.uuid, + exp, data->ocd_connect_flags, data->ocd_version, + data->ocd_grant, data->ocd_index); + return -EPROTO; + } + } + + cfs_spin_lock(&exp->exp_lock); exp->exp_connect_flags = data->ocd_connect_flags; - spin_unlock(&exp->exp_lock); + cfs_spin_unlock(&exp->exp_lock); data->ocd_version = LUSTRE_VERSION_CODE; exp->exp_mdt_data.med_ibits_known = data->ocd_ibits_known; } @@ -4949,9 +4935,8 @@ static int mdt_connect_internal(struct obd_export *exp, return -EBADE; } - if (mdt->mdt_som_conf && - !(exp->exp_connect_flags & OBD_CONNECT_MDS_MDS) && - !(exp->exp_connect_flags & OBD_CONNECT_SOM)) { + if (mdt->mdt_som_conf && !exp_connect_som(exp) && + !(exp->exp_connect_flags & OBD_CONNECT_MDS_MDS)) { CWARN("%s: MDS has SOM enabled, but client does not support " "it\n", mdt->mdt_md_dev.md_lu_dev.ld_obd->obd_name); return -EBADE; @@ -4968,14 +4953,14 @@ static int mdt_connect_check_sptlrpc(struct mdt_device *mdt, int rc = 0; if (exp->exp_flvr.sf_rpc == SPTLRPC_FLVR_INVALID) { - read_lock(&mdt->mdt_sptlrpc_lock); + cfs_read_lock(&mdt->mdt_sptlrpc_lock); sptlrpc_target_choose_flavor(&mdt->mdt_sptlrpc_rset, req->rq_sp_from, req->rq_peer.nid, &flvr); - read_unlock(&mdt->mdt_sptlrpc_lock); + cfs_read_unlock(&mdt->mdt_sptlrpc_lock); - spin_lock(&exp->exp_lock); + cfs_spin_lock(&exp->exp_lock); exp->exp_sp_peer = req->rq_sp_from; exp->exp_flvr = flvr; @@ -4989,7 +4974,7 @@ static int mdt_connect_check_sptlrpc(struct mdt_device *mdt, rc = -EACCES; } - spin_unlock(&exp->exp_lock); + cfs_spin_unlock(&exp->exp_lock); } else { if (exp->exp_sp_peer != req->rq_sp_from) { CERROR("RPC source %s doesn't match %s\n", @@ -5012,7 +4997,6 @@ static int mdt_obd_connect(const struct lu_env *env, void *localdata) { struct mdt_thread_info *info; - struct lsd_client_data *lcd; struct obd_export *lexp; struct lustre_handle conn = { 0 }; struct mdt_device *mdt; @@ -5039,33 +5023,30 @@ static int mdt_obd_connect(const struct lu_env *env, if (rc) GOTO(out, rc); + if (OBD_FAIL_CHECK(OBD_FAIL_TGT_RCVG_FLAG)) + lustre_msg_add_op_flags(req->rq_repmsg, MSG_CONNECT_RECOVERING); + rc = mdt_connect_internal(lexp, mdt, data); if (rc == 0) { - OBD_ALLOC_PTR(lcd); - if (lcd != NULL) { - struct mdt_thread_info *mti; - mti = lu_context_key_get(&env->le_ctx, - &mdt_thread_key); - LASSERT(mti != NULL); - mti->mti_exp = lexp; - memcpy(lcd->lcd_uuid, cluuid, sizeof lcd->lcd_uuid); - lexp->exp_mdt_data.med_lcd = lcd; - rc = mdt_client_new(env, mdt); - if (rc != 0) { - OBD_FREE_PTR(lcd); - lexp->exp_mdt_data.med_lcd = NULL; - } else { - mdt_export_stats_init(obd, lexp, localdata); - } - } else - rc = -ENOMEM; + struct mdt_thread_info *mti; + struct lsd_client_data *lcd = lexp->exp_target_data.ted_lcd; + LASSERT(lcd); + mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key); + LASSERT(mti != NULL); + mti->mti_exp = lexp; + memcpy(lcd->lcd_uuid, cluuid, sizeof lcd->lcd_uuid); + rc = mdt_client_new(env, mdt); + if (rc == 0) + mdt_export_stats_init(obd, lexp, localdata); } out: - if (rc != 0) + if (rc != 0) { class_disconnect(lexp); - else + *exp = NULL; + } else { *exp = lexp; + } RETURN(rc); } @@ -5111,17 +5092,17 @@ static int mdt_export_cleanup(struct obd_export *exp) int rc = 0; ENTRY; - spin_lock(&med->med_open_lock); - while (!list_empty(&med->med_open_head)) { - struct list_head *tmp = med->med_open_head.next; - mfd = list_entry(tmp, struct mdt_file_data, mfd_list); + cfs_spin_lock(&med->med_open_lock); + while (!cfs_list_empty(&med->med_open_head)) { + cfs_list_t *tmp = med->med_open_head.next; + mfd = cfs_list_entry(tmp, struct mdt_file_data, mfd_list); /* Remove mfd handle so it can't be found again. * We are consuming the mfd_list reference here. */ class_handle_unhash(&mfd->mfd_handle); - list_move_tail(&mfd->mfd_list, &closing_list); + cfs_list_move_tail(&mfd->mfd_list, &closing_list); } - spin_unlock(&med->med_open_lock); + cfs_spin_unlock(&med->med_open_lock); mdt = mdt_dev(obd->obd_lu_dev); LASSERT(mdt != NULL); @@ -5136,44 +5117,49 @@ static int mdt_export_cleanup(struct obd_export *exp) info->mti_mdt = mdt; info->mti_exp = exp; - if (!list_empty(&closing_list)) { + if (!cfs_list_empty(&closing_list)) { struct md_attr *ma = &info->mti_attr; int lmm_size; int cookie_size; lmm_size = mdt->mdt_max_mdsize; - OBD_ALLOC(ma->ma_lmm, lmm_size); + OBD_ALLOC_LARGE(ma->ma_lmm, lmm_size); if (ma->ma_lmm == NULL) GOTO(out_lmm, rc = -ENOMEM); cookie_size = mdt->mdt_max_cookiesize; - OBD_ALLOC(ma->ma_cookie, cookie_size); + OBD_ALLOC_LARGE(ma->ma_cookie, cookie_size); if (ma->ma_cookie == NULL) GOTO(out_cookie, rc = -ENOMEM); /* Close any open files (which may also cause orphan unlinking). */ - list_for_each_entry_safe(mfd, n, &closing_list, mfd_list) { - list_del_init(&mfd->mfd_list); + cfs_list_for_each_entry_safe(mfd, n, &closing_list, mfd_list) { + cfs_list_del_init(&mfd->mfd_list); memset(&ma->ma_attr, 0, sizeof(ma->ma_attr)); ma->ma_lmm_size = lmm_size; ma->ma_cookie_size = cookie_size; ma->ma_need = 0; /* It is not for setattr, just tell MDD to send * DESTROY RPC to OSS if needed */ - ma->ma_attr_flags = MDS_CLOSE_CLEANUP; ma->ma_valid = MA_FLAGS; + ma->ma_attr_flags = MDS_CLOSE_CLEANUP; + /* Don't unlink orphan on failover umount, LU-184 */ + if (exp->exp_flags & OBD_OPT_FAILOVER) + ma->ma_attr_flags |= MDS_KEEP_ORPHAN; mdt_mfd_close(info, mfd); } - OBD_FREE(ma->ma_cookie, cookie_size); + OBD_FREE_LARGE(ma->ma_cookie, cookie_size); ma->ma_cookie = NULL; out_cookie: - OBD_FREE(ma->ma_lmm, lmm_size); + OBD_FREE_LARGE(ma->ma_lmm, lmm_size); ma->ma_lmm = NULL; } out_lmm: info->mti_mdt = NULL; /* cleanup client slot early */ - mdt_client_del(&env, mdt); + /* Do not erase record for recoverable client. */ + if (!(exp->exp_flags & OBD_OPT_FAILOVER) || exp->exp_failed) + mdt_client_del(&env, mdt); lu_env_fini(&env); RETURN(rc); @@ -5204,13 +5190,16 @@ static int mdt_init_export(struct obd_export *exp) ENTRY; CFS_INIT_LIST_HEAD(&med->med_open_head); - spin_lock_init(&med->med_open_lock); - sema_init(&med->med_idmap_sem, 1); + cfs_spin_lock_init(&med->med_open_lock); + cfs_sema_init(&med->med_idmap_sem, 1); med->med_idmap = NULL; - spin_lock(&exp->exp_lock); + cfs_spin_lock(&exp->exp_lock); exp->exp_connecting = 1; - spin_unlock(&exp->exp_lock); - rc = ldlm_init_export(exp); + cfs_spin_unlock(&exp->exp_lock); + rc = lut_client_alloc(exp); + if (rc == 0) + rc = ldlm_init_export(exp); + if (rc) CERROR("Error %d while initializing export\n", rc); RETURN(rc); @@ -5228,9 +5217,10 @@ static int mdt_destroy_export(struct obd_export *exp) target_destroy_export(exp); ldlm_destroy_export(exp); + lut_client_free(exp); - LASSERT(list_empty(&exp->exp_outstanding_replies)); - LASSERT(list_empty(&exp->exp_mdt_data.med_open_head)); + LASSERT(cfs_list_empty(&exp->exp_outstanding_replies)); + LASSERT(cfs_list_empty(&exp->exp_mdt_data.med_open_head)); if (obd_uuid_equals(&exp->exp_client_uuid, &exp->exp_obd->obd_uuid)) RETURN(0); @@ -5240,15 +5230,23 @@ static int mdt_destroy_export(struct obd_export *exp) static void mdt_allow_cli(struct mdt_device *m, unsigned int flag) { if (flag & CONFIG_LOG) - m->mdt_fl_cfglog = 1; + cfs_set_bit(MDT_FL_CFGLOG, &m->mdt_state); /* also notify active event */ if (flag & CONFIG_SYNC) - m->mdt_fl_synced = 1; + cfs_set_bit(MDT_FL_SYNCED, &m->mdt_state); + + if (cfs_test_bit(MDT_FL_CFGLOG, &m->mdt_state) && + cfs_test_bit(MDT_FL_SYNCED, &m->mdt_state)) { + struct obd_device *obd = m->mdt_md_dev.md_lu_dev.ld_obd; - if (m->mdt_fl_cfglog && m->mdt_fl_synced) /* Open for clients */ - m->mdt_md_dev.md_lu_dev.ld_obd->obd_no_conn = 0; + if (obd->obd_no_conn) { + cfs_spin_lock(&obd->obd_dev_lock); + obd->obd_no_conn = 0; + cfs_spin_unlock(&obd->obd_dev_lock); + } + } } static int mdt_upcall(const struct lu_env *env, struct md_device *md, @@ -5269,7 +5267,8 @@ static int mdt_upcall(const struct lu_env *env, struct md_device *md, m->mdt_max_mdsize, m->mdt_max_cookiesize); mdt_allow_cli(m, CONFIG_SYNC); if (data) - (*(__u64 *)data) = m->mdt_mount_count; + (*(__u64 *)data) = + m->mdt_lut.lut_obd->u.obt.obt_mount_count; break; case MD_NO_TRANS: mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key); @@ -5278,7 +5277,7 @@ static int mdt_upcall(const struct lu_env *env, struct md_device *md, break; case MD_LOV_CONFIG: /* Check that MDT is not yet configured */ - LASSERT(!m->mdt_fl_cfglog); + LASSERT(!cfs_test_bit(MDT_FL_CFGLOG, &m->mdt_state)); break; #ifdef HAVE_QUOTA_SUPPORT case MD_LOV_QUOTA: @@ -5295,11 +5294,11 @@ static int mdt_upcall(const struct lu_env *env, struct md_device *md, RETURN(rc); } -static int mdt_obd_notify(struct obd_device *host, +static int mdt_obd_notify(struct obd_device *obd, struct obd_device *watched, enum obd_notify_event ev, void *data) { - struct mdt_device *mdt = mdt_dev(host->obd_lu_dev); + struct mdt_device *mdt = mdt_dev(obd->obd_lu_dev); #ifdef HAVE_QUOTA_SUPPORT struct md_device *next = mdt->mdt_child; #endif @@ -5328,7 +5327,7 @@ static int mdt_rpc_fid2path(struct mdt_thread_info *info, void *key, struct getinfo_fid2path *fpout, *fpin; int rc = 0; - fpin = key + size_round(sizeof(KEY_FID2PATH)); + fpin = key + cfs_size_round(sizeof(KEY_FID2PATH)); fpout = val; if (ptlrpc_req_need_swab(info->mti_pill->rc_req)) @@ -5457,6 +5456,7 @@ static int mdt_ioc_version_get(struct mdt_thread_info *mti, void *karg) struct mdt_lock_handle *lh; int rc; ENTRY; + CDEBUG(D_IOCTL, "getting version for "DFID"\n", PFID(fid)); if (!fid_is_sane(fid)) RETURN(-EINVAL); @@ -5477,16 +5477,12 @@ static int mdt_ioc_version_get(struct mdt_thread_info *mti, void *karg) */ CERROR("nonlocal object "DFID"\n", PFID(fid)); } else if (rc == 0) { + *(__u64 *)data->ioc_inlbuf2 = ENOENT_VERSION; rc = -ENOENT; - CDEBUG(D_IOCTL, "no such object: "DFID"\n", PFID(fid)); } else { version = mo_version_get(mti->mti_env, mdt_object_child(obj)); - if (version < 0) { - rc = (int)version; - } else { - *(__u64 *)data->ioc_inlbuf2 = version; - rc = 0; - } + *(__u64 *)data->ioc_inlbuf2 = version; + rc = 0; } mdt_object_unlock_put(mti, obj, lh, 1); RETURN(rc); @@ -5549,16 +5545,12 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len, int mdt_postrecov(const struct lu_env *env, struct mdt_device *mdt) { struct lu_device *ld = md2lu_dev(mdt->mdt_child); - struct obd_device *obd = mdt2obd_dev(mdt); #ifdef HAVE_QUOTA_SUPPORT + struct obd_device *obd = mdt2obd_dev(mdt); struct md_device *next = mdt->mdt_child; #endif - int rc, lost; + int rc; ENTRY; - /* if some clients didn't participate in recovery then we can possibly - * lost sequence. Now we should increase sequence for safe value */ - lost = obd->obd_max_recoverable_clients - obd->obd_connected_clients; - mdt_seq_adjust(env, mdt, lost); rc = ld->ld_ops->ldo_recovery_complete(env, ld); #ifdef HAVE_QUOTA_SUPPORT @@ -5588,7 +5580,7 @@ int mdt_obd_postrecov(struct obd_device *obd) */ int mdt_hsm_copytool_send(struct obd_export *exp) { - struct lnl_hdr *lh; + struct kuc_hdr *lh; struct hsm_action_list *hal; struct hsm_action_item *hai; int rc, len; @@ -5598,15 +5590,15 @@ int mdt_hsm_copytool_send(struct obd_export *exp) libcfs_nid2str(exp->exp_connection->c_peer.nid)); len = sizeof(*lh) + sizeof(*hal) + MTI_NAME_MAXLEN + - /* for mockup below */ 2 * size_round(sizeof(*hai)); + /* for mockup below */ 2 * cfs_size_round(sizeof(*hai)); OBD_ALLOC(lh, len); if (lh == NULL) RETURN(-ENOMEM); - lh->lnl_magic = LNL_MAGIC; - lh->lnl_transport = LNL_TRANSPORT_HSM; - lh->lnl_msgtype = HMT_ACTION_LIST; - lh->lnl_msglen = len; + lh->kuc_magic = KUC_MAGIC; + lh->kuc_transport = KUC_TRANSPORT_HSM; + lh->kuc_msgtype = HMT_ACTION_LIST; + lh->kuc_msglen = len; hal = (struct hsm_action_list *)(lh + 1); hal->hal_version = HAL_VERSION; @@ -5627,10 +5619,11 @@ int mdt_hsm_copytool_send(struct obd_export *exp) /* Uses the ldlm reverse import; this rpc will be seen by the ldlm_callback_handler */ - rc = target_set_info_rpc(exp->exp_imp_reverse, LDLM_SET_INFO, - sizeof(KEY_HSM_COPYTOOL_SEND), - KEY_HSM_COPYTOOL_SEND, - len, lh, NULL); + rc = do_set_info_async(exp->exp_imp_reverse, + LDLM_SET_INFO, LUSTRE_OBD_VERSION, + sizeof(KEY_HSM_COPYTOOL_SEND), + KEY_HSM_COPYTOOL_SEND, + len, lh, NULL); OBD_FREE(lh, len); @@ -5786,7 +5779,17 @@ static int __init mdt_mod_init(void) llo_local_obj_register(&mdt_last_recv); - mdt_num_threads = MDT_NUM_THREADS; + if (mdt_num_threads > 0) { + if (mdt_num_threads > MDT_MAX_THREADS) + mdt_num_threads = MDT_MAX_THREADS; + if (mdt_num_threads < MDT_MIN_THREADS) + mdt_num_threads = MDT_MIN_THREADS; + mdt_max_threads = mdt_min_threads = mdt_num_threads; + } else { + mdt_max_threads = MDT_MAX_THREADS; + mdt_min_threads = MDT_MIN_THREADS; + } + lprocfs_mdt_init_vars(&lvars); rc = class_register_type(&mdt_obd_device_ops, NULL, lvars.module_vars, LUSTRE_MDT_NAME,