b=13220

r=alex,vitaly

Change the ldlm_lock's bitlock to a real spinlock to avoid it to be overwritten when operating on l_flags.

lustre/include/lustre_dlm.h

@@ -128,10 +128,6 @@ typedef enum {
  * w/o involving separate thread. in order to decrease cs rate */
 #define LDLM_FL_ATOMIC_CB      0x4000000
 
-/* while this flag is set, the lock can't change resource */
-#define LDLM_FL_LOCK_PROTECT   0x8000000
-#define LDLM_FL_LOCK_PROTECT_BIT  27
-
 /* It may happen that a client initiate 2 operations, e.g. unlink and mkdir,
  * such that server send blocking ast for conflict locks to this client for
  * the 1st operation, whereas the 2nd operation has canceled this lock and
@@ -273,6 +269,10 @@ struct ldlm_lock {
         struct portals_handle l_handle; // must be first in the structure
         atomic_t              l_refc;
 
+        /* internal spinlock protects l_resource.  we should hold this lock 
+         * first before grabbing res_lock.*/
+        spinlock_t            l_lock;
+
         /* ldlm_lock_change_resource() can change this */
         struct ldlm_resource *l_resource;
 
@@ -299,13 +299,11 @@ struct ldlm_lock {
         struct obd_export    *l_export;
         struct obd_export    *l_conn_export;
 
-        /* protected by lr_lock */
-        __u32                 l_flags;
-
         struct lustre_handle  l_remote_handle;
         ldlm_policy_data_t    l_policy_data;
 
         /* protected by lr_lock */
+        __u32                 l_flags;
         __u32                 l_readers;
         __u32                 l_writers;
         __u8                  l_destroyed;
@@ -332,7 +330,6 @@ struct ldlm_lock {
         cfs_time_t            l_callback_timeout; /* jiffies */
 
         __u32                 l_pid;            /* pid which created this lock */
-        __u32                 l_pidb;           /* who holds LOCK_PROTECT_BIT */
 
         /* for ldlm_add_ast_work_item() */
         struct list_head      l_bl_ast;
@@ -652,60 +649,7 @@ static inline void check_res_locked(struct ldlm_resource *res)
 {
         LASSERT_SPIN_LOCKED(&res->lr_lock);
 }
-#ifdef __KERNEL__
-# if LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,60)
-static inline void lock_bitlock(struct ldlm_lock *lock)
-{
-        bit_spin_lock(LDLM_FL_LOCK_PROTECT_BIT, (void *) &lock->l_flags);
-        LASSERT(lock->l_pidb == 0);
-        lock->l_pidb = current->pid;
-}
-
-static inline void unlock_bitlock(struct ldlm_lock *lock)
-{
-        LASSERT(lock->l_pidb == current->pid);
-        lock->l_pidb = 0;
-        bit_spin_unlock(LDLM_FL_LOCK_PROTECT_BIT, (void *) &lock->l_flags);
-}
-#else /* LINUX_VERSION_CODE < KERNEL_VERSION(2,5,60) */
-static inline void lock_bitlock(struct ldlm_lock *lock)
-{
-#if defined(CONFIG_SMP) || defined(CONFIG_DEBUG_SPINLOCK)
-        /* bit_spin_lock(LDLM_FL_LOCK_PROTECT_BIT, (void *)&lock->l_flags);*/
-        while (test_and_set_bit(LDLM_FL_LOCK_PROTECT_BIT, &lock->l_flags)) {
-                while (test_bit(LDLM_FL_LOCK_PROTECT_BIT, &lock->l_flags))
-                        cpu_relax();
-        }
-#endif
-
-        LASSERT(lock->l_pidb == 0);
-        lock->l_pidb = current->pid;
-}
-
-static inline void unlock_bitlock(struct ldlm_lock *lock)
-{
-        LASSERT(lock->l_pidb == current->pid);
-        lock->l_pidb = 0;
-
-#if defined(CONFIG_SMP) || defined(CONFIG_DEBUG_SPINLOCK)
-        /* bit_spin_unlock(LDLM_FL_LOCK_PROTECT_BIT, (void *)&lock->l_flags);*/
-        BUG_ON(!test_bit(LDLM_FL_LOCK_PROTECT_BIT, &lock->l_flags));
-        smp_mb__before_clear_bit();
-        clear_bit(LDLM_FL_LOCK_PROTECT_BIT, &lock->l_flags);
-#endif
-}
-#endif /* LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,60) */
-#else /* !__KERNEL__ */
-static inline void lock_bitlock(struct ldlm_lock *lock)
-{
-        lock->l_flags |= 1 << LDLM_FL_LOCK_PROTECT_BIT;
-}
 
-static inline void unlock_bitlock(struct ldlm_lock *lock)
-{
-        lock->l_flags &= ~(1 << LDLM_FL_LOCK_PROTECT_BIT);
-}
-#endif /* __KERNEL__ */
 struct ldlm_resource * lock_res_and_lock(struct ldlm_lock *lock);
 void unlock_res_and_lock(struct ldlm_lock *lock);
 

lustre/ldlm/l_lock.c

@@ -49,7 +49,7 @@ struct ldlm_resource * lock_res_and_lock(struct ldlm_lock *lock)
                 return res;
         } 
 
-        lock_bitlock(lock);
+        spin_lock(&lock->l_lock);
         res = lock->l_resource;
         lock_res(res);
         return res;
@@ -66,6 +66,6 @@ void unlock_res_and_lock(struct ldlm_lock *lock)
         }
 
         unlock_res(res);
-        unlock_bitlock(lock);
+        spin_unlock(&lock->l_lock);
 }
 

lustre/ldlm/ldlm_lock.c

@@ -292,6 +292,7 @@ static struct ldlm_lock *ldlm_lock_new(struct ldlm_resource *resource)
         if (lock == NULL)
                 RETURN(NULL);
 
+        spin_lock_init(&lock->l_lock);
         lock->l_resource = ldlm_resource_getref(resource);
 
         atomic_set(&lock->l_refc, 2);
@@ -303,7 +304,6 @@ static struct ldlm_lock *ldlm_lock_new(struct ldlm_resource *resource)
         CFS_INIT_LIST_HEAD(&lock->l_cp_ast);
         cfs_waitq_init(&lock->l_waitq);
         lock->l_blocking_lock = NULL;
-        lock->l_pidb = 0;
         lock->l_sl_mode.prev = NULL;
         lock->l_sl_mode.next = NULL;
         lock->l_sl_policy.prev = NULL;
@@ -353,9 +353,8 @@ int ldlm_lock_change_resource(struct ldlm_namespace *ns, struct ldlm_lock *lock,
                        sizeof(lock->l_resource->lr_name)) != 0);
         lock_res(newres);
         lock->l_resource = newres;
-        unlock_res(newres);
         unlock_res(oldres);
-        unlock_bitlock(lock);
+        unlock_res_and_lock(lock);
 
         /* ...and the flowers are still standing! */
         ldlm_resource_putref(oldres);