LU-7713 osd: osd-zfs should serialize destroy vs. others

otherwise we can get unexpected EEXIST from DMU at any time.
sanityn/91 hits this regularly.

Change-Id: I948413a0689f1ceae7f073a1e33adef023eb274c
Signed-off-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Reviewed-on: http://review.whamcloud.com/18155
Tested-by: Jenkins
Reviewed-by: John L. Hammond <john.hammond@intel.com>
Reviewed-by: Nathaniel Clark <nathaniel.l.clark@intel.com>
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/mdd/mdd_dir.c b/lustre/mdd/mdd_dir.c
index 22c3f2b..eab1629 100644 (file)
--- a/lustre/mdd/mdd_dir.c
+++ b/lustre/mdd/mdd_dir.c
@@ -1593,7 +1593,7 @@ static bool mdd_hsm_archive_exists(const struct lu_env *env,
                if (rc < 0)
                        RETURN(false);
 
                if (rc < 0)
                        RETURN(false);
 

-               ma->ma_valid = MA_HSM;
+               ma->ma_valid |= MA_HSM;

        }
        if (ma->ma_hsm.mh_flags & HS_EXISTS)
                RETURN(true);
        }
        if (ma->ma_hsm.mh_flags & HS_EXISTS)
                RETURN(true);


@@ -1621,7 +1621,7 @@ static int mdd_unlink(const struct lu_env *env, struct md_object *pobj,
        struct mdd_object *mdd_cobj = NULL;
        struct mdd_device *mdd = mdo2mdd(pobj);
        struct thandle    *handle;
        struct mdd_object *mdd_cobj = NULL;
        struct mdd_device *mdd = mdo2mdd(pobj);
        struct thandle    *handle;

-       int rc, is_dir = 0;
+       int rc, is_dir = 0, cl_flags = 0;

        ENTRY;
 
        /* cobj == NULL means only delete name entry */
        ENTRY;
 
        /* cobj == NULL means only delete name entry */


@@ -1642,6 +1642,11 @@ static int mdd_unlink(const struct lu_env *env, struct md_object *pobj,
                        RETURN(rc);
 
                is_dir = S_ISDIR(cattr->la_mode);
                        RETURN(rc);
 
                is_dir = S_ISDIR(cattr->la_mode);

+               /* search for an existing archive.
+                * we should check ahead as the object
+                * can be destroyed in this transaction */
+               if (mdd_hsm_archive_exists(env, mdd_cobj, ma))
+                       cl_flags |= CLF_UNLINK_HSM_EXISTS;

        }
 
        rc = mdd_unlink_sanity_check(env, mdd_pobj, pattr, mdd_cobj, cattr);
        }
 
        rc = mdd_unlink_sanity_check(env, mdd_pobj, pattr, mdd_cobj, cattr);


