// SPDX-License-Identifier: CDDL-1.0
/*
 * 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 https://opensource.org/licenses/CDDL-1.0.
 * 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) 2011, 2018 by Delphix. All rights reserved.
 * Copyright (c) 2014, Joyent, Inc. All rights reserved.
 * Copyright 2014 HybridCluster. All rights reserved.
 * Copyright 2016 RackTop Systems.
 * Copyright (c) 2016 Actifio, Inc. All rights reserved.
 * Copyright (c) 2019, 2024, Klara, Inc.
 * Copyright (c) 2019, Allan Jude
 */

#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/spa_impl.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/dmu_recv.h>
#include <sys/dsl_destroy.h>
#include <sys/blkptr.h>
#include <sys/dsl_bookmark.h>
#include <sys/zfeature.h>
#include <sys/bqueue.h>
#include <sys/zvol.h>
#include <sys/policy.h>
#include <sys/objlist.h>
#ifdef _KERNEL
#include <sys/zfs_vfsops.h>
#endif

/* Set this tunable to TRUE to replace corrupt data with 0x2f5baddb10c */
static int zfs_send_corrupt_data = B_FALSE;
/*
 * This tunable controls the amount of data (measured in bytes) that will be
 * prefetched by zfs send.  If the main thread is blocking on reads that haven't
 * completed, this variable might need to be increased.  If instead the main
 * thread is issuing new reads because the prefetches have fallen out of the
 * cache, this may need to be decreased.
 */
static uint_t zfs_send_queue_length = SPA_MAXBLOCKSIZE;
/*
 * This tunable controls the length of the queues that zfs send worker threads
 * use to communicate.  If the send_main_thread is blocking on these queues,
 * this variable may need to be increased.  If there is a significant slowdown
 * at the start of a send as these threads consume all the available IO
 * resources, this variable may need to be decreased.
 */
static uint_t zfs_send_no_prefetch_queue_length = 1024 * 1024;
/*
 * These tunables control the fill fraction of the queues by zfs send.  The fill
 * fraction controls the frequency with which threads have to be cv_signaled.
 * If a lot of cpu time is being spent on cv_signal, then these should be tuned
 * down.  If the queues empty before the signalled thread can catch up, then
 * these should be tuned up.
 */
static uint_t zfs_send_queue_ff = 20;
static uint_t zfs_send_no_prefetch_queue_ff = 20;

/*
 * Use this to override the recordsize calculation for fast zfs send estimates.
 */
static uint_t zfs_override_estimate_recordsize = 0;

/* Set this tunable to FALSE to disable setting of DRR_FLAG_FREERECORDS */
static const boolean_t zfs_send_set_freerecords_bit = B_TRUE;

/* Set this tunable to FALSE is disable sending unmodified spill blocks. */
static int zfs_send_unmodified_spill_blocks = B_TRUE;

static inline boolean_t
overflow_multiply(uint64_t a, uint64_t b, uint64_t *c)
{
	uint64_t temp = a * b;
	if (b != 0 && temp / b != a)
		return (B_FALSE);
	*c = temp;
	return (B_TRUE);
}

struct send_thread_arg {
	bqueue_t	q;
	objset_t	*os;		/* Objset to traverse */
	uint64_t	fromtxg;	/* Traverse from this txg */
	int		flags;		/* flags to pass to traverse_dataset */
	int		error_code;
	boolean_t	cancel;
	zbookmark_phys_t resume;
	uint64_t	*num_blocks_visited;
};

struct redact_list_thread_arg {
	boolean_t		cancel;
	bqueue_t		q;
	zbookmark_phys_t	resume;
	redaction_list_t	*rl;
	boolean_t		mark_redact;
	int			error_code;
	uint64_t		*num_blocks_visited;
};

struct send_merge_thread_arg {
	bqueue_t			q;
	objset_t			*os;
	struct redact_list_thread_arg	*from_arg;
	struct send_thread_arg		*to_arg;
	struct redact_list_thread_arg	*redact_arg;
	int				error;
	boolean_t			cancel;
};

struct send_range {
	boolean_t		eos_marker; /* Marks the end of the stream */
	uint64_t		object;
	uint64_t		start_blkid;
	uint64_t		end_blkid;
	bqueue_node_t		ln;
	enum type {DATA, HOLE, OBJECT, OBJECT_RANGE, REDACT,
	    PREVIOUSLY_REDACTED} type;
	union {
		struct srd {
			dmu_object_type_t	obj_type;
			uint32_t		datablksz; // logical size
			uint32_t		datasz; // payload size
			blkptr_t		bp;
			arc_buf_t		*abuf;
			abd_t			*abd;
			kmutex_t		lock;
			kcondvar_t		cv;
			boolean_t		io_outstanding;
			boolean_t		io_compressed;
			int			io_err;
		} data;
		struct srh {
			uint32_t		datablksz;
		} hole;
		struct sro {
			/*
			 * This is a pointer because embedding it in the
			 * struct causes these structures to be massively larger
			 * for all range types; this makes the code much less
			 * memory efficient.
			 */
			dnode_phys_t		*dnp;
			blkptr_t		bp;
			/* Piggyback unmodified spill block */
			struct send_range	*spill_range;
		} object;
		struct srr {
			uint32_t		datablksz;
		} redact;
		struct sror {
			blkptr_t		bp;
		} object_range;
	} sru;
};

/*
 * The list of data whose inclusion in a send stream can be pending from
 * one call to backup_cb to another.  Multiple calls to dump_free(),
 * dump_freeobjects(), and dump_redact() can be aggregated into a single
 * DRR_FREE, DRR_FREEOBJECTS, or DRR_REDACT replay record.
 */
typedef enum {
	PENDING_NONE,
	PENDING_FREE,
	PENDING_FREEOBJECTS,
	PENDING_REDACT
} dmu_pendop_t;

typedef struct dmu_send_cookie {
	dmu_replay_record_t *dsc_drr;
	dmu_send_outparams_t *dsc_dso;
	offset_t *dsc_off;
	objset_t *dsc_os;
	zio_cksum_t dsc_zc;
	uint64_t dsc_toguid;
	uint64_t dsc_fromtxg;
	int dsc_err;
	dmu_pendop_t dsc_pending_op;
	uint64_t dsc_featureflags;
	uint64_t dsc_last_data_object;
	uint64_t dsc_last_data_offset;
	uint64_t dsc_resume_object;
	uint64_t dsc_resume_offset;
	boolean_t dsc_sent_begin;
	boolean_t dsc_sent_end;
} dmu_send_cookie_t;

static int do_dump(dmu_send_cookie_t *dscp, struct send_range *range);

static void
range_free(struct send_range *range)
{
	if (range->type == OBJECT) {
		size_t size = sizeof (dnode_phys_t) *
		    (range->sru.object.dnp->dn_extra_slots + 1);
		kmem_free(range->sru.object.dnp, size);
		if (range->sru.object.spill_range)
			range_free(range->sru.object.spill_range);
	} else if (range->type == DATA) {
		mutex_enter(&range->sru.data.lock);
		while (range->sru.data.io_outstanding)
			cv_wait(&range->sru.data.cv, &range->sru.data.lock);
		if (range->sru.data.abd != NULL)
			abd_free(range->sru.data.abd);
		if (range->sru.data.abuf != NULL) {
			arc_buf_destroy(range->sru.data.abuf,
			    &range->sru.data.abuf);
		}
		mutex_exit(&range->sru.data.lock);

		cv_destroy(&range->sru.data.cv);
		mutex_destroy(&range->sru.data.lock);
	}
	kmem_free(range, sizeof (*range));
}

/*
 * For all record types except BEGIN, fill in the checksum (overlaid in
 * drr_u.drr_checksum.drr_checksum).  The checksum verifies everything
 * up to the start of the checksum itself.
 */
static int
dump_record(dmu_send_cookie_t *dscp, void *payload, int payload_len)
{
	dmu_send_outparams_t *dso = dscp->dsc_dso;
	ASSERT3U(offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
	    ==, sizeof (dmu_replay_record_t) - sizeof (zio_cksum_t));
	(void) fletcher_4_incremental_native(dscp->dsc_drr,
	    offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
	    &dscp->dsc_zc);
	if (dscp->dsc_drr->drr_type == DRR_BEGIN) {
		dscp->dsc_sent_begin = B_TRUE;
	} else {
		ASSERT(ZIO_CHECKSUM_IS_ZERO(&dscp->dsc_drr->drr_u.
		    drr_checksum.drr_checksum));
		dscp->dsc_drr->drr_u.drr_checksum.drr_checksum = dscp->dsc_zc;
	}
	if (dscp->dsc_drr->drr_type == DRR_END) {
		dscp->dsc_sent_end = B_TRUE;
	}
	(void) fletcher_4_incremental_native(&dscp->dsc_drr->
	    drr_u.drr_checksum.drr_checksum,
	    sizeof (zio_cksum_t), &dscp->dsc_zc);
	*dscp->dsc_off += sizeof (dmu_replay_record_t);
	dscp->dsc_err = dso->dso_outfunc(dscp->dsc_os, dscp->dsc_drr,
	    sizeof (dmu_replay_record_t), dso->dso_arg);
	if (dscp->dsc_err != 0)
		return (SET_ERROR(EINTR));
	if (payload_len != 0) {
		*dscp->dsc_off += payload_len;
		/*
		 * payload is null when dso_dryrun == B_TRUE (i.e. when we're
		 * doing a send size calculation)
		 */
		if (payload != NULL) {
			(void) fletcher_4_incremental_native(
			    payload, payload_len, &dscp->dsc_zc);
		}

		/*
		 * The code does not rely on this (len being a multiple of 8).
		 * We keep this assertion because of the corresponding assertion
		 * in receive_read().  Keeping this assertion ensures that we do
		 * not inadvertently break backwards compatibility (causing the
		 * assertion in receive_read() to trigger on old software).
		 *
		 * Raw sends cannot be received on old software, and so can
		 * bypass this assertion.
		 */

		ASSERT((payload_len % 8 == 0) ||
		    (dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW));

		dscp->dsc_err = dso->dso_outfunc(dscp->dsc_os, payload,
		    payload_len, dso->dso_arg);
		if (dscp->dsc_err != 0)
			return (SET_ERROR(EINTR));
	}
	return (0);
}

/*
 * Fill in the drr_free struct, or perform aggregation if the previous record is
 * also a free record, and the two are adjacent.
 *
 * Note that we send free records even for a full send, because we want to be
 * able to receive a full send as a clone, which requires a list of all the free
 * and freeobject records that were generated on the source.
 */
static int
dump_free(dmu_send_cookie_t *dscp, uint64_t object, uint64_t offset,
    uint64_t length)
{
	struct drr_free *drrf = &(dscp->dsc_drr->drr_u.drr_free);

	/*
	 * When we receive a free record, dbuf_free_range() assumes
	 * that the receiving system doesn't have any dbufs in the range
	 * being freed.  This is always true because there is a one-record
	 * constraint: we only send one WRITE record for any given
	 * object,offset.  We know that the one-record constraint is
	 * true because we always send data in increasing order by
	 * object,offset.
	 *
	 * If the increasing-order constraint ever changes, we should find
	 * another way to assert that the one-record constraint is still
	 * satisfied.
	 */
	ASSERT(object > dscp->dsc_last_data_object ||
	    (object == dscp->dsc_last_data_object &&
	    offset > dscp->dsc_last_data_offset));

	/*
	 * 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 (dscp->dsc_pending_op != PENDING_NONE &&
	    dscp->dsc_pending_op != PENDING_FREE) {
		if (dump_record(dscp, NULL, 0) != 0)
			return (SET_ERROR(EINTR));
		dscp->dsc_pending_op = PENDING_NONE;
	}

	if (dscp->dsc_pending_op == PENDING_FREE) {
		/*
		 * 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) {
			if (offset + length < offset || length == UINT64_MAX)
				drrf->drr_length = UINT64_MAX;
			else
				drrf->drr_length += length;
			return (0);
		} else {
			/* not a continuation.  Push out pending record */
			if (dump_record(dscp, NULL, 0) != 0)
				return (SET_ERROR(EINTR));
			dscp->dsc_pending_op = PENDING_NONE;
		}
	}
	/* create a FREE record and make it pending */
	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
	dscp->dsc_drr->drr_type = DRR_FREE;
	drrf->drr_object = object;
	drrf->drr_offset = offset;
	if (offset + length < offset)
		drrf->drr_length = DMU_OBJECT_END;
	else
		drrf->drr_length = length;
	drrf->drr_toguid = dscp->dsc_toguid;
	if (length == DMU_OBJECT_END) {
		if (dump_record(dscp, NULL, 0) != 0)
			return (SET_ERROR(EINTR));
	} else {
		dscp->dsc_pending_op = PENDING_FREE;
	}

	return (0);
}

/*
 * Fill in the drr_redact struct, or perform aggregation if the previous record
 * is also a redaction record, and the two are adjacent.
 */
static int
dump_redact(dmu_send_cookie_t *dscp, uint64_t object, uint64_t offset,
    uint64_t length)
{
	struct drr_redact *drrr = &dscp->dsc_drr->drr_u.drr_redact;

	/*
	 * If there is a pending op, but it's not PENDING_REDACT, push it out,
	 * since free block aggregation can only be done for blocks of the
	 * same type (i.e., DRR_REDACT records can only be aggregated with
	 * other DRR_REDACT records).
	 */
	if (dscp->dsc_pending_op != PENDING_NONE &&
	    dscp->dsc_pending_op != PENDING_REDACT) {
		if (dump_record(dscp, NULL, 0) != 0)
			return (SET_ERROR(EINTR));
		dscp->dsc_pending_op = PENDING_NONE;
	}

	if (dscp->dsc_pending_op == PENDING_REDACT) {
		/*
		 * Check to see whether this redacted block can be aggregated
		 * with pending one.
		 */
		if (drrr->drr_object == object && drrr->drr_offset +
		    drrr->drr_length == offset) {
			drrr->drr_length += length;
			return (0);
		} else {
			/* not a continuation.  Push out pending record */
			if (dump_record(dscp, NULL, 0) != 0)
				return (SET_ERROR(EINTR));
			dscp->dsc_pending_op = PENDING_NONE;
		}
	}
	/* create a REDACT record and make it pending */
	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
	dscp->dsc_drr->drr_type = DRR_REDACT;
	drrr->drr_object = object;
	drrr->drr_offset = offset;
	drrr->drr_length = length;
	drrr->drr_toguid = dscp->dsc_toguid;
	dscp->dsc_pending_op = PENDING_REDACT;

	return (0);
}

