/*

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

static char *dsl_reaper = "the grim reaper";

static dsl_checkfunc_t dsl_dataset_destroy_begin_check;

static dsl_syncfunc_t dsl_dataset_destroy_begin_sync;

static dsl_syncfunc_t dsl_dataset_set_reservation_sync;

#define	SWITCH64(x, y) \

	{ \

		uint64_t __tmp = (x); \

		(x) = (y); \

		(y) = __tmp; \

	}

#define	DS_REF_MAX	(1ULL << 62)

#define	DSL_DEADLIST_BLOCKSIZE	SPA_MAXBLOCKSIZE

#define	DSL_DATASET_IS_DESTROYED(ds)	((ds)->ds_owner == dsl_reaper)

/*

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

	if (ds == NULL) {

		/*

		 * Account for the meta-objset space in its placeholder

		 * dsl_dir.

		 */

		ASSERT3U(compressed, ==, uncompressed); /* it's all metadata */

		dsl_dir_diduse_space(tx->tx_pool->dp_mos_dir, DD_USED_HEAD,

		    used, compressed, uncompressed, tx);

		dsl_dir_dirty(tx->tx_pool->dp_mos_dir, tx);

		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_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) {

		/*

		 * Account for the meta-objset space in its placeholder

		 * dataset.

		 */

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

		dsl_dir_diduse_space(tx->tx_pool->dp_mos_dir, DD_USED_HEAD,

		    -used, -compressed, -uncompressed, tx);

		dsl_dir_dirty(tx->tx_pool->dp_mos_dir, tx);

		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_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 || DSL_DATASET_IS_DESTROYED(ds));

	unique_remove(ds->ds_fsid_guid);

	if (ds->ds_objset != NULL)

		dmu_objset_evict(ds->ds_objset);

	if (ds->ds_prev) {

		dsl_dataset_drop_ref(ds->ds_prev, ds);

		ds->ds_prev = NULL;

	}

	bplist_destroy(&ds->ds_pending_deadlist);

	if (db != NULL) {

		dsl_deadlist_close(&ds->ds_deadlist);

	} else {

		ASSERT(ds->ds_deadlist.dl_dbuf == NULL);

		ASSERT(!ds->ds_deadlist.dl_oldfmt);

	}

	if (ds->ds_dir)

		dsl_dir_close(ds->ds_dir, ds);

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

	mutex_destroy(&ds->ds_lock);

	mutex_destroy(&ds->ds_recvlock);

	mutex_destroy(&ds->ds_opening_lock);

	rw_destroy(&ds->ds_rwlock);

	cv_destroy(&ds->ds_exclusive_cv);

	kmem_free(ds, sizeof (dsl_dataset_t));

}

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

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

}

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

}

static int

dsl_dataset_snap_remove(dsl_dataset_t *ds, 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);

}

static int

dsl_dataset_get_ref(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(RW_LOCK_HELD(&dp->dp_config_rwlock) ||

	    dsl_pool_sync_context(dp));

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

	if (err)

		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)

		return (EINVAL);

	ds = dmu_buf_get_user(dbuf);

	if (ds == NULL) {

		dsl_dataset_t *winner;

		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_recvlock, NULL, MUTEX_DEFAULT, NULL);

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

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

		rw_init(&ds->ds_rwlock, 0, 0, 0);

		cv_init(&ds->ds_exclusive_cv, NULL, CV_DEFAULT, NULL);

		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_open_obj(dp,

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

		}

		if (err) {

			mutex_destroy(&ds->ds_lock);

			mutex_destroy(&ds->ds_recvlock);

			mutex_destroy(&ds->ds_opening_lock);

			rw_destroy(&ds->ds_rwlock);

			cv_destroy(&ds->ds_exclusive_cv);

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

				err = dsl_dataset_get_ref(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)) {

			/*

			 * In sync context, we're called with either no lock

			 * or with the write lock.  If we're not syncing,

			 * we're always called with the read lock held.

			 */

			boolean_t need_lock =

			    !RW_WRITE_HELD(&dp->dp_config_rwlock) &&

			    dsl_pool_sync_context(dp);

			if (need_lock)

				rw_enter(&dp->dp_config_rwlock, RW_READER);

			err = dsl_prop_get_ds(ds,

			    "refreservation", sizeof (uint64_t), 1,

			    &ds->ds_reserved, NULL);

			if (err == 0) {

				err = dsl_prop_get_ds(ds,

				    "refquota", sizeof (uint64_t), 1,

				    &ds->ds_quota, NULL);

			}

			if (need_lock)

				rw_exit(&dp->dp_config_rwlock);

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

		}

		if (err || winner) {

			bplist_destroy(&ds->ds_pending_deadlist);

			dsl_deadlist_close(&ds->ds_deadlist);

			if (ds->ds_prev)

				dsl_dataset_drop_ref(ds->ds_prev, ds);

			dsl_dir_close(ds->ds_dir, ds);

			mutex_destroy(&ds->ds_lock);

			mutex_destroy(&ds->ds_recvlock);

			mutex_destroy(&ds->ds_opening_lock);

			rw_destroy(&ds->ds_rwlock);

			cv_destroy(&ds->ds_exclusive_cv);

			kmem_free(ds, sizeof (dsl_dataset_t));

			if (err) {

				dmu_buf_rele(dbuf, tag);