GCC Code Coverage Report

include/simdb/
Coverage:
low: ≥ 0%
medium: ≥ 75.0%
high: ≥ 90.0%
Lines:
94 / 98
95.9%
Branches:
64 / 126
50.8%
Functions:
14 / 14
100.0%
pipeline/PipelineManager.hpp
Line Branch Exec Source
1 // <PipelineManager.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/PipelineSnooper.hpp"
9 #include "simdb/pipeline/PollingThread.hpp"
10 #include "simdb/pipeline/ThreadMerger.hpp"
11
12 #include <iostream>
13
14 namespace simdb {
15 class App;
16 }
17
18 namespace simdb::pipeline {
19
20 /*!
21 * \class PipelineManager
22 *
23 * \brief Manages all Pipeline instances and their PollingThreads for an
24 * AppManager (or unit test). Creates pipelines, merges threads
25 * (minimizeThreads), opens threads, and provides async DB access.
26 */
27 class PipelineManager
28 {
29 public:
30 /// \brief Construct with the DatabaseManager to be used by the pipelines.
31 /// \param db_mgr Non-null DatabaseManager to be used by the pipelines.
32 9 PipelineManager(DatabaseManager* db_mgr) :
33 9 db_mgr_(db_mgr)
34 {
35 9 }
36
37 /// \brief Return the AsyncDatabaseAccessor for async DB work; only valid after openPipelines().
38 /// \throws DBException if called before openPipelines().
39 AsyncDatabaseAccessor* getAsyncDatabaseAccessor()
40 {
41 checkOpen_();
42 simdb_assert(!threads_opened_, "Cannot access the AsyncDatabaseAccessor before "
43 "calling openPipelines()");
44 return async_db_accessor_;
45 }
46
47 /// \brief Create and own a new Pipeline with the given name and owning App.
48 /// \param name Pipeline name. Only used for reporting purposes.
49 /// \param app The App that owns this pipeline.
50 /// \return Raw pointer to the new Pipeline (manager retains ownership).
51 13 Pipeline* createPipeline(const std::string& name, const App* app)
52 {
53
1/1
✓ Branch 2 → 3 taken 13 times.
 13 checkOpen_();
54
1/1
✓ Branch 3 → 4 taken 13 times.
 13 auto pipeline = std::make_unique<Pipeline>(db_mgr_, name, app);
55
1/1
✓ Branch 5 → 6 taken 13 times.
 13 pipelines_.emplace_back(std::move(pipeline));
56 26 return pipelines_.back().get();
57 13 }
58
59 /// \brief Return pointers to all created pipelines.
60 9 std::vector<Pipeline*> getPipelines()
61 {
62 9 checkOpen_();
63
64 9 std::vector<Pipeline*> pipelines;
65
2/2
✓ Branch 11 → 6 taken 13 times.
✓ Branch 11 → 12 taken 9 times.
 22 for (auto& pipeline : pipelines_)
66 {
67
1/1
✓ Branch 8 → 9 taken 13 times.
 13 pipelines.push_back(pipeline.get());
68 }
69 9 return pipelines;
70 ✗ }
71
72 /// \brief Create a snooper for iterating stages with a key and snooped object type.
73 /// \return Unique_ptr to a PipelineSnooper<KeyType, SnoopedType>.
74 template <typename KeyType, typename SnoopedType>
75 1 std::unique_ptr<PipelineSnooper<KeyType, SnoopedType>> createSnooper()
76 {
77
1/1
✓ Branch 2 → 3 taken 1 time.
 1 return std::make_unique<PipelineSnooper<KeyType, SnoopedType>>(this);
78 }
79
80 /// \brief Merge all apps' pipeline threads into a minimal set; call at most once.
81 /// \throws DBException if called more than once.
82 void minimizeThreads()
83 {
84 simdb_assert(!thread_merger_, "You can only call minimizeThreads() method once.");
85
86 thread_merger_ = std::make_unique<ThreadMerger>(pipelines_);
87 thread_merger_->mergeAllAppThreads();
88 }
89
90 /// \brief Mark one app's pipeline threads for merging (call before openPipelines()).
91 void minimizeThreads(const App* app)
92 {
93 simdb_assert(thread_merger_, "Cannot merge a single app's pipeline threads");
94 thread_merger_->addAppForMerging(app);
95 }
96
97 /// \brief Mark multiple apps' pipeline threads for merging (variadic).
98 template <typename... Apps> void minimizeThreads(const App* app, Apps&&... rest)
99 {
100 if (!thread_merger_)
101 {
102 thread_merger_ = std::make_unique<ThreadMerger>(pipelines_);
103 }
104 thread_merger_->addAppForMerging(app);
105 minimizeThreads(std::forward<Apps>(rest)...);
106 }
107
108 /// \brief Create and open all polling threads (after stages are added and optionally minimizeThreads).
109 9 void openPipelines()
110 {
111 9 checkOpen_();
112
113
1/2
✓ Branch 4 → 5 taken 9 times.
✗ Branch 4 → 9 not taken.
 9 if (!thread_merger_)
114 {
115
1/1
✓ Branch 5 → 6 taken 9 times.
 9 thread_merger_ = std::make_unique<ThreadMerger>(pipelines_);
116 }
117 9 thread_merger_->performMerge(polling_threads_);
118
119 // Now that all threads are created, give the async DB accessor to all
120 // non-DB stages in all pipelines.
121
1/2
✓ Branch 23 → 13 taken 29 times.
✗ Branch 23 → 24 not taken.
 29 for (auto& thread : polling_threads_)
122 {
123
3/4
✓ Branch 15 → 16 taken 29 times.
✗ Branch 15 → 17 not taken.
✓ Branch 18 → 19 taken 9 times.
✓ Branch 18 → 21 taken 20 times.
 29 if (auto db_thread = dynamic_cast<DatabaseThread*>(thread.get()))
124 {
125 9 async_db_accessor_ = db_thread->getAsyncDatabaseAccessor();
126 9 break;
127 }
128 }
129
130
1/2
✓ Branch 24 → 25 taken 9 times.
✗ Branch 24 → 51 not taken.
 9 if (async_db_accessor_)
131 {
132
2/2
✓ Branch 49 → 27 taken 29 times.
✓ Branch 49 → 50 taken 9 times.
 38 for (auto& thread : polling_threads_)
133 {
134
3/4
✓ Branch 29 → 30 taken 29 times.
✗ Branch 29 → 31 not taken.
✓ Branch 32 → 33 taken 20 times.
✓ Branch 32 → 47 taken 9 times.
 29 if (!dynamic_cast<DatabaseThread*>(thread.get()))
135 {
136
2/2
✓ Branch 45 → 37 taken 20 times.
✓ Branch 45 → 46 taken 20 times.
 40 for (auto runnable : thread->getRunnables())
137 {
138
2/4
✓ Branch 38 → 39 taken 20 times.
✗ Branch 38 → 40 not taken.
✓ Branch 41 → 42 taken 20 times.
✗ Branch 41 → 43 not taken.
 20 if (auto stage = dynamic_cast<Stage*>(runnable))
139 {
140 20 stage->setAsyncDatabaseAccessor_(async_db_accessor_);
141 }
142 }
143 }
144 }
145 }
146
147 // Now open all threads for simulation
148
2/2
✓ Branch 58 → 53 taken 29 times.
✓ Branch 58 → 59 taken 9 times.
 38 for (auto& thread : polling_threads_)
149 {
150
1/1
✓ Branch 55 → 56 taken 29 times.
 29 thread->open();
151 }
152 9 threads_opened_ = true;
153 9 }
154
155 /// \brief Temporarily disable all pipeline runnables (and optionally pause threads);
156 /// re-enabled when the returnedobject is destroyed.
157 /// \param disable_threads_too If true, also pause polling threads; if false, only
158 /// disable runnables.
159 /// \return A ScopedRunnableDisabler, or nullptr if a disabler is already active (nested
160 /// calls are no-ops).
161 10000 std::unique_ptr<ScopedRunnableDisabler> scopedDisableAll(bool disable_threads_too = true)
162 {
163
1/2
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 4 taken 10000 times.
 10000 if (disabler_active_)
164 {
165 ✗ return nullptr;
166 }
167
168
1/1
✓ Branch 4 → 5 taken 10000 times.
 10000 getDisablerRunnables_();
169
1/1
✓ Branch 5 → 6 taken 10000 times.
 10000 getDisablerThreads_();
170
171 10000 std::unique_ptr<ScopedRunnableDisabler> disabler;
172
1/2
✓ Branch 6 → 7 taken 10000 times.
✗ Branch 6 → 10 not taken.
 10000 if (disable_threads_too)
173 {
174
2/4
✓ Branch 7 → 8 taken 10000 times.
✓ Branch 8 → 9 taken 10000 times.
✗ Branch 17 → 18 not taken.
✗ Branch 17 → 19 not taken.
 10000 disabler.reset(new ScopedRunnableDisabler(this, disabler_runnables_, disabler_threads_));
175 } else
176 {
177
0/4
✗ Branch 10 → 11 not taken.
✗ Branch 11 → 12 not taken.
✗ Branch 20 → 21 not taken.
✗ Branch 20 → 22 not taken.
 ✗ disabler.reset(new ScopedRunnableDisabler(this, disabler_runnables_));
178 }
179
180 10000 disabler_active_ = true;
181 10000 return disabler;
182 10000 }
183
184 /// \brief Close all threads, flush runnables, and print performance reports.
185 9 void postSimLoopTeardown()
186 {
187
1/1
✓ Branch 2 → 3 taken 9 times.
 9 checkOpen_();
188
189 29 auto close_thread = [&](PollingThread* thread) {
190 29 thread->close();
191
1/2
✓ Branch 4 → 5 taken 29 times.
✗ Branch 4 → 6 not taken.
 29 if (thread->printPerfReport())
192 {
193 29 std::cout << "\n\n";
194 }
195 29 };
196
197 9 auto it = polling_threads_.begin();
198
2/2
✓ Branch 11 → 5 taken 29 times.
✓ Branch 11 → 12 taken 9 times.
 38 while (it != polling_threads_.end())
199 {
200
1/1
✓ Branch 7 → 8 taken 29 times.
 29 close_thread(it->get());
201 29 ++it;
202 }
203
204 bool continue_while;
205
2/2
✓ Branch 22 → 23 taken 6 times.
✓ Branch 22 → 24 taken 9 times.
 15 do
206 {
207 15 continue_while = false;
208
209 15 it = polling_threads_.begin();
210
2/2
✓ Branch 21 → 15 taken 49 times.
✓ Branch 21 → 22 taken 15 times.
 64 while (it != polling_threads_.end())
211 {
212
1/1
✓ Branch 17 → 18 taken 49 times.
 49 continue_while |= (*it)->flushRunnables();
213 49 ++it;
214 }
215 } while (continue_while);
216
217 9 closed_ = true;
218 9 }
219
220 private:
221 /// Associated DatabaseManager.
222 DatabaseManager* db_mgr_ = nullptr;
223
224 /// Instantiated pipelines.
225 std::vector<std::unique_ptr<Pipeline>> pipelines_;
226
227 /// Instantiated threads.
228 std::vector<std::unique_ptr<PollingThread>> polling_threads_;
229
230 /// Threads that we give to the ScopedRunnableDisabler.
231 std::vector<PollingThread*> disabler_threads_;
232
233 /// Runnables that we give to the ScopedRunnableDisabler.
234 std::vector<Runnable*> disabler_runnables_;
235
236 /// Flag saying whether a ScopedRunnableDisabler is active.
237 /// Used in order to short-circuit nested disablers.
238 bool disabler_active_ = false;
239
240 /// Flag used to prevent AsyncDatabaseAccessor from being
241 /// accessed until threads are opened/finalized.
242 bool threads_opened_ = false;
243
244 /// Cached AsyncDatabaseAccessor for async DB queries.
245 AsyncDatabaseAccessor* async_db_accessor_ = nullptr;
246
247 /// Used to perform minimizeThread() to share threads
248 /// between concurrently running apps.
249 std::unique_ptr<ThreadMerger> thread_merger_;
250
251 10000 void getDisablerThreads_()
252 {
253
2/2
✓ Branch 3 → 4 taken 9998 times.
✓ Branch 3 → 5 taken 2 times.
 10000 if (!disabler_threads_.empty())
254 {
255 9998 return;
256 }
257
258
2/2
✓ Branch 12 → 7 taken 7 times.
✓ Branch 12 → 13 taken 2 times.
 9 for (auto& thread : polling_threads_)
259 {
260
1/1
✓ Branch 9 → 10 taken 7 times.
 7 disabler_threads_.push_back(thread.get());
261 }
262
263 // Ensure unique
264
1/1
✓ Branch 15 → 16 taken 2 times.
 2 auto it = std::unique(disabler_threads_.begin(), disabler_threads_.end());
265
1/12
✗ Branch 18 → 19 not taken.
✓ Branch 18 → 37 taken 2 times.
✗ Branch 21 → 22 not taken.
✗ Branch 22 → 23 not taken.
✗ Branch 24 → 25 not taken.
✗ Branch 28 → 29 not taken.
✗ Branch 29 → 30 not taken.
✗ Branch 30 → 31 not taken.
✗ Branch 31 → 32 not taken.
✗ Branch 32 → 33 not taken.
✗ Branch 33 → 34 not taken.
✗ Branch 35 → 36 not taken.
 2 simdb_assert(it == disabler_threads_.end(), "Internal error: duplicate threads found in disabler_threads_");
266 }
267
268 10000 void getDisablerRunnables_()
269 {
270
2/2
✓ Branch 3 → 4 taken 9998 times.
✓ Branch 3 → 5 taken 2 times.
 10000 if (!disabler_runnables_.empty())
271 {
272 9998 return;
273 }
274
275
2/2
✓ Branch 17 → 7 taken 7 times.
✓ Branch 17 → 18 taken 2 times.
 9 for (auto& thread : polling_threads_)
276 {
277 7 const auto& runnables = thread->getRunnables();
278
1/1
✓ Branch 14 → 15 taken 7 times.
 7 disabler_runnables_.insert(disabler_runnables_.end(), runnables.begin(), runnables.end());
279 }
280
281 // Ensure unique
282
1/1
✓ Branch 20 → 21 taken 2 times.
 2 auto it = std::unique(disabler_runnables_.begin(), disabler_runnables_.end());
283
1/12
✗ Branch 23 → 24 not taken.
✓ Branch 23 → 42 taken 2 times.
✗ Branch 26 → 27 not taken.
✗ Branch 27 → 28 not taken.
✗ Branch 29 → 30 not taken.
✗ Branch 33 → 34 not taken.
✗ Branch 34 → 35 not taken.
✗ Branch 35 → 36 not taken.
✗ Branch 36 → 37 not taken.
✗ Branch 37 → 38 not taken.
✗ Branch 38 → 39 not taken.
✗ Branch 40 → 41 not taken.
 2 simdb_assert(it == disabler_runnables_.end(), "Internal error: duplicate runnables found in "
284 "disabler_runnables_");
285 }
286
287 /// Get a notification when a disabler goes out of scope.
288 friend class ScopedRunnableDisabler;
289 10000 void onDisablerDestruction_()
290 {
291
1/12
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 21 taken 10000 times.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 19 → 20 not taken.
 10000 simdb_assert(disabler_active_, "Internal error: no disabler active in "
292 "onDisablerDestruction_()");
293 10000 disabler_active_ = false;
294 10000 }
295
296 /// Flag saying whether postSimLoopTeardown() was called.
297 bool closed_ = false;
298
299 /// Validate that no APIs are called after closing the pipelines
300
1/12
✗ Branch 2 → 3 not taken.
✓ Branch 2 → 21 taken 40 times.
✗ Branch 5 → 6 not taken.
✗ Branch 6 → 7 not taken.
✗ Branch 8 → 9 not taken.
✗ Branch 12 → 13 not taken.
✗ Branch 13 → 14 not taken.
✗ Branch 14 → 15 not taken.
✗ Branch 15 → 16 not taken.
✗ Branch 16 → 17 not taken.
✗ Branch 17 → 18 not taken.
✗ Branch 19 → 20 not taken.
 40 void checkOpen_() const { simdb_assert(!closed_, "PipelineManager has been closed"); }
301 };
302
303 /// Defined here so we can avoid circular includes
304 10000 inline void ScopedRunnableDisabler::notifyPipelineMgrReenabled_()
305 {
306 10000 pipeline_mgr_->onDisablerDestruction_();
307 10000 }
308
309 /// Defined here so we can avoid circular includes
310 template <typename KeyType, typename SnoopedType>
311 9999 bool PipelineSnooper<KeyType, SnoopedType>::snoopAllStages(const KeyType& key, SnoopedType& snooped_obj,
312 bool disable_pipeline)
313 {
314
2/3
✓ Branch 2 → 3 taken 9999 times.
✗ Branch 2 → 4 not taken.
✓ Branch 3 → 5 taken 9999 times.
 9999 std::unique_ptr<ScopedRunnableDisabler> disabler = disable_pipeline ? pipeline_mgr_->scopedDisableAll() : nullptr;
315
316
1/2
✓ Branch 13 → 7 taken 39292 times.
✗ Branch 13 → 14 not taken.
 39292 for (auto& cb : callbacks_)
317 {
318
3/3
✓ Branch 8 → 9 taken 39292 times.
✓ Branch 9 → 10 taken 9999 times.
✓ Branch 9 → 11 taken 29293 times.
 39292 if (cb(key, snooped_obj))
319 {
320 9999 return true;
321 }
322 }
323 ✗ return false;
324 9999 }
325
326 } // namespace simdb::pipeline
327