GCC Code Coverage Report

include/simdb/
Coverage:
low: ≥ 0%
medium: ≥ 75.0%
high: ≥ 90.0%
Lines:
102 / 107
95.3%
Branches:
62 / 96
64.6%
Functions:
21 / 22
95.5%
pipeline/DatabaseThread.hpp
Line Branch Exec Source
1 // <DatabaseThread.hpp> -*- C++ -*-
2
3 #pragma once
4
5 #include "simdb/Assert.hpp"
6 #include "simdb/pipeline/AsyncDatabaseAccessor.hpp"
7 #include "simdb/pipeline/PollingThread.hpp"
8 #include "simdb/sqlite/DatabaseManager.hpp"
9 #include "simdb/utils/ConcurrentQueue.hpp"
10
11 namespace simdb {
12 class AsyncDatabaseAccessor;
13 class DatabaseManager;
14 } // namespace simdb
15
16 namespace simdb::pipeline {
17
18 /*!
19 * \class DatabaseThread
20 *
21 * \brief PollingThread dedicated to database stages. Runs all its Runnables
22 * inside batched BEGIN/COMMIT transactions for performance, and
23 * provides AsyncDatabaseAccessor for other threads to submit work
24 * (eval) that runs on this thread. Exactly one DatabaseThread exists
25 * for every pipeline associated with the same DatabaseManager (.db file).
26 */
27 class DatabaseThread : public PollingThread, private AsyncDatabaseAccessHandler
28 {
29 public:
30 /// \brief Construct the database thread for the given DatabaseManager.
31 /// \param db_mgr DatabaseManager used for safeTransaction and async tasks.
32 9 DatabaseThread(DatabaseManager* db_mgr) :
33 9 db_mgr_(db_mgr),
34
1/1
✓ Branch 6 → 7 taken 9 times.
 9 dormant_thread_(db_mgr)
35 {
36 9 }
37
38 18 ~DatabaseThread() noexcept = default;
39
40 /// \brief Return the AsyncDatabaseAccessor for submitting work to this thread.
41 9 AsyncDatabaseAccessor* getAsyncDatabaseAccessor() { return &db_accessor_; }
42
43 /// \brief Print a performance report (sleep vs work %) for this thread.
44 /// Also append DB-thread-specific perf numbers.
45 9 bool printPerfReport() const noexcept override final
46 {
47 9 auto printed = PollingThread::printPerfReport();
48
5/6
✓ Branch 3 → 4 taken 9 times.
✗ Branch 3 → 7 not taken.
✓ Branch 5 → 6 taken 7 times.
✓ Branch 5 → 7 taken 2 times.
✓ Branch 8 → 9 taken 7 times.
✓ Branch 8 → 16 taken 2 times.
 9 if (printed && db_perf_report_.largest_file_change)
49 {
50 7 std::cout << " Num committed transactions: " << db_perf_report_.transactions_issued << "\n";
51 7 std::cout << " Largest transaction change: " << *db_perf_report_.largest_file_change << " bytes\n";
52 }
53 9 return printed;
54 }
55
56 private:
57 /// Overridden from AsyncDatabaseAccessHandler
58 4 void eval(AsyncDatabaseTaskPtr&& task, double timeout_seconds = 0) override final
59 {
60 4 dormant_thread_.eval(std::move(task), timeout_seconds);
61 4 }
62
63 /// Overridden from PollingThread
64 11115 bool run_(bool force) override final
65 {
66 11115 bool did_work = false;
67
68 11115 std::error_code ec;
69
1/1
✓ Branch 4 → 5 taken 11115 times.
 11115 int64_t starting_db_bytes = std::filesystem::file_size(db_mgr_->getDatabaseFilePath(), ec);
70 11115 bool perf_good = !static_cast<bool>(ec);
71
72 // Put all runnables (AsyncDbWriters) in a BEGIN/COMMIT block
73 // for fastest insert performance.
74
2/2
✓ Branch 9 → 10 taken 11115 times.
✓ Branch 10 → 11 taken 11115 times.
 22230 db_mgr_->safeTransaction([&]() {
75 bool continue_while;
76 31688 do
77 {
78 31688 continue_while = false;
79
2/2
✓ Branch 18 → 6 taken 50375 times.
✓ Branch 18 → 19 taken 31686 times.
 82061 for (auto runnable : getRunnables())
80 {
81
2/2
✓ Branch 8 → 9 taken 7531 times.
✓ Branch 8 → 10 taken 42844 times.
 50375 if (!runnable->enabled())
82 {
83 7531 continue;
84 }
85
86 // We call processOne() here instead of processAll() to use
87 // a smaller granularity of tasks to "inject" break
88 // statements more frequently in the event of pending async
89 // DB access requests.
90
3/3
✓ Branch 10 → 11 taken 42844 times.
✓ Branch 11 → 12 taken 24691 times.
✓ Branch 11 → 13 taken 18153 times.
 42844 if (runnable->processOne(force) == PipelineAction::PROCEED)
91 {
92 24691 continue_while = true;
93 }
94
95 // Give as high priority as possible for async DB access
96
3/3
✓ Branch 13 → 14 taken 42844 times.
✓ Branch 14 → 15 taken 2 times.
✓ Branch 14 → 16 taken 42842 times.
 42844 if (dormant_thread_.hasTasks())
97 {
98 2 continue_while = false;
99 2 break;
100 }
101 }
102
2/2
✓ Branch 19 → 20 taken 20573 times.
✓ Branch 19 → 21 taken 11115 times.
 31688 did_work |= continue_while;
103 } while (continue_while);
104 11115 });
105
106
3/4
✓ Branch 12 → 13 taken 11115 times.
✗ Branch 12 → 21 not taken.
✓ Branch 13 → 14 taken 1062 times.
✓ Branch 13 → 21 taken 10053 times.
 11115 if (perf_good && did_work)
107 {
108
1/1
✓ Branch 15 → 16 taken 1062 times.
 1062 int64_t current_db_bytes = std::filesystem::file_size(db_mgr_->getDatabaseFilePath(), ec);
109 1062 perf_good &= !static_cast<bool>(ec);
110
1/2
✓ Branch 19 → 20 taken 1062 times.
✗ Branch 19 → 21 not taken.
 1062 if (perf_good)
111 {
112 1062 recordTransaction_(starting_db_bytes, current_db_bytes);
113 }
114 }
115
116 11115 return did_work;
117 }
118
119 /// Overridden from PollingThread
120 9 void open() override
121 {
122 9 PollingThread::open();
123 9 dormant_thread_.open();
124 9 }
125
126 /// Overridden from PollingThread
127 9 void close() noexcept override
128 {
129 9 PollingThread::close();
130 9 dormant_thread_.close();
131 9 }
132
133 /// Overridden from PollingThread
134 15 bool flushRunnables() override
135 {
136 15 bool did_work = false;
137
138
1/1
✓ Branch 4 → 5 taken 15 times.
 30 db_mgr_->safeTransaction([&] { did_work = PollingThread::flushRunnables(); });
139
140 15 return did_work;
141 }
142
143 /*!
144 * \class DormantThread
145 *
146 * \brief Thread that sleeps until an async DB task is enqueued; then runs
147 * the task in a safeTransaction() and goes back to sleep. Used to
148 * service AsyncDatabaseAccessor::eval() without blocking the main
149 * DB polling loop.
150 */
151 class DormantThread
152 {
153 public:
154 9 explicit DormantThread(DatabaseManager* db_mgr) :
155
1/1
✓ Branch 5 → 6 taken 9 times.
 9 db_mgr_(db_mgr)
156 {
157 9 }
158
159 9 ~DormantThread() noexcept { close(); }
160
161 /// \brief Start the dormant thread (waits on condition variable for tasks).
162 9 void open()
163 {
164
1/2
✓ Branch 3 → 4 taken 9 times.
✗ Branch 3 → 9 not taken.
 9 if (!thread_)
165 {
166 9 stop_ = false;
167
1/1
✓ Branch 5 → 6 taken 9 times.
 9 thread_ = std::make_unique<std::thread>(&DormantThread::loop_, this);
168 }
169 9 }
170
171 /// \brief Stop the thread and join.
172 18 void close() noexcept
173 {
174 {
175 18 std::lock_guard<std::mutex> lock(mutex_);
176 18 stop_ = true;
177 18 }
178 18 cond_var_.notify_one();
179
180
2/2
✓ Branch 7 → 8 taken 9 times.
✓ Branch 7 → 14 taken 9 times.
 18 if (thread_)
181 {
182
1/2
✓ Branch 10 → 11 taken 9 times.
✗ Branch 10 → 13 not taken.
 9 if (thread_->joinable())
183 {
184 9 thread_->join();
185 }
186 9 thread_.reset();
187 }
188 18 }
189
190 /// \brief Enqueue a task and optionally block until it completes (or timeout).
191 /// \param task Task to run on this thread inside safeTransaction().
192 /// \param timeout_seconds If > 0, block up to this many seconds; throw on timeout.
193 /// \throws DBException on timeout or if the task throws.
194 4 void eval(AsyncDatabaseTaskPtr&& task, double timeout_seconds = 0)
195 {
196
1/2
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 9 taken 4 times.
 4 if (stop_)
197 {
198 // If we are already stopped, we cannot use std::future. We have
199 // to evaluate the task right here.
200
0/1
✗ Branch 6 → 7 not taken.
 ✗ db_mgr_->safeTransaction([&]() { task->func(db_mgr_); });
201 ✗ return;
202 }
203
204
1/1
✓ Branch 10 → 11 taken 4 times.
 4 std::future<std::string> fut = task->exception_reason.get_future();
205
1/1
✓ Branch 12 → 13 taken 4 times.
 4 pending_async_db_tasks_.emplace(std::move(task));
206 4 cond_var_.notify_one();
207
208
1/2
✓ Branch 14 → 15 taken 4 times.
✗ Branch 14 → 36 not taken.
 4 if (timeout_seconds > 0)
209 {
210
1/1
✓ Branch 16 → 17 taken 4 times.
 4 auto status = fut.wait_for(std::chrono::duration<double>(timeout_seconds));
211
1/12
✗ Branch 17 → 18 not taken.
✓ Branch 17 → 36 taken 4 times.
✗ Branch 20 → 21 not taken.
✗ Branch 21 → 22 not taken.
✗ Branch 23 → 24 not taken.
✗ Branch 27 → 28 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 34 → 35 not taken.
 4 simdb_assert(status != std::future_status::timeout, "Timed out waiting for async DB task to complete");
212 }
213
214
1/1
✓ Branch 36 → 37 taken 4 times.
 4 auto exception_reason = fut.get();
215
1/12
✗ Branch 38 → 39 not taken.
✓ Branch 38 → 57 taken 4 times.
✗ Branch 41 → 42 not taken.
✗ Branch 42 → 43 not taken.
✗ Branch 44 → 45 not taken.
✗ Branch 48 → 49 not taken.
✗ Branch 49 → 50 not taken.
✗ Branch 50 → 51 not taken.
✗ Branch 51 → 52 not taken.
✗ Branch 52 → 53 not taken.
✗ Branch 53 → 54 not taken.
✗ Branch 55 → 56 not taken.
 4 simdb_assert(exception_reason.empty(), exception_reason);
216 4 }
217
218 /// \brief Return true if there are pending async DB tasks in the queue.
219 42844 bool hasTasks() const { return !pending_async_db_tasks_.empty(); }
220
221 private:
222 /// \brief Main loop that runs until the thread is stopped.
223 9 void loop_()
224 {
225 while (true)
226 {
227
1/1
✓ Branch 3 → 4 taken 13 times.
 13 std::unique_lock<std::mutex> lock(mutex_);
228
5/5
✓ Branch 3 → 4 taken 21 times.
✓ Branch 3 → 6 taken 4 times.
✓ Branch 4 → 5 taken 13 times.
✓ Branch 5 → 6 taken 9 times.
✓ Branch 5 → 7 taken 12 times.
 38 cond_var_.wait(lock, [this] { return !pending_async_db_tasks_.empty() || stop_; });
229
230 // Assume we just woke up due to an async DB access request.
231 // Immediately start a transaction so the DB polling thread
232 // blocks until we are done responding to the requests.
233
1/1
✓ Branch 7 → 8 taken 13 times.
 13 db_mgr_->safeTransaction([&]() {
234 13 AsyncDatabaseTaskPtr async_task;
235
3/3
✓ Branch 14 → 15 taken 17 times.
✓ Branch 15 → 3 taken 4 times.
✓ Branch 15 → 16 taken 13 times.
 17 while (pending_async_db_tasks_.try_pop(async_task))
236 {
237 try
238 {
239
1/1
✓ Branch 4 → 5 taken 4 times.
 4 async_task->func(db_mgr_);
240
2/2
✓ Branch 8 → 9 taken 4 times.
✓ Branch 9 → 10 taken 4 times.
 12 async_task->exception_reason.set_value("");
241
0/2
✗ Branch 24 → 25 not taken.
✗ Branch 24 → 26 not taken.
 ✗ } catch (const std::exception& ex)
242 {
243
0/2
✗ Branch 31 → 32 not taken.
✗ Branch 32 → 33 not taken.
 ✗ async_task->exception_reason.set_value(ex.what());
244 ✗ }
245 }
246 13 });
247
248
2/2
✓ Branch 10 → 11 taken 9 times.
✓ Branch 10 → 12 taken 4 times.
 13 if (stop_)
249 {
250 9 break;
251 }
252 17 }
253 9 }
254
255 DatabaseManager* db_mgr_ = nullptr;
256 std::unique_ptr<std::thread> thread_;
257 std::mutex mutex_;
258 std::condition_variable cond_var_;
259 std::atomic<bool> stop_;
260 ConcurrentQueue<AsyncDatabaseTaskPtr> pending_async_db_tasks_;
261 };
262
263 DatabaseManager* db_mgr_ = nullptr;
264 AsyncDatabaseAccessor db_accessor_{this};
265 std::vector<Runnable*> polling_runnables_;
266 DormantThread dormant_thread_;
267
268 struct DbPerfReport
269 {
270 uint64_t transactions_issued = 0;
271 std::optional<int64_t> largest_file_change;
272 } db_perf_report_;
273
274 1062 void recordTransaction_(int64_t prev_bytes, int64_t curr_bytes)
275 {
276 1062 ++db_perf_report_.transactions_issued;
277 1062 const auto delta = curr_bytes - prev_bytes;
278
6/6
✓ Branch 3 → 4 taken 1055 times.
✓ Branch 3 → 8 taken 7 times.
✓ Branch 7 → 8 taken 11 times.
✓ Branch 7 → 9 taken 1044 times.
✓ Branch 10 → 11 taken 18 times.
✓ Branch 10 → 13 taken 1044 times.
 1062 if (!db_perf_report_.largest_file_change || std::abs(delta) > std::abs(*db_perf_report_.largest_file_change))
279 {
280 18 db_perf_report_.largest_file_change = delta;
281 }
282 1062 }
283 };
284
285 } // namespace simdb::pipeline
286