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