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