INSTINCT Code Coverage Report

Directory: src/
File: internal/FlowExecutor.cpp
Date: 2025-07-19 10:51:51
Exec Total Coverage
Lines: 120 128 93.8%
Functions: 11 11 100.0%
Branches: 148 256 57.8%
Line Branch Exec Source
1 // This file is part of INSTINCT, the INS Toolkit for Integrated
2 // Navigation Concepts and Training by the Institute of Navigation of
3 // the University of Stuttgart, Germany.
4 //
5 // This Source Code Form is subject to the terms of the Mozilla Public
6 // License, v. 2.0. If a copy of the MPL was not distributed with this
7 // file, You can obtain one at https://mozilla.org/MPL/2.0/.
8
9 #include "FlowExecutor.hpp"
10
11 // <boost/asio.hpp> needs to be included before <winsock.h> (even though not used in this file)
12 // https://stackoverflow.com/questions/9750344/boostasio-winsock-and-winsock-2-compatibility-issue
13 #ifdef _WIN32
14 // Set the proper SDK version before including boost/Asio
15 #include <SDKDDKVer.h>
16 // Note boost/ASIO includes Windows.h.
17 #include <boost/asio.hpp>
18 #endif //_WIN32
19
20 #include "Navigation/GNSS/Positioning/AntexReader.hpp"
21 #include "util/Logger.hpp"
22 #include "Navigation/Time/InsTime.hpp"
23
24 #include "internal/Node/Node.hpp"
25
26 #include "internal/NodeManager.hpp"
27 namespace nm = NAV::NodeManager;
28 #include "internal/ConfigManager.hpp"
29 #include "util/Time/TimeBase.hpp"
30
31 #include <chrono>
32 #include <map>
33 #include <variant>
34 #include <memory>
35
36 #include <thread>
37 #include <atomic>
38 #include <mutex>
39 #include <condition_variable>
40
41 #ifdef TESTING
42 #include <catch2/catch_test_macros.hpp>
43 #endif
44
45 /* -------------------------------------------------------------------------------------------------------- */
46 /* Private Members */
47 /* -------------------------------------------------------------------------------------------------------- */
48
49 namespace NAV::FlowExecutor
50 {
51 namespace
52 {
53
54 std::mutex _mutex;
55 std::condition_variable _cv;
56
57 enum class State : uint8_t
58 {
59 Idle,
60 Starting,
61 Running,
62 Stopping,
63 };
64 State _state = State::Idle;
65
66 std::thread _thd;
67 std::atomic<size_t> _activeNodes{ 0 };
68 std::chrono::time_point<std::chrono::steady_clock> _startTime;
69
70 /* -------------------------------------------------------------------------------------------------------- */
71 /* Private Function Declarations */
72 /* -------------------------------------------------------------------------------------------------------- */
73
74 } // namespace
75
76 /// @brief Main task of the thread
77 void execute();
78
79 } // namespace NAV::FlowExecutor
80
81 /* -------------------------------------------------------------------------------------------------------- */
82 /* Function Definitions */
83 /* -------------------------------------------------------------------------------------------------------- */
84
85 228 bool NAV::FlowExecutor::isRunning() noexcept
86 {
87 228 std::scoped_lock<std::mutex> lk(_mutex);
88 456 return _state != State::Idle;
89 228 }
90
91 113 void NAV::FlowExecutor::start()
92 {
93 LOG_TRACE("called");
94
95 113 stop();
96
97 {
98
1/2
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
 113 std::scoped_lock<std::mutex> lk(_mutex);
99 113 _state = State::Starting;
100 113 }
101
102
1/2
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
 113 _thd = std::thread(execute);
103 113 }
104
105 114 void NAV::FlowExecutor::stop()
106 {
107 LOG_TRACE("called");
108
109
2/2
✓ Branch 1 taken 1 times.
✓ Branch 2 taken 113 times.
 114 if (isRunning())
110 {
111 {
112
1/2
✓ Branch 1 taken 1 times.
✗ Branch 2 not taken.
 1 std::scoped_lock<std::mutex> lk(_mutex);
113
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 1 if (_state == State::Running || _state == State::Starting)
114 {
115 1 _state = State::Stopping;
116 1 _cv.notify_all();
117 }
118 1 }
119
120 1 waitForFinish();
121 }
122
123
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 114 times.
 114 if (_thd.joinable()) { _thd.join(); }
124 114 }
125
126 113 void NAV::FlowExecutor::waitForFinish()
127 {
128 LOG_TRACE("Waiting for finish of FlowExecutor...");
129 {
130
1/2
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
 113 std::unique_lock lk(_mutex);
131
1/2
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
 451 _cv.wait(lk, [] { return _state == State::Idle; });
132 113 }
133
134 {
135
1/2
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
 113 std::scoped_lock<std::mutex> lk(_mutex);
136
2/4
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 113 times.
✗ Branch 5 not taken.
 113 if (_thd.joinable()) { _thd.join(); }
137 113 }
138 LOG_TRACE("FlowExecutor finished.");
139 113 }
140
141 345 void NAV::FlowExecutor::deregisterNode([[maybe_unused]] const Node* node)
142 {
143
2/4
✓ Branch 2 taken 345 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 345 times.
✗ Branch 7 not taken.
 345 LOG_DEBUG("Node {} finished.", node->nameId());
144 345 _activeNodes--;
145
146
2/2
✓ Branch 1 taken 112 times.
✓ Branch 2 taken 233 times.
 345 if (_activeNodes == 0)
147 {
148
1/2
✓ Branch 1 taken 112 times.
✗ Branch 2 not taken.
 112 std::scoped_lock<std::mutex> lk(_mutex);
149 112 _state = State::Stopping;
150 112 _cv.notify_all();
151 112 }
152 345 }
153
154 113 void NAV::FlowExecutor::execute()
155 {
156 LOG_TRACE("called");
157
158 113 AntexReader::Get().reset();
159
160
3/4
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
✓ Branch 8 taken 355 times.
✓ Branch 9 taken 113 times.
 468 for (Node* node : nm::m_Nodes())
161 {
162
2/2
✓ Branch 4 taken 301 times.
✓ Branch 5 taken 355 times.
 656 for (auto& inputPin : node->inputPins)
163 {
164 301 inputPin.queue.clear();
165 301 inputPin.queueBlocked = false;
166 }
167 355 node->pollEvents.clear();
168 }
169
170
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 113 times.
 113 if (!nm::InitializeAllNodes()) // This wakes the threads
171 {
172 ✗ std::scoped_lock<std::mutex> lk(_mutex);
173 ✗ _state = State::Idle;
174 ✗ _cv.notify_all();
175 ✗ return;
176 ✗ }
177
178
1/2
✓ Branch 3 taken 113 times.
✗ Branch 4 not taken.
 113 LOG_INFO("=====================================================");
179 113 bool realTimeMode = std::any_of(nm::m_Nodes().begin(), nm::m_Nodes().end(), [](const Node* node) {
180
5/6
✓ Branch 0 taken 352 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 351 times.
✓ Branch 4 taken 1 times.
✓ Branch 5 taken 1 times.
✓ Branch 6 taken 350 times.
 352 return node && !node->isDisabled() && node->_onlyRealTime;
181 });
182
3/4
✓ Branch 1 taken 1 times.
✓ Branch 2 taken 112 times.
✓ Branch 5 taken 113 times.
✗ Branch 6 not taken.
 113 LOG_INFO("Executing in {} mode", realTimeMode ? "real-time" : "post-processing");
183
184
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 112 times.
 113 util::time::SetMode(realTimeMode ? util::time::Mode::REAL_TIME : util::time::Mode::POST_PROCESSING);
185 113 _activeNodes = 0;
186
187
3/4
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
✓ Branch 8 taken 355 times.
✓ Branch 9 taken 113 times.
 468 for (Node* node : nm::m_Nodes())
188 {
189
8/10
✓ Branch 0 taken 355 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 351 times.
✓ Branch 4 taken 4 times.
✓ Branch 6 taken 351 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 1 times.
✓ Branch 9 taken 350 times.
✓ Branch 10 taken 5 times.
✓ Branch 11 taken 350 times.
 355 if (node == nullptr || node->kind == Node::Kind::GroupBox || !node->isInitialized()) { continue; }
190
191 {
192
1/2
✓ Branch 1 taken 350 times.
✗ Branch 2 not taken.
 350 std::scoped_lock<std::mutex> lk(_mutex);
193
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 350 times.
 350 if (_state != State::Starting) { break; }
194 350 }
195
196
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 344 times.
 350 node->_mode = realTimeMode ? Node::Mode::REAL_TIME : Node::Mode::POST_PROCESSING;
197
2/2
✓ Branch 0 taken 344 times.
✓ Branch 1 taken 6 times.
 350 if (!realTimeMode)
198 {
199 344 _activeNodes += 1;
200 LOG_TRACE("Putting node '{}' into post-processing mode and adding to active nodes.", node->nameId());
201 }
202 {
203
1/2
✓ Branch 1 taken 350 times.
✗ Branch 2 not taken.
 350 std::scoped_lock<std::mutex> guard(node->_configWindowMutex);
204
1/2
✓ Branch 1 taken 350 times.
✗ Branch 2 not taken.
 350 node->resetNode();
205 350 }
206
2/2
✓ Branch 1 taken 292 times.
✓ Branch 2 taken 350 times.
 642 for (size_t i = 0; i < node->outputPins.size(); i++) // for (auto& outputPin : node->outputPins)
207 {
208 292 auto& outputPin = node->outputPins[i];
209
7/8
✓ Branch 1 taken 224 times.
✓ Branch 2 taken 68 times.
✓ Branch 4 taken 224 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 215 times.
✓ Branch 7 taken 9 times.
✓ Branch 8 taken 215 times.
✓ Branch 9 taken 77 times.
 292 if (outputPin.type == Pin::Type::Flow && outputPin.isPinLinked())
210 {
211 215 outputPin.noMoreDataAvailable = false;
212 LOG_TRACE(" Setting pin '{}' to hasDataAvailable", outputPin.name);
213 }
214
215
2/2
✓ Branch 1 taken 116 times.
✓ Branch 2 taken 176 times.
 292 if (std::holds_alternative<OutputPin::PollDataFunc>(outputPin.data))
216 {
217 LOG_TRACE(" Adding pin '{}' to data poll event list.", outputPin.name);
218
1/2
✓ Branch 4 taken 116 times.
✗ Branch 5 not taken.
 116 node->pollEvents.insert(std::make_pair(InsTime(), std::make_pair(&outputPin, i)));
219 }
220
2/2
✓ Branch 1 taken 24 times.
✓ Branch 2 taken 152 times.
 176 else if (std::holds_alternative<OutputPin::PeekPollDataFunc>(outputPin.data))
221 {
222 24 if (auto* callback = std::get_if<OutputPin::PeekPollDataFunc>(&outputPin.data);
223
2/4
✓ Branch 1 taken 18 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 18 times.
✗ Branch 4 not taken.
 42 !outputPin.noMoreDataAvailable && callback != nullptr && *callback != nullptr
224
6/8
✓ Branch 0 taken 18 times.
✓ Branch 1 taken 6 times.
✓ Branch 3 taken 18 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 18 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 18 times.
✓ Branch 8 taken 6 times.
 42 && std::ranges::any_of(outputPin.links, [](const OutputPin::OutgoingLink& link) {
225 18 return link.connectedNode->isInitialized();
226 }))
227 {
228
3/6
✗ Branch 0 not taken.
✓ Branch 1 taken 18 times.
✓ Branch 3 taken 18 times.
✗ Branch 4 not taken.
✓ Branch 6 taken 18 times.
✗ Branch 7 not taken.
 18 if (auto obs = (node->**callback)(i, true)) // Peek the data
229 {
230 LOG_TRACE(" Adding pin '{}' to data poll event list with time {}.", outputPin.name, obs->insTime);
231
1/2
✓ Branch 4 taken 18 times.
✗ Branch 5 not taken.
 18 node->pollEvents.insert(std::make_pair(obs->insTime, std::make_pair(&outputPin, i)));
232 18 }
233 }
234 }
235 }
236 }
237
238 {
239
1/2
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
 113 std::scoped_lock<std::mutex> lk(_mutex);
240
1/2
✓ Branch 0 taken 113 times.
✗ Branch 1 not taken.
 113 if (_state == State::Starting)
241 {
242
1/2
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
 113 nm::EnableAllCallbacks();
243 113 _state = State::Running;
244 }
245 113 }
246
247 113 _startTime = std::chrono::steady_clock::now();
248
1/2
✓ Branch 3 taken 113 times.
✗ Branch 4 not taken.
 113 LOG_INFO("Execution started");
249
250 113 bool anyNodeRunning = false;
251
3/4
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
✓ Branch 8 taken 355 times.
✓ Branch 9 taken 113 times.
 468 for (Node* node : nm::m_Nodes()) // Search for node pins with data callbacks
252 {
253
8/10
✓ Branch 0 taken 355 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 351 times.
✓ Branch 4 taken 4 times.
✓ Branch 6 taken 351 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 350 times.
✓ Branch 9 taken 1 times.
✓ Branch 10 taken 350 times.
✓ Branch 11 taken 5 times.
 355 if (node != nullptr && node->kind != Node::Kind::GroupBox && node->isInitialized())
254 {
255 350 anyNodeRunning = true;
256
3/6
✓ Branch 1 taken 350 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 350 times.
✗ Branch 5 not taken.
✓ Branch 8 taken 350 times.
✗ Branch 9 not taken.
 350 LOG_DEBUG("Waking up node {}", node->nameId());
257
1/2
✓ Branch 1 taken 350 times.
✗ Branch 2 not taken.
 350 node->wakeWorker();
258 }
259 }
260
261
1/2
✓ Branch 0 taken 113 times.
✗ Branch 1 not taken.
 113 if (anyNodeRunning)
262 {
263 // Wait for the nodes to finish
264 113 bool timeout = true;
265 113 auto timeoutDuration = std::chrono::minutes(1);
266
2/2
✓ Branch 0 taken 120 times.
✓ Branch 1 taken 112 times.
 232 while (timeout)
267 {
268
1/2
✓ Branch 1 taken 120 times.
✗ Branch 2 not taken.
 120 std::unique_lock lk(_mutex);
269
270
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 119 times.
 120 if (realTimeMode)
271 {
272
1/2
✓ Branch 1 taken 1 times.
✗ Branch 2 not taken.
 3 _cv.wait(lk, [] { return _state == State::Stopping; });
273 1 break;
274 }
275
276
1/2
✓ Branch 1 taken 119 times.
✗ Branch 2 not taken.
 357 timeout = !_cv.wait_for(lk, timeoutDuration, [] { return _state == State::Stopping; });
277
4/6
✓ Branch 0 taken 7 times.
✓ Branch 1 taken 112 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 7 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 119 times.
 119 if (timeout && _activeNodes == 0)
278 {
279 ✗ LOG_ERROR("FlowExecutor had a timeout, but all nodes finished already.");
280 #ifdef TESTING
281 ✗ FAIL("The FlowExecutor should not have a timeout when all nodes are finished already.");
282 #endif
283 ✗ break;
284 }
285 120 }
286 }
287
288 // Deinitialize
289
1/2
✓ Branch 3 taken 113 times.
✗ Branch 4 not taken.
 113 LOG_DEBUG("Stopping FlowExecutor...");
290 113 nm::DisableAllCallbacks();
291 113 nm::ClearAllNodeQueues();
292
293
3/4
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
✓ Branch 8 taken 355 times.
✓ Branch 9 taken 113 times.
 468 for (Node* node : nm::m_Nodes())
294 {
295
8/10
✓ Branch 0 taken 355 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 351 times.
✓ Branch 4 taken 4 times.
✓ Branch 6 taken 351 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 1 times.
✓ Branch 9 taken 350 times.
✓ Branch 10 taken 5 times.
✓ Branch 11 taken 350 times.
 355 if (node == nullptr || node->kind == Node::Kind::GroupBox || !node->isInitialized()) { continue; }
296
297 350 node->_mode = Node::Mode::REAL_TIME;
298
2/2
✓ Branch 5 taken 292 times.
✓ Branch 6 taken 350 times.
 642 for (auto& outputPin : node->outputPins)
299 {
300 292 outputPin.noMoreDataAvailable = true;
301 }
302
1/2
✓ Branch 1 taken 350 times.
✗ Branch 2 not taken.
 350 node->flush();
303 }
304
305
4/10
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 113 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 112 times.
✓ Branch 7 taken 1 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
✗ Branch 11 not taken.
✗ Branch 12 not taken.
 339 if (!ConfigManager::Get<bool>("nogui")
306
14/34
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 113 times.
✗ Branch 4 not taken.
✓ Branch 6 taken 113 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 113 times.
✗ Branch 10 not taken.
✓ Branch 11 taken 113 times.
✗ Branch 12 not taken.
✓ Branch 14 taken 113 times.
✗ Branch 15 not taken.
✓ Branch 17 taken 113 times.
✗ Branch 18 not taken.
✓ Branch 19 taken 112 times.
✓ Branch 20 taken 1 times.
✓ Branch 21 taken 113 times.
✗ Branch 22 not taken.
✓ Branch 24 taken 113 times.
✗ Branch 25 not taken.
✓ Branch 26 taken 113 times.
✗ Branch 27 not taken.
✓ Branch 29 taken 113 times.
✗ Branch 30 not taken.
✓ Branch 31 taken 113 times.
✗ Branch 32 not taken.
✗ Branch 33 not taken.
✗ Branch 34 not taken.
✗ Branch 36 not taken.
✗ Branch 37 not taken.
✗ Branch 38 not taken.
✗ Branch 39 not taken.
✗ Branch 41 not taken.
✗ Branch 42 not taken.
 678 || (!ConfigManager::Get<bool>("sigterm") && !ConfigManager::Get<size_t>("duration")))
307 {
308 112 auto finish = std::chrono::steady_clock::now();
309
2/4
✓ Branch 1 taken 112 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 112 times.
✗ Branch 5 not taken.
 112 [[maybe_unused]] std::chrono::duration<double> elapsed = finish - _startTime;
310
2/4
✓ Branch 1 taken 112 times.
✗ Branch 2 not taken.
✓ Branch 6 taken 112 times.
✗ Branch 7 not taken.
 112 LOG_INFO("Elapsed time: {:.2f} s", elapsed.count());
311 }
312
313 113 _activeNodes = 0;
314 LOG_TRACE("FlowExecutor deinitialized.");
315 {
316
1/2
✓ Branch 1 taken 113 times.
✗ Branch 2 not taken.
 113 std::scoped_lock<std::mutex> lk(_mutex);
317 113 _state = State::Idle;
318 113 _cv.notify_all();
319 113 }
320
321 LOG_TRACE("Execute thread finished.");
322 }
323