--- old/usr/src/uts/common/fs/zfs/dmu_send.c	Mon Aug  5 17:26:49 2019
+++ new/usr/src/uts/common/fs/zfs/dmu_send.c	Mon Aug  5 17:26:49 2019
@@ -20,10 +20,10 @@
  */
 /*
  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
- * Copyright 2011 Nexenta Systems, Inc. All rights reserved.
  * Copyright (c) 2011, 2015 by Delphix. All rights reserved.
  * Copyright (c) 2014, Joyent, Inc. All rights reserved.
  * Copyright 2014 HybridCluster. All rights reserved.
+ * Copyright 2017 Nexenta Systems, Inc. All rights reserved.
  * Copyright 2016 RackTop Systems.
  * Copyright (c) 2014 Integros [integros.com]
  */
@@ -54,8 +54,11 @@
 #include <sys/blkptr.h>
 #include <sys/dsl_bookmark.h>
 #include <sys/zfeature.h>
+#include <sys/autosnap.h>
 #include <sys/bqueue.h>
 
+#include "zfs_errno.h"
+
 /* Set this tunable to TRUE to replace corrupt data with 0x2f5baddb10c */
 int zfs_send_corrupt_data = B_FALSE;
 int zfs_send_queue_length = 16 * 1024 * 1024;
@@ -110,11 +113,22 @@
 	 */
 
 	ASSERT0(len % 8);
+	ASSERT(buf != NULL);
 
-	dsp->dsa_err = vn_rdwr(UIO_WRITE, dsp->dsa_vp,
-	    (caddr_t)buf, len,
-	    0, UIO_SYSSPACE, FAPPEND, RLIM64_INFINITY, CRED(), &resid);
-
+	dsp->dsa_err = 0;
+	if (!dsp->sendsize) {
+		/* if vp is NULL, then the send is from krrp */
+		if (dsp->dsa_vp != NULL) {
+			dsp->dsa_err = vn_rdwr(UIO_WRITE, dsp->dsa_vp,
+			    (caddr_t)buf, len,
+			    0, UIO_SYSSPACE, FAPPEND, RLIM64_INFINITY,
+			    CRED(), &resid);
+		} else {
+			ASSERT(dsp->dsa_krrp_task != NULL);
+			dsp->dsa_err = dmu_krrp_buffer_write(buf, len,
+			    dsp->dsa_krrp_task);
+		}
+	}
 	mutex_enter(&ds->ds_sendstream_lock);
 	*dsp->dsa_off += len;
 	mutex_exit(&ds->ds_sendstream_lock);
@@ -122,6 +136,17 @@
 	return (dsp->dsa_err);
 }
 
