X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fosd-zfs%2Fosd_object.c;h=b50258559696056d57d57cbb91d2383bf0a908c0;hb=a95e9e1c4f845249f83498f9650aa9c49a0dca60;hp=bd128515fcb7b1821ebc1cc837ee5127681f5f61;hpb=d948d10f0d68f8675cd8f80032642382c7811846;p=fs%2Flustre-release.git

diff --git a/lustre/osd-zfs/osd_object.c b/lustre/osd-zfs/osd_object.c
index bd12851..b502585 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) 2011, 2012 Whamcloud, Inc.
+ * Copyright (c) 2012, 2013, Intel Corporation.
  * Use is subject to license terms.
  */
 /*
@@ -42,14 +42,10 @@
  * Author: Johann Lombardi <johann@whamcloud.com>
  */
 
-#ifndef EXPORT_SYMTAB
-# define EXPORT_SYMTAB
-#endif
 #define DEBUG_SUBSYSTEM S_OSD
 
 #include <lustre_ver.h>
 #include <libcfs/libcfs.h>
-#include <lustre_fsfilt.h>
 #include <obd_support.h>
 #include <lustre_net.h>
 #include <obd.h>
@@ -79,7 +75,7 @@ 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)
@@ -124,10 +120,12 @@ osd_object_sa_dirty_add(struct osd_object *obj, struct osd_thandle *oh)
 	if (!cfs_list_empty(&obj->oo_sa_linkage))
 		return;
 
-	cfs_down(&oh->ot_sa_lock);
+	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);
-	cfs_up(&oh->ot_sa_lock);
+	write_unlock(&obj->oo_attr_lock);
+	up(&oh->ot_sa_lock);
 }
 
 /*
@@ -137,14 +135,16 @@ void osd_object_sa_dirty_rele(struct osd_thandle *oh)
 {
 	struct osd_object *obj;
 
-	cfs_down(&oh->ot_sa_lock);
+	down(&oh->ot_sa_lock);
 	while (!cfs_list_empty(&oh->ot_sa_list)) {
 		obj = cfs_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);
+		write_unlock(&obj->oo_attr_lock);
 	}
-	cfs_up(&oh->ot_sa_lock);
+	up(&oh->ot_sa_lock);
 }
 
 /*
@@ -230,7 +230,7 @@ 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)) {
@@ -286,7 +286,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;
 
@@ -295,9 +295,9 @@ struct lu_object *osd_object_alloc(const struct lu_env *env,
 		mo->oo_dt.do_ops = &osd_obj_ops;
 		l->lo_ops = &osd_lu_obj_ops;
 		CFS_INIT_LIST_HEAD(&mo->oo_sa_linkage);
-		cfs_init_rwsem(&mo->oo_sem);
-		cfs_sema_init(&mo->oo_guard, 1);
-		cfs_rwlock_init(&mo->oo_attr_lock);
+		init_rwsem(&mo->oo_sem);
+		sema_init(&mo->oo_guard, 1);
+		rwlock_init(&mo->oo_attr_lock);
 		return l;
 	} else {
 		return NULL;
@@ -339,12 +339,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.
@@ -360,27 +393,35 @@ 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 {
+		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);
 }
 
@@ -458,6 +499,7 @@ static int osd_declare_object_destroy(const struct lu_env *env,
 	struct osd_device	*osd = osd_obj2dev(obj);
 	struct osd_thandle	*oh;
 	uint64_t		 zapid;
+	int			 rc;
 	ENTRY;
 
 	LASSERT(th != NULL);
@@ -480,15 +522,24 @@ static int osd_declare_object_destroy(const struct lu_env *env,
 	dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
 	dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, 0, buf);
 
-	RETURN(0);
+	/* one less inode */
+	rc = osd_declare_quota(env, osd, obj->oo_attr.la_uid,
+			       obj->oo_attr.la_gid, -1, oh, false, NULL, false);
+	if (rc)
+		RETURN(rc);
+
+	/* data to be truncated */
+	rc = osd_declare_quota(env, osd, obj->oo_attr.la_uid,
+			       obj->oo_attr.la_gid, 0, oh, true, NULL, false);
+	RETURN(rc);
 }
 
 int __osd_object_free(udmu_objset_t *uos, uint64_t oid, dmu_tx_t *tx)
 {
 	LASSERT(uos->objects != 0);
-	cfs_spin_lock(&uos->lock);
+	spin_lock(&uos->lock);
 	uos->objects--;
-	cfs_spin_unlock(&uos->lock);
+	spin_unlock(&uos->lock);
 
 	return -dmu_object_free(uos->os, oid, tx);
 }
