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