Whamcloud - gitweb
b=18551 adapt ldlm_namespace to new cfs_hash
authorLiang Zhen <liang@whamcloud.com>
Fri, 19 Nov 2010 14:44:02 +0000 (22:44 +0800)
committerVitaly Fertman <vitaly.fertman@oracle.com>
Tue, 23 Nov 2010 21:13:54 +0000 (00:13 +0300)
Use new cfs_hash to implement ldlm_namespace

o=Liang Zhen
i=Andreas Dilger
i=Oleg Drokin
i=Vitaly Fertman

13 files changed:
lustre/include/lustre_dlm.h
lustre/ldlm/ldlm_extent.c
lustre/ldlm/ldlm_flock.c
lustre/ldlm/ldlm_lib.c
lustre/ldlm/ldlm_lock.c
lustre/ldlm/ldlm_lockd.c
lustre/ldlm/ldlm_pool.c
lustre/ldlm/ldlm_request.c
lustre/ldlm/ldlm_resource.c
lustre/mdt/mdt_handler.c
lustre/mgs/mgs_handler.c
lustre/obdecho/echo.c
lustre/obdfilter/filter.c

index 2c6ed28..79f8d08 100644 (file)
@@ -261,7 +261,7 @@ static inline int lockmode_compat(ldlm_mode_t exist_mode, ldlm_mode_t new_mode)
  *     led_lock
  *
  * lr_lock
  *     led_lock
  *
  * lr_lock
- *     ns_unused_lock
+ *     ns_lock
  *
  * lr_lvb_sem
  *     lr_lock
  *
  * lr_lvb_sem
  *     lr_lock
