xref: /illumos-gate/usr/src/uts/common/fs/zfs/dmu_send.c (revision 946342a2)
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 http://www.opensolaris.org/os/licensing.
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, 2015 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) 2014 Integros [integros.com]
29  */
30 
31 #include <sys/dmu.h>
32 #include <sys/dmu_impl.h>
33 #include <sys/dmu_tx.h>
34 #include <sys/dbuf.h>
35 #include <sys/dnode.h>
36 #include <sys/zfs_context.h>
37 #include <sys/dmu_objset.h>
38 #include <sys/dmu_traverse.h>
39 #include <sys/dsl_dataset.h>
40 #include <sys/dsl_dir.h>
41 #include <sys/dsl_prop.h>
42 #include <sys/dsl_pool.h>
43 #include <sys/dsl_synctask.h>
44 #include <sys/zfs_ioctl.h>
45 #include <sys/zap.h>
46 #include <sys/zio_checksum.h>
47 #include <sys/zfs_znode.h>
48 #include <zfs_fletcher.h>
49 #include <sys/avl.h>
50 #include <sys/ddt.h>
51 #include <sys/zfs_onexit.h>
52 #include <sys/dmu_send.h>
53 #include <sys/dsl_destroy.h>
54 #include <sys/blkptr.h>
55 #include <sys/dsl_bookmark.h>
56 #include <sys/zfeature.h>
57 #include <sys/bqueue.h>
58 
59 /* Set this tunable to TRUE to replace corrupt data with 0x2f5baddb10c */
60 int zfs_send_corrupt_data = B_FALSE;
61 int zfs_send_queue_length = 16 * 1024 * 1024;
62 int zfs_recv_queue_length = 16 * 1024 * 1024;
63 /* Set this tunable to FALSE to disable setting of DRR_FLAG_FREERECORDS */
64 int zfs_send_set_freerecords_bit = B_TRUE;
65 
66 static char *dmu_recv_tag = "dmu_recv_tag";
67 const char *recv_clone_name = "%recv";
68 
69 /*
70  * Use this to override the recordsize calculation for fast zfs send estimates.
71  */
72 uint64_t zfs_override_estimate_recordsize = 0;
73 
74 #define	BP_SPAN(datablkszsec, indblkshift, level) \
75 	(((uint64_t)datablkszsec) << (SPA_MINBLOCKSHIFT + \
76 	(level) * (indblkshift - SPA_BLKPTRSHIFT)))
77 
78 static void byteswap_record(dmu_replay_record_t *drr);
79 
80 struct send_thread_arg {
81 	bqueue_t	q;
82 	dsl_dataset_t	*ds;		/* Dataset to traverse */
83 	uint64_t	fromtxg;	/* Traverse from this txg */
84 	int		flags;		/* flags to pass to traverse_dataset */
85 	int		error_code;
86 	boolean_t	cancel;
87 	zbookmark_phys_t resume;
88 };
89 
90 struct send_block_record {
91 	boolean_t		eos_marker; /* Marks the end of the stream */
92 	blkptr_t		bp;
93 	zbookmark_phys_t	zb;
94 	uint8_t			indblkshift;
95 	uint16_t		datablkszsec;
96 	bqueue_node_t		ln;
97 };
98 
99 static int
100 dump_bytes(dmu_sendarg_t *dsp, void *buf, int len)
101 {
102 	dsl_dataset_t *ds = dmu_objset_ds(dsp->dsa_os);
103 	ssize_t resid; /* have to get resid to get detailed errno */
104 
105 	/*
106 	 * The code does not rely on this (len being a multiple of 8).  We keep
107 	 * this assertion because of the corresponding assertion in
108 	 * receive_read().  Keeping this assertion ensures that we do not
109 	 * inadvertently break backwards compatibility (causing the assertion
110 	 * in receive_read() to trigger on old software).
111 	 *
112 	 * Removing the assertions could be rolled into a new feature that uses
113 	 * data that isn't 8-byte aligned; if the assertions were removed, a
114 	 * feature flag would have to be added.
115 	 */
116 
117 	ASSERT0(len % 8);
118 
119 	dsp->dsa_err = vn_rdwr(UIO_WRITE, dsp->dsa_vp,
120 	    (caddr_t)buf, len,
121 	    0, UIO_SYSSPACE, FAPPEND, RLIM64_INFINITY, CRED(), &resid);
122 
123 	mutex_enter(&ds->ds_sendstream_lock);
124 	*dsp->dsa_off += len;
125 	mutex_exit(&ds->ds_sendstream_lock);
126 
127 	return (dsp->dsa_err);
128 }
129 
130 /*
131  * For all record types except BEGIN, fill in the checksum (overlaid in
132  * drr_u.drr_checksum.drr_checksum).  The checksum verifies everything
133  * up to the start of the checksum itself.
134  */
135 static int
136 dump_record(dmu_sendarg_t *dsp, void *payload, int payload_len)
137 {
138 	ASSERT3U(offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
139 	    ==, sizeof (dmu_replay_record_t) - sizeof (zio_cksum_t));
140 	(void) fletcher_4_incremental_native(dsp->dsa_drr,
141 	    offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
142 	    &dsp->dsa_zc);
143 	if (dsp->dsa_drr->drr_type == DRR_BEGIN) {
144 		dsp->dsa_sent_begin = B_TRUE;
145 	} else {
146 		ASSERT(ZIO_CHECKSUM_IS_ZERO(&dsp->dsa_drr->drr_u.
147 		    drr_checksum.drr_checksum));
148 		dsp->dsa_drr->drr_u.drr_checksum.drr_checksum = dsp->dsa_zc;
149 	}
150 	if (dsp->dsa_drr->drr_type == DRR_END) {
151 		dsp->dsa_sent_end = B_TRUE;
152 	}
153 	(void) fletcher_4_incremental_native(&dsp->dsa_drr->
154 	    drr_u.drr_checksum.drr_checksum,
155 	    sizeof (zio_cksum_t), &dsp->dsa_zc);
156 	if (dump_bytes(dsp, dsp->dsa_drr, sizeof (dmu_replay_record_t)) != 0)
157 		return (SET_ERROR(EINTR));
158 	if (payload_len != 0) {
159 		(void) fletcher_4_incremental_native(payload, payload_len,
160 		    &dsp->dsa_zc);
161 		if (dump_bytes(dsp, payload, payload_len) != 0)
162 			return (SET_ERROR(EINTR));
163 	}
164 	return (0);
165 }
166 
167 /*
168  * Fill in the drr_free struct, or perform aggregation if the previous record is
169  * also a free record, and the two are adjacent.
170  *
171  * Note that we send free records even for a full send, because we want to be
172  * able to receive a full send as a clone, which requires a list of all the free
173  * and freeobject records that were generated on the source.
174  */
175 static int
176 dump_free(dmu_sendarg_t *dsp, uint64_t object, uint64_t offset,
177     uint64_t length)
178 {
179 	struct drr_free *drrf = &(dsp->dsa_drr->drr_u.drr_free);
180 
181 	/*
182 	 * When we receive a free record, dbuf_free_range() assumes
183 	 * that the receiving system doesn't have any dbufs in the range
184 	 * being freed.  This is always true because there is a one-record
185 	 * constraint: we only send one WRITE record for any given
186 	 * object,offset.  We know that the one-record constraint is
187 	 * true because we always send data in increasing order by
188 	 * object,offset.
189 	 *
190 	 * If the increasing-order constraint ever changes, we should find
191 	 * another way to assert that the one-record constraint is still
192 	 * satisfied.
193 	 */
194 	ASSERT(object > dsp->dsa_last_data_object ||
195 	    (object == dsp->dsa_last_data_object &&
196 	    offset > dsp->dsa_last_data_offset));
197 
198 	if (length != -1ULL && offset + length < offset)
199 		length = -1ULL;
200 
201 	/*
202 	 * If there is a pending op, but it's not PENDING_FREE, push it out,
203 	 * since free block aggregation can only be done for blocks of the
204 	 * same type (i.e., DRR_FREE records can only be aggregated with
205 	 * other DRR_FREE records.  DRR_FREEOBJECTS records can only be
206 	 * aggregated with other DRR_FREEOBJECTS records.
207 	 */
208 	if (dsp->dsa_pending_op != PENDING_NONE &&
209 	    dsp->dsa_pending_op != PENDING_FREE) {
210 		if (dump_record(dsp, NULL, 0) != 0)
211 			return (SET_ERROR(EINTR));
212 		dsp->dsa_pending_op = PENDING_NONE;
213 	}
214 
215 	if (dsp->dsa_pending_op == PENDING_FREE) {
216 		/*
217 		 * There should never be a PENDING_FREE if length is -1
218 		 * (because dump_dnode is the only place where this
219 		 * function is called with a -1, and only after flushing
220 		 * any pending record).
221 		 */
222 		ASSERT(length != -1ULL);
223 		/*
224 		 * Check to see whether this free block can be aggregated
225 		 * with pending one.
226 		 */
227 		if (drrf->drr_object == object && drrf->drr_offset +
228 		    drrf->drr_length == offset) {
229 			drrf->drr_length += length;
230 			return (0);
231 		} else {
232 			/* not a continuation.  Push out pending record */
233 			if (dump_record(dsp, NULL, 0) != 0)
234 				return (SET_ERROR(EINTR));
235 			dsp->dsa_pending_op = PENDING_NONE;
236 		}
237 	}
238 	/* create a FREE record and make it pending */
239 	bzero(dsp->dsa_drr, sizeof (dmu_replay_record_t));
240 	dsp->dsa_drr->drr_type = DRR_FREE;
241 	drrf->drr_object = object;
242 	drrf->drr_offset = offset;
243 	drrf->drr_length = length;
244 	drrf->drr_toguid = dsp->dsa_toguid;
245 	if (length == -1ULL) {
246 		if (dump_record(dsp, NULL, 0) != 0)
247 			return (SET_ERROR(EINTR));
248 	} else {
249 		dsp->dsa_pending_op = PENDING_FREE;
250 	}
251 
252 	return (0);
253 }
254 
255 static int
256 dump_write(dmu_sendarg_t *dsp, dmu_object_type_t type,
257     uint64_t object, uint64_t offset, int lsize, int psize, const blkptr_t *bp,
258     void *data)
259 {
260 	uint64_t payload_size;
261 	struct drr_write *drrw = &(dsp->dsa_drr->drr_u.drr_write);
262 
263 	/*
264 	 * We send data in increasing object, offset order.
265 	 * See comment in dump_free() for details.
266 	 */
267 	ASSERT(object > dsp->dsa_last_data_object ||
268 	    (object == dsp->dsa_last_data_object &&
269 	    offset > dsp->dsa_last_data_offset));
270 	dsp->dsa_last_data_object = object;
271 	dsp->dsa_last_data_offset = offset + lsize - 1;
272 
273 	/*
274 	 * If there is any kind of pending aggregation (currently either
275 	 * a grouping of free objects or free blocks), push it out to
276 	 * the stream, since aggregation can't be done across operations
277 	 * of different types.
278 	 */
279 	if (dsp->dsa_pending_op != PENDING_NONE) {
280 		if (dump_record(dsp, NULL, 0) != 0)
281 			return (SET_ERROR(EINTR));
282 		dsp->dsa_pending_op = PENDING_NONE;
283 	}
284 	/* write a WRITE record */
285 	bzero(dsp->dsa_drr, sizeof (dmu_replay_record_t));
286 	dsp->dsa_drr->drr_type = DRR_WRITE;
287 	drrw->drr_object = object;
288 	drrw->drr_type = type;
289 	drrw->drr_offset = offset;
290 	drrw->drr_toguid = dsp->dsa_toguid;
291 	drrw->drr_logical_size = lsize;
292 
293 	/* only set the compression fields if the buf is compressed */
294 	if (lsize != psize) {
295 		ASSERT(dsp->dsa_featureflags & DMU_BACKUP_FEATURE_COMPRESSED);
296 		ASSERT(!BP_IS_EMBEDDED(bp));
297 		ASSERT(!BP_SHOULD_BYTESWAP(bp));
298 		ASSERT(!DMU_OT_IS_METADATA(BP_GET_TYPE(bp)));
299 		ASSERT3U(BP_GET_COMPRESS(bp), !=, ZIO_COMPRESS_OFF);
300 		ASSERT3S(psize, >, 0);
301 		ASSERT3S(lsize, >=, psize);
302 
303 		drrw->drr_compressiontype = BP_GET_COMPRESS(bp);
304 		drrw->drr_compressed_size = psize;
305 		payload_size = drrw->drr_compressed_size;
306 	} else {
307 		payload_size = drrw->drr_logical_size;
308 	}
309 
310 	if (bp == NULL || BP_IS_EMBEDDED(bp)) {
311 		/*
312 		 * There's no pre-computed checksum for partial-block
313 		 * writes or embedded BP's, so (like
314 		 * fletcher4-checkummed blocks) userland will have to
315 		 * compute a dedup-capable checksum itself.
316 		 */
317 		drrw->drr_checksumtype = ZIO_CHECKSUM_OFF;
318 	} else {
319 		drrw->drr_checksumtype = BP_GET_CHECKSUM(bp);
320 		if (zio_checksum_table[drrw->drr_checksumtype].ci_flags &
321 		    ZCHECKSUM_FLAG_DEDUP)
322 			drrw->drr_checksumflags |= DRR_CHECKSUM_DEDUP;
323 		DDK_SET_LSIZE(&drrw->drr_key, BP_GET_LSIZE(bp));
324 		DDK_SET_PSIZE(&drrw->drr_key, BP_GET_PSIZE(bp));
325 		DDK_SET_COMPRESS(&drrw->drr_key, BP_GET_COMPRESS(bp));
326 		drrw->drr_key.ddk_cksum = bp->blk_cksum;
327 	}
328 
329 	if (dump_record(dsp, data, payload_size) != 0)
330 		return (SET_ERROR(EINTR));
331 	return (0);
332 }
333 
334 static int
335 dump_write_embedded(dmu_sendarg_t *dsp, uint64_t object, uint64_t offset,
336     int blksz, const blkptr_t *bp)
337 {
338 	char buf[BPE_PAYLOAD_SIZE];
339 	struct drr_write_embedded *drrw =
340 	    &(dsp->dsa_drr->drr_u.drr_write_embedded);
341 
342 	if (dsp->dsa_pending_op != PENDING_NONE) {
343 		if (dump_record(dsp, NULL, 0) != 0)
344 			return (EINTR);
345 		dsp->dsa_pending_op = PENDING_NONE;
346 	}
347 
348 	ASSERT(BP_IS_EMBEDDED(bp));
349 
350 	bzero(dsp->dsa_drr, sizeof (dmu_replay_record_t));
351 	dsp->dsa_drr->drr_type = DRR_WRITE_EMBEDDED;
352 	drrw->drr_object = object;
353 	drrw->drr_offset = offset;
354 	drrw->drr_length = blksz;
355 	drrw->drr_toguid = dsp->dsa_toguid;
356 	drrw->drr_compression = BP_GET_COMPRESS(bp);
357 	drrw->drr_etype = BPE_GET_ETYPE(bp);
358 	drrw->drr_lsize = BPE_GET_LSIZE(bp);
359 	drrw->drr_psize = BPE_GET_PSIZE(bp);
360 
361 	decode_embedded_bp_compressed(bp, buf);
362 
363 	if (dump_record(dsp, buf, P2ROUNDUP(drrw->drr_psize, 8)) != 0)
364 		return (EINTR);
365 	return (0);
366 }
367 
368 static int
369 dump_spill(dmu_sendarg_t *dsp, uint64_t object, int blksz, void *data)
370 {
371 	struct drr_spill *drrs = &(dsp->dsa_drr->drr_u.drr_spill);
372 
373 	if (dsp->dsa_pending_op != PENDING_NONE) {
374 		if (dump_record(dsp, NULL, 0) != 0)
375 			return (SET_ERROR(EINTR));
376 		dsp->dsa_pending_op = PENDING_NONE;
377 	}
378 
379 	/* write a SPILL record */
380 	bzero(dsp->dsa_drr, sizeof (dmu_replay_record_t));
381 	dsp->dsa_drr->drr_type = DRR_SPILL;
382 	drrs->drr_object = object;
383 	drrs->drr_length = blksz;
384 	drrs->drr_toguid = dsp->dsa_toguid;
385 
386 	if (dump_record(dsp, data, blksz) != 0)
387 		return (SET_ERROR(EINTR));
388 	return (0);
389 }
390 
391 static int
392 dump_freeobjects(dmu_sendarg_t *dsp, uint64_t firstobj, uint64_t numobjs)
393 {
394 	struct drr_freeobjects *drrfo = &(dsp->dsa_drr->drr_u.drr_freeobjects);
395 
396 	/*
397 	 * If there is a pending op, but it's not PENDING_FREEOBJECTS,
398 	 * push it out, since free block aggregation can only be done for
399 	 * blocks of the same type (i.e., DRR_FREE records can only be
400 	 * aggregated with other DRR_FREE records.  DRR_FREEOBJECTS records
401 	 * can only be aggregated with other DRR_FREEOBJECTS records.
402 	 */
403 	if (dsp->dsa_pending_op != PENDING_NONE &&
404 	    dsp->dsa_pending_op != PENDING_FREEOBJECTS) {
405 		if (dump_record(dsp, NULL, 0) != 0)
406 			return (SET_ERROR(EINTR));
407 		dsp->dsa_pending_op = PENDING_NONE;
408 	}
409 	if (dsp->dsa_pending_op == PENDING_FREEOBJECTS) {
410 		/*
411 		 * See whether this free object array can be aggregated
412 		 * with pending one
413 		 */
414 		if (drrfo->drr_firstobj + drrfo->drr_numobjs == firstobj) {
415 			drrfo->drr_numobjs += numobjs;
416 			return (0);
417 		} else {
418 			/* can't be aggregated.  Push out pending record */
419 			if (dump_record(dsp, NULL, 0) != 0)
420 				return (SET_ERROR(EINTR));
421 			dsp->dsa_pending_op = PENDING_NONE;
422 		}
423 	}
424 
425 	/* write a FREEOBJECTS record */
426 	bzero(dsp->dsa_drr, sizeof (dmu_replay_record_t));
427 	dsp->dsa_drr->drr_type = DRR_FREEOBJECTS;
428 	drrfo->drr_firstobj = firstobj;
429 	drrfo->drr_numobjs = numobjs;
430 	drrfo->drr_toguid = dsp->dsa_toguid;
431 
432 	dsp->dsa_pending_op = PENDING_FREEOBJECTS;
433 
434 	return (0);
435 }
436 
437 static int
438 dump_dnode(dmu_sendarg_t *dsp, uint64_t object, dnode_phys_t *dnp)
439 {
440 	struct drr_object *drro = &(dsp->dsa_drr->drr_u.drr_object);
441 
442 	if (object < dsp->dsa_resume_object) {
443 		/*
444 		 * Note: when resuming, we will visit all the dnodes in
445 		 * the block of dnodes that we are resuming from.  In
446 		 * this case it's unnecessary to send the dnodes prior to
447 		 * the one we are resuming from.  We should be at most one
448 		 * block's worth of dnodes behind the resume point.
449 		 */
450 		ASSERT3U(dsp->dsa_resume_object - object, <,
451 		    1 << (DNODE_BLOCK_SHIFT - DNODE_SHIFT));
452 		return (0);
453 	}
454 
455 	if (dnp == NULL || dnp->dn_type == DMU_OT_NONE)
456 		return (dump_freeobjects(dsp, object, 1));
457 
458 	if (dsp->dsa_pending_op != PENDING_NONE) {
459 		if (dump_record(dsp, NULL, 0) != 0)
460 			return (SET_ERROR(EINTR));
461 		dsp->dsa_pending_op = PENDING_NONE;
462 	}
463 
464 	/* write an OBJECT record */
465 	bzero(dsp->dsa_drr, sizeof (dmu_replay_record_t));
466 	dsp->dsa_drr->drr_type = DRR_OBJECT;
467 	drro->drr_object = object;
468 	drro->drr_type = dnp->dn_type;
469 	drro->drr_bonustype = dnp->dn_bonustype;
470 	drro->drr_blksz = dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT;
471 	drro->drr_bonuslen = dnp->dn_bonuslen;
472 	drro->drr_dn_slots = dnp->dn_extra_slots + 1;
473 	drro->drr_checksumtype = dnp->dn_checksum;
474 	drro->drr_compress = dnp->dn_compress;
475 	drro->drr_toguid = dsp->dsa_toguid;
476 
477 	if (!(dsp->dsa_featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS) &&
478 	    drro->drr_blksz > SPA_OLD_MAXBLOCKSIZE)
479 		drro->drr_blksz = SPA_OLD_MAXBLOCKSIZE;
480 
481 	if (dump_record(dsp, DN_BONUS(dnp),
482 	    P2ROUNDUP(dnp->dn_bonuslen, 8)) != 0) {
483 		return (SET_ERROR(EINTR));
484 	}
485 
486 	/* Free anything past the end of the file. */
487 	if (dump_free(dsp, object, (dnp->dn_maxblkid + 1) *
488 	    (dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT), -1ULL) != 0)
489 		return (SET_ERROR(EINTR));
490 	if (dsp->dsa_err != 0)
491 		return (SET_ERROR(EINTR));
492 	return (0);
493 }
494 
495 static boolean_t
496 backup_do_embed(dmu_sendarg_t *dsp, const blkptr_t *bp)
497 {
498 	if (!BP_IS_EMBEDDED(bp))
499 		return (B_FALSE);
500 
501 	/*
502 	 * Compression function must be legacy, or explicitly enabled.
503 	 */
504 	if ((BP_GET_COMPRESS(bp) >= ZIO_COMPRESS_LEGACY_FUNCTIONS &&
505 	    !(dsp->dsa_featureflags & DMU_BACKUP_FEATURE_LZ4)))
506 		return (B_FALSE);
507 
508 	/*
509 	 * Embed type must be explicitly enabled.
510 	 */
511 	switch (BPE_GET_ETYPE(bp)) {
512 	case BP_EMBEDDED_TYPE_DATA:
513 		if (dsp->dsa_featureflags & DMU_BACKUP_FEATURE_EMBED_DATA)
514 			return (B_TRUE);
515 		break;
516 	default:
517 		return (B_FALSE);
518 	}
519 	return (B_FALSE);
520 }
521 
522 /*
523  * This is the callback function to traverse_dataset that acts as the worker
524  * thread for dmu_send_impl.
525  */
526 /*ARGSUSED*/
527 static int
528 send_cb(spa_t *spa, zilog_t *zilog, const blkptr_t *bp,
529     const zbookmark_phys_t *zb, const struct dnode_phys *dnp, void *arg)
530 {
531 	struct send_thread_arg *sta = arg;
532 	struct send_block_record *record;
533 	uint64_t record_size;
534 	int err = 0;
535 
536 	ASSERT(zb->zb_object == DMU_META_DNODE_OBJECT ||
537 	    zb->zb_object >= sta->resume.zb_object);
538 
539 	if (sta->cancel)
540 		return (SET_ERROR(EINTR));
541 
542 	if (bp == NULL) {
543 		ASSERT3U(zb->zb_level, ==, ZB_DNODE_LEVEL);
544 		return (0);
545 	} else if (zb->zb_level < 0) {
546 		return (0);
547 	}
548 
549 	record = kmem_zalloc(sizeof (struct send_block_record), KM_SLEEP);
550 	record->eos_marker = B_FALSE;
551 	record->bp = *bp;
552 	record->zb = *zb;
553 	record->indblkshift = dnp->dn_indblkshift;
554 	record->datablkszsec = dnp->dn_datablkszsec;
555 	record_size = dnp->dn_datablkszsec << SPA_MINBLOCKSHIFT;
556 	bqueue_enqueue(&sta->q, record, record_size);
557 
558 	return (err);
559 }
560 
561 /*
562  * This function kicks off the traverse_dataset.  It also handles setting the
563  * error code of the thread in case something goes wrong, and pushes the End of
564  * Stream record when the traverse_dataset call has finished.  If there is no
565  * dataset to traverse, the thread immediately pushes End of Stream marker.
566  */
567 static void
568 send_traverse_thread(void *arg)
569 {
570 	struct send_thread_arg *st_arg = arg;
571 	int err;
572 	struct send_block_record *data;
573 
574 	if (st_arg->ds != NULL) {
575 		err = traverse_dataset_resume(st_arg->ds,
576 		    st_arg->fromtxg, &st_arg->resume,
577 		    st_arg->flags, send_cb, st_arg);
578 
579 		if (err != EINTR)
580 			st_arg->error_code = err;
581 	}
582 	data = kmem_zalloc(sizeof (*data), KM_SLEEP);
583 	data->eos_marker = B_TRUE;
584 	bqueue_enqueue(&st_arg->q, data, 1);
585 	thread_exit();
586 }
587 
588 /*
589  * This function actually handles figuring out what kind of record needs to be
590  * dumped, reading the data (which has hopefully been prefetched), and calling
591  * the appropriate helper function.
592  */
593 static int
594 do_dump(dmu_sendarg_t *dsa, struct send_block_record *data)
595 {
596 	dsl_dataset_t *ds = dmu_objset_ds(dsa->dsa_os);
597 	const blkptr_t *bp = &data->bp;
598 	const zbookmark_phys_t *zb = &data->zb;
599 	uint8_t indblkshift = data->indblkshift;
600 	uint16_t dblkszsec = data->datablkszsec;
601 	spa_t *spa = ds->ds_dir->dd_pool->dp_spa;
602 	dmu_object_type_t type = bp ? BP_GET_TYPE(bp) : DMU_OT_NONE;
603 	int err = 0;
604 
605 	ASSERT3U(zb->zb_level, >=, 0);
606 
607 	ASSERT(zb->zb_object == DMU_META_DNODE_OBJECT ||
608 	    zb->zb_object >= dsa->dsa_resume_object);
609 
610 	if (zb->zb_object != DMU_META_DNODE_OBJECT &&
611 	    DMU_OBJECT_IS_SPECIAL(zb->zb_object)) {
612 		return (0);
613 	} else if (BP_IS_HOLE(bp) &&
614 	    zb->zb_object == DMU_META_DNODE_OBJECT) {
615 		uint64_t span = BP_SPAN(dblkszsec, indblkshift, zb->zb_level);
616 		uint64_t dnobj = (zb->zb_blkid * span) >> DNODE_SHIFT;
617 		err = dump_freeobjects(dsa, dnobj, span >> DNODE_SHIFT);
618 	} else if (BP_IS_HOLE(bp)) {
619 		uint64_t span = BP_SPAN(dblkszsec, indblkshift, zb->zb_level);
620 		uint64_t offset = zb->zb_blkid * span;
621 		err = dump_free(dsa, zb->zb_object, offset, span);
622 	} else if (zb->zb_level > 0 || type == DMU_OT_OBJSET) {
623 		return (0);
624 	} else if (type == DMU_OT_DNODE) {
625 		int epb = BP_GET_LSIZE(bp) >> DNODE_SHIFT;
626 		arc_flags_t aflags = ARC_FLAG_WAIT;
627 		arc_buf_t *abuf;
628 
629 		ASSERT0(zb->zb_level);
630 
631 		if (arc_read(NULL, spa, bp, arc_getbuf_func, &abuf,
632 		    ZIO_PRIORITY_ASYNC_READ, ZIO_FLAG_CANFAIL,
633 		    &aflags, zb) != 0)
634 			return (SET_ERROR(EIO));
635 
636 		dnode_phys_t *blk = abuf->b_data;
637 		uint64_t dnobj = zb->zb_blkid * epb;
638 		for (int i = 0; i < epb; i += blk[i].dn_extra_slots + 1) {
639 			err = dump_dnode(dsa, dnobj + i, blk + i);
640 			if (err != 0)
641 				break;
642 		}
643 		arc_buf_destroy(abuf, &abuf);
644 	} else if (type == DMU_OT_SA) {
645 		arc_flags_t aflags = ARC_FLAG_WAIT;
646 		arc_buf_t *abuf;
647 		int blksz = BP_GET_LSIZE(bp);
648 
649 		if (arc_read(NULL, spa, bp, arc_getbuf_func, &abuf,
650 		    ZIO_PRIORITY_ASYNC_READ, ZIO_FLAG_CANFAIL,
651 		    &aflags, zb) != 0)
652 			return (SET_ERROR(EIO));
653 
654 		err = dump_spill(dsa, zb->zb_object, blksz, abuf->b_data);
655 		arc_buf_destroy(abuf, &abuf);
656 	} else if (backup_do_embed(dsa, bp)) {
657 		/* it's an embedded level-0 block of a regular object */
658 		int blksz = dblkszsec << SPA_MINBLOCKSHIFT;
659 		ASSERT0(zb->zb_level);
660 		err = dump_write_embedded(dsa, zb->zb_object,
661 		    zb->zb_blkid * blksz, blksz, bp);
662 	} else {
663 		/* it's a level-0 block of a regular object */
664 		arc_flags_t aflags = ARC_FLAG_WAIT;
665 		arc_buf_t *abuf;
666 		int blksz = dblkszsec << SPA_MINBLOCKSHIFT;
667 		uint64_t offset;
668 
669 		/*
670 		 * If we have large blocks stored on disk but the send flags
671 		 * don't allow us to send large blocks, we split the data from
672 		 * the arc buf into chunks.
673 		 */
674 		boolean_t split_large_blocks = blksz > SPA_OLD_MAXBLOCKSIZE &&
675 		    !(dsa->dsa_featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS);
676 		/*
677 		 * We should only request compressed data from the ARC if all
678 		 * the following are true:
679 		 *  - stream compression was requested
680 		 *  - we aren't splitting large blocks into smaller chunks
681 		 *  - the data won't need to be byteswapped before sending
682 		 *  - this isn't an embedded block
683 		 *  - this isn't metadata (if receiving on a different endian
684 		 *    system it can be byteswapped more easily)
685 		 */
686 		boolean_t request_compressed =
687 		    (dsa->dsa_featureflags & DMU_BACKUP_FEATURE_COMPRESSED) &&
688 		    !split_large_blocks && !BP_SHOULD_BYTESWAP(bp) &&
689 		    !BP_IS_EMBEDDED(bp) && !DMU_OT_IS_METADATA(BP_GET_TYPE(bp));
690 
691 		ASSERT0(zb->zb_level);
692 		ASSERT(zb->zb_object > dsa->dsa_resume_object ||
693 		    (zb->zb_object == dsa->dsa_resume_object &&
694 		    zb->zb_blkid * blksz >= dsa->dsa_resume_offset));
695 
696 		ASSERT0(zb->zb_level);
697 		ASSERT(zb->zb_object > dsa->dsa_resume_object ||
698 		    (zb->zb_object == dsa->dsa_resume_object &&
699 		    zb->zb_blkid * blksz >= dsa->dsa_resume_offset));
700 
701 		ASSERT3U(blksz, ==, BP_GET_LSIZE(bp));
702 
703 		enum zio_flag zioflags = ZIO_FLAG_CANFAIL;
704 		if (request_compressed)
705 			zioflags |= ZIO_FLAG_RAW;
706 		if (arc_read(NULL, spa, bp, arc_getbuf_func, &abuf,
707 		    ZIO_PRIORITY_ASYNC_READ, zioflags, &aflags, zb) != 0) {
708 			if (zfs_send_corrupt_data) {
709 				/* Send a block filled with 0x"zfs badd bloc" */
710 				abuf = arc_alloc_buf(spa, &abuf, ARC_BUFC_DATA,
711 				    blksz);
712 				uint64_t *ptr;
713 				for (ptr = abuf->b_data;
714 				    (char *)ptr < (char *)abuf->b_data + blksz;
715 				    ptr++)
716 					*ptr = 0x2f5baddb10cULL;
717 			} else {
718 				return (SET_ERROR(EIO));
719 			}
720 		}
721 
722 		offset = zb->zb_blkid * blksz;
723 
724 		if (split_large_blocks) {
725 			ASSERT3U(arc_get_compression(abuf), ==,
726 			    ZIO_COMPRESS_OFF);
727 			char *buf = abuf->b_data;
728 			while (blksz > 0 && err == 0) {
729 				int n = MIN(blksz, SPA_OLD_MAXBLOCKSIZE);
730 				err = dump_write(dsa, type, zb->zb_object,
731 				    offset, n, n, NULL, buf);
732 				offset += n;
733 				buf += n;
734 				blksz -= n;
735 			}
736 		} else {
737 			err = dump_write(dsa, type, zb->zb_object, offset,
738 			    blksz, arc_buf_size(abuf), bp, abuf->b_data);
739 		}
740 		arc_buf_destroy(abuf, &abuf);
741 	}
742 
743 	ASSERT(err == 0 || err == EINTR);
744 	return (err);
745 }
746 
747 /*
748  * Pop the new data off the queue, and free the old data.
749  */
750 static struct send_block_record *
751 get_next_record(bqueue_t *bq, struct send_block_record *data)
752 {
753 	struct send_block_record *tmp = bqueue_dequeue(bq);
754 	kmem_free(data, sizeof (*data));
755 	return (tmp);
756 }
757 
758 /*
759  * Actually do the bulk of the work in a zfs send.
760  *
761  * Note: Releases dp using the specified tag.
762  */
763 static int
764 dmu_send_impl(void *tag, dsl_pool_t *dp, dsl_dataset_t *to_ds,
765     zfs_bookmark_phys_t *ancestor_zb, boolean_t is_clone,
766     boolean_t embedok, boolean_t large_block_ok, boolean_t compressok,
767     int outfd, uint64_t resumeobj, uint64_t resumeoff,
768     vnode_t *vp, offset_t *off)
769 {
770 	objset_t *os;
771 	dmu_replay_record_t *drr;
772 	dmu_sendarg_t *dsp;
773 	int err;
774 	uint64_t fromtxg = 0;
775 	uint64_t featureflags = 0;
776 	struct send_thread_arg to_arg = { 0 };
777 
778 	err = dmu_objset_from_ds(to_ds, &os);
779 	if (err != 0) {
780 		dsl_pool_rele(dp, tag);
781 		return (err);
782 	}
783 
784 	drr = kmem_zalloc(sizeof (dmu_replay_record_t), KM_SLEEP);
785 	drr->drr_type = DRR_BEGIN;
786 	drr->drr_u.drr_begin.drr_magic = DMU_BACKUP_MAGIC;
787 	DMU_SET_STREAM_HDRTYPE(drr->drr_u.drr_begin.drr_versioninfo,
788 	    DMU_SUBSTREAM);
789 
790 #ifdef _KERNEL
791 	if (dmu_objset_type(os) == DMU_OST_ZFS) {
792 		uint64_t version;
793 		if (zfs_get_zplprop(os, ZFS_PROP_VERSION, &version) != 0) {
794 			kmem_free(drr, sizeof (dmu_replay_record_t));
795 			dsl_pool_rele(dp, tag);
796 			return (SET_ERROR(EINVAL));
797 		}
798 		if (version >= ZPL_VERSION_SA) {
799 			featureflags |= DMU_BACKUP_FEATURE_SA_SPILL;
800 		}
801 	}
802 #endif
803 
804 	if (large_block_ok && to_ds->ds_feature_inuse[SPA_FEATURE_LARGE_BLOCKS])
805 		featureflags |= DMU_BACKUP_FEATURE_LARGE_BLOCKS;
806 	if (to_ds->ds_feature_inuse[SPA_FEATURE_LARGE_DNODE])
807 		featureflags |= DMU_BACKUP_FEATURE_LARGE_DNODE;
808 	if (embedok &&
809 	    spa_feature_is_active(dp->dp_spa, SPA_FEATURE_EMBEDDED_DATA)) {
810 		featureflags |= DMU_BACKUP_FEATURE_EMBED_DATA;
811 		if (spa_feature_is_active(dp->dp_spa, SPA_FEATURE_LZ4_COMPRESS))
812 			featureflags |= DMU_BACKUP_FEATURE_LZ4;
813 	}
814 	if (compressok) {
815 		featureflags |= DMU_BACKUP_FEATURE_COMPRESSED;
816 	}
817 	if ((featureflags &
818 	    (DMU_BACKUP_FEATURE_EMBED_DATA | DMU_BACKUP_FEATURE_COMPRESSED)) !=
819 	    0 && spa_feature_is_active(dp->dp_spa, SPA_FEATURE_LZ4_COMPRESS)) {
820 		featureflags |= DMU_BACKUP_FEATURE_LZ4;
821 	}
822 
823 	if (resumeobj != 0 || resumeoff != 0) {
824 		featureflags |= DMU_BACKUP_FEATURE_RESUMING;
825 	}
826 
827 	DMU_SET_FEATUREFLAGS(drr->drr_u.drr_begin.drr_versioninfo,
828 	    featureflags);
829 
830 	drr->drr_u.drr_begin.drr_creation_time =
831 	    dsl_dataset_phys(to_ds)->ds_creation_time;
832 	drr->drr_u.drr_begin.drr_type = dmu_objset_type(os);
833 	if (is_clone)
834 		drr->drr_u.drr_begin.drr_flags |= DRR_FLAG_CLONE;
835 	drr->drr_u.drr_begin.drr_toguid = dsl_dataset_phys(to_ds)->ds_guid;
836 	if (dsl_dataset_phys(to_ds)->ds_flags & DS_FLAG_CI_DATASET)
837 		drr->drr_u.drr_begin.drr_flags |= DRR_FLAG_CI_DATA;
838 	if (zfs_send_set_freerecords_bit)
839 		drr->drr_u.drr_begin.drr_flags |= DRR_FLAG_FREERECORDS;
840 
841 	if (ancestor_zb != NULL) {
842 		drr->drr_u.drr_begin.drr_fromguid =
843 		    ancestor_zb->zbm_guid;
844 		fromtxg = ancestor_zb->zbm_creation_txg;
845 	}
846 	dsl_dataset_name(to_ds, drr->drr_u.drr_begin.drr_toname);
847 	if (!to_ds->ds_is_snapshot) {
848 		(void) strlcat(drr->drr_u.drr_begin.drr_toname, "@--head--",
849 		    sizeof (drr->drr_u.drr_begin.drr_toname));
850 	}
851 
852 	dsp = kmem_zalloc(sizeof (dmu_sendarg_t), KM_SLEEP);
853 
854 	dsp->dsa_drr = drr;
855 	dsp->dsa_vp = vp;
856 	dsp->dsa_outfd = outfd;
857 	dsp->dsa_proc = curproc;
858 	dsp->dsa_os = os;
859 	dsp->dsa_off = off;
860 	dsp->dsa_toguid = dsl_dataset_phys(to_ds)->ds_guid;
861 	dsp->dsa_pending_op = PENDING_NONE;
862 	dsp->dsa_featureflags = featureflags;
863 	dsp->dsa_resume_object = resumeobj;
864 	dsp->dsa_resume_offset = resumeoff;
865 
866 	mutex_enter(&to_ds->ds_sendstream_lock);
867 	list_insert_head(&to_ds->ds_sendstreams, dsp);
868 	mutex_exit(&to_ds->ds_sendstream_lock);
869 
870 	dsl_dataset_long_hold(to_ds, FTAG);
871 	dsl_pool_rele(dp, tag);
872 
873 	void *payload = NULL;
874 	size_t payload_len = 0;
875 	if (resumeobj != 0 || resumeoff != 0) {
876 		dmu_object_info_t to_doi;
877 		err = dmu_object_info(os, resumeobj, &to_doi);
878 		if (err != 0)
879 			goto out;
880 		SET_BOOKMARK(&to_arg.resume, to_ds->ds_object, resumeobj, 0,
881 		    resumeoff / to_doi.doi_data_block_size);
882 
883 		nvlist_t *nvl = fnvlist_alloc();
884 		fnvlist_add_uint64(nvl, "resume_object", resumeobj);
885 		fnvlist_add_uint64(nvl, "resume_offset", resumeoff);
886 		payload = fnvlist_pack(nvl, &payload_len);
887 		drr->drr_payloadlen = payload_len;
888 		fnvlist_free(nvl);
889 	}
890 
891 	err = dump_record(dsp, payload, payload_len);
892 	fnvlist_pack_free(payload, payload_len);
893 	if (err != 0) {
894 		err = dsp->dsa_err;
895 		goto out;
896 	}
897 
898 	err = bqueue_init(&to_arg.q, zfs_send_queue_length,
899 	    offsetof(struct send_block_record, ln));
900 	to_arg.error_code = 0;
901 	to_arg.cancel = B_FALSE;
902 	to_arg.ds = to_ds;
903 	to_arg.fromtxg = fromtxg;
904 	to_arg.flags = TRAVERSE_PRE | TRAVERSE_PREFETCH;
905 	(void) thread_create(NULL, 0, send_traverse_thread, &to_arg, 0, curproc,
906 	    TS_RUN, minclsyspri);
907 
908 	struct send_block_record *to_data;
909 	to_data = bqueue_dequeue(&to_arg.q);
910 
911 	while (!to_data->eos_marker && err == 0) {
912 		err = do_dump(dsp, to_data);
913 		to_data = get_next_record(&to_arg.q, to_data);
914 		if (issig(JUSTLOOKING) && issig(FORREAL))
915 			err = EINTR;
916 	}
917 
918 	if (err != 0) {
919 		to_arg.cancel = B_TRUE;
920 		while (!to_data->eos_marker) {
921 			to_data = get_next_record(&to_arg.q, to_data);
922 		}
923 	}
924 	kmem_free(to_data, sizeof (*to_data));
925 
926 	bqueue_destroy(&to_arg.q);
927 
928 	if (err == 0 && to_arg.error_code != 0)
929 		err = to_arg.error_code;
930 
931 	if (err != 0)
932 		goto out;
933 
934 	if (dsp->dsa_pending_op != PENDING_NONE)
935 		if (dump_record(dsp, NULL, 0) != 0)
936 			err = SET_ERROR(EINTR);
937 
938 	if (err != 0) {
939 		if (err == EINTR && dsp->dsa_err != 0)
940 			err = dsp->dsa_err;
941 		goto out;
942 	}
943 
944 	bzero(drr, sizeof (dmu_replay_record_t));
945 	drr->drr_type = DRR_END;
946 	drr->drr_u.drr_end.drr_checksum = dsp->dsa_zc;
947 	drr->drr_u.drr_end.drr_toguid = dsp->dsa_toguid;
948 
949 	if (dump_record(dsp, NULL, 0) != 0)
950 		err = dsp->dsa_err;
951 
952 out:
953 	mutex_enter(&to_ds->ds_sendstream_lock);
954 	list_remove(&to_ds->ds_sendstreams, dsp);
955 	mutex_exit(&to_ds->ds_sendstream_lock);
956 
957 	VERIFY(err != 0 || (dsp->dsa_sent_begin && dsp->dsa_sent_end));
958 
959 	kmem_free(drr, sizeof (dmu_replay_record_t));
960 	kmem_free(dsp, sizeof (dmu_sendarg_t));
961 
962 	dsl_dataset_long_rele(to_ds, FTAG);
963 
964 	return (err);
965 }
966 
967 int
968 dmu_send_obj(const char *pool, uint64_t tosnap, uint64_t fromsnap,
969     boolean_t embedok, boolean_t large_block_ok, boolean_t compressok,
970     int outfd, vnode_t *vp, offset_t *off)
971 {
972 	dsl_pool_t *dp;
973 	dsl_dataset_t *ds;
974 	dsl_dataset_t *fromds = NULL;
975 	int err;
976 
977 	err = dsl_pool_hold(pool, FTAG, &dp);
978 	if (err != 0)
979 		return (err);
980 
981 	err = dsl_dataset_hold_obj(dp, tosnap, FTAG, &ds);
982 	if (err != 0) {
983 		dsl_pool_rele(dp, FTAG);
984 		return (err);
985 	}
986 
987 	if (fromsnap != 0) {
988 		zfs_bookmark_phys_t zb;
989 		boolean_t is_clone;
990 
991 		err = dsl_dataset_hold_obj(dp, fromsnap, FTAG, &fromds);
992 		if (err != 0) {
993 			dsl_dataset_rele(ds, FTAG);
994 			dsl_pool_rele(dp, FTAG);
995 			return (err);
996 		}
997 		if (!dsl_dataset_is_before(ds, fromds, 0))
998 			err = SET_ERROR(EXDEV);
999 		zb.zbm_creation_time =
1000 		    dsl_dataset_phys(fromds)->ds_creation_time;
1001 		zb.zbm_creation_txg = dsl_dataset_phys(fromds)->ds_creation_txg;
1002 		zb.zbm_guid = dsl_dataset_phys(fromds)->ds_guid;
1003 		is_clone = (fromds->ds_dir != ds->ds_dir);
1004 		dsl_dataset_rele(fromds, FTAG);
1005 		err = dmu_send_impl(FTAG, dp, ds, &zb, is_clone,
1006 		    embedok, large_block_ok, compressok, outfd, 0, 0, vp, off);
1007 	} else {
1008 		err = dmu_send_impl(FTAG, dp, ds, NULL, B_FALSE,
1009 		    embedok, large_block_ok, compressok, outfd, 0, 0, vp, off);
1010 	}
1011 	dsl_dataset_rele(ds, FTAG);
1012 	return (err);
1013 }
1014 
1015 int
1016 dmu_send(const char *tosnap, const char *fromsnap, boolean_t embedok,
1017     boolean_t large_block_ok, boolean_t compressok, int outfd,
1018     uint64_t resumeobj, uint64_t resumeoff,
1019     vnode_t *vp, offset_t *off)
1020 {
1021 	dsl_pool_t *dp;
1022 	dsl_dataset_t *ds;
1023 	int err;
1024 	boolean_t owned = B_FALSE;
1025 
1026 	if (fromsnap != NULL && strpbrk(fromsnap, "@#") == NULL)
1027 		return (SET_ERROR(EINVAL));
1028 
1029 	err = dsl_pool_hold(tosnap, FTAG, &dp);
1030 	if (err != 0)
1031 		return (err);
1032 
1033 	if (strchr(tosnap, '@') == NULL && spa_writeable(dp->dp_spa)) {
1034 		/*
1035 		 * We are sending a filesystem or volume.  Ensure
1036 		 * that it doesn't change by owning the dataset.
1037 		 */
1038 		err = dsl_dataset_own(dp, tosnap, FTAG, &ds);
1039 		owned = B_TRUE;
1040 	} else {
1041 		err = dsl_dataset_hold(dp, tosnap, FTAG, &ds);
1042 	}
1043 	if (err != 0) {
1044 		dsl_pool_rele(dp, FTAG);
1045 		return (err);
1046 	}
1047 
1048 	if (fromsnap != NULL) {
1049 		zfs_bookmark_phys_t zb;
1050 		boolean_t is_clone = B_FALSE;
1051 		int fsnamelen = strchr(tosnap, '@') - tosnap;
1052 
1053 		/*
1054 		 * If the fromsnap is in a different filesystem, then
1055 		 * mark the send stream as a clone.
1056 		 */
1057 		if (strncmp(tosnap, fromsnap, fsnamelen) != 0 ||
1058 		    (fromsnap[fsnamelen] != '@' &&
1059 		    fromsnap[fsnamelen] != '#')) {
1060 			is_clone = B_TRUE;
1061 		}
1062 
1063 		if (strchr(fromsnap, '@')) {
1064 			dsl_dataset_t *fromds;
1065 			err = dsl_dataset_hold(dp, fromsnap, FTAG, &fromds);
1066 			if (err == 0) {
1067 				if (!dsl_dataset_is_before(ds, fromds, 0))
1068 					err = SET_ERROR(EXDEV);
1069 				zb.zbm_creation_time =
1070 				    dsl_dataset_phys(fromds)->ds_creation_time;
1071 				zb.zbm_creation_txg =
1072 				    dsl_dataset_phys(fromds)->ds_creation_txg;
1073 				zb.zbm_guid = dsl_dataset_phys(fromds)->ds_guid;
1074 				is_clone = (ds->ds_dir != fromds->ds_dir);
1075 				dsl_dataset_rele(fromds, FTAG);
1076 			}
1077 		} else {
1078 			err = dsl_bookmark_lookup(dp, fromsnap, ds, &zb);
1079 		}
1080 		if (err != 0) {
1081 			dsl_dataset_rele(ds, FTAG);
1082 			dsl_pool_rele(dp, FTAG);
1083 			return (err);
1084 		}
1085 		err = dmu_send_impl(FTAG, dp, ds, &zb, is_clone,
1086 		    embedok, large_block_ok, compressok,
1087 		    outfd, resumeobj, resumeoff, vp, off);
1088 	} else {
1089 		err = dmu_send_impl(FTAG, dp, ds, NULL, B_FALSE,
1090 		    embedok, large_block_ok, compressok,
1091 		    outfd, resumeobj, resumeoff, vp, off);
1092 	}
1093 	if (owned)
1094 		dsl_dataset_disown(ds, FTAG);
1095 	else
1096 		dsl_dataset_rele(ds, FTAG);
1097 	return (err);
1098 }
1099 
1100 static int
1101 dmu_adjust_send_estimate_for_indirects(dsl_dataset_t *ds, uint64_t uncompressed,
1102     uint64_t compressed, boolean_t stream_compressed, uint64_t *sizep)
1103 {
1104 	int err = 0;
1105 	uint64_t size;
1106 	/*
1107 	 * Assume that space (both on-disk and in-stream) is dominated by
1108 	 * data.  We will adjust for indirect blocks and the copies property,
1109 	 * but ignore per-object space used (eg, dnodes and DRR_OBJECT records).
1110 	 */
1111 	uint64_t recordsize;
1112 	uint64_t record_count;
1113 	objset_t *os;
1114 	VERIFY0(dmu_objset_from_ds(ds, &os));
1115 
1116 	/* Assume all (uncompressed) blocks are recordsize. */
1117 	if (zfs_override_estimate_recordsize != 0) {
1118 		recordsize = zfs_override_estimate_recordsize;
1119 	} else if (os->os_phys->os_type == DMU_OST_ZVOL) {
1120 		err = dsl_prop_get_int_ds(ds,
1121 		    zfs_prop_to_name(ZFS_PROP_VOLBLOCKSIZE), &recordsize);
1122 	} else {
1123 		err = dsl_prop_get_int_ds(ds,
1124 		    zfs_prop_to_name(ZFS_PROP_RECORDSIZE), &recordsize);
1125 	}
1126 	if (err != 0)
1127 		return (err);
1128 	record_count = uncompressed / recordsize;
1129 
1130 	/*
1131 	 * If we're estimating a send size for a compressed stream, use the
1132 	 * compressed data size to estimate the stream size. Otherwise, use the
1133 	 * uncompressed data size.
1134 	 */
1135 	size = stream_compressed ? compressed : uncompressed;
1136 
1137 	/*
1138 	 * Subtract out approximate space used by indirect blocks.
1139 	 * Assume most space is used by data blocks (non-indirect, non-dnode).
1140 	 * Assume no ditto blocks or internal fragmentation.
1141 	 *
1142 	 * Therefore, space used by indirect blocks is sizeof(blkptr_t) per
1143 	 * block.
1144 	 */
1145 	size -= record_count * sizeof (blkptr_t);
1146 
1147 	/* Add in the space for the record associated with each block. */
1148 	size += record_count * sizeof (dmu_replay_record_t);
1149 
1150 	*sizep = size;
1151 
1152 	return (0);
1153 }
1154 
1155 int
1156 dmu_send_estimate(dsl_dataset_t *ds, dsl_dataset_t *fromds,
1157     boolean_t stream_compressed, uint64_t *sizep)
1158 {
1159 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
1160 	int err;
1161 	uint64_t uncomp, comp;
1162 
1163 	ASSERT(dsl_pool_config_held(dp));
1164 
1165 	/* tosnap must be a snapshot */
1166 	if (!ds->ds_is_snapshot)
1167 		return (SET_ERROR(EINVAL));
1168 
1169 	/* fromsnap, if provided, must be a snapshot */
1170 	if (fromds != NULL && !fromds->ds_is_snapshot)
1171 		return (SET_ERROR(EINVAL));
1172 
1173 	/*
1174 	 * fromsnap must be an earlier snapshot from the same fs as tosnap,
1175 	 * or the origin's fs.
1176 	 */
1177 	if (fromds != NULL && !dsl_dataset_is_before(ds, fromds, 0))
1178 		return (SET_ERROR(EXDEV));
1179 
1180 	/* Get compressed and uncompressed size estimates of changed data. */
1181 	if (fromds == NULL) {
1182 		uncomp = dsl_dataset_phys(ds)->ds_uncompressed_bytes;
1183 		comp = dsl_dataset_phys(ds)->ds_compressed_bytes;
1184 	} else {
1185 		uint64_t used;
1186 		err = dsl_dataset_space_written(fromds, ds,
1187 		    &used, &comp, &uncomp);
1188 		if (err != 0)
1189 			return (err);
1190 	}
1191 
1192 	err = dmu_adjust_send_estimate_for_indirects(ds, uncomp, comp,
1193 	    stream_compressed, sizep);
1194 	/*
1195 	 * Add the size of the BEGIN and END records to the estimate.
1196 	 */
1197 	*sizep += 2 * sizeof (dmu_replay_record_t);
1198 	return (err);
1199 }
1200 
1201 struct calculate_send_arg {
1202 	uint64_t uncompressed;
1203 	uint64_t compressed;
1204 };
1205 
1206 /*
1207  * Simple callback used to traverse the blocks of a snapshot and sum their
1208  * uncompressed and compressed sizes.
1209  */
1210 /* ARGSUSED */
1211 static int
1212 dmu_calculate_send_traversal(spa_t *spa, zilog_t *zilog, const blkptr_t *bp,
1213     const zbookmark_phys_t *zb, const dnode_phys_t *dnp, void *arg)
1214 {
1215 	struct calculate_send_arg *space = arg;
1216 	if (bp != NULL && !BP_IS_HOLE(bp)) {
1217 		space->uncompressed += BP_GET_UCSIZE(bp);
1218 		space->compressed += BP_GET_PSIZE(bp);
1219 	}
1220 	return (0);
1221 }
1222 
1223 /*
1224  * Given a desination snapshot and a TXG, calculate the approximate size of a
1225  * send stream sent from that TXG. from_txg may be zero, indicating that the
1226  * whole snapshot will be sent.
1227  */
1228 int
1229 dmu_send_estimate_from_txg(dsl_dataset_t *ds, uint64_t from_txg,
1230     boolean_t stream_compressed, uint64_t *sizep)
1231 {
1232 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
1233 	int err;
1234 	struct calculate_send_arg size = { 0 };
1235 
1236 	ASSERT(dsl_pool_config_held(dp));
1237 
1238 	/* tosnap must be a snapshot */
1239 	if (!ds->ds_is_snapshot)
1240 		return (SET_ERROR(EINVAL));
1241 
1242 	/* verify that from_txg is before the provided snapshot was taken */
1243 	if (from_txg >= dsl_dataset_phys(ds)->ds_creation_txg) {
1244 		return (SET_ERROR(EXDEV));
1245 	}
1246 
1247 	/*
1248 	 * traverse the blocks of the snapshot with birth times after
1249 	 * from_txg, summing their uncompressed size
1250 	 */
1251 	err = traverse_dataset(ds, from_txg, TRAVERSE_POST,
1252 	    dmu_calculate_send_traversal, &size);
1253 	if (err)
1254 		return (err);
1255 
1256 	err = dmu_adjust_send_estimate_for_indirects(ds, size.uncompressed,
1257 	    size.compressed, stream_compressed, sizep);
1258 	return (err);
1259 }
1260 
1261 typedef struct dmu_recv_begin_arg {
1262 	const char *drba_origin;
1263 	dmu_recv_cookie_t *drba_cookie;
1264 	cred_t *drba_cred;
1265 	uint64_t drba_snapobj;
1266 } dmu_recv_begin_arg_t;
1267 
1268 static int
1269 recv_begin_check_existing_impl(dmu_recv_begin_arg_t *drba, dsl_dataset_t *ds,
1270     uint64_t fromguid)
1271 {
1272 	uint64_t val;
1273 	int error;
1274 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
1275 
1276 	/* temporary clone name must not exist */
1277 	error = zap_lookup(dp->dp_meta_objset,
1278 	    dsl_dir_phys(ds->ds_dir)->dd_child_dir_zapobj, recv_clone_name,
1279 	    8, 1, &val);
1280 	if (error != ENOENT)
1281 		return (error == 0 ? EBUSY : error);
1282 
1283 	/* new snapshot name must not exist */
1284 	error = zap_lookup(dp->dp_meta_objset,
1285 	    dsl_dataset_phys(ds)->ds_snapnames_zapobj,
1286 	    drba->drba_cookie->drc_tosnap, 8, 1, &val);
1287 	if (error != ENOENT)
1288 		return (error == 0 ? EEXIST : error);
1289 
1290 	/*
1291 	 * Check snapshot limit before receiving. We'll recheck again at the
1292 	 * end, but might as well abort before receiving if we're already over
1293 	 * the limit.
1294 	 *
1295 	 * Note that we do not check the file system limit with
1296 	 * dsl_dir_fscount_check because the temporary %clones don't count
1297 	 * against that limit.
1298 	 */
1299 	error = dsl_fs_ss_limit_check(ds->ds_dir, 1, ZFS_PROP_SNAPSHOT_LIMIT,
1300 	    NULL, drba->drba_cred);
1301 	if (error != 0)
1302 		return (error);
1303 
1304 	if (fromguid != 0) {
1305 		dsl_dataset_t *snap;
1306 		uint64_t obj = dsl_dataset_phys(ds)->ds_prev_snap_obj;
1307 
1308 		/* Find snapshot in this dir that matches fromguid. */
1309 		while (obj != 0) {
1310 			error = dsl_dataset_hold_obj(dp, obj, FTAG,
1311 			    &snap);
1312 			if (error != 0)
1313 				return (SET_ERROR(ENODEV));
1314 			if (snap->ds_dir != ds->ds_dir) {
1315 				dsl_dataset_rele(snap, FTAG);
1316 				return (SET_ERROR(ENODEV));
1317 			}
1318 			if (dsl_dataset_phys(snap)->ds_guid == fromguid)
1319 				break;
1320 			obj = dsl_dataset_phys(snap)->ds_prev_snap_obj;
1321 			dsl_dataset_rele(snap, FTAG);
1322 		}
1323 		if (obj == 0)
1324 			return (SET_ERROR(ENODEV));
1325 
1326 		if (drba->drba_cookie->drc_force) {
1327 			drba->drba_snapobj = obj;
1328 		} else {
1329 			/*
1330 			 * If we are not forcing, there must be no
1331 			 * changes since fromsnap.
1332 			 */
1333 			if (dsl_dataset_modified_since_snap(ds, snap)) {
1334 				dsl_dataset_rele(snap, FTAG);
1335 				return (SET_ERROR(ETXTBSY));
1336 			}
1337 			drba->drba_snapobj = ds->ds_prev->ds_object;
1338 		}
1339 
1340 		dsl_dataset_rele(snap, FTAG);
1341 	} else {
1342 		/* if full, then must be forced */
1343 		if (!drba->drba_cookie->drc_force)
1344 			return (SET_ERROR(EEXIST));
1345 		/* start from $ORIGIN@$ORIGIN, if supported */
1346 		drba->drba_snapobj = dp->dp_origin_snap != NULL ?
1347 		    dp->dp_origin_snap->ds_object : 0;
1348 	}
1349 
1350 	return (0);
1351 
1352 }
1353 
1354 static int
1355 dmu_recv_begin_check(void *arg, dmu_tx_t *tx)
1356 {
1357 	dmu_recv_begin_arg_t *drba = arg;
1358 	dsl_pool_t *dp = dmu_tx_pool(tx);
1359 	struct drr_begin *drrb = drba->drba_cookie->drc_drrb;
1360 	uint64_t fromguid = drrb->drr_fromguid;
1361 	int flags = drrb->drr_flags;
1362 	int error;
1363 	uint64_t featureflags = DMU_GET_FEATUREFLAGS(drrb->drr_versioninfo);
1364 	dsl_dataset_t *ds;
1365 	const char *tofs = drba->drba_cookie->drc_tofs;
1366 
1367 	/* already checked */
1368 	ASSERT3U(drrb->drr_magic, ==, DMU_BACKUP_MAGIC);
1369 	ASSERT(!(featureflags & DMU_BACKUP_FEATURE_RESUMING));
1370 
1371 	if (DMU_GET_STREAM_HDRTYPE(drrb->drr_versioninfo) ==
1372 	    DMU_COMPOUNDSTREAM ||
1373 	    drrb->drr_type >= DMU_OST_NUMTYPES ||
1374 	    ((flags & DRR_FLAG_CLONE) && drba->drba_origin == NULL))
1375 		return (SET_ERROR(EINVAL));
1376 
1377 	/* Verify pool version supports SA if SA_SPILL feature set */
1378 	if ((featureflags & DMU_BACKUP_FEATURE_SA_SPILL) &&
1379 	    spa_version(dp->dp_spa) < SPA_VERSION_SA)
1380 		return (SET_ERROR(ENOTSUP));
1381 
1382 	if (drba->drba_cookie->drc_resumable &&
1383 	    !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_EXTENSIBLE_DATASET))
1384 		return (SET_ERROR(ENOTSUP));
1385 
1386 	/*
1387 	 * The receiving code doesn't know how to translate a WRITE_EMBEDDED
1388 	 * record to a plain WRITE record, so the pool must have the
1389 	 * EMBEDDED_DATA feature enabled if the stream has WRITE_EMBEDDED
1390 	 * records.  Same with WRITE_EMBEDDED records that use LZ4 compression.
1391 	 */
1392 	if ((featureflags & DMU_BACKUP_FEATURE_EMBED_DATA) &&
1393 	    !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_EMBEDDED_DATA))
1394 		return (SET_ERROR(ENOTSUP));
1395 	if ((featureflags & DMU_BACKUP_FEATURE_LZ4) &&
1396 	    !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_LZ4_COMPRESS))
1397 		return (SET_ERROR(ENOTSUP));
1398 
1399 	/*
1400 	 * The receiving code doesn't know how to translate large blocks
1401 	 * to smaller ones, so the pool must have the LARGE_BLOCKS
1402 	 * feature enabled if the stream has LARGE_BLOCKS. Same with
1403 	 * large dnodes.
1404 	 */
1405 	if ((featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS) &&
1406 	    !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_LARGE_BLOCKS))
1407 		return (SET_ERROR(ENOTSUP));
1408 	if ((featureflags & DMU_BACKUP_FEATURE_LARGE_DNODE) &&
1409 	    !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_LARGE_DNODE))
1410 		return (SET_ERROR(ENOTSUP));
1411 
1412 	error = dsl_dataset_hold(dp, tofs, FTAG, &ds);
1413 	if (error == 0) {
1414 		/* target fs already exists; recv into temp clone */
1415 
1416 		/* Can't recv a clone into an existing fs */
1417 		if (flags & DRR_FLAG_CLONE || drba->drba_origin) {
1418 			dsl_dataset_rele(ds, FTAG);
1419 			return (SET_ERROR(EINVAL));
1420 		}
1421 
1422 		error = recv_begin_check_existing_impl(drba, ds, fromguid);
1423 		dsl_dataset_rele(ds, FTAG);
1424 	} else if (error == ENOENT) {
1425 		/* target fs does not exist; must be a full backup or clone */
1426 		char buf[ZFS_MAX_DATASET_NAME_LEN];
1427 
1428 		/*
1429 		 * If it's a non-clone incremental, we are missing the
1430 		 * target fs, so fail the recv.
1431 		 */
1432 		if (fromguid != 0 && !(flags & DRR_FLAG_CLONE ||
1433 		    drba->drba_origin))
1434 			return (SET_ERROR(ENOENT));
1435 
1436 		/*
1437 		 * If we're receiving a full send as a clone, and it doesn't
1438 		 * contain all the necessary free records and freeobject
1439 		 * records, reject it.
1440 		 */
1441 		if (fromguid == 0 && drba->drba_origin &&
1442 		    !(flags & DRR_FLAG_FREERECORDS))
1443 			return (SET_ERROR(EINVAL));
1444 
1445 		/* Open the parent of tofs */
1446 		ASSERT3U(strlen(tofs), <, sizeof (buf));
1447 		(void) strlcpy(buf, tofs, strrchr(tofs, '/') - tofs + 1);
1448 		error = dsl_dataset_hold(dp, buf, FTAG, &ds);
1449 		if (error != 0)
1450 			return (error);
1451 
1452 		/*
1453 		 * Check filesystem and snapshot limits before receiving. We'll
1454 		 * recheck snapshot limits again at the end (we create the
1455 		 * filesystems and increment those counts during begin_sync).
1456 		 */
1457 		error = dsl_fs_ss_limit_check(ds->ds_dir, 1,
1458 		    ZFS_PROP_FILESYSTEM_LIMIT, NULL, drba->drba_cred);
1459 		if (error != 0) {
1460 			dsl_dataset_rele(ds, FTAG);
1461 			return (error);
1462 		}
1463 
1464 		error = dsl_fs_ss_limit_check(ds->ds_dir, 1,
1465 		    ZFS_PROP_SNAPSHOT_LIMIT, NULL, drba->drba_cred);
1466 		if (error != 0) {
1467 			dsl_dataset_rele(ds, FTAG);
1468 			return (error);
1469 		}
1470 
1471 		if (drba->drba_origin != NULL) {
1472 			dsl_dataset_t *origin;
1473 			error = dsl_dataset_hold(dp, drba->drba_origin,
1474 			    FTAG, &origin);
1475 			if (error != 0) {
1476 				dsl_dataset_rele(ds, FTAG);
1477 				return (error);
1478 			}
1479 			if (!origin->ds_is_snapshot) {
1480 				dsl_dataset_rele(origin, FTAG);
1481 				dsl_dataset_rele(ds, FTAG);
1482 				return (SET_ERROR(EINVAL));
1483 			}
1484 			if (dsl_dataset_phys(origin)->ds_guid != fromguid &&
1485 			    fromguid != 0) {
1486 				dsl_dataset_rele(origin, FTAG);
1487 				dsl_dataset_rele(ds, FTAG);
1488 				return (SET_ERROR(ENODEV));
1489 			}
1490 			dsl_dataset_rele(origin, FTAG);
1491 		}
1492 		dsl_dataset_rele(ds, FTAG);
1493 		error = 0;
1494 	}
1495 	return (error);
1496 }
1497 
1498 static void
1499 dmu_recv_begin_sync(void *arg, dmu_tx_t *tx)
1500 {
1501 	dmu_recv_begin_arg_t *drba = arg;
1502 	dsl_pool_t *dp = dmu_tx_pool(tx);
1503 	objset_t *mos = dp->dp_meta_objset;
1504 	struct drr_begin *drrb = drba->drba_cookie->drc_drrb;
1505 	const char *tofs = drba->drba_cookie->drc_tofs;
1506 	dsl_dataset_t *ds, *newds;
1507 	uint64_t dsobj;
1508 	int error;
1509 	uint64_t crflags = 0;
1510 
1511 	if (drrb->drr_flags & DRR_FLAG_CI_DATA)
1512 		crflags |= DS_FLAG_CI_DATASET;
1513 
1514 	error = dsl_dataset_hold(dp, tofs, FTAG, &ds);
1515 	if (error == 0) {
1516 		/* create temporary clone */
1517 		dsl_dataset_t *snap = NULL;
1518 		if (drba->drba_snapobj != 0) {
1519 			VERIFY0(dsl_dataset_hold_obj(dp,
1520 			    drba->drba_snapobj, FTAG, &snap));
1521 		}
1522 		dsobj = dsl_dataset_create_sync(ds->ds_dir, recv_clone_name,
1523 		    snap, crflags, drba->drba_cred, tx);
1524 		if (drba->drba_snapobj != 0)
1525 			dsl_dataset_rele(snap, FTAG);
1526 		dsl_dataset_rele(ds, FTAG);
1527 	} else {
1528 		dsl_dir_t *dd;
1529 		const char *tail;
1530 		dsl_dataset_t *origin = NULL;
1531 
1532 		VERIFY0(dsl_dir_hold(dp, tofs, FTAG, &dd, &tail));
1533 
1534 		if (drba->drba_origin != NULL) {
1535 			VERIFY0(dsl_dataset_hold(dp, drba->drba_origin,
1536 			    FTAG, &origin));
1537 		}
1538 
1539 		/* Create new dataset. */
1540 		dsobj = dsl_dataset_create_sync(dd,
1541 		    strrchr(tofs, '/') + 1,
1542 		    origin, crflags, drba->drba_cred, tx);
1543 		if (origin != NULL)
1544 			dsl_dataset_rele(origin, FTAG);
1545 		dsl_dir_rele(dd, FTAG);
1546 		drba->drba_cookie->drc_newfs = B_TRUE;
1547 	}
1548 	VERIFY0(dsl_dataset_own_obj(dp, dsobj, dmu_recv_tag, &newds));
1549 
1550 	if (drba->drba_cookie->drc_resumable) {
1551 		dsl_dataset_zapify(newds, tx);
1552 		if (drrb->drr_fromguid != 0) {
1553 			VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_FROMGUID,
1554 			    8, 1, &drrb->drr_fromguid, tx));
1555 		}
1556 		VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_TOGUID,
1557 		    8, 1, &drrb->drr_toguid, tx));
1558 		VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_TONAME,
1559 		    1, strlen(drrb->drr_toname) + 1, drrb->drr_toname, tx));
1560 		uint64_t one = 1;
1561 		uint64_t zero = 0;
1562 		VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_OBJECT,
1563 		    8, 1, &one, tx));
1564 		VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_OFFSET,
1565 		    8, 1, &zero, tx));
1566 		VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_BYTES,
1567 		    8, 1, &zero, tx));
1568 		if (DMU_GET_FEATUREFLAGS(drrb->drr_versioninfo) &
1569 		    DMU_BACKUP_FEATURE_LARGE_BLOCKS) {
1570 			VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_LARGEBLOCK,
1571 			    8, 1, &one, tx));
1572 		}
1573 		if (DMU_GET_FEATUREFLAGS(drrb->drr_versioninfo) &
1574 		    DMU_BACKUP_FEATURE_EMBED_DATA) {
1575 			VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_EMBEDOK,
1576 			    8, 1, &one, tx));
1577 		}
1578 		if (DMU_GET_FEATUREFLAGS(drrb->drr_versioninfo) &
1579 		    DMU_BACKUP_FEATURE_COMPRESSED) {
1580 			VERIFY0(zap_add(mos, dsobj, DS_FIELD_RESUME_COMPRESSOK,
1581 			    8, 1, &one, tx));
1582 		}
1583 	}
1584 
1585 	dmu_buf_will_dirty(newds->ds_dbuf, tx);
1586 	dsl_dataset_phys(newds)->ds_flags |= DS_FLAG_INCONSISTENT;
1587 
1588 	/*
1589 	 * If we actually created a non-clone, we need to create the
1590 	 * objset in our new dataset.
1591 	 */
1592 	rrw_enter(&newds->ds_bp_rwlock, RW_READER, FTAG);
1593 	if (BP_IS_HOLE(dsl_dataset_get_blkptr(newds))) {
1594 		(void) dmu_objset_create_impl(dp->dp_spa,
1595 		    newds, dsl_dataset_get_blkptr(newds), drrb->drr_type, tx);
1596 	}
1597 	rrw_exit(&newds->ds_bp_rwlock, FTAG);
1598 
1599 	drba->drba_cookie->drc_ds = newds;
1600 
1601 	spa_history_log_internal_ds(newds, "receive", tx, "");
1602 }
1603 
1604 static int
1605 dmu_recv_resume_begin_check(void *arg, dmu_tx_t *tx)
1606 {
1607 	dmu_recv_begin_arg_t *drba = arg;
1608 	dsl_pool_t *dp = dmu_tx_pool(tx);
1609 	struct drr_begin *drrb = drba->drba_cookie->drc_drrb;
1610 	int error;
1611 	uint64_t featureflags = DMU_GET_FEATUREFLAGS(drrb->drr_versioninfo);
1612 	dsl_dataset_t *ds;
1613 	const char *tofs = drba->drba_cookie->drc_tofs;
1614 
1615 	/* 6 extra bytes for /%recv */
1616 	char recvname[ZFS_MAX_DATASET_NAME_LEN + 6];
1617 
1618 	/* already checked */
1619 	ASSERT3U(drrb->drr_magic, ==, DMU_BACKUP_MAGIC);
1620 	ASSERT(featureflags & DMU_BACKUP_FEATURE_RESUMING);
1621 
1622 	if (DMU_GET_STREAM_HDRTYPE(drrb->drr_versioninfo) ==
1623 	    DMU_COMPOUNDSTREAM ||
1624 	    drrb->drr_type >= DMU_OST_NUMTYPES)
1625 		return (SET_ERROR(EINVAL));
1626 
1627 	/* Verify pool version supports SA if SA_SPILL feature set */
1628 	if ((featureflags & DMU_BACKUP_FEATURE_SA_SPILL) &&
1629 	    spa_version(dp->dp_spa) < SPA_VERSION_SA)
1630 		return (SET_ERROR(ENOTSUP));
1631 
1632 	/*
1633 	 * The receiving code doesn't know how to translate a WRITE_EMBEDDED
1634 	 * record to a plain WRITE record, so the pool must have the
1635 	 * EMBEDDED_DATA feature enabled if the stream has WRITE_EMBEDDED
1636 	 * records.  Same with WRITE_EMBEDDED records that use LZ4 compression.
1637 	 */
1638 	if ((featureflags & DMU_BACKUP_FEATURE_EMBED_DATA) &&
1639 	    !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_EMBEDDED_DATA))
1640 		return (SET_ERROR(ENOTSUP));
1641 	if ((featureflags & DMU_BACKUP_FEATURE_LZ4) &&
1642 	    !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_LZ4_COMPRESS))
1643 		return (SET_ERROR(ENOTSUP));
1644 
1645 	/*
1646 	 * The receiving code doesn't know how to translate large blocks
1647 	 * to smaller ones, so the pool must have the LARGE_BLOCKS
1648 	 * feature enabled if the stream has LARGE_BLOCKS. Same with
1649 	 * large dnodes.
1650 	 */
1651 	if ((featureflags & DMU_BACKUP_FEATURE_LARGE_BLOCKS) &&
1652 	    !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_LARGE_BLOCKS))
1653 		return (SET_ERROR(ENOTSUP));
1654 	if ((featureflags & DMU_BACKUP_FEATURE_LARGE_DNODE) &&
1655 	    !spa_feature_is_enabled(dp->dp_spa, SPA_FEATURE_LARGE_DNODE))
1656 		return (SET_ERROR(ENOTSUP));
1657 
1658 	(void) snprintf(recvname, sizeof (recvname), "%s/%s",
1659 	    tofs, recv_clone_name);
1660 
1661 	if (dsl_dataset_hold(dp, recvname, FTAG, &ds) != 0) {
1662 		/* %recv does not exist; continue in tofs */
1663 		error = dsl_dataset_hold(dp, tofs, FTAG, &ds);
1664 		if (error != 0)
1665 			return (error);
1666 	}
1667 
1668 	/* check that ds is marked inconsistent */
1669 	if (!DS_IS_INCONSISTENT(ds)) {
1670 		dsl_dataset_rele(ds, FTAG);
1671 		return (SET_ERROR(EINVAL));
1672 	}
1673 
1674 	/* check that there is resuming data, and that the toguid matches */
1675 	if (!dsl_dataset_is_zapified(ds)) {
1676 		dsl_dataset_rele(ds, FTAG);
1677 		return (SET_ERROR(EINVAL));
1678 	}
1679 	uint64_t val;
1680 	error = zap_lookup(dp->dp_meta_objset, ds->ds_object,
1681 	    DS_FIELD_RESUME_TOGUID, sizeof (val), 1, &val);
1682 	if (error != 0 || drrb->drr_toguid != val) {
1683 		dsl_dataset_rele(ds, FTAG);
1684 		return (SET_ERROR(EINVAL));
1685 	}
1686 
1687 	/*
1688 	 * Check if the receive is still running.  If so, it will be owned.
1689 	 * Note that nothing else can own the dataset (e.g. after the receive
1690 	 * fails) because it will be marked inconsistent.
1691 	 */
1692 	if (dsl_dataset_has_owner(ds)) {
1693 		dsl_dataset_rele(ds, FTAG);
1694 		return (SET_ERROR(EBUSY));
1695 	}
1696 
1697 	/* There should not be any snapshots of this fs yet. */
1698 	if (ds->ds_prev != NULL && ds->ds_prev->ds_dir == ds->ds_dir) {
1699 		dsl_dataset_rele(ds, FTAG);
1700 		return (SET_ERROR(EINVAL));
1701 	}
1702 
1703 	/*
1704 	 * Note: resume point will be checked when we process the first WRITE
1705 	 * record.
1706 	 */
1707 
1708 	/* check that the origin matches */
1709 	val = 0;
1710 	(void) zap_lookup(dp->dp_meta_objset, ds->ds_object,
1711 	    DS_FIELD_RESUME_FROMGUID, sizeof (val), 1, &val);
1712 	if (drrb->drr_fromguid != val) {
1713 		dsl_dataset_rele(ds, FTAG);
1714 		return (SET_ERROR(EINVAL));
1715 	}
1716 
1717 	dsl_dataset_rele(ds, FTAG);
1718 	return (0);
1719 }
1720 
1721 static void
1722 dmu_recv_resume_begin_sync(void *arg, dmu_tx_t *tx)
1723 {
1724 	dmu_recv_begin_arg_t *drba = arg;
1725 	dsl_pool_t *dp = dmu_tx_pool(tx);
1726 	const char *tofs = drba->drba_cookie->drc_tofs;
1727 	dsl_dataset_t *ds;
1728 	uint64_t dsobj;
1729 	/* 6 extra bytes for /%recv */
1730 	char recvname[ZFS_MAX_DATASET_NAME_LEN + 6];
1731 
1732 	(void) snprintf(recvname, sizeof (recvname), "%s/%s",
1733 	    tofs, recv_clone_name);
1734 
1735 	if (dsl_dataset_hold(dp, recvname, FTAG, &ds) != 0) {
1736 		/* %recv does not exist; continue in tofs */
1737 		VERIFY0(dsl_dataset_hold(dp, tofs, FTAG, &ds));
1738 		drba->drba_cookie->drc_newfs = B_TRUE;
1739 	}
1740 
1741 	/* clear the inconsistent flag so that we can own it */
1742 	ASSERT(DS_IS_INCONSISTENT(ds));
1743 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
1744 	dsl_dataset_phys(ds)->ds_flags &= ~DS_FLAG_INCONSISTENT;
1745 	dsobj = ds->ds_object;
1746 	dsl_dataset_rele(ds, FTAG);
1747 
1748 	VERIFY0(dsl_dataset_own_obj(dp, dsobj, dmu_recv_tag, &ds));
1749 
1750 	dmu_buf_will_dirty(ds->ds_dbuf, tx);
1751 	dsl_dataset_phys(ds)->ds_flags |= DS_FLAG_INCONSISTENT;
1752 
1753 	rrw_enter(&ds->ds_bp_rwlock, RW_READER, FTAG);
1754 	ASSERT(!BP_IS_HOLE(dsl_dataset_get_blkptr(ds)));
1755 	rrw_exit(&ds->ds_bp_rwlock, FTAG);
1756 
1757 	drba->drba_cookie->drc_ds = ds;
1758 
1759 	spa_history_log_internal_ds(ds, "resume receive", tx, "");
1760 }
1761 
1762 /*
1763  * NB: callers *MUST* call dmu_recv_stream() if dmu_recv_begin()
1764  * succeeds; otherwise we will leak the holds on the datasets.
1765  */
1766 int
1767 dmu_recv_begin(char *tofs, char *tosnap, dmu_replay_record_t *drr_begin,
1768     boolean_t force, boolean_t resumable, char *origin, dmu_recv_cookie_t *drc)
1769 {
1770 	dmu_recv_begin_arg_t drba = { 0 };
1771 
1772 	bzero(drc, sizeof (dmu_recv_cookie_t));
1773 	drc->drc_drr_begin = drr_begin;
1774 	drc->drc_drrb = &drr_begin->drr_u.drr_begin;
1775 	drc->drc_tosnap = tosnap;
1776 	drc->drc_tofs = tofs;
1777 	drc->drc_force = force;
1778 	drc->drc_resumable = resumable;
1779 	drc->drc_cred = CRED();
1780 	drc->drc_clone = (origin != NULL);
1781 
1782 	if (drc->drc_drrb->drr_magic == BSWAP_64(DMU_BACKUP_MAGIC)) {
1783 		drc->drc_byteswap = B_TRUE;
1784 		(void) fletcher_4_incremental_byteswap(drr_begin,
1785 		    sizeof (dmu_replay_record_t), &drc->drc_cksum);
1786 		byteswap_record(drr_begin);
1787 	} else if (drc->drc_drrb->drr_magic == DMU_BACKUP_MAGIC) {
1788 		(void) fletcher_4_incremental_native(drr_begin,
1789 		    sizeof (dmu_replay_record_t), &drc->drc_cksum);
1790 	} else {
1791 		return (SET_ERROR(EINVAL));
1792 	}
1793 
1794 	drba.drba_origin = origin;
1795 	drba.drba_cookie = drc;
1796 	drba.drba_cred = CRED();
1797 
1798 	if (DMU_GET_FEATUREFLAGS(drc->drc_drrb->drr_versioninfo) &
1799 	    DMU_BACKUP_FEATURE_RESUMING) {
1800 		return (dsl_sync_task(tofs,
1801 		    dmu_recv_resume_begin_check, dmu_recv_resume_begin_sync,
1802 		    &drba, 5, ZFS_SPACE_CHECK_NORMAL));
1803 	} else  {
1804 		return (dsl_sync_task(tofs,
1805 		    dmu_recv_begin_check, dmu_recv_begin_sync,
1806 		    &drba, 5, ZFS_SPACE_CHECK_NORMAL));
1807 	}
1808 }
1809 
1810 struct receive_record_arg {
1811 	dmu_replay_record_t header;
1812 	void *payload; /* Pointer to a buffer containing the payload */
1813 	/*
1814 	 * If the record is a write, pointer to the arc_buf_t containing the
1815 	 * payload.
1816 	 */
1817 	arc_buf_t *write_buf;
1818 	int payload_size;
1819 	uint64_t bytes_read; /* bytes read from stream when record created */
1820 	boolean_t eos_marker; /* Marks the end of the stream */
1821 	bqueue_node_t node;
1822 };
1823 
1824 struct receive_writer_arg {
1825 	objset_t *os;
1826 	boolean_t byteswap;
1827 	bqueue_t q;
1828 
1829 	/*
1830 	 * These three args are used to signal to the main thread that we're
1831 	 * done.
1832 	 */
1833 	kmutex_t mutex;
1834 	kcondvar_t cv;
1835 	boolean_t done;
1836 
1837 	int err;
1838 	/* A map from guid to dataset to help handle dedup'd streams. */
1839 	avl_tree_t *guid_to_ds_map;
1840 	boolean_t resumable;
1841 	uint64_t last_object;
1842 	uint64_t last_offset;
1843 	uint64_t max_object; /* highest object ID referenced in stream */
1844 	uint64_t bytes_read; /* bytes read when current record created */
1845 };
1846 
1847 struct objlist {
1848 	list_t list; /* List of struct receive_objnode. */
1849 	/*
1850 	 * Last object looked up. Used to assert that objects are being looked
1851 	 * up in ascending order.
1852 	 */
1853 	uint64_t last_lookup;
1854 };
1855 
1856 struct receive_objnode {
1857 	list_node_t node;
1858 	uint64_t object;
1859 };
1860 
1861 struct receive_arg {
1862 	objset_t *os;
1863 	vnode_t *vp; /* The vnode to read the stream from */
1864 	uint64_t voff; /* The current offset in the stream */
1865 	uint64_t bytes_read;
1866 	/*
1867 	 * A record that has had its payload read in, but hasn't yet been handed
1868 	 * off to the worker thread.
1869 	 */
1870 	struct receive_record_arg *rrd;
1871 	/* A record that has had its header read in, but not its payload. */
1872 	struct receive_record_arg *next_rrd;
1873 	zio_cksum_t cksum;
1874 	zio_cksum_t prev_cksum;
1875 	int err;
1876 	boolean_t byteswap;
1877 	/* Sorted list of objects not to issue prefetches for. */
1878 	struct objlist ignore_objlist;
1879 };
1880 
1881 typedef struct guid_map_entry {
1882 	uint64_t	guid;
1883 	dsl_dataset_t	*gme_ds;
1884 	avl_node_t	avlnode;
1885 } guid_map_entry_t;
1886 
1887 static int
1888 guid_compare(const void *arg1, const void *arg2)
1889 {
1890 	const guid_map_entry_t *gmep1 = arg1;
1891 	const guid_map_entry_t *gmep2 = arg2;
1892 
1893 	if (gmep1->guid < gmep2->guid)
1894 		return (-1);
1895 	else if (gmep1->guid > gmep2->guid)
1896 		return (1);
1897 	return (0);
1898 }
1899 
1900 static void
1901 free_guid_map_onexit(void *arg)
1902 {
1903 	avl_tree_t *ca = arg;
1904 	void *cookie = NULL;
1905 	guid_map_entry_t *gmep;
1906 
1907 	while ((gmep = avl_destroy_nodes(ca, &cookie)) != NULL) {
1908 		dsl_dataset_long_rele(gmep->gme_ds, gmep);
1909 		dsl_dataset_rele(gmep->gme_ds, gmep);
1910 		kmem_free(gmep, sizeof (guid_map_entry_t));
1911 	}
1912 	avl_destroy(ca);
1913 	kmem_free(ca, sizeof (avl_tree_t));
1914 }
1915 
1916 static int
1917 receive_read(struct receive_arg *ra, int len, void *buf)
1918 {
1919 	int done = 0;
1920 
1921 	/*
1922 	 * The code doesn't rely on this (lengths being multiples of 8).  See
1923 	 * comment in dump_bytes.
1924 	 */
1925 	ASSERT0(len % 8);
1926 
1927 	while (done < len) {
1928 		ssize_t resid;
1929 
1930 		ra->err = vn_rdwr(UIO_READ, ra->vp,
1931 		    (char *)buf + done, len - done,
1932 		    ra->voff, UIO_SYSSPACE, FAPPEND,
1933 		    RLIM64_INFINITY, CRED(), &resid);
1934 
1935 		if (resid == len - done) {
1936 			/*
1937 			 * Note: ECKSUM indicates that the receive
1938 			 * was interrupted and can potentially be resumed.
1939 			 */
1940 			ra->err = SET_ERROR(ECKSUM);
1941 		}
1942 		ra->voff += len - done - resid;
1943 		done = len - resid;
1944 		if (ra->err != 0)
1945 			return (ra->err);
1946 	}
1947 
1948 	ra->bytes_read += len;
1949 
1950 	ASSERT3U(done, ==, len);
1951 	return (0);
1952 }
1953 
1954 static void
1955 byteswap_record(dmu_replay_record_t *drr)
1956 {
1957 #define	DO64(X) (drr->drr_u.X = BSWAP_64(drr->drr_u.X))
1958 #define	DO32(X) (drr->drr_u.X = BSWAP_32(drr->drr_u.X))
1959 	drr->drr_type = BSWAP_32(drr->drr_type);
1960 	drr->drr_payloadlen = BSWAP_32(drr->drr_payloadlen);
1961 
1962 	switch (drr->drr_type) {
1963 	case DRR_BEGIN:
1964 		DO64(drr_begin.drr_magic);
1965 		DO64(drr_begin.drr_versioninfo);
1966 		DO64(drr_begin.drr_creation_time);
1967 		DO32(drr_begin.drr_type);
1968 		DO32(drr_begin.drr_flags);
1969 		DO64(drr_begin.drr_toguid);
1970 		DO64(drr_begin.drr_fromguid);
1971 		break;
1972 	case DRR_OBJECT:
1973 		DO64(drr_object.drr_object);
1974 		DO32(drr_object.drr_type);
1975 		DO32(drr_object.drr_bonustype);
1976 		DO32(drr_object.drr_blksz);
1977 		DO32(drr_object.drr_bonuslen);
1978 		DO64(drr_object.drr_toguid);
1979 		break;
1980 	case DRR_FREEOBJECTS:
1981 		DO64(drr_freeobjects.drr_firstobj);
1982 		DO64(drr_freeobjects.drr_numobjs);
1983 		DO64(drr_freeobjects.drr_toguid);
1984 		break;
1985 	case DRR_WRITE:
1986 		DO64(drr_write.drr_object);
1987 		DO32(drr_write.drr_type);
1988 		DO64(drr_write.drr_offset);
1989 		DO64(drr_write.drr_logical_size);
1990 		DO64(drr_write.drr_toguid);
1991 		ZIO_CHECKSUM_BSWAP(&drr->drr_u.drr_write.drr_key.ddk_cksum);
1992 		DO64(drr_write.drr_key.ddk_prop);
1993 		DO64(drr_write.drr_compressed_size);
1994 		break;
1995 	case DRR_WRITE_BYREF:
1996 		DO64(drr_write_byref.drr_object);
1997 		DO64(drr_write_byref.drr_offset);
1998 		DO64(drr_write_byref.drr_length);
1999 		DO64(drr_write_byref.drr_toguid);
2000 		DO64(drr_write_byref.drr_refguid);
2001 		DO64(drr_write_byref.drr_refobject);
2002 		DO64(drr_write_byref.drr_refoffset);
2003 		ZIO_CHECKSUM_BSWAP(&drr->drr_u.drr_write_byref.
2004 		    drr_key.ddk_cksum);
2005 		DO64(drr_write_byref.drr_key.ddk_prop);
2006 		break;
2007 	case DRR_WRITE_EMBEDDED:
2008 		DO64(drr_write_embedded.drr_object);
2009 		DO64(drr_write_embedded.drr_offset);
2010 		DO64(drr_write_embedded.drr_length);
2011 		DO64(drr_write_embedded.drr_toguid);
2012 		DO32(drr_write_embedded.drr_lsize);
2013 		DO32(drr_write_embedded.drr_psize);
2014 		break;
2015 	case DRR_FREE:
2016 		DO64(drr_free.drr_object);
2017 		DO64(drr_free.drr_offset);
2018 		DO64(drr_free.drr_length);
2019 		DO64(drr_free.drr_toguid);
2020 		break;
2021 	case DRR_SPILL:
2022 		DO64(drr_spill.drr_object);
2023 		DO64(drr_spill.drr_length);
2024 		DO64(drr_spill.drr_toguid);
2025 		break;
2026 	case DRR_END:
2027 		DO64(drr_end.drr_toguid);
2028 		ZIO_CHECKSUM_BSWAP(&drr->drr_u.drr_end.drr_checksum);
2029 		break;
2030 	}
2031 
2032 	if (drr->drr_type != DRR_BEGIN) {
2033 		ZIO_CHECKSUM_BSWAP(&drr->drr_u.drr_checksum.drr_checksum);
2034 	}
2035 
2036 #undef DO64
2037 #undef DO32
2038 }
2039 
2040 static inline uint8_t
2041 deduce_nblkptr(dmu_object_type_t bonus_type, uint64_t bonus_size)
2042 {
2043 	if (bonus_type == DMU_OT_SA) {
2044 		return (1);
2045 	} else {
2046 		return (1 +
2047 		    ((DN_OLD_MAX_BONUSLEN -
2048 		    MIN(DN_OLD_MAX_BONUSLEN, bonus_size)) >> SPA_BLKPTRSHIFT));
2049 	}
2050 }
2051 
2052 static void
2053 save_resume_state(struct receive_writer_arg *rwa,
2054     uint64_t object, uint64_t offset, dmu_tx_t *tx)
2055 {
2056 	int txgoff = dmu_tx_get_txg(tx) & TXG_MASK;
2057 
2058 	if (!rwa->resumable)
2059 		return;
2060 
2061 	/*
2062 	 * We use ds_resume_bytes[] != 0 to indicate that we need to
2063 	 * update this on disk, so it must not be 0.
2064 	 */
2065 	ASSERT(rwa->bytes_read != 0);
2066 
2067 	/*
2068 	 * We only resume from write records, which have a valid
2069 	 * (non-meta-dnode) object number.
2070 	 */
2071 	ASSERT(object != 0);
2072 
2073 	/*
2074 	 * For resuming to work correctly, we must receive records in order,
2075 	 * sorted by object,offset.  This is checked by the callers, but
2076 	 * assert it here for good measure.
2077 	 */
2078 	ASSERT3U(object, >=, rwa->os->os_dsl_dataset->ds_resume_object[txgoff]);
2079 	ASSERT(object != rwa->os->os_dsl_dataset->ds_resume_object[txgoff] ||
2080 	    offset >= rwa->os->os_dsl_dataset->ds_resume_offset[txgoff]);
2081 	ASSERT3U(rwa->bytes_read, >=,
2082 	    rwa->os->os_dsl_dataset->ds_resume_bytes[txgoff]);
2083 
2084 	rwa->os->os_dsl_dataset->ds_resume_object[txgoff] = object;
2085 	rwa->os->os_dsl_dataset->ds_resume_offset[txgoff] = offset;
2086 	rwa->os->os_dsl_dataset->ds_resume_bytes[txgoff] = rwa->bytes_read;
2087 }
2088 
2089 static int
2090 receive_object(struct receive_writer_arg *rwa, struct drr_object *drro,
2091     void *data)
2092 {
2093 	dmu_object_info_t doi;
2094 	dmu_tx_t *tx;
2095 	uint64_t object;
2096 	int err;
2097 	uint8_t dn_slots = drro->drr_dn_slots != 0 ?
2098 	    drro->drr_dn_slots : DNODE_MIN_SLOTS;
2099 
2100 	if (drro->drr_type == DMU_OT_NONE ||
2101 	    !DMU_OT_IS_VALID(drro->drr_type) ||
2102 	    !DMU_OT_IS_VALID(drro->drr_bonustype) ||
2103 	    drro->drr_checksumtype >= ZIO_CHECKSUM_FUNCTIONS ||
2104 	    drro->drr_compress >= ZIO_COMPRESS_FUNCTIONS ||
2105 	    P2PHASE(drro->drr_blksz, SPA_MINBLOCKSIZE) ||
2106 	    drro->drr_blksz < SPA_MINBLOCKSIZE ||
2107 	    drro->drr_blksz > spa_maxblocksize(dmu_objset_spa(rwa->os)) ||
2108 	    drro->drr_bonuslen >
2109 	    DN_BONUS_SIZE(spa_maxdnodesize(dmu_objset_spa(rwa->os))) ||
2110 	    dn_slots >
2111 	    (spa_maxdnodesize(dmu_objset_spa(rwa->os)) >> DNODE_SHIFT)) {
2112 		return (SET_ERROR(EINVAL));
2113 	}
2114 
2115 	err = dmu_object_info(rwa->os, drro->drr_object, &doi);
2116 
2117 	if (err != 0 && err != ENOENT && err != EEXIST)
2118 		return (SET_ERROR(EINVAL));
2119 
2120 	if (drro->drr_object > rwa->max_object)
2121 		rwa->max_object = drro->drr_object;
2122 
2123 	/*
2124 	 * If we are losing blkptrs or changing the block size this must
2125 	 * be a new file instance.  We must clear out the previous file
2126 	 * contents before we can change this type of metadata in the dnode.
2127 	 */
2128 	if (err == 0) {
2129 		int nblkptr;
2130 
2131 		object = drro->drr_object;
2132 
2133 		nblkptr = deduce_nblkptr(drro->drr_bonustype,
2134 		    drro->drr_bonuslen);
2135 
2136 		if (drro->drr_blksz != doi.doi_data_block_size ||
2137 		    nblkptr < doi.doi_nblkptr ||
2138 		    dn_slots != doi.doi_dnodesize >> DNODE_SHIFT) {
2139 			err = dmu_free_long_range(rwa->os, drro->drr_object,
2140 			    0, DMU_OBJECT_END);
2141 			if (err != 0)
2142 				return (SET_ERROR(EINVAL));
2143 		}
2144 	} else if (err == EEXIST) {
2145 		/*
2146 		 * The object requested is currently an interior slot of a
2147 		 * multi-slot dnode. This will be resolved when the next txg
2148 		 * is synced out, since the send stream will have told us
2149 		 * to free this slot when we freed the associated dnode
2150 		 * earlier in the stream.
2151 		 */
2152 		txg_wait_synced(dmu_objset_pool(rwa->os), 0);
2153 		object = drro->drr_object;
2154 	} else {
2155 		/* object is free and we are about to allocate a new one */
2156 		object = DMU_NEW_OBJECT;
2157 	}
2158 
2159 	/*
2160 	 * If this is a multi-slot dnode there is a chance that this
2161 	 * object will expand into a slot that is already used by
2162 	 * another object from the previous snapshot. We must free
2163 	 * these objects before we attempt to allocate the new dnode.
2164 	 */
2165 	if (dn_slots > 1) {
2166 		boolean_t need_sync = B_FALSE;
2167 
2168 		for (uint64_t slot = drro->drr_object + 1;
2169 		    slot < drro->drr_object + dn_slots;
2170 		    slot++) {
2171 			dmu_object_info_t slot_doi;
2172 
2173 			err = dmu_object_info(rwa->os, slot, &slot_doi);
2174 			if (err == ENOENT || err == EEXIST)
2175 				continue;
2176 			else if (err != 0)
2177 				return (err);
2178 
2179 			err = dmu_free_long_object(rwa->os, slot);
2180 
2181 			if (err != 0)
2182 				return (err);
2183 
2184 			need_sync = B_TRUE;
2185 		}
2186 
2187 		if (need_sync)
2188 			txg_wait_synced(dmu_objset_pool(rwa->os), 0);
2189 	}
2190 
2191 	tx = dmu_tx_create(rwa->os);
2192 	dmu_tx_hold_bonus(tx, object);
2193 	err = dmu_tx_assign(tx, TXG_WAIT);
2194 	if (err != 0) {
2195 		dmu_tx_abort(tx);
2196 		return (err);
2197 	}
2198 
2199 	if (object == DMU_NEW_OBJECT) {
2200 		/* currently free, want to be allocated */
2201 		err = dmu_object_claim_dnsize(rwa->os, drro->drr_object,
2202 		    drro->drr_type, drro->drr_blksz,
2203 		    drro->drr_bonustype, drro->drr_bonuslen,
2204 		    dn_slots << DNODE_SHIFT, tx);
2205 	} else if (drro->drr_type != doi.doi_type ||
2206 	    drro->drr_blksz != doi.doi_data_block_size ||
2207 	    drro->drr_bonustype != doi.doi_bonus_type ||
2208 	    drro->drr_bonuslen != doi.doi_bonus_size ||
2209 	    drro->drr_dn_slots != (doi.doi_dnodesize >> DNODE_SHIFT)) {
2210 		/* currently allocated, but with different properties */
2211 		err = dmu_object_reclaim_dnsize(rwa->os, drro->drr_object,
2212 		    drro->drr_type, drro->drr_blksz,
2213 		    drro->drr_bonustype, drro->drr_bonuslen,
2214 		    drro->drr_dn_slots << DNODE_SHIFT, tx);
2215 	}
2216 	if (err != 0) {
2217 		dmu_tx_commit(tx);
2218 		return (SET_ERROR(EINVAL));
2219 	}
2220 
2221 	dmu_object_set_checksum(rwa->os, drro->drr_object,
2222 	    drro->drr_checksumtype, tx);
2223 	dmu_object_set_compress(rwa->os, drro->drr_object,
2224 	    drro->drr_compress, tx);
2225 
2226 	if (data != NULL) {
2227 		dmu_buf_t *db;
2228 
2229 		VERIFY0(dmu_bonus_hold(rwa->os, drro->drr_object, FTAG, &db));
2230 		dmu_buf_will_dirty(db, tx);
2231 
2232 		ASSERT3U(db->db_size, >=, drro->drr_bonuslen);
2233 		bcopy(data, db->db_data, drro->drr_bonuslen);
2234 		if (rwa->byteswap) {
2235 			dmu_object_byteswap_t byteswap =
2236 			    DMU_OT_BYTESWAP(drro->drr_bonustype);
2237 			dmu_ot_byteswap[byteswap].ob_func(db->db_data,
2238 			    drro->drr_bonuslen);
2239 		}
2240 		dmu_buf_rele(db, FTAG);
2241 	}
2242 	dmu_tx_commit(tx);
2243 
2244 	return (0);
2245 }
2246 
2247 /* ARGSUSED */
2248 static int
2249 receive_freeobjects(struct receive_writer_arg *rwa,
2250     struct drr_freeobjects *drrfo)
2251 {
2252 	uint64_t obj;
2253 	int next_err = 0;
2254 
2255 	if (drrfo->drr_firstobj + drrfo->drr_numobjs < drrfo->drr_firstobj)
2256 		return (SET_ERROR(EINVAL));
2257 
2258 	for (obj = drrfo->drr_firstobj == 0 ? 1 : drrfo->drr_firstobj;
2259 	    obj < drrfo->drr_firstobj + drrfo->drr_numobjs && next_err == 0;
2260 	    next_err = dmu_object_next(rwa->os, &obj, FALSE, 0)) {
2261 		int err;
2262 
2263 		err = dmu_object_info(rwa->os, obj, NULL);
2264 		if (err == ENOENT)
2265 			continue;
2266 		else if (err != 0)
2267 			return (err);
2268 
2269 		err = dmu_free_long_object(rwa->os, obj);
2270 		if (err != 0)
2271 			return (err);
2272 
2273 		if (obj > rwa->max_object)
2274 			rwa->max_object = obj;
2275 	}
2276 	if (next_err != ESRCH)
2277 		return (next_err);
2278 	return (0);
2279 }
2280 
2281 static int
2282 receive_write(struct receive_writer_arg *rwa, struct drr_write *drrw,
2283     arc_buf_t *abuf)
2284 {
2285 	dmu_tx_t *tx;
2286 	int err;
2287 
2288 	if (drrw->drr_offset + drrw->drr_logical_size < drrw->drr_offset ||
2289 	    !DMU_OT_IS_VALID(drrw->drr_type))
2290 		return (SET_ERROR(EINVAL));
2291 
2292 	/*
2293 	 * For resuming to work, records must be in increasing order
2294 	 * by (object, offset).
2295 	 */
2296 	if (drrw->drr_object < rwa->last_object ||
2297 	    (drrw->drr_object == rwa->last_object &&
2298 	    drrw->drr_offset < rwa->last_offset)) {
2299 		return (SET_ERROR(EINVAL));
2300 	}
2301 	rwa->last_object = drrw->drr_object;
2302 	rwa->last_offset = drrw->drr_offset;
2303 
2304 	if (rwa->last_object > rwa->max_object)
2305 		rwa->max_object = rwa->last_object;
2306 
2307 	if (dmu_object_info(rwa->os, drrw->drr_object, NULL) != 0)
2308 		return (SET_ERROR(EINVAL));
2309 
2310 	tx = dmu_tx_create(rwa->os);
2311 
2312 	dmu_tx_hold_write(tx, drrw->drr_object,
2313 	    drrw->drr_offset, drrw->drr_logical_size);
2314 	err = dmu_tx_assign(tx, TXG_WAIT);
2315 	if (err != 0) {
2316 		dmu_tx_abort(tx);
2317 		return (err);
2318 	}
2319 	if (rwa->byteswap) {
2320 		dmu_object_byteswap_t byteswap =
2321 		    DMU_OT_BYTESWAP(drrw->drr_type);
2322 		dmu_ot_byteswap[byteswap].ob_func(abuf->b_data,
2323 		    DRR_WRITE_PAYLOAD_SIZE(drrw));
2324 	}
2325 
2326 	/* use the bonus buf to look up the dnode in dmu_assign_arcbuf */
2327 	dmu_buf_t *bonus;
2328 	if (dmu_bonus_hold(rwa->os, drrw->drr_object, FTAG, &bonus) != 0)
2329 		return (SET_ERROR(EINVAL));
2330 	dmu_assign_arcbuf(bonus, drrw->drr_offset, abuf, tx);
2331 
2332 	/*
2333 	 * Note: If the receive fails, we want the resume stream to start
2334 	 * with the same record that we last successfully received (as opposed
2335 	 * to the next record), so that we can verify that we are
2336 	 * resuming from the correct location.
2337 	 */
2338 	save_resume_state(rwa, drrw->drr_object, drrw->drr_offset, tx);
2339 	dmu_tx_commit(tx);
2340 	dmu_buf_rele(bonus, FTAG);
2341 
2342 	return (0);
2343 }
2344 
2345 /*
2346  * Handle a DRR_WRITE_BYREF record.  This record is used in dedup'ed
2347  * streams to refer to a copy of the data that is already on the
2348  * system because it came in earlier in the stream.  This function
2349  * finds the earlier copy of the data, and uses that copy instead of
2350  * data from the stream to fulfill this write.
2351  */
2352 static int
2353 receive_write_byref(struct receive_writer_arg *rwa,
2354     struct drr_write_byref *drrwbr)
2355 {
2356 	dmu_tx_t *tx;
2357 	int err;
2358 	guid_map_entry_t gmesrch;
2359 	guid_map_entry_t *gmep;
2360 	avl_index_t where;
2361 	objset_t *ref_os = NULL;
2362 	dmu_buf_t *dbp;
2363 
2364 	if (drrwbr->drr_offset + drrwbr->drr_length < drrwbr->drr_offset)
2365 		return (SET_ERROR(EINVAL));
2366 
2367 	/*
2368 	 * If the GUID of the referenced dataset is different from the
2369 	 * GUID of the target dataset, find the referenced dataset.
2370 	 */
2371 	if (drrwbr->drr_toguid != drrwbr->drr_refguid) {
2372 		gmesrch.guid = drrwbr->drr_refguid;
2373 		if ((gmep = avl_find(rwa->guid_to_ds_map, &gmesrch,
2374 		    &where)) == NULL) {
2375 			return (SET_ERROR(EINVAL));
2376 		}
2377 		if (dmu_objset_from_ds(gmep->gme_ds, &ref_os))
2378 			return (SET_ERROR(EINVAL));
2379 	} else {
2380 		ref_os = rwa->os;
2381 	}
2382 
2383 	if (drrwbr->drr_object > rwa->max_object)
2384 		rwa->max_object = drrwbr->drr_object;
2385 
2386 	err = dmu_buf_hold(ref_os, drrwbr->drr_refobject,
2387 	    drrwbr->drr_refoffset, FTAG, &dbp, DMU_READ_PREFETCH);
2388 	if (err != 0)
2389 		return (err);
2390 
2391 	tx = dmu_tx_create(rwa->os);
2392 
2393 	dmu_tx_hold_write(tx, drrwbr->drr_object,
2394 	    drrwbr->drr_offset, drrwbr->drr_length);
2395 	err = dmu_tx_assign(tx, TXG_WAIT);
2396 	if (err != 0) {
2397 		dmu_tx_abort(tx);
2398 		return (err);
2399 	}
2400 	dmu_write(rwa->os, drrwbr->drr_object,
2401 	    drrwbr->drr_offset, drrwbr->drr_length, dbp->db_data, tx);
2402 	dmu_buf_rele(dbp, FTAG);
2403 
2404 	/* See comment in restore_write. */
2405 	save_resume_state(rwa, drrwbr->drr_object, drrwbr->drr_offset, tx);
2406 	dmu_tx_commit(tx);
2407 	return (0);
2408 }
2409 
2410 static int
2411 receive_write_embedded(struct receive_writer_arg *rwa,
2412     struct drr_write_embedded *drrwe, void *data)
2413 {
2414 	dmu_tx_t *tx;
2415 	int err;
2416 
2417 	if (drrwe->drr_offset + drrwe->drr_length < drrwe->drr_offset)
2418 		return (EINVAL);
2419 
2420 	if (drrwe->drr_psize > BPE_PAYLOAD_SIZE)
2421 		return (EINVAL);
2422 
2423 	if (drrwe->drr_etype >= NUM_BP_EMBEDDED_TYPES)
2424 		return (EINVAL);
2425 	if (drrwe->drr_compression >= ZIO_COMPRESS_FUNCTIONS)
2426 		return (EINVAL);
2427 
2428 	if (drrwe->drr_object > rwa->max_object)
2429 		rwa->max_object = drrwe->drr_object;
2430 
2431 	tx = dmu_tx_create(rwa->os);
2432 
2433 	dmu_tx_hold_write(tx, drrwe->drr_object,
2434 	    drrwe->drr_offset, drrwe->drr_length);
2435 	err = dmu_tx_assign(tx, TXG_WAIT);
2436 	if (err != 0) {
2437 		dmu_tx_abort(tx);
2438 		return (err);
2439 	}
2440 
2441 	dmu_write_embedded(rwa->os, drrwe->drr_object,
2442 	    drrwe->drr_offset, data, drrwe->drr_etype,
2443 	    drrwe->drr_compression, drrwe->drr_lsize, drrwe->drr_psize,
2444 	    rwa->byteswap ^ ZFS_HOST_BYTEORDER, tx);
2445 
2446 	/* See comment in restore_write. */
2447 	save_resume_state(rwa, drrwe->drr_object, drrwe->drr_offset, tx);
2448 	dmu_tx_commit(tx);
2449 	return (0);
2450 }
2451 
2452 static int
2453 receive_spill(struct receive_writer_arg *rwa, struct drr_spill *drrs,
2454     void *data)
2455 {
2456 	dmu_tx_t *tx;
2457 	dmu_buf_t *db, *db_spill;
2458 	int err;
2459 
2460 	if (drrs->drr_length < SPA_MINBLOCKSIZE ||
2461 	    drrs->drr_length > spa_maxblocksize(dmu_objset_spa(rwa->os)))
2462 		return (SET_ERROR(EINVAL));
2463 
2464 	if (dmu_object_info(rwa->os, drrs->drr_object, NULL) != 0)
2465 		return (SET_ERROR(EINVAL));
2466 
2467 	if (drrs->drr_object > rwa->max_object)
2468 		rwa->max_object = drrs->drr_object;
2469 
2470 	VERIFY0(dmu_bonus_hold(rwa->os, drrs->drr_object, FTAG, &db));
2471 	if ((err = dmu_spill_hold_by_bonus(db, FTAG, &db_spill)) != 0) {
2472 		dmu_buf_rele(db, FTAG);
2473 		return (err);
2474 	}
2475 
2476 	tx = dmu_tx_create(rwa->os);
2477 
2478 	dmu_tx_hold_spill(tx, db->db_object);
2479 
2480 	err = dmu_tx_assign(tx, TXG_WAIT);
2481 	if (err != 0) {
2482 		dmu_buf_rele(db, FTAG);
2483 		dmu_buf_rele(db_spill, FTAG);
2484 		dmu_tx_abort(tx);
2485 		return (err);
2486 	}
2487 	dmu_buf_will_dirty(db_spill, tx);
2488 
2489 	if (db_spill->db_size < drrs->drr_length)
2490 		VERIFY(0 == dbuf_spill_set_blksz(db_spill,
2491 		    drrs->drr_length, tx));
2492 	bcopy(data, db_spill->db_data, drrs->drr_length);
2493 
2494 	dmu_buf_rele(db, FTAG);
2495 	dmu_buf_rele(db_spill, FTAG);
2496 
2497 	dmu_tx_commit(tx);
2498 	return (0);
2499 }
2500 
2501 /* ARGSUSED */
2502 static int
2503 receive_free(struct receive_writer_arg *rwa, struct drr_free *drrf)
2504 {
2505 	int err;
2506 
2507 	if (drrf->drr_length != -1ULL &&
2508 	    drrf->drr_offset + drrf->drr_length < drrf->drr_offset)
2509 		return (SET_ERROR(EINVAL));
2510 
2511 	if (dmu_object_info(rwa->os, drrf->drr_object, NULL) != 0)
2512 		return (SET_ERROR(EINVAL));
2513 
2514 	if (drrf->drr_object > rwa->max_object)
2515 		rwa->max_object = drrf->drr_object;
2516 
2517 	err = dmu_free_long_range(rwa->os, drrf->drr_object,
2518 	    drrf->drr_offset, drrf->drr_length);
2519 
2520 	return (err);
2521 }
2522 
2523 /* used to destroy the drc_ds on error */
2524 static void
2525 dmu_recv_cleanup_ds(dmu_recv_cookie_t *drc)
2526 {
2527 	if (drc->drc_resumable) {
2528 		/* wait for our resume state to be written to disk */
2529 		txg_wait_synced(drc->drc_ds->ds_dir->dd_pool, 0);
2530 		dsl_dataset_disown(drc->drc_ds, dmu_recv_tag);
2531 	} else {
2532 		char name[ZFS_MAX_DATASET_NAME_LEN];
2533 		dsl_dataset_name(drc->drc_ds, name);
2534 		dsl_dataset_disown(drc->drc_ds, dmu_recv_tag);
2535 		(void) dsl_destroy_head(name);
2536 	}
2537 }
2538 
2539 static void
2540 receive_cksum(struct receive_arg *ra, int len, void *buf)
2541 {
2542 	if (ra->byteswap) {
2543 		(void) fletcher_4_incremental_byteswap(buf, len, &ra->cksum);
2544 	} else {
2545 		(void) fletcher_4_incremental_native(buf, len, &ra->cksum);
2546 	}
2547 }
2548 
2549 /*
2550  * Read the payload into a buffer of size len, and update the current record's
2551  * payload field.
2552  * Allocate ra->next_rrd and read the next record's header into
2553  * ra->next_rrd->header.
2554  * Verify checksum of payload and next record.
2555  */
2556 static int
2557 receive_read_payload_and_next_header(struct receive_arg *ra, int len, void *buf)
2558 {
2559 	int err;
2560 
2561 	if (len != 0) {
2562 		ASSERT3U(len, <=, SPA_MAXBLOCKSIZE);
2563 		err = receive_read(ra, len, buf);
2564 		if (err != 0)
2565 			return (err);
2566 		receive_cksum(ra, len, buf);
2567 
2568 		/* note: rrd is NULL when reading the begin record's payload */
2569 		if (ra->rrd != NULL) {
2570 			ra->rrd->payload = buf;
2571 			ra->rrd->payload_size = len;
2572 			ra->rrd->bytes_read = ra->bytes_read;
2573 		}
2574 	}
2575 
2576 	ra->prev_cksum = ra->cksum;
2577 
2578 	ra->next_rrd = kmem_zalloc(sizeof (*ra->next_rrd), KM_SLEEP);
2579 	err = receive_read(ra, sizeof (ra->next_rrd->header),
2580 	    &ra->next_rrd->header);
2581 	ra->next_rrd->bytes_read = ra->bytes_read;
2582 	if (err != 0) {
2583 		kmem_free(ra->next_rrd, sizeof (*ra->next_rrd));
2584 		ra->next_rrd = NULL;
2585 		return (err);
2586 	}
2587 	if (ra->next_rrd->header.drr_type == DRR_BEGIN) {
2588 		kmem_free(ra->next_rrd, sizeof (*ra->next_rrd));
2589 		ra->next_rrd = NULL;
2590 		return (SET_ERROR(EINVAL));
2591 	}
2592 
2593 	/*
2594 	 * Note: checksum is of everything up to but not including the
2595 	 * checksum itself.
2596 	 */
2597 	ASSERT3U(offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
2598 	    ==, sizeof (dmu_replay_record_t) - sizeof (zio_cksum_t));
2599 	receive_cksum(ra,
2600 	    offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
2601 	    &ra->next_rrd->header);
2602 
2603 	zio_cksum_t cksum_orig =
2604 	    ra->next_rrd->header.drr_u.drr_checksum.drr_checksum;
2605 	zio_cksum_t *cksump =
2606 	    &ra->next_rrd->header.drr_u.drr_checksum.drr_checksum;
2607 
2608 	if (ra->byteswap)
2609 		byteswap_record(&ra->next_rrd->header);
2610 
2611 	if ((!ZIO_CHECKSUM_IS_ZERO(cksump)) &&
2612 	    !ZIO_CHECKSUM_EQUAL(ra->cksum, *cksump)) {
2613 		kmem_free(ra->next_rrd, sizeof (*ra->next_rrd));
2614 		ra->next_rrd = NULL;
2615 		return (SET_ERROR(ECKSUM));
2616 	}
2617 
2618 	receive_cksum(ra, sizeof (cksum_orig), &cksum_orig);
2619 
2620 	return (0);
2621 }
2622 
2623 static void
2624 objlist_create(struct objlist *list)
2625 {
2626 	list_create(&list->list, sizeof (struct receive_objnode),
2627 	    offsetof(struct receive_objnode, node));
2628 	list->last_lookup = 0;
2629 }
2630 
2631 static void
2632 objlist_destroy(struct objlist *list)
2633 {
2634 	for (struct receive_objnode *n = list_remove_head(&list->list);
2635 	    n != NULL; n = list_remove_head(&list->list)) {
2636 		kmem_free(n, sizeof (*n));
2637 	}
2638 	list_destroy(&list->list);
2639 }
2640 
2641 /*
2642  * This function looks through the objlist to see if the specified object number
2643  * is contained in the objlist.  In the process, it will remove all object
2644  * numbers in the list that are smaller than the specified object number.  Thus,
2645  * any lookup of an object number smaller than a previously looked up object
2646  * number will always return false; therefore, all lookups should be done in
2647  * ascending order.
2648  */
2649 static boolean_t
2650 objlist_exists(struct objlist *list, uint64_t object)
2651 {
2652 	struct receive_objnode *node = list_head(&list->list);
2653 	ASSERT3U(object, >=, list->last_lookup);
2654 	list->last_lookup = object;
2655 	while (node != NULL && node->object < object) {
2656 		VERIFY3P(node, ==, list_remove_head(&list->list));
2657 		kmem_free(node, sizeof (*node));
2658 		node = list_head(&list->list);
2659 	}
2660 	return (node != NULL && node->object == object);
2661 }
2662 
2663 /*
2664  * The objlist is a list of object numbers stored in ascending order.  However,
2665  * the insertion of new object numbers does not seek out the correct location to
2666  * store a new object number; instead, it appends it to the list for simplicity.
2667  * Thus, any users must take care to only insert new object numbers in ascending
2668  * order.
2669  */
2670 static void
2671 objlist_insert(struct objlist *list, uint64_t object)
2672 {
2673 	struct receive_objnode *node = kmem_zalloc(sizeof (*node), KM_SLEEP);
2674 	node->object = object;
2675 #ifdef ZFS_DEBUG
2676 	struct receive_objnode *last_object = list_tail(&list->list);
2677 	uint64_t last_objnum = (last_object != NULL ? last_object->object : 0);
2678 	ASSERT3U(node->object, >, last_objnum);
2679 #endif
2680 	list_insert_tail(&list->list, node);
2681 }
2682 
2683 /*
2684  * Issue the prefetch reads for any necessary indirect blocks.
2685  *
2686  * We use the object ignore list to tell us whether or not to issue prefetches
2687  * for a given object.  We do this for both correctness (in case the blocksize
2688  * of an object has changed) and performance (if the object doesn't exist, don't
2689  * needlessly try to issue prefetches).  We also trim the list as we go through
2690  * the stream to prevent it from growing to an unbounded size.
2691  *
2692  * The object numbers within will always be in sorted order, and any write
2693  * records we see will also be in sorted order, but they're not sorted with
2694  * respect to each other (i.e. we can get several object records before
2695  * receiving each object's write records).  As a result, once we've reached a
2696  * given object number, we can safely remove any reference to lower object
2697  * numbers in the ignore list. In practice, we receive up to 32 object records
2698  * before receiving write records, so the list can have up to 32 nodes in it.
2699  */
2700 /* ARGSUSED */
2701 static void
2702 receive_read_prefetch(struct receive_arg *ra,
2703     uint64_t object, uint64_t offset, uint64_t length)
2704 {
2705 	if (!objlist_exists(&ra->ignore_objlist, object)) {
2706 		dmu_prefetch(ra->os, object, 1, offset, length,
2707 		    ZIO_PRIORITY_SYNC_READ);
2708 	}
2709 }
2710 
2711 /*
2712  * Read records off the stream, issuing any necessary prefetches.
2713  */
2714 static int
2715 receive_read_record(struct receive_arg *ra)
2716 {
2717 	int err;
2718 
2719 	switch (ra->rrd->header.drr_type) {
2720 	case DRR_OBJECT:
2721 	{
2722 		struct drr_object *drro = &ra->rrd->header.drr_u.drr_object;
2723 		uint32_t size = P2ROUNDUP(drro->drr_bonuslen, 8);
2724 		void *buf = kmem_zalloc(size, KM_SLEEP);
2725 		dmu_object_info_t doi;
2726 		err = receive_read_payload_and_next_header(ra, size, buf);
2727 		if (err != 0) {
2728 			kmem_free(buf, size);
2729 			return (err);
2730 		}
2731 		err = dmu_object_info(ra->os, drro->drr_object, &doi);
2732 		/*
2733 		 * See receive_read_prefetch for an explanation why we're
2734 		 * storing this object in the ignore_obj_list.
2735 		 */
2736 		if (err == ENOENT ||
2737 		    (err == 0 && doi.doi_data_block_size != drro->drr_blksz)) {
2738 			objlist_insert(&ra->ignore_objlist, drro->drr_object);
2739 			err = 0;
2740 		}
2741 		return (err);
2742 	}
2743 	case DRR_FREEOBJECTS:
2744 	{
2745 		err = receive_read_payload_and_next_header(ra, 0, NULL);
2746 		return (err);
2747 	}
2748 	case DRR_WRITE:
2749 	{
2750 		struct drr_write *drrw = &ra->rrd->header.drr_u.drr_write;
2751 		arc_buf_t *abuf;
2752 		boolean_t is_meta = DMU_OT_IS_METADATA(drrw->drr_type);
2753 		if (DRR_WRITE_COMPRESSED(drrw)) {
2754 			ASSERT3U(drrw->drr_compressed_size, >, 0);
2755 			ASSERT3U(drrw->drr_logical_size, >=,
2756 			    drrw->drr_compressed_size);
2757 			ASSERT(!is_meta);
2758 			abuf = arc_loan_compressed_buf(
2759 			    dmu_objset_spa(ra->os),
2760 			    drrw->drr_compressed_size, drrw->drr_logical_size,
2761 			    drrw->drr_compressiontype);
2762 		} else {
2763 			abuf = arc_loan_buf(dmu_objset_spa(ra->os),
2764 			    is_meta, drrw->drr_logical_size);
2765 		}
2766 
2767 		err = receive_read_payload_and_next_header(ra,
2768 		    DRR_WRITE_PAYLOAD_SIZE(drrw), abuf->b_data);
2769 		if (err != 0) {
2770 			dmu_return_arcbuf(abuf);
2771 			return (err);
2772 		}
2773 		ra->rrd->write_buf = abuf;
2774 		receive_read_prefetch(ra, drrw->drr_object, drrw->drr_offset,
2775 		    drrw->drr_logical_size);
2776 		return (err);
2777 	}
2778 	case DRR_WRITE_BYREF:
2779 	{
2780 		struct drr_write_byref *drrwb =
2781 		    &ra->rrd->header.drr_u.drr_write_byref;
2782 		err = receive_read_payload_and_next_header(ra, 0, NULL);
2783 		receive_read_prefetch(ra, drrwb->drr_object, drrwb->drr_offset,
2784 		    drrwb->drr_length);
2785 		return (err);
2786 	}
2787 	case DRR_WRITE_EMBEDDED:
2788 	{
2789 		struct drr_write_embedded *drrwe =
2790 		    &ra->rrd->header.drr_u.drr_write_embedded;
2791 		uint32_t size = P2ROUNDUP(drrwe->drr_psize, 8);
2792 		void *buf = kmem_zalloc(size, KM_SLEEP);
2793 
2794 		err = receive_read_payload_and_next_header(ra, size, buf);
2795 		if (err != 0) {
2796 			kmem_free(buf, size);
2797 			return (err);
2798 		}
2799 
2800 		receive_read_prefetch(ra, drrwe->drr_object, drrwe->drr_offset,
2801 		    drrwe->drr_length);
2802 		return (err);
2803 	}
2804 	case DRR_FREE:
2805 	{
2806 		/*
2807 		 * It might be beneficial to prefetch indirect blocks here, but
2808 		 * we don't really have the data to decide for sure.
2809 		 */
2810 		err = receive_read_payload_and_next_header(ra, 0, NULL);
2811 		return (err);
2812 	}
2813 	case DRR_END:
2814 	{
2815 		struct drr_end *drre = &ra->rrd->header.drr_u.drr_end;
2816 		if (!ZIO_CHECKSUM_EQUAL(ra->prev_cksum, drre->drr_checksum))
2817 			return (SET_ERROR(ECKSUM));
2818 		return (0);
2819 	}
2820 	case DRR_SPILL:
2821 	{
2822 		struct drr_spill *drrs = &ra->rrd->header.drr_u.drr_spill;
2823 		void *buf = kmem_zalloc(drrs->drr_length, KM_SLEEP);
2824 		err = receive_read_payload_and_next_header(ra, drrs->drr_length,
2825 		    buf);
2826 		if (err != 0)
2827 			kmem_free(buf, drrs->drr_length);
2828 		return (err);
2829 	}
2830 	default:
2831 		return (SET_ERROR(EINVAL));
2832 	}
2833 }
2834 
2835 /*
2836  * Commit the records to the pool.
2837  */
2838 static int
2839 receive_process_record(struct receive_writer_arg *rwa,
2840     struct receive_record_arg *rrd)
2841 {
2842 	int err;
2843 
2844 	/* Processing in order, therefore bytes_read should be increasing. */
2845 	ASSERT3U(rrd->bytes_read, >=, rwa->bytes_read);
2846 	rwa->bytes_read = rrd->bytes_read;
2847 
2848 	switch (rrd->header.drr_type) {
2849 	case DRR_OBJECT:
2850 	{
2851 		struct drr_object *drro = &rrd->header.drr_u.drr_object;
2852 		err = receive_object(rwa, drro, rrd->payload);
2853 		kmem_free(rrd->payload, rrd->payload_size);
2854 		rrd->payload = NULL;
2855 		return (err);
2856 	}
2857 	case DRR_FREEOBJECTS:
2858 	{
2859 		struct drr_freeobjects *drrfo =
2860 		    &rrd->header.drr_u.drr_freeobjects;
2861 		return (receive_freeobjects(rwa, drrfo));
2862 	}
2863 	case DRR_WRITE:
2864 	{
2865 		struct drr_write *drrw = &rrd->header.drr_u.drr_write;
2866 		err = receive_write(rwa, drrw, rrd->write_buf);
2867 		/* if receive_write() is successful, it consumes the arc_buf */
2868 		if (err != 0)
2869 			dmu_return_arcbuf(rrd->write_buf);
2870 		rrd->write_buf = NULL;
2871 		rrd->payload = NULL;
2872 		return (err);
2873 	}
2874 	case DRR_WRITE_BYREF:
2875 	{
2876 		struct drr_write_byref *drrwbr =
2877 		    &rrd->header.drr_u.drr_write_byref;
2878 		return (receive_write_byref(rwa, drrwbr));
2879 	}
2880 	case DRR_WRITE_EMBEDDED:
2881 	{
2882 		struct drr_write_embedded *drrwe =
2883 		    &rrd->header.drr_u.drr_write_embedded;
2884 		err = receive_write_embedded(rwa, drrwe, rrd->payload);
2885 		kmem_free(rrd->payload, rrd->payload_size);
2886 		rrd->payload = NULL;
2887 		return (err);
2888 	}
2889 	case DRR_FREE:
2890 	{
2891 		struct drr_free *drrf = &rrd->header.drr_u.drr_free;
2892 		return (receive_free(rwa, drrf));
2893 	}
2894 	case DRR_SPILL:
2895 	{
2896 		struct drr_spill *drrs = &rrd->header.drr_u.drr_spill;
2897 		err = receive_spill(rwa, drrs, rrd->payload);
2898 		kmem_free(rrd->payload, rrd->payload_size);
2899 		rrd->payload = NULL;
2900 		return (err);
2901 	}
2902 	default:
2903 		return (SET_ERROR(EINVAL));
2904 	}
2905 }
2906 
2907 /*
2908  * dmu_recv_stream's worker thread; pull records off the queue, and then call
2909  * receive_process_record  When we're done, signal the main thread and exit.
2910  */
2911 static void
2912 receive_writer_thread(void *arg)
2913 {
2914 	struct receive_writer_arg *rwa = arg;
2915 	struct receive_record_arg *rrd;
2916 	for (rrd = bqueue_dequeue(&rwa->q); !rrd->eos_marker;
2917 	    rrd = bqueue_dequeue(&rwa->q)) {
2918 		/*
2919 		 * If there's an error, the main thread will stop putting things
2920 		 * on the queue, but we need to clear everything in it before we
2921 		 * can exit.
2922 		 */
2923 		if (rwa->err == 0) {
2924 			rwa->err = receive_process_record(rwa, rrd);
2925 		} else if (rrd->write_buf != NULL) {
2926 			dmu_return_arcbuf(rrd->write_buf);
2927 			rrd->write_buf = NULL;
2928 			rrd->payload = NULL;
2929 		} else if (rrd->payload != NULL) {
2930 			kmem_free(rrd->payload, rrd->payload_size);
2931 			rrd->payload = NULL;
2932 		}
2933 		kmem_free(rrd, sizeof (*rrd));
2934 	}
2935 	kmem_free(rrd, sizeof (*rrd));
2936 	mutex_enter(&rwa->mutex);
2937 	rwa->done = B_TRUE;
2938 	cv_signal(&rwa->cv);
2939 	mutex_exit(&rwa->mutex);
2940 	thread_exit();
2941 }
2942 
2943 static int
2944 resume_check(struct receive_arg *ra, nvlist_t *begin_nvl)
2945 {
2946 	uint64_t val;
2947 	objset_t *mos = dmu_objset_pool(ra->os)->dp_meta_objset;
2948 	uint64_t dsobj = dmu_objset_id(ra->os);
2949 	uint64_t resume_obj, resume_off;
2950 
2951 	if (nvlist_lookup_uint64(begin_nvl,
2952 	    "resume_object", &resume_obj) != 0 ||
2953 	    nvlist_lookup_uint64(begin_nvl,
2954 	    "resume_offset", &resume_off) != 0) {
2955 		return (SET_ERROR(EINVAL));
2956 	}
2957 	VERIFY0(zap_lookup(mos, dsobj,
2958 	    DS_FIELD_RESUME_OBJECT, sizeof (val), 1, &val));
2959 	if (resume_obj != val)
2960 		return (SET_ERROR(EINVAL));
2961 	VERIFY0(zap_lookup(mos, dsobj,
2962 	    DS_FIELD_RESUME_OFFSET, sizeof (val), 1, &val));
2963 	if (resume_off != val)
2964 		return (SET_ERROR(EINVAL));
2965 
2966 	return (0);
2967 }
2968 
2969 /*
2970  * Read in the stream's records, one by one, and apply them to the pool.  There
2971  * are two threads involved; the thread that calls this function will spin up a
2972  * worker thread, read the records off the stream one by one, and issue
2973  * prefetches for any necessary indirect blocks.  It will then push the records
2974  * onto an internal blocking queue.  The worker thread will pull the records off
2975  * the queue, and actually write the data into the DMU.  This way, the worker
2976  * thread doesn't have to wait for reads to complete, since everything it needs
2977  * (the indirect blocks) will be prefetched.
2978  *
2979  * NB: callers *must* call dmu_recv_end() if this succeeds.
2980  */
2981 int
2982 dmu_recv_stream(dmu_recv_cookie_t *drc, vnode_t *vp, offset_t *voffp,
2983     int cleanup_fd, uint64_t *action_handlep)
2984 {
2985 	int err = 0;
2986 	struct receive_arg ra = { 0 };
2987 	struct receive_writer_arg rwa = { 0 };
2988 	int featureflags;
2989 	nvlist_t *begin_nvl = NULL;
2990 
2991 	ra.byteswap = drc->drc_byteswap;
2992 	ra.cksum = drc->drc_cksum;
2993 	ra.vp = vp;
2994 	ra.voff = *voffp;
2995 
2996 	if (dsl_dataset_is_zapified(drc->drc_ds)) {
2997 		(void) zap_lookup(drc->drc_ds->ds_dir->dd_pool->dp_meta_objset,
2998 		    drc->drc_ds->ds_object, DS_FIELD_RESUME_BYTES,
2999 		    sizeof (ra.bytes_read), 1, &ra.bytes_read);
3000 	}
3001 
3002 	objlist_create(&ra.ignore_objlist);
3003 
3004 	/* these were verified in dmu_recv_begin */
3005 	ASSERT3U(DMU_GET_STREAM_HDRTYPE(drc->drc_drrb->drr_versioninfo), ==,
3006 	    DMU_SUBSTREAM);
3007 	ASSERT3U(drc->drc_drrb->drr_type, <, DMU_OST_NUMTYPES);
3008 
3009 	/*
3010 	 * Open the objset we are modifying.
3011 	 */
3012 	VERIFY0(dmu_objset_from_ds(drc->drc_ds, &ra.os));
3013 
3014 	ASSERT(dsl_dataset_phys(drc->drc_ds)->ds_flags & DS_FLAG_INCONSISTENT);
3015 
3016 	featureflags = DMU_GET_FEATUREFLAGS(drc->drc_drrb->drr_versioninfo);
3017 
3018 	/* if this stream is dedup'ed, set up the avl tree for guid mapping */
3019 	if (featureflags & DMU_BACKUP_FEATURE_DEDUP) {
3020 		minor_t minor;
3021 
3022 		if (cleanup_fd == -1) {
3023 			ra.err = SET_ERROR(EBADF);
3024 			goto out;
3025 		}
3026 		ra.err = zfs_onexit_fd_hold(cleanup_fd, &minor);
3027 		if (ra.err != 0) {
3028 			cleanup_fd = -1;
3029 			goto out;
3030 		}
3031 
3032 		if (*action_handlep == 0) {
3033 			rwa.guid_to_ds_map =
3034 			    kmem_alloc(sizeof (avl_tree_t), KM_SLEEP);
3035 			avl_create(rwa.guid_to_ds_map, guid_compare,
3036 			    sizeof (guid_map_entry_t),
3037 			    offsetof(guid_map_entry_t, avlnode));
3038 			err = zfs_onexit_add_cb(minor,
3039 			    free_guid_map_onexit, rwa.guid_to_ds_map,
3040 			    action_handlep);
3041 			if (ra.err != 0)
3042 				goto out;
3043 		} else {
3044 			err = zfs_onexit_cb_data(minor, *action_handlep,
3045 			    (void **)&rwa.guid_to_ds_map);
3046 			if (ra.err != 0)
3047 				goto out;
3048 		}
3049 
3050 		drc->drc_guid_to_ds_map = rwa.guid_to_ds_map;
3051 	}
3052 
3053 	uint32_t payloadlen = drc->drc_drr_begin->drr_payloadlen;
3054 	void *payload = NULL;
3055 	if (payloadlen != 0)
3056 		payload = kmem_alloc(payloadlen, KM_SLEEP);
3057 
3058 	err = receive_read_payload_and_next_header(&ra, payloadlen, payload);
3059 	if (err != 0) {
3060 		if (payloadlen != 0)
3061 			kmem_free(payload, payloadlen);
3062 		goto out;
3063 	}
3064 	if (payloadlen != 0) {
3065 		err = nvlist_unpack(payload, payloadlen, &begin_nvl, KM_SLEEP);
3066 		kmem_free(payload, payloadlen);
3067 		if (err != 0)
3068 			goto out;
3069 	}
3070 
3071 	if (featureflags & DMU_BACKUP_FEATURE_RESUMING) {
3072 		err = resume_check(&ra, begin_nvl);
3073 		if (err != 0)
3074 			goto out;
3075 	}
3076 
3077 	(void) bqueue_init(&rwa.q, zfs_recv_queue_length,
3078 	    offsetof(struct receive_record_arg, node));
3079 	cv_init(&rwa.cv, NULL, CV_DEFAULT, NULL);
3080 	mutex_init(&rwa.mutex, NULL, MUTEX_DEFAULT, NULL);
3081 	rwa.os = ra.os;
3082 	rwa.byteswap = drc->drc_byteswap;
3083 	rwa.resumable = drc->drc_resumable;
3084 
3085 	(void) thread_create(NULL, 0, receive_writer_thread, &rwa, 0, curproc,
3086 	    TS_RUN, minclsyspri);
3087 	/*
3088 	 * We're reading rwa.err without locks, which is safe since we are the
3089 	 * only reader, and the worker thread is the only writer.  It's ok if we
3090 	 * miss a write for an iteration or two of the loop, since the writer
3091 	 * thread will keep freeing records we send it until we send it an eos
3092 	 * marker.
3093 	 *
3094 	 * We can leave this loop in 3 ways:  First, if rwa.err is
3095 	 * non-zero.  In that case, the writer thread will free the rrd we just
3096 	 * pushed.  Second, if  we're interrupted; in that case, either it's the
3097 	 * first loop and ra.rrd was never allocated, or it's later, and ra.rrd
3098 	 * has been handed off to the writer thread who will free it.  Finally,
3099 	 * if receive_read_record fails or we're at the end of the stream, then
3100 	 * we free ra.rrd and exit.
3101 	 */
3102 	while (rwa.err == 0) {
3103 		if (issig(JUSTLOOKING) && issig(FORREAL)) {
3104 			err = SET_ERROR(EINTR);
3105 			break;
3106 		}
3107 
3108 		ASSERT3P(ra.rrd, ==, NULL);
3109 		ra.rrd = ra.next_rrd;
3110 		ra.next_rrd = NULL;
3111 		/* Allocates and loads header into ra.next_rrd */
3112 		err = receive_read_record(&ra);
3113 
3114 		if (ra.rrd->header.drr_type == DRR_END || err != 0) {
3115 			kmem_free(ra.rrd, sizeof (*ra.rrd));
3116 			ra.rrd = NULL;
3117 			break;
3118 		}
3119 
3120 		bqueue_enqueue(&rwa.q, ra.rrd,
3121 		    sizeof (struct receive_record_arg) + ra.rrd->payload_size);
3122 		ra.rrd = NULL;
3123 	}
3124 	if (ra.next_rrd == NULL)
3125 		ra.next_rrd = kmem_zalloc(sizeof (*ra.next_rrd), KM_SLEEP);
3126 	ra.next_rrd->eos_marker = B_TRUE;
3127 	bqueue_enqueue(&rwa.q, ra.next_rrd, 1);
3128 
3129 	mutex_enter(&rwa.mutex);
3130 	while (!rwa.done) {
3131 		cv_wait(&rwa.cv, &rwa.mutex);
3132 	}
3133 	mutex_exit(&rwa.mutex);
3134 
3135 	/*
3136 	 * If we are receiving a full stream as a clone, all object IDs which
3137 	 * are greater than the maximum ID referenced in the stream are
3138 	 * by definition unused and must be freed. Note that it's possible that
3139 	 * we've resumed this send and the first record we received was the END
3140 	 * record. In that case, max_object would be 0, but we shouldn't start
3141 	 * freeing all objects from there; instead we should start from the
3142 	 * resumeobj.
3143 	 */
3144 	if (drc->drc_clone && drc->drc_drrb->drr_fromguid == 0) {
3145 		uint64_t obj;
3146 		if (nvlist_lookup_uint64(begin_nvl, "resume_object", &obj) != 0)
3147 			obj = 0;
3148 		if (rwa.max_object > obj)
3149 			obj = rwa.max_object;
3150 		obj++;
3151 		int free_err = 0;
3152 		int next_err = 0;
3153 
3154 		while (next_err == 0) {
3155 			free_err = dmu_free_long_object(rwa.os, obj);
3156 			if (free_err != 0 && free_err != ENOENT)
3157 				break;
3158 
3159 			next_err = dmu_object_next(rwa.os, &obj, FALSE, 0);
3160 		}
3161 
3162 		if (err == 0) {
3163 			if (free_err != 0 && free_err != ENOENT)
3164 				err = free_err;
3165 			else if (next_err != ESRCH)
3166 				err = next_err;
3167 		}
3168 	}
3169 
3170 	cv_destroy(&rwa.cv);
3171 	mutex_destroy(&rwa.mutex);
3172 	bqueue_destroy(&rwa.q);
3173 	if (err == 0)
3174 		err = rwa.err;
3175 
3176 out:
3177 	nvlist_free(begin_nvl);
3178 	if ((featureflags & DMU_BACKUP_FEATURE_DEDUP) && (cleanup_fd != -1))
3179 		zfs_onexit_fd_rele(cleanup_fd);
3180 
3181 	if (err != 0) {
3182 		/*
3183 		 * Clean up references. If receive is not resumable,
3184 		 * destroy what we created, so we don't leave it in
3185 		 * the inconsistent state.
3186 		 */
3187 		dmu_recv_cleanup_ds(drc);
3188 	}
3189 
3190 	*voffp = ra.voff;
3191 	objlist_destroy(&ra.ignore_objlist);
3192 	return (err);
3193 }
3194 
3195 static int
3196 dmu_recv_end_check(void *arg, dmu_tx_t *tx)
3197 {
3198 	dmu_recv_cookie_t *drc = arg;
3199 	dsl_pool_t *dp = dmu_tx_pool(tx);
3200 	int error;
3201 
3202 	ASSERT3P(drc->drc_ds->ds_owner, ==, dmu_recv_tag);
3203 
3204 	if (!drc->drc_newfs) {
3205 		dsl_dataset_t *origin_head;
3206 
3207 		error = dsl_dataset_hold(dp, drc->drc_tofs, FTAG, &origin_head);
3208 		if (error != 0)
3209 			return (error);
3210 		if (drc->drc_force) {
3211 			/*
3212 			 * We will destroy any snapshots in tofs (i.e. before
3213 			 * origin_head) that are after the origin (which is
3214 			 * the snap before drc_ds, because drc_ds can not
3215 			 * have any snaps of its own).
3216 			 */
3217 			uint64_t obj;
3218 
3219 			obj = dsl_dataset_phys(origin_head)->ds_prev_snap_obj;
3220 			while (obj !=
3221 			    dsl_dataset_phys(drc->drc_ds)->ds_prev_snap_obj) {
3222 				dsl_dataset_t *snap;
3223 				error = dsl_dataset_hold_obj(dp, obj, FTAG,
3224 				    &snap);
3225 				if (error != 0)
3226 					break;
3227 				if (snap->ds_dir != origin_head->ds_dir)
3228 					error = SET_ERROR(EINVAL);
3229 				if (error == 0)  {
3230 					error = dsl_destroy_snapshot_check_impl(
3231 					    snap, B_FALSE);
3232 				}
3233 				obj = dsl_dataset_phys(snap)->ds_prev_snap_obj;
3234 				dsl_dataset_rele(snap, FTAG);
3235 				if (error != 0)
3236 					break;
3237 			}
3238 			if (error != 0) {
3239 				dsl_dataset_rele(origin_head, FTAG);
3240 				return (error);
3241 			}
3242 		}
3243 		error = dsl_dataset_clone_swap_check_impl(drc->drc_ds,
3244 		    origin_head, drc->drc_force, drc->drc_owner, tx);
3245 		if (error != 0) {
3246 			dsl_dataset_rele(origin_head, FTAG);
3247 			return (error);
3248 		}
3249 		error = dsl_dataset_snapshot_check_impl(origin_head,
3250 		    drc->drc_tosnap, tx, B_TRUE, 1, drc->drc_cred);
3251 		dsl_dataset_rele(origin_head, FTAG);
3252 		if (error != 0)
3253 			return (error);
3254 
3255 		error = dsl_destroy_head_check_impl(drc->drc_ds, 1);
3256 	} else {
3257 		error = dsl_dataset_snapshot_check_impl(drc->drc_ds,
3258 		    drc->drc_tosnap, tx, B_TRUE, 1, drc->drc_cred);
3259 	}
3260 	return (error);
3261 }
3262 
3263 static void
3264 dmu_recv_end_sync(void *arg, dmu_tx_t *tx)
3265 {
3266 	dmu_recv_cookie_t *drc = arg;
3267 	dsl_pool_t *dp = dmu_tx_pool(tx);
3268 
3269 	spa_history_log_internal_ds(drc->drc_ds, "finish receiving",
3270 	    tx, "snap=%s", drc->drc_tosnap);
3271 
3272 	if (!drc->drc_newfs) {
3273 		dsl_dataset_t *origin_head;
3274 
3275 		VERIFY0(dsl_dataset_hold(dp, drc->drc_tofs, FTAG,
3276 		    &origin_head));
3277 
3278 		if (drc->drc_force) {
3279 			/*
3280 			 * Destroy any snapshots of drc_tofs (origin_head)
3281 			 * after the origin (the snap before drc_ds).
3282 			 */
3283 			uint64_t obj;
3284 
3285 			obj = dsl_dataset_phys(origin_head)->ds_prev_snap_obj;
3286 			while (obj !=
3287 			    dsl_dataset_phys(drc->drc_ds)->ds_prev_snap_obj) {
3288 				dsl_dataset_t *snap;
3289 				VERIFY0(dsl_dataset_hold_obj(dp, obj, FTAG,
3290 				    &snap));
3291 				ASSERT3P(snap->ds_dir, ==, origin_head->ds_dir);
3292 				obj = dsl_dataset_phys(snap)->ds_prev_snap_obj;
3293 				dsl_destroy_snapshot_sync_impl(snap,
3294 				    B_FALSE, tx);
3295 				dsl_dataset_rele(snap, FTAG);
3296 			}
3297 		}
3298 		VERIFY3P(drc->drc_ds->ds_prev, ==,
3299 		    origin_head->ds_prev);
3300 
3301 		dsl_dataset_clone_swap_sync_impl(drc->drc_ds,
3302 		    origin_head, tx);
3303 		dsl_dataset_snapshot_sync_impl(origin_head,
3304 		    drc->drc_tosnap, tx);
3305 
3306 		/* set snapshot's creation time and guid */
3307 		dmu_buf_will_dirty(origin_head->ds_prev->ds_dbuf, tx);
3308 		dsl_dataset_phys(origin_head->ds_prev)->ds_creation_time =
3309 		    drc->drc_drrb->drr_creation_time;
3310 		dsl_dataset_phys(origin_head->ds_prev)->ds_guid =
3311 		    drc->drc_drrb->drr_toguid;
3312 		dsl_dataset_phys(origin_head->ds_prev)->ds_flags &=
3313 		    ~DS_FLAG_INCONSISTENT;
3314 
3315 		dmu_buf_will_dirty(origin_head->ds_dbuf, tx);
3316 		dsl_dataset_phys(origin_head)->ds_flags &=
3317 		    ~DS_FLAG_INCONSISTENT;
3318 
3319 		drc->drc_newsnapobj =
3320 		    dsl_dataset_phys(origin_head)->ds_prev_snap_obj;
3321 
3322 		dsl_dataset_rele(origin_head, FTAG);
3323 		dsl_destroy_head_sync_impl(drc->drc_ds, tx);
3324 
3325 		if (drc->drc_owner != NULL)
3326 			VERIFY3P(origin_head->ds_owner, ==, drc->drc_owner);
3327 	} else {
3328 		dsl_dataset_t *ds = drc->drc_ds;
3329 
3330 		dsl_dataset_snapshot_sync_impl(ds, drc->drc_tosnap, tx);
3331 
3332 		/* set snapshot's creation time and guid */
3333 		dmu_buf_will_dirty(ds->ds_prev->ds_dbuf, tx);
3334 		dsl_dataset_phys(ds->ds_prev)->ds_creation_time =
3335 		    drc->drc_drrb->drr_creation_time;
3336 		dsl_dataset_phys(ds->ds_prev)->ds_guid =
3337 		    drc->drc_drrb->drr_toguid;
3338 		dsl_dataset_phys(ds->ds_prev)->ds_flags &=
3339 		    ~DS_FLAG_INCONSISTENT;
3340 
3341 		dmu_buf_will_dirty(ds->ds_dbuf, tx);
3342 		dsl_dataset_phys(ds)->ds_flags &= ~DS_FLAG_INCONSISTENT;
3343 		if (dsl_dataset_has_resume_receive_state(ds)) {
3344 			(void) zap_remove(dp->dp_meta_objset, ds->ds_object,
3345 			    DS_FIELD_RESUME_FROMGUID, tx);
3346 			(void) zap_remove(dp->dp_meta_objset, ds->ds_object,
3347 			    DS_FIELD_RESUME_OBJECT, tx);
3348 			(void) zap_remove(dp->dp_meta_objset, ds->ds_object,
3349 			    DS_FIELD_RESUME_OFFSET, tx);
3350 			(void) zap_remove(dp->dp_meta_objset, ds->ds_object,
3351 			    DS_FIELD_RESUME_BYTES, tx);
3352 			(void) zap_remove(dp->dp_meta_objset, ds->ds_object,
3353 			    DS_FIELD_RESUME_TOGUID, tx);
3354 			(void) zap_remove(dp->dp_meta_objset, ds->ds_object,
3355 			    DS_FIELD_RESUME_TONAME, tx);
3356 		}
3357 		drc->drc_newsnapobj =
3358 		    dsl_dataset_phys(drc->drc_ds)->ds_prev_snap_obj;
3359 	}
3360 	/*
3361 	 * Release the hold from dmu_recv_begin.  This must be done before
3362 	 * we return to open context, so that when we free the dataset's dnode,
3363 	 * we can evict its bonus buffer.
3364 	 */
3365 	dsl_dataset_disown(drc->drc_ds, dmu_recv_tag);
3366 	drc->drc_ds = NULL;
3367 }
3368 
3369 static int
3370 add_ds_to_guidmap(const char *name, avl_tree_t *guid_map, uint64_t snapobj)
3371 {
3372 	dsl_pool_t *dp;
3373 	dsl_dataset_t *snapds;
3374 	guid_map_entry_t *gmep;
3375 	int err;
3376 
3377 	ASSERT(guid_map != NULL);
3378 
3379 	err = dsl_pool_hold(name, FTAG, &dp);
3380 	if (err != 0)
3381 		return (err);
3382 	gmep = kmem_alloc(sizeof (*gmep), KM_SLEEP);
3383 	err = dsl_dataset_hold_obj(dp, snapobj, gmep, &snapds);
3384 	if (err == 0) {
3385 		gmep->guid = dsl_dataset_phys(snapds)->ds_guid;
3386 		gmep->gme_ds = snapds;
3387 		avl_add(guid_map, gmep);
3388 		dsl_dataset_long_hold(snapds, gmep);
3389 	} else {
3390 		kmem_free(gmep, sizeof (*gmep));
3391 	}
3392 
3393 	dsl_pool_rele(dp, FTAG);
3394 	return (err);
3395 }
3396 
3397 static int dmu_recv_end_modified_blocks = 3;
3398 
3399 static int
3400 dmu_recv_existing_end(dmu_recv_cookie_t *drc)
3401 {
3402 #ifdef _KERNEL
3403 	/*
3404 	 * We will be destroying the ds; make sure its origin is unmounted if
3405 	 * necessary.
3406 	 */
3407 	char name[ZFS_MAX_DATASET_NAME_LEN];
3408 	dsl_dataset_name(drc->drc_ds, name);
3409 	zfs_destroy_unmount_origin(name);
3410 #endif
3411 
3412 	return (dsl_sync_task(drc->drc_tofs,
3413 	    dmu_recv_end_check, dmu_recv_end_sync, drc,
3414 	    dmu_recv_end_modified_blocks, ZFS_SPACE_CHECK_NORMAL));
3415 }
3416 
3417 static int
3418 dmu_recv_new_end(dmu_recv_cookie_t *drc)
3419 {
3420 	return (dsl_sync_task(drc->drc_tofs,
3421 	    dmu_recv_end_check, dmu_recv_end_sync, drc,
3422 	    dmu_recv_end_modified_blocks, ZFS_SPACE_CHECK_NORMAL));
3423 }
3424 
3425 int
3426 dmu_recv_end(dmu_recv_cookie_t *drc, void *owner)
3427 {
3428 	int error;
3429 
3430 	drc->drc_owner = owner;
3431 
3432 	if (drc->drc_newfs)
3433 		error = dmu_recv_new_end(drc);
3434 	else
3435 		error = dmu_recv_existing_end(drc);
3436 
3437 	if (error != 0) {
3438 		dmu_recv_cleanup_ds(drc);
3439 	} else if (drc->drc_guid_to_ds_map != NULL) {
3440 		(void) add_ds_to_guidmap(drc->drc_tofs,
3441 		    drc->drc_guid_to_ds_map,
3442 		    drc->drc_newsnapobj);
3443 	}
3444 	return (error);
3445 }
3446 
3447 /*
3448  * Return TRUE if this objset is currently being received into.
3449  */
3450 boolean_t
3451 dmu_objset_is_receiving(objset_t *os)
3452 {
3453 	return (os->os_dsl_dataset != NULL &&
3454 	    os->os_dsl_dataset->ds_owner == dmu_recv_tag);
3455 }
3456