X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;ds=sidebyside;f=lustre%2Ffid%2Ffid_handler.c;h=06196e66b971ea430b91db51ab81517573a95971;hb=HEAD;hp=11a9a54e45b3c6e7c1183a132693382759515be9;hpb=7817e4c785d075aae76b635dcf799064590833b0;p=fs%2Flustre-release.git diff --git a/lustre/fid/fid_handler.c b/lustre/fid/fid_handler.c index 11a9a54..95684d9 100644 --- a/lustre/fid/fid_handler.c +++ b/lustre/fid/fid_handler.c @@ -15,11 +15,7 @@ * * You should have received a copy of the GNU General Public License * version 2 along with this program; If not, see - * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf - * - * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara, - * CA 95054 USA or visit www.sun.com if you need additional information or - * have any questions. + * http://www.gnu.org/licenses/gpl-2.0.html * * GPL HEADER END */ @@ -27,11 +23,10 @@ * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2011, 2014, Intel Corporation. + * Copyright (c) 2011, 2017, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ - * Lustre is a trademark of Sun Microsystems, Inc. * * lustre/fid/fid_handler.c * @@ -52,55 +47,53 @@ #include #include "fid_internal.h" -static void seq_server_proc_fini(struct lu_server_seq *seq); - /* Assigns client to sequence controller node. */ int seq_server_set_cli(const struct lu_env *env, struct lu_server_seq *seq, struct lu_client_seq *cli) { - int rc = 0; - ENTRY; + int rc = 0; + ENTRY; - /* - * Ask client for new range, assign that range to ->seq_space and write - * seq state to backing store should be atomic. - */ + /* + * Ask client for new range, assign that range to ->seq_space and write + * seq state to backing store should be atomic. + */ mutex_lock(&seq->lss_mutex); - if (cli == NULL) { + if (!cli) { CDEBUG(D_INFO, "%s: Detached sequence client\n", seq->lss_name); seq->lss_cli = NULL; GOTO(out_up, rc = 0); } - if (seq->lss_cli != NULL) { - CDEBUG(D_HA, "%s: Sequence controller is already " - "assigned\n", seq->lss_name); + if (seq->lss_cli) { + CDEBUG(D_HA, "%s: Sequence controller is already assigned\n", + seq->lss_name); GOTO(out_up, rc = -EEXIST); } - CDEBUG(D_INFO, "%s: Attached sequence controller %s\n", - seq->lss_name, cli->lcs_name); + CDEBUG(D_INFO, "%s: Attached sequence controller %s\n", + seq->lss_name, cli->lcs_name); seq->lss_cli = cli; cli->lcs_space.lsr_index = seq->lss_site->ss_node_id; EXIT; out_up: mutex_unlock(&seq->lss_mutex); - return rc; + return rc; } EXPORT_SYMBOL(seq_server_set_cli); /* * allocate \a w units of sequence from range \a from. */ static inline void range_alloc(struct lu_seq_range *to, - struct lu_seq_range *from, - __u64 width) + struct lu_seq_range *from, + __u64 width) { - width = min(range_space(from), width); - to->lsr_start = from->lsr_start; - to->lsr_end = from->lsr_start + width; - from->lsr_start += width; + width = min(lu_seq_range_space(from), width); + to->lsr_start = from->lsr_start; + to->lsr_end = from->lsr_start + width; + from->lsr_start += width; } /** @@ -111,16 +104,16 @@ static inline void range_alloc(struct lu_seq_range *to, */ static int __seq_server_alloc_super(struct lu_server_seq *seq, - struct lu_seq_range *out, - const struct lu_env *env) + struct lu_seq_range *out, + const struct lu_env *env) { struct lu_seq_range *space = &seq->lss_space; int rc; ENTRY; - LASSERT(range_is_sane(space)); + LASSERT(lu_seq_range_is_sane(space)); - if (range_is_exhausted(space)) { + if (lu_seq_range_is_exhausted(space)) { CERROR("%s: Sequences space is exhausted\n", seq->lss_name); RETURN(-ENOSPC); @@ -137,17 +130,17 @@ static int __seq_server_alloc_super(struct lu_server_seq *seq, } int seq_server_alloc_super(struct lu_server_seq *seq, - struct lu_seq_range *out, - const struct lu_env *env) + struct lu_seq_range *out, + const struct lu_env *env) { - int rc; - ENTRY; + int rc; + ENTRY; mutex_lock(&seq->lss_mutex); - rc = __seq_server_alloc_super(seq, out, env); + rc = __seq_server_alloc_super(seq, out, env); mutex_unlock(&seq->lss_mutex); - RETURN(rc); + RETURN(rc); } int seq_server_alloc_spec(struct lu_server_seq *seq, @@ -186,17 +179,17 @@ int seq_server_alloc_spec(struct lu_server_seq *seq, } static int __seq_set_init(const struct lu_env *env, - struct lu_server_seq *seq) + struct lu_server_seq *seq) { - struct lu_seq_range *space = &seq->lss_space; - int rc; + struct lu_seq_range *space = &seq->lss_space; + int rc; - range_alloc(&seq->lss_lowater_set, space, seq->lss_set_width); - range_alloc(&seq->lss_hiwater_set, space, seq->lss_set_width); + range_alloc(&seq->lss_lowater_set, space, seq->lss_set_width); + range_alloc(&seq->lss_hiwater_set, space, seq->lss_set_width); - rc = seq_store_update(env, seq, NULL, 1); + rc = seq_store_update(env, seq, NULL, 1); - return rc; + return rc; } /* @@ -211,66 +204,77 @@ static int __seq_set_init(const struct lu_env *env, * * when lss_lowater_set reaches the end it is replaced with hiwater one and * a write operation is initiated to allocate new hiwater range. - * if last seq write opearion is still not commited, current operation is + * if last seq write opearion is still not committed, current operation is * flaged as sync write op. */ static int range_alloc_set(const struct lu_env *env, - struct lu_seq_range *out, - struct lu_server_seq *seq) + struct lu_seq_range *out, + struct lu_server_seq *seq) { - struct lu_seq_range *space = &seq->lss_space; - struct lu_seq_range *loset = &seq->lss_lowater_set; - struct lu_seq_range *hiset = &seq->lss_hiwater_set; - int rc = 0; + struct lu_seq_range *space = &seq->lss_space; + struct lu_seq_range *loset = &seq->lss_lowater_set; + struct lu_seq_range *hiset = &seq->lss_hiwater_set; + int rc = 0; - if (range_is_zero(loset)) - __seq_set_init(env, seq); + if (lu_seq_range_is_zero(loset)) + __seq_set_init(env, seq); - if (OBD_FAIL_CHECK(OBD_FAIL_SEQ_ALLOC)) /* exhaust set */ - loset->lsr_start = loset->lsr_end; + if (CFS_FAIL_CHECK(OBD_FAIL_SEQ_ALLOC)) /* exhaust set */ + loset->lsr_start = loset->lsr_end; - if (range_is_exhausted(loset)) { - /* reached high water mark. */ + if (lu_seq_range_is_exhausted(loset)) { + /* reached high water mark. */ struct lu_device *dev = seq->lss_site->ss_lu->ls_top_dev; - int obd_num_clients = dev->ld_obd->obd_num_exports; - __u64 set_sz; + int obd_num_clients = dev->ld_obd->obd_num_exports; + __u64 set_sz; - /* calculate new seq width based on number of clients */ - set_sz = max(seq->lss_set_width, - obd_num_clients * seq->lss_width); - set_sz = min(range_space(space), set_sz); + /* calculate new seq width based on number of clients */ + set_sz = max(seq->lss_set_width, + obd_num_clients * seq->lss_width); + set_sz = min(lu_seq_range_space(space), set_sz); - /* Switch to hiwater range now */ - *loset = *hiset; - /* allocate new hiwater range */ - range_alloc(hiset, space, set_sz); + /* Switch to hiwater range now */ + *loset = *hiset; + /* allocate new hiwater range */ + range_alloc(hiset, space, set_sz); - /* update ondisk seq with new *space */ - rc = seq_store_update(env, seq, NULL, seq->lss_need_sync); - } + /* update ondisk seq with new *space */ + rc = seq_store_update(env, seq, NULL, seq->lss_need_sync); + } - LASSERTF(!range_is_exhausted(loset) || range_is_sane(loset), - DRANGE"\n", PRANGE(loset)); + LASSERTF(!lu_seq_range_is_exhausted(loset) || + lu_seq_range_is_sane(loset), + DRANGE"\n", PRANGE(loset)); - if (rc == 0) - range_alloc(out, loset, seq->lss_width); + if (rc == 0) + range_alloc(out, loset, seq->lss_width); - RETURN(rc); + RETURN(rc); } -static int __seq_server_alloc_meta(struct lu_server_seq *seq, - struct lu_seq_range *out, - const struct lu_env *env) +/** + * Check if the sequence server has sequence avaible + * + * Check if the sequence server has sequence avaible, if not, then + * allocating super sequence from sequence manager (MDT0). + * + * \param[in] env execution environment + * \param[in] seq server sequence + * + * \retval negative errno if allocating new sequence fails + * \retval 0 if there is enough sequence or allocating + * new sequence succeeds + */ +int seq_server_check_and_alloc_super(const struct lu_env *env, + struct lu_server_seq *seq) { struct lu_seq_range *space = &seq->lss_space; int rc = 0; ENTRY; - LASSERT(range_is_sane(space)); - /* Check if available space ends and allocate new super seq */ - if (range_is_exhausted(space)) { + if (lu_seq_range_is_exhausted(space)) { if (!seq->lss_cli) { CERROR("%s: No sequence controller is attached.\n", seq->lss_name); @@ -279,15 +283,16 @@ static int __seq_server_alloc_meta(struct lu_server_seq *seq, rc = seq_client_alloc_super(seq->lss_cli, env); if (rc) { - CERROR("%s: Can't allocate super-sequence, rc %d\n", + CDEBUG(D_HA, + "%s: Can't allocate super-sequence: rc = %d\n", seq->lss_name, rc); RETURN(rc); } /* Saving new range to allocation space. */ *space = seq->lss_cli->lcs_space; - LASSERT(range_is_sane(space)); - if (seq->lss_cli->lcs_srv == NULL) { + LASSERT(lu_seq_range_is_sane(space)); + if (!seq->lss_cli->lcs_srv) { struct lu_server_fld *fld; /* Insert it to the local FLDB */ @@ -298,40 +303,74 @@ static int __seq_server_alloc_meta(struct lu_server_seq *seq, } } + if (lu_seq_range_is_zero(&seq->lss_lowater_set)) + __seq_set_init(env, seq); + + RETURN(rc); +} +EXPORT_SYMBOL(seq_server_check_and_alloc_super); + +static int __seq_server_alloc_meta(struct lu_server_seq *seq, + struct lu_seq_range *out, + const struct lu_env *env) +{ + struct lu_seq_range *space = &seq->lss_space; + int rc = 0; + + ENTRY; + + LASSERT(lu_seq_range_is_sane(space)); + + rc = seq_server_check_and_alloc_super(env, seq); + if (rc < 0) { + if (rc == -EINPROGRESS) { + static int printed; + + if (printed++ % 8 == 0) + LCONSOLE_INFO("%s: Waiting to contact MDT0000 to allocate super-sequence: rc = %d\n", + seq->lss_name, rc); + } else { + CERROR("%s: Allocated super-sequence failed: rc = %d\n", + seq->lss_name, rc); + } + RETURN(rc); + } + rc = range_alloc_set(env, out, seq); if (rc != 0) { CERROR("%s: Allocated meta-sequence failed: rc = %d\n", - seq->lss_name, rc); + seq->lss_name, rc); RETURN(rc); } CDEBUG(D_INFO, "%s: Allocated meta-sequence " DRANGE"\n", - seq->lss_name, PRANGE(out)); + seq->lss_name, PRANGE(out)); RETURN(rc); } int seq_server_alloc_meta(struct lu_server_seq *seq, - struct lu_seq_range *out, - const struct lu_env *env) + struct lu_seq_range *out, + const struct lu_env *env) { - int rc; - ENTRY; + int rc; + ENTRY; mutex_lock(&seq->lss_mutex); - rc = __seq_server_alloc_meta(seq, out, env); + rc = __seq_server_alloc_meta(seq, out, env); mutex_unlock(&seq->lss_mutex); - RETURN(rc); + RETURN(rc); } EXPORT_SYMBOL(seq_server_alloc_meta); static int seq_server_handle(struct lu_site *site, - const struct lu_env *env, - __u32 opc, struct lu_seq_range *out) + const struct lu_env *env, + __u32 opc, struct lu_seq_range *out) { int rc; struct seq_server_site *ss_site; + struct dt_device *dev; ENTRY; ss_site = lu_site2seq(site); @@ -339,18 +378,30 @@ static int seq_server_handle(struct lu_site *site, switch (opc) { case SEQ_ALLOC_META: if (!ss_site->ss_server_seq) { - CERROR("Sequence server is not " - "initialized\n"); - RETURN(-EINVAL); + rc = -EINVAL; + CERROR("Sequence server is not initialized: rc = %d\n", + rc); + RETURN(rc); } + + dev = lu2dt_dev(ss_site->ss_server_seq->lss_obj->do_lu.lo_dev); + if (dev->dd_rdonly) + RETURN(-EROFS); + rc = seq_server_alloc_meta(ss_site->ss_server_seq, out, env); break; case SEQ_ALLOC_SUPER: if (!ss_site->ss_control_seq) { - CERROR("Sequence controller is not " - "initialized\n"); - RETURN(-EINVAL); + rc = -EINVAL; + CERROR("Sequence controller is not initialized: rc = %d\n", + rc); + RETURN(rc); } + + dev = lu2dt_dev(ss_site->ss_control_seq->lss_obj->do_lu.lo_dev); + if (dev->dd_rdonly) + RETURN(-EROFS); + rc = seq_server_alloc_super(ss_site->ss_control_seq, out, env); break; default: @@ -375,16 +426,17 @@ static int seq_handler(struct tgt_session_info *tsi) LASSERT(site != NULL); opc = req_capsule_client_get(tsi->tsi_pill, &RMF_SEQ_OPC); - if (opc != NULL) { + if (opc) { out = req_capsule_server_get(tsi->tsi_pill, &RMF_SEQ_RANGE); - if (out == NULL) + if (!out) RETURN(err_serious(-EPROTO)); tmp = req_capsule_client_get(tsi->tsi_pill, &RMF_SEQ_RANGE); - /* seq client passed mdt id, we need to pass that using out - * range parameter */ - + /* + * seq client passed mdt id, we need to pass that using out + * range parameter + */ out->lsr_index = tmp->lsr_index; out->lsr_flags = tmp->lsr_flags; rc = seq_server_handle(site, tsi->tsi_env, *opc, out); @@ -396,7 +448,7 @@ static int seq_handler(struct tgt_session_info *tsi) } struct tgt_handler seq_handlers[] = { -TGT_SEQ_HDL(HABEO_REFERO, SEQ_QUERY, seq_handler), +TGT_SEQ_HDL(HAS_REPLY, SEQ_QUERY, seq_handler), }; EXPORT_SYMBOL(seq_handlers); @@ -406,61 +458,24 @@ LU_KEY_INIT_FINI(seq, struct seq_thread_info); /* context key: seq_thread_key */ LU_CONTEXT_KEY_DEFINE(seq, LCT_MD_THREAD | LCT_DT_THREAD); -extern const struct file_operations seq_fld_proc_seq_fops; +static void seq_server_debugfs_fini(struct lu_server_seq *seq) +{ + debugfs_remove_recursive(seq->lss_debugfs_entry); +} -static int seq_server_proc_init(struct lu_server_seq *seq) +static void seq_server_debugfs_init(struct lu_server_seq *seq) { -#ifdef CONFIG_PROC_FS - int rc; ENTRY; - seq->lss_proc_dir = lprocfs_seq_register(seq->lss_name, - seq_type_proc_dir, - NULL, NULL); - if (IS_ERR(seq->lss_proc_dir)) { - rc = PTR_ERR(seq->lss_proc_dir); - RETURN(rc); - } - - rc = lprocfs_seq_add_vars(seq->lss_proc_dir, - seq_server_proc_list, seq); - if (rc) { - CERROR("%s: Can't init sequence manager " - "proc, rc %d\n", seq->lss_name, rc); - GOTO(out_cleanup, rc); - } - - if (seq->lss_type == LUSTRE_SEQ_CONTROLLER) { - rc = lprocfs_seq_create(seq->lss_proc_dir, "fldb", 0644, - &seq_fld_proc_seq_fops, seq); - if (rc) { - CERROR("%s: Can't create fldb for sequence manager " - "proc: rc = %d\n", seq->lss_name, rc); - GOTO(out_cleanup, rc); - } - } - - RETURN(0); + seq->lss_debugfs_entry = debugfs_create_dir(seq->lss_name, + seq_debugfs_dir); -out_cleanup: - seq_server_proc_fini(seq); - return rc; -#else /* !CONFIG_PROC_FS */ - return 0; -#endif /* CONFIG_PROC_FS */ -} + ldebugfs_add_vars(seq->lss_debugfs_entry, + seq_server_debugfs_list, seq); -static void seq_server_proc_fini(struct lu_server_seq *seq) -{ -#ifdef CONFIG_PROC_FS - ENTRY; - if (seq->lss_proc_dir != NULL) { - if (!IS_ERR(seq->lss_proc_dir)) - lprocfs_remove(&seq->lss_proc_dir); - seq->lss_proc_dir = NULL; - } - EXIT; -#endif /* CONFIG_PROC_FS */ + if (seq->lss_type == LUSTRE_SEQ_CONTROLLER) + debugfs_create_file("fldb", 0644, seq->lss_debugfs_entry, + seq, &seq_fld_debugfs_seq_fops); } int seq_server_init(const struct lu_env *env, @@ -478,96 +493,93 @@ int seq_server_init(const struct lu_env *env, LASSERT(ss != NULL); LASSERT(ss->ss_lu != NULL); - /* A compile-time check for FIDs that used to be in lustre_idl.h - * but is moved here to remove CLASSERT/LASSERT in that header. + /* * Check all lu_fid fields are converted in fid_cpu_to_le() and friends - * and that there is no padding added by compiler to the struct. */ + * and that there is no padding added by compiler to the struct. + */ { struct lu_fid tst; - CLASSERT(sizeof(tst) == sizeof(tst.f_seq) + - sizeof(tst.f_oid) + sizeof(tst.f_ver)); + BUILD_BUG_ON(sizeof(tst) != sizeof(tst.f_seq) + + sizeof(tst.f_oid) + sizeof(tst.f_ver)); } seq->lss_cli = NULL; seq->lss_type = type; seq->lss_site = ss; - range_init(&seq->lss_space); + lu_seq_range_init(&seq->lss_space); - range_init(&seq->lss_lowater_set); - range_init(&seq->lss_hiwater_set); - seq->lss_set_width = LUSTRE_SEQ_BATCH_WIDTH; + lu_seq_range_init(&seq->lss_lowater_set); + lu_seq_range_init(&seq->lss_hiwater_set); + seq->lss_set_width = LUSTRE_SEQ_BATCH_WIDTH; mutex_init(&seq->lss_mutex); - seq->lss_width = is_srv ? - LUSTRE_SEQ_META_WIDTH : LUSTRE_SEQ_SUPER_WIDTH; + seq->lss_width = is_srv ? + LUSTRE_SEQ_META_WIDTH : LUSTRE_SEQ_SUPER_WIDTH; - snprintf(seq->lss_name, sizeof(seq->lss_name), - "%s-%s", (is_srv ? "srv" : "ctl"), prefix); + snprintf(seq->lss_name, sizeof(seq->lss_name), + "%s-%s", (is_srv ? "srv" : "ctl"), prefix); - rc = seq_store_init(seq, env, dev); - if (rc) - GOTO(out, rc); - /* Request backing store for saved sequence info. */ - rc = seq_store_read(seq, env); - if (rc == -ENODATA) { + rc = seq_store_init(seq, env, dev); + if (rc) + GOTO(out, rc); + /* Request backing store for saved sequence info. */ + rc = seq_store_read(seq, env); + if (rc == -ENODATA) { - /* Nothing is read, init by default value. */ - seq->lss_space = is_srv ? - LUSTRE_SEQ_ZERO_RANGE: - LUSTRE_SEQ_SPACE_RANGE; + /* Nothing is read, init by default value. */ + seq->lss_space = is_srv ? + LUSTRE_SEQ_ZERO_RANGE : + LUSTRE_SEQ_SPACE_RANGE; seq->lss_space.lsr_index = ss->ss_node_id; - LCONSOLE_INFO("%s: No data found " - "on store. Initialize space\n", + LCONSOLE_INFO("%s: No data found on store. Initialize space.\n", seq->lss_name); - rc = seq_store_update(env, seq, NULL, 0); - if (rc) { - CERROR("%s: Can't write space data, " - "rc %d\n", seq->lss_name, rc); - } - } else if (rc) { - CERROR("%s: Can't read space data, rc %d\n", - seq->lss_name, rc); - GOTO(out, rc); - } - - if (is_srv) { - LASSERT(range_is_sane(&seq->lss_space)); - } else { - LASSERT(!range_is_zero(&seq->lss_space) && - range_is_sane(&seq->lss_space)); - } - - rc = seq_server_proc_init(seq); - if (rc) - GOTO(out, rc); - - EXIT; + rc = seq_store_update(env, seq, NULL, 0); + if (rc) { + CERROR("%s: Can't write space data: rc = %d\n", + seq->lss_name, rc); + } + } else if (rc) { + CERROR("%s: Can't read space data: rc = %d\n", + seq->lss_name, rc); + GOTO(out, rc); + } + + if (is_srv) { + LASSERT(lu_seq_range_is_sane(&seq->lss_space)); + } else { + LASSERT(!lu_seq_range_is_zero(&seq->lss_space) && + lu_seq_range_is_sane(&seq->lss_space)); + } + + seq_server_debugfs_init(seq); + + EXIT; out: - if (rc) - seq_server_fini(seq, env); - return rc; + if (rc) + seq_server_fini(seq, env); + return rc; } EXPORT_SYMBOL(seq_server_init); void seq_server_fini(struct lu_server_seq *seq, - const struct lu_env *env) + const struct lu_env *env) { - ENTRY; + ENTRY; - seq_server_proc_fini(seq); - seq_store_fini(seq, env); + seq_server_debugfs_fini(seq); + seq_store_fini(seq, env); - EXIT; + EXIT; } EXPORT_SYMBOL(seq_server_fini); int seq_site_fini(const struct lu_env *env, struct seq_server_site *ss) { - if (ss == NULL) + if (!ss) RETURN(0); if (ss->ss_server_seq) {