/*

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

 * Use is subject to license terms.

 */

#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 boolean_t zil_empty(zilog_t *zilog);

static int

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

{

	const dva_t *dva1 = &((zil_bp_node_t *)x1)->zn_dva;

	const dva_t *dva2 = &((zil_bp_node_t *)x2)->zn_dva;

	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_bp_tree_init(zilog_t *zilog)

{

	avl_create(&zilog->zl_bp_tree, zil_bp_compare,

	    sizeof (zil_bp_node_t), offsetof(zil_bp_node_t, zn_node));

}

static void

zil_bp_tree_fini(zilog_t *zilog)

{

	avl_tree_t *t = &zilog->zl_bp_tree;

	zil_bp_node_t *zn;

	void *cookie = NULL;

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

		kmem_free(zn, sizeof (zil_bp_node_t));

	avl_destroy(t);

}

int

zil_bp_tree_add(zilog_t *zilog, const blkptr_t *bp)

{

	avl_tree_t *t = &zilog->zl_bp_tree;

	const dva_t *dva = BP_IDENTITY(bp);

	zil_bp_node_t *zn;

	avl_index_t where;

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

		return (EEXIST);

	zn = kmem_alloc(sizeof (zil_bp_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 and make sure it's valid.

 */

static int

zil_read_log_block(zilog_t *zilog, const blkptr_t *bp, blkptr_t *nbp, void *dst)

{

	enum zio_flag zio_flags = ZIO_FLAG_CANFAIL;

	uint32_t aflags = ARC_WAIT;

	arc_buf_t *abuf = NULL;

	zbookmark_t zb;

	int error;

	if (zilog->zl_header->zh_claim_txg == 0)

		zio_flags |= ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB;

	if (!(zilog->zl_header->zh_flags & ZIL_CLAIM_LR_SEQ_VALID))

		zio_flags |= ZIO_FLAG_SPECULATIVE;

	SET_BOOKMARK(&zb, bp->blk_cksum.zc_word[ZIL_ZC_OBJSET],

	    ZB_ZIL_OBJECT, ZB_ZIL_LEVEL, bp->blk_cksum.zc_word[ZIL_ZC_SEQ]);

	error = arc_read_nolock(NULL, zilog->zl_spa, bp, arc_getbuf_func, &abuf,

	    ZIO_PRIORITY_SYNC_READ, zio_flags, &aflags, &zb);

	if (error == 0) {

		char *data = abuf->b_data;

		uint64_t size = BP_GET_LSIZE(bp);

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

		zio_cksum_t cksum = bp->blk_cksum;

		bcopy(data, dst, size);

		*nbp = ztp->zit_next_blk;

		/*

		 * Validate the checksummed log block.

		 *

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

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

		 *

		 * Also check the log chain linkage and size used.

		 */

		cksum.zc_word[ZIL_ZC_SEQ]++;

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

		    sizeof (cksum)) || BP_IS_HOLE(&ztp->zit_next_blk) ||

		    (ztp->zit_nused > (size - sizeof (zil_trailer_t))))

			error = ECKSUM;

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

	}

	return (error);

}

/*

 * Read a TX_WRITE log data block.

 */

static int

zil_read_log_data(zilog_t *zilog, const lr_write_t *lr, void *wbuf)

{

	enum zio_flag zio_flags = ZIO_FLAG_CANFAIL;

	const blkptr_t *bp = &lr->lr_blkptr;

	uint32_t aflags = ARC_WAIT;

	arc_buf_t *abuf = NULL;

	zbookmark_t zb;

	int error;

	if (BP_IS_HOLE(bp)) {

		if (wbuf != NULL)

			bzero(wbuf, MAX(BP_GET_LSIZE(bp), lr->lr_length));

		return (0);

	}

	if (zilog->zl_header->zh_claim_txg == 0)

		zio_flags |= ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB;

	SET_BOOKMARK(&zb, dmu_objset_id(zilog->zl_os), lr->lr_foid,

	    ZB_ZIL_LEVEL, lr->lr_offset / BP_GET_LSIZE(bp));

	error = arc_read_nolock(NULL, zilog->zl_spa, bp, arc_getbuf_func, &abuf,

	    ZIO_PRIORITY_SYNC_READ, zio_flags, &aflags, &zb);

	if (error == 0) {

		if (wbuf != NULL)

			bcopy(abuf->b_data, wbuf, arc_buf_size(abuf));

		(void) arc_buf_remove_ref(abuf, &abuf);

	}

	return (error);

}

/*

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

 */

int

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;

	boolean_t claimed = !!zh->zh_claim_txg;

	uint64_t claim_blk_seq = claimed ? zh->zh_claim_blk_seq : UINT64_MAX;

	uint64_t claim_lr_seq = claimed ? zh->zh_claim_lr_seq : UINT64_MAX;

	uint64_t max_blk_seq = 0;

	uint64_t max_lr_seq = 0;

	uint64_t blk_count = 0;

	uint64_t lr_count = 0;

	blkptr_t blk, next_blk;

	char *lrbuf, *lrp;

	int error = 0;

	/*

	 * Old logs didn't record the maximum zh_claim_lr_seq.

	 */

	if (!(zh->zh_flags & ZIL_CLAIM_LR_SEQ_VALID))

		claim_lr_seq = UINT64_MAX;

	/*

	 * 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.

	 */

	lrbuf = zio_buf_alloc(SPA_MAXBLOCKSIZE);

	zil_bp_tree_init(zilog);

	for (blk = zh->zh_log; !BP_IS_HOLE(&blk); blk = next_blk) {

		zil_trailer_t *ztp =

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

		uint64_t blk_seq = blk.blk_cksum.zc_word[ZIL_ZC_SEQ];

		int reclen;

		if (blk_seq > claim_blk_seq)

			break;

		if ((error = parse_blk_func(zilog, &blk, arg, txg)) != 0)

			break;

		ASSERT(max_blk_seq < blk_seq);

		max_blk_seq = blk_seq;

		blk_count++;

		if (max_lr_seq == claim_lr_seq && max_blk_seq == claim_blk_seq)

			break;

		error = zil_read_log_block(zilog, &blk, &next_blk, lrbuf);

		if (error)

			break;

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

			if (lr->lrc_seq > claim_lr_seq)

				goto done;

			if ((error = parse_lr_func(zilog, lr, arg, txg)) != 0)

				goto done;

			ASSERT(max_lr_seq < lr->lrc_seq);

			max_lr_seq = lr->lrc_seq;

			lr_count++;

		}

	}

done:

	zilog->zl_parse_error = error;

	zilog->zl_parse_blk_seq = max_blk_seq;

	zilog->zl_parse_lr_seq = max_lr_seq;

	zilog->zl_parse_blk_count = blk_count;

	zilog->zl_parse_lr_count = lr_count;

	ASSERT(!claimed || !(zh->zh_flags & ZIL_CLAIM_LR_SEQ_VALID) ||

	    (max_blk_seq == claim_blk_seq && max_lr_seq == claim_lr_seq));

	zil_bp_tree_fini(zilog);

	zio_buf_free(lrbuf, SPA_MAXBLOCKSIZE);

	return (error);

}