+static int
+dump_bytes_with_checksum(dmu_sendarg_t *dsp, void *buf, int len)
+{
+	if (!dsp->sendsize && (dsp->dsa_krrp_task == NULL ||
+	    dsp->dsa_krrp_task->buffer_args.force_cksum)) {
+		(void) fletcher_4_incremental_native(buf, len, &dsp->dsa_zc);
+	}
+
+	return (dump_bytes(dsp, buf, len));
+}
+
 /*
  * For all record types except BEGIN, fill in the checksum (overlaid in
  * drr_u.drr_checksum.drr_checksum).  The checksum verifies everything
@@ -130,30 +155,41 @@
 static int
 dump_record(dmu_sendarg_t *dsp, void *payload, int payload_len)
 {
+	boolean_t do_checksum = (dsp->dsa_krrp_task == NULL ||
+	    dsp->dsa_krrp_task->buffer_args.force_cksum);
+
 	ASSERT3U(offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
 	    ==, sizeof (dmu_replay_record_t) - sizeof (zio_cksum_t));
-	(void) fletcher_4_incremental_native(dsp->dsa_drr,
-	    offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
-	    &dsp->dsa_zc);
+
 	if (dsp->dsa_drr->drr_type == DRR_BEGIN) {
 		dsp->dsa_sent_begin = B_TRUE;
-	} else {
-		ASSERT(ZIO_CHECKSUM_IS_ZERO(&dsp->dsa_drr->drr_u.
-		    drr_checksum.drr_checksum));
-		dsp->dsa_drr->drr_u.drr_checksum.drr_checksum = dsp->dsa_zc;
 	}
+
 	if (dsp->dsa_drr->drr_type == DRR_END) {
 		dsp->dsa_sent_end = B_TRUE;
 	}
-	(void) fletcher_4_incremental_native(&dsp->dsa_drr->
-	    drr_u.drr_checksum.drr_checksum,
-	    sizeof (zio_cksum_t), &dsp->dsa_zc);
+
+	if (!dsp->sendsize && do_checksum) {
+		(void) fletcher_4_incremental_native(dsp->dsa_drr,
+		    offsetof(dmu_replay_record_t,
+		    drr_u.drr_checksum.drr_checksum),
+		    &dsp->dsa_zc);
+		if (dsp->dsa_drr->drr_type != DRR_BEGIN) {
+			ASSERT(ZIO_CHECKSUM_IS_ZERO(&dsp->dsa_drr->drr_u.
+			    drr_checksum.drr_checksum));
+			dsp->dsa_drr->drr_u.drr_checksum.drr_checksum =
+			    dsp->dsa_zc;
+		}
+
+		(void) fletcher_4_incremental_native(&dsp->dsa_drr->
+		    drr_u.drr_checksum.drr_checksum,
+		    sizeof (zio_cksum_t), &dsp->dsa_zc);
+	}
+
 	if (dump_bytes(dsp, dsp->dsa_drr, sizeof (dmu_replay_record_t)) != 0)
 		return (SET_ERROR(EINTR));
 	if (payload_len != 0) {
-		(void) fletcher_4_incremental_native(payload, payload_len,
-		    &dsp->dsa_zc);
-		if (dump_bytes(dsp, payload, payload_len) != 0)
+		if (dump_bytes_with_checksum(dsp, payload, payload_len) != 0)
 			return (SET_ERROR(EINTR));
 	}
 	return (0);
@@ -361,9 +397,14 @@
 }
 
 static int
-dump_spill(dmu_sendarg_t *dsp, uint64_t object, int blksz, void *data)
+dump_spill(dmu_sendarg_t *dsp, uint64_t object,
+    const blkptr_t *bp, const zbookmark_phys_t *zb)
 {
+	int rc = 0;
 	struct drr_spill *drrs = &(dsp->dsa_drr->drr_u.drr_spill);
+	enum arc_flags aflags = ARC_FLAG_WAIT;
+	int blksz = BP_GET_LSIZE(bp);
+	arc_buf_t *abuf;
 
 	if (dsp->dsa_pending_op != PENDING_NONE) {
 		if (dump_record(dsp, NULL, 0) != 0)
@@ -378,8 +419,38 @@
 	drrs->drr_length = blksz;
 	drrs->drr_toguid = dsp->dsa_toguid;
 
-	if (dump_record(dsp, data, blksz) != 0)
+	if (dump_record(dsp, NULL, 0))
 		return (SET_ERROR(EINTR));
+
+	/*
+	 * if dsa_krrp task is not NULL, then the send is from krrp and we can
+	 * try to bypass copying data to an intermediate buffer.
+	 */
+	if (!dsp->sendsize && dsp->dsa_krrp_task != NULL) {
+		rc = dmu_krrp_direct_arc_read(dsp->dsa_os->os_spa,
+		    dsp->dsa_krrp_task, &dsp->dsa_zc, bp);
+		/*
+		 * rc == 0 means that we successfully copy
+		 * the data directly from ARC to krrp buffer
+		 * rc != 0 && rc != EINTR means that we cannot
+		 * zerocopy the data and need to use slow-path
+		 */
+		if (rc == 0 || rc == EINTR)
+			return (rc);
+
+		ASSERT3U(rc, ==, ENODATA);
+	}
+
+	if (arc_read(NULL, dsp->dsa_os->os_spa, bp, arc_getbuf_func, &abuf,
+	    ZIO_PRIORITY_ASYNC_READ, ZIO_FLAG_CANFAIL,
+	    &aflags, zb) != 0)
+		return (SET_ERROR(EIO));
+
+	rc = dump_bytes_with_checksum(dsp, abuf->b_data, blksz);
+	arc_buf_destroy(abuf, &abuf);
+	if (rc != 0)
+		return (SET_ERROR(EINTR));
+
 	return (0);
 }
 
