Log block allocation. Addresses #27.

git-svn-id: file:///svn/tokudb@1737 c7de825b-a66e-492c-adef-691d508d4ae1

newbrt/brt.c

@@ -38,7 +38,7 @@
 
 extern long long n_items_malloced;
 
-static DISKOFF malloc_diskblock (BRT brt, int size);
+static int malloc_diskblock (DISKOFF *res, BRT brt, int size, TOKUTXN);
 //static void verify_local_fingerprint_nonleaf (BRTNODE node);
 
 /* Frees a node, including all the stuff in the hash table. */
@@ -237,14 +237,16 @@ int kvpair_compare (const void *av, const void *bv) {
 #endif
 
 /* Forgot to handle the case where there is something in the freelist. */
-static DISKOFF malloc_diskblock_header_is_in_memory (BRT brt, int size) {
+static int malloc_diskblock_header_is_in_memory (DISKOFF *res, BRT brt, int size, TOKUTXN txn) {
     DISKOFF result = brt->h->unused_memory;
     brt->h->unused_memory+=size;
     brt->h->dirty = 1;
-    return result;
+    int r = toku_log_changeunusedmemory(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), result, brt->h->unused_memory);
+    *res = result;
+    return r;
 }
 
-DISKOFF malloc_diskblock (BRT brt, int size) {
+int malloc_diskblock (DISKOFF *res, BRT brt, int size, TOKUTXN txn) {
 #if 0
     int r = read_and_pin_brt_header(brt->fd, &brt->h);
     assert(r==0);
@@ -255,7 +257,7 @@ DISKOFF malloc_diskblock (BRT brt, int size) {
 	return result;
     }
 #else
-    return malloc_diskblock_header_is_in_memory(brt,size);
+    return malloc_diskblock_header_is_in_memory(res, brt,size, txn);
 #endif
 }
 
@@ -303,7 +305,9 @@ static void initialize_brtnode (BRT t, BRTNODE n, DISKOFF nodename, int height)
 static void create_new_brtnode (BRT t, BRTNODE *result, int height, TOKUTXN txn) {
     TAGMALLOC(BRTNODE, n);
     int r;
-    DISKOFF name = malloc_diskblock(t, t->h->nodesize);
+    DISKOFF name;
+    r = malloc_diskblock(&name, t, t->h->nodesize, txn);
+    assert(r==0);
     assert(n);
     assert(t->h->nodesize>0);
     //printf("%s:%d malloced %lld (and malloc again=%lld)\n", __FILE__, __LINE__, name, malloc_diskblock(t, t->nodesize));
@@ -1448,7 +1452,8 @@ int toku_brt_open(BRT t, const char *fname, const char *fname_in_env, const char
 	    t->h->n_named_roots++;
 	    if ((t->h->names[t->h->n_named_roots-1] = toku_strdup(dbname)) == 0)                                { assert(errno==ENOMEM); r=ENOMEM; goto died_after_read_and_pin; }
 	    //printf("%s:%d t=%p\n", __FILE__, __LINE__, t);
-	    t->h->roots[t->h->n_named_roots-1] = malloc_diskblock_header_is_in_memory(t, t->h->nodesize);
+	    r = malloc_diskblock_header_is_in_memory(&t->h->roots[t->h->n_named_roots-1], t, t->h->nodesize, txn);
+	    if (r!=0) goto died_after_read_and_pin;
 	    t->h->dirty = 1;
 	    if ((r=setup_brt_root_node(t, t->h->roots[t->h->n_named_roots-1], txn))!=0) goto died_after_read_and_pin;
 	}
@@ -1593,8 +1598,18 @@ static int brt_init_new_root(BRT brt, BRTNODE nodea, BRTNODE nodeb, DBT splitk,
     int r;
     int new_height = nodea->height+1;
     int new_nodesize = brt->h->nodesize;
-    DISKOFF newroot_diskoff=malloc_diskblock(brt, new_nodesize);
+    DISKOFF newroot_diskoff;
+    r=malloc_diskblock(&newroot_diskoff, brt, new_nodesize, txn);
+    assert(r==0);
     assert(newroot);
+    if (brt->database_name==0) {
+	toku_log_changeunnamedroot(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), *rootp, newroot_diskoff);
+    } else {
+	BYTESTRING bs;
+	bs.len = 1+strlen(brt->database_name);
+	bs.data = brt->database_name;
+	toku_log_changenamedroot(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), bs, *rootp, newroot_diskoff);
+    }
     *rootp=newroot_diskoff;
     brt->h->dirty=1;
     initialize_brtnode (brt, newroot, newroot_diskoff, new_height);
@@ -1614,7 +1629,6 @@ static int brt_init_new_root(BRT brt, BRTNODE nodea, BRTNODE nodeb, DBT splitk,
     //verify_local_fingerprint_nonleaf(nodeb);
     r=toku_log_newbrtnode(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), newroot_diskoff, new_height, new_nodesize, (brt->flags&TOKU_DB_DUPSORT)!=0, newroot->rand4fingerprint);
     if (r!=0) return r;
-    printf("doing addchild\n");
     r=toku_log_addchild(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), newroot_diskoff, 0);
     if (r!=0) return r;
     r=toku_log_addchild(txn, toku_txn_get_txnid(txn), toku_cachefile_filenum(brt->cf), newroot_diskoff, 1);

newbrt/logformat.c

@@ -13,6 +13,7 @@
 #include <ctype.h>
 #include <stdarg.h>
 #include <stdio.h>
+#include <stdlib.h>
 #include <string.h>
 #include <sys/stat.h>
 #include <sys/types.h>
@@ -60,6 +61,22 @@ const struct logtype logtypes[] = {
 			   {"u_int8_t", "is_dup_sort", 0},
 			   {"u_int32_t", "rand4fingerprint", 0},
 			   NULLFIELD}},