@@ -654,7 +705,7 @@ static void osd_object_read_lock(const struct lu_env *env,
 
 	LASSERT(osd_invariant(obj));
 
-	cfs_down_read(&obj->oo_sem);
+	down_read(&obj->oo_sem);
 }
 
 static void osd_object_write_lock(const struct lu_env *env,
@@ -664,7 +715,7 @@ static void osd_object_write_lock(const struct lu_env *env,
 
 	LASSERT(osd_invariant(obj));
 
-	cfs_down_write(&obj->oo_sem);
+	down_write(&obj->oo_sem);
 }
 
 static void osd_object_read_unlock(const struct lu_env *env,
@@ -673,7 +724,7 @@ static void osd_object_read_unlock(const struct lu_env *env,
 	struct osd_object *obj = osd_dt_obj(dt);
 
 	LASSERT(osd_invariant(obj));
-	cfs_up_read(&obj->oo_sem);
+	up_read(&obj->oo_sem);
 }
 
 static void osd_object_write_unlock(const struct lu_env *env,
@@ -682,7 +733,7 @@ static void osd_object_write_unlock(const struct lu_env *env,
         struct osd_object *obj = osd_dt_obj(dt);
 
         LASSERT(osd_invariant(obj));
-        cfs_up_write(&obj->oo_sem);
+	up_write(&obj->oo_sem);
 }
 
 static int osd_object_write_locked(const struct lu_env *env,
@@ -693,9 +744,9 @@ static int osd_object_write_locked(const struct lu_env *env,
 
 	LASSERT(osd_invariant(obj));
 
-	if (cfs_down_write_trylock(&obj->oo_sem)) {
+	if (down_write_trylock(&obj->oo_sem)) {
 		rc = 0;
-		cfs_up_write(&obj->oo_sem);
+		up_write(&obj->oo_sem);
 	}
 	return rc;
 }
@@ -713,14 +764,18 @@ static int osd_attr_get(const struct lu_env *env,
 	LASSERT(osd_invariant(obj));
 	LASSERT(obj->oo_db);
 
-	cfs_read_lock(&obj->oo_attr_lock);
+	read_lock(&obj->oo_attr_lock);
 	*attr = obj->oo_attr;
-	cfs_read_unlock(&obj->oo_attr_lock);
+	read_unlock(&obj->oo_attr_lock);
 
 	/* with ZFS_DEBUG zrl_add_debug() called by DB_DNODE_ENTER()
 	 * from within sa_object_size() can block on a mutex, so
 	 * we can't call sa_object_size() holding rwlock */
 	sa_object_size(obj->oo_sa_hdl, &blksize, &blocks);
+	/* we do not control size of indices, so always calculate
+	 * it from number of blocks reported by DMU */
+	if (S_ISDIR(attr->la_mode))
+		attr->la_size = 512 * blocks;
 	/* Block size may be not set; suggest maximal I/O transfers. */
 	if (blksize == 0)
 		blksize = 1ULL << SPA_MAXBLOCKSHIFT;
@@ -732,15 +787,80 @@ static int osd_attr_get(const struct lu_env *env,
 	return 0;
 }
 
+/* Simple wrapper on top of qsd API which implement quota transfer for osd
+ * setattr needs. As a reminder, only the root user can change ownership of
+ * a file, that's why EDQUOT & EINPROGRESS errors are discarded */
+static inline int qsd_transfer(const struct lu_env *env,
+			       struct qsd_instance *qsd,
+			       struct lquota_trans *trans, int qtype,
+			       __u64 orig_id, __u64 new_id, __u64 bspace,
+			       struct lquota_id_info *qi)
+{
+	int	rc;
+
+	if (unlikely(qsd == NULL))
+		return 0;
+
+	LASSERT(qtype >= 0 && qtype < MAXQUOTAS);
+	qi->lqi_type = qtype;
+
+	/* inode accounting */
+	qi->lqi_is_blk = false;
+
+	/* one more inode for the new owner ... */
+	qi->lqi_id.qid_uid = new_id;
+	qi->lqi_space      = 1;
+	rc = qsd_op_begin(env, qsd, trans, qi, NULL);
+	if (rc == -EDQUOT || rc == -EINPROGRESS)
+		rc = 0;
+	if (rc)
+		return rc;
+
+	/* 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 == -EINPROGRESS)
+		rc = 0;
+	if (rc)
+		return rc;
+
+	/* block accounting */
+	qi->lqi_is_blk = true;
+
+	/* more blocks for the new owner ... */
+	qi->lqi_id.qid_uid = new_id;
+	qi->lqi_space      = bspace;
+	rc = qsd_op_begin(env, qsd, trans, qi, NULL);
+	if (rc == -EDQUOT || rc == -EINPROGRESS)
+		rc = 0;
+	if (rc)
+		return rc;
+
+	/* and finally less blocks for the current owner */
+	qi->lqi_id.qid_uid = orig_id;
+	qi->lqi_space      = -bspace;
+	rc = qsd_op_begin(env, qsd, trans, qi, NULL);
+	/* can't get EDQUOT when reducing usage */
+	if (rc == -EINPROGRESS)
+		rc = 0;
+	return rc;
+}
+
 static int osd_declare_attr_set(const struct lu_env *env,
 				struct dt_object *dt,
 				const struct lu_attr *attr,
 				struct thandle *handle)
 {
+	struct osd_thread_info	*info = osd_oti_get(env);
 	char			*buf = osd_oti_get(env)->oti_str;
 	struct osd_object	*obj = osd_dt_obj(dt);
 	struct osd_device	*osd = osd_obj2dev(obj);
 	struct osd_thandle	*oh;
+	uint64_t		 bspace;
+	uint32_t		 blksize;
+	int			 rc;
 	ENTRY;
 
 	if (!dt_object_exists(dt)) {
@@ -754,17 +874,43 @@ 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);
 
 	if (attr && attr->la_valid & LA_UID) {
 		/* account for user inode tracking ZAP update */
 		dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
 		dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, TRUE, buf);
+
+		/* quota enforcement for user */
+		if (attr->la_uid != obj->oo_attr.la_uid) {
+			rc = qsd_transfer(env, osd->od_quota_slave,
+					  &oh->ot_quota_trans, USRQUOTA,
+					  obj->oo_attr.la_uid, attr->la_uid,
+					  bspace, &info->oti_qi);
+			if (rc)
+				RETURN(rc);
+		}
 	}
 	if (attr && attr->la_valid & LA_GID) {
 		/* account for user inode tracking ZAP update */
 		dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
 		dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, TRUE, buf);
+
+		/* quota enforcement for group */
+		if (attr->la_gid != obj->oo_attr.la_gid) {
+			rc = qsd_transfer(env, osd->od_quota_slave,
+					  &oh->ot_quota_trans, GRPQUOTA,
+					  obj->oo_attr.la_gid, attr->la_gid,
+					  bspace, &info->oti_qi);
+			if (rc)
+				RETURN(rc);
+		}
 	}
 
 	RETURN(0);
@@ -840,7 +986,7 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
 				obj->oo_attr.la_gid, rc);
 	}
 
-	cfs_write_lock(&obj->oo_attr_lock);
+	write_lock(&obj->oo_attr_lock);
 	cnt = 0;
 	if (la->la_valid & LA_ATIME) {
 		osa->atime[0] = obj->oo_attr.la_atime = la->la_atime;
@@ -881,7 +1027,10 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
 				 &osa->rdev, 8);
 	}
 	if (la->la_valid & LA_FLAGS) {
-		osa->flags = obj->oo_attr.la_flags = la->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(uos), NULL,
 				 &osa->flags, 8);
 	}
@@ -896,7 +1045,7 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
 				 &osa->gid, 8);
 	}
 	obj->oo_attr.la_valid |= la->la_valid;