@@ -391,11 +391,41 @@ typedef enum {
 #define NS_DEFAULT_CONTENTION_SECONDS 2
 #define NS_DEFAULT_CONTENDED_LOCKS 32
 
 #define NS_DEFAULT_CONTENTION_SECONDS 2
 #define NS_DEFAULT_CONTENDED_LOCKS 32
 
+struct ldlm_ns_bucket {
+        /** refer back */
+        struct ldlm_namespace      *nsb_namespace;
+        /** estimated lock callback time */
+        struct adaptive_timeout     nsb_at_estimate;
+};
+
+enum {
+        /** ldlm namespace lock stats */
+        LDLM_NSS_LOCKS          = 0,
+        LDLM_NSS_LAST
+};
+
+typedef enum {
+        /** invalide type */
+        LDLM_NS_TYPE_UNKNOWN    = 0,
+        /** mdc namespace */
+        LDLM_NS_TYPE_MDC,
+        /** mds namespace */
+        LDLM_NS_TYPE_MDT,
+        /** osc namespace */
+        LDLM_NS_TYPE_OSC,
+        /** ost namespace */
+        LDLM_NS_TYPE_OST,
+        /** mgc namespace */
+        LDLM_NS_TYPE_MGC,
+        /** mgs namespace */
+        LDLM_NS_TYPE_MGT,
+} ldlm_ns_type_t;
+
 struct ldlm_namespace {
         /**
 struct ldlm_namespace {
         /**
-         * Namespace name. Used for logging, etc.
+         * Backward link to obd, required for ldlm pool to store new SLV.
          */
          */
-        char                  *ns_name;
+        struct obd_device     *ns_obd;
 
         /**
          * Is this a client-side lock tree?
 
         /**
          * Is this a client-side lock tree?
@@ -403,31 +433,30 @@ struct ldlm_namespace {
         ldlm_side_t            ns_client;
 
         /**
         ldlm_side_t            ns_client;
 
         /**
-         * Namespce connect flags supported by server (may be changed via proc,
-         * lru resize may be disabled/enabled).
+         * resource hash
          */
          */
-        __u64                  ns_connect_flags;
+        cfs_hash_t            *ns_rs_hash;
 
 
-         /**
-          * Client side orig connect flags supported by server.
-          */
-        __u64                  ns_orig_connect_flags;
+        /**
+         * serialize
+         */
+        cfs_spinlock_t         ns_lock;
 
         /**
 
         /**
-         * Hash table for namespace.
+         * big refcount (by bucket)
          */
          */
-        cfs_list_t            *ns_hash;
-        cfs_spinlock_t         ns_hash_lock;
+        cfs_atomic_t           ns_bref;
 
 
-         /**
-          * Count of resources in the hash.
-          */
-        __u32                  ns_refcount;
+        /**
+         * Namespce connect flags supported by server (may be changed via proc,
+         * lru resize may be disabled/enabled).
+         */
+        __u64                  ns_connect_flags;
 
          /**
 
          /**
-          * All root resources in namespace.
+          * Client side orig connect flags supported by server.
           */
           */
-        cfs_list_t             ns_root_list;
+        __u64                  ns_orig_connect_flags;
 
         /**
          * Position in global namespace list.
 
         /**
          * Position in global namespace list.
@@ -439,7 +468,6 @@ struct ldlm_namespace {
          */
         cfs_list_t             ns_unused_list;
         int                    ns_nr_unused;
          */
         cfs_list_t             ns_unused_list;
         int                    ns_nr_unused;
-        cfs_spinlock_t         ns_unused_lock;
 
         unsigned int           ns_max_unused;
         unsigned int           ns_max_age;
 
         unsigned int           ns_max_unused;
         unsigned int           ns_max_age;
@@ -454,8 +482,6 @@ struct ldlm_namespace {
          */
         cfs_time_t             ns_next_dump;
 
          */
         cfs_time_t             ns_next_dump;
 
-        cfs_atomic_t           ns_locks;
-        __u64                  ns_resources;
         ldlm_res_policy        ns_policy;
         struct ldlm_valblock_ops *ns_lvbo;
         void                  *ns_lvbp;
         ldlm_res_policy        ns_policy;
         struct ldlm_valblock_ops *ns_lvbo;
         void                  *ns_lvbp;
@@ -479,16 +505,12 @@ struct ldlm_namespace {
          * Limit size of nolock requests, in bytes.
          */
         unsigned               ns_max_nolock_size;
          * Limit size of nolock requests, in bytes.
          */
         unsigned               ns_max_nolock_size;
-
-        /**
-         * Backward link to obd, required for ldlm pool to store new SLV.
-         */
-        struct obd_device     *ns_obd;
-
-        struct adaptive_timeout ns_at_estimate;/* estimated lock callback time*/
-
         /* callback to cancel locks before replaying it during recovery */
         ldlm_cancel_for_recovery ns_cancel_for_recovery;
         /* callback to cancel locks before replaying it during recovery */
         ldlm_cancel_for_recovery ns_cancel_for_recovery;
+        /**
+         * ldlm lock stats
+         */
+        struct lprocfs_stats  *ns_stats;
 };
 
 static inline int ns_is_client(struct ldlm_namespace *ns)
 };
 
 static inline int ns_is_client(struct ldlm_namespace *ns)
@@ -524,16 +546,6 @@ static inline void ns_register_cancel(struct ldlm_namespace *ns,
         ns->ns_cancel_for_recovery = arg;
 }
 
         ns->ns_cancel_for_recovery = arg;
 }
 
-/*
- *
- * Resource hash table
- *
- */
-
-#define RES_HASH_BITS 12
-#define RES_HASH_SIZE (1UL << RES_HASH_BITS)
-#define RES_HASH_MASK (RES_HASH_SIZE - 1)
-
 struct ldlm_lock;
 
 typedef int (*ldlm_blocking_callback)(struct ldlm_lock *lock,
 struct ldlm_lock;
 
 typedef int (*ldlm_blocking_callback)(struct ldlm_lock *lock,
@@ -748,12 +760,10 @@ struct ldlm_lock {
 };
 
 struct ldlm_resource {
 };
 
 struct ldlm_resource {
-        struct ldlm_namespace *lr_namespace;
+        struct ldlm_ns_bucket *lr_ns_bucket;
 
         /* protected by ns_hash_lock */
 
         /* protected by ns_hash_lock */
-        cfs_list_t             lr_hash;
-        cfs_list_t             lr_childof;  /* part of ns_root_list if root res,
-                                             * part of lr_children if child */
+        cfs_hlist_node_t       lr_hash;
         cfs_spinlock_t         lr_lock;
 
         /* protected by lr_lock */
         cfs_spinlock_t         lr_lock;
 
         /* protected by lr_lock */
@@ -782,10 +792,16 @@ struct ldlm_resource {
         struct inode          *lr_lvb_inode;
 };
 
         struct inode          *lr_lvb_inode;
 };
 
+static inline char *
+ldlm_ns_name(struct ldlm_namespace *ns)
+{
+        return ns->ns_rs_hash->hs_name;
+}
+
 static inline struct ldlm_namespace *
 ldlm_res_to_ns(struct ldlm_resource *res)
 {
 static inline struct ldlm_namespace *
 ldlm_res_to_ns(struct ldlm_resource *res)
 {
-        return res->lr_namespace;
+        return res->lr_ns_bucket->nsb_namespace;
 }
 
 static inline struct ldlm_namespace *
 }
 
 static inline struct ldlm_namespace *
@@ -797,13 +813,13 @@ ldlm_lock_to_ns(struct ldlm_lock *lock)
 static inline char *
 ldlm_lock_to_ns_name(struct ldlm_lock *lock)
 {
 static inline char *
 ldlm_lock_to_ns_name(struct ldlm_lock *lock)
 {
-        return ldlm_lock_to_ns(lock)->ns_name;
+        return ldlm_ns_name(ldlm_lock_to_ns(lock));
 }
 
 static inline struct adaptive_timeout *
 ldlm_lock_to_ns_at(struct ldlm_lock *lock)
 {
 }
 
 static inline struct adaptive_timeout *
 ldlm_lock_to_ns_at(struct ldlm_lock *lock)
 {
-        return &ldlm_lock_to_ns(lock)->ns_at_estimate;
+        return &lock->l_resource->lr_ns_bucket->nsb_at_estimate;
 }
 
 struct ldlm_ast_work {
 }
 
 struct ldlm_ast_work {
@@ -894,10 +910,8 @@ typedef int (*ldlm_res_iterator_t)(struct ldlm_resource *, void *);
 
 int ldlm_resource_foreach(struct ldlm_resource *res, ldlm_iterator_t iter,
                           void *closure);
 
 int ldlm_resource_foreach(struct ldlm_resource *res, ldlm_iterator_t iter,
                           void *closure);
-int ldlm_namespace_foreach(struct ldlm_namespace *ns, ldlm_iterator_t iter,
-                           void *closure);
-int ldlm_namespace_foreach_res(struct ldlm_namespace *ns,
-                               ldlm_res_iterator_t iter, void *closure);
+void ldlm_namespace_foreach(struct ldlm_namespace *ns, ldlm_iterator_t iter,
+                            void *closure);
 
 int ldlm_replay_locks(struct obd_import *imp);
 int ldlm_resource_iterate(struct ldlm_namespace *, const struct ldlm_res_id *,
 
 int ldlm_replay_locks(struct obd_import *imp);
 int ldlm_resource_iterate(struct ldlm_namespace *, const struct ldlm_res_id *,
@@ -1043,7 +1057,8 @@ void ldlm_unlink_lock_skiplist(struct ldlm_lock *req);
 /* resource.c */
 struct ldlm_namespace *
 ldlm_namespace_new(struct obd_device *obd, char *name,
 /* resource.c */
 struct ldlm_namespace *
 ldlm_namespace_new(struct obd_device *obd, char *name,
-                   ldlm_side_t client, ldlm_appetite_t apt);
+                   ldlm_side_t client, ldlm_appetite_t apt,
+                   ldlm_ns_type_t ns_type);
 int ldlm_namespace_cleanup(struct ldlm_namespace *ns, int flags);
 void ldlm_namespace_free(struct ldlm_namespace *ns,
                          struct obd_import *imp, int force);
 int ldlm_namespace_cleanup(struct ldlm_namespace *ns, int flags);
 void ldlm_namespace_free(struct ldlm_namespace *ns,
                          struct obd_import *imp, int force);
@@ -1051,10 +1066,8 @@ void ldlm_namespace_register(struct ldlm_namespace *ns, ldlm_side_t client);
 void ldlm_namespace_unregister(struct ldlm_namespace *ns, ldlm_side_t client);
 void ldlm_namespace_move_locked(struct ldlm_namespace *ns, ldlm_side_t client);
 struct ldlm_namespace *ldlm_namespace_first_locked(ldlm_side_t client);
 void ldlm_namespace_unregister(struct ldlm_namespace *ns, ldlm_side_t client);
 void ldlm_namespace_move_locked(struct ldlm_namespace *ns, ldlm_side_t client);
 struct ldlm_namespace *ldlm_namespace_first_locked(ldlm_side_t client);
-void ldlm_namespace_get_locked(struct ldlm_namespace *ns);
-void ldlm_namespace_put_locked(struct ldlm_namespace *ns, int wakeup);
 void ldlm_namespace_get(struct ldlm_namespace *ns);
 void ldlm_namespace_get(struct ldlm_namespace *ns);
-void ldlm_namespace_put(struct ldlm_namespace *ns, int wakeup);
+void ldlm_namespace_put(struct ldlm_namespace *ns);
 int ldlm_proc_setup(void);
 #ifdef LPROCFS
 void ldlm_proc_cleanup(void);
 int ldlm_proc_setup(void);
 #ifdef LPROCFS
 void ldlm_proc_cleanup(void);
index a8aa5ec..8232d6f 100644 (file)
@@ -760,7 +760,7 @@ out:
 
 /* When a lock is cancelled by a client, the KMS may undergo change if this
  * is the "highest lock".  This function returns the new KMS value.
 
 /* When a lock is cancelled by a client, the KMS may undergo change if this
  * is the "highest lock".  This function returns the new KMS value.
- * Caller must hold ns_lock already.
+ * Caller must hold lr_lock already.
  *
  * NB: A lock on [x,y] protects a KMS of up to y + 1 bytes! */
 __u64 ldlm_extent_shift_kms(struct ldlm_lock *lock, __u64 old_kms)
  *
  * NB: A lock on [x,y] protects a KMS of up to y + 1 bytes! */
 __u64 ldlm_extent_shift_kms(struct ldlm_lock *lock, __u64 old_kms)
index d22e99e..32837d5 100644 (file)
@@ -373,7 +373,7 @@ reprocess:
                  * it must see the original lock data in the reply. */
 
                 /* XXX - if ldlm_lock_new() can sleep we should
                  * it must see the original lock data in the reply. */
 
                 /* XXX - if ldlm_lock_new() can sleep we should
-                 * release the ns_lock, allocate the new lock,
+                 * release the lr_lock, allocate the new lock,
                  * and restart processing this lock. */
                 if (!new2) {
                         unlock_res_and_lock(req);
                  * and restart processing this lock. */
                 if (!new2) {
                         unlock_res_and_lock(req);
index 5530278..b49c28a 100644 (file)
@@ -214,6 +214,7 @@ int client_obd_setup(struct obd_device *obddev, struct lustre_cfg *lcfg)
         struct obd_uuid server_uuid;
         int rq_portal, rp_portal, connect_op;
         char *name = obddev->obd_type->typ_name;
         struct obd_uuid server_uuid;
         int rq_portal, rp_portal, connect_op;
         char *name = obddev->obd_type->typ_name;
+        ldlm_ns_type_t ns_type = LDLM_NS_TYPE_UNKNOWN;
         int rc;
         ENTRY;
 
         int rc;
         ENTRY;
 
@@ -225,12 +226,16 @@ int client_obd_setup(struct obd_device *obddev, struct lustre_cfg *lcfg)
                 connect_op = OST_CONNECT;
                 cli->cl_sp_me = LUSTRE_SP_CLI;
                 cli->cl_sp_to = LUSTRE_SP_OST;
                 connect_op = OST_CONNECT;
                 cli->cl_sp_me = LUSTRE_SP_CLI;
                 cli->cl_sp_to = LUSTRE_SP_OST;
+                ns_type = LDLM_NS_TYPE_OSC;
+
         } else if (!strcmp(name, LUSTRE_MDC_NAME)) {
                 rq_portal = MDS_REQUEST_PORTAL;
                 rp_portal = MDC_REPLY_PORTAL;
                 connect_op = MDS_CONNECT;
                 cli->cl_sp_me = LUSTRE_SP_CLI;
                 cli->cl_sp_to = LUSTRE_SP_MDT;
         } else if (!strcmp(name, LUSTRE_MDC_NAME)) {
                 rq_portal = MDS_REQUEST_PORTAL;
                 rp_portal = MDC_REPLY_PORTAL;
                 connect_op = MDS_CONNECT;
                 cli->cl_sp_me = LUSTRE_SP_CLI;
                 cli->cl_sp_to = LUSTRE_SP_MDT;
+                ns_type = LDLM_NS_TYPE_MDC;
+
         } else if (!strcmp(name, LUSTRE_MGC_NAME)) {
                 rq_portal = MGS_REQUEST_PORTAL;
                 rp_portal = MGC_REPLY_PORTAL;
         } else if (!strcmp(name, LUSTRE_MGC_NAME)) {
                 rq_portal = MGS_REQUEST_PORTAL;
                 rp_portal = MGC_REPLY_PORTAL;
@@ -238,6 +243,8 @@ int client_obd_setup(struct obd_device *obddev, struct lustre_cfg *lcfg)
                 cli->cl_sp_me = LUSTRE_SP_MGC;
                 cli->cl_sp_to = LUSTRE_SP_MGS;
                 cli->cl_flvr_mgc.sf_rpc = SPTLRPC_FLVR_INVALID;
                 cli->cl_sp_me = LUSTRE_SP_MGC;
                 cli->cl_sp_to = LUSTRE_SP_MGS;
                 cli->cl_flvr_mgc.sf_rpc = SPTLRPC_FLVR_INVALID;
+                ns_type = LDLM_NS_TYPE_MGC;
+
         } else {
                 CERROR("unknown client OBD type \"%s\", can't setup\n",
                        name);
         } else {
                 CERROR("unknown client OBD type \"%s\", can't setup\n",
                        name);
@@ -366,7 +373,8 @@ int client_obd_setup(struct obd_device *obddev, struct lustre_cfg *lcfg)
 
         obddev->obd_namespace = ldlm_namespace_new(obddev, obddev->obd_name,
                                                    LDLM_NAMESPACE_CLIENT,
 
         obddev->obd_namespace = ldlm_namespace_new(obddev, obddev->obd_name,
                                                    LDLM_NAMESPACE_CLIENT,
-                                                   LDLM_NAMESPACE_GREEDY);
+                                                   LDLM_NAMESPACE_GREEDY,
+                                                   ns_type);
         if (obddev->obd_namespace == NULL) {
                 CERROR("Unable to create client namespace - %s\n",
                        obddev->obd_name);
         if (obddev->obd_namespace == NULL) {
                 CERROR("Unable to create client namespace - %s\n",
                        obddev->obd_name);
index 4779bde..d788767 100644 (file)
@@ -159,7 +159,8 @@ void ldlm_lock_put(struct ldlm_lock *lock)
                 LASSERT(cfs_list_empty(&lock->l_res_link));
                 LASSERT(cfs_list_empty(&lock->l_pending_chain));
 
                 LASSERT(cfs_list_empty(&lock->l_res_link));
                 LASSERT(cfs_list_empty(&lock->l_pending_chain));
 
-                cfs_atomic_dec(&ldlm_res_to_ns(res)->ns_locks);
+                lprocfs_counter_decr(ldlm_res_to_ns(res)->ns_stats,
+                                     LDLM_NSS_LOCKS);
                 lu_ref_del(&res->lr_reference, "lock", lock);
                 ldlm_resource_putref(res);
                 lock->l_resource = NULL;
                 lu_ref_del(&res->lr_reference, "lock", lock);
                 ldlm_resource_putref(res);
                 lock->l_resource = NULL;
@@ -208,9 +209,9 @@ int ldlm_lock_remove_from_lru(struct ldlm_lock *lock)
                 RETURN(0);
         }
 
                 RETURN(0);
         }
 
-        cfs_spin_lock(&ns->ns_unused_lock);
+        cfs_spin_lock(&ns->ns_lock);
         rc = ldlm_lock_remove_from_lru_nolock(lock);
         rc = ldlm_lock_remove_from_lru_nolock(lock);
-        cfs_spin_unlock(&ns->ns_unused_lock);
+        cfs_spin_unlock(&ns->ns_lock);
         EXIT;
         return rc;
 }
         EXIT;
         return rc;
 }
@@ -232,9 +233,9 @@ void ldlm_lock_add_to_lru(struct ldlm_lock *lock)
         struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
 
         ENTRY;
         struct ldlm_namespace *ns = ldlm_lock_to_ns(lock);
 
         ENTRY;
-        cfs_spin_lock(&ns->ns_unused_lock);
+        cfs_spin_lock(&ns->ns_lock);
         ldlm_lock_add_to_lru_nolock(lock);
         ldlm_lock_add_to_lru_nolock(lock);
-        cfs_spin_unlock(&ns->ns_unused_lock);
+        cfs_spin_unlock(&ns->ns_lock);
         EXIT;
 }
 
         EXIT;
 }
 
@@ -249,12 +250,12 @@ void ldlm_lock_touch_in_lru(struct ldlm_lock *lock)
                 return;
         }
 
                 return;
         }
 
-        cfs_spin_lock(&ns->ns_unused_lock);
+        cfs_spin_lock(&ns->ns_lock);
         if (!cfs_list_empty(&lock->l_lru)) {
                 ldlm_lock_remove_from_lru_nolock(lock);
                 ldlm_lock_add_to_lru_nolock(lock);
         }
         if (!cfs_list_empty(&lock->l_lru)) {
                 ldlm_lock_remove_from_lru_nolock(lock);
                 ldlm_lock_add_to_lru_nolock(lock);
         }
-        cfs_spin_unlock(&ns->ns_unused_lock);
+        cfs_spin_unlock(&ns->ns_lock);
         EXIT;
 }
 
         EXIT;
 }
 
@@ -377,7 +378,8 @@ static struct ldlm_lock *ldlm_lock_new(struct ldlm_resource *resource)
         CFS_INIT_LIST_HEAD(&lock->l_sl_policy);
         CFS_INIT_HLIST_NODE(&lock->l_exp_hash);
 
         CFS_INIT_LIST_HEAD(&lock->l_sl_policy);
         CFS_INIT_HLIST_NODE(&lock->l_exp_hash);
 
-        cfs_atomic_inc(&ldlm_res_to_ns(resource)->ns_locks);
+        lprocfs_counter_incr(ldlm_res_to_ns(resource)->ns_stats,
+                             LDLM_NSS_LOCKS);
         CFS_INIT_LIST_HEAD(&lock->l_handle.h_link);
         class_handle_hash(&lock->l_handle, lock_handle_addref);
 
         CFS_INIT_LIST_HEAD(&lock->l_handle.h_link);
         class_handle_hash(&lock->l_handle, lock_handle_addref);
 
@@ -421,9 +423,10 @@ int ldlm_lock_change_resource(struct ldlm_namespace *ns, struct ldlm_lock *lock,
         unlock_res_and_lock(lock);
 
         newres = ldlm_resource_get(ns, NULL, new_resid, type, 1);
         unlock_res_and_lock(lock);
 
         newres = ldlm_resource_get(ns, NULL, new_resid, type, 1);
-        lu_ref_add(&newres->lr_reference, "lock", lock);
         if (newres == NULL)
                 RETURN(-ENOMEM);
         if (newres == NULL)
                 RETURN(-ENOMEM);
+
+        lu_ref_add(&newres->lr_reference, "lock", lock);
         /*
          * To flip the lock from the old to the new resource, lock, oldres and
          * newres have to be locked. Resource spin-locks are nested within
         /*
          * To flip the lock from the old to the new resource, lock, oldres and
          * newres have to be locked. Resource spin-locks are nested within
@@ -1540,40 +1543,25 @@ static int reprocess_one_queue(struct ldlm_resource *res, void *closure)
         return LDLM_ITER_CONTINUE;
 }
 
         return LDLM_ITER_CONTINUE;
 }
 
-void ldlm_reprocess_all_ns(struct ldlm_namespace *ns)
+static int ldlm_reprocess_res(cfs_hash_t *hs, cfs_hash_bd_t *bd,
+                              cfs_hlist_node_t *hnode, void *arg)
 {
 {
-        cfs_list_t *tmp;
-        int i, rc;
+        struct ldlm_resource *res = cfs_hash_object(hs, hnode);
+        int    rc;
 
 
-        if (ns == NULL)
-                return;
+        rc = reprocess_one_queue(res, arg);
+
+        return rc == LDLM_ITER_STOP;
+}
 
 
+void ldlm_reprocess_all_ns(struct ldlm_namespace *ns)
+{
         ENTRY;
         ENTRY;
-        cfs_spin_lock(&ns->ns_hash_lock);
-        for (i = 0; i < RES_HASH_SIZE; i++) {
-                tmp = ns->ns_hash[i].next;
-                while (tmp != &(ns->ns_hash[i])) {
-                        struct ldlm_resource *res =
-                                cfs_list_entry(tmp, struct ldlm_resource,
-                                               lr_hash);
-
-                        ldlm_resource_getref(res);
-                        cfs_spin_unlock(&ns->ns_hash_lock);
-                        LDLM_RESOURCE_ADDREF(res);
-
-                        rc = reprocess_one_queue(res, NULL);
-
-                        LDLM_RESOURCE_DELREF(res);
-                        cfs_spin_lock(&ns->ns_hash_lock);
-                        tmp = tmp->next;
-                        ldlm_resource_putref_locked(res);
-
-                        if (rc == LDLM_ITER_STOP)
-                                GOTO(out, rc);
-                }
+
+        if (ns != NULL) {
+                cfs_hash_for_each_nolock(ns->ns_rs_hash,
+                                         ldlm_reprocess_res, NULL);
         }
         }
- out:
-        cfs_spin_unlock(&ns->ns_hash_lock);
         EXIT;
 }
 
         EXIT;
 }
 
@@ -1722,7 +1710,6 @@ void ldlm_cancel_locks_for_export(struct obd_export *exp)
  */
 void ldlm_lock_downgrade(struct ldlm_lock *lock, int new_mode)
 {
  */
 void ldlm_lock_downgrade(struct ldlm_lock *lock, int new_mode)
 {
-        struct ldlm_namespace *ns;
         ENTRY;
 
         LASSERT(lock->l_granted_mode & (LCK_PW | LCK_EX));
         ENTRY;
 
         LASSERT(lock->l_granted_mode & (LCK_PW | LCK_EX));
@@ -1734,8 +1721,7 @@ void ldlm_lock_downgrade(struct ldlm_lock *lock, int new_mode)
          * Remove the lock from pool as it will be added again in
          * ldlm_grant_lock() called below.
          */
          * Remove the lock from pool as it will be added again in
          * ldlm_grant_lock() called below.
          */
-        ns = ldlm_lock_to_ns(lock);
-        ldlm_pool_del(&ns->ns_pool, lock);
+        ldlm_pool_del(&ldlm_lock_to_ns(lock)->ns_pool, lock);
 
         lock->l_req_mode = new_mode;
         ldlm_grant_lock(lock, NULL);
 
         lock->l_req_mode = new_mode;
         ldlm_grant_lock(lock, NULL);
@@ -1818,7 +1804,7 @@ struct ldlm_resource *ldlm_lock_convert(struct ldlm_lock *lock, int new_mode,
 
                         ldlm_grant_lock(lock, &rpc_list);
                         granted = 1;
 
                         ldlm_grant_lock(lock, &rpc_list);
                         granted = 1;
-                        /* FIXME: completion handling not with ns_lock held ! */
+                        /* FIXME: completion handling not with lr_lock held ! */
                         if (lock->l_completion_ast)
                                 lock->l_completion_ast(lock, 0, NULL);
                 }
                         if (lock->l_completion_ast)
                                 lock->l_completion_ast(lock, 0, NULL);
                 }
index 1cd25f4..14fe1c9 100644 (file)
@@ -2681,7 +2681,6 @@ EXPORT_SYMBOL(ldlm_cli_cancel_req);
 EXPORT_SYMBOL(ldlm_replay_locks);
 EXPORT_SYMBOL(ldlm_resource_foreach);
 EXPORT_SYMBOL(ldlm_namespace_foreach);
 EXPORT_SYMBOL(ldlm_replay_locks);
 EXPORT_SYMBOL(ldlm_resource_foreach);
 EXPORT_SYMBOL(ldlm_namespace_foreach);
-EXPORT_SYMBOL(ldlm_namespace_foreach_res);
 EXPORT_SYMBOL(ldlm_resource_iterate);
 EXPORT_SYMBOL(ldlm_cancel_resource_local);
 EXPORT_SYMBOL(ldlm_cli_cancel_list_local);
 EXPORT_SYMBOL(ldlm_resource_iterate);
 EXPORT_SYMBOL(ldlm_cancel_resource_local);
 EXPORT_SYMBOL(ldlm_cli_cancel_list_local);
index 9de7e73..b42ed55 100644 (file)
@@ -546,9 +546,9 @@ static int ldlm_cli_pool_shrink(struct ldlm_pool *pl,
          */
         ldlm_cli_pool_pop_slv(pl);
 
          */
         ldlm_cli_pool_pop_slv(pl);
 
-        cfs_spin_lock(&ns->ns_unused_lock);
+        cfs_spin_lock(&ns->ns_lock);
         unused = ns->ns_nr_unused;
         unused = ns->ns_nr_unused;
-        cfs_spin_unlock(&ns->ns_unused_lock);
+        cfs_spin_unlock(&ns->ns_lock);
         
         if (nr) {
                 canceled = ldlm_cancel_lru(ns, nr, LDLM_SYNC, 
         
         if (nr) {
                 canceled = ldlm_cancel_lru(ns, nr, LDLM_SYNC, 
@@ -735,10 +735,11 @@ static int ldlm_pool_proc_init(struct ldlm_pool *pl)
         if (!var_name)
                 RETURN(-ENOMEM);
 
         if (!var_name)
                 RETURN(-ENOMEM);
 
-        parent_ns_proc = lprocfs_srch(ldlm_ns_proc_dir, ns->ns_name);
+        parent_ns_proc = lprocfs_srch(ldlm_ns_proc_dir,
+                                      ldlm_ns_name(ns));
         if (parent_ns_proc == NULL) {
                 CERROR("%s: proc entry is not initialized\n",
         if (parent_ns_proc == NULL) {
                 CERROR("%s: proc entry is not initialized\n",
-                       ns->ns_name);
+                       ldlm_ns_name(ns));
                 GOTO(out_free_name, rc = -EINVAL);
         }
         pl->pl_proc_dir = lprocfs_register("pool", parent_ns_proc,
                 GOTO(out_free_name, rc = -EINVAL);
         }
         pl->pl_proc_dir = lprocfs_register("pool", parent_ns_proc,
@@ -884,7 +885,7 @@ int ldlm_pool_init(struct ldlm_pool *pl, struct ldlm_namespace *ns,
         pl->pl_grant_plan = LDLM_POOL_GP(LDLM_POOL_HOST_L);
 
         snprintf(pl->pl_name, sizeof(pl->pl_name), "ldlm-pool-%s-%d",
         pl->pl_grant_plan = LDLM_POOL_GP(LDLM_POOL_HOST_L);
 
         snprintf(pl->pl_name, sizeof(pl->pl_name), "ldlm-pool-%s-%d",
-                 ns->ns_name, idx);
+                 ldlm_ns_name(ns), idx);
 
         if (client == LDLM_NAMESPACE_SERVER) {
                 pl->pl_ops = &ldlm_srv_pool_ops;
 
         if (client == LDLM_NAMESPACE_SERVER) {
                 pl->pl_ops = &ldlm_srv_pool_ops;
@@ -1105,7 +1106,7 @@ static int ldlm_pools_shrink(ldlm_side_t client, int nr,
                 ldlm_namespace_move_locked(ns, client);
                 cfs_mutex_up(ldlm_namespace_lock(client));
                 total += ldlm_pool_shrink(&ns->ns_pool, 0, gfp_mask);
                 ldlm_namespace_move_locked(ns, client);
                 cfs_mutex_up(ldlm_namespace_lock(client));
                 total += ldlm_pool_shrink(&ns->ns_pool, 0, gfp_mask);
-                ldlm_namespace_put(ns, 1);
+                ldlm_namespace_put(ns);
         }
 
         if (nr == 0 || total == 0) {
         }
 
         if (nr == 0 || total == 0) {
@@ -1144,7 +1145,7 @@ static int ldlm_pools_shrink(ldlm_side_t client, int nr,
                 cancel = 1 + nr_locks * nr / total;
                 ldlm_pool_shrink(&ns->ns_pool, cancel, gfp_mask);
                 cached += ldlm_pool_granted(&ns->ns_pool);
                 cancel = 1 + nr_locks * nr / total;
                 ldlm_pool_shrink(&ns->ns_pool, cancel, gfp_mask);
                 cached += ldlm_pool_granted(&ns->ns_pool);
-                ldlm_namespace_put(ns, 1);
+                ldlm_namespace_put(ns);
         }
         cl_env_reexit(cookie);
         return cached;
         }
         cl_env_reexit(cookie);
         return cached;
@@ -1258,18 +1259,18 @@ void ldlm_pools_recalc(ldlm_side_t client)
                 }
                 ns = ldlm_namespace_first_locked(client);
 
                 }
                 ns = ldlm_namespace_first_locked(client);
 
-                cfs_spin_lock(&ns->ns_hash_lock);
+                cfs_spin_lock(&ns->ns_lock);
                 /*
                  * skip ns which is being freed, and we don't want to increase
                  * its refcount again, not even temporarily. bz21519.
                  */
                 /*
                  * skip ns which is being freed, and we don't want to increase
                  * its refcount again, not even temporarily. bz21519.
                  */
-                if (ns->ns_refcount == 0) {
+                if (cfs_atomic_read(&ns->ns_bref) == 0) {
                         skip = 1;
                 } else {
                         skip = 0;
                         skip = 1;
                 } else {
                         skip = 0;
-                        ldlm_namespace_get_locked(ns);
+                        ldlm_namespace_get(ns);
                 }
                 }
-                cfs_spin_unlock(&ns->ns_hash_lock);
+                cfs_spin_unlock(&ns->ns_lock);
 
                 ldlm_namespace_move_locked(ns, client);
                 cfs_mutex_up(ldlm_namespace_lock(client));
 
                 ldlm_namespace_move_locked(ns, client);
                 cfs_mutex_up(ldlm_namespace_lock(client));
@@ -1279,7 +1280,7 @@ void ldlm_pools_recalc(ldlm_side_t client)
                  */
                 if (!skip) {
                         ldlm_pool_recalc(&ns->ns_pool);
                  */
                 if (!skip) {
                         ldlm_pool_recalc(&ns->ns_pool);
-                        ldlm_namespace_put(ns, 1);
+                        ldlm_namespace_put(ns);
                 }
         }
 }
                 }
         }
 }
index 0f30f16..de1885a 100644 (file)
@@ -337,7 +337,7 @@ int ldlm_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
         lock_res_and_lock(lock);
         /* Get this: if ldlm_blocking_ast is racing with intent_policy, such
          * that ldlm_blocking_ast is called just before intent_policy method
         lock_res_and_lock(lock);
         /* Get this: if ldlm_blocking_ast is racing with intent_policy, such
          * that ldlm_blocking_ast is called just before intent_policy method
-         * takes the ns_lock, then by the time we get the lock, we might not
+         * takes the lr_lock, then by the time we get the lock, we might not
          * be the correct blocking function anymore.  So check, and return
          * early, if so. */
         if (lock->l_blocking_ast != ldlm_blocking_ast) {
          * be the correct blocking function anymore.  So check, and return
          * early, if so. */
         if (lock->l_blocking_ast != ldlm_blocking_ast) {
@@ -1525,7 +1525,7 @@ static int ldlm_prepare_lru_list(struct ldlm_namespace *ns, cfs_list_t *cancels,
         int added = 0, unused, remained;
         ENTRY;
 
         int added = 0, unused, remained;
         ENTRY;
 
-        cfs_spin_lock(&ns->ns_unused_lock);
+        cfs_spin_lock(&ns->ns_lock);
         unused = ns->ns_nr_unused;
         remained = unused;
 
         unused = ns->ns_nr_unused;
         remained = unused;
 
@@ -1567,7 +1567,7 @@ static int ldlm_prepare_lru_list(struct ldlm_namespace *ns, cfs_list_t *cancels,
                         break;
 
                 LDLM_LOCK_GET(lock);
                         break;
 
                 LDLM_LOCK_GET(lock);
-                cfs_spin_unlock(&ns->ns_unused_lock);
+                cfs_spin_unlock(&ns->ns_lock);
                 lu_ref_add(&lock->l_reference, __FUNCTION__, cfs_current());
 
                 /* Pass the lock through the policy filter and see if it
                 lu_ref_add(&lock->l_reference, __FUNCTION__, cfs_current());
 
                 /* Pass the lock through the policy filter and see if it
@@ -1588,14 +1588,14 @@ static int ldlm_prepare_lru_list(struct ldlm_namespace *ns, cfs_list_t *cancels,
                         lu_ref_del(&lock->l_reference,
                                    __FUNCTION__, cfs_current());
                         LDLM_LOCK_RELEASE(lock);
                         lu_ref_del(&lock->l_reference,
                                    __FUNCTION__, cfs_current());
                         LDLM_LOCK_RELEASE(lock);
-                        cfs_spin_lock(&ns->ns_unused_lock);
+                        cfs_spin_lock(&ns->ns_lock);
                         break;
                 }
                 if (result == LDLM_POLICY_SKIP_LOCK) {
                         lu_ref_del(&lock->l_reference,
                                    __FUNCTION__, cfs_current());
                         LDLM_LOCK_RELEASE(lock);
                         break;
                 }
                 if (result == LDLM_POLICY_SKIP_LOCK) {
                         lu_ref_del(&lock->l_reference,
                                    __FUNCTION__, cfs_current());
                         LDLM_LOCK_RELEASE(lock);
-                        cfs_spin_lock(&ns->ns_unused_lock);
+                        cfs_spin_lock(&ns->ns_lock);
                         continue;
                 }
 
                         continue;
                 }
 
@@ -1612,7 +1612,7 @@ static int ldlm_prepare_lru_list(struct ldlm_namespace *ns, cfs_list_t *cancels,
                         lu_ref_del(&lock->l_reference,
                                    __FUNCTION__, cfs_current());
                         LDLM_LOCK_RELEASE(lock);
                         lu_ref_del(&lock->l_reference,
                                    __FUNCTION__, cfs_current());
                         LDLM_LOCK_RELEASE(lock);
-                        cfs_spin_lock(&ns->ns_unused_lock);
+                        cfs_spin_lock(&ns->ns_lock);
                         continue;
                 }
                 LASSERT(!lock->l_readers && !lock->l_writers);
                         continue;
                 }
                 LASSERT(!lock->l_readers && !lock->l_writers);
@@ -1634,7 +1634,7 @@ static int ldlm_prepare_lru_list(struct ldlm_namespace *ns, cfs_list_t *cancels,
 
                 /* We can't re-add to l_lru as it confuses the
                  * refcounting in ldlm_lock_remove_from_lru() if an AST
 
                 /* We can't re-add to l_lru as it confuses the
                  * refcounting in ldlm_lock_remove_from_lru() if an AST
-                 * arrives after we drop ns_lock below. We use l_bl_ast
+                 * arrives after we drop lr_lock below. We use l_bl_ast
                  * and can't use l_pending_chain as it is used both on
                  * server and client nevertheless bug 5666 says it is
                  * used only on server */
                  * and can't use l_pending_chain as it is used both on
                  * server and client nevertheless bug 5666 says it is
                  * used only on server */
@@ -1642,11 +1642,11 @@ static int ldlm_prepare_lru_list(struct ldlm_namespace *ns, cfs_list_t *cancels,
                 cfs_list_add(&lock->l_bl_ast, cancels);
                 unlock_res_and_lock(lock);
                 lu_ref_del(&lock->l_reference, __FUNCTION__, cfs_current());
                 cfs_list_add(&lock->l_bl_ast, cancels);
                 unlock_res_and_lock(lock);
                 lu_ref_del(&lock->l_reference, __FUNCTION__, cfs_current());
-                cfs_spin_lock(&ns->ns_unused_lock);
+                cfs_spin_lock(&ns->ns_lock);
                 added++;
                 unused--;
         }
                 added++;
                 unused--;
         }
-        cfs_spin_unlock(&ns->ns_unused_lock);
+        cfs_spin_unlock(&ns->ns_lock);
         RETURN(added);
 }
 
         RETURN(added);
 }
 
@@ -1824,16 +1824,27 @@ int ldlm_cli_cancel_unused_resource(struct ldlm_namespace *ns,
         RETURN(0);
 }
 
         RETURN(0);
 }
 
-static inline int have_no_nsresource(struct ldlm_namespace *ns)
-{
-        int no_resource = 0;
-
-        cfs_spin_lock(&ns->ns_hash_lock);
-        if (ns->ns_resources == 0)
-                no_resource = 1;
-        cfs_spin_unlock(&ns->ns_hash_lock);
+struct ldlm_cli_cancel_arg {
+        int     lc_flags;
+        void   *lc_opaque;
+};
 
 
-        RETURN(no_resource);
+static int ldlm_cli_hash_cancel_unused(cfs_hash_t *hs, cfs_hash_bd_t *bd,
+                                       cfs_hlist_node_t *hnode, void *arg)
+{
+        struct ldlm_resource           *res = cfs_hash_object(hs, hnode);
+        struct ldlm_cli_cancel_arg     *lc = arg;
+        int                             rc;
+
+        rc = ldlm_cli_cancel_unused_resource(ldlm_res_to_ns(res), &res->lr_name,
+                                             NULL, LCK_MINMODE,
+                                             lc->lc_flags, lc->lc_opaque);
+        if (rc != 0) {
+                CERROR("ldlm_cli_cancel_unused ("LPU64"): %d\n",
+                       res->lr_name.name[0], rc);
+        }
+        /* must return 0 for hash iteration */
+        return 0;
 }
 
 /* Cancel all locks on a namespace (or a specific resource, if given)
 }
 
 /* Cancel all locks on a namespace (or a specific resource, if given)
@@ -1845,48 +1856,25 @@ int ldlm_cli_cancel_unused(struct ldlm_namespace *ns,
                            const struct ldlm_res_id *res_id,
                            ldlm_cancel_flags_t flags, void *opaque)
 {
                            const struct ldlm_res_id *res_id,
                            ldlm_cancel_flags_t flags, void *opaque)
 {
-        int i;
+        struct ldlm_cli_cancel_arg arg = {
+                .lc_flags       = flags,
+                .lc_opaque      = opaque,
+        };
+
         ENTRY;
 
         if (ns == NULL)
                 RETURN(ELDLM_OK);
 
         ENTRY;
 
         if (ns == NULL)
                 RETURN(ELDLM_OK);
 
-        if (res_id)
+        if (res_id != NULL) {
                 RETURN(ldlm_cli_cancel_unused_resource(ns, res_id, NULL,
                                                        LCK_MINMODE, flags,
                                                        opaque));
                 RETURN(ldlm_cli_cancel_unused_resource(ns, res_id, NULL,
                                                        LCK_MINMODE, flags,
                                                        opaque));
-
-        cfs_spin_lock(&ns->ns_hash_lock);
-        for (i = 0; i < RES_HASH_SIZE; i++) {
-                cfs_list_t *tmp;
-                tmp = ns->ns_hash[i].next;
-                while (tmp != &(ns->ns_hash[i])) {
-                        struct ldlm_resource *res;
-                        int rc;
-
-                        res = cfs_list_entry(tmp, struct ldlm_resource,
-                                             lr_hash);
-                        ldlm_resource_getref(res);
-                        cfs_spin_unlock(&ns->ns_hash_lock);
-
-                        LDLM_RESOURCE_ADDREF(res);
-                        rc = ldlm_cli_cancel_unused_resource(ns, &res->lr_name,
-                                                             NULL, LCK_MINMODE,
-                                                             flags, opaque);
-
-                        if (rc)
-                                CERROR("ldlm_cli_cancel_unused ("LPU64"): %d\n",
-                                       res->lr_name.name[0], rc);
-
-                        LDLM_RESOURCE_DELREF(res);
-                        cfs_spin_lock(&ns->ns_hash_lock);
-                        tmp = tmp->next;
-                        ldlm_resource_putref_locked(res);
-                }
+        } else {
+                cfs_hash_for_each_nolock(ns->ns_rs_hash,
+                                         ldlm_cli_hash_cancel_unused, &arg);
+                RETURN(ELDLM_OK);
         }
         }
-        cfs_spin_unlock(&ns->ns_hash_lock);
-
-        RETURN(ELDLM_OK);
 }
 
 /* Lock iterators. */
 }
 
 /* Lock iterators. */
@@ -1940,49 +1928,25 @@ static int ldlm_iter_helper(struct ldlm_lock *lock, void *closure)
         return helper->iter(lock, helper->closure);
 }
 
         return helper->iter(lock, helper->closure);
 }
 
-static int ldlm_res_iter_helper(struct ldlm_resource *res, void *closure)
+static int ldlm_res_iter_helper(cfs_hash_t *hs, cfs_hash_bd_t *bd,
+                                cfs_hlist_node_t *hnode, void *arg)
+
 {
 {
-        return ldlm_resource_foreach(res, ldlm_iter_helper, closure);
+        struct ldlm_resource *res = cfs_hash_object(hs, hnode);
+
+        return ldlm_resource_foreach(res, ldlm_iter_helper, arg) ==
+               LDLM_ITER_STOP;
 }
 
 }
 
-int ldlm_namespace_foreach(struct ldlm_namespace *ns, ldlm_iterator_t iter,
-                           void *closure)
+void ldlm_namespace_foreach(struct ldlm_namespace *ns,
+                            ldlm_iterator_t iter, void *closure)
+
 {
         struct iter_helper_data helper = { iter: iter, closure: closure };
 {
         struct iter_helper_data helper = { iter: iter, closure: closure };
-        return ldlm_namespace_foreach_res(ns, ldlm_res_iter_helper, &helper);
-}
 
 
-int ldlm_namespace_foreach_res(struct ldlm_namespace *ns,
-                               ldlm_res_iterator_t iter, void *closure)
-{
-        int i, rc = LDLM_ITER_CONTINUE;
-        struct ldlm_resource *res;
-        cfs_list_t *tmp;
+        cfs_hash_for_each_nolock(ns->ns_rs_hash,
+                                 ldlm_res_iter_helper, &helper);
 
 
-        ENTRY;
-        cfs_spin_lock(&ns->ns_hash_lock);
-        for (i = 0; i < RES_HASH_SIZE; i++) {
-                tmp = ns->ns_hash[i].next;
-                while (tmp != &(ns->ns_hash[i])) {
-                        res = cfs_list_entry(tmp, struct ldlm_resource,
-                                             lr_hash);
-                        ldlm_resource_getref(res);
-                        cfs_spin_unlock(&ns->ns_hash_lock);
-                        LDLM_RESOURCE_ADDREF(res);
-
-                        rc = iter(res, closure);
-
-                        LDLM_RESOURCE_DELREF(res);
-                        cfs_spin_lock(&ns->ns_hash_lock);
-                        tmp = tmp->next;
-                        ldlm_resource_putref_locked(res);
-                        if (rc == LDLM_ITER_STOP)
-                                GOTO(out, rc);
-                }
-        }
- out:
-        cfs_spin_unlock(&ns->ns_hash_lock);
-        RETURN(rc);
 }
 
 /* non-blocking function to manipulate a lock whose cb_data is being put away.
 }
 
 /* non-blocking function to manipulate a lock whose cb_data is being put away.
@@ -2184,8 +2148,8 @@ static void ldlm_cancel_unused_locks_for_replay(struct ldlm_namespace *ns)
         CFS_LIST_HEAD(cancels);
 
         CDEBUG(D_DLMTRACE, "Dropping as many unused locks as possible before"
         CFS_LIST_HEAD(cancels);
 
         CDEBUG(D_DLMTRACE, "Dropping as many unused locks as possible before"
-                           "replay for namespace %s (%d)\n", ns->ns_name,
-                           ns->ns_nr_unused);
+                           "replay for namespace %s (%d)\n",
+                           ldlm_ns_name(ns), ns->ns_nr_unused);
 
         /* We don't need to care whether or not LRU resize is enabled
          * because the LDLM_CANCEL_NO_WAIT policy doesn't use the
 
         /* We don't need to care whether or not LRU resize is enabled
          * because the LDLM_CANCEL_NO_WAIT policy doesn't use the
@@ -2194,7 +2158,7 @@ static void ldlm_cancel_unused_locks_for_replay(struct ldlm_namespace *ns)
                                          LCF_LOCAL, LDLM_CANCEL_NO_WAIT);
 
         CDEBUG(D_DLMTRACE, "Canceled %d unused locks from namespace %s\n",
                                          LCF_LOCAL, LDLM_CANCEL_NO_WAIT);
 
         CDEBUG(D_DLMTRACE, "Canceled %d unused locks from namespace %s\n",
-                           canceled, ns->ns_name);
+                           canceled, ldlm_ns_name(ns));
 }
 
 int ldlm_replay_locks(struct obd_import *imp)
 }
 
 int ldlm_replay_locks(struct obd_import *imp)
@@ -2218,7 +2182,7 @@ int ldlm_replay_locks(struct obd_import *imp)
         if (ldlm_cancel_unused_locks_before_replay)
                 ldlm_cancel_unused_locks_for_replay(ns);
 
         if (ldlm_cancel_unused_locks_before_replay)
                 ldlm_cancel_unused_locks_for_replay(ns);
 
-        (void)ldlm_namespace_foreach(ns, ldlm_chain_lock_for_replay, &list);
+        ldlm_namespace_foreach(ns, ldlm_chain_lock_for_replay, &list);
 
         cfs_list_for_each_entry_safe(lock, next, &list, l_pending_chain) {
                 cfs_list_del_init(&lock->l_pending_chain);
 
         cfs_list_for_each_entry_safe(lock, next, &list, l_pending_chain) {
                 cfs_list_del_init(&lock->l_pending_chain);
index 9860694..0d34e40 100644 (file)
@@ -46,6 +46,7 @@
 # include <liblustre.h>
 #endif
 
 # include <liblustre.h>
 #endif
 
+#include <lustre_fid.h>
 #include <obd_class.h>
 #include "ldlm_internal.h"
 
 #include <obd_class.h>
 #include "ldlm_internal.h"
 
@@ -139,6 +140,31 @@ void ldlm_proc_cleanup(void)
                 lprocfs_remove(&ldlm_type_proc_dir);
 }
 
                 lprocfs_remove(&ldlm_type_proc_dir);
 }
 
+static int lprocfs_rd_ns_resources(char *page, char **start, off_t off,
+                                   int count, int *eof, void *data)
+{
+        struct ldlm_namespace *ns  = data;
+        __u64                  res = 0;
+        cfs_hash_bd_t          bd;
+        int                    i;
+
+        /* result is not strictly consistant */
+        cfs_hash_for_each_bucket(ns->ns_rs_hash, &bd, i)
+                res += cfs_hash_bd_count_get(&bd);
+        return lprocfs_rd_u64(page, start, off, count, eof, &res);
+}
+
+static int lprocfs_rd_ns_locks(char *page, char **start, off_t off,
+                               int count, int *eof, void *data)
+{
+        struct ldlm_namespace *ns = data;
+        __u64                  locks;
+
+        locks = lprocfs_stats_collector(ns->ns_stats, LDLM_NSS_LOCKS,
+                                        LPROCFS_FIELDS_FLAGS_SUM);
+        return lprocfs_rd_u64(page, start, off, count, eof, &locks);
+}
+
 static int lprocfs_rd_lru_size(char *page, char **start, off_t off,
                                int count, int *eof, void *data)
 {
 static int lprocfs_rd_lru_size(char *page, char **start, off_t off,
                                int count, int *eof, void *data)
 {
@@ -165,7 +191,7 @@ static int lprocfs_wr_lru_size(struct file *file, const char *buffer,
         if (strncmp(dummy, "clear", 5) == 0) {
                 CDEBUG(D_DLMTRACE,
                        "dropping all unused locks from namespace %s\n",
         if (strncmp(dummy, "clear", 5) == 0) {
                 CDEBUG(D_DLMTRACE,
                        "dropping all unused locks from namespace %s\n",
-                       ns->ns_name);
+                       ldlm_ns_name(ns));
                 if (ns_connect_lru_resize(ns)) {
                         int canceled, unused  = ns->ns_nr_unused;
 
                 if (ns_connect_lru_resize(ns)) {
                         int canceled, unused  = ns->ns_nr_unused;
 
@@ -205,19 +231,21 @@ static int lprocfs_wr_lru_size(struct file *file, const char *buffer,
 
                 CDEBUG(D_DLMTRACE,
                        "changing namespace %s unused locks from %u to %u\n",
 
                 CDEBUG(D_DLMTRACE,
                        "changing namespace %s unused locks from %u to %u\n",
-                       ns->ns_name, ns->ns_nr_unused, (unsigned int)tmp);
+                       ldlm_ns_name(ns), ns->ns_nr_unused,
+                       (unsigned int)tmp);
                 ldlm_cancel_lru(ns, tmp, LDLM_ASYNC, LDLM_CANCEL_PASSED);
 
                 if (!lru_resize) {
                         CDEBUG(D_DLMTRACE,
                                "disable lru_resize for namespace %s\n",
                 ldlm_cancel_lru(ns, tmp, LDLM_ASYNC, LDLM_CANCEL_PASSED);
 
                 if (!lru_resize) {
                         CDEBUG(D_DLMTRACE,
                                "disable lru_resize for namespace %s\n",
-                               ns->ns_name);
+                               ldlm_ns_name(ns));
                         ns->ns_connect_flags &= ~OBD_CONNECT_LRU_RESIZE;
                 }
         } else {
                 CDEBUG(D_DLMTRACE,
                        "changing namespace %s max_unused from %u to %u\n",
                         ns->ns_connect_flags &= ~OBD_CONNECT_LRU_RESIZE;
                 }
         } else {
                 CDEBUG(D_DLMTRACE,
                        "changing namespace %s max_unused from %u to %u\n",
-                       ns->ns_name, ns->ns_max_unused, (unsigned int)tmp);
+                       ldlm_ns_name(ns), ns->ns_max_unused,
+                       (unsigned int)tmp);
                 ns->ns_max_unused = (unsigned int)tmp;
                 ldlm_cancel_lru(ns, 0, LDLM_ASYNC, LDLM_CANCEL_PASSED);
 
                 ns->ns_max_unused = (unsigned int)tmp;
                 ldlm_cancel_lru(ns, 0, LDLM_ASYNC, LDLM_CANCEL_PASSED);
 
@@ -227,7 +255,7 @@ static int lprocfs_wr_lru_size(struct file *file, const char *buffer,
                     (ns->ns_orig_connect_flags & OBD_CONNECT_LRU_RESIZE)) {
                         CDEBUG(D_DLMTRACE,
                                "enable lru_resize for namespace %s\n",
                     (ns->ns_orig_connect_flags & OBD_CONNECT_LRU_RESIZE)) {
                         CDEBUG(D_DLMTRACE,
                                "enable lru_resize for namespace %s\n",
-                               ns->ns_name);
+                               ldlm_ns_name(ns));
                         ns->ns_connect_flags |= OBD_CONNECT_LRU_RESIZE;
                 }
         }
                         ns->ns_connect_flags |= OBD_CONNECT_LRU_RESIZE;
                 }
         }
@@ -235,150 +263,365 @@ static int lprocfs_wr_lru_size(struct file *file, const char *buffer,
         return count;
 }
 
         return count;
 }
 
-void ldlm_proc_namespace(struct ldlm_namespace *ns)
+void ldlm_namespace_proc_unregister(struct ldlm_namespace *ns)
+{
+        struct proc_dir_entry *dir;
+
+        dir = lprocfs_srch(ldlm_ns_proc_dir, ldlm_ns_name(ns));
+        if (dir == NULL) {
+                CERROR("dlm namespace %s has no procfs dir?\n",
+                       ldlm_ns_name(ns));
+        } else {
+                lprocfs_remove(&dir);
+        }
+
+        if (ns->ns_stats != NULL)
+                lprocfs_free_stats(&ns->ns_stats);
+}
+
+int ldlm_namespace_proc_register(struct ldlm_namespace *ns)
 {
         struct lprocfs_vars lock_vars[2];
         char lock_name[MAX_STRING_SIZE + 1];
 
         LASSERT(ns != NULL);
 {
         struct lprocfs_vars lock_vars[2];
         char lock_name[MAX_STRING_SIZE + 1];
 
         LASSERT(ns != NULL);
-        LASSERT(ns->ns_name != NULL);
+        LASSERT(ns->ns_rs_hash != NULL);
+
+        ns->ns_stats = lprocfs_alloc_stats(LDLM_NSS_LAST, 0);
+        if (ns->ns_stats == NULL)
+                return -ENOMEM;
+
+        lprocfs_counter_init(ns->ns_stats, LDLM_NSS_LOCKS,
+                             LPROCFS_CNTR_AVGMINMAX, "locks", "locks");
 
         lock_name[MAX_STRING_SIZE] = '\0';
 
         memset(lock_vars, 0, sizeof(lock_vars));
         lock_vars[0].name = lock_name;
 
 
         lock_name[MAX_STRING_SIZE] = '\0';
 
         memset(lock_vars, 0, sizeof(lock_vars));
         lock_vars[0].name = lock_name;
 
-        snprintf(lock_name, MAX_STRING_SIZE, "%s/resource_count", ns->ns_name);
-        lock_vars[0].data = &ns->ns_refcount;
-        lock_vars[0].read_fptr = lprocfs_rd_atomic;
+        snprintf(lock_name, MAX_STRING_SIZE, "%s/resource_count",
+                 ldlm_ns_name(ns));
+        lock_vars[0].data = ns;
+        lock_vars[0].read_fptr = lprocfs_rd_ns_resources;
         lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
         lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
-        snprintf(lock_name, MAX_STRING_SIZE, "%s/lock_count", ns->ns_name);
-        lock_vars[0].data = &ns->ns_locks;
-        lock_vars[0].read_fptr = lprocfs_rd_atomic;
+        snprintf(lock_name, MAX_STRING_SIZE, "%s/lock_count",
+                 ldlm_ns_name(ns));
+        lock_vars[0].data = ns;
+        lock_vars[0].read_fptr = lprocfs_rd_ns_locks;
         lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
         if (ns_is_client(ns)) {
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lock_unused_count",
         lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
         if (ns_is_client(ns)) {
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lock_unused_count",
-                         ns->ns_name);
+                         ldlm_ns_name(ns));
                 lock_vars[0].data = &ns->ns_nr_unused;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lru_size",
                 lock_vars[0].data = &ns->ns_nr_unused;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lru_size",
-                         ns->ns_name);
+                         ldlm_ns_name(ns));
                 lock_vars[0].data = ns;
                 lock_vars[0].read_fptr = lprocfs_rd_lru_size;
                 lock_vars[0].write_fptr = lprocfs_wr_lru_size;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lru_max_age",
                 lock_vars[0].data = ns;
                 lock_vars[0].read_fptr = lprocfs_rd_lru_size;
                 lock_vars[0].write_fptr = lprocfs_wr_lru_size;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lru_max_age",
-                         ns->ns_name);
+                         ldlm_ns_name(ns));
                 lock_vars[0].data = &ns->ns_max_age;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lock_vars[0].write_fptr = lprocfs_wr_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
         } else {
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/ctime_age_limit",
                 lock_vars[0].data = &ns->ns_max_age;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lock_vars[0].write_fptr = lprocfs_wr_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
         } else {
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/ctime_age_limit",
-                         ns->ns_name);
+                         ldlm_ns_name(ns));
                 lock_vars[0].data = &ns->ns_ctime_age_limit;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lock_vars[0].write_fptr = lprocfs_wr_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lock_timeouts",
                 lock_vars[0].data = &ns->ns_ctime_age_limit;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lock_vars[0].write_fptr = lprocfs_wr_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/lock_timeouts",
-                         ns->ns_name);
+                         ldlm_ns_name(ns));
                 lock_vars[0].data = &ns->ns_timeouts;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/max_nolock_bytes",
                 lock_vars[0].data = &ns->ns_timeouts;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/max_nolock_bytes",
-                         ns->ns_name);
+                         ldlm_ns_name(ns));
                 lock_vars[0].data = &ns->ns_max_nolock_size;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lock_vars[0].write_fptr = lprocfs_wr_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/contention_seconds",
                 lock_vars[0].data = &ns->ns_max_nolock_size;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lock_vars[0].write_fptr = lprocfs_wr_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/contention_seconds",
-                         ns->ns_name);
+                         ldlm_ns_name(ns));
                 lock_vars[0].data = &ns->ns_contention_time;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lock_vars[0].write_fptr = lprocfs_wr_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/contended_locks",
                 lock_vars[0].data = &ns->ns_contention_time;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lock_vars[0].write_fptr = lprocfs_wr_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
 
                 snprintf(lock_name, MAX_STRING_SIZE, "%s/contended_locks",
-                         ns->ns_name);
+                         ldlm_ns_name(ns));
                 lock_vars[0].data = &ns->ns_contended_locks;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lock_vars[0].write_fptr = lprocfs_wr_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
         }
                 lock_vars[0].data = &ns->ns_contended_locks;
                 lock_vars[0].read_fptr = lprocfs_rd_uint;
                 lock_vars[0].write_fptr = lprocfs_wr_uint;
                 lprocfs_add_vars(ldlm_ns_proc_dir, lock_vars, 0);
         }
