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 +++++++++++++++++++++++++++++++++++ 1 file changed, 128 insertions(+) create mode 100644 include/framework/scheduling/Epoch.h (limited to 'include/framework/scheduling/Epoch.h') 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; +}; +} -- cgit v1.2.3 From 39ae3e0441d8297a09197aba98bd494b5ada12c1 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Mon, 30 Oct 2023 14:17:59 -0400 Subject: Concurrency updates + fixes for compile errors --- include/framework/scheduling/Epoch.h | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index a1f865c..fe63c86 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -41,7 +41,7 @@ public: assert(m_active_jobs.load() == 0); for (auto buf : m_buffers) { - buf.release_reference(); + buf->release_reference(); } if (m_structure) { @@ -111,6 +111,8 @@ public: if (m_structure) { epoch->m_structure = m_structure->copy(); } + + return epoch; } private: -- cgit v1.2.3 From 32aeedbaf6584eb71126cbe92cb42e93b65d69d3 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Mon, 30 Oct 2023 14:47:35 -0400 Subject: Epoch/DynamicExtension: added cv to epoch retirement check Instead of busy waiting on the active job count, a condition variable is now used to wait for all active jobs to finish before freeing an epoch's resources. --- include/framework/scheduling/Epoch.h | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index fe63c86..87463bd 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -62,6 +62,11 @@ public: void end_job() { m_active_jobs.fetch_add(-1); + + if (m_active_jobs.load() == 0) { + std::unique_lock lk(m_cv_lock); + m_active_cv.notify_all(); + } } size_t get_active_job_num() { @@ -115,10 +120,35 @@ public: return epoch; } + /* + * + */ + bool retirable() { + /* if epoch is currently active, then it cannot be retired */ + if (m_active) { + return false; + } + + /* + * if the epoch has active jobs but is not itself active, + * wait for them to finish and return true. If there are + * not active jobs, return true immediately + */ + while (m_active_jobs > 0) { + std::unique_lock lk(m_cv_lock); + m_active_cv.wait(lk); + } + + return true; + } + private: Structure *m_structure; std::vector m_buffers; + std::condition_variable m_active_cv; + std::mutex m_cv_lock; + /* * The number of currently active jobs * (queries/merges) operating on this @@ -126,5 +156,6 @@ private: * when this number is 0. */ std::atomic m_active_jobs; + bool m_active; }; } -- cgit v1.2.3 From d2279e1b96d352a0af1d425dcaaf93e8a26a8d52 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Mon, 30 Oct 2023 17:15:05 -0400 Subject: General Comment + Consistency updates --- include/framework/scheduling/Epoch.h | 1 - 1 file changed, 1 deletion(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index 87463bd..03cbb62 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -2,7 +2,6 @@ * include/framework/scheduling/Epoch.h * * Copyright (C) 2023 Douglas B. Rumbaugh - * Dong Xie * * All rights reserved. Published under the Modified BSD License. * -- cgit v1.2.3 From 62792753bb4df2515e5e2d8cc48bca568c5379fd Mon Sep 17 00:00:00 2001 From: "Douglas B. Rumbaugh" Date: Tue, 31 Oct 2023 11:48:56 -0400 Subject: Epoch: Creating an epoch now takes references on buffers + versions When an epoch is created using the constructor Epoch(Structure, Buffer), it will call take_reference() on both. This was necessary to ensure that the destructor doesn't fail, as it releases references and fails if the refcnt is 0. It releases the user of the object from the burden of manually taking references in this situation. --- include/framework/scheduling/Epoch.h | 2 ++ 1 file changed, 2 insertions(+) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index 03cbb62..6bbf927 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -33,6 +33,8 @@ public: , m_structure(structure) , m_active_jobs(0) { + structure->take_reference(); + buff->take_reference(); m_buffers.push_back(buff); } -- cgit v1.2.3 From 68ae6279476e7d37837ac06474fb558e50ce6706 Mon Sep 17 00:00:00 2001 From: "Douglas B. Rumbaugh" Date: Tue, 31 Oct 2023 12:41:55 -0400 Subject: Fixes for various bugs under SerialScheduler --- include/framework/scheduling/Epoch.h | 31 ++++++++++++++++++++++++++----- 1 file changed, 26 insertions(+), 5 deletions(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index 6bbf927..f4aefe9 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -22,16 +22,20 @@ private: typedef ExtensionStructure Structure; typedef BufferView BufView; public: - Epoch() + Epoch(size_t number=0) : m_buffers() , m_structure(nullptr) , m_active_jobs(0) + , m_active(true) + , m_epoch_number(number) {} - Epoch(Structure *structure, Buffer *buff) + Epoch(size_t number, Structure *structure, Buffer *buff) : m_buffers() , m_structure(structure) , m_active_jobs(0) + , m_active(true) + , m_epoch_number(number) { structure->take_reference(); buff->take_reference(); @@ -62,6 +66,7 @@ public: } void end_job() { + assert(m_active_jobs.load() > 0); m_active_jobs.fetch_add(-1); if (m_active_jobs.load() == 0) { @@ -74,6 +79,10 @@ public: return m_active_jobs.load(); } + size_t get_epoch_number() { + return m_epoch_number; + } + Structure *get_structure() { return m_structure; } @@ -109,18 +118,29 @@ public: /* * 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. + * a copy of the m_structure, rather than a reference to the same one. The epoch number of + * the new epoch will be set to the provided argument. */ - Epoch *clone() { - auto epoch = new Epoch(); + Epoch *clone(size_t number) { + auto epoch = new Epoch(number); epoch->m_buffers = m_buffers; if (m_structure) { epoch->m_structure = m_structure->copy(); + /* the copy routine returns a structure with 0 references */ + epoch->m_structure->take_reference(); + } + + for (auto b : m_buffers) { + b->take_reference(); } return epoch; } + void set_inactive() { + m_active = false; + } + /* * */ @@ -158,5 +178,6 @@ private: */ std::atomic m_active_jobs; bool m_active; + size_t m_epoch_number; }; } -- cgit v1.2.3 From ca1605a9924e27ccbacb33d04ccdb4326e7abe74 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Mon, 6 Nov 2023 12:37:06 -0500 Subject: Epoch: Adjusted add empty buffer behavior Add empty buffer now supports a CAS-like operation, where it will only add a buffer if the currently active one is still the same as when the decision to add a buffer was made. This is to support adding new buffers on insert outside of the merge-lock, so that multiple concurrent threads cannot add multiple new empty buffers. --- include/framework/scheduling/Epoch.h | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index f4aefe9..58fe6cd 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -54,11 +54,25 @@ public: } } - void add_buffer(Buffer *buf) { + Buffer *add_buffer(Buffer *buf, Buffer *cur_buf=nullptr) { assert(buf); + /* + * if a current buffer is specified, only add the + * new buffer if the active buffer is the current, + * otherwise just return the active buffer (poor man's + * CAS). + */ + if (cur_buf) { + auto active_buf = get_active_buffer(); + if (active_buf != cur_buf) { + return active_buf; + } + } + buf->take_reference(); m_buffers.push_back(buf); + return buf; } void start_job() { -- cgit v1.2.3 From 254f8aa85ea8962e5c11d8b475a171883c22f168 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Mon, 6 Nov 2023 12:39:35 -0500 Subject: DynamicExtension: internal_append fixes Fixed a few bugs with concurrent operation of internal_append, as well as enabled the spawning of multiple empty buffers while merges are currently active. --- include/framework/scheduling/Epoch.h | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index 58fe6cd..0ebbde9 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -25,6 +25,7 @@ public: Epoch(size_t number=0) : m_buffers() , m_structure(nullptr) + , m_active_merge(false) , m_active_jobs(0) , m_active(true) , m_epoch_number(number) @@ -34,6 +35,7 @@ public: : m_buffers() , m_structure(structure) , m_active_jobs(0) + , m_active_merge(false) , m_active(true) , m_epoch_number(number) { @@ -151,6 +153,31 @@ public: return epoch; } + /* + * Check if a merge can be started from this Epoch. + * At present, without concurrent merging, this simply + * checks if there is currently a scheduled merge based + * on this Epoch. If there is, returns false. If there + * isn't, return true and set a flag indicating that + * there is an active merge. + */ + bool prepare_merge() { + auto old = m_active_merge.load(); + if (old) { + return false; + } + + // FIXME: this needs cleaned up + while (!m_active_merge.compare_exchange_strong(old, true)) { + old = m_active_merge.load(); + if (old) { + return false; + } + } + + return true; + } + void set_inactive() { m_active = false; } @@ -184,6 +211,8 @@ private: std::condition_variable m_active_cv; std::mutex m_cv_lock; + std::atomic m_active_merge; + /* * The number of currently active jobs * (queries/merges) operating on this -- cgit v1.2.3 From 357cab549c2ed33970562b84ff6f83923742343d Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Tue, 7 Nov 2023 15:34:24 -0500 Subject: Comment and License updates --- include/framework/scheduling/Epoch.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index 0ebbde9..9193b06 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -3,7 +3,7 @@ * * Copyright (C) 2023 Douglas B. Rumbaugh * - * All rights reserved. Published under the Modified BSD License. + * Distributed under the Modified BSD License. * */ #pragma once -- cgit v1.2.3 From 39d22316be1708073e4fe1f708814cc801ecdc69 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Thu, 9 Nov 2023 11:08:34 -0500 Subject: Fixed various concurrency bugs 1. The system should now cleanly shutdown when the DynamicExtension object is destroyed. Before now, this would lead to use-after-frees and/or deadlocks. 2. Improved synchronization on mutable buffer structure management to fix the issue of the framework losing track of buffers during Epoch changeovers. --- include/framework/scheduling/Epoch.h | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index 9193b06..fc08d57 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -47,9 +47,14 @@ public: ~Epoch() { assert(m_active_jobs.load() == 0); - for (auto buf : m_buffers) { - buf->release_reference(); - } + /* FIXME: this is needed to keep the destructor from + * sometimes locking up here. But there *shouldn't* be + * any threads waiting on this signal at object destruction, + * so something else is going on here that needs looked into + */ + //m_active_cv.notify_all(); + + clear_buffers(); if (m_structure) { m_structure->release_reference(); @@ -59,6 +64,7 @@ public: Buffer *add_buffer(Buffer *buf, Buffer *cur_buf=nullptr) { assert(buf); + std::unique_lock m_buffer_lock; /* * if a current buffer is specified, only add the * new buffer if the active buffer is the current, @@ -108,6 +114,7 @@ public: } BufView get_buffer_view() { + std::unique_lock m_buffer_lock; return BufView(m_buffers); } @@ -123,6 +130,7 @@ public: * releasing all references in the process. */ size_t clear_buffers() { + std::unique_lock m_buffer_lock; size_t buf_cnt = m_buffers.size(); for (auto buf : m_buffers) { if (buf) buf->release_reference(); @@ -138,6 +146,7 @@ public: * the new epoch will be set to the provided argument. */ Epoch *clone(size_t number) { + std::unique_lock m_buffer_lock; auto epoch = new Epoch(number); epoch->m_buffers = m_buffers; if (m_structure) { @@ -196,8 +205,8 @@ public: * wait for them to finish and return true. If there are * not active jobs, return true immediately */ - while (m_active_jobs > 0) { - std::unique_lock lk(m_cv_lock); + std::unique_lock lk(m_cv_lock); + while (m_active_jobs.load() > 0) { m_active_cv.wait(lk); } @@ -211,6 +220,8 @@ private: std::condition_variable m_active_cv; std::mutex m_cv_lock; + std::mutex m_buffer_lock; + std::atomic m_active_merge; /* -- cgit v1.2.3 From 3c127eda69295cb306739bdd3c5ddccff6026a8d Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Wed, 13 Dec 2023 12:39:54 -0500 Subject: Refactoring: corrected a number of names and added more comments --- include/framework/scheduling/Epoch.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index fc08d57..4e1b8a2 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -170,7 +170,7 @@ public: * isn't, return true and set a flag indicating that * there is an active merge. */ - bool prepare_merge() { + bool prepare_reconstruction() { auto old = m_active_merge.load(); if (old) { return false; -- cgit v1.2.3 From aac0bb661af8fae38d3ce08d6078cb4d9dfcb575 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Fri, 12 Jan 2024 14:10:11 -0500 Subject: Initial integration of new buffering scheme into framework It isn't working right now (lotsa test failures), but we're to the debugging phase now. --- include/framework/scheduling/Epoch.h | 80 ++++++++---------------------------- 1 file changed, 16 insertions(+), 64 deletions(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index 4e1b8a2..ca85fe2 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -8,6 +8,9 @@ */ #pragma once +#include +#include + #include "framework/structure/MutableBuffer.h" #include "framework/structure/ExtensionStructure.h" #include "framework/structure/BufferView.h" @@ -20,10 +23,10 @@ class Epoch { private: typedef MutableBuffer Buffer; typedef ExtensionStructure Structure; - typedef BufferView BufView; + typedef BufferView BufView; public: Epoch(size_t number=0) - : m_buffers() + : m_buffer(nullptr) , m_structure(nullptr) , m_active_merge(false) , m_active_jobs(0) @@ -31,8 +34,8 @@ public: , m_epoch_number(number) {} - Epoch(size_t number, Structure *structure, Buffer *buff) - : m_buffers() + Epoch(size_t number, Structure *structure, Buffer *buff) + : m_buffer(buff) , m_structure(structure) , m_active_jobs(0) , m_active_merge(false) @@ -40,8 +43,6 @@ public: , m_epoch_number(number) { structure->take_reference(); - buff->take_reference(); - m_buffers.push_back(buff); } ~Epoch() { @@ -54,35 +55,11 @@ public: */ //m_active_cv.notify_all(); - clear_buffers(); - if (m_structure) { m_structure->release_reference(); } } - Buffer *add_buffer(Buffer *buf, Buffer *cur_buf=nullptr) { - assert(buf); - - std::unique_lock m_buffer_lock; - /* - * if a current buffer is specified, only add the - * new buffer if the active buffer is the current, - * otherwise just return the active buffer (poor man's - * CAS). - */ - if (cur_buf) { - auto active_buf = get_active_buffer(); - if (active_buf != cur_buf) { - return active_buf; - } - } - - buf->take_reference(); - m_buffers.push_back(buf); - return buf; - } - void start_job() { m_active_jobs.fetch_add(1); } @@ -109,36 +86,10 @@ public: return m_structure; } - std::vector &get_buffers() { - return m_buffers; - } - - BufView get_buffer_view() { - std::unique_lock m_buffer_lock; - return BufView(m_buffers); - } - - Buffer *get_active_buffer() { - if (m_buffers.size() == 0) return nullptr; - - return m_buffers[m_buffers.size() - 1]; + BufView get_buffer() { + return m_buffer->get_buffer_view(); } - /* - * 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() { - std::unique_lock m_buffer_lock; - 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 @@ -148,17 +99,14 @@ public: Epoch *clone(size_t number) { std::unique_lock m_buffer_lock; auto epoch = new Epoch(number); - epoch->m_buffers = m_buffers; + epoch->m_buffer = m_buffer; + if (m_structure) { epoch->m_structure = m_structure->copy(); /* the copy routine returns a structure with 0 references */ epoch->m_structure->take_reference(); } - for (auto b : m_buffers) { - b->take_reference(); - } - return epoch; } @@ -213,9 +161,13 @@ public: return true; } + bool advance_buffer_head(size_t head) { + return m_buffer->advance_head(head); + } + private: Structure *m_structure; - std::vector m_buffers; + Buffer *m_buffer; std::condition_variable m_active_cv; std::mutex m_cv_lock; -- cgit v1.2.3 From cf178ae74a76b780b655a447531d2114f9f81d98 Mon Sep 17 00:00:00 2001 From: "Douglas B. Rumbaugh" Date: Mon, 15 Jan 2024 14:01:36 -0500 Subject: Various single-threaded bug fixes --- include/framework/scheduling/Epoch.h | 14 ++++++++++++++ 1 file changed, 14 insertions(+) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index ca85fe2..b005ff6 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -60,6 +60,16 @@ public: } } + + /* + * Epochs are *not* copyable or movable. Only one can exist, and all users of + * it work with pointers + */ + Epoch(const Epoch&) = delete; + Epoch(Epoch&&) = delete; + Epoch &operator=(const Epoch&) = delete; + Epoch &operator=(Epoch&&) = delete; + void start_job() { m_active_jobs.fetch_add(1); } @@ -90,6 +100,10 @@ public: return m_buffer->get_buffer_view(); } + BufView get_flush_buffer() { + return m_buffer->get_flush_buffer_view(); + } + /* * Returns a new Epoch object that is a copy of this one. The new object will also contain -- cgit v1.2.3 From 138c793b0a58577713d98c98bb140cf1d9c79bee Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Wed, 17 Jan 2024 18:22:00 -0500 Subject: Multiple concurrency bug fixes A poorly organized commit with fixes for a variety of bugs that were causing missing records. The core problems all appear to be fixed, though there is an outstanding problem with tombstones not being completely canceled. A very small number are appearing in the wrong order during the static structure test. --- include/framework/scheduling/Epoch.h | 51 +++++++++++++++++------------------- 1 file changed, 24 insertions(+), 27 deletions(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index b005ff6..45ee17d 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -32,15 +32,17 @@ public: , m_active_jobs(0) , m_active(true) , m_epoch_number(number) + , m_buffer_head(0) {} - Epoch(size_t number, Structure *structure, Buffer *buff) + Epoch(size_t number, Structure *structure, Buffer *buff, size_t head) : m_buffer(buff) , m_structure(structure) , m_active_jobs(0) , m_active_merge(false) , m_active(true) , m_epoch_number(number) + , m_buffer_head(head) { structure->take_reference(); } @@ -48,22 +50,21 @@ public: ~Epoch() { assert(m_active_jobs.load() == 0); - /* FIXME: this is needed to keep the destructor from - * sometimes locking up here. But there *shouldn't* be - * any threads waiting on this signal at object destruction, - * so something else is going on here that needs looked into + /* FIXME: this is needed to keep the destructor from sometimes locking + * up here. But there *shouldn't* be any threads waiting on this signal + * at object destruction, so something else is going on here that needs + * looked into */ - //m_active_cv.notify_all(); + // m_active_cv.notify_all(); if (m_structure) { m_structure->release_reference(); } } - - /* - * Epochs are *not* copyable or movable. Only one can exist, and all users of - * it work with pointers + /* + * Epochs are *not* copyable or movable. Only one can exist, and all users + * of it work with pointers */ Epoch(const Epoch&) = delete; Epoch(Epoch&&) = delete; @@ -97,23 +98,20 @@ public: } BufView get_buffer() { - return m_buffer->get_buffer_view(); - } - - BufView get_flush_buffer() { - return m_buffer->get_flush_buffer_view(); + return m_buffer->get_buffer_view(m_buffer_head); } - /* - * 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. The epoch number of - * the new epoch will be set to the provided argument. + * 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. The epoch number of the new epoch will be set to the + * provided argument. */ Epoch *clone(size_t number) { std::unique_lock m_buffer_lock; auto epoch = new Epoch(number); epoch->m_buffer = m_buffer; + epoch->m_buffer_head = m_buffer_head; if (m_structure) { epoch->m_structure = m_structure->copy(); @@ -125,12 +123,10 @@ public: } /* - * Check if a merge can be started from this Epoch. - * At present, without concurrent merging, this simply - * checks if there is currently a scheduled merge based - * on this Epoch. If there is, returns false. If there - * isn't, return true and set a flag indicating that - * there is an active merge. + * Check if a merge can be started from this Epoch. At present, without + * concurrent merging, this simply checks if there is currently a scheduled + * merge based on this Epoch. If there is, returns false. If there isn't, + * return true and set a flag indicating that there is an active merge. */ bool prepare_reconstruction() { auto old = m_active_merge.load(); @@ -176,7 +172,8 @@ public: } bool advance_buffer_head(size_t head) { - return m_buffer->advance_head(head); + m_buffer_head = head; + return m_buffer->advance_head(m_buffer_head); } private: @@ -187,7 +184,6 @@ private: std::mutex m_cv_lock; std::mutex m_buffer_lock; - std::atomic m_active_merge; /* @@ -199,5 +195,6 @@ private: std::atomic m_active_jobs; bool m_active; size_t m_epoch_number; + size_t m_buffer_head; }; } -- cgit v1.2.3 From f3b7428cfa7f9364c5a8bc85107db3a7cccd53bc Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Wed, 31 Jan 2024 18:41:17 -0500 Subject: Adjusted epoch transition methodology --- include/framework/scheduling/Epoch.h | 55 ------------------------------------ 1 file changed, 55 deletions(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index 45ee17d..7b533b6 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -29,8 +29,6 @@ public: : m_buffer(nullptr) , m_structure(nullptr) , m_active_merge(false) - , m_active_jobs(0) - , m_active(true) , m_epoch_number(number) , m_buffer_head(0) {} @@ -38,9 +36,7 @@ public: Epoch(size_t number, Structure *structure, Buffer *buff, size_t head) : m_buffer(buff) , m_structure(structure) - , m_active_jobs(0) , m_active_merge(false) - , m_active(true) , m_epoch_number(number) , m_buffer_head(head) { @@ -48,8 +44,6 @@ public: } ~Epoch() { - assert(m_active_jobs.load() == 0); - /* FIXME: this is needed to keep the destructor from sometimes locking * up here. But there *shouldn't* be any threads waiting on this signal * at object destruction, so something else is going on here that needs @@ -71,24 +65,6 @@ public: Epoch &operator=(const Epoch&) = delete; Epoch &operator=(Epoch&&) = delete; - void start_job() { - m_active_jobs.fetch_add(1); - } - - void end_job() { - assert(m_active_jobs.load() > 0); - m_active_jobs.fetch_add(-1); - - if (m_active_jobs.load() == 0) { - std::unique_lock lk(m_cv_lock); - m_active_cv.notify_all(); - } - } - - size_t get_active_job_num() { - return m_active_jobs.load(); - } - size_t get_epoch_number() { return m_epoch_number; } @@ -145,32 +121,6 @@ public: return true; } - void set_inactive() { - m_active = false; - } - - /* - * - */ - bool retirable() { - /* if epoch is currently active, then it cannot be retired */ - if (m_active) { - return false; - } - - /* - * if the epoch has active jobs but is not itself active, - * wait for them to finish and return true. If there are - * not active jobs, return true immediately - */ - std::unique_lock lk(m_cv_lock); - while (m_active_jobs.load() > 0) { - m_active_cv.wait(lk); - } - - return true; - } - bool advance_buffer_head(size_t head) { m_buffer_head = head; return m_buffer->advance_head(m_buffer_head); @@ -180,9 +130,6 @@ private: Structure *m_structure; Buffer *m_buffer; - std::condition_variable m_active_cv; - std::mutex m_cv_lock; - std::mutex m_buffer_lock; std::atomic m_active_merge; @@ -192,8 +139,6 @@ private: * epoch. An epoch can only be retired * when this number is 0. */ - std::atomic m_active_jobs; - bool m_active; size_t m_epoch_number; size_t m_buffer_head; }; -- cgit v1.2.3 From 10b4425e842d10b7fbfa85978969ed4591d6b98e Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Wed, 7 Feb 2024 10:56:52 -0500 Subject: Fully implemented Query concept and adjusted queries to use it --- include/framework/scheduling/Epoch.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index 7b533b6..48b7742 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -18,7 +18,7 @@ namespace de { -template +template Q, LayoutPolicy L> class Epoch { private: typedef MutableBuffer Buffer; -- cgit v1.2.3 From 2c5d549b3618b9ea72e6eece4cb4f3da5a6811a8 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Wed, 7 Feb 2024 13:42:34 -0500 Subject: Fully realized shard concept interface --- include/framework/scheduling/Epoch.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index 48b7742..e58bd11 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -18,7 +18,7 @@ namespace de { -template Q, LayoutPolicy L> +template S, QueryInterface Q, LayoutPolicy L> class Epoch { private: typedef MutableBuffer Buffer; -- cgit v1.2.3 From 402fc269c0aaa671d84a6d15918735ad4b90e6b2 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Fri, 9 Feb 2024 12:30:21 -0500 Subject: Comment updates/fixes --- include/framework/scheduling/Epoch.h | 7 ------- 1 file changed, 7 deletions(-) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index e58bd11..3ffa145 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -44,13 +44,6 @@ public: } ~Epoch() { - /* FIXME: this is needed to keep the destructor from sometimes locking - * up here. But there *shouldn't* be any threads waiting on this signal - * at object destruction, so something else is going on here that needs - * looked into - */ - // m_active_cv.notify_all(); - if (m_structure) { m_structure->release_reference(); } -- cgit v1.2.3 From 3ddafd3b9ac089252814af87cb7d9fe534cf59a4 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Fri, 9 Feb 2024 13:09:05 -0500 Subject: Removed centralized version structure --- include/framework/scheduling/Epoch.h | 5 +++++ 1 file changed, 5 insertions(+) (limited to 'include/framework/scheduling/Epoch.h') diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h index 3ffa145..9377fb0 100644 --- a/include/framework/scheduling/Epoch.h +++ b/include/framework/scheduling/Epoch.h @@ -47,6 +47,11 @@ public: if (m_structure) { m_structure->release_reference(); } + + if (m_structure->get_reference_count() == 0) { + delete m_structure; + } + } /* -- cgit v1.2.3