/*

 * CDDL HEADER START

 *

 * The contents of this file are subject to the terms of the

 * Common Development and Distribution License, Version 1.0 only

 * (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 2005 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>

/*

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

 */

/*

 * These global ZIL switches affect all pools

 */

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

int zil_always = 0;	/* make every transaction synchronous */

int zil_purge = 0;	/* at pool open, just throw everything away */

int zil_noflush = 0;	/* don't flush write cache buffers on disks */

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

}

/*

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

 */

static int

zil_read_log_block(zilog_t *zilog, blkptr_t *bp, char *buf)

{

	uint64_t blksz = BP_GET_LSIZE(bp);

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

	zio_cksum_t cksum;

	int error;

	error = zio_wait(zio_read(NULL, zilog->zl_spa, bp, buf, blksz,

	    NULL, NULL, ZIO_PRIORITY_SYNC_READ,

	    ZIO_FLAG_CANFAIL | ZIO_FLAG_SPECULATIVE));

	if (error) {

		dprintf_bp(bp, "zilog %p bp %p read failed, error %d: ",

		    zilog, bp, error);

		return (error);

	}

	if (BP_SHOULD_BYTESWAP(bp))

		byteswap_uint64_array(buf, blksz);

	/*

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

	 * the next block should be: <logid[0], logid[1], objset id, seq + 1>.

	 */

	cksum = bp->blk_cksum;

	cksum.zc_word[3]++;

	if (bcmp(&cksum, &ztp->zit_next_blk.blk_cksum, sizeof (cksum)) != 0) {

		dprintf_bp(bp, "zilog %p bp %p stale pointer: ", zilog, bp);

		return (ESTALE);

	}

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

		dprintf_bp(bp, "zilog %p bp %p hole: ", zilog, bp);

		return (ENOENT);

	}

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

		dprintf("zilog %p bp %p nused exceeds blksz\n", zilog, bp);

		return (EOVERFLOW);

	}

	dprintf_bp(bp, "zilog %p bp %p good block: ", zilog, bp);

	return (0);

}

/*

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

 */

void

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)

{

	blkptr_t blk;

	char *lrbuf, *lrp;

	zil_trailer_t *ztp;

	int reclen, error;

	blk = zilog->zl_header->zh_log;

	if (BP_IS_HOLE(&blk))

		return;

	/*

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

	 */

	zil_dva_tree_init(&zilog->zl_dva_tree);

	lrbuf = zio_buf_alloc(SPA_MAXBLOCKSIZE);

	for (;;) {

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

		if (parse_blk_func != NULL)

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

		if (error)

			break;

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

		blk = ztp->zit_next_blk;

		if (parse_lr_func == NULL)

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

		}

	}

	zio_buf_free(lrbuf, SPA_MAXBLOCKSIZE);

	zil_dva_tree_fini(&zilog->zl_dva_tree);

}

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

	dprintf_bp(bp, "first_txg %llu: ", first_txg);

	/*

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

{

	lwb_t *lwb;

	uint64_t txg;

	dmu_tx_t *tx;

	blkptr_t blk;

	int error;

	ASSERT(zilog->zl_header->zh_claim_txg == 0);

	ASSERT(zilog->zl_header->zh_replay_seq == 0);

	/*

	 * Initialize the log header block.

	 */

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

	/*

	 * Allocate the first log block and assign its checksum verifier.

	 */

	error = zio_alloc_blk(zilog->zl_spa, ZIO_CHECKSUM_ZILOG,

	    ZIL_MIN_BLKSZ, &blk, txg);

	if (error == 0) {

		ZIO_SET_CHECKSUM(&blk.blk_cksum,

		    spa_get_random(-1ULL), spa_get_random(-1ULL),

		    dmu_objset_id(zilog->zl_os), 1ULL);

		/*

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

		 */

		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_seq = 0;

		lwb->lwb_state = UNWRITTEN;

		mutex_enter(&zilog->zl_lock);

		list_insert_tail(&zilog->zl_lwb_list, lwb);

		mutex_exit(&zilog->zl_lock);

	}

	dmu_tx_commit(tx);

	txg_wait_synced(zilog->zl_dmu_pool, txg);

}

/*

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

 */

void

zil_destroy(zilog_t *zilog)

{

	dmu_tx_t *tx;

	uint64_t txg;

	mutex_enter(&zilog->zl_destroy_lock);

	if (BP_IS_HOLE(&zilog->zl_header->zh_log)) {

		mutex_exit(&zilog->zl_destroy_lock);

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

	zil_parse(zilog, zil_free_log_block, zil_free_log_record, tx,

	    zilog->zl_header->zh_claim_txg);

	zilog->zl_destroy_txg = txg;

	dmu_tx_commit(tx);

	txg_wait_synced(zilog->zl_dmu_pool, txg);

	mutex_exit(&zilog->zl_destroy_lock);

}

void

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_STANDARD, &os);

	if (error) {

		cmn_err(CE_WARN, "can't process intent log for %s", osname);

		return;

	}

	zilog = dmu_objset_zil(os);

	zh = zilog->zl_header;

	/*

	 * Claim all log blocks if we haven't already done so.

	 */

	ASSERT3U(zh->zh_claim_txg, <=, first_txg);

	if (zh->zh_claim_txg == 0 && !BP_IS_HOLE(&zh->zh_log)) {

		zh->zh_claim_txg = first_txg;

		zil_parse(zilog, zil_claim_log_block, zil_claim_log_record,

		    tx, first_txg);

		dsl_dataset_dirty(dmu_objset_ds(os), tx);

	}

	ASSERT3U(first_txg, ==, (spa_last_synced_txg(zilog->zl_spa) + 1));

	dmu_objset_close(os);

}

