/**
 * Implementation of the llog_operations::lop_create
 *
 * This function creates the llog according with llog_handle::lgh_obj
 * and llog_handle::lgh_name.
 *
 * \param[in] env	execution environment
 * \param[in] res	llog handle of the current llog
 * \param[in] th	current transaction handle
 *
 * \retval		0 on successful create
 * \retval		negative value on error
 */
static int llog_osd_create(const struct lu_env *env, struct llog_handle *res,
			   struct thandle *th)
{
	struct llog_thread_info *lgi = llog_info(env);
	struct dt_insert_rec	*rec = &lgi->lgi_dt_rec;
	struct local_oid_storage *los;
	struct dt_object        *o;
	int                      rc = 0;

	ENTRY;

	LASSERT(env);
	o = res->lgh_obj;
	LASSERT(o);

	/* llog can be already created */
	if (dt_object_exists(o))
		RETURN(-EEXIST);

	los = res->private_data;
	LASSERT(los);

	dt_write_lock(env, o, 0);
	if (!dt_object_exists(o))
		rc = llog_osd_create_new_object(env, los, o, th);
	else
		rc = -EEXIST;

	dt_write_unlock(env, o);
	if (rc)
		RETURN(rc);

	if (res->lgh_name) {
		struct dt_object *llog_dir;

		llog_dir = llog_osd_dir_get(env, res->lgh_ctxt);
		if (IS_ERR(llog_dir))
			RETURN(PTR_ERR(llog_dir));

		logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
		rec->rec_fid = &lgi->lgi_fid;
		rec->rec_type = S_IFREG;
		dt_read_lock(env, llog_dir, 0);
		rc = dt_insert(env, llog_dir, (struct dt_rec *)rec,
			       (struct dt_key *)res->lgh_name,
			       th, BYPASS_CAPA, 1);
		dt_read_unlock(env, llog_dir);
		lu_object_put(env, &llog_dir->do_lu);
		if (rc)
			CERROR("%s: can't create named llog %s: rc = %d\n",
			       o->do_lu.lo_dev->ld_obd->obd_name,
			       res->lgh_name, rc);
	}
	RETURN(rc);
}

/*
 * 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;

	LASSERT(osd_invariant(obj));
	LASSERT(dt_object_exists(dt));
	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);
	return rc;
}

/*
 * Reserve enough credits to update a record in a quota index file.
 *
 * \param env - is the environment passed by the caller
 * \param th  - is the transaction to use for disk writes
 * \param obj - is the on-disk index where quota settings are stored.
 * \param id  - is the key to be updated
 *
 * \retval    - 0 on success, appropriate error on failure
 */
int lquota_disk_declare_write(const struct lu_env *env, struct thandle *th,
			      struct dt_object *obj, union lquota_id *id)
{
	struct lquota_thread_info	*qti = lquota_info(env);
	struct dt_key			*key = (struct dt_key *)&id->qid_uid;
	int				 rc;
	ENTRY;

	LASSERT(dt_object_exists(obj));
	LASSERT(obj->do_index_ops != NULL);

	/* speculative delete declaration in case there is already an existing
	 * record in the index */
	rc = dt_declare_delete(env, obj, key, th);
	if (rc)
		RETURN(rc);

	/* declare insertion of updated record */
	rc = dt_declare_insert(env, obj, (struct dt_rec *)&qti->qti_rec, key,
			       th);
	if (rc)
		RETURN(rc);

	/* we might have to update the version of the global index too */
	rc = dt_declare_version_set(env, obj, th);

	RETURN(rc);
}

int osd_xattr_get(const struct lu_env *env, struct dt_object *dt,
		  struct lu_buf *buf, const char *name)
{
	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);
}

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

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

	RETURN(rc);
}

