diff --git a/include/simdb/apps/AppManager.hpp b/include/simdb/apps/AppManager.hpp index 0088b8f3..be254184 100644 --- a/include/simdb/apps/AppManager.hpp +++ b/include/simdb/apps/AppManager.hpp @@ -306,35 +306,6 @@ class AppManager return nullptr; } - /// Optionally call this method after initializePipelines(), but before - /// openPipelines(). This will reduce the number of non-database threads - /// to the minimum across all app pipelines. - /// - /// Note that you can either call minimizeThreads() OR minimizeThreads(app1, - /// app2, ...) but you cannot call both. - void minimizeThreads() - { - if (!pipeline_mgr_) - { - throw DBException("Pipeline manager not set - did you call " - "initializePipelines()?"); - } - pipeline_mgr_->minimizeThreads(); - } - - /// Optionally call this method after initializePipelines(), but before - /// openPipelines(). This will share the minimum number of non-database - /// threads across the given apps' pipelines. - template void minimizeThreads(const App* app, Apps&&... rest) - { - if (!pipeline_mgr_) - { - throw DBException("Pipeline manager not set - did you call " - "initializePipelines()?"); - } - pipeline_mgr_->minimizeThreads(app, std::forward(rest)...); - } - /// \brief Set the order in which app lifecycle hooks are invoked. /// /// postInit(), preTeardown(), and postTeardown() are called in this order @@ -552,8 +523,7 @@ class AppManager std::cout << std::endl; } - /// Call this once after initializePipelines() (and after minimizeThreads() - /// if you called that too). + /// Call this once after initializePipelines(). void openPipelines_() { PROFILE_APP_PHASE @@ -994,8 +964,7 @@ class AppManagers } } - /// Call this once after initializePipelines() (and after minimizeThreads() - /// if you called that too). + /// Call this once after initializePipelines(). void openPipelines() { for (auto& [app_mgr, _] : getAllManagers()) diff --git a/include/simdb/pipeline/DatabaseThread.hpp b/include/simdb/pipeline/DatabaseThread.hpp index 90671bb3..421d0c4b 100644 --- a/include/simdb/pipeline/DatabaseThread.hpp +++ b/include/simdb/pipeline/DatabaseThread.hpp @@ -39,6 +39,20 @@ class DatabaseThread : public PollingThread, private AsyncDatabaseAccessHandler /// \brief Return the AsyncDatabaseAccessor for submitting work to this thread. AsyncDatabaseAccessor* getAsyncDatabaseAccessor() { return &db_accessor_; } + /// \brief Start the database polling thread and dormant async task thread. + void open() override + { + PollingThread::open(); + dormant_thread_.open(); + } + + /// \brief Stop and join the database polling thread and dormant async task thread. + void close() noexcept override + { + PollingThread::close(); + dormant_thread_.close(); + } + private: /// Overridden from AsyncDatabaseAccessHandler void eval(AsyncDatabaseTaskPtr&& task, double timeout_seconds = 0) override final @@ -88,20 +102,6 @@ class DatabaseThread : public PollingThread, private AsyncDatabaseAccessHandler return did_work; } - /// Overridden from PollingThread - void open() override - { - PollingThread::open(); - dormant_thread_.open(); - } - - /// Overridden from PollingThread - void close() noexcept override - { - PollingThread::close(); - dormant_thread_.close(); - } - /// Overridden from PollingThread bool flushRunnables() override { diff --git a/include/simdb/pipeline/Pipeline.hpp b/include/simdb/pipeline/Pipeline.hpp index f6f36594..fb89c9e7 100644 --- a/include/simdb/pipeline/Pipeline.hpp +++ b/include/simdb/pipeline/Pipeline.hpp @@ -135,11 +135,12 @@ class Pipeline return queue_repo_.getOutPortQueue(port_full_name); } - /// \brief Assign each stage to a PollingThread (or the shared DatabaseThread); call after noMoreBindings(). - /// \param threads Vector to which new PollingThreads may be appended. + /// \brief Assign each stage to the thread pool or the shared DatabaseThread; call after noMoreBindings(). + /// \param pool Thread pool for non-database stages. /// \param database_thread Single shared DatabaseThread for all DatabaseStages (created if null). - void assignStageThreads(std::vector>& threads, - std::unique_ptr& database_thread) + /// \param global_order Running index across all pipelines for pool runnable ordering. + void assignStageThreads(PollingThreadPool& pool, std::unique_ptr& database_thread, + size_t& global_order) { if (state_ != State::BINDINGS_COMPLETE) { @@ -149,9 +150,9 @@ class Pipeline queue_repo_.validateQueues(); - for (auto& [stage_name, stage] : stages_) + for (const auto& stage_name : stages_in_order_) { - stage->assignThread_(db_mgr_, threads, database_thread); + stages_.at(stage_name)->assignThread_(db_mgr_, pool, database_thread, global_order); } state_ = State::FINALIZED; diff --git a/include/simdb/pipeline/PipelineManager.hpp b/include/simdb/pipeline/PipelineManager.hpp index 2e1e73f3..7b0564c7 100644 --- a/include/simdb/pipeline/PipelineManager.hpp +++ b/include/simdb/pipeline/PipelineManager.hpp @@ -5,8 +5,7 @@ #include "simdb/pipeline/DatabaseThread.hpp" #include "simdb/pipeline/Pipeline.hpp" #include "simdb/pipeline/PipelineSnooper.hpp" -#include "simdb/pipeline/PollingThread.hpp" -#include "simdb/pipeline/ThreadMerger.hpp" +#include "simdb/pipeline/PollingThreadPool.hpp" #include @@ -19,9 +18,9 @@ namespace simdb::pipeline { /*! * \class PipelineManager * - * \brief Manages all Pipeline instances and their PollingThreads for an - * AppManager (or unit test). Creates pipelines, merges threads - * (minimizeThreads), opens threads, and provides async DB access. + * \brief Manages all Pipeline instances, a PollingThreadPool for non-database + * stages, and a dedicated DatabaseThread for database stages. Creates + * pipelines, opens threads, and provides async DB access. */ class PipelineManager { @@ -79,83 +78,37 @@ class PipelineManager return std::make_unique>(this); } - /// \brief Merge all apps' pipeline threads into a minimal set; call at most once. - /// \throws DBException if called more than once. - void minimizeThreads() - { - if (thread_merger_) - { - throw DBException("You can only call minimizeThreads() method once."); - } - - thread_merger_ = std::make_unique(pipelines_); - thread_merger_->mergeAllAppThreads(); - } - - /// \brief Mark one app's pipeline threads for merging (call before openPipelines()). - void minimizeThreads(const App* app) - { - if (!thread_merger_) - { - throw DBException("Cannot merge a single app's pipeline threads"); - } - thread_merger_->addAppForMerging(app); - } - - /// \brief Mark multiple apps' pipeline threads for merging (variadic). - template void minimizeThreads(const App* app, Apps&&... rest) - { - if (!thread_merger_) - { - thread_merger_ = std::make_unique(pipelines_); - } - thread_merger_->addAppForMerging(app); - minimizeThreads(std::forward(rest)...); - } - - /// \brief Create and open all polling threads (after stages are added and optionally minimizeThreads). + /// \brief Register stages with the thread pool and open all polling threads. void openPipelines() { checkOpen_(); - if (!thread_merger_) + size_t global_order = 0; + for (auto& pipeline : pipelines_) { - thread_merger_ = std::make_unique(pipelines_); + pipeline->assignStageThreads(thread_pool_, database_thread_, global_order); } - thread_merger_->performMerge(polling_threads_); - // Now that all threads are created, give the async DB accessor to all - // non-DB stages in all pipelines. - for (auto& thread : polling_threads_) + if (database_thread_) { - if (auto db_thread = dynamic_cast(thread.get())) - { - async_db_accessor_ = db_thread->getAsyncDatabaseAccessor(); - break; - } + async_db_accessor_ = database_thread_->getAsyncDatabaseAccessor(); } if (async_db_accessor_) { - for (auto& thread : polling_threads_) + for (auto runnable : thread_pool_.getRegisteredRunnables()) { - if (!dynamic_cast(thread.get())) + if (auto stage = dynamic_cast(runnable)) { - for (auto runnable : thread->getRunnables()) - { - if (auto stage = dynamic_cast(runnable)) - { - stage->setAsyncDatabaseAccessor_(async_db_accessor_); - } - } + stage->setAsyncDatabaseAccessor_(async_db_accessor_); } } } - // Now open all threads for simulation - for (auto& thread : polling_threads_) + thread_pool_.open(); + if (database_thread_) { - thread->open(); + database_thread_->open(); } threads_opened_ = true; } @@ -189,22 +142,21 @@ class PipelineManager return disabler; } - /// \brief Close all threads, flush runnables, and print performance reports. + /// \brief Close all threads, flush runnables, and print the pool performance report. void postSimLoopTeardown() { checkOpen_(); - auto close_thread = [&](PollingThread* thread) { - thread->close(); - thread->printPerfReport(); - std::cout << "\n\n"; - }; + auto threads = thread_pool_.getWorkerThreads(); + if (database_thread_) + { + threads.push_back(database_thread_.get()); + } - auto it = polling_threads_.begin(); - while (it != polling_threads_.end()) + thread_pool_.close(); + if (database_thread_) { - close_thread(it->get()); - ++it; + database_thread_->close(); } bool continue_while; @@ -212,14 +164,13 @@ class PipelineManager { continue_while = false; - it = polling_threads_.begin(); - while (it != polling_threads_.end()) + for (auto thread : threads) { - continue_while |= (*it)->flushRunnables(); - ++it; + continue_while |= thread->flushRunnables(); } } while (continue_while); + thread_pool_.printPerfReport(); closed_ = true; } @@ -230,8 +181,11 @@ class PipelineManager /// Instantiated pipelines. std::vector> pipelines_; - /// Instantiated threads. - std::vector> polling_threads_; + /// Pool of worker threads for non-database stages. + PollingThreadPool thread_pool_; + + /// Dedicated database thread (never part of the pool). + std::unique_ptr database_thread_; /// Threads that we give to the ScopedRunnableDisabler. std::vector disabler_threads_; @@ -250,10 +204,6 @@ class PipelineManager /// Cached AsyncDatabaseAccessor for async DB queries. AsyncDatabaseAccessor* async_db_accessor_ = nullptr; - /// Used to perform minimizeThread() to share threads - /// between concurrently running apps. - std::unique_ptr thread_merger_; - void getDisablerThreads_() { if (!disabler_threads_.empty()) @@ -261,9 +211,10 @@ class PipelineManager return; } - for (auto& thread : polling_threads_) + disabler_threads_ = thread_pool_.getWorkerThreads(); + if (database_thread_) { - disabler_threads_.push_back(thread.get()); + disabler_threads_.push_back(database_thread_.get()); } // Ensure unique @@ -281,10 +232,14 @@ class PipelineManager return; } - for (auto& thread : polling_threads_) + for (auto runnable : thread_pool_.getRegisteredRunnables()) + { + disabler_runnables_.push_back(runnable); + } + if (database_thread_) { - const auto& runnables = thread->getRunnables(); - disabler_runnables_.insert(disabler_runnables_.end(), runnables.begin(), runnables.end()); + const auto& db_runnables = database_thread_->getRunnables(); + disabler_runnables_.insert(disabler_runnables_.end(), db_runnables.begin(), db_runnables.end()); } // Ensure unique diff --git a/include/simdb/pipeline/PollingThread.hpp b/include/simdb/pipeline/PollingThread.hpp index 63b036eb..30bc9f91 100644 --- a/include/simdb/pipeline/PollingThread.hpp +++ b/include/simdb/pipeline/PollingThread.hpp @@ -4,26 +4,46 @@ #include "simdb/Exceptions.hpp" #include "simdb/pipeline/Runnable.hpp" -#include "simdb/utils/StreamFormatters.hpp" +#include #include #include #include #include -#include -#include #include +#include #include +#include #include namespace simdb::pipeline { +class PollingThreadPool; + +/// \brief Point-in-time metrics exported by a PollingThread for pool load balancing. +struct PollingThreadMetrics +{ + size_t num_runnables = 0; + uint64_t num_poll_cycles_with_work = 0; + double total_sleep_seconds = 0.0; + double elapsed_seconds = 0.0; + bool is_running = false; + bool is_paused = false; +}; + +/// \brief Per-Runnable PROCEED/SLEEP poll counts since the last pool rebalance reset. +struct RunnablePollMetrics +{ + uint64_t proceed_count = 0; + uint64_t sleep_count = 0; +}; + /*! * \class PollingThread * * \brief Thread that repeatedly polls its Runnables for work; when none do * work, it sleeps for a fixed interval before polling again. Supports - * pause/resume and performance reporting. Base for DatabaseThread. + * pause/resume. Base for DatabaseThread. */ class PollingThread { @@ -57,6 +77,91 @@ class PollingThread /// \brief Return the number of Runnables on this thread. size_t getNumRunnables() const { return runnables_.size(); } + /// \brief Export metrics for PollingThreadPool load balancing. + PollingThreadMetrics getMetrics() const noexcept + { + PollingThreadMetrics metrics; + metrics.num_runnables = runnables_.size(); + metrics.num_poll_cycles_with_work = num_times_run_; + metrics.total_sleep_seconds = total_sleep_seconds_; + metrics.is_running = is_running_; + metrics.is_paused = paused_; + if (is_running_) + { + auto now = std::chrono::high_resolution_clock::now(); + metrics.elapsed_seconds = std::chrono::duration(now - start_).count(); + } + return metrics; + } + + /// \brief Return PROCEED/SLEEP poll counts for \p runnable since the last rebalance reset. + RunnablePollMetrics getRunnablePollMetrics(Runnable* runnable) const + { + const auto it = runnable_poll_metrics_.find(runnable); + if (it == runnable_poll_metrics_.end()) + { + return {}; + } + return it->second; + } + + /// \brief Sum of PROCEED poll counts across all runnables on this thread. + uint64_t getTotalProceedPolls() const + { + uint64_t total = 0; + for (const auto& [_, metrics] : runnable_poll_metrics_) + { + total += metrics.proceed_count; + } + return total; + } + + /// \brief Sum of SLEEP poll counts across all runnables on this thread. + uint64_t getTotalSleepPolls() const + { + uint64_t total = 0; + for (const auto& [_, metrics] : runnable_poll_metrics_) + { + total += metrics.sleep_count; + } + return total; + } + + /// \brief Clear per-runnable poll counters (called by PollingThreadPool after each rebalance). + void resetPollMetrics() { runnable_poll_metrics_.clear(); } + + /// \brief Add a Runnable while this thread is paused (used during pool migration). + /// \pre paused() == true + /// \throws DBException if the thread is not paused. + void addRunnableWhilePaused(Runnable* runnable) + { + if (!paused_) + { + throw DBException("addRunnableWhilePaused() requires a paused PollingThread"); + } + runnables_.emplace_back(runnable); + } + + /// \brief Remove a Runnable while this thread is paused (used during pool migration). + /// \pre paused() == true + /// \return true if \p runnable was found and removed. + /// \throws DBException if the thread is not paused. + bool removeRunnableWhilePaused(Runnable* runnable) + { + if (!paused_) + { + throw DBException("removeRunnableWhilePaused() requires a paused PollingThread"); + } + auto it = std::find(runnables_.begin(), runnables_.end(), runnable); + if (it == runnables_.end()) + { + return false; + } + runnables_.erase(it); + runnable_poll_metrics_.erase(runnable); + return true; + } + /// \brief Reorder this thread's Runnables to match the order in \p runnables (only those /// that belong to this thread). void ensureRelativeOrder(const std::vector& runnables) @@ -77,29 +182,39 @@ class PollingThread virtual bool flushRunnables() { bool did_work = false; - for (auto runnable : runnables_) + while (true) { - if (!runnable->enabled()) + bool processed = false; + for (auto runner : runnables_) { - continue; - } + if (!runner->enabled()) + { + continue; + } - if (runnable->processAll(true) == PipelineAction::PROCEED) + const auto action = runner->processOne(true); + auto& poll_metrics = runnable_poll_metrics_[runner]; + if (action == PipelineAction::PROCEED) + { + processed = true; + ++poll_metrics.proceed_count; + } else + { + ++poll_metrics.sleep_count; + } + } + if (!processed) { - did_work = true; + break; } + did_work = true; } return did_work; } - /// \brief Start the polling thread (must have at least one Runnable). + /// \brief Start the polling thread. virtual void open() { - if (runnables_.empty()) - { - return; - } - if (!thread_) { stop_requested_ = false; @@ -176,40 +291,6 @@ class PollingThread pause_cv_.notify_all(); } - /// \brief Print a performance report (sleep vs work %) for this thread. - void printPerfReport() const noexcept - { - if (runnables_.empty()) - { - return; - } - - if (is_running_) - { - return; - } - - auto now = std::chrono::high_resolution_clock::now(); - const std::chrono::duration dur = now - start_; - const auto total_elap_seconds = dur.count(); - const auto pct_time_sleeping = (total_sleep_seconds_ / total_elap_seconds) * 100; - const auto pct_time_working = 100 - pct_time_sleeping; - - std::cout << "Thread containing:\n"; - for (const auto runnable : runnables_) - { - runnable->print(std::cout, 4); - } - - [[maybe_unused]] ios_format_saver fmt_saver(std::cout); - std::cout << "\n"; - std::cout << " Performance report:\n"; - std::cout << " Num times run: " << num_times_run_ << "\n"; - std::cout << " Pct time sleeping: " << std::fixed << std::setprecision(1) << pct_time_sleeping << "%\n"; - std::cout << " Pct time working: " << std::fixed << std::setprecision(1) << pct_time_working << "%\n"; - std::cout << "\n"; - } - private: void loop_() { @@ -271,9 +352,15 @@ class PollingThread continue; } - if (runner->processOne(force) == PipelineAction::PROCEED) + const auto action = runner->processOne(force); + auto& poll_metrics = runnable_poll_metrics_[runner]; + if (action == PipelineAction::PROCEED) { processed = true; + ++poll_metrics.proceed_count; + } else + { + ++poll_metrics.sleep_count; } } if (!processed) @@ -302,6 +389,9 @@ class PollingThread std::chrono::high_resolution_clock::time_point start_; uint64_t num_times_run_ = 0; double total_sleep_seconds_ = 0; + std::unordered_map runnable_poll_metrics_; + + friend class PollingThreadPool; }; /// Defined here so we can avoid circular includes diff --git a/include/simdb/pipeline/PollingThreadPool.hpp b/include/simdb/pipeline/PollingThreadPool.hpp new file mode 100644 index 00000000..cb65fc99 --- /dev/null +++ b/include/simdb/pipeline/PollingThreadPool.hpp @@ -0,0 +1,1082 @@ +// -*- C++ -*- + +#pragma once + +#include "simdb/Exceptions.hpp" +#include "simdb/pipeline/DatabaseThread.hpp" +#include "simdb/pipeline/PollingThread.hpp" +#include "simdb/pipeline/Runnable.hpp" +#include "simdb/utils/StreamFormatters.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace simdb::pipeline { + +/// \brief Rolling load snapshot for one pool worker thread. +struct ThreadLoadSnapshot +{ + PollingThread* thread = nullptr; + size_t num_runnables = 0; + uint64_t num_poll_cycles_with_work = 0; + double pct_time_sleeping = 0.0; + double pct_time_working = 0.0; + /// Fraction of recent runnable polls that returned PROCEED (0..1). + double recent_busy_ratio = 0.0; + bool is_running = false; + bool is_paused = false; +}; + +/// \brief Metadata for a non-database Runnable registered with the pool. +struct RunnableRegistration +{ + Runnable* runnable = nullptr; + size_t interval_ms = 100; + /// Global pipeline definition order; used to preserve relative ordering + /// within a worker's runnable list (replaces ensureRelativeOrder()). + size_t global_order = 0; + PollingThread* assigned_thread = nullptr; + uint64_t migration_count = 0; + uint64_t lifetime_proceed_count = 0; + uint64_t lifetime_sleep_count = 0; +}; + +/*! + * \class PollingThreadPool + * + * \brief Owns a dynamic set of PollingThread workers and automatically + * balances non-database Runnables across them. Grows and shrinks the + * worker count based on observed load. DatabaseThread and its runnables + * never participate in this pool. + * + * Typical lifecycle (via PipelineManager): + * 1. registerRunnable() for each non-DB stage (before open) + * 2. open() — create initial workers, distribute runnables, start balancer + * 3. runtime — balancer steals/migrates runnables and resizes pool + * 4. close() — stop balancer, drain workers, join all threads + */ +class PollingThreadPool +{ +public: + /// \brief Tunables for pool sizing and steal decisions. + struct Config + { + /// Minimum number of worker PollingThreads kept open while the pool is running. + size_t min_threads = 1; + + /// Maximum number of worker PollingThreads the pool may create. + size_t max_threads = 0; // 0 => std::thread::hardware_concurrency() + + /// Default sleep interval for newly created workers (ms). + size_t default_interval_ms = 100; + + /// How often the balancer re-evaluates load and may steal or resize. + std::chrono::milliseconds rebalance_period_ms{500}; + + /// Steal source: thread recent_busy_ratio must exceed this (0..1). + double steal_busy_threshold = 0.70; + + /// Steal destination: thread recent_busy_ratio must be below this (0..1). + double steal_idle_threshold = 0.30; + + /// Grow pool when every worker exceeds this busy ratio (0..1). + double grow_all_busy_threshold = 0.80; + + /// Shrink candidate: worker busy ratio below this and runnable count <= shrink_max_runnables. + double shrink_idle_threshold = 0.10; + + /// Max runnables on a thread eligible for shrink (typically 0 or 1). + size_t shrink_max_runnables = 0; + }; + + explicit PollingThreadPool(Config config) : + config_(std::move(config)) + { + if (config_.max_threads == 0) + { + config_.max_threads = std::max(size_t{1}, static_cast(std::thread::hardware_concurrency())); + } + if (config_.min_threads > config_.max_threads) + { + throw DBException("Pool min threads cannot be greater than pool max threads"); + } + if (config_.min_threads == 0) + { + std::cout << "Zero pool min threads ignored; using 1 min thread" << std::endl; + config_.min_threads = 1; + } + } + + PollingThreadPool() : + PollingThreadPool(Config{}) + { + } + + /// \brief Register a pool-eligible Runnable before open(). + /// \param runnable Non-null stage or other non-DB runnable. + /// \param interval_ms Polling sleep interval when this runnable has no work. + /// \param global_order Position in global pipeline stage definition order. + /// \throws DBException if called after open() or if runnable is already registered. + void registerRunnable(Runnable* runnable, size_t interval_ms, size_t global_order) + { + if (is_open_) + { + throw DBException("Cannot register runnables after PollingThreadPool::open()"); + } + if (!runnable) + { + throw DBException("PollingThreadPool::registerRunnable() requires non-null runnable"); + } + for (const auto& reg : registrations_) + { + if (reg.runnable == runnable) + { + throw DBException("Runnable already registered with PollingThreadPool"); + } + } + registrations_.push_back({runnable, interval_ms, global_order, nullptr}); + } + + /// \brief Create workers, assign runnables, and start the balancer thread. + /// \throws DBException if already open. + void open(); + + /// \brief Stop the balancer, close all workers, and join threads. + void close() noexcept; + + /// \brief Return true after open() and before close(). + bool isOpen() const { return is_open_; } + + /// \brief Return all worker threads (for pause/disable integration). + std::vector getWorkerThreads() const; + + /// \brief Return registered runnables in global definition order. + std::vector getRegisteredRunnables() const; + + /// \brief Return the pool configuration. + const Config& getConfig() const { return config_; } + + /// \brief Force an immediate rebalance (primarily for tests). + void rebalanceNow(); + + /// \brief Print end-of-sim pool summary and final worker layout. + /// \note Call before close() so worker utilization metrics are still available. + void printPerfReport(std::ostream& os = std::cout); + +private: + struct IntervalWorkerGroup + { + size_t interval_ms = 100; + std::vector> workers; + }; + + void rebalanceLoop_(); + void rebalanceOnce_(); + void createInitialWorkers_(); + void distributeInitialRunnables_(); + void ensureRelativeOrderOnThread_(PollingThread* thread); + ThreadLoadSnapshot snapshotThread_(PollingThread* thread) const; + std::vector snapshotAllWorkers_() const; + IntervalWorkerGroup* findWorkerGroupForInterval_(size_t interval_ms); + const IntervalWorkerGroup* findWorkerGroupForThread_(PollingThread* thread) const; + size_t totalWorkerCount_() const; + + PollingThread* findStealDestination_(const ThreadLoadSnapshot& source, + const std::vector& snapshots) const; + Runnable* findStealCandidate_(PollingThread* source_thread) const; + + void migrateRunnable_(Runnable* runnable, PollingThread* from, PollingThread* to); + void growPool_(size_t interval_ms); + void shrinkPool_(const std::vector& snapshots); + void accumulatePollMetricsFromWorkers_(); + void resetWorkerPollMetrics_(); + double runnableProceedPct_(const RunnableRegistration& reg) const; + double workerProceedPct_(PollingThread* worker) const; + std::string formatWorkerLabel_(PollingThread* thread) const; + std::unique_ptr createWorker_(size_t interval_ms); + + Config config_; + std::vector registrations_; + std::vector worker_groups_; + + size_t initial_worker_count_ = 0; + size_t peak_worker_count_ = 0; + uint64_t num_rebalances_ = 0; + uint64_t num_steals_ = 0; + uint64_t num_grows_ = 0; + uint64_t num_shrinks_ = 0; + uint64_t num_migrations_ = 0; + uint64_t grow_blocked_at_max_ = 0; + + std::unique_ptr balancer_thread_; + std::mutex pool_mutex_; + std::atomic is_open_{false}; + std::atomic stop_balancer_{false}; +}; + +inline void PollingThreadPool::open() +{ + if (is_open_) + { + throw DBException("PollingThreadPool::open() called more than once"); + } + if (registrations_.empty()) + { + throw DBException("PollingThreadPool::open() requires at least one registered runnable"); + } + + std::lock_guard lock(pool_mutex_); + + createInitialWorkers_(); + distributeInitialRunnables_(); + initial_worker_count_ = totalWorkerCount_(); + peak_worker_count_ = initial_worker_count_; + + for (auto& group : worker_groups_) + { + for (auto& worker : group.workers) + { + worker->open(); + } + } + + stop_balancer_ = false; + balancer_thread_ = std::make_unique(&PollingThreadPool::rebalanceLoop_, this); + is_open_ = true; +} + +inline void PollingThreadPool::close() noexcept +{ + if (!is_open_) + { + return; + } + + stop_balancer_ = true; + if (balancer_thread_ && balancer_thread_->joinable()) + { + balancer_thread_->join(); + } + balancer_thread_.reset(); + + for (auto& group : worker_groups_) + { + for (auto& worker : group.workers) + { + worker->close(); + } + } + + for (auto& reg : registrations_) + { + reg.assigned_thread = nullptr; + } + + is_open_ = false; +} + +inline std::vector PollingThreadPool::getWorkerThreads() const +{ + std::vector threads; + for (const auto& group : worker_groups_) + { + for (const auto& worker : group.workers) + { + threads.push_back(worker.get()); + } + } + return threads; +} + +inline std::vector PollingThreadPool::getRegisteredRunnables() const +{ + std::vector runnables; + runnables.reserve(registrations_.size()); + for (const auto& reg : registrations_) + { + runnables.push_back(reg.runnable); + } + return runnables; +} + +inline void PollingThreadPool::rebalanceNow() +{ + if (!is_open_) + { + throw DBException("PollingThreadPool::rebalanceNow() requires an open pool"); + } + rebalanceOnce_(); +} + +inline void PollingThreadPool::rebalanceLoop_() +{ + while (!stop_balancer_) + { + std::this_thread::sleep_for(config_.rebalance_period_ms); + if (stop_balancer_) + { + break; + } + rebalanceOnce_(); + } +} + +inline void PollingThreadPool::rebalanceOnce_() +{ + std::lock_guard lock(pool_mutex_); + if (worker_groups_.empty()) + { + return; + } + + ++num_rebalances_; + + const auto snapshots = snapshotAllWorkers_(); + + // Steal one runnable from an overloaded worker to an idle peer (same interval group). + for (const auto& source : snapshots) + { + if (source.recent_busy_ratio <= config_.steal_busy_threshold) + { + continue; + } + if (source.num_runnables <= 1) + { + continue; + } + + PollingThread* dest = findStealDestination_(source, snapshots); + if (!dest) + { + continue; + } + + Runnable* candidate = findStealCandidate_(source.thread); + if (!candidate) + { + continue; + } + + migrateRunnable_(candidate, source.thread, dest); + ++num_steals_; + break; + } + + // Grow when every worker is busy and the pool is below its cap. + bool all_busy = !snapshots.empty(); + for (const auto& snap : snapshots) + { + if (snap.recent_busy_ratio <= config_.grow_all_busy_threshold) + { + all_busy = false; + break; + } + } + if (all_busy) + { + if (totalWorkerCount_() < config_.max_threads) + { + size_t grow_interval_ms = worker_groups_.front().interval_ms; + double busiest_avg = -1.0; + for (const auto& group : worker_groups_) + { + double sum_busy = 0.0; + size_t count = 0; + for (const auto& snap : snapshots) + { + const auto snap_group = findWorkerGroupForThread_(snap.thread); + if (snap_group && snap_group->interval_ms == group.interval_ms) + { + sum_busy += snap.recent_busy_ratio; + ++count; + } + } + if (count == 0) + { + continue; + } + const double avg_busy = sum_busy / static_cast(count); + if (avg_busy > busiest_avg) + { + busiest_avg = avg_busy; + grow_interval_ms = group.interval_ms; + } + } + growPool_(grow_interval_ms); + } else + { + ++grow_blocked_at_max_; + } + } + + shrinkPool_(snapshotAllWorkers_()); + resetWorkerPollMetrics_(); +} + +inline void PollingThreadPool::createInitialWorkers_() +{ + std::set intervals; + for (const auto& reg : registrations_) + { + intervals.insert(reg.interval_ms); + } + + const size_t num_groups = std::max(size_t{1}, intervals.size()); + const size_t workers_per_group = std::max(size_t{1}, config_.min_threads / num_groups); + + for (size_t interval_ms : intervals) + { + IntervalWorkerGroup group; + group.interval_ms = interval_ms; + for (size_t i = 0; i < workers_per_group; ++i) + { + group.workers.push_back(std::make_unique(interval_ms)); + } + worker_groups_.push_back(std::move(group)); + } + + while (totalWorkerCount_() < config_.min_threads) + { + const auto interval_ms = worker_groups_.front().interval_ms; + auto worker = std::make_unique(interval_ms); + worker_groups_.front().workers.emplace_back(std::move(worker)); + } +} + +inline PollingThreadPool::IntervalWorkerGroup* PollingThreadPool::findWorkerGroupForInterval_(size_t interval_ms) +{ + for (auto& group : worker_groups_) + { + if (group.interval_ms == interval_ms) + { + return &group; + } + } + return nullptr; +} + +inline const PollingThreadPool::IntervalWorkerGroup* +PollingThreadPool::findWorkerGroupForThread_(PollingThread* thread) const +{ + for (const auto& group : worker_groups_) + { + for (const auto& worker : group.workers) + { + if (worker.get() == thread) + { + return &group; + } + } + } + return nullptr; +} + +inline size_t PollingThreadPool::totalWorkerCount_() const +{ + size_t count = 0; + for (const auto& group : worker_groups_) + { + count += group.workers.size(); + } + return count; +} + +inline void PollingThreadPool::distributeInitialRunnables_() +{ + std::map worker_idx_by_interval; + for (auto& reg : registrations_) + { + auto group = findWorkerGroupForInterval_(reg.interval_ms); + if (!group || group->workers.empty()) + { + throw DBException("No worker group for runnable interval"); + } + + size_t& worker_idx = worker_idx_by_interval[reg.interval_ms]; + auto& worker = group->workers.at(worker_idx % group->workers.size()); + worker->addRunnable(reg.runnable); + reg.assigned_thread = worker.get(); + + ++worker_idx; + } + + for (auto& group : worker_groups_) + { + for (auto& worker : group.workers) + { + ensureRelativeOrderOnThread_(worker.get()); + } + } +} + +inline void PollingThreadPool::ensureRelativeOrderOnThread_(PollingThread* thread) +{ + std::vector ordered; + ordered.reserve(registrations_.size()); + for (const auto& reg : registrations_) + { + ordered.push_back(reg.runnable); + } + thread->ensureRelativeOrder(ordered); +} + +inline ThreadLoadSnapshot PollingThreadPool::snapshotThread_(PollingThread* thread) const +{ + ThreadLoadSnapshot snapshot; + snapshot.thread = thread; + const auto metrics = thread->getMetrics(); + snapshot.num_runnables = metrics.num_runnables; + snapshot.num_poll_cycles_with_work = metrics.num_poll_cycles_with_work; + snapshot.is_running = metrics.is_running; + snapshot.is_paused = metrics.is_paused; + if (metrics.elapsed_seconds > 0.0) + { + snapshot.pct_time_sleeping = (metrics.total_sleep_seconds / metrics.elapsed_seconds) * 100.0; + snapshot.pct_time_working = 100.0 - snapshot.pct_time_sleeping; + } + + const uint64_t proceed_polls = thread->getTotalProceedPolls(); + const uint64_t sleep_polls = thread->getTotalSleepPolls(); + const uint64_t total_polls = proceed_polls + sleep_polls; + if (total_polls > 0) + { + snapshot.recent_busy_ratio = static_cast(proceed_polls) / static_cast(total_polls); + } else if (metrics.elapsed_seconds > 0.0) + { + snapshot.recent_busy_ratio = snapshot.pct_time_working / 100.0; + } + return snapshot; +} + +inline std::vector PollingThreadPool::snapshotAllWorkers_() const +{ + std::vector snapshots; + for (const auto& group : worker_groups_) + { + for (const auto& worker : group.workers) + { + snapshots.push_back(snapshotThread_(worker.get())); + } + } + return snapshots; +} + +inline PollingThread* PollingThreadPool::findStealDestination_(const ThreadLoadSnapshot& source, + const std::vector& snapshots) const +{ + const auto source_group = findWorkerGroupForThread_(source.thread); + if (!source_group) + { + return nullptr; + } + + PollingThread* best = nullptr; + double best_idle = 1.0; + for (const auto& snap : snapshots) + { + if (snap.thread == source.thread) + { + continue; + } + const auto dest_group = findWorkerGroupForThread_(snap.thread); + if (!dest_group || dest_group->interval_ms != source_group->interval_ms) + { + continue; + } + if (snap.recent_busy_ratio < config_.steal_idle_threshold && snap.recent_busy_ratio < best_idle) + { + best_idle = snap.recent_busy_ratio; + best = snap.thread; + } + } + return best; +} + +inline Runnable* PollingThreadPool::findStealCandidate_(PollingThread* source_thread) const +{ + Runnable* best = nullptr; + double best_busy_ratio = -1.0; + size_t best_global_order = 0; + + for (const auto& reg : registrations_) + { + if (reg.assigned_thread != source_thread) + { + continue; + } + + const auto poll_metrics = source_thread->getRunnablePollMetrics(reg.runnable); + const uint64_t total_polls = poll_metrics.proceed_count + poll_metrics.sleep_count; + if (total_polls > 0) + { + const double busy_ratio = + static_cast(poll_metrics.proceed_count) / static_cast(total_polls); + if (busy_ratio > best_busy_ratio) + { + best_busy_ratio = busy_ratio; + best = reg.runnable; + } + continue; + } + + if (!best || reg.global_order > best_global_order) + { + best = reg.runnable; + best_global_order = reg.global_order; + } + } + + return best; +} + +inline void PollingThreadPool::migrateRunnable_(Runnable* runnable, PollingThread* from, PollingThread* to) +{ + from->pause(); + if (!from->removeRunnableWhilePaused(runnable)) + { + from->resume(); + throw DBException("Failed to remove runnable during migration"); + } + to->pause(); + to->addRunnableWhilePaused(runnable); + ensureRelativeOrderOnThread_(to); + to->resume(); + from->resume(); + + for (auto& reg : registrations_) + { + if (reg.runnable == runnable) + { + reg.assigned_thread = to; + ++reg.migration_count; + break; + } + } + + ++num_migrations_; +} + +inline void PollingThreadPool::growPool_(size_t interval_ms) +{ + auto worker_count = totalWorkerCount_(); + if (worker_count >= config_.max_threads) + { + return; + } + + auto group = findWorkerGroupForInterval_(interval_ms); + if (!group) + { + return; + } + + auto worker = createWorker_(interval_ms); + worker->open(); + group->workers.emplace_back(std::move(worker)); + + ++num_grows_; + peak_worker_count_ = std::max(peak_worker_count_, worker_count); +} + +inline void PollingThreadPool::shrinkPool_(const std::vector& snapshots) +{ + if (totalWorkerCount_() <= config_.min_threads) + { + return; + } + + for (const auto& snap : snapshots) + { + if (snap.recent_busy_ratio >= config_.shrink_idle_threshold) + { + continue; + } + if (snap.num_runnables > config_.shrink_max_runnables) + { + continue; + } + + PollingThread* thread = snap.thread; + IntervalWorkerGroup* group = nullptr; + for (auto& candidate_group : worker_groups_) + { + for (const auto& worker : candidate_group.workers) + { + if (worker.get() == thread) + { + group = &candidate_group; + break; + } + } + if (group) + { + break; + } + } + if (!group) + { + continue; + } + + if (snap.num_runnables > 0) + { + if (group->workers.size() <= 1) + { + continue; + } + + PollingThread* dest = nullptr; + double best_idle = 1.0; + for (const auto& other_snap : snapshots) + { + if (other_snap.thread == thread) + { + continue; + } + const auto other_group = findWorkerGroupForThread_(other_snap.thread); + if (!other_group || other_group->interval_ms != group->interval_ms) + { + continue; + } + if (other_snap.recent_busy_ratio < best_idle) + { + best_idle = other_snap.recent_busy_ratio; + dest = other_snap.thread; + } + } + if (!dest) + { + continue; + } + + std::vector to_migrate; + for (const auto& reg : registrations_) + { + if (reg.assigned_thread == thread) + { + to_migrate.push_back(reg.runnable); + } + } + for (Runnable* runnable : to_migrate) + { + migrateRunnable_(runnable, thread, dest); + } + } + + thread->close(); + for (auto it = group->workers.begin(); it != group->workers.end(); ++it) + { + if (it->get() == thread) + { + group->workers.erase(it); + break; + } + } + ++num_shrinks_; + return; + } +} + +inline void PollingThreadPool::accumulatePollMetricsFromWorkers_() +{ + for (const auto& group : worker_groups_) + { + for (const auto& worker : group.workers) + { + for (Runnable* runnable : worker->getRunnables()) + { + const auto window = worker->getRunnablePollMetrics(runnable); + if (window.proceed_count == 0 && window.sleep_count == 0) + { + continue; + } + + for (auto& reg : registrations_) + { + if (reg.runnable == runnable) + { + reg.lifetime_proceed_count += window.proceed_count; + reg.lifetime_sleep_count += window.sleep_count; + break; + } + } + } + } + } +} + +inline void PollingThreadPool::resetWorkerPollMetrics_() +{ + accumulatePollMetricsFromWorkers_(); + for (auto& group : worker_groups_) + { + for (auto& worker : group.workers) + { + worker->resetPollMetrics(); + } + } +} + +inline double PollingThreadPool::runnableProceedPct_(const RunnableRegistration& reg) const +{ + uint64_t proceed = reg.lifetime_proceed_count; + uint64_t sleep = reg.lifetime_sleep_count; + + PollingThread* worker = reg.assigned_thread; + if (!worker) + { + for (const auto& group : worker_groups_) + { + for (const auto& candidate : group.workers) + { + const auto& runnables = candidate->getRunnables(); + if (std::find(runnables.begin(), runnables.end(), reg.runnable) != runnables.end()) + { + worker = candidate.get(); + break; + } + } + if (worker) + { + break; + } + } + } + + if (worker) + { + const auto window = worker->getRunnablePollMetrics(reg.runnable); + proceed += window.proceed_count; + sleep += window.sleep_count; + } + + const uint64_t total = proceed + sleep; + if (total == 0) + { + return -1.0; + } + return (static_cast(proceed) / static_cast(total)) * 100.0; +} + +inline double PollingThreadPool::workerProceedPct_(PollingThread* worker) const +{ + if (!worker) + { + return -1.0; + } + + uint64_t proceed = 0; + uint64_t sleep = 0; + for (Runnable* runnable : worker->getRunnables()) + { + for (const auto& reg : registrations_) + { + if (reg.runnable != runnable) + { + continue; + } + proceed += reg.lifetime_proceed_count; + sleep += reg.lifetime_sleep_count; + break; + } + + const auto window = worker->getRunnablePollMetrics(runnable); + proceed += window.proceed_count; + sleep += window.sleep_count; + } + + const uint64_t total = proceed + sleep; + if (total == 0) + { + return -1.0; + } + return (static_cast(proceed) / static_cast(total)) * 100.0; +} + +inline std::string PollingThreadPool::formatWorkerLabel_(PollingThread* thread) const +{ + for (const auto& group : worker_groups_) + { + size_t worker_idx = 0; + for (const auto& worker : group.workers) + { + if (worker.get() == thread) + { + return "[" + std::to_string(group.interval_ms) + "ms] #" + std::to_string(worker_idx); + } + ++worker_idx; + } + } + return "unknown"; +} + +inline void PollingThreadPool::printPerfReport(std::ostream& os) +{ + std::lock_guard lock(pool_mutex_); + + if (initial_worker_count_ == 0 && worker_groups_.empty()) + { + return; + } + + std::vector paused_workers; + for (auto& group : worker_groups_) + { + for (auto& worker : group.workers) + { + const auto metrics = worker->getMetrics(); + if (metrics.is_running && !worker->paused()) + { + worker->pause(); + paused_workers.push_back(worker.get()); + } + } + } + + const auto resume_workers = [&paused_workers]() { + for (auto worker : paused_workers) + { + worker->resume(); + } + }; + + const size_t final_worker_count = totalWorkerCount_(); + + [[maybe_unused]] ios_format_saver fmt_saver(os); + os << "PollingThreadPool performance report\n\n"; + + os << " Summary:\n"; + os << " Workers: " << initial_worker_count_ << " -> peak " << peak_worker_count_ << " -> final " + << final_worker_count << " (min=" << config_.min_threads << " max=" << config_.max_threads << ")\n"; + os << " Registered runnables: " << registrations_.size() << "\n"; + os << " Rebalance cycles: " << num_rebalances_ << "\n"; + os << " Steals: " << num_steals_ << "\n"; + os << " Grows: " << num_grows_ << "\n"; + os << " Shrinks: " << num_shrinks_ << "\n"; + os << " Runnable migrations: " << num_migrations_ << "\n"; + os << " Grow blocked (max): " << grow_blocked_at_max_ << "\n"; + + os << "\n Final worker layout:\n"; + for (const auto& group : worker_groups_) + { + size_t worker_idx = 0; + for (const auto& worker : group.workers) + { + os << " Worker [" << group.interval_ms << "ms] #" << worker_idx << ": "; + + const auto& runnables = worker->getRunnables(); + if (runnables.empty()) + { + os << "(empty)"; + } else + { + for (size_t i = 0; i < runnables.size(); ++i) + { + if (i > 0) + { + os << ", "; + } + os << runnables[i]->getDescription(); + } + } + + const auto snap = snapshotThread_(worker.get()); + const double worker_proceed_pct = workerProceedPct_(worker.get()); + if (worker_proceed_pct >= 0.0) + { + os << " (" << std::fixed << std::setprecision(1) << worker_proceed_pct << "% proceed polls)\n"; + } else if (snap.pct_time_working > 0.0) + { + os << " (" << std::fixed << std::setprecision(1) << snap.pct_time_working << "% working)\n"; + } else + { + os << " (n/a)\n"; + } + ++worker_idx; + } + } + + std::vector sorted_regs; + sorted_regs.reserve(registrations_.size()); + for (const auto& reg : registrations_) + { + sorted_regs.push_back(®); + } + + std::sort(sorted_regs.begin(), sorted_regs.end(), [](const RunnableRegistration* a, const RunnableRegistration* b) { + return a->global_order < b->global_order; + }); + + os << "\n Runnables:\n"; + accumulatePollMetricsFromWorkers_(); + for (auto& group : worker_groups_) + { + for (auto& worker : group.workers) + { + worker->resetPollMetrics(); + } + } + for (const RunnableRegistration* reg : sorted_regs) + { + PollingThread* worker = reg->assigned_thread; + if (!worker) + { + for (const auto& group : worker_groups_) + { + for (const auto& candidate : group.workers) + { + const auto& runnables = candidate->getRunnables(); + if (std::find(runnables.begin(), runnables.end(), reg->runnable) != runnables.end()) + { + worker = candidate.get(); + break; + } + } + if (worker) + { + break; + } + } + } + + const double proceed_pct = runnableProceedPct_(*reg); + os << " " << reg->runnable->getDescription() << ": "; + if (proceed_pct >= 0.0) + { + os << std::fixed << std::setprecision(1) << proceed_pct << "% proceed polls"; + } else + { + os << "n/a proceed polls"; + } + + os << ", final worker " << formatWorkerLabel_(worker) << ", migrated " << reg->migration_count << " times\n"; + } + + os << "\n Configuration:\n"; + os << " rebalance_period_ms: " << config_.rebalance_period_ms.count() << "ms\n"; + os << " steal_busy_threshold: " << config_.steal_busy_threshold << "\n"; + os << " steal_idle_threshold: " << config_.steal_idle_threshold << "\n"; + os << " grow_all_busy_threshold: " << config_.grow_all_busy_threshold << "\n"; + os << " shrink_idle_threshold: " << config_.shrink_idle_threshold << "\n"; + os << " shrink_max_runnables: " << config_.shrink_max_runnables << "\n"; + os << "\n"; + + resume_workers(); +} + +inline std::unique_ptr PollingThreadPool::createWorker_(size_t interval_ms) +{ + return std::make_unique(interval_ms); +} + +} // namespace simdb::pipeline diff --git a/include/simdb/pipeline/Stage.hpp b/include/simdb/pipeline/Stage.hpp index a45d33c0..d90cbc46 100644 --- a/include/simdb/pipeline/Stage.hpp +++ b/include/simdb/pipeline/Stage.hpp @@ -5,7 +5,7 @@ #include "simdb/Exceptions.hpp" #include "simdb/pipeline/DatabaseAccessor.hpp" #include "simdb/pipeline/DatabaseThread.hpp" -#include "simdb/pipeline/PollingThread.hpp" +#include "simdb/pipeline/PollingThreadPool.hpp" #include "simdb/pipeline/QueueRepo.hpp" #include "simdb/pipeline/Runnable.hpp" #include @@ -25,8 +25,8 @@ class Stage : public Runnable { protected: /// \brief Construct with the polling interval (ms) for the thread when no work is done. - /// \param interval_milliseconds Sleep time for the PollingThread; non-database stages - /// that share a thread must use the same interval. + /// \param interval_milliseconds Sleep time for the PollingThread; stages sharing a + /// pool worker must use the same interval (workers are grouped by interval). Stage(size_t interval_milliseconds = 100) : interval_milliseconds_(interval_milliseconds) { @@ -53,11 +53,10 @@ class Stage : public Runnable void mergeQueueRepo_(PipelineQueueRepo& master_repo) { master_repo.merge(queue_repo_); } - virtual void assignThread_(DatabaseManager*, std::vector>& threads, - std::unique_ptr&) + virtual void assignThread_(DatabaseManager*, PollingThreadPool& pool, std::unique_ptr&, + size_t& global_order) { - threads.emplace_back(std::make_unique(interval_milliseconds_)); - threads.back()->addRunnable(this); + pool.registerRunnable(this, interval_milliseconds_, global_order++); } void setAsyncDatabaseAccessor_(AsyncDatabaseAccessor* async_db_accessor) { async_db_accessor_ = async_db_accessor; } @@ -143,9 +142,10 @@ template class DatabaseStage : public DatabaseStageBase } private: - void assignThread_(DatabaseManager* db_mgr, std::vector>&, - std::unique_ptr& database_thread) override final + void assignThread_(DatabaseManager* db_mgr, PollingThreadPool&, std::unique_ptr& database_thread, + size_t& global_order) override final { + (void)global_order; // Prepare the DatabaseAccessor db_accessor_ = std::make_unique(db_mgr); diff --git a/include/simdb/pipeline/ThreadMerger.hpp b/include/simdb/pipeline/ThreadMerger.hpp deleted file mode 100644 index 195c9f9b..00000000 --- a/include/simdb/pipeline/ThreadMerger.hpp +++ /dev/null @@ -1,336 +0,0 @@ -// -*- C++ -*- - -#pragma once - -#include "simdb/Exceptions.hpp" -#include "simdb/pipeline/DatabaseThread.hpp" -#include "simdb/pipeline/Pipeline.hpp" -#include "simdb/pipeline/PollingThread.hpp" - -#include -#include -#include - -namespace simdb::pipeline { - -/*! - * \class ThreadMerger - * - * \brief Merges non-database PollingThreads across pipelines when - * PipelineManager::minimizeThreads() or minimizeThreads(app,...) was - * called before openPipelines(). Creates a minimal set of threads. - * Exactly one DatabaseThread is shared by all pipeline(s) DatabaseStages - * if any. - */ -class ThreadMerger -{ -public: - /// \brief Construct with the pipelines that will contribute stages. - /// \param pipelines All pipelines (from PipelineManager). - ThreadMerger(const std::vector>& pipelines) : - pipelines_(pipelines) - { - } - - /// \brief Mark one app's non-database stages for merging; cannot mix with mergeAllAppThreads(). - /// \param app App whose pipeline threads should be merged. - /// \throws DBException if already finalized or if mergeAllAppThreads() was used. - void addAppForMerging(const App* app) - { - if (!accepting_apps_) - { - throw DBException("ThreadMerger already finalized merge"); - } - - if (merge_all_apps_) - { - throw DBException("Can either call addAppForMerging() or " - "mergeAllAppThreads(), not both"); - } - - auto it = std::find(apps_to_merge_.begin(), apps_to_merge_.end(), app); - if (it != apps_to_merge_.end()) - { - throw DBException("Already tracking app for thread merging"); - } - apps_to_merge_.push_back(app); - } - - /// \brief Merge all apps' non-database threads into a minimal set; cannot mix with addAppForMerging(). - void mergeAllAppThreads() { merge_all_apps_ = true; } - - /// \brief Build the final list of PollingThreads (and one DatabaseThread); call once from - /// PipelineManager::openPipelines(). - /// \param polling_threads Output vector; merged threads are appended (caller takes ownership). - void performMerge(std::vector>& polling_threads) - { - if (apps_to_merge_.empty() && !merge_all_apps_) - { - createThreadsWithoutMerging_(polling_threads); - } else if (apps_to_merge_.empty() || apps_to_merge_.size() == pipelines_.size()) - { - // TODO cnyce: I think this logic breaks down for apps that have - // more than one pipeline. - createThreadsAndMergeAll_(polling_threads); - } else - { - createThreadsAndMergeSpecificApps_(polling_threads); - } - - // Sanity check that we did not create more than one database thread. - size_t num_db_threads = 0; - for (const auto& thread : polling_threads) - { - if (dynamic_cast(thread.get())) - { - ++num_db_threads; - } - } - - if (num_db_threads > 1) - { - throw DBException("Internal error - we ended up creating ") - << num_db_threads << " database threads! Only one is allowed."; - } - - // Sanity check that every runnable (stage) exists only on one thread. - std::map> threads_by_runnable; - for (auto& thread : polling_threads) - { - for (auto runnable : thread->getRunnables()) - { - threads_by_runnable[runnable].push_back(thread.get()); - } - } - - for (const auto& [runnable, threads] : threads_by_runnable) - { - if (threads.size() != 1) - { - throw DBException("Internal error - assigned pipeline stage to " - "more than one thread"); - } - } - - // Reorder stages in polling threads to ensure that the relative - // ordering of stages is the same in the threads as they were originally - // defined in the app createPipeline() method. - std::vector ordered_runnables; - for (auto& pipeline : pipelines_) - { - for (auto& [stage_name, stage] : pipeline->getOrderedStages()) - { - ordered_runnables.push_back(stage); - } - } - - for (auto& thread : polling_threads) - { - thread->ensureRelativeOrder(ordered_runnables); - } - - accepting_apps_ = false; - } - -private: - void createThreadsWithoutMerging_(std::vector>& polling_threads) - { - std::unique_ptr database_thread; - for (auto& pipeline : pipelines_) - { - pipeline->assignStageThreads(polling_threads, database_thread); - } - - // Add the dedicated database thread - if (database_thread) - { - polling_threads.emplace_back(std::move(database_thread)); - } - } - - void createThreadsAndMergeAll_(std::vector>& polling_threads) - { - // Create all threads for each app before merge. - std::map>> polling_threads_by_app; - std::map pipelines_by_app; - std::unique_ptr database_thread; - for (auto& pipeline : pipelines_) - { - auto& app_polling_threads = polling_threads_by_app[pipeline->getOwningApp()]; - pipeline->assignStageThreads(app_polling_threads, database_thread); - pipelines_by_app[pipeline->getOwningApp()] = pipeline.get(); - } - - // Verify that all non-database stages are using the same polling - // interval. - std::set intervals; - for (const auto& [app, app_polling_threads] : polling_threads_by_app) - { - for (const auto& thread : app_polling_threads) - { - intervals.insert(thread->getIntervalMilliseconds()); - } - } - - if (intervals.size() != 1) - { - throw DBException("In order to merge threads, all must agree on " - "their polling interval."); - } - - // Find the maximum number of polling threads needed for all apps. - size_t max_polling_threads = 0; - for (const auto& [app, app_polling_threads] : polling_threads_by_app) - { - max_polling_threads = std::max(max_polling_threads, app_polling_threads.size()); - } - - // Create the max number of polling threads needed for all apps. - while (max_polling_threads--) - { - polling_threads.emplace_back(std::make_unique()); - } - - // Reassign all the polling_threads_by_app stages (runnables) to the - // merged thread list in a round-robin fashion. - size_t thread_idx = 0; - for (const auto& [app, app_polling_threads] : polling_threads_by_app) - { - auto app_pipeline = pipelines_by_app.at(app); - for (const auto& [name, stage] : app_pipeline->getOrderedStages()) - { - // Remember we are only merging non-database stages - if (dynamic_cast(stage)) - { - continue; - } - - auto& dest_thread = polling_threads.at(thread_idx); - dest_thread->addRunnable(stage); - - ++thread_idx; - if (thread_idx == polling_threads.size()) - { - thread_idx = 0; - } - } - } - - // Add the dedicated database thread - if (database_thread) - { - polling_threads.emplace_back(std::move(database_thread)); - } - } - - void createThreadsAndMergeSpecificApps_(std::vector>& polling_threads) - { - // TODO cnyce: This logic is not tested well enough to enable this - // feature yet. - throw DBException("Method not yet supported"); - - // Create all threads for each app before merge. - std::map>> polling_threads_by_app; - std::map pipelines_by_app; - std::unique_ptr database_thread; - for (auto& pipeline : pipelines_) - { - auto& app_polling_threads = polling_threads_by_app[pipeline->getOwningApp()]; - pipeline->assignStageThreads(app_polling_threads, database_thread); - pipelines_by_app[pipeline->getOwningApp()] = pipeline.get(); - } - - // Find the maximum number of polling threads needed for all merged - // apps. - size_t max_polling_threads_merged = 0; - for (const auto& [app, app_polling_threads] : polling_threads_by_app) - { - auto it = std::find(apps_to_merge_.begin(), apps_to_merge_.end(), app); - auto merge = (it != apps_to_merge_.end()); - if (merge) - { - max_polling_threads_merged = std::max(max_polling_threads_merged, app_polling_threads.size()); - } - } - - // Add the number of polling threads needed for all unmerged apps. - size_t total_num_polling_threads = max_polling_threads_merged; - for (const auto& [app, app_polling_threads] : polling_threads_by_app) - { - auto it = std::find(apps_to_merge_.begin(), apps_to_merge_.end(), app); - auto merge = (it != apps_to_merge_.end()); - if (!merge) - { - total_num_polling_threads += app_polling_threads.size(); - } - } - - // Create all polling threads needed. - while (total_num_polling_threads--) - { - polling_threads.emplace_back(std::make_unique()); - } - - // Assign stages from unmerged apps into the polling threads. - size_t thread_idx = 0; - for (const auto& [app, app_polling_threads] : polling_threads_by_app) - { - auto it = std::find(apps_to_merge_.begin(), apps_to_merge_.end(), app); - auto merge = (it != apps_to_merge_.end()); - if (!merge) - { - auto app_pipeline = pipelines_by_app.at(app); - for (const auto& [name, stage] : app_pipeline->getOrderedStages()) - { - // We are only merging threads for non-DB stages - if (dynamic_cast(stage)) - { - continue; - } - - auto& dest_thread = polling_threads.at(thread_idx); - dest_thread->addRunnable(stage); - - ++thread_idx; - if (thread_idx == polling_threads.size()) - { - thread_idx = 0; - } - } - } - } - - // Perform merge - for (auto app : apps_to_merge_) - { - auto app_pipeline = pipelines_by_app.at(app); - for (const auto& [name, stage] : app_pipeline->getOrderedStages()) - { - // Instead of a blind round-robin, sort the polling threads such - // that the one with the fewest assigned stages gets the next - // stage. - std::sort( - polling_threads.begin(), polling_threads.end(), - [](const std::unique_ptr& thread1, const std::unique_ptr& thread2) { - return thread1->getNumRunnables() < thread2->getNumRunnables(); - }); - - auto& dest_thread = polling_threads.front(); - dest_thread->addRunnable(stage); - } - } - - // Add the dedicated database thread - if (database_thread) - { - polling_threads.emplace_back(std::move(database_thread)); - } - } - - const std::vector>& pipelines_; - std::vector apps_to_merge_; - bool accepting_apps_ = true; - bool merge_all_apps_ = false; -}; - -} // namespace simdb::pipeline