/*

 * 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 (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.

 * Copyright 2011 Nexenta Systems, Inc. All rights reserved.

 * Copyright (c) 2013 by Delphix. All rights reserved.

 * Copyright (c) 2012, Joyent, Inc. All rights reserved.

 */

#include <sys/dmu.h>

#include <sys/dmu_impl.h>

#include <sys/dmu_tx.h>

#include <sys/dbuf.h>

#include <sys/dnode.h>

#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_prop.h>

#include <sys/dsl_pool.h>

#include <sys/dsl_synctask.h>

#include <sys/zfs_ioctl.h>

#include <sys/zap.h>

#include <sys/zio_checksum.h>

#include <sys/zfs_znode.h>

#include <zfs_fletcher.h>

#include <sys/avl.h>

#include <sys/ddt.h>

#include <sys/zfs_onexit.h>

#include <sys/dmu_send.h>

#include <sys/dsl_destroy.h>

/* Set this tunable to TRUE to replace corrupt data with 0x2f5baddb10c */

int zfs_send_corrupt_data = B_FALSE;

static char *dmu_recv_tag = "dmu_recv_tag";

static const char *recv_clone_name = "%recv";

static int

dump_bytes(dmu_sendarg_t *dsp, void *buf, int len)

{

	dsl_dataset_t *ds = dsp->dsa_os->os_dsl_dataset;

	ssize_t resid; /* have to get resid to get detailed errno */

	ASSERT0(len % 8);

	fletcher_4_incremental_native(buf, len, &dsp->dsa_zc);

	dsp->dsa_err = vn_rdwr(UIO_WRITE, dsp->dsa_vp,

	    (caddr_t)buf, len,

	    0, UIO_SYSSPACE, FAPPEND, RLIM64_INFINITY, CRED(), &resid);

	mutex_enter(&ds->ds_sendstream_lock);

	*dsp->dsa_off += len;

	mutex_exit(&ds->ds_sendstream_lock);

	return (dsp->dsa_err);

}

static int

dump_free(dmu_sendarg_t *dsp, uint64_t object, uint64_t offset,

    uint64_t length)

{

	struct drr_free *drrf = &(dsp->dsa_drr->drr_u.drr_free);

	if (length != -1ULL && offset + length < offset)

		length = -1ULL;

	/*

	 * If there is a pending op, but it's not PENDING_FREE, push it out,

	 * since free block aggregation can only be done for blocks of the

	 * same type (i.e., DRR_FREE records can only be aggregated with

	 * other DRR_FREE records.  DRR_FREEOBJECTS records can only be

	 * aggregated with other DRR_FREEOBJECTS records.

	 */

	if (dsp->dsa_pending_op != PENDING_NONE &&

	    dsp->dsa_pending_op != PENDING_FREE) {

		if (dump_bytes(dsp, dsp->dsa_drr,

		    sizeof (dmu_replay_record_t)) != 0)

			return (SET_ERROR(EINTR));

		dsp->dsa_pending_op = PENDING_NONE;

	}

	if (dsp->dsa_pending_op == PENDING_FREE) {

		/*

		 * There should never be a PENDING_FREE if length is -1

		 * (because dump_dnode is the only place where this

		 * function is called with a -1, and only after flushing

		 * any pending record).

		 */

		ASSERT(length != -1ULL);

		/*

		 * Check to see whether this free block can be aggregated

		 * with pending one.

		 */

		if (drrf->drr_object == object && drrf->drr_offset +

		    drrf->drr_length == offset) {

			drrf->drr_length += length;

			return (0);

		} else {

			/* not a continuation.  Push out pending record */

			if (dump_bytes(dsp, dsp->dsa_drr,

			    sizeof (dmu_replay_record_t)) != 0)

				return (SET_ERROR(EINTR));

			dsp->dsa_pending_op = PENDING_NONE;

		}

	}

	/* create a FREE record and make it pending */

	bzero(dsp->dsa_drr, sizeof (dmu_replay_record_t));

	dsp->dsa_drr->drr_type = DRR_FREE;

	drrf->drr_object = object;

	drrf->drr_offset = offset;

	drrf->drr_length = length;

	drrf->drr_toguid = dsp->dsa_toguid;

	if (length == -1ULL) {

		if (dump_bytes(dsp, dsp->dsa_drr,

		    sizeof (dmu_replay_record_t)) != 0)

			return (SET_ERROR(EINTR));

	} else {

		dsp->dsa_pending_op = PENDING_FREE;

	}

	return (0);

}

