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 2013 Nexenta Systems, Inc. All rights reserved.
24 * Copyright (c) 2011, 2017 by Delphix. All rights reserved.
25 * Copyright (c) 2014 Integros [integros.com]
26 */
27
28 /* Portions Copyright 2010 Robert Milkowski */
29
30 #include <sys/zfs_context.h>
31 #include <sys/spa.h>
32 #include <sys/dmu.h>
33 #include <sys/zap.h>
34 #include <sys/arc.h>
35 #include <sys/stat.h>
36 #include <sys/resource.h>
37 #include <sys/zil.h>
38 #include <sys/zil_impl.h>
39 #include <sys/dsl_dataset.h>
40 #include <sys/vdev_impl.h>
41 #include <sys/dmu_tx.h>
42 #include <sys/dsl_pool.h>
43 #include <sys/abd.h>
44
45 /*
46 * The ZFS Intent Log (ZIL) saves "transaction records" (itxs) of system
47 * calls that change the file system. Each itx has enough information to
48 * be able to replay them after a system crash, power loss, or
49 * equivalent failure mode. These are stored in memory until either:
50 *
51 * 1. they are committed to the pool by the DMU transaction group
52 * (txg), at which point they can be discarded; or
53 * 2. they are committed to the on-disk ZIL for the dataset being
54 * modified (e.g. due to an fsync, O_DSYNC, or other synchronous
55 * requirement).
56 *
57 * In the event of a crash or power loss, the itxs contained by each
58 * dataset's on-disk ZIL will be replayed when that dataset is first
59 * instantianted (e.g. if the dataset is a normal fileystem, when it is
60 * first mounted).
61 *
62 * As hinted at above, there is one ZIL per dataset (both the in-memory
63 * representation, and the on-disk representation). The on-disk format
64 * consists of 3 parts:
65 *
66 * - a single, per-dataset, ZIL header; which points to a chain of
67 * - zero or more ZIL blocks; each of which contains
68 * - zero or more ZIL records
69 *
70 * A ZIL record holds the information necessary to replay a single
71 * system call transaction. A ZIL block can hold many ZIL records, and
72 * the blocks are chained together, similarly to a singly linked list.
73 *
74 * Each ZIL block contains a block pointer (blkptr_t) to the next ZIL
75 * block in the chain, and the ZIL header points to the first block in
76 * the chain.
77 *
78 * Note, there is not a fixed place in the pool to hold these ZIL
79 * blocks; they are dynamically allocated and freed as needed from the
80 * blocks available on the pool, though they can be preferentially
81 * allocated from a dedicated "log" vdev.
82 */
83
84 /*
85 * This controls the amount of time that a ZIL block (lwb) will remain
86 * "open" when it isn't "full", and it has a thread waiting for it to be
87 * committed to stable storage. Please refer to the zil_commit_waiter()
88 * function (and the comments within it) for more details.
89 */
90 int zfs_commit_timeout_pct = 5;
91
92 /*
93 * Disable intent logging replay. This global ZIL switch affects all pools.
94 */
95 int zil_replay_disable = 0; /* disable intent logging replay */
96
97 /*
98 * Tunable parameter for debugging or performance analysis. Setting
99 * zfs_nocacheflush will cause corruption on power loss if a volatile
100 * out-of-order write cache is enabled.
101 */
102 boolean_t zfs_nocacheflush = B_FALSE;
103
104 /*
105 * Limit SLOG write size per commit executed with synchronous priority.
106 * Any writes above that will be executed with lower (asynchronous) priority
107 * to limit potential SLOG device abuse by single active ZIL writer.
108 */
109 uint64_t zil_slog_bulk = 768 * 1024;
110
111 static kmem_cache_t *zil_lwb_cache;
112 static kmem_cache_t *zil_zcw_cache;
113
114 static void zil_async_to_sync(zilog_t *zilog, uint64_t foid);
115
116 #define LWB_EMPTY(lwb) ((BP_GET_LSIZE(&lwb->lwb_blk) - \
117 sizeof (zil_chain_t)) == (lwb->lwb_sz - lwb->lwb_nused))
118
119 static int
120 zil_bp_compare(const void *x1, const void *x2)
121 {
122 const dva_t *dva1 = &((zil_bp_node_t *)x1)->zn_dva;
123 const dva_t *dva2 = &((zil_bp_node_t *)x2)->zn_dva;
124
125 if (DVA_GET_VDEV(dva1) < DVA_GET_VDEV(dva2))
126 return (-1);
127 if (DVA_GET_VDEV(dva1) > DVA_GET_VDEV(dva2))
128 return (1);
129
130 if (DVA_GET_OFFSET(dva1) < DVA_GET_OFFSET(dva2))
131 return (-1);
132 if (DVA_GET_OFFSET(dva1) > DVA_GET_OFFSET(dva2))
133 return (1);
134
135 return (0);
136 }
137
138 static void
139 zil_bp_tree_init(zilog_t *zilog)
140 {
141 avl_create(&zilog->zl_bp_tree, zil_bp_compare,
142 sizeof (zil_bp_node_t), offsetof(zil_bp_node_t, zn_node));
143 }
144
145 static void
146 zil_bp_tree_fini(zilog_t *zilog)
147 {
148 avl_tree_t *t = &zilog->zl_bp_tree;
149 zil_bp_node_t *zn;
150 void *cookie = NULL;
151
152 while ((zn = avl_destroy_nodes(t, &cookie)) != NULL)
153 kmem_free(zn, sizeof (zil_bp_node_t));
154
155 avl_destroy(t);
156 }
157
158 int
159 zil_bp_tree_add(zilog_t *zilog, const blkptr_t *bp)
160 {
161 avl_tree_t *t = &zilog->zl_bp_tree;
162 const dva_t *dva;
163 zil_bp_node_t *zn;
164 avl_index_t where;
165
166 if (BP_IS_EMBEDDED(bp))
167 return (0);
168
169 dva = BP_IDENTITY(bp);
170
171 if (avl_find(t, dva, &where) != NULL)
172 return (SET_ERROR(EEXIST));
173
174 zn = kmem_alloc(sizeof (zil_bp_node_t), KM_SLEEP);
175 zn->zn_dva = *dva;
176 avl_insert(t, zn, where);
177
178 return (0);
179 }
180
181 static zil_header_t *
182 zil_header_in_syncing_context(zilog_t *zilog)
183 {
184 return ((zil_header_t *)zilog->zl_header);
185 }
186
187 static void
188 zil_init_log_chain(zilog_t *zilog, blkptr_t *bp)
189 {
190 zio_cksum_t *zc = &bp->blk_cksum;
191
192 zc->zc_word[ZIL_ZC_GUID_0] = spa_get_random(-1ULL);
193 zc->zc_word[ZIL_ZC_GUID_1] = spa_get_random(-1ULL);
194 zc->zc_word[ZIL_ZC_OBJSET] = dmu_objset_id(zilog->zl_os);
195 zc->zc_word[ZIL_ZC_SEQ] = 1ULL;
196 }
197
198 /*
199 * Read a log block and make sure it's valid.
200 */
201 static int
202 zil_read_log_block(zilog_t *zilog, const blkptr_t *bp, blkptr_t *nbp, void *dst,
203 char **end)
204 {
205 enum zio_flag zio_flags = ZIO_FLAG_CANFAIL;
206 arc_flags_t aflags = ARC_FLAG_WAIT;
207 arc_buf_t *abuf = NULL;
208 zbookmark_phys_t zb;
209 int error;
210
211 if (zilog->zl_header->zh_claim_txg == 0)
212 zio_flags |= ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB;
213
214 if (!(zilog->zl_header->zh_flags & ZIL_CLAIM_LR_SEQ_VALID))
215 zio_flags |= ZIO_FLAG_SPECULATIVE;
216
217 SET_BOOKMARK(&zb, bp->blk_cksum.zc_word[ZIL_ZC_OBJSET],
218 ZB_ZIL_OBJECT, ZB_ZIL_LEVEL, bp->blk_cksum.zc_word[ZIL_ZC_SEQ]);
219
220 error = arc_read(NULL, zilog->zl_spa, bp, arc_getbuf_func, &abuf,
221 ZIO_PRIORITY_SYNC_READ, zio_flags, &aflags, &zb);
222
223 if (error == 0) {
224 zio_cksum_t cksum = bp->blk_cksum;
225
226 /*
227 * Validate the checksummed log block.
228 *
229 * Sequence numbers should be... sequential. The checksum
230 * verifier for the next block should be bp's checksum plus 1.
231 *
232 * Also check the log chain linkage and size used.
233 */
234 cksum.zc_word[ZIL_ZC_SEQ]++;
235
236 if (BP_GET_CHECKSUM(bp) == ZIO_CHECKSUM_ZILOG2) {
237 zil_chain_t *zilc = abuf->b_data;
238 char *lr = (char *)(zilc + 1);
239 uint64_t len = zilc->zc_nused - sizeof (zil_chain_t);
240
241 if (bcmp(&cksum, &zilc->zc_next_blk.blk_cksum,
242 sizeof (cksum)) || BP_IS_HOLE(&zilc->zc_next_blk)) {
243 error = SET_ERROR(ECKSUM);
244 } else {
245 ASSERT3U(len, <=, SPA_OLD_MAXBLOCKSIZE);
246 bcopy(lr, dst, len);
247 *end = (char *)dst + len;
248 *nbp = zilc->zc_next_blk;
249 }
250 } else {
251 char *lr = abuf->b_data;
252 uint64_t size = BP_GET_LSIZE(bp);
253 zil_chain_t *zilc = (zil_chain_t *)(lr + size) - 1;
254
255 if (bcmp(&cksum, &zilc->zc_next_blk.blk_cksum,
256 sizeof (cksum)) || BP_IS_HOLE(&zilc->zc_next_blk) ||
257 (zilc->zc_nused > (size - sizeof (*zilc)))) {
258 error = SET_ERROR(ECKSUM);
259 } else {
260 ASSERT3U(zilc->zc_nused, <=,
261 SPA_OLD_MAXBLOCKSIZE);
262 bcopy(lr, dst, zilc->zc_nused);
263 *end = (char *)dst + zilc->zc_nused;
264 *nbp = zilc->zc_next_blk;
265 }
266 }
267
268 arc_buf_destroy(abuf, &abuf);
269 }
270
271 return (error);
272 }
273
274 /*
275 * Read a TX_WRITE log data block.
276 */
277 static int
278 zil_read_log_data(zilog_t *zilog, const lr_write_t *lr, void *wbuf)
279 {
280 enum zio_flag zio_flags = ZIO_FLAG_CANFAIL;
281 const blkptr_t *bp = &lr->lr_blkptr;
282 arc_flags_t aflags = ARC_FLAG_WAIT;
283 arc_buf_t *abuf = NULL;
284 zbookmark_phys_t zb;
285 int error;
286
287 if (BP_IS_HOLE(bp)) {
288 if (wbuf != NULL)
289 bzero(wbuf, MAX(BP_GET_LSIZE(bp), lr->lr_length));
290 return (0);
291 }
292
293 if (zilog->zl_header->zh_claim_txg == 0)
294 zio_flags |= ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB;
295
296 SET_BOOKMARK(&zb, dmu_objset_id(zilog->zl_os), lr->lr_foid,
297 ZB_ZIL_LEVEL, lr->lr_offset / BP_GET_LSIZE(bp));
298
299 error = arc_read(NULL, zilog->zl_spa, bp, arc_getbuf_func, &abuf,
300 ZIO_PRIORITY_SYNC_READ, zio_flags, &aflags, &zb);
301
302 if (error == 0) {
303 if (wbuf != NULL)
304 bcopy(abuf->b_data, wbuf, arc_buf_size(abuf));
305 arc_buf_destroy(abuf, &abuf);
306 }
307
308 return (error);
309 }
310
311 /*
312 * Parse the intent log, and call parse_func for each valid record within.
313 */
314 int
315 zil_parse(zilog_t *zilog, zil_parse_blk_func_t *parse_blk_func,
316 zil_parse_lr_func_t *parse_lr_func, void *arg, uint64_t txg)
317 {
318 const zil_header_t *zh = zilog->zl_header;
319 boolean_t claimed = !!zh->zh_claim_txg;
320 uint64_t claim_blk_seq = claimed ? zh->zh_claim_blk_seq : UINT64_MAX;
321 uint64_t claim_lr_seq = claimed ? zh->zh_claim_lr_seq : UINT64_MAX;
322 uint64_t max_blk_seq = 0;
323 uint64_t max_lr_seq = 0;
324 uint64_t blk_count = 0;
325 uint64_t lr_count = 0;
326 blkptr_t blk, next_blk;
327 char *lrbuf, *lrp;
328 int error = 0;
329
330 /*
331 * Old logs didn't record the maximum zh_claim_lr_seq.
332 */
333 if (!(zh->zh_flags & ZIL_CLAIM_LR_SEQ_VALID))
334 claim_lr_seq = UINT64_MAX;
335
336 /*
337 * Starting at the block pointed to by zh_log we read the log chain.
338 * For each block in the chain we strongly check that block to
339 * ensure its validity. We stop when an invalid block is found.
340 * For each block pointer in the chain we call parse_blk_func().
341 * For each record in each valid block we call parse_lr_func().
342 * If the log has been claimed, stop if we encounter a sequence
343 * number greater than the highest claimed sequence number.
344 */
345 lrbuf = zio_buf_alloc(SPA_OLD_MAXBLOCKSIZE);
346 zil_bp_tree_init(zilog);
347
348 for (blk = zh->zh_log; !BP_IS_HOLE(&blk); blk = next_blk) {
349 uint64_t blk_seq = blk.blk_cksum.zc_word[ZIL_ZC_SEQ];
350 int reclen;
351 char *end;
352
353 if (blk_seq > claim_blk_seq)
354 break;
355 if ((error = parse_blk_func(zilog, &blk, arg, txg)) != 0)
356 break;
357 ASSERT3U(max_blk_seq, <, blk_seq);
358 max_blk_seq = blk_seq;
359 blk_count++;
360
361 if (max_lr_seq == claim_lr_seq && max_blk_seq == claim_blk_seq)
362 break;
363
364 error = zil_read_log_block(zilog, &blk, &next_blk, lrbuf, &end);
365 if (error != 0)
366 break;
367
368 for (lrp = lrbuf; lrp < end; lrp += reclen) {
369 lr_t *lr = (lr_t *)lrp;
370 reclen = lr->lrc_reclen;
371 ASSERT3U(reclen, >=, sizeof (lr_t));
372 if (lr->lrc_seq > claim_lr_seq)
373 goto done;
374 if ((error = parse_lr_func(zilog, lr, arg, txg)) != 0)
375 goto done;
376 ASSERT3U(max_lr_seq, <, lr->lrc_seq);
377 max_lr_seq = lr->lrc_seq;
378 lr_count++;
379 }
380 }
381 done:
382 zilog->zl_parse_error = error;
383 zilog->zl_parse_blk_seq = max_blk_seq;
384 zilog->zl_parse_lr_seq = max_lr_seq;
385 zilog->zl_parse_blk_count = blk_count;
386 zilog->zl_parse_lr_count = lr_count;
387
388 ASSERT(!claimed || !(zh->zh_flags & ZIL_CLAIM_LR_SEQ_VALID) ||
389 (max_blk_seq == claim_blk_seq && max_lr_seq == claim_lr_seq));
390
391 zil_bp_tree_fini(zilog);
392 zio_buf_free(lrbuf, SPA_OLD_MAXBLOCKSIZE);
393
394 return (error);
395 }
396
397 static int
398 zil_claim_log_block(zilog_t *zilog, blkptr_t *bp, void *tx, uint64_t first_txg)
399 {
400 /*
401 * Claim log block if not already committed and not already claimed.
402 * If tx == NULL, just verify that the block is claimable.
403 */
404 if (BP_IS_HOLE(bp) || bp->blk_birth < first_txg ||
405 zil_bp_tree_add(zilog, bp) != 0)
406 return (0);
407
408 return (zio_wait(zio_claim(NULL, zilog->zl_spa,
409 tx == NULL ? 0 : first_txg, bp, spa_claim_notify, NULL,
410 ZIO_FLAG_CANFAIL | ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB)));
411 }
412
413 static int
414 zil_claim_log_record(zilog_t *zilog, lr_t *lrc, void *tx, uint64_t first_txg)
415 {
416 lr_write_t *lr = (lr_write_t *)lrc;
417 int error;
418
419 if (lrc->lrc_txtype != TX_WRITE)
420 return (0);
421
422 /*
423 * If the block is not readable, don't claim it. This can happen
424 * in normal operation when a log block is written to disk before
425 * some of the dmu_sync() blocks it points to. In this case, the
426 * transaction cannot have been committed to anyone (we would have
427 * waited for all writes to be stable first), so it is semantically
428 * correct to declare this the end of the log.
429 */
430 if (lr->lr_blkptr.blk_birth >= first_txg &&
431 (error = zil_read_log_data(zilog, lr, NULL)) != 0)
432 return (error);
433 return (zil_claim_log_block(zilog, &lr->lr_blkptr, tx, first_txg));
434 }
435
436 /* ARGSUSED */
437 static int
438 zil_free_log_block(zilog_t *zilog, blkptr_t *bp, void *tx, uint64_t claim_txg)
439 {
440 zio_free_zil(zilog->zl_spa, dmu_tx_get_txg(tx), bp);
441
442 return (0);
443 }
444
445 static int
446 zil_free_log_record(zilog_t *zilog, lr_t *lrc, void *tx, uint64_t claim_txg)
447 {
448 lr_write_t *lr = (lr_write_t *)lrc;
449 blkptr_t *bp = &lr->lr_blkptr;
450
451 /*
452 * If we previously claimed it, we need to free it.
453 */
454 if (claim_txg != 0 && lrc->lrc_txtype == TX_WRITE &&
455 bp->blk_birth >= claim_txg && zil_bp_tree_add(zilog, bp) == 0 &&
456 !BP_IS_HOLE(bp))
457 zio_free(zilog->zl_spa, dmu_tx_get_txg(tx), bp);
458
459 return (0);
460 }
461
462 static int
463 zil_lwb_vdev_compare(const void *x1, const void *x2)
464 {
465 const uint64_t v1 = ((zil_vdev_node_t *)x1)->zv_vdev;
466 const uint64_t v2 = ((zil_vdev_node_t *)x2)->zv_vdev;
467
468 if (v1 < v2)
469 return (-1);
470 if (v1 > v2)
471 return (1);
472
473 return (0);
474 }
475
476 static lwb_t *
477 zil_alloc_lwb(zilog_t *zilog, blkptr_t *bp, boolean_t slog, uint64_t txg)
478 {
479 lwb_t *lwb;
480
481 lwb = kmem_cache_alloc(zil_lwb_cache, KM_SLEEP);
482 lwb->lwb_zilog = zilog;
483 lwb->lwb_blk = *bp;
484 lwb->lwb_slog = slog;
485 lwb->lwb_state = LWB_STATE_CLOSED;
486 lwb->lwb_buf = zio_buf_alloc(BP_GET_LSIZE(bp));
487 lwb->lwb_max_txg = txg;
488 lwb->lwb_write_zio = NULL;
489 lwb->lwb_root_zio = NULL;
490 lwb->lwb_tx = NULL;
491 lwb->lwb_issued_timestamp = 0;
492 if (BP_GET_CHECKSUM(bp) == ZIO_CHECKSUM_ZILOG2) {
493 lwb->lwb_nused = sizeof (zil_chain_t);
494 lwb->lwb_sz = BP_GET_LSIZE(bp);
495 } else {
496 lwb->lwb_nused = 0;
497 lwb->lwb_sz = BP_GET_LSIZE(bp) - sizeof (zil_chain_t);
498 }
499
500 mutex_enter(&zilog->zl_lock);
501 list_insert_tail(&zilog->zl_lwb_list, lwb);
502 mutex_exit(&zilog->zl_lock);
503
504 ASSERT(!MUTEX_HELD(&lwb->lwb_vdev_lock));
505 ASSERT(avl_is_empty(&lwb->lwb_vdev_tree));
506 VERIFY(list_is_empty(&lwb->lwb_waiters));
507
508 return (lwb);
509 }
510
511 static void
512 zil_free_lwb(zilog_t *zilog, lwb_t *lwb)
513 {
514 ASSERT(MUTEX_HELD(&zilog->zl_lock));
515 ASSERT(!MUTEX_HELD(&lwb->lwb_vdev_lock));
516 VERIFY(list_is_empty(&lwb->lwb_waiters));
517 ASSERT(avl_is_empty(&lwb->lwb_vdev_tree));
518 ASSERT3P(lwb->lwb_write_zio, ==, NULL);
519 ASSERT3P(lwb->lwb_root_zio, ==, NULL);
520 ASSERT3U(lwb->lwb_max_txg, <=, spa_syncing_txg(zilog->zl_spa));
521 ASSERT(lwb->lwb_state == LWB_STATE_CLOSED ||
522 lwb->lwb_state == LWB_STATE_FLUSH_DONE);
523
524 /*
525 * Clear the zilog's field to indicate this lwb is no longer
526 * valid, and prevent use-after-free errors.
527 */
528 if (zilog->zl_last_lwb_opened == lwb)
529 zilog->zl_last_lwb_opened = NULL;
530
531 kmem_cache_free(zil_lwb_cache, lwb);
532 }
533
534 /*
535 * Called when we create in-memory log transactions so that we know
536 * to cleanup the itxs at the end of spa_sync().
537 */
538 void
539 zilog_dirty(zilog_t *zilog, uint64_t txg)
540 {
541 dsl_pool_t *dp = zilog->zl_dmu_pool;
542 dsl_dataset_t *ds = dmu_objset_ds(zilog->zl_os);
543
544 ASSERT(spa_writeable(zilog->zl_spa));
545
546 if (ds->ds_is_snapshot)
547 panic("dirtying snapshot!");
548
549 if (txg_list_add(&dp->dp_dirty_zilogs, zilog, txg)) {
550 /* up the hold count until we can be written out */
551 dmu_buf_add_ref(ds->ds_dbuf, zilog);
552
553 zilog->zl_dirty_max_txg = MAX(txg, zilog->zl_dirty_max_txg);
554 }
555 }
556
557 /*
558 * Determine if the zil is dirty in the specified txg. Callers wanting to
559 * ensure that the dirty state does not change must hold the itxg_lock for
560 * the specified txg. Holding the lock will ensure that the zil cannot be
561 * dirtied (zil_itx_assign) or cleaned (zil_clean) while we check its current
562 * state.
563 */
564 boolean_t
565 zilog_is_dirty_in_txg(zilog_t *zilog, uint64_t txg)
566 {
567 dsl_pool_t *dp = zilog->zl_dmu_pool;
568
569 if (txg_list_member(&dp->dp_dirty_zilogs, zilog, txg & TXG_MASK))
570 return (B_TRUE);
571 return (B_FALSE);
572 }
573
574 /*
575 * Determine if the zil is dirty. The zil is considered dirty if it has
576 * any pending itx records that have not been cleaned by zil_clean().
577 */
578 boolean_t
579 zilog_is_dirty(zilog_t *zilog)
580 {
581 dsl_pool_t *dp = zilog->zl_dmu_pool;
582
583 for (int t = 0; t < TXG_SIZE; t++) {
584 if (txg_list_member(&dp->dp_dirty_zilogs, zilog, t))
585 return (B_TRUE);
586 }
587 return (B_FALSE);
588 }
589
590 /*
591 * Create an on-disk intent log.
592 */
593 static lwb_t *
594 zil_create(zilog_t *zilog)
595 {
596 const zil_header_t *zh = zilog->zl_header;
597 lwb_t *lwb = NULL;
598 uint64_t txg = 0;
599 dmu_tx_t *tx = NULL;
600 blkptr_t blk;
601 int error = 0;
602 boolean_t slog = FALSE;
603
604 /*
605 * Wait for any previous destroy to complete.
606 */
607 txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
608
609 ASSERT(zh->zh_claim_txg == 0);
610 ASSERT(zh->zh_replay_seq == 0);
611
612 blk = zh->zh_log;
613
614 /*
615 * Allocate an initial log block if:
616 * - there isn't one already
617 * - the existing block is the wrong endianess
618 */
619 if (BP_IS_HOLE(&blk) || BP_SHOULD_BYTESWAP(&blk)) {
620 tx = dmu_tx_create(zilog->zl_os);
621 VERIFY0(dmu_tx_assign(tx, TXG_WAIT));
622 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
623 txg = dmu_tx_get_txg(tx);
624
625 if (!BP_IS_HOLE(&blk)) {
626 zio_free_zil(zilog->zl_spa, txg, &blk);
627 BP_ZERO(&blk);
628 }
629
630 error = zio_alloc_zil(zilog->zl_spa, txg, &blk, NULL,
631 ZIL_MIN_BLKSZ, &slog);
632
633 if (error == 0)
634 zil_init_log_chain(zilog, &blk);
635 }
636
637 /*
638 * Allocate a log write block (lwb) for the first log block.
639 */
640 if (error == 0)
641 lwb = zil_alloc_lwb(zilog, &blk, slog, txg);
642
643 /*
644 * If we just allocated the first log block, commit our transaction
645 * and wait for zil_sync() to stuff the block poiner into zh_log.
646 * (zh is part of the MOS, so we cannot modify it in open context.)
647 */
648 if (tx != NULL) {
649 dmu_tx_commit(tx);
650 txg_wait_synced(zilog->zl_dmu_pool, txg);
651 }
652
653 ASSERT(bcmp(&blk, &zh->zh_log, sizeof (blk)) == 0);
654
655 return (lwb);
656 }
657
658 /*
659 * In one tx, free all log blocks and clear the log header. If keep_first
660 * is set, then we're replaying a log with no content. We want to keep the
661 * first block, however, so that the first synchronous transaction doesn't
662 * require a txg_wait_synced() in zil_create(). We don't need to
663 * txg_wait_synced() here either when keep_first is set, because both
664 * zil_create() and zil_destroy() will wait for any in-progress destroys
665 * to complete.
666 */
667 void
668 zil_destroy(zilog_t *zilog, boolean_t keep_first)
669 {
670 const zil_header_t *zh = zilog->zl_header;
671 lwb_t *lwb;
672 dmu_tx_t *tx;
673 uint64_t txg;
674
675 /*
676 * Wait for any previous destroy to complete.
677 */
678 txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
679
680 zilog->zl_old_header = *zh; /* debugging aid */
681
682 if (BP_IS_HOLE(&zh->zh_log))
683 return;
684
685 tx = dmu_tx_create(zilog->zl_os);
686 VERIFY0(dmu_tx_assign(tx, TXG_WAIT));
687 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
688 txg = dmu_tx_get_txg(tx);
689
690 mutex_enter(&zilog->zl_lock);
691
692 ASSERT3U(zilog->zl_destroy_txg, <, txg);
693 zilog->zl_destroy_txg = txg;
694 zilog->zl_keep_first = keep_first;
695
696 if (!list_is_empty(&zilog->zl_lwb_list)) {
697 ASSERT(zh->zh_claim_txg == 0);
698 VERIFY(!keep_first);
699 while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) {
700 list_remove(&zilog->zl_lwb_list, lwb);
701 if (lwb->lwb_buf != NULL)
702 zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
703 zio_free(zilog->zl_spa, txg, &lwb->lwb_blk);
704 zil_free_lwb(zilog, lwb);
705 }
706 } else if (!keep_first) {
707 zil_destroy_sync(zilog, tx);
708 }
709 mutex_exit(&zilog->zl_lock);
710
711 dmu_tx_commit(tx);
712 }
713
714 void
715 zil_destroy_sync(zilog_t *zilog, dmu_tx_t *tx)
716 {
717 ASSERT(list_is_empty(&zilog->zl_lwb_list));
718 (void) zil_parse(zilog, zil_free_log_block,
719 zil_free_log_record, tx, zilog->zl_header->zh_claim_txg);
720 }
721
722 int
723 zil_claim(dsl_pool_t *dp, dsl_dataset_t *ds, void *txarg)
724 {
725 dmu_tx_t *tx = txarg;
726 uint64_t first_txg = dmu_tx_get_txg(tx);
727 zilog_t *zilog;
728 zil_header_t *zh;
729 objset_t *os;
730 int error;
731
732 error = dmu_objset_own_obj(dp, ds->ds_object,
733 DMU_OST_ANY, B_FALSE, FTAG, &os);
734 if (error != 0) {
735 /*
736 * EBUSY indicates that the objset is inconsistent, in which
737 * case it can not have a ZIL.
738 */
739 if (error != EBUSY) {
740 cmn_err(CE_WARN, "can't open objset for %llu, error %u",
741 (unsigned long long)ds->ds_object, error);
742 }
743 return (0);
744 }
745
746 zilog = dmu_objset_zil(os);
747 zh = zil_header_in_syncing_context(zilog);
748
749 if (spa_get_log_state(zilog->zl_spa) == SPA_LOG_CLEAR) {
750 if (!BP_IS_HOLE(&zh->zh_log))
751 zio_free_zil(zilog->zl_spa, first_txg, &zh->zh_log);
752 BP_ZERO(&zh->zh_log);
753 dsl_dataset_dirty(dmu_objset_ds(os), tx);
754 dmu_objset_disown(os, FTAG);
755 return (0);
756 }
757
758 /*
759 * Claim all log blocks if we haven't already done so, and remember
760 * the highest claimed sequence number. This ensures that if we can
761 * read only part of the log now (e.g. due to a missing device),
762 * but we can read the entire log later, we will not try to replay
763 * or destroy beyond the last block we successfully claimed.
764 */
765 ASSERT3U(zh->zh_claim_txg, <=, first_txg);
766 if (zh->zh_claim_txg == 0 && !BP_IS_HOLE(&zh->zh_log)) {
767 (void) zil_parse(zilog, zil_claim_log_block,
768 zil_claim_log_record, tx, first_txg);
769 zh->zh_claim_txg = first_txg;
770 zh->zh_claim_blk_seq = zilog->zl_parse_blk_seq;
771 zh->zh_claim_lr_seq = zilog->zl_parse_lr_seq;
772 if (zilog->zl_parse_lr_count || zilog->zl_parse_blk_count > 1)
773 zh->zh_flags |= ZIL_REPLAY_NEEDED;
774 zh->zh_flags |= ZIL_CLAIM_LR_SEQ_VALID;
775 dsl_dataset_dirty(dmu_objset_ds(os), tx);
776 }
777
778 ASSERT3U(first_txg, ==, (spa_last_synced_txg(zilog->zl_spa) + 1));
779 dmu_objset_disown(os, FTAG);
780 return (0);
781 }
782
783 /*
784 * Check the log by walking the log chain.
785 * Checksum errors are ok as they indicate the end of the chain.
786 * Any other error (no device or read failure) returns an error.
787 */
788 /* ARGSUSED */
789 int
790 zil_check_log_chain(dsl_pool_t *dp, dsl_dataset_t *ds, void *tx)
791 {
792 zilog_t *zilog;
793 objset_t *os;
794 blkptr_t *bp;
795 int error;
796
797 ASSERT(tx == NULL);
798
799 error = dmu_objset_from_ds(ds, &os);
800 if (error != 0) {
801 cmn_err(CE_WARN, "can't open objset %llu, error %d",
802 (unsigned long long)ds->ds_object, error);
803 return (0);
804 }
805
806 zilog = dmu_objset_zil(os);
807 bp = (blkptr_t *)&zilog->zl_header->zh_log;
808
809 /*
810 * Check the first block and determine if it's on a log device
811 * which may have been removed or faulted prior to loading this
812 * pool. If so, there's no point in checking the rest of the log
813 * as its content should have already been synced to the pool.
814 */
815 if (!BP_IS_HOLE(bp)) {
816 vdev_t *vd;
817 boolean_t valid = B_TRUE;
818
819 spa_config_enter(os->os_spa, SCL_STATE, FTAG, RW_READER);
820 vd = vdev_lookup_top(os->os_spa, DVA_GET_VDEV(&bp->blk_dva[0]));
821 if (vd->vdev_islog && vdev_is_dead(vd))
822 valid = vdev_log_state_valid(vd);
823 spa_config_exit(os->os_spa, SCL_STATE, FTAG);
824
825 if (!valid)
826 return (0);
827 }
828
829 /*
830 * Because tx == NULL, zil_claim_log_block() will not actually claim
831 * any blocks, but just determine whether it is possible to do so.
832 * In addition to checking the log chain, zil_claim_log_block()
833 * will invoke zio_claim() with a done func of spa_claim_notify(),
834 * which will update spa_max_claim_txg. See spa_load() for details.
835 */
836 error = zil_parse(zilog, zil_claim_log_block, zil_claim_log_record, tx,
837 zilog->zl_header->zh_claim_txg ? -1ULL : spa_first_txg(os->os_spa));
838
839 return ((error == ECKSUM || error == ENOENT) ? 0 : error);
840 }
841
842 /*
843 * When an itx is "skipped", this function is used to properly mark the
844 * waiter as "done, and signal any thread(s) waiting on it. An itx can
845 * be skipped (and not committed to an lwb) for a variety of reasons,
846 * one of them being that the itx was committed via spa_sync(), prior to
847 * it being committed to an lwb; this can happen if a thread calling
848 * zil_commit() is racing with spa_sync().
849 */
850 static void
851 zil_commit_waiter_skip(zil_commit_waiter_t *zcw)
852 {
853 mutex_enter(&zcw->zcw_lock);
854 ASSERT3B(zcw->zcw_done, ==, B_FALSE);
855 zcw->zcw_done = B_TRUE;
856 cv_broadcast(&zcw->zcw_cv);
857 mutex_exit(&zcw->zcw_lock);
858 }
859
860 /*
861 * This function is used when the given waiter is to be linked into an
862 * lwb's "lwb_waiter" list; i.e. when the itx is committed to the lwb.
863 * At this point, the waiter will no longer be referenced by the itx,
864 * and instead, will be referenced by the lwb.
865 */
866 static void
867 zil_commit_waiter_link_lwb(zil_commit_waiter_t *zcw, lwb_t *lwb)
868 {
869 /*
870 * The lwb_waiters field of the lwb is protected by the zilog's
871 * zl_lock, thus it must be held when calling this function.
872 */
873 ASSERT(MUTEX_HELD(&lwb->lwb_zilog->zl_lock));
874
875 mutex_enter(&zcw->zcw_lock);
876 ASSERT(!list_link_active(&zcw->zcw_node));
877 ASSERT3P(zcw->zcw_lwb, ==, NULL);
878 ASSERT3P(lwb, !=, NULL);
879 ASSERT(lwb->lwb_state == LWB_STATE_OPENED ||
880 lwb->lwb_state == LWB_STATE_ISSUED ||
881 lwb->lwb_state == LWB_STATE_WRITE_DONE);
882
883 list_insert_tail(&lwb->lwb_waiters, zcw);
884 zcw->zcw_lwb = lwb;
885 mutex_exit(&zcw->zcw_lock);
886 }
887
888 /*
889 * This function is used when zio_alloc_zil() fails to allocate a ZIL
890 * block, and the given waiter must be linked to the "nolwb waiters"
891 * list inside of zil_process_commit_list().
892 */
893 static void
894 zil_commit_waiter_link_nolwb(zil_commit_waiter_t *zcw, list_t *nolwb)
895 {
896 mutex_enter(&zcw->zcw_lock);
897 ASSERT(!list_link_active(&zcw->zcw_node));
898 ASSERT3P(zcw->zcw_lwb, ==, NULL);
899 list_insert_tail(nolwb, zcw);
900 mutex_exit(&zcw->zcw_lock);
901 }
902
903 void
904 zil_lwb_add_block(lwb_t *lwb, const blkptr_t *bp)
905 {
906 avl_tree_t *t = &lwb->lwb_vdev_tree;
907 avl_index_t where;
908 zil_vdev_node_t *zv, zvsearch;
909 int ndvas = BP_GET_NDVAS(bp);
910 int i;
911
912 if (zfs_nocacheflush)
913 return;
914
915 mutex_enter(&lwb->lwb_vdev_lock);
916 for (i = 0; i < ndvas; i++) {
917 zvsearch.zv_vdev = DVA_GET_VDEV(&bp->blk_dva[i]);
918 if (avl_find(t, &zvsearch, &where) == NULL) {
919 zv = kmem_alloc(sizeof (*zv), KM_SLEEP);
920 zv->zv_vdev = zvsearch.zv_vdev;
921 avl_insert(t, zv, where);
922 }
923 }
924 mutex_exit(&lwb->lwb_vdev_lock);
925 }
926
927 static void
928 zil_lwb_flush_defer(lwb_t *lwb, lwb_t *nlwb)
929 {
930 avl_tree_t *src = &lwb->lwb_vdev_tree;
931 avl_tree_t *dst = &nlwb->lwb_vdev_tree;
932 void *cookie = NULL;
933 zil_vdev_node_t *zv;
934
935 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_WRITE_DONE);
936 ASSERT3S(nlwb->lwb_state, !=, LWB_STATE_WRITE_DONE);
937 ASSERT3S(nlwb->lwb_state, !=, LWB_STATE_FLUSH_DONE);
938
939 /*
940 * While 'lwb' is at a point in its lifetime where lwb_vdev_tree does
941 * not need the protection of lwb_vdev_lock (it will only be modified
942 * while holding zilog->zl_lock) as its writes and those of its
943 * children have all completed. The younger 'nlwb' may be waiting on
944 * future writes to additional vdevs.
945 */
946 mutex_enter(&nlwb->lwb_vdev_lock);
947 /*
948 * Tear down the 'lwb' vdev tree, ensuring that entries which do not
949 * exist in 'nlwb' are moved to it, freeing any would-be duplicates.
950 */
951 while ((zv = avl_destroy_nodes(src, &cookie)) != NULL) {
952 avl_index_t where;
953
954 if (avl_find(dst, zv, &where) == NULL) {
955 avl_insert(dst, zv, where);
956 } else {
957 kmem_free(zv, sizeof (*zv));
958 }
959 }
960 mutex_exit(&nlwb->lwb_vdev_lock);
961 }
962
963 void
964 zil_lwb_add_txg(lwb_t *lwb, uint64_t txg)
965 {
966 lwb->lwb_max_txg = MAX(lwb->lwb_max_txg, txg);
967 }
968
969 /*
970 * This function is a called after all vdevs associated with a given lwb
971 * write have completed their DKIOCFLUSHWRITECACHE command; or as soon
972 * as the lwb write completes, if "zil_nocacheflush" is set. Further,
973 * all "previous" lwb's will have completed before this function is
974 * called; i.e. this function is called for all previous lwbs before
975 * it's called for "this" lwb (enforced via zio the dependencies
976 * configured in zil_lwb_set_zio_dependency()).
977 *
978 * The intention is for this function to be called as soon as the
979 * contents of an lwb are considered "stable" on disk, and will survive
980 * any sudden loss of power. At this point, any threads waiting for the
981 * lwb to reach this state are signalled, and the "waiter" structures
982 * are marked "done".
983 */
984 static void
985 zil_lwb_flush_vdevs_done(zio_t *zio)
986 {
987 lwb_t *lwb = zio->io_private;
988 zilog_t *zilog = lwb->lwb_zilog;
989 dmu_tx_t *tx = lwb->lwb_tx;
990 zil_commit_waiter_t *zcw;
991
992 spa_config_exit(zilog->zl_spa, SCL_STATE, lwb);
993
994 zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
995
996 mutex_enter(&zilog->zl_lock);
997
998 /*
999 * Ensure the lwb buffer pointer is cleared before releasing the
1000 * txg. If we have had an allocation failure and the txg is
1001 * waiting to sync then we want zil_sync() to remove the lwb so
1002 * that it's not picked up as the next new one in
1003 * zil_process_commit_list(). zil_sync() will only remove the
1004 * lwb if lwb_buf is null.
1005 */
1006 lwb->lwb_buf = NULL;
1007 lwb->lwb_tx = NULL;
1008
1009 ASSERT3U(lwb->lwb_issued_timestamp, >, 0);
1010 zilog->zl_last_lwb_latency = gethrtime() - lwb->lwb_issued_timestamp;
1011
1012 lwb->lwb_root_zio = NULL;
1013
1014 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_WRITE_DONE);
1015 lwb->lwb_state = LWB_STATE_FLUSH_DONE;
1016
1017 if (zilog->zl_last_lwb_opened == lwb) {
1018 /*
1019 * Remember the highest committed log sequence number
1020 * for ztest. We only update this value when all the log
1021 * writes succeeded, because ztest wants to ASSERT that
1022 * it got the whole log chain.
1023 */
1024 zilog->zl_commit_lr_seq = zilog->zl_lr_seq;
1025 }
1026
1027 while ((zcw = list_head(&lwb->lwb_waiters)) != NULL) {
1028 mutex_enter(&zcw->zcw_lock);
1029
1030 ASSERT(list_link_active(&zcw->zcw_node));
1031 list_remove(&lwb->lwb_waiters, zcw);
1032
1033 ASSERT3P(zcw->zcw_lwb, ==, lwb);
1034 zcw->zcw_lwb = NULL;
1035
1036 zcw->zcw_zio_error = zio->io_error;
1037
1038 ASSERT3B(zcw->zcw_done, ==, B_FALSE);
1039 zcw->zcw_done = B_TRUE;
1040 cv_broadcast(&zcw->zcw_cv);
1041
1042 mutex_exit(&zcw->zcw_lock);
1043 }
1044
1045 mutex_exit(&zilog->zl_lock);
1046
1047 /*
1048 * Now that we've written this log block, we have a stable pointer
1049 * to the next block in the chain, so it's OK to let the txg in
1050 * which we allocated the next block sync.
1051 */
1052 dmu_tx_commit(tx);
1053 }
1054
1055 /*
1056 * This is called when an lwb's write zio completes. The callback's
1057 * purpose is to issue the DKIOCFLUSHWRITECACHE commands for the vdevs
1058 * in the lwb's lwb_vdev_tree. The tree will contain the vdevs involved
1059 * in writing out this specific lwb's data, and in the case that cache
1060 * flushes have been deferred, vdevs involved in writing the data for
1061 * previous lwbs. The writes corresponding to all the vdevs in the
1062 * lwb_vdev_tree will have completed by the time this is called, due to
1063 * the zio dependencies configured in zil_lwb_set_zio_dependency(),
1064 * which takes deferred flushes into account. The lwb will be "done"
1065 * once zil_lwb_flush_vdevs_done() is called, which occurs in the zio
1066 * completion callback for the lwb's root zio.
1067 */
1068 static void
1069 zil_lwb_write_done(zio_t *zio)
1070 {
1071 lwb_t *lwb = zio->io_private;
1072 spa_t *spa = zio->io_spa;
1073 zilog_t *zilog = lwb->lwb_zilog;
1074 avl_tree_t *t = &lwb->lwb_vdev_tree;
1075 void *cookie = NULL;
1076 zil_vdev_node_t *zv;
1077 lwb_t *nlwb;
1078
1079 ASSERT3S(spa_config_held(spa, SCL_STATE, RW_READER), !=, 0);
1080
1081 ASSERT(BP_GET_COMPRESS(zio->io_bp) == ZIO_COMPRESS_OFF);
1082 ASSERT(BP_GET_TYPE(zio->io_bp) == DMU_OT_INTENT_LOG);
1083 ASSERT(BP_GET_LEVEL(zio->io_bp) == 0);
1084 ASSERT(BP_GET_BYTEORDER(zio->io_bp) == ZFS_HOST_BYTEORDER);
1085 ASSERT(!BP_IS_GANG(zio->io_bp));
1086 ASSERT(!BP_IS_HOLE(zio->io_bp));
1087 ASSERT(BP_GET_FILL(zio->io_bp) == 0);
1088
1089 abd_put(zio->io_abd);
1090
1091 mutex_enter(&zilog->zl_lock);
1092 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_ISSUED);
1093 lwb->lwb_state = LWB_STATE_WRITE_DONE;
1094 lwb->lwb_write_zio = NULL;
1095 nlwb = list_next(&zilog->zl_lwb_list, lwb);
1096 mutex_exit(&zilog->zl_lock);
1097
1098 if (avl_numnodes(t) == 0)
1099 return;
1100
1101 /*
1102 * If there was an IO error, we're not going to call zio_flush()
1103 * on these vdevs, so we simply empty the tree and free the
1104 * nodes. We avoid calling zio_flush() since there isn't any
1105 * good reason for doing so, after the lwb block failed to be
1106 * written out.
1107 */
1108 if (zio->io_error != 0) {
1109 while ((zv = avl_destroy_nodes(t, &cookie)) != NULL)
1110 kmem_free(zv, sizeof (*zv));
1111 return;
1112 }
1113
1114 /*
1115 * If this lwb does not have any threads waiting for it to
1116 * complete, we want to defer issuing the DKIOCFLUSHWRITECACHE
1117 * command to the vdevs written to by "this" lwb, and instead
1118 * rely on the "next" lwb to handle the DKIOCFLUSHWRITECACHE
1119 * command for those vdevs. Thus, we merge the vdev tree of
1120 * "this" lwb with the vdev tree of the "next" lwb in the list,
1121 * and assume the "next" lwb will handle flushing the vdevs (or
1122 * deferring the flush(s) again).
1123 *
1124 * This is a useful performance optimization, especially for
1125 * workloads with lots of async write activity and few sync
1126 * write and/or fsync activity, as it has the potential to
1127 * coalesce multiple flush commands to a vdev into one.
1128 */
1129 if (list_head(&lwb->lwb_waiters) == NULL && nlwb != NULL) {
1130 zil_lwb_flush_defer(lwb, nlwb);
1131 ASSERT(avl_is_empty(&lwb->lwb_vdev_tree));
1132 return;
1133 }
1134
1135 while ((zv = avl_destroy_nodes(t, &cookie)) != NULL) {
1136 vdev_t *vd = vdev_lookup_top(spa, zv->zv_vdev);
1137 if (vd != NULL)
1138 zio_flush(lwb->lwb_root_zio, vd);
1139 kmem_free(zv, sizeof (*zv));
1140 }
1141 }
1142
1143 static void
1144 zil_lwb_set_zio_dependency(zilog_t *zilog, lwb_t *lwb)
1145 {
1146 lwb_t *last_lwb_opened = zilog->zl_last_lwb_opened;
1147
1148 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock));
1149 ASSERT(MUTEX_HELD(&zilog->zl_lock));
1150
1151 /*
1152 * The zilog's "zl_last_lwb_opened" field is used to build the
1153 * lwb/zio dependency chain, which is used to preserve the
1154 * ordering of lwb completions that is required by the semantics
1155 * of the ZIL. Each new lwb zio becomes a parent of the
1156 * "previous" lwb zio, such that the new lwb's zio cannot
1157 * complete until the "previous" lwb's zio completes.
1158 *
1159 * This is required by the semantics of zil_commit(); the commit
1160 * waiters attached to the lwbs will be woken in the lwb zio's
1161 * completion callback, so this zio dependency graph ensures the
1162 * waiters are woken in the correct order (the same order the
1163 * lwbs were created).
1164 */
1165 if (last_lwb_opened != NULL &&
1166 last_lwb_opened->lwb_state != LWB_STATE_FLUSH_DONE) {
1167 ASSERT(last_lwb_opened->lwb_state == LWB_STATE_OPENED ||
1168 last_lwb_opened->lwb_state == LWB_STATE_ISSUED ||
1169 last_lwb_opened->lwb_state == LWB_STATE_WRITE_DONE);
1170
1171 ASSERT3P(last_lwb_opened->lwb_root_zio, !=, NULL);
1172 zio_add_child(lwb->lwb_root_zio,
1173 last_lwb_opened->lwb_root_zio);
1174
1175 /*
1176 * If the previous lwb's write hasn't already completed,
1177 * we also want to order the completion of the lwb write
1178 * zios (above, we only order the completion of the lwb
1179 * root zios). This is required because of how we can
1180 * defer the DKIOCFLUSHWRITECACHE commands for each lwb.
1181 *
1182 * When the DKIOCFLUSHWRITECACHE commands are defered,
1183 * the previous lwb will rely on this lwb to flush the
1184 * vdevs written to by that previous lwb. Thus, we need
1185 * to ensure this lwb doesn't issue the flush until
1186 * after the previous lwb's write completes. We ensure
1187 * this ordering by setting the zio parent/child
1188 * relationship here.
1189 *
1190 * Without this relationship on the lwb's write zio,
1191 * it's possible for this lwb's write to complete prior
1192 * to the previous lwb's write completing; and thus, the
1193 * vdevs for the previous lwb would be flushed prior to
1194 * that lwb's data being written to those vdevs (the
1195 * vdevs are flushed in the lwb write zio's completion
1196 * handler, zil_lwb_write_done()).
1197 */
1198 if (last_lwb_opened->lwb_state != LWB_STATE_WRITE_DONE) {
1199 ASSERT(last_lwb_opened->lwb_state == LWB_STATE_OPENED ||
1200 last_lwb_opened->lwb_state == LWB_STATE_ISSUED);
1201
1202 ASSERT3P(last_lwb_opened->lwb_write_zio, !=, NULL);
1203 zio_add_child(lwb->lwb_write_zio,
1204 last_lwb_opened->lwb_write_zio);
1205 }
1206 }
1207 }
1208
1209
1210 /*
1211 * This function's purpose is to "open" an lwb such that it is ready to
1212 * accept new itxs being committed to it. To do this, the lwb's zio
1213 * structures are created, and linked to the lwb. This function is
1214 * idempotent; if the passed in lwb has already been opened, this
1215 * function is essentially a no-op.
1216 */
1217 static void
1218 zil_lwb_write_open(zilog_t *zilog, lwb_t *lwb)
1219 {
1220 zbookmark_phys_t zb;
1221 zio_priority_t prio;
1222
1223 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock));
1224 ASSERT3P(lwb, !=, NULL);
1225 EQUIV(lwb->lwb_root_zio == NULL, lwb->lwb_state == LWB_STATE_CLOSED);
1226 EQUIV(lwb->lwb_root_zio != NULL, lwb->lwb_state == LWB_STATE_OPENED);
1227
1228 SET_BOOKMARK(&zb, lwb->lwb_blk.blk_cksum.zc_word[ZIL_ZC_OBJSET],
1229 ZB_ZIL_OBJECT, ZB_ZIL_LEVEL,
1230 lwb->lwb_blk.blk_cksum.zc_word[ZIL_ZC_SEQ]);
1231
1232 if (lwb->lwb_root_zio == NULL) {
1233 abd_t *lwb_abd = abd_get_from_buf(lwb->lwb_buf,
1234 BP_GET_LSIZE(&lwb->lwb_blk));
1235
1236 if (!lwb->lwb_slog || zilog->zl_cur_used <= zil_slog_bulk)
1237 prio = ZIO_PRIORITY_SYNC_WRITE;
1238 else
1239 prio = ZIO_PRIORITY_ASYNC_WRITE;
1240
1241 lwb->lwb_root_zio = zio_root(zilog->zl_spa,
1242 zil_lwb_flush_vdevs_done, lwb, ZIO_FLAG_CANFAIL);
1243 ASSERT3P(lwb->lwb_root_zio, !=, NULL);
1244
1245 lwb->lwb_write_zio = zio_rewrite(lwb->lwb_root_zio,
1246 zilog->zl_spa, 0, &lwb->lwb_blk, lwb_abd,
1247 BP_GET_LSIZE(&lwb->lwb_blk), zil_lwb_write_done, lwb,
1248 prio, ZIO_FLAG_CANFAIL | ZIO_FLAG_DONT_PROPAGATE, &zb);
1249 ASSERT3P(lwb->lwb_write_zio, !=, NULL);
1250
1251 lwb->lwb_state = LWB_STATE_OPENED;
1252
1253 mutex_enter(&zilog->zl_lock);
1254 zil_lwb_set_zio_dependency(zilog, lwb);
1255 zilog->zl_last_lwb_opened = lwb;
1256 mutex_exit(&zilog->zl_lock);
1257 }
1258
1259 ASSERT3P(lwb->lwb_root_zio, !=, NULL);
1260 ASSERT3P(lwb->lwb_write_zio, !=, NULL);
1261 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_OPENED);
1262 }
1263
1264 /*
1265 * Define a limited set of intent log block sizes.
1266 *
1267 * These must be a multiple of 4KB. Note only the amount used (again
1268 * aligned to 4KB) actually gets written. However, we can't always just
1269 * allocate SPA_OLD_MAXBLOCKSIZE as the slog space could be exhausted.
1270 */
1271 uint64_t zil_block_buckets[] = {
1272 4096, /* non TX_WRITE */
1273 8192+4096, /* data base */
1274 32*1024 + 4096, /* NFS writes */
1275 UINT64_MAX
1276 };
1277
1278 /*
1279 * Start a log block write and advance to the next log block.
1280 * Calls are serialized.
1281 */
1282 static lwb_t *
1283 zil_lwb_write_issue(zilog_t *zilog, lwb_t *lwb)
1284 {
1285 lwb_t *nlwb = NULL;
1286 zil_chain_t *zilc;
1287 spa_t *spa = zilog->zl_spa;
1288 blkptr_t *bp;
1289 dmu_tx_t *tx;
1290 uint64_t txg;
1291 uint64_t zil_blksz, wsz;
1292 int i, error;
1293 boolean_t slog;
1294
1295 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock));
1296 ASSERT3P(lwb->lwb_root_zio, !=, NULL);
1297 ASSERT3P(lwb->lwb_write_zio, !=, NULL);
1298 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_OPENED);
1299
1300 if (BP_GET_CHECKSUM(&lwb->lwb_blk) == ZIO_CHECKSUM_ZILOG2) {
1301 zilc = (zil_chain_t *)lwb->lwb_buf;
1302 bp = &zilc->zc_next_blk;
1303 } else {
1304 zilc = (zil_chain_t *)(lwb->lwb_buf + lwb->lwb_sz);
1305 bp = &zilc->zc_next_blk;
1306 }
1307
1308 ASSERT(lwb->lwb_nused <= lwb->lwb_sz);
1309
1310 /*
1311 * Allocate the next block and save its address in this block
1312 * before writing it in order to establish the log chain.
1313 * Note that if the allocation of nlwb synced before we wrote
1314 * the block that points at it (lwb), we'd leak it if we crashed.
1315 * Therefore, we don't do dmu_tx_commit() until zil_lwb_write_done().
1316 * We dirty the dataset to ensure that zil_sync() will be called
1317 * to clean up in the event of allocation failure or I/O failure.
1318 */
1319
1320 tx = dmu_tx_create(zilog->zl_os);
1321
1322 /*
1323 * Since we are not going to create any new dirty data and we can even
1324 * help with clearing the existing dirty data, we should not be subject
1325 * to the dirty data based delays.
1326 * We (ab)use TXG_WAITED to bypass the delay mechanism.
1327 * One side effect from using TXG_WAITED is that dmu_tx_assign() can
1328 * fail if the pool is suspended. Those are dramatic circumstances,
1329 * so we return NULL to signal that the normal ZIL processing is not
1330 * possible and txg_wait_synced() should be used to ensure that the data
1331 * is on disk.
1332 */
1333 error = dmu_tx_assign(tx, TXG_WAITED);
1334 if (error != 0) {
1335 ASSERT3S(error, ==, EIO);
1336 dmu_tx_abort(tx);
1337 return (NULL);
1338 }
1339 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
1340 txg = dmu_tx_get_txg(tx);
1341
1342 lwb->lwb_tx = tx;
1343
1344 /*
1345 * Log blocks are pre-allocated. Here we select the size of the next
1346 * block, based on size used in the last block.
1347 * - first find the smallest bucket that will fit the block from a
1348 * limited set of block sizes. This is because it's faster to write
1349 * blocks allocated from the same metaslab as they are adjacent or
1350 * close.
1351 * - next find the maximum from the new suggested size and an array of
1352 * previous sizes. This lessens a picket fence effect of wrongly
1353 * guesssing the size if we have a stream of say 2k, 64k, 2k, 64k
1354 * requests.
1355 *
1356 * Note we only write what is used, but we can't just allocate
1357 * the maximum block size because we can exhaust the available
1358 * pool log space.
1359 */
1360 zil_blksz = zilog->zl_cur_used + sizeof (zil_chain_t);
1361 for (i = 0; zil_blksz > zil_block_buckets[i]; i++)
1362 continue;
1363 zil_blksz = zil_block_buckets[i];
1364 if (zil_blksz == UINT64_MAX)
1365 zil_blksz = SPA_OLD_MAXBLOCKSIZE;
1366 zilog->zl_prev_blks[zilog->zl_prev_rotor] = zil_blksz;
1367 for (i = 0; i < ZIL_PREV_BLKS; i++)
1368 zil_blksz = MAX(zil_blksz, zilog->zl_prev_blks[i]);
1369 zilog->zl_prev_rotor = (zilog->zl_prev_rotor + 1) & (ZIL_PREV_BLKS - 1);
1370
1371 BP_ZERO(bp);
1372
1373 /* pass the old blkptr in order to spread log blocks across devs */
1374 error = zio_alloc_zil(spa, txg, bp, &lwb->lwb_blk, zil_blksz, &slog);
1375 if (error == 0) {
1376 ASSERT3U(bp->blk_birth, ==, txg);
1377 bp->blk_cksum = lwb->lwb_blk.blk_cksum;
1378 bp->blk_cksum.zc_word[ZIL_ZC_SEQ]++;
1379
1380 /*
1381 * Allocate a new log write block (lwb).
1382 */
1383 nlwb = zil_alloc_lwb(zilog, bp, slog, txg);
1384 }
1385
1386 if (BP_GET_CHECKSUM(&lwb->lwb_blk) == ZIO_CHECKSUM_ZILOG2) {
1387 /* For Slim ZIL only write what is used. */
1388 wsz = P2ROUNDUP_TYPED(lwb->lwb_nused, ZIL_MIN_BLKSZ, uint64_t);
1389 ASSERT3U(wsz, <=, lwb->lwb_sz);
1390 zio_shrink(lwb->lwb_write_zio, wsz);
1391
1392 } else {
1393 wsz = lwb->lwb_sz;
1394 }
1395
1396 zilc->zc_pad = 0;
1397 zilc->zc_nused = lwb->lwb_nused;
1398 zilc->zc_eck.zec_cksum = lwb->lwb_blk.blk_cksum;
1399
1400 /*
1401 * clear unused data for security
1402 */
1403 bzero(lwb->lwb_buf + lwb->lwb_nused, wsz - lwb->lwb_nused);
1404
1405 spa_config_enter(zilog->zl_spa, SCL_STATE, lwb, RW_READER);
1406
1407 zil_lwb_add_block(lwb, &lwb->lwb_blk);
1408 lwb->lwb_issued_timestamp = gethrtime();
1409 lwb->lwb_state = LWB_STATE_ISSUED;
1410
1411 zio_nowait(lwb->lwb_root_zio);
1412 zio_nowait(lwb->lwb_write_zio);
1413
1414 /*
1415 * If there was an allocation failure then nlwb will be null which
1416 * forces a txg_wait_synced().
1417 */
1418 return (nlwb);
1419 }
1420
1421 static lwb_t *
1422 zil_lwb_commit(zilog_t *zilog, itx_t *itx, lwb_t *lwb)
1423 {
1424 lr_t *lrcb, *lrc;
1425 lr_write_t *lrwb, *lrw;
1426 char *lr_buf;
1427 uint64_t dlen, dnow, lwb_sp, reclen, txg;
1428
1429 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock));
1430 ASSERT3P(lwb, !=, NULL);
1431 ASSERT3P(lwb->lwb_buf, !=, NULL);
1432
1433 zil_lwb_write_open(zilog, lwb);
1434
1435 lrc = &itx->itx_lr;
1436 lrw = (lr_write_t *)lrc;
1437
1438 /*
1439 * A commit itx doesn't represent any on-disk state; instead
1440 * it's simply used as a place holder on the commit list, and
1441 * provides a mechanism for attaching a "commit waiter" onto the
1442 * correct lwb (such that the waiter can be signalled upon
1443 * completion of that lwb). Thus, we don't process this itx's
1444 * log record if it's a commit itx (these itx's don't have log
1445 * records), and instead link the itx's waiter onto the lwb's
1446 * list of waiters.
1447 *
1448 * For more details, see the comment above zil_commit().
1449 */
1450 if (lrc->lrc_txtype == TX_COMMIT) {
1451 mutex_enter(&zilog->zl_lock);
1452 zil_commit_waiter_link_lwb(itx->itx_private, lwb);
1453 itx->itx_private = NULL;
1454 mutex_exit(&zilog->zl_lock);
1455 return (lwb);
1456 }
1457
1458 if (lrc->lrc_txtype == TX_WRITE && itx->itx_wr_state == WR_NEED_COPY) {
1459 dlen = P2ROUNDUP_TYPED(
1460 lrw->lr_length, sizeof (uint64_t), uint64_t);
1461 } else {
1462 dlen = 0;
1463 }
1464 reclen = lrc->lrc_reclen;
1465 zilog->zl_cur_used += (reclen + dlen);
1466 txg = lrc->lrc_txg;
1467
1468 ASSERT3U(zilog->zl_cur_used, <, UINT64_MAX - (reclen + dlen));
1469
1470 cont:
1471 /*
1472 * If this record won't fit in the current log block, start a new one.
1473 * For WR_NEED_COPY optimize layout for minimal number of chunks.
1474 */
1475 lwb_sp = lwb->lwb_sz - lwb->lwb_nused;
1476 if (reclen > lwb_sp || (reclen + dlen > lwb_sp &&
1477 lwb_sp < ZIL_MAX_WASTE_SPACE && (dlen % ZIL_MAX_LOG_DATA == 0 ||
1478 lwb_sp < reclen + dlen % ZIL_MAX_LOG_DATA))) {
1479 lwb = zil_lwb_write_issue(zilog, lwb);
1480 if (lwb == NULL)
1481 return (NULL);
1482 zil_lwb_write_open(zilog, lwb);
1483 ASSERT(LWB_EMPTY(lwb));
1484 lwb_sp = lwb->lwb_sz - lwb->lwb_nused;
1485 ASSERT3U(reclen + MIN(dlen, sizeof (uint64_t)), <=, lwb_sp);
1486 }
1487
1488 dnow = MIN(dlen, lwb_sp - reclen);
1489 lr_buf = lwb->lwb_buf + lwb->lwb_nused;
1490 bcopy(lrc, lr_buf, reclen);
1491 lrcb = (lr_t *)lr_buf; /* Like lrc, but inside lwb. */
1492 lrwb = (lr_write_t *)lrcb; /* Like lrw, but inside lwb. */
1493
1494 /*
1495 * If it's a write, fetch the data or get its blkptr as appropriate.
1496 */
1497 if (lrc->lrc_txtype == TX_WRITE) {
1498 if (txg > spa_freeze_txg(zilog->zl_spa))
1499 txg_wait_synced(zilog->zl_dmu_pool, txg);
1500 if (itx->itx_wr_state != WR_COPIED) {
1501 char *dbuf;
1502 int error;
1503
1504 if (itx->itx_wr_state == WR_NEED_COPY) {
1505 dbuf = lr_buf + reclen;
1506 lrcb->lrc_reclen += dnow;
1507 if (lrwb->lr_length > dnow)
1508 lrwb->lr_length = dnow;
1509 lrw->lr_offset += dnow;
1510 lrw->lr_length -= dnow;
1511 } else {
1512 ASSERT(itx->itx_wr_state == WR_INDIRECT);
1513 dbuf = NULL;
1514 }
1515
1516 /*
1517 * We pass in the "lwb_write_zio" rather than
1518 * "lwb_root_zio" so that the "lwb_write_zio"
1519 * becomes the parent of any zio's created by
1520 * the "zl_get_data" callback. The vdevs are
1521 * flushed after the "lwb_write_zio" completes,
1522 * so we want to make sure that completion
1523 * callback waits for these additional zio's,
1524 * such that the vdevs used by those zio's will
1525 * be included in the lwb's vdev tree, and those
1526 * vdevs will be properly flushed. If we passed
1527 * in "lwb_root_zio" here, then these additional
1528 * vdevs may not be flushed; e.g. if these zio's
1529 * completed after "lwb_write_zio" completed.
1530 */
1531 error = zilog->zl_get_data(itx->itx_private,
1532 lrwb, dbuf, lwb, lwb->lwb_write_zio);
1533
1534 if (error == EIO) {
1535 txg_wait_synced(zilog->zl_dmu_pool, txg);
1536 return (lwb);
1537 }
1538 if (error != 0) {
1539 ASSERT(error == ENOENT || error == EEXIST ||
1540 error == EALREADY);
1541 return (lwb);
1542 }
1543 }
1544 }
1545
1546 /*
1547 * We're actually making an entry, so update lrc_seq to be the
1548 * log record sequence number. Note that this is generally not
1549 * equal to the itx sequence number because not all transactions
1550 * are synchronous, and sometimes spa_sync() gets there first.
1551 */
1552 lrcb->lrc_seq = ++zilog->zl_lr_seq;
1553 lwb->lwb_nused += reclen + dnow;
1554
1555 zil_lwb_add_txg(lwb, txg);
1556
1557 ASSERT3U(lwb->lwb_nused, <=, lwb->lwb_sz);
1558 ASSERT0(P2PHASE(lwb->lwb_nused, sizeof (uint64_t)));
1559
1560 dlen -= dnow;
1561 if (dlen > 0) {
1562 zilog->zl_cur_used += reclen;
1563 goto cont;
1564 }
1565
1566 return (lwb);
1567 }
1568
1569 itx_t *
1570 zil_itx_create(uint64_t txtype, size_t lrsize)
1571 {
1572 itx_t *itx;
1573
1574 lrsize = P2ROUNDUP_TYPED(lrsize, sizeof (uint64_t), size_t);
1575
1576 itx = kmem_alloc(offsetof(itx_t, itx_lr) + lrsize, KM_SLEEP);
1577 itx->itx_lr.lrc_txtype = txtype;
1578 itx->itx_lr.lrc_reclen = lrsize;
1579 itx->itx_lr.lrc_seq = 0; /* defensive */
1580 itx->itx_sync = B_TRUE; /* default is synchronous */
1581
1582 return (itx);
1583 }
1584
1585 void
1586 zil_itx_destroy(itx_t *itx)
1587 {
1588 kmem_free(itx, offsetof(itx_t, itx_lr) + itx->itx_lr.lrc_reclen);
1589 }
1590
1591 /*
1592 * Free up the sync and async itxs. The itxs_t has already been detached
1593 * so no locks are needed.
1594 */
1595 static void
1596 zil_itxg_clean(itxs_t *itxs)
1597 {
1598 itx_t *itx;
1599 list_t *list;
1600 avl_tree_t *t;
1601 void *cookie;
1602 itx_async_node_t *ian;
1603
1604 list = &itxs->i_sync_list;
1605 while ((itx = list_head(list)) != NULL) {
1606 /*
1607 * In the general case, commit itxs will not be found
1608 * here, as they'll be committed to an lwb via
1609 * zil_lwb_commit(), and free'd in that function. Having
1610 * said that, it is still possible for commit itxs to be
1611 * found here, due to the following race:
1612 *
1613 * - a thread calls zil_commit() which assigns the
1614 * commit itx to a per-txg i_sync_list
1615 * - zil_itxg_clean() is called (e.g. via spa_sync())
1616 * while the waiter is still on the i_sync_list
1617 *
1618 * There's nothing to prevent syncing the txg while the
1619 * waiter is on the i_sync_list. This normally doesn't
1620 * happen because spa_sync() is slower than zil_commit(),
1621 * but if zil_commit() calls txg_wait_synced() (e.g.
1622 * because zil_create() or zil_commit_writer_stall() is
1623 * called) we will hit this case.
1624 */
1625 if (itx->itx_lr.lrc_txtype == TX_COMMIT)
1626 zil_commit_waiter_skip(itx->itx_private);
1627
1628 list_remove(list, itx);
1629 zil_itx_destroy(itx);
1630 }
1631
1632 cookie = NULL;
1633 t = &itxs->i_async_tree;
1634 while ((ian = avl_destroy_nodes(t, &cookie)) != NULL) {
1635 list = &ian->ia_list;
1636 while ((itx = list_head(list)) != NULL) {
1637 list_remove(list, itx);
1638 /* commit itxs should never be on the async lists. */
1639 ASSERT3U(itx->itx_lr.lrc_txtype, !=, TX_COMMIT);
1640 zil_itx_destroy(itx);
1641 }
1642 list_destroy(list);
1643 kmem_free(ian, sizeof (itx_async_node_t));
1644 }
1645 avl_destroy(t);
1646
1647 kmem_free(itxs, sizeof (itxs_t));
1648 }
1649
1650 static int
1651 zil_aitx_compare(const void *x1, const void *x2)
1652 {
1653 const uint64_t o1 = ((itx_async_node_t *)x1)->ia_foid;
1654 const uint64_t o2 = ((itx_async_node_t *)x2)->ia_foid;
1655
1656 if (o1 < o2)
1657 return (-1);
1658 if (o1 > o2)
1659 return (1);
1660
1661 return (0);
1662 }
1663
1664 /*
1665 * Remove all async itx with the given oid.
1666 */
1667 static void
1668 zil_remove_async(zilog_t *zilog, uint64_t oid)
1669 {
1670 uint64_t otxg, txg;
1671 itx_async_node_t *ian;
1672 avl_tree_t *t;
1673 avl_index_t where;
1674 list_t clean_list;
1675 itx_t *itx;
1676
1677 ASSERT(oid != 0);
1678 list_create(&clean_list, sizeof (itx_t), offsetof(itx_t, itx_node));
1679
1680 if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) /* ziltest support */
1681 otxg = ZILTEST_TXG;
1682 else
1683 otxg = spa_last_synced_txg(zilog->zl_spa) + 1;
1684
1685 for (txg = otxg; txg < (otxg + TXG_CONCURRENT_STATES); txg++) {
1686 itxg_t *itxg = &zilog->zl_itxg[txg & TXG_MASK];
1687
1688 mutex_enter(&itxg->itxg_lock);
1689 if (itxg->itxg_txg != txg) {
1690 mutex_exit(&itxg->itxg_lock);
1691 continue;
1692 }
1693
1694 /*
1695 * Locate the object node and append its list.
1696 */
1697 t = &itxg->itxg_itxs->i_async_tree;
1698 ian = avl_find(t, &oid, &where);
1699 if (ian != NULL)
1700 list_move_tail(&clean_list, &ian->ia_list);
1701 mutex_exit(&itxg->itxg_lock);
1702 }
1703 while ((itx = list_head(&clean_list)) != NULL) {
1704 list_remove(&clean_list, itx);
1705 /* commit itxs should never be on the async lists. */
1706 ASSERT3U(itx->itx_lr.lrc_txtype, !=, TX_COMMIT);
1707 zil_itx_destroy(itx);
1708 }
1709 list_destroy(&clean_list);
1710 }
1711
1712 void
1713 zil_itx_assign(zilog_t *zilog, itx_t *itx, dmu_tx_t *tx)
1714 {
1715 uint64_t txg;
1716 itxg_t *itxg;
1717 itxs_t *itxs, *clean = NULL;
1718
1719 /*
1720 * Object ids can be re-instantiated in the next txg so
1721 * remove any async transactions to avoid future leaks.
1722 * This can happen if a fsync occurs on the re-instantiated
1723 * object for a WR_INDIRECT or WR_NEED_COPY write, which gets
1724 * the new file data and flushes a write record for the old object.
1725 */
1726 if ((itx->itx_lr.lrc_txtype & ~TX_CI) == TX_REMOVE)
1727 zil_remove_async(zilog, itx->itx_oid);
1728
1729 /*
1730 * Ensure the data of a renamed file is committed before the rename.
1731 */
1732 if ((itx->itx_lr.lrc_txtype & ~TX_CI) == TX_RENAME)
1733 zil_async_to_sync(zilog, itx->itx_oid);
1734
1735 if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX)
1736 txg = ZILTEST_TXG;
1737 else
1738 txg = dmu_tx_get_txg(tx);
1739
1740 itxg = &zilog->zl_itxg[txg & TXG_MASK];
1741 mutex_enter(&itxg->itxg_lock);
1742 itxs = itxg->itxg_itxs;
1743 if (itxg->itxg_txg != txg) {
1744 if (itxs != NULL) {
1745 /*
1746 * The zil_clean callback hasn't got around to cleaning
1747 * this itxg. Save the itxs for release below.
1748 * This should be rare.
1749 */
1750 zfs_dbgmsg("zil_itx_assign: missed itx cleanup for "
1751 "txg %llu", itxg->itxg_txg);
1752 clean = itxg->itxg_itxs;
1753 }
1754 itxg->itxg_txg = txg;
1755 itxs = itxg->itxg_itxs = kmem_zalloc(sizeof (itxs_t), KM_SLEEP);
1756
1757 list_create(&itxs->i_sync_list, sizeof (itx_t),
1758 offsetof(itx_t, itx_node));
1759 avl_create(&itxs->i_async_tree, zil_aitx_compare,
1760 sizeof (itx_async_node_t),
1761 offsetof(itx_async_node_t, ia_node));
1762 }
1763 if (itx->itx_sync) {
1764 list_insert_tail(&itxs->i_sync_list, itx);
1765 } else {
1766 avl_tree_t *t = &itxs->i_async_tree;
1767 uint64_t foid = ((lr_ooo_t *)&itx->itx_lr)->lr_foid;
1768 itx_async_node_t *ian;
1769 avl_index_t where;
1770
1771 ian = avl_find(t, &foid, &where);
1772 if (ian == NULL) {
1773 ian = kmem_alloc(sizeof (itx_async_node_t), KM_SLEEP);
1774 list_create(&ian->ia_list, sizeof (itx_t),
1775 offsetof(itx_t, itx_node));
1776 ian->ia_foid = foid;
1777 avl_insert(t, ian, where);
1778 }
1779 list_insert_tail(&ian->ia_list, itx);
1780 }
1781
1782 itx->itx_lr.lrc_txg = dmu_tx_get_txg(tx);
1783
1784 /*
1785 * We don't want to dirty the ZIL using ZILTEST_TXG, because
1786 * zil_clean() will never be called using ZILTEST_TXG. Thus, we
1787 * need to be careful to always dirty the ZIL using the "real"
1788 * TXG (not itxg_txg) even when the SPA is frozen.
1789 */
1790 zilog_dirty(zilog, dmu_tx_get_txg(tx));
1791 mutex_exit(&itxg->itxg_lock);
1792
1793 /* Release the old itxs now we've dropped the lock */
1794 if (clean != NULL)
1795 zil_itxg_clean(clean);
1796 }
1797
1798 /*
1799 * If there are any in-memory intent log transactions which have now been
1800 * synced then start up a taskq to free them. We should only do this after we
1801 * have written out the uberblocks (i.e. txg has been comitted) so that
1802 * don't inadvertently clean out in-memory log records that would be required
1803 * by zil_commit().
1804 */
1805 void
1806 zil_clean(zilog_t *zilog, uint64_t synced_txg)
1807 {
1808 itxg_t *itxg = &zilog->zl_itxg[synced_txg & TXG_MASK];
1809 itxs_t *clean_me;
1810
1811 ASSERT3U(synced_txg, <, ZILTEST_TXG);
1812
1813 mutex_enter(&itxg->itxg_lock);
1814 if (itxg->itxg_itxs == NULL || itxg->itxg_txg == ZILTEST_TXG) {
1815 mutex_exit(&itxg->itxg_lock);
1816 return;
1817 }
1818 ASSERT3U(itxg->itxg_txg, <=, synced_txg);
1819 ASSERT3U(itxg->itxg_txg, !=, 0);
1820 clean_me = itxg->itxg_itxs;
1821 itxg->itxg_itxs = NULL;
1822 itxg->itxg_txg = 0;
1823 mutex_exit(&itxg->itxg_lock);
1824 /*
1825 * Preferably start a task queue to free up the old itxs but
1826 * if taskq_dispatch can't allocate resources to do that then
1827 * free it in-line. This should be rare. Note, using TQ_SLEEP
1828 * created a bad performance problem.
1829 */
1830 ASSERT3P(zilog->zl_dmu_pool, !=, NULL);
1831 ASSERT3P(zilog->zl_dmu_pool->dp_zil_clean_taskq, !=, NULL);
1832 if (taskq_dispatch(zilog->zl_dmu_pool->dp_zil_clean_taskq,
1833 (void (*)(void *))zil_itxg_clean, clean_me, TQ_NOSLEEP) == NULL)
1834 zil_itxg_clean(clean_me);
1835 }
1836
1837 /*
1838 * This function will traverse the queue of itxs that need to be
1839 * committed, and move them onto the ZIL's zl_itx_commit_list.
1840 */
1841 static void
1842 zil_get_commit_list(zilog_t *zilog)
1843 {
1844 uint64_t otxg, txg;
1845 list_t *commit_list = &zilog->zl_itx_commit_list;
1846
1847 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock));
1848
1849 if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) /* ziltest support */
1850 otxg = ZILTEST_TXG;
1851 else
1852 otxg = spa_last_synced_txg(zilog->zl_spa) + 1;
1853
1854 /*
1855 * This is inherently racy, since there is nothing to prevent
1856 * the last synced txg from changing. That's okay since we'll
1857 * only commit things in the future.
1858 */
1859 for (txg = otxg; txg < (otxg + TXG_CONCURRENT_STATES); txg++) {
1860 itxg_t *itxg = &zilog->zl_itxg[txg & TXG_MASK];
1861
1862 mutex_enter(&itxg->itxg_lock);
1863 if (itxg->itxg_txg != txg) {
1864 mutex_exit(&itxg->itxg_lock);
1865 continue;
1866 }
1867
1868 /*
1869 * If we're adding itx records to the zl_itx_commit_list,
1870 * then the zil better be dirty in this "txg". We can assert
1871 * that here since we're holding the itxg_lock which will
1872 * prevent spa_sync from cleaning it. Once we add the itxs
1873 * to the zl_itx_commit_list we must commit it to disk even
1874 * if it's unnecessary (i.e. the txg was synced).
1875 */
1876 ASSERT(zilog_is_dirty_in_txg(zilog, txg) ||
1877 spa_freeze_txg(zilog->zl_spa) != UINT64_MAX);
1878 list_move_tail(commit_list, &itxg->itxg_itxs->i_sync_list);
1879
1880 mutex_exit(&itxg->itxg_lock);
1881 }
1882 }
1883
1884 /*
1885 * Move the async itxs for a specified object to commit into sync lists.
1886 */
1887 static void
1888 zil_async_to_sync(zilog_t *zilog, uint64_t foid)
1889 {
1890 uint64_t otxg, txg;
1891 itx_async_node_t *ian;
1892 avl_tree_t *t;
1893 avl_index_t where;
1894
1895 if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) /* ziltest support */
1896 otxg = ZILTEST_TXG;
1897 else
1898 otxg = spa_last_synced_txg(zilog->zl_spa) + 1;
1899
1900 /*
1901 * This is inherently racy, since there is nothing to prevent
1902 * the last synced txg from changing.
1903 */
1904 for (txg = otxg; txg < (otxg + TXG_CONCURRENT_STATES); txg++) {
1905 itxg_t *itxg = &zilog->zl_itxg[txg & TXG_MASK];
1906
1907 mutex_enter(&itxg->itxg_lock);
1908 if (itxg->itxg_txg != txg) {
1909 mutex_exit(&itxg->itxg_lock);
1910 continue;
1911 }
1912
1913 /*
1914 * If a foid is specified then find that node and append its
1915 * list. Otherwise walk the tree appending all the lists
1916 * to the sync list. We add to the end rather than the
1917 * beginning to ensure the create has happened.
1918 */
1919 t = &itxg->itxg_itxs->i_async_tree;
1920 if (foid != 0) {
1921 ian = avl_find(t, &foid, &where);
1922 if (ian != NULL) {
1923 list_move_tail(&itxg->itxg_itxs->i_sync_list,
1924 &ian->ia_list);
1925 }
1926 } else {
1927 void *cookie = NULL;
1928
1929 while ((ian = avl_destroy_nodes(t, &cookie)) != NULL) {
1930 list_move_tail(&itxg->itxg_itxs->i_sync_list,
1931 &ian->ia_list);
1932 list_destroy(&ian->ia_list);
1933 kmem_free(ian, sizeof (itx_async_node_t));
1934 }
1935 }
1936 mutex_exit(&itxg->itxg_lock);
1937 }
1938 }
1939
1940 /*
1941 * This function will prune commit itxs that are at the head of the
1942 * commit list (it won't prune past the first non-commit itx), and
1943 * either: a) attach them to the last lwb that's still pending
1944 * completion, or b) skip them altogether.
1945 *
1946 * This is used as a performance optimization to prevent commit itxs
1947 * from generating new lwbs when it's unnecessary to do so.
1948 */
1949 static void
1950 zil_prune_commit_list(zilog_t *zilog)
1951 {
1952 itx_t *itx;
1953
1954 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock));
1955
1956 while (itx = list_head(&zilog->zl_itx_commit_list)) {
1957 lr_t *lrc = &itx->itx_lr;
1958 if (lrc->lrc_txtype != TX_COMMIT)
1959 break;
1960
1961 mutex_enter(&zilog->zl_lock);
1962
1963 lwb_t *last_lwb = zilog->zl_last_lwb_opened;
1964 if (last_lwb == NULL ||
1965 last_lwb->lwb_state == LWB_STATE_FLUSH_DONE) {
1966 /*
1967 * All of the itxs this waiter was waiting on
1968 * must have already completed (or there were
1969 * never any itx's for it to wait on), so it's
1970 * safe to skip this waiter and mark it done.
1971 */
1972 zil_commit_waiter_skip(itx->itx_private);
1973 } else {
1974 zil_commit_waiter_link_lwb(itx->itx_private, last_lwb);
1975 itx->itx_private = NULL;
1976 }
1977
1978 mutex_exit(&zilog->zl_lock);
1979
1980 list_remove(&zilog->zl_itx_commit_list, itx);
1981 zil_itx_destroy(itx);
1982 }
1983
1984 IMPLY(itx != NULL, itx->itx_lr.lrc_txtype != TX_COMMIT);
1985 }
1986
1987 static void
1988 zil_commit_writer_stall(zilog_t *zilog)
1989 {
1990 /*
1991 * When zio_alloc_zil() fails to allocate the next lwb block on
1992 * disk, we must call txg_wait_synced() to ensure all of the
1993 * lwbs in the zilog's zl_lwb_list are synced and then freed (in
1994 * zil_sync()), such that any subsequent ZIL writer (i.e. a call
1995 * to zil_process_commit_list()) will have to call zil_create(),
1996 * and start a new ZIL chain.
1997 *
1998 * Since zil_alloc_zil() failed, the lwb that was previously
1999 * issued does not have a pointer to the "next" lwb on disk.
2000 * Thus, if another ZIL writer thread was to allocate the "next"
2001 * on-disk lwb, that block could be leaked in the event of a
2002 * crash (because the previous lwb on-disk would not point to
2003 * it).
2004 *
2005 * We must hold the zilog's zl_issuer_lock while we do this, to
2006 * ensure no new threads enter zil_process_commit_list() until
2007 * all lwb's in the zl_lwb_list have been synced and freed
2008 * (which is achieved via the txg_wait_synced() call).
2009 */
2010 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock));
2011 txg_wait_synced(zilog->zl_dmu_pool, 0);
2012 ASSERT3P(list_tail(&zilog->zl_lwb_list), ==, NULL);
2013 }
2014
2015 /*
2016 * This function will traverse the commit list, creating new lwbs as
2017 * needed, and committing the itxs from the commit list to these newly
2018 * created lwbs. Additionally, as a new lwb is created, the previous
2019 * lwb will be issued to the zio layer to be written to disk.
2020 */
2021 static void
2022 zil_process_commit_list(zilog_t *zilog)
2023 {
2024 spa_t *spa = zilog->zl_spa;
2025 list_t nolwb_waiters;
2026 lwb_t *lwb;
2027 itx_t *itx;
2028
2029 ASSERT(MUTEX_HELD(&zilog->zl_issuer_lock));
2030
2031 /*
2032 * Return if there's nothing to commit before we dirty the fs by
2033 * calling zil_create().
2034 */
2035 if (list_head(&zilog->zl_itx_commit_list) == NULL)
2036 return;
2037
2038 list_create(&nolwb_waiters, sizeof (zil_commit_waiter_t),
2039 offsetof(zil_commit_waiter_t, zcw_node));
2040
2041 lwb = list_tail(&zilog->zl_lwb_list);
2042 if (lwb == NULL) {
2043 lwb = zil_create(zilog);
2044 } else {
2045 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_ISSUED);
2046 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_WRITE_DONE);
2047 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_FLUSH_DONE);
2048 }
2049
2050 while (itx = list_head(&zilog->zl_itx_commit_list)) {
2051 lr_t *lrc = &itx->itx_lr;
2052 uint64_t txg = lrc->lrc_txg;
2053
2054 ASSERT3U(txg, !=, 0);
2055
2056 if (lrc->lrc_txtype == TX_COMMIT) {
2057 DTRACE_PROBE2(zil__process__commit__itx,
2058 zilog_t *, zilog, itx_t *, itx);
2059 } else {
2060 DTRACE_PROBE2(zil__process__normal__itx,
2061 zilog_t *, zilog, itx_t *, itx);
2062 }
2063
2064 boolean_t synced = txg <= spa_last_synced_txg(spa);
2065 boolean_t frozen = txg > spa_freeze_txg(spa);
2066
2067 /*
2068 * If the txg of this itx has already been synced out, then
2069 * we don't need to commit this itx to an lwb. This is
2070 * because the data of this itx will have already been
2071 * written to the main pool. This is inherently racy, and
2072 * it's still ok to commit an itx whose txg has already
2073 * been synced; this will result in a write that's
2074 * unnecessary, but will do no harm.
2075 *
2076 * With that said, we always want to commit TX_COMMIT itxs
2077 * to an lwb, regardless of whether or not that itx's txg
2078 * has been synced out. We do this to ensure any OPENED lwb
2079 * will always have at least one zil_commit_waiter_t linked
2080 * to the lwb.
2081 *
2082 * As a counter-example, if we skipped TX_COMMIT itx's
2083 * whose txg had already been synced, the following
2084 * situation could occur if we happened to be racing with
2085 * spa_sync:
2086 *
2087 * 1. we commit a non-TX_COMMIT itx to an lwb, where the
2088 * itx's txg is 10 and the last synced txg is 9.
2089 * 2. spa_sync finishes syncing out txg 10.
2090 * 3. we move to the next itx in the list, it's a TX_COMMIT
2091 * whose txg is 10, so we skip it rather than committing
2092 * it to the lwb used in (1).
2093 *
2094 * If the itx that is skipped in (3) is the last TX_COMMIT
2095 * itx in the commit list, than it's possible for the lwb
2096 * used in (1) to remain in the OPENED state indefinitely.
2097 *
2098 * To prevent the above scenario from occuring, ensuring
2099 * that once an lwb is OPENED it will transition to ISSUED
2100 * and eventually DONE, we always commit TX_COMMIT itx's to
2101 * an lwb here, even if that itx's txg has already been
2102 * synced.
2103 *
2104 * Finally, if the pool is frozen, we _always_ commit the
2105 * itx. The point of freezing the pool is to prevent data
2106 * from being written to the main pool via spa_sync, and
2107 * instead rely solely on the ZIL to persistently store the
2108 * data; i.e. when the pool is frozen, the last synced txg
2109 * value can't be trusted.
2110 */
2111 if (frozen || !synced || lrc->lrc_txtype == TX_COMMIT) {
2112 if (lwb != NULL) {
2113 lwb = zil_lwb_commit(zilog, itx, lwb);
2114 } else if (lrc->lrc_txtype == TX_COMMIT) {
2115 ASSERT3P(lwb, ==, NULL);
2116 zil_commit_waiter_link_nolwb(
2117 itx->itx_private, &nolwb_waiters);
2118 }
2119 }
2120
2121 list_remove(&zilog->zl_itx_commit_list, itx);
2122 zil_itx_destroy(itx);
2123 }
2124
2125 if (lwb == NULL) {
2126 /*
2127 * This indicates zio_alloc_zil() failed to allocate the
2128 * "next" lwb on-disk. When this happens, we must stall
2129 * the ZIL write pipeline; see the comment within
2130 * zil_commit_writer_stall() for more details.
2131 */
2132 zil_commit_writer_stall(zilog);
2133
2134 /*
2135 * Additionally, we have to signal and mark the "nolwb"
2136 * waiters as "done" here, since without an lwb, we
2137 * can't do this via zil_lwb_flush_vdevs_done() like
2138 * normal.
2139 */
2140 zil_commit_waiter_t *zcw;
2141 while (zcw = list_head(&nolwb_waiters)) {
2142 zil_commit_waiter_skip(zcw);
2143 list_remove(&nolwb_waiters, zcw);
2144 }
2145 } else {
2146 ASSERT(list_is_empty(&nolwb_waiters));
2147 ASSERT3P(lwb, !=, NULL);
2148 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_ISSUED);
2149 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_WRITE_DONE);
2150 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_FLUSH_DONE);
2151
2152 /*
2153 * At this point, the ZIL block pointed at by the "lwb"
2154 * variable is in one of the following states: "closed"
2155 * or "open".
2156 *
2157 * If its "closed", then no itxs have been committed to
2158 * it, so there's no point in issuing its zio (i.e.
2159 * it's "empty").
2160 *
2161 * If its "open" state, then it contains one or more
2162 * itxs that eventually need to be committed to stable
2163 * storage. In this case we intentionally do not issue
2164 * the lwb's zio to disk yet, and instead rely on one of
2165 * the following two mechanisms for issuing the zio:
2166 *
2167 * 1. Ideally, there will be more ZIL activity occuring
2168 * on the system, such that this function will be
2169 * immediately called again (not necessarily by the same
2170 * thread) and this lwb's zio will be issued via
2171 * zil_lwb_commit(). This way, the lwb is guaranteed to
2172 * be "full" when it is issued to disk, and we'll make
2173 * use of the lwb's size the best we can.
2174 *
2175 * 2. If there isn't sufficient ZIL activity occuring on
2176 * the system, such that this lwb's zio isn't issued via
2177 * zil_lwb_commit(), zil_commit_waiter() will issue the
2178 * lwb's zio. If this occurs, the lwb is not guaranteed
2179 * to be "full" by the time its zio is issued, and means
2180 * the size of the lwb was "too large" given the amount
2181 * of ZIL activity occuring on the system at that time.
2182 *
2183 * We do this for a couple of reasons:
2184 *
2185 * 1. To try and reduce the number of IOPs needed to
2186 * write the same number of itxs. If an lwb has space
2187 * available in it's buffer for more itxs, and more itxs
2188 * will be committed relatively soon (relative to the
2189 * latency of performing a write), then it's beneficial
2190 * to wait for these "next" itxs. This way, more itxs
2191 * can be committed to stable storage with fewer writes.
2192 *
2193 * 2. To try and use the largest lwb block size that the
2194 * incoming rate of itxs can support. Again, this is to
2195 * try and pack as many itxs into as few lwbs as
2196 * possible, without significantly impacting the latency
2197 * of each individual itx.
2198 */
2199 }
2200 }
2201
2202 /*
2203 * This function is responsible for ensuring the passed in commit waiter
2204 * (and associated commit itx) is committed to an lwb. If the waiter is
2205 * not already committed to an lwb, all itxs in the zilog's queue of
2206 * itxs will be processed. The assumption is the passed in waiter's
2207 * commit itx will found in the queue just like the other non-commit
2208 * itxs, such that when the entire queue is processed, the waiter will
2209 * have been commited to an lwb.
2210 *
2211 * The lwb associated with the passed in waiter is not guaranteed to
2212 * have been issued by the time this function completes. If the lwb is
2213 * not issued, we rely on future calls to zil_commit_writer() to issue
2214 * the lwb, or the timeout mechanism found in zil_commit_waiter().
2215 */
2216 static void
2217 zil_commit_writer(zilog_t *zilog, zil_commit_waiter_t *zcw)
2218 {
2219 ASSERT(!MUTEX_HELD(&zilog->zl_lock));
2220 ASSERT(spa_writeable(zilog->zl_spa));
2221
2222 mutex_enter(&zilog->zl_issuer_lock);
2223
2224 if (zcw->zcw_lwb != NULL || zcw->zcw_done) {
2225 /*
2226 * It's possible that, while we were waiting to acquire
2227 * the "zl_issuer_lock", another thread committed this
2228 * waiter to an lwb. If that occurs, we bail out early,
2229 * without processing any of the zilog's queue of itxs.
2230 *
2231 * On certain workloads and system configurations, the
2232 * "zl_issuer_lock" can become highly contended. In an
2233 * attempt to reduce this contention, we immediately drop
2234 * the lock if the waiter has already been processed.
2235 *
2236 * We've measured this optimization to reduce CPU spent
2237 * contending on this lock by up to 5%, using a system
2238 * with 32 CPUs, low latency storage (~50 usec writes),
2239 * and 1024 threads performing sync writes.
2240 */
2241 goto out;
2242 }
2243
2244 zil_get_commit_list(zilog);
2245 zil_prune_commit_list(zilog);
2246 zil_process_commit_list(zilog);
2247
2248 out:
2249 mutex_exit(&zilog->zl_issuer_lock);
2250 }
2251
2252 static void
2253 zil_commit_waiter_timeout(zilog_t *zilog, zil_commit_waiter_t *zcw)
2254 {
2255 ASSERT(!MUTEX_HELD(&zilog->zl_issuer_lock));
2256 ASSERT(MUTEX_HELD(&zcw->zcw_lock));
2257 ASSERT3B(zcw->zcw_done, ==, B_FALSE);
2258
2259 lwb_t *lwb = zcw->zcw_lwb;
2260 ASSERT3P(lwb, !=, NULL);
2261 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_CLOSED);
2262
2263 /*
2264 * If the lwb has already been issued by another thread, we can
2265 * immediately return since there's no work to be done (the
2266 * point of this function is to issue the lwb). Additionally, we
2267 * do this prior to acquiring the zl_issuer_lock, to avoid
2268 * acquiring it when it's not necessary to do so.
2269 */
2270 if (lwb->lwb_state == LWB_STATE_ISSUED ||
2271 lwb->lwb_state == LWB_STATE_WRITE_DONE ||
2272 lwb->lwb_state == LWB_STATE_FLUSH_DONE)
2273 return;
2274
2275 /*
2276 * In order to call zil_lwb_write_issue() we must hold the
2277 * zilog's "zl_issuer_lock". We can't simply acquire that lock,
2278 * since we're already holding the commit waiter's "zcw_lock",
2279 * and those two locks are aquired in the opposite order
2280 * elsewhere.
2281 */
2282 mutex_exit(&zcw->zcw_lock);
2283 mutex_enter(&zilog->zl_issuer_lock);
2284 mutex_enter(&zcw->zcw_lock);
2285
2286 /*
2287 * Since we just dropped and re-acquired the commit waiter's
2288 * lock, we have to re-check to see if the waiter was marked
2289 * "done" during that process. If the waiter was marked "done",
2290 * the "lwb" pointer is no longer valid (it can be free'd after
2291 * the waiter is marked "done"), so without this check we could
2292 * wind up with a use-after-free error below.
2293 */
2294 if (zcw->zcw_done)
2295 goto out;
2296
2297 ASSERT3P(lwb, ==, zcw->zcw_lwb);
2298
2299 /*
2300 * We've already checked this above, but since we hadn't acquired
2301 * the zilog's zl_issuer_lock, we have to perform this check a
2302 * second time while holding the lock.
2303 *
2304 * We don't need to hold the zl_lock since the lwb cannot transition
2305 * from OPENED to ISSUED while we hold the zl_issuer_lock. The lwb
2306 * _can_ transition from ISSUED to DONE, but it's OK to race with
2307 * that transition since we treat the lwb the same, whether it's in
2308 * the ISSUED or DONE states.
2309 *
2310 * The important thing, is we treat the lwb differently depending on
2311 * if it's ISSUED or OPENED, and block any other threads that might
2312 * attempt to issue this lwb. For that reason we hold the
2313 * zl_issuer_lock when checking the lwb_state; we must not call
2314 * zil_lwb_write_issue() if the lwb had already been issued.
2315 *
2316 * See the comment above the lwb_state_t structure definition for
2317 * more details on the lwb states, and locking requirements.
2318 */
2319 if (lwb->lwb_state == LWB_STATE_ISSUED ||
2320 lwb->lwb_state == LWB_STATE_WRITE_DONE ||
2321 lwb->lwb_state == LWB_STATE_FLUSH_DONE)
2322 goto out;
2323
2324 ASSERT3S(lwb->lwb_state, ==, LWB_STATE_OPENED);
2325
2326 /*
2327 * As described in the comments above zil_commit_waiter() and
2328 * zil_process_commit_list(), we need to issue this lwb's zio
2329 * since we've reached the commit waiter's timeout and it still
2330 * hasn't been issued.
2331 */
2332 lwb_t *nlwb = zil_lwb_write_issue(zilog, lwb);
2333
2334 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_OPENED);
2335
2336 /*
2337 * Since the lwb's zio hadn't been issued by the time this thread
2338 * reached its timeout, we reset the zilog's "zl_cur_used" field
2339 * to influence the zil block size selection algorithm.
2340 *
2341 * By having to issue the lwb's zio here, it means the size of the
2342 * lwb was too large, given the incoming throughput of itxs. By
2343 * setting "zl_cur_used" to zero, we communicate this fact to the
2344 * block size selection algorithm, so it can take this informaiton
2345 * into account, and potentially select a smaller size for the
2346 * next lwb block that is allocated.
2347 */
2348 zilog->zl_cur_used = 0;
2349
2350 if (nlwb == NULL) {
2351 /*
2352 * When zil_lwb_write_issue() returns NULL, this
2353 * indicates zio_alloc_zil() failed to allocate the
2354 * "next" lwb on-disk. When this occurs, the ZIL write
2355 * pipeline must be stalled; see the comment within the
2356 * zil_commit_writer_stall() function for more details.
2357 *
2358 * We must drop the commit waiter's lock prior to
2359 * calling zil_commit_writer_stall() or else we can wind
2360 * up with the following deadlock:
2361 *
2362 * - This thread is waiting for the txg to sync while
2363 * holding the waiter's lock; txg_wait_synced() is
2364 * used within txg_commit_writer_stall().
2365 *
2366 * - The txg can't sync because it is waiting for this
2367 * lwb's zio callback to call dmu_tx_commit().
2368 *
2369 * - The lwb's zio callback can't call dmu_tx_commit()
2370 * because it's blocked trying to acquire the waiter's
2371 * lock, which occurs prior to calling dmu_tx_commit()
2372 */
2373 mutex_exit(&zcw->zcw_lock);
2374 zil_commit_writer_stall(zilog);
2375 mutex_enter(&zcw->zcw_lock);
2376 }
2377
2378 out:
2379 mutex_exit(&zilog->zl_issuer_lock);
2380 ASSERT(MUTEX_HELD(&zcw->zcw_lock));
2381 }
2382
2383 /*
2384 * This function is responsible for performing the following two tasks:
2385 *
2386 * 1. its primary responsibility is to block until the given "commit
2387 * waiter" is considered "done".
2388 *
2389 * 2. its secondary responsibility is to issue the zio for the lwb that
2390 * the given "commit waiter" is waiting on, if this function has
2391 * waited "long enough" and the lwb is still in the "open" state.
2392 *
2393 * Given a sufficient amount of itxs being generated and written using
2394 * the ZIL, the lwb's zio will be issued via the zil_lwb_commit()
2395 * function. If this does not occur, this secondary responsibility will
2396 * ensure the lwb is issued even if there is not other synchronous
2397 * activity on the system.
2398 *
2399 * For more details, see zil_process_commit_list(); more specifically,
2400 * the comment at the bottom of that function.
2401 */
2402 static void
2403 zil_commit_waiter(zilog_t *zilog, zil_commit_waiter_t *zcw)
2404 {
2405 ASSERT(!MUTEX_HELD(&zilog->zl_lock));
2406 ASSERT(!MUTEX_HELD(&zilog->zl_issuer_lock));
2407 ASSERT(spa_writeable(zilog->zl_spa));
2408
2409 mutex_enter(&zcw->zcw_lock);
2410
2411 /*
2412 * The timeout is scaled based on the lwb latency to avoid
2413 * significantly impacting the latency of each individual itx.
2414 * For more details, see the comment at the bottom of the
2415 * zil_process_commit_list() function.
2416 */
2417 int pct = MAX(zfs_commit_timeout_pct, 1);
2418 hrtime_t sleep = (zilog->zl_last_lwb_latency * pct) / 100;
2419 hrtime_t wakeup = gethrtime() + sleep;
2420 boolean_t timedout = B_FALSE;
2421
2422 while (!zcw->zcw_done) {
2423 ASSERT(MUTEX_HELD(&zcw->zcw_lock));
2424
2425 lwb_t *lwb = zcw->zcw_lwb;
2426
2427 /*
2428 * Usually, the waiter will have a non-NULL lwb field here,
2429 * but it's possible for it to be NULL as a result of
2430 * zil_commit() racing with spa_sync().
2431 *
2432 * When zil_clean() is called, it's possible for the itxg
2433 * list (which may be cleaned via a taskq) to contain
2434 * commit itxs. When this occurs, the commit waiters linked
2435 * off of these commit itxs will not be committed to an
2436 * lwb. Additionally, these commit waiters will not be
2437 * marked done until zil_commit_waiter_skip() is called via
2438 * zil_itxg_clean().
2439 *
2440 * Thus, it's possible for this commit waiter (i.e. the
2441 * "zcw" variable) to be found in this "in between" state;
2442 * where it's "zcw_lwb" field is NULL, and it hasn't yet
2443 * been skipped, so it's "zcw_done" field is still B_FALSE.
2444 */
2445 IMPLY(lwb != NULL, lwb->lwb_state != LWB_STATE_CLOSED);
2446
2447 if (lwb != NULL && lwb->lwb_state == LWB_STATE_OPENED) {
2448 ASSERT3B(timedout, ==, B_FALSE);
2449
2450 /*
2451 * If the lwb hasn't been issued yet, then we
2452 * need to wait with a timeout, in case this
2453 * function needs to issue the lwb after the
2454 * timeout is reached; responsibility (2) from
2455 * the comment above this function.
2456 */
2457 clock_t timeleft = cv_timedwait_hires(&zcw->zcw_cv,
2458 &zcw->zcw_lock, wakeup, USEC2NSEC(1),
2459 CALLOUT_FLAG_ABSOLUTE);
2460
2461 if (timeleft >= 0 || zcw->zcw_done)
2462 continue;
2463
2464 timedout = B_TRUE;
2465 zil_commit_waiter_timeout(zilog, zcw);
2466
2467 if (!zcw->zcw_done) {
2468 /*
2469 * If the commit waiter has already been
2470 * marked "done", it's possible for the
2471 * waiter's lwb structure to have already
2472 * been freed. Thus, we can only reliably
2473 * make these assertions if the waiter
2474 * isn't done.
2475 */
2476 ASSERT3P(lwb, ==, zcw->zcw_lwb);
2477 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_OPENED);
2478 }
2479 } else {
2480 /*
2481 * If the lwb isn't open, then it must have already
2482 * been issued. In that case, there's no need to
2483 * use a timeout when waiting for the lwb to
2484 * complete.
2485 *
2486 * Additionally, if the lwb is NULL, the waiter
2487 * will soon be signalled and marked done via
2488 * zil_clean() and zil_itxg_clean(), so no timeout
2489 * is required.
2490 */
2491
2492 IMPLY(lwb != NULL,
2493 lwb->lwb_state == LWB_STATE_ISSUED ||
2494 lwb->lwb_state == LWB_STATE_WRITE_DONE ||
2495 lwb->lwb_state == LWB_STATE_FLUSH_DONE);
2496 cv_wait(&zcw->zcw_cv, &zcw->zcw_lock);
2497 }
2498 }
2499
2500 mutex_exit(&zcw->zcw_lock);
2501 }
2502
2503 static zil_commit_waiter_t *
2504 zil_alloc_commit_waiter()
2505 {
2506 zil_commit_waiter_t *zcw = kmem_cache_alloc(zil_zcw_cache, KM_SLEEP);
2507
2508 cv_init(&zcw->zcw_cv, NULL, CV_DEFAULT, NULL);
2509 mutex_init(&zcw->zcw_lock, NULL, MUTEX_DEFAULT, NULL);
2510 list_link_init(&zcw->zcw_node);
2511 zcw->zcw_lwb = NULL;
2512 zcw->zcw_done = B_FALSE;
2513 zcw->zcw_zio_error = 0;
2514
2515 return (zcw);
2516 }
2517
2518 static void
2519 zil_free_commit_waiter(zil_commit_waiter_t *zcw)
2520 {
2521 ASSERT(!list_link_active(&zcw->zcw_node));
2522 ASSERT3P(zcw->zcw_lwb, ==, NULL);
2523 ASSERT3B(zcw->zcw_done, ==, B_TRUE);
2524 mutex_destroy(&zcw->zcw_lock);
2525 cv_destroy(&zcw->zcw_cv);
2526 kmem_cache_free(zil_zcw_cache, zcw);
2527 }
2528
2529 /*
2530 * This function is used to create a TX_COMMIT itx and assign it. This
2531 * way, it will be linked into the ZIL's list of synchronous itxs, and
2532 * then later committed to an lwb (or skipped) when
2533 * zil_process_commit_list() is called.
2534 */
2535 static void
2536 zil_commit_itx_assign(zilog_t *zilog, zil_commit_waiter_t *zcw)
2537 {
2538 dmu_tx_t *tx = dmu_tx_create(zilog->zl_os);
2539 VERIFY0(dmu_tx_assign(tx, TXG_WAIT));
2540
2541 itx_t *itx = zil_itx_create(TX_COMMIT, sizeof (lr_t));
2542 itx->itx_sync = B_TRUE;
2543 itx->itx_private = zcw;
2544
2545 zil_itx_assign(zilog, itx, tx);
2546
2547 dmu_tx_commit(tx);
2548 }
2549
2550 /*
2551 * Commit ZFS Intent Log transactions (itxs) to stable storage.
2552 *
2553 * When writing ZIL transactions to the on-disk representation of the
2554 * ZIL, the itxs are committed to a Log Write Block (lwb). Multiple
2555 * itxs can be committed to a single lwb. Once a lwb is written and
2556 * committed to stable storage (i.e. the lwb is written, and vdevs have
2557 * been flushed), each itx that was committed to that lwb is also
2558 * considered to be committed to stable storage.
2559 *
2560 * When an itx is committed to an lwb, the log record (lr_t) contained
2561 * by the itx is copied into the lwb's zio buffer, and once this buffer
2562 * is written to disk, it becomes an on-disk ZIL block.
2563 *
2564 * As itxs are generated, they're inserted into the ZIL's queue of
2565 * uncommitted itxs. The semantics of zil_commit() are such that it will
2566 * block until all itxs that were in the queue when it was called, are
2567 * committed to stable storage.
2568 *
2569 * If "foid" is zero, this means all "synchronous" and "asynchronous"
2570 * itxs, for all objects in the dataset, will be committed to stable
2571 * storage prior to zil_commit() returning. If "foid" is non-zero, all
2572 * "synchronous" itxs for all objects, but only "asynchronous" itxs
2573 * that correspond to the foid passed in, will be committed to stable
2574 * storage prior to zil_commit() returning.
2575 *
2576 * Generally speaking, when zil_commit() is called, the consumer doesn't
2577 * actually care about _all_ of the uncommitted itxs. Instead, they're
2578 * simply trying to waiting for a specific itx to be committed to disk,
2579 * but the interface(s) for interacting with the ZIL don't allow such
2580 * fine-grained communication. A better interface would allow a consumer
2581 * to create and assign an itx, and then pass a reference to this itx to
2582 * zil_commit(); such that zil_commit() would return as soon as that
2583 * specific itx was committed to disk (instead of waiting for _all_
2584 * itxs to be committed).
2585 *
2586 * When a thread calls zil_commit() a special "commit itx" will be
2587 * generated, along with a corresponding "waiter" for this commit itx.
2588 * zil_commit() will wait on this waiter's CV, such that when the waiter
2589 * is marked done, and signalled, zil_commit() will return.
2590 *
2591 * This commit itx is inserted into the queue of uncommitted itxs. This
2592 * provides an easy mechanism for determining which itxs were in the
2593 * queue prior to zil_commit() having been called, and which itxs were
2594 * added after zil_commit() was called.
2595 *
2596 * The commit it is special; it doesn't have any on-disk representation.
2597 * When a commit itx is "committed" to an lwb, the waiter associated
2598 * with it is linked onto the lwb's list of waiters. Then, when that lwb
2599 * completes, each waiter on the lwb's list is marked done and signalled
2600 * -- allowing the thread waiting on the waiter to return from zil_commit().
2601 *
2602 * It's important to point out a few critical factors that allow us
2603 * to make use of the commit itxs, commit waiters, per-lwb lists of
2604 * commit waiters, and zio completion callbacks like we're doing:
2605 *
2606 * 1. The list of waiters for each lwb is traversed, and each commit
2607 * waiter is marked "done" and signalled, in the zio completion
2608 * callback of the lwb's zio[*].
2609 *
2610 * * Actually, the waiters are signalled in the zio completion
2611 * callback of the root zio for the DKIOCFLUSHWRITECACHE commands
2612 * that are sent to the vdevs upon completion of the lwb zio.
2613 *
2614 * 2. When the itxs are inserted into the ZIL's queue of uncommitted
2615 * itxs, the order in which they are inserted is preserved[*]; as
2616 * itxs are added to the queue, they are added to the tail of
2617 * in-memory linked lists.
2618 *
2619 * When committing the itxs to lwbs (to be written to disk), they
2620 * are committed in the same order in which the itxs were added to
2621 * the uncommitted queue's linked list(s); i.e. the linked list of
2622 * itxs to commit is traversed from head to tail, and each itx is
2623 * committed to an lwb in that order.
2624 *
2625 * * To clarify:
2626 *
2627 * - the order of "sync" itxs is preserved w.r.t. other
2628 * "sync" itxs, regardless of the corresponding objects.
2629 * - the order of "async" itxs is preserved w.r.t. other
2630 * "async" itxs corresponding to the same object.
2631 * - the order of "async" itxs is *not* preserved w.r.t. other
2632 * "async" itxs corresponding to different objects.
2633 * - the order of "sync" itxs w.r.t. "async" itxs (or vice
2634 * versa) is *not* preserved, even for itxs that correspond
2635 * to the same object.
2636 *
2637 * For more details, see: zil_itx_assign(), zil_async_to_sync(),
2638 * zil_get_commit_list(), and zil_process_commit_list().
2639 *
2640 * 3. The lwbs represent a linked list of blocks on disk. Thus, any
2641 * lwb cannot be considered committed to stable storage, until its
2642 * "previous" lwb is also committed to stable storage. This fact,
2643 * coupled with the fact described above, means that itxs are
2644 * committed in (roughly) the order in which they were generated.
2645 * This is essential because itxs are dependent on prior itxs.
2646 * Thus, we *must not* deem an itx as being committed to stable
2647 * storage, until *all* prior itxs have also been committed to
2648 * stable storage.
2649 *
2650 * To enforce this ordering of lwb zio's, while still leveraging as
2651 * much of the underlying storage performance as possible, we rely
2652 * on two fundamental concepts:
2653 *
2654 * 1. The creation and issuance of lwb zio's is protected by
2655 * the zilog's "zl_issuer_lock", which ensures only a single
2656 * thread is creating and/or issuing lwb's at a time
2657 * 2. The "previous" lwb is a child of the "current" lwb
2658 * (leveraging the zio parent-child depenency graph)
2659 *
2660 * By relying on this parent-child zio relationship, we can have
2661 * many lwb zio's concurrently issued to the underlying storage,
2662 * but the order in which they complete will be the same order in
2663 * which they were created.
2664 */
2665 void
2666 zil_commit(zilog_t *zilog, uint64_t foid)
2667 {
2668 /*
2669 * We should never attempt to call zil_commit on a snapshot for
2670 * a couple of reasons:
2671 *
2672 * 1. A snapshot may never be modified, thus it cannot have any
2673 * in-flight itxs that would have modified the dataset.
2674 *
2675 * 2. By design, when zil_commit() is called, a commit itx will
2676 * be assigned to this zilog; as a result, the zilog will be
2677 * dirtied. We must not dirty the zilog of a snapshot; there's
2678 * checks in the code that enforce this invariant, and will
2679 * cause a panic if it's not upheld.
2680 */
2681 ASSERT3B(dmu_objset_is_snapshot(zilog->zl_os), ==, B_FALSE);
2682
2683 if (zilog->zl_sync == ZFS_SYNC_DISABLED)
2684 return;
2685
2686 if (!spa_writeable(zilog->zl_spa)) {
2687 /*
2688 * If the SPA is not writable, there should never be any
2689 * pending itxs waiting to be committed to disk. If that
2690 * weren't true, we'd skip writing those itxs out, and
2691 * would break the sematics of zil_commit(); thus, we're
2692 * verifying that truth before we return to the caller.
2693 */
2694 ASSERT(list_is_empty(&zilog->zl_lwb_list));
2695 ASSERT3P(zilog->zl_last_lwb_opened, ==, NULL);
2696 for (int i = 0; i < TXG_SIZE; i++)
2697 ASSERT3P(zilog->zl_itxg[i].itxg_itxs, ==, NULL);
2698 return;
2699 }
2700
2701 /*
2702 * If the ZIL is suspended, we don't want to dirty it by calling
2703 * zil_commit_itx_assign() below, nor can we write out
2704 * lwbs like would be done in zil_commit_write(). Thus, we
2705 * simply rely on txg_wait_synced() to maintain the necessary
2706 * semantics, and avoid calling those functions altogether.
2707 */
2708 if (zilog->zl_suspend > 0) {
2709 txg_wait_synced(zilog->zl_dmu_pool, 0);
2710 return;
2711 }
2712
2713 zil_commit_impl(zilog, foid);
2714 }
2715
2716 void
2717 zil_commit_impl(zilog_t *zilog, uint64_t foid)
2718 {
2719 /*
2720 * Move the "async" itxs for the specified foid to the "sync"
2721 * queues, such that they will be later committed (or skipped)
2722 * to an lwb when zil_process_commit_list() is called.
2723 *
2724 * Since these "async" itxs must be committed prior to this
2725 * call to zil_commit returning, we must perform this operation
2726 * before we call zil_commit_itx_assign().
2727 */
2728 zil_async_to_sync(zilog, foid);
2729
2730 /*
2731 * We allocate a new "waiter" structure which will initially be
2732 * linked to the commit itx using the itx's "itx_private" field.
2733 * Since the commit itx doesn't represent any on-disk state,
2734 * when it's committed to an lwb, rather than copying the its
2735 * lr_t into the lwb's buffer, the commit itx's "waiter" will be
2736 * added to the lwb's list of waiters. Then, when the lwb is
2737 * committed to stable storage, each waiter in the lwb's list of
2738 * waiters will be marked "done", and signalled.
2739 *
2740 * We must create the waiter and assign the commit itx prior to
2741 * calling zil_commit_writer(), or else our specific commit itx
2742 * is not guaranteed to be committed to an lwb prior to calling
2743 * zil_commit_waiter().
2744 */
2745 zil_commit_waiter_t *zcw = zil_alloc_commit_waiter();
2746 zil_commit_itx_assign(zilog, zcw);
2747
2748 zil_commit_writer(zilog, zcw);
2749 zil_commit_waiter(zilog, zcw);
2750
2751 if (zcw->zcw_zio_error != 0) {
2752 /*
2753 * If there was an error writing out the ZIL blocks that
2754 * this thread is waiting on, then we fallback to
2755 * relying on spa_sync() to write out the data this
2756 * thread is waiting on. Obviously this has performance
2757 * implications, but the expectation is for this to be
2758 * an exceptional case, and shouldn't occur often.
2759 */
2760 DTRACE_PROBE2(zil__commit__io__error,
2761 zilog_t *, zilog, zil_commit_waiter_t *, zcw);
2762 txg_wait_synced(zilog->zl_dmu_pool, 0);
2763 }
2764
2765 zil_free_commit_waiter(zcw);
2766 }
2767
2768 /*
2769 * Called in syncing context to free committed log blocks and update log header.
2770 */
2771 void
2772 zil_sync(zilog_t *zilog, dmu_tx_t *tx)
2773 {
2774 zil_header_t *zh = zil_header_in_syncing_context(zilog);
2775 uint64_t txg = dmu_tx_get_txg(tx);
2776 spa_t *spa = zilog->zl_spa;
2777 uint64_t *replayed_seq = &zilog->zl_replayed_seq[txg & TXG_MASK];
2778 lwb_t *lwb;
2779
2780 /*
2781 * We don't zero out zl_destroy_txg, so make sure we don't try
2782 * to destroy it twice.
2783 */
2784 if (spa_sync_pass(spa) != 1)
2785 return;
2786
2787 mutex_enter(&zilog->zl_lock);
2788
2789 ASSERT(zilog->zl_stop_sync == 0);
2790
2791 if (*replayed_seq != 0) {
2792 ASSERT(zh->zh_replay_seq < *replayed_seq);
2793 zh->zh_replay_seq = *replayed_seq;
2794 *replayed_seq = 0;
2795 }
2796
2797 if (zilog->zl_destroy_txg == txg) {
2798 blkptr_t blk = zh->zh_log;
2799
2800 ASSERT(list_head(&zilog->zl_lwb_list) == NULL);
2801
2802 bzero(zh, sizeof (zil_header_t));
2803 bzero(zilog->zl_replayed_seq, sizeof (zilog->zl_replayed_seq));
2804
2805 if (zilog->zl_keep_first) {
2806 /*
2807 * If this block was part of log chain that couldn't
2808 * be claimed because a device was missing during
2809 * zil_claim(), but that device later returns,
2810 * then this block could erroneously appear valid.
2811 * To guard against this, assign a new GUID to the new
2812 * log chain so it doesn't matter what blk points to.
2813 */
2814 zil_init_log_chain(zilog, &blk);
2815 zh->zh_log = blk;
2816 }
2817 }
2818
2819 while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) {
2820 zh->zh_log = lwb->lwb_blk;
2821 if (lwb->lwb_buf != NULL || lwb->lwb_max_txg > txg)
2822 break;
2823 list_remove(&zilog->zl_lwb_list, lwb);
2824 zio_free(spa, txg, &lwb->lwb_blk);
2825 zil_free_lwb(zilog, lwb);
2826
2827 /*
2828 * If we don't have anything left in the lwb list then
2829 * we've had an allocation failure and we need to zero
2830 * out the zil_header blkptr so that we don't end
2831 * up freeing the same block twice.
2832 */
2833 if (list_head(&zilog->zl_lwb_list) == NULL)
2834 BP_ZERO(&zh->zh_log);
2835 }
2836 mutex_exit(&zilog->zl_lock);
2837 }
2838
2839 /* ARGSUSED */
2840 static int
2841 zil_lwb_cons(void *vbuf, void *unused, int kmflag)
2842 {
2843 lwb_t *lwb = vbuf;
2844 list_create(&lwb->lwb_waiters, sizeof (zil_commit_waiter_t),
2845 offsetof(zil_commit_waiter_t, zcw_node));
2846 avl_create(&lwb->lwb_vdev_tree, zil_lwb_vdev_compare,
2847 sizeof (zil_vdev_node_t), offsetof(zil_vdev_node_t, zv_node));
2848 mutex_init(&lwb->lwb_vdev_lock, NULL, MUTEX_DEFAULT, NULL);
2849 return (0);
2850 }
2851
2852 /* ARGSUSED */
2853 static void
2854 zil_lwb_dest(void *vbuf, void *unused)
2855 {
2856 lwb_t *lwb = vbuf;
2857 mutex_destroy(&lwb->lwb_vdev_lock);
2858 avl_destroy(&lwb->lwb_vdev_tree);
2859 list_destroy(&lwb->lwb_waiters);
2860 }
2861
2862 void
2863 zil_init(void)
2864 {
2865 zil_lwb_cache = kmem_cache_create("zil_lwb_cache",
2866 sizeof (lwb_t), 0, zil_lwb_cons, zil_lwb_dest, NULL, NULL, NULL, 0);
2867
2868 zil_zcw_cache = kmem_cache_create("zil_zcw_cache",
2869 sizeof (zil_commit_waiter_t), 0, NULL, NULL, NULL, NULL, NULL, 0);
2870 }
2871
2872 void
2873 zil_fini(void)
2874 {
2875 kmem_cache_destroy(zil_zcw_cache);
2876 kmem_cache_destroy(zil_lwb_cache);
2877 }
2878
2879 void
2880 zil_set_sync(zilog_t *zilog, uint64_t sync)
2881 {
2882 zilog->zl_sync = sync;
2883 }
2884
2885 void
2886 zil_set_logbias(zilog_t *zilog, uint64_t logbias)
2887 {
2888 zilog->zl_logbias = logbias;
2889 }
2890
2891 zilog_t *
2892 zil_alloc(objset_t *os, zil_header_t *zh_phys)
2893 {
2894 zilog_t *zilog;
2895
2896 zilog = kmem_zalloc(sizeof (zilog_t), KM_SLEEP);
2897
2898 zilog->zl_header = zh_phys;
2899 zilog->zl_os = os;
2900 zilog->zl_spa = dmu_objset_spa(os);
2901 zilog->zl_dmu_pool = dmu_objset_pool(os);
2902 zilog->zl_destroy_txg = TXG_INITIAL - 1;
2903 zilog->zl_logbias = dmu_objset_logbias(os);
2904 zilog->zl_sync = dmu_objset_syncprop(os);
2905 zilog->zl_dirty_max_txg = 0;
2906 zilog->zl_last_lwb_opened = NULL;
2907 zilog->zl_last_lwb_latency = 0;
2908
2909 mutex_init(&zilog->zl_lock, NULL, MUTEX_DEFAULT, NULL);
2910 mutex_init(&zilog->zl_issuer_lock, NULL, MUTEX_DEFAULT, NULL);
2911
2912 for (int i = 0; i < TXG_SIZE; i++) {
2913 mutex_init(&zilog->zl_itxg[i].itxg_lock, NULL,
2914 MUTEX_DEFAULT, NULL);
2915 }
2916
2917 list_create(&zilog->zl_lwb_list, sizeof (lwb_t),
2918 offsetof(lwb_t, lwb_node));
2919
2920 list_create(&zilog->zl_itx_commit_list, sizeof (itx_t),
2921 offsetof(itx_t, itx_node));
2922
2923 cv_init(&zilog->zl_cv_suspend, NULL, CV_DEFAULT, NULL);
2924
2925 return (zilog);
2926 }
2927
2928 void
2929 zil_free(zilog_t *zilog)
2930 {
2931 zilog->zl_stop_sync = 1;
2932
2933 ASSERT0(zilog->zl_suspend);
2934 ASSERT0(zilog->zl_suspending);
2935
2936 ASSERT(list_is_empty(&zilog->zl_lwb_list));
2937 list_destroy(&zilog->zl_lwb_list);
2938
2939 ASSERT(list_is_empty(&zilog->zl_itx_commit_list));
2940 list_destroy(&zilog->zl_itx_commit_list);
2941
2942 for (int i = 0; i < TXG_SIZE; i++) {
2943 /*
2944 * It's possible for an itx to be generated that doesn't dirty
2945 * a txg (e.g. ztest TX_TRUNCATE). So there's no zil_clean()
2946 * callback to remove the entry. We remove those here.
2947 *
2948 * Also free up the ziltest itxs.
2949 */
2950 if (zilog->zl_itxg[i].itxg_itxs)
2951 zil_itxg_clean(zilog->zl_itxg[i].itxg_itxs);
2952 mutex_destroy(&zilog->zl_itxg[i].itxg_lock);
2953 }
2954
2955 mutex_destroy(&zilog->zl_issuer_lock);
2956 mutex_destroy(&zilog->zl_lock);
2957
2958 cv_destroy(&zilog->zl_cv_suspend);
2959
2960 kmem_free(zilog, sizeof (zilog_t));
2961 }
2962
2963 /*
2964 * Open an intent log.
2965 */
2966 zilog_t *
2967 zil_open(objset_t *os, zil_get_data_t *get_data)
2968 {
2969 zilog_t *zilog = dmu_objset_zil(os);
2970
2971 ASSERT3P(zilog->zl_get_data, ==, NULL);
2972 ASSERT3P(zilog->zl_last_lwb_opened, ==, NULL);
2973 ASSERT(list_is_empty(&zilog->zl_lwb_list));
2974
2975 zilog->zl_get_data = get_data;
2976
2977 return (zilog);
2978 }
2979
2980 /*
2981 * Close an intent log.
2982 */
2983 void
2984 zil_close(zilog_t *zilog)
2985 {
2986 lwb_t *lwb;
2987 uint64_t txg;
2988
2989 if (!dmu_objset_is_snapshot(zilog->zl_os)) {
2990 zil_commit(zilog, 0);
2991 } else {
2992 ASSERT3P(list_tail(&zilog->zl_lwb_list), ==, NULL);
2993 ASSERT0(zilog->zl_dirty_max_txg);
2994 ASSERT3B(zilog_is_dirty(zilog), ==, B_FALSE);
2995 }
2996
2997 mutex_enter(&zilog->zl_lock);
2998 lwb = list_tail(&zilog->zl_lwb_list);
2999 if (lwb == NULL)
3000 txg = zilog->zl_dirty_max_txg;
3001 else
3002 txg = MAX(zilog->zl_dirty_max_txg, lwb->lwb_max_txg);
3003 mutex_exit(&zilog->zl_lock);
3004
3005 /*
3006 * We need to use txg_wait_synced() to wait long enough for the
3007 * ZIL to be clean, and to wait for all pending lwbs to be
3008 * written out.
3009 */
3010 if (txg != 0)
3011 txg_wait_synced(zilog->zl_dmu_pool, txg);
3012
3013 if (zilog_is_dirty(zilog))
3014 zfs_dbgmsg("zil (%p) is dirty, txg %llu", zilog, txg);
3015 VERIFY(!zilog_is_dirty(zilog));
3016
3017 zilog->zl_get_data = NULL;
3018
3019 /*
3020 * We should have only one lwb left on the list; remove it now.
3021 */
3022 mutex_enter(&zilog->zl_lock);
3023 lwb = list_head(&zilog->zl_lwb_list);
3024 if (lwb != NULL) {
3025 ASSERT3P(lwb, ==, list_tail(&zilog->zl_lwb_list));
3026 ASSERT3S(lwb->lwb_state, !=, LWB_STATE_ISSUED);
3027 list_remove(&zilog->zl_lwb_list, lwb);
3028 zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
3029 zil_free_lwb(zilog, lwb);
3030 }
3031 mutex_exit(&zilog->zl_lock);
3032 }
3033
3034 static char *suspend_tag = "zil suspending";
3035
3036 /*
3037 * Suspend an intent log. While in suspended mode, we still honor
3038 * synchronous semantics, but we rely on txg_wait_synced() to do it.
3039 * On old version pools, we suspend the log briefly when taking a
3040 * snapshot so that it will have an empty intent log.
3041 *
3042 * Long holds are not really intended to be used the way we do here --
3043 * held for such a short time. A concurrent caller of dsl_dataset_long_held()
3044 * could fail. Therefore we take pains to only put a long hold if it is
3045 * actually necessary. Fortunately, it will only be necessary if the
3046 * objset is currently mounted (or the ZVOL equivalent). In that case it
3047 * will already have a long hold, so we are not really making things any worse.
3048 *
3049 * Ideally, we would locate the existing long-holder (i.e. the zfsvfs_t or
3050 * zvol_state_t), and use their mechanism to prevent their hold from being
3051 * dropped (e.g. VFS_HOLD()). However, that would be even more pain for
3052 * very little gain.
3053 *
3054 * if cookiep == NULL, this does both the suspend & resume.
3055 * Otherwise, it returns with the dataset "long held", and the cookie
3056 * should be passed into zil_resume().
3057 */
3058 int
3059 zil_suspend(const char *osname, void **cookiep)
3060 {
3061 objset_t *os;
3062 zilog_t *zilog;
3063 const zil_header_t *zh;
3064 int error;
3065
3066 error = dmu_objset_hold(osname, suspend_tag, &os);
3067 if (error != 0)
3068 return (error);
3069 zilog = dmu_objset_zil(os);
3070
3071 mutex_enter(&zilog->zl_lock);
3072 zh = zilog->zl_header;
3073
3074 if (zh->zh_flags & ZIL_REPLAY_NEEDED) { /* unplayed log */
3075 mutex_exit(&zilog->zl_lock);
3076 dmu_objset_rele(os, suspend_tag);
3077 return (SET_ERROR(EBUSY));
3078 }
3079
3080 /*
3081 * Don't put a long hold in the cases where we can avoid it. This
3082 * is when there is no cookie so we are doing a suspend & resume
3083 * (i.e. called from zil_vdev_offline()), and there's nothing to do
3084 * for the suspend because it's already suspended, or there's no ZIL.
3085 */
3086 if (cookiep == NULL && !zilog->zl_suspending &&
3087 (zilog->zl_suspend > 0 || BP_IS_HOLE(&zh->zh_log))) {
3088 mutex_exit(&zilog->zl_lock);
3089 dmu_objset_rele(os, suspend_tag);
3090 return (0);
3091 }
3092
3093 dsl_dataset_long_hold(dmu_objset_ds(os), suspend_tag);
3094 dsl_pool_rele(dmu_objset_pool(os), suspend_tag);
3095
3096 zilog->zl_suspend++;
3097
3098 if (zilog->zl_suspend > 1) {
3099 /*
3100 * Someone else is already suspending it.
3101 * Just wait for them to finish.
3102 */
3103
3104 while (zilog->zl_suspending)
3105 cv_wait(&zilog->zl_cv_suspend, &zilog->zl_lock);
3106 mutex_exit(&zilog->zl_lock);
3107
3108 if (cookiep == NULL)
3109 zil_resume(os);
3110 else
3111 *cookiep = os;
3112 return (0);
3113 }
3114
3115 /*
3116 * If there is no pointer to an on-disk block, this ZIL must not
3117 * be active (e.g. filesystem not mounted), so there's nothing
3118 * to clean up.
3119 */
3120 if (BP_IS_HOLE(&zh->zh_log)) {
3121 ASSERT(cookiep != NULL); /* fast path already handled */
3122
3123 *cookiep = os;
3124 mutex_exit(&zilog->zl_lock);
3125 return (0);
3126 }
3127
3128 zilog->zl_suspending = B_TRUE;
3129 mutex_exit(&zilog->zl_lock);
3130
3131 /*
3132 * We need to use zil_commit_impl to ensure we wait for all
3133 * LWB_STATE_OPENED and LWB_STATE_ISSUED lwb's to be committed
3134 * to disk before proceeding. If we used zil_commit instead, it
3135 * would just call txg_wait_synced(), because zl_suspend is set.
3136 * txg_wait_synced() doesn't wait for these lwb's to be
3137 * LWB_STATE_FLUSH_DONE before returning.
3138 */
3139 zil_commit_impl(zilog, 0);
3140
3141 /*
3142 * Now that we've ensured all lwb's are LWB_STATE_FLUSH_DONE, we
3143 * use txg_wait_synced() to ensure the data from the zilog has
3144 * migrated to the main pool before calling zil_destroy().
3145 */
3146 txg_wait_synced(zilog->zl_dmu_pool, 0);
3147
3148 zil_destroy(zilog, B_FALSE);
3149
3150 mutex_enter(&zilog->zl_lock);
3151 zilog->zl_suspending = B_FALSE;
3152 cv_broadcast(&zilog->zl_cv_suspend);
3153 mutex_exit(&zilog->zl_lock);
3154
3155 if (cookiep == NULL)
3156 zil_resume(os);
3157 else
3158 *cookiep = os;
3159 return (0);
3160 }
3161
3162 void
3163 zil_resume(void *cookie)
3164 {
3165 objset_t *os = cookie;
3166 zilog_t *zilog = dmu_objset_zil(os);
3167
3168 mutex_enter(&zilog->zl_lock);
3169 ASSERT(zilog->zl_suspend != 0);
3170 zilog->zl_suspend--;
3171 mutex_exit(&zilog->zl_lock);
3172 dsl_dataset_long_rele(dmu_objset_ds(os), suspend_tag);
3173 dsl_dataset_rele(dmu_objset_ds(os), suspend_tag);
3174 }
3175
3176 typedef struct zil_replay_arg {
3177 zil_replay_func_t **zr_replay;
3178 void *zr_arg;
3179 boolean_t zr_byteswap;
3180 char *zr_lr;
3181 } zil_replay_arg_t;
3182
3183 static int
3184 zil_replay_error(zilog_t *zilog, lr_t *lr, int error)
3185 {
3186 char name[ZFS_MAX_DATASET_NAME_LEN];
3187
3188 zilog->zl_replaying_seq--; /* didn't actually replay this one */
3189
3190 dmu_objset_name(zilog->zl_os, name);
3191
3192 cmn_err(CE_WARN, "ZFS replay transaction error %d, "
3193 "dataset %s, seq 0x%llx, txtype %llu %s\n", error, name,
3194 (u_longlong_t)lr->lrc_seq,
3195 (u_longlong_t)(lr->lrc_txtype & ~TX_CI),
3196 (lr->lrc_txtype & TX_CI) ? "CI" : "");
3197
3198 return (error);
3199 }
3200
3201 static int
3202 zil_replay_log_record(zilog_t *zilog, lr_t *lr, void *zra, uint64_t claim_txg)
3203 {
3204 zil_replay_arg_t *zr = zra;
3205 const zil_header_t *zh = zilog->zl_header;
3206 uint64_t reclen = lr->lrc_reclen;
3207 uint64_t txtype = lr->lrc_txtype;
3208 int error = 0;
3209
3210 zilog->zl_replaying_seq = lr->lrc_seq;
3211
3212 if (lr->lrc_seq <= zh->zh_replay_seq) /* already replayed */
3213 return (0);
3214
3215 if (lr->lrc_txg < claim_txg) /* already committed */
3216 return (0);
3217
3218 /* Strip case-insensitive bit, still present in log record */
3219 txtype &= ~TX_CI;
3220
3221 if (txtype == 0 || txtype >= TX_MAX_TYPE)
3222 return (zil_replay_error(zilog, lr, EINVAL));
3223
3224 /*
3225 * If this record type can be logged out of order, the object
3226 * (lr_foid) may no longer exist. That's legitimate, not an error.
3227 */
3228 if (TX_OOO(txtype)) {
3229 error = dmu_object_info(zilog->zl_os,
3230 ((lr_ooo_t *)lr)->lr_foid, NULL);
3231 if (error == ENOENT || error == EEXIST)
3232 return (0);
3233 }
3234
3235 /*
3236 * Make a copy of the data so we can revise and extend it.
3237 */
3238 bcopy(lr, zr->zr_lr, reclen);
3239
3240 /*
3241 * If this is a TX_WRITE with a blkptr, suck in the data.
3242 */
3243 if (txtype == TX_WRITE && reclen == sizeof (lr_write_t)) {
3244 error = zil_read_log_data(zilog, (lr_write_t *)lr,
3245 zr->zr_lr + reclen);
3246 if (error != 0)
3247 return (zil_replay_error(zilog, lr, error));
3248 }
3249
3250 /*
3251 * The log block containing this lr may have been byteswapped
3252 * so that we can easily examine common fields like lrc_txtype.
3253 * However, the log is a mix of different record types, and only the
3254 * replay vectors know how to byteswap their records. Therefore, if
3255 * the lr was byteswapped, undo it before invoking the replay vector.
3256 */
3257 if (zr->zr_byteswap)
3258 byteswap_uint64_array(zr->zr_lr, reclen);
3259
3260 /*
3261 * We must now do two things atomically: replay this log record,
3262 * and update the log header sequence number to reflect the fact that
3263 * we did so. At the end of each replay function the sequence number
3264 * is updated if we are in replay mode.
3265 */
3266 error = zr->zr_replay[txtype](zr->zr_arg, zr->zr_lr, zr->zr_byteswap);
3267 if (error != 0) {
3268 /*
3269 * The DMU's dnode layer doesn't see removes until the txg
3270 * commits, so a subsequent claim can spuriously fail with
3271 * EEXIST. So if we receive any error we try syncing out
3272 * any removes then retry the transaction. Note that we
3273 * specify B_FALSE for byteswap now, so we don't do it twice.
3274 */
3275 txg_wait_synced(spa_get_dsl(zilog->zl_spa), 0);
3276 error = zr->zr_replay[txtype](zr->zr_arg, zr->zr_lr, B_FALSE);
3277 if (error != 0)
3278 return (zil_replay_error(zilog, lr, error));
3279 }
3280 return (0);
3281 }
3282
3283 /* ARGSUSED */
3284 static int
3285 zil_incr_blks(zilog_t *zilog, blkptr_t *bp, void *arg, uint64_t claim_txg)
3286 {
3287 zilog->zl_replay_blks++;
3288
3289 return (0);
3290 }
3291
3292 /*
3293 * If this dataset has a non-empty intent log, replay it and destroy it.
3294 */
3295 void
3296 zil_replay(objset_t *os, void *arg, zil_replay_func_t *replay_func[TX_MAX_TYPE])
3297 {
3298 zilog_t *zilog = dmu_objset_zil(os);
3299 const zil_header_t *zh = zilog->zl_header;
3300 zil_replay_arg_t zr;
3301
3302 if ((zh->zh_flags & ZIL_REPLAY_NEEDED) == 0) {
3303 zil_destroy(zilog, B_TRUE);
3304 return;
3305 }
3306
3307 zr.zr_replay = replay_func;
3308 zr.zr_arg = arg;
3309 zr.zr_byteswap = BP_SHOULD_BYTESWAP(&zh->zh_log);
3310 zr.zr_lr = kmem_alloc(2 * SPA_MAXBLOCKSIZE, KM_SLEEP);
3311
3312 /*
3313 * Wait for in-progress removes to sync before starting replay.
3314 */
3315 txg_wait_synced(zilog->zl_dmu_pool, 0);
3316
3317 zilog->zl_replay = B_TRUE;
3318 zilog->zl_replay_time = ddi_get_lbolt();
3319 ASSERT(zilog->zl_replay_blks == 0);
3320 (void) zil_parse(zilog, zil_incr_blks, zil_replay_log_record, &zr,
3321 zh->zh_claim_txg);
3322 kmem_free(zr.zr_lr, 2 * SPA_MAXBLOCKSIZE);
3323
3324 zil_destroy(zilog, B_FALSE);
3325 txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
3326 zilog->zl_replay = B_FALSE;
3327 }
3328
3329 boolean_t
3330 zil_replaying(zilog_t *zilog, dmu_tx_t *tx)
3331 {
3332 if (zilog->zl_sync == ZFS_SYNC_DISABLED)
3333 return (B_TRUE);
3334
3335 if (zilog->zl_replay) {
3336 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
3337 zilog->zl_replayed_seq[dmu_tx_get_txg(tx) & TXG_MASK] =
3338 zilog->zl_replaying_seq;
3339 return (B_TRUE);
3340 }
3341
3342 return (B_FALSE);
3343 }
3344
3345 /* ARGSUSED */
3346 int
3347 zil_vdev_offline(const char *osname, void *arg)
3348 {
3349 int error;
3350
3351 error = zil_suspend(osname, NULL);
3352 if (error != 0)
3353 return (SET_ERROR(EEXIST));
3354 return (0);
3355 }