FT-304 Move the blocktable into its own class, fix tests.

ft/ft-cachetable-wrappers.cc

@@ -105,10 +105,10 @@ ftnode_get_key_and_fullhash(
     void* extra)
 {
     FT ft = (FT) extra;
-    BLOCKNUM name;
-    toku_allocate_blocknum(ft->blocktable, &name, ft);
-    *cachekey = name;
-    *fullhash = toku_cachetable_hash(ft->cf, name);
+    BLOCKNUM blocknum;
+    ft->blocktable.allocate_blocknum(&blocknum, ft);
+    *cachekey = blocknum;
+    *fullhash = toku_cachetable_hash(ft->cf, blocknum);
 }
 
 void
@@ -116,7 +116,7 @@ cachetable_put_empty_node_with_dep_nodes(
     FT ft,
     uint32_t num_dependent_nodes,
     FTNODE* dependent_nodes,
-    BLOCKNUM* name, //output
+    BLOCKNUM* blocknum, //output
     uint32_t* fullhash, //output
     FTNODE* result)
 {
@@ -138,7 +138,7 @@ cachetable_put_empty_node_with_dep_nodes(
         num_dependent_nodes,
         dependent_pairs,
         dependent_dirty_bits,
-        name,
+        blocknum,
         fullhash,
         toku_ftnode_save_ct_pair);
     *result = new_node;
@@ -154,13 +154,13 @@ create_new_ftnode_with_dep_nodes(
     FTNODE* dependent_nodes)
 {
     uint32_t fullhash = 0;
-    BLOCKNUM name;
+    BLOCKNUM blocknum;
 
     cachetable_put_empty_node_with_dep_nodes(
         ft,
         num_dependent_nodes,
         dependent_nodes,
-        &name,
+        &blocknum,
         &fullhash,
         result);
 
@@ -171,7 +171,7 @@ create_new_ftnode_with_dep_nodes(
 
     toku_initialize_empty_ftnode(
         *result,
-        name,
+        blocknum,
         height,
         n_children,
         ft->h->layout_version,

ft/ft-flusher.cc

@@ -1338,7 +1338,7 @@ maybe_merge_pinned_nodes(
 
 static void merge_remove_key_callback(BLOCKNUM *bp, bool for_checkpoint, void *extra) {
     FT ft = (FT) extra;
-    toku_free_blocknum(ft->blocktable, bp, ft, for_checkpoint);
+    ft->blocktable.free_blocknum(bp, ft, for_checkpoint);
 }
 
 //
@@ -1517,7 +1517,7 @@ void toku_ft_flush_some_child(FT ft, FTNODE parent, struct flusher_advice *fa)
 
     // get the child into memory
     BLOCKNUM targetchild = BP_BLOCKNUM(parent, childnum);
-    toku_verify_blocknum_allocated(ft->blocktable, targetchild);
+    ft->blocktable.verify_blocknum_allocated(targetchild);
     uint32_t childfullhash = compute_child_fullhash(ft->cf, parent, childnum);
     FTNODE child;
     struct ftnode_fetch_extra bfe;

ft/ft-internal.h

@@ -101,6 +101,7 @@ PATENT RIGHTS GRANT:
 #include "ft/ft.h"
 #include "ft/ft-ops.h"
 #include "ft/node.h"
+#include "ft/serialize/block_table.h"
 #include "ft/txn/rollback.h"
 
 // Symbol TOKUDB_REVISION is not defined by fractal-tree makefiles, so
@@ -111,7 +112,6 @@ PATENT RIGHTS GRANT:
 #error
 #endif
 
-struct block_table;
 struct ft_search;
 
 enum { FT_DEFAULT_FANOUT = 16 };
@@ -225,7 +225,7 @@ struct ft {
     // These are not read-only:
 
     // protected by blocktable lock
-    struct block_table *blocktable;
+    struct block_table blocktable;
 
     // protected by atomic builtins
     STAT64INFO_S in_memory_stats;

ft/ft-ops.cc

@@ -1927,7 +1927,7 @@ static void push_something_in_subtree(
 
         {
             const BLOCKNUM child_blocknum = BP_BLOCKNUM(subtree_root, childnum);
-            toku_verify_blocknum_allocated(ft->blocktable, child_blocknum);
+            ft->blocktable.verify_blocknum_allocated(child_blocknum);
             const uint32_t child_fullhash = toku_cachetable_hash(ft->cf, child_blocknum);
 
             FTNODE child;
@@ -3088,10 +3088,11 @@ ft_handle_open(FT_HANDLE ft_h, const char *fname_in_env, int is_create, int only
         toku_txn_maybe_note_ft(txn, ft);
     }
 
-    //Opening an ft may restore to previous checkpoint.         Truncate if necessary.
+    // Opening an ft may restore to previous checkpoint.
+    // Truncate if necessary.
     {
         int fd = toku_cachefile_get_fd (ft->cf);
-        toku_maybe_truncate_file_on_open(ft->blocktable, fd);
+        ft->blocktable.maybe_truncate_file_on_open(fd);
     }
 
     r = 0;
@@ -4528,17 +4529,15 @@ toku_dump_ftnode (FILE *file, FT_HANDLE ft_handle, BLOCKNUM blocknum, int depth,
     return result;
 }
 
-int toku_dump_ft (FILE *f, FT_HANDLE ft_handle) {
-    int r;
-    assert(ft_handle->ft);
-    toku_dump_translation_table(f, ft_handle->ft->blocktable);
-    {
-        uint32_t fullhash = 0;
-        CACHEKEY root_key;
-        toku_calculate_root_offset_pointer(ft_handle->ft, &root_key, &fullhash);
-        r = toku_dump_ftnode(f, ft_handle, root_key, 0, 0, 0);
-    }
-    return r;
+int toku_dump_ft(FILE *f, FT_HANDLE ft_handle) {
+    FT ft = ft_handle->ft;
+    invariant_notnull(ft);
+    ft->blocktable.dump_translation_table(f);
+
+    uint32_t fullhash = 0;
+    CACHEKEY root_key;
+    toku_calculate_root_offset_pointer(ft_handle->ft, &root_key, &fullhash);
+    return toku_dump_ftnode(f, ft_handle, root_key, 0, 0, 0);
 }
 
 int toku_ft_layer_init(void) {
@@ -4630,18 +4629,15 @@ void toku_ft_unlink(FT_HANDLE handle) {
     toku_cachefile_unlink_on_close(cf);
 }
 
-int
-toku_ft_get_fragmentation(FT_HANDLE ft_handle, TOKU_DB_FRAGMENTATION report) {
-    int r;
-
+int toku_ft_get_fragmentation(FT_HANDLE ft_handle, TOKU_DB_FRAGMENTATION report) {
     int fd = toku_cachefile_get_fd(ft_handle->ft->cf);
     toku_ft_lock(ft_handle->ft);
 
     int64_t file_size;
-    r = toku_os_get_file_size(fd, &file_size);
-    if (r==0) {
+    int r = toku_os_get_file_size(fd, &file_size);
+    if (r == 0) {
         report->file_size_bytes = file_size;
-        toku_block_table_get_fragmentation_unlocked(ft_handle->ft->blocktable, report);
+        ft_handle->ft->blocktable.get_fragmentation_unlocked(report);
     }
     toku_ft_unlock(ft_handle->ft);
     return r;

ft/ft.cc

@@ -121,7 +121,7 @@ ft_destroy(FT ft) {
     //header and checkpoint_header have same Blocktable pointer
     //cannot destroy since it is still in use by CURRENT
     assert(ft->h->type == FT_CURRENT);
-    toku_blocktable_destroy(&ft->blocktable);
+    ft->blocktable.destroy();
     ft->cmp.destroy();
     toku_destroy_dbt(&ft->descriptor.dbt);
     toku_destroy_dbt(&ft->cmp_descriptor.dbt);
@@ -203,7 +203,7 @@ static void ft_begin_checkpoint (LSN checkpoint_lsn, void *header_v) {
     assert(ft->checkpoint_header == NULL);
     ft_copy_for_checkpoint_unlocked(ft, checkpoint_lsn);
     ft->h->dirty = 0;             // this is only place this bit is cleared        (in currentheader)
-    toku_block_translation_note_start_checkpoint_unlocked(ft->blocktable);
+    ft->blocktable.note_start_checkpoint_unlocked();
     toku_ft_unlock (ft);
 }
 
@@ -239,8 +239,6 @@ ft_hack_highest_unused_msn_for_upgrade_for_checkpoint(FT ft) {
 static void ft_checkpoint (CACHEFILE cf, int fd, void *header_v) {
     FT ft = (FT) header_v;
     FT_HEADER ch = ft->checkpoint_header;
-    //printf("%s:%d allocated_limit=%lu writing queue to %lu\n", __FILE__, __LINE__,
-    //             block_allocator_allocated_limit(h->block_allocator), h->unused_blocks.b*h->nodesize);
     assert(ch);
     assert(ch->type == FT_CHECKPOINT_INPROGRESS);
     if (ch->dirty) {            // this is only place this bit is tested (in checkpoint_header)
@@ -255,16 +253,15 @@ static void ft_checkpoint (CACHEFILE cf, int fd, void *header_v) {
         ft_hack_highest_unused_msn_for_upgrade_for_checkpoint(ft);
                                                              
         // write translation and header to disk (or at least to OS internal buffer)
-        toku_serialize_ft_to(fd, ch, ft->blocktable, ft->cf);
+        toku_serialize_ft_to(fd, ch, &ft->blocktable, ft->cf);
         ch->dirty = 0;                      // this is only place this bit is cleared (in checkpoint_header)
         
         // fsync the cachefile
         toku_cachefile_fsync(cf);
         ft->h->checkpoint_count++;        // checkpoint succeeded, next checkpoint will save to alternate header location
         ft->h->checkpoint_lsn = ch->checkpoint_lsn;  //Header updated.
-    } 
-    else {
-        toku_block_translation_note_skipped_checkpoint(ft->blocktable);
+    } else {
+        ft->blocktable.note_skipped_checkpoint();
     }
 }
 
@@ -272,14 +269,12 @@ static void ft_checkpoint (CACHEFILE cf, int fd, void *header_v) {
 // free unused disk space 
 // (i.e. tell BlockAllocator to liberate blocks used by previous checkpoint).
 // Must have access to fd (protected)
-static void ft_end_checkpoint (CACHEFILE UU(cachefile), int fd, void *header_v) {
+static void ft_end_checkpoint(CACHEFILE UU(cf), int fd, void *header_v) {
     FT ft = (FT) header_v;
     assert(ft->h->type == FT_CURRENT);
-    toku_block_translation_note_end_checkpoint(ft->blocktable, fd);
-    if (ft->checkpoint_header) {
-        toku_free(ft->checkpoint_header);
-        ft->checkpoint_header = NULL;
-    }
+    ft->blocktable.note_end_checkpoint(fd);
+    toku_free(ft->checkpoint_header);
+    ft->checkpoint_header = nullptr;
 }
 
 // maps to cf->close_userdata
@@ -407,7 +402,7 @@ static void ft_init(FT ft, FT_OPTIONS options, CACHEFILE cf) {
                                 ft_note_pin_by_checkpoint,
                                 ft_note_unpin_by_checkpoint);
 
-    toku_block_verify_no_free_blocknums(ft->blocktable);
+    ft->blocktable.verify_no_free_blocknums();
 }
 
 
@@ -456,8 +451,8 @@ void toku_ft_create(FT *ftp, FT_OPTIONS options, CACHEFILE cf, TOKUTXN txn) {
     toku_ft_init_reflock(ft);
 
     // Assign blocknum for root block, also dirty the header
-    toku_blocktable_create_new(&ft->blocktable);
-    toku_allocate_blocknum(ft->blocktable, &ft->h->root_blocknum, ft);
+    ft->blocktable.create();
+    ft->blocktable.allocate_blocknum(&ft->h->root_blocknum, ft);
 
     ft_init(ft, options, cf);
 
@@ -875,14 +870,13 @@ toku_ft_stat64 (FT ft, struct ftstat64_s *s) {
     s->verify_time_sec = ft->h->time_of_last_verification;    
 }
 
-void
-toku_ft_get_fractal_tree_info64(FT ft, struct ftinfo64 *s) {
-    toku_blocktable_get_info64(ft->blocktable, s);
+void toku_ft_get_fractal_tree_info64(FT ft, struct ftinfo64 *info) {
+    ft->blocktable.get_info64(info);
 }
 
 int toku_ft_iterate_fractal_tree_block_map(FT ft, int (*iter)(uint64_t,int64_t,int64_t,int64_t,int64_t,void*), void *iter_extra) {
     uint64_t this_checkpoint_count = ft->h->checkpoint_count;
-    return toku_blocktable_iterate_translation_tables(ft->blocktable, this_checkpoint_count, iter, iter_extra);
+    return ft->blocktable.iterate_translation_tables(this_checkpoint_count, iter, iter_extra);
 }
 
 void 
@@ -908,7 +902,7 @@ toku_ft_update_descriptor_with_fd(FT ft, DESCRIPTOR desc, int fd) {
     // make space for the new descriptor and write it out to disk
     DISKOFF offset, size;
     size = toku_serialize_descriptor_size(desc) + 4;
-    toku_realloc_descriptor_on_disk(ft->blocktable, size, &offset, ft, fd);
+    ft->blocktable.realloc_descriptor_on_disk(size, &offset, ft, fd);
     toku_serialize_descriptor_contents_to_fd(fd, desc, offset);
 
     // cleanup the old descriptor and set the in-memory descriptor to the new one
@@ -1086,7 +1080,7 @@ void toku_ft_get_garbage(FT ft, uint64_t *total_space, uint64_t *used_space) {
         .total_space = 0,
         .used_space = 0
     };
-    toku_blocktable_iterate(ft->blocktable, TRANSLATION_CHECKPOINTED, garbage_helper, &info, true, true);
+    ft->blocktable.iterate(block_table::TRANSLATION_CHECKPOINTED, garbage_helper, &info, true, true);
     *total_space = info.total_space;
     *used_space = info.used_space;
 }

ft/ft.h

@@ -111,6 +111,9 @@ void toku_ft_destroy_reflock(FT ft);
 void toku_ft_grab_reflock(FT ft);
 void toku_ft_release_reflock(FT ft);
 
+void toku_ft_lock(struct ft *ft);
+void toku_ft_unlock(struct ft *ft);
+
 void toku_ft_create(FT *ftp, FT_OPTIONS options, CACHEFILE cf, TOKUTXN txn);
 void toku_ft_free (FT ft);
 

ft/logger/logger.cc

@@ -269,32 +269,30 @@ bool toku_logger_rollback_is_open (TOKULOGGER logger) {
 
 #define MAX_CACHED_ROLLBACK_NODES 4096
 
-void
-toku_logger_initialize_rollback_cache(TOKULOGGER logger, FT ft) {
-    toku_free_unused_blocknums(ft->blocktable, ft->h->root_blocknum);
+void toku_logger_initialize_rollback_cache(TOKULOGGER logger, FT ft) {
+    ft->blocktable.free_unused_blocknums(ft->h->root_blocknum);
     logger->rollback_cache.init(MAX_CACHED_ROLLBACK_NODES);
 }
 
-int
-toku_logger_open_rollback(TOKULOGGER logger, CACHETABLE cachetable, bool create) {
+int toku_logger_open_rollback(TOKULOGGER logger, CACHETABLE cachetable, bool create) {
     assert(logger->is_open);
     assert(!logger->rollback_cachefile);
 
-    FT_HANDLE t = NULL;   // Note, there is no DB associated with this FT.
-    toku_ft_handle_create(&t);
-    int r = toku_ft_handle_open(t, toku_product_name_strings.rollback_cachefile, create, create, cachetable, nullptr);
+    FT_HANDLE ft_handle = nullptr;   // Note, there is no DB associated with this FT.
+    toku_ft_handle_create(&ft_handle);
+    int r = toku_ft_handle_open(ft_handle, toku_product_name_strings.rollback_cachefile, create, create, cachetable, nullptr);
     if (r == 0) {
-        logger->rollback_cachefile = t->ft->cf;
-        toku_logger_initialize_rollback_cache(logger, t->ft);
-
-        //Verify it is empty
-        //Must have no data blocks (rollback logs or otherwise).
-        toku_block_verify_no_data_blocks_except_root(t->ft->blocktable, t->ft->h->root_blocknum);
-        bool is_empty;
-        is_empty = toku_ft_is_empty_fast(t);
+        FT ft = ft_handle->ft;
+        logger->rollback_cachefile = ft->cf;
+        toku_logger_initialize_rollback_cache(logger, ft_handle->ft);
+
+        // Verify it is empty
+        // Must have no data blocks (rollback logs or otherwise).
+        ft->blocktable.verify_no_data_blocks_except_root(ft->h->root_blocknum);
+        bool is_empty = toku_ft_is_empty_fast(ft_handle);
         assert(is_empty);
     } else {
-        toku_ft_handle_close(t);
+        toku_ft_handle_close(ft_handle);
     }
     return r;
 }
@@ -314,9 +312,9 @@ void toku_logger_close_rollback_check_empty(TOKULOGGER logger, bool clean_shutdo
             if (clean_shutdown) {
                 //Verify it is safe to close it.
                 assert(!ft->h->dirty);  //Must not be dirty.
-                toku_free_unused_blocknums(ft->blocktable, ft->h->root_blocknum);
-                //Must have no data blocks (rollback logs or otherwise).
-                toku_block_verify_no_data_blocks_except_root(ft->blocktable, ft->h->root_blocknum);
+                ft->blocktable.free_unused_blocknums(ft->h->root_blocknum);
+                // Must have no data blocks (rollback logs or otherwise).
+                ft->blocktable.verify_no_data_blocks_except_root(ft->h->root_blocknum);
                 assert(!ft->h->dirty);
             } else {
                 ft->h->dirty = 0;

ft/serialize/block_allocator.cc

@@ -114,7 +114,7 @@ static inline bool ba_trace_enabled() {
 #endif
 }
 
-void block_allocator::create(uint64_t reserve_at_beginning, uint64_t alignment) {
+void block_allocator::_create_internal(uint64_t reserve_at_beginning, uint64_t alignment) {
     // the alignment must be at least 512 and aligned with 512 to work with direct I/O
     assert(alignment >= 512 && (alignment % 512) == 0);
 
@@ -127,7 +127,10 @@ void block_allocator::create(uint64_t reserve_at_beginning, uint64_t alignment)
     _strategy = BA_STRATEGY_FIRST_FIT;
 
     VALIDATE();
+}
 
+void block_allocator::create(uint64_t reserve_at_beginning, uint64_t alignment) {
+    _create_internal(reserve_at_beginning, alignment);
     if (ba_trace_enabled()) {
         fprintf(stderr, "ba_trace_create %p\n", this);
     }
@@ -161,41 +164,6 @@ void block_allocator::grow_blocks_array() {
     grow_blocks_array_by(1);
 }
 
-void block_allocator::merge_blockpairs_into(uint64_t d, struct blockpair dst[],
-                                            uint64_t s, const struct blockpair src[])
-{
-    uint64_t tail = d+s;
-    while (d > 0 && s > 0) {
-        struct blockpair       *dp = &dst[d - 1];
-        struct blockpair const *sp = &src[s - 1];
-        struct blockpair       *tp = &dst[tail - 1];
-        assert(tail > 0);
-        if (dp->offset > sp->offset) {
-            *tp = *dp;
-            d--;
-            tail--;
-        } else {
-            *tp = *sp;
-            s--;
-            tail--;
-        }
-    }
-    while (d > 0) {
-        struct blockpair *dp = &dst[d - 1];
-        struct blockpair *tp = &dst[tail - 1];
-        *tp = *dp;
-        d--;
-        tail--;
-    }
-    while (s > 0) {
-        struct blockpair const *sp = &src[s - 1];
-        struct blockpair       *tp = &dst[tail - 1];
-        *tp = *sp;
-        s--;
-        tail--;
-    }
-}
-
 int block_allocator::compare_blockpairs(const void *av, const void *bv) {
     const struct blockpair *a = (const struct blockpair *) av;
     const struct blockpair *b = (const struct blockpair *) bv;
@@ -208,30 +176,25 @@ int block_allocator::compare_blockpairs(const void *av, const void *bv) {
     }
 }
 
-// See the documentation in block_allocator.h
-void block_allocator::alloc_blocks_at(uint64_t n_blocks, struct blockpair pairs[]) {
-    VALIDATE();
-    qsort(pairs, n_blocks, sizeof(*pairs), compare_blockpairs);
-    for (uint64_t i = 0; i < n_blocks; i++) {
-        assert(pairs[i].offset >= _reserve_at_beginning);
-        assert(pairs[i].offset % _alignment == 0);
-        _n_bytes_in_use += pairs[i].size;
+void block_allocator::create_from_blockpairs(uint64_t reserve_at_beginning, uint64_t alignment,
+                                             struct blockpair *pairs, uint64_t n_blocks) {
+    _create_internal(reserve_at_beginning, alignment);
+
+    for (uint64_t i = 0; i < _n_blocks; i++) {
         // Allocator does not support size 0 blocks. See block_allocator_free_block.
         invariant(pairs[i].size > 0);
+        invariant(pairs[i].offset >= _reserve_at_beginning);
+        invariant(pairs[i].offset % _alignment == 0);
+
+        _n_bytes_in_use += pairs[i].size;
     }
-    grow_blocks_array_by(n_blocks);
-    merge_blockpairs_into(_n_blocks, _blocks_array, n_blocks, pairs);
-    _n_blocks += n_blocks;
-    VALIDATE();
-}
+    _n_blocks = n_blocks;
 
-void block_allocator::alloc_block_at(uint64_t size, uint64_t offset) {
-    struct blockpair p(offset, size);
+    grow_blocks_array_by(_n_blocks);
+    memcpy(_blocks_array, pairs, _n_blocks * sizeof(struct blockpair));
+    qsort(_blocks_array, _n_blocks, sizeof(struct blockpair), compare_blockpairs);
 
-    // Just do a linear search for the block.
-    // This data structure is a sorted array (no gaps or anything), so the search isn't really making this any slower than the insertion.
-    // To speed up the insertion when opening a file, we provide the block_allocator_alloc_blocks_at function.
-    alloc_blocks_at(1, &p);
+    VALIDATE();
 }
 
 // Effect: align a value by rounding up.

ft/serialize/block_allocator.h

@@ -128,6 +128,14 @@ public:
         BA_STRATEGY_FIRST_FIT = 1
     };
 
+    struct blockpair {
+        uint64_t offset;
+        uint64_t size;
+        blockpair(uint64_t o, uint64_t s) :
+            offset(o), size(s) {
+        }
+    };
+
     // Effect: Create a block allocator, in which the first RESERVE_AT_BEGINNING bytes are not put into a block.
     //         The default allocation strategy is first fit (BA_STRATEGY_FIRST_FIT)
     //  All blocks be start on a multiple of ALIGNMENT.
@@ -137,6 +145,19 @@ public:
     //  alignment (IN)                   Block alignment.
     void create(uint64_t reserve_at_beginning, uint64_t alignment);
 
+    // Effect: Create a block allocator, in which the first RESERVE_AT_BEGINNING bytes are not put into a block.
+    //         The default allocation strategy is first fit (BA_STRATEGY_FIRST_FIT)
+    //         The allocator is initialized to contain `n_blocks' of blockpairs, taken from `pairs'
+    //  All blocks be start on a multiple of ALIGNMENT.
+    //  Aborts if we run out of memory.
+    // Parameters
+    //  pairs,                           unowned array of pairs to copy
+    //  n_blocks,                        Size of pairs array
+    //  reserve_at_beginning (IN)        Size of reserved block at beginning.  This size does not have to be aligned.
+    //  alignment (IN)                   Block alignment.
+    void create_from_blockpairs(uint64_t reserve_at_beginning, uint64_t alignment,
+                                struct blockpair *pairs, uint64_t n_blocks);
+
     // Effect: Destroy this block allocator
     void destroy();
 
@@ -144,35 +165,10 @@ public:
     // Requires: No other threads are operating on this block allocator
     void set_strategy(enum allocation_strategy strategy);
 
-    // Effect: Allocate a block of the specified size at a particular offset.
-    //  Aborts if anything goes wrong.
-    //  The performance of this function may be as bad as Theta(N), where N is the number of blocks currently in use.
-    // Usage note: To allocate several blocks (e.g., when opening a FT),  use block_allocator_alloc_blocks_at().
-    // Requires: The resulting block may not overlap any other allocated block.
-    //  And the offset must be a multiple of the block alignment.
-    // Parameters:
-    //  size (IN):   The size of the block.
-    //  offset (IN): The location of the block.
-    void alloc_block_at(uint64_t size, uint64_t offset);
-
-    struct blockpair {
-        uint64_t offset;
-        uint64_t size;
-        blockpair(uint64_t o, uint64_t s) :
-            offset(o), size(s) {
-        }
-    };
-
-    // Effect: Take pairs in any order, and add them all, as if we did block_allocator_alloc_block() on each pair.
-    //  This should run in time O(N + M log M) where N is the number of blocks in ba, and M is the number of new blocks.
-    // Modifies: pairs (sorts them).
-    void alloc_blocks_at(uint64_t n_blocks, blockpair *pairs);
-
     // Effect: Allocate a block of the specified size at an address chosen by the allocator.
     //  Aborts if anything goes wrong.
     //  The block address will be a multiple of the alignment.
     // Parameters:
-    //  ba (IN/OUT):  The block allocator.   (Modifies ba.)
     //  size (IN):    The size of the block.  (The size does not have to be aligned.)
     //  offset (OUT): The location of the block.
     void alloc_block(uint64_t size, uint64_t *offset);
@@ -180,14 +176,12 @@ public:
     // Effect: Free the block at offset.
     // Requires: There must be a block currently allocated at that offset.
     // Parameters:
-    //  ba (IN/OUT): The block allocator.  (Modifies ba.)
     //  offset (IN): The offset of the block.
     void free_block(uint64_t offset);
 
     // Effect: Return the size of the block that starts at offset.
     // Requires: There must be a block currently allocated at that offset.
     // Parameters:
-    //  ba (IN/OUT): The block allocator.  (Modifies ba.)
     //  offset (IN): The offset of the block.
     uint64_t block_size(uint64_t offset);
 
@@ -221,18 +215,8 @@ public:
     //  report->checkpoint_bytes_additional is ignored on return
     void get_statistics(TOKU_DB_FRAGMENTATION report);
 
-    // Effect: Merge dst[d] and src[s] into dst[d+s], merging in place.
-    //   Initially dst and src hold sorted arrays (sorted by increasing offset).
-    //   Finally dst contains all d+s elements sorted in order.
-    // Requires: 
-    //   dst and src are sorted.
-    //   dst must be large enough (sizeof(dst) >= d && sizeof(src) >= s)
-    //   No blocks may overlap.
-    // Rationale: This is exposed so it can be tested by a glass box tester.
-    static void merge_blockpairs_into(uint64_t d, struct blockpair dst[],
-                                      uint64_t s, const struct blockpair src[]);
-
 private:
+    void _create_internal(uint64_t reserve_at_beginning, uint64_t alignment);
     void grow_blocks_array_by(uint64_t n_to_add);
     void grow_blocks_array();
     int64_t find_block(uint64_t offset);

ft/serialize/ft-serialize.cc

@@ -161,12 +161,12 @@ deserialize_descriptor_from_rbuf(struct rbuf *rb, DESCRIPTOR desc, int layout_ve
 }
 
 static int
-deserialize_descriptor_from(int fd, BLOCK_TABLE bt, DESCRIPTOR desc, int layout_version) {
+deserialize_descriptor_from(int fd, block_table *bt, DESCRIPTOR desc, int layout_version) {
     int r = 0;
     DISKOFF offset;
     DISKOFF size;
-    unsigned char *dbuf = NULL;
-    toku_get_descriptor_offset_size(bt, &offset, &size);
+    unsigned char *dbuf = nullptr;
+    bt->get_descriptor_offset_size(&offset, &size);
     memset(desc, 0, sizeof(*desc));
     if (size > 0) {
         lazy_assert(size>=4); //4 for checksum
@@ -274,11 +274,10 @@ int deserialize_ft_versioned(int fd, struct rbuf *rb, FT *ftp, uint32_t version)
             assert(readsz <= (ssize_t)size_to_read);
         }
         // Create table and read in data.
-        r = toku_blocktable_create_from_buffer(fd,
-                                               &ft->blocktable,
-                                               translation_address_on_disk,
-                                               translation_size_on_disk,
-                                               tbuf);
+        r = ft->blocktable.create_from_buffer(fd,
+                                              translation_address_on_disk,
+                                              translation_size_on_disk,
+                                              tbuf);
         toku_free(tbuf);
         if (r != 0) {
             goto exit;
@@ -426,7 +425,7 @@ int deserialize_ft_versioned(int fd, struct rbuf *rb, FT *ftp, uint32_t version)
     }
 
     invariant((uint32_t) ft->layout_version_read_from_disk == version);
-    r = deserialize_descriptor_from(fd, ft->blocktable, &ft->descriptor, version);
+    r = deserialize_descriptor_from(fd, &ft->blocktable, &ft->descriptor, version);
     if (r != 0) {
         goto exit;
     }
@@ -804,18 +803,20 @@ void toku_serialize_ft_to_wbuf (
     lazy_assert(wbuf->ndone == wbuf->size);
 }
 
-void toku_serialize_ft_to (int fd, FT_HEADER h, BLOCK_TABLE blocktable, CACHEFILE cf) {
+void toku_serialize_ft_to(int fd, FT_HEADER h, block_table *bt, CACHEFILE cf) {
     lazy_assert(h->type==FT_CHECKPOINT_INPROGRESS);
     struct wbuf w_translation;
     int64_t size_translation;
     int64_t address_translation;
 
-    //Must serialize translation first, to get address,size for header.
-    toku_serialize_translation_to_wbuf(blocktable, fd, &w_translation,
-                                               &address_translation,
-                                               &size_translation);
-    assert(size_translation == w_translation.ndone); // the bytes written are the size
-    assert(w_translation.size % 512 == 0);           // the number of bytes available in the buffer is 0 mod 512, and those last bytes are all initialized.
+    // Must serialize translation first, to get address,size for header.
+    bt->serialize_translation_to_wbuf(fd, &w_translation,
+                                      &address_translation,
+                                      &size_translation);
+    assert(size_translation == w_translation.ndone);
+
+    // the number of bytes available in the buffer is 0 mod 512, and those last bytes are all initialized.
+    assert(w_translation.size % 512 == 0);
 
     struct wbuf w_main;
     size_t size_main       = toku_serialize_ft_size(h);

ft/serialize/ft_node-serialize.cc

@@ -845,8 +845,8 @@ toku_serialize_ftnode_to (int fd, BLOCKNUM blocknum, FTNODE node, FTNODE_DISK_DA
     invariant(blocknum.b>=0);
     DISKOFF offset;
 
-    toku_blocknum_realloc_on_disk(ft->blocktable, blocknum, n_to_write, &offset,
-                                  ft, fd, for_checkpoint); //dirties h
+    ft->blocktable.realloc_on_disk(blocknum, n_to_write, &offset,
+                                   ft, fd, for_checkpoint); //dirties h
 
     tokutime_t t0 = toku_time_now();
     toku_os_full_pwrite(fd, compressed_buf, n_to_write, offset);
@@ -1085,7 +1085,7 @@ void read_block_from_fd_into_rbuf(
 {
     // get the file offset and block size for the block
     DISKOFF offset, size;
-    toku_translate_blocknum_to_offset_size(ft->blocktable, blocknum, &offset, &size);
+    ft->blocktable.translate_blocknum_to_offset_size(blocknum, &offset, &size);
     DISKOFF size_aligned = roundup_to_multiple(512, size);
     uint8_t *XMALLOC_N_ALIGNED(512, size_aligned, raw_block);
     rbuf_init(rb, raw_block, size);
@@ -1101,11 +1101,12 @@ static const int read_header_heuristic_max = 32*1024;
 #define MIN(a,b) (((a)>(b)) ? (b) : (a))
 #endif
 
-static void read_ftnode_header_from_fd_into_rbuf_if_small_enough (int fd, BLOCKNUM blocknum, FT ft, struct rbuf *rb, struct ftnode_fetch_extra *bfe)
 // Effect: If the header part of the node is small enough, then read it into the rbuf.  The rbuf will be allocated to be big enough in any case.
-{
+static void read_ftnode_header_from_fd_into_rbuf_if_small_enough(int fd, BLOCKNUM blocknum,
+                                                                 FT ft, struct rbuf *rb,
+                                                                 struct ftnode_fetch_extra *bfe) {
     DISKOFF offset, size;
-    toku_translate_blocknum_to_offset_size(ft->blocktable, blocknum, &offset, &size);
+    ft->blocktable.translate_blocknum_to_offset_size(blocknum, &offset, &size);
     DISKOFF read_size = roundup_to_multiple(512, MIN(read_header_heuristic_max, size));
     uint8_t *XMALLOC_N_ALIGNED(512, roundup_to_multiple(512, size), raw_block);
     rbuf_init(rb, raw_block, read_size);
@@ -1937,10 +1938,8 @@ deserialize_and_upgrade_ftnode(FTNODE node,
     // we read the different sub-sections.
     // get the file offset and block size for the block
     DISKOFF offset, size;
-    toku_translate_blocknum_to_offset_size(bfe->ft->blocktable,
-                                           blocknum,
-                                           &offset,
-                                           &size);
+    bfe->ft->blocktable.translate_blocknum_to_offset_size(blocknum, &offset, &size);
+
     struct rbuf rb;
     r = read_and_decompress_block_from_fd_into_rbuf(fd,
                                                     blocknum,
@@ -2218,16 +2217,13 @@ toku_deserialize_bp_from_disk(FTNODE node, FTNODE_DISK_DATA ndd, int childnum, i
     // 
     // get the file offset and block size for the block
     DISKOFF node_offset, total_node_disk_size;
-    toku_translate_blocknum_to_offset_size(
-        bfe->ft->blocktable, 
-        node->blocknum, 
-        &node_offset, 
-        &total_node_disk_size
-        );
+    bfe->ft->blocktable.translate_blocknum_to_offset_size(node->blocknum, &node_offset, &total_node_disk_size);
 
     uint32_t curr_offset = BP_START(ndd, childnum);
-    uint32_t curr_size   = BP_SIZE (ndd, childnum);
-    struct rbuf rb = {.buf = NULL, .size = 0, .ndone = 0};
+    uint32_t curr_size = BP_SIZE (ndd, childnum);
+
+    struct rbuf rb;
+    rbuf_init(&rb, nullptr, 0);
 
     uint32_t pad_at_beginning = (node_offset+curr_offset)%512;
     uint32_t padded_size = roundup_to_multiple(512, pad_at_beginning + curr_size);
@@ -2530,20 +2526,22 @@ toku_serialize_rollback_log_to (int fd, ROLLBACK_LOG_NODE log, SERIALIZED_ROLLBA
         serialized_log = &serialized_local;
         toku_serialize_rollback_log_to_memory_uncompressed(log, serialized_log);
     }
+
     BLOCKNUM blocknum = serialized_log->blocknum;
+    invariant(blocknum.b >= 0);
 
-    //Compress and malloc buffer to write
+    // Compress and malloc buffer to write
     serialize_uncompressed_block_to_memory(serialized_log->data,
-            serialized_log->n_sub_blocks, serialized_log->sub_block,
-            ft->h->compression_method, &n_to_write, &compressed_buf);
+                                           serialized_log->n_sub_blocks,
+                                           serialized_log->sub_block,
+                                           ft->h->compression_method,
+                                           &n_to_write, &compressed_buf);
 
-    {
-        lazy_assert(blocknum.b>=0);
-        DISKOFF offset;
-        toku_blocknum_realloc_on_disk(ft->blocktable, blocknum, n_to_write, &offset,
-                                      ft, fd, for_checkpoint); //dirties h
-        toku_os_full_pwrite(fd, compressed_buf, n_to_write, offset);
-    }
+    // Dirties the ft
+    DISKOFF offset;
+    ft->blocktable.realloc_on_disk(blocknum, n_to_write, &offset,
+                                   ft, fd, for_checkpoint);
+    toku_os_full_pwrite(fd, compressed_buf, n_to_write, offset);
     toku_free(compressed_buf);
     if (!is_serialized) {
         toku_static_serialized_rollback_log_destroy(&serialized_local);
@@ -2803,16 +2801,19 @@ cleanup:
     return r;
 }
 
-// Read rollback log node from file into struct.  Perform version upgrade if necessary.
-int
-toku_deserialize_rollback_log_from (int fd, BLOCKNUM blocknum, ROLLBACK_LOG_NODE *logp, FT ft) {
+// Read rollback log node from file into struct.
+// Perform version upgrade if necessary.
+int toku_deserialize_rollback_log_from(int fd, BLOCKNUM blocknum, ROLLBACK_LOG_NODE *logp, FT ft) {
     int layout_version = 0;
     int r;
-    struct rbuf rb = {.buf = NULL, .size = 0, .ndone = 0};
+
+    struct rbuf rb;
+    rbuf_init(&rb, nullptr, 0);
 
     // get the file offset and block size for the block
     DISKOFF offset, size;
-    toku_translate_blocknum_to_offset_size(ft->blocktable, blocknum, &offset, &size);
+    ft->blocktable.translate_blocknum_to_offset_size(blocknum, &offset, &size);
+
     // if the size is 0, then the blocknum is unused
     if (size == 0) {
         // blocknum is unused, just create an empty one and get out
@@ -2838,7 +2839,9 @@ toku_deserialize_rollback_log_from (int fd, BLOCKNUM blocknum, ROLLBACK_LOG_NODE
     r = deserialize_rollback_log_from_rbuf_versioned(layout_version, blocknum, logp, &rb);
 
 cleanup:
-    if (rb.buf) toku_free(rb.buf);
+    if (rb.buf) {
+        toku_free(rb.buf);
+    }
     return r;
 }
 

ft/tests/block_allocator_test.cc

@@ -90,12 +90,6 @@ PATENT RIGHTS GRANT:
 
 #include "test.h"
 
-static void ba_alloc_at(block_allocator *ba, uint64_t size, uint64_t offset) {
-    ba->validate();
-    ba->alloc_block_at(size * 512, offset * 512);
-    ba->validate();
-}
-
 static void ba_alloc(block_allocator *ba, uint64_t size, uint64_t *answer) {
     ba->validate();
     uint64_t actual_answer;
@@ -133,28 +127,8 @@ static void
 test_ba0 (void) {
     block_allocator allocator;
     block_allocator *ba = &allocator;
-    uint64_t b0, b1;
     ba->create(100*512, 1*512);
     assert(ba->allocated_limit()==100*512);
-    ba_alloc_at(ba, 50, 100);
-    assert(ba->allocated_limit()==150*512);
-    ba_alloc_at(ba, 25, 150);
-    ba_alloc   (ba, 10, &b0);
-    ba_check_l (ba, 0, 0,   100);
-    ba_check_l (ba, 1, 100,  50);
-    ba_check_l (ba, 2, 150,  25);
-    ba_check_l (ba, 3, b0,  10);
-    ba_check_none (ba, 4);
-    assert(b0==175);
-    ba_free(ba, 150);
-    ba_alloc_at(ba, 10, 150);
-    ba_alloc(ba, 10, &b0);
-    assert(b0==160);
-    ba_alloc(ba, 10, &b0);
-    ba_alloc(ba, 113, &b1);
-    assert(113*512==ba->block_size(b1 *512));
-    assert(10 *512==ba->block_size(b0 *512));
-    assert(50 *512==ba->block_size(100*512));
 
     uint64_t b2, b3, b4, b5, b6, b7;
     ba_alloc(ba, 100, &b2);     

ft/tests/ft-bfe-query.cc

@@ -422,22 +422,22 @@ test_prefetching(void) {
                  16);
     ft_h->cmp.create(int64_key_cmp, nullptr);
     ft->ft = ft_h;
-    toku_blocktable_create_new(&ft_h->blocktable);
+    ft_h->blocktable.create();
     { int r_truncate = ftruncate(fd, 0); CKERR(r_truncate); }
     //Want to use block #20
     BLOCKNUM b = make_blocknum(0);
     while (b.b < 20) {
-        toku_allocate_blocknum(ft_h->blocktable, &b, ft_h);
+        ft_h->blocktable.allocate_blocknum(&b, ft_h);
     }
     assert(b.b == 20);
 
     {
         DISKOFF offset;
         DISKOFF size;
-        toku_blocknum_realloc_on_disk(ft_h->blocktable, b, 100, &offset, ft_h, fd, false);
+        ft_h->blocktable.realloc_on_disk(b, 100, &offset, ft_h, fd, false);
         assert(offset==(DISKOFF)block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
 
-        toku_translate_blocknum_to_offset_size(ft_h->blocktable, b, &offset, &size);
+        ft_h->blocktable.translate_blocknum_to_offset_size(b, &offset, &size);
         assert(offset == (DISKOFF)block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
         assert(size   == 100);
     }
@@ -450,8 +450,8 @@ test_prefetching(void) {
 
     toku_destroy_ftnode_internals(&sn);
 
-    toku_block_free(ft_h->blocktable, block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
-    toku_blocktable_destroy(&ft_h->blocktable);
+    ft_h->blocktable.block_free(block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
+    ft_h->blocktable.destroy();
     ft_h->cmp.destroy();
     toku_free(ft_h->h);
     toku_free(ft_h);

ft/tests/ft-clock-test.cc

@@ -358,22 +358,22 @@ test_serialize_nonleaf(void) {
     ft_h->cmp.create(string_key_cmp, nullptr);
     ft->ft = ft_h;
     
-    toku_blocktable_create_new(&ft_h->blocktable);
+    ft_h->blocktable.create();
     { int r_truncate = ftruncate(fd, 0); CKERR(r_truncate); }
     //Want to use block #20
     BLOCKNUM b = make_blocknum(0);
     while (b.b < 20) {
-        toku_allocate_blocknum(ft_h->blocktable, &b, ft_h);
+        ft_h->blocktable.allocate_blocknum(&b, ft_h);
     }
     assert(b.b == 20);
 
     {
         DISKOFF offset;
         DISKOFF size;
-        toku_blocknum_realloc_on_disk(ft_h->blocktable, b, 100, &offset, ft_h, fd, false);
+        ft_h->blocktable.realloc_on_disk(b, 100, &offset, ft_h, fd, false);
         assert(offset==(DISKOFF)block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
 
-        toku_translate_blocknum_to_offset_size(ft_h->blocktable, b, &offset, &size);
+        ft_h->blocktable.translate_blocknum_to_offset_size(b, &offset, &size);
         assert(offset == (DISKOFF)block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
         assert(size   == 100);
     }
@@ -387,8 +387,8 @@ test_serialize_nonleaf(void) {
     toku_destroy_ftnode_internals(&sn);
     toku_free(ndd);
 
-    toku_block_free(ft_h->blocktable, block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
-    toku_blocktable_destroy(&ft_h->blocktable);
+    ft_h->blocktable.block_free(block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
+    ft_h->blocktable.destroy();
     toku_free(ft_h->h);
     ft_h->cmp.destroy();
     toku_free(ft_h);
@@ -438,22 +438,22 @@ test_serialize_leaf(void) {
                  16);
     ft->ft = ft_h;
     
-    toku_blocktable_create_new(&ft_h->blocktable);
+    ft_h->blocktable.create();
     { int r_truncate = ftruncate(fd, 0); CKERR(r_truncate); }
     //Want to use block #20
     BLOCKNUM b = make_blocknum(0);
     while (b.b < 20) {
-        toku_allocate_blocknum(ft_h->blocktable, &b, ft_h);
+        ft_h->blocktable.allocate_blocknum(&b, ft_h);
     }
     assert(b.b == 20);
 
     {
         DISKOFF offset;
         DISKOFF size;
-        toku_blocknum_realloc_on_disk(ft_h->blocktable, b, 100, &offset, ft_h, fd, false);
+        ft_h->blocktable.realloc_on_disk(b, 100, &offset, ft_h, fd, false);
         assert(offset==(DISKOFF)block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
 
-        toku_translate_blocknum_to_offset_size(ft_h->blocktable, b, &offset, &size);
+        ft_h->blocktable.translate_blocknum_to_offset_size(b, &offset, &size);
         assert(offset == (DISKOFF)block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
         assert(size   == 100);
     }
@@ -466,8 +466,8 @@ test_serialize_leaf(void) {
 
     toku_destroy_ftnode_internals(&sn);
 
-    toku_block_free(ft_h->blocktable, block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
-    toku_blocktable_destroy(&ft_h->blocktable);
+    ft_h->blocktable.block_free(block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
+    ft_h->blocktable.destroy();
     toku_free(ft_h->h);
     toku_free(ft_h);
     toku_free(ft);

ft/tests/ft-serialize-benchmark.cc

@@ -198,22 +198,22 @@ test_serialize_leaf(int valsize, int nelts, double entropy, int ser_runs, int de
     ft_h->cmp.create(long_key_cmp, nullptr);
     ft->ft = ft_h;
     
-    toku_blocktable_create_new(&ft_h->blocktable);
+    ft_h->blocktable.create();
     { int r_truncate = ftruncate(fd, 0); CKERR(r_truncate); }
     //Want to use block #20
     BLOCKNUM b = make_blocknum(0);
     while (b.b < 20) {
-        toku_allocate_blocknum(ft_h->blocktable, &b, ft_h);
+        ft_h->blocktable.allocate_blocknum(&b, ft_h);
     }
     assert(b.b == 20);
 
     {
         DISKOFF offset;
         DISKOFF size;
-        toku_blocknum_realloc_on_disk(ft_h->blocktable, b, 100, &offset, ft_h, fd, false);
+        ft_h->blocktable.realloc_on_disk(b, 100, &offset, ft_h, fd, false);
         assert(offset==(DISKOFF)block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
 
-        toku_translate_blocknum_to_offset_size(ft_h->blocktable, b, &offset, &size);
+        ft_h->blocktable.translate_blocknum_to_offset_size(b, &offset, &size);
         assert(offset == (DISKOFF)block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
         assert(size   == 100);
     }
@@ -277,8 +277,8 @@ test_serialize_leaf(int valsize, int nelts, double entropy, int ser_runs, int de
 
     toku_ftnode_free(&sn);
 
-    toku_block_free(ft_h->blocktable, block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
-    toku_blocktable_destroy(&ft_h->blocktable);
+    ft_h->blocktable.block_free(block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
+    ft_h->blocktable.destroy();
     ft_h->cmp.destroy();
     toku_free(ft_h->h);
     toku_free(ft_h);
@@ -361,22 +361,22 @@ test_serialize_nonleaf(int valsize, int nelts, double entropy, int ser_runs, int
     ft_h->cmp.create(long_key_cmp, nullptr);
     ft->ft = ft_h;
     
-    toku_blocktable_create_new(&ft_h->blocktable);
+    ft_h->blocktable.create();
     { int r_truncate = ftruncate(fd, 0); CKERR(r_truncate); }
     //Want to use block #20
     BLOCKNUM b = make_blocknum(0);
     while (b.b < 20) {
-        toku_allocate_blocknum(ft_h->blocktable, &b, ft_h);
+        ft_h->blocktable.allocate_blocknum(&b, ft_h);
     }
     assert(b.b == 20);
 
     {
         DISKOFF offset;
         DISKOFF size;
-        toku_blocknum_realloc_on_disk(ft_h->blocktable, b, 100, &offset, ft_h, fd, false);
+        ft_h->blocktable.realloc_on_disk(b, 100, &offset, ft_h, fd, false);
         assert(offset==(DISKOFF)block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
 
-        toku_translate_blocknum_to_offset_size(ft_h->blocktable, b, &offset, &size);
+        ft_h->blocktable.translate_blocknum_to_offset_size(b, &offset, &size);
         assert(offset == (DISKOFF)block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
         assert(size   == 100);
     }
@@ -412,8 +412,8 @@ test_serialize_nonleaf(int valsize, int nelts, double entropy, int ser_runs, int
     toku_ftnode_free(&dn);
     toku_destroy_ftnode_internals(&sn);
 
-    toku_block_free(ft_h->blocktable, block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
-    toku_blocktable_destroy(&ft_h->blocktable);
+    ft_h->blocktable.block_free(block_allocator::BLOCK_ALLOCATOR_TOTAL_HEADER_RESERVE);
+    ft_h->blocktable.destroy();
     toku_free(ft_h->h);
     ft_h->cmp.destroy();
     toku_free(ft_h);

ft/tests/test_block_allocator_merge.cc

-/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
-// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
-#ident "$Id$"
-/*
-COPYING CONDITIONS NOTICE:
-
-  This program is free software; you can redistribute it and/or modify
-  it under the terms of version 2 of the GNU General Public License as
-  published by the Free Software Foundation, and provided that the
-  following conditions are met:
-
-      * Redistributions of source code must retain this COPYING
-        CONDITIONS NOTICE, the COPYRIGHT NOTICE (below), the
-        DISCLAIMER (below), the UNIVERSITY PATENT NOTICE (below), the
-        PATENT MARKING NOTICE (below), and the PATENT RIGHTS
-        GRANT (below).
-
-      * Redistributions in binary form must reproduce this COPYING
-        CONDITIONS NOTICE, the COPYRIGHT NOTICE (below), the
-        DISCLAIMER (below), the UNIVERSITY PATENT NOTICE (below), the
-        PATENT MARKING NOTICE (below), and the PATENT RIGHTS
-        GRANT (below) in the documentation and/or other materials
-        provided with the distribution.
-
-  You should have received a copy of the GNU General Public License
-  along with this program; if not, write to the Free Software
-  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
-  02110-1301, USA.
-
-COPYRIGHT NOTICE:
-
-  TokuDB, Tokutek Fractal Tree Indexing Library.
-  Copyright (C) 2007-2013 Tokutek, Inc.
-
-DISCLAIMER:
-
-  This program is distributed in the hope that it will be useful, but
-  WITHOUT ANY WARRANTY; without even the implied warranty of
-  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
-  General Public License for more details.
-
-UNIVERSITY PATENT NOTICE:
-
-  The technology is licensed by the Massachusetts Institute of
-  Technology, Rutgers State University of New Jersey, and the Research
-  Foundation of State University of New York at Stony Brook under
-  United States of America Serial No. 11/760379 and to the patents
-  and/or patent applications resulting from it.
-
-PATENT MARKING NOTICE:
-
-  This software is covered by US Patent No. 8,185,551.
-  This software is covered by US Patent No. 8,489,638.
-
-PATENT RIGHTS GRANT:
-
-  "THIS IMPLEMENTATION" means the copyrightable works distributed by
-  Tokutek as part of the Fractal Tree project.
-
-  "PATENT CLAIMS" means the claims of patents that are owned or
-  licensable by Tokutek, both currently or in the future; and that in
-  the absence of this license would be infringed by THIS
-  IMPLEMENTATION or by using or running THIS IMPLEMENTATION.
-
-  "PATENT CHALLENGE" shall mean a challenge to the validity,
-  patentability, enforceability and/or non-infringement of any of the
-  PATENT CLAIMS or otherwise opposing any of the PATENT CLAIMS.
-
-  Tokutek hereby grants to you, for the term and geographical scope of
-  the PATENT CLAIMS, a non-exclusive, no-charge, royalty-free,
-  irrevocable (except as stated in this section) patent license to
-  make, have made, use, offer to sell, sell, import, transfer, and
-  otherwise run, modify, and propagate the contents of THIS
-  IMPLEMENTATION, where such license applies only to the PATENT
-  CLAIMS.  This grant does not include claims that would be infringed
-  only as a consequence of further modifications of THIS
-  IMPLEMENTATION.  If you or your agent or licensee institute or order
-  or agree to the institution of patent litigation against any entity
-  (including a cross-claim or counterclaim in a lawsuit) alleging that
-  THIS IMPLEMENTATION constitutes direct or contributory patent
-  infringement, or inducement of patent infringement, then any rights
-  granted to you under this License shall terminate as of the date
-  such litigation is filed.  If you or your agent or exclusive
-  licensee institute or order or agree to the institution of a PATENT
-  CHALLENGE, then Tokutek may terminate any rights granted to you
-  under this License.
-*/
-
-#ident "Copyright (c) 2009-2013 Tokutek Inc.  All rights reserved."
-#include "ft/serialize/block_allocator.h"
-#include <memory.h>
-#include <assert.h>
-// Test the merger.
-
-int verbose = 0;
-
-static void
-print_array (uint64_t n, const struct block_allocator::blockpair a[/*n*/]) {
-    printf("{");
-    for (uint64_t i=0; i<n; i++) printf(" %016lx", (long)a[i].offset);
-    printf("}\n");
-}
-
-static int
-compare_blockpairs (const void *av, const void *bv) {
-    const struct block_allocator::blockpair *CAST_FROM_VOIDP(a, av);
-    const struct block_allocator::blockpair *CAST_FROM_VOIDP(b, bv);
-    if (a->offset < b->offset) return -1;
-    if (a->offset > b->offset) return +1;
-    return 0;
-}
-
-static void
-test_merge (uint64_t an, const struct block_allocator::blockpair a[/*an*/],
-	    uint64_t bn, const struct block_allocator::blockpair b[/*bn*/]) {
-    if (verbose>1) { printf("a:"); print_array(an, a); }
-    if (verbose>1) { printf("b:"); print_array(bn, b); }
-    struct block_allocator::blockpair *MALLOC_N(an+bn, q);
-    struct block_allocator::blockpair *MALLOC_N(an+bn, m);
-    if (q==0 || m==0) {
-	fprintf(stderr, "malloc failed, continuing\n");
-	goto malloc_failed;
-    }
-    for (uint64_t i=0; i<an; i++) {
-	q[i] = m[i] = a[i];
-    }
-    for (uint64_t i=0; i<bn; i++) {
-	q[an+i] = b[i];
-    }
-    if (verbose) printf("qsort\n");
-    qsort(q, an+bn, sizeof(*q), compare_blockpairs);
-    if (verbose>1) { printf("q:"); print_array(an+bn, q); }
-    if (verbose) printf("merge\n");
-    block_allocator::merge_blockpairs_into(an, m, bn, b);
-    if (verbose) printf("compare\n");
-    if (verbose>1) { printf("m:"); print_array(an+bn, m); }
-    for (uint64_t i=0; i<an+bn; i++) {
-	assert(q[i].offset == m[i].offset);
-    }
- malloc_failed:
-    toku_free(q);
-    toku_free(m);
-}
-
-static uint64_t
-compute_a (uint64_t i, int mode) {
-    if (mode==0) return (((uint64_t)random()) << 32) + i;
-    if (mode==1) return 2*i;
-    if (mode==2) return i;
-    if (mode==3) return (1LL<<50) + i;
-    abort();
-}
-static uint64_t
-compute_b (uint64_t i, int mode) {
-    if (mode==0) return (((uint64_t)random()) << 32) + i;
-    if (mode==1) return 2*i+1;
-    if (mode==2) return (1LL<<50) + i;
-    if (mode==3) return i;
-    abort();
-}
-    
-
-static void
-test_merge_n_m (uint64_t n, uint64_t m, int mode)
-{
-    struct block_allocator::blockpair *MALLOC_N(n, na);
-    struct block_allocator::blockpair *MALLOC_N(m, ma);
-    if (na==0 || ma==0) {
-	fprintf(stderr, "malloc failed, continuing\n");
-	goto malloc_failed;
-    }
-    if (verbose) printf("Filling a[%" PRIu64 "]\n", n);
-    for (uint64_t i=0; i<n; i++) {
-	na[i].offset = compute_a(i, mode);
-    }
-    if (verbose) printf("Filling b[%" PRIu64 "]\n", m);
-    for (uint64_t i=0; i<m; i++) {
-	if (verbose && i % (1+m/10) == 0) { printf("."); fflush(stdout); }
-	ma[i].offset = compute_b(i, mode);
-    }
-    qsort(na, n, sizeof(*na), compare_blockpairs);
-    qsort(ma, m, sizeof(*ma), compare_blockpairs);
-    if (verbose) fprintf(stderr, "\ntest_merge\n");
-    test_merge(n, na, m, ma);
- malloc_failed:
-    toku_free(na);
-    toku_free(ma);
-}
-
-static void
-test_big_merge (void) {
-    uint64_t G = 1024LL * 1024LL * 1024LL;
-    if (toku_os_get_phys_memory_size() < 40 * G) {
-	fprintf(stderr, "Skipping big merge because there is only %4.1fGiB physical memory\n", toku_os_get_phys_memory_size()/(1024.0*1024.0*1024.0));
-    } else {
-	uint64_t twoG = 2*G;
-
-	uint64_t an = twoG;
-	uint64_t bn = 1;
-	struct block_allocator::blockpair *MALLOC_N(an+bn, a);
-        struct block_allocator::blockpair *MALLOC_N(bn,    b);
-        if (a == nullptr) {
-            fprintf(stderr, "%s:%u malloc failed, continuing\n", __FUNCTION__, __LINE__);
-            goto malloc_failed;
-        }
-        if (b == nullptr) {
-            fprintf(stderr, "%s:%u malloc failed, continuing\n", __FUNCTION__, __LINE__);
-            goto malloc_failed;
-        }
-        assert(a);
-        assert(b);
-        for (uint64_t i=0; i<an; i++) a[i].offset=i+1;
-        b[0].offset = 0;
-        block_allocator::merge_blockpairs_into(an, a, bn, b);
-        for (uint64_t i=0; i<an+bn; i++) assert(a[i].offset == i);
-    malloc_failed:
-        toku_free(a);
-        toku_free(b);
-    }
-}
-
-int main (int argc __attribute__((__unused__)), char *argv[] __attribute__((__unused__))) {
-    test_merge_n_m(4, 4, 0);
-    test_merge_n_m(16, 16, 0);
-    test_merge_n_m(0, 100, 0);
-    test_merge_n_m(100, 0, 0);
-    test_merge_n_m(1000000, 1000000, 0);
-    // Cannot run this on my laptop, or even on pointy
-#if 0
-    uint64_t too_big = 1024LL * 1024LL * 1024LL * 2;
-    test_merge_n_m(too_big, too_big);
-    test_merge_n_m(1, too_big, 0);
-#endif
-    test_big_merge();
-    return 0;
-}

ft/txn/rollback-ct-callbacks.cc

@@ -126,8 +126,7 @@ toku_rollback_flush_unused_log(
 {
     if (write_me) {
         DISKOFF offset;
-        toku_blocknum_realloc_on_disk(ft->blocktable, logname, 0, &offset,
-                                      ft, fd, for_checkpoint);
+        ft->blocktable.realloc_on_disk(logname, 0, &offset, ft, fd, for_checkpoint);
     }
     if (!keep_me && !is_clone) {
         toku_free(log);

ft/txn/rollback.cc

@@ -98,7 +98,7 @@ PATENT RIGHTS GRANT:
 
 static void rollback_unpin_remove_callback(CACHEKEY* cachekey, bool for_checkpoint, void* extra) {
     FT CAST_FROM_VOIDP(ft, extra);
-    toku_free_blocknum(ft->blocktable, cachekey, ft, for_checkpoint);
+    ft->blocktable.free_blocknum(cachekey, ft, for_checkpoint);
 }
 
 void toku_rollback_log_unpin_and_remove(TOKUTXN txn, ROLLBACK_LOG_NODE log) {
@@ -216,7 +216,7 @@ static void rollback_log_create (
     CACHEFILE cf = txn->logger->rollback_cachefile;
     FT CAST_FROM_VOIDP(ft, toku_cachefile_get_userdata(cf));
     rollback_initialize_for_txn(log, txn, previous);
-    toku_allocate_blocknum(ft->blocktable, &log->blocknum, ft);
+    ft->blocktable.allocate_blocknum(&log->blocknum, ft);
     const uint32_t hash = toku_cachetable_hash(ft->cf, log->blocknum);
     *result = log;
     toku_cachetable_put(cf, log->blocknum, hash,

tools/ftverify.cc

@@ -412,10 +412,8 @@ cleanup:
 // Passes our check_block() function to be called as we iterate over
 // the block table.  This will print any interesting failures and
 // update us on our progress.
-static void
-check_block_table(int fd, BLOCK_TABLE bt, struct ft *h)
-{
-    int64_t num_blocks = toku_block_get_blocks_in_use_unlocked(bt);
+static void check_block_table(int fd, block_table *bt, struct ft *h) {
+    int64_t num_blocks = bt->get_blocks_in_use_unlocked();
     printf("Starting verification of checkpoint containing");
     printf(" %" PRId64 " blocks.\n", num_blocks);
     fflush(stdout);
@@ -425,13 +423,11 @@ check_block_table(int fd, BLOCK_TABLE bt, struct ft *h)
 					     .blocks_failed = 0,
 					     .total_blocks = num_blocks,
 					     .h = h };
-    int r = 0;
-    r = toku_blocktable_iterate(bt, 
-				TRANSLATION_CURRENT,
-				check_block,
-				&extra,
-				true,
-				true);
+    int r = bt->iterate(block_table::TRANSLATION_CURRENT,
+                    check_block,
+                    &extra,
+                    true,
+                    true);
     if (r != 0) {
         // We can print more information here if necessary.
     }
@@ -493,11 +489,11 @@ main(int argc, char const * const argv[])
     // walk over the block table and check blocks
     if (h1) {
         printf("Checking dictionary from header 1.\n");
-        check_block_table(dictfd, h1->blocktable, h1);
+        check_block_table(dictfd, &h1->blocktable, h1);
     }
     if (h2) {
         printf("Checking dictionary from header 2.\n");
-        check_block_table(dictfd, h2->blocktable, h2);
+        check_block_table(dictfd, &h2->blocktable, h2);
     }
     if (h1 == NULL && h2 == NULL) {
         printf("Both headers have a corruption and could not be used.\n");

tools/tokuftdump.cc

@@ -237,7 +237,7 @@ static void dump_node(int fd, BLOCKNUM blocknum, FT ft) {
     assert(n!=0);
     printf("ftnode\n");
     DISKOFF disksize, diskoffset;
-    toku_translate_blocknum_to_offset_size(ft->blocktable, blocknum, &diskoffset, &disksize);
+    ft->blocktable.translate_blocknum_to_offset_size(blocknum, &diskoffset, &disksize);
     printf(" diskoffset  =%" PRId64 "\n", diskoffset);
     printf(" disksize    =%" PRId64 "\n", disksize);
     printf(" serialize_size =%u\n", toku_serialize_ftnode_size(n));
@@ -334,13 +334,13 @@ ok:
 }
 
 static void dump_block_translation(FT ft, uint64_t offset) {
-    toku_blocknum_dump_translation(ft->blocktable, make_blocknum(offset));
+    ft->blocktable.blocknum_dump_translation(make_blocknum(offset));
 }
 
 static void dump_fragmentation(int UU(f), FT ft, int tsv) {
     int64_t used_space;
     int64_t total_space;
-    toku_blocktable_internal_fragmentation(ft->blocktable, &total_space, &used_space);
+    ft->blocktable.internal_fragmentation(&total_space, &used_space);
     int64_t fragsizes = total_space - used_space;
 
     if (tsv) {
@@ -386,8 +386,8 @@ static void dump_nodesizes(int fd, FT ft) {
     memset(&info, 0, sizeof(info));
     info.fd = fd;
     info.ft = ft;
-    toku_blocktable_iterate(ft->blocktable, TRANSLATION_CHECKPOINTED,
-                            nodesizes_helper, &info, true, true);
+    ft->blocktable.iterate(block_table::TRANSLATION_CHECKPOINTED,
+                           nodesizes_helper, &info, true, true);
     printf("leafblocks\t%" PRIu64 "\n", info.leafblocks);
     printf("blocksizes\t%" PRIu64 "\n", info.blocksizes);
     printf("leafsizes\t%" PRIu64 "\n", info.leafsizes);
@@ -476,7 +476,7 @@ static void verify_block(unsigned char *cp, uint64_t file_offset, uint64_t size)
 
 static void dump_block(int fd, BLOCKNUM blocknum, FT ft) {
     DISKOFF offset, size;
-    toku_translate_blocknum_to_offset_size(ft->blocktable, blocknum, &offset, &size);
+    ft->blocktable.translate_blocknum_to_offset_size(blocknum, &offset, &size);
     printf("%" PRId64 " at %" PRId64 " size %" PRId64 "\n", blocknum.b, offset, size);
 
     unsigned char *CAST_FROM_VOIDP(vp, toku_malloc(size));
@@ -688,22 +688,22 @@ int main (int argc, const char *const argv[]) {
             dump_fragmentation(fd, ft, do_tsv);
         }
         if (do_translation_table) {
-            toku_dump_translation_table_pretty(stdout, ft->blocktable);
+            ft->blocktable.dump_translation_table_pretty(stdout);
         }
         if (do_garbage) {
             dump_garbage_stats(fd, ft);
         }
         if (!do_header && !do_rootnode && !do_fragmentation && !do_translation_table && !do_garbage) {
             printf("Block translation:");
-            toku_dump_translation_table(stdout, ft->blocktable);
+            ft->blocktable.dump_translation_table(stdout);
 
             dump_header(ft);
             
             struct __dump_node_extra info;
             info.fd = fd;
             info.ft = ft;
-            toku_blocktable_iterate(ft->blocktable, TRANSLATION_CHECKPOINTED,
-                                    dump_node_wrapper, &info, true, true);
+            ft->blocktable.iterate(block_table::TRANSLATION_CHECKPOINTED,
+                                   dump_node_wrapper, &info, true, true);
         }
     }
     toku_cachefile_close(&cf, false, ZERO_LSN);