1fa9e4066Sahrens /* 2fa9e4066Sahrens * CDDL HEADER START 3fa9e4066Sahrens * 4fa9e4066Sahrens * The contents of this file are subject to the terms of the 55ad82045Snd * Common Development and Distribution License (the "License"). 65ad82045Snd * You may not use this file except in compliance with the License. 7fa9e4066Sahrens * 8fa9e4066Sahrens * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE 9fa9e4066Sahrens * or http://www.opensolaris.org/os/licensing. 10fa9e4066Sahrens * See the License for the specific language governing permissions 11fa9e4066Sahrens * and limitations under the License. 12fa9e4066Sahrens * 13fa9e4066Sahrens * When distributing Covered Code, include this CDDL HEADER in each 14fa9e4066Sahrens * file and include the License file at usr/src/OPENSOLARIS.LICENSE. 15fa9e4066Sahrens * If applicable, add the following below this CDDL HEADER, with the 16fa9e4066Sahrens * fields enclosed by brackets "[]" replaced with your own identifying 17fa9e4066Sahrens * information: Portions Copyright [yyyy] [name of copyright owner] 18fa9e4066Sahrens * 19fa9e4066Sahrens * CDDL HEADER END 20fa9e4066Sahrens */ 21fa9e4066Sahrens /* 228f38d419Sek * Copyright 2008 Sun Microsystems, Inc. All rights reserved. 23fa9e4066Sahrens * Use is subject to license terms. 24fa9e4066Sahrens */ 25fa9e4066Sahrens 26fa9e4066Sahrens #pragma ident "%Z%%M% %I% %E% SMI" 27fa9e4066Sahrens 28fa9e4066Sahrens #include <sys/zfs_context.h> 29fa9e4066Sahrens #include <sys/txg_impl.h> 30fa9e4066Sahrens #include <sys/dmu_impl.h> 31fa9e4066Sahrens #include <sys/dsl_pool.h> 32fa9e4066Sahrens #include <sys/callb.h> 33fa9e4066Sahrens 34fa9e4066Sahrens /* 35fa9e4066Sahrens * Pool-wide transaction groups. 36fa9e4066Sahrens */ 37fa9e4066Sahrens 38fa9e4066Sahrens static void txg_sync_thread(dsl_pool_t *dp); 39fa9e4066Sahrens static void txg_quiesce_thread(dsl_pool_t *dp); 40fa9e4066Sahrens 411ab7f2deSmaybee int zfs_txg_timeout = 30; /* max seconds worth of delta per txg */ 421ab7f2deSmaybee int zfs_txg_synctime = 5; /* target seconds to sync a txg */ 431ab7f2deSmaybee 441ab7f2deSmaybee int zfs_write_limit_shift = 3; /* 1/8th of physical memory */ 451ab7f2deSmaybee 461ab7f2deSmaybee uint64_t zfs_write_limit_min = 32 << 20; /* min write limit is 32MB */ 471ab7f2deSmaybee uint64_t zfs_write_limit_max = 0; /* max data payload per txg */ 481ab7f2deSmaybee uint64_t zfs_write_limit_inflated = 0; 49fa9e4066Sahrens 50fa9e4066Sahrens /* 51fa9e4066Sahrens * Prepare the txg subsystem. 52fa9e4066Sahrens */ 53fa9e4066Sahrens void 54fa9e4066Sahrens txg_init(dsl_pool_t *dp, uint64_t txg) 55fa9e4066Sahrens { 56fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 575ad82045Snd int c; 58fa9e4066Sahrens bzero(tx, sizeof (tx_state_t)); 59fa9e4066Sahrens 60fa9e4066Sahrens tx->tx_cpu = kmem_zalloc(max_ncpus * sizeof (tx_cpu_t), KM_SLEEP); 61fa9e4066Sahrens 628f38d419Sek for (c = 0; c < max_ncpus; c++) { 638f38d419Sek int i; 648f38d419Sek 655ad82045Snd mutex_init(&tx->tx_cpu[c].tc_lock, NULL, MUTEX_DEFAULT, NULL); 668f38d419Sek for (i = 0; i < TXG_SIZE; i++) { 678f38d419Sek cv_init(&tx->tx_cpu[c].tc_cv[i], NULL, CV_DEFAULT, 688f38d419Sek NULL); 698f38d419Sek } 708f38d419Sek } 715ad82045Snd 72fa9e4066Sahrens rw_init(&tx->tx_suspend, NULL, RW_DEFAULT, NULL); 735ad82045Snd mutex_init(&tx->tx_sync_lock, NULL, MUTEX_DEFAULT, NULL); 74fa9e4066Sahrens 75fa9e4066Sahrens tx->tx_open_txg = txg; 76fa9e4066Sahrens } 77fa9e4066Sahrens 78fa9e4066Sahrens /* 79fa9e4066Sahrens * Close down the txg subsystem. 80fa9e4066Sahrens */ 81fa9e4066Sahrens void 82fa9e4066Sahrens txg_fini(dsl_pool_t *dp) 83fa9e4066Sahrens { 84fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 855ad82045Snd int c; 86fa9e4066Sahrens 87fa9e4066Sahrens ASSERT(tx->tx_threads == 0); 88fa9e4066Sahrens 89fa9e4066Sahrens rw_destroy(&tx->tx_suspend); 905ad82045Snd mutex_destroy(&tx->tx_sync_lock); 915ad82045Snd 928f38d419Sek for (c = 0; c < max_ncpus; c++) { 938f38d419Sek int i; 948f38d419Sek 955ad82045Snd mutex_destroy(&tx->tx_cpu[c].tc_lock); 968f38d419Sek for (i = 0; i < TXG_SIZE; i++) 978f38d419Sek cv_destroy(&tx->tx_cpu[c].tc_cv[i]); 988f38d419Sek } 99fa9e4066Sahrens 100fa9e4066Sahrens kmem_free(tx->tx_cpu, max_ncpus * sizeof (tx_cpu_t)); 101fa9e4066Sahrens 102fa9e4066Sahrens bzero(tx, sizeof (tx_state_t)); 103fa9e4066Sahrens } 104fa9e4066Sahrens 105fa9e4066Sahrens /* 106fa9e4066Sahrens * Start syncing transaction groups. 107fa9e4066Sahrens */ 108fa9e4066Sahrens void 109fa9e4066Sahrens txg_sync_start(dsl_pool_t *dp) 110fa9e4066Sahrens { 111fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 112fa9e4066Sahrens 113fa9e4066Sahrens mutex_enter(&tx->tx_sync_lock); 114fa9e4066Sahrens 115fa9e4066Sahrens dprintf("pool %p\n", dp); 116fa9e4066Sahrens 117fa9e4066Sahrens ASSERT(tx->tx_threads == 0); 118fa9e4066Sahrens 1191ab7f2deSmaybee tx->tx_threads = 2; 120fa9e4066Sahrens 121fa9e4066Sahrens tx->tx_quiesce_thread = thread_create(NULL, 0, txg_quiesce_thread, 122fa9e4066Sahrens dp, 0, &p0, TS_RUN, minclsyspri); 123fa9e4066Sahrens 124*088f3894Sahrens /* 125*088f3894Sahrens * The sync thread can need a larger-than-default stack size on 126*088f3894Sahrens * 32-bit x86. This is due in part to nested pools and 127*088f3894Sahrens * scrub_visitbp() recursion. 128*088f3894Sahrens */ 129*088f3894Sahrens tx->tx_sync_thread = thread_create(NULL, 12<<10, txg_sync_thread, 130fa9e4066Sahrens dp, 0, &p0, TS_RUN, minclsyspri); 131fa9e4066Sahrens 132fa9e4066Sahrens mutex_exit(&tx->tx_sync_lock); 133fa9e4066Sahrens } 134fa9e4066Sahrens 135fa9e4066Sahrens static void 136fa9e4066Sahrens txg_thread_enter(tx_state_t *tx, callb_cpr_t *cpr) 137fa9e4066Sahrens { 138fa9e4066Sahrens CALLB_CPR_INIT(cpr, &tx->tx_sync_lock, callb_generic_cpr, FTAG); 139fa9e4066Sahrens mutex_enter(&tx->tx_sync_lock); 140fa9e4066Sahrens } 141fa9e4066Sahrens 142fa9e4066Sahrens static void 143fa9e4066Sahrens txg_thread_exit(tx_state_t *tx, callb_cpr_t *cpr, kthread_t **tpp) 144fa9e4066Sahrens { 145fa9e4066Sahrens ASSERT(*tpp != NULL); 146fa9e4066Sahrens *tpp = NULL; 147fa9e4066Sahrens tx->tx_threads--; 148fa9e4066Sahrens cv_broadcast(&tx->tx_exit_cv); 149fa9e4066Sahrens CALLB_CPR_EXIT(cpr); /* drops &tx->tx_sync_lock */ 150fa9e4066Sahrens thread_exit(); 151fa9e4066Sahrens } 152fa9e4066Sahrens 153fa9e4066Sahrens static void 1541ab7f2deSmaybee txg_thread_wait(tx_state_t *tx, callb_cpr_t *cpr, kcondvar_t *cv, uint64_t time) 155fa9e4066Sahrens { 156fa9e4066Sahrens CALLB_CPR_SAFE_BEGIN(cpr); 157fa9e4066Sahrens 1581ab7f2deSmaybee if (time) 1591ab7f2deSmaybee (void) cv_timedwait(cv, &tx->tx_sync_lock, lbolt + time); 160fa9e4066Sahrens else 161fa9e4066Sahrens cv_wait(cv, &tx->tx_sync_lock); 162fa9e4066Sahrens 163fa9e4066Sahrens CALLB_CPR_SAFE_END(cpr, &tx->tx_sync_lock); 164fa9e4066Sahrens } 165fa9e4066Sahrens 166fa9e4066Sahrens /* 167fa9e4066Sahrens * Stop syncing transaction groups. 168fa9e4066Sahrens */ 169fa9e4066Sahrens void 170fa9e4066Sahrens txg_sync_stop(dsl_pool_t *dp) 171fa9e4066Sahrens { 172fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 173fa9e4066Sahrens 174fa9e4066Sahrens dprintf("pool %p\n", dp); 175fa9e4066Sahrens /* 176fa9e4066Sahrens * Finish off any work in progress. 177fa9e4066Sahrens */ 1781ab7f2deSmaybee ASSERT(tx->tx_threads == 2); 179fa9e4066Sahrens txg_wait_synced(dp, 0); 180fa9e4066Sahrens 181fa9e4066Sahrens /* 1821ab7f2deSmaybee * Wake all sync threads and wait for them to die. 183fa9e4066Sahrens */ 184fa9e4066Sahrens mutex_enter(&tx->tx_sync_lock); 185fa9e4066Sahrens 1861ab7f2deSmaybee ASSERT(tx->tx_threads == 2); 187fa9e4066Sahrens 188fa9e4066Sahrens tx->tx_exiting = 1; 189fa9e4066Sahrens 190fa9e4066Sahrens cv_broadcast(&tx->tx_quiesce_more_cv); 191fa9e4066Sahrens cv_broadcast(&tx->tx_quiesce_done_cv); 192fa9e4066Sahrens cv_broadcast(&tx->tx_sync_more_cv); 193fa9e4066Sahrens 194fa9e4066Sahrens while (tx->tx_threads != 0) 195fa9e4066Sahrens cv_wait(&tx->tx_exit_cv, &tx->tx_sync_lock); 196fa9e4066Sahrens 197fa9e4066Sahrens tx->tx_exiting = 0; 198fa9e4066Sahrens 199fa9e4066Sahrens mutex_exit(&tx->tx_sync_lock); 200fa9e4066Sahrens } 201fa9e4066Sahrens 202fa9e4066Sahrens uint64_t 203fa9e4066Sahrens txg_hold_open(dsl_pool_t *dp, txg_handle_t *th) 204fa9e4066Sahrens { 205fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 206fa9e4066Sahrens tx_cpu_t *tc = &tx->tx_cpu[CPU_SEQID]; 207fa9e4066Sahrens uint64_t txg; 208fa9e4066Sahrens 209fa9e4066Sahrens mutex_enter(&tc->tc_lock); 210fa9e4066Sahrens 211fa9e4066Sahrens txg = tx->tx_open_txg; 212fa9e4066Sahrens tc->tc_count[txg & TXG_MASK]++; 213fa9e4066Sahrens 214fa9e4066Sahrens th->th_cpu = tc; 215fa9e4066Sahrens th->th_txg = txg; 216fa9e4066Sahrens 217fa9e4066Sahrens return (txg); 218fa9e4066Sahrens } 219fa9e4066Sahrens 220fa9e4066Sahrens void 221fa9e4066Sahrens txg_rele_to_quiesce(txg_handle_t *th) 222fa9e4066Sahrens { 223fa9e4066Sahrens tx_cpu_t *tc = th->th_cpu; 224fa9e4066Sahrens 225fa9e4066Sahrens mutex_exit(&tc->tc_lock); 226fa9e4066Sahrens } 227fa9e4066Sahrens 228fa9e4066Sahrens void 229fa9e4066Sahrens txg_rele_to_sync(txg_handle_t *th) 230fa9e4066Sahrens { 231fa9e4066Sahrens tx_cpu_t *tc = th->th_cpu; 232fa9e4066Sahrens int g = th->th_txg & TXG_MASK; 233fa9e4066Sahrens 234fa9e4066Sahrens mutex_enter(&tc->tc_lock); 235fa9e4066Sahrens ASSERT(tc->tc_count[g] != 0); 236fa9e4066Sahrens if (--tc->tc_count[g] == 0) 237fa9e4066Sahrens cv_broadcast(&tc->tc_cv[g]); 238fa9e4066Sahrens mutex_exit(&tc->tc_lock); 239fa9e4066Sahrens 240fa9e4066Sahrens th->th_cpu = NULL; /* defensive */ 241fa9e4066Sahrens } 242fa9e4066Sahrens 243fa9e4066Sahrens static void 244fa9e4066Sahrens txg_quiesce(dsl_pool_t *dp, uint64_t txg) 245fa9e4066Sahrens { 246fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 247fa9e4066Sahrens int g = txg & TXG_MASK; 248fa9e4066Sahrens int c; 249fa9e4066Sahrens 250fa9e4066Sahrens /* 251fa9e4066Sahrens * Grab all tx_cpu locks so nobody else can get into this txg. 252fa9e4066Sahrens */ 253fa9e4066Sahrens for (c = 0; c < max_ncpus; c++) 254fa9e4066Sahrens mutex_enter(&tx->tx_cpu[c].tc_lock); 255fa9e4066Sahrens 256fa9e4066Sahrens ASSERT(txg == tx->tx_open_txg); 257fa9e4066Sahrens tx->tx_open_txg++; 258fa9e4066Sahrens 259fa9e4066Sahrens /* 260fa9e4066Sahrens * Now that we've incremented tx_open_txg, we can let threads 261fa9e4066Sahrens * enter the next transaction group. 262fa9e4066Sahrens */ 263fa9e4066Sahrens for (c = 0; c < max_ncpus; c++) 264fa9e4066Sahrens mutex_exit(&tx->tx_cpu[c].tc_lock); 265fa9e4066Sahrens 266fa9e4066Sahrens /* 267fa9e4066Sahrens * Quiesce the transaction group by waiting for everyone to txg_exit(). 268fa9e4066Sahrens */ 269fa9e4066Sahrens for (c = 0; c < max_ncpus; c++) { 270fa9e4066Sahrens tx_cpu_t *tc = &tx->tx_cpu[c]; 271fa9e4066Sahrens mutex_enter(&tc->tc_lock); 272fa9e4066Sahrens while (tc->tc_count[g] != 0) 273fa9e4066Sahrens cv_wait(&tc->tc_cv[g], &tc->tc_lock); 274fa9e4066Sahrens mutex_exit(&tc->tc_lock); 275fa9e4066Sahrens } 276fa9e4066Sahrens } 277fa9e4066Sahrens 278fa9e4066Sahrens static void 279fa9e4066Sahrens txg_sync_thread(dsl_pool_t *dp) 280fa9e4066Sahrens { 281fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 282fa9e4066Sahrens callb_cpr_t cpr; 2831ab7f2deSmaybee uint64_t timeout, start, delta, timer; 2841ab7f2deSmaybee int target; 285fa9e4066Sahrens 286fa9e4066Sahrens txg_thread_enter(tx, &cpr); 287fa9e4066Sahrens 2881ab7f2deSmaybee start = delta = 0; 2891ab7f2deSmaybee timeout = zfs_txg_timeout * hz; 290fa9e4066Sahrens for (;;) { 2911ab7f2deSmaybee uint64_t txg, written; 292fa9e4066Sahrens 293fa9e4066Sahrens /* 294fa9e4066Sahrens * We sync when there's someone waiting on us, or the 2951ab7f2deSmaybee * quiesce thread has handed off a txg to us, or we have 2961ab7f2deSmaybee * reached our timeout. 297fa9e4066Sahrens */ 2981ab7f2deSmaybee timer = (delta >= timeout ? 0 : timeout - delta); 2991ab7f2deSmaybee while (!tx->tx_exiting && timer > 0 && 300fa9e4066Sahrens tx->tx_synced_txg >= tx->tx_sync_txg_waiting && 301fa9e4066Sahrens tx->tx_quiesced_txg == 0) { 302fa9e4066Sahrens dprintf("waiting; tx_synced=%llu waiting=%llu dp=%p\n", 303fa9e4066Sahrens tx->tx_synced_txg, tx->tx_sync_txg_waiting, dp); 3041ab7f2deSmaybee txg_thread_wait(tx, &cpr, &tx->tx_sync_more_cv, timer); 3051ab7f2deSmaybee delta = lbolt - start; 3061ab7f2deSmaybee timer = (delta > timeout ? 0 : timeout - delta); 307fa9e4066Sahrens } 308fa9e4066Sahrens 309fa9e4066Sahrens /* 310fa9e4066Sahrens * Wait until the quiesce thread hands off a txg to us, 311fa9e4066Sahrens * prompting it to do so if necessary. 312fa9e4066Sahrens */ 313fa9e4066Sahrens while (!tx->tx_exiting && tx->tx_quiesced_txg == 0) { 314fa9e4066Sahrens if (tx->tx_quiesce_txg_waiting < tx->tx_open_txg+1) 315fa9e4066Sahrens tx->tx_quiesce_txg_waiting = tx->tx_open_txg+1; 316fa9e4066Sahrens cv_broadcast(&tx->tx_quiesce_more_cv); 317fa9e4066Sahrens txg_thread_wait(tx, &cpr, &tx->tx_quiesce_done_cv, 0); 318fa9e4066Sahrens } 319fa9e4066Sahrens 320fa9e4066Sahrens if (tx->tx_exiting) 321fa9e4066Sahrens txg_thread_exit(tx, &cpr, &tx->tx_sync_thread); 322fa9e4066Sahrens 323fa9e4066Sahrens rw_enter(&tx->tx_suspend, RW_WRITER); 324fa9e4066Sahrens 325fa9e4066Sahrens /* 326fa9e4066Sahrens * Consume the quiesced txg which has been handed off to 327fa9e4066Sahrens * us. This may cause the quiescing thread to now be 328fa9e4066Sahrens * able to quiesce another txg, so we must signal it. 329fa9e4066Sahrens */ 330fa9e4066Sahrens txg = tx->tx_quiesced_txg; 331fa9e4066Sahrens tx->tx_quiesced_txg = 0; 332fa9e4066Sahrens tx->tx_syncing_txg = txg; 333fa9e4066Sahrens cv_broadcast(&tx->tx_quiesce_more_cv); 334fa9e4066Sahrens rw_exit(&tx->tx_suspend); 335fa9e4066Sahrens 336fa9e4066Sahrens dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n", 3378f38d419Sek txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting); 338fa9e4066Sahrens mutex_exit(&tx->tx_sync_lock); 3391ab7f2deSmaybee start = lbolt; 340fa9e4066Sahrens spa_sync(dp->dp_spa, txg); 341745cd3c5Smaybee delta = (lbolt - start) + 1; 3421ab7f2deSmaybee 3431ab7f2deSmaybee written = dp->dp_space_towrite[txg & TXG_MASK]; 3441ab7f2deSmaybee dp->dp_space_towrite[txg & TXG_MASK] = 0; 3451ab7f2deSmaybee ASSERT(dp->dp_tempreserved[txg & TXG_MASK] == 0); 3461ab7f2deSmaybee 3471ab7f2deSmaybee /* 3481ab7f2deSmaybee * If the write limit max has not been explicitly set, set it 3491ab7f2deSmaybee * to a fraction of available phisical memory (default 1/8th). 3501ab7f2deSmaybee * Note that we must inflate the limit because the spa 3511ab7f2deSmaybee * inflates write sizes to account for data replication. 3521ab7f2deSmaybee * Check this each sync phase to catch changing memory size. 3531ab7f2deSmaybee */ 3541ab7f2deSmaybee if (zfs_write_limit_inflated == 0 || 3551ab7f2deSmaybee (zfs_write_limit_shift && zfs_write_limit_max != 3561ab7f2deSmaybee physmem * PAGESIZE >> zfs_write_limit_shift)) { 3571ab7f2deSmaybee zfs_write_limit_max = 3581ab7f2deSmaybee physmem * PAGESIZE >> zfs_write_limit_shift; 3591ab7f2deSmaybee zfs_write_limit_inflated = 3601ab7f2deSmaybee spa_get_asize(dp->dp_spa, zfs_write_limit_max); 3611ab7f2deSmaybee if (zfs_write_limit_min > zfs_write_limit_inflated) 3621ab7f2deSmaybee zfs_write_limit_inflated = zfs_write_limit_min; 3631ab7f2deSmaybee } 3641ab7f2deSmaybee 3651ab7f2deSmaybee /* 3661ab7f2deSmaybee * Attempt to keep the sync time consistant by adjusting the 3671ab7f2deSmaybee * amount of write traffic allowed into each transaction group. 3681ab7f2deSmaybee */ 3691ab7f2deSmaybee target = zfs_txg_synctime * hz; 3701ab7f2deSmaybee if (delta > target) { 3711ab7f2deSmaybee uint64_t old = MIN(dp->dp_write_limit, written); 3721ab7f2deSmaybee 3731ab7f2deSmaybee dp->dp_write_limit = MAX(zfs_write_limit_min, 3741ab7f2deSmaybee old * target / delta); 3751ab7f2deSmaybee } else if (written >= dp->dp_write_limit && 3761ab7f2deSmaybee delta >> 3 < target >> 3) { 3771ab7f2deSmaybee uint64_t rescale = 3781ab7f2deSmaybee MIN((100 * target) / delta, 200); 3791ab7f2deSmaybee 3801ab7f2deSmaybee dp->dp_write_limit = MIN(zfs_write_limit_inflated, 3811ab7f2deSmaybee written * rescale / 100); 3821ab7f2deSmaybee } 3831ab7f2deSmaybee 384fa9e4066Sahrens mutex_enter(&tx->tx_sync_lock); 385fa9e4066Sahrens rw_enter(&tx->tx_suspend, RW_WRITER); 386fa9e4066Sahrens tx->tx_synced_txg = txg; 387fa9e4066Sahrens tx->tx_syncing_txg = 0; 388fa9e4066Sahrens rw_exit(&tx->tx_suspend); 389fa9e4066Sahrens cv_broadcast(&tx->tx_sync_done_cv); 390fa9e4066Sahrens } 391fa9e4066Sahrens } 392fa9e4066Sahrens 393fa9e4066Sahrens static void 394fa9e4066Sahrens txg_quiesce_thread(dsl_pool_t *dp) 395fa9e4066Sahrens { 396fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 397fa9e4066Sahrens callb_cpr_t cpr; 398fa9e4066Sahrens 399fa9e4066Sahrens txg_thread_enter(tx, &cpr); 400fa9e4066Sahrens 401fa9e4066Sahrens for (;;) { 402fa9e4066Sahrens uint64_t txg; 403fa9e4066Sahrens 404fa9e4066Sahrens /* 405fa9e4066Sahrens * We quiesce when there's someone waiting on us. 406fa9e4066Sahrens * However, we can only have one txg in "quiescing" or 407fa9e4066Sahrens * "quiesced, waiting to sync" state. So we wait until 408fa9e4066Sahrens * the "quiesced, waiting to sync" txg has been consumed 409fa9e4066Sahrens * by the sync thread. 410fa9e4066Sahrens */ 411fa9e4066Sahrens while (!tx->tx_exiting && 412fa9e4066Sahrens (tx->tx_open_txg >= tx->tx_quiesce_txg_waiting || 413fa9e4066Sahrens tx->tx_quiesced_txg != 0)) 414fa9e4066Sahrens txg_thread_wait(tx, &cpr, &tx->tx_quiesce_more_cv, 0); 415fa9e4066Sahrens 416fa9e4066Sahrens if (tx->tx_exiting) 417fa9e4066Sahrens txg_thread_exit(tx, &cpr, &tx->tx_quiesce_thread); 418fa9e4066Sahrens 419fa9e4066Sahrens txg = tx->tx_open_txg; 420fa9e4066Sahrens dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n", 421fa9e4066Sahrens txg, tx->tx_quiesce_txg_waiting, 422fa9e4066Sahrens tx->tx_sync_txg_waiting); 423fa9e4066Sahrens mutex_exit(&tx->tx_sync_lock); 424fa9e4066Sahrens txg_quiesce(dp, txg); 425fa9e4066Sahrens mutex_enter(&tx->tx_sync_lock); 426fa9e4066Sahrens 427fa9e4066Sahrens /* 428fa9e4066Sahrens * Hand this txg off to the sync thread. 429fa9e4066Sahrens */ 430fa9e4066Sahrens dprintf("quiesce done, handing off txg %llu\n", txg); 431fa9e4066Sahrens tx->tx_quiesced_txg = txg; 432fa9e4066Sahrens cv_broadcast(&tx->tx_sync_more_cv); 433fa9e4066Sahrens cv_broadcast(&tx->tx_quiesce_done_cv); 434fa9e4066Sahrens } 435fa9e4066Sahrens } 436fa9e4066Sahrens 4371ab7f2deSmaybee /* 4381ab7f2deSmaybee * Delay this thread by 'ticks' if we are still in the open transaction 4391ab7f2deSmaybee * group and there is already a waiting txg quiesing or quiesced. Abort 4401ab7f2deSmaybee * the delay if this txg stalls or enters the quiesing state. 4411ab7f2deSmaybee */ 4421ab7f2deSmaybee void 4431ab7f2deSmaybee txg_delay(dsl_pool_t *dp, uint64_t txg, int ticks) 4441ab7f2deSmaybee { 4451ab7f2deSmaybee tx_state_t *tx = &dp->dp_tx; 4461ab7f2deSmaybee int timeout = lbolt + ticks; 4471ab7f2deSmaybee 4481ab7f2deSmaybee /* don't delay if this txg could transition to quiesing immediately */ 4491ab7f2deSmaybee if (tx->tx_open_txg > txg || 4501ab7f2deSmaybee tx->tx_syncing_txg == txg-1 || tx->tx_synced_txg == txg-1) 4511ab7f2deSmaybee return; 4521ab7f2deSmaybee 4531ab7f2deSmaybee mutex_enter(&tx->tx_sync_lock); 4541ab7f2deSmaybee if (tx->tx_open_txg > txg || tx->tx_synced_txg == txg-1) { 4551ab7f2deSmaybee mutex_exit(&tx->tx_sync_lock); 4561ab7f2deSmaybee return; 4571ab7f2deSmaybee } 4581ab7f2deSmaybee 4591ab7f2deSmaybee while (lbolt < timeout && 4601ab7f2deSmaybee tx->tx_syncing_txg < txg-1 && !txg_stalled(dp)) 4611ab7f2deSmaybee (void) cv_timedwait(&tx->tx_quiesce_more_cv, &tx->tx_sync_lock, 4621ab7f2deSmaybee timeout); 4631ab7f2deSmaybee 4641ab7f2deSmaybee mutex_exit(&tx->tx_sync_lock); 4651ab7f2deSmaybee } 4661ab7f2deSmaybee 467fa9e4066Sahrens void 468fa9e4066Sahrens txg_wait_synced(dsl_pool_t *dp, uint64_t txg) 469fa9e4066Sahrens { 470fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 471fa9e4066Sahrens 472fa9e4066Sahrens mutex_enter(&tx->tx_sync_lock); 4731ab7f2deSmaybee ASSERT(tx->tx_threads == 2); 474fa9e4066Sahrens if (txg == 0) 475fa9e4066Sahrens txg = tx->tx_open_txg; 476fa9e4066Sahrens if (tx->tx_sync_txg_waiting < txg) 477fa9e4066Sahrens tx->tx_sync_txg_waiting = txg; 478fa9e4066Sahrens dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n", 479fa9e4066Sahrens txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting); 480fa9e4066Sahrens while (tx->tx_synced_txg < txg) { 481fa9e4066Sahrens dprintf("broadcasting sync more " 482fa9e4066Sahrens "tx_synced=%llu waiting=%llu dp=%p\n", 483fa9e4066Sahrens tx->tx_synced_txg, tx->tx_sync_txg_waiting, dp); 484fa9e4066Sahrens cv_broadcast(&tx->tx_sync_more_cv); 485fa9e4066Sahrens cv_wait(&tx->tx_sync_done_cv, &tx->tx_sync_lock); 486fa9e4066Sahrens } 487fa9e4066Sahrens mutex_exit(&tx->tx_sync_lock); 488fa9e4066Sahrens } 489fa9e4066Sahrens 490fa9e4066Sahrens void 491fa9e4066Sahrens txg_wait_open(dsl_pool_t *dp, uint64_t txg) 492fa9e4066Sahrens { 493fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 494fa9e4066Sahrens 495fa9e4066Sahrens mutex_enter(&tx->tx_sync_lock); 4961ab7f2deSmaybee ASSERT(tx->tx_threads == 2); 497fa9e4066Sahrens if (txg == 0) 498fa9e4066Sahrens txg = tx->tx_open_txg + 1; 499fa9e4066Sahrens if (tx->tx_quiesce_txg_waiting < txg) 500fa9e4066Sahrens tx->tx_quiesce_txg_waiting = txg; 501fa9e4066Sahrens dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n", 502fa9e4066Sahrens txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting); 503fa9e4066Sahrens while (tx->tx_open_txg < txg) { 504fa9e4066Sahrens cv_broadcast(&tx->tx_quiesce_more_cv); 505fa9e4066Sahrens cv_wait(&tx->tx_quiesce_done_cv, &tx->tx_sync_lock); 506fa9e4066Sahrens } 507fa9e4066Sahrens mutex_exit(&tx->tx_sync_lock); 508fa9e4066Sahrens } 509fa9e4066Sahrens 510*088f3894Sahrens boolean_t 511fa9e4066Sahrens txg_stalled(dsl_pool_t *dp) 512fa9e4066Sahrens { 513fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 514fa9e4066Sahrens return (tx->tx_quiesce_txg_waiting > tx->tx_open_txg); 515fa9e4066Sahrens } 516fa9e4066Sahrens 517*088f3894Sahrens boolean_t 518*088f3894Sahrens txg_sync_waiting(dsl_pool_t *dp) 519*088f3894Sahrens { 520*088f3894Sahrens tx_state_t *tx = &dp->dp_tx; 521*088f3894Sahrens 522*088f3894Sahrens return (tx->tx_syncing_txg <= tx->tx_sync_txg_waiting || 523*088f3894Sahrens tx->tx_quiesced_txg != 0); 524*088f3894Sahrens } 525*088f3894Sahrens 526fa9e4066Sahrens void 527fa9e4066Sahrens txg_suspend(dsl_pool_t *dp) 528fa9e4066Sahrens { 529fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 530fa9e4066Sahrens /* XXX some code paths suspend when they are already suspended! */ 531fa9e4066Sahrens rw_enter(&tx->tx_suspend, RW_READER); 532fa9e4066Sahrens } 533fa9e4066Sahrens 534fa9e4066Sahrens void 535fa9e4066Sahrens txg_resume(dsl_pool_t *dp) 536fa9e4066Sahrens { 537fa9e4066Sahrens tx_state_t *tx = &dp->dp_tx; 538fa9e4066Sahrens rw_exit(&tx->tx_suspend); 539fa9e4066Sahrens } 540fa9e4066Sahrens 541fa9e4066Sahrens /* 542fa9e4066Sahrens * Per-txg object lists. 543fa9e4066Sahrens */ 544fa9e4066Sahrens void 545fa9e4066Sahrens txg_list_create(txg_list_t *tl, size_t offset) 546fa9e4066Sahrens { 547fa9e4066Sahrens int t; 548fa9e4066Sahrens 549fa9e4066Sahrens mutex_init(&tl->tl_lock, NULL, MUTEX_DEFAULT, NULL); 550fa9e4066Sahrens 551fa9e4066Sahrens tl->tl_offset = offset; 552fa9e4066Sahrens 553fa9e4066Sahrens for (t = 0; t < TXG_SIZE; t++) 554fa9e4066Sahrens tl->tl_head[t] = NULL; 555fa9e4066Sahrens } 556fa9e4066Sahrens 557fa9e4066Sahrens void 558fa9e4066Sahrens txg_list_destroy(txg_list_t *tl) 559fa9e4066Sahrens { 560fa9e4066Sahrens int t; 561fa9e4066Sahrens 562fa9e4066Sahrens for (t = 0; t < TXG_SIZE; t++) 563fa9e4066Sahrens ASSERT(txg_list_empty(tl, t)); 564fa9e4066Sahrens 565fa9e4066Sahrens mutex_destroy(&tl->tl_lock); 566fa9e4066Sahrens } 567fa9e4066Sahrens 568fa9e4066Sahrens int 569fa9e4066Sahrens txg_list_empty(txg_list_t *tl, uint64_t txg) 570fa9e4066Sahrens { 571fa9e4066Sahrens return (tl->tl_head[txg & TXG_MASK] == NULL); 572fa9e4066Sahrens } 573fa9e4066Sahrens 574fa9e4066Sahrens /* 575fa9e4066Sahrens * Add an entry to the list. 576fa9e4066Sahrens * Returns 0 if it's a new entry, 1 if it's already there. 577fa9e4066Sahrens */ 578fa9e4066Sahrens int 579fa9e4066Sahrens txg_list_add(txg_list_t *tl, void *p, uint64_t txg) 580fa9e4066Sahrens { 581fa9e4066Sahrens int t = txg & TXG_MASK; 582fa9e4066Sahrens txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset); 583fa9e4066Sahrens int already_on_list; 584fa9e4066Sahrens 585fa9e4066Sahrens mutex_enter(&tl->tl_lock); 586fa9e4066Sahrens already_on_list = tn->tn_member[t]; 587fa9e4066Sahrens if (!already_on_list) { 588fa9e4066Sahrens tn->tn_member[t] = 1; 589fa9e4066Sahrens tn->tn_next[t] = tl->tl_head[t]; 590fa9e4066Sahrens tl->tl_head[t] = tn; 591fa9e4066Sahrens } 592fa9e4066Sahrens mutex_exit(&tl->tl_lock); 593fa9e4066Sahrens 594fa9e4066Sahrens return (already_on_list); 595fa9e4066Sahrens } 596fa9e4066Sahrens 597fa9e4066Sahrens /* 598fa9e4066Sahrens * Remove the head of the list and return it. 599fa9e4066Sahrens */ 600fa9e4066Sahrens void * 601fa9e4066Sahrens txg_list_remove(txg_list_t *tl, uint64_t txg) 602fa9e4066Sahrens { 603fa9e4066Sahrens int t = txg & TXG_MASK; 604fa9e4066Sahrens txg_node_t *tn; 605fa9e4066Sahrens void *p = NULL; 606fa9e4066Sahrens 607fa9e4066Sahrens mutex_enter(&tl->tl_lock); 608fa9e4066Sahrens if ((tn = tl->tl_head[t]) != NULL) { 609fa9e4066Sahrens p = (char *)tn - tl->tl_offset; 610fa9e4066Sahrens tl->tl_head[t] = tn->tn_next[t]; 611fa9e4066Sahrens tn->tn_next[t] = NULL; 612fa9e4066Sahrens tn->tn_member[t] = 0; 613fa9e4066Sahrens } 614fa9e4066Sahrens mutex_exit(&tl->tl_lock); 615fa9e4066Sahrens 616fa9e4066Sahrens return (p); 617fa9e4066Sahrens } 618fa9e4066Sahrens 619fa9e4066Sahrens /* 620fa9e4066Sahrens * Remove a specific item from the list and return it. 621fa9e4066Sahrens */ 622fa9e4066Sahrens void * 623fa9e4066Sahrens txg_list_remove_this(txg_list_t *tl, void *p, uint64_t txg) 624fa9e4066Sahrens { 625fa9e4066Sahrens int t = txg & TXG_MASK; 626fa9e4066Sahrens txg_node_t *tn, **tp; 627fa9e4066Sahrens 628fa9e4066Sahrens mutex_enter(&tl->tl_lock); 629fa9e4066Sahrens 630fa9e4066Sahrens for (tp = &tl->tl_head[t]; (tn = *tp) != NULL; tp = &tn->tn_next[t]) { 631fa9e4066Sahrens if ((char *)tn - tl->tl_offset == p) { 632fa9e4066Sahrens *tp = tn->tn_next[t]; 633fa9e4066Sahrens tn->tn_next[t] = NULL; 634fa9e4066Sahrens tn->tn_member[t] = 0; 635fa9e4066Sahrens mutex_exit(&tl->tl_lock); 636fa9e4066Sahrens return (p); 637fa9e4066Sahrens } 638fa9e4066Sahrens } 639fa9e4066Sahrens 640fa9e4066Sahrens mutex_exit(&tl->tl_lock); 641fa9e4066Sahrens 642fa9e4066Sahrens return (NULL); 643fa9e4066Sahrens } 644fa9e4066Sahrens 645fa9e4066Sahrens int 646fa9e4066Sahrens txg_list_member(txg_list_t *tl, void *p, uint64_t txg) 647fa9e4066Sahrens { 648fa9e4066Sahrens int t = txg & TXG_MASK; 649fa9e4066Sahrens txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset); 650fa9e4066Sahrens 651fa9e4066Sahrens return (tn->tn_member[t]); 652fa9e4066Sahrens } 653fa9e4066Sahrens 654fa9e4066Sahrens /* 655fa9e4066Sahrens * Walk a txg list -- only safe if you know it's not changing. 656fa9e4066Sahrens */ 657fa9e4066Sahrens void * 658fa9e4066Sahrens txg_list_head(txg_list_t *tl, uint64_t txg) 659fa9e4066Sahrens { 660fa9e4066Sahrens int t = txg & TXG_MASK; 661fa9e4066Sahrens txg_node_t *tn = tl->tl_head[t]; 662fa9e4066Sahrens 663fa9e4066Sahrens return (tn == NULL ? NULL : (char *)tn - tl->tl_offset); 664fa9e4066Sahrens } 665fa9e4066Sahrens 666fa9e4066Sahrens void * 667fa9e4066Sahrens txg_list_next(txg_list_t *tl, void *p, uint64_t txg) 668fa9e4066Sahrens { 669fa9e4066Sahrens int t = txg & TXG_MASK; 670fa9e4066Sahrens txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset); 671fa9e4066Sahrens 672fa9e4066Sahrens tn = tn->tn_next[t]; 673fa9e4066Sahrens 674fa9e4066Sahrens return (tn == NULL ? NULL : (char *)tn - tl->tl_offset); 675fa9e4066Sahrens } 676