static int osd_attr_get(const struct lu_env *env,
			struct dt_object *dt,
			struct lu_attr *attr)
{
	struct osd_object	*obj = osd_dt_obj(dt);
	uint64_t		 blocks;
	uint32_t		 blksize;

	LASSERT(dt_object_exists(dt));
	LASSERT(osd_invariant(obj));
	LASSERT(obj->oo_db);

	read_lock(&obj->oo_attr_lock);
	*attr = obj->oo_attr;
	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 = osd_spa_maxblocksize(
			dmu_objset_spa(osd_obj2dev(obj)->od_os));

	attr->la_blksize = blksize;
	attr->la_blocks = blocks;
	attr->la_valid |= LA_BLOCKS | LA_BLKSIZE;

	return 0;
}

/**
 * Restore iterator from cookie. if the \a hash isn't found,
 * restore the first valid record.
 *
 * \param  di    - osd iterator
 * \param  hash  - iterator location cookie
 *
 * \retval +ve  - di points to exact matched key
 * \retval  0   - di points to the first valid record
 * \retval -ve  - failure
 */
static int osd_it_acct_load(const struct lu_env *env,
			    const struct dt_it *di, __u64 hash)
{
	struct osd_it_quota	*it  = (struct osd_it_quota *)di;
	struct osd_device	*osd = osd_obj2dev(it->oiq_obj);
	zap_attribute_t		*za = &osd_oti_get(env)->oti_za;
	zap_cursor_t		*zc;
	int			 rc;
	ENTRY;

	/* create new cursor pointing to the new hash */
	rc = osd_zap_cursor_init(&zc, osd->od_os, it->oiq_oid, hash);
	if (rc)
		RETURN(rc);
	osd_zap_cursor_fini(it->oiq_zc);
	it->oiq_zc = zc;
	it->oiq_reset = 0;

	rc = -zap_cursor_retrieve(it->oiq_zc, za);
	if (rc == 0)
		rc = 1;
	else if (rc == -ENOENT)
		rc = 0;

	RETURN(rc);
}

static dmu_buf_t *osd_mkreg(const struct lu_env *env, struct osd_object *obj,
			    struct lu_attr *la, uint64_t parent,
			    struct osd_thandle *oh)
{
	dmu_buf_t	  *db;
	int		   rc;
	struct osd_device *osd = osd_obj2dev(obj);

	LASSERT(S_ISREG(la->la_mode));
	rc = __osd_object_create(env, obj, &db, oh->ot_tx, la, parent);
	if (rc)
		return ERR_PTR(rc);

	if (!lu_device_is_md(osd2lu_dev(osd))) {
		/* uses 4K as default block size because clients write data
		 * with page size that is 4K at minimum */
		rc = -dmu_object_set_blocksize(osd->od_os, db->db_object,
					       4096, 0, oh->ot_tx);
		if (unlikely(rc)) {
			CERROR("%s: can't change blocksize: %d\n",
			       osd->od_svname, rc);
			return ERR_PTR(rc);
		}
	}

	return db;
}

/*
 * Concurrency: @dt is write locked.
 */
static int osd_object_ref_del(const struct lu_env *env,
			      struct dt_object *dt,
			      struct thandle *handle)
{
	struct osd_object	*obj = osd_dt_obj(dt);
	struct osd_thandle	*oh;
	struct osd_device	*osd = osd_obj2dev(obj);
	uint64_t		 nlink;
	int			 rc;

	ENTRY;

	LASSERT(osd_invariant(obj));
	LASSERT(dt_object_exists(dt));
	LASSERT(obj->oo_sa_hdl != NULL);

	oh = container_of0(handle, struct osd_thandle, ot_super);
	LASSERT(!lu_object_is_dying(dt->do_lu.lo_header));

	write_lock(&obj->oo_attr_lock);
	nlink = --obj->oo_attr.la_nlink;
	write_unlock(&obj->oo_attr_lock);

