From f0cc06c7388acb32b86301965c5b2547e4e3b919 Mon Sep 17 00:00:00 2001 From: untodesu Date: Thu, 11 Sep 2025 13:51:50 +0500 Subject: Displace threading into core (qfortress graft) --- core/CMakeLists.txt | 4 +- core/pch.hh | 2 + core/threading.cc | 127 ++++++++++++++++++++++++++++++++++++++++++++++++++++ core/threading.hh | 50 +++++++++++++++++++++ 4 files changed, 182 insertions(+), 1 deletion(-) create mode 100644 core/threading.cc create mode 100644 core/threading.hh (limited to 'core') 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 #include +#include + #include #include 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_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(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(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(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 m_status; + std::future 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 +void submit(AT&&... args); +} // namespace threading + +template +inline void threading::submit(AT&&... args) +{ + threading::detail::submit_new(new T(args...)); +} + +#endif // CORE_THREADING_HH -- cgit