@@ -636,17 +707,12 @@
 		}
 		arc_buf_destroy(abuf, &abuf);
 	} else if (type == DMU_OT_SA) {
-		arc_flags_t aflags = ARC_FLAG_WAIT;
-		arc_buf_t *abuf;
-		int blksz = BP_GET_LSIZE(bp);
-
-		if (arc_read(NULL, spa, bp, arc_getbuf_func, &abuf,
-		    ZIO_PRIORITY_ASYNC_READ, ZIO_FLAG_CANFAIL,
-		    &aflags, zb) != 0)
-			return (SET_ERROR(EIO));
-
-		err = dump_spill(dsa, zb->zb_object, blksz, abuf->b_data);
-		arc_buf_destroy(abuf, &abuf);
+		/*
+		 * The upstream code has arc_read() call here, but we moved
+		 * it to dump_spill() since we want to take advantage of
+		 * zero copy of the buffer if possible
+		 */
+		err = dump_spill(dsa, zb->zb_object, bp, zb);
 	} else if (backup_do_embed(dsa, bp)) {
 		/* it's an embedded level-0 block of a regular object */
 		int blksz = dblkszsec << SPA_MINBLOCKSHIFT;
@@ -687,11 +753,6 @@
 		    (zb->zb_object == dsa->dsa_resume_object &&
 		    zb->zb_blkid * blksz >= dsa->dsa_resume_offset));
 
-		ASSERT0(zb->zb_level);
-		ASSERT(zb->zb_object > dsa->dsa_resume_object ||
-		    (zb->zb_object == dsa->dsa_resume_object &&
-		    zb->zb_blkid * blksz >= dsa->dsa_resume_offset));
-
 		ASSERT3U(blksz, ==, BP_GET_LSIZE(bp));
 
 		enum zio_flag zioflags = ZIO_FLAG_CANFAIL;
@@ -724,7 +785,6 @@
 				err = dump_write(dsa, type, zb->zb_object,
 				    offset, n, n, NULL, buf);
 				offset += n;
