Branch data Line data Source code
1 : : /*
2 : : * Copyright (C) 2026 by Thun Lu. All rights reserved.
3 : : * Author: Thun Lu <thun.lu@zohomail.cn>
4 : : * Repo: https://github.com/thun-res/vlink
5 : : * _ __ __ _ __
6 : : * | | / / / / (_) ____ / /__
7 : : * | | / / / / / / / __ \ / //_/
8 : : * | |/ / / /___ / / / / / / / ,<
9 : : * |___/ /_____/ /_/ /_/ /_/ /_/|_|
10 : : *
11 : : * Licensed under the Apache License, Version 2.0 (the "License");
12 : : * you may not use this file except in compliance with the License.
13 : : * You may obtain a copy of the License at
14 : : *
15 : : * http://www.apache.org/licenses/LICENSE-2.0
16 : : *
17 : : * Unless required by applicable law or agreed to in writing, software
18 : : * distributed under the License is distributed on an "AS IS" BASIS,
19 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 : : * See the License for the specific language governing permissions and
21 : : * limitations under the License.
22 : : */
23 : :
24 : : #include "./base/graph_task.h"
25 : :
26 : : #include <algorithm>
27 : : #include <atomic>
28 : : #include <cstddef>
29 : : #include <limits>
30 : : #include <memory>
31 : : #include <mutex>
32 : : #include <new>
33 : : #include <shared_mutex>
34 : : #include <sstream>
35 : : #include <stack>
36 : : #include <string>
37 : : #include <unordered_map>
38 : : #include <unordered_set>
39 : : #include <utility>
40 : : #include <vector>
41 : :
42 : : #include "./base/condition_variable.h"
43 : : #include "./base/helpers.h"
44 : : #include "./base/logger.h"
45 : : #include "./base/memory_resource.h"
46 : :
47 : : namespace vlink {
48 : :
49 : : static std::atomic<uint32_t> global_graph_task_count = 0;
50 : :
51 : 78 : static std::recursive_mutex& topology_mutex() {
52 : : static std::recursive_mutex mtx;
53 : 78 : return mtx;
54 : : }
55 : :
56 : : // GraphTask::Impl
57 : : struct GraphTask::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
58 : : alignas(64) std::atomic<size_t> pending_index{0};
59 : : alignas(64) std::atomic<size_t> active_index{0};
60 : : alignas(64) std::atomic<bool> is_ready{false};
61 : : std::atomic<bool> is_enable{false};
62 : : std::atomic<GraphTask::Status> status{GraphTask::kStatusInActive};
63 : :
64 : : std::atomic<uint32_t> max_recursion_depth{10'000};
65 : : std::atomic<GraphTask::Policy> policy{GraphTask::kPolicyOnce};
66 : : std::atomic<uint16_t> priority{100};
67 : : std::atomic<int> condition_number{0};
68 : :
69 : : std::vector<std::weak_ptr<GraphTask>> precede_task_list;
70 : : std::vector<std::weak_ptr<GraphTask>> succeed_task_list;
71 : :
72 : : mutable std::mutex mtx;
73 : : vlink::ConditionVariable cv;
74 : :
75 : : std::shared_mutex shared_mtx;
76 : :
77 : : std::string name;
78 : : std::string group_name;
79 : :
80 : : GraphTask::Callback callback;
81 : : GraphTask::ConditionCallback condition_callback;
82 : :
83 : : std::mutex status_callbacks_mtx;
84 : : std::atomic<uint32_t> next_status_callback_id{1};
85 : : std::unordered_map<uint32_t, std::shared_ptr<GraphTask::StatusCallback>> status_callbacks;
86 : :
87 : : bool is_condition_task{false};
88 : : };
89 : :
90 : : template <typename TypeT>
91 : : struct GraphTask::SharedAllocator {
92 : : using value_type = TypeT;
93 : : using is_always_equal = std::true_type;
94 : :
95 : : SharedAllocator() noexcept = default;
96 : :
97 : : template <typename OtherT>
98 : : // NOLINTNEXTLINE(google-explicit-constructor)
99 : 230 : SharedAllocator(const SharedAllocator<OtherT>&) noexcept {}
100 : :
101 [ + - ]: 115 : [[nodiscard]] TypeT* allocate(std::size_t count) { return std::allocator<TypeT>{}.allocate(count); }
102 : :
103 : 115 : void deallocate(TypeT* ptr, std::size_t count) noexcept { std::allocator<TypeT>{}.deallocate(ptr, count); }
104 : :
105 : : template <typename OtherT, typename... ArgsT>
106 : 115 : void construct(OtherT* ptr, ArgsT&&... args) {
107 [ + - ]: 115 : ::new (static_cast<void*>(ptr)) OtherT(std::forward<ArgsT>(args)...);
108 : 115 : }
109 : :
110 : : template <typename OtherT>
111 : 115 : void destroy(OtherT* ptr) noexcept {
112 : 115 : ptr->~OtherT();
113 : 115 : }
114 : :
115 : : template <typename OtherT>
116 : : // NOLINTNEXTLINE(readability-identifier-naming)
117 : : struct rebind {
118 : : using other = SharedAllocator<OtherT>;
119 : : };
120 : :
121 : : template <typename OtherT>
122 : : constexpr bool operator==(const SharedAllocator<OtherT>&) const noexcept {
123 : : return true;
124 : : }
125 : :
126 : : template <typename OtherT>
127 : : constexpr bool operator!=(const SharedAllocator<OtherT>&) const noexcept {
128 : : return false;
129 : : }
130 : : };
131 : :
132 : : // GraphTask
133 : 5 : std::shared_ptr<GraphTask> GraphTask::create(Callback&& callback, int condition_number) {
134 : 10 : return std::allocate_shared<GraphTask>(SharedAllocator<GraphTask>{}, PrivateToken{}, std::move(callback),
135 [ + - ]: 10 : condition_number);
136 : : }
137 : :
138 : 100 : std::shared_ptr<GraphTask> GraphTask::create(const std::string& name, Callback&& callback, int condition_number) {
139 : 200 : return std::allocate_shared<GraphTask>(SharedAllocator<GraphTask>{}, PrivateToken{}, name, std::move(callback),
140 [ + - ]: 200 : condition_number);
141 : : }
142 : :
143 : 2 : std::shared_ptr<GraphTask> GraphTask::create_condition(ConditionCallback&& callback, int condition_number) {
144 : 4 : return std::allocate_shared<GraphTask>(SharedAllocator<GraphTask>{}, PrivateToken{}, std::move(callback),
145 [ + - ]: 4 : condition_number);
146 : : }
147 : :
148 : 8 : std::shared_ptr<GraphTask> GraphTask::create_condition(const std::string& name, ConditionCallback&& callback,
149 : : int condition_number) {
150 : 16 : return std::allocate_shared<GraphTask>(SharedAllocator<GraphTask>{}, PrivateToken{}, name, std::move(callback),
151 [ + - ]: 16 : condition_number);
152 : : }
153 : :
154 : 4 : void GraphTask::cancel() {
155 [ + - ]: 4 : std::lock_guard topology_lock(topology_mutex());
156 : :
157 [ + - ]: 4 : std::stack<std::shared_ptr<GraphTask>> task_stack;
158 [ + - + - ]: 4 : task_stack.emplace(shared_from_this());
159 : :
160 [ + + ]: 10 : while (!task_stack.empty()) {
161 : 6 : auto current_task = task_stack.top();
162 : 6 : task_stack.pop();
163 : :
164 : 6 : std::vector<std::weak_ptr<GraphTask>> succ_snapshot;
165 : :
166 : : {
167 [ + - ]: 6 : std::lock_guard lock(current_task->impl_->mtx);
168 : :
169 [ + + ]: 6 : if (current_task->impl_->status.load(std::memory_order_acquire) == kStatusInActive) {
170 : 2 : continue;
171 : : }
172 : :
173 [ + - ]: 4 : succ_snapshot = current_task->impl_->succeed_task_list;
174 [ + + ]: 6 : }
175 : :
176 [ + - ]: 4 : current_task->update_status(kStatusInActive);
177 [ + - ]: 4 : current_task->mark_ready(false);
178 : :
179 [ + + ]: 6 : for (const auto& weak_task : succ_snapshot) {
180 [ + - ]: 2 : if (auto task_ptr = weak_task.lock()) {
181 [ + - ]: 2 : task_stack.emplace(task_ptr);
182 : 2 : }
183 : : }
184 [ + + + + ]: 8 : }
185 : 4 : }
186 : :
187 : 59 : void GraphTask::precede(const std::shared_ptr<GraphTask>& task) {
188 [ + + + + : 59 : if VUNLIKELY (!task || task.get() == this) {
+ + ]
189 [ + - - + ]: 4 : VLOG_F("GraphTask: Invalid task for precede.");
190 : 2 : return;
191 : : }
192 : :
193 [ + - ]: 57 : std::lock_guard topology_lock(topology_mutex());
194 : :
195 : 57 : std::unique_lock lock1(this->impl_->mtx, std::defer_lock);
196 : 57 : std::unique_lock lock2(task->impl_->mtx, std::defer_lock);
197 [ + - ]: 57 : std::lock(lock1, lock2);
198 : :
199 [ + - ]: 57 : if (std::find_if(impl_->succeed_task_list.begin(), impl_->succeed_task_list.end(), [&task](const auto& weak_task) {
200 : 13 : return weak_task.lock() == task;
201 [ + + ]: 114 : }) != impl_->succeed_task_list.end()) {
202 [ + - - + ]: 2 : VLOG_F("GraphTask: Task already added.");
203 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
204 : : }
205 : :
206 [ + - + + ]: 56 : if VUNLIKELY (reaches_via_successors(task.get(), task->impl_->succeed_task_list, this)) {
207 [ + - + - ]: 4 : VLOG_E("GraphTask: precede would create a cycle; edge rejected.");
208 : 2 : return;
209 : : }
210 : :
211 [ + - ]: 54 : impl_->succeed_task_list.emplace_back(task);
212 [ + - + - ]: 54 : task->impl_->precede_task_list.emplace_back(shared_from_this());
213 [ + + + + : 63 : }
+ + ]
214 : :
215 : 14 : void GraphTask::succeed(const std::shared_ptr<GraphTask>& task) {
216 [ + + - + : 14 : if VUNLIKELY (!task || task.get() == this) {
+ + ]
217 [ + - - + ]: 2 : VLOG_F("GraphTask: Invalid task for succeed.");
218 : 1 : return;
219 : : }
220 : :
221 [ + - ]: 13 : std::lock_guard topology_lock(topology_mutex());
222 : :
223 : 13 : std::unique_lock lock1(this->impl_->mtx, std::defer_lock);
224 : 13 : std::unique_lock lock2(task->impl_->mtx, std::defer_lock);
225 [ + - ]: 13 : std::lock(lock1, lock2);
226 : :
227 [ + - ]: 13 : if (std::find_if(impl_->precede_task_list.begin(), impl_->precede_task_list.end(), [&task](const auto& weak_task) {
228 : 3 : return weak_task.lock() == task;
229 [ + + ]: 26 : }) != impl_->precede_task_list.end()) {
230 [ + - - + ]: 2 : VLOG_F("GraphTask: Task already added.");
231 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
232 : : }
233 : :
234 [ + - + + ]: 12 : if VUNLIKELY (reaches_via_successors(this, impl_->succeed_task_list, task.get())) {
235 [ + - + - ]: 2 : VLOG_E("GraphTask: succeed would create a cycle; edge rejected.");
236 : 1 : return;
237 : : }
238 : :
239 [ + - ]: 11 : impl_->precede_task_list.emplace_back(task);
240 [ + - + - ]: 11 : task->impl_->succeed_task_list.emplace_back(shared_from_this());
241 [ + + + + : 17 : }
+ + ]
242 : :
243 : 267 : uint32_t GraphTask::register_status_callback(StatusCallback&& callback) {
244 [ + + ]: 267 : if VUNLIKELY (!callback) {
245 : 1 : return 0;
246 : : }
247 : :
248 [ + - ]: 266 : std::lock_guard lock(impl_->status_callbacks_mtx);
249 : :
250 : 266 : uint32_t id = 0;
251 : :
252 [ + - ]: 266 : for (uint32_t attempts = 0; attempts < std::numeric_limits<uint32_t>::max();
253 : : ++attempts) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
254 : 266 : id = impl_->next_status_callback_id.fetch_add(1, std::memory_order_relaxed);
255 : :
256 [ - + ]: 266 : if VUNLIKELY (id == 0) {
257 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
258 : : }
259 : :
260 [ + - + - ]: 266 : if VLIKELY (impl_->status_callbacks.find(id) == impl_->status_callbacks.end()) {
261 : 266 : break;
262 : : }
263 : : }
264 : :
265 [ - + ]: 266 : if VUNLIKELY (id == 0) {
266 : : VLOG_E("GraphTask: status_callback id space exhausted."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
267 : : return 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
268 : : }
269 : :
270 [ + - + - ]: 266 : impl_->status_callbacks.emplace(id, MemoryResource::make_shared<StatusCallback>(std::move(callback)));
271 : :
272 : 266 : return id;
273 : 266 : }
274 : :
275 : 258 : bool GraphTask::unregister_status_callback(uint32_t id) {
276 [ + - ]: 258 : std::lock_guard lock(impl_->status_callbacks_mtx);
277 [ + - ]: 516 : return impl_->status_callbacks.erase(id) > 0;
278 : 258 : }
279 : :
280 : 1 : void GraphTask::clear_status_callbacks() {
281 [ + - ]: 1 : std::lock_guard lock(impl_->status_callbacks_mtx);
282 : 1 : impl_->status_callbacks.clear();
283 : 1 : }
284 : :
285 : 1 : void GraphTask::set_name(const std::string& name) {
286 [ + - ]: 1 : std::lock_guard lock(impl_->shared_mtx);
287 [ + - ]: 1 : impl_->name = name;
288 : 1 : }
289 : :
290 : 3 : void GraphTask::set_group_name(const std::string& name) {
291 [ + - ]: 3 : std::lock_guard lock(impl_->shared_mtx);
292 [ + - ]: 3 : impl_->group_name = name;
293 : 3 : }
294 : :
295 : 6 : void GraphTask::set_condition_number(int condition_number) {
296 : 6 : impl_->condition_number.store(condition_number, std::memory_order_relaxed);
297 : 6 : }
298 : :
299 : 1 : void GraphTask::set_priority(uint16_t priority) { impl_->priority.store(priority, std::memory_order_relaxed); }
300 : :
301 : 2 : void GraphTask::set_max_recursion_depth(uint32_t depth) {
302 : 2 : impl_->max_recursion_depth.store(depth, std::memory_order_relaxed);
303 : 2 : }
304 : :
305 : 8 : void GraphTask::set_policy(Policy policy) { impl_->policy.store(policy, std::memory_order_relaxed); }
306 : :
307 : 11 : std::string GraphTask::get_name() const {
308 [ + - ]: 11 : std::shared_lock lock(impl_->shared_mtx);
309 [ + - ]: 22 : return impl_->name;
310 : 11 : }
311 : :
312 : 7 : std::string GraphTask::get_group_name() const {
313 [ + - ]: 7 : std::shared_lock lock(impl_->shared_mtx);
314 [ + - ]: 14 : return impl_->group_name;
315 : 7 : }
316 : :
317 : 8 : int GraphTask::get_condition_number() const { return impl_->condition_number.load(std::memory_order_relaxed); }
318 : :
319 : 2 : uint16_t GraphTask::get_priority() const { return impl_->priority.load(std::memory_order_relaxed); }
320 : :
321 : 1 : uint32_t GraphTask::get_max_recursion_depth() const {
322 : 2 : return impl_->max_recursion_depth.load(std::memory_order_relaxed);
323 : : }
324 : :
325 : 3 : GraphTask::Policy GraphTask::get_policy() const { return impl_->policy.load(std::memory_order_relaxed); }
326 : :
327 : 81 : GraphTask::Status GraphTask::get_status() const { return impl_->status.load(std::memory_order_acquire); }
328 : :
329 : 3 : void GraphTask::remove_precede_task(const std::shared_ptr<GraphTask>& task) {
330 [ + + ]: 3 : if VUNLIKELY (!task) {
331 [ + - - + ]: 2 : VLOG_F("GraphTask: Invalid task provided to remove_precede_task.");
332 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
333 : : }
334 : :
335 [ + - ]: 2 : std::lock_guard topology_lock(topology_mutex());
336 : :
337 : 2 : std::unique_lock lock1(this->impl_->mtx, std::defer_lock);
338 : 2 : std::unique_lock lock2(task->impl_->mtx, std::defer_lock);
339 [ + - ]: 2 : std::lock(lock1, lock2);
340 : :
341 [ + - ]: 2 : auto iter_succeed = std::remove_if(impl_->succeed_task_list.begin(), impl_->succeed_task_list.end(),
342 : 4 : [&task](const std::weak_ptr<GraphTask>& weak_task) {
343 : 2 : auto locked_task = weak_task.lock();
344 : 2 : return locked_task == task;
345 : 2 : });
346 : :
347 [ + + ]: 2 : if VLIKELY (iter_succeed != impl_->succeed_task_list.end()) {
348 [ + - ]: 1 : impl_->succeed_task_list.erase(iter_succeed, impl_->succeed_task_list.end());
349 : : } else {
350 [ + - - + ]: 2 : VLOG_F("GraphTask: Task not found in succeed_task_list.");
351 : : }
352 : :
353 [ + - ]: 1 : auto iter_precede = std::remove_if(task->impl_->precede_task_list.begin(), task->impl_->precede_task_list.end(),
354 : 2 : [this](const std::weak_ptr<GraphTask>& weak_task) {
355 : 1 : auto locked_task = weak_task.lock();
356 : 1 : return locked_task.get() == this;
357 : 1 : });
358 : :
359 [ + - ]: 1 : if VLIKELY (iter_precede != task->impl_->precede_task_list.end()) {
360 [ + - ]: 1 : task->impl_->precede_task_list.erase(iter_precede, task->impl_->precede_task_list.end());
361 : : } else {
362 : : VLOG_F("GraphTask: Current task not found in task's precede_task_list."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
363 : : }
364 : 4 : }
365 : :
366 : 3 : void GraphTask::remove_succeed_task(const std::shared_ptr<GraphTask>& task) {
367 [ + + ]: 3 : if VUNLIKELY (!task) {
368 [ + - - + ]: 2 : VLOG_F("GraphTask: Invalid task provided to remove_succeed_task.");
369 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
370 : : }
371 : :
372 [ + - ]: 2 : std::lock_guard topology_lock(topology_mutex());
373 : :
374 : 2 : std::unique_lock lock1(this->impl_->mtx, std::defer_lock);
375 : 2 : std::unique_lock lock2(task->impl_->mtx, std::defer_lock);
376 [ + - ]: 2 : std::lock(lock1, lock2);
377 : :
378 [ + - ]: 2 : auto iter_precede = std::remove_if(impl_->precede_task_list.begin(), impl_->precede_task_list.end(),
379 : 2 : [&task](const auto& weak_task) { return weak_task.lock() == task; });
380 : :
381 [ + + ]: 2 : if (iter_precede != impl_->precede_task_list.end()) {
382 [ + - ]: 1 : impl_->precede_task_list.erase(iter_precede, impl_->precede_task_list.end());
383 : : }
384 : :
385 [ + - ]: 2 : auto iter_succeed = std::remove_if(task->impl_->succeed_task_list.begin(), task->impl_->succeed_task_list.end(),
386 : 1 : [this](const auto& weak_task) { return weak_task.lock().get() == this; });
387 : :
388 [ + + ]: 2 : if (iter_succeed != task->impl_->succeed_task_list.end()) {
389 [ + - ]: 1 : task->impl_->succeed_task_list.erase(iter_succeed, task->impl_->succeed_task_list.end());
390 : : }
391 : 2 : }
392 : :
393 : 9 : std::vector<std::weak_ptr<GraphTask>> GraphTask::get_precede_task_list() const {
394 [ + - ]: 9 : std::lock_guard lock(impl_->mtx);
395 [ + - ]: 18 : return impl_->precede_task_list;
396 : 9 : }
397 : :
398 : 20 : std::vector<std::weak_ptr<GraphTask>> GraphTask::get_succeed_task_list() const {
399 [ + - ]: 20 : std::lock_guard lock(impl_->mtx);
400 [ + - ]: 40 : return impl_->succeed_task_list;
401 : 20 : }
402 : :
403 : 13 : bool GraphTask::is_condition_task() const { return impl_->is_condition_task; }
404 : :
405 [ + - ]: 5 : GraphTask::GraphTask(PrivateToken, Callback&& callback, int condition_number) : impl_(std::make_unique<Impl>()) {
406 [ + - + - ]: 5 : impl_->name = "Task_" + std::to_string(global_graph_task_count.fetch_add(1, std::memory_order_relaxed));
407 : 5 : impl_->condition_number.store(condition_number, std::memory_order_relaxed);
408 : 5 : impl_->is_condition_task = false;
409 : 5 : impl_->callback = std::move(callback);
410 : 5 : }
411 : :
412 : 100 : GraphTask::GraphTask(PrivateToken, const std::string& name, Callback&& callback, int condition_number)
413 [ + - ]: 100 : : impl_(std::make_unique<Impl>()) {
414 [ + - ]: 100 : impl_->name = name;
415 : 100 : impl_->condition_number.store(condition_number, std::memory_order_relaxed);
416 : 100 : impl_->is_condition_task = false;
417 : 100 : impl_->callback = std::move(callback);
418 : 100 : }
419 : :
420 : 2 : GraphTask::GraphTask(PrivateToken, ConditionCallback&& callback, int condition_number)
421 [ + - ]: 2 : : impl_(std::make_unique<Impl>()) {
422 [ + - + - ]: 2 : impl_->name = "Task_" + std::to_string(global_graph_task_count.fetch_add(1, std::memory_order_relaxed));
423 : 2 : impl_->condition_number.store(condition_number, std::memory_order_relaxed);
424 : 2 : impl_->is_condition_task = true;
425 : 2 : impl_->condition_callback = std::move(callback);
426 : 2 : }
427 : :
428 : 8 : GraphTask::GraphTask(PrivateToken, const std::string& name, ConditionCallback&& callback, int condition_number)
429 [ + - ]: 8 : : impl_(std::make_unique<Impl>()) {
430 [ + - ]: 8 : impl_->name = name;
431 : 8 : impl_->condition_number.store(condition_number, std::memory_order_relaxed);
432 : 8 : impl_->is_condition_task = true;
433 : 8 : impl_->condition_callback = std::move(callback);
434 : 8 : }
435 : :
436 : 115 : GraphTask::~GraphTask() = default;
437 : :
438 : 24 : void GraphTask::process_and_traverse(FindTaskCallback&& callback) {
439 : 24 : uint32_t recursion_count = 0;
440 : :
441 [ + - ]: 24 : std::stack<std::shared_ptr<GraphTask>> task_stack;
442 : :
443 [ + - + - ]: 24 : task_stack.emplace(shared_from_this());
444 : :
445 : 24 : std::unordered_map<GraphTask*, int> pending_count_map;
446 : 24 : std::unordered_map<GraphTask*, std::vector<std::shared_ptr<GraphTask>>> successor_map;
447 : 24 : std::unordered_set<GraphTask*> processed;
448 : :
449 : 24 : std::vector<std::shared_ptr<GraphTask>> top_task_list;
450 : :
451 [ + + ]: 83 : while (!task_stack.empty()) {
452 : 59 : auto current_task = task_stack.top();
453 : 59 : task_stack.pop();
454 : :
455 [ + - - + ]: 59 : if (!processed.insert(current_task.get()).second) {
456 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
457 : : }
458 : :
459 : : {
460 [ + - ]: 59 : std::lock_guard lock(current_task->impl_->mtx);
461 : :
462 [ + - ]: 59 : clear_invalid_task(current_task);
463 : :
464 [ + + ]: 59 : if (recursion_count == 0) {
465 : 24 : current_task->impl_->is_ready.store(true, std::memory_order_release);
466 : 24 : current_task->impl_->is_enable.store(true, std::memory_order_release);
467 : 24 : current_task->impl_->active_index.store(0U, std::memory_order_release);
468 : :
469 [ + - ]: 24 : auto& sub_pending_count = pending_count_map[current_task.get()];
470 : 24 : current_task->impl_->pending_index.store(++sub_pending_count, std::memory_order_release);
471 : :
472 [ + - ]: 24 : top_task_list.emplace_back(current_task);
473 : : }
474 : :
475 [ + + ]: 98 : for (const auto& task : current_task->impl_->succeed_task_list) {
476 : 39 : auto task_ptr = task.lock();
477 : :
478 [ - + ]: 39 : if VUNLIKELY (!task_ptr) {
479 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
480 : : }
481 : :
482 [ + - + - ]: 39 : successor_map[current_task.get()].emplace_back(task_ptr);
483 : :
484 [ + - ]: 39 : bool first_seen = (pending_count_map.find(task_ptr.get()) == pending_count_map.end());
485 : :
486 [ + - ]: 39 : auto& sub_pending_count = pending_count_map[task_ptr.get()];
487 : 39 : task_ptr->impl_->pending_index.store(++sub_pending_count, std::memory_order_release);
488 : :
489 [ + + ]: 39 : if (first_seen) {
490 : 35 : task_ptr->impl_->is_ready.store(false, std::memory_order_release);
491 : 35 : task_ptr->impl_->is_enable.store(false, std::memory_order_release);
492 : 35 : task_ptr->impl_->active_index.store(0U, std::memory_order_release);
493 : :
494 [ + - ]: 35 : top_task_list.emplace_back(task_ptr);
495 [ + - ]: 35 : task_stack.emplace(task_ptr);
496 : : }
497 : :
498 [ - + ]: 78 : if VUNLIKELY (recursion_count++ >= impl_->max_recursion_depth.load(std::memory_order_relaxed)) {
499 : : CLOG_F("GraphTask: Recursion detection exceeds the upper limit (%d).", // LCOV_EXCL_LINE GCOVR_EXCL_LINE
500 : : impl_->max_recursion_depth.load(std::memory_order_relaxed));
501 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
502 : : }
503 [ + - - ]: 39 : }
504 [ + - ]: 59 : }
505 [ + - - ]: 59 : }
506 : :
507 : 24 : std::vector<std::shared_ptr<GraphTask>> ready_task_list;
508 : 24 : std::vector<std::shared_ptr<GraphTask>> sorted_task_list;
509 [ + - ]: 24 : ready_task_list.reserve(top_task_list.size());
510 [ + - ]: 24 : sorted_task_list.reserve(top_task_list.size());
511 [ + - ]: 24 : ready_task_list.emplace_back(top_task_list.front());
512 : :
513 [ + + ]: 83 : for (size_t index = 0; index < ready_task_list.size(); ++index) {
514 : 59 : const auto& current_task = ready_task_list[index];
515 [ + - ]: 59 : sorted_task_list.emplace_back(current_task);
516 : :
517 [ + - + + ]: 98 : for (const auto& successor : successor_map[current_task.get()]) {
518 [ + - ]: 39 : auto pending_iter = pending_count_map.find(successor.get());
519 [ + - + + : 39 : if (pending_iter != pending_count_map.end() && --pending_iter->second == 0) {
+ + ]
520 [ + - ]: 35 : ready_task_list.emplace_back(successor);
521 : : }
522 : : }
523 : : }
524 : :
525 [ - + ]: 24 : if VUNLIKELY (sorted_task_list.size() != top_task_list.size()) {
526 : : CLOG_E("GraphTask: Failed to produce a topological task order."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
527 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
528 : : }
529 : :
530 : 24 : top_task_list = std::move(sorted_task_list);
531 : :
532 [ + + ]: 83 : for (const auto& top_task : top_task_list) {
533 [ + - ]: 59 : top_task->update_status(kStatusPending);
534 : : }
535 : :
536 [ + + ]: 83 : for (const auto& top_task : top_task_list) {
537 [ + - ]: 59 : callback(top_task);
538 : : }
539 [ + - + - : 24 : }
+ - + - +
- + - +
- ]
540 : :
541 : 6 : bool GraphTask::has_cycle() const {
542 : 6 : std::unordered_set<const GraphTask*> visited;
543 : 6 : std::unordered_set<const GraphTask*> recursion_stack;
544 : 6 : uint32_t depth = 0;
545 : 6 : const uint32_t max_depth = impl_->max_recursion_depth.load(std::memory_order_relaxed);
546 : :
547 [ + + ]: 11 : return detect_cycle(this, visited, recursion_stack, depth, max_depth);
548 : 7 : }
549 : :
550 : 68 : bool GraphTask::reaches_via_successors(const GraphTask* start_node,
551 : : const std::vector<std::weak_ptr<GraphTask>>& start_successors,
552 : : const GraphTask* target) const {
553 : 68 : std::unordered_set<const GraphTask*> visited;
554 [ + - ]: 68 : visited.insert(start_node);
555 [ + - ]: 68 : visited.insert(target);
556 : :
557 [ + - ]: 68 : std::stack<std::shared_ptr<GraphTask>> stack;
558 : :
559 [ + + ]: 75 : for (const auto& w : start_successors) {
560 : 7 : auto p = w.lock();
561 : :
562 [ - + ]: 7 : if VUNLIKELY (!p) {
563 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
564 : : }
565 : :
566 [ - + ]: 7 : if (p.get() == target) {
567 : : return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
568 : : }
569 : :
570 [ + - + - ]: 7 : if (visited.insert(p.get()).second) {
571 [ + - ]: 7 : stack.push(std::move(p));
572 : : }
573 [ + - - ]: 7 : }
574 : :
575 : 68 : const uint32_t max_depth = impl_->max_recursion_depth.load(std::memory_order_relaxed);
576 : 68 : uint32_t visit_count = 0;
577 : :
578 [ + + ]: 72 : while (!stack.empty()) {
579 : 7 : auto cur = std::move(stack.top());
580 : 7 : stack.pop();
581 : :
582 [ - + ]: 7 : if VUNLIKELY (++visit_count > max_depth) {
583 : : CLOG_F("GraphTask: reaches() exceeded max_recursion_depth (%u).", max_depth); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
584 : : return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
585 : : }
586 : :
587 : 7 : std::vector<std::weak_ptr<GraphTask>> succ_copy;
588 : : {
589 [ + - ]: 7 : std::lock_guard lock(cur->impl_->mtx);
590 [ + - ]: 7 : succ_copy = cur->impl_->succeed_task_list;
591 : 7 : }
592 : :
593 [ + + ]: 7 : for (const auto& w : succ_copy) {
594 : 3 : auto p = w.lock();
595 : :
596 [ - + ]: 3 : if VUNLIKELY (!p) {
597 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
598 : : }
599 : :
600 [ + - ]: 3 : if (p.get() == target) {
601 : 3 : return true;
602 : : }
603 : :
604 : : if (visited.insert(p.get()).second) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
605 : : stack.push(std::move(p)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
606 : : }
607 [ - - + ]: 3 : }
608 [ + + + + ]: 10 : }
609 : :
610 : 65 : return false;
611 : 68 : }
612 : :
613 : 3 : std::string GraphTask::export_to_dot() const {
614 [ + - ]: 3 : std::ostringstream dot_stream;
615 : :
616 [ + - ]: 3 : dot_stream << "digraph TaskGraph {\n";
617 : :
618 [ + - ]: 3 : dot_stream << " node [fontname=\"Arial\"];\n";
619 : :
620 : 3 : std::unordered_map<std::string, std::vector<const GraphTask*>> groups;
621 : :
622 : 3 : std::unordered_set<const GraphTask*> visited;
623 : :
624 : 6 : Function<void(const GraphTask*)> traverse = [&visited, &groups, &dot_stream,
625 : : &traverse](const GraphTask* task) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
626 [ - + ]: 6 : if (visited.count(task)) {
627 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
628 : : }
629 : :
630 : 6 : visited.insert(task);
631 : :
632 : : {
633 [ + - + - ]: 6 : groups[task->get_group_name()].emplace_back(task);
634 : : }
635 : :
636 : : {
637 [ + - ]: 6 : std::lock_guard lock(task->impl_->mtx);
638 : :
639 [ + + ]: 9 : for (const auto& succeed_task_weak : task->impl_->succeed_task_list) {
640 : 3 : auto succeed_task = succeed_task_weak.lock();
641 : :
642 [ - + ]: 3 : if VUNLIKELY (!succeed_task) {
643 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
644 : : }
645 : :
646 [ + - + + ]: 3 : if (task->is_condition_task()) {
647 : 2 : dot_stream << " \"" << task->impl_->name << "\" -> \"" << succeed_task->impl_->name
648 [ + - + - : 4 : << "\" [style=dashed, arrowhead=vee];\n";
+ - + - +
- ]
649 : : } else {
650 [ + - + - : 1 : dot_stream << " \"" << task->impl_->name << "\" -> \"" << succeed_task->impl_->name << "\";\n";
+ - + - +
- ]
651 : : }
652 : :
653 [ + - ]: 3 : traverse(succeed_task.get());
654 [ + - ]: 3 : }
655 : 6 : }
656 [ + - ]: 3 : };
657 : :
658 [ + - ]: 3 : traverse(this);
659 : :
660 : 3 : std::string title_label;
661 : 3 : std::string extra_label;
662 : 3 : std::string color_label;
663 : :
664 [ + + ]: 7 : for (const auto& [group_name, tasks] : groups) {
665 [ + + ]: 4 : if (!group_name.empty()) {
666 [ + - + - : 1 : dot_stream << " subgraph cluster_" << group_name << " {\n";
+ - ]
667 [ + - + - : 1 : dot_stream << " label = \"" << group_name << "\";\n";
+ - ]
668 [ + - ]: 1 : dot_stream << " style = filled;\n";
669 [ + - ]: 1 : dot_stream << " color = lightgray;\n";
670 : : }
671 : :
672 [ + + ]: 10 : for (const auto* task : tasks) {
673 [ + - ]: 6 : const std::string& name = task->get_name();
674 : :
675 [ + + ]: 6 : if (task->impl_->policy.load(std::memory_order_relaxed) == kPolicyOnce) {
676 : 4 : extra_label.clear();
677 [ + - ]: 4 : color_label = "lightgray";
678 [ + + ]: 2 : } else if (task->impl_->policy.load(std::memory_order_relaxed) == kPolicyMultiple) {
679 [ + - ]: 1 : extra_label = "\\n[Multiple]";
680 [ + - ]: 1 : color_label = "lightgreen";
681 [ + - ]: 1 : } else if (task->impl_->policy.load(std::memory_order_relaxed) ==
682 : : kPolicyWaitAll) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
683 [ + - ]: 1 : extra_label = "\\n[Waitfor]";
684 [ + - ]: 1 : color_label = "lightblue";
685 : : } else {
686 : : extra_label.clear(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
687 : : color_label = "lightgray"; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
688 : : }
689 : :
690 [ + - + + ]: 6 : if (task->is_condition_task()) {
691 [ + - ]: 1 : title_label = "style=dashed, shape=diamond, style=filled";
692 [ + - + - ]: 1 : Helpers::replace_string(color_label, "light", "");
693 : : } else {
694 [ + - ]: 5 : title_label = "style=ellipse, style=filled";
695 : : }
696 : :
697 [ + - ]: 6 : dot_stream << " \"";
698 [ + - ]: 6 : dot_stream << name;
699 [ + - ]: 6 : dot_stream << "\" [";
700 [ + - ]: 6 : dot_stream << title_label;
701 [ + - ]: 6 : dot_stream << ", color=";
702 [ + - ]: 6 : dot_stream << color_label;
703 [ + - ]: 6 : dot_stream << ", label=\"";
704 [ + - ]: 6 : dot_stream << name;
705 [ + - ]: 6 : dot_stream << extra_label;
706 [ + - ]: 6 : dot_stream << "\"];\n";
707 : 6 : }
708 : :
709 [ + + ]: 4 : if (!group_name.empty()) {
710 [ + - ]: 1 : dot_stream << " }\n";
711 : : }
712 : : }
713 : :
714 [ + - ]: 3 : dot_stream << "}\n";
715 : :
716 [ + - ]: 6 : return dot_stream.str();
717 : 3 : }
718 : :
719 : 59 : int GraphTask::invoke(bool once) {
720 [ + + ]: 59 : if (once) {
721 [ + + - + : 106 : if (!impl_->is_enable.load(std::memory_order_acquire) ||
+ + ]
722 : 48 : impl_->status.load(std::memory_order_acquire) != kStatusPending) {
723 : 10 : return -1;
724 : : }
725 : : }
726 : :
727 [ + - ]: 49 : update_status(kStatusRunning);
728 : :
729 : 49 : std::string name_copy;
730 : :
731 : : {
732 [ + - ]: 49 : std::shared_lock lock(impl_->shared_mtx);
733 : :
734 [ + - ]: 49 : name_copy = impl_->name;
735 : 49 : }
736 : :
737 [ + + + - : 49 : if (!impl_->is_condition_task && impl_->callback) {
+ + ]
738 : : try {
739 [ + + ]: 43 : impl_->callback();
740 [ + + ]: 2 : } catch (const std::exception& e) {
741 [ + - + - ]: 2 : CLOG_E("GraphTask: callback (%s) threw an exception: %s.", name_copy.c_str(), e.what());
742 : 2 : } catch (...) {
743 [ + - + - ]: 2 : CLOG_E("GraphTask: callback (%s) threw a non-std exception.", name_copy.c_str());
744 [ + - ]: 1 : }
745 : :
746 [ + - ]: 43 : update_status(kStatusDone);
747 : :
748 : 43 : return 0;
749 : : }
750 : :
751 [ + - + - : 6 : if (impl_->is_condition_task && impl_->condition_callback) {
+ - ]
752 : 6 : int ret = 0;
753 : 6 : bool failed = false;
754 : :
755 : : try {
756 [ + + ]: 6 : ret = impl_->condition_callback();
757 [ + + ]: 2 : } catch (const std::exception& e) {
758 : 1 : failed = true;
759 [ + - + - ]: 2 : CLOG_E("GraphTask: condition_callback (%s) threw an exception: %s.", name_copy.c_str(), e.what());
760 : 2 : } catch (...) {
761 : 1 : failed = true;
762 [ + - + - ]: 2 : CLOG_E("GraphTask: condition_callback (%s) threw a non-std exception.", name_copy.c_str());
763 [ + - ]: 1 : }
764 : :
765 [ + - ]: 6 : update_status(kStatusDone);
766 : :
767 [ + + ]: 6 : if (failed) {
768 : 2 : return std::numeric_limits<int>::max();
769 : : }
770 : :
771 : 4 : return ret;
772 : : }
773 : :
774 : : update_status(kStatusDone); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
775 : :
776 : : return -1; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
777 : 49 : }
778 : :
779 : 35 : void GraphTask::wait() {
780 [ + - ]: 35 : std::unique_lock lock(impl_->mtx);
781 : :
782 : 35 : impl_->cv.wait(lock, [this]() -> bool {
783 [ - + - - ]: 35 : return impl_->is_ready.load(std::memory_order_acquire) ||
784 : 35 : impl_->status.load(std::memory_order_acquire) == kStatusInActive;
785 : : });
786 : 35 : }
787 : :
788 : 48 : void GraphTask::notify(int condition_number) {
789 : 48 : std::vector<std::weak_ptr<GraphTask>> invoke_list;
790 : 48 : std::vector<std::shared_ptr<GraphTask>> skip_list;
791 : :
792 : : {
793 [ + - ]: 48 : std::lock_guard lock(impl_->mtx);
794 : :
795 [ + + ]: 86 : for (const auto& task : impl_->succeed_task_list) {
796 : 38 : auto task_ptr = task.lock();
797 : :
798 [ - + ]: 38 : if VUNLIKELY (!task_ptr) {
799 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
800 : : }
801 : :
802 [ + + ]: 76 : if (condition_number != task_ptr->impl_->condition_number.load(std::memory_order_relaxed)) {
803 [ + - ]: 8 : if (impl_->is_condition_task) {
804 : 8 : bool has_active = false;
805 [ + - ]: 8 : const bool ready = task_ptr->mark_predecessor_satisfied(false, &has_active);
806 : :
807 [ + - + - ]: 8 : if (ready && !has_active) {
808 [ + - ]: 8 : task_ptr->mark_ready(false);
809 [ + - ]: 8 : skip_list.emplace_back(task_ptr);
810 : : // LCOV_EXCL_START GCOVR_EXCL_START
811 : : } else if (ready && task_ptr->impl_->policy.load(std::memory_order_relaxed) == kPolicyWaitAll) {
812 : : task_ptr->mark_ready(true);
813 : : }
814 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
815 : : }
816 : :
817 : 8 : continue;
818 : 8 : }
819 : :
820 [ + + ]: 30 : if (task_ptr->impl_->policy.load(std::memory_order_relaxed) == kPolicyOnce) {
821 [ + - ]: 22 : task_ptr->mark_predecessor_satisfied(true, nullptr);
822 [ + - ]: 22 : task_ptr->mark_ready(true);
823 [ + + ]: 8 : } else if (task_ptr->impl_->policy.load(std::memory_order_relaxed) == kPolicyMultiple) {
824 [ + - ]: 1 : task_ptr->mark_predecessor_satisfied(true, nullptr);
825 [ + - ]: 1 : task_ptr->mark_ready(false);
826 : :
827 [ + - ]: 1 : invoke_list.emplace_back(task);
828 [ + - ]: 7 : } else if (task_ptr->impl_->policy.load(std::memory_order_relaxed) ==
829 : : kPolicyWaitAll) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
830 : 7 : bool has_active = false;
831 : :
832 [ + - + + : 7 : if (!task_ptr->mark_predecessor_satisfied(true, &has_active) || !has_active) {
- + + + ]
833 : 4 : continue;
834 : : }
835 : :
836 [ + - ]: 3 : task_ptr->mark_ready(true);
837 : : }
838 [ + + ]: 38 : }
839 : 48 : }
840 : :
841 [ + + ]: 56 : for (const auto& task_ptr : skip_list) {
842 [ + - ]: 8 : task_ptr->update_status(kStatusInActive);
843 [ + - ]: 8 : task_ptr->notify_skip();
844 : : }
845 : :
846 [ + + ]: 49 : for (const auto& task : invoke_list) {
847 : 1 : auto task_ptr = task.lock();
848 : :
849 [ - + ]: 1 : if VUNLIKELY (!task_ptr) {
850 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
851 : : }
852 : :
853 [ + - ]: 1 : task_ptr->invoke(false);
854 [ + - ]: 1 : }
855 : 48 : }
856 : :
857 : 9 : void GraphTask::notify_skip() {
858 : 9 : std::vector<std::weak_ptr<GraphTask>> succ_snapshot;
859 : 9 : std::vector<std::shared_ptr<GraphTask>> skip_list;
860 : :
861 : : {
862 [ + - ]: 9 : std::lock_guard lock(impl_->mtx);
863 [ + - ]: 9 : succ_snapshot = impl_->succeed_task_list;
864 : 9 : }
865 : :
866 [ + + ]: 10 : for (const auto& task : succ_snapshot) {
867 : 1 : auto task_ptr = task.lock();
868 : :
869 [ - + ]: 1 : if VUNLIKELY (!task_ptr) {
870 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
871 : : }
872 : :
873 : 1 : bool has_active = false;
874 [ + - ]: 1 : const bool ready = task_ptr->mark_predecessor_satisfied(false, &has_active);
875 : :
876 [ - + ]: 1 : if (!ready) {
877 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
878 : : }
879 : :
880 [ + - ]: 1 : if (!has_active) {
881 [ + - ]: 1 : task_ptr->mark_ready(false);
882 [ + - ]: 1 : skip_list.emplace_back(task_ptr);
883 : : // LCOV_EXCL_START GCOVR_EXCL_START
884 : : } else if (task_ptr->impl_->policy.load(std::memory_order_relaxed) == kPolicyWaitAll) {
885 : : task_ptr->mark_ready(true);
886 : : }
887 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
888 [ + - ]: 1 : }
889 : :
890 [ + + ]: 10 : for (const auto& task_ptr : skip_list) {
891 [ + - ]: 1 : task_ptr->update_status(kStatusInActive);
892 [ + - ]: 1 : task_ptr->notify_skip();
893 : : }
894 : 9 : }
895 : :
896 : 39 : bool GraphTask::mark_predecessor_satisfied(bool active, bool* has_active) {
897 [ + + ]: 39 : if (active) {
898 : 30 : impl_->active_index.fetch_add(1U, std::memory_order_acq_rel);
899 : : }
900 : :
901 : 39 : size_t expected = impl_->pending_index.load(std::memory_order_acquire);
902 : :
903 [ + - ]: 39 : while (expected > 0U) {
904 : 39 : const size_t desired = expected - 1U;
905 : :
906 [ + - ]: 78 : if (impl_->pending_index.compare_exchange_weak(expected, desired,
907 : : std::memory_order_acq_rel, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
908 : : std::memory_order_acquire)) {
909 [ + + ]: 39 : if (has_active != nullptr) {
910 : 32 : *has_active = impl_->active_index.load(std::memory_order_acquire) > 0U;
911 : : }
912 : :
913 : 39 : return desired == 0U;
914 : : }
915 : : }
916 : :
917 : : if (has_active != nullptr) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
918 : : *has_active = impl_->active_index.load(std::memory_order_acquire) > 0U; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
919 : : }
920 : :
921 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
922 : : }
923 : :
924 : 39 : void GraphTask::mark_ready(bool enable) {
925 : : {
926 [ + - ]: 39 : std::lock_guard lock(impl_->mtx);
927 : 39 : impl_->is_ready.store(true, std::memory_order_release);
928 : 39 : impl_->is_enable.store(enable, std::memory_order_release);
929 : 39 : }
930 : :
931 : 39 : impl_->cv.notify_all();
932 : 39 : }
933 : :
934 : 170 : void GraphTask::update_status(Status status) {
935 [ - + ]: 170 : if VUNLIKELY (impl_->status.load(std::memory_order_acquire) == status) {
936 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
937 : : }
938 : :
939 : 170 : impl_->status.store(status, std::memory_order_release);
940 : :
941 : 170 : std::string name_copy;
942 : :
943 : : {
944 [ + - ]: 170 : std::shared_lock lock(impl_->shared_mtx);
945 : :
946 [ + - ]: 170 : name_copy = impl_->name;
947 : 170 : }
948 : :
949 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
950 [ + - ]: 170 : std::pmr::vector<std::shared_ptr<StatusCallback>> callbacks(&MemoryResource::global_instance());
951 : : #else
952 : : std::vector<std::shared_ptr<StatusCallback>> callbacks;
953 : : #endif
954 : :
955 : : {
956 [ + - ]: 170 : std::lock_guard lock(impl_->status_callbacks_mtx);
957 : :
958 [ + - ]: 170 : callbacks.reserve(impl_->status_callbacks.size());
959 : :
960 [ + + ]: 194 : for (auto& [id, cb] : impl_->status_callbacks) {
961 : : (void)id;
962 : :
963 [ + - ]: 24 : callbacks.emplace_back(cb);
964 : : }
965 : 170 : }
966 : :
967 [ + + ]: 194 : for (auto& cb : callbacks) {
968 [ + - - + : 24 : if VUNLIKELY (!cb || !(*cb)) {
- + ]
969 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
970 : : }
971 : :
972 : : try {
973 [ + + ]: 24 : (*cb)(name_copy, status);
974 [ + + ]: 6 : } catch (const std::exception& e) {
975 [ + - + - ]: 6 : CLOG_E("GraphTask: status_callback (%s) threw an exception: %s.", name_copy.c_str(), e.what());
976 : 6 : } catch (...) {
977 [ + - + - ]: 6 : CLOG_E("GraphTask: status_callback (%s) threw a non-std exception.", name_copy.c_str());
978 [ + - ]: 3 : }
979 : : }
980 : 170 : }
981 : :
982 : 11 : bool GraphTask::detect_cycle(const GraphTask* task, std::unordered_set<const GraphTask*>& visited,
983 : : std::unordered_set<const GraphTask*>& recursion_stack, uint32_t& depth,
984 : : uint32_t max_depth) const {
985 [ + - - + ]: 11 : if VUNLIKELY (recursion_stack.count(task)) {
986 : : return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
987 : : }
988 : :
989 [ + - - + ]: 11 : if (visited.count(task)) {
990 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
991 : : }
992 : :
993 [ + + ]: 11 : if VUNLIKELY (++depth > max_depth) {
994 [ + - - + ]: 2 : CLOG_F("GraphTask: detect_cycle exceeded max_recursion_depth (%u).", max_depth);
995 : : return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
996 : : }
997 : :
998 [ + - ]: 10 : visited.insert(task);
999 [ + - ]: 10 : recursion_stack.insert(task);
1000 : :
1001 : 10 : std::vector<std::weak_ptr<GraphTask>> succ_copy;
1002 : : {
1003 [ + - ]: 10 : std::lock_guard lock(task->impl_->mtx);
1004 [ + - ]: 10 : succ_copy = task->impl_->succeed_task_list;
1005 : 10 : }
1006 : :
1007 [ + + ]: 14 : for (const auto& weak_task : succ_copy) {
1008 : 5 : auto next_task = weak_task.lock();
1009 : :
1010 [ - + ]: 5 : if VUNLIKELY (!next_task) {
1011 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1012 : : }
1013 : :
1014 [ + + - + ]: 5 : if VUNLIKELY (detect_cycle(next_task.get(), visited, recursion_stack, depth, max_depth)) {
1015 : : return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1016 : : }
1017 [ + - - ]: 5 : }
1018 : :
1019 [ + - ]: 9 : recursion_stack.erase(task);
1020 : 9 : --depth;
1021 : :
1022 : 9 : return false;
1023 : 10 : }
1024 : :
1025 : 59 : void GraphTask::clear_invalid_task(const std::shared_ptr<GraphTask>& task) {
1026 : : {
1027 [ + - ]: 177 : task->impl_->precede_task_list.erase(
1028 [ + - ]: 59 : std::remove_if(task->impl_->precede_task_list.begin(), task->impl_->precede_task_list.end(),
1029 : 39 : [](const std::weak_ptr<GraphTask>& weak_task) { return weak_task.expired(); }),
1030 : 59 : task->impl_->precede_task_list.end());
1031 : :
1032 [ + - ]: 177 : task->impl_->succeed_task_list.erase(
1033 [ + - ]: 59 : std::remove_if(task->impl_->succeed_task_list.begin(), task->impl_->succeed_task_list.end(),
1034 : 39 : [](const std::weak_ptr<GraphTask>& weak_task) { return weak_task.expired(); }),
1035 : 59 : task->impl_->succeed_task_list.end());
1036 : : }
1037 : 59 : }
1038 : :
1039 : : } // namespace vlink
|
