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/FIFOScheduler.h | 96 ++++++++++++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 include/framework/scheduling/FIFOScheduler.h (limited to 'include/framework/scheduling/FIFOScheduler.h') 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); + } +}; + +} -- cgit v1.2.3 From 40b87b74f2bf4e93fdc9dabd6eab9175187fb63c Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Mon, 30 Oct 2023 14:47:16 -0400 Subject: FIFOScheduler: correctly protect m_cv with a lock --- include/framework/scheduling/FIFOScheduler.h | 5 +++++ 1 file changed, 5 insertions(+) (limited to 'include/framework/scheduling/FIFOScheduler.h') diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h index 878bb81..7ccab26 100644 --- a/include/framework/scheduling/FIFOScheduler.h +++ b/include/framework/scheduling/FIFOScheduler.h @@ -44,13 +44,18 @@ public: ~FIFOScheduler() { shutdown(); + std::unique_lock lk(m_cv_lock); m_cv.notify_all(); + lk.release(); + 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)); + + std::unique_lock lk(m_cv_lock); m_cv.notify_all(); } -- 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/FIFOScheduler.h | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) (limited to 'include/framework/scheduling/FIFOScheduler.h') diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h index 7ccab26..5425c4f 100644 --- a/include/framework/scheduling/FIFOScheduler.h +++ b/include/framework/scheduling/FIFOScheduler.h @@ -1,8 +1,7 @@ /* - * include/framework/Scheduler.h + * include/framework/scheduling/FIFOScheduler.h * * Copyright (C) 2023 Douglas B. Rumbaugh - * Dong Xie * * All rights reserved. Published under the Modified BSD License. * -- cgit v1.2.3 From 786a1cf5ab76f94a1adece48c1de53fb32e4551e Mon Sep 17 00:00:00 2001 From: "Douglas B. Rumbaugh" Date: Tue, 31 Oct 2023 11:54:09 -0400 Subject: FIFOScheduler: fixed a few synchronization issues --- include/framework/scheduling/FIFOScheduler.h | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) (limited to 'include/framework/scheduling/FIFOScheduler.h') diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h index 5425c4f..91a72b3 100644 --- a/include/framework/scheduling/FIFOScheduler.h +++ b/include/framework/scheduling/FIFOScheduler.h @@ -43,23 +43,20 @@ public: ~FIFOScheduler() { shutdown(); - std::unique_lock lk(m_cv_lock); m_cv.notify_all(); - lk.release(); - m_sched_thrd.join(); } void schedule_job(std::function job, size_t size, void *args) { + std::unique_lock lk(m_cv_lock); size_t ts = m_counter.fetch_add(1); m_task_queue.push(Task(size, ts, job, args)); - std::unique_lock lk(m_cv_lock); m_cv.notify_all(); } void shutdown() { - m_shutdown = true; + m_shutdown.store(true); } private: @@ -68,7 +65,7 @@ private: size_t m_memory_budget; size_t m_thrd_cnt; - bool m_shutdown; + std::atomic m_shutdown; std::atomic m_counter; std::mutex m_cv_lock; @@ -80,6 +77,7 @@ private: std::atomic m_used_memory; void schedule_next() { + assert(m_task_queue.size() > 0); auto t = m_task_queue.pop(); t(); } @@ -92,8 +90,7 @@ private: while (m_task_queue.size() > 0 && m_used_thrds.load() < m_thrd_cnt) { schedule_next(); } - cv_lock.unlock(); - } while(!m_shutdown); + } while(!m_shutdown.load()); } }; -- cgit v1.2.3 From 4e4cf858122ca6c1ae6d5f635e839089769fee38 Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Mon, 6 Nov 2023 10:01:23 -0500 Subject: Scheduling: Switched over to a thread pool model --- include/framework/scheduling/FIFOScheduler.h | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) (limited to 'include/framework/scheduling/FIFOScheduler.h') diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h index 91a72b3..1521eb6 100644 --- a/include/framework/scheduling/FIFOScheduler.h +++ b/include/framework/scheduling/FIFOScheduler.h @@ -24,20 +24,26 @@ #include "framework/structure/ExtensionStructure.h" #include "framework/scheduling/Task.h" +#include "ctpl/ctpl.h" #include "psu-ds/LockedPriorityQueue.h" namespace de { + class FIFOScheduler { +private: + static const size_t DEFAULT_MAX_THREADS = 8; + 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_thrd_cnt((thread_cnt) ? thread_cnt: DEFAULT_MAX_THREADS) , m_used_memory(0) , m_used_thrds(0) , m_shutdown(false) { m_sched_thrd = std::thread(&FIFOScheduler::run, this); + m_thrd_pool.resize(m_thrd_cnt); } ~FIFOScheduler() { @@ -72,6 +78,7 @@ private: std::condition_variable m_cv; std::thread m_sched_thrd; + ctpl::thread_pool m_thrd_pool; std::atomic m_used_thrds; std::atomic m_used_memory; @@ -79,7 +86,7 @@ private: void schedule_next() { assert(m_task_queue.size() > 0); auto t = m_task_queue.pop(); - t(); + m_thrd_pool.push(t); } void run() { @@ -87,7 +94,7 @@ private: 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) { + while (m_task_queue.size() > 0 && m_thrd_pool.n_idle() > 0) { schedule_next(); } } while(!m_shutdown.load()); -- 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/FIFOScheduler.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'include/framework/scheduling/FIFOScheduler.h') diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h index 1521eb6..ba62f9e 100644 --- a/include/framework/scheduling/FIFOScheduler.h +++ b/include/framework/scheduling/FIFOScheduler.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/FIFOScheduler.h | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) (limited to 'include/framework/scheduling/FIFOScheduler.h') diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h index ba62f9e..4cdc436 100644 --- a/include/framework/scheduling/FIFOScheduler.h +++ b/include/framework/scheduling/FIFOScheduler.h @@ -47,9 +47,10 @@ public: } ~FIFOScheduler() { - shutdown(); + if (!m_shutdown.load()) { + shutdown(); + } - m_cv.notify_all(); m_sched_thrd.join(); } @@ -63,6 +64,8 @@ public: void shutdown() { m_shutdown.store(true); + m_thrd_pool.stop(true); + m_cv.notify_all(); } private: -- cgit v1.2.3 From 38693c342558628c75e0ab0d23c32a95a499ed8b Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Fri, 19 Jan 2024 15:58:04 -0500 Subject: Initial rough-out of internal statistics tracker Need to figure out the best way to do the detailed tracking in a concurrent manner. I was thinking just an event log, with parsing routines for extracting statistics. But that'll be pretty slow. --- include/framework/scheduling/FIFOScheduler.h | 28 ++++++++++++++-------------- 1 file changed, 14 insertions(+), 14 deletions(-) (limited to 'include/framework/scheduling/FIFOScheduler.h') diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h index 4cdc436..513a3a2 100644 --- a/include/framework/scheduling/FIFOScheduler.h +++ b/include/framework/scheduling/FIFOScheduler.h @@ -8,21 +8,11 @@ */ #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 "framework/scheduling/statistics.h" #include "ctpl/ctpl.h" #include "psu-ds/LockedPriorityQueue.h" @@ -54,10 +44,12 @@ public: m_sched_thrd.join(); } - void schedule_job(std::function job, size_t size, void *args) { + void schedule_job(std::function job, size_t size, void *args, size_t type=0) { std::unique_lock lk(m_cv_lock); size_t ts = m_counter.fetch_add(1); - m_task_queue.push(Task(size, ts, job, args)); + + m_stats.job_queued(ts, type, size); + m_task_queue.push(Task(size, ts, job, args, type, &m_stats)); m_cv.notify_all(); } @@ -68,6 +60,10 @@ public: m_cv.notify_all(); } + void print_statistics() { + m_stats.print_statistics(); + } + private: psudb::LockedPriorityQueue m_task_queue; @@ -86,9 +82,13 @@ private: std::atomic m_used_thrds; std::atomic m_used_memory; + SchedulerStatistics m_stats; + void schedule_next() { assert(m_task_queue.size() > 0); auto t = m_task_queue.pop(); + m_stats.job_scheduled(t.m_timestamp); + m_thrd_pool.push(t); } -- cgit v1.2.3 From 4d0d26bfef684566a371ca7c87ba84df52f25ccc Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Mon, 22 Jan 2024 10:42:58 -0500 Subject: FIFOScheduler: added automake wakeup Sometimes, when the max thread count is exceeded, it is possible for the scheduler to lock up. This is because the scheduler is only run when a new job is put into the queue, and so it is possible for a job to be blocked by thread limitations and be left sitting in the queue. If the main program is waiting on this job to finish before scheduling a new one, then the system deadlocks. I added a second background thread to the scheduler that wakes the scheduler up every 20us to resolve this and prevent these deadlocks. --- include/framework/scheduling/FIFOScheduler.h | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) (limited to 'include/framework/scheduling/FIFOScheduler.h') diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h index 513a3a2..b77a8a1 100644 --- a/include/framework/scheduling/FIFOScheduler.h +++ b/include/framework/scheduling/FIFOScheduler.h @@ -10,7 +10,7 @@ #include #include - +#include #include "framework/scheduling/Task.h" #include "framework/scheduling/statistics.h" @@ -19,6 +19,8 @@ namespace de { +using namespace std::literals::chrono_literals; + class FIFOScheduler { private: @@ -33,6 +35,7 @@ public: , m_shutdown(false) { m_sched_thrd = std::thread(&FIFOScheduler::run, this); + m_sched_wakeup_thrd = std::thread(&FIFOScheduler::periodic_wakeup, this); m_thrd_pool.resize(m_thrd_cnt); } @@ -77,6 +80,7 @@ private: std::condition_variable m_cv; std::thread m_sched_thrd; + std::thread m_sched_wakeup_thrd; ctpl::thread_pool m_thrd_pool; std::atomic m_used_thrds; @@ -84,6 +88,13 @@ private: SchedulerStatistics m_stats; + void periodic_wakeup() { + do { + std::this_thread::sleep_for(10us); + m_cv.notify_all(); + } while (!m_shutdown.load()); + } + void schedule_next() { assert(m_task_queue.size() > 0); auto t = m_task_queue.pop(); -- cgit v1.2.3 From b1e4182825e6c162571b7cc4efaf8bc44055b49c Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Mon, 22 Jan 2024 12:10:54 -0500 Subject: Adjusted recon_benchmark and properly shutdown FIFOScheduler --- include/framework/scheduling/FIFOScheduler.h | 1 + 1 file changed, 1 insertion(+) (limited to 'include/framework/scheduling/FIFOScheduler.h') diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h index b77a8a1..0df4d3c 100644 --- a/include/framework/scheduling/FIFOScheduler.h +++ b/include/framework/scheduling/FIFOScheduler.h @@ -45,6 +45,7 @@ public: } m_sched_thrd.join(); + m_sched_wakeup_thrd.join(); } void schedule_job(std::function job, size_t size, void *args, size_t type=0) { -- cgit v1.2.3 From d166465dcca3550cb8f3263e0f5b5189a69d531a Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Wed, 31 Jan 2024 13:29:49 -0500 Subject: Temporary thread affinity for reconstruction --- include/framework/scheduling/FIFOScheduler.h | 1 + 1 file changed, 1 insertion(+) (limited to 'include/framework/scheduling/FIFOScheduler.h') diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h index 0df4d3c..c6baf9b 100644 --- a/include/framework/scheduling/FIFOScheduler.h +++ b/include/framework/scheduling/FIFOScheduler.h @@ -114,6 +114,7 @@ private: } } while(!m_shutdown.load()); } + }; } -- 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/FIFOScheduler.h | 9 +++++++++ 1 file changed, 9 insertions(+) (limited to 'include/framework/scheduling/FIFOScheduler.h') diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h index c6baf9b..3ed4f49 100644 --- a/include/framework/scheduling/FIFOScheduler.h +++ b/include/framework/scheduling/FIFOScheduler.h @@ -5,6 +5,15 @@ * * Distributed under the Modified BSD License. * + * This scheduler runs just concurrently, using a standard FIFO queue to + * determine which jobs to run next. If more jobs are scheduled than there + * are available threads, the excess will stall until a thread becomes + * available and then run in the order they were received by the scheduler. + * + * TODO: We need to set up a custom threadpool based on jthreads to support + * thread preemption for a later phase of this project. That will allow us + * to avoid blocking epoch transitions on long-running queries, or to pause + * reconstructions on demand. */ #pragma once -- cgit v1.2.3