/*

 * 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 (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.

 * Copyright (c) 2013 by Delphix. All rights reserved.

 */

/* Copyright (c) 2013 by Saso Kiselkov. All rights reserved. */

#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>

#include <sys/zio_compress.h>

#include <sys/sa.h>

#ifdef _KERNEL

#include <sys/vmsystm.h>

#include <sys/zfs_znode.h>

#endif

/*

 * Enable/disable nopwrite feature.

 */

int zfs_nopwrite_enabled = 1;

const dmu_object_type_info_t dmu_ot[DMU_OT_NUMTYPES] = {

	{	DMU_BSWAP_UINT8,	TRUE,	"unallocated"		},

	{	DMU_BSWAP_ZAP,		TRUE,	"object directory"	},

	{	DMU_BSWAP_UINT64,	TRUE,	"object array"		},

	{	DMU_BSWAP_UINT8,	TRUE,	"packed nvlist"		},

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

	{	DMU_BSWAP_UINT64,	TRUE,	"bpobj"			},

	{	DMU_BSWAP_UINT64,	TRUE,	"bpobj header"		},

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

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

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

	{	DMU_BSWAP_DNODE,	TRUE,	"DMU dnode"		},

	{	DMU_BSWAP_OBJSET,	TRUE,	"DMU objset"		},

	{	DMU_BSWAP_UINT64,	TRUE,	"DSL directory"		},

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

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

	{	DMU_BSWAP_ZAP,		TRUE,	"DSL props"		},

	{	DMU_BSWAP_UINT64,	TRUE,	"DSL dataset"		},

	{	DMU_BSWAP_ZNODE,	TRUE,	"ZFS znode"		},

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

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

	{	DMU_BSWAP_ZAP,		TRUE,	"ZFS directory"		},

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

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

	{	DMU_BSWAP_UINT8,	FALSE,	"zvol object"		},

	{	DMU_BSWAP_ZAP,		TRUE,	"zvol prop"		},

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

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

	{	DMU_BSWAP_ZAP,		TRUE,	"other ZAP"		},

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

	{	DMU_BSWAP_UINT8,	TRUE,	"SPA history"		},

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

	{	DMU_BSWAP_ZAP,		TRUE,	"Pool properties"	},

	{	DMU_BSWAP_ZAP,		TRUE,	"DSL permissions"	},

	{	DMU_BSWAP_ACL,		TRUE,	"ZFS ACL"		},

	{	DMU_BSWAP_UINT8,	TRUE,	"ZFS SYSACL"		},

	{	DMU_BSWAP_UINT8,	TRUE,	"FUID table"		},

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

	{	DMU_BSWAP_ZAP,		TRUE,	"DSL dataset next clones"},

	{	DMU_BSWAP_ZAP,		TRUE,	"scan work queue"	},

	{	DMU_BSWAP_ZAP,		TRUE,	"ZFS user/group used"	},

	{	DMU_BSWAP_ZAP,		TRUE,	"ZFS user/group quota"	},

	{	DMU_BSWAP_ZAP,		TRUE,	"snapshot refcount tags"},

	{	DMU_BSWAP_ZAP,		TRUE,	"DDT ZAP algorithm"	},

	{	DMU_BSWAP_ZAP,		TRUE,	"DDT statistics"	},

	{	DMU_BSWAP_UINT8,	TRUE,	"System attributes"	},

	{	DMU_BSWAP_ZAP,		TRUE,	"SA master node"	},

	{	DMU_BSWAP_ZAP,		TRUE,	"SA attr registration"	},

	{	DMU_BSWAP_ZAP,		TRUE,	"SA attr layouts"	},

	{	DMU_BSWAP_ZAP,		TRUE,	"scan translations"	},

	{	DMU_BSWAP_UINT8,	FALSE,	"deduplicated block"	},

	{	DMU_BSWAP_ZAP,		TRUE,	"DSL deadlist map"	},

	{	DMU_BSWAP_UINT64,	TRUE,	"DSL deadlist map hdr"	},

	{	DMU_BSWAP_ZAP,		TRUE,	"DSL dir clones"	},

	{	DMU_BSWAP_UINT64,	TRUE,	"bpobj subobj"		}

};

