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.

3 files changed, 137 insertions, 9 deletions

diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h
new file mode 100644
index 0000000..a1f865c
--- /dev/null
+++ b/include/framework/scheduling/Epoch.h
@@ -0,0 +1,128 @@
+/*
+ * include/framework/scheduling/Epoch.h
+ *
+ * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu> 
+ *                    Dong Xie <dongx@psu.edu>
+ *
+ * All rights reserved. Published under the Modified BSD License.
+ *
+ */
+#pragma once
+
+#include "framework/structure/MutableBuffer.h"
+#include "framework/structure/ExtensionStructure.h"
+#include "framework/structure/BufferView.h"
+
+namespace de {
+
+
+template <RecordInterface R, ShardInterface S, QueryInterface Q, LayoutPolicy L>
+class Epoch {
+private:
+    typedef MutableBuffer<R> Buffer;
+    typedef ExtensionStructure<R, S, Q, L> Structure;
+    typedef BufferView<R, Q> BufView;
+public:
+    Epoch()
+        : m_buffers()
+        , m_structure(nullptr)
+        , m_active_jobs(0) 
+    {}
+
+    Epoch(Structure *structure, Buffer *buff) 
+        : m_buffers()
+        , m_structure(structure)
+        , m_active_jobs(0) 
+    {
+        m_buffers.push_back(buff);
+    }
+
+    ~Epoch() {
+        assert(m_active_jobs.load() == 0);
+
+        for (auto buf : m_buffers) {
+            buf.release_reference();
+        }
+
+        if (m_structure) {
+            m_structure->release_reference();
+        }
+    }
+
+    void add_buffer(Buffer *buf) {
+        assert(buf);
+
+        buf->take_reference();
+        m_buffers.push_back(buf);
+    }
+
+    void start_job() {
+        m_active_jobs.fetch_add(1);
+    }
+
+    void end_job() {
+        m_active_jobs.fetch_add(-1);
+    }
+
+    size_t get_active_job_num() {
+        return m_active_jobs.load();
+    }
+
+    Structure *get_structure() {
+        return m_structure;
+    }
+
+    std::vector<Buffer *> &get_buffers() {
+        return m_buffers;
+    }
+
+    BufView get_buffer_view() {
+        return BufView(m_buffers);
+    }
+
+    Buffer *get_active_buffer() {
+        if (m_buffers.size() == 0) return nullptr;
+
+        return m_buffers[m_buffers.size() - 1];
+    }
+
+    /*
+     * Return the number of buffers in this epoch at
+     * time of call, and then clear the buffer vector,
+     * releasing all references in the process.
+     */
+    size_t clear_buffers() {
+        size_t buf_cnt = m_buffers.size();
+        for (auto buf : m_buffers) {
+            if (buf) buf->release_reference();
+        }
+
+        m_buffers.clear();
+        return buf_cnt;
+    }
+
+    /*
+     * Returns a new Epoch object that is a copy of this one. The new object will also contain
+     * a copy of the m_structure, rather than a reference to the same one.
+     */
+    Epoch *clone() {
+        auto epoch = new Epoch();
+        epoch->m_buffers = m_buffers;
+        if (m_structure) {
+            epoch->m_structure = m_structure->copy();
+        }
+    }
+
+private:
+    Structure *m_structure;
+    std::vector<Buffer *> m_buffers;
+
+    /*
+     * The number of currently active jobs
+     * (queries/merges) operating on this
+     * epoch. An epoch can only be retired
+     * when this number is 0.
+     */
+    std::atomic<size_t> m_active_jobs;
+};
+}
diff --git a/include/framework/scheduling/SerialScheduler.h b/include/framework/scheduling/FIFOScheduler.h
index da2bb8e..878bb81 100644
--- a/include/framework/scheduling/SerialScheduler.h
+++ b/include/framework/scheduling/FIFOScheduler.h
@@ -29,19 +29,19 @@
 
 namespace de {
 
-class SerialScheduler {
+class FIFOScheduler {
 public:
-    SerialScheduler(size_t memory_budget, size_t thread_cnt)
+    FIFOScheduler(size_t memory_budget, size_t thread_cnt)
       : m_memory_budget((memory_budget) ? memory_budget : UINT64_MAX)
       , m_thrd_cnt((thread_cnt) ? thread_cnt: UINT64_MAX)
       , m_used_memory(0)
       , m_used_thrds(0)
       , m_shutdown(false)
     {
-        m_sched_thrd = std::thread(&SerialScheduler::run, this);
+        m_sched_thrd = std::thread(&FIFOScheduler::run, this);
     }
 
-    ~SerialScheduler() {
+    ~FIFOScheduler() {
         shutdown();
 
         m_cv.notify_all();
diff --git a/include/framework/scheduling/Task.h b/include/framework/scheduling/Task.h
index 518159d..94c4d0a 100644
--- a/include/framework/scheduling/Task.h
+++ b/include/framework/scheduling/Task.h
@@ -8,20 +8,20 @@
 #include <functional>
 
 #include "framework/util/Configuration.h"
+#include "framework/scheduling/Epoch.h"
 
 namespace de {
 
+template <RecordInterface R, ShardInterface S, QueryInterface Q, LayoutPolicy L>
 struct MergeArgs {
-    void *version;
-    void *buffer;
+    Epoch<R, S, Q, L> *epoch;
     std::vector<MergeTask> merges;
     std::promise<bool> result;
 };
 
-template <typename R>
+template <RecordInterface R, ShardInterface S, QueryInterface Q, LayoutPolicy L>
 struct QueryArgs {
-    void *version;
-    void *buffer;
+    Epoch<R, S, Q, L> *epoch;
     std::promise<std::vector<R>> result_set;
     void *query_parms;
 };