LU-1876 hsm: bugfix for deadlock in glimpse and layout change

This dead lock is as follows: glimpse calls ccc_inode_lsm_get()
inside glimpse IO, it will be blocked at grabbing lo_type_guard
sem, because another thread is changing layout which holds
lo_type_guard but waiting for all active IO to complete.

The basic rule: any attempt to grab lo_type_guard inside an IO is
generically risky.

Signed-off-by: Jinshan Xiong <jinshan.xiong@intel.com>
Change-Id: Ib41c03f0a73e2ef7cd37e7689c2d55f7fd0484bf
Reviewed-on: http://review.whamcloud.com/5158
Reviewed-by: jacques-Charles Lafoucriere <jacques-charles.lafoucriere@cea.fr>
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Johann Lombardi <johann.lombardi@intel.com>

diff --git a/lustre/lclient/glimpse.c b/lustre/lclient/glimpse.c
index c4b44de..b5762a6 100644 (file)
--- a/lustre/lclient/glimpse.c
+++ b/lustre/lclient/glimpse.c
@@ -157,7 +157,7 @@ int cl_glimpse_lock(const struct lu_env *env, struct cl_io *io,
                         LASSERT(agl == 0);
                         result = cl_wait(env, lock);
                         if (result == 0) {
-                                cl_merge_lvb(inode);
+                               cl_merge_lvb(env, inode);
                                 if (cl_isize_read(inode) > 0 &&
                                     inode->i_blocks == 0) {
                                         /*
@@ -173,7 +173,7 @@ int cl_glimpse_lock(const struct lu_env *env, struct cl_io *io,
                         cl_lock_release(env, lock, "glimpse", cfs_current());
                 } else {
                         CDEBUG(D_DLMTRACE, "No objects for inode\n");
-                       cl_merge_lvb(inode);
+                       cl_merge_lvb(env, inode);
                 }
         }
 
@@ -279,7 +279,7 @@ int cl_local_size(struct inode *inode)
                 descr->cld_obj = clob;
                 lock = cl_lock_peek(env, io, descr, "localsize", cfs_current());
                 if (lock != NULL) {
-                        cl_merge_lvb(inode);
+                       cl_merge_lvb(env, inode);
                         cl_unuse(env, lock);
                         cl_lock_release(env, lock, "localsize", cfs_current());
                         result = 0;

diff --git a/lustre/liblustre/llite_lib.h b/lustre/liblustre/llite_lib.h
index 88b69b9..c587ad5 100644 (file)
--- a/lustre/liblustre/llite_lib.h
+++ b/lustre/liblustre/llite_lib.h
@@ -329,7 +329,7 @@ static inline struct ext2_dirent *ext2_next_entry(struct ext2_dirent *p)
         return (struct ext2_dirent*)((char*) p + le16_to_cpu(p->rec_len));
 }
 
-int llu_merge_lvb(struct inode *inode);
+int llu_merge_lvb(const struct lu_env *env, struct inode *inode);
 
 static inline void inode_init_lvb(struct inode *inode, struct ost_lvb *lvb)
 {
@@ -415,9 +415,9 @@ static inline void cl_isize_unlock(struct inode *inode)
 {
 }
 
-static inline int cl_merge_lvb(struct inode *inode)
+static inline int cl_merge_lvb(const struct lu_env *env, struct inode *inode)
 {
-        return llu_merge_lvb(inode);
+       return llu_merge_lvb(env, inode);
 }
 
 #define cl_inode_atime(inode) (llu_i2stat(inode)->st_atime)

diff --git a/lustre/liblustre/rw.c b/lustre/liblustre/rw.c
index 1263f56..8caa98f 100644 (file)
--- a/lustre/liblustre/rw.c
+++ b/lustre/liblustre/rw.c
@@ -203,38 +203,41 @@ static int llu_glimpse_callback(struct ldlm_lock *lock, void *reqp)
         return rc;
 }
 
-int llu_merge_lvb(struct inode *inode)
+int llu_merge_lvb(const struct lu_env *env, struct inode *inode)
 {
        struct llu_inode_info *lli = llu_i2info(inode);
-       struct llu_sb_info *sbi = llu_i2sbi(inode);
+       struct cl_object *obj = lli->lli_clob;
        struct intnl_stat *st = llu_i2stat(inode);
+       struct cl_attr *attr = ccc_env_thread_attr(env);
        struct ost_lvb lvb;
-       struct lov_stripe_md *lsm;
        int rc;
        ENTRY;
 
-       lsm = ccc_inode_lsm_get(inode);
-       if (lsm == NULL)
-               RETURN(0);
-
-       lov_stripe_lock(lsm);
-        inode_init_lvb(inode, &lvb);
-        /* merge timestamps the most resently obtained from mds with
-           timestamps obtained from osts */
-        lvb.lvb_atime = lli->lli_lvb.lvb_atime;
-        lvb.lvb_mtime = lli->lli_lvb.lvb_mtime;
-        lvb.lvb_ctime = lli->lli_lvb.lvb_ctime;
-       rc = obd_merge_lvb(sbi->ll_dt_exp, lsm, &lvb, 0);
-        st->st_size = lvb.lvb_size;
-        st->st_blocks = lvb.lvb_blocks;
-        /* handle st_blocks overflow gracefully */
-        if (st->st_blocks < lvb.lvb_blocks)
-                st->st_blocks = ~0UL;
-        st->st_mtime = lvb.lvb_mtime;
-        st->st_atime = lvb.lvb_atime;
-        st->st_ctime = lvb.lvb_ctime;
-       lov_stripe_unlock(lsm);
-       ccc_inode_lsm_put(inode, lsm);
+       /* merge timestamps the most recently obtained from mds with
+          timestamps obtained from osts */
+       LTIME_S(inode->i_atime) = lli->lli_lvb.lvb_atime;
+       LTIME_S(inode->i_mtime) = lli->lli_lvb.lvb_mtime;
+       LTIME_S(inode->i_ctime) = lli->lli_lvb.lvb_ctime;
+
+       inode_init_lvb(inode, &lvb);
+
+       cl_object_attr_lock(obj);
+       rc = cl_object_attr_get(env, obj, attr);
+       cl_object_attr_unlock(obj);
+       if (rc == 0) {
+               if (lvb.lvb_atime < attr->cat_atime)
+                       lvb.lvb_atime = attr->cat_atime;
+               if (lvb.lvb_ctime < attr->cat_ctime)
+                       lvb.lvb_ctime = attr->cat_ctime;
+               if (lvb.lvb_mtime < attr->cat_mtime)
+                       lvb.lvb_mtime = attr->cat_mtime;
+
+               st->st_size = lvb.lvb_size;
+               st->st_blocks = lvb.lvb_blocks;
+               st->st_mtime = lvb.lvb_mtime;
+               st->st_atime = lvb.lvb_atime;
+               st->st_ctime = lvb.lvb_ctime;
+       }
 
        RETURN(rc);
 }

diff --git a/lustre/llite/file.c b/lustre/llite/file.c
index 067261b..45f2131 100644 (file)
--- a/lustre/llite/file.c
+++ b/lustre/llite/file.c
@@ -767,38 +767,47 @@ int ll_inode_getattr(struct inode *inode, struct obdo *obdo,
        RETURN(rc);
 }
 
-int ll_merge_lvb(struct inode *inode)
+int ll_merge_lvb(const struct lu_env *env, struct inode *inode)
 {
        struct ll_inode_info *lli = ll_i2info(inode);
-       struct ll_sb_info *sbi = ll_i2sbi(inode);
-       struct lov_stripe_md *lsm;
+       struct cl_object *obj = lli->lli_clob;
+       struct cl_attr *attr = ccc_env_thread_attr(env);
        struct ost_lvb lvb;
        int rc = 0;
 
        ENTRY;
 
-       lsm = ccc_inode_lsm_get(inode);
        ll_inode_size_lock(inode);
+       /* merge timestamps the most recently obtained from mds with
+          timestamps obtained from osts */
+       LTIME_S(inode->i_atime) = lli->lli_lvb.lvb_atime;
+       LTIME_S(inode->i_mtime) = lli->lli_lvb.lvb_mtime;
+       LTIME_S(inode->i_ctime) = lli->lli_lvb.lvb_ctime;
        inode_init_lvb(inode, &lvb);
 
-       /* merge timestamps the most resently obtained from mds with
-          timestamps obtained from osts */
-       lvb.lvb_atime = lli->lli_lvb.lvb_atime;
-       lvb.lvb_mtime = lli->lli_lvb.lvb_mtime;
-       lvb.lvb_ctime = lli->lli_lvb.lvb_ctime;
-       if (lsm != NULL) {
-               rc = obd_merge_lvb(sbi->ll_dt_exp, lsm, &lvb, 0);
-               cl_isize_write_nolock(inode, lvb.lvb_size);
+       cl_object_attr_lock(obj);
+       rc = cl_object_attr_get(env, obj, attr);
+       cl_object_attr_unlock(obj);
+
+       if (rc == 0) {
+               if (lvb.lvb_atime < attr->cat_atime)
+                       lvb.lvb_atime = attr->cat_atime;
+               if (lvb.lvb_ctime < attr->cat_ctime)
+                       lvb.lvb_ctime = attr->cat_ctime;
+               if (lvb.lvb_mtime < attr->cat_mtime)
+                       lvb.lvb_mtime = attr->cat_mtime;
 
                CDEBUG(D_VFSTRACE, DFID" updating i_size "LPU64"\n",
-                               PFID(&lli->lli_fid), lvb.lvb_size);
-               inode->i_blocks = lvb.lvb_blocks;
+                               PFID(&lli->lli_fid), attr->cat_size);
+               cl_isize_write_nolock(inode, attr->cat_size);
+
+               inode->i_blocks = attr->cat_blocks;
+
+               LTIME_S(inode->i_mtime) = lvb.lvb_mtime;
+               LTIME_S(inode->i_atime) = lvb.lvb_atime;
+               LTIME_S(inode->i_ctime) = lvb.lvb_ctime;
        }
-       LTIME_S(inode->i_mtime) = lvb.lvb_mtime;
-       LTIME_S(inode->i_atime) = lvb.lvb_atime;
-       LTIME_S(inode->i_ctime) = lvb.lvb_ctime;
        ll_inode_size_unlock(inode);
-       ccc_inode_lsm_put(inode, lsm);
 
        RETURN(rc);
 }

diff --git a/lustre/llite/llite_internal.h b/lustre/llite/llite_internal.h
index cc03923..ee76a52 100644 (file)
--- a/lustre/llite/llite_internal.h
+++ b/lustre/llite/llite_internal.h
@@ -780,7 +780,7 @@ int ll_fsync(struct file *file, struct dentry *dentry, int data);
 #endif
 int ll_do_fiemap(struct inode *inode, struct ll_user_fiemap *fiemap,
               int num_bytes);
-int ll_merge_lvb(struct inode *inode);
+int ll_merge_lvb(const struct lu_env *env, struct inode *inode);
 int ll_get_grouplock(struct inode *inode, struct file *file, unsigned long arg);
 int ll_put_grouplock(struct inode *inode, struct file *file, unsigned long arg);
 int ll_fid2path(struct inode *inode, void *arg);
@@ -1436,9 +1436,9 @@ static inline void cl_isize_write(struct inode *inode, loff_t kms)
 
 #define cl_isize_read(inode)             i_size_read(inode)
 
-static inline int cl_merge_lvb(struct inode *inode)
+static inline int cl_merge_lvb(const struct lu_env *env, struct inode *inode)
 {
-        return ll_merge_lvb(inode);
+       return ll_merge_lvb(env, inode);
 }
 
 #define cl_inode_atime(inode) LTIME_S((inode)->i_atime)

diff --git a/lustre/lov/lov_object.c b/lustre/lov/lov_object.c
index c1cca70..f4f2861 100644 (file)
--- a/lustre/lov/lov_object.c
+++ b/lustre/lov/lov_object.c
@@ -389,12 +389,10 @@ static int lov_attr_get_empty(const struct lu_env *env, struct cl_object *obj,
 static int lov_attr_get_raid0(const struct lu_env *env, struct cl_object *obj,
                               struct cl_attr *attr)
 {
-       struct lov_object       *lov = cl2lov(obj);
+       struct lov_object       *lov = cl2lov(obj);
        struct lov_layout_raid0 *r0 = lov_r0(lov);
-       struct lov_stripe_md    *lsm = lov->lo_lsm;
-        struct ost_lvb          *lvb = &lov_env_info(env)->lti_lvb;
-        __u64                    kms;
-        int                      result = 0;
+       struct cl_attr          *lov_attr = &r0->lo_attr;
+       int                      result = 0;
 
         ENTRY;
 
@@ -408,38 +406,51 @@ static int lov_attr_get_raid0(const struct lu_env *env, struct cl_object *obj,
         * can't go if locks exist. */
        /* LASSERT(cfs_atomic_read(&lsm->lsm_refc) > 1); */
 
-        if (!r0->lo_attr_valid) {
-                /*
-                 * Fill LVB with attributes already initialized by the upper
-                 * layer.
-                 */
-                cl_attr2lvb(lvb, attr);
-                kms = attr->cat_kms;
-
-                /*
-                 * XXX that should be replaced with a loop over sub-objects,
-                 * doing cl_object_attr_get() on them. But for now, let's
-                 * reuse old lov code.
-                 */
-
-                /*
-                 * XXX take lsm spin-lock to keep lov_merge_lvb_kms()
-                 * happy. It's not needed, because new code uses
-                 * ->coh_attr_guard spin-lock to protect consistency of
-                 * sub-object attributes.
-                 */
-                lov_stripe_lock(lsm);
-                result = lov_merge_lvb_kms(lsm, lvb, &kms);
-                lov_stripe_unlock(lsm);
-                if (result == 0) {
-                        cl_lvb2attr(attr, lvb);
-                        attr->cat_kms = kms;
-                        r0->lo_attr_valid = 1;
-                        r0->lo_attr = *attr;
-                }
-        } else
-                *attr = r0->lo_attr;
-        RETURN(result);
+       if (!r0->lo_attr_valid) {
+               struct lov_stripe_md    *lsm = lov->lo_lsm;
+               struct ost_lvb          *lvb = &lov_env_info(env)->lti_lvb;
+               __u64                    kms = 0;
+
+               memset(lvb, 0, sizeof(*lvb));
+               /* XXX: timestamps can be negative by sanity:test_39m,
+                * how can it be? */
+               lvb->lvb_atime = LLONG_MIN;
+               lvb->lvb_ctime = LLONG_MIN;
+               lvb->lvb_mtime = LLONG_MIN;
+
+               /*
+                * XXX that should be replaced with a loop over sub-objects,
+                * doing cl_object_attr_get() on them. But for now, let's
+                * reuse old lov code.
+                */
+
+               /*
+                * XXX take lsm spin-lock to keep lov_merge_lvb_kms()
+                * happy. It's not needed, because new code uses
+                * ->coh_attr_guard spin-lock to protect consistency of
+                * sub-object attributes.
+                */
+               lov_stripe_lock(lsm);
+               result = lov_merge_lvb_kms(lsm, lvb, &kms);
+               lov_stripe_unlock(lsm);
+               if (result == 0) {
+                       cl_lvb2attr(lov_attr, lvb);
+                       lov_attr->cat_kms = kms;
+                       r0->lo_attr_valid = 1;
+               }
+       }
+       if (result == 0) { /* merge results */
+               attr->cat_blocks = lov_attr->cat_blocks;
+               attr->cat_size = lov_attr->cat_size;
+               attr->cat_kms = lov_attr->cat_kms;
+               if (attr->cat_atime < lov_attr->cat_atime)
+                       attr->cat_atime = lov_attr->cat_atime;
+               if (attr->cat_ctime < lov_attr->cat_ctime)
+                       attr->cat_ctime = lov_attr->cat_ctime;
+               if (attr->cat_mtime < lov_attr->cat_mtime)
+                       attr->cat_mtime = lov_attr->cat_mtime;
+       }
+       RETURN(result);
 }
 
 const static struct lov_layout_operations lov_dispatch[] = {