Skip to content

M
mariadb
Commit 8192256d authored by John Esmet's avatar John Esmet Committed by Yoni Fogel
Browse files
Options

[t:4749] [t:4878] [t:4929] [t:4947] merging these changes to main. 


git-svn-id: file:///svn/toku/tokudb@44058 c7de825b-a66e-492c-adef-691d508d4ae1
parent 971862bb
Show whitespace changes
Inline Side-by-side
Showing with 298 additions and 414 deletions
ft/log.h
View file @ 8192256d
......@@ -19,8 +19,8 @@
extern "C" {
#endif
typedef void(*voidfp)(void *thunk);
typedef void(*YIELDF)(voidfp, void *fpthunk, void *yieldthunk);
// typedef void(*voidfp)(void *thunk);
// typedef void(*YIELDF)(voidfp, void *fpthunk, void *yieldthunk);
struct roll_entry;
......
ft/logformat.c
View file @ 8192256d
......@@ -298,10 +298,10 @@ generate_log_struct (void) {
fprintf(hf, "};\n");
fprintf(hf, "int toku_rollback_%s (", lt->name);
DO_FIELDS(field_type, lt, fprintf(hf, "%s %s,", field_type->type, field_type->name));
fprintf(hf, "TOKUTXN txn, YIELDF yield, void*yield_v, LSN oplsn);\n");
fprintf(hf, "TOKUTXN txn, LSN oplsn);\n");
fprintf(hf, "int toku_commit_%s (", lt->name);
DO_FIELDS(field_type, lt, fprintf(hf, "%s %s,", field_type->type, field_type->name));
fprintf(hf, "TOKUTXN txn, YIELDF yield, void*yield_v, LSN oplsn);\n");
fprintf(hf, "TOKUTXN txn, LSN oplsn);\n");
});
fprintf(hf, "struct log_entry {\n");
fprintf(hf, " enum lt_cmd cmd;\n");
......
ft/recover.c
View file @ 8192256d
......@@ -260,12 +260,6 @@ static const char *recover_state(RECOVER_ENV renv) {
return scan_state_string(&renv->ss);
}
// function supplied to transaction commit and abort
// No yielding is necessary, but it must call the f function if provided.
static void recover_yield(voidfp f, void *fpthunk, void *UU(yieldthunk)) {
if (f) f(fpthunk);
}
// Open the file if it is not already open. If it is already open, then do nothing.
static int internal_recover_fopen_or_fcreate (RECOVER_ENV renv, BOOL must_create, int UU(mode), BYTESTRING *bs_iname, FILENUM filenum, u_int32_t treeflags,
TOKUTXN txn, uint32_t nodesize, uint32_t basementnodesize, enum toku_compression_method compression_method, LSN max_acceptable_lsn) {
......@@ -663,7 +657,7 @@ static int toku_recover_xcommit (struct logtype_xcommit *l, RECOVER_ENV renv) {
assert(txn!=NULL);
// commit the transaction
r = toku_txn_commit_with_lsn(txn, TRUE, recover_yield, NULL, l->lsn,
r = toku_txn_commit_with_lsn(txn, TRUE, l->lsn,
NULL, NULL);
assert(r == 0);
......@@ -711,7 +705,7 @@ static int toku_recover_xabort (struct logtype_xabort *l, RECOVER_ENV renv) {
assert(txn!=NULL);
// abort the transaction
r = toku_txn_abort_with_lsn(txn, recover_yield, NULL, l->lsn, NULL, NULL);
r = toku_txn_abort_with_lsn(txn, l->lsn, NULL, NULL);
assert(r == 0);
// close the transaction
......@@ -1269,7 +1263,7 @@ static void recover_abort_live_txns(RECOVER_ENV renv) {
int r = find_an_unprepared_txn(renv, &txn);
if (r==0) {
// abort the transaction
r = toku_txn_abort_txn(txn, recover_yield, NULL, NULL, NULL);
r = toku_txn_abort_txn(txn, NULL, NULL);
assert(r == 0);
// close the transaction
......
ft/roll.c
View file @ 8192256d
......@@ -28,8 +28,6 @@
int
toku_commit_fdelete (FILENUM filenum,
TOKUTXN txn,
YIELDF UU(yield),
void *UU(yield_v),
LSN UU(oplsn)) //oplsn is the lsn of the commit
{
int r;
......@@ -64,9 +62,6 @@ toku_commit_fdelete (FILENUM filenum,
// after processing rollback entries. As a result, we may be unlinking a file
// here as part of a transactoin that may abort if we do not fsync the log.
// So, we fsync the log here.
//
// Because committing fdeletes should be a rare operation, we do not bother
// yielding the ydb lock before performing the fsync.
if (txn->logger) {
r = toku_logger_fsync_if_lsn_not_fsynced(txn->logger, txn->do_fsync_lsn);
assert_zero(r);
......@@ -90,8 +85,6 @@ done:
int
toku_rollback_fdelete (FILENUM UU(filenum),
TOKUTXN UU(txn),
YIELDF UU(yield),
void* UU(yield_v),
LSN UU(oplsn)) //oplsn is the lsn of the abort
{
//Rolling back an fdelete is an no-op.
......@@ -102,8 +95,6 @@ int
toku_commit_fcreate (FILENUM UU(filenum),
BYTESTRING UU(bs_fname),
TOKUTXN UU(txn),
YIELDF UU(yield),
void *UU(yield_v),
LSN UU(oplsn))
{
return 0;
......@@ -113,8 +104,6 @@ int
toku_rollback_fcreate (FILENUM filenum,
BYTESTRING UU(bs_fname),
TOKUTXN txn,
YIELDF UU(yield),
void* UU(yield_v),
LSN UU(oplsn))
{
int r;
......@@ -219,7 +208,7 @@ static int do_nothing_with_filenum(TOKUTXN UU(txn), FILENUM UU(filenum)) {
}
int toku_commit_cmdinsert (FILENUM filenum, BYTESTRING UU(key), TOKUTXN txn, YIELDF UU(yield), void *UU(yieldv), LSN UU(oplsn)) {
int toku_commit_cmdinsert (FILENUM filenum, BYTESTRING UU(key), TOKUTXN txn, LSN UU(oplsn)) {
#if TOKU_DO_COMMIT_CMD_INSERT
return do_insertion (FT_COMMIT_ANY, filenum, key, 0, txn, oplsn, FALSE);
#else
......@@ -231,8 +220,6 @@ int
toku_rollback_cmdinsert (FILENUM filenum,
BYTESTRING key,
TOKUTXN txn,
YIELDF UU(yield),
void * UU(yieldv),
LSN oplsn)
{
return do_insertion (FT_ABORT_ANY, filenum, key, 0, txn, oplsn, FALSE);
......@@ -242,8 +229,6 @@ int
toku_commit_cmdupdate(FILENUM filenum,
BYTESTRING key,
TOKUTXN txn,
YIELDF UU(yield),
void * UU(yieldv),
LSN oplsn)
{
return do_insertion(FT_COMMIT_ANY, filenum, key, 0, txn, oplsn, FALSE);
......@@ -253,8 +238,6 @@ int
toku_rollback_cmdupdate(FILENUM filenum,
BYTESTRING key,
TOKUTXN txn,
YIELDF UU(yield),
void * UU(yieldv),
LSN oplsn)
{
return do_insertion(FT_ABORT_ANY, filenum, key, 0, txn, oplsn, FALSE);
......@@ -264,8 +247,6 @@ int
toku_commit_cmdupdatebroadcast(FILENUM filenum,
BOOL is_resetting_op,
TOKUTXN txn,
YIELDF UU(yield),
void * UU(yieldv),
LSN oplsn)
{
// if is_resetting_op, reset root_xid_that_created in
......@@ -282,8 +263,6 @@ int
toku_rollback_cmdupdatebroadcast(FILENUM filenum,
BOOL UU(is_resetting_op),
TOKUTXN txn,
YIELDF UU(yield),
void * UU(yieldv),
LSN oplsn)
{
BYTESTRING nullkey = { 0, NULL };
......@@ -294,8 +273,6 @@ int
toku_commit_cmddelete (FILENUM filenum,
BYTESTRING key,
TOKUTXN txn,
YIELDF UU(yield),
void * UU(yieldv),
LSN oplsn)
{
#if TOKU_DO_COMMIT_CMD_DELETE
......@@ -310,8 +287,6 @@ int
toku_rollback_cmddelete (FILENUM filenum,
BYTESTRING key,
TOKUTXN txn,
YIELDF UU(yield),
void * UU(yieldv),
LSN oplsn)
{
return do_insertion (FT_ABORT_ANY, filenum, key, 0, txn, oplsn, FALSE);
......@@ -325,13 +300,10 @@ toku_apply_rollinclude (TXNID xid,
BLOCKNUM spilled_tail,
uint32_t spilled_tail_hash,
TOKUTXN txn,
YIELDF yield,
void * yieldv,
LSN oplsn,
apply_rollback_item func) {
int r = 0;
struct roll_entry *item;
int count=0;
BLOCKNUM next_log = spilled_tail;
uint32_t next_log_hash = spilled_tail_hash;
......@@ -350,10 +322,8 @@ toku_apply_rollinclude (TXNID xid,
while ((item=log->newest_logentry)) {
log->newest_logentry = item->prev;
r = func(txn, item, yield, yieldv, oplsn);
r = func(txn, item, oplsn);
if (r!=0) return r;
count++;
if (count%2 == 0) yield(NULL, NULL, yieldv);
}
if (next_log.b == spilled_head.b) {
assert(!found_head);
......@@ -386,14 +356,12 @@ toku_commit_rollinclude (TXNID xid,
BLOCKNUM spilled_tail,
uint32_t spilled_tail_hash,
TOKUTXN txn,
YIELDF yield,
void * yieldv,
LSN oplsn) {
int r;
r = toku_apply_rollinclude(xid, num_nodes,
spilled_head, spilled_head_hash,
spilled_tail, spilled_tail_hash,
txn, yield, yieldv, oplsn,
txn, oplsn,
toku_commit_rollback_item);
return r;
}
......@@ -406,14 +374,12 @@ toku_rollback_rollinclude (TXNID xid,
BLOCKNUM spilled_tail,
uint32_t spilled_tail_hash,
TOKUTXN txn,
YIELDF yield,
void * yieldv,
LSN oplsn) {
int r;
r = toku_apply_rollinclude(xid, num_nodes,
spilled_head, spilled_head_hash,
spilled_tail, spilled_tail_hash,
txn, yield, yieldv, oplsn,
txn, oplsn,
toku_abort_rollback_item);
return r;
}
......@@ -422,8 +388,6 @@ int
toku_commit_load (FILENUM old_filenum,
BYTESTRING UU(new_iname),
TOKUTXN txn,
YIELDF UU(yield),
void *UU(yield_v),
LSN UU(oplsn))
{
int r;
......@@ -452,9 +416,6 @@ toku_commit_load (FILENUM old_filenum,
// after processing rollback entries. As a result, we may be unlinking a file
// here as part of a transactoin that may abort if we do not fsync the log.
// So, we fsync the log here.
//
// Because committing fdeletes should be a rare operation, we do not bother
// yielding the ydb lock before performing the fsync.
if (txn->logger) {
r = toku_logger_fsync_if_lsn_not_fsynced(txn->logger, txn->do_fsync_lsn);
lazy_assert(r == 0);
......@@ -469,8 +430,6 @@ int
toku_rollback_load (FILENUM UU(old_filenum),
BYTESTRING new_iname,
TOKUTXN txn,
YIELDF UU(yield),
void *UU(yield_v),
LSN UU(oplsn))
{
int r;
......@@ -501,8 +460,6 @@ toku_rollback_load (FILENUM UU(old_filenum),
int
toku_commit_hot_index (FILENUMS UU(hot_index_filenums),
TOKUTXN UU(txn),
YIELDF UU(yield),
void * UU(yield_v),
LSN UU(oplsn))
{
// nothing
......@@ -512,8 +469,6 @@ toku_commit_hot_index (FILENUMS UU(hot_index_filenums),
int
toku_rollback_hot_index (FILENUMS UU(hot_index_filenums),
TOKUTXN UU(txn),
YIELDF UU(yield),
void * UU(yield_v),
LSN UU(oplsn))
{
return 0;
......@@ -523,8 +478,6 @@ int
toku_commit_dictionary_redirect (FILENUM UU(old_filenum),
FILENUM UU(new_filenum),
TOKUTXN UU(txn),
YIELDF UU(yield),
void * UU(yield_v),
LSN UU(oplsn)) //oplsn is the lsn of the commit
{
//Redirect only has meaning during normal operation (NOT during recovery).
......@@ -538,8 +491,6 @@ int
toku_rollback_dictionary_redirect (FILENUM old_filenum,
FILENUM new_filenum,
TOKUTXN txn,
YIELDF UU(yield),
void * UU(yield_v),
LSN UU(oplsn)) //oplsn is the lsn of the abort
{
int r = 0;
......@@ -566,8 +517,6 @@ int
toku_commit_change_fdescriptor(FILENUM filenum,
BYTESTRING UU(old_descriptor),
TOKUTXN txn,
YIELDF UU(yield),
void * UU(yieldv),
LSN UU(oplsn))
{
return do_nothing_with_filenum(txn, filenum);
......@@ -577,8 +526,6 @@ int
toku_rollback_change_fdescriptor(FILENUM filenum,
BYTESTRING old_descriptor,
TOKUTXN txn,
YIELDF UU(yield),
void * UU(yieldv),
LSN UU(oplsn))
{
CACHEFILE cf;
......
ft/rollback.c
View file @ 8192256d
......@@ -18,9 +18,9 @@ poll_txn_progress_function(TOKUTXN txn, uint8_t is_commit, uint8_t stall_for_che
}
}
int toku_commit_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv, LSN lsn) {
int toku_commit_rollback_item (TOKUTXN txn, struct roll_entry *item, LSN lsn) {
int r=0;
rolltype_dispatch_assign(item, toku_commit_, r, txn, yield, yieldv, lsn);
rolltype_dispatch_assign(item, toku_commit_, r, txn, lsn);
txn->num_rollentries_processed++;
if (txn->num_rollentries_processed % 1024 == 0) {
poll_txn_progress_function(txn, TRUE, FALSE);
......@@ -28,9 +28,9 @@ int toku_commit_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yiel
return r;
}
int toku_abort_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv, LSN lsn) {
int toku_abort_rollback_item (TOKUTXN txn, struct roll_entry *item, LSN lsn) {
int r=0;
rolltype_dispatch_assign(item, toku_rollback_, r, txn, yield, yieldv, lsn);
rolltype_dispatch_assign(item, toku_rollback_, r, txn, lsn);
txn->num_rollentries_processed++;
if (txn->num_rollentries_processed % 1024 == 0) {
poll_txn_progress_function(txn, FALSE, FALSE);
......@@ -68,14 +68,13 @@ void toku_rollback_log_unpin_and_remove(TOKUTXN txn, ROLLBACK_LOG_NODE log) {
}
static int
apply_txn (TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn,
apply_txn (TOKUTXN txn, LSN lsn,
apply_rollback_item func) {
int r = 0;
// do the commit/abort calls and free everything
// we do the commit/abort calls in reverse order too.
struct roll_entry *item;
//printf("%s:%d abort\n", __FILE__, __LINE__);
int count=0;
BLOCKNUM next_log = ROLLBACK_NONE;
uint32_t next_log_hash = 0;
......@@ -105,18 +104,8 @@ apply_txn (TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn,
if (func) {
while ((item=log->newest_logentry)) {
log->newest_logentry = item->prev;
r = func(txn, item, yield, yieldv, lsn);
r = func(txn, item, lsn);
if (r!=0) return r;
count++;
// We occassionally yield here to prevent transactions
// from hogging the log. This yield will allow other
// threads to grab the ydb lock. However, we don't
// want any transaction doing more than one log
// operation to always yield the ydb lock, as it must
// wait for the ydb lock to be released to proceed.
if (count % 8 == 0) {
yield(NULL, NULL, yieldv);
}
}
}
if (next_log.b == txn->spilled_rollback_head.b) {
......@@ -197,7 +186,7 @@ static int note_ft_used_in_txns_parent(OMTVALUE hv, u_int32_t UU(index), void*tx
//Commit each entry in the rollback log.
//If the transaction has a parent, it just promotes its information to its parent.
int toku_rollback_commit(TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn) {
int toku_rollback_commit(TOKUTXN txn, LSN lsn) {
int r=0;
if (txn->parent!=0) {
// First we must put a rollinclude entry into the parent if we spilled
......@@ -275,16 +264,16 @@ int toku_rollback_commit(TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn) {
txn->parent->force_fsync_on_commit |= txn->force_fsync_on_commit;
txn->parent->num_rollentries += txn->num_rollentries;
} else {
r = apply_txn(txn, yield, yieldv, lsn, toku_commit_rollback_item);
r = apply_txn(txn, lsn, toku_commit_rollback_item);
assert(r==0);
}
return r;
}
int toku_rollback_abort(TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn) {
int toku_rollback_abort(TOKUTXN txn, LSN lsn) {
int r;
r = apply_txn(txn, yield, yieldv, lsn, toku_abort_rollback_item);
r = apply_txn(txn, lsn, toku_abort_rollback_item);
assert(r==0);
return r;
}
......
ft/rollback.h
View file @ 8192256d
......@@ -14,8 +14,8 @@ extern "C" {
#endif
void toku_poll_txn_progress_function(TOKUTXN txn, uint8_t is_commit, uint8_t stall_for_checkpoint);
int toku_rollback_commit(TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn);
int toku_rollback_abort(TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn);
int toku_rollback_commit(TOKUTXN txn, LSN lsn);
int toku_rollback_abort(TOKUTXN txn, LSN lsn);
// these functions assert internally that they succeed
......@@ -38,9 +38,9 @@ void toku_maybe_prefetch_previous_rollback_log(TOKUTXN txn, ROLLBACK_LOG_NODE lo
// unpin and rmove a rollback log from the cachetable
void toku_rollback_log_unpin_and_remove(TOKUTXN txn, ROLLBACK_LOG_NODE log);
typedef int(*apply_rollback_item)(TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv, LSN lsn);
int toku_commit_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv, LSN lsn);
int toku_abort_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv, LSN lsn);
typedef int(*apply_rollback_item)(TOKUTXN txn, struct roll_entry *item, LSN lsn);
int toku_commit_rollback_item (TOKUTXN txn, struct roll_entry *item, LSN lsn);
int toku_abort_rollback_item (TOKUTXN txn, struct roll_entry *item, LSN lsn);
void *toku_malloc_in_rollback(ROLLBACK_LOG_NODE log, size_t size);
void *toku_memdup_in_rollback(ROLLBACK_LOG_NODE log, const void *v, size_t len);
......
ft/tests/is_empty.c
View file @ 8192256d
......@@ -13,11 +13,6 @@
#define TESTDIR __SRCFILE__ ".dir"
#define FILENAME "test0.ft_handle"
static void do_yield (voidfp f, void *fv, void *UU(v)) {
if (f) f(fv);
}
static void test_it (int N) {
FT_HANDLE brt;
int r;
......@@ -42,7 +37,7 @@ static void test_it (int N) {
r = toku_open_ft_handle(FILENAME, 1, &brt, 1024, 256, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, toku_builtin_compare_fun); CKERR(r);
r = toku_txn_commit_txn(txn, FALSE, do_yield, NULL, NULL, NULL); CKERR(r);
r = toku_txn_commit_txn(txn, FALSE, NULL, NULL); CKERR(r);
toku_txn_close_txn(txn);
r = toku_checkpoint(ct, logger, NULL, NULL, NULL, NULL, CLIENT_CHECKPOINT); CKERR(r);
......@@ -52,7 +47,7 @@ static void test_it (int N) {
for (int i=0; i<N; i++) {
r = toku_txn_begin_txn((DB_TXN*)NULL, (TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT); CKERR(r);
r = toku_open_ft_handle(FILENAME, 0, &brt, 1024, 256, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, toku_builtin_compare_fun); CKERR(r);
r = toku_txn_commit_txn(txn, FALSE, do_yield, NULL, NULL, NULL); CKERR(r);
r = toku_txn_commit_txn(txn, FALSE, NULL, NULL); CKERR(r);
toku_txn_close_txn(txn);
r = toku_txn_begin_txn((DB_TXN*)NULL, (TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT); CKERR(r);
......@@ -63,7 +58,7 @@ static void test_it (int N) {
memset(val, 'v', sizeof(val));
val[sizeof(val)-1]=0;
r = toku_ft_insert(brt, toku_fill_dbt(&k, key, 1+strlen(key)), toku_fill_dbt(&v, val, 1+strlen(val)), txn);
r = toku_txn_commit_txn(txn, FALSE, do_yield, NULL, NULL, NULL); CKERR(r);
r = toku_txn_commit_txn(txn, FALSE, NULL, NULL); CKERR(r);
toku_txn_close_txn(txn);
......@@ -75,7 +70,7 @@ static void test_it (int N) {
for (int i=0; i<N; i++) {
r = toku_txn_begin_txn((DB_TXN*)NULL, (TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT); CKERR(r);
r = toku_open_ft_handle(FILENAME, 0, &brt, 1024, 256, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, toku_builtin_compare_fun); CKERR(r);
r = toku_txn_commit_txn(txn, FALSE, do_yield, NULL, NULL, NULL); CKERR(r);
r = toku_txn_commit_txn(txn, FALSE, NULL, NULL); CKERR(r);
toku_txn_close_txn(txn);
r = toku_txn_begin_txn((DB_TXN*)NULL, (TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT); CKERR(r);
......@@ -90,7 +85,7 @@ static void test_it (int N) {
assert(!is_empty);
}
r = toku_txn_commit_txn(txn, FALSE, do_yield, NULL, NULL, NULL); CKERR(r);
r = toku_txn_commit_txn(txn, FALSE, NULL, NULL); CKERR(r);
toku_txn_close_txn(txn);
......@@ -101,7 +96,7 @@ static void test_it (int N) {
}
r = toku_txn_begin_txn((DB_TXN*)NULL, (TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT); CKERR(r);
r = toku_open_ft_handle(FILENAME, 0, &brt, 1024, 256, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, toku_builtin_compare_fun); CKERR(r);
r = toku_txn_commit_txn(txn, FALSE, do_yield, NULL, NULL, NULL); CKERR(r);
r = toku_txn_commit_txn(txn, FALSE, NULL, NULL); CKERR(r);
toku_txn_close_txn(txn);
if (0) {
......
ft/tests/le-cursor-provdel.c
View file @ 8192256d
......@@ -17,12 +17,6 @@ static int test_ft_cursor_keycompare(DB *desc __attribute__((unused)), const DBT
return toku_keycompare(a->data, a->size, b->data, b->size);
}
static void
txn_yield(voidfp UU(f), void *UU(fv), void *UU(v)) {
if (f)
f(fv);
}
// create a tree and populate it with n rows
static void
create_populate_tree(const char *logdir, const char *fname, int n) {
......@@ -49,7 +43,7 @@ create_populate_tree(const char *logdir, const char *fname, int n) {
error = toku_open_ft_handle(fname, 1, &brt, 1<<12, 1<<9, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, test_ft_cursor_keycompare);
assert(error == 0);
error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
error = toku_txn_commit_txn(txn, TRUE, NULL, NULL);
assert(error == 0);
toku_txn_close_txn(txn);
......@@ -69,7 +63,7 @@ create_populate_tree(const char *logdir, const char *fname, int n) {
assert(error == 0);
}
error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
error = toku_txn_commit_txn(txn, TRUE, NULL, NULL);
assert(error == 0);
toku_txn_close_txn(txn);
......@@ -115,7 +109,7 @@ test_provdel(const char *logdir, const char *fname, int n) {
error = toku_open_ft_handle(fname, 1, &brt, 1<<12, 1<<9, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, test_ft_cursor_keycompare);
assert(error == 0);
error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
error = toku_txn_commit_txn(txn, TRUE, NULL, NULL);
assert(error == 0);
toku_txn_close_txn(txn);
......@@ -166,11 +160,11 @@ test_provdel(const char *logdir, const char *fname, int n) {
error = le_cursor_close(cursor);
assert(error == 0);
error = toku_txn_commit_txn(cursortxn, TRUE, txn_yield, NULL, NULL, NULL);
error = toku_txn_commit_txn(cursortxn, TRUE, NULL, NULL);
assert(error == 0);
toku_txn_close_txn(cursortxn);
error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
error = toku_txn_commit_txn(txn, TRUE, NULL, NULL);
assert(error == 0);
toku_txn_close_txn(txn);
......
ft/tests/le-cursor-right.c
View file @ 8192256d
......@@ -21,12 +21,6 @@ test_keycompare(DB* UU(desc), const DBT *a, const DBT *b) {
return toku_keycompare(a->data, a->size, b->data, b->size);
}
static void
txn_yield(voidfp UU(f), void *UU(fv), void *UU(v)) {
if (f)
f(fv);
}
// create a tree and populate it with n rows
static void
create_populate_tree(const char *logdir, const char *fname, int n) {
......@@ -53,7 +47,7 @@ create_populate_tree(const char *logdir, const char *fname, int n) {
error = toku_open_ft_handle(fname, 1, &brt, 1<<12, 1<<9, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, test_keycompare);
assert(error == 0);
error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
error = toku_txn_commit_txn(txn, TRUE, NULL, NULL);
assert(error == 0);
toku_txn_close_txn(txn);
......@@ -73,7 +67,7 @@ create_populate_tree(const char *logdir, const char *fname, int n) {
assert(error == 0);
}
error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
error = toku_txn_commit_txn(txn, TRUE, NULL, NULL);
assert(error == 0);
toku_txn_close_txn(txn);
......
ft/tests/le-cursor-walk.c
View file @ 8192256d
......@@ -17,12 +17,6 @@ static int test_ft_cursor_keycompare(DB *db __attribute__((unused)), const DBT *
return toku_keycompare(a->data, a->size, b->data, b->size);
}
static void
txn_yield(voidfp UU(f), void *UU(fv), void *UU(v)) {
if (f)
f(fv);
}
// create a tree and populate it with n rows
static void
create_populate_tree(const char *logdir, const char *fname, int n) {
......@@ -49,7 +43,7 @@ create_populate_tree(const char *logdir, const char *fname, int n) {
error = toku_open_ft_handle(fname, 1, &brt, 1<<12, 1<<9, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, test_ft_cursor_keycompare);
assert(error == 0);
error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
error = toku_txn_commit_txn(txn, TRUE, NULL, NULL);
assert(error == 0);
toku_txn_close_txn(txn);
......@@ -69,7 +63,7 @@ create_populate_tree(const char *logdir, const char *fname, int n) {
assert(error == 0);
}
error = toku_txn_commit_txn(txn, TRUE, txn_yield, NULL, NULL, NULL);
error = toku_txn_commit_txn(txn, TRUE, NULL, NULL);
assert(error == 0);
toku_txn_close_txn(txn);
......
ft/txn.c
View file @ 8192256d
......@@ -182,12 +182,12 @@ toku_txn_load_txninfo (TOKUTXN txn, TXNINFO info) {
return 0;
}
int toku_txn_commit_txn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv,
int toku_txn_commit_txn(TOKUTXN txn, int nosync,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra)
// Effect: Doesn't close the txn, just performs the commit operations.
// If release_multi_operation_client_lock is true, then unlock that lock (even if an error path is taken)
{
return toku_txn_commit_with_lsn(txn, nosync, yield, yieldv, ZERO_LSN,
return toku_txn_commit_with_lsn(txn, nosync, ZERO_LSN,
poll, poll_extra);
}
......@@ -206,7 +206,7 @@ BOOL toku_txn_requires_checkpoint(TOKUTXN txn) {
return (!txn->parent && txn->checkpoint_needed_before_commit);
}
//Called during a yield (ydb lock NOT held).
//TODO(yoni): inline this function manually
static void
log_xcommit(void *thunk) {
struct xcommit_info *info = thunk;
......@@ -214,7 +214,7 @@ log_xcommit(void *thunk) {
info->r = toku_log_xcommit(txn->logger, &txn->do_fsync_lsn, 0, txn->txnid64); // exits holding neither of the tokulogger locks.
}
int toku_txn_commit_with_lsn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv, LSN oplsn,
int toku_txn_commit_with_lsn(TOKUTXN txn, int nosync, LSN oplsn,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra)
// Effect: Among other things: if release_multi_operation_client_lock is true, then unlock that lock (even if an error path is taken)
{
......@@ -245,21 +245,21 @@ int toku_txn_commit_with_lsn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv
r = info.r;
}
if (r==0) {
r = toku_rollback_commit(txn, yield, yieldv, oplsn);
r = toku_rollback_commit(txn, oplsn);
STATUS_VALUE(TXN_COMMIT)++;
}
return r;
}
int toku_txn_abort_txn(TOKUTXN txn, YIELDF yield, void *yieldv,
int toku_txn_abort_txn(TOKUTXN txn,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra)
// Effect: Doesn't close the txn, just performs the abort operations.
// If release_multi_operation_client_lock is true, then unlock that lock (even if an error path is taken)
{
return toku_txn_abort_with_lsn(txn, yield, yieldv, ZERO_LSN, poll, poll_extra);
return toku_txn_abort_with_lsn(txn, ZERO_LSN, poll, poll_extra);
}
int toku_txn_abort_with_lsn(TOKUTXN txn, YIELDF yield, void *yieldv, LSN oplsn,
int toku_txn_abort_with_lsn(TOKUTXN txn, LSN oplsn,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra)
// Effect: Ammong other things, if release_multi_operation_client_lock is true, then unlock that lock (even if an error path is taken)
{
......@@ -271,7 +271,7 @@ int toku_txn_abort_with_lsn(TOKUTXN txn, YIELDF yield, void *yieldv, LSN oplsn,
txn->do_fsync = FALSE;
r = toku_log_xabort(txn->logger, &txn->do_fsync_lsn, 0, txn->txnid64);
if (r==0) {
r = toku_rollback_abort(txn, yield, yieldv, oplsn);
r = toku_rollback_abort(txn, oplsn);
STATUS_VALUE(TXN_ABORT)++;
}
return r;
......@@ -320,11 +320,12 @@ static void do_txn_fsync_log(void *thunk) {
info->r = toku_logger_fsync_if_lsn_not_fsynced(info->logger, info->do_fsync_lsn);
}
int toku_txn_maybe_fsync_log(TOKULOGGER logger, LSN do_fsync_lsn, BOOL do_fsync, YIELDF yield, void *yieldv) {
int toku_txn_maybe_fsync_log(TOKULOGGER logger, LSN do_fsync_lsn, BOOL do_fsync) {
int r = 0;
if (logger && do_fsync) {
struct txn_fsync_log_info info = { .logger = logger, .do_fsync_lsn = do_fsync_lsn };
yield(do_txn_fsync_log, &info, yieldv);
//TODO(yoni): inline do_txn_fsync_log here
do_txn_fsync_log(&info);
r = info.r;
}
return r;
......
ft/txn.h
View file @ 8192256d
......@@ -37,15 +37,15 @@ int toku_txn_create_txn(TOKUTXN *txn_ptr, TOKUTXN parent, TOKULOGGER logger, TXN
int toku_txn_load_txninfo (TOKUTXN txn, TXNINFO info);
int toku_txn_commit_txn (TOKUTXN txn, int nosync, YIELDF yield, void *yieldv,
int toku_txn_commit_txn (TOKUTXN txn, int nosync,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra);
BOOL toku_txn_requires_checkpoint(TOKUTXN txn);
int toku_txn_commit_with_lsn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv, LSN oplsn,
int toku_txn_commit_with_lsn(TOKUTXN txn, int nosync, LSN oplsn,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra);
int toku_txn_abort_txn(TOKUTXN txn, YIELDF yield, void *yieldv,
int toku_txn_abort_txn(TOKUTXN txn,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra);
int toku_txn_abort_with_lsn(TOKUTXN txn, YIELDF yield, void *yieldv, LSN oplsn,
int toku_txn_abort_with_lsn(TOKUTXN txn, LSN oplsn,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra);
int toku_txn_prepare_txn (TOKUTXN txn, TOKU_XA_XID *xid) __attribute__((warn_unused_result));
......@@ -54,7 +54,7 @@ int toku_txn_prepare_txn (TOKUTXN txn, TOKU_XA_XID *xid) __attribute__((warn_unu
void toku_txn_get_prepared_xa_xid (TOKUTXN, TOKU_XA_XID *);
// Effect: Fill in the XID information for a transaction. The caller allocates the XID and the function fills in values.
int toku_txn_maybe_fsync_log(TOKULOGGER logger, LSN do_fsync_lsn, BOOL do_fsync, YIELDF yield, void *yieldv);
int toku_txn_maybe_fsync_log(TOKULOGGER logger, LSN do_fsync_lsn, BOOL do_fsync);
void toku_txn_get_fsync_info(TOKUTXN ttxn, BOOL* do_fsync, LSN* do_fsync_lsn);
......
src/loader.c
View file @ 8192256d
......@@ -254,7 +254,7 @@ int toku_loader_create_loader(DB_ENV *env,
loader->i->ekeys = NULL;
loader->i->evals = NULL;
LSN load_lsn;
r = ydb_load_inames(env, txn, N, dbs, new_inames_in_env, &load_lsn, use_ft_loader);
r = locked_load_inames(env, txn, N, dbs, new_inames_in_env, &load_lsn, use_ft_loader);
if ( r!=0 ) {
toku_free(new_inames_in_env);
toku_free(brts);
......
src/ydb-internal.h
View file @ 8192256d
......@@ -73,6 +73,7 @@ struct __toku_db_env_internal {
DB *directory; // Maps dnames to inames
DB *persistent_environment; // Stores environment settings, can be used for upgrade
OMT open_dbs; // Stores open db handles, sorted first by dname and then by numerical value of pointer to the db (arbitrarily assigned memory location)
toku_mutex_t open_dbs_lock; // lock that protects the OMT of open dbs.
char *real_data_dir; // data dir used when the env is opened (relative to cwd, or absolute with leading /)
char *real_log_dir; // log dir used when the env is opened (relative to cwd, or absolute with leading /)
......
src/ydb.c
View file @ 8192256d
......@@ -31,7 +31,6 @@ const char *toku_copyright_string = "Copyright (c) 2007-2009 Tokutek Inc. All r
#include <ft/key.h>
#include "loader.h"
#include "indexer.h"
#include "ydb_load.h"
#include <ft/ftloader.h>
#include <ft/log_header.h>
#include <ft/ft.h>
......@@ -396,7 +395,6 @@ static void keep_cachetable_callback (DB_ENV *env, CACHETABLE cachetable)
static int
ydb_do_recovery (DB_ENV *env) {
assert(env->i->real_log_dir);
toku_ydb_unlock();
int r = tokudb_recover(env,
toku_keep_prepared_txn_callback,
keep_cachetable_callback,
......@@ -405,7 +403,6 @@ ydb_do_recovery (DB_ENV *env) {
env->i->update_function,
env->i->generate_row_for_put, env->i->generate_row_for_del,
env->i->cachetable_size);
toku_ydb_lock();
return r;
}
......@@ -726,9 +723,7 @@ static int
ydb_maybe_upgrade_env (DB_ENV *env, LSN * last_lsn_of_clean_shutdown_read_from_log, BOOL * upgrade_in_progress) {
int r = 0;
if (env->i->open_flags & DB_INIT_TXN && env->i->open_flags & DB_INIT_LOG) {
toku_ydb_unlock();
r = toku_maybe_upgrade_log(env->i->dir, env->i->real_log_dir, last_lsn_of_clean_shutdown_read_from_log, upgrade_in_progress);
toku_ydb_lock();
}
return r;
}
......@@ -949,7 +944,7 @@ toku_env_open(DB_ENV * env, const char *home, u_int32_t flags, int mode) {
DB_TXN *txn=NULL;
if (using_txns) {
r = toku_txn_begin(env, 0, &txn, 0, 1, true);
r = locked_txn_begin(env, 0, &txn, 0);
assert_zero(r);
}
......@@ -997,13 +992,11 @@ toku_env_open(DB_ENV * env, const char *home, u_int32_t flags, int mode) {
assert_zero(r);
}
if (using_txns) {
r = toku_txn_commit(txn, 0, NULL, NULL, false);
r = locked_txn_commit(txn, 0);
assert_zero(r);
}
toku_ydb_unlock();
r = toku_checkpoint(env->i->cachetable, env->i->logger, NULL, NULL, NULL, NULL, STARTUP_CHECKPOINT);
assert_zero(r);
toku_ydb_lock();
env_fs_poller(env); // get the file system state at startup
env_fs_init_minicron(env);
cleanup:
......@@ -1064,7 +1057,6 @@ toku_env_close(DB_ENV * env, u_int32_t flags) {
}
}
if (env->i->cachetable) {
toku_ydb_unlock(); // ydb lock must not be held when shutting down minicron
toku_cachetable_minicron_shutdown(env->i->cachetable);
if (env->i->logger) {
r = toku_checkpoint(env->i->cachetable, env->i->logger, NULL, NULL, NULL, NULL, SHUTDOWN_CHECKPOINT);
......@@ -1093,7 +1085,6 @@ toku_env_close(DB_ENV * env, u_int32_t flags) {
goto panic_and_quit_early;
}
}
toku_ydb_lock();
r=toku_cachetable_close(&env->i->cachetable);
if (r) {
err_msg = "Cannot close environment (cachetable close error)\n";
......@@ -1136,6 +1127,7 @@ toku_env_close(DB_ENV * env, u_int32_t flags) {
toku_free(env->i->real_tmp_dir);
if (env->i->open_dbs)
toku_omt_destroy(&env->i->open_dbs);
toku_mutex_destroy(&env->i->open_dbs_lock);
if (env->i->dir)
toku_free(env->i->dir);
//Immediately before freeing internal environment unlock the directories
......@@ -1189,21 +1181,59 @@ toku_env_set_cachesize(DB_ENV * env, u_int32_t gbytes, u_int32_t bytes, int ncac
static int
locked_env_dbremove(DB_ENV * env, DB_TXN *txn, const char *fname, const char *dbname, u_int32_t flags) {
toku_multi_operation_client_lock(); //Cannot begin checkpoint
toku_ydb_lock();
int r = toku_env_dbremove(env, txn, fname, dbname, flags);
toku_ydb_unlock();
toku_multi_operation_client_unlock(); //Can now begin checkpoint
int ret, r;
HANDLE_ILLEGAL_WORKING_PARENT_TXN(env, txn);
DB_TXN *child_txn = NULL;
int using_txns = env->i->open_flags & DB_INIT_TXN;
if (using_txns) {
ret = locked_txn_begin(env, txn, &child_txn, DB_TXN_NOSYNC);
invariant_zero(ret);
}
// cannot begin a checkpoint
toku_multi_operation_client_lock();
r = toku_env_dbremove(env, child_txn, fname, dbname, flags);
toku_multi_operation_client_unlock();
if (using_txns) {
if (r == 0) { // commit
ret = locked_txn_commit(child_txn, DB_TXN_NOSYNC);
invariant_zero(ret);
} else {
ret = locked_txn_abort(child_txn);
invariant_zero(ret);
}
}
return r;
}
static int
locked_env_dbrename(DB_ENV *env, DB_TXN *txn, const char *fname, const char *dbname, const char *newname, u_int32_t flags) {
toku_multi_operation_client_lock(); //Cannot begin checkpoint
toku_ydb_lock();
int r = toku_env_dbrename(env, txn, fname, dbname, newname, flags);
toku_ydb_unlock();
toku_multi_operation_client_unlock(); //Can now begin checkpoint
int ret, r;
HANDLE_ILLEGAL_WORKING_PARENT_TXN(env, txn);
DB_TXN *child_txn = NULL;
int using_txns = env->i->open_flags & DB_INIT_TXN;
if (using_txns) {
ret = locked_txn_begin(env, txn, &child_txn, DB_TXN_NOSYNC);
invariant_zero(ret);
}
// cannot begin a checkpoint
toku_multi_operation_client_lock();
r = toku_env_dbrename(env, child_txn, fname, dbname, newname, flags);
toku_multi_operation_client_unlock();
if (using_txns) {
if (r == 0) {
ret = locked_txn_commit(child_txn, DB_TXN_NOSYNC);
invariant_zero(ret);
} else {
ret = locked_txn_abort(child_txn);
invariant_zero(ret);
}
}
return r;
}
......@@ -1454,16 +1484,6 @@ toku_env_txn_stat(DB_ENV * env, DB_TXN_STAT ** UU(statp), u_int32_t UU(flags)) {
return 1;
}
static int
locked_env_open(DB_ENV * env, const char *home, u_int32_t flags, int mode) {
toku_ydb_lock(); int r = toku_env_open(env, home, flags, mode); toku_ydb_unlock(); return r;
}
static int
locked_env_close(DB_ENV * env, u_int32_t flags) {
toku_ydb_lock(); int r = toku_env_close(env, flags); toku_ydb_unlock(); return r;
}
static int
toku_env_txn_xa_recover (DB_ENV *env, TOKU_XA_XID xids[/*count*/], long count, /*out*/ long *retp, u_int32_t flags) {
struct tokulogger_preplist *MALLOC_N(count,preps);
......@@ -2391,8 +2411,6 @@ toku_env_create(DB_ENV ** envp, u_int32_t flags) {
SENV(cleaner_get_period);
SENV(cleaner_set_iterations);
SENV(cleaner_get_iterations);
SENV(open);
SENV(close);
SENV(txn_recover);
SENV(txn_xa_recover);
SENV(get_txn_from_xid);
......@@ -2432,6 +2450,8 @@ toku_env_create(DB_ENV ** envp, u_int32_t flags) {
result->update_multiple = env_update_multiple;
// unlocked methods
result->open = toku_env_open;
result->close = toku_env_close;
result->txn_checkpoint = toku_env_txn_checkpoint;
result->checkpointing_postpone = env_checkpointing_postpone;
result->checkpointing_resume = env_checkpointing_resume;
......@@ -2471,6 +2491,7 @@ toku_env_create(DB_ENV ** envp, u_int32_t flags) {
assert(result->i->ltm);
r = toku_omt_create(&result->i->open_dbs);
toku_mutex_init(&result->i->open_dbs_lock, NULL);
assert_zero(r);
assert(result->i->open_dbs);
......@@ -2488,9 +2509,7 @@ cleanup:
int
DB_ENV_CREATE_FUN (DB_ENV ** envp, u_int32_t flags) {
toku_ydb_lock();
int r = toku_env_create(envp, flags);
toku_ydb_unlock();
return r;
}
......@@ -2523,6 +2542,7 @@ find_db_by_db (OMTVALUE v, void *dbv) {
// Tell env that there is a new db handle (with non-unique dname in db->i-dname)
void
env_note_db_opened(DB_ENV *env, DB *db) {
toku_mutex_lock(&env->i->open_dbs_lock);
assert(db->i->dname); // internal (non-user) dictionary has no dname
int r;
OMTVALUE dbv;
......@@ -2535,14 +2555,15 @@ env_note_db_opened(DB_ENV *env, DB *db) {
assert(r==DB_NOTFOUND); //Must not already be there.
r = toku_omt_insert_at(env->i->open_dbs, db, idx);
assert_zero(r);
toku_mutex_unlock(&env->i->open_dbs_lock);
}
void
env_note_db_closed(DB_ENV *env, DB *db)
// Effect: Tell the DB_ENV that the DB is no longer in use by the user of the API. The DB may still be in use by the fractal tree internals.
{
assert(db->i->dname);
assert(toku_omt_size(env->i->open_dbs));
void
env_note_db_closed(DB_ENV *env, DB *db) {
toku_mutex_lock(&env->i->open_dbs_lock);
assert(db->i->dname); // internal (non-user) dictionary has no dname
assert(toku_omt_size(env->i->open_dbs) > 0);
int r;
OMTVALUE dbv;
uint32_t idx;
......@@ -2553,6 +2574,7 @@ env_note_db_closed(DB_ENV *env, DB *db)
r = toku_omt_delete_at(env->i->open_dbs, idx);
STATUS_VALUE(YDB_LAYER_NUM_OPEN_DBS) = toku_omt_size(env->i->open_dbs);
assert_zero(r);
toku_mutex_unlock(&env->i->open_dbs_lock);
}
static int
......@@ -2572,6 +2594,7 @@ env_is_db_with_dname_open(DB_ENV *env, const char *dname) {
BOOL rval;
OMTVALUE dbv;
uint32_t idx;
toku_mutex_lock(&env->i->open_dbs_lock);
r = toku_omt_find_zero(env->i->open_dbs, find_open_db_by_dname, (void*)dname, &dbv, &idx);
if (r==0) {
DB *db = dbv;
......@@ -2582,6 +2605,7 @@ env_is_db_with_dname_open(DB_ENV *env, const char *dname) {
assert(r==DB_NOTFOUND);
rval = FALSE;
}
toku_mutex_unlock(&env->i->open_dbs_lock);
return rval;
}
......@@ -2618,8 +2642,7 @@ finalize_file_removal(DICTIONARY_ID dict_id, void * extra) {
// returns: true if we could open, lock, and close a dictionary
// with the given dname, false otherwise.
static bool
can_acquire_table_lock(DB_ENV *env, DB_TXN *txn, const char *iname_in_env)
{
can_acquire_table_lock(DB_ENV *env, DB_TXN *txn, const char *iname_in_env) {
int r;
bool got_lock = false;
DB *db;
......@@ -2627,9 +2650,6 @@ can_acquire_table_lock(DB_ENV *env, DB_TXN *txn, const char *iname_in_env)
r = toku_db_create(&db, env, 0);
assert_zero(r);
r = db_open_iname(db, txn, iname_in_env, 0, 0);
if (r != 0) {
printf("%s had db_open_iname return %d\n", __FUNCTION__, r);
}
assert_zero(r);
r = toku_db_pre_acquire_table_lock(db, txn);
if (r == 0) {
......@@ -2647,43 +2667,37 @@ int
toku_env_dbremove(DB_ENV * env, DB_TXN *txn, const char *fname, const char *dbname, u_int32_t flags) {
int r;
HANDLE_PANICKED_ENV(env);
HANDLE_ILLEGAL_WORKING_PARENT_TXN(env, txn);
if (!env_opened(env)) return EINVAL;
if (dbname!=NULL)
return env_dbremove_subdb(env, txn, fname, dbname, flags);
if (!env_opened(env) || flags != 0) {
return EINVAL;
}
if (dbname != NULL) {
// env_dbremove_subdb() converts (fname, dbname) to dname
return env_dbremove_subdb(env, txn, fname, dbname, flags);
}
const char * dname = fname;
assert(dbname == NULL);
if (flags!=0) return EINVAL;
// We check for an open db here as a "fast path" to error.
// We'll need to check again below to be sure.
if (env_is_db_with_dname_open(env, dname))
if (env_is_db_with_dname_open(env, dname)) {
return toku_ydb_do_error(env, EINVAL, "Cannot remove dictionary with an open handle.\n");
}
DBT dname_dbt;
DBT iname_dbt;
toku_fill_dbt(&dname_dbt, dname, strlen(dname)+1);
init_dbt_realloc(&iname_dbt); // sets iname_dbt.data = NULL
int using_txns = env->i->open_flags & DB_INIT_TXN;
DB_TXN *child = NULL;
// begin child (unless transactionless)
if (using_txns) {
r = toku_txn_begin(env, txn, &child, DB_TXN_NOSYNC, 1, true);
assert_zero(r);
}
// get iname
r = toku_db_get(env->i->directory, child, &dname_dbt, &iname_dbt, DB_SERIALIZABLE); // allocates memory for iname
r = toku_db_get(env->i->directory, txn, &dname_dbt, &iname_dbt, DB_SERIALIZABLE); // allocates memory for iname
char *iname = iname_dbt.data;
DB *db = NULL;
if (r == DB_NOTFOUND) {
r = ENOENT;
} else if (r == 0) {
// remove (dname,iname) from directory
r = toku_db_del(env->i->directory, child, &dname_dbt, DB_DELETE_ANY, TRUE);
r = toku_db_del(env->i->directory, txn, &dname_dbt, DB_DELETE_ANY, TRUE);
if (r != 0) {
goto exit;
}
......@@ -2691,7 +2705,7 @@ toku_env_dbremove(DB_ENV * env, DB_TXN *txn, const char *fname, const char *dbna
assert_zero(r);
r = db_open_iname(db, txn, iname, 0, 0);
assert_zero(r);
if (using_txns) {
if (txn) {
// Now that we have a writelock on dname, verify that there are still no handles open. (to prevent race conditions)
if (env_is_db_with_dname_open(env, dname)) {
r = toku_ydb_do_error(env, EINVAL, "Cannot remove dictionary with an open handle.\n");
......@@ -2707,11 +2721,11 @@ toku_env_dbremove(DB_ENV * env, DB_TXN *txn, const char *fname, const char *dbna
// otherwise, we're okay in marking this ft as remove on
// commit. no new handles can open for this dictionary
// because the txn has directory write locks on the dname
if (toku_db_pre_acquire_table_lock(db, child) != 0) {
if (toku_db_pre_acquire_table_lock(db, txn) != 0) {
r = DB_LOCK_NOTGRANTED;
} else {
// The ft will be removed when the txn commits
r = toku_ft_remove_on_commit(db->i->ft_handle, db_txn_struct_i(child)->tokutxn);
r = toku_ft_remove_on_commit(db->i->ft_handle, db_txn_struct_i(txn)->tokutxn);
assert_zero(r);
}
}
......@@ -2726,17 +2740,6 @@ exit:
int ret = toku_db_close(db);
assert(ret == 0);
}
if (using_txns) {
// close txn
if (r == 0) { // commit
r = toku_txn_commit(child, DB_TXN_NOSYNC, NULL, NULL, false);
invariant(r==0); // TODO panic
}
else { // abort
int r2 = toku_txn_abort(child, NULL, NULL, false);
invariant(r2==0); // TODO panic
}
}
if (iname) {
toku_free(iname);
}
......@@ -2769,22 +2772,25 @@ int
toku_env_dbrename(DB_ENV *env, DB_TXN *txn, const char *fname, const char *dbname, const char *newname, u_int32_t flags) {
int r;
HANDLE_PANICKED_ENV(env);
HANDLE_ILLEGAL_WORKING_PARENT_TXN(env, txn);
if (!env_opened(env)) return EINVAL;
if (dbname!=NULL)
return env_dbrename_subdb(env, txn, fname, dbname, newname, flags);
if (!env_opened(env) || flags != 0) {
return EINVAL;
}
if (dbname != NULL) {
// env_dbrename_subdb() converts (fname, dbname) to dname and (fname, newname) to newdname
return env_dbrename_subdb(env, txn, fname, dbname, newname, flags);
}
const char * dname = fname;
assert(dbname == NULL);
if (flags != 0) return EINVAL;
// We check for open dnames for the old and new name as a "fast path" to error.
// We will need to check these again later.
if (env_is_db_with_dname_open(env, dname))
if (env_is_db_with_dname_open(env, dname)) {
return toku_ydb_do_error(env, EINVAL, "Cannot rename dictionary with an open handle.\n");
if (env_is_db_with_dname_open(env, newname))
}
if (env_is_db_with_dname_open(env, newname)) {
return toku_ydb_do_error(env, EINVAL, "Cannot rename dictionary; Dictionary with target name has an open handle.\n");
}
DBT old_dname_dbt;
DBT new_dname_dbt;
......@@ -2793,30 +2799,22 @@ toku_env_dbrename(DB_ENV *env, DB_TXN *txn, const char *fname, const char *dbnam
toku_fill_dbt(&new_dname_dbt, newname, strlen(newname)+1);
init_dbt_realloc(&iname_dbt); // sets iname_dbt.data = NULL
int using_txns = env->i->open_flags & DB_INIT_TXN;
DB_TXN *child = NULL;
// begin child (unless transactionless)
if (using_txns) {
r = toku_txn_begin(env, txn, &child, DB_TXN_NOSYNC, 1, true);
assert_zero(r);
}
// get iname
r = toku_db_get(env->i->directory, child, &old_dname_dbt, &iname_dbt, DB_SERIALIZABLE); // allocates memory for iname
r = toku_db_get(env->i->directory, txn, &old_dname_dbt, &iname_dbt, DB_SERIALIZABLE); // allocates memory for iname
char *iname = iname_dbt.data;
if (r == DB_NOTFOUND) {
r = ENOENT;
} else if (r == 0) {
// verify that newname does not already exist
r = db_getf_set(env->i->directory, child, DB_SERIALIZABLE, &new_dname_dbt, ydb_getf_do_nothing, NULL);
r = db_getf_set(env->i->directory, txn, DB_SERIALIZABLE, &new_dname_dbt, ydb_getf_do_nothing, NULL);
if (r == 0) {
r = EEXIST;
}
else if (r == DB_NOTFOUND) {
// remove old (dname,iname) and insert (newname,iname) in directory
r = toku_db_del(env->i->directory, child, &old_dname_dbt, DB_DELETE_ANY, TRUE);
r = toku_db_del(env->i->directory, txn, &old_dname_dbt, DB_DELETE_ANY, TRUE);
if (r != 0) { goto exit; }
r = toku_db_put(env->i->directory, child, &new_dname_dbt, &iname_dbt, 0, TRUE);
r = toku_db_put(env->i->directory, txn, &new_dname_dbt, &iname_dbt, 0, TRUE);
if (r != 0) { goto exit; }
//Now that we have writelocks on both dnames, verify that there are still no handles open. (to prevent race conditions)
......@@ -2839,7 +2837,7 @@ toku_env_dbrename(DB_ENV *env, DB_TXN *txn, const char *fname, const char *dbnam
// otherwise, we're okay in marking this ft as remove on
// commit. no new handles can open for this dictionary
// because the txn has directory write locks on the dname
if (!can_acquire_table_lock(env, child, iname)) {
if (txn && !can_acquire_table_lock(env, txn, iname)) {
r = DB_LOCK_NOTGRANTED;
}
// We don't do anything at the ft or cachetable layer for rename.
......@@ -2848,17 +2846,6 @@ toku_env_dbrename(DB_ENV *env, DB_TXN *txn, const char *fname, const char *dbnam
}
exit:
if (using_txns) {
// close txn
if (r == 0) { // commit
r = toku_txn_commit(child, DB_TXN_NOSYNC, NULL, NULL, false);
invariant(r==0); // TODO panic
}
else { // abort
int r2 = toku_txn_abort(child, NULL, NULL, false);
invariant(r2==0); // TODO panic
}
}
if (iname) {
toku_free(iname);
}
......@@ -2867,9 +2854,7 @@ exit:
int
DB_CREATE_FUN (DB ** db, DB_ENV * env, u_int32_t flags) {
toku_ydb_lock();
int r = toku_db_create(db, env, flags);
toku_ydb_unlock();
return r;
}
......
src/ydb_db.c
View file @ 8192256d
......@@ -143,7 +143,6 @@ toku_db_close(DB * db) {
return r;
}
///////////
//db_getf_XXX is equivalent to c_getf_XXX, without a persistent cursor
......@@ -229,18 +228,17 @@ db_open_subdb(DB * db, DB_TXN * txn, const char *fname, const char *dbname, DBTY
static int
toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *dbname, DBTYPE dbtype, u_int32_t flags, int mode) {
HANDLE_PANICKED_DB(db);
HANDLE_DB_ILLEGAL_WORKING_PARENT_TXN(db, txn);
if (dbname!=NULL)
if (dbname != NULL) {
return db_open_subdb(db, txn, fname, dbname, dbtype, flags, mode);
}
// at this point fname is the dname
//This code ONLY supports single-db files.
assert(dbname==NULL);
assert(dbname == NULL);
const char * dname = fname; // db_open_subdb() converts (fname, dbname) to dname
////////////////////////////// do some level of parameter checking.
u_int32_t unused_flags = flags;
int using_txns = db->dbenv->i->open_flags & DB_INIT_TXN;
int r;
if (dbtype!=DB_BTREE && dbtype!=DB_UNKNOWN) return EINVAL;
int is_db_excl = flags & DB_EXCL; unused_flags&=~DB_EXCL;
......@@ -252,7 +250,6 @@ toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *dbname, DBTYP
unused_flags&=~DB_READ_COMMITTED;
unused_flags&=~DB_SERIALIZABLE;
if (unused_flags & ~DB_THREAD) return EINVAL; // unknown flags
if (is_db_excl && !is_db_create) return EINVAL;
if (dbtype==DB_UNKNOWN && is_db_excl) return EINVAL;
......@@ -262,16 +259,11 @@ toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *dbname, DBTYP
if (r != 0)
return r;
if (db_opened(db))
return EINVAL; /* It was already open. */
//////////////////////////////
DB_TXN *child = NULL;
// begin child (unless transactionless)
if (using_txns) {
r = toku_txn_begin(db->dbenv, txn, &child, DB_TXN_NOSYNC, 1, true);
assert(r==0);
if (db_opened(db)) {
// it was already open
return EINVAL;
}
//////////////////////////////
// convert dname to iname
// - look up dname, get iname
......@@ -280,21 +272,20 @@ toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *dbname, DBTYP
DBT iname_dbt; // holds iname_in_env
toku_fill_dbt(&dname_dbt, dname, strlen(dname)+1);
init_dbt_realloc(&iname_dbt); // sets iname_dbt.data = NULL
r = toku_db_get(db->dbenv->i->directory, child, &dname_dbt, &iname_dbt, DB_SERIALIZABLE); // allocates memory for iname
r = toku_db_get(db->dbenv->i->directory, txn, &dname_dbt, &iname_dbt, DB_SERIALIZABLE); // allocates memory for iname
char *iname = iname_dbt.data;
if (r==DB_NOTFOUND && !is_db_create)
if (r == DB_NOTFOUND && !is_db_create) {
r = ENOENT;
else if (r==0 && is_db_excl) {
} else if (r==0 && is_db_excl) {
r = EEXIST;
}
else if (r==DB_NOTFOUND) {
} else if (r == DB_NOTFOUND) {
char hint[strlen(dname) + 1];
// create iname and make entry in directory
u_int64_t id = 0;
if (using_txns) {
id = toku_txn_get_txnid(db_txn_struct_i(child)->tokutxn);
if (txn) {
id = toku_txn_get_txnid(db_txn_struct_i(txn)->tokutxn);
}
create_iname_hint(dname, hint);
iname = create_iname(db->dbenv, id, hint, NULL, -1); // allocated memory for iname
......@@ -305,33 +296,21 @@ toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *dbname, DBTYP
// directory read lock is grabbed in toku_db_get above
//
u_int32_t put_flags = 0 | ((is_db_hot_index) ? DB_PRELOCKED_WRITE : 0);
r = toku_db_put(db->dbenv->i->directory, child, &dname_dbt, &iname_dbt, put_flags, TRUE);
r = toku_db_put(db->dbenv->i->directory, txn, &dname_dbt, &iname_dbt, put_flags, TRUE);
}
// we now have an iname
if (r == 0) {
r = db_open_iname(db, child, iname, flags, mode);
if (r==0) {
r = db_open_iname(db, txn, iname, flags, mode);
if (r == 0) {
db->i->dname = toku_xstrdup(dname);
env_note_db_opened(db->dbenv, db); // tell env that a new db handle is open (using dname)
}
}
// free string holding iname
if (iname) toku_free(iname);
if (using_txns) {
// close txn
if (r == 0) { // commit
r = toku_txn_commit(child, DB_TXN_NOSYNC, NULL, NULL, false);
invariant(r==0); // TODO panic
}
else { // abort
int r2 = toku_txn_abort(child, NULL, NULL, false);
invariant(r2==0); // TODO panic
if (iname) {
toku_free(iname);
}
}
return r;
}
......@@ -621,9 +600,10 @@ toku_db_pre_acquire_table_lock(DB *db, DB_TXN *txn) {
static int
locked_db_close(DB * db, u_int32_t UU(flags)) {
toku_ydb_lock();
// cannot begin a checkpoint
toku_multi_operation_client_lock();
int r = toku_db_close(db);
toku_ydb_unlock();
toku_multi_operation_client_unlock();
return r;
}
......@@ -649,11 +629,31 @@ autotxn_db_getf_set (DB *db, DB_TXN *txn, u_int32_t flags, DBT *key, YDB_CALLBAC
static int
locked_db_open(DB *db, DB_TXN *txn, const char *fname, const char *dbname, DBTYPE dbtype, u_int32_t flags, int mode) {
toku_multi_operation_client_lock(); //Cannot begin checkpoint
toku_ydb_lock();
int r = toku_db_open(db, txn, fname, dbname, dbtype, flags & ~DB_AUTO_COMMIT, mode);
toku_ydb_unlock();
toku_multi_operation_client_unlock(); //Can now begin checkpoint
int ret, r;
HANDLE_DB_ILLEGAL_WORKING_PARENT_TXN(db, txn);
DB_ENV *env = db->dbenv;
DB_TXN *child_txn = NULL;
int using_txns = env->i->open_flags & DB_INIT_TXN;
if (using_txns) {
ret = locked_txn_begin(env, txn, &child_txn, DB_TXN_NOSYNC);
invariant_zero(ret);
}
// cannot begin a checkpoint
toku_multi_operation_client_lock();
r = toku_db_open(db, child_txn, fname, dbname, dbtype, flags & ~DB_AUTO_COMMIT, mode);
toku_multi_operation_client_unlock();
if (using_txns) {
if (r == 0) {
ret = locked_txn_commit(child_txn, DB_TXN_NOSYNC);
invariant_zero(ret);
} else {
ret = locked_txn_abort(child_txn);
invariant_zero(ret);
}
}
return r;
}
......@@ -871,11 +871,7 @@ toku_db_create(DB ** db, DB_ENV * env, u_int32_t flags) {
return 0;
}
/* Following functions (ydb_load_xxx()) are used by loader:
*/
// When the loader is created, it makes this call.
// When the loader is created, it makes this call (toku_env_load_inames).
// For each dictionary to be loaded, replace old iname in directory
// with a newly generated iname. This will also take a write lock
// on the directory entries. The write lock will be released when
......@@ -887,33 +883,29 @@ toku_db_create(DB ** db, DB_ENV * env, u_int32_t flags) {
// If "mark_as_loader" is true, then include a mark in the iname
// to indicate that the file is created by the brt loader.
// Return 0 on success (could fail if write lock not available).
int
ydb_load_inames(DB_ENV * env, DB_TXN * txn, int N, DB * dbs[N], char * new_inames_in_env[N], LSN *load_lsn, BOOL mark_as_loader) {
int rval;
static int
load_inames(DB_ENV * env, DB_TXN * txn, int N, DB * dbs[N], char * new_inames_in_env[N], LSN *load_lsn, BOOL mark_as_loader) {
int rval = 0;
int i;
int using_txns = env->i->open_flags & DB_INIT_TXN;
DB_TXN * child = NULL;
TXNID xid = 0;
DBT dname_dbt; // holds dname
DBT iname_dbt; // holds new iname
char * mark;
if (mark_as_loader)
if (mark_as_loader) {
mark = "B";
else
} else {
mark = "P";
}
for (i=0; i<N; i++) {
new_inames_in_env[i] = NULL;
}
// begin child (unless transactionless)
if (using_txns) {
rval = toku_txn_begin(env, txn, &child, DB_TXN_NOSYNC, 1, true);
assert(rval == 0);
xid = toku_txn_get_txnid(db_txn_struct_i(child)->tokutxn);
if (txn) {
xid = toku_txn_get_txnid(db_txn_struct_i(txn)->tokutxn);
}
for (i = 0; i < N; i++) {
char * dname = dbs[i]->i->dname;
......@@ -924,12 +916,12 @@ ydb_load_inames(DB_ENV * env, DB_TXN * txn, int N, DB * dbs[N], char * new_iname
char * new_iname = create_iname(env, xid, hint, mark, i); // allocates memory for iname_in_env
new_inames_in_env[i] = new_iname;
toku_fill_dbt(&iname_dbt, new_iname, strlen(new_iname) + 1); // iname_in_env goes in directory
rval = toku_db_put(env->i->directory, child, &dname_dbt, &iname_dbt, 0, TRUE);
rval = toku_db_put(env->i->directory, txn, &dname_dbt, &iname_dbt, 0, TRUE);
if (rval) break;
}
// Generate load log entries.
if (!rval && using_txns) {
if (!rval && txn) {
TOKUTXN ttxn = db_txn_struct_i(txn)->tokutxn;
int do_fsync = 0;
LSN *get_lsn = NULL;
......@@ -944,17 +936,33 @@ ydb_load_inames(DB_ENV * env, DB_TXN * txn, int N, DB * dbs[N], char * new_iname
if (rval) break;
}
}
return rval;
}
int
locked_load_inames(DB_ENV * env, DB_TXN * txn, int N, DB * dbs[N], char * new_inames_in_env[N], LSN *load_lsn, BOOL mark_as_loader) {
int ret, r;
DB_TXN *child_txn = NULL;
int using_txns = env->i->open_flags & DB_INIT_TXN;
if (using_txns) {
// close txn
if (rval == 0) { // all well so far, commit child
rval = toku_txn_commit(child, DB_TXN_NOSYNC, NULL, NULL, false);
assert(rval==0);
}
else { // abort child
int r2 = toku_txn_abort(child, NULL, NULL, false);
assert(r2==0);
for (i=0; i<N; i++) {
ret = locked_txn_begin(env, txn, &child_txn, DB_TXN_NOSYNC);
invariant_zero(ret);
}
// cannot begin a checkpoint
toku_multi_operation_client_lock();
r = load_inames(env, child_txn, N, dbs, new_inames_in_env, load_lsn, mark_as_loader);
toku_multi_operation_client_unlock();
if (using_txns) {
if (r == 0) {
ret = locked_txn_commit(child_txn, DB_TXN_NOSYNC);
invariant_zero(ret);
} else {
ret = locked_txn_abort(child_txn);
invariant_zero(ret);
for (int i = 0; i < N; i++) {
if (new_inames_in_env[i]) {
toku_free(new_inames_in_env[i]);
new_inames_in_env[i] = NULL;
......@@ -962,10 +970,11 @@ ydb_load_inames(DB_ENV * env, DB_TXN * txn, int N, DB * dbs[N], char * new_iname
}
}
}
return r;
return rval;
}
#undef STATUS_VALUE
#include <valgrind/helgrind.h>
......
src/ydb_load.h
View file @ 8192256d
......@@ -8,8 +8,6 @@
/* ydb functions used by loader
*/
// When the loader is created, it makes this call.
// For each dictionary to be loaded, replace old iname in directory
// with a newly generated iname. This will also take a write lock
......@@ -22,13 +20,12 @@
// If "mark_as_loader" is true, then include a mark in the iname
// to indicate that the file is created by the brt loader.
// Return 0 on success (could fail if write lock not available).
int ydb_load_inames(DB_ENV * env,
int locked_load_inames(DB_ENV * env,
DB_TXN * txn,
int N,
DB * dbs[/*N*/],
/*out*/ char * new_inames_in_env[N],
DB * dbs[N],
char * new_inames_in_env[N], /* out */
LSN *load_lsn,
BOOL mark_as_loader);
#endif
src/ydb_txn.c
View file @ 8192256d
......@@ -13,6 +13,11 @@
#include <valgrind/helgrind.h>
#include "ft/txn_manager.h"
static int toku_txn_begin(DB_ENV *env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags);
static int toku_txn_commit(DB_TXN * txn, u_int32_t flags, TXN_PROGRESS_POLL_FUNCTION poll,
void *poll_extra, bool release_multi_operation_client_lock);
static int toku_txn_abort(DB_TXN * txn, TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra);
static int
toku_txn_release_locks(DB_TXN* txn) {
assert(txn);
......@@ -40,16 +45,6 @@ toku_txn_release_locks(DB_TXN* txn) {
return r;
}
// Yield the lock so someone else can work, and then reacquire the lock.
// Useful while processing commit or rollback logs, to allow others to access the system.
static void
ydb_yield (voidfp f, void *fv, void *UU(v)) {
toku_ydb_unlock();
if (f)
f(fv);
toku_ydb_lock();
}
static void
toku_txn_destroy(DB_TXN *txn) {
(void) __sync_fetch_and_sub(&txn->mgrp->i->open_txns, 1);
......@@ -64,9 +59,8 @@ toku_txn_destroy(DB_TXN *txn) {
static int
toku_txn_commit_only(DB_TXN * txn, u_int32_t flags,
TXN_PROGRESS_POLL_FUNCTION poll, void* poll_extra,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra,
bool release_multi_operation_client_lock) {
if (!txn) return EINVAL;
HANDLE_PANICKED_ENV(txn->mgrp);
//Recursively kill off children
if (db_txn_struct_i(txn)->child) {
......@@ -96,12 +90,10 @@ toku_txn_commit_only(DB_TXN * txn, u_int32_t flags,
int r;
if (flags!=0) {
// frees the tokutxn
// Calls ydb_yield(NULL) occasionally
r = toku_txn_abort_txn(db_txn_struct_i(txn)->tokutxn, ydb_yield, NULL, poll, poll_extra);
r = toku_txn_abort_txn(db_txn_struct_i(txn)->tokutxn, poll, poll_extra);
} else {
// frees the tokutxn
// Calls ydb_yield(NULL) occasionally
r = toku_txn_commit_txn(db_txn_struct_i(txn)->tokutxn, nosync, ydb_yield, NULL,
r = toku_txn_commit_txn(db_txn_struct_i(txn)->tokutxn, nosync,
poll, poll_extra);
}
if (r!=0 && !toku_env_is_panicked(txn->mgrp)) {
......@@ -139,8 +131,7 @@ toku_txn_commit_only(DB_TXN * txn, u_int32_t flags,
// modified by both this transaction and B, it'll read B's value, even
// though it cannot read this transaction's value, which comes below
// B's value on the leafentry's stack. This behavior is incorrect.
// All of this happens while the ydb lock is yielded. This causes a failure
// in the test_stress tests.
// This causes a failure in the test_stress tests.
//
toku_txn_get_fsync_info(ttxn, &do_fsync, &do_fsync_lsn);
toku_txn_complete_txn(ttxn);
......@@ -154,7 +145,7 @@ toku_txn_commit_only(DB_TXN * txn, u_int32_t flags,
if (release_multi_operation_client_lock) {
toku_multi_operation_client_unlock();
}
toku_txn_maybe_fsync_log(logger, do_fsync_lsn, do_fsync, ydb_yield, NULL);
toku_txn_maybe_fsync_log(logger, do_fsync_lsn, do_fsync);
//Promote list to parent (dbs that must close before abort)
if (txn->parent) {
......@@ -175,9 +166,9 @@ toku_txn_commit_only(DB_TXN * txn, u_int32_t flags,
return r;
}
int
static int
toku_txn_commit(DB_TXN * txn, u_int32_t flags,
TXN_PROGRESS_POLL_FUNCTION poll, void* poll_extra,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra,
bool release_multi_operation_client_lock) {
int r = toku_txn_commit_only(txn, flags, poll, poll_extra, release_multi_operation_client_lock);
toku_txn_destroy(txn);
......@@ -194,8 +185,7 @@ toku_txn_id(DB_TXN * txn) {
static int
toku_txn_abort_only(DB_TXN * txn,
TXN_PROGRESS_POLL_FUNCTION poll, void* poll_extra,
bool release_multi_operation_client_lock) {
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra) {
HANDLE_PANICKED_ENV(txn->mgrp);
//Recursively kill off children (abort or commit are both correct, commit is cheaper)
if (db_txn_struct_i(txn)->child) {
......@@ -217,7 +207,7 @@ toku_txn_abort_only(DB_TXN * txn,
//All dbs that must close before abort, must now be closed
assert(toku_list_empty(&db_txn_struct_i(txn)->dbs_that_must_close_before_abort));
int r = toku_txn_abort_txn(db_txn_struct_i(txn)->tokutxn, ydb_yield, NULL, poll, poll_extra);
int r = toku_txn_abort_txn(db_txn_struct_i(txn)->tokutxn, poll, poll_extra);
if (r!=0 && !toku_env_is_panicked(txn->mgrp)) {
env_panic(txn->mgrp, r, "Error during abort.\n");
}
......@@ -225,13 +215,6 @@ toku_txn_abort_only(DB_TXN * txn,
assert_zero(r);
r = toku_txn_release_locks(txn);
toku_txn_complete_txn(db_txn_struct_i(txn)->tokutxn);
// this lock must be released after toku_txn_complete_txn because
// this lock must be held until the references to the open FTs is released
// begin checkpoint logs these associations, so we must be protect
// the changing of these associations with checkpointing
if (release_multi_operation_client_lock) {
toku_multi_operation_client_unlock();
}
return r;
}
......@@ -257,7 +240,7 @@ toku_txn_xa_prepare (DB_TXN *txn, TOKU_XA_XID *xid) {
LSN do_fsync_lsn;
bool do_fsync;
toku_txn_get_fsync_info(ttxn, &do_fsync, &do_fsync_lsn);
toku_txn_maybe_fsync_log(logger, do_fsync_lsn, do_fsync, ydb_yield, NULL);
toku_txn_maybe_fsync_log(logger, do_fsync_lsn, do_fsync);
return r;
}
......@@ -272,28 +255,26 @@ toku_txn_prepare (DB_TXN *txn, u_int8_t gid[DB_GID_SIZE]) {
return toku_txn_xa_prepare(txn, &xid);
}
int
static int
toku_txn_abort(DB_TXN * txn,
TXN_PROGRESS_POLL_FUNCTION poll, void* poll_extra,
bool release_multi_operation_client_lock) {
int r = toku_txn_abort_only(txn, poll, poll_extra, release_multi_operation_client_lock);
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra) {
int r = toku_txn_abort_only(txn, poll, poll_extra);
toku_txn_destroy(txn);
return r;
}
// Create a new transaction.
// Called without holding the ydb lock.
int
locked_txn_begin(DB_ENV *env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags) {
int r = toku_txn_begin(env, stxn, txn, flags, 0, false);
toku_multi_operation_client_lock();
int r = toku_txn_begin(env, stxn, txn, flags);
toku_multi_operation_client_unlock();
return r;
}
static u_int32_t
locked_txn_id(DB_TXN *txn) {
toku_ydb_lock();
u_int32_t r = toku_txn_id(txn);
toku_ydb_unlock();
return r;
}
......@@ -305,9 +286,7 @@ toku_txn_txn_stat (DB_TXN *txn, struct txn_stat **txn_stat) {
static int
locked_txn_txn_stat (DB_TXN *txn, struct txn_stat **txn_stat) {
toku_ydb_lock();
int r = toku_txn_txn_stat(txn, txn_stat);
toku_ydb_unlock();
return r;
}
......@@ -318,22 +297,21 @@ locked_txn_commit_with_progress(DB_TXN *txn, u_int32_t flags,
if (toku_txn_requires_checkpoint(ttxn)) {
toku_checkpoint(txn->mgrp->i->cachetable, txn->mgrp->i->logger, NULL, NULL, NULL, NULL, TXN_COMMIT_CHECKPOINT);
}
toku_multi_operation_client_lock(); //Cannot checkpoint during a commit.
toku_ydb_lock();
int r = toku_txn_commit_only(txn, flags, poll, poll_extra, true); // the final 'true' says to release the multi_operation_client_lock
toku_ydb_unlock();
toku_txn_destroy(txn);
// cannot begin a checkpoint. the multi operation lock is taken here,
// but released in toku_txn_commit_only. this way, we don't hold it
// while we fsync the log.
toku_multi_operation_client_lock();
int r = toku_txn_commit(txn, flags, poll, poll_extra, true);
return r;
}
static int
locked_txn_abort_with_progress(DB_TXN *txn,
TXN_PROGRESS_POLL_FUNCTION poll, void* poll_extra) {
toku_multi_operation_client_lock(); //Cannot checkpoint during an abort.
toku_ydb_lock();
int r = toku_txn_abort_only(txn, poll, poll_extra, true); // the final 'true' says to release the multi_operation_client_lock
toku_ydb_unlock();
toku_txn_destroy(txn);
// cannot begin a checkpoint
toku_multi_operation_client_lock();
int r = toku_txn_abort(txn, poll, poll_extra);
toku_multi_operation_client_unlock();
return r;
}
......@@ -351,16 +329,22 @@ locked_txn_abort(DB_TXN *txn) {
static int
locked_txn_prepare (DB_TXN *txn, u_int8_t gid[DB_GID_SIZE]) {
toku_ydb_lock(); int r = toku_txn_prepare (txn, gid); toku_ydb_unlock(); return r;
toku_multi_operation_client_lock();
int r = toku_txn_prepare(txn, gid);
toku_multi_operation_client_unlock();
return r;
}
static int
locked_txn_xa_prepare (DB_TXN *txn, TOKU_XA_XID *xid) {
toku_ydb_lock(); int r = toku_txn_xa_prepare (txn, xid); toku_ydb_unlock(); return r;
toku_multi_operation_client_lock();
int r = toku_txn_xa_prepare(txn, xid);
toku_multi_operation_client_unlock();
return r;
}
int
toku_txn_begin(DB_ENV *env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags, bool internal, bool holds_ydb_lock) {
static int
toku_txn_begin(DB_ENV *env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags) {
HANDLE_PANICKED_ENV(env);
HANDLE_ILLEGAL_WORKING_PARENT_TXN(env, stxn); //Cannot create child while child already exists.
if (!toku_logger_is_open(env->i->logger))
......@@ -388,7 +372,7 @@ toku_txn_begin(DB_ENV *env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags, bool
}
flags &= ~iso_flags;
if (internal && stxn) {
if (stxn) {
child_isolation = db_txn_struct_i(stxn)->iso;
}
else {
......@@ -504,18 +488,14 @@ toku_txn_begin(DB_ENV *env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags, bool
snapshot_type,
result
);
if (r != 0)
if (r != 0) {
toku_free(result);
return r;
if (!holds_ydb_lock) {
toku_ydb_lock();
}
toku_txn_manager_start_txn(
toku_logger_get_txn_manager(env->i->logger),
db_txn_struct_i(result)->tokutxn
);
if (!holds_ydb_lock) {
toku_ydb_unlock();
}
//Add to the list of children for the parent.
if (result->parent) {
......
src/ydb_txn.h
View file @ 8192256d
......@@ -9,13 +9,17 @@
extern "C" {
#endif
// begin, commit, and abort use the multi operation lock
// internally to synchronize with begin checkpoint. callers
// should not hold the multi operation lock.
int locked_txn_begin(DB_ENV *env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags);
int toku_txn_begin(DB_ENV *env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags, bool internal, bool holds_ydb_lock);
int toku_txn_commit(DB_TXN * txn, u_int32_t flags, TXN_PROGRESS_POLL_FUNCTION, void*, bool release_multi_operation_client_lock);
int toku_txn_abort(DB_TXN * txn, TXN_PROGRESS_POLL_FUNCTION, void*, bool release_multi_operation_client_lock);
int locked_txn_commit(DB_TXN *txn, u_int32_t flags);
int locked_txn_abort(DB_TXN *txn);
void toku_keep_prepared_txn_callback (DB_ENV *env, TOKUTXN tokutxn);
void toku_keep_prepared_txn_callback(DB_ENV *env, TOKUTXN tokutxn);
#if defined(__cplusplus)
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7