/*

 * 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 2008 Sun Microsystems, Inc.  All rights reserved.

 * Use is subject to license terms.

 */

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

#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_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/sunddi.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_checkfunc_t dsl_dataset_rollback_check;

static dsl_syncfunc_t dsl_dataset_rollback_sync;

static dsl_syncfunc_t dsl_dataset_set_reservation_sync;

#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, blkptr_t *bp, dmu_tx_t *tx)

{

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

	int compressed = BP_GET_PSIZE(bp);

	int uncompressed = BP_GET_UCSIZE(bp);

	int64_t delta;

	dprintf_bp(bp, "born, ds=%p\n", 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);

	ASSERT3U(BP_GET_TYPE(bp), <, DMU_OT_NUMTYPES);

	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,

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

	delta = parent_delta(ds, used);

	ds->ds_phys->ds_used_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, delta, compressed, uncompressed, tx);

}

int

dsl_dataset_block_kill(dsl_dataset_t *ds, blkptr_t *bp, zio_t *pio,

    dmu_tx_t *tx)

{

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

	int compressed = BP_GET_PSIZE(bp);

	int uncompressed = BP_GET_UCSIZE(bp);

	ASSERT(dmu_tx_is_syncing(tx));

	/* No block pointer => nothing to free */

	if (BP_IS_HOLE(bp))

		return (0);

	ASSERT(used > 0);

	if (ds == NULL) {

		int err;

		/*

		 * Account for the meta-objset space in its placeholder

		 * dataset.

		 */

		err = dsl_free(pio, tx->tx_pool,

		    tx->tx_txg, bp, NULL, NULL, pio ? ARC_NOWAIT: ARC_WAIT);

		ASSERT(err == 0);

		dsl_dir_diduse_space(tx->tx_pool->dp_mos_dir,

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

	dmu_buf_will_dirty(ds->ds_dbuf, tx);

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

		int err;

		int64_t delta;

		dprintf_bp(bp, "freeing: %s", "");

		err = dsl_free(pio, tx->tx_pool,

		    tx->tx_txg, bp, NULL, NULL, pio ? ARC_NOWAIT: ARC_WAIT);

		ASSERT(err == 0);

		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,

		    delta, -compressed, -uncompressed, tx);

	} else {

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

		VERIFY(0 == bplist_enqueue(&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);

		}

	}

	mutex_enter(&ds->ds_lock);

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

	ds->ds_phys->ds_used_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));

}

int

dsl_dataset_block_freeable(dsl_dataset_t *ds, uint64_t blk_birth)

{

	return (blk_birth > dsl_dataset_prev_snap_txg(ds));

}

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

	dprintf_ds(ds, "evicting %s\n", "");

	unique_remove(ds->ds_fsid_guid);

	if (ds->ds_user_ptr != NULL)

		ds->ds_user_evict_func(ds, ds->ds_user_ptr);

	if (ds->ds_prev) {

		dsl_dataset_drop_ref(ds->ds_prev, ds);

		ds->ds_prev = NULL;

	}

	bplist_close(&ds->ds_deadlist);

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

	mutex_destroy(&ds->ds_deadlist.bpl_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;

	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;

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

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

		mutex_init(&ds->ds_deadlist.bpl_lock, NULL, MUTEX_DEFAULT,

		    NULL);

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

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

		err = bplist_open(&ds->ds_deadlist,

		    mos, ds->ds_phys->ds_deadlist_obj);

		if (err == 0) {

			err = dsl_dir_open_obj(dp,

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

		}

		if (err) {

			/*

			 * we don't really need to close the blist if we

			 * just opened it.

			 */

			mutex_destroy(&ds->ds_lock);

			mutex_destroy(&ds->ds_opening_lock);

			mutex_destroy(&ds->ds_deadlist.bpl_lock);

			rw_destroy(&ds->ds_rwlock);

			cv_destroy(&ds->ds_exclusive_cv);

			kmem_free(ds, sizeof (dsl_dataset_t));

			dmu_buf_rele(dbuf, tag);

			return (err);

		}

		if (ds->ds_dir->dd_phys->dd_head_dataset_obj == dsobj) {

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

			    "refreservation", sizeof (uint64_t), 1,

			    &ds->ds_reserved, NULL);

			if (err == 0) {

				    "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_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_opening_lock);

			mutex_destroy(&ds->ds_deadlist.bpl_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);

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

	mutex_enter(&ds->ds_lock);

	if (!dsl_pool_sync_context(dp) && DSL_DATASET_IS_DESTROYED(ds)) {

		mutex_exit(&ds->ds_lock);

		dmu_buf_rele(ds->ds_dbuf, tag);

		return (ENOENT);

	}

	mutex_exit(&ds->ds_lock);

	*dsp = ds;

	return (0);

}

static int

dsl_dataset_hold_ref(dsl_dataset_t *ds, void *tag)

{

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

	/*

	 * In syncing context we don't want the rwlock lock: there

	 * may be an existing writer waiting for sync phase to

	 * finish.  We don't need to worry about such writers, since

	 * sync phase is single-threaded, so the writer can't be

	 * doing anything while we are active.

	 */

	if (dsl_pool_sync_context(dp)) {

		ASSERT(!DSL_DATASET_IS_DESTROYED(ds));

		return (0);

	}

	/*

	 * Normal users will hold the ds_rwlock as a READER until they

	 * are finished (i.e., call dsl_dataset_rele()).  "Owners" will

	 * drop their READER lock after they set the ds_owner field.

	 *

	 * If the dataset is being destroyed, the destroy thread will

	 * obtain a WRITER lock for exclusive access after it's done its

	 * open-context work and then change the ds_owner to

	 * dsl_reaper once destruction is assured.  So threads

	 * may block here temporarily, until the "destructability" of

	 * the dataset is determined.

	 */

	ASSERT(!RW_WRITE_HELD(&dp->dp_config_rwlock));

	mutex_enter(&ds->ds_lock);

	while (!rw_tryenter(&ds->ds_rwlock, RW_READER)) {

		rw_exit(&dp->dp_config_rwlock);

		cv_wait(&ds->ds_exclusive_cv, &ds->ds_lock);

		if (DSL_DATASET_IS_DESTROYED(ds)) {

			mutex_exit(&ds->ds_lock);

			dsl_dataset_drop_ref(ds, tag);

			rw_enter(&dp->dp_config_rwlock, RW_READER);

			return (ENOENT);

		}

		rw_enter(&dp->dp_config_rwlock, RW_READER);

	}