xref: /titanic_50/usr/src/uts/common/fs/zfs/zil.c (revision 8cdd6a74847b5ae6ed26727528bdf6b139cf7552)
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 2009 Sun Microsystems, Inc.  All rights reserved.
23  * Use is subject to license terms.
24  */
25 
26 #include <sys/zfs_context.h>
27 #include <sys/spa.h>
28 #include <sys/spa_impl.h>
29 #include <sys/dmu.h>
30 #include <sys/zap.h>
31 #include <sys/arc.h>
32 #include <sys/stat.h>
33 #include <sys/resource.h>
34 #include <sys/zil.h>
35 #include <sys/zil_impl.h>
36 #include <sys/dsl_dataset.h>
37 #include <sys/vdev.h>
38 #include <sys/dmu_tx.h>
39 
40 /*
41  * The zfs intent log (ZIL) saves transaction records of system calls
42  * that change the file system in memory with enough information
43  * to be able to replay them. These are stored in memory until
44  * either the DMU transaction group (txg) commits them to the stable pool
45  * and they can be discarded, or they are flushed to the stable log
46  * (also in the pool) due to a fsync, O_DSYNC or other synchronous
47  * requirement. In the event of a panic or power fail then those log
48  * records (transactions) are replayed.
49  *
50  * There is one ZIL per file system. Its on-disk (pool) format consists
51  * of 3 parts:
52  *
53  * 	- ZIL header
54  * 	- ZIL blocks
55  * 	- ZIL records
56  *
57  * A log record holds a system call transaction. Log blocks can
58  * hold many log records and the blocks are chained together.
59  * Each ZIL block contains a block pointer (blkptr_t) to the next
60  * ZIL block in the chain. The ZIL header points to the first
61  * block in the chain. Note there is not a fixed place in the pool
62  * to hold blocks. They are dynamically allocated and freed as
63  * needed from the blocks available. Figure X shows the ZIL structure:
64  */
65 
66 /*
67  * This global ZIL switch affects all pools
68  */
69 int zil_disable = 0;	/* disable intent logging */
70 
71 /*
72  * Tunable parameter for debugging or performance analysis.  Setting
73  * zfs_nocacheflush will cause corruption on power loss if a volatile
74  * out-of-order write cache is enabled.
75  */
76 boolean_t zfs_nocacheflush = B_FALSE;
77 
78 static kmem_cache_t *zil_lwb_cache;
79 
80 static boolean_t zil_empty(zilog_t *zilog);
81 
82 static int
83 zil_dva_compare(const void *x1, const void *x2)
84 {
85 	const dva_t *dva1 = x1;
86 	const dva_t *dva2 = x2;
87 
88 	if (DVA_GET_VDEV(dva1) < DVA_GET_VDEV(dva2))
89 		return (-1);
90 	if (DVA_GET_VDEV(dva1) > DVA_GET_VDEV(dva2))
91 		return (1);
92 
93 	if (DVA_GET_OFFSET(dva1) < DVA_GET_OFFSET(dva2))
94 		return (-1);
95 	if (DVA_GET_OFFSET(dva1) > DVA_GET_OFFSET(dva2))
96 		return (1);
97 
98 	return (0);
99 }
100 
101 static void
102 zil_dva_tree_init(avl_tree_t *t)
103 {
104 	avl_create(t, zil_dva_compare, sizeof (zil_dva_node_t),
105 	    offsetof(zil_dva_node_t, zn_node));
106 }
107 
108 static void
109 zil_dva_tree_fini(avl_tree_t *t)
110 {
111 	zil_dva_node_t *zn;
112 	void *cookie = NULL;
113 
114 	while ((zn = avl_destroy_nodes(t, &cookie)) != NULL)
115 		kmem_free(zn, sizeof (zil_dva_node_t));
116 
117 	avl_destroy(t);
118 }
119 
120 static int
121 zil_dva_tree_add(avl_tree_t *t, dva_t *dva)
122 {
123 	zil_dva_node_t *zn;
124 	avl_index_t where;
125 
126 	if (avl_find(t, dva, &where) != NULL)
127 		return (EEXIST);
128 
129 	zn = kmem_alloc(sizeof (zil_dva_node_t), KM_SLEEP);
130 	zn->zn_dva = *dva;
131 	avl_insert(t, zn, where);
132 
133 	return (0);
134 }
135 
136 static zil_header_t *
137 zil_header_in_syncing_context(zilog_t *zilog)
138 {
139 	return ((zil_header_t *)zilog->zl_header);
140 }
141 
142 static void
143 zil_init_log_chain(zilog_t *zilog, blkptr_t *bp)
144 {
145 	zio_cksum_t *zc = &bp->blk_cksum;
146 
147 	zc->zc_word[ZIL_ZC_GUID_0] = spa_get_random(-1ULL);
148 	zc->zc_word[ZIL_ZC_GUID_1] = spa_get_random(-1ULL);
149 	zc->zc_word[ZIL_ZC_OBJSET] = dmu_objset_id(zilog->zl_os);
150 	zc->zc_word[ZIL_ZC_SEQ] = 1ULL;
151 }
152 
153 /*
154  * Read a log block, make sure it's valid, and byteswap it if necessary.
155  */
156 static int
157 zil_read_log_block(zilog_t *zilog, const blkptr_t *bp, arc_buf_t **abufpp)
158 {
159 	blkptr_t blk = *bp;
160 	zbookmark_t zb;
161 	uint32_t aflags = ARC_WAIT;
162 	int error;
163 
164 	zb.zb_objset = bp->blk_cksum.zc_word[ZIL_ZC_OBJSET];
165 	zb.zb_object = 0;
166 	zb.zb_level = -1;
167 	zb.zb_blkid = bp->blk_cksum.zc_word[ZIL_ZC_SEQ];
168 
169 	*abufpp = NULL;
170 
171 	/*
172 	 * We shouldn't be doing any scrubbing while we're doing log
173 	 * replay, it's OK to not lock.
174 	 */
175 	error = arc_read_nolock(NULL, zilog->zl_spa, &blk,
176 	    arc_getbuf_func, abufpp, ZIO_PRIORITY_SYNC_READ, ZIO_FLAG_CANFAIL |
177 	    ZIO_FLAG_SPECULATIVE | ZIO_FLAG_SCRUB, &aflags, &zb);
178 
179 	if (error == 0) {
180 		char *data = (*abufpp)->b_data;
181 		uint64_t blksz = BP_GET_LSIZE(bp);
182 		zil_trailer_t *ztp = (zil_trailer_t *)(data + blksz) - 1;
183 		zio_cksum_t cksum = bp->blk_cksum;
184 
185 		/*
186 		 * Validate the checksummed log block.
187 		 *
188 		 * Sequence numbers should be... sequential.  The checksum
189 		 * verifier for the next block should be bp's checksum plus 1.
190 		 *
191 		 * Also check the log chain linkage and size used.
192 		 */
193 		cksum.zc_word[ZIL_ZC_SEQ]++;
194 
195 		if (bcmp(&cksum, &ztp->zit_next_blk.blk_cksum,
196 		    sizeof (cksum)) || BP_IS_HOLE(&ztp->zit_next_blk) ||
197 		    (ztp->zit_nused > (blksz - sizeof (zil_trailer_t)))) {
198 			error = ECKSUM;
199 		}
200 
201 		if (error) {
202 			VERIFY(arc_buf_remove_ref(*abufpp, abufpp) == 1);
203 			*abufpp = NULL;
204 		}
205 	}
206 
207 	dprintf("error %d on %llu:%llu\n", error, zb.zb_objset, zb.zb_blkid);
208 
209 	return (error);
210 }
211 
212 /*
213  * Parse the intent log, and call parse_func for each valid record within.
214  * Return the highest sequence number.
215  */
216 uint64_t
217 zil_parse(zilog_t *zilog, zil_parse_blk_func_t *parse_blk_func,
218     zil_parse_lr_func_t *parse_lr_func, void *arg, uint64_t txg)
219 {
220 	const zil_header_t *zh = zilog->zl_header;
221 	uint64_t claim_seq = zh->zh_claim_seq;
222 	uint64_t seq = 0;
223 	uint64_t max_seq = 0;
224 	blkptr_t blk = zh->zh_log;
225 	arc_buf_t *abuf;
226 	char *lrbuf, *lrp;
227 	zil_trailer_t *ztp;
228 	int reclen, error;
229 
230 	if (BP_IS_HOLE(&blk))
231 		return (max_seq);
232 
233 	/*
234 	 * Starting at the block pointed to by zh_log we read the log chain.
235 	 * For each block in the chain we strongly check that block to
236 	 * ensure its validity.  We stop when an invalid block is found.
237 	 * For each block pointer in the chain we call parse_blk_func().
238 	 * For each record in each valid block we call parse_lr_func().
239 	 * If the log has been claimed, stop if we encounter a sequence
240 	 * number greater than the highest claimed sequence number.
241 	 */
242 	zil_dva_tree_init(&zilog->zl_dva_tree);
243 	for (;;) {
244 		seq = blk.blk_cksum.zc_word[ZIL_ZC_SEQ];
245 
246 		if (claim_seq != 0 && seq > claim_seq)
247 			break;
248 
249 		ASSERT(max_seq < seq);
250 		max_seq = seq;
251 
252 		error = zil_read_log_block(zilog, &blk, &abuf);
253 
254 		if (parse_blk_func != NULL)
255 			parse_blk_func(zilog, &blk, arg, txg);
256 
257 		if (error)
258 			break;
259 
260 		lrbuf = abuf->b_data;
261 		ztp = (zil_trailer_t *)(lrbuf + BP_GET_LSIZE(&blk)) - 1;
262 		blk = ztp->zit_next_blk;
263 
264 		if (parse_lr_func == NULL) {
265 			VERIFY(arc_buf_remove_ref(abuf, &abuf) == 1);
266 			continue;
267 		}
268 
269 		for (lrp = lrbuf; lrp < lrbuf + ztp->zit_nused; lrp += reclen) {
270 			lr_t *lr = (lr_t *)lrp;
271 			reclen = lr->lrc_reclen;
272 			ASSERT3U(reclen, >=, sizeof (lr_t));
273 			parse_lr_func(zilog, lr, arg, txg);
274 		}
275 		VERIFY(arc_buf_remove_ref(abuf, &abuf) == 1);
276 	}
277 	zil_dva_tree_fini(&zilog->zl_dva_tree);
278 
279 	return (max_seq);
280 }
281 
282 /* ARGSUSED */
283 static void
284 zil_claim_log_block(zilog_t *zilog, blkptr_t *bp, void *tx, uint64_t first_txg)
285 {
286 	spa_t *spa = zilog->zl_spa;
287 	int err;
288 
289 	/*
290 	 * Claim log block if not already committed and not already claimed.
291 	 */
292 	if (bp->blk_birth >= first_txg &&
293 	    zil_dva_tree_add(&zilog->zl_dva_tree, BP_IDENTITY(bp)) == 0) {
294 		err = zio_wait(zio_claim(NULL, spa, first_txg, bp, NULL, NULL,
295 		    ZIO_FLAG_MUSTSUCCEED));
296 		ASSERT(err == 0);
297 	}
298 }
299 
300 static void
301 zil_claim_log_record(zilog_t *zilog, lr_t *lrc, void *tx, uint64_t first_txg)
302 {
303 	if (lrc->lrc_txtype == TX_WRITE) {
304 		lr_write_t *lr = (lr_write_t *)lrc;
305 		zil_claim_log_block(zilog, &lr->lr_blkptr, tx, first_txg);
306 	}
307 }
308 
309 /* ARGSUSED */
310 static void
311 zil_free_log_block(zilog_t *zilog, blkptr_t *bp, void *tx, uint64_t claim_txg)
312 {
313 	zio_free_blk(zilog->zl_spa, bp, dmu_tx_get_txg(tx));
314 }
315 
316 static void
317 zil_free_log_record(zilog_t *zilog, lr_t *lrc, void *tx, uint64_t claim_txg)
318 {
319 	/*
320 	 * If we previously claimed it, we need to free it.
321 	 */
322 	if (claim_txg != 0 && lrc->lrc_txtype == TX_WRITE) {
323 		lr_write_t *lr = (lr_write_t *)lrc;
324 		blkptr_t *bp = &lr->lr_blkptr;
325 		if (bp->blk_birth >= claim_txg &&
326 		    !zil_dva_tree_add(&zilog->zl_dva_tree, BP_IDENTITY(bp))) {
327 			(void) arc_free(NULL, zilog->zl_spa,
328 			    dmu_tx_get_txg(tx), bp, NULL, NULL, ARC_WAIT);
329 		}
330 	}
331 }
332 
333 /*
334  * Create an on-disk intent log.
335  */
336 static void
337 zil_create(zilog_t *zilog)
338 {
339 	const zil_header_t *zh = zilog->zl_header;
340 	lwb_t *lwb;
341 	uint64_t txg = 0;
342 	dmu_tx_t *tx = NULL;
343 	blkptr_t blk;
344 	int error = 0;
345 
346 	/*
347 	 * Wait for any previous destroy to complete.
348 	 */
349 	txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
350 
351 	ASSERT(zh->zh_claim_txg == 0);
352 	ASSERT(zh->zh_replay_seq == 0);
353 
354 	blk = zh->zh_log;
355 
356 	/*
357 	 * If we don't already have an initial log block or we have one
358 	 * but it's the wrong endianness then allocate one.
359 	 */
360 	if (BP_IS_HOLE(&blk) || BP_SHOULD_BYTESWAP(&blk)) {
361 		tx = dmu_tx_create(zilog->zl_os);
362 		(void) dmu_tx_assign(tx, TXG_WAIT);
363 		dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
364 		txg = dmu_tx_get_txg(tx);
365 
366 		if (!BP_IS_HOLE(&blk)) {
367 			zio_free_blk(zilog->zl_spa, &blk, txg);
368 			BP_ZERO(&blk);
369 		}
370 
371 		error = zio_alloc_blk(zilog->zl_spa, ZIL_MIN_BLKSZ, &blk,
372 		    NULL, txg, zilog->zl_logbias == ZFS_LOGBIAS_LATENCY);
373 
374 		if (error == 0)
375 			zil_init_log_chain(zilog, &blk);
376 	}
377 
378 	/*
379 	 * Allocate a log write buffer (lwb) for the first log block.
380 	 */
381 	if (error == 0) {
382 		lwb = kmem_cache_alloc(zil_lwb_cache, KM_SLEEP);
383 		lwb->lwb_zilog = zilog;
384 		lwb->lwb_blk = blk;
385 		lwb->lwb_nused = 0;
386 		lwb->lwb_sz = BP_GET_LSIZE(&lwb->lwb_blk);
387 		lwb->lwb_buf = zio_buf_alloc(lwb->lwb_sz);
388 		lwb->lwb_max_txg = txg;
389 		lwb->lwb_zio = NULL;
390 
391 		mutex_enter(&zilog->zl_lock);
392 		list_insert_tail(&zilog->zl_lwb_list, lwb);
393 		mutex_exit(&zilog->zl_lock);
394 	}
395 
396 	/*
397 	 * If we just allocated the first log block, commit our transaction
398 	 * and wait for zil_sync() to stuff the block poiner into zh_log.
399 	 * (zh is part of the MOS, so we cannot modify it in open context.)
400 	 */
401 	if (tx != NULL) {
402 		dmu_tx_commit(tx);
403 		txg_wait_synced(zilog->zl_dmu_pool, txg);
404 	}
405 
406 	ASSERT(bcmp(&blk, &zh->zh_log, sizeof (blk)) == 0);
407 }
408 
409 /*
410  * In one tx, free all log blocks and clear the log header.
411  * If keep_first is set, then we're replaying a log with no content.
412  * We want to keep the first block, however, so that the first
413  * synchronous transaction doesn't require a txg_wait_synced()
414  * in zil_create().  We don't need to txg_wait_synced() here either
415  * when keep_first is set, because both zil_create() and zil_destroy()
416  * will wait for any in-progress destroys to complete.
417  */
418 void
419 zil_destroy(zilog_t *zilog, boolean_t keep_first)
420 {
421 	const zil_header_t *zh = zilog->zl_header;
422 	lwb_t *lwb;
423 	dmu_tx_t *tx;
424 	uint64_t txg;
425 
426 	/*
427 	 * Wait for any previous destroy to complete.
428 	 */
429 	txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
430 
431 	if (BP_IS_HOLE(&zh->zh_log))
432 		return;
433 
434 	tx = dmu_tx_create(zilog->zl_os);
435 	(void) dmu_tx_assign(tx, TXG_WAIT);
436 	dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
437 	txg = dmu_tx_get_txg(tx);
438 
439 	mutex_enter(&zilog->zl_lock);
440 
441 	ASSERT3U(zilog->zl_destroy_txg, <, txg);
442 	zilog->zl_destroy_txg = txg;
443 
444 	if (!list_is_empty(&zilog->zl_lwb_list)) {
445 		ASSERT(zh->zh_claim_txg == 0);
446 		zilog->zl_keep_first = B_FALSE;
447 		while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) {
448 			list_remove(&zilog->zl_lwb_list, lwb);
449 			if (lwb->lwb_buf != NULL)
450 				zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
451 			zio_free_blk(zilog->zl_spa, &lwb->lwb_blk, txg);
452 			kmem_cache_free(zil_lwb_cache, lwb);
453 		}
454 	} else {
455 		zilog->zl_keep_first = keep_first;
456 		if (zh->zh_flags & ZIL_REPLAY_NEEDED) {
457 			ASSERT(!keep_first);
458 			(void) zil_parse(zilog, zil_free_log_block,
459 			    zil_free_log_record, tx, zh->zh_claim_txg);
460 		} else {
461 			/*
462 			 * Would like to assert zil_empty() but that
463 			 * would force us to read the log chain which
464 			 * requires us to do I/O to the log. This is
465 			 * overkill since we really just want to destroy
466 			 * the chain anyway.
467 			 */
468 			if (!keep_first) {
469 				blkptr_t bp = zh->zh_log;
470 				zio_free_blk(zilog->zl_spa, &bp, txg);
471 			}
472 		}
473 	}
474 	mutex_exit(&zilog->zl_lock);
475 
476 	dmu_tx_commit(tx);
477 }
478 
479 /*
480  * return true if the initial log block is not valid
481  */
482 static boolean_t
483 zil_empty(zilog_t *zilog)
484 {
485 	const zil_header_t *zh = zilog->zl_header;
486 	arc_buf_t *abuf = NULL;
487 
488 	if (BP_IS_HOLE(&zh->zh_log))
489 		return (B_TRUE);
490 
491 	if (zil_read_log_block(zilog, &zh->zh_log, &abuf) != 0)
492 		return (B_TRUE);
493 
494 	VERIFY(arc_buf_remove_ref(abuf, &abuf) == 1);
495 	return (B_FALSE);
496 }
497 
498 int
499 zil_claim(char *osname, void *txarg)
500 {
501 	dmu_tx_t *tx = txarg;
502 	uint64_t first_txg = dmu_tx_get_txg(tx);
503 	zilog_t *zilog;
504 	zil_header_t *zh;
505 	objset_t *os;
506 	int error;
507 
508 	error = dmu_objset_hold(osname, FTAG, &os);
509 	if (error) {
510 		cmn_err(CE_WARN, "can't open objset for %s", osname);
511 		return (0);
512 	}
513 
514 	zilog = dmu_objset_zil(os);
515 	zh = zil_header_in_syncing_context(zilog);
516 
517 	if (zilog->zl_spa->spa_log_state == SPA_LOG_CLEAR) {
518 		if (!BP_IS_HOLE(&zh->zh_log))
519 			zio_free_blk(zilog->zl_spa, &zh->zh_log, first_txg);
520 		BP_ZERO(&zh->zh_log);
521 		dsl_dataset_dirty(dmu_objset_ds(os), tx);
522 	}
523 
524 	/*
525 	 * Record here whether the zil has any records to replay.
526 	 * If the header block pointer is null or the block points
527 	 * to the stubby then we know there are no valid log records.
528 	 * We use the header to store this state as the the zilog gets
529 	 * freed later in dmu_objset_close().
530 	 * The flags (and the rest of the header fields) are cleared in
531 	 * zil_sync() as a result of a zil_destroy(), after replaying the log.
532 	 *
533 	 * Note, the intent log can be empty but still need the
534 	 * stubby to be claimed.
535 	 */
536 	if (!zil_empty(zilog)) {
537 		zh->zh_flags |= ZIL_REPLAY_NEEDED;
538 		dsl_dataset_dirty(dmu_objset_ds(os), tx);
539 	}
540 
541 	/*
542 	 * Claim all log blocks if we haven't already done so, and remember
543 	 * the highest claimed sequence number.  This ensures that if we can
544 	 * read only part of the log now (e.g. due to a missing device),
545 	 * but we can read the entire log later, we will not try to replay
546 	 * or destroy beyond the last block we successfully claimed.
547 	 */
548 	ASSERT3U(zh->zh_claim_txg, <=, first_txg);
549 	if (zh->zh_claim_txg == 0 && !BP_IS_HOLE(&zh->zh_log)) {
550 		zh->zh_claim_txg = first_txg;
551 		zh->zh_claim_seq = zil_parse(zilog, zil_claim_log_block,
552 		    zil_claim_log_record, tx, first_txg);
553 		dsl_dataset_dirty(dmu_objset_ds(os), tx);
554 	}
555 
556 	ASSERT3U(first_txg, ==, (spa_last_synced_txg(zilog->zl_spa) + 1));
557 	dmu_objset_rele(os, FTAG);
558 	return (0);
559 }
560 
561 /*
562  * Check the log by walking the log chain.
563  * Checksum errors are ok as they indicate the end of the chain.
564  * Any other error (no device or read failure) returns an error.
565  */
566 /* ARGSUSED */
567 int
568 zil_check_log_chain(char *osname, void *txarg)
569 {
570 	zilog_t *zilog;
571 	zil_header_t *zh;
572 	blkptr_t blk;
573 	arc_buf_t *abuf;
574 	objset_t *os;
575 	char *lrbuf;
576 	zil_trailer_t *ztp;
577 	int error;
578 
579 	error = dmu_objset_hold(osname, FTAG, &os);
580 	if (error) {
581 		cmn_err(CE_WARN, "can't open objset for %s", osname);
582 		return (0);
583 	}
584 
585 	zilog = dmu_objset_zil(os);
586 	zh = zil_header_in_syncing_context(zilog);
587 	blk = zh->zh_log;
588 	if (BP_IS_HOLE(&blk)) {
589 		dmu_objset_rele(os, FTAG);
590 		return (0); /* no chain */
591 	}
592 
593 	for (;;) {
594 		error = zil_read_log_block(zilog, &blk, &abuf);
595 		if (error)
596 			break;
597 		lrbuf = abuf->b_data;
598 		ztp = (zil_trailer_t *)(lrbuf + BP_GET_LSIZE(&blk)) - 1;
599 		blk = ztp->zit_next_blk;
600 		VERIFY(arc_buf_remove_ref(abuf, &abuf) == 1);
601 	}
602 	dmu_objset_rele(os, FTAG);
603 	if (error == ECKSUM)
604 		return (0); /* normal end of chain */
605 	return (error);
606 }
607 
608 static int
609 zil_vdev_compare(const void *x1, const void *x2)
610 {
611 	uint64_t v1 = ((zil_vdev_node_t *)x1)->zv_vdev;
612 	uint64_t v2 = ((zil_vdev_node_t *)x2)->zv_vdev;
613 
614 	if (v1 < v2)
615 		return (-1);
616 	if (v1 > v2)
617 		return (1);
618 
619 	return (0);
620 }
621 
622 void
623 zil_add_block(zilog_t *zilog, blkptr_t *bp)
624 {
625 	avl_tree_t *t = &zilog->zl_vdev_tree;
626 	avl_index_t where;
627 	zil_vdev_node_t *zv, zvsearch;
628 	int ndvas = BP_GET_NDVAS(bp);
629 	int i;
630 
631 	if (zfs_nocacheflush)
632 		return;
633 
634 	ASSERT(zilog->zl_writer);
635 
636 	/*
637 	 * Even though we're zl_writer, we still need a lock because the
638 	 * zl_get_data() callbacks may have dmu_sync() done callbacks
639 	 * that will run concurrently.
640 	 */
641 	mutex_enter(&zilog->zl_vdev_lock);
642 	for (i = 0; i < ndvas; i++) {
643 		zvsearch.zv_vdev = DVA_GET_VDEV(&bp->blk_dva[i]);
644 		if (avl_find(t, &zvsearch, &where) == NULL) {
645 			zv = kmem_alloc(sizeof (*zv), KM_SLEEP);
646 			zv->zv_vdev = zvsearch.zv_vdev;
647 			avl_insert(t, zv, where);
648 		}
649 	}
650 	mutex_exit(&zilog->zl_vdev_lock);
651 }
652 
653 void
654 zil_flush_vdevs(zilog_t *zilog)
655 {
656 	spa_t *spa = zilog->zl_spa;
657 	avl_tree_t *t = &zilog->zl_vdev_tree;
658 	void *cookie = NULL;
659 	zil_vdev_node_t *zv;
660 	zio_t *zio;
661 
662 	ASSERT(zilog->zl_writer);
663 
664 	/*
665 	 * We don't need zl_vdev_lock here because we're the zl_writer,
666 	 * and all zl_get_data() callbacks are done.
667 	 */
668 	if (avl_numnodes(t) == 0)
669 		return;
670 
671 	spa_config_enter(spa, SCL_STATE, FTAG, RW_READER);
672 
673 	zio = zio_root(spa, NULL, NULL, ZIO_FLAG_CANFAIL);
674 
675 	while ((zv = avl_destroy_nodes(t, &cookie)) != NULL) {
676 		vdev_t *vd = vdev_lookup_top(spa, zv->zv_vdev);
677 		if (vd != NULL)
678 			zio_flush(zio, vd);
679 		kmem_free(zv, sizeof (*zv));
680 	}
681 
682 	/*
683 	 * Wait for all the flushes to complete.  Not all devices actually
684 	 * support the DKIOCFLUSHWRITECACHE ioctl, so it's OK if it fails.
685 	 */
686 	(void) zio_wait(zio);
687 
688 	spa_config_exit(spa, SCL_STATE, FTAG);
689 }
690 
691 /*
692  * Function called when a log block write completes
693  */
694 static void
695 zil_lwb_write_done(zio_t *zio)
696 {
697 	lwb_t *lwb = zio->io_private;
698 	zilog_t *zilog = lwb->lwb_zilog;
699 
700 	ASSERT(BP_GET_COMPRESS(zio->io_bp) == ZIO_COMPRESS_OFF);
701 	ASSERT(BP_GET_CHECKSUM(zio->io_bp) == ZIO_CHECKSUM_ZILOG);
702 	ASSERT(BP_GET_TYPE(zio->io_bp) == DMU_OT_INTENT_LOG);
703 	ASSERT(BP_GET_LEVEL(zio->io_bp) == 0);
704 	ASSERT(BP_GET_BYTEORDER(zio->io_bp) == ZFS_HOST_BYTEORDER);
705 	ASSERT(!BP_IS_GANG(zio->io_bp));
706 	ASSERT(!BP_IS_HOLE(zio->io_bp));
707 	ASSERT(zio->io_bp->blk_fill == 0);
708 
709 	/*
710 	 * Ensure the lwb buffer pointer is cleared before releasing
711 	 * the txg. If we have had an allocation failure and
712 	 * the txg is waiting to sync then we want want zil_sync()
713 	 * to remove the lwb so that it's not picked up as the next new
714 	 * one in zil_commit_writer(). zil_sync() will only remove
715 	 * the lwb if lwb_buf is null.
716 	 */
717 	zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
718 	mutex_enter(&zilog->zl_lock);
719 	lwb->lwb_buf = NULL;
720 	if (zio->io_error)
721 		zilog->zl_log_error = B_TRUE;
722 
723 	/*
724 	 * Now that we've written this log block, we have a stable pointer
725 	 * to the next block in the chain, so it's OK to let the txg in
726 	 * which we allocated the next block sync. We still have the
727 	 * zl_lock to ensure zil_sync doesn't kmem free the lwb.
728 	 */
729 	txg_rele_to_sync(&lwb->lwb_txgh);
730 	mutex_exit(&zilog->zl_lock);
731 }
732 
733 /*
734  * Initialize the io for a log block.
735  */
736 static void
737 zil_lwb_write_init(zilog_t *zilog, lwb_t *lwb)
738 {
739 	zbookmark_t zb;
740 
741 	zb.zb_objset = lwb->lwb_blk.blk_cksum.zc_word[ZIL_ZC_OBJSET];
742 	zb.zb_object = 0;
743 	zb.zb_level = -1;
744 	zb.zb_blkid = lwb->lwb_blk.blk_cksum.zc_word[ZIL_ZC_SEQ];
745 
746 	if (zilog->zl_root_zio == NULL) {
747 		zilog->zl_root_zio = zio_root(zilog->zl_spa, NULL, NULL,
748 		    ZIO_FLAG_CANFAIL);
749 	}
750 	if (lwb->lwb_zio == NULL) {
751 		lwb->lwb_zio = zio_rewrite(zilog->zl_root_zio, zilog->zl_spa,
752 		    0, &lwb->lwb_blk, lwb->lwb_buf, lwb->lwb_sz,
753 		    zil_lwb_write_done, lwb, ZIO_PRIORITY_LOG_WRITE,
754 		    ZIO_FLAG_CANFAIL | ZIO_FLAG_DONT_PROPAGATE, &zb);
755 	}
756 }
757 
758 /*
759  * Use the slog as long as the logbias is 'latency' and the current commit size
760  * is less than the limit or the total list size is less than 2X the limit.
761  * Limit checking is disabled by setting zil_slog_limit to UINT64_MAX.
762  */
763 uint64_t zil_slog_limit = 1024 * 1024;
764 #define	USE_SLOG(zilog) (((zilog)->zl_logbias == ZFS_LOGBIAS_LATENCY) && \
765 	(((zilog)->zl_cur_used < zil_slog_limit) || \
766 	((zilog)->zl_itx_list_sz < (zil_slog_limit << 1))))
767 
768 /*
769  * Start a log block write and advance to the next log block.
770  * Calls are serialized.
771  */
772 static lwb_t *
773 zil_lwb_write_start(zilog_t *zilog, lwb_t *lwb)
774 {
775 	lwb_t *nlwb;
776 	zil_trailer_t *ztp = (zil_trailer_t *)(lwb->lwb_buf + lwb->lwb_sz) - 1;
777 	spa_t *spa = zilog->zl_spa;
778 	blkptr_t *bp = &ztp->zit_next_blk;
779 	uint64_t txg;
780 	uint64_t zil_blksz;
781 	int error;
782 
783 	ASSERT(lwb->lwb_nused <= ZIL_BLK_DATA_SZ(lwb));
784 
785 	/*
786 	 * Allocate the next block and save its address in this block
787 	 * before writing it in order to establish the log chain.
788 	 * Note that if the allocation of nlwb synced before we wrote
789 	 * the block that points at it (lwb), we'd leak it if we crashed.
790 	 * Therefore, we don't do txg_rele_to_sync() until zil_lwb_write_done().
791 	 */
792 	txg = txg_hold_open(zilog->zl_dmu_pool, &lwb->lwb_txgh);
793 	txg_rele_to_quiesce(&lwb->lwb_txgh);
794 
795 	/*
796 	 * Pick a ZIL blocksize. We request a size that is the
797 	 * maximum of the previous used size, the current used size and
798 	 * the amount waiting in the queue.
799 	 */
800 	zil_blksz = MAX(zilog->zl_prev_used,
801 	    zilog->zl_cur_used + sizeof (*ztp));
802 	zil_blksz = MAX(zil_blksz, zilog->zl_itx_list_sz + sizeof (*ztp));
803 	zil_blksz = P2ROUNDUP_TYPED(zil_blksz, ZIL_MIN_BLKSZ, uint64_t);
804 	if (zil_blksz > ZIL_MAX_BLKSZ)
805 		zil_blksz = ZIL_MAX_BLKSZ;
806 
807 	BP_ZERO(bp);
808 	/* pass the old blkptr in order to spread log blocks across devs */
809 	error = zio_alloc_blk(spa, zil_blksz, bp, &lwb->lwb_blk, txg,
810 	    USE_SLOG(zilog));
811 	if (error) {
812 		dmu_tx_t *tx = dmu_tx_create_assigned(zilog->zl_dmu_pool, txg);
813 
814 		/*
815 		 * We dirty the dataset to ensure that zil_sync() will
816 		 * be called to remove this lwb from our zl_lwb_list.
817 		 * Failing to do so, may leave an lwb with a NULL lwb_buf
818 		 * hanging around on the zl_lwb_list.
819 		 */
820 		dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
821 		dmu_tx_commit(tx);
822 
823 		/*
824 		 * Since we've just experienced an allocation failure so we
825 		 * terminate the current lwb and send it on its way.
826 		 */
827 		ztp->zit_pad = 0;
828 		ztp->zit_nused = lwb->lwb_nused;
829 		ztp->zit_bt.zbt_cksum = lwb->lwb_blk.blk_cksum;
830 		zio_nowait(lwb->lwb_zio);
831 
832 		/*
833 		 * By returning NULL the caller will call tx_wait_synced()
834 		 */
835 		return (NULL);
836 	}
837 
838 	ASSERT3U(bp->blk_birth, ==, txg);
839 	ztp->zit_pad = 0;
840 	ztp->zit_nused = lwb->lwb_nused;
841 	ztp->zit_bt.zbt_cksum = lwb->lwb_blk.blk_cksum;
842 	bp->blk_cksum = lwb->lwb_blk.blk_cksum;
843 	bp->blk_cksum.zc_word[ZIL_ZC_SEQ]++;
844 
845 	/*
846 	 * Allocate a new log write buffer (lwb).
847 	 */
848 	nlwb = kmem_cache_alloc(zil_lwb_cache, KM_SLEEP);
849 
850 	nlwb->lwb_zilog = zilog;
851 	nlwb->lwb_blk = *bp;
852 	nlwb->lwb_nused = 0;
853 	nlwb->lwb_sz = BP_GET_LSIZE(&nlwb->lwb_blk);
854 	nlwb->lwb_buf = zio_buf_alloc(nlwb->lwb_sz);
855 	nlwb->lwb_max_txg = txg;
856 	nlwb->lwb_zio = NULL;
857 
858 	/*
859 	 * Put new lwb at the end of the log chain
860 	 */
861 	mutex_enter(&zilog->zl_lock);
862 	list_insert_tail(&zilog->zl_lwb_list, nlwb);
863 	mutex_exit(&zilog->zl_lock);
864 
865 	/* Record the block for later vdev flushing */
866 	zil_add_block(zilog, &lwb->lwb_blk);
867 
868 	/*
869 	 * kick off the write for the old log block
870 	 */
871 	dprintf_bp(&lwb->lwb_blk, "lwb %p txg %llu: ", lwb, txg);
872 	ASSERT(lwb->lwb_zio);
873 	zio_nowait(lwb->lwb_zio);
874 
875 	return (nlwb);
876 }
877 
878 static lwb_t *
879 zil_lwb_commit(zilog_t *zilog, itx_t *itx, lwb_t *lwb)
880 {
881 	lr_t *lrc = &itx->itx_lr; /* common log record */
882 	lr_write_t *lr = (lr_write_t *)lrc;
883 	uint64_t txg = lrc->lrc_txg;
884 	uint64_t reclen = lrc->lrc_reclen;
885 	uint64_t dlen;
886 
887 	if (lwb == NULL)
888 		return (NULL);
889 	ASSERT(lwb->lwb_buf != NULL);
890 
891 	if (lrc->lrc_txtype == TX_WRITE && itx->itx_wr_state == WR_NEED_COPY)
892 		dlen = P2ROUNDUP_TYPED(
893 		    lr->lr_length, sizeof (uint64_t), uint64_t);
894 	else
895 		dlen = 0;
896 
897 	zilog->zl_cur_used += (reclen + dlen);
898 
899 	zil_lwb_write_init(zilog, lwb);
900 
901 	/*
902 	 * If this record won't fit in the current log block, start a new one.
903 	 */
904 	if (lwb->lwb_nused + reclen + dlen > ZIL_BLK_DATA_SZ(lwb)) {
905 		lwb = zil_lwb_write_start(zilog, lwb);
906 		if (lwb == NULL)
907 			return (NULL);
908 		zil_lwb_write_init(zilog, lwb);
909 		ASSERT(lwb->lwb_nused == 0);
910 		if (reclen + dlen > ZIL_BLK_DATA_SZ(lwb)) {
911 			txg_wait_synced(zilog->zl_dmu_pool, txg);
912 			return (lwb);
913 		}
914 	}
915 
916 	/*
917 	 * Update the lrc_seq, to be log record sequence number. See zil.h
918 	 * Then copy the record to the log buffer.
919 	 */
920 	lrc->lrc_seq = ++zilog->zl_lr_seq; /* we are single threaded */
921 	bcopy(lrc, lwb->lwb_buf + lwb->lwb_nused, reclen);
922 
923 	/*
924 	 * If it's a write, fetch the data or get its blkptr as appropriate.
925 	 */
926 	if (lrc->lrc_txtype == TX_WRITE) {
927 		if (txg > spa_freeze_txg(zilog->zl_spa))
928 			txg_wait_synced(zilog->zl_dmu_pool, txg);
929 		if (itx->itx_wr_state != WR_COPIED) {
930 			char *dbuf;
931 			int error;
932 
933 			/* alignment is guaranteed */
934 			lr = (lr_write_t *)(lwb->lwb_buf + lwb->lwb_nused);
935 			if (dlen) {
936 				ASSERT(itx->itx_wr_state == WR_NEED_COPY);
937 				dbuf = lwb->lwb_buf + lwb->lwb_nused + reclen;
938 				lr->lr_common.lrc_reclen += dlen;
939 			} else {
940 				ASSERT(itx->itx_wr_state == WR_INDIRECT);
941 				dbuf = NULL;
942 			}
943 			error = zilog->zl_get_data(
944 			    itx->itx_private, lr, dbuf, lwb->lwb_zio);
945 			if (error == EIO) {
946 				txg_wait_synced(zilog->zl_dmu_pool, txg);
947 				return (lwb);
948 			}
949 			if (error) {
950 				ASSERT(error == ENOENT || error == EEXIST ||
951 				    error == EALREADY);
952 				return (lwb);
953 			}
954 		}
955 	}
956 
957 	lwb->lwb_nused += reclen + dlen;
958 	lwb->lwb_max_txg = MAX(lwb->lwb_max_txg, txg);
959 	ASSERT3U(lwb->lwb_nused, <=, ZIL_BLK_DATA_SZ(lwb));
960 	ASSERT3U(P2PHASE(lwb->lwb_nused, sizeof (uint64_t)), ==, 0);
961 
962 	return (lwb);
963 }
964 
965 itx_t *
966 zil_itx_create(uint64_t txtype, size_t lrsize)
967 {
968 	itx_t *itx;
969 
970 	lrsize = P2ROUNDUP_TYPED(lrsize, sizeof (uint64_t), size_t);
971 
972 	itx = kmem_alloc(offsetof(itx_t, itx_lr) + lrsize, KM_SLEEP);
973 	itx->itx_lr.lrc_txtype = txtype;
974 	itx->itx_lr.lrc_reclen = lrsize;
975 	itx->itx_sod = lrsize; /* if write & WR_NEED_COPY will be increased */
976 	itx->itx_lr.lrc_seq = 0;	/* defensive */
977 
978 	return (itx);
979 }
980 
981 uint64_t
982 zil_itx_assign(zilog_t *zilog, itx_t *itx, dmu_tx_t *tx)
983 {
984 	uint64_t seq;
985 
986 	ASSERT(itx->itx_lr.lrc_seq == 0);
987 
988 	mutex_enter(&zilog->zl_lock);
989 	list_insert_tail(&zilog->zl_itx_list, itx);
990 	zilog->zl_itx_list_sz += itx->itx_sod;
991 	itx->itx_lr.lrc_txg = dmu_tx_get_txg(tx);
992 	itx->itx_lr.lrc_seq = seq = ++zilog->zl_itx_seq;
993 	mutex_exit(&zilog->zl_lock);
994 
995 	return (seq);
996 }
997 
998 /*
999  * Free up all in-memory intent log transactions that have now been synced.
1000  */
1001 static void
1002 zil_itx_clean(zilog_t *zilog)
1003 {
1004 	uint64_t synced_txg = spa_last_synced_txg(zilog->zl_spa);
1005 	uint64_t freeze_txg = spa_freeze_txg(zilog->zl_spa);
1006 	list_t clean_list;
1007 	itx_t *itx;
1008 
1009 	list_create(&clean_list, sizeof (itx_t), offsetof(itx_t, itx_node));
1010 
1011 	mutex_enter(&zilog->zl_lock);
1012 	/* wait for a log writer to finish walking list */
1013 	while (zilog->zl_writer) {
1014 		cv_wait(&zilog->zl_cv_writer, &zilog->zl_lock);
1015 	}
1016 
1017 	/*
1018 	 * Move the sync'd log transactions to a separate list so we can call
1019 	 * kmem_free without holding the zl_lock.
1020 	 *
1021 	 * There is no need to set zl_writer as we don't drop zl_lock here
1022 	 */
1023 	while ((itx = list_head(&zilog->zl_itx_list)) != NULL &&
1024 	    itx->itx_lr.lrc_txg <= MIN(synced_txg, freeze_txg)) {
1025 		list_remove(&zilog->zl_itx_list, itx);
1026 		zilog->zl_itx_list_sz -= itx->itx_sod;
1027 		list_insert_tail(&clean_list, itx);
1028 	}
1029 	cv_broadcast(&zilog->zl_cv_writer);
1030 	mutex_exit(&zilog->zl_lock);
1031 
1032 	/* destroy sync'd log transactions */
1033 	while ((itx = list_head(&clean_list)) != NULL) {
1034 		list_remove(&clean_list, itx);
1035 		kmem_free(itx, offsetof(itx_t, itx_lr)
1036 		    + itx->itx_lr.lrc_reclen);
1037 	}
1038 	list_destroy(&clean_list);
1039 }
1040 
1041 /*
1042  * If there are any in-memory intent log transactions which have now been
1043  * synced then start up a taskq to free them.
1044  */
1045 void
1046 zil_clean(zilog_t *zilog)
1047 {
1048 	itx_t *itx;
1049 
1050 	mutex_enter(&zilog->zl_lock);
1051 	itx = list_head(&zilog->zl_itx_list);
1052 	if ((itx != NULL) &&
1053 	    (itx->itx_lr.lrc_txg <= spa_last_synced_txg(zilog->zl_spa))) {
1054 		(void) taskq_dispatch(zilog->zl_clean_taskq,
1055 		    (task_func_t *)zil_itx_clean, zilog, TQ_NOSLEEP);
1056 	}
1057 	mutex_exit(&zilog->zl_lock);
1058 }
1059 
1060 static void
1061 zil_commit_writer(zilog_t *zilog, uint64_t seq, uint64_t foid)
1062 {
1063 	uint64_t txg;
1064 	uint64_t commit_seq = 0;
1065 	itx_t *itx, *itx_next = (itx_t *)-1;
1066 	lwb_t *lwb;
1067 	spa_t *spa;
1068 
1069 	zilog->zl_writer = B_TRUE;
1070 	ASSERT(zilog->zl_root_zio == NULL);
1071 	spa = zilog->zl_spa;
1072 
1073 	if (zilog->zl_suspend) {
1074 		lwb = NULL;
1075 	} else {
1076 		lwb = list_tail(&zilog->zl_lwb_list);
1077 		if (lwb == NULL) {
1078 			/*
1079 			 * Return if there's nothing to flush before we
1080 			 * dirty the fs by calling zil_create()
1081 			 */
1082 			if (list_is_empty(&zilog->zl_itx_list)) {
1083 				zilog->zl_writer = B_FALSE;
1084 				return;
1085 			}
1086 			mutex_exit(&zilog->zl_lock);
1087 			zil_create(zilog);
1088 			mutex_enter(&zilog->zl_lock);
1089 			lwb = list_tail(&zilog->zl_lwb_list);
1090 		}
1091 	}
1092 
1093 	/* Loop through in-memory log transactions filling log blocks. */
1094 	DTRACE_PROBE1(zil__cw1, zilog_t *, zilog);
1095 	for (;;) {
1096 		/*
1097 		 * Find the next itx to push:
1098 		 * Push all transactions related to specified foid and all
1099 		 * other transactions except TX_WRITE, TX_TRUNCATE,
1100 		 * TX_SETATTR and TX_ACL for all other files.
1101 		 */
1102 		if (itx_next != (itx_t *)-1)
1103 			itx = itx_next;
1104 		else
1105 			itx = list_head(&zilog->zl_itx_list);
1106 		for (; itx != NULL; itx = list_next(&zilog->zl_itx_list, itx)) {
1107 			if (foid == 0) /* push all foids? */
1108 				break;
1109 			if (itx->itx_sync) /* push all O_[D]SYNC */
1110 				break;
1111 			switch (itx->itx_lr.lrc_txtype) {
1112 			case TX_SETATTR:
1113 			case TX_WRITE:
1114 			case TX_TRUNCATE:
1115 			case TX_ACL:
1116 				/* lr_foid is same offset for these records */
1117 				if (((lr_write_t *)&itx->itx_lr)->lr_foid
1118 				    != foid) {
1119 					continue; /* skip this record */
1120 				}
1121 			}
1122 			break;
1123 		}
1124 		if (itx == NULL)
1125 			break;
1126 
1127 		if ((itx->itx_lr.lrc_seq > seq) &&
1128 		    ((lwb == NULL) || (lwb->lwb_nused == 0) ||
1129 		    (lwb->lwb_nused + itx->itx_sod > ZIL_BLK_DATA_SZ(lwb)))) {
1130 			break;
1131 		}
1132 
1133 		/*
1134 		 * Save the next pointer.  Even though we soon drop
1135 		 * zl_lock all threads that may change the list
1136 		 * (another writer or zil_itx_clean) can't do so until
1137 		 * they have zl_writer.
1138 		 */
1139 		itx_next = list_next(&zilog->zl_itx_list, itx);
1140 		list_remove(&zilog->zl_itx_list, itx);
1141 		zilog->zl_itx_list_sz -= itx->itx_sod;
1142 		mutex_exit(&zilog->zl_lock);
1143 		txg = itx->itx_lr.lrc_txg;
1144 		ASSERT(txg);
1145 
1146 		if (txg > spa_last_synced_txg(spa) ||
1147 		    txg > spa_freeze_txg(spa))
1148 			lwb = zil_lwb_commit(zilog, itx, lwb);
1149 		kmem_free(itx, offsetof(itx_t, itx_lr)
1150 		    + itx->itx_lr.lrc_reclen);
1151 		mutex_enter(&zilog->zl_lock);
1152 	}
1153 	DTRACE_PROBE1(zil__cw2, zilog_t *, zilog);
1154 	/* determine commit sequence number */
1155 	itx = list_head(&zilog->zl_itx_list);
1156 	if (itx)
1157 		commit_seq = itx->itx_lr.lrc_seq;
1158 	else
1159 		commit_seq = zilog->zl_itx_seq;
1160 	mutex_exit(&zilog->zl_lock);
1161 
1162 	/* write the last block out */
1163 	if (lwb != NULL && lwb->lwb_zio != NULL)
1164 		lwb = zil_lwb_write_start(zilog, lwb);
1165 
1166 	zilog->zl_prev_used = zilog->zl_cur_used;
1167 	zilog->zl_cur_used = 0;
1168 
1169 	/*
1170 	 * Wait if necessary for the log blocks to be on stable storage.
1171 	 */
1172 	if (zilog->zl_root_zio) {
1173 		DTRACE_PROBE1(zil__cw3, zilog_t *, zilog);
1174 		(void) zio_wait(zilog->zl_root_zio);
1175 		zilog->zl_root_zio = NULL;
1176 		DTRACE_PROBE1(zil__cw4, zilog_t *, zilog);
1177 		zil_flush_vdevs(zilog);
1178 	}
1179 
1180 	if (zilog->zl_log_error || lwb == NULL) {
1181 		zilog->zl_log_error = 0;
1182 		txg_wait_synced(zilog->zl_dmu_pool, 0);
1183 	}
1184 
1185 	mutex_enter(&zilog->zl_lock);
1186 	zilog->zl_writer = B_FALSE;
1187 
1188 	ASSERT3U(commit_seq, >=, zilog->zl_commit_seq);
1189 	zilog->zl_commit_seq = commit_seq;
1190 }
1191 
1192 /*
1193  * Push zfs transactions to stable storage up to the supplied sequence number.
1194  * If foid is 0 push out all transactions, otherwise push only those
1195  * for that file or might have been used to create that file.
1196  */
1197 void
1198 zil_commit(zilog_t *zilog, uint64_t seq, uint64_t foid)
1199 {
1200 	if (zilog == NULL || seq == 0)
1201 		return;
1202 
1203 	mutex_enter(&zilog->zl_lock);
1204 
1205 	seq = MIN(seq, zilog->zl_itx_seq);	/* cap seq at largest itx seq */
1206 
1207 	while (zilog->zl_writer) {
1208 		cv_wait(&zilog->zl_cv_writer, &zilog->zl_lock);
1209 		if (seq < zilog->zl_commit_seq) {
1210 			mutex_exit(&zilog->zl_lock);
1211 			return;
1212 		}
1213 	}
1214 	zil_commit_writer(zilog, seq, foid); /* drops zl_lock */
1215 	/* wake up others waiting on the commit */
1216 	cv_broadcast(&zilog->zl_cv_writer);
1217 	mutex_exit(&zilog->zl_lock);
1218 }
1219 
1220 /*
1221  * Called in syncing context to free committed log blocks and update log header.
1222  */
1223 void
1224 zil_sync(zilog_t *zilog, dmu_tx_t *tx)
1225 {
1226 	zil_header_t *zh = zil_header_in_syncing_context(zilog);
1227 	uint64_t txg = dmu_tx_get_txg(tx);
1228 	spa_t *spa = zilog->zl_spa;
1229 	lwb_t *lwb;
1230 
1231 	/*
1232 	 * We don't zero out zl_destroy_txg, so make sure we don't try
1233 	 * to destroy it twice.
1234 	 */
1235 	if (spa_sync_pass(spa) != 1)
1236 		return;
1237 
1238 	mutex_enter(&zilog->zl_lock);
1239 
1240 	ASSERT(zilog->zl_stop_sync == 0);
1241 
1242 	zh->zh_replay_seq = zilog->zl_replayed_seq[txg & TXG_MASK];
1243 
1244 	if (zilog->zl_destroy_txg == txg) {
1245 		blkptr_t blk = zh->zh_log;
1246 
1247 		ASSERT(list_head(&zilog->zl_lwb_list) == NULL);
1248 
1249 		bzero(zh, sizeof (zil_header_t));
1250 		bzero(zilog->zl_replayed_seq, sizeof (zilog->zl_replayed_seq));
1251 
1252 		if (zilog->zl_keep_first) {
1253 			/*
1254 			 * If this block was part of log chain that couldn't
1255 			 * be claimed because a device was missing during
1256 			 * zil_claim(), but that device later returns,
1257 			 * then this block could erroneously appear valid.
1258 			 * To guard against this, assign a new GUID to the new
1259 			 * log chain so it doesn't matter what blk points to.
1260 			 */
1261 			zil_init_log_chain(zilog, &blk);
1262 			zh->zh_log = blk;
1263 		}
1264 	}
1265 
1266 	while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) {
1267 		zh->zh_log = lwb->lwb_blk;
1268 		if (lwb->lwb_buf != NULL || lwb->lwb_max_txg > txg)
1269 			break;
1270 		list_remove(&zilog->zl_lwb_list, lwb);
1271 		zio_free_blk(spa, &lwb->lwb_blk, txg);
1272 		kmem_cache_free(zil_lwb_cache, lwb);
1273 
1274 		/*
1275 		 * If we don't have anything left in the lwb list then
1276 		 * we've had an allocation failure and we need to zero
1277 		 * out the zil_header blkptr so that we don't end
1278 		 * up freeing the same block twice.
1279 		 */
1280 		if (list_head(&zilog->zl_lwb_list) == NULL)
1281 			BP_ZERO(&zh->zh_log);
1282 	}
1283 	mutex_exit(&zilog->zl_lock);
1284 }
1285 
1286 void
1287 zil_init(void)
1288 {
1289 	zil_lwb_cache = kmem_cache_create("zil_lwb_cache",
1290 	    sizeof (struct lwb), 0, NULL, NULL, NULL, NULL, NULL, 0);
1291 }
1292 
1293 void
1294 zil_fini(void)
1295 {
1296 	kmem_cache_destroy(zil_lwb_cache);
1297 }
1298 
1299 void
1300 zil_set_logbias(zilog_t *zilog, uint64_t logbias)
1301 {
1302 	zilog->zl_logbias = logbias;
1303 }
1304 
1305 zilog_t *
1306 zil_alloc(objset_t *os, zil_header_t *zh_phys)
1307 {
1308 	zilog_t *zilog;
1309 
1310 	zilog = kmem_zalloc(sizeof (zilog_t), KM_SLEEP);
1311 
1312 	zilog->zl_header = zh_phys;
1313 	zilog->zl_os = os;
1314 	zilog->zl_spa = dmu_objset_spa(os);
1315 	zilog->zl_dmu_pool = dmu_objset_pool(os);
1316 	zilog->zl_destroy_txg = TXG_INITIAL - 1;
1317 	zilog->zl_logbias = dmu_objset_logbias(os);
1318 
1319 	mutex_init(&zilog->zl_lock, NULL, MUTEX_DEFAULT, NULL);
1320 
1321 	list_create(&zilog->zl_itx_list, sizeof (itx_t),
1322 	    offsetof(itx_t, itx_node));
1323 
1324 	list_create(&zilog->zl_lwb_list, sizeof (lwb_t),
1325 	    offsetof(lwb_t, lwb_node));
1326 
1327 	mutex_init(&zilog->zl_vdev_lock, NULL, MUTEX_DEFAULT, NULL);
1328 
1329 	avl_create(&zilog->zl_vdev_tree, zil_vdev_compare,
1330 	    sizeof (zil_vdev_node_t), offsetof(zil_vdev_node_t, zv_node));
1331 
1332 	cv_init(&zilog->zl_cv_writer, NULL, CV_DEFAULT, NULL);
1333 	cv_init(&zilog->zl_cv_suspend, NULL, CV_DEFAULT, NULL);
1334 
1335 	return (zilog);
1336 }
1337 
1338 void
1339 zil_free(zilog_t *zilog)
1340 {
1341 	lwb_t *lwb;
1342 
1343 	zilog->zl_stop_sync = 1;
1344 
1345 	while ((lwb = list_head(&zilog->zl_lwb_list)) != NULL) {
1346 		list_remove(&zilog->zl_lwb_list, lwb);
1347 		if (lwb->lwb_buf != NULL)
1348 			zio_buf_free(lwb->lwb_buf, lwb->lwb_sz);
1349 		kmem_cache_free(zil_lwb_cache, lwb);
1350 	}
1351 	list_destroy(&zilog->zl_lwb_list);
1352 
1353 	avl_destroy(&zilog->zl_vdev_tree);
1354 	mutex_destroy(&zilog->zl_vdev_lock);
1355 
1356 	ASSERT(list_head(&zilog->zl_itx_list) == NULL);
1357 	list_destroy(&zilog->zl_itx_list);
1358 	mutex_destroy(&zilog->zl_lock);
1359 
1360 	cv_destroy(&zilog->zl_cv_writer);
1361 	cv_destroy(&zilog->zl_cv_suspend);
1362 
1363 	kmem_free(zilog, sizeof (zilog_t));
1364 }
1365 
1366 /*
1367  * Open an intent log.
1368  */
1369 zilog_t *
1370 zil_open(objset_t *os, zil_get_data_t *get_data)
1371 {
1372 	zilog_t *zilog = dmu_objset_zil(os);
1373 
1374 	zilog->zl_get_data = get_data;
1375 	zilog->zl_clean_taskq = taskq_create("zil_clean", 1, minclsyspri,
1376 	    2, 2, TASKQ_PREPOPULATE);
1377 
1378 	return (zilog);
1379 }
1380 
1381 /*
1382  * Close an intent log.
1383  */
1384 void
1385 zil_close(zilog_t *zilog)
1386 {
1387 	/*
1388 	 * If the log isn't already committed, mark the objset dirty
1389 	 * (so zil_sync() will be called) and wait for that txg to sync.
1390 	 */
1391 	if (!zil_is_committed(zilog)) {
1392 		uint64_t txg;
1393 		dmu_tx_t *tx = dmu_tx_create(zilog->zl_os);
1394 		(void) dmu_tx_assign(tx, TXG_WAIT);
1395 		dsl_dataset_dirty(dmu_objset_ds(zilog->zl_os), tx);
1396 		txg = dmu_tx_get_txg(tx);
1397 		dmu_tx_commit(tx);
1398 		txg_wait_synced(zilog->zl_dmu_pool, txg);
1399 	}
1400 
1401 	taskq_destroy(zilog->zl_clean_taskq);
1402 	zilog->zl_clean_taskq = NULL;
1403 	zilog->zl_get_data = NULL;
1404 
1405 	zil_itx_clean(zilog);
1406 	ASSERT(list_head(&zilog->zl_itx_list) == NULL);
1407 }
1408 
1409 /*
1410  * Suspend an intent log.  While in suspended mode, we still honor
1411  * synchronous semantics, but we rely on txg_wait_synced() to do it.
1412  * We suspend the log briefly when taking a snapshot so that the snapshot
1413  * contains all the data it's supposed to, and has an empty intent log.
1414  */
1415 int
1416 zil_suspend(zilog_t *zilog)
1417 {
1418 	const zil_header_t *zh = zilog->zl_header;
1419 
1420 	mutex_enter(&zilog->zl_lock);
1421 	if (zh->zh_flags & ZIL_REPLAY_NEEDED) {		/* unplayed log */
1422 		mutex_exit(&zilog->zl_lock);
1423 		return (EBUSY);
1424 	}
1425 	if (zilog->zl_suspend++ != 0) {
1426 		/*
1427 		 * Someone else already began a suspend.
1428 		 * Just wait for them to finish.
1429 		 */
1430 		while (zilog->zl_suspending)
1431 			cv_wait(&zilog->zl_cv_suspend, &zilog->zl_lock);
1432 		mutex_exit(&zilog->zl_lock);
1433 		return (0);
1434 	}
1435 	zilog->zl_suspending = B_TRUE;
1436 	mutex_exit(&zilog->zl_lock);
1437 
1438 	zil_commit(zilog, UINT64_MAX, 0);
1439 
1440 	/*
1441 	 * Wait for any in-flight log writes to complete.
1442 	 */
1443 	mutex_enter(&zilog->zl_lock);
1444 	while (zilog->zl_writer)
1445 		cv_wait(&zilog->zl_cv_writer, &zilog->zl_lock);
1446 	mutex_exit(&zilog->zl_lock);
1447 
1448 	zil_destroy(zilog, B_FALSE);
1449 
1450 	mutex_enter(&zilog->zl_lock);
1451 	zilog->zl_suspending = B_FALSE;
1452 	cv_broadcast(&zilog->zl_cv_suspend);
1453 	mutex_exit(&zilog->zl_lock);
1454 
1455 	return (0);
1456 }
1457 
1458 void
1459 zil_resume(zilog_t *zilog)
1460 {
1461 	mutex_enter(&zilog->zl_lock);
1462 	ASSERT(zilog->zl_suspend != 0);
1463 	zilog->zl_suspend--;
1464 	mutex_exit(&zilog->zl_lock);
1465 }
1466 
1467 /*
1468  * Read in the data for the dmu_sync()ed block, and change the log
1469  * record to write this whole block.
1470  */
1471 void
1472 zil_get_replay_data(zilog_t *zilog, lr_write_t *lr)
1473 {
1474 	blkptr_t *wbp = &lr->lr_blkptr;
1475 	char *wbuf = (char *)(lr + 1); /* data follows lr_write_t */
1476 	uint64_t blksz;
1477 
1478 	if (BP_IS_HOLE(wbp)) {	/* compressed to a hole */
1479 		blksz = BP_GET_LSIZE(&lr->lr_blkptr);
1480 		/*
1481 		 * If the blksz is zero then we must be replaying a log
1482 		 * from an version prior to setting the blksize of null blocks.
1483 		 * So we just zero the actual write size reqeusted.
1484 		 */
1485 		if (blksz == 0) {
1486 			bzero(wbuf, lr->lr_length);
1487 			return;
1488 		}
1489 		bzero(wbuf, blksz);
1490 	} else {
1491 		/*
1492 		 * A subsequent write may have overwritten this block, in which
1493 		 * case wbp may have been been freed and reallocated, and our
1494 		 * read of wbp may fail with a checksum error.  We can safely
1495 		 * ignore this because the later write will provide the
1496 		 * correct data.
1497 		 */
1498 		zbookmark_t zb;
1499 
1500 		zb.zb_objset = dmu_objset_id(zilog->zl_os);
1501 		zb.zb_object = lr->lr_foid;
1502 		zb.zb_level = 0;
1503 		zb.zb_blkid = -1; /* unknown */
1504 
1505 		blksz = BP_GET_LSIZE(&lr->lr_blkptr);
1506 		(void) zio_wait(zio_read(NULL, zilog->zl_spa, wbp, wbuf, blksz,
1507 		    NULL, NULL, ZIO_PRIORITY_SYNC_READ,
1508 		    ZIO_FLAG_CANFAIL | ZIO_FLAG_SPECULATIVE, &zb));
1509 	}
1510 	lr->lr_offset -= lr->lr_offset % blksz;
1511 	lr->lr_length = blksz;
1512 }
1513 
1514 typedef struct zil_replay_arg {
1515 	objset_t	*zr_os;
1516 	zil_replay_func_t **zr_replay;
1517 	void		*zr_arg;
1518 	boolean_t	zr_byteswap;
1519 	char		*zr_lrbuf;
1520 } zil_replay_arg_t;
1521 
1522 static void
1523 zil_replay_log_record(zilog_t *zilog, lr_t *lr, void *zra, uint64_t claim_txg)
1524 {
1525 	zil_replay_arg_t *zr = zra;
1526 	const zil_header_t *zh = zilog->zl_header;
1527 	uint64_t reclen = lr->lrc_reclen;
1528 	uint64_t txtype = lr->lrc_txtype;
1529 	char *name;
1530 	int pass, error;
1531 
1532 	if (!zilog->zl_replay)			/* giving up */
1533 		return;
1534 
1535 	if (lr->lrc_txg < claim_txg)		/* already committed */
1536 		return;
1537 
1538 	if (lr->lrc_seq <= zh->zh_replay_seq)	/* already replayed */
1539 		return;
1540 
1541 	/* Strip case-insensitive bit, still present in log record */
1542 	txtype &= ~TX_CI;
1543 
1544 	if (txtype == 0 || txtype >= TX_MAX_TYPE) {
1545 		error = EINVAL;
1546 		goto bad;
1547 	}
1548 
1549 	/*
1550 	 * Make a copy of the data so we can revise and extend it.
1551 	 */
1552 	bcopy(lr, zr->zr_lrbuf, reclen);
1553 
1554 	/*
1555 	 * The log block containing this lr may have been byteswapped
1556 	 * so that we can easily examine common fields like lrc_txtype.
1557 	 * However, the log is a mix of different data types, and only the
1558 	 * replay vectors know how to byteswap their records.  Therefore, if
1559 	 * the lr was byteswapped, undo it before invoking the replay vector.
1560 	 */
1561 	if (zr->zr_byteswap)
1562 		byteswap_uint64_array(zr->zr_lrbuf, reclen);
1563 
1564 	/*
1565 	 * We must now do two things atomically: replay this log record,
1566 	 * and update the log header sequence number to reflect the fact that
1567 	 * we did so. At the end of each replay function the sequence number
1568 	 * is updated if we are in replay mode.
1569 	 */
1570 	for (pass = 1; pass <= 2; pass++) {
1571 		zilog->zl_replaying_seq = lr->lrc_seq;
1572 		/* Only byteswap (if needed) on the 1st pass.  */
1573 		error = zr->zr_replay[txtype](zr->zr_arg, zr->zr_lrbuf,
1574 		    zr->zr_byteswap && pass == 1);
1575 
1576 		if (!error)
1577 			return;
1578 
1579 		/*
1580 		 * The DMU's dnode layer doesn't see removes until the txg
1581 		 * commits, so a subsequent claim can spuriously fail with
1582 		 * EEXIST. So if we receive any error we try syncing out
1583 		 * any removes then retry the transaction.
1584 		 */
1585 		if (pass == 1)
1586 			txg_wait_synced(spa_get_dsl(zilog->zl_spa), 0);
1587 	}
1588 
1589 bad:
1590 	ASSERT(error);
1591 	name = kmem_alloc(MAXNAMELEN, KM_SLEEP);
1592 	dmu_objset_name(zr->zr_os, name);
1593 	cmn_err(CE_WARN, "ZFS replay transaction error %d, "
1594 	    "dataset %s, seq 0x%llx, txtype %llu %s\n",
1595 	    error, name, (u_longlong_t)lr->lrc_seq, (u_longlong_t)txtype,
1596 	    (lr->lrc_txtype & TX_CI) ? "CI" : "");
1597 	zilog->zl_replay = B_FALSE;
1598 	kmem_free(name, MAXNAMELEN);
1599 }
1600 
1601 /* ARGSUSED */
1602 static void
1603 zil_incr_blks(zilog_t *zilog, blkptr_t *bp, void *arg, uint64_t claim_txg)
1604 {
1605 	zilog->zl_replay_blks++;
1606 }
1607 
1608 /*
1609  * If this dataset has a non-empty intent log, replay it and destroy it.
1610  */
1611 void
1612 zil_replay(objset_t *os, void *arg, zil_replay_func_t *replay_func[TX_MAX_TYPE])
1613 {
1614 	zilog_t *zilog = dmu_objset_zil(os);
1615 	const zil_header_t *zh = zilog->zl_header;
1616 	zil_replay_arg_t zr;
1617 
1618 	if ((zh->zh_flags & ZIL_REPLAY_NEEDED) == 0) {
1619 		zil_destroy(zilog, B_TRUE);
1620 		return;
1621 	}
1622 
1623 	zr.zr_os = os;
1624 	zr.zr_replay = replay_func;
1625 	zr.zr_arg = arg;
1626 	zr.zr_byteswap = BP_SHOULD_BYTESWAP(&zh->zh_log);
1627 	zr.zr_lrbuf = kmem_alloc(2 * SPA_MAXBLOCKSIZE, KM_SLEEP);
1628 
1629 	/*
1630 	 * Wait for in-progress removes to sync before starting replay.
1631 	 */
1632 	txg_wait_synced(zilog->zl_dmu_pool, 0);
1633 
1634 	zilog->zl_replay = B_TRUE;
1635 	zilog->zl_replay_time = lbolt;
1636 	ASSERT(zilog->zl_replay_blks == 0);
1637 	(void) zil_parse(zilog, zil_incr_blks, zil_replay_log_record, &zr,
1638 	    zh->zh_claim_txg);
1639 	kmem_free(zr.zr_lrbuf, 2 * SPA_MAXBLOCKSIZE);
1640 
1641 	zil_destroy(zilog, B_FALSE);
1642 	txg_wait_synced(zilog->zl_dmu_pool, zilog->zl_destroy_txg);
1643 	zilog->zl_replay = B_FALSE;
1644 }
1645 
1646 /*
1647  * Report whether all transactions are committed
1648  */
1649 int
1650 zil_is_committed(zilog_t *zilog)
1651 {
1652 	lwb_t *lwb;
1653 	int ret;
1654 
1655 	mutex_enter(&zilog->zl_lock);
1656 	while (zilog->zl_writer)
1657 		cv_wait(&zilog->zl_cv_writer, &zilog->zl_lock);
1658 
1659 	/* recent unpushed intent log transactions? */
1660 	if (!list_is_empty(&zilog->zl_itx_list)) {
1661 		ret = B_FALSE;
1662 		goto out;
1663 	}
1664 
1665 	/* intent log never used? */
1666 	lwb = list_head(&zilog->zl_lwb_list);
1667 	if (lwb == NULL) {
1668 		ret = B_TRUE;
1669 		goto out;
1670 	}
1671 
1672 	/*
1673 	 * more than 1 log buffer means zil_sync() hasn't yet freed
1674 	 * entries after a txg has committed
1675 	 */
1676 	if (list_next(&zilog->zl_lwb_list, lwb)) {
1677 		ret = B_FALSE;
1678 		goto out;
1679 	}
1680 
1681 	ASSERT(zil_empty(zilog));
1682 	ret = B_TRUE;
1683 out:
1684 	cv_broadcast(&zilog->zl_cv_writer);
1685 	mutex_exit(&zilog->zl_lock);
1686 	return (ret);
1687 }
1688 
1689 /* ARGSUSED */
1690 int
1691 zil_vdev_offline(char *osname, void *arg)
1692 {
1693 	objset_t *os;
1694 	zilog_t *zilog;
1695 	int error;
1696 
1697 	error = dmu_objset_hold(osname, FTAG, &os);
1698 	if (error)
1699 		return (error);
1700 
1701 	zilog = dmu_objset_zil(os);
1702 	if (zil_suspend(zilog) != 0)
1703 		error = EEXIST;
1704 	else
1705 		zil_resume(zilog);
1706 	dmu_objset_rele(os, FTAG);
1707 	return (error);
1708 }
1709