static int
dmu_dump_write(dmu_send_cookie_t *dscp, dmu_object_type_t type, uint64_t object,
    uint64_t offset, int lsize, int psize, const blkptr_t *bp,
    boolean_t io_compressed, void *data)
{
	uint64_t payload_size;
	boolean_t raw = (dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW);
	struct drr_write *drrw = &(dscp->dsc_drr->drr_u.drr_write);

	/*
	 * We send data in increasing object, offset order.
	 * See comment in dump_free() for details.
	 */
	ASSERT(object > dscp->dsc_last_data_object ||
	    (object == dscp->dsc_last_data_object &&
	    offset > dscp->dsc_last_data_offset));
	dscp->dsc_last_data_object = object;
	dscp->dsc_last_data_offset = offset + lsize - 1;

	/*
	 * 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 (dscp->dsc_pending_op != PENDING_NONE) {
		if (dump_record(dscp, NULL, 0) != 0)
			return (SET_ERROR(EINTR));
		dscp->dsc_pending_op = PENDING_NONE;
	}
	/* write a WRITE record */
	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
	dscp->dsc_drr->drr_type = DRR_WRITE;
	drrw->drr_object = object;
	drrw->drr_type = type;
	drrw->drr_offset = offset;
	drrw->drr_toguid = dscp->dsc_toguid;
	drrw->drr_logical_size = lsize;

	/* only set the compression fields if the buf is compressed or raw */
	boolean_t compressed =
	    (bp != NULL ? BP_GET_COMPRESS(bp) != ZIO_COMPRESS_OFF &&
	    io_compressed : lsize != psize);
	if (raw || compressed) {
		ASSERT(bp != NULL);
		ASSERT(raw || dscp->dsc_featureflags &
		    DMU_BACKUP_FEATURE_COMPRESSED);
		ASSERT(!BP_IS_EMBEDDED(bp));
		ASSERT3S(psize, >, 0);

		if (raw) {
			ASSERT(BP_IS_PROTECTED(bp));

			/*
			 * This is a raw protected block so we need to pass
			 * along everything the receiving side will need to
			 * interpret this block, including the byteswap, salt,
			 * IV, and MAC.
			 */
			if (BP_SHOULD_BYTESWAP(bp))
				drrw->drr_flags |= DRR_RAW_BYTESWAP;
			zio_crypt_decode_params_bp(bp, drrw->drr_salt,
			    drrw->drr_iv);
			zio_crypt_decode_mac_bp(bp, drrw->drr_mac);
		} else {
			/* this is a compressed block */
			ASSERT(dscp->dsc_featureflags &
			    DMU_BACKUP_FEATURE_COMPRESSED);
			ASSERT(!BP_SHOULD_BYTESWAP(bp));
			ASSERT(!DMU_OT_IS_METADATA(BP_GET_TYPE(bp)));
			ASSERT3U(BP_GET_COMPRESS(bp), !=, ZIO_COMPRESS_OFF);
			ASSERT3S(lsize, >=, psize);
		}

		/* set fields common to compressed and raw sends */
		drrw->drr_compressiontype = BP_GET_COMPRESS(bp);
		drrw->drr_compressed_size = psize;
		payload_size = drrw->drr_compressed_size;
	} else {
		payload_size = drrw->drr_logical_size;
	}

	if (bp == NULL || BP_IS_EMBEDDED(bp) || (BP_IS_PROTECTED(bp) && !raw)) {
		/*
		 * There's no pre-computed checksum for partial-block writes,
		 * embedded BP's, or encrypted BP's that are being sent as
		 * plaintext, so (like fletcher4-checksummed blocks) userland
		 * will have to compute a dedup-capable checksum itself.
		 */
		drrw->drr_checksumtype = ZIO_CHECKSUM_OFF;
	} else {
		drrw->drr_checksumtype = BP_GET_CHECKSUM(bp);
		if (zio_checksum_table[drrw->drr_checksumtype].ci_flags &
		    ZCHECKSUM_FLAG_DEDUP)
			drrw->drr_flags |= 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));
		DDK_SET_CRYPT(&drrw->drr_key, BP_IS_PROTECTED(bp));
		drrw->drr_key.ddk_cksum = bp->blk_cksum;
	}

	if (dump_record(dscp, data, payload_size) != 0)
		return (SET_ERROR(EINTR));
	return (0);
}

static int
dump_write_embedded(dmu_send_cookie_t *dscp, uint64_t object, uint64_t offset,
    int blksz, const blkptr_t *bp)
{
	char buf[BPE_PAYLOAD_SIZE];
	struct drr_write_embedded *drrw =
	    &(dscp->dsc_drr->drr_u.drr_write_embedded);

	if (dscp->dsc_pending_op != PENDING_NONE) {
		if (dump_record(dscp, NULL, 0) != 0)
			return (SET_ERROR(EINTR));
		dscp->dsc_pending_op = PENDING_NONE;
	}

	ASSERT(BP_IS_EMBEDDED(bp));

	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
	dscp->dsc_drr->drr_type = DRR_WRITE_EMBEDDED;
	drrw->drr_object = object;
	drrw->drr_offset = offset;
	drrw->drr_length = blksz;
	drrw->drr_toguid = dscp->dsc_toguid;
	drrw->drr_compression = BP_GET_COMPRESS(bp);
	drrw->drr_etype = BPE_GET_ETYPE(bp);
	drrw->drr_lsize = BPE_GET_LSIZE(bp);
	drrw->drr_psize = BPE_GET_PSIZE(bp);

	decode_embedded_bp_compressed(bp, buf);

	uint32_t psize = drrw->drr_psize;
	uint32_t rsize = P2ROUNDUP(psize, 8);

	if (psize != rsize)
		memset(buf + psize, 0, rsize - psize);

	if (dump_record(dscp, buf, rsize) != 0)
		return (SET_ERROR(EINTR));
	return (0);
}

static int
dump_spill(dmu_send_cookie_t *dscp, const blkptr_t *bp, uint64_t object,
    void *data)
{
	struct drr_spill *drrs = &(dscp->dsc_drr->drr_u.drr_spill);
	uint64_t blksz = BP_GET_LSIZE(bp);
	uint64_t payload_size = blksz;

	if (dscp->dsc_pending_op != PENDING_NONE) {
		if (dump_record(dscp, NULL, 0) != 0)
			return (SET_ERROR(EINTR));
		dscp->dsc_pending_op = PENDING_NONE;
	}

	/* write a SPILL record */
	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
	dscp->dsc_drr->drr_type = DRR_SPILL;
	drrs->drr_object = object;
	drrs->drr_length = blksz;
	drrs->drr_toguid = dscp->dsc_toguid;

	/* See comment in piggyback_unmodified_spill() for full details */
	if (zfs_send_unmodified_spill_blocks &&
	    (BP_GET_LOGICAL_BIRTH(bp) <= dscp->dsc_fromtxg)) {
		drrs->drr_flags |= DRR_SPILL_UNMODIFIED;
	}

	/* handle raw send fields */
	if (dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW) {
		ASSERT(BP_IS_PROTECTED(bp));

		if (BP_SHOULD_BYTESWAP(bp))
			drrs->drr_flags |= DRR_RAW_BYTESWAP;
		drrs->drr_compressiontype = BP_GET_COMPRESS(bp);
		drrs->drr_compressed_size = BP_GET_PSIZE(bp);
		zio_crypt_decode_params_bp(bp, drrs->drr_salt, drrs->drr_iv);
		zio_crypt_decode_mac_bp(bp, drrs->drr_mac);
		payload_size = drrs->drr_compressed_size;
	}

	if (dump_record(dscp, data, payload_size) != 0)
		return (SET_ERROR(EINTR));
	return (0);
}

static int
dump_freeobjects(dmu_send_cookie_t *dscp, uint64_t firstobj, uint64_t numobjs)
{
	struct drr_freeobjects *drrfo = &(dscp->dsc_drr->drr_u.drr_freeobjects);
	uint64_t maxobj = DNODES_PER_BLOCK *
	    (DMU_META_DNODE(dscp->dsc_os)->dn_maxblkid + 1);

	/*
	 * ZoL < 0.7 does not handle large FREEOBJECTS records correctly,
	 * leading to zfs recv never completing. to avoid this issue, don't
	 * send FREEOBJECTS records for object IDs which cannot exist on the
	 * receiving side.
	 */
	if (maxobj > 0) {
		if (maxobj <= firstobj)
			return (0);

		if (maxobj < firstobj + numobjs)
			numobjs = maxobj - firstobj;
	}

	/*
	 * 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 (dscp->dsc_pending_op != PENDING_NONE &&
	    dscp->dsc_pending_op != PENDING_FREEOBJECTS) {
		if (dump_record(dscp, NULL, 0) != 0)
			return (SET_ERROR(EINTR));
		dscp->dsc_pending_op = PENDING_NONE;
	}

	if (dscp->dsc_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_record(dscp, NULL, 0) != 0)
				return (SET_ERROR(EINTR));
			dscp->dsc_pending_op = PENDING_NONE;
		}
	}

	/* write a FREEOBJECTS record */
	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
	dscp->dsc_drr->drr_type = DRR_FREEOBJECTS;
	drrfo->drr_firstobj = firstobj;
	drrfo->drr_numobjs = numobjs;
	drrfo->drr_toguid = dscp->dsc_toguid;

	dscp->dsc_pending_op = PENDING_FREEOBJECTS;

	return (0);
}

static int
dump_dnode(dmu_send_cookie_t *dscp, const blkptr_t *bp, uint64_t object,
    dnode_phys_t *dnp)
{
	struct drr_object *drro = &(dscp->dsc_drr->drr_u.drr_object);
	int bonuslen;

	if (object < dscp->dsc_resume_object) {
		/*
		 * Note: when resuming, we will visit all the dnodes in
		 * the block of dnodes that we are resuming from.  In
		 * this case it's unnecessary to send the dnodes prior to
		 * the one we are resuming from.  We should be at most one
		 * block's worth of dnodes behind the resume point.
		 */
		ASSERT3U(dscp->dsc_resume_object - object, <,
		    1 << (DNODE_BLOCK_SHIFT - DNODE_SHIFT));
		return (0);
	}

	if (dnp == NULL || dnp->dn_type == DMU_OT_NONE)
		return (dump_freeobjects(dscp, object, 1));

	if (dscp->dsc_pending_op != PENDING_NONE) {
		if (dump_record(dscp, NULL, 0) != 0)
			return (SET_ERROR(EINTR));
		dscp->dsc_pending_op = PENDING_NONE;
	}

	/* write an OBJECT record */
	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
	dscp->dsc_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_dn_slots = dnp->dn_extra_slots + 1;
	drro->drr_checksumtype = dnp->dn_checksum;
	drro->drr_compress = dnp->dn_compress;
	drro->drr_toguid = dscp->dsc_toguid;

	if (!(dscp->dsc_featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS) &&
	    drro->drr_blksz > SPA_OLD_MAXBLOCKSIZE)
		drro->drr_blksz = SPA_OLD_MAXBLOCKSIZE;

	bonuslen = P2ROUNDUP(dnp->dn_bonuslen, 8);

	if ((dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW)) {
		ASSERT(BP_IS_ENCRYPTED(bp));

		if (BP_SHOULD_BYTESWAP(bp))
			drro->drr_flags |= DRR_RAW_BYTESWAP;

		/* needed for reconstructing dnp on recv side */
		drro->drr_maxblkid = dnp->dn_maxblkid;
		drro->drr_indblkshift = dnp->dn_indblkshift;
		drro->drr_nlevels = dnp->dn_nlevels;
		drro->drr_nblkptr = dnp->dn_nblkptr;

		/*
		 * Since we encrypt the entire bonus area, the (raw) part
		 * beyond the bonuslen is actually nonzero, so we need
		 * to send it.
		 */
		if (bonuslen != 0) {
			if (drro->drr_bonuslen > DN_MAX_BONUS_LEN(dnp))
				return (SET_ERROR(EINVAL));
			drro->drr_raw_bonuslen = DN_MAX_BONUS_LEN(dnp);
			bonuslen = drro->drr_raw_bonuslen;
		}
	}

	/*
	 * DRR_OBJECT_SPILL is set for every dnode which references a
	 * spill block.	 This allows the receiving pool to definitively
	 * determine when a spill block should be kept or freed.
	 */
	if (dnp->dn_flags & DNODE_FLAG_SPILL_BLKPTR)
		drro->drr_flags |= DRR_OBJECT_SPILL;

	if (dump_record(dscp, DN_BONUS(dnp), bonuslen) != 0)
		return (SET_ERROR(EINTR));

	/* Free anything past the end of the file. */
	if (dump_free(dscp, object, (dnp->dn_maxblkid + 1) *
	    (dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT), DMU_OBJECT_END) != 0)
		return (SET_ERROR(EINTR));

	if (dscp->dsc_err != 0)
		return (SET_ERROR(EINTR));

	return (0);
}

static int
dump_object_range(dmu_send_cookie_t *dscp, const blkptr_t *bp,
    uint64_t firstobj, uint64_t numslots)
{
	struct drr_object_range *drror =
	    &(dscp->dsc_drr->drr_u.drr_object_range);

	/* we only use this record type for raw sends */
	ASSERT(BP_IS_PROTECTED(bp));
	ASSERT(dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW);
	ASSERT3U(BP_GET_COMPRESS(bp), ==, ZIO_COMPRESS_OFF);
	ASSERT3U(BP_GET_TYPE(bp), ==, DMU_OT_DNODE);
	ASSERT0(BP_GET_LEVEL(bp));

	if (dscp->dsc_pending_op != PENDING_NONE) {
		if (dump_record(dscp, NULL, 0) != 0)
			return (SET_ERROR(EINTR));
		dscp->dsc_pending_op = PENDING_NONE;
	}

	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
	dscp->dsc_drr->drr_type = DRR_OBJECT_RANGE;
	drror->drr_firstobj = firstobj;
	drror->drr_numslots = numslots;
	drror->drr_toguid = dscp->dsc_toguid;
	if (BP_SHOULD_BYTESWAP(bp))
		drror->drr_flags |= DRR_RAW_BYTESWAP;
	zio_crypt_decode_params_bp(bp, drror->drr_salt, drror->drr_iv);
	zio_crypt_decode_mac_bp(bp, drror->drr_mac);

	if (dump_record(dscp, NULL, 0) != 0)
		return (SET_ERROR(EINTR));
	return (0);
}

static boolean_t
send_do_embed(const blkptr_t *bp, uint64_t featureflags)
{
	if (!BP_IS_EMBEDDED(bp))
		return (B_FALSE);

	/*
	 * Compression function must be legacy, or explicitly enabled.
	 */
	if ((BP_GET_COMPRESS(bp) >= ZIO_COMPRESS_LEGACY_FUNCTIONS &&
	    !(featureflags & DMU_BACKUP_FEATURE_LZ4)))
		return (B_FALSE);

	/*
	 * If we have not set the ZSTD feature flag, we can't send ZSTD
	 * compressed embedded blocks, as the receiver may not support them.
	 */
	if ((BP_GET_COMPRESS(bp) == ZIO_COMPRESS_ZSTD &&
	    !(featureflags & DMU_BACKUP_FEATURE_ZSTD)))
		return (B_FALSE);

	/*
	 * Embed type must be explicitly enabled.
	 */
	switch (BPE_GET_ETYPE(bp)) {
	case BP_EMBEDDED_TYPE_DATA:
		if (featureflags & DMU_BACKUP_FEATURE_EMBED_DATA)
			return (B_TRUE);
		break;
	default:
		return (B_FALSE);
	}
	return (B_FALSE);
}

