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