xref: /illumos-gate/usr/src/uts/common/fs/zfs/txg.c (revision 0689f76c08c5e553ff25ac43a852b56c430bb61e)
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, clock_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 	DTRACE_PROBE2(txg__quiescing, dsl_pool_t *, dp, uint64_t, txg);
357 	DTRACE_PROBE2(txg__opened, dsl_pool_t *, dp, uint64_t, tx->tx_open_txg);
358 
359 	/*
360 	 * Now that we've incremented tx_open_txg, we can let threads
361 	 * enter the next transaction group.
362 	 */
363 	for (c = 0; c < max_ncpus; c++)
364 		mutex_exit(&tx->tx_cpu[c].tc_lock);
365 
366 	/*
367 	 * Quiesce the transaction group by waiting for everyone to txg_exit().
368 	 */
369 	for (c = 0; c < max_ncpus; c++) {
370 		tx_cpu_t *tc = &tx->tx_cpu[c];
371 		mutex_enter(&tc->tc_lock);
372 		while (tc->tc_count[g] != 0)
373 			cv_wait(&tc->tc_cv[g], &tc->tc_lock);
374 		mutex_exit(&tc->tc_lock);
375 	}
376 }
377 
378 static void
379 txg_do_callbacks(list_t *cb_list)
380 {
381 	dmu_tx_do_callbacks(cb_list, 0);
382 
383 	list_destroy(cb_list);
384 
385 	kmem_free(cb_list, sizeof (list_t));
386 }
387 
388 /*
389  * Dispatch the commit callbacks registered on this txg to worker threads.
390  */
391 static void
392 txg_dispatch_callbacks(dsl_pool_t *dp, uint64_t txg)
393 {
394 	int c;
395 	tx_state_t *tx = &dp->dp_tx;
396 	list_t *cb_list;
397 
398 	for (c = 0; c < max_ncpus; c++) {
399 		tx_cpu_t *tc = &tx->tx_cpu[c];
400 		/* No need to lock tx_cpu_t at this point */
401 
402 		int g = txg & TXG_MASK;
403 
404 		if (list_is_empty(&tc->tc_callbacks[g]))
405 			continue;
406 
407 		if (tx->tx_commit_cb_taskq == NULL) {
408 			/*
409 			 * Commit callback taskq hasn't been created yet.
410 			 */
411 			tx->tx_commit_cb_taskq = taskq_create("tx_commit_cb",
412 			    max_ncpus, minclsyspri, max_ncpus, max_ncpus * 2,
413 			    TASKQ_PREPOPULATE);
414 		}
415 
416 		cb_list = kmem_alloc(sizeof (list_t), KM_SLEEP);
417 		list_create(cb_list, sizeof (dmu_tx_callback_t),
418 		    offsetof(dmu_tx_callback_t, dcb_node));
419 
420 		list_move_tail(&tc->tc_callbacks[g], cb_list);
421 
422 		(void) taskq_dispatch(tx->tx_commit_cb_taskq, (task_func_t *)
423 		    txg_do_callbacks, cb_list, TQ_SLEEP);
424 	}
425 }
426 
427 static void
428 txg_sync_thread(dsl_pool_t *dp)
429 {
430 	spa_t *spa = dp->dp_spa;
431 	tx_state_t *tx = &dp->dp_tx;
432 	callb_cpr_t cpr;
433 	uint64_t start, delta;
434 
435 	txg_thread_enter(tx, &cpr);
436 
437 	start = delta = 0;
438 	for (;;) {
439 		uint64_t timer, timeout = zfs_txg_timeout * hz;
440 		uint64_t txg;
441 
442 		/*
443 		 * We sync when we're scanning, there's someone waiting
444 		 * on us, or the quiesce thread has handed off a txg to
445 		 * us, or we have reached our timeout.
446 		 */
447 		timer = (delta >= timeout ? 0 : timeout - delta);
448 		while (!dsl_scan_active(dp->dp_scan) &&
449 		    !tx->tx_exiting && timer > 0 &&
450 		    tx->tx_synced_txg >= tx->tx_sync_txg_waiting &&
451 		    tx->tx_quiesced_txg == 0) {
452 			dprintf("waiting; tx_synced=%llu waiting=%llu dp=%p\n",
453 			    tx->tx_synced_txg, tx->tx_sync_txg_waiting, dp);
454 			txg_thread_wait(tx, &cpr, &tx->tx_sync_more_cv, timer);
455 			delta = ddi_get_lbolt() - start;
456 			timer = (delta > timeout ? 0 : timeout - delta);
457 		}
458 
459 		/*
460 		 * Wait until the quiesce thread hands off a txg to us,
461 		 * prompting it to do so if necessary.
462 		 */
463 		while (!tx->tx_exiting && tx->tx_quiesced_txg == 0) {
464 			if (tx->tx_quiesce_txg_waiting < tx->tx_open_txg+1)
465 				tx->tx_quiesce_txg_waiting = tx->tx_open_txg+1;
466 			cv_broadcast(&tx->tx_quiesce_more_cv);
467 			txg_thread_wait(tx, &cpr, &tx->tx_quiesce_done_cv, 0);
468 		}
469 
470 		if (tx->tx_exiting)
471 			txg_thread_exit(tx, &cpr, &tx->tx_sync_thread);
472 
473 		/*
474 		 * Consume the quiesced txg which has been handed off to
475 		 * us.  This may cause the quiescing thread to now be
476 		 * able to quiesce another txg, so we must signal it.
477 		 */
478 		txg = tx->tx_quiesced_txg;
479 		tx->tx_quiesced_txg = 0;
480 		tx->tx_syncing_txg = txg;
481 		DTRACE_PROBE2(txg__syncing, dsl_pool_t *, dp, uint64_t, txg);
482 		cv_broadcast(&tx->tx_quiesce_more_cv);
483 
484 		dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
485 		    txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting);
486 		mutex_exit(&tx->tx_sync_lock);
487 
488 		start = ddi_get_lbolt();
489 		spa_sync(spa, txg);
490 		delta = ddi_get_lbolt() - start;
491 
492 		mutex_enter(&tx->tx_sync_lock);
493 		tx->tx_synced_txg = txg;
494 		tx->tx_syncing_txg = 0;
495 		DTRACE_PROBE2(txg__synced, dsl_pool_t *, dp, uint64_t, txg);
496 		cv_broadcast(&tx->tx_sync_done_cv);
497 
498 		/*
499 		 * Dispatch commit callbacks to worker threads.
500 		 */
501 		txg_dispatch_callbacks(dp, txg);
502 	}
503 }
504 
505 static void
506 txg_quiesce_thread(dsl_pool_t *dp)
507 {
508 	tx_state_t *tx = &dp->dp_tx;
509 	callb_cpr_t cpr;
510 
511 	txg_thread_enter(tx, &cpr);
512 
513 	for (;;) {
514 		uint64_t txg;
515 
516 		/*
517 		 * We quiesce when there's someone waiting on us.
518 		 * However, we can only have one txg in "quiescing" or
519 		 * "quiesced, waiting to sync" state.  So we wait until
520 		 * the "quiesced, waiting to sync" txg has been consumed
521 		 * by the sync thread.
522 		 */
523 		while (!tx->tx_exiting &&
524 		    (tx->tx_open_txg >= tx->tx_quiesce_txg_waiting ||
525 		    tx->tx_quiesced_txg != 0))
526 			txg_thread_wait(tx, &cpr, &tx->tx_quiesce_more_cv, 0);
527 
528 		if (tx->tx_exiting)
529 			txg_thread_exit(tx, &cpr, &tx->tx_quiesce_thread);
530 
531 		txg = tx->tx_open_txg;
532 		dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
533 		    txg, tx->tx_quiesce_txg_waiting,
534 		    tx->tx_sync_txg_waiting);
535 		mutex_exit(&tx->tx_sync_lock);
536 		txg_quiesce(dp, txg);
537 		mutex_enter(&tx->tx_sync_lock);
538 
539 		/*
540 		 * Hand this txg off to the sync thread.
541 		 */
542 		dprintf("quiesce done, handing off txg %llu\n", txg);
543 		tx->tx_quiesced_txg = txg;
544 		DTRACE_PROBE2(txg__quiesced, dsl_pool_t *, dp, uint64_t, txg);
545 		cv_broadcast(&tx->tx_sync_more_cv);
546 		cv_broadcast(&tx->tx_quiesce_done_cv);
547 	}
548 }
549 
550 /*
551  * Delay this thread by delay nanoseconds if we are still in the open
552  * transaction group and there is already a waiting txg quiesing or quiesced.
553  * Abort the delay if this txg stalls or enters the quiesing state.
554  */
555 void
556 txg_delay(dsl_pool_t *dp, uint64_t txg, hrtime_t delay, hrtime_t resolution)
557 {
558 	tx_state_t *tx = &dp->dp_tx;
559 	hrtime_t start = gethrtime();
560 
561 	/* don't delay if this txg could transition to quiesing immediately */
562 	if (tx->tx_open_txg > txg ||
563 	    tx->tx_syncing_txg == txg-1 || tx->tx_synced_txg == txg-1)
564 		return;
565 
566 	mutex_enter(&tx->tx_sync_lock);
567 	if (tx->tx_open_txg > txg || tx->tx_synced_txg == txg-1) {
568 		mutex_exit(&tx->tx_sync_lock);
569 		return;
570 	}
571 
572 	while (gethrtime() - start < delay &&
573 	    tx->tx_syncing_txg < txg-1 && !txg_stalled(dp)) {
574 		(void) cv_timedwait_hires(&tx->tx_quiesce_more_cv,
575 		    &tx->tx_sync_lock, delay, resolution, 0);
576 	}
577 
578 	mutex_exit(&tx->tx_sync_lock);
579 }
580 
581 void
582 txg_wait_synced(dsl_pool_t *dp, uint64_t txg)
583 {
584 	tx_state_t *tx = &dp->dp_tx;
585 
586 	ASSERT(!dsl_pool_config_held(dp));
587 
588 	mutex_enter(&tx->tx_sync_lock);
589 	ASSERT(tx->tx_threads == 2);
590 	if (txg == 0)
591 		txg = tx->tx_open_txg + TXG_DEFER_SIZE;
592 	if (tx->tx_sync_txg_waiting < txg)
593 		tx->tx_sync_txg_waiting = txg;
594 	dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
595 	    txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting);
596 	while (tx->tx_synced_txg < txg) {
597 		dprintf("broadcasting sync more "
598 		    "tx_synced=%llu waiting=%llu dp=%p\n",
599 		    tx->tx_synced_txg, tx->tx_sync_txg_waiting, dp);
600 		cv_broadcast(&tx->tx_sync_more_cv);
601 		cv_wait(&tx->tx_sync_done_cv, &tx->tx_sync_lock);
602 	}
603 	mutex_exit(&tx->tx_sync_lock);
604 }
605 
606 void
607 txg_wait_open(dsl_pool_t *dp, uint64_t txg)
608 {
609 	tx_state_t *tx = &dp->dp_tx;
610 
611 	ASSERT(!dsl_pool_config_held(dp));
612 
613 	mutex_enter(&tx->tx_sync_lock);
614 	ASSERT(tx->tx_threads == 2);
615 	if (txg == 0)
616 		txg = tx->tx_open_txg + 1;
617 	if (tx->tx_quiesce_txg_waiting < txg)
618 		tx->tx_quiesce_txg_waiting = txg;
619 	dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
620 	    txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting);
621 	while (tx->tx_open_txg < txg) {
622 		cv_broadcast(&tx->tx_quiesce_more_cv);
623 		cv_wait(&tx->tx_quiesce_done_cv, &tx->tx_sync_lock);
624 	}
625 	mutex_exit(&tx->tx_sync_lock);
626 }
627 
628 boolean_t
629 txg_stalled(dsl_pool_t *dp)
630 {
631 	tx_state_t *tx = &dp->dp_tx;
632 	return (tx->tx_quiesce_txg_waiting > tx->tx_open_txg);
633 }
634 
635 boolean_t
636 txg_sync_waiting(dsl_pool_t *dp)
637 {
638 	tx_state_t *tx = &dp->dp_tx;
639 
640 	return (tx->tx_syncing_txg <= tx->tx_sync_txg_waiting ||
641 	    tx->tx_quiesced_txg != 0);
642 }
643 
644 /*
645  * Per-txg object lists.
646  */
647 void
648 txg_list_create(txg_list_t *tl, size_t offset)
649 {
650 	int t;
651 
652 	mutex_init(&tl->tl_lock, NULL, MUTEX_DEFAULT, NULL);
653 
654 	tl->tl_offset = offset;
655 
656 	for (t = 0; t < TXG_SIZE; t++)
657 		tl->tl_head[t] = NULL;
658 }
659 
660 void
661 txg_list_destroy(txg_list_t *tl)
662 {
663 	int t;
664 
665 	for (t = 0; t < TXG_SIZE; t++)
666 		ASSERT(txg_list_empty(tl, t));
667 
668 	mutex_destroy(&tl->tl_lock);
669 }
670 
671 boolean_t
672 txg_list_empty(txg_list_t *tl, uint64_t txg)
673 {
674 	return (tl->tl_head[txg & TXG_MASK] == NULL);
675 }
676 
677 /*
678  * Add an entry to the list (unless it's already on the list).
679  * Returns B_TRUE if it was actually added.
680  */
681 boolean_t
682 txg_list_add(txg_list_t *tl, void *p, uint64_t txg)
683 {
684 	int t = txg & TXG_MASK;
685 	txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
686 	boolean_t add;
687 
688 	mutex_enter(&tl->tl_lock);
689 	add = (tn->tn_member[t] == 0);
690 	if (add) {
691 		tn->tn_member[t] = 1;
692 		tn->tn_next[t] = tl->tl_head[t];
693 		tl->tl_head[t] = tn;
694 	}
695 	mutex_exit(&tl->tl_lock);
696 
697 	return (add);
698 }
699 
700 /*
701  * Add an entry to the end of the list, unless it's already on the list.
702  * (walks list to find end)
703  * Returns B_TRUE if it was actually added.
704  */
705 boolean_t
706 txg_list_add_tail(txg_list_t *tl, void *p, uint64_t txg)
707 {
708 	int t = txg & TXG_MASK;
709 	txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
710 	boolean_t add;
711 
712 	mutex_enter(&tl->tl_lock);
713 	add = (tn->tn_member[t] == 0);
714 	if (add) {
715 		txg_node_t **tp;
716 
717 		for (tp = &tl->tl_head[t]; *tp != NULL; tp = &(*tp)->tn_next[t])
718 			continue;
719 
720 		tn->tn_member[t] = 1;
721 		tn->tn_next[t] = NULL;
722 		*tp = tn;
723 	}
724 	mutex_exit(&tl->tl_lock);
725 
726 	return (add);
727 }
728 
729 /*
730  * Remove the head of the list and return it.
731  */
732 void *
733 txg_list_remove(txg_list_t *tl, uint64_t txg)
734 {
735 	int t = txg & TXG_MASK;
736 	txg_node_t *tn;
737 	void *p = NULL;
738 
739 	mutex_enter(&tl->tl_lock);
740 	if ((tn = tl->tl_head[t]) != NULL) {
741 		p = (char *)tn - tl->tl_offset;
742 		tl->tl_head[t] = tn->tn_next[t];
743 		tn->tn_next[t] = NULL;
744 		tn->tn_member[t] = 0;
745 	}
746 	mutex_exit(&tl->tl_lock);
747 
748 	return (p);
749 }
750 
751 /*
752  * Remove a specific item from the list and return it.
753  */
754 void *
755 txg_list_remove_this(txg_list_t *tl, void *p, uint64_t txg)
756 {
757 	int t = txg & TXG_MASK;
758 	txg_node_t *tn, **tp;
759 
760 	mutex_enter(&tl->tl_lock);
761 
762 	for (tp = &tl->tl_head[t]; (tn = *tp) != NULL; tp = &tn->tn_next[t]) {
763 		if ((char *)tn - tl->tl_offset == p) {
764 			*tp = tn->tn_next[t];
765 			tn->tn_next[t] = NULL;
766 			tn->tn_member[t] = 0;
767 			mutex_exit(&tl->tl_lock);
768 			return (p);
769 		}
770 	}
771 
772 	mutex_exit(&tl->tl_lock);
773 
774 	return (NULL);
775 }
776 
777 boolean_t
778 txg_list_member(txg_list_t *tl, void *p, uint64_t txg)
779 {
780 	int t = txg & TXG_MASK;
781 	txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
782 
783 	return (tn->tn_member[t] != 0);
784 }
785 
786 /*
787  * Walk a txg list -- only safe if you know it's not changing.
788  */
789 void *
790 txg_list_head(txg_list_t *tl, uint64_t txg)
791 {
792 	int t = txg & TXG_MASK;
793 	txg_node_t *tn = tl->tl_head[t];
794 
795 	return (tn == NULL ? NULL : (char *)tn - tl->tl_offset);
796 }
797 
798 void *
799 txg_list_next(txg_list_t *tl, void *p, uint64_t txg)
800 {
801 	int t = txg & TXG_MASK;
802 	txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
803 
804 	tn = tn->tn_next[t];
805 
806 	return (tn == NULL ? NULL : (char *)tn - tl->tl_offset);
807 }
808