/*
 * This function actually handles figuring out what kind of record needs to be
 * dumped, and calling the appropriate helper function.  In most cases,
 * the data has already been read by send_reader_thread().
 */
static int
do_dump(dmu_send_cookie_t *dscp, struct send_range *range)
{
	int err = 0;
	switch (range->type) {
	case OBJECT:
		err = dump_dnode(dscp, &range->sru.object.bp, range->object,
		    range->sru.object.dnp);
		/* Dump piggybacked unmodified spill block */
		if (!err && range->sru.object.spill_range)
			err = do_dump(dscp, range->sru.object.spill_range);
		return (err);
	case OBJECT_RANGE: {
		ASSERT3U(range->start_blkid + 1, ==, range->end_blkid);
		if (!(dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW)) {
			return (0);
		}
		uint64_t epb = BP_GET_LSIZE(&range->sru.object_range.bp) >>
		    DNODE_SHIFT;
		uint64_t firstobj = range->start_blkid * epb;
		err = dump_object_range(dscp, &range->sru.object_range.bp,
		    firstobj, epb);
		break;
	}
	case REDACT: {
		struct srr *srrp = &range->sru.redact;
		err = dump_redact(dscp, range->object, range->start_blkid *
		    srrp->datablksz, (range->end_blkid - range->start_blkid) *
		    srrp->datablksz);
		return (err);
	}
	case DATA: {
		struct srd *srdp = &range->sru.data;
		blkptr_t *bp = &srdp->bp;
		spa_t *spa =
		    dmu_objset_spa(dscp->dsc_os);

		ASSERT3U(srdp->datablksz, ==, BP_GET_LSIZE(bp));
		ASSERT3U(range->start_blkid + 1, ==, range->end_blkid);

		if (send_do_embed(bp, dscp->dsc_featureflags)) {
			err = dump_write_embedded(dscp, range->object,
			    range->start_blkid * srdp->datablksz,
			    srdp->datablksz, bp);
			return (err);
		}
		ASSERT(range->object > dscp->dsc_resume_object ||
		    (range->object == dscp->dsc_resume_object &&
		    (range->start_blkid == DMU_SPILL_BLKID ||
		    range->start_blkid * srdp->datablksz >=
		    dscp->dsc_resume_offset)));
		/* it's a level-0 block of a regular object */

		mutex_enter(&srdp->lock);
		while (srdp->io_outstanding)
			cv_wait(&srdp->cv, &srdp->lock);
		err = srdp->io_err;
		mutex_exit(&srdp->lock);

		if (err != 0) {
			if (zfs_send_corrupt_data &&
			    !dscp->dsc_dso->dso_dryrun) {
				/*
				 * Send a block filled with 0x"zfs badd bloc"
				 */
				srdp->abuf = arc_alloc_buf(spa, &srdp->abuf,
				    ARC_BUFC_DATA, srdp->datablksz);
				uint64_t *ptr;
				for (ptr = srdp->abuf->b_data;
				    (char *)ptr < (char *)srdp->abuf->b_data +
				    srdp->datablksz; ptr++)
					*ptr = 0x2f5baddb10cULL;
			} else {
				return (SET_ERROR(EIO));
			}
		}

		ASSERT(dscp->dsc_dso->dso_dryrun ||
		    srdp->abuf != NULL || srdp->abd != NULL);

		char *data = NULL;
		if (srdp->abd != NULL) {
			data = abd_to_buf(srdp->abd);
			ASSERT3P(srdp->abuf, ==, NULL);
		} else if (srdp->abuf != NULL) {
			data = srdp->abuf->b_data;
		}

		if (BP_GET_TYPE(bp) == DMU_OT_SA) {
			ASSERT3U(range->start_blkid, ==, DMU_SPILL_BLKID);
			err = dump_spill(dscp, bp, range->object, data);
			return (err);
		}

		uint64_t offset = range->start_blkid * srdp->datablksz;

		/*
		 * If we have large blocks stored on disk but the send flags
		 * don't allow us to send large blocks, we split the data from
		 * the arc buf into chunks.
		 */
		if (srdp->datablksz > SPA_OLD_MAXBLOCKSIZE &&
		    !(dscp->dsc_featureflags &
		    DMU_BACKUP_FEATURE_LARGE_BLOCKS)) {
			while (srdp->datablksz > 0 && err == 0) {
				int n = MIN(srdp->datablksz,
				    SPA_OLD_MAXBLOCKSIZE);
				err = dmu_dump_write(dscp, srdp->obj_type,
				    range->object, offset, n, n, NULL, B_FALSE,
				    data);
				offset += n;
				/*
				 * When doing dry run, data==NULL is used as a
				 * sentinel value by
				 * dmu_dump_write()->dump_record().
				 */
				if (data != NULL)
					data += n;
				srdp->datablksz -= n;
			}
		} else {
			err = dmu_dump_write(dscp, srdp->obj_type,
			    range->object, offset,
			    srdp->datablksz, srdp->datasz, bp,
			    srdp->io_compressed, data);
		}
		return (err);
	}
	case HOLE: {
		struct srh *srhp = &range->sru.hole;
		if (range->object == DMU_META_DNODE_OBJECT) {
			uint32_t span = srhp->datablksz >> DNODE_SHIFT;
			uint64_t first_obj = range->start_blkid * span;
			uint64_t numobj = range->end_blkid * span - first_obj;
			return (dump_freeobjects(dscp, first_obj, numobj));
		}
		uint64_t offset = 0;

		/*
		 * If this multiply overflows, we don't need to send this block.
		 * Even if it has a birth time, it can never not be a hole, so
		 * we don't need to send records for it.
		 */
		if (!overflow_multiply(range->start_blkid, srhp->datablksz,
		    &offset)) {
			return (0);
		}
		uint64_t len = 0;

		if (!overflow_multiply(range->end_blkid, srhp->datablksz, &len))
			len = UINT64_MAX;
		len = len - offset;
		return (dump_free(dscp, range->object, offset, len));
	}
	default:
		panic("Invalid range type in do_dump: %d", range->type);
	}
	return (err);
}

static struct send_range *
range_alloc(enum type type, uint64_t object, uint64_t start_blkid,
    uint64_t end_blkid, boolean_t eos)
{
	struct send_range *range = kmem_alloc(sizeof (*range), KM_SLEEP);
	range->type = type;
	range->object = object;
	range->start_blkid = start_blkid;
	range->end_blkid = end_blkid;
	range->eos_marker = eos;
	if (type == DATA) {
		range->sru.data.abd = NULL;
		range->sru.data.abuf = NULL;
		mutex_init(&range->sru.data.lock, NULL, MUTEX_DEFAULT, NULL);
		cv_init(&range->sru.data.cv, NULL, CV_DEFAULT, NULL);
		range->sru.data.io_outstanding = 0;
		range->sru.data.io_err = 0;
		range->sru.data.io_compressed = B_FALSE;
	} else if (type == OBJECT) {
		range->sru.object.spill_range = NULL;
	}
	return (range);
}

/*
 * This is the callback function to traverse_dataset that acts as a worker
 * thread for dmu_send_impl.
 */
static int
send_cb(spa_t *spa, zilog_t *zilog, const blkptr_t *bp,
    const zbookmark_phys_t *zb, const struct dnode_phys *dnp, void *arg)
{
	(void) zilog;
	struct send_thread_arg *sta = arg;
	struct send_range *record;

	ASSERT(zb->zb_object == DMU_META_DNODE_OBJECT ||
	    zb->zb_object >= sta->resume.zb_object);

	/*
	 * All bps of an encrypted os should have the encryption bit set.
	 * If this is not true it indicates tampering and we report an error.
	 */
	if (sta->os->os_encrypted &&
	    !BP_IS_HOLE(bp) && !BP_USES_CRYPT(bp)) {
		spa_log_error(spa, zb, BP_GET_LOGICAL_BIRTH(bp));
		return (SET_ERROR(EIO));
	}

	if (sta->cancel)
		return (SET_ERROR(EINTR));
	if (zb->zb_object != DMU_META_DNODE_OBJECT &&
	    DMU_OBJECT_IS_SPECIAL(zb->zb_object))
		return (0);
	atomic_inc_64(sta->num_blocks_visited);

	if (zb->zb_level == ZB_DNODE_LEVEL) {
		if (zb->zb_object == DMU_META_DNODE_OBJECT)
			return (0);
		record = range_alloc(OBJECT, zb->zb_object, 0, 0, B_FALSE);
		record->sru.object.bp = *bp;
		size_t size  = sizeof (*dnp) * (dnp->dn_extra_slots + 1);
		record->sru.object.dnp = kmem_alloc(size, KM_SLEEP);
		memcpy(record->sru.object.dnp, dnp, size);
		bqueue_enqueue(&sta->q, record, sizeof (*record));
		return (0);
	}
	if (zb->zb_level == 0 && zb->zb_object == DMU_META_DNODE_OBJECT &&
	    !BP_IS_HOLE(bp)) {
		record = range_alloc(OBJECT_RANGE, 0, zb->zb_blkid,
		    zb->zb_blkid + 1, B_FALSE);
		record->sru.object_range.bp = *bp;
		bqueue_enqueue(&sta->q, record, sizeof (*record));
		return (0);
	}
	if (zb->zb_level < 0 || (zb->zb_level > 0 && !BP_IS_HOLE(bp)))
		return (0);
	if (zb->zb_object == DMU_META_DNODE_OBJECT && !BP_IS_HOLE(bp))
		return (0);

	uint64_t span = bp_span_in_blocks(dnp->dn_indblkshift, zb->zb_level);
	uint64_t start;

	/*
	 * If this multiply overflows, we don't need to send this block.
	 * Even if it has a birth time, it can never not be a hole, so
	 * we don't need to send records for it.
	 */
	if (!overflow_multiply(span, zb->zb_blkid, &start) || (!(zb->zb_blkid ==
	    DMU_SPILL_BLKID || DMU_OT_IS_METADATA(dnp->dn_type)) &&
	    span * zb->zb_blkid > dnp->dn_maxblkid)) {
		ASSERT(BP_IS_HOLE(bp));
		return (0);
	}

	if (zb->zb_blkid == DMU_SPILL_BLKID)
		ASSERT3U(BP_GET_TYPE(bp), ==, DMU_OT_SA);

	enum type record_type = DATA;
	if (BP_IS_HOLE(bp))
		record_type = HOLE;
	else if (BP_IS_REDACTED(bp))
		record_type = REDACT;
	else
		record_type = DATA;

	record = range_alloc(record_type, zb->zb_object, start,
	    (start + span < start ? 0 : start + span), B_FALSE);

	uint64_t datablksz = (zb->zb_blkid == DMU_SPILL_BLKID ?
	    BP_GET_LSIZE(bp) : dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT);

	if (BP_IS_HOLE(bp)) {
		record->sru.hole.datablksz = datablksz;
	} else if (BP_IS_REDACTED(bp)) {
		record->sru.redact.datablksz = datablksz;
	} else {
		record->sru.data.datablksz = datablksz;
		record->sru.data.obj_type = dnp->dn_type;
		record->sru.data.bp = *bp;
	}

	bqueue_enqueue(&sta->q, record, sizeof (*record));
	return (0);
}

struct redact_list_cb_arg {
	uint64_t *num_blocks_visited;
	bqueue_t *q;
	boolean_t *cancel;
	boolean_t mark_redact;
};

static int
redact_list_cb(redact_block_phys_t *rb, void *arg)
{
	struct redact_list_cb_arg *rlcap = arg;

	atomic_inc_64(rlcap->num_blocks_visited);
	if (*rlcap->cancel)
		return (-1);

	struct send_range *data = range_alloc(REDACT, rb->rbp_object,
	    rb->rbp_blkid, rb->rbp_blkid + redact_block_get_count(rb), B_FALSE);
	ASSERT3U(data->end_blkid, >, rb->rbp_blkid);
	if (rlcap->mark_redact) {
		data->type = REDACT;
		data->sru.redact.datablksz = redact_block_get_size(rb);
	} else {
		data->type = PREVIOUSLY_REDACTED;
	}
	bqueue_enqueue(rlcap->q, data, sizeof (*data));

	return (0);
}

/*
 * This function kicks off the traverse_dataset.  It also handles setting the
 * error code of the thread in case something goes wrong, and pushes the End of
 * Stream record when the traverse_dataset call has finished.
 */
static __attribute__((noreturn)) void
send_traverse_thread(void *arg)
{
	struct send_thread_arg *st_arg = arg;
	int err = 0;
	struct send_range *data;
	fstrans_cookie_t cookie = spl_fstrans_mark();

	err = traverse_dataset_resume(st_arg->os->os_dsl_dataset,
	    st_arg->fromtxg, &st_arg->resume,
	    st_arg->flags, send_cb, st_arg);

	if (err != EINTR)
		st_arg->error_code = err;
	data = range_alloc(DATA, 0, 0, 0, B_TRUE);
	bqueue_enqueue_flush(&st_arg->q, data, sizeof (*data));
	spl_fstrans_unmark(cookie);
	thread_exit();
}

/*
 * Utility function that causes End of Stream records to compare after of all
 * others, so that other threads' comparison logic can stay simple.
 */
static int __attribute__((unused))
send_range_after(const struct send_range *from, const struct send_range *to)
{
	if (from->eos_marker == B_TRUE)
		return (1);
	if (to->eos_marker == B_TRUE)
		return (-1);

	uint64_t from_obj = from->object;
	uint64_t from_end_obj = from->object + 1;
	uint64_t to_obj = to->object;
	uint64_t to_end_obj = to->object + 1;
	if (from_obj == 0) {
		ASSERT(from->type == HOLE || from->type == OBJECT_RANGE);
		from_obj = from->start_blkid << DNODES_PER_BLOCK_SHIFT;
		from_end_obj = from->end_blkid << DNODES_PER_BLOCK_SHIFT;
	}
	if (to_obj == 0) {
		ASSERT(to->type == HOLE || to->type == OBJECT_RANGE);
		to_obj = to->start_blkid << DNODES_PER_BLOCK_SHIFT;
		to_end_obj = to->end_blkid << DNODES_PER_BLOCK_SHIFT;
	}

	if (from_end_obj <= to_obj)
		return (-1);
	if (from_obj >= to_end_obj)
		return (1);
	int64_t cmp = TREE_CMP(to->type == OBJECT_RANGE, from->type ==
	    OBJECT_RANGE);
	if (unlikely(cmp))
		return (cmp);
	cmp = TREE_CMP(to->type == OBJECT, from->type == OBJECT);
	if (unlikely(cmp))
		return (cmp);
	if (from->end_blkid <= to->start_blkid)
		return (-1);
	if (from->start_blkid >= to->end_blkid)
		return (1);
	return (0);
}

