Automatic load balancing with dynamic thread pool

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 <typename... Apps> 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<Apps>(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())

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
     {

include/simdb/pipeline/Pipeline.hpp

@@ -135,11 +135,12 @@ class Pipeline
         return queue_repo_.getOutPortQueue<T>(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<std::unique_ptr<PollingThread>>& threads,
-                            std::unique_ptr<DatabaseThread>& database_thread)
+    /// \param global_order Running index across all pipelines for pool runnable ordering.
+    void assignStageThreads(PollingThreadPool& pool, std::unique_ptr<DatabaseThread>& 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;

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 <iostream>
 
@@ -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<PipelineSnooper<KeyType, SnoopedType>>(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<ThreadMerger>(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 <typename... Apps> void minimizeThreads(const App* app, Apps&&... rest)
-    {
-        if (!thread_merger_)
-        {
-            thread_merger_ = std::make_unique<ThreadMerger>(pipelines_);
-        }
-        thread_merger_->addAppForMerging(app);
-        minimizeThreads(std::forward<Apps>(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<ThreadMerger>(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<DatabaseThread*>(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<DatabaseThread*>(thread.get()))
+                if (auto stage = dynamic_cast<Stage*>(runnable))
                 {
-                    for (auto runnable : thread->getRunnables())
-                    {
-                        if (auto stage = dynamic_cast<Stage*>(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,37 +142,35 @@ 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;
         do
         {
             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<std::unique_ptr<Pipeline>> pipelines_;
 
-    /// Instantiated threads.
-    std::vector<std::unique_ptr<PollingThread>> polling_threads_;
+    /// Pool of worker threads for non-database stages.
+    PollingThreadPool thread_pool_;
+
+    /// Dedicated database thread (never part of the pool).
+    std::unique_ptr<DatabaseThread> database_thread_;
 
     /// Threads that we give to the ScopedRunnableDisabler.
     std::vector<PollingThread*> disabler_threads_;
@@ -250,20 +204,17 @@ 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<ThreadMerger> thread_merger_;
-
     void getDisablerThreads_()
     {
         if (!disabler_threads_.empty())
         {
             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