- Back out the ldlm_everything_lock patch, which causes a deadlock on cleanup

- Move backslashes in lustre_lite.h macros to fit in 80 columns
- Added a couple of structure fields that support my LOV work

diff --git a/lustre/include/linux/lustre_dlm.h b/lustre/include/linux/lustre_dlm.h
index 4324ee2..c07d6ff 100644 (file)
--- a/lustre/include/linux/lustre_dlm.h
+++ b/lustre/include/linux/lustre_dlm.h
@@ -93,6 +93,7 @@ struct ldlm_namespace {
         struct list_head      *ns_hash; /* hash table for ns */
         __u32                  ns_refcount; /* count of resources in the hash */
         struct list_head       ns_root_list; /* all root resources in ns */
+        struct lustre_lock     ns_lock; /* protects hash, refcount, list */
         struct list_head       ns_list_chain; /* position in global NS list */
         struct proc_dir_entry *ns_proc_dir;
 
@@ -265,9 +266,6 @@ do {                                                                          \
 #define LDLM_DEBUG_NOLOCK(format, a...)                 \
         CDEBUG(D_DLMTRACE, "### " format "\n" , ## a)
 
-/* For internal LDLM use */
-extern struct lustre_lock ldlm_everything_lock;
-
 /* ldlm_extent.c */
 int ldlm_extent_compat(struct ldlm_lock *, struct ldlm_lock *);
 int ldlm_extent_policy(struct ldlm_lock *, void *, ldlm_mode_t, void *);

diff --git a/lustre/include/linux/lustre_idl.h b/lustre/include/linux/lustre_idl.h
index 94c991b..caeffef 100644 (file)
--- a/lustre/include/linux/lustre_idl.h
+++ b/lustre/include/linux/lustre_idl.h
@@ -427,6 +427,7 @@ struct mds_rec_rename {
 
 struct lov_desc {
         __u32 ld_tgt_count;                /* how many OBD's */
+        __u32 ld_active_tgt_count;         /* how many active */
         __u32 ld_default_stripe_count;     /* how many objects are used */
         __u64 ld_default_stripe_size;      /* in bytes */
         __u64 ld_default_stripe_offset;    /* in bytes */

diff --git a/lustre/include/linux/lustre_lite.h b/lustre/include/linux/lustre_lite.h
index 83f3e27..b0b7917 100644 (file)
--- a/lustre/include/linux/lustre_lite.h
+++ b/lustre/include/linux/lustre_lite.h
@@ -58,7 +58,7 @@ struct ll_inode_info {
 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
 static inline struct ll_inode_info *LL_I(struct inode *inode)
 {
-       return container_of(inode, struct ll_inode_info, lli_vfs_inode);
+        return container_of(inode, struct ll_inode_info, lli_vfs_inode);
 }
 #endif
 
@@ -170,41 +170,36 @@ int ll_set_dd(struct dentry *de);
 /****
 
 I originally implmented these as functions, then realized a macro
-would be more helpful for debugging, so the CDEBUG messages show 
+would be more helpful for debugging, so the CDEBUG messages show
 the current calling function.  The orignal functions are in llite/dcache.c
 
 int ll_save_intent(struct dentry * de, struct lookup_intent * it);
 struct lookup_intent * ll_get_intent(struct dentry * de);
 ****/
 
-#define LL_SAVE_INTENT(de, it)                                                  \
-do {                                                                            \
-        LASSERT(ll_d2d(de) != NULL);                                            \
-                                                                                \
-        down(&ll_d2d(de)->lld_it_sem);                                          \
-        de->d_it = it;                                                          \
-        CDEBUG(D_DENTRY, "D_IT DOWN dentry %p fsdata %p intent: %s sem %d\n",   \
-               de, ll_d2d(de), ldlm_it2str(de->d_it->it_op),                    \
-               atomic_read(&(ll_d2d(de)->lld_it_sem.count)));                   \
-                                                                                \
+#define LL_SAVE_INTENT(de, it)                                                 \
+do {                                                                           \
+        LASSERT(ll_d2d(de) != NULL);                                           \
+                                                                               \
+        down(&ll_d2d(de)->lld_it_sem);                                         \
+        de->d_it = it;                                                         \
+        CDEBUG(D_DENTRY, "D_IT DOWN dentry %p fsdata %p intent: %s sem %d\n",  \
+               de, ll_d2d(de), ldlm_it2str(de->d_it->it_op),                   \
+               atomic_read(&(ll_d2d(de)->lld_it_sem.count)));                  \
 } while(0)
 
-#define LL_GET_INTENT(de, it)                                                           \
-do {                                                                                    \
-        it = de->d_it;                                                                  \
-                                                                                        \
-        LASSERT(ll_d2d(de) != NULL);                                                    \
-                                                                                        \
-        CDEBUG(D_DENTRY, "D_IT UP dentry %p fsdata %p intent: %s\n", de, ll_d2d(de),    \
-               ldlm_it2str(de->d_it->it_op));                                           \
-        de->d_it = NULL;                                                                \
-        up(&ll_d2d(de)->lld_it_sem);                                                    \
-                                                                                        \
-                                                                                        \
+#define LL_GET_INTENT(de, it)                                                  \
+do {                                                                           \
+        it = de->d_it;                                                         \
+                                                                               \
+        LASSERT(ll_d2d(de) != NULL);                                           \
+                                                                               \
+        CDEBUG(D_DENTRY, "D_IT UP dentry %p fsdata %p intent: %s\n",           \
+               de, ll_d2d(de), ldlm_it2str(de->d_it->it_op));                  \
+        de->d_it = NULL;                                                       \
+        up(&ll_d2d(de)->lld_it_sem);                                           \
 } while(0)
 
-
-
 /* dir.c */
 extern struct file_operations ll_dir_operations;
 extern struct inode_operations ll_dir_inode_operations;

diff --git a/lustre/include/linux/obd.h b/lustre/include/linux/obd.h
index c253ca6..fa95441 100644 (file)
--- a/lustre/include/linux/obd.h
+++ b/lustre/include/linux/obd.h
@@ -186,9 +186,11 @@ struct ost_obd {
 struct lov_tgt_desc {
         obd_uuid_t uuid;
         struct lustre_handle conn;
+        int active; /* is this target available for requests, etc */
 };
 
 struct lov_obd {
+        spinlock_t lov_lock;
         struct obd_device *mdcobd;
         struct lov_desc desc;
         int bufsize;

diff --git a/lustre/ldlm/ldlm_extent.c b/lustre/ldlm/ldlm_extent.c
index 16c24fd..1f952cd 100644 (file)
--- a/lustre/ldlm/ldlm_extent.c
+++ b/lustre/ldlm/ldlm_extent.c
@@ -79,11 +79,11 @@ int ldlm_extent_policy(struct ldlm_lock *lock, void *req_cookie,
         if (!res)
                 LBUG();
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&res->lr_namespace->ns_lock);
         policy_internal(&res->lr_granted, req_ex, &new_ex, mode);
         policy_internal(&res->lr_converting, req_ex, &new_ex, mode);
         policy_internal(&res->lr_waiting, req_ex, &new_ex, mode);
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&res->lr_namespace->ns_lock);
 
         memcpy(&lock->l_extent, &new_ex, sizeof(new_ex));
 

diff --git a/lustre/ldlm/ldlm_lock.c b/lustre/ldlm/ldlm_lock.c
index 97a8770..a361cab 100644 (file)
--- a/lustre/ldlm/ldlm_lock.c
+++ b/lustre/ldlm/ldlm_lock.c
@@ -30,8 +30,7 @@
 #include <linux/lustre_mds.h>
 #include <linux/obd_class.h>
 
-/* this lock protects ldlm_handle2lock's integrity */
-struct lustre_lock ldlm_everything_lock;
+//struct lustre_lock ldlm_everything_lock;
 
 /* lock types */
 char *ldlm_lockname[] = {
@@ -132,10 +131,10 @@ void ldlm_unregister_intent(void)
  */
 struct ldlm_lock *ldlm_lock_get(struct ldlm_lock *lock)
 {
-        l_lock(&ldlm_everything_lock);
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
         lock->l_refc++;
         ldlm_resource_getref(lock->l_resource);
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
         return lock;
 }
 
@@ -144,7 +143,7 @@ void ldlm_lock_put(struct ldlm_lock *lock)
         struct ldlm_namespace *ns = lock->l_resource->lr_namespace;
         ENTRY;
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&ns->ns_lock);
         lock->l_refc--;
         //LDLM_DEBUG(lock, "after refc--");
         if (lock->l_refc < 0)
@@ -158,21 +157,24 @@ void ldlm_lock_put(struct ldlm_lock *lock)
                 LDLM_LOCK_PUT(lock->l_parent);
 
         if (lock->l_refc == 0 && (lock->l_flags & LDLM_FL_DESTROYED)) {
+                l_unlock(&ns->ns_lock);
                 LDLM_DEBUG(lock, "final lock_put on destroyed lock, freeing");
 
+                //spin_lock(&ldlm_handle_lock);
                 spin_lock(&ns->ns_counter_lock);
                 ns->ns_locks--;
                 spin_unlock(&ns->ns_counter_lock);
 
+                lock->l_resource = NULL;
                 lock->l_random = DEAD_HANDLE_MAGIC;
                 if (lock->l_export && lock->l_export->exp_connection)
                         ptlrpc_put_connection(lock->l_export->exp_connection);
                 kmem_cache_free(ldlm_lock_slab, lock);
-                l_unlock(&ldlm_everything_lock);
+                //spin_unlock(&ldlm_handle_lock);
                 CDEBUG(D_MALLOC, "kfreed 'lock': %d at %p (tot 0).\n",
                        sizeof(*lock), lock);
         } else
-                l_unlock(&ldlm_everything_lock);
+                l_unlock(&ns->ns_lock);
 
         EXIT;
 }
@@ -180,7 +182,7 @@ void ldlm_lock_put(struct ldlm_lock *lock)
 void ldlm_lock_destroy(struct ldlm_lock *lock)
 {
         ENTRY;
-        l_lock(&ldlm_everything_lock);
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
 
         if (!list_empty(&lock->l_children)) {
                 LDLM_DEBUG(lock, "still has children (%p)!",
@@ -199,7 +201,7 @@ void ldlm_lock_destroy(struct ldlm_lock *lock)
         }
 
         if (lock->l_flags & LDLM_FL_DESTROYED) {
-                l_unlock(&ldlm_everything_lock);
+                l_unlock(&lock->l_resource->lr_namespace->ns_lock);
                 EXIT;
                 return;
         }
@@ -214,7 +216,7 @@ void ldlm_lock_destroy(struct ldlm_lock *lock)
         if (lock->l_export && lock->l_completion_ast)
                 lock->l_completion_ast(lock, 0);
 
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
         LDLM_LOCK_PUT(lock);
         EXIT;
 }
@@ -256,10 +258,10 @@ static struct ldlm_lock *ldlm_lock_new(struct ldlm_lock *parent,
         spin_unlock(&resource->lr_namespace->ns_counter_lock);
 
         if (parent != NULL) {
-                l_lock(&ldlm_everything_lock);
+                l_lock(&parent->l_resource->lr_namespace->ns_lock);
                 lock->l_parent = parent;
                 list_add(&lock->l_childof, &parent->l_children);
-                l_unlock(&ldlm_everything_lock);
+                l_unlock(&parent->l_resource->lr_namespace->ns_lock);
         }
 
         CDEBUG(D_MALLOC, "kmalloced 'lock': %d at "
@@ -276,11 +278,11 @@ int ldlm_lock_change_resource(struct ldlm_lock *lock, __u64 new_resid[3])
         int type, i;
         ENTRY;
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&ns->ns_lock);
         if (memcmp(new_resid, lock->l_resource->lr_name,
                    sizeof(lock->l_resource->lr_name)) == 0) {
                 /* Nothing to do */
-                l_unlock(&ldlm_everything_lock);
+                l_unlock(&ns->ns_lock);
                 RETURN(0);
         }
 
@@ -304,7 +306,7 @@ int ldlm_lock_change_resource(struct ldlm_lock *lock, __u64 new_resid[3])
         /* compensate for the initial get above.. */
         ldlm_resource_put(lock->l_resource);
 
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&ns->ns_lock);
         RETURN(0);
 }
 
@@ -327,29 +329,30 @@ struct ldlm_lock *__ldlm_handle2lock(struct lustre_handle *handle,
         if (!handle || !handle->addr)
                 RETURN(NULL);
 
-        l_lock(&ldlm_everything_lock);
+        //spin_lock(&ldlm_handle_lock);
         lock = (struct ldlm_lock *)(unsigned long)(handle->addr);
         if (!kmem_cache_validate(ldlm_lock_slab, (void *)lock)) {
                 CERROR("bogus lock %p\n", lock);
-                GOTO(out, retval);
+                GOTO(out2, retval);
         }
 
         if (lock->l_random != handle->cookie) {
                 CERROR("bogus cookie: lock %p has "LPX64" vs. handle "LPX64"\n",
                        lock, lock->l_random, handle->cookie);
-                GOTO(out, NULL);
+                GOTO(out2, NULL);
         }
         if (!lock->l_resource) {
                 CERROR("trying to lock bogus resource: lock %p\n", lock);
                 LDLM_DEBUG(lock, "ldlm_handle2lock(%p)", lock);
-                GOTO(out, retval);
+                GOTO(out2, retval);
         }
         if (!lock->l_resource->lr_namespace) {
                 CERROR("trying to lock bogus namespace: lock %p\n", lock);
                 LDLM_DEBUG(lock, "ldlm_handle2lock(%p)", lock);
-                GOTO(out, retval);
+                GOTO(out2, retval);
         }
 
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
         if (strict && lock->l_flags & LDLM_FL_DESTROYED) {
                 CERROR("lock already destroyed: lock %p\n", lock);
                 LDLM_DEBUG(lock, "ldlm_handle2lock(%p)", lock);
@@ -361,7 +364,9 @@ struct ldlm_lock *__ldlm_handle2lock(struct lustre_handle *handle,
                 CERROR("lock disappeared below us!!! %p\n", lock);
         EXIT;
  out:
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
+ out2:
+        //spin_unlock(&ldlm_handle_lock);
         return retval;
 }
 
@@ -385,7 +390,7 @@ static void ldlm_add_ast_work_item(struct ldlm_lock *lock,
         struct ldlm_ast_work *w;
         ENTRY;
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
         if (new && (lock->l_flags & LDLM_FL_AST_SENT))
                 GOTO(out, 0);
 
@@ -404,7 +409,7 @@ static void ldlm_add_ast_work_item(struct ldlm_lock *lock,
         w->w_lock = LDLM_LOCK_GET(lock);
         list_add(&w->w_list, lock->l_resource->lr_tmp);
       out:
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
         return;
 }
 
@@ -420,12 +425,12 @@ void ldlm_lock_addref(struct lustre_handle *lockh, __u32 mode)
 /* only called for local locks */
 void ldlm_lock_addref_internal(struct ldlm_lock *lock, __u32 mode)
 {
-        l_lock(&ldlm_everything_lock);
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
         if (mode == LCK_NL || mode == LCK_CR || mode == LCK_PR)
                 lock->l_readers++;
         else
                 lock->l_writers++;
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
         LDLM_LOCK_GET(lock);
         LDLM_DEBUG(lock, "ldlm_lock_addref(%s)", ldlm_lockname[mode]);
 }
@@ -440,7 +445,7 @@ void ldlm_lock_decref(struct lustre_handle *lockh, __u32 mode)
                 LBUG();
 
         LDLM_DEBUG(lock, "ldlm_lock_decref(%s)", ldlm_lockname[mode]);
-        l_lock(&ldlm_everything_lock);
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
         if (mode == LCK_NL || mode == LCK_CR || mode == LCK_PR)
                 lock->l_readers--;
         else
@@ -456,13 +461,13 @@ void ldlm_lock_decref(struct lustre_handle *lockh, __u32 mode)
                                "warning\n");
 
                 LDLM_DEBUG(lock, "final decref done on cbpending lock");
-                l_unlock(&ldlm_everything_lock);
+                l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 
                 /* FIXME: need a real 'desc' here */
                 lock->l_blocking_ast(lock, NULL, lock->l_data,
                                      lock->l_data_len, LDLM_CB_BLOCKING);
         } else
-                l_unlock(&ldlm_everything_lock);
+                l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 
         LDLM_LOCK_PUT(lock);    /* matches the ldlm_lock_get in addref */
         LDLM_LOCK_PUT(lock);    /* matches the handle2lock above */
@@ -511,7 +516,7 @@ static int ldlm_lock_compat(struct ldlm_lock *lock, int send_cbs)
         int rc;
         ENTRY;
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
         rc = ldlm_lock_compat_list(lock, send_cbs,
                                    &lock->l_resource->lr_granted);
         /* FIXME: should we be sending ASTs to converting? */
@@ -519,7 +524,7 @@ static int ldlm_lock_compat(struct ldlm_lock *lock, int send_cbs)
                 rc = ldlm_lock_compat_list
                         (lock, send_cbs, &lock->l_resource->lr_converting);
 
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
         RETURN(rc);
 }
 
@@ -531,7 +536,7 @@ void ldlm_grant_lock(struct ldlm_lock *lock)
         struct ldlm_resource *res = lock->l_resource;
         ENTRY;
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
         ldlm_resource_add_lock(res, &res->lr_granted, lock);
         lock->l_granted_mode = lock->l_req_mode;
 
@@ -541,7 +546,7 @@ void ldlm_grant_lock(struct ldlm_lock *lock)
         if (lock->l_completion_ast) {
                 ldlm_add_ast_work_item(lock, NULL);
         }
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
         EXIT;
 }
 
@@ -593,7 +598,8 @@ int ldlm_lock_match(struct ldlm_namespace *ns, __u64 * res_id, __u32 type,
         if (res == NULL)
                 RETURN(0);
 
-        l_lock(&ldlm_everything_lock);
+        ns = res->lr_namespace;
+        l_lock(&ns->ns_lock);
 
         if ((lock = search_queue(&res->lr_granted, mode, cookie)))
                 GOTO(out, rc = 1);
@@ -605,7 +611,7 @@ int ldlm_lock_match(struct ldlm_namespace *ns, __u64 * res_id, __u32 type,
         EXIT;
       out:
         ldlm_resource_put(res);
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&ns->ns_lock);
 
         if (lock) {
                 ldlm_lock2handle(lock, lockh);
@@ -685,7 +691,7 @@ ldlm_error_t ldlm_lock_enqueue(struct ldlm_lock * lock,
         lock->l_cookie = cookie;
         lock->l_cookie_len = cookie_len;
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&res->lr_namespace->ns_lock);
         if (local && lock->l_req_mode == lock->l_granted_mode) {
                 /* The server returned a blocked lock, but it was granted before
                  * we got a chance to actually enqueue it.  We don't need to do
@@ -732,7 +738,7 @@ ldlm_error_t ldlm_lock_enqueue(struct ldlm_lock * lock,
         ldlm_grant_lock(lock);
         EXIT;
       out:
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&res->lr_namespace->ns_lock);
         /* Don't set 'completion_ast' until here so that if the lock is granted
          * immediately we don't do an unnecessary completion call. */
         lock->l_completion_ast = completion;
@@ -800,7 +806,7 @@ void ldlm_reprocess_all(struct ldlm_resource *res)
                 return;
         }
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&res->lr_namespace->ns_lock);
         res->lr_tmp = &rpc_list;
 
         ldlm_reprocess_queue(res, &res->lr_converting);
@@ -808,7 +814,7 @@ void ldlm_reprocess_all(struct ldlm_resource *res)
                 ldlm_reprocess_queue(res, &res->lr_waiting);
 
         res->lr_tmp = NULL;
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&res->lr_namespace->ns_lock);
 
         ldlm_run_ast_work(&rpc_list);
         EXIT;
@@ -816,13 +822,13 @@ void ldlm_reprocess_all(struct ldlm_resource *res)
 
 void ldlm_cancel_callback(struct ldlm_lock *lock)
 {
-        l_lock(&ldlm_everything_lock);
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
         if (!(lock->l_flags & LDLM_FL_CANCEL)) {
                 lock->l_flags |= LDLM_FL_CANCEL;
                 lock->l_blocking_ast(lock, NULL, lock->l_data,
                                      lock->l_data_len, LDLM_CB_CANCELING);
         }
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 }
 
 void ldlm_lock_cancel(struct ldlm_lock *lock)
@@ -834,7 +840,7 @@ void ldlm_lock_cancel(struct ldlm_lock *lock)
         res = lock->l_resource;
         ns = res->lr_namespace;
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&ns->ns_lock);
         if (lock->l_readers || lock->l_writers)
                 CDEBUG(D_INFO, "lock still has references (%d readers, %d "
                        "writers)\n", lock->l_readers, lock->l_writers);
@@ -844,7 +850,7 @@ void ldlm_lock_cancel(struct ldlm_lock *lock)
         ldlm_del_waiting_lock(lock);
         ldlm_resource_unlink_lock(lock);
         ldlm_lock_destroy(lock);
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&ns->ns_lock);
         EXIT;
 }
 
@@ -876,7 +882,7 @@ struct ldlm_resource *ldlm_lock_convert(struct ldlm_lock *lock, int new_mode,
         res = lock->l_resource;
         ns = res->lr_namespace;
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&ns->ns_lock);
 
         lock->l_req_mode = new_mode;
         ldlm_resource_unlink_lock(lock);
@@ -906,7 +912,7 @@ struct ldlm_resource *ldlm_lock_convert(struct ldlm_lock *lock, int new_mode,
                 *flags |= LDLM_FL_BLOCK_CONV;
         }
 
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&ns->ns_lock);
 
         if (granted)
                 ldlm_run_ast_work(&rpc_list);

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index 21f8801..a6abc5e 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -267,10 +267,10 @@ int ldlm_handle_enqueue(struct ptlrpc_request *req)
 
         lock->l_export = req->rq_export;
         if (lock->l_export) {
-                l_lock(&ldlm_everything_lock);
+                l_lock(&lock->l_resource->lr_namespace->ns_lock);
                 list_add(&lock->l_export_chain,
                          &lock->l_export->exp_ldlm_data.led_held_locks);
-                l_unlock(&ldlm_everything_lock);
+                l_unlock(&lock->l_resource->lr_namespace->ns_lock);
         }
 
         EXIT;
@@ -399,10 +399,10 @@ static int ldlm_handle_bl_callback(struct ptlrpc_request *req)
 
         LDLM_DEBUG(lock, "client blocking AST callback handler START");
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
         lock->l_flags |= LDLM_FL_CBPENDING;
         do_ast = (!lock->l_readers && !lock->l_writers);
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 
         if (do_ast) {
                 LDLM_DEBUG(lock, "already unused, calling "
@@ -441,7 +441,7 @@ static int ldlm_handle_cp_callback(struct ptlrpc_request *req)
 
         LDLM_DEBUG(lock, "client completion callback handler START");
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
 
         /* If we receive the completion AST before the actual enqueue returned,
          * then we might need to switch lock modes, resources, or extents. */
@@ -462,7 +462,7 @@ static int ldlm_handle_cp_callback(struct ptlrpc_request *req)
         lock->l_resource->lr_tmp = &ast_list;
         ldlm_grant_lock(lock);
         lock->l_resource->lr_tmp = NULL;
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
         LDLM_DEBUG(lock, "callback handler finished, about to run_ast_work");
         LDLM_LOCK_PUT(lock);
 
@@ -669,8 +669,6 @@ static int __init ldlm_init(void)
                 return -ENOMEM;
         }
 
-        l_lock_init(&ldlm_everything_lock);
-
         return 0;
 }
 
@@ -715,7 +713,6 @@ EXPORT_SYMBOL(ldlm_namespace_dump);
 EXPORT_SYMBOL(ldlm_cancel_locks_for_export);
 EXPORT_SYMBOL(l_lock);
 EXPORT_SYMBOL(l_unlock);
-EXPORT_SYMBOL(ldlm_everything_lock);
 
 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 MODULE_DESCRIPTION("Lustre Lock Management Module v0.1");

diff --git a/lustre/ldlm/ldlm_request.c b/lustre/ldlm/ldlm_request.c
index be8a581..90ca711 100644 (file)
--- a/lustre/ldlm/ldlm_request.c
+++ b/lustre/ldlm/ldlm_request.c
@@ -426,10 +426,10 @@ int ldlm_cli_cancel(struct lustre_handle *lockh)
         if (lock->l_connh) {
                 LDLM_DEBUG(lock, "client-side cancel");
                 /* Set this flag to prevent others from getting new references*/
-                l_lock(&ldlm_everything_lock);
+                l_lock(&lock->l_resource->lr_namespace->ns_lock);
                 lock->l_flags |= LDLM_FL_CBPENDING;
                 ldlm_cancel_callback(lock);
-                l_unlock(&ldlm_everything_lock);
+                l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 
                 req = ptlrpc_prep_req(class_conn2cliimp(lock->l_connh),
                                       LDLM_CANCEL, 1, &size, NULL);
@@ -485,7 +485,7 @@ int ldlm_cli_cancel_unused(struct ldlm_namespace *ns, __u64 *res_id,
                 RETURN(0);
         }
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&ns->ns_lock);
         list_for_each(tmp, &res->lr_granted) {
                 struct ldlm_lock *lock;
                 lock = list_entry(tmp, struct ldlm_lock, l_res_link);
@@ -506,7 +506,7 @@ int ldlm_cli_cancel_unused(struct ldlm_namespace *ns, __u64 *res_id,
                 w->w_lock = LDLM_LOCK_GET(lock);
                 list_add(&w->w_list, &list);
         }
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&ns->ns_lock);
 
         list_for_each_safe(tmp, next, &list) {
                 struct lustre_handle lockh;

diff --git a/lustre/ldlm/ldlm_resource.c b/lustre/ldlm/ldlm_resource.c
index 8a7e03d..683f1c9 100644 (file)
--- a/lustre/ldlm/ldlm_resource.c
+++ b/lustre/ldlm/ldlm_resource.c
@@ -79,6 +79,7 @@ struct ldlm_namespace *ldlm_namespace_new(char *name, __u32 client)
         strcpy(ns->ns_name, name);
 
         INIT_LIST_HEAD(&ns->ns_root_list);
+        l_lock_init(&ns->ns_lock);
         ns->ns_refcount = 0;
         ns->ns_client = client;
         spin_lock_init(&ns->ns_counter_lock);
@@ -158,7 +159,7 @@ int ldlm_namespace_cleanup(struct ldlm_namespace *ns, int local_only)
 {
         int i;
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&ns->ns_lock);
         for (i = 0; i < RES_HASH_SIZE; i++) {
                 struct list_head *tmp, *pos;
                 list_for_each_safe(tmp, pos, &(ns->ns_hash[i])) {
@@ -185,7 +186,7 @@ int ldlm_namespace_cleanup(struct ldlm_namespace *ns, int local_only)
                         }
                 }
         }
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&ns->ns_lock);
 
         return ELDLM_OK;
 }
@@ -317,7 +318,7 @@ struct ldlm_resource *ldlm_resource_get(struct ldlm_namespace *ns,
                 RETURN(NULL);
         }
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&ns->ns_lock);
         bucket = ns->ns_hash + ldlm_hash_fn(parent, name);
 
         list_for_each(tmp, bucket) {
@@ -334,7 +335,7 @@ struct ldlm_resource *ldlm_resource_get(struct ldlm_namespace *ns,
 
         if (res == NULL && create)
                 res = ldlm_resource_add(ns, parent, name, type);
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&ns->ns_lock);
 
         RETURN(res);
 }
