/*

 * 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/zfs_context.h>

#include <sys/spa.h>

#include <sys/dmu.h>

#include <sys/zap.h>

#include <sys/arc.h>

#include <sys/stat.h>

#include <sys/resource.h>

#include <sys/zil.h>

#include <sys/zil_impl.h>

#include <sys/dsl_dataset.h>

#include <sys/vdev.h>

#include <sys/dmu_tx.h>

/*

 * The zfs intent log (ZIL) saves transaction records of system calls

 * that change the file system in memory with enough information

 * to be able to replay them. These are stored in memory until

 * either the DMU transaction group (txg) commits them to the stable pool

 * and they can be discarded, or they are flushed to the stable log

 * (also in the pool) due to a fsync, O_DSYNC or other synchronous

 * requirement. In the event of a panic or power fail then those log

 * records (transactions) are replayed.

 *

 * There is one ZIL per file system. Its on-disk (pool) format consists

 * of 3 parts:

 *

 * 	- ZIL header

 * 	- ZIL blocks

 * 	- ZIL records

 *

 * A log record holds a system call transaction. Log blocks can

 * hold many log records and the blocks are chained together.

 * Each ZIL block contains a block pointer (blkptr_t) to the next

 * ZIL block in the chain. The ZIL header points to the first

 * block in the chain. Note there is not a fixed place in the pool

 * to hold blocks. They are dynamically allocated and freed as

 * needed from the blocks available. Figure X shows the ZIL structure:

 */

/*

 * This global ZIL switch affects all pools

 */

int zil_disable = 0;	/* disable intent logging */

/*

 * Tunable parameter for debugging or performance analysis.  Setting

 * zfs_nocacheflush will cause corruption on power loss if a volatile

 * out-of-order write cache is enabled.

 */

boolean_t zfs_nocacheflush = B_FALSE;

static kmem_cache_t *zil_lwb_cache;

static int

zil_dva_compare(const void *x1, const void *x2)

{

	const dva_t *dva1 = x1;

	const dva_t *dva2 = x2;

	if (DVA_GET_VDEV(dva1) < DVA_GET_VDEV(dva2))

		return (-1);

	if (DVA_GET_VDEV(dva1) > DVA_GET_VDEV(dva2))

		return (1);

	if (DVA_GET_OFFSET(dva1) < DVA_GET_OFFSET(dva2))

		return (-1);

	if (DVA_GET_OFFSET(dva1) > DVA_GET_OFFSET(dva2))

		return (1);

	return (0);

}

static void

zil_dva_tree_init(avl_tree_t *t)

{

	avl_create(t, zil_dva_compare, sizeof (zil_dva_node_t),

	    offsetof(zil_dva_node_t, zn_node));

}

static void

zil_dva_tree_fini(avl_tree_t *t)

{

	zil_dva_node_t *zn;

	void *cookie = NULL;

	while ((zn = avl_destroy_nodes(t, &cookie)) != NULL)

		kmem_free(zn, sizeof (zil_dva_node_t));

	avl_destroy(t);

}

static int

zil_dva_tree_add(avl_tree_t *t, dva_t *dva)

{

	zil_dva_node_t *zn;

	avl_index_t where;

	if (avl_find(t, dva, &where) != NULL)

		return (EEXIST);

	zn = kmem_alloc(sizeof (zil_dva_node_t), KM_SLEEP);

	zn->zn_dva = *dva;

	avl_insert(t, zn, where);

	return (0);

}

static zil_header_t *

zil_header_in_syncing_context(zilog_t *zilog)

{

	return ((zil_header_t *)zilog->zl_header);

}

static void

zil_init_log_chain(zilog_t *zilog, blkptr_t *bp)

{

	zio_cksum_t *zc = &bp->blk_cksum;

	zc->zc_word[ZIL_ZC_GUID_0] = spa_get_random(-1ULL);

	zc->zc_word[ZIL_ZC_GUID_1] = spa_get_random(-1ULL);

	zc->zc_word[ZIL_ZC_OBJSET] = dmu_objset_id(zilog->zl_os);

	zc->zc_word[ZIL_ZC_SEQ] = 1ULL;

}

/*

 * Read a log block, make sure it's valid, and byteswap it if necessary.

 */

static int