-	cfs_write_unlock(&obj->oo_attr_lock);
+	write_unlock(&obj->oo_attr_lock);
 
 	rc = osd_object_sa_bulk_update(obj, bulk, cnt, oh);
 
@@ -911,11 +1060,13 @@ 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, cfs_umode_t child_mode)
+			struct dt_object *parent, struct dt_object *child,
+			umode_t child_mode)
 {
 	LASSERT(ah);
 
 	memset(ah, 0, sizeof(*ah));
+	ah->dah_parent = parent;
 	ah->dah_mode = child_mode;
 }
 
@@ -932,6 +1083,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);
@@ -982,17 +1134,21 @@ static int osd_declare_object_create(const struct lu_env *env,
 
 	dmu_tx_hold_sa_create(oh->ot_tx, ZFS_SA_BASE_ATTR_SIZE);
 
-	RETURN(0);
+	__osd_xattr_declare_set(env, obj, sizeof(struct lustre_mdt_attrs),
+				XATTR_NAME_LMA, oh);
+
+	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, udmu_objset_t *uos, 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;
@@ -1002,9 +1158,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;
@@ -1014,7 +1167,7 @@ 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 */
@@ -1066,17 +1219,13 @@ out:
  * 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)
+			dmu_buf_t **dbp, dmu_tx_t *tx, struct lu_attr *la,
+			uint64_t parent, void *tag)
 {
 	uint64_t oid;
 	int	 rc;
 
 	LASSERT(tag);
-	cfs_spin_lock(&uos->lock);
-	uos->objects++;
-	cfs_spin_unlock(&uos->lock);
-
 	/* Assert that the transaction has been assigned to a
 	   transaction group. */
 	LASSERT(tx->tx_txg != 0);
@@ -1085,14 +1234,24 @@ int __osd_object_create(const struct lu_env *env, udmu_objset_t *uos,
 	oid = dmu_object_alloc(uos->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;
+	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, uos, oid, tx, la, parent);
+	if (rc != 0) {
+		sa_buf_rele(*dbp, tag);
+		*dbp = NULL;
+		dmu_object_free(uos->os, oid, tx);
+		return rc;
+	}
+
+	spin_lock(&uos->lock);
+	uos->objects++;
+	spin_unlock(&uos->lock);
+	return 0;
 }
 
 /*
@@ -1107,16 +1266,17 @@ int __osd_object_create(const struct lu_env *env, udmu_objset_t *uos,
  * a conversion from the different internal ZAP hash formats being used. */
 int __osd_zap_create(const struct lu_env *env, udmu_objset_t *uos,
 		     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,
+		     void *tag, zap_flags_t flags)
 {
 	uint64_t oid;
 	int	 rc;
 
 	LASSERT(tag);
 
-	cfs_spin_lock(&uos->lock);
+	spin_lock(&uos->lock);
 	uos->objects++;
-	cfs_spin_unlock(&uos->lock);
+	spin_unlock(&uos->lock);
 
 	/* Assert that the transaction has been assigned to a
 	   transaction group. */
@@ -1134,11 +1294,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, uos, 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;
@@ -1149,35 +1310,37 @@ static dmu_buf_t *osd_mkidx(const struct lu_env *env, struct osd_device *osd,
 	 * 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);
+			      parent, osd_obj_tag, 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);
+			      parent, osd_obj_tag, 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);
+				 parent, osd_obj_tag);
 	if (rc)
 		return ERR_PTR(rc);
 
