GCC Code Coverage Report

include/simdb/
Coverage:
low: ≥ 0%
medium: ≥ 75.0%
high: ≥ 90.0%
Lines:
136 / 152
89.5%
Branches:
84 / 117
71.8%
Functions:
18 / 20
90.0%
pipeline/PollingThread.hpp
Line Branch Exec Source
1 // <PollingThread.hpp> -*- C++ -*-
2
3 #pragma once
4
5 #include "simdb/Assert.hpp"
6 #include "simdb/pipeline/Runnable.hpp"
7 #include "simdb/utils/StreamFormatters.hpp"
8
9 #include <atomic>
10 #include <chrono>
11 #include <condition_variable>
12 #include <future>
13 #include <iomanip>
14 #include <iostream>
15 #include <mutex>
16 #include <thread>
17 #include <vector>
18
19 namespace simdb::pipeline {
20
21 /*!
22 * \class PollingThread
23 *
24 * \brief Thread that repeatedly polls its Runnables for work; when none do
25 * work, it sleeps for a fixed interval before polling again. Supports
26 * pause/resume and performance reporting. Base for DatabaseThread.
27 */
28 class PollingThread
29 {
30 public:
31 /// \brief Create a thread with the given sleep interval when no work is done.
32 /// \param interval_milliseconds How long to sleep when no Runnable had work (default 100ms).
33 29 PollingThread(const size_t interval_milliseconds = 100) :
34
1/1
✓ Branch 9 → 10 taken 29 times.
 29 interval_ms_(interval_milliseconds)
35 {
36 29 }
37
38 49 virtual ~PollingThread() noexcept { close(); }
39
40 /// \brief Return the sleep interval in milliseconds (when no work is done).
41 ✗ size_t getIntervalMilliseconds() const { return interval_ms_; }
42
43 /// \brief Add a Runnable to this thread; must not be called while the thread is running.
44 /// \throws DBException if called while the thread is running.
45 32 void addRunnable(Runnable* runnable)
46 {
47
1/12
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 22 taken 32 times.
✗ Branch 6 → 7 not taken.
✗ Branch 7 → 8 not taken.
✗ Branch 9 → 10 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 18 → 19 not taken.
✗ Branch 20 → 21 not taken.
 32 simdb_assert(!is_running_, "Cannot add runnables while thread is running");
48 32 runnables_.emplace_back(runnable);
49 32 runnable->thread_ = this;
50 32 }
51
52 /// \brief Return the Runnables on this thread.
53 31744 const std::vector<Runnable*>& getRunnables() const { return runnables_; }
54
55 /// \brief Return the number of Runnables on this thread.
56 size_t getNumRunnables() const { return runnables_.size(); }
57
58 /// \brief Reorder this thread's Runnables to match the order in \p runnables (only those
59 /// that belong to this thread).
60 29 void ensureRelativeOrder(const std::vector<Runnable*>& runnables)
61 {
62
1/1
✓ Branch 4 → 5 taken 29 times.
 29 const std::set<Runnable*> my_runnables(runnables_.begin(), runnables_.end());
63 29 std::vector<Runnable*> ordered_runnables;
64
2/2
✓ Branch 14 → 8 taken 114 times.
✓ Branch 14 → 15 taken 29 times.
 143 for (auto runnable : runnables)
65 {
66
3/3
✓ Branch 9 → 10 taken 114 times.
✓ Branch 10 → 11 taken 32 times.
✓ Branch 10 → 12 taken 82 times.
 114 if (my_runnables.count(runnable))
67 {
68
1/1
✓ Branch 11 → 12 taken 32 times.
 32 ordered_runnables.push_back(runnable);
69 }
70 }
71 29 std::swap(ordered_runnables, runnables_);
72 29 }
73
74 /// \brief Call processAll(true) on all enabled Runnables; return true if any did work.
75 49 virtual bool flushRunnables()
76 {
77 49 bool did_work = false;
78
2/2
✓ Branch 13 → 4 taken 52 times.
✓ Branch 13 → 14 taken 49 times.
 101 for (auto runnable : runnables_)
79 {
80
1/2
✗ Branch 6 → 7 not taken.
✓ Branch 6 → 8 taken 52 times.
 52 if (!runnable->enabled())
81 {
82 ✗ continue;
83 }
84
85
3/3
✓ Branch 8 → 9 taken 52 times.
✓ Branch 9 → 10 taken 11 times.
✓ Branch 9 → 11 taken 41 times.
 52 if (runnable->processAll(true) == PipelineAction::PROCEED)
86 {
87 11 did_work = true;
88 }
89 }
90 49 return did_work;
91 }
92
93 /// \brief Start the polling thread (must have at least one Runnable).
94 29 virtual void open()
95 {
96
1/2
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 5 taken 29 times.
 29 if (runnables_.empty())
97 {
98 ✗ return;
99 }
100
101
1/2
✓ Branch 6 → 7 taken 29 times.
✗ Branch 6 → 15 not taken.
 29 if (!thread_)
102 {
103 29 stop_requested_ = false;
104 29 paused_ = false;
105 29 is_running_ = true;
106 29 start_ = std::chrono::high_resolution_clock::now();
107
1/1
✓ Branch 11 → 12 taken 29 times.
 29 thread_ = std::make_unique<std::thread>(&PollingThread::loop_, this);
108 }
109 }
110
111 /// \brief Stop the thread and join.
112 /// \note Meant to be called from the main thread.
113 58 virtual void close() noexcept
114 {
115
2/2
✓ Branch 3 → 4 taken 29 times.
✓ Branch 3 → 5 taken 29 times.
 58 if (!thread_)
116 {
117 29 return;
118 }
119
120 {
121 29 std::lock_guard<std::mutex> lock(pause_mutex_);
122 29 stop_requested_ = true;
123 29 paused_ = false; // Unpause in case it was paused
124 29 }
125 29 pause_cv_.notify_all();
126
127
1/2
✓ Branch 12 → 13 taken 29 times.
✗ Branch 12 → 15 not taken.
 29 if (thread_->joinable())
128 {
129 29 thread_->join();
130 }
131 29 is_running_ = false;
132 29 thread_.reset();
133 }
134
135 /// \brief Pause the polling loop; blocks until the thread has acknowledged it is paused.
136 39999 void pause()
137 {
138
3/6
✓ Branch 3 → 4 taken 39999 times.
✗ Branch 3 → 6 not taken.
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 7 taken 39999 times.
✗ Branch 8 → 9 not taken.
✓ Branch 8 → 10 taken 39999 times.
 39999 if (!is_running_ || paused_)
139 {
140 ✗ return;
141 }
142
143 {
144
1/1
✓ Branch 10 → 11 taken 39999 times.
 39999 std::lock_guard<std::mutex> lock(pause_mutex_);
145 39999 paused_ = true;
146
147 // Reset and prepare for acknowledgment
148
1/1
✓ Branch 12 → 13 taken 39999 times.
 39999 paused_promise_ = std::promise<void>();
149 39999 has_pending_pause_ack_ = true;
150 39999 }
151
152 39999 pause_cv_.notify_all(); // Wake the thread so it can acknowledge
153
154 // Wait for thread to acknowledge it is paused
155
2/2
✓ Branch 17 → 18 taken 39999 times.
✓ Branch 18 → 19 taken 39999 times.
 39999 paused_promise_.get_future().wait();
156 }
157
158 /// \brief Return true if the thread is currently paused.
159 39999 bool paused() { return paused_; }
160
161 /// \brief Resume the polling loop after a pause.
162 39999 void resume()
163 {
164
3/6
✓ Branch 3 → 4 taken 39999 times.
✗ Branch 3 → 6 not taken.
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 7 taken 39999 times.
✗ Branch 8 → 9 not taken.
✓ Branch 8 → 10 taken 39999 times.
 39999 if (!is_running_ || !paused_)
165 {
166 ✗ return;
167 }
168
169 {
170
1/1
✓ Branch 10 → 11 taken 39999 times.
 39999 std::lock_guard<std::mutex> lock(pause_mutex_);
171 39999 paused_ = false;
172 39999 }
173
174 39999 pause_cv_.notify_all();
175 }
176
177 /// \brief Print a performance report (sleep vs work %) for this thread.
178 29 virtual bool printPerfReport() const noexcept
179 {
180
1/2
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 5 taken 29 times.
 29 if (runnables_.empty())
181 {
182 ✗ return false;
183 }
184
185
1/2
✗ Branch 6 → 7 not taken.
✓ Branch 6 → 8 taken 29 times.
 29 if (is_running_)
186 {
187 ✗ return false;
188 }
189
190 29 const auto pct_time_sleeping = getSleepPct();
191 29 const auto pct_time_working = 100 - pct_time_sleeping;
192
193 29 std::cout << "Thread containing:\n";
194
2/2
✓ Branch 16 → 12 taken 32 times.
✓ Branch 16 → 17 taken 29 times.
 61 for (const auto runnable : runnables_)
195 {
196 32 runnable->print(std::cout, 4);
197 }
198
199 29 [[maybe_unused]] ios_format_saver fmt_saver(std::cout);
200 29 std::cout << "\n";
201 29 std::cout << " Performance report:\n";
202 29 std::cout << " Pct time sleeping: " << std::fixed << std::setprecision(1) << pct_time_sleeping << "%\n";
203 29 std::cout << " Pct time working: " << std::fixed << std::setprecision(1) << pct_time_working << "%\n";
204
205 29 return true;
206 29 }
207
208 25717 double getSleepPct() const
209 {
210 25717 auto now = std::chrono::high_resolution_clock::now();
211
2/2
✓ Branch 3 → 4 taken 25717 times.
✓ Branch 4 → 5 taken 25717 times.
 25717 const std::chrono::duration<double> dur = now - start_;
212 25717 const auto total_elap_seconds = dur.count();
213 25717 const auto pct_time_sleeping = (total_sleep_seconds_ / total_elap_seconds) * 100;
214 25717 return pct_time_sleeping;
215 }
216
217 private:
218 29 void loop_()
219 {
220
2/2
✓ Branch 40 → 3 taken 40720 times.
✓ Branch 40 → 41 taken 29 times.
 40778 while (!stop_requested_)
221 {
222 // Pause handling
223 {
224
1/1
✓ Branch 3 → 4 taken 40720 times.
 40720 std::unique_lock<std::mutex> lock(pause_mutex_);
225
5/6
✓ Branch 11 → 12 taken 39999 times.
✓ Branch 11 → 15 taken 40720 times.
✓ Branch 13 → 14 taken 39999 times.
✗ Branch 13 → 15 not taken.
✓ Branch 16 → 5 taken 39999 times.
✓ Branch 16 → 17 taken 40720 times.
 80719 while (paused_ && !stop_requested_)
226 {
227
1/2
✓ Branch 5 → 6 taken 39999 times.
✗ Branch 5 → 8 not taken.
 39999 if (has_pending_pause_ack_)
228 {
229
1/1
✓ Branch 6 → 7 taken 39999 times.
 39999 paused_promise_.set_value();
230 39999 has_pending_pause_ack_ = false;
231 }
232
1/1
✓ Branch 8 → 9 taken 39999 times.
 39999 pause_cv_.wait(lock);
233 }
234
235
1/2
✗ Branch 18 → 19 not taken.
✓ Branch 18 → 20 taken 40720 times.
 40720 if (stop_requested_)
236 {
237 ✗ break;
238 }
239 40720 }
240
241
2/2
✓ Branch 25 → 26 taken 1168 times.
✓ Branch 25 → 28 taken 39552 times.
 40720 if (!run_(false))
242 {
243 // Sleep for a fixed amount of time before polling all runnables
244 // again but wake early if paused or stop is requested
245 39552 auto sleep_start = std::chrono::high_resolution_clock::now();
246
1/1
✓ Branch 29 → 30 taken 39552 times.
 39552 std::unique_lock<std::mutex> lock(pause_mutex_);
247
1/1
✓ Branch 31 → 32 taken 39552 times.
 39552 pause_cv_.wait_for(lock, std::chrono::milliseconds(interval_ms_),
248
4/4
✓ Branch 3 → 4 taken 38845 times.
✓ Branch 3 → 6 taken 39374 times.
✓ Branch 5 → 6 taken 24 times.
✓ Branch 5 → 7 taken 38821 times.
 78219 [this] { return paused_ || stop_requested_; });
249
250 39552 auto sleep_end = std::chrono::high_resolution_clock::now();
251
2/2
✓ Branch 33 → 34 taken 39552 times.
✓ Branch 34 → 35 taken 39552 times.
 39552 auto duration_us = std::chrono::duration_cast<std::chrono::microseconds>(sleep_end - sleep_start);
252 39552 total_sleep_seconds_ += duration_us.count() / 1'000'000.0;
253 39552 }
254 }
255
256 // Flush
257
2/2
✓ Branch 43 → 44 taken 5 times.
✓ Branch 43 → 45 taken 29 times.
 34 while (run_(true))
258 {
259 }
260 29 }
261
262 29639 virtual bool run_(bool force)
263 {
264 29639 bool did_work = false;
265 while (true)
266 {
267 54373 bool processed = false;
268
2/2
✓ Branch 14 → 5 taken 54373 times.
✓ Branch 14 → 15 taken 54373 times.
 108746 for (auto runner : runnables_)
269 {
270
2/2
✓ Branch 7 → 8 taken 13023 times.
✓ Branch 7 → 9 taken 41350 times.
 54373 if (!runner->enabled())
271 {
272 13023 continue;
273 }
274
275
3/3
✓ Branch 9 → 10 taken 41350 times.
✓ Branch 10 → 11 taken 24734 times.
✓ Branch 10 → 12 taken 16616 times.
 41350 if (runner->processOne(force) == PipelineAction::PROCEED)
276 {
277 24734 processed = true;
278 }
279 }
280
2/2
✓ Branch 15 → 16 taken 29639 times.
✓ Branch 15 → 17 taken 24734 times.
 54373 if (!processed)
281 {
282 29639 break;
283 }
284 24734 did_work = true;
285 24734 }
286 29639 return did_work;
287 }
288
289 const size_t interval_ms_;
290 std::vector<Runnable*> runnables_;
291 std::unique_ptr<std::thread> thread_;
292
293 std::mutex pause_mutex_;
294 std::condition_variable pause_cv_;
295
296 std::atomic<bool> is_running_ = false;
297 std::atomic<bool> paused_ = false;
298 std::atomic<bool> stop_requested_ = false;
299
300 std::promise<void> paused_promise_;
301 bool has_pending_pause_ack_ = false;
302
303 std::chrono::high_resolution_clock::time_point start_;
304 double total_sleep_seconds_ = 0;
305 };
306
307 /// Defined here so we can avoid circular includes
308 10000 inline ScopedRunnableDisabler::ScopedRunnableDisabler(PipelineManager* pipeline_mgr,
309 const std::vector<Runnable*>& runnables,
310 10000 const std::vector<PollingThread*>& polling_threads) :
311 10000 pipeline_mgr_(pipeline_mgr)
312 {
313 // Disable runnables
314
2/2
✓ Branch 13 → 6 taken 39999 times.
✓ Branch 13 → 14 taken 10000 times.
 49999 for (auto r : runnables)
315 {
316
1/2
✓ Branch 8 → 9 taken 39999 times.
✗ Branch 8 → 11 not taken.
 39999 if (r->enabled())
317 {
318 39999 r->enable(false);
319
1/1
✓ Branch 10 → 11 taken 39999 times.
 39999 disabled_runnables_.push_back(r);
320 }
321 }
322
323 // Pause polling threads
324
2/2
✓ Branch 23 → 16 taken 39999 times.
✓ Branch 23 → 24 taken 10000 times.
 49999 for (auto pt : polling_threads)
325 {
326
1/2
✓ Branch 18 → 19 taken 39999 times.
✗ Branch 18 → 21 not taken.
 39999 if (!pt->paused())
327 {
328
1/1
✓ Branch 19 → 20 taken 39999 times.
 39999 pt->pause();
329
1/1
✓ Branch 20 → 21 taken 39999 times.
 39999 paused_threads_.push_back(pt);
330 }
331 }
332 10000 }
333
334 /// Defined here so we can avoid circular includes
335 ✗ inline ScopedRunnableDisabler::ScopedRunnableDisabler(PipelineManager* pipeline_mgr,
336 ✗ const std::vector<Runnable*>& runnables) :
337 ✗ pipeline_mgr_(pipeline_mgr)
338 {
339 // Disable runnables
340
0/2
✗ Branch 13 → 6 not taken.
✗ Branch 13 → 14 not taken.
 ✗ for (auto r : runnables)
341 {
342
0/2
✗ Branch 8 → 9 not taken.
✗ Branch 8 → 11 not taken.
 ✗ if (r->enabled())
343 {
344 ✗ r->enable(false);
345
0/1
✗ Branch 10 → 11 not taken.
 ✗ disabled_runnables_.push_back(r);
346 }
347 }
348 ✗ }
349
350 /// Defined here so we can avoid circular includes
351 10000 inline ScopedRunnableDisabler::~ScopedRunnableDisabler()
352 {
353
2/2
✓ Branch 8 → 4 taken 39999 times.
✓ Branch 8 → 9 taken 10000 times.
 49999 for (auto pt : paused_threads_)
354 {
355 39999 pt->resume();
356 }
357
358
2/2
✓ Branch 15 → 11 taken 39999 times.
✓ Branch 15 → 16 taken 10000 times.
 49999 for (auto r : disabled_runnables_)
359 {
360 39999 r->enable(true);
361 }
362
363 10000 notifyPipelineMgrReenabled_();
364 10000 }
365
366 } // namespace simdb::pipeline
367