/*
 * 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;

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

out:
	up_read(&obj->oo_guard);
	RETURN(rc);
}

/**
 * Look-up accounting object to collect space usage information for user
 * or group.
 *
 * \param env  - is the environment passed by the caller
 * \param dev  - is the dt_device storing the accounting object
 * \param type - is the quota type, either USRQUOTA or GRPQUOTA
 */
struct dt_object *acct_obj_lookup(const struct lu_env *env,
				  struct dt_device *dev, int type)
{
	struct lquota_thread_info	*qti = lquota_info(env);
	struct dt_object		*obj = NULL;
	ENTRY;

	lu_local_obj_fid(&qti->qti_fid, qtype2acct_oid(type));

	/* lookup the accounting object */
	obj = dt_locate(env, dev, &qti->qti_fid);
	if (IS_ERR(obj))
		RETURN(obj);

	if (!dt_object_exists(obj)) {
		dt_object_put(env, obj);
		RETURN(ERR_PTR(-ENOENT));
	}

	if (obj->do_index_ops == NULL) {
		int rc;

		/* set up indexing operations */
		rc = obj->do_ops->do_index_try(env, obj, &dt_acct_features);
		if (rc) {
			CERROR("%s: failed to set up indexing operations for %s"
			       " acct object rc:%d\n",
			       dev->dd_lu_dev.ld_obd->obd_name,
			       qtype_name(type), rc);
			dt_object_put(env, obj);
			RETURN(ERR_PTR(rc));
		}
	}
	RETURN(obj);
}

int osd_xattr_del(const struct lu_env *env, struct dt_object *dt,
		  const char *name, struct thandle *handle)
{
	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_write(&obj->oo_guard);
	rc = __osd_xattr_del(env, obj, name, oh);
	up_write(&obj->oo_guard);

	RETURN(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);
}

/**
 * Write the special file which contains the list of llog catalogs IDs
 *
 * This function writes the CATALOG file which contains the array of llog
 * catalogs IDs. It is used mostly to store OSP llogs indexed by OST/MDT
 * number.
 *
 * \param[in]  env	execution environment
 * \param[in]  d	corresponding storage device
 * \param[in]  idx	position to start from, usually OST/MDT index
 * \param[in]  count	how many catalog IDs to write
 * \param[out] idarray	the buffer with the data to write.
 * \param[in]  fid	LLOG_CATALOGS_OID for CATALOG object
 *
 * \retval		0 on successful write of catalog IDs
 * \retval		negative value on error
 */
int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d,
			  int idx, int count, struct llog_catid *idarray,
			  const struct lu_fid *fid)
{
	struct llog_thread_info	*lgi = llog_info(env);
	struct dt_object	*o = NULL;
	struct thandle		*th;
	int			 rc, size;

	if (count == 0)
		RETURN(0);

	LASSERT(d);

	size = sizeof(*idarray) * count;
	lgi->lgi_off = idx * sizeof(*idarray);
	lgi->lgi_fid = *fid;

	o = dt_locate(env, d, &lgi->lgi_fid);
	if (IS_ERR(o))
		RETURN(PTR_ERR(o));

	if (!dt_object_exists(o))
		GOTO(out, rc = -ENOENT);

	rc = dt_attr_get(env, o, &lgi->lgi_attr, BYPASS_CAPA);
	if (rc)
		GOTO(out, rc);

	if (!S_ISREG(lgi->lgi_attr.la_mode)) {
		CERROR("%s: CATALOGS is not a regular file!: mode = %o\n",
		       o->do_lu.lo_dev->ld_obd->obd_name,
		       lgi->lgi_attr.la_mode);
		GOTO(out, rc = -ENOENT);
	}

	th = dt_trans_create(env, d);
	if (IS_ERR(th))
		GOTO(out, rc = PTR_ERR(th));

	lgi->lgi_buf.lb_len = size;
	lgi->lgi_buf.lb_buf = idarray;
	rc = dt_declare_record_write(env, o, &lgi->lgi_buf, lgi->lgi_off, th);
	if (rc)
		GOTO(out, rc);

	rc = dt_trans_start_local(env, d, th);
	if (rc)
		GOTO(out_trans, rc);

	rc = dt_record_write(env, o, &lgi->lgi_buf, &lgi->lgi_off, th);
	if (rc)
		CDEBUG(D_INODE, "can't write CATALOGS at index %d: rc = %d\n",
		       idx, rc);
out_trans:
	dt_trans_stop(env, d, th);
out:
	lu_object_put(env, &o->do_lu);
	RETURN(rc);
}