/*
 * Pop the new data off the queue, check that the records we receive are in
 * the right order, but do not free the old data.  This is used so that the
 * records can be sent on to the main thread without copying the data.
 */
static struct send_range *
get_next_range_nofree(bqueue_t *bq, struct send_range *prev)
{
	struct send_range *next = bqueue_dequeue(bq);
	ASSERT3S(send_range_after(prev, next), ==, -1);
	return (next);
}

/*
 * Pop the new data off the queue, check that the records we receive are in
 * the right order, and free the old data.
 */
static struct send_range *
get_next_range(bqueue_t *bq, struct send_range *prev)
{
	struct send_range *next = get_next_range_nofree(bq, prev);
	range_free(prev);
	return (next);
}

static __attribute__((noreturn)) void
redact_list_thread(void *arg)
{
	struct redact_list_thread_arg *rlt_arg = arg;
	struct send_range *record;
	fstrans_cookie_t cookie = spl_fstrans_mark();
	if (rlt_arg->rl != NULL) {
		struct redact_list_cb_arg rlcba = {0};
		rlcba.cancel = &rlt_arg->cancel;
		rlcba.q = &rlt_arg->q;
		rlcba.num_blocks_visited = rlt_arg->num_blocks_visited;
		rlcba.mark_redact = rlt_arg->mark_redact;
		int err = dsl_redaction_list_traverse(rlt_arg->rl,
		    &rlt_arg->resume, redact_list_cb, &rlcba);
		if (err != EINTR)
			rlt_arg->error_code = err;
	}
	record = range_alloc(DATA, 0, 0, 0, B_TRUE);
	bqueue_enqueue_flush(&rlt_arg->q, record, sizeof (*record));
	spl_fstrans_unmark(cookie);

	thread_exit();
}

/*
 * Compare the start point of the two provided ranges. End of stream ranges
 * compare last, objects compare before any data or hole inside that object and
 * multi-object holes that start at the same object.
 */
static int
send_range_start_compare(struct send_range *r1, struct send_range *r2)
{
	uint64_t r1_objequiv = r1->object;
	uint64_t r1_l0equiv = r1->start_blkid;
	uint64_t r2_objequiv = r2->object;
	uint64_t r2_l0equiv = r2->start_blkid;
	int64_t cmp = TREE_CMP(r1->eos_marker, r2->eos_marker);
	if (unlikely(cmp))
		return (cmp);
	if (r1->object == 0) {
		r1_objequiv = r1->start_blkid * DNODES_PER_BLOCK;
		r1_l0equiv = 0;
	}
	if (r2->object == 0) {
		r2_objequiv = r2->start_blkid * DNODES_PER_BLOCK;
		r2_l0equiv = 0;
	}

	cmp = TREE_CMP(r1_objequiv, r2_objequiv);
	if (likely(cmp))
		return (cmp);
	cmp = TREE_CMP(r2->type == OBJECT_RANGE, r1->type == OBJECT_RANGE);
	if (unlikely(cmp))
		return (cmp);
	cmp = TREE_CMP(r2->type == OBJECT, r1->type == OBJECT);
	if (unlikely(cmp))
		return (cmp);

	return (TREE_CMP(r1_l0equiv, r2_l0equiv));
}

enum q_idx {
	REDACT_IDX = 0,
	TO_IDX,
	FROM_IDX,
	NUM_THREADS
};

/*
 * This function returns the next range the send_merge_thread should operate on.
 * The inputs are two arrays; the first one stores the range at the front of the
 * queues stored in the second one.  The ranges are sorted in descending
 * priority order; the metadata from earlier ranges overrules metadata from
 * later ranges.  out_mask is used to return which threads the ranges came from;
 * bit i is set if ranges[i] started at the same place as the returned range.
 *
 * This code is not hardcoded to compare a specific number of threads; it could
 * be used with any number, just by changing the q_idx enum.
 *
 * The "next range" is the one with the earliest start; if two starts are equal,
 * the highest-priority range is the next to operate on.  If a higher-priority
 * range starts in the middle of the first range, then the first range will be
 * truncated to end where the higher-priority range starts, and we will operate
 * on that one next time.   In this way, we make sure that each block covered by
 * some range gets covered by a returned range, and each block covered is
 * returned using the metadata of the highest-priority range it appears in.
 *
 * For example, if the three ranges at the front of the queues were [2,4),
 * [3,5), and [1,3), then the ranges returned would be [1,2) with the metadata
 * from the third range, [2,4) with the metadata from the first range, and then
 * [4,5) with the metadata from the second.
 */
static struct send_range *
find_next_range(struct send_range **ranges, bqueue_t **qs, uint64_t *out_mask)
{
	int idx = 0; // index of the range with the earliest start
	int i;
	uint64_t bmask = 0;
	for (i = 1; i < NUM_THREADS; i++) {
		if (send_range_start_compare(ranges[i], ranges[idx]) < 0)
			idx = i;
	}
	if (ranges[idx]->eos_marker) {
		struct send_range *ret = range_alloc(DATA, 0, 0, 0, B_TRUE);
		*out_mask = 0;
		return (ret);
	}
	/*
	 * Find all the ranges that start at that same point.
	 */
	for (i = 0; i < NUM_THREADS; i++) {
		if (send_range_start_compare(ranges[i], ranges[idx]) == 0)
			bmask |= 1 << i;
	}
	*out_mask = bmask;
	/*
	 * OBJECT_RANGE records only come from the TO thread, and should always
	 * be treated as overlapping with nothing and sent on immediately.  They
	 * are only used in raw sends, and are never redacted.
	 */
	if (ranges[idx]->type == OBJECT_RANGE) {
		ASSERT3U(idx, ==, TO_IDX);
		ASSERT3U(*out_mask, ==, 1 << TO_IDX);
		struct send_range *ret = ranges[idx];
		ranges[idx] = get_next_range_nofree(qs[idx], ranges[idx]);
		return (ret);
	}
	/*
	 * Find the first start or end point after the start of the first range.
	 */
	uint64_t first_change = ranges[idx]->end_blkid;
	for (i = 0; i < NUM_THREADS; i++) {
		if (i == idx || ranges[i]->eos_marker ||
		    ranges[i]->object > ranges[idx]->object ||
		    ranges[i]->object == DMU_META_DNODE_OBJECT)
			continue;
		ASSERT3U(ranges[i]->object, ==, ranges[idx]->object);
		if (first_change > ranges[i]->start_blkid &&
		    (bmask & (1 << i)) == 0)
			first_change = ranges[i]->start_blkid;
		else if (first_change > ranges[i]->end_blkid)
			first_change = ranges[i]->end_blkid;
	}
	/*
	 * Update all ranges to no longer overlap with the range we're
	 * returning. All such ranges must start at the same place as the range
	 * being returned, and end at or after first_change. Thus we update
	 * their start to first_change. If that makes them size 0, then free
	 * them and pull a new range from that thread.
	 */
	for (i = 0; i < NUM_THREADS; i++) {
		if (i == idx || (bmask & (1 << i)) == 0)
			continue;
		ASSERT3U(first_change, >, ranges[i]->start_blkid);
		ranges[i]->start_blkid = first_change;
		ASSERT3U(ranges[i]->start_blkid, <=, ranges[i]->end_blkid);
		if (ranges[i]->start_blkid == ranges[i]->end_blkid)
			ranges[i] = get_next_range(qs[i], ranges[i]);
	}
	/*
	 * Short-circuit the simple case; if the range doesn't overlap with
	 * anything else, or it only overlaps with things that start at the same
	 * place and are longer, send it on.
	 */
	if (first_change == ranges[idx]->end_blkid) {
		struct send_range *ret = ranges[idx];
		ranges[idx] = get_next_range_nofree(qs[idx], ranges[idx]);
		return (ret);
	}

	/*
	 * Otherwise, return a truncated copy of ranges[idx] and move the start
	 * of ranges[idx] back to first_change.
	 */
	struct send_range *ret = kmem_alloc(sizeof (*ret), KM_SLEEP);
	*ret = *ranges[idx];
	ret->end_blkid = first_change;
	ranges[idx]->start_blkid = first_change;
	return (ret);
}

#define	FROM_AND_REDACT_BITS ((1 << REDACT_IDX) | (1 << FROM_IDX))

/*
 * Merge the results from the from thread and the to thread, and then hand the
 * records off to send_prefetch_thread to prefetch them.  If this is not a
 * send from a redaction bookmark, the from thread will push an end of stream
 * record and stop, and we'll just send everything that was changed in the
 * to_ds since the ancestor's creation txg. If it is, then since
 * traverse_dataset has a canonical order, we can compare each change as
 * they're pulled off the queues.  That will give us a stream that is
 * appropriately sorted, and covers all records.  In addition, we pull the
 * data from the redact_list_thread and use that to determine which blocks
 * should be redacted.
 */
static __attribute__((noreturn)) void
send_merge_thread(void *arg)
{
	struct send_merge_thread_arg *smt_arg = arg;
	struct send_range *front_ranges[NUM_THREADS];
	bqueue_t *queues[NUM_THREADS];
	int err = 0;
	fstrans_cookie_t cookie = spl_fstrans_mark();

	if (smt_arg->redact_arg == NULL) {
		front_ranges[REDACT_IDX] =
		    kmem_zalloc(sizeof (struct send_range), KM_SLEEP);
		front_ranges[REDACT_IDX]->eos_marker = B_TRUE;
		front_ranges[REDACT_IDX]->type = REDACT;
		queues[REDACT_IDX] = NULL;
	} else {
		front_ranges[REDACT_IDX] =
		    bqueue_dequeue(&smt_arg->redact_arg->q);
		queues[REDACT_IDX] = &smt_arg->redact_arg->q;
	}
	front_ranges[TO_IDX] = bqueue_dequeue(&smt_arg->to_arg->q);
	queues[TO_IDX] = &smt_arg->to_arg->q;
	front_ranges[FROM_IDX] = bqueue_dequeue(&smt_arg->from_arg->q);
	queues[FROM_IDX] = &smt_arg->from_arg->q;
	uint64_t mask = 0;
	struct send_range *range;
	for (range = find_next_range(front_ranges, queues, &mask);
	    !range->eos_marker && err == 0 && !smt_arg->cancel;
	    range = find_next_range(front_ranges, queues, &mask)) {
		/*
		 * If the range in question was in both the from redact bookmark
		 * and the bookmark we're using to redact, then don't send it.
		 * It's already redacted on the receiving system, so a redaction
		 * record would be redundant.
		 */
		if ((mask & FROM_AND_REDACT_BITS) == FROM_AND_REDACT_BITS) {
			ASSERT3U(range->type, ==, REDACT);
			range_free(range);
			continue;
		}
		bqueue_enqueue(&smt_arg->q, range, sizeof (*range));

		if (smt_arg->to_arg->error_code != 0) {
			err = smt_arg->to_arg->error_code;
		} else if (smt_arg->from_arg->error_code != 0) {
			err = smt_arg->from_arg->error_code;
		} else if (smt_arg->redact_arg != NULL &&
		    smt_arg->redact_arg->error_code != 0) {
			err = smt_arg->redact_arg->error_code;
		}
	}
	if (smt_arg->cancel && err == 0)
		err = SET_ERROR(EINTR);
	smt_arg->error = err;
	if (smt_arg->error != 0) {
		smt_arg->to_arg->cancel = B_TRUE;
		smt_arg->from_arg->cancel = B_TRUE;
		if (smt_arg->redact_arg != NULL)
			smt_arg->redact_arg->cancel = B_TRUE;
	}
	for (int i = 0; i < NUM_THREADS; i++) {
		while (!front_ranges[i]->eos_marker) {
			front_ranges[i] = get_next_range(queues[i],
			    front_ranges[i]);
		}
		range_free(front_ranges[i]);
	}
	range->eos_marker = B_TRUE;
	bqueue_enqueue_flush(&smt_arg->q, range, 1);
	spl_fstrans_unmark(cookie);
	thread_exit();
}

struct send_reader_thread_arg {
	struct send_merge_thread_arg *smta;
	bqueue_t q;
	boolean_t cancel;
	boolean_t issue_reads;
	uint64_t featureflags;
	int error;
};

static void
dmu_send_read_done(zio_t *zio)
{
	struct send_range *range = zio->io_private;

	mutex_enter(&range->sru.data.lock);
	if (zio->io_error != 0) {
		abd_free(range->sru.data.abd);
		range->sru.data.abd = NULL;
		range->sru.data.io_err = zio->io_error;
	}

	ASSERT(range->sru.data.io_outstanding);
	range->sru.data.io_outstanding = B_FALSE;
	cv_broadcast(&range->sru.data.cv);
	mutex_exit(&range->sru.data.lock);
}

static void
issue_data_read(struct send_reader_thread_arg *srta, struct send_range *range)
{
	struct srd *srdp = &range->sru.data;
	blkptr_t *bp = &srdp->bp;
	objset_t *os = srta->smta->os;

	ASSERT3U(range->type, ==, DATA);
	ASSERT3U(range->start_blkid + 1, ==, range->end_blkid);
	/*
	 * If we have large blocks stored on disk but
	 * the send flags don't allow us to send large
	 * blocks, we split the data from the arc buf
	 * into chunks.
	 */
	boolean_t split_large_blocks =
	    srdp->datablksz > SPA_OLD_MAXBLOCKSIZE &&
	    !(srta->featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS);
	/*
	 * We should only request compressed data from the ARC if all
	 * the following are true:
	 *  - stream compression was requested
	 *  - we aren't splitting large blocks into smaller chunks
	 *  - the data won't need to be byteswapped before sending
	 *  - this isn't an embedded block
	 *  - this isn't metadata (if receiving on a different endian
	 *    system it can be byteswapped more easily)
	 */
	boolean_t request_compressed =
	    (srta->featureflags & DMU_BACKUP_FEATURE_COMPRESSED) &&
	    !split_large_blocks && !BP_SHOULD_BYTESWAP(bp) &&
	    !BP_IS_EMBEDDED(bp) && !DMU_OT_IS_METADATA(BP_GET_TYPE(bp));

	zio_flag_t zioflags = ZIO_FLAG_CANFAIL;

	if (srta->featureflags & DMU_BACKUP_FEATURE_RAW) {
		zioflags |= ZIO_FLAG_RAW;
		srdp->io_compressed = B_TRUE;
	} else if (request_compressed) {
		zioflags |= ZIO_FLAG_RAW_COMPRESS;
		srdp->io_compressed = B_TRUE;
	}

	srdp->datasz = (zioflags & ZIO_FLAG_RAW_COMPRESS) ?
	    BP_GET_PSIZE(bp) : BP_GET_LSIZE(bp);

	if (!srta->issue_reads)
		return;
	if (BP_IS_REDACTED(bp))
		return;
	if (send_do_embed(bp, srta->featureflags))
		return;

	zbookmark_phys_t zb = {
	    .zb_objset = dmu_objset_id(os),
	    .zb_object = range->object,
	    .zb_level = 0,
	    .zb_blkid = range->start_blkid,
	};

	arc_flags_t aflags = ARC_FLAG_CACHED_ONLY;

	int arc_err = arc_read(NULL, os->os_spa, bp,
	    arc_getbuf_func, &srdp->abuf, ZIO_PRIORITY_ASYNC_READ,
	    zioflags, &aflags, &zb);
	/*
	 * If the data is not already cached in the ARC, we read directly
	 * from zio.  This avoids the performance overhead of adding a new
	 * entry to the ARC, and we also avoid polluting the ARC cache with
	 * data that is not likely to be used in the future.
	 */
	if (arc_err != 0) {
		srdp->abd = abd_alloc_linear(srdp->datasz, B_FALSE);
		srdp->io_outstanding = B_TRUE;
		zio_nowait(zio_read(NULL, os->os_spa, bp, srdp->abd,
		    srdp->datasz, dmu_send_read_done, range,
		    ZIO_PRIORITY_ASYNC_READ, zioflags, &zb));
	}
}