static int

dump_data(dmu_sendarg_t *dsp, dmu_object_type_t type,

    uint64_t object, uint64_t offset, int blksz, const blkptr_t *bp, void *data)

{

	struct drr_write *drrw = &(dsp->dsa_drr->drr_u.drr_write);

	/*

	 * If there is any kind of pending aggregation (currently either

	 * a grouping of free objects or free blocks), push it out to

	 * the stream, since aggregation can't be done across operations

	 * of different types.

	 */

	if (dsp->dsa_pending_op != PENDING_NONE) {

		if (dump_bytes(dsp, dsp->dsa_drr,

		    sizeof (dmu_replay_record_t)) != 0)

			return (SET_ERROR(EINTR));

		dsp->dsa_pending_op = PENDING_NONE;

	}

	/* write a DATA record */

	bzero(dsp->dsa_drr, sizeof (dmu_replay_record_t));

	dsp->dsa_drr->drr_type = DRR_WRITE;

	drrw->drr_object = object;

	drrw->drr_type = type;

	drrw->drr_offset = offset;

	drrw->drr_length = blksz;

	drrw->drr_toguid = dsp->dsa_toguid;

	drrw->drr_checksumtype = BP_GET_CHECKSUM(bp);

	if (zio_checksum_table[drrw->drr_checksumtype].ci_dedup)

		drrw->drr_checksumflags |= DRR_CHECKSUM_DEDUP;

	DDK_SET_LSIZE(&drrw->drr_key, BP_GET_LSIZE(bp));

	DDK_SET_PSIZE(&drrw->drr_key, BP_GET_PSIZE(bp));

	DDK_SET_COMPRESS(&drrw->drr_key, BP_GET_COMPRESS(bp));

	drrw->drr_key.ddk_cksum = bp->blk_cksum;

	if (dump_bytes(dsp, dsp->dsa_drr, sizeof (dmu_replay_record_t)) != 0)

		return (SET_ERROR(EINTR));

	if (dump_bytes(dsp, data, blksz) != 0)

		return (SET_ERROR(EINTR));

	return (0);

}

static int

dump_spill(dmu_sendarg_t *dsp, uint64_t object, int blksz, void *data)

{

	struct drr_spill *drrs = &(dsp->dsa_drr->drr_u.drr_spill);

	if (dsp->dsa_pending_op != PENDING_NONE) {

		if (dump_bytes(dsp, dsp->dsa_drr,

		    sizeof (dmu_replay_record_t)) != 0)

			return (SET_ERROR(EINTR));

		dsp->dsa_pending_op = PENDING_NONE;

	}

	/* write a SPILL record */

	bzero(dsp->dsa_drr, sizeof (dmu_replay_record_t));

	dsp->dsa_drr->drr_type = DRR_SPILL;

	drrs->drr_object = object;

	drrs->drr_length = blksz;

	drrs->drr_toguid = dsp->dsa_toguid;

	if (dump_bytes(dsp, dsp->dsa_drr, sizeof (dmu_replay_record_t)))

		return (SET_ERROR(EINTR));

	if (dump_bytes(dsp, data, blksz))

		return (SET_ERROR(EINTR));

	return (0);

}

static int

dump_freeobjects(dmu_sendarg_t *dsp, uint64_t firstobj, uint64_t numobjs)

