/*

 * 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) 2012, Joyent, Inc. All rights reserved.

 */

#include <sys/dmu_objset.h>

#include <sys/dsl_dataset.h>

#include <sys/dsl_dir.h>

#include <sys/dsl_prop.h>

#include <sys/dsl_synctask.h>

#include <sys/dmu_traverse.h>

#include <sys/dmu_impl.h>

#include <sys/dmu_tx.h>

#include <sys/arc.h>

#include <sys/zio.h>

#include <sys/zap.h>

#include <sys/zfeature.h>

#include <sys/unique.h>

#include <sys/zfs_context.h>

#include <sys/zfs_ioctl.h>

#include <sys/spa.h>

#include <sys/zfs_znode.h>

#include <sys/zfs_onexit.h>

#include <sys/zvol.h>

#include <sys/dsl_scan.h>

#include <sys/dsl_deadlist.h>

#include <sys/dsl_destroy.h>

#include <sys/dsl_userhold.h>

#define	SWITCH64(x, y) \

	{ \

		uint64_t __tmp = (x); \

		(x) = (y); \

		(y) = __tmp; \

	}

#define	DS_REF_MAX	(1ULL << 62)

#define	DSL_DEADLIST_BLOCKSIZE	SPA_MAXBLOCKSIZE

/*

 * Figure out how much of this delta should be propogated to the dsl_dir

 * layer.  If there's a refreservation, that space has already been

 * partially accounted for in our ancestors.

 */

static int64_t

parent_delta(dsl_dataset_t *ds, int64_t delta)

{

	uint64_t old_bytes, new_bytes;

	if (ds->ds_reserved == 0)

		return (delta);

	old_bytes = MAX(ds->ds_phys->ds_unique_bytes, ds->ds_reserved);

	new_bytes = MAX(ds->ds_phys->ds_unique_bytes + delta, ds->ds_reserved);

	ASSERT3U(ABS((int64_t)(new_bytes - old_bytes)), <=, ABS(delta));

	return (new_bytes - old_bytes);

}

void

dsl_dataset_block_born(dsl_dataset_t *ds, const blkptr_t *bp, dmu_tx_t *tx)

{

	int used = bp_get_dsize_sync(tx->tx_pool->dp_spa, bp);

	int compressed = BP_GET_PSIZE(bp);

	int uncompressed = BP_GET_UCSIZE(bp);

	int64_t delta;

	dprintf_bp(bp, "ds=%p", ds);

	ASSERT(dmu_tx_is_syncing(tx));

	/* It could have been compressed away to nothing */

	if (BP_IS_HOLE(bp))

		return;

	ASSERT(BP_GET_TYPE(bp) != DMU_OT_NONE);

	ASSERT(DMU_OT_IS_VALID(BP_GET_TYPE(bp)));

	if (ds == NULL) {

		dsl_pool_mos_diduse_space(tx->tx_pool,

		    used, compressed, uncompressed);

		return;

	}

	dmu_buf_will_dirty(ds->ds_dbuf, tx);

	mutex_enter(&ds->ds_dir->dd_lock);

	mutex_enter(&ds->ds_lock);

	delta = parent_delta(ds, used);

	ds->ds_phys->ds_referenced_bytes += used;

	ds->ds_phys->ds_compressed_bytes += compressed;

	ds->ds_phys->ds_uncompressed_bytes += uncompressed;

	ds->ds_phys->ds_unique_bytes += used;

	mutex_exit(&ds->ds_lock);

	dsl_dir_diduse_space(ds->ds_dir, DD_USED_HEAD, delta,

	    compressed, uncompressed, tx);

	dsl_dir_transfer_space(ds->ds_dir, used - delta,

	    DD_USED_REFRSRV, DD_USED_HEAD, tx);

	mutex_exit(&ds->ds_dir->dd_lock);

}

int

dsl_dataset_block_kill(dsl_dataset_t *ds, const blkptr_t *bp, dmu_tx_t *tx,

    boolean_t async)