/*
 * Create a new record with the given values.
 */
static void
enqueue_range(struct send_reader_thread_arg *srta, bqueue_t *q, dnode_t *dn,
    uint64_t blkid, uint64_t count, const blkptr_t *bp, uint32_t datablksz)
{
	enum type range_type = (bp == NULL || BP_IS_HOLE(bp) ? HOLE :
	    (BP_IS_REDACTED(bp) ? REDACT : DATA));

	struct send_range *range = range_alloc(range_type, dn->dn_object,
	    blkid, blkid + count, B_FALSE);

	if (blkid == DMU_SPILL_BLKID) {
		ASSERT3P(bp, !=, NULL);
		ASSERT3U(BP_GET_TYPE(bp), ==, DMU_OT_SA);
	}

	switch (range_type) {
	case HOLE:
		range->sru.hole.datablksz = datablksz;
		break;
	case DATA:
		ASSERT3U(count, ==, 1);
		range->sru.data.datablksz = datablksz;
		range->sru.data.obj_type = dn->dn_type;
		range->sru.data.bp = *bp;
		issue_data_read(srta, range);
		break;
	case REDACT:
		range->sru.redact.datablksz = datablksz;
		break;
	default:
		break;
	}
	bqueue_enqueue(q, range, datablksz);
}

/*
 * Send DRR_SPILL records for unmodified spill blocks.	This is useful
 * because changing certain attributes of the object (e.g. blocksize)
 * can cause old versions of ZFS to incorrectly remove a spill block.
 * Including these records in the stream forces an up to date version
 * to always be written ensuring they're never lost.  Current versions
 * of the code which understand the DRR_FLAG_SPILL_BLOCK feature can
 * ignore these unmodified spill blocks.
 *
 * We piggyback the spill_range to dnode range instead of enqueueing it
 * so send_range_after won't complain.
 */
static uint64_t
piggyback_unmodified_spill(struct send_reader_thread_arg *srta,
    struct send_range *range)
{
	ASSERT3U(range->type, ==, OBJECT);

	dnode_phys_t *dnp = range->sru.object.dnp;
	uint64_t fromtxg = srta->smta->to_arg->fromtxg;

	if (!zfs_send_unmodified_spill_blocks ||
	    !(dnp->dn_flags & DNODE_FLAG_SPILL_BLKPTR) ||
	    !(BP_GET_LOGICAL_BIRTH(DN_SPILL_BLKPTR(dnp)) <= fromtxg))
		return (0);

	blkptr_t *bp = DN_SPILL_BLKPTR(dnp);
	struct send_range *spill_range = range_alloc(DATA, range->object,
	    DMU_SPILL_BLKID, DMU_SPILL_BLKID+1, B_FALSE);
	spill_range->sru.data.bp = *bp;
	spill_range->sru.data.obj_type = dnp->dn_type;
	spill_range->sru.data.datablksz = BP_GET_LSIZE(bp);

	issue_data_read(srta, spill_range);
	range->sru.object.spill_range = spill_range;

	return (BP_GET_LSIZE(bp));
}

/*
 * This thread is responsible for two things: First, it retrieves the correct
 * blkptr in the to ds if we need to send the data because of something from
 * the from thread.  As a result of this, we're the first ones to discover that
 * some indirect blocks can be discarded because they're not holes. Second,
 * it issues prefetches for the data we need to send.
 */
static __attribute__((noreturn)) void
send_reader_thread(void *arg)
{
	struct send_reader_thread_arg *srta = arg;
	struct send_merge_thread_arg *smta = srta->smta;
	bqueue_t *inq = &smta->q;
	bqueue_t *outq = &srta->q;
	objset_t *os = smta->os;
	fstrans_cookie_t cookie = spl_fstrans_mark();
	struct send_range *range = bqueue_dequeue(inq);
	int err = 0;

	/*
	 * If the record we're analyzing is from a redaction bookmark from the
	 * fromds, then we need to know whether or not it exists in the tods so
	 * we know whether to create records for it or not. If it does, we need
	 * the datablksz so we can generate an appropriate record for it.
	 * Finally, if it isn't redacted, we need the blkptr so that we can send
	 * a WRITE record containing the actual data.
	 */
	uint64_t last_obj = UINT64_MAX;
	uint64_t last_obj_exists = B_TRUE;
	while (!range->eos_marker && !srta->cancel && smta->error == 0 &&
	    err == 0) {
		uint64_t spill = 0;
		switch (range->type) {
		case DATA:
			issue_data_read(srta, range);
			bqueue_enqueue(outq, range, range->sru.data.datablksz);
			range = get_next_range_nofree(inq, range);
			break;
		case OBJECT:
			spill = piggyback_unmodified_spill(srta, range);
			zfs_fallthrough;
		case HOLE:
		case OBJECT_RANGE:
		case REDACT: // Redacted blocks must exist
			bqueue_enqueue(outq, range, sizeof (*range) + spill);
			range = get_next_range_nofree(inq, range);
			break;
		case PREVIOUSLY_REDACTED: {
			/*
			 * This entry came from the "from bookmark" when
			 * sending from a bookmark that has a redaction
			 * list.  We need to check if this object/blkid
			 * exists in the target ("to") dataset, and if
			 * not then we drop this entry.  We also need
			 * to fill in the block pointer so that we know
			 * what to prefetch.
			 *
			 * To accomplish the above, we first cache whether or
			 * not the last object we examined exists.  If it
			 * doesn't, we can drop this record. If it does, we hold
			 * the dnode and use it to call dbuf_dnode_findbp. We do
			 * this instead of dbuf_bookmark_findbp because we will
			 * often operate on large ranges, and holding the dnode
			 * once is more efficient.
			 */
			boolean_t object_exists = B_TRUE;
			/*
			 * If the data is redacted, we only care if it exists,
			 * so that we don't send records for objects that have
			 * been deleted.
			 */
			dnode_t *dn;
			if (range->object == last_obj && !last_obj_exists) {
				/*
				 * If we're still examining the same object as
				 * previously, and it doesn't exist, we don't
				 * need to call dbuf_bookmark_findbp.
				 */
				object_exists = B_FALSE;
			} else {
				err = dnode_hold(os, range->object, FTAG, &dn);
				if (err == ENOENT) {
					object_exists = B_FALSE;
					err = 0;
				}
				last_obj = range->object;
				last_obj_exists = object_exists;
			}

			if (err != 0) {
				break;
			} else if (!object_exists) {
				/*
				 * The block was modified, but doesn't
				 * exist in the to dataset; if it was
				 * deleted in the to dataset, then we'll
				 * visit the hole bp for it at some point.
				 */
				range = get_next_range(inq, range);
				continue;
			}
			uint64_t file_max =
			    MIN(dn->dn_maxblkid, range->end_blkid);
			/*
			 * The object exists, so we need to try to find the
			 * blkptr for each block in the range we're processing.
			 */
			rw_enter(&dn->dn_struct_rwlock, RW_READER);
			for (uint64_t blkid = range->start_blkid;
			    blkid < file_max; blkid++) {
				blkptr_t bp;
				uint32_t datablksz =
				    dn->dn_phys->dn_datablkszsec <<
				    SPA_MINBLOCKSHIFT;
				uint64_t offset = blkid * datablksz;
				/*
				 * This call finds the next non-hole block in
				 * the object. This is to prevent a
				 * performance problem where we're unredacting
				 * a large hole. Using dnode_next_offset to
				 * skip over the large hole avoids iterating
				 * over every block in it.
				 */
				err = dnode_next_offset(dn, DNODE_FIND_HAVELOCK,
				    &offset, 1, 1, 0);
				if (err == ESRCH) {
					offset = UINT64_MAX;
					err = 0;
				} else if (err != 0) {
					break;
				}
				if (offset != blkid * datablksz) {
					/*
					 * if there is a hole from here
					 * (blkid) to offset
					 */
					offset = MIN(offset, file_max *
					    datablksz);
					uint64_t nblks = (offset / datablksz) -
					    blkid;
					enqueue_range(srta, outq, dn, blkid,
					    nblks, NULL, datablksz);
					blkid += nblks;
				}
				if (blkid >= file_max)
					break;
				err = dbuf_dnode_findbp(dn, 0, blkid, &bp,
				    NULL, NULL);
				if (err != 0)
					break;
				ASSERT(!BP_IS_HOLE(&bp));
				enqueue_range(srta, outq, dn, blkid, 1, &bp,
				    datablksz);
			}
			rw_exit(&dn->dn_struct_rwlock);
			dnode_rele(dn, FTAG);
			range = get_next_range(inq, range);
		}
		}
	}
	if (srta->cancel || err != 0) {
		smta->cancel = B_TRUE;
		srta->error = err;
	} else if (smta->error != 0) {
		srta->error = smta->error;
	}
	while (!range->eos_marker)
		range = get_next_range(inq, range);

	bqueue_enqueue_flush(outq, range, 1);
	spl_fstrans_unmark(cookie);
	thread_exit();
}

#define	NUM_SNAPS_NOT_REDACTED UINT64_MAX

struct dmu_send_params {
	/* Pool args */
	const void *tag; // Tag dp was held with, will be used to release dp.
	dsl_pool_t *dp;
	/* To snapshot args */
	const char *tosnap;
	dsl_dataset_t *to_ds;
	/* From snapshot args */
	zfs_bookmark_phys_t ancestor_zb;
	uint64_t *fromredactsnaps;
	/* NUM_SNAPS_NOT_REDACTED if not sending from redaction bookmark */
	uint64_t numfromredactsnaps;
	/* Stream params */
	boolean_t is_clone;
	boolean_t embedok;
	boolean_t large_block_ok;
	boolean_t compressok;
	boolean_t rawok;
	boolean_t savedok;
	uint64_t resumeobj;
	uint64_t resumeoff;
	uint64_t saved_guid;
	zfs_bookmark_phys_t *redactbook;
	/* Stream output params */
	dmu_send_outparams_t *dso;

	/* Stream progress params */
	offset_t *off;
	int outfd;
	char saved_toname[MAXNAMELEN];
};

static int
setup_featureflags(struct dmu_send_params *dspp, objset_t *os,
    uint64_t *featureflags)
{
	dsl_dataset_t *to_ds = dspp->to_ds;
	dsl_pool_t *dp = dspp->dp;

	if (dmu_objset_type(os) == DMU_OST_ZFS) {
		uint64_t version;
		if (zfs_get_zplprop(os, ZFS_PROP_VERSION, &version) != 0)
			return (SET_ERROR(EINVAL));

		if (version >= ZPL_VERSION_SA)
			*featureflags |= DMU_BACKUP_FEATURE_SA_SPILL;
	}

	/* raw sends imply large_block_ok */
	if ((dspp->rawok || dspp->large_block_ok) &&
	    dsl_dataset_feature_is_active(to_ds, SPA_FEATURE_LARGE_BLOCKS)) {
		*featureflags |= DMU_BACKUP_FEATURE_LARGE_BLOCKS;
	}

	/* encrypted datasets will not have embedded blocks */
	if ((dspp->embedok || dspp->rawok) && !os->os_encrypted &&
	    spa_feature_is_active(dp->dp_spa, SPA_FEATURE_EMBEDDED_DATA)) {
		*featureflags |= DMU_BACKUP_FEATURE_EMBED_DATA;
	}

	/* raw send implies compressok */
	if (dspp->compressok || dspp->rawok)
		*featureflags |= DMU_BACKUP_FEATURE_COMPRESSED;

	if (dspp->rawok && os->os_encrypted)
		*featureflags |= DMU_BACKUP_FEATURE_RAW;

	if ((*featureflags &
	    (DMU_BACKUP_FEATURE_EMBED_DATA | DMU_BACKUP_FEATURE_COMPRESSED |
	    DMU_BACKUP_FEATURE_RAW)) != 0 &&
	    spa_feature_is_active(dp->dp_spa, SPA_FEATURE_LZ4_COMPRESS)) {
		*featureflags |= DMU_BACKUP_FEATURE_LZ4;
	}

	/*
	 * We specifically do not include DMU_BACKUP_FEATURE_EMBED_DATA here to
	 * allow sending ZSTD compressed datasets to a receiver that does not
	 * support ZSTD
	 */
	if ((*featureflags &
	    (DMU_BACKUP_FEATURE_COMPRESSED | DMU_BACKUP_FEATURE_RAW)) != 0 &&
	    dsl_dataset_feature_is_active(to_ds, SPA_FEATURE_ZSTD_COMPRESS)) {
		*featureflags |= DMU_BACKUP_FEATURE_ZSTD;
	}

	if (dspp->resumeobj != 0 || dspp->resumeoff != 0) {
		*featureflags |= DMU_BACKUP_FEATURE_RESUMING;
	}

	if (dspp->redactbook != NULL) {
		*featureflags |= DMU_BACKUP_FEATURE_REDACTED;
	}

	if (dsl_dataset_feature_is_active(to_ds, SPA_FEATURE_LARGE_DNODE)) {
		*featureflags |= DMU_BACKUP_FEATURE_LARGE_DNODE;
	}

	if (dsl_dataset_feature_is_active(to_ds, SPA_FEATURE_LONGNAME)) {
		*featureflags |= DMU_BACKUP_FEATURE_LONGNAME;
	}

	if (dsl_dataset_feature_is_active(to_ds, SPA_FEATURE_LARGE_MICROZAP)) {
		/*
		 * We must never split a large microzap block, so we can only
		 * send large microzaps if LARGE_BLOCKS is already enabled.
		 */
		if (!(*featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS))
			return (SET_ERROR(ZFS_ERR_STREAM_LARGE_MICROZAP));
		*featureflags |= DMU_BACKUP_FEATURE_LARGE_MICROZAP;
	}

	return (0);
}

