/*

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

#include <sys/dmu_traverse.h>

#include <sys/dsl_dataset.h>

#include <sys/dsl_dir.h>

#include <sys/dsl_pool.h>

#include <sys/dnode.h>

#include <sys/spa.h>

#include <sys/zio.h>

#include <sys/dmu_impl.h>

#define	BP_SPAN_SHIFT(level, width)	((level) * (width))

#define	BP_EQUAL(b1, b2)				\

	(DVA_EQUAL(BP_IDENTITY(b1), BP_IDENTITY(b2)) &&	\

	(b1)->blk_birth == (b2)->blk_birth)

/*

 * Compare two bookmarks.

 *

 * For ADVANCE_PRE, the visitation order is:

 *

 *	objset 0, 1, 2, ..., ZB_MAXOBJSET.

 *	object 0, 1, 2, ..., ZB_MAXOBJECT.

 *	blkoff 0, 1, 2, ...

 *	level ZB_MAXLEVEL, ..., 2, 1, 0.

 *

 * where blkoff = blkid << BP_SPAN_SHIFT(level, width), and thus a valid

 * ordering vector is:

 *

 *	< objset, object, blkoff, -level >

 *

 * For ADVANCE_POST, the starting offsets aren't sequential but ending

 * offsets [blkoff = (blkid + 1) << BP_SPAN_SHIFT(level, width)] are.

 * The visitation order is:

 *

 *	objset 1, 2, ..., ZB_MAXOBJSET, 0.

 *	object 1, 2, ..., ZB_MAXOBJECT, 0.

 *	blkoff 1, 2, ...

 *	level 0, 1, 2, ..., ZB_MAXLEVEL.

 *

 * and thus a valid ordering vector is:

 *

 *	< objset - 1, object - 1, blkoff, level >

 *

 * Both orderings can be expressed as:

 *

 *	< objset + bias, object + bias, blkoff, level ^ bias >

 *

 * where 'bias' is either 0 or -1 (for ADVANCE_PRE or ADVANCE_POST)

 * and 'blkoff' is (blkid - bias) << BP_SPAN_SHIFT(level, wshift).

 *

 * Special case: an objset's osphys is represented as level -1 of object 0.

 * It is always either the very first or very last block we visit in an objset.

 * Therefore, if either bookmark's level is -1, level alone determines order.

 */

static int

compare_bookmark(zbookmark_t *szb, zbookmark_t *ezb, dnode_phys_t *dnp,

    int advance)

{

	int bias = (advance & ADVANCE_PRE) ? 0 : -1;

	uint64_t sblkoff, eblkoff;

	int slevel, elevel, wshift;

	if (szb->zb_objset + bias < ezb->zb_objset + bias)

		return (-1);

	if (szb->zb_objset + bias > ezb->zb_objset + bias)

		return (1);

	slevel = szb->zb_level;

	elevel = ezb->zb_level;

	if ((slevel | elevel) < 0)

		return ((slevel ^ bias) - (elevel ^ bias));

	if (szb->zb_object + bias < ezb->zb_object + bias)

		return (-1);

	if (szb->zb_object + bias > ezb->zb_object + bias)

		return (1);

	if (dnp == NULL)

		return (0);

	wshift = dnp->dn_indblkshift - SPA_BLKPTRSHIFT;

	sblkoff = (szb->zb_blkid - bias) << BP_SPAN_SHIFT(slevel, wshift);

	eblkoff = (ezb->zb_blkid - bias) << BP_SPAN_SHIFT(elevel, wshift);

	if (sblkoff < eblkoff)

		return (-1);

	if (sblkoff > eblkoff)

		return (1);

	return ((elevel ^ bias) - (slevel ^ bias));

}

#define	SET_BOOKMARK(zb, objset, object, level, blkid)	\

{							\

	(zb)->zb_objset = objset;			\

	(zb)->zb_object = object;			\

	(zb)->zb_level = level;				\

	(zb)->zb_blkid = blkid;				\

}

#define	SET_BOOKMARK_LB(zb, level, blkid)		\

{							\

	(zb)->zb_level = level;				\

	(zb)->zb_blkid = blkid;				\

}

static int

advance_objset(zseg_t *zseg, uint64_t objset, int advance)

