Fix zfs_txg_timeout module parameter
[zfs.git] / module / zfs / txg.c
1 /*
2  * CDDL HEADER START
3  *
4  * The contents of this file are subject to the terms of the
5  * Common Development and Distribution License (the "License").
6  * You may not use this file except in compliance with the License.
7  *
8  * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
9  * or http://www.opensolaris.org/os/licensing.
10  * See the License for the specific language governing permissions
11  * and limitations under the License.
12  *
13  * When distributing Covered Code, include this CDDL HEADER in each
14  * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
15  * If applicable, add the following below this CDDL HEADER, with the
16  * fields enclosed by brackets "[]" replaced with your own identifying
17  * information: Portions Copyright [yyyy] [name of copyright owner]
18  *
19  * CDDL HEADER END
20  */
21 /*
22  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
23  */
24
25 #include <sys/zfs_context.h>
26 #include <sys/txg_impl.h>
27 #include <sys/dmu_impl.h>
28 #include <sys/dmu_tx.h>
29 #include <sys/dsl_pool.h>
30 #include <sys/dsl_scan.h>
31 #include <sys/callb.h>
32
33 /*
34  * Pool-wide transaction groups.
35  */
36
37 static void txg_sync_thread(dsl_pool_t *dp);
38 static void txg_quiesce_thread(dsl_pool_t *dp);
39
40 int zfs_txg_timeout = 5;        /* max seconds worth of delta per txg */
41
42 /*
43  * Prepare the txg subsystem.
44  */
45 void
46 txg_init(dsl_pool_t *dp, uint64_t txg)
47 {
48         tx_state_t *tx = &dp->dp_tx;
49         int c;
50         bzero(tx, sizeof (tx_state_t));
51
52         tx->tx_cpu = vmem_zalloc(max_ncpus * sizeof (tx_cpu_t), KM_SLEEP);
53
54         for (c = 0; c < max_ncpus; c++) {
55                 int i;
56
57                 mutex_init(&tx->tx_cpu[c].tc_lock, NULL, MUTEX_DEFAULT, NULL);
58                 for (i = 0; i < TXG_SIZE; i++) {
59                         cv_init(&tx->tx_cpu[c].tc_cv[i], NULL, CV_DEFAULT,
60                             NULL);
61                         list_create(&tx->tx_cpu[c].tc_callbacks[i],
62                             sizeof (dmu_tx_callback_t),
63                             offsetof(dmu_tx_callback_t, dcb_node));
64                 }
65         }
66
67         mutex_init(&tx->tx_sync_lock, NULL, MUTEX_DEFAULT, NULL);
68
69         cv_init(&tx->tx_sync_more_cv, NULL, CV_DEFAULT, NULL);
70         cv_init(&tx->tx_sync_done_cv, NULL, CV_DEFAULT, NULL);
71         cv_init(&tx->tx_quiesce_more_cv, NULL, CV_DEFAULT, NULL);
72         cv_init(&tx->tx_quiesce_done_cv, NULL, CV_DEFAULT, NULL);
73         cv_init(&tx->tx_exit_cv, NULL, CV_DEFAULT, NULL);
74
75         tx->tx_open_txg = txg;
76 }
77
78 /*
79  * Close down the txg subsystem.
80  */
81 void
82 txg_fini(dsl_pool_t *dp)
83 {
84         tx_state_t *tx = &dp->dp_tx;
85         int c;
86
87         ASSERT(tx->tx_threads == 0);
88
89         mutex_destroy(&tx->tx_sync_lock);
90
91         cv_destroy(&tx->tx_sync_more_cv);
92         cv_destroy(&tx->tx_sync_done_cv);
93         cv_destroy(&tx->tx_quiesce_more_cv);
94         cv_destroy(&tx->tx_quiesce_done_cv);
95         cv_destroy(&tx->tx_exit_cv);
96
97         for (c = 0; c < max_ncpus; c++) {
98                 int i;
99
100                 mutex_destroy(&tx->tx_cpu[c].tc_lock);
101                 for (i = 0; i < TXG_SIZE; i++) {
102                         cv_destroy(&tx->tx_cpu[c].tc_cv[i]);
103                         list_destroy(&tx->tx_cpu[c].tc_callbacks[i]);
104                 }
105         }
106
107         if (tx->tx_commit_cb_taskq != NULL)
108                 taskq_destroy(tx->tx_commit_cb_taskq);
109
110         vmem_free(tx->tx_cpu, max_ncpus * sizeof (tx_cpu_t));
111
112         bzero(tx, sizeof (tx_state_t));
113 }
114
115 /*
116  * Start syncing transaction groups.
117  */
118 void
119 txg_sync_start(dsl_pool_t *dp)
120 {
121         tx_state_t *tx = &dp->dp_tx;
122
123         mutex_enter(&tx->tx_sync_lock);
124
125         dprintf("pool %p\n", dp);
126
127         ASSERT(tx->tx_threads == 0);
128
129         tx->tx_threads = 2;
130
131         tx->tx_quiesce_thread = thread_create(NULL, 0, txg_quiesce_thread,
132             dp, 0, &p0, TS_RUN, minclsyspri);
133
134         /*
135          * The sync thread can need a larger-than-default stack size on
136          * 32-bit x86.  This is due in part to nested pools and
137          * scrub_visitbp() recursion.
138          */
139         tx->tx_sync_thread = thread_create(NULL, 32<<10, txg_sync_thread,
140             dp, 0, &p0, TS_RUN, minclsyspri);
141
142         mutex_exit(&tx->tx_sync_lock);
143 }
144
145 static void
146 txg_thread_enter(tx_state_t *tx, callb_cpr_t *cpr)
147 {
148         CALLB_CPR_INIT(cpr, &tx->tx_sync_lock, callb_generic_cpr, FTAG);
149         mutex_enter(&tx->tx_sync_lock);
150 }
151
152 static void
153 txg_thread_exit(tx_state_t *tx, callb_cpr_t *cpr, kthread_t **tpp)
154 {
155         ASSERT(*tpp != NULL);
156         *tpp = NULL;
157         tx->tx_threads--;
158         cv_broadcast(&tx->tx_exit_cv);
159         CALLB_CPR_EXIT(cpr);            /* drops &tx->tx_sync_lock */
160         thread_exit();
161 }
162
163 static void
164 txg_thread_wait(tx_state_t *tx, callb_cpr_t *cpr, kcondvar_t *cv, uint64_t time)
165 {
166         CALLB_CPR_SAFE_BEGIN(cpr);
167
168         if (time)
169                 (void) cv_timedwait_interruptible(cv, &tx->tx_sync_lock,
170                     ddi_get_lbolt() + time);
171         else
172                 cv_wait_interruptible(cv, &tx->tx_sync_lock);
173
174         CALLB_CPR_SAFE_END(cpr, &tx->tx_sync_lock);
175 }
176
177 /*
178  * Stop syncing transaction groups.
179  */
180 void
181 txg_sync_stop(dsl_pool_t *dp)
182 {
183         tx_state_t *tx = &dp->dp_tx;
184
185         dprintf("pool %p\n", dp);
186         /*
187          * Finish off any work in progress.
188          */
189         ASSERT(tx->tx_threads == 2);
190
191         /*
192          * We need to ensure that we've vacated the deferred space_maps.
193          */
194         txg_wait_synced(dp, tx->tx_open_txg + TXG_DEFER_SIZE);
195
196         /*
197          * Wake all sync threads and wait for them to die.
198          */
199         mutex_enter(&tx->tx_sync_lock);
200
201         ASSERT(tx->tx_threads == 2);
202
203         tx->tx_exiting = 1;
204
205         cv_broadcast(&tx->tx_quiesce_more_cv);
206         cv_broadcast(&tx->tx_quiesce_done_cv);
207         cv_broadcast(&tx->tx_sync_more_cv);
208
209         while (tx->tx_threads != 0)
210                 cv_wait(&tx->tx_exit_cv, &tx->tx_sync_lock);
211
212         tx->tx_exiting = 0;
213
214         mutex_exit(&tx->tx_sync_lock);
215 }
216
217 uint64_t
218 txg_hold_open(dsl_pool_t *dp, txg_handle_t *th)
219 {
220         tx_state_t *tx = &dp->dp_tx;
221         tx_cpu_t *tc;
222         uint64_t txg;
223
224         /*
225          * It appears the processor id is simply used as a "random"
226          * number to index into the array, and there isn't any other
227          * significance to the chosen tx_cpu. Because.. Why not use
228          * the current cpu to index into the array?
229          */
230         kpreempt_disable();
231         tc = &tx->tx_cpu[CPU_SEQID];
232         kpreempt_enable();
233
234         mutex_enter(&tc->tc_lock);
235
236         txg = tx->tx_open_txg;
237         tc->tc_count[txg & TXG_MASK]++;
238
239         th->th_cpu = tc;
240         th->th_txg = txg;
241
242         return (txg);
243 }
244
245 void
246 txg_rele_to_quiesce(txg_handle_t *th)
247 {
248         tx_cpu_t *tc = th->th_cpu;
249
250         mutex_exit(&tc->tc_lock);
251 }
252
253 void
254 txg_register_callbacks(txg_handle_t *th, list_t *tx_callbacks)
255 {
256         tx_cpu_t *tc = th->th_cpu;
257         int g = th->th_txg & TXG_MASK;
258
259         mutex_enter(&tc->tc_lock);
260         list_move_tail(&tc->tc_callbacks[g], tx_callbacks);
261         mutex_exit(&tc->tc_lock);
262 }
263
264 void
265 txg_rele_to_sync(txg_handle_t *th)
266 {
267         tx_cpu_t *tc = th->th_cpu;
268         int g = th->th_txg & TXG_MASK;
269
270         mutex_enter(&tc->tc_lock);
271         ASSERT(tc->tc_count[g] != 0);
272         if (--tc->tc_count[g] == 0)
273                 cv_broadcast(&tc->tc_cv[g]);
274         mutex_exit(&tc->tc_lock);
275
276         th->th_cpu = NULL;      /* defensive */
277 }
278
279 static void
280 txg_quiesce(dsl_pool_t *dp, uint64_t txg)
281 {
282         tx_state_t *tx = &dp->dp_tx;
283         int g = txg & TXG_MASK;
284         int c;
285
286         /*
287          * Grab all tx_cpu locks so nobody else can get into this txg.
288          */
289         for (c = 0; c < max_ncpus; c++)
290                 mutex_enter(&tx->tx_cpu[c].tc_lock);
291
292         ASSERT(txg == tx->tx_open_txg);
293         tx->tx_open_txg++;
294
295         /*
296          * Now that we've incremented tx_open_txg, we can let threads
297          * enter the next transaction group.
298          */
299         for (c = 0; c < max_ncpus; c++)
300                 mutex_exit(&tx->tx_cpu[c].tc_lock);
301
302         /*
303          * Quiesce the transaction group by waiting for everyone to txg_exit().
304          */
305         for (c = 0; c < max_ncpus; c++) {
306                 tx_cpu_t *tc = &tx->tx_cpu[c];
307                 mutex_enter(&tc->tc_lock);
308                 while (tc->tc_count[g] != 0)
309                         cv_wait(&tc->tc_cv[g], &tc->tc_lock);
310                 mutex_exit(&tc->tc_lock);
311         }
312 }
313
314 static void
315 txg_do_callbacks(list_t *cb_list)
316 {
317         dmu_tx_do_callbacks(cb_list, 0);
318
319         list_destroy(cb_list);
320
321         kmem_free(cb_list, sizeof (list_t));
322 }
323
324 /*
325  * Dispatch the commit callbacks registered on this txg to worker threads.
326  */
327 static void
328 txg_dispatch_callbacks(dsl_pool_t *dp, uint64_t txg)
329 {
330         int c;
331         tx_state_t *tx = &dp->dp_tx;
332         list_t *cb_list;
333
334         for (c = 0; c < max_ncpus; c++) {
335                 tx_cpu_t *tc = &tx->tx_cpu[c];
336                 /* No need to lock tx_cpu_t at this point */
337
338                 int g = txg & TXG_MASK;
339
340                 if (list_is_empty(&tc->tc_callbacks[g]))
341                         continue;
342
343                 if (tx->tx_commit_cb_taskq == NULL) {
344                         /*
345                          * Commit callback taskq hasn't been created yet.
346                          */
347                         tx->tx_commit_cb_taskq = taskq_create("tx_commit_cb",
348                             100, minclsyspri, max_ncpus, INT_MAX,
349                             TASKQ_THREADS_CPU_PCT | TASKQ_PREPOPULATE);
350                 }
351
352                 cb_list = kmem_alloc(sizeof (list_t), KM_PUSHPAGE);
353                 list_create(cb_list, sizeof (dmu_tx_callback_t),
354                     offsetof(dmu_tx_callback_t, dcb_node));
355
356                 list_move_tail(cb_list, &tc->tc_callbacks[g]);
357
358                 (void) taskq_dispatch(tx->tx_commit_cb_taskq, (task_func_t *)
359                     txg_do_callbacks, cb_list, TQ_SLEEP);
360         }
361 }
362
363 /*
364  * Wait for pending commit callbacks of already-synced transactions to finish
365  * processing.
366  * Calling this function from within a commit callback will deadlock.
367  */
368 void
369 txg_wait_callbacks(dsl_pool_t *dp)
370 {
371         tx_state_t *tx = &dp->dp_tx;
372
373         if (tx->tx_commit_cb_taskq != NULL)
374                 taskq_wait(tx->tx_commit_cb_taskq);
375 }
376
377 static void
378 txg_sync_thread(dsl_pool_t *dp)
379 {
380         spa_t *spa = dp->dp_spa;
381         tx_state_t *tx = &dp->dp_tx;
382         callb_cpr_t cpr;
383         uint64_t start, delta;
384
385 #ifdef _KERNEL
386         /*
387          * Annotate this process with a flag that indicates that it is
388          * unsafe to use KM_SLEEP during memory allocations due to the
389          * potential for a deadlock.  KM_PUSHPAGE should be used instead.
390          */
391         current->flags |= PF_NOFS;
392 #endif /* _KERNEL */
393
394         txg_thread_enter(tx, &cpr);
395
396         start = delta = 0;
397         for (;;) {
398                 uint64_t timer, timeout;
399                 uint64_t txg;
400
401                 timeout = zfs_txg_timeout * hz;
402
403                 /*
404                  * We sync when we're scanning, there's someone waiting
405                  * on us, or the quiesce thread has handed off a txg to
406                  * us, or we have reached our timeout.
407                  */
408                 timer = (delta >= timeout ? 0 : timeout - delta);
409                 while (!dsl_scan_active(dp->dp_scan) &&
410                     !tx->tx_exiting && timer > 0 &&
411                     tx->tx_synced_txg >= tx->tx_sync_txg_waiting &&
412                     tx->tx_quiesced_txg == 0) {
413                         dprintf("waiting; tx_synced=%llu waiting=%llu dp=%p\n",
414                             tx->tx_synced_txg, tx->tx_sync_txg_waiting, dp);
415                         txg_thread_wait(tx, &cpr, &tx->tx_sync_more_cv, timer);
416                         delta = ddi_get_lbolt() - start;
417                         timer = (delta > timeout ? 0 : timeout - delta);
418                 }
419
420                 /*
421                  * Wait until the quiesce thread hands off a txg to us,
422                  * prompting it to do so if necessary.
423                  */
424                 while (!tx->tx_exiting && tx->tx_quiesced_txg == 0) {
425                         if (tx->tx_quiesce_txg_waiting < tx->tx_open_txg+1)
426                                 tx->tx_quiesce_txg_waiting = tx->tx_open_txg+1;
427                         cv_broadcast(&tx->tx_quiesce_more_cv);
428                         txg_thread_wait(tx, &cpr, &tx->tx_quiesce_done_cv, 0);
429                 }
430
431                 if (tx->tx_exiting)
432                         txg_thread_exit(tx, &cpr, &tx->tx_sync_thread);
433
434                 /*
435                  * Consume the quiesced txg which has been handed off to
436                  * us.  This may cause the quiescing thread to now be
437                  * able to quiesce another txg, so we must signal it.
438                  */
439                 txg = tx->tx_quiesced_txg;
440                 tx->tx_quiesced_txg = 0;
441                 tx->tx_syncing_txg = txg;
442                 cv_broadcast(&tx->tx_quiesce_more_cv);
443
444                 dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
445                     txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting);
446                 mutex_exit(&tx->tx_sync_lock);
447
448                 start = ddi_get_lbolt();
449                 spa_sync(spa, txg);
450                 delta = ddi_get_lbolt() - start;
451
452                 mutex_enter(&tx->tx_sync_lock);
453                 tx->tx_synced_txg = txg;
454                 tx->tx_syncing_txg = 0;
455                 cv_broadcast(&tx->tx_sync_done_cv);
456
457                 /*
458                  * Dispatch commit callbacks to worker threads.
459                  */
460                 txg_dispatch_callbacks(dp, txg);
461         }
462 }
463
464 static void
465 txg_quiesce_thread(dsl_pool_t *dp)
466 {
467         tx_state_t *tx = &dp->dp_tx;
468         callb_cpr_t cpr;
469
470         txg_thread_enter(tx, &cpr);
471
472         for (;;) {
473                 uint64_t txg;
474
475                 /*
476                  * We quiesce when there's someone waiting on us.
477                  * However, we can only have one txg in "quiescing" or
478                  * "quiesced, waiting to sync" state.  So we wait until
479                  * the "quiesced, waiting to sync" txg has been consumed
480                  * by the sync thread.
481                  */
482                 while (!tx->tx_exiting &&
483                     (tx->tx_open_txg >= tx->tx_quiesce_txg_waiting ||
484                     tx->tx_quiesced_txg != 0))
485                         txg_thread_wait(tx, &cpr, &tx->tx_quiesce_more_cv, 0);
486
487                 if (tx->tx_exiting)
488                         txg_thread_exit(tx, &cpr, &tx->tx_quiesce_thread);
489
490                 txg = tx->tx_open_txg;
491                 dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
492                     txg, tx->tx_quiesce_txg_waiting,
493                     tx->tx_sync_txg_waiting);
494                 mutex_exit(&tx->tx_sync_lock);
495                 txg_quiesce(dp, txg);
496                 mutex_enter(&tx->tx_sync_lock);
497
498                 /*
499                  * Hand this txg off to the sync thread.
500                  */
501                 dprintf("quiesce done, handing off txg %llu\n", txg);
502                 tx->tx_quiesced_txg = txg;
503                 cv_broadcast(&tx->tx_sync_more_cv);
504                 cv_broadcast(&tx->tx_quiesce_done_cv);
505         }
506 }
507
508 /*
509  * Delay this thread by 'ticks' if we are still in the open transaction
510  * group and there is already a waiting txg quiesing or quiesced.  Abort
511  * the delay if this txg stalls or enters the quiesing state.
512  */
513 void
514 txg_delay(dsl_pool_t *dp, uint64_t txg, int ticks)
515 {
516         tx_state_t *tx = &dp->dp_tx;
517         clock_t timeout = ddi_get_lbolt() + ticks;
518
519         /* don't delay if this txg could transition to quiesing immediately */
520         if (tx->tx_open_txg > txg ||
521             tx->tx_syncing_txg == txg-1 || tx->tx_synced_txg == txg-1)
522                 return;
523
524         mutex_enter(&tx->tx_sync_lock);
525         if (tx->tx_open_txg > txg || tx->tx_synced_txg == txg-1) {
526                 mutex_exit(&tx->tx_sync_lock);
527                 return;
528         }
529
530         while (ddi_get_lbolt() < timeout &&
531             tx->tx_syncing_txg < txg-1 && !txg_stalled(dp))
532                 (void) cv_timedwait(&tx->tx_quiesce_more_cv, &tx->tx_sync_lock,
533                     timeout);
534
535         DMU_TX_STAT_BUMP(dmu_tx_delay);
536
537         mutex_exit(&tx->tx_sync_lock);
538 }
539
540 void
541 txg_wait_synced(dsl_pool_t *dp, uint64_t txg)
542 {
543         tx_state_t *tx = &dp->dp_tx;
544
545         mutex_enter(&tx->tx_sync_lock);
546         ASSERT(tx->tx_threads == 2);
547         if (txg == 0)
548                 txg = tx->tx_open_txg + TXG_DEFER_SIZE;
549         if (tx->tx_sync_txg_waiting < txg)
550                 tx->tx_sync_txg_waiting = txg;
551         dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
552             txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting);
553         while (tx->tx_synced_txg < txg) {
554                 dprintf("broadcasting sync more "
555                     "tx_synced=%llu waiting=%llu dp=%p\n",
556                     tx->tx_synced_txg, tx->tx_sync_txg_waiting, dp);
557                 cv_broadcast(&tx->tx_sync_more_cv);
558                 cv_wait(&tx->tx_sync_done_cv, &tx->tx_sync_lock);
559         }
560         mutex_exit(&tx->tx_sync_lock);
561 }
562
563 void
564 txg_wait_open(dsl_pool_t *dp, uint64_t txg)
565 {
566         tx_state_t *tx = &dp->dp_tx;
567
568         mutex_enter(&tx->tx_sync_lock);
569         ASSERT(tx->tx_threads == 2);
570         if (txg == 0)
571                 txg = tx->tx_open_txg + 1;
572         if (tx->tx_quiesce_txg_waiting < txg)
573                 tx->tx_quiesce_txg_waiting = txg;
574         dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
575             txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting);
576         while (tx->tx_open_txg < txg) {
577                 cv_broadcast(&tx->tx_quiesce_more_cv);
578                 cv_wait(&tx->tx_quiesce_done_cv, &tx->tx_sync_lock);
579         }
580         mutex_exit(&tx->tx_sync_lock);
581 }
582
583 boolean_t
584 txg_stalled(dsl_pool_t *dp)
585 {
586         tx_state_t *tx = &dp->dp_tx;
587         return (tx->tx_quiesce_txg_waiting > tx->tx_open_txg);
588 }
589
590 boolean_t
591 txg_sync_waiting(dsl_pool_t *dp)
592 {
593         tx_state_t *tx = &dp->dp_tx;
594
595         return (tx->tx_syncing_txg <= tx->tx_sync_txg_waiting ||
596             tx->tx_quiesced_txg != 0);
597 }
598
599 /*
600  * Per-txg object lists.
601  */
602 void
603 txg_list_create(txg_list_t *tl, size_t offset)
604 {
605         int t;
606
607         mutex_init(&tl->tl_lock, NULL, MUTEX_DEFAULT, NULL);
608
609         tl->tl_offset = offset;
610
611         for (t = 0; t < TXG_SIZE; t++)
612                 tl->tl_head[t] = NULL;
613 }
614
615 void
616 txg_list_destroy(txg_list_t *tl)
617 {
618         int t;
619
620         for (t = 0; t < TXG_SIZE; t++)
621                 ASSERT(txg_list_empty(tl, t));
622
623         mutex_destroy(&tl->tl_lock);
624 }
625
626 int
627 txg_list_empty(txg_list_t *tl, uint64_t txg)
628 {
629         return (tl->tl_head[txg & TXG_MASK] == NULL);
630 }
631
632 /*
633  * Add an entry to the list.
634  * Returns 0 if it's a new entry, 1 if it's already there.
635  */
636 int
637 txg_list_add(txg_list_t *tl, void *p, uint64_t txg)
638 {
639         int t = txg & TXG_MASK;
640         txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
641         int already_on_list;
642
643         mutex_enter(&tl->tl_lock);
644         already_on_list = tn->tn_member[t];
645         if (!already_on_list) {
646                 tn->tn_member[t] = 1;
647                 tn->tn_next[t] = tl->tl_head[t];
648                 tl->tl_head[t] = tn;
649         }
650         mutex_exit(&tl->tl_lock);
651
652         return (already_on_list);
653 }
654
655 /*
656  * Add an entry to the end of the list (walks list to find end).
657  * Returns 0 if it's a new entry, 1 if it's already there.
658  */
659 int
660 txg_list_add_tail(txg_list_t *tl, void *p, uint64_t txg)
661 {
662         int t = txg & TXG_MASK;
663         txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
664         int already_on_list;
665
666         mutex_enter(&tl->tl_lock);
667         already_on_list = tn->tn_member[t];
668         if (!already_on_list) {
669                 txg_node_t **tp;
670
671                 for (tp = &tl->tl_head[t]; *tp != NULL; tp = &(*tp)->tn_next[t])
672                         continue;
673
674                 tn->tn_member[t] = 1;
675                 tn->tn_next[t] = NULL;
676                 *tp = tn;
677         }
678         mutex_exit(&tl->tl_lock);
679
680         return (already_on_list);
681 }
682
683 /*
684  * Remove the head of the list and return it.
685  */
686 void *
687 txg_list_remove(txg_list_t *tl, uint64_t txg)
688 {
689         int t = txg & TXG_MASK;
690         txg_node_t *tn;
691         void *p = NULL;
692
693         mutex_enter(&tl->tl_lock);
694         if ((tn = tl->tl_head[t]) != NULL) {
695                 p = (char *)tn - tl->tl_offset;
696                 tl->tl_head[t] = tn->tn_next[t];
697                 tn->tn_next[t] = NULL;
698                 tn->tn_member[t] = 0;
699         }
700         mutex_exit(&tl->tl_lock);
701
702         return (p);
703 }
704
705 /*
706  * Remove a specific item from the list and return it.
707  */
708 void *
709 txg_list_remove_this(txg_list_t *tl, void *p, uint64_t txg)
710 {
711         int t = txg & TXG_MASK;
712         txg_node_t *tn, **tp;
713
714         mutex_enter(&tl->tl_lock);
715
716         for (tp = &tl->tl_head[t]; (tn = *tp) != NULL; tp = &tn->tn_next[t]) {
717                 if ((char *)tn - tl->tl_offset == p) {
718                         *tp = tn->tn_next[t];
719                         tn->tn_next[t] = NULL;
720                         tn->tn_member[t] = 0;
721                         mutex_exit(&tl->tl_lock);
722                         return (p);
723                 }
724         }
725
726         mutex_exit(&tl->tl_lock);
727
728         return (NULL);
729 }
730
731 int
732 txg_list_member(txg_list_t *tl, void *p, uint64_t txg)
733 {
734         int t = txg & TXG_MASK;
735         txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
736
737         return (tn->tn_member[t]);
738 }
739
740 /*
741  * Walk a txg list -- only safe if you know it's not changing.
742  */
743 void *
744 txg_list_head(txg_list_t *tl, uint64_t txg)
745 {
746         int t = txg & TXG_MASK;
747         txg_node_t *tn = tl->tl_head[t];
748
749         return (tn == NULL ? NULL : (char *)tn - tl->tl_offset);
750 }
751
752 void *
753 txg_list_next(txg_list_t *tl, void *p, uint64_t txg)
754 {
755         int t = txg & TXG_MASK;
756         txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
757
758         tn = tn->tn_next[t];
759
760         return (tn == NULL ? NULL : (char *)tn - tl->tl_offset);
761 }
762
763 #if defined(_KERNEL) && defined(HAVE_SPL)
764 EXPORT_SYMBOL(txg_init);
765 EXPORT_SYMBOL(txg_fini);
766 EXPORT_SYMBOL(txg_sync_start);
767 EXPORT_SYMBOL(txg_sync_stop);
768 EXPORT_SYMBOL(txg_hold_open);
769 EXPORT_SYMBOL(txg_rele_to_quiesce);
770 EXPORT_SYMBOL(txg_rele_to_sync);
771 EXPORT_SYMBOL(txg_register_callbacks);
772 EXPORT_SYMBOL(txg_delay);
773 EXPORT_SYMBOL(txg_wait_synced);
774 EXPORT_SYMBOL(txg_wait_open);
775 EXPORT_SYMBOL(txg_wait_callbacks);
776 EXPORT_SYMBOL(txg_stalled);
777 EXPORT_SYMBOL(txg_sync_waiting);
778
779 module_param(zfs_txg_timeout, int, 0644);
780 MODULE_PARM_DESC(zfs_txg_timeout, "Max seconds worth of delta per txg");
781 #endif