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