X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_handler.c;h=671ac7a82fea9bdf470972b3f9d4f40a283266fb;hp=b4c747a46ff0bb15a53dc4513b44a1086f4d2909;hb=99aec0dc17de5f980588c08d8befd0f3119db0ea;hpb=08b2bc314e2eef39edebd87616ea3a372718b310 diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c index b4c747a..671ac7a 100644 --- a/lustre/mdt/mdt_handler.c +++ b/lustre/mdt/mdt_handler.c @@ -1,6 +1,4 @@ -/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- - * vim:expandtab:shiftwidth=8:tabstop=8: - * +/* * GPL HEADER START * * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. @@ -26,8 +24,10 @@ * 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, 2012, Whamcloud, Inc. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -46,9 +46,6 @@ * Author: Yury Umanets */ -#ifndef EXPORT_SYMTAB -# define EXPORT_SYMTAB -#endif #define DEBUG_SUBSYSTEM S_MDS #include @@ -100,8 +97,6 @@ ldlm_mode_t mdt_dlm_lock_modes[] = { * Initialized in mdt_mod_init(). */ 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, @@ -210,7 +205,7 @@ void mdt_lock_pdo_init(struct mdt_lock_handle *lh, ldlm_mode_t lm, lh->mlh_reg_mode = lm; lh->mlh_type = MDT_PDO_LOCK; - if (name != NULL) { + if (name != NULL && (name[0] != '\0')) { LASSERT(namelen > 0); lh->mlh_pdo_hash = full_name_hash(name, namelen); } else { @@ -341,19 +336,20 @@ static int mdt_getstatus(struct mdt_thread_info *info) static int mdt_statfs(struct mdt_thread_info *info) { - struct md_device *next = info->mti_mdt->mdt_child; - struct ptlrpc_service *svc; - struct obd_statfs *osfs; - int rc; + struct ptlrpc_request *req = mdt_info_req(info); + struct md_device *next = info->mti_mdt->mdt_child; + struct ptlrpc_service_part *svcpt; + struct obd_statfs *osfs; + int rc; - ENTRY; + ENTRY; - svc = info->mti_pill->rc_req->rq_rqbd->rqbd_service; + svcpt = info->mti_pill->rc_req->rq_rqbd->rqbd_svcpt; - /* This will trigger a watchdog timeout */ - OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_STATFS_LCW_SLEEP, - (MDT_SERVICE_WATCHDOG_FACTOR * - at_get(&svc->srv_at_estimate)) + 1); + /* This will trigger a watchdog timeout */ + OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_STATFS_LCW_SLEEP, + (MDT_SERVICE_WATCHDOG_FACTOR * + at_get(&svcpt->scp_at_estimate)) + 1); rc = mdt_check_ucred(info); if (rc) @@ -363,10 +359,12 @@ static int mdt_statfs(struct mdt_thread_info *info) rc = err_serious(-ENOMEM); } else { osfs = req_capsule_server_get(info->mti_pill, &RMF_OBD_STATFS); - rc = next->md_ops->mdo_statfs(info->mti_env, next, - &info->mti_u.ksfs); - statfs_pack(osfs, &info->mti_u.ksfs); + rc = next->md_ops->mdo_statfs(info->mti_env, next, osfs); } + + if (rc == 0) + mdt_counter_incr(req, LPROC_MDT_STATFS); + RETURN(rc); } @@ -400,20 +398,9 @@ static void mdt_pack_size2body(struct mdt_thread_info *info, void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b, const struct lu_attr *attr, const struct lu_fid *fid) { - struct md_attr *ma = &info->mti_attr; - - /*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; + struct md_attr *ma = &info->mti_attr; - /* if no object is allocated on osts, the size on mds is valid. b=22272 */ - if (ma->ma_lmm_size == 0) - b->valid |= OBD_MD_FLSIZE; + LASSERT(ma->ma_valid & MA_INODE); b->atime = attr->la_atime; b->mtime = attr->la_mtime; @@ -427,6 +414,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; @@ -442,6 +446,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, @@ -451,6 +459,32 @@ static inline int mdt_body_has_lov(const struct lu_attr *la, (S_ISDIR(la->la_mode) && (body->valid & OBD_MD_FLDIREA )) ); } +void mdt_client_compatibility(struct mdt_thread_info *info) +{ + struct mdt_body *body; + struct ptlrpc_request *req = mdt_info_req(info); + struct obd_export *exp = req->rq_export; + struct md_attr *ma = &info->mti_attr; + struct lu_attr *la = &ma->ma_attr; + ENTRY; + + if (exp->exp_connect_flags & OBD_CONNECT_LAYOUTLOCK) + /* the client can deal with 16-bit lmm_stripe_count */ + RETURN_EXIT; + + body = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY); + + if (!mdt_body_has_lov(la, body)) + RETURN_EXIT; + + /* now we have a reply with a lov for a client not compatible with the + * layout lock so we have to clean the layout generation number */ + if (S_ISREG(la->la_mode)) + ma->ma_lmm->lmm_layout_gen = 0; + EXIT; +} + + static int mdt_getattr_internal(struct mdt_thread_info *info, struct mdt_object *o, int ma_need) { @@ -478,7 +512,7 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, /* This object is located on remote node.*/ repbody->fid1 = *mdt_object_fid(o); repbody->valid = OBD_MD_FLID | OBD_MD_MDS; - RETURN(0); + GOTO(out, rc = 0); } buffer->lb_buf = req_capsule_server_get(pill, &RMF_MDT_MD); @@ -534,27 +568,38 @@ 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; - buffer->lb_len = reqbody->eadatasize; + /* eadatasize from client includes NULL-terminator, so + * there is no need to read it */ + buffer->lb_len = reqbody->eadatasize - 1; rc = mo_readlink(env, next, buffer); if (unlikely(rc <= 0)) { CERROR("readlink failed: %d\n", rc); rc = -EFAULT; } else { - if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READLINK_EPROTO)) - rc -= 2; - repbody->valid |= OBD_MD_LINKNAME; - repbody->eadatasize = rc; - /* NULL terminate */ - ((char*)ma->ma_lmm)[rc - 1] = 0; - CDEBUG(D_INODE, "symlink dest %s, len = %d\n", - (char*)ma->ma_lmm, rc); - rc = 0; + int print_limit = min_t(int, CFS_PAGE_SIZE - 128, rc); + + if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READLINK_EPROTO)) + rc -= 2; + repbody->valid |= OBD_MD_LINKNAME; + /* we need to report back size with NULL-terminator + * because client expects that */ + repbody->eadatasize = rc + 1; + if (repbody->eadatasize != reqbody->eadatasize) + CERROR("Read shorter symlink %d, expected %d\n", + rc, reqbody->eadatasize - 1); + /* NULL terminate */ + ((char *)ma->ma_lmm)[rc] = 0; + + /* If the total CDEBUG() size is larger than a page, it + * will print a warning to the console, avoid this by + * printing just the last part of the symlink. */ + CDEBUG(D_INODE, "symlink dest %s%.*s, len = %d\n", + print_limit < rc ? "..." : "", print_limit, + (char *)ma->ma_lmm + rc - print_limit, rc); + rc = 0; } } @@ -623,6 +668,11 @@ static int mdt_getattr_internal(struct mdt_thread_info *info, RETURN(rc); repbody->valid |= OBD_MD_FLMDSCAPA; } + +out: + if (rc == 0) + mdt_counter_incr(req, LPROC_MDT_GETATTR); + RETURN(rc); } @@ -665,8 +715,7 @@ static int mdt_getattr(struct mdt_thread_info *info) struct mdt_body *reqbody; struct mdt_body *repbody; mode_t mode; - int md_size; - int rc; + int rc, rc2; ENTRY; reqbody = req_capsule_client_get(pill, &RMF_MDT_BODY); @@ -684,13 +733,12 @@ static int mdt_getattr(struct mdt_thread_info *info) LASSERT(lu_object_assert_exists(&obj->mot_obj.mo_lu)); mode = lu_object_attr(&obj->mot_obj.mo_lu); - if (S_ISLNK(mode) && (reqbody->valid & OBD_MD_LINKNAME) && - (reqbody->eadatasize > info->mti_mdt->mdt_max_mdsize)) - md_size = reqbody->eadatasize; - else - md_size = info->mti_mdt->mdt_max_mdsize; - req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, md_size); + /* old clients may not report needed easize, use max value then */ + req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, + reqbody->eadatasize == 0 ? + info->mti_mdt->mdt_max_mdsize : + reqbody->eadatasize); rc = req_capsule_server_pack(pill); if (unlikely(rc != 0)) @@ -721,7 +769,10 @@ static int mdt_getattr(struct mdt_thread_info *info) mdt_exit_ucred(info); EXIT; out_shrink: - mdt_shrink_reply(info); + mdt_client_compatibility(info); + rc2 = mdt_fix_reply(info); + if (rc == 0) + rc = rc2; return rc; } @@ -770,6 +821,7 @@ static int mdt_raw_lookup(struct mdt_thread_info *info, LASSERT(!info->mti_cross_ref); /* Only got the fid of this obj by name */ + fid_zero(child_fid); rc = mdo_lookup(info->mti_env, next, lname, child_fid, &info->mti_spec); #if 0 @@ -811,7 +863,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, struct lu_name *lname = NULL; const char *name = NULL; int namelen = 0; - struct mdt_lock_handle *lhp; + struct mdt_lock_handle *lhp = NULL; struct ldlm_lock *lock; struct ldlm_res_id *res_id; int is_resent; @@ -841,13 +893,17 @@ 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), - ldlm_rep); + PFID(mdt_object_fid(parent)), + PFID(&reqbody->fid2), ldlm_rep); } else { lname = mdt_name(info->mti_env, (char *)name, namelen); CDEBUG(D_INODE, "getattr with lock for "DFID"/%s, " @@ -857,16 +913,16 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, } mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_EXECD); - rc = mdt_object_exists(parent); - if (unlikely(rc == 0)) { - LU_OBJECT_DEBUG(D_WARNING, info->mti_env, - &parent->mot_obj.mo_lu, - "Parent doesn't exist!\n"); - RETURN(-ESTALE); - } else if (!info->mti_cross_ref) { - LASSERTF(rc > 0, "Parent "DFID" is on remote server\n", - PFID(mdt_object_fid(parent))); - } + rc = mdt_object_exists(parent); + if (unlikely(rc == 0)) { + LU_OBJECT_DEBUG(D_INODE, info->mti_env, + &parent->mot_obj.mo_lu, + "Parent doesn't exist!\n"); + RETURN(-ESTALE); + } else if (!info->mti_cross_ref) { + LASSERTF(rc > 0, "Parent "DFID" is on remote server\n", + PFID(mdt_object_fid(parent))); + } if (lname) { rc = mdt_raw_lookup(info, parent, lname, ldlm_rep); if (rc != 0) { @@ -917,23 +973,27 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info, RETURN(rc); } - /* step 1: lock parent */ - lhp = &info->mti_lh[MDT_LH_PARENT]; - mdt_lock_pdo_init(lhp, LCK_PR, name, namelen); - rc = mdt_object_lock(info, parent, lhp, MDS_INODELOCK_UPDATE, - MDT_LOCAL_LOCK); - - if (unlikely(rc != 0)) - RETURN(rc); - if (lname) { + /* step 1: lock parent only if parent is a directory */ + if (S_ISDIR(lu_object_attr(&parent->mot_obj.mo_lu))) { + 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 */ + fid_zero(child_fid); rc = mdo_lookup(info->mti_env, next, lname, child_fid, &info->mti_spec); if (rc != 0) { if (rc == -ENOENT) - mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG); + mdt_set_disposition(info, ldlm_rep, + DISP_LOOKUP_NEG); GOTO(out_parent, rc); } else mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS); @@ -977,36 +1037,48 @@ 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); + if (child_bits == MDS_INODELOCK_LAYOUT) + mdt_lock_reg_init(lhc, LCK_CR); + else + mdt_lock_reg_init(lhc, LCK_PR); if (mdt_object_exists(child) == 0) { - LU_OBJECT_DEBUG(D_WARNING, info->mti_env, + LU_OBJECT_DEBUG(D_INODE, 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); + + /* layout lock is used only on regular files */ + if ((ma->ma_valid & MA_INODE) && + (ma->ma_attr.la_valid & LA_MODE) && + !S_ISREG(ma->ma_attr.la_mode)) + child_bits &= ~MDS_INODELOCK_LAYOUT; + + /* 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); @@ -1047,7 +1119,8 @@ relock: out_child: mdt_object_put(info->mti_env, child); out_parent: - mdt_object_unlock(info, parent, lhp, 1); + if (lhp) + mdt_object_unlock(info, parent, lhp, 1); return rc; } @@ -1057,7 +1130,7 @@ static int mdt_getattr_name(struct mdt_thread_info *info) struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_CHILD]; struct mdt_body *reqbody; struct mdt_body *repbody; - int rc; + int rc, rc2; ENTRY; reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY); @@ -1082,7 +1155,10 @@ static int mdt_getattr_name(struct mdt_thread_info *info) mdt_exit_ucred(info); EXIT; out_shrink: - mdt_shrink_reply(info); + mdt_client_compatibility(info); + rc2 = mdt_fix_reply(info); + if (rc == 0) + rc = rc2; return rc; } @@ -1193,7 +1269,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; @@ -1201,7 +1277,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; @@ -1211,58 +1286,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 @@ -1470,8 +1503,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); @@ -1484,11 +1521,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: @@ -1509,19 +1546,26 @@ static int mdt_reint_internal(struct mdt_thread_info *info, __u32 op) { struct req_capsule *pill = info->mti_pill; - struct mdt_device *mdt = info->mti_mdt; struct md_quota *mq = md_quota(info->mti_env); struct mdt_body *repbody; - int rc = 0; + int rc = 0, rc2; ENTRY; - /* pack reply */ + + 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, - mdt->mdt_max_mdsize); + info->mti_rr.rr_eadatalen); + if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER)) req_capsule_set_size(pill, &RMF_LOGCOOKIES, RCL_SERVER, - mdt->mdt_max_cookiesize); + info->mti_mdt->mdt_max_cookiesize); rc = req_capsule_server_pack(pill); if (rc != 0) { @@ -1536,27 +1580,13 @@ static int mdt_reint_internal(struct mdt_thread_info *info, repbody->aclsize = 0; } - if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNPACK)) - GOTO(out_shrink, rc = err_serious(-EFAULT)); - - rc = mdt_reint_unpack(info, op); - if (rc != 0) { - CERROR("Can't unpack reint, rc %d\n", rc); - GOTO(out_shrink, rc = err_serious(rc)); - } - 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 == 1) { + 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); - if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER)) - req_capsule_set_size(pill, &RMF_LOGCOOKIES, RCL_SERVER, - 0); - } - rc = mdt_init_ucred_reint(info); if (rc) GOTO(out_shrink, rc); @@ -1575,7 +1605,10 @@ static int mdt_reint_internal(struct mdt_thread_info *info, out_ucred: mdt_exit_ucred(info); out_shrink: - mdt_shrink_reply(info); + mdt_client_compatibility(info); + rc2 = mdt_fix_reply(info); + if (rc == 0) + rc = rc2; return rc; } @@ -1663,6 +1696,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; @@ -1710,6 +1744,9 @@ static int mdt_sync(struct mdt_thread_info *info) } else rc = err_serious(rc); } + if (rc == 0) + mdt_counter_incr(req, LPROC_MDT_SYNC); + RETURN(rc); } @@ -1975,7 +2012,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; /* @@ -1985,17 +2021,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; @@ -2045,7 +2070,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; } @@ -2056,6 +2081,24 @@ static struct mdt_object *mdt_obj(struct lu_object *o) return container_of0(o, struct mdt_object, mot_obj.mo_lu); } +struct mdt_object *mdt_object_new(const struct lu_env *env, + struct mdt_device *d, + const struct lu_fid *f) +{ + struct lu_object_conf conf = { .loc_flags = LOC_F_NEW }; + struct lu_object *o; + struct mdt_object *m; + ENTRY; + + CDEBUG(D_INFO, "Allocate object for "DFID"\n", PFID(f)); + o = lu_object_find(env, &d->mdt_md_dev.md_lu_dev, f, &conf); + if (unlikely(IS_ERR(o))) + m = (struct mdt_object *)o; + else + m = mdt_obj(o); + RETURN(m); +} + struct mdt_object *mdt_object_find(const struct lu_env *env, struct mdt_device *d, const struct lu_fid *f) @@ -2138,7 +2181,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; @@ -2164,7 +2207,7 @@ int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc, if (lock->l_req_mode == LCK_COS && lock->l_blocking_lock != NULL) { struct lu_env env; - rc = lu_env_init(&env, LCT_MD_THREAD); + rc = lu_env_init(&env, LCT_LOCAL); if (unlikely(rc != 0)) CWARN("lu_env initialization failed with rc = %d," "cannot start asynchronous commit\n", rc); @@ -2209,7 +2252,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); } } @@ -2259,7 +2303,7 @@ int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o, else if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_PDO_LOCK)) && lh->mlh_pdo_hash != 0 && (lh->mlh_reg_mode == LCK_PW || lh->mlh_reg_mode == LCK_EX)) { - OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PDO_LOCK, 10); + OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PDO_LOCK, 15); } RETURN(rc); @@ -2486,16 +2530,13 @@ static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags) rc = 0; if (rc == 0 && (flags & HABEO_REFERO)) { - struct mdt_device *mdt = info->mti_mdt; - /* Pack reply. */ - if (req_capsule_has_field(pill, &RMF_MDT_MD, RCL_SERVER)) req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, - mdt->mdt_max_mdsize); + info->mti_body->eadatasize); if (req_capsule_has_field(pill, &RMF_LOGCOOKIES, RCL_SERVER)) req_capsule_set_size(pill, &RMF_LOGCOOKIES, RCL_SERVER, - mdt->mdt_max_cookiesize); + info->mti_mdt->mdt_max_cookiesize); rc = req_capsule_server_pack(pill); } @@ -2565,10 +2606,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; } @@ -2625,7 +2681,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(); @@ -2689,10 +2745,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; @@ -3056,6 +3109,7 @@ enum mdt_it_code { MDT_IT_UNLINK, MDT_IT_TRUNC, MDT_IT_GETXATTR, + MDT_IT_LAYOUT, MDT_IT_NR }; @@ -3126,6 +3180,11 @@ static struct mdt_it_flavor { .it_fmt = NULL, .it_flags = 0, .it_act = NULL + }, + [MDT_IT_LAYOUT] = { + .it_fmt = &RQF_LDLM_INTENT_GETATTR, + .it_flags = HABEO_REFERO, + .it_act = mdt_intent_getattr } }; @@ -3280,7 +3339,7 @@ static int mdt_intent_getattr(enum mdt_it_code opcode, struct ptlrpc_request *req; struct mdt_body *reqbody; struct mdt_body *repbody; - int rc; + int rc, rc2; ENTRY; reqbody = req_capsule_client_get(info->mti_pill, &RMF_MDT_BODY); @@ -3301,8 +3360,18 @@ static int mdt_intent_getattr(enum mdt_it_code opcode, case MDT_IT_GETATTR: child_bits = MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE; break; + case MDT_IT_LAYOUT: { + static int printed = 0; + + if (!printed) { + CERROR("layout lock not supported by this version\n"); + printed = 1; + } + GOTO(out_shrink, rc = -EINVAL); + break; + } default: - CERROR("Unhandled till now"); + CERROR("Unsupported intent (%d)\n", opcode); GOTO(out_shrink, rc = -EINVAL); } @@ -3333,7 +3402,10 @@ static int mdt_intent_getattr(enum mdt_it_code opcode, out_ucred: mdt_exit_ucred(info); out_shrink: - mdt_shrink_reply(info); + mdt_client_compatibility(info); + rc2 = mdt_fix_reply(info); + if (rc == 0) + rc = rc2; return rc; } @@ -3388,6 +3460,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 */ /* @@ -3408,6 +3484,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); @@ -3450,6 +3528,9 @@ static int mdt_intent_code(long itcode) case IT_GETXATTR: rc = MDT_IT_GETXATTR; break; + case IT_LAYOUT: + rc = MDT_IT_LAYOUT; + break; default: CERROR("Unknown intent opcode: %ld\n", itcode); rc = -EINVAL; @@ -3517,23 +3598,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 { @@ -3546,25 +3623,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) { @@ -3868,250 +3926,311 @@ static void mdt_stop_ptlrpc_service(struct mdt_device *m) static int mdt_start_ptlrpc_service(struct mdt_device *m) { - int rc; static struct ptlrpc_service_conf conf; cfs_proc_dir_entry_t *procfs_entry; - ENTRY; - - procfs_entry = m->mdt_md_dev.md_lu_dev.ld_obd->obd_proc_entry; - - conf = (typeof(conf)) { - .psc_nbufs = MDS_NBUFS, - .psc_bufsize = MDS_BUFSIZE, - .psc_max_req_size = MDS_MAXREQSIZE, - .psc_max_reply_size = MDS_MAXREPSIZE, - .psc_req_portal = MDS_REQUEST_PORTAL, - .psc_rep_portal = MDC_REPLY_PORTAL, - .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - /* - * We'd like to have a mechanism to set this on a per-device - * basis, but alas... - */ - .psc_min_threads = mdt_min_threads, - .psc_max_threads = mdt_max_threads, - .psc_ctx_tags = LCT_MD_THREAD - }; - - m->mdt_ldlm_client = &m->mdt_md_dev.md_lu_dev.ld_obd->obd_ldlm_client; - ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL, - "mdt_ldlm_client", m->mdt_ldlm_client); - - m->mdt_regular_service = - ptlrpc_init_svc_conf(&conf, mdt_regular_handle, LUSTRE_MDT_NAME, - procfs_entry, target_print_req, - LUSTRE_MDT_NAME); - if (m->mdt_regular_service == NULL) - RETURN(-ENOMEM); - - rc = ptlrpc_start_threads(NULL, m->mdt_regular_service); - if (rc) - GOTO(err_mdt_svc, rc); - - /* - * readpage service configuration. Parameters have to be adjusted, - * ideally. - */ - conf = (typeof(conf)) { - .psc_nbufs = MDS_NBUFS, - .psc_bufsize = MDS_BUFSIZE, - .psc_max_req_size = MDS_MAXREQSIZE, - .psc_max_reply_size = MDS_MAXREPSIZE, - .psc_req_portal = MDS_READPAGE_PORTAL, - .psc_rep_portal = MDC_REPLY_PORTAL, - .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = mdt_min_threads, - .psc_max_threads = mdt_max_threads, - .psc_ctx_tags = LCT_MD_THREAD - }; - m->mdt_readpage_service = - ptlrpc_init_svc_conf(&conf, mdt_readpage_handle, - LUSTRE_MDT_NAME "_readpage", - procfs_entry, target_print_req,"mdt_rdpg"); - - if (m->mdt_readpage_service == NULL) { - CERROR("failed to start readpage service\n"); - GOTO(err_mdt_svc, rc = -ENOMEM); + int rc = 0; + ENTRY; + + m->mdt_ldlm_client = &m->mdt_md_dev.md_lu_dev.ld_obd->obd_ldlm_client; + ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL, + "mdt_ldlm_client", m->mdt_ldlm_client); + + procfs_entry = m->mdt_md_dev.md_lu_dev.ld_obd->obd_proc_entry; + + conf = (typeof(conf)) { + .psc_name = LUSTRE_MDT_NAME, + .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, + .psc_buf = { + .bc_nbufs = MDS_NBUFS, + .bc_buf_size = MDS_BUFSIZE, + .bc_req_max_size = MDS_MAXREQSIZE, + .bc_rep_max_size = MDS_MAXREPSIZE, + .bc_req_portal = MDS_REQUEST_PORTAL, + .bc_rep_portal = MDC_REPLY_PORTAL, + }, + /* + * We'd like to have a mechanism to set this on a per-device + * basis, but alas... + */ + .psc_thr = { + .tc_thr_name = LUSTRE_MDT_NAME, + .tc_nthrs_min = MDT_MIN_THREADS, + .tc_nthrs_max = MDT_MAX_THREADS, + .tc_nthrs_user = mdt_num_threads, + .tc_ctx_tags = LCT_MD_THREAD, + }, + .psc_ops = { + .so_req_handler = mdt_regular_handle, + .so_req_printer = target_print_req, + }, + }; + m->mdt_regular_service = ptlrpc_register_service(&conf, procfs_entry); + if (IS_ERR(m->mdt_regular_service)) { + rc = PTR_ERR(m->mdt_regular_service); + CERROR("failed to start regular mdt service: %d\n", rc); + m->mdt_regular_service = NULL; + + RETURN(rc); + } + + /* + * readpage service configuration. Parameters have to be adjusted, + * ideally. + */ + memset(&conf, 0, sizeof(conf)); + conf = (typeof(conf)) { + .psc_name = LUSTRE_MDT_NAME "_readpage", + .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, + .psc_buf = { + .bc_nbufs = MDS_NBUFS, + .bc_buf_size = MDS_BUFSIZE, + .bc_req_max_size = MDS_MAXREQSIZE, + .bc_rep_max_size = MDS_MAXREPSIZE, + .bc_req_portal = MDS_READPAGE_PORTAL, + .bc_rep_portal = MDC_REPLY_PORTAL, + }, + .psc_thr = { + .tc_thr_name = "mdt_rdpg", + .tc_nthrs_min = MDT_MIN_THREADS, + .tc_nthrs_max = MDT_MAX_THREADS, + .tc_nthrs_user = mdt_num_threads, + .tc_ctx_tags = LCT_MD_THREAD, + }, + .psc_ops = { + .so_req_handler = mdt_readpage_handle, + .so_req_printer = target_print_req, + }, + }; + m->mdt_readpage_service = ptlrpc_register_service(&conf, procfs_entry); + if (IS_ERR(m->mdt_readpage_service)) { + rc = PTR_ERR(m->mdt_readpage_service); + CERROR("failed to start readpage service: %d\n", rc); + m->mdt_readpage_service = NULL; + + GOTO(err_mdt_svc, rc); } - rc = ptlrpc_start_threads(NULL, m->mdt_readpage_service); - /* * setattr service configuration. + * + * XXX To keep the compatibility with old client(< 2.2), we need to + * preserve this portal for a certain time, it should be removed + * eventually. LU-617. */ - conf = (typeof(conf)) { - .psc_nbufs = MDS_NBUFS, - .psc_bufsize = MDS_BUFSIZE, - .psc_max_req_size = MDS_MAXREQSIZE, - .psc_max_reply_size = MDS_MAXREPSIZE, - .psc_req_portal = MDS_SETATTR_PORTAL, - .psc_rep_portal = MDC_REPLY_PORTAL, - .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = mdt_min_threads, - .psc_max_threads = mdt_max_threads, - .psc_ctx_tags = LCT_MD_THREAD - }; - - m->mdt_setattr_service = - ptlrpc_init_svc_conf(&conf, mdt_regular_handle, - LUSTRE_MDT_NAME "_setattr", - procfs_entry, target_print_req,"mdt_attr"); - - if (!m->mdt_setattr_service) { - CERROR("failed to start setattr service\n"); - GOTO(err_mdt_svc, rc = -ENOMEM); - } - - rc = ptlrpc_start_threads(NULL, m->mdt_setattr_service); - if (rc) - GOTO(err_mdt_svc, rc); - - /* - * sequence controller service configuration - */ - conf = (typeof(conf)) { - .psc_nbufs = MDS_NBUFS, - .psc_bufsize = MDS_BUFSIZE, - .psc_max_req_size = SEQ_MAXREQSIZE, - .psc_max_reply_size = SEQ_MAXREPSIZE, - .psc_req_portal = SEQ_CONTROLLER_PORTAL, - .psc_rep_portal = MDC_REPLY_PORTAL, - .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = mdt_min_threads, - .psc_max_threads = mdt_max_threads, - .psc_ctx_tags = LCT_MD_THREAD|LCT_DT_THREAD - }; - - m->mdt_mdsc_service = - ptlrpc_init_svc_conf(&conf, mdt_mdsc_handle, - LUSTRE_MDT_NAME"_mdsc", - procfs_entry, target_print_req,"mdt_mdsc"); - if (!m->mdt_mdsc_service) { - CERROR("failed to start seq controller service\n"); - GOTO(err_mdt_svc, rc = -ENOMEM); - } - - rc = ptlrpc_start_threads(NULL, m->mdt_mdsc_service); - if (rc) - GOTO(err_mdt_svc, rc); - - /* - * metadata sequence server service configuration - */ - conf = (typeof(conf)) { - .psc_nbufs = MDS_NBUFS, - .psc_bufsize = MDS_BUFSIZE, - .psc_max_req_size = SEQ_MAXREQSIZE, - .psc_max_reply_size = SEQ_MAXREPSIZE, - .psc_req_portal = SEQ_METADATA_PORTAL, - .psc_rep_portal = MDC_REPLY_PORTAL, - .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = mdt_min_threads, - .psc_max_threads = mdt_max_threads, - .psc_ctx_tags = LCT_MD_THREAD|LCT_DT_THREAD + memset(&conf, 0, sizeof(conf)); + conf = (typeof(conf)) { + .psc_name = LUSTRE_MDT_NAME "_setattr", + .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, + .psc_buf = { + .bc_nbufs = MDS_NBUFS, + .bc_buf_size = MDS_BUFSIZE, + .bc_req_max_size = MDS_MAXREQSIZE, + .bc_rep_max_size = MDS_MAXREPSIZE, + .bc_req_portal = MDS_SETATTR_PORTAL, + .bc_rep_portal = MDC_REPLY_PORTAL, + }, + .psc_thr = { + .tc_thr_name = "mdt_attr", + .tc_nthrs_min = MDT_MIN_THREADS, + .tc_nthrs_max = MDT_MAX_THREADS, + .tc_nthrs_user = mdt_num_threads, + .tc_ctx_tags = LCT_MD_THREAD, + }, + .psc_ops = { + .so_req_handler = mdt_regular_handle, + .so_req_printer = target_print_req, + }, + }; + m->mdt_setattr_service = ptlrpc_register_service(&conf, procfs_entry); + if (IS_ERR(m->mdt_setattr_service)) { + rc = PTR_ERR(m->mdt_setattr_service); + CERROR("failed to start setattr service: %d\n", rc); + m->mdt_setattr_service = NULL; + + GOTO(err_mdt_svc, rc); + } + + /* + * sequence controller service configuration + */ + memset(&conf, 0, sizeof(conf)); + conf = (typeof(conf)) { + .psc_name = LUSTRE_MDT_NAME "_mdsc", + .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, + .psc_buf = { + .bc_nbufs = MDS_NBUFS, + .bc_buf_size = MDS_BUFSIZE, + .bc_req_max_size = SEQ_MAXREQSIZE, + .bc_rep_max_size = SEQ_MAXREPSIZE, + .bc_req_portal = SEQ_CONTROLLER_PORTAL, + .bc_rep_portal = MDC_REPLY_PORTAL, + }, + .psc_thr = { + .tc_thr_name = "mdt_mdsc", + .tc_nthrs_min = MDT_MIN_THREADS, + .tc_nthrs_max = MDT_MAX_THREADS, + .tc_nthrs_user = mdt_num_threads, + .tc_ctx_tags = LCT_MD_THREAD, + }, + .psc_ops = { + .so_req_handler = mdt_mdsc_handle, + .so_req_printer = target_print_req, + }, + }; + m->mdt_mdsc_service = ptlrpc_register_service(&conf, procfs_entry); + if (IS_ERR(m->mdt_mdsc_service)) { + rc = PTR_ERR(m->mdt_mdsc_service); + CERROR("failed to start seq controller service: %d\n", rc); + m->mdt_mdsc_service = NULL; + + GOTO(err_mdt_svc, rc); + } + + /* + * metadata sequence server service configuration + */ + memset(&conf, 0, sizeof(conf)); + conf = (typeof(conf)) { + .psc_name = LUSTRE_MDT_NAME "_mdss", + .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, + .psc_buf = { + .bc_nbufs = MDS_NBUFS, + .bc_buf_size = MDS_BUFSIZE, + .bc_req_max_size = SEQ_MAXREQSIZE, + .bc_rep_max_size = SEQ_MAXREPSIZE, + .bc_req_portal = SEQ_METADATA_PORTAL, + .bc_rep_portal = MDC_REPLY_PORTAL, + }, + .psc_thr = { + .tc_thr_name = "mdt_mdss", + .tc_nthrs_min = MDT_MIN_THREADS, + .tc_nthrs_max = MDT_MAX_THREADS, + .tc_nthrs_user = mdt_num_threads, + .tc_ctx_tags = LCT_MD_THREAD | LCT_DT_THREAD + }, + .psc_ops = { + .so_req_handler = mdt_mdss_handle, + .so_req_printer = target_print_req, + }, }; - - m->mdt_mdss_service = - ptlrpc_init_svc_conf(&conf, mdt_mdss_handle, - LUSTRE_MDT_NAME"_mdss", - procfs_entry, target_print_req,"mdt_mdss"); - if (!m->mdt_mdss_service) { - CERROR("failed to start metadata seq server service\n"); - GOTO(err_mdt_svc, rc = -ENOMEM); - } - - rc = ptlrpc_start_threads(NULL, m->mdt_mdss_service); - if (rc) - GOTO(err_mdt_svc, rc); - - - /* - * Data sequence server service configuration. We want to have really - * cluster-wide sequences space. This is why we start only one sequence - * controller which manages space. - */ - conf = (typeof(conf)) { - .psc_nbufs = MDS_NBUFS, - .psc_bufsize = MDS_BUFSIZE, - .psc_max_req_size = SEQ_MAXREQSIZE, - .psc_max_reply_size = SEQ_MAXREPSIZE, - .psc_req_portal = SEQ_DATA_PORTAL, - .psc_rep_portal = OSC_REPLY_PORTAL, - .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = mdt_min_threads, - .psc_max_threads = mdt_max_threads, - .psc_ctx_tags = LCT_MD_THREAD|LCT_DT_THREAD + m->mdt_mdss_service = ptlrpc_register_service(&conf, procfs_entry); + if (IS_ERR(m->mdt_mdss_service)) { + rc = PTR_ERR(m->mdt_mdss_service); + CERROR("failed to start metadata seq server service: %d\n", rc); + m->mdt_mdss_service = NULL; + + GOTO(err_mdt_svc, rc); + } + + /* + * Data sequence server service configuration. We want to have really + * cluster-wide sequences space. This is why we start only one sequence + * controller which manages space. + */ + memset(&conf, 0, sizeof(conf)); + conf = (typeof(conf)) { + .psc_name = LUSTRE_MDT_NAME "_dtss", + .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, + .psc_buf = { + .bc_nbufs = MDS_NBUFS, + .bc_buf_size = MDS_BUFSIZE, + .bc_req_max_size = SEQ_MAXREQSIZE, + .bc_rep_max_size = SEQ_MAXREPSIZE, + .bc_req_portal = SEQ_DATA_PORTAL, + .bc_rep_portal = OSC_REPLY_PORTAL, + }, + .psc_thr = { + .tc_thr_name = "mdt_dtss", + .tc_nthrs_min = MDT_MIN_THREADS, + .tc_nthrs_max = MDT_MAX_THREADS, + .tc_nthrs_user = mdt_num_threads, + .tc_ctx_tags = LCT_MD_THREAD | LCT_DT_THREAD + }, + .psc_ops = { + .so_req_handler = mdt_dtss_handle, + .so_req_printer = target_print_req, + }, }; - - m->mdt_dtss_service = - ptlrpc_init_svc_conf(&conf, mdt_dtss_handle, - LUSTRE_MDT_NAME"_dtss", - procfs_entry, target_print_req,"mdt_dtss"); - if (!m->mdt_dtss_service) { - CERROR("failed to start data seq server service\n"); - GOTO(err_mdt_svc, rc = -ENOMEM); - } - - rc = ptlrpc_start_threads(NULL, m->mdt_dtss_service); - if (rc) - GOTO(err_mdt_svc, rc); - - /* FLD service start */ - conf = (typeof(conf)) { - .psc_nbufs = MDS_NBUFS, - .psc_bufsize = MDS_BUFSIZE, - .psc_max_req_size = FLD_MAXREQSIZE, - .psc_max_reply_size = FLD_MAXREPSIZE, - .psc_req_portal = FLD_REQUEST_PORTAL, - .psc_rep_portal = MDC_REPLY_PORTAL, + m->mdt_dtss_service = ptlrpc_register_service(&conf, procfs_entry); + if (IS_ERR(m->mdt_dtss_service)) { + rc = PTR_ERR(m->mdt_dtss_service); + CERROR("failed to start data seq server service: %d\n", rc); + m->mdt_dtss_service = NULL; + + GOTO(err_mdt_svc, rc); + } + + /* FLD service start */ + memset(&conf, 0, sizeof(conf)); + conf = (typeof(conf)) { + .psc_name = LUSTRE_MDT_NAME "_fld", .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = mdt_min_threads, - .psc_max_threads = mdt_max_threads, - .psc_ctx_tags = LCT_DT_THREAD|LCT_MD_THREAD - }; - - m->mdt_fld_service = - ptlrpc_init_svc_conf(&conf, mdt_fld_handle, - LUSTRE_MDT_NAME"_fld", - procfs_entry, target_print_req, "mdt_fld"); - if (!m->mdt_fld_service) { - CERROR("failed to start fld service\n"); - GOTO(err_mdt_svc, rc = -ENOMEM); + .psc_buf = { + .bc_nbufs = MDS_NBUFS, + .bc_buf_size = MDS_BUFSIZE, + .bc_req_max_size = FLD_MAXREQSIZE, + .bc_rep_max_size = FLD_MAXREPSIZE, + .bc_req_portal = FLD_REQUEST_PORTAL, + .bc_rep_portal = MDC_REPLY_PORTAL, + }, + .psc_thr = { + .tc_thr_name = "mdt_fld", + .tc_nthrs_min = MDT_MIN_THREADS, + .tc_nthrs_max = MDT_MAX_THREADS, + .tc_nthrs_user = mdt_num_threads, + .tc_ctx_tags = LCT_DT_THREAD | LCT_MD_THREAD + }, + .psc_ops = { + .so_req_handler = mdt_fld_handle, + .so_req_printer = target_print_req, + }, + }; + m->mdt_fld_service = ptlrpc_register_service(&conf, procfs_entry); + if (IS_ERR(m->mdt_fld_service)) { + rc = PTR_ERR(m->mdt_fld_service); + CERROR("failed to start fld service: %d\n", rc); + m->mdt_fld_service = NULL; + + GOTO(err_mdt_svc, rc); + } + + /* + * mds-mds service configuration. Separate portal is used to allow + * mds-mds requests be not blocked during recovery. + */ + memset(&conf, 0, sizeof(conf)); + conf = (typeof(conf)) { + .psc_name = LUSTRE_MDT_NAME "_mds", + .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, + .psc_buf = { + .bc_nbufs = MDS_NBUFS, + .bc_buf_size = MDS_BUFSIZE, + .bc_req_max_size = MDS_MAXREQSIZE, + .bc_rep_max_size = MDS_MAXREPSIZE, + .bc_req_portal = MDS_MDS_PORTAL, + .bc_rep_portal = MDC_REPLY_PORTAL, + }, + .psc_thr = { + .tc_thr_name = "mdt_mds", + .tc_nthrs_min = MDT_MIN_THREADS, + .tc_nthrs_max = MDT_MAX_THREADS, + .tc_nthrs_user = mdt_num_threads, + .tc_ctx_tags = LCT_MD_THREAD, + }, + .psc_ops = { + .so_req_handler = mdt_xmds_handle, + .so_req_printer = target_print_req, + }, + }; + m->mdt_xmds_service = ptlrpc_register_service(&conf, procfs_entry); + if (IS_ERR(m->mdt_xmds_service)) { + rc = PTR_ERR(m->mdt_xmds_service); + CERROR("failed to start xmds service: %d\n", rc); + m->mdt_xmds_service = NULL; + + GOTO(err_mdt_svc, rc); } - rc = ptlrpc_start_threads(NULL, m->mdt_fld_service); - if (rc) - GOTO(err_mdt_svc, rc); - - /* - * mds-mds service configuration. Separate portal is used to allow - * mds-mds requests be not blocked during recovery. - */ - conf = (typeof(conf)) { - .psc_nbufs = MDS_NBUFS, - .psc_bufsize = MDS_BUFSIZE, - .psc_max_req_size = MDS_MAXREQSIZE, - .psc_max_reply_size = MDS_MAXREPSIZE, - .psc_req_portal = MDS_MDS_PORTAL, - .psc_rep_portal = MDC_REPLY_PORTAL, - .psc_watchdog_factor = MDT_SERVICE_WATCHDOG_FACTOR, - .psc_min_threads = mdt_min_threads, - .psc_max_threads = mdt_max_threads, - .psc_ctx_tags = LCT_MD_THREAD - }; - m->mdt_xmds_service = - ptlrpc_init_svc_conf(&conf, mdt_xmds_handle, - LUSTRE_MDT_NAME "_mds", - procfs_entry, target_print_req,"mdt_xmds"); - - if (m->mdt_xmds_service == NULL) { - CERROR("failed to start xmds service\n"); - GOTO(err_mdt_svc, rc = -ENOMEM); - } - - rc = ptlrpc_start_threads(NULL, m->mdt_xmds_service); - if (rc) - GOTO(err_mdt_svc, rc); - EXIT; err_mdt_svc: if (rc) @@ -4206,8 +4325,11 @@ static struct lu_device *mdt_layer_setup(struct lu_env *env, lu_device_get(d); lu_ref_add(&d->ld_reference, "lu-stack", &lu_site_init); - RETURN(d); + cfs_spin_lock(&d->ld_site->ls_ld_lock); + cfs_list_add_tail(&d->ld_linkage, &d->ld_site->ls_ld_linkage); + cfs_spin_unlock(&d->ld_site->ls_ld_lock); + RETURN(d); out_alloc: ldt->ldt_ops->ldto_device_free(env, d); type->typ_refcnt--; @@ -4344,6 +4466,9 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) mdt_obd_llog_cleanup(obd); obd_exports_barrier(obd); obd_zombie_barrier(); + + mdt_procfs_fini(m); + #ifdef HAVE_QUOTA_SUPPORT next->md_ops->mdo_quota.mqo_cleanup(env, next); #endif @@ -4379,10 +4504,6 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m) */ mdt_stack_fini(env, m, md2lu_dev(m->mdt_child)); - lprocfs_free_per_client_stats(obd); - lprocfs_free_obd_stats(obd); - mdt_procfs_fini(m); - if (ls) { struct md_site *mite; @@ -4420,49 +4541,6 @@ static int mdt_adapt_sptlrpc_conf(struct obd_device *obd, int initial) return 0; } -static void fsoptions_to_mdt_flags(struct mdt_device *m, char *options) -{ - char *p = options; - - m->mdt_opts.mo_mds_capa = 1; - m->mdt_opts.mo_oss_capa = 1; -#ifdef CONFIG_FS_POSIX_ACL - /* ACLs should be enabled by default (b=13829) */ - m->mdt_opts.mo_acl = 1; - LCONSOLE_INFO("Enabling ACL\n"); -#else - m->mdt_opts.mo_acl = 0; - LCONSOLE_INFO("Disabling ACL\n"); -#endif - - if (!options) - return; - - while (*options) { - int len; - - while (*p && *p != ',') - p++; - - len = p - options; - if ((len == sizeof("user_xattr") - 1) && - (memcmp(options, "user_xattr", len) == 0)) { - m->mdt_opts.mo_user_xattr = 1; - LCONSOLE_INFO("Enabling user_xattr\n"); - } else if ((len == sizeof("nouser_xattr") - 1) && - (memcmp(options, "nouser_xattr", len) == 0)) { - m->mdt_opts.mo_user_xattr = 0; - LCONSOLE_INFO("Disabling user_xattr\n"); - } else if ((len == sizeof("noacl") - 1) && - (memcmp(options, "noacl", len) == 0)) { - m->mdt_opts.mo_acl = 0; - LCONSOLE_INFO("Disabling ACL\n"); - } - - options = ++p; - } -} - int mdt_postrecov(const struct lu_env *, struct mdt_device *); static int mdt_init0(const struct lu_env *env, struct mdt_device *m, @@ -4478,11 +4556,10 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, struct lu_site *s; struct md_site *mite; const char *identity_upcall = "NONE"; -#ifdef HAVE_QUOTA_SUPPORT struct md_device *next; -#endif int rc; int node_id; + mntopt_t mntopts; ENTRY; md_device_init(&m->mdt_md_dev, ldt); @@ -4504,12 +4581,11 @@ 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; + m->mdt_max_mdsize = MAX_MD_SIZE; /* 4 stripes */ m->mdt_max_cookiesize = sizeof(struct llog_cookie); + m->mdt_som_conf = 0; - m->mdt_opts.mo_user_xattr = 0; - m->mdt_opts.mo_acl = 0; m->mdt_opts.mo_cos = MDT_COS_DEFAULT; lmi = server_get_mount_2(dev); if (lmi == NULL) { @@ -4517,7 +4593,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, RETURN(-EFAULT); } else { lsi = s2lsi(lmi->lmi_sb); - fsoptions_to_mdt_flags(m, lsi->lsi_lmd->lmd_opts); /* CMD is supported only in IAM mode */ ldd = lsi->lsi_ldd; LASSERT(num); @@ -4526,14 +4601,8 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, CERROR("CMD Operation not allowed in IOP mode\n"); GOTO(err_lmi, rc = -EINVAL); } - /* Read recovery timeouts */ - if (lsi->lsi_lmd && lsi->lsi_lmd->lmd_recovery_time_soft) - obd->obd_recovery_timeout = - lsi->lsi_lmd->lmd_recovery_time_soft; - if (lsi->lsi_lmd && lsi->lsi_lmd->lmd_recovery_time_hard) - obd->obd_recovery_time_hard = - lsi->lsi_lmd->lmd_recovery_time_hard; + obd->u.obt.obt_magic = OBT_MAGIC; } cfs_rwlock_init(&m->mdt_sptlrpc_lock); @@ -4541,6 +4610,8 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, cfs_spin_lock_init(&m->mdt_ioepoch_lock); m->mdt_opts.mo_compat_resname = 0; + m->mdt_opts.mo_mds_capa = 1; + m->mdt_opts.mo_oss_capa = 1; m->mdt_capa_timeout = CAPA_TIMEOUT; m->mdt_capa_alg = CAPA_HMAC_ALG_SHA1; m->mdt_ck_timeout = CAPA_KEY_TIMEOUT; @@ -4568,12 +4639,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, GOTO(err_free_site, rc); } - rc = mdt_procfs_init(m, dev); - if (rc) { - CERROR("Can't init MDT lprocfs, rc %d\n", rc); - GOTO(err_fini_proc, rc); - } - /* set server index */ lu_site2md(s)->ms_node_id = node_id; @@ -4597,7 +4662,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, rc = mdt_stack_init((struct lu_env *)env, m, cfg, lmi); if (rc) { CERROR("Can't init device stack, rc %d\n", rc); - GOTO(err_fini_proc, rc); + GOTO(err_lu_site, rc); } rc = lut_init(env, &m->mdt_lut, obd, m->mdt_bottom); @@ -4616,7 +4681,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); @@ -4624,19 +4690,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, /* set obd_namespace for compatibility with old code */ obd->obd_namespace = m->mdt_namespace; - /* XXX: to support suppgid for ACL, we enable identity_upcall - * by default, otherwise, maybe got unexpected -EACCESS. */ - if (m->mdt_opts.mo_acl) - identity_upcall = MDT_IDENTITY_UPCALL_PATH; - - m->mdt_identity_cache = upcall_cache_init(obd->obd_name, identity_upcall, - &mdt_identity_upcall_cache_ops); - if (IS_ERR(m->mdt_identity_cache)) { - rc = PTR_ERR(m->mdt_identity_cache); - m->mdt_identity_cache = NULL; - GOTO(err_free_ns, rc); - } - cfs_timer_init(&m->mdt_ck_timer, mdt_ck_timer_callback, m); rc = mdt_ck_thread_start(m); @@ -4657,8 +4710,8 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, mdt_adapt_sptlrpc_conf(obd, 1); -#ifdef HAVE_QUOTA_SUPPORT next = m->mdt_child; +#ifdef HAVE_QUOTA_SUPPORT rc = next->md_ops->mdo_quota.mqo_setup(env, next, lmi->lmi_mnt); if (rc) GOTO(err_llog_cleanup, rc); @@ -4667,11 +4720,45 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, server_put_mount_2(dev, lmi->lmi_mnt); lmi = NULL; + rc = next->md_ops->mdo_iocontrol(env, next, OBD_IOC_GET_MNTOPT, 0, + &mntopts); + if (rc) + GOTO(err_quota, rc); + + if (mntopts & MNTOPT_USERXATTR) + m->mdt_opts.mo_user_xattr = 1; + else + m->mdt_opts.mo_user_xattr = 0; + + if (mntopts & MNTOPT_ACL) + m->mdt_opts.mo_acl = 1; + else + m->mdt_opts.mo_acl = 0; + + /* XXX: to support suppgid for ACL, we enable identity_upcall + * by default, otherwise, maybe got unexpected -EACCESS. */ + if (m->mdt_opts.mo_acl) + identity_upcall = MDT_IDENTITY_UPCALL_PATH; + + m->mdt_identity_cache = upcall_cache_init(obd->obd_name,identity_upcall, + &mdt_identity_upcall_cache_ops); + if (IS_ERR(m->mdt_identity_cache)) { + rc = PTR_ERR(m->mdt_identity_cache); + m->mdt_identity_cache = NULL; + GOTO(err_quota, rc); + } + target_recovery_init(&m->mdt_lut, mdt_recovery_handle); + rc = mdt_procfs_init(m, dev); + if (rc) { + CERROR("Can't init MDT lprocfs, rc %d\n", rc); + GOTO(err_recovery, rc); + } + rc = mdt_start_ptlrpc_service(m); if (rc) - GOTO(err_recovery, rc); + GOTO(err_procfs, rc); ping_evictor_start(); @@ -4695,8 +4782,13 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m, err_stop_service: ping_evictor_stop(); mdt_stop_ptlrpc_service(m); +err_procfs: + mdt_procfs_fini(m); err_recovery: target_recovery_fini(obd); + upcall_cache_cleanup(m->mdt_identity_cache); + m->mdt_identity_cache = NULL; +err_quota: #ifdef HAVE_QUOTA_SUPPORT next->md_ops->mdo_quota.mqo_cleanup(env, next); #endif @@ -4709,8 +4801,6 @@ err_capa: cfs_timer_disarm(&m->mdt_ck_timer); mdt_ck_thread_stop(m); err_free_ns: - upcall_cache_cleanup(m->mdt_identity_cache); - m->mdt_identity_cache = NULL; ldlm_namespace_free(m->mdt_namespace, NULL, 0); obd->obd_namespace = m->mdt_namespace = NULL; err_fini_seq: @@ -4721,8 +4811,7 @@ err_lut: lut_fini(env, &m->mdt_lut); err_fini_stack: mdt_stack_fini(env, m, md2lu_dev(m->mdt_child)); -err_fini_proc: - mdt_procfs_fini(m); +err_lu_site: lu_site_fini(s); err_free_site: OBD_FREE_PTR(mite); @@ -4749,15 +4838,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); @@ -4806,7 +4901,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_mutex_init(&mo->mot_ioepoch_mutex); + cfs_mutex_init(&mo->mot_lov_mutex); RETURN(o); } else RETURN(NULL); @@ -4871,7 +4967,8 @@ static const struct lu_object_operations mdt_obj_ops = { .loo_object_print = mdt_object_print }; -static int mdt_obd_set_info_async(struct obd_export *exp, +static int mdt_obd_set_info_async(const struct lu_env *env, + struct obd_export *exp, __u32 keylen, void *key, __u32 vallen, void *val, struct ptlrpc_request_set *set) @@ -4915,6 +5012,24 @@ static int mdt_connect_internal(struct obd_export *exp, if (!mdt->mdt_som_conf) data->ocd_connect_flags &= ~OBD_CONNECT_SOM; + if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) { + data->ocd_brw_size = min(data->ocd_brw_size, + (__u32)(PTLRPC_MAX_BRW_PAGES << CFS_PAGE_SHIFT)); + if (data->ocd_brw_size == 0) { + CERROR("%s: cli %s/%p ocd_connect_flags: "LPX64 + " ocd_version: %x ocd_grant: %d " + "ocd_index: %u ocd_brw_size is " + "unexpectedly zero, network data " + "corruption? Refusing connection of this" + " client\n", + exp->exp_obd->obd_name, + exp->exp_client_uuid.uuid, + exp, data->ocd_connect_flags, data->ocd_version, + data->ocd_grant, data->ocd_index); + return -EPROTO; + } + } + cfs_spin_lock(&exp->exp_lock); exp->exp_connect_flags = data->ocd_connect_flags; cfs_spin_unlock(&exp->exp_lock); @@ -4999,7 +5114,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; @@ -5026,30 +5140,28 @@ 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_target_data.ted_lcd = lcd; - rc = mdt_client_new(env, mdt); - if (rc == 0) - mdt_export_stats_init(obd, lexp, localdata); - } else { - rc = -ENOMEM; - } + struct lsd_client_data *lcd = lexp->exp_target_data.ted_lcd; + + LASSERT(lcd); + info->mti_exp = lexp; + memcpy(lcd->lcd_uuid, cluuid, sizeof lcd->lcd_uuid); + rc = lut_client_new(env, lexp); + 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); } @@ -5083,6 +5195,7 @@ static int mdt_obd_reconnect(const struct lu_env *env, RETURN(rc); } + static int mdt_export_cleanup(struct obd_export *exp) { struct mdt_export_data *med = &exp->exp_mdt_data; @@ -5126,12 +5239,12 @@ static int mdt_export_cleanup(struct obd_export *exp) int cookie_size; lmm_size = mdt->mdt_max_mdsize; - OBD_ALLOC(ma->ma_lmm, lmm_size); + OBD_ALLOC_LARGE(ma->ma_lmm, lmm_size); if (ma->ma_lmm == NULL) GOTO(out_lmm, rc = -ENOMEM); cookie_size = mdt->mdt_max_cookiesize; - OBD_ALLOC(ma->ma_cookie, cookie_size); + OBD_ALLOC_LARGE(ma->ma_cookie, cookie_size); if (ma->ma_cookie == NULL) GOTO(out_cookie, rc = -ENOMEM); @@ -5144,22 +5257,25 @@ static int mdt_export_cleanup(struct obd_export *exp) ma->ma_need = 0; /* It is not for setattr, just tell MDD to send * DESTROY RPC to OSS if needed */ - ma->ma_attr_flags = MDS_CLOSE_CLEANUP; ma->ma_valid = MA_FLAGS; + ma->ma_attr_flags = MDS_CLOSE_CLEANUP; + /* Don't unlink orphan on failover umount, LU-184 */ + if (exp->exp_flags & OBD_OPT_FAILOVER) + ma->ma_attr_flags |= MDS_KEEP_ORPHAN; mdt_mfd_close(info, mfd); } - OBD_FREE(ma->ma_cookie, cookie_size); + OBD_FREE_LARGE(ma->ma_cookie, cookie_size); ma->ma_cookie = NULL; out_cookie: - OBD_FREE(ma->ma_lmm, lmm_size); + OBD_FREE_LARGE(ma->ma_lmm, lmm_size); ma->ma_lmm = NULL; } out_lmm: info->mti_mdt = NULL; /* cleanup client slot early */ /* Do not erase record for recoverable client. */ - if (!obd->obd_fail || exp->exp_failed) - mdt_client_del(&env, mdt); + if (!(exp->exp_flags & OBD_OPT_FAILOVER) || exp->exp_failed) + lut_client_del(&env, exp); lu_env_fini(&env); RETURN(rc); @@ -5191,37 +5307,56 @@ static int mdt_init_export(struct obd_export *exp) CFS_INIT_LIST_HEAD(&med->med_open_head); cfs_spin_lock_init(&med->med_open_lock); - cfs_sema_init(&med->med_idmap_sem, 1); + cfs_mutex_init(&med->med_idmap_mutex); med->med_idmap = NULL; cfs_spin_lock(&exp->exp_lock); exp->exp_connecting = 1; cfs_spin_unlock(&exp->exp_lock); - rc = ldlm_init_export(exp); + + /* self-export doesn't need client data and ldlm initialization */ + if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid, + &exp->exp_client_uuid))) + RETURN(0); + + rc = lut_client_alloc(exp); if (rc) - CERROR("Error %d while initializing export\n", rc); + GOTO(err, rc); + + rc = ldlm_init_export(exp); + if (rc) + GOTO(err_free, rc); + RETURN(rc); + +err_free: + lut_client_free(exp); +err: + CERROR("%s: Error %d while initializing export\n", + exp->exp_obd->obd_name, rc); + return rc; } static int mdt_destroy_export(struct obd_export *exp) { - struct mdt_export_data *med; - int rc = 0; ENTRY; - med = &exp->exp_mdt_data; if (exp_connect_rmtclient(exp)) mdt_cleanup_idmap(&exp->exp_mdt_data); target_destroy_export(exp); + /* destroy can be called from failed obd_setup, so + * checking uuid is safer than obd_self_export */ + if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid, + &exp->exp_client_uuid))) + RETURN(0); + ldlm_destroy_export(exp); + lut_client_free(exp); LASSERT(cfs_list_empty(&exp->exp_outstanding_replies)); LASSERT(cfs_list_empty(&exp->exp_mdt_data.med_open_head)); - if (obd_uuid_equals(&exp->exp_client_uuid, &exp->exp_obd->obd_uuid)) - RETURN(0); - lut_client_free(exp); - RETURN(rc); + RETURN(0); } static void mdt_allow_cli(struct mdt_device *m, unsigned int flag) @@ -5236,12 +5371,12 @@ static void mdt_allow_cli(struct mdt_device *m, unsigned int flag) if (cfs_test_bit(MDT_FL_CFGLOG, &m->mdt_state) && cfs_test_bit(MDT_FL_SYNCED, &m->mdt_state)) { struct obd_device *obd = m->mdt_md_dev.md_lu_dev.ld_obd; - + /* Open for clients */ if (obd->obd_no_conn) { - cfs_spin_lock_bh(&obd->obd_processing_task_lock); + cfs_spin_lock(&obd->obd_dev_lock); obd->obd_no_conn = 0; - cfs_spin_unlock_bh(&obd->obd_processing_task_lock); + cfs_spin_unlock(&obd->obd_dev_lock); } } } @@ -5264,7 +5399,8 @@ static int mdt_upcall(const struct lu_env *env, struct md_device *md, m->mdt_max_mdsize, m->mdt_max_cookiesize); mdt_allow_cli(m, CONFIG_SYNC); if (data) - (*(__u64 *)data) = m->mdt_lut.lut_mount_count; + (*(__u64 *)data) = + m->mdt_lut.lut_obd->u.obt.obt_mount_count; break; case MD_NO_TRANS: mti = lu_context_key_get(&env->le_ctx, &mdt_thread_key); @@ -5290,11 +5426,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 @@ -5352,7 +5488,7 @@ static int mdt_fid2path(const struct lu_env *env, struct mdt_device *mdt, obj = mdt_object_find(env, mdt, &fp->gf_fid); if (obj == NULL || IS_ERR(obj)) { - CDEBUG(D_IOCTL, "no object "DFID": %ld\n",PFID(&fp->gf_fid), + CDEBUG(D_IOCTL, "no object "DFID": %ld\n", PFID(&fp->gf_fid), PTR_ERR(obj)); RETURN(-EINVAL); } @@ -5452,6 +5588,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); @@ -5472,16 +5609,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; - } + version = dt_version_get(mti->mti_env, mdt_obj2dt(obj)); + *(__u64 *)data->ioc_inlbuf2 = version; + rc = 0; } mdt_object_unlock_put(mti, obj, lh, 1); RETURN(rc); @@ -5508,7 +5641,7 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len, rc = mdt_device_sync(&env, mdt); break; case OBD_IOC_SET_READONLY: - dt->dd_ops->dt_ro(&env, dt); + rc = dt->dd_ops->dt_ro(&env, dt); break; case OBD_IOC_ABORT_RECOVERY: CERROR("Aborting recovery for device %s\n", obd->obd_name); @@ -5544,16 +5677,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 @@ -5697,15 +5826,6 @@ LU_KEY_INIT_FINI(mdt, struct mdt_thread_info); /* context key: mdt_thread_key */ LU_CONTEXT_KEY_DEFINE(mdt, LCT_MD_THREAD); -/* context key constructor/destructor: mdt_txn_key_init, mdt_txn_key_fini */ -LU_KEY_INIT_FINI(mdt_txn, struct mdt_txn_info); - -struct lu_context_key mdt_txn_key = { - .lct_tags = LCT_TX_HANDLE, - .lct_init = mdt_txn_key_init, - .lct_fini = mdt_txn_key_fini -}; - struct md_ucred *mdt_ucred(const struct mdt_thread_info *info) { return md_ucred(info->mti_env); @@ -5725,7 +5845,7 @@ void mdt_enable_cos(struct mdt_device *mdt, int val) int rc; mdt->mdt_opts.mo_cos = !!val; - rc = lu_env_init(&env, LCT_MD_THREAD); + rc = lu_env_init(&env, LCT_LOCAL); if (unlikely(rc != 0)) { CWARN("lu_env initialization failed with rc = %d," "cannot sync\n", rc); @@ -5748,7 +5868,7 @@ int mdt_cos_is_enabled(struct mdt_device *mdt) } /* type constructor/destructor: mdt_type_init, mdt_type_fini */ -LU_TYPE_INIT_FINI(mdt, &mdt_thread_key, &mdt_txn_key); +LU_TYPE_INIT_FINI(mdt, &mdt_thread_key); static struct lu_device_type_operations mdt_device_type_ops = { .ldto_init = mdt_type_init, @@ -5769,32 +5889,11 @@ static struct lu_device_type mdt_device_type = { .ldt_ctx_tags = LCT_MD_THREAD }; -static struct lu_local_obj_desc mdt_last_recv = { - .llod_name = LAST_RCVD, - .llod_oid = MDT_LAST_RECV_OID, - .llod_is_index = 0, -}; - static int __init mdt_mod_init(void) { struct lprocfs_static_vars lvars; int rc; - llo_local_obj_register(&mdt_last_recv); - - 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; - if (mdt_min_threads < MDT_NUM_THREADS) - mdt_min_threads = MDT_NUM_THREADS; - } - lprocfs_mdt_init_vars(&lvars); rc = class_register_type(&mdt_obd_device_ops, NULL, lvars.module_vars, LUSTRE_MDT_NAME, @@ -5805,7 +5904,6 @@ static int __init mdt_mod_init(void) static void __exit mdt_mod_exit(void) { - llo_local_obj_unregister(&mdt_last_recv); class_unregister_type(LUSTRE_MDT_NAME); }