int osd_index_try(const struct lu_env *env, struct dt_object *dt,
		const struct dt_index_features *feat)
{
	struct osd_object *obj = osd_dt_obj(dt);
	ENTRY;

	LASSERT(dt_object_exists(dt));

	/*
	 * XXX: implement support for fixed-size keys sorted with natural
	 *      numerical way (not using internal hash value)
	 */
	if (feat->dif_flags & DT_IND_RANGE)
		RETURN(-ERANGE);

	if (unlikely(feat == &dt_otable_features))
		/* do not support oi scrub yet. */
		RETURN(-ENOTSUPP);

	LASSERT(obj->oo_db != NULL);
	if (likely(feat == &dt_directory_features)) {
		if (udmu_object_is_zap(obj->oo_db))
			dt->do_index_ops = &osd_dir_ops;
		else
			RETURN(-ENOTDIR);
	} else if (unlikely(feat == &dt_acct_features)) {
		LASSERT(fid_is_acct(lu_object_fid(&dt->do_lu)));
		dt->do_index_ops = &osd_acct_index_ops;
	} else if (udmu_object_is_zap(obj->oo_db) &&
		   dt->do_index_ops == NULL) {
		/* For index file, we don't support variable key & record sizes
		 * and the key has to be unique */
		if ((feat->dif_flags & ~DT_IND_UPDATE) != 0)
			RETURN(-EINVAL);

		if (feat->dif_keysize_max > ZAP_MAXNAMELEN)
			RETURN(-E2BIG);
		if (feat->dif_keysize_max != feat->dif_keysize_min)
			RETURN(-EINVAL);

		/* As for the record size, it should be a multiple of 8 bytes
		 * and smaller than the maximum value length supported by ZAP.
		 */
		if (feat->dif_recsize_max > ZAP_MAXVALUELEN)
			RETURN(-E2BIG);
		if (feat->dif_recsize_max != feat->dif_recsize_min)
			RETURN(-EINVAL);

		obj->oo_keysize = feat->dif_keysize_max;
		obj->oo_recsize = feat->dif_recsize_max;
		obj->oo_recusize = 1;

		/* ZFS prefers to work with array of 64bits */
		if ((obj->oo_recsize & 7) == 0) {
			obj->oo_recsize >>= 3;
			obj->oo_recusize = 8;
		}
		dt->do_index_ops = &osd_index_ops;
	}

static int osp_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
				const struct lu_attr *attr, struct thandle *th)
{
	struct osp_device	*d = lu2osp_dev(dt->do_lu.lo_dev);
	struct osp_object	*o = dt2osp_obj(dt);
	int			 rc = 0;

	ENTRY;

	/*
	 * Usually we don't allow server stack to manipulate size
	 * but there is a special case when striping is created
	 * late, after stripless file got truncated to non-zero.
	 *
	 * In this case we do the following:
	 *
	 * 1) grab id in declare - this can lead to leaked OST objects
	 *    but we don't currently have proper mechanism and the only
	 *    options we have are to do truncate RPC holding transaction
	 *    open (very bad) or to grab id in declare at cost of leaked
	 *    OST object in same very rare unfortunate case (just bad)
	 *    notice 1.6-2.0 do assignment outside of running transaction
	 *    all the time, meaning many more chances for leaked objects.
	 *
	 * 2) send synchronous truncate RPC with just assigned id
	 */

	/* there are few places in MDD code still passing NULL
	 * XXX: to be fixed soon */
	if (attr == NULL)
		RETURN(0);

	if (attr->la_valid & LA_SIZE && attr->la_size > 0 &&
	    fid_is_zero(lu_object_fid(&o->opo_obj.do_lu))) {
		LASSERT(!dt_object_exists(dt));
		osp_object_assign_fid(env, d, o);
		rc = osp_object_truncate(env, dt, attr->la_size);
		if (rc)
			RETURN(rc);
	}

	if (o->opo_new) {
		/* no need in logging for new objects being created */
		RETURN(0);
	}

	if (!(attr->la_valid & (LA_UID | LA_GID)))
		RETURN(0);

	/*
	 * track all UID/GID changes via llog
	 */
	rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);

	RETURN(rc);
}

