Make some tests run better. Addresses #1195.

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

newbrt/brt-verify.c

@@ -82,10 +82,10 @@ static void verify_pair (bytevec key, unsigned int keylen,
     DBT k1,k2;
     if (thislorange) assert(brt->compare_fun(brt->db,
                                              toku_fill_dbt(&k1,thislorange,thislolen),
-                                             toku_fill_dbt(&k2,key,keylen)) < 0);
+                                             toku_fill_dbt(&k2,key,keylen)) <= 0);
     if (thishirange && (brt->compare_fun(brt->db,
                                          toku_fill_dbt(&k1,key,keylen),
-                                         toku_fill_dbt(&k2,thishirange,thishilen)) > 0)) {
+                                         toku_fill_dbt(&k2,thishirange,thishilen)) >= 0)) {
         printf("%s:%d in buffer %d key %s is bigger than %s\n", __FILE__, __LINE__, i, (char*)key, (char*)thishirange);
         *vparg->resultp = 1;
     }
@@ -140,7 +140,10 @@ int toku_verify_brtnode (BRT brt, BLOCKNUM blocknum, bytevec lorange, ITEMLEN lo
 		thishilen  =toku_brt_pivot_key_len(brt, node->u.n.childkeys[i]);
 	    }
             struct verify_pair_arg vparg = { brt, i, thislorange, thislolen, thishirange, thishilen, &result };
-            toku_fifo_iterate(BNC_BUFFER(node,i), verify_pair, &vparg);
+	    if (!(brt->flags & TOKU_DB_DUP)) {
+		// verify_pair doesn't work for dupsort
+		toku_fifo_iterate(BNC_BUFFER(node,i), verify_pair, &vparg);
+	    }
 	}
 	//if (lorange) printf("%s:%d lorange=%s\n", __FILE__, __LINE__, (char*)lorange);
 	//if (hirange) printf("%s:%d lorange=%s\n", __FILE__, __LINE__, (char*)hirange);

newbrt/brt.c