static dmu_replay_record_t *
create_begin_record(struct dmu_send_params *dspp, objset_t *os,
    uint64_t featureflags)
{
	dmu_replay_record_t *drr = kmem_zalloc(sizeof (dmu_replay_record_t),
	    KM_SLEEP);
	drr->drr_type = DRR_BEGIN;

	struct drr_begin *drrb = &drr->drr_u.drr_begin;
	dsl_dataset_t *to_ds = dspp->to_ds;

	drrb->drr_magic = DMU_BACKUP_MAGIC;
	drrb->drr_creation_time = dsl_dataset_phys(to_ds)->ds_creation_time;
	drrb->drr_type = dmu_objset_type(os);
	drrb->drr_toguid = dsl_dataset_phys(to_ds)->ds_guid;
	drrb->drr_fromguid = dspp->ancestor_zb.zbm_guid;

	DMU_SET_STREAM_HDRTYPE(drrb->drr_versioninfo, DMU_SUBSTREAM);
	DMU_SET_FEATUREFLAGS(drrb->drr_versioninfo, featureflags);

	if (dspp->is_clone)
		drrb->drr_flags |= DRR_FLAG_CLONE;
	if (dsl_dataset_phys(dspp->to_ds)->ds_flags & DS_FLAG_CI_DATASET)
		drrb->drr_flags |= DRR_FLAG_CI_DATA;
	if (zfs_send_set_freerecords_bit)
		drrb->drr_flags |= DRR_FLAG_FREERECORDS;
	drr->drr_u.drr_begin.drr_flags |= DRR_FLAG_SPILL_BLOCK;

	if (dspp->savedok) {
		drrb->drr_toguid = dspp->saved_guid;
		strlcpy(drrb->drr_toname, dspp->saved_toname,
		    sizeof (drrb->drr_toname));
	} else {
		dsl_dataset_name(to_ds, drrb->drr_toname);
		if (!to_ds->ds_is_snapshot) {
			(void) strlcat(drrb->drr_toname, "@--head--",
			    sizeof (drrb->drr_toname));
		}
	}
	return (drr);
}

static void
setup_to_thread(struct send_thread_arg *to_arg, objset_t *to_os,
    dmu_sendstatus_t *dssp, uint64_t fromtxg, boolean_t rawok)
{
	VERIFY0(bqueue_init(&to_arg->q, zfs_send_no_prefetch_queue_ff,
	    MAX(zfs_send_no_prefetch_queue_length, 2 * zfs_max_recordsize),
	    offsetof(struct send_range, ln)));
	to_arg->error_code = 0;
	to_arg->cancel = B_FALSE;
	to_arg->os = to_os;
	to_arg->fromtxg = fromtxg;
	to_arg->flags = TRAVERSE_PRE | TRAVERSE_PREFETCH_METADATA;
	if (rawok)
		to_arg->flags |= TRAVERSE_NO_DECRYPT;
	if (zfs_send_corrupt_data)
		to_arg->flags |= TRAVERSE_HARD;
	to_arg->num_blocks_visited = &dssp->dss_blocks;
	(void) thread_create(NULL, 0, send_traverse_thread, to_arg, 0,
	    curproc, TS_RUN, minclsyspri);
}

static void
setup_from_thread(struct redact_list_thread_arg *from_arg,
    redaction_list_t *from_rl, dmu_sendstatus_t *dssp)
{
	VERIFY0(bqueue_init(&from_arg->q, zfs_send_no_prefetch_queue_ff,
	    MAX(zfs_send_no_prefetch_queue_length, 2 * zfs_max_recordsize),
	    offsetof(struct send_range, ln)));
	from_arg->error_code = 0;
	from_arg->cancel = B_FALSE;
	from_arg->rl = from_rl;
	from_arg->mark_redact = B_FALSE;
	from_arg->num_blocks_visited = &dssp->dss_blocks;
	/*
	 * If from_ds is null, send_traverse_thread just returns success and
	 * enqueues an eos marker.
	 */
	(void) thread_create(NULL, 0, redact_list_thread, from_arg, 0,
	    curproc, TS_RUN, minclsyspri);
}

static void
setup_redact_list_thread(struct redact_list_thread_arg *rlt_arg,
    struct dmu_send_params *dspp, redaction_list_t *rl, dmu_sendstatus_t *dssp)
{
	if (dspp->redactbook == NULL)
		return;

	rlt_arg->cancel = B_FALSE;
	VERIFY0(bqueue_init(&rlt_arg->q, zfs_send_no_prefetch_queue_ff,
	    MAX(zfs_send_no_prefetch_queue_length, 2 * zfs_max_recordsize),
	    offsetof(struct send_range, ln)));
	rlt_arg->error_code = 0;
	rlt_arg->mark_redact = B_TRUE;
	rlt_arg->rl = rl;
	rlt_arg->num_blocks_visited = &dssp->dss_blocks;

	(void) thread_create(NULL, 0, redact_list_thread, rlt_arg, 0,
	    curproc, TS_RUN, minclsyspri);
}

static void
setup_merge_thread(struct send_merge_thread_arg *smt_arg,
    struct dmu_send_params *dspp, struct redact_list_thread_arg *from_arg,
    struct send_thread_arg *to_arg, struct redact_list_thread_arg *rlt_arg,
    objset_t *os)
{
	VERIFY0(bqueue_init(&smt_arg->q, zfs_send_no_prefetch_queue_ff,
	    MAX(zfs_send_no_prefetch_queue_length, 2 * zfs_max_recordsize),
	    offsetof(struct send_range, ln)));
	smt_arg->cancel = B_FALSE;
	smt_arg->error = 0;
	smt_arg->from_arg = from_arg;
	smt_arg->to_arg = to_arg;
	if (dspp->redactbook != NULL)
		smt_arg->redact_arg = rlt_arg;

	smt_arg->os = os;
	(void) thread_create(NULL, 0, send_merge_thread, smt_arg, 0, curproc,
	    TS_RUN, minclsyspri);
}

static void
setup_reader_thread(struct send_reader_thread_arg *srt_arg,
    struct dmu_send_params *dspp, struct send_merge_thread_arg *smt_arg,
    uint64_t featureflags)
{
	VERIFY0(bqueue_init(&srt_arg->q, zfs_send_queue_ff,
	    MAX(zfs_send_queue_length, 2 * zfs_max_recordsize),
	    offsetof(struct send_range, ln)));
	srt_arg->smta = smt_arg;
	srt_arg->issue_reads = !dspp->dso->dso_dryrun;
	srt_arg->featureflags = featureflags;
	(void) thread_create(NULL, 0, send_reader_thread, srt_arg, 0,
	    curproc, TS_RUN, minclsyspri);
}

static int
setup_resume_points(struct dmu_send_params *dspp,
    struct send_thread_arg *to_arg, struct redact_list_thread_arg *from_arg,
    struct redact_list_thread_arg *rlt_arg,
    struct send_merge_thread_arg *smt_arg, boolean_t resuming, objset_t *os,
    redaction_list_t *redact_rl, nvlist_t *nvl)
{
	(void) smt_arg;
	dsl_dataset_t *to_ds = dspp->to_ds;
	int err = 0;

	uint64_t obj = 0;
	uint64_t blkid = 0;
	if (resuming) {
		obj = dspp->resumeobj;
		dmu_object_info_t to_doi;
		err = dmu_object_info(os, obj, &to_doi);
		if (err != 0)
			return (err);

		blkid = dspp->resumeoff / to_doi.doi_data_block_size;
	}
	/*
	 * If we're resuming a redacted send, we can skip to the appropriate
	 * point in the redaction bookmark by binary searching through it.
	 */
	if (redact_rl != NULL) {
		SET_BOOKMARK(&rlt_arg->resume, to_ds->ds_object, obj, 0, blkid);
	}

	SET_BOOKMARK(&to_arg->resume, to_ds->ds_object, obj, 0, blkid);
	if (nvlist_exists(nvl, BEGINNV_REDACT_FROM_SNAPS)) {
		uint64_t objset = dspp->ancestor_zb.zbm_redaction_obj;
		/*
		 * Note: If the resume point is in an object whose
		 * blocksize is different in the from vs to snapshots,
		 * we will have divided by the "wrong" blocksize.
		 * However, in this case fromsnap's send_cb() will
		 * detect that the blocksize has changed and therefore
		 * ignore this object.
		 *
		 * If we're resuming a send from a redaction bookmark,
		 * we still cannot accidentally suggest blocks behind
		 * the to_ds.  In addition, we know that any blocks in
		 * the object in the to_ds will have to be sent, since
		 * the size changed.  Therefore, we can't cause any harm
		 * this way either.
		 */
		SET_BOOKMARK(&from_arg->resume, objset, obj, 0, blkid);
	}
	if (resuming) {
		fnvlist_add_uint64(nvl, BEGINNV_RESUME_OBJECT, dspp->resumeobj);
		fnvlist_add_uint64(nvl, BEGINNV_RESUME_OFFSET, dspp->resumeoff);
	}
	return (0);
}

static dmu_sendstatus_t *
setup_send_progress(struct dmu_send_params *dspp)
{
	dmu_sendstatus_t *dssp = kmem_zalloc(sizeof (*dssp), KM_SLEEP);
	dssp->dss_outfd = dspp->outfd;
	dssp->dss_off = dspp->off;
	dssp->dss_proc = curproc;
	mutex_enter(&dspp->to_ds->ds_sendstream_lock);
	list_insert_head(&dspp->to_ds->ds_sendstreams, dssp);
	mutex_exit(&dspp->to_ds->ds_sendstream_lock);
	return (dssp);
}

/*
 * Actually do the bulk of the work in a zfs send.
 *
 * The idea is that we want to do a send from ancestor_zb to to_ds.  We also
 * want to not send any data that has been modified by all the datasets in
 * redactsnaparr, and store the list of blocks that are redacted in this way in
 * a bookmark named redactbook, created on the to_ds.  We do this by creating
 * several worker threads, whose function is described below.
 *
 * There are three cases.
 * The first case is a redacted zfs send.  In this case there are 5 threads.
 * The first thread is the to_ds traversal thread: it calls dataset_traverse on
 * the to_ds and finds all the blocks that have changed since ancestor_zb (if
 * it's a full send, that's all blocks in the dataset).  It then sends those
 * blocks on to the send merge thread. The redact list thread takes the data
 * from the redaction bookmark and sends those blocks on to the send merge
 * thread.  The send merge thread takes the data from the to_ds traversal
 * thread, and combines it with the redaction records from the redact list
 * thread.  If a block appears in both the to_ds's data and the redaction data,
 * the send merge thread will mark it as redacted and send it on to the prefetch
 * thread.  Otherwise, the send merge thread will send the block on to the
 * prefetch thread unchanged. The prefetch thread will issue prefetch reads for
 * any data that isn't redacted, and then send the data on to the main thread.
 * The main thread behaves the same as in a normal send case, issuing demand
 * reads for data blocks and sending out records over the network
 *
 * The graphic below diagrams the flow of data in the case of a redacted zfs
 * send.  Each box represents a thread, and each line represents the flow of
 * data.
 *
 *             Records from the |
 *           redaction bookmark |
 * +--------------------+       |  +---------------------------+
 * |                    |       v  | Send Merge Thread         |
 * | Redact List Thread +----------> Apply redaction marks to  |
 * |                    |          | records as specified by   |
 * +--------------------+          | redaction ranges          |
 *                                 +----^---------------+------+
 *                                      |               | Merged data
 *                                      |               |
 *                                      |  +------------v--------+
 *                                      |  | Prefetch Thread     |
 * +--------------------+               |  | Issues prefetch     |
 * | to_ds Traversal    |               |  | reads of data blocks|
 * | Thread (finds      +---------------+  +------------+--------+
 * | candidate blocks)  |  Blocks modified              | Prefetched data
 * +--------------------+  by to_ds since               |
 *                         ancestor_zb     +------------v----+
 *                                         | Main Thread     |  File Descriptor
 *                                         | Sends data over +->(to zfs receive)
 *                                         | wire            |
 *                                         +-----------------+
 *
 * The second case is an incremental send from a redaction bookmark.  The to_ds
 * traversal thread and the main thread behave the same as in the redacted
 * send case.  The new thread is the from bookmark traversal thread.  It
 * iterates over the redaction list in the redaction bookmark, and enqueues
 * records for each block that was redacted in the original send.  The send
 * merge thread now has to merge the data from the two threads.  For details
 * about that process, see the header comment of send_merge_thread().  Any data
 * it decides to send on will be prefetched by the prefetch thread.  Note that
 * you can perform a redacted send from a redaction bookmark; in that case,
 * the data flow behaves very similarly to the flow in the redacted send case,
 * except with the addition of the bookmark traversal thread iterating over the
 * redaction bookmark.  The send_merge_thread also has to take on the
 * responsibility of merging the redact list thread's records, the bookmark
 * traversal thread's records, and the to_ds records.
 *
 * +---------------------+
 * |                     |
 * | Redact List Thread  +--------------+
 * |                     |              |
 * +---------------------+              |
 *        Blocks in redaction list      | Ranges modified by every secure snap
 *        of from bookmark              | (or EOS if not readcted)
 *                                      |
 * +---------------------+   |     +----v----------------------+
 * | bookmark Traversal  |   v     | Send Merge Thread         |
 * | Thread (finds       +---------> Merges bookmark, rlt, and |
 * | candidate blocks)   |         | to_ds send records        |
 * +---------------------+         +----^---------------+------+
 *                                      |               | Merged data
 *                                      |  +------------v--------+
 *                                      |  | Prefetch Thread     |
 * +--------------------+               |  | Issues prefetch     |
 * | to_ds Traversal    |               |  | reads of data blocks|
 * | Thread (finds      +---------------+  +------------+--------+
 * | candidate blocks)  |  Blocks modified              | Prefetched data
 * +--------------------+  by to_ds since  +------------v----+
 *                         ancestor_zb     | Main Thread     |  File Descriptor
 *                                         | Sends data over +->(to zfs receive)
 *                                         | wire            |
 *                                         +-----------------+
 *
 * The final case is a simple zfs full or incremental send.  The to_ds traversal
 * thread behaves the same as always. The redact list thread is never started.
 * The send merge thread takes all the blocks that the to_ds traversal thread
 * sends it, prefetches the data, and sends the blocks on to the main thread.
 * The main thread sends the data over the wire.
 *
 * To keep performance acceptable, we want to prefetch the data in the worker
 * threads.  While the to_ds thread could simply use the TRAVERSE_PREFETCH
 * feature built into traverse_dataset, the combining and deletion of records
 * due to redaction and sends from redaction bookmarks mean that we could
 * issue many unnecessary prefetches.  As a result, we only prefetch data
 * after we've determined that the record is not going to be redacted.  To
 * prevent the prefetching from getting too far ahead of the main thread, the
 * blocking queues that are used for communication are capped not by the
 * number of entries in the queue, but by the sum of the size of the
 * prefetches associated with them.  The limit on the amount of data that the
 * thread can prefetch beyond what the main thread has reached is controlled
 * by the global variable zfs_send_queue_length.  In addition, to prevent poor
 * performance in the beginning of a send, we also limit the distance ahead
 * that the traversal threads can be.  That distance is controlled by the
 * zfs_send_no_prefetch_queue_length tunable.
 *
 * Note: Releases dp using the specified tag.
 */
