xref: /illumos-gate/usr/src/uts/common/fs/zfs/txg.c (revision a6f561b4aee75d0d028e7b36b151c8ed8a86bc76)
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  * Portions Copyright 2011 Martin Matuska
24  * Copyright (c) 2013 by Delphix. All rights reserved.
25  */
26 
27 #include <sys/zfs_context.h>
28 #include <sys/txg_impl.h>
29 #include <sys/dmu_impl.h>
30 #include <sys/dmu_tx.h>
31 #include <sys/dsl_pool.h>
32 #include <sys/dsl_scan.h>
33 #include <sys/callb.h>
34 
35 /*
36  * ZFS Transaction Groups
37  * ----------------------
38  *
39  * ZFS transaction groups are, as the name implies, groups of transactions
40  * that act on persistent state. ZFS asserts consistency at the granularity of
41  * these transaction groups. Each successive transaction group (txg) is
42  * assigned a 64-bit consecutive identifier. There are three active
43  * transaction group states: open, quiescing, or syncing. At any given time,
44  * there may be an active txg associated with each state; each active txg may
45  * either be processing, or blocked waiting to enter the next state. There may
46  * be up to three active txgs, and there is always a txg in the open state
47  * (though it may be blocked waiting to enter the quiescing state). In broad
48  * strokes, transactions — operations that change in-memory structures — are
49  * accepted into the txg in the open state, and are completed while the txg is
50  * in the open or quiescing states. The accumulated changes are written to
51  * disk in the syncing state.
52  *
53  * Open
54  *
55  * When a new txg becomes active, it first enters the open state. New
56  * transactions — updates to in-memory structures — are assigned to the
57  * currently open txg. There is always a txg in the open state so that ZFS can
58  * accept new changes (though the txg may refuse new changes if it has hit
59  * some limit). ZFS advances the open txg to the next state for a variety of
60  * reasons such as it hitting a time or size threshold, or the execution of an
61  * administrative action that must be completed in the syncing state.
62  *
63  * Quiescing
64  *
65  * After a txg exits the open state, it enters the quiescing state. The
66  * quiescing state is intended to provide a buffer between accepting new
67  * transactions in the open state and writing them out to stable storage in
68  * the syncing state. While quiescing, transactions can continue their
69  * operation without delaying either of the other states. Typically, a txg is
70  * in the quiescing state very briefly since the operations are bounded by
71  * software latencies rather than, say, slower I/O latencies. After all
72  * transactions complete, the txg is ready to enter the next state.
73  *
74  * Syncing
75  *
76  * In the syncing state, the in-memory state built up during the open and (to
77  * a lesser degree) the quiescing states is written to stable storage. The
78  * process of writing out modified data can, in turn modify more data. For
79  * example when we write new blocks, we need to allocate space for them; those
80  * allocations modify metadata (space maps)... which themselves must be
81  * written to stable storage. During the sync state, ZFS iterates, writing out
82  * data until it converges and all in-memory changes have been written out.
83  * The first such pass is the largest as it encompasses all the modified user
84  * data (as opposed to filesystem metadata). Subsequent passes typically have
85  * far less data to write as they consist exclusively of filesystem metadata.
86  *
87  * To ensure convergence, after a certain number of passes ZFS begins
88  * overwriting locations on stable storage that had been allocated earlier in
89  * the syncing state (and subsequently freed). ZFS usually allocates new
90  * blocks to optimize for large, continuous, writes. For the syncing state to
91  * converge however it must complete a pass where no new blocks are allocated
92  * since each allocation requires a modification of persistent metadata.
93  * Further, to hasten convergence, after a prescribed number of passes, ZFS
94  * also defers frees, and stops compressing.
95  *
96  * In addition to writing out user data, we must also execute synctasks during
97  * the syncing context. A synctask is the mechanism by which some
98  * administrative activities work such as creating and destroying snapshots or
99  * datasets. Note that when a synctask is initiated it enters the open txg,
100  * and ZFS then pushes that txg as quickly as possible to completion of the
101  * syncing state in order to reduce the latency of the administrative
102  * activity. To complete the syncing state, ZFS writes out a new uberblock,
103  * the root of the tree of blocks that comprise all state stored on the ZFS
104  * pool. Finally, if there is a quiesced txg waiting, we signal that it can
105  * now transition to the syncing state.
106  */
107 
108 static void txg_sync_thread(dsl_pool_t *dp);
109 static void txg_quiesce_thread(dsl_pool_t *dp);
110 
111 int zfs_txg_timeout = 5;	/* max seconds worth of delta per txg */
112 
113 /*
114  * Prepare the txg subsystem.
115  */
116 void
117 txg_init(dsl_pool_t *dp, uint64_t txg)
118 {
119 	tx_state_t *tx = &dp->dp_tx;
120 	int c;
121 	bzero(tx, sizeof (tx_state_t));
122 
123 	tx->tx_cpu = kmem_zalloc(max_ncpus * sizeof (tx_cpu_t), KM_SLEEP);
124 
125 	for (c = 0; c < max_ncpus; c++) {
126 		int i;
127 
128 		mutex_init(&tx->tx_cpu[c].tc_lock, NULL, MUTEX_DEFAULT, NULL);
129 		for (i = 0; i < TXG_SIZE; i++) {
130 			cv_init(&tx->tx_cpu[c].tc_cv[i], NULL, CV_DEFAULT,
131 			    NULL);
132 			list_create(&tx->tx_cpu[c].tc_callbacks[i],
133 			    sizeof (dmu_tx_callback_t),
134 			    offsetof(dmu_tx_callback_t, dcb_node));
135 		}
136 	}
137 
138 	mutex_init(&tx->tx_sync_lock, NULL, MUTEX_DEFAULT, NULL);
139 
140 	cv_init(&tx->tx_sync_more_cv, NULL, CV_DEFAULT, NULL);
141 	cv_init(&tx->tx_sync_done_cv, NULL, CV_DEFAULT, NULL);
142 	cv_init(&tx->tx_quiesce_more_cv, NULL, CV_DEFAULT, NULL);
143 	cv_init(&tx->tx_quiesce_done_cv, NULL, CV_DEFAULT, NULL);
144 	cv_init(&tx->tx_exit_cv, NULL, CV_DEFAULT, NULL);
145 
146 	tx->tx_open_txg = txg;
147 }
148 
149 /*
150  * Close down the txg subsystem.
151  */
152 void
153 txg_fini(dsl_pool_t *dp)
154 {
155 	tx_state_t *tx = &dp->dp_tx;
156 	int c;
157 
158 	ASSERT(tx->tx_threads == 0);
159 
160 	mutex_destroy(&tx->tx_sync_lock);
161 
162 	cv_destroy(&tx->tx_sync_more_cv);
163 	cv_destroy(&tx->tx_sync_done_cv);
164 	cv_destroy(&tx->tx_quiesce_more_cv);
165 	cv_destroy(&tx->tx_quiesce_done_cv);
166 	cv_destroy(&tx->tx_exit_cv);
167 
168 	for (c = 0; c < max_ncpus; c++) {
169 		int i;
170 
171 		mutex_destroy(&tx->tx_cpu[c].tc_lock);
172 		for (i = 0; i < TXG_SIZE; i++) {
173 			cv_destroy(&tx->tx_cpu[c].tc_cv[i]);
174 			list_destroy(&tx->tx_cpu[c].tc_callbacks[i]);
175 		}
176 	}
177 
178 	if (tx->tx_commit_cb_taskq != NULL)
179 		taskq_destroy(tx->tx_commit_cb_taskq);
180 
181 	kmem_free(tx->tx_cpu, max_ncpus * sizeof (tx_cpu_t));
182 
183 	bzero(tx, sizeof (tx_state_t));
184 }
185 
186 /*
187  * Start syncing transaction groups.
188  */
189 void
190 txg_sync_start(dsl_pool_t *dp)
191 {
192 	tx_state_t *tx = &dp->dp_tx;
193 
194 	mutex_enter(&tx->tx_sync_lock);
195 
196 	dprintf("pool %p\n", dp);
197 
198 	ASSERT(tx->tx_threads == 0);
199 
200 	tx->tx_threads = 2;
201 
202 	tx->tx_quiesce_thread = thread_create(NULL, 0, txg_quiesce_thread,
203 	    dp, 0, &p0, TS_RUN, minclsyspri);
204 
205 	/*
206 	 * The sync thread can need a larger-than-default stack size on
207 	 * 32-bit x86.  This is due in part to nested pools and
208 	 * scrub_visitbp() recursion.
209 	 */
210 	tx->tx_sync_thread = thread_create(NULL, 32<<10, txg_sync_thread,
211 	    dp, 0, &p0, TS_RUN, minclsyspri);
212 
213 	mutex_exit(&tx->tx_sync_lock);
214 }
215 
216 static void
217 txg_thread_enter(tx_state_t *tx, callb_cpr_t *cpr)
218 {
219 	CALLB_CPR_INIT(cpr, &tx->tx_sync_lock, callb_generic_cpr, FTAG);
220 	mutex_enter(&tx->tx_sync_lock);
221 }
222 
223 static void
224 txg_thread_exit(tx_state_t *tx, callb_cpr_t *cpr, kthread_t **tpp)
225 {
226 	ASSERT(*tpp != NULL);
227 	*tpp = NULL;
228 	tx->tx_threads--;
229 	cv_broadcast(&tx->tx_exit_cv);
230 	CALLB_CPR_EXIT(cpr);		/* drops &tx->tx_sync_lock */
231 	thread_exit();
232 }
233 
234 static void
235 txg_thread_wait(tx_state_t *tx, callb_cpr_t *cpr, kcondvar_t *cv, uint64_t time)
236 {
237 	CALLB_CPR_SAFE_BEGIN(cpr);
238 
239 	if (time)
240 		(void) cv_timedwait(cv, &tx->tx_sync_lock,
241 		    ddi_get_lbolt() + time);
242 	else
243 		cv_wait(cv, &tx->tx_sync_lock);
244 
245 	CALLB_CPR_SAFE_END(cpr, &tx->tx_sync_lock);
246 }
247 
248 /*
249  * Stop syncing transaction groups.
250  */
251 void
252 txg_sync_stop(dsl_pool_t *dp)
253 {
254 	tx_state_t *tx = &dp->dp_tx;
255 
256 	dprintf("pool %p\n", dp);
257 	/*
258 	 * Finish off any work in progress.
259 	 */
260 	ASSERT(tx->tx_threads == 2);
261 
262 	/*
263 	 * We need to ensure that we've vacated the deferred space_maps.
264 	 */
265 	txg_wait_synced(dp, tx->tx_open_txg + TXG_DEFER_SIZE);
266 
267 	/*
268 	 * Wake all sync threads and wait for them to die.
269 	 */
270 	mutex_enter(&tx->tx_sync_lock);
271 
272 	ASSERT(tx->tx_threads == 2);
273 
274 	tx->tx_exiting = 1;
275 
276 	cv_broadcast(&tx->tx_quiesce_more_cv);
277 	cv_broadcast(&tx->tx_quiesce_done_cv);
278 	cv_broadcast(&tx->tx_sync_more_cv);
279 
280 	while (tx->tx_threads != 0)
281 		cv_wait(&tx->tx_exit_cv, &tx->tx_sync_lock);
282 
283 	tx->tx_exiting = 0;
284 
285 	mutex_exit(&tx->tx_sync_lock);
286 }
287 
288 uint64_t
289 txg_hold_open(dsl_pool_t *dp, txg_handle_t *th)
290 {
291 	tx_state_t *tx = &dp->dp_tx;
292 	tx_cpu_t *tc = &tx->tx_cpu[CPU_SEQID];
293 	uint64_t txg;
294 
295 	mutex_enter(&tc->tc_lock);
296 
297 	txg = tx->tx_open_txg;
298 	tc->tc_count[txg & TXG_MASK]++;
299 
300 	th->th_cpu = tc;
301 	th->th_txg = txg;
302 
303 	return (txg);
304 }
305 
306 void
307 txg_rele_to_quiesce(txg_handle_t *th)
308 {
309 	tx_cpu_t *tc = th->th_cpu;
310 
311 	mutex_exit(&tc->tc_lock);
312 }
313 
314 void
315 txg_register_callbacks(txg_handle_t *th, list_t *tx_callbacks)
316 {
317 	tx_cpu_t *tc = th->th_cpu;
318 	int g = th->th_txg & TXG_MASK;
319 
320 	mutex_enter(&tc->tc_lock);
321 	list_move_tail(&tc->tc_callbacks[g], tx_callbacks);
322 	mutex_exit(&tc->tc_lock);
323 }
324 
325 void
326 txg_rele_to_sync(txg_handle_t *th)
327 {
328 	tx_cpu_t *tc = th->th_cpu;
329 	int g = th->th_txg & TXG_MASK;
330 
331 	mutex_enter(&tc->tc_lock);
332 	ASSERT(tc->tc_count[g] != 0);
333 	if (--tc->tc_count[g] == 0)
334 		cv_broadcast(&tc->tc_cv[g]);
335 	mutex_exit(&tc->tc_lock);
336 
337 	th->th_cpu = NULL;	/* defensive */
338 }
339 
340 static void
341 txg_quiesce(dsl_pool_t *dp, uint64_t txg)
342 {
343 	tx_state_t *tx = &dp->dp_tx;
344 	int g = txg & TXG_MASK;
345 	int c;
346 
347 	/*
348 	 * Grab all tx_cpu locks so nobody else can get into this txg.
349 	 */
350 	for (c = 0; c < max_ncpus; c++)
351 		mutex_enter(&tx->tx_cpu[c].tc_lock);
352 
353 	ASSERT(txg == tx->tx_open_txg);
354 	tx->tx_open_txg++;
355 
356 	/*
357 	 * Now that we've incremented tx_open_txg, we can let threads
358 	 * enter the next transaction group.
359 	 */
360 	for (c = 0; c < max_ncpus; c++)
361 		mutex_exit(&tx->tx_cpu[c].tc_lock);
362 
363 	/*
364 	 * Quiesce the transaction group by waiting for everyone to txg_exit().
365 	 */
366 	for (c = 0; c < max_ncpus; c++) {
367 		tx_cpu_t *tc = &tx->tx_cpu[c];
368 		mutex_enter(&tc->tc_lock);
369 		while (tc->tc_count[g] != 0)
370 			cv_wait(&tc->tc_cv[g], &tc->tc_lock);
371 		mutex_exit(&tc->tc_lock);
372 	}
373 }
374 
375 static void
376 txg_do_callbacks(list_t *cb_list)
377 {
378 	dmu_tx_do_callbacks(cb_list, 0);
379 
380 	list_destroy(cb_list);
381 
382 	kmem_free(cb_list, sizeof (list_t));
383 }
384 
385 /*
386  * Dispatch the commit callbacks registered on this txg to worker threads.
387  */
388 static void
389 txg_dispatch_callbacks(dsl_pool_t *dp, uint64_t txg)
390 {
391 	int c;
392 	tx_state_t *tx = &dp->dp_tx;
393 	list_t *cb_list;
394 
395 	for (c = 0; c < max_ncpus; c++) {
396 		tx_cpu_t *tc = &tx->tx_cpu[c];
397 		/* No need to lock tx_cpu_t at this point */
398 
399 		int g = txg & TXG_MASK;
400 
401 		if (list_is_empty(&tc->tc_callbacks[g]))
402 			continue;
403 
404 		if (tx->tx_commit_cb_taskq == NULL) {
405 			/*
406 			 * Commit callback taskq hasn't been created yet.
407 			 */
408 			tx->tx_commit_cb_taskq = taskq_create("tx_commit_cb",
409 			    max_ncpus, minclsyspri, max_ncpus, max_ncpus * 2,
410 			    TASKQ_PREPOPULATE);
411 		}
412 
413 		cb_list = kmem_alloc(sizeof (list_t), KM_SLEEP);
414 		list_create(cb_list, sizeof (dmu_tx_callback_t),
415 		    offsetof(dmu_tx_callback_t, dcb_node));
416 
417 		list_move_tail(&tc->tc_callbacks[g], cb_list);
418 
419 		(void) taskq_dispatch(tx->tx_commit_cb_taskq, (task_func_t *)
420 		    txg_do_callbacks, cb_list, TQ_SLEEP);
421 	}
422 }
423 
424 static void
425 txg_sync_thread(dsl_pool_t *dp)
426 {
427 	spa_t *spa = dp->dp_spa;
428 	tx_state_t *tx = &dp->dp_tx;
429 	callb_cpr_t cpr;
430 	uint64_t start, delta;
431 
432 	txg_thread_enter(tx, &cpr);
433 
434 	start = delta = 0;
435 	for (;;) {
436 		uint64_t timer, timeout = zfs_txg_timeout * hz;
437 		uint64_t txg;
438 
439 		/*
440 		 * We sync when we're scanning, there's someone waiting
441 		 * on us, or the quiesce thread has handed off a txg to
442 		 * us, or we have reached our timeout.
443 		 */
444 		timer = (delta >= timeout ? 0 : timeout - delta);
445 		while (!dsl_scan_active(dp->dp_scan) &&
446 		    !tx->tx_exiting && timer > 0 &&
447 		    tx->tx_synced_txg >= tx->tx_sync_txg_waiting &&
448 		    tx->tx_quiesced_txg == 0) {
449 			dprintf("waiting; tx_synced=%llu waiting=%llu dp=%p\n",
450 			    tx->tx_synced_txg, tx->tx_sync_txg_waiting, dp);
451 			txg_thread_wait(tx, &cpr, &tx->tx_sync_more_cv, timer);
452 			delta = ddi_get_lbolt() - start;
453 			timer = (delta > timeout ? 0 : timeout - delta);
454 		}
455 
456 		/*
457 		 * Wait until the quiesce thread hands off a txg to us,
458 		 * prompting it to do so if necessary.
459 		 */
460 		while (!tx->tx_exiting && tx->tx_quiesced_txg == 0) {
461 			if (tx->tx_quiesce_txg_waiting < tx->tx_open_txg+1)
462 				tx->tx_quiesce_txg_waiting = tx->tx_open_txg+1;
463 			cv_broadcast(&tx->tx_quiesce_more_cv);
464 			txg_thread_wait(tx, &cpr, &tx->tx_quiesce_done_cv, 0);
465 		}
466 
467 		if (tx->tx_exiting)
468 			txg_thread_exit(tx, &cpr, &tx->tx_sync_thread);
469 
470 		/*
471 		 * Consume the quiesced txg which has been handed off to
472 		 * us.  This may cause the quiescing thread to now be
473 		 * able to quiesce another txg, so we must signal it.
474 		 */
475 		txg = tx->tx_quiesced_txg;
476 		tx->tx_quiesced_txg = 0;
477 		tx->tx_syncing_txg = txg;
478 		cv_broadcast(&tx->tx_quiesce_more_cv);
479 
480 		dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
481 		    txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting);
482 		mutex_exit(&tx->tx_sync_lock);
483 
484 		start = ddi_get_lbolt();
485 		spa_sync(spa, txg);
486 		delta = ddi_get_lbolt() - start;
487 
488 		mutex_enter(&tx->tx_sync_lock);
489 		tx->tx_synced_txg = txg;
490 		tx->tx_syncing_txg = 0;
491 		cv_broadcast(&tx->tx_sync_done_cv);
492 
493 		/*
494 		 * Dispatch commit callbacks to worker threads.
495 		 */
496 		txg_dispatch_callbacks(dp, txg);
497 	}
498 }
499 
500 static void
501 txg_quiesce_thread(dsl_pool_t *dp)
502 {
503 	tx_state_t *tx = &dp->dp_tx;
504 	callb_cpr_t cpr;
505 
506 	txg_thread_enter(tx, &cpr);
507 
508 	for (;;) {
509 		uint64_t txg;
510 
511 		/*
512 		 * We quiesce when there's someone waiting on us.
513 		 * However, we can only have one txg in "quiescing" or
514 		 * "quiesced, waiting to sync" state.  So we wait until
515 		 * the "quiesced, waiting to sync" txg has been consumed
516 		 * by the sync thread.
517 		 */
518 		while (!tx->tx_exiting &&
519 		    (tx->tx_open_txg >= tx->tx_quiesce_txg_waiting ||
520 		    tx->tx_quiesced_txg != 0))
521 			txg_thread_wait(tx, &cpr, &tx->tx_quiesce_more_cv, 0);
522 
523 		if (tx->tx_exiting)
524 			txg_thread_exit(tx, &cpr, &tx->tx_quiesce_thread);
525 
526 		txg = tx->tx_open_txg;
527 		dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
528 		    txg, tx->tx_quiesce_txg_waiting,
529 		    tx->tx_sync_txg_waiting);
530 		mutex_exit(&tx->tx_sync_lock);
531 		txg_quiesce(dp, txg);
532 		mutex_enter(&tx->tx_sync_lock);
533 
534 		/*
535 		 * Hand this txg off to the sync thread.
536 		 */
537 		dprintf("quiesce done, handing off txg %llu\n", txg);
538 		tx->tx_quiesced_txg = txg;
539 		cv_broadcast(&tx->tx_sync_more_cv);
540 		cv_broadcast(&tx->tx_quiesce_done_cv);
541 	}
542 }
543 
544 /*
545  * Delay this thread by 'ticks' if we are still in the open transaction
546  * group and there is already a waiting txg quiesing or quiesced.  Abort
547  * the delay if this txg stalls or enters the quiesing state.
548  */
549 void
550 txg_delay(dsl_pool_t *dp, uint64_t txg, int ticks)
551 {
552 	tx_state_t *tx = &dp->dp_tx;
553 	clock_t timeout = ddi_get_lbolt() + ticks;
554 
555 	/* don't delay if this txg could transition to quiesing immediately */
556 	if (tx->tx_open_txg > txg ||
557 	    tx->tx_syncing_txg == txg-1 || tx->tx_synced_txg == txg-1)
558 		return;
559 
560 	mutex_enter(&tx->tx_sync_lock);
561 	if (tx->tx_open_txg > txg || tx->tx_synced_txg == txg-1) {
562 		mutex_exit(&tx->tx_sync_lock);
563 		return;
564 	}
565 
566 	while (ddi_get_lbolt() < timeout &&
567 	    tx->tx_syncing_txg < txg-1 && !txg_stalled(dp))
568 		(void) cv_timedwait(&tx->tx_quiesce_more_cv, &tx->tx_sync_lock,
569 		    timeout);
570 
571 	mutex_exit(&tx->tx_sync_lock);
572 }
573 
574 void
575 txg_wait_synced(dsl_pool_t *dp, uint64_t txg)
576 {
577 	tx_state_t *tx = &dp->dp_tx;
578 
579 	mutex_enter(&tx->tx_sync_lock);
580 	ASSERT(tx->tx_threads == 2);
581 	if (txg == 0)
582 		txg = tx->tx_open_txg + TXG_DEFER_SIZE;
583 	if (tx->tx_sync_txg_waiting < txg)
584 		tx->tx_sync_txg_waiting = txg;
585 	dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
586 	    txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting);
587 	while (tx->tx_synced_txg < txg) {
588 		dprintf("broadcasting sync more "
589 		    "tx_synced=%llu waiting=%llu dp=%p\n",
590 		    tx->tx_synced_txg, tx->tx_sync_txg_waiting, dp);
591 		cv_broadcast(&tx->tx_sync_more_cv);
592 		cv_wait(&tx->tx_sync_done_cv, &tx->tx_sync_lock);
593 	}
594 	mutex_exit(&tx->tx_sync_lock);
595 }
596 
597 void
598 txg_wait_open(dsl_pool_t *dp, uint64_t txg)
599 {
600 	tx_state_t *tx = &dp->dp_tx;
601 
602 	mutex_enter(&tx->tx_sync_lock);
603 	ASSERT(tx->tx_threads == 2);
604 	if (txg == 0)
605 		txg = tx->tx_open_txg + 1;
606 	if (tx->tx_quiesce_txg_waiting < txg)
607 		tx->tx_quiesce_txg_waiting = txg;
608 	dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
609 	    txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting);
610 	while (tx->tx_open_txg < txg) {
611 		cv_broadcast(&tx->tx_quiesce_more_cv);
612 		cv_wait(&tx->tx_quiesce_done_cv, &tx->tx_sync_lock);
613 	}
614 	mutex_exit(&tx->tx_sync_lock);
615 }
616 
617 boolean_t
618 txg_stalled(dsl_pool_t *dp)
619 {
620 	tx_state_t *tx = &dp->dp_tx;
621 	return (tx->tx_quiesce_txg_waiting > tx->tx_open_txg);
622 }
623 
624 boolean_t
625 txg_sync_waiting(dsl_pool_t *dp)
626 {
627 	tx_state_t *tx = &dp->dp_tx;
628 
629 	return (tx->tx_syncing_txg <= tx->tx_sync_txg_waiting ||
630 	    tx->tx_quiesced_txg != 0);
631 }
632 
633 /*
634  * Per-txg object lists.
635  */
636 void
637 txg_list_create(txg_list_t *tl, size_t offset)
638 {
639 	int t;
640 
641 	mutex_init(&tl->tl_lock, NULL, MUTEX_DEFAULT, NULL);
642 
643 	tl->tl_offset = offset;
644 
645 	for (t = 0; t < TXG_SIZE; t++)
646 		tl->tl_head[t] = NULL;
647 }
648 
649 void
650 txg_list_destroy(txg_list_t *tl)
651 {
652 	int t;
653 
654 	for (t = 0; t < TXG_SIZE; t++)
655 		ASSERT(txg_list_empty(tl, t));
656 
657 	mutex_destroy(&tl->tl_lock);
658 }
659 
660 boolean_t
661 txg_list_empty(txg_list_t *tl, uint64_t txg)
662 {
663 	return (tl->tl_head[txg & TXG_MASK] == NULL);
664 }
665 
666 /*
667  * Add an entry to the list.
668  * Returns 0 if it's a new entry, 1 if it's already there.
669  */
670 int
671 txg_list_add(txg_list_t *tl, void *p, uint64_t txg)
672 {
673 	int t = txg & TXG_MASK;
674 	txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
675 	int already_on_list;
676 
677 	mutex_enter(&tl->tl_lock);
678 	already_on_list = tn->tn_member[t];
679 	if (!already_on_list) {
680 		tn->tn_member[t] = 1;
681 		tn->tn_next[t] = tl->tl_head[t];
682 		tl->tl_head[t] = tn;
683 	}
684 	mutex_exit(&tl->tl_lock);
685 
686 	return (already_on_list);
687 }
688 
689 /*
690  * Add an entry to the end of the list (walks list to find end).
691  * Returns 0 if it's a new entry, 1 if it's already there.
692  */
693 int
694 txg_list_add_tail(txg_list_t *tl, void *p, uint64_t txg)
695 {
696 	int t = txg & TXG_MASK;
697 	txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
698 	int already_on_list;
699 
700 	mutex_enter(&tl->tl_lock);
701 	already_on_list = tn->tn_member[t];
702 	if (!already_on_list) {
703 		txg_node_t **tp;
704 
705 		for (tp = &tl->tl_head[t]; *tp != NULL; tp = &(*tp)->tn_next[t])
706 			continue;
707 
708 		tn->tn_member[t] = 1;
709 		tn->tn_next[t] = NULL;
710 		*tp = tn;
711 	}
712 	mutex_exit(&tl->tl_lock);
713 
714 	return (already_on_list);
715 }
716 
717 /*
718  * Remove the head of the list and return it.
719  */
720 void *
721 txg_list_remove(txg_list_t *tl, uint64_t txg)
722 {
723 	int t = txg & TXG_MASK;
724 	txg_node_t *tn;
725 	void *p = NULL;
726 
727 	mutex_enter(&tl->tl_lock);
728 	if ((tn = tl->tl_head[t]) != NULL) {
729 		p = (char *)tn - tl->tl_offset;
730 		tl->tl_head[t] = tn->tn_next[t];
731 		tn->tn_next[t] = NULL;
732 		tn->tn_member[t] = 0;
733 	}
734 	mutex_exit(&tl->tl_lock);
735 
736 	return (p);
737 }
738 
739 /*
740  * Remove a specific item from the list and return it.
741  */
742 void *
743 txg_list_remove_this(txg_list_t *tl, void *p, uint64_t txg)
744 {
745 	int t = txg & TXG_MASK;
746 	txg_node_t *tn, **tp;
747 
748 	mutex_enter(&tl->tl_lock);
749 
750 	for (tp = &tl->tl_head[t]; (tn = *tp) != NULL; tp = &tn->tn_next[t]) {
751 		if ((char *)tn - tl->tl_offset == p) {
752 			*tp = tn->tn_next[t];
753 			tn->tn_next[t] = NULL;
754 			tn->tn_member[t] = 0;
755 			mutex_exit(&tl->tl_lock);
756 			return (p);
757 		}
758 	}
759 
760 	mutex_exit(&tl->tl_lock);
761 
762 	return (NULL);
763 }
764 
765 int
766 txg_list_member(txg_list_t *tl, void *p, uint64_t txg)
767 {
768 	int t = txg & TXG_MASK;
769 	txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
770 
771 	return (tn->tn_member[t]);
772 }
773 
774 /*
775  * Walk a txg list -- only safe if you know it's not changing.
776  */
777 void *
778 txg_list_head(txg_list_t *tl, uint64_t txg)
779 {
780 	int t = txg & TXG_MASK;
781 	txg_node_t *tn = tl->tl_head[t];
782 
783 	return (tn == NULL ? NULL : (char *)tn - tl->tl_offset);
784 }
785 
786 void *
787 txg_list_next(txg_list_t *tl, void *p, uint64_t txg)
788 {
789 	int t = txg & TXG_MASK;
790 	txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
791 
792 	tn = tn->tn_next[t];
793 
794 	return (tn == NULL ? NULL : (char *)tn - tl->tl_offset);
795 }
796