const dmu_object_byteswap_info_t dmu_ot_byteswap[DMU_BSWAP_NUMFUNCS] = {

	{	byteswap_uint8_array,	"uint8"		},

	{	byteswap_uint16_array,	"uint16"	},

	{	byteswap_uint32_array,	"uint32"	},

	{	byteswap_uint64_array,	"uint64"	},

	{	zap_byteswap,		"zap"		},

	{	dnode_buf_byteswap,	"dnode"		},

	{	dmu_objset_byteswap,	"objset"	},

	{	zfs_znode_byteswap,	"znode"		},

	{	zfs_oldacl_byteswap,	"oldacl"	},

	{	zfs_acl_byteswap,	"acl"		}

};

int

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

    void *tag, dmu_buf_t **dbp, int flags)

{

	dnode_t *dn;

	uint64_t blkid;

	dmu_buf_impl_t *db;

	int err;

	int db_flags = DB_RF_CANFAIL;

	if (flags & DMU_READ_NO_PREFETCH)

		db_flags |= DB_RF_NOPREFETCH;

	err = dnode_hold(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 = SET_ERROR(EIO);

	} else {

		err = dbuf_read(db, NULL, db_flags);

		if (err) {

			dbuf_rele(db, tag);

			db = NULL;

		}

	}

	dnode_rele(dn, FTAG);

	*dbp = &db->db; /* NULL db plus first field offset is NULL */

	return (err);

}

int

dmu_bonus_max(void)

{

	return (DN_MAX_BONUSLEN);

}

int

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

{

	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;

	dnode_t *dn;

	int error;

	DB_DNODE_ENTER(db);

	dn = DB_DNODE(db);

	if (dn->dn_bonus != db) {

		error = SET_ERROR(EINVAL);

	} else if (newsize < 0 || newsize > db_fake->db_size) {

		error = SET_ERROR(EINVAL);

	} else {

		dnode_setbonuslen(dn, newsize, tx);

		error = 0;

	}

	DB_DNODE_EXIT(db);

	return (error);

}

int

dmu_set_bonustype(dmu_buf_t *db_fake, dmu_object_type_t type, dmu_tx_t *tx)

{

	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;

	dnode_t *dn;

	int error;

	DB_DNODE_ENTER(db);

	dn = DB_DNODE(db);

	if (!DMU_OT_IS_VALID(type)) {

		error = SET_ERROR(EINVAL);

	} else if (dn->dn_bonus != db) {

		error = SET_ERROR(EINVAL);

	} else {

		dnode_setbonus_type(dn, type, tx);

		error = 0;

	}

	DB_DNODE_EXIT(db);

	return (error);

}

dmu_object_type_t

dmu_get_bonustype(dmu_buf_t *db_fake)

{

	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;

	dnode_t *dn;

	dmu_object_type_t type;

	DB_DNODE_ENTER(db);

	dn = DB_DNODE(db);

	type = dn->dn_bonustype;

	DB_DNODE_EXIT(db);

	return (type);

}

int

dmu_rm_spill(objset_t *os, uint64_t object, dmu_tx_t *tx)

{

	dnode_t *dn;

	int error;

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

	dbuf_rm_spill(dn, tx);

	rw_enter(&dn->dn_struct_rwlock, RW_WRITER);

	dnode_rm_spill(dn, tx);

	rw_exit(&dn->dn_struct_rwlock);

	dnode_rele(dn, FTAG);

	return (error);

}

/*

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

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

		(void) atomic_inc_32_nv(&dn->dn_dbufs_count);

	}

	/*

	 * Wait to drop dn_struct_rwlock until after adding the bonus dbuf's

	 * hold and incrementing the dbuf count to ensure that dnode_move() sees

	 * a dnode hold for every dbuf.

	 */

	rw_exit(&dn->dn_struct_rwlock);

	dnode_rele(dn, FTAG);

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

	*dbp = &db->db;

	return (0);

}

/*

 * returns ENOENT, EIO, or 0.

 *

 * This interface will allocate a blank spill dbuf when a spill blk

 * doesn't already exist on the dnode.

 *

 * if you only want to find an already existing spill db, then

 * dmu_spill_hold_existing() should be used.

 */

