[t:3627], merge last of milestone 3 to main

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

newbrt/brt-internal.h

@@ -80,6 +80,54 @@ add_estimates (struct subtree_estimates *a, struct subtree_estimates *b) {
     a->dsize += b->dsize;
 }
 
+enum brtnode_fetch_type {
+    brtnode_fetch_none=1, 
+    brtnode_fetch_subset,
+    brtnode_fetch_all
+};
+
+struct brtnode_fetch_extra {
+    enum brtnode_fetch_type type;
+    // needed for reading a node off disk
+    struct brt_header *h;
+    // used in the case where type == brtnode_fetch_extra
+    // parameters needed to find out which child needs to be decompressed (so it can be read)
+    brt_search_t* search;
+    BRT brt;
+    // this value will be set during the fetch_callback call by toku_brtnode_fetch_callback or toku_brtnode_pf_req_callback
+    // thi callbacks need to evaluate this anyway, so we cache it here so the search code does not reevaluate it
+    int child_to_read;
+};
+
+static inline void fill_bfe_for_full_read(struct brtnode_fetch_extra *bfe, struct brt_header *h) {
+    bfe->type = brtnode_fetch_all;
+    bfe->h = h;
+    bfe->search = NULL;
+    bfe->brt = NULL;
+    bfe->child_to_read = -1;
+};
+
+static inline void fill_bfe_for_subset_read(
+    struct brtnode_fetch_extra *bfe, 
+    struct brt_header *h,
+    BRT brt,
+    brt_search_t* search
+    ) 
+{
+    bfe->type = brtnode_fetch_subset;
+    bfe->h = h;
+    bfe->search = search;
+    bfe->brt = brt;
+    bfe->child_to_read = -1;
+};
+
+static inline void fill_bfe_for_min_read(struct brtnode_fetch_extra *bfe, struct brt_header *h) {
+    bfe->type = brtnode_fetch_none;
+    bfe->h = h;
+    bfe->search = NULL;
+    bfe->brt = NULL;
+    bfe->child_to_read = -1;
+};
 
 struct brtnode_nonleaf_childinfo {
     FIFO         buffer;
@@ -168,6 +216,7 @@ struct brtnode {
     unsigned int    totalchildkeylens;
     struct kv_pair **childkeys;   /* Pivot keys.  Child 0's keys are <= childkeys[0].  Child 1's keys are <= childkeys[1].
                                                                         Child 1's keys are > childkeys[0]. */
+    u_int32_t bp_offset; // offset on disk to where the partitions start
     // array of brtnode partitions
     // each one is associated with a child
     // for internal nodes, the ith partition corresponds to the ith message buffer
@@ -274,7 +323,8 @@ int toku_serialize_rollback_log_to (int fd, BLOCKNUM blocknum, ROLLBACK_LOG_NODE
                                     struct brt_header *h, int n_workitems, int n_threads,
                                     BOOL for_checkpoint);
 int toku_deserialize_rollback_log_from (int fd, BLOCKNUM blocknum, u_int32_t fullhash, ROLLBACK_LOG_NODE *logp, struct brt_header *h);
-int toku_deserialize_brtnode_from (int fd, BLOCKNUM off, u_int32_t /*fullhash*/, BRTNODE *brtnode, struct brt_header *h);
+void toku_deserialize_bp_from_compressed(BRTNODE node, int childnum);
+int toku_deserialize_brtnode_from (int fd, BLOCKNUM off, u_int32_t /*fullhash*/, BRTNODE *brtnode, struct brtnode_fetch_extra* bfe);
 unsigned int toku_serialize_brtnode_size(BRTNODE node); /* How much space will it take? */
 int toku_keycompare (bytevec key1, ITEMLEN key1len, bytevec key2, ITEMLEN key2len);
 
@@ -361,15 +411,25 @@ struct pivot_bounds {
     struct kv_pair const * const upper_bound_inclusive; // NULL to indicate negative or positive infinity (which are in practice exclusive since there are now transfinite keys in messages).
 };
 
+int
+toku_brt_search_which_child(
+    BRT brt, 
+    BRTNODE node, 
+    brt_search_t *search
+    );
+u_int8_t 
+toku_brtnode_partition_state (struct brtnode_fetch_extra* bfe, int childnum);
 // logs the memory allocation, but not the creation of the new node
 void toku_create_new_brtnode (BRT t, BRTNODE *result, int height, int n_children);
 int toku_pin_brtnode (BRT brt, BLOCKNUM blocknum, u_int32_t fullhash,
 		      UNLOCKERS unlockers,
 		      ANCESTORS ancestors, struct pivot_bounds const * const pbounds,
+                      struct brtnode_fetch_extra *bfe,
 		      BRTNODE *node_p)
     __attribute__((__warn_unused_result__));
 void toku_pin_brtnode_holding_lock (BRT brt, BLOCKNUM blocknum, u_int32_t fullhash,
 				   ANCESTORS ancestors, struct pivot_bounds const * const pbounds,
+                                   struct brtnode_fetch_extra *bfe,
 				   BRTNODE *node_p);
 void toku_unpin_brtnode (BRT brt, BRTNODE node);
 unsigned int toku_brtnode_which_child (BRTNODE node , const DBT *k, BRT t)

newbrt/brt-test-helpers.c

@@ -77,6 +77,8 @@ int toku_testsetup_get_sersize(BRT brt, BLOCKNUM diskoff) // Return the size on
 {
     assert(testsetup_initialized);
     void *node_v;
+    struct brtnode_fetch_extra bfe;
+    fill_bfe_for_full_read(&bfe, brt->h);
     int r  = toku_cachetable_get_and_pin(
         brt->cf, diskoff, 
         toku_cachetable_hash(brt->cf, diskoff), 
@@ -87,7 +89,7 @@ int toku_testsetup_get_sersize(BRT brt, BLOCKNUM diskoff) // Return the size on
         toku_brtnode_pe_callback, 
         toku_brtnode_pf_req_callback,
         toku_brtnode_pf_callback,
-        brt->h, 
+        &bfe, 
         brt->h
         );
     assert(r==0);
@@ -102,6 +104,8 @@ int toku_testsetup_insert_to_leaf (BRT brt, BLOCKNUM blocknum, char *key, int ke
 
     assert(testsetup_initialized);
     
+    struct brtnode_fetch_extra bfe;
+    fill_bfe_for_full_read(&bfe, brt->h);
     r = toku_cachetable_get_and_pin(
         brt->cf, 
         blocknum, 
@@ -113,7 +117,7 @@ int toku_testsetup_insert_to_leaf (BRT brt, BLOCKNUM blocknum, char *key, int ke
 	toku_brtnode_pe_callback, 
         toku_brtnode_pf_req_callback,
         toku_brtnode_pf_callback,
-	brt->h, 
+	&bfe, 
 	brt->h
 	);
     if (r!=0) return r;
@@ -169,6 +173,8 @@ int toku_testsetup_insert_to_nonleaf (BRT brt, BLOCKNUM blocknum, enum brt_msg_t
 
     assert(testsetup_initialized);
 
+    struct brtnode_fetch_extra bfe;
+    fill_bfe_for_full_read(&bfe, brt->h);
     r = toku_cachetable_get_and_pin(
         brt->cf, 
         blocknum, 
@@ -180,7 +186,7 @@ int toku_testsetup_insert_to_nonleaf (BRT brt, BLOCKNUM blocknum, enum brt_msg_t
 	toku_brtnode_pe_callback, 
         toku_brtnode_pf_req_callback,
         toku_brtnode_pf_callback,
-	brt->h, 
+	&bfe, 
 	brt->h
 	);
     if (r!=0) return r;

newbrt/brt-verify.c

@@ -113,6 +113,8 @@ toku_verify_brtnode (BRT brt,
     
     u_int32_t fullhash = toku_cachetable_hash(brt->cf, blocknum);
     {
+        struct brtnode_fetch_extra bfe;
+        fill_bfe_for_full_read(&bfe, brt->h);
         int r = toku_cachetable_get_and_pin(
             brt->cf, 
             blocknum, 
@@ -124,7 +126,7 @@ toku_verify_brtnode (BRT brt,
             toku_brtnode_pe_callback, 
             toku_brtnode_pf_req_callback,
             toku_brtnode_pf_callback,
-            brt->h, 
+            &bfe, 
             brt->h
             );
         assert_zero(r); // this is a bad failure if it happens.

newbrt/brtdump.c

@@ -118,7 +118,9 @@ print_le (OMTVALUE lev, u_int32_t UU(idx), void *UU(v)) {
 static void
 dump_node (int f, BLOCKNUM blocknum, struct brt_header *h) {
     BRTNODE n;
-    int r = toku_deserialize_brtnode_from (f, blocknum, 0 /*pass zero for hash, it doesn't matter*/, &n, h);
+    struct brtnode_fetch_extra bfe;
+    fill_bfe_for_full_read(&bfe, h);
+    int r = toku_deserialize_brtnode_from (f, blocknum, 0 /*pass zero for hash, it doesn't matter*/, &n, &bfe);
     assert(r==0);
     assert(n!=0);
     printf("brtnode\n");
@@ -227,7 +229,9 @@ static int
 fragmentation_helper(BLOCKNUM b, int64_t size, int64_t UU(address), void *extra) {
     frag_help_extra *info = extra;
     BRTNODE n;
-    int r = toku_deserialize_brtnode_from(info->f, b, 0 /*pass zero for hash, it doesn't matter*/, &n, info->h);
+    struct brtnode_fetch_extra bfe;
+    fill_bfe_for_full_read(&bfe, info->h);
+    int r = toku_deserialize_brtnode_from(info->f, b, 0 /*pass zero for hash, it doesn't matter*/, &n, &bfe);
     if (r==0) {
         info->blocksizes += size;
         if (n->height == 0) {
@@ -266,14 +270,14 @@ get_unaligned_uint32(unsigned char *p) {
     return *(u_int32_t *)p;
 }
 
-struct sub_block {
+struct dump_sub_block {
   u_int32_t compressed_size;
   u_int32_t uncompressed_size;
   u_int32_t xsum;
 };
 
 static void
-sub_block_deserialize(struct sub_block *sb, unsigned char *sub_block_header) {
+sub_block_deserialize(struct dump_sub_block *sb, unsigned char *sub_block_header) {
     sb->compressed_size = toku_dtoh32(get_unaligned_uint32(sub_block_header+0));
     sb->uncompressed_size = toku_dtoh32(get_unaligned_uint32(sub_block_header+4));
     sb->xsum = toku_dtoh32(get_unaligned_uint32(sub_block_header+8));
@@ -288,7 +292,7 @@ verify_block(unsigned char *cp, u_int64_t file_offset, u_int64_t size) {
 
     unsigned char *sub_block_header = &cp[node_header];
     u_int32_t n_sub_blocks = toku_dtoh32(get_unaligned_uint32(&sub_block_header[0]));
-    u_int32_t header_length = node_header + n_sub_blocks * sizeof (struct sub_block);
+    u_int32_t header_length = node_header + n_sub_blocks * sizeof (struct dump_sub_block);
     header_length += sizeof (u_int32_t); // CRC
     if (header_length > size) {
         printf("header length too big: %u\n", header_length);
@@ -302,11 +306,11 @@ verify_block(unsigned char *cp, u_int64_t file_offset, u_int64_t size) {
     }
 
     // deserialize the sub block header
-    struct sub_block sub_block[n_sub_blocks];
+    struct dump_sub_block sub_block[n_sub_blocks];
     sub_block_header += sizeof (u_int32_t);
     for (u_int32_t i = 0 ; i < n_sub_blocks; i++) {
         sub_block_deserialize(&sub_block[i], sub_block_header);
-        sub_block_header += sizeof (struct sub_block);
+        sub_block_header += sizeof (struct dump_sub_block);
     }
 
     // verify the sub block header

newbrt/cachetable.c

@@ -1544,8 +1544,12 @@ int toku_cachetable_get_and_pin (
                     cachetable_waittime += get_tnow() - t0;
                 }
 	        t0 = get_tnow();
+                long old_size = p->size;
                 long size = 0;
                 int r = pf_callback(p->value, read_extraargs, &size);
+                p->size = size;
+                ct->size_current += size;
+                ct->size_current -= old_size;
                 lazy_assert_zero(r);                
                 cachetable_waittime += get_tnow() - t0;
                 rwlock_write_unlock(&p->rwlock);
@@ -1816,11 +1820,18 @@ int toku_cachetable_get_and_pin_nonblocking (
                         if (ct->ydb_unlock_callback) ct->ydb_unlock_callback();
                         // Now wait for the I/O to occur.
                         rwlock_write_lock(&p->rwlock, ct->mutex);
+                        cachetable_unlock(ct);
+                        long old_size = p->size;
                         long size = 0;
                         int r = pf_callback(p->value, read_extraargs, &size);
                         lazy_assert_zero(r);
+                        cachetable_lock(ct);
+                        p->size = size;
+                        ct->size_current += size;
+                        ct->size_current -= old_size;
                         rwlock_write_unlock(&p->rwlock);
                         cachetable_unlock(ct);
+                        if (ct->ydb_lock_callback) ct->ydb_lock_callback();
                         return TOKUDB_TRY_AGAIN;
                     }
                     rwlock_read_lock(&p->rwlock, ct->mutex);

newbrt/tests/brt-serialize-test.c

@@ -160,7 +160,9 @@ test_serialize_leaf_with_large_pivots(void) {
     r = toku_serialize_brtnode_to(fd, make_blocknum(20), &sn, brt->h, 1, 1, FALSE);
     assert(r==0);
 
-    r = toku_deserialize_brtnode_from(fd, make_blocknum(20), 0/*pass zero for hash*/, &dn, brt_h);
+    struct brtnode_fetch_extra bfe;
+    fill_bfe_for_full_read(&bfe, brt_h);
+    r = toku_deserialize_brtnode_from(fd, make_blocknum(20), 0/*pass zero for hash*/, &dn, &bfe);
     assert(r==0);
 
     assert(dn->thisnodename.b==20);
@@ -281,7 +283,9 @@ test_serialize_leaf_with_many_rows(void) {
     r = toku_serialize_brtnode_to(fd, make_blocknum(20), &sn, brt->h, 1, 1, FALSE);
     assert(r==0);
 
-    r = toku_deserialize_brtnode_from(fd, make_blocknum(20), 0/*pass zero for hash*/, &dn, brt_h);
+    struct brtnode_fetch_extra bfe;
+    fill_bfe_for_full_read(&bfe, brt_h);
+    r = toku_deserialize_brtnode_from(fd, make_blocknum(20), 0/*pass zero for hash*/, &dn, &bfe);
     assert(r==0);
 
     assert(dn->thisnodename.b==20);
@@ -408,7 +412,9 @@ test_serialize_leaf_with_large_rows(void) {
     r = toku_serialize_brtnode_to(fd, make_blocknum(20), &sn, brt->h, 1, 1, FALSE);
     assert(r==0);
 
-    r = toku_deserialize_brtnode_from(fd, make_blocknum(20), 0/*pass zero for hash*/, &dn, brt_h);
+    struct brtnode_fetch_extra bfe;
+    fill_bfe_for_full_read(&bfe, brt_h);
+    r = toku_deserialize_brtnode_from(fd, make_blocknum(20), 0/*pass zero for hash*/, &dn, &bfe);
     assert(r==0);
 
     assert(dn->thisnodename.b==20);
@@ -538,7 +544,9 @@ test_serialize_leaf_with_empty_basement_nodes(void) {
     r = toku_serialize_brtnode_to(fd, make_blocknum(20), &sn, brt->h, 1, 1, FALSE);
     assert(r==0);
 
-    r = toku_deserialize_brtnode_from(fd, make_blocknum(20), 0/*pass zero for hash*/, &dn, brt_h);
+    struct brtnode_fetch_extra bfe;
+    fill_bfe_for_full_read(&bfe, brt_h);
+    r = toku_deserialize_brtnode_from(fd, make_blocknum(20), 0/*pass zero for hash*/, &dn, &bfe);
     assert(r==0);
 
     assert(dn->thisnodename.b==20);
@@ -667,7 +675,9 @@ test_serialize_leaf(void) {
     r = toku_serialize_brtnode_to(fd, make_blocknum(20), &sn, brt->h, 1, 1, FALSE);
     assert(r==0);
 
-    r = toku_deserialize_brtnode_from(fd, make_blocknum(20), 0/*pass zero for hash*/, &dn, brt_h);
+    struct brtnode_fetch_extra bfe;
+    fill_bfe_for_full_read(&bfe, brt_h);
+    r = toku_deserialize_brtnode_from(fd, make_blocknum(20), 0/*pass zero for hash*/, &dn, &bfe);
     assert(r==0);
 
     assert(dn->thisnodename.b==20);
@@ -814,7 +824,9 @@ test_serialize_nonleaf(void) {
     assert(sn.max_msn_applied_to_node_in_memory.msn == TESTMSNMEMVAL);
     
 
-    r = toku_deserialize_brtnode_from(fd, make_blocknum(20), 0/*pass zero for hash*/, &dn, brt_h);
+    struct brtnode_fetch_extra bfe;
+    fill_bfe_for_full_read(&bfe, brt_h);
+    r = toku_deserialize_brtnode_from(fd, make_blocknum(20), 0/*pass zero for hash*/, &dn, &bfe);
     assert(r==0);
 
     assert(dn->thisnodename.b==20);