#3211 #3185 test brt verify and repair refs[t:3211] refs[t:3185]

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

newbrt/brt-internal.h

@@ -257,6 +257,12 @@ void toku_serialize_descriptor_contents_to_wbuf(struct wbuf *wb, const DESCRIPTO
 
 void toku_brtnode_free (BRTNODE *node);
 
+// append a child node to a parent node
+void toku_brt_nonleaf_append_child(BRTNODE node, BRTNODE child, struct kv_pair *pivotkey, size_t pivotkeysize);
+
+// append a cmd to a nonleaf node child buffer
+void toku_brt_append_to_child_buffer(BRTNODE node, int childnum, int type, XIDS xids, DBT *key, DBT *val);
+
 #if 1
 #define DEADBEEF ((void*)0xDEADBEEF)
 #else
@@ -385,6 +391,7 @@ typedef struct le_status {
 
 void toku_le_get_status(LE_STATUS);
 
+int brt_leaf_apply_cmd_once (BRTNODE node, const BRT_MSG cmd, u_int32_t idx, LEAFENTRY le, TOKULOGGER logger);
 
 C_END
 

newbrt/brt.c

@@ -770,6 +770,38 @@ int toku_create_new_brtnode (BRT t, BRTNODE *result, int height, size_t mpsize)
     return 0;
 }
 
+static void
+init_childinfo(BRTNODE node, int childnum, BRTNODE child) {
+    BNC_BLOCKNUM(node,childnum) = child->thisnodename;
+    BNC_HAVE_FULLHASH(node,childnum) = FALSE;
+    BNC_NBYTESINBUF(node,childnum) = 0;
+    BNC_SUBTREE_FINGERPRINT(node,childnum) = 0;
+    BNC_SUBTREE_ESTIMATES(node,childnum) = zero_estimates;
+    int r = toku_fifo_create(&BNC_BUFFER(node,childnum));
+    resource_assert_zero(r);
+}
+
+static void
+init_childkey(BRTNODE node, int childnum, struct kv_pair *pivotkey, size_t pivotkeysize) {
+    node->u.n.childkeys[childnum] = pivotkey;
+    node->u.n.totalchildkeylens += pivotkeysize;
+}
+
+// append a child node to a parent node
+void
+toku_brt_nonleaf_append_child(BRTNODE node, BRTNODE child, struct kv_pair *pivotkey, size_t pivotkeysize) {
+    int childnum = node->u.n.n_children;
+    node->u.n.n_children++;
+    XREALLOC_N(node->u.n.n_children, node->u.n.childinfos);
+    init_childinfo(node, childnum, child);
+    XREALLOC_N(node->u.n.n_children-1, node->u.n.childkeys);
+    if (pivotkey) {
+        invariant(childnum > 0);
+        init_childkey(node, childnum-1, pivotkey, pivotkeysize);
+    }
+    node->dirty = 1;
+}
+
 static int
 fill_buf (OMTVALUE lev, u_int32_t idx, void *varray)
 {
@@ -1414,7 +1446,7 @@ brt_leaf_delete_leafentry (BRTNODE node, u_int32_t idx, LEAFENTRY le)
     return r;
 }
 
-static int
+int
 brt_leaf_apply_cmd_once (BRTNODE node, const BRT_MSG cmd,
                          u_int32_t idx, LEAFENTRY le, TOKULOGGER logger)
 // Effect: Apply cmd to leafentry
@@ -1771,27 +1803,26 @@ static int brt_nonleaf_cmd_once_to_child (BRT t, BRTNODE node, unsigned int chil
     }
 
  put_in_fifo:
+    toku_brt_append_to_child_buffer(node, childnum, cmd->type, cmd->xids, cmd->u.id.key, cmd->u.id.val);
 
-    {
-        //TODO: Determine if we like direct access to type, to key, to val, 
-        int type = cmd->type;
-        DBT *k = cmd->u.id.key;
-        DBT *v = cmd->u.id.val;
-
-	node->local_fingerprint += node->rand4fingerprint * toku_calc_fingerprint_cmd(type, cmd->xids, k->data, k->size, v->data, v->size);
-        int diff = k->size + v->size + KEY_VALUE_OVERHEAD + BRT_CMD_OVERHEAD + xids_get_serialize_size(cmd->xids);
-        int r=toku_fifo_enq(BNC_BUFFER(node,childnum), k->data, k->size, v->data, v->size, type, cmd->xids);
+    verify_local_fingerprint_nonleaf(node);
+
+    return 0;
+}
+
+// append a cmd to a nonleaf node's child buffer
+void
+toku_brt_append_to_child_buffer(BRTNODE node, int childnum, int type, XIDS xids, DBT *key, DBT *val) {
+    node->local_fingerprint += node->rand4fingerprint * toku_calc_fingerprint_cmd(type, xids, key->data, key->size, val->data, val->size);
+    int diff = key->size + val->size + KEY_VALUE_OVERHEAD + BRT_CMD_OVERHEAD + xids_get_serialize_size(xids);
+    int r = toku_fifo_enq(BNC_BUFFER(node,childnum), key->data, key->size, val->data, val->size, type, xids);
     lazy_assert_zero(r);
     node->u.n.n_bytes_in_buffers += diff;
     BNC_NBYTESINBUF(node, childnum) += diff;
     node->dirty = 1;
-    }
-
-    verify_local_fingerprint_nonleaf(node);
-
-    return 0;
 }
 
