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