xref: /freebsd/sys/contrib/openzfs/module/zfs/dmu_send.c (revision e6bfd18d21b225af6a0ed67ceeaf1293b7b9eba5)
1 /*
2  * CDDL HEADER START
3  *
4  * The contents of this file are subject to the terms of the
5  * Common Development and Distribution License (the "License").
6  * You may not use this file except in compliance with the License.
7  *
8  * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
9  * or https://opensource.org/licenses/CDDL-1.0.
10  * See the License for the specific language governing permissions
11  * and limitations under the License.
12  *
13  * When distributing Covered Code, include this CDDL HEADER in each
14  * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
15  * If applicable, add the following below this CDDL HEADER, with the
16  * fields enclosed by brackets "[]" replaced with your own identifying
17  * information: Portions Copyright [yyyy] [name of copyright owner]
18  *
19  * CDDL HEADER END
20  */
21 /*
22  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
23  * Copyright 2011 Nexenta Systems, Inc. All rights reserved.
24  * Copyright (c) 2011, 2018 by Delphix. All rights reserved.
25  * Copyright (c) 2014, Joyent, Inc. All rights reserved.
26  * Copyright 2014 HybridCluster. All rights reserved.
27  * Copyright 2016 RackTop Systems.
28  * Copyright (c) 2016 Actifio, Inc. All rights reserved.
29  * Copyright (c) 2019, Klara Inc.
30  * Copyright (c) 2019, Allan Jude
31  */
32 
33 #include <sys/dmu.h>
34 #include <sys/dmu_impl.h>
35 #include <sys/dmu_tx.h>
36 #include <sys/dbuf.h>
37 #include <sys/dnode.h>
38 #include <sys/zfs_context.h>
39 #include <sys/dmu_objset.h>
40 #include <sys/dmu_traverse.h>
41 #include <sys/dsl_dataset.h>
42 #include <sys/dsl_dir.h>
43 #include <sys/dsl_prop.h>
44 #include <sys/dsl_pool.h>
45 #include <sys/dsl_synctask.h>
46 #include <sys/spa_impl.h>
47 #include <sys/zfs_ioctl.h>
48 #include <sys/zap.h>
49 #include <sys/zio_checksum.h>
50 #include <sys/zfs_znode.h>
51 #include <zfs_fletcher.h>
52 #include <sys/avl.h>
53 #include <sys/ddt.h>
54 #include <sys/zfs_onexit.h>
55 #include <sys/dmu_send.h>
56 #include <sys/dmu_recv.h>
57 #include <sys/dsl_destroy.h>
58 #include <sys/blkptr.h>
59 #include <sys/dsl_bookmark.h>
60 #include <sys/zfeature.h>
61 #include <sys/bqueue.h>
62 #include <sys/zvol.h>
63 #include <sys/policy.h>
64 #include <sys/objlist.h>
65 #ifdef _KERNEL
66 #include <sys/zfs_vfsops.h>
67 #endif
68 
69 /* Set this tunable to TRUE to replace corrupt data with 0x2f5baddb10c */
70 static int zfs_send_corrupt_data = B_FALSE;
71 /*
72  * This tunable controls the amount of data (measured in bytes) that will be
73  * prefetched by zfs send.  If the main thread is blocking on reads that haven't
74  * completed, this variable might need to be increased.  If instead the main
75  * thread is issuing new reads because the prefetches have fallen out of the
76  * cache, this may need to be decreased.
77  */
78 static uint_t zfs_send_queue_length = SPA_MAXBLOCKSIZE;
79 /*
80  * This tunable controls the length of the queues that zfs send worker threads
81  * use to communicate.  If the send_main_thread is blocking on these queues,
82  * this variable may need to be increased.  If there is a significant slowdown
83  * at the start of a send as these threads consume all the available IO
84  * resources, this variable may need to be decreased.
85  */
86 static uint_t zfs_send_no_prefetch_queue_length = 1024 * 1024;
87 /*
88  * These tunables control the fill fraction of the queues by zfs send.  The fill
89  * fraction controls the frequency with which threads have to be cv_signaled.
90  * If a lot of cpu time is being spent on cv_signal, then these should be tuned
91  * down.  If the queues empty before the signalled thread can catch up, then
92  * these should be tuned up.
93  */
94 static uint_t zfs_send_queue_ff = 20;
95 static uint_t zfs_send_no_prefetch_queue_ff = 20;
96 
97 /*
98  * Use this to override the recordsize calculation for fast zfs send estimates.
99  */
100 static uint_t zfs_override_estimate_recordsize = 0;
101 
102 /* Set this tunable to FALSE to disable setting of DRR_FLAG_FREERECORDS */
103 static const boolean_t zfs_send_set_freerecords_bit = B_TRUE;
104 
105 /* Set this tunable to FALSE is disable sending unmodified spill blocks. */
106 static int zfs_send_unmodified_spill_blocks = B_TRUE;
107 
108 static inline boolean_t
109 overflow_multiply(uint64_t a, uint64_t b, uint64_t *c)
110 {
111 	uint64_t temp = a * b;
112 	if (b != 0 && temp / b != a)
113 		return (B_FALSE);
114 	*c = temp;
115 	return (B_TRUE);
116 }
117 
118 struct send_thread_arg {
119 	bqueue_t	q;
120 	objset_t	*os;		/* Objset to traverse */
121 	uint64_t	fromtxg;	/* Traverse from this txg */
122 	int		flags;		/* flags to pass to traverse_dataset */
123 	int		error_code;
124 	boolean_t	cancel;
125 	zbookmark_phys_t resume;
126 	uint64_t	*num_blocks_visited;
127 };
128 
129 struct redact_list_thread_arg {
130 	boolean_t		cancel;
131 	bqueue_t		q;
132 	zbookmark_phys_t	resume;
133 	redaction_list_t	*rl;
134 	boolean_t		mark_redact;
135 	int			error_code;
136 	uint64_t		*num_blocks_visited;
137 };
138 
139 struct send_merge_thread_arg {
140 	bqueue_t			q;
141 	objset_t			*os;
142 	struct redact_list_thread_arg	*from_arg;
143 	struct send_thread_arg		*to_arg;
144 	struct redact_list_thread_arg	*redact_arg;
145 	int				error;
146 	boolean_t			cancel;
147 };
148 
149 struct send_range {
150 	boolean_t		eos_marker; /* Marks the end of the stream */
151 	uint64_t		object;
152 	uint64_t		start_blkid;
153 	uint64_t		end_blkid;
154 	bqueue_node_t		ln;
155 	enum type {DATA, HOLE, OBJECT, OBJECT_RANGE, REDACT,
156 	    PREVIOUSLY_REDACTED} type;
157 	union {
158 		struct srd {
159 			dmu_object_type_t	obj_type;
160 			uint32_t		datablksz; // logical size
161 			uint32_t		datasz; // payload size
162 			blkptr_t		bp;
163 			arc_buf_t		*abuf;
164 			abd_t			*abd;
165 			kmutex_t		lock;
166 			kcondvar_t		cv;
167 			boolean_t		io_outstanding;
168 			boolean_t		io_compressed;
169 			int			io_err;
170 		} data;
171 		struct srh {
172 			uint32_t		datablksz;
173 		} hole;
174 		struct sro {
175 			/*
176 			 * This is a pointer because embedding it in the
177 			 * struct causes these structures to be massively larger
178 			 * for all range types; this makes the code much less
179 			 * memory efficient.
180 			 */
181 			dnode_phys_t		*dnp;
182 			blkptr_t		bp;
183 		} object;
184 		struct srr {
185 			uint32_t		datablksz;
186 		} redact;
187 		struct sror {
188 			blkptr_t		bp;
189 		} object_range;
190 	} sru;
191 };
192 
193 /*
194  * The list of data whose inclusion in a send stream can be pending from
195  * one call to backup_cb to another.  Multiple calls to dump_free(),
196  * dump_freeobjects(), and dump_redact() can be aggregated into a single
197  * DRR_FREE, DRR_FREEOBJECTS, or DRR_REDACT replay record.
198  */
199 typedef enum {
200 	PENDING_NONE,
201 	PENDING_FREE,
202 	PENDING_FREEOBJECTS,
203 	PENDING_REDACT
204 } dmu_pendop_t;
205 
206 typedef struct dmu_send_cookie {
207 	dmu_replay_record_t *dsc_drr;
208 	dmu_send_outparams_t *dsc_dso;
209 	offset_t *dsc_off;
210 	objset_t *dsc_os;
211 	zio_cksum_t dsc_zc;
212 	uint64_t dsc_toguid;
213 	uint64_t dsc_fromtxg;
214 	int dsc_err;
215 	dmu_pendop_t dsc_pending_op;
216 	uint64_t dsc_featureflags;
217 	uint64_t dsc_last_data_object;
218 	uint64_t dsc_last_data_offset;
219 	uint64_t dsc_resume_object;
220 	uint64_t dsc_resume_offset;
221 	boolean_t dsc_sent_begin;
222 	boolean_t dsc_sent_end;
223 } dmu_send_cookie_t;
224 
225 static int do_dump(dmu_send_cookie_t *dscp, struct send_range *range);
226 
227 static void
228 range_free(struct send_range *range)
229 {
230 	if (range->type == OBJECT) {
231 		size_t size = sizeof (dnode_phys_t) *
232 		    (range->sru.object.dnp->dn_extra_slots + 1);
233 		kmem_free(range->sru.object.dnp, size);
234 	} else if (range->type == DATA) {
235 		mutex_enter(&range->sru.data.lock);
236 		while (range->sru.data.io_outstanding)
237 			cv_wait(&range->sru.data.cv, &range->sru.data.lock);
238 		if (range->sru.data.abd != NULL)
239 			abd_free(range->sru.data.abd);
240 		if (range->sru.data.abuf != NULL) {
241 			arc_buf_destroy(range->sru.data.abuf,
242 			    &range->sru.data.abuf);
243 		}
244 		mutex_exit(&range->sru.data.lock);
245 
246 		cv_destroy(&range->sru.data.cv);
247 		mutex_destroy(&range->sru.data.lock);
248 	}
249 	kmem_free(range, sizeof (*range));
250 }
251 
252 /*
253  * For all record types except BEGIN, fill in the checksum (overlaid in
254  * drr_u.drr_checksum.drr_checksum).  The checksum verifies everything
255  * up to the start of the checksum itself.
256  */
257 static int
258 dump_record(dmu_send_cookie_t *dscp, void *payload, int payload_len)
259 {
260 	dmu_send_outparams_t *dso = dscp->dsc_dso;
261 	ASSERT3U(offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
262 	    ==, sizeof (dmu_replay_record_t) - sizeof (zio_cksum_t));
263 	(void) fletcher_4_incremental_native(dscp->dsc_drr,
264 	    offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
265 	    &dscp->dsc_zc);
266 	if (dscp->dsc_drr->drr_type == DRR_BEGIN) {
267 		dscp->dsc_sent_begin = B_TRUE;
268 	} else {
269 		ASSERT(ZIO_CHECKSUM_IS_ZERO(&dscp->dsc_drr->drr_u.
270 		    drr_checksum.drr_checksum));
271 		dscp->dsc_drr->drr_u.drr_checksum.drr_checksum = dscp->dsc_zc;
272 	}
273 	if (dscp->dsc_drr->drr_type == DRR_END) {
274 		dscp->dsc_sent_end = B_TRUE;
275 	}
276 	(void) fletcher_4_incremental_native(&dscp->dsc_drr->
277 	    drr_u.drr_checksum.drr_checksum,
278 	    sizeof (zio_cksum_t), &dscp->dsc_zc);
279 	*dscp->dsc_off += sizeof (dmu_replay_record_t);
280 	dscp->dsc_err = dso->dso_outfunc(dscp->dsc_os, dscp->dsc_drr,
281 	    sizeof (dmu_replay_record_t), dso->dso_arg);
282 	if (dscp->dsc_err != 0)
283 		return (SET_ERROR(EINTR));
284 	if (payload_len != 0) {
285 		*dscp->dsc_off += payload_len;
286 		/*
287 		 * payload is null when dso_dryrun == B_TRUE (i.e. when we're
288 		 * doing a send size calculation)
289 		 */
290 		if (payload != NULL) {
291 			(void) fletcher_4_incremental_native(
292 			    payload, payload_len, &dscp->dsc_zc);
293 		}
294 
295 		/*
296 		 * The code does not rely on this (len being a multiple of 8).
297 		 * We keep this assertion because of the corresponding assertion
298 		 * in receive_read().  Keeping this assertion ensures that we do
299 		 * not inadvertently break backwards compatibility (causing the
300 		 * assertion in receive_read() to trigger on old software).
301 		 *
302 		 * Raw sends cannot be received on old software, and so can
303 		 * bypass this assertion.
304 		 */
305 
306 		ASSERT((payload_len % 8 == 0) ||
307 		    (dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW));
308 
309 		dscp->dsc_err = dso->dso_outfunc(dscp->dsc_os, payload,
310 		    payload_len, dso->dso_arg);
311 		if (dscp->dsc_err != 0)
312 			return (SET_ERROR(EINTR));
313 	}
314 	return (0);
315 }
316 
317 /*
318  * Fill in the drr_free struct, or perform aggregation if the previous record is
319  * also a free record, and the two are adjacent.
320  *
321  * Note that we send free records even for a full send, because we want to be
322  * able to receive a full send as a clone, which requires a list of all the free
323  * and freeobject records that were generated on the source.
324  */
325 static int
326 dump_free(dmu_send_cookie_t *dscp, uint64_t object, uint64_t offset,
327     uint64_t length)
328 {
329 	struct drr_free *drrf = &(dscp->dsc_drr->drr_u.drr_free);
330 
331 	/*
332 	 * When we receive a free record, dbuf_free_range() assumes
333 	 * that the receiving system doesn't have any dbufs in the range
334 	 * being freed.  This is always true because there is a one-record
335 	 * constraint: we only send one WRITE record for any given
336 	 * object,offset.  We know that the one-record constraint is
337 	 * true because we always send data in increasing order by
338 	 * object,offset.
339 	 *
340 	 * If the increasing-order constraint ever changes, we should find
341 	 * another way to assert that the one-record constraint is still
342 	 * satisfied.
343 	 */
344 	ASSERT(object > dscp->dsc_last_data_object ||
345 	    (object == dscp->dsc_last_data_object &&
346 	    offset > dscp->dsc_last_data_offset));
347 
348 	/*
349 	 * If there is a pending op, but it's not PENDING_FREE, push it out,
350 	 * since free block aggregation can only be done for blocks of the
351 	 * same type (i.e., DRR_FREE records can only be aggregated with
352 	 * other DRR_FREE records.  DRR_FREEOBJECTS records can only be
353 	 * aggregated with other DRR_FREEOBJECTS records).
354 	 */
355 	if (dscp->dsc_pending_op != PENDING_NONE &&
356 	    dscp->dsc_pending_op != PENDING_FREE) {
357 		if (dump_record(dscp, NULL, 0) != 0)
358 			return (SET_ERROR(EINTR));
359 		dscp->dsc_pending_op = PENDING_NONE;
360 	}
361 
362 	if (dscp->dsc_pending_op == PENDING_FREE) {
363 		/*
364 		 * Check to see whether this free block can be aggregated
365 		 * with pending one.
366 		 */
367 		if (drrf->drr_object == object && drrf->drr_offset +
368 		    drrf->drr_length == offset) {
369 			if (offset + length < offset || length == UINT64_MAX)
370 				drrf->drr_length = UINT64_MAX;
371 			else
372 				drrf->drr_length += length;
373 			return (0);
374 		} else {
375 			/* not a continuation.  Push out pending record */
376 			if (dump_record(dscp, NULL, 0) != 0)
377 				return (SET_ERROR(EINTR));
378 			dscp->dsc_pending_op = PENDING_NONE;
379 		}
380 	}
381 	/* create a FREE record and make it pending */
382 	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
383 	dscp->dsc_drr->drr_type = DRR_FREE;
384 	drrf->drr_object = object;
385 	drrf->drr_offset = offset;
386 	if (offset + length < offset)
387 		drrf->drr_length = DMU_OBJECT_END;
388 	else
389 		drrf->drr_length = length;
390 	drrf->drr_toguid = dscp->dsc_toguid;
391 	if (length == DMU_OBJECT_END) {
392 		if (dump_record(dscp, NULL, 0) != 0)
393 			return (SET_ERROR(EINTR));
394 	} else {
395 		dscp->dsc_pending_op = PENDING_FREE;
396 	}
397 
398 	return (0);
399 }
400 
401 /*
402  * Fill in the drr_redact struct, or perform aggregation if the previous record
403  * is also a redaction record, and the two are adjacent.
404  */
405 static int
406 dump_redact(dmu_send_cookie_t *dscp, uint64_t object, uint64_t offset,
407     uint64_t length)
408 {
409 	struct drr_redact *drrr = &dscp->dsc_drr->drr_u.drr_redact;
410 
411 	/*
412 	 * If there is a pending op, but it's not PENDING_REDACT, push it out,
413 	 * since free block aggregation can only be done for blocks of the
414 	 * same type (i.e., DRR_REDACT records can only be aggregated with
415 	 * other DRR_REDACT records).
416 	 */
417 	if (dscp->dsc_pending_op != PENDING_NONE &&
418 	    dscp->dsc_pending_op != PENDING_REDACT) {
419 		if (dump_record(dscp, NULL, 0) != 0)
420 			return (SET_ERROR(EINTR));
421 		dscp->dsc_pending_op = PENDING_NONE;
422 	}
423 
424 	if (dscp->dsc_pending_op == PENDING_REDACT) {
425 		/*
426 		 * Check to see whether this redacted block can be aggregated
427 		 * with pending one.
428 		 */
429 		if (drrr->drr_object == object && drrr->drr_offset +
430 		    drrr->drr_length == offset) {
431 			drrr->drr_length += length;
432 			return (0);
433 		} else {
434 			/* not a continuation.  Push out pending record */
435 			if (dump_record(dscp, NULL, 0) != 0)
436 				return (SET_ERROR(EINTR));
437 			dscp->dsc_pending_op = PENDING_NONE;
438 		}
439 	}
440 	/* create a REDACT record and make it pending */
441 	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
442 	dscp->dsc_drr->drr_type = DRR_REDACT;
443 	drrr->drr_object = object;
444 	drrr->drr_offset = offset;
445 	drrr->drr_length = length;
446 	drrr->drr_toguid = dscp->dsc_toguid;
447 	dscp->dsc_pending_op = PENDING_REDACT;
448 
449 	return (0);
450 }
451 
452 static int
453 dmu_dump_write(dmu_send_cookie_t *dscp, dmu_object_type_t type, uint64_t object,
454     uint64_t offset, int lsize, int psize, const blkptr_t *bp,
455     boolean_t io_compressed, void *data)
456 {
457 	uint64_t payload_size;
458 	boolean_t raw = (dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW);
459 	struct drr_write *drrw = &(dscp->dsc_drr->drr_u.drr_write);
460 
461 	/*
462 	 * We send data in increasing object, offset order.
463 	 * See comment in dump_free() for details.
464 	 */
465 	ASSERT(object > dscp->dsc_last_data_object ||
466 	    (object == dscp->dsc_last_data_object &&
467 	    offset > dscp->dsc_last_data_offset));
468 	dscp->dsc_last_data_object = object;
469 	dscp->dsc_last_data_offset = offset + lsize - 1;
470 
471 	/*
472 	 * If there is any kind of pending aggregation (currently either
473 	 * a grouping of free objects or free blocks), push it out to
474 	 * the stream, since aggregation can't be done across operations
475 	 * of different types.
476 	 */
477 	if (dscp->dsc_pending_op != PENDING_NONE) {
478 		if (dump_record(dscp, NULL, 0) != 0)
479 			return (SET_ERROR(EINTR));
480 		dscp->dsc_pending_op = PENDING_NONE;
481 	}
482 	/* write a WRITE record */
483 	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
484 	dscp->dsc_drr->drr_type = DRR_WRITE;
485 	drrw->drr_object = object;
486 	drrw->drr_type = type;
487 	drrw->drr_offset = offset;
488 	drrw->drr_toguid = dscp->dsc_toguid;
489 	drrw->drr_logical_size = lsize;
490 
491 	/* only set the compression fields if the buf is compressed or raw */
492 	boolean_t compressed =
493 	    (bp != NULL ? BP_GET_COMPRESS(bp) != ZIO_COMPRESS_OFF &&
494 	    io_compressed : lsize != psize);
495 	if (raw || compressed) {
496 		ASSERT(bp != NULL);
497 		ASSERT(raw || dscp->dsc_featureflags &
498 		    DMU_BACKUP_FEATURE_COMPRESSED);
499 		ASSERT(!BP_IS_EMBEDDED(bp));
500 		ASSERT3S(psize, >, 0);
501 
502 		if (raw) {
503 			ASSERT(BP_IS_PROTECTED(bp));
504 
505 			/*
506 			 * This is a raw protected block so we need to pass
507 			 * along everything the receiving side will need to
508 			 * interpret this block, including the byteswap, salt,
509 			 * IV, and MAC.
510 			 */
511 			if (BP_SHOULD_BYTESWAP(bp))
512 				drrw->drr_flags |= DRR_RAW_BYTESWAP;
513 			zio_crypt_decode_params_bp(bp, drrw->drr_salt,
514 			    drrw->drr_iv);
515 			zio_crypt_decode_mac_bp(bp, drrw->drr_mac);
516 		} else {
517 			/* this is a compressed block */
518 			ASSERT(dscp->dsc_featureflags &
519 			    DMU_BACKUP_FEATURE_COMPRESSED);
520 			ASSERT(!BP_SHOULD_BYTESWAP(bp));
521 			ASSERT(!DMU_OT_IS_METADATA(BP_GET_TYPE(bp)));
522 			ASSERT3U(BP_GET_COMPRESS(bp), !=, ZIO_COMPRESS_OFF);
523 			ASSERT3S(lsize, >=, psize);
524 		}
525 
526 		/* set fields common to compressed and raw sends */
527 		drrw->drr_compressiontype = BP_GET_COMPRESS(bp);
528 		drrw->drr_compressed_size = psize;
529 		payload_size = drrw->drr_compressed_size;
530 	} else {
531 		payload_size = drrw->drr_logical_size;
532 	}
533 
534 	if (bp == NULL || BP_IS_EMBEDDED(bp) || (BP_IS_PROTECTED(bp) && !raw)) {
535 		/*
536 		 * There's no pre-computed checksum for partial-block writes,
537 		 * embedded BP's, or encrypted BP's that are being sent as
538 		 * plaintext, so (like fletcher4-checksummed blocks) userland
539 		 * will have to compute a dedup-capable checksum itself.
540 		 */
541 		drrw->drr_checksumtype = ZIO_CHECKSUM_OFF;
542 	} else {
543 		drrw->drr_checksumtype = BP_GET_CHECKSUM(bp);
544 		if (zio_checksum_table[drrw->drr_checksumtype].ci_flags &
545 		    ZCHECKSUM_FLAG_DEDUP)
546 			drrw->drr_flags |= DRR_CHECKSUM_DEDUP;
547 		DDK_SET_LSIZE(&drrw->drr_key, BP_GET_LSIZE(bp));
548 		DDK_SET_PSIZE(&drrw->drr_key, BP_GET_PSIZE(bp));
549 		DDK_SET_COMPRESS(&drrw->drr_key, BP_GET_COMPRESS(bp));
550 		DDK_SET_CRYPT(&drrw->drr_key, BP_IS_PROTECTED(bp));
551 		drrw->drr_key.ddk_cksum = bp->blk_cksum;
552 	}
553 
554 	if (dump_record(dscp, data, payload_size) != 0)
555 		return (SET_ERROR(EINTR));
556 	return (0);
557 }
558 
559 static int
560 dump_write_embedded(dmu_send_cookie_t *dscp, uint64_t object, uint64_t offset,
561     int blksz, const blkptr_t *bp)
562 {
563 	char buf[BPE_PAYLOAD_SIZE];
564 	struct drr_write_embedded *drrw =
565 	    &(dscp->dsc_drr->drr_u.drr_write_embedded);
566 
567 	if (dscp->dsc_pending_op != PENDING_NONE) {
568 		if (dump_record(dscp, NULL, 0) != 0)
569 			return (SET_ERROR(EINTR));
570 		dscp->dsc_pending_op = PENDING_NONE;
571 	}
572 
573 	ASSERT(BP_IS_EMBEDDED(bp));
574 
575 	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
576 	dscp->dsc_drr->drr_type = DRR_WRITE_EMBEDDED;
577 	drrw->drr_object = object;
578 	drrw->drr_offset = offset;
579 	drrw->drr_length = blksz;
580 	drrw->drr_toguid = dscp->dsc_toguid;
581 	drrw->drr_compression = BP_GET_COMPRESS(bp);
582 	drrw->drr_etype = BPE_GET_ETYPE(bp);
583 	drrw->drr_lsize = BPE_GET_LSIZE(bp);
584 	drrw->drr_psize = BPE_GET_PSIZE(bp);
585 
586 	decode_embedded_bp_compressed(bp, buf);
587 
588 	uint32_t psize = drrw->drr_psize;
589 	uint32_t rsize = P2ROUNDUP(psize, 8);
590 
591 	if (psize != rsize)
592 		memset(buf + psize, 0, rsize - psize);
593 
594 	if (dump_record(dscp, buf, rsize) != 0)
595 		return (SET_ERROR(EINTR));
596 	return (0);
597 }
598 
599 static int
600 dump_spill(dmu_send_cookie_t *dscp, const blkptr_t *bp, uint64_t object,
601     void *data)
602 {
603 	struct drr_spill *drrs = &(dscp->dsc_drr->drr_u.drr_spill);
604 	uint64_t blksz = BP_GET_LSIZE(bp);
605 	uint64_t payload_size = blksz;
606 
607 	if (dscp->dsc_pending_op != PENDING_NONE) {
608 		if (dump_record(dscp, NULL, 0) != 0)
609 			return (SET_ERROR(EINTR));
610 		dscp->dsc_pending_op = PENDING_NONE;
611 	}
612 
613 	/* write a SPILL record */
614 	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
615 	dscp->dsc_drr->drr_type = DRR_SPILL;
616 	drrs->drr_object = object;
617 	drrs->drr_length = blksz;
618 	drrs->drr_toguid = dscp->dsc_toguid;
619 
620 	/* See comment in dump_dnode() for full details */
621 	if (zfs_send_unmodified_spill_blocks &&
622 	    (bp->blk_birth <= dscp->dsc_fromtxg)) {
623 		drrs->drr_flags |= DRR_SPILL_UNMODIFIED;
624 	}
625 
626 	/* handle raw send fields */
627 	if (dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW) {
628 		ASSERT(BP_IS_PROTECTED(bp));
629 
630 		if (BP_SHOULD_BYTESWAP(bp))
631 			drrs->drr_flags |= DRR_RAW_BYTESWAP;
632 		drrs->drr_compressiontype = BP_GET_COMPRESS(bp);
633 		drrs->drr_compressed_size = BP_GET_PSIZE(bp);
634 		zio_crypt_decode_params_bp(bp, drrs->drr_salt, drrs->drr_iv);
635 		zio_crypt_decode_mac_bp(bp, drrs->drr_mac);
636 		payload_size = drrs->drr_compressed_size;
637 	}
638 
639 	if (dump_record(dscp, data, payload_size) != 0)
640 		return (SET_ERROR(EINTR));
641 	return (0);
642 }
643 
644 static int
645 dump_freeobjects(dmu_send_cookie_t *dscp, uint64_t firstobj, uint64_t numobjs)
646 {
647 	struct drr_freeobjects *drrfo = &(dscp->dsc_drr->drr_u.drr_freeobjects);
648 	uint64_t maxobj = DNODES_PER_BLOCK *
649 	    (DMU_META_DNODE(dscp->dsc_os)->dn_maxblkid + 1);
650 
651 	/*
652 	 * ZoL < 0.7 does not handle large FREEOBJECTS records correctly,
653 	 * leading to zfs recv never completing. to avoid this issue, don't
654 	 * send FREEOBJECTS records for object IDs which cannot exist on the
655 	 * receiving side.
656 	 */
657 	if (maxobj > 0) {
658 		if (maxobj <= firstobj)
659 			return (0);
660 
661 		if (maxobj < firstobj + numobjs)
662 			numobjs = maxobj - firstobj;
663 	}
664 
665 	/*
666 	 * If there is a pending op, but it's not PENDING_FREEOBJECTS,
667 	 * push it out, since free block aggregation can only be done for
668 	 * blocks of the same type (i.e., DRR_FREE records can only be
669 	 * aggregated with other DRR_FREE records.  DRR_FREEOBJECTS records
670 	 * can only be aggregated with other DRR_FREEOBJECTS records).
671 	 */
672 	if (dscp->dsc_pending_op != PENDING_NONE &&
673 	    dscp->dsc_pending_op != PENDING_FREEOBJECTS) {
674 		if (dump_record(dscp, NULL, 0) != 0)
675 			return (SET_ERROR(EINTR));
676 		dscp->dsc_pending_op = PENDING_NONE;
677 	}
678 
679 	if (dscp->dsc_pending_op == PENDING_FREEOBJECTS) {
680 		/*
681 		 * See whether this free object array can be aggregated
682 		 * with pending one
683 		 */
684 		if (drrfo->drr_firstobj + drrfo->drr_numobjs == firstobj) {
685 			drrfo->drr_numobjs += numobjs;
686 			return (0);
687 		} else {
688 			/* can't be aggregated.  Push out pending record */
689 			if (dump_record(dscp, NULL, 0) != 0)
690 				return (SET_ERROR(EINTR));
691 			dscp->dsc_pending_op = PENDING_NONE;
692 		}
693 	}
694 
695 	/* write a FREEOBJECTS record */
696 	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
697 	dscp->dsc_drr->drr_type = DRR_FREEOBJECTS;
698 	drrfo->drr_firstobj = firstobj;
699 	drrfo->drr_numobjs = numobjs;
700 	drrfo->drr_toguid = dscp->dsc_toguid;
701 
702 	dscp->dsc_pending_op = PENDING_FREEOBJECTS;
703 
704 	return (0);
705 }
706 
707 static int
708 dump_dnode(dmu_send_cookie_t *dscp, const blkptr_t *bp, uint64_t object,
709     dnode_phys_t *dnp)
710 {
711 	struct drr_object *drro = &(dscp->dsc_drr->drr_u.drr_object);
712 	int bonuslen;
713 
714 	if (object < dscp->dsc_resume_object) {
715 		/*
716 		 * Note: when resuming, we will visit all the dnodes in
717 		 * the block of dnodes that we are resuming from.  In
718 		 * this case it's unnecessary to send the dnodes prior to
719 		 * the one we are resuming from.  We should be at most one
720 		 * block's worth of dnodes behind the resume point.
721 		 */
722 		ASSERT3U(dscp->dsc_resume_object - object, <,
723 		    1 << (DNODE_BLOCK_SHIFT - DNODE_SHIFT));
724 		return (0);
725 	}
726 
727 	if (dnp == NULL || dnp->dn_type == DMU_OT_NONE)
728 		return (dump_freeobjects(dscp, object, 1));
729 
730 	if (dscp->dsc_pending_op != PENDING_NONE) {
731 		if (dump_record(dscp, NULL, 0) != 0)
732 			return (SET_ERROR(EINTR));
733 		dscp->dsc_pending_op = PENDING_NONE;
734 	}
735 
736 	/* write an OBJECT record */
737 	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
738 	dscp->dsc_drr->drr_type = DRR_OBJECT;
739 	drro->drr_object = object;
740 	drro->drr_type = dnp->dn_type;
741 	drro->drr_bonustype = dnp->dn_bonustype;
742 	drro->drr_blksz = dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT;
743 	drro->drr_bonuslen = dnp->dn_bonuslen;
744 	drro->drr_dn_slots = dnp->dn_extra_slots + 1;
745 	drro->drr_checksumtype = dnp->dn_checksum;
746 	drro->drr_compress = dnp->dn_compress;
747 	drro->drr_toguid = dscp->dsc_toguid;
748 
749 	if (!(dscp->dsc_featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS) &&
750 	    drro->drr_blksz > SPA_OLD_MAXBLOCKSIZE)
751 		drro->drr_blksz = SPA_OLD_MAXBLOCKSIZE;
752 
753 	bonuslen = P2ROUNDUP(dnp->dn_bonuslen, 8);
754 
755 	if ((dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW)) {
756 		ASSERT(BP_IS_ENCRYPTED(bp));
757 
758 		if (BP_SHOULD_BYTESWAP(bp))
759 			drro->drr_flags |= DRR_RAW_BYTESWAP;
760 
761 		/* needed for reconstructing dnp on recv side */
762 		drro->drr_maxblkid = dnp->dn_maxblkid;
763 		drro->drr_indblkshift = dnp->dn_indblkshift;
764 		drro->drr_nlevels = dnp->dn_nlevels;
765 		drro->drr_nblkptr = dnp->dn_nblkptr;
766 
767 		/*
768 		 * Since we encrypt the entire bonus area, the (raw) part
769 		 * beyond the bonuslen is actually nonzero, so we need
770 		 * to send it.
771 		 */
772 		if (bonuslen != 0) {
773 			if (drro->drr_bonuslen > DN_MAX_BONUS_LEN(dnp))
774 				return (SET_ERROR(EINVAL));
775 			drro->drr_raw_bonuslen = DN_MAX_BONUS_LEN(dnp);
776 			bonuslen = drro->drr_raw_bonuslen;
777 		}
778 	}
779 
780 	/*
781 	 * DRR_OBJECT_SPILL is set for every dnode which references a
782 	 * spill block.	 This allows the receiving pool to definitively
783 	 * determine when a spill block should be kept or freed.
784 	 */
785 	if (dnp->dn_flags & DNODE_FLAG_SPILL_BLKPTR)
786 		drro->drr_flags |= DRR_OBJECT_SPILL;
787 
788 	if (dump_record(dscp, DN_BONUS(dnp), bonuslen) != 0)
789 		return (SET_ERROR(EINTR));
790 
791 	/* Free anything past the end of the file. */
792 	if (dump_free(dscp, object, (dnp->dn_maxblkid + 1) *
793 	    (dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT), DMU_OBJECT_END) != 0)
794 		return (SET_ERROR(EINTR));
795 
796 	/*
797 	 * Send DRR_SPILL records for unmodified spill blocks.	This is useful
798 	 * because changing certain attributes of the object (e.g. blocksize)
799 	 * can cause old versions of ZFS to incorrectly remove a spill block.
800 	 * Including these records in the stream forces an up to date version
801 	 * to always be written ensuring they're never lost.  Current versions
802 	 * of the code which understand the DRR_FLAG_SPILL_BLOCK feature can
803 	 * ignore these unmodified spill blocks.
804 	 */
805 	if (zfs_send_unmodified_spill_blocks &&
806 	    (dnp->dn_flags & DNODE_FLAG_SPILL_BLKPTR) &&
807 	    (DN_SPILL_BLKPTR(dnp)->blk_birth <= dscp->dsc_fromtxg)) {
808 		struct send_range record;
809 		blkptr_t *bp = DN_SPILL_BLKPTR(dnp);
810 
811 		memset(&record, 0, sizeof (struct send_range));
812 		record.type = DATA;
813 		record.object = object;
814 		record.eos_marker = B_FALSE;
815 		record.start_blkid = DMU_SPILL_BLKID;
816 		record.end_blkid = record.start_blkid + 1;
817 		record.sru.data.bp = *bp;
818 		record.sru.data.obj_type = dnp->dn_type;
819 		record.sru.data.datablksz = BP_GET_LSIZE(bp);
820 
821 		if (do_dump(dscp, &record) != 0)
822 			return (SET_ERROR(EINTR));
823 	}
824 
825 	if (dscp->dsc_err != 0)
826 		return (SET_ERROR(EINTR));
827 
828 	return (0);
829 }
830 
831 static int
832 dump_object_range(dmu_send_cookie_t *dscp, const blkptr_t *bp,
833     uint64_t firstobj, uint64_t numslots)
834 {
835 	struct drr_object_range *drror =
836 	    &(dscp->dsc_drr->drr_u.drr_object_range);
837 
838 	/* we only use this record type for raw sends */
839 	ASSERT(BP_IS_PROTECTED(bp));
840 	ASSERT(dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW);
841 	ASSERT3U(BP_GET_COMPRESS(bp), ==, ZIO_COMPRESS_OFF);
842 	ASSERT3U(BP_GET_TYPE(bp), ==, DMU_OT_DNODE);
843 	ASSERT0(BP_GET_LEVEL(bp));
844 
845 	if (dscp->dsc_pending_op != PENDING_NONE) {
846 		if (dump_record(dscp, NULL, 0) != 0)
847 			return (SET_ERROR(EINTR));
848 		dscp->dsc_pending_op = PENDING_NONE;
849 	}
850 
851 	memset(dscp->dsc_drr, 0, sizeof (dmu_replay_record_t));
852 	dscp->dsc_drr->drr_type = DRR_OBJECT_RANGE;
853 	drror->drr_firstobj = firstobj;
854 	drror->drr_numslots = numslots;
855 	drror->drr_toguid = dscp->dsc_toguid;
856 	if (BP_SHOULD_BYTESWAP(bp))
857 		drror->drr_flags |= DRR_RAW_BYTESWAP;
858 	zio_crypt_decode_params_bp(bp, drror->drr_salt, drror->drr_iv);
859 	zio_crypt_decode_mac_bp(bp, drror->drr_mac);
860 
861 	if (dump_record(dscp, NULL, 0) != 0)
862 		return (SET_ERROR(EINTR));
863 	return (0);
864 }
865 
866 static boolean_t
867 send_do_embed(const blkptr_t *bp, uint64_t featureflags)
868 {
869 	if (!BP_IS_EMBEDDED(bp))
870 		return (B_FALSE);
871 
872 	/*
873 	 * Compression function must be legacy, or explicitly enabled.
874 	 */
875 	if ((BP_GET_COMPRESS(bp) >= ZIO_COMPRESS_LEGACY_FUNCTIONS &&
876 	    !(featureflags & DMU_BACKUP_FEATURE_LZ4)))
877 		return (B_FALSE);
878 
879 	/*
880 	 * If we have not set the ZSTD feature flag, we can't send ZSTD
881 	 * compressed embedded blocks, as the receiver may not support them.
882 	 */
883 	if ((BP_GET_COMPRESS(bp) == ZIO_COMPRESS_ZSTD &&
884 	    !(featureflags & DMU_BACKUP_FEATURE_ZSTD)))
885 		return (B_FALSE);
886 
887 	/*
888 	 * Embed type must be explicitly enabled.
889 	 */
890 	switch (BPE_GET_ETYPE(bp)) {
891 	case BP_EMBEDDED_TYPE_DATA:
892 		if (featureflags & DMU_BACKUP_FEATURE_EMBED_DATA)
893 			return (B_TRUE);
894 		break;
895 	default:
896 		return (B_FALSE);
897 	}
898 	return (B_FALSE);
899 }
900 
901 /*
902  * This function actually handles figuring out what kind of record needs to be
903  * dumped, and calling the appropriate helper function.  In most cases,
904  * the data has already been read by send_reader_thread().
905  */
906 static int
907 do_dump(dmu_send_cookie_t *dscp, struct send_range *range)
908 {
909 	int err = 0;
910 	switch (range->type) {
911 	case OBJECT:
912 		err = dump_dnode(dscp, &range->sru.object.bp, range->object,
913 		    range->sru.object.dnp);
914 		return (err);
915 	case OBJECT_RANGE: {
916 		ASSERT3U(range->start_blkid + 1, ==, range->end_blkid);
917 		if (!(dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW)) {
918 			return (0);
919 		}
920 		uint64_t epb = BP_GET_LSIZE(&range->sru.object_range.bp) >>
921 		    DNODE_SHIFT;
922 		uint64_t firstobj = range->start_blkid * epb;
923 		err = dump_object_range(dscp, &range->sru.object_range.bp,
924 		    firstobj, epb);
925 		break;
926 	}
927 	case REDACT: {
928 		struct srr *srrp = &range->sru.redact;
929 		err = dump_redact(dscp, range->object, range->start_blkid *
930 		    srrp->datablksz, (range->end_blkid - range->start_blkid) *
931 		    srrp->datablksz);
932 		return (err);
933 	}
934 	case DATA: {
935 		struct srd *srdp = &range->sru.data;
936 		blkptr_t *bp = &srdp->bp;
937 		spa_t *spa =
938 		    dmu_objset_spa(dscp->dsc_os);
939 
940 		ASSERT3U(srdp->datablksz, ==, BP_GET_LSIZE(bp));
941 		ASSERT3U(range->start_blkid + 1, ==, range->end_blkid);
942 		if (BP_GET_TYPE(bp) == DMU_OT_SA) {
943 			arc_flags_t aflags = ARC_FLAG_WAIT;
944 			zio_flag_t zioflags = ZIO_FLAG_CANFAIL;
945 
946 			if (dscp->dsc_featureflags & DMU_BACKUP_FEATURE_RAW) {
947 				ASSERT(BP_IS_PROTECTED(bp));
948 				zioflags |= ZIO_FLAG_RAW;
949 			}
950 
951 			zbookmark_phys_t zb;
952 			ASSERT3U(range->start_blkid, ==, DMU_SPILL_BLKID);
953 			zb.zb_objset = dmu_objset_id(dscp->dsc_os);
954 			zb.zb_object = range->object;
955 			zb.zb_level = 0;
956 			zb.zb_blkid = range->start_blkid;
957 
958 			arc_buf_t *abuf = NULL;
959 			if (!dscp->dsc_dso->dso_dryrun && arc_read(NULL, spa,
960 			    bp, arc_getbuf_func, &abuf, ZIO_PRIORITY_ASYNC_READ,
961 			    zioflags, &aflags, &zb) != 0)
962 				return (SET_ERROR(EIO));
963 
964 			err = dump_spill(dscp, bp, zb.zb_object,
965 			    (abuf == NULL ? NULL : abuf->b_data));
966 			if (abuf != NULL)
967 				arc_buf_destroy(abuf, &abuf);
968 			return (err);
969 		}
970 		if (send_do_embed(bp, dscp->dsc_featureflags)) {
971 			err = dump_write_embedded(dscp, range->object,
972 			    range->start_blkid * srdp->datablksz,
973 			    srdp->datablksz, bp);
974 			return (err);
975 		}
976 		ASSERT(range->object > dscp->dsc_resume_object ||
977 		    (range->object == dscp->dsc_resume_object &&
978 		    range->start_blkid * srdp->datablksz >=
979 		    dscp->dsc_resume_offset));
980 		/* it's a level-0 block of a regular object */
981 
982 		mutex_enter(&srdp->lock);
983 		while (srdp->io_outstanding)
984 			cv_wait(&srdp->cv, &srdp->lock);
985 		err = srdp->io_err;
986 		mutex_exit(&srdp->lock);
987 
988 		if (err != 0) {
989 			if (zfs_send_corrupt_data &&
990 			    !dscp->dsc_dso->dso_dryrun) {
991 				/*
992 				 * Send a block filled with 0x"zfs badd bloc"
993 				 */
994 				srdp->abuf = arc_alloc_buf(spa, &srdp->abuf,
995 				    ARC_BUFC_DATA, srdp->datablksz);
996 				uint64_t *ptr;
997 				for (ptr = srdp->abuf->b_data;
998 				    (char *)ptr < (char *)srdp->abuf->b_data +
999 				    srdp->datablksz; ptr++)
1000 					*ptr = 0x2f5baddb10cULL;
1001 			} else {
1002 				return (SET_ERROR(EIO));
1003 			}
1004 		}
1005 
1006 		ASSERT(dscp->dsc_dso->dso_dryrun ||
1007 		    srdp->abuf != NULL || srdp->abd != NULL);
1008 
1009 		uint64_t offset = range->start_blkid * srdp->datablksz;
1010 
1011 		char *data = NULL;
1012 		if (srdp->abd != NULL) {
1013 			data = abd_to_buf(srdp->abd);
1014 			ASSERT3P(srdp->abuf, ==, NULL);
1015 		} else if (srdp->abuf != NULL) {
1016 			data = srdp->abuf->b_data;
1017 		}
1018 
1019 		/*
1020 		 * If we have large blocks stored on disk but the send flags
1021 		 * don't allow us to send large blocks, we split the data from
1022 		 * the arc buf into chunks.
1023 		 */
1024 		if (srdp->datablksz > SPA_OLD_MAXBLOCKSIZE &&
1025 		    !(dscp->dsc_featureflags &
1026 		    DMU_BACKUP_FEATURE_LARGE_BLOCKS)) {
1027 			while (srdp->datablksz > 0 && err == 0) {
1028 				int n = MIN(srdp->datablksz,
1029 				    SPA_OLD_MAXBLOCKSIZE);
1030 				err = dmu_dump_write(dscp, srdp->obj_type,
1031 				    range->object, offset, n, n, NULL, B_FALSE,
1032 				    data);
1033 				offset += n;
1034 				/*
1035 				 * When doing dry run, data==NULL is used as a
1036 				 * sentinel value by
1037 				 * dmu_dump_write()->dump_record().
1038 				 */
1039 				if (data != NULL)
1040 					data += n;
1041 				srdp->datablksz -= n;
1042 			}
1043 		} else {
1044 			err = dmu_dump_write(dscp, srdp->obj_type,
1045 			    range->object, offset,
1046 			    srdp->datablksz, srdp->datasz, bp,
1047 			    srdp->io_compressed, data);
1048 		}
1049 		return (err);
1050 	}
1051 	case HOLE: {
1052 		struct srh *srhp = &range->sru.hole;
1053 		if (range->object == DMU_META_DNODE_OBJECT) {
1054 			uint32_t span = srhp->datablksz >> DNODE_SHIFT;
1055 			uint64_t first_obj = range->start_blkid * span;
1056 			uint64_t numobj = range->end_blkid * span - first_obj;
1057 			return (dump_freeobjects(dscp, first_obj, numobj));
1058 		}
1059 		uint64_t offset = 0;
1060 
1061 		/*
1062 		 * If this multiply overflows, we don't need to send this block.
1063 		 * Even if it has a birth time, it can never not be a hole, so
1064 		 * we don't need to send records for it.
1065 		 */
1066 		if (!overflow_multiply(range->start_blkid, srhp->datablksz,
1067 		    &offset)) {
1068 			return (0);
1069 		}
1070 		uint64_t len = 0;
1071 
1072 		if (!overflow_multiply(range->end_blkid, srhp->datablksz, &len))
1073 			len = UINT64_MAX;
1074 		len = len - offset;
1075 		return (dump_free(dscp, range->object, offset, len));
1076 	}
1077 	default:
1078 		panic("Invalid range type in do_dump: %d", range->type);
1079 	}
1080 	return (err);
1081 }
1082 
1083 static struct send_range *
1084 range_alloc(enum type type, uint64_t object, uint64_t start_blkid,
1085     uint64_t end_blkid, boolean_t eos)
1086 {
1087 	struct send_range *range = kmem_alloc(sizeof (*range), KM_SLEEP);
1088 	range->type = type;
1089 	range->object = object;
1090 	range->start_blkid = start_blkid;
1091 	range->end_blkid = end_blkid;
1092 	range->eos_marker = eos;
1093 	if (type == DATA) {
1094 		range->sru.data.abd = NULL;
1095 		range->sru.data.abuf = NULL;
1096 		mutex_init(&range->sru.data.lock, NULL, MUTEX_DEFAULT, NULL);
1097 		cv_init(&range->sru.data.cv, NULL, CV_DEFAULT, NULL);
1098 		range->sru.data.io_outstanding = 0;
1099 		range->sru.data.io_err = 0;
1100 		range->sru.data.io_compressed = B_FALSE;
1101 	}
1102 	return (range);
1103 }
1104 
1105 /*
1106  * This is the callback function to traverse_dataset that acts as a worker
1107  * thread for dmu_send_impl.
1108  */
1109 static int
1110 send_cb(spa_t *spa, zilog_t *zilog, const blkptr_t *bp,
1111     const zbookmark_phys_t *zb, const struct dnode_phys *dnp, void *arg)
1112 {
1113 	(void) zilog;
1114 	struct send_thread_arg *sta = arg;
1115 	struct send_range *record;
1116 
1117 	ASSERT(zb->zb_object == DMU_META_DNODE_OBJECT ||
1118 	    zb->zb_object >= sta->resume.zb_object);
1119 
1120 	/*
1121 	 * All bps of an encrypted os should have the encryption bit set.
1122 	 * If this is not true it indicates tampering and we report an error.
1123 	 */
1124 	if (sta->os->os_encrypted &&
1125 	    !BP_IS_HOLE(bp) && !BP_USES_CRYPT(bp)) {
1126 		spa_log_error(spa, zb, &bp->blk_birth);
1127 		zfs_panic_recover("unencrypted block in encrypted "
1128 		    "object set %llu", dmu_objset_id(sta->os));
1129 		return (SET_ERROR(EIO));
1130 	}
1131 
1132 	if (sta->cancel)
1133 		return (SET_ERROR(EINTR));
1134 	if (zb->zb_object != DMU_META_DNODE_OBJECT &&
1135 	    DMU_OBJECT_IS_SPECIAL(zb->zb_object))
1136 		return (0);
1137 	atomic_inc_64(sta->num_blocks_visited);
1138 
1139 	if (zb->zb_level == ZB_DNODE_LEVEL) {
1140 		if (zb->zb_object == DMU_META_DNODE_OBJECT)
1141 			return (0);
1142 		record = range_alloc(OBJECT, zb->zb_object, 0, 0, B_FALSE);
1143 		record->sru.object.bp = *bp;
1144 		size_t size  = sizeof (*dnp) * (dnp->dn_extra_slots + 1);
1145 		record->sru.object.dnp = kmem_alloc(size, KM_SLEEP);
1146 		memcpy(record->sru.object.dnp, dnp, size);
1147 		bqueue_enqueue(&sta->q, record, sizeof (*record));
1148 		return (0);
1149 	}
1150 	if (zb->zb_level == 0 && zb->zb_object == DMU_META_DNODE_OBJECT &&
1151 	    !BP_IS_HOLE(bp)) {
1152 		record = range_alloc(OBJECT_RANGE, 0, zb->zb_blkid,
1153 		    zb->zb_blkid + 1, B_FALSE);
1154 		record->sru.object_range.bp = *bp;
1155 		bqueue_enqueue(&sta->q, record, sizeof (*record));
1156 		return (0);
1157 	}
1158 	if (zb->zb_level < 0 || (zb->zb_level > 0 && !BP_IS_HOLE(bp)))
1159 		return (0);
1160 	if (zb->zb_object == DMU_META_DNODE_OBJECT && !BP_IS_HOLE(bp))
1161 		return (0);
1162 
1163 	uint64_t span = bp_span_in_blocks(dnp->dn_indblkshift, zb->zb_level);
1164 	uint64_t start;
1165 
1166 	/*
1167 	 * If this multiply overflows, we don't need to send this block.
1168 	 * Even if it has a birth time, it can never not be a hole, so
1169 	 * we don't need to send records for it.
1170 	 */
1171 	if (!overflow_multiply(span, zb->zb_blkid, &start) || (!(zb->zb_blkid ==
1172 	    DMU_SPILL_BLKID || DMU_OT_IS_METADATA(dnp->dn_type)) &&
1173 	    span * zb->zb_blkid > dnp->dn_maxblkid)) {
1174 		ASSERT(BP_IS_HOLE(bp));
1175 		return (0);
1176 	}
1177 
1178 	if (zb->zb_blkid == DMU_SPILL_BLKID)
1179 		ASSERT3U(BP_GET_TYPE(bp), ==, DMU_OT_SA);
1180 
1181 	enum type record_type = DATA;
1182 	if (BP_IS_HOLE(bp))
1183 		record_type = HOLE;
1184 	else if (BP_IS_REDACTED(bp))
1185 		record_type = REDACT;
1186 	else
1187 		record_type = DATA;
1188 
1189 	record = range_alloc(record_type, zb->zb_object, start,
1190 	    (start + span < start ? 0 : start + span), B_FALSE);
1191 
1192 	uint64_t datablksz = (zb->zb_blkid == DMU_SPILL_BLKID ?
1193 	    BP_GET_LSIZE(bp) : dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT);
1194 
1195 	if (BP_IS_HOLE(bp)) {
1196 		record->sru.hole.datablksz = datablksz;
1197 	} else if (BP_IS_REDACTED(bp)) {
1198 		record->sru.redact.datablksz = datablksz;
1199 	} else {
1200 		record->sru.data.datablksz = datablksz;
1201 		record->sru.data.obj_type = dnp->dn_type;
1202 		record->sru.data.bp = *bp;
1203 	}
1204 
1205 	bqueue_enqueue(&sta->q, record, sizeof (*record));
1206 	return (0);
1207 }
1208 
1209 struct redact_list_cb_arg {
1210 	uint64_t *num_blocks_visited;
1211 	bqueue_t *q;
1212 	boolean_t *cancel;
1213 	boolean_t mark_redact;
1214 };
1215 
1216 static int
1217 redact_list_cb(redact_block_phys_t *rb, void *arg)
1218 {
1219 	struct redact_list_cb_arg *rlcap = arg;
1220 
1221 	atomic_inc_64(rlcap->num_blocks_visited);
1222 	if (*rlcap->cancel)
1223 		return (-1);
1224 
1225 	struct send_range *data = range_alloc(REDACT, rb->rbp_object,
1226 	    rb->rbp_blkid, rb->rbp_blkid + redact_block_get_count(rb), B_FALSE);
1227 	ASSERT3U(data->end_blkid, >, rb->rbp_blkid);
1228 	if (rlcap->mark_redact) {
1229 		data->type = REDACT;
1230 		data->sru.redact.datablksz = redact_block_get_size(rb);
1231 	} else {
1232 		data->type = PREVIOUSLY_REDACTED;
1233 	}
1234 	bqueue_enqueue(rlcap->q, data, sizeof (*data));
1235 
1236 	return (0);
1237 }
1238 
1239 /*
1240  * This function kicks off the traverse_dataset.  It also handles setting the
1241  * error code of the thread in case something goes wrong, and pushes the End of
1242  * Stream record when the traverse_dataset call has finished.
1243  */
1244 static __attribute__((noreturn)) void
1245 send_traverse_thread(void *arg)
1246 {
1247 	struct send_thread_arg *st_arg = arg;
1248 	int err = 0;
1249 	struct send_range *data;
1250 	fstrans_cookie_t cookie = spl_fstrans_mark();
1251 
1252 	err = traverse_dataset_resume(st_arg->os->os_dsl_dataset,
1253 	    st_arg->fromtxg, &st_arg->resume,
1254 	    st_arg->flags, send_cb, st_arg);
1255 
1256 	if (err != EINTR)
1257 		st_arg->error_code = err;
1258 	data = range_alloc(DATA, 0, 0, 0, B_TRUE);
1259 	bqueue_enqueue_flush(&st_arg->q, data, sizeof (*data));
1260 	spl_fstrans_unmark(cookie);
1261 	thread_exit();
1262 }
1263 
1264 /*
1265  * Utility function that causes End of Stream records to compare after of all
1266  * others, so that other threads' comparison logic can stay simple.
1267  */
1268 static int __attribute__((unused))
1269 send_range_after(const struct send_range *from, const struct send_range *to)
1270 {
1271 	if (from->eos_marker == B_TRUE)
1272 		return (1);
1273 	if (to->eos_marker == B_TRUE)
1274 		return (-1);
1275 
1276 	uint64_t from_obj = from->object;
1277 	uint64_t from_end_obj = from->object + 1;
1278 	uint64_t to_obj = to->object;
1279 	uint64_t to_end_obj = to->object + 1;
1280 	if (from_obj == 0) {
1281 		ASSERT(from->type == HOLE || from->type == OBJECT_RANGE);
1282 		from_obj = from->start_blkid << DNODES_PER_BLOCK_SHIFT;
1283 		from_end_obj = from->end_blkid << DNODES_PER_BLOCK_SHIFT;
1284 	}
1285 	if (to_obj == 0) {
1286 		ASSERT(to->type == HOLE || to->type == OBJECT_RANGE);
1287 		to_obj = to->start_blkid << DNODES_PER_BLOCK_SHIFT;
1288 		to_end_obj = to->end_blkid << DNODES_PER_BLOCK_SHIFT;
1289 	}
1290 
1291 	if (from_end_obj <= to_obj)
1292 		return (-1);
1293 	if (from_obj >= to_end_obj)
1294 		return (1);
1295 	int64_t cmp = TREE_CMP(to->type == OBJECT_RANGE, from->type ==
1296 	    OBJECT_RANGE);
1297 	if (unlikely(cmp))
1298 		return (cmp);
1299 	cmp = TREE_CMP(to->type == OBJECT, from->type == OBJECT);
1300 	if (unlikely(cmp))
1301 		return (cmp);
1302 	if (from->end_blkid <= to->start_blkid)
1303 		return (-1);
1304 	if (from->start_blkid >= to->end_blkid)
1305 		return (1);
1306 	return (0);
1307 }
1308 
1309 /*
1310  * Pop the new data off the queue, check that the records we receive are in
1311  * the right order, but do not free the old data.  This is used so that the
1312  * records can be sent on to the main thread without copying the data.
1313  */
1314 static struct send_range *
1315 get_next_range_nofree(bqueue_t *bq, struct send_range *prev)
1316 {
1317 	struct send_range *next = bqueue_dequeue(bq);
1318 	ASSERT3S(send_range_after(prev, next), ==, -1);
1319 	return (next);
1320 }
1321 
1322 /*
1323  * Pop the new data off the queue, check that the records we receive are in
1324  * the right order, and free the old data.
1325  */
1326 static struct send_range *
1327 get_next_range(bqueue_t *bq, struct send_range *prev)
1328 {
1329 	struct send_range *next = get_next_range_nofree(bq, prev);
1330 	range_free(prev);
1331 	return (next);
1332 }
1333 
1334 static __attribute__((noreturn)) void
1335 redact_list_thread(void *arg)
1336 {
1337 	struct redact_list_thread_arg *rlt_arg = arg;
1338 	struct send_range *record;
1339 	fstrans_cookie_t cookie = spl_fstrans_mark();
1340 	if (rlt_arg->rl != NULL) {
1341 		struct redact_list_cb_arg rlcba = {0};
1342 		rlcba.cancel = &rlt_arg->cancel;
1343 		rlcba.q = &rlt_arg->q;
1344 		rlcba.num_blocks_visited = rlt_arg->num_blocks_visited;
1345 		rlcba.mark_redact = rlt_arg->mark_redact;
1346 		int err = dsl_redaction_list_traverse(rlt_arg->rl,
1347 		    &rlt_arg->resume, redact_list_cb, &rlcba);
1348 		if (err != EINTR)
1349 			rlt_arg->error_code = err;
1350 	}
1351 	record = range_alloc(DATA, 0, 0, 0, B_TRUE);
1352 	bqueue_enqueue_flush(&rlt_arg->q, record, sizeof (*record));
1353 	spl_fstrans_unmark(cookie);
1354 
1355 	thread_exit();
1356 }
1357 
1358 /*
1359  * Compare the start point of the two provided ranges. End of stream ranges
1360  * compare last, objects compare before any data or hole inside that object and
1361  * multi-object holes that start at the same object.
1362  */
1363 static int
1364 send_range_start_compare(struct send_range *r1, struct send_range *r2)
1365 {
1366 	uint64_t r1_objequiv = r1->object;
1367 	uint64_t r1_l0equiv = r1->start_blkid;
1368 	uint64_t r2_objequiv = r2->object;
1369 	uint64_t r2_l0equiv = r2->start_blkid;
1370 	int64_t cmp = TREE_CMP(r1->eos_marker, r2->eos_marker);
1371 	if (unlikely(cmp))
1372 		return (cmp);
1373 	if (r1->object == 0) {
1374 		r1_objequiv = r1->start_blkid * DNODES_PER_BLOCK;
1375 		r1_l0equiv = 0;
1376 	}
1377 	if (r2->object == 0) {
1378 		r2_objequiv = r2->start_blkid * DNODES_PER_BLOCK;
1379 		r2_l0equiv = 0;
1380 	}
1381 
1382 	cmp = TREE_CMP(r1_objequiv, r2_objequiv);
1383 	if (likely(cmp))
1384 		return (cmp);
1385 	cmp = TREE_CMP(r2->type == OBJECT_RANGE, r1->type == OBJECT_RANGE);
1386 	if (unlikely(cmp))
1387 		return (cmp);
1388 	cmp = TREE_CMP(r2->type == OBJECT, r1->type == OBJECT);
1389 	if (unlikely(cmp))
1390 		return (cmp);
1391 
1392 	return (TREE_CMP(r1_l0equiv, r2_l0equiv));
1393 }
1394 
1395 enum q_idx {
1396 	REDACT_IDX = 0,
1397 	TO_IDX,
1398 	FROM_IDX,
1399 	NUM_THREADS
1400 };
1401 
1402 /*
1403  * This function returns the next range the send_merge_thread should operate on.
1404  * The inputs are two arrays; the first one stores the range at the front of the
1405  * queues stored in the second one.  The ranges are sorted in descending
1406  * priority order; the metadata from earlier ranges overrules metadata from
1407  * later ranges.  out_mask is used to return which threads the ranges came from;
1408  * bit i is set if ranges[i] started at the same place as the returned range.
1409  *
1410  * This code is not hardcoded to compare a specific number of threads; it could
1411  * be used with any number, just by changing the q_idx enum.
1412  *
1413  * The "next range" is the one with the earliest start; if two starts are equal,
1414  * the highest-priority range is the next to operate on.  If a higher-priority
1415  * range starts in the middle of the first range, then the first range will be
1416  * truncated to end where the higher-priority range starts, and we will operate
1417  * on that one next time.   In this way, we make sure that each block covered by
1418  * some range gets covered by a returned range, and each block covered is
1419  * returned using the metadata of the highest-priority range it appears in.
1420  *
1421  * For example, if the three ranges at the front of the queues were [2,4),
1422  * [3,5), and [1,3), then the ranges returned would be [1,2) with the metadata
1423  * from the third range, [2,4) with the metadata from the first range, and then
1424  * [4,5) with the metadata from the second.
1425  */
1426 static struct send_range *
1427 find_next_range(struct send_range **ranges, bqueue_t **qs, uint64_t *out_mask)
1428 {
1429 	int idx = 0; // index of the range with the earliest start
1430 	int i;
1431 	uint64_t bmask = 0;
1432 	for (i = 1; i < NUM_THREADS; i++) {
1433 		if (send_range_start_compare(ranges[i], ranges[idx]) < 0)
1434 			idx = i;
1435 	}
1436 	if (ranges[idx]->eos_marker) {
1437 		struct send_range *ret = range_alloc(DATA, 0, 0, 0, B_TRUE);
1438 		*out_mask = 0;
1439 		return (ret);
1440 	}
1441 	/*
1442 	 * Find all the ranges that start at that same point.
1443 	 */
1444 	for (i = 0; i < NUM_THREADS; i++) {
1445 		if (send_range_start_compare(ranges[i], ranges[idx]) == 0)
1446 			bmask |= 1 << i;
1447 	}
1448 	*out_mask = bmask;
1449 	/*
1450 	 * OBJECT_RANGE records only come from the TO thread, and should always
1451 	 * be treated as overlapping with nothing and sent on immediately.  They
1452 	 * are only used in raw sends, and are never redacted.
1453 	 */
1454 	if (ranges[idx]->type == OBJECT_RANGE) {
1455 		ASSERT3U(idx, ==, TO_IDX);
1456 		ASSERT3U(*out_mask, ==, 1 << TO_IDX);
1457 		struct send_range *ret = ranges[idx];
1458 		ranges[idx] = get_next_range_nofree(qs[idx], ranges[idx]);
1459 		return (ret);
1460 	}
1461 	/*
1462 	 * Find the first start or end point after the start of the first range.
1463 	 */
1464 	uint64_t first_change = ranges[idx]->end_blkid;
1465 	for (i = 0; i < NUM_THREADS; i++) {
1466 		if (i == idx || ranges[i]->eos_marker ||
1467 		    ranges[i]->object > ranges[idx]->object ||
1468 		    ranges[i]->object == DMU_META_DNODE_OBJECT)
1469 			continue;
1470 		ASSERT3U(ranges[i]->object, ==, ranges[idx]->object);
1471 		if (first_change > ranges[i]->start_blkid &&
1472 		    (bmask & (1 << i)) == 0)
1473 			first_change = ranges[i]->start_blkid;
1474 		else if (first_change > ranges[i]->end_blkid)
1475 			first_change = ranges[i]->end_blkid;
1476 	}
1477 	/*
1478 	 * Update all ranges to no longer overlap with the range we're
1479 	 * returning. All such ranges must start at the same place as the range
1480 	 * being returned, and end at or after first_change. Thus we update
1481 	 * their start to first_change. If that makes them size 0, then free
1482 	 * them and pull a new range from that thread.
1483 	 */
1484 	for (i = 0; i < NUM_THREADS; i++) {
1485 		if (i == idx || (bmask & (1 << i)) == 0)
1486 			continue;
1487 		ASSERT3U(first_change, >, ranges[i]->start_blkid);
1488 		ranges[i]->start_blkid = first_change;
1489 		ASSERT3U(ranges[i]->start_blkid, <=, ranges[i]->end_blkid);
1490 		if (ranges[i]->start_blkid == ranges[i]->end_blkid)
1491 			ranges[i] = get_next_range(qs[i], ranges[i]);
1492 	}
1493 	/*
1494 	 * Short-circuit the simple case; if the range doesn't overlap with
1495 	 * anything else, or it only overlaps with things that start at the same
1496 	 * place and are longer, send it on.
1497 	 */
1498 	if (first_change == ranges[idx]->end_blkid) {
1499 		struct send_range *ret = ranges[idx];
1500 		ranges[idx] = get_next_range_nofree(qs[idx], ranges[idx]);
1501 		return (ret);
1502 	}
1503 
1504 	/*
1505 	 * Otherwise, return a truncated copy of ranges[idx] and move the start
1506 	 * of ranges[idx] back to first_change.
1507 	 */
1508 	struct send_range *ret = kmem_alloc(sizeof (*ret), KM_SLEEP);
1509 	*ret = *ranges[idx];
1510 	ret->end_blkid = first_change;
1511 	ranges[idx]->start_blkid = first_change;
1512 	return (ret);
1513 }
1514 
1515 #define	FROM_AND_REDACT_BITS ((1 << REDACT_IDX) | (1 << FROM_IDX))
1516 
1517 /*
1518  * Merge the results from the from thread and the to thread, and then hand the
1519  * records off to send_prefetch_thread to prefetch them.  If this is not a
1520  * send from a redaction bookmark, the from thread will push an end of stream
1521  * record and stop, and we'll just send everything that was changed in the
1522  * to_ds since the ancestor's creation txg. If it is, then since
1523  * traverse_dataset has a canonical order, we can compare each change as
1524  * they're pulled off the queues.  That will give us a stream that is
1525  * appropriately sorted, and covers all records.  In addition, we pull the
1526  * data from the redact_list_thread and use that to determine which blocks
1527  * should be redacted.
1528  */
1529 static __attribute__((noreturn)) void
1530 send_merge_thread(void *arg)
1531 {
1532 	struct send_merge_thread_arg *smt_arg = arg;
1533 	struct send_range *front_ranges[NUM_THREADS];
1534 	bqueue_t *queues[NUM_THREADS];
1535 	int err = 0;
1536 	fstrans_cookie_t cookie = spl_fstrans_mark();
1537 
1538 	if (smt_arg->redact_arg == NULL) {
1539 		front_ranges[REDACT_IDX] =
1540 		    kmem_zalloc(sizeof (struct send_range), KM_SLEEP);
1541 		front_ranges[REDACT_IDX]->eos_marker = B_TRUE;
1542 		front_ranges[REDACT_IDX]->type = REDACT;
1543 		queues[REDACT_IDX] = NULL;
1544 	} else {
1545 		front_ranges[REDACT_IDX] =
1546 		    bqueue_dequeue(&smt_arg->redact_arg->q);
1547 		queues[REDACT_IDX] = &smt_arg->redact_arg->q;
1548 	}
1549 	front_ranges[TO_IDX] = bqueue_dequeue(&smt_arg->to_arg->q);
1550 	queues[TO_IDX] = &smt_arg->to_arg->q;
1551 	front_ranges[FROM_IDX] = bqueue_dequeue(&smt_arg->from_arg->q);
1552 	queues[FROM_IDX] = &smt_arg->from_arg->q;
1553 	uint64_t mask = 0;
1554 	struct send_range *range;
1555 	for (range = find_next_range(front_ranges, queues, &mask);
1556 	    !range->eos_marker && err == 0 && !smt_arg->cancel;
1557 	    range = find_next_range(front_ranges, queues, &mask)) {
1558 		/*
1559 		 * If the range in question was in both the from redact bookmark
1560 		 * and the bookmark we're using to redact, then don't send it.
1561 		 * It's already redacted on the receiving system, so a redaction
1562 		 * record would be redundant.
1563 		 */
1564 		if ((mask & FROM_AND_REDACT_BITS) == FROM_AND_REDACT_BITS) {
1565 			ASSERT3U(range->type, ==, REDACT);
1566 			range_free(range);
1567 			continue;
1568 		}
1569 		bqueue_enqueue(&smt_arg->q, range, sizeof (*range));
1570 
1571 		if (smt_arg->to_arg->error_code != 0) {
1572 			err = smt_arg->to_arg->error_code;
1573 		} else if (smt_arg->from_arg->error_code != 0) {
1574 			err = smt_arg->from_arg->error_code;
1575 		} else if (smt_arg->redact_arg != NULL &&
1576 		    smt_arg->redact_arg->error_code != 0) {
1577 			err = smt_arg->redact_arg->error_code;
1578 		}
1579 	}
1580 	if (smt_arg->cancel && err == 0)
1581 		err = SET_ERROR(EINTR);
1582 	smt_arg->error = err;
1583 	if (smt_arg->error != 0) {
1584 		smt_arg->to_arg->cancel = B_TRUE;
1585 		smt_arg->from_arg->cancel = B_TRUE;
1586 		if (smt_arg->redact_arg != NULL)
1587 			smt_arg->redact_arg->cancel = B_TRUE;
1588 	}
1589 	for (int i = 0; i < NUM_THREADS; i++) {
1590 		while (!front_ranges[i]->eos_marker) {
1591 			front_ranges[i] = get_next_range(queues[i],
1592 			    front_ranges[i]);
1593 		}
1594 		range_free(front_ranges[i]);
1595 	}
1596 	range->eos_marker = B_TRUE;
1597 	bqueue_enqueue_flush(&smt_arg->q, range, 1);
1598 	spl_fstrans_unmark(cookie);
1599 	thread_exit();
1600 }
1601 
1602 struct send_reader_thread_arg {
1603 	struct send_merge_thread_arg *smta;
1604 	bqueue_t q;
1605 	boolean_t cancel;
1606 	boolean_t issue_reads;
1607 	uint64_t featureflags;
1608 	int error;
1609 };
1610 
1611 static void
1612 dmu_send_read_done(zio_t *zio)
1613 {
1614 	struct send_range *range = zio->io_private;
1615 
1616 	mutex_enter(&range->sru.data.lock);
1617 	if (zio->io_error != 0) {
1618 		abd_free(range->sru.data.abd);
1619 		range->sru.data.abd = NULL;
1620 		range->sru.data.io_err = zio->io_error;
1621 	}
1622 
1623 	ASSERT(range->sru.data.io_outstanding);
1624 	range->sru.data.io_outstanding = B_FALSE;
1625 	cv_broadcast(&range->sru.data.cv);
1626 	mutex_exit(&range->sru.data.lock);
1627 }
1628 
1629 static void
1630 issue_data_read(struct send_reader_thread_arg *srta, struct send_range *range)
1631 {
1632 	struct srd *srdp = &range->sru.data;
1633 	blkptr_t *bp = &srdp->bp;
1634 	objset_t *os = srta->smta->os;
1635 
1636 	ASSERT3U(range->type, ==, DATA);
1637 	ASSERT3U(range->start_blkid + 1, ==, range->end_blkid);
1638 	/*
1639 	 * If we have large blocks stored on disk but
1640 	 * the send flags don't allow us to send large
1641 	 * blocks, we split the data from the arc buf
1642 	 * into chunks.
1643 	 */
1644 	boolean_t split_large_blocks =
1645 	    srdp->datablksz > SPA_OLD_MAXBLOCKSIZE &&
1646 	    !(srta->featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS);
1647 	/*
1648 	 * We should only request compressed data from the ARC if all
1649 	 * the following are true:
1650 	 *  - stream compression was requested
1651 	 *  - we aren't splitting large blocks into smaller chunks
1652 	 *  - the data won't need to be byteswapped before sending
1653 	 *  - this isn't an embedded block
1654 	 *  - this isn't metadata (if receiving on a different endian
1655 	 *    system it can be byteswapped more easily)
1656 	 */
1657 	boolean_t request_compressed =
1658 	    (srta->featureflags & DMU_BACKUP_FEATURE_COMPRESSED) &&
1659 	    !split_large_blocks && !BP_SHOULD_BYTESWAP(bp) &&
1660 	    !BP_IS_EMBEDDED(bp) && !DMU_OT_IS_METADATA(BP_GET_TYPE(bp));
1661 
1662 	zio_flag_t zioflags = ZIO_FLAG_CANFAIL;
1663 
1664 	if (srta->featureflags & DMU_BACKUP_FEATURE_RAW) {
1665 		zioflags |= ZIO_FLAG_RAW;
1666 		srdp->io_compressed = B_TRUE;
1667 	} else if (request_compressed) {
1668 		zioflags |= ZIO_FLAG_RAW_COMPRESS;
1669 		srdp->io_compressed = B_TRUE;
1670 	}
1671 
1672 	srdp->datasz = (zioflags & ZIO_FLAG_RAW_COMPRESS) ?
1673 	    BP_GET_PSIZE(bp) : BP_GET_LSIZE(bp);
1674 
1675 	if (!srta->issue_reads)
1676 		return;
1677 	if (BP_IS_REDACTED(bp))
1678 		return;
1679 	if (send_do_embed(bp, srta->featureflags))
1680 		return;
1681 
1682 	zbookmark_phys_t zb = {
1683 	    .zb_objset = dmu_objset_id(os),
1684 	    .zb_object = range->object,
1685 	    .zb_level = 0,
1686 	    .zb_blkid = range->start_blkid,
1687 	};
1688 
1689 	arc_flags_t aflags = ARC_FLAG_CACHED_ONLY;
1690 
1691 	int arc_err = arc_read(NULL, os->os_spa, bp,
1692 	    arc_getbuf_func, &srdp->abuf, ZIO_PRIORITY_ASYNC_READ,
1693 	    zioflags, &aflags, &zb);
1694 	/*
1695 	 * If the data is not already cached in the ARC, we read directly
1696 	 * from zio.  This avoids the performance overhead of adding a new
1697 	 * entry to the ARC, and we also avoid polluting the ARC cache with
1698 	 * data that is not likely to be used in the future.
1699 	 */
1700 	if (arc_err != 0) {
1701 		srdp->abd = abd_alloc_linear(srdp->datasz, B_FALSE);
1702 		srdp->io_outstanding = B_TRUE;
1703 		zio_nowait(zio_read(NULL, os->os_spa, bp, srdp->abd,
1704 		    srdp->datasz, dmu_send_read_done, range,
1705 		    ZIO_PRIORITY_ASYNC_READ, zioflags, &zb));
1706 	}
1707 }
1708 
1709 /*
1710  * Create a new record with the given values.
1711  */
1712 static void
1713 enqueue_range(struct send_reader_thread_arg *srta, bqueue_t *q, dnode_t *dn,
1714     uint64_t blkid, uint64_t count, const blkptr_t *bp, uint32_t datablksz)
1715 {
1716 	enum type range_type = (bp == NULL || BP_IS_HOLE(bp) ? HOLE :
1717 	    (BP_IS_REDACTED(bp) ? REDACT : DATA));
1718 
1719 	struct send_range *range = range_alloc(range_type, dn->dn_object,
1720 	    blkid, blkid + count, B_FALSE);
1721 
1722 	if (blkid == DMU_SPILL_BLKID) {
1723 		ASSERT3P(bp, !=, NULL);
1724 		ASSERT3U(BP_GET_TYPE(bp), ==, DMU_OT_SA);
1725 	}
1726 
1727 	switch (range_type) {
1728 	case HOLE:
1729 		range->sru.hole.datablksz = datablksz;
1730 		break;
1731 	case DATA:
1732 		ASSERT3U(count, ==, 1);
1733 		range->sru.data.datablksz = datablksz;
1734 		range->sru.data.obj_type = dn->dn_type;
1735 		range->sru.data.bp = *bp;
1736 		issue_data_read(srta, range);
1737 		break;
1738 	case REDACT:
1739 		range->sru.redact.datablksz = datablksz;
1740 		break;
1741 	default:
1742 		break;
1743 	}
1744 	bqueue_enqueue(q, range, datablksz);
1745 }
1746 
1747 /*
1748  * This thread is responsible for two things: First, it retrieves the correct
1749  * blkptr in the to ds if we need to send the data because of something from
1750  * the from thread.  As a result of this, we're the first ones to discover that
1751  * some indirect blocks can be discarded because they're not holes. Second,
1752  * it issues prefetches for the data we need to send.
1753  */
1754 static __attribute__((noreturn)) void
1755 send_reader_thread(void *arg)
1756 {
1757 	struct send_reader_thread_arg *srta = arg;
1758 	struct send_merge_thread_arg *smta = srta->smta;
1759 	bqueue_t *inq = &smta->q;
1760 	bqueue_t *outq = &srta->q;
1761 	objset_t *os = smta->os;
1762 	fstrans_cookie_t cookie = spl_fstrans_mark();
1763 	struct send_range *range = bqueue_dequeue(inq);
1764 	int err = 0;
1765 
1766 	/*
1767 	 * If the record we're analyzing is from a redaction bookmark from the
1768 	 * fromds, then we need to know whether or not it exists in the tods so
1769 	 * we know whether to create records for it or not. If it does, we need
1770 	 * the datablksz so we can generate an appropriate record for it.
1771 	 * Finally, if it isn't redacted, we need the blkptr so that we can send
1772 	 * a WRITE record containing the actual data.
1773 	 */
1774 	uint64_t last_obj = UINT64_MAX;
1775 	uint64_t last_obj_exists = B_TRUE;
1776 	while (!range->eos_marker && !srta->cancel && smta->error == 0 &&
1777 	    err == 0) {
1778 		switch (range->type) {
1779 		case DATA:
1780 			issue_data_read(srta, range);
1781 			bqueue_enqueue(outq, range, range->sru.data.datablksz);
1782 			range = get_next_range_nofree(inq, range);
1783 			break;
1784 		case HOLE:
1785 		case OBJECT:
1786 		case OBJECT_RANGE:
1787 		case REDACT: // Redacted blocks must exist
1788 			bqueue_enqueue(outq, range, sizeof (*range));
1789 			range = get_next_range_nofree(inq, range);
1790 			break;
1791 		case PREVIOUSLY_REDACTED: {
1792 			/*
1793 			 * This entry came from the "from bookmark" when
1794 			 * sending from a bookmark that has a redaction
1795 			 * list.  We need to check if this object/blkid
1796 			 * exists in the target ("to") dataset, and if
1797 			 * not then we drop this entry.  We also need
1798 			 * to fill in the block pointer so that we know
1799 			 * what to prefetch.
1800 			 *
1801 			 * To accomplish the above, we first cache whether or
1802 			 * not the last object we examined exists.  If it
1803 			 * doesn't, we can drop this record. If it does, we hold
1804 			 * the dnode and use it to call dbuf_dnode_findbp. We do
1805 			 * this instead of dbuf_bookmark_findbp because we will
1806 			 * often operate on large ranges, and holding the dnode
1807 			 * once is more efficient.
1808 			 */
1809 			boolean_t object_exists = B_TRUE;
1810 			/*
1811 			 * If the data is redacted, we only care if it exists,
1812 			 * so that we don't send records for objects that have
1813 			 * been deleted.
1814 			 */
1815 			dnode_t *dn;
1816 			if (range->object == last_obj && !last_obj_exists) {
1817 				/*
1818 				 * If we're still examining the same object as
1819 				 * previously, and it doesn't exist, we don't
1820 				 * need to call dbuf_bookmark_findbp.
1821 				 */
1822 				object_exists = B_FALSE;
1823 			} else {
1824 				err = dnode_hold(os, range->object, FTAG, &dn);
1825 				if (err == ENOENT) {
1826 					object_exists = B_FALSE;
1827 					err = 0;
1828 				}
1829 				last_obj = range->object;
1830 				last_obj_exists = object_exists;
1831 			}
1832 
1833 			if (err != 0) {
1834 				break;
1835 			} else if (!object_exists) {
1836 				/*
1837 				 * The block was modified, but doesn't
1838 				 * exist in the to dataset; if it was
1839 				 * deleted in the to dataset, then we'll
1840 				 * visit the hole bp for it at some point.
1841 				 */
1842 				range = get_next_range(inq, range);
1843 				continue;
1844 			}
1845 			uint64_t file_max =
1846 			    MIN(dn->dn_maxblkid, range->end_blkid);
1847 			/*
1848 			 * The object exists, so we need to try to find the
1849 			 * blkptr for each block in the range we're processing.
1850 			 */
1851 			rw_enter(&dn->dn_struct_rwlock, RW_READER);
1852 			for (uint64_t blkid = range->start_blkid;
1853 			    blkid < file_max; blkid++) {
1854 				blkptr_t bp;
1855 				uint32_t datablksz =
1856 				    dn->dn_phys->dn_datablkszsec <<
1857 				    SPA_MINBLOCKSHIFT;
1858 				uint64_t offset = blkid * datablksz;
1859 				/*
1860 				 * This call finds the next non-hole block in
1861 				 * the object. This is to prevent a
1862 				 * performance problem where we're unredacting
1863 				 * a large hole. Using dnode_next_offset to
1864 				 * skip over the large hole avoids iterating
1865 				 * over every block in it.
1866 				 */
1867 				err = dnode_next_offset(dn, DNODE_FIND_HAVELOCK,
1868 				    &offset, 1, 1, 0);
1869 				if (err == ESRCH) {
1870 					offset = UINT64_MAX;
1871 					err = 0;
1872 				} else if (err != 0) {
1873 					break;
1874 				}
1875 				if (offset != blkid * datablksz) {
1876 					/*
1877 					 * if there is a hole from here
1878 					 * (blkid) to offset
1879 					 */
1880 					offset = MIN(offset, file_max *
1881 					    datablksz);
1882 					uint64_t nblks = (offset / datablksz) -
1883 					    blkid;
1884 					enqueue_range(srta, outq, dn, blkid,
1885 					    nblks, NULL, datablksz);
1886 					blkid += nblks;
1887 				}
1888 				if (blkid >= file_max)
1889 					break;
1890 				err = dbuf_dnode_findbp(dn, 0, blkid, &bp,
1891 				    NULL, NULL);
1892 				if (err != 0)
1893 					break;
1894 				ASSERT(!BP_IS_HOLE(&bp));
1895 				enqueue_range(srta, outq, dn, blkid, 1, &bp,
1896 				    datablksz);
1897 			}
1898 			rw_exit(&dn->dn_struct_rwlock);
1899 			dnode_rele(dn, FTAG);
1900 			range = get_next_range(inq, range);
1901 		}
1902 		}
1903 	}
1904 	if (srta->cancel || err != 0) {
1905 		smta->cancel = B_TRUE;
1906 		srta->error = err;
1907 	} else if (smta->error != 0) {
1908 		srta->error = smta->error;
1909 	}
1910 	while (!range->eos_marker)
1911 		range = get_next_range(inq, range);
1912 
1913 	bqueue_enqueue_flush(outq, range, 1);
1914 	spl_fstrans_unmark(cookie);
1915 	thread_exit();
1916 }
1917 
1918 #define	NUM_SNAPS_NOT_REDACTED UINT64_MAX
1919 
1920 struct dmu_send_params {
1921 	/* Pool args */
1922 	const void *tag; // Tag dp was held with, will be used to release dp.
1923 	dsl_pool_t *dp;
1924 	/* To snapshot args */
1925 	const char *tosnap;
1926 	dsl_dataset_t *to_ds;
1927 	/* From snapshot args */
1928 	zfs_bookmark_phys_t ancestor_zb;
1929 	uint64_t *fromredactsnaps;
1930 	/* NUM_SNAPS_NOT_REDACTED if not sending from redaction bookmark */
1931 	uint64_t numfromredactsnaps;
1932 	/* Stream params */
1933 	boolean_t is_clone;
1934 	boolean_t embedok;
1935 	boolean_t large_block_ok;
1936 	boolean_t compressok;
1937 	boolean_t rawok;
1938 	boolean_t savedok;
1939 	uint64_t resumeobj;
1940 	uint64_t resumeoff;
1941 	uint64_t saved_guid;
1942 	zfs_bookmark_phys_t *redactbook;
1943 	/* Stream output params */
1944 	dmu_send_outparams_t *dso;
1945 
1946 	/* Stream progress params */
1947 	offset_t *off;
1948 	int outfd;
1949 	char saved_toname[MAXNAMELEN];
1950 };
1951 
1952 static int
1953 setup_featureflags(struct dmu_send_params *dspp, objset_t *os,
1954     uint64_t *featureflags)
1955 {
1956 	dsl_dataset_t *to_ds = dspp->to_ds;
1957 	dsl_pool_t *dp = dspp->dp;
1958 #ifdef _KERNEL
1959 	if (dmu_objset_type(os) == DMU_OST_ZFS) {
1960 		uint64_t version;
1961 		if (zfs_get_zplprop(os, ZFS_PROP_VERSION, &version) != 0)
1962 			return (SET_ERROR(EINVAL));
1963 
1964 		if (version >= ZPL_VERSION_SA)
1965 			*featureflags |= DMU_BACKUP_FEATURE_SA_SPILL;
1966 	}
1967 #endif
1968 
1969 	/* raw sends imply large_block_ok */
1970 	if ((dspp->rawok || dspp->large_block_ok) &&
1971 	    dsl_dataset_feature_is_active(to_ds, SPA_FEATURE_LARGE_BLOCKS)) {
1972 		*featureflags |= DMU_BACKUP_FEATURE_LARGE_BLOCKS;
1973 	}
1974 
1975 	/* encrypted datasets will not have embedded blocks */
1976 	if ((dspp->embedok || dspp->rawok) && !os->os_encrypted &&
1977 	    spa_feature_is_active(dp->dp_spa, SPA_FEATURE_EMBEDDED_DATA)) {
1978 		*featureflags |= DMU_BACKUP_FEATURE_EMBED_DATA;
1979 	}
1980 
1981 	/* raw send implies compressok */
1982 	if (dspp->compressok || dspp->rawok)
1983 		*featureflags |= DMU_BACKUP_FEATURE_COMPRESSED;
1984 
1985 	if (dspp->rawok && os->os_encrypted)
1986 		*featureflags |= DMU_BACKUP_FEATURE_RAW;
1987 
1988 	if ((*featureflags &
1989 	    (DMU_BACKUP_FEATURE_EMBED_DATA | DMU_BACKUP_FEATURE_COMPRESSED |
1990 	    DMU_BACKUP_FEATURE_RAW)) != 0 &&
1991 	    spa_feature_is_active(dp->dp_spa, SPA_FEATURE_LZ4_COMPRESS)) {
1992 		*featureflags |= DMU_BACKUP_FEATURE_LZ4;
1993 	}
1994 
1995 	/*
1996 	 * We specifically do not include DMU_BACKUP_FEATURE_EMBED_DATA here to
1997 	 * allow sending ZSTD compressed datasets to a receiver that does not
1998 	 * support ZSTD
1999 	 */
2000 	if ((*featureflags &
2001 	    (DMU_BACKUP_FEATURE_COMPRESSED | DMU_BACKUP_FEATURE_RAW)) != 0 &&
2002 	    dsl_dataset_feature_is_active(to_ds, SPA_FEATURE_ZSTD_COMPRESS)) {
2003 		*featureflags |= DMU_BACKUP_FEATURE_ZSTD;
2004 	}
2005 
2006 	if (dspp->resumeobj != 0 || dspp->resumeoff != 0) {
2007 		*featureflags |= DMU_BACKUP_FEATURE_RESUMING;
2008 	}
2009 
2010 	if (dspp->redactbook != NULL) {
2011 		*featureflags |= DMU_BACKUP_FEATURE_REDACTED;
2012 	}
2013 
2014 	if (dsl_dataset_feature_is_active(to_ds, SPA_FEATURE_LARGE_DNODE)) {
2015 		*featureflags |= DMU_BACKUP_FEATURE_LARGE_DNODE;
2016 	}
2017 	return (0);
2018 }
2019 
2020 static dmu_replay_record_t *
2021 create_begin_record(struct dmu_send_params *dspp, objset_t *os,
2022     uint64_t featureflags)
2023 {
2024 	dmu_replay_record_t *drr = kmem_zalloc(sizeof (dmu_replay_record_t),
2025 	    KM_SLEEP);
2026 	drr->drr_type = DRR_BEGIN;
2027 
2028 	struct drr_begin *drrb = &drr->drr_u.drr_begin;
2029 	dsl_dataset_t *to_ds = dspp->to_ds;
2030 
2031 	drrb->drr_magic = DMU_BACKUP_MAGIC;
2032 	drrb->drr_creation_time = dsl_dataset_phys(to_ds)->ds_creation_time;
2033 	drrb->drr_type = dmu_objset_type(os);
2034 	drrb->drr_toguid = dsl_dataset_phys(to_ds)->ds_guid;
2035 	drrb->drr_fromguid = dspp->ancestor_zb.zbm_guid;
2036 
2037 	DMU_SET_STREAM_HDRTYPE(drrb->drr_versioninfo, DMU_SUBSTREAM);
2038 	DMU_SET_FEATUREFLAGS(drrb->drr_versioninfo, featureflags);
2039 
2040 	if (dspp->is_clone)
2041 		drrb->drr_flags |= DRR_FLAG_CLONE;
2042 	if (dsl_dataset_phys(dspp->to_ds)->ds_flags & DS_FLAG_CI_DATASET)
2043 		drrb->drr_flags |= DRR_FLAG_CI_DATA;
2044 	if (zfs_send_set_freerecords_bit)
2045 		drrb->drr_flags |= DRR_FLAG_FREERECORDS;
2046 	drr->drr_u.drr_begin.drr_flags |= DRR_FLAG_SPILL_BLOCK;
2047 
2048 	if (dspp->savedok) {
2049 		drrb->drr_toguid = dspp->saved_guid;
2050 		strlcpy(drrb->drr_toname, dspp->saved_toname,
2051 		    sizeof (drrb->drr_toname));
2052 	} else {
2053 		dsl_dataset_name(to_ds, drrb->drr_toname);
2054 		if (!to_ds->ds_is_snapshot) {
2055 			(void) strlcat(drrb->drr_toname, "@--head--",
2056 			    sizeof (drrb->drr_toname));
2057 		}
2058 	}
2059 	return (drr);
2060 }
2061 
2062 static void
2063 setup_to_thread(struct send_thread_arg *to_arg, objset_t *to_os,
2064     dmu_sendstatus_t *dssp, uint64_t fromtxg, boolean_t rawok)
2065 {
2066 	VERIFY0(bqueue_init(&to_arg->q, zfs_send_no_prefetch_queue_ff,
2067 	    MAX(zfs_send_no_prefetch_queue_length, 2 * zfs_max_recordsize),
2068 	    offsetof(struct send_range, ln)));
2069 	to_arg->error_code = 0;
2070 	to_arg->cancel = B_FALSE;
2071 	to_arg->os = to_os;
2072 	to_arg->fromtxg = fromtxg;
2073 	to_arg->flags = TRAVERSE_PRE | TRAVERSE_PREFETCH_METADATA;
2074 	if (rawok)
2075 		to_arg->flags |= TRAVERSE_NO_DECRYPT;
2076 	if (zfs_send_corrupt_data)
2077 		to_arg->flags |= TRAVERSE_HARD;
2078 	to_arg->num_blocks_visited = &dssp->dss_blocks;
2079 	(void) thread_create(NULL, 0, send_traverse_thread, to_arg, 0,
2080 	    curproc, TS_RUN, minclsyspri);
2081 }
2082 
2083 static void
2084 setup_from_thread(struct redact_list_thread_arg *from_arg,
2085     redaction_list_t *from_rl, dmu_sendstatus_t *dssp)
2086 {
2087 	VERIFY0(bqueue_init(&from_arg->q, zfs_send_no_prefetch_queue_ff,
2088 	    MAX(zfs_send_no_prefetch_queue_length, 2 * zfs_max_recordsize),
2089 	    offsetof(struct send_range, ln)));
2090 	from_arg->error_code = 0;
2091 	from_arg->cancel = B_FALSE;
2092 	from_arg->rl = from_rl;
2093 	from_arg->mark_redact = B_FALSE;
2094 	from_arg->num_blocks_visited = &dssp->dss_blocks;
2095 	/*
2096 	 * If from_ds is null, send_traverse_thread just returns success and
2097 	 * enqueues an eos marker.
2098 	 */
2099 	(void) thread_create(NULL, 0, redact_list_thread, from_arg, 0,
2100 	    curproc, TS_RUN, minclsyspri);
2101 }
2102 
2103 static void
2104 setup_redact_list_thread(struct redact_list_thread_arg *rlt_arg,
2105     struct dmu_send_params *dspp, redaction_list_t *rl, dmu_sendstatus_t *dssp)
2106 {
2107 	if (dspp->redactbook == NULL)
2108 		return;
2109 
2110 	rlt_arg->cancel = B_FALSE;
2111 	VERIFY0(bqueue_init(&rlt_arg->q, zfs_send_no_prefetch_queue_ff,
2112 	    MAX(zfs_send_no_prefetch_queue_length, 2 * zfs_max_recordsize),
2113 	    offsetof(struct send_range, ln)));
2114 	rlt_arg->error_code = 0;
2115 	rlt_arg->mark_redact = B_TRUE;
2116 	rlt_arg->rl = rl;
2117 	rlt_arg->num_blocks_visited = &dssp->dss_blocks;
2118 
2119 	(void) thread_create(NULL, 0, redact_list_thread, rlt_arg, 0,
2120 	    curproc, TS_RUN, minclsyspri);
2121 }
2122 
2123 static void
2124 setup_merge_thread(struct send_merge_thread_arg *smt_arg,
2125     struct dmu_send_params *dspp, struct redact_list_thread_arg *from_arg,
2126     struct send_thread_arg *to_arg, struct redact_list_thread_arg *rlt_arg,
2127     objset_t *os)
2128 {
2129 	VERIFY0(bqueue_init(&smt_arg->q, zfs_send_no_prefetch_queue_ff,
2130 	    MAX(zfs_send_no_prefetch_queue_length, 2 * zfs_max_recordsize),
2131 	    offsetof(struct send_range, ln)));
2132 	smt_arg->cancel = B_FALSE;
2133 	smt_arg->error = 0;
2134 	smt_arg->from_arg = from_arg;
2135 	smt_arg->to_arg = to_arg;
2136 	if (dspp->redactbook != NULL)
2137 		smt_arg->redact_arg = rlt_arg;
2138 
2139 	smt_arg->os = os;
2140 	(void) thread_create(NULL, 0, send_merge_thread, smt_arg, 0, curproc,
2141 	    TS_RUN, minclsyspri);
2142 }
2143 
2144 static void
2145 setup_reader_thread(struct send_reader_thread_arg *srt_arg,
2146     struct dmu_send_params *dspp, struct send_merge_thread_arg *smt_arg,
2147     uint64_t featureflags)
2148 {
2149 	VERIFY0(bqueue_init(&srt_arg->q, zfs_send_queue_ff,
2150 	    MAX(zfs_send_queue_length, 2 * zfs_max_recordsize),
2151 	    offsetof(struct send_range, ln)));
2152 	srt_arg->smta = smt_arg;
2153 	srt_arg->issue_reads = !dspp->dso->dso_dryrun;
2154 	srt_arg->featureflags = featureflags;
2155 	(void) thread_create(NULL, 0, send_reader_thread, srt_arg, 0,
2156 	    curproc, TS_RUN, minclsyspri);
2157 }
2158 
2159 static int
2160 setup_resume_points(struct dmu_send_params *dspp,
2161     struct send_thread_arg *to_arg, struct redact_list_thread_arg *from_arg,
2162     struct redact_list_thread_arg *rlt_arg,
2163     struct send_merge_thread_arg *smt_arg, boolean_t resuming, objset_t *os,
2164     redaction_list_t *redact_rl, nvlist_t *nvl)
2165 {
2166 	(void) smt_arg;
2167 	dsl_dataset_t *to_ds = dspp->to_ds;
2168 	int err = 0;
2169 
2170 	uint64_t obj = 0;
2171 	uint64_t blkid = 0;
2172 	if (resuming) {
2173 		obj = dspp->resumeobj;
2174 		dmu_object_info_t to_doi;
2175 		err = dmu_object_info(os, obj, &to_doi);
2176 		if (err != 0)
2177 			return (err);
2178 
2179 		blkid = dspp->resumeoff / to_doi.doi_data_block_size;
2180 	}
2181 	/*
2182 	 * If we're resuming a redacted send, we can skip to the appropriate
2183 	 * point in the redaction bookmark by binary searching through it.
2184 	 */
2185 	if (redact_rl != NULL) {
2186 		SET_BOOKMARK(&rlt_arg->resume, to_ds->ds_object, obj, 0, blkid);
2187 	}
2188 
2189 	SET_BOOKMARK(&to_arg->resume, to_ds->ds_object, obj, 0, blkid);
2190 	if (nvlist_exists(nvl, BEGINNV_REDACT_FROM_SNAPS)) {
2191 		uint64_t objset = dspp->ancestor_zb.zbm_redaction_obj;
2192 		/*
2193 		 * Note: If the resume point is in an object whose
2194 		 * blocksize is different in the from vs to snapshots,
2195 		 * we will have divided by the "wrong" blocksize.
2196 		 * However, in this case fromsnap's send_cb() will
2197 		 * detect that the blocksize has changed and therefore
2198 		 * ignore this object.
2199 		 *
2200 		 * If we're resuming a send from a redaction bookmark,
2201 		 * we still cannot accidentally suggest blocks behind
2202 		 * the to_ds.  In addition, we know that any blocks in
2203 		 * the object in the to_ds will have to be sent, since
2204 		 * the size changed.  Therefore, we can't cause any harm
2205 		 * this way either.
2206 		 */
2207 		SET_BOOKMARK(&from_arg->resume, objset, obj, 0, blkid);
2208 	}
2209 	if (resuming) {
2210 		fnvlist_add_uint64(nvl, BEGINNV_RESUME_OBJECT, dspp->resumeobj);
2211 		fnvlist_add_uint64(nvl, BEGINNV_RESUME_OFFSET, dspp->resumeoff);
2212 	}
2213 	return (0);
2214 }
2215 
2216 static dmu_sendstatus_t *
2217 setup_send_progress(struct dmu_send_params *dspp)
2218 {
2219 	dmu_sendstatus_t *dssp = kmem_zalloc(sizeof (*dssp), KM_SLEEP);
2220 	dssp->dss_outfd = dspp->outfd;
2221 	dssp->dss_off = dspp->off;
2222 	dssp->dss_proc = curproc;
2223 	mutex_enter(&dspp->to_ds->ds_sendstream_lock);
2224 	list_insert_head(&dspp->to_ds->ds_sendstreams, dssp);
2225 	mutex_exit(&dspp->to_ds->ds_sendstream_lock);
2226 	return (dssp);
2227 }
2228 
2229 /*
2230  * Actually do the bulk of the work in a zfs send.
2231  *
2232  * The idea is that we want to do a send from ancestor_zb to to_ds.  We also
2233  * want to not send any data that has been modified by all the datasets in
2234  * redactsnaparr, and store the list of blocks that are redacted in this way in
2235  * a bookmark named redactbook, created on the to_ds.  We do this by creating
2236  * several worker threads, whose function is described below.
2237  *
2238  * There are three cases.
2239  * The first case is a redacted zfs send.  In this case there are 5 threads.
2240  * The first thread is the to_ds traversal thread: it calls dataset_traverse on
2241  * the to_ds and finds all the blocks that have changed since ancestor_zb (if
2242  * it's a full send, that's all blocks in the dataset).  It then sends those
2243  * blocks on to the send merge thread. The redact list thread takes the data
2244  * from the redaction bookmark and sends those blocks on to the send merge
2245  * thread.  The send merge thread takes the data from the to_ds traversal
2246  * thread, and combines it with the redaction records from the redact list
2247  * thread.  If a block appears in both the to_ds's data and the redaction data,
2248  * the send merge thread will mark it as redacted and send it on to the prefetch
2249  * thread.  Otherwise, the send merge thread will send the block on to the
2250  * prefetch thread unchanged. The prefetch thread will issue prefetch reads for
2251  * any data that isn't redacted, and then send the data on to the main thread.
2252  * The main thread behaves the same as in a normal send case, issuing demand
2253  * reads for data blocks and sending out records over the network
2254  *
2255  * The graphic below diagrams the flow of data in the case of a redacted zfs
2256  * send.  Each box represents a thread, and each line represents the flow of
2257  * data.
2258  *
2259  *             Records from the |
2260  *           redaction bookmark |
2261  * +--------------------+       |  +---------------------------+
2262  * |                    |       v  | Send Merge Thread         |
2263  * | Redact List Thread +----------> Apply redaction marks to  |
2264  * |                    |          | records as specified by   |
2265  * +--------------------+          | redaction ranges          |
2266  *                                 +----^---------------+------+
2267  *                                      |               | Merged data
2268  *                                      |               |
2269  *                                      |  +------------v--------+
2270  *                                      |  | Prefetch Thread     |
2271  * +--------------------+               |  | Issues prefetch     |
2272  * | to_ds Traversal    |               |  | reads of data blocks|
2273  * | Thread (finds      +---------------+  +------------+--------+
2274  * | candidate blocks)  |  Blocks modified              | Prefetched data
2275  * +--------------------+  by to_ds since               |
2276  *                         ancestor_zb     +------------v----+
2277  *                                         | Main Thread     |  File Descriptor
2278  *                                         | Sends data over +->(to zfs receive)
2279  *                                         | wire            |
2280  *                                         +-----------------+
2281  *
2282  * The second case is an incremental send from a redaction bookmark.  The to_ds
2283  * traversal thread and the main thread behave the same as in the redacted
2284  * send case.  The new thread is the from bookmark traversal thread.  It
2285  * iterates over the redaction list in the redaction bookmark, and enqueues
2286  * records for each block that was redacted in the original send.  The send
2287  * merge thread now has to merge the data from the two threads.  For details
2288  * about that process, see the header comment of send_merge_thread().  Any data
2289  * it decides to send on will be prefetched by the prefetch thread.  Note that
2290  * you can perform a redacted send from a redaction bookmark; in that case,
2291  * the data flow behaves very similarly to the flow in the redacted send case,
2292  * except with the addition of the bookmark traversal thread iterating over the
2293  * redaction bookmark.  The send_merge_thread also has to take on the
2294  * responsibility of merging the redact list thread's records, the bookmark
2295  * traversal thread's records, and the to_ds records.
2296  *
2297  * +---------------------+
2298  * |                     |
2299  * | Redact List Thread  +--------------+
2300  * |                     |              |
2301  * +---------------------+              |
2302  *        Blocks in redaction list      | Ranges modified by every secure snap
2303  *        of from bookmark              | (or EOS if not readcted)
2304  *                                      |
2305  * +---------------------+   |     +----v----------------------+
2306  * | bookmark Traversal  |   v     | Send Merge Thread         |
2307  * | Thread (finds       +---------> Merges bookmark, rlt, and |
2308  * | candidate blocks)   |         | to_ds send records        |
2309  * +---------------------+         +----^---------------+------+
2310  *                                      |               | Merged data
2311  *                                      |  +------------v--------+
2312  *                                      |  | Prefetch Thread     |
2313  * +--------------------+               |  | Issues prefetch     |
2314  * | to_ds Traversal    |               |  | reads of data blocks|
2315  * | Thread (finds      +---------------+  +------------+--------+
2316  * | candidate blocks)  |  Blocks modified              | Prefetched data
2317  * +--------------------+  by to_ds since  +------------v----+
2318  *                         ancestor_zb     | Main Thread     |  File Descriptor
2319  *                                         | Sends data over +->(to zfs receive)
2320  *                                         | wire            |
2321  *                                         +-----------------+
2322  *
2323  * The final case is a simple zfs full or incremental send.  The to_ds traversal
2324  * thread behaves the same as always. The redact list thread is never started.
2325  * The send merge thread takes all the blocks that the to_ds traversal thread
2326  * sends it, prefetches the data, and sends the blocks on to the main thread.
2327  * The main thread sends the data over the wire.
2328  *
2329  * To keep performance acceptable, we want to prefetch the data in the worker
2330  * threads.  While the to_ds thread could simply use the TRAVERSE_PREFETCH
2331  * feature built into traverse_dataset, the combining and deletion of records
2332  * due to redaction and sends from redaction bookmarks mean that we could
2333  * issue many unnecessary prefetches.  As a result, we only prefetch data
2334  * after we've determined that the record is not going to be redacted.  To
2335  * prevent the prefetching from getting too far ahead of the main thread, the
2336  * blocking queues that are used for communication are capped not by the
2337  * number of entries in the queue, but by the sum of the size of the
2338  * prefetches associated with them.  The limit on the amount of data that the
2339  * thread can prefetch beyond what the main thread has reached is controlled
2340  * by the global variable zfs_send_queue_length.  In addition, to prevent poor
2341  * performance in the beginning of a send, we also limit the distance ahead
2342  * that the traversal threads can be.  That distance is controlled by the
2343  * zfs_send_no_prefetch_queue_length tunable.
2344  *
2345  * Note: Releases dp using the specified tag.
2346  */
2347 static int
2348 dmu_send_impl(struct dmu_send_params *dspp)
2349 {
2350 	objset_t *os;
2351 	dmu_replay_record_t *drr;
2352 	dmu_sendstatus_t *dssp;
2353 	dmu_send_cookie_t dsc = {0};
2354 	int err;
2355 	uint64_t fromtxg = dspp->ancestor_zb.zbm_creation_txg;
2356 	uint64_t featureflags = 0;
2357 	struct redact_list_thread_arg *from_arg;
2358 	struct send_thread_arg *to_arg;
2359 	struct redact_list_thread_arg *rlt_arg;
2360 	struct send_merge_thread_arg *smt_arg;
2361 	struct send_reader_thread_arg *srt_arg;
2362 	struct send_range *range;
2363 	redaction_list_t *from_rl = NULL;
2364 	redaction_list_t *redact_rl = NULL;
2365 	boolean_t resuming = (dspp->resumeobj != 0 || dspp->resumeoff != 0);
2366 	boolean_t book_resuming = resuming;
2367 
2368 	dsl_dataset_t *to_ds = dspp->to_ds;
2369 	zfs_bookmark_phys_t *ancestor_zb = &dspp->ancestor_zb;
2370 	dsl_pool_t *dp = dspp->dp;
2371 	const void *tag = dspp->tag;
2372 
2373 	err = dmu_objset_from_ds(to_ds, &os);
2374 	if (err != 0) {
2375 		dsl_pool_rele(dp, tag);
2376 		return (err);
2377 	}
2378 
2379 	/*
2380 	 * If this is a non-raw send of an encrypted ds, we can ensure that
2381 	 * the objset_phys_t is authenticated. This is safe because this is
2382 	 * either a snapshot or we have owned the dataset, ensuring that
2383 	 * it can't be modified.
2384 	 */
2385 	if (!dspp->rawok && os->os_encrypted &&
2386 	    arc_is_unauthenticated(os->os_phys_buf)) {
2387 		zbookmark_phys_t zb;
2388 
2389 		SET_BOOKMARK(&zb, to_ds->ds_object, ZB_ROOT_OBJECT,
2390 		    ZB_ROOT_LEVEL, ZB_ROOT_BLKID);
2391 		err = arc_untransform(os->os_phys_buf, os->os_spa,
2392 		    &zb, B_FALSE);
2393 		if (err != 0) {
2394 			dsl_pool_rele(dp, tag);
2395 			return (err);
2396 		}
2397 
2398 		ASSERT0(arc_is_unauthenticated(os->os_phys_buf));
2399 	}
2400 
2401 	if ((err = setup_featureflags(dspp, os, &featureflags)) != 0) {
2402 		dsl_pool_rele(dp, tag);
2403 		return (err);
2404 	}
2405 
2406 	/*
2407 	 * If we're doing a redacted send, hold the bookmark's redaction list.
2408 	 */
2409 	if (dspp->redactbook != NULL) {
2410 		err = dsl_redaction_list_hold_obj(dp,
2411 		    dspp->redactbook->zbm_redaction_obj, FTAG,
2412 		    &redact_rl);
2413 		if (err != 0) {
2414 			dsl_pool_rele(dp, tag);
2415 			return (SET_ERROR(EINVAL));
2416 		}
2417 		dsl_redaction_list_long_hold(dp, redact_rl, FTAG);
2418 	}
2419 
2420 	/*
2421 	 * If we're sending from a redaction bookmark, hold the redaction list
2422 	 * so that we can consider sending the redacted blocks.
2423 	 */
2424 	if (ancestor_zb->zbm_redaction_obj != 0) {
2425 		err = dsl_redaction_list_hold_obj(dp,
2426 		    ancestor_zb->zbm_redaction_obj, FTAG, &from_rl);
2427 		if (err != 0) {
2428 			if (redact_rl != NULL) {
2429 				dsl_redaction_list_long_rele(redact_rl, FTAG);
2430 				dsl_redaction_list_rele(redact_rl, FTAG);
2431 			}
2432 			dsl_pool_rele(dp, tag);
2433 			return (SET_ERROR(EINVAL));
2434 		}
2435 		dsl_redaction_list_long_hold(dp, from_rl, FTAG);
2436 	}
2437 
2438 	dsl_dataset_long_hold(to_ds, FTAG);
2439 
2440 	from_arg = kmem_zalloc(sizeof (*from_arg), KM_SLEEP);
2441 	to_arg = kmem_zalloc(sizeof (*to_arg), KM_SLEEP);
2442 	rlt_arg = kmem_zalloc(sizeof (*rlt_arg), KM_SLEEP);
2443 	smt_arg = kmem_zalloc(sizeof (*smt_arg), KM_SLEEP);
2444 	srt_arg = kmem_zalloc(sizeof (*srt_arg), KM_SLEEP);
2445 
2446 	drr = create_begin_record(dspp, os, featureflags);
2447 	dssp = setup_send_progress(dspp);
2448 
2449 	dsc.dsc_drr = drr;
2450 	dsc.dsc_dso = dspp->dso;
2451 	dsc.dsc_os = os;
2452 	dsc.dsc_off = dspp->off;
2453 	dsc.dsc_toguid = dsl_dataset_phys(to_ds)->ds_guid;
2454 	dsc.dsc_fromtxg = fromtxg;
2455 	dsc.dsc_pending_op = PENDING_NONE;
2456 	dsc.dsc_featureflags = featureflags;
2457 	dsc.dsc_resume_object = dspp->resumeobj;
2458 	dsc.dsc_resume_offset = dspp->resumeoff;
2459 
2460 	dsl_pool_rele(dp, tag);
2461 
2462 	void *payload = NULL;
2463 	size_t payload_len = 0;
2464 	nvlist_t *nvl = fnvlist_alloc();
2465 
2466 	/*
2467 	 * If we're doing a redacted send, we include the snapshots we're
2468 	 * redacted with respect to so that the target system knows what send
2469 	 * streams can be correctly received on top of this dataset. If we're
2470 	 * instead sending a redacted dataset, we include the snapshots that the
2471 	 * dataset was created with respect to.
2472 	 */
2473 	if (dspp->redactbook != NULL) {
2474 		fnvlist_add_uint64_array(nvl, BEGINNV_REDACT_SNAPS,
2475 		    redact_rl->rl_phys->rlp_snaps,
2476 		    redact_rl->rl_phys->rlp_num_snaps);
2477 	} else if (dsl_dataset_feature_is_active(to_ds,
2478 	    SPA_FEATURE_REDACTED_DATASETS)) {
2479 		uint64_t *tods_guids;
2480 		uint64_t length;
2481 		VERIFY(dsl_dataset_get_uint64_array_feature(to_ds,
2482 		    SPA_FEATURE_REDACTED_DATASETS, &length, &tods_guids));
2483 		fnvlist_add_uint64_array(nvl, BEGINNV_REDACT_SNAPS, tods_guids,
2484 		    length);
2485 	}
2486 
2487 	/*
2488 	 * If we're sending from a redaction bookmark, then we should retrieve
2489 	 * the guids of that bookmark so we can send them over the wire.
2490 	 */
2491 	if (from_rl != NULL) {
2492 		fnvlist_add_uint64_array(nvl, BEGINNV_REDACT_FROM_SNAPS,
2493 		    from_rl->rl_phys->rlp_snaps,
2494 		    from_rl->rl_phys->rlp_num_snaps);
2495 	}
2496 
2497 	/*
2498 	 * If the snapshot we're sending from is redacted, include the redaction
2499 	 * list in the stream.
2500 	 */
2501 	if (dspp->numfromredactsnaps != NUM_SNAPS_NOT_REDACTED) {
2502 		ASSERT3P(from_rl, ==, NULL);
2503 		fnvlist_add_uint64_array(nvl, BEGINNV_REDACT_FROM_SNAPS,
2504 		    dspp->fromredactsnaps, (uint_t)dspp->numfromredactsnaps);
2505 		if (dspp->numfromredactsnaps > 0) {
2506 			kmem_free(dspp->fromredactsnaps,
2507 			    dspp->numfromredactsnaps * sizeof (uint64_t));
2508 			dspp->fromredactsnaps = NULL;
2509 		}
2510 	}
2511 
2512 	if (resuming || book_resuming) {
2513 		err = setup_resume_points(dspp, to_arg, from_arg,
2514 		    rlt_arg, smt_arg, resuming, os, redact_rl, nvl);
2515 		if (err != 0)
2516 			goto out;
2517 	}
2518 
2519 	if (featureflags & DMU_BACKUP_FEATURE_RAW) {
2520 		uint64_t ivset_guid = ancestor_zb->zbm_ivset_guid;
2521 		nvlist_t *keynvl = NULL;
2522 		ASSERT(os->os_encrypted);
2523 
2524 		err = dsl_crypto_populate_key_nvlist(os, ivset_guid,
2525 		    &keynvl);
2526 		if (err != 0) {
2527 			fnvlist_free(nvl);
2528 			goto out;
2529 		}
2530 
2531 		fnvlist_add_nvlist(nvl, "crypt_keydata", keynvl);
2532 		fnvlist_free(keynvl);
2533 	}
2534 
2535 	if (!nvlist_empty(nvl)) {
2536 		payload = fnvlist_pack(nvl, &payload_len);
2537 		drr->drr_payloadlen = payload_len;
2538 	}
2539 
2540 	fnvlist_free(nvl);
2541 	err = dump_record(&dsc, payload, payload_len);
2542 	fnvlist_pack_free(payload, payload_len);
2543 	if (err != 0) {
2544 		err = dsc.dsc_err;
2545 		goto out;
2546 	}
2547 
2548 	setup_to_thread(to_arg, os, dssp, fromtxg, dspp->rawok);
2549 	setup_from_thread(from_arg, from_rl, dssp);
2550 	setup_redact_list_thread(rlt_arg, dspp, redact_rl, dssp);
2551 	setup_merge_thread(smt_arg, dspp, from_arg, to_arg, rlt_arg, os);
2552 	setup_reader_thread(srt_arg, dspp, smt_arg, featureflags);
2553 
2554 	range = bqueue_dequeue(&srt_arg->q);
2555 	while (err == 0 && !range->eos_marker) {
2556 		err = do_dump(&dsc, range);
2557 		range = get_next_range(&srt_arg->q, range);
2558 		if (issig(JUSTLOOKING) && issig(FORREAL))
2559 			err = SET_ERROR(EINTR);
2560 	}
2561 
2562 	/*
2563 	 * If we hit an error or are interrupted, cancel our worker threads and
2564 	 * clear the queue of any pending records.  The threads will pass the
2565 	 * cancel up the tree of worker threads, and each one will clean up any
2566 	 * pending records before exiting.
2567 	 */
2568 	if (err != 0) {
2569 		srt_arg->cancel = B_TRUE;
2570 		while (!range->eos_marker) {
2571 			range = get_next_range(&srt_arg->q, range);
2572 		}
2573 	}
2574 	range_free(range);
2575 
2576 	bqueue_destroy(&srt_arg->q);
2577 	bqueue_destroy(&smt_arg->q);
2578 	if (dspp->redactbook != NULL)
2579 		bqueue_destroy(&rlt_arg->q);
2580 	bqueue_destroy(&to_arg->q);
2581 	bqueue_destroy(&from_arg->q);
2582 
2583 	if (err == 0 && srt_arg->error != 0)
2584 		err = srt_arg->error;
2585 
2586 	if (err != 0)
2587 		goto out;
2588 
2589 	if (dsc.dsc_pending_op != PENDING_NONE)
2590 		if (dump_record(&dsc, NULL, 0) != 0)
2591 			err = SET_ERROR(EINTR);
2592 
2593 	if (err != 0) {
2594 		if (err == EINTR && dsc.dsc_err != 0)
2595 			err = dsc.dsc_err;
2596 		goto out;
2597 	}
2598 
2599 	/*
2600 	 * Send the DRR_END record if this is not a saved stream.
2601 	 * Otherwise, the omitted DRR_END record will signal to
2602 	 * the receive side that the stream is incomplete.
2603 	 */
2604 	if (!dspp->savedok) {
2605 		memset(drr, 0, sizeof (dmu_replay_record_t));
2606 		drr->drr_type = DRR_END;
2607 		drr->drr_u.drr_end.drr_checksum = dsc.dsc_zc;
2608 		drr->drr_u.drr_end.drr_toguid = dsc.dsc_toguid;
2609 
2610 		if (dump_record(&dsc, NULL, 0) != 0)
2611 			err = dsc.dsc_err;
2612 	}
2613 out:
2614 	mutex_enter(&to_ds->ds_sendstream_lock);
2615 	list_remove(&to_ds->ds_sendstreams, dssp);
2616 	mutex_exit(&to_ds->ds_sendstream_lock);
2617 
2618 	VERIFY(err != 0 || (dsc.dsc_sent_begin &&
2619 	    (dsc.dsc_sent_end || dspp->savedok)));
2620 
2621 	kmem_free(drr, sizeof (dmu_replay_record_t));
2622 	kmem_free(dssp, sizeof (dmu_sendstatus_t));
2623 	kmem_free(from_arg, sizeof (*from_arg));
2624 	kmem_free(to_arg, sizeof (*to_arg));
2625 	kmem_free(rlt_arg, sizeof (*rlt_arg));
2626 	kmem_free(smt_arg, sizeof (*smt_arg));
2627 	kmem_free(srt_arg, sizeof (*srt_arg));
2628 
2629 	dsl_dataset_long_rele(to_ds, FTAG);
2630 	if (from_rl != NULL) {
2631 		dsl_redaction_list_long_rele(from_rl, FTAG);
2632 		dsl_redaction_list_rele(from_rl, FTAG);
2633 	}
2634 	if (redact_rl != NULL) {
2635 		dsl_redaction_list_long_rele(redact_rl, FTAG);
2636 		dsl_redaction_list_rele(redact_rl, FTAG);
2637 	}
2638 
2639 	return (err);
2640 }
2641 
2642 int
2643 dmu_send_obj(const char *pool, uint64_t tosnap, uint64_t fromsnap,
2644     boolean_t embedok, boolean_t large_block_ok, boolean_t compressok,
2645     boolean_t rawok, boolean_t savedok, int outfd, offset_t *off,
2646     dmu_send_outparams_t *dsop)
2647 {
2648 	int err;
2649 	dsl_dataset_t *fromds;
2650 	ds_hold_flags_t dsflags;
2651 	struct dmu_send_params dspp = {0};
2652 	dspp.embedok = embedok;
2653 	dspp.large_block_ok = large_block_ok;
2654 	dspp.compressok = compressok;
2655 	dspp.outfd = outfd;
2656 	dspp.off = off;
2657 	dspp.dso = dsop;
2658 	dspp.tag = FTAG;
2659 	dspp.rawok = rawok;
2660 	dspp.savedok = savedok;
2661 
2662 	dsflags = (rawok) ? DS_HOLD_FLAG_NONE : DS_HOLD_FLAG_DECRYPT;
2663 	err = dsl_pool_hold(pool, FTAG, &dspp.dp);
2664 	if (err != 0)
2665 		return (err);
2666 
2667 	err = dsl_dataset_hold_obj_flags(dspp.dp, tosnap, dsflags, FTAG,
2668 	    &dspp.to_ds);
2669 	if (err != 0) {
2670 		dsl_pool_rele(dspp.dp, FTAG);
2671 		return (err);
2672 	}
2673 
2674 	if (fromsnap != 0) {
2675 		err = dsl_dataset_hold_obj_flags(dspp.dp, fromsnap, dsflags,
2676 		    FTAG, &fromds);
2677 		if (err != 0) {
2678 			dsl_dataset_rele_flags(dspp.to_ds, dsflags, FTAG);
2679 			dsl_pool_rele(dspp.dp, FTAG);
2680 			return (err);
2681 		}
2682 		dspp.ancestor_zb.zbm_guid = dsl_dataset_phys(fromds)->ds_guid;
2683 		dspp.ancestor_zb.zbm_creation_txg =
2684 		    dsl_dataset_phys(fromds)->ds_creation_txg;
2685 		dspp.ancestor_zb.zbm_creation_time =
2686 		    dsl_dataset_phys(fromds)->ds_creation_time;
2687 
2688 		if (dsl_dataset_is_zapified(fromds)) {
2689 			(void) zap_lookup(dspp.dp->dp_meta_objset,
2690 			    fromds->ds_object, DS_FIELD_IVSET_GUID, 8, 1,
2691 			    &dspp.ancestor_zb.zbm_ivset_guid);
2692 		}
2693 
2694 		/* See dmu_send for the reasons behind this. */
2695 		uint64_t *fromredact;
2696 
2697 		if (!dsl_dataset_get_uint64_array_feature(fromds,
2698 		    SPA_FEATURE_REDACTED_DATASETS,
2699 		    &dspp.numfromredactsnaps,
2700 		    &fromredact)) {
2701 			dspp.numfromredactsnaps = NUM_SNAPS_NOT_REDACTED;
2702 		} else if (dspp.numfromredactsnaps > 0) {
2703 			uint64_t size = dspp.numfromredactsnaps *
2704 			    sizeof (uint64_t);
2705 			dspp.fromredactsnaps = kmem_zalloc(size, KM_SLEEP);
2706 			memcpy(dspp.fromredactsnaps, fromredact, size);
2707 		}
2708 
2709 		boolean_t is_before =
2710 		    dsl_dataset_is_before(dspp.to_ds, fromds, 0);
2711 		dspp.is_clone = (dspp.to_ds->ds_dir !=
2712 		    fromds->ds_dir);
2713 		dsl_dataset_rele(fromds, FTAG);
2714 		if (!is_before) {
2715 			dsl_pool_rele(dspp.dp, FTAG);
2716 			err = SET_ERROR(EXDEV);
2717 		} else {
2718 			err = dmu_send_impl(&dspp);
2719 		}
2720 	} else {
2721 		dspp.numfromredactsnaps = NUM_SNAPS_NOT_REDACTED;
2722 		err = dmu_send_impl(&dspp);
2723 	}
2724 	if (dspp.fromredactsnaps)
2725 		kmem_free(dspp.fromredactsnaps,
2726 		    dspp.numfromredactsnaps * sizeof (uint64_t));
2727 
2728 	dsl_dataset_rele(dspp.to_ds, FTAG);
2729 	return (err);
2730 }
2731 
2732 int
2733 dmu_send(const char *tosnap, const char *fromsnap, boolean_t embedok,
2734     boolean_t large_block_ok, boolean_t compressok, boolean_t rawok,
2735     boolean_t savedok, uint64_t resumeobj, uint64_t resumeoff,
2736     const char *redactbook, int outfd, offset_t *off,
2737     dmu_send_outparams_t *dsop)
2738 {
2739 	int err = 0;
2740 	ds_hold_flags_t dsflags;
2741 	boolean_t owned = B_FALSE;
2742 	dsl_dataset_t *fromds = NULL;
2743 	zfs_bookmark_phys_t book = {0};
2744 	struct dmu_send_params dspp = {0};
2745 
2746 	dsflags = (rawok) ? DS_HOLD_FLAG_NONE : DS_HOLD_FLAG_DECRYPT;
2747 	dspp.tosnap = tosnap;
2748 	dspp.embedok = embedok;
2749 	dspp.large_block_ok = large_block_ok;
2750 	dspp.compressok = compressok;
2751 	dspp.outfd = outfd;
2752 	dspp.off = off;
2753 	dspp.dso = dsop;
2754 	dspp.tag = FTAG;
2755 	dspp.resumeobj = resumeobj;
2756 	dspp.resumeoff = resumeoff;
2757 	dspp.rawok = rawok;
2758 	dspp.savedok = savedok;
2759 
2760 	if (fromsnap != NULL && strpbrk(fromsnap, "@#") == NULL)
2761 		return (SET_ERROR(EINVAL));
2762 
2763 	err = dsl_pool_hold(tosnap, FTAG, &dspp.dp);
2764 	if (err != 0)
2765 		return (err);
2766 
2767 	if (strchr(tosnap, '@') == NULL && spa_writeable(dspp.dp->dp_spa)) {
2768 		/*
2769 		 * We are sending a filesystem or volume.  Ensure
2770 		 * that it doesn't change by owning the dataset.
2771 		 */
2772 
2773 		if (savedok) {
2774 			/*
2775 			 * We are looking for the dataset that represents the
2776 			 * partially received send stream. If this stream was
2777 			 * received as a new snapshot of an existing dataset,
2778 			 * this will be saved in a hidden clone named
2779 			 * "<pool>/<dataset>/%recv". Otherwise, the stream
2780 			 * will be saved in the live dataset itself. In
2781 			 * either case we need to use dsl_dataset_own_force()
2782 			 * because the stream is marked as inconsistent,
2783 			 * which would normally make it unavailable to be
2784 			 * owned.
2785 			 */
2786 			char *name = kmem_asprintf("%s/%s", tosnap,
2787 			    recv_clone_name);
2788 			err = dsl_dataset_own_force(dspp.dp, name, dsflags,
2789 			    FTAG, &dspp.to_ds);
2790 			if (err == ENOENT) {
2791 				err = dsl_dataset_own_force(dspp.dp, tosnap,
2792 				    dsflags, FTAG, &dspp.to_ds);
2793 			}
2794 
2795 			if (err == 0) {
2796 				err = zap_lookup(dspp.dp->dp_meta_objset,
2797 				    dspp.to_ds->ds_object,
2798 				    DS_FIELD_RESUME_TOGUID, 8, 1,
2799 				    &dspp.saved_guid);
2800 			}
2801 
2802 			if (err == 0) {
2803 				err = zap_lookup(dspp.dp->dp_meta_objset,
2804 				    dspp.to_ds->ds_object,
2805 				    DS_FIELD_RESUME_TONAME, 1,
2806 				    sizeof (dspp.saved_toname),
2807 				    dspp.saved_toname);
2808 			}
2809 			if (err != 0)
2810 				dsl_dataset_disown(dspp.to_ds, dsflags, FTAG);
2811 
2812 			kmem_strfree(name);
2813 		} else {
2814 			err = dsl_dataset_own(dspp.dp, tosnap, dsflags,
2815 			    FTAG, &dspp.to_ds);
2816 		}
2817 		owned = B_TRUE;
2818 	} else {
2819 		err = dsl_dataset_hold_flags(dspp.dp, tosnap, dsflags, FTAG,
2820 		    &dspp.to_ds);
2821 	}
2822 
2823 	if (err != 0) {
2824 		dsl_pool_rele(dspp.dp, FTAG);
2825 		return (err);
2826 	}
2827 
2828 	if (redactbook != NULL) {
2829 		char path[ZFS_MAX_DATASET_NAME_LEN];
2830 		(void) strlcpy(path, tosnap, sizeof (path));
2831 		char *at = strchr(path, '@');
2832 		if (at == NULL) {
2833 			err = EINVAL;
2834 		} else {
2835 			(void) snprintf(at, sizeof (path) - (at - path), "#%s",
2836 			    redactbook);
2837 			err = dsl_bookmark_lookup(dspp.dp, path,
2838 			    NULL, &book);
2839 			dspp.redactbook = &book;
2840 		}
2841 	}
2842 
2843 	if (err != 0) {
2844 		dsl_pool_rele(dspp.dp, FTAG);
2845 		if (owned)
2846 			dsl_dataset_disown(dspp.to_ds, dsflags, FTAG);
2847 		else
2848 			dsl_dataset_rele_flags(dspp.to_ds, dsflags, FTAG);
2849 		return (err);
2850 	}
2851 
2852 	if (fromsnap != NULL) {
2853 		zfs_bookmark_phys_t *zb = &dspp.ancestor_zb;
2854 		int fsnamelen;
2855 		if (strpbrk(tosnap, "@#") != NULL)
2856 			fsnamelen = strpbrk(tosnap, "@#") - tosnap;
2857 		else
2858 			fsnamelen = strlen(tosnap);
2859 
2860 		/*
2861 		 * If the fromsnap is in a different filesystem, then
2862 		 * mark the send stream as a clone.
2863 		 */
2864 		if (strncmp(tosnap, fromsnap, fsnamelen) != 0 ||
2865 		    (fromsnap[fsnamelen] != '@' &&
2866 		    fromsnap[fsnamelen] != '#')) {
2867 			dspp.is_clone = B_TRUE;
2868 		}
2869 
2870 		if (strchr(fromsnap, '@') != NULL) {
2871 			err = dsl_dataset_hold(dspp.dp, fromsnap, FTAG,
2872 			    &fromds);
2873 
2874 			if (err != 0) {
2875 				ASSERT3P(fromds, ==, NULL);
2876 			} else {
2877 				/*
2878 				 * We need to make a deep copy of the redact
2879 				 * snapshots of the from snapshot, because the
2880 				 * array will be freed when we evict from_ds.
2881 				 */
2882 				uint64_t *fromredact;
2883 				if (!dsl_dataset_get_uint64_array_feature(
2884 				    fromds, SPA_FEATURE_REDACTED_DATASETS,
2885 				    &dspp.numfromredactsnaps,
2886 				    &fromredact)) {
2887 					dspp.numfromredactsnaps =
2888 					    NUM_SNAPS_NOT_REDACTED;
2889 				} else if (dspp.numfromredactsnaps > 0) {
2890 					uint64_t size =
2891 					    dspp.numfromredactsnaps *
2892 					    sizeof (uint64_t);
2893 					dspp.fromredactsnaps = kmem_zalloc(size,
2894 					    KM_SLEEP);
2895 					memcpy(dspp.fromredactsnaps, fromredact,
2896 					    size);
2897 				}
2898 				if (!dsl_dataset_is_before(dspp.to_ds, fromds,
2899 				    0)) {
2900 					err = SET_ERROR(EXDEV);
2901 				} else {
2902 					zb->zbm_creation_txg =
2903 					    dsl_dataset_phys(fromds)->
2904 					    ds_creation_txg;
2905 					zb->zbm_creation_time =
2906 					    dsl_dataset_phys(fromds)->
2907 					    ds_creation_time;
2908 					zb->zbm_guid =
2909 					    dsl_dataset_phys(fromds)->ds_guid;
2910 					zb->zbm_redaction_obj = 0;
2911 
2912 					if (dsl_dataset_is_zapified(fromds)) {
2913 						(void) zap_lookup(
2914 						    dspp.dp->dp_meta_objset,
2915 						    fromds->ds_object,
2916 						    DS_FIELD_IVSET_GUID, 8, 1,
2917 						    &zb->zbm_ivset_guid);
2918 					}
2919 				}
2920 				dsl_dataset_rele(fromds, FTAG);
2921 			}
2922 		} else {
2923 			dspp.numfromredactsnaps = NUM_SNAPS_NOT_REDACTED;
2924 			err = dsl_bookmark_lookup(dspp.dp, fromsnap, dspp.to_ds,
2925 			    zb);
2926 			if (err == EXDEV && zb->zbm_redaction_obj != 0 &&
2927 			    zb->zbm_guid ==
2928 			    dsl_dataset_phys(dspp.to_ds)->ds_guid)
2929 				err = 0;
2930 		}
2931 
2932 		if (err == 0) {
2933 			/* dmu_send_impl will call dsl_pool_rele for us. */
2934 			err = dmu_send_impl(&dspp);
2935 		} else {
2936 			if (dspp.fromredactsnaps)
2937 				kmem_free(dspp.fromredactsnaps,
2938 				    dspp.numfromredactsnaps *
2939 				    sizeof (uint64_t));
2940 			dsl_pool_rele(dspp.dp, FTAG);
2941 		}
2942 	} else {
2943 		dspp.numfromredactsnaps = NUM_SNAPS_NOT_REDACTED;
2944 		err = dmu_send_impl(&dspp);
2945 	}
2946 	if (owned)
2947 		dsl_dataset_disown(dspp.to_ds, dsflags, FTAG);
2948 	else
2949 		dsl_dataset_rele_flags(dspp.to_ds, dsflags, FTAG);
2950 	return (err);
2951 }
2952 
2953 static int
2954 dmu_adjust_send_estimate_for_indirects(dsl_dataset_t *ds, uint64_t uncompressed,
2955     uint64_t compressed, boolean_t stream_compressed, uint64_t *sizep)
2956 {
2957 	int err = 0;
2958 	uint64_t size;
2959 	/*
2960 	 * Assume that space (both on-disk and in-stream) is dominated by
2961 	 * data.  We will adjust for indirect blocks and the copies property,
2962 	 * but ignore per-object space used (eg, dnodes and DRR_OBJECT records).
2963 	 */
2964 
2965 	uint64_t recordsize;
2966 	uint64_t record_count;
2967 	objset_t *os;
2968 	VERIFY0(dmu_objset_from_ds(ds, &os));
2969 
2970 	/* Assume all (uncompressed) blocks are recordsize. */
2971 	if (zfs_override_estimate_recordsize != 0) {
2972 		recordsize = zfs_override_estimate_recordsize;
2973 	} else if (os->os_phys->os_type == DMU_OST_ZVOL) {
2974 		err = dsl_prop_get_int_ds(ds,
2975 		    zfs_prop_to_name(ZFS_PROP_VOLBLOCKSIZE), &recordsize);
2976 	} else {
2977 		err = dsl_prop_get_int_ds(ds,
2978 		    zfs_prop_to_name(ZFS_PROP_RECORDSIZE), &recordsize);
2979 	}
2980 	if (err != 0)
2981 		return (err);
2982 	record_count = uncompressed / recordsize;
2983 
2984 	/*
2985 	 * If we're estimating a send size for a compressed stream, use the
2986 	 * compressed data size to estimate the stream size. Otherwise, use the
2987 	 * uncompressed data size.
2988 	 */
2989 	size = stream_compressed ? compressed : uncompressed;
2990 
2991 	/*
2992 	 * Subtract out approximate space used by indirect blocks.
2993 	 * Assume most space is used by data blocks (non-indirect, non-dnode).
2994 	 * Assume no ditto blocks or internal fragmentation.
2995 	 *
2996 	 * Therefore, space used by indirect blocks is sizeof(blkptr_t) per
2997 	 * block.
2998 	 */
2999 	size -= record_count * sizeof (blkptr_t);
3000 
3001 	/* Add in the space for the record associated with each block. */
3002 	size += record_count * sizeof (dmu_replay_record_t);
3003 
3004 	*sizep = size;
3005 
3006 	return (0);
3007 }
3008 
3009 int
3010 dmu_send_estimate_fast(dsl_dataset_t *origds, dsl_dataset_t *fromds,
3011     zfs_bookmark_phys_t *frombook, boolean_t stream_compressed,
3012     boolean_t saved, uint64_t *sizep)
3013 {
3014 	int err;
3015 	dsl_dataset_t *ds = origds;
3016 	uint64_t uncomp, comp;
3017 
3018 	ASSERT(dsl_pool_config_held(origds->ds_dir->dd_pool));
3019 	ASSERT(fromds == NULL || frombook == NULL);
3020 
3021 	/*
3022 	 * If this is a saved send we may actually be sending
3023 	 * from the %recv clone used for resuming.
3024 	 */
3025 	if (saved) {
3026 		objset_t *mos = origds->ds_dir->dd_pool->dp_meta_objset;
3027 		uint64_t guid;
3028 		char dsname[ZFS_MAX_DATASET_NAME_LEN + 6];
3029 
3030 		dsl_dataset_name(origds, dsname);
3031 		(void) strcat(dsname, "/");
3032 		(void) strlcat(dsname, recv_clone_name, sizeof (dsname));
3033 
3034 		err = dsl_dataset_hold(origds->ds_dir->dd_pool,
3035 		    dsname, FTAG, &ds);
3036 		if (err != ENOENT && err != 0) {
3037 			return (err);
3038 		} else if (err == ENOENT) {
3039 			ds = origds;
3040 		}
3041 
3042 		/* check that this dataset has partially received data */
3043 		err = zap_lookup(mos, ds->ds_object,
3044 		    DS_FIELD_RESUME_TOGUID, 8, 1, &guid);
3045 		if (err != 0) {
3046 			err = SET_ERROR(err == ENOENT ? EINVAL : err);
3047 			goto out;
3048 		}
3049 
3050 		err = zap_lookup(mos, ds->ds_object,
3051 		    DS_FIELD_RESUME_TONAME, 1, sizeof (dsname), dsname);
3052 		if (err != 0) {
3053 			err = SET_ERROR(err == ENOENT ? EINVAL : err);
3054 			goto out;
3055 		}
3056 	}
3057 
3058 	/* tosnap must be a snapshot or the target of a saved send */
3059 	if (!ds->ds_is_snapshot && ds == origds)
3060 		return (SET_ERROR(EINVAL));
3061 
3062 	if (fromds != NULL) {
3063 		uint64_t used;
3064 		if (!fromds->ds_is_snapshot) {
3065 			err = SET_ERROR(EINVAL);
3066 			goto out;
3067 		}
3068 
3069 		if (!dsl_dataset_is_before(ds, fromds, 0)) {
3070 			err = SET_ERROR(EXDEV);
3071 			goto out;
3072 		}
3073 
3074 		err = dsl_dataset_space_written(fromds, ds, &used, &comp,
3075 		    &uncomp);
3076 		if (err != 0)
3077 			goto out;
3078 	} else if (frombook != NULL) {
3079 		uint64_t used;
3080 		err = dsl_dataset_space_written_bookmark(frombook, ds, &used,
3081 		    &comp, &uncomp);
3082 		if (err != 0)
3083 			goto out;
3084 	} else {
3085 		uncomp = dsl_dataset_phys(ds)->ds_uncompressed_bytes;
3086 		comp = dsl_dataset_phys(ds)->ds_compressed_bytes;
3087 	}
3088 
3089 	err = dmu_adjust_send_estimate_for_indirects(ds, uncomp, comp,
3090 	    stream_compressed, sizep);
3091 	/*
3092 	 * Add the size of the BEGIN and END records to the estimate.
3093 	 */
3094 	*sizep += 2 * sizeof (dmu_replay_record_t);
3095 
3096 out:
3097 	if (ds != origds)
3098 		dsl_dataset_rele(ds, FTAG);
3099 	return (err);
3100 }
3101 
3102 ZFS_MODULE_PARAM(zfs_send, zfs_send_, corrupt_data, INT, ZMOD_RW,
3103 	"Allow sending corrupt data");
3104 
3105 ZFS_MODULE_PARAM(zfs_send, zfs_send_, queue_length, UINT, ZMOD_RW,
3106 	"Maximum send queue length");
3107 
3108 ZFS_MODULE_PARAM(zfs_send, zfs_send_, unmodified_spill_blocks, INT, ZMOD_RW,
3109 	"Send unmodified spill blocks");
3110 
3111 ZFS_MODULE_PARAM(zfs_send, zfs_send_, no_prefetch_queue_length, UINT, ZMOD_RW,
3112 	"Maximum send queue length for non-prefetch queues");
3113 
3114 ZFS_MODULE_PARAM(zfs_send, zfs_send_, queue_ff, UINT, ZMOD_RW,
3115 	"Send queue fill fraction");
3116 
3117 ZFS_MODULE_PARAM(zfs_send, zfs_send_, no_prefetch_queue_ff, UINT, ZMOD_RW,
3118 	"Send queue fill fraction for non-prefetch queues");
3119 
3120 ZFS_MODULE_PARAM(zfs_send, zfs_, override_estimate_recordsize, UINT, ZMOD_RW,
3121 	"Override block size estimate with fixed size");
3122