--- old/usr/src/uts/common/fs/zfs/zio.c	Mon Aug  5 17:27:13 2019
+++ new/usr/src/uts/common/fs/zfs/zio.c	Mon Aug  5 17:27:13 2019
@@ -18,11 +18,12 @@
  *
  * CDDL HEADER END
  */
+
 /*
  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
  * Copyright (c) 2011, 2017 by Delphix. All rights reserved.
- * Copyright (c) 2011 Nexenta Systems, Inc. All rights reserved.
  * Copyright (c) 2014 Integros [integros.com]
+ * Copyright 2017 Nexenta Systems, Inc. All rights reserved.
  */
 
 #include <sys/sysmacros.h>
@@ -39,10 +40,17 @@
 #include <sys/arc.h>
 #include <sys/ddt.h>
 #include <sys/blkptr.h>
+#include <sys/special.h>
+#include <sys/blkptr.h>
 #include <sys/zfeature.h>
+#include <sys/dkioc_free_util.h>
+#include <sys/dsl_scan.h>
+
 #include <sys/metaslab_impl.h>
 #include <sys/abd.h>
 
+extern int zfs_txg_timeout;
+
 /*
  * ==========================================================================
  * I/O type descriptions
@@ -69,12 +77,10 @@
 extern vmem_t *zio_alloc_arena;
 #endif
 
-#define	ZIO_PIPELINE_CONTINUE		0x100
-#define	ZIO_PIPELINE_STOP		0x101
-
 #define	BP_SPANB(indblkshift, level) \
 	(((uint64_t)1) << ((level) * ((indblkshift) - SPA_BLKPTRSHIFT)))
 #define	COMPARE_META_LEVEL	0x80000000ul
+
 /*
  * The following actions directly effect the spa's sync-to-convergence logic.
  * The values below define the sync pass when we start performing the action.
@@ -105,6 +111,21 @@
 int zio_buf_debug_limit = 0;
 #endif
 
+/*
+ * Fault insertion for stress testing
+ */
+int zio_faulty_vdev_enabled = 0;
+uint64_t zio_faulty_vdev_guid;
+uint64_t zio_faulty_vdev_delay_us = 1000000;	/* 1 second */
+
+/*
+ * Tunable to allow for debugging SCSI UNMAP/SATA TRIM calls. Disabling
+ * it will prevent ZFS from attempting to issue DKIOCFREE ioctls to the
+ * underlying storage.
+ */
+boolean_t zfs_trim = B_TRUE;
+uint64_t zfs_trim_min_ext_sz = 1 << 20;	/* 1 MB */
+
 static void zio_taskq_dispatch(zio_t *, zio_taskq_type_t, boolean_t);
 
 void
@@ -180,6 +201,7 @@
 	}
 
 	zio_inject_init();
+
 }
 
 void
@@ -442,26 +464,21 @@
 }
 
 static boolean_t