zil_read_log_block(zilog_t *zilog, const blkptr_t *bp, arc_buf_t **abufpp)

{

	blkptr_t blk = *bp;

	zbookmark_t zb;

	uint32_t aflags = ARC_WAIT;

	int error;

	zb.zb_objset = bp->blk_cksum.zc_word[ZIL_ZC_OBJSET];

	zb.zb_object = 0;

	zb.zb_level = -1;

	zb.zb_blkid = bp->blk_cksum.zc_word[ZIL_ZC_SEQ];

	*abufpp = NULL;

	/*

	 * We shouldn't be doing any scrubbing while we're doing log

	 * replay, it's OK to not lock.

	 */

	error = arc_read_nolock(NULL, zilog->zl_spa, &blk,

	    arc_getbuf_func, abufpp, ZIO_PRIORITY_SYNC_READ, ZIO_FLAG_CANFAIL |

	    ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB, &aflags, &zb);

	if (error == 0) {

		char *data = (*abufpp)->b_data;

		uint64_t blksz = BP_GET_LSIZE(bp);

		zil_trailer_t *ztp = (zil_trailer_t *)(data + blksz) - 1;

		zio_cksum_t cksum = bp->blk_cksum;

		/*

		 * Sequence numbers should be... sequential.  The checksum

		 * verifier for the next block should be bp's checksum plus 1.

		 */

		cksum.zc_word[ZIL_ZC_SEQ]++;

		if (bcmp(&cksum, &ztp->zit_next_blk.blk_cksum, sizeof (cksum)))

			error = ESTALE;

		else if (BP_IS_HOLE(&ztp->zit_next_blk))

			error = ENOENT;

		else if (ztp->zit_nused > (blksz - sizeof (zil_trailer_t)))

			error = EOVERFLOW;

		if (error) {

			VERIFY(arc_buf_remove_ref(*abufpp, abufpp) == 1);

			*abufpp = NULL;

		}

	}

	dprintf("error %d on %llu:%llu\n", error, zb.zb_objset, zb.zb_blkid);

	return (error);

}

/*

 * Parse the intent log, and call parse_func for each valid record within.

 * Return the highest sequence number.

 */

uint64_t

zil_parse(zilog_t *zilog, zil_parse_blk_func_t *parse_blk_func,

    zil_parse_lr_func_t *parse_lr_func, void *arg, uint64_t txg)

{

	const zil_header_t *zh = zilog->zl_header;

	uint64_t claim_seq = zh->zh_claim_seq;

	uint64_t seq = 0;

	uint64_t max_seq = 0;

	blkptr_t blk = zh->zh_log;

	arc_buf_t *abuf;

	char *lrbuf, *lrp;

	zil_trailer_t *ztp;

	int reclen, error;

	if (BP_IS_HOLE(&blk))

		return (max_seq);

	/*

	 * Starting at the block pointed to by zh_log we read the log chain.

	 * For each block in the chain we strongly check that block to

	 * ensure its validity.  We stop when an invalid block is found.

	 * For each block pointer in the chain we call parse_blk_func().

	 * For each record in each valid block we call parse_lr_func().

	 * If the log has been claimed, stop if we encounter a sequence

	 * number greater than the highest claimed sequence number.

	 */

	zil_dva_tree_init(&zilog->zl_dva_tree);

	for (;;) {

		seq = blk.blk_cksum.zc_word[ZIL_ZC_SEQ];

		if (claim_seq != 0 && seq > claim_seq)

			break;

		ASSERT(max_seq < seq);

		max_seq = seq;

		error = zil_read_log_block(zilog, &blk, &abuf);

		if (parse_blk_func != NULL)

			parse_blk_func(zilog, &blk, arg, txg);

		if (error)

			break;

		lrbuf = abuf->b_data;

		ztp = (zil_trailer_t *)(lrbuf + BP_GET_LSIZE(&blk)) - 1;

		blk = ztp->zit_next_blk;

		if (parse_lr_func == NULL) {

			VERIFY(arc_buf_remove_ref(abuf, &abuf) == 1);

			continue;

		}

		for (lrp = lrbuf; lrp < lrbuf + ztp->zit_nused; lrp += reclen) {

			lr_t *lr = (lr_t *)lrp;

			reclen = lr->lrc_reclen;

			ASSERT3U(reclen, >=, sizeof (lr_t));

			parse_lr_func(zilog, lr, arg, txg);

		}

		VERIFY(arc_buf_remove_ref(abuf, &abuf) == 1);

	}

	zil_dva_tree_fini(&zilog->zl_dva_tree);

	return (max_seq);

}