+        return 0;
 }
 #undef MAX_STRING_SIZE
 }
 #undef MAX_STRING_SIZE
-#else
-#define ldlm_proc_namespace(ns) do {} while (0)
+#else /* LPROCFS */
+
+#define ldlm_namespace_proc_unregister(ns)      ({;})
+#define ldlm_namespace_proc_register(ns)        ({0;})
+
 #endif /* LPROCFS */
 
 #endif /* LPROCFS */
 
+static unsigned ldlm_res_hop_hash(cfs_hash_t *hs, void *key, unsigned mask)
+{
+        struct ldlm_res_id     *id  = key;
+        unsigned                val = 0;
+        unsigned                i;
+
+        for (i = 0; i < RES_NAME_SIZE; i++)
+                val += id->name[i];
+        return val & mask;
+}
+
+static unsigned ldlm_res_hop_fid_hash(cfs_hash_t *hs, void *key, unsigned mask)
+{
+        struct ldlm_res_id *id = key;
+        struct lu_fid       fid;
+        __u64               hash;
+
+        fid.f_seq = id->name[LUSTRE_RES_ID_SEQ_OFF];
+        fid.f_oid = (__u32)id->name[LUSTRE_RES_ID_OID_OFF];
+        fid.f_ver = (__u32)id->name[LUSTRE_RES_ID_VER_OFF];
+
+        hash = fid_flatten(&fid);
+        hash = cfs_hash_long(hash, hs->hs_bkt_bits);
+        /* ignore a few low bits */
+        if (id->name[LUSTRE_RES_ID_HSH_OFF] != 0)
+                hash += id->name[LUSTRE_RES_ID_HSH_OFF] >> 5;
+        else
+                hash = hash >> 5;
+        hash <<= hs->hs_cur_bits - hs->hs_bkt_bits;
+        hash |= ldlm_res_hop_hash(hs, key, CFS_HASH_NBKT(hs) - 1);
+
+        return hash & mask;
+}
+
+static void *ldlm_res_hop_key(cfs_hlist_node_t *hnode)
+{
+        struct ldlm_resource   *res;
+
+        res = cfs_hlist_entry(hnode, struct ldlm_resource, lr_hash);
+        return &res->lr_name;
+}
+
+static int ldlm_res_eq(const struct ldlm_res_id *res0,
+                       const struct ldlm_res_id *res1)
+{
+        return !memcmp(res0, res1, sizeof(*res0));
+}
+
+static int ldlm_res_hop_keycmp(void *key, cfs_hlist_node_t *hnode)
+{
+        struct ldlm_resource   *res;
+
+        res = cfs_hlist_entry(hnode, struct ldlm_resource, lr_hash);
+        return ldlm_res_eq((const struct ldlm_res_id *)key,
+                           (const struct ldlm_res_id *)&res->lr_name);
+}
+
+static void *ldlm_res_hop_object(cfs_hlist_node_t *hnode)
+{
+        return cfs_hlist_entry(hnode, struct ldlm_resource, lr_hash);
+}
+
+static void ldlm_res_hop_get_locked(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
+{
+        struct ldlm_resource *res;
+
+        res = cfs_hlist_entry(hnode, struct ldlm_resource, lr_hash);
+        ldlm_resource_getref(res);
+        LDLM_RESOURCE_ADDREF(res);
+}
+
+static void ldlm_res_hop_put_locked(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
+{
+        struct ldlm_resource *res;
+
+        res = cfs_hlist_entry(hnode, struct ldlm_resource, lr_hash);
+        /* cfs_hash_for_each_nolock is the only chance we call it */
+        LDLM_RESOURCE_DELREF(res);
+        ldlm_resource_putref_locked(res);
+}
+
+static void ldlm_res_hop_put(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
+{
+        struct ldlm_resource *res;
+
+        res = cfs_hlist_entry(hnode, struct ldlm_resource, lr_hash);
+        LDLM_RESOURCE_DELREF(res);
+        ldlm_resource_putref(res);
+}
+
+cfs_hash_ops_t ldlm_ns_hash_ops = {
+        .hs_hash        = ldlm_res_hop_hash,
+        .hs_key         = ldlm_res_hop_key,
+        .hs_keycmp      = ldlm_res_hop_keycmp,
+        .hs_keycpy      = NULL,
+        .hs_object      = ldlm_res_hop_object,
+        .hs_get         = ldlm_res_hop_get_locked,
+        .hs_put_locked  = ldlm_res_hop_put_locked,
+        .hs_put         = ldlm_res_hop_put
+};
+
+cfs_hash_ops_t ldlm_ns_fid_hash_ops = {
+        .hs_hash        = ldlm_res_hop_fid_hash,
+        .hs_key         = ldlm_res_hop_key,
+        .hs_keycmp      = ldlm_res_hop_keycmp,
+        .hs_keycpy      = NULL,
+        .hs_object      = ldlm_res_hop_object,
+        .hs_get         = ldlm_res_hop_get_locked,
+        .hs_put_locked  = ldlm_res_hop_put_locked,
+        .hs_put         = ldlm_res_hop_put
+};
+
+typedef struct {
+        ldlm_ns_type_t  nsd_type;
+        /** hash bucket bits */
+        unsigned        nsd_bkt_bits;
+        /** hash bits */
+        unsigned        nsd_all_bits;
+        /** hash operations */
+        cfs_hash_ops_t *nsd_hops;
+} ldlm_ns_hash_def_t;
+
+ldlm_ns_hash_def_t ldlm_ns_hash_defs[] =
+{
+        {
+                .nsd_type       = LDLM_NS_TYPE_MDC,
+                .nsd_bkt_bits   = 11,
+                .nsd_all_bits   = 15,
+                .nsd_hops       = &ldlm_ns_fid_hash_ops,
+        },
+        {
+                .nsd_type       = LDLM_NS_TYPE_MDT,
+                .nsd_bkt_bits   = 14,
+                .nsd_all_bits   = 21,
+                .nsd_hops       = &ldlm_ns_fid_hash_ops,
+        },
+        {
+                .nsd_type       = LDLM_NS_TYPE_OSC,
+                .nsd_bkt_bits   = 8,
+                .nsd_all_bits   = 12,
+                .nsd_hops       = &ldlm_ns_hash_ops,
+        },
+        {
+                .nsd_type       = LDLM_NS_TYPE_OST,
+                .nsd_bkt_bits   = 11,
+                .nsd_all_bits   = 17,
+                .nsd_hops       = &ldlm_ns_hash_ops,
+        },
+        {
+                .nsd_type       = LDLM_NS_TYPE_MGC,
+                .nsd_bkt_bits   = 4,
+                .nsd_all_bits   = 4,
+                .nsd_hops       = &ldlm_ns_hash_ops,
+        },
+        {
+                .nsd_type       = LDLM_NS_TYPE_MGT,
+                .nsd_bkt_bits   = 4,
+                .nsd_all_bits   = 4,
+                .nsd_hops       = &ldlm_ns_hash_ops,
+        },
+        {
+                .nsd_type       = LDLM_NS_TYPE_UNKNOWN,
+        },
+};
+
 struct ldlm_namespace *ldlm_namespace_new(struct obd_device *obd, char *name,
 struct ldlm_namespace *ldlm_namespace_new(struct obd_device *obd, char *name,
-                                          ldlm_side_t client, ldlm_appetite_t apt)
+                                          ldlm_side_t client,
+                                          ldlm_appetite_t apt,
+                                          ldlm_ns_type_t ns_type)
 {
         struct ldlm_namespace *ns = NULL;
 {
         struct ldlm_namespace *ns = NULL;
-        cfs_list_t *bucket;
-        int rc, idx, namelen;
+        struct ldlm_ns_bucket *nsb;
+        ldlm_ns_hash_def_t    *nsd;
+        cfs_hash_bd_t          bd;
+        int                    idx;
+        int                    rc;
         ENTRY;
 
         ENTRY;
 
+        LASSERT(obd != NULL);
+
         rc = ldlm_get_ref();
         if (rc) {
                 CERROR("ldlm_get_ref failed: %d\n", rc);
                 RETURN(NULL);
         }
 
         rc = ldlm_get_ref();
         if (rc) {
                 CERROR("ldlm_get_ref failed: %d\n", rc);
                 RETURN(NULL);
         }
 
+        for (idx = 0;;idx++) {
+                nsd = &ldlm_ns_hash_defs[idx];
+                if (nsd->nsd_type == LDLM_NS_TYPE_UNKNOWN) {
+                        CERROR("Unknown type %d for ns %s\n", ns_type, name);
+                        GOTO(out_ref, NULL);
+                }
+
+                if (nsd->nsd_type == ns_type)
+                        break;
+        }
+
         OBD_ALLOC_PTR(ns);
         if (!ns)
                 GOTO(out_ref, NULL);
 
         OBD_ALLOC_PTR(ns);
         if (!ns)
                 GOTO(out_ref, NULL);
 
-        OBD_VMALLOC(ns->ns_hash, sizeof(*ns->ns_hash) * RES_HASH_SIZE);
-        if (!ns->ns_hash)
+        ns->ns_rs_hash = cfs_hash_create(name,
+                                         nsd->nsd_all_bits, nsd->nsd_all_bits,
+                                         nsd->nsd_bkt_bits, sizeof(*nsb),
+                                         CFS_HASH_MIN_THETA,
+                                         CFS_HASH_MAX_THETA,
+                                         nsd->nsd_hops,
+                                         CFS_HASH_DEPTH |
+                                         CFS_HASH_BIGNAME |
+                                         CFS_HASH_SPIN_BKTLOCK |
+                                         CFS_HASH_NO_ITEMREF);
+        if (ns->ns_rs_hash == NULL)
                 GOTO(out_ns, NULL);
 
                 GOTO(out_ns, NULL);
 
-        ns->ns_appetite = apt;
-
-        LASSERT(obd != NULL);
-        ns->ns_obd = obd;
-
-        namelen = strlen(name);
-        OBD_ALLOC(ns->ns_name, namelen + 1);
-        if (!ns->ns_name)
-                GOTO(out_hash, NULL);
+        cfs_hash_for_each_bucket(ns->ns_rs_hash, &bd, idx) {
+                nsb = cfs_hash_bd_extra_get(ns->ns_rs_hash, &bd);
+                at_init(&nsb->nsb_at_estimate, ldlm_enqueue_min, 0);
+                nsb->nsb_namespace = ns;
+        }
 
 
-        strcpy(ns->ns_name, name);
+        ns->ns_obd      = obd;
+        ns->ns_appetite = apt;
+        ns->ns_client   = client;
 
 
-        CFS_INIT_LIST_HEAD(&ns->ns_root_list);
         CFS_INIT_LIST_HEAD(&ns->ns_list_chain);
         CFS_INIT_LIST_HEAD(&ns->ns_list_chain);
-        ns->ns_refcount = 0;
-        ns->ns_client = client;
-        cfs_spin_lock_init(&ns->ns_hash_lock);
-        cfs_atomic_set(&ns->ns_locks, 0);
-        ns->ns_resources = 0;
+        CFS_INIT_LIST_HEAD(&ns->ns_unused_list);
+        cfs_spin_lock_init(&ns->ns_lock);
+        cfs_atomic_set(&ns->ns_bref, 0);
         cfs_waitq_init(&ns->ns_waitq);
         cfs_waitq_init(&ns->ns_waitq);
-        ns->ns_max_nolock_size = NS_DEFAULT_MAX_NOLOCK_BYTES;
-        ns->ns_contention_time = NS_DEFAULT_CONTENTION_SECONDS;
-        ns->ns_contended_locks = NS_DEFAULT_CONTENDED_LOCKS;
 
 
-        for (bucket = ns->ns_hash + RES_HASH_SIZE - 1; bucket >= ns->ns_hash;
-             bucket--)
-                CFS_INIT_LIST_HEAD(bucket);
+        ns->ns_max_nolock_size    = NS_DEFAULT_MAX_NOLOCK_BYTES;
+        ns->ns_contention_time    = NS_DEFAULT_CONTENTION_SECONDS;
+        ns->ns_contended_locks    = NS_DEFAULT_CONTENDED_LOCKS;
 
 
-        CFS_INIT_LIST_HEAD(&ns->ns_unused_list);
-        ns->ns_nr_unused = 0;
-        ns->ns_max_unused = LDLM_DEFAULT_LRU_SIZE;
-        ns->ns_max_age = LDLM_DEFAULT_MAX_ALIVE;
-        ns->ns_ctime_age_limit = LDLM_CTIME_AGE_LIMIT;
-        ns->ns_timeouts = 0;
-        cfs_spin_lock_init(&ns->ns_unused_lock);
+        ns->ns_nr_unused          = 0;
+        ns->ns_max_unused         = LDLM_DEFAULT_LRU_SIZE;
+        ns->ns_max_age            = LDLM_DEFAULT_MAX_ALIVE;
+        ns->ns_ctime_age_limit    = LDLM_CTIME_AGE_LIMIT;
+        ns->ns_timeouts           = 0;
         ns->ns_orig_connect_flags = 0;
         ns->ns_orig_connect_flags = 0;
-        ns->ns_connect_flags = 0;
-        ldlm_proc_namespace(ns);
+        ns->ns_connect_flags      = 0;
+        rc = ldlm_namespace_proc_register(ns);
+        if (rc != 0) {
+                CERROR("Can't initialize ns proc, rc %d\n", rc);
+                GOTO(out_hash, rc);
+        }
 
         idx = cfs_atomic_read(ldlm_namespace_nr(client));
         rc = ldlm_pool_init(&ns->ns_pool, ns, idx, client);
 
         idx = cfs_atomic_read(ldlm_namespace_nr(client));
         rc = ldlm_pool_init(&ns->ns_pool, ns, idx, client);
@@ -387,15 +630,13 @@ struct ldlm_namespace *ldlm_namespace_new(struct obd_device *obd, char *name,
                 GOTO(out_proc, rc);
         }
 
                 GOTO(out_proc, rc);
         }
 
-        at_init(&ns->ns_at_estimate, ldlm_enqueue_min, 0);
-
         ldlm_namespace_register(ns, client);
         RETURN(ns);
 out_proc:
         ldlm_namespace_register(ns, client);
         RETURN(ns);
 out_proc:
+        ldlm_namespace_proc_unregister(ns);
         ldlm_namespace_cleanup(ns, 0);
         ldlm_namespace_cleanup(ns, 0);
-        OBD_FREE(ns->ns_name, namelen + 1);
 out_hash:
 out_hash:
-        OBD_VFREE(ns->ns_hash, sizeof(*ns->ns_hash) * RES_HASH_SIZE);
+        cfs_hash_putref(ns->ns_rs_hash);
 out_ns:
         OBD_FREE_PTR(ns);
 out_ref:
 out_ns:
         OBD_FREE_PTR(ns);
 out_ref:
@@ -407,17 +648,13 @@ extern struct ldlm_lock *ldlm_lock_get(struct ldlm_lock *lock);
 
 /* If flags contains FL_LOCAL_ONLY, don't try to tell the server, just cleanup.
  * This is currently only used for recovery, and we make certain assumptions
 
 /* If flags contains FL_LOCAL_ONLY, don't try to tell the server, just cleanup.
  * This is currently only used for recovery, and we make certain assumptions
- * as a result--notably, that we shouldn't cancel locks with refs. -phil
- *
- * Called with the ns_lock held. */
+ * as a result--notably, that we shouldn't cancel locks with refs. -phil */
 static void cleanup_resource(struct ldlm_resource *res, cfs_list_t *q,
                              int flags)
 {
         cfs_list_t *tmp;
         int rc = 0, client = ns_is_client(ldlm_res_to_ns(res));
         int local_only = (flags & LDLM_FL_LOCAL_ONLY);
 static void cleanup_resource(struct ldlm_resource *res, cfs_list_t *q,
                              int flags)
 {
         cfs_list_t *tmp;
         int rc = 0, client = ns_is_client(ldlm_res_to_ns(res));
         int local_only = (flags & LDLM_FL_LOCAL_ONLY);
-        ENTRY;
-
 
         do {
                 struct ldlm_lock *lock = NULL;
 
         do {
                 struct ldlm_lock *lock = NULL;
@@ -482,62 +719,50 @@ static void cleanup_resource(struct ldlm_resource *res, cfs_list_t *q,
                 }
                 LDLM_LOCK_RELEASE(lock);
         } while (1);
                 }
                 LDLM_LOCK_RELEASE(lock);
         } while (1);
+}
 
 
-        EXIT;
+static int ldlm_resource_clean(cfs_hash_t *hs, cfs_hash_bd_t *bd,
+                               cfs_hlist_node_t *hnode, void *arg)
+{
+        struct ldlm_resource *res = cfs_hash_object(hs, hnode);
+        int    flags = (int)(unsigned long)arg;
+
+        cleanup_resource(res, &res->lr_granted, flags);
+        cleanup_resource(res, &res->lr_converting, flags);
+        cleanup_resource(res, &res->lr_waiting, flags);
+
+        return 0;
 }
 
 }
 
-int ldlm_namespace_cleanup(struct ldlm_namespace *ns, int flags)
+static int ldlm_resource_complain(cfs_hash_t *hs, cfs_hash_bd_t *bd,
+                                  cfs_hlist_node_t *hnode, void *arg)
 {
 {
-        cfs_list_t *tmp;
-        int i;
+        struct ldlm_resource  *res = cfs_hash_object(hs, hnode);
 
 
+        CERROR("Namespace %s resource refcount nonzero "
+               "(%d) after lock cleanup; forcing "
+               "cleanup.\n",
+               ldlm_ns_name(ldlm_res_to_ns(res)),
+               cfs_atomic_read(&res->lr_refcount) - 1);
+
+        CERROR("Resource: %p ("LPU64"/"LPU64"/"LPU64"/"
+               LPU64") (rc: %d)\n", res,
+               res->lr_name.name[0], res->lr_name.name[1],
+               res->lr_name.name[2], res->lr_name.name[3],
+               cfs_atomic_read(&res->lr_refcount) - 1);
+        return 0;
+}
+
+int ldlm_namespace_cleanup(struct ldlm_namespace *ns, int flags)
+{
         if (ns == NULL) {
                 CDEBUG(D_INFO, "NULL ns, skipping cleanup\n");
                 return ELDLM_OK;
         }
 
         if (ns == NULL) {
                 CDEBUG(D_INFO, "NULL ns, skipping cleanup\n");
                 return ELDLM_OK;
         }
 
-        for (i = 0; i < RES_HASH_SIZE; i++) {
-                cfs_spin_lock(&ns->ns_hash_lock);
-                tmp = ns->ns_hash[i].next;
-                while (tmp != &(ns->ns_hash[i])) {
-                        struct ldlm_resource *res;
-                        res = cfs_list_entry(tmp, struct ldlm_resource,
-                                             lr_hash);
-                        ldlm_resource_getref(res);
-                        cfs_spin_unlock(&ns->ns_hash_lock);
-                        LDLM_RESOURCE_ADDREF(res);
-
-                        cleanup_resource(res, &res->lr_granted, flags);
-                        cleanup_resource(res, &res->lr_converting, flags);
-                        cleanup_resource(res, &res->lr_waiting, flags);
-
-                        cfs_spin_lock(&ns->ns_hash_lock);
-                        tmp = tmp->next;
-
-                        /* XXX: former stuff caused issues in case of race
-                         * between ldlm_namespace_cleanup() and lockd() when
-                         * client gets blocking ast when lock gets distracted by
-                         * server. This is 1_4 branch solution, let's see how
-                         * will it behave. */
-                        LDLM_RESOURCE_DELREF(res);
-                        if (!ldlm_resource_putref_locked(res)) {
-                                CERROR("Namespace %s resource refcount nonzero "
-                                       "(%d) after lock cleanup; forcing "
-                                       "cleanup.\n",
-                                       ns->ns_name,
-                                       cfs_atomic_read(&res->lr_refcount));
-                                CERROR("Resource: %p ("LPU64"/"LPU64"/"LPU64"/"
-                                       LPU64") (rc: %d)\n", res,
-                                       res->lr_name.name[0],
-                                       res->lr_name.name[1],
-                                       res->lr_name.name[2],
-                                       res->lr_name.name[3],
-                                       cfs_atomic_read(&res->lr_refcount));
-                        }
-                }
-                cfs_spin_unlock(&ns->ns_hash_lock);
-        }
-
+        cfs_hash_for_each_nolock(ns->ns_rs_hash, ldlm_resource_clean,
+                                 (void *)(unsigned long)flags);
+        cfs_hash_for_each_nolock(ns->ns_rs_hash, ldlm_resource_complain, NULL);
         return ELDLM_OK;
 }
 
         return ELDLM_OK;
 }
 
@@ -548,38 +773,38 @@ static int __ldlm_namespace_free(struct ldlm_namespace *ns, int force)
         /* At shutdown time, don't call the cancellation callback */
         ldlm_namespace_cleanup(ns, force ? LDLM_FL_LOCAL_ONLY : 0);
 
         /* At shutdown time, don't call the cancellation callback */
         ldlm_namespace_cleanup(ns, force ? LDLM_FL_LOCAL_ONLY : 0);
 
-        if (ns->ns_refcount > 0) {
+        if (cfs_atomic_read(&ns->ns_bref) > 0) {
                 struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
                 int rc;
                 CDEBUG(D_DLMTRACE,
                        "dlm namespace %s free waiting on refcount %d\n",
                 struct l_wait_info lwi = LWI_INTR(LWI_ON_SIGNAL_NOOP, NULL);
                 int rc;
                 CDEBUG(D_DLMTRACE,
                        "dlm namespace %s free waiting on refcount %d\n",
-                       ns->ns_name, ns->ns_refcount);
+                       ldlm_ns_name(ns), cfs_atomic_read(&ns->ns_bref));
 force_wait:
                 if (force)
                         lwi = LWI_TIMEOUT(obd_timeout * CFS_HZ / 4, NULL, NULL);
 
                 rc = l_wait_event(ns->ns_waitq,
 force_wait:
                 if (force)
                         lwi = LWI_TIMEOUT(obd_timeout * CFS_HZ / 4, NULL, NULL);
 
                 rc = l_wait_event(ns->ns_waitq,
-                                  ns->ns_refcount == 0, &lwi);
+                                  cfs_atomic_read(&ns->ns_bref) == 0, &lwi);
 
                 /* Forced cleanups should be able to reclaim all references,
                  * so it's safe to wait forever... we can't leak locks... */
                 if (force && rc == -ETIMEDOUT) {
                         LCONSOLE_ERROR("Forced cleanup waiting for %s "
                                        "namespace with %d resources in use, "
 
                 /* Forced cleanups should be able to reclaim all references,
                  * so it's safe to wait forever... we can't leak locks... */
                 if (force && rc == -ETIMEDOUT) {
                         LCONSOLE_ERROR("Forced cleanup waiting for %s "
                                        "namespace with %d resources in use, "
-                                       "(rc=%d)\n", ns->ns_name,
-                                       ns->ns_refcount, rc);
+                                       "(rc=%d)\n", ldlm_ns_name(ns),
+                                       cfs_atomic_read(&ns->ns_bref), rc);
                         GOTO(force_wait, rc);
                 }
 
                         GOTO(force_wait, rc);
                 }
 
-                if (ns->ns_refcount) {
+                if (cfs_atomic_read(&ns->ns_bref)) {
                         LCONSOLE_ERROR("Cleanup waiting for %s namespace "
                                        "with %d resources in use, (rc=%d)\n",
                         LCONSOLE_ERROR("Cleanup waiting for %s namespace "
                                        "with %d resources in use, (rc=%d)\n",
-                                       ns->ns_name,
-                                       ns->ns_refcount, rc);
+                                       ldlm_ns_name(ns),
+                                       cfs_atomic_read(&ns->ns_bref), rc);
                         RETURN(ELDLM_NAMESPACE_EXISTS);
                 }
                         RETURN(ELDLM_NAMESPACE_EXISTS);
                 }
-                CDEBUG(D_DLMTRACE,
-                       "dlm namespace %s free done waiting\n", ns->ns_name);
+                CDEBUG(D_DLMTRACE, "dlm namespace %s free done waiting\n",
+                       ldlm_ns_name(ns));
         }
 
         RETURN(ELDLM_OK);
         }
 
         RETURN(ELDLM_OK);
@@ -651,22 +876,8 @@ void ldlm_namespace_free_post(struct ldlm_namespace *ns)
          */
         ldlm_pool_fini(&ns->ns_pool);
 
          */
         ldlm_pool_fini(&ns->ns_pool);
 
-#ifdef LPROCFS
-        {
-                struct proc_dir_entry *dir;
-                dir = lprocfs_srch(ldlm_ns_proc_dir, ns->ns_name);
-                if (dir == NULL) {
-                        CERROR("dlm namespace %s has no procfs dir?\n",
-                               ns->ns_name);
-                } else {
-                        lprocfs_remove(&dir);
-                }
-        }
-#endif
-
-        OBD_VFREE(ns->ns_hash, sizeof(*ns->ns_hash) * RES_HASH_SIZE);
-        OBD_FREE(ns->ns_name, strlen(ns->ns_name) + 1);
-
+        ldlm_namespace_proc_unregister(ns);
+        cfs_hash_putref(ns->ns_rs_hash);
         /*
          * Namespace \a ns should be not on list in this time, otherwise this
          * will cause issues realted to using freed \a ns in pools thread.
         /*
          * Namespace \a ns should be not on list in this time, otherwise this
          * will cause issues realted to using freed \a ns in pools thread.
@@ -703,32 +914,17 @@ void ldlm_namespace_free(struct ldlm_namespace *ns,
         ldlm_namespace_free_post(ns);
 }
 
         ldlm_namespace_free_post(ns);
 }
 
-
-void ldlm_namespace_get_locked(struct ldlm_namespace *ns)
-{
-        ns->ns_refcount++;
-}
-
 void ldlm_namespace_get(struct ldlm_namespace *ns)
 {
 void ldlm_namespace_get(struct ldlm_namespace *ns)
 {
-        cfs_spin_lock(&ns->ns_hash_lock);
-        ldlm_namespace_get_locked(ns);
-        cfs_spin_unlock(&ns->ns_hash_lock);
+        cfs_atomic_inc(&ns->ns_bref);
 }
 
 }
 
-void ldlm_namespace_put_locked(struct ldlm_namespace *ns, int wakeup)
+void ldlm_namespace_put(struct ldlm_namespace *ns)
 {
 {
-        LASSERT(ns->ns_refcount > 0);
-        ns->ns_refcount--;
-        if (ns->ns_refcount == 0 && wakeup)
+        if (cfs_atomic_dec_and_lock(&ns->ns_bref, &ns->ns_lock)) {
                 cfs_waitq_signal(&ns->ns_waitq);
                 cfs_waitq_signal(&ns->ns_waitq);
-}
-
-void ldlm_namespace_put(struct ldlm_namespace *ns, int wakeup)
-{
-        cfs_spin_lock(&ns->ns_hash_lock);
-        ldlm_namespace_put_locked(ns, wakeup);
-        cfs_spin_unlock(&ns->ns_hash_lock);
+                cfs_spin_unlock(&ns->ns_lock);
+        }
 }
 
 /* Register @ns in the list of namespaces */
 }
 
 /* Register @ns in the list of namespaces */
@@ -772,19 +968,6 @@ struct ldlm_namespace *ldlm_namespace_first_locked(ldlm_side_t client)
         return container_of(ldlm_namespace_list(client)->next,
                 struct ldlm_namespace, ns_list_chain);
 }
         return container_of(ldlm_namespace_list(client)->next,
                 struct ldlm_namespace, ns_list_chain);
 }
