LU-2240 mds: Assign special fid sequence to root.

[fs/lustre-release.git] / lustre / fld / fld_handler.c

diff --git a/lustre/fld/fld_handler.c b/lustre/fld/fld_handler.c
index 4e4404f..7fe5d7e 100644 (file)
--- a/lustre/fld/fld_handler.c
+++ b/lustre/fld/fld_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, 2012, Whamcloud, Inc.
+ * Copyright (c) 2011, 2013, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -72,7 +72,7 @@
 LU_KEY_INIT_FINI(fld, struct fld_thread_info);
 
 /* context key: fld_thread_key */
-LU_CONTEXT_KEY_DEFINE(fld, LCT_MD_THREAD|LCT_DT_THREAD);
+LU_CONTEXT_KEY_DEFINE(fld, LCT_MD_THREAD | LCT_DT_THREAD | LCT_MG_THREAD);
 
 cfs_proc_dir_entry_t *fld_type_proc_dir = NULL;
 
@@ -145,13 +145,15 @@ int fld_server_lookup(const struct lu_env *env, struct lu_server_fld *fld,
         int rc;
         ENTRY;
 
-        info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
-        erange = &info->fti_lrange;
+       info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
+       LASSERT(info != NULL);
+       erange = &info->fti_lrange;
 
         /* Lookup it in the cache. */
         rc = fld_cache_lookup(fld->lsf_cache, seq, erange);
         if (rc == 0) {
-                if (unlikely(erange->lsr_flags != range->lsr_flags)) {
+               if (unlikely(erange->lsr_flags != range->lsr_flags) &&
+                   range->lsr_flags != -1) {
                         CERROR("FLD cache found a range "DRANGE" doesn't "
                                "match the requested flag %x\n",
                                PRANGE(erange), range->lsr_flags);
@@ -173,6 +175,7 @@ int fld_server_lookup(const struct lu_env *env, struct lu_server_fld *fld,
                 * This is temporary solution, long term solution is fld
                 * replication on all mdt servers.
                 */
+               range->lsr_start = seq;
                rc = fld_client_rpc(fld->lsf_control_exp,
                                    range, FLD_LOOKUP);
                if (rc == 0)
@@ -236,16 +239,17 @@ static int fld_req_handle(struct ptlrpc_request *req,
                         RETURN(err_serious(-EPROTO));
                 *out = *in;
 
-                /* For old 2.0 client, the 'lsr_flags' is uninitialized.
-                 * Set it as 'LU_SEQ_RANGE_MDT' by default.
-                 * Old 2.0 liblustre client cannot talk with new 2.1 server. */
-                if (!(exp->exp_connect_flags & OBD_CONNECT_64BITHASH) &&
-                    !exp->exp_libclient)
-                        out->lsr_flags = LU_SEQ_RANGE_MDT;
-
-                rc = fld_server_handle(lu_site2md(site)->ms_server_fld,
-                                       req->rq_svc_thread->t_env,
-                                       *opc, out, info);
+               /* For old 2.0 client, the 'lsr_flags' is uninitialized.
+                * Set it as 'LU_SEQ_RANGE_MDT' by default. */
+               if (!(exp_connect_flags(exp) & OBD_CONNECT_64BITHASH) &&
+                   !(exp_connect_flags(exp) & OBD_CONNECT_MDS_MDS) &&
+                   !(exp_connect_flags(exp) & OBD_CONNECT_LIGHTWEIGHT) &&
+                   !exp->exp_libclient)
+                       out->lsr_flags = LU_SEQ_RANGE_MDT;
+
+               rc = fld_server_handle(lu_site2seq(site)->ss_server_fld,
+                                      req->rq_svc_thread->t_env,
+                                      *opc, out, info);
         } else
                 rc = err_serious(-EPROTO);
 
@@ -305,26 +309,26 @@ EXPORT_SYMBOL(fld_query);
 int fid_is_local(const struct lu_env *env,
                  struct lu_site *site, const struct lu_fid *fid)
 {
-        int result;
-        struct md_site *msite;
-        struct lu_seq_range *range;
-        struct fld_thread_info *info;
-        ENTRY;
+       int result;
+       struct seq_server_site *ss_site;
+       struct lu_seq_range *range;
+       struct fld_thread_info *info;
+       ENTRY;
 
-        info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
-        range = &info->fti_lrange;
+       info = lu_context_key_get(&env->le_ctx, &fld_thread_key);
+       range = &info->fti_lrange;
 
-        result = 1; /* conservatively assume fid is local */
-        msite = lu_site2md(site);
-        if (msite->ms_client_fld != NULL) {
-                int rc;
+       result = 1; /* conservatively assume fid is local */
+       ss_site = lu_site2seq(site);
+       if (ss_site->ss_client_fld != NULL) {
+               int rc;
 
-                rc = fld_cache_lookup(msite->ms_client_fld->lcf_cache,
-                                      fid_seq(fid), range);
-                if (rc == 0)
-                        result = (range->lsr_index == msite->ms_node_id);
-        }
-        return result;
+               rc = fld_cache_lookup(ss_site->ss_client_fld->lcf_cache,
+                                     fid_seq(fid), range);
+               if (rc == 0)
+                       result = (range->lsr_index == ss_site->ss_node_id);
+       }
+       return result;
 }
 EXPORT_SYMBOL(fid_is_local);
 
@@ -377,7 +381,8 @@ static void fld_server_proc_fini(struct lu_server_fld *fld)
 #endif
 
 int fld_server_init(const struct lu_env *env, struct lu_server_fld *fld,
-                   struct dt_device *dt, const char *prefix, int mds_node_id)
+                   struct dt_device *dt, const char *prefix, int mds_node_id,
+                   __u32 lsr_flags)
 {
         int cache_size, cache_threshold;
         struct lu_seq_range range;
@@ -402,7 +407,7 @@ int fld_server_init(const struct lu_env *env, struct lu_server_fld *fld,
                 GOTO(out, rc);
         }
 
-        if (!mds_node_id) {
+       if (!mds_node_id && lsr_flags == LU_SEQ_RANGE_MDT) {
                rc = fld_index_init(env, fld, dt);
                 if (rc)
                         GOTO(out, rc);
@@ -415,13 +420,15 @@ int fld_server_init(const struct lu_env *env, struct lu_server_fld *fld,
 
         fld->lsf_control_exp = NULL;
 
-        /* Insert reserved sequence number of ".lustre" into fld cache. */
-        range.lsr_start = FID_SEQ_DOT_LUSTRE;
-        range.lsr_end = FID_SEQ_DOT_LUSTRE + 1;
-        range.lsr_index = 0;
-        range.lsr_flags = LU_SEQ_RANGE_MDT;
-       fld_cache_insert(fld->lsf_cache, &range);
-
+       if (lsr_flags == LU_SEQ_RANGE_MDT) {
+               /* Insert reserved sequence of "ROOT" and ".lustre"
+                * into fld cache. */
+               range.lsr_start = FID_SEQ_LOCAL_FILE;
+               range.lsr_end = FID_SEQ_DOT_LUSTRE + 1;
+               range.lsr_index = 0;
+               range.lsr_flags = lsr_flags;
+               fld_cache_insert(fld->lsf_cache, &range);
+       }
         EXIT;
 out:
        if (rc)