/* ARGSUSED */

static void

zil_claim_log_block(zilog_t *zilog, blkptr_t *bp, void *tx, uint64_t first_txg)

{

	spa_t *spa = zilog->zl_spa;

	int err;

	/*

	 * Claim log block if not already committed and not already claimed.

	 */

	if (bp->blk_birth >= first_txg &&

	    zil_dva_tree_add(&zilog->zl_dva_tree, BP_IDENTITY(bp)) == 0) {

		err = zio_wait(zio_claim(NULL, spa, first_txg, bp, NULL, NULL));

		ASSERT(err == 0);

	}

}

static void

zil_claim_log_record(zilog_t *zilog, lr_t *lrc, void *tx, uint64_t first_txg)

{

	if (lrc->lrc_txtype == TX_WRITE) {

		lr_write_t *lr = (lr_write_t *)lrc;

		zil_claim_log_block(zilog, &lr->lr_blkptr, tx, first_txg);

	}

}

/* ARGSUSED */

static void

zil_free_log_block(zilog_t *zilog, blkptr_t *bp, void *tx, uint64_t claim_txg)

{

	zio_free_blk(zilog->zl_spa, bp, dmu_tx_get_txg(tx));

}

static void

zil_free_log_record(zilog_t *zilog, lr_t *lrc, void *tx, uint64_t claim_txg)

{

	/*

	 * If we previously claimed it, we need to free it.

	 */

	if (claim_txg != 0 && lrc->lrc_txtype == TX_WRITE) {

		lr_write_t *lr = (lr_write_t *)lrc;

		blkptr_t *bp = &lr->lr_blkptr;

		if (bp->blk_birth >= claim_txg &&

		    !zil_dva_tree_add(&zilog->zl_dva_tree, BP_IDENTITY(bp))) {

			(void) arc_free(NULL, zilog->zl_spa,

			    dmu_tx_get_txg(tx), bp, NULL, NULL, ARC_WAIT);

		}

	}

}

/*

 * Create an on-disk intent log.

 */

static void

zil_create(zilog_t *zilog)

{

	const zil_header_t *zh = zilog->zl_header;

	lwb_t *lwb;

	uint64_t txg = 0;

	dmu_tx_t *tx = NULL;

	blkptr_t blk;

	int error = 0;

	/*

	 * Wait for any previous destroy to complete.

	 */

	txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);

	ASSERT(zh->zh_claim_txg == 0);

	ASSERT(zh->zh_replay_seq == 0);

	blk = zh->zh_log;

	/*

	 * If we don't already have an initial log block, allocate one now.

	 */

	if (BP_IS_HOLE(&blk)) {

		tx = dmu_tx_create(zilog->zl_os);

		(void) dmu_tx_assign(tx, TXG_WAIT);

		dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);

		txg = dmu_tx_get_txg(tx);

		error = zio_alloc_blk(zilog->zl_spa, ZIL_MIN_BLKSZ, &blk,

		    NULL, txg);

		if (error == 0)

			zil_init_log_chain(zilog, &blk);

	}

	/*

	 * Allocate a log write buffer (lwb) for the first log block.

	 */

	if (error == 0) {

		lwb = kmem_cache_alloc(zil_lwb_cache, KM_SLEEP);

		lwb->lwb_zilog = zilog;

		lwb->lwb_blk = blk;

		lwb->lwb_nused = 0;

		lwb->lwb_sz = BP_GET_LSIZE(&lwb->lwb_blk);

		lwb->lwb_buf = zio_buf_alloc(lwb->lwb_sz);

		lwb->lwb_max_txg = txg;

		lwb->lwb_zio = NULL;

		mutex_enter(&zilog->zl_lock);

		list_insert_tail(&zilog->zl_lwb_list, lwb);

		mutex_exit(&zilog->zl_lock);

	}

	/*

	 * If we just allocated the first log block, commit our transaction

	 * and wait for zil_sync() to stuff the block poiner into zh_log.

	 * (zh is part of the MOS, so we cannot modify it in open context.)

	 */

	if (tx != NULL) {

		dmu_tx_commit(tx);

		txg_wait_synced(zilog->zl_dmu_pool, txg);

	}

	ASSERT(bcmp(&blk, &zh->zh_log, sizeof (blk)) == 0);

}