{

	zbookmark_t *zb = &zseg->seg_start;

	if (advance & ADVANCE_PRE) {

		if (objset >= ZB_MAXOBJSET)

			return (ERANGE);

		SET_BOOKMARK(zb, objset, 0, -1, 0);

	} else {

		if (objset >= ZB_MAXOBJSET)

			objset = 0;

		SET_BOOKMARK(zb, objset, 1, 0, 0);

	}

	if (compare_bookmark(zb, &zseg->seg_end, NULL, advance) > 0)

		return (ERANGE);

	return (EAGAIN);

}

static int

advance_object(zseg_t *zseg, uint64_t object, int advance)

{

	zbookmark_t *zb = &zseg->seg_start;

	if (advance & ADVANCE_PRE) {

		if (object >= ZB_MAXOBJECT) {

			SET_BOOKMARK(zb, zb->zb_objset + 1, 0, -1, 0);

		} else {

			SET_BOOKMARK(zb, zb->zb_objset, object, ZB_MAXLEVEL, 0);

		}

	} else {

		if (zb->zb_object == 0) {

			SET_BOOKMARK(zb, zb->zb_objset, 0, -1, 0);

		} else {

			if (object >= ZB_MAXOBJECT)

				object = 0;

			SET_BOOKMARK(zb, zb->zb_objset, object, 0, 0);

		}

	}

	if (compare_bookmark(zb, &zseg->seg_end, NULL, advance) > 0)

		return (ERANGE);

	return (EAGAIN);

}

static int

advance_from_osphys(zseg_t *zseg, int advance)

{

	zbookmark_t *zb = &zseg->seg_start;

	ASSERT(zb->zb_object == 0);

	ASSERT(zb->zb_level == -1);

	ASSERT(zb->zb_blkid == 0);

	if (advance & ADVANCE_PRE) {

		SET_BOOKMARK_LB(zb, ZB_MAXLEVEL, 0);

	} else {

		if (zb->zb_objset == 0)

			return (ERANGE);

		SET_BOOKMARK(zb, zb->zb_objset + 1, 1, 0, 0);

	}

	if (compare_bookmark(zb, &zseg->seg_end, NULL, advance) > 0)

		return (ERANGE);

	return (EAGAIN);

}

static int

advance_block(zseg_t *zseg, dnode_phys_t *dnp, int rc, int advance)

{

	zbookmark_t *zb = &zseg->seg_start;

	int wshift = dnp->dn_indblkshift - SPA_BLKPTRSHIFT;

	int maxlevel = dnp->dn_nlevels - 1;

	int level = zb->zb_level;

	uint64_t blkid = zb->zb_blkid;

	if (advance & ADVANCE_PRE) {

		if (level > 0 && rc == 0) {

			level--;

			blkid <<= wshift;

		} else {

			blkid++;

			if ((blkid << BP_SPAN_SHIFT(level, wshift)) >

			    dnp->dn_maxblkid)

				return (ERANGE);

			while (level < maxlevel) {

				if (P2PHASE(blkid, 1ULL << wshift))

					break;

				blkid >>= wshift;

				level++;

			}

		}

	} else {

		if (level >= maxlevel || P2PHASE(blkid + 1, 1ULL << wshift)) {

			blkid = (blkid + 1) << BP_SPAN_SHIFT(level, wshift);

			level = 0;

		} else {

			blkid >>= wshift;

			level++;

		}

		while ((blkid << BP_SPAN_SHIFT(level, wshift)) >

		    dnp->dn_maxblkid) {

			if (level == maxlevel)

				return (ERANGE);

			blkid >>= wshift;

			level++;

		}

	}

	SET_BOOKMARK_LB(zb, level, blkid);

	if (compare_bookmark(zb, &zseg->seg_end, dnp, advance) > 0)

		return (ERANGE);

	return (EAGAIN);

}

static int

traverse_callback(traverse_handle_t *th, zseg_t *zseg, traverse_blk_cache_t *bc)