static int
dmu_send_impl(struct dmu_send_params *dspp)
{
	objset_t *os;
	dmu_replay_record_t *drr;
	dmu_sendstatus_t *dssp;
	dmu_send_cookie_t dsc = {0};
	int err;
	uint64_t fromtxg = dspp->ancestor_zb.zbm_creation_txg;
	uint64_t featureflags = 0;
	struct redact_list_thread_arg *from_arg;
	struct send_thread_arg *to_arg;
	struct redact_list_thread_arg *rlt_arg;
	struct send_merge_thread_arg *smt_arg;
	struct send_reader_thread_arg *srt_arg;
	struct send_range *range;
	redaction_list_t *from_rl = NULL;
	redaction_list_t *redact_rl = NULL;
	boolean_t resuming = (dspp->resumeobj != 0 || dspp->resumeoff != 0);
	boolean_t book_resuming = resuming;

	dsl_dataset_t *to_ds = dspp->to_ds;
	zfs_bookmark_phys_t *ancestor_zb = &dspp->ancestor_zb;
	dsl_pool_t *dp = dspp->dp;
	const void *tag = dspp->tag;

	err = dmu_objset_from_ds(to_ds, &os);
	if (err != 0) {
		dsl_pool_rele(dp, tag);
		return (err);
	}

	/*
	 * If this is a non-raw send of an encrypted ds, we can ensure that
	 * the objset_phys_t is authenticated. This is safe because this is
	 * either a snapshot or we have owned the dataset, ensuring that
	 * it can't be modified.
	 */
	if (!dspp->rawok && os->os_encrypted &&
	    arc_is_unauthenticated(os->os_phys_buf)) {
		zbookmark_phys_t zb;

		SET_BOOKMARK(&zb, to_ds->ds_object, ZB_ROOT_OBJECT,
		    ZB_ROOT_LEVEL, ZB_ROOT_BLKID);
		err = arc_untransform(os->os_phys_buf, os->os_spa,
		    &zb, B_FALSE);
		if (err != 0) {
			dsl_pool_rele(dp, tag);
			return (err);
		}

		ASSERT0(arc_is_unauthenticated(os->os_phys_buf));
	}

	if ((err = setup_featureflags(dspp, os, &featureflags)) != 0) {
		dsl_pool_rele(dp, tag);
		return (err);
	}

	/*
	 * If we're doing a redacted send, hold the bookmark's redaction list.
	 */
	if (dspp->redactbook != NULL) {
		err = dsl_redaction_list_hold_obj(dp,
		    dspp->redactbook->zbm_redaction_obj, FTAG,
		    &redact_rl);
		if (err != 0) {
			dsl_pool_rele(dp, tag);
			return (SET_ERROR(EINVAL));
		}
		dsl_redaction_list_long_hold(dp, redact_rl, FTAG);
	}

	/*
	 * If we're sending from a redaction bookmark, hold the redaction list
	 * so that we can consider sending the redacted blocks.
	 */
	if (ancestor_zb->zbm_redaction_obj != 0) {
		err = dsl_redaction_list_hold_obj(dp,
		    ancestor_zb->zbm_redaction_obj, FTAG, &from_rl);
		if (err != 0) {
			if (redact_rl != NULL) {
				dsl_redaction_list_long_rele(redact_rl, FTAG);
				dsl_redaction_list_rele(redact_rl, FTAG);
			}
			dsl_pool_rele(dp, tag);
			return (SET_ERROR(EINVAL));
		}
		dsl_redaction_list_long_hold(dp, from_rl, FTAG);
	}

	dsl_dataset_long_hold(to_ds, FTAG);

	from_arg = kmem_zalloc(sizeof (*from_arg), KM_SLEEP);
	to_arg = kmem_zalloc(sizeof (*to_arg), KM_SLEEP);
	rlt_arg = kmem_zalloc(sizeof (*rlt_arg), KM_SLEEP);
	smt_arg = kmem_zalloc(sizeof (*smt_arg), KM_SLEEP);
	srt_arg = kmem_zalloc(sizeof (*srt_arg), KM_SLEEP);

	drr = create_begin_record(dspp, os, featureflags);
	dssp = setup_send_progress(dspp);

	dsc.dsc_drr = drr;
	dsc.dsc_dso = dspp->dso;
	dsc.dsc_os = os;
	dsc.dsc_off = dspp->off;
	dsc.dsc_toguid = dsl_dataset_phys(to_ds)->ds_guid;
	dsc.dsc_fromtxg = fromtxg;
	dsc.dsc_pending_op = PENDING_NONE;
	dsc.dsc_featureflags = featureflags;
	dsc.dsc_resume_object = dspp->resumeobj;
	dsc.dsc_resume_offset = dspp->resumeoff;

	dsl_pool_rele(dp, tag);

	void *payload = NULL;
	size_t payload_len = 0;
	nvlist_t *nvl = fnvlist_alloc();

	/*
	 * If we're doing a redacted send, we include the snapshots we're
	 * redacted with respect to so that the target system knows what send
	 * streams can be correctly received on top of this dataset. If we're
	 * instead sending a redacted dataset, we include the snapshots that the
	 * dataset was created with respect to.
	 */
	if (dspp->redactbook != NULL) {
		fnvlist_add_uint64_array(nvl, BEGINNV_REDACT_SNAPS,
		    redact_rl->rl_phys->rlp_snaps,
		    redact_rl->rl_phys->rlp_num_snaps);
	} else if (dsl_dataset_feature_is_active(to_ds,
	    SPA_FEATURE_REDACTED_DATASETS)) {
		uint64_t *tods_guids;
		uint64_t length;
		VERIFY(dsl_dataset_get_uint64_array_feature(to_ds,
		    SPA_FEATURE_REDACTED_DATASETS, &length, &tods_guids));
		fnvlist_add_uint64_array(nvl, BEGINNV_REDACT_SNAPS, tods_guids,
		    length);
	}

	/*
	 * If we're sending from a redaction bookmark, then we should retrieve
	 * the guids of that bookmark so we can send them over the wire.
	 */
	if (from_rl != NULL) {
		fnvlist_add_uint64_array(nvl, BEGINNV_REDACT_FROM_SNAPS,
		    from_rl->rl_phys->rlp_snaps,
		    from_rl->rl_phys->rlp_num_snaps);
	}

	/*
	 * If the snapshot we're sending from is redacted, include the redaction
	 * list in the stream.
	 */
	if (dspp->numfromredactsnaps != NUM_SNAPS_NOT_REDACTED) {
		ASSERT3P(from_rl, ==, NULL);
		fnvlist_add_uint64_array(nvl, BEGINNV_REDACT_FROM_SNAPS,
		    dspp->fromredactsnaps, (uint_t)dspp->numfromredactsnaps);
		if (dspp->numfromredactsnaps > 0) {
			kmem_free(dspp->fromredactsnaps,
			    dspp->numfromredactsnaps * sizeof (uint64_t));
			dspp->fromredactsnaps = NULL;
		}
	}

	if (resuming || book_resuming) {
		err = setup_resume_points(dspp, to_arg, from_arg,
		    rlt_arg, smt_arg, resuming, os, redact_rl, nvl);
		if (err != 0)
			goto out;
	}

	if (featureflags & DMU_BACKUP_FEATURE_RAW) {
		uint64_t ivset_guid = ancestor_zb->zbm_ivset_guid;
		nvlist_t *keynvl = NULL;
		ASSERT(os->os_encrypted);

		err = dsl_crypto_populate_key_nvlist(os, ivset_guid,
		    &keynvl);
		if (err != 0) {
			fnvlist_free(nvl);
			goto out;
		}

		fnvlist_add_nvlist(nvl, "crypt_keydata", keynvl);
		fnvlist_free(keynvl);
	}

	if (!nvlist_empty(nvl)) {
		payload = fnvlist_pack(nvl, &payload_len);
		drr->drr_payloadlen = payload_len;
	}

	fnvlist_free(nvl);
	err = dump_record(&dsc, payload, payload_len);
	fnvlist_pack_free(payload, payload_len);
	if (err != 0) {
		err = dsc.dsc_err;
		goto out;
	}

	setup_to_thread(to_arg, os, dssp, fromtxg, dspp->rawok);
	setup_from_thread(from_arg, from_rl, dssp);
	setup_redact_list_thread(rlt_arg, dspp, redact_rl, dssp);
	setup_merge_thread(smt_arg, dspp, from_arg, to_arg, rlt_arg, os);
	setup_reader_thread(srt_arg, dspp, smt_arg, featureflags);

	range = bqueue_dequeue(&srt_arg->q);
	while (err == 0 && !range->eos_marker) {
		err = do_dump(&dsc, range);
		range = get_next_range(&srt_arg->q, range);
		if (issig())
			err = SET_ERROR(EINTR);
	}

	/*
	 * If we hit an error or are interrupted, cancel our worker threads and
	 * clear the queue of any pending records.  The threads will pass the
	 * cancel up the tree of worker threads, and each one will clean up any
	 * pending records before exiting.
	 */
	if (err != 0) {
		srt_arg->cancel = B_TRUE;
		while (!range->eos_marker) {
			range = get_next_range(&srt_arg->q, range);
		}
	}
	range_free(range);

	bqueue_destroy(&srt_arg->q);
	bqueue_destroy(&smt_arg->q);
	if (dspp->redactbook != NULL)
		bqueue_destroy(&rlt_arg->q);
	bqueue_destroy(&to_arg->q);
	bqueue_destroy(&from_arg->q);

	if (err == 0 && srt_arg->error != 0)
		err = srt_arg->error;

	if (err != 0)
		goto out;

	if (dsc.dsc_pending_op != PENDING_NONE)
		if (dump_record(&dsc, NULL, 0) != 0)
			err = SET_ERROR(EINTR);

	if (err != 0) {
		if (err == EINTR && dsc.dsc_err != 0)
			err = dsc.dsc_err;
		goto out;
	}

	/*
	 * Send the DRR_END record if this is not a saved stream.
	 * Otherwise, the omitted DRR_END record will signal to
	 * the receive side that the stream is incomplete.
	 */
	if (!dspp->savedok) {
		memset(drr, 0, sizeof (dmu_replay_record_t));
		drr->drr_type = DRR_END;
		drr->drr_u.drr_end.drr_checksum = dsc.dsc_zc;
		drr->drr_u.drr_end.drr_toguid = dsc.dsc_toguid;

		if (dump_record(&dsc, NULL, 0) != 0)
			err = dsc.dsc_err;
	}
out:
	mutex_enter(&to_ds->ds_sendstream_lock);
	list_remove(&to_ds->ds_sendstreams, dssp);
	mutex_exit(&to_ds->ds_sendstream_lock);

	VERIFY(err != 0 || (dsc.dsc_sent_begin &&
	    (dsc.dsc_sent_end || dspp->savedok)));

	kmem_free(drr, sizeof (dmu_replay_record_t));
	kmem_free(dssp, sizeof (dmu_sendstatus_t));
	kmem_free(from_arg, sizeof (*from_arg));
	kmem_free(to_arg, sizeof (*to_arg));
	kmem_free(rlt_arg, sizeof (*rlt_arg));
	kmem_free(smt_arg, sizeof (*smt_arg));
	kmem_free(srt_arg, sizeof (*srt_arg));

	dsl_dataset_long_rele(to_ds, FTAG);
	if (from_rl != NULL) {
		dsl_redaction_list_long_rele(from_rl, FTAG);
		dsl_redaction_list_rele(from_rl, FTAG);
	}
	if (redact_rl != NULL) {
		dsl_redaction_list_long_rele(redact_rl, FTAG);
		dsl_redaction_list_rele(redact_rl, FTAG);
	}

	return (err);
}

int
dmu_send_obj(const char *pool, uint64_t tosnap, uint64_t fromsnap,
    boolean_t embedok, boolean_t large_block_ok, boolean_t compressok,
    boolean_t rawok, boolean_t savedok, int outfd, offset_t *off,
    dmu_send_outparams_t *dsop)
{
	int err;
	dsl_dataset_t *fromds;
	ds_hold_flags_t dsflags;
	struct dmu_send_params dspp = {0};
	dspp.embedok = embedok;
	dspp.large_block_ok = large_block_ok;
	dspp.compressok = compressok;
	dspp.outfd = outfd;
	dspp.off = off;
	dspp.dso = dsop;
	dspp.tag = FTAG;
	dspp.rawok = rawok;
	dspp.savedok = savedok;

	dsflags = (rawok) ? DS_HOLD_FLAG_NONE : DS_HOLD_FLAG_DECRYPT;
	err = dsl_pool_hold(pool, FTAG, &dspp.dp);
	if (err != 0)
		return (err);

	err = dsl_dataset_hold_obj_flags(dspp.dp, tosnap, dsflags, FTAG,
	    &dspp.to_ds);
	if (err != 0) {
		dsl_pool_rele(dspp.dp, FTAG);
		return (err);
	}

	if (fromsnap != 0) {
		err = dsl_dataset_hold_obj(dspp.dp, fromsnap, FTAG, &fromds);

		if (err != 0) {
			dsl_dataset_rele_flags(dspp.to_ds, dsflags, FTAG);
			dsl_pool_rele(dspp.dp, FTAG);
			return (err);
		}
		dspp.ancestor_zb.zbm_guid = dsl_dataset_phys(fromds)->ds_guid;
		dspp.ancestor_zb.zbm_creation_txg =
		    dsl_dataset_phys(fromds)->ds_creation_txg;
		dspp.ancestor_zb.zbm_creation_time =
		    dsl_dataset_phys(fromds)->ds_creation_time;

		if (dsl_dataset_is_zapified(fromds)) {
			(void) zap_lookup(dspp.dp->dp_meta_objset,
			    fromds->ds_object, DS_FIELD_IVSET_GUID, 8, 1,
			    &dspp.ancestor_zb.zbm_ivset_guid);
		}

		/* See dmu_send for the reasons behind this. */
		uint64_t *fromredact;

		if (!dsl_dataset_get_uint64_array_feature(fromds,
		    SPA_FEATURE_REDACTED_DATASETS,
		    &dspp.numfromredactsnaps,
		    &fromredact)) {
			dspp.numfromredactsnaps = NUM_SNAPS_NOT_REDACTED;
		} else if (dspp.numfromredactsnaps > 0) {
			uint64_t size = dspp.numfromredactsnaps *
			    sizeof (uint64_t);
			dspp.fromredactsnaps = kmem_zalloc(size, KM_SLEEP);
			memcpy(dspp.fromredactsnaps, fromredact, size);
		}

		boolean_t is_before =
		    dsl_dataset_is_before(dspp.to_ds, fromds, 0);
		dspp.is_clone = (dspp.to_ds->ds_dir !=
		    fromds->ds_dir);
		dsl_dataset_rele(fromds, FTAG);
		if (!is_before) {
			dsl_pool_rele(dspp.dp, FTAG);
			err = SET_ERROR(EXDEV);
		} else {
			err = dmu_send_impl(&dspp);
		}
	} else {
		dspp.numfromredactsnaps = NUM_SNAPS_NOT_REDACTED;
		err = dmu_send_impl(&dspp);
	}
	if (dspp.fromredactsnaps)
		kmem_free(dspp.fromredactsnaps,
		    dspp.numfromredactsnaps * sizeof (uint64_t));

	dsl_dataset_rele_flags(dspp.to_ds, dsflags, FTAG);
	return (err);
}

