[t:3928] serialize freshness bit too

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

newbrt/brt-internal.h

@@ -37,7 +37,7 @@ C_BEGIN
 enum { TREE_FANOUT = BRT_FANOUT };
 enum { KEY_VALUE_OVERHEAD = 8 }; /* Must store the two lengths. */
 enum { OMT_ITEM_OVERHEAD = 0 }; /* No overhead for the OMT item.  The PMA needed to know the idx, but the OMT doesn't. */
-enum { BRT_CMD_OVERHEAD = (1 + sizeof(MSN))     // the type plus MSN
+enum { BRT_CMD_OVERHEAD = (2 + sizeof(MSN))     // the type plus freshness plus MSN
 };
 
 enum { BRT_DEFAULT_NODE_SIZE = 1 << 22 };

newbrt/brt-serialize.c

@@ -260,10 +260,11 @@ serialize_nonleaf_childinfo(NONLEAF_CHILDINFO bnc, struct wbuf *wb)
     // serialize the FIFO, first the number of entries, then the elements
     wbuf_nocrc_int(wb, toku_bnc_n_entries(bnc));
     FIFO_ITERATE(
-        bnc->buffer, key, keylen, data, datalen, type, msn, xids, UU(is_fresh),
+        bnc->buffer, key, keylen, data, datalen, type, msn, xids, is_fresh,
         {
             invariant((int)type>=0 && type<256);
             wbuf_nocrc_char(wb, (unsigned char)type);
+            wbuf_nocrc_char(wb, (unsigned char)is_fresh);
             wbuf_MSN(wb, msn);
             wbuf_nocrc_xids(wb, xids);
             wbuf_nocrc_bytes(wb, key, keylen);
@@ -869,12 +870,13 @@ deserialize_child_buffer(NONLEAF_CHILDINFO bnc, struct rbuf *rbuf,
     int r;
     int n_bytes_in_buffer = 0;
     int n_in_this_buffer = rbuf_int(rbuf);
-    void **offsets;
+    void **fresh_offsets, **stale_offsets;
     void **broadcast_offsets;
-    int noffsets = 0;
+    int nfresh = 0, nstale = 0;
     int nbroadcast_offsets = 0;
     if (cmp) {
-        MALLOC_N(n_in_this_buffer, offsets);
+        MALLOC_N(n_in_this_buffer, stale_offsets);
+        MALLOC_N(n_in_this_buffer, fresh_offsets);
         MALLOC_N(n_in_this_buffer, broadcast_offsets);
     }
     for (int i = 0; i < n_in_this_buffer; i++) {
@@ -883,6 +885,7 @@ deserialize_child_buffer(NONLEAF_CHILDINFO bnc, struct rbuf *rbuf,
         // this is weird but it's necessary to pass icc and gcc together
         unsigned char ctype = rbuf_char(rbuf);
         enum brt_msg_type type = (enum brt_msg_type) ctype;
+        bool is_fresh = rbuf_char(rbuf);
         MSN msn = rbuf_msn(rbuf);
         XIDS xids;
         xids_create_from_buffer(rbuf, &xids);
@@ -892,8 +895,13 @@ deserialize_child_buffer(NONLEAF_CHILDINFO bnc, struct rbuf *rbuf,
         long *dest;
         if (cmp) {
             if (brt_msg_type_applies_once(type)) {
-                dest = (long *) &offsets[noffsets];
-                noffsets++;
+                if (is_fresh) {
+                    dest = (long *) &fresh_offsets[nfresh];
+                    nfresh++;
+                } else {
+                    dest = (long *) &stale_offsets[nstale];
+                    nstale++;
+                }
             } else if (brt_msg_type_applies_all(type) || brt_msg_type_does_nothing(type)) {
                 dest = (long *) &broadcast_offsets[nbroadcast_offsets];
                 nbroadcast_offsets++;
@@ -903,7 +911,7 @@ deserialize_child_buffer(NONLEAF_CHILDINFO bnc, struct rbuf *rbuf,
         } else {
             dest = NULL;
         }
-        r = toku_fifo_enq(bnc->buffer, key, keylen, val, vallen, type, msn, xids, true, dest); /* Copies the data into the fifo */
+        r = toku_fifo_enq(bnc->buffer, key, keylen, val, vallen, type, msn, xids, is_fresh, dest); /* Copies the data into the fifo */
         lazy_assert_zero(r);
         n_bytes_in_buffer += keylen + vallen + KEY_VALUE_OVERHEAD + BRT_CMD_OVERHEAD + xids_get_serialize_size(xids);
         //printf("Inserted\n");
@@ -913,10 +921,15 @@ deserialize_child_buffer(NONLEAF_CHILDINFO bnc, struct rbuf *rbuf,
 
     if (cmp) {
         struct toku_fifo_entry_key_msn_cmp_extra extra = { .cmp_extra = cmp_extra, .cmp = cmp, .fifo = bnc->buffer };
-        r = mergesort_r(offsets, noffsets, sizeof offsets[0], &extra, toku_fifo_entry_key_msn_cmp);
+        r = mergesort_r(fresh_offsets, nfresh, sizeof fresh_offsets[0], &extra, toku_fifo_entry_key_msn_cmp);
         assert_zero(r);
         toku_omt_destroy(&bnc->fresh_message_tree);
-        r = toku_omt_create_steal_sorted_array(&bnc->fresh_message_tree, &offsets, noffsets, n_in_this_buffer);
+        r = toku_omt_create_steal_sorted_array(&bnc->fresh_message_tree, &fresh_offsets, nfresh, n_in_this_buffer);
+        assert_zero(r);
+        r = mergesort_r(stale_offsets, nstale, sizeof stale_offsets[0], &extra, toku_fifo_entry_key_msn_cmp);
+        assert_zero(r);
+        toku_omt_destroy(&bnc->stale_message_tree);
+        r = toku_omt_create_steal_sorted_array(&bnc->stale_message_tree, &stale_offsets, nstale, n_in_this_buffer);
         assert_zero(r);
         toku_omt_destroy(&bnc->broadcast_list);
         r = toku_omt_create_steal_sorted_array(&bnc->broadcast_list, &broadcast_offsets, nbroadcast_offsets, n_in_this_buffer);

newbrt/tests/brt-serialize-test.c

@@ -1113,7 +1113,7 @@ test_serialize_nonleaf(enum brtnode_verify_type bft) {
     CKERR(r);
 
     r = toku_bnc_insert_msg(BNC(&sn, 0), "a", 2, "aval", 5, BRT_NONE, next_dummymsn(), xids_0, true, NULL, string_key_cmp); assert_zero(r);
-    r = toku_bnc_insert_msg(BNC(&sn, 0), "b", 2, "bval", 5, BRT_NONE, next_dummymsn(), xids_123, true, NULL, string_key_cmp); assert_zero(r);
+    r = toku_bnc_insert_msg(BNC(&sn, 0), "b", 2, "bval", 5, BRT_NONE, next_dummymsn(), xids_123, false, NULL, string_key_cmp); assert_zero(r);
     r = toku_bnc_insert_msg(BNC(&sn, 1), "x", 2, "xval", 5, BRT_NONE, next_dummymsn(), xids_234, true, NULL, string_key_cmp); assert_zero(r);
     BNC(&sn, 0)->n_bytes_in_buffer = 2*(BRT_CMD_OVERHEAD+KEY_VALUE_OVERHEAD+2+5) + xids_get_serialize_size(xids_0) + xids_get_serialize_size(xids_123);
     BNC(&sn, 1)->n_bytes_in_buffer = 1*(BRT_CMD_OVERHEAD+KEY_VALUE_OVERHEAD+2+5) + xids_get_serialize_size(xids_234);
@@ -1194,7 +1194,7 @@ test_serialize_nonleaf(enum brtnode_verify_type bft) {
     assert(strcmp(dest_key, "a") == 0);
     assert(strcmp(src_val, "aval") == 0);
     assert(strcmp(dest_val, "aval") == 0);
-    assert(dest_is_fresh);
+    assert(src_is_fresh == dest_is_fresh);
     r = toku_fifo_deq(src_fifo_1);
     assert(r==0);
     r = toku_fifo_deq(dest_fifo_1);
@@ -1213,7 +1213,7 @@ test_serialize_nonleaf(enum brtnode_verify_type bft) {
     assert(strcmp(dest_key, "b") == 0);
     assert(strcmp(src_val, "bval") == 0);
     assert(strcmp(dest_val, "bval") == 0);
-    assert(dest_is_fresh);
+    assert(src_is_fresh == dest_is_fresh);
     r = toku_fifo_deq(src_fifo_1);
     assert(r==0);
     r = toku_fifo_deq(dest_fifo_1);
@@ -1237,7 +1237,7 @@ test_serialize_nonleaf(enum brtnode_verify_type bft) {
     assert(strcmp(dest_key, "x") == 0);
     assert(strcmp(src_val, "xval") == 0);
     assert(strcmp(dest_val, "xval") == 0);
-    assert(dest_is_fresh);
+    assert(src_is_fresh == dest_is_fresh);
     r = toku_fifo_deq(src_fifo_2);
     assert(r==0);
     r = toku_fifo_deq(dest_fifo_2);