	rc = osd_object_sa_update(obj, SA_ZPL_LINKS(osd), &nlink, 8, oh);
	return rc;
}

int osd_xattr_del(const struct lu_env *env, struct dt_object *dt,
		const char *name, struct thandle *handle,
		struct lustre_capa *capa)
{
	struct osd_object  *obj = osd_dt_obj(dt);
	struct osd_thandle *oh;
	int                 rc;
	ENTRY;

	LASSERT(handle != NULL);
	LASSERT(obj->oo_db != NULL);
	LASSERT(osd_invariant(obj));
	LASSERT(dt_object_exists(dt));
	oh = container_of0(handle, struct osd_thandle, ot_super);
	LASSERT(oh->ot_tx != NULL);

	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);
	rc = __osd_xattr_del(env, obj, name, oh);
	up(&obj->oo_guard);

	RETURN(rc);
}

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);
	udmu_objset_t     *uos = &osd->od_objset;
	uint64_t           xa_data_obj;
	int                rc;

	/* try remove xattr from SA at first */
	rc = __osd_sa_xattr_del(env, obj, name, oh);
	if (rc != -ENOENT)
		return rc;

	if (obj->oo_xattr == ZFS_NO_OBJECT)
		return 0;

	rc = -zap_lookup(uos->os, obj->oo_xattr, name, sizeof(uint64_t), 1,
			&xa_data_obj);
	if (rc == -ENOENT) {
		rc = 0;
	} else if (rc == 0) {
		/*
		 * Entry exists.
		 * We'll delete the existing object and ZAP entry.
		 */
		rc = __osd_object_free(uos, xa_data_obj, oh->ot_tx);
		if (rc)
			return rc;

		rc = -zap_remove(uos->os, obj->oo_xattr, name, oh->ot_tx);
	}

	return rc;
}

int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
		  const struct lu_buf *buf, const char *name, int fl,
		  struct thandle *handle, struct lustre_capa *capa)
{
	struct osd_object  *obj = osd_dt_obj(dt);
	struct osd_thandle *oh;
	int rc = 0;
	ENTRY;

	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 ||
	     strcmp(name, POSIX_ACL_XATTR_DEFAULT) == 0))
		RETURN(-EOPNOTSUPP);

	oh = container_of0(handle, struct osd_thandle, ot_super);

	down(&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, capa);
	up(&obj->oo_guard);

	RETURN(rc);
}

static int osd_index_it_load(const struct lu_env *env, const struct dt_it *di,
			     __u64 hash)
{
	struct osd_zap_it *it = (struct osd_zap_it *)di;
	struct osd_object *obj = it->ozi_obj;
	struct osd_device *osd = osd_obj2dev(obj);
	zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
	int                rc;
	ENTRY;

	/* close the current cursor */
	zap_cursor_fini(it->ozi_zc);

	/* create a new one starting at hash */
	memset(it->ozi_zc, 0, sizeof(*it->ozi_zc));
	zap_cursor_init_serialized(it->ozi_zc, osd->od_objset.os,
				   obj->oo_db->db_object, hash);
	it->ozi_reset = 0;

	rc = -zap_cursor_retrieve(it->ozi_zc, za);
	if (rc == 0)
		RETURN(+1);
	else if (rc == -ENOENT)
		RETURN(0);

	RETURN(rc);
}

int osd_xattr_get(const struct lu_env *env, struct dt_object *dt,
		struct lu_buf *buf, const char *name,
		struct lustre_capa *capa)
{
	struct osd_object  *obj  = osd_dt_obj(dt);
	int                 rc, size = 0;
	ENTRY;

	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);
	rc = __osd_xattr_get(env, obj, buf, name, &size);
	up(&obj->oo_guard);

	if (rc == -ENOENT)
		rc = -ENODATA;
	else if (rc == 0)
		rc = size;
	RETURN(rc);
}