-zio_wait_for_children(zio_t *zio, uint8_t childbits, enum zio_wait_type wait)
+zio_wait_for_children(zio_t *zio, enum zio_child child, enum zio_wait_type wait)
 {
+	uint64_t *countp = &zio->io_children[child][wait];
 	boolean_t waiting = B_FALSE;
 
 	mutex_enter(&zio->io_lock);
 	ASSERT(zio->io_stall == NULL);
-	for (int c = 0; c < ZIO_CHILD_TYPES; c++) {
-		if (!(ZIO_CHILD_BIT_IS_SET(childbits, c)))
-			continue;
-
-		uint64_t *countp = &zio->io_children[c][wait];
-		if (*countp != 0) {
-			zio->io_stage >>= 1;
-			ASSERT3U(zio->io_stage, !=, ZIO_STAGE_OPEN);
-			zio->io_stall = countp;
-			waiting = B_TRUE;
-			break;
-		}
+	if (*countp != 0) {
+		zio->io_stage >>= 1;
+		ASSERT3U(zio->io_stage, !=, ZIO_STAGE_OPEN);
+		zio->io_stall = countp;
+		waiting = B_TRUE;
 	}
 	mutex_exit(&zio->io_lock);
+
 	return (waiting);
 }
 
@@ -619,11 +636,16 @@
 		zio->io_bookmark = *zb;
 
 	if (pio != NULL) {
+		zio->io_mc = pio->io_mc;
 		if (zio->io_logical == NULL)
 			zio->io_logical = pio->io_logical;
 		if (zio->io_child_type == ZIO_CHILD_GANG)
 			zio->io_gang_leader = pio->io_gang_leader;
 		zio_add_child(pio, zio);
+
+		/* copy the smartcomp setting when creating child zio's */
+		bcopy(&pio->io_smartcomp, &zio->io_smartcomp,
+		    sizeof (zio->io_smartcomp));
 	}
 
 	return (zio);
@@ -662,6 +684,14 @@
 void
 zfs_blkptr_verify(spa_t *spa, const blkptr_t *bp)
 {
+	/*
+	 * SPECIAL-BP has two DVAs, but DVA[0] in this case is a
+	 * temporary DVA, and after migration only the DVA[1]
+	 * contains valid data. Therefore, we start walking for
+	 * these BPs from DVA[1].
+	 */
+	int start_dva = BP_IS_SPECIAL(bp) ? 1 : 0;
+
 	if (!DMU_OT_IS_VALID(BP_GET_TYPE(bp))) {
 		zfs_panic_recover("blkptr at %p has invalid TYPE %llu",
 		    bp, (longlong_t)BP_GET_TYPE(bp));
@@ -693,13 +723,6 @@
 	}
 
 	/*
-	 * Do not verify individual DVAs if the config is not trusted. This
-	 * will be done once the zio is executed in vdev_mirror_map_alloc.
-	 */
-	if (!spa->spa_trust_config)
-		return;
-
-	/*
 	 * Pool-specific checks.
 	 *
 	 * Note: it would be nice to verify that the blk_birth and
@@ -707,7 +730,7 @@
 	 * allows the birth time of log blocks (and dmu_sync()-ed blocks
 	 * that are in the log) to be arbitrarily large.
 	 */
-	for (int i = 0; i < BP_GET_NDVAS(bp); i++) {
+	for (int i = start_dva; i < BP_GET_NDVAS(bp); i++) {
 		uint64_t vdevid = DVA_GET_VDEV(&bp->blk_dva[i]);
 		if (vdevid >= spa->spa_root_vdev->vdev_children) {
 			zfs_panic_recover("blkptr at %p DVA %u has invalid "
@@ -748,36 +771,6 @@
 	}
 }
 
-boolean_t
-zfs_dva_valid(spa_t *spa, const dva_t *dva, const blkptr_t *bp)
-{
-	uint64_t vdevid = DVA_GET_VDEV(dva);
-
-	if (vdevid >= spa->spa_root_vdev->vdev_children)
-		return (B_FALSE);
-
-	vdev_t *vd = spa->spa_root_vdev->vdev_child[vdevid];
-	if (vd == NULL)
-		return (B_FALSE);
-
-	if (vd->vdev_ops == &vdev_hole_ops)
-		return (B_FALSE);
-
-	if (vd->vdev_ops == &vdev_missing_ops) {
-		return (B_FALSE);
-	}
-
-	uint64_t offset = DVA_GET_OFFSET(dva);
-	uint64_t asize = DVA_GET_ASIZE(dva);
-
-	if (BP_IS_GANG(bp))
-		asize = vdev_psize_to_asize(vd, SPA_GANGBLOCKSIZE);
-	if (offset + asize > vd->vdev_asize)
-		return (B_FALSE);
-
-	return (B_TRUE);
-}
-
 zio_t *
 zio_read(zio_t *pio, spa_t *spa, const blkptr_t *bp,
     abd_t *data, uint64_t size, zio_done_func_t *done, void *private,
@@ -802,7 +795,8 @@
     zio_done_func_t *ready, zio_done_func_t *children_ready,
     zio_done_func_t *physdone, zio_done_func_t *done,
     void *private, zio_priority_t priority, enum zio_flag flags,
-    const zbookmark_phys_t *zb)
+    const zbookmark_phys_t *zb,
+    const zio_smartcomp_info_t *smartcomp)
 {
 	zio_t *zio;
 
@@ -824,6 +818,8 @@
 	zio->io_children_ready = children_ready;
 	zio->io_physdone = physdone;
 	zio->io_prop = *zp;
+	if (smartcomp != NULL)
+		bcopy(smartcomp, &zio->io_smartcomp, sizeof (*smartcomp));
 
 	/*
 	 * Data can be NULL if we are going to call zio_write_override() to
@@ -875,8 +871,6 @@
 zio_free(spa_t *spa, uint64_t txg, const blkptr_t *bp)
 {
 
-	zfs_blkptr_verify(spa, bp);
-
 	/*
 	 * The check for EMBEDDED is a performance optimization.  We
 	 * process the free here (by ignoring it) rather than
@@ -917,6 +911,7 @@
 
 	metaslab_check_free(spa, bp);
 	arc_freed(spa, bp);
+	dsl_scan_freed(spa, bp);
 
 	/*
 	 * GANG and DEDUP blocks can induce a read (for the gang block header,
@@ -939,7 +934,7 @@
 {
 	zio_t *zio;
 
-	zfs_blkptr_verify(spa, bp);
+	dprintf_bp(bp, "claiming in txg %llu", txg);
 
 	if (BP_IS_EMBEDDED(bp))
 		return (zio_null(pio, spa, NULL, NULL, NULL, 0));
@@ -968,9 +963,10 @@
 	return (zio);
 }
 
-zio_t *
-zio_ioctl(zio_t *pio, spa_t *spa, vdev_t *vd, int cmd,
-    zio_done_func_t *done, void *private, enum zio_flag flags)
+static zio_t *
+zio_ioctl_with_pipeline(zio_t *pio, spa_t *spa, vdev_t *vd, int cmd,
+    zio_done_func_t *done, void *private, enum zio_flag flags,
+    enum zio_stage pipeline)
 {
 	zio_t *zio;
 	int c;
@@ -978,15 +974,26 @@
 	if (vd->vdev_children == 0) {
 		zio = zio_create(pio, spa, 0, NULL, NULL, 0, 0, done, private,
 		    ZIO_TYPE_IOCTL, ZIO_PRIORITY_NOW, flags, vd, 0, NULL,
-		    ZIO_STAGE_OPEN, ZIO_IOCTL_PIPELINE);
+		    ZIO_STAGE_OPEN, pipeline);
 
 		zio->io_cmd = cmd;
 	} else {
-		zio = zio_null(pio, spa, NULL, NULL, NULL, flags);
-
-		for (c = 0; c < vd->vdev_children; c++)
-			zio_nowait(zio_ioctl(zio, spa, vd->vdev_child[c], cmd,
-			    done, private, flags));
+		zio = zio_null(pio, spa, vd, done, private, flags);
+		/*
+		 * DKIOCFREE ioctl's need some special handling on interior
+		 * vdevs. If the device provides an ops function to handle
+		 * recomputing dkioc_free extents, then we call it.
+		 * Otherwise the default behavior applies, which simply fans
+		 * out the ioctl to all component vdevs.
+		 */
+		if (cmd == DKIOCFREE && vd->vdev_ops->vdev_op_trim != NULL) {
+			vd->vdev_ops->vdev_op_trim(vd, zio, private);
+		} else {
+			for (c = 0; c < vd->vdev_children; c++)
+				zio_nowait(zio_ioctl_with_pipeline(zio,
+				    spa, vd->vdev_child[c], cmd, NULL,
+				    private, flags, pipeline));
+		}
 	}
 
 	return (zio);
@@ -993,6 +1000,118 @@
 }
 
 zio_t *
+zio_ioctl(zio_t *pio, spa_t *spa, vdev_t *vd, int cmd,
+    zio_done_func_t *done, void *private, enum zio_flag flags)
+{
+	return (zio_ioctl_with_pipeline(pio, spa, vd, cmd, done,
+	    private, flags, ZIO_IOCTL_PIPELINE));
+}
+
+/*
+ * Callback for when a trim zio has completed. This simply frees the
+ * dkioc_free_list_t extent list of the DKIOCFREE ioctl.
+ */
+static void
+zio_trim_done(zio_t *zio)
+{
+	VERIFY(zio->io_private != NULL);
+	dfl_free(zio->io_private);
+}
+
+static void
+zio_trim_check(uint64_t start, uint64_t len, void *msp)
+{
+	metaslab_t *ms = msp;
+	boolean_t held = MUTEX_HELD(&ms->ms_lock);
+	if (!held)
+		mutex_enter(&ms->ms_lock);
+	ASSERT(ms->ms_trimming_ts != NULL);
+	ASSERT(range_tree_contains(ms->ms_trimming_ts->ts_tree,
+	    start - VDEV_LABEL_START_SIZE, len));
+	if (!held)
+		mutex_exit(&ms->ms_lock);
+}
+
+/*
+ * Takes a bunch of freed extents and tells the underlying vdevs that the
+ * space associated with these extents can be released.
+ * This is used by flash storage to pre-erase blocks for rapid reuse later
+ * and thin-provisioned block storage to reclaim unused blocks.
+ */
+zio_t *
+zio_trim(spa_t *spa, vdev_t *vd, struct range_tree *tree,
+    zio_done_func_t *done, void *private, enum zio_flag flags,
+    int trim_flags, metaslab_t *msp)
+{
+	dkioc_free_list_t *dfl = NULL;
+	range_seg_t *rs;
+	uint64_t rs_idx;
+	uint64_t num_exts;
+	uint64_t bytes_issued = 0, bytes_skipped = 0, exts_skipped = 0;
+	/*
+	 * We need this to invoke the caller's `done' callback with the
+	 * correct io_private (not the dkioc_free_list_t, which is needed
+	 * by the underlying DKIOCFREE ioctl).
+	 */
+	zio_t *sub_pio = zio_root(spa, done, private, flags);
+
+	ASSERT(range_tree_space(tree) != 0);
+
+	if (!zfs_trim)
+		return (sub_pio);
+
+	num_exts = avl_numnodes(&tree->rt_root);
+	dfl = kmem_zalloc(DFL_SZ(num_exts), KM_SLEEP);
+	dfl->dfl_flags = trim_flags;
+	dfl->dfl_num_exts = num_exts;
+	dfl->dfl_offset = VDEV_LABEL_START_SIZE;
+	if (msp) {
+		dfl->dfl_ck_func = zio_trim_check;
+		dfl->dfl_ck_arg = msp;
+	}
+
+	for (rs = avl_first(&tree->rt_root), rs_idx = 0; rs != NULL;
+	    rs = AVL_NEXT(&tree->rt_root, rs)) {
+		uint64_t len = rs->rs_end - rs->rs_start;
+
+		if (len < zfs_trim_min_ext_sz) {
+			bytes_skipped += len;
+			exts_skipped++;
+			continue;
+		}
+
+		dfl->dfl_exts[rs_idx].dfle_start = rs->rs_start;
+		dfl->dfl_exts[rs_idx].dfle_length = len;
+
+		// check we're a multiple of the vdev ashift
+		ASSERT0(dfl->dfl_exts[rs_idx].dfle_start &
+		    ((1 << vd->vdev_ashift) - 1));
+		ASSERT0(dfl->dfl_exts[rs_idx].dfle_length &
+		    ((1 << vd->vdev_ashift) - 1));
+
+		rs_idx++;
+		bytes_issued += len;
+	}
+
+	spa_trimstats_update(spa, rs_idx, bytes_issued, exts_skipped,
+	    bytes_skipped);
+
+	/* the zfs_trim_min_ext_sz filter may have shortened the list */
+	if (dfl->dfl_num_exts != rs_idx) {
+		dkioc_free_list_t *dfl2 = kmem_zalloc(DFL_SZ(rs_idx), KM_SLEEP);
+		bcopy(dfl, dfl2, DFL_SZ(rs_idx));
+		dfl2->dfl_num_exts = rs_idx;
+		dfl_free(dfl);
+		dfl = dfl2;
+	}
+
+	zio_nowait(zio_ioctl_with_pipeline(sub_pio, spa, vd, DKIOCFREE,
+	    zio_trim_done, dfl, ZIO_FLAG_CANFAIL | ZIO_FLAG_DONT_PROPAGATE |
+	    ZIO_FLAG_DONT_RETRY, ZIO_TRIM_PIPELINE));
+	return (sub_pio);
+}
+
+zio_t *
 zio_read_phys(zio_t *pio, vdev_t *vd, uint64_t offset, uint64_t size,
     abd_t *data, int checksum, zio_done_func_t *done, void *private,
     zio_priority_t priority, enum zio_flag flags, boolean_t labels)
@@ -1058,27 +1177,9 @@
 	enum zio_stage pipeline = ZIO_VDEV_CHILD_PIPELINE;
 	zio_t *zio;
 
-	/*
-	 * vdev child I/Os do not propagate their error to the parent.
-	 * Therefore, for correct operation the caller *must* check for
-	 * and handle the error in the child i/o's done callback.
-	 * The only exceptions are i/os that we don't care about
-	 * (OPTIONAL or REPAIR).
-	 */
-	ASSERT((flags & ZIO_FLAG_OPTIONAL) || (flags & ZIO_FLAG_IO_REPAIR) ||
-	    done != NULL);
+	ASSERT(vd->vdev_parent ==
+	    (pio->io_vd ? pio->io_vd : pio->io_spa->spa_root_vdev));
 
-	/*
-	 * In the common case, where the parent zio was to a normal vdev,
-	 * the child zio must be to a child vdev of that vdev.  Otherwise,
-	 * the child zio must be to a top-level vdev.
-	 */
-	if (pio->io_vd != NULL && pio->io_vd->vdev_ops != &vdev_indirect_ops) {
-		ASSERT3P(vd->vdev_parent, ==, pio->io_vd);
-	} else {
-		ASSERT3P(vd, ==, vd->vdev_top);
-	}
-
 	if (type == ZIO_TYPE_READ && bp != NULL) {
 		/*
 		 * If we have the bp, then the child should perform the
@@ -1090,12 +1191,10 @@
 		pio->io_pipeline &= ~ZIO_STAGE_CHECKSUM_VERIFY;
 	}
 
-	if (vd->vdev_ops->vdev_op_leaf) {
-		ASSERT0(vd->vdev_children);
+	if (vd->vdev_children == 0)
 		offset += VDEV_LABEL_START_SIZE;
-	}
 
-	flags |= ZIO_VDEV_CHILD_FLAGS(pio);
+	flags |= ZIO_VDEV_CHILD_FLAGS(pio) | ZIO_FLAG_DONT_PROPAGATE;
 
 	/*
 	 * If we've decided to do a repair, the write is not speculative --
@@ -1112,7 +1211,7 @@
 	 */
 	if (flags & ZIO_FLAG_IO_ALLOCATING &&
 	    (vd != vd->vdev_top || (flags & ZIO_FLAG_IO_RETRY))) {
-		metaslab_class_t *mc = spa_normal_class(pio->io_spa);
+		metaslab_class_t *mc = pio->io_mc;
 
 		ASSERT(mc->mc_alloc_throttle_enabled);
 		ASSERT(type == ZIO_TYPE_WRITE);
@@ -1193,8 +1292,6 @@
 {
 	blkptr_t *bp = zio->io_bp;
 
-	ASSERT3P(zio->io_bp, ==, &zio->io_bp_copy);
-
 	if (BP_GET_COMPRESS(bp) != ZIO_COMPRESS_OFF &&
 	    zio->io_child_type == ZIO_CHILD_LOGICAL &&
 	    !(zio->io_flags & ZIO_FLAG_RAW)) {
@@ -1213,10 +1310,9 @@
 		abd_return_buf_copy(zio->io_abd, data, psize);
 	} else {
 		ASSERT(!BP_IS_EMBEDDED(bp));
-		ASSERT3P(zio->io_bp, ==, &zio->io_bp_copy);
 	}
 
-	if (!DMU_OT_IS_METADATA(BP_GET_TYPE(bp)) && BP_GET_LEVEL(bp) == 0)
+	if (!BP_IS_METADATA(bp))
 		zio->io_flags |= ZIO_FLAG_DONT_CACHE;
 
 	if (BP_GET_TYPE(bp) == DMU_OT_DDT_ZAP)
@@ -1304,10 +1400,9 @@
 	 * If our children haven't all reached the ready stage,
 	 * wait for them and then repeat this pipeline stage.
 	 */
-	if (zio_wait_for_children(zio, ZIO_CHILD_LOGICAL_BIT |
-	    ZIO_CHILD_GANG_BIT, ZIO_WAIT_READY)) {
+	if (zio_wait_for_children(zio, ZIO_CHILD_GANG, ZIO_WAIT_READY) ||
+	    zio_wait_for_children(zio, ZIO_CHILD_LOGICAL, ZIO_WAIT_READY))
 		return (ZIO_PIPELINE_STOP);
-	}
 
 	if (!IO_IS_ALLOCATING(zio))
 		return (ZIO_PIPELINE_CONTINUE);
@@ -1349,8 +1444,10 @@
 		    spa_max_replication(spa)) == BP_GET_NDVAS(bp));
 	}
 
+	DTRACE_PROBE1(zio_compress_ready, zio_t *, zio);
 	/* If it's a compressed write that is not raw, compress the buffer. */
-	if (compress != ZIO_COMPRESS_OFF && psize == lsize) {
+	if (compress != ZIO_COMPRESS_OFF && psize == lsize &&
+	    ZIO_SHOULD_COMPRESS(zio)) {
 		void *cbuf = zio_buf_alloc(lsize);
 		psize = zio_compress_data(compress, zio->io_abd, cbuf, lsize);
 		if (psize == 0 || psize == lsize) {
@@ -1369,6 +1466,12 @@
 			zio->io_pipeline = ZIO_INTERLOCK_PIPELINE;
 			ASSERT(spa_feature_is_active(spa,
 			    SPA_FEATURE_EMBEDDED_DATA));
+			if (zio->io_smartcomp.sc_result != NULL) {
+				zio->io_smartcomp.sc_result(
+				    zio->io_smartcomp.sc_userinfo, zio);
+			} else {
+				ASSERT(zio->io_smartcomp.sc_ask == NULL);
+			}
 			return (ZIO_PIPELINE_CONTINUE);
 		} else {
 			/*
@@ -1396,6 +1499,13 @@
 			}
 		}
 
+		if (zio->io_smartcomp.sc_result != NULL) {
+			zio->io_smartcomp.sc_result(
+			    zio->io_smartcomp.sc_userinfo, zio);
+		} else {
+			ASSERT(zio->io_smartcomp.sc_ask == NULL);
+		}
+
 		/*
 		 * We were unable to handle this as an override bp, treat
 		 * it as a regular write I/O.
@@ -1405,6 +1515,17 @@
 		zio->io_pipeline = zio->io_orig_pipeline;
 	} else {
 		ASSERT3U(psize, !=, 0);
+
+		/*
+		 * We are here because of:
+		 *	- compress == ZIO_COMPRESS_OFF
+		 *	- SmartCompression decides don't compress this data
+		 *	- this is a RAW-write
+		 *
+		 *	In case of RAW-write we should not override "compress"
+		 */
+		if ((zio->io_flags & ZIO_FLAG_RAW) == 0)
+			compress = ZIO_COMPRESS_OFF;
 	}
 
 	/*
@@ -1437,6 +1558,10 @@
 		}
 		zio->io_pipeline = ZIO_INTERLOCK_PIPELINE;
 	} else {
+		if (zp->zp_dedup) {
+			/* check the best-effort dedup setting */
+			zio_best_effort_dedup(zio);
+		}
 		ASSERT(zp->zp_checksum != ZIO_CHECKSUM_GANG_HEADER);
 		BP_SET_LSIZE(bp, lsize);
 		BP_SET_TYPE(bp, zp->zp_type);
@@ -1470,8 +1595,6 @@
 			zio->io_pipeline = ZIO_DDT_FREE_PIPELINE;
 	}
 
-	ASSERT3P(zio->io_bp, ==, &zio->io_bp_copy);
-
 	return (ZIO_PIPELINE_CONTINUE);
 }
 
@@ -1506,7 +1629,8 @@
 	 * If this is a high priority I/O, then use the high priority taskq if
 	 * available.
 	 */
-	if (zio->io_priority == ZIO_PRIORITY_NOW &&
+	if ((zio->io_priority == ZIO_PRIORITY_NOW ||
+	    zio->io_priority == ZIO_PRIORITY_SYNC_WRITE) &&
 	    spa->spa_zio_taskq[t][q + 1].stqs_count != 0)
 		q++;
 
@@ -1633,6 +1757,7 @@
 
 	while (zio->io_stage < ZIO_STAGE_DONE) {
 		enum zio_stage pipeline = zio->io_pipeline;
+		enum zio_stage old_stage = zio->io_stage;
 		enum zio_stage stage = zio->io_stage;
 		int rv;
 
@@ -1670,6 +1795,12 @@
 		if (rv == ZIO_PIPELINE_STOP)
 			return;
 
+		if (rv == ZIO_PIPELINE_RESTART_STAGE) {
+			zio->io_stage = old_stage;
+			(void) zio_issue_async(zio);
+			return;
+		}
+
 		ASSERT(rv == ZIO_PIPELINE_CONTINUE);
 	}
 }
@@ -2150,9 +2281,8 @@
 {
 	blkptr_t *bp = zio->io_bp;
 
-	if (zio_wait_for_children(zio, ZIO_CHILD_GANG_BIT, ZIO_WAIT_DONE)) {
+	if (zio_wait_for_children(zio, ZIO_CHILD_GANG, ZIO_WAIT_DONE))
 		return (ZIO_PIPELINE_STOP);
-	}
 
 	ASSERT(BP_IS_GANG(bp) && zio->io_gang_leader == zio);
 	ASSERT(zio->io_child_type > ZIO_CHILD_GANG);
@@ -2208,7 +2338,7 @@
 zio_write_gang_block(zio_t *pio)
 {
 	spa_t *spa = pio->io_spa;
-	metaslab_class_t *mc = spa_normal_class(spa);
+	metaslab_class_t *mc = pio->io_mc;
 	blkptr_t *bp = pio->io_bp;
 	zio_t *gio = pio->io_gang_leader;
 	zio_t *zio;
@@ -2305,8 +2435,11 @@
 		    abd_get_offset(pio->io_abd, pio->io_size - resid), lsize,
 		    lsize, &zp, zio_write_gang_member_ready, NULL, NULL,
 		    zio_write_gang_done, &gn->gn_child[g], pio->io_priority,
-		    ZIO_GANG_CHILD_FLAGS(pio), &pio->io_bookmark);
+		    ZIO_GANG_CHILD_FLAGS(pio), &pio->io_bookmark,
+		    &pio->io_smartcomp);
 
+		cio->io_mc = mc;
+
 		if (pio->io_flags & ZIO_FLAG_IO_ALLOCATING) {
 			ASSERT(pio->io_priority == ZIO_PRIORITY_ASYNC_WRITE);
 			ASSERT(!(pio->io_flags & ZIO_FLAG_NODATA));
@@ -2473,9 +2606,8 @@
 {
 	blkptr_t *bp = zio->io_bp;
 
-	if (zio_wait_for_children(zio, ZIO_CHILD_DDT_BIT, ZIO_WAIT_DONE)) {
+	if (zio_wait_for_children(zio, ZIO_CHILD_DDT, ZIO_WAIT_DONE))
 		return (ZIO_PIPELINE_STOP);
-	}
 
 	ASSERT(BP_GET_DEDUP(bp));
 	ASSERT(BP_GET_PSIZE(bp) == zio->io_size);
@@ -2507,6 +2639,7 @@
 	return (ZIO_PIPELINE_CONTINUE);
 }
 
+/* ARGSUSED */
 static boolean_t
 zio_ddt_collision(zio_t *zio, ddt_t *ddt, ddt_entry_t *dde)
 {
@@ -2544,7 +2677,7 @@
 
 			ddt_bp_fill(ddp, &blk, ddp->ddp_phys_birth);
 
-			ddt_exit(ddt);
+			dde_exit(dde);
 
 			/*
 			 * Intuitively, it would make more sense to compare
@@ -2575,7 +2708,7 @@
 				arc_buf_destroy(abuf, &abuf);
 			}
 
-			ddt_enter(ddt);
+			dde_enter(dde);
 			return (error != 0);
 		}
 	}
@@ -2587,7 +2720,6 @@
 zio_ddt_child_write_ready(zio_t *zio)
 {
 	int p = zio->io_prop.zp_copies;
-	ddt_t *ddt = ddt_select(zio->io_spa, zio->io_bp);
 	ddt_entry_t *dde = zio->io_private;
 	ddt_phys_t *ddp = &dde->dde_phys[p];
 	zio_t *pio;
@@ -2595,7 +2727,7 @@
 	if (zio->io_error)
 		return;
 
-	ddt_enter(ddt);
+	dde_enter(dde);
 
 	ASSERT(dde->dde_lead_zio[p] == zio);
 
@@ -2605,7 +2737,7 @@
 	while ((pio = zio_walk_parents(zio, &zl)) != NULL)
 		ddt_bp_fill(ddp, pio->io_bp, zio->io_txg);
 
-	ddt_exit(ddt);
+	dde_exit(dde);
 }
 
 static void
@@ -2612,11 +2744,10 @@
 zio_ddt_child_write_done(zio_t *zio)
 {
 	int p = zio->io_prop.zp_copies;
-	ddt_t *ddt = ddt_select(zio->io_spa, zio->io_bp);
 	ddt_entry_t *dde = zio->io_private;
 	ddt_phys_t *ddp = &dde->dde_phys[p];
 
-	ddt_enter(ddt);
+	dde_enter(dde);
 
 	ASSERT(ddp->ddp_refcnt == 0);
 	ASSERT(dde->dde_lead_zio[p] == zio);
@@ -2630,7 +2761,7 @@
 		ddt_phys_clear(ddp);
 	}
 
-	ddt_exit(ddt);
+	dde_exit(dde);
 }
 
 static void
@@ -2644,7 +2775,7 @@
 	ddt_phys_t *ddp = &dde->dde_phys[p];
 	ddt_key_t *ddk = &dde->dde_key;
 
-	ddt_enter(ddt);
+	dde_enter(dde);
 
 	ASSERT(ddp->ddp_refcnt == 0);
 	ASSERT(dde->dde_lead_zio[p] == zio);
@@ -2659,7 +2790,7 @@
 		ddt_phys_fill(ddp, bp);
 	}
 
-	ddt_exit(ddt);
+	dde_exit(dde);
 }
 
 static int
@@ -2682,10 +2813,32 @@
 	ASSERT(BP_IS_HOLE(bp) || zio->io_bp_override);
 	ASSERT(!(zio->io_bp_override && (zio->io_flags & ZIO_FLAG_RAW)));
 
-	ddt_enter(ddt);
 	dde = ddt_lookup(ddt, bp, B_TRUE);
-	ddp = &dde->dde_phys[p];
 
+	/*
+	 * If we're not using special tier, for each new DDE that's not on disk:
+	 * disable dedup if we have exhausted "allowed" DDT L2/ARC space
+	 */
+	if ((dde->dde_state & DDE_NEW) && !spa->spa_usesc &&
+	    (zfs_ddt_limit_type != DDT_NO_LIMIT || zfs_ddt_byte_ceiling != 0)) {
+		/* turn off dedup if we need to stop DDT growth */
+		if (spa_enable_dedup_cap(spa)) {
+			dde->dde_state |= DDE_DONT_SYNC;
+
+			/* disable dedup and use the ordinary write pipeline */
+			zio_pop_transforms(zio);
+			zp->zp_dedup = zp->zp_dedup_verify = B_FALSE;
+			zio->io_stage = ZIO_STAGE_OPEN;
+			zio->io_pipeline = ZIO_WRITE_PIPELINE;
+			zio->io_bp_override = NULL;
+			BP_ZERO(bp);
+			dde_exit(dde);
+
+			return (ZIO_PIPELINE_CONTINUE);
+		}
+	}
+	ASSERT(!(dde->dde_state & DDE_DONT_SYNC));
+
 	if (zp->zp_dedup_verify && zio_ddt_collision(zio, ddt, dde)) {
 		/*
 		 * If we're using a weak checksum, upgrade to a strong checksum
@@ -2705,10 +2858,11 @@
 		}
 		ASSERT(!BP_GET_DEDUP(bp));
 		zio->io_pipeline = ZIO_WRITE_PIPELINE;
-		ddt_exit(ddt);
+		dde_exit(dde);
 		return (ZIO_PIPELINE_CONTINUE);
 	}
 
+	ddp = &dde->dde_phys[p];
 	ditto_copies = ddt_ditto_copies_needed(ddt, dde, ddp);
 	ASSERT(ditto_copies < SPA_DVAS_PER_BP);
 
@@ -2731,7 +2885,7 @@
 			zio->io_pipeline = ZIO_WRITE_PIPELINE;
 			zio->io_bp_override = NULL;
 			BP_ZERO(bp);
-			ddt_exit(ddt);
+			dde_exit(dde);
 			return (ZIO_PIPELINE_CONTINUE);
 		}
 
@@ -2738,7 +2892,7 @@
 		dio = zio_write(zio, spa, txg, bp, zio->io_orig_abd,
 		    zio->io_orig_size, zio->io_orig_size, &czp, NULL, NULL,
 		    NULL, zio_ddt_ditto_write_done, dde, zio->io_priority,
-		    ZIO_DDT_CHILD_FLAGS(zio), &zio->io_bookmark);
+		    ZIO_DDT_CHILD_FLAGS(zio), &zio->io_bookmark, NULL);
 
 		zio_push_transform(dio, zio->io_abd, zio->io_size, 0, NULL);
 		dde->dde_lead_zio[DDT_PHYS_DITTO] = dio;
@@ -2761,13 +2915,13 @@
 		    zio->io_orig_size, zio->io_orig_size, zp,
 		    zio_ddt_child_write_ready, NULL, NULL,
 		    zio_ddt_child_write_done, dde, zio->io_priority,
-		    ZIO_DDT_CHILD_FLAGS(zio), &zio->io_bookmark);
+		    ZIO_DDT_CHILD_FLAGS(zio), &zio->io_bookmark, NULL);
 
 		zio_push_transform(cio, zio->io_abd, zio->io_size, 0, NULL);
 		dde->dde_lead_zio[p] = cio;
 	}
 
-	ddt_exit(ddt);
+	dde_exit(dde);
 
 	if (cio)
 		zio_nowait(cio);
@@ -2791,11 +2945,11 @@
 	ASSERT(BP_GET_DEDUP(bp));
 	ASSERT(zio->io_child_type == ZIO_CHILD_LOGICAL);
 
-	ddt_enter(ddt);
 	freedde = dde = ddt_lookup(ddt, bp, B_TRUE);
 	ddp = ddt_phys_select(dde, bp);
-	ddt_phys_decref(ddp);
-	ddt_exit(ddt);
+	if (ddp)
+		ddt_phys_decref(ddp);
+	dde_exit(dde);
 
 	return (ZIO_PIPELINE_CONTINUE);
 }
@@ -2807,13 +2961,13 @@
  */
 
 static zio_t *
-zio_io_to_allocate(spa_t *spa)
+zio_io_to_allocate(metaslab_class_t *mc)
 {
 	zio_t *zio;
 
-	ASSERT(MUTEX_HELD(&spa->spa_alloc_lock));
+	ASSERT(MUTEX_HELD(&mc->mc_alloc_lock));
 
-	zio = avl_first(&spa->spa_alloc_tree);
+	zio = avl_first(&mc->mc_alloc_tree);
 	if (zio == NULL)
 		return (NULL);
 
@@ -2823,12 +2977,12 @@
 	 * Try to place a reservation for this zio. If we're unable to
 	 * reserve then we throttle.
 	 */
-	if (!metaslab_class_throttle_reserve(spa_normal_class(spa),
+	if (!metaslab_class_throttle_reserve(mc,
 	    zio->io_prop.zp_copies, zio, 0)) {
 		return (NULL);
 	}
 
-	avl_remove(&spa->spa_alloc_tree, zio);
+	avl_remove(&mc->mc_alloc_tree, zio);
 	ASSERT3U(zio->io_stage, <, ZIO_STAGE_DVA_ALLOCATE);
 
 	return (zio);
@@ -2840,8 +2994,15 @@
 	spa_t *spa = zio->io_spa;
 	zio_t *nio;
 
+	/* We need to use parent's MetaslabClass */
+	if (zio->io_mc == NULL) {
+		zio->io_mc = spa_select_class(spa, zio);
+		if (zio->io_prop.zp_usewbc)
+			return (ZIO_PIPELINE_CONTINUE);
+	}
+
 	if (zio->io_priority == ZIO_PRIORITY_SYNC_WRITE ||
-	    !spa_normal_class(zio->io_spa)->mc_alloc_throttle_enabled ||
+	    !zio->io_mc->mc_alloc_throttle_enabled ||
 	    zio->io_child_type == ZIO_CHILD_GANG ||
 	    zio->io_flags & ZIO_FLAG_NODATA) {
 		return (ZIO_PIPELINE_CONTINUE);
@@ -2852,13 +3013,13 @@
 	ASSERT3U(zio->io_queued_timestamp, >, 0);
 	ASSERT(zio->io_stage == ZIO_STAGE_DVA_THROTTLE);
 
-	mutex_enter(&spa->spa_alloc_lock);
+	mutex_enter(&zio->io_mc->mc_alloc_lock);
 
 	ASSERT(zio->io_type == ZIO_TYPE_WRITE);
-	avl_add(&spa->spa_alloc_tree, zio);
+	avl_add(&zio->io_mc->mc_alloc_tree, zio);
 
-	nio = zio_io_to_allocate(zio->io_spa);
-	mutex_exit(&spa->spa_alloc_lock);
+	nio = zio_io_to_allocate(zio->io_mc);
+	mutex_exit(&zio->io_mc->mc_alloc_lock);
 
 	if (nio == zio)
 		return (ZIO_PIPELINE_CONTINUE);
@@ -2879,13 +3040,13 @@
 }
 
 void
-zio_allocate_dispatch(spa_t *spa)
+zio_allocate_dispatch(metaslab_class_t *mc)
 {
 	zio_t *zio;
 
-	mutex_enter(&spa->spa_alloc_lock);
-	zio = zio_io_to_allocate(spa);
-	mutex_exit(&spa->spa_alloc_lock);
+	mutex_enter(&mc->mc_alloc_lock);
+	zio = zio_io_to_allocate(mc);
+	mutex_exit(&mc->mc_alloc_lock);
 	if (zio == NULL)
 		return;
 
@@ -2898,7 +3059,8 @@
 zio_dva_allocate(zio_t *zio)
 {
 	spa_t *spa = zio->io_spa;
-	metaslab_class_t *mc = spa_normal_class(spa);
+	metaslab_class_t *mc = zio->io_mc;
+
 	blkptr_t *bp = zio->io_bp;
 	int error;
 	int flags = 0;
@@ -2914,13 +3076,14 @@
 	ASSERT3U(zio->io_prop.zp_copies, <=, spa_max_replication(spa));
 	ASSERT3U(zio->io_size, ==, BP_GET_PSIZE(bp));
 
-	if (zio->io_flags & ZIO_FLAG_NODATA) {
+	if (zio->io_flags & ZIO_FLAG_NODATA || zio->io_prop.zp_usewbc) {
 		flags |= METASLAB_DONT_THROTTLE;
 	}
 	if (zio->io_flags & ZIO_FLAG_GANG_CHILD) {
 		flags |= METASLAB_GANG_CHILD;
 	}
-	if (zio->io_priority == ZIO_PRIORITY_ASYNC_WRITE) {
+	if (zio->io_priority == ZIO_PRIORITY_ASYNC_WRITE &&
+	    zio->io_flags & ZIO_FLAG_IO_ALLOCATING) {
 		flags |= METASLAB_ASYNC_ALLOC;
 	}
 
@@ -2928,12 +3091,30 @@
 	    zio->io_prop.zp_copies, zio->io_txg, NULL, flags,
 	    &zio->io_alloc_list, zio);
 
+#ifdef _KERNEL
+	DTRACE_PROBE6(zio_dva_allocate,
+	    uint64_t, DVA_GET_VDEV(&bp->blk_dva[0]),
+	    uint64_t, DVA_GET_VDEV(&bp->blk_dva[1]),
+	    uint64_t, BP_GET_LEVEL(bp),
+	    boolean_t, BP_IS_SPECIAL(bp),
+	    boolean_t, BP_IS_METADATA(bp),
+	    int, error);
+#endif
+
 	if (error != 0) {
 		spa_dbgmsg(spa, "%s: metaslab allocation failure: zio %p, "
 		    "size %llu, error %d", spa_name(spa), zio, zio->io_size,
 		    error);
-		if (error == ENOSPC && zio->io_size > SPA_MINBLOCKSIZE)
+		if (error == ENOSPC && zio->io_size > SPA_MINBLOCKSIZE) {
+			if (zio->io_prop.zp_usewbc) {
+				zio->io_prop.zp_usewbc = B_FALSE;
+				zio->io_prop.zp_usesc = B_FALSE;
+				zio->io_mc = spa_normal_class(spa);
+			}
+
 			return (zio_write_gang_block(zio));
+		}
+
 		zio->io_error = error;
 	}
 
@@ -2991,21 +3172,64 @@
 {
 	int error = 1;
 	zio_alloc_list_t io_alloc_list;
+	spa_meta_placement_t *mp = &spa->spa_meta_policy;
 
 	ASSERT(txg > spa_syncing_txg(spa));
 
 	metaslab_trace_init(&io_alloc_list);
-	error = metaslab_alloc(spa, spa_log_class(spa), size, new_bp, 1,
-	    txg, old_bp, METASLAB_HINTBP_AVOID, &io_alloc_list, NULL);
-	if (error == 0) {
-		*slog = TRUE;
-	} else {
+
+	/*
+	 * ZIL blocks are always contiguous (i.e. not gang blocks)
+	 * so we set the METASLAB_HINTBP_AVOID flag so that they
+	 * don't "fast gang" when allocating them.
+	 * If the caller indicates that slog is not to be used
+	 * (via use_slog)
+	 * separate allocation class will not indeed be used,
+	 * independently of whether this is log or special
+	 */
+
+	if (spa_has_slogs(spa)) {
+		error = metaslab_alloc(spa, spa_log_class(spa),
+		    size, new_bp, 1, txg, old_bp,
+		    METASLAB_HINTBP_AVOID, &io_alloc_list, NULL);
+
+		DTRACE_PROBE2(zio_alloc_zil_log,
+		    spa_t *, spa, int, error);
+
+		if (error == 0)
+			*slog = TRUE;
+	}
+
+	/*
+	 * use special when failed to allocate from the regular
+	 * slog, but only if allowed and if the special used
+	 * space is  below watermarks
+	 */
+	if (error != 0 && spa_can_special_be_used(spa) &&
+	    mp->spa_sync_to_special != SYNC_TO_SPECIAL_DISABLED) {
+		error = metaslab_alloc(spa, spa_special_class(spa),
+		    size, new_bp, 1, txg, old_bp,
+		    METASLAB_HINTBP_AVOID, &io_alloc_list, NULL);
+
+		DTRACE_PROBE2(zio_alloc_zil_special,
+		    spa_t *, spa, int, error);
+
+		if (error == 0)
+			*slog = FALSE;
+	}
+
+	if (error != 0) {
 		error = metaslab_alloc(spa, spa_normal_class(spa), size,
 		    new_bp, 1, txg, old_bp, METASLAB_HINTBP_AVOID,
 		    &io_alloc_list, NULL);
+
+		DTRACE_PROBE2(zio_alloc_zil_normal,
+		    spa_t *, spa, int, error);
+
 		if (error == 0)
 			*slog = FALSE;
 	}
+
 	metaslab_trace_fini(&io_alloc_list);
 
 	if (error == 0) {
@@ -3062,6 +3286,8 @@
 	vdev_t *vd = zio->io_vd;
 	uint64_t align;
 	spa_t *spa = zio->io_spa;
+	zio_type_t type = zio->io_type;
+	zio->io_vd_timestamp = gethrtime();
 
 	ASSERT(zio->io_error == 0);
 	ASSERT(zio->io_child_error[ZIO_CHILD_VDEV] == 0);
@@ -3078,36 +3304,7 @@
 	}
 
 	ASSERT3P(zio->io_logical, !=, zio);
-	if (zio->io_type == ZIO_TYPE_WRITE) {
-		ASSERT(spa->spa_trust_config);
 
-		if (zio->io_vd->vdev_removing) {
-			ASSERT(zio->io_flags &
-			    (ZIO_FLAG_PHYSICAL | ZIO_FLAG_SELF_HEAL |
-			    ZIO_FLAG_INDUCE_DAMAGE));
-		}
-	}
-
-	/*
-	 * We keep track of time-sensitive I/Os so that the scan thread
-	 * can quickly react to certain workloads.  In particular, we care
-	 * about non-scrubbing, top-level reads and writes with the following
-	 * characteristics:
-	 *	- synchronous writes of user data to non-slog devices
-	 *	- any reads of user data
-	 * When these conditions are met, adjust the timestamp of spa_last_io
-	 * which allows the scan thread to adjust its workload accordingly.
-	 */
-	if (!(zio->io_flags & ZIO_FLAG_SCAN_THREAD) && zio->io_bp != NULL &&
-	    vd == vd->vdev_top && !vd->vdev_islog &&
-	    zio->io_bookmark.zb_objset != DMU_META_OBJSET &&
-	    zio->io_txg != spa_syncing_txg(spa)) {
-		uint64_t old = spa->spa_last_io;
-		uint64_t new = ddi_get_lbolt64();
-		if (old != new)
-			(void) atomic_cas_64(&spa->spa_last_io, old, new);
-	}
-
 	align = 1ULL << vd->vdev_top->vdev_ashift;
 
 	if (!(zio->io_flags & ZIO_FLAG_PHYSICAL) &&
@@ -3116,7 +3313,7 @@
 		uint64_t asize = P2ROUNDUP(zio->io_size, align);
 		abd_t *abuf = abd_alloc_sametype(zio->io_abd, asize);
 		ASSERT(vd == vd->vdev_top);
-		if (zio->io_type == ZIO_TYPE_WRITE) {
+		if (type == ZIO_TYPE_WRITE) {
 			abd_copy(abuf, zio->io_abd, zio->io_size);
 			abd_zero_off(abuf, zio->io_size, asize - zio->io_size);
 		}
@@ -3139,7 +3336,7 @@
 		ASSERT0(P2PHASE(zio->io_size, SPA_MINBLOCKSIZE));
 	}
 
-	VERIFY(zio->io_type != ZIO_TYPE_WRITE || spa_writeable(spa));
+	VERIFY(type != ZIO_TYPE_WRITE || spa_writeable(spa));
 
 	/*
 	 * If this is a repair I/O, and there's no self-healing involved --
@@ -3158,15 +3355,14 @@
 	    !(zio->io_flags & ZIO_FLAG_SELF_HEAL) &&
 	    zio->io_txg != 0 &&	/* not a delegated i/o */
 	    !vdev_dtl_contains(vd, DTL_PARTIAL, zio->io_txg, 1)) {
-		ASSERT(zio->io_type == ZIO_TYPE_WRITE);
+		ASSERT(type == ZIO_TYPE_WRITE);
 		zio_vdev_io_bypass(zio);
 		return (ZIO_PIPELINE_CONTINUE);
 	}
 
 	if (vd->vdev_ops->vdev_op_leaf &&
-	    (zio->io_type == ZIO_TYPE_READ || zio->io_type == ZIO_TYPE_WRITE)) {
-
-		if (zio->io_type == ZIO_TYPE_READ && vdev_cache_read(zio))
+	    (type == ZIO_TYPE_READ || type == ZIO_TYPE_WRITE)) {
+		if (type == ZIO_TYPE_READ && vdev_cache_read(zio))
 			return (ZIO_PIPELINE_CONTINUE);
 
 		if ((zio = vdev_queue_io(zio)) == NULL)
@@ -3177,6 +3373,15 @@
 			zio_interrupt(zio);
 			return (ZIO_PIPELINE_STOP);
 		}
+
+		/*
+		 * Insert a fault simulation delay for a particular vdev.
+		 */
+		if (zio_faulty_vdev_enabled &&
+		    (zio->io_vd->vdev_guid == zio_faulty_vdev_guid)) {
+			delay(NSEC_TO_TICK(zio_faulty_vdev_delay_us *
+			    (NANOSEC / MICROSEC)));
+		}
 	}
 
 	vd->vdev_ops->vdev_op_io_start(zio);
@@ -3190,14 +3395,12 @@
 	vdev_ops_t *ops = vd ? vd->vdev_ops : &vdev_mirror_ops;
 	boolean_t unexpected_error = B_FALSE;
 
-	if (zio_wait_for_children(zio, ZIO_CHILD_VDEV_BIT, ZIO_WAIT_DONE)) {
+	if (zio_wait_for_children(zio, ZIO_CHILD_VDEV, ZIO_WAIT_DONE))
 		return (ZIO_PIPELINE_STOP);
-	}
 
 	ASSERT(zio->io_type == ZIO_TYPE_READ || zio->io_type == ZIO_TYPE_WRITE);
 
 	if (vd != NULL && vd->vdev_ops->vdev_op_leaf) {
-
 		vdev_queue_io_done(zio);
 
 		if (zio->io_type == ZIO_TYPE_WRITE)
@@ -3224,6 +3427,12 @@
 	if (unexpected_error)
 		VERIFY(vdev_probe(vd, zio) == NULL);
 
+	/*
+	 * Measure delta between start and end of the I/O in nanoseconds.
+	 * XXX: Handle overflow.
+	 */
+	zio->io_vd_timestamp = gethrtime() - zio->io_vd_timestamp;
+
 	return (ZIO_PIPELINE_CONTINUE);
 }
 
@@ -3258,9 +3467,8 @@
 {
 	vdev_t *vd = zio->io_vd;
 
-	if (zio_wait_for_children(zio, ZIO_CHILD_VDEV_BIT, ZIO_WAIT_DONE)) {
+	if (zio_wait_for_children(zio, ZIO_CHILD_VDEV, ZIO_WAIT_DONE))
 		return (ZIO_PIPELINE_STOP);
-	}
 
 	if (vd == NULL && !(zio->io_flags & ZIO_FLAG_CONFIG_WRITER))
 		spa_config_exit(zio->io_spa, SCL_ZIO, zio);
@@ -3475,10 +3683,9 @@
 	zio_t *pio, *pio_next;
 	zio_link_t *zl = NULL;
 
-	if (zio_wait_for_children(zio, ZIO_CHILD_GANG_BIT | ZIO_CHILD_DDT_BIT,
-	    ZIO_WAIT_READY)) {
+	if (zio_wait_for_children(zio, ZIO_CHILD_GANG, ZIO_WAIT_READY) ||
+	    zio_wait_for_children(zio, ZIO_CHILD_DDT, ZIO_WAIT_READY))
 		return (ZIO_PIPELINE_STOP);
-	}
 
 	if (zio->io_ready) {
 		ASSERT(IO_IS_ALLOCATING(zio));
@@ -3502,10 +3709,9 @@
 			 * We were unable to allocate anything, unreserve and
 			 * issue the next I/O to allocate.
 			 */
-			metaslab_class_throttle_unreserve(
-			    spa_normal_class(zio->io_spa),
+			metaslab_class_throttle_unreserve(zio->io_mc,
 			    zio->io_prop.zp_copies, zio);
-			zio_allocate_dispatch(zio->io_spa);
+			zio_allocate_dispatch(zio->io_mc);
 		}
 	}
 
@@ -3591,8 +3797,7 @@
 	metaslab_group_alloc_decrement(zio->io_spa, vd->vdev_id, pio, flags);
 	mutex_exit(&pio->io_lock);
 
-	metaslab_class_throttle_unreserve(spa_normal_class(zio->io_spa),
-	    1, pio);
+	metaslab_class_throttle_unreserve(pio->io_mc, 1, pio);
 
 	/*
 	 * Call into the pipeline to see if there is more work that
@@ -3599,7 +3804,7 @@
 	 * needs to be done. If there is work to be done it will be
 	 * dispatched to another taskq thread.
 	 */
-	zio_allocate_dispatch(zio->io_spa);
+	zio_allocate_dispatch(pio->io_mc);
 }
 
 static int
@@ -3611,7 +3816,7 @@
 	vdev_t *vd = zio->io_vd;
 	uint64_t psize = zio->io_size;
 	zio_t *pio, *pio_next;
-	metaslab_class_t *mc = spa_normal_class(spa);
+	metaslab_class_t *mc = zio->io_mc;
 	zio_link_t *zl = NULL;
 
 	/*
@@ -3618,9 +3823,11 @@
 	 * If our children haven't all completed,
 	 * wait for them and then repeat this pipeline stage.
 	 */
-	if (zio_wait_for_children(zio, ZIO_CHILD_ALL_BITS, ZIO_WAIT_DONE)) {
+	if (zio_wait_for_children(zio, ZIO_CHILD_VDEV, ZIO_WAIT_DONE) ||
+	    zio_wait_for_children(zio, ZIO_CHILD_GANG, ZIO_WAIT_DONE) ||
+	    zio_wait_for_children(zio, ZIO_CHILD_DDT, ZIO_WAIT_DONE) ||
+	    zio_wait_for_children(zio, ZIO_CHILD_LOGICAL, ZIO_WAIT_DONE))
 		return (ZIO_PIPELINE_STOP);
-	}
 
 	/*
 	 * If the allocation throttle is enabled, then update the accounting.
@@ -3910,6 +4117,34 @@
 	return (ZIO_PIPELINE_STOP);
 }
 
+zio_t *
+zio_wbc(zio_type_t type, vdev_t *vd, abd_t *data,
+    uint64_t size, uint64_t offset)
+{
+	zio_t *zio = NULL;
+
+	switch (type) {
+	case ZIO_TYPE_WRITE:
+		zio = zio_create(NULL, vd->vdev_spa, 0, NULL, data, size,
+		    size, NULL, NULL, ZIO_TYPE_WRITE, ZIO_PRIORITY_ASYNC_WRITE,
+		    ZIO_FLAG_PHYSICAL, vd, offset,
+		    NULL, ZIO_STAGE_OPEN, ZIO_WRITE_PHYS_PIPELINE);
+		break;
+	case ZIO_TYPE_READ:
+		zio = zio_create(NULL, vd->vdev_spa, 0, NULL, data, size,
+		    size, NULL, NULL, ZIO_TYPE_READ, ZIO_PRIORITY_ASYNC_READ,
+		    ZIO_FLAG_DONT_CACHE | ZIO_FLAG_PHYSICAL, vd, offset,
+		    NULL, ZIO_STAGE_OPEN, ZIO_READ_PHYS_PIPELINE);
+		break;
+	default:
+		ASSERT(0);
+	}
+
+	zio->io_prop.zp_checksum = ZIO_CHECKSUM_OFF;
+
+	return (zio);
+}
+
 /*
  * ==========================================================================
  * I/O pipeline definition