{

	if (BP_IS_HOLE(bp))

		return (0);

	ASSERT(dmu_tx_is_syncing(tx));

	ASSERT(bp->blk_birth <= tx->tx_txg);

	int used = bp_get_dsize_sync(tx->tx_pool->dp_spa, bp);

	int compressed = BP_GET_PSIZE(bp);

	int uncompressed = BP_GET_UCSIZE(bp);

	ASSERT(used > 0);

	if (ds == NULL) {

		dsl_free(tx->tx_pool, tx->tx_txg, bp);

		dsl_pool_mos_diduse_space(tx->tx_pool,

		    -used, -compressed, -uncompressed);

		return (used);

	}

	ASSERT3P(tx->tx_pool, ==, ds->ds_dir->dd_pool);

	ASSERT(!dsl_dataset_is_snapshot(ds));

	dmu_buf_will_dirty(ds->ds_dbuf, tx);

	if (bp->blk_birth > ds->ds_phys->ds_prev_snap_txg) {

		int64_t delta;

		dprintf_bp(bp, "freeing ds=%llu", ds->ds_object);

		dsl_free(tx->tx_pool, tx->tx_txg, bp);

		mutex_enter(&ds->ds_dir->dd_lock);

		mutex_enter(&ds->ds_lock);

		ASSERT(ds->ds_phys->ds_unique_bytes >= used ||

		    !DS_UNIQUE_IS_ACCURATE(ds));

		delta = parent_delta(ds, -used);

		ds->ds_phys->ds_unique_bytes -= used;

		mutex_exit(&ds->ds_lock);

		dsl_dir_diduse_space(ds->ds_dir, DD_USED_HEAD,

		    delta, -compressed, -uncompressed, tx);

		dsl_dir_transfer_space(ds->ds_dir, -used - delta,

		    DD_USED_REFRSRV, DD_USED_HEAD, tx);

		mutex_exit(&ds->ds_dir->dd_lock);

	} else {

		dprintf_bp(bp, "putting on dead list: %s", "");

		if (async) {

			/*

			 * We are here as part of zio's write done callback,

			 * which means we're a zio interrupt thread.  We can't

			 * call dsl_deadlist_insert() now because it may block

			 * waiting for I/O.  Instead, put bp on the deferred

			 * queue and let dsl_pool_sync() finish the job.

			 */

			bplist_append(&ds->ds_pending_deadlist, bp);

		} else {

			dsl_deadlist_insert(&ds->ds_deadlist, bp, tx);

		}

		ASSERT3U(ds->ds_prev->ds_object, ==,

		    ds->ds_phys->ds_prev_snap_obj);

		ASSERT(ds->ds_prev->ds_phys->ds_num_children > 0);

		/* if (bp->blk_birth > prev prev snap txg) prev unique += bs */

		if (ds->ds_prev->ds_phys->ds_next_snap_obj ==

		    ds->ds_object && bp->blk_birth >

		    ds->ds_prev->ds_phys->ds_prev_snap_txg) {

			dmu_buf_will_dirty(ds->ds_prev->ds_dbuf, tx);

			mutex_enter(&ds->ds_prev->ds_lock);

			ds->ds_prev->ds_phys->ds_unique_bytes += used;

			mutex_exit(&ds->ds_prev->ds_lock);

		}

		if (bp->blk_birth > ds->ds_dir->dd_origin_txg) {

			dsl_dir_transfer_space(ds->ds_dir, used,

			    DD_USED_HEAD, DD_USED_SNAP, tx);

		}

	}

	mutex_enter(&ds->ds_lock);

	ASSERT3U(ds->ds_phys->ds_referenced_bytes, >=, used);

	ds->ds_phys->ds_referenced_bytes -= used;

	ASSERT3U(ds->ds_phys->ds_compressed_bytes, >=, compressed);

	ds->ds_phys->ds_compressed_bytes -= compressed;

	ASSERT3U(ds->ds_phys->ds_uncompressed_bytes, >=, uncompressed);

	ds->ds_phys->ds_uncompressed_bytes -= uncompressed;

	mutex_exit(&ds->ds_lock);

	return (used);

}

uint64_t

dsl_dataset_prev_snap_txg(dsl_dataset_t *ds)

{

	uint64_t trysnap = 0;

	if (ds == NULL)

		return (0);

	/*

	 * The snapshot creation could fail, but that would cause an

	 * incorrect FALSE return, which would only result in an

	 * overestimation of the amount of space that an operation would

	 * consume, which is OK.

	 *

	 * There's also a small window where we could miss a pending

	 * snapshot, because we could set the sync task in the quiescing

	 * phase.  So this should only be used as a guess.

	 */

	if (ds->ds_trysnap_txg >

	    spa_last_synced_txg(ds->ds_dir->dd_pool->dp_spa))

		trysnap = ds->ds_trysnap_txg;

	return (MAX(ds->ds_phys->ds_prev_snap_txg, trysnap));

}

boolean_t

dsl_dataset_block_freeable(dsl_dataset_t *ds, const blkptr_t *bp,

    uint64_t blk_birth)

{

	if (blk_birth <= dsl_dataset_prev_snap_txg(ds))

		return (B_FALSE);

	ddt_prefetch(dsl_dataset_get_spa(ds), bp);

	return (B_TRUE);

}

/* ARGSUSED */

static void

dsl_dataset_evict(dmu_buf_t *db, void *dsv)