void

zil_add_vdev(zilog_t *zilog, uint64_t vdev, uint64_t seq)

{

	zil_vdev_t *zv;

	if (zil_noflush)

		return;

	ASSERT(MUTEX_HELD(&zilog->zl_lock));

	zv = kmem_alloc(sizeof (zil_vdev_t), KM_SLEEP);

	zv->vdev = vdev;

	zv->seq = seq;

	list_insert_tail(&zilog->zl_vdev_list, zv);

}

void

zil_flush_vdevs(zilog_t *zilog, uint64_t seq)

{

	vdev_t *vd;

	zil_vdev_t *zv, *zv2;

	zio_t *zio;

	spa_t *spa;

	uint64_t vdev;

	if (zil_noflush)

		return;

	ASSERT(MUTEX_HELD(&zilog->zl_lock));

	spa = zilog->zl_spa;

	zio = NULL;

	while ((zv = list_head(&zilog->zl_vdev_list)) != NULL &&

	    zv->seq <= seq) {

		vdev = zv->vdev;

		list_remove(&zilog->zl_vdev_list, zv);

		kmem_free(zv, sizeof (zil_vdev_t));

		/*

		 * remove all chained entries <= seq with same vdev

		 */

		zv = list_head(&zilog->zl_vdev_list);

		while (zv && zv->seq <= seq) {

			zv2 = list_next(&zilog->zl_vdev_list, zv);

			if (zv->vdev == vdev) {

				list_remove(&zilog->zl_vdev_list, zv);

				kmem_free(zv, sizeof (zil_vdev_t));

			}

			zv = zv2;

		}

		/* flush the write cache for this vdev */

		mutex_exit(&zilog->zl_lock);

		if (zio == NULL)

			zio = zio_root(spa, NULL, NULL, ZIO_FLAG_CANFAIL);

		vd = vdev_lookup_top(spa, vdev);

		ASSERT(vd);

		(void) zio_nowait(zio_ioctl(zio, spa, vd, DKIOCFLUSHWRITECACHE,

		    NULL, NULL, ZIO_PRIORITY_NOW,

		    ZIO_FLAG_CANFAIL | ZIO_FLAG_DONT_RETRY));

		mutex_enter(&zilog->zl_lock);

	}

	/*

	 * Wait for all the flushes to complete.  Not all devices actually

	 * support the DKIOCFLUSHWRITECACHE ioctl, so it's OK if it fails.

	 */

	if (zio != NULL) {

		mutex_exit(&zilog->zl_lock);

		(void) zio_wait(zio);

		mutex_enter(&zilog->zl_lock);

	}

}

/*

 * Function called when a log block write completes

 */

static void

zil_lwb_write_done(zio_t *zio)

{

	lwb_t *prev;

	lwb_t *lwb = zio->io_private;

	zilog_t *zilog = lwb->lwb_zilog;

	uint64_t max_seq;

	/*

	 * Now that we've written this log block, we have a stable pointer

	 * to the next block in the chain, so it's OK to let the txg in

	 * which we allocated the next block sync.

	 */

	txg_rele_to_sync(&lwb->lwb_txgh);

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

	mutex_enter(&zilog->zl_lock);

	lwb->lwb_buf = NULL;

	if (zio->io_error) {

		zilog->zl_log_error = B_TRUE;

		mutex_exit(&zilog->zl_lock);

		cv_broadcast(&zilog->zl_cv_seq);

		return;

	}

	prev = list_prev(&zilog->zl_lwb_list, lwb);

	if (prev && prev->lwb_state != SEQ_COMPLETE) {

		/* There's an unwritten buffer in the chain before this one */

		lwb->lwb_state = SEQ_INCOMPLETE;

		mutex_exit(&zilog->zl_lock);

		return;

	}

	max_seq = lwb->lwb_seq;

	lwb->lwb_state = SEQ_COMPLETE;

	/*

	 * We must also follow up the chain for already written buffers

	 * to see if we can set zl_ss_seq even higher.

	 */

	while (lwb = list_next(&zilog->zl_lwb_list, lwb)) {

		if (lwb->lwb_state != SEQ_INCOMPLETE)

			break;

		lwb->lwb_state = SEQ_COMPLETE;

		/* lwb_seq will be zero if we've written an empty buffer */

		if (lwb->lwb_seq) {

			ASSERT3U(max_seq, <, lwb->lwb_seq);

			max_seq = lwb->lwb_seq;

		}

	}

	zilog->zl_ss_seq = MAX(max_seq, zilog->zl_ss_seq);

	mutex_exit(&zilog->zl_lock);