From db1ef0a322f41314abd37b5ec4ad153d63c9b405 Mon Sep 17 00:00:00 2001 From: Jian Yu Date: Fri, 27 Jan 2017 19:36:02 +0800 Subject: [PATCH] LU-9045 osp: Revert "LU-8840 osp: handle EA cache properly" The patch caused test failures tracked in LU-9045 and LU-9048. This reverts commit 555d02f47401340182b47b3245a657b52fc3e68a. Test-Parameters: mdtfilesystemtype=ldiskfs ostfilesystemtype=ldiskfs \ mdscount=2 mdtcount=4 \ testlist=conf-sanity,conf-sanity,sanity-lfsck,sanity-lfsck,sanity-hsm,sanity-hsm Change-Id: I3d922abd76b441f10ed0446e5528644a38211949 Signed-off-by: Jian Yu Reviewed-on: https://review.whamcloud.com/25134 Tested-by: Jenkins Reviewed-by: Oleg Drokin Tested-by: Oleg Drokin --- lustre/lfsck/lfsck_namespace.c | 5 +- lustre/osp/osp_object.c | 263 ++++++++++++++++++++--------------------- lustre/target/out_handler.c | 17 ++- 3 files changed, 136 insertions(+), 149 deletions(-) diff --git a/lustre/lfsck/lfsck_namespace.c b/lustre/lfsck/lfsck_namespace.c index 88a9ab2..7140ef0 100644 --- a/lustre/lfsck/lfsck_namespace.c +++ b/lustre/lfsck/lfsck_namespace.c @@ -2688,8 +2688,7 @@ again: * When the LFSCK runs again, if the dangling name is still * there, the LFSCK should move the orphan directory object * back to the normal namespace. */ - if (!lpf && !fid_is_zero(pfid) && - !lu_fid_eq(pfid, &tfid) && once) { + if (!lpf && !lu_fid_eq(pfid, &tfid) && once) { linkea_next_entry(ldata); continue; } @@ -2740,7 +2739,7 @@ again: * directory contains the specified child, but such * parent does not match the dotdot name entry, then * trust the linkEA. */ - if (!fid_is_zero(pfid) && !lu_fid_eq(pfid, pfid2)) { + if (!lu_fid_eq(pfid, pfid2)) { *type = LNIT_UNMATCHED_PAIRS; rc = lfsck_namespace_repair_unmatched_pairs(env, com, child, pfid2, cname); diff --git a/lustre/osp/osp_object.c b/lustre/osp/osp_object.c index 0781eb9..1fcbac7 100644 --- a/lustre/osp/osp_object.c +++ b/lustre/osp/osp_object.c @@ -126,28 +126,11 @@ static inline __u32 osp_dev2node(struct osp_device *osp) return osp->opd_storage->dd_lu_dev.ld_site->ld_seq_site->ss_node_id; } -static inline const char *osp_dto2name(struct osp_object *obj) -{ - return obj->opo_obj.do_lu.lo_dev->ld_obd->obd_name; -} - static inline bool is_ost_obj(struct lu_object *lo) { return !lu2osp_dev(lo->lo_dev)->opd_connect_mdt; } -static inline void __osp_oac_xattr_assignment(struct osp_object *obj, - struct osp_xattr_entry *oxe, - const struct lu_buf *buf) -{ - if (buf->lb_len > 0) - memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len); - - oxe->oxe_vallen = buf->lb_len; - oxe->oxe_exist = 1; - oxe->oxe_ready = 1; -} - /** * Assign FID to the OST object. * @@ -172,24 +155,6 @@ static void osp_object_assign_fid(const struct lu_env *env, lu_object_assign_fid(env, &o->opo_obj.do_lu, &osi->osi_fid); } -#define OXE_DEFAULT_LEN 16 - -/** - * Release reference from the OSP object extended attribute entry. - * - * If it is the last reference, then free the entry. - * - * \param[in] oxe pointer to the OSP object extended attribute entry. - */ -static inline void osp_oac_xattr_put(struct osp_xattr_entry *oxe) -{ - if (atomic_dec_and_test(&oxe->oxe_ref)) { - LASSERT(list_empty(&oxe->oxe_list)); - - OBD_FREE(oxe, oxe->oxe_buflen); - } -} - /** * Find the named extended attribute in the OSP object attributes cache. * @@ -270,10 +235,9 @@ static struct osp_xattr_entry * osp_oac_xattr_find_or_add(struct osp_object *obj, const char *name, size_t len) { struct osp_xattr_entry *oxe; - struct osp_xattr_entry *tmp = NULL; - size_t namelen = strlen(name); - size_t size = sizeof(*oxe) + namelen + 1 + - (len ? len : OXE_DEFAULT_LEN); + struct osp_xattr_entry *tmp = NULL; + size_t namelen = strlen(name); + size_t size = sizeof(*oxe) + namelen + 1 + len; oxe = osp_oac_xattr_find(obj, name, false); if (oxe != NULL) @@ -308,87 +272,67 @@ osp_oac_xattr_find_or_add(struct osp_object *obj, const char *name, size_t len) } /** - * Assign the cached OST-object's EA with the given value. + * Add the given extended attribute to the OSP object attributes cache. * - * If the current EA entry in cache has not enough space to hold the new - * value, remove it, create a new one, then assign with the given value. + * If there is an old extended attributed entry with the same name, + * remove it from the cache and return it via the parameter \a poxe. * * \param[in] obj pointer to the OSP object - * \param[in] oxe pointer to the cached EA entry to be assigned - * \param[in] buf pointer to the buffer with new EA value - * - * \retval pointer to the new created EA entry in cache if - * current entry is not big enough; otherwise, the - * input 'oxe' will be returned. + * \param[in,out] poxe double pointer to the OSP object extended attribute + * entry: the new extended attribute entry is transferred + * via such pointer target, and if old the extended + * attribute entry exists, then it will be returned back + * via such pointer target. + * \param[in] len the length of the (new) extended attribute value + * + * \retval pointer to the new extended attribute entry + * \retval NULL for failure cases. */ static struct osp_xattr_entry * -osp_oac_xattr_assignment(struct osp_object *obj, struct osp_xattr_entry *oxe, - const struct lu_buf *buf) +osp_oac_xattr_replace(struct osp_object *obj, + struct osp_xattr_entry **poxe, size_t len) { - struct osp_xattr_entry *new = NULL; - struct osp_xattr_entry *old = NULL; - int namelen = oxe->oxe_namelen; - bool unlink_only = false; + struct osp_xattr_entry *oxe; + size_t namelen = (*poxe)->oxe_namelen; + size_t size = sizeof(*oxe) + namelen + 1 + len; - if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) { - size_t size = sizeof(*oxe) + namelen + 1 + buf->lb_len; - - OBD_ALLOC(new, size); - if (likely(new)) { - INIT_LIST_HEAD(&new->oxe_list); - new->oxe_buflen = size; - new->oxe_namelen = namelen; - memcpy(new->oxe_buf, oxe->oxe_buf, namelen); - new->oxe_value = new->oxe_buf + namelen + 1; - /* One ref is for the caller, - * the other is for the entry on the list. */ - atomic_set(&new->oxe_ref, 2); - __osp_oac_xattr_assignment(obj, new, buf); - } else { - unlink_only = true; - CWARN("%s: cannot update cached xattr %.*s of "DFID"\n", - osp_dto2name(obj), namelen, oxe->oxe_buf, - PFID(lu_object_fid(&obj->opo_obj.do_lu))); - } - } + OBD_ALLOC(oxe, size); + if (unlikely(oxe == NULL)) + return NULL; - spin_lock(&obj->opo_lock); - old = osp_oac_xattr_find_locked(obj, oxe->oxe_buf, namelen); - if (likely(old)) { - if (new) { - /* Unlink the 'old'. */ - list_del_init(&old->oxe_list); + INIT_LIST_HEAD(&oxe->oxe_list); + oxe->oxe_buflen = size; + oxe->oxe_namelen = namelen; + memcpy(oxe->oxe_buf, (*poxe)->oxe_buf, namelen); + oxe->oxe_value = oxe->oxe_buf + namelen + 1; + /* One ref is for the caller, the other is for the entry on the list. */ + atomic_set(&oxe->oxe_ref, 2); - /* Drop the ref for 'old' on list. */ - osp_oac_xattr_put(old); + spin_lock(&obj->opo_lock); + *poxe = osp_oac_xattr_find_locked(obj, oxe->oxe_buf, namelen); + LASSERT(*poxe != NULL); - /* Insert 'new' into list. */ - list_add_tail(&new->oxe_list, &obj->opo_xattr_list); + list_del_init(&(*poxe)->oxe_list); + list_add_tail(&oxe->oxe_list, &obj->opo_xattr_list); + spin_unlock(&obj->opo_lock); - /* Drop the ref for current using. */ - osp_oac_xattr_put(oxe); - oxe = new; - } else if (unlink_only) { - /* Unlink the 'old'. */ - list_del_init(&old->oxe_list); + return oxe; +} - /* Drop the ref for 'old' on list. */ - osp_oac_xattr_put(old); - } else { - __osp_oac_xattr_assignment(obj, oxe, buf); - } - } else if (new) { - /* Someone unlinked the 'old' by race, need NOT to assign - * for unlinked 'oxe', just add the 'new' one. */ - list_add_tail(&new->oxe_list, &obj->opo_xattr_list); +/** + * Release reference from the OSP object extended attribute entry. + * + * If it is the last reference, then free the entry. + * + * \param[in] oxe pointer to the OSP object extended attribute entry. + */ +static inline void osp_oac_xattr_put(struct osp_xattr_entry *oxe) +{ + if (atomic_dec_and_test(&oxe->oxe_ref)) { + LASSERT(list_empty(&oxe->oxe_list)); - /* Drop the ref for current using. */ - osp_oac_xattr_put(oxe); - oxe = new; + OBD_FREE(oxe, oxe->oxe_buflen); } - spin_unlock(&obj->opo_lock); - - return oxe; } /** @@ -768,18 +712,13 @@ static int osp_xattr_get_interpterer(const struct lu_env *env, struct osp_xattr_entry *oxe = data; struct lu_buf *rbuf = &osp_env_info(env)->osi_lb2; - if (!rc) { + if (rc == 0) { size_t len = sizeof(*oxe) + oxe->oxe_namelen + 1; rc = object_update_result_data_get(reply, rbuf, index); if (rc < 0 || rbuf->lb_len > (oxe->oxe_buflen - len)) { spin_lock(&obj->opo_lock); - if (unlikely(rc == -ENODATA)) { - oxe->oxe_exist = 0; - oxe->oxe_ready = 1; - } else { - oxe->oxe_ready = 0; - } + oxe->oxe_ready = 0; spin_unlock(&obj->opo_lock); osp_oac_xattr_put(oxe); @@ -787,7 +726,10 @@ static int osp_xattr_get_interpterer(const struct lu_env *env, } spin_lock(&obj->opo_lock); - __osp_oac_xattr_assignment(obj, oxe, rbuf); + oxe->oxe_vallen = rbuf->lb_len; + memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len); + oxe->oxe_exist = 1; + oxe->oxe_ready = 1; spin_unlock(&obj->opo_lock); } else if (rc == -ENOENT || rc == -ENODATA) { spin_lock(&obj->opo_lock); @@ -835,14 +777,16 @@ static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt, LASSERT(buf != NULL); LASSERT(name != NULL); + namelen = strlen(name); + + /* If only for xattr size, return directly. */ if (unlikely(buf->lb_len == 0)) - return -EINVAL; + return 0; oxe = osp_oac_xattr_find_or_add(obj, name, buf->lb_len); if (oxe == NULL) return -ENOMEM; - namelen = strlen(name); mutex_lock(&osp->opd_async_requests_mutex); rc = osp_insert_async_request(env, OUT_XATTR_GET, obj, 1, &namelen, (const void **)&name, @@ -909,8 +853,8 @@ int osp_xattr_get(const struct lu_env *env, struct dt_object *dt, struct ptlrpc_request *req = NULL; struct object_update_reply *reply; struct osp_xattr_entry *oxe = NULL; - const char *dname = osp_dto2name(obj); - int rc = 0; + const char *dname = dt->do_lu.lo_dev->ld_obd->obd_name; + int rc = 0; ENTRY; LASSERT(buf != NULL); @@ -1013,15 +957,15 @@ unlock: if (rc < 0) GOTO(out, rc); - /* The peer should have converted '0' as '-ENODATA'. */ - if (unlikely(!rc)) - GOTO(out, rc = -EINVAL); - - /* For detecting EA size. */ - if (!buf->lb_buf) + if (buf->lb_buf == NULL) GOTO(out, rc); - if (!oxe) { + if (unlikely(buf->lb_len < rbuf->lb_len)) + GOTO(out, rc = -ERANGE); + + memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len); + + if (oxe == NULL) { oxe = osp_oac_xattr_find_or_add(obj, name, rbuf->lb_len); if (oxe == NULL) { CWARN("%s: Fail to add xattr (%s) to " @@ -1032,18 +976,38 @@ unlock: } } - oxe = osp_oac_xattr_assignment(obj, oxe, rbuf); + if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < rbuf->lb_len) { + struct osp_xattr_entry *old = oxe; + struct osp_xattr_entry *tmp; - GOTO(out, rc); + tmp = osp_oac_xattr_replace(obj, &old, rbuf->lb_len); + osp_oac_xattr_put(oxe); + oxe = tmp; + if (tmp == NULL) { + CWARN("%s: Fail to update xattr (%s) to " + "cache for "DFID": rc = %d\n", + dname, name, PFID(lu_object_fid(&dt->do_lu)), rc); + spin_lock(&obj->opo_lock); + old->oxe_ready = 0; + spin_unlock(&obj->opo_lock); -out: - if (rc > 0 && buf->lb_buf) { - if (unlikely(buf->lb_len < rbuf->lb_len)) - rc = -ERANGE; - else - memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len); + GOTO(out, rc); + } + + /* Drop the ref for entry on list. */ + osp_oac_xattr_put(old); } + spin_lock(&obj->opo_lock); + oxe->oxe_vallen = rbuf->lb_len; + memcpy(oxe->oxe_value, rbuf->lb_buf, rbuf->lb_len); + oxe->oxe_exist = 1; + oxe->oxe_ready = 1; + spin_unlock(&obj->opo_lock); + + GOTO(out, rc); + +out: if (req != NULL) ptlrpc_req_finished(req); @@ -1141,14 +1105,41 @@ int osp_xattr_set(const struct lu_env *env, struct dt_object *dt, oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len); if (oxe == NULL) { CWARN("%s: cannot cache xattr '%s' of "DFID"\n", - osp_dto2name(o), name, PFID(lu_object_fid(&dt->do_lu))); + dt->do_lu.lo_dev->ld_obd->obd_name, + name, PFID(lu_object_fid(&dt->do_lu))); RETURN(0); } - oxe = osp_oac_xattr_assignment(o, oxe, buf); - if (oxe) + if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) { + struct osp_xattr_entry *old = oxe; + struct osp_xattr_entry *tmp; + + tmp = osp_oac_xattr_replace(o, &old, buf->lb_len); osp_oac_xattr_put(oxe); + oxe = tmp; + if (tmp == NULL) { + CWARN("%s: cannot update cached xattr '%s' of "DFID"\n", + dt->do_lu.lo_dev->ld_obd->obd_name, + name, PFID(lu_object_fid(&dt->do_lu))); + spin_lock(&o->opo_lock); + old->oxe_ready = 0; + spin_unlock(&o->opo_lock); + + RETURN(0); + } + + /* Drop the ref for entry on list. */ + osp_oac_xattr_put(old); + } + + spin_lock(&o->opo_lock); + oxe->oxe_vallen = buf->lb_len; + memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len); + oxe->oxe_exist = 1; + oxe->oxe_ready = 1; + spin_unlock(&o->opo_lock); + osp_oac_xattr_put(oxe); RETURN(0); } diff --git a/lustre/target/out_handler.c b/lustre/target/out_handler.c index 7781ff3..1cf8c21 100644 --- a/lustre/target/out_handler.c +++ b/lustre/target/out_handler.c @@ -277,22 +277,19 @@ static int out_xattr_get(struct tgt_session_info *tsi) lbuf->lb_len = (int)tti->tti_u.update.tti_update->ou_result_size; lbuf->lb_buf = update_result->our_data; if (lbuf->lb_len == 0) - lbuf->lb_buf = NULL; + lbuf->lb_buf = 0; dt_read_lock(env, obj, MOR_TGT_CHILD); rc = dt_xattr_get(env, obj, lbuf, name); dt_read_unlock(env, obj); - if (rc <= 0) { + if (rc < 0) lbuf->lb_len = 0; - if (unlikely(!rc)) - rc = -ENODATA; - } else if (lbuf->lb_buf) { - lbuf->lb_len = rc; - } - - CDEBUG(D_INFO, "%s: "DFID" get xattr %s len %d: rc = %d\n", + CDEBUG(D_INFO, "%s: "DFID" get xattr %s len %d\n", tgt_name(tsi->tsi_tgt), PFID(lu_object_fid(&obj->do_lu)), - name, (int)lbuf->lb_len, rc); + name, (int)lbuf->lb_len); + GOTO(out, rc); + +out: object_update_result_insert(reply, lbuf->lb_buf, lbuf->lb_len, idx, rc); RETURN(0); } -- 1.8.3.1