-				buf += n;
 				blksz -= n;
 			}
 		} else {
@@ -755,11 +815,11 @@
  * Note: Releases dp using the specified tag.
  */
 static int
-dmu_send_impl(void *tag, dsl_pool_t *dp, dsl_dataset_t *to_ds,
+dmu_send_impl_ss(void *tag, dsl_pool_t *dp, dsl_dataset_t *to_ds,
     zfs_bookmark_phys_t *ancestor_zb, boolean_t is_clone,
     boolean_t embedok, boolean_t large_block_ok, boolean_t compressok,
-    int outfd, uint64_t resumeobj, uint64_t resumeoff,
-    vnode_t *vp, offset_t *off)
+    int outfd, uint64_t resumeobj, uint64_t resumeoff, vnode_t *vp,
+    offset_t *off, boolean_t sendsize, dmu_krrp_task_t *krrp_task)
 {
 	objset_t *os;
 	dmu_replay_record_t *drr;
@@ -850,8 +910,10 @@
 	dsp->dsa_os = os;
 	dsp->dsa_off = off;
 	dsp->dsa_toguid = dsl_dataset_phys(to_ds)->ds_guid;
+	dsp->dsa_krrp_task = krrp_task;
 	dsp->dsa_pending_op = PENDING_NONE;
 	dsp->dsa_featureflags = featureflags;
+	dsp->sendsize = sendsize;
 	dsp->dsa_resume_object = resumeobj;
 	dsp->dsa_resume_offset = resumeoff;
 
@@ -903,7 +965,7 @@
 	while (!to_data->eos_marker && err == 0) {
 		err = do_dump(dsp, to_data);
 		to_data = get_next_record(&to_arg.q, to_data);
-		if (issig(JUSTLOOKING) && issig(FORREAL))
+		if (vp != NULL && issig(JUSTLOOKING) && issig(FORREAL))
 			err = EINTR;
 	}
 
@@ -957,9 +1019,21 @@
 }
 
 int
+dmu_send_impl(void *tag, dsl_pool_t *dp, dsl_dataset_t *to_ds,
+    zfs_bookmark_phys_t *ancestor_zb, boolean_t is_clone, boolean_t embedok,
+    boolean_t large_block_ok, boolean_t compressok, int outfd,
+    uint64_t resumeobj, uint64_t resumeoff, vnode_t *vp, offset_t *off,
+    dmu_krrp_task_t *krrp_task)
+{
+	return (dmu_send_impl_ss(tag, dp, to_ds, ancestor_zb, is_clone,
+	    embedok, large_block_ok, compressok, outfd, resumeobj, resumeoff,
+	    vp, off, B_FALSE, krrp_task));
+}
+
+int
 dmu_send_obj(const char *pool, uint64_t tosnap, uint64_t fromsnap,
     boolean_t embedok, boolean_t large_block_ok, boolean_t compressok,
-    int outfd, vnode_t *vp, offset_t *off)
+    int outfd, vnode_t *vp, offset_t *off, boolean_t sendsize)
 {
 	dsl_pool_t *dp;
 	dsl_dataset_t *ds;
@@ -994,11 +1068,13 @@
 		zb.zbm_guid = dsl_dataset_phys(fromds)->ds_guid;
 		is_clone = (fromds->ds_dir != ds->ds_dir);
 		dsl_dataset_rele(fromds, FTAG);
-		err = dmu_send_impl(FTAG, dp, ds, &zb, is_clone,
-		    embedok, large_block_ok, compressok, outfd, 0, 0, vp, off);
+		err = dmu_send_impl_ss(FTAG, dp, ds, &zb, is_clone,
+		    embedok, large_block_ok, compressok, outfd, 0, 0, vp, off,
+			sendsize, NULL);
 	} else {
-		err = dmu_send_impl(FTAG, dp, ds, NULL, B_FALSE,
-		    embedok, large_block_ok, compressok, outfd, 0, 0, vp, off);
+		err = dmu_send_impl_ss(FTAG, dp, ds, NULL, B_FALSE,
+		    embedok, large_block_ok, compressok, outfd, 0, 0, vp, off,
+			sendsize, NULL);
 	}
 	dsl_dataset_rele(ds, FTAG);
 	return (err);
@@ -1075,12 +1151,12 @@
 			return (err);
 		}
 		err = dmu_send_impl(FTAG, dp, ds, &zb, is_clone,
-		    embedok, large_block_ok, compressok,
-		    outfd, resumeobj, resumeoff, vp, off);
+		    embedok, large_block_ok, compressok, outfd,
+		    resumeobj, resumeoff, vp, off, NULL);
 	} else {
 		err = dmu_send_impl(FTAG, dp, ds, NULL, B_FALSE,
-		    embedok, large_block_ok, compressok,
-		    outfd, resumeobj, resumeoff, vp, off);
+		    embedok, large_block_ok, compressok, outfd,
+		    resumeobj, resumeoff, vp, off, NULL);
 	}
 	if (owned)
 		dsl_dataset_disown(ds, FTAG);