static int osd_declare_object_destroy(const struct lu_env *env,
				      struct dt_object *dt,
				      struct thandle *th)
{
	char			*buf = osd_oti_get(env)->oti_str;
	const struct lu_fid	*fid = lu_object_fid(&dt->do_lu);
	struct osd_object	*obj = osd_dt_obj(dt);
	struct osd_device	*osd = osd_obj2dev(obj);
	struct osd_thandle	*oh;
	int			 rc;
	uint64_t		 zapid;
	ENTRY;

	LASSERT(th != NULL);
	LASSERT(dt_object_exists(dt));

	oh = container_of0(th, struct osd_thandle, ot_super);
	LASSERT(oh->ot_tx != NULL);

	/* declare that we'll remove object from fid-dnode mapping */
	zapid = osd_get_name_n_idx(env, osd, fid, buf);
	dmu_tx_hold_bonus(oh->ot_tx, zapid);
	dmu_tx_hold_zap(oh->ot_tx, zapid, FALSE, buf);

	osd_declare_xattrs_destroy(env, obj, oh);

	/* declare that we'll remove object from inode accounting ZAPs */
	dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
	dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, FALSE, buf);
	dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
	dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, FALSE, buf);

	/* 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);
	if (rc)
		RETURN(rc);

	osd_object_set_destroy_type(obj);
	if (obj->oo_destroy == OSD_DESTROY_SYNC)
		dmu_tx_hold_free(oh->ot_tx, obj->oo_db->db_object,
				 0, DMU_OBJECT_END);
	else
		dmu_tx_hold_zap(oh->ot_tx, osd->od_unlinkedid, TRUE, NULL);

	RETURN(0);
}

/**
 * Implementation of the llog_operations::lop_declare_create
 *
 * This function declares the llog create. It declares also name insert
 * into llog directory in case of named llog.
 *
 * \param[in] env	execution environment
 * \param[in] res	llog handle of the current llog
 * \param[in] th	current transaction handle
 *
 * \retval		0 on successful create declaration
 * \retval		negative value on error
 */
static int llog_osd_declare_create(const struct lu_env *env,
				   struct llog_handle *res, struct thandle *th)
{
	struct llog_thread_info		*lgi = llog_info(env);
	struct dt_insert_rec		*rec = &lgi->lgi_dt_rec;
	struct local_oid_storage	*los;
	struct dt_object		*o;
	int				 rc;

	ENTRY;

	LASSERT(res->lgh_obj);
	LASSERT(th);

	/* object can be created by another thread */
	o = res->lgh_obj;
	if (dt_object_exists(o))
		RETURN(0);

	los = res->private_data;
	LASSERT(los);

	rc = llog_osd_declare_new_object(env, los, o, th);
	if (rc)
		RETURN(rc);

	/* do not declare header initialization here as it's declared
	 * in llog_osd_declare_write_rec() which is always called */