{

	/*

	 * Before we issue the callback, prune against maxtxg.

	 *

	 * We prune against mintxg before we get here because it's a big win.

	 * If a given block was born in txg 37, then we know that the entire

	 * subtree below that block must have been born in txg 37 or earlier.

	 * We can therefore lop off huge branches of the tree as we go.

	 *

	 * There's no corresponding optimization for maxtxg because knowing

	 * that bp->blk_birth >= maxtxg doesn't imply anything about the bp's

	 * children.  In fact, the copy-on-write design of ZFS ensures that

	 * top-level blocks will pretty much always be new.

	 *

	 * Therefore, in the name of simplicity we don't prune against

	 * maxtxg until the last possible moment -- that being right now.

	 */

	if (bc->bc_errno == 0 && bc->bc_blkptr.blk_birth >= zseg->seg_maxtxg)

		return (0);

	/*

	 * Debugging: verify that the order we visit things agrees with the

	 * order defined by compare_bookmark().  We don't check this for

	 * log blocks because there's no defined ordering for them; they're

	 * always visited (or not) as part of visiting the objset_phys_t.

	 */

	if (bc->bc_errno == 0 && bc != &th->th_zil_cache) {

		zbookmark_t *zb = &bc->bc_bookmark;

		zbookmark_t *szb = &zseg->seg_start;

		zbookmark_t *ezb = &zseg->seg_end;

		zbookmark_t *lzb = &th->th_lastcb;

		dnode_phys_t *dnp = bc->bc_dnode;

		ASSERT(compare_bookmark(zb, ezb, dnp, th->th_advance) <= 0);

		ASSERT(compare_bookmark(zb, szb, dnp, th->th_advance) == 0);

		ASSERT(compare_bookmark(lzb, zb, dnp, th->th_advance) < 0 ||

		    lzb->zb_level == ZB_NO_LEVEL);

		*lzb = *zb;

	}

	th->th_callbacks++;

	return (th->th_func(bc, th->th_spa, th->th_arg));

}

static int

traverse_read(traverse_handle_t *th, traverse_blk_cache_t *bc, blkptr_t *bp,

	dnode_phys_t *dnp)

{

	zbookmark_t *zb = &bc->bc_bookmark;

	int error;

	th->th_hits++;

	bc->bc_dnode = dnp;

	bc->bc_errno = 0;

	if (BP_EQUAL(&bc->bc_blkptr, bp))

		return (0);

	bc->bc_blkptr = *bp;

	if (bc->bc_data == NULL)

		return (0);

	if (BP_IS_HOLE(bp)) {

		ASSERT(th->th_advance & ADVANCE_HOLES);

		return (0);

	}

	if (compare_bookmark(zb, &th->th_noread, dnp, 0) == 0) {

		error = EIO;

	} else if (arc_tryread(th->th_spa, bp, bc->bc_data) == 0) {

		error = 0;

		th->th_arc_hits++;

	} else {

		error = zio_wait(zio_read(NULL, th->th_spa, bp, bc->bc_data,

		    BP_GET_LSIZE(bp), NULL, NULL, ZIO_PRIORITY_SYNC_READ,

		    th->th_zio_flags | ZIO_FLAG_DONT_CACHE, zb));

		if (BP_SHOULD_BYTESWAP(bp) && error == 0)

			(zb->zb_level > 0 ? byteswap_uint64_array :

			    dmu_ot[BP_GET_TYPE(bp)].ot_byteswap)(bc->bc_data,

			    BP_GET_LSIZE(bp));

		th->th_reads++;

	}

	if (error) {

		bc->bc_errno = error;

		error = traverse_callback(th, NULL, bc);

		ASSERT(error == EAGAIN || error == EINTR || error == ERESTART);

		bc->bc_blkptr.blk_birth = -1ULL;

	}

	dprintf("cache %02x error %d <%llu, %llu, %d, %llx>\n",

	    bc - &th->th_cache[0][0], error,

	    zb->zb_objset, zb->zb_object, zb->zb_level, zb->zb_blkid);

	return (error);

}

static int

find_block(traverse_handle_t *th, zseg_t *zseg, dnode_phys_t *dnp, int depth)

{

	zbookmark_t *zb = &zseg->seg_start;

	traverse_blk_cache_t *bc;

	blkptr_t *bp = dnp->dn_blkptr;

	int i, first, level;

	int nbp = dnp->dn_nblkptr;

	int minlevel = zb->zb_level;

	int maxlevel = dnp->dn_nlevels - 1;

	int wshift = dnp->dn_indblkshift - SPA_BLKPTRSHIFT;

	int bp_shift = BP_SPAN_SHIFT(maxlevel - minlevel, wshift);

	uint64_t blkid = zb->zb_blkid >> bp_shift;

	int do_holes = (th->th_advance & ADVANCE_HOLES) && depth == ZB_DN_CACHE;

	int rc;

	if (minlevel > maxlevel || blkid >= nbp)

		return (ERANGE);

	for (level = maxlevel; level >= minlevel; level--) {

		first = P2PHASE(blkid, 1ULL << wshift);

		for (i = first; i < nbp; i++)

			if (bp[i].blk_birth > zseg->seg_mintxg ||

			    BP_IS_HOLE(&bp[i]) && do_holes)

				break;

		if (i != first) {

			i--;

			SET_BOOKMARK_LB(zb, level, blkid + (i - first));

			return (ENOTBLK);

		}

		bc = &th->th_cache[depth][level];

		SET_BOOKMARK(&bc->bc_bookmark, zb->zb_objset, zb->zb_object,

		    level, blkid);

		if (rc = traverse_read(th, bc, bp + i, dnp)) {

			if (rc != EAGAIN) {

				SET_BOOKMARK_LB(zb, level, blkid);

			}

			return (rc);

		}

		if (BP_IS_HOLE(&bp[i])) {

			SET_BOOKMARK_LB(zb, level, blkid);

			th->th_lastcb.zb_level = ZB_NO_LEVEL;

			return (0);

		}

		nbp = 1 << wshift;

		bp = bc->bc_data;

		bp_shift -= wshift;

		blkid = zb->zb_blkid >> bp_shift;

	}

	return (0);

}

