osd/mdd: new (weaker) locking for ->do_attr_{g,e}set(), see message to colibri@ for...

diff --git a/lustre/mdd/mdd_dir.c b/lustre/mdd/mdd_dir.c
index e812a01..90d612e 100644 (file)
--- a/lustre/mdd/mdd_dir.c
+++ b/lustre/mdd/mdd_dir.c
@@ -355,7 +355,7 @@ static int __mdd_index_delete(const struct lu_env *env, struct mdd_object *pobj,
 
         mdd_lproc_time_start(mdo2mdd(&pobj->mod_obj), &start,
                              LPROC_MDD_INDEX_DELETE);
-        
+
         if (dt_try_as_dir(env, next)) {
                 rc = next->do_index_ops->dio_delete(env, next,
                                                     (struct dt_key *)name,
@@ -425,9 +425,9 @@ static int mdd_link(const struct lu_env *env, struct md_object *tgt_obj,
                                      mdd_object_capa(env, mdd_tobj));
         if (rc)
                 GOTO(out_unlock, rc);
-        
+
         mdd_ref_add_internal(env, mdd_sobj, handle);
-       
+
         *la = ma->ma_attr;
         la->la_valid = LA_CTIME | LA_MTIME;
         rc = mdd_attr_set_internal_locked(env, mdd_tobj, la, handle, 0);
@@ -583,16 +583,15 @@ static int mdd_unlink(const struct lu_env *env, struct md_object *pobj,
 
         *la = ma->ma_attr;
         la->la_valid = LA_CTIME | LA_MTIME;
-        rc = mdd_attr_set_internal_locked(env, mdd_pobj, la,
-                                          handle, 0);
+        rc = mdd_attr_set_internal_locked(env, mdd_pobj, la, handle, 0);
         if (rc)
                 GOTO(cleanup, rc);
-                
+
         la->la_valid = LA_CTIME;
         rc = mdd_attr_set_internal(env, mdd_cobj, la, handle, 0);
         if (rc)
                 GOTO(cleanup, rc);
-        
+
         rc = mdd_finish_unlink(env, mdd_cobj, ma, handle);
 
         if (rc == 0)
@@ -658,8 +657,7 @@ static int mdd_name_insert(const struct lu_env *env, struct md_object *pobj,
 
         la->la_ctime = la->la_atime = CURRENT_SECONDS;
         la->la_valid = LA_ATIME | LA_CTIME;
-        rc = mdd_attr_set_internal_locked(env, mdd_obj, la,
-                                          handle, 0);
+        rc = mdd_attr_set_internal_locked(env, mdd_obj, la, handle, 0);
         EXIT;
 out_unlock:
         mdd_pdo_write_unlock(env, mdd_obj, dlh);
@@ -723,8 +721,7 @@ static int mdd_name_remove(const struct lu_env *env,
 
         la->la_ctime = la->la_mtime = CURRENT_SECONDS;
         la->la_valid = LA_CTIME | LA_MTIME;
-        rc = mdd_attr_set_internal_locked(env, mdd_obj, la,
-                                          handle, 0);
+        rc = mdd_attr_set_internal_locked(env, mdd_obj, la, handle, 0);
         EXIT;
 out_unlock:
         mdd_pdo_write_unlock(env, mdd_obj, dlh);
@@ -947,7 +944,7 @@ int mdd_object_initialize(const struct lu_env *env, const struct lu_fid *pfid,
 
         /*
          * Update attributes for child.
-         *  
+         *
          * FIXME:
          *  (1) the valid bits should be converted between Lustre and Linux;
          *  (2) maybe, the child attributes should be set in OSD when creation.
@@ -1062,7 +1059,7 @@ static int mdd_create(const struct lu_env *env,
         ENTRY;
 
         mdd_lproc_time_start(mdd, &start, LPROC_MDD_CREATE);
-        
+
         /*
          * Two operations have to be performed:
          *
@@ -1167,7 +1164,7 @@ static int mdd_create(const struct lu_env *env,
                 GOTO(cleanup, rc);
 
         inserted = 1;
-        
+
         /* Replay creates has objects already. */
         if (spec->u.sp_ea.no_lov_create) {
                 CDEBUG(D_INFO, "we already have lov ea\n");
@@ -1230,7 +1227,7 @@ cleanup:
                         mdd_write_unlock(env, son);
                 }
         }
-        
+
         /* Finish mdd_lov_create() stuff */
         mdd_lov_create_finish(env, mdd, rc);
         if (lmm && !spec->u.sp_ea.no_lov_create)
@@ -1402,17 +1399,16 @@ static int mdd_rename(const struct lu_env *env,
         mdd_sobj = mdd_object_find(env, mdd, lf);
         if (mdd_sobj) {
                 la->la_valid = LA_CTIME;
-                
+
                 /* XXX: How to update ctime for remote sobj? */
-                rc = mdd_attr_set_internal_locked(env, mdd_sobj, la,
-                                                  handle, 1);
+                rc = mdd_attr_set_internal_locked(env, mdd_sobj, la, handle, 1);
                 if (rc)
                         GOTO(cleanup, rc);
         }
         if (tobj && lu_object_exists(&tobj->mo_lu)) {
                 mdd_write_lock(env, mdd_tobj);
                 mdd_ref_del_internal(env, mdd_tobj, handle);
-                
+
                 /* Remove dot reference. */
                 if (is_dir)
                         mdd_ref_del_internal(env, mdd_tobj, handle);

diff --git a/lustre/mdd/mdd_object.c b/lustre/mdd/mdd_object.c
index fca738d..ccf0d21 100644 (file)
--- a/lustre/mdd/mdd_object.c
+++ b/lustre/mdd/mdd_object.c
@@ -315,7 +315,7 @@ static int __mdd_lmv_get(const struct lu_env *env,
 
         if (ma->ma_valid & MA_LMV)
                 RETURN(0);
-        
+
         rc = mdd_get_md(env, mdd_obj, ma->ma_lmv, &ma->ma_lmv_size,
                         MDS_LMV_MD_NAME);
         if (rc > 0) {
@@ -491,9 +491,16 @@ int mdd_attr_set_internal_locked(const struct lu_env *env,
                                  struct thandle *handle, int needacl)
 {
         int rc;
-        mdd_write_lock(env, o);
+
+        needacl = needacl && (attr->la_valid & LA_MODE);
+
+        if (needacl)
+                mdd_write_lock(env, o);
+
         rc = mdd_attr_set_internal(env, o, attr, handle, needacl);
-        mdd_write_unlock(env, o);
+
+        if (needacl)
+                mdd_write_unlock(env, o);
         return rc;
 }
 
@@ -878,12 +885,12 @@ int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
 
         mdd_write_lock(env, mdd_obj);
         rc = __mdd_xattr_del(env, mdd, md2mdd_obj(obj), name, handle);
+        mdd_write_unlock(env, mdd_obj);
         if (rc == 0) {
                 la_copy->la_ctime = CURRENT_SECONDS;
                 la_copy->la_valid = LA_CTIME;
                 rc = mdd_attr_set_internal(env, mdd_obj, la_copy, handle, 0);
         }
-        mdd_write_unlock(env, mdd_obj);
 
         mdd_trans_stop(env, mdd, rc, handle);
 
@@ -1006,7 +1013,7 @@ static int mdd_object_create(const struct lu_env *env,
 
                 CDEBUG(D_INFO, "Set slave ea "DFID", eadatalen %d, rc %d\n",
                        PFID(mdo2fid(mdd_obj)), spec->u.sp_ea.eadatalen, rc);
-                
+
                 rc = mdd_attr_set_internal(env, mdd_obj, &ma->ma_attr, handle, 0);
         } else {
 #ifdef CONFIG_FS_POSIX_ACL
@@ -1058,13 +1065,14 @@ static int mdd_ref_add(const struct lu_env *env,
 
         mdd_write_lock(env, mdd_obj);
         rc = mdd_link_sanity_check(env, NULL, mdd_obj);
-        if (rc == 0) {
+        if (rc == 0)
                 mdd_ref_add_internal(env, mdd_obj, handle);
+        mdd_write_unlock(env, mdd_obj);
+        if (rc == 0) {
                 la_copy->la_ctime = CURRENT_SECONDS;
                 la_copy->la_valid = LA_CTIME;
                 rc = mdd_attr_set_internal(env, mdd_obj, la_copy, handle, 0);
         }
-        mdd_write_unlock(env, mdd_obj);
         mdd_trans_stop(env, mdd, 0, handle);
 
         RETURN(rc);
@@ -1269,7 +1277,7 @@ static int mdd_dir_page_build(const struct lu_env *env, int first,
                 recsize = (sizeof(*ent) + len + 3) & ~3;
                 hash = iops->store(env, it);
                 *end = hash;
-                
+
                 CDEBUG(D_INFO, "%p %p %d "DFID": %#8.8x (%d) \"%*.*s\"\n",
                        name, ent, nob, PFID(fid2), hash, len, len, len, name);
 

diff --git a/lustre/osd/osd_handler.c b/lustre/osd/osd_handler.c
index 1f643e3..42df505 100644 (file)
--- a/lustre/osd/osd_handler.c
+++ b/lustre/osd/osd_handler.c
@@ -86,6 +86,8 @@ struct osd_object {
         struct rw_semaphore    oo_sem;
         struct iam_container   oo_container;
         struct iam_descr       oo_descr;
+        /* protects inode attributes */
+        spinlock_t             oo_guard;
         const struct lu_env   *oo_owner;
 };
 
@@ -153,9 +155,9 @@ static int   osd_device_init   (const struct lu_env *env,
 static int   osd_fid_lookup    (const struct lu_env *env,
                                 struct osd_object *obj,
                                 const struct lu_fid *fid);
-static int   osd_inode_getattr (const struct lu_env *env,
+static void  osd_inode_getattr (const struct lu_env *env,
                                 struct inode *inode, struct lu_attr *attr);
-static int   osd_inode_setattr (const struct lu_env *env,
+static void  osd_inode_setattr (const struct lu_env *env,
                                 struct inode *inode, const struct lu_attr *attr);
 static int   osd_param_is_sane (const struct osd_device *dev,
                                 const struct txn_param *param);
@@ -335,6 +337,7 @@ static struct lu_object *osd_object_alloc(const struct lu_env *env,
                 mo->oo_dt.do_ops = &osd_obj_ops;
                 l->lo_ops = &osd_lu_obj_ops;
                 init_rwsem(&mo->oo_sem);
+                spin_lock_init(&mo->oo_guard);
                 return l;
         } else
                 return NULL;
@@ -923,12 +926,14 @@ static int osd_attr_get(const struct lu_env *env,
 
         LASSERT(dt_object_exists(dt));
         LASSERT(osd_invariant(obj));
-        LASSERT(osd_read_locked(env, obj) || osd_write_locked(env, obj));
 
         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
                 return -EACCES;
 
-        return osd_inode_getattr(env, obj->oo_inode, attr);
+        spin_lock(&obj->oo_guard);
+        osd_inode_getattr(env, obj->oo_inode, attr);
+        spin_unlock(&obj->oo_guard);
+        return 0;
 }
 
 static int osd_attr_set(const struct lu_env *env,
@@ -942,12 +947,16 @@ static int osd_attr_set(const struct lu_env *env,
         LASSERT(handle != NULL);
         LASSERT(dt_object_exists(dt));
         LASSERT(osd_invariant(obj));
-        LASSERT(osd_write_locked(env, obj));
 
         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
                 return -EACCES;
 
-        return osd_inode_setattr(env, obj->oo_inode, attr);
+        spin_lock(&obj->oo_guard);
+        osd_inode_setattr(env, obj->oo_inode, attr);
+        spin_unlock(&obj->oo_guard);
+
+        mark_inode_dirty(obj->oo_inode);
+        return 0;
 }
 
 static struct timespec *osd_inode_time(const struct lu_env *env,
@@ -962,8 +971,8 @@ static struct timespec *osd_inode_time(const struct lu_env *env,
         return t;
 }
 
-static int osd_inode_setattr(const struct lu_env *env,
-                             struct inode *inode, const struct lu_attr *attr)
+static void osd_inode_setattr(const struct lu_env *env,
+                              struct inode *inode, const struct lu_attr *attr)
 {
         __u64 bits;
 
@@ -1001,8 +1010,6 @@ static int osd_inode_setattr(const struct lu_env *env,
                 li->i_flags = (li->i_flags & ~LDISKFS_FL_USER_MODIFIABLE) |
                         (attr->la_flags & LDISKFS_FL_USER_MODIFIABLE);
         }
-        mark_inode_dirty(inode);
-        return 0;
 }
 
 /*
@@ -1219,12 +1226,16 @@ static void osd_object_ref_add(const struct lu_env *env,
         LASSERT(osd_write_locked(env, obj));
         LASSERT(th != NULL);
 
+        spin_lock(&obj->oo_guard);
         if (inode->i_nlink < LDISKFS_LINK_MAX) {
                 inode->i_nlink ++;
+                spin_unlock(&obj->oo_guard);
                 mark_inode_dirty(inode);
-        } else
+        } else {
+                spin_unlock(&obj->oo_guard);
                 LU_OBJECT_DEBUG(D_ERROR, env, &dt->do_lu,
                                 "Overflowed nlink\n");
+        }
         LASSERT(osd_invariant(obj));
 }
 
@@ -1243,12 +1254,16 @@ static void osd_object_ref_del(const struct lu_env *env,
         LASSERT(osd_write_locked(env, obj));
         LASSERT(th != NULL);
 
+        spin_lock(&obj->oo_guard);
         if (inode->i_nlink > 0) {
                 inode->i_nlink --;
+                spin_unlock(&obj->oo_guard);
                 mark_inode_dirty(inode);
-        } else
+        } else {
+                spin_unlock(&obj->oo_guard);
                 LU_OBJECT_DEBUG(D_ERROR, env, &dt->do_lu,
                                 "Underflowed nlink\n");
+        }
         LASSERT(osd_invariant(obj));
 }
 
@@ -2276,8 +2291,8 @@ static int osd_fid_lookup(const struct lu_env *env,
         RETURN(result);
 }
 
-static int osd_inode_getattr(const struct lu_env *env,
-                             struct inode *inode, struct lu_attr *attr)
+static void osd_inode_getattr(const struct lu_env *env,
+                              struct inode *inode, struct lu_attr *attr)
 {
         attr->la_valid      |= LA_ATIME | LA_MTIME | LA_CTIME | LA_MODE |
                                LA_SIZE | LA_BLOCKS | LA_UID | LA_GID |
@@ -2295,7 +2310,6 @@ static int osd_inode_getattr(const struct lu_env *env,
         attr->la_nlink      = inode->i_nlink;
         attr->la_rdev       = inode->i_rdev;
         attr->la_blksize    = inode->i_blksize;
-        return 0;
 }
 
 /*