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