+    {"changeunnamedroot", 'u', FA{{"TXNID",   "txnid", 0},
+				 {"FILENUM", "filenum", 0},
+				 {"DISKOFF", "oldroot", 0},
+				 {"DISKOFF", "newroot", 0},
+				 NULLFIELD}},
+    {"changenamedroot", 'n', FA{{"TXNID",   "txnid", 0},
+				{"FILENUM", "filenum", 0},
+				{"BYTESTRING", "name", 0},
+				{"DISKOFF", "oldroot", 0},
+				{"DISKOFF", "newroot", 0},
+				NULLFIELD}},
+    {"changeunusedmemory", 'm',  FA{{"TXNID",   "txnid", 0},
+				    {"FILENUM", "filenum", 0},
+				    {"DISKOFF", "oldunused", 0},
+				    {"DISKOFF", "newunused", 0},
+				    NULLFIELD}},
     {"addchild", 'c', FA{{"TXNID",   "txnid", 0},
 			 {"FILENUM", "filenum", 0},
 			 {"DISKOFF", "diskoff", 0},
@@ -129,7 +146,9 @@ void fprintf2 (FILE *f1, FILE *f2, const char *format, ...) {
 FILE *hf=0, *cf=0;
 
 void generate_lt_enum (void) {
+    char used_cmds[256];
     int count=0;
+    memset(used_cmds, 0, 256);
     fprintf(hf, "enum lt_cmd {");
     DO_LOGTYPES(lt,
 		({
@@ -137,6 +156,8 @@ void generate_lt_enum (void) {
 		    count++;
 		    fprintf(hf, "\n");
 		    fprintf(hf,"    LT_%-16s = '%c'", lt->name, lt->command);
+		    if (used_cmds[(unsigned char)lt->command]!=0) { fprintf(stderr, "%s:%d Command %d (%c) was used twice\n", __FILE__, __LINE__, lt->command, lt->command); abort(); }
+		    used_cmds[(unsigned char)lt->command]=1;
 		}));
     fprintf(hf, "\n};\n\n");
 }

newbrt/roll.c

@@ -207,17 +207,24 @@ int toku_rollback_newbrtnode (struct logtype_newbrtnode *le, TOKUTXN txn) {
 }
 
 
-int toku_rollback_addchild (struct logtype_addchild *le, TOKUTXN txn) ABORTIT
-void toku_recover_addchild (struct logtype_addchild *le) {
-    printf("%s:%d %s\n", __FILE__, __LINE__, __FUNCTION__);
+static void recover_setup_node (FILENUM filenum, DISKOFF diskoff, CACHEFILE *cf, BRTNODE *resultnode) {
     struct cf_pair *pair;
-    int r = find_cachefile(le->filenum, &pair);
+    int r = find_cachefile(filenum, &pair);
     assert(r==0);
-    void *node_v;
     assert(pair->brt);
-    r = toku_cachetable_get_and_pin(pair->cf, le->diskoff, &node_v, NULL, toku_brtnode_flush_callback, toku_brtnode_fetch_callback, pair->brt);
+    void *node_v;
+    r = toku_cachetable_get_and_pin(pair->cf, diskoff, &node_v, NULL, toku_brtnode_flush_callback, toku_brtnode_fetch_callback, pair->brt);
     assert(r==0);
     BRTNODE node = node_v;
+    *resultnode = node;
+    *cf = pair->cf;
+}
+
+int toku_rollback_addchild (struct logtype_addchild *le, TOKUTXN txn) ABORTIT
+void toku_recover_addchild (struct logtype_addchild *le) {
+    CACHEFILE cf;
+    BRTNODE node;
+    recover_setup_node(le->filenum, le->diskoff, &cf, &node);
     assert(node->height>0);
     assert(le->childnum <= (unsigned)node->u.n.n_children);
     unsigned int i;
@@ -234,11 +241,11 @@ void toku_recover_addchild (struct logtype_addchild *le) {
     node->u.n.childinfos[le->childnum].subtree_fingerprint = 0;
     node->u.n.childkeys [le->childnum] = 0;
     node->u.n.children  [le->childnum] = -1;
-    r= toku_fifo_create(&node->u.n.buffers[le->childnum]); assert(r==0);
+    int r= toku_fifo_create(&node->u.n.buffers[le->childnum]); assert(r==0);
     node->u.n.n_bytes_in_buffer[le->childnum] = 0;
     node->u.n.n_cursors[le->childnum] = 0;
     node->u.n.n_children++;
-    r = toku_cachetable_unpin(pair->cf, le->diskoff, 1, toku_serialize_brtnode_size(node));
+    r = toku_cachetable_unpin(cf, le->diskoff, 1, toku_serialize_brtnode_size(node));
     assert(r==0);
 }
 
@@ -433,3 +440,28 @@ int toku_rollback_pmadistribute (struct logtype_pmadistribute *le, TOKUTXN txn)
 int toku_rollback_fheader (struct logtype_fheader *le, TOKUTXN txn)             ABORTIT
 int toku_rollback_resizepma (struct logtype_resizepma *le, TOKUTXN txn)         ABORTIT
 
+void toku_recover_changeunnamedroot (struct logtype_changeunnamedroot *le) {
+    struct cf_pair *pair;
+    int r = find_cachefile(le->filenum, &pair);
+    assert(r==0);
+    assert(pair->brt);
+    r = toku_read_and_pin_brt_header(pair->cf, &pair->brt->h);
+    assert(r==0);
+    pair->brt->h->unnamed_root = le->newroot;
+    r = toku_unpin_brt_header(pair->brt);
+}
+void toku_recover_changenamedroot (struct logtype_changenamedroot *le) { le=le; assert(0); }
+int toku_rollback_changeunnamedroot (struct logtype_changeunnamedroot *le, TOKUTXN txn) ABORTIT
+int toku_rollback_changenamedroot (struct logtype_changenamedroot *le, TOKUTXN txn) ABORTIT
+
+void toku_recover_changeunusedmemory (struct logtype_changeunusedmemory *le)  {
+    struct cf_pair *pair;
+    int r = find_cachefile(le->filenum, &pair);
+    assert(r==0);
+    assert(pair->brt);
+    r = toku_read_and_pin_brt_header(pair->cf, &pair->brt->h);
+    assert(r==0);
+    pair->brt->h->unused_memory = le->newunused;
+    r = toku_unpin_brt_header(pair->brt);
+}
+int toku_rollback_changeunusedmemory (struct logtype_changeunusedmemory *le, TOKUTXN txn) ABORTIT