static int

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

{

	/*

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

	 * If tx == NULL, just verify that the block is claimable.

	 */

	if (bp->blk_birth < first_txg || zil_bp_tree_add(zilog, bp) != 0)

		return (0);

	return (zio_wait(zio_claim(NULL, zilog->zl_spa,

	    tx == NULL ? 0 : first_txg, bp, spa_claim_notify, NULL,

	    ZIO_FLAG_CANFAIL | ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB)));

}

static int

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

{

	lr_write_t *lr = (lr_write_t *)lrc;

	int error;

	if (lrc->lrc_txtype != TX_WRITE)

		return (0);

	/*

	 * If the block is not readable, don't claim it.  This can happen

	 * in normal operation when a log block is written to disk before

	 * some of the dmu_sync() blocks it points to.  In this case, the

	 * transaction cannot have been committed to anyone (we would have

	 * waited for all writes to be stable first), so it is semantically

	 * correct to declare this the end of the log.

	 */

	if (lr->lr_blkptr.blk_birth >= first_txg &&

	    (error = zil_read_log_data(zilog, lr, NULL)) != 0)

		return (error);

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

}

/* ARGSUSED */

static int

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

{

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

	return (0);

}

static int

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

{

	lr_write_t *lr = (lr_write_t *)lrc;

	blkptr_t *bp = &lr->lr_blkptr;

	/*

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

	 */

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

	    bp->blk_birth >= claim_txg && zil_bp_tree_add(zilog, bp) == 0)

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

	return (0);

}

/*

 * 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 or we have one

	 * but it's the wrong endianness then allocate one.

	 */

	if (BP_IS_HOLE(&blk) || BP_SHOULD_BYTESWAP(&blk)) {

		tx = dmu_tx_create(zilog->zl_os);

		VERIFY(dmu_tx_assign(tx, TXG_WAIT) == 0);

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

		txg = dmu_tx_get_txg(tx);

		if (!BP_IS_HOLE(&blk)) {

			zio_free_zil(zilog->zl_spa, txg, &blk);

			BP_ZERO(&blk);

		}

		error = zio_alloc_zil(zilog->zl_spa, txg, &blk, NULL,

		    ZIL_MIN_BLKSZ, zilog->zl_logbias == ZFS_LOGBIAS_LATENCY);

		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;

		lwb->lwb_tx = 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);

	zilog->zl_old_header = *zh;		/* debugging aid */

	if (BP_IS_HOLE(&zh->zh_log))

		return;