Add FASTWRITE algorithm for synchronous writes.
[zfs.git] / module / zfs / zil.c
1 /*
2  * CDDL HEADER START
3  *
4  * The contents of this file are subject to the terms of the
5  * Common Development and Distribution License (the "License").
6  * You may not use this file except in compliance with the License.
7  *
8  * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
9  * or http://www.opensolaris.org/os/licensing.
10  * See the License for the specific language governing permissions
11  * and limitations under the License.
12  *
13  * When distributing Covered Code, include this CDDL HEADER in each
14  * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
15  * If applicable, add the following below this CDDL HEADER, with the
16  * fields enclosed by brackets "[]" replaced with your own identifying
17  * information: Portions Copyright [yyyy] [name of copyright owner]
18  *
19  * CDDL HEADER END
20  */
21 /*
22  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
23  * Copyright (c) 2011 by Delphix. All rights reserved.
24  */
25
26 /* Portions Copyright 2010 Robert Milkowski */
27
28 #include <sys/zfs_context.h>
29 #include <sys/spa.h>
30 #include <sys/dmu.h>
31 #include <sys/zap.h>
32 #include <sys/arc.h>
33 #include <sys/stat.h>
34 #include <sys/resource.h>
35 #include <sys/zil.h>
36 #include <sys/zil_impl.h>
37 #include <sys/dsl_dataset.h>
38 #include <sys/vdev_impl.h>
39 #include <sys/dmu_tx.h>
40 #include <sys/dsl_pool.h>
41 #include <sys/metaslab.h>
42
43 /*
44  * The zfs intent log (ZIL) saves transaction records of system calls
45  * that change the file system in memory with enough information
46  * to be able to replay them. These are stored in memory until
47  * either the DMU transaction group (txg) commits them to the stable pool
48  * and they can be discarded, or they are flushed to the stable log
49  * (also in the pool) due to a fsync, O_DSYNC or other synchronous
50  * requirement. In the event of a panic or power fail then those log
51  * records (transactions) are replayed.
52  *
53  * There is one ZIL per file system. Its on-disk (pool) format consists
54  * of 3 parts:
55  *
56  *      - ZIL header
57  *      - ZIL blocks
58  *      - ZIL records
59  *
60  * A log record holds a system call transaction. Log blocks can
61  * hold many log records and the blocks are chained together.
62  * Each ZIL block contains a block pointer (blkptr_t) to the next
63  * ZIL block in the chain. The ZIL header points to the first
64  * block in the chain. Note there is not a fixed place in the pool
65  * to hold blocks. They are dynamically allocated and freed as
66  * needed from the blocks available. Figure X shows the ZIL structure:
67  */
68
69 /*
70  * See zil.h for more information about these fields.
71  */
72 zil_stats_t zil_stats = {
73         { "zil_commit_count",              KSTAT_DATA_UINT64 },
74         { "zil_commit_writer_count",       KSTAT_DATA_UINT64 },
75         { "zil_itx_count",                 KSTAT_DATA_UINT64 },
76         { "zil_itx_indirect_count",        KSTAT_DATA_UINT64 },
77         { "zil_itx_indirect_bytes",        KSTAT_DATA_UINT64 },
78         { "zil_itx_copied_count",          KSTAT_DATA_UINT64 },
79         { "zil_itx_copied_bytes",          KSTAT_DATA_UINT64 },
80         { "zil_itx_needcopy_count",        KSTAT_DATA_UINT64 },
81         { "zil_itx_needcopy_bytes",        KSTAT_DATA_UINT64 },
82         { "zil_itx_metaslab_normal_count", KSTAT_DATA_UINT64 },
83         { "zil_itx_metaslab_normal_bytes", KSTAT_DATA_UINT64 },
84         { "zil_itx_metaslab_slog_count",   KSTAT_DATA_UINT64 },
85         { "zil_itx_metaslab_slog_bytes",   KSTAT_DATA_UINT64 },
86 };
87
88 static kstat_t *zil_ksp;
89
90 /*
91  * This global ZIL switch affects all pools
92  */
93 int zil_replay_disable = 0;    /* disable intent logging replay */
94
95 /*
96  * Tunable parameter for debugging or performance analysis.  Setting
97  * zfs_nocacheflush will cause corruption on power loss if a volatile
98  * out-of-order write cache is enabled.
99  */
100 int zfs_nocacheflush = 0;
101
102 static kmem_cache_t *zil_lwb_cache;
103
104 static void zil_async_to_sync(zilog_t *zilog, uint64_t foid);
105
106 #define LWB_EMPTY(lwb) ((BP_GET_LSIZE(&lwb->lwb_blk) - \
107     sizeof (zil_chain_t)) == (lwb->lwb_sz - lwb->lwb_nused))
108
109
110 /*
111  * ziltest is by and large an ugly hack, but very useful in
112  * checking replay without tedious work.
113  * When running ziltest we want to keep all itx's and so maintain
114  * a single list in the zl_itxg[] that uses a high txg: ZILTEST_TXG
115  * We subtract TXG_CONCURRENT_STATES to allow for common code.
116  */
117 #define ZILTEST_TXG (UINT64_MAX - TXG_CONCURRENT_STATES)
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 = BP_IDENTITY(bp);
163         zil_bp_node_t *zn;
164         avl_index_t where;
165
166         if (avl_find(t, dva, &where) != NULL)
167                 return (EEXIST);
168
169         zn = kmem_alloc(sizeof (zil_bp_node_t), KM_PUSHPAGE);
170         zn->zn_dva = *dva;
171         avl_insert(t, zn, where);
172
173         return (0);
174 }
175
176 static zil_header_t *
177 zil_header_in_syncing_context(zilog_t *zilog)
178 {
179         return ((zil_header_t *)zilog->zl_header);
180 }
181
182 static void
183 zil_init_log_chain(zilog_t *zilog, blkptr_t *bp)
184 {
185         zio_cksum_t *zc = &bp->blk_cksum;
186
187         zc->zc_word[ZIL_ZC_GUID_0] = spa_get_random(-1ULL);
188         zc->zc_word[ZIL_ZC_GUID_1] = spa_get_random(-1ULL);
189         zc->zc_word[ZIL_ZC_OBJSET] = dmu_objset_id(zilog->zl_os);
190         zc->zc_word[ZIL_ZC_SEQ] = 1ULL;
191 }
192
193 /*
194  * Read a log block and make sure it's valid.
195  */
196 static int
197 zil_read_log_block(zilog_t *zilog, const blkptr_t *bp, blkptr_t *nbp, void *dst,
198     char **end)
199 {
200         enum zio_flag zio_flags = ZIO_FLAG_CANFAIL;
201         uint32_t aflags = ARC_WAIT;
202         arc_buf_t *abuf = NULL;
203         zbookmark_t zb;
204         int error;
205
206         if (zilog->zl_header->zh_claim_txg == 0)
207                 zio_flags |= ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB;
208
209         if (!(zilog->zl_header->zh_flags & ZIL_CLAIM_LR_SEQ_VALID))
210                 zio_flags |= ZIO_FLAG_SPECULATIVE;
211
212         SET_BOOKMARK(&zb, bp->blk_cksum.zc_word[ZIL_ZC_OBJSET],
213             ZB_ZIL_OBJECT, ZB_ZIL_LEVEL, bp->blk_cksum.zc_word[ZIL_ZC_SEQ]);
214
215         error = dsl_read_nolock(NULL, zilog->zl_spa, bp, arc_getbuf_func, &abuf,
216             ZIO_PRIORITY_SYNC_READ, zio_flags, &aflags, &zb);
217
218         if (error == 0) {
219                 zio_cksum_t cksum = bp->blk_cksum;
220
221                 /*
222                  * Validate the checksummed log block.
223                  *
224                  * Sequence numbers should be... sequential.  The checksum
225                  * verifier for the next block should be bp's checksum plus 1.
226                  *
227                  * Also check the log chain linkage and size used.
228                  */
229                 cksum.zc_word[ZIL_ZC_SEQ]++;
230
231                 if (BP_GET_CHECKSUM(bp) == ZIO_CHECKSUM_ZILOG2) {
232                         zil_chain_t *zilc = abuf->b_data;
233                         char *lr = (char *)(zilc + 1);
234                         uint64_t len = zilc->zc_nused - sizeof (zil_chain_t);
235
236                         if (bcmp(&cksum, &zilc->zc_next_blk.blk_cksum,
237                             sizeof (cksum)) || BP_IS_HOLE(&zilc->zc_next_blk)) {
238                                 error = ECKSUM;
239                         } else {
240                                 bcopy(lr, dst, len);
241                                 *end = (char *)dst + len;
242                                 *nbp = zilc->zc_next_blk;
243                         }
244                 } else {
245                         char *lr = abuf->b_data;
246                         uint64_t size = BP_GET_LSIZE(bp);
247                         zil_chain_t *zilc = (zil_chain_t *)(lr + size) - 1;
248
249                         if (bcmp(&cksum, &zilc->zc_next_blk.blk_cksum,
250                             sizeof (cksum)) || BP_IS_HOLE(&zilc->zc_next_blk) ||
251                             (zilc->zc_nused > (size - sizeof (*zilc)))) {
252                                 error = ECKSUM;
253                         } else {
254                                 bcopy(lr, dst, zilc->zc_nused);
255                                 *end = (char *)dst + zilc->zc_nused;
256                                 *nbp = zilc->zc_next_blk;
257                         }
258                 }
259
260                 VERIFY(arc_buf_remove_ref(abuf, &abuf) == 1);
261         }
262
263         return (error);
264 }
265
266 /*
267  * Read a TX_WRITE log data block.
268  */
269 static int
270 zil_read_log_data(zilog_t *zilog, const lr_write_t *lr, void *wbuf)
271 {
272         enum zio_flag zio_flags = ZIO_FLAG_CANFAIL;
273         const blkptr_t *bp = &lr->lr_blkptr;
274         uint32_t aflags = ARC_WAIT;
275         arc_buf_t *abuf = NULL;
276         zbookmark_t zb;
277         int error;
278
279         if (BP_IS_HOLE(bp)) {
280                 if (wbuf != NULL)
281                         bzero(wbuf, MAX(BP_GET_LSIZE(bp), lr->lr_length));
282                 return (0);
283         }
284
285         if (zilog->zl_header->zh_claim_txg == 0)
286                 zio_flags |= ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB;
287
288         SET_BOOKMARK(&zb, dmu_objset_id(zilog->zl_os), lr->lr_foid,
289             ZB_ZIL_LEVEL, lr->lr_offset / BP_GET_LSIZE(bp));
290
291         error = arc_read_nolock(NULL, zilog->zl_spa, bp, arc_getbuf_func, &abuf,
292             ZIO_PRIORITY_SYNC_READ, zio_flags, &aflags, &zb);
293
294         if (error == 0) {
295                 if (wbuf != NULL)
296                         bcopy(abuf->b_data, wbuf, arc_buf_size(abuf));
297                 (void) arc_buf_remove_ref(abuf, &abuf);
298         }
299
300         return (error);
301 }
302
303 /*
304  * Parse the intent log, and call parse_func for each valid record within.
305  */
306 int
307 zil_parse(zilog_t *zilog, zil_parse_blk_func_t *parse_blk_func,
308     zil_parse_lr_func_t *parse_lr_func, void *arg, uint64_t txg)
309 {
310         const zil_header_t *zh = zilog->zl_header;
311         boolean_t claimed = !!zh->zh_claim_txg;
312         uint64_t claim_blk_seq = claimed ? zh->zh_claim_blk_seq : UINT64_MAX;
313         uint64_t claim_lr_seq = claimed ? zh->zh_claim_lr_seq : UINT64_MAX;
314         uint64_t max_blk_seq = 0;
315         uint64_t max_lr_seq = 0;
316         uint64_t blk_count = 0;
317         uint64_t lr_count = 0;
318         blkptr_t blk, next_blk;
319         char *lrbuf, *lrp;
320         int error = 0;
321
322         bzero(&next_blk, sizeof(blkptr_t));
323
324         /*
325          * Old logs didn't record the maximum zh_claim_lr_seq.
326          */
327         if (!(zh->zh_flags & ZIL_CLAIM_LR_SEQ_VALID))
328                 claim_lr_seq = UINT64_MAX;
329
330         /*
331          * Starting at the block pointed to by zh_log we read the log chain.
332          * For each block in the chain we strongly check that block to
333          * ensure its validity.  We stop when an invalid block is found.
334          * For each block pointer in the chain we call parse_blk_func().
335          * For each record in each valid block we call parse_lr_func().
336          * If the log has been claimed, stop if we encounter a sequence
337          * number greater than the highest claimed sequence number.
338          */
339         lrbuf = zio_buf_alloc(SPA_MAXBLOCKSIZE);
340         zil_bp_tree_init(zilog);
341
342         for (blk = zh->zh_log; !BP_IS_HOLE(&blk); blk = next_blk) {
343                 uint64_t blk_seq = blk.blk_cksum.zc_word[ZIL_ZC_SEQ];
344                 int reclen;
345                 char *end = NULL;
346
347                 if (blk_seq > claim_blk_seq)
348                         break;
349                 if ((error = parse_blk_func(zilog, &blk, arg, txg)) != 0)
350                         break;
351                 ASSERT3U(max_blk_seq, <, blk_seq);
352                 max_blk_seq = blk_seq;
353                 blk_count++;
354
355                 if (max_lr_seq == claim_lr_seq && max_blk_seq == claim_blk_seq)
356                         break;
357
358                 error = zil_read_log_block(zilog, &blk, &next_blk, lrbuf, &end);
359                 if (error)
360                         break;
361
362                 for (lrp = lrbuf; lrp < end; lrp += reclen) {
363                         lr_t *lr = (lr_t *)lrp;
364                         reclen = lr->lrc_reclen;
365                         ASSERT3U(reclen, >=, sizeof (lr_t));
366                         if (lr->lrc_seq > claim_lr_seq)
367                                 goto done;
368                         if ((error = parse_lr_func(zilog, lr, arg, txg)) != 0)
369                                 goto done;
370                         ASSERT3U(max_lr_seq, <, lr->lrc_seq);
371                         max_lr_seq = lr->lrc_seq;
372                         lr_count++;
373                 }
374         }
375 done:
376         zilog->zl_parse_error = error;
377         zilog->zl_parse_blk_seq = max_blk_seq;
378         zilog->zl_parse_lr_seq = max_lr_seq;
379         zilog->zl_parse_blk_count = blk_count;
380         zilog->zl_parse_lr_count = lr_count;
381
382         ASSERT(!claimed || !(zh->zh_flags & ZIL_CLAIM_LR_SEQ_VALID) ||
383             (max_blk_seq == claim_blk_seq && max_lr_seq == claim_lr_seq));
384
385         zil_bp_tree_fini(zilog);
386         zio_buf_free(lrbuf, SPA_MAXBLOCKSIZE);
387
388         return (error);
389 }
390
391 static int
392 zil_claim_log_block(zilog_t *zilog, blkptr_t *bp, void *tx, uint64_t first_txg)
393 {
394         /*
395          * Claim log block if not already committed and not already claimed.
396          * If tx == NULL, just verify that the block is claimable.
397          */
398         if (bp->blk_birth < first_txg || zil_bp_tree_add(zilog, bp) != 0)
399                 return (0);
400
401         return (zio_wait(zio_claim(NULL, zilog->zl_spa,
402             tx == NULL ? 0 : first_txg, bp, spa_claim_notify, NULL,
403             ZIO_FLAG_CANFAIL | ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB)));
404 }
405
406 static int
407 zil_claim_log_record(zilog_t *zilog, lr_t *lrc, void *tx, uint64_t first_txg)
408 {
409         lr_write_t *lr = (lr_write_t *)lrc;
410         int error;
411
412         if (lrc->lrc_txtype != TX_WRITE)
413                 return (0);
414
415         /*
416          * If the block is not readable, don't claim it.  This can happen
417          * in normal operation when a log block is written to disk before
418          * some of the dmu_sync() blocks it points to.  In this case, the
419          * transaction cannot have been committed to anyone (we would have
420          * waited for all writes to be stable first), so it is semantically
421          * correct to declare this the end of the log.
422          */
423         if (lr->lr_blkptr.blk_birth >= first_txg &&
424             (error = zil_read_log_data(zilog, lr, NULL)) != 0)
425                 return (error);
426         return (zil_claim_log_block(zilog, &lr->lr_blkptr, tx, first_txg));
427 }
428
429 /* ARGSUSED */
430 static int
431 zil_free_log_block(zilog_t *zilog, blkptr_t *bp, void *tx, uint64_t claim_txg)
432 {
433         zio_free_zil(zilog->zl_spa, dmu_tx_get_txg(tx), bp);
434
435         return (0);
436 }
437
438 static int
439 zil_free_log_record(zilog_t *zilog, lr_t *lrc, void *tx, uint64_t claim_txg)
440 {
441         lr_write_t *lr = (lr_write_t *)lrc;
442         blkptr_t *bp = &lr->lr_blkptr;
443
444         /*
445          * If we previously claimed it, we need to free it.
446          */
447         if (claim_txg != 0 && lrc->lrc_txtype == TX_WRITE &&
448             bp->blk_birth >= claim_txg && zil_bp_tree_add(zilog, bp) == 0)
449                 zio_free(zilog->zl_spa, dmu_tx_get_txg(tx), bp);
450
451         return (0);
452 }
453
454 static lwb_t *
455 zil_alloc_lwb(zilog_t *zilog, blkptr_t *bp, uint64_t txg, boolean_t fastwrite)
456 {
457         lwb_t *lwb;
458
459         lwb = kmem_cache_alloc(zil_lwb_cache, KM_PUSHPAGE);
460         lwb->lwb_zilog = zilog;
461         lwb->lwb_blk = *bp;
462         lwb->lwb_fastwrite = fastwrite;
463         lwb->lwb_buf = zio_buf_alloc(BP_GET_LSIZE(bp));
464         lwb->lwb_max_txg = txg;
465         lwb->lwb_zio = NULL;
466         lwb->lwb_tx = NULL;
467         if (BP_GET_CHECKSUM(bp) == ZIO_CHECKSUM_ZILOG2) {
468                 lwb->lwb_nused = sizeof (zil_chain_t);
469                 lwb->lwb_sz = BP_GET_LSIZE(bp);
470         } else {
471                 lwb->lwb_nused = 0;
472                 lwb->lwb_sz = BP_GET_LSIZE(bp) - sizeof (zil_chain_t);
473         }
474
475         mutex_enter(&zilog->zl_lock);
476         list_insert_tail(&zilog->zl_lwb_list, lwb);
477         mutex_exit(&zilog->zl_lock);
478
479         return (lwb);
480 }
481
482 /*
483  * Create an on-disk intent log.
484  */
485 static lwb_t *
486 zil_create(zilog_t *zilog)
487 {
488         const zil_header_t *zh = zilog->zl_header;
489         lwb_t *lwb = NULL;
490         uint64_t txg = 0;
491         dmu_tx_t *tx = NULL;
492         blkptr_t blk;
493         int error = 0;
494         boolean_t fastwrite = FALSE;
495
496         /*
497          * Wait for any previous destroy to complete.
498          */
499         txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
500
501         ASSERT(zh->zh_claim_txg == 0);
502         ASSERT(zh->zh_replay_seq == 0);
503
504         blk = zh->zh_log;
505
506         /*
507          * Allocate an initial log block if:
508          *    - there isn't one already
509          *    - the existing block is the wrong endianess
510          */
511         if (BP_IS_HOLE(&blk) || BP_SHOULD_BYTESWAP(&blk)) {
512                 tx = dmu_tx_create(zilog->zl_os);
513                 VERIFY(dmu_tx_assign(tx, TXG_WAIT) == 0);
514                 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
515                 txg = dmu_tx_get_txg(tx);
516
517                 if (!BP_IS_HOLE(&blk)) {
518                         zio_free_zil(zilog->zl_spa, txg, &blk);
519                         BP_ZERO(&blk);
520                 }
521
522                 error = zio_alloc_zil(zilog->zl_spa, txg, &blk,
523                     ZIL_MIN_BLKSZ, zilog->zl_logbias == ZFS_LOGBIAS_LATENCY);
524                 fastwrite = TRUE;
525
526                 if (error == 0)
527                         zil_init_log_chain(zilog, &blk);
528         }
529
530         /*
531          * Allocate a log write buffer (lwb) for the first log block.
532          */
533         if (error == 0)
534                 lwb = zil_alloc_lwb(zilog, &blk, txg, fastwrite);
535
536         /*
537          * If we just allocated the first log block, commit our transaction
538          * and wait for zil_sync() to stuff the block poiner into zh_log.
539          * (zh is part of the MOS, so we cannot modify it in open context.)
540          */
541         if (tx != NULL) {
542                 dmu_tx_commit(tx);
543                 txg_wait_synced(zilog->zl_dmu_pool, txg);
544         }
545
546         ASSERT(bcmp(&blk, &zh->zh_log, sizeof (blk)) == 0);
547
548         return (lwb);
549 }
550
551 /*
552  * In one tx, free all log blocks and clear the log header.
553  * If keep_first is set, then we're replaying a log with no content.
554  * We want to keep the first block, however, so that the first
555  * synchronous transaction doesn't require a txg_wait_synced()
556  * in zil_create().  We don't need to txg_wait_synced() here either
557  * when keep_first is set, because both zil_create() and zil_destroy()
558  * will wait for any in-progress destroys to complete.
559  */
560 void
561 zil_destroy(zilog_t *zilog, boolean_t keep_first)
562 {
563         const zil_header_t *zh = zilog->zl_header;
564         lwb_t *lwb;
565         dmu_tx_t *tx;
566         uint64_t txg;
567
568         /*
569          * Wait for any previous destroy to complete.
570          */
571         txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
572
573         zilog->zl_old_header = *zh;             /* debugging aid */
574
575         if (BP_IS_HOLE(&zh->zh_log))
576                 return;
577
578         tx = dmu_tx_create(zilog->zl_os);
579         VERIFY(dmu_tx_assign(tx, TXG_WAIT) == 0);
580         dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
581         txg = dmu_tx_get_txg(tx);
582
583         mutex_enter(&zilog->zl_lock);
584
585         ASSERT3U(zilog->zl_destroy_txg, <, txg);
586         zilog->zl_destroy_txg = txg;
587         zilog->zl_keep_first = keep_first;
588
589         if (!list_is_empty(&zilog->zl_lwb_list)) {
590                 ASSERT(zh->zh_claim_txg == 0);
591                 VERIFY(!keep_first);
592                 while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) {
593                         ASSERT(lwb->lwb_zio == NULL);
594                         if (lwb->lwb_fastwrite)
595                                 metaslab_fastwrite_unmark(zilog->zl_spa,
596                                     &lwb->lwb_blk);
597                         list_remove(&zilog->zl_lwb_list, lwb);
598                         if (lwb->lwb_buf != NULL)
599                                 zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
600                         zio_free_zil(zilog->zl_spa, txg, &lwb->lwb_blk);
601                         kmem_cache_free(zil_lwb_cache, lwb);
602                 }
603         } else if (!keep_first) {
604                 (void) zil_parse(zilog, zil_free_log_block,
605                     zil_free_log_record, tx, zh->zh_claim_txg);
606         }
607         mutex_exit(&zilog->zl_lock);
608
609         dmu_tx_commit(tx);
610 }
611
612 int
613 zil_claim(const char *osname, void *txarg)
614 {
615         dmu_tx_t *tx = txarg;
616         uint64_t first_txg = dmu_tx_get_txg(tx);
617         zilog_t *zilog;
618         zil_header_t *zh;
619         objset_t *os;
620         int error;
621
622         error = dmu_objset_hold(osname, FTAG, &os);
623         if (error) {
624                 cmn_err(CE_WARN, "can't open objset for %s", osname);
625                 return (0);
626         }
627
628         zilog = dmu_objset_zil(os);
629         zh = zil_header_in_syncing_context(zilog);
630
631         if (spa_get_log_state(zilog->zl_spa) == SPA_LOG_CLEAR) {
632                 if (!BP_IS_HOLE(&zh->zh_log))
633                         zio_free_zil(zilog->zl_spa, first_txg, &zh->zh_log);
634                 BP_ZERO(&zh->zh_log);
635                 dsl_dataset_dirty(dmu_objset_ds(os), tx);
636                 dmu_objset_rele(os, FTAG);
637                 return (0);
638         }
639
640         /*
641          * Claim all log blocks if we haven't already done so, and remember
642          * the highest claimed sequence number.  This ensures that if we can
643          * read only part of the log now (e.g. due to a missing device),
644          * but we can read the entire log later, we will not try to replay
645          * or destroy beyond the last block we successfully claimed.
646          */
647         ASSERT3U(zh->zh_claim_txg, <=, first_txg);
648         if (zh->zh_claim_txg == 0 && !BP_IS_HOLE(&zh->zh_log)) {
649                 (void) zil_parse(zilog, zil_claim_log_block,
650                     zil_claim_log_record, tx, first_txg);
651                 zh->zh_claim_txg = first_txg;
652                 zh->zh_claim_blk_seq = zilog->zl_parse_blk_seq;
653                 zh->zh_claim_lr_seq = zilog->zl_parse_lr_seq;
654                 if (zilog->zl_parse_lr_count || zilog->zl_parse_blk_count > 1)
655                         zh->zh_flags |= ZIL_REPLAY_NEEDED;
656                 zh->zh_flags |= ZIL_CLAIM_LR_SEQ_VALID;
657                 dsl_dataset_dirty(dmu_objset_ds(os), tx);
658         }
659
660         ASSERT3U(first_txg, ==, (spa_last_synced_txg(zilog->zl_spa) + 1));
661         dmu_objset_rele(os, FTAG);
662         return (0);
663 }
664
665 /*
666  * Check the log by walking the log chain.
667  * Checksum errors are ok as they indicate the end of the chain.
668  * Any other error (no device or read failure) returns an error.
669  */
670 int
671 zil_check_log_chain(const char *osname, void *tx)
672 {
673         zilog_t *zilog;
674         objset_t *os;
675         blkptr_t *bp;
676         int error;
677
678         ASSERT(tx == NULL);
679
680         error = dmu_objset_hold(osname, FTAG, &os);
681         if (error) {
682                 cmn_err(CE_WARN, "can't open objset for %s", osname);
683                 return (0);
684         }
685
686         zilog = dmu_objset_zil(os);
687         bp = (blkptr_t *)&zilog->zl_header->zh_log;
688
689         /*
690          * Check the first block and determine if it's on a log device
691          * which may have been removed or faulted prior to loading this
692          * pool.  If so, there's no point in checking the rest of the log
693          * as its content should have already been synced to the pool.
694          */
695         if (!BP_IS_HOLE(bp)) {
696                 vdev_t *vd;
697                 boolean_t valid = B_TRUE;
698
699                 spa_config_enter(os->os_spa, SCL_STATE, FTAG, RW_READER);
700                 vd = vdev_lookup_top(os->os_spa, DVA_GET_VDEV(&bp->blk_dva[0]));
701                 if (vd->vdev_islog && vdev_is_dead(vd))
702                         valid = vdev_log_state_valid(vd);
703                 spa_config_exit(os->os_spa, SCL_STATE, FTAG);
704
705                 if (!valid) {
706                         dmu_objset_rele(os, FTAG);
707                         return (0);
708                 }
709         }
710
711         /*
712          * Because tx == NULL, zil_claim_log_block() will not actually claim
713          * any blocks, but just determine whether it is possible to do so.
714          * In addition to checking the log chain, zil_claim_log_block()
715          * will invoke zio_claim() with a done func of spa_claim_notify(),
716          * which will update spa_max_claim_txg.  See spa_load() for details.
717          */
718         error = zil_parse(zilog, zil_claim_log_block, zil_claim_log_record, tx,
719             zilog->zl_header->zh_claim_txg ? -1ULL : spa_first_txg(os->os_spa));
720
721         dmu_objset_rele(os, FTAG);
722
723         return ((error == ECKSUM || error == ENOENT) ? 0 : error);
724 }
725
726 static int
727 zil_vdev_compare(const void *x1, const void *x2)
728 {
729         const uint64_t v1 = ((zil_vdev_node_t *)x1)->zv_vdev;
730         const uint64_t v2 = ((zil_vdev_node_t *)x2)->zv_vdev;
731
732         if (v1 < v2)
733                 return (-1);
734         if (v1 > v2)
735                 return (1);
736
737         return (0);
738 }
739
740 void
741 zil_add_block(zilog_t *zilog, const blkptr_t *bp)
742 {
743         avl_tree_t *t = &zilog->zl_vdev_tree;
744         avl_index_t where;
745         zil_vdev_node_t *zv, zvsearch;
746         int ndvas = BP_GET_NDVAS(bp);
747         int i;
748
749         if (zfs_nocacheflush)
750                 return;
751
752         ASSERT(zilog->zl_writer);
753
754         /*
755          * Even though we're zl_writer, we still need a lock because the
756          * zl_get_data() callbacks may have dmu_sync() done callbacks
757          * that will run concurrently.
758          */
759         mutex_enter(&zilog->zl_vdev_lock);
760         for (i = 0; i < ndvas; i++) {
761                 zvsearch.zv_vdev = DVA_GET_VDEV(&bp->blk_dva[i]);
762                 if (avl_find(t, &zvsearch, &where) == NULL) {
763                         zv = kmem_alloc(sizeof (*zv), KM_PUSHPAGE);
764                         zv->zv_vdev = zvsearch.zv_vdev;
765                         avl_insert(t, zv, where);
766                 }
767         }
768         mutex_exit(&zilog->zl_vdev_lock);
769 }
770
771 static void
772 zil_flush_vdevs(zilog_t *zilog)
773 {
774         spa_t *spa = zilog->zl_spa;
775         avl_tree_t *t = &zilog->zl_vdev_tree;
776         void *cookie = NULL;
777         zil_vdev_node_t *zv;
778         zio_t *zio;
779
780         ASSERT(zilog->zl_writer);
781
782         /*
783          * We don't need zl_vdev_lock here because we're the zl_writer,
784          * and all zl_get_data() callbacks are done.
785          */
786         if (avl_numnodes(t) == 0)
787                 return;
788
789         spa_config_enter(spa, SCL_STATE, FTAG, RW_READER);
790
791         zio = zio_root(spa, NULL, NULL, ZIO_FLAG_CANFAIL);
792
793         while ((zv = avl_destroy_nodes(t, &cookie)) != NULL) {
794                 vdev_t *vd = vdev_lookup_top(spa, zv->zv_vdev);
795                 if (vd != NULL)
796                         zio_flush(zio, vd);
797                 kmem_free(zv, sizeof (*zv));
798         }
799
800         /*
801          * Wait for all the flushes to complete.  Not all devices actually
802          * support the DKIOCFLUSHWRITECACHE ioctl, so it's OK if it fails.
803          */
804         (void) zio_wait(zio);
805
806         spa_config_exit(spa, SCL_STATE, FTAG);
807 }
808
809 /*
810  * Function called when a log block write completes
811  */
812 static void
813 zil_lwb_write_done(zio_t *zio)
814 {
815         lwb_t *lwb = zio->io_private;
816         zilog_t *zilog = lwb->lwb_zilog;
817         dmu_tx_t *tx = lwb->lwb_tx;
818
819         ASSERT(BP_GET_COMPRESS(zio->io_bp) == ZIO_COMPRESS_OFF);
820         ASSERT(BP_GET_TYPE(zio->io_bp) == DMU_OT_INTENT_LOG);
821         ASSERT(BP_GET_LEVEL(zio->io_bp) == 0);
822         ASSERT(BP_GET_BYTEORDER(zio->io_bp) == ZFS_HOST_BYTEORDER);
823         ASSERT(!BP_IS_GANG(zio->io_bp));
824         ASSERT(!BP_IS_HOLE(zio->io_bp));
825         ASSERT(zio->io_bp->blk_fill == 0);
826
827         /*
828          * Ensure the lwb buffer pointer is cleared before releasing
829          * the txg. If we have had an allocation failure and
830          * the txg is waiting to sync then we want want zil_sync()
831          * to remove the lwb so that it's not picked up as the next new
832          * one in zil_commit_writer(). zil_sync() will only remove
833          * the lwb if lwb_buf is null.
834          */
835         zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
836         mutex_enter(&zilog->zl_lock);
837         lwb->lwb_zio = NULL;
838         lwb->lwb_fastwrite = FALSE;
839         lwb->lwb_buf = NULL;
840         lwb->lwb_tx = NULL;
841         mutex_exit(&zilog->zl_lock);
842
843         /*
844          * Now that we've written this log block, we have a stable pointer
845          * to the next block in the chain, so it's OK to let the txg in
846          * which we allocated the next block sync.
847          */
848         dmu_tx_commit(tx);
849 }
850
851 /*
852  * Initialize the io for a log block.
853  */
854 static void
855 zil_lwb_write_init(zilog_t *zilog, lwb_t *lwb)
856 {
857         zbookmark_t zb;
858
859         SET_BOOKMARK(&zb, lwb->lwb_blk.blk_cksum.zc_word[ZIL_ZC_OBJSET],
860             ZB_ZIL_OBJECT, ZB_ZIL_LEVEL,
861             lwb->lwb_blk.blk_cksum.zc_word[ZIL_ZC_SEQ]);
862
863         if (zilog->zl_root_zio == NULL) {
864                 zilog->zl_root_zio = zio_root(zilog->zl_spa, NULL, NULL,
865                     ZIO_FLAG_CANFAIL);
866         }
867
868         /* Lock so zil_sync() doesn't fastwrite_unmark after zio is created */
869         mutex_enter(&zilog->zl_lock);
870         if (lwb->lwb_zio == NULL) {
871                 if (!lwb->lwb_fastwrite) {
872                         metaslab_fastwrite_mark(zilog->zl_spa, &lwb->lwb_blk);
873                         lwb->lwb_fastwrite = 1;
874                 }
875                 lwb->lwb_zio = zio_rewrite(zilog->zl_root_zio, zilog->zl_spa,
876                     0, &lwb->lwb_blk, lwb->lwb_buf, BP_GET_LSIZE(&lwb->lwb_blk),
877                     zil_lwb_write_done, lwb, ZIO_PRIORITY_LOG_WRITE,
878                     ZIO_FLAG_CANFAIL | ZIO_FLAG_DONT_PROPAGATE |
879                     ZIO_FLAG_FASTWRITE, &zb);
880         }
881         mutex_exit(&zilog->zl_lock);
882 }
883
884 /*
885  * Define a limited set of intent log block sizes.
886  * These must be a multiple of 4KB. Note only the amount used (again
887  * aligned to 4KB) actually gets written. However, we can't always just
888  * allocate SPA_MAXBLOCKSIZE as the slog space could be exhausted.
889  */
890 uint64_t zil_block_buckets[] = {
891     4096,               /* non TX_WRITE */
892     8192+4096,          /* data base */
893     32*1024 + 4096,     /* NFS writes */
894     UINT64_MAX
895 };
896
897 /*
898  * Use the slog as long as the logbias is 'latency' and the current commit size
899  * is less than the limit or the total list size is less than 2X the limit.
900  * Limit checking is disabled by setting zil_slog_limit to UINT64_MAX.
901  */
902 unsigned long zil_slog_limit = 1024 * 1024;
903 #define USE_SLOG(zilog) (((zilog)->zl_logbias == ZFS_LOGBIAS_LATENCY) && \
904         (((zilog)->zl_cur_used < zil_slog_limit) || \
905         ((zilog)->zl_itx_list_sz < (zil_slog_limit << 1))))
906
907 /*
908  * Start a log block write and advance to the next log block.
909  * Calls are serialized.
910  */
911 static lwb_t *
912 zil_lwb_write_start(zilog_t *zilog, lwb_t *lwb)
913 {
914         lwb_t *nlwb = NULL;
915         zil_chain_t *zilc;
916         spa_t *spa = zilog->zl_spa;
917         blkptr_t *bp;
918         dmu_tx_t *tx;
919         uint64_t txg;
920         uint64_t zil_blksz, wsz;
921         int i, error;
922         boolean_t use_slog;
923
924         if (BP_GET_CHECKSUM(&lwb->lwb_blk) == ZIO_CHECKSUM_ZILOG2) {
925                 zilc = (zil_chain_t *)lwb->lwb_buf;
926                 bp = &zilc->zc_next_blk;
927         } else {
928                 zilc = (zil_chain_t *)(lwb->lwb_buf + lwb->lwb_sz);
929                 bp = &zilc->zc_next_blk;
930         }
931
932         ASSERT(lwb->lwb_nused <= lwb->lwb_sz);
933
934         /*
935          * Allocate the next block and save its address in this block
936          * before writing it in order to establish the log chain.
937          * Note that if the allocation of nlwb synced before we wrote
938          * the block that points at it (lwb), we'd leak it if we crashed.
939          * Therefore, we don't do dmu_tx_commit() until zil_lwb_write_done().
940          * We dirty the dataset to ensure that zil_sync() will be called
941          * to clean up in the event of allocation failure or I/O failure.
942          */
943         tx = dmu_tx_create(zilog->zl_os);
944         VERIFY(dmu_tx_assign(tx, TXG_WAIT) == 0);
945         dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
946         txg = dmu_tx_get_txg(tx);
947
948         lwb->lwb_tx = tx;
949
950         /*
951          * Log blocks are pre-allocated. Here we select the size of the next
952          * block, based on size used in the last block.
953          * - first find the smallest bucket that will fit the block from a
954          *   limited set of block sizes. This is because it's faster to write
955          *   blocks allocated from the same metaslab as they are adjacent or
956          *   close.
957          * - next find the maximum from the new suggested size and an array of
958          *   previous sizes. This lessens a picket fence effect of wrongly
959          *   guesssing the size if we have a stream of say 2k, 64k, 2k, 64k
960          *   requests.
961          *
962          * Note we only write what is used, but we can't just allocate
963          * the maximum block size because we can exhaust the available
964          * pool log space.
965          */
966         zil_blksz = zilog->zl_cur_used + sizeof (zil_chain_t);
967         for (i = 0; zil_blksz > zil_block_buckets[i]; i++)
968                 continue;
969         zil_blksz = zil_block_buckets[i];
970         if (zil_blksz == UINT64_MAX)
971                 zil_blksz = SPA_MAXBLOCKSIZE;
972         zilog->zl_prev_blks[zilog->zl_prev_rotor] = zil_blksz;
973         for (i = 0; i < ZIL_PREV_BLKS; i++)
974                 zil_blksz = MAX(zil_blksz, zilog->zl_prev_blks[i]);
975         zilog->zl_prev_rotor = (zilog->zl_prev_rotor + 1) & (ZIL_PREV_BLKS - 1);
976
977         BP_ZERO(bp);
978         use_slog = USE_SLOG(zilog);
979         error = zio_alloc_zil(spa, txg, bp, zil_blksz, USE_SLOG(zilog));
980         if (use_slog)
981         {
982                 ZIL_STAT_BUMP(zil_itx_metaslab_slog_count);
983                 ZIL_STAT_INCR(zil_itx_metaslab_slog_bytes, lwb->lwb_nused);
984         }
985         else
986         {
987                 ZIL_STAT_BUMP(zil_itx_metaslab_normal_count);
988                 ZIL_STAT_INCR(zil_itx_metaslab_normal_bytes, lwb->lwb_nused);
989         }
990         if (!error) {
991                 ASSERT3U(bp->blk_birth, ==, txg);
992                 bp->blk_cksum = lwb->lwb_blk.blk_cksum;
993                 bp->blk_cksum.zc_word[ZIL_ZC_SEQ]++;
994
995                 /*
996                  * Allocate a new log write buffer (lwb).
997                  */
998                 nlwb = zil_alloc_lwb(zilog, bp, txg, TRUE);
999
1000                 /* Record the block for later vdev flushing */
1001                 zil_add_block(zilog, &lwb->lwb_blk);
1002         }
1003
1004         if (BP_GET_CHECKSUM(&lwb->lwb_blk) == ZIO_CHECKSUM_ZILOG2) {
1005                 /* For Slim ZIL only write what is used. */
1006                 wsz = P2ROUNDUP_TYPED(lwb->lwb_nused, ZIL_MIN_BLKSZ, uint64_t);
1007                 ASSERT3U(wsz, <=, lwb->lwb_sz);
1008                 zio_shrink(lwb->lwb_zio, wsz);
1009
1010         } else {
1011                 wsz = lwb->lwb_sz;
1012         }
1013
1014         zilc->zc_pad = 0;
1015         zilc->zc_nused = lwb->lwb_nused;
1016         zilc->zc_eck.zec_cksum = lwb->lwb_blk.blk_cksum;
1017
1018         /*
1019          * clear unused data for security
1020          */
1021         bzero(lwb->lwb_buf + lwb->lwb_nused, wsz - lwb->lwb_nused);
1022
1023         zio_nowait(lwb->lwb_zio); /* Kick off the write for the old log block */
1024
1025         /*
1026          * If there was an allocation failure then nlwb will be null which
1027          * forces a txg_wait_synced().
1028          */
1029         return (nlwb);
1030 }
1031
1032 static lwb_t *
1033 zil_lwb_commit(zilog_t *zilog, itx_t *itx, lwb_t *lwb)
1034 {
1035         lr_t *lrc = &itx->itx_lr; /* common log record */
1036         lr_write_t *lrw = (lr_write_t *)lrc;
1037         char *lr_buf;
1038         uint64_t txg = lrc->lrc_txg;
1039         uint64_t reclen = lrc->lrc_reclen;
1040         uint64_t dlen = 0;
1041
1042         if (lwb == NULL)
1043                 return (NULL);
1044
1045         ASSERT(lwb->lwb_buf != NULL);
1046
1047         if (lrc->lrc_txtype == TX_WRITE && itx->itx_wr_state == WR_NEED_COPY)
1048                 dlen = P2ROUNDUP_TYPED(
1049                     lrw->lr_length, sizeof (uint64_t), uint64_t);
1050
1051         zilog->zl_cur_used += (reclen + dlen);
1052
1053         zil_lwb_write_init(zilog, lwb);
1054
1055         /*
1056          * If this record won't fit in the current log block, start a new one.
1057          */
1058         if (lwb->lwb_nused + reclen + dlen > lwb->lwb_sz) {
1059                 lwb = zil_lwb_write_start(zilog, lwb);
1060                 if (lwb == NULL)
1061                         return (NULL);
1062                 zil_lwb_write_init(zilog, lwb);
1063                 ASSERT(LWB_EMPTY(lwb));
1064                 if (lwb->lwb_nused + reclen + dlen > lwb->lwb_sz) {
1065                         txg_wait_synced(zilog->zl_dmu_pool, txg);
1066                         return (lwb);
1067                 }
1068         }
1069
1070         lr_buf = lwb->lwb_buf + lwb->lwb_nused;
1071         bcopy(lrc, lr_buf, reclen);
1072         lrc = (lr_t *)lr_buf;
1073         lrw = (lr_write_t *)lrc;
1074
1075         ZIL_STAT_BUMP(zil_itx_count);
1076
1077         /*
1078          * If it's a write, fetch the data or get its blkptr as appropriate.
1079          */
1080         if (lrc->lrc_txtype == TX_WRITE) {
1081                 if (txg > spa_freeze_txg(zilog->zl_spa))
1082                         txg_wait_synced(zilog->zl_dmu_pool, txg);
1083                 if (itx->itx_wr_state == WR_COPIED) {
1084                         ZIL_STAT_BUMP(zil_itx_copied_count);
1085                         ZIL_STAT_INCR(zil_itx_copied_bytes, lrw->lr_length);
1086                 } else {
1087                         char *dbuf;
1088                         int error;
1089
1090                         if (dlen) {
1091                                 ASSERT(itx->itx_wr_state == WR_NEED_COPY);
1092                                 dbuf = lr_buf + reclen;
1093                                 lrw->lr_common.lrc_reclen += dlen;
1094                                 ZIL_STAT_BUMP(zil_itx_needcopy_count);
1095                                 ZIL_STAT_INCR(zil_itx_needcopy_bytes, lrw->lr_length);
1096                         } else {
1097                                 ASSERT(itx->itx_wr_state == WR_INDIRECT);
1098                                 dbuf = NULL;
1099                                 ZIL_STAT_BUMP(zil_itx_indirect_count);
1100                                 ZIL_STAT_INCR(zil_itx_indirect_bytes, lrw->lr_length);
1101                         }
1102                         error = zilog->zl_get_data(
1103                             itx->itx_private, lrw, dbuf, lwb->lwb_zio);
1104                         if (error == EIO) {
1105                                 txg_wait_synced(zilog->zl_dmu_pool, txg);
1106                                 return (lwb);
1107                         }
1108                         if (error) {
1109                                 ASSERT(error == ENOENT || error == EEXIST ||
1110                                     error == EALREADY);
1111                                 return (lwb);
1112                         }
1113                 }
1114         }
1115
1116         /*
1117          * We're actually making an entry, so update lrc_seq to be the
1118          * log record sequence number.  Note that this is generally not
1119          * equal to the itx sequence number because not all transactions
1120          * are synchronous, and sometimes spa_sync() gets there first.
1121          */
1122         lrc->lrc_seq = ++zilog->zl_lr_seq; /* we are single threaded */
1123         lwb->lwb_nused += reclen + dlen;
1124         lwb->lwb_max_txg = MAX(lwb->lwb_max_txg, txg);
1125         ASSERT3U(lwb->lwb_nused, <=, lwb->lwb_sz);
1126         ASSERT3U(P2PHASE(lwb->lwb_nused, sizeof (uint64_t)), ==, 0);
1127
1128         return (lwb);
1129 }
1130
1131 itx_t *
1132 zil_itx_create(uint64_t txtype, size_t lrsize)
1133 {
1134         itx_t *itx;
1135
1136         lrsize = P2ROUNDUP_TYPED(lrsize, sizeof (uint64_t), size_t);
1137
1138         itx = kmem_alloc(offsetof(itx_t, itx_lr) + lrsize,
1139             KM_PUSHPAGE | KM_NODEBUG);
1140         itx->itx_lr.lrc_txtype = txtype;
1141         itx->itx_lr.lrc_reclen = lrsize;
1142         itx->itx_sod = lrsize; /* if write & WR_NEED_COPY will be increased */
1143         itx->itx_lr.lrc_seq = 0;        /* defensive */
1144         itx->itx_sync = B_TRUE;         /* default is synchronous */
1145
1146         return (itx);
1147 }
1148
1149 void
1150 zil_itx_destroy(itx_t *itx)
1151 {
1152         kmem_free(itx, offsetof(itx_t, itx_lr) + itx->itx_lr.lrc_reclen);
1153 }
1154
1155 /*
1156  * Free up the sync and async itxs. The itxs_t has already been detached
1157  * so no locks are needed.
1158  */
1159 static void
1160 zil_itxg_clean(itxs_t *itxs)
1161 {
1162         itx_t *itx;
1163         list_t *list;
1164         avl_tree_t *t;
1165         void *cookie;
1166         itx_async_node_t *ian;
1167
1168         list = &itxs->i_sync_list;
1169         while ((itx = list_head(list)) != NULL) {
1170                 list_remove(list, itx);
1171                 kmem_free(itx, offsetof(itx_t, itx_lr) +
1172                     itx->itx_lr.lrc_reclen);
1173         }
1174
1175         cookie = NULL;
1176         t = &itxs->i_async_tree;
1177         while ((ian = avl_destroy_nodes(t, &cookie)) != NULL) {
1178                 list = &ian->ia_list;
1179                 while ((itx = list_head(list)) != NULL) {
1180                         list_remove(list, itx);
1181                         kmem_free(itx, offsetof(itx_t, itx_lr) +
1182                             itx->itx_lr.lrc_reclen);
1183                 }
1184                 list_destroy(list);
1185                 kmem_free(ian, sizeof (itx_async_node_t));
1186         }
1187         avl_destroy(t);
1188
1189         kmem_free(itxs, sizeof (itxs_t));
1190 }
1191
1192 static int
1193 zil_aitx_compare(const void *x1, const void *x2)
1194 {
1195         const uint64_t o1 = ((itx_async_node_t *)x1)->ia_foid;
1196         const uint64_t o2 = ((itx_async_node_t *)x2)->ia_foid;
1197
1198         if (o1 < o2)
1199                 return (-1);
1200         if (o1 > o2)
1201                 return (1);
1202
1203         return (0);
1204 }
1205
1206 /*
1207  * Remove all async itx with the given oid.
1208  */
1209 static void
1210 zil_remove_async(zilog_t *zilog, uint64_t oid)
1211 {
1212         uint64_t otxg, txg;
1213         itx_async_node_t *ian;
1214         avl_tree_t *t;
1215         avl_index_t where;
1216         list_t clean_list;
1217         itx_t *itx;
1218
1219         ASSERT(oid != 0);
1220         list_create(&clean_list, sizeof (itx_t), offsetof(itx_t, itx_node));
1221
1222         if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) /* ziltest support */
1223                 otxg = ZILTEST_TXG;
1224         else
1225                 otxg = spa_last_synced_txg(zilog->zl_spa) + 1;
1226
1227         for (txg = otxg; txg < (otxg + TXG_CONCURRENT_STATES); txg++) {
1228                 itxg_t *itxg = &zilog->zl_itxg[txg & TXG_MASK];
1229
1230                 mutex_enter(&itxg->itxg_lock);
1231                 if (itxg->itxg_txg != txg) {
1232                         mutex_exit(&itxg->itxg_lock);
1233                         continue;
1234                 }
1235
1236                 /*
1237                  * Locate the object node and append its list.
1238                  */
1239                 t = &itxg->itxg_itxs->i_async_tree;
1240                 ian = avl_find(t, &oid, &where);
1241                 if (ian != NULL)
1242                         list_move_tail(&clean_list, &ian->ia_list);
1243                 mutex_exit(&itxg->itxg_lock);
1244         }
1245         while ((itx = list_head(&clean_list)) != NULL) {
1246                 list_remove(&clean_list, itx);
1247                 kmem_free(itx, offsetof(itx_t, itx_lr) +
1248                     itx->itx_lr.lrc_reclen);
1249         }
1250         list_destroy(&clean_list);
1251 }
1252
1253 void
1254 zil_itx_assign(zilog_t *zilog, itx_t *itx, dmu_tx_t *tx)
1255 {
1256         uint64_t txg;
1257         itxg_t *itxg;
1258         itxs_t *itxs, *clean = NULL;
1259
1260         /*
1261          * Object ids can be re-instantiated in the next txg so
1262          * remove any async transactions to avoid future leaks.
1263          * This can happen if a fsync occurs on the re-instantiated
1264          * object for a WR_INDIRECT or WR_NEED_COPY write, which gets
1265          * the new file data and flushes a write record for the old object.
1266          */
1267         if ((itx->itx_lr.lrc_txtype & ~TX_CI) == TX_REMOVE)
1268                 zil_remove_async(zilog, itx->itx_oid);
1269
1270         /*
1271          * Ensure the data of a renamed file is committed before the rename.
1272          */
1273         if ((itx->itx_lr.lrc_txtype & ~TX_CI) == TX_RENAME)
1274                 zil_async_to_sync(zilog, itx->itx_oid);
1275
1276         if (spa_freeze_txg(zilog->zl_spa) !=  UINT64_MAX)
1277                 txg = ZILTEST_TXG;
1278         else
1279                 txg = dmu_tx_get_txg(tx);
1280
1281         itxg = &zilog->zl_itxg[txg & TXG_MASK];
1282         mutex_enter(&itxg->itxg_lock);
1283         itxs = itxg->itxg_itxs;
1284         if (itxg->itxg_txg != txg) {
1285                 if (itxs != NULL) {
1286                         /*
1287                          * The zil_clean callback hasn't got around to cleaning
1288                          * this itxg. Save the itxs for release below.
1289                          * This should be rare.
1290                          */
1291                         atomic_add_64(&zilog->zl_itx_list_sz, -itxg->itxg_sod);
1292                         itxg->itxg_sod = 0;
1293                         clean = itxg->itxg_itxs;
1294                 }
1295                 ASSERT(itxg->itxg_sod == 0);
1296                 itxg->itxg_txg = txg;
1297                 itxs = itxg->itxg_itxs = kmem_zalloc(sizeof (itxs_t), KM_PUSHPAGE);
1298
1299                 list_create(&itxs->i_sync_list, sizeof (itx_t),
1300                     offsetof(itx_t, itx_node));
1301                 avl_create(&itxs->i_async_tree, zil_aitx_compare,
1302                     sizeof (itx_async_node_t),
1303                     offsetof(itx_async_node_t, ia_node));
1304         }
1305         if (itx->itx_sync) {
1306                 list_insert_tail(&itxs->i_sync_list, itx);
1307                 atomic_add_64(&zilog->zl_itx_list_sz, itx->itx_sod);
1308                 itxg->itxg_sod += itx->itx_sod;
1309         } else {
1310                 avl_tree_t *t = &itxs->i_async_tree;
1311                 uint64_t foid = ((lr_ooo_t *)&itx->itx_lr)->lr_foid;
1312                 itx_async_node_t *ian;
1313                 avl_index_t where;
1314
1315                 ian = avl_find(t, &foid, &where);
1316                 if (ian == NULL) {
1317                         ian = kmem_alloc(sizeof (itx_async_node_t), KM_PUSHPAGE);
1318                         list_create(&ian->ia_list, sizeof (itx_t),
1319                             offsetof(itx_t, itx_node));
1320                         ian->ia_foid = foid;
1321                         avl_insert(t, ian, where);
1322                 }
1323                 list_insert_tail(&ian->ia_list, itx);
1324         }
1325
1326         itx->itx_lr.lrc_txg = dmu_tx_get_txg(tx);
1327         mutex_exit(&itxg->itxg_lock);
1328
1329         /* Release the old itxs now we've dropped the lock */
1330         if (clean != NULL)
1331                 zil_itxg_clean(clean);
1332 }
1333
1334 /*
1335  * If there are any in-memory intent log transactions which have now been
1336  * synced then start up a taskq to free them.
1337  */
1338 void
1339 zil_clean(zilog_t *zilog, uint64_t synced_txg)
1340 {
1341         itxg_t *itxg = &zilog->zl_itxg[synced_txg & TXG_MASK];
1342         itxs_t *clean_me;
1343
1344         mutex_enter(&itxg->itxg_lock);
1345         if (itxg->itxg_itxs == NULL || itxg->itxg_txg == ZILTEST_TXG) {
1346                 mutex_exit(&itxg->itxg_lock);
1347                 return;
1348         }
1349         ASSERT3U(itxg->itxg_txg, <=, synced_txg);
1350         ASSERT(itxg->itxg_txg != 0);
1351         ASSERT(zilog->zl_clean_taskq != NULL);
1352         atomic_add_64(&zilog->zl_itx_list_sz, -itxg->itxg_sod);
1353         itxg->itxg_sod = 0;
1354         clean_me = itxg->itxg_itxs;
1355         itxg->itxg_itxs = NULL;
1356         itxg->itxg_txg = 0;
1357         mutex_exit(&itxg->itxg_lock);
1358         /*
1359          * Preferably start a task queue to free up the old itxs but
1360          * if taskq_dispatch can't allocate resources to do that then
1361          * free it in-line. This should be rare. Note, using TQ_SLEEP
1362          * created a bad performance problem.
1363          */
1364         if (taskq_dispatch(zilog->zl_clean_taskq,
1365             (void (*)(void *))zil_itxg_clean, clean_me, TQ_NOSLEEP) == 0)
1366                 zil_itxg_clean(clean_me);
1367 }
1368
1369 /*
1370  * Get the list of itxs to commit into zl_itx_commit_list.
1371  */
1372 static void
1373 zil_get_commit_list(zilog_t *zilog)
1374 {
1375         uint64_t otxg, txg;
1376         list_t *commit_list = &zilog->zl_itx_commit_list;
1377         uint64_t push_sod = 0;
1378
1379         if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) /* ziltest support */
1380                 otxg = ZILTEST_TXG;
1381         else
1382                 otxg = spa_last_synced_txg(zilog->zl_spa) + 1;
1383
1384         for (txg = otxg; txg < (otxg + TXG_CONCURRENT_STATES); txg++) {
1385                 itxg_t *itxg = &zilog->zl_itxg[txg & TXG_MASK];
1386
1387                 mutex_enter(&itxg->itxg_lock);
1388                 if (itxg->itxg_txg != txg) {
1389                         mutex_exit(&itxg->itxg_lock);
1390                         continue;
1391                 }
1392
1393                 list_move_tail(commit_list, &itxg->itxg_itxs->i_sync_list);
1394                 push_sod += itxg->itxg_sod;
1395                 itxg->itxg_sod = 0;
1396
1397                 mutex_exit(&itxg->itxg_lock);
1398         }
1399         atomic_add_64(&zilog->zl_itx_list_sz, -push_sod);
1400 }
1401
1402 /*
1403  * Move the async itxs for a specified object to commit into sync lists.
1404  */
1405 static void
1406 zil_async_to_sync(zilog_t *zilog, uint64_t foid)
1407 {
1408         uint64_t otxg, txg;
1409         itx_async_node_t *ian;
1410         avl_tree_t *t;
1411         avl_index_t where;
1412
1413         if (spa_freeze_txg(zilog->zl_spa) != UINT64_MAX) /* ziltest support */
1414                 otxg = ZILTEST_TXG;
1415         else
1416                 otxg = spa_last_synced_txg(zilog->zl_spa) + 1;
1417
1418         for (txg = otxg; txg < (otxg + TXG_CONCURRENT_STATES); txg++) {
1419                 itxg_t *itxg = &zilog->zl_itxg[txg & TXG_MASK];
1420
1421                 mutex_enter(&itxg->itxg_lock);
1422                 if (itxg->itxg_txg != txg) {
1423                         mutex_exit(&itxg->itxg_lock);
1424                         continue;
1425                 }
1426
1427                 /*
1428                  * If a foid is specified then find that node and append its
1429                  * list. Otherwise walk the tree appending all the lists
1430                  * to the sync list. We add to the end rather than the
1431                  * beginning to ensure the create has happened.
1432                  */
1433                 t = &itxg->itxg_itxs->i_async_tree;
1434                 if (foid != 0) {
1435                         ian = avl_find(t, &foid, &where);
1436                         if (ian != NULL) {
1437                                 list_move_tail(&itxg->itxg_itxs->i_sync_list,
1438                                     &ian->ia_list);
1439                         }
1440                 } else {
1441                         void *cookie = NULL;
1442
1443                         while ((ian = avl_destroy_nodes(t, &cookie)) != NULL) {
1444                                 list_move_tail(&itxg->itxg_itxs->i_sync_list,
1445                                     &ian->ia_list);
1446                                 list_destroy(&ian->ia_list);
1447                                 kmem_free(ian, sizeof (itx_async_node_t));
1448                         }
1449                 }
1450                 mutex_exit(&itxg->itxg_lock);
1451         }
1452 }
1453
1454 static void
1455 zil_commit_writer(zilog_t *zilog)
1456 {
1457         uint64_t txg;
1458         itx_t *itx;
1459         lwb_t *lwb;
1460         spa_t *spa = zilog->zl_spa;
1461         int error = 0;
1462
1463         ASSERT(zilog->zl_root_zio == NULL);
1464
1465         mutex_exit(&zilog->zl_lock);
1466
1467         zil_get_commit_list(zilog);
1468
1469         /*
1470          * Return if there's nothing to commit before we dirty the fs by
1471          * calling zil_create().
1472          */
1473         if (list_head(&zilog->zl_itx_commit_list) == NULL) {
1474                 mutex_enter(&zilog->zl_lock);
1475                 return;
1476         }
1477
1478         if (zilog->zl_suspend) {
1479                 lwb = NULL;
1480         } else {
1481                 lwb = list_tail(&zilog->zl_lwb_list);
1482                 if (lwb == NULL)
1483                         lwb = zil_create(zilog);
1484         }
1485
1486         DTRACE_PROBE1(zil__cw1, zilog_t *, zilog);
1487         while ((itx = list_head(&zilog->zl_itx_commit_list))) {
1488                 txg = itx->itx_lr.lrc_txg;
1489                 ASSERT(txg);
1490
1491                 if (txg > spa_last_synced_txg(spa) || txg > spa_freeze_txg(spa))
1492                         lwb = zil_lwb_commit(zilog, itx, lwb);
1493                 list_remove(&zilog->zl_itx_commit_list, itx);
1494                 kmem_free(itx, offsetof(itx_t, itx_lr)
1495                     + itx->itx_lr.lrc_reclen);
1496         }
1497         DTRACE_PROBE1(zil__cw2, zilog_t *, zilog);
1498
1499         /* write the last block out */
1500         if (lwb != NULL && lwb->lwb_zio != NULL)
1501                 lwb = zil_lwb_write_start(zilog, lwb);
1502
1503         zilog->zl_cur_used = 0;
1504
1505         /*
1506          * Wait if necessary for the log blocks to be on stable storage.
1507          */
1508         if (zilog->zl_root_zio) {
1509                 error = zio_wait(zilog->zl_root_zio);
1510                 zilog->zl_root_zio = NULL;
1511                 zil_flush_vdevs(zilog);
1512         }
1513
1514         if (error || lwb == NULL)
1515                 txg_wait_synced(zilog->zl_dmu_pool, 0);
1516
1517         mutex_enter(&zilog->zl_lock);
1518
1519         /*
1520          * Remember the highest committed log sequence number for ztest.
1521          * We only update this value when all the log writes succeeded,
1522          * because ztest wants to ASSERT that it got the whole log chain.
1523          */
1524         if (error == 0 && lwb != NULL)
1525                 zilog->zl_commit_lr_seq = zilog->zl_lr_seq;
1526 }
1527
1528 /*
1529  * Commit zfs transactions to stable storage.
1530  * If foid is 0 push out all transactions, otherwise push only those
1531  * for that object or might reference that object.
1532  *
1533  * itxs are committed in batches. In a heavily stressed zil there will be
1534  * a commit writer thread who is writing out a bunch of itxs to the log
1535  * for a set of committing threads (cthreads) in the same batch as the writer.
1536  * Those cthreads are all waiting on the same cv for that batch.
1537  *
1538  * There will also be a different and growing batch of threads that are
1539  * waiting to commit (qthreads). When the committing batch completes
1540  * a transition occurs such that the cthreads exit and the qthreads become
1541  * cthreads. One of the new cthreads becomes the writer thread for the
1542  * batch. Any new threads arriving become new qthreads.
1543  *
1544  * Only 2 condition variables are needed and there's no transition
1545  * between the two cvs needed. They just flip-flop between qthreads
1546  * and cthreads.
1547  *
1548  * Using this scheme we can efficiently wakeup up only those threads
1549  * that have been committed.
1550  */
1551 void
1552 zil_commit(zilog_t *zilog, uint64_t foid)
1553 {
1554         uint64_t mybatch;
1555
1556         if (zilog->zl_sync == ZFS_SYNC_DISABLED)
1557                 return;
1558
1559         ZIL_STAT_BUMP(zil_commit_count);
1560
1561         /* move the async itxs for the foid to the sync queues */
1562         zil_async_to_sync(zilog, foid);
1563
1564         mutex_enter(&zilog->zl_lock);
1565         mybatch = zilog->zl_next_batch;
1566         while (zilog->zl_writer) {
1567                 cv_wait(&zilog->zl_cv_batch[mybatch & 1], &zilog->zl_lock);
1568                 if (mybatch <= zilog->zl_com_batch) {
1569                         mutex_exit(&zilog->zl_lock);
1570                         return;
1571                 }
1572         }
1573
1574         zilog->zl_next_batch++;
1575         zilog->zl_writer = B_TRUE;
1576         ZIL_STAT_BUMP(zil_commit_writer_count);
1577         zil_commit_writer(zilog);
1578         zilog->zl_com_batch = mybatch;
1579         zilog->zl_writer = B_FALSE;
1580
1581         /* wake up one thread to become the next writer */
1582         cv_signal(&zilog->zl_cv_batch[(mybatch+1) & 1]);
1583
1584         /* wake up all threads waiting for this batch to be committed */
1585         cv_broadcast(&zilog->zl_cv_batch[mybatch & 1]);
1586
1587         mutex_exit(&zilog->zl_lock);
1588 }
1589
1590 /*
1591  * Called in syncing context to free committed log blocks and update log header.
1592  */
1593 void
1594 zil_sync(zilog_t *zilog, dmu_tx_t *tx)
1595 {
1596         zil_header_t *zh = zil_header_in_syncing_context(zilog);
1597         uint64_t txg = dmu_tx_get_txg(tx);
1598         spa_t *spa = zilog->zl_spa;
1599         uint64_t *replayed_seq = &zilog->zl_replayed_seq[txg & TXG_MASK];
1600         lwb_t *lwb;
1601
1602         /*
1603          * We don't zero out zl_destroy_txg, so make sure we don't try
1604          * to destroy it twice.
1605          */
1606         if (spa_sync_pass(spa) != 1)
1607                 return;
1608
1609         mutex_enter(&zilog->zl_lock);
1610
1611         ASSERT(zilog->zl_stop_sync == 0);
1612
1613         if (*replayed_seq != 0) {
1614                 ASSERT(zh->zh_replay_seq < *replayed_seq);
1615                 zh->zh_replay_seq = *replayed_seq;
1616                 *replayed_seq = 0;
1617         }
1618
1619         if (zilog->zl_destroy_txg == txg) {
1620                 blkptr_t blk = zh->zh_log;
1621
1622                 ASSERT(list_head(&zilog->zl_lwb_list) == NULL);
1623
1624                 bzero(zh, sizeof (zil_header_t));
1625                 bzero(zilog->zl_replayed_seq, sizeof (zilog->zl_replayed_seq));
1626
1627                 if (zilog->zl_keep_first) {
1628                         /*
1629                          * If this block was part of log chain that couldn't
1630                          * be claimed because a device was missing during
1631                          * zil_claim(), but that device later returns,
1632                          * then this block could erroneously appear valid.
1633                          * To guard against this, assign a new GUID to the new
1634                          * log chain so it doesn't matter what blk points to.
1635                          */
1636                         zil_init_log_chain(zilog, &blk);
1637                         zh->zh_log = blk;
1638                 }
1639         }
1640
1641         while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) {
1642                 zh->zh_log = lwb->lwb_blk;
1643                 if (lwb->lwb_buf != NULL || lwb->lwb_max_txg > txg)
1644                         break;
1645
1646                 ASSERT(lwb->lwb_zio == NULL);
1647
1648                 list_remove(&zilog->zl_lwb_list, lwb);
1649                 zio_free_zil(spa, txg, &lwb->lwb_blk);
1650                 kmem_cache_free(zil_lwb_cache, lwb);
1651
1652                 /*
1653                  * If we don't have anything left in the lwb list then
1654                  * we've had an allocation failure and we need to zero
1655                  * out the zil_header blkptr so that we don't end
1656                  * up freeing the same block twice.
1657                  */
1658                 if (list_head(&zilog->zl_lwb_list) == NULL)
1659                         BP_ZERO(&zh->zh_log);
1660         }
1661
1662         /*
1663          * Remove fastwrite on any blocks that have been pre-allocated for
1664          * the next commit. This prevents fastwrite counter pollution by
1665          * unused, long-lived LWBs.
1666          */
1667         for (; lwb != NULL; lwb = list_next(&zilog->zl_lwb_list, lwb)) {
1668                 if (lwb->lwb_fastwrite && !lwb->lwb_zio) {
1669                         metaslab_fastwrite_unmark(zilog->zl_spa, &lwb->lwb_blk);
1670                         lwb->lwb_fastwrite = 0;
1671                 }
1672         }
1673
1674         mutex_exit(&zilog->zl_lock);
1675 }
1676
1677 void
1678 zil_init(void)
1679 {
1680         zil_lwb_cache = kmem_cache_create("zil_lwb_cache",
1681             sizeof (struct lwb), 0, NULL, NULL, NULL, NULL, NULL, 0);
1682
1683         zil_ksp = kstat_create("zfs", 0, "zil", "misc",
1684             KSTAT_TYPE_NAMED, sizeof(zil_stats) / sizeof(kstat_named_t),
1685             KSTAT_FLAG_VIRTUAL);
1686
1687         if (zil_ksp != NULL) {
1688                 zil_ksp->ks_data = &zil_stats;
1689                 kstat_install(zil_ksp);
1690         }
1691 }
1692
1693 void
1694 zil_fini(void)
1695 {
1696         kmem_cache_destroy(zil_lwb_cache);
1697
1698         if (zil_ksp != NULL) {
1699                 kstat_delete(zil_ksp);
1700                 zil_ksp = NULL;
1701         }
1702 }
1703
1704 void
1705 zil_set_sync(zilog_t *zilog, uint64_t sync)
1706 {
1707         zilog->zl_sync = sync;
1708 }
1709
1710 void
1711 zil_set_logbias(zilog_t *zilog, uint64_t logbias)
1712 {
1713         zilog->zl_logbias = logbias;
1714 }
1715
1716 zilog_t *
1717 zil_alloc(objset_t *os, zil_header_t *zh_phys)
1718 {
1719         zilog_t *zilog;
1720         int i;
1721
1722         zilog = kmem_zalloc(sizeof (zilog_t), KM_PUSHPAGE);
1723
1724         zilog->zl_header = zh_phys;
1725         zilog->zl_os = os;
1726         zilog->zl_spa = dmu_objset_spa(os);
1727         zilog->zl_dmu_pool = dmu_objset_pool(os);
1728         zilog->zl_destroy_txg = TXG_INITIAL - 1;
1729         zilog->zl_logbias = dmu_objset_logbias(os);
1730         zilog->zl_sync = dmu_objset_syncprop(os);
1731         zilog->zl_next_batch = 1;
1732
1733         mutex_init(&zilog->zl_lock, NULL, MUTEX_DEFAULT, NULL);
1734
1735         for (i = 0; i < TXG_SIZE; i++) {
1736                 mutex_init(&zilog->zl_itxg[i].itxg_lock, NULL,
1737                     MUTEX_DEFAULT, NULL);
1738         }
1739
1740         list_create(&zilog->zl_lwb_list, sizeof (lwb_t),
1741             offsetof(lwb_t, lwb_node));
1742
1743         list_create(&zilog->zl_itx_commit_list, sizeof (itx_t),
1744             offsetof(itx_t, itx_node));
1745
1746         mutex_init(&zilog->zl_vdev_lock, NULL, MUTEX_DEFAULT, NULL);
1747
1748         avl_create(&zilog->zl_vdev_tree, zil_vdev_compare,
1749             sizeof (zil_vdev_node_t), offsetof(zil_vdev_node_t, zv_node));
1750
1751         cv_init(&zilog->zl_cv_writer, NULL, CV_DEFAULT, NULL);
1752         cv_init(&zilog->zl_cv_suspend, NULL, CV_DEFAULT, NULL);
1753         cv_init(&zilog->zl_cv_batch[0], NULL, CV_DEFAULT, NULL);
1754         cv_init(&zilog->zl_cv_batch[1], NULL, CV_DEFAULT, NULL);
1755
1756         return (zilog);
1757 }
1758
1759 void
1760 zil_free(zilog_t *zilog)
1761 {
1762         int i;
1763
1764         zilog->zl_stop_sync = 1;
1765
1766         ASSERT(list_is_empty(&zilog->zl_lwb_list));
1767         list_destroy(&zilog->zl_lwb_list);
1768
1769         avl_destroy(&zilog->zl_vdev_tree);
1770         mutex_destroy(&zilog->zl_vdev_lock);
1771
1772         ASSERT(list_is_empty(&zilog->zl_itx_commit_list));
1773         list_destroy(&zilog->zl_itx_commit_list);
1774
1775         for (i = 0; i < TXG_SIZE; i++) {
1776                 /*
1777                  * It's possible for an itx to be generated that doesn't dirty
1778                  * a txg (e.g. ztest TX_TRUNCATE). So there's no zil_clean()
1779                  * callback to remove the entry. We remove those here.
1780                  *
1781                  * Also free up the ziltest itxs.
1782                  */
1783                 if (zilog->zl_itxg[i].itxg_itxs)
1784                         zil_itxg_clean(zilog->zl_itxg[i].itxg_itxs);
1785                 mutex_destroy(&zilog->zl_itxg[i].itxg_lock);
1786         }
1787
1788         mutex_destroy(&zilog->zl_lock);
1789
1790         cv_destroy(&zilog->zl_cv_writer);
1791         cv_destroy(&zilog->zl_cv_suspend);
1792         cv_destroy(&zilog->zl_cv_batch[0]);
1793         cv_destroy(&zilog->zl_cv_batch[1]);
1794
1795         kmem_free(zilog, sizeof (zilog_t));
1796 }
1797
1798 /*
1799  * Open an intent log.
1800  */
1801 zilog_t *
1802 zil_open(objset_t *os, zil_get_data_t *get_data)
1803 {
1804         zilog_t *zilog = dmu_objset_zil(os);
1805
1806         ASSERT(zilog->zl_clean_taskq == NULL);
1807         ASSERT(zilog->zl_get_data == NULL);
1808         ASSERT(list_is_empty(&zilog->zl_lwb_list));
1809
1810         zilog->zl_get_data = get_data;
1811         zilog->zl_clean_taskq = taskq_create("zil_clean", 1, minclsyspri,
1812             2, 2, TASKQ_PREPOPULATE);
1813
1814         return (zilog);
1815 }
1816
1817 /*
1818  * Close an intent log.
1819  */
1820 void
1821 zil_close(zilog_t *zilog)
1822 {
1823         lwb_t *lwb;
1824         uint64_t txg = 0;
1825
1826         zil_commit(zilog, 0); /* commit all itx */
1827
1828         /*
1829          * The lwb_max_txg for the stubby lwb will reflect the last activity
1830          * for the zil.  After a txg_wait_synced() on the txg we know all the
1831          * callbacks have occurred that may clean the zil.  Only then can we
1832          * destroy the zl_clean_taskq.
1833          */
1834         mutex_enter(&zilog->zl_lock);
1835         lwb = list_tail(&zilog->zl_lwb_list);
1836         if (lwb != NULL)
1837                 txg = lwb->lwb_max_txg;
1838         mutex_exit(&zilog->zl_lock);
1839         if (txg)
1840                 txg_wait_synced(zilog->zl_dmu_pool, txg);
1841
1842         taskq_destroy(zilog->zl_clean_taskq);
1843         zilog->zl_clean_taskq = NULL;
1844         zilog->zl_get_data = NULL;
1845
1846         /*
1847          * We should have only one LWB left on the list; remove it now.
1848          */
1849         mutex_enter(&zilog->zl_lock);
1850         lwb = list_head(&zilog->zl_lwb_list);
1851         if (lwb != NULL) {
1852                 ASSERT(lwb == list_tail(&zilog->zl_lwb_list));
1853                 ASSERT(lwb->lwb_zio == NULL);
1854                 if (lwb->lwb_fastwrite)
1855                         metaslab_fastwrite_unmark(zilog->zl_spa, &lwb->lwb_blk);
1856                 list_remove(&zilog->zl_lwb_list, lwb);
1857                 zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
1858                 kmem_cache_free(zil_lwb_cache, lwb);
1859         }
1860         mutex_exit(&zilog->zl_lock);
1861 }
1862
1863 /*
1864  * Suspend an intent log.  While in suspended mode, we still honor
1865  * synchronous semantics, but we rely on txg_wait_synced() to do it.
1866  * We suspend the log briefly when taking a snapshot so that the snapshot
1867  * contains all the data it's supposed to, and has an empty intent log.
1868  */
1869 int
1870 zil_suspend(zilog_t *zilog)
1871 {
1872         const zil_header_t *zh = zilog->zl_header;
1873
1874         mutex_enter(&zilog->zl_lock);
1875         if (zh->zh_flags & ZIL_REPLAY_NEEDED) {         /* unplayed log */
1876                 mutex_exit(&zilog->zl_lock);
1877                 return (EBUSY);
1878         }
1879         if (zilog->zl_suspend++ != 0) {
1880                 /*
1881                  * Someone else already began a suspend.
1882                  * Just wait for them to finish.
1883                  */
1884                 while (zilog->zl_suspending)
1885                         cv_wait(&zilog->zl_cv_suspend, &zilog->zl_lock);
1886                 mutex_exit(&zilog->zl_lock);
1887                 return (0);
1888         }
1889         zilog->zl_suspending = B_TRUE;
1890         mutex_exit(&zilog->zl_lock);
1891
1892         zil_commit(zilog, 0);
1893
1894         zil_destroy(zilog, B_FALSE);
1895
1896         mutex_enter(&zilog->zl_lock);
1897         zilog->zl_suspending = B_FALSE;
1898         cv_broadcast(&zilog->zl_cv_suspend);
1899         mutex_exit(&zilog->zl_lock);
1900
1901         return (0);
1902 }
1903
1904 void
1905 zil_resume(zilog_t *zilog)
1906 {
1907         mutex_enter(&zilog->zl_lock);
1908         ASSERT(zilog->zl_suspend != 0);
1909         zilog->zl_suspend--;
1910         mutex_exit(&zilog->zl_lock);
1911 }
1912
1913 typedef struct zil_replay_arg {
1914         zil_replay_func_t **zr_replay;
1915         void            *zr_arg;
1916         boolean_t       zr_byteswap;
1917         char            *zr_lr;
1918 } zil_replay_arg_t;
1919
1920 static int
1921 zil_replay_error(zilog_t *zilog, lr_t *lr, int error)
1922 {
1923         char name[MAXNAMELEN];
1924
1925         zilog->zl_replaying_seq--;      /* didn't actually replay this one */
1926
1927         dmu_objset_name(zilog->zl_os, name);
1928
1929         cmn_err(CE_WARN, "ZFS replay transaction error %d, "
1930             "dataset %s, seq 0x%llx, txtype %llu %s\n", error, name,
1931             (u_longlong_t)lr->lrc_seq,
1932             (u_longlong_t)(lr->lrc_txtype & ~TX_CI),
1933             (lr->lrc_txtype & TX_CI) ? "CI" : "");
1934
1935         return (error);
1936 }
1937
1938 static int
1939 zil_replay_log_record(zilog_t *zilog, lr_t *lr, void *zra, uint64_t claim_txg)
1940 {
1941         zil_replay_arg_t *zr = zra;
1942         const zil_header_t *zh = zilog->zl_header;
1943         uint64_t reclen = lr->lrc_reclen;
1944         uint64_t txtype = lr->lrc_txtype;
1945         int error = 0;
1946
1947         zilog->zl_replaying_seq = lr->lrc_seq;
1948
1949         if (lr->lrc_seq <= zh->zh_replay_seq)   /* already replayed */
1950                 return (0);
1951
1952         if (lr->lrc_txg < claim_txg)            /* already committed */
1953                 return (0);
1954
1955         /* Strip case-insensitive bit, still present in log record */
1956         txtype &= ~TX_CI;
1957
1958         if (txtype == 0 || txtype >= TX_MAX_TYPE)
1959                 return (zil_replay_error(zilog, lr, EINVAL));
1960
1961         /*
1962          * If this record type can be logged out of order, the object
1963          * (lr_foid) may no longer exist.  That's legitimate, not an error.
1964          */
1965         if (TX_OOO(txtype)) {
1966                 error = dmu_object_info(zilog->zl_os,
1967                     ((lr_ooo_t *)lr)->lr_foid, NULL);
1968                 if (error == ENOENT || error == EEXIST)
1969                         return (0);
1970         }
1971
1972         /*
1973          * Make a copy of the data so we can revise and extend it.
1974          */
1975         bcopy(lr, zr->zr_lr, reclen);
1976
1977         /*
1978          * If this is a TX_WRITE with a blkptr, suck in the data.
1979          */
1980         if (txtype == TX_WRITE && reclen == sizeof (lr_write_t)) {
1981                 error = zil_read_log_data(zilog, (lr_write_t *)lr,
1982                     zr->zr_lr + reclen);
1983                 if (error)
1984                         return (zil_replay_error(zilog, lr, error));
1985         }
1986
1987         /*
1988          * The log block containing this lr may have been byteswapped
1989          * so that we can easily examine common fields like lrc_txtype.
1990          * However, the log is a mix of different record types, and only the
1991          * replay vectors know how to byteswap their records.  Therefore, if
1992          * the lr was byteswapped, undo it before invoking the replay vector.
1993          */
1994         if (zr->zr_byteswap)
1995                 byteswap_uint64_array(zr->zr_lr, reclen);
1996
1997         /*
1998          * We must now do two things atomically: replay this log record,
1999          * and update the log header sequence number to reflect the fact that
2000          * we did so. At the end of each replay function the sequence number
2001          * is updated if we are in replay mode.
2002          */
2003         error = zr->zr_replay[txtype](zr->zr_arg, zr->zr_lr, zr->zr_byteswap);
2004         if (error) {
2005                 /*
2006                  * The DMU's dnode layer doesn't see removes until the txg
2007                  * commits, so a subsequent claim can spuriously fail with
2008                  * EEXIST. So if we receive any error we try syncing out
2009                  * any removes then retry the transaction.  Note that we
2010                  * specify B_FALSE for byteswap now, so we don't do it twice.
2011                  */
2012                 txg_wait_synced(spa_get_dsl(zilog->zl_spa), 0);
2013                 error = zr->zr_replay[txtype](zr->zr_arg, zr->zr_lr, B_FALSE);
2014                 if (error)
2015                         return (zil_replay_error(zilog, lr, error));
2016         }
2017         return (0);
2018 }
2019
2020 /* ARGSUSED */
2021 static int
2022 zil_incr_blks(zilog_t *zilog, blkptr_t *bp, void *arg, uint64_t claim_txg)
2023 {
2024         zilog->zl_replay_blks++;
2025
2026         return (0);
2027 }
2028
2029 /*
2030  * If this dataset has a non-empty intent log, replay it and destroy it.
2031  */
2032 void
2033 zil_replay(objset_t *os, void *arg, zil_replay_func_t *replay_func[TX_MAX_TYPE])
2034 {
2035         zilog_t *zilog = dmu_objset_zil(os);
2036         const zil_header_t *zh = zilog->zl_header;
2037         zil_replay_arg_t zr;
2038
2039         if ((zh->zh_flags & ZIL_REPLAY_NEEDED) == 0) {
2040                 zil_destroy(zilog, B_TRUE);
2041                 return;
2042         }
2043
2044         zr.zr_replay = replay_func;
2045         zr.zr_arg = arg;
2046         zr.zr_byteswap = BP_SHOULD_BYTESWAP(&zh->zh_log);
2047         zr.zr_lr = vmem_alloc(2 * SPA_MAXBLOCKSIZE, KM_PUSHPAGE);
2048
2049         /*
2050          * Wait for in-progress removes to sync before starting replay.
2051          */
2052         txg_wait_synced(zilog->zl_dmu_pool, 0);
2053
2054         zilog->zl_replay = B_TRUE;
2055         zilog->zl_replay_time = ddi_get_lbolt();
2056         ASSERT(zilog->zl_replay_blks == 0);
2057         (void) zil_parse(zilog, zil_incr_blks, zil_replay_log_record, &zr,
2058             zh->zh_claim_txg);
2059         vmem_free(zr.zr_lr, 2 * SPA_MAXBLOCKSIZE);
2060
2061         zil_destroy(zilog, B_FALSE);
2062         txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
2063         zilog->zl_replay = B_FALSE;
2064 }
2065
2066 boolean_t
2067 zil_replaying(zilog_t *zilog, dmu_tx_t *tx)
2068 {
2069         if (zilog->zl_sync == ZFS_SYNC_DISABLED)
2070                 return (B_TRUE);
2071
2072         if (zilog->zl_replay) {
2073                 dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
2074                 zilog->zl_replayed_seq[dmu_tx_get_txg(tx) & TXG_MASK] =
2075                     zilog->zl_replaying_seq;
2076                 return (B_TRUE);
2077         }
2078
2079         return (B_FALSE);
2080 }
2081
2082 /* ARGSUSED */
2083 int
2084 zil_vdev_offline(const char *osname, void *arg)
2085 {
2086         objset_t *os;
2087         zilog_t *zilog;
2088         int error;
2089
2090         error = dmu_objset_hold(osname, FTAG, &os);
2091         if (error)
2092                 return (error);
2093
2094         zilog = dmu_objset_zil(os);
2095         if (zil_suspend(zilog) != 0)
2096                 error = EEXIST;
2097         else
2098                 zil_resume(zilog);
2099         dmu_objset_rele(os, FTAG);
2100         return (error);
2101 }
2102
2103 #if defined(_KERNEL) && defined(HAVE_SPL)
2104 module_param(zil_replay_disable, int, 0644);
2105 MODULE_PARM_DESC(zil_replay_disable, "Disable intent logging replay");
2106
2107 module_param(zfs_nocacheflush, int, 0644);
2108 MODULE_PARM_DESC(zfs_nocacheflush, "Disable cache flushes");
2109
2110 module_param(zil_slog_limit, ulong, 0644);
2111 MODULE_PARM_DESC(zil_slog_limit, "Max commit bytes to separate log device");
2112 #endif