+
 /* find the leftmost child that may contain the key */
 unsigned int toku_brtnode_which_child (BRTNODE node , DBT *k, BRT t) {
     lazy_assert(node->height>0);

newbrt/tests/make-tree.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2011 Tokutek Inc.  All rights reserved."
+
+// generate fractal trees with a given height, fanout, and number of leaf elements per leaf.
+// jam the child buffers with inserts.
+// this code can be used as a template to build broken trees
+
+#include "includes.h"
+#include "test.h"
+
+static BRTNODE
+make_node(BRT brt, int height) {
+    BRTNODE node = NULL;
+    int r = toku_create_new_brtnode(brt, &node, height, 0);
+    assert(r == 0);
+    return node;
+}
+
+static void
+append_leaf(BRTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen) {
+    assert(leafnode->height == 0);
+
+    DBT thekey; toku_fill_dbt(&thekey, key, keylen);
+    DBT theval; toku_fill_dbt(&theval, val, vallen);
+
+    // get an index that we can use to create a new leaf entry
+    uint32_t idx = toku_omt_size(leafnode->u.l.buffer);
+
+    // apply an insert to the leaf node
+    BRT_MSG_S cmd = { BRT_INSERT, xids_get_root_xids(), .u.id = { &thekey, &theval } };
+    int r = brt_leaf_apply_cmd_once(leafnode, &cmd, idx, NULL, NULL);
+    assert(r == 0);
+
+    // dont forget to dirty the node
+    leafnode->dirty = 1;
+}
+
+static void
+populate_leaf(BRTNODE leafnode, int seq, int n, int *minkey, int *maxkey) {
+    for (int i = 0; i < n; i++) {
+        int k = htonl(seq + i);
+        int v = seq + i;
+        append_leaf(leafnode, &k, sizeof k, &v, sizeof v);
+    }
+    *minkey = htonl(seq);
+    *maxkey = htonl(seq + n - 1);
+}
+
+static void
+insert_into_child_buffer(BRTNODE node, int childnum, int minkey, int maxkey) {
+    for (unsigned int val = htonl(minkey); val <= htonl(maxkey); val++) {
+        unsigned int key = htonl(val);
+        DBT thekey; toku_fill_dbt(&thekey, &key, sizeof key);
+        DBT theval; toku_fill_dbt(&theval, &val, sizeof val);
+        toku_brt_append_to_child_buffer(node, childnum, BRT_INSERT, xids_get_root_xids(), &thekey, &theval);
+    }
+}
+
+static BRTNODE
+make_tree(BRT brt, int height, int fanout, int nperleaf, int *seq, int *minkey, int *maxkey) {
+    BRTNODE node;
+    if (height == 0) {
+        node = make_node(brt, 0);
+        populate_leaf(node, *seq, nperleaf, minkey, maxkey);
+        *seq += nperleaf;
+    } else {
+        node = make_node(brt, height);
+        int minkeys[fanout], maxkeys[fanout];
+        for (int childnum = 0; childnum < fanout; childnum++) {
+            BRTNODE child = make_tree(brt, height-1, fanout, nperleaf, seq, &minkeys[childnum], &maxkeys[childnum]);
+            if (childnum == 0) 
+                toku_brt_nonleaf_append_child(node, child, NULL, 0);
+            else {
+                int k = maxkeys[childnum-1]; // use the max of the left tree
+                struct kv_pair *pivotkey = kv_pair_malloc(&k, sizeof k, NULL, 0);
+                toku_brt_nonleaf_append_child(node, child, pivotkey, sizeof k);
+            }
+            int r = toku_unpin_brtnode(brt, child);
+            assert(r == 0);
+            insert_into_child_buffer(node, childnum, minkeys[childnum], maxkeys[childnum]);
+        }
+        *minkey = minkeys[0];
+        *maxkey = maxkeys[0];
+        for (int i = 1; i < fanout; i++) {
+            if (memcmp(minkey, &minkeys[i], sizeof minkeys[i]) > 0)
+                *minkey = minkeys[i];
+            if (memcmp(maxkey, &maxkeys[i], sizeof maxkeys[i]) < 0)
+                *maxkey = maxkeys[i];
+        }
+    }
+    return node;
+}
+
+static UU() void
+deleted_row(UU() DB *db, UU() DBT *key, UU() DBT *val) {
+}
+
+static void 
+test_make_tree(int height, int fanout, int nperleaf, int do_verify) {
+    int r;
+
+    // cleanup
+    char fname[]= __FILE__ ".brt";
+    r = unlink(fname);
+    assert(r == 0 || (r == -1 && errno == ENOENT));
+
+    // create a cachetable
+    CACHETABLE ct = NULL;
+    r = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER);
+    assert(r == 0);
+
+    // create the brt
+    TOKUTXN null_txn = NULL;
+    DB *null_db = NULL;
+    BRT brt = NULL;
+    r = toku_open_brt(fname, 1, &brt, 1024, ct, null_txn, toku_builtin_compare_fun, null_db);
+    assert(r == 0);
+
+    // make a tree
+    int seq = 0, minkey, maxkey;
+    BRTNODE newroot = make_tree(brt, height, fanout, nperleaf, &seq, &minkey, &maxkey);
+
+    // discard the old root block
+    u_int32_t fullhash = 0;
+    CACHEKEY *rootp;
+    rootp = toku_calculate_root_offset_pointer(brt, &fullhash);
+
+    // set the new root to point to the new tree
+    *rootp = newroot->thisnodename;
+
+    // unpin the new root
+    r = toku_unpin_brtnode(brt, newroot);
+    assert(r == 0);
+
+    if (do_verify) {
+        r = toku_verify_brt(brt);
+        assert(r == 0);
+    }
+
+    // flush to the file system
+    r = toku_close_brt(brt, 0);     
+    assert(r == 0);
+
+    // shutdown the cachetable
+    r = toku_cachetable_close(&ct);
+    assert(r == 0);
+}
+
+static int
+usage(void) {
+    return 1;
+}
+
+int
+test_main (int argc , const char *argv[]) {
+    int height = 1;
+    int fanout = 2;
+    int nperleaf = 8;
+    int do_verify = 1;
+    for (int i = 1; i < argc; i++) {
+        const char *arg = argv[i];
+        if (strcmp(arg, "-v") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose = 0;
+            continue;
+        }
+        if (strcmp(arg, "--height") == 0 && i+1 < argc) {
+            height = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--fanout") == 0 && i+1 < argc) {
+            fanout = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--nperleaf") == 0 && i+1 < argc) {
+            nperleaf = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--verify") == 0 && i+1 < argc) {
+            do_verify = atoi(argv[++i]);
+            continue;
+        }
+        return usage();
+    }
+    test_make_tree(height, fanout, nperleaf, do_verify);
+    return 0;
+}