int

dmu_spill_hold_by_dnode(dnode_t *dn, uint32_t flags, void *tag, dmu_buf_t **dbp)

{

	dmu_buf_impl_t *db = NULL;

	int err;

	if ((flags & DB_RF_HAVESTRUCT) == 0)

		rw_enter(&dn->dn_struct_rwlock, RW_READER);

	db = dbuf_hold(dn, DMU_SPILL_BLKID, tag);

	if ((flags & DB_RF_HAVESTRUCT) == 0)

		rw_exit(&dn->dn_struct_rwlock);

	ASSERT(db != NULL);

	err = dbuf_read(db, NULL, flags);

	if (err == 0)

		*dbp = &db->db;

	else

		dbuf_rele(db, tag);

	return (err);

}

int

dmu_spill_hold_existing(dmu_buf_t *bonus, void *tag, dmu_buf_t **dbp)

{

	dmu_buf_impl_t *db = (dmu_buf_impl_t *)bonus;

	dnode_t *dn;

	int err;

	DB_DNODE_ENTER(db);

	dn = DB_DNODE(db);

	if (spa_version(dn->dn_objset->os_spa) < SPA_VERSION_SA) {

		err = SET_ERROR(EINVAL);

	} else {

		rw_enter(&dn->dn_struct_rwlock, RW_READER);

		if (!dn->dn_have_spill) {

			err = SET_ERROR(ENOENT);

		} else {

			err = dmu_spill_hold_by_dnode(dn,

			    DB_RF_HAVESTRUCT | DB_RF_CANFAIL, tag, dbp);

		}

		rw_exit(&dn->dn_struct_rwlock);

	}

	DB_DNODE_EXIT(db);

	return (err);

}

int

dmu_spill_hold_by_bonus(dmu_buf_t *bonus, void *tag, dmu_buf_t **dbp)

{

	dmu_buf_impl_t *db = (dmu_buf_impl_t *)bonus;

	dnode_t *dn;

	int err;

	DB_DNODE_ENTER(db);

	dn = DB_DNODE(db);

	err = dmu_spill_hold_by_dnode(dn, DB_RF_CANFAIL, tag, dbp);

	DB_DNODE_EXIT(db);

	return (err);

}

/*

 * 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, uint32_t flags)

{

	dsl_pool_t *dp = NULL;

	dmu_buf_t **dbp;

	uint64_t blkid, nblks, i;

	uint32_t dbuf_flags;

	int err;

	zio_t *zio;

	hrtime_t start;

	ASSERT(length <= DMU_MAX_ACCESS);

	dbuf_flags = DB_RF_CANFAIL | DB_RF_NEVERWAIT | DB_RF_HAVESTRUCT;

	if (flags & DMU_READ_NO_PREFETCH || length > zfetch_array_rd_sz)

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

			rw_exit(&dn->dn_struct_rwlock);

			return (SET_ERROR(EIO));

		}

		nblks = 1;

	}

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

	if (dn->dn_objset->os_dsl_dataset)

		dp = dn->dn_objset->os_dsl_dataset->ds_dir->dd_pool;

	start = gethrtime();

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

	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 (SET_ERROR(EIO));

		}

		/* initiate async i/o */

		if (read) {

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

		}

		dbp[i] = &db->db;

	}

	rw_exit(&dn->dn_struct_rwlock);

	/* wait for async i/o */

	err = zio_wait(zio);

	/* track read overhead when we are in sync context */

	if (dp && dsl_pool_sync_context(dp))

		dp->dp_read_overhead += gethrtime() - start;

	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 = SET_ERROR(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, object, FTAG, &dn);

	if (err)

		return (err);

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

	    numbufsp, dbpp, DMU_READ_PREFETCH);

	dnode_rele(dn, FTAG);

	return (err);

}

int

dmu_buf_hold_array_by_bonus(dmu_buf_t *db_fake, uint64_t offset,

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

{

	dmu_buf_impl_t *db = (dmu_buf_impl_t *)db_fake;

	dnode_t *dn;

	int err;

	DB_DNODE_ENTER(db);

	dn = DB_DNODE(db);