-static __u32 ldlm_hash_fn(struct ldlm_resource *parent,
-                          const struct ldlm_res_id *name)
-{
-        __u32 hash = 0;
-        int i;
-
-        for (i = 0; i < RES_NAME_SIZE; i++)
-                hash += name->name[i];
-
-        hash += (__u32)((unsigned long)parent >> 4);
-
-        return (hash & RES_HASH_MASK);
-}
 
 static struct ldlm_resource *ldlm_resource_new(void)
 {
 
 static struct ldlm_resource *ldlm_resource_new(void)
 {
@@ -795,9 +978,6 @@ static struct ldlm_resource *ldlm_resource_new(void)
         if (res == NULL)
                 return NULL;
 
         if (res == NULL)
                 return NULL;
 
-        memset(res, 0, sizeof(*res));
-
-        CFS_INIT_LIST_HEAD(&res->lr_childof);
         CFS_INIT_LIST_HEAD(&res->lr_granted);
         CFS_INIT_LIST_HEAD(&res->lr_converting);
         CFS_INIT_LIST_HEAD(&res->lr_waiting);
         CFS_INIT_LIST_HEAD(&res->lr_granted);
         CFS_INIT_LIST_HEAD(&res->lr_converting);
         CFS_INIT_LIST_HEAD(&res->lr_waiting);
@@ -820,75 +1000,78 @@ static struct ldlm_resource *ldlm_resource_new(void)
         return res;
 }
 
         return res;
 }
 
