X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Ffid%2Ffid_handler.c;h=98c94f16a7920a5a4c905ea03628e95e3046c5ed;hb=05300cbf466d97368b00b49d57839186e5662687;hp=becd6da910586ed4c618e9f5ab362186a1181f57;hpb=eb5923c2d24f40a4116495c1c59eb9a6e2ed55a1;p=fs%2Flustre-release.git

diff --git a/lustre/fid/fid_handler.c b/lustre/fid/fid_handler.c
index becd6da..98c94f1 100644
--- a/lustre/fid/fid_handler.c
+++ b/lustre/fid/fid_handler.c
@@ -27,7 +27,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, 2015, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -47,7 +47,6 @@
 #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>
@@ -56,9 +55,8 @@
 static void seq_server_proc_fini(struct lu_server_seq *seq);
 
 /* Assigns client to sequence controller node. */
-int seq_server_set_cli(struct lu_server_seq *seq,
-                       struct lu_client_seq *cli,
-                       const struct lu_env *env)
+int seq_server_set_cli(const struct lu_env *env, struct lu_server_seq *seq,
+		       struct lu_client_seq *cli)
 {
         int rc = 0;
         ENTRY;
@@ -69,12 +67,11 @@ int seq_server_set_cli(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 "
@@ -97,13 +94,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;
 }
 
 /**
@@ -121,9 +118,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);
@@ -153,6 +150,41 @@ int seq_server_alloc_super(struct lu_server_seq *seq,
         RETURN(rc);
 }
 
+int seq_server_alloc_spec(struct lu_server_seq *seq,
+			  struct lu_seq_range *spec,
+			  const struct lu_env *env)
+{
+	struct lu_seq_range *space = &seq->lss_space;
+	int rc = -ENOSPC;
+	ENTRY;
+
+	/*
+	 * In some cases (like recovery after a disaster)
+	 * we may need to allocate sequences manually
+	 * Notice some sequences can be lost if requested
+	 * range doesn't start at the beginning of current
+	 * free space. Also notice it's not possible now
+	 * to allocate sequences out of natural order.
+	 */
+	if (spec->lsr_start >= spec->lsr_end)
+		RETURN(-EINVAL);
+	if (spec->lsr_flags != LU_SEQ_RANGE_MDT &&
+	    spec->lsr_flags != LU_SEQ_RANGE_OST)
+		RETURN(-EINVAL);
+
+	mutex_lock(&seq->lss_mutex);
+	if (spec->lsr_start >= space->lsr_start) {
+		space->lsr_start = spec->lsr_end;
+		rc = seq_store_update(env, seq, spec, 1 /* sync */);
+
+		LCONSOLE_INFO("%s: "DRANGE" sequences allocated: rc = %d \n",
+			      seq->lss_name, PRANGE(spec), rc);
+	}
+	mutex_unlock(&seq->lss_mutex);
+
+	RETURN(rc);
+}
+
 static int __seq_set_init(const struct lu_env *env,
                             struct lu_server_seq *seq)
 {
@@ -183,62 +215,73 @@ static int __seq_set_init(const struct lu_env *env,
  * 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);
@@ -247,14 +290,48 @@ 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;
+
+			/* Insert it to the local FLDB */
+			fld = seq->lss_site->ss_server_fld;
+			mutex_lock(&fld->lsf_lock);
+			rc = fld_insert_entry(env, fld, space);
+			mutex_unlock(&fld->lsf_lock);
+		}
+	}
+
+	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) {
+		CERROR("%s: Allocated super-sequence failed: rc = %d\n",
+			seq->lss_name, rc);
+		RETURN(rc);
 	}
 
 	rc = range_alloc_set(env, out, seq);
@@ -365,42 +442,52 @@ 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)
 {
-#ifdef LPROCFS
+#ifdef CONFIG_PROC_FS
+	int rc;
+	ENTRY;
 
-        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_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);
-        }
+	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 = 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);
-        }
+	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);
+	RETURN(0);
 
 out_cleanup:
-        seq_server_proc_fini(seq);
-        return rc;
-#else /* LPROCFS */
+	seq_server_proc_fini(seq);
+	return rc;
+#else /* !CONFIG_PROC_FS */
 	return 0;
-#endif
+#endif /* CONFIG_PROC_FS */
 }
 
 static void seq_server_proc_fini(struct lu_server_seq *seq)
 {
-#ifdef LPROCFS
+#ifdef CONFIG_PROC_FS
         ENTRY;
         if (seq->lss_proc_dir != NULL) {
                 if (!IS_ERR(seq->lss_proc_dir))
@@ -408,15 +495,15 @@ static void seq_server_proc_fini(struct lu_server_seq *seq)
                 seq->lss_proc_dir = NULL;
         }
         EXIT;
-#endif /* LPROCFS */
+#endif /* CONFIG_PROC_FS */
 }
 
-int seq_server_init(struct lu_server_seq *seq,
+int seq_server_init(const struct lu_env *env,
+		    struct lu_server_seq *seq,
 		    struct dt_device *dev,
 		    const char *prefix,
 		    enum lu_mgr_type type,
-		    struct seq_server_site *ss,
-		    const struct lu_env *env)
+		    struct seq_server_site *ss)
 {
 	int rc, is_srv = (type == LUSTRE_SEQ_SERVER);
 	ENTRY;
@@ -440,11 +527,11 @@ int seq_server_init(struct lu_server_seq *seq,
 	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);
 
@@ -466,7 +553,6 @@ int seq_server_init(struct lu_server_seq *seq,
                         LUSTRE_SEQ_ZERO_RANGE:
                         LUSTRE_SEQ_SPACE_RANGE;
 
-		LASSERT(ss != NULL);
 		seq->lss_space.lsr_index = ss->ss_node_id;
 		LCONSOLE_INFO("%s: No data found "
 			      "on store. Initialize space\n",
@@ -483,12 +569,12 @@ int seq_server_init(struct lu_server_seq *seq,
                 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);
         if (rc)