Displace threading into core (qfortress graft)

4 files changed, 182 insertions, 1 deletions

diff --git a/core/CMakeLists.txt b/core/CMakeLists.txt
index edcab85..16cc4f0 100644
--- a/core/CMakeLists.txt
+++ b/core/CMakeLists.txt
@@ -6,12 +6,14 @@ configure_file("${CMAKE_CURRENT_LIST_DIR}/version.cc.in" "${CMAKE_CURRENT_LIST_D
 
 add_library(core STATIC
     "${CMAKE_CURRENT_LIST_DIR}/pch.hh"
+    "${CMAKE_CURRENT_LIST_DIR}/threading.cc"
+    "${CMAKE_CURRENT_LIST_DIR}/threading.hh"
     "${CMAKE_CURRENT_LIST_DIR}/version.cc"
     "${CMAKE_CURRENT_LIST_DIR}/version.hh")
 target_compile_features(core PUBLIC cxx_std_20)
 target_include_directories(core PUBLIC "${PROJECT_SOURCE_DIR}")
 target_precompile_headers(core PRIVATE "${CMAKE_CURRENT_LIST_DIR}/pch.hh")
-target_link_libraries(core PUBLIC enet emhash glm physfs spdlog stb)
+target_link_libraries(core PUBLIC enet emhash glm physfs spdlog stb thread_pool)
 
 add_subdirectory(config)
 add_subdirectory(io)
diff --git a/core/pch.hh b/core/pch.hh
index 924b1dd..3a013bd 100644
--- a/core/pch.hh
+++ b/core/pch.hh
@@ -25,6 +25,8 @@
 #include <unordered_set>
 #include <vector>
 
+#include <BS_thread_pool.hpp>
+
 #include <emhash/hash_table8.hpp>
 
 #include <enet/enet.h>
diff --git a/core/threading.cc b/core/threading.cc
new file mode 100644
index 0000000..4c7f8c8
--- /dev/null
+++ b/core/threading.cc
@@ -0,0 +1,127 @@
+#include "core/pch.hh"
+
+#include "core/threading.hh"
+
+#include "core/io/cmdline.hh"
+#include "core/math/constexpr.hh"
+
+constexpr static std::string_view DEFAULT_POOL_SIZE_ARG = "4";
+
+static BS::light_thread_pool* thread_pool;
+static std::deque<Task*> task_deque;
+
+static void task_process(Task* task)
+{
+    task->set_status(task_status::PROCESSING);
+    task->process();
+
+    if(task->get_status() == task_status::PROCESSING) {
+        // If the task status is still PROCESSING
+        // it can be deduced it hasn't been cancelled
+        task->set_status(task_status::COMPLETED);
+    }
+}
+
+task_status Task::get_status(void) const
+{
+    return m_status;
+}
+
+void Task::set_status(task_status status)
+{
+    m_status = status;
+}
+
+void threading::init(void)
+{
+    auto argument = io::cmdline::get("threads", DEFAULT_POOL_SIZE_ARG);
+    auto num_concurrent_threads = std::thread::hardware_concurrency();
+    unsigned int thread_pool_size;
+
+    if(num_concurrent_threads && 0 == argument.compare("max")) {
+        // Use the maximum available number of concurrent
+        // hardware threads provided by the implementation
+        thread_pool_size = num_concurrent_threads;
+    }
+    else {
+        if(num_concurrent_threads) {
+            auto result = std::from_chars(argument.data(), argument.data() + argument.size(), thread_pool_size);
+
+            if(result.ec == std::errc()) {
+                thread_pool_size = math::clamp<unsigned int>(thread_pool_size, 1U, num_concurrent_threads);
+            }
+            else {
+                thread_pool_size = 4U;
+            }
+        }
+        else {
+            auto result = std::from_chars(argument.data(), argument.data() + argument.size(), thread_pool_size);
+
+            if(result.ec == std::errc()) {
+                thread_pool_size = math::max<unsigned int>(thread_pool_size, 1U);
+            }
+            else {
+                thread_pool_size = 4U;
+            }
+        }
+    }
+
+    spdlog::info("threading: using {} threads for pooling tasks", thread_pool_size);
+
+    thread_pool = new BS::light_thread_pool(thread_pool_size);
+
+    task_deque.clear();
+}
+
+void threading::shutdown(void)
+{
+    for(auto task : task_deque) {
+        auto status = task->get_status();
+        if((status != task_status::CANCELLED) || (status != task_status::COMPLETED)) {
+            task->set_status(task_status::CANCELLED);
+        }
+    }
+
+    thread_pool->purge();
+    thread_pool->wait();
+
+    for(auto task : task_deque)
+        delete task;
+    task_deque.clear();
+
+    delete thread_pool;
+}
+
+void threading::update(void)
+{
+    auto task_iter = task_deque.cbegin();
+
+    while(task_iter != task_deque.cend()) {
+        auto task_ptr = *task_iter;
+        auto status = task_ptr->get_status();
+
+        if(status == task_status::CANCELLED) {
+            delete task_ptr;
+            task_iter = task_deque.erase(task_iter);
+            continue;
+        }
+
+        if(status == task_status::COMPLETED) {
+            task_ptr->finalize();
+            delete task_ptr;
+            task_iter = task_deque.erase(task_iter);
+            continue;
+        }
+
+        task_iter = std::next(task_iter);
+    }
+}
+
+void threading::detail::submit_new(Task* task)
+{
+    task->set_status(task_status::ENQUEUED);
+
+    static_cast<void>(thread_pool->submit_task(std::bind(&task_process, task)));
+
+    task_deque.push_back(task);
+}
diff --git a/core/threading.hh b/core/threading.hh
new file mode 100644
index 0000000..bd359ad
--- /dev/null
+++ b/core/threading.hh
@@ -0,0 +1,50 @@
+#ifndef CORE_THREADING_HH
+#define CORE_THREADING_HH 1
+#pragma once
+
+enum class task_status : unsigned int {
+    ENQUEUED = 0x0000U,
+    PROCESSING = 0x0001U,
+    COMPLETED = 0x0002U,
+    CANCELLED = 0x0004U,
+};
+
+class Task {
+public:
+    virtual ~Task(void) = default;
+    virtual void process(void) = 0;
+    virtual void finalize(void) = 0;
+
+    task_status get_status(void) const;
+    void set_status(task_status status);
+
+protected:
+    std::atomic<task_status> m_status;
+    std::future<void> m_future;
+};
+
+namespace threading
+{
+void init(void);
+void shutdown(void);
+void update(void);
+} // namespace threading
+
+namespace threading::detail
+{
+void submit_new(Task* task);
+} // namespace threading::detail
+
+namespace threading
+{
+template<typename T, typename... AT>
+void submit(AT&&... args);
+} // namespace threading
+
+template<typename T, typename... AT>
+inline void threading::submit(AT&&... args)
+{
+    threading::detail::submit_new(new T(args...));
+}
+
+#endif // CORE_THREADING_HH