{

	struct drr_freeobjects *drrfo = &(dsp->dsa_drr->drr_u.drr_freeobjects);

	/*

	 * If there is a pending op, but it's not PENDING_FREEOBJECTS,

	 * push it out, since free block aggregation can only be done for

	 * blocks of the same type (i.e., DRR_FREE records can only be

	 * aggregated with other DRR_FREE records.  DRR_FREEOBJECTS records

	 * can only be aggregated with other DRR_FREEOBJECTS records.

	 */

	if (dsp->dsa_pending_op != PENDING_NONE &&

	    dsp->dsa_pending_op != PENDING_FREEOBJECTS) {

		if (dump_bytes(dsp, dsp->dsa_drr,

		    sizeof (dmu_replay_record_t)) != 0)

			return (SET_ERROR(EINTR));

		dsp->dsa_pending_op = PENDING_NONE;

	}

	if (dsp->dsa_pending_op == PENDING_FREEOBJECTS) {

		/*

		 * See whether this free object array can be aggregated

		 * with pending one

		 */

		if (drrfo->drr_firstobj + drrfo->drr_numobjs == firstobj) {

			drrfo->drr_numobjs += numobjs;

			return (0);

		} else {

			/* can't be aggregated.  Push out pending record */

			if (dump_bytes(dsp, dsp->dsa_drr,

			    sizeof (dmu_replay_record_t)) != 0)

				return (SET_ERROR(EINTR));

			dsp->dsa_pending_op = PENDING_NONE;

		}

	}

	/* write a FREEOBJECTS record */

	bzero(dsp->dsa_drr, sizeof (dmu_replay_record_t));

	dsp->dsa_drr->drr_type = DRR_FREEOBJECTS;

	drrfo->drr_firstobj = firstobj;

	drrfo->drr_numobjs = numobjs;

	drrfo->drr_toguid = dsp->dsa_toguid;

	dsp->dsa_pending_op = PENDING_FREEOBJECTS;

	return (0);

}

static int

dump_dnode(dmu_sendarg_t *dsp, uint64_t object, dnode_phys_t *dnp)

{

	struct drr_object *drro = &(dsp->dsa_drr->drr_u.drr_object);

	if (dnp == NULL || dnp->dn_type == DMU_OT_NONE)

		return (dump_freeobjects(dsp, object, 1));

	if (dsp->dsa_pending_op != PENDING_NONE) {

		if (dump_bytes(dsp, dsp->dsa_drr,

		    sizeof (dmu_replay_record_t)) != 0)

			return (SET_ERROR(EINTR));

		dsp->dsa_pending_op = PENDING_NONE;

	}

	/* write an OBJECT record */

	bzero(dsp->dsa_drr, sizeof (dmu_replay_record_t));

	dsp->dsa_drr->drr_type = DRR_OBJECT;

	drro->drr_object = object;

	drro->drr_type = dnp->dn_type;

	drro->drr_bonustype = dnp->dn_bonustype;

	drro->drr_blksz = dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT;

	drro->drr_bonuslen = dnp->dn_bonuslen;

	drro->drr_checksumtype = dnp->dn_checksum;

	drro->drr_compress = dnp->dn_compress;

	drro->drr_toguid = dsp->dsa_toguid;

	if (dump_bytes(dsp, dsp->dsa_drr, sizeof (dmu_replay_record_t)) != 0)

		return (SET_ERROR(EINTR));

	if (dump_bytes(dsp, DN_BONUS(dnp), P2ROUNDUP(dnp->dn_bonuslen, 8)) != 0)

		return (SET_ERROR(EINTR));

	/* free anything past the end of the file */

	if (dump_free(dsp, object, (dnp->dn_maxblkid + 1) *

	    (dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT), -1ULL))

		return (SET_ERROR(EINTR));

	if (dsp->dsa_err != 0)

		return (SET_ERROR(EINTR));

	return (0);

}

#define	BP_SPAN(dnp, level) \

	(((uint64_t)dnp->dn_datablkszsec) << (SPA_MINBLOCKSHIFT + \

	(level) * (dnp->dn_indblkshift - SPA_BLKPTRSHIFT)))

/* ARGSUSED */

static int

backup_cb(spa_t *spa, zilog_t *zilog, const blkptr_t *bp,

    const zbookmark_t *zb, const dnode_phys_t *dnp, void *arg)

