- fixes in fld:

  * added exports list and chosing one of them using seq number and some hash function (round-robin is now in use, should be dht later);
  * small cleanups about ENTRY/EXIT macros using.

diff --git a/lustre/fld/fld_handler.c b/lustre/fld/fld_handler.c
index 8649359..b27da00 100644 (file)
--- a/lustre/fld/fld_handler.c
+++ b/lustre/fld/fld_handler.c
@@ -61,7 +61,7 @@ enum {
         FLD_HTABLE_MASK = FLD_HTABLE_SIZE - 1
 };
 
-static __u32 fld_hash(__u64 lu_seq)
+static __u32 fld_cache_hash(__u64 lu_seq)
 {
         return lu_seq;
 }
@@ -76,7 +76,7 @@ fld_cache_insert(struct fld_cache_info *fld_cache,
         int rc = 0;
         ENTRY;
 
-        bucket = fld_cache->fld_hash + (fld_hash(lu_seq) &
+        bucket = fld_cache->fld_hash + (fld_cache_hash(lu_seq) &
                                         fld_cache->fld_hash_mask);
 
         OBD_ALLOC_PTR(fld);
@@ -110,7 +110,7 @@ fld_cache_lookup(struct fld_cache_info *fld_cache, __u64 lu_seq)
         struct fld_cache *fld;
         ENTRY;
 
-        bucket = fld_cache->fld_hash + (fld_hash(lu_seq) &
+        bucket = fld_cache->fld_hash + (fld_cache_hash(lu_seq) &
                                         fld_cache->fld_hash_mask);
 
         spin_lock(&fld_cache->fld_lock);
@@ -133,7 +133,7 @@ fld_cache_delete(struct fld_cache_info *fld_cache, __u64 lu_seq)
         struct fld_cache *fld;
         ENTRY;
 
-        bucket = fld_cache->fld_hash + (fld_hash(lu_seq) &
+        bucket = fld_cache->fld_hash + (fld_cache_hash(lu_seq) &
                                         fld_cache->fld_hash_mask);
 
         spin_lock(&fld_cache->fld_lock);
@@ -150,43 +150,137 @@ fld_cache_delete(struct fld_cache_info *fld_cache, __u64 lu_seq)
 }
 #endif
 
-static int dht_mdt_hash(__u64 seq)
+static int fld_rrb_hash(struct lu_client_fld *fld, __u64 seq)
 {
-        return 0;
+        return seq % fld->fld_count;
 }
 
+static int fld_dht_hash(struct lu_client_fld *fld, __u64 seq)
+{
+        CWARN("using Round Robin hash func for while\n");
+        return seq % fld->fld_count;
+}
+
+static struct lu_fld_hash fld_hash[2] = {
+        {
+                .fh_name = "DHT",
+                .fh_func = fld_dht_hash
+        },
+        {
+                .fh_name = "Round Robin",
+                .fh_func = fld_rrb_hash
+        }        
+};
+
 static struct obd_export *
 fld_client_get_exp(struct lu_client_fld *fld, __u64 seq)
 {
-        int seq_mds;
+        struct obd_export *fld_exp;
+        int count = 0, hash;
+        ENTRY;
 
-        seq_mds = dht_mdt_hash(seq);
-        CDEBUG(D_INFO, "mds number %d\n", seq_mds);
+        hash = fld->fld_hash->fh_func(fld, seq);
+
+        spin_lock(&fld->fld_lock);
+        list_for_each_entry(fld_exp, &fld->fld_exports, exp_obd_chain) {
+                if (count == hash)
+                        break;
+                count++;
+        }
+        spin_unlock(&fld->fld_lock);
 
-        /* XXX: get exp according to lu_seq */
-        return fld->fld_exp;
+        RETURN(fld_exp);
 }
 
-int fld_client_init(struct lu_client_fld *fld,
-                    struct obd_export *exp)
+/* add export to FLD. This is usually done by CMM and LMV as they are main users
+ * of FLD module. */
+int fld_client_add_export(struct lu_client_fld *fld,
+                          struct obd_export *exp)
+{
+        struct obd_export *fld_exp;
+        ENTRY;
+
+        spin_lock(&fld->fld_lock);
+        list_for_each_entry(fld_exp, &fld->fld_exports, exp_obd_chain) {
+                if (obd_uuid_equals(&fld_exp->exp_client_uuid,
+                                    &exp->exp_client_uuid))
+                {
+                        spin_unlock(&fld->fld_lock);
+                        RETURN(-EEXIST);
+                }
+        }
+        
+        fld_exp = class_export_get(exp);
+        list_add_tail(&exp->exp_obd_chain,
+                      &fld->fld_exports);
+        fld->fld_count++;
+        
+        spin_unlock(&fld->fld_lock);
+        
+        RETURN(0);
+}
+
+/* remove export from FLD */
+int fld_client_del_export(struct lu_client_fld *fld,
+                          struct obd_export *exp)
+{
+        struct obd_export *fld_exp;
+        struct obd_export *tmp;
+        ENTRY;
+
+        spin_lock(&fld->fld_lock);
+        list_for_each_entry_safe(fld_exp, tmp, &fld->fld_exports, exp_obd_chain) {
+                if (obd_uuid_equals(&fld_exp->exp_client_uuid,
+                                    &exp->exp_client_uuid))
+                {
+                        fld->fld_count--;
+                        list_del(&fld_exp->exp_obd_chain);
+                        class_export_get(fld_exp);
+
+                        spin_unlock(&fld->fld_lock);
+                        RETURN(0);
+                }
+        }
+        spin_unlock(&fld->fld_lock);
+        
+        RETURN(-ENOENT);
+}
+
+int fld_client_init(struct lu_client_fld *fld, int hash)
 {
         int rc = 0;
         ENTRY;
 
-        LASSERT(exp != NULL);
-        fld->fld_exp = class_export_get(exp);
-        CDEBUG(D_INFO, "Client FLD initialized\n");
+        LASSERT(fld != NULL);
+
+        if (hash < 0 || hash >= LUSTRE_CLI_FLD_HASH_LAST) {
+                CERROR("wrong hash function 0x%x\n", hash);
+                RETURN(-EINVAL);
+        }
+        
+        INIT_LIST_HEAD(&fld->fld_exports);
+        spin_lock_init(&fld->fld_lock);
+        fld->fld_hash = &fld_hash[hash];
+        fld->fld_count = 0;
         
+        CDEBUG(D_INFO, "Client FLD initialized, using %s\n",
+               fld->fld_hash->fh_name);
         RETURN(rc);
 }
 
 void fld_client_fini(struct lu_client_fld *fld)
 {
+        struct obd_export *fld_exp;
+        struct obd_export *tmp;
         ENTRY;
-        if (fld->fld_exp != NULL) {
-                class_export_put(fld->fld_exp);
-                fld->fld_exp = NULL;
+
+        spin_lock(&fld->fld_lock);
+        list_for_each_entry_safe(fld_exp, tmp, &fld->fld_exports, exp_obd_chain) {
+                fld->fld_count--;
+                list_del(&fld_exp->exp_obd_chain);
+                class_export_get(fld_exp);
         }
+        spin_unlock(&fld->fld_lock);
         CDEBUG(D_INFO, "Client FLD finalized\n");
         EXIT;
 }

diff --git a/lustre/fld/fld_iam.c b/lustre/fld/fld_iam.c
index 50d9b53..712ee43 100644 (file)
--- a/lustre/fld/fld_iam.c
+++ b/lustre/fld/fld_iam.c
@@ -77,18 +77,21 @@ static void *fld_key_init(const struct lu_context *ctx,
                           struct lu_context_key *key)
 {
         struct fld_thread_info *info;
+        ENTRY;
 
         OBD_ALLOC_PTR(info);
         if (info == NULL)
                 info = ERR_PTR(-ENOMEM);
-        return info;
+        RETURN(info);
 }
 
 static void fld_key_fini(const struct lu_context *ctx,
                          struct lu_context_key *key, void *data)
 {
         struct fld_thread_info *info = data;
+        ENTRY;
         OBD_FREE_PTR(info);
+        EXIT;
 }
 
 static int fld_key_registered = 0;
@@ -101,21 +104,27 @@ static struct lu_context_key fld_thread_key = {
 static struct dt_key *fld_key(const struct lu_context *ctx,
                               const fidseq_t seq_num)
 {
-        struct fld_thread_info *info = lu_context_key_get(ctx, &fld_thread_key);
+        struct fld_thread_info *info;
+        ENTRY;
+        
+        info = lu_context_key_get(ctx, &fld_thread_key);
         LASSERT(info != NULL);
 
         info->fti_key = cpu_to_be64(seq_num);
-        return (void *)&info->fti_key;
+        RETURN((void *)&info->fti_key);
 }
 
 static struct dt_rec *fld_rec(const struct lu_context *ctx,
                               const mdsno_t mds_num)
 {
-        struct fld_thread_info *info = lu_context_key_get(ctx, &fld_thread_key);
+        struct fld_thread_info *info;
+        ENTRY;
+        
+        info = lu_context_key_get(ctx, &fld_thread_key);
         LASSERT(info != NULL);
 
         info->fti_rec = cpu_to_be64(mds_num);
-        return (void *)&info->fti_rec;
+        RETURN((void *)&info->fti_rec);
 }
 
 int fld_handle_insert(struct lu_server_fld *fld,
@@ -126,8 +135,8 @@ int fld_handle_insert(struct lu_server_fld *fld,
         struct dt_object *dt_obj = fld->fld_obj;
         struct txn_param txn;
         struct thandle *th;
-        int    rc;
-
+        int rc;
+        ENTRY;
 
         /*stub here, will fix it later*/
         txn.tp_credits = FLD_TXN_INDEX_INSERT_CREDITS;
@@ -150,8 +159,8 @@ int fld_handle_delete(struct lu_server_fld *fld,
         struct dt_object *dt_obj = fld->fld_obj;
         struct txn_param txn;
         struct thandle *th;
-        int    rc;
-
+        int rc;
+        ENTRY;
 
         txn.tp_credits = FLD_TXN_INDEX_DELETE_CREDITS;
         th = dt->dd_ops->dt_trans_start(ctx, dt, &txn);
@@ -166,16 +175,17 @@ int fld_handle_lookup(struct lu_server_fld *fld,
                       const struct lu_context *ctx,
                       fidseq_t seq_num, mdsno_t *mds_num)
 {
-        int result;
-
+        int rc;
+        ENTRY;
+        
         struct dt_object *dt_obj = fld->fld_obj;
         struct dt_rec    *rec = fld_rec(ctx, 0);
 
-        result = dt_obj->do_index_ops->dio_lookup(ctx, dt_obj, rec,
-                                                  fld_key(ctx, seq_num));
-        if (result == 0)
+        rc = dt_obj->do_index_ops->dio_lookup(ctx, dt_obj, rec,
+                                              fld_key(ctx, seq_num));
+        if (rc == 0)
                 *mds_num = be64_to_cpu(*(__u64 *)rec);
-        return result;
+        RETURN(rc);
 }
 
 int fld_iam_init(struct lu_server_fld *fld,
@@ -184,7 +194,6 @@ int fld_iam_init(struct lu_server_fld *fld,
         struct dt_device *dt = fld->fld_dt;
         struct dt_object *dt_obj;
         int rc;
-
         ENTRY;
 
         if (fld_key_registered == 0) {
@@ -225,6 +234,7 @@ int fld_iam_init(struct lu_server_fld *fld,
 void fld_iam_fini(struct lu_server_fld *fld,
                   const struct lu_context *ctx)
 {
+        ENTRY;
         if (!IS_ERR(fld->fld_cookie) && fld->fld_cookie != NULL) {
                 fld->fld_dt->dd_ops->dt_index_fini(ctx, fld->fld_cookie);
                 fld->fld_cookie = NULL;
@@ -234,7 +244,8 @@ void fld_iam_fini(struct lu_server_fld *fld,
                 fld->fld_obj = NULL;
         }
         if (fld_key_registered > 0) {
-                if (-- fld_key_registered == 0)
+                if (--fld_key_registered == 0)
                         lu_context_key_degister(&fld_thread_key);
         }
+        EXIT;
 }

diff --git a/lustre/include/lustre_fld.h b/lustre/include/lustre_fld.h
index eeee541..7be9143 100644 (file)
--- a/lustre/include/lustre_fld.h
+++ b/lustre/include/lustre_fld.h
@@ -23,9 +23,25 @@
 #ifndef __LINUX_FLD_H
 #define __LINUX_FLD_H
 
+struct lu_client_fld;
+struct lu_server_fld;
+
 /*
  * FLD (Fid Location Database) interface.
  */
+enum {
+        LUSTRE_CLI_FLD_HASH_DHT = 0,
+        LUSTRE_CLI_FLD_HASH_RRB,
+        LUSTRE_CLI_FLD_HASH_LAST
+};
+        
+typedef int (*fld_hash_func_t) (struct lu_client_fld *, __u64);
+
+struct lu_fld_hash {
+        const char              *fh_name;
+        fld_hash_func_t          fh_func;
+};
+
 struct lu_server_fld {
         struct proc_dir_entry   *fld_proc_entry;
         struct ptlrpc_service   *fld_service;
@@ -37,7 +53,10 @@ struct lu_server_fld {
 
 struct lu_client_fld {
         struct proc_dir_entry   *fld_proc_entry;
-        struct obd_export       *fld_exp;
+        struct list_head         fld_exports;
+        struct lu_fld_hash      *fld_hash;
+        int                      fld_count;
+        spinlock_t               fld_lock;
 };
 
 /* server methods */
@@ -50,10 +69,16 @@ void fld_server_fini(struct lu_server_fld *fld,
 
 /* client methods */
 int fld_client_init(struct lu_client_fld *fld,
-                    struct obd_export *exp);
+                    int hash);
 
 void fld_client_fini(struct lu_client_fld *fld);
 
+int fld_client_add_export(struct lu_client_fld *fld,
+                          struct obd_export *exp);
+
+int fld_client_del_export(struct lu_client_fld *fld,
+                          struct obd_export *exp);
+
 int fld_client_create(struct lu_client_fld *fld,
                       __u64 seq, __u64 mds_num);