static dmu_buf_t *osd_mkreg(const struct lu_env *env, struct osd_object *obj,
			    struct lu_attr *la, uint64_t parent,
			    struct osd_thandle *oh)
{
	dmu_buf_t	  *db;
	int		   rc;
	struct osd_device *osd = osd_obj2dev(obj);

	LASSERT(S_ISREG(la->la_mode));
	rc = __osd_object_create(env, obj, &db, oh->ot_tx, la, parent);
	if (rc)
		return ERR_PTR(rc);

	/*
	 * XXX: This heuristic is non-optimal.  It would be better to
	 * increase the blocksize up to osd->od_max_blksz during the write.
	 * This is exactly how the ZPL behaves and it ensures that the right
	 * blocksize is selected based on the file size rather than the
	 * making broad assumptions based on the osd type.
	 */
	if (!lu_device_is_md(osd2lu_dev(osd))) {
		rc = -dmu_object_set_blocksize(osd->od_os, db->db_object,
					       osd->od_max_blksz, 0, oh->ot_tx);
		if (unlikely(rc)) {
			CERROR("%s: can't change blocksize: %d\n",
			       osd->od_svname, rc);
			return ERR_PTR(rc);
		}
	}

	return db;
}

static struct dt_it *osd_index_it_init(const struct lu_env *env,
				       struct dt_object *dt,
				       __u32 unused,
				       struct lustre_capa *capa)
{
	struct osd_thread_info  *info = osd_oti_get(env);
	struct osd_zap_it       *it;
	struct osd_object       *obj = osd_dt_obj(dt);
	struct osd_device       *osd = osd_obj2dev(obj);
	struct lu_object        *lo  = &dt->do_lu;
	ENTRY;

	/* XXX: check capa ? */

	LASSERT(lu_object_exists(lo));
	LASSERT(obj->oo_db);
	LASSERT(udmu_object_is_zap(obj->oo_db));
	LASSERT(info);

	it = &info->oti_it_zap;

	if (udmu_zap_cursor_init(&it->ozi_zc, &osd->od_objset,
				 obj->oo_db->db_object, 0))
		RETURN(ERR_PTR(-ENOMEM));

	it->ozi_obj   = obj;
	it->ozi_capa  = capa;
	it->ozi_reset = 1;
	lu_object_get(lo);

	RETURN((struct dt_it *)it);
}

/*
 * return status :
 *  rc == 0 -> end of directory.
 *  rc >  0 -> ok, proceed.
 *  rc <  0 -> error.  ( EOVERFLOW  can be masked.)
 */
static int osd_dir_it_load(const struct lu_env *env,
			const struct dt_it *di, __u64 hash)
{
	struct osd_zap_it *it = (struct osd_zap_it *)di;
	struct osd_object *obj = it->ozi_obj;
	struct osd_device *osd = osd_obj2dev(obj);
	zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
	int		   rc;
	ENTRY;

	udmu_zap_cursor_fini(it->ozi_zc);
	if (udmu_zap_cursor_init(&it->ozi_zc, &osd->od_objset,
				 obj->oo_db->db_object, hash))
		RETURN(-ENOMEM);

	if (hash <= 2) {
		it->ozi_pos = hash;
		rc = +1;
	} else {
		it->ozi_pos = 3;
		/* to return whether the end has been reached */
		rc = osd_index_retrieve_skip_dots(it, za);
		if (rc == 0)
			rc = +1;
		else if (rc == -ENOENT)
			rc = 0;
	}

	RETURN(rc);
}

/*
 * Set an extended attribute.
 * This transaction must have called udmu_xattr_declare_set() first.
 *
 * Returns 0 on success or a negative error number on failure.
 *
 * No locking is done here.
 */
static int
__osd_sa_xattr_update(const struct lu_env *env, struct osd_object *obj,
			struct osd_thandle *oh)
{
	struct osd_device *osd = osd_obj2dev(obj);
	udmu_objset_t     *uos = &osd->od_objset;
	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 = sa_spill_alloc(KM_SLEEP);
	if (dxattr == NULL)
		RETURN(-ENOMEM);