{

	dmu_sendarg_t *dsp = arg;

	dmu_object_type_t type = bp ? BP_GET_TYPE(bp) : DMU_OT_NONE;

	int err = 0;

	if (issig(JUSTLOOKING) && issig(FORREAL))

		return (SET_ERROR(EINTR));

	if (zb->zb_object != DMU_META_DNODE_OBJECT &&

	    DMU_OBJECT_IS_SPECIAL(zb->zb_object)) {

		return (0);

	} else if (bp == NULL && zb->zb_object == DMU_META_DNODE_OBJECT) {

		uint64_t span = BP_SPAN(dnp, zb->zb_level);

		uint64_t dnobj = (zb->zb_blkid * span) >> DNODE_SHIFT;

		err = dump_freeobjects(dsp, dnobj, span >> DNODE_SHIFT);

	} else if (bp == NULL) {

		uint64_t span = BP_SPAN(dnp, zb->zb_level);

		err = dump_free(dsp, zb->zb_object, zb->zb_blkid * span, span);

	} else if (zb->zb_level > 0 || type == DMU_OT_OBJSET) {

		return (0);

	} else if (type == DMU_OT_DNODE) {

		dnode_phys_t *blk;

		int i;

		int blksz = BP_GET_LSIZE(bp);

		uint32_t aflags = ARC_WAIT;

		arc_buf_t *abuf;

		if (arc_read(NULL, spa, bp, arc_getbuf_func, &abuf,

		    ZIO_PRIORITY_ASYNC_READ, ZIO_FLAG_CANFAIL,

		    &aflags, zb) != 0)

			return (SET_ERROR(EIO));

		blk = abuf->b_data;

		for (i = 0; i < blksz >> DNODE_SHIFT; i++) {

			uint64_t dnobj = (zb->zb_blkid <<

			    (DNODE_BLOCK_SHIFT - DNODE_SHIFT)) + i;

			err = dump_dnode(dsp, dnobj, blk+i);

			if (err != 0)

				break;

		}

		(void) arc_buf_remove_ref(abuf, &abuf);

	} else if (type == DMU_OT_SA) {

		uint32_t aflags = ARC_WAIT;

		arc_buf_t *abuf;

		int blksz = BP_GET_LSIZE(bp);

		if (arc_read(NULL, spa, bp, arc_getbuf_func, &abuf,

		    ZIO_PRIORITY_ASYNC_READ, ZIO_FLAG_CANFAIL,

		    &aflags, zb) != 0)

			return (SET_ERROR(EIO));

		err = dump_spill(dsp, zb->zb_object, blksz, abuf->b_data);

		(void) arc_buf_remove_ref(abuf, &abuf);

	} else { /* it's a level-0 block of a regular object */

		uint32_t aflags = ARC_WAIT;

		arc_buf_t *abuf;

		int blksz = BP_GET_LSIZE(bp);

		if (arc_read(NULL, spa, bp, arc_getbuf_func, &abuf,

		    ZIO_PRIORITY_ASYNC_READ, ZIO_FLAG_CANFAIL,

		    &aflags, zb) != 0) {

			if (zfs_send_corrupt_data) {

				/* Send a block filled with 0x"zfs badd bloc" */

				abuf = arc_buf_alloc(spa, blksz, &abuf,

				    ARC_BUFC_DATA);

				uint64_t *ptr;

				for (ptr = abuf->b_data;

				    (char *)ptr < (char *)abuf->b_data + blksz;

				    ptr++)

					*ptr = 0x2f5baddb10c;

			} else {

				return (SET_ERROR(EIO));

			}

		}

		err = dump_data(dsp, type, zb->zb_object, zb->zb_blkid * blksz,

		    blksz, bp, abuf->b_data);

		(void) arc_buf_remove_ref(abuf, &abuf);

	}

	ASSERT(err == 0 || err == EINTR);

	return (err);

}

/*

 * Releases dp, ds, and fromds, using the specified tag.

 */

static int

dmu_send_impl(void *tag, dsl_pool_t *dp, dsl_dataset_t *ds,

    dsl_dataset_t *fromds, int outfd, vnode_t *vp, offset_t *off)

