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