merge 1766b to main. addresses #1766

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

newbrt/brt-internal.h

@@ -281,7 +281,7 @@ struct brt_cursor {
 };
 
 // logs the memory allocation, but not the creation of the new node
-int toku_create_new_brtnode (BRT t, BRTNODE *result, int height);
+int toku_create_new_brtnode (BRT t, BRTNODE *result, int height, size_t mpsize);
 int toku_unpin_brtnode (BRT brt, BRTNODE node);
 unsigned int toku_brtnode_which_child (BRTNODE node , DBT *k, DBT *d, BRT t);
 

newbrt/brt-test-helpers.c

@@ -8,7 +8,7 @@ int toku_testsetup_leaf(BRT brt, BLOCKNUM *blocknum) {
     BRTNODE node;
     int r = toku_read_brt_header_and_store_in_cachefile(brt->cf, &brt->h);
     if (r!=0) return r;
-    toku_create_new_brtnode(brt, &node, 0);
+    toku_create_new_brtnode(brt, &node, 0, 0);
 
     *blocknum = node->thisnodename;
     r = toku_unpin_brtnode(brt, node);
@@ -22,7 +22,7 @@ int toku_testsetup_nonleaf (BRT brt, int height, BLOCKNUM *blocknum, int n_child
     assert(n_children<=BRT_FANOUT);
     int r = toku_read_brt_header_and_store_in_cachefile(brt->cf, &brt->h);
     if (r!=0) return r;
-    toku_create_new_brtnode(brt, &node, height);
+    toku_create_new_brtnode(brt, &node, height, 0);
     node->u.n.n_children=n_children;
     MALLOC_N(n_children+1, node->u.n.childinfos);
     MALLOC_N(n_children,   node->u.n.childkeys);

newbrt/brt.c

@@ -195,8 +195,10 @@ static enum reactivity
 get_leaf_reactivity (BRTNODE node) {
     assert(node->height==0);
     unsigned int size = toku_serialize_brtnode_size(node);
-    if (size     > node->nodesize) return RE_FISSIBLE;
-    if ((size*4) < node->nodesize) return RE_FUSIBLE;
+    if (size     > node->nodesize && toku_omt_size(node->u.l.buffer) > 1) 
+        return RE_FISSIBLE;
+    if ((size*4) < node->nodesize)     
+        return RE_FUSIBLE;
     return RE_STABLE;
 }
 
@@ -649,7 +651,7 @@ toku_brtheader_free (struct brt_header *h) {
 }
 
 static void
-initialize_empty_brtnode (BRT t, BRTNODE n, BLOCKNUM nodename, int height)
+initialize_empty_brtnode (BRT t, BRTNODE n, BLOCKNUM nodename, int height, size_t suggest_mpsize)
 // Effect: Fill in N as an empty brtnode.
 {
     n->tag = TYP_BRTNODE;
@@ -680,7 +682,10 @@ initialize_empty_brtnode (BRT t, BRTNODE n, BLOCKNUM nodename, int height)
         r = toku_leaflock_borrow(&n->u.l.leaflock);
         assert(r==0);
         {
-            u_int32_t mpsize = mp_pool_size_for_nodesize(n->nodesize);
+            // mpsize = max(suggest_mpsize, mp_pool_size_for_nodesize)
+            size_t mpsize = mp_pool_size_for_nodesize(n->nodesize);
+            if (mpsize < suggest_mpsize) 
+                mpsize = suggest_mpsize;
             void *mp = toku_malloc(mpsize);
             assert(mp);
             toku_mempool_init(&n->u.l.buffer_mempool, mp, mpsize);
@@ -711,7 +716,7 @@ brt_init_new_root(BRT brt, BRTNODE nodea, BRTNODE nodeb, DBT splitk, CACHEKEY *r
     newroot->ever_been_written = 0;
     toku_log_changeunnamedroot(logger, (LSN*)0, 0, toku_cachefile_filenum(brt->cf), *rootp, newroot_diskoff);
     *rootp=newroot_diskoff;
-    initialize_empty_brtnode (brt, newroot, newroot_diskoff, new_height);
+    initialize_empty_brtnode (brt, newroot, newroot_diskoff, new_height, 0);
     //printf("new_root %lld %d %lld %lld\n", newroot_diskoff, newroot->height, nodea->thisnodename, nodeb->thisnodename);
     newroot->u.n.n_children=2;
     MALLOC_N(3, newroot->u.n.childinfos);
@@ -749,7 +754,7 @@ brt_init_new_root(BRT brt, BRTNODE nodea, BRTNODE nodeb, DBT splitk, CACHEKEY *r
 }
 
 // logs the memory allocation, but not the creation of the new node
-int toku_create_new_brtnode (BRT t, BRTNODE *result, int height) {
+int toku_create_new_brtnode (BRT t, BRTNODE *result, int height, size_t mpsize) {
     TAGMALLOC(BRTNODE, n);
     int r;
     BLOCKNUM name;
@@ -757,7 +762,7 @@ int toku_create_new_brtnode (BRT t, BRTNODE *result, int height) {
     assert(n);
     assert(t->h->nodesize>0);
     n->ever_been_written = 0;
-    initialize_empty_brtnode(t, n, name, height);
+    initialize_empty_brtnode(t, n, name, height, mpsize);
     *result = n;
     assert(n->nodesize>0);
     //    n->brt            = t;
@@ -791,7 +796,7 @@ brtleaf_split (BRT t, BRTNODE node, BRTNODE *nodea, BRTNODE *nodeb, DBT *splitk)
 
     assert(node->height==0);
     assert(t->h->nodesize>=node->nodesize); /* otherwise we might be in trouble because the nodesize shrank. */
-    toku_create_new_brtnode(t, &B, 0);
+    toku_create_new_brtnode(t, &B, 0, toku_mempool_get_size(&node->u.l.buffer_mempool));
     assert(B->nodesize>0);
     assert(node->nodesize>0);
     //printf("%s:%d A is at %lld\n", __FILE__, __LINE__, A->thisnodename);
@@ -940,7 +945,7 @@ brt_nonleaf_split (BRT t, BRTNODE node, BRTNODE *nodea, BRTNODE *nodeb, DBT *spl
     assert(node->height>0);
     assert(node->u.n.n_children>=2); // Otherwise, how do we split?  We need at least two children to split. */
     assert(t->h->nodesize>=node->nodesize); /* otherwise we might be in trouble because the nodesize shrank. */
-    toku_create_new_brtnode(t, &B, node->height);
+    toku_create_new_brtnode(t, &B, node->height, 0);
     MALLOC_N(n_children_in_b+1, B->u.n.childinfos);
     MALLOC_N(n_children_in_b, B->u.n.childkeys);
     B->u.n.n_children   =n_children_in_b;
@@ -2770,7 +2775,7 @@ static int setup_initial_brt_root_node (BRT t, BLOCKNUM blocknum) {
     assert(node);
     node->ever_been_written = 0;
     //printf("%s:%d\n", __FILE__, __LINE__);
-    initialize_empty_brtnode(t, node, blocknum, 0);
+    initialize_empty_brtnode(t, node, blocknum, 0, 0);
     //    node->brt = t;
     if (0) {
         printf("%s:%d for tree %p node %p mdict_create--> %p\n", __FILE__, __LINE__, t, node, node->u.l.buffer);

src/ydb.c

@@ -3113,12 +3113,12 @@ db_put_check_size_constraints(DB *db, DBT *key, DBT *val) {
     u_int32_t limit;
 
     if (dupsort) {
-        limit = nodesize / (2*BRT_FANOUT-1);
-        if (key->size + val->size >= limit)
+        limit = nodesize / BRT_FANOUT;
+        if (key->size + val->size > limit)
             r = toku_ydb_do_error(db->dbenv, EINVAL, "The largest (key + val) item allowed is %u bytes", limit-1);
     } else {
-        limit = nodesize / (3*BRT_FANOUT-1);
-        if (key->size >= limit || val->size >= limit)
+        limit = nodesize / BRT_FANOUT;
+        if (key->size > limit || val->size > nodesize)
             r = toku_ydb_do_error(db->dbenv, EINVAL, "The largest key or val item allowed is %u bytes", limit-1);
     }
     return r;