X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fosd-zfs%2Fosd_object.c;h=e66d2e03bc25ccbd150ca11916f5a82c6c4f2b1c;hp=86e9d8763413256cecf368186742fdfd0b449f40;hb=26995a3588e72d65f609b12772d24a879c9deb7f;hpb=08aa217ce49aba1ded52e0f7adb8a607035123fd diff --git a/lustre/osd-zfs/osd_object.c b/lustre/osd-zfs/osd_object.c index 86e9d87..e66d2e0 100644 --- a/lustre/osd-zfs/osd_object.c +++ b/lustre/osd-zfs/osd_object.c @@ -28,7 +28,7 @@ * Use is subject to license terms. */ /* - * Copyright (c) 2012, Intel Corporation. + * Copyright (c) 2012, 2013, Intel Corporation. * Use is subject to license terms. */ /* @@ -42,14 +42,10 @@ * Author: Johann Lombardi */ -#ifndef EXPORT_SYMTAB -# define EXPORT_SYMTAB -#endif #define DEBUG_SUBSYSTEM S_OSD #include #include -#include #include #include #include @@ -72,14 +68,14 @@ #include #include -static char *osd_obj_tag = "osd_object"; +char *osd_obj_tag = "osd_object"; static struct dt_object_operations osd_obj_ops; static struct lu_object_operations osd_lu_obj_ops; extern struct dt_body_operations osd_body_ops; static struct dt_object_operations osd_obj_otable_it_ops; -extern cfs_mem_cache_t *osd_object_kmem; +extern struct kmem_cache *osd_object_kmem; static void osd_object_sa_fini(struct osd_object *obj) @@ -91,20 +87,20 @@ osd_object_sa_fini(struct osd_object *obj) } static int -osd_object_sa_init(struct osd_object *obj, udmu_objset_t *uos) +osd_object_sa_init(struct osd_object *obj, struct osd_device *o) { int rc; LASSERT(obj->oo_sa_hdl == NULL); LASSERT(obj->oo_db != NULL); - rc = -sa_handle_get(uos->os, obj->oo_db->db_object, obj, + rc = -sa_handle_get(o->od_os, obj->oo_db->db_object, obj, SA_HDL_PRIVATE, &obj->oo_sa_hdl); if (rc) return rc; /* Cache the xattr object id, valid for the life of the object */ - rc = -sa_lookup(obj->oo_sa_hdl, SA_ZPL_XATTR(uos), &obj->oo_xattr, 8); + rc = -sa_lookup(obj->oo_sa_hdl, SA_ZPL_XATTR(o), &obj->oo_xattr, 8); if (rc == -ENOENT) { obj->oo_xattr = ZFS_NO_OBJECT; rc = 0; @@ -121,13 +117,13 @@ osd_object_sa_init(struct osd_object *obj, udmu_objset_t *uos) static void osd_object_sa_dirty_add(struct osd_object *obj, struct osd_thandle *oh) { - if (!cfs_list_empty(&obj->oo_sa_linkage)) + if (!list_empty(&obj->oo_sa_linkage)) return; down(&oh->ot_sa_lock); write_lock(&obj->oo_attr_lock); - if (likely(cfs_list_empty(&obj->oo_sa_linkage))) - cfs_list_add(&obj->oo_sa_linkage, &oh->ot_sa_list); + if (likely(list_empty(&obj->oo_sa_linkage))) + list_add(&obj->oo_sa_linkage, &oh->ot_sa_list); write_unlock(&obj->oo_attr_lock); up(&oh->ot_sa_lock); } @@ -140,12 +136,12 @@ void osd_object_sa_dirty_rele(struct osd_thandle *oh) struct osd_object *obj; down(&oh->ot_sa_lock); - while (!cfs_list_empty(&oh->ot_sa_list)) { - obj = cfs_list_entry(oh->ot_sa_list.next, + while (!list_empty(&oh->ot_sa_list)) { + obj = list_entry(oh->ot_sa_list.next, struct osd_object, oo_sa_linkage); sa_spill_rele(obj->oo_sa_hdl); write_lock(&obj->oo_attr_lock); - cfs_list_del_init(&obj->oo_sa_linkage); + list_del_init(&obj->oo_sa_linkage); write_unlock(&obj->oo_attr_lock); } up(&oh->ot_sa_lock); @@ -189,7 +185,7 @@ osd_object_sa_bulk_update(struct osd_object *obj, sa_bulk_attr_t *attrs, /* * Retrieve the attributes of a DMU object */ -int __osd_object_attr_get(const struct lu_env *env, udmu_objset_t *uos, +int __osd_object_attr_get(const struct lu_env *env, struct osd_device *o, struct osd_object *obj, struct lu_attr *la) { struct osa_attr *osa = &osd_oti_get(env)->oti_osa; @@ -201,7 +197,7 @@ int __osd_object_attr_get(const struct lu_env *env, udmu_objset_t *uos, LASSERT(obj->oo_db != NULL); - rc = -sa_handle_get(uos->os, obj->oo_db->db_object, NULL, + rc = -sa_handle_get(o->od_os, obj->oo_db->db_object, NULL, SA_HDL_PRIVATE, &sa_hdl); if (rc) RETURN(rc); @@ -213,15 +209,15 @@ int __osd_object_attr_get(const struct lu_env *env, udmu_objset_t *uos, la->la_valid |= LA_ATIME | LA_MTIME | LA_CTIME | LA_MODE | LA_TYPE | LA_SIZE | LA_UID | LA_GID | LA_FLAGS | LA_NLINK; - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(uos), NULL, osa->atime, 16); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(uos), NULL, osa->mtime, 16); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(uos), NULL, osa->ctime, 16); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(uos), NULL, &osa->mode, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(uos), NULL, &osa->size, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(uos), NULL, &osa->nlink, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(uos), NULL, &osa->uid, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(uos), NULL, &osa->gid, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(uos), NULL, &osa->flags, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(o), NULL, osa->atime, 16); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(o), NULL, osa->mtime, 16); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(o), NULL, osa->ctime, 16); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(o), NULL, &osa->mode, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(o), NULL, &osa->size, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(o), NULL, &osa->nlink, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(o), NULL, &osa->uid, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(o), NULL, &osa->gid, 8); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(o), NULL, &osa->flags, 8); rc = -sa_bulk_lookup(sa_hdl, bulk, cnt); if (rc) @@ -234,11 +230,11 @@ int __osd_object_attr_get(const struct lu_env *env, udmu_objset_t *uos, la->la_uid = osa->uid; la->la_gid = osa->gid; la->la_nlink = osa->nlink; - la->la_flags = osa->flags; + la->la_flags = attrs_zfs2fs(osa->flags); la->la_size = osa->size; if (S_ISCHR(la->la_mode) || S_ISBLK(la->la_mode)) { - rc = -sa_lookup(sa_hdl, SA_ZPL_RDEV(uos), &osa->rdev, 8); + rc = -sa_lookup(sa_hdl, SA_ZPL_RDEV(o), &osa->rdev, 8); if (rc) GOTO(out_bulk, rc); la->la_rdev = osa->rdev; @@ -253,21 +249,19 @@ out_sa: } int __osd_obj2dbuf(const struct lu_env *env, objset_t *os, - uint64_t oid, dmu_buf_t **dbp, void *tag) + uint64_t oid, dmu_buf_t **dbp) { dmu_object_info_t *doi = &osd_oti_get(env)->oti_doi; int rc; - LASSERT(tag); - - rc = -sa_buf_hold(os, oid, tag, dbp); + rc = -sa_buf_hold(os, oid, osd_obj_tag, dbp); if (rc) return rc; dmu_object_info_from_db(*dbp, doi); if (unlikely (oid != DMU_USERUSED_OBJECT && oid != DMU_GROUPUSED_OBJECT && doi->doi_bonus_type != DMU_OT_SA)) { - sa_buf_rele(*dbp, tag); + sa_buf_rele(*dbp, osd_obj_tag); *dbp = NULL; return -EINVAL; } @@ -290,7 +284,7 @@ struct lu_object *osd_object_alloc(const struct lu_env *env, { struct osd_object *mo; - OBD_SLAB_ALLOC_PTR_GFP(mo, osd_object_kmem, CFS_ALLOC_IO); + OBD_SLAB_ALLOC_PTR_GFP(mo, osd_object_kmem, GFP_NOFS); if (mo != NULL) { struct lu_object *l; @@ -298,7 +292,7 @@ struct lu_object *osd_object_alloc(const struct lu_env *env, dt_object_init(&mo->oo_dt, NULL, d); mo->oo_dt.do_ops = &osd_obj_ops; l->lo_ops = &osd_lu_obj_ops; - CFS_INIT_LIST_HEAD(&mo->oo_sa_linkage); + INIT_LIST_HEAD(&mo->oo_sa_linkage); init_rwsem(&mo->oo_sem); sema_init(&mo->oo_guard, 1); rwlock_init(&mo->oo_attr_lock); @@ -314,7 +308,6 @@ struct lu_object *osd_object_alloc(const struct lu_env *env, int osd_object_init0(const struct lu_env *env, struct osd_object *obj) { struct osd_device *osd = osd_obj2dev(obj); - udmu_objset_t *uos = &osd->od_objset; const struct lu_fid *fid = lu_object_fid(&obj->oo_dt.do_lu); int rc = 0; ENTRY; @@ -324,13 +317,12 @@ int osd_object_init0(const struct lu_env *env, struct osd_object *obj) /* object exist */ - rc = osd_object_sa_init(obj, uos); + rc = osd_object_sa_init(obj, osd); if (rc) RETURN(rc); /* cache attrs in object */ - rc = __osd_object_attr_get(env, &osd->od_objset, - obj, &obj->oo_attr); + rc = __osd_object_attr_get(env, osd, obj, &obj->oo_attr); if (rc) RETURN(rc); @@ -343,12 +335,45 @@ int osd_object_init0(const struct lu_env *env, struct osd_object *obj) */ obj->oo_dt.do_lu.lo_header->loh_attr |= obj->oo_attr.la_mode & S_IFMT; - cfs_mb(); + smp_mb(); obj->oo_dt.do_lu.lo_header->loh_attr |= LOHA_EXISTS; RETURN(0); } +static int osd_check_lma(const struct lu_env *env, struct osd_object *obj) +{ + struct osd_thread_info *info = osd_oti_get(env); + struct lu_buf buf; + int rc; + struct lustre_mdt_attrs *lma; + ENTRY; + + CLASSERT(sizeof(info->oti_buf) >= sizeof(*lma)); + lma = (struct lustre_mdt_attrs *)info->oti_buf; + buf.lb_buf = lma; + buf.lb_len = sizeof(info->oti_buf); + + rc = osd_xattr_get(env, &obj->oo_dt, &buf, XATTR_NAME_LMA, BYPASS_CAPA); + if (rc > 0) { + rc = 0; + lustre_lma_swab(lma); + if (unlikely((lma->lma_incompat & ~LMA_INCOMPAT_SUPP) || + CFS_FAIL_CHECK(OBD_FAIL_OSD_LMA_INCOMPAT))) { + CWARN("%s: unsupported incompat LMA feature(s) %#x for " + "fid = "DFID"\n", osd_obj2dev(obj)->od_svname, + lma->lma_incompat & ~LMA_INCOMPAT_SUPP, + PFID(lu_object_fid(&obj->oo_dt.do_lu))); + rc = -EOPNOTSUPP; + } + } else if (rc == -ENODATA) { + /* haven't initialize LMA xattr */ + rc = 0; + } + + RETURN(rc); +} + /* * Concurrency: no concurrent access is possible that early in object * life-cycle. @@ -364,27 +389,34 @@ static int osd_object_init(const struct lu_env *env, struct lu_object *l, LASSERT(osd_invariant(obj)); + if (fid_is_otable_it(&l->lo_header->loh_fid)) { + obj->oo_dt.do_ops = &osd_obj_otable_it_ops; + l->lo_header->loh_attr |= LOHA_EXISTS; + RETURN(0); + } + rc = osd_fid_lookup(env, osd, lu_object_fid(l), &oid); if (rc == 0) { LASSERT(obj->oo_db == NULL); - rc = __osd_obj2dbuf(env, osd->od_objset.os, oid, - &obj->oo_db, osd_obj_tag); - if (rc == 0) { - LASSERT(obj->oo_db); - rc = osd_object_init0(env, obj); - } else { + rc = __osd_obj2dbuf(env, osd->od_os, oid, &obj->oo_db); + if (rc != 0) { CERROR("%s: lookup "DFID"/"LPX64" failed: rc = %d\n", osd->od_svname, PFID(lu_object_fid(l)), oid, rc); + GOTO(out, rc); } + LASSERT(obj->oo_db); + rc = osd_object_init0(env, obj); + if (rc != 0) + GOTO(out, rc); + + rc = osd_check_lma(env, obj); + if (rc != 0) + GOTO(out, rc); } else if (rc == -ENOENT) { - if (fid_is_otable_it(&l->lo_header->loh_fid)) { - obj->oo_dt.do_ops = &osd_obj_otable_it_ops; - /* LFSCK iterator object is special without inode */ - l->lo_header->loh_attr |= LOHA_EXISTS; - } rc = 0; } LASSERT(osd_invariant(obj)); +out: RETURN(rc); } @@ -407,7 +439,6 @@ static void __osd_declare_object_destroy(const struct lu_env *env, struct osd_thandle *oh) { struct osd_device *osd = osd_obj2dev(obj); - udmu_objset_t *uos = &osd->od_objset; dmu_buf_t *db = obj->oo_db; zap_attribute_t *za = &osd_oti_get(env)->oti_za; uint64_t oid = db->db_object, xid; @@ -423,7 +454,7 @@ static void __osd_declare_object_destroy(const struct lu_env *env, dmu_tx_hold_free(tx, oid, 0, DMU_OBJECT_END); - rc = -udmu_zap_cursor_init(&zc, uos, oid, 0); + rc = osd_zap_cursor_init(&zc, osd->od_os, oid, 0); if (rc) goto out; @@ -431,7 +462,7 @@ static void __osd_declare_object_destroy(const struct lu_env *env, BUG_ON(za->za_integer_length != sizeof(uint64_t)); BUG_ON(za->za_num_integers != 1); - rc = -zap_lookup(uos->os, obj->oo_xattr, za->za_name, + rc = -zap_lookup(osd->od_os, obj->oo_xattr, za->za_name, sizeof(uint64_t), 1, &xid); if (rc) { CERROR("%s: xattr lookup failed: rc = %d\n", @@ -445,7 +476,7 @@ static void __osd_declare_object_destroy(const struct lu_env *env, if (rc == -ENOENT) rc = 0; out_err: - udmu_zap_cursor_fini(zc); + osd_zap_cursor_fini(zc); } out: if (rc && tx->tx_err == 0) @@ -497,16 +528,6 @@ static int osd_declare_object_destroy(const struct lu_env *env, RETURN(rc); } -int __osd_object_free(udmu_objset_t *uos, uint64_t oid, dmu_tx_t *tx) -{ - LASSERT(uos->objects != 0); - spin_lock(&uos->lock); - uos->objects--; - spin_unlock(&uos->lock); - - return -dmu_object_free(uos->os, oid, tx); -} - /* * Delete a DMU object * @@ -521,7 +542,6 @@ static int __osd_object_destroy(const struct lu_env *env, dmu_tx_t *tx, void *tag) { struct osd_device *osd = osd_obj2dev(obj); - udmu_objset_t *uos = &osd->od_objset; uint64_t xid; zap_attribute_t *za = &osd_oti_get(env)->oti_za; zap_cursor_t *zc; @@ -533,35 +553,35 @@ static int __osd_object_destroy(const struct lu_env *env, /* zap holding xattrs */ if (obj->oo_xattr != ZFS_NO_OBJECT) { - rc = -udmu_zap_cursor_init(&zc, uos, obj->oo_xattr, 0); + rc = osd_zap_cursor_init(&zc, osd->od_os, obj->oo_xattr, 0); if (rc) return rc; while ((rc = -zap_cursor_retrieve(zc, za)) == 0) { BUG_ON(za->za_integer_length != sizeof(uint64_t)); BUG_ON(za->za_num_integers != 1); - rc = -zap_lookup(uos->os, obj->oo_xattr, za->za_name, + rc = -zap_lookup(osd->od_os, obj->oo_xattr, za->za_name, sizeof(uint64_t), 1, &xid); if (rc) { CERROR("%s: lookup xattr %s failed: rc = %d\n", osd->od_svname, za->za_name, rc); continue; } - rc = __osd_object_free(uos, xid, tx); + rc = -dmu_object_free(osd->od_os, xid, tx); if (rc) CERROR("%s: fetch xattr %s failed: rc = %d\n", osd->od_svname, za->za_name, rc); zap_cursor_advance(zc); } - udmu_zap_cursor_fini(zc); + osd_zap_cursor_fini(zc); - rc = __osd_object_free(uos, obj->oo_xattr, tx); + rc = -dmu_object_free(osd->od_os, obj->oo_xattr, tx); if (rc) CERROR("%s: freeing xattr failed: rc = %d\n", osd->od_svname, rc); } - return __osd_object_free(uos, obj->oo_db->db_object, tx); + return -dmu_object_free(osd->od_os, obj->oo_db->db_object, tx); } static int osd_object_destroy(const struct lu_env *env, @@ -588,7 +608,7 @@ static int osd_object_destroy(const struct lu_env *env, zapid = osd_get_name_n_idx(env, osd, fid, buf); /* remove obj ref from index dir (it depends) */ - rc = -zap_remove(osd->od_objset.os, zapid, buf, oh->ot_tx); + rc = -zap_remove(osd->od_os, zapid, buf, oh->ot_tx); if (rc) { CERROR("%s: zap_remove() failed: rc = %d\n", osd->od_svname, rc); @@ -598,13 +618,13 @@ static int osd_object_destroy(const struct lu_env *env, /* Remove object from inode accounting. It is not fatal for the destroy * operation if something goes wrong while updating accounting, but we * still log an error message to notify the administrator */ - rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid, + rc = -zap_increment_int(osd->od_os, osd->od_iusr_oid, obj->oo_attr.la_uid, -1, oh->ot_tx); if (rc) CERROR("%s: failed to remove "DFID" from accounting ZAP for usr" " %d: rc = %d\n", osd->od_svname, PFID(fid), obj->oo_attr.la_uid, rc); - rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid, + rc = -zap_increment_int(osd->od_os, osd->od_igrp_oid, obj->oo_attr.la_gid, -1, oh->ot_tx); if (rc) CERROR("%s: failed to remove "DFID" from accounting ZAP for grp" @@ -637,7 +657,7 @@ static void osd_object_delete(const struct lu_env *env, struct lu_object *l) obj->oo_sa_xattr = NULL; } sa_buf_rele(obj->oo_db, osd_obj_tag); - cfs_list_del(&obj->oo_sa_linkage); + list_del(&obj->oo_sa_linkage); obj->oo_db = NULL; } } @@ -782,8 +802,9 @@ static inline int qsd_transfer(const struct lu_env *env, /* and one less inode for the current id */ qi->lqi_id.qid_uid = orig_id;; qi->lqi_space = -1; + /* can't get EDQUOT when reducing usage */ rc = qsd_op_begin(env, qsd, trans, qi, NULL); - if (rc == -EDQUOT || rc == -EINPROGRESS) + if (rc == -EINPROGRESS) rc = 0; if (rc) return rc; @@ -804,7 +825,8 @@ static inline int qsd_transfer(const struct lu_env *env, qi->lqi_id.qid_uid = orig_id; qi->lqi_space = -bspace; rc = qsd_op_begin(env, qsd, trans, qi, NULL); - if (rc == -EDQUOT || rc == -EINPROGRESS) + /* can't get EDQUOT when reducing usage */ + if (rc == -EINPROGRESS) rc = 0; return rc; } @@ -835,7 +857,10 @@ static int osd_declare_attr_set(const struct lu_env *env, oh = container_of0(handle, struct osd_thandle, ot_super); 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); sa_object_size(obj->oo_sa_hdl, &blksize, &bspace); bspace = toqb(bspace * blksize); @@ -887,10 +912,10 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt, { struct osd_object *obj = osd_dt_obj(dt); struct osd_device *osd = osd_obj2dev(obj); - udmu_objset_t *uos = &osd->od_objset; struct osd_thandle *oh; struct osa_attr *osa = &osd_oti_get(env)->oti_osa; sa_bulk_attr_t *bulk; + __u64 valid = la->la_valid; int cnt; int rc = 0; @@ -905,7 +930,11 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt, transaction group. */ LASSERT(oh->ot_tx->tx_txg != 0); - if (la->la_valid == 0) + /* Only allow set size for regular file */ + if (!S_ISREG(dt->do_lu.lo_header->loh_attr)) + valid &= ~(LA_SIZE | LA_BLOCKS); + + if (valid == 0) RETURN(0); OBD_ALLOC(bulk, sizeof(sa_bulk_attr_t) * 10); @@ -913,30 +942,30 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt, RETURN(-ENOMEM); /* do both accounting updates outside oo_attr_lock below */ - if ((la->la_valid & LA_UID) && (la->la_uid != obj->oo_attr.la_uid)) { + if ((valid & LA_UID) && (la->la_uid != obj->oo_attr.la_uid)) { /* Update user accounting. Failure isn't fatal, but we still * log an error message */ - rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid, + rc = -zap_increment_int(osd->od_os, osd->od_iusr_oid, la->la_uid, 1, oh->ot_tx); if (rc) CERROR("%s: failed to update accounting ZAP for user " "%d (%d)\n", osd->od_svname, la->la_uid, rc); - rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid, + rc = -zap_increment_int(osd->od_os, osd->od_iusr_oid, obj->oo_attr.la_uid, -1, oh->ot_tx); if (rc) CERROR("%s: failed to update accounting ZAP for user " "%d (%d)\n", osd->od_svname, obj->oo_attr.la_uid, rc); } - if ((la->la_valid & LA_GID) && (la->la_gid != obj->oo_attr.la_gid)) { + if ((valid & LA_GID) && (la->la_gid != obj->oo_attr.la_gid)) { /* Update group accounting. Failure isn't fatal, but we still * log an error message */ - rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid, + rc = -zap_increment_int(osd->od_os, osd->od_igrp_oid, la->la_gid, 1, oh->ot_tx); if (rc) CERROR("%s: failed to update accounting ZAP for user " "%d (%d)\n", osd->od_svname, la->la_gid, rc); - rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid, + rc = -zap_increment_int(osd->od_os, osd->od_igrp_oid, obj->oo_attr.la_gid, -1, oh->ot_tx); if (rc) CERROR("%s: failed to update accounting ZAP for user " @@ -946,60 +975,63 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt, write_lock(&obj->oo_attr_lock); cnt = 0; - if (la->la_valid & LA_ATIME) { + if (valid & LA_ATIME) { osa->atime[0] = obj->oo_attr.la_atime = la->la_atime; - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(uos), NULL, + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(osd), NULL, osa->atime, 16); } - if (la->la_valid & LA_MTIME) { + if (valid & LA_MTIME) { osa->mtime[0] = obj->oo_attr.la_mtime = la->la_mtime; - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(uos), NULL, + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(osd), NULL, osa->mtime, 16); } - if (la->la_valid & LA_CTIME) { + if (valid & LA_CTIME) { osa->ctime[0] = obj->oo_attr.la_ctime = la->la_ctime; - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(uos), NULL, + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(osd), NULL, osa->ctime, 16); } - if (la->la_valid & LA_MODE) { + if (valid & LA_MODE) { /* mode is stored along with type, so read it first */ obj->oo_attr.la_mode = (obj->oo_attr.la_mode & S_IFMT) | (la->la_mode & ~S_IFMT); osa->mode = obj->oo_attr.la_mode; - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(uos), NULL, + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(osd), NULL, &osa->mode, 8); } - if (la->la_valid & LA_SIZE) { + if (valid & LA_SIZE) { osa->size = obj->oo_attr.la_size = la->la_size; - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(uos), NULL, + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(osd), NULL, &osa->size, 8); } - if (la->la_valid & LA_NLINK) { + if (valid & LA_NLINK) { osa->nlink = obj->oo_attr.la_nlink = la->la_nlink; - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(uos), NULL, + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(osd), NULL, &osa->nlink, 8); } - if (la->la_valid & LA_RDEV) { + if (valid & LA_RDEV) { osa->rdev = obj->oo_attr.la_rdev = la->la_rdev; - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(uos), NULL, + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(osd), NULL, &osa->rdev, 8); } - if (la->la_valid & LA_FLAGS) { - osa->flags = obj->oo_attr.la_flags = la->la_flags; - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(uos), NULL, + if (valid & LA_FLAGS) { + osa->flags = attrs_fs2zfs(la->la_flags); + /* many flags are not supported by zfs, so ensure a good cached + * copy */ + obj->oo_attr.la_flags = attrs_zfs2fs(osa->flags); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(osd), NULL, &osa->flags, 8); } - if (la->la_valid & LA_UID) { + if (valid & LA_UID) { osa->uid = obj->oo_attr.la_uid = la->la_uid; - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(uos), NULL, + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(osd), NULL, &osa->uid, 8); } - if (la->la_valid & LA_GID) { + if (valid & LA_GID) { osa->gid = obj->oo_attr.la_gid = la->la_gid; - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(uos), NULL, + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(osd), NULL, &osa->gid, 8); } - obj->oo_attr.la_valid |= la->la_valid; + obj->oo_attr.la_valid |= valid; write_unlock(&obj->oo_attr_lock); rc = osd_object_sa_bulk_update(obj, bulk, cnt, oh); @@ -1016,11 +1048,12 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt, static void osd_ah_init(const struct lu_env *env, struct dt_allocation_hint *ah, struct dt_object *parent, struct dt_object *child, - cfs_umode_t child_mode) + umode_t child_mode) { LASSERT(ah); memset(ah, 0, sizeof(*ah)); + ah->dah_parent = parent; ah->dah_mode = child_mode; } @@ -1037,6 +1070,7 @@ static int osd_declare_object_create(const struct lu_env *env, struct osd_device *osd = osd_obj2dev(obj); struct osd_thandle *oh; uint64_t zapid; + int rc; ENTRY; LASSERT(dof); @@ -1058,6 +1092,7 @@ static int osd_declare_object_create(const struct lu_env *env, switch (dof->dof_type) { case DFT_DIR: + dt->do_index_ops = &osd_dir_ops; case DFT_INDEX: /* for zap create */ dmu_tx_hold_zap(oh->ot_tx, DMU_NEW_OBJECT, 1, NULL); @@ -1090,18 +1125,19 @@ static int osd_declare_object_create(const struct lu_env *env, __osd_xattr_declare_set(env, obj, sizeof(struct lustre_mdt_attrs), XATTR_NAME_LMA, oh); - RETURN(osd_declare_quota(env, osd, attr->la_uid, attr->la_gid, 1, oh, - false, NULL, false)); + rc = osd_declare_quota(env, osd, attr->la_uid, attr->la_gid, 1, oh, + false, NULL, false); + RETURN(rc); } -int __osd_attr_init(const struct lu_env *env, udmu_objset_t *uos, - uint64_t oid, dmu_tx_t *tx, struct lu_attr *la) +int __osd_attr_init(const struct lu_env *env, struct osd_device *osd, + uint64_t oid, dmu_tx_t *tx, struct lu_attr *la, + uint64_t parent) { sa_bulk_attr_t *bulk; sa_handle_t *sa_hdl; struct osa_attr *osa = &osd_oti_get(env)->oti_osa; uint64_t gen; - uint64_t parent; uint64_t crtime[2]; timestruc_t now; int cnt; @@ -1111,9 +1147,6 @@ int __osd_attr_init(const struct lu_env *env, udmu_objset_t *uos, gen = dmu_tx_get_txg(tx); ZFS_TIME_ENCODE(&now, crtime); - /* XXX: this should be real id of parent for ZPL access, but we have no - * such info in OSD, probably it can be part of dt_object_format */ - parent = 0; osa->atime[0] = la->la_atime; osa->ctime[0] = la->la_ctime; @@ -1123,11 +1156,11 @@ int __osd_attr_init(const struct lu_env *env, udmu_objset_t *uos, osa->gid = la->la_gid; osa->rdev = la->la_rdev; osa->nlink = la->la_nlink; - osa->flags = la->la_flags; + osa->flags = attrs_fs2zfs(la->la_flags); osa->size = la->la_size; /* Now add in all of the "SA" attributes */ - rc = -sa_handle_get(uos->os, oid, NULL, SA_HDL_PRIVATE, &sa_hdl); + rc = -sa_handle_get(osd->od_os, oid, NULL, SA_HDL_PRIVATE, &sa_hdl); if (rc) return rc; @@ -1147,19 +1180,19 @@ int __osd_attr_init(const struct lu_env *env, udmu_objset_t *uos, * work around the problem. See ORI-610. */ cnt = 0; - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(uos), NULL, &osa->mode, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(uos), NULL, &osa->size, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GEN(uos), NULL, &gen, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(uos), NULL, &osa->uid, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(uos), NULL, &osa->gid, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PARENT(uos), NULL, &parent, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(uos), NULL, &osa->flags, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(uos), NULL, osa->atime, 16); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(uos), NULL, osa->mtime, 16); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(uos), NULL, osa->ctime, 16); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CRTIME(uos), NULL, crtime, 16); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(uos), NULL, &osa->nlink, 8); - SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(uos), NULL, &osa->rdev, 8); + 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, &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); + SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(osd), NULL, &osa->rdev, 8); rc = -sa_replace_all_by_template(sa_hdl, bulk, cnt, tx); @@ -1174,34 +1207,36 @@ out: * dmu_tx_hold_bonus(tx, DMU_NEW_OBJECT) called and then assigned * to a transaction group. */ -int __osd_object_create(const struct lu_env *env, udmu_objset_t *uos, - dmu_buf_t **dbp, dmu_tx_t *tx, - struct lu_attr *la, void *tag) +int __osd_object_create(const struct lu_env *env, struct osd_device *osd, + dmu_buf_t **dbp, dmu_tx_t *tx, struct lu_attr *la, + uint64_t parent) { uint64_t oid; int rc; - LASSERT(tag); - spin_lock(&uos->lock); - uos->objects++; - spin_unlock(&uos->lock); - /* Assert that the transaction has been assigned to a transaction group. */ LASSERT(tx->tx_txg != 0); /* Create a new DMU object. */ - oid = dmu_object_alloc(uos->os, DMU_OT_PLAIN_FILE_CONTENTS, 0, + oid = dmu_object_alloc(osd->od_os, DMU_OT_PLAIN_FILE_CONTENTS, 0, DMU_OT_SA, DN_MAX_BONUSLEN, tx); - rc = -sa_buf_hold(uos->os, oid, tag, dbp); - if (rc) - return rc; + rc = -sa_buf_hold(osd->od_os, oid, osd_obj_tag, dbp); + LASSERTF(rc == 0, "sa_buf_hold "LPU64" failed: %d\n", oid, rc); LASSERT(la->la_valid & LA_MODE); la->la_size = 0; la->la_nlink = 1; - return __osd_attr_init(env, uos, oid, tx, la); + rc = __osd_attr_init(env, osd, oid, tx, la, parent); + if (rc != 0) { + sa_buf_rele(*dbp, osd_obj_tag); + *dbp = NULL; + dmu_object_free(osd->od_os, oid, tx); + return rc; + } + + return 0; } /* @@ -1214,28 +1249,24 @@ int __osd_object_create(const struct lu_env *env, udmu_objset_t *uos, * will also require a FAT ZAP. If there is a new type of micro ZAP created * then we might need to re-evaluate the use of this flag and instead do * a conversion from the different internal ZAP hash formats being used. */ -int __osd_zap_create(const struct lu_env *env, udmu_objset_t *uos, +int __osd_zap_create(const struct lu_env *env, struct osd_device *osd, dmu_buf_t **zap_dbp, dmu_tx_t *tx, - struct lu_attr *la, void *tag, zap_flags_t flags) + struct lu_attr *la, uint64_t parent, zap_flags_t flags) { uint64_t oid; int rc; - LASSERT(tag); - - spin_lock(&uos->lock); - uos->objects++; - spin_unlock(&uos->lock); - /* Assert that the transaction has been assigned to a transaction group. */ LASSERT(tx->tx_txg != 0); - oid = zap_create_flags(uos->os, 0, flags | ZAP_FLAG_HASH64, - DMU_OT_DIRECTORY_CONTENTS, 12, 12, + oid = zap_create_flags(osd->od_os, 0, flags | ZAP_FLAG_HASH64, + DMU_OT_DIRECTORY_CONTENTS, + 14, /* == ZFS fzap_default_block_shift */ + DN_MAX_INDBLKSHIFT, /* indirect block shift */ DMU_OT_SA, DN_MAX_BONUSLEN, tx); - rc = -sa_buf_hold(uos->os, oid, tag, zap_dbp); + rc = -sa_buf_hold(osd->od_os, oid, osd_obj_tag, zap_dbp); if (rc) return rc; @@ -1243,11 +1274,12 @@ int __osd_zap_create(const struct lu_env *env, udmu_objset_t *uos, la->la_size = 2; la->la_nlink = 1; - return __osd_attr_init(env, uos, oid, tx, la); + return __osd_attr_init(env, osd, oid, tx, la, parent); } static dmu_buf_t *osd_mkidx(const struct lu_env *env, struct osd_device *osd, - struct lu_attr *la, struct osd_thandle *oh) + struct lu_attr *la, uint64_t parent, + struct osd_thandle *oh) { dmu_buf_t *db; int rc; @@ -1257,36 +1289,36 @@ static dmu_buf_t *osd_mkidx(const struct lu_env *env, struct osd_device *osd, * We set ZAP_FLAG_UINT64_KEY to let ZFS know than we are going to use * binary keys */ LASSERT(S_ISREG(la->la_mode)); - rc = __osd_zap_create(env, &osd->od_objset, &db, oh->ot_tx, la, - osd_obj_tag, ZAP_FLAG_UINT64_KEY); + rc = __osd_zap_create(env, osd, &db, oh->ot_tx, la, parent, + ZAP_FLAG_UINT64_KEY); if (rc) return ERR_PTR(rc); return db; } static dmu_buf_t *osd_mkdir(const struct lu_env *env, struct osd_device *osd, - struct lu_attr *la, struct osd_thandle *oh) + struct lu_attr *la, uint64_t parent, + struct osd_thandle *oh) { dmu_buf_t *db; int rc; LASSERT(S_ISDIR(la->la_mode)); - rc = __osd_zap_create(env, &osd->od_objset, &db, oh->ot_tx, la, - osd_obj_tag, 0); + rc = __osd_zap_create(env, osd, &db, oh->ot_tx, la, parent, 0); if (rc) return ERR_PTR(rc); return db; } static dmu_buf_t* osd_mkreg(const struct lu_env *env, struct osd_device *osd, - struct lu_attr *la, struct osd_thandle *oh) + struct lu_attr *la, uint64_t parent, + struct osd_thandle *oh) { dmu_buf_t *db; int rc; LASSERT(S_ISREG(la->la_mode)); - rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la, - osd_obj_tag); + rc = __osd_object_create(env, osd, &db, oh->ot_tx, la, parent); if (rc) return ERR_PTR(rc); @@ -1295,7 +1327,7 @@ static dmu_buf_t* osd_mkreg(const struct lu_env *env, struct osd_device *osd, * a method in OSD API to control this from OFD/MDD */ if (!lu_device_is_md(osd2lu_dev(osd))) { - rc = -dmu_object_set_blocksize(osd->od_objset.os, + rc = -dmu_object_set_blocksize(osd->od_os, db->db_object, 128 << 10, 0, oh->ot_tx); if (unlikely(rc)) { @@ -1309,21 +1341,22 @@ static dmu_buf_t* osd_mkreg(const struct lu_env *env, struct osd_device *osd, } static dmu_buf_t *osd_mksym(const struct lu_env *env, struct osd_device *osd, - struct lu_attr *la, struct osd_thandle *oh) + struct lu_attr *la, uint64_t parent, + struct osd_thandle *oh) { dmu_buf_t *db; int rc; LASSERT(S_ISLNK(la->la_mode)); - rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la, - osd_obj_tag); + rc = __osd_object_create(env, osd, &db, oh->ot_tx, la, parent); if (rc) return ERR_PTR(rc); return db; } static dmu_buf_t *osd_mknod(const struct lu_env *env, struct osd_device *osd, - struct lu_attr *la, struct osd_thandle *oh) + struct lu_attr *la, uint64_t parent, + struct osd_thandle *oh) { dmu_buf_t *db; int rc; @@ -1332,15 +1365,17 @@ static dmu_buf_t *osd_mknod(const struct lu_env *env, struct osd_device *osd, if (S_ISCHR(la->la_mode) || S_ISBLK(la->la_mode)) la->la_valid |= LA_RDEV; - rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la, - osd_obj_tag); + rc = __osd_object_create(env, osd, &db, oh->ot_tx, la, parent); if (rc) return ERR_PTR(rc); return db; } -typedef dmu_buf_t *(*osd_obj_type_f)(const struct lu_env *env, struct osd_device *osd, - struct lu_attr *la, struct osd_thandle *oh); +typedef dmu_buf_t *(*osd_obj_type_f)(const struct lu_env *env, + struct osd_device *osd, + struct lu_attr *la, + uint64_t parent, + struct osd_thandle *oh); static osd_obj_type_f osd_create_type_f(enum dt_format_type type) { @@ -1380,7 +1415,7 @@ static inline int osd_init_lma(const struct lu_env *env, struct osd_object *obj, struct lu_buf buf; int rc; - lustre_lma_init(lma, fid); + lustre_lma_init(lma, fid, 0, 0); lustre_lma_swab(lma); buf.lb_buf = lma; buf.lb_len = sizeof(*lma); @@ -1430,9 +1465,15 @@ static int osd_object_create(const struct lu_env *env, struct dt_object *dt, LASSERT(obj->oo_db == NULL); - db = osd_create_type_f(dof->dof_type)(env, osd, attr, oh); + /* to follow ZFS on-disk format we need + * to initialize parent dnode properly */ + zapid = 0; + if (hint && hint->dah_parent) + zapid = osd_dt_obj(hint->dah_parent)->oo_db->db_object; + + db = osd_create_type_f(dof->dof_type)(env, osd, attr, zapid, oh); if (IS_ERR(db)) - GOTO(out, rc = PTR_ERR(th)); + GOTO(out, rc = PTR_ERR(db)); zde->zde_pad = 0; zde->zde_dnode = db->db_object; @@ -1440,19 +1481,19 @@ static int osd_object_create(const struct lu_env *env, struct dt_object *dt, zapid = osd_get_name_n_idx(env, osd, fid, buf); - rc = -zap_add(osd->od_objset.os, zapid, buf, 8, 1, zde, oh->ot_tx); + rc = -zap_add(osd->od_os, zapid, buf, 8, 1, zde, oh->ot_tx); if (rc) GOTO(out, rc); /* Add new object to inode accounting. * Errors are not considered as fatal */ - rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid, + rc = -zap_increment_int(osd->od_os, osd->od_iusr_oid, (attr->la_valid & LA_UID) ? attr->la_uid : 0, 1, oh->ot_tx); if (rc) CERROR("%s: failed to add "DFID" to accounting ZAP for usr %d " "(%d)\n", osd->od_svname, PFID(fid), attr->la_uid, rc); - rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid, + rc = -zap_increment_int(osd->od_os, osd->od_igrp_oid, (attr->la_valid & LA_GID) ? attr->la_gid : 0, 1, oh->ot_tx); if (rc) @@ -1495,7 +1536,6 @@ static int osd_object_ref_add(const struct lu_env *env, struct osd_object *obj = osd_dt_obj(dt); struct osd_thandle *oh; struct osd_device *osd = osd_obj2dev(obj); - udmu_objset_t *uos = &osd->od_objset; uint64_t nlink; int rc; @@ -1511,7 +1551,7 @@ static int osd_object_ref_add(const struct lu_env *env, nlink = ++obj->oo_attr.la_nlink; write_unlock(&obj->oo_attr_lock); - rc = osd_object_sa_update(obj, SA_ZPL_LINKS(uos), &nlink, 8, oh); + rc = osd_object_sa_update(obj, SA_ZPL_LINKS(osd), &nlink, 8, oh); return rc; } @@ -1532,7 +1572,6 @@ static int osd_object_ref_del(const struct lu_env *env, struct osd_object *obj = osd_dt_obj(dt); struct osd_thandle *oh; struct osd_device *osd = osd_obj2dev(obj); - udmu_objset_t *uos = &osd->od_objset; uint64_t nlink; int rc; @@ -1549,7 +1588,7 @@ static int osd_object_ref_del(const struct lu_env *env, nlink = --obj->oo_attr.la_nlink; write_unlock(&obj->oo_attr_lock); - rc = osd_object_sa_update(obj, SA_ZPL_LINKS(uos), &nlink, 8, oh); + rc = osd_object_sa_update(obj, SA_ZPL_LINKS(osd), &nlink, 8, oh); return rc; } @@ -1692,14 +1731,17 @@ static struct obd_capa *osd_capa_get(const struct lu_env *env, RETURN(oc); } -static int osd_object_sync(const struct lu_env *env, struct dt_object *dt) +static int osd_object_sync(const struct lu_env *env, struct dt_object *dt, + __u64 start, __u64 end) { struct osd_device *osd = osd_obj2dev(osd_dt_obj(dt)); ENTRY; /* XXX: no other option than syncing the whole filesystem until we - * support ZIL */ - txg_wait_synced(dmu_objset_pool(osd->od_objset.os), 0ULL); + * support ZIL. If the object tracked the txg that it was last + * modified in, it could pass that txg here instead of "0". Maybe + * the changes are already committed, so no wait is needed at all? */ + txg_wait_synced(dmu_objset_pool(osd->od_os), 0ULL); RETURN(0); }