pipeline/ThreadMerger.hpp
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | // <ThreadMerger.hpp> -*- C++ -*- | ||
| 2 | |||
| 3 | #pragma once | ||
| 4 | |||
| 5 | #include "simdb/Assert.hpp" | ||
| 6 | #include "simdb/pipeline/DatabaseThread.hpp" | ||
| 7 | #include "simdb/pipeline/Pipeline.hpp" | ||
| 8 | #include "simdb/pipeline/PollingThread.hpp" | ||
| 9 | |||
| 10 | #include <algorithm> | ||
| 11 | #include <memory> | ||
| 12 | #include <vector> | ||
| 13 | |||
| 14 | namespace simdb::pipeline { | ||
| 15 | |||
| 16 | /*! | ||
| 17 | * \class ThreadMerger | ||
| 18 | * | ||
| 19 | * \brief Merges non-database PollingThreads across pipelines when | ||
| 20 | * PipelineManager::minimizeThreads() or minimizeThreads(app,...) was | ||
| 21 | * called before openPipelines(). Creates a minimal set of threads. | ||
| 22 | * Exactly one DatabaseThread is shared by all pipeline(s) DatabaseStages | ||
| 23 | * if any. | ||
| 24 | */ | ||
| 25 | class ThreadMerger | ||
| 26 | { | ||
| 27 | public: | ||
| 28 | /// \brief Construct with the pipelines that will contribute stages. | ||
| 29 | /// \param pipelines All pipelines (from PipelineManager). | ||
| 30 | 9 | ThreadMerger(const std::vector<std::unique_ptr<Pipeline>>& pipelines) : | |
| 31 | 9 | pipelines_(pipelines) | |
| 32 | { | ||
| 33 | 9 | } | |
| 34 | |||
| 35 | /// \brief Mark one app's non-database stages for merging; cannot mix with mergeAllAppThreads(). | ||
| 36 | /// \param app App whose pipeline threads should be merged. | ||
| 37 | /// \throws DBException if already finalized or if mergeAllAppThreads() was used. | ||
| 38 | void addAppForMerging(const App* app) | ||
| 39 | { | ||
| 40 | simdb_assert(accepting_apps_, "ThreadMerger already finalized merge"); | ||
| 41 | |||
| 42 | simdb_assert(!merge_all_apps_, "Can either call addAppForMerging() or " | ||
| 43 | "mergeAllAppThreads(), not both"); | ||
| 44 | |||
| 45 | auto it = std::find(apps_to_merge_.begin(), apps_to_merge_.end(), app); | ||
| 46 | simdb_assert(it == apps_to_merge_.end(), "Already tracking app for thread merging"); | ||
| 47 | apps_to_merge_.push_back(app); | ||
| 48 | } | ||
| 49 | |||
| 50 | /// \brief Merge all apps' non-database threads into a minimal set; cannot mix with addAppForMerging(). | ||
| 51 | void mergeAllAppThreads() { merge_all_apps_ = true; } | ||
| 52 | |||
| 53 | /// \brief Build the final list of PollingThreads (and one DatabaseThread); call once from | ||
| 54 | /// PipelineManager::openPipelines(). | ||
| 55 | /// \param polling_threads Output vector; merged threads are appended (caller takes ownership). | ||
| 56 | 9 | void performMerge(std::vector<std::unique_ptr<PollingThread>>& polling_threads) | |
| 57 | { | ||
| 58 | 9 | if (apps_to_merge_.empty() && !merge_all_apps_) | |
| 59 | { | ||
| 60 | 9 | createThreadsWithoutMerging_(polling_threads); | |
| 61 | ✗ | } else if (apps_to_merge_.empty() || apps_to_merge_.size() == pipelines_.size()) | |
| 62 | { | ||
| 63 | // TODO cnyce: I think this logic breaks down for apps that have | ||
| 64 | // more than one pipeline. | ||
| 65 | ✗ | createThreadsAndMergeAll_(polling_threads); | |
| 66 | } else | ||
| 67 | { | ||
| 68 | ✗ | createThreadsAndMergeSpecificApps_(polling_threads); | |
| 69 | } | ||
| 70 | |||
| 71 | // Sanity check that we did not create more than one database thread. | ||
| 72 | 9 | size_t num_db_threads = 0; | |
| 73 | 38 | for (const auto& thread : polling_threads) | |
| 74 | { | ||
| 75 | 29 | if (dynamic_cast<const DatabaseThread*>(thread.get())) | |
| 76 | { | ||
| 77 | 9 | ++num_db_threads; | |
| 78 | } | ||
| 79 | } | ||
| 80 | |||
| 81 | 9 | simdb_assert(num_db_threads <= 1, "Internal error - we ended up creating " | |
| 82 | << num_db_threads << " database threads! Only one is allowed."); | ||
| 83 | |||
| 84 | // Sanity check that every runnable (stage) exists only on one thread. | ||
| 85 | 9 | std::map<Runnable*, std::vector<PollingThread*>> threads_by_runnable; | |
| 86 | 38 | for (auto& thread : polling_threads) | |
| 87 | { | ||
| 88 | 61 | for (auto runnable : thread->getRunnables()) | |
| 89 | { | ||
| 90 | 32 | threads_by_runnable[runnable].push_back(thread.get()); | |
| 91 | } | ||
| 92 | } | ||
| 93 | |||
| 94 | 41 | for (const auto& [runnable, threads] : threads_by_runnable) | |
| 95 | { | ||
| 96 | 32 | simdb_assert(threads.size() == 1, "Internal error - assigned pipeline stage to " | |
| 97 | "more than one thread"); | ||
| 98 | } | ||
| 99 | |||
| 100 | // Reorder stages in polling threads to ensure that the relative | ||
| 101 | // ordering of stages is the same in the threads as they were originally | ||
| 102 | // defined in the app createPipeline() method. | ||
| 103 | 9 | std::vector<Runnable*> ordered_runnables; | |
| 104 | 22 | for (auto& pipeline : pipelines_) | |
| 105 | { | ||
| 106 | 45 | for (auto& [stage_name, stage] : pipeline->getOrderedStages()) | |
| 107 | { | ||
| 108 | 32 | ordered_runnables.push_back(stage); | |
| 109 | 13 | } | |
| 110 | } | ||
| 111 | |||
| 112 | 38 | for (auto& thread : polling_threads) | |
| 113 | { | ||
| 114 | 29 | thread->ensureRelativeOrder(ordered_runnables); | |
| 115 | } | ||
| 116 | |||
| 117 | 9 | accepting_apps_ = false; | |
| 118 | 9 | } | |
| 119 | |||
| 120 | private: | ||
| 121 | 9 | void createThreadsWithoutMerging_(std::vector<std::unique_ptr<PollingThread>>& polling_threads) | |
| 122 | { | ||
| 123 | 9 | std::unique_ptr<DatabaseThread> database_thread; | |
| 124 | 22 | for (auto& pipeline : pipelines_) | |
| 125 | { | ||
| 126 | 13 | pipeline->assignStageThreads(polling_threads, database_thread); | |
| 127 | } | ||
| 128 | |||
| 129 | // Add the dedicated database thread | ||
| 130 | 9 | if (database_thread) | |
| 131 | { | ||
| 132 | 9 | polling_threads.emplace_back(std::move(database_thread)); | |
| 133 | } | ||
| 134 | 9 | } | |
| 135 | |||
| 136 | ✗ | void createThreadsAndMergeAll_(std::vector<std::unique_ptr<PollingThread>>& polling_threads) | |
| 137 | { | ||
| 138 | // Create all threads for each app before merge. | ||
| 139 | ✗ | std::map<const App*, std::vector<std::unique_ptr<PollingThread>>> polling_threads_by_app; | |
| 140 | ✗ | std::map<const App*, Pipeline*> pipelines_by_app; | |
| 141 | ✗ | std::unique_ptr<DatabaseThread> database_thread; | |
| 142 | ✗ | for (auto& pipeline : pipelines_) | |
| 143 | { | ||
| 144 | ✗ | auto& app_polling_threads = polling_threads_by_app[pipeline->getOwningApp()]; | |
| 145 | ✗ | pipeline->assignStageThreads(app_polling_threads, database_thread); | |
| 146 | ✗ | pipelines_by_app[pipeline->getOwningApp()] = pipeline.get(); | |
| 147 | } | ||
| 148 | |||
| 149 | // Verify that all non-database stages are using the same polling | ||
| 150 | // interval. | ||
| 151 | ✗ | std::set<size_t> intervals; | |
| 152 | ✗ | for (const auto& [app, app_polling_threads] : polling_threads_by_app) | |
| 153 | { | ||
| 154 | ✗ | for (const auto& thread : app_polling_threads) | |
| 155 | { | ||
| 156 | ✗ | intervals.insert(thread->getIntervalMilliseconds()); | |
| 157 | } | ||
| 158 | } | ||
| 159 | |||
| 160 | ✗ | simdb_assert(intervals.size() == 1, "In order to merge threads, all must agree on " | |
| 161 | "their polling interval."); | ||
| 162 | |||
| 163 | // Find the maximum number of polling threads needed for all apps. | ||
| 164 | ✗ | size_t max_polling_threads = 0; | |
| 165 | ✗ | for (const auto& [app, app_polling_threads] : polling_threads_by_app) | |
| 166 | { | ||
| 167 | ✗ | max_polling_threads = std::max(max_polling_threads, app_polling_threads.size()); | |
| 168 | } | ||
| 169 | |||
| 170 | // Create the max number of polling threads needed for all apps. | ||
| 171 | ✗ | while (max_polling_threads--) | |
| 172 | { | ||
| 173 | ✗ | polling_threads.emplace_back(std::make_unique<PollingThread>()); | |
| 174 | } | ||
| 175 | |||
| 176 | // Reassign all the polling_threads_by_app stages (runnables) to the | ||
| 177 | // merged thread list in a round-robin fashion. | ||
| 178 | ✗ | size_t thread_idx = 0; | |
| 179 | ✗ | for (const auto& [app, app_polling_threads] : polling_threads_by_app) | |
| 180 | { | ||
| 181 | ✗ | auto app_pipeline = pipelines_by_app.at(app); | |
| 182 | ✗ | for (const auto& [name, stage] : app_pipeline->getOrderedStages()) | |
| 183 | { | ||
| 184 | // Remember we are only merging non-database stages | ||
| 185 | ✗ | if (dynamic_cast<DatabaseStageBase*>(stage)) | |
| 186 | { | ||
| 187 | ✗ | continue; | |
| 188 | } | ||
| 189 | |||
| 190 | ✗ | auto& dest_thread = polling_threads.at(thread_idx); | |
| 191 | ✗ | dest_thread->addRunnable(stage); | |
| 192 | |||
| 193 | ✗ | ++thread_idx; | |
| 194 | ✗ | if (thread_idx == polling_threads.size()) | |
| 195 | { | ||
| 196 | ✗ | thread_idx = 0; | |
| 197 | } | ||
| 198 | ✗ | } | |
| 199 | } | ||
| 200 | |||
| 201 | // Add the dedicated database thread | ||
| 202 | ✗ | if (database_thread) | |
| 203 | { | ||
| 204 | ✗ | polling_threads.emplace_back(std::move(database_thread)); | |
| 205 | } | ||
| 206 | ✗ | } | |
| 207 | |||
| 208 | ✗ | void createThreadsAndMergeSpecificApps_(std::vector<std::unique_ptr<PollingThread>>& polling_threads) | |
| 209 | { | ||
| 210 | // TODO cnyce: This logic is not tested well enough to enable this | ||
| 211 | // feature yet. | ||
| 212 | ✗ | throw DBException("Method not yet supported"); | |
| 213 | |||
| 214 | // Create all threads for each app before merge. | ||
| 215 | std::map<const App*, std::vector<std::unique_ptr<PollingThread>>> polling_threads_by_app; | ||
| 216 | std::map<const App*, Pipeline*> pipelines_by_app; | ||
| 217 | std::unique_ptr<DatabaseThread> database_thread; | ||
| 218 | for (auto& pipeline : pipelines_) | ||
| 219 | { | ||
| 220 | auto& app_polling_threads = polling_threads_by_app[pipeline->getOwningApp()]; | ||
| 221 | pipeline->assignStageThreads(app_polling_threads, database_thread); | ||
| 222 | pipelines_by_app[pipeline->getOwningApp()] = pipeline.get(); | ||
| 223 | } | ||
| 224 | |||
| 225 | // Find the maximum number of polling threads needed for all merged | ||
| 226 | // apps. | ||
| 227 | size_t max_polling_threads_merged = 0; | ||
| 228 | for (const auto& [app, app_polling_threads] : polling_threads_by_app) | ||
| 229 | { | ||
| 230 | auto it = std::find(apps_to_merge_.begin(), apps_to_merge_.end(), app); | ||
| 231 | auto merge = (it != apps_to_merge_.end()); | ||
| 232 | if (merge) | ||
| 233 | { | ||
| 234 | max_polling_threads_merged = std::max(max_polling_threads_merged, app_polling_threads.size()); | ||
| 235 | } | ||
| 236 | } | ||
| 237 | |||
| 238 | // Add the number of polling threads needed for all unmerged apps. | ||
| 239 | size_t total_num_polling_threads = max_polling_threads_merged; | ||
| 240 | for (const auto& [app, app_polling_threads] : polling_threads_by_app) | ||
| 241 | { | ||
| 242 | auto it = std::find(apps_to_merge_.begin(), apps_to_merge_.end(), app); | ||
| 243 | auto merge = (it != apps_to_merge_.end()); | ||
| 244 | if (!merge) | ||
| 245 | { | ||
| 246 | total_num_polling_threads += app_polling_threads.size(); | ||
| 247 | } | ||
| 248 | } | ||
| 249 | |||
| 250 | // Create all polling threads needed. | ||
| 251 | while (total_num_polling_threads--) | ||
| 252 | { | ||
| 253 | polling_threads.emplace_back(std::make_unique<PollingThread>()); | ||
| 254 | } | ||
| 255 | |||
| 256 | // Assign stages from unmerged apps into the polling threads. | ||
| 257 | size_t thread_idx = 0; | ||
| 258 | for (const auto& [app, app_polling_threads] : polling_threads_by_app) | ||
| 259 | { | ||
| 260 | auto it = std::find(apps_to_merge_.begin(), apps_to_merge_.end(), app); | ||
| 261 | auto merge = (it != apps_to_merge_.end()); | ||
| 262 | if (!merge) | ||
| 263 | { | ||
| 264 | auto app_pipeline = pipelines_by_app.at(app); | ||
| 265 | for (const auto& [name, stage] : app_pipeline->getOrderedStages()) | ||
| 266 | { | ||
| 267 | // We are only merging threads for non-DB stages | ||
| 268 | if (dynamic_cast<DatabaseStageBase*>(stage)) | ||
| 269 | { | ||
| 270 | continue; | ||
| 271 | } | ||
| 272 | |||
| 273 | auto& dest_thread = polling_threads.at(thread_idx); | ||
| 274 | dest_thread->addRunnable(stage); | ||
| 275 | |||
| 276 | ++thread_idx; | ||
| 277 | if (thread_idx == polling_threads.size()) | ||
| 278 | { | ||
| 279 | thread_idx = 0; | ||
| 280 | } | ||
| 281 | } | ||
| 282 | } | ||
| 283 | } | ||
| 284 | |||
| 285 | // Perform merge | ||
| 286 | for (auto app : apps_to_merge_) | ||
| 287 | { | ||
| 288 | auto app_pipeline = pipelines_by_app.at(app); | ||
| 289 | for (const auto& [name, stage] : app_pipeline->getOrderedStages()) | ||
| 290 | { | ||
| 291 | // Instead of a blind round-robin, sort the polling threads such | ||
| 292 | // that the one with the fewest assigned stages gets the next | ||
| 293 | // stage. | ||
| 294 | std::sort( | ||
| 295 | polling_threads.begin(), polling_threads.end(), | ||
| 296 | [](const std::unique_ptr<PollingThread>& thread1, const std::unique_ptr<PollingThread>& thread2) { | ||
| 297 | return thread1->getNumRunnables() < thread2->getNumRunnables(); | ||
| 298 | }); | ||
| 299 | |||
| 300 | auto& dest_thread = polling_threads.front(); | ||
| 301 | dest_thread->addRunnable(stage); | ||
| 302 | } | ||
| 303 | } | ||
| 304 | |||
| 305 | // Add the dedicated database thread | ||
| 306 | if (database_thread) | ||
| 307 | { | ||
| 308 | polling_threads.emplace_back(std::move(database_thread)); | ||
| 309 | } | ||
| 310 | } | ||
| 311 | |||
| 312 | const std::vector<std::unique_ptr<Pipeline>>& pipelines_; | ||
| 313 | std::vector<const App*> apps_to_merge_; | ||
| 314 | bool accepting_apps_ = true; | ||
| 315 | bool merge_all_apps_ = false; | ||
| 316 | }; | ||
| 317 | |||
| 318 | } // namespace simdb::pipeline | ||
| 319 |