xref: /illumos-gate/usr/src/uts/common/fs/zfs/zio.c (revision c717a56157ae0e6fca6a1e3689ae1edc385716a3)
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 2007 Sun Microsystems, Inc.  All rights reserved.
23  * Use is subject to license terms.
24  */
25 
26 #pragma ident	"%Z%%M%	%I%	%E% SMI"
27 
28 #include <sys/zfs_context.h>
29 #include <sys/fm/fs/zfs.h>
30 #include <sys/spa.h>
31 #include <sys/txg.h>
32 #include <sys/spa_impl.h>
33 #include <sys/vdev_impl.h>
34 #include <sys/zio_impl.h>
35 #include <sys/zio_compress.h>
36 #include <sys/zio_checksum.h>
37 
38 /*
39  * ==========================================================================
40  * I/O priority table
41  * ==========================================================================
42  */
43 uint8_t zio_priority_table[ZIO_PRIORITY_TABLE_SIZE] = {
44 	0,	/* ZIO_PRIORITY_NOW		*/
45 	0,	/* ZIO_PRIORITY_SYNC_READ	*/
46 	0,	/* ZIO_PRIORITY_SYNC_WRITE	*/
47 	6,	/* ZIO_PRIORITY_ASYNC_READ	*/
48 	4,	/* ZIO_PRIORITY_ASYNC_WRITE	*/
49 	4,	/* ZIO_PRIORITY_FREE		*/
50 	0,	/* ZIO_PRIORITY_CACHE_FILL	*/
51 	0,	/* ZIO_PRIORITY_LOG_WRITE	*/
52 	10,	/* ZIO_PRIORITY_RESILVER	*/
53 	20,	/* ZIO_PRIORITY_SCRUB		*/
54 };
55 
56 /*
57  * ==========================================================================
58  * I/O type descriptions
59  * ==========================================================================
60  */
61 char *zio_type_name[ZIO_TYPES] = {
62 	"null", "read", "write", "free", "claim", "ioctl" };
63 
64 /* At or above this size, force gang blocking - for testing */
65 uint64_t zio_gang_bang = SPA_MAXBLOCKSIZE + 1;
66 
67 typedef struct zio_sync_pass {
68 	int	zp_defer_free;		/* defer frees after this pass */
69 	int	zp_dontcompress;	/* don't compress after this pass */
70 	int	zp_rewrite;		/* rewrite new bps after this pass */
71 } zio_sync_pass_t;
72 
73 zio_sync_pass_t zio_sync_pass = {
74 	1,	/* zp_defer_free */
75 	4,	/* zp_dontcompress */
76 	1,	/* zp_rewrite */
77 };
78 
79 /*
80  * ==========================================================================
81  * I/O kmem caches
82  * ==========================================================================
83  */
84 kmem_cache_t *zio_buf_cache[SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT];
85 kmem_cache_t *zio_data_buf_cache[SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT];
86 
87 #ifdef _KERNEL
88 extern vmem_t *zio_alloc_arena;
89 #endif
90 
91 void
92 zio_init(void)
93 {
94 	size_t c;
95 	vmem_t *data_alloc_arena = NULL;
96 
97 #ifdef _KERNEL
98 	data_alloc_arena = zio_alloc_arena;
99 #endif
100 
101 	/*
102 	 * For small buffers, we want a cache for each multiple of
103 	 * SPA_MINBLOCKSIZE.  For medium-size buffers, we want a cache
104 	 * for each quarter-power of 2.  For large buffers, we want
105 	 * a cache for each multiple of PAGESIZE.
106 	 */
107 	for (c = 0; c < SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT; c++) {
108 		size_t size = (c + 1) << SPA_MINBLOCKSHIFT;
109 		size_t p2 = size;
110 		size_t align = 0;
111 
112 		while (p2 & (p2 - 1))
113 			p2 &= p2 - 1;
114 
115 		if (size <= 4 * SPA_MINBLOCKSIZE) {
116 			align = SPA_MINBLOCKSIZE;
117 		} else if (P2PHASE(size, PAGESIZE) == 0) {
118 			align = PAGESIZE;
119 		} else if (P2PHASE(size, p2 >> 2) == 0) {
120 			align = p2 >> 2;
121 		}
122 
123 		if (align != 0) {
124 			char name[36];
125 			(void) sprintf(name, "zio_buf_%lu", (ulong_t)size);
126 			zio_buf_cache[c] = kmem_cache_create(name, size,
127 			    align, NULL, NULL, NULL, NULL, NULL, KMC_NODEBUG);
128 
129 			(void) sprintf(name, "zio_data_buf_%lu", (ulong_t)size);
130 			zio_data_buf_cache[c] = kmem_cache_create(name, size,
131 			    align, NULL, NULL, NULL, NULL, data_alloc_arena,
132 			    KMC_NODEBUG);
133 
134 			dprintf("creating cache for size %5lx align %5lx\n",
135 			    size, align);
136 		}
137 	}
138 
139 	while (--c != 0) {
140 		ASSERT(zio_buf_cache[c] != NULL);
141 		if (zio_buf_cache[c - 1] == NULL)
142 			zio_buf_cache[c - 1] = zio_buf_cache[c];
143 
144 		ASSERT(zio_data_buf_cache[c] != NULL);
145 		if (zio_data_buf_cache[c - 1] == NULL)
146 			zio_data_buf_cache[c - 1] = zio_data_buf_cache[c];
147 	}
148 
149 	zio_inject_init();
150 }
151 
152 void
153 zio_fini(void)
154 {
155 	size_t c;
156 	kmem_cache_t *last_cache = NULL;
157 	kmem_cache_t *last_data_cache = NULL;
158 
159 	for (c = 0; c < SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT; c++) {
160 		if (zio_buf_cache[c] != last_cache) {
161 			last_cache = zio_buf_cache[c];
162 			kmem_cache_destroy(zio_buf_cache[c]);
163 		}
164 		zio_buf_cache[c] = NULL;
165 
166 		if (zio_data_buf_cache[c] != last_data_cache) {
167 			last_data_cache = zio_data_buf_cache[c];
168 			kmem_cache_destroy(zio_data_buf_cache[c]);
169 		}
170 		zio_data_buf_cache[c] = NULL;
171 	}
172 
173 	zio_inject_fini();
174 }
175 
176 /*
177  * ==========================================================================
178  * Allocate and free I/O buffers
179  * ==========================================================================
180  */
181 
182 /*
183  * Use zio_buf_alloc to allocate ZFS metadata.  This data will appear in a
184  * crashdump if the kernel panics, so use it judiciously.  Obviously, it's
185  * useful to inspect ZFS metadata, but if possible, we should avoid keeping
186  * excess / transient data in-core during a crashdump.
187  */
188 void *
189 zio_buf_alloc(size_t size)
190 {
191 	size_t c = (size - 1) >> SPA_MINBLOCKSHIFT;
192 
193 	ASSERT(c < SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT);
194 
195 	return (kmem_cache_alloc(zio_buf_cache[c], KM_SLEEP));
196 }
197 
198 /*
199  * Use zio_data_buf_alloc to allocate data.  The data will not appear in a
200  * crashdump if the kernel panics.  This exists so that we will limit the amount
201  * of ZFS data that shows up in a kernel crashdump.  (Thus reducing the amount
202  * of kernel heap dumped to disk when the kernel panics)
203  */
204 void *
205 zio_data_buf_alloc(size_t size)
206 {
207 	size_t c = (size - 1) >> SPA_MINBLOCKSHIFT;
208 
209 	ASSERT(c < SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT);
210 
211 	return (kmem_cache_alloc(zio_data_buf_cache[c], KM_SLEEP));
212 }
213 
214 void
215 zio_buf_free(void *buf, size_t size)
216 {
217 	size_t c = (size - 1) >> SPA_MINBLOCKSHIFT;
218 
219 	ASSERT(c < SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT);
220 
221 	kmem_cache_free(zio_buf_cache[c], buf);
222 }
223 
224 void
225 zio_data_buf_free(void *buf, size_t size)
226 {
227 	size_t c = (size - 1) >> SPA_MINBLOCKSHIFT;
228 
229 	ASSERT(c < SPA_MAXBLOCKSIZE >> SPA_MINBLOCKSHIFT);
230 
231 	kmem_cache_free(zio_data_buf_cache[c], buf);
232 }
233 
234 /*
235  * ==========================================================================
236  * Push and pop I/O transform buffers
237  * ==========================================================================
238  */
239 static void
240 zio_push_transform(zio_t *zio, void *data, uint64_t size, uint64_t bufsize)
241 {
242 	zio_transform_t *zt = kmem_alloc(sizeof (zio_transform_t), KM_SLEEP);
243 
244 	zt->zt_data = data;
245 	zt->zt_size = size;
246 	zt->zt_bufsize = bufsize;
247 
248 	zt->zt_next = zio->io_transform_stack;
249 	zio->io_transform_stack = zt;
250 
251 	zio->io_data = data;
252 	zio->io_size = size;
253 }
254 
255 static void
256 zio_pop_transform(zio_t *zio, void **data, uint64_t *size, uint64_t *bufsize)
257 {
258 	zio_transform_t *zt = zio->io_transform_stack;
259 
260 	*data = zt->zt_data;
261 	*size = zt->zt_size;
262 	*bufsize = zt->zt_bufsize;
263 
264 	zio->io_transform_stack = zt->zt_next;
265 	kmem_free(zt, sizeof (zio_transform_t));
266 
267 	if ((zt = zio->io_transform_stack) != NULL) {
268 		zio->io_data = zt->zt_data;
269 		zio->io_size = zt->zt_size;
270 	}
271 }
272 
273 static void
274 zio_clear_transform_stack(zio_t *zio)
275 {
276 	void *data;
277 	uint64_t size, bufsize;
278 
279 	ASSERT(zio->io_transform_stack != NULL);
280 
281 	zio_pop_transform(zio, &data, &size, &bufsize);
282 	while (zio->io_transform_stack != NULL) {
283 		zio_buf_free(data, bufsize);
284 		zio_pop_transform(zio, &data, &size, &bufsize);
285 	}
286 }
287 
288 /*
289  * ==========================================================================
290  * Create the various types of I/O (read, write, free)
291  * ==========================================================================
292  */
293 static zio_t *
294 zio_create(zio_t *pio, spa_t *spa, uint64_t txg, blkptr_t *bp,
295     void *data, uint64_t size, zio_done_func_t *done, void *private,
296     zio_type_t type, int priority, int flags, uint8_t stage, uint32_t pipeline)
297 {
298 	zio_t *zio;
299 
300 	ASSERT3U(size, <=, SPA_MAXBLOCKSIZE);
301 	ASSERT(P2PHASE(size, SPA_MINBLOCKSIZE) == 0);
302 
303 	zio = kmem_zalloc(sizeof (zio_t), KM_SLEEP);
304 	zio->io_parent = pio;
305 	zio->io_spa = spa;
306 	zio->io_txg = txg;
307 	if (bp != NULL) {
308 		zio->io_bp = bp;
309 		zio->io_bp_copy = *bp;
310 		zio->io_bp_orig = *bp;
311 	}
312 	zio->io_done = done;
313 	zio->io_private = private;
314 	zio->io_type = type;
315 	zio->io_priority = priority;
316 	zio->io_stage = stage;
317 	zio->io_pipeline = pipeline;
318 	zio->io_async_stages = ZIO_ASYNC_PIPELINE_STAGES;
319 	zio->io_timestamp = lbolt64;
320 	zio->io_flags = flags;
321 	mutex_init(&zio->io_lock, NULL, MUTEX_DEFAULT, NULL);
322 	zio_push_transform(zio, data, size, size);
323 
324 	/*
325 	 * Note on config lock:
326 	 *
327 	 * If CONFIG_HELD is set, then the caller already has the config
328 	 * lock, so we don't need it for this io.
329 	 *
330 	 * We set CONFIG_GRABBED to indicate that we have grabbed the
331 	 * config lock on behalf of this io, so it should be released
332 	 * in zio_done.
333 	 *
334 	 * Unless CONFIG_HELD is set, we will grab the config lock for
335 	 * any top-level (parent-less) io, *except* NULL top-level ios.
336 	 * The NULL top-level ios rarely have any children, so we delay
337 	 * grabbing the lock until the first child is added (but it is
338 	 * still grabbed on behalf of the top-level i/o, so additional
339 	 * children don't need to also grab it).  This greatly reduces
340 	 * contention on the config lock.
341 	 */
342 	if (pio == NULL) {
343 		if (type != ZIO_TYPE_NULL &&
344 		    !(flags & ZIO_FLAG_CONFIG_HELD)) {
345 			spa_config_enter(zio->io_spa, RW_READER, zio);
346 			zio->io_flags |= ZIO_FLAG_CONFIG_GRABBED;
347 		}
348 		zio->io_root = zio;
349 	} else {
350 		zio->io_root = pio->io_root;
351 		if (!(flags & ZIO_FLAG_NOBOOKMARK))
352 			zio->io_logical = pio->io_logical;
353 		mutex_enter(&pio->io_lock);
354 		if (pio->io_parent == NULL &&
355 		    pio->io_type == ZIO_TYPE_NULL &&
356 		    !(pio->io_flags & ZIO_FLAG_CONFIG_GRABBED) &&
357 		    !(pio->io_flags & ZIO_FLAG_CONFIG_HELD)) {
358 			pio->io_flags |= ZIO_FLAG_CONFIG_GRABBED;
359 			spa_config_enter(zio->io_spa, RW_READER, pio);
360 		}
361 		if (stage < ZIO_STAGE_READY)
362 			pio->io_children_notready++;
363 		pio->io_children_notdone++;
364 		zio->io_sibling_next = pio->io_child;
365 		zio->io_sibling_prev = NULL;
366 		if (pio->io_child != NULL)
367 			pio->io_child->io_sibling_prev = zio;
368 		pio->io_child = zio;
369 		zio->io_ndvas = pio->io_ndvas;
370 		mutex_exit(&pio->io_lock);
371 	}
372 
373 	return (zio);
374 }
375 
376 zio_t *
377 zio_null(zio_t *pio, spa_t *spa, zio_done_func_t *done, void *private,
378 	int flags)
379 {
380 	zio_t *zio;
381 
382 	zio = zio_create(pio, spa, 0, NULL, NULL, 0, done, private,
383 	    ZIO_TYPE_NULL, ZIO_PRIORITY_NOW, flags, ZIO_STAGE_OPEN,
384 	    ZIO_WAIT_FOR_CHILDREN_PIPELINE);
385 
386 	return (zio);
387 }
388 
389 zio_t *
390 zio_root(spa_t *spa, zio_done_func_t *done, void *private, int flags)
391 {
392 	return (zio_null(NULL, spa, done, private, flags));
393 }
394 
395 zio_t *
396 zio_read(zio_t *pio, spa_t *spa, blkptr_t *bp, void *data,
397     uint64_t size, zio_done_func_t *done, void *private,
398     int priority, int flags, zbookmark_t *zb)
399 {
400 	zio_t *zio;
401 
402 	ASSERT3U(size, ==, BP_GET_LSIZE(bp));
403 
404 	zio = zio_create(pio, spa, bp->blk_birth, bp, data, size, done, private,
405 	    ZIO_TYPE_READ, priority, flags | ZIO_FLAG_USER,
406 	    ZIO_STAGE_OPEN, ZIO_READ_PIPELINE);
407 	zio->io_bookmark = *zb;
408 
409 	zio->io_logical = zio;
410 
411 	/*
412 	 * Work off our copy of the bp so the caller can free it.
413 	 */
414 	zio->io_bp = &zio->io_bp_copy;
415 
416 	if (BP_GET_COMPRESS(bp) != ZIO_COMPRESS_OFF) {
417 		uint64_t csize = BP_GET_PSIZE(bp);
418 		void *cbuf = zio_buf_alloc(csize);
419 
420 		zio_push_transform(zio, cbuf, csize, csize);
421 		zio->io_pipeline |= 1U << ZIO_STAGE_READ_DECOMPRESS;
422 	}
423 
424 	if (BP_IS_GANG(bp)) {
425 		uint64_t gsize = SPA_GANGBLOCKSIZE;
426 		void *gbuf = zio_buf_alloc(gsize);
427 
428 		zio_push_transform(zio, gbuf, gsize, gsize);
429 		zio->io_pipeline |= 1U << ZIO_STAGE_READ_GANG_MEMBERS;
430 	}
431 
432 	return (zio);
433 }
434 
435 zio_t *
436 zio_write(zio_t *pio, spa_t *spa, int checksum, int compress, int ncopies,
437     uint64_t txg, blkptr_t *bp, void *data, uint64_t size,
438     zio_done_func_t *ready, zio_done_func_t *done, void *private, int priority,
439     int flags, zbookmark_t *zb)
440 {
441 	zio_t *zio;
442 
443 	ASSERT(checksum >= ZIO_CHECKSUM_OFF &&
444 	    checksum < ZIO_CHECKSUM_FUNCTIONS);
445 
446 	ASSERT(compress >= ZIO_COMPRESS_OFF &&
447 	    compress < ZIO_COMPRESS_FUNCTIONS);
448 
449 	zio = zio_create(pio, spa, txg, bp, data, size, done, private,
450 	    ZIO_TYPE_WRITE, priority, flags | ZIO_FLAG_USER,
451 	    ZIO_STAGE_OPEN, ZIO_WRITE_PIPELINE);
452 
453 	zio->io_ready = ready;
454 
455 	zio->io_bookmark = *zb;
456 
457 	zio->io_logical = zio;
458 
459 	zio->io_checksum = checksum;
460 	zio->io_compress = compress;
461 	zio->io_ndvas = ncopies;
462 
463 	if (compress != ZIO_COMPRESS_OFF)
464 		zio->io_async_stages |= 1U << ZIO_STAGE_WRITE_COMPRESS;
465 
466 	if (bp->blk_birth != txg) {
467 		/* XXX the bp usually (always?) gets re-zeroed later */
468 		BP_ZERO(bp);
469 		BP_SET_LSIZE(bp, size);
470 		BP_SET_PSIZE(bp, size);
471 	} else {
472 		/* Make sure someone doesn't change their mind on overwrites */
473 		ASSERT(MIN(zio->io_ndvas + BP_IS_GANG(bp),
474 		    spa_max_replication(spa)) == BP_GET_NDVAS(bp));
475 	}
476 
477 	return (zio);
478 }
479 
480 zio_t *
481 zio_rewrite(zio_t *pio, spa_t *spa, int checksum,
482     uint64_t txg, blkptr_t *bp, void *data, uint64_t size,
483     zio_done_func_t *done, void *private, int priority, int flags,
484     zbookmark_t *zb)
485 {
486 	zio_t *zio;
487 
488 	zio = zio_create(pio, spa, txg, bp, data, size, done, private,
489 	    ZIO_TYPE_WRITE, priority, flags | ZIO_FLAG_USER,
490 	    ZIO_STAGE_OPEN, ZIO_REWRITE_PIPELINE);
491 
492 	zio->io_bookmark = *zb;
493 	zio->io_checksum = checksum;
494 	zio->io_compress = ZIO_COMPRESS_OFF;
495 
496 	if (pio != NULL)
497 		ASSERT3U(zio->io_ndvas, <=, BP_GET_NDVAS(bp));
498 
499 	return (zio);
500 }
501 
502 static zio_t *
503 zio_write_allocate(zio_t *pio, spa_t *spa, int checksum,
504     uint64_t txg, blkptr_t *bp, void *data, uint64_t size,
505     zio_done_func_t *done, void *private, int priority, int flags)
506 {
507 	zio_t *zio;
508 
509 	BP_ZERO(bp);
510 	BP_SET_LSIZE(bp, size);
511 	BP_SET_PSIZE(bp, size);
512 	BP_SET_COMPRESS(bp, ZIO_COMPRESS_OFF);
513 
514 	zio = zio_create(pio, spa, txg, bp, data, size, done, private,
515 	    ZIO_TYPE_WRITE, priority, flags,
516 	    ZIO_STAGE_OPEN, ZIO_WRITE_ALLOCATE_PIPELINE);
517 
518 	zio->io_checksum = checksum;
519 	zio->io_compress = ZIO_COMPRESS_OFF;
520 
521 	return (zio);
522 }
523 
524 zio_t *
525 zio_free(zio_t *pio, spa_t *spa, uint64_t txg, blkptr_t *bp,
526     zio_done_func_t *done, void *private)
527 {
528 	zio_t *zio;
529 
530 	ASSERT(!BP_IS_HOLE(bp));
531 
532 	if (txg == spa->spa_syncing_txg &&
533 	    spa->spa_sync_pass > zio_sync_pass.zp_defer_free) {
534 		bplist_enqueue_deferred(&spa->spa_sync_bplist, bp);
535 		return (zio_null(pio, spa, NULL, NULL, 0));
536 	}
537 
538 	zio = zio_create(pio, spa, txg, bp, NULL, 0, done, private,
539 	    ZIO_TYPE_FREE, ZIO_PRIORITY_FREE, ZIO_FLAG_USER,
540 	    ZIO_STAGE_OPEN, ZIO_FREE_PIPELINE);
541 
542 	zio->io_bp = &zio->io_bp_copy;
543 
544 	return (zio);
545 }
546 
547 zio_t *
548 zio_claim(zio_t *pio, spa_t *spa, uint64_t txg, blkptr_t *bp,
549     zio_done_func_t *done, void *private)
550 {
551 	zio_t *zio;
552 
553 	/*
554 	 * A claim is an allocation of a specific block.  Claims are needed
555 	 * to support immediate writes in the intent log.  The issue is that
556 	 * immediate writes contain committed data, but in a txg that was
557 	 * *not* committed.  Upon opening the pool after an unclean shutdown,
558 	 * the intent log claims all blocks that contain immediate write data
559 	 * so that the SPA knows they're in use.
560 	 *
561 	 * All claims *must* be resolved in the first txg -- before the SPA
562 	 * starts allocating blocks -- so that nothing is allocated twice.
563 	 */
564 	ASSERT3U(spa->spa_uberblock.ub_rootbp.blk_birth, <, spa_first_txg(spa));
565 	ASSERT3U(spa_first_txg(spa), <=, txg);
566 
567 	zio = zio_create(pio, spa, txg, bp, NULL, 0, done, private,
568 	    ZIO_TYPE_CLAIM, ZIO_PRIORITY_NOW, 0,
569 	    ZIO_STAGE_OPEN, ZIO_CLAIM_PIPELINE);
570 
571 	zio->io_bp = &zio->io_bp_copy;
572 
573 	return (zio);
574 }
575 
576 zio_t *
577 zio_ioctl(zio_t *pio, spa_t *spa, vdev_t *vd, int cmd,
578     zio_done_func_t *done, void *private, int priority, int flags)
579 {
580 	zio_t *zio;
581 	int c;
582 
583 	if (vd->vdev_children == 0) {
584 		zio = zio_create(pio, spa, 0, NULL, NULL, 0, done, private,
585 		    ZIO_TYPE_IOCTL, priority, flags,
586 		    ZIO_STAGE_OPEN, ZIO_IOCTL_PIPELINE);
587 
588 		zio->io_vd = vd;
589 		zio->io_cmd = cmd;
590 	} else {
591 		zio = zio_null(pio, spa, NULL, NULL, flags);
592 
593 		for (c = 0; c < vd->vdev_children; c++)
594 			zio_nowait(zio_ioctl(zio, spa, vd->vdev_child[c], cmd,
595 			    done, private, priority, flags));
596 	}
597 
598 	return (zio);
599 }
600 
601 static void
602 zio_phys_bp_init(vdev_t *vd, blkptr_t *bp, uint64_t offset, uint64_t size,
603     int checksum)
604 {
605 	ASSERT(vd->vdev_children == 0);
606 
607 	ASSERT(size <= SPA_MAXBLOCKSIZE);
608 	ASSERT(P2PHASE(size, SPA_MINBLOCKSIZE) == 0);
609 	ASSERT(P2PHASE(offset, SPA_MINBLOCKSIZE) == 0);
610 
611 	ASSERT(offset + size <= VDEV_LABEL_START_SIZE ||
612 	    offset >= vd->vdev_psize - VDEV_LABEL_END_SIZE);
613 	ASSERT3U(offset + size, <=, vd->vdev_psize);
614 
615 	BP_ZERO(bp);
616 
617 	BP_SET_LSIZE(bp, size);
618 	BP_SET_PSIZE(bp, size);
619 
620 	BP_SET_CHECKSUM(bp, checksum);
621 	BP_SET_COMPRESS(bp, ZIO_COMPRESS_OFF);
622 	BP_SET_BYTEORDER(bp, ZFS_HOST_BYTEORDER);
623 
624 	if (checksum != ZIO_CHECKSUM_OFF)
625 		ZIO_SET_CHECKSUM(&bp->blk_cksum, offset, 0, 0, 0);
626 }
627 
628 zio_t *
629 zio_read_phys(zio_t *pio, vdev_t *vd, uint64_t offset, uint64_t size,
630     void *data, int checksum, zio_done_func_t *done, void *private,
631     int priority, int flags)
632 {
633 	zio_t *zio;
634 	blkptr_t blk;
635 
636 	zio_phys_bp_init(vd, &blk, offset, size, checksum);
637 
638 	zio = zio_create(pio, vd->vdev_spa, 0, &blk, data, size, done, private,
639 	    ZIO_TYPE_READ, priority, flags | ZIO_FLAG_PHYSICAL,
640 	    ZIO_STAGE_OPEN, ZIO_READ_PHYS_PIPELINE);
641 
642 	zio->io_vd = vd;
643 	zio->io_offset = offset;
644 
645 	/*
646 	 * Work off our copy of the bp so the caller can free it.
647 	 */
648 	zio->io_bp = &zio->io_bp_copy;
649 
650 	return (zio);
651 }
652 
653 zio_t *
654 zio_write_phys(zio_t *pio, vdev_t *vd, uint64_t offset, uint64_t size,
655     void *data, int checksum, zio_done_func_t *done, void *private,
656     int priority, int flags)
657 {
658 	zio_block_tail_t *zbt;
659 	void *wbuf;
660 	zio_t *zio;
661 	blkptr_t blk;
662 
663 	zio_phys_bp_init(vd, &blk, offset, size, checksum);
664 
665 	zio = zio_create(pio, vd->vdev_spa, 0, &blk, data, size, done, private,
666 	    ZIO_TYPE_WRITE, priority, flags | ZIO_FLAG_PHYSICAL,
667 	    ZIO_STAGE_OPEN, ZIO_WRITE_PHYS_PIPELINE);
668 
669 	zio->io_vd = vd;
670 	zio->io_offset = offset;
671 
672 	zio->io_bp = &zio->io_bp_copy;
673 	zio->io_checksum = checksum;
674 
675 	if (zio_checksum_table[checksum].ci_zbt) {
676 		/*
677 		 * zbt checksums are necessarily destructive -- they modify
678 		 * one word of the write buffer to hold the verifier/checksum.
679 		 * Therefore, we must make a local copy in case the data is
680 		 * being written to multiple places.
681 		 */
682 		wbuf = zio_buf_alloc(size);
683 		bcopy(data, wbuf, size);
684 		zio_push_transform(zio, wbuf, size, size);
685 
686 		zbt = (zio_block_tail_t *)((char *)wbuf + size) - 1;
687 		zbt->zbt_cksum = blk.blk_cksum;
688 	}
689 
690 	return (zio);
691 }
692 
693 /*
694  * Create a child I/O to do some work for us.  It has no associated bp.
695  */
696 zio_t *
697 zio_vdev_child_io(zio_t *zio, blkptr_t *bp, vdev_t *vd, uint64_t offset,
698 	void *data, uint64_t size, int type, int priority, int flags,
699 	zio_done_func_t *done, void *private)
700 {
701 	uint32_t pipeline = ZIO_VDEV_CHILD_PIPELINE;
702 	zio_t *cio;
703 
704 	if (type == ZIO_TYPE_READ && bp != NULL) {
705 		/*
706 		 * If we have the bp, then the child should perform the
707 		 * checksum and the parent need not.  This pushes error
708 		 * detection as close to the leaves as possible and
709 		 * eliminates redundant checksums in the interior nodes.
710 		 */
711 		pipeline |= 1U << ZIO_STAGE_CHECKSUM_VERIFY;
712 		zio->io_pipeline &= ~(1U << ZIO_STAGE_CHECKSUM_VERIFY);
713 	}
714 
715 	cio = zio_create(zio, zio->io_spa, zio->io_txg, bp, data, size,
716 	    done, private, type, priority,
717 	    (zio->io_flags & ZIO_FLAG_VDEV_INHERIT) | ZIO_FLAG_CANFAIL | flags,
718 	    ZIO_STAGE_VDEV_IO_START - 1, pipeline);
719 
720 	cio->io_vd = vd;
721 	cio->io_offset = offset;
722 
723 	return (cio);
724 }
725 
726 /*
727  * ==========================================================================
728  * Initiate I/O, either sync or async
729  * ==========================================================================
730  */
731 int
732 zio_wait(zio_t *zio)
733 {
734 	int error;
735 
736 	ASSERT(zio->io_stage == ZIO_STAGE_OPEN);
737 
738 	zio->io_waiter = curthread;
739 
740 	zio_next_stage_async(zio);
741 
742 	mutex_enter(&zio->io_lock);
743 	while (zio->io_stalled != ZIO_STAGE_DONE)
744 		cv_wait(&zio->io_cv, &zio->io_lock);
745 	mutex_exit(&zio->io_lock);
746 
747 	error = zio->io_error;
748 	mutex_destroy(&zio->io_lock);
749 	kmem_free(zio, sizeof (zio_t));
750 
751 	return (error);
752 }
753 
754 void
755 zio_nowait(zio_t *zio)
756 {
757 	zio_next_stage_async(zio);
758 }
759 
760 /*
761  * ==========================================================================
762  * I/O pipeline interlocks: parent/child dependency scoreboarding
763  * ==========================================================================
764  */
765 static void
766 zio_wait_for_children(zio_t *zio, uint32_t stage, uint64_t *countp)
767 {
768 	mutex_enter(&zio->io_lock);
769 	if (*countp == 0) {
770 		ASSERT(zio->io_stalled == 0);
771 		mutex_exit(&zio->io_lock);
772 		zio_next_stage(zio);
773 	} else {
774 		zio->io_stalled = stage;
775 		mutex_exit(&zio->io_lock);
776 	}
777 }
778 
779 static void
780 zio_notify_parent(zio_t *zio, uint32_t stage, uint64_t *countp)
781 {
782 	zio_t *pio = zio->io_parent;
783 
784 	mutex_enter(&pio->io_lock);
785 	if (pio->io_error == 0 && !(zio->io_flags & ZIO_FLAG_DONT_PROPAGATE))
786 		pio->io_error = zio->io_error;
787 	if (--*countp == 0 && pio->io_stalled == stage) {
788 		pio->io_stalled = 0;
789 		mutex_exit(&pio->io_lock);
790 		zio_next_stage_async(pio);
791 	} else {
792 		mutex_exit(&pio->io_lock);
793 	}
794 }
795 
796 static void
797 zio_wait_children_ready(zio_t *zio)
798 {
799 	zio_wait_for_children(zio, ZIO_STAGE_WAIT_CHILDREN_READY,
800 	    &zio->io_children_notready);
801 }
802 
803 void
804 zio_wait_children_done(zio_t *zio)
805 {
806 	zio_wait_for_children(zio, ZIO_STAGE_WAIT_CHILDREN_DONE,
807 	    &zio->io_children_notdone);
808 }
809 
810 static void
811 zio_ready(zio_t *zio)
812 {
813 	zio_t *pio = zio->io_parent;
814 
815 	if (zio->io_ready)
816 		zio->io_ready(zio);
817 
818 	if (pio != NULL)
819 		zio_notify_parent(zio, ZIO_STAGE_WAIT_CHILDREN_READY,
820 		    &pio->io_children_notready);
821 
822 	if (zio->io_bp)
823 		zio->io_bp_copy = *zio->io_bp;
824 
825 	zio_next_stage(zio);
826 }
827 
828 static void
829 zio_done(zio_t *zio)
830 {
831 	zio_t *pio = zio->io_parent;
832 	spa_t *spa = zio->io_spa;
833 	blkptr_t *bp = zio->io_bp;
834 	vdev_t *vd = zio->io_vd;
835 
836 	ASSERT(zio->io_children_notready == 0);
837 	ASSERT(zio->io_children_notdone == 0);
838 
839 	if (bp != NULL) {
840 		ASSERT(bp->blk_pad[0] == 0);
841 		ASSERT(bp->blk_pad[1] == 0);
842 		ASSERT(bp->blk_pad[2] == 0);
843 		ASSERT(bcmp(bp, &zio->io_bp_copy, sizeof (blkptr_t)) == 0);
844 		if (zio->io_type == ZIO_TYPE_WRITE && !BP_IS_HOLE(bp) &&
845 		    !(zio->io_flags & ZIO_FLAG_IO_REPAIR)) {
846 			ASSERT(!BP_SHOULD_BYTESWAP(bp));
847 			if (zio->io_ndvas != 0)
848 				ASSERT3U(zio->io_ndvas, <=, BP_GET_NDVAS(bp));
849 			ASSERT(BP_COUNT_GANG(bp) == 0 ||
850 			    (BP_COUNT_GANG(bp) == BP_GET_NDVAS(bp)));
851 		}
852 	}
853 
854 	if (vd != NULL)
855 		vdev_stat_update(zio);
856 
857 	if (zio->io_error) {
858 		/*
859 		 * If this I/O is attached to a particular vdev,
860 		 * generate an error message describing the I/O failure
861 		 * at the block level.  We ignore these errors if the
862 		 * device is currently unavailable.
863 		 */
864 		if (zio->io_error != ECKSUM && vd != NULL && !vdev_is_dead(vd))
865 			zfs_ereport_post(FM_EREPORT_ZFS_IO,
866 			    zio->io_spa, vd, zio, 0, 0);
867 
868 		if ((zio->io_error == EIO ||
869 		    !(zio->io_flags & ZIO_FLAG_SPECULATIVE)) &&
870 		    zio->io_logical == zio) {
871 			/*
872 			 * For root I/O requests, tell the SPA to log the error
873 			 * appropriately.  Also, generate a logical data
874 			 * ereport.
875 			 */
876 			spa_log_error(zio->io_spa, zio);
877 
878 			zfs_ereport_post(FM_EREPORT_ZFS_DATA,
879 			    zio->io_spa, NULL, zio, 0, 0);
880 		}
881 
882 		/*
883 		 * For I/O requests that cannot fail, panic appropriately.
884 		 */
885 		if (!(zio->io_flags & ZIO_FLAG_CANFAIL)) {
886 			char *blkbuf;
887 
888 			blkbuf = kmem_alloc(BP_SPRINTF_LEN, KM_NOSLEEP);
889 			if (blkbuf) {
890 				sprintf_blkptr(blkbuf, BP_SPRINTF_LEN,
891 				    bp ? bp : &zio->io_bp_copy);
892 			}
893 			panic("ZFS: %s (%s on %s off %llx: zio %p %s): error "
894 			    "%d", zio->io_error == ECKSUM ?
895 			    "bad checksum" : "I/O failure",
896 			    zio_type_name[zio->io_type],
897 			    vdev_description(vd),
898 			    (u_longlong_t)zio->io_offset,
899 			    zio, blkbuf ? blkbuf : "", zio->io_error);
900 		}
901 	}
902 	zio_clear_transform_stack(zio);
903 
904 	if (zio->io_done)
905 		zio->io_done(zio);
906 
907 	ASSERT(zio->io_delegate_list == NULL);
908 	ASSERT(zio->io_delegate_next == NULL);
909 
910 	if (pio != NULL) {
911 		zio_t *next, *prev;
912 
913 		mutex_enter(&pio->io_lock);
914 		next = zio->io_sibling_next;
915 		prev = zio->io_sibling_prev;
916 		if (next != NULL)
917 			next->io_sibling_prev = prev;
918 		if (prev != NULL)
919 			prev->io_sibling_next = next;
920 		if (pio->io_child == zio)
921 			pio->io_child = next;
922 		mutex_exit(&pio->io_lock);
923 
924 		zio_notify_parent(zio, ZIO_STAGE_WAIT_CHILDREN_DONE,
925 		    &pio->io_children_notdone);
926 	}
927 
928 	/*
929 	 * Note: this I/O is now done, and will shortly be
930 	 * kmem_free()'d, so there is no need to clear this (or any
931 	 * other) flag.
932 	 */
933 	if (zio->io_flags & ZIO_FLAG_CONFIG_GRABBED)
934 		spa_config_exit(spa, zio);
935 
936 	if (zio->io_waiter != NULL) {
937 		mutex_enter(&zio->io_lock);
938 		ASSERT(zio->io_stage == ZIO_STAGE_DONE);
939 		zio->io_stalled = zio->io_stage;
940 		cv_broadcast(&zio->io_cv);
941 		mutex_exit(&zio->io_lock);
942 	} else {
943 		kmem_free(zio, sizeof (zio_t));
944 	}
945 }
946 
947 /*
948  * ==========================================================================
949  * Compression support
950  * ==========================================================================
951  */
952 static void
953 zio_write_compress(zio_t *zio)
954 {
955 	int compress = zio->io_compress;
956 	blkptr_t *bp = zio->io_bp;
957 	void *cbuf;
958 	uint64_t lsize = zio->io_size;
959 	uint64_t csize = lsize;
960 	uint64_t cbufsize = 0;
961 	int pass;
962 
963 	if (bp->blk_birth == zio->io_txg) {
964 		/*
965 		 * We're rewriting an existing block, which means we're
966 		 * working on behalf of spa_sync().  For spa_sync() to
967 		 * converge, it must eventually be the case that we don't
968 		 * have to allocate new blocks.  But compression changes
969 		 * the blocksize, which forces a reallocate, and makes
970 		 * convergence take longer.  Therefore, after the first
971 		 * few passes, stop compressing to ensure convergence.
972 		 */
973 		pass = spa_sync_pass(zio->io_spa);
974 		if (pass > zio_sync_pass.zp_dontcompress)
975 			compress = ZIO_COMPRESS_OFF;
976 	} else {
977 		ASSERT(BP_IS_HOLE(bp));
978 		pass = 1;
979 	}
980 
981 	if (compress != ZIO_COMPRESS_OFF)
982 		if (!zio_compress_data(compress, zio->io_data, zio->io_size,
983 		    &cbuf, &csize, &cbufsize))
984 			compress = ZIO_COMPRESS_OFF;
985 
986 	if (compress != ZIO_COMPRESS_OFF && csize != 0)
987 		zio_push_transform(zio, cbuf, csize, cbufsize);
988 
989 	/*
990 	 * The final pass of spa_sync() must be all rewrites, but the first
991 	 * few passes offer a trade-off: allocating blocks defers convergence,
992 	 * but newly allocated blocks are sequential, so they can be written
993 	 * to disk faster.  Therefore, we allow the first few passes of
994 	 * spa_sync() to reallocate new blocks, but force rewrites after that.
995 	 * There should only be a handful of blocks after pass 1 in any case.
996 	 */
997 	if (bp->blk_birth == zio->io_txg && BP_GET_PSIZE(bp) == csize &&
998 	    pass > zio_sync_pass.zp_rewrite) {
999 		ASSERT(csize != 0);
1000 		BP_SET_LSIZE(bp, lsize);
1001 		BP_SET_COMPRESS(bp, compress);
1002 		zio->io_pipeline = ZIO_REWRITE_PIPELINE;
1003 	} else {
1004 		if (bp->blk_birth == zio->io_txg) {
1005 			ASSERT3U(BP_GET_LSIZE(bp), ==, lsize);
1006 			bzero(bp, sizeof (blkptr_t));
1007 		}
1008 		if (csize == 0) {
1009 			BP_ZERO(bp);
1010 			zio->io_pipeline = ZIO_WAIT_FOR_CHILDREN_PIPELINE;
1011 		} else {
1012 			ASSERT3U(BP_GET_NDVAS(bp), ==, 0);
1013 			BP_SET_LSIZE(bp, lsize);
1014 			BP_SET_PSIZE(bp, csize);
1015 			BP_SET_COMPRESS(bp, compress);
1016 			zio->io_pipeline = ZIO_WRITE_ALLOCATE_PIPELINE;
1017 		}
1018 	}
1019 
1020 	zio_next_stage(zio);
1021 }
1022 
1023 static void
1024 zio_read_decompress(zio_t *zio)
1025 {
1026 	blkptr_t *bp = zio->io_bp;
1027 	void *data;
1028 	uint64_t size;
1029 	uint64_t bufsize;
1030 	int compress = BP_GET_COMPRESS(bp);
1031 
1032 	ASSERT(compress != ZIO_COMPRESS_OFF);
1033 
1034 	zio_pop_transform(zio, &data, &size, &bufsize);
1035 
1036 	if (zio_decompress_data(compress, data, size,
1037 	    zio->io_data, zio->io_size))
1038 		zio->io_error = EIO;
1039 
1040 	zio_buf_free(data, bufsize);
1041 
1042 	zio_next_stage(zio);
1043 }
1044 
1045 /*
1046  * ==========================================================================
1047  * Gang block support
1048  * ==========================================================================
1049  */
1050 static void
1051 zio_gang_pipeline(zio_t *zio)
1052 {
1053 	/*
1054 	 * By default, the pipeline assumes that we're dealing with a gang
1055 	 * block.  If we're not, strip out any gang-specific stages.
1056 	 */
1057 	if (!BP_IS_GANG(zio->io_bp))
1058 		zio->io_pipeline &= ~ZIO_GANG_STAGES;
1059 
1060 	zio_next_stage(zio);
1061 }
1062 
1063 static void
1064 zio_gang_byteswap(zio_t *zio)
1065 {
1066 	ASSERT(zio->io_size == SPA_GANGBLOCKSIZE);
1067 
1068 	if (BP_SHOULD_BYTESWAP(zio->io_bp))
1069 		byteswap_uint64_array(zio->io_data, zio->io_size);
1070 }
1071 
1072 static void
1073 zio_get_gang_header(zio_t *zio)
1074 {
1075 	blkptr_t *bp = zio->io_bp;
1076 	uint64_t gsize = SPA_GANGBLOCKSIZE;
1077 	void *gbuf = zio_buf_alloc(gsize);
1078 
1079 	ASSERT(BP_IS_GANG(bp));
1080 
1081 	zio_push_transform(zio, gbuf, gsize, gsize);
1082 
1083 	zio_nowait(zio_create(zio, zio->io_spa, bp->blk_birth, bp, gbuf, gsize,
1084 	    NULL, NULL, ZIO_TYPE_READ, zio->io_priority,
1085 	    zio->io_flags & ZIO_FLAG_GANG_INHERIT,
1086 	    ZIO_STAGE_OPEN, ZIO_READ_PIPELINE));
1087 
1088 	zio_wait_children_done(zio);
1089 }
1090 
1091 static void
1092 zio_read_gang_members(zio_t *zio)
1093 {
1094 	zio_gbh_phys_t *gbh;
1095 	uint64_t gsize, gbufsize, loff, lsize;
1096 	int i;
1097 
1098 	ASSERT(BP_IS_GANG(zio->io_bp));
1099 
1100 	zio_gang_byteswap(zio);
1101 	zio_pop_transform(zio, (void **)&gbh, &gsize, &gbufsize);
1102 
1103 	for (loff = 0, i = 0; loff != zio->io_size; loff += lsize, i++) {
1104 		blkptr_t *gbp = &gbh->zg_blkptr[i];
1105 		lsize = BP_GET_PSIZE(gbp);
1106 
1107 		ASSERT(BP_GET_COMPRESS(gbp) == ZIO_COMPRESS_OFF);
1108 		ASSERT3U(lsize, ==, BP_GET_LSIZE(gbp));
1109 		ASSERT3U(loff + lsize, <=, zio->io_size);
1110 		ASSERT(i < SPA_GBH_NBLKPTRS);
1111 		ASSERT(!BP_IS_HOLE(gbp));
1112 
1113 		zio_nowait(zio_read(zio, zio->io_spa, gbp,
1114 		    (char *)zio->io_data + loff, lsize, NULL, NULL,
1115 		    zio->io_priority, zio->io_flags & ZIO_FLAG_GANG_INHERIT,
1116 		    &zio->io_bookmark));
1117 	}
1118 
1119 	zio_buf_free(gbh, gbufsize);
1120 	zio_wait_children_done(zio);
1121 }
1122 
1123 static void
1124 zio_rewrite_gang_members(zio_t *zio)
1125 {
1126 	zio_gbh_phys_t *gbh;
1127 	uint64_t gsize, gbufsize, loff, lsize;
1128 	int i;
1129 
1130 	ASSERT(BP_IS_GANG(zio->io_bp));
1131 	ASSERT3U(zio->io_size, ==, SPA_GANGBLOCKSIZE);
1132 
1133 	zio_gang_byteswap(zio);
1134 	zio_pop_transform(zio, (void **)&gbh, &gsize, &gbufsize);
1135 
1136 	ASSERT(gsize == gbufsize);
1137 
1138 	for (loff = 0, i = 0; loff != zio->io_size; loff += lsize, i++) {
1139 		blkptr_t *gbp = &gbh->zg_blkptr[i];
1140 		lsize = BP_GET_PSIZE(gbp);
1141 
1142 		ASSERT(BP_GET_COMPRESS(gbp) == ZIO_COMPRESS_OFF);
1143 		ASSERT3U(lsize, ==, BP_GET_LSIZE(gbp));
1144 		ASSERT3U(loff + lsize, <=, zio->io_size);
1145 		ASSERT(i < SPA_GBH_NBLKPTRS);
1146 		ASSERT(!BP_IS_HOLE(gbp));
1147 
1148 		zio_nowait(zio_rewrite(zio, zio->io_spa, zio->io_checksum,
1149 		    zio->io_txg, gbp, (char *)zio->io_data + loff, lsize,
1150 		    NULL, NULL, zio->io_priority, zio->io_flags,
1151 		    &zio->io_bookmark));
1152 	}
1153 
1154 	zio_push_transform(zio, gbh, gsize, gbufsize);
1155 	zio_wait_children_ready(zio);
1156 }
1157 
1158 static void
1159 zio_free_gang_members(zio_t *zio)
1160 {
1161 	zio_gbh_phys_t *gbh;
1162 	uint64_t gsize, gbufsize;
1163 	int i;
1164 
1165 	ASSERT(BP_IS_GANG(zio->io_bp));
1166 
1167 	zio_gang_byteswap(zio);
1168 	zio_pop_transform(zio, (void **)&gbh, &gsize, &gbufsize);
1169 
1170 	for (i = 0; i < SPA_GBH_NBLKPTRS; i++) {
1171 		blkptr_t *gbp = &gbh->zg_blkptr[i];
1172 
1173 		if (BP_IS_HOLE(gbp))
1174 			continue;
1175 		zio_nowait(zio_free(zio, zio->io_spa, zio->io_txg,
1176 		    gbp, NULL, NULL));
1177 	}
1178 
1179 	zio_buf_free(gbh, gbufsize);
1180 	zio_next_stage(zio);
1181 }
1182 
1183 static void
1184 zio_claim_gang_members(zio_t *zio)
1185 {
1186 	zio_gbh_phys_t *gbh;
1187 	uint64_t gsize, gbufsize;
1188 	int i;
1189 
1190 	ASSERT(BP_IS_GANG(zio->io_bp));
1191 
1192 	zio_gang_byteswap(zio);
1193 	zio_pop_transform(zio, (void **)&gbh, &gsize, &gbufsize);
1194 
1195 	for (i = 0; i < SPA_GBH_NBLKPTRS; i++) {
1196 		blkptr_t *gbp = &gbh->zg_blkptr[i];
1197 		if (BP_IS_HOLE(gbp))
1198 			continue;
1199 		zio_nowait(zio_claim(zio, zio->io_spa, zio->io_txg,
1200 		    gbp, NULL, NULL));
1201 	}
1202 
1203 	zio_buf_free(gbh, gbufsize);
1204 	zio_next_stage(zio);
1205 }
1206 
1207 static void
1208 zio_write_allocate_gang_member_done(zio_t *zio)
1209 {
1210 	zio_t *pio = zio->io_parent;
1211 	dva_t *cdva = zio->io_bp->blk_dva;
1212 	dva_t *pdva = pio->io_bp->blk_dva;
1213 	uint64_t asize;
1214 	int d;
1215 
1216 	ASSERT3U(pio->io_ndvas, ==, zio->io_ndvas);
1217 	ASSERT3U(BP_GET_NDVAS(zio->io_bp), <=, BP_GET_NDVAS(pio->io_bp));
1218 	ASSERT3U(zio->io_ndvas, <=, BP_GET_NDVAS(zio->io_bp));
1219 	ASSERT3U(pio->io_ndvas, <=, BP_GET_NDVAS(pio->io_bp));
1220 
1221 	mutex_enter(&pio->io_lock);
1222 	for (d = 0; d < BP_GET_NDVAS(pio->io_bp); d++) {
1223 		ASSERT(DVA_GET_GANG(&pdva[d]));
1224 		asize = DVA_GET_ASIZE(&pdva[d]);
1225 		asize += DVA_GET_ASIZE(&cdva[d]);
1226 		DVA_SET_ASIZE(&pdva[d], asize);
1227 	}
1228 	mutex_exit(&pio->io_lock);
1229 }
1230 
1231 static void
1232 zio_write_allocate_gang_members(zio_t *zio)
1233 {
1234 	blkptr_t *bp = zio->io_bp;
1235 	dva_t *dva = bp->blk_dva;
1236 	spa_t *spa = zio->io_spa;
1237 	zio_gbh_phys_t *gbh;
1238 	uint64_t txg = zio->io_txg;
1239 	uint64_t resid = zio->io_size;
1240 	uint64_t maxalloc = P2ROUNDUP(zio->io_size >> 1, SPA_MINBLOCKSIZE);
1241 	uint64_t gsize, loff, lsize;
1242 	uint32_t gbps_left;
1243 	int ndvas = zio->io_ndvas;
1244 	int gbh_ndvas = MIN(ndvas + 1, spa_max_replication(spa));
1245 	int error;
1246 	int i, d;
1247 
1248 	gsize = SPA_GANGBLOCKSIZE;
1249 	gbps_left = SPA_GBH_NBLKPTRS;
1250 
1251 	error = metaslab_alloc(spa, gsize, bp, gbh_ndvas, txg, NULL, B_FALSE);
1252 	if (error == ENOSPC)
1253 		panic("can't allocate gang block header");
1254 	ASSERT(error == 0);
1255 
1256 	for (d = 0; d < gbh_ndvas; d++)
1257 		DVA_SET_GANG(&dva[d], 1);
1258 
1259 	bp->blk_birth = txg;
1260 
1261 	gbh = zio_buf_alloc(gsize);
1262 	bzero(gbh, gsize);
1263 
1264 	/* We need to test multi-level gang blocks */
1265 	if (maxalloc >= zio_gang_bang && (lbolt & 0x1) == 0)
1266 		maxalloc = MAX(maxalloc >> 2, SPA_MINBLOCKSIZE);
1267 
1268 	for (loff = 0, i = 0; loff != zio->io_size;
1269 	    loff += lsize, resid -= lsize, gbps_left--, i++) {
1270 		blkptr_t *gbp = &gbh->zg_blkptr[i];
1271 		dva = gbp->blk_dva;
1272 
1273 		ASSERT(gbps_left != 0);
1274 		maxalloc = MIN(maxalloc, resid);
1275 
1276 		while (resid <= maxalloc * gbps_left) {
1277 			error = metaslab_alloc(spa, maxalloc, gbp, ndvas,
1278 			    txg, bp, B_FALSE);
1279 			if (error == 0)
1280 				break;
1281 			ASSERT3U(error, ==, ENOSPC);
1282 			if (maxalloc == SPA_MINBLOCKSIZE)
1283 				panic("really out of space");
1284 			maxalloc = P2ROUNDUP(maxalloc >> 1, SPA_MINBLOCKSIZE);
1285 		}
1286 
1287 		if (resid <= maxalloc * gbps_left) {
1288 			lsize = maxalloc;
1289 			BP_SET_LSIZE(gbp, lsize);
1290 			BP_SET_PSIZE(gbp, lsize);
1291 			BP_SET_COMPRESS(gbp, ZIO_COMPRESS_OFF);
1292 			gbp->blk_birth = txg;
1293 			zio_nowait(zio_rewrite(zio, spa,
1294 			    zio->io_checksum, txg, gbp,
1295 			    (char *)zio->io_data + loff, lsize,
1296 			    zio_write_allocate_gang_member_done, NULL,
1297 			    zio->io_priority, zio->io_flags,
1298 			    &zio->io_bookmark));
1299 		} else {
1300 			lsize = P2ROUNDUP(resid / gbps_left, SPA_MINBLOCKSIZE);
1301 			ASSERT(lsize != SPA_MINBLOCKSIZE);
1302 			zio_nowait(zio_write_allocate(zio, spa,
1303 			    zio->io_checksum, txg, gbp,
1304 			    (char *)zio->io_data + loff, lsize,
1305 			    zio_write_allocate_gang_member_done, NULL,
1306 			    zio->io_priority, zio->io_flags));
1307 		}
1308 	}
1309 
1310 	ASSERT(resid == 0 && loff == zio->io_size);
1311 
1312 	zio->io_pipeline |= 1U << ZIO_STAGE_GANG_CHECKSUM_GENERATE;
1313 
1314 	zio_push_transform(zio, gbh, gsize, gsize);
1315 	/*
1316 	 * As much as we'd like this to be zio_wait_children_ready(),
1317 	 * updating our ASIZE doesn't happen until the io_done callback,
1318 	 * so we have to wait for that to finish in order for our BP
1319 	 * to be stable.
1320 	 */
1321 	zio_wait_children_done(zio);
1322 }
1323 
1324 /*
1325  * ==========================================================================
1326  * Allocate and free blocks
1327  * ==========================================================================
1328  */
1329 static void
1330 zio_dva_allocate(zio_t *zio)
1331 {
1332 	blkptr_t *bp = zio->io_bp;
1333 	int error;
1334 
1335 	ASSERT(BP_IS_HOLE(bp));
1336 	ASSERT3U(BP_GET_NDVAS(bp), ==, 0);
1337 	ASSERT3U(zio->io_ndvas, >, 0);
1338 	ASSERT3U(zio->io_ndvas, <=, spa_max_replication(zio->io_spa));
1339 
1340 	/* For testing, make some blocks above a certain size be gang blocks */
1341 	if (zio->io_size >= zio_gang_bang && (lbolt & 0x3) == 0) {
1342 		zio_write_allocate_gang_members(zio);
1343 		return;
1344 	}
1345 
1346 	ASSERT3U(zio->io_size, ==, BP_GET_PSIZE(bp));
1347 
1348 	error = metaslab_alloc(zio->io_spa, zio->io_size, bp, zio->io_ndvas,
1349 	    zio->io_txg, NULL, B_FALSE);
1350 
1351 	if (error == 0) {
1352 		bp->blk_birth = zio->io_txg;
1353 	} else if (error == ENOSPC) {
1354 		if (zio->io_size == SPA_MINBLOCKSIZE)
1355 			panic("really, truly out of space");
1356 		zio_write_allocate_gang_members(zio);
1357 		return;
1358 	} else {
1359 		zio->io_error = error;
1360 	}
1361 	zio_next_stage(zio);
1362 }
1363 
1364 static void
1365 zio_dva_free(zio_t *zio)
1366 {
1367 	blkptr_t *bp = zio->io_bp;
1368 
1369 	metaslab_free(zio->io_spa, bp, zio->io_txg, B_FALSE);
1370 
1371 	BP_ZERO(bp);
1372 
1373 	zio_next_stage(zio);
1374 }
1375 
1376 static void
1377 zio_dva_claim(zio_t *zio)
1378 {
1379 	zio->io_error = metaslab_claim(zio->io_spa, zio->io_bp, zio->io_txg);
1380 
1381 	zio_next_stage(zio);
1382 }
1383 
1384 /*
1385  * ==========================================================================
1386  * Read and write to physical devices
1387  * ==========================================================================
1388  */
1389 
1390 static void
1391 zio_vdev_io_start(zio_t *zio)
1392 {
1393 	vdev_t *vd = zio->io_vd;
1394 	vdev_t *tvd = vd ? vd->vdev_top : NULL;
1395 	blkptr_t *bp = zio->io_bp;
1396 	uint64_t align;
1397 
1398 	if (vd == NULL) {
1399 		/* The mirror_ops handle multiple DVAs in a single BP */
1400 		vdev_mirror_ops.vdev_op_io_start(zio);
1401 		return;
1402 	}
1403 
1404 	align = 1ULL << tvd->vdev_ashift;
1405 
1406 	if (zio->io_retries == 0 && vd == tvd)
1407 		zio->io_flags |= ZIO_FLAG_FAILFAST;
1408 
1409 	if (!(zio->io_flags & ZIO_FLAG_PHYSICAL) &&
1410 	    vd->vdev_children == 0) {
1411 		zio->io_flags |= ZIO_FLAG_PHYSICAL;
1412 		zio->io_offset += VDEV_LABEL_START_SIZE;
1413 	}
1414 
1415 	if (P2PHASE(zio->io_size, align) != 0) {
1416 		uint64_t asize = P2ROUNDUP(zio->io_size, align);
1417 		char *abuf = zio_buf_alloc(asize);
1418 		ASSERT(vd == tvd);
1419 		if (zio->io_type == ZIO_TYPE_WRITE) {
1420 			bcopy(zio->io_data, abuf, zio->io_size);
1421 			bzero(abuf + zio->io_size, asize - zio->io_size);
1422 		}
1423 		zio_push_transform(zio, abuf, asize, asize);
1424 		ASSERT(!(zio->io_flags & ZIO_FLAG_SUBBLOCK));
1425 		zio->io_flags |= ZIO_FLAG_SUBBLOCK;
1426 	}
1427 
1428 	ASSERT(P2PHASE(zio->io_offset, align) == 0);
1429 	ASSERT(P2PHASE(zio->io_size, align) == 0);
1430 	ASSERT(bp == NULL ||
1431 	    P2ROUNDUP(ZIO_GET_IOSIZE(zio), align) == zio->io_size);
1432 	ASSERT(zio->io_type != ZIO_TYPE_WRITE || (spa_mode & FWRITE));
1433 
1434 	vdev_io_start(zio);
1435 
1436 	/* zio_next_stage_async() gets called from io completion interrupt */
1437 }
1438 
1439 static void
1440 zio_vdev_io_done(zio_t *zio)
1441 {
1442 	if (zio->io_vd == NULL)
1443 		/* The mirror_ops handle multiple DVAs in a single BP */
1444 		vdev_mirror_ops.vdev_op_io_done(zio);
1445 	else
1446 		vdev_io_done(zio);
1447 }
1448 
1449 /* XXPOLICY */
1450 boolean_t
1451 zio_should_retry(zio_t *zio)
1452 {
1453 	vdev_t *vd = zio->io_vd;
1454 
1455 	if (zio->io_error == 0)
1456 		return (B_FALSE);
1457 	if (zio->io_delegate_list != NULL)
1458 		return (B_FALSE);
1459 	if (vd && vd != vd->vdev_top)
1460 		return (B_FALSE);
1461 	if (zio->io_flags & ZIO_FLAG_DONT_RETRY)
1462 		return (B_FALSE);
1463 	if (zio->io_retries > 0)
1464 		return (B_FALSE);
1465 
1466 	return (B_TRUE);
1467 }
1468 
1469 static void
1470 zio_vdev_io_assess(zio_t *zio)
1471 {
1472 	vdev_t *vd = zio->io_vd;
1473 	vdev_t *tvd = vd ? vd->vdev_top : NULL;
1474 
1475 	ASSERT(zio->io_vsd == NULL);
1476 
1477 	if (zio->io_flags & ZIO_FLAG_SUBBLOCK) {
1478 		void *abuf;
1479 		uint64_t asize;
1480 		ASSERT(vd == tvd);
1481 		zio_pop_transform(zio, &abuf, &asize, &asize);
1482 		if (zio->io_type == ZIO_TYPE_READ)
1483 			bcopy(abuf, zio->io_data, zio->io_size);
1484 		zio_buf_free(abuf, asize);
1485 		zio->io_flags &= ~ZIO_FLAG_SUBBLOCK;
1486 	}
1487 
1488 	if (zio_injection_enabled && !zio->io_error)
1489 		zio->io_error = zio_handle_fault_injection(zio, EIO);
1490 
1491 	/*
1492 	 * If the I/O failed, determine whether we should attempt to retry it.
1493 	 */
1494 	/* XXPOLICY */
1495 	if (zio_should_retry(zio)) {
1496 		ASSERT(tvd == vd);
1497 
1498 		zio->io_retries++;
1499 		zio->io_error = 0;
1500 		zio->io_flags &= ZIO_FLAG_VDEV_INHERIT |
1501 		    ZIO_FLAG_CONFIG_GRABBED;
1502 		/* XXPOLICY */
1503 		zio->io_flags &= ~ZIO_FLAG_FAILFAST;
1504 		zio->io_flags |= ZIO_FLAG_DONT_CACHE;
1505 		zio->io_stage = ZIO_STAGE_VDEV_IO_START - 1;
1506 
1507 		dprintf("retry #%d for %s to %s offset %llx\n",
1508 		    zio->io_retries, zio_type_name[zio->io_type],
1509 		    vdev_description(vd), zio->io_offset);
1510 
1511 		zio_next_stage_async(zio);
1512 		return;
1513 	}
1514 
1515 	if (zio->io_error != 0 && zio->io_error != ECKSUM &&
1516 	    !(zio->io_flags & ZIO_FLAG_SPECULATIVE) && vd) {
1517 		/*
1518 		 * Poor man's hotplug support.  Even if we're done retrying this
1519 		 * I/O, try to reopen the vdev to see if it's still attached.
1520 		 * To avoid excessive thrashing, we only try it once a minute.
1521 		 * This also has the effect of detecting when missing devices
1522 		 * have come back, by polling the device once a minute.
1523 		 *
1524 		 * We need to do this asynchronously because we can't grab
1525 		 * all the necessary locks way down here.
1526 		 */
1527 		if (gethrtime() - vd->vdev_last_try > 60ULL * NANOSEC) {
1528 			vd->vdev_last_try = gethrtime();
1529 			tvd->vdev_reopen_wanted = 1;
1530 			spa_async_request(vd->vdev_spa, SPA_ASYNC_REOPEN);
1531 		}
1532 	}
1533 
1534 	zio_next_stage(zio);
1535 }
1536 
1537 void
1538 zio_vdev_io_reissue(zio_t *zio)
1539 {
1540 	ASSERT(zio->io_stage == ZIO_STAGE_VDEV_IO_START);
1541 	ASSERT(zio->io_error == 0);
1542 
1543 	zio->io_stage--;
1544 }
1545 
1546 void
1547 zio_vdev_io_redone(zio_t *zio)
1548 {
1549 	ASSERT(zio->io_stage == ZIO_STAGE_VDEV_IO_DONE);
1550 
1551 	zio->io_stage--;
1552 }
1553 
1554 void
1555 zio_vdev_io_bypass(zio_t *zio)
1556 {
1557 	ASSERT(zio->io_stage == ZIO_STAGE_VDEV_IO_START);
1558 	ASSERT(zio->io_error == 0);
1559 
1560 	zio->io_flags |= ZIO_FLAG_IO_BYPASS;
1561 	zio->io_stage = ZIO_STAGE_VDEV_IO_ASSESS - 1;
1562 }
1563 
1564 /*
1565  * ==========================================================================
1566  * Generate and verify checksums
1567  * ==========================================================================
1568  */
1569 static void
1570 zio_checksum_generate(zio_t *zio)
1571 {
1572 	int checksum = zio->io_checksum;
1573 	blkptr_t *bp = zio->io_bp;
1574 
1575 	ASSERT3U(zio->io_size, ==, BP_GET_PSIZE(bp));
1576 
1577 	BP_SET_CHECKSUM(bp, checksum);
1578 	BP_SET_BYTEORDER(bp, ZFS_HOST_BYTEORDER);
1579 
1580 	zio_checksum(checksum, &bp->blk_cksum, zio->io_data, zio->io_size);
1581 
1582 	zio_next_stage(zio);
1583 }
1584 
1585 static void
1586 zio_gang_checksum_generate(zio_t *zio)
1587 {
1588 	zio_cksum_t zc;
1589 	zio_gbh_phys_t *gbh = zio->io_data;
1590 
1591 	ASSERT(BP_IS_GANG(zio->io_bp));
1592 	ASSERT3U(zio->io_size, ==, SPA_GANGBLOCKSIZE);
1593 
1594 	zio_set_gang_verifier(zio, &gbh->zg_tail.zbt_cksum);
1595 
1596 	zio_checksum(ZIO_CHECKSUM_GANG_HEADER, &zc, zio->io_data, zio->io_size);
1597 
1598 	zio_next_stage(zio);
1599 }
1600 
1601 static void
1602 zio_checksum_verify(zio_t *zio)
1603 {
1604 	if (zio->io_bp != NULL) {
1605 		zio->io_error = zio_checksum_error(zio);
1606 		if (zio->io_error && !(zio->io_flags & ZIO_FLAG_SPECULATIVE))
1607 			zfs_ereport_post(FM_EREPORT_ZFS_CHECKSUM,
1608 			    zio->io_spa, zio->io_vd, zio, 0, 0);
1609 	}
1610 
1611 	zio_next_stage(zio);
1612 }
1613 
1614 /*
1615  * Called by RAID-Z to ensure we don't compute the checksum twice.
1616  */
1617 void
1618 zio_checksum_verified(zio_t *zio)
1619 {
1620 	zio->io_pipeline &= ~(1U << ZIO_STAGE_CHECKSUM_VERIFY);
1621 }
1622 
1623 /*
1624  * Set the external verifier for a gang block based on stuff in the bp
1625  */
1626 void
1627 zio_set_gang_verifier(zio_t *zio, zio_cksum_t *zcp)
1628 {
1629 	blkptr_t *bp = zio->io_bp;
1630 
1631 	zcp->zc_word[0] = DVA_GET_VDEV(BP_IDENTITY(bp));
1632 	zcp->zc_word[1] = DVA_GET_OFFSET(BP_IDENTITY(bp));
1633 	zcp->zc_word[2] = bp->blk_birth;
1634 	zcp->zc_word[3] = 0;
1635 }
1636 
1637 /*
1638  * ==========================================================================
1639  * Define the pipeline
1640  * ==========================================================================
1641  */
1642 typedef void zio_pipe_stage_t(zio_t *zio);
1643 
1644 static void
1645 zio_badop(zio_t *zio)
1646 {
1647 	panic("Invalid I/O pipeline stage %u for zio %p", zio->io_stage, zio);
1648 }
1649 
1650 zio_pipe_stage_t *zio_pipeline[ZIO_STAGE_DONE + 2] = {
1651 	zio_badop,
1652 	zio_wait_children_ready,
1653 	zio_write_compress,
1654 	zio_checksum_generate,
1655 	zio_gang_pipeline,
1656 	zio_get_gang_header,
1657 	zio_rewrite_gang_members,
1658 	zio_free_gang_members,
1659 	zio_claim_gang_members,
1660 	zio_dva_allocate,
1661 	zio_dva_free,
1662 	zio_dva_claim,
1663 	zio_gang_checksum_generate,
1664 	zio_ready,
1665 	zio_vdev_io_start,
1666 	zio_vdev_io_done,
1667 	zio_vdev_io_assess,
1668 	zio_wait_children_done,
1669 	zio_checksum_verify,
1670 	zio_read_gang_members,
1671 	zio_read_decompress,
1672 	zio_done,
1673 	zio_badop
1674 };
1675 
1676 /*
1677  * Move an I/O to the next stage of the pipeline and execute that stage.
1678  * There's no locking on io_stage because there's no legitimate way for
1679  * multiple threads to be attempting to process the same I/O.
1680  */
1681 void
1682 zio_next_stage(zio_t *zio)
1683 {
1684 	uint32_t pipeline = zio->io_pipeline;
1685 
1686 	ASSERT(!MUTEX_HELD(&zio->io_lock));
1687 
1688 	if (zio->io_error) {
1689 		dprintf("zio %p vdev %s offset %llx stage %d error %d\n",
1690 		    zio, vdev_description(zio->io_vd),
1691 		    zio->io_offset, zio->io_stage, zio->io_error);
1692 		if (((1U << zio->io_stage) & ZIO_VDEV_IO_PIPELINE) == 0)
1693 			pipeline &= ZIO_ERROR_PIPELINE_MASK;
1694 	}
1695 
1696 	while (((1U << ++zio->io_stage) & pipeline) == 0)
1697 		continue;
1698 
1699 	ASSERT(zio->io_stage <= ZIO_STAGE_DONE);
1700 	ASSERT(zio->io_stalled == 0);
1701 
1702 	zio_pipeline[zio->io_stage](zio);
1703 }
1704 
1705 void
1706 zio_next_stage_async(zio_t *zio)
1707 {
1708 	taskq_t *tq;
1709 	uint32_t pipeline = zio->io_pipeline;
1710 
1711 	ASSERT(!MUTEX_HELD(&zio->io_lock));
1712 
1713 	if (zio->io_error) {
1714 		dprintf("zio %p vdev %s offset %llx stage %d error %d\n",
1715 		    zio, vdev_description(zio->io_vd),
1716 		    zio->io_offset, zio->io_stage, zio->io_error);
1717 		if (((1U << zio->io_stage) & ZIO_VDEV_IO_PIPELINE) == 0)
1718 			pipeline &= ZIO_ERROR_PIPELINE_MASK;
1719 	}
1720 
1721 	while (((1U << ++zio->io_stage) & pipeline) == 0)
1722 		continue;
1723 
1724 	ASSERT(zio->io_stage <= ZIO_STAGE_DONE);
1725 	ASSERT(zio->io_stalled == 0);
1726 
1727 	/*
1728 	 * For performance, we'll probably want two sets of task queues:
1729 	 * per-CPU issue taskqs and per-CPU completion taskqs.  The per-CPU
1730 	 * part is for read performance: since we have to make a pass over
1731 	 * the data to checksum it anyway, we want to do this on the same CPU
1732 	 * that issued the read, because (assuming CPU scheduling affinity)
1733 	 * that thread is probably still there.  Getting this optimization
1734 	 * right avoids performance-hostile cache-to-cache transfers.
1735 	 *
1736 	 * Note that having two sets of task queues is also necessary for
1737 	 * correctness: if all of the issue threads get bogged down waiting
1738 	 * for dependent reads (e.g. metaslab freelist) to complete, then
1739 	 * there won't be any threads available to service I/O completion
1740 	 * interrupts.
1741 	 */
1742 	if ((1U << zio->io_stage) & zio->io_async_stages) {
1743 		if (zio->io_stage < ZIO_STAGE_VDEV_IO_DONE)
1744 			tq = zio->io_spa->spa_zio_issue_taskq[zio->io_type];
1745 		else
1746 			tq = zio->io_spa->spa_zio_intr_taskq[zio->io_type];
1747 		(void) taskq_dispatch(tq,
1748 		    (task_func_t *)zio_pipeline[zio->io_stage], zio, TQ_SLEEP);
1749 	} else {
1750 		zio_pipeline[zio->io_stage](zio);
1751 	}
1752 }
1753 
1754 /*
1755  * Try to allocate an intent log block.  Return 0 on success, errno on failure.
1756  */
1757 int
1758 zio_alloc_blk(spa_t *spa, uint64_t size, blkptr_t *new_bp, blkptr_t *old_bp,
1759     uint64_t txg)
1760 {
1761 	int error;
1762 
1763 	spa_config_enter(spa, RW_READER, FTAG);
1764 
1765 	/*
1766 	 * We were passed the previous log blocks dva_t in bp->blk_dva[0].
1767 	 */
1768 	error = metaslab_alloc(spa, size, new_bp, 1, txg, old_bp, B_TRUE);
1769 
1770 	if (error == 0) {
1771 		BP_SET_LSIZE(new_bp, size);
1772 		BP_SET_PSIZE(new_bp, size);
1773 		BP_SET_COMPRESS(new_bp, ZIO_COMPRESS_OFF);
1774 		BP_SET_CHECKSUM(new_bp, ZIO_CHECKSUM_ZILOG);
1775 		BP_SET_TYPE(new_bp, DMU_OT_INTENT_LOG);
1776 		BP_SET_LEVEL(new_bp, 0);
1777 		BP_SET_BYTEORDER(new_bp, ZFS_HOST_BYTEORDER);
1778 		new_bp->blk_birth = txg;
1779 	}
1780 
1781 	spa_config_exit(spa, FTAG);
1782 
1783 	return (error);
1784 }
1785 
1786 /*
1787  * Free an intent log block.  We know it can't be a gang block, so there's
1788  * nothing to do except metaslab_free() it.
1789  */
1790 void
1791 zio_free_blk(spa_t *spa, blkptr_t *bp, uint64_t txg)
1792 {
1793 	ASSERT(!BP_IS_GANG(bp));
1794 
1795 	spa_config_enter(spa, RW_READER, FTAG);
1796 
1797 	metaslab_free(spa, bp, txg, B_FALSE);
1798 
1799 	spa_config_exit(spa, FTAG);
1800 }
1801