From f9ca359284357d145819beb08b316e932f7a3060 Mon Sep 17 00:00:00 2001 From: Alex Zhuravlev Date: Tue, 26 Jan 2016 18:47:29 +0300 Subject: [PATCH] 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 Reviewed-on: http://review.whamcloud.com/18155 Tested-by: Jenkins Reviewed-by: John L. Hammond Reviewed-by: Nathaniel Clark Reviewed-by: Andreas Dilger Tested-by: Maloo Reviewed-by: Oleg Drokin --- lustre/mdd/mdd_dir.c | 34 +++++++++------- lustre/osd-zfs/osd_index.c | 24 ++++++----- lustre/osd-zfs/osd_internal.h | 16 ++++++-- lustre/osd-zfs/osd_object.c | 93 ++++++++++++++++++++++++++++--------------- lustre/osd-zfs/osd_xattr.c | 48 ++++++++++++---------- 5 files changed, 134 insertions(+), 81 deletions(-) diff --git a/lustre/mdd/mdd_dir.c b/lustre/mdd/mdd_dir.c index 22c3f2b..eab1629 100644 --- 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); - ma->ma_valid = MA_HSM; + ma->ma_valid |= MA_HSM; } 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; - int rc, is_dir = 0; + int rc, is_dir = 0, cl_flags = 0; 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); + /* 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); @@ -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); - - /* fetch updated nlink */ 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 */ @@ -1743,14 +1747,10 @@ cleanup: 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; - /* 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, @@ -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); + /* 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); @@ -2967,11 +2972,10 @@ static int mdd_rename(const struct lu_env *env, ma->ma_attr = *tattr; ma->ma_valid |= MA_INODE; - if (tattr->la_nlink == 0) { + if (tattr->la_nlink == 0) 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; diff --git a/lustre/osd-zfs/osd_index.c b/lustre/osd-zfs/osd_index.c index 9780e43..705c86d 100644 --- 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); + int rc = 0; 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) - RETURN(-ERANGE); + GOTO(out, rc = -ERANGE); 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); @@ -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 - 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; @@ -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) - RETURN(-EINVAL); + GOTO(out, rc = -EINVAL); if (feat->dif_keysize_max > ZAP_MAXNAMELEN) - RETURN(-E2BIG); + GOTO(out, rc = -E2BIG); 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) - RETURN(-E2BIG); + GOTO(out, rc = -E2BIG); 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; @@ -1668,5 +1671,8 @@ int osd_index_try(const struct lu_env *env, struct dt_object *dt, 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 --- 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; - struct list_head oo_unlinked_linkage; + /* used to implement osd_object_*_{lock|unlock} */ 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; - /* 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; @@ -520,6 +526,10 @@ osd_xattr_set_internal(const struct lu_env *env, struct osd_object *obj, { 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) diff --git a/lustre/osd-zfs/osd_object.c b/lustre/osd-zfs/osd_object.c index 7935fab..b65a282 100644 --- 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); - 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; @@ -466,17 +466,16 @@ osd_object_unlinked_add(struct osd_object *obj, struct osd_thandle *oh) { int rc = -EBUSY; - down(&obj->oo_guard); - 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; } - up(&obj->oo_guard); - 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. */ - 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; } - up(&obj->oo_guard); + up_write(&obj->oo_guard); } 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; + down_write(&obj->oo_guard); + + if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed)) + GOTO(out, rc = -ENOENT); + 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); @@ -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; + up_write(&obj->oo_guard); 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; + 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); @@ -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; - return 0; +out: + up_read(&obj->oo_guard); + return rc; } /* 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; - int rc; + int rc = 0; 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); + 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) - 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); @@ -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) - RETURN(rc); + GOTO(out, rc); } } 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) - 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; + + down_read(&obj->oo_guard); + if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed)) + GOTO(out, rc = -ENOENT); + LASSERT(handle != NULL); - LASSERT(dt_object_exists(dt)); 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) - RETURN(0); + GOTO(out, rc = 0); 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) - 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)) { @@ -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); +out: + up_read(&obj->oo_guard); 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 */ - down(&obj->oo_guard); + down_write(&obj->oo_guard); + + if (unlikely(dt_object_exists(dt))) + GOTO(out, rc = -EEXIST); LASSERT(osd_invariant(obj)); - LASSERT(!dt_object_exists(dt)); 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: - up(&obj->oo_guard); + up_write(&obj->oo_guard); RETURN(rc); } @@ -1595,8 +1611,9 @@ static int osd_object_ref_add(const struct lu_env *env, 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); @@ -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); - return rc; + +out: + up_read(&obj->oo_guard); + RETURN(rc); } 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; + down_read(&obj->oo_guard); + + if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed)) + GOTO(out, rc = -ENOENT); + LASSERT(osd_invariant(obj)); - LASSERT(dt_object_exists(dt)); 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); + +out: + up_read(&obj->oo_guard); RETURN(rc); } diff --git a/lustre/osd-zfs/osd_xattr.c b/lustre/osd-zfs/osd_xattr.c index a4458cd..1be8363 100644 --- 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; + 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) @@ -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(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); - down(&obj->oo_guard); + down_read(&obj->oo_guard); rc = __osd_xattr_get(env, obj, buf, name, &size); - up(&obj->oo_guard); + up_read(&obj->oo_guard); 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; + if (unlikely(obj->oo_destroyed)) + return; + 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); - down(&obj->oo_guard); + down_read(&obj->oo_guard); __osd_xattr_declare_set(env, obj, buf->lb_len, name, oh); - up(&obj->oo_guard); + up_read(&obj->oo_guard); 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(dt_object_exists(dt)); - LASSERT(obj->oo_db); 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); - 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); - up(&obj->oo_guard); + up_write(&obj->oo_guard); RETURN(rc); } @@ -660,22 +663,22 @@ int osd_declare_xattr_del(const struct lu_env *env, struct dt_object *dt, 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); - 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); } -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; @@ -691,13 +694,16 @@ int __osd_sa_xattr_del(const struct lu_env *env, struct osd_object *obj, 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; + 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) @@ -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); - down(&obj->oo_guard); + down_write(&obj->oo_guard); rc = __osd_xattr_del(env, obj, name, oh); - up(&obj->oo_guard); + up_write(&obj->oo_guard); 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)); - down(&obj->oo_guard); + down_read(&obj->oo_guard); 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: - up(&obj->oo_guard); + up_read(&obj->oo_guard); RETURN(rc); } -- 1.8.3.1