int
dmu_send(const char *tosnap, const char *fromsnap, boolean_t embedok,
    boolean_t large_block_ok, boolean_t compressok, boolean_t rawok,
    boolean_t savedok, uint64_t resumeobj, uint64_t resumeoff,
    const char *redactbook, int outfd, offset_t *off,
    dmu_send_outparams_t *dsop)
{
	int err = 0;
	ds_hold_flags_t dsflags;
	boolean_t owned = B_FALSE;
	dsl_dataset_t *fromds = NULL;
	zfs_bookmark_phys_t book = {0};
	struct dmu_send_params dspp = {0};

	dsflags = (rawok) ? DS_HOLD_FLAG_NONE : DS_HOLD_FLAG_DECRYPT;
	dspp.tosnap = tosnap;
	dspp.embedok = embedok;
	dspp.large_block_ok = large_block_ok;
	dspp.compressok = compressok;
	dspp.outfd = outfd;
	dspp.off = off;
	dspp.dso = dsop;
	dspp.tag = FTAG;
	dspp.resumeobj = resumeobj;
	dspp.resumeoff = resumeoff;
	dspp.rawok = rawok;
	dspp.savedok = savedok;

	if (fromsnap != NULL && strpbrk(fromsnap, "@#") == NULL)
		return (SET_ERROR(EINVAL));

	err = dsl_pool_hold(tosnap, FTAG, &dspp.dp);
	if (err != 0)
		return (err);

	if (strchr(tosnap, '@') == NULL && spa_writeable(dspp.dp->dp_spa)) {
		/*
		 * We are sending a filesystem or volume.  Ensure
		 * that it doesn't change by owning the dataset.
		 */

		if (savedok) {
			/*
			 * We are looking for the dataset that represents the
			 * partially received send stream. If this stream was
			 * received as a new snapshot of an existing dataset,
			 * this will be saved in a hidden clone named
			 * "<pool>/<dataset>/%recv". Otherwise, the stream
			 * will be saved in the live dataset itself. In
			 * either case we need to use dsl_dataset_own_force()
			 * because the stream is marked as inconsistent,
			 * which would normally make it unavailable to be
			 * owned.
			 */
			char *name = kmem_asprintf("%s/%s", tosnap,
			    recv_clone_name);
			err = dsl_dataset_own_force(dspp.dp, name, dsflags,
			    FTAG, &dspp.to_ds);
			if (err == ENOENT) {
				err = dsl_dataset_own_force(dspp.dp, tosnap,
				    dsflags, FTAG, &dspp.to_ds);
			}

			if (err == 0) {
				owned = B_TRUE;
				err = zap_lookup(dspp.dp->dp_meta_objset,
				    dspp.to_ds->ds_object,
				    DS_FIELD_RESUME_TOGUID, 8, 1,
				    &dspp.saved_guid);
			}

			if (err == 0) {
				err = zap_lookup(dspp.dp->dp_meta_objset,
				    dspp.to_ds->ds_object,
				    DS_FIELD_RESUME_TONAME, 1,
				    sizeof (dspp.saved_toname),
				    dspp.saved_toname);
			}
			/* Only disown if there was an error in the lookups */
			if (owned && (err != 0))
				dsl_dataset_disown(dspp.to_ds, dsflags, FTAG);

			kmem_strfree(name);
		} else {
			err = dsl_dataset_own(dspp.dp, tosnap, dsflags,
			    FTAG, &dspp.to_ds);
			if (err == 0)
				owned = B_TRUE;
		}
	} else {
		err = dsl_dataset_hold_flags(dspp.dp, tosnap, dsflags, FTAG,
		    &dspp.to_ds);
	}

	if (err != 0) {
		/* Note: dsl dataset is not owned at this point */
		dsl_pool_rele(dspp.dp, FTAG);
		return (err);
	}

	if (redactbook != NULL) {
		char path[ZFS_MAX_DATASET_NAME_LEN];
		(void) strlcpy(path, tosnap, sizeof (path));
		char *at = strchr(path, '@');
		if (at == NULL) {
			err = EINVAL;
		} else {
			(void) snprintf(at, sizeof (path) - (at - path), "#%s",
			    redactbook);
			err = dsl_bookmark_lookup(dspp.dp, path,
			    NULL, &book);
			dspp.redactbook = &book;
		}
	}

	if (err != 0) {
		dsl_pool_rele(dspp.dp, FTAG);
		if (owned)
			dsl_dataset_disown(dspp.to_ds, dsflags, FTAG);
		else
			dsl_dataset_rele_flags(dspp.to_ds, dsflags, FTAG);
		return (err);
	}

	if (fromsnap != NULL) {
		zfs_bookmark_phys_t *zb = &dspp.ancestor_zb;
		int fsnamelen;
		if (strpbrk(tosnap, "@#") != NULL)
			fsnamelen = strpbrk(tosnap, "@#") - tosnap;
		else
			fsnamelen = strlen(tosnap);

		/*
		 * If the fromsnap is in a different filesystem, then
		 * mark the send stream as a clone.
		 */
		if (strncmp(tosnap, fromsnap, fsnamelen) != 0 ||
		    (fromsnap[fsnamelen] != '@' &&
		    fromsnap[fsnamelen] != '#')) {
			dspp.is_clone = B_TRUE;
		}

		if (strchr(fromsnap, '@') != NULL) {
			err = dsl_dataset_hold(dspp.dp, fromsnap, FTAG,
			    &fromds);

			if (err != 0) {
				ASSERT3P(fromds, ==, NULL);
			} else {
				/*
				 * We need to make a deep copy of the redact
				 * snapshots of the from snapshot, because the
				 * array will be freed when we evict from_ds.
				 */
				uint64_t *fromredact;
				if (!dsl_dataset_get_uint64_array_feature(
				    fromds, SPA_FEATURE_REDACTED_DATASETS,
				    &dspp.numfromredactsnaps,
				    &fromredact)) {
					dspp.numfromredactsnaps =
					    NUM_SNAPS_NOT_REDACTED;
				} else if (dspp.numfromredactsnaps > 0) {
					uint64_t size =
					    dspp.numfromredactsnaps *
					    sizeof (uint64_t);
					dspp.fromredactsnaps = kmem_zalloc(size,
					    KM_SLEEP);
					memcpy(dspp.fromredactsnaps, fromredact,
					    size);
				}
				if (!dsl_dataset_is_before(dspp.to_ds, fromds,
				    0)) {
					err = SET_ERROR(EXDEV);
				} else {
					zb->zbm_creation_txg =
					    dsl_dataset_phys(fromds)->
					    ds_creation_txg;
					zb->zbm_creation_time =
					    dsl_dataset_phys(fromds)->
					    ds_creation_time;
					zb->zbm_guid =
					    dsl_dataset_phys(fromds)->ds_guid;
					zb->zbm_redaction_obj = 0;

					if (dsl_dataset_is_zapified(fromds)) {
						(void) zap_lookup(
						    dspp.dp->dp_meta_objset,
						    fromds->ds_object,
						    DS_FIELD_IVSET_GUID, 8, 1,
						    &zb->zbm_ivset_guid);
					}
				}
				dsl_dataset_rele(fromds, FTAG);
			}
		} else {
			dspp.numfromredactsnaps = NUM_SNAPS_NOT_REDACTED;
			err = dsl_bookmark_lookup(dspp.dp, fromsnap, dspp.to_ds,
			    zb);
			if (err == EXDEV && zb->zbm_redaction_obj != 0 &&
			    zb->zbm_guid ==
			    dsl_dataset_phys(dspp.to_ds)->ds_guid)
				err = 0;
		}

		if (err == 0) {
			/* dmu_send_impl will call dsl_pool_rele for us. */
			err = dmu_send_impl(&dspp);
		} else {
			if (dspp.fromredactsnaps)
				kmem_free(dspp.fromredactsnaps,
				    dspp.numfromredactsnaps *
				    sizeof (uint64_t));
			dsl_pool_rele(dspp.dp, FTAG);
		}
	} else {
		dspp.numfromredactsnaps = NUM_SNAPS_NOT_REDACTED;
		err = dmu_send_impl(&dspp);
	}
	if (owned)
		dsl_dataset_disown(dspp.to_ds, dsflags, FTAG);
	else
		dsl_dataset_rele_flags(dspp.to_ds, dsflags, FTAG);
	return (err);
}

static int
dmu_adjust_send_estimate_for_indirects(dsl_dataset_t *ds, uint64_t uncompressed,
    uint64_t compressed, boolean_t stream_compressed, uint64_t *sizep)
{
	int err = 0;
	uint64_t size;
	/*
	 * Assume that space (both on-disk and in-stream) is dominated by
	 * data.  We will adjust for indirect blocks and the copies property,
	 * but ignore per-object space used (eg, dnodes and DRR_OBJECT records).
	 */

	uint64_t recordsize;
	uint64_t record_count;
	objset_t *os;
	VERIFY0(dmu_objset_from_ds(ds, &os));

	/* Assume all (uncompressed) blocks are recordsize. */
	if (zfs_override_estimate_recordsize != 0) {
		recordsize = zfs_override_estimate_recordsize;
	} else if (os->os_phys->os_type == DMU_OST_ZVOL) {
		err = dsl_prop_get_int_ds(ds,
		    zfs_prop_to_name(ZFS_PROP_VOLBLOCKSIZE), &recordsize);
	} else {
		err = dsl_prop_get_int_ds(ds,
		    zfs_prop_to_name(ZFS_PROP_RECORDSIZE), &recordsize);
	}
	if (err != 0)
		return (err);
	record_count = uncompressed / recordsize;

	/*
	 * If we're estimating a send size for a compressed stream, use the
	 * compressed data size to estimate the stream size. Otherwise, use the
	 * uncompressed data size.
	 */
	size = stream_compressed ? compressed : uncompressed;

	/*
	 * Subtract out approximate space used by indirect blocks.
	 * Assume most space is used by data blocks (non-indirect, non-dnode).
	 * Assume no ditto blocks or internal fragmentation.
	 *
	 * Therefore, space used by indirect blocks is sizeof(blkptr_t) per
	 * block.
	 */
	size -= record_count * sizeof (blkptr_t);

	/* Add in the space for the record associated with each block. */
	size += record_count * sizeof (dmu_replay_record_t);

	*sizep = size;

	return (0);
}

int
dmu_send_estimate_fast(dsl_dataset_t *origds, dsl_dataset_t *fromds,
    zfs_bookmark_phys_t *frombook, boolean_t stream_compressed,
    boolean_t saved, uint64_t *sizep)
{
	int err;
	dsl_dataset_t *ds = origds;
	uint64_t uncomp, comp;

	ASSERT(dsl_pool_config_held(origds->ds_dir->dd_pool));
	ASSERT(fromds == NULL || frombook == NULL);

	/*
	 * If this is a saved send we may actually be sending
	 * from the %recv clone used for resuming.
	 */
	if (saved) {
		objset_t *mos = origds->ds_dir->dd_pool->dp_meta_objset;
		uint64_t guid;
		char dsname[ZFS_MAX_DATASET_NAME_LEN + 6];

		dsl_dataset_name(origds, dsname);
		(void) strcat(dsname, "/");
		(void) strlcat(dsname, recv_clone_name, sizeof (dsname));

		err = dsl_dataset_hold(origds->ds_dir->dd_pool,
		    dsname, FTAG, &ds);
		if (err != ENOENT && err != 0) {
			return (err);
		} else if (err == ENOENT) {
			ds = origds;
		}

		/* check that this dataset has partially received data */
		err = zap_lookup(mos, ds->ds_object,
		    DS_FIELD_RESUME_TOGUID, 8, 1, &guid);
		if (err != 0) {
			err = SET_ERROR(err == ENOENT ? EINVAL : err);
			goto out;
		}

		err = zap_lookup(mos, ds->ds_object,
		    DS_FIELD_RESUME_TONAME, 1, sizeof (dsname), dsname);
		if (err != 0) {
			err = SET_ERROR(err == ENOENT ? EINVAL : err);
			goto out;
		}
	}

	/* tosnap must be a snapshot or the target of a saved send */
	if (!ds->ds_is_snapshot && ds == origds)
		return (SET_ERROR(EINVAL));

	if (fromds != NULL) {
		uint64_t used;
		if (!fromds->ds_is_snapshot) {
			err = SET_ERROR(EINVAL);
			goto out;
		}

		if (!dsl_dataset_is_before(ds, fromds, 0)) {
			err = SET_ERROR(EXDEV);
			goto out;
		}

		err = dsl_dataset_space_written(fromds, ds, &used, &comp,
		    &uncomp);
		if (err != 0)
			goto out;
	} else if (frombook != NULL) {
		uint64_t used;
		err = dsl_dataset_space_written_bookmark(frombook, ds, &used,
		    &comp, &uncomp);
		if (err != 0)
			goto out;
	} else {
		uncomp = dsl_dataset_phys(ds)->ds_uncompressed_bytes;
		comp = dsl_dataset_phys(ds)->ds_compressed_bytes;
	}

	err = dmu_adjust_send_estimate_for_indirects(ds, uncomp, comp,
	    stream_compressed, sizep);
	/*
	 * Add the size of the BEGIN and END records to the estimate.
	 */
	*sizep += 2 * sizeof (dmu_replay_record_t);

out:
	if (ds != origds)
		dsl_dataset_rele(ds, FTAG);
	return (err);
}

ZFS_MODULE_PARAM(zfs_send, zfs_send_, corrupt_data, INT, ZMOD_RW,
	"Allow sending corrupt data");

ZFS_MODULE_PARAM(zfs_send, zfs_send_, queue_length, UINT, ZMOD_RW,
	"Maximum send queue length");

ZFS_MODULE_PARAM(zfs_send, zfs_send_, unmodified_spill_blocks, INT, ZMOD_RW,
	"Send unmodified spill blocks");

ZFS_MODULE_PARAM(zfs_send, zfs_send_, no_prefetch_queue_length, UINT, ZMOD_RW,
	"Maximum send queue length for non-prefetch queues");

ZFS_MODULE_PARAM(zfs_send, zfs_send_, queue_ff, UINT, ZMOD_RW,
	"Send queue fill fraction");

ZFS_MODULE_PARAM(zfs_send, zfs_send_, no_prefetch_queue_ff, UINT, ZMOD_RW,
	"Send queue fill fraction for non-prefetch queues");

ZFS_MODULE_PARAM(zfs_send, zfs_, override_estimate_recordsize, UINT, ZMOD_RW,
	"Override block size estimate with fixed size");