	rc = -nvlist_pack(obj->oo_sa_xattr, &dxattr, &sa_size,
				NV_ENCODE_XDR, KM_SLEEP);
	if (rc)
		GOTO(out_free, rc);

	rc = osd_object_sa_update(obj, SA_ZPL_DXATTR(uos), dxattr, sa_size, oh);
out_free:
	sa_spill_free(dxattr);
	RETURN(rc);
}

static int osd_dir_lookup(const struct lu_env *env, struct dt_object *dt,
			  struct dt_rec *rec, const struct dt_key *key,
			  struct lustre_capa *capa)
{
	struct osd_thread_info *oti = osd_oti_get(env);
	struct osd_object  *obj = osd_dt_obj(dt);
	struct osd_device  *osd = osd_obj2dev(obj);
	char		   *name = (char *)key;
	int                 rc;
	ENTRY;

	LASSERT(udmu_object_is_zap(obj->oo_db));

	if (name[0] == '.') {
		if (name[1] == 0) {
			const struct lu_fid *f = lu_object_fid(&dt->do_lu);
			memcpy(rec, f, sizeof(*f));
			RETURN(1);
		} else if (name[1] == '.' && name[2] == 0) {
			rc = osd_find_parent_fid(env, dt, (struct lu_fid *)rec);
			RETURN(rc == 0 ? 1 : rc);
		}
	}

	rc = -zap_lookup(osd->od_objset.os, obj->oo_db->db_object,
			 (char *)key, 8, sizeof(oti->oti_zde) / 8,
			 (void *)&oti->oti_zde);
	memcpy(rec, &oti->oti_zde.lzd_fid, sizeof(struct lu_fid));

	RETURN(rc == 0 ? 1 : rc);
}

static int osd_index_insert(const struct lu_env *env, struct dt_object *dt,
			    const struct dt_rec *rec, const struct dt_key *key,
			    struct thandle *th, struct lustre_capa *capa,
			    int ignore_quota)
{
	struct osd_object  *obj = osd_dt_obj(dt);
	struct osd_device  *osd = osd_obj2dev(obj);
	struct osd_thandle *oh;
	__u64		   *k = osd_oti_get(env)->oti_key64;
	int                 rc;
	ENTRY;

	LASSERT(obj->oo_db);
	LASSERT(dt_object_exists(dt));
	LASSERT(osd_invariant(obj));
	LASSERT(th != NULL);

	oh = container_of0(th, struct osd_thandle, ot_super);

	rc = osd_prepare_key_uint64(obj, k, key);

	/* Insert (key,oid) into ZAP */
	rc = -zap_add_uint64(osd->od_objset.os, obj->oo_db->db_object,
			     k, rc, obj->oo_recusize, obj->oo_recsize,
			     (void *)rec, oh->ot_tx);
	RETURN(rc);
}

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);
	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: shouldn't matter.
 */
int osd_object_init0(const struct lu_env *env, struct osd_object *obj)
{
	struct osd_device	*osd = osd_obj2dev(obj);
	const struct lu_fid	*fid = lu_object_fid(&obj->oo_dt.do_lu);
	int			 rc = 0;
	ENTRY;

	if (obj->oo_db == NULL)
		RETURN(0);

	/* object exist */

	rc = osd_object_sa_init(obj, osd);
	if (rc)
		RETURN(rc);

	/* cache attrs in object */
	rc = __osd_object_attr_get(env, osd, obj, &obj->oo_attr);
	if (rc)
		RETURN(rc);

	if (likely(!fid_is_acct(fid)))
		/* no body operations for accounting objects */
		obj->oo_dt.do_body_ops = &osd_body_ops;

	/*
	 * initialize object before marking it existing
	 */
	obj->oo_dt.do_lu.lo_header->loh_attr |= obj->oo_attr.la_mode & S_IFMT;

	smp_mb();
	obj->oo_dt.do_lu.lo_header->loh_attr |= LOHA_EXISTS;

	RETURN(0);
}

