/*

 * CDDL HEADER START

 *

 * The contents of this file are subject to the terms of the

 * Common Development and Distribution License (the "License").

 * You may not use this file except in compliance with the License.

 *

 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE

 * or http://www.opensolaris.org/os/licensing.

 * See the License for the specific language governing permissions

 * and limitations under the License.

 *

 * When distributing Covered Code, include this CDDL HEADER in each

 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.

 * If applicable, add the following below this CDDL HEADER, with the

 * fields enclosed by brackets "[]" replaced with your own identifying

 * information: Portions Copyright [yyyy] [name of copyright owner]

 *

 * CDDL HEADER END

 */

/*

 * Copyright 2007 Sun Microsystems, Inc.  All rights reserved.

 * Use is subject to license terms.

 */

#pragma ident	"%Z%%M%	%I%	%E% SMI"

#include <sys/dmu.h>

#include <sys/dmu_impl.h>

#include <sys/dmu_tx.h>

#include <sys/dbuf.h>

#include <sys/dnode.h>

#include <sys/zfs_context.h>

#include <sys/dmu_objset.h>

#include <sys/dmu_traverse.h>

#include <sys/dsl_dataset.h>

#include <sys/dsl_dir.h>

#include <sys/dsl_pool.h>

#include <sys/dsl_synctask.h>

#include <sys/dsl_prop.h>

#include <sys/dmu_zfetch.h>

#include <sys/zfs_ioctl.h>

#include <sys/zap.h>

#include <sys/zio_checksum.h>

#ifdef _KERNEL

#include <sys/vmsystm.h>

#endif

const dmu_object_type_info_t dmu_ot[DMU_OT_NUMTYPES] = {

	{	byteswap_uint8_array,	TRUE,	"unallocated"		},

	{	zap_byteswap,		TRUE,	"object directory"	},

	{	byteswap_uint64_array,	TRUE,	"object array"		},

	{	byteswap_uint8_array,	TRUE,	"packed nvlist"		},

	{	byteswap_uint64_array,	TRUE,	"packed nvlist size"	},

	{	byteswap_uint64_array,	TRUE,	"bplist"		},

	{	byteswap_uint64_array,	TRUE,	"bplist header"		},

	{	byteswap_uint64_array,	TRUE,	"SPA space map header"	},

	{	byteswap_uint64_array,	TRUE,	"SPA space map"		},

	{	byteswap_uint64_array,	TRUE,	"ZIL intent log"	},

	{	dnode_buf_byteswap,	TRUE,	"DMU dnode"		},

	{	dmu_objset_byteswap,	TRUE,	"DMU objset"		},

	{	byteswap_uint64_array,	TRUE,	"DSL directory"		},

	{	zap_byteswap,		TRUE,	"DSL directory child map"},

	{	zap_byteswap,		TRUE,	"DSL dataset snap map"	},

	{	zap_byteswap,		TRUE,	"DSL props"		},

	{	byteswap_uint64_array,	TRUE,	"DSL dataset"		},

	{	zfs_znode_byteswap,	TRUE,	"ZFS znode"		},

	{	zfs_oldacl_byteswap,	TRUE,	"ZFS V0 ACL"		},

	{	byteswap_uint8_array,	FALSE,	"ZFS plain file"	},

	{	zap_byteswap,		TRUE,	"ZFS directory"		},

	{	zap_byteswap,		TRUE,	"ZFS master node"	},

	{	zap_byteswap,		TRUE,	"ZFS delete queue"	},

	{	byteswap_uint8_array,	FALSE,	"zvol object"		},

	{	zap_byteswap,		TRUE,	"zvol prop"		},

	{	byteswap_uint8_array,	FALSE,	"other uint8[]"		},

	{	byteswap_uint64_array,	FALSE,	"other uint64[]"	},

	{	zap_byteswap,		TRUE,	"other ZAP"		},

	{	zap_byteswap,		TRUE,	"persistent error log"	},

	{	byteswap_uint8_array,	TRUE,	"SPA history"		},

	{	byteswap_uint64_array,	TRUE,	"SPA history offsets"	},

	{	zap_byteswap,		TRUE,	"Pool properties"	},

	{	zap_byteswap,		TRUE,	"DSL permissions"	},

	{	zfs_acl_byteswap,	TRUE,	"ZFS ACL"		},

	{	byteswap_uint8_array,	TRUE,	"ZFS SYSACL"		},

	{	byteswap_uint8_array,	TRUE,	"FUID table"		},

	{	byteswap_uint64_array,	TRUE,	"FUID table size"	},

};

