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