From 3afacb7702e6d8fa67749a2a41dc776d315e02a9 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Mon, 23 Oct 2023 17:43:22 -0400
Subject: Began moving to an explicit epoch-based system

I started moving over to an explicit Epoch based system, which has
necessitated a ton of changes throughout the code base. This will
ultimately allow for a much cleaner set of abstractions for managing
concurrency.
---
 include/framework/structure/BufferView.h         | 124 +++++++++++++++++++++++
 include/framework/structure/ExtensionStructure.h |  26 +++++
 include/framework/structure/MutableBuffer.h      |   4 +
 3 files changed, 154 insertions(+)
 create mode 100644 include/framework/structure/BufferView.h

(limited to 'include/framework/structure')

diff --git a/include/framework/structure/BufferView.h b/include/framework/structure/BufferView.h
new file mode 100644
index 0000000..1efc1ac
--- /dev/null
+++ b/include/framework/structure/BufferView.h
@@ -0,0 +1,124 @@
+/*
+ * include/framework/structure/BufferView.h
+ *
+ * Copyright (C) 2023 Douglas Rumbaugh <drumbaugh@psu.edu> 
+ *
+ * All rights reserved. Published under the Modified BSD License.
+ *
+ */
+#pragma once
+
+#include <cstdlib>
+#include <atomic>
+#include <condition_variable>
+#include <cassert>
+#include <numeric>
+#include <algorithm>
+#include <type_traits>
+
+#include "psu-util/alignment.h"
+#include "util/bf_config.h"
+#include "psu-ds/BloomFilter.h"
+#include "psu-ds/Alias.h"
+#include "psu-util/timer.h"
+#include "framework/interface/Record.h"
+#include "framework/structure/MutableBuffer.h"
+#include "framework/interface/Query.h"
+
+namespace de {
+
+template <RecordInterface R, QueryInterface Q>
+class BufferView {
+    typedef MutableBuffer<R> Buffer;
+public:
+    BufferView() = default;
+
+    BufferView(std::vector<Buffer*> buffers) 
+        : m_buffers(buffers)
+        , m_cutoff(buffers[buffers->size()-1]->get_record_count())
+    {}
+
+    ~BufferView() = default;
+
+    bool delete_record(const R& rec) {
+        auto res = false;
+        for (auto buf : m_buffers) {
+            res = buf->delete_record(rec);
+            if (res) return true;
+        }
+        return false;
+    }
+
+    bool check_tombstone(const R& rec) {
+        auto res = false;
+        for (auto buf : m_buffers) {
+            res = buf->check_tombstone(rec);
+            if (res) return true;
+        }
+        return false;
+    }
+
+    size_t get_record_count() {
+        size_t reccnt = 0;
+        for (auto buf : m_buffers) {
+            reccnt += buf->get_record_count();
+        }
+        return reccnt;
+    }
+    
+    size_t get_capacity() {
+        return m_buffers[0]->get_capacity();
+    }
+
+    bool is_full() {
+        return m_buffers[m_buffers.size() - 1]->is_full();
+    }
+
+    size_t get_tombstone_count() {
+        size_t tscnt = 0;
+        for (auto buf : m_buffers) {
+            tscnt += buf->get_tombstone_count();
+        }
+        return tscnt;
+    }
+
+    size_t get_memory_usage() {
+        size_t mem = 0;
+        for (auto buf : m_buffers) {
+            mem += buf->get_memory_usage();
+        }
+        return mem;
+    }
+
+    size_t get_aux_memory_usage() {
+        size_t mem = 0;
+        for (auto buf : m_buffers) {
+            mem += buf->get_aux_memory_usage();
+        }
+        return mem;
+    }
+
+    size_t get_tombstone_capacity() {
+        return m_buffers[0]->get_tombstone_capacity();
+    }
+
+    std::vector<void *> get_buffer_states(void *parms) {
+        std::vector<void *> states;
+
+        for (auto buf : m_buffers) {
+            states.push_back(Q::get_buffer_query_state(buf, parms));
+        }
+
+        return states;
+    }
+
+    std::vector<Buffer *> &get_buffers() {
+        return m_buffers;
+    }
+
+private:
+    std::vector<Buffer *> m_buffers;
+    size_t m_cutoff;
+};
+
+}
diff --git a/include/framework/structure/ExtensionStructure.h b/include/framework/structure/ExtensionStructure.h
index 8344518..de965ae 100644
--- a/include/framework/structure/ExtensionStructure.h
+++ b/include/framework/structure/ExtensionStructure.h
@@ -302,12 +302,38 @@ public:
         m_levels[incoming_level] = std::shared_ptr<InternalLevel<R, Shard, Q>>(new InternalLevel<R, Shard, Q>(incoming_level, (L == LayoutPolicy::LEVELING) ? 1 : m_scale_factor));
     }
 
+    bool take_reference() {
+        m_refcnt.fetch_add(1);
+        return true;
+    }
+
+    bool release_reference() {
+        assert(m_refcnt.load() > 0);
+        m_refcnt.fetch_add(-1);
+        return true;
+    }
+
+    size_t get_reference_count() {
+        return m_refcnt.load();
+    }
+
+    std::vector<void *> get_query_states(std::vector<std::pair<ShardID, Shard*>> &shards, void *parms) {
+        std::vector<void*> states;
+
+        for (auto &level : m_levels) {
+            level->get_query_states(shards, states, parms);
+        }
+
+        return states;
+    }
 
 private:
     size_t m_scale_factor;
     double m_max_delete_prop;
     size_t m_buffer_size;
 
+    std::atomic<size_t> m_refcnt;
+
     std::vector<std::shared_ptr<InternalLevel<R, S, Q>>> m_levels;
 
     /*
diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 4e0b5c2..974dc28 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -217,6 +217,10 @@ public:
         return true;
     }
 
+    size_t get_reference_count() {
+        return m_refcnt.load();
+    }
+
     bool active_merge() {
         return m_merging.load();
     }
-- 
cgit v1.2.3