Ported ISAMTree over to new buffer setup

I may still play with the shard from shards constructor, and queries
need some work yet too.

1 files changed, 65 insertions, 60 deletions

diff --git a/include/shard/ISAMTree.h b/include/shard/ISAMTree.h
index e11c899..6b2f6b5 100644
--- a/include/shard/ISAMTree.h
+++ b/include/shard/ISAMTree.h
@@ -13,8 +13,6 @@
 
 #include <vector>
 #include <cassert>
-#include <queue>
-#include <memory>
 
 #include "framework/ShardRequirements.h"
 
@@ -27,52 +25,54 @@ using psudb::CACHELINE_SIZE;
 using psudb::BloomFilter;
 using psudb::PriorityQueue;
 using psudb::queue_record;
-using psudb::Alias;
 
 namespace de {
 
-thread_local size_t mrun_cancelations = 0;
-
-template <RecordInterface R>
+template <KVPInterface R>
 class ISAMTree {
 private:
 
 typedef decltype(R::key) K;
 typedef decltype(R::value) V;
 
-constexpr static size_t inmem_isam_node_size = 256;
-constexpr static size_t inmem_isam_fanout = inmem_isam_node_size / (sizeof(K) + sizeof(char*));
+constexpr static size_t NODE_SZ = 256;
+constexpr static size_t INTERNAL_FANOUT = NODE_SZ / (sizeof(K) + sizeof(byte*));
 
 struct InternalNode {
-    K keys[inmem_isam_fanout];
-    char* child[inmem_isam_fanout];
+    K keys[INTERNAL_FANOUT];
+    byte* child[INTERNAL_FANOUT];
 };
 
-constexpr static size_t inmem_isam_leaf_fanout = inmem_isam_node_size / sizeof(R);
-constexpr static size_t inmem_isam_node_keyskip = sizeof(K) * inmem_isam_fanout;
-
-static_assert(sizeof(InternalNode) == inmem_isam_node_size, "node size does not match");
+static_assert(sizeof(InternalNode) == NODE_SZ, "node size does not match");
 
-public:
-    ISAMTree(MutableBuffer<R>* buffer)
-    :m_reccnt(0), m_tombstone_cnt(0), m_isam_nodes(nullptr), m_deleted_cnt(0) {
+constexpr static size_t LEAF_FANOUT = NODE_SZ / sizeof(R);
 
-        m_bf = new BloomFilter<R>(BF_FPR, buffer->get_tombstone_count(), BF_HASH_FUNCS);
-
-        m_alloc_size = (buffer->get_record_count() * sizeof(Wrapped<R>)) + (CACHELINE_SIZE - (buffer->get_record_count() * sizeof(Wrapped<R>)) % CACHELINE_SIZE);
-        assert(m_alloc_size % CACHELINE_SIZE == 0);
-        m_data = (Wrapped<R>*)std::aligned_alloc(CACHELINE_SIZE, m_alloc_size);
 
+public:
+    ISAMTree(BufferView<R> buffer)
+        : m_bf(new BloomFilter<R>(BF_FPR, buffer.get_tombstone_count(), BF_HASH_FUNCS))
+        , m_isam_nodes(nullptr)
+        , m_root(nullptr)
+        , m_reccnt(0)
+        , m_tombstone_cnt(0)
+        , m_internal_node_cnt(0)
+        , m_deleted_cnt(0)
+        , m_alloc_size(0)
+        , m_data(nullptr)
+    {
         TIMER_INIT();
 
-        size_t offset = 0;
-        m_reccnt = 0;
-        auto base = buffer->get_data();
-        auto stop = base + buffer->get_record_count();
+        m_alloc_size = psudb::sf_aligned_alloc(CACHELINE_SIZE, buffer.get_record_count() * sizeof(Wrapped<R>), (byte**) &m_data);
 
         TIMER_START();
+        auto temp_buffer = (Wrapped<R> *) psudb::sf_aligned_alloc(CACHELINE_SIZE, buffer.get_record_count() * sizeof(Wrapped<R>));
+        buffer.copy_to_buffer((byte *) temp_buffer);
+
+        auto base = temp_buffer;
+        auto stop = base + buffer.get_record_count();
         std::sort(base, stop, std::less<Wrapped<R>>());
         TIMER_STOP();
+
         auto sort_time = TIMER_RESULT();
 
         TIMER_START();
@@ -80,7 +80,6 @@ public:
             if (!base->is_tombstone() && (base + 1 < stop)
                 && base->rec == (base + 1)->rec  && (base + 1)->is_tombstone()) {
                 base += 2;
-                mrun_cancelations++;
                 continue;
             } else if (base->is_deleted()) {
                 base += 1;
@@ -109,10 +108,21 @@ public:
         }
         TIMER_STOP();
         auto level_time = TIMER_RESULT();
+
+        free(temp_buffer);
     }
 
     ISAMTree(ISAMTree** runs, size_t len)
-    : m_reccnt(0), m_tombstone_cnt(0), m_deleted_cnt(0), m_isam_nodes(nullptr) {
+        : m_bf(nullptr) 
+        , m_isam_nodes(nullptr)
+        , m_root(nullptr)
+        , m_reccnt(0)
+        , m_tombstone_cnt(0)
+        , m_internal_node_cnt(0)
+        , m_deleted_cnt(0)
+        , m_alloc_size(0)
+        , m_data(nullptr)
+    {
         std::vector<Cursor<Wrapped<R>>> cursors;
         cursors.reserve(len);
 
@@ -139,8 +149,6 @@ public:
         assert(m_alloc_size % CACHELINE_SIZE == 0);
         m_data = (Wrapped<R>*)std::aligned_alloc(CACHELINE_SIZE, m_alloc_size);
 
-        size_t offset = 0;
-        
         while (pq.size()) {
             auto now = pq.peek();
             auto next = pq.size() > 1 ? pq.peek(1) : queue_record<Wrapped<R>>{nullptr, 0};
@@ -173,9 +181,9 @@ public:
     }
 
     ~ISAMTree() {
-        if (m_data) free(m_data);
-        if (m_isam_nodes) free(m_isam_nodes);
-        if (m_bf) delete m_bf;
+        free(m_data);
+        free(m_isam_nodes);
+        delete m_bf;
     }
 
     Wrapped<R> *point_lookup(const R &rec, bool filter=false) {
@@ -214,25 +222,25 @@ public:
     }
 
     size_t get_memory_usage() {
-        return m_internal_node_cnt * inmem_isam_node_size + m_alloc_size;
+        return m_alloc_size;
     }
 
     size_t get_aux_memory_usage() {
-        return 0;
+        return m_bf->memory_usage();
     }
 
     size_t get_lower_bound(const K& key) const {
         const InternalNode* now = m_root;
-        while (!is_leaf(reinterpret_cast<const char*>(now))) {
+        while (!is_leaf(reinterpret_cast<const byte*>(now))) {
             const InternalNode* next = nullptr;
-            for (size_t i = 0; i < inmem_isam_fanout - 1; ++i) {
+            for (size_t i = 0; i < INTERNAL_FANOUT - 1; ++i) {
                 if (now->child[i + 1] == nullptr || key <= now->keys[i]) {
                     next = reinterpret_cast<InternalNode*>(now->child[i]);
                     break;
                 }
             }
 
-            now = next ? next : reinterpret_cast<const InternalNode*>(now->child[inmem_isam_fanout - 1]);
+            now = next ? next : reinterpret_cast<const InternalNode*>(now->child[INTERNAL_FANOUT - 1]);
         }
 
         const Wrapped<R>* pos = reinterpret_cast<const Wrapped<R>*>(now);
@@ -243,16 +251,16 @@ public:
 
     size_t get_upper_bound(const K& key) const {
         const InternalNode* now = m_root;
-        while (!is_leaf(reinterpret_cast<const char*>(now))) {
+        while (!is_leaf(reinterpret_cast<const byte*>(now))) {
             const InternalNode* next = nullptr;
-            for (size_t i = 0; i < inmem_isam_fanout - 1; ++i) {
+            for (size_t i = 0; i < INTERNAL_FANOUT - 1; ++i) {
                 if (now->child[i + 1] == nullptr || key < now->keys[i]) {
                     next = reinterpret_cast<InternalNode*>(now->child[i]);
                     break;
                 }
             }
 
-            now = next ? next : reinterpret_cast<const InternalNode*>(now->child[inmem_isam_fanout - 1]);
+            now = next ? next : reinterpret_cast<const InternalNode*>(now->child[INTERNAL_FANOUT - 1]);
         }
 
         const Wrapped<R>* pos = reinterpret_cast<const Wrapped<R>*>(now);
@@ -264,20 +272,17 @@ public:
 
 private:
     void build_internal_levels() {
-        size_t n_leaf_nodes = m_reccnt / inmem_isam_leaf_fanout + (m_reccnt % inmem_isam_leaf_fanout != 0);
+        size_t n_leaf_nodes = m_reccnt / LEAF_FANOUT + (m_reccnt % LEAF_FANOUT != 0);
+
         size_t level_node_cnt = n_leaf_nodes;
         size_t node_cnt = 0;
         do {
-            level_node_cnt = level_node_cnt / inmem_isam_fanout + (level_node_cnt % inmem_isam_fanout != 0);
+            level_node_cnt = level_node_cnt / INTERNAL_FANOUT + (level_node_cnt % INTERNAL_FANOUT != 0);
             node_cnt += level_node_cnt;
         } while (level_node_cnt > 1);
 
-        m_alloc_size = (node_cnt * inmem_isam_node_size) + (CACHELINE_SIZE - (node_cnt * inmem_isam_node_size) % CACHELINE_SIZE);
-        assert(m_alloc_size % CACHELINE_SIZE == 0);
-
-        m_isam_nodes = (InternalNode*)std::aligned_alloc(CACHELINE_SIZE, m_alloc_size);
+        m_alloc_size += psudb::sf_aligned_calloc(CACHELINE_SIZE, node_cnt, NODE_SZ, (byte**) &m_isam_nodes);
         m_internal_node_cnt = node_cnt;
-        memset(m_isam_nodes, 0, node_cnt * inmem_isam_node_size);
 
         InternalNode* current_node = m_isam_nodes;
 
@@ -285,16 +290,16 @@ private:
         const Wrapped<R>* leaf_stop = m_data + m_reccnt;
         while (leaf_base < leaf_stop) {
             size_t fanout = 0;
-            for (size_t i = 0; i < inmem_isam_fanout; ++i) {
-                auto rec_ptr = leaf_base + inmem_isam_leaf_fanout * i;
+            for (size_t i = 0; i < INTERNAL_FANOUT; ++i) {
+                auto rec_ptr = leaf_base + LEAF_FANOUT * i;
                 if (rec_ptr >= leaf_stop) break;
-                const Wrapped<R>* sep_key = std::min(rec_ptr + inmem_isam_leaf_fanout - 1, leaf_stop - 1);
+                const Wrapped<R>* sep_key = std::min(rec_ptr + LEAF_FANOUT - 1, leaf_stop - 1);
                 current_node->keys[i] = sep_key->rec.key;
-                current_node->child[i] = (char*)rec_ptr;
+                current_node->child[i] = (byte*)rec_ptr;
                 ++fanout;
             }
             current_node++;
-            leaf_base += fanout * inmem_isam_leaf_fanout;
+            leaf_base += fanout * LEAF_FANOUT;
         }
 
         auto level_start = m_isam_nodes;
@@ -304,12 +309,12 @@ private:
             auto now = level_start;
             while (now < level_stop) {
                 size_t child_cnt = 0;
-                for (size_t i = 0; i < inmem_isam_fanout; ++i) {
+                for (size_t i = 0; i < INTERNAL_FANOUT; ++i) {
                     auto node_ptr = now + i;
                     ++child_cnt;
                     if (node_ptr >= level_stop) break;
-                    current_node->keys[i] = node_ptr->keys[inmem_isam_fanout - 1];
-                    current_node->child[i] = (char*)node_ptr;
+                    current_node->keys[i] = node_ptr->keys[INTERNAL_FANOUT - 1];
+                    current_node->child[i] = (byte*)node_ptr;
                 }
                 now += child_cnt;
                 current_node++;
@@ -323,12 +328,10 @@ private:
         m_root = level_start;
     }
 
-    bool is_leaf(const char* ptr) const {
-        return ptr >= (const char*)m_data && ptr < (const char*)(m_data + m_reccnt);
+    bool is_leaf(const byte* ptr) const {
+        return ptr >= (const byte*)m_data && ptr < (const byte*)(m_data + m_reccnt);
     }
 
-    // Members: sorted data, internal ISAM levels, reccnt;
-    Wrapped<R>* m_data;
     psudb::BloomFilter<R> *m_bf;
     InternalNode* m_isam_nodes;
     InternalNode* m_root;
@@ -337,5 +340,7 @@ private:
     size_t m_internal_node_cnt;
     size_t m_deleted_cnt;
     size_t m_alloc_size;
+
+    Wrapped<R>* m_data;
 };
 }