X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Ffid%2Ffid_handler.c;h=0b898473abbfa094003b13b3c905b989307d2dfd;hp=95b25b9bb841d6709d4fa63febce5d5ca0073857;hb=1a7720934dfb3105afd2f025c953bea2167d4e5d;hpb=519a65ddc04673022124f421e4809f8a87f790d7

diff --git a/lustre/fid/fid_handler.c b/lustre/fid/fid_handler.c
index 95b25b9..0b89847 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,7 +23,7 @@
  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2011, 2013, Intel Corporation.
+ * Copyright (c) 2011, 2017, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -47,14 +43,11 @@
 #include <obd.h>
 #include <obd_class.h>
 #include <dt_object.h>
-#include <md_object.h>
 #include <obd_support.h>
 #include <lustre_req_layout.h>
 #include <lustre_fid.h>
 #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)
@@ -68,12 +61,11 @@ int seq_server_set_cli(const struct lu_env *env, struct lu_server_seq *seq,
          */
 	mutex_lock(&seq->lss_mutex);
 
-        if (cli == NULL) {
-                CDEBUG(D_INFO, "%s: Detached sequence client %s\n",
-                       seq->lss_name, cli->lcs_name);
-                seq->lss_cli = cli;
-                GOTO(out_up, rc = 0);
-        }
+	if (cli == NULL) {
+		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 "
@@ -96,13 +88,13 @@ 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;
 }
 
 /**
@@ -120,9 +112,9 @@ static int __seq_server_alloc_super(struct lu_server_seq *seq,
 	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);
@@ -213,66 +205,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 (OBD_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);
@@ -281,14 +284,14 @@ 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",
-			       seq->lss_name, rc);
+			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));
+		LASSERT(lu_seq_range_is_sane(space));
 		if (seq->lss_cli->lcs_srv == NULL) {
 			struct lu_server_fld *fld;
 
@@ -300,6 +303,40 @@ 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\n",
+					      seq->lss_name);
+		} 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",
@@ -334,6 +371,7 @@ static int seq_server_handle(struct lu_site *site,
 {
 	int rc;
 	struct seq_server_site *ss_site;
+	struct dt_device *dev;
 	ENTRY;
 
 	ss_site = lu_site2seq(site);
@@ -345,6 +383,11 @@ static int seq_server_handle(struct lu_site *site,
 			       "initialized\n");
 			RETURN(-EINVAL);
 		}
+
+		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:
@@ -353,6 +396,11 @@ static int seq_server_handle(struct lu_site *site,
 			       "initialized\n");
 			RETURN(-EINVAL);
 		}
+
+		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:
@@ -398,7 +446,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);
 
@@ -408,62 +456,50 @@ 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 int seq_server_proc_init(struct lu_server_seq *seq)
+static void seq_server_debugfs_fini(struct lu_server_seq *seq)
 {
-#ifdef LPROCFS
+	if (!IS_ERR_OR_NULL(seq->lss_debugfs_entry))
+		ldebugfs_remove(&seq->lss_debugfs_entry);
+}
 
-        int rc;
-        ENTRY;
+static int seq_server_debugfs_init(struct lu_server_seq *seq)
+{
+	int rc;
+	ENTRY;
 
-        seq->lss_proc_dir = lprocfs_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);
-        }
+	seq->lss_debugfs_entry = ldebugfs_register(seq->lss_name,
+						   seq_debugfs_dir,
+						   NULL, NULL);
+	if (IS_ERR_OR_NULL(seq->lss_debugfs_entry)) {
+		rc = seq->lss_debugfs_entry ? PTR_ERR(seq->lss_debugfs_entry)
+					    : -ENOMEM;
+		seq->lss_debugfs_entry = NULL;
+		RETURN(rc);
+	}
 
-        rc = lprocfs_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);
-        }
+	rc = ldebugfs_add_vars(seq->lss_debugfs_entry,
+			       seq_server_debugfs_list, seq);
+	if (rc) {
+		CERROR("%s: Can't init sequence manager debugfs, 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);
+		rc = ldebugfs_seq_create(seq->lss_debugfs_entry, "fldb", 0644,
+					 &seq_fld_debugfs_seq_fops, seq);
 		if (rc) {
-			CERROR("%s: Can't create fldb for sequence manager "
-			       "proc: rc = %d\n", seq->lss_name, rc);
+			CERROR("%s: Can't create fldb for sequence manager debugfs: rc = %d\n",
+			       seq->lss_name, rc);
 			GOTO(out_cleanup, rc);
 		}
 	}
 
-        RETURN(0);
+	RETURN(0);
 
 out_cleanup:
-        seq_server_proc_fini(seq);
-        return rc;
-#else /* LPROCFS */
-	return 0;
-#endif
-}
-
-static void seq_server_proc_fini(struct lu_server_seq *seq)
-{
-#ifdef LPROCFS
-        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 /* LPROCFS */
+	seq_server_debugfs_fini(seq);
+	return rc;
 }
 
 int seq_server_init(const struct lu_env *env,
@@ -495,11 +531,11 @@ int seq_server_init(const struct lu_env *env,
 	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);
 
@@ -537,14 +573,14 @@ int seq_server_init(const struct lu_env *env,
                 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));
-        }
+	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));
+	}
 
-        rc  = seq_server_proc_init(seq);
+	rc  = seq_server_debugfs_init(seq);
         if (rc)
                 GOTO(out, rc);
 
@@ -561,7 +597,7 @@ void seq_server_fini(struct lu_server_seq *seq,
 {
         ENTRY;
 
-        seq_server_proc_fini(seq);
+	seq_server_debugfs_fini(seq);
         seq_store_fini(seq, env);
 
         EXIT;