-/* must be called with hash lock held */
-static struct ldlm_resource *
-ldlm_resource_find(struct ldlm_namespace *ns, const struct ldlm_res_id *name,
-                   __u32 hash)
+/* Args: unlocked namespace
+ *  * Locks: takes and releases NS hash-lock and res->lr_lock
+ *   * Returns: referenced, unlocked ldlm_resource or NULL */
+struct ldlm_resource *
+ldlm_resource_get(struct ldlm_namespace *ns, struct ldlm_resource *parent,
+                  const struct ldlm_res_id *name, ldlm_type_t type, int create)
 {
 {
-        cfs_list_t *bucket, *tmp;
+        cfs_hlist_node_t     *hnode;
         struct ldlm_resource *res;
         struct ldlm_resource *res;
+        cfs_hash_bd_t         bd;
+        __u64                 version;
 
 
-        LASSERT_SPIN_LOCKED(&ns->ns_hash_lock);
-        bucket = ns->ns_hash + hash;
+        LASSERT(ns != NULL);
+        LASSERT(parent == NULL);
+        LASSERT(ns->ns_rs_hash != NULL);
+        LASSERT(name->name[0] != 0);
 
 
-        cfs_list_for_each(tmp, bucket) {
-                res = cfs_list_entry(tmp, struct ldlm_resource, lr_hash);
-                if (memcmp(&res->lr_name, name, sizeof(res->lr_name)) == 0)
-                        return res;
+        cfs_hash_bd_get_and_lock(ns->ns_rs_hash, (void *)name, &bd, 0);
+        hnode = cfs_hash_bd_lookup_locked(ns->ns_rs_hash, &bd, (void *)name);
+        if (hnode != NULL) {
+                cfs_hash_bd_unlock(ns->ns_rs_hash, &bd, 0);
+                res = cfs_hlist_entry(hnode, struct ldlm_resource, lr_hash);
+                /* synchronize WRT resource creation */
+                if (ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
+                        cfs_down(&res->lr_lvb_sem);
+                        cfs_up(&res->lr_lvb_sem);
+                }
+                return res;
         }
 
         }
 
-        return NULL;
-}
+        version = cfs_hash_bd_version_get(&bd);
+        cfs_hash_bd_unlock(ns->ns_rs_hash, &bd, 0);
 
 
-/* Args: locked namespace
- * Returns: newly-allocated, referenced, unlocked resource */
-static struct ldlm_resource *
-ldlm_resource_add(struct ldlm_namespace *ns, struct ldlm_resource *parent,
-                  const struct ldlm_res_id *name, __u32 hash, ldlm_type_t type)
-{
-        cfs_list_t *bucket;
-        struct ldlm_resource *res, *old_res;
-        ENTRY;
+        if (create == 0)
+                return NULL;
 
         LASSERTF(type >= LDLM_MIN_TYPE && type < LDLM_MAX_TYPE,
                  "type: %d\n", type);
 
         LASSERTF(type >= LDLM_MIN_TYPE && type < LDLM_MAX_TYPE,
                  "type: %d\n", type);
-
         res = ldlm_resource_new();
         if (!res)
         res = ldlm_resource_new();
         if (!res)
-                RETURN(NULL);
+                return NULL;
 
 
-        res->lr_name = *name;
-        res->lr_namespace = ns;
-        res->lr_type = type;
+        res->lr_ns_bucket  = cfs_hash_bd_extra_get(ns->ns_rs_hash, &bd);
+        res->lr_name       = *name;
+        res->lr_type       = type;
         res->lr_most_restr = LCK_NL;
 
         res->lr_most_restr = LCK_NL;
 
-        cfs_spin_lock(&ns->ns_hash_lock);
-        old_res = ldlm_resource_find(ns, name, hash);
-        if (old_res) {
+        cfs_hash_bd_lock(ns->ns_rs_hash, &bd, 1);
+        hnode = (version == cfs_hash_bd_version_get(&bd)) ?  NULL :
+                cfs_hash_bd_lookup_locked(ns->ns_rs_hash, &bd, (void *)name);
+
+        if (hnode != NULL) {
                 /* someone won the race and added the resource before */
                 /* someone won the race and added the resource before */
-                ldlm_resource_getref(old_res);
-                cfs_spin_unlock(&ns->ns_hash_lock);
+                cfs_hash_bd_unlock(ns->ns_rs_hash, &bd, 1);
                 /* clean lu_ref for failed resource */
                 lu_ref_fini(&res->lr_reference);
                 OBD_SLAB_FREE(res, ldlm_resource_slab, sizeof *res);
                 /* clean lu_ref for failed resource */
                 lu_ref_fini(&res->lr_reference);
                 OBD_SLAB_FREE(res, ldlm_resource_slab, sizeof *res);
+
+                res = cfs_hlist_entry(hnode, struct ldlm_resource, lr_hash);
                 /* synchronize WRT resource creation */
                 if (ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
                 /* synchronize WRT resource creation */
                 if (ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
-                        cfs_down(&old_res->lr_lvb_sem);
-                        cfs_up(&old_res->lr_lvb_sem);
+                        cfs_down(&res->lr_lvb_sem);
+                        cfs_up(&res->lr_lvb_sem);
                 }
                 }
-                RETURN(old_res);
+                return res;
         }
         }
-
         /* we won! let's add the resource */
         /* we won! let's add the resource */
-        bucket = ns->ns_hash + hash;
-        cfs_list_add(&res->lr_hash, bucket);
-        ns->ns_resources++;
-        ldlm_namespace_get_locked(ns);
-
-        LASSERT(parent == NULL); /* legacy... */
-        cfs_list_add(&res->lr_childof, &ns->ns_root_list);
-        cfs_spin_unlock(&ns->ns_hash_lock);
+        cfs_hash_bd_add_locked(ns->ns_rs_hash, &bd, &res->lr_hash);
+        if (cfs_hash_bd_count_get(&bd) == 1)
+                ldlm_namespace_get(ns);
 
 
+        cfs_hash_bd_unlock(ns->ns_rs_hash, &bd, 1);
         if (ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
                 int rc;
 
         if (ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
                 int rc;
 
@@ -901,43 +1084,7 @@ ldlm_resource_add(struct ldlm_namespace *ns, struct ldlm_resource *parent,
                 cfs_up(&res->lr_lvb_sem);
         }
 
                 cfs_up(&res->lr_lvb_sem);
         }
 
-        RETURN(res);
-}
-
-/* Args: unlocked namespace
- * Locks: takes and releases ns->ns_lock and res->lr_lock
- * Returns: referenced, unlocked ldlm_resource or NULL */
-struct ldlm_resource *
-ldlm_resource_get(struct ldlm_namespace *ns, struct ldlm_resource *parent,
-                  const struct ldlm_res_id *name, ldlm_type_t type, int create)
-{
-        __u32 hash = ldlm_hash_fn(parent, name);
-        struct ldlm_resource *res = NULL;
-        ENTRY;
-
-        LASSERT(ns != NULL);
-        LASSERT(ns->ns_hash != NULL);
-        LASSERT(name->name[0] != 0);
-
-        cfs_spin_lock(&ns->ns_hash_lock);
-        res = ldlm_resource_find(ns, name, hash);
-        if (res) {
-                ldlm_resource_getref(res);
-                cfs_spin_unlock(&ns->ns_hash_lock);
-                /* synchronize WRT resource creation */
-                if (ns->ns_lvbo && ns->ns_lvbo->lvbo_init) {
-                        cfs_down(&res->lr_lvb_sem);
-                        cfs_up(&res->lr_lvb_sem);
-                }
-                RETURN(res);
-        }
-        cfs_spin_unlock(&ns->ns_hash_lock);
-
-        if (create == 0)
-                RETURN(NULL);
-
-        res = ldlm_resource_add(ns, parent, name, hash, type);
-        RETURN(res);
+        return res;
 }
 
 struct ldlm_resource *ldlm_resource_getref(struct ldlm_resource *res)
 }
 
 struct ldlm_resource *ldlm_resource_getref(struct ldlm_resource *res)
@@ -950,11 +1097,10 @@ struct ldlm_resource *ldlm_resource_getref(struct ldlm_resource *res)
         return res;
 }
 
         return res;
 }
 