@@ -354,11 +355,11 @@ int ldlm_resource_put(struct ldlm_resource *res)
                 struct ldlm_namespace *ns = res->lr_namespace;
                 ENTRY;
 
-                l_lock(&ldlm_everything_lock);
+                l_lock(&ns->ns_lock);
 
                 if (atomic_read(&res->lr_refcount) != 0) {
                         /* We lost the race. */
-                        l_unlock(&ldlm_everything_lock);
+                        l_unlock(&ns->ns_lock);
                         goto out;
                 }
 
@@ -379,7 +380,7 @@ int ldlm_resource_put(struct ldlm_resource *res)
                 list_del(&res->lr_childof);
 
                 kmem_cache_free(ldlm_resource_slab, res);
-                l_unlock(&ldlm_everything_lock);
+                l_unlock(&ns->ns_lock);
 
                 spin_lock(&ns->ns_counter_lock);
                 ns->ns_resources--;
@@ -399,7 +400,7 @@ int ldlm_resource_put(struct ldlm_resource *res)
 void ldlm_resource_add_lock(struct ldlm_resource *res, struct list_head *head,
                             struct ldlm_lock *lock)
 {
-        l_lock(&ldlm_everything_lock);
+        l_lock(&res->lr_namespace->ns_lock);
 
         ldlm_resource_dump(res);
         ldlm_lock_dump(lock);
@@ -408,14 +409,14 @@ void ldlm_resource_add_lock(struct ldlm_resource *res, struct list_head *head,
                 LBUG();
 
         list_add(&lock->l_res_link, head);
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&res->lr_namespace->ns_lock);
 }
 
 void ldlm_resource_unlink_lock(struct ldlm_lock *lock)
 {
-        l_lock(&ldlm_everything_lock);
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
         list_del_init(&lock->l_res_link);
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 }
 
 void ldlm_res2desc(struct ldlm_resource *res, struct ldlm_resource_desc *desc)