static int

get_dnode(traverse_handle_t *th, uint64_t objset, dnode_phys_t *mdn,

    uint64_t *objectp, dnode_phys_t **dnpp, uint64_t txg, int type, int depth)

{

	zseg_t zseg;

	zbookmark_t *zb = &zseg.seg_start;

	uint64_t object = *objectp;

	int i, rc;

	SET_BOOKMARK(zb, objset, 0, 0, object / DNODES_PER_BLOCK);

	SET_BOOKMARK(&zseg.seg_end, objset, 0, 0, ZB_MAXBLKID);

	zseg.seg_mintxg = txg;

	zseg.seg_maxtxg = -1ULL;

	for (;;) {

		rc = find_block(th, &zseg, mdn, depth);

		if (rc == EAGAIN || rc == EINTR || rc == ERANGE)

			break;

		if (rc == 0 && zb->zb_level == 0) {

			dnode_phys_t *dnp = th->th_cache[depth][0].bc_data;

			for (i = 0; i < DNODES_PER_BLOCK; i++) {

				object = (zb->zb_blkid * DNODES_PER_BLOCK) + i;

				if (object >= *objectp &&

				    dnp[i].dn_type != DMU_OT_NONE &&

				    (type == -1 || dnp[i].dn_type == type)) {

					*objectp = object;

					*dnpp = &dnp[i];

					return (0);

				}

			}

		}

		rc = advance_block(&zseg, mdn, rc, ADVANCE_PRE);

		if (rc == ERANGE)

			break;

	}

	if (rc == ERANGE)

		*objectp = ZB_MAXOBJECT;

	return (rc);

}

/* ARGSUSED */

static void

traverse_zil_block(zilog_t *zilog, blkptr_t *bp, void *arg, uint64_t claim_txg)

{

	traverse_handle_t *th = arg;

	traverse_blk_cache_t *bc = &th->th_zil_cache;

	zbookmark_t *zb = &bc->bc_bookmark;

	zseg_t *zseg = list_head(&th->th_seglist);

	if (bp->blk_birth <= zseg->seg_mintxg)

		return;

	if (claim_txg != 0 || bp->blk_birth < spa_first_txg(th->th_spa)) {

		zb->zb_object = 0;

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

		bc->bc_blkptr = *bp;

		(void) traverse_callback(th, zseg, bc);

	}

}

/* ARGSUSED */

static void

traverse_zil_record(zilog_t *zilog, lr_t *lrc, void *arg, uint64_t claim_txg)

{

	traverse_handle_t *th = arg;

	traverse_blk_cache_t *bc = &th->th_zil_cache;

	zbookmark_t *zb = &bc->bc_bookmark;

	zseg_t *zseg = list_head(&th->th_seglist);

	if (lrc->lrc_txtype == TX_WRITE) {

		lr_write_t *lr = (lr_write_t *)lrc;

		blkptr_t *bp = &lr->lr_blkptr;

		if (bp->blk_birth <= zseg->seg_mintxg)

			return;

		if (claim_txg != 0 && bp->blk_birth >= claim_txg) {

			zb->zb_object = lr->lr_foid;

			zb->zb_blkid = lr->lr_offset / BP_GET_LSIZE(bp);

			bc->bc_blkptr = *bp;

			(void) traverse_callback(th, zseg, bc);

		}

	}

}

static void

traverse_zil(traverse_handle_t *th, traverse_blk_cache_t *bc)

{

	spa_t *spa = th->th_spa;

	dsl_pool_t *dp = spa_get_dsl(spa);