-void __ldlm_resource_putref_final(struct ldlm_resource *res)
+static void __ldlm_resource_putref_final(cfs_hash_bd_t *bd,
+                                         struct ldlm_resource *res)
 {
 {
-        struct ldlm_namespace *ns = ldlm_res_to_ns(res);
-
-        LASSERT_SPIN_LOCKED(&ns->ns_hash_lock);
+        struct ldlm_ns_bucket *nsb = res->lr_ns_bucket;
 
         if (!cfs_list_empty(&res->lr_granted)) {
                 ldlm_resource_dump(D_ERROR, res);
 
         if (!cfs_list_empty(&res->lr_granted)) {
                 ldlm_resource_dump(D_ERROR, res);
@@ -971,57 +1117,62 @@ void __ldlm_resource_putref_final(struct ldlm_resource *res)
                 LBUG();
         }
 
                 LBUG();
         }
 
-        /* Pass 0 here to not wake ->ns_waitq up yet, we will do it few
-         * lines below when all children are freed. */
-        ldlm_namespace_put_locked(ns, 0);
-        cfs_list_del_init(&res->lr_hash);
-        cfs_list_del_init(&res->lr_childof);
+        cfs_hash_bd_del_locked(nsb->nsb_namespace->ns_rs_hash,
+                               bd, &res->lr_hash);
         lu_ref_fini(&res->lr_reference);
         lu_ref_fini(&res->lr_reference);
-
-        ns->ns_resources--;
-        if (ns->ns_resources == 0)
-                cfs_waitq_signal(&ns->ns_waitq);
-}
-
-int ldlm_resource_putref_internal(struct ldlm_resource *res, int locked)
-{
-        struct ldlm_namespace *ns = ldlm_res_to_ns(res);
-        ENTRY;
-
-        CDEBUG(D_INFO, "putref res: %p count: %d\n", res,
-               cfs_atomic_read(&res->lr_refcount) - 1);
-        LASSERTF(cfs_atomic_read(&res->lr_refcount) > 0, "%d",
-                 cfs_atomic_read(&res->lr_refcount));
-        LASSERTF(cfs_atomic_read(&res->lr_refcount) < LI_POISON, "%d",
-                 cfs_atomic_read(&res->lr_refcount));
-
-        if (locked && !cfs_atomic_dec_and_test(&res->lr_refcount))
-                RETURN(0);
-        if (!locked && !cfs_atomic_dec_and_lock(&res->lr_refcount,
-                                            &ns->ns_hash_lock))
-                RETURN(0);
-
-        __ldlm_resource_putref_final(res);
-
-        if (!locked)
-                cfs_spin_unlock(&ns->ns_hash_lock);
-
-        if (ns->ns_lvbo && ns->ns_lvbo->lvbo_free)
-                ns->ns_lvbo->lvbo_free(res);
-
-        OBD_SLAB_FREE(res, ldlm_resource_slab, sizeof *res);
-
-        RETURN(1);
+        if (cfs_hash_bd_count_get(bd) == 0)
+                ldlm_namespace_put(nsb->nsb_namespace);
 }
 
 }
 