	if (res->lgh_name) {
		struct dt_object *llog_dir;

		llog_dir = llog_osd_dir_get(env, res->lgh_ctxt);
		if (IS_ERR(llog_dir))
			RETURN(PTR_ERR(llog_dir));
		logid_to_fid(&res->lgh_id, &lgi->lgi_fid);
		rec->rec_fid = &lgi->lgi_fid;
		rec->rec_type = S_IFREG;
		rc = dt_declare_insert(env, llog_dir,
				       (struct dt_rec *)rec,
				       (struct dt_key *)res->lgh_name, th);
		lu_object_put(env, &llog_dir->do_lu);
		if (rc)
			CERROR("%s: can't declare named llog %s: rc = %d\n",
			       o->do_lu.lo_dev->ld_obd->obd_name,
			       res->lgh_name, rc);
	}
	RETURN(rc);
}

/**
 * Copy an extended attribute into the buffer provided, or compute
 * the required buffer size if \a buf is NULL.
 *
 * On success, the number of bytes used or required is stored in \a sizep.
 *
 * Note that no locking is done here.
 *
 * \param[in] env      execution environment
 * \param[in] obj      object for which to retrieve xattr
 * \param[out] buf     buffer to store xattr value in
 * \param[in] name     name of xattr to copy
 * \param[out] sizep   bytes used or required to store xattr
 *
 * \retval 0           on success
 * \retval negative    negated errno on failure
 */
int __osd_xattr_get(const struct lu_env *env, struct osd_object *obj,
		    struct lu_buf *buf, const char *name, int *sizep)
{
	int rc;

	if (unlikely(!dt_object_exists(&obj->oo_dt) || obj->oo_destroyed))
		return -ENOENT;

	/* check SA_ZPL_DXATTR first then fallback to directory xattr */
	rc = __osd_sa_xattr_get(env, obj, buf, name, sizep);
	if (rc != -ENOENT)
		return rc;

	return __osd_xattr_get_large(env, osd_obj2dev(obj), obj->oo_xattr,
				     buf, name, sizep);
}

/*
 * Retrieve quota settings from disk for a particular identifier.
 *
 * \param env - is the environment passed by the caller
 * \param obj - is the on-disk index where quota settings are stored.
 * \param id  - is the key to be updated
 * \param rec - is the output record where to store quota settings.
 *
 * \retval    - 0 on success, appropriate error on failure
 */
int lquota_disk_read(const struct lu_env *env, struct dt_object *obj,
		     union lquota_id *id, struct dt_rec *rec)
{
	int	rc;
	ENTRY;

	LASSERT(dt_object_exists(obj));
	LASSERT(obj->do_index_ops != NULL);

	/* lookup on-disk record from index file */
	dt_read_lock(env, obj, 0);
	rc = dt_lookup(env, obj, rec, (struct dt_key *)&id->qid_uid);
	dt_read_unlock(env, obj);

	RETURN(rc);
}

static ssize_t osd_write(const struct lu_env *env, struct dt_object *dt,
			const struct lu_buf *buf, loff_t *pos,
			struct thandle *th, int ignore_quota)
{
	struct osd_object  *obj  = osd_dt_obj(dt);
	struct osd_device  *osd = osd_obj2dev(obj);
	struct osd_thandle *oh;
	uint64_t            offset = *pos;
	int                 rc;

	ENTRY;

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

	LASSERT(th != NULL);
	oh = container_of0(th, struct osd_thandle, ot_super);

	record_start_io(osd, WRITE, 0);

	dmu_write(osd->od_os, obj->oo_db->db_object, offset,
		(uint64_t)buf->lb_len, buf->lb_buf, oh->ot_tx);
	write_lock(&obj->oo_attr_lock);
	if (obj->oo_attr.la_size < offset + buf->lb_len) {
		obj->oo_attr.la_size = offset + buf->lb_len;
		write_unlock(&obj->oo_attr_lock);
		/* osd_object_sa_update() will be copying directly from oo_attr
		 * into dbuf.  any update within a single txg will copy the
		 * most actual */
		rc = osd_object_sa_update(obj, SA_ZPL_SIZE(osd),
					&obj->oo_attr.la_size, 8, oh);
		if (unlikely(rc))
			GOTO(out, rc);
	} else {
		write_unlock(&obj->oo_attr_lock);
	}

