Skip to content

M
mariadb
Commit 14d67581 authored by Yoni Fogel's avatar Yoni Fogel
Browse files
Options

closes[t:2391] Merge 2391 branch to main 

git-svn-id: file:///svn/toku/tokudb@18153 c7de825b-a66e-492c-adef-691d508d4ae1
parent 0bdd64ae
Show whitespace changes
Inline Side-by-side
Showing with 565 additions and 50 deletions
buildheader/db.h_4_1
View file @ 14d67581
......@@ -346,6 +346,13 @@ struct __toku_db_txn_active {
char __toku_dummy0[4];
DB_LSN lsn; /* 32-bit offset=8 size=8, 64=bit offset=8 size=8 */
};
typedef struct __toku_txn_progress {
uint64_t entries_total;
uint64_t entries_processed;
uint8_t is_commit;
uint8_t stalled_on_checkpoint;
} *TOKU_TXN_PROGRESS, TOKU_TXN_PROGRESS_S;
typedef void(*TXN_PROGRESS_POLL_FUNCTION)(TOKU_TXN_PROGRESS, void*);
struct txn_stat {
u_int64_t rolltmp_raw_count;
};
......@@ -354,7 +361,9 @@ struct __toku_db_txn {
DB_TXN *parent; /* 32-bit offset=4 size=4, 64=bit offset=8 size=8 */
int (*txn_stat)(DB_TXN *, struct txn_stat **);
struct { void *next, *prev; } open_txns;
void* __toku_dummy0[7];
int (*commit_with_progress)(DB_TXN*, uint32_t, TXN_PROGRESS_POLL_FUNCTION, void*);
int (*abort_with_progress)(DB_TXN*, TXN_PROGRESS_POLL_FUNCTION, void*);
void* __toku_dummy0[5];
char __toku_dummy1[24];
void *api_internal; /* 32-bit offset=68 size=4, 64=bit offset=112 size=8 */
void* __toku_dummy2[1];
......
buildheader/db.h_4_3
View file @ 14d67581
......@@ -362,6 +362,13 @@ struct __toku_db_txn_active {
DB_LSN lsn; /* 32-bit offset=8 size=8, 64=bit offset=8 size=8 */
char __toku_dummy1[132]; /* Padding at the end */
};
typedef struct __toku_txn_progress {
uint64_t entries_total;
uint64_t entries_processed;
uint8_t is_commit;
uint8_t stalled_on_checkpoint;
} *TOKU_TXN_PROGRESS, TOKU_TXN_PROGRESS_S;
typedef void(*TXN_PROGRESS_POLL_FUNCTION)(TOKU_TXN_PROGRESS, void*);
struct txn_stat {
u_int64_t rolltmp_raw_count;
};
......@@ -370,7 +377,9 @@ struct __toku_db_txn {
DB_TXN *parent; /* 32-bit offset=4 size=4, 64=bit offset=8 size=8 */
int (*txn_stat)(DB_TXN *, struct txn_stat **);
struct { void *next, *prev; } open_txns;
void* __toku_dummy0[12];
int (*commit_with_progress)(DB_TXN*, uint32_t, TXN_PROGRESS_POLL_FUNCTION, void*);
int (*abort_with_progress)(DB_TXN*, TXN_PROGRESS_POLL_FUNCTION, void*);
void* __toku_dummy0[10];
char __toku_dummy1[24];
void *api_internal; /* 32-bit offset=88 size=4, 64=bit offset=152 size=8 */
void* __toku_dummy2[2];
......
buildheader/db.h_4_4
View file @ 14d67581
......@@ -369,6 +369,13 @@ struct __toku_db_txn_active {
DB_LSN lsn; /* 32-bit offset=16 size=8, 64=bit offset=24 size=8 */
char __toku_dummy2[184]; /* Padding at the end */
};
typedef struct __toku_txn_progress {
uint64_t entries_total;
uint64_t entries_processed;
uint8_t is_commit;
uint8_t stalled_on_checkpoint;
} *TOKU_TXN_PROGRESS, TOKU_TXN_PROGRESS_S;
typedef void(*TXN_PROGRESS_POLL_FUNCTION)(TOKU_TXN_PROGRESS, void*);
struct txn_stat {
u_int64_t rolltmp_raw_count;
};
......@@ -377,7 +384,9 @@ struct __toku_db_txn {
DB_TXN *parent; /* 32-bit offset=4 size=4, 64=bit offset=8 size=8 */
int (*txn_stat)(DB_TXN *, struct txn_stat **);
struct { void *next, *prev; } open_txns;
void* __toku_dummy0[15];
int (*commit_with_progress)(DB_TXN*, uint32_t, TXN_PROGRESS_POLL_FUNCTION, void*);
int (*abort_with_progress)(DB_TXN*, TXN_PROGRESS_POLL_FUNCTION, void*);
void* __toku_dummy0[13];
char __toku_dummy1[8];
void *api_internal; /* 32-bit offset=84 size=4, 64=bit offset=160 size=8 */
void* __toku_dummy2[2];
......
buildheader/db.h_4_5
View file @ 14d67581
......@@ -369,6 +369,13 @@ struct __toku_db_txn_active {
DB_LSN lsn; /* 32-bit offset=16 size=8, 64=bit offset=24 size=8 */
char __toku_dummy2[200]; /* Padding at the end */
};
typedef struct __toku_txn_progress {
uint64_t entries_total;
uint64_t entries_processed;
uint8_t is_commit;
uint8_t stalled_on_checkpoint;
} *TOKU_TXN_PROGRESS, TOKU_TXN_PROGRESS_S;
typedef void(*TXN_PROGRESS_POLL_FUNCTION)(TOKU_TXN_PROGRESS, void*);
struct txn_stat {
u_int64_t rolltmp_raw_count;
};
......@@ -377,7 +384,9 @@ struct __toku_db_txn {
DB_TXN *parent; /* 32-bit offset=4 size=4, 64=bit offset=8 size=8 */
int (*txn_stat)(DB_TXN *, struct txn_stat **);
struct { void *next, *prev; } open_txns;
void* __toku_dummy0[15];
int (*commit_with_progress)(DB_TXN*, uint32_t, TXN_PROGRESS_POLL_FUNCTION, void*);
int (*abort_with_progress)(DB_TXN*, TXN_PROGRESS_POLL_FUNCTION, void*);
void* __toku_dummy0[13];
char __toku_dummy1[8];
void *api_internal; /* 32-bit offset=84 size=4, 64=bit offset=160 size=8 */
void* __toku_dummy2[2];
......
buildheader/db.h_4_6
View file @ 14d67581
......@@ -373,6 +373,13 @@ struct __toku_db_txn_active {
DB_LSN lsn; /* 32-bit offset=16 size=8, 64=bit offset=24 size=8 */
char __toku_dummy2[200]; /* Padding at the end */
};
typedef struct __toku_txn_progress {
uint64_t entries_total;
uint64_t entries_processed;
uint8_t is_commit;
uint8_t stalled_on_checkpoint;
} *TOKU_TXN_PROGRESS, TOKU_TXN_PROGRESS_S;
typedef void(*TXN_PROGRESS_POLL_FUNCTION)(TOKU_TXN_PROGRESS, void*);
struct txn_stat {
u_int64_t rolltmp_raw_count;
};
......@@ -381,7 +388,9 @@ struct __toku_db_txn {
DB_TXN *parent; /* 32-bit offset=4 size=4, 64=bit offset=8 size=8 */
int (*txn_stat)(DB_TXN *, struct txn_stat **);
struct { void *next, *prev; } open_txns;
void* __toku_dummy0[16];
int (*commit_with_progress)(DB_TXN*, uint32_t, TXN_PROGRESS_POLL_FUNCTION, void*);
int (*abort_with_progress)(DB_TXN*, TXN_PROGRESS_POLL_FUNCTION, void*);
void* __toku_dummy0[14];
char __toku_dummy1[8];
void *api_internal; /* 32-bit offset=88 size=4, 64=bit offset=168 size=8 */
void* __toku_dummy2[2];
......
buildheader/make_db_h.c
View file @ 14d67581
......@@ -458,7 +458,6 @@ int main (int argc __attribute__((__unused__)), char *argv[] __attribute__((__un
printf(" u_int64_t fsync_time; /* total time required to fsync */ \n");
printf("} ENGINE_STATUS;\n");
print_dbtype();
// print_db_notices();
print_defines();
......@@ -544,10 +543,21 @@ int main (int argc __attribute__((__unused__)), char *argv[] __attribute__((__un
print_struct("db_txn_active", 0, db_txn_active_fields32, db_txn_active_fields64, sizeof(db_txn_active_fields32)/sizeof(db_txn_active_fields32[0]), 0);
assert(sizeof(db_txn_fields32)==sizeof(db_txn_fields64));
{
//txn progress info
printf("typedef struct __toku_txn_progress {\n");
printf(" uint64_t entries_total;\n");
printf(" uint64_t entries_processed;\n");
printf(" uint8_t is_commit;\n");
printf(" uint8_t stalled_on_checkpoint;\n");
printf("} *TOKU_TXN_PROGRESS, TOKU_TXN_PROGRESS_S;\n");
printf("typedef void(*TXN_PROGRESS_POLL_FUNCTION)(TOKU_TXN_PROGRESS, void*);\n");
printf("struct txn_stat {\n u_int64_t rolltmp_raw_count;\n};\n");
const char *extra[] = {
"int (*txn_stat)(DB_TXN *, struct txn_stat **)",
"struct { void *next, *prev; } open_txns",
"int (*commit_with_progress)(DB_TXN*, uint32_t, TXN_PROGRESS_POLL_FUNCTION, void*)",
"int (*abort_with_progress)(DB_TXN*, TXN_PROGRESS_POLL_FUNCTION, void*)",
NULL,
};
print_struct("db_txn", INTERNAL_AT_END, db_txn_fields32, db_txn_fields64, sizeof(db_txn_fields32)/sizeof(db_txn_fields32[0]), extra);
......
buildheader/tdb.h
View file @ 14d67581
......@@ -320,6 +320,13 @@ struct __toku_db_txn_active {
u_int32_t txnid;
DB_LSN lsn;
};
typedef struct __toku_txn_progress {
uint64_t entries_total;
uint64_t entries_processed;
uint8_t is_commit;
uint8_t stalled_on_checkpoint;
} *TOKU_TXN_PROGRESS, TOKU_TXN_PROGRESS_S;
typedef void(*TXN_PROGRESS_POLL_FUNCTION)(TOKU_TXN_PROGRESS, void*);
struct txn_stat {
u_int64_t rolltmp_raw_count;
};
......@@ -328,6 +335,8 @@ struct __toku_db_txn {
DB_TXN *parent;
int (*txn_stat)(DB_TXN *, struct txn_stat **);
struct { void *next, *prev; } open_txns;
int (*commit_with_progress)(DB_TXN*, uint32_t, TXN_PROGRESS_POLL_FUNCTION, void*);
int (*abort_with_progress)(DB_TXN*, TXN_PROGRESS_POLL_FUNCTION, void*);
void *api_internal;
int (*abort) (DB_TXN *);
int (*commit) (DB_TXN*, u_int32_t);
......
include/db.h
View file @ 14d67581
......@@ -320,6 +320,13 @@ struct __toku_db_txn_active {
u_int32_t txnid;
DB_LSN lsn;
};
typedef struct __toku_txn_progress {
uint64_t entries_total;
uint64_t entries_processed;
uint8_t is_commit;
uint8_t stalled_on_checkpoint;
} *TOKU_TXN_PROGRESS, TOKU_TXN_PROGRESS_S;
typedef void(*TXN_PROGRESS_POLL_FUNCTION)(TOKU_TXN_PROGRESS, void*);
struct txn_stat {
u_int64_t rolltmp_raw_count;
};
......@@ -328,6 +335,8 @@ struct __toku_db_txn {
DB_TXN *parent;
int (*txn_stat)(DB_TXN *, struct txn_stat **);
struct { void *next, *prev; } open_txns;
int (*commit_with_progress)(DB_TXN*, uint32_t, TXN_PROGRESS_POLL_FUNCTION, void*);
int (*abort_with_progress)(DB_TXN*, TXN_PROGRESS_POLL_FUNCTION, void*);
void *api_internal;
int (*abort) (DB_TXN *);
int (*commit) (DB_TXN*, u_int32_t);
......
newbrt/log-internal.h
View file @ 14d67581
......@@ -120,6 +120,10 @@ struct tokutxn {
XIDS xids; //Represents the xid list
BOOL force_fsync_on_commit; //This transaction NEEDS an fsync once (if) it commits. (commit means root txn)
BOOL has_done_work; //If this transaction has not done work, there is no need to fsync.
TXN_PROGRESS_POLL_FUNCTION progress_poll_fun;
void * progress_poll_fun_extra;
uint64_t num_rollentries;
uint64_t num_rollentries_processed;
};
static inline int toku_logsizeof_u_int8_t (u_int32_t v __attribute__((__unused__))) {
......
newbrt/logformat.c
View file @ 14d67581
......@@ -548,6 +548,7 @@ generate_rollbacks (void) {
fprintf(cf, " txn->newest_logentry = v;\n");
fprintf(cf, " txn->rollentry_resident_bytecount += rollback_fsize;\n");
fprintf(cf, " txn->rollentry_raw_count += rollback_fsize;\n");
fprintf(cf, " txn->num_rollentries++;\n");
fprintf(cf, " return toku_maybe_spill_rollbacks(txn);\n}\n");
});
......
newbrt/recover.c
View file @ 14d67581
......@@ -930,7 +930,7 @@ static int toku_recover_commit (struct logtype_commit *l, RECOVER_ENV renv) {
}
// commit the transaction
r = toku_txn_commit_with_lsn(txn, TRUE, recover_yield, NULL, l->lsn);
r = toku_txn_commit_with_lsn(txn, TRUE, recover_yield, NULL, l->lsn, NULL, NULL);
assert(r == 0);
// close the transaction
......@@ -958,7 +958,7 @@ static int toku_recover_xabort (struct logtype_xabort *l, RECOVER_ENV renv) {
}
// abort the transaction
r = toku_txn_abort_with_lsn(txn, recover_yield, NULL, l->lsn);
r = toku_txn_abort_with_lsn(txn, recover_yield, NULL, l->lsn, NULL, NULL);
assert(r == 0);
// close the transaction
......@@ -1089,7 +1089,7 @@ static void recover_abort_live_txns(RECOVER_ENV renv) {
TOKUTXN txn = (TOKUTXN) v;
// abort the transaction
r = toku_txn_abort_txn(txn, recover_yield, NULL);
r = toku_txn_abort_txn(txn, recover_yield, NULL, NULL, NULL);
assert(r == 0);
// close the transaction
......
newbrt/roll.c
View file @ 14d67581
......@@ -345,7 +345,9 @@ toku_rollback_tablelock_on_empty_table (FILENUM filenum,
LSN UU(oplsn))
{
//TODO: Replace truncate function with something that doesn't need to mess with checkpoints.
toku_poll_txn_progress_function(txn, FALSE, TRUE);
yield(toku_checkpoint_safe_client_lock, yield_v);
toku_poll_txn_progress_function(txn, FALSE, FALSE);
// on rollback we have to make the file be empty, since we locked an empty table, and then may have done things to it.
CACHEFILE cf;
......
newbrt/rollback.c
View file @ 14d67581
......@@ -7,17 +7,34 @@
static void note_txn_closing (TOKUTXN txn);
void
toku_poll_txn_progress_function(TOKUTXN txn, uint8_t is_commit, uint8_t stall_for_checkpoint) {
if (txn->progress_poll_fun) {
TOKU_TXN_PROGRESS_S progress = {
.entries_total = txn->num_rollentries,
.entries_processed = txn->num_rollentries_processed,
.is_commit = is_commit,
.stalled_on_checkpoint = stall_for_checkpoint};
txn->progress_poll_fun(&progress, txn->progress_poll_fun_extra);
}
}
int toku_commit_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv, LSN lsn) {
int r=0;
rolltype_dispatch_assign(item, toku_commit_, r, txn, yield, yieldv, lsn);
txn->num_rollentries_processed++;
if (txn->num_rollentries_processed % 1024 == 0)
toku_poll_txn_progress_function(txn, TRUE, FALSE);
return r;
}
int toku_abort_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv, LSN lsn) {
int r=0;
rolltype_dispatch_assign(item, toku_rollback_, r, txn, yield, yieldv, lsn);
if (r!=0) return r;
return 0;
txn->num_rollentries_processed++;
if (txn->num_rollentries_processed % 1024 == 0)
toku_poll_txn_progress_function(txn, FALSE, FALSE);
return r;
}
void toku_rollback_txn_close (TOKUTXN txn) {
......@@ -156,6 +173,7 @@ int toku_rollback_commit(TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn) {
//save that in the parent. Since the commit really happens in the root txn.
txn->parent->force_fsync_on_commit |= txn->force_fsync_on_commit;
txn->parent->has_done_work |= txn->has_done_work;
txn->parent->num_rollentries += txn->num_rollentries;
} else {
// do the commit calls and free everything
// we do the commit calls in reverse order too.
......
newbrt/rollback.h
View file @ 14d67581
......@@ -9,6 +9,7 @@
// these routines in rollback.c
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);
void toku_rollback_txn_close (TOKUTXN txn);
......
newbrt/txn.c
View file @ 14d67581
......@@ -36,6 +36,10 @@ int toku_txn_begin_with_xid (TOKUTXN parent_tokutxn, TOKUTXN *tokutxn, TOKULOGGE
result->oldest_logentry = result->newest_logentry = 0;
result->rollentry_arena = memarena_create();
result->num_rollentries = 0;
result->num_rollentries_processed = 0;
result->progress_poll_fun = NULL;
result->progress_poll_fun_extra = NULL;
if (toku_omt_size(logger->live_txns) == 0) {
assert(logger->oldest_living_xid == TXNID_NONE_LIVING);
......@@ -73,17 +77,22 @@ died:
// Doesn't close the txn, just performs the commit operations.
int toku_txn_commit_txn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv) {
return toku_txn_commit_with_lsn(txn, nosync, yield, yieldv, ZERO_LSN);
int toku_txn_commit_txn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra) {
return toku_txn_commit_with_lsn(txn, nosync, yield, yieldv, ZERO_LSN, poll, poll_extra);
}
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, YIELDF yield, void *yieldv, LSN oplsn,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra) {
int r;
// panic handled in log_commit
//Child transactions do not actually 'commit'. They promote their changes to parent, so no need to fsync if this txn has a parent.
int do_fsync = !txn->parent && (txn->force_fsync_on_commit || (!nosync && txn->has_done_work));
txn->progress_poll_fun = poll;
txn->progress_poll_fun_extra = poll_extra;
r = toku_log_commit(txn->logger, (LSN*)0, do_fsync, txn->txnid64); // exits holding neither of the tokulogger locks.
if (r!=0)
return r;
......@@ -92,15 +101,19 @@ int toku_txn_commit_with_lsn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv
}
// Doesn't close the txn, just performs the abort operations.
int toku_txn_abort_txn(TOKUTXN txn, YIELDF yield, void *yieldv) {
return toku_txn_abort_with_lsn(txn, yield, yieldv, ZERO_LSN);
int toku_txn_abort_txn(TOKUTXN txn, YIELDF yield, void *yieldv,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra) {
return toku_txn_abort_with_lsn(txn, yield, yieldv, 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, YIELDF yield, void *yieldv, LSN oplsn,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra) {
//printf("%s:%d aborting\n", __FILE__, __LINE__);
// Must undo everything. Must undo it all in reverse order.
// Build the reverse list
//printf("%s:%d abort\n", __FILE__, __LINE__);
txn->progress_poll_fun = poll;
txn->progress_poll_fun_extra = poll_extra;
int r=0;
r = toku_log_xabort(txn->logger, (LSN*)0, 0, txn->txnid64);
if (r!=0)
......
newbrt/txn.h
View file @ 14d67581
......@@ -8,11 +8,15 @@
int toku_txn_begin_txn (TOKUTXN parent_tokutxn, TOKUTXN *tokutxn, TOKULOGGER logger);
int toku_txn_begin_with_xid (TOKUTXN parent_tokutxn, TOKUTXN *tokutxn, TOKULOGGER logger, TXNID xid);
int toku_txn_commit_txn (TOKUTXN txn, int nosync, YIELDF yield, void *yieldv);
int toku_txn_commit_with_lsn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv, LSN oplsn);
int toku_txn_abort_txn(TOKUTXN txn, YIELDF yield, void *yieldv);
int toku_txn_abort_with_lsn(TOKUTXN txn, YIELDF yield, void *yieldv, LSN oplsn);
int toku_txn_commit_txn (TOKUTXN txn, int nosync, YIELDF yield, void *yieldv,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra);
int toku_txn_commit_with_lsn(TOKUTXN txn, int nosync, YIELDF yield, void *yieldv, LSN oplsn,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra);
int toku_txn_abort_txn(TOKUTXN txn, YIELDF yield, void *yieldv,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra);
int toku_txn_abort_with_lsn(TOKUTXN txn, YIELDF yield, void *yieldv, LSN oplsn,
TXN_PROGRESS_POLL_FUNCTION poll, void *poll_extra);
void toku_txn_close_txn(TOKUTXN txn);
XIDS toku_txn_get_xids (TOKUTXN);
......
src/tests/progress.c 0 → 100644
View file @ 14d67581
/* -*- mode: C; c-basic-offset: 4 -*- */
#ident "Copyright (c) 2010 Tokutek Inc. All rights reserved."
#include "test.h"
/*
- ydb layer test of progress report on commit, abort.
- test1:
create two txns
perform operations (inserts and deletes)
commit or abort inner txn
if abort, verify progress callback was called with correct args
if commit, verify progress callback was not called
commit or abort outer txn
verify progress callback was called with correct args
Note: inner loop ends with commit, so when outer loop completes,
it should be called for all operations performed by inner loop.
perform_ops {
for i = 0 -> 5 {
for j = 0 -> 1023
if (j & 0x20) insert
else delete
}
verify (n) {
verify that callback was called n times with correct args
}
test1:
for c0 = 0, 1 {
for c1 = 0, 1 {
begin txn0
perform_ops (txn0)
begin txn1
perform ops (tnx1)
if c1
abort txn1
verify (n)
else
commit txn1
verify (0)
}
if c0
abort txn0
verify (2n)
else
commit txn0
verify (2n)
}
- test2
- create empty dictionary
- begin txn
- lock empty dictionary (full range lock)
- abort
- verify that callback was called twice, first with stalled-on-checkpoint true, then with stalled-on-checkpoint false
*/
#define DICT_0 "dict_0.db"
static DB_ENV *env = NULL;
static DB_TXN *txn_parent = NULL;
static DB_TXN *txn_child = NULL;
static DB *db;
static char *dname = DICT_0;
static DBT key;
static DBT val;
static void start_txn(void);
static void commit_txn(int);
static void open_db(void);
static void close_db(void);
static void insert(void);
static void delete(void);
static void
start_env(void) {
assert(env==NULL);
system("rm -rf " ENVDIR);
toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO);
int r;
r = db_env_create(&env, 0);
CKERR(r);
r = env->open(env, ENVDIR, DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_MPOOL|DB_INIT_TXN|DB_CREATE|DB_PRIVATE, S_IRWXU+S_IRWXG+S_IRWXO);
CKERR(r);
dname = DICT_0;
dbt_init(&key, "key", strlen("key")+1);
dbt_init(&val, "val", strlen("val")+1);
open_db();
close_db();
}
static void
end_env(void) {
int r;
r=env->close(env, 0);
CKERR(r);
env = NULL;
}
static void
start_txn(void) {
assert(env!=NULL);
int r;
if (!txn_parent) {
r=env->txn_begin(env, 0, &txn_parent, 0);
}
else {
assert(!txn_child);
r=env->txn_begin(env, txn_parent, &txn_child, 0);
}
CKERR(r);
}
struct progress_expect {
int num_calls;
uint8_t is_commit_expected;
uint8_t stalled_on_checkpoint_expected;
uint64_t min_entries_total_expected;
uint64_t last_entries_processed;
};
static void poll(TOKU_TXN_PROGRESS progress, void *extra) {
struct progress_expect *info = extra;
info->num_calls++;
assert(progress->is_commit == info->is_commit_expected);
assert(progress->stalled_on_checkpoint == info->stalled_on_checkpoint_expected);
assert(progress->entries_total >= info->min_entries_total_expected);
assert(progress->entries_processed == 1024 + info->last_entries_processed);
info->last_entries_processed = progress->entries_processed;
}
//expect_number_polls is number of times polling function should be called.
static void
abort_txn(int expect_number_polls) {
assert(env!=NULL);
DB_TXN *txn;
BOOL child;
if (txn_child) {
txn = txn_child;
child = TRUE;
}
else {
txn = txn_parent;
child = FALSE;
}
assert(txn);
struct progress_expect extra = {
.num_calls = 0,
.is_commit_expected = 0,
.stalled_on_checkpoint_expected = 0,
.min_entries_total_expected = expect_number_polls * 1024,
.last_entries_processed = 0
};
int r;
r=txn->abort_with_progress(txn, poll, &extra);
CKERR(r);
assert(extra.num_calls == expect_number_polls);
if (child)
txn_child = NULL;
else
txn_parent = NULL;
}
static void
commit_txn(int expect_number_polls) {
assert(env!=NULL);
DB_TXN *txn;
BOOL child;
if (txn_child) {
txn = txn_child;
child = TRUE;
}
else {
txn = txn_parent;
child = FALSE;
}
assert(txn);
if (child)
assert(expect_number_polls == 0);
struct progress_expect extra = {
.num_calls = 0,
.is_commit_expected = 1,
.stalled_on_checkpoint_expected = 0,
.min_entries_total_expected = expect_number_polls * 1024,
.last_entries_processed = 0
};
int r;
r=txn->commit_with_progress(txn, 0, poll, &extra);
CKERR(r);
assert(extra.num_calls == expect_number_polls);
if (child)
txn_child = NULL;
else
txn_parent = NULL;
}
static void
open_db(void) {
assert(env!=NULL);
assert(db == NULL);
int r;
r = db_create(&db, env, 0);
CKERR(r);
r=db->open(db, NULL, dname, 0, DB_BTREE, DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO);
CKERR(r);
}
static void
close_db(void) {
assert(env!=NULL);
assert(db != NULL);
int r;
r = db->close(db, 0);
CKERR(r);
db = NULL;
}
static void
insert(void) {
assert(env!=NULL);
assert(db!=NULL);
DB_TXN *txn = txn_child ? txn_child : txn_parent;
assert(txn);
int r=db->put(db, txn,
&key,
&val,
DB_YESOVERWRITE);
CKERR(r);
}
static void
delete(void) {
assert(env!=NULL);
assert(db!=NULL);
DB_TXN *txn = txn_child ? txn_child : txn_parent;
assert(txn);
int r=db->del(db, txn,
&key,
DB_DELETE_ANY);
CKERR(r);
}
static void
perform_ops(int n) {
int i;
int j;
for (i = 0; i < n; i++) {
for (j = 0; j < 1024; j++) {
if (j & 0x20)
delete();
else
insert();
}
}
}
static void
progress_test_1(int n, int commit) {
start_env();
open_db();
{
start_txn();
{
start_txn();
perform_ops(n);
abort_txn(n);
}
{
start_txn();
perform_ops(n);
commit_txn(0);
}
perform_ops(n);
if (commit)
commit_txn(2*n);
else
abort_txn(2*n);
}
close_db();
end_env();
}
struct progress_stall_expect {
int num_calls;
BOOL has_been_stalled;
};
static void stall_poll(TOKU_TXN_PROGRESS progress, void *extra) {
struct progress_stall_expect *info = extra;
info->num_calls++;
assert(info->num_calls <= 2);
assert(progress->is_commit == FALSE);
if (!info->has_been_stalled) {
assert(info->num_calls==1);
assert(progress->stalled_on_checkpoint);
info->has_been_stalled = TRUE;
}
else {
assert(info->num_calls==2);
assert(!progress->stalled_on_checkpoint);
}
}
static void
abort_txn_stall_checkpoint(void) {
assert(env!=NULL);
assert(txn_parent);
assert(!txn_child);
struct progress_stall_expect extra = {
.num_calls = 0,
.has_been_stalled = FALSE
};
int r;
r=txn_parent->abort_with_progress(txn_parent, stall_poll, &extra);
CKERR(r);
assert(extra.num_calls == 2);
txn_parent = NULL;
}
static void
lock(void) {
assert(env!=NULL);
assert(db!=NULL);
assert(txn_parent);
assert(!txn_child);
int r=db->pre_acquire_table_lock(db, txn_parent);
CKERR(r);
}
static void
progress_test_2(void) {
start_env();
open_db();
start_txn();
lock();
abort_txn_stall_checkpoint();
close_db();
end_env();
}
int
test_main (int argc, char *argv[])
{
parse_args(argc, argv);
int commit;
for (commit = 0; commit <= 1; commit++) {
progress_test_1(4, commit);
}
progress_test_2();
return 0;
}
src/ydb.c
View file @ 14d67581
......@@ -133,7 +133,7 @@ static void env_remove_open_txn(DB_ENV *UU(env), DB_TXN *txn) {
toku_list_remove((struct toku_list *) (void *) &txn->open_txns);
}
static int toku_txn_abort(DB_TXN * txn);
static int toku_txn_abort(DB_TXN * txn, TXN_PROGRESS_POLL_FUNCTION, void*);
/* db methods */
static inline int db_opened(DB *db) {
......@@ -366,7 +366,7 @@ static int toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *db
static int toku_env_txn_checkpoint(DB_ENV * env, u_int32_t kbyte, u_int32_t min, u_int32_t flags);
static int toku_db_close(DB * db, u_int32_t flags);
static int toku_txn_begin(DB_ENV *env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags, int internal);
static int toku_txn_commit(DB_TXN * txn, u_int32_t flags);
static int toku_txn_commit(DB_TXN * txn, u_int32_t flags, TXN_PROGRESS_POLL_FUNCTION, void*);
static int db_open_iname(DB * db, DB_TXN * txn, const char *iname, u_int32_t flags, int mode);
static void finalize_file_removal(int fd, void * extra);
......@@ -710,7 +710,7 @@ toku_env_open(DB_ENV * env, const char *home, u_int32_t flags, int mode) {
assert(r==0);
}
if (using_txns) {
r = toku_txn_commit(txn, 0);
r = toku_txn_commit(txn, 0, NULL, NULL);
assert(r==0);
}
toku_ydb_unlock();
......@@ -1602,13 +1602,14 @@ static void ydb_yield (voidfp f, void *UU(v)) {
toku_ydb_lock();
}
static int toku_txn_commit(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) {
if (!txn) return EINVAL;
HANDLE_PANICKED_ENV(txn->mgrp);
//Recursively kill off children
if (db_txn_struct_i(txn)->child) {
//commit of child sets the child pointer to NULL
int r_child = toku_txn_commit(db_txn_struct_i(txn)->child, flags);
int r_child = toku_txn_commit(db_txn_struct_i(txn)->child, flags, NULL, NULL);
if (r_child !=0 && !toku_env_is_panicked(txn->mgrp)) {
txn->mgrp->i->is_panicked = r_child;
txn->mgrp->i->panic_string = toku_strdup("Recursive child commit failed during parent commit.\n");
......@@ -1636,12 +1637,12 @@ static int toku_txn_commit(DB_TXN * txn, u_int32_t flags) {
// frees the tokutxn
// Calls ydb_yield(NULL) occasionally
//r = toku_logger_abort(db_txn_struct_i(txn)->tokutxn, ydb_yield, NULL);
r = toku_txn_abort_txn(db_txn_struct_i(txn)->tokutxn, ydb_yield, NULL);
r = toku_txn_abort_txn(db_txn_struct_i(txn)->tokutxn, ydb_yield, NULL, poll, poll_extra);
else
// frees the tokutxn
// Calls ydb_yield(NULL) occasionally
//r = toku_logger_commit(db_txn_struct_i(txn)->tokutxn, nosync, ydb_yield, NULL);
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, ydb_yield, NULL, poll, poll_extra);
if (r!=0 && !toku_env_is_panicked(txn->mgrp)) {
txn->mgrp->i->is_panicked = r;
......@@ -1689,12 +1690,13 @@ static u_int32_t toku_txn_id(DB_TXN * txn) {
return -1;
}
static int toku_txn_abort(DB_TXN * txn) {
static int toku_txn_abort(DB_TXN * txn,
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) {
//commit of child sets the child pointer to NULL
int r_child = toku_txn_commit(db_txn_struct_i(txn)->child, DB_TXN_NOSYNC);
int r_child = toku_txn_commit(db_txn_struct_i(txn)->child, DB_TXN_NOSYNC, NULL, NULL);
if (r_child !=0 && !toku_env_is_panicked(txn->mgrp)) {
txn->mgrp->i->is_panicked = r_child;
txn->mgrp->i->panic_string = toku_strdup("Recursive child commit failed during parent abort.\n");
......@@ -1714,7 +1716,7 @@ static int toku_txn_abort(DB_TXN * txn) {
assert(toku_list_empty(&db_txn_struct_i(txn)->dbs_that_must_close_before_abort));
//int r = toku_logger_abort(db_txn_struct_i(txn)->tokutxn, ydb_yield, NULL);
int r = toku_txn_abort_txn(db_txn_struct_i(txn)->tokutxn, ydb_yield, NULL);
int r = toku_txn_abort_txn(db_txn_struct_i(txn)->tokutxn, ydb_yield, NULL, poll, poll_extra);
if (r!=0 && !toku_env_is_panicked(txn->mgrp)) {
txn->mgrp->i->is_panicked = r;
txn->mgrp->i->panic_string = toku_strdup("Error during abort.\n");
......@@ -1750,20 +1752,34 @@ static int locked_txn_stat (DB_TXN *txn, struct txn_stat **txn_stat) {
toku_ydb_lock(); u_int32_t r = toku_txn_stat(txn, txn_stat); toku_ydb_unlock(); return r;
}
static int locked_txn_commit(DB_TXN *txn, u_int32_t flags) {
static int locked_txn_commit_with_progress(DB_TXN *txn, u_int32_t flags,
TXN_PROGRESS_POLL_FUNCTION poll, void* poll_extra) {
toku_multi_operation_client_lock(); //Cannot checkpoint during a commit.
toku_ydb_lock(); int r = toku_txn_commit(txn, flags); toku_ydb_unlock();
toku_ydb_lock(); int r = toku_txn_commit(txn, flags, poll, poll_extra); toku_ydb_unlock();
toku_multi_operation_client_unlock(); //Cannot checkpoint during a commit.
return r;
}
static int locked_txn_abort(DB_TXN *txn) {
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(txn); toku_ydb_unlock();
toku_ydb_lock(); int r = toku_txn_abort(txn, poll, poll_extra); toku_ydb_unlock();
toku_multi_operation_client_unlock(); //Cannot checkpoint during an abort.
return r;
}
static int locked_txn_commit(DB_TXN *txn, u_int32_t flags) {
int r;
r = locked_txn_commit_with_progress(txn, flags, NULL, NULL);
return r;
}
static int locked_txn_abort(DB_TXN *txn) {
int r;
r = locked_txn_abort_with_progress(txn, NULL, NULL);
return r;
}
static int toku_txn_begin(DB_ENV *env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags, int internal) {
HANDLE_PANICKED_ENV(env);
HANDLE_ILLEGAL_WORKING_PARENT_TXN(env, stxn); //Cannot create child while child already exists.
......@@ -1813,11 +1829,17 @@ static int toku_txn_begin(DB_ENV *env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t f
memset(result, 0, result_size);
//toku_ydb_notef("parent=%p flags=0x%x\n", stxn, flags);
result->mgrp = env;
result->abort = locked_txn_abort;
result->commit = locked_txn_commit;
result->id = locked_txn_id;
result->parent = stxn;
#define STXN(name) result->name = locked_txn_ ## name
STXN(abort);
STXN(commit);
STXN(abort_with_progress);
STXN(commit_with_progress);
STXN(id);
#undef STXN
result->txn_stat = locked_txn_stat;
result->parent = stxn;
#if !TOKUDB_NATIVE_H
MALLOC(db_txn_struct_i(result));
if (!db_txn_struct_i(result)) {
......@@ -4087,11 +4109,11 @@ toku_db_open(DB * db, DB_TXN * txn, const char *fname, const char *dbname, DBTYP
if (using_txns) {
// close txn
if (r == 0) { // commit
r = toku_txn_commit(child, DB_TXN_NOSYNC);
r = toku_txn_commit(child, DB_TXN_NOSYNC, NULL, NULL);
assert(r==0); // TODO panic
}
else { // abort
int r2 = toku_txn_abort(child);
int r2 = toku_txn_abort(child, NULL, NULL);
assert(r2==0); // TODO panic
}
}
......@@ -4475,11 +4497,11 @@ toku_env_dbremove(DB_ENV * env, DB_TXN *txn, const char *fname, const char *dbna
if (using_txns) {
// close txn
if (r == 0) { // commit
r = toku_txn_commit(child, DB_TXN_NOSYNC);
r = toku_txn_commit(child, DB_TXN_NOSYNC, NULL, NULL);
assert(r==0); // TODO panic
}
else { // abort
int r2 = toku_txn_abort(child);
int r2 = toku_txn_abort(child, NULL, NULL);
assert(r2==0); // TODO panic
}
}
......@@ -4596,11 +4618,11 @@ toku_env_dbrename(DB_ENV *env, DB_TXN *txn, const char *fname, const char *dbnam
if (using_txns) {
// close txn
if (r == 0) { // commit
r = toku_txn_commit(child, DB_TXN_NOSYNC);
r = toku_txn_commit(child, DB_TXN_NOSYNC, NULL, NULL);
assert(r==0); // TODO panic
}
else { // abort
int r2 = toku_txn_abort(child);
int r2 = toku_txn_abort(child, NULL, NULL);
assert(r2==0); // TODO panic
}
}
......@@ -4809,8 +4831,8 @@ static inline int toku_db_construct_autotxn(DB* db, DB_TXN **txn, BOOL* changed,
static inline int toku_db_destruct_autotxn(DB_TXN *txn, int r, BOOL changed) {
if (!changed) return r;
if (r==0) return toku_txn_commit(txn, 0);
toku_txn_abort(txn);
if (r==0) return toku_txn_commit(txn, 0, NULL, NULL);
toku_txn_abort(txn, NULL, NULL);
return r;
}
......@@ -5300,11 +5322,11 @@ ydb_load_inames(DB_ENV * env, DB_TXN * txn, int N, DB * dbs[N], char * new_iname
if (using_txns) {
// close txn
if (rval == 0) { // all well so far, commit child
rval = toku_txn_commit(child, DB_TXN_NOSYNC);
rval = toku_txn_commit(child, DB_TXN_NOSYNC, NULL, NULL);
assert(rval==0);
}
else { // abort child
int r2 = toku_txn_abort(child);
int r2 = toku_txn_abort(child, NULL, NULL);
assert(r2==0);
for (i=0; i<N; i++) {
if (new_inames[i]) {
......
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