X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fosd-zfs%2Fosd_xattr.c;h=aaa94abb6f926d0eb27c0be7c8412fafcff1eaf4;hb=refs%2Fchanges%2F93%2F27093%2F40;hp=0ea99773bf0e092257d8d6cccbd33b41f5068c98;hpb=ead6df2feee9c143b617cb60e50e403c955bd401;p=fs%2Flustre-release.git diff --git a/lustre/osd-zfs/osd_xattr.c b/lustre/osd-zfs/osd_xattr.c index 0ea9977..aaa94ab 100644 --- a/lustre/osd-zfs/osd_xattr.c +++ b/lustre/osd-zfs/osd_xattr.c @@ -15,11 +15,7 @@ * * You should have received a copy of the GNU General Public License * version 2 along with this program; If not, see - * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf - * - * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara, - * CA 95054 USA or visit www.sun.com if you need additional information or - * have any questions. + * http://www.gnu.org/licenses/gpl-2.0.html * * GPL HEADER END */ @@ -27,7 +23,7 @@ * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2012, 2015, Intel Corporation. + * Copyright (c) 2012, 2016, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -42,7 +38,6 @@ #define DEBUG_SUBSYSTEM S_OSD -#include #include #include #include @@ -69,20 +64,12 @@ #include -int __osd_xattr_load(struct osd_device *osd, uint64_t dnode, nvlist_t **sa) +int __osd_xattr_load(struct osd_device *osd, sa_handle_t *hdl, nvlist_t **sa) { - sa_handle_t *sa_hdl; char *buf; int rc, size; - if (unlikely(dnode == ZFS_NO_OBJECT)) - return -ENOENT; - - rc = -sa_handle_get(osd->od_os, dnode, NULL, SA_HDL_PRIVATE, &sa_hdl); - if (rc) - return rc; - - rc = -sa_size(sa_hdl, SA_ZPL_DXATTR(osd), &size); + rc = -sa_size(hdl, SA_ZPL_DXATTR(osd), &size); if (rc) { if (rc == -ENOENT) rc = -nvlist_alloc(sa, NV_UNIQUE_NAME, KM_SLEEP); @@ -94,24 +81,22 @@ int __osd_xattr_load(struct osd_device *osd, uint64_t dnode, nvlist_t **sa) rc = -ENOMEM; goto out_sa; } - rc = -sa_lookup(sa_hdl, SA_ZPL_DXATTR(osd), buf, size); + rc = -sa_lookup(hdl, SA_ZPL_DXATTR(osd), buf, size); if (rc == 0) rc = -nvlist_unpack(buf, size, sa, KM_SLEEP); osd_zio_buf_free(buf, size); out_sa: - sa_handle_destroy(sa_hdl); return rc; } -static inline int __osd_xattr_cache(const struct lu_env *env, - struct osd_object *obj) +static inline int __osd_xattr_cache(struct osd_object *obj) { - LASSERT(obj->oo_sa_xattr == NULL); - LASSERT(obj->oo_db != NULL); - - return __osd_xattr_load(osd_obj2dev(obj), obj->oo_db->db_object, - &obj->oo_sa_xattr); + LASSERT(obj->oo_sa_hdl); + if (obj->oo_sa_xattr != NULL) + return 0; + return __osd_xattr_load(osd_obj2dev(obj), + obj->oo_sa_hdl, &obj->oo_sa_xattr); } static int @@ -119,19 +104,15 @@ __osd_sa_xattr_get(const struct lu_env *env, struct osd_object *obj, const struct lu_buf *buf, const char *name, int *sizep) { uchar_t *nv_value; - int rc; - - LASSERT(obj->oo_sa_hdl); + int rc = 0; - if (obj->oo_sa_xattr == NULL) { - rc = __osd_xattr_cache(env, obj); - if (rc) - return rc; - } + rc = __osd_xattr_cache(obj); + if (rc) + return rc; LASSERT(obj->oo_sa_xattr); - rc = -nvlist_lookup_byte_array(obj->oo_sa_xattr, name, &nv_value, - sizep); + rc = -nvlist_lookup_byte_array(obj->oo_sa_xattr, name, + &nv_value, sizep); if (rc) return rc; @@ -151,8 +132,8 @@ int __osd_xattr_get_large(const struct lu_env *env, struct osd_device *osd, uint64_t xattr, struct lu_buf *buf, const char *name, int *sizep) { - dmu_buf_t *xa_data_db; - sa_handle_t *sa_hdl = NULL; + dnode_t *xa_data_dn; + sa_handle_t *sa_hdl = NULL; uint64_t xa_data_obj, size; int rc; @@ -166,7 +147,7 @@ int __osd_xattr_get_large(const struct lu_env *env, struct osd_device *osd, if (rc) return rc; - rc = __osd_obj2dbuf(env, osd->od_os, xa_data_obj, &xa_data_db); + rc = __osd_obj2dnode(osd->od_os, xa_data_obj, &xa_data_dn); if (rc) return rc; @@ -196,13 +177,13 @@ int __osd_xattr_get_large(const struct lu_env *env, struct osd_device *osd, goto out; } - rc = -dmu_read(osd->od_os, xa_data_db->db_object, 0, + rc = -dmu_read(osd->od_os, xa_data_dn->dn_object, 0, size, buf->lb_buf, DMU_READ_PREFETCH); out: sa_handle_destroy(sa_hdl); out_rele: - dmu_buf_rele(xa_data_db, FTAG); + osd_dnode_rele(xa_data_dn); return rc; } @@ -248,7 +229,7 @@ int osd_xattr_get(const struct lu_env *env, struct dt_object *dt, int rc, size = 0; ENTRY; - LASSERT(obj->oo_db != NULL); + LASSERT(obj->oo_dn != NULL); LASSERT(osd_invariant(obj)); if (!osd_obj2dev(obj)->od_posix_acl && @@ -313,8 +294,8 @@ void __osd_xattr_declare_set(const struct lu_env *env, struct osd_object *obj, int vallen, const char *name, struct osd_thandle *oh) { - dmu_buf_t *db = obj->oo_db; - dmu_tx_t *tx = oh->ot_tx; + dmu_tx_t *tx = oh->ot_tx; + int bonuslen; if (unlikely(obj->oo_destroyed)) return; @@ -330,15 +311,17 @@ void __osd_xattr_declare_set(const struct lu_env *env, struct osd_object *obj, /* XXX: it should be possible to skip spill * declaration if specific EA is part of * bonus and doesn't grow */ - dmu_tx_hold_spill(tx, db->db_object); + dmu_tx_hold_spill(tx, obj->oo_dn->dn_object); return; } + bonuslen = osd_obj_bonuslen(obj); + /* the object doesn't exist, but we've declared bonus * in osd_declare_object_create() yet */ - if (obj->oo_ea_in_bonus > DN_MAX_BONUSLEN) { + if (obj->oo_ea_in_bonus > bonuslen) { /* spill has been declared already */ - } else if (obj->oo_ea_in_bonus + vallen > DN_MAX_BONUSLEN) { + } else if (obj->oo_ea_in_bonus + vallen > bonuslen) { /* we're about to exceed bonus, let's declare spill */ dmu_tx_hold_spill(tx, DMU_NEW_OBJECT); } @@ -363,6 +346,127 @@ int osd_declare_xattr_set(const struct lu_env *env, struct dt_object *dt, RETURN(0); } +int __osd_sa_attr_init(const struct lu_env *env, struct osd_object *obj, + struct osd_thandle *oh) +{ + sa_bulk_attr_t *bulk = osd_oti_get(env)->oti_attr_bulk; + struct osa_attr *osa = &osd_oti_get(env)->oti_osa; + struct lu_buf *lb = &osd_oti_get(env)->oti_xattr_lbuf; + struct osd_device *osd = osd_obj2dev(obj); + uint64_t crtime[2], gen; + timestruc_t now; + size_t size; + int rc, cnt; + + obj->oo_late_xattr = 0; + obj->oo_late_attr_set = 0; + + gen = dmu_tx_get_txg(oh->ot_tx); + gethrestime(&now); + ZFS_TIME_ENCODE(&now, crtime); + + osa->atime[0] = obj->oo_attr.la_atime; + osa->ctime[0] = obj->oo_attr.la_ctime; + osa->mtime[0] = obj->oo_attr.la_mtime; + osa->mode = obj->oo_attr.la_mode; + osa->uid = obj->oo_attr.la_uid; + osa->gid = obj->oo_attr.la_gid; + osa->rdev = obj->oo_attr.la_rdev; + osa->nlink = obj->oo_attr.la_nlink; + osa->flags = attrs_fs2zfs(obj->oo_attr.la_flags); + osa->size = obj->oo_attr.la_size; +#ifdef ZFS_PROJINHERIT + if (osd->od_projectused_dn) { + if (obj->oo_attr.la_valid & LA_PROJID) + osa->projid = obj->oo_attr.la_projid; + else + osa->projid = ZFS_DEFAULT_PROJID; + osa->flags |= ZFS_PROJID; + obj->oo_with_projid = 1; + } +#endif + + cnt = 0; + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(osd), NULL, &osa->mode, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(osd), NULL, &osa->size, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GEN(osd), NULL, &gen, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(osd), NULL, &osa->uid, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(osd), NULL, &osa->gid, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PARENT(osd), NULL, + &obj->oo_parent, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(osd), NULL, &osa->flags, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(osd), NULL, osa->atime, 16); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(osd), NULL, osa->mtime, 16); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(osd), NULL, osa->ctime, 16); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CRTIME(osd), NULL, crtime, 16); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(osd), NULL, &osa->nlink, 8); +#ifdef ZFS_PROJINHERIT + if (osd->od_projectused_dn) + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PROJID(osd), NULL, + &osa->projid, 8); +#endif + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(osd), NULL, &osa->rdev, 8); + LASSERT(cnt <= ARRAY_SIZE(osd_oti_get(env)->oti_attr_bulk)); + + /* Update the SA for additions, modifications, and removals. */ + rc = -nvlist_size(obj->oo_sa_xattr, &size, NV_ENCODE_XDR); + if (rc) + return rc; + + lu_buf_check_and_alloc(lb, size); + if (lb->lb_buf == NULL) { + CERROR("%s: can't allocate buffer for xattr update\n", + osd->od_svname); + return -ENOMEM; + } + + rc = -nvlist_pack(obj->oo_sa_xattr, (char **)&lb->lb_buf, &size, + NV_ENCODE_XDR, KM_SLEEP); + if (rc) + return rc; + + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_DXATTR(osd), NULL, lb->lb_buf, size); + + rc = -sa_replace_all_by_template(obj->oo_sa_hdl, bulk, cnt, oh->ot_tx); + + return rc; +} + +int __osd_sa_xattr_update(const struct lu_env *env, struct osd_object *obj, + struct osd_thandle *oh) +{ + struct lu_buf *lb = &osd_oti_get(env)->oti_xattr_lbuf; + struct osd_device *osd = osd_obj2dev(obj); + char *dxattr; + size_t size; + int rc; + + obj->oo_late_xattr = 0; + + /* Update the SA for additions, modifications, and removals. */ + rc = -nvlist_size(obj->oo_sa_xattr, &size, NV_ENCODE_XDR); + if (rc) + return rc; + + lu_buf_check_and_alloc(lb, size); + if (lb->lb_buf == NULL) { + CERROR("%s: can't allocate buffer for xattr update\n", + osd->od_svname); + return -ENOMEM; + } + + dxattr = lb->lb_buf; + rc = -nvlist_pack(obj->oo_sa_xattr, &dxattr, &size, + NV_ENCODE_XDR, KM_SLEEP); + if (rc) + return rc; + LASSERT(dxattr == lb->lb_buf); + + sa_update(obj->oo_sa_hdl, SA_ZPL_DXATTR(osd), dxattr, size, oh->ot_tx); + + return 0; +} + /* * Set an extended attribute. * This transaction must have called udmu_xattr_declare_set() first. @@ -371,37 +475,20 @@ int osd_declare_xattr_set(const struct lu_env *env, struct dt_object *dt, * * No locking is done here. */ -static int -__osd_sa_xattr_update(const struct lu_env *env, struct osd_object *obj, - struct osd_thandle *oh) +int __osd_sa_xattr_schedule_update(const struct lu_env *env, + struct osd_object *obj, + struct osd_thandle *oh) { - struct osd_device *osd = osd_obj2dev(obj); - char *dxattr; - size_t sa_size; - int rc; - ENTRY; LASSERT(obj->oo_sa_hdl); LASSERT(obj->oo_sa_xattr); - /* Update the SA for additions, modifications, and removals. */ - rc = -nvlist_size(obj->oo_sa_xattr, &sa_size, NV_ENCODE_XDR); - if (rc) - return rc; - - dxattr = osd_zio_buf_alloc(sa_size); - if (dxattr == NULL) - RETURN(-ENOMEM); + /* schedule batched SA update in osd_object_sa_dirty_rele() */ + obj->oo_late_xattr = 1; + osd_object_sa_dirty_add(obj, oh); - rc = -nvlist_pack(obj->oo_sa_xattr, &dxattr, &sa_size, - NV_ENCODE_XDR, KM_SLEEP); - if (rc) - GOTO(out_free, rc); + RETURN(0); - rc = osd_object_sa_update(obj, SA_ZPL_DXATTR(osd), dxattr, sa_size, oh); -out_free: - osd_zio_buf_free(dxattr, sa_size); - RETURN(rc); } int __osd_sa_xattr_set(const struct lu_env *env, struct osd_object *obj, @@ -414,12 +501,9 @@ int __osd_sa_xattr_set(const struct lu_env *env, struct osd_object *obj, int rc; int too_big = 0; - LASSERT(obj->oo_sa_hdl); - if (obj->oo_sa_xattr == NULL) { - rc = __osd_xattr_cache(env, obj); - if (rc) - return rc; - } + rc = __osd_xattr_cache(obj); + if (rc) + return rc; LASSERT(obj->oo_sa_xattr); /* Limited to 32k to keep nvpair memory allocations small */ @@ -448,7 +532,7 @@ int __osd_sa_xattr_set(const struct lu_env *env, struct osd_object *obj, DATA_TYPE_BYTE_ARRAY); if (rc < 0) return rc; - rc = __osd_sa_xattr_update(env, obj, oh); + rc = __osd_sa_xattr_schedule_update(env, obj, oh); return rc == 0 ? -EFBIG : rc; } } else if (rc == -ENOENT) { @@ -479,7 +563,13 @@ int __osd_sa_xattr_set(const struct lu_env *env, struct osd_object *obj, if (rc) return rc; - rc = __osd_sa_xattr_update(env, obj, oh); + /* batch updates only for just created dnodes where we + * used to set number of EAs in a single transaction */ + if (obj->oo_dn->dn_allocated_txg == oh->ot_tx->tx_txg) + rc = __osd_sa_xattr_schedule_update(env, obj, oh); + else + rc = __osd_sa_xattr_update(env, obj, oh); + return rc; } @@ -489,8 +579,8 @@ __osd_xattr_set(const struct lu_env *env, struct osd_object *obj, struct osd_thandle *oh) { struct osd_device *osd = osd_obj2dev(obj); - dmu_buf_t *xa_zap_db = NULL; - dmu_buf_t *xa_data_db = NULL; + dnode_t *xa_zap_dn = NULL; + dnode_t *xa_data_dn = NULL; uint64_t xa_data_obj; sa_handle_t *sa_hdl = NULL; dmu_tx_t *tx = oh->ot_tx; @@ -504,11 +594,11 @@ __osd_xattr_set(const struct lu_env *env, struct osd_object *obj, la->la_valid = LA_MODE; la->la_mode = S_IFDIR | S_IRUGO | S_IWUSR | S_IXUGO; - rc = __osd_zap_create(env, osd, &xa_zap_db, tx, la, 0); + rc = __osd_zap_create(env, osd, &xa_zap_dn, tx, la, 0, 0); if (rc) return rc; - obj->oo_xattr = xa_zap_db->db_object; + obj->oo_xattr = xa_zap_dn->dn_object; rc = osd_object_sa_update(obj, SA_ZPL_XATTR(osd), &obj->oo_xattr, 8, oh); if (rc) @@ -526,8 +616,7 @@ __osd_xattr_set(const struct lu_env *env, struct osd_object *obj, * Entry already exists. * We'll truncate the existing object. */ - rc = __osd_obj2dbuf(env, osd->od_os, xa_data_obj, - &xa_data_db); + rc = __osd_obj2dnode(osd->od_os, xa_data_obj, &xa_data_dn); if (rc) goto out; @@ -540,7 +629,7 @@ __osd_xattr_set(const struct lu_env *env, struct osd_object *obj, if (rc) goto out_sa; - rc = -dmu_free_range(osd->od_os, xa_data_db->db_object, + rc = -dmu_free_range(osd->od_os, xa_data_dn->dn_object, 0, DMU_OBJECT_END, tx); if (rc) goto out_sa; @@ -559,10 +648,10 @@ __osd_xattr_set(const struct lu_env *env, struct osd_object *obj, la->la_valid = LA_MODE; la->la_mode = S_IFREG | S_IRUGO | S_IWUSR; - rc = __osd_object_create(env, obj, &xa_data_db, tx, la); + rc = __osd_object_create(env, obj, &xa_data_dn, tx, la); if (rc) goto out; - xa_data_obj = xa_data_db->db_object; + xa_data_obj = xa_data_dn->dn_object; rc = -sa_handle_get(osd->od_os, xa_data_obj, NULL, SA_HDL_PRIVATE, &sa_hdl); @@ -587,10 +676,10 @@ __osd_xattr_set(const struct lu_env *env, struct osd_object *obj, out_sa: sa_handle_destroy(sa_hdl); out: - if (xa_data_db != NULL) - dmu_buf_rele(xa_data_db, FTAG); - if (xa_zap_db != NULL) - dmu_buf_rele(xa_zap_db, FTAG); + if (xa_data_dn != NULL) + osd_dnode_rele(xa_data_dn); + if (xa_zap_dn != NULL) + osd_dnode_rele(xa_zap_dn); return rc; } @@ -612,10 +701,6 @@ int osd_xattr_set(const struct lu_env *env, struct dt_object *dt, strcmp(name, XATTR_NAME_POSIX_ACL_DEFAULT) == 0)) RETURN(-EOPNOTSUPP); - if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LINKEA_OVERFLOW) && - strcmp(name, XATTR_NAME_LINK) == 0) - RETURN(-ENOSPC); - oh = container_of0(handle, struct osd_thandle, ot_super); down_write(&obj->oo_guard); @@ -675,7 +760,7 @@ int osd_declare_xattr_del(const struct lu_env *env, struct dt_object *dt, oh = container_of0(handle, struct osd_thandle, ot_super); LASSERT(oh->ot_tx != NULL); - LASSERT(obj->oo_db != NULL); + LASSERT(obj->oo_dn != NULL); down_read(&obj->oo_guard); if (likely(dt_object_exists(&obj->oo_dt) && !obj->oo_destroyed)) @@ -690,15 +775,13 @@ static int __osd_sa_xattr_del(const struct lu_env *env, struct osd_object *obj, { int rc; - if (obj->oo_sa_xattr == NULL) { - rc = __osd_xattr_cache(env, obj); - if (rc) - return rc; - } + rc = __osd_xattr_cache(obj); + if (rc) + return rc; rc = -nvlist_remove(obj->oo_sa_xattr, name, DATA_TYPE_BYTE_ARRAY); if (rc == 0) - rc = __osd_sa_xattr_update(env, obj, oh); + rc = __osd_sa_xattr_schedule_update(env, obj, oh); return rc; } @@ -748,7 +831,7 @@ int osd_xattr_del(const struct lu_env *env, struct dt_object *dt, ENTRY; LASSERT(handle != NULL); - LASSERT(obj->oo_db != NULL); + LASSERT(obj->oo_dn != NULL); LASSERT(osd_invariant(obj)); LASSERT(dt_object_exists(dt)); oh = container_of0(handle, struct osd_thandle, ot_super); @@ -850,7 +933,7 @@ int osd_xattrs_destroy(const struct lu_env *env, rc = -dmu_object_free(osd->od_os, obj->oo_xattr, tx); if (rc) - CERROR("%s: free xattr "LPU64" failed: rc = %d\n", + CERROR("%s: free xattr %llu failed: rc = %d\n", osd->od_svname, obj->oo_xattr, rc); return rc; @@ -864,13 +947,9 @@ osd_sa_xattr_list(const struct lu_env *env, struct osd_object *obj, int len, counted = 0; int rc = 0; - if (obj->oo_sa_xattr == NULL) { - rc = __osd_xattr_cache(env, obj); - if (rc) - return rc; - } - - LASSERT(obj->oo_sa_xattr); + rc = __osd_xattr_cache(obj); + if (rc) + return rc; while ((nvp = nvlist_next_nvpair(obj->oo_sa_xattr, nvp)) != NULL) { const char *name = nvpair_name(nvp); @@ -902,7 +981,7 @@ int osd_xattr_list(const struct lu_env *env, struct dt_object *dt, int rc, counted; ENTRY; - LASSERT(obj->oo_db != NULL); + LASSERT(obj->oo_dn != NULL); LASSERT(osd_invariant(obj)); LASSERT(dt_object_exists(dt));