int

dmu_buf_hold(objset_t *os, uint64_t object, uint64_t offset,

    void *tag, dmu_buf_t **dbp)

{

	dnode_t *dn;

	uint64_t blkid;

	dmu_buf_impl_t *db;

	int err;

	err = dnode_hold(os->os, object, FTAG, &dn);

	if (err)

		return (err);

	blkid = dbuf_whichblock(dn, offset);

	rw_enter(&dn->dn_struct_rwlock, RW_READER);

	db = dbuf_hold(dn, blkid, tag);

	rw_exit(&dn->dn_struct_rwlock);

	if (db == NULL) {

		err = EIO;

	} else {

		err = dbuf_read(db, NULL, DB_RF_CANFAIL);

		if (err) {

			dbuf_rele(db, tag);

			db = NULL;

		}

	}

	dnode_rele(dn, FTAG);

	*dbp = &db->db;

	return (err);

}

int

dmu_bonus_max(void)

{

	return (DN_MAX_BONUSLEN);

}

int

dmu_set_bonus(dmu_buf_t *db, int newsize, dmu_tx_t *tx)

{

	dnode_t *dn = ((dmu_buf_impl_t *)db)->db_dnode;

	if (dn->dn_bonus != (dmu_buf_impl_t *)db)

		return (EINVAL);

	if (newsize < 0 || newsize > db->db_size)

		return (EINVAL);

	dnode_setbonuslen(dn, newsize, tx);

	return (0);

}

/*

 * returns ENOENT, EIO, or 0.

 */

int

dmu_bonus_hold(objset_t *os, uint64_t object, void *tag, dmu_buf_t **dbp)

{

	dnode_t *dn;

	dmu_buf_impl_t *db;

	int error;

	error = dnode_hold(os->os, object, FTAG, &dn);

	if (error)

		return (error);

	rw_enter(&dn->dn_struct_rwlock, RW_READER);

	if (dn->dn_bonus == NULL) {

		rw_exit(&dn->dn_struct_rwlock);

		rw_enter(&dn->dn_struct_rwlock, RW_WRITER);

		if (dn->dn_bonus == NULL)

			dbuf_create_bonus(dn);

	}

	db = dn->dn_bonus;

	rw_exit(&dn->dn_struct_rwlock);

	/* as long as the bonus buf is held, the dnode will be held */

	if (refcount_add(&db->db_holds, tag) == 1)

		VERIFY(dnode_add_ref(dn, db));

	dnode_rele(dn, FTAG);

	VERIFY(0 == dbuf_read(db, NULL, DB_RF_MUST_SUCCEED));

	*dbp = &db->db;

	return (0);

}

/*

 * Note: longer-term, we should modify all of the dmu_buf_*() interfaces

 * to take a held dnode rather than <os, object> -- the lookup is wasteful,

 * and can induce severe lock contention when writing to several files

 * whose dnodes are in the same block.

 */

static int

dmu_buf_hold_array_by_dnode(dnode_t *dn, uint64_t offset,

    uint64_t length, int read, void *tag, int *numbufsp, dmu_buf_t ***dbpp)