+/* Returns 1 if the resource was freed, 0 if it remains. */
 int ldlm_resource_putref(struct ldlm_resource *res)
 {
 int ldlm_resource_putref(struct ldlm_resource *res)
 {
-        return ldlm_resource_putref_internal(res, 0);
+        struct ldlm_namespace *ns = ldlm_res_to_ns(res);
+        int ref = cfs_atomic_read(&res->lr_refcount);
+        cfs_hash_bd_t   bd;
+
+        CDEBUG(D_INFO, "putref res: %p count: %d\n", res, ref - 1);
+        LASSERTF(ref > 0 && ref < LI_POISON, "%d", ref);
+        cfs_hash_bd_get(ns->ns_rs_hash, &res->lr_name, &bd);
+        if (cfs_hash_bd_dec_and_lock(ns->ns_rs_hash, &bd, &res->lr_refcount)) {
+                __ldlm_resource_putref_final(&bd, res);
+                cfs_hash_bd_unlock(ns->ns_rs_hash, &bd, 1);
+                if (ns->ns_lvbo && ns->ns_lvbo->lvbo_free)
+                        ns->ns_lvbo->lvbo_free(res);
+                OBD_SLAB_FREE(res, ldlm_resource_slab, sizeof *res);
+                return 1;
+        }
+        return 0;
 }
 
 }
 
