Skip to content

M
MariaDB
Commit 24c21821 authored by Bradley C. Kuszmaul's avatar Bradley C. Kuszmaul
Browse files
Options

Merge logging 

git-svn-id: file:///svn/tokudb@371 c7de825b-a66e-492c-adef-691d508d4ae1
parent 55ced3bf
Hide whitespace changes
Inline Side-by-side
Showing with 708 additions and 294 deletions
buildheader/make_db_h.c
View file @ 24c21821
......@@ -227,7 +227,9 @@ int main (int argc __attribute__((__unused__)), char *argv[] __attribute__((__un
printf("#ifdef _TOKUDB_WRAP_H\n#define txn_begin txn_begin_tokudb\n#endif\n");
printf("int db_env_create(DB_ENV **, u_int32_t);\n");
printf("int db_create(DB **, DB_ENV *, u_int32_t);\n");
printf("char *db_strerror(int);\n");
printf("#if defined(__cplusplus)\n}\n#endif\n");
printf("#endif\n");
return 0;
}
include/db.h
View file @ 24c21821
......@@ -204,6 +204,7 @@ struct __toku_dbt {
#endif
int db_env_create(DB_ENV **, u_int32_t);
int db_create(DB **, DB_ENV *, u_int32_t);
char *db_strerror(int);
#if defined(__cplusplus)
}
#endif
......
newbrt/Makefile
View file @ 24c21821
......@@ -23,7 +23,7 @@ endif
# When debugging, try: valgrind --show-reachable=yes --leak-check=full ./brt-test
default: bins
default: bins libs
# Put these one-per-line so that if we insert a new one the svn diff can understand it better.
# Also keep them sorted.
REGRESSION_TESTS = \
......@@ -42,6 +42,7 @@ BINS = $(REGRESSION_TESTS) \
randdb4 \
# This line intentially kept commented so I can have a \ on the end of the previous line
libs: log.o
bins: $(BINS)
check: bins
$(DTOOL) ./ybt-test
......@@ -62,14 +63,14 @@ check-fanout:
# pma: PROF_FLAGS=-fprofile-arcs -ftest-coverage
key.o: brttypes.h key.h
pma-test.o: pma-internal.h pma.h yerror.h memory.h ../include/db.h
pma-test: pma.o memory.o key.o ybt.o
pma-test.o: pma-internal.h pma.h yerror.h memory.h ../include/db.h list.h kv-pair.h brttypes.h ybt.h yerror.h
pma-test: pma.o memory.o key.o ybt.o log.o
pma.o: pma.h yerror.h pma-internal.h memory.h key.h ybt.h brttypes.h ../include/db.h
ybt.o: ybt.h brttypes.h ../include/db.h
ybt-test: ybt-test.o ybt.o memory.o
ybt-test.o: ybt.h ../include/db.h
cachetable.o: cachetable.h hashfun.h
brt-test: ybt.o brt.o hashtable.o pma.o memory.o brt-serialize.o cachetable.o header-io.o ybt.o key.o primes.o
brt-test: ybt.o brt.o hashtable.o pma.o memory.o brt-serialize.o cachetable.o header-io.o ybt.o key.o primes.o log.o
brt-test.o brt.o: brt.h ../include/db.h hashtable.h pma.h brttypes.h cachetable.h
brt-serialize-test.o: pma.h yerror.h brt.h ../include/db.h memory.h hashtable.h brttypes.h brt-internal.h
brt.o: brt.h ../include/db.h mdict.h pma.h brttypes.h memory.h brt-internal.h cachetable.h hashtable.h
......@@ -85,7 +86,7 @@ brt-bigtest: memory.o ybt.o brt.o pma.o cachetable.o key.o hashtable.o brt-seria
brt-bigtest.o: brt.h ../include/db.h
log-test: log.o memory.o
brt-serialize-test: brt-serialize-test.o brt-serialize.o memory.o hashtable.o pma.o key.o ybt.o brt.o cachetable.o primes.o
brt-serialize-test: brt-serialize-test.o brt-serialize.o memory.o hashtable.o pma.o key.o ybt.o brt.o cachetable.o primes.o log.o
cachetable-test.o: cachetable.h memory.h
cachetable-test: cachetable.o memory.o cachetable-test.o
......@@ -93,7 +94,7 @@ cachetable-test: cachetable.o memory.o cachetable-test.o
cachetable-test2.o: cachetable.h memory.h
cachetable-test2: cachetable.o memory.o cachetable-test2.o
benchmark-test: benchmark-test.o ybt.o memory.o brt.o pma.o cachetable.o key.o hashtable.o brt-serialize.o primes.o
benchmark-test: benchmark-test.o ybt.o memory.o brt.o pma.o cachetable.o key.o hashtable.o brt-serialize.o primes.o log.o
benchmark-test.o: brt.h ../include/db.h
clean:
......
newbrt/benchmark-test.c
View file @ 24c21821
......@@ -44,7 +44,7 @@ void insert (long long v) {
DBT kt, vt;
long_long_to_array(kc, v);
long_long_to_array(vc, v);
brt_insert(t, fill_dbt(&kt, kc, 8), fill_dbt(&vt, vc, 8), 0);
brt_insert(t, fill_dbt(&kt, kc, 8), fill_dbt(&vt, vc, 8), 0, 0);
}
void serial_insert_from (long long from) {
......
newbrt/brt-internal.h
View file @ 24c21821
......@@ -4,8 +4,6 @@
#include "brt.h"
//#include "pma.h"
typedef long long diskoff; /* Offset in a disk. -1 is the NULL pointer. */
#ifndef BRT_FANOUT
#define BRT_FANOUT 16
#endif
......@@ -76,7 +74,7 @@ struct brt {
};
/* serialization code */
int serialize_brtnode_to(int fd, diskoff off, diskoff size, BRTNODE node);
void serialize_brtnode_to(int fd, diskoff off, diskoff size, BRTNODE node);
int deserialize_brtnode_from (int fd, diskoff off, BRTNODE *brtnode, int nodesize);
unsigned int serialize_brtnode_size(BRTNODE node); /* How much space will it take? */
int keycompare (bytevec key1, ITEMLEN key1len, bytevec key2, ITEMLEN key2len);
......
newbrt/brt-serialize-test.c
View file @ 24c21821
......@@ -32,7 +32,7 @@ void test_serialize(void) {
r = toku_hash_insert(sn.u.n.htables[1], "x", 2, "xval", 5, BRT_NONE); assert(r==0);
sn.u.n.n_bytes_in_hashtables = 3*(BRT_CMD_OVERHEAD+KEY_VALUE_OVERHEAD+2+5);
r = serialize_brtnode_to(fd, sn.nodesize*20, sn.nodesize, &sn); assert(r==0);
serialize_brtnode_to(fd, sn.nodesize*20, sn.nodesize, &sn); assert(r==0);
r = deserialize_brtnode_from(fd, nodesize*20, &dn, nodesize);
......
newbrt/brt-serialize.c
View file @ 24c21821
......@@ -73,13 +73,13 @@ unsigned int serialize_brtnode_size (BRTNODE node) {
return result;
}
int serialize_brtnode_to(int fd, diskoff off, diskoff size, BRTNODE node) {
void serialize_brtnode_to(int fd, diskoff off, diskoff size, BRTNODE node) {
struct wbuf w;
int i;
unsigned int calculated_size = serialize_brtnode_size(node);
int r;
char buf[size];
assert(size>0);
if ((r=wbuf_init(&w, size))) return r;
wbuf_init(&w, buf, size);
//printf("%s:%d serializing %lld w height=%d p0=%p\n", __FILE__, __LINE__, off, node->height, node->mdicts[0]);
wbuf_int(&w, calculated_size);
wbuf_int(&w, node->height);
......@@ -124,8 +124,6 @@ int serialize_brtnode_to(int fd, diskoff off, diskoff size, BRTNODE node) {
//printf("%s:%d wrote %d bytes for %lld size=%lld\n", __FILE__, __LINE__, w.ndone, off, size);
assert(w.ndone<=size);
toku_free(w.buf);
return 0;
}
int deserialize_brtnode_from (int fd, diskoff off, BRTNODE *brtnode, int nodesize) {
......
newbrt/brt-test.c
View file @ 24c21821
......@@ -11,6 +11,9 @@
#include <unistd.h>
#include <arpa/inet.h>
TOKUTXN const null_txn = 0;
DB * const null_db = 0;
extern long long n_items_malloced;
static void test0 (void) {
......@@ -49,7 +52,7 @@ static void test1 (void) {
unlink(fname);
r = open_brt(fname, 0, 1, &t, 1024, ct, default_compare_fun);
assert(r==0);
brt_insert(t, fill_dbt(&k, "hello", 6), fill_dbt(&v, "there", 6), 0);
brt_insert(t, fill_dbt(&k, "hello", 6), fill_dbt(&v, "there", 6), null_db, null_txn);
{
r = brt_lookup(t, fill_dbt(&k, "hello", 6), init_dbt(&v), 0);
assert(r==0);
......@@ -81,7 +84,7 @@ static void test2 (int memcheck) {
char key[100],val[100];
snprintf(key,100,"hello%d",i);
snprintf(val,100,"there%d",i);
brt_insert(t, fill_dbt(&k, key, 1+strlen(key)), fill_dbt(&v, val, 1+strlen(val)), 0);
brt_insert(t, fill_dbt(&k, key, 1+strlen(key)), fill_dbt(&v, val, 1+strlen(val)), null_db, null_txn);
//printf("%s:%d did insert %d\n", __FILE__, __LINE__, i);
if (0) {
brt_flush(t);
......@@ -119,7 +122,7 @@ static void test3 (int nodesize, int count, int memcheck) {
DBT k,v;
snprintf(key,100,"hello%d",i);
snprintf(val,100,"there%d",i);
brt_insert(t, fill_dbt(&k, key, 1+strlen(key)), fill_dbt(&v, val, 1+strlen(val)), 0);
brt_insert(t, fill_dbt(&k, key, 1+strlen(key)), fill_dbt(&v, val, 1+strlen(val)), null_db, null_txn);
}
r = close_brt(t); assert(r==0);
r = cachetable_close(&ct); assert(r==0);
......@@ -150,7 +153,7 @@ static void test4 (int nodesize, int count, int memcheck) {
DBT k,v;
snprintf(key,100,"hello%d",rv);
snprintf(val,100,"there%d",i);
brt_insert(t, fill_dbt(&k, key, 1+strlen(key)), fill_dbt(&v, val, 1+strlen(val)), 0);
brt_insert(t, fill_dbt(&k, key, 1+strlen(key)), fill_dbt(&v, val, 1+strlen(val)), null_db, null_txn);
}
r = close_brt(t); assert(r==0);
r = cachetable_close(&ct); assert(r==0);
......@@ -185,7 +188,7 @@ static void test5 (void) {
snprintf(key, 100, "key%d", rk);
snprintf(val, 100, "val%d", rv);
DBT k,v;
brt_insert(t, fill_dbt(&k, key, 1+strlen(key)), fill_dbt(&v, val, 1+strlen(val)), 0);
brt_insert(t, fill_dbt(&k, key, 1+strlen(key)), fill_dbt(&v, val, 1+strlen(val)), null_db, null_txn);
}
printf("\n");
for (i=0; i<limit/2; i++) {
......@@ -245,9 +248,9 @@ static void test_multiple_files_of_size (int size) {
DBT k,v;
snprintf(key, 100, "key%d", i);
snprintf(val, 100, "val%d", i);
brt_insert(t0, fill_dbt(&k, key, 1+strlen(key)), fill_dbt(&v, val, 1+strlen(val)), 0);
brt_insert(t0, fill_dbt(&k, key, 1+strlen(key)), fill_dbt(&v, val, 1+strlen(val)), null_db, null_txn);
snprintf(val, 100, "Val%d", i);
brt_insert(t1, fill_dbt(&k, key, 1+strlen(key)), fill_dbt(&v, val, 1+strlen(val)), 0);
brt_insert(t1, fill_dbt(&k, key, 1+strlen(key)), fill_dbt(&v, val, 1+strlen(val)), null_db, null_txn);
}
//verify_brt(t0);
//dump_brt(t0);
......@@ -310,7 +313,7 @@ static void test_named_db (void) {
r = open_brt(n0, "db1", 1, &t0, 1<<12, ct, default_compare_fun); assert(r==0);
brt_insert(t0, fill_dbt(&k, "good", 5), fill_dbt(&v, "day", 4), 0); assert(r==0);
brt_insert(t0, fill_dbt(&k, "good", 5), fill_dbt(&v, "day", 4), null_db, null_txn); assert(r==0);
r = close_brt(t0); assert(r==0);
r = cachetable_close(&ct); assert(r==0);
......@@ -347,8 +350,8 @@ static void test_multiple_dbs (void) {
r = open_brt(n0, "db1", 1, &t0, 1<<12, ct, default_compare_fun); assert(r==0);
r = open_brt(n1, "db2", 1, &t1, 1<<12, ct, default_compare_fun); assert(r==0);
brt_insert(t0, fill_dbt(&k, "good", 5), fill_dbt(&v, "grief", 6), 0); assert(r==0);
brt_insert(t1, fill_dbt(&k, "bad", 4), fill_dbt(&v, "night", 6), 0); assert(r==0);
brt_insert(t0, fill_dbt(&k, "good", 5), fill_dbt(&v, "grief", 6), null_db, null_txn); assert(r==0);
brt_insert(t1, fill_dbt(&k, "bad", 4), fill_dbt(&v, "night", 6), null_db, null_txn); assert(r==0);
r = close_brt(t0); assert(r==0);
r = close_brt(t1); assert(r==0);
......@@ -408,7 +411,7 @@ static void test_multiple_dbs_many (void) {
DBT kdbt,vdbt;
snprintf(k, 20, "key%d", i);
snprintf(v, 20, "val%d", i);
brt_insert(trees[i], fill_dbt(&kdbt, k, strlen(k)+1), fill_dbt(&vdbt, v, strlen(v)+1), 0);
brt_insert(trees[i], fill_dbt(&kdbt, k, strlen(k)+1), fill_dbt(&vdbt, v, strlen(v)+1), null_db, null_txn);
}
for (i=0; i<MANYN; i++) {
r = close_brt(trees[i]); assert(r==0);
......@@ -437,7 +440,7 @@ static void test_multiple_brts_one_db_one_file (void) {
DBT kb, vb;
snprintf(k, 20, "key%d", i);
snprintf(v, 20, "val%d", i);
brt_insert(trees[i], fill_dbt(&kb, k, strlen(k)+1), fill_dbt(&vb, v, strlen(v)+1), 0);
brt_insert(trees[i], fill_dbt(&kb, k, strlen(k)+1), fill_dbt(&vb, v, strlen(v)+1), null_db, null_txn);
}
for (i=0; i<MANYN; i++) {
char k[20],vexpect[20];
......@@ -484,7 +487,7 @@ static void test_read_what_was_written (void) {
r = open_brt(n, 0, 0, &brt, 1<<12, ct, default_compare_fun); assert(r==0);
/* See if we can put something in it. */
brt_insert(brt, fill_dbt(&k, "hello", 6), fill_dbt(&v, "there", 6), 0);
brt_insert(brt, fill_dbt(&k, "hello", 6), fill_dbt(&v, "there", 6), null_db, null_txn);
r = close_brt(brt); assert(r==0);
r = cachetable_close(&ct); assert(r==0);
......@@ -516,7 +519,7 @@ static void test_read_what_was_written (void) {
int verify_result=verify_brt(brt);;
assert(verify_result==0);
}
brt_insert(brt, fill_dbt(&k, key, strlen(key)+1), fill_dbt(&v, val, strlen(val)+1), 0);
brt_insert(brt, fill_dbt(&k, key, strlen(key)+1), fill_dbt(&v, val, strlen(val)+1), null_db, null_txn);
if (i<600) {
int verify_result=verify_brt(brt);
if (verify_result) {
......@@ -620,10 +623,10 @@ void test_cursor_last_empty(void) {
//printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
init_dbt(&vbt);
//printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
r = brt_cursor_get(cursor, &kbt, &vbt, DB_LAST, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_LAST, null_db, null_txn);
//printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
assert(r==DB_NOTFOUND);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_FIRST, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_FIRST, null_db, null_txn);
//printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
assert(r==DB_NOTFOUND);
r = close_brt(brt);
......@@ -647,8 +650,8 @@ void test_cursor_next (void) {
//printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
r = open_brt(n, 0, 1, &brt, 1<<12, ct, default_compare_fun); assert(r==0);
//printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
r = brt_insert(brt, fill_dbt(&kbt, "hello", 6), fill_dbt(&vbt, "there", 6), 0);
r = brt_insert(brt, fill_dbt(&kbt, "byebye", 7), fill_dbt(&vbt, "byenow", 7), 0);
r = brt_insert(brt, fill_dbt(&kbt, "hello", 6), fill_dbt(&vbt, "there", 6), null_db, null_txn);
r = brt_insert(brt, fill_dbt(&kbt, "byebye", 7), fill_dbt(&vbt, "byenow", 7), null_db, null_txn);
printf("%s:%d calling brt_cursor(...)\n", __FILE__, __LINE__);
r = brt_cursor(brt, &cursor); assert(r==0);
init_dbt(&kbt);
......@@ -657,7 +660,7 @@ void test_cursor_next (void) {
//printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
printf("%s:%d calling brt_cursor_get(...)\n", __FILE__, __LINE__);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, null_db, null_txn);
printf("%s:%d called brt_cursor_get(...)\n", __FILE__, __LINE__);
//printf("%s:%d %d alloced\n", __FILE__, __LINE__, get_n_items_malloced()); print_malloced_items();
assert(r==0);
......@@ -666,14 +669,14 @@ void test_cursor_next (void) {
assert(vbt.size==7);
assert(memcmp(vbt.data, "byenow", 7)==0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, null_db, null_txn);
assert(r==0);
assert(kbt.size==6);
assert(memcmp(kbt.data, "hello", 6)==0);
assert(vbt.size==6);
assert(memcmp(vbt.data, "there", 6)==0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, null_db, null_txn);
assert(r==DB_NOTFOUND);
r = close_brt(brt);
......@@ -732,7 +735,7 @@ static void test_wrongendian_compare (int wrong_p, unsigned int N) {
r = brt_create_cachetable(&ct, 0); assert(r==0);
printf("%s:%d WRONG=%d\n", __FILE__, __LINE__, wrong_p);
if (0) {
if (0) { // ???? Why is this commented out?
r = open_brt(n, 0, 1, &brt, 1<<20, ct, wrong_p ? wrong_compare_fun : default_compare_fun); assert(r==0);
for (i=1; i<257; i+=255) {
unsigned char a[4],b[4];
......@@ -745,14 +748,14 @@ static void test_wrongendian_compare (int wrong_p, unsigned int N) {
printf("%s:%d insert: %02x%02x%02x%02x -> %02x%02x%02x%02x\n", __FILE__, __LINE__,
((char*)kbt.data)[0], ((char*)kbt.data)[1], ((char*)kbt.data)[2], ((char*)kbt.data)[3],
((char*)vbt.data)[0], ((char*)vbt.data)[1], ((char*)vbt.data)[2], ((char*)vbt.data)[3]);
r = brt_insert(brt, &kbt, &vbt, &nonce_db);
r = brt_insert(brt, &kbt, &vbt, &nonce_db, null_txn);
assert(r==0);
}
r = brt_cursor(brt, &cursor); assert(r==0);
for (i=0; i<2; i++) {
init_dbt(&kbt); init_dbt(&vbt);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, null_db, null_txn);
assert(r==0);
assert(kbt.size==4 && vbt.size==4);
printf("%s:%d %02x%02x%02x%02x -> %02x%02x%02x%02x\n", __FILE__, __LINE__,
......@@ -778,7 +781,7 @@ static void test_wrongendian_compare (int wrong_p, unsigned int N) {
if (0) printf("%s:%d insert: %02x%02x%02x%02x -> %02x%02x%02x%02x\n", __FILE__, __LINE__,
((unsigned char*)kbt.data)[0], ((unsigned char*)kbt.data)[1], ((unsigned char*)kbt.data)[2], ((unsigned char*)kbt.data)[3],
((unsigned char*)vbt.data)[0], ((unsigned char*)vbt.data)[1], ((unsigned char*)vbt.data)[2], ((unsigned char*)vbt.data)[3]);
r = brt_insert(brt, &kbt, &vbt, &nonce_db);
r = brt_insert(brt, &kbt, &vbt, &nonce_db, null_txn);
assert(r==0);
}
r = brt_cursor(brt, &cursor); assert(r==0);
......@@ -787,7 +790,7 @@ static void test_wrongendian_compare (int wrong_p, unsigned int N) {
for (i=0; i<N; i++) {
int this;
init_dbt(&kbt); init_dbt(&vbt);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, null_db, null_txn);
assert(r==0);
assert(kbt.size==4 && vbt.size==4);
if (0) printf("%s:%d %02x%02x%02x%02x -> %02x%02x%02x%02x\n", __FILE__, __LINE__,
......@@ -822,7 +825,7 @@ void assert_cursor_notfound(BRT brt, int position) {
init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &kbt, &vbt, position, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, position, null_db, null_txn);
assert(r == DB_NOTFOUND);
r = brt_cursor_close(cursor);
......@@ -841,7 +844,7 @@ void assert_cursor_value(BRT brt, int position, long long value) {
if (test_cursor_debug) printf("key: ");
init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &kbt, &vbt, position, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, position, null_db, null_txn);
assert(r == 0);
if (test_cursor_debug) printf("%s ", (char*)kbt.data);
assert(vbt.size == sizeof v);
......@@ -867,7 +870,7 @@ void assert_cursor_first_last(BRT brt, long long firstv, long long lastv) {
if (test_cursor_debug) printf("first key: ");
init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &kbt, &vbt, DB_FIRST, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_FIRST, null_db, null_txn);
assert(r == 0);
if (test_cursor_debug) printf("%s ", (char*)kbt.data);
assert(vbt.size == sizeof v);
......@@ -880,7 +883,7 @@ void assert_cursor_first_last(BRT brt, long long firstv, long long lastv) {
if (test_cursor_debug) printf("last key:");
init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &kbt, &vbt, DB_LAST, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_LAST, null_db, null_txn);
assert(r == 0);
if (test_cursor_debug)printf("%s ", (char*)kbt.data);
assert(vbt.size == sizeof v);
......@@ -920,7 +923,7 @@ void test_brt_cursor_first(int n) {
fill_dbt(&kbt, key, strlen(key)+1);
v = i;
fill_dbt(&vbt, &v, sizeof v);
r = brt_insert(brt, &kbt, &vbt, 0);
r = brt_insert(brt, &kbt, &vbt, 0, 0);
assert(r==0);
}
......@@ -962,7 +965,7 @@ void test_brt_cursor_last(int n) {
fill_dbt(&kbt, key, strlen(key)+1);
v = i;
fill_dbt(&vbt, &v, sizeof v);
r = brt_insert(brt, &kbt, &vbt, 0);
r = brt_insert(brt, &kbt, &vbt, 0, 0);
assert(r==0);
}
......@@ -1004,7 +1007,7 @@ void test_brt_cursor_first_last(int n) {
fill_dbt(&kbt, key, strlen(key)+1);
v = i;
fill_dbt(&vbt, &v, sizeof v);
r = brt_insert(brt, &kbt, &vbt, 0);
r = brt_insert(brt, &kbt, &vbt, 0, 0);
assert(r==0);
}
......@@ -1047,7 +1050,7 @@ void test_brt_cursor_rfirst(int n) {
fill_dbt(&kbt, key, strlen(key)+1);
v = i;
fill_dbt(&vbt, &v, sizeof v);
r = brt_insert(brt, &kbt, &vbt, 0);
r = brt_insert(brt, &kbt, &vbt, 0, 0);
assert(r==0);
}
......@@ -1078,7 +1081,7 @@ void assert_cursor_walk(BRT brt, int n) {
init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, null_db, null_txn);
if (r != 0)
break;
if (test_cursor_debug) printf("%s ", (char*)kbt.data);
......@@ -1121,7 +1124,7 @@ void test_brt_cursor_walk(int n) {
fill_dbt(&kbt, key, strlen(key)+1);
v = i;
fill_dbt(&vbt, &v, sizeof v);
r = brt_insert(brt, &kbt, &vbt, 0);
r = brt_insert(brt, &kbt, &vbt, 0, 0);
assert(r==0);
}
......@@ -1150,7 +1153,7 @@ void assert_cursor_rwalk(BRT brt, int n) {
init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &kbt, &vbt, DB_PREV, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_PREV, null_db, null_txn);
if (r != 0)
break;
if (test_cursor_debug) printf("%s ", (char*)kbt.data);
......@@ -1193,7 +1196,7 @@ void test_brt_cursor_rwalk(int n) {
fill_dbt(&kbt, &k, sizeof k);
v = i;
fill_dbt(&vbt, &v, sizeof v);
r = brt_insert(brt, &kbt, &vbt, 0);
r = brt_insert(brt, &kbt, &vbt, 0, 0);
assert(r==0);
}
......@@ -1224,7 +1227,7 @@ void assert_cursor_walk_inorder(BRT brt, int n) {
init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, null_db, null_txn);
if (r != 0)
break;
if (test_cursor_debug) printf("%s ", (char*)kbt.data);
......@@ -1278,7 +1281,7 @@ void test_brt_cursor_rand(int n) {
printf("dup");
continue;
}
r = brt_insert(brt, &kbt, &vbt, 0);
r = brt_insert(brt, &kbt, &vbt, 0, 0);
assert(r==0);
break;
}
......@@ -1322,7 +1325,7 @@ void test_brt_cursor_split(int n) {
fill_dbt(&kbt, key, strlen(key)+1);
v = keyseqnum;
fill_dbt(&vbt, &v, sizeof v);
r = brt_insert(brt, &kbt, &vbt, 0);
r = brt_insert(brt, &kbt, &vbt, 0, 0);
assert(r==0);
}
......@@ -1333,7 +1336,7 @@ void test_brt_cursor_split(int n) {
for (i=0; i<n/2; i++) {
init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, null_db, null_txn);
assert(r==0);
if (test_cursor_debug) printf("%s ", (char*)kbt.data);
toku_free(kbt.data);
......@@ -1348,7 +1351,7 @@ void test_brt_cursor_split(int n) {
fill_dbt(&kbt, key, strlen(key)+1);
v = keyseqnum;
fill_dbt(&vbt, &v, sizeof v);
r = brt_insert(brt, &kbt, &vbt, 0);
r = brt_insert(brt, &kbt, &vbt, 0, 0);
assert(r==0);
}
......@@ -1356,7 +1359,7 @@ void test_brt_cursor_split(int n) {
for (;;) {
init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, 0);
r = brt_cursor_get(cursor, &kbt, &vbt, DB_NEXT, null_db, null_txn);
if (r != 0)
break;
if (test_cursor_debug) printf("%s ", (char*)kbt.data);
......@@ -1459,7 +1462,7 @@ void test_multiple_brt_cursor_walk(int n) {
v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
r = brt_insert(brt, &key, &val, 0);
r = brt_insert(brt, &key, &val, 0, 0);
assert(r == 0);
/* point cursor i / cursor_gap to the current last key i */
......@@ -1467,7 +1470,7 @@ void test_multiple_brt_cursor_walk(int n) {
c = i / cursor_gap;
init_dbt(&key); key.flags = DB_DBT_MALLOC;
init_dbt(&val); val.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursors[c], &key, &val, DB_LAST, 0);
r = brt_cursor_get(cursors[c], &key, &val, DB_LAST, null_db, null_txn);
assert(r == 0);
toku_free(key.data);
toku_free(val.data);
......@@ -1479,7 +1482,7 @@ void test_multiple_brt_cursor_walk(int n) {
for (c=0; c<ncursors; c++) {
init_dbt(&key); key.flags = DB_DBT_MALLOC;
init_dbt(&val); val.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursors[c], &key, &val, DB_NEXT, 0);
r = brt_cursor_get(cursors[c], &key, &val, DB_NEXT, null_db, null_txn);
if (r == DB_NOTFOUND) {
/* we already consumed 1 previously */
assert(i == cursor_gap-1);
......@@ -1534,7 +1537,7 @@ void test_brt_cursor_set(int n, int cursor_op) {
v = 10*i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
r = brt_insert(brt, &key, &val, 0);
r = brt_insert(brt, &key, &val, 0, 0);
assert(r == 0);
}
......@@ -1549,7 +1552,7 @@ void test_brt_cursor_set(int n, int cursor_op) {
k = htonl(v);
fill_dbt(&key, &k, sizeof k);
init_dbt(&val); val.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &key, &val, cursor_op, 0);
r = brt_cursor_get(cursor, &key, &val, cursor_op, null_db, null_txn);
assert(r == 0);
assert(val.size == sizeof vv);
memcpy(&vv, val.data, val.size);
......@@ -1564,7 +1567,7 @@ void test_brt_cursor_set(int n, int cursor_op) {
k = htonl(i);
fill_dbt(&key, &k, sizeof k);
init_dbt(&val); val.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &key, &val, DB_SET, 0);
r = brt_cursor_get(cursor, &key, &val, DB_SET, null_db, null_txn);
assert(r == DB_NOTFOUND);
}
......@@ -1606,7 +1609,7 @@ void test_brt_cursor_set_range(int n) {
v = 10*i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
r = brt_insert(brt, &key, &val, 0);
r = brt_insert(brt, &key, &val, 0, 0);
assert(r == 0);
}
......@@ -1622,7 +1625,7 @@ void test_brt_cursor_set_range(int n) {
k = htonl(v);
fill_dbt(&key, &k, sizeof k);
init_dbt(&val); val.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &key, &val, DB_SET_RANGE, 0);
r = brt_cursor_get(cursor, &key, &val, DB_SET_RANGE, null_db, null_txn);
if (v > max_key)
/* there is no smallest key if v > the max key */
assert(r == DB_NOTFOUND);
......@@ -1675,7 +1678,7 @@ void test_brt_cursor_delete(int n) {
v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = brt_insert(brt, &key, &val, 0);
error = brt_insert(brt, &key, &val, 0, 0);
assert(error == 0);
}
......@@ -1683,7 +1686,7 @@ void test_brt_cursor_delete(int n) {
for (;;) {
init_dbt(&key); key.flags = DB_DBT_MALLOC;
init_dbt(&val); val.flags = DB_DBT_MALLOC;
error = brt_cursor_get(cursor, &key, &val, DB_NEXT, 0);
error = brt_cursor_get(cursor, &key, &val, DB_NEXT, null_db, null_txn);
if (error == DB_NOTFOUND)
break;
assert(error == 0);
......@@ -1735,7 +1738,7 @@ void test_brt_cursor_get_both(int n) {
v = n+1;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, 0);
error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, null_db, null_txn);
assert(error == DB_NOTFOUND);
int i;
......@@ -1745,7 +1748,7 @@ void test_brt_cursor_get_both(int n) {
v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = brt_insert(brt, &key, &val, 0);
error = brt_insert(brt, &key, &val, 0, 0);
assert(error == 0);
}
......@@ -1754,7 +1757,7 @@ void test_brt_cursor_get_both(int n) {
v = n-1;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, 0);
error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, null_db, null_txn);
assert(error == DB_NOTFOUND);
/* verify that key match but data mismatch fails */
......@@ -1763,7 +1766,7 @@ void test_brt_cursor_get_both(int n) {
v = i+1;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, 0);
error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, null_db, null_txn);
assert(error == DB_NOTFOUND);
}
......@@ -1773,7 +1776,7 @@ void test_brt_cursor_get_both(int n) {
v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, 0);
error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, null_db, null_txn);
assert(error == 0);
#ifdef DB_CURRENT
init_dbt(&key); key.flags = DB_DBT_MALLOC;
......@@ -1794,7 +1797,7 @@ void test_brt_cursor_get_both(int n) {
v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, 0);
error = brt_cursor_get(cursor, &key, &val, DB_GET_BOTH, null_db, null_txn);
assert(error == DB_NOTFOUND);
}
......@@ -1878,7 +1881,7 @@ void test_large_kv(int bsize, int ksize, int vsize) {
fill_dbt(&key, k, ksize);
fill_dbt(&val, v, vsize);
r = brt_insert(t, &key, &val, 0);
r = brt_insert(t, &key, &val, 0, 0);
assert(r == 0);
toku_free(k);
......@@ -1955,7 +1958,7 @@ void test_brt_delete_present(int n) {
k = htonl(i); v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
r = brt_insert(t, &key, &val, 0);
r = brt_insert(t, &key, &val, 0, 0);
assert(r == 0);
}
......@@ -1984,7 +1987,7 @@ void test_brt_delete_present(int n) {
init_dbt(&key); key.flags = DB_DBT_MALLOC;
init_dbt(&val); val.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &key, &val, DB_FIRST, 0);
r = brt_cursor_get(cursor, &key, &val, DB_FIRST, null_db, null_txn);
assert(r != 0);
r = brt_cursor_close(cursor);
......@@ -2017,7 +2020,7 @@ void test_brt_delete_not_present(int n) {
k = htonl(i); v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
r = brt_insert(t, &key, &val, 0);
r = brt_insert(t, &key, &val, 0, 0);
assert(r == 0);
}
......@@ -2064,7 +2067,7 @@ void test_brt_delete_cursor_first(int n) {
k = htonl(i); v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
r = brt_insert(t, &key, &val, 0);
r = brt_insert(t, &key, &val, 0, 0);
assert(r == 0);
}
......@@ -2093,7 +2096,7 @@ void test_brt_delete_cursor_first(int n) {
init_dbt(&key); key.flags = DB_DBT_MALLOC;
init_dbt(&val); val.flags = DB_DBT_MALLOC;
r = brt_cursor_get(cursor, &key, &val, DB_FIRST, 0);
r = brt_cursor_get(cursor, &key, &val, DB_FIRST, null_db, null_txn);
assert(r == 0);
int vv;
assert(val.size == sizeof vv);
......
newbrt/brt.c
View file @ 24c21821
......@@ -79,8 +79,7 @@ void brtnode_flush_callback (CACHEFILE cachefile, diskoff nodename, void *brtnod
assert(brtnode->thisnodename==nodename);
//printf("%s:%d %p->mdict[0]=%p\n", __FILE__, __LINE__, brtnode, brtnode->mdicts[0]);
if (write_me) {
int r=serialize_brtnode_to(cachefile_fd(cachefile), brtnode->thisnodename, brtnode->nodesize, brtnode);
assert(r==0); // ?????
serialize_brtnode_to(cachefile_fd(cachefile), brtnode->thisnodename, brtnode->nodesize, brtnode);
}
//printf("%s:%d %p->mdict[0]=%p\n", __FILE__, __LINE__, brtnode, brtnode->mdicts[0]);
if (!keep_me) {
......@@ -520,14 +519,16 @@ void find_heaviest_data (BRTNODE node, int *childnum_ret, KVPAIR *pairs_ret, int
#endif
static int brtnode_put_cmd (BRT t, BRTNODE node, BRT_CMD *cmd,
int *did_split, BRTNODE *nodea, BRTNODE *nodeb,
DBT *split,
int debug);
int *did_split, BRTNODE *nodea, BRTNODE *nodeb,
DBT *split,
int debug,
TOKUTXN txn);
/* key is not in the hashtable in node. Either put the key-value pair in the child, or put it in the node. */
static int push_brt_cmd_down_only_if_it_wont_push_more_else_put_here (BRT t, BRTNODE node, BRTNODE child,
BRT_CMD *cmd,
int childnum_of_node) {
BRT_CMD *cmd,
int childnum_of_node,
TOKUTXN txn) {
assert(node->height>0); /* Not a leaf. */
DBT *k = cmd->u.id.key;
DBT *v = cmd->u.id.val;
......@@ -548,8 +549,9 @@ static int push_brt_cmd_down_only_if_it_wont_push_more_else_put_here (BRT t, BRT
init_dbt(&againk);
//printf("%s:%d hello!\n", __FILE__, __LINE__);
int r = brtnode_put_cmd(t, child, cmd,
&again_split, &againa, &againb, &againk,
0);
&again_split, &againa, &againb, &againk,
0,
txn);
if (r!=0) return r;
assert(again_split==0); /* I only did the insert if I knew it wouldn't push down, and hence wouldn't split. */
return r;
......@@ -560,17 +562,19 @@ static int push_brt_cmd_down_only_if_it_wont_push_more_else_put_here (BRT t, BRT
}
static int push_a_brt_cmd_down (BRT t, BRTNODE node, BRTNODE child, int childnum,
BRT_CMD *cmd,
int *child_did_split, BRTNODE *childa, BRTNODE *childb,
DBT *childsplitk) {
BRT_CMD *cmd,
int *child_did_split, BRTNODE *childa, BRTNODE *childb,
DBT *childsplitk,
TOKUTXN txn) {
//if (debug) printf("%s:%d %*sinserting down\n", __FILE__, __LINE__, debug, "");
//printf("%s:%d hello!\n", __FILE__, __LINE__);
assert(node->height>0);
{
int r = brtnode_put_cmd(t, child, cmd,
child_did_split, childa, childb, childsplitk,
0);
child_did_split, childa, childb, childsplitk,
0,
txn);
if (r!=0) return r;
}
......@@ -607,7 +611,8 @@ static int handle_split_of_child (BRT t, BRTNODE node, int childnum,
int *did_split, BRTNODE *nodea, BRTNODE *nodeb,
DBT *splitk,
void *app_private,
DB *db) {
DB *db,
TOKUTXN txn) {
assert(node->height>0);
assert(0 <= childnum && childnum < node->u.n.n_children);
HASHTABLE old_h = node->u.n.htables[childnum];
......@@ -667,9 +672,9 @@ static int handle_split_of_child (BRT t, BRTNODE node, int childnum,
BRT_CMD brtcmd;
brtcmd.type = type; brtcmd.u.id.key = &skd; brtcmd.u.id.val = &svd; brtcmd.u.id.db = db;
if (t->compare_fun(db, &skd, childsplitk)<=0) {
r=push_brt_cmd_down_only_if_it_wont_push_more_else_put_here(t, node, childa, &brtcmd, childnum);
r=push_brt_cmd_down_only_if_it_wont_push_more_else_put_here(t, node, childa, &brtcmd, childnum, txn);
} else {
r=push_brt_cmd_down_only_if_it_wont_push_more_else_put_here(t, node, childb, &brtcmd, childnum+1);
r=push_brt_cmd_down_only_if_it_wont_push_more_else_put_here(t, node, childb, &brtcmd, childnum+1, txn);
}
if (r!=0) return r;
}));
......@@ -708,11 +713,12 @@ static int handle_split_of_child (BRT t, BRTNODE node, int childnum,
}
static int push_some_brt_cmds_down (BRT t, BRTNODE node, int childnum,
int *did_split, BRTNODE *nodea, BRTNODE *nodeb,
DBT *splitk,
int debug,
void *app_private,
DB *db) {
int *did_split, BRTNODE *nodea, BRTNODE *nodeb,
DBT *splitk,
int debug,
void *app_private,
DB *db,
TOKUTXN txn) {
void *childnode_v;
BRTNODE child;
int r;
......@@ -762,9 +768,10 @@ static int push_some_brt_cmds_down (BRT t, BRTNODE node, int childnum,
if (debug) printf("%s:%d %*spush down %s\n", __FILE__, __LINE__, debug, "", (char*)key);
r = push_a_brt_cmd_down (t, node, child, childnum,
&brtcmd,
&child_did_split, &childa, &childb,
&childsplitk);
&brtcmd,
&child_did_split, &childa, &childb,
&childsplitk,
txn);
if (0){
unsigned int sum=0;
......@@ -782,7 +789,7 @@ static int push_some_brt_cmds_down (BRT t, BRTNODE node, int childnum,
r=handle_split_of_child (t, node, childnum,
childa, childb, &childsplitk,
did_split, nodea, nodeb, splitk,
app_private, db);
app_private, db, txn);
return r; /* Don't do any more pushing if the child splits. */
}
}
......@@ -800,7 +807,7 @@ int debugp1 (int debug) {
return debug ? debug+1 : 0;
}
static int brtnode_maybe_push_down(BRT t, BRTNODE node, int *did_split, BRTNODE *nodea, BRTNODE *nodeb, DBT *splitk, int debug, void *app_private, DB *db)
static int brtnode_maybe_push_down(BRT t, BRTNODE node, int *did_split, BRTNODE *nodea, BRTNODE *nodeb, DBT *splitk, int debug, void *app_private, DB *db, TOKUTXN txn)
/* If the buffer is too full, then push down. Possibly the child will split. That may make us split. */
{
assert(node->height>0);
......@@ -816,7 +823,7 @@ static int brtnode_maybe_push_down(BRT t, BRTNODE node, int *did_split, BRTNODE
find_heaviest_child(node, &childnum);
if (0) printf("%s:%d %*spush some down from %lld into %lld (child %d)\n", __FILE__, __LINE__, debug, "", node->thisnodename, node->u.n.children[childnum], childnum);
assert(node->u.n.children[childnum]!=0);
int r = push_some_brt_cmds_down(t, node, childnum, did_split, nodea, nodeb, splitk, debugp1(debug), app_private, db);
int r = push_some_brt_cmds_down(t, node, childnum, did_split, nodea, nodeb, splitk, debugp1(debug), app_private, db, txn);
if (r!=0) return r;
assert(*did_split==0 || *did_split==1);
if (debug) printf("%s:%d %*sdid push_some_brt_cmds_down did_split=%d\n", __FILE__, __LINE__, debug, "", *did_split);
......@@ -841,15 +848,16 @@ static int brtnode_maybe_push_down(BRT t, BRTNODE node, int *did_split, BRTNODE
#define INSERT_ALL_AT_ONCE
static int brt_leaf_put_cmd (BRT t, BRTNODE node, BRT_CMD *cmd,
int *did_split, BRTNODE *nodea, BRTNODE *nodeb, DBT *splitk,
int debug) {
int *did_split, BRTNODE *nodea, BRTNODE *nodeb, DBT *splitk,
int debug,
TOKUTXN txn) {
if (cmd->type == BRT_INSERT) {
DBT *k = cmd->u.id.key;
DBT *v = cmd->u.id.val;
DB *db = cmd->u.id.db;
#ifdef INSERT_ALL_AT_ONCE
int replaced_v_size;
enum pma_errors pma_status = pma_insert_or_replace(node->u.l.buffer, k, v, db, &replaced_v_size);
enum pma_errors pma_status = pma_insert_or_replace(node->u.l.buffer, k, v, &replaced_v_size, db, txn, node->thisnodename);
assert(pma_status==BRT_OK);
//printf("replaced_v_size=%d\n", replaced_v_size);
if (replaced_v_size>=0) {
......@@ -922,9 +930,10 @@ static unsigned int brtnode_which_child (BRTNODE node , DBT *k, BRT t, DB *db) {
static int brt_nonleaf_put_cmd (BRT t, BRTNODE node, BRT_CMD *cmd,
int *did_split, BRTNODE *nodea, BRTNODE *nodeb,
DBT *splitk,
int debug) {
int *did_split, BRTNODE *nodea, BRTNODE *nodeb,
DBT *splitk,
int debug,
TOKUTXN txn) {
bytevec olddata;
ITEMLEN olddatalen;
unsigned int childnum;
......@@ -956,14 +965,14 @@ static int brt_nonleaf_put_cmd (BRT t, BRTNODE node, BRT_CMD *cmd,
child_did_split = 0;
r = brtnode_put_cmd(t, child, cmd,
&child_did_split, &childa, &childb, &childsplitk, 0);
&child_did_split, &childa, &childb, &childsplitk, 0, txn);
assert(r == 0);
if (child_did_split) {
if (0) printf("brt_nonleaf_insert child_split %p\n", child);
r = handle_split_of_child(t, node, childnum,
childa, childb, &childsplitk,
did_split, nodea, nodeb, splitk,
k->app_private, db);
childa, childb, &childsplitk,
did_split, nodea, nodeb, splitk,
k->app_private, db, txn);
assert(r == 0);
} else {
r = cachetable_unpin_size(t->cf, child->thisnodename, child->dirty, brtnode_size(child));
......@@ -994,13 +1003,13 @@ static int brt_nonleaf_put_cmd (BRT t, BRTNODE node, BRT_CMD *cmd,
BRTNODE childa, childb;
DBT childsplitk;
int r = brtnode_put_cmd(t, child, cmd,
&child_did_split, &childa, &childb, &childsplitk, 0);
&child_did_split, &childa, &childb, &childsplitk, 0, txn);
if (r!=0) return r;
if (child_did_split) {
r=handle_split_of_child(t, node, childnum,
childa, childb, &childsplitk,
did_split, nodea, nodeb, splitk,
k->app_private, db);
k->app_private, db, txn);
if (r!=0) return r;
} else {
cachetable_unpin_size(t->cf, child->thisnodename, child->dirty, brtnode_size(child));
......@@ -1031,7 +1040,7 @@ static int brt_nonleaf_put_cmd (BRT t, BRTNODE node, BRT_CMD *cmd,
brtnode_set_dirty(node);
}
if (debug) printf("%s:%d %*sDoing maybe_push_down\n", __FILE__, __LINE__, debug, "");
int r = brtnode_maybe_push_down(t, node, did_split, nodea, nodeb, splitk, debugp1(debug), k->app_private, db);
int r = brtnode_maybe_push_down(t, node, did_split, nodea, nodeb, splitk, debugp1(debug), k->app_private, db, txn);
if (r!=0) return r;
if (debug) printf("%s:%d %*sDid maybe_push_down\n", __FILE__, __LINE__, debug, "");
if (*did_split) {
......@@ -1052,16 +1061,17 @@ static int brt_nonleaf_put_cmd (BRT t, BRTNODE node, BRT_CMD *cmd,
static int brtnode_put_cmd (BRT t, BRTNODE node, BRT_CMD *cmd,
int *did_split, BRTNODE *nodea, BRTNODE *nodeb, DBT *splitk,
int debug) {
int *did_split, BRTNODE *nodea, BRTNODE *nodeb, DBT *splitk,
int debug,
TOKUTXN txn) {
if (node->height==0) {
return brt_leaf_put_cmd(t, node, cmd,
did_split, nodea, nodeb, splitk,
debug);
did_split, nodea, nodeb, splitk,
debug, txn);
} else {
return brt_nonleaf_put_cmd(t, node, cmd,
did_split, nodea, nodeb, splitk,
debug);
did_split, nodea, nodeb, splitk,
debug, txn);
}
}
......@@ -1283,7 +1293,7 @@ int brt_init_new_root(BRT brt, BRTNODE nodea, BRTNODE nodeb, DBT splitk, CACHEKE
return 0;
}
int brt_root_put_cmd(BRT brt, BRT_CMD *cmd) {
int brt_root_put_cmd(BRT brt, BRT_CMD *cmd, TOKUTXN txn) {
void *node_v;
BRTNODE node;
CACHEKEY *rootp;
......@@ -1307,8 +1317,9 @@ int brt_root_put_cmd(BRT brt, BRT_CMD *cmd) {
if (debug) printf("%s:%d node inserting\n", __FILE__, __LINE__);
did_split = 0;
result = brtnode_put_cmd(brt, node, cmd,
&did_split, &nodea, &nodeb, &splitk,
debug);
&did_split, &nodea, &nodeb, &splitk,
debug,
txn);
if (debug) printf("%s:%d did_insert\n", __FILE__, __LINE__);
if (did_split) {
//printf("%s:%d did_split=%d nodeb=%p nodeb->thisnodename=%lld nodeb->nodesize=%d\n", __FILE__, __LINE__, did_split, nodeb, nodeb->thisnodename, nodeb->nodesize);
......@@ -1333,7 +1344,7 @@ int brt_root_put_cmd(BRT brt, BRT_CMD *cmd) {
return result;
}
int brt_insert (BRT brt, DBT *key, DBT *val, DB* db) {
int brt_insert (BRT brt, DBT *key, DBT *val, DB* db, TOKUTXN txn) {
int r;
BRT_CMD brtcmd;
......@@ -1341,7 +1352,7 @@ int brt_insert (BRT brt, DBT *key, DBT *val, DB* db) {
brtcmd.u.id.key = key;
brtcmd.u.id.val = val;
brtcmd.u.id.db = db;
r = brt_root_put_cmd(brt, &brtcmd);
r = brt_root_put_cmd(brt, &brtcmd, txn);
return r;
}
......@@ -1416,7 +1427,6 @@ int brt_lookup (BRT brt, DBT *k, DBT *v, DB *db) {
return 0;
}
int brt_delete(BRT brt, DBT *key, DB *db) {
int r;
BRT_CMD brtcmd;
......@@ -1428,7 +1438,7 @@ int brt_delete(BRT brt, DBT *key, DB *db) {
brtcmd.u.id.key = key;
brtcmd.u.id.val = &val;
brtcmd.u.id.db = db;
r = brt_root_put_cmd(brt, &brtcmd);
r = brt_root_put_cmd(brt, &brtcmd, 0);
return r;
}
......@@ -1631,7 +1641,7 @@ int brt_flush_debug = 0;
* then reflect the node split up the cursor path towards the tree root.
* If the root is reached then create a new root
*/
void brt_flush_child(BRT t, BRTNODE node, int childnum, BRT_CURSOR cursor) {
void brt_flush_child(BRT t, BRTNODE node, int childnum, BRT_CURSOR cursor, TOKUTXN txn) {
int r;
int child_did_split;
BRTNODE childa, childb;
......@@ -1644,7 +1654,7 @@ void brt_flush_child(BRT t, BRTNODE node, int childnum, BRT_CURSOR cursor) {
init_dbt(&child_splitk);
r = push_some_brt_cmds_down(t, node, childnum,
&child_did_split, &childa, &childb, &child_splitk, brt_flush_debug, 0, 0);
&child_did_split, &childa, &childb, &child_splitk, brt_flush_debug, 0, 0, txn);
assert(r == 0);
if (brt_flush_debug) {
printf("brt_flush_child done %lld %d\n", node->thisnodename, childnum);
......@@ -1676,9 +1686,9 @@ void brt_flush_child(BRT t, BRTNODE node, int childnum, BRT_CURSOR cursor) {
upnode = cursor->path[i];
childnum = cursor->pathcnum[i];
r = handle_split_of_child(t, upnode, childnum,
childa, childb, &child_splitk,
&child_did_split, &childa, &childb, &child_splitk,
0, 0);
childa, childb, &child_splitk,
&child_did_split, &childa, &childb, &child_splitk,
0, 0, txn);
assert(r == 0);
}
}
......@@ -1821,18 +1831,16 @@ void brt_cursor_nonleaf_expand(BRT_CURSOR cursor, BRT t __attribute__((unused)),
assert(node->height > 0);
#if 0
i = cursor->path_len - node->height - 1;
if (i < 0)
i = cursor->path_len - 1;
if (i >= 0 && cursor->path[i] == node) {
#else
// i = cursor->path_len - node->height - 1;
// if (i < 0)
// i = cursor->path_len - 1;
// if (i >= 0 && cursor->path[i] == node) {
// }
if (0) brt_cursor_print(cursor);
for (i = 0; i < cursor->path_len; i++)
if (cursor->path[i] == node)
break;
if (i < cursor->path_len) {
#endif
if (cursor->pathcnum[i] < childnum)
return;
if (cursor->pathcnum[i] > childnum) {
......@@ -1864,17 +1872,14 @@ void brt_cursor_nonleaf_split(BRT_CURSOR cursor, BRT t, BRTNODE oldnode, BRTNODE
int childnum;
assert(oldnode->height > 0 && left->height > 0 && right->height > 0);
#if 0
i = cursor->path_len - oldnode->height - 1;
if (i < 0)
i = cursor->path_len - 1;
if (i >= 0 && cursor->path[i] == oldnode) {
#else
// i = cursor->path_len - oldnode->height - 1;
// if (i < 0)
// i = cursor->path_len - 1;
// if (i >= 0 && cursor->path[i] == oldnode) {
for (i = 0; i < cursor->path_len; i++)
if (cursor->path[i] == oldnode)
break;
if (i < cursor->path_len) {
#endif
childnum = cursor->pathcnum[i];
brt_node_remove_cursor(oldnode, childnum, cursor);
if (childnum < left->u.n.n_children) {
......@@ -1885,7 +1890,7 @@ void brt_cursor_nonleaf_split(BRT_CURSOR cursor, BRT t, BRTNODE oldnode, BRTNODE
}
if (0) printf("brt_cursor_nonleaf_split %p oldnode %lld newnode %lld\n",
cursor, oldnode->thisnodename, newnode->thisnodename);
cursor, oldnode->thisnodename, newnode->thisnodename);
r = cachetable_unpin_size(t->cf, oldnode->thisnodename, oldnode->dirty, brtnode_size(oldnode));
assert(r == 0);
......@@ -1959,7 +1964,7 @@ void brt_cursor_print(BRT_CURSOR cursor) {
printf("\n");
}
int brtcurs_set_position_last (BRT_CURSOR cursor, diskoff off) {
int brtcurs_set_position_last (BRT_CURSOR cursor, diskoff off, TOKUTXN txn) {
BRT brt=cursor->brt;
void *node_v;
......@@ -1982,7 +1987,7 @@ int brtcurs_set_position_last (BRT_CURSOR cursor, diskoff off) {
cursor->pathcnum[cursor->path_len-1] = childnum;
brt_node_add_cursor(node, childnum, cursor);
if (node->u.n.n_bytes_in_hashtable[childnum] > 0) {
brt_flush_child(cursor->brt, node, childnum, cursor);
brt_flush_child(cursor->brt, node, childnum, cursor, txn);
/*
* the flush may have been partially successfull. it may have also
* changed the tree such that the current node have expanded or been
......@@ -1993,7 +1998,7 @@ int brtcurs_set_position_last (BRT_CURSOR cursor, diskoff off) {
brt_node_remove_cursor(node, childnum, cursor);
goto try_last_child;
}
r=brtcurs_set_position_last (cursor, node->u.n.children[childnum]);
r=brtcurs_set_position_last (cursor, node->u.n.children[childnum], txn);
if (r == 0)
return 0;
assert(node == cursor->path[cursor->path_len-1]);
......@@ -2020,7 +2025,7 @@ int brtcurs_set_position_last (BRT_CURSOR cursor, diskoff off) {
}
}
int brtcurs_set_position_first (BRT_CURSOR cursor, diskoff off) {
int brtcurs_set_position_first (BRT_CURSOR cursor, diskoff off, TOKUTXN txn) {
BRT brt=cursor->brt;
void *node_v;
......@@ -2043,7 +2048,7 @@ int brtcurs_set_position_first (BRT_CURSOR cursor, diskoff off) {
cursor->pathcnum[cursor->path_len-1] = childnum;
brt_node_add_cursor(node, childnum, cursor);
if (node->u.n.n_bytes_in_hashtable[childnum] > 0) {
brt_flush_child(cursor->brt, node, childnum, cursor);
brt_flush_child(cursor->brt, node, childnum, cursor, txn);
/*
* the flush may have been partially successfull. it may have also
* changed the tree such that the current node have expanded or been
......@@ -2054,7 +2059,7 @@ int brtcurs_set_position_first (BRT_CURSOR cursor, diskoff off) {
brt_node_remove_cursor(node, childnum, cursor);
goto try_first_child;
}
r=brtcurs_set_position_first (cursor, node->u.n.children[childnum]);
r=brtcurs_set_position_first (cursor, node->u.n.children[childnum], txn);
if (r == 0)
return r;
assert(node == cursor->path[cursor->path_len-1]);
......@@ -2082,7 +2087,7 @@ int brtcurs_set_position_first (BRT_CURSOR cursor, diskoff off) {
}
}
int brtcurs_set_position_next2(BRT_CURSOR cursor) {
int brtcurs_set_position_next2(BRT_CURSOR cursor, TOKUTXN txn) {
BRTNODE node;
int childnum;
int r;
......@@ -2113,11 +2118,11 @@ int brtcurs_set_position_next2(BRT_CURSOR cursor) {
more = node->u.n.n_bytes_in_hashtable[childnum];
if (more == 0)
break;
brt_flush_child(cursor->brt, node, childnum, cursor);
brt_flush_child(cursor->brt, node, childnum, cursor, txn);
node = cursor->path[cursor->path_len-1];
childnum = cursor->pathcnum[cursor->path_len-1];
}
r = brtcurs_set_position_first(cursor, node->u.n.children[childnum]);
r = brtcurs_set_position_first(cursor, node->u.n.children[childnum], txn);
if (r == 0)
return 0;
assert(node == cursor->path[cursor->path_len-1]);
......@@ -2125,11 +2130,11 @@ int brtcurs_set_position_next2(BRT_CURSOR cursor) {
childnum += 1;
}
return brtcurs_set_position_next2(cursor);
return brtcurs_set_position_next2(cursor, txn);
}
/* requires that the cursor is initialized. */
int brtcurs_set_position_next (BRT_CURSOR cursor) {
int brtcurs_set_position_next (BRT_CURSOR cursor, TOKUTXN txn) {
int r = pma_cursor_set_position_next(cursor->pmacurs);
if (r==DB_NOTFOUND) {
/* We fell off the end of the pma. */
......@@ -2137,12 +2142,12 @@ int brtcurs_set_position_next (BRT_CURSOR cursor) {
/* Part of the trickyness is we need to leave the cursor pointing at the current (possibly deleted) value if there is no next value. */
r = pma_cursor_free(&cursor->pmacurs);
assert(r == 0);
return brtcurs_set_position_next2(cursor);
return brtcurs_set_position_next2(cursor, txn);
}
return 0;
}
int brtcurs_set_position_prev2(BRT_CURSOR cursor) {
int brtcurs_set_position_prev2(BRT_CURSOR cursor, TOKUTXN txn) {
BRTNODE node;
int childnum;
int r;
......@@ -2173,11 +2178,11 @@ int brtcurs_set_position_prev2(BRT_CURSOR cursor) {
more = node->u.n.n_bytes_in_hashtable[childnum];
if (more == 0)
break;
brt_flush_child(cursor->brt, node, childnum, cursor);
brt_flush_child(cursor->brt, node, childnum, cursor, txn);
node = cursor->path[cursor->path_len-1];
childnum = cursor->pathcnum[cursor->path_len-1];
}
r = brtcurs_set_position_last(cursor, node->u.n.children[childnum]);
r = brtcurs_set_position_last(cursor, node->u.n.children[childnum], txn);
if (r == 0)
return 0;
assert(node == cursor->path[cursor->path_len-1]);
......@@ -2185,22 +2190,22 @@ int brtcurs_set_position_prev2(BRT_CURSOR cursor) {
childnum -= 1;
}
return brtcurs_set_position_prev2(cursor);
return brtcurs_set_position_prev2(cursor, txn);
}
int brtcurs_set_position_prev (BRT_CURSOR cursor) {
int brtcurs_set_position_prev (BRT_CURSOR cursor, TOKUTXN txn) {
int r = pma_cursor_set_position_prev(cursor->pmacurs);
if (r==DB_NOTFOUND) {
if (cursor->path_len==1)
return DB_NOTFOUND;
r = pma_cursor_free(&cursor->pmacurs);
assert(r == 0);
return brtcurs_set_position_prev2(cursor);
return brtcurs_set_position_prev2(cursor, txn);
}
return 0;
}
int brtcurs_set_key(BRT_CURSOR cursor, diskoff off, DBT *key, DBT *val, int flag, DB *db) {
int brtcurs_set_key(BRT_CURSOR cursor, diskoff off, DBT *key, DBT *val, int flag, DB *db, TOKUTXN txn) {
BRT brt = cursor->brt;
void *node_v;
int r;
......@@ -2221,7 +2226,7 @@ int brtcurs_set_key(BRT_CURSOR cursor, diskoff off, DBT *key, DBT *val, int flag
brt_node_add_cursor(node, childnum, cursor);
int more = node->u.n.n_bytes_in_hashtable[childnum];
if (more > 0) {
brt_flush_child(cursor->brt, node, childnum, cursor);
brt_flush_child(cursor->brt, node, childnum, cursor, txn);
node = cursor->path[cursor->path_len-1];
childnum = cursor->pathcnum[cursor->path_len-1];
brt_node_remove_cursor(node, childnum, cursor);
......@@ -2230,7 +2235,7 @@ int brtcurs_set_key(BRT_CURSOR cursor, diskoff off, DBT *key, DBT *val, int flag
}
break;
}
r = brtcurs_set_key(cursor, node->u.n.children[childnum], key, val, flag, db);
r = brtcurs_set_key(cursor, node->u.n.children[childnum], key, val, flag, db, txn);
if (r != 0)
brt_node_remove_cursor(node, childnum, cursor);
} else {
......@@ -2260,7 +2265,7 @@ int brtcurs_set_key(BRT_CURSOR cursor, diskoff off, DBT *key, DBT *val, int flag
return r;
}
int brtcurs_set_range(BRT_CURSOR cursor, diskoff off, DBT *key, DB *db) {
int brtcurs_set_range(BRT_CURSOR cursor, diskoff off, DBT *key, DB *db, TOKUTXN txn) {
BRT brt = cursor->brt;
void *node_v;
int r;
......@@ -2283,7 +2288,7 @@ int brtcurs_set_range(BRT_CURSOR cursor, diskoff off, DBT *key, DB *db) {
brt_node_add_cursor(node, childnum, cursor);
int more = node->u.n.n_bytes_in_hashtable[childnum];
if (more > 0) {
brt_flush_child(cursor->brt, node, childnum, cursor);
brt_flush_child(cursor->brt, node, childnum, cursor, txn);
node = cursor->path[cursor->path_len-1];
childnum = cursor->pathcnum[cursor->path_len-1];
brt_node_remove_cursor(node, childnum, cursor);
......@@ -2291,7 +2296,7 @@ int brtcurs_set_range(BRT_CURSOR cursor, diskoff off, DBT *key, DB *db) {
}
break;
}
r = brtcurs_set_range(cursor, node->u.n.children[childnum], key, db);
r = brtcurs_set_range(cursor, node->u.n.children[childnum], key, db, txn);
if (r != 0) {
node = cursor->path[cursor->path_len-1];
childnum = cursor->pathcnum[cursor->path_len-1];
......@@ -2358,7 +2363,7 @@ static void assert_cursor_path(BRT_CURSOR cursor) {
assert(node->height == 0);
}
int brt_cursor_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int flags, DB *db) {
int brt_cursor_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int flags, DB *db, TOKUTXN txn) {
int do_rmw=0;
int r;
CACHEKEY *rootp;
......@@ -2380,7 +2385,7 @@ static void assert_cursor_path(BRT_CURSOR cursor) {
do_db_last:
r=unpin_cursor(cursor); if (r!=0) goto died0;
assert(cursor->pmacurs == 0);
r=brtcurs_set_position_last(cursor, *rootp); if (r!=0) goto died0;
r=brtcurs_set_position_last(cursor, *rootp, txn); if (r!=0) goto died0;
r=pma_cursor_get_current(cursor->pmacurs, kbt, vbt);
if (r == 0) assert_cursor_path(cursor);
break;
......@@ -2388,28 +2393,28 @@ static void assert_cursor_path(BRT_CURSOR cursor) {
do_db_first:
r=unpin_cursor(cursor); if (r!=0) goto died0;
assert(cursor->pmacurs == 0);
r=brtcurs_set_position_first(cursor, *rootp); if (r!=0) goto died0;
r=brtcurs_set_position_first(cursor, *rootp, txn); if (r!=0) goto died0;
r=pma_cursor_get_current(cursor->pmacurs, kbt, vbt);
if (r == 0) assert_cursor_path(cursor);
break;
case DB_NEXT:
if (cursor->path_len<=0)
goto do_db_first;
r=brtcurs_set_position_next(cursor); if (r!=0) goto died0;
r=brtcurs_set_position_next(cursor, txn); if (r!=0) goto died0;
r=pma_cursor_get_current(cursor->pmacurs, kbt, vbt); if (r!=0) goto died0;
if (r == 0) assert_cursor_path(cursor);
break;
case DB_PREV:
if (cursor->path_len<= 0)
goto do_db_last;
r = brtcurs_set_position_prev(cursor); if (r!=0) goto died0;
r = brtcurs_set_position_prev(cursor, txn); if (r!=0) goto died0;
r = pma_cursor_get_current(cursor->pmacurs, kbt, vbt); if (r!=0) goto died0;
if (r == 0) assert_cursor_path(cursor);
break;
case DB_SET:
r = unpin_cursor(cursor);
assert(r == 0);
r = brtcurs_set_key(cursor, *rootp, kbt, vbt, DB_SET, db);
r = brtcurs_set_key(cursor, *rootp, kbt, vbt, DB_SET, db, txn);
if (r != 0) goto died0;
r = pma_cursor_get_current(cursor->pmacurs, kbt, vbt);
if (r != 0) goto died0;
......@@ -2417,13 +2422,13 @@ static void assert_cursor_path(BRT_CURSOR cursor) {
case DB_GET_BOTH:
r = unpin_cursor(cursor);
assert(r == 0);
r = brtcurs_set_key(cursor, *rootp, kbt, vbt, DB_GET_BOTH, db);
r = brtcurs_set_key(cursor, *rootp, kbt, vbt, DB_GET_BOTH, db, txn);
if (r != 0) goto died0;
break;
case DB_SET_RANGE:
r = unpin_cursor(cursor);
assert(r == 0);
r = brtcurs_set_range(cursor, *rootp, kbt, db);
r = brtcurs_set_range(cursor, *rootp, kbt, db, txn);
if (r != 0) goto died0;
r = pma_cursor_get_current(cursor->pmacurs, kbt, vbt);
if (r != 0) goto died0;
......
newbrt/brt.h
View file @ 24c21821
......@@ -8,11 +8,12 @@
#include "ybt.h"
#include "../include/db.h"
#include "cachetable.h"
#include "log.h"
typedef struct brt *BRT;
int open_brt (const char *fname, const char *dbname, int is_create, BRT *, int nodesize, CACHETABLE, int(*)(DB*,const DBT*,const DBT*));
//int brt_create (BRT **, int nodesize, int n_nodes_in_cache); /* the nodesize and n_nodes in cache really should be separately configured. */
//int brt_open (BRT *, char *fname, char *dbname);
int brt_insert (BRT brt, DBT *k, DBT *v, DB*db);
int brt_insert (BRT, DBT *, DBT *, DB*, TOKUTXN);
int brt_lookup (BRT brt, DBT *k, DBT *v, DB*db);
int brt_delete (BRT brt, DBT *k, DB *db);
int close_brt (BRT);
......@@ -35,7 +36,7 @@ int show_brt_blocknumbers(BRT);
typedef struct brt_cursor *BRT_CURSOR;
int brt_cursor (BRT, BRT_CURSOR*);
int brt_cursor_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int brtc_flags, DB *db);
int brt_cursor_get (BRT_CURSOR cursor, DBT *kbt, DBT *vbt, int brtc_flags, DB *db, TOKUTXN);
int brt_cursor_delete(BRT_CURSOR cursor, int flags);
int brt_cursor_close (BRT_CURSOR curs);
......
newbrt/brttypes.h
View file @ 24c21821
......@@ -5,4 +5,8 @@
typedef unsigned int ITEMLEN;
typedef const void *bytevec;
//typedef const void *bytevec;
typedef long long diskoff; /* Offset in a disk. -1 is the NULL pointer. */
typedef long long TXNID;
#endif
newbrt/kv-pair.h
View file @ 24c21821
#ifndef KV_PAIR_H
#define KV_PAIR_H
#include "memory.h"
#include <string.h>
/*
* the key value pair contains a key and a value in a contiguous space. the
* key is right after the length fields and the value is right after the key.
......@@ -40,6 +46,10 @@ static inline void kv_pair_free(struct kv_pair *pair) {
static inline void *kv_pair_key(struct kv_pair *pair) {
return pair->key;
}
static inline const void *kv_pair_key_const(const struct kv_pair *pair) {
return pair->key;
}
static inline int kv_pair_keylen(struct kv_pair *pair) {
return pair->keylen;
......@@ -48,6 +58,9 @@ static inline int kv_pair_keylen(struct kv_pair *pair) {
static inline void *kv_pair_val(struct kv_pair *pair) {
return pair->key + pair->keylen;
}
static inline const void *kv_pair_val_const(const struct kv_pair *pair) {
return pair->key + pair->keylen;
}
static inline int kv_pair_vallen(struct kv_pair *pair) {
return pair->vallen;
......@@ -79,3 +92,4 @@ struct kv_pair_tag {
struct kv_pair *pair;
int oldtag, newtag;
};
#endif
newbrt/log-internal.h
View file @ 24c21821
#include "yerror.h"
#include <stdio.h>
#include "log.h"
#include <sys/types.h>
#define LOGGER_BUF_SIZE (1<<20)
typedef struct tokulogger *TOKULOGGER;
struct tokulogger {
enum typ_tag tag;
char *directory;
......@@ -14,3 +16,10 @@ struct tokulogger {
};
int tokulogger_find_next_unused_log_file(const char *directory, long long *result);
enum { LT_INSERT_WITH_NO_OVERWRITE = 'I', LT_DELETE = 'D', LT_COMMIT = 'C' };
struct tokutxn {
u_int64_t txnid64;
TOKULOGGER logger;
};
newbrt/log.c
View file @ 24c21821
#include "brttypes.h"
#include "log-internal.h"
#include "wbuf.h"
#include "memory.h"
#include <dirent.h>
#include <errno.h>
......@@ -22,7 +24,8 @@ int tokulogger_find_next_unused_log_file(const char *directory, long long *resul
if (r==1 && thisl>max) max=thisl;
}
*result=max+1;
return 0;
int r = closedir(d);
return r;
}
int tokulogger_create_and_open_logger (const char *directory, TOKULOGGER *resultp) {
......@@ -37,20 +40,22 @@ int tokulogger_create_and_open_logger (const char *directory, TOKULOGGER *result
return nexti;
}
result->directory = toku_strdup(directory);
if (result->directory!=0) goto died0;
if (result->directory==0) goto died0;
result->fd = -1;
result->next_log_file_number = nexti;
result->n_in_buf = 0;
*resultp=result;
return 0;
return tokulogger_log_bytes(result, 0, "");
}
int tokulogger_log_bytes(TOKULOGGER logger, int nbytes, char *bytes) {
int tokulogger_log_bytes(TOKULOGGER logger, int nbytes, void *bytes) {
int r;
printf("%s:%d logging %d bytes\n", __FILE__, __LINE__, nbytes);
if (logger->fd==-1) {
int fnamelen = strlen(logger->directory)+50;
char fname[fnamelen];
snprintf(fname, fnamelen, "%s/log%012llu.tokulog", logger->directory, logger->next_log_file_number);
printf("%s:%d creat(%s, ...)\n", __FILE__, __LINE__, fname);
logger->fd = creat(fname, O_EXCL | 0700);
if (logger->fd==-1) return errno;
}
......@@ -78,8 +83,6 @@ int tokulogger_log_bytes(TOKULOGGER logger, int nbytes, char *bytes) {
return 0;
}
enum { LT_INSERT_WITH_NO_OVERWITE = 'I' };
// Log an insertion of a key-value pair into a particular node of the tree.
int tokulogger_log_brt_insert_with_no_overwrite (TOKULOGGER logger,
TXNID txnid,
......@@ -89,19 +92,34 @@ int tokulogger_log_brt_insert_with_no_overwrite (TOKULOGGER logger,
unsigned char *val,
int vallen) {
int buflen=30+keylen+vallen;
char buf[buflen];
WBUF wbuf;
int r;
r = wbuf_create(&wbuf, buf, buflen) ;
if (r!=0) return r;
wbuf_byte(&wbuf, LT_INSERT_WITH_NO_OVERWRITE);
wbuf_txnid(&wbuf, txnind);
unsigned char buf[buflen];
struct wbuf wbuf;
wbuf_init(&wbuf, buf, buflen) ;
wbuf_char(&wbuf, LT_INSERT_WITH_NO_OVERWRITE);
wbuf_txnid(&wbuf, txnid);
wbuf_diskoff(&wbuf, diskoff);
wbuf_bytes(&wbuf, key, keylen);
wbuf_bytes(&wbuf, val, vallen);
return tokulogger_log_wbuf(logger, &wbuf);
return tokulogger_log_bytes(logger, wbuf.ndone, wbuf.buf);
}
int tokulogger_log_close(TOKULOGGER *loggerp) {
TOKULOGGER logger = *loggerp;
int r = 0;
if (logger->fd!=-1) {
printf("%s:%d n_in_buf=%d\n", __FILE__, __LINE__, logger->n_in_buf);
if (logger->n_in_buf>0) {
r = write(logger->fd, logger->buf, logger->n_in_buf);
if (r==-1) return errno;
}
r = close(logger->fd);
}
toku_free(logger->directory);
toku_free(logger);
*loggerp=0;
return r;
}
#if 0
int tokulogger_log_brt_remove (TOKULOGGER logger,
TXNID txnid,
diskoff diskoff,
......@@ -109,5 +127,58 @@ int tokulogger_log_brt_remove (TOKULOGGER logger,
int keylen,
unsigned char *val,
int vallen) {
n
}
#endif
int tokulogger_log_phys_add_or_delete_in_leaf (TOKUTXN txn, diskoff diskoff, int is_add, const struct kv_pair *pair) {
if (txn==0) return 0;
int keylen = pair->keylen;
int vallen = pair->vallen;
int buflen=30+keylen+vallen;
unsigned char buf[buflen];
struct wbuf wbuf;
wbuf_init(&wbuf, buf, buflen) ;
wbuf_char(&wbuf, is_add ? LT_INSERT_WITH_NO_OVERWRITE : LT_DELETE);
wbuf_txnid(&wbuf, txn->txnid64);
wbuf_diskoff(&wbuf, diskoff);
wbuf_bytes(&wbuf, kv_pair_key_const(pair), keylen);
wbuf_bytes(&wbuf, kv_pair_val_const(pair), vallen);
return tokulogger_log_bytes(txn->logger, wbuf.ndone, wbuf.buf);
}
int tokulogger_fsync (TOKULOGGER logger) {
if (logger->n_in_buf>0) {
int r = write(logger->fd, logger->buf, logger->n_in_buf);
if (r==-1) return errno;
logger->n_in_buf=0;
}
{
int r = fsync(logger->fd);
if (r!=0) return errno;
}
return 0;
}
int tokulogger_log_commit (TOKUTXN txn) {
struct wbuf wbuf;
int buflen =30;
unsigned char buf[buflen];
wbuf_init(&wbuf, buf, buflen);
wbuf_char(&wbuf, LT_COMMIT);
wbuf_txnid(&wbuf, txn->txnid64);
int r = tokulogger_log_bytes(txn->logger, wbuf.ndone, wbuf.buf);
if (r!=0) return r;
return tokulogger_fsync(txn->logger);
}
int tokutxn_begin (TOKUTXN *tokutxn, TXNID txnid64, TOKULOGGER logger) {
TAGMALLOC(TOKUTXN, result);
if (result==0) return errno;
result->txnid64 = txnid64;
result->logger = logger;
*tokutxn = result;
return 0;
}
newbrt/log.h 0 → 100644
View file @ 24c21821
#ifndef TOKULOGGGER_H
#define TOKULOGGGER_H
#include "kv-pair.h"
typedef struct tokulogger *TOKULOGGER;
typedef struct tokutxn *TOKUTXN;
int tokulogger_create_and_open_logger (const char *directory, TOKULOGGER *resultp);
int tokulogger_log_bytes(TOKULOGGER logger, int nbytes, void *bytes);
int tokulogger_log_close(TOKULOGGER *logger);
int tokulogger_log_phys_add_or_delete_in_leaf (TOKUTXN txn, diskoff diskoff, int is_add, const struct kv_pair *pair);
int tokulogger_log_commit (TOKUTXN txn);
int tokutxn_begin (TOKUTXN *, TXNID txnid64, TOKULOGGER logger);
#endif
newbrt/logdump.c 0 → 100644
View file @ 24c21821
/* Dump the log from stdin to stdout. */
#include <stdio.h>
#include "brttypes.h"
#include "log-internal.h"
#include <sys/types.h>
#include <ctype.h>
#include <stdlib.h>
u_int32_t get_uint32 (void) {
u_int32_t a = getchar();
u_int32_t b = getchar();
u_int32_t c = getchar();
u_int32_t d = getchar();
return (a<<24)|(b<<16)|(c<<8)|d;
}
u_int64_t get_uint64 (void) {
u_int32_t hi = get_uint32();
u_int32_t lo = get_uint32();
return ((((long long)hi) << 32)
|
lo);
}
void transcribe_txnid (void) {
long long value = get_uint64();
printf(" txnid=%lld", value);
}
void transcribe_diskoff (void) {
long long value = get_uint64();
printf(" diskoff=%lld", value);
}
void transcribe_key_or_data (char *what) {
u_int32_t l = get_uint32();
unsigned int i;
printf(" %s(%d):\"", what, l);
for (i=0; i<l; i++) {
u_int32_t c = getchar();
if (c=='\\') printf("\\\\");
else if (c=='\n') printf("\\n");
else if (c==' ') printf("\\ ");
else if (c=='"') printf("\"\"");
else if (isprint(c)) printf("%c", c);
else printf("\\%02x", c);
}
printf("\"");
}
int main (int argc __attribute__((__unused__)), char *argv[] __attribute__((__unused__))) {
int cmd;
while ((cmd=getchar())!=EOF) {
switch (cmd) {
case LT_INSERT_WITH_NO_OVERWRITE:
printf("INSERT_WITH_NO_OVERWRITE:");
transcribe_txnid();
transcribe_diskoff();
transcribe_key_or_data("key");
transcribe_key_or_data("data");
printf("\n");
break;
case LT_DELETE:
printf("DELETE:");
transcribe_txnid();
transcribe_diskoff();
transcribe_key_or_data("key");
transcribe_key_or_data("data");
printf("\n");
break;
case LT_COMMIT:
printf("COMMIT:");
transcribe_txnid();
printf("\n");
break;
default:
printf("Huh?");
abort();
}
}
return 0;
}
newbrt/memory.h
View file @ 24c21821
#ifndef MEMORY_H
#define MEMORY_H
//#include <stdlib.h>
/* Tokutek memory allocation functions and macros.
......@@ -66,3 +69,4 @@ int get_n_items_malloced(void); /* How many items are malloc'd but not free'd.
void print_malloced_items(void); /* Try to print some malloced-but-not-freed items. May be a noop. */
void malloc_report (void); /* report on statistics about number of mallocs. Maybe a no-op. */
#endif
newbrt/pma-test.c
View file @ 24c21821
......@@ -10,6 +10,12 @@
#include "kv-pair.h"
#include "pma-internal.h"
TOKUTXN const null_txn = 0;
DB * const null_db = 0;
const diskoff null_diskoff = -1;
#define NULL_ARGS null_db, null_txn, null_diskoff
static void test_make_space_at (void) {
PMA pma;
char *key;
......@@ -256,7 +262,7 @@ static void test_pma_random_pick (void) {
assert(r==0);
r = pma_random_pick(pma, &key, &keylen, &val, &vallen);
assert(r==DB_NOTFOUND);
r = pma_insert(pma, fill_dbt(&k, "hello", 6), fill_dbt(&v, "there", 6), 0);
r = pma_insert(pma, fill_dbt(&k, "hello", 6), fill_dbt(&v, "there", 6), NULL_ARGS);
assert(r==BRT_OK);
r = pma_random_pick(pma, &key, &keylen, &val, &vallen);
assert(r==0);
......@@ -271,7 +277,7 @@ static void test_pma_random_pick (void) {
r = pma_random_pick(pma, &key, &keylen, &val, &vallen);
assert(r==DB_NOTFOUND);
r = pma_insert(pma, fill_dbt(&k, "hello", 6), fill_dbt(&v, "there", 6), 0);
r = pma_insert(pma, fill_dbt(&k, "hello", 6), fill_dbt(&v, "there", 6), NULL_ARGS);
assert(r==BRT_OK);
......@@ -281,13 +287,13 @@ static void test_pma_random_pick (void) {
assert(strcmp(key,"hello")==0);
assert(strcmp(val,"there")==0);
r = pma_insert(pma, fill_dbt(&k, "aaa", 4), fill_dbt(&v, "athere", 7), 0); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aab", 4), fill_dbt(&v, "bthere", 7), 0); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aac", 4), fill_dbt(&v, "cthere", 7), 0); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aad", 4), fill_dbt(&v, "dthere", 7), 0); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aae", 4), fill_dbt(&v, "ethere", 7), 0); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aaf", 4), fill_dbt(&v, "fthere", 7), 0); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aag", 4), fill_dbt(&v, "gthere", 7), 0); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aaa", 4), fill_dbt(&v, "athere", 7), NULL_ARGS); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aab", 4), fill_dbt(&v, "bthere", 7), NULL_ARGS); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aac", 4), fill_dbt(&v, "cthere", 7), NULL_ARGS); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aad", 4), fill_dbt(&v, "dthere", 7), NULL_ARGS); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aae", 4), fill_dbt(&v, "ethere", 7), NULL_ARGS); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aaf", 4), fill_dbt(&v, "fthere", 7), NULL_ARGS); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "aag", 4), fill_dbt(&v, "gthere", 7), NULL_ARGS); assert(r==BRT_OK);
r = pma_delete(pma, fill_dbt(&k, "aaa", 4), 0); assert(r==BRT_OK);
r = pma_delete(pma, fill_dbt(&k, "aab", 4), 0); assert(r==BRT_OK);
r = pma_delete(pma, fill_dbt(&k, "aac", 4), 0); assert(r==BRT_OK);
......@@ -313,7 +319,7 @@ static void test_find_insert (void) {
r=pma_lookup(pma, fill_dbt(&k, "aaa", 3), &v, 0);
assert(r==DB_NOTFOUND);
r=pma_insert(pma, fill_dbt(&k, "aaa", 3), fill_dbt(&v, "aaadata", 7), 0);
r=pma_insert(pma, fill_dbt(&k, "aaa", 3), fill_dbt(&v, "aaadata", 7), NULL_ARGS);
assert(r==BRT_OK);
init_dbt(&v);
......@@ -323,7 +329,7 @@ static void test_find_insert (void) {
assert(keycompare(v.data,v.size,"aaadata", 7)==0);
//toku_free(v.data); v.data=0;
r=pma_insert(pma, fill_dbt(&k, "bbb", 4), fill_dbt(&v, "bbbdata", 8), 0);
r=pma_insert(pma, fill_dbt(&k, "bbb", 4), fill_dbt(&v, "bbbdata", 8), NULL_ARGS);
assert(r==BRT_OK);
init_dbt(&v);
......@@ -338,7 +344,7 @@ static void test_find_insert (void) {
assert((unsigned long)pma->pairs[pma_index_limit(pma)]==0xdeadbeefL);
r=pma_insert(pma, fill_dbt(&k, "00000", 6), fill_dbt(&v, "d0", 3), 0);
r=pma_insert(pma, fill_dbt(&k, "00000", 6), fill_dbt(&v, "d0", 3), NULL_ARGS);
assert(r==BRT_OK);
assert((unsigned long)pma->pairs[pma_index_limit(pma)]==0xdeadbeefL);
......@@ -354,7 +360,7 @@ static void test_find_insert (void) {
snprintf(string,10,"%05d",i);
snprintf(dstring,10,"d%d", i);
printf("Inserting %d: string=%s dstring=%s\n", i, string, dstring);
r=pma_insert(pma, fill_dbt(&k, string, strlen(string)+1), fill_dbt(&v, dstring, strlen(dstring)+1), 0);
r=pma_insert(pma, fill_dbt(&k, string, strlen(string)+1), fill_dbt(&v, dstring, strlen(dstring)+1), NULL_ARGS);
assert(r==BRT_OK);
}
}
......@@ -382,11 +388,11 @@ static void test_pma_iterate (void) {
int r;
DBT k,v;
pma_create(&pma, default_compare_fun);
r=pma_insert(pma, fill_dbt(&k, "42", 3), fill_dbt(&v, "-19", 4), 0);
r=pma_insert(pma, fill_dbt(&k, "42", 3), fill_dbt(&v, "-19", 4), NULL_ARGS);
assert(r==BRT_OK);
test_pma_iterate_internal(pma, 42, -19);
r=pma_insert(pma, fill_dbt(&k, "12", 3), fill_dbt(&v, "-100", 5), 0);
r=pma_insert(pma, fill_dbt(&k, "12", 3), fill_dbt(&v, "-100", 5), NULL_ARGS);
assert(r==BRT_OK);
test_pma_iterate_internal(pma, 42+12, -19-100);
r=pma_free(&pma); assert(r==0); assert(pma==0);
......@@ -400,9 +406,9 @@ static void test_pma_iterate2 (void) {
DBT k,v;
r=pma_create(&pma0, default_compare_fun); assert(r==0);
r=pma_create(&pma1, default_compare_fun); assert(r==0);
pma_insert(pma0, fill_dbt(&k, "a", 2), fill_dbt(&v, "aval", 5), 0);
pma_insert(pma0, fill_dbt(&k, "b", 2), fill_dbt(&v, "bval", 5), 0);
pma_insert(pma1, fill_dbt(&k, "x", 2), fill_dbt(&v, "xval", 5), 0);
pma_insert(pma0, fill_dbt(&k, "a", 2), fill_dbt(&v, "aval", 5), NULL_ARGS);
pma_insert(pma0, fill_dbt(&k, "b", 2), fill_dbt(&v, "bval", 5), NULL_ARGS);
pma_insert(pma1, fill_dbt(&k, "x", 2), fill_dbt(&v, "xval", 5), NULL_ARGS);
PMA_ITERATE(pma0,kv __attribute__((__unused__)),kl,dv __attribute__((__unused__)),dl, (n_items++,sum+=kl+dl));
PMA_ITERATE(pma1,kv __attribute__((__unused__)),kl,dv __attribute__((__unused__)), dl, (n_items++,sum+=kl+dl));
assert(sum==21);
......@@ -479,9 +485,9 @@ void test_pma_cursor_3 (void) {
DBT key,val;
DBT k,v;
r=pma_create(&pma, default_compare_fun); assert(r==0);
r=pma_insert(pma, fill_dbt(&k, "x", 2), fill_dbt(&v, "xx", 3), 0); assert(r==BRT_OK);
r=pma_insert(pma, fill_dbt(&k, "m", 2), fill_dbt(&v, "mm", 3), 0); assert(r==BRT_OK);
r=pma_insert(pma, fill_dbt(&k, "aa", 3), fill_dbt(&v,"a", 2), 0); assert(r==BRT_OK);
r=pma_insert(pma, fill_dbt(&k, "x", 2), fill_dbt(&v, "xx", 3), NULL_ARGS); assert(r==BRT_OK);
r=pma_insert(pma, fill_dbt(&k, "m", 2), fill_dbt(&v, "mm", 3), NULL_ARGS); assert(r==BRT_OK);
r=pma_insert(pma, fill_dbt(&k, "aa", 3), fill_dbt(&v,"a", 2), NULL_ARGS); assert(r==BRT_OK);
init_dbt(&key); key.flags=DB_DBT_REALLOC;
init_dbt(&val); val.flags=DB_DBT_REALLOC;
r=pma_cursor(pma, &c); assert(r==0); assert(c!=0);
......@@ -552,7 +558,7 @@ void test_pma_cursor_4 (void) {
v = i;
fill_dbt(&dbtv, &v, sizeof v);
error = pma_insert(pma, &dbtk, &dbtv, 0);
error = pma_insert(pma, &dbtk, &dbtv, NULL_ARGS);
assert(error == BRT_OK);
}
assert(pma_n_entries(pma) == 4);
......@@ -588,7 +594,7 @@ void test_pma_cursor_4 (void) {
v = i;
fill_dbt(&dbtv, &v, sizeof v);
error = pma_insert(pma, &dbtk, &dbtv, 0);
error = pma_insert(pma, &dbtk, &dbtv, NULL_ARGS);
assert(error == BRT_OK);
}
assert(pma_n_entries(pma) == 8);
......@@ -625,7 +631,7 @@ void test_pma_cursor_delete(int n) {
k = i; v = -i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = pma_insert(pma, &key, &val, 0);
error = pma_insert(pma, &key, &val, NULL_ARGS);
assert(error == 0);
}
......@@ -725,10 +731,10 @@ void test_pma_compare_fun (int wrong_endian_p) {
int i;
DBT k,v;
r = pma_create(&pma, wrong_endian_p ? wrong_endian_compare_fun : default_compare_fun); assert(r==0);
r = pma_insert(pma, fill_dbt(&k, "10", 3), fill_dbt(&v, "10v", 4), 0); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "00", 3), fill_dbt(&v, "00v", 4), 0); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "01", 3), fill_dbt(&v, "01v", 4), 0); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "11", 3), fill_dbt(&v, "11v", 4), 0); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "10", 3), fill_dbt(&v, "10v", 4), NULL_ARGS); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "00", 3), fill_dbt(&v, "00v", 4), NULL_ARGS); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "01", 3), fill_dbt(&v, "01v", 4), NULL_ARGS); assert(r==BRT_OK);
r = pma_insert(pma, fill_dbt(&k, "11", 3), fill_dbt(&v, "11v", 4), NULL_ARGS); assert(r==BRT_OK);
init_dbt(&key); key.flags=DB_DBT_REALLOC;
init_dbt(&val); val.flags=DB_DBT_REALLOC;
r=pma_cursor(pma, &c); assert(r==0); assert(c!=0);
......@@ -780,7 +786,7 @@ void test_pma_split_n(int n) {
v = i;
fill_dbt(&dbtv, &v, sizeof v);
error = pma_insert(pmaa, &dbtk, &dbtv, 0);
error = pma_insert(pmaa, &dbtk, &dbtv, NULL_ARGS);
assert(error == BRT_OK);
}
......@@ -833,7 +839,7 @@ void test_pma_split_varkey(void) {
v = i;
fill_dbt(&dbtv, &v, sizeof v);
error = pma_insert(pmaa, &dbtk, &dbtv, 0);
error = pma_insert(pmaa, &dbtk, &dbtv, NULL_ARGS);
assert(error == BRT_OK);
}
n = i;
......@@ -944,7 +950,7 @@ void test_pma_split_cursor(void) {
v = i;
fill_dbt(&dbtv, &v, sizeof v);
error = pma_insert(pmaa, &dbtk, &dbtv, 0);
error = pma_insert(pmaa, &dbtk, &dbtv, NULL_ARGS);
assert(error == BRT_OK);
}
assert(pma_n_entries(pmaa) == 16);
......@@ -1118,13 +1124,13 @@ void test_pma_insert_or_replace(void) {
int n_diff=-2;
r = pma_create(&pma, default_compare_fun);
assert(r==0);
r = pma_insert_or_replace(pma, fill_dbt(&dbtk, "aaa", 4), fill_dbt(&dbtv, "zzz", 4), 0, &n_diff);
r = pma_insert_or_replace(pma, fill_dbt(&dbtk, "aaa", 4), fill_dbt(&dbtv, "zzz", 4), &n_diff, NULL_ARGS);
assert(r==0); assert(n_diff==-1);
r = pma_lookup(pma, fill_dbt(&dbtk, "aaa", 4), init_dbt(&dbtv), 0);
assert(r==0); assert(dbtv.size==4); assert(memcmp(dbtv.data, "zzz", 4)==0);
r = pma_insert_or_replace(pma, fill_dbt(&dbtk, "bbbb", 5), fill_dbt(&dbtv, "ww", 3), 0, &n_diff);
r = pma_insert_or_replace(pma, fill_dbt(&dbtk, "bbbb", 5), fill_dbt(&dbtv, "ww", 3), &n_diff, NULL_ARGS);
assert(r==0); assert(n_diff==-1);
r = pma_lookup(pma, fill_dbt(&dbtk, "aaa", 4), init_dbt(&dbtv), 0);
......@@ -1133,7 +1139,7 @@ void test_pma_insert_or_replace(void) {
r = pma_lookup(pma, fill_dbt(&dbtk, "bbbb", 5), init_dbt(&dbtv), 0);
assert(r==0); assert(dbtv.size==3); assert(memcmp(dbtv.data, "ww", 3)==0);
r = pma_insert_or_replace(pma, fill_dbt(&dbtk, "bbbb", 5), fill_dbt(&dbtv, "xxxx", 5), 0, &n_diff);
r = pma_insert_or_replace(pma, fill_dbt(&dbtk, "bbbb", 5), fill_dbt(&dbtv, "xxxx", 5), &n_diff, NULL_ARGS);
assert(r==0); assert(n_diff==3);
r = pma_lookup(pma, fill_dbt(&dbtk, "aaa", 4), init_dbt(&dbtv), 0);
......@@ -1169,7 +1175,7 @@ void test_pma_delete_shrink(int n) {
fill_dbt(&key, k, strlen(k)+1);
v = i;
fill_dbt(&val, &v, sizeof v);
r = pma_insert(pma, &key, &val, 0);
r = pma_insert(pma, &key, &val, NULL_ARGS);
assert(r == 0);
}
......@@ -1218,7 +1224,7 @@ void test_pma_delete_random(int n) {
fill_dbt(&key, k, strlen(k)+1);
v = keys[i];
fill_dbt(&val, &v, sizeof v);
r = pma_insert(pma, &key, &val, 0);
r = pma_insert(pma, &key, &val, NULL_ARGS);
assert(r == 0);
}
......@@ -1289,7 +1295,7 @@ void test_pma_delete_cursor(int n) {
fill_dbt(&key, k, strlen(k)+1);
v = i;
fill_dbt(&val, &v, sizeof v);
r = pma_insert(pma, &key, &val, 0);
r = pma_insert(pma, &key, &val, NULL_ARGS);
assert(r == 0);
}
......@@ -1355,7 +1361,7 @@ void test_pma_delete_insert() {
k = 1; v = 1;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = pma_insert(pma, &key, &val, 0);
error = pma_insert(pma, &key, &val, NULL_ARGS);
assert(error == 0);
error = pma_cursor_set_position_first(pmacursor);
......@@ -1377,7 +1383,7 @@ void test_pma_delete_insert() {
k = 1; v = 2;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = pma_insert(pma, &key, &val, 0);
error = pma_insert(pma, &key, &val, NULL_ARGS);
assert(error == 0);
assert_cursor_equal(pmacursor, 2);
......@@ -1408,7 +1414,7 @@ void test_pma_double_delete() {
k = 1; v = 1;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = pma_insert(pma, &key, &val, 0);
error = pma_insert(pma, &key, &val, NULL_ARGS);
assert(error == 0);
error = pma_cursor_set_position_first(pmacursor);
......@@ -1451,7 +1457,7 @@ void test_pma_cursor_first_delete_last() {
v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = pma_insert(pma, &key, &val, 0);
error = pma_insert(pma, &key, &val, NULL_ARGS);
assert(error == 0);
}
assert(pma_n_entries(pma) == 2);
......@@ -1499,7 +1505,7 @@ void test_pma_cursor_last_delete_first() {
v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = pma_insert(pma, &key, &val, 0);
error = pma_insert(pma, &key, &val, NULL_ARGS);
assert(error == 0);
}
assert(pma_n_entries(pma) == 2);
......@@ -1554,9 +1560,9 @@ void test_pma_already_there() {
k = 1; v = 1;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = pma_insert(pma, &key, &val, 0);
error = pma_insert(pma, &key, &val, NULL_ARGS);
assert(error == 0);
error = pma_insert(pma, &key, &val, 0);
error = pma_insert(pma, &key, &val, NULL_ARGS);
assert(error == BRT_ALREADY_THERE);
error = pma_free(&pma);
......@@ -1582,7 +1588,7 @@ void test_pma_cursor_set_key() {
v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = pma_insert(pma, &key, &val, 0);
error = pma_insert(pma, &key, &val, NULL_ARGS);
assert(error == 0);
}
......@@ -1638,7 +1644,7 @@ void test_pma_cursor_set_range() {
v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = pma_insert(pma, &key, &val, 0);
error = pma_insert(pma, &key, &val, NULL_ARGS);
assert(error == 0);
}
......@@ -1707,7 +1713,7 @@ void test_pma_cursor_delete_under() {
v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = pma_insert(pma, &key, &val, 0);
error = pma_insert(pma, &key, &val, NULL_ARGS);
assert(error == 0);
}
......@@ -1772,7 +1778,7 @@ void test_pma_cursor_set_both() {
v = i;
fill_dbt(&key, &k, sizeof k);
fill_dbt(&val, &v, sizeof v);
error = pma_insert(pma, &key, &val, 0);
error = pma_insert(pma, &key, &val, NULL_ARGS);
assert(error == 0);
}
......
newbrt/pma.c
View file @ 24c21821
......@@ -14,6 +14,7 @@
#include "list.h"
#include "kv-pair.h"
#include "pma-internal.h"
#include "log.h"
/* get KEY_VALUE_OVERHEAD */
#include "brt-internal.h"
......@@ -648,7 +649,7 @@ int pma_free (PMA *pmap) {
}
/* Copies keylen and datalen */
int pma_insert (PMA pma, DBT *k, DBT *v, DB* db) {
int pma_insert (PMA pma, DBT *k, DBT *v, DB* db, TOKUTXN txn, diskoff diskoff) {
int idx = pmainternal_find(pma, k, db);
if (idx < pma_index_limit(pma) && pma->pairs[idx]) {
DBT k2;
......@@ -656,10 +657,11 @@ int pma_insert (PMA pma, DBT *k, DBT *v, DB* db) {
if (0==pma->compare_fun(db, k, fill_dbt(&k2, kv->key, kv->keylen))) {
if (kv_pair_deleted(pma->pairs[idx])) {
pma->pairs[idx] = kv_pair_realloc_same_key(kv, v->data, v->size);
return BRT_OK;
int r = tokulogger_log_phys_add_or_delete_in_leaf(txn, diskoff, 0, pma->pairs[idx]);
return r;
} else
return BRT_ALREADY_THERE; /* It is already here. Return an error. */
}
}
}
if (kv_pair_inuse(pma->pairs[idx])) {
idx = pmainternal_make_space_at (pma, idx); /* returns the new idx. */
......@@ -668,7 +670,7 @@ int pma_insert (PMA pma, DBT *k, DBT *v, DB* db) {
pma->pairs[idx] = kv_pair_malloc(k->data, k->size, v->data, v->size);
assert(pma->pairs[idx]);
pma->n_pairs_present++;
return BRT_OK;
return tokulogger_log_phys_add_or_delete_in_leaf(txn, diskoff, 1, pma->pairs[idx]);
}
int pma_delete (PMA pma, DBT *k, DB *db) {
......@@ -770,20 +772,26 @@ void __pma_delete_at(PMA pma, int here) {
toku_free(newpairs);
}
int pma_insert_or_replace (PMA pma, DBT *k, DBT *v, DB *db,
int *replaced_v_size /* If it is a replacement, set to the size of the old value, otherwise set to -1. */
) {
int pma_insert_or_replace (PMA pma, DBT *k, DBT *v,
int *replaced_v_size, /* If it is a replacement, set to the size of the old value, otherwise set to -1. */
DB *db, TOKUTXN txn, diskoff diskoff) {
//printf("%s:%d v->size=%d\n", __FILE__, __LINE__, v->size);
int idx = pmainternal_find(pma, k, db);
struct kv_pair *kv;
int r;
if (idx < pma_index_limit(pma) && (kv = pma->pairs[idx])) {
DBT k2;
// printf("%s:%d\n", __FILE__, __LINE__);
kv = kv_pair_ptr(kv);
if (0==pma->compare_fun(db, k, fill_dbt(&k2, kv->key, kv->keylen))) {
if (!kv_pair_deleted(pma->pairs[idx]))
if (!kv_pair_deleted(pma->pairs[idx])) {
*replaced_v_size = kv->vallen;
r=tokulogger_log_phys_add_or_delete_in_leaf(txn, diskoff, 0, kv);
if (r!=0) return r;
}
pma->pairs[idx] = kv_pair_realloc_same_key(kv, v->data, v->size);
return BRT_OK;
r = tokulogger_log_phys_add_or_delete_in_leaf(txn, diskoff, 0, pma->pairs[idx]);
return r;
}
}
if (kv_pair_inuse(pma->pairs[idx])) {
......@@ -795,7 +803,9 @@ int pma_insert_or_replace (PMA pma, DBT *k, DBT *v, DB *db,
assert(pma->pairs[idx]);
pma->n_pairs_present++;
*replaced_v_size = -1;
return BRT_OK;
//printf("%s:%d txn=%p\n", __FILE__, __LINE__, txn);
r = tokulogger_log_phys_add_or_delete_in_leaf(txn, diskoff, 1, pma->pairs[idx]);
return r;
}
void pma_iterate (PMA pma, void(*f)(bytevec,ITEMLEN,bytevec,ITEMLEN, void*), void*v) {
......
newbrt/pma.h
View file @ 24c21821
......@@ -4,6 +4,8 @@
#include "brttypes.h"
#include "ybt.h"
#include "yerror.h"
#include "../include/db.h"
#include "log.h"
/* An in-memory Packed Memory Array dictionary. */
/* There is a built-in-cursor. */
......@@ -26,15 +28,15 @@ int pma_n_entries (PMA);
/* Duplicates the key and keylen. */
//enum pma_errors pma_insert (PMA, bytevec key, ITEMLEN keylen, bytevec data, ITEMLEN datalen);
// The DB pointer is there so that the comparison function can be called.
enum pma_errors pma_insert (PMA, DBT*, DBT*, DB*);
enum pma_errors pma_insert (PMA, DBT*, DBT*, DB*, TOKUTXN txn, diskoff);
/* This returns an error if the key is NOT present. */
int pma_replace (PMA, bytevec key, ITEMLEN keylen, bytevec data, ITEMLEN datalen);
/* This returns an error if the key is NOT present. */
int pma_delete (PMA, DBT *, DB*);
int pma_insert_or_replace (PMA pma, DBT *k, DBT *v, DB *db,
int *replaced_v_size /* If it is a replacement, set to the size of the old value, otherwise set to -1. */
);
int pma_insert_or_replace (PMA pma, DBT *k, DBT *v,
int *replaced_v_size, /* If it is a replacement, set to the size of the old value, otherwise set to -1. */
DB *db, TOKUTXN txn, diskoff);
/* Exposes internals of the PMA by returning a pointer to the guts.
......
newbrt/wbuf.h
View file @ 24c21821
......@@ -4,6 +4,7 @@
#include <assert.h>
#include <string.h>
#include <errno.h>
#include "memory.h"
/* When serializing a value, write it into a buffer. */
/* This code requires that the buffer be big enough to hold whatever you put into it. */
......@@ -15,11 +16,10 @@ struct wbuf {
unsigned int ndone;
};
static int wbuf_init (struct wbuf *w, diskoff size) {
w->buf=toku_malloc(size);
static void wbuf_init (struct wbuf *w, void *buf, diskoff size) {
w->buf=buf;
w->size=size;
w->ndone=0;
return errno;
}
/* Write a character. */
......@@ -56,8 +56,17 @@ static void wbuf_bytes (struct wbuf *w, bytevec bytes_bv, int nbytes) {
#endif
}
static void wbuf_ulonglong (struct wbuf *w, unsigned long long ull) {
wbuf_int(w, ull>>32);
wbuf_int(w, ull&0xFFFFFFFF);
}
static void wbuf_diskoff (struct wbuf *w, diskoff off) {
wbuf_int(w, off>>32);
wbuf_int(w, off&0xFFFFFFFF);
wbuf_ulonglong(w, off);
}
static inline void wbuf_txnid (struct wbuf *w, TXNID tid) {
wbuf_ulonglong(w, tid);
}
#endif
newbrt/yerror.h
View file @ 24c21821
......@@ -3,4 +3,6 @@ enum pma_errors { BRT_OK=0, BRT_ALREADY_THERE = -2, BRT_KEYEMPTY=-3 };
enum typ_tag { TYP_BRTNODE = 0xdead0001,
TYP_CACHETABLE, TYP_PAIR, /* for cachetables */
TYP_PMA,
TYP_TOKULOGGER };
TYP_TOKULOGGER,
TYP_TOKUTXN
};
src/Makefile
View file @ 24c21821
......@@ -6,9 +6,8 @@ install: libdb.so
clean:
rm -rf *.so *.o
ydb.o: ../include/db.h ../newbrt/cachetable.h
DBBINS = ydb.o ../newbrt/brt.o ../newbrt/brt-serialize.o ../newbrt/cachetable.o ../newbrt/hashtable.o ../newbrt/header-io.o ../newbrt/key.o ../newbrt/memory.o ../newbrt/pma.o ../newbrt/ybt.o ../newbrt/primes.o
ydb.o: ../include/db.h ../newbrt/cachetable.h ../newbrt/brt.h
DBBINS = ydb.o ../newbrt/brt.o ../newbrt/brt-serialize.o ../newbrt/cachetable.o ../newbrt/hashtable.o ../newbrt/header-io.o ../newbrt/key.o ../newbrt/memory.o ../newbrt/pma.o ../newbrt/ybt.o ../newbrt/primes.o ../newbrt/log.o
libdb.so: $(DBBINS)
cc $(CPPFLAGS) $(DBBINS) -shared -o libdb.so $(CFLAGS)
libdb.a(ydb.o): ydb.o
src/tests/Makefile 0 → 100644
View file @ 24c21821
CFLAGS = -Wall -Werror -O2 -g
CPPFLAGS = -I../../include
LOADLIBES = -L../ -ldb -Wl,-rpath,..
test_log1.bdb: test_log1
test_log1 test_log0: ../libdb.so
check_log0: ./test_log0
valgrind --quiet ./test_log0
test -f dir.test_log0/log000000000000.tokulog
check_log1: ./test_log1
valgrind --quiet ./test_log1
check_db_close_no_open: ./test_db_close_no_open
valgrind --quiet ./test_db_close_no_open
.PHONY: check_log0 make_libs
make_libs:
cd ..;make
check: make_libs check_log0 check_log1 check_db_close_no_open
test_log1.bdb: test_log1.c
cc -Wall -Werror -O2 -g test_log1.c -o $@ -ldb -DDBVERSION=\"bdb\"
test_log1.bdb_link: test_log1.c
cc -Wall -Werror -O2 -g test_log1.c -o $@ -ldb $(CPPFLAGS)
test_log1.tokudb_link: test_log1.c
cc -Wall -Werror -O2 -g test_log1.c -o $@ $(LOADLIBES)
src/tests/test_db_close_no_open.c 0 → 100644
View file @ 24c21821
/* Simple test of logging. Can I start a TokuDB with logging enabled? */
#include <assert.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <db.h>
#define DIR "dir.test_db_close_no_open"
DB_ENV *env;
DB *db;
int main (int argc, char *argv[]) {
int r;
system("rm -rf " DIR);
r=mkdir(DIR, 0777); assert(r==0);
r=db_env_create(&env, 0); assert(r==0);
r=env->open(env, DIR, DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_MPOOL|DB_INIT_TXN|DB_PRIVATE|DB_CREATE, 0777); assert(r==0);
r=db_create(&db, env, 0); assert(r==0);
r=db->close(db, 0); assert(r==0);
r=env->close(env, 0); assert(r==0);
return 0;
}
src/tests/test_log0.c 0 → 100644
View file @ 24c21821
/* Simple test of logging. Can I start a TokuDB with logging enabled? */
#include <assert.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <db.h>
#define DIR "dir.test_log0"
DB_ENV *env;
int main (int argc, char *argv[]) {
int r;
system("rm -rf " DIR);
r=mkdir(DIR, 0777); assert(r==0);
r=db_env_create(&env, 0); assert(r==0);
r=env->open(env, DIR, DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_MPOOL|DB_INIT_TXN|DB_PRIVATE|DB_CREATE, 0777); assert(r==0);
r=env->close(env, 0); assert(r==0);
return 0;
}
src/tests/test_log1.c 0 → 100644
View file @ 24c21821
/* Simple test of logging. Can I start a TokuDB with logging enabled? */
#include <assert.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <db.h>
#include <string.h>
#include <stdio.h>
#ifndef DBVERSION
#define DBVERSION "toku"
#endif
#define DIR "dir.test_log1." DBVERSION
DB_ENV *env;
DB *db;
DB_TXN *tid;
#define CKERR(r) if (r!=0) fprintf(stderr, "%s:%d error %d %s\n", __FILE__, __LINE__, r, db_strerror(r)); assert(r==0);
int main (int argc, char *argv[]) {
int r;
system("rm -rf " DIR);
r=mkdir(DIR, 0777); assert(r==0);
r=db_env_create(&env, 0); assert(r==0);
r=env->open(env, DIR, DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_MPOOL|DB_INIT_TXN|DB_PRIVATE|DB_CREATE, 0777); CKERR(r);
r=db_create(&db, env, 0); CKERR(r);
r=env->txn_begin(env, 0, &tid, 0); assert(r==0);
r=db->open(db, tid, "foo.db", 0, DB_BTREE, DB_CREATE, 0777); CKERR(r);
{
DBT key,data;
memset(&key, 0, sizeof(key));
memset(&data, 0, sizeof(data));
key.data="hello";
key.size=6;
data.data="there";
data.size=6;
r=db->put(db, tid, &key, &data, 0);
CKERR(r);
}
r=tid->commit(tid, 0); assert(r==0);
r=db->close(db, 0); assert(r==0);
r=env->close(env, 0); assert(r==0);
return 0;
}
src/ydb.c
View file @ 24c21821
......@@ -15,6 +15,8 @@
#include <unistd.h>
#include "cachetable.h"
#include "log.h"
#include "memory.h"
struct db_header {
int n_databases; // Or there can be >=1 named databases. This is the count.
......@@ -38,12 +40,17 @@ struct __toku_db_internal {
};
static inline void *malloc_zero(size_t size) {
void *vp = malloc(size);
void *vp = toku_malloc(size);
if (vp)
memset(vp, 0, size);
return vp;
}
struct __toku_db_txn_internal {
//TXNID txnid64; /* A sixty-four bit txn id. */
TOKUTXN tokutxn;
};
void __toku_db_env_err (const DB_ENV *env __attribute__((__unused__)), int error, const char *fmt, ...) {
va_list ap;
va_start(ap, fmt);
......@@ -86,6 +93,7 @@ struct __toku_db_env_internal {
int files_array_limit; // How big is *files ?
struct ydb_file **files;
CACHETABLE cachetable;
TOKULOGGER logger;
};
int __toku_db_env_open (DB_ENV *env, const char *home, u_int32_t flags, int mode) {
......@@ -97,17 +105,28 @@ int __toku_db_env_open (DB_ENV *env, const char *home, u_int32_t flags, int mod
print_flags(flags);
assert(DB_PRIVATE & flags); // This means that we don't have to do anything with shared memory. And that's good enough for mysql.
r = brt_create_cachetable(&env->i->cachetable, 32);
assert(r==0);
if (flags & (DB_INIT_TXN | DB_INIT_LOG)) {
r = tokulogger_create_and_open_logger(env->i->dir, &env->i->logger);
} else {
env->i->dir = 0;
}
return 0;
}
int __toku_db_env_close (DB_ENV * env, u_int32_t flags) {
cachetable_close(&env->i->cachetable);
free(env->i->dir);
free(env->i->files);
free(env->i);
free(env);
if (env->i->logger) {
tokulogger_log_close(&env->i->logger);
}
toku_free(env->i->dir);
toku_free(env->i->files);
toku_free(env->i);
toku_free(env);
return 0;
}
int __toku_db_env_log_archive (DB_ENV *env, char **list[], u_int32_t flags) {
......@@ -225,6 +244,11 @@ int db_env_create (DB_ENV **envp, u_int32_t flags) {
int __toku_db_txn_commit (DB_TXN *txn, u_int32_t flags) {
notef("flags=%d\n", flags);
if (!txn) return -1;
int r = tokulogger_log_commit(txn->i->tokutxn);
if (r!=0) return r;
if (txn->i) toku_free(txn->i);
toku_free(txn);
return 0;
}
......@@ -233,11 +257,16 @@ u_int32_t __toku_db_txn_id (DB_TXN *txn) {
abort();
}
static TXNID next_txn=0;
int txn_begin (DB_ENV *env, DB_TXN *stxn, DB_TXN **txn, u_int32_t flags) {
DB_TXN *result = malloc_zero(sizeof(*result));
notef("parent=%p flags=0x%x\n", stxn, flags);
result->commit = __toku_db_txn_commit;
result->id = __toku_db_txn_id;
result->i = malloc(sizeof(*result->i));
int r = tokutxn_begin(&result->i->tokutxn, next_txn++, env->i->logger);
if (r!=0) return r;
*txn = result;
return 0;
}
......@@ -247,9 +276,12 @@ int txn_abort (DB_TXN *txn) {
abort();
}
#if 0
int txn_commit (DB_TXN *txn, u_int32_t flags) {
return 0;
printf("%s:%d\n", __FILE__, __LINE__);
return tokulogger_log_commit(txn->i->tokutxn);
}
#endif
int log_compare (const DB_LSN *a, const DB_LSN *b) {
fprintf(stderr, "%s:%d log_compare(%p,%p)\n", __FILE__, __LINE__, a, b);
......@@ -257,31 +289,35 @@ int log_compare (const DB_LSN *a, const DB_LSN *b) {
}
int __toku_db_close (DB *db, u_int32_t flags) {
int r = close_brt(db->i->brt);
int r = 0;
if (db->i->brt) {
r = close_brt(db->i->brt);
}
printf("%s:%d %d=__toku_db_close(%p)\n", __FILE__, __LINE__, r, db);
db->i->freed = 1;
free(db->i->database_name);
free(db->i->full_fname);
free(db->i);
free(db);
toku_free(db->i->database_name);
toku_free(db->i->full_fname);
toku_free(db->i);
toku_free(db);
return r;
}
struct __toku_dbc_internal {
BRT_CURSOR c;
DB *db;
DB_TXN *txn;
};
int __toku_c_get (DBC *c, DBT *key, DBT *data, u_int32_t flag) {
int r = brt_cursor_get(c->i->c, key, data, flag, c->i->db);
int r = brt_cursor_get(c->i->c, key, data, flag, c->i->db, c->i->txn->i->tokutxn);
return r;
}
int __toku_c_close (DBC *c) {
int r = brt_cursor_close(c->i->c);
printf("%s:%d %d=__toku_c_close(%p)\n", __FILE__, __LINE__, r, c);
free(c->i);
free(c);
toku_free(c->i);
toku_free(c);
return r;
}
......@@ -299,6 +335,7 @@ int __toku_db_cursor (DB *db, DB_TXN *txn, DBC **c, u_int32_t flags) {
result->c_del = __toku_c_del;
result->i = malloc_zero(sizeof(*result->i));
result->i->db = db;
result->i->txn = txn;
r = brt_cursor(db->i->brt, &result->i->c);
assert(r==0);
*c = result;
......@@ -328,7 +365,7 @@ char *construct_full_name (const char *dir, const char *fname) {
int dirlen = strlen(dir);
int fnamelen = strlen(fname);
int len = dirlen+fnamelen+2; // One for the / between (which may not be there). One for the trailing null.
char *result = malloc(len);
char *result = toku_malloc(len);
int l;
printf("%s:%d len(%d)=%d+%d+2\n", __FILE__, __LINE__, len, dirlen, fnamelen);
assert(result);
......@@ -388,8 +425,9 @@ int __toku_db_open (DB *db, DB_TXN *txn, const char *fname, const char *dbname,
assert(r==0);
return 0;
}
int __toku_db_put (DB *db, DB_TXN *txn, DBT *key, DBT *data, u_int32_t flags) {
int r = brt_insert(db->i->brt, key, data, db);
int r = brt_insert(db->i->brt, key, data, db, txn->i->tokutxn);
//printf("%s:%d %d=__toku_db_put(...)\n", __FILE__, __LINE__, r);
return r;
}
......@@ -465,3 +503,22 @@ int db_create (DB **db, DB_ENV *env, u_int32_t flags) {
*db = result;
return 0;
}
char *db_strerror (int error) {
if (error == 0) return "Success: 0";
if (error > 0) {
char *result = strerror(error);
if (result) return result;
unknown:
{
static char unknown_result[100]; // Race condition if two threads call this at the same time. However even in a bad case, it should be some sort of nul-terminated string.
snprintf(unknown_result, 100, "Unknown error code: %d", error);
return unknown_result;
}
}
switch (error) {
default:
goto unknown;
}
}
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