{

	dmu_buf_t **dbp;

	uint64_t blkid, nblks, i;

	uint32_t flags;

	int err;

	zio_t *zio;

	ASSERT(length <= DMU_MAX_ACCESS);

	flags = DB_RF_CANFAIL | DB_RF_NEVERWAIT;

	if (length > zfetch_array_rd_sz)

		flags |= DB_RF_NOPREFETCH;

	rw_enter(&dn->dn_struct_rwlock, RW_READER);

	if (dn->dn_datablkshift) {

		int blkshift = dn->dn_datablkshift;

		nblks = (P2ROUNDUP(offset+length, 1ULL<<blkshift) -

		    P2ALIGN(offset, 1ULL<<blkshift)) >> blkshift;

	} else {

		if (offset + length > dn->dn_datablksz) {

			zfs_panic_recover("zfs: accessing past end of object "

			    "%llx/%llx (size=%u access=%llu+%llu)",

			    (longlong_t)dn->dn_objset->

			    os_dsl_dataset->ds_object,

			    (longlong_t)dn->dn_object, dn->dn_datablksz,

			    (longlong_t)offset, (longlong_t)length);

			return (EIO);

		}

		nblks = 1;

	}

	dbp = kmem_zalloc(sizeof (dmu_buf_t *) * nblks, KM_SLEEP);

	zio = zio_root(dn->dn_objset->os_spa, NULL, NULL, TRUE);

	blkid = dbuf_whichblock(dn, offset);

	for (i = 0; i < nblks; i++) {

		dmu_buf_impl_t *db = dbuf_hold(dn, blkid+i, tag);

		if (db == NULL) {

			rw_exit(&dn->dn_struct_rwlock);

			dmu_buf_rele_array(dbp, nblks, tag);

			zio_nowait(zio);

			return (EIO);

		}

		/* initiate async i/o */

		if (read) {

			rw_exit(&dn->dn_struct_rwlock);

			(void) dbuf_read(db, zio, flags);

			rw_enter(&dn->dn_struct_rwlock, RW_READER);

		}

		dbp[i] = &db->db;

	}

	rw_exit(&dn->dn_struct_rwlock);

	/* wait for async i/o */

	err = zio_wait(zio);

	if (err) {

		dmu_buf_rele_array(dbp, nblks, tag);

		return (err);

	}

	/* wait for other io to complete */

	if (read) {

		for (i = 0; i < nblks; i++) {

			dmu_buf_impl_t *db = (dmu_buf_impl_t *)dbp[i];

			mutex_enter(&db->db_mtx);

			while (db->db_state == DB_READ ||

			    db->db_state == DB_FILL)

				cv_wait(&db->db_changed, &db->db_mtx);

			if (db->db_state == DB_UNCACHED)

				err = EIO;

			mutex_exit(&db->db_mtx);

			if (err) {

				dmu_buf_rele_array(dbp, nblks, tag);

				return (err);

			}

		}

	}

	*numbufsp = nblks;

	*dbpp = dbp;

	return (0);

}

static int

dmu_buf_hold_array(objset_t *os, uint64_t object, uint64_t offset,

    uint64_t length, int read, void *tag, int *numbufsp, dmu_buf_t ***dbpp)

{

	dnode_t *dn;

	int err;

	err = dnode_hold(os->os, object, FTAG, &dn);

	if (err)

		return (err);

	err = dmu_buf_hold_array_by_dnode(dn, offset, length, read, tag,

	    numbufsp, dbpp);

	dnode_rele(dn, FTAG);

	return (err);

}

int

dmu_buf_hold_array_by_bonus(dmu_buf_t *db, uint64_t offset,

    uint64_t length, int read, void *tag, int *numbufsp, dmu_buf_t ***dbpp)

{

	dnode_t *dn = ((dmu_buf_impl_t *)db)->db_dnode;

	int err;

	err = dmu_buf_hold_array_by_dnode(dn, offset, length, read, tag,

	    numbufsp, dbpp);

	return (err);

}

void

dmu_buf_rele_array(dmu_buf_t **dbp_fake, int numbufs, void *tag)

{

	int i;

	dmu_buf_impl_t **dbp = (dmu_buf_impl_t **)dbp_fake;

	if (numbufs == 0)

		return;

	for (i = 0; i < numbufs; i++) {

		if (dbp[i])

			dbuf_rele(dbp[i], tag);

	}

	kmem_free(dbp, sizeof (dmu_buf_t *) * numbufs);

}

void

dmu_prefetch(objset_t *os, uint64_t object, uint64_t offset, uint64_t len)

{

	dnode_t *dn;

	uint64_t blkid;

	int nblks, i, err;

	if (zfs_prefetch_disable)

		return;

	if (len == 0) {  /* they're interested in the bonus buffer */

		dn = os->os->os_meta_dnode;

		if (object == 0 || object >= DN_MAX_OBJECT)

			return;

		rw_enter(&dn->dn_struct_rwlock, RW_READER);

		blkid = dbuf_whichblock(dn, object * sizeof (dnode_phys_t));

		dbuf_prefetch(dn, blkid);

		rw_exit(&dn->dn_struct_rwlock);

		return;

	}

	/*

	 * XXX - Note, if the dnode for the requested object is not

	 * already cached, we will do a *synchronous* read in the

	 * dnode_hold() call.  The same is true for any indirects.

	 */

	err = dnode_hold(os->os, object, FTAG, &dn);

	if (err != 0)

		return;

	rw_enter(&dn->dn_struct_rwlock, RW_READER);

	if (dn->dn_datablkshift) {

		int blkshift = dn->dn_datablkshift;

		nblks = (P2ROUNDUP(offset+len, 1<<blkshift) -

		    P2ALIGN(offset, 1<<blkshift)) >> blkshift;

	} else {

		nblks = (offset < dn->dn_datablksz);

	}

	if (nblks != 0) {

		blkid = dbuf_whichblock(dn, offset);

		for (i = 0; i < nblks; i++)

			dbuf_prefetch(dn, blkid+i);

	}

	rw_exit(&dn->dn_struct_rwlock);

	dnode_rele(dn, FTAG);

}