@@ -1719,11 +1724,10 @@ static int mdd_unlink(const struct lu_env *env, struct md_object *pobj,
        ma->ma_attr = *cattr;
        ma->ma_valid |= MA_INODE;
        rc = mdd_finish_unlink(env, mdd_cobj, ma, mdd_pobj, lname, handle);
        ma->ma_attr = *cattr;
        ma->ma_valid |= MA_INODE;
        rc = mdd_finish_unlink(env, mdd_cobj, ma, mdd_pobj, lname, handle);

-
-       /* fetch updated nlink */

        if (rc != 0)
                GOTO(cleanup, rc);
 
        if (rc != 0)
                GOTO(cleanup, rc);
 

+       /* fetch updated nlink */

        rc = mdd_la_get(env, mdd_cobj, cattr);
        /* if object is removed then we can't get its attrs,
         * use last get */
        rc = mdd_la_get(env, mdd_cobj, cattr);
        /* if object is removed then we can't get its attrs,
         * use last get */


@@ -1743,14 +1747,10 @@ cleanup:
                mdd_write_unlock(env, mdd_cobj);
 
        if (rc == 0) {
                mdd_write_unlock(env, mdd_cobj);
 
        if (rc == 0) {

-               int cl_flags = 0;
-
-               if (cattr->la_nlink == 0) {
+               if (cattr->la_nlink == 0)

                        cl_flags |= CLF_UNLINK_LAST;
                        cl_flags |= CLF_UNLINK_LAST;

-                       /* search for an existing archive */
-                       if (mdd_hsm_archive_exists(env, mdd_cobj, ma))
-                               cl_flags |= CLF_UNLINK_HSM_EXISTS;
-               }
+               else
+                       cl_flags &= ~CLF_UNLINK_HSM_EXISTS;

 
                rc = mdd_changelog_ns_store(env, mdd,
                        is_dir ? CL_RMDIR : CL_UNLINK, cl_flags,
 
                rc = mdd_changelog_ns_store(env, mdd,
                        is_dir ? CL_RMDIR : CL_UNLINK, cl_flags,


@@ -2804,6 +2804,11 @@ static int mdd_rename(const struct lu_env *env,
                rc = mdd_la_get(env, mdd_tobj, tattr);
                if (rc)
                        GOTO(out_pending, rc);
                rc = mdd_la_get(env, mdd_tobj, tattr);
                if (rc)
                        GOTO(out_pending, rc);

+               /* search for an existing archive.
+                * we should check ahead as the object
+                * can be destroyed in this transaction */
+               if (mdd_hsm_archive_exists(env, mdd_tobj, ma))
+                       cl_flags |= CLF_RENAME_LAST_EXISTS;

        }
 
        rc = mdd_la_get(env, mdd_tpobj, tpattr);
        }
 
        rc = mdd_la_get(env, mdd_tpobj, tpattr);


@@ -2967,11 +2972,10 @@ static int mdd_rename(const struct lu_env *env,
                ma->ma_attr = *tattr;
                ma->ma_valid |= MA_INODE;
 
                ma->ma_attr = *tattr;
                ma->ma_valid |= MA_INODE;
 

-               if (tattr->la_nlink == 0) {
+               if (tattr->la_nlink == 0)

                        cl_flags |= CLF_RENAME_LAST;
                        cl_flags |= CLF_RENAME_LAST;

-                       if (mdd_hsm_archive_exists(env, mdd_tobj, ma))
-                               cl_flags |= CLF_RENAME_LAST_EXISTS;
-               }
+               else
+                       cl_flags &= ~CLF_RENAME_LAST_EXISTS;

         }
 
        la->la_valid = LA_CTIME | LA_MTIME;
         }
 
        la->la_valid = LA_CTIME | LA_MTIME;

diff --git a/lustre/osd-zfs/osd_index.c b/lustre/osd-zfs/osd_index.c
index 9780e43..705c86d 100644 (file)
--- a/lustre/osd-zfs/osd_index.c
+++ b/lustre/osd-zfs/osd_index.c
@@ -1614,18 +1614,21 @@ int osd_index_try(const struct lu_env *env, struct dt_object *dt,
                const struct dt_index_features *feat)
 {
        struct osd_object *obj = osd_dt_obj(dt);
                const struct dt_index_features *feat)
 {
        struct osd_object *obj = osd_dt_obj(dt);

+       int rc = 0;

        ENTRY;
 
        ENTRY;
 

+       down_read(&obj->oo_guard);
+

        /*
         * XXX: implement support for fixed-size keys sorted with natural
         *      numerical way (not using internal hash value)
         */
        if (feat->dif_flags & DT_IND_RANGE)
        /*
         * XXX: implement support for fixed-size keys sorted with natural
         *      numerical way (not using internal hash value)
         */
        if (feat->dif_flags & DT_IND_RANGE)

-               RETURN(-ERANGE);
+               GOTO(out, rc = -ERANGE);

 
        if (unlikely(feat == &dt_otable_features)) {
                dt->do_index_ops = &osd_zfs_otable_ops;
 
        if (unlikely(feat == &dt_otable_features)) {
                dt->do_index_ops = &osd_zfs_otable_ops;

-               RETURN(0);
+               GOTO(out, rc = 0);

        }
 
        LASSERT(!dt_object_exists(dt) || obj->oo_db != NULL);
        }
 
        LASSERT(!dt_object_exists(dt) || obj->oo_db != NULL);


@@ -1633,7 +1636,7 @@ int osd_index_try(const struct lu_env *env, struct dt_object *dt,
                if (!dt_object_exists(dt) || osd_object_is_zap(obj->oo_db))
                        dt->do_index_ops = &osd_dir_ops;
                else
                if (!dt_object_exists(dt) || osd_object_is_zap(obj->oo_db))
                        dt->do_index_ops = &osd_dir_ops;
                else

-                       RETURN(-ENOTDIR);
+                       GOTO(out, rc = -ENOTDIR);

        } else if (unlikely(feat == &dt_acct_features)) {
                LASSERT(fid_is_acct(lu_object_fid(&dt->do_lu)));
                dt->do_index_ops = &osd_acct_index_ops;
        } else if (unlikely(feat == &dt_acct_features)) {
                LASSERT(fid_is_acct(lu_object_fid(&dt->do_lu)));
                dt->do_index_ops = &osd_acct_index_ops;


@@ -1641,20 +1644,20 @@ int osd_index_try(const struct lu_env *env, struct dt_object *dt,
                /* For index file, we don't support variable key & record sizes
                 * and the key has to be unique */
                if ((feat->dif_flags & ~DT_IND_UPDATE) != 0)
                /* For index file, we don't support variable key & record sizes
                 * and the key has to be unique */
                if ((feat->dif_flags & ~DT_IND_UPDATE) != 0)

-                       RETURN(-EINVAL);
+                       GOTO(out, rc = -EINVAL);

 
                if (feat->dif_keysize_max > ZAP_MAXNAMELEN)
 
                if (feat->dif_keysize_max > ZAP_MAXNAMELEN)

-                       RETURN(-E2BIG);
+                       GOTO(out, rc = -E2BIG);

                if (feat->dif_keysize_max != feat->dif_keysize_min)
                if (feat->dif_keysize_max != feat->dif_keysize_min)

-                       RETURN(-EINVAL);
+                       GOTO(out, rc = -EINVAL);

 
                /* As for the record size, it should be a multiple of 8 bytes
                 * and smaller than the maximum value length supported by ZAP.
                 */
                if (feat->dif_recsize_max > ZAP_MAXVALUELEN)
 
                /* As for the record size, it should be a multiple of 8 bytes
                 * and smaller than the maximum value length supported by ZAP.
                 */
                if (feat->dif_recsize_max > ZAP_MAXVALUELEN)

-                       RETURN(-E2BIG);
+                       GOTO(out, rc = -E2BIG);

                if (feat->dif_recsize_max != feat->dif_recsize_min)
                if (feat->dif_recsize_max != feat->dif_recsize_min)

-                       RETURN(-EINVAL);
+                       GOTO(out, rc = -EINVAL);

 
                obj->oo_keysize = feat->dif_keysize_max;
                obj->oo_recsize = feat->dif_recsize_max;
 
                obj->oo_keysize = feat->dif_keysize_max;
                obj->oo_recsize = feat->dif_recsize_max;


@@ -1668,5 +1671,8 @@ int osd_index_try(const struct lu_env *env, struct dt_object *dt,
                dt->do_index_ops = &osd_index_ops;
        }
 
                dt->do_index_ops = &osd_index_ops;
        }
 

-       RETURN(0);
+out:
+       up_read(&obj->oo_guard);
+
+       RETURN(rc);

 }
 }

diff --git a/lustre/osd-zfs/osd_internal.h b/lustre/osd-zfs/osd_internal.h
index 527eb79..fa0c3f2 100644 (file)
--- a/lustre/osd-zfs/osd_internal.h
+++ b/lustre/osd-zfs/osd_internal.h
@@ -318,16 +318,22 @@ struct osd_object {
        sa_handle_t             *oo_sa_hdl;
        nvlist_t                *oo_sa_xattr;
        struct list_head         oo_sa_linkage;
        sa_handle_t             *oo_sa_hdl;
        nvlist_t                *oo_sa_xattr;
        struct list_head         oo_sa_linkage;

-       struct list_head         oo_unlinked_linkage;

 
 

+       /* used to implement osd_object_*_{lock|unlock} */

        struct rw_semaphore      oo_sem;
 
        struct rw_semaphore      oo_sem;
 

+       /* to serialize some updates: destroy vs. others,
+        * xattr_set, etc */
+       struct rw_semaphore      oo_guard;
+
+       /* protected by oo_guard */
+       struct list_head         oo_unlinked_linkage;
+

        /* cached attributes */
        rwlock_t                 oo_attr_lock;
        struct lu_attr           oo_attr;
 
        /* cached attributes */
        rwlock_t                 oo_attr_lock;
        struct lu_attr           oo_attr;
 

-       /* protects extended attributes and oo_unlinked_linkage */
-       struct semaphore         oo_guard;
+       /* external dnode holding large EAs, protected by oo_guard */

        uint64_t                 oo_xattr;
        enum osd_destroy_type    oo_destroy;
 
        uint64_t                 oo_xattr;
        enum osd_destroy_type    oo_destroy;
 


@@ -520,6 +526,10 @@ osd_xattr_set_internal(const struct lu_env *env, struct osd_object *obj,
 {
        int rc;
 
 {
        int rc;
 

+       if (unlikely(!dt_object_exists(&obj->oo_dt) || obj->oo_destroyed))
+               return -ENOENT;
+
+       LASSERT(obj->oo_db);

        if (osd_obj2dev(obj)->od_xattr_in_sa) {
                rc = __osd_sa_xattr_set(env, obj, buf, name, fl, oh);
                if (rc == -EFBIG)
        if (osd_obj2dev(obj)->od_xattr_in_sa) {
                rc = __osd_sa_xattr_set(env, obj, buf, name, fl, oh);
                if (rc == -EFBIG)

diff --git a/lustre/osd-zfs/osd_object.c b/lustre/osd-zfs/osd_object.c
index 7935fab..b65a282 100644 (file)
--- a/lustre/osd-zfs/osd_object.c
+++ b/lustre/osd-zfs/osd_object.c
@@ -315,7 +315,7 @@ struct lu_object *osd_object_alloc(const struct lu_env *env,
                INIT_LIST_HEAD(&mo->oo_sa_linkage);
                INIT_LIST_HEAD(&mo->oo_unlinked_linkage);
                init_rwsem(&mo->oo_sem);
                INIT_LIST_HEAD(&mo->oo_sa_linkage);
                INIT_LIST_HEAD(&mo->oo_unlinked_linkage);
                init_rwsem(&mo->oo_sem);

-               sema_init(&mo->oo_guard, 1);
+               init_rwsem(&mo->oo_guard);

                rwlock_init(&mo->oo_attr_lock);
                mo->oo_destroy = OSD_DESTROY_NONE;
                return l;
                rwlock_init(&mo->oo_attr_lock);
                mo->oo_destroy = OSD_DESTROY_NONE;
                return l;


@@ -466,17 +466,16 @@ osd_object_unlinked_add(struct osd_object *obj, struct osd_thandle *oh)
 {
        int rc = -EBUSY;
 
 {
        int rc = -EBUSY;
 

-       down(&obj->oo_guard);
-

        LASSERT(obj->oo_destroy == OSD_DESTROY_ASYNC);
 
        LASSERT(obj->oo_destroy == OSD_DESTROY_ASYNC);
 

+       /* the object is supposed to be exclusively locked by
+        * the caller (osd_object_destroy()), while the transaction
+        * (oh) is per-thread and not shared */

        if (likely(list_empty(&obj->oo_unlinked_linkage))) {
                list_add(&obj->oo_unlinked_linkage, &oh->ot_unlinked_list);
                rc = 0;
        }
 
        if (likely(list_empty(&obj->oo_unlinked_linkage))) {
                list_add(&obj->oo_unlinked_linkage, &oh->ot_unlinked_list);
                rc = 0;
        }
 

-       up(&obj->oo_guard);
-

        return rc;
 }
 
        return rc;
 }
 


@@ -493,14 +492,14 @@ osd_object_set_destroy_type(struct osd_object *obj)
         * Lock-less OST_WRITE can race with OST_DESTROY, so set destroy type
         * only once and use it consistently thereafter.
         */
         * Lock-less OST_WRITE can race with OST_DESTROY, so set destroy type
         * only once and use it consistently thereafter.
         */

-       down(&obj->oo_guard);
+       down_write(&obj->oo_guard);

        if (obj->oo_destroy == OSD_DESTROY_NONE) {
                if (obj->oo_attr.la_size <= osd_sync_destroy_max_size)
                        obj->oo_destroy = OSD_DESTROY_SYNC;
                else /* Larger objects are destroyed asynchronously */
                        obj->oo_destroy = OSD_DESTROY_ASYNC;
        }
        if (obj->oo_destroy == OSD_DESTROY_NONE) {
                if (obj->oo_attr.la_size <= osd_sync_destroy_max_size)
                        obj->oo_destroy = OSD_DESTROY_SYNC;
                else /* Larger objects are destroyed asynchronously */
                        obj->oo_destroy = OSD_DESTROY_ASYNC;
        }

-       up(&obj->oo_guard);
+       up_write(&obj->oo_guard);

 }
 
 static int osd_declare_object_destroy(const struct lu_env *env,
 }
 
 static int osd_declare_object_destroy(const struct lu_env *env,


@@ -569,9 +568,12 @@ static int osd_object_destroy(const struct lu_env *env,
        uint64_t                 oid, zapid;
        ENTRY;
 
        uint64_t                 oid, zapid;
        ENTRY;
 

+       down_write(&obj->oo_guard);
+
+       if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
+               GOTO(out, rc = -ENOENT);
+

        LASSERT(obj->oo_db != NULL);
        LASSERT(obj->oo_db != NULL);

-       LASSERT(dt_object_exists(dt));
-       LASSERT(!lu_object_is_dying(dt->do_lu.lo_header));

 
        oh = container_of0(th, struct osd_thandle, ot_super);
        LASSERT(oh != NULL);
 
        oh = container_of0(th, struct osd_thandle, ot_super);
        LASSERT(oh != NULL);


@@ -642,6 +644,7 @@ out:
        set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
        if (rc == 0)
                obj->oo_destroyed = 1;
        set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
        if (rc == 0)
                obj->oo_destroyed = 1;

+       up_write(&obj->oo_guard);

        RETURN (0);
 }
 
        RETURN (0);
 }
 


@@ -740,11 +743,12 @@ static int osd_attr_get(const struct lu_env *env,
        struct osd_object       *obj = osd_dt_obj(dt);
        uint64_t                 blocks;
        uint32_t                 blksize;
        struct osd_object       *obj = osd_dt_obj(dt);
        uint64_t                 blocks;
        uint32_t                 blksize;

+       int                      rc = 0;

 
 

-       if (unlikely(!dt_object_exists(dt)))
-               return -ENOENT;
-       if (unlikely(obj->oo_destroyed))
-               return -ENOENT;
+       down_read(&obj->oo_guard);
+
+       if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
+               GOTO(out, rc = -ENOENT);

 
        LASSERT(osd_invariant(obj));
        LASSERT(obj->oo_db);
 
        LASSERT(osd_invariant(obj));
        LASSERT(obj->oo_db);


@@ -772,7 +776,9 @@ static int osd_attr_get(const struct lu_env *env,
        attr->la_blocks = blocks;
        attr->la_valid |= LA_BLOCKS | LA_BLKSIZE;
 
        attr->la_blocks = blocks;
        attr->la_valid |= LA_BLOCKS | LA_BLKSIZE;
 

-       return 0;
+out:
+       up_read(&obj->oo_guard);
+       return rc;

 }
 
 /* Simple wrapper on top of qsd API which implement quota transfer for osd
 }
 
 /* Simple wrapper on top of qsd API which implement quota transfer for osd


@@ -848,24 +854,24 @@ static int osd_declare_attr_set(const struct lu_env *env,
        struct osd_thandle      *oh;
        uint64_t                 bspace;
        uint32_t                 blksize;
        struct osd_thandle      *oh;
        uint64_t                 bspace;
        uint32_t                 blksize;

-       int                      rc;
+       int                      rc = 0;

        ENTRY;
 
        ENTRY;
 

-       if (!dt_object_exists(dt)) {
-               /* XXX: sanity check that object creation is declared */
-               RETURN(0);
-       }

 
        LASSERT(handle != NULL);
        LASSERT(osd_invariant(obj));
 
        oh = container_of0(handle, struct osd_thandle, ot_super);
 
 
        LASSERT(handle != NULL);
        LASSERT(osd_invariant(obj));
 
        oh = container_of0(handle, struct osd_thandle, ot_super);
 

+       down_read(&obj->oo_guard);
+       if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
+               GOTO(out, rc = 0);
+

        LASSERT(obj->oo_sa_hdl != NULL);
        LASSERT(oh->ot_tx != NULL);
        dmu_tx_hold_sa(oh->ot_tx, obj->oo_sa_hdl, 0);
        if (oh->ot_tx->tx_err != 0)
        LASSERT(obj->oo_sa_hdl != NULL);
        LASSERT(oh->ot_tx != NULL);
        dmu_tx_hold_sa(oh->ot_tx, obj->oo_sa_hdl, 0);
        if (oh->ot_tx->tx_err != 0)

-               RETURN(-oh->ot_tx->tx_err);
+               GOTO(out, rc = -oh->ot_tx->tx_err);

 
        sa_object_size(obj->oo_sa_hdl, &blksize, &bspace);
        bspace = toqb(bspace * blksize);
 
        sa_object_size(obj->oo_sa_hdl, &blksize, &bspace);
        bspace = toqb(bspace * blksize);


@@ -885,7 +891,7 @@ static int osd_declare_attr_set(const struct lu_env *env,
                                          obj->oo_attr.la_uid, attr->la_uid,
                                          bspace, &info->oti_qi);
                        if (rc)
                                          obj->oo_attr.la_uid, attr->la_uid,
                                          bspace, &info->oti_qi);
                        if (rc)

-                               RETURN(rc);
+                               GOTO(out, rc);

                }
        }
        if (attr && attr->la_valid & LA_GID) {
                }
        }
        if (attr && attr->la_valid & LA_GID) {


@@ -900,11 +906,13 @@ static int osd_declare_attr_set(const struct lu_env *env,
                                          obj->oo_attr.la_gid, attr->la_gid,
                                          bspace, &info->oti_qi);
                        if (rc)
                                          obj->oo_attr.la_gid, attr->la_gid,
                                          bspace, &info->oti_qi);
                        if (rc)

-                               RETURN(rc);
+                               GOTO(out, rc);

                }
        }
 
                }
        }
 

-       RETURN(0);
+out:
+       up_read(&obj->oo_guard);
+       RETURN(rc);

 }
 
 /*
 }
 
 /*


@@ -928,8 +936,12 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
        int                      rc = 0;
 
        ENTRY;
        int                      rc = 0;
 
        ENTRY;

+
+       down_read(&obj->oo_guard);
+       if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
+               GOTO(out, rc = -ENOENT);
+

        LASSERT(handle != NULL);
        LASSERT(handle != NULL);

-       LASSERT(dt_object_exists(dt));

        LASSERT(osd_invariant(obj));
        LASSERT(obj->oo_sa_hdl);
 
        LASSERT(osd_invariant(obj));
        LASSERT(obj->oo_sa_hdl);
 


@@ -943,7 +955,7 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
                valid &= ~(LA_SIZE | LA_BLOCKS);
 
        if (valid == 0)
                valid &= ~(LA_SIZE | LA_BLOCKS);
 
        if (valid == 0)

-               RETURN(0);
+               GOTO(out, rc = 0);

 
        if (valid & LA_FLAGS) {
                struct lustre_mdt_attrs *lma;
 
        if (valid & LA_FLAGS) {
                struct lustre_mdt_attrs *lma;


@@ -980,7 +992,7 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
 
        OBD_ALLOC(bulk, sizeof(sa_bulk_attr_t) * 10);
        if (bulk == NULL)
 
        OBD_ALLOC(bulk, sizeof(sa_bulk_attr_t) * 10);
        if (bulk == NULL)

-               RETURN(-ENOMEM);
+               GOTO(out, rc = -ENOMEM);

 
        /* do both accounting updates outside oo_attr_lock below */
        if ((valid & LA_UID) && (la->la_uid != obj->oo_attr.la_uid)) {
 
        /* do both accounting updates outside oo_attr_lock below */
        if ((valid & LA_UID) && (la->la_uid != obj->oo_attr.la_uid)) {


@@ -1078,6 +1090,8 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
        rc = osd_object_sa_bulk_update(obj, bulk, cnt, oh);
 
        OBD_FREE(bulk, sizeof(sa_bulk_attr_t) * 10);
        rc = osd_object_sa_bulk_update(obj, bulk, cnt, oh);
 
        OBD_FREE(bulk, sizeof(sa_bulk_attr_t) * 10);

+out:
+       up_read(&obj->oo_guard);

        RETURN(rc);
 }
 
        RETURN(rc);
 }
 