{

	dsl_dataset_t *ds = dsv;

	ASSERT(ds->ds_owner == NULL);

	unique_remove(ds->ds_fsid_guid);

	if (ds->ds_objset != NULL)

		dmu_objset_evict(ds->ds_objset);

	if (ds->ds_prev) {

		dsl_dataset_rele(ds->ds_prev, ds);

		ds->ds_prev = NULL;

	}

	bplist_destroy(&ds->ds_pending_deadlist);

	if (ds->ds_phys->ds_deadlist_obj != 0)

		dsl_deadlist_close(&ds->ds_deadlist);

	if (ds->ds_dir)

		dsl_dir_rele(ds->ds_dir, ds);

	ASSERT(!list_link_active(&ds->ds_synced_link));

	mutex_destroy(&ds->ds_lock);

	mutex_destroy(&ds->ds_opening_lock);

	refcount_destroy(&ds->ds_longholds);

	kmem_free(ds, sizeof (dsl_dataset_t));

}

int

dsl_dataset_get_snapname(dsl_dataset_t *ds)

{

	dsl_dataset_phys_t *headphys;

	int err;

	dmu_buf_t *headdbuf;

	dsl_pool_t *dp = ds->ds_dir->dd_pool;

	objset_t *mos = dp->dp_meta_objset;

	if (ds->ds_snapname[0])

		return (0);

	if (ds->ds_phys->ds_next_snap_obj == 0)

		return (0);

	err = dmu_bonus_hold(mos, ds->ds_dir->dd_phys->dd_head_dataset_obj,

	    FTAG, &headdbuf);

	if (err != 0)

		return (err);

	headphys = headdbuf->db_data;

	err = zap_value_search(dp->dp_meta_objset,

	    headphys->ds_snapnames_zapobj, ds->ds_object, 0, ds->ds_snapname);

	dmu_buf_rele(headdbuf, FTAG);

	return (err);

}

int

dsl_dataset_snap_lookup(dsl_dataset_t *ds, const char *name, uint64_t *value)

{

	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;

	uint64_t snapobj = ds->ds_phys->ds_snapnames_zapobj;

	matchtype_t mt;

	int err;

	if (ds->ds_phys->ds_flags & DS_FLAG_CI_DATASET)

		mt = MT_FIRST;

	else

		mt = MT_EXACT;

	err = zap_lookup_norm(mos, snapobj, name, 8, 1,

	    value, mt, NULL, 0, NULL);

	if (err == ENOTSUP && mt == MT_FIRST)

		err = zap_lookup(mos, snapobj, name, 8, 1, value);

	return (err);

}

int

dsl_dataset_snap_remove(dsl_dataset_t *ds, const char *name, dmu_tx_t *tx)

{

	objset_t *mos = ds->ds_dir->dd_pool->dp_meta_objset;

	uint64_t snapobj = ds->ds_phys->ds_snapnames_zapobj;

	matchtype_t mt;

	int err;

	dsl_dir_snap_cmtime_update(ds->ds_dir);

	if (ds->ds_phys->ds_flags & DS_FLAG_CI_DATASET)

		mt = MT_FIRST;

	else

		mt = MT_EXACT;

	err = zap_remove_norm(mos, snapobj, name, mt, tx);

	if (err == ENOTSUP && mt == MT_FIRST)

		err = zap_remove(mos, snapobj, name, tx);

	return (err);

}

int

dsl_dataset_hold_obj(dsl_pool_t *dp, uint64_t dsobj, void *tag,

    dsl_dataset_t **dsp)

