From 3afacb7702e6d8fa67749a2a41dc776d315e02a9 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh 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/scheduling/Epoch.h | 128 +++++++++++++++++++++++++ include/framework/scheduling/FIFOScheduler.h | 96 +++++++++++++++++++ include/framework/scheduling/SerialScheduler.h | 96 ------------------- include/framework/scheduling/Task.h | 10 +- 4 files changed, 229 insertions(+), 101 deletions(-) create mode 100644 include/framework/scheduling/Epoch.h create mode 100644 include/framework/scheduling/FIFOScheduler.h delete mode 100644 include/framework/scheduling/SerialScheduler.h (limited to 'include/framework/scheduling') 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 + * Dong Xie + * + * 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 +class Epoch { +private: + typedef MutableBuffer Buffer; + typedef ExtensionStructure Structure; + typedef BufferView 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 &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 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 m_active_jobs; +}; +} diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h new file mode 100644 index 0000000..878bb81 --- /dev/null +++ b/include/framework/scheduling/FIFOScheduler.h @@ -0,0 +1,96 @@ +/* + * include/framework/Scheduler.h + * + * Copyright (C) 2023 Douglas B. Rumbaugh + * Dong Xie + * + * All rights reserved. Published under the Modified BSD License. + * + */ +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "util/types.h" +#include "framework/interface/Shard.h" +#include "framework/interface/Query.h" +#include "framework/interface/Record.h" +#include "framework/structure/MutableBuffer.h" +#include "framework/util/Configuration.h" +#include "framework/structure/ExtensionStructure.h" +#include "framework/scheduling/Task.h" + +#include "psu-ds/LockedPriorityQueue.h" + +namespace de { + +class FIFOScheduler { +public: + 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(&FIFOScheduler::run, this); + } + + ~FIFOScheduler() { + shutdown(); + + m_cv.notify_all(); + m_sched_thrd.join(); + } + + void schedule_job(std::function job, size_t size, void *args) { + size_t ts = m_counter.fetch_add(1); + m_task_queue.push(Task(size, ts, job, args)); + m_cv.notify_all(); + } + + void shutdown() { + m_shutdown = true; + } + +private: + psudb::LockedPriorityQueue m_task_queue; + + size_t m_memory_budget; + size_t m_thrd_cnt; + + bool m_shutdown; + + std::atomic m_counter; + std::mutex m_cv_lock; + std::condition_variable m_cv; + + std::thread m_sched_thrd; + + std::atomic m_used_thrds; + std::atomic m_used_memory; + + void schedule_next() { + auto t = m_task_queue.pop(); + t(); + } + + void run() { + do { + std::unique_lock cv_lock(m_cv_lock); + m_cv.wait(cv_lock); + + while (m_task_queue.size() > 0 && m_used_thrds.load() < m_thrd_cnt) { + schedule_next(); + } + cv_lock.unlock(); + } while(!m_shutdown); + } +}; + +} diff --git a/include/framework/scheduling/SerialScheduler.h b/include/framework/scheduling/SerialScheduler.h deleted file mode 100644 index da2bb8e..0000000 --- a/include/framework/scheduling/SerialScheduler.h +++ /dev/null @@ -1,96 +0,0 @@ -/* - * include/framework/Scheduler.h - * - * Copyright (C) 2023 Douglas B. Rumbaugh - * Dong Xie - * - * All rights reserved. Published under the Modified BSD License. - * - */ -#pragma once - -#include -#include -#include -#include -#include -#include - -#include "util/types.h" -#include "framework/interface/Shard.h" -#include "framework/interface/Query.h" -#include "framework/interface/Record.h" -#include "framework/structure/MutableBuffer.h" -#include "framework/util/Configuration.h" -#include "framework/structure/ExtensionStructure.h" -#include "framework/scheduling/Task.h" - -#include "psu-ds/LockedPriorityQueue.h" - -namespace de { - -class SerialScheduler { -public: - SerialScheduler(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); - } - - ~SerialScheduler() { - shutdown(); - - m_cv.notify_all(); - m_sched_thrd.join(); - } - - void schedule_job(std::function job, size_t size, void *args) { - size_t ts = m_counter.fetch_add(1); - m_task_queue.push(Task(size, ts, job, args)); - m_cv.notify_all(); - } - - void shutdown() { - m_shutdown = true; - } - -private: - psudb::LockedPriorityQueue m_task_queue; - - size_t m_memory_budget; - size_t m_thrd_cnt; - - bool m_shutdown; - - std::atomic m_counter; - std::mutex m_cv_lock; - std::condition_variable m_cv; - - std::thread m_sched_thrd; - - std::atomic m_used_thrds; - std::atomic m_used_memory; - - void schedule_next() { - auto t = m_task_queue.pop(); - t(); - } - - void run() { - do { - std::unique_lock cv_lock(m_cv_lock); - m_cv.wait(cv_lock); - - while (m_task_queue.size() > 0 && m_used_thrds.load() < m_thrd_cnt) { - schedule_next(); - } - cv_lock.unlock(); - } while(!m_shutdown); - } -}; - -} 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 #include "framework/util/Configuration.h" +#include "framework/scheduling/Epoch.h" namespace de { +template struct MergeArgs { - void *version; - void *buffer; + Epoch *epoch; std::vector merges; std::promise result; }; -template +template struct QueryArgs { - void *version; - void *buffer; + Epoch *epoch; std::promise> result_set; void *query_parms; }; -- cgit v1.2.3