@@ -1503,10 +1517,12 @@ static int osd_object_create(const struct lu_env *env, struct dt_object *dt,
        /* concurrent create declarations should not see
         * the object inconsistent (db, attr, etc).
         * in regular cases acquisition should be cheap */
        /* concurrent create declarations should not see
         * the object inconsistent (db, attr, etc).
         * in regular cases acquisition should be cheap */

-       down(&obj->oo_guard);
+       down_write(&obj->oo_guard);
+
+       if (unlikely(dt_object_exists(dt)))
+               GOTO(out, rc = -EEXIST);

 
        LASSERT(osd_invariant(obj));
 
        LASSERT(osd_invariant(obj));

-       LASSERT(!dt_object_exists(dt));

        LASSERT(dof != NULL);
 
        LASSERT(th != NULL);
        LASSERT(dof != NULL);
 
        LASSERT(th != NULL);


@@ -1569,7 +1585,7 @@ static int osd_object_create(const struct lu_env *env, struct dt_object *dt,
        }
 
 out:
        }
 
 out:

-       up(&obj->oo_guard);
+       up_write(&obj->oo_guard);

        RETURN(rc);
 }
 
        RETURN(rc);
 }
 


@@ -1595,8 +1611,9 @@ static int osd_object_ref_add(const struct lu_env *env,
 
        ENTRY;
 
 
        ENTRY;
 

-       if (!dt_object_exists(dt))
-               RETURN(-ENOENT);
+       down_read(&obj->oo_guard);
+       if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
+               GOTO(out, rc = -ENOENT);

 
        LASSERT(osd_invariant(obj));
        LASSERT(obj->oo_sa_hdl != NULL);
 
        LASSERT(osd_invariant(obj));
        LASSERT(obj->oo_sa_hdl != NULL);


@@ -1608,7 +1625,10 @@ static int osd_object_ref_add(const struct lu_env *env,
        write_unlock(&obj->oo_attr_lock);
 
        rc = osd_object_sa_update(obj, SA_ZPL_LINKS(osd), &nlink, 8, oh);
        write_unlock(&obj->oo_attr_lock);
 
        rc = osd_object_sa_update(obj, SA_ZPL_LINKS(osd), &nlink, 8, oh);

-       return rc;
+
+out:
+       up_read(&obj->oo_guard);
+       RETURN(rc);

 }
 
 static int osd_declare_object_ref_del(const struct lu_env *env,
 }
 
 static int osd_declare_object_ref_del(const struct lu_env *env,


@@ -1633,8 +1653,12 @@ static int osd_object_ref_del(const struct lu_env *env,
 
        ENTRY;
 
 
        ENTRY;
 

+       down_read(&obj->oo_guard);
+
+       if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
+               GOTO(out, rc = -ENOENT);
+

        LASSERT(osd_invariant(obj));
        LASSERT(osd_invariant(obj));

-       LASSERT(dt_object_exists(dt));

        LASSERT(obj->oo_sa_hdl != NULL);
 
        oh = container_of0(handle, struct osd_thandle, ot_super);
        LASSERT(obj->oo_sa_hdl != NULL);
 
        oh = container_of0(handle, struct osd_thandle, ot_super);


@@ -1645,6 +1669,9 @@ static int osd_object_ref_del(const struct lu_env *env,
        write_unlock(&obj->oo_attr_lock);
 
        rc = osd_object_sa_update(obj, SA_ZPL_LINKS(osd), &nlink, 8, oh);
        write_unlock(&obj->oo_attr_lock);
 
        rc = osd_object_sa_update(obj, SA_ZPL_LINKS(osd), &nlink, 8, oh);

+
+out:
+       up_read(&obj->oo_guard);

        RETURN(rc);
 }
 
        RETURN(rc);
 }
 

diff --git a/lustre/osd-zfs/osd_xattr.c b/lustre/osd-zfs/osd_xattr.c
index a4458cd..1be8363 100644 (file)
--- a/lustre/osd-zfs/osd_xattr.c
+++ b/lustre/osd-zfs/osd_xattr.c
@@ -229,6 +229,9 @@ int __osd_xattr_get(const struct lu_env *env, struct osd_object *obj,
 {
        int rc;
 
 {
        int rc;
 

+       if (unlikely(!dt_object_exists(&obj->oo_dt) || obj->oo_destroyed))
+               return -ENOENT;
+

        /* check SA_ZPL_DXATTR first then fallback to directory xattr */
        rc = __osd_sa_xattr_get(env, obj, buf, name, sizep);
        if (rc != -ENOENT)
        /* check SA_ZPL_DXATTR first then fallback to directory xattr */
        rc = __osd_sa_xattr_get(env, obj, buf, name, sizep);
        if (rc != -ENOENT)


@@ -247,16 +250,15 @@ int osd_xattr_get(const struct lu_env *env, struct dt_object *dt,
 
        LASSERT(obj->oo_db != NULL);
        LASSERT(osd_invariant(obj));
 
        LASSERT(obj->oo_db != NULL);
        LASSERT(osd_invariant(obj));

-       LASSERT(dt_object_exists(dt));

 
        if (!osd_obj2dev(obj)->od_posix_acl &&
            (strcmp(name, POSIX_ACL_XATTR_ACCESS) == 0 ||
             strcmp(name, POSIX_ACL_XATTR_DEFAULT) == 0))
                RETURN(-EOPNOTSUPP);
 
 
        if (!osd_obj2dev(obj)->od_posix_acl &&
            (strcmp(name, POSIX_ACL_XATTR_ACCESS) == 0 ||
             strcmp(name, POSIX_ACL_XATTR_DEFAULT) == 0))
                RETURN(-EOPNOTSUPP);
 

-       down(&obj->oo_guard);
+       down_read(&obj->oo_guard);

        rc = __osd_xattr_get(env, obj, buf, name, &size);
        rc = __osd_xattr_get(env, obj, buf, name, &size);

-       up(&obj->oo_guard);
+       up_read(&obj->oo_guard);

 
        if (rc == -ENOENT)
                rc = -ENODATA;
 
        if (rc == -ENOENT)
                rc = -ENODATA;


@@ -276,6 +278,9 @@ void __osd_xattr_declare_set(const struct lu_env *env, struct osd_object *obj,
        int                rc = 0;
        int                here;
 
        int                rc = 0;
        int                here;
 

+       if (unlikely(obj->oo_destroyed))
+               return;
+

        here = dt_object_exists(&obj->oo_dt);
 
        /* object may be not yet created */
        here = dt_object_exists(&obj->oo_dt);
 
        /* object may be not yet created */


@@ -341,9 +346,9 @@ int osd_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
        LASSERT(handle != NULL);
        oh = container_of0(handle, struct osd_thandle, ot_super);
 
        LASSERT(handle != NULL);
        oh = container_of0(handle, struct osd_thandle, ot_super);
 

-       down(&obj->oo_guard);
+       down_read(&obj->oo_guard);

        __osd_xattr_declare_set(env, obj, buf->lb_len, name, oh);
        __osd_xattr_declare_set(env, obj, buf->lb_len, name, oh);

-       up(&obj->oo_guard);
+       up_read(&obj->oo_guard);

 
        RETURN(0);
 }
 
        RETURN(0);
 }


@@ -593,8 +598,6 @@ int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
 
        LASSERT(handle != NULL);
        LASSERT(osd_invariant(obj));
 
        LASSERT(handle != NULL);
        LASSERT(osd_invariant(obj));

-       LASSERT(dt_object_exists(dt));
-       LASSERT(obj->oo_db);

 
        if (!osd_obj2dev(obj)->od_posix_acl &&
            (strcmp(name, POSIX_ACL_XATTR_ACCESS) == 0 ||
 
        if (!osd_obj2dev(obj)->od_posix_acl &&
            (strcmp(name, POSIX_ACL_XATTR_ACCESS) == 0 ||


@@ -607,11 +610,11 @@ int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
 
        oh = container_of0(handle, struct osd_thandle, ot_super);
 
 
        oh = container_of0(handle, struct osd_thandle, ot_super);
 

-       down(&obj->oo_guard);
+       down_write(&obj->oo_guard);

        CDEBUG(D_INODE, "Setting xattr %s with size %d\n",
                name, (int)buf->lb_len);
        rc = osd_xattr_set_internal(env, obj, buf, name, fl, oh);
        CDEBUG(D_INODE, "Setting xattr %s with size %d\n",
                name, (int)buf->lb_len);
        rc = osd_xattr_set_internal(env, obj, buf, name, fl, oh);

-       up(&obj->oo_guard);
+       up_write(&obj->oo_guard);

 
        RETURN(rc);
 }
 
        RETURN(rc);
 }


@@ -660,22 +663,22 @@ int osd_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
        ENTRY;
 
        LASSERT(handle != NULL);
        ENTRY;
 
        LASSERT(handle != NULL);

-       LASSERT(dt_object_exists(dt));

        LASSERT(osd_invariant(obj));
 
        oh = container_of0(handle, struct osd_thandle, ot_super);
        LASSERT(oh->ot_tx != NULL);
        LASSERT(obj->oo_db != NULL);
 
        LASSERT(osd_invariant(obj));
 
        oh = container_of0(handle, struct osd_thandle, ot_super);
        LASSERT(oh->ot_tx != NULL);
        LASSERT(obj->oo_db != NULL);
 

-       down(&obj->oo_guard);
-       __osd_xattr_declare_del(env, obj, name, oh);
-       up(&obj->oo_guard);
+       down_read(&obj->oo_guard);
+       if (likely(dt_object_exists(&obj->oo_dt) && !obj->oo_destroyed))
+               __osd_xattr_declare_del(env, obj, name, oh);
+       up_read(&obj->oo_guard);

 
        RETURN(0);
 }
 
 
        RETURN(0);
 }
 

-int __osd_sa_xattr_del(const struct lu_env *env, struct osd_object *obj,
-                      const char *name, struct osd_thandle *oh)
+static int __osd_sa_xattr_del(const struct lu_env *env, struct osd_object *obj,
+                             const char *name, struct osd_thandle *oh)

 {
        int rc;
 
 {
        int rc;
 


@@ -691,13 +694,16 @@ int __osd_sa_xattr_del(const struct lu_env *env, struct osd_object *obj,
        return rc;
 }
 
        return rc;
 }
 

-int __osd_xattr_del(const struct lu_env *env, struct osd_object *obj,
-                   const char *name, struct osd_thandle *oh)
+static int __osd_xattr_del(const struct lu_env *env, struct osd_object *obj,
+                          const char *name, struct osd_thandle *oh)

 {
        struct osd_device *osd = osd_obj2dev(obj);
        uint64_t           xa_data_obj;
        int                rc;
 
 {
        struct osd_device *osd = osd_obj2dev(obj);
        uint64_t           xa_data_obj;
        int                rc;
 

+       if (unlikely(!dt_object_exists(&obj->oo_dt) || obj->oo_destroyed))
+               return -ENOENT;
+

        /* try remove xattr from SA at first */
        rc = __osd_sa_xattr_del(env, obj, name, oh);
        if (rc != -ENOENT)
        /* try remove xattr from SA at first */
        rc = __osd_sa_xattr_del(env, obj, name, oh);
        if (rc != -ENOENT)


@@ -745,9 +751,9 @@ int osd_xattr_del(const struct lu_env *env, struct dt_object *dt,
             strcmp(name, POSIX_ACL_XATTR_DEFAULT) == 0))
                RETURN(-EOPNOTSUPP);
 
             strcmp(name, POSIX_ACL_XATTR_DEFAULT) == 0))
                RETURN(-EOPNOTSUPP);
 

-       down(&obj->oo_guard);
+       down_write(&obj->oo_guard);

        rc = __osd_xattr_del(env, obj, name, oh);
        rc = __osd_xattr_del(env, obj, name, oh);

-       up(&obj->oo_guard);
+       up_write(&obj->oo_guard);

 
        RETURN(rc);
 }
 
        RETURN(rc);
 }


@@ -892,7 +898,7 @@ int osd_xattr_list(const struct lu_env *env, struct dt_object *dt,
        LASSERT(osd_invariant(obj));
        LASSERT(dt_object_exists(dt));
 
        LASSERT(osd_invariant(obj));
        LASSERT(dt_object_exists(dt));
 

-       down(&obj->oo_guard);
+       down_read(&obj->oo_guard);

 
        rc = osd_sa_xattr_list(env, obj, lb);
        if (rc < 0)
 
        rc = osd_sa_xattr_list(env, obj, lb);
        if (rc < 0)


@@ -936,7 +942,7 @@ int osd_xattr_list(const struct lu_env *env, struct dt_object *dt,
 out_fini:
        osd_zap_cursor_fini(zc);
 out:
 out_fini:
        osd_zap_cursor_fini(zc);
 out:

-       up(&obj->oo_guard);
+       up_read(&obj->oo_guard);

        RETURN(rc);
 
 }
        RETURN(rc);
 
 }