{

	objset_t *mos = dp->dp_meta_objset;

	dmu_buf_t *dbuf;

	dsl_dataset_t *ds;

	int err;

	dmu_object_info_t doi;

	ASSERT(dsl_pool_config_held(dp));

	err = dmu_bonus_hold(mos, dsobj, tag, &dbuf);

	if (err != 0)

		return (err);

	/* Make sure dsobj has the correct object type. */

	dmu_object_info_from_db(dbuf, &doi);

	if (doi.doi_type != DMU_OT_DSL_DATASET) {

		dmu_buf_rele(dbuf, tag);

		return (SET_ERROR(EINVAL));

	}

	ds = dmu_buf_get_user(dbuf);

	if (ds == NULL) {

		dsl_dataset_t *winner = NULL;

		ds = kmem_zalloc(sizeof (dsl_dataset_t), KM_SLEEP);

		ds->ds_dbuf = dbuf;

		ds->ds_object = dsobj;

		ds->ds_phys = dbuf->db_data;

		mutex_init(&ds->ds_lock, NULL, MUTEX_DEFAULT, NULL);

		mutex_init(&ds->ds_opening_lock, NULL, MUTEX_DEFAULT, NULL);

		mutex_init(&ds->ds_sendstream_lock, NULL, MUTEX_DEFAULT, NULL);

		refcount_create(&ds->ds_longholds);

		bplist_create(&ds->ds_pending_deadlist);

		dsl_deadlist_open(&ds->ds_deadlist,

		    mos, ds->ds_phys->ds_deadlist_obj);

		list_create(&ds->ds_sendstreams, sizeof (dmu_sendarg_t),

		    offsetof(dmu_sendarg_t, dsa_link));

		if (err == 0) {

			err = dsl_dir_hold_obj(dp,

			    ds->ds_phys->ds_dir_obj, NULL, ds, &ds->ds_dir);

		}

		if (err != 0) {

			mutex_destroy(&ds->ds_lock);

			mutex_destroy(&ds->ds_opening_lock);

			refcount_destroy(&ds->ds_longholds);

			bplist_destroy(&ds->ds_pending_deadlist);

			dsl_deadlist_close(&ds->ds_deadlist);

			kmem_free(ds, sizeof (dsl_dataset_t));

			dmu_buf_rele(dbuf, tag);

			return (err);

		}

		if (!dsl_dataset_is_snapshot(ds)) {

			ds->ds_snapname[0] = '\0';

			if (ds->ds_phys->ds_prev_snap_obj != 0) {

				err = dsl_dataset_hold_obj(dp,

				    ds->ds_phys->ds_prev_snap_obj,

				    ds, &ds->ds_prev);

			}

		} else {

			if (zfs_flags & ZFS_DEBUG_SNAPNAMES)

				err = dsl_dataset_get_snapname(ds);

			if (err == 0 && ds->ds_phys->ds_userrefs_obj != 0) {

				err = zap_count(

				    ds->ds_dir->dd_pool->dp_meta_objset,

				    ds->ds_phys->ds_userrefs_obj,

				    &ds->ds_userrefs);

			}

		}

		if (err == 0 && !dsl_dataset_is_snapshot(ds)) {

			err = dsl_prop_get_int_ds(ds,

			    zfs_prop_to_name(ZFS_PROP_REFRESERVATION),

			    &ds->ds_reserved);

			if (err == 0) {

				err = dsl_prop_get_int_ds(ds,

				    zfs_prop_to_name(ZFS_PROP_REFQUOTA),

				    &ds->ds_quota);

			}

		} else {

			ds->ds_reserved = ds->ds_quota = 0;

		}

		if (err != 0 || (winner = dmu_buf_set_user_ie(dbuf, ds,

		    &ds->ds_phys, dsl_dataset_evict)) != NULL) {

			bplist_destroy(&ds->ds_pending_deadlist);

			dsl_deadlist_close(&ds->ds_deadlist);

			if (ds->ds_prev)

				dsl_dataset_rele(ds->ds_prev, ds);

			dsl_dir_rele(ds->ds_dir, ds);

			mutex_destroy(&ds->ds_lock);

			mutex_destroy(&ds->ds_opening_lock);

			refcount_destroy(&ds->ds_longholds);

			kmem_free(ds, sizeof (dsl_dataset_t));

			if (err != 0) {

				dmu_buf_rele(dbuf, tag);

				return (err);

			}

			ds = winner;

		} else {

			ds->ds_fsid_guid =

			    unique_insert(ds->ds_phys->ds_fsid_guid);

		}

	}

	ASSERT3P(ds->ds_dbuf, ==, dbuf);

	ASSERT3P(ds->ds_phys, ==, dbuf->db_data);

	ASSERT(ds->ds_phys->ds_prev_snap_obj != 0 ||

	    spa_version(dp->dp_spa) < SPA_VERSION_ORIGIN ||

	    dp->dp_origin_snap == NULL || ds == dp->dp_origin_snap);

	*dsp = ds;

	return (0);

}

int

dsl_dataset_hold(dsl_pool_t *dp, const char *name,

    void *tag, dsl_dataset_t **dsp)

{

	dsl_dir_t *dd;

	const char *snapname;

	uint64_t obj;

	int err = 0;

	err = dsl_dir_hold(dp, name, FTAG, &dd, &snapname);

	if (err != 0)

		return (err);

	ASSERT(dsl_pool_config_held(dp));

	obj = dd->dd_phys->dd_head_dataset_obj;

	if (obj != 0)

		err = dsl_dataset_hold_obj(dp, obj, tag, dsp);

	else

		err = SET_ERROR(ENOENT);

	/* we may be looking for a snapshot */

	if (err == 0 && snapname != NULL) {

		dsl_dataset_t *ds;

		if (*snapname++ != '@') {

			dsl_dataset_rele(*dsp, tag);

			dsl_dir_rele(dd, FTAG);

			return (SET_ERROR(ENOENT));

		}