@@ -234,6 +234,7 @@ int toku_brt_debug_mode = 0;
 #endif
 
 static u_int32_t compute_child_fullhash (CACHEFILE cf, BRTNODE node, int childnum) {
+    assert(node->height>0 && childnum<node->u.n.n_children);
     switch (BNC_HAVE_FULLHASH(node, childnum)) {
     case TRUE:
 	{
@@ -1078,7 +1079,7 @@ handle_split_of_child (BRT t, BRTNODE node, int childnum,
 }
 
 static int
-brt_split_child (BRT t, BRTNODE node, int childnum, TOKULOGGER logger)
+brt_split_child (BRT t, BRTNODE node, int childnum, TOKULOGGER logger, BOOL *did_react)
 {
     if (0) {
 	printf("%s:%d Node %" PRIu64 "->u.n.n_children=%d estimates=", __FILE__, __LINE__, node->thisnodename.b, node->u.n.n_children);
@@ -1128,6 +1129,7 @@ brt_split_child (BRT t, BRTNODE node, int childnum, TOKULOGGER logger)
 	if (r!=0) return r;
     }
     // printf("%s:%d child did split\n", __FILE__, __LINE__);
+    *did_react = TRUE;
     {
 	int r = handle_split_of_child (t, node, childnum, nodea, nodeb, &splitk, logger);
 	if (0) {
@@ -1688,7 +1690,7 @@ brt_nonleaf_cmd_many (BRT t, BRTNODE node, BRT_CMD cmd, TOKULOGGER logger,
 //  The re_array[i] gets set to the reactivity of any modified child i.  (And there may be several such children.)
 {
     /* find all children that need a copy of the command */
-    unsigned int *MALLOC_N(node->u.n.n_children, sendchild);
+    unsigned int sendchild[node->u.n.n_children];
     unsigned int delidx = 0;
 #define sendchild_append(i) \
         if (delidx == 0 || sendchild[delidx-1] != i) sendchild[delidx++] = i;
@@ -1724,7 +1726,6 @@ brt_nonleaf_cmd_many (BRT t, BRTNODE node, BRT_CMD cmd, TOKULOGGER logger,
     }
     r=0;
  return_r:
-    toku_free(sendchild);
     return r;
 }
 
@@ -1928,7 +1929,7 @@ maybe_merge_pinned_nodes (BRT t,
 }
 
 static int
-brt_merge_child (BRT t, BRTNODE node, int childnum_to_merge, BOOL *did_io, TOKULOGGER logger)
+brt_merge_child (BRT t, BRTNODE node, int childnum_to_merge, BOOL *did_io, TOKULOGGER logger, BOOL *did_react)
 {
     if (node->u.n.n_children < 2) return 0; // if no siblings, we are merged as best we can.
 
@@ -1991,6 +1992,7 @@ brt_merge_child (BRT t, BRTNODE node, int childnum_to_merge, BOOL *did_io, TOKUL
 	r = maybe_merge_pinned_nodes(t, node, childnuma, node->u.n.childkeys[childnuma], childa, childb, logger, &did_merge, &splitk_kvpair);
 	if (childa->height>0) { int i; for (i=0; i+1<childa->u.n.n_children; i++) assert(childa->u.n.childkeys[i]); }
 	//(toku_verify_counts(childa), toku_verify_estimates(t,childa));
+	*did_react = did_merge;
 	if (did_merge) assert(!splitk_kvpair); else assert(splitk_kvpair);
 	if (r!=0) goto return_r;
 
@@ -2035,13 +2037,13 @@ brt_merge_child (BRT t, BRTNODE node, int childnum_to_merge, BOOL *did_io, TOKUL
     }
 
 static int
-brt_handle_maybe_reactive_child(BRT t, BRTNODE node, int childnum, enum reactivity re, BOOL *did_io, TOKULOGGER logger) {
+brt_handle_maybe_reactive_child(BRT t, BRTNODE node, int childnum, enum reactivity re, BOOL *did_io, TOKULOGGER logger, BOOL *did_react) {
     switch (re) {
     case RE_STABLE: return 0;
     case RE_FISSIBLE:
-	return brt_split_child(t, node, childnum, logger);
+	return brt_split_child(t, node, childnum, logger, did_react);
     case RE_FUSIBLE:
-	return brt_merge_child(t, node, childnum, did_io, logger);
+	return brt_merge_child(t, node, childnum, did_io, logger, did_react);
     }
     abort(); // this cannot happen
 }
@@ -2186,7 +2188,7 @@ brtnode_put_cmd (BRT t, BRTNODE node, BRT_CMD cmd, TOKULOGGER logger, enum react
     if (node->height==0) {
 	return brt_leaf_put_cmd(t, node, cmd, logger, re);
     } else {
-	enum reactivity *MALLOC_N(node->u.n.n_children, child_re);
+	enum reactivity child_re[node->u.n.n_children];
 	{ int i; for (i=0; i<node->u.n.n_children; i++) child_re[i]=RE_STABLE; }
 	int r = brt_nonleaf_put_cmd(t, node, cmd, logger, child_re, did_io);
 	if (r!=0) goto return_r;
@@ -2203,12 +2205,12 @@ brtnode_put_cmd (BRT t, BRTNODE node, BRT_CMD cmd, TOKULOGGER logger, enum react
 	int original_n_children = node->u.n.n_children;
 	for (i=0; i<original_n_children; i++) {
 	    int childnum = original_n_children - 1 -i;
-	    r = brt_handle_maybe_reactive_child(t, node, childnum, child_re[childnum], did_io, logger);
+	    BOOL did_react; // ignore the result.
+	    r = brt_handle_maybe_reactive_child(t, node, childnum, child_re[childnum], did_io, logger, &did_react);
 	    if (r!=0) break;
 	    if (*did_io) break;
 	}
     return_r:
-	toku_free(child_re);
 	*re = get_nonleaf_reactivity(node);
 	return r;
     }
@@ -3158,7 +3160,10 @@ static int
 brt_search_node (BRT brt, BRTNODE node, brt_search_t *search, DBT *newkey, DBT *newval, enum reactivity *re, TOKULOGGER logger, OMTCURSOR omtcursor);
 
 /* search in a node's child */
-static int brt_search_child(BRT brt, BRTNODE node, int childnum, brt_search_t *search, DBT *newkey, DBT *newval, enum reactivity *parent_re, TOKULOGGER logger, OMTCURSOR omtcursor) {
+static int brt_search_child(BRT brt, BRTNODE node, int childnum, brt_search_t *search, DBT *newkey, DBT *newval, enum reactivity *parent_re, TOKULOGGER logger, OMTCURSOR omtcursor, BOOL *did_react)
+// Effect: Search in a node's child.
+//  If we change the shape, set *did_react = TRUE.  Else set *did_react = FALSE.
+{
 
     /* if the child's buffer is not empty then empty it */
     if (BNC_NBYTESINBUF(node, childnum) > 0) {
@@ -3180,15 +3185,16 @@ static int brt_search_child(BRT brt, BRTNODE node, int childnum, brt_search_t *s
     BRTNODE childnode = node_v;
     enum reactivity child_re = RE_STABLE;
     int r = brt_search_node(brt, childnode, search, newkey, newval, &child_re, logger, omtcursor);
-    if (r!=0) goto return_r;
-
-    BOOL did_io = FALSE;
-    r = brt_handle_maybe_reactive_child(brt, node, childnum, child_re, &did_io, logger);
+    // Even if r is reactive, we want to handle the maybe reactive child.
+    {
+	BOOL did_io = FALSE;
+	int rr = brt_handle_maybe_reactive_child(brt, node, childnum, child_re, &did_io, logger, did_react);
+	if (rr!=0) r = rr; // if we got an error, then return rr.  Else we will return the r from brt_search_node().
+    }
 
- return_r:
     {
 	int rr = toku_cachetable_unpin(brt->cf, childnode->thisnodename, childnode->fullhash, childnode->dirty, brtnode_memory_size(childnode)); 
-	assert(rr == 0);
+	if (rr!=0) r = rr;
     }
     *parent_re = get_nonleaf_reactivity(node);
     
@@ -3196,30 +3202,37 @@ static int brt_search_child(BRT brt, BRTNODE node, int childnum, brt_search_t *s
 }
 
 static int brt_search_nonleaf_node(BRT brt, BRTNODE node, brt_search_t *search, DBT *newkey, DBT *newval, enum reactivity *re, TOKULOGGER logger, OMTCURSOR omtcursor) {
-    int c;
-
-    /* binary search is overkill for a small array */
-    int child[node->u.n.n_children];
-
-    /* scan left to right or right to left depending on the search direction */
-    for (c = 0; c < node->u.n.n_children; c++) 
-        child[c] = search->direction & BRT_SEARCH_LEFT ? c : node->u.n.n_children - 1 - c;
-
-    for (c = 0; c < node->u.n.n_children-1; c++) {
-        int p = search->direction & BRT_SEARCH_LEFT ? child[c] : child[c] - 1;
-        struct kv_pair *pivot = node->u.n.childkeys[p];
-        DBT pivotkey, pivotval;
-        if (search->compare(search, 
-                            toku_fill_dbt(&pivotkey, kv_pair_key(pivot), kv_pair_keylen(pivot)), 
-                            brt->flags & TOKU_DB_DUPSORT ? toku_fill_dbt(&pivotval, kv_pair_val(pivot), kv_pair_vallen(pivot)): 0)) {
-            int r = brt_search_child(brt, node, child[c], search, newkey, newval, re, logger, omtcursor);
-	    assert(r != EAGAIN);
-	    if (r == 0) return r;
-        }
-    }
+ again:
+    {
+
+	int c;
+
+	/* binary search is overkill for a small array */
+	int child[node->u.n.n_children];
+
+	/* scan left to right or right to left depending on the search direction */
+	for (c = 0; c < node->u.n.n_children; c++) 
+	    child[c] = search->direction & BRT_SEARCH_LEFT ? c : node->u.n.n_children - 1 - c;
+
+	for (c = 0; c < node->u.n.n_children-1; c++) {
+	    int p = search->direction & BRT_SEARCH_LEFT ? child[c] : child[c] - 1;
+	    struct kv_pair *pivot = node->u.n.childkeys[p];
+	    DBT pivotkey, pivotval;
+	    if (search->compare(search, 
+				toku_fill_dbt(&pivotkey, kv_pair_key(pivot), kv_pair_keylen(pivot)), 
+				brt->flags & TOKU_DB_DUPSORT ? toku_fill_dbt(&pivotval, kv_pair_val(pivot), kv_pair_vallen(pivot)): 0)) {
+		BOOL did_change_shape;
+		int r = brt_search_child(brt, node, child[c], search, newkey, newval, re, logger, omtcursor, &did_change_shape);
+		assert(r != EAGAIN);
+		if (r == 0) return r;
+		if (did_change_shape) goto again;
+	    }
+	}
     
-    /* check the first (left) or last (right) node if nothing has been found */
-    return brt_search_child(brt, node, child[c], search, newkey, newval, re, logger, omtcursor);
+	/* check the first (left) or last (right) node if nothing has been found */
+	BOOL did_change_shape; // ignore this
+	return brt_search_child(brt, node, child[c], search, newkey, newval, re, logger, omtcursor, &did_change_shape);
+    }
 }
 
 static int

newbrt/tests/Makefile

@@ -137,6 +137,7 @@ REGRESSION_TESTS = \
 	cachetable-debug-test \
 	cachetable-debug-test \
 	cachetable-unpin-and-remove-test \
+	dup-delete-all \
 	fifo-test \
 	list-test \
 	keyrange \

newbrt/tests/dup-delete-all.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2008 Tokutek Inc.  All rights reserved."
+
+/* Insert N duplicates into a BRT.
+ * Delete them with a single delete.
+ * Close the BRT.
+ * Check to see that the BRT is empty.
+ */
+
+#include "includes.h"
+#include "test.h"
+
+#include <unistd.h>
+
+static TOKUTXN const null_txn = 0;
+static DB * const null_db = 0;
+
+static void test_delete_all (void) {
+    char fname[]= __FILE__ ".brt";
+    u_int32_t limit =200;
+    unlink(fname);
+    CACHETABLE ct;
+    int r = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER);                                assert(r==0);
+    BRT t;
+    r = toku_brt_create(&t); assert(r==0);
+    r = toku_brt_set_flags(t, TOKU_DB_DUP + TOKU_DB_DUPSORT); assert(r == 0);
+    r = toku_brt_set_nodesize(t, 4096); assert(r == 0);
+    r = toku_brt_open(t, fname, fname, 0, 1, 1, ct, null_txn, (DB*)0); assert(r==0);
+    u_int32_t i;
+    for (i=0; i<limit; i++) {
+	char key[100];
+	char val[100];
+	snprintf(key, 100, "%03u", limit/2);
+	snprintf(val, 100, "%03u", i);
+	DBT k,v;
+	r = toku_brt_insert(t, toku_fill_dbt(&k, key, 1+strlen(key)), toku_fill_dbt(&v,val, 1+strlen(val)), null_txn);
+	assert(r == 0);
+    }
+
+    //printf("Initial insert done\n"); toku_dump_brt(stdout, t);
+
+    // Now reopen the DB to force non-leaf buffering.
+    r = toku_close_brt(t, 0);          assert(r==0);
+    r = toku_cachetable_close(&ct);    assert(r==0);
+    r = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER);                                assert(r==0);
+
+    r = toku_brt_create(&t); assert(r==0);
+    r = toku_brt_open(t, fname, fname, 0, 0, 0, ct, null_txn, (DB*)0); assert(r==0);
+
+    // Don't do a dump here, because that will warm the cachetable.  We want subsequent inserts to be buffered at the root.
+
+    // Insert some more stuff
+    if (1) {
+	u_int32_t j;
+	for (j=0; j<1; j++) {
+	    char key[100];
+	    char val[100];
+	    snprintf(key, 100, "%03u", limit/2);
+	    snprintf(val, 100, "%03u", limit+j);
+	    DBT k,v;
+	    r = toku_brt_insert(t, toku_fill_dbt(&k, key, 1+strlen(key)), toku_fill_dbt(&v,val, 1+strlen(val)), null_txn);
+	    assert(r == 0);
+	}
+    }
+
+    // Don't do a dump here, because that will warm the cachetable.
+
+    // Delete everything
+    {
+	char key[100];
+	DBT k;
+	snprintf(key, 100, "%03u", limit/2);
+	r = toku_brt_delete(t, toku_fill_dbt(&k, key, 1+strlen(key)), null_txn);
+	assert(r == 0);
+    }
+
+    //printf("Deleted\n"); toku_dump_brt(stdout, t);
+
+    // Now use a cursor to see if it is all empty
+    {
+	BRT_CURSOR cursor = 0;
+	r = toku_brt_cursor(t, &cursor, 0); assert(r==0);
+	DBT kbt, vbt;
+	toku_init_dbt(&kbt); kbt.flags = DB_DBT_MALLOC;
+	toku_init_dbt(&vbt); vbt.flags = DB_DBT_MALLOC;
+	r = toku_brt_cursor_get(cursor, &kbt, &vbt, DB_FIRST, null_txn);
+	assert(r == DB_NOTFOUND);
+
+	r = toku_brt_cursor_close(cursor);
+	assert(r==0);
+    }
+
+    //printf("Looked\n"); toku_dump_brt(stdout, t);
+
+    r = toku_close_brt(t, 0);          assert(r==0);
+    r = toku_cachetable_close(&ct);    assert(r==0);
+}
+
+int main (int argc , const char *argv[]) {
+    default_parse_args(argc, argv);
+
+    test_delete_all();
+    if (verbose) printf("test ok\n");
+    return 0;
+}
+