+/* Returns 1 if the resource was freed, 0 if it remains. */
 int ldlm_resource_putref_locked(struct ldlm_resource *res)
 {
 int ldlm_resource_putref_locked(struct ldlm_resource *res)
 {
-        return ldlm_resource_putref_internal(res, 1);
+        struct ldlm_namespace *ns = ldlm_res_to_ns(res);
+        int ref = cfs_atomic_read(&res->lr_refcount);
+
+        CDEBUG(D_INFO, "putref res: %p count: %d\n", res, ref - 1);
+        LASSERTF(ref > 0 && ref < LI_POISON, "%d", ref);
+        if (cfs_atomic_dec_and_test(&res->lr_refcount)) {
+                cfs_hash_bd_t bd;
+
+                cfs_hash_bd_get(ldlm_res_to_ns(res)->ns_rs_hash,
+                                &res->lr_name, &bd);
+                __ldlm_resource_putref_final(&bd, res);
+                cfs_hash_bd_unlock(ns->ns_rs_hash, &bd, 1);
+                /* NB: ns_rs_hash is created with CFS_HASH_NO_ITEMREF,
+                 * so we should never be here while calling cfs_hash_del,
+                 * cfs_hash_for_each_nolock is the only case we can get
+                 * here, which is safe to release cfs_hash_bd_lock.
+                 */
+                if (ns->ns_lvbo && ns->ns_lvbo->lvbo_free)
+                        ns->ns_lvbo->lvbo_free(res);
+                OBD_SLAB_FREE(res, ldlm_resource_slab, sizeof *res);
+
+                cfs_hash_bd_lock(ns->ns_rs_hash, &bd, 1);
+                return 1;
+        }
+        return 0;
 }
 
 void ldlm_resource_add_lock(struct ldlm_resource *res, cfs_list_t *head,
 }
 
 void ldlm_resource_add_lock(struct ldlm_resource *res, cfs_list_t *head,
@@ -1101,41 +1252,37 @@ void ldlm_dump_all_namespaces(ldlm_side_t client, int level)
         cfs_mutex_up(ldlm_namespace_lock(client));
 }
 
         cfs_mutex_up(ldlm_namespace_lock(client));
 }
 
-void ldlm_namespace_dump(int level, struct ldlm_namespace *ns)
+static int ldlm_res_hash_dump(cfs_hash_t *hs, cfs_hash_bd_t *bd,
+                              cfs_hlist_node_t *hnode, void *arg)
 {
 {
-        cfs_list_t *tmp;
+        struct ldlm_resource *res = cfs_hash_object(hs, hnode);
+        int    level = (int)(unsigned long)arg;
+
+        lock_res(res);
+        ldlm_resource_dump(level, res);
+        unlock_res(res);
 
 
+        return 0;
+}
+
+void ldlm_namespace_dump(int level, struct ldlm_namespace *ns)
+{
         if (!((libcfs_debug | D_ERROR) & level))
                 return;
 
         CDEBUG(level, "--- Namespace: %s (rc: %d, side: %s)\n",
         if (!((libcfs_debug | D_ERROR) & level))
                 return;
 
         CDEBUG(level, "--- Namespace: %s (rc: %d, side: %s)\n",
-               ns->ns_name, ns->ns_refcount,
+               ldlm_ns_name(ns), cfs_atomic_read(&ns->ns_bref),
                ns_is_client(ns) ? "client" : "server");
 
         if (cfs_time_before(cfs_time_current(), ns->ns_next_dump))
                 return;
 
                ns_is_client(ns) ? "client" : "server");
 
         if (cfs_time_before(cfs_time_current(), ns->ns_next_dump))
                 return;
 
-        cfs_spin_lock(&ns->ns_hash_lock);
-        tmp = ns->ns_root_list.next;
-        while (tmp != &ns->ns_root_list) {
-                struct ldlm_resource *res;
-                res = cfs_list_entry(tmp, struct ldlm_resource, lr_childof);
-
-                ldlm_resource_getref(res);
-                cfs_spin_unlock(&ns->ns_hash_lock);
-                LDLM_RESOURCE_ADDREF(res);
-
-                lock_res(res);
-                ldlm_resource_dump(level, res);
-                unlock_res(res);
-
-                LDLM_RESOURCE_DELREF(res);
-                cfs_spin_lock(&ns->ns_hash_lock);
-                tmp = tmp->next;
-                ldlm_resource_putref_locked(res);
-        }
+        cfs_hash_for_each_nolock(ns->ns_rs_hash,
+                                 ldlm_res_hash_dump,
+                                 (void *)(unsigned long)level);
+        cfs_spin_lock(&ns->ns_lock);
         ns->ns_next_dump = cfs_time_shift(10);
         ns->ns_next_dump = cfs_time_shift(10);
-        cfs_spin_unlock(&ns->ns_hash_lock);
+        cfs_spin_unlock(&ns->ns_lock);
 }
 
 void ldlm_resource_dump(int level, struct ldlm_resource *res)
 }
 
 void ldlm_resource_dump(int level, struct ldlm_resource *res)
index 06faa97..da97a53 100644 (file)
@@ -4610,7 +4610,8 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
                  LUSTRE_MDT_NAME"-%p", m);
         m->mdt_namespace = ldlm_namespace_new(obd, info->mti_u.ns_name,
                                               LDLM_NAMESPACE_SERVER,
                  LUSTRE_MDT_NAME"-%p", m);
         m->mdt_namespace = ldlm_namespace_new(obd, info->mti_u.ns_name,
                                               LDLM_NAMESPACE_SERVER,
-                                              LDLM_NAMESPACE_GREEDY);
+                                              LDLM_NAMESPACE_GREEDY,
+                                              LDLM_NS_TYPE_MDT);
         if (m->mdt_namespace == NULL)
                 GOTO(err_fini_seq, rc = -ENOMEM);
 
         if (m->mdt_namespace == NULL)
                 GOTO(err_fini_seq, rc = -ENOMEM);
 
index b3dfef2..770cab8 100644 (file)
@@ -197,8 +197,10 @@ static int mgs_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
         }
 
         /* namespace for mgs llog */
         }
 
         /* namespace for mgs llog */
-        obd->obd_namespace = ldlm_namespace_new(obd ,"MGS", LDLM_NAMESPACE_SERVER,
-                                                LDLM_NAMESPACE_MODEST);
+        obd->obd_namespace = ldlm_namespace_new(obd ,"MGS",
+                                                LDLM_NAMESPACE_SERVER,
+                                                LDLM_NAMESPACE_MODEST,
+                                                LDLM_NS_TYPE_MGT);
         if (obd->obd_namespace == NULL)
                 GOTO(err_ops, rc = -ENOMEM);
 
         if (obd->obd_namespace == NULL)
                 GOTO(err_ops, rc = -ENOMEM);
 
index f17e36f..6e8a459 100644 (file)
@@ -560,7 +560,8 @@ static int echo_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
         sprintf(ns_name, "echotgt-%s", obd->obd_uuid.uuid);
         obd->obd_namespace = ldlm_namespace_new(obd, ns_name,
                                                 LDLM_NAMESPACE_SERVER,
         sprintf(ns_name, "echotgt-%s", obd->obd_uuid.uuid);
         obd->obd_namespace = ldlm_namespace_new(obd, ns_name,
                                                 LDLM_NAMESPACE_SERVER,
-                                                LDLM_NAMESPACE_MODEST);
+                                                LDLM_NAMESPACE_MODEST,
+                                                LDLM_NS_TYPE_OST);
         if (obd->obd_namespace == NULL) {
                 LBUG();
                 RETURN(-ENOMEM);
         if (obd->obd_namespace == NULL) {
                 LBUG();
                 RETURN(-ENOMEM);
index c6b330f..0057fe4 100644 (file)
@@ -1724,7 +1724,7 @@ static int filter_intent_policy(struct ldlm_namespace *ns,
         if (rc == LDLM_ITER_CONTINUE) {
                 /* do not grant locks to the liblustre clients: they cannot
                  * handle ASTs robustly.  We need to do this while still
         if (rc == LDLM_ITER_CONTINUE) {
                 /* do not grant locks to the liblustre clients: they cannot
                  * handle ASTs robustly.  We need to do this while still
-                 * holding ns_lock to avoid the lock remaining on the res_link
+                 * holding lr_lock to avoid the lock remaining on the res_link
                  * list (and potentially being added to l_pending_list by an
                  * AST) when we are going to drop this lock ASAP. */
                 if (lock->l_export->exp_libclient ||
                  * list (and potentially being added to l_pending_list by an
                  * AST) when we are going to drop this lock ASAP. */
                 if (lock->l_export->exp_libclient ||
@@ -1747,7 +1747,7 @@ static int filter_intent_policy(struct ldlm_namespace *ns,
         *reply_lvb = *res_lvb;
 
         /*
         *reply_lvb = *res_lvb;
 
         /*
-         * ->ns_lock guarantees that no new locks are granted, and,
+         * lr_lock guarantees that no new locks are granted, and,
          * therefore, that res->lr_lvb_data cannot increase beyond the
          * end of already granted lock. As a result, it is safe to
          * check against "stale" reply_lvb->lvb_size value without
          * therefore, that res->lr_lvb_data cannot increase beyond the
          * end of already granted lock. As a result, it is safe to
          * check against "stale" reply_lvb->lvb_size value without
@@ -2059,8 +2059,10 @@ int filter_common_setup(struct obd_device *obd, struct lustre_cfg* lcfg,
                 GOTO(err_post, rc = -ENOMEM);
 
         sprintf(ns_name, "filter-%s", obd->obd_uuid.uuid);
                 GOTO(err_post, rc = -ENOMEM);
 
         sprintf(ns_name, "filter-%s", obd->obd_uuid.uuid);
-        obd->obd_namespace = ldlm_namespace_new(obd, ns_name, LDLM_NAMESPACE_SERVER,
-                                                LDLM_NAMESPACE_GREEDY);
+        obd->obd_namespace = ldlm_namespace_new(obd, ns_name,
+                                                LDLM_NAMESPACE_SERVER,
+                                                LDLM_NAMESPACE_GREEDY,
+                                                LDLM_NS_TYPE_OST);
         if (obd->obd_namespace == NULL)
                 GOTO(err_post, rc = -ENOMEM);
         obd->obd_namespace->ns_lvbp = obd;
         if (obd->obd_namespace == NULL)
                 GOTO(err_post, rc = -ENOMEM);
         obd->obd_namespace->ns_lvbp = obd;