@@ -1200,21 +1363,23 @@ 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);
+				 parent, osd_obj_tag);
 	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;
@@ -1224,14 +1389,17 @@ static dmu_buf_t *osd_mknod(const struct lu_env *env, struct osd_device *osd,
 		la->la_valid |= LA_RDEV;
 
 	rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
-				 osd_obj_tag);
+				 parent, osd_obj_tag);
 	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)
 {
@@ -1263,6 +1431,24 @@ static osd_obj_type_f osd_create_type_f(enum dt_format_type type)
 /*
  * Primitives for directory (i.e. ZAP) handling
  */
+static inline int osd_init_lma(const struct lu_env *env, struct osd_object *obj,
+			       const struct lu_fid *fid, struct osd_thandle *oh)
+{
+	struct osd_thread_info	*info = osd_oti_get(env);
+	struct lustre_mdt_attrs	*lma = &info->oti_mdt_attrs;
+	struct lu_buf		 buf;
+	int rc;
+
+	lustre_lma_init(lma, fid, 0, 0);
+	lustre_lma_swab(lma);
+	buf.lb_buf = lma;
+	buf.lb_len = sizeof(*lma);
+
+	rc = osd_xattr_set_internal(env, obj, &buf, XATTR_NAME_LMA,
+				    LU_XATTR_CREATE, oh, BYPASS_CAPA);
+
+	return rc;
+}
 
 /*
  * Concurrency: @dt is write locked.
@@ -1288,7 +1474,7 @@ 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 */
-	cfs_down(&obj->oo_guard);
+	down(&obj->oo_guard);
 
 	LASSERT(osd_invariant(obj));
 	LASSERT(!dt_object_exists(dt));
@@ -1303,9 +1489,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;
@@ -1338,8 +1530,16 @@ static int osd_object_create(const struct lu_env *env, struct dt_object *dt,
 	LASSERT(ergo(rc == 0, dt_object_exists(dt)));
 	LASSERT(osd_invariant(obj));
 
+	rc = osd_init_lma(env, obj, fid, oh);
+	if (rc) {
+		CERROR("%s: can not set LMA on "DFID": rc = %d\n",
+		       osd->od_svname, PFID(fid), rc);
+		/* ignore errors during LMA initialization */
+		rc = 0;
+	}
+
 out:
-	cfs_up(&obj->oo_guard);
+	up(&obj->oo_guard);
 	RETURN(rc);
 }
 
@@ -1372,9 +1572,9 @@ static int osd_object_ref_add(const struct lu_env *env,
 
 	oh = container_of0(handle, struct osd_thandle, ot_super);
 
-	cfs_write_lock(&obj->oo_attr_lock);
+	write_lock(&obj->oo_attr_lock);
 	nlink = ++obj->oo_attr.la_nlink;
-	cfs_write_unlock(&obj->oo_attr_lock);
+	write_unlock(&obj->oo_attr_lock);
 
 	rc = osd_object_sa_update(obj, SA_ZPL_LINKS(uos), &nlink, 8, oh);
 	return rc;
@@ -1410,9 +1610,9 @@ static int osd_object_ref_del(const struct lu_env *env,
 	oh = container_of0(handle, struct osd_thandle, ot_super);
 	LASSERT(!lu_object_is_dying(dt->do_lu.lo_header));
 
-	cfs_write_lock(&obj->oo_attr_lock);
+	write_lock(&obj->oo_attr_lock);
 	nlink = --obj->oo_attr.la_nlink;
-	cfs_write_unlock(&obj->oo_attr_lock);
+	write_unlock(&obj->oo_attr_lock);
 
 	rc = osd_object_sa_update(obj, SA_ZPL_LINKS(uos), &nlink, 8, oh);
 	return rc;
@@ -1436,14 +1636,14 @@ static int capa_is_sane(const struct lu_env *env, struct osd_device *dev,
 		RETURN(rc);
 	}
 
-	cfs_spin_lock(&capa_lock);
+	spin_lock(&capa_lock);
 	for (i = 0; i < 2; i++) {
 		if (keys[i].lk_keyid == capa->lc_keyid) {
 			oti->oti_capa_key = keys[i];
 			break;
 		}
 	}
-	cfs_spin_unlock(&capa_lock);
+	spin_unlock(&capa_lock);
 
 	if (i == 2) {
 		DEBUG_CAPA(D_ERROR, capa, "no matched capa key");
@@ -1539,9 +1739,9 @@ static struct obd_capa *osd_capa_get(const struct lu_env *env,
 		RETURN(oc);
 	}
 
-	cfs_spin_lock(&capa_lock);
+	spin_lock(&capa_lock);
 	*key = dev->od_capa_keys[1];
-	cfs_spin_unlock(&capa_lock);
+	spin_unlock(&capa_lock);
 
 	capa->lc_keyid = key->lk_keyid;
 	capa->lc_expiry = cfs_time_current_sec() + dev->od_capa_timeout;
@@ -1557,13 +1757,16 @@ 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 */
+	 * 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_objset.os), 0ULL);
 
 	RETURN(0);