xref: /illumos-gate/usr/src/uts/common/fs/zfs/zio.c (revision 86714001)
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 (c) 2011, 2017 by Delphix. All rights reserved.
24  * Copyright (c) 2011 Nexenta Systems, Inc. All rights reserved.
25  * Copyright (c) 2014 Integros [integros.com]
26  */
27 
28 #include <sys/sysmacros.h>
29 #include <sys/zfs_context.h>
30 #include <sys/fm/fs/zfs.h>
31 #include <sys/spa.h>
32 #include <sys/txg.h>
33 #include <sys/spa_impl.h>
34 #include <sys/vdev_impl.h>
35 #include <sys/zio_impl.h>
36 #include <sys/zio_compress.h>
37 #include <sys/zio_checksum.h>
38 #include <sys/dmu_objset.h>
39 #include <sys/arc.h>
40 #include <sys/ddt.h>
41 #include <sys/blkptr.h>
42 #include <sys/zfeature.h>
43 #include <sys/metaslab_impl.h>
44 #include <sys/abd.h>
45 
46 /*
47  * ==========================================================================
48  * I/O type descriptions
49  * ==========================================================================
50  */
51 const char *zio_type_name[ZIO_TYPES] = {
52 	"zio_null", "zio_read", "zio_write", "zio_free", "zio_claim",
53 	"zio_ioctl"
54 };
55 
56 boolean_t zio_dva_throttle_enabled = B_TRUE;
57 
58 /*
59  * ==========================================================================
60  * I/O kmem caches
61  * ==========================================================================
62  */
63 kmem_cache_t *zio_cache;
64 kmem_cache_t *zio_link_cache;
65 kmem_cache_t *zio_buf_cache[SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT];
66 kmem_cache_t *zio_data_buf_cache[SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT];
67 
68 #ifdef _KERNEL
69 extern vmem_t *zio_alloc_arena;
70 #endif
71 
72 #define	ZIO_PIPELINE_CONTINUE		0x100
73 #define	ZIO_PIPELINE_STOP		0x101
74 
75 #define	BP_SPANB(indblkshift, level) \
76 	(((uint64_t)1) << ((level) * ((indblkshift) - SPA_BLKPTRSHIFT)))
77 #define	COMPARE_META_LEVEL	0x80000000ul
78 /*
79  * The following actions directly effect the spa's sync-to-convergence logic.
80  * The values below define the sync pass when we start performing the action.
81  * Care should be taken when changing these values as they directly impact
82  * spa_sync() performance. Tuning these values may introduce subtle performance
83  * pathologies and should only be done in the context of performance analysis.
84  * These tunables will eventually be removed and replaced with #defines once
85  * enough analysis has been done to determine optimal values.
86  *
87  * The 'zfs_sync_pass_deferred_free' pass must be greater than 1 to ensure that
88  * regular blocks are not deferred.
89  */
90 int zfs_sync_pass_deferred_free = 2; /* defer frees starting in this pass */
91 int zfs_sync_pass_dont_compress = 5; /* don't compress starting in this pass */
92 int zfs_sync_pass_rewrite = 2; /* rewrite new bps starting in this pass */
93 
94 /*
95  * An allocating zio is one that either currently has the DVA allocate
96  * stage set or will have it later in its lifetime.
97  */
98 #define	IO_IS_ALLOCATING(zio) ((zio)->io_orig_pipeline & ZIO_STAGE_DVA_ALLOCATE)
99 
100 boolean_t	zio_requeue_io_start_cut_in_line = B_TRUE;
101 
102 #ifdef ZFS_DEBUG
103 int zio_buf_debug_limit = 16384;
104 #else
105 int zio_buf_debug_limit = 0;
106 #endif
107 
108 static void zio_taskq_dispatch(zio_t *, zio_taskq_type_t, boolean_t);
109 
110 void
111 zio_init(void)
112 {
113 	size_t c;
114 	vmem_t *data_alloc_arena = NULL;
115 
116 #ifdef _KERNEL
117 	data_alloc_arena = zio_alloc_arena;
118 #endif
119 	zio_cache = kmem_cache_create("zio_cache",
120 	    sizeof (zio_t), 0, NULL, NULL, NULL, NULL, NULL, 0);
121 	zio_link_cache = kmem_cache_create("zio_link_cache",
122 	    sizeof (zio_link_t), 0, NULL, NULL, NULL, NULL, NULL, 0);
123 
124 	/*
125 	 * For small buffers, we want a cache for each multiple of
126 	 * SPA_MINBLOCKSIZE.  For larger buffers, we want a cache
127 	 * for each quarter-power of 2.
128 	 */
129 	for (c = 0; c < SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT; c++) {
130 		size_t size = (c + 1) << SPA_MINBLOCKSHIFT;
131 		size_t p2 = size;
132 		size_t align = 0;
133 		size_t cflags = (size > zio_buf_debug_limit) ? KMC_NODEBUG : 0;
134 
135 		while (!ISP2(p2))
136 			p2 &= p2 - 1;
137 
138 #ifndef _KERNEL
139 		/*
140 		 * If we are using watchpoints, put each buffer on its own page,
141 		 * to eliminate the performance overhead of trapping to the
142 		 * kernel when modifying a non-watched buffer that shares the
143 		 * page with a watched buffer.
144 		 */
145 		if (arc_watch && !IS_P2ALIGNED(size, PAGESIZE))
146 			continue;
147 #endif
148 		if (size <= 4 * SPA_MINBLOCKSIZE) {
149 			align = SPA_MINBLOCKSIZE;
150 		} else if (IS_P2ALIGNED(size, p2 >> 2)) {
151 			align = MIN(p2 >> 2, PAGESIZE);
152 		}
153 
154 		if (align != 0) {
155 			char name[36];
156 			(void) sprintf(name, "zio_buf_%lu", (ulong_t)size);
157 			zio_buf_cache[c] = kmem_cache_create(name, size,
158 			    align, NULL, NULL, NULL, NULL, NULL, cflags);
159 
160 			/*
161 			 * Since zio_data bufs do not appear in crash dumps, we
162 			 * pass KMC_NOTOUCH so that no allocator metadata is
163 			 * stored with the buffers.
164 			 */
165 			(void) sprintf(name, "zio_data_buf_%lu", (ulong_t)size);
166 			zio_data_buf_cache[c] = kmem_cache_create(name, size,
167 			    align, NULL, NULL, NULL, NULL, data_alloc_arena,
168 			    cflags | KMC_NOTOUCH);
169 		}
170 	}
171 
172 	while (--c != 0) {
173 		ASSERT(zio_buf_cache[c] != NULL);
174 		if (zio_buf_cache[c - 1] == NULL)
175 			zio_buf_cache[c - 1] = zio_buf_cache[c];
176 
177 		ASSERT(zio_data_buf_cache[c] != NULL);
178 		if (zio_data_buf_cache[c - 1] == NULL)
179 			zio_data_buf_cache[c - 1] = zio_data_buf_cache[c];
180 	}
181 
182 	zio_inject_init();
183 }
184 
185 void
186 zio_fini(void)
187 {
188 	size_t c;
189 	kmem_cache_t *last_cache = NULL;
190 	kmem_cache_t *last_data_cache = NULL;
191 
192 	for (c = 0; c < SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT; c++) {
193 		if (zio_buf_cache[c] != last_cache) {
194 			last_cache = zio_buf_cache[c];
195 			kmem_cache_destroy(zio_buf_cache[c]);
196 		}
197 		zio_buf_cache[c] = NULL;
198 
199 		if (zio_data_buf_cache[c] != last_data_cache) {
200 			last_data_cache = zio_data_buf_cache[c];
201 			kmem_cache_destroy(zio_data_buf_cache[c]);
202 		}
203 		zio_data_buf_cache[c] = NULL;
204 	}
205 
206 	kmem_cache_destroy(zio_link_cache);
207 	kmem_cache_destroy(zio_cache);
208 
209 	zio_inject_fini();
210 }
211 
212 /*
213  * ==========================================================================
214  * Allocate and free I/O buffers
215  * ==========================================================================
216  */
217 
218 /*
219  * Use zio_buf_alloc to allocate ZFS metadata.  This data will appear in a
220  * crashdump if the kernel panics, so use it judiciously.  Obviously, it's
221  * useful to inspect ZFS metadata, but if possible, we should avoid keeping
222  * excess / transient data in-core during a crashdump.
223  */
224 void *
225 zio_buf_alloc(size_t size)
226 {
227 	size_t c = (size - 1) >> SPA_MINBLOCKSHIFT;
228 
229 	VERIFY3U(c, <, SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT);
230 
231 	return (kmem_cache_alloc(zio_buf_cache[c], KM_PUSHPAGE));
232 }
233 
234 /*
235  * Use zio_data_buf_alloc to allocate data.  The data will not appear in a
236  * crashdump if the kernel panics.  This exists so that we will limit the amount
237  * of ZFS data that shows up in a kernel crashdump.  (Thus reducing the amount
238  * of kernel heap dumped to disk when the kernel panics)
239  */
240 void *
241 zio_data_buf_alloc(size_t size)
242 {
243 	size_t c = (size - 1) >> SPA_MINBLOCKSHIFT;
244 
245 	VERIFY3U(c, <, SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT);
246 
247 	return (kmem_cache_alloc(zio_data_buf_cache[c], KM_PUSHPAGE));
248 }
249 
250 void
251 zio_buf_free(void *buf, size_t size)
252 {
253 	size_t c = (size - 1) >> SPA_MINBLOCKSHIFT;
254 
255 	VERIFY3U(c, <, SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT);
256 
257 	kmem_cache_free(zio_buf_cache[c], buf);
258 }
259 
260 void
261 zio_data_buf_free(void *buf, size_t size)
262 {
263 	size_t c = (size - 1) >> SPA_MINBLOCKSHIFT;
264 
265 	VERIFY3U(c, <, SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT);
266 
267 	kmem_cache_free(zio_data_buf_cache[c], buf);
268 }
269 
270 /*
271  * ==========================================================================
272  * Push and pop I/O transform buffers
273  * ==========================================================================
274  */
275 void
276 zio_push_transform(zio_t *zio, abd_t *data, uint64_t size, uint64_t bufsize,
277     zio_transform_func_t *transform)
278 {
279 	zio_transform_t *zt = kmem_alloc(sizeof (zio_transform_t), KM_SLEEP);
280 
281 	/*
282 	 * Ensure that anyone expecting this zio to contain a linear ABD isn't
283 	 * going to get a nasty surprise when they try to access the data.
284 	 */
285 	IMPLY(abd_is_linear(zio->io_abd), abd_is_linear(data));
286 
287 	zt->zt_orig_abd = zio->io_abd;
288 	zt->zt_orig_size = zio->io_size;
289 	zt->zt_bufsize = bufsize;
290 	zt->zt_transform = transform;
291 
292 	zt->zt_next = zio->io_transform_stack;
293 	zio->io_transform_stack = zt;
294 
295 	zio->io_abd = data;
296 	zio->io_size = size;
297 }
298 
299 void
300 zio_pop_transforms(zio_t *zio)
301 {
302 	zio_transform_t *zt;
303 
304 	while ((zt = zio->io_transform_stack) != NULL) {
305 		if (zt->zt_transform != NULL)
306 			zt->zt_transform(zio,
307 			    zt->zt_orig_abd, zt->zt_orig_size);
308 
309 		if (zt->zt_bufsize != 0)
310 			abd_free(zio->io_abd);
311 
312 		zio->io_abd = zt->zt_orig_abd;
313 		zio->io_size = zt->zt_orig_size;
314 		zio->io_transform_stack = zt->zt_next;
315 
316 		kmem_free(zt, sizeof (zio_transform_t));
317 	}
318 }
319 
320 /*
321  * ==========================================================================
322  * I/O transform callbacks for subblocks and decompression
323  * ==========================================================================
324  */
325 static void
326 zio_subblock(zio_t *zio, abd_t *data, uint64_t size)
327 {
328 	ASSERT(zio->io_size > size);
329 
330 	if (zio->io_type == ZIO_TYPE_READ)
331 		abd_copy(data, zio->io_abd, size);
332 }
333 
334 static void
335 zio_decompress(zio_t *zio, abd_t *data, uint64_t size)
336 {
337 	if (zio->io_error == 0) {
338 		void *tmp = abd_borrow_buf(data, size);
339 		int ret = zio_decompress_data(BP_GET_COMPRESS(zio->io_bp),
340 		    zio->io_abd, tmp, zio->io_size, size);
341 		abd_return_buf_copy(data, tmp, size);
342 
343 		if (ret != 0)
344 			zio->io_error = SET_ERROR(EIO);
345 	}
346 }
347 
348 /*
349  * ==========================================================================
350  * I/O parent/child relationships and pipeline interlocks
351  * ==========================================================================
352  */
353 zio_t *
354 zio_walk_parents(zio_t *cio, zio_link_t **zl)
355 {
356 	list_t *pl = &cio->io_parent_list;
357 
358 	*zl = (*zl == NULL) ? list_head(pl) : list_next(pl, *zl);
359 	if (*zl == NULL)
360 		return (NULL);
361 
362 	ASSERT((*zl)->zl_child == cio);
363 	return ((*zl)->zl_parent);
364 }
365 
366 zio_t *
367 zio_walk_children(zio_t *pio, zio_link_t **zl)
368 {
369 	list_t *cl = &pio->io_child_list;
370 
371 	*zl = (*zl == NULL) ? list_head(cl) : list_next(cl, *zl);
372 	if (*zl == NULL)
373 		return (NULL);
374 
375 	ASSERT((*zl)->zl_parent == pio);
376 	return ((*zl)->zl_child);
377 }
378 
379 zio_t *
380 zio_unique_parent(zio_t *cio)
381 {
382 	zio_link_t *zl = NULL;
383 	zio_t *pio = zio_walk_parents(cio, &zl);
384 
385 	VERIFY3P(zio_walk_parents(cio, &zl), ==, NULL);
386 	return (pio);
387 }
388 
389 void
390 zio_add_child(zio_t *pio, zio_t *cio)
391 {
392 	zio_link_t *zl = kmem_cache_alloc(zio_link_cache, KM_SLEEP);
393 
394 	/*
395 	 * Logical I/Os can have logical, gang, or vdev children.
396 	 * Gang I/Os can have gang or vdev children.
397 	 * Vdev I/Os can only have vdev children.
398 	 * The following ASSERT captures all of these constraints.
399 	 */
400 	ASSERT3S(cio->io_child_type, <=, pio->io_child_type);
401 
402 	zl->zl_parent = pio;
403 	zl->zl_child = cio;
404 
405 	mutex_enter(&cio->io_lock);
406 	mutex_enter(&pio->io_lock);
407 
408 	ASSERT(pio->io_state[ZIO_WAIT_DONE] == 0);
409 
410 	for (int w = 0; w < ZIO_WAIT_TYPES; w++)
411 		pio->io_children[cio->io_child_type][w] += !cio->io_state[w];
412 
413 	list_insert_head(&pio->io_child_list, zl);
414 	list_insert_head(&cio->io_parent_list, zl);
415 
416 	pio->io_child_count++;
417 	cio->io_parent_count++;
418 
419 	mutex_exit(&pio->io_lock);
420 	mutex_exit(&cio->io_lock);
421 }
422 
423 static void
424 zio_remove_child(zio_t *pio, zio_t *cio, zio_link_t *zl)
425 {
426 	ASSERT(zl->zl_parent == pio);
427 	ASSERT(zl->zl_child == cio);
428 
429 	mutex_enter(&cio->io_lock);
430 	mutex_enter(&pio->io_lock);
431 
432 	list_remove(&pio->io_child_list, zl);
433 	list_remove(&cio->io_parent_list, zl);
434 
435 	pio->io_child_count--;
436 	cio->io_parent_count--;
437 
438 	mutex_exit(&pio->io_lock);
439 	mutex_exit(&cio->io_lock);
440 
441 	kmem_cache_free(zio_link_cache, zl);
442 }
443 
444 static boolean_t
445 zio_wait_for_children(zio_t *zio, uint8_t childbits, enum zio_wait_type wait)
446 {
447 	boolean_t waiting = B_FALSE;
448 
449 	mutex_enter(&zio->io_lock);
450 	ASSERT(zio->io_stall == NULL);
451 	for (int c = 0; c < ZIO_CHILD_TYPES; c++) {
452 		if (!(ZIO_CHILD_BIT_IS_SET(childbits, c)))
453 			continue;
454 
455 		uint64_t *countp = &zio->io_children[c][wait];
456 		if (*countp != 0) {
457 			zio->io_stage >>= 1;
458 			ASSERT3U(zio->io_stage, !=, ZIO_STAGE_OPEN);
459 			zio->io_stall = countp;
460 			waiting = B_TRUE;
461 			break;
462 		}
463 	}
464 	mutex_exit(&zio->io_lock);
465 	return (waiting);
466 }
467 
468 static void
469 zio_notify_parent(zio_t *pio, zio_t *zio, enum zio_wait_type wait)
470 {
471 	uint64_t *countp = &pio->io_children[zio->io_child_type][wait];
472 	int *errorp = &pio->io_child_error[zio->io_child_type];
473 
474 	mutex_enter(&pio->io_lock);
475 	if (zio->io_error && !(zio->io_flags & ZIO_FLAG_DONT_PROPAGATE))
476 		*errorp = zio_worst_error(*errorp, zio->io_error);
477 	pio->io_reexecute |= zio->io_reexecute;
478 	ASSERT3U(*countp, >, 0);
479 
480 	(*countp)--;
481 
482 	if (*countp == 0 && pio->io_stall == countp) {
483 		zio_taskq_type_t type =
484 		    pio->io_stage < ZIO_STAGE_VDEV_IO_START ? ZIO_TASKQ_ISSUE :
485 		    ZIO_TASKQ_INTERRUPT;
486 		pio->io_stall = NULL;
487 		mutex_exit(&pio->io_lock);
488 		/*
489 		 * Dispatch the parent zio in its own taskq so that
490 		 * the child can continue to make progress. This also
491 		 * prevents overflowing the stack when we have deeply nested
492 		 * parent-child relationships.
493 		 */
494 		zio_taskq_dispatch(pio, type, B_FALSE);
495 	} else {
496 		mutex_exit(&pio->io_lock);
497 	}
498 }
499 
500 static void
501 zio_inherit_child_errors(zio_t *zio, enum zio_child c)
502 {
503 	if (zio->io_child_error[c] != 0 && zio->io_error == 0)
504 		zio->io_error = zio->io_child_error[c];
505 }
506 
507 int
508 zio_bookmark_compare(const void *x1, const void *x2)
509 {
510 	const zio_t *z1 = x1;
511 	const zio_t *z2 = x2;
512 
513 	if (z1->io_bookmark.zb_objset < z2->io_bookmark.zb_objset)
514 		return (-1);
515 	if (z1->io_bookmark.zb_objset > z2->io_bookmark.zb_objset)
516 		return (1);
517 
518 	if (z1->io_bookmark.zb_object < z2->io_bookmark.zb_object)
519 		return (-1);
520 	if (z1->io_bookmark.zb_object > z2->io_bookmark.zb_object)
521 		return (1);
522 
523 	if (z1->io_bookmark.zb_level < z2->io_bookmark.zb_level)
524 		return (-1);
525 	if (z1->io_bookmark.zb_level > z2->io_bookmark.zb_level)
526 		return (1);
527 
528 	if (z1->io_bookmark.zb_blkid < z2->io_bookmark.zb_blkid)
529 		return (-1);
530 	if (z1->io_bookmark.zb_blkid > z2->io_bookmark.zb_blkid)
531 		return (1);
532 
533 	if (z1 < z2)
534 		return (-1);
535 	if (z1 > z2)
536 		return (1);
537 
538 	return (0);
539 }
540 
541 /*
542  * ==========================================================================
543  * Create the various types of I/O (read, write, free, etc)
544  * ==========================================================================
545  */
546 static zio_t *
547 zio_create(zio_t *pio, spa_t *spa, uint64_t txg, const blkptr_t *bp,
548     abd_t *data, uint64_t lsize, uint64_t psize, zio_done_func_t *done,
549     void *private, zio_type_t type, zio_priority_t priority,
550     enum zio_flag flags, vdev_t *vd, uint64_t offset,
551     const zbookmark_phys_t *zb, enum zio_stage stage, enum zio_stage pipeline)
552 {
553 	zio_t *zio;
554 
555 	ASSERT3U(psize, <=, SPA_MAXBLOCKSIZE);
556 	ASSERT(P2PHASE(psize, SPA_MINBLOCKSIZE) == 0);
557 	ASSERT(P2PHASE(offset, SPA_MINBLOCKSIZE) == 0);
558 
559 	ASSERT(!vd || spa_config_held(spa, SCL_STATE_ALL, RW_READER));
560 	ASSERT(!bp || !(flags & ZIO_FLAG_CONFIG_WRITER));
561 	ASSERT(vd || stage == ZIO_STAGE_OPEN);
562 
563 	IMPLY(lsize != psize, (flags & ZIO_FLAG_RAW) != 0);
564 
565 	zio = kmem_cache_alloc(zio_cache, KM_SLEEP);
566 	bzero(zio, sizeof (zio_t));
567 
568 	mutex_init(&zio->io_lock, NULL, MUTEX_DEFAULT, NULL);
569 	cv_init(&zio->io_cv, NULL, CV_DEFAULT, NULL);
570 
571 	list_create(&zio->io_parent_list, sizeof (zio_link_t),
572 	    offsetof(zio_link_t, zl_parent_node));
573 	list_create(&zio->io_child_list, sizeof (zio_link_t),
574 	    offsetof(zio_link_t, zl_child_node));
575 	metaslab_trace_init(&zio->io_alloc_list);
576 
577 	if (vd != NULL)
578 		zio->io_child_type = ZIO_CHILD_VDEV;
579 	else if (flags & ZIO_FLAG_GANG_CHILD)
580 		zio->io_child_type = ZIO_CHILD_GANG;
581 	else if (flags & ZIO_FLAG_DDT_CHILD)
582 		zio->io_child_type = ZIO_CHILD_DDT;
583 	else
584 		zio->io_child_type = ZIO_CHILD_LOGICAL;
585 
586 	if (bp != NULL) {
587 		zio->io_bp = (blkptr_t *)bp;
588 		zio->io_bp_copy = *bp;
589 		zio->io_bp_orig = *bp;
590 		if (type != ZIO_TYPE_WRITE ||
591 		    zio->io_child_type == ZIO_CHILD_DDT)
592 			zio->io_bp = &zio->io_bp_copy;	/* so caller can free */
593 		if (zio->io_child_type == ZIO_CHILD_LOGICAL)
594 			zio->io_logical = zio;
595 		if (zio->io_child_type > ZIO_CHILD_GANG && BP_IS_GANG(bp))
596 			pipeline |= ZIO_GANG_STAGES;
597 	}
598 
599 	zio->io_spa = spa;
600 	zio->io_txg = txg;
601 	zio->io_done = done;
602 	zio->io_private = private;
603 	zio->io_type = type;
604 	zio->io_priority = priority;
605 	zio->io_vd = vd;
606 	zio->io_offset = offset;
607 	zio->io_orig_abd = zio->io_abd = data;
608 	zio->io_orig_size = zio->io_size = psize;
609 	zio->io_lsize = lsize;
610 	zio->io_orig_flags = zio->io_flags = flags;
611 	zio->io_orig_stage = zio->io_stage = stage;
612 	zio->io_orig_pipeline = zio->io_pipeline = pipeline;
613 	zio->io_pipeline_trace = ZIO_STAGE_OPEN;
614 
615 	zio->io_state[ZIO_WAIT_READY] = (stage >= ZIO_STAGE_READY);
616 	zio->io_state[ZIO_WAIT_DONE] = (stage >= ZIO_STAGE_DONE);
617 
618 	if (zb != NULL)
619 		zio->io_bookmark = *zb;
620 
621 	if (pio != NULL) {
622 		if (zio->io_logical == NULL)
623 			zio->io_logical = pio->io_logical;
624 		if (zio->io_child_type == ZIO_CHILD_GANG)
625 			zio->io_gang_leader = pio->io_gang_leader;
626 		zio_add_child(pio, zio);
627 	}
628 
629 	return (zio);
630 }
631 
632 static void
633 zio_destroy(zio_t *zio)
634 {
635 	metaslab_trace_fini(&zio->io_alloc_list);
636 	list_destroy(&zio->io_parent_list);
637 	list_destroy(&zio->io_child_list);
638 	mutex_destroy(&zio->io_lock);
639 	cv_destroy(&zio->io_cv);
640 	kmem_cache_free(zio_cache, zio);
641 }
642 
643 zio_t *
644 zio_null(zio_t *pio, spa_t *spa, vdev_t *vd, zio_done_func_t *done,
645     void *private, enum zio_flag flags)
646 {
647 	zio_t *zio;
648 
649 	zio = zio_create(pio, spa, 0, NULL, NULL, 0, 0, done, private,
650 	    ZIO_TYPE_NULL, ZIO_PRIORITY_NOW, flags, vd, 0, NULL,
651 	    ZIO_STAGE_OPEN, ZIO_INTERLOCK_PIPELINE);
652 
653 	return (zio);
654 }
655 
656 zio_t *
657 zio_root(spa_t *spa, zio_done_func_t *done, void *private, enum zio_flag flags)
658 {
659 	return (zio_null(NULL, spa, NULL, done, private, flags));
660 }
661 
662 void
663 zfs_blkptr_verify(spa_t *spa, const blkptr_t *bp)
664 {
665 	if (!DMU_OT_IS_VALID(BP_GET_TYPE(bp))) {
666 		zfs_panic_recover("blkptr at %p has invalid TYPE %llu",
667 		    bp, (longlong_t)BP_GET_TYPE(bp));
668 	}
669 	if (BP_GET_CHECKSUM(bp) >= ZIO_CHECKSUM_FUNCTIONS ||
670 	    BP_GET_CHECKSUM(bp) <= ZIO_CHECKSUM_ON) {
671 		zfs_panic_recover("blkptr at %p has invalid CHECKSUM %llu",
672 		    bp, (longlong_t)BP_GET_CHECKSUM(bp));
673 	}
674 	if (BP_GET_COMPRESS(bp) >= ZIO_COMPRESS_FUNCTIONS ||
675 	    BP_GET_COMPRESS(bp) <= ZIO_COMPRESS_ON) {
676 		zfs_panic_recover("blkptr at %p has invalid COMPRESS %llu",
677 		    bp, (longlong_t)BP_GET_COMPRESS(bp));
678 	}
679 	if (BP_GET_LSIZE(bp) > SPA_MAXBLOCKSIZE) {
680 		zfs_panic_recover("blkptr at %p has invalid LSIZE %llu",
681 		    bp, (longlong_t)BP_GET_LSIZE(bp));
682 	}
683 	if (BP_GET_PSIZE(bp) > SPA_MAXBLOCKSIZE) {
684 		zfs_panic_recover("blkptr at %p has invalid PSIZE %llu",
685 		    bp, (longlong_t)BP_GET_PSIZE(bp));
686 	}
687 
688 	if (BP_IS_EMBEDDED(bp)) {
689 		if (BPE_GET_ETYPE(bp) > NUM_BP_EMBEDDED_TYPES) {
690 			zfs_panic_recover("blkptr at %p has invalid ETYPE %llu",
691 			    bp, (longlong_t)BPE_GET_ETYPE(bp));
692 		}
693 	}
694 
695 	/*
696 	 * Do not verify individual DVAs if the config is not trusted. This
697 	 * will be done once the zio is executed in vdev_mirror_map_alloc.
698 	 */
699 	if (!spa->spa_trust_config)
700 		return;
701 
702 	/*
703 	 * Pool-specific checks.
704 	 *
705 	 * Note: it would be nice to verify that the blk_birth and
706 	 * BP_PHYSICAL_BIRTH() are not too large.  However, spa_freeze()
707 	 * allows the birth time of log blocks (and dmu_sync()-ed blocks
708 	 * that are in the log) to be arbitrarily large.
709 	 */
710 	for (int i = 0; i < BP_GET_NDVAS(bp); i++) {
711 		uint64_t vdevid = DVA_GET_VDEV(&bp->blk_dva[i]);
712 		if (vdevid >= spa->spa_root_vdev->vdev_children) {
713 			zfs_panic_recover("blkptr at %p DVA %u has invalid "
714 			    "VDEV %llu",
715 			    bp, i, (longlong_t)vdevid);
716 			continue;
717 		}
718 		vdev_t *vd = spa->spa_root_vdev->vdev_child[vdevid];
719 		if (vd == NULL) {
720 			zfs_panic_recover("blkptr at %p DVA %u has invalid "
721 			    "VDEV %llu",
722 			    bp, i, (longlong_t)vdevid);
723 			continue;
724 		}
725 		if (vd->vdev_ops == &vdev_hole_ops) {
726 			zfs_panic_recover("blkptr at %p DVA %u has hole "
727 			    "VDEV %llu",
728 			    bp, i, (longlong_t)vdevid);
729 			continue;
730 		}
731 		if (vd->vdev_ops == &vdev_missing_ops) {
732 			/*
733 			 * "missing" vdevs are valid during import, but we
734 			 * don't have their detailed info (e.g. asize), so
735 			 * we can't perform any more checks on them.
736 			 */
737 			continue;
738 		}
739 		uint64_t offset = DVA_GET_OFFSET(&bp->blk_dva[i]);
740 		uint64_t asize = DVA_GET_ASIZE(&bp->blk_dva[i]);
741 		if (BP_IS_GANG(bp))
742 			asize = vdev_psize_to_asize(vd, SPA_GANGBLOCKSIZE);
743 		if (offset + asize > vd->vdev_asize) {
744 			zfs_panic_recover("blkptr at %p DVA %u has invalid "
745 			    "OFFSET %llu",
746 			    bp, i, (longlong_t)offset);
747 		}
748 	}
749 }
750 
751 boolean_t
752 zfs_dva_valid(spa_t *spa, const dva_t *dva, const blkptr_t *bp)
753 {
754 	uint64_t vdevid = DVA_GET_VDEV(dva);
755 
756 	if (vdevid >= spa->spa_root_vdev->vdev_children)
757 		return (B_FALSE);
758 
759 	vdev_t *vd = spa->spa_root_vdev->vdev_child[vdevid];
760 	if (vd == NULL)
761 		return (B_FALSE);
762 
763 	if (vd->vdev_ops == &vdev_hole_ops)
764 		return (B_FALSE);
765 
766 	if (vd->vdev_ops == &vdev_missing_ops) {
767 		return (B_FALSE);
768 	}
769 
770 	uint64_t offset = DVA_GET_OFFSET(dva);
771 	uint64_t asize = DVA_GET_ASIZE(dva);
772 
773 	if (BP_IS_GANG(bp))
774 		asize = vdev_psize_to_asize(vd, SPA_GANGBLOCKSIZE);
775 	if (offset + asize > vd->vdev_asize)
776 		return (B_FALSE);
777 
778 	return (B_TRUE);
779 }
780 
781 zio_t *
782 zio_read(zio_t *pio, spa_t *spa, const blkptr_t *bp,
783     abd_t *data, uint64_t size, zio_done_func_t *done, void *private,
784     zio_priority_t priority, enum zio_flag flags, const zbookmark_phys_t *zb)
785 {
786 	zio_t *zio;
787 
788 	zfs_blkptr_verify(spa, bp);
789 
790 	zio = zio_create(pio, spa, BP_PHYSICAL_BIRTH(bp), bp,
791 	    data, size, size, done, private,
792 	    ZIO_TYPE_READ, priority, flags, NULL, 0, zb,
793 	    ZIO_STAGE_OPEN, (flags & ZIO_FLAG_DDT_CHILD) ?
794 	    ZIO_DDT_CHILD_READ_PIPELINE : ZIO_READ_PIPELINE);
795 
796 	return (zio);
797 }
798 
799 zio_t *
800 zio_write(zio_t *pio, spa_t *spa, uint64_t txg, blkptr_t *bp,
801     abd_t *data, uint64_t lsize, uint64_t psize, const zio_prop_t *zp,
802     zio_done_func_t *ready, zio_done_func_t *children_ready,
803     zio_done_func_t *physdone, zio_done_func_t *done,
804     void *private, zio_priority_t priority, enum zio_flag flags,
805     const zbookmark_phys_t *zb)
806 {
807 	zio_t *zio;
808 
809 	ASSERT(zp->zp_checksum >= ZIO_CHECKSUM_OFF &&
810 	    zp->zp_checksum < ZIO_CHECKSUM_FUNCTIONS &&
811 	    zp->zp_compress >= ZIO_COMPRESS_OFF &&
812 	    zp->zp_compress < ZIO_COMPRESS_FUNCTIONS &&
813 	    DMU_OT_IS_VALID(zp->zp_type) &&
814 	    zp->zp_level < 32 &&
815 	    zp->zp_copies > 0 &&
816 	    zp->zp_copies <= spa_max_replication(spa));
817 
818 	zio = zio_create(pio, spa, txg, bp, data, lsize, psize, done, private,
819 	    ZIO_TYPE_WRITE, priority, flags, NULL, 0, zb,
820 	    ZIO_STAGE_OPEN, (flags & ZIO_FLAG_DDT_CHILD) ?
821 	    ZIO_DDT_CHILD_WRITE_PIPELINE : ZIO_WRITE_PIPELINE);
822 
823 	zio->io_ready = ready;
824 	zio->io_children_ready = children_ready;
825 	zio->io_physdone = physdone;
826 	zio->io_prop = *zp;
827 
828 	/*
829 	 * Data can be NULL if we are going to call zio_write_override() to
830 	 * provide the already-allocated BP.  But we may need the data to
831 	 * verify a dedup hit (if requested).  In this case, don't try to
832 	 * dedup (just take the already-allocated BP verbatim).
833 	 */
834 	if (data == NULL && zio->io_prop.zp_dedup_verify) {
835 		zio->io_prop.zp_dedup = zio->io_prop.zp_dedup_verify = B_FALSE;
836 	}
837 
838 	return (zio);
839 }
840 
841 zio_t *
842 zio_rewrite(zio_t *pio, spa_t *spa, uint64_t txg, blkptr_t *bp, abd_t *data,
843     uint64_t size, zio_done_func_t *done, void *private,
844     zio_priority_t priority, enum zio_flag flags, zbookmark_phys_t *zb)
845 {
846 	zio_t *zio;
847 
848 	zio = zio_create(pio, spa, txg, bp, data, size, size, done, private,
849 	    ZIO_TYPE_WRITE, priority, flags | ZIO_FLAG_IO_REWRITE, NULL, 0, zb,
850 	    ZIO_STAGE_OPEN, ZIO_REWRITE_PIPELINE);
851 
852 	return (zio);
853 }
854 
855 void
856 zio_write_override(zio_t *zio, blkptr_t *bp, int copies, boolean_t nopwrite)
857 {
858 	ASSERT(zio->io_type == ZIO_TYPE_WRITE);
859 	ASSERT(zio->io_child_type == ZIO_CHILD_LOGICAL);
860 	ASSERT(zio->io_stage == ZIO_STAGE_OPEN);
861 	ASSERT(zio->io_txg == spa_syncing_txg(zio->io_spa));
862 
863 	/*
864 	 * We must reset the io_prop to match the values that existed
865 	 * when the bp was first written by dmu_sync() keeping in mind
866 	 * that nopwrite and dedup are mutually exclusive.
867 	 */
868 	zio->io_prop.zp_dedup = nopwrite ? B_FALSE : zio->io_prop.zp_dedup;
869 	zio->io_prop.zp_nopwrite = nopwrite;
870 	zio->io_prop.zp_copies = copies;
871 	zio->io_bp_override = bp;
872 }
873 
874 void
875 zio_free(spa_t *spa, uint64_t txg, const blkptr_t *bp)
876 {
877 
878 	zfs_blkptr_verify(spa, bp);
879 
880 	/*
881 	 * The check for EMBEDDED is a performance optimization.  We
882 	 * process the free here (by ignoring it) rather than
883 	 * putting it on the list and then processing it in zio_free_sync().
884 	 */
885 	if (BP_IS_EMBEDDED(bp))
886 		return;
887 	metaslab_check_free(spa, bp);
888 
889 	/*
890 	 * Frees that are for the currently-syncing txg, are not going to be
891 	 * deferred, and which will not need to do a read (i.e. not GANG or
892 	 * DEDUP), can be processed immediately.  Otherwise, put them on the
893 	 * in-memory list for later processing.
894 	 */
895 	if (BP_IS_GANG(bp) || BP_GET_DEDUP(bp) ||
896 	    txg != spa->spa_syncing_txg ||
897 	    spa_sync_pass(spa) >= zfs_sync_pass_deferred_free) {
898 		bplist_append(&spa->spa_free_bplist[txg & TXG_MASK], bp);
899 	} else {
900 		VERIFY0(zio_wait(zio_free_sync(NULL, spa, txg, bp, 0)));
901 	}
902 }
903 
904 zio_t *
905 zio_free_sync(zio_t *pio, spa_t *spa, uint64_t txg, const blkptr_t *bp,
906     enum zio_flag flags)
907 {
908 	zio_t *zio;
909 	enum zio_stage stage = ZIO_FREE_PIPELINE;
910 
911 	ASSERT(!BP_IS_HOLE(bp));
912 	ASSERT(spa_syncing_txg(spa) == txg);
913 	ASSERT(spa_sync_pass(spa) < zfs_sync_pass_deferred_free);
914 
915 	if (BP_IS_EMBEDDED(bp))
916 		return (zio_null(pio, spa, NULL, NULL, NULL, 0));
917 
918 	metaslab_check_free(spa, bp);
919 	arc_freed(spa, bp);
920 
921 	/*
922 	 * GANG and DEDUP blocks can induce a read (for the gang block header,
923 	 * or the DDT), so issue them asynchronously so that this thread is
924 	 * not tied up.
925 	 */
926 	if (BP_IS_GANG(bp) || BP_GET_DEDUP(bp))
927 		stage |= ZIO_STAGE_ISSUE_ASYNC;
928 
929 	zio = zio_create(pio, spa, txg, bp, NULL, BP_GET_PSIZE(bp),
930 	    BP_GET_PSIZE(bp), NULL, NULL, ZIO_TYPE_FREE, ZIO_PRIORITY_NOW,
931 	    flags, NULL, 0, NULL, ZIO_STAGE_OPEN, stage);
932 
933 	return (zio);
934 }
935 
936 zio_t *
937 zio_claim(zio_t *pio, spa_t *spa, uint64_t txg, const blkptr_t *bp,
938     zio_done_func_t *done, void *private, enum zio_flag flags)
939 {
940 	zio_t *zio;
941 
942 	zfs_blkptr_verify(spa, bp);
943 
944 	if (BP_IS_EMBEDDED(bp))
945 		return (zio_null(pio, spa, NULL, NULL, NULL, 0));
946 
947 	/*
948 	 * A claim is an allocation of a specific block.  Claims are needed
949 	 * to support immediate writes in the intent log.  The issue is that
950 	 * immediate writes contain committed data, but in a txg that was
951 	 * *not* committed.  Upon opening the pool after an unclean shutdown,
952 	 * the intent log claims all blocks that contain immediate write data
953 	 * so that the SPA knows they're in use.
954 	 *
955 	 * All claims *must* be resolved in the first txg -- before the SPA
956 	 * starts allocating blocks -- so that nothing is allocated twice.
957 	 * If txg == 0 we just verify that the block is claimable.
958 	 */
959 	ASSERT3U(spa->spa_uberblock.ub_rootbp.blk_birth, <,
960 	    spa_min_claim_txg(spa));
961 	ASSERT(txg == spa_min_claim_txg(spa) || txg == 0);
962 	ASSERT(!BP_GET_DEDUP(bp) || !spa_writeable(spa));	/* zdb(1M) */
963 
964 	zio = zio_create(pio, spa, txg, bp, NULL, BP_GET_PSIZE(bp),
965 	    BP_GET_PSIZE(bp), done, private, ZIO_TYPE_CLAIM, ZIO_PRIORITY_NOW,
966 	    flags, NULL, 0, NULL, ZIO_STAGE_OPEN, ZIO_CLAIM_PIPELINE);
967 	ASSERT0(zio->io_queued_timestamp);
968 
969 	return (zio);
970 }
971 
972 zio_t *
973 zio_ioctl(zio_t *pio, spa_t *spa, vdev_t *vd, int cmd,
974     zio_done_func_t *done, void *private, enum zio_flag flags)
975 {
976 	zio_t *zio;
977 	int c;
978 
979 	if (vd->vdev_children == 0) {
980 		zio = zio_create(pio, spa, 0, NULL, NULL, 0, 0, done, private,
981 		    ZIO_TYPE_IOCTL, ZIO_PRIORITY_NOW, flags, vd, 0, NULL,
982 		    ZIO_STAGE_OPEN, ZIO_IOCTL_PIPELINE);
983 
984 		zio->io_cmd = cmd;
985 	} else {
986 		zio = zio_null(pio, spa, NULL, NULL, NULL, flags);
987 
988 		for (c = 0; c < vd->vdev_children; c++)
989 			zio_nowait(zio_ioctl(zio, spa, vd->vdev_child[c], cmd,
990 			    done, private, flags));
991 	}
992 
993 	return (zio);
994 }
995 
996 zio_t *
997 zio_read_phys(zio_t *pio, vdev_t *vd, uint64_t offset, uint64_t size,
998     abd_t *data, int checksum, zio_done_func_t *done, void *private,
999     zio_priority_t priority, enum zio_flag flags, boolean_t labels)
1000 {
1001 	zio_t *zio;
1002 
1003 	ASSERT(vd->vdev_children == 0);
1004 	ASSERT(!labels || offset + size <= VDEV_LABEL_START_SIZE ||
1005 	    offset >= vd->vdev_psize - VDEV_LABEL_END_SIZE);
1006 	ASSERT3U(offset + size, <=, vd->vdev_psize);
1007 
1008 	zio = zio_create(pio, vd->vdev_spa, 0, NULL, data, size, size, done,
1009 	    private, ZIO_TYPE_READ, priority, flags | ZIO_FLAG_PHYSICAL, vd,
1010 	    offset, NULL, ZIO_STAGE_OPEN, ZIO_READ_PHYS_PIPELINE);
1011 
1012 	zio->io_prop.zp_checksum = checksum;
1013 
1014 	return (zio);
1015 }
1016 
1017 zio_t *
1018 zio_write_phys(zio_t *pio, vdev_t *vd, uint64_t offset, uint64_t size,
1019     abd_t *data, int checksum, zio_done_func_t *done, void *private,
1020     zio_priority_t priority, enum zio_flag flags, boolean_t labels)
1021 {
1022 	zio_t *zio;
1023 
1024 	ASSERT(vd->vdev_children == 0);
1025 	ASSERT(!labels || offset + size <= VDEV_LABEL_START_SIZE ||
1026 	    offset >= vd->vdev_psize - VDEV_LABEL_END_SIZE);
1027 	ASSERT3U(offset + size, <=, vd->vdev_psize);
1028 
1029 	zio = zio_create(pio, vd->vdev_spa, 0, NULL, data, size, size, done,
1030 	    private, ZIO_TYPE_WRITE, priority, flags | ZIO_FLAG_PHYSICAL, vd,
1031 	    offset, NULL, ZIO_STAGE_OPEN, ZIO_WRITE_PHYS_PIPELINE);
1032 
1033 	zio->io_prop.zp_checksum = checksum;
1034 
1035 	if (zio_checksum_table[checksum].ci_flags & ZCHECKSUM_FLAG_EMBEDDED) {
1036 		/*
1037 		 * zec checksums are necessarily destructive -- they modify
1038 		 * the end of the write buffer to hold the verifier/checksum.
1039 		 * Therefore, we must make a local copy in case the data is
1040 		 * being written to multiple places in parallel.
1041 		 */
1042 		abd_t *wbuf = abd_alloc_sametype(data, size);
1043 		abd_copy(wbuf, data, size);
1044 
1045 		zio_push_transform(zio, wbuf, size, size, NULL);
1046 	}
1047 
1048 	return (zio);
1049 }
1050 
1051 /*
1052  * Create a child I/O to do some work for us.
1053  */
1054 zio_t *
1055 zio_vdev_child_io(zio_t *pio, blkptr_t *bp, vdev_t *vd, uint64_t offset,
1056     abd_t *data, uint64_t size, int type, zio_priority_t priority,
1057     enum zio_flag flags, zio_done_func_t *done, void *private)
1058 {
1059 	enum zio_stage pipeline = ZIO_VDEV_CHILD_PIPELINE;
1060 	zio_t *zio;
1061 
1062 	/*
1063 	 * vdev child I/Os do not propagate their error to the parent.
1064 	 * Therefore, for correct operation the caller *must* check for
1065 	 * and handle the error in the child i/o's done callback.
1066 	 * The only exceptions are i/os that we don't care about
1067 	 * (OPTIONAL or REPAIR).
1068 	 */
1069 	ASSERT((flags & ZIO_FLAG_OPTIONAL) || (flags & ZIO_FLAG_IO_REPAIR) ||
1070 	    done != NULL);
1071 
1072 	/*
1073 	 * In the common case, where the parent zio was to a normal vdev,
1074 	 * the child zio must be to a child vdev of that vdev.  Otherwise,
1075 	 * the child zio must be to a top-level vdev.
1076 	 */
1077 	if (pio->io_vd != NULL && pio->io_vd->vdev_ops != &vdev_indirect_ops) {
1078 		ASSERT3P(vd->vdev_parent, ==, pio->io_vd);
1079 	} else {
1080 		ASSERT3P(vd, ==, vd->vdev_top);
1081 	}
1082 
1083 	if (type == ZIO_TYPE_READ && bp != NULL) {
1084 		/*
1085 		 * If we have the bp, then the child should perform the
1086 		 * checksum and the parent need not.  This pushes error
1087 		 * detection as close to the leaves as possible and
1088 		 * eliminates redundant checksums in the interior nodes.
1089 		 */
1090 		pipeline |= ZIO_STAGE_CHECKSUM_VERIFY;
1091 		pio->io_pipeline &= ~ZIO_STAGE_CHECKSUM_VERIFY;
1092 	}
1093 
1094 	if (vd->vdev_ops->vdev_op_leaf) {
1095 		ASSERT0(vd->vdev_children);
1096 		offset += VDEV_LABEL_START_SIZE;
1097 	}
1098 
1099 	flags |= ZIO_VDEV_CHILD_FLAGS(pio);
1100 
1101 	/*
1102 	 * If we've decided to do a repair, the write is not speculative --
1103 	 * even if the original read was.
1104 	 */
1105 	if (flags & ZIO_FLAG_IO_REPAIR)
1106 		flags &= ~ZIO_FLAG_SPECULATIVE;
1107 
1108 	/*
1109 	 * If we're creating a child I/O that is not associated with a
1110 	 * top-level vdev, then the child zio is not an allocating I/O.
1111 	 * If this is a retried I/O then we ignore it since we will
1112 	 * have already processed the original allocating I/O.
1113 	 */
1114 	if (flags & ZIO_FLAG_IO_ALLOCATING &&
1115 	    (vd != vd->vdev_top || (flags & ZIO_FLAG_IO_RETRY))) {
1116 		metaslab_class_t *mc = spa_normal_class(pio->io_spa);
1117 
1118 		ASSERT(mc->mc_alloc_throttle_enabled);
1119 		ASSERT(type == ZIO_TYPE_WRITE);
1120 		ASSERT(priority == ZIO_PRIORITY_ASYNC_WRITE);
1121 		ASSERT(!(flags & ZIO_FLAG_IO_REPAIR));
1122 		ASSERT(!(pio->io_flags & ZIO_FLAG_IO_REWRITE) ||
1123 		    pio->io_child_type == ZIO_CHILD_GANG);
1124 
1125 		flags &= ~ZIO_FLAG_IO_ALLOCATING;
1126 	}
1127 
1128 	zio = zio_create(pio, pio->io_spa, pio->io_txg, bp, data, size, size,
1129 	    done, private, type, priority, flags, vd, offset, &pio->io_bookmark,
1130 	    ZIO_STAGE_VDEV_IO_START >> 1, pipeline);
1131 	ASSERT3U(zio->io_child_type, ==, ZIO_CHILD_VDEV);
1132 
1133 	zio->io_physdone = pio->io_physdone;
1134 	if (vd->vdev_ops->vdev_op_leaf && zio->io_logical != NULL)
1135 		zio->io_logical->io_phys_children++;
1136 
1137 	return (zio);
1138 }
1139 
1140 zio_t *
1141 zio_vdev_delegated_io(vdev_t *vd, uint64_t offset, abd_t *data, uint64_t size,
1142     int type, zio_priority_t priority, enum zio_flag flags,
1143     zio_done_func_t *done, void *private)
1144 {
1145 	zio_t *zio;
1146 
1147 	ASSERT(vd->vdev_ops->vdev_op_leaf);
1148 
1149 	zio = zio_create(NULL, vd->vdev_spa, 0, NULL,
1150 	    data, size, size, done, private, type, priority,
1151 	    flags | ZIO_FLAG_CANFAIL | ZIO_FLAG_DONT_RETRY | ZIO_FLAG_DELEGATED,
1152 	    vd, offset, NULL,
1153 	    ZIO_STAGE_VDEV_IO_START >> 1, ZIO_VDEV_CHILD_PIPELINE);
1154 
1155 	return (zio);
1156 }
1157 
1158 void
1159 zio_flush(zio_t *zio, vdev_t *vd)
1160 {
1161 	zio_nowait(zio_ioctl(zio, zio->io_spa, vd, DKIOCFLUSHWRITECACHE,
1162 	    NULL, NULL,
1163 	    ZIO_FLAG_CANFAIL | ZIO_FLAG_DONT_PROPAGATE | ZIO_FLAG_DONT_RETRY));
1164 }
1165 
1166 void
1167 zio_shrink(zio_t *zio, uint64_t size)
1168 {
1169 	ASSERT3P(zio->io_executor, ==, NULL);
1170 	ASSERT3P(zio->io_orig_size, ==, zio->io_size);
1171 	ASSERT3U(size, <=, zio->io_size);
1172 
1173 	/*
1174 	 * We don't shrink for raidz because of problems with the
1175 	 * reconstruction when reading back less than the block size.
1176 	 * Note, BP_IS_RAIDZ() assumes no compression.
1177 	 */
1178 	ASSERT(BP_GET_COMPRESS(zio->io_bp) == ZIO_COMPRESS_OFF);
1179 	if (!BP_IS_RAIDZ(zio->io_bp)) {
1180 		/* we are not doing a raw write */
1181 		ASSERT3U(zio->io_size, ==, zio->io_lsize);
1182 		zio->io_orig_size = zio->io_size = zio->io_lsize = size;
1183 	}
1184 }
1185 
1186 /*
1187  * ==========================================================================
1188  * Prepare to read and write logical blocks
1189  * ==========================================================================
1190  */
1191 
1192 static int
1193 zio_read_bp_init(zio_t *zio)
1194 {
1195 	blkptr_t *bp = zio->io_bp;
1196 
1197 	ASSERT3P(zio->io_bp, ==, &zio->io_bp_copy);
1198 
1199 	if (BP_GET_COMPRESS(bp) != ZIO_COMPRESS_OFF &&
1200 	    zio->io_child_type == ZIO_CHILD_LOGICAL &&
1201 	    !(zio->io_flags & ZIO_FLAG_RAW)) {
1202 		uint64_t psize =
1203 		    BP_IS_EMBEDDED(bp) ? BPE_GET_PSIZE(bp) : BP_GET_PSIZE(bp);
1204 		zio_push_transform(zio, abd_alloc_sametype(zio->io_abd, psize),
1205 		    psize, psize, zio_decompress);
1206 	}
1207 
1208 	if (BP_IS_EMBEDDED(bp) && BPE_GET_ETYPE(bp) == BP_EMBEDDED_TYPE_DATA) {
1209 		zio->io_pipeline = ZIO_INTERLOCK_PIPELINE;
1210 
1211 		int psize = BPE_GET_PSIZE(bp);
1212 		void *data = abd_borrow_buf(zio->io_abd, psize);
1213 		decode_embedded_bp_compressed(bp, data);
1214 		abd_return_buf_copy(zio->io_abd, data, psize);
1215 	} else {
1216 		ASSERT(!BP_IS_EMBEDDED(bp));
1217 		ASSERT3P(zio->io_bp, ==, &zio->io_bp_copy);
1218 	}
1219 
1220 	if (!DMU_OT_IS_METADATA(BP_GET_TYPE(bp)) && BP_GET_LEVEL(bp) == 0)
1221 		zio->io_flags |= ZIO_FLAG_DONT_CACHE;
1222 
1223 	if (BP_GET_TYPE(bp) == DMU_OT_DDT_ZAP)
1224 		zio->io_flags |= ZIO_FLAG_DONT_CACHE;
1225 
1226 	if (BP_GET_DEDUP(bp) && zio->io_child_type == ZIO_CHILD_LOGICAL)
1227 		zio->io_pipeline = ZIO_DDT_READ_PIPELINE;
1228 
1229 	return (ZIO_PIPELINE_CONTINUE);
1230 }
1231 
1232 static int
1233 zio_write_bp_init(zio_t *zio)
1234 {
1235 	if (!IO_IS_ALLOCATING(zio))
1236 		return (ZIO_PIPELINE_CONTINUE);
1237 
1238 	ASSERT(zio->io_child_type != ZIO_CHILD_DDT);
1239 
1240 	if (zio->io_bp_override) {
1241 		blkptr_t *bp = zio->io_bp;
1242 		zio_prop_t *zp = &zio->io_prop;
1243 
1244 		ASSERT(bp->blk_birth != zio->io_txg);
1245 		ASSERT(BP_GET_DEDUP(zio->io_bp_override) == 0);
1246 
1247 		*bp = *zio->io_bp_override;
1248 		zio->io_pipeline = ZIO_INTERLOCK_PIPELINE;
1249 
1250 		if (BP_IS_EMBEDDED(bp))
1251 			return (ZIO_PIPELINE_CONTINUE);
1252 
1253 		/*
1254 		 * If we've been overridden and nopwrite is set then
1255 		 * set the flag accordingly to indicate that a nopwrite
1256 		 * has already occurred.
1257 		 */
1258 		if (!BP_IS_HOLE(bp) && zp->zp_nopwrite) {
1259 			ASSERT(!zp->zp_dedup);
1260 			ASSERT3U(BP_GET_CHECKSUM(bp), ==, zp->zp_checksum);
1261 			zio->io_flags |= ZIO_FLAG_NOPWRITE;
1262 			return (ZIO_PIPELINE_CONTINUE);
1263 		}
1264 
1265 		ASSERT(!zp->zp_nopwrite);
1266 
1267 		if (BP_IS_HOLE(bp) || !zp->zp_dedup)
1268 			return (ZIO_PIPELINE_CONTINUE);
1269 
1270 		ASSERT((zio_checksum_table[zp->zp_checksum].ci_flags &
1271 		    ZCHECKSUM_FLAG_DEDUP) || zp->zp_dedup_verify);
1272 
1273 		if (BP_GET_CHECKSUM(bp) == zp->zp_checksum) {
1274 			BP_SET_DEDUP(bp, 1);
1275 			zio->io_pipeline |= ZIO_STAGE_DDT_WRITE;
1276 			return (ZIO_PIPELINE_CONTINUE);
1277 		}
1278 
1279 		/*
1280 		 * We were unable to handle this as an override bp, treat
1281 		 * it as a regular write I/O.
1282 		 */
1283 		zio->io_bp_override = NULL;
1284 		*bp = zio->io_bp_orig;
1285 		zio->io_pipeline = zio->io_orig_pipeline;
1286 	}
1287 
1288 	return (ZIO_PIPELINE_CONTINUE);
1289 }
1290 
1291 static int
1292 zio_write_compress(zio_t *zio)
1293 {
1294 	spa_t *spa = zio->io_spa;
1295 	zio_prop_t *zp = &zio->io_prop;
1296 	enum zio_compress compress = zp->zp_compress;
1297 	blkptr_t *bp = zio->io_bp;
1298 	uint64_t lsize = zio->io_lsize;
1299 	uint64_t psize = zio->io_size;
1300 	int pass = 1;
1301 
1302 	EQUIV(lsize != psize, (zio->io_flags & ZIO_FLAG_RAW) != 0);
1303 
1304 	/*
1305 	 * If our children haven't all reached the ready stage,
1306 	 * wait for them and then repeat this pipeline stage.
1307 	 */
1308 	if (zio_wait_for_children(zio, ZIO_CHILD_LOGICAL_BIT |
1309 	    ZIO_CHILD_GANG_BIT, ZIO_WAIT_READY)) {
1310 		return (ZIO_PIPELINE_STOP);
1311 	}
1312 
1313 	if (!IO_IS_ALLOCATING(zio))
1314 		return (ZIO_PIPELINE_CONTINUE);
1315 
1316 	if (zio->io_children_ready != NULL) {
1317 		/*
1318 		 * Now that all our children are ready, run the callback
1319 		 * associated with this zio in case it wants to modify the
1320 		 * data to be written.
1321 		 */
1322 		ASSERT3U(zp->zp_level, >, 0);
1323 		zio->io_children_ready(zio);
1324 	}
1325 
1326 	ASSERT(zio->io_child_type != ZIO_CHILD_DDT);
1327 	ASSERT(zio->io_bp_override == NULL);
1328 
1329 	if (!BP_IS_HOLE(bp) && bp->blk_birth == zio->io_txg) {
1330 		/*
1331 		 * We're rewriting an existing block, which means we're
1332 		 * working on behalf of spa_sync().  For spa_sync() to
1333 		 * converge, it must eventually be the case that we don't
1334 		 * have to allocate new blocks.  But compression changes
1335 		 * the blocksize, which forces a reallocate, and makes
1336 		 * convergence take longer.  Therefore, after the first
1337 		 * few passes, stop compressing to ensure convergence.
1338 		 */
1339 		pass = spa_sync_pass(spa);
1340 
1341 		ASSERT(zio->io_txg == spa_syncing_txg(spa));
1342 		ASSERT(zio->io_child_type == ZIO_CHILD_LOGICAL);
1343 		ASSERT(!BP_GET_DEDUP(bp));
1344 
1345 		if (pass >= zfs_sync_pass_dont_compress)
1346 			compress = ZIO_COMPRESS_OFF;
1347 
1348 		/* Make sure someone doesn't change their mind on overwrites */
1349 		ASSERT(BP_IS_EMBEDDED(bp) || MIN(zp->zp_copies + BP_IS_GANG(bp),
1350 		    spa_max_replication(spa)) == BP_GET_NDVAS(bp));
1351 	}
1352 
1353 	/* If it's a compressed write that is not raw, compress the buffer. */
1354 	if (compress != ZIO_COMPRESS_OFF && psize == lsize) {
1355 		void *cbuf = zio_buf_alloc(lsize);
1356 		psize = zio_compress_data(compress, zio->io_abd, cbuf, lsize);
1357 		if (psize == 0 || psize == lsize) {
1358 			compress = ZIO_COMPRESS_OFF;
1359 			zio_buf_free(cbuf, lsize);
1360 		} else if (!zp->zp_dedup && psize <= BPE_PAYLOAD_SIZE &&
1361 		    zp->zp_level == 0 && !DMU_OT_HAS_FILL(zp->zp_type) &&
1362 		    spa_feature_is_enabled(spa, SPA_FEATURE_EMBEDDED_DATA)) {
1363 			encode_embedded_bp_compressed(bp,
1364 			    cbuf, compress, lsize, psize);
1365 			BPE_SET_ETYPE(bp, BP_EMBEDDED_TYPE_DATA);
1366 			BP_SET_TYPE(bp, zio->io_prop.zp_type);
1367 			BP_SET_LEVEL(bp, zio->io_prop.zp_level);
1368 			zio_buf_free(cbuf, lsize);
1369 			bp->blk_birth = zio->io_txg;
1370 			zio->io_pipeline = ZIO_INTERLOCK_PIPELINE;
1371 			ASSERT(spa_feature_is_active(spa,
1372 			    SPA_FEATURE_EMBEDDED_DATA));
1373 			return (ZIO_PIPELINE_CONTINUE);
1374 		} else {
1375 			/*
1376 			 * Round up compressed size up to the ashift
1377 			 * of the smallest-ashift device, and zero the tail.
1378 			 * This ensures that the compressed size of the BP
1379 			 * (and thus compressratio property) are correct,
1380 			 * in that we charge for the padding used to fill out
1381 			 * the last sector.
1382 			 */
1383 			ASSERT3U(spa->spa_min_ashift, >=, SPA_MINBLOCKSHIFT);
1384 			size_t rounded = (size_t)P2ROUNDUP(psize,
1385 			    1ULL << spa->spa_min_ashift);
1386 			if (rounded >= lsize) {
1387 				compress = ZIO_COMPRESS_OFF;
1388 				zio_buf_free(cbuf, lsize);
1389 				psize = lsize;
1390 			} else {
1391 				abd_t *cdata = abd_get_from_buf(cbuf, lsize);
1392 				abd_take_ownership_of_buf(cdata, B_TRUE);
1393 				abd_zero_off(cdata, psize, rounded - psize);
1394 				psize = rounded;
1395 				zio_push_transform(zio, cdata,
1396 				    psize, lsize, NULL);
1397 			}
1398 		}
1399 
1400 		/*
1401 		 * We were unable to handle this as an override bp, treat
1402 		 * it as a regular write I/O.
1403 		 */
1404 		zio->io_bp_override = NULL;
1405 		*bp = zio->io_bp_orig;
1406 		zio->io_pipeline = zio->io_orig_pipeline;
1407 	} else {
1408 		ASSERT3U(psize, !=, 0);
1409 	}
1410 
1411 	/*
1412 	 * The final pass of spa_sync() must be all rewrites, but the first
1413 	 * few passes offer a trade-off: allocating blocks defers convergence,
1414 	 * but newly allocated blocks are sequential, so they can be written
1415 	 * to disk faster.  Therefore, we allow the first few passes of
1416 	 * spa_sync() to allocate new blocks, but force rewrites after that.
1417 	 * There should only be a handful of blocks after pass 1 in any case.
1418 	 */
1419 	if (!BP_IS_HOLE(bp) && bp->blk_birth == zio->io_txg &&
1420 	    BP_GET_PSIZE(bp) == psize &&
1421 	    pass >= zfs_sync_pass_rewrite) {
1422 		ASSERT(psize != 0);
1423 		enum zio_stage gang_stages = zio->io_pipeline & ZIO_GANG_STAGES;
1424 		zio->io_pipeline = ZIO_REWRITE_PIPELINE | gang_stages;
1425 		zio->io_flags |= ZIO_FLAG_IO_REWRITE;
1426 	} else {
1427 		BP_ZERO(bp);
1428 		zio->io_pipeline = ZIO_WRITE_PIPELINE;
1429 	}
1430 
1431 	if (psize == 0) {
1432 		if (zio->io_bp_orig.blk_birth != 0 &&
1433 		    spa_feature_is_active(spa, SPA_FEATURE_HOLE_BIRTH)) {
1434 			BP_SET_LSIZE(bp, lsize);
1435 			BP_SET_TYPE(bp, zp->zp_type);
1436 			BP_SET_LEVEL(bp, zp->zp_level);
1437 			BP_SET_BIRTH(bp, zio->io_txg, 0);
1438 		}
1439 		zio->io_pipeline = ZIO_INTERLOCK_PIPELINE;
1440 	} else {
1441 		ASSERT(zp->zp_checksum != ZIO_CHECKSUM_GANG_HEADER);
1442 		BP_SET_LSIZE(bp, lsize);
1443 		BP_SET_TYPE(bp, zp->zp_type);
1444 		BP_SET_LEVEL(bp, zp->zp_level);
1445 		BP_SET_PSIZE(bp, psize);
1446 		BP_SET_COMPRESS(bp, compress);
1447 		BP_SET_CHECKSUM(bp, zp->zp_checksum);
1448 		BP_SET_DEDUP(bp, zp->zp_dedup);
1449 		BP_SET_BYTEORDER(bp, ZFS_HOST_BYTEORDER);
1450 		if (zp->zp_dedup) {
1451 			ASSERT(zio->io_child_type == ZIO_CHILD_LOGICAL);
1452 			ASSERT(!(zio->io_flags & ZIO_FLAG_IO_REWRITE));
1453 			zio->io_pipeline = ZIO_DDT_WRITE_PIPELINE;
1454 		}
1455 		if (zp->zp_nopwrite) {
1456 			ASSERT(zio->io_child_type == ZIO_CHILD_LOGICAL);
1457 			ASSERT(!(zio->io_flags & ZIO_FLAG_IO_REWRITE));
1458 			zio->io_pipeline |= ZIO_STAGE_NOP_WRITE;
1459 		}
1460 	}
1461 	return (ZIO_PIPELINE_CONTINUE);
1462 }
1463 
1464 static int
1465 zio_free_bp_init(zio_t *zio)
1466 {
1467 	blkptr_t *bp = zio->io_bp;
1468 
1469 	if (zio->io_child_type == ZIO_CHILD_LOGICAL) {
1470 		if (BP_GET_DEDUP(bp))
1471 			zio->io_pipeline = ZIO_DDT_FREE_PIPELINE;
1472 	}
1473 
1474 	ASSERT3P(zio->io_bp, ==, &zio->io_bp_copy);
1475 
1476 	return (ZIO_PIPELINE_CONTINUE);
1477 }
1478 
1479 /*
1480  * ==========================================================================
1481  * Execute the I/O pipeline
1482  * ==========================================================================
1483  */
1484 
1485 static void
1486 zio_taskq_dispatch(zio_t *zio, zio_taskq_type_t q, boolean_t cutinline)
1487 {
1488 	spa_t *spa = zio->io_spa;
1489 	zio_type_t t = zio->io_type;
1490 	int flags = (cutinline ? TQ_FRONT : 0);
1491 
1492 	/*
1493 	 * If we're a config writer or a probe, the normal issue and
1494 	 * interrupt threads may all be blocked waiting for the config lock.
1495 	 * In this case, select the otherwise-unused taskq for ZIO_TYPE_NULL.
1496 	 */
1497 	if (zio->io_flags & (ZIO_FLAG_CONFIG_WRITER | ZIO_FLAG_PROBE))
1498 		t = ZIO_TYPE_NULL;
1499 
1500 	/*
1501 	 * A similar issue exists for the L2ARC write thread until L2ARC 2.0.
1502 	 */
1503 	if (t == ZIO_TYPE_WRITE && zio->io_vd && zio->io_vd->vdev_aux)
1504 		t = ZIO_TYPE_NULL;
1505 
1506 	/*
1507 	 * If this is a high priority I/O, then use the high priority taskq if
1508 	 * available.
1509 	 */
1510 	if (zio->io_priority == ZIO_PRIORITY_NOW &&
1511 	    spa->spa_zio_taskq[t][q + 1].stqs_count != 0)
1512 		q++;
1513 
1514 	ASSERT3U(q, <, ZIO_TASKQ_TYPES);
1515 
1516 	/*
1517 	 * NB: We are assuming that the zio can only be dispatched
1518 	 * to a single taskq at a time.  It would be a grievous error
1519 	 * to dispatch the zio to another taskq at the same time.
1520 	 */
1521 	ASSERT(zio->io_tqent.tqent_next == NULL);
1522 	spa_taskq_dispatch_ent(spa, t, q, (task_func_t *)zio_execute, zio,
1523 	    flags, &zio->io_tqent);
1524 }
1525 
1526 static boolean_t
1527 zio_taskq_member(zio_t *zio, zio_taskq_type_t q)
1528 {
1529 	kthread_t *executor = zio->io_executor;
1530 	spa_t *spa = zio->io_spa;
1531 
1532 	for (zio_type_t t = 0; t < ZIO_TYPES; t++) {
1533 		spa_taskqs_t *tqs = &spa->spa_zio_taskq[t][q];
1534 		uint_t i;
1535 		for (i = 0; i < tqs->stqs_count; i++) {
1536 			if (taskq_member(tqs->stqs_taskq[i], executor))
1537 				return (B_TRUE);
1538 		}
1539 	}
1540 
1541 	return (B_FALSE);
1542 }
1543 
1544 static int
1545 zio_issue_async(zio_t *zio)
1546 {
1547 	zio_taskq_dispatch(zio, ZIO_TASKQ_ISSUE, B_FALSE);
1548 
1549 	return (ZIO_PIPELINE_STOP);
1550 }
1551 
1552 void
1553 zio_interrupt(zio_t *zio)
1554 {
1555 	zio_taskq_dispatch(zio, ZIO_TASKQ_INTERRUPT, B_FALSE);
1556 }
1557 
1558 void
1559 zio_delay_interrupt(zio_t *zio)
1560 {
1561 	/*
1562 	 * The timeout_generic() function isn't defined in userspace, so
1563 	 * rather than trying to implement the function, the zio delay
1564 	 * functionality has been disabled for userspace builds.
1565 	 */
1566 
1567 #ifdef _KERNEL
1568 	/*
1569 	 * If io_target_timestamp is zero, then no delay has been registered
1570 	 * for this IO, thus jump to the end of this function and "skip" the
1571 	 * delay; issuing it directly to the zio layer.
1572 	 */
1573 	if (zio->io_target_timestamp != 0) {
1574 		hrtime_t now = gethrtime();
1575 
1576 		if (now >= zio->io_target_timestamp) {
1577 			/*
1578 			 * This IO has already taken longer than the target
1579 			 * delay to complete, so we don't want to delay it
1580 			 * any longer; we "miss" the delay and issue it
1581 			 * directly to the zio layer. This is likely due to
1582 			 * the target latency being set to a value less than
1583 			 * the underlying hardware can satisfy (e.g. delay
1584 			 * set to 1ms, but the disks take 10ms to complete an
1585 			 * IO request).
1586 			 */
1587 
1588 			DTRACE_PROBE2(zio__delay__miss, zio_t *, zio,
1589 			    hrtime_t, now);
1590 
1591 			zio_interrupt(zio);
1592 		} else {
1593 			hrtime_t diff = zio->io_target_timestamp - now;
1594 
1595 			DTRACE_PROBE3(zio__delay__hit, zio_t *, zio,
1596 			    hrtime_t, now, hrtime_t, diff);
1597 
1598 			(void) timeout_generic(CALLOUT_NORMAL,
1599 			    (void (*)(void *))zio_interrupt, zio, diff, 1, 0);
1600 		}
1601 
1602 		return;
1603 	}
1604 #endif
1605 
1606 	DTRACE_PROBE1(zio__delay__skip, zio_t *, zio);
1607 	zio_interrupt(zio);
1608 }
1609 
1610 /*
1611  * Execute the I/O pipeline until one of the following occurs:
1612  *
1613  *	(1) the I/O completes
1614  *	(2) the pipeline stalls waiting for dependent child I/Os
1615  *	(3) the I/O issues, so we're waiting for an I/O completion interrupt
1616  *	(4) the I/O is delegated by vdev-level caching or aggregation
1617  *	(5) the I/O is deferred due to vdev-level queueing
1618  *	(6) the I/O is handed off to another thread.
1619  *
1620  * In all cases, the pipeline stops whenever there's no CPU work; it never
1621  * burns a thread in cv_wait().
1622  *
1623  * There's no locking on io_stage because there's no legitimate way
1624  * for multiple threads to be attempting to process the same I/O.
1625  */
1626 static zio_pipe_stage_t *zio_pipeline[];
1627 
1628 void
1629 zio_execute(zio_t *zio)
1630 {
1631 	zio->io_executor = curthread;
1632 
1633 	ASSERT3U(zio->io_queued_timestamp, >, 0);
1634 
1635 	while (zio->io_stage < ZIO_STAGE_DONE) {
1636 		enum zio_stage pipeline = zio->io_pipeline;
1637 		enum zio_stage stage = zio->io_stage;
1638 		int rv;
1639 
1640 		ASSERT(!MUTEX_HELD(&zio->io_lock));
1641 		ASSERT(ISP2(stage));
1642 		ASSERT(zio->io_stall == NULL);
1643 
1644 		do {
1645 			stage <<= 1;
1646 		} while ((stage & pipeline) == 0);
1647 
1648 		ASSERT(stage <= ZIO_STAGE_DONE);
1649 
1650 		/*
1651 		 * If we are in interrupt context and this pipeline stage
1652 		 * will grab a config lock that is held across I/O,
1653 		 * or may wait for an I/O that needs an interrupt thread
1654 		 * to complete, issue async to avoid deadlock.
1655 		 *
1656 		 * For VDEV_IO_START, we cut in line so that the io will
1657 		 * be sent to disk promptly.
1658 		 */
1659 		if ((stage & ZIO_BLOCKING_STAGES) && zio->io_vd == NULL &&
1660 		    zio_taskq_member(zio, ZIO_TASKQ_INTERRUPT)) {
1661 			boolean_t cut = (stage == ZIO_STAGE_VDEV_IO_START) ?
1662 			    zio_requeue_io_start_cut_in_line : B_FALSE;
1663 			zio_taskq_dispatch(zio, ZIO_TASKQ_ISSUE, cut);
1664 			return;
1665 		}
1666 
1667 		zio->io_stage = stage;
1668 		zio->io_pipeline_trace |= zio->io_stage;
1669 		rv = zio_pipeline[highbit64(stage) - 1](zio);
1670 
1671 		if (rv == ZIO_PIPELINE_STOP)
1672 			return;
1673 
1674 		ASSERT(rv == ZIO_PIPELINE_CONTINUE);
1675 	}
1676 }
1677 
1678 /*
1679  * ==========================================================================
1680  * Initiate I/O, either sync or async
1681  * ==========================================================================
1682  */
1683 int
1684 zio_wait(zio_t *zio)
1685 {
1686 	int error;
1687 
1688 	ASSERT3P(zio->io_stage, ==, ZIO_STAGE_OPEN);
1689 	ASSERT3P(zio->io_executor, ==, NULL);
1690 
1691 	zio->io_waiter = curthread;
1692 	ASSERT0(zio->io_queued_timestamp);
1693 	zio->io_queued_timestamp = gethrtime();
1694 
1695 	zio_execute(zio);
1696 
1697 	mutex_enter(&zio->io_lock);
1698 	while (zio->io_executor != NULL)
1699 		cv_wait(&zio->io_cv, &zio->io_lock);
1700 	mutex_exit(&zio->io_lock);
1701 
1702 	error = zio->io_error;
1703 	zio_destroy(zio);
1704 
1705 	return (error);
1706 }
1707 
1708 void
1709 zio_nowait(zio_t *zio)
1710 {
1711 	ASSERT3P(zio->io_executor, ==, NULL);
1712 
1713 	if (zio->io_child_type == ZIO_CHILD_LOGICAL &&
1714 	    zio_unique_parent(zio) == NULL) {
1715 		/*
1716 		 * This is a logical async I/O with no parent to wait for it.
1717 		 * We add it to the spa_async_root_zio "Godfather" I/O which
1718 		 * will ensure they complete prior to unloading the pool.
1719 		 */
1720 		spa_t *spa = zio->io_spa;
1721 
1722 		zio_add_child(spa->spa_async_zio_root[CPU_SEQID], zio);
1723 	}
1724 
1725 	ASSERT0(zio->io_queued_timestamp);
1726 	zio->io_queued_timestamp = gethrtime();
1727 	zio_execute(zio);
1728 }
1729 
1730 /*
1731  * ==========================================================================
1732  * Reexecute, cancel, or suspend/resume failed I/O
1733  * ==========================================================================
1734  */
1735 
1736 static void
1737 zio_reexecute(zio_t *pio)
1738 {
1739 	zio_t *cio, *cio_next;
1740 
1741 	ASSERT(pio->io_child_type == ZIO_CHILD_LOGICAL);
1742 	ASSERT(pio->io_orig_stage == ZIO_STAGE_OPEN);
1743 	ASSERT(pio->io_gang_leader == NULL);
1744 	ASSERT(pio->io_gang_tree == NULL);
1745 
1746 	pio->io_flags = pio->io_orig_flags;
1747 	pio->io_stage = pio->io_orig_stage;
1748 	pio->io_pipeline = pio->io_orig_pipeline;
1749 	pio->io_reexecute = 0;
1750 	pio->io_flags |= ZIO_FLAG_REEXECUTED;
1751 	pio->io_pipeline_trace = 0;
1752 	pio->io_error = 0;
1753 	for (int w = 0; w < ZIO_WAIT_TYPES; w++)
1754 		pio->io_state[w] = 0;
1755 	for (int c = 0; c < ZIO_CHILD_TYPES; c++)
1756 		pio->io_child_error[c] = 0;
1757 
1758 	if (IO_IS_ALLOCATING(pio))
1759 		BP_ZERO(pio->io_bp);
1760 
1761 	/*
1762 	 * As we reexecute pio's children, new children could be created.
1763 	 * New children go to the head of pio's io_child_list, however,
1764 	 * so we will (correctly) not reexecute them.  The key is that
1765 	 * the remainder of pio's io_child_list, from 'cio_next' onward,
1766 	 * cannot be affected by any side effects of reexecuting 'cio'.
1767 	 */
1768 	zio_link_t *zl = NULL;
1769 	for (cio = zio_walk_children(pio, &zl); cio != NULL; cio = cio_next) {
1770 		cio_next = zio_walk_children(pio, &zl);
1771 		mutex_enter(&pio->io_lock);
1772 		for (int w = 0; w < ZIO_WAIT_TYPES; w++)
1773 			pio->io_children[cio->io_child_type][w]++;
1774 		mutex_exit(&pio->io_lock);
1775 		zio_reexecute(cio);
1776 	}
1777 
1778 	/*
1779 	 * Now that all children have been reexecuted, execute the parent.
1780 	 * We don't reexecute "The Godfather" I/O here as it's the
1781 	 * responsibility of the caller to wait on it.
1782 	 */
1783 	if (!(pio->io_flags & ZIO_FLAG_GODFATHER)) {
1784 		pio->io_queued_timestamp = gethrtime();
1785 		zio_execute(pio);
1786 	}
1787 }
1788 
1789 void
1790 zio_suspend(spa_t *spa, zio_t *zio)
1791 {
1792 	if (spa_get_failmode(spa) == ZIO_FAILURE_MODE_PANIC)
1793 		fm_panic("Pool '%s' has encountered an uncorrectable I/O "
1794 		    "failure and the failure mode property for this pool "
1795 		    "is set to panic.", spa_name(spa));
1796 
1797 	zfs_ereport_post(FM_EREPORT_ZFS_IO_FAILURE, spa, NULL, NULL, 0, 0);
1798 
1799 	mutex_enter(&spa->spa_suspend_lock);
1800 
1801 	if (spa->spa_suspend_zio_root == NULL)
1802 		spa->spa_suspend_zio_root = zio_root(spa, NULL, NULL,
1803 		    ZIO_FLAG_CANFAIL | ZIO_FLAG_SPECULATIVE |
1804 		    ZIO_FLAG_GODFATHER);
1805 
1806 	spa->spa_suspended = B_TRUE;
1807 
1808 	if (zio != NULL) {
1809 		ASSERT(!(zio->io_flags & ZIO_FLAG_GODFATHER));
1810 		ASSERT(zio != spa->spa_suspend_zio_root);
1811 		ASSERT(zio->io_child_type == ZIO_CHILD_LOGICAL);
1812 		ASSERT(zio_unique_parent(zio) == NULL);
1813 		ASSERT(zio->io_stage == ZIO_STAGE_DONE);
1814 		zio_add_child(spa->spa_suspend_zio_root, zio);
1815 	}
1816 
1817 	mutex_exit(&spa->spa_suspend_lock);
1818 }
1819 
1820 int
1821 zio_resume(spa_t *spa)
1822 {
1823 	zio_t *pio;
1824 
1825 	/*
1826 	 * Reexecute all previously suspended i/o.
1827 	 */
1828 	mutex_enter(&spa->spa_suspend_lock);
1829 	spa->spa_suspended = B_FALSE;
1830 	cv_broadcast(&spa->spa_suspend_cv);
1831 	pio = spa->spa_suspend_zio_root;
1832 	spa->spa_suspend_zio_root = NULL;
1833 	mutex_exit(&spa->spa_suspend_lock);
1834 
1835 	if (pio == NULL)
1836 		return (0);
1837 
1838 	zio_reexecute(pio);
1839 	return (zio_wait(pio));
1840 }
1841 
1842 void
1843 zio_resume_wait(spa_t *spa)
1844 {
1845 	mutex_enter(&spa->spa_suspend_lock);
1846 	while (spa_suspended(spa))
1847 		cv_wait(&spa->spa_suspend_cv, &spa->spa_suspend_lock);
1848 	mutex_exit(&spa->spa_suspend_lock);
1849 }
1850 
1851 /*
1852  * ==========================================================================
1853  * Gang blocks.
1854  *
1855  * A gang block is a collection of small blocks that looks to the DMU
1856  * like one large block.  When zio_dva_allocate() cannot find a block
1857  * of the requested size, due to either severe fragmentation or the pool
1858  * being nearly full, it calls zio_write_gang_block() to construct the
1859  * block from smaller fragments.
1860  *
1861  * A gang block consists of a gang header (zio_gbh_phys_t) and up to
1862  * three (SPA_GBH_NBLKPTRS) gang members.  The gang header is just like
1863  * an indirect block: it's an array of block pointers.  It consumes
1864  * only one sector and hence is allocatable regardless of fragmentation.
1865  * The gang header's bps point to its gang members, which hold the data.
1866  *
1867  * Gang blocks are self-checksumming, using the bp's <vdev, offset, txg>
1868  * as the verifier to ensure uniqueness of the SHA256 checksum.
1869  * Critically, the gang block bp's blk_cksum is the checksum of the data,
1870  * not the gang header.  This ensures that data block signatures (needed for
1871  * deduplication) are independent of how the block is physically stored.
1872  *
1873  * Gang blocks can be nested: a gang member may itself be a gang block.
1874  * Thus every gang block is a tree in which root and all interior nodes are
1875  * gang headers, and the leaves are normal blocks that contain user data.
1876  * The root of the gang tree is called the gang leader.
1877  *
1878  * To perform any operation (read, rewrite, free, claim) on a gang block,
1879  * zio_gang_assemble() first assembles the gang tree (minus data leaves)
1880  * in the io_gang_tree field of the original logical i/o by recursively
1881  * reading the gang leader and all gang headers below it.  This yields
1882  * an in-core tree containing the contents of every gang header and the
1883  * bps for every constituent of the gang block.
1884  *
1885  * With the gang tree now assembled, zio_gang_issue() just walks the gang tree
1886  * and invokes a callback on each bp.  To free a gang block, zio_gang_issue()
1887  * calls zio_free_gang() -- a trivial wrapper around zio_free() -- for each bp.
1888  * zio_claim_gang() provides a similarly trivial wrapper for zio_claim().
1889  * zio_read_gang() is a wrapper around zio_read() that omits reading gang
1890  * headers, since we already have those in io_gang_tree.  zio_rewrite_gang()
1891  * performs a zio_rewrite() of the data or, for gang headers, a zio_rewrite()
1892  * of the gang header plus zio_checksum_compute() of the data to update the
1893  * gang header's blk_cksum as described above.
1894  *
1895  * The two-phase assemble/issue model solves the problem of partial failure --
1896  * what if you'd freed part of a gang block but then couldn't read the
1897  * gang header for another part?  Assembling the entire gang tree first
1898  * ensures that all the necessary gang header I/O has succeeded before
1899  * starting the actual work of free, claim, or write.  Once the gang tree
1900  * is assembled, free and claim are in-memory operations that cannot fail.
1901  *
1902  * In the event that a gang write fails, zio_dva_unallocate() walks the
1903  * gang tree to immediately free (i.e. insert back into the space map)
1904  * everything we've allocated.  This ensures that we don't get ENOSPC
1905  * errors during repeated suspend/resume cycles due to a flaky device.
1906  *
1907  * Gang rewrites only happen during sync-to-convergence.  If we can't assemble
1908  * the gang tree, we won't modify the block, so we can safely defer the free
1909  * (knowing that the block is still intact).  If we *can* assemble the gang
1910  * tree, then even if some of the rewrites fail, zio_dva_unallocate() will free
1911  * each constituent bp and we can allocate a new block on the next sync pass.
1912  *
1913  * In all cases, the gang tree allows complete recovery from partial failure.
1914  * ==========================================================================
1915  */
1916 
1917 static void
1918 zio_gang_issue_func_done(zio_t *zio)
1919 {
1920 	abd_put(zio->io_abd);
1921 }
1922 
1923 static zio_t *
1924 zio_read_gang(zio_t *pio, blkptr_t *bp, zio_gang_node_t *gn, abd_t *data,
1925     uint64_t offset)
1926 {
1927 	if (gn != NULL)
1928 		return (pio);
1929 
1930 	return (zio_read(pio, pio->io_spa, bp, abd_get_offset(data, offset),
1931 	    BP_GET_PSIZE(bp), zio_gang_issue_func_done,
1932 	    NULL, pio->io_priority, ZIO_GANG_CHILD_FLAGS(pio),
1933 	    &pio->io_bookmark));
1934 }
1935 
1936 static zio_t *
1937 zio_rewrite_gang(zio_t *pio, blkptr_t *bp, zio_gang_node_t *gn, abd_t *data,
1938     uint64_t offset)
1939 {
1940 	zio_t *zio;
1941 
1942 	if (gn != NULL) {
1943 		abd_t *gbh_abd =
1944 		    abd_get_from_buf(gn->gn_gbh, SPA_GANGBLOCKSIZE);
1945 		zio = zio_rewrite(pio, pio->io_spa, pio->io_txg, bp,
1946 		    gbh_abd, SPA_GANGBLOCKSIZE, zio_gang_issue_func_done, NULL,
1947 		    pio->io_priority, ZIO_GANG_CHILD_FLAGS(pio),
1948 		    &pio->io_bookmark);
1949 		/*
1950 		 * As we rewrite each gang header, the pipeline will compute
1951 		 * a new gang block header checksum for it; but no one will
1952 		 * compute a new data checksum, so we do that here.  The one
1953 		 * exception is the gang leader: the pipeline already computed
1954 		 * its data checksum because that stage precedes gang assembly.
1955 		 * (Presently, nothing actually uses interior data checksums;
1956 		 * this is just good hygiene.)
1957 		 */
1958 		if (gn != pio->io_gang_leader->io_gang_tree) {
1959 			abd_t *buf = abd_get_offset(data, offset);
1960 
1961 			zio_checksum_compute(zio, BP_GET_CHECKSUM(bp),
1962 			    buf, BP_GET_PSIZE(bp));
1963 
1964 			abd_put(buf);
1965 		}
1966 		/*
1967 		 * If we are here to damage data for testing purposes,
1968 		 * leave the GBH alone so that we can detect the damage.
1969 		 */
1970 		if (pio->io_gang_leader->io_flags & ZIO_FLAG_INDUCE_DAMAGE)
1971 			zio->io_pipeline &= ~ZIO_VDEV_IO_STAGES;
1972 	} else {
1973 		zio = zio_rewrite(pio, pio->io_spa, pio->io_txg, bp,
1974 		    abd_get_offset(data, offset), BP_GET_PSIZE(bp),
1975 		    zio_gang_issue_func_done, NULL, pio->io_priority,
1976 		    ZIO_GANG_CHILD_FLAGS(pio), &pio->io_bookmark);
1977 	}
1978 
1979 	return (zio);
1980 }
1981 
1982 /* ARGSUSED */
1983 static zio_t *
1984 zio_free_gang(zio_t *pio, blkptr_t *bp, zio_gang_node_t *gn, abd_t *data,
1985     uint64_t offset)
1986 {
1987 	return (zio_free_sync(pio, pio->io_spa, pio->io_txg, bp,
1988 	    ZIO_GANG_CHILD_FLAGS(pio)));
1989 }
1990 
1991 /* ARGSUSED */
1992 static zio_t *
1993 zio_claim_gang(zio_t *pio, blkptr_t *bp, zio_gang_node_t *gn, abd_t *data,
1994     uint64_t offset)
1995 {
1996 	return (zio_claim(pio, pio->io_spa, pio->io_txg, bp,
1997 	    NULL, NULL, ZIO_GANG_CHILD_FLAGS(pio)));
1998 }
1999 
2000 static zio_gang_issue_func_t *zio_gang_issue_func[ZIO_TYPES] = {
2001 	NULL,
2002 	zio_read_gang,
2003 	zio_rewrite_gang,
2004 	zio_free_gang,
2005 	zio_claim_gang,
2006 	NULL
2007 };
2008 
2009 static void zio_gang_tree_assemble_done(zio_t *zio);
2010 
2011 static zio_gang_node_t *
2012 zio_gang_node_alloc(zio_gang_node_t **gnpp)
2013 {
2014 	zio_gang_node_t *gn;
2015 
2016 	ASSERT(*gnpp == NULL);
2017 
2018 	gn = kmem_zalloc(sizeof (*gn), KM_SLEEP);
2019 	gn->gn_gbh = zio_buf_alloc(SPA_GANGBLOCKSIZE);
2020 	*gnpp = gn;
2021 
2022 	return (gn);
2023 }
2024 
2025 static void
2026 zio_gang_node_free(zio_gang_node_t **gnpp)
2027 {
2028 	zio_gang_node_t *gn = *gnpp;
2029 
2030 	for (int g = 0; g < SPA_GBH_NBLKPTRS; g++)
2031 		ASSERT(gn->gn_child[g] == NULL);
2032 
2033 	zio_buf_free(gn->gn_gbh, SPA_GANGBLOCKSIZE);
2034 	kmem_free(gn, sizeof (*gn));
2035 	*gnpp = NULL;
2036 }
2037 
2038 static void
2039 zio_gang_tree_free(zio_gang_node_t **gnpp)
2040 {
2041 	zio_gang_node_t *gn = *gnpp;
2042 
2043 	if (gn == NULL)
2044 		return;
2045 
2046 	for (int g = 0; g < SPA_GBH_NBLKPTRS; g++)
2047 		zio_gang_tree_free(&gn->gn_child[g]);
2048 
2049 	zio_gang_node_free(gnpp);
2050 }
2051 
2052 static void
2053 zio_gang_tree_assemble(zio_t *gio, blkptr_t *bp, zio_gang_node_t **gnpp)
2054 {
2055 	zio_gang_node_t *gn = zio_gang_node_alloc(gnpp);
2056 	abd_t *gbh_abd = abd_get_from_buf(gn->gn_gbh, SPA_GANGBLOCKSIZE);
2057 
2058 	ASSERT(gio->io_gang_leader == gio);
2059 	ASSERT(BP_IS_GANG(bp));
2060 
2061 	zio_nowait(zio_read(gio, gio->io_spa, bp, gbh_abd, SPA_GANGBLOCKSIZE,
2062 	    zio_gang_tree_assemble_done, gn, gio->io_priority,
2063 	    ZIO_GANG_CHILD_FLAGS(gio), &gio->io_bookmark));
2064 }
2065 
2066 static void
2067 zio_gang_tree_assemble_done(zio_t *zio)
2068 {
2069 	zio_t *gio = zio->io_gang_leader;
2070 	zio_gang_node_t *gn = zio->io_private;
2071 	blkptr_t *bp = zio->io_bp;
2072 
2073 	ASSERT(gio == zio_unique_parent(zio));
2074 	ASSERT(zio->io_child_count == 0);
2075 
2076 	if (zio->io_error)
2077 		return;
2078 
2079 	/* this ABD was created from a linear buf in zio_gang_tree_assemble */
2080 	if (BP_SHOULD_BYTESWAP(bp))
2081 		byteswap_uint64_array(abd_to_buf(zio->io_abd), zio->io_size);
2082 
2083 	ASSERT3P(abd_to_buf(zio->io_abd), ==, gn->gn_gbh);
2084 	ASSERT(zio->io_size == SPA_GANGBLOCKSIZE);
2085 	ASSERT(gn->gn_gbh->zg_tail.zec_magic == ZEC_MAGIC);
2086 
2087 	abd_put(zio->io_abd);
2088 
2089 	for (int g = 0; g < SPA_GBH_NBLKPTRS; g++) {
2090 		blkptr_t *gbp = &gn->gn_gbh->zg_blkptr[g];
2091 		if (!BP_IS_GANG(gbp))
2092 			continue;
2093 		zio_gang_tree_assemble(gio, gbp, &gn->gn_child[g]);
2094 	}
2095 }
2096 
2097 static void
2098 zio_gang_tree_issue(zio_t *pio, zio_gang_node_t *gn, blkptr_t *bp, abd_t *data,
2099     uint64_t offset)
2100 {
2101 	zio_t *gio = pio->io_gang_leader;
2102 	zio_t *zio;
2103 
2104 	ASSERT(BP_IS_GANG(bp) == !!gn);
2105 	ASSERT(BP_GET_CHECKSUM(bp) == BP_GET_CHECKSUM(gio->io_bp));
2106 	ASSERT(BP_GET_LSIZE(bp) == BP_GET_PSIZE(bp) || gn == gio->io_gang_tree);
2107 
2108 	/*
2109 	 * If you're a gang header, your data is in gn->gn_gbh.
2110 	 * If you're a gang member, your data is in 'data' and gn == NULL.
2111 	 */
2112 	zio = zio_gang_issue_func[gio->io_type](pio, bp, gn, data, offset);
2113 
2114 	if (gn != NULL) {
2115 		ASSERT(gn->gn_gbh->zg_tail.zec_magic == ZEC_MAGIC);
2116 
2117 		for (int g = 0; g < SPA_GBH_NBLKPTRS; g++) {
2118 			blkptr_t *gbp = &gn->gn_gbh->zg_blkptr[g];
2119 			if (BP_IS_HOLE(gbp))
2120 				continue;
2121 			zio_gang_tree_issue(zio, gn->gn_child[g], gbp, data,
2122 			    offset);
2123 			offset += BP_GET_PSIZE(gbp);
2124 		}
2125 	}
2126 
2127 	if (gn == gio->io_gang_tree)
2128 		ASSERT3U(gio->io_size, ==, offset);
2129 
2130 	if (zio != pio)
2131 		zio_nowait(zio);
2132 }
2133 
2134 static int
2135 zio_gang_assemble(zio_t *zio)
2136 {
2137 	blkptr_t *bp = zio->io_bp;
2138 
2139 	ASSERT(BP_IS_GANG(bp) && zio->io_gang_leader == NULL);
2140 	ASSERT(zio->io_child_type > ZIO_CHILD_GANG);
2141 
2142 	zio->io_gang_leader = zio;
2143 
2144 	zio_gang_tree_assemble(zio, bp, &zio->io_gang_tree);
2145 
2146 	return (ZIO_PIPELINE_CONTINUE);
2147 }
2148 
2149 static int
2150 zio_gang_issue(zio_t *zio)
2151 {
2152 	blkptr_t *bp = zio->io_bp;
2153 
2154 	if (zio_wait_for_children(zio, ZIO_CHILD_GANG_BIT, ZIO_WAIT_DONE)) {
2155 		return (ZIO_PIPELINE_STOP);
2156 	}
2157 
2158 	ASSERT(BP_IS_GANG(bp) && zio->io_gang_leader == zio);
2159 	ASSERT(zio->io_child_type > ZIO_CHILD_GANG);
2160 
2161 	if (zio->io_child_error[ZIO_CHILD_GANG] == 0)
2162 		zio_gang_tree_issue(zio, zio->io_gang_tree, bp, zio->io_abd,
2163 		    0);
2164 	else
2165 		zio_gang_tree_free(&zio->io_gang_tree);
2166 
2167 	zio->io_pipeline = ZIO_INTERLOCK_PIPELINE;
2168 
2169 	return (ZIO_PIPELINE_CONTINUE);
2170 }
2171 
2172 static void
2173 zio_write_gang_member_ready(zio_t *zio)
2174 {
2175 	zio_t *pio = zio_unique_parent(zio);
2176 	zio_t *gio = zio->io_gang_leader;
2177 	dva_t *cdva = zio->io_bp->blk_dva;
2178 	dva_t *pdva = pio->io_bp->blk_dva;
2179 	uint64_t asize;
2180 
2181 	if (BP_IS_HOLE(zio->io_bp))
2182 		return;
2183 
2184 	ASSERT(BP_IS_HOLE(&zio->io_bp_orig));
2185 
2186 	ASSERT(zio->io_child_type == ZIO_CHILD_GANG);
2187 	ASSERT3U(zio->io_prop.zp_copies, ==, gio->io_prop.zp_copies);
2188 	ASSERT3U(zio->io_prop.zp_copies, <=, BP_GET_NDVAS(zio->io_bp));
2189 	ASSERT3U(pio->io_prop.zp_copies, <=, BP_GET_NDVAS(pio->io_bp));
2190 	ASSERT3U(BP_GET_NDVAS(zio->io_bp), <=, BP_GET_NDVAS(pio->io_bp));
2191 
2192 	mutex_enter(&pio->io_lock);
2193 	for (int d = 0; d < BP_GET_NDVAS(zio->io_bp); d++) {
2194 		ASSERT(DVA_GET_GANG(&pdva[d]));
2195 		asize = DVA_GET_ASIZE(&pdva[d]);
2196 		asize += DVA_GET_ASIZE(&cdva[d]);
2197 		DVA_SET_ASIZE(&pdva[d], asize);
2198 	}
2199 	mutex_exit(&pio->io_lock);
2200 }
2201 
2202 static void
2203 zio_write_gang_done(zio_t *zio)
2204 {
2205 	abd_put(zio->io_abd);
2206 }
2207 
2208 static int
2209 zio_write_gang_block(zio_t *pio)
2210 {
2211 	spa_t *spa = pio->io_spa;
2212 	metaslab_class_t *mc = spa_normal_class(spa);
2213 	blkptr_t *bp = pio->io_bp;
2214 	zio_t *gio = pio->io_gang_leader;
2215 	zio_t *zio;
2216 	zio_gang_node_t *gn, **gnpp;
2217 	zio_gbh_phys_t *gbh;
2218 	abd_t *gbh_abd;
2219 	uint64_t txg = pio->io_txg;
2220 	uint64_t resid = pio->io_size;
2221 	uint64_t lsize;
2222 	int copies = gio->io_prop.zp_copies;
2223 	int gbh_copies = MIN(copies + 1, spa_max_replication(spa));
2224 	zio_prop_t zp;
2225 	int error;
2226 
2227 	int flags = METASLAB_HINTBP_FAVOR | METASLAB_GANG_HEADER;
2228 	if (pio->io_flags & ZIO_FLAG_IO_ALLOCATING) {
2229 		ASSERT(pio->io_priority == ZIO_PRIORITY_ASYNC_WRITE);
2230 		ASSERT(!(pio->io_flags & ZIO_FLAG_NODATA));
2231 
2232 		flags |= METASLAB_ASYNC_ALLOC;
2233 		VERIFY(refcount_held(&mc->mc_alloc_slots, pio));
2234 
2235 		/*
2236 		 * The logical zio has already placed a reservation for
2237 		 * 'copies' allocation slots but gang blocks may require
2238 		 * additional copies. These additional copies
2239 		 * (i.e. gbh_copies - copies) are guaranteed to succeed
2240 		 * since metaslab_class_throttle_reserve() always allows
2241 		 * additional reservations for gang blocks.
2242 		 */
2243 		VERIFY(metaslab_class_throttle_reserve(mc, gbh_copies - copies,
2244 		    pio, flags));
2245 	}
2246 
2247 	error = metaslab_alloc(spa, mc, SPA_GANGBLOCKSIZE,
2248 	    bp, gbh_copies, txg, pio == gio ? NULL : gio->io_bp, flags,
2249 	    &pio->io_alloc_list, pio);
2250 	if (error) {
2251 		if (pio->io_flags & ZIO_FLAG_IO_ALLOCATING) {
2252 			ASSERT(pio->io_priority == ZIO_PRIORITY_ASYNC_WRITE);
2253 			ASSERT(!(pio->io_flags & ZIO_FLAG_NODATA));
2254 
2255 			/*
2256 			 * If we failed to allocate the gang block header then
2257 			 * we remove any additional allocation reservations that
2258 			 * we placed here. The original reservation will
2259 			 * be removed when the logical I/O goes to the ready
2260 			 * stage.
2261 			 */
2262 			metaslab_class_throttle_unreserve(mc,
2263 			    gbh_copies - copies, pio);
2264 		}
2265 		pio->io_error = error;
2266 		return (ZIO_PIPELINE_CONTINUE);
2267 	}
2268 
2269 	if (pio == gio) {
2270 		gnpp = &gio->io_gang_tree;
2271 	} else {
2272 		gnpp = pio->io_private;
2273 		ASSERT(pio->io_ready == zio_write_gang_member_ready);
2274 	}
2275 
2276 	gn = zio_gang_node_alloc(gnpp);
2277 	gbh = gn->gn_gbh;
2278 	bzero(gbh, SPA_GANGBLOCKSIZE);
2279 	gbh_abd = abd_get_from_buf(gbh, SPA_GANGBLOCKSIZE);
2280 
2281 	/*
2282 	 * Create the gang header.
2283 	 */
2284 	zio = zio_rewrite(pio, spa, txg, bp, gbh_abd, SPA_GANGBLOCKSIZE,
2285 	    zio_write_gang_done, NULL, pio->io_priority,
2286 	    ZIO_GANG_CHILD_FLAGS(pio), &pio->io_bookmark);
2287 
2288 	/*
2289 	 * Create and nowait the gang children.
2290 	 */
2291 	for (int g = 0; resid != 0; resid -= lsize, g++) {
2292 		lsize = P2ROUNDUP(resid / (SPA_GBH_NBLKPTRS - g),
2293 		    SPA_MINBLOCKSIZE);
2294 		ASSERT(lsize >= SPA_MINBLOCKSIZE && lsize <= resid);
2295 
2296 		zp.zp_checksum = gio->io_prop.zp_checksum;
2297 		zp.zp_compress = ZIO_COMPRESS_OFF;
2298 		zp.zp_type = DMU_OT_NONE;
2299 		zp.zp_level = 0;
2300 		zp.zp_copies = gio->io_prop.zp_copies;
2301 		zp.zp_dedup = B_FALSE;
2302 		zp.zp_dedup_verify = B_FALSE;
2303 		zp.zp_nopwrite = B_FALSE;
2304 
2305 		zio_t *cio = zio_write(zio, spa, txg, &gbh->zg_blkptr[g],
2306 		    abd_get_offset(pio->io_abd, pio->io_size - resid), lsize,
2307 		    lsize, &zp, zio_write_gang_member_ready, NULL, NULL,
2308 		    zio_write_gang_done, &gn->gn_child[g], pio->io_priority,
2309 		    ZIO_GANG_CHILD_FLAGS(pio), &pio->io_bookmark);
2310 
2311 		if (pio->io_flags & ZIO_FLAG_IO_ALLOCATING) {
2312 			ASSERT(pio->io_priority == ZIO_PRIORITY_ASYNC_WRITE);
2313 			ASSERT(!(pio->io_flags & ZIO_FLAG_NODATA));
2314 
2315 			/*
2316 			 * Gang children won't throttle but we should
2317 			 * account for their work, so reserve an allocation
2318 			 * slot for them here.
2319 			 */
2320 			VERIFY(metaslab_class_throttle_reserve(mc,
2321 			    zp.zp_copies, cio, flags));
2322 		}
2323 		zio_nowait(cio);
2324 	}
2325 
2326 	/*
2327 	 * Set pio's pipeline to just wait for zio to finish.
2328 	 */
2329 	pio->io_pipeline = ZIO_INTERLOCK_PIPELINE;
2330 
2331 	zio_nowait(zio);
2332 
2333 	return (ZIO_PIPELINE_CONTINUE);
2334 }
2335 
2336 /*
2337  * The zio_nop_write stage in the pipeline determines if allocating a
2338  * new bp is necessary.  The nopwrite feature can handle writes in
2339  * either syncing or open context (i.e. zil writes) and as a result is
2340  * mutually exclusive with dedup.
2341  *
2342  * By leveraging a cryptographically secure checksum, such as SHA256, we
2343  * can compare the checksums of the new data and the old to determine if
2344  * allocating a new block is required.  Note that our requirements for
2345  * cryptographic strength are fairly weak: there can't be any accidental
2346  * hash collisions, but we don't need to be secure against intentional
2347  * (malicious) collisions.  To trigger a nopwrite, you have to be able
2348  * to write the file to begin with, and triggering an incorrect (hash
2349  * collision) nopwrite is no worse than simply writing to the file.
2350  * That said, there are no known attacks against the checksum algorithms
2351  * used for nopwrite, assuming that the salt and the checksums
2352  * themselves remain secret.
2353  */
2354 static int
2355 zio_nop_write(zio_t *zio)
2356 {
2357 	blkptr_t *bp = zio->io_bp;
2358 	blkptr_t *bp_orig = &zio->io_bp_orig;
2359 	zio_prop_t *zp = &zio->io_prop;
2360 
2361 	ASSERT(BP_GET_LEVEL(bp) == 0);
2362 	ASSERT(!(zio->io_flags & ZIO_FLAG_IO_REWRITE));
2363 	ASSERT(zp->zp_nopwrite);
2364 	ASSERT(!zp->zp_dedup);
2365 	ASSERT(zio->io_bp_override == NULL);
2366 	ASSERT(IO_IS_ALLOCATING(zio));
2367 
2368 	/*
2369 	 * Check to see if the original bp and the new bp have matching
2370 	 * characteristics (i.e. same checksum, compression algorithms, etc).
2371 	 * If they don't then just continue with the pipeline which will
2372 	 * allocate a new bp.
2373 	 */
2374 	if (BP_IS_HOLE(bp_orig) ||
2375 	    !(zio_checksum_table[BP_GET_CHECKSUM(bp)].ci_flags &
2376 	    ZCHECKSUM_FLAG_NOPWRITE) ||
2377 	    BP_GET_CHECKSUM(bp) != BP_GET_CHECKSUM(bp_orig) ||
2378 	    BP_GET_COMPRESS(bp) != BP_GET_COMPRESS(bp_orig) ||
2379 	    BP_GET_DEDUP(bp) != BP_GET_DEDUP(bp_orig) ||
2380 	    zp->zp_copies != BP_GET_NDVAS(bp_orig))
2381 		return (ZIO_PIPELINE_CONTINUE);
2382 
2383 	/*
2384 	 * If the checksums match then reset the pipeline so that we
2385 	 * avoid allocating a new bp and issuing any I/O.
2386 	 */
2387 	if (ZIO_CHECKSUM_EQUAL(bp->blk_cksum, bp_orig->blk_cksum)) {
2388 		ASSERT(zio_checksum_table[zp->zp_checksum].ci_flags &
2389 		    ZCHECKSUM_FLAG_NOPWRITE);
2390 		ASSERT3U(BP_GET_PSIZE(bp), ==, BP_GET_PSIZE(bp_orig));
2391 		ASSERT3U(BP_GET_LSIZE(bp), ==, BP_GET_LSIZE(bp_orig));
2392 		ASSERT(zp->zp_compress != ZIO_COMPRESS_OFF);
2393 		ASSERT(bcmp(&bp->blk_prop, &bp_orig->blk_prop,
2394 		    sizeof (uint64_t)) == 0);
2395 
2396 		*bp = *bp_orig;
2397 		zio->io_pipeline = ZIO_INTERLOCK_PIPELINE;
2398 		zio->io_flags |= ZIO_FLAG_NOPWRITE;
2399 	}
2400 
2401 	return (ZIO_PIPELINE_CONTINUE);
2402 }
2403 
2404 /*
2405  * ==========================================================================
2406  * Dedup
2407  * ==========================================================================
2408  */
2409 static void
2410 zio_ddt_child_read_done(zio_t *zio)
2411 {
2412 	blkptr_t *bp = zio->io_bp;
2413 	ddt_entry_t *dde = zio->io_private;
2414 	ddt_phys_t *ddp;
2415 	zio_t *pio = zio_unique_parent(zio);
2416 
2417 	mutex_enter(&pio->io_lock);
2418 	ddp = ddt_phys_select(dde, bp);
2419 	if (zio->io_error == 0)
2420 		ddt_phys_clear(ddp);	/* this ddp doesn't need repair */
2421 
2422 	if (zio->io_error == 0 && dde->dde_repair_abd == NULL)
2423 		dde->dde_repair_abd = zio->io_abd;
2424 	else
2425 		abd_free(zio->io_abd);
2426 	mutex_exit(&pio->io_lock);
2427 }
2428 
2429 static int
2430 zio_ddt_read_start(zio_t *zio)
2431 {
2432 	blkptr_t *bp = zio->io_bp;
2433 
2434 	ASSERT(BP_GET_DEDUP(bp));
2435 	ASSERT(BP_GET_PSIZE(bp) == zio->io_size);
2436 	ASSERT(zio->io_child_type == ZIO_CHILD_LOGICAL);
2437 
2438 	if (zio->io_child_error[ZIO_CHILD_DDT]) {
2439 		ddt_t *ddt = ddt_select(zio->io_spa, bp);
2440 		ddt_entry_t *dde = ddt_repair_start(ddt, bp);
2441 		ddt_phys_t *ddp = dde->dde_phys;
2442 		ddt_phys_t *ddp_self = ddt_phys_select(dde, bp);
2443 		blkptr_t blk;
2444 
2445 		ASSERT(zio->io_vsd == NULL);
2446 		zio->io_vsd = dde;
2447 
2448 		if (ddp_self == NULL)
2449 			return (ZIO_PIPELINE_CONTINUE);
2450 
2451 		for (int p = 0; p < DDT_PHYS_TYPES; p++, ddp++) {
2452 			if (ddp->ddp_phys_birth == 0 || ddp == ddp_self)
2453 				continue;
2454 			ddt_bp_create(ddt->ddt_checksum, &dde->dde_key, ddp,
2455 			    &blk);
2456 			zio_nowait(zio_read(zio, zio->io_spa, &blk,
2457 			    abd_alloc_for_io(zio->io_size, B_TRUE),
2458 			    zio->io_size, zio_ddt_child_read_done, dde,
2459 			    zio->io_priority, ZIO_DDT_CHILD_FLAGS(zio) |
2460 			    ZIO_FLAG_DONT_PROPAGATE, &zio->io_bookmark));
2461 		}
2462 		return (ZIO_PIPELINE_CONTINUE);
2463 	}
2464 
2465 	zio_nowait(zio_read(zio, zio->io_spa, bp,
2466 	    zio->io_abd, zio->io_size, NULL, NULL, zio->io_priority,
2467 	    ZIO_DDT_CHILD_FLAGS(zio), &zio->io_bookmark));
2468 
2469 	return (ZIO_PIPELINE_CONTINUE);
2470 }
2471 
2472 static int
2473 zio_ddt_read_done(zio_t *zio)
2474 {
2475 	blkptr_t *bp = zio->io_bp;
2476 
2477 	if (zio_wait_for_children(zio, ZIO_CHILD_DDT_BIT, ZIO_WAIT_DONE)) {
2478 		return (ZIO_PIPELINE_STOP);
2479 	}
2480 
2481 	ASSERT(BP_GET_DEDUP(bp));
2482 	ASSERT(BP_GET_PSIZE(bp) == zio->io_size);
2483 	ASSERT(zio->io_child_type == ZIO_CHILD_LOGICAL);
2484 
2485 	if (zio->io_child_error[ZIO_CHILD_DDT]) {
2486 		ddt_t *ddt = ddt_select(zio->io_spa, bp);
2487 		ddt_entry_t *dde = zio->io_vsd;
2488 		if (ddt == NULL) {
2489 			ASSERT(spa_load_state(zio->io_spa) != SPA_LOAD_NONE);
2490 			return (ZIO_PIPELINE_CONTINUE);
2491 		}
2492 		if (dde == NULL) {
2493 			zio->io_stage = ZIO_STAGE_DDT_READ_START >> 1;
2494 			zio_taskq_dispatch(zio, ZIO_TASKQ_ISSUE, B_FALSE);
2495 			return (ZIO_PIPELINE_STOP);
2496 		}
2497 		if (dde->dde_repair_abd != NULL) {
2498 			abd_copy(zio->io_abd, dde->dde_repair_abd,
2499 			    zio->io_size);
2500 			zio->io_child_error[ZIO_CHILD_DDT] = 0;
2501 		}
2502 		ddt_repair_done(ddt, dde);
2503 		zio->io_vsd = NULL;
2504 	}
2505 
2506 	ASSERT(zio->io_vsd == NULL);
2507 
2508 	return (ZIO_PIPELINE_CONTINUE);
2509 }
2510 
2511 static boolean_t
2512 zio_ddt_collision(zio_t *zio, ddt_t *ddt, ddt_entry_t *dde)
2513 {
2514 	spa_t *spa = zio->io_spa;
2515 	boolean_t do_raw = (zio->io_flags & ZIO_FLAG_RAW);
2516 
2517 	/* We should never get a raw, override zio */
2518 	ASSERT(!(zio->io_bp_override && do_raw));
2519 
2520 	/*
2521 	 * Note: we compare the original data, not the transformed data,
2522 	 * because when zio->io_bp is an override bp, we will not have
2523 	 * pushed the I/O transforms.  That's an important optimization
2524 	 * because otherwise we'd compress/encrypt all dmu_sync() data twice.
2525 	 */
2526 	for (int p = DDT_PHYS_SINGLE; p <= DDT_PHYS_TRIPLE; p++) {
2527 		zio_t *lio = dde->dde_lead_zio[p];
2528 
2529 		if (lio != NULL) {
2530 			return (lio->io_orig_size != zio->io_orig_size ||
2531 			    abd_cmp(zio->io_orig_abd, lio->io_orig_abd,
2532 			    zio->io_orig_size) != 0);
2533 		}
2534 	}
2535 
2536 	for (int p = DDT_PHYS_SINGLE; p <= DDT_PHYS_TRIPLE; p++) {
2537 		ddt_phys_t *ddp = &dde->dde_phys[p];
2538 
2539 		if (ddp->ddp_phys_birth != 0) {
2540 			arc_buf_t *abuf = NULL;
2541 			arc_flags_t aflags = ARC_FLAG_WAIT;
2542 			int zio_flags = ZIO_FLAG_CANFAIL | ZIO_FLAG_SPECULATIVE;
2543 			blkptr_t blk = *zio->io_bp;
2544 			int error;
2545 
2546 			ddt_bp_fill(ddp, &blk, ddp->ddp_phys_birth);
2547 
2548 			ddt_exit(ddt);
2549 
2550 			/*
2551 			 * Intuitively, it would make more sense to compare
2552 			 * io_abd than io_orig_abd in the raw case since you
2553 			 * don't want to look at any transformations that have
2554 			 * happened to the data. However, for raw I/Os the
2555 			 * data will actually be the same in io_abd and
2556 			 * io_orig_abd, so all we have to do is issue this as
2557 			 * a raw ARC read.
2558 			 */
2559 			if (do_raw) {
2560 				zio_flags |= ZIO_FLAG_RAW;
2561 				ASSERT3U(zio->io_size, ==, zio->io_orig_size);
2562 				ASSERT0(abd_cmp(zio->io_abd, zio->io_orig_abd,
2563 				    zio->io_size));
2564 				ASSERT3P(zio->io_transform_stack, ==, NULL);
2565 			}
2566 
2567 			error = arc_read(NULL, spa, &blk,
2568 			    arc_getbuf_func, &abuf, ZIO_PRIORITY_SYNC_READ,
2569 			    zio_flags, &aflags, &zio->io_bookmark);
2570 
2571 			if (error == 0) {
2572 				if (arc_buf_size(abuf) != zio->io_orig_size ||
2573 				    abd_cmp_buf(zio->io_orig_abd, abuf->b_data,
2574 				    zio->io_orig_size) != 0)
2575 					error = SET_ERROR(EEXIST);
2576 				arc_buf_destroy(abuf, &abuf);
2577 			}
2578 
2579 			ddt_enter(ddt);
2580 			return (error != 0);
2581 		}
2582 	}
2583 
2584 	return (B_FALSE);
2585 }
2586 
2587 static void
2588 zio_ddt_child_write_ready(zio_t *zio)
2589 {
2590 	int p = zio->io_prop.zp_copies;
2591 	ddt_t *ddt = ddt_select(zio->io_spa, zio->io_bp);
2592 	ddt_entry_t *dde = zio->io_private;
2593 	ddt_phys_t *ddp = &dde->dde_phys[p];
2594 	zio_t *pio;
2595 
2596 	if (zio->io_error)
2597 		return;
2598 
2599 	ddt_enter(ddt);
2600 
2601 	ASSERT(dde->dde_lead_zio[p] == zio);
2602 
2603 	ddt_phys_fill(ddp, zio->io_bp);
2604 
2605 	zio_link_t *zl = NULL;
2606 	while ((pio = zio_walk_parents(zio, &zl)) != NULL)
2607 		ddt_bp_fill(ddp, pio->io_bp, zio->io_txg);
2608 
2609 	ddt_exit(ddt);
2610 }
2611 
2612 static void
2613 zio_ddt_child_write_done(zio_t *zio)
2614 {
2615 	int p = zio->io_prop.zp_copies;
2616 	ddt_t *ddt = ddt_select(zio->io_spa, zio->io_bp);
2617 	ddt_entry_t *dde = zio->io_private;
2618 	ddt_phys_t *ddp = &dde->dde_phys[p];
2619 
2620 	ddt_enter(ddt);
2621 
2622 	ASSERT(ddp->ddp_refcnt == 0);
2623 	ASSERT(dde->dde_lead_zio[p] == zio);
2624 	dde->dde_lead_zio[p] = NULL;
2625 
2626 	if (zio->io_error == 0) {
2627 		zio_link_t *zl = NULL;
2628 		while (zio_walk_parents(zio, &zl) != NULL)
2629 			ddt_phys_addref(ddp);
2630 	} else {
2631 		ddt_phys_clear(ddp);
2632 	}
2633 
2634 	ddt_exit(ddt);
2635 }
2636 
2637 static void
2638 zio_ddt_ditto_write_done(zio_t *zio)
2639 {
2640 	int p = DDT_PHYS_DITTO;
2641 	zio_prop_t *zp = &zio->io_prop;
2642 	blkptr_t *bp = zio->io_bp;
2643 	ddt_t *ddt = ddt_select(zio->io_spa, bp);
2644 	ddt_entry_t *dde = zio->io_private;
2645 	ddt_phys_t *ddp = &dde->dde_phys[p];
2646 	ddt_key_t *ddk = &dde->dde_key;
2647 
2648 	ddt_enter(ddt);
2649 
2650 	ASSERT(ddp->ddp_refcnt == 0);
2651 	ASSERT(dde->dde_lead_zio[p] == zio);
2652 	dde->dde_lead_zio[p] = NULL;
2653 
2654 	if (zio->io_error == 0) {
2655 		ASSERT(ZIO_CHECKSUM_EQUAL(bp->blk_cksum, ddk->ddk_cksum));
2656 		ASSERT(zp->zp_copies < SPA_DVAS_PER_BP);
2657 		ASSERT(zp->zp_copies == BP_GET_NDVAS(bp) - BP_IS_GANG(bp));
2658 		if (ddp->ddp_phys_birth != 0)
2659 			ddt_phys_free(ddt, ddk, ddp, zio->io_txg);
2660 		ddt_phys_fill(ddp, bp);
2661 	}
2662 
2663 	ddt_exit(ddt);
2664 }
2665 
2666 static int
2667 zio_ddt_write(zio_t *zio)
2668 {
2669 	spa_t *spa = zio->io_spa;
2670 	blkptr_t *bp = zio->io_bp;
2671 	uint64_t txg = zio->io_txg;
2672 	zio_prop_t *zp = &zio->io_prop;
2673 	int p = zp->zp_copies;
2674 	int ditto_copies;
2675 	zio_t *cio = NULL;
2676 	zio_t *dio = NULL;
2677 	ddt_t *ddt = ddt_select(spa, bp);
2678 	ddt_entry_t *dde;
2679 	ddt_phys_t *ddp;
2680 
2681 	ASSERT(BP_GET_DEDUP(bp));
2682 	ASSERT(BP_GET_CHECKSUM(bp) == zp->zp_checksum);
2683 	ASSERT(BP_IS_HOLE(bp) || zio->io_bp_override);
2684 	ASSERT(!(zio->io_bp_override && (zio->io_flags & ZIO_FLAG_RAW)));
2685 
2686 	ddt_enter(ddt);
2687 	dde = ddt_lookup(ddt, bp, B_TRUE);
2688 	ddp = &dde->dde_phys[p];
2689 
2690 	if (zp->zp_dedup_verify && zio_ddt_collision(zio, ddt, dde)) {
2691 		/*
2692 		 * If we're using a weak checksum, upgrade to a strong checksum
2693 		 * and try again.  If we're already using a strong checksum,
2694 		 * we can't resolve it, so just convert to an ordinary write.
2695 		 * (And automatically e-mail a paper to Nature?)
2696 		 */
2697 		if (!(zio_checksum_table[zp->zp_checksum].ci_flags &
2698 		    ZCHECKSUM_FLAG_DEDUP)) {
2699 			zp->zp_checksum = spa_dedup_checksum(spa);
2700 			zio_pop_transforms(zio);
2701 			zio->io_stage = ZIO_STAGE_OPEN;
2702 			BP_ZERO(bp);
2703 		} else {
2704 			zp->zp_dedup = B_FALSE;
2705 			BP_SET_DEDUP(bp, B_FALSE);
2706 		}
2707 		ASSERT(!BP_GET_DEDUP(bp));
2708 		zio->io_pipeline = ZIO_WRITE_PIPELINE;
2709 		ddt_exit(ddt);
2710 		return (ZIO_PIPELINE_CONTINUE);
2711 	}
2712 
2713 	ditto_copies = ddt_ditto_copies_needed(ddt, dde, ddp);
2714 	ASSERT(ditto_copies < SPA_DVAS_PER_BP);
2715 
2716 	if (ditto_copies > ddt_ditto_copies_present(dde) &&
2717 	    dde->dde_lead_zio[DDT_PHYS_DITTO] == NULL) {
2718 		zio_prop_t czp = *zp;
2719 
2720 		czp.zp_copies = ditto_copies;
2721 
2722 		/*
2723 		 * If we arrived here with an override bp, we won't have run
2724 		 * the transform stack, so we won't have the data we need to
2725 		 * generate a child i/o.  So, toss the override bp and restart.
2726 		 * This is safe, because using the override bp is just an
2727 		 * optimization; and it's rare, so the cost doesn't matter.
2728 		 */
2729 		if (zio->io_bp_override) {
2730 			zio_pop_transforms(zio);
2731 			zio->io_stage = ZIO_STAGE_OPEN;
2732 			zio->io_pipeline = ZIO_WRITE_PIPELINE;
2733 			zio->io_bp_override = NULL;
2734 			BP_ZERO(bp);
2735 			ddt_exit(ddt);
2736 			return (ZIO_PIPELINE_CONTINUE);
2737 		}
2738 
2739 		dio = zio_write(zio, spa, txg, bp, zio->io_orig_abd,
2740 		    zio->io_orig_size, zio->io_orig_size, &czp, NULL, NULL,
2741 		    NULL, zio_ddt_ditto_write_done, dde, zio->io_priority,
2742 		    ZIO_DDT_CHILD_FLAGS(zio), &zio->io_bookmark);
2743 
2744 		zio_push_transform(dio, zio->io_abd, zio->io_size, 0, NULL);
2745 		dde->dde_lead_zio[DDT_PHYS_DITTO] = dio;
2746 	}
2747 
2748 	if (ddp->ddp_phys_birth != 0 || dde->dde_lead_zio[p] != NULL) {
2749 		if (ddp->ddp_phys_birth != 0)
2750 			ddt_bp_fill(ddp, bp, txg);
2751 		if (dde->dde_lead_zio[p] != NULL)
2752 			zio_add_child(zio, dde->dde_lead_zio[p]);
2753 		else
2754 			ddt_phys_addref(ddp);
2755 	} else if (zio->io_bp_override) {
2756 		ASSERT(bp->blk_birth == txg);
2757 		ASSERT(BP_EQUAL(bp, zio->io_bp_override));
2758 		ddt_phys_fill(ddp, bp);
2759 		ddt_phys_addref(ddp);
2760 	} else {
2761 		cio = zio_write(zio, spa, txg, bp, zio->io_orig_abd,
2762 		    zio->io_orig_size, zio->io_orig_size, zp,
2763 		    zio_ddt_child_write_ready, NULL, NULL,
2764 		    zio_ddt_child_write_done, dde, zio->io_priority,
2765 		    ZIO_DDT_CHILD_FLAGS(zio), &zio->io_bookmark);
2766 
2767 		zio_push_transform(cio, zio->io_abd, zio->io_size, 0, NULL);
2768 		dde->dde_lead_zio[p] = cio;
2769 	}
2770 
2771 	ddt_exit(ddt);
2772 
2773 	if (cio)
2774 		zio_nowait(cio);
2775 	if (dio)
2776 		zio_nowait(dio);
2777 
2778 	return (ZIO_PIPELINE_CONTINUE);
2779 }
2780 
2781 ddt_entry_t *freedde; /* for debugging */
2782 
2783 static int
2784 zio_ddt_free(zio_t *zio)
2785 {
2786 	spa_t *spa = zio->io_spa;
2787 	blkptr_t *bp = zio->io_bp;
2788 	ddt_t *ddt = ddt_select(spa, bp);
2789 	ddt_entry_t *dde;
2790 	ddt_phys_t *ddp;
2791 
2792 	ASSERT(BP_GET_DEDUP(bp));
2793 	ASSERT(zio->io_child_type == ZIO_CHILD_LOGICAL);
2794 
2795 	ddt_enter(ddt);
2796 	freedde = dde = ddt_lookup(ddt, bp, B_TRUE);
2797 	ddp = ddt_phys_select(dde, bp);
2798 	ddt_phys_decref(ddp);
2799 	ddt_exit(ddt);
2800 
2801 	return (ZIO_PIPELINE_CONTINUE);
2802 }
2803 
2804 /*
2805  * ==========================================================================
2806  * Allocate and free blocks
2807  * ==========================================================================
2808  */
2809 
2810 static zio_t *
2811 zio_io_to_allocate(spa_t *spa)
2812 {
2813 	zio_t *zio;
2814 
2815 	ASSERT(MUTEX_HELD(&spa->spa_alloc_lock));
2816 
2817 	zio = avl_first(&spa->spa_alloc_tree);
2818 	if (zio == NULL)
2819 		return (NULL);
2820 
2821 	ASSERT(IO_IS_ALLOCATING(zio));
2822 
2823 	/*
2824 	 * Try to place a reservation for this zio. If we're unable to
2825 	 * reserve then we throttle.
2826 	 */
2827 	if (!metaslab_class_throttle_reserve(spa_normal_class(spa),
2828 	    zio->io_prop.zp_copies, zio, 0)) {
2829 		return (NULL);
2830 	}
2831 
2832 	avl_remove(&spa->spa_alloc_tree, zio);
2833 	ASSERT3U(zio->io_stage, <, ZIO_STAGE_DVA_ALLOCATE);
2834 
2835 	return (zio);
2836 }
2837 
2838 static int
2839 zio_dva_throttle(zio_t *zio)
2840 {
2841 	spa_t *spa = zio->io_spa;
2842 	zio_t *nio;
2843 
2844 	if (zio->io_priority == ZIO_PRIORITY_SYNC_WRITE ||
2845 	    !spa_normal_class(zio->io_spa)->mc_alloc_throttle_enabled ||
2846 	    zio->io_child_type == ZIO_CHILD_GANG ||
2847 	    zio->io_flags & ZIO_FLAG_NODATA) {
2848 		return (ZIO_PIPELINE_CONTINUE);
2849 	}
2850 
2851 	ASSERT(zio->io_child_type > ZIO_CHILD_GANG);
2852 
2853 	ASSERT3U(zio->io_queued_timestamp, >, 0);
2854 	ASSERT(zio->io_stage == ZIO_STAGE_DVA_THROTTLE);
2855 
2856 	mutex_enter(&spa->spa_alloc_lock);
2857 
2858 	ASSERT(zio->io_type == ZIO_TYPE_WRITE);
2859 	avl_add(&spa->spa_alloc_tree, zio);
2860 
2861 	nio = zio_io_to_allocate(zio->io_spa);
2862 	mutex_exit(&spa->spa_alloc_lock);
2863 
2864 	if (nio == zio)
2865 		return (ZIO_PIPELINE_CONTINUE);
2866 
2867 	if (nio != NULL) {
2868 		ASSERT(nio->io_stage == ZIO_STAGE_DVA_THROTTLE);
2869 		/*
2870 		 * We are passing control to a new zio so make sure that
2871 		 * it is processed by a different thread. We do this to
2872 		 * avoid stack overflows that can occur when parents are
2873 		 * throttled and children are making progress. We allow
2874 		 * it to go to the head of the taskq since it's already
2875 		 * been waiting.
2876 		 */
2877 		zio_taskq_dispatch(nio, ZIO_TASKQ_ISSUE, B_TRUE);
2878 	}
2879 	return (ZIO_PIPELINE_STOP);
2880 }
2881 
2882 void
2883 zio_allocate_dispatch(spa_t *spa)
2884 {
2885 	zio_t *zio;
2886 
2887 	mutex_enter(&spa->spa_alloc_lock);
2888 	zio = zio_io_to_allocate(spa);
2889 	mutex_exit(&spa->spa_alloc_lock);
2890 	if (zio == NULL)
2891 		return;
2892 
2893 	ASSERT3U(zio->io_stage, ==, ZIO_STAGE_DVA_THROTTLE);
2894 	ASSERT0(zio->io_error);
2895 	zio_taskq_dispatch(zio, ZIO_TASKQ_ISSUE, B_TRUE);
2896 }
2897 
2898 static int
2899 zio_dva_allocate(zio_t *zio)
2900 {
2901 	spa_t *spa = zio->io_spa;
2902 	metaslab_class_t *mc = spa_normal_class(spa);
2903 	blkptr_t *bp = zio->io_bp;
2904 	int error;
2905 	int flags = 0;
2906 
2907 	if (zio->io_gang_leader == NULL) {
2908 		ASSERT(zio->io_child_type > ZIO_CHILD_GANG);
2909 		zio->io_gang_leader = zio;
2910 	}
2911 
2912 	ASSERT(BP_IS_HOLE(bp));
2913 	ASSERT0(BP_GET_NDVAS(bp));
2914 	ASSERT3U(zio->io_prop.zp_copies, >, 0);
2915 	ASSERT3U(zio->io_prop.zp_copies, <=, spa_max_replication(spa));
2916 	ASSERT3U(zio->io_size, ==, BP_GET_PSIZE(bp));
2917 
2918 	if (zio->io_flags & ZIO_FLAG_NODATA) {
2919 		flags |= METASLAB_DONT_THROTTLE;
2920 	}
2921 	if (zio->io_flags & ZIO_FLAG_GANG_CHILD) {
2922 		flags |= METASLAB_GANG_CHILD;
2923 	}
2924 	if (zio->io_priority == ZIO_PRIORITY_ASYNC_WRITE) {
2925 		flags |= METASLAB_ASYNC_ALLOC;
2926 	}
2927 
2928 	error = metaslab_alloc(spa, mc, zio->io_size, bp,
2929 	    zio->io_prop.zp_copies, zio->io_txg, NULL, flags,
2930 	    &zio->io_alloc_list, zio);
2931 
2932 	if (error != 0) {
2933 		spa_dbgmsg(spa, "%s: metaslab allocation failure: zio %p, "
2934 		    "size %llu, error %d", spa_name(spa), zio, zio->io_size,
2935 		    error);
2936 		if (error == ENOSPC && zio->io_size > SPA_MINBLOCKSIZE)
2937 			return (zio_write_gang_block(zio));
2938 		zio->io_error = error;
2939 	}
2940 
2941 	return (ZIO_PIPELINE_CONTINUE);
2942 }
2943 
2944 static int
2945 zio_dva_free(zio_t *zio)
2946 {
2947 	metaslab_free(zio->io_spa, zio->io_bp, zio->io_txg, B_FALSE);
2948 
2949 	return (ZIO_PIPELINE_CONTINUE);
2950 }
2951 
2952 static int
2953 zio_dva_claim(zio_t *zio)
2954 {
2955 	int error;
2956 
2957 	error = metaslab_claim(zio->io_spa, zio->io_bp, zio->io_txg);
2958 	if (error)
2959 		zio->io_error = error;
2960 
2961 	return (ZIO_PIPELINE_CONTINUE);
2962 }
2963 
2964 /*
2965  * Undo an allocation.  This is used by zio_done() when an I/O fails
2966  * and we want to give back the block we just allocated.
2967  * This handles both normal blocks and gang blocks.
2968  */
2969 static void
2970 zio_dva_unallocate(zio_t *zio, zio_gang_node_t *gn, blkptr_t *bp)
2971 {
2972 	ASSERT(bp->blk_birth == zio->io_txg || BP_IS_HOLE(bp));
2973 	ASSERT(zio->io_bp_override == NULL);
2974 
2975 	if (!BP_IS_HOLE(bp))
2976 		metaslab_free(zio->io_spa, bp, bp->blk_birth, B_TRUE);
2977 
2978 	if (gn != NULL) {
2979 		for (int g = 0; g < SPA_GBH_NBLKPTRS; g++) {
2980 			zio_dva_unallocate(zio, gn->gn_child[g],
2981 			    &gn->gn_gbh->zg_blkptr[g]);
2982 		}
2983 	}
2984 }
2985 
2986 /*
2987  * Try to allocate an intent log block.  Return 0 on success, errno on failure.
2988  */
2989 int
2990 zio_alloc_zil(spa_t *spa, uint64_t txg, blkptr_t *new_bp, blkptr_t *old_bp,
2991     uint64_t size, boolean_t *slog)
2992 {
2993 	int error = 1;
2994 	zio_alloc_list_t io_alloc_list;
2995 
2996 	ASSERT(txg > spa_syncing_txg(spa));
2997 
2998 	metaslab_trace_init(&io_alloc_list);
2999 	error = metaslab_alloc(spa, spa_log_class(spa), size, new_bp, 1,
3000 	    txg, old_bp, METASLAB_HINTBP_AVOID, &io_alloc_list, NULL);
3001 	if (error == 0) {
3002 		*slog = TRUE;
3003 	} else {
3004 		error = metaslab_alloc(spa, spa_normal_class(spa), size,
3005 		    new_bp, 1, txg, old_bp, METASLAB_HINTBP_AVOID,
3006 		    &io_alloc_list, NULL);
3007 		if (error == 0)
3008 			*slog = FALSE;
3009 	}
3010 	metaslab_trace_fini(&io_alloc_list);
3011 
3012 	if (error == 0) {
3013 		BP_SET_LSIZE(new_bp, size);
3014 		BP_SET_PSIZE(new_bp, size);
3015 		BP_SET_COMPRESS(new_bp, ZIO_COMPRESS_OFF);
3016 		BP_SET_CHECKSUM(new_bp,
3017 		    spa_version(spa) >= SPA_VERSION_SLIM_ZIL
3018 		    ? ZIO_CHECKSUM_ZILOG2 : ZIO_CHECKSUM_ZILOG);
3019 		BP_SET_TYPE(new_bp, DMU_OT_INTENT_LOG);
3020 		BP_SET_LEVEL(new_bp, 0);
3021 		BP_SET_DEDUP(new_bp, 0);
3022 		BP_SET_BYTEORDER(new_bp, ZFS_HOST_BYTEORDER);
3023 	} else {
3024 		zfs_dbgmsg("%s: zil block allocation failure: "
3025 		    "size %llu, error %d", spa_name(spa), size, error);
3026 	}
3027 
3028 	return (error);
3029 }
3030 
3031 /*
3032  * ==========================================================================
3033  * Read and write to physical devices
3034  * ==========================================================================
3035  */
3036 
3037 
3038 /*
3039  * Issue an I/O to the underlying vdev. Typically the issue pipeline
3040  * stops after this stage and will resume upon I/O completion.
3041  * However, there are instances where the vdev layer may need to
3042  * continue the pipeline when an I/O was not issued. Since the I/O
3043  * that was sent to the vdev layer might be different than the one
3044  * currently active in the pipeline (see vdev_queue_io()), we explicitly
3045  * force the underlying vdev layers to call either zio_execute() or
3046  * zio_interrupt() to ensure that the pipeline continues with the correct I/O.
3047  */
3048 static int
3049 zio_vdev_io_start(zio_t *zio)
3050 {
3051 	vdev_t *vd = zio->io_vd;
3052 	uint64_t align;
3053 	spa_t *spa = zio->io_spa;
3054 
3055 	ASSERT(zio->io_error == 0);
3056 	ASSERT(zio->io_child_error[ZIO_CHILD_VDEV] == 0);
3057 
3058 	if (vd == NULL) {
3059 		if (!(zio->io_flags & ZIO_FLAG_CONFIG_WRITER))
3060 			spa_config_enter(spa, SCL_ZIO, zio, RW_READER);
3061 
3062 		/*
3063 		 * The mirror_ops handle multiple DVAs in a single BP.
3064 		 */
3065 		vdev_mirror_ops.vdev_op_io_start(zio);
3066 		return (ZIO_PIPELINE_STOP);
3067 	}
3068 
3069 	ASSERT3P(zio->io_logical, !=, zio);
3070 	if (zio->io_type == ZIO_TYPE_WRITE) {
3071 		ASSERT(spa->spa_trust_config);
3072 
3073 		if (zio->io_vd->vdev_removing) {
3074 			ASSERT(zio->io_flags &
3075 			    (ZIO_FLAG_PHYSICAL | ZIO_FLAG_SELF_HEAL |
3076 			    ZIO_FLAG_INDUCE_DAMAGE));
3077 		}
3078 	}
3079 
3080 	/*
3081 	 * We keep track of time-sensitive I/Os so that the scan thread
3082 	 * can quickly react to certain workloads.  In particular, we care
3083 	 * about non-scrubbing, top-level reads and writes with the following
3084 	 * characteristics:
3085 	 *	- synchronous writes of user data to non-slog devices
3086 	 *	- any reads of user data
3087 	 * When these conditions are met, adjust the timestamp of spa_last_io
3088 	 * which allows the scan thread to adjust its workload accordingly.
3089 	 */
3090 	if (!(zio->io_flags & ZIO_FLAG_SCAN_THREAD) && zio->io_bp != NULL &&
3091 	    vd == vd->vdev_top && !vd->vdev_islog &&
3092 	    zio->io_bookmark.zb_objset != DMU_META_OBJSET &&
3093 	    zio->io_txg != spa_syncing_txg(spa)) {
3094 		uint64_t old = spa->spa_last_io;
3095 		uint64_t new = ddi_get_lbolt64();
3096 		if (old != new)
3097 			(void) atomic_cas_64(&spa->spa_last_io, old, new);
3098 	}
3099 
3100 	align = 1ULL << vd->vdev_top->vdev_ashift;
3101 
3102 	if (!(zio->io_flags & ZIO_FLAG_PHYSICAL) &&
3103 	    P2PHASE(zio->io_size, align) != 0) {
3104 		/* Transform logical writes to be a full physical block size. */
3105 		uint64_t asize = P2ROUNDUP(zio->io_size, align);
3106 		abd_t *abuf = abd_alloc_sametype(zio->io_abd, asize);
3107 		ASSERT(vd == vd->vdev_top);
3108 		if (zio->io_type == ZIO_TYPE_WRITE) {
3109 			abd_copy(abuf, zio->io_abd, zio->io_size);
3110 			abd_zero_off(abuf, zio->io_size, asize - zio->io_size);
3111 		}
3112 		zio_push_transform(zio, abuf, asize, asize, zio_subblock);
3113 	}
3114 
3115 	/*
3116 	 * If this is not a physical io, make sure that it is properly aligned
3117 	 * before proceeding.
3118 	 */
3119 	if (!(zio->io_flags & ZIO_FLAG_PHYSICAL)) {
3120 		ASSERT0(P2PHASE(zio->io_offset, align));
3121 		ASSERT0(P2PHASE(zio->io_size, align));
3122 	} else {
3123 		/*
3124 		 * For physical writes, we allow 512b aligned writes and assume
3125 		 * the device will perform a read-modify-write as necessary.
3126 		 */
3127 		ASSERT0(P2PHASE(zio->io_offset, SPA_MINBLOCKSIZE));
3128 		ASSERT0(P2PHASE(zio->io_size, SPA_MINBLOCKSIZE));
3129 	}
3130 
3131 	VERIFY(zio->io_type != ZIO_TYPE_WRITE || spa_writeable(spa));
3132 
3133 	/*
3134 	 * If this is a repair I/O, and there's no self-healing involved --
3135 	 * that is, we're just resilvering what we expect to resilver --
3136 	 * then don't do the I/O unless zio's txg is actually in vd's DTL.
3137 	 * This prevents spurious resilvering with nested replication.
3138 	 * For example, given a mirror of mirrors, (A+B)+(C+D), if only
3139 	 * A is out of date, we'll read from C+D, then use the data to
3140 	 * resilver A+B -- but we don't actually want to resilver B, just A.
3141 	 * The top-level mirror has no way to know this, so instead we just
3142 	 * discard unnecessary repairs as we work our way down the vdev tree.
3143 	 * The same logic applies to any form of nested replication:
3144 	 * ditto + mirror, RAID-Z + replacing, etc.  This covers them all.
3145 	 */
3146 	if ((zio->io_flags & ZIO_FLAG_IO_REPAIR) &&
3147 	    !(zio->io_flags & ZIO_FLAG_SELF_HEAL) &&
3148 	    zio->io_txg != 0 &&	/* not a delegated i/o */
3149 	    !vdev_dtl_contains(vd, DTL_PARTIAL, zio->io_txg, 1)) {
3150 		ASSERT(zio->io_type == ZIO_TYPE_WRITE);
3151 		zio_vdev_io_bypass(zio);
3152 		return (ZIO_PIPELINE_CONTINUE);
3153 	}
3154 
3155 	if (vd->vdev_ops->vdev_op_leaf &&
3156 	    (zio->io_type == ZIO_TYPE_READ || zio->io_type == ZIO_TYPE_WRITE)) {
3157 
3158 		if (zio->io_type == ZIO_TYPE_READ && vdev_cache_read(zio))
3159 			return (ZIO_PIPELINE_CONTINUE);
3160 
3161 		if ((zio = vdev_queue_io(zio)) == NULL)
3162 			return (ZIO_PIPELINE_STOP);
3163 
3164 		if (!vdev_accessible(vd, zio)) {
3165 			zio->io_error = SET_ERROR(ENXIO);
3166 			zio_interrupt(zio);
3167 			return (ZIO_PIPELINE_STOP);
3168 		}
3169 	}
3170 
3171 	vd->vdev_ops->vdev_op_io_start(zio);
3172 	return (ZIO_PIPELINE_STOP);
3173 }
3174 
3175 static int
3176 zio_vdev_io_done(zio_t *zio)
3177 {
3178 	vdev_t *vd = zio->io_vd;
3179 	vdev_ops_t *ops = vd ? vd->vdev_ops : &vdev_mirror_ops;
3180 	boolean_t unexpected_error = B_FALSE;
3181 
3182 	if (zio_wait_for_children(zio, ZIO_CHILD_VDEV_BIT, ZIO_WAIT_DONE)) {
3183 		return (ZIO_PIPELINE_STOP);
3184 	}
3185 
3186 	ASSERT(zio->io_type == ZIO_TYPE_READ || zio->io_type == ZIO_TYPE_WRITE);
3187 
3188 	if (vd != NULL && vd->vdev_ops->vdev_op_leaf) {
3189 
3190 		vdev_queue_io_done(zio);
3191 
3192 		if (zio->io_type == ZIO_TYPE_WRITE)
3193 			vdev_cache_write(zio);
3194 
3195 		if (zio_injection_enabled && zio->io_error == 0)
3196 			zio->io_error = zio_handle_device_injection(vd,
3197 			    zio, EIO);
3198 
3199 		if (zio_injection_enabled && zio->io_error == 0)
3200 			zio->io_error = zio_handle_label_injection(zio, EIO);
3201 
3202 		if (zio->io_error) {
3203 			if (!vdev_accessible(vd, zio)) {
3204 				zio->io_error = SET_ERROR(ENXIO);
3205 			} else {
3206 				unexpected_error = B_TRUE;
3207 			}
3208 		}
3209 	}
3210 
3211 	ops->vdev_op_io_done(zio);
3212 
3213 	if (unexpected_error)
3214 		VERIFY(vdev_probe(vd, zio) == NULL);
3215 
3216 	return (ZIO_PIPELINE_CONTINUE);
3217 }
3218 
3219 /*
3220  * For non-raidz ZIOs, we can just copy aside the bad data read from the
3221  * disk, and use that to finish the checksum ereport later.
3222  */
3223 static void
3224 zio_vsd_default_cksum_finish(zio_cksum_report_t *zcr,
3225     const void *good_buf)
3226 {
3227 	/* no processing needed */
3228 	zfs_ereport_finish_checksum(zcr, good_buf, zcr->zcr_cbdata, B_FALSE);
3229 }
3230 
3231 /*ARGSUSED*/
3232 void
3233 zio_vsd_default_cksum_report(zio_t *zio, zio_cksum_report_t *zcr, void *ignored)
3234 {
3235 	void *buf = zio_buf_alloc(zio->io_size);
3236 
3237 	abd_copy_to_buf(buf, zio->io_abd, zio->io_size);
3238 
3239 	zcr->zcr_cbinfo = zio->io_size;
3240 	zcr->zcr_cbdata = buf;
3241 	zcr->zcr_finish = zio_vsd_default_cksum_finish;
3242 	zcr->zcr_free = zio_buf_free;
3243 }
3244 
3245 static int
3246 zio_vdev_io_assess(zio_t *zio)
3247 {
3248 	vdev_t *vd = zio->io_vd;
3249 
3250 	if (zio_wait_for_children(zio, ZIO_CHILD_VDEV_BIT, ZIO_WAIT_DONE)) {
3251 		return (ZIO_PIPELINE_STOP);
3252 	}
3253 
3254 	if (vd == NULL && !(zio->io_flags & ZIO_FLAG_CONFIG_WRITER))
3255 		spa_config_exit(zio->io_spa, SCL_ZIO, zio);
3256 
3257 	if (zio->io_vsd != NULL) {
3258 		zio->io_vsd_ops->vsd_free(zio);
3259 		zio->io_vsd = NULL;
3260 	}
3261 
3262 	if (zio_injection_enabled && zio->io_error == 0)
3263 		zio->io_error = zio_handle_fault_injection(zio, EIO);
3264 
3265 	/*
3266 	 * If the I/O failed, determine whether we should attempt to retry it.
3267 	 *
3268 	 * On retry, we cut in line in the issue queue, since we don't want
3269 	 * compression/checksumming/etc. work to prevent our (cheap) IO reissue.
3270 	 */
3271 	if (zio->io_error && vd == NULL &&
3272 	    !(zio->io_flags & (ZIO_FLAG_DONT_RETRY | ZIO_FLAG_IO_RETRY))) {
3273 		ASSERT(!(zio->io_flags & ZIO_FLAG_DONT_QUEUE));	/* not a leaf */
3274 		ASSERT(!(zio->io_flags & ZIO_FLAG_IO_BYPASS));	/* not a leaf */
3275 		zio->io_error = 0;
3276 		zio->io_flags |= ZIO_FLAG_IO_RETRY |
3277 		    ZIO_FLAG_DONT_CACHE | ZIO_FLAG_DONT_AGGREGATE;
3278 		zio->io_stage = ZIO_STAGE_VDEV_IO_START >> 1;
3279 		zio_taskq_dispatch(zio, ZIO_TASKQ_ISSUE,
3280 		    zio_requeue_io_start_cut_in_line);
3281 		return (ZIO_PIPELINE_STOP);
3282 	}
3283 
3284 	/*
3285 	 * If we got an error on a leaf device, convert it to ENXIO
3286 	 * if the device is not accessible at all.
3287 	 */
3288 	if (zio->io_error && vd != NULL && vd->vdev_ops->vdev_op_leaf &&
3289 	    !vdev_accessible(vd, zio))
3290 		zio->io_error = SET_ERROR(ENXIO);
3291 
3292 	/*
3293 	 * If we can't write to an interior vdev (mirror or RAID-Z),
3294 	 * set vdev_cant_write so that we stop trying to allocate from it.
3295 	 */
3296 	if (zio->io_error == ENXIO && zio->io_type == ZIO_TYPE_WRITE &&
3297 	    vd != NULL && !vd->vdev_ops->vdev_op_leaf) {
3298 		vd->vdev_cant_write = B_TRUE;
3299 	}
3300 
3301 	/*
3302 	 * If a cache flush returns ENOTSUP or ENOTTY, we know that no future
3303 	 * attempts will ever succeed. In this case we set a persistent bit so
3304 	 * that we don't bother with it in the future.
3305 	 */
3306 	if ((zio->io_error == ENOTSUP || zio->io_error == ENOTTY) &&
3307 	    zio->io_type == ZIO_TYPE_IOCTL &&
3308 	    zio->io_cmd == DKIOCFLUSHWRITECACHE && vd != NULL)
3309 		vd->vdev_nowritecache = B_TRUE;
3310 
3311 	if (zio->io_error)
3312 		zio->io_pipeline = ZIO_INTERLOCK_PIPELINE;
3313 
3314 	if (vd != NULL && vd->vdev_ops->vdev_op_leaf &&
3315 	    zio->io_physdone != NULL) {
3316 		ASSERT(!(zio->io_flags & ZIO_FLAG_DELEGATED));
3317 		ASSERT(zio->io_child_type == ZIO_CHILD_VDEV);
3318 		zio->io_physdone(zio->io_logical);
3319 	}
3320 
3321 	return (ZIO_PIPELINE_CONTINUE);
3322 }
3323 
3324 void
3325 zio_vdev_io_reissue(zio_t *zio)
3326 {
3327 	ASSERT(zio->io_stage == ZIO_STAGE_VDEV_IO_START);
3328 	ASSERT(zio->io_error == 0);
3329 
3330 	zio->io_stage >>= 1;
3331 }
3332 
3333 void
3334 zio_vdev_io_redone(zio_t *zio)
3335 {
3336 	ASSERT(zio->io_stage == ZIO_STAGE_VDEV_IO_DONE);
3337 
3338 	zio->io_stage >>= 1;
3339 }
3340 
3341 void
3342 zio_vdev_io_bypass(zio_t *zio)
3343 {
3344 	ASSERT(zio->io_stage == ZIO_STAGE_VDEV_IO_START);
3345 	ASSERT(zio->io_error == 0);
3346 
3347 	zio->io_flags |= ZIO_FLAG_IO_BYPASS;
3348 	zio->io_stage = ZIO_STAGE_VDEV_IO_ASSESS >> 1;
3349 }
3350 
3351 /*
3352  * ==========================================================================
3353  * Generate and verify checksums
3354  * ==========================================================================
3355  */
3356 static int
3357 zio_checksum_generate(zio_t *zio)
3358 {
3359 	blkptr_t *bp = zio->io_bp;
3360 	enum zio_checksum checksum;
3361 
3362 	if (bp == NULL) {
3363 		/*
3364 		 * This is zio_write_phys().
3365 		 * We're either generating a label checksum, or none at all.
3366 		 */
3367 		checksum = zio->io_prop.zp_checksum;
3368 
3369 		if (checksum == ZIO_CHECKSUM_OFF)
3370 			return (ZIO_PIPELINE_CONTINUE);
3371 
3372 		ASSERT(checksum == ZIO_CHECKSUM_LABEL);
3373 	} else {
3374 		if (BP_IS_GANG(bp) && zio->io_child_type == ZIO_CHILD_GANG) {
3375 			ASSERT(!IO_IS_ALLOCATING(zio));
3376 			checksum = ZIO_CHECKSUM_GANG_HEADER;
3377 		} else {
3378 			checksum = BP_GET_CHECKSUM(bp);
3379 		}
3380 	}
3381 
3382 	zio_checksum_compute(zio, checksum, zio->io_abd, zio->io_size);
3383 
3384 	return (ZIO_PIPELINE_CONTINUE);
3385 }
3386 
3387 static int
3388 zio_checksum_verify(zio_t *zio)
3389 {
3390 	zio_bad_cksum_t info;
3391 	blkptr_t *bp = zio->io_bp;
3392 	int error;
3393 
3394 	ASSERT(zio->io_vd != NULL);
3395 
3396 	if (bp == NULL) {
3397 		/*
3398 		 * This is zio_read_phys().
3399 		 * We're either verifying a label checksum, or nothing at all.
3400 		 */
3401 		if (zio->io_prop.zp_checksum == ZIO_CHECKSUM_OFF)
3402 			return (ZIO_PIPELINE_CONTINUE);
3403 
3404 		ASSERT(zio->io_prop.zp_checksum == ZIO_CHECKSUM_LABEL);
3405 	}
3406 
3407 	if ((error = zio_checksum_error(zio, &info)) != 0) {
3408 		zio->io_error = error;
3409 		if (error == ECKSUM &&
3410 		    !(zio->io_flags & ZIO_FLAG_SPECULATIVE)) {
3411 			zfs_ereport_start_checksum(zio->io_spa,
3412 			    zio->io_vd, zio, zio->io_offset,
3413 			    zio->io_size, NULL, &info);
3414 		}
3415 	}
3416 
3417 	return (ZIO_PIPELINE_CONTINUE);
3418 }
3419 
3420 /*
3421  * Called by RAID-Z to ensure we don't compute the checksum twice.
3422  */
3423 void
3424 zio_checksum_verified(zio_t *zio)
3425 {
3426 	zio->io_pipeline &= ~ZIO_STAGE_CHECKSUM_VERIFY;
3427 }
3428 
3429 /*
3430  * ==========================================================================
3431  * Error rank.  Error are ranked in the order 0, ENXIO, ECKSUM, EIO, other.
3432  * An error of 0 indicates success.  ENXIO indicates whole-device failure,
3433  * which may be transient (e.g. unplugged) or permament.  ECKSUM and EIO
3434  * indicate errors that are specific to one I/O, and most likely permanent.
3435  * Any other error is presumed to be worse because we weren't expecting it.
3436  * ==========================================================================
3437  */
3438 int
3439 zio_worst_error(int e1, int e2)
3440 {
3441 	static int zio_error_rank[] = { 0, ENXIO, ECKSUM, EIO };
3442 	int r1, r2;
3443 
3444 	for (r1 = 0; r1 < sizeof (zio_error_rank) / sizeof (int); r1++)
3445 		if (e1 == zio_error_rank[r1])
3446 			break;
3447 
3448 	for (r2 = 0; r2 < sizeof (zio_error_rank) / sizeof (int); r2++)
3449 		if (e2 == zio_error_rank[r2])
3450 			break;
3451 
3452 	return (r1 > r2 ? e1 : e2);
3453 }
3454 
3455 /*
3456  * ==========================================================================
3457  * I/O completion
3458  * ==========================================================================
3459  */
3460 static int
3461 zio_ready(zio_t *zio)
3462 {
3463 	blkptr_t *bp = zio->io_bp;
3464 	zio_t *pio, *pio_next;
3465 	zio_link_t *zl = NULL;
3466 
3467 	if (zio_wait_for_children(zio, ZIO_CHILD_GANG_BIT | ZIO_CHILD_DDT_BIT,
3468 	    ZIO_WAIT_READY)) {
3469 		return (ZIO_PIPELINE_STOP);
3470 	}
3471 
3472 	if (zio->io_ready) {
3473 		ASSERT(IO_IS_ALLOCATING(zio));
3474 		ASSERT(bp->blk_birth == zio->io_txg || BP_IS_HOLE(bp) ||
3475 		    (zio->io_flags & ZIO_FLAG_NOPWRITE));
3476 		ASSERT(zio->io_children[ZIO_CHILD_GANG][ZIO_WAIT_READY] == 0);
3477 
3478 		zio->io_ready(zio);
3479 	}
3480 
3481 	if (bp != NULL && bp != &zio->io_bp_copy)
3482 		zio->io_bp_copy = *bp;
3483 
3484 	if (zio->io_error != 0) {
3485 		zio->io_pipeline = ZIO_INTERLOCK_PIPELINE;
3486 
3487 		if (zio->io_flags & ZIO_FLAG_IO_ALLOCATING) {
3488 			ASSERT(IO_IS_ALLOCATING(zio));
3489 			ASSERT(zio->io_priority == ZIO_PRIORITY_ASYNC_WRITE);
3490 			/*
3491 			 * We were unable to allocate anything, unreserve and
3492 			 * issue the next I/O to allocate.
3493 			 */
3494 			metaslab_class_throttle_unreserve(
3495 			    spa_normal_class(zio->io_spa),
3496 			    zio->io_prop.zp_copies, zio);
3497 			zio_allocate_dispatch(zio->io_spa);
3498 		}
3499 	}
3500 
3501 	mutex_enter(&zio->io_lock);
3502 	zio->io_state[ZIO_WAIT_READY] = 1;
3503 	pio = zio_walk_parents(zio, &zl);
3504 	mutex_exit(&zio->io_lock);
3505 
3506 	/*
3507 	 * As we notify zio's parents, new parents could be added.
3508 	 * New parents go to the head of zio's io_parent_list, however,
3509 	 * so we will (correctly) not notify them.  The remainder of zio's
3510 	 * io_parent_list, from 'pio_next' onward, cannot change because
3511 	 * all parents must wait for us to be done before they can be done.
3512 	 */
3513 	for (; pio != NULL; pio = pio_next) {
3514 		pio_next = zio_walk_parents(zio, &zl);
3515 		zio_notify_parent(pio, zio, ZIO_WAIT_READY);
3516 	}
3517 
3518 	if (zio->io_flags & ZIO_FLAG_NODATA) {
3519 		if (BP_IS_GANG(bp)) {
3520 			zio->io_flags &= ~ZIO_FLAG_NODATA;
3521 		} else {
3522 			ASSERT((uintptr_t)zio->io_abd < SPA_MAXBLOCKSIZE);
3523 			zio->io_pipeline &= ~ZIO_VDEV_IO_STAGES;
3524 		}
3525 	}
3526 
3527 	if (zio_injection_enabled &&
3528 	    zio->io_spa->spa_syncing_txg == zio->io_txg)
3529 		zio_handle_ignored_writes(zio);
3530 
3531 	return (ZIO_PIPELINE_CONTINUE);
3532 }
3533 
3534 /*
3535  * Update the allocation throttle accounting.
3536  */
3537 static void
3538 zio_dva_throttle_done(zio_t *zio)
3539 {
3540 	zio_t *lio = zio->io_logical;
3541 	zio_t *pio = zio_unique_parent(zio);
3542 	vdev_t *vd = zio->io_vd;
3543 	int flags = METASLAB_ASYNC_ALLOC;
3544 
3545 	ASSERT3P(zio->io_bp, !=, NULL);
3546 	ASSERT3U(zio->io_type, ==, ZIO_TYPE_WRITE);
3547 	ASSERT3U(zio->io_priority, ==, ZIO_PRIORITY_ASYNC_WRITE);
3548 	ASSERT3U(zio->io_child_type, ==, ZIO_CHILD_VDEV);
3549 	ASSERT(vd != NULL);
3550 	ASSERT3P(vd, ==, vd->vdev_top);
3551 	ASSERT(!(zio->io_flags & (ZIO_FLAG_IO_REPAIR | ZIO_FLAG_IO_RETRY)));
3552 	ASSERT(zio->io_flags & ZIO_FLAG_IO_ALLOCATING);
3553 	ASSERT(!(lio->io_flags & ZIO_FLAG_IO_REWRITE));
3554 	ASSERT(!(lio->io_orig_flags & ZIO_FLAG_NODATA));
3555 
3556 	/*
3557 	 * Parents of gang children can have two flavors -- ones that
3558 	 * allocated the gang header (will have ZIO_FLAG_IO_REWRITE set)
3559 	 * and ones that allocated the constituent blocks. The allocation
3560 	 * throttle needs to know the allocating parent zio so we must find
3561 	 * it here.
3562 	 */
3563 	if (pio->io_child_type == ZIO_CHILD_GANG) {
3564 		/*
3565 		 * If our parent is a rewrite gang child then our grandparent
3566 		 * would have been the one that performed the allocation.
3567 		 */
3568 		if (pio->io_flags & ZIO_FLAG_IO_REWRITE)
3569 			pio = zio_unique_parent(pio);
3570 		flags |= METASLAB_GANG_CHILD;
3571 	}
3572 
3573 	ASSERT(IO_IS_ALLOCATING(pio));
3574 	ASSERT3P(zio, !=, zio->io_logical);
3575 	ASSERT(zio->io_logical != NULL);
3576 	ASSERT(!(zio->io_flags & ZIO_FLAG_IO_REPAIR));
3577 	ASSERT0(zio->io_flags & ZIO_FLAG_NOPWRITE);
3578 
3579 	mutex_enter(&pio->io_lock);
3580 	metaslab_group_alloc_decrement(zio->io_spa, vd->vdev_id, pio, flags);
3581 	mutex_exit(&pio->io_lock);
3582 
3583 	metaslab_class_throttle_unreserve(spa_normal_class(zio->io_spa),
3584 	    1, pio);
3585 
3586 	/*
3587 	 * Call into the pipeline to see if there is more work that
3588 	 * needs to be done. If there is work to be done it will be
3589 	 * dispatched to another taskq thread.
3590 	 */
3591 	zio_allocate_dispatch(zio->io_spa);
3592 }
3593 
3594 static int
3595 zio_done(zio_t *zio)
3596 {
3597 	spa_t *spa = zio->io_spa;
3598 	zio_t *lio = zio->io_logical;
3599 	blkptr_t *bp = zio->io_bp;
3600 	vdev_t *vd = zio->io_vd;
3601 	uint64_t psize = zio->io_size;
3602 	zio_t *pio, *pio_next;
3603 	metaslab_class_t *mc = spa_normal_class(spa);
3604 	zio_link_t *zl = NULL;
3605 
3606 	/*
3607 	 * If our children haven't all completed,
3608 	 * wait for them and then repeat this pipeline stage.
3609 	 */
3610 	if (zio_wait_for_children(zio, ZIO_CHILD_ALL_BITS, ZIO_WAIT_DONE)) {
3611 		return (ZIO_PIPELINE_STOP);
3612 	}
3613 
3614 	/*
3615 	 * If the allocation throttle is enabled, then update the accounting.
3616 	 * We only track child I/Os that are part of an allocating async
3617 	 * write. We must do this since the allocation is performed
3618 	 * by the logical I/O but the actual write is done by child I/Os.
3619 	 */
3620 	if (zio->io_flags & ZIO_FLAG_IO_ALLOCATING &&
3621 	    zio->io_child_type == ZIO_CHILD_VDEV) {
3622 		ASSERT(mc->mc_alloc_throttle_enabled);
3623 		zio_dva_throttle_done(zio);
3624 	}
3625 
3626 	/*
3627 	 * If the allocation throttle is enabled, verify that
3628 	 * we have decremented the refcounts for every I/O that was throttled.
3629 	 */
3630 	if (zio->io_flags & ZIO_FLAG_IO_ALLOCATING) {
3631 		ASSERT(zio->io_type == ZIO_TYPE_WRITE);
3632 		ASSERT(zio->io_priority == ZIO_PRIORITY_ASYNC_WRITE);
3633 		ASSERT(bp != NULL);
3634 		metaslab_group_alloc_verify(spa, zio->io_bp, zio);
3635 		VERIFY(refcount_not_held(&mc->mc_alloc_slots, zio));
3636 	}
3637 
3638 	for (int c = 0; c < ZIO_CHILD_TYPES; c++)
3639 		for (int w = 0; w < ZIO_WAIT_TYPES; w++)
3640 			ASSERT(zio->io_children[c][w] == 0);
3641 
3642 	if (bp != NULL && !BP_IS_EMBEDDED(bp)) {
3643 		ASSERT(bp->blk_pad[0] == 0);
3644 		ASSERT(bp->blk_pad[1] == 0);
3645 		ASSERT(bcmp(bp, &zio->io_bp_copy, sizeof (blkptr_t)) == 0 ||
3646 		    (bp == zio_unique_parent(zio)->io_bp));
3647 		if (zio->io_type == ZIO_TYPE_WRITE && !BP_IS_HOLE(bp) &&
3648 		    zio->io_bp_override == NULL &&
3649 		    !(zio->io_flags & ZIO_FLAG_IO_REPAIR)) {
3650 			ASSERT(!BP_SHOULD_BYTESWAP(bp));
3651 			ASSERT3U(zio->io_prop.zp_copies, <=, BP_GET_NDVAS(bp));
3652 			ASSERT(BP_COUNT_GANG(bp) == 0 ||
3653 			    (BP_COUNT_GANG(bp) == BP_GET_NDVAS(bp)));
3654 		}
3655 		if (zio->io_flags & ZIO_FLAG_NOPWRITE)
3656 			VERIFY(BP_EQUAL(bp, &zio->io_bp_orig));
3657 	}
3658 
3659 	/*
3660 	 * If there were child vdev/gang/ddt errors, they apply to us now.
3661 	 */
3662 	zio_inherit_child_errors(zio, ZIO_CHILD_VDEV);
3663 	zio_inherit_child_errors(zio, ZIO_CHILD_GANG);
3664 	zio_inherit_child_errors(zio, ZIO_CHILD_DDT);
3665 
3666 	/*
3667 	 * If the I/O on the transformed data was successful, generate any
3668 	 * checksum reports now while we still have the transformed data.
3669 	 */
3670 	if (zio->io_error == 0) {
3671 		while (zio->io_cksum_report != NULL) {
3672 			zio_cksum_report_t *zcr = zio->io_cksum_report;
3673 			uint64_t align = zcr->zcr_align;
3674 			uint64_t asize = P2ROUNDUP(psize, align);
3675 			char *abuf = NULL;
3676 			abd_t *adata = zio->io_abd;
3677 
3678 			if (asize != psize) {
3679 				adata = abd_alloc_linear(asize, B_TRUE);
3680 				abd_copy(adata, zio->io_abd, psize);
3681 				abd_zero_off(adata, psize, asize - psize);
3682 			}
3683 
3684 			if (adata != NULL)
3685 				abuf = abd_borrow_buf_copy(adata, asize);
3686 
3687 			zio->io_cksum_report = zcr->zcr_next;
3688 			zcr->zcr_next = NULL;
3689 			zcr->zcr_finish(zcr, abuf);
3690 			zfs_ereport_free_checksum(zcr);
3691 
3692 			if (adata != NULL)
3693 				abd_return_buf(adata, abuf, asize);
3694 
3695 			if (asize != psize)
3696 				abd_free(adata);
3697 		}
3698 	}
3699 
3700 	zio_pop_transforms(zio);	/* note: may set zio->io_error */
3701 
3702 	vdev_stat_update(zio, psize);
3703 
3704 	if (zio->io_error) {
3705 		/*
3706 		 * If this I/O is attached to a particular vdev,
3707 		 * generate an error message describing the I/O failure
3708 		 * at the block level.  We ignore these errors if the
3709 		 * device is currently unavailable.
3710 		 */
3711 		if (zio->io_error != ECKSUM && vd != NULL && !vdev_is_dead(vd))
3712 			zfs_ereport_post(FM_EREPORT_ZFS_IO, spa, vd, zio, 0, 0);
3713 
3714 		if ((zio->io_error == EIO || !(zio->io_flags &
3715 		    (ZIO_FLAG_SPECULATIVE | ZIO_FLAG_DONT_PROPAGATE))) &&
3716 		    zio == lio) {
3717 			/*
3718 			 * For logical I/O requests, tell the SPA to log the
3719 			 * error and generate a logical data ereport.
3720 			 */
3721 			spa_log_error(spa, zio);
3722 			zfs_ereport_post(FM_EREPORT_ZFS_DATA, spa, NULL, zio,
3723 			    0, 0);
3724 		}
3725 	}
3726 
3727 	if (zio->io_error && zio == lio) {
3728 		/*
3729 		 * Determine whether zio should be reexecuted.  This will
3730 		 * propagate all the way to the root via zio_notify_parent().
3731 		 */
3732 		ASSERT(vd == NULL && bp != NULL);
3733 		ASSERT(zio->io_child_type == ZIO_CHILD_LOGICAL);
3734 
3735 		if (IO_IS_ALLOCATING(zio) &&
3736 		    !(zio->io_flags & ZIO_FLAG_CANFAIL)) {
3737 			if (zio->io_error != ENOSPC)
3738 				zio->io_reexecute |= ZIO_REEXECUTE_NOW;
3739 			else
3740 				zio->io_reexecute |= ZIO_REEXECUTE_SUSPEND;
3741 		}
3742 
3743 		if ((zio->io_type == ZIO_TYPE_READ ||
3744 		    zio->io_type == ZIO_TYPE_FREE) &&
3745 		    !(zio->io_flags & ZIO_FLAG_SCAN_THREAD) &&
3746 		    zio->io_error == ENXIO &&
3747 		    spa_load_state(spa) == SPA_LOAD_NONE &&
3748 		    spa_get_failmode(spa) != ZIO_FAILURE_MODE_CONTINUE)
3749 			zio->io_reexecute |= ZIO_REEXECUTE_SUSPEND;
3750 
3751 		if (!(zio->io_flags & ZIO_FLAG_CANFAIL) && !zio->io_reexecute)
3752 			zio->io_reexecute |= ZIO_REEXECUTE_SUSPEND;
3753 
3754 		/*
3755 		 * Here is a possibly good place to attempt to do
3756 		 * either combinatorial reconstruction or error correction
3757 		 * based on checksums.  It also might be a good place
3758 		 * to send out preliminary ereports before we suspend
3759 		 * processing.
3760 		 */
3761 	}
3762 
3763 	/*
3764 	 * If there were logical child errors, they apply to us now.
3765 	 * We defer this until now to avoid conflating logical child
3766 	 * errors with errors that happened to the zio itself when
3767 	 * updating vdev stats and reporting FMA events above.
3768 	 */
3769 	zio_inherit_child_errors(zio, ZIO_CHILD_LOGICAL);
3770 
3771 	if ((zio->io_error || zio->io_reexecute) &&
3772 	    IO_IS_ALLOCATING(zio) && zio->io_gang_leader == zio &&
3773 	    !(zio->io_flags & (ZIO_FLAG_IO_REWRITE | ZIO_FLAG_NOPWRITE)))
3774 		zio_dva_unallocate(zio, zio->io_gang_tree, bp);
3775 
3776 	zio_gang_tree_free(&zio->io_gang_tree);
3777 
3778 	/*
3779 	 * Godfather I/Os should never suspend.
3780 	 */
3781 	if ((zio->io_flags & ZIO_FLAG_GODFATHER) &&
3782 	    (zio->io_reexecute & ZIO_REEXECUTE_SUSPEND))
3783 		zio->io_reexecute = 0;
3784 
3785 	if (zio->io_reexecute) {
3786 		/*
3787 		 * This is a logical I/O that wants to reexecute.
3788 		 *
3789 		 * Reexecute is top-down.  When an i/o fails, if it's not
3790 		 * the root, it simply notifies its parent and sticks around.
3791 		 * The parent, seeing that it still has children in zio_done(),
3792 		 * does the same.  This percolates all the way up to the root.
3793 		 * The root i/o will reexecute or suspend the entire tree.
3794 		 *
3795 		 * This approach ensures that zio_reexecute() honors
3796 		 * all the original i/o dependency relationships, e.g.
3797 		 * parents not executing until children are ready.
3798 		 */
3799 		ASSERT(zio->io_child_type == ZIO_CHILD_LOGICAL);
3800 
3801 		zio->io_gang_leader = NULL;
3802 
3803 		mutex_enter(&zio->io_lock);
3804 		zio->io_state[ZIO_WAIT_DONE] = 1;
3805 		mutex_exit(&zio->io_lock);
3806 
3807 		/*
3808 		 * "The Godfather" I/O monitors its children but is
3809 		 * not a true parent to them. It will track them through
3810 		 * the pipeline but severs its ties whenever they get into
3811 		 * trouble (e.g. suspended). This allows "The Godfather"
3812 		 * I/O to return status without blocking.
3813 		 */
3814 		zl = NULL;
3815 		for (pio = zio_walk_parents(zio, &zl); pio != NULL;
3816 		    pio = pio_next) {
3817 			zio_link_t *remove_zl = zl;
3818 			pio_next = zio_walk_parents(zio, &zl);
3819 
3820 			if ((pio->io_flags & ZIO_FLAG_GODFATHER) &&
3821 			    (zio->io_reexecute & ZIO_REEXECUTE_SUSPEND)) {
3822 				zio_remove_child(pio, zio, remove_zl);
3823 				zio_notify_parent(pio, zio, ZIO_WAIT_DONE);
3824 			}
3825 		}
3826 
3827 		if ((pio = zio_unique_parent(zio)) != NULL) {
3828 			/*
3829 			 * We're not a root i/o, so there's nothing to do
3830 			 * but notify our parent.  Don't propagate errors
3831 			 * upward since we haven't permanently failed yet.
3832 			 */
3833 			ASSERT(!(zio->io_flags & ZIO_FLAG_GODFATHER));
3834 			zio->io_flags |= ZIO_FLAG_DONT_PROPAGATE;
3835 			zio_notify_parent(pio, zio, ZIO_WAIT_DONE);
3836 		} else if (zio->io_reexecute & ZIO_REEXECUTE_SUSPEND) {
3837 			/*
3838 			 * We'd fail again if we reexecuted now, so suspend
3839 			 * until conditions improve (e.g. device comes online).
3840 			 */
3841 			zio_suspend(spa, zio);
3842 		} else {
3843 			/*
3844 			 * Reexecution is potentially a huge amount of work.
3845 			 * Hand it off to the otherwise-unused claim taskq.
3846 			 */
3847 			ASSERT(zio->io_tqent.tqent_next == NULL);
3848 			spa_taskq_dispatch_ent(spa, ZIO_TYPE_CLAIM,
3849 			    ZIO_TASKQ_ISSUE, (task_func_t *)zio_reexecute, zio,
3850 			    0, &zio->io_tqent);
3851 		}
3852 		return (ZIO_PIPELINE_STOP);
3853 	}
3854 
3855 	ASSERT(zio->io_child_count == 0);
3856 	ASSERT(zio->io_reexecute == 0);
3857 	ASSERT(zio->io_error == 0 || (zio->io_flags & ZIO_FLAG_CANFAIL));
3858 
3859 	/*
3860 	 * Report any checksum errors, since the I/O is complete.
3861 	 */
3862 	while (zio->io_cksum_report != NULL) {
3863 		zio_cksum_report_t *zcr = zio->io_cksum_report;
3864 		zio->io_cksum_report = zcr->zcr_next;
3865 		zcr->zcr_next = NULL;
3866 		zcr->zcr_finish(zcr, NULL);
3867 		zfs_ereport_free_checksum(zcr);
3868 	}
3869 
3870 	/*
3871 	 * It is the responsibility of the done callback to ensure that this
3872 	 * particular zio is no longer discoverable for adoption, and as
3873 	 * such, cannot acquire any new parents.
3874 	 */
3875 	if (zio->io_done)
3876 		zio->io_done(zio);
3877 
3878 	mutex_enter(&zio->io_lock);
3879 	zio->io_state[ZIO_WAIT_DONE] = 1;
3880 	mutex_exit(&zio->io_lock);
3881 
3882 	zl = NULL;
3883 	for (pio = zio_walk_parents(zio, &zl); pio != NULL; pio = pio_next) {
3884 		zio_link_t *remove_zl = zl;
3885 		pio_next = zio_walk_parents(zio, &zl);
3886 		zio_remove_child(pio, zio, remove_zl);
3887 		zio_notify_parent(pio, zio, ZIO_WAIT_DONE);
3888 	}
3889 
3890 	if (zio->io_waiter != NULL) {
3891 		mutex_enter(&zio->io_lock);
3892 		zio->io_executor = NULL;
3893 		cv_broadcast(&zio->io_cv);
3894 		mutex_exit(&zio->io_lock);
3895 	} else {
3896 		zio_destroy(zio);
3897 	}
3898 
3899 	return (ZIO_PIPELINE_STOP);
3900 }
3901 
3902 /*
3903  * ==========================================================================
3904  * I/O pipeline definition
3905  * ==========================================================================
3906  */
3907 static zio_pipe_stage_t *zio_pipeline[] = {
3908 	NULL,
3909 	zio_read_bp_init,
3910 	zio_write_bp_init,
3911 	zio_free_bp_init,
3912 	zio_issue_async,
3913 	zio_write_compress,
3914 	zio_checksum_generate,
3915 	zio_nop_write,
3916 	zio_ddt_read_start,
3917 	zio_ddt_read_done,
3918 	zio_ddt_write,
3919 	zio_ddt_free,
3920 	zio_gang_assemble,
3921 	zio_gang_issue,
3922 	zio_dva_throttle,
3923 	zio_dva_allocate,
3924 	zio_dva_free,
3925 	zio_dva_claim,
3926 	zio_ready,
3927 	zio_vdev_io_start,
3928 	zio_vdev_io_done,
3929 	zio_vdev_io_assess,
3930 	zio_checksum_verify,
3931 	zio_done
3932 };
3933 
3934 
3935 
3936 
3937 /*
3938  * Compare two zbookmark_phys_t's to see which we would reach first in a
3939  * pre-order traversal of the object tree.
3940  *
3941  * This is simple in every case aside from the meta-dnode object. For all other
3942  * objects, we traverse them in order (object 1 before object 2, and so on).
3943  * However, all of these objects are traversed while traversing object 0, since
3944  * the data it points to is the list of objects.  Thus, we need to convert to a
3945  * canonical representation so we can compare meta-dnode bookmarks to
3946  * non-meta-dnode bookmarks.
3947  *
3948  * We do this by calculating "equivalents" for each field of the zbookmark.
3949  * zbookmarks outside of the meta-dnode use their own object and level, and
3950  * calculate the level 0 equivalent (the first L0 blkid that is contained in the
3951  * blocks this bookmark refers to) by multiplying their blkid by their span
3952  * (the number of L0 blocks contained within one block at their level).
3953  * zbookmarks inside the meta-dnode calculate their object equivalent
3954  * (which is L0equiv * dnodes per data block), use 0 for their L0equiv, and use
3955  * level + 1<<31 (any value larger than a level could ever be) for their level.
3956  * This causes them to always compare before a bookmark in their object
3957  * equivalent, compare appropriately to bookmarks in other objects, and to
3958  * compare appropriately to other bookmarks in the meta-dnode.
3959  */
3960 int
3961 zbookmark_compare(uint16_t dbss1, uint8_t ibs1, uint16_t dbss2, uint8_t ibs2,
3962     const zbookmark_phys_t *zb1, const zbookmark_phys_t *zb2)
3963 {
3964 	/*
3965 	 * These variables represent the "equivalent" values for the zbookmark,
3966 	 * after converting zbookmarks inside the meta dnode to their
3967 	 * normal-object equivalents.
3968 	 */
3969 	uint64_t zb1obj, zb2obj;
3970 	uint64_t zb1L0, zb2L0;
3971 	uint64_t zb1level, zb2level;
3972 
3973 	if (zb1->zb_object == zb2->zb_object &&
3974 	    zb1->zb_level == zb2->zb_level &&
3975 	    zb1->zb_blkid == zb2->zb_blkid)
3976 		return (0);
3977 
3978 	/*
3979 	 * BP_SPANB calculates the span in blocks.
3980 	 */
3981 	zb1L0 = (zb1->zb_blkid) * BP_SPANB(ibs1, zb1->zb_level);
3982 	zb2L0 = (zb2->zb_blkid) * BP_SPANB(ibs2, zb2->zb_level);
3983 
3984 	if (zb1->zb_object == DMU_META_DNODE_OBJECT) {
3985 		zb1obj = zb1L0 * (dbss1 << (SPA_MINBLOCKSHIFT - DNODE_SHIFT));
3986 		zb1L0 = 0;
3987 		zb1level = zb1->zb_level + COMPARE_META_LEVEL;
3988 	} else {
3989 		zb1obj = zb1->zb_object;
3990 		zb1level = zb1->zb_level;
3991 	}
3992 
3993 	if (zb2->zb_object == DMU_META_DNODE_OBJECT) {
3994 		zb2obj = zb2L0 * (dbss2 << (SPA_MINBLOCKSHIFT - DNODE_SHIFT));
3995 		zb2L0 = 0;
3996 		zb2level = zb2->zb_level + COMPARE_META_LEVEL;
3997 	} else {
3998 		zb2obj = zb2->zb_object;
3999 		zb2level = zb2->zb_level;
4000 	}
4001 
4002 	/* Now that we have a canonical representation, do the comparison. */
4003 	if (zb1obj != zb2obj)
4004 		return (zb1obj < zb2obj ? -1 : 1);
4005 	else if (zb1L0 != zb2L0)
4006 		return (zb1L0 < zb2L0 ? -1 : 1);
4007 	else if (zb1level != zb2level)
4008 		return (zb1level > zb2level ? -1 : 1);
4009 	/*
4010 	 * This can (theoretically) happen if the bookmarks have the same object
4011 	 * and level, but different blkids, if the block sizes are not the same.
4012 	 * There is presently no way to change the indirect block sizes
4013 	 */
4014 	return (0);
4015 }
4016 
4017 /*
4018  *  This function checks the following: given that last_block is the place that
4019  *  our traversal stopped last time, does that guarantee that we've visited
4020  *  every node under subtree_root?  Therefore, we can't just use the raw output
4021  *  of zbookmark_compare.  We have to pass in a modified version of
4022  *  subtree_root; by incrementing the block id, and then checking whether
4023  *  last_block is before or equal to that, we can tell whether or not having
4024  *  visited last_block implies that all of subtree_root's children have been
4025  *  visited.
4026  */
4027 boolean_t
4028 zbookmark_subtree_completed(const dnode_phys_t *dnp,
4029     const zbookmark_phys_t *subtree_root, const zbookmark_phys_t *last_block)
4030 {
4031 	zbookmark_phys_t mod_zb = *subtree_root;
4032 	mod_zb.zb_blkid++;
4033 	ASSERT(last_block->zb_level == 0);
4034 
4035 	/* The objset_phys_t isn't before anything. */
4036 	if (dnp == NULL)
4037 		return (B_FALSE);
4038 
4039 	/*
4040 	 * We pass in 1ULL << (DNODE_BLOCK_SHIFT - SPA_MINBLOCKSHIFT) for the
4041 	 * data block size in sectors, because that variable is only used if
4042 	 * the bookmark refers to a block in the meta-dnode.  Since we don't
4043 	 * know without examining it what object it refers to, and there's no
4044 	 * harm in passing in this value in other cases, we always pass it in.
4045 	 *
4046 	 * We pass in 0 for the indirect block size shift because zb2 must be
4047 	 * level 0.  The indirect block size is only used to calculate the span
4048 	 * of the bookmark, but since the bookmark must be level 0, the span is
4049 	 * always 1, so the math works out.
4050 	 *
4051 	 * If you make changes to how the zbookmark_compare code works, be sure
4052 	 * to make sure that this code still works afterwards.
4053 	 */
4054 	return (zbookmark_compare(dnp->dn_datablkszsec, dnp->dn_indblkshift,
4055 	    1ULL << (DNODE_BLOCK_SHIFT - SPA_MINBLOCKSHIFT), 0, &mod_zb,
4056 	    last_block) <= 0);
4057 }
4058