txg.c revision 8f38d41910063e19709864b025684a228961299f
0N/A/*
1879N/A * CDDL HEADER START
0N/A *
0N/A * The contents of this file are subject to the terms of the
0N/A * Common Development and Distribution License (the "License").
0N/A * You may not use this file except in compliance with the License.
0N/A *
0N/A * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
0N/A * or http://www.opensolaris.org/os/licensing.
0N/A * See the License for the specific language governing permissions
0N/A * and limitations under the License.
0N/A *
0N/A * When distributing Covered Code, include this CDDL HEADER in each
0N/A * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
0N/A * If applicable, add the following below this CDDL HEADER, with the
0N/A * fields enclosed by brackets "[]" replaced with your own identifying
0N/A * information: Portions Copyright [yyyy] [name of copyright owner]
0N/A *
1472N/A * CDDL HEADER END
1472N/A */
1472N/A/*
0N/A * Copyright 2008 Sun Microsystems, Inc. All rights reserved.
0N/A * Use is subject to license terms.
0N/A */
1879N/A
1879N/A#pragma ident "%Z%%M% %I% %E% SMI"
1879N/A
1879N/A#include <sys/zfs_context.h>
1879N/A#include <sys/txg_impl.h>
1879N/A#include <sys/dmu_impl.h>
1879N/A#include <sys/dsl_pool.h>
1879N/A#include <sys/callb.h>
1879N/A
0N/A/*
0N/A * Pool-wide transaction groups.
0N/A */
0N/A
0N/Astatic void txg_sync_thread(dsl_pool_t *dp);
0N/Astatic void txg_quiesce_thread(dsl_pool_t *dp);
0N/Astatic void txg_timelimit_thread(dsl_pool_t *dp);
0N/A
0N/Aint txg_time = 5; /* max 5 seconds worth of delta per txg */
0N/A
0N/A/*
0N/A * Prepare the txg subsystem.
0N/A */
0N/Avoid
0N/Atxg_init(dsl_pool_t *dp, uint64_t txg)
0N/A{
0N/A tx_state_t *tx = &dp->dp_tx;
0N/A int c;
0N/A bzero(tx, sizeof (tx_state_t));
0N/A
0N/A tx->tx_cpu = kmem_zalloc(max_ncpus * sizeof (tx_cpu_t), KM_SLEEP);
0N/A
0N/A for (c = 0; c < max_ncpus; c++) {
0N/A int i;
0N/A
0N/A mutex_init(&tx->tx_cpu[c].tc_lock, NULL, MUTEX_DEFAULT, NULL);
0N/A for (i = 0; i < TXG_SIZE; i++) {
0N/A cv_init(&tx->tx_cpu[c].tc_cv[i], NULL, CV_DEFAULT,
0N/A NULL);
0N/A }
0N/A }
0N/A
0N/A rw_init(&tx->tx_suspend, NULL, RW_DEFAULT, NULL);
0N/A mutex_init(&tx->tx_sync_lock, NULL, MUTEX_DEFAULT, NULL);
0N/A
0N/A tx->tx_open_txg = txg;
0N/A}
0N/A
0N/A/*
0N/A * Close down the txg subsystem.
0N/A */
0N/Avoid
0N/Atxg_fini(dsl_pool_t *dp)
0N/A{
0N/A tx_state_t *tx = &dp->dp_tx;
0N/A int c;
0N/A
0N/A ASSERT(tx->tx_threads == 0);
0N/A
0N/A rw_destroy(&tx->tx_suspend);
0N/A mutex_destroy(&tx->tx_sync_lock);
0N/A
0N/A for (c = 0; c < max_ncpus; c++) {
0N/A int i;
0N/A
0N/A mutex_destroy(&tx->tx_cpu[c].tc_lock);
0N/A for (i = 0; i < TXG_SIZE; i++)
0N/A cv_destroy(&tx->tx_cpu[c].tc_cv[i]);
0N/A }
0N/A
0N/A kmem_free(tx->tx_cpu, max_ncpus * sizeof (tx_cpu_t));
0N/A
0N/A bzero(tx, sizeof (tx_state_t));
0N/A}
0N/A
0N/A/*
0N/A * Start syncing transaction groups.
0N/A */
0N/Avoid
0N/Atxg_sync_start(dsl_pool_t *dp)
0N/A{
0N/A tx_state_t *tx = &dp->dp_tx;
0N/A
0N/A mutex_enter(&tx->tx_sync_lock);
0N/A
1605N/A dprintf("pool %p\n", dp);
1605N/A
1605N/A ASSERT(tx->tx_threads == 0);
1605N/A
0N/A tx->tx_threads = 3;
0N/A
0N/A tx->tx_quiesce_thread = thread_create(NULL, 0, txg_quiesce_thread,
0N/A dp, 0, &p0, TS_RUN, minclsyspri);
0N/A
0N/A tx->tx_sync_thread = thread_create(NULL, 0, txg_sync_thread,
0N/A dp, 0, &p0, TS_RUN, minclsyspri);
0N/A
0N/A tx->tx_timelimit_thread = thread_create(NULL, 0, txg_timelimit_thread,
0N/A dp, 0, &p0, TS_RUN, minclsyspri);
1605N/A
1605N/A mutex_exit(&tx->tx_sync_lock);
1605N/A}
1605N/A
0N/Astatic void
0N/Atxg_thread_enter(tx_state_t *tx, callb_cpr_t *cpr)
0N/A{
432N/A CALLB_CPR_INIT(cpr, &tx->tx_sync_lock, callb_generic_cpr, FTAG);
432N/A mutex_enter(&tx->tx_sync_lock);
432N/A}
432N/A
432N/Astatic void
432N/Atxg_thread_exit(tx_state_t *tx, callb_cpr_t *cpr, kthread_t **tpp)
0N/A{
0N/A ASSERT(*tpp != NULL);
0N/A *tpp = NULL;
0N/A tx->tx_threads--;
0N/A cv_broadcast(&tx->tx_exit_cv);
0N/A CALLB_CPR_EXIT(cpr); /* drops &tx->tx_sync_lock */
0N/A thread_exit();
0N/A}
0N/A
0N/Astatic void
432N/Atxg_thread_wait(tx_state_t *tx, callb_cpr_t *cpr, kcondvar_t *cv, int secmax)
432N/A{
432N/A CALLB_CPR_SAFE_BEGIN(cpr);
432N/A
432N/A if (secmax)
0N/A (void) cv_timedwait(cv, &tx->tx_sync_lock, lbolt + secmax * hz);
0N/A else
0N/A cv_wait(cv, &tx->tx_sync_lock);
0N/A
0N/A CALLB_CPR_SAFE_END(cpr, &tx->tx_sync_lock);
0N/A}
0N/A
0N/A/*
0N/A * Stop syncing transaction groups.
0N/A */
0N/Avoid
0N/Atxg_sync_stop(dsl_pool_t *dp)
0N/A{
0N/A tx_state_t *tx = &dp->dp_tx;
0N/A
0N/A dprintf("pool %p\n", dp);
0N/A /*
0N/A * Finish off any work in progress.
0N/A */
0N/A ASSERT(tx->tx_threads == 3);
0N/A txg_wait_synced(dp, 0);
0N/A
0N/A /*
0N/A * Wake all 3 sync threads (one per state) and wait for them to die.
0N/A */
0N/A mutex_enter(&tx->tx_sync_lock);
0N/A
0N/A ASSERT(tx->tx_threads == 3);
0N/A
0N/A tx->tx_exiting = 1;
0N/A
0N/A cv_broadcast(&tx->tx_quiesce_more_cv);
0N/A cv_broadcast(&tx->tx_quiesce_done_cv);
0N/A cv_broadcast(&tx->tx_sync_more_cv);
0N/A cv_broadcast(&tx->tx_timeout_exit_cv);
0N/A
0N/A while (tx->tx_threads != 0)
0N/A cv_wait(&tx->tx_exit_cv, &tx->tx_sync_lock);
432N/A
0N/A tx->tx_exiting = 0;
0N/A
432N/A mutex_exit(&tx->tx_sync_lock);
432N/A}
432N/A
0N/Auint64_t
0N/Atxg_hold_open(dsl_pool_t *dp, txg_handle_t *th)
0N/A{
0N/A tx_state_t *tx = &dp->dp_tx;
0N/A tx_cpu_t *tc = &tx->tx_cpu[CPU_SEQID];
0N/A uint64_t txg;
0N/A
0N/A mutex_enter(&tc->tc_lock);
0N/A
0N/A txg = tx->tx_open_txg;
0N/A tc->tc_count[txg & TXG_MASK]++;
0N/A
0N/A th->th_cpu = tc;
0N/A th->th_txg = txg;
0N/A
0N/A return (txg);
0N/A}
0N/A
0N/Avoid
0N/Atxg_rele_to_quiesce(txg_handle_t *th)
0N/A{
0N/A tx_cpu_t *tc = th->th_cpu;
0N/A
0N/A mutex_exit(&tc->tc_lock);
0N/A}
0N/A
0N/Avoid
0N/Atxg_rele_to_sync(txg_handle_t *th)
0N/A{
0N/A tx_cpu_t *tc = th->th_cpu;
0N/A int g = th->th_txg & TXG_MASK;
0N/A
0N/A mutex_enter(&tc->tc_lock);
0N/A ASSERT(tc->tc_count[g] != 0);
0N/A if (--tc->tc_count[g] == 0)
0N/A cv_broadcast(&tc->tc_cv[g]);
0N/A mutex_exit(&tc->tc_lock);
0N/A
0N/A th->th_cpu = NULL; /* defensive */
0N/A}
0N/A
0N/Astatic void
0N/Atxg_quiesce(dsl_pool_t *dp, uint64_t txg)
0N/A{
0N/A tx_state_t *tx = &dp->dp_tx;
0N/A int g = txg & TXG_MASK;
0N/A int c;
0N/A
0N/A /*
0N/A * Grab all tx_cpu locks so nobody else can get into this txg.
0N/A */
0N/A for (c = 0; c < max_ncpus; c++)
0N/A mutex_enter(&tx->tx_cpu[c].tc_lock);
0N/A
0N/A ASSERT(txg == tx->tx_open_txg);
0N/A tx->tx_open_txg++;
0N/A
0N/A /*
0N/A * Now that we've incremented tx_open_txg, we can let threads
0N/A * enter the next transaction group.
0N/A */
0N/A for (c = 0; c < max_ncpus; c++)
0N/A mutex_exit(&tx->tx_cpu[c].tc_lock);
0N/A
0N/A /*
0N/A * Quiesce the transaction group by waiting for everyone to txg_exit().
0N/A */
0N/A for (c = 0; c < max_ncpus; c++) {
0N/A tx_cpu_t *tc = &tx->tx_cpu[c];
0N/A mutex_enter(&tc->tc_lock);
0N/A while (tc->tc_count[g] != 0)
0N/A cv_wait(&tc->tc_cv[g], &tc->tc_lock);
0N/A mutex_exit(&tc->tc_lock);
0N/A }
0N/A}
0N/A
0N/Astatic void
0N/Atxg_sync_thread(dsl_pool_t *dp)
0N/A{
0N/A tx_state_t *tx = &dp->dp_tx;
0N/A callb_cpr_t cpr;
0N/A
0N/A txg_thread_enter(tx, &cpr);
0N/A
0N/A for (;;) {
0N/A uint64_t txg;
0N/A
0N/A /*
0N/A * We sync when there's someone waiting on us, or the
0N/A * quiesce thread has handed off a txg to us.
0N/A */
0N/A while (!tx->tx_exiting &&
0N/A tx->tx_synced_txg >= tx->tx_sync_txg_waiting &&
0N/A tx->tx_quiesced_txg == 0) {
0N/A dprintf("waiting; tx_synced=%llu waiting=%llu dp=%p\n",
0N/A tx->tx_synced_txg, tx->tx_sync_txg_waiting, dp);
0N/A txg_thread_wait(tx, &cpr, &tx->tx_sync_more_cv, 0);
0N/A }
0N/A
0N/A /*
0N/A * Wait until the quiesce thread hands off a txg to us,
0N/A * prompting it to do so if necessary.
1080N/A */
1080N/A while (!tx->tx_exiting && tx->tx_quiesced_txg == 0) {
1080N/A if (tx->tx_quiesce_txg_waiting < tx->tx_open_txg+1)
1080N/A tx->tx_quiesce_txg_waiting = tx->tx_open_txg+1;
1080N/A cv_broadcast(&tx->tx_quiesce_more_cv);
1080N/A txg_thread_wait(tx, &cpr, &tx->tx_quiesce_done_cv, 0);
1080N/A }
1080N/A
1080N/A if (tx->tx_exiting)
1080N/A txg_thread_exit(tx, &cpr, &tx->tx_sync_thread);
1080N/A
0N/A rw_enter(&tx->tx_suspend, RW_WRITER);
0N/A
0N/A /*
0N/A * Consume the quiesced txg which has been handed off to
0N/A * us. This may cause the quiescing thread to now be
0N/A * able to quiesce another txg, so we must signal it.
0N/A */
0N/A txg = tx->tx_quiesced_txg;
0N/A tx->tx_quiesced_txg = 0;
0N/A tx->tx_syncing_txg = txg;
0N/A cv_broadcast(&tx->tx_quiesce_more_cv);
0N/A rw_exit(&tx->tx_suspend);
0N/A
0N/A dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
0N/A txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting);
0N/A mutex_exit(&tx->tx_sync_lock);
0N/A spa_sync(dp->dp_spa, txg);
0N/A mutex_enter(&tx->tx_sync_lock);
0N/A rw_enter(&tx->tx_suspend, RW_WRITER);
0N/A tx->tx_synced_txg = txg;
0N/A tx->tx_syncing_txg = 0;
0N/A rw_exit(&tx->tx_suspend);
0N/A cv_broadcast(&tx->tx_sync_done_cv);
0N/A }
0N/A}
0N/A
0N/Astatic void
0N/Atxg_quiesce_thread(dsl_pool_t *dp)
0N/A{
0N/A tx_state_t *tx = &dp->dp_tx;
0N/A callb_cpr_t cpr;
0N/A
0N/A txg_thread_enter(tx, &cpr);
0N/A
0N/A for (;;) {
0N/A uint64_t txg;
0N/A
0N/A /*
0N/A * We quiesce when there's someone waiting on us.
0N/A * However, we can only have one txg in "quiescing" or
0N/A * "quiesced, waiting to sync" state. So we wait until
0N/A * the "quiesced, waiting to sync" txg has been consumed
0N/A * by the sync thread.
0N/A */
0N/A while (!tx->tx_exiting &&
0N/A (tx->tx_open_txg >= tx->tx_quiesce_txg_waiting ||
0N/A tx->tx_quiesced_txg != 0))
0N/A txg_thread_wait(tx, &cpr, &tx->tx_quiesce_more_cv, 0);
0N/A
0N/A if (tx->tx_exiting)
0N/A txg_thread_exit(tx, &cpr, &tx->tx_quiesce_thread);
0N/A
0N/A txg = tx->tx_open_txg;
0N/A dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
0N/A txg, tx->tx_quiesce_txg_waiting,
0N/A tx->tx_sync_txg_waiting);
0N/A mutex_exit(&tx->tx_sync_lock);
0N/A txg_quiesce(dp, txg);
0N/A mutex_enter(&tx->tx_sync_lock);
0N/A
0N/A /*
0N/A * Hand this txg off to the sync thread.
0N/A */
0N/A dprintf("quiesce done, handing off txg %llu\n", txg);
1879N/A tx->tx_quiesced_txg = txg;
1879N/A cv_broadcast(&tx->tx_sync_more_cv);
cv_broadcast(&tx->tx_quiesce_done_cv);
}
}
void
txg_wait_synced(dsl_pool_t *dp, uint64_t txg)
{
tx_state_t *tx = &dp->dp_tx;
mutex_enter(&tx->tx_sync_lock);
ASSERT(tx->tx_threads == 3);
if (txg == 0)
txg = tx->tx_open_txg;
if (tx->tx_sync_txg_waiting < txg)
tx->tx_sync_txg_waiting = txg;
dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting);
while (tx->tx_synced_txg < txg) {
dprintf("broadcasting sync more "
"tx_synced=%llu waiting=%llu dp=%p\n",
tx->tx_synced_txg, tx->tx_sync_txg_waiting, dp);
cv_broadcast(&tx->tx_sync_more_cv);
cv_wait(&tx->tx_sync_done_cv, &tx->tx_sync_lock);
}
mutex_exit(&tx->tx_sync_lock);
}
void
txg_wait_open(dsl_pool_t *dp, uint64_t txg)
{
tx_state_t *tx = &dp->dp_tx;
mutex_enter(&tx->tx_sync_lock);
ASSERT(tx->tx_threads == 3);
if (txg == 0)
txg = tx->tx_open_txg + 1;
if (tx->tx_quiesce_txg_waiting < txg)
tx->tx_quiesce_txg_waiting = txg;
dprintf("txg=%llu quiesce_txg=%llu sync_txg=%llu\n",
txg, tx->tx_quiesce_txg_waiting, tx->tx_sync_txg_waiting);
while (tx->tx_open_txg < txg) {
cv_broadcast(&tx->tx_quiesce_more_cv);
cv_wait(&tx->tx_quiesce_done_cv, &tx->tx_sync_lock);
}
mutex_exit(&tx->tx_sync_lock);
}
static void
txg_timelimit_thread(dsl_pool_t *dp)
{
tx_state_t *tx = &dp->dp_tx;
callb_cpr_t cpr;
txg_thread_enter(tx, &cpr);
while (!tx->tx_exiting) {
uint64_t txg = tx->tx_open_txg + 1;
txg_thread_wait(tx, &cpr, &tx->tx_timeout_exit_cv, txg_time);
if (tx->tx_quiesce_txg_waiting < txg)
tx->tx_quiesce_txg_waiting = txg;
while (!tx->tx_exiting && tx->tx_open_txg < txg) {
dprintf("pushing out %llu\n", txg);
cv_broadcast(&tx->tx_quiesce_more_cv);
txg_thread_wait(tx, &cpr, &tx->tx_quiesce_done_cv, 0);
}
}
txg_thread_exit(tx, &cpr, &tx->tx_timelimit_thread);
}
int
txg_stalled(dsl_pool_t *dp)
{
tx_state_t *tx = &dp->dp_tx;
return (tx->tx_quiesce_txg_waiting > tx->tx_open_txg);
}
void
txg_suspend(dsl_pool_t *dp)
{
tx_state_t *tx = &dp->dp_tx;
/* XXX some code paths suspend when they are already suspended! */
rw_enter(&tx->tx_suspend, RW_READER);
}
void
txg_resume(dsl_pool_t *dp)
{
tx_state_t *tx = &dp->dp_tx;
rw_exit(&tx->tx_suspend);
}
/*
* Per-txg object lists.
*/
void
txg_list_create(txg_list_t *tl, size_t offset)
{
int t;
mutex_init(&tl->tl_lock, NULL, MUTEX_DEFAULT, NULL);
tl->tl_offset = offset;
for (t = 0; t < TXG_SIZE; t++)
tl->tl_head[t] = NULL;
}
void
txg_list_destroy(txg_list_t *tl)
{
int t;
for (t = 0; t < TXG_SIZE; t++)
ASSERT(txg_list_empty(tl, t));
mutex_destroy(&tl->tl_lock);
}
int
txg_list_empty(txg_list_t *tl, uint64_t txg)
{
return (tl->tl_head[txg & TXG_MASK] == NULL);
}
/*
* Add an entry to the list.
* Returns 0 if it's a new entry, 1 if it's already there.
*/
int
txg_list_add(txg_list_t *tl, void *p, uint64_t txg)
{
int t = txg & TXG_MASK;
txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
int already_on_list;
mutex_enter(&tl->tl_lock);
already_on_list = tn->tn_member[t];
if (!already_on_list) {
tn->tn_member[t] = 1;
tn->tn_next[t] = tl->tl_head[t];
tl->tl_head[t] = tn;
}
mutex_exit(&tl->tl_lock);
return (already_on_list);
}
/*
* Remove the head of the list and return it.
*/
void *
txg_list_remove(txg_list_t *tl, uint64_t txg)
{
int t = txg & TXG_MASK;
txg_node_t *tn;
void *p = NULL;
mutex_enter(&tl->tl_lock);
if ((tn = tl->tl_head[t]) != NULL) {
p = (char *)tn - tl->tl_offset;
tl->tl_head[t] = tn->tn_next[t];
tn->tn_next[t] = NULL;
tn->tn_member[t] = 0;
}
mutex_exit(&tl->tl_lock);
return (p);
}
/*
* Remove a specific item from the list and return it.
*/
void *
txg_list_remove_this(txg_list_t *tl, void *p, uint64_t txg)
{
int t = txg & TXG_MASK;
txg_node_t *tn, **tp;
mutex_enter(&tl->tl_lock);
for (tp = &tl->tl_head[t]; (tn = *tp) != NULL; tp = &tn->tn_next[t]) {
if ((char *)tn - tl->tl_offset == p) {
*tp = tn->tn_next[t];
tn->tn_next[t] = NULL;
tn->tn_member[t] = 0;
mutex_exit(&tl->tl_lock);
return (p);
}
}
mutex_exit(&tl->tl_lock);
return (NULL);
}
int
txg_list_member(txg_list_t *tl, void *p, uint64_t txg)
{
int t = txg & TXG_MASK;
txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
return (tn->tn_member[t]);
}
/*
* Walk a txg list -- only safe if you know it's not changing.
*/
void *
txg_list_head(txg_list_t *tl, uint64_t txg)
{
int t = txg & TXG_MASK;
txg_node_t *tn = tl->tl_head[t];
return (tn == NULL ? NULL : (char *)tn - tl->tl_offset);
}
void *
txg_list_next(txg_list_t *tl, void *p, uint64_t txg)
{
int t = txg & TXG_MASK;
txg_node_t *tn = (txg_node_t *)((char *)p + tl->tl_offset);
tn = tn->tn_next[t];
return (tn == NULL ? NULL : (char *)tn - tl->tl_offset);
}