refs #46, LOTS of refactoring done.

Isolate mempool and OMT into a new class, bndata.
Remove key from the leafentry.

ft/CMakeLists.txt

@@ -26,6 +26,7 @@ set(FT_SOURCES
   background_job_manager
   block_allocator
   block_table
+  bndata
   cachetable
   checkpoint
   compress

ft/ft-internal.h

@@ -114,6 +114,7 @@ PATENT RIGHTS GRANT:
 #include "compress.h"
 #include <util/mempool.h>
 #include <util/omt.h>
+#include "bndata.h"
 
 #ifndef FT_FANOUT
 #define FT_FANOUT 16
@@ -244,10 +245,7 @@ uint32_t get_leaf_num_entries(FTNODE node);
 
 // data of an available partition of a leaf ftnode
 struct ftnode_leaf_basement_node {
-    OMT buffer;                     // pointers to individual leaf entries
-    struct mempool buffer_mempool;  // storage for all leaf entries
-    unsigned int n_bytes_in_buffer; // How many bytes to represent the OMT (including the per-key overheads, ...
-                                    // ... but not including the overheads for the node. 
+    bn_data data_buffer;
     unsigned int seqinsert;         // number of sequential inserts to this leaf 
     MSN max_msn_applied;            // max message sequence number applied
     bool stale_ancestor_messages_applied;
@@ -450,9 +448,8 @@ static inline void set_BSB(FTNODE node, int i, SUB_BLOCK sb) {
 // ftnode leaf basementnode macros, 
 #define BLB_MAX_MSN_APPLIED(node,i) (BLB(node,i)->max_msn_applied)
 #define BLB_MAX_DSN_APPLIED(node,i) (BLB(node,i)->max_dsn_applied)
-#define BLB_BUFFER(node,i) (BLB(node,i)->buffer)
-#define BLB_BUFFER_MEMPOOL(node,i) (BLB(node,i)->buffer_mempool)
-#define BLB_NBYTESINBUF(node,i) (BLB(node,i)->n_bytes_in_buffer)
+#define BLB_DATA(node,i) (&(BLB(node,i)->data_buffer))
+#define BLB_NBYTESINDATA(node,i) (BLB_DATA(node,i)->get_disk_size())
 #define BLB_SEQINSERT(node,i) (BLB(node,i)->seqinsert)
 
 /* pivot flags  (must fit in 8 bits) */
@@ -742,6 +739,7 @@ bool toku_ftnode_pf_req_callback(void* ftnode_pv, void* read_extraargs);
 int toku_ftnode_pf_callback(void* ftnode_pv, void* UU(disk_data), void* read_extraargs, int fd, PAIR_ATTR* sizep);
 int toku_ftnode_cleaner_callback( void *ftnode_pv, BLOCKNUM blocknum, uint32_t fullhash, void *extraargs);
 void toku_evict_bn_from_memory(FTNODE node, int childnum, FT h);
+BASEMENTNODE toku_detach_bn(FTNODE node, int childnum);
 
 // Given pinned node and pinned child, split child into two
 // and update node with information about its new child.
@@ -773,17 +771,6 @@ static inline CACHETABLE_WRITE_CALLBACK get_write_callbacks_for_node(FT h) {
 
 static const FTNODE null_ftnode=0;
 
-// Values to be used to update ftcursor if a search is successful.
-struct ft_cursor_leaf_info_to_be {
-    uint32_t index;
-    OMT       omt;
-};
-
-// Values to be used to pin a leaf for shortcut searches
-struct ft_cursor_leaf_info {
-    struct ft_cursor_leaf_info_to_be  to_be;
-};
-
 /* a brt cursor is represented as a kv pair in a tree */
 struct ft_cursor {
     struct toku_list cursors_link;
@@ -799,7 +786,6 @@ struct ft_cursor {
     int out_of_range_error;
     int direction;
     TOKUTXN ttxn;
-    struct ft_cursor_leaf_info  leaf_info;
 };
 
 //
@@ -1041,23 +1027,9 @@ int toku_testsetup_insert_to_leaf (FT_HANDLE brt, BLOCKNUM, const char *key, int
 int toku_testsetup_insert_to_nonleaf (FT_HANDLE brt, BLOCKNUM, enum ft_msg_type, const char *key, int keylen, const char *val, int vallen);
 void toku_pin_node_with_min_bfe(FTNODE* node, BLOCKNUM b, FT_HANDLE t);
 
-// These two go together to do lookups in a ftnode using the keys in a command.
-struct cmd_leafval_heaviside_extra {
-    ft_compare_func compare_fun;
-    DESCRIPTOR desc;
-    DBT const * const key;
-};
-int toku_cmd_leafval_heaviside (OMTVALUE leafentry, void *extra)
-    __attribute__((__warn_unused_result__));
-
 // toku_ft_root_put_cmd() accepts non-constant cmd because this is where we set the msn
 void toku_ft_root_put_cmd(FT h, FT_MSG_S * cmd, TXNID oldest_referenced_xid, GC_INFO gc_info);
 
-void *mempool_malloc_from_omt(OMT *omtp, struct mempool *mp, size_t size, void **maybe_free);
-// Effect: Allocate a new object of size SIZE in MP.  If MP runs out of space, allocate new a new mempool space, and copy all the items
-//  from the OMT (which items refer to items in the old mempool) into the new mempool.
-//  If MAYBE_FREE is NULL then free the old mempool's space.
-//  Otherwise, store the old mempool's space in maybe_free.
 void
 toku_get_node_for_verify(
     BLOCKNUM blocknum,

ft/ft-search.h

@@ -103,7 +103,7 @@ struct ft_search;
    the compare function should be a step function from 0 to 1 for a left to right search
    and 1 to 0 for a right to left search */
 
-typedef int (*ft_search_compare_func_t)(struct ft_search */*so*/, DBT *);
+typedef int (*ft_search_compare_func_t)(const struct ft_search &, const DBT *);
 
 /* the search object contains the compare function, search direction, and the kv pair that
    is used in the compare function.  the context is the user's private data */

ft/ft-verify.cc

@@ -107,24 +107,6 @@ compare_pairs (FT_HANDLE brt, const DBT *a, const DBT *b) {
     return cmp;
 }
 
-static int 
-compare_leafentries (FT_HANDLE brt, LEAFENTRY a, LEAFENTRY b) {
-    DBT x,y;
-    FAKE_DB(db, &brt->ft->cmp_descriptor);
-    int cmp = brt->ft->compare_fun(&db,
-                               toku_fill_dbt(&x, le_key(a), le_keylen(a)),
-                               toku_fill_dbt(&y, le_key(b), le_keylen(b)));
-    return cmp;
-}
-
-static int 
-compare_pair_to_leafentry (FT_HANDLE brt, const DBT *a, LEAFENTRY b) {
-    DBT y;
-    FAKE_DB(db, &brt->ft->cmp_descriptor);
-    int cmp = brt->ft->compare_fun(&db, a, toku_fill_dbt(&y, le_key(b), le_keylen(b)));
-    return cmp;
-}
-
 static int 
 compare_pair_to_key (FT_HANDLE brt, const DBT *a, bytevec key, ITEMLEN keylen) {
     DBT y;
@@ -166,12 +148,12 @@ verify_msg_in_child_buffer(FT_HANDLE brt, enum ft_msg_type type, MSN msn, byteve
     return result;
 }
 
-static LEAFENTRY 
-get_ith_leafentry (BASEMENTNODE bn, int i) {
-    OMTVALUE le_v;
-    int r = toku_omt_fetch(bn->buffer, i, &le_v);
-    invariant(r == 0); // this is a bad failure if it happens.
-    return (LEAFENTRY)le_v;
+static DBT
+get_ith_key_dbt (BASEMENTNODE bn, int i) {
+    DBT kdbt;
+    int r = bn->data_buffer.fetch_le_key_and_len(i, &kdbt.size, &kdbt.data);
+    invariant_zero(r); // this is a bad failure if it happens.
+    return kdbt;
 }
 
 #define VERIFY_ASSERTION(predicate, i, string) ({                                                                              \
@@ -441,26 +423,26 @@ toku_verify_ftnode_internal(FT_HANDLE brt,
         }
         else {
             BASEMENTNODE bn = BLB(node, i);
-            for (uint32_t j = 0; j < toku_omt_size(bn->buffer); j++) {
+            for (uint32_t j = 0; j < bn->data_buffer.omt_size(); j++) {
                 VERIFY_ASSERTION((rootmsn.msn >= this_msn.msn), 0, "leaf may have latest msn, but cannot be greater than root msn");
-                LEAFENTRY le = get_ith_leafentry(bn, j);
+                DBT kdbt = get_ith_key_dbt(bn, j);
                 if (curr_less_pivot) {
-                    int compare = compare_pair_to_leafentry(brt, curr_less_pivot, le);
+                    int compare = compare_pairs(brt, curr_less_pivot, &kdbt);
                     VERIFY_ASSERTION(compare < 0, j, "The leafentry is >= the lower-bound pivot");
                 }
                 if (curr_geq_pivot) {
-                    int compare = compare_pair_to_leafentry(brt, curr_geq_pivot, le);
+                    int compare = compare_pairs(brt, curr_geq_pivot, &kdbt);
                     VERIFY_ASSERTION(compare >= 0, j, "The leafentry is < the upper-bound pivot");
                 }
                 if (0 < j) {
-                    LEAFENTRY prev_le = get_ith_leafentry(bn, j-1);
-                    int compare = compare_leafentries(brt, prev_le, le);
+                    DBT prev_key_dbt = get_ith_key_dbt(bn, j-1);
+                    int compare = compare_pairs(brt, &prev_key_dbt, &kdbt);
                     VERIFY_ASSERTION(compare < 0, j, "Adjacent leafentries are out of order");
                 }
             }
         }
-    } 
-    
+    }
+
 done:
     return result;
 }

ft/ft.cc

@@ -1028,11 +1028,12 @@ struct garbage_helper_extra {
 };
 
 static int
-garbage_leafentry_helper(OMTVALUE v, uint32_t UU(idx), void *extra) {
-    struct garbage_helper_extra *CAST_FROM_VOIDP(info, extra);
-    LEAFENTRY CAST_FROM_VOIDP(le, v);
-    info->total_space += leafentry_disksize(le);
-    info->used_space += LE_CLEAN_MEMSIZE(le_latest_keylen(le), le_latest_vallen(le));
+garbage_leafentry_helper(const void* key UU(), const uint32_t keylen, const LEAFENTRY & le, uint32_t UU(idx), struct garbage_helper_extra * const info) {
+    //TODO #warning need to reanalyze for split
+    info->total_space += leafentry_disksize(le) + keylen + sizeof(keylen);
+    if (!le_latest_is_del(le)) {
+        info->used_space += LE_CLEAN_MEMSIZE(le_latest_vallen(le)) + keylen + sizeof(keylen);
+    }
     return 0;
 }
 
@@ -1052,8 +1053,8 @@ garbage_helper(BLOCKNUM blocknum, int64_t UU(size), int64_t UU(address), void *e
         goto exit;
     }
     for (int i = 0; i < node->n_children; ++i) {
-        BASEMENTNODE bn = BLB(node, i);
-        r = toku_omt_iterate(bn->buffer, garbage_leafentry_helper, info);
+        BN_DATA bd = BLB_DATA(node, i);
+        r = bd->omt_iterate<struct garbage_helper_extra, garbage_leafentry_helper>(info);
         if (r != 0) {
             goto exit;
         }

ft/ftloader.cc

@@ -1236,9 +1236,9 @@ static TXNID leafentry_xid(FTLOADER bl, int which_db) {
 size_t ft_loader_leafentry_size(size_t key_size, size_t val_size, TXNID xid) {
     size_t s = 0;
     if (xid == TXNID_NONE)
-        s = LE_CLEAN_MEMSIZE(key_size, val_size);
+        s = LE_CLEAN_MEMSIZE(val_size) + key_size + sizeof(uint32_t);
     else
-        s = LE_MVCC_COMMITTED_MEMSIZE(key_size, val_size);
+        s = LE_MVCC_COMMITTED_MEMSIZE(val_size) + key_size + sizeof(uint32_t);
     return s;
 }
 
@@ -2906,7 +2906,7 @@ static void add_pair_to_leafnode (struct leaf_buf *lbuf, unsigned char *key, int
     // #3588 TODO just make a clean ule and append it to the omt
     // #3588 TODO can do the rebalancing here and avoid a lot of work later
     FTNODE leafnode = lbuf->node;
-    uint32_t idx = toku_omt_size(BLB_BUFFER(leafnode, 0));
+    uint32_t idx = BLB_DATA(leafnode, 0)->omt_size();
     DBT thekey = { .data = key, .size = (uint32_t) keylen }; 
     DBT theval = { .data = val, .size = (uint32_t) vallen };
     FT_MSG_S cmd = { .type = FT_INSERT,

ft/fttypes.h

@@ -233,6 +233,7 @@ typedef struct cachetable *CACHETABLE;
 typedef struct cachefile *CACHEFILE;
 typedef struct ctpair *PAIR;
 typedef class checkpointer *CHECKPOINTER;
+typedef class bn_data *BN_DATA;
 
 /* tree command types */
 enum ft_msg_type {
@@ -334,7 +335,7 @@ struct ft_msg {
 // Message sent into brt to implement command (insert, delete, etc.)
 // This structure supports nested transactions, and obsoletes ft_msg.
 typedef struct ft_msg FT_MSG_S;
-typedef const struct ft_msg *FT_MSG;
+typedef struct ft_msg *FT_MSG;
 
 typedef int (*ft_compare_func)(DB *, const DBT *, const DBT *);
 typedef void (*setval_func)(const DBT *, void *);

ft/leafentry.cc

@@ -91,10 +91,6 @@ PATENT RIGHTS GRANT:
 #include "wbuf.h"
 #include "leafentry.h"
 
-void wbuf_LEAFENTRY(struct wbuf *w, LEAFENTRY le) {
-    wbuf_literal_bytes(w, le, leafentry_disksize(le));
-}
-
 void wbuf_nocrc_LEAFENTRY(struct wbuf *w, LEAFENTRY le) {
     wbuf_nocrc_literal_bytes(w, le, leafentry_disksize(le));
 }

ft/leafentry.h

@@ -132,18 +132,17 @@ PATENT RIGHTS GRANT:
 enum { LE_CLEAN = 0, LE_MVCC = 1 };
 
 // This is an on-disk format.  static_asserts verify everything is packed and aligned correctly.
-struct __attribute__ ((__packed__)) leafentry {
+struct leafentry {
     struct leafentry_clean {
         uint32_t vallen;
-        uint8_t  key_val[0];     //Actual key, then actual val
+        uint8_t  val[0];     //actual val
     }; // For the case where LEAFENTRY->type is LE_CLEAN
     static_assert(4 == sizeof(leafentry::leafentry_clean), "leafentry_clean size is wrong");
-    static_assert(4 == __builtin_offsetof(leafentry::leafentry_clean, key_val), "key_val is in the wrong place");
+    static_assert(4 == __builtin_offsetof(leafentry::leafentry_clean, val), "val is in the wrong place");
     struct __attribute__ ((__packed__)) leafentry_mvcc {
         uint32_t num_cxrs; // number of committed transaction records
         uint8_t  num_pxrs; // number of provisional transaction records
-        uint8_t key_xrs[0]; //Actual key,
-                             //then TXNIDs of XRs relevant for reads:
+        uint8_t xrs[0];      //then TXNIDs of XRs relevant for reads:
                              //  if provisional XRs exist, store OUTERMOST TXNID
                              //  store committed TXNIDs, from most recently committed to least recently committed (newest first)
                              //then lengths of XRs relevant for reads (length is at most 1<<31, MSB is 1 for insert, 0 for delete):
@@ -167,37 +166,33 @@ struct __attribute__ ((__packed__)) leafentry {
                              //   (innermost data and length with insert/delete flag are stored above, cannot be a placeholder)
     }; // For the case where LEAFENTRY->type is LE_MVCC
     static_assert(5 == sizeof(leafentry::leafentry_mvcc), "leafentry_mvcc size is wrong");
-    static_assert(5 == __builtin_offsetof(leafentry::leafentry_mvcc, key_xrs), "key_xrs is in the wrong place");
+    static_assert(5 == __builtin_offsetof(leafentry::leafentry_mvcc, xrs), "xrs is in the wrong place");
 
     uint8_t  type;    // type is LE_CLEAN or LE_MVCC
-    uint32_t keylen;
+    //uint32_t keylen;
     union __attribute__ ((__packed__)) {
         struct leafentry_clean clean;
         struct leafentry_mvcc mvcc;
     } u;
 };
-static_assert(10 == sizeof(leafentry), "leafentry size is wrong");
-static_assert(5 == __builtin_offsetof(leafentry, u), "union is in the wrong place");
+static_assert(6 == sizeof(leafentry), "leafentry size is wrong");
+static_assert(1 == __builtin_offsetof(leafentry, u), "union is in the wrong place");
 
-#define LE_CLEAN_MEMSIZE(_keylen, _vallen)                       \
+#define LE_CLEAN_MEMSIZE(_vallen)                       \
     (sizeof(((LEAFENTRY)NULL)->type)            /* type */       \
-    +sizeof(((LEAFENTRY)NULL)->keylen)          /* keylen */     \
     +sizeof(((LEAFENTRY)NULL)->u.clean.vallen)  /* vallen */     \
-    +(_keylen)                                    /* actual key */     \
     +(_vallen))                                   /* actual val */
 
 #define LE_MVCC_COMMITTED_HEADER_MEMSIZE                          \
     (sizeof(((LEAFENTRY)NULL)->type)            /* type */        \
-    +sizeof(((LEAFENTRY)NULL)->keylen)          /* keylen */      \
     +sizeof(((LEAFENTRY)NULL)->u.mvcc.num_cxrs) /* committed */   \
     +sizeof(((LEAFENTRY)NULL)->u.mvcc.num_pxrs) /* provisional */ \
     +sizeof(TXNID)                              /* transaction */ \
     +sizeof(uint32_t)                           /* length+bit */  \
     +sizeof(uint32_t))                          /* length+bit */ 
 
-#define LE_MVCC_COMMITTED_MEMSIZE(_keylen, _vallen)    \
+#define LE_MVCC_COMMITTED_MEMSIZE(_vallen)    \
     (LE_MVCC_COMMITTED_HEADER_MEMSIZE                \
-    +(_keylen)                        /* actual key */ \
     +(_vallen))                       /* actual val */
 
 
@@ -208,25 +203,20 @@ typedef struct leafentry_13 *LEAFENTRY_13;
 // TODO: consistency among names is very poor.
 //
 
+// TODO: rename this helper function for deserialization
+size_t leafentry_rest_memsize(uint32_t num_puxrs, uint32_t num_cuxrs, uint8_t* start);
 size_t leafentry_memsize (LEAFENTRY le); // the size of a leafentry in memory.
 size_t leafentry_disksize (LEAFENTRY le); // this is the same as logsizeof_LEAFENTRY.  The size of a leafentry on disk.
-void wbuf_LEAFENTRY(struct wbuf *w, LEAFENTRY le);
 void wbuf_nocrc_LEAFENTRY(struct wbuf *w, LEAFENTRY le);
-int print_leafentry (FILE *outf, LEAFENTRY v); // Print a leafentry out in human-readable form.
+int print_klpair (FILE *outf, const void* key, uint32_t keylen, LEAFENTRY v); // Print a leafentry out in human-readable form.
 
 int le_latest_is_del(LEAFENTRY le); // Return true if it is a provisional delete.
 bool le_is_clean(LEAFENTRY le); //Return how many xids exist (0 does not count)
 bool le_has_xids(LEAFENTRY le, XIDS xids); // Return true transaction represented by xids is still provisional in this leafentry (le's xid stack is a superset or equal to xids)
-uint32_t le_latest_keylen (LEAFENTRY le); // Return the latest keylen.
 void*     le_latest_val (LEAFENTRY le); // Return the latest val (return NULL for provisional deletes)
 uint32_t le_latest_vallen (LEAFENTRY le); // Return the latest vallen.  Returns 0 for provisional deletes.
 void* le_latest_val_and_len (LEAFENTRY le, uint32_t *len);
 
- // Return any key or value (even if it's only provisional).
-void* le_key (LEAFENTRY le);
-uint32_t le_keylen (LEAFENTRY le);
-void* le_key_and_len (LEAFENTRY le, uint32_t *len);
-
 uint64_t le_outermost_uncommitted_xid (LEAFENTRY le);
 
 //Callback contract:
@@ -246,35 +236,35 @@ leafentry_disksize_13(LEAFENTRY_13 le);
 
 int
 toku_le_upgrade_13_14(LEAFENTRY_13 old_leafentry, // NULL if there was no stored data.
+                      void** keyp,
+                      uint32_t* keylen,
                       size_t *new_leafentry_memorysize,
-                      LEAFENTRY *new_leafentry_p,
-                      OMT *omtp,
-                      struct mempool *mp);
-
-void toku_le_apply_msg(FT_MSG   msg,
-                       LEAFENTRY old_leafentry, // NULL if there was no stored data.
-                       TXNID oldest_referenced_xid,
-                       GC_INFO gc_info,
-                       size_t *new_leafentry_memorysize,
-                       LEAFENTRY *new_leafentry_p,
-                       OMT *omtp,
-                       struct mempool *mp,
-                       void **maybe_free,
-                       int64_t * numbytes_delta_p);
+                      LEAFENTRY *new_leafentry_p);
+
+void
+toku_le_apply_msg(FT_MSG   msg,
+                  LEAFENTRY old_leafentry, // NULL if there was no stored data.
+                  bn_data* data_buffer, // bn_data storing leafentry, if NULL, means there is no bn_data
+                  uint32_t idx, // index in data_buffer where leafentry is stored (and should be replaced
+                  TXNID oldest_referenced_xid,
+                  GC_INFO gc_info,
+                  LEAFENTRY *new_leafentry_p,
+                  int64_t * numbytes_delta_p);
 
 bool toku_le_worth_running_garbage_collection(LEAFENTRY le, TXNID oldest_referenced_xid_known);
 
-void toku_le_garbage_collect(LEAFENTRY old_leaf_entry,
-                             LEAFENTRY *new_leaf_entry,
-                             size_t *new_leaf_entry_memory_size,
-                             OMT *omtp,
-                             struct mempool *mp,
-                             void **maybe_free,
-                             const xid_omt_t &snapshot_xids,
-                             const rx_omt_t &referenced_xids,
-                             const xid_omt_t &live_root_txns,
-                             TXNID oldest_referenced_xid_known,
-                             int64_t * numbytes_delta_p);
+void
+toku_le_garbage_collect(LEAFENTRY old_leaf_entry,
+                        bn_data* data_buffer,
+                        uint32_t idx,
+                        void* keyp,
+                        uint32_t keylen,
+                        LEAFENTRY *new_leaf_entry,
+                        const xid_omt_t &snapshot_xids,
+                        const rx_omt_t &referenced_xids,
+                        const xid_omt_t &live_root_txns,
+                        TXNID oldest_referenced_xid_known,
+                        int64_t * numbytes_delta_p);
 
 #endif /* TOKU_LEAFENTRY_H */
 

ft/tests/ft-clock-test.cc

@@ -109,44 +109,31 @@ string_key_cmp(DB *UU(e), const DBT *a, const DBT *b)
     return strcmp(s, t);
 }
 
-static int omt_cmp(OMTVALUE p, void *q)
-{
-    LEAFENTRY CAST_FROM_VOIDP(a, p);
-    LEAFENTRY CAST_FROM_VOIDP(b, q);
-    void *ak, *bk;
-    uint32_t al, bl;
-    ak = le_key_and_len(a, &al);
-    bk = le_key_and_len(b, &bl);
-    int l = MIN(al, bl);
-    int c = memcmp(ak, bk, l);
-    if (c < 0) { return -1; }
-    if (c > 0) { return +1; }
-    int d = al - bl;
-    if (d < 0) { return -1; }
-    if (d > 0) { return +1; }
-    else       { return  0; }
-}
-
-static LEAFENTRY
-le_fastmalloc(const char *key, int keylen, const char *val, int vallen)
+static void
+le_add_to_bn(bn_data* bn, uint32_t idx, const  char *key, int keylen, const char *val, int vallen)
 {
-    LEAFENTRY CAST_FROM_VOIDP(r, toku_malloc(sizeof(r->type) + sizeof(r->keylen) + sizeof(r->u.clean.vallen) +
-                                             keylen + vallen));
+    LEAFENTRY r = NULL;
+    uint32_t size_needed = LE_CLEAN_MEMSIZE(vallen);
+    bn->get_space_for_insert(
+        idx, 
+        key,
+        keylen,
+        size_needed,
+        &r
+        );
     resource_assert(r);
     r->type = LE_CLEAN;
-    r->keylen = keylen;
     r->u.clean.vallen = vallen;
-    memcpy(&r->u.clean.key_val[0], key, keylen);
-    memcpy(&r->u.clean.key_val[keylen], val, vallen);
-    return r;
+    memcpy(r->u.clean.val, val, vallen);
 }
 
-static LEAFENTRY
-le_malloc(const char *key, const char *val)
+
+static void
+le_malloc(bn_data* bn, uint32_t idx, const char *key, const char *val)
 {
     int keylen = strlen(key) + 1;
     int vallen = strlen(val) + 1;
-    return le_fastmalloc(key, keylen, val, vallen);
+    le_add_to_bn(bn, idx, key, keylen, val, vallen);
 }
 
 
@@ -219,7 +206,7 @@ test1(int fd, FT brt_h, FTNODE *dn) {
 }
 
 
-static int search_cmp(struct ft_search* UU(so), DBT* UU(key)) {
+static int search_cmp(const struct ft_search& UU(so), const DBT* UU(key)) {
     return 0;
 }
 
@@ -430,10 +417,6 @@ test_serialize_leaf(void) {
     sn.n_children = 2;
     sn.dirty = 1;
     sn.oldest_referenced_xid_known = TXNID_NONE;
-    LEAFENTRY elts[3];
-    elts[0] = le_malloc("a", "aval");
-    elts[1] = le_malloc("b", "bval");
-    elts[2] = le_malloc("x", "xval");
     MALLOC_N(sn.n_children, sn.bp);
     MALLOC_N(1, sn.childkeys);
     toku_memdup_dbt(&sn.childkeys[0], "b", 2);
@@ -442,11 +425,9 @@ test_serialize_leaf(void) {
     BP_STATE(&sn,1) = PT_AVAIL;
     set_BLB(&sn, 0, toku_create_empty_bn());
     set_BLB(&sn, 1, toku_create_empty_bn());
-    r = toku_omt_insert(BLB_BUFFER(&sn, 0), elts[0], omt_cmp, elts[0], NULL); assert(r==0);
-    r = toku_omt_insert(BLB_BUFFER(&sn, 0), elts[1], omt_cmp, elts[1], NULL); assert(r==0);
-    r = toku_omt_insert(BLB_BUFFER(&sn, 1), elts[2], omt_cmp, elts[2], NULL); assert(r==0);
-    BLB_NBYTESINBUF(&sn, 0) = 2*(KEY_VALUE_OVERHEAD+2+5) + toku_omt_size(BLB_BUFFER(&sn, 0));
-    BLB_NBYTESINBUF(&sn, 1) = 1*(KEY_VALUE_OVERHEAD+2+5) + toku_omt_size(BLB_BUFFER(&sn, 1));
+    le_malloc(BLB_DATA(&sn, 0), 0, "a", "aval");
+    le_malloc(BLB_DATA(&sn, 0), 1, "b", "bval");
+    le_malloc(BLB_DATA(&sn, 1), 0, "x", "xval");
 
     FT_HANDLE XMALLOC(brt);
     FT XCALLOC(brt_h);
@@ -488,12 +469,7 @@ test_serialize_leaf(void) {
     for (int i = 0; i < sn.n_children-1; ++i) {
         toku_free(sn.childkeys[i].data);
     }
-    for (int i = 0; i < 3; ++i) {
-        toku_free(elts[i]);
-    }
     for (int i = 0; i < sn.n_children; i++) {
-        struct mempool * mp = &(BLB_BUFFER_MEMPOOL(&sn, i));
-	toku_mempool_destroy(mp);
         destroy_basement_node(BLB(&sn, i));
     }
     toku_free(sn.bp);

ft/tests/ft-serialize-benchmark.cc

@@ -99,36 +99,22 @@ PATENT RIGHTS GRANT:
 #endif
 const double USECS_PER_SEC = 1000000.0;
 
-static int omt_cmp(OMTVALUE p, void *q)
-{
-    LEAFENTRY CAST_FROM_VOIDP(a, p);
-    LEAFENTRY CAST_FROM_VOIDP(b, q);
-    void *ak, *bk;
-    uint32_t al, bl;
-    ak = le_key_and_len(a, &al);
-    bk = le_key_and_len(b, &bl);
-    int l = MIN(al, bl);
-    int c = memcmp(ak, bk, l);
-    if (c < 0) { return -1; }
-    if (c > 0) { return +1; }
-    int d = al - bl;
-    if (d < 0) { return -1; }
-    if (d > 0) { return +1; }
-    else       { return  0; }
-}
-
-static LEAFENTRY
-le_fastmalloc(char *key, int keylen, char *val, int vallen)
+static void
+le_add_to_bn(bn_data* bn, uint32_t idx, char *key, int keylen, char *val, int vallen)
 {
-    LEAFENTRY CAST_FROM_VOIDP(r, toku_malloc(sizeof(r->type) + sizeof(r->keylen) + sizeof(r->u.clean.vallen) +
-                                             keylen + vallen));
+    LEAFENTRY r = NULL;
+    uint32_t size_needed = LE_CLEAN_MEMSIZE(vallen);
+    bn->get_space_for_insert(
+        idx, 
+        key,
+        keylen,
+        size_needed,
+        &r
+        );
     resource_assert(r);
     r->type = LE_CLEAN;
-    r->keylen = keylen;
     r->u.clean.vallen = vallen;
-    memcpy(&r->u.clean.key_val[0], key, keylen);
-    memcpy(&r->u.clean.key_val[keylen], val, vallen);
-    return r;
+    memcpy(r->u.clean.val, val, vallen);
 }
 
 static int
@@ -167,8 +153,6 @@ test_serialize_leaf(int valsize, int nelts, double entropy) {
         set_BLB(sn, i, toku_create_empty_bn());
     }
     int nperbn = nelts / sn->n_children;
-    LEAFENTRY les[nelts];
-    memset(les, 0, sizeof les);
     for (int ck = 0; ck < sn->n_children; ++ck) {
         long k;
         for (long i = 0; i < nperbn; ++i) {
@@ -181,10 +165,15 @@ test_serialize_leaf(int valsize, int nelts, double entropy) {
                 c += sizeof(*p);
             }
             memset(&buf[c], 0, valsize - c);
-            les[k] = le_fastmalloc((char *)&k, sizeof k, buf, sizeof buf);
-            r = toku_omt_insert(BLB_BUFFER(sn, ck), les[k], omt_cmp, les[k], NULL); assert(r==0);
+            le_add_to_bn(
+                BLB_DATA(sn,ck),
+                i,
+                (char *)&k, 
+                sizeof k, 
+                buf, 
+                sizeof buf
+                );
         }
-        BLB_NBYTESINBUF(sn, ck) = nperbn*(KEY_VALUE_OVERHEAD+(sizeof(long)+valsize)) + toku_omt_size(BLB_BUFFER(sn, ck));
         if (ck < 7) {
             toku_memdup_dbt(&sn->childkeys[ck], &k, sizeof k);
             sn->totalchildkeylens += sizeof k;
@@ -249,12 +238,6 @@ test_serialize_leaf(int valsize, int nelts, double entropy) {
            );
 
     toku_ftnode_free(&dn);
-
-    for (int i = 0; i < nelts; ++i) {
-        if (les[i]) {
-            toku_free(les[i]);
-        }
-    }
     toku_ftnode_free(&sn);
 
     toku_block_free(brt_h->blocktable, BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);

ft/tests/le-cursor-provdel.cc

@@ -98,17 +98,17 @@ static TOKUTXN const null_txn = 0;
 static DB * const null_db = 0;
 
 static int
-get_next_callback(ITEMLEN UU(keylen), bytevec UU(key), ITEMLEN vallen, bytevec val, void *extra, bool lock_only) {
-    DBT *CAST_FROM_VOIDP(val_dbt, extra);
+get_next_callback(ITEMLEN keylen, bytevec key, ITEMLEN vallen UU(), bytevec val UU(), void *extra, bool lock_only) {
+    DBT *CAST_FROM_VOIDP(key_dbt, extra);
     if (!lock_only) {
-        toku_dbt_set(vallen, val, val_dbt, NULL);
+        toku_dbt_set(keylen, key, key_dbt, NULL);
     }
     return 0;
 }
 
 static int
-le_cursor_get_next(LE_CURSOR cursor, DBT *val) {
-    int r = toku_le_cursor_next(cursor, get_next_callback, val);
+le_cursor_get_next(LE_CURSOR cursor, DBT *key) {
+    int r = toku_le_cursor_next(cursor, get_next_callback, key);
     return r;
 }
 
@@ -243,15 +243,12 @@ test_provdel(const char *logdir, const char *fname, int n) {
 
     int i;
     for (i=0; ; i++) {
-        error = le_cursor_get_next(cursor, &val);
+        error = le_cursor_get_next(cursor, &key);
         if (error != 0) 
             break;
         
-        LEAFENTRY le = (LEAFENTRY) val.data;
-        assert(le->type == LE_MVCC);
-        assert(le->keylen == sizeof (int));
-        int ii;
-        memcpy(&ii, le->u.mvcc.key_xrs, le->keylen);
+        assert(key.size == sizeof (int));
+        int ii = *(int *)key.data;
         assert((int) toku_htonl(n-i-1) == ii);
     }
     assert(i == n);

ft/tests/le-cursor-right.cc

@@ -101,10 +101,10 @@ static TOKUTXN const null_txn = 0;
 static DB * const null_db = 0;
 
 static int
-get_next_callback(ITEMLEN UU(keylen), bytevec UU(key), ITEMLEN vallen, bytevec val, void *extra, bool lock_only) {
-    DBT *CAST_FROM_VOIDP(val_dbt, extra);
+get_next_callback(ITEMLEN keylen, bytevec key, ITEMLEN vallen UU(), bytevec val UU(), void *extra, bool lock_only) {
+    DBT *CAST_FROM_VOIDP(key_dbt, extra);
     if (!lock_only) {
-        toku_dbt_set(vallen, val, val_dbt, NULL);
+        toku_dbt_set(keylen, key, key_dbt, NULL);
     }
     return 0;
 }
@@ -247,17 +247,13 @@ test_neg_infinity(const char *fname, int n) {
 
     int i;
     for (i = n-1; ; i--) {
-        error = le_cursor_get_next(cursor, &val);
+        error = le_cursor_get_next(cursor, &key);
         if (error != 0) 
             break;
         
-        LEAFENTRY le = (LEAFENTRY) val.data;
-        assert(le->type == LE_MVCC);
-        assert(le->keylen == sizeof (int));
-        int ii;
-        memcpy(&ii, le->u.mvcc.key_xrs, le->keylen);
+        assert(key.size == sizeof (int));
+        int ii = *(int *)key.data;
         assert((int) toku_htonl(i) == ii);
-
     }
     assert(i == -1);
 
@@ -306,17 +302,12 @@ test_between(const char *fname, int n) {
     int i;
     for (i = 0; ; i++) {
         // move the LE_CURSOR forward
-        error = le_cursor_get_next(cursor, &val);
+        error = le_cursor_get_next(cursor, &key);
         if (error != 0) 
             break;
         
-        LEAFENTRY le = (LEAFENTRY) val.data;
-        assert(le->type == LE_MVCC);
-        assert(le->keylen == sizeof (int));
-        int ii;
-        memcpy(&ii, le->u.mvcc.key_xrs, le->keylen);
-        // hot indexer runs in reverse, therefore need
-        // to check n-i-1
+        assert(key.size == sizeof (int));
+        int ii = *(int *)key.data;
         assert((int) toku_htonl(n-i-1) == ii);
 
         // test 0 .. i-1 

ft/tests/le-cursor-walk.cc

@@ -99,10 +99,10 @@ static TOKUTXN const null_txn = 0;
 static DB * const null_db = 0;
 
 static int
-get_next_callback(ITEMLEN UU(keylen), bytevec UU(key), ITEMLEN vallen, bytevec val, void *extra, bool lock_only) {
-    DBT *CAST_FROM_VOIDP(val_dbt, extra);
+get_next_callback(ITEMLEN keylen, bytevec key, ITEMLEN vallen UU(), bytevec val UU(), void *extra, bool lock_only) {
+    DBT *CAST_FROM_VOIDP(key_dbt, extra);
     if (!lock_only) {
-        toku_dbt_set(vallen, val, val_dbt, NULL);
+        toku_dbt_set(keylen, key, key_dbt, NULL);
     }
     return 0;
 }
@@ -207,19 +207,13 @@ walk_tree(const char *fname, int n) {
     toku_init_dbt(&val); val.flags = DB_DBT_REALLOC;
 
     int i;
-    for (i = 0; ; i++) {
-        error = TOKUDB_TRY_AGAIN;
-        while (error == TOKUDB_TRY_AGAIN) {
-            error = le_cursor_get_next(cursor, &val);
-        }
+    for (i=0; ; i++) {
+        error = le_cursor_get_next(cursor, &key);
         if (error != 0) 
             break;
-
-        LEAFENTRY le = (LEAFENTRY) val.data;
-        assert(le->type == LE_MVCC);
-        assert(le->keylen == sizeof (int));
-        int ii;
-        memcpy(&ii, le->u.mvcc.key_xrs, le->keylen);
+        
+        assert(key.size == sizeof (int));
+        int ii = *(int *)key.data;
         assert((int) toku_htonl(n-i-1) == ii);
     }
     assert(i == n);

ft/tests/make-tree.cc

@@ -118,7 +118,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     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(BLB_BUFFER(leafnode, 0));
+    uint32_t idx = BLB_DATA(leafnode, 0)->omt_size();
 
     MSN msn = next_dummymsn();
 

ft/tests/test-checkpoint-during-flush.cc

@@ -320,10 +320,10 @@ doit (bool after_child_pin) {
     assert(!node->dirty);
     assert(node->n_children == 1);
     if (after_child_pin) {
-        assert(BLB_NBYTESINBUF(node,0) > 0);
+        assert(BLB_NBYTESINDATA(node,0) > 0);
     }
     else {
-        assert(BLB_NBYTESINBUF(node,0) == 0);
+        assert(BLB_NBYTESINDATA(node,0) == 0);
     }
     toku_unpin_ftnode_off_client_thread(c_ft->ft, node);
 

ft/tests/test-checkpoint-during-merge.cc

@@ -347,7 +347,7 @@ doit (int state) {
         assert(node->height == 0);
         assert(!node->dirty);
         assert(node->n_children == 1);
-        assert(toku_omt_size(BLB_BUFFER(node,0)) == 1);
+        assert(BLB_DATA(node, 0)->omt_size() == 1);
         toku_unpin_ftnode_off_client_thread(c_ft->ft, node);
 
         toku_pin_ftnode_off_client_thread(
@@ -363,7 +363,7 @@ doit (int state) {
         assert(node->height == 0);
         assert(!node->dirty);
         assert(node->n_children == 1);
-        assert(toku_omt_size(BLB_BUFFER(node,0)) == 1);
+        assert(BLB_DATA(node, 0)->omt_size() == 1);
         toku_unpin_ftnode_off_client_thread(c_ft->ft, node);
     }
     else if (state == ft_flush_aflter_merge || state == flt_flush_before_unpin_remove) {
@@ -380,7 +380,7 @@ doit (int state) {
         assert(node->height == 0);
         assert(!node->dirty);
         assert(node->n_children == 1);
-        assert(toku_omt_size(BLB_BUFFER(node,0)) == 2);
+        assert(BLB_DATA(node, 0)->omt_size() == 2);
         toku_unpin_ftnode_off_client_thread(c_ft->ft, node);
     }
     else {

ft/tests/test-checkpoint-during-rebalance.cc

@@ -358,7 +358,7 @@ doit (int state) {
     assert(node->height == 0);
     assert(!node->dirty);
     assert(node->n_children == 1);
-    assert(toku_omt_size(BLB_BUFFER(node,0)) == 2);
+    assert(BLB_DATA(node, 0)->omt_size() == 2);
     toku_unpin_ftnode_off_client_thread(c_ft->ft, node);
     
     toku_pin_ftnode_off_client_thread(
@@ -374,7 +374,7 @@ doit (int state) {
     assert(node->height == 0);
     assert(!node->dirty);
     assert(node->n_children == 1);
-    assert(toku_omt_size(BLB_BUFFER(node,0)) == 2);
+    assert(BLB_DATA(node, 0)->omt_size() == 2);
     toku_unpin_ftnode_off_client_thread(c_ft->ft, node);
 
 

ft/tests/test-checkpoint-during-split.cc

@@ -341,7 +341,7 @@ doit (bool after_split) {
         assert(node->height == 0);
         assert(!node->dirty);
         assert(node->n_children == 1);
-        assert(toku_omt_size(BLB_BUFFER(node,0)) == 1);
+        assert(BLB_DATA(node, 0)->omt_size() == 1);
         toku_unpin_ftnode_off_client_thread(c_ft->ft, node);
 
         toku_pin_ftnode_off_client_thread(
@@ -357,7 +357,7 @@ doit (bool after_split) {
         assert(node->height == 0);
         assert(!node->dirty);
         assert(node->n_children == 1);
-        assert(toku_omt_size(BLB_BUFFER(node,0)) == 1);
+        assert(BLB_DATA(node, 0)->omt_size() == 1);
         toku_unpin_ftnode_off_client_thread(c_ft->ft, node);
     }
     else {
@@ -374,7 +374,7 @@ doit (bool after_split) {
         assert(node->height == 0);
         assert(!node->dirty);
         assert(node->n_children == 1);
-        assert(toku_omt_size(BLB_BUFFER(node,0)) == 2);
+        assert(BLB_DATA(node, 0)->omt_size() == 2);
         toku_unpin_ftnode_off_client_thread(c_ft->ft, node);
     }
 

ft/tests/test-leafentry-child-txn.cc

@@ -95,12 +95,10 @@ PATENT RIGHTS GRANT:
 #include "ule.h"
 #include "ule-internal.h"
 
-static void init_empty_ule(ULE ule, DBT *key) {
+static void init_empty_ule(ULE ule) {
     ule->num_cuxrs = 0;
     ule->num_puxrs = 0;
     ule->uxrs = ule->uxrs_static;
-    ule->keylen   = key->size;
-    ule->keyp     = key->data;
 }
 
 static void add_committed_entry(ULE ule, DBT *val, TXNID xid) {
@@ -155,7 +153,7 @@ run_test(void) {
     r = xids_create_child(msg_xids_1, &msg_xids_2, child_id);
     assert(r==0);
 
-    init_empty_ule(&ule_initial, &key);
+    init_empty_ule(&ule_initial);
     add_committed_entry(&ule_initial, &val, 0);
     val.data = &val_data_two;
     // make the TXNID match the child id of xids

ft/tests/verify-bad-msn.cc

@@ -121,7 +121,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     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(BLB_BUFFER(leafnode, 0));
+    uint32_t idx = BLB_DATA(leafnode, 0)->omt_size();
 
     MSN msn = next_dummymsn();
 

ft/tests/verify-bad-pivots.cc

@@ -110,7 +110,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     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(BLB_BUFFER(leafnode, 0));
+    uint32_t idx = BLB_DATA(leafnode, 0)->omt_size();
 
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();

ft/tests/verify-dup-in-leaf.cc

@@ -111,7 +111,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     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(BLB_BUFFER(leafnode, 0));
+    uint32_t idx = BLB_DATA(leafnode, 0)->omt_size();
 
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();

ft/tests/verify-dup-pivots.cc

@@ -110,7 +110,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     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(BLB_BUFFER(leafnode, 0));
+    uint32_t idx = BLB_DATA(leafnode, 0)->omt_size();
 
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();

ft/tests/verify-misrouted-msgs.cc

@@ -111,7 +111,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     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(BLB_BUFFER(leafnode, 0));
+    uint32_t idx = BLB_DATA(leafnode, 0)->omt_size();
 
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();

ft/tests/verify-unsorted-leaf.cc

@@ -113,7 +113,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     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(BLB_BUFFER(leafnode, 0));
+    uint32_t idx = BLB_DATA(leafnode, 0)->omt_size();
 
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();

ft/tests/verify-unsorted-pivots.cc

@@ -110,7 +110,7 @@ append_leaf(FTNODE leafnode, void *key, size_t keylen, void *val, size_t vallen)
     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(BLB_BUFFER(leafnode, 0));
+    uint32_t idx = BLB_DATA(leafnode, 0)->omt_size();
 
     // apply an insert to the leaf node
     MSN msn = next_dummymsn();

ft/ule-internal.h

@@ -128,8 +128,6 @@ typedef struct uxr {     // unpacked transaction record
 typedef struct ule {     // unpacked leaf entry
     uint32_t  num_puxrs;   // how many of uxrs[] are provisional
     uint32_t  num_cuxrs;   // how many of uxrs[] are committed
-    uint32_t  keylen;
-    void *    keyp;
     UXR_S     uxrs_static[MAX_TRANSACTION_RECORDS*2];    // uxrs[0] is oldest committed (txn commit time, not txn start time), uxrs[num_cuxrs] is outermost provisional value (if any exist/num_puxrs > 0)
     UXR       uxrs;                                      //If num_cuxrs < MAX_TRANSACTION_RECORDS then &uxrs_static[0].
                                                          //Otherwise we use a dynamically allocated array of size num_cuxrs + 1 + MAX_TRANSATION_RECORD.
@@ -143,12 +141,15 @@ void test_msg_modify_ule(ULE ule, FT_MSG msg);
 //////////////////////////////////////////////////////////////////////////////////////
 //Functions exported for test purposes only (used internally for non-test purposes).
 void le_unpack(ULE ule,  LEAFENTRY le);
-int le_pack(ULE ule,                            // data to be packed into new leafentry
-	    size_t *new_leafentry_memorysize,
-	    LEAFENTRY * const new_leafentry_p,  // this is what this function creates
-	    OMT *omtp,
-	    struct mempool *mp,
-	    void **maybe_free);
+int
+le_pack(ULE ule, // data to be packed into new leafentry
+        bn_data* data_buffer,
+        uint32_t idx,
+        void* keyp,
+        uint32_t keylen,
+        uint32_t old_le_size,
+        LEAFENTRY * const new_leafentry_p // this is what this function creates
+        );
 
 
 size_t le_memsize_from_ule (ULE ule);

ft/ule.h

@@ -116,8 +116,6 @@ uint32_t ule_get_num_provisional(ULEHANDLE ule);
 UXRHANDLE ule_get_uxr(ULEHANDLE ule, uint64_t ith);
 int ule_is_committed(ULEHANDLE ule, uint64_t ith);
 int ule_is_provisional(ULEHANDLE ule, uint64_t ith);
-void *ule_get_key(ULEHANDLE ule);
-uint32_t ule_get_keylen(ULEHANDLE ule);
 
 bool uxr_is_insert(UXRHANDLE uxr);
 bool uxr_is_delete(UXRHANDLE uxr);

src/indexer-internal.h

@@ -113,6 +113,8 @@ struct ule_prov_info {
     // is responsible for cleaning up the leafentry and ule when done.
     LEAFENTRY le;
     ULEHANDLE ule;
+    void* key;
+    uint32_t keylen;
     // provisional txn info for the ule
     uint32_t num_provisional;
     uint32_t num_committed;
@@ -147,7 +149,7 @@ struct __toku_indexer_internal {
     DBT_ARRAY *hot_vals;
 
     // test functions
-    int (*undo_do)(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule);
+    int (*undo_do)(DB_INDEXER *indexer, DB *hotdb, DBT* key, ULEHANDLE ule);
     TOKUTXN_STATE (*test_xid_state)(DB_INDEXER *indexer, TXNID xid);
     void (*test_lock_key)(DB_INDEXER *indexer, TXNID xid, DB *hotdb, DBT *key);
     int (*test_delete_provisional)(DB_INDEXER *indexer, DB *hotdb, DBT *hotkey, XIDS xids);
@@ -164,6 +166,6 @@ void indexer_undo_do_init(DB_INDEXER *indexer);
 
 void indexer_undo_do_destroy(DB_INDEXER *indexer);
 
-int indexer_undo_do(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule, struct ule_prov_info *prov_info, DBT_ARRAY *hot_keys, DBT_ARRAY *hot_vals);
+int indexer_undo_do(DB_INDEXER *indexer, DB *hotdb, struct ule_prov_info *prov_info, DBT_ARRAY *hot_keys, DBT_ARRAY *hot_vals);
 
 #endif

src/indexer-undo-do.cc

@@ -156,7 +156,7 @@ static int indexer_append_xid(DB_INDEXER *indexer, TXNID xid, XIDS *xids_result)
 
 static bool indexer_find_prev_xr(DB_INDEXER *indexer, ULEHANDLE ule, uint64_t xrindex, uint64_t *prev_xrindex);
 
-static int indexer_generate_hot_keys_vals(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule, UXRHANDLE uxr, DBT_ARRAY *hotkeys, DBT_ARRAY *hotvals);
+static int indexer_generate_hot_keys_vals(DB_INDEXER *indexer, DB *hotdb, struct ule_prov_info* prov_info, UXRHANDLE uxr, DBT_ARRAY *hotkeys, DBT_ARRAY *hotvals);
 static int indexer_ft_delete_provisional(DB_INDEXER *indexer, DB *hotdb, DBT *hotkey, XIDS xids, TOKUTXN txn);
 static int indexer_ft_delete_committed(DB_INDEXER *indexer, DB *hotdb, DBT *hotkey, XIDS xids);
 static int indexer_ft_insert_provisional(DB_INDEXER *indexer, DB *hotdb, DBT *hotkey, DBT *hotval, XIDS xids, TOKUTXN txn);
@@ -193,8 +193,9 @@ indexer_undo_do_destroy(DB_INDEXER *indexer) {
 }
 
 static int
-indexer_undo_do_committed(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule, DBT_ARRAY *hot_keys, DBT_ARRAY *hot_vals) {
+indexer_undo_do_committed(DB_INDEXER *indexer, DB *hotdb, struct ule_prov_info *prov_info, DBT_ARRAY *hot_keys, DBT_ARRAY *hot_vals) {
     int result = 0;
+    ULEHANDLE ule = prov_info->ule;
 
     // init the xids to the root xid
     XIDS xids = xids_get_root_xids();
@@ -225,7 +226,7 @@ indexer_undo_do_committed(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule, DBT_ARR
                 ; // do nothing
             } else if (uxr_is_insert(prevuxr)) {
                 // generate the hot delete key
-                result = indexer_generate_hot_keys_vals(indexer, hotdb, ule, prevuxr, hot_keys, NULL);
+                result = indexer_generate_hot_keys_vals(indexer, hotdb, prov_info, prevuxr, hot_keys, NULL);
                 if (result == 0) {
                     paranoid_invariant(hot_keys->size <= hot_keys->capacity);
                     for (uint32_t i = 0; i < hot_keys->size; i++) {
@@ -251,7 +252,7 @@ indexer_undo_do_committed(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule, DBT_ARR
             ; // do nothing
         } else if (uxr_is_insert(uxr)) {
             // generate the hot insert key and val
-            result = indexer_generate_hot_keys_vals(indexer, hotdb, ule, uxr, hot_keys, hot_vals);
+            result = indexer_generate_hot_keys_vals(indexer, hotdb, prov_info, uxr, hot_keys, hot_vals);
             if (result == 0) {
                 paranoid_invariant(hot_keys->size == hot_vals->size);
                 paranoid_invariant(hot_keys->size <= hot_keys->capacity);
@@ -304,9 +305,10 @@ exit:
 }
 
 static int
-indexer_undo_do_provisional(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule, struct ule_prov_info *prov_info, DBT_ARRAY *hot_keys, DBT_ARRAY *hot_vals) {
+indexer_undo_do_provisional(DB_INDEXER *indexer, DB *hotdb, struct ule_prov_info *prov_info, DBT_ARRAY *hot_keys, DBT_ARRAY *hot_vals) {
     int result = 0;
     indexer_commit_keys_set_empty(&indexer->i->commit_keys);
+    ULEHANDLE ule = prov_info->ule;
 
     // init the xids to the root xid
     XIDS xids = xids_get_root_xids();
@@ -380,7 +382,7 @@ indexer_undo_do_provisional(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule, struc
                 ; // do nothing
             } else if (uxr_is_insert(prevuxr)) {
                 // generate the hot delete key
-                result = indexer_generate_hot_keys_vals(indexer, hotdb, ule, prevuxr, hot_keys, NULL);
+                result = indexer_generate_hot_keys_vals(indexer, hotdb, prov_info, prevuxr, hot_keys, NULL);
                 if (result == 0) {
                     paranoid_invariant(hot_keys->size <= hot_keys->capacity);
                     for (uint32_t i = 0; i < hot_keys->size; i++) {
@@ -419,7 +421,7 @@ indexer_undo_do_provisional(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule, struc
             ; // do nothing
         } else if (uxr_is_insert(uxr)) {
             // generate the hot insert key and val
-            result = indexer_generate_hot_keys_vals(indexer, hotdb, ule, uxr, hot_keys, hot_vals);
+            result = indexer_generate_hot_keys_vals(indexer, hotdb, prov_info, uxr, hot_keys, hot_vals);
             if (result == 0) {
                 paranoid_invariant(hot_keys->size == hot_vals->size);
                 paranoid_invariant(hot_keys->size <= hot_keys->capacity);
@@ -474,10 +476,10 @@ exit:
 }
 
 int
-indexer_undo_do(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule, struct ule_prov_info *prov_info, DBT_ARRAY *hot_keys, DBT_ARRAY *hot_vals) {
-    int result = indexer_undo_do_committed(indexer, hotdb, ule, hot_keys, hot_vals);
+indexer_undo_do(DB_INDEXER *indexer, DB *hotdb, struct ule_prov_info *prov_info, DBT_ARRAY *hot_keys, DBT_ARRAY *hot_vals) {
+    int result = indexer_undo_do_committed(indexer, hotdb, prov_info, hot_keys, hot_vals);
     if (result == 0) {
-        result = indexer_undo_do_provisional(indexer, hotdb, ule, prov_info, hot_keys, hot_vals);
+        result = indexer_undo_do_provisional(indexer, hotdb, prov_info, hot_keys, hot_vals);
     }
     if (indexer->i->test_only_flags == INDEXER_TEST_ONLY_ERROR_CALLBACK)  {
         result = EINVAL;
@@ -523,12 +525,12 @@ indexer_append_xid(DB_INDEXER *UU(indexer), TXNID xid, XIDS *xids_result) {
 }
 
 static int
-indexer_generate_hot_keys_vals(DB_INDEXER *indexer, DB *hotdb, ULEHANDLE ule, UXRHANDLE uxr, DBT_ARRAY *hotkeys, DBT_ARRAY *hotvals) {
+indexer_generate_hot_keys_vals(DB_INDEXER *indexer, DB *hotdb, struct ule_prov_info *prov_info, UXRHANDLE uxr, DBT_ARRAY *hotkeys, DBT_ARRAY *hotvals) {
     int result = 0;
 
     // setup the source key
     DBT srckey;
-    toku_fill_dbt(&srckey, ule_get_key(ule), ule_get_keylen(ule));
+    toku_fill_dbt(&srckey, prov_info->key, prov_info->keylen);
 
     // setup the source val
     DBT srcval;

util/mempool.h

@@ -123,7 +123,7 @@ void toku_mempool_copy_construct(struct mempool *mp, const void * const data_sou
 
 /* initialize the memory pool with the base address and size of a
    contiguous chunk of memory */
-void toku_mempool_init(struct mempool *mp, void *base, size_t size);
+void toku_mempool_init(struct mempool *mp, void *base, size_t free_offset, size_t size);
 
 /* allocate memory and construct mempool
  */