int

dmu_free_range(objset_t *os, uint64_t object, uint64_t offset,

    uint64_t size, dmu_tx_t *tx)

{

	dnode_t *dn;

	int err = dnode_hold(os->os, object, FTAG, &dn);

	if (err)

		return (err);

	ASSERT(offset < UINT64_MAX);

	ASSERT(size == -1ULL || size <= UINT64_MAX - offset);

	dnode_free_range(dn, offset, size, tx);

	dnode_rele(dn, FTAG);

	return (0);

}

int

dmu_read(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,

    void *buf)

{

	dnode_t *dn;

	dmu_buf_t **dbp;

	int numbufs, i, err;

	err = dnode_hold(os->os, object, FTAG, &dn);

	if (err)

		return (err);

	/*

	 * Deal with odd block sizes, where there can't be data past the first

	 * block.  If we ever do the tail block optimization, we will need to

	 * handle that here as well.

	 */

	if (dn->dn_datablkshift == 0) {

		int newsz = offset > dn->dn_datablksz ? 0 :

		    MIN(size, dn->dn_datablksz - offset);

		bzero((char *)buf + newsz, size - newsz);

		size = newsz;

	}

	while (size > 0) {

		uint64_t mylen = MIN(size, DMU_MAX_ACCESS / 2);

		/*

		 * NB: we could do this block-at-a-time, but it's nice

		 * to be reading in parallel.

		 */

		err = dmu_buf_hold_array_by_dnode(dn, offset, mylen,

		    TRUE, FTAG, &numbufs, &dbp);

		if (err)

			break;

		for (i = 0; i < numbufs; i++) {

			int tocpy;

			int bufoff;

			dmu_buf_t *db = dbp[i];

			ASSERT(size > 0);

			bufoff = offset - db->db_offset;

			tocpy = (int)MIN(db->db_size - bufoff, size);

			bcopy((char *)db->db_data + bufoff, buf, tocpy);

			offset += tocpy;

			size -= tocpy;

			buf = (char *)buf + tocpy;

		}

		dmu_buf_rele_array(dbp, numbufs, FTAG);

	}

	dnode_rele(dn, FTAG);

	return (err);

}

void

dmu_write(objset_t *os, uint64_t object, uint64_t offset, uint64_t size,

    const void *buf, dmu_tx_t *tx)

{

	dmu_buf_t **dbp;

	int numbufs, i;

	if (size == 0)

		return;

	VERIFY(0 == dmu_buf_hold_array(os, object, offset, size,

	    FALSE, FTAG, &numbufs, &dbp));

	for (i = 0; i < numbufs; i++) {

		int tocpy;

		int bufoff;

		dmu_buf_t *db = dbp[i];

		ASSERT(size > 0);

		bufoff = offset - db->db_offset;

		tocpy = (int)MIN(db->db_size - bufoff, size);

		ASSERT(i == 0 || i == numbufs-1 || tocpy == db->db_size);

		if (tocpy == db->db_size)

			dmu_buf_will_fill(db, tx);

		else

			dmu_buf_will_dirty(db, tx);

		bcopy(buf, (char *)db->db_data + bufoff, tocpy);

		if (tocpy == db->db_size)

			dmu_buf_fill_done(db, tx);

		offset += tocpy;

		size -= tocpy;

		buf = (char *)buf + tocpy;

	}

	dmu_buf_rele_array(dbp, numbufs, FTAG);

}

#ifdef _KERNEL

int

dmu_read_uio(objset_t *os, uint64_t object, uio_t *uio, uint64_t size)

{

	dmu_buf_t **dbp;

	int numbufs, i, err;

	/*

	 * NB: we could do this block-at-a-time, but it's nice

	 * to be reading in parallel.

	 */

	err = dmu_buf_hold_array(os, object, uio->uio_loffset, size, TRUE, FTAG,

	    &numbufs, &dbp);

	if (err)

		return (err);

	for (i = 0; i < numbufs; i++) {

		int tocpy;

		int bufoff;

		dmu_buf_t *db = dbp[i];

		ASSERT(size > 0);