	*pos += buf->lb_len;
	rc = buf->lb_len;

out:
	record_end_io(osd, WRITE, 0, buf->lb_len,
		      buf->lb_len >> PAGE_CACHE_SHIFT);

	RETURN(rc);
}

static ssize_t osd_declare_write(const struct lu_env *env, struct dt_object *dt,
				const struct lu_buf *buf, loff_t pos,
				struct thandle *th)
{
	struct osd_object  *obj  = osd_dt_obj(dt);
	struct osd_device  *osd = osd_obj2dev(obj);
	struct osd_thandle *oh;
	uint64_t            oid;
	ENTRY;

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

	/* in some cases declare can race with creation (e.g. llog)
	 * and we need to wait till object is initialized. notice
	 * LOHA_EXISTs is supposed to be the last step in the
	 * initialization */

	/* declare possible size change. notice we can't check
	 * current size here as another thread can change it */

	if (dt_object_exists(dt)) {
		LASSERT(obj->oo_db);
		oid = obj->oo_db->db_object;

		dmu_tx_hold_sa(oh->ot_tx, obj->oo_sa_hdl, 0);
	} else {
		oid = DMU_NEW_OBJECT;
		dmu_tx_hold_sa_create(oh->ot_tx, ZFS_SA_BASE_ATTR_SIZE);
	}

	/* XXX: we still miss for append declaration support in ZFS
	 *	-1 means append which is used by llog mostly, llog
	 *	can grow upto LLOG_MIN_CHUNK_SIZE*8 records */
	if (pos == -1)
		pos = max_t(loff_t, 256 * 8 * LLOG_MIN_CHUNK_SIZE,
			    obj->oo_attr.la_size + (2 << 20));
	dmu_tx_hold_write(oh->ot_tx, oid, pos, buf->lb_len);

	/* dt_declare_write() is usually called for system objects, such
	 * as llog or last_rcvd files. We needn't enforce quota on those
	 * objects, so always set the lqi_space as 0. */
	RETURN(osd_declare_quota(env, osd, obj->oo_attr.la_uid,
				 obj->oo_attr.la_gid, 0, oh, true, NULL,
				 false));
}

int llog_destroy(const struct lu_env *env, struct llog_handle *handle)
{
	struct llog_operations	*lop;
	struct dt_device	*dt;
	struct thandle		*th;
	int rc;

	ENTRY;

	rc = llog_handle2ops(handle, &lop);
	if (rc < 0)
		RETURN(rc);
	if (lop->lop_destroy == NULL)
		RETURN(-EOPNOTSUPP);

	if (handle->lgh_obj == NULL) {
		/* if lgh_obj == NULL, then it is from client side destroy */
		rc = lop->lop_destroy(env, handle, NULL);
		RETURN(rc);
	}

	if (!dt_object_exists(handle->lgh_obj))
		RETURN(0);

	dt = lu2dt_dev(handle->lgh_obj->do_lu.lo_dev);

	th = dt_trans_create(env, dt);
	if (IS_ERR(th))
		RETURN(PTR_ERR(th));

	rc = llog_declare_destroy(env, handle, th);
	if (rc != 0)
		GOTO(out_trans, rc);

	rc = dt_trans_start_local(env, dt, th);
	if (rc < 0)
		GOTO(out_trans, rc);

	rc = lop->lop_destroy(env, handle, th);

out_trans:
	dt_trans_stop(env, dt, th);

	RETURN(rc);
}

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;
	int			 rc = 0;

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

	read_lock(&obj->oo_attr_lock);
	*attr = obj->oo_attr;
	if (obj->oo_lma_flags & LUSTRE_ORPHAN_FL)
		attr->la_flags |= LUSTRE_ORPHAN_FL;
	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;

out:
	up_read(&obj->oo_guard);
	return rc;
}