/*
 * Concurrency: @dt is write locked.
 */
static int osd_object_ref_add(const struct lu_env *env,
			      struct dt_object *dt,
			      struct thandle *handle)
{
	struct osd_object	*obj = osd_dt_obj(dt);
	struct osd_thandle	*oh;
	struct osd_device	*osd = osd_obj2dev(obj);
	uint64_t		 nlink;
	int rc;

	ENTRY;

	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);

	oh = container_of0(handle, struct osd_thandle, ot_super);

	write_lock(&obj->oo_attr_lock);
	nlink = ++obj->oo_attr.la_nlink;
	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);
}

/**
 * Return space usage consumed by a given uid or gid.
 * Block usage is accurrate since it is maintained by DMU itself.
 * However, DMU does not provide inode accounting, so the #inodes in use
 * is estimated from the block usage and statfs information.
 *
 * \param env   - is the environment passed by the caller
 * \param dtobj - is the accounting object
 * \param dtrec - is the record to fill with space usage information
 * \param dtkey - is the id the of the user or group for which we would
 *                like to access disk usage.
 * \param capa - is the capability, not used.
 *
 * \retval +ve - success : exact match
 * \retval -ve - failure
 */
static int osd_acct_index_lookup(const struct lu_env *env,
				struct dt_object *dtobj,
				struct dt_rec *dtrec,
				const struct dt_key *dtkey,
				struct lustre_capa *capa)
{
	struct osd_thread_info	*info = osd_oti_get(env);
	char			*buf  = info->oti_buf;
	struct lquota_acct_rec	*rec  = (struct lquota_acct_rec *)dtrec;
	struct osd_object	*obj = osd_dt_obj(dtobj);
	struct osd_device	*osd = osd_obj2dev(obj);
	int			 rc;
	uint64_t		 oid;
	ENTRY;

	rec->bspace = rec->ispace = 0;

	/* convert the 64-bit uid/gid into a string */
	sprintf(buf, "%llx", *((__u64 *)dtkey));
	/* fetch DMU object ID (DMU_USERUSED_OBJECT/DMU_GROUPUSED_OBJECT) to be
	 * used */
	oid = osd_quota_fid2dmu(lu_object_fid(&dtobj->do_lu));

	/* disk usage (in bytes) is maintained by DMU.
	 * DMU_USERUSED_OBJECT/DMU_GROUPUSED_OBJECT are special objects which
	 * not associated with any dmu_but_t (see dnode_special_open()).
	 * As a consequence, we cannot use udmu_zap_lookup() here since it
	 * requires a valid oo_db. */
	rc = -zap_lookup(osd->od_objset.os, oid, buf, sizeof(uint64_t), 1,
			&rec->bspace);
	if (rc == -ENOENT)
		/* user/group has not created anything yet */
		CDEBUG(D_QUOTA, "%s: id %s not found in DMU accounting ZAP\n",
		       osd->od_svname, buf);
	else if (rc)
		RETURN(rc);

	if (osd->od_quota_iused_est) {
		if (rec->bspace != 0)
			/* estimate #inodes in use */
			rec->ispace = udmu_objset_user_iused(&osd->od_objset,
							     rec->bspace);
		RETURN(+1);
	}

	/* as for inode accounting, it is not maintained by DMU, so we just
	 * use our own ZAP to track inode usage */
	rc = -zap_lookup(osd->od_objset.os, obj->oo_db->db_object,
			 buf, sizeof(uint64_t), 1, &rec->ispace);
	if (rc == -ENOENT)
		/* user/group has not created any file yet */
		CDEBUG(D_QUOTA, "%s: id %s not found in accounting ZAP\n",
		       osd->od_svname, buf);
	else if (rc)
		RETURN(rc);

	RETURN(+1);
}

static inline int __osd_xattr_cache(const struct lu_env *env,
				    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);
}