@@ -1257,25 +1333,50 @@
 
 static int
 recv_begin_check_existing_impl(dmu_recv_begin_arg_t *drba, dsl_dataset_t *ds,
-    uint64_t fromguid)
+    uint64_t fromguid, dmu_tx_t *tx)
 {
 	uint64_t val;
 	int error;
 	dsl_pool_t *dp = ds->ds_dir->dd_pool;
 
-	/* temporary clone name must not exist */
-	error = zap_lookup(dp->dp_meta_objset,
-	    dsl_dir_phys(ds->ds_dir)->dd_child_dir_zapobj, recv_clone_name,
-	    8, 1, &val);
-	if (error != ENOENT)
-		return (error == 0 ? EBUSY : error);
+	if (dmu_tx_is_syncing(tx)) {
+		/* temporary clone name must not exist */
+		error = zap_lookup(dp->dp_meta_objset,
+		    dsl_dir_phys(ds->ds_dir)->dd_child_dir_zapobj,
+		    recv_clone_name, 8, 1, &val);
+		if (error == 0) {
+			dsl_dataset_t *tds;
 
+			/* check that if it is currently used */
+			error = dsl_dataset_own_obj(dp, val, FTAG, &tds);
+			if (!error) {
+				char name[ZFS_MAX_DATASET_NAME_LEN];
+
+				dsl_dataset_name(tds, name);
+				dsl_dataset_disown(tds, FTAG);
+
+				error = dsl_dataset_hold(dp, name, FTAG, &tds);
+				if (!error) {
+					dsl_destroy_head_sync_impl(tds, tx);
+					dsl_dataset_rele(tds, FTAG);
+					error = ENOENT;
+				}
+			} else {
+				error = 0;
+			}
+		}
+		if (error != ENOENT) {
+			return (error == 0 ?
+			    SET_ERROR(EBUSY) : SET_ERROR(error));
+		}
+	}
+
 	/* new snapshot name must not exist */
 	error = zap_lookup(dp->dp_meta_objset,
 	    dsl_dataset_phys(ds)->ds_snapnames_zapobj,
 	    drba->drba_cookie->drc_tosnap, 8, 1, &val);
 	if (error != ENOENT)
-		return (error == 0 ? EEXIST : error);
+		return (error == 0 ? SET_ERROR(EEXIST) : SET_ERROR(error));
 
 	/*
 	 * Check snapshot limit before receiving. We'll recheck again at the
@@ -1399,6 +1500,22 @@
 	if (error == 0) {
 		/* target fs already exists; recv into temp clone */
 
+		if (spa_feature_is_active(dp->dp_spa, SPA_FEATURE_WBC)) {
+			objset_t *os = NULL;
+
+			error = dmu_objset_from_ds(ds, &os);
+			if (error) {
+				dsl_dataset_rele(ds, FTAG);
+				return (error);
+			}
+
+			/* Recv is impossible into DS that uses WBC */
+			if (os->os_wbc_mode != ZFS_WBC_MODE_OFF) {
+				dsl_dataset_rele(ds, FTAG);
+				return (SET_ERROR(EKZFS_WBCNOTSUP));
+			}
+		}
+
 		/* Can't recv a clone into an existing fs */
 		if (flags & DRR_FLAG_CLONE || drba->drba_origin) {
 			dsl_dataset_rele(ds, FTAG);
@@ -1405,7 +1522,7 @@
 			return (SET_ERROR(EINVAL));
 		}
 
-		error = recv_begin_check_existing_impl(drba, ds, fromguid);
+		error = recv_begin_check_existing_impl(drba, ds, fromguid, tx);
 		dsl_dataset_rele(ds, FTAG);
 	} else if (error == ENOENT) {
 		/* target fs does not exist; must be a full backup or clone */
@@ -1435,6 +1552,22 @@
 		if (error != 0)
 			return (error);
 
+		if (spa_feature_is_active(dp->dp_spa, SPA_FEATURE_WBC)) {
+			objset_t *os = NULL;
+
+			error = dmu_objset_from_ds(ds, &os);
+			if (error) {
+				dsl_dataset_rele(ds, FTAG);
+				return (error);
+			}
+
+			/* Recv is impossible into DS that uses WBC */
+			if (os->os_wbc_mode != ZFS_WBC_MODE_OFF) {
+				dsl_dataset_rele(ds, FTAG);
+				return (SET_ERROR(EKZFS_WBCNOTSUP));
+			}
+		}
+
 		/*
 		 * Check filesystem and snapshot limits before receiving. We'll
 		 * recheck snapshot limits again at the end (we create the
@@ -1649,7 +1782,7 @@
 		dsl_dataset_rele(ds, FTAG);
 		return (SET_ERROR(EINVAL));
 	}
-	uint64_t val;
+	uint64_t val = 0;
 	error = zap_lookup(dp->dp_meta_objset, ds->ds_object,
 	    DS_FIELD_RESUME_TOGUID, sizeof (val), 1, &val);
 	if (error != 0 || drrb->drr_toguid != val) {
@@ -1738,7 +1871,8 @@
  */
 int
 dmu_recv_begin(char *tofs, char *tosnap, dmu_replay_record_t *drr_begin,
-    boolean_t force, boolean_t resumable, char *origin, dmu_recv_cookie_t *drc)
+    boolean_t force, boolean_t resumable, boolean_t force_cksum,
+    char *origin, dmu_recv_cookie_t *drc)
 {
 	dmu_recv_begin_arg_t drba = { 0 };
 
@@ -1753,12 +1887,19 @@
 
 	if (drc->drc_drrb->drr_magic == BSWAP_64(DMU_BACKUP_MAGIC)) {
 		drc->drc_byteswap = B_TRUE;
-		(void) fletcher_4_incremental_byteswap(drr_begin,
-		    sizeof (dmu_replay_record_t), &drc->drc_cksum);
-		byteswap_record(drr_begin);
+
+		/* on-wire checksum can be disabled for krrp */
+		if (force_cksum) {
+			(void) fletcher_4_incremental_byteswap(drr_begin,
+			    sizeof (dmu_replay_record_t), &drc->drc_cksum);
+			byteswap_record(drr_begin);
+		}
 	} else if (drc->drc_drrb->drr_magic == DMU_BACKUP_MAGIC) {
-		(void) fletcher_4_incremental_native(drr_begin,
-		    sizeof (dmu_replay_record_t), &drc->drc_cksum);
+		/* on-wire checksum can be disabled for krrp */
+		if (force_cksum) {
+			(void) fletcher_4_incremental_native(drr_begin,
+			    sizeof (dmu_replay_record_t), &drc->drc_cksum);
+		}
 	} else {
 		return (SET_ERROR(EINVAL));
 	}
@@ -1842,6 +1983,7 @@
 	struct receive_record_arg *next_rrd;
 	zio_cksum_t cksum;
 	zio_cksum_t prev_cksum;
+	dmu_krrp_task_t *krrp_task;
 	int err;
 	boolean_t byteswap;
 	/* Sorted list of objects not to issue prefetches for. */
@@ -1894,25 +2036,37 @@
 	 */
 	ASSERT0(len % 8);
 
-	while (done < len) {
-		ssize_t resid;
+	/*
+	 * if vp is NULL, then the send is from krrp and we can try to bypass
+	 * copying data to an intermediate buffer.
+	 */
+	if (ra->vp != NULL) {
+		while (done < len) {
+			ssize_t resid = 0;
 
-		ra->err = vn_rdwr(UIO_READ, ra->vp,
-		    (char *)buf + done, len - done,
-		    ra->voff, UIO_SYSSPACE, FAPPEND,
-		    RLIM64_INFINITY, CRED(), &resid);
-
-		if (resid == len - done) {
-			/*
-			 * Note: ECKSUM indicates that the receive
-			 * was interrupted and can potentially be resumed.
-			 */
-			ra->err = SET_ERROR(ECKSUM);
+			ra->err = vn_rdwr(UIO_READ, ra->vp,
+			    (char *)buf + done, len - done,
+			    ra->voff, UIO_SYSSPACE, FAPPEND,
+			    RLIM64_INFINITY, CRED(), &resid);
+			if (resid == len - done) {
+				/*
+				 * Note: ECKSUM indicates that the receive was
+				 * interrupted and can potentially be resumed.
+				 */
+				ra->err = SET_ERROR(ECKSUM);
+			}
+			ra->voff += len - done - resid;
+			done = len - resid;
+			if (ra->err != 0)
+				return (ra->err);
 		}
-		ra->voff += len - done - resid;
-		done = len - resid;
+	} else {
+		ASSERT(ra->krrp_task != NULL);
+		ra->err = dmu_krrp_buffer_read(buf, len, ra->krrp_task);
 		if (ra->err != 0)
 			return (ra->err);
+
+		done = len;
 	}
 
 	ra->bytes_read += len;
@@ -2218,6 +2372,7 @@
 		dmu_tx_abort(tx);
 		return (err);
 	}
+
 	if (rwa->byteswap) {
 		dmu_object_byteswap_t byteswap =
 		    DMU_OT_BYTESWAP(drrw->drr_type);
@@ -2447,6 +2602,8 @@
 receive_read_payload_and_next_header(struct receive_arg *ra, int len, void *buf)
 {
 	int err;
+	boolean_t checksum_enable = (ra->krrp_task == NULL ||
+	    ra->krrp_task->buffer_args.force_cksum);
 
 	if (len != 0) {
 		ASSERT3U(len, <=, SPA_MAXBLOCKSIZE);
@@ -2480,33 +2637,37 @@
 		return (SET_ERROR(EINVAL));
 	}
 
-	/*
-	 * Note: checksum is of everything up to but not including the
-	 * checksum itself.
-	 */
-	ASSERT3U(offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
-	    ==, sizeof (dmu_replay_record_t) - sizeof (zio_cksum_t));
-	receive_cksum(ra,
-	    offsetof(dmu_replay_record_t, drr_u.drr_checksum.drr_checksum),
-	    &ra->next_rrd->header);
+	if (checksum_enable) {
+		/*
+		 * Note: checksum is of everything up to but not including the
+		 * checksum itself.
+		 */
+		ASSERT3U(offsetof(dmu_replay_record_t,
+		    drr_u.drr_checksum.drr_checksum),
+		    ==, sizeof (dmu_replay_record_t) - sizeof (zio_cksum_t));
+		receive_cksum(ra,
+		    offsetof(dmu_replay_record_t,
+		    drr_u.drr_checksum.drr_checksum),
+		    &ra->next_rrd->header);
 
-	zio_cksum_t cksum_orig =
-	    ra->next_rrd->header.drr_u.drr_checksum.drr_checksum;
-	zio_cksum_t *cksump =
-	    &ra->next_rrd->header.drr_u.drr_checksum.drr_checksum;
+		zio_cksum_t cksum_orig =
+		    ra->next_rrd->header.drr_u.drr_checksum.drr_checksum;
+		zio_cksum_t *cksump =
+		    &ra->next_rrd->header.drr_u.drr_checksum.drr_checksum;
 
-	if (ra->byteswap)
-		byteswap_record(&ra->next_rrd->header);
+		if (ra->byteswap)
+			byteswap_record(&ra->next_rrd->header);
 
-	if ((!ZIO_CHECKSUM_IS_ZERO(cksump)) &&
-	    !ZIO_CHECKSUM_EQUAL(ra->cksum, *cksump)) {
-		kmem_free(ra->next_rrd, sizeof (*ra->next_rrd));
-		ra->next_rrd = NULL;
-		return (SET_ERROR(ECKSUM));
+		if ((!ZIO_CHECKSUM_IS_ZERO(cksump)) &&
+		    !ZIO_CHECKSUM_EQUAL(ra->cksum, *cksump)) {
+			kmem_free(ra->next_rrd, sizeof (*ra->next_rrd));
+			ra->next_rrd = NULL;
+			return (SET_ERROR(ECKSUM));
+		}
+
+		receive_cksum(ra, sizeof (cksum_orig), &cksum_orig);
 	}
 
-	receive_cksum(ra, sizeof (cksum_orig), &cksum_orig);
-
 	return (0);
 }
 
@@ -2702,9 +2863,13 @@
 	}
 	case DRR_END:
 	{
-		struct drr_end *drre = &ra->rrd->header.drr_u.drr_end;
-		if (!ZIO_CHECKSUM_EQUAL(ra->prev_cksum, drre->drr_checksum))
-			return (SET_ERROR(ECKSUM));
+		if (ra->krrp_task == NULL ||
+		    ra->krrp_task->buffer_args.force_cksum) {
+			struct drr_end *drre = &ra->rrd->header.drr_u.drr_end;
+			if (!ZIO_CHECKSUM_EQUAL(ra->prev_cksum,
+			    drre->drr_checksum))
+				return (SET_ERROR(ECKSUM));
+		}
 		return (0);
 	}
 	case DRR_SPILL:
@@ -2870,7 +3035,7 @@
  */
 int
 dmu_recv_stream(dmu_recv_cookie_t *drc, vnode_t *vp, offset_t *voffp,
-    int cleanup_fd, uint64_t *action_handlep)
+    int cleanup_fd, uint64_t *action_handlep, dmu_krrp_task_t *krrp_task)
 {
 	int err = 0;
 	struct receive_arg ra = { 0 };
@@ -2882,6 +3047,7 @@
 	ra.cksum = drc->drc_cksum;
 	ra.vp = vp;
 	ra.voff = *voffp;
+	ra.krrp_task = krrp_task;
 
 	if (dsl_dataset_is_zapified(drc->drc_ds)) {
 		(void) zap_lookup(drc->drc_ds->ds_dir->dd_pool->dp_meta_objset,
@@ -2990,7 +3156,7 @@
 	 * we free ra.rrd and exit.
 	 */
 	while (rwa.err == 0) {
-		if (issig(JUSTLOOKING) && issig(FORREAL)) {
+		if (vp && issig(JUSTLOOKING) && issig(FORREAL)) {
 			err = SET_ERROR(EINTR);
 			break;
 		}
@@ -3056,6 +3222,18 @@
 
 	ASSERT3P(drc->drc_ds->ds_owner, ==, dmu_recv_tag);
 
+	if (spa_feature_is_active(dp->dp_spa, SPA_FEATURE_WBC)) {
+		objset_t *os = NULL;
+
+		error  = dmu_objset_from_ds(drc->drc_ds, &os);
+		if (error)
+			return (error);
+
+		/* Recv is impossible into DS that uses WBC */
+		if (os->os_wbc_mode != ZFS_WBC_MODE_OFF)
+			return (SET_ERROR(EKZFS_WBCNOTSUP));
+	}
+
 	if (!drc->drc_newfs) {
 		dsl_dataset_t *origin_head;
 
@@ -3112,6 +3290,19 @@
 		error = dsl_dataset_snapshot_check_impl(drc->drc_ds,
 		    drc->drc_tosnap, tx, B_TRUE, 1, drc->drc_cred);
 	}
+
+	if (dmu_tx_is_syncing(tx) && drc->drc_krrp_task != NULL) {
+		const char *token =
+		    drc->drc_krrp_task->buffer_args.to_ds;
+		const char *cookie = drc->drc_krrp_task->cookie;
+		dsl_pool_t *dp = tx->tx_pool;
+
+		if (*token != '\0') {
+			error = zap_update(dp->dp_meta_objset,
+			    DMU_POOL_DIRECTORY_OBJECT, token, 1,
+			    strlen(cookie) + 1, cookie, tx);
+		}
+	}
 	return (error);
 }
 