/*

 * In one tx, free all log blocks and clear the log header.

 * If keep_first is set, then we're replaying a log with no content.

 * We want to keep the first block, however, so that the first

 * synchronous transaction doesn't require a txg_wait_synced()

 * in zil_create().  We don't need to txg_wait_synced() here either

 * when keep_first is set, because both zil_create() and zil_destroy()

 * will wait for any in-progress destroys to complete.

 */

void

zil_destroy(zilog_t *zilog, boolean_t keep_first)

{

	const zil_header_t *zh = zilog->zl_header;

	lwb_t *lwb;

	dmu_tx_t *tx;

	uint64_t txg;

	/*

	 * Wait for any previous destroy to complete.

	 */

	txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);

	if (BP_IS_HOLE(&zh->zh_log))

		return;

	tx = dmu_tx_create(zilog->zl_os);

	(void) dmu_tx_assign(tx, TXG_WAIT);

	dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);

	txg = dmu_tx_get_txg(tx);

	mutex_enter(&zilog->zl_lock);

	/*

	 * It is possible for the ZIL to get the previously mounted zilog

	 * structure of the same dataset if quickly remounted and the dbuf

	 * eviction has not completed. In this case we can see a non

	 * empty lwb list and keep_first will be set. We fix this by

	 * clearing the keep_first. This will be slower but it's very rare.

	 */

	if (!list_is_empty(&zilog->zl_lwb_list) && keep_first)

		keep_first = B_FALSE;

	ASSERT3U(zilog->zl_destroy_txg, <, txg);

	zilog->zl_destroy_txg = txg;

	zilog->zl_keep_first = keep_first;

	if (!list_is_empty(&zilog->zl_lwb_list)) {

		ASSERT(zh->zh_claim_txg == 0);

		ASSERT(!keep_first);

		while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) {

			list_remove(&zilog->zl_lwb_list, lwb);

			if (lwb->lwb_buf != NULL)

				zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);

			zio_free_blk(zilog->zl_spa, &lwb->lwb_blk, txg);

			kmem_cache_free(zil_lwb_cache, lwb);

		}

	} else {

		if (!keep_first) {

			(void) zil_parse(zilog, zil_free_log_block,

			    zil_free_log_record, tx, zh->zh_claim_txg);

		}

	}

	mutex_exit(&zilog->zl_lock);

	dmu_tx_commit(tx);

}

/*

 * zil_rollback_destroy() is only called by the rollback code.

 * We already have a syncing tx. Rollback has exclusive access to the

 * dataset, so we don't have to worry about concurrent zil access.

 * The actual freeing of any log blocks occurs in zil_sync() later in

 * this txg syncing phase.

 */

void

zil_rollback_destroy(zilog_t *zilog, dmu_tx_t *tx)

{

	const zil_header_t *zh = zilog->zl_header;

	uint64_t txg;

	if (BP_IS_HOLE(&zh->zh_log))

		return;

	txg = dmu_tx_get_txg(tx);

	ASSERT3U(zilog->zl_destroy_txg, <, txg);

	zilog->zl_destroy_txg = txg;

	zilog->zl_keep_first = B_FALSE;

	/*

	 * Ensure there's no outstanding ZIL IO.  No lwbs or just the

	 * unused one that allocated in advance is ok.

	 */

	ASSERT(zilog->zl_lwb_list.list_head.list_next ==

	    zilog->zl_lwb_list.list_head.list_prev);

	(void) zil_parse(zilog, zil_free_log_block, zil_free_log_record,

	    tx, zh->zh_claim_txg);

}

int

zil_claim(char *osname, void *txarg)

{

	dmu_tx_t *tx = txarg;

	uint64_t first_txg = dmu_tx_get_txg(tx);

	zilog_t *zilog;

	zil_header_t *zh;

	objset_t *os;

	int error;

	error = dmu_objset_open(osname, DMU_OST_ANY, DS_MODE_USER, &os);

	if (error) {