LU-22 ldlm_resource::lr_lvb_data is protected by wrong lock

Use lr_lock to protect lr_lvbo_data while lr_lvbo_sem is only for
lvbo_init serialization.

Bugzilla: 24336
Change-Id: Ic3e77e99d8b3a3ca277adbc6548c254969e9761a
Signed-off-by: Bobi Jam <bobijam@whamcloud.com>
Reviewed-on: http://review.whamcloud.com/379
Tested-by: Hudson
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/include/lustre_dlm.h b/lustre/include/lustre_dlm.h
index 79f8d08..01ba8f8 100644 (file)
--- a/lustre/include/lustre_dlm.h
+++ b/lustre/include/lustre_dlm.h
@@ -778,8 +778,10 @@ struct ldlm_resource {
         struct ldlm_interval_tree lr_itree[LCK_MODE_NUM];  /* interval trees*/
 
         /* Server-side-only lock value block elements */
         struct ldlm_interval_tree lr_itree[LCK_MODE_NUM];  /* interval trees*/
 
         /* Server-side-only lock value block elements */

+        /** to serialize lvbo_init */

         cfs_semaphore_t        lr_lvb_sem;
         __u32                  lr_lvb_len;
         cfs_semaphore_t        lr_lvb_sem;
         __u32                  lr_lvb_len;

+        /** protect by lr_lock */

         void                  *lr_lvb_data;
 
         /* when the resource was considered as contended */
         void                  *lr_lvb_data;
 
         /* when the resource was considered as contended */

diff --git a/lustre/ldlm/ldlm_lockd.c b/lustre/ldlm/ldlm_lockd.c
index f5fc1f8..cfedb83 100644 (file)
--- a/lustre/ldlm/ldlm_lockd.c
+++ b/lustre/ldlm/ldlm_lockd.c
@@ -851,10 +851,10 @@ int ldlm_server_completion_ast(struct ldlm_lock *lock, int flags, void *data)
         if (lock->l_resource->lr_lvb_len) {
                 void *lvb = req_capsule_client_get(&req->rq_pill, &RMF_DLM_LVB);
 
         if (lock->l_resource->lr_lvb_len) {
                 void *lvb = req_capsule_client_get(&req->rq_pill, &RMF_DLM_LVB);
 

-                cfs_down(&lock->l_resource->lr_lvb_sem);
+                lock_res(lock->l_resource);

                 memcpy(lvb, lock->l_resource->lr_lvb_data,
                        lock->l_resource->lr_lvb_len);
                 memcpy(lvb, lock->l_resource->lr_lvb_data,
                        lock->l_resource->lr_lvb_len);

-                cfs_up(&lock->l_resource->lr_lvb_sem);
+                unlock_res(lock->l_resource);

         }
 
         LDLM_DEBUG(lock, "server preparing completion AST (after %lds wait)",
         }
 
         LDLM_DEBUG(lock, "server preparing completion AST (after %lds wait)",


@@ -1137,6 +1137,9 @@ existing_lock:
                  * local_lock_enqueue by the policy function. */
                 cookie = req;
         } else {
                  * local_lock_enqueue by the policy function. */
                 cookie = req;
         } else {

+                /* based on the assumption that lvb size never changes during
+                 * resource life time otherwise it need resource->lr_lock's
+                 * protection */

                 if (lock->l_resource->lr_lvb_len) {
                         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB,
                                              RCL_SERVER,
                 if (lock->l_resource->lr_lvb_len) {
                         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB,
                                              RCL_SERVER,


@@ -1244,17 +1247,19 @@ existing_lock:
 
                 if (rc == 0) {
                         if (lock->l_resource->lr_lvb_len > 0) {
 
                 if (rc == 0) {
                         if (lock->l_resource->lr_lvb_len > 0) {

+                                /* MDT path won't handle lr_lvb_data, so
+                                 * lock/unlock better be contained in the
+                                 * if block */

                                 void *lvb;
 
                                 lvb = req_capsule_server_get(&req->rq_pill,
                                                              &RMF_DLM_LVB);
                                 LASSERTF(lvb != NULL, "req %p, lock %p\n",
                                          req, lock);
                                 void *lvb;
 
                                 lvb = req_capsule_server_get(&req->rq_pill,
                                                              &RMF_DLM_LVB);
                                 LASSERTF(lvb != NULL, "req %p, lock %p\n",
                                          req, lock);

-
-                                cfs_down(&lock->l_resource->lr_lvb_sem);
+                                lock_res(lock->l_resource);

                                 memcpy(lvb, lock->l_resource->lr_lvb_data,
                                        lock->l_resource->lr_lvb_len);
                                 memcpy(lvb, lock->l_resource->lr_lvb_data,
                                        lock->l_resource->lr_lvb_len);

-                                cfs_up(&lock->l_resource->lr_lvb_sem);
+                                unlock_res(lock->l_resource);

                         }
                 } else {
                         lock_res_and_lock(lock);
                         }
                 } else {
                         lock_res_and_lock(lock);

diff --git a/lustre/obdfilter/filter_lvb.c b/lustre/obdfilter/filter_lvb.c
index 5010eb6..080db86 100644 (file)
--- a/lustre/obdfilter/filter_lvb.c
+++ b/lustre/obdfilter/filter_lvb.c
@@ -95,7 +95,7 @@ static int filter_lvbo_init(struct ldlm_resource *res)
                LPU64")\n", obd->obd_name, res->lr_name.name[1],
                res->lr_name.name[0]);
 
                LPU64")\n", obd->obd_name, res->lr_name.name[1],
                res->lr_name.name[0]);
 

-        dentry = filter_fid2dentry(obd, NULL, res->lr_name.name[1], 
+        dentry = filter_fid2dentry(obd, NULL, res->lr_name.name[1],

                                               res->lr_name.name[0]);
         if (IS_ERR(dentry)) {
                 rc = PTR_ERR(dentry);
                                               res->lr_name.name[0]);
         if (IS_ERR(dentry)) {
                 rc = PTR_ERR(dentry);


@@ -129,8 +129,8 @@ out_dentry:
 
 /* This will be called in two ways:
  *
 
 /* This will be called in two ways:
  *

- *   m != NULL : called by the DLM itself after a glimpse callback
- *   m == NULL : called by the filter after a disk write
+ *   r != NULL : called by the DLM itself after a glimpse callback
+ *   r == NULL : called by the filter after a disk write

  *
  *   If 'increase_only' is true, don't allow values to move backwards.
  */
  *
  *   If 'increase_only' is true, don't allow values to move backwards.
  */


@@ -141,11 +141,12 @@ static int filter_lvbo_update(struct ldlm_resource *res,
         struct ost_lvb *lvb;
         struct obd_device *obd;
         struct inode *inode;
         struct ost_lvb *lvb;
         struct obd_device *obd;
         struct inode *inode;

+        struct inode *tmpinode = NULL;

         ENTRY;
 
         LASSERT(res);
 
         ENTRY;
 
         LASSERT(res);
 

-        cfs_down(&res->lr_lvb_sem);
+        lock_res(res);

         lvb = res->lr_lvb_data;
         if (lvb == NULL) {
                 CERROR("No lvb when running lvbo_update!\n");
         lvb = res->lr_lvb_data;
         if (lvb == NULL) {
                 CERROR("No lvb when running lvbo_update!\n");


@@ -195,18 +196,28 @@ static int filter_lvbo_update(struct ldlm_resource *res,
         LASSERT(obd);
 
         inode = res->lr_lvb_inode;
         LASSERT(obd);
 
         inode = res->lr_lvb_inode;

-        /* filter_fid2dentry could fail */
-        if (unlikely(!inode)) {
+        /* filter_fid2dentry could fail, esp. in OBD_FAIL_OST_ENOENT test case */
+        if (unlikely(inode == NULL)) {

                 struct dentry *dentry;
 
                 struct dentry *dentry;
 

-                dentry = filter_fid2dentry(obd, NULL, res->lr_name.name[1], 
+                unlock_res(res);
+
+                dentry = filter_fid2dentry(obd, NULL, res->lr_name.name[1],

                                            res->lr_name.name[0]);
                 if (IS_ERR(dentry))
                                            res->lr_name.name[0]);
                 if (IS_ERR(dentry))

-                        GOTO(out, rc = PTR_ERR(dentry));
+                        RETURN(PTR_ERR(dentry));

 
                 if (dentry->d_inode)
 
                 if (dentry->d_inode)

-                        inode = res->lr_lvb_inode = igrab(dentry->d_inode);
+                        tmpinode = igrab(dentry->d_inode);

                 f_dput(dentry);
                 f_dput(dentry);

+                /* tmpinode could be NULL, but it does not matter if other
+                 * have set res->lr_lvb_inode */
+                lock_res(res);
+                if (res->lr_lvb_inode == NULL) {
+                        res->lr_lvb_inode = tmpinode;
+                        tmpinode = NULL;
+                }
+                inode = res->lr_lvb_inode;

         }
 
         if (!inode || !inode->i_nlink)
         }
 
         if (!inode || !inode->i_nlink)


@@ -245,7 +256,9 @@ static int filter_lvbo_update(struct ldlm_resource *res,
         }
 
 out:
         }
 
 out:

-        cfs_up(&res->lr_lvb_sem);
+        unlock_res(res);
+        if (tmpinode)
+                iput(tmpinode);

         return rc;
 }
 
         return rc;
 }