{

	objset_t *os;

	dmu_replay_record_t *drr;

	dmu_sendarg_t *dsp;

	int err;

	uint64_t fromtxg = 0;

	if (fromds != NULL && !dsl_dataset_is_before(ds, fromds)) {

		dsl_dataset_rele(fromds, tag);

		dsl_dataset_rele(ds, tag);

		dsl_pool_rele(dp, tag);

		return (SET_ERROR(EXDEV));

	}

	err = dmu_objset_from_ds(ds, &os);

	if (err != 0) {

		if (fromds != NULL)

			dsl_dataset_rele(fromds, tag);

		dsl_dataset_rele(ds, tag);

		dsl_pool_rele(dp, tag);

		return (err);

	}

	drr = kmem_zalloc(sizeof (dmu_replay_record_t), KM_SLEEP);

	drr->drr_type = DRR_BEGIN;

	drr->drr_u.drr_begin.drr_magic = DMU_BACKUP_MAGIC;

	DMU_SET_STREAM_HDRTYPE(drr->drr_u.drr_begin.drr_versioninfo,

	    DMU_SUBSTREAM);

#ifdef _KERNEL

	if (dmu_objset_type(os) == DMU_OST_ZFS) {

		uint64_t version;

		if (zfs_get_zplprop(os, ZFS_PROP_VERSION, &version) != 0) {

			kmem_free(drr, sizeof (dmu_replay_record_t));

			if (fromds != NULL)

				dsl_dataset_rele(fromds, tag);

			dsl_dataset_rele(ds, tag);

			dsl_pool_rele(dp, tag);

			return (SET_ERROR(EINVAL));

		}

		if (version >= ZPL_VERSION_SA) {

			DMU_SET_FEATUREFLAGS(

			    drr->drr_u.drr_begin.drr_versioninfo,

			    DMU_BACKUP_FEATURE_SA_SPILL);

		}

	}

#endif

	drr->drr_u.drr_begin.drr_creation_time =

	    ds->ds_phys->ds_creation_time;

	drr->drr_u.drr_begin.drr_type = dmu_objset_type(os);

	if (fromds != NULL && ds->ds_dir != fromds->ds_dir)

		drr->drr_u.drr_begin.drr_flags |= DRR_FLAG_CLONE;

	drr->drr_u.drr_begin.drr_toguid = ds->ds_phys->ds_guid;

	if (ds->ds_phys->ds_flags & DS_FLAG_CI_DATASET)

		drr->drr_u.drr_begin.drr_flags |= DRR_FLAG_CI_DATA;

	if (fromds != NULL)

		drr->drr_u.drr_begin.drr_fromguid = fromds->ds_phys->ds_guid;

	dsl_dataset_name(ds, drr->drr_u.drr_begin.drr_toname);

	if (fromds != NULL) {

		fromtxg = fromds->ds_phys->ds_creation_txg;

		dsl_dataset_rele(fromds, tag);

		fromds = NULL;

	}

	dsp = kmem_zalloc(sizeof (dmu_sendarg_t), KM_SLEEP);

	dsp->dsa_drr = drr;

	dsp->dsa_vp = vp;

	dsp->dsa_outfd = outfd;

	dsp->dsa_proc = curproc;

	dsp->dsa_os = os;

	dsp->dsa_off = off;

	dsp->dsa_toguid = ds->ds_phys->ds_guid;

	ZIO_SET_CHECKSUM(&dsp->dsa_zc, 0, 0, 0, 0);

	dsp->dsa_pending_op = PENDING_NONE;

	mutex_enter(&ds->ds_sendstream_lock);

	list_insert_head(&ds->ds_sendstreams, dsp);

	mutex_exit(&ds->ds_sendstream_lock);

	dsl_dataset_long_hold(ds, FTAG);

	dsl_pool_rele(dp, tag);

	if (dump_bytes(dsp, drr, sizeof (dmu_replay_record_t)) != 0) {

		err = dsp->dsa_err;

		goto out;

	}