@@ -444,7 +445,7 @@ void ldlm_namespace_dump(struct ldlm_namespace *ns)
 {
         struct list_head *tmp;
 
-        l_lock(&ldlm_everything_lock);
+        l_lock(&ns->ns_lock);
         CDEBUG(D_OTHER, "--- Namespace: %s (rc: %d, client: %d)\n", ns->ns_name,
                ns->ns_refcount, ns->ns_client);
 
@@ -456,7 +457,7 @@ void ldlm_namespace_dump(struct ldlm_namespace *ns)
                  * them recursively. */
                 ldlm_resource_dump(res);
         }
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&ns->ns_lock);
 }
 
 void ldlm_resource_dump(struct ldlm_resource *res)

diff --git a/lustre/mds/handler.c b/lustre/mds/handler.c
index c44c087..58d2882 100644 (file)
--- a/lustre/mds/handler.c
+++ b/lustre/mds/handler.c
@@ -443,12 +443,11 @@ int mds_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
                 RETURN(0);
         }
 
-        /* XXX layering violation!  When this goes, so can the
-         * EXPORT_SYMBOL(ldlm_everything_lock) in ldlm/ldlm_lockd.c -phil*/
-        l_lock(&ldlm_everything_lock);
+        /* XXX layering violation!  -phil */
+        l_lock(&lock->l_resource->lr_namespace->ns_lock);
         lock->l_flags |= LDLM_FL_CBPENDING;
         do_ast = (!lock->l_readers && !lock->l_writers);
-        l_unlock(&ldlm_everything_lock);
+        l_unlock(&lock->l_resource->lr_namespace->ns_lock);
 
         if (do_ast) {
                 struct lustre_handle lockh;