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/thread_pool.h"
25 : :
26 : : #include <algorithm>
27 : : #include <atomic>
28 : : #include <deque>
29 : : #include <mutex>
30 : : #include <optional>
31 : : #include <string>
32 : : #include <thread>
33 : : #include <tuple>
34 : : #include <utility>
35 : : #include <vector>
36 : :
37 : : #include "./base/condition_variable.h"
38 : : #include "./base/logger.h"
39 : : #include "./base/memory_pool.h"
40 : : #include "./base/memory_resource.h"
41 : : #include "./base/mpmc_queue.h"
42 : : #include "./base/utils.h"
43 : :
44 : : namespace vlink {
45 : :
46 : : static constexpr size_t kMaxTaskSize = 10000U;
47 : : static constexpr int kMaxLockfreePushRetry = 32;
48 : :
49 : : // ThreadPoolGlobal
50 : : struct ThreadPoolGlobal final {
51 : : std::atomic<int> instance_index{0};
52 : :
53 : 67 : static ThreadPoolGlobal& get() {
54 : : static ThreadPoolGlobal instance;
55 : :
56 : 67 : return instance;
57 : : }
58 : :
59 : : private:
60 : : ThreadPoolGlobal() = default;
61 : : };
62 : :
63 : : // ThreadPool::Impl
64 : : struct ThreadPool::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
65 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
66 : : using NormalTaskTuple = std::tuple<bool, ThreadPool::Callback>;
67 : : using LockfreeTaskTuple = std::tuple<ThreadPool::Callback>;
68 : : using NormalQueue = std::pmr::deque<NormalTaskTuple>;
69 : : using LockfreeQueue = MpmcQueue<LockfreeTaskTuple>;
70 : : #else
71 : : using NormalTaskTuple = std::tuple<bool, ThreadPool::Callback>;
72 : : using LockfreeTaskTuple = std::tuple<ThreadPool::Callback>;
73 : : using NormalQueue = std::deque<NormalTaskTuple>;
74 : : using LockfreeQueue = MpmcQueue<LockfreeTaskTuple>;
75 : : #endif
76 : :
77 : : alignas(64) std::atomic_bool quit_flag{false};
78 : : alignas(64) std::atomic_size_t lockfree_task_count{0U};
79 : : alignas(64) std::atomic_size_t lockfree_producer_count{0U};
80 : :
81 : : std::string name;
82 : : size_t thread_count{0};
83 : : ThreadPool::Type type{ThreadPool::kNormalType};
84 : : std::atomic<ThreadPool::Strategy> strategy{ThreadPool::kOptimizationStrategy};
85 : : std::vector<std::thread> threads;
86 : :
87 : : std::optional<NormalQueue> normal_queue;
88 : : std::optional<LockfreeQueue> lockfree_queue;
89 : :
90 : : ConditionVariable cv;
91 : : std::mutex mtx;
92 : : };
93 : :
94 : : // ThreadPool
95 : 40 : ThreadPool::ThreadPool(size_t thread_count) : impl_(MemoryResource::make_shared<Impl>()) {
96 : 40 : impl_->name =
97 [ + - + - ]: 120 : "ThreadPool_" + std::to_string(ThreadPoolGlobal::get().instance_index.fetch_add(1, std::memory_order_relaxed));
98 : 40 : impl_->thread_count = thread_count;
99 : :
100 [ + - ]: 40 : MemoryPool::global_instance();
101 : :
102 [ + - ]: 40 : init();
103 : 40 : }
104 : :
105 : 27 : ThreadPool::ThreadPool(size_t thread_count, Type type) : impl_(MemoryResource::make_shared<Impl>()) {
106 : 27 : impl_->name =
107 [ + - + - ]: 81 : "ThreadPool_" + std::to_string(ThreadPoolGlobal::get().instance_index.fetch_add(1, std::memory_order_relaxed));
108 : 27 : impl_->thread_count = thread_count;
109 : 27 : impl_->type = type;
110 : :
111 [ + - ]: 27 : MemoryPool::global_instance();
112 : :
113 [ + - ]: 27 : init();
114 : 27 : }
115 : :
116 : 67 : ThreadPool::~ThreadPool() { shutdown(); }
117 : :
118 : 2 : void ThreadPool::set_name(const std::string& name) { impl_->name = name; }
119 : :
120 : 2 : const std::string& ThreadPool::get_name() const { return impl_->name; }
121 : :
122 : 3 : ThreadPool::Type ThreadPool::get_type() const { return impl_->type; }
123 : :
124 : 3 : ThreadPool::Strategy ThreadPool::get_strategy() const { return impl_->strategy.load(std::memory_order_acquire); }
125 : :
126 : 11 : void ThreadPool::set_strategy(Strategy strategy) { impl_->strategy.store(strategy, std::memory_order_release); }
127 : :
128 : 899 : bool ThreadPool::post_task(Callback&& callback) {
129 : 899 : return push_task(std::move(callback), true, TaskOverflowPolicy::kUseDispatcherStrategy);
130 : : }
131 : :
132 : 24 : TaskHandle ThreadPool::post_task_handle(Callback&& callback, const PostTaskOptions& options) {
133 [ + - ]: 24 : auto handle = TaskHandle::make_task_handle(options.cancellation_token);
134 [ + - ]: 24 : TaskHandle::mark_task_queued(handle);
135 : :
136 [ + + ]: 24 : if (handle.state() == TaskExecutionState::kCancelled) {
137 : 2 : return handle;
138 : : }
139 : :
140 [ + + - + : 22 : if VUNLIKELY (impl_->type == kLockfreeType && options.drop_policy == TaskDropPolicy::kProtected) {
- + ]
141 : : CLOG_W("ThreadPool: TaskDropPolicy::kProtected is ignored by lock-free queues (%s).", // LCOV_EXCL_LINE
142 : : // GCOVR_EXCL_LINE
143 : : impl_->name.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
144 : : }
145 : :
146 [ + - ]: 22 : auto tracked = TaskHandle::make_tracked_task(handle, std::move(callback));
147 : 22 : const bool droppable = options.drop_policy == TaskDropPolicy::kDroppable;
148 : :
149 [ + - + + : 22 : if VUNLIKELY (!push_task(std::move(tracked), droppable, options.overflow_policy, &handle) && !handle.is_done()) {
- + - + ]
150 : : TaskHandle::mark_task_rejected(handle); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
151 : : }
152 : :
153 : 22 : return handle;
154 : 22 : }
155 : :
156 : 4 : bool ThreadPool::drop_one_normal_task() {
157 [ + + ]: 5 : for (auto iter = impl_->normal_queue->begin(); iter != impl_->normal_queue->end(); ++iter) {
158 [ + + ]: 4 : if (std::get<0>(*iter)) {
159 [ + - ]: 3 : impl_->normal_queue->erase(iter);
160 : 3 : return true;
161 : : }
162 : : }
163 : :
164 : 1 : return false;
165 : : }
166 : :
167 : 1 : bool ThreadPool::drop_one_lockfree_task(bool keep_reserved) {
168 : 1 : Impl::LockfreeTaskTuple task;
169 : :
170 [ - + ]: 1 : if (!impl_->lockfree_queue->try_pop(task)) {
171 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
172 : : }
173 : :
174 [ - + ]: 1 : if (!keep_reserved) {
175 : : release_lockfree_task(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
176 : : }
177 : :
178 : 1 : return true;
179 : 1 : }
180 : :
181 : 455 : bool ThreadPool::reserve_lockfree_task(bool* was_empty) {
182 : 455 : auto count = impl_->lockfree_task_count.load(std::memory_order_acquire);
183 [ + - ]: 459 : const auto max_count = get_max_task_count();
184 : :
185 [ + + ]: 488 : while (count < max_count) {
186 [ + + ]: 940 : if (impl_->lockfree_task_count.compare_exchange_weak(count, count + 1U, std::memory_order_acq_rel,
187 : : std::memory_order_acquire)) {
188 [ + - ]: 436 : if (was_empty != nullptr) {
189 : 436 : *was_empty = count == 0U;
190 : : }
191 : :
192 : 436 : return true;
193 : : }
194 : : }
195 : :
196 : 17 : return false;
197 : : }
198 : :
199 : 0 : void ThreadPool::release_lockfree_task() {
200 : 0 : impl_->lockfree_task_count.fetch_sub(1U, std::memory_order_acq_rel);
201 : : } // LCOV_EXCL_LINE GCOVR_EXCL_LINE
202 : :
203 : 435 : bool ThreadPool::push_lockfree_task(Callback&& callback) {
204 [ + - ]: 435 : for (int retry = 0; retry < kMaxLockfreePushRetry; ++retry) {
205 [ + - ]: 435 : if VLIKELY (impl_->lockfree_queue->try_push(std::forward_as_tuple(std::move(callback)))) {
206 : 436 : return true;
207 : : }
208 : :
209 : : // LCOV_EXCL_START GCOVR_EXCL_START
210 : : Utils::yield_cpu();
211 : : }
212 : :
213 : : CLOG_E("ThreadPool: Failed to push lockfree task after %d retries (%s).", kMaxLockfreePushRetry, impl_->name.c_str());
214 : :
215 : : return false;
216 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
217 : : }
218 : :
219 : 923 : bool ThreadPool::push_task(Callback&& callback, bool droppable, TaskOverflowPolicy overflow_policy,
220 : : const TaskHandle* submit_handle) {
221 : 1630 : auto is_cancelled = [submit_handle]() -> bool {
222 [ + + + + ]: 1546 : return submit_handle != nullptr && submit_handle->state() == TaskExecutionState::kCancelled;
223 : 923 : };
224 : :
225 : 16 : auto reject = [submit_handle]() -> bool {
226 [ + + + - : 8 : if (submit_handle != nullptr && !submit_handle->is_done()) {
+ + ]
227 : 4 : TaskHandle::mark_task_rejected(*submit_handle);
228 : : }
229 : :
230 : 8 : return false;
231 : 923 : };
232 : :
233 [ + + ]: 923 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
234 [ + - ]: 4 : return reject();
235 : : }
236 : :
237 : 919 : bool is_full = false;
238 : 919 : int retry_cnt = 0;
239 : :
240 [ + + ]: 919 : if (impl_->type == kNormalType) {
241 [ + + ]: 556 : do {
242 : : {
243 [ + - ]: 559 : std::lock_guard lock(impl_->mtx);
244 : :
245 [ + + ]: 563 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
246 [ + - ]: 1 : return reject();
247 : : }
248 : :
249 [ + + ]: 562 : if VUNLIKELY (is_cancelled()) {
250 : 1 : return false;
251 : : }
252 : :
253 [ + - ]: 561 : is_full = impl_->normal_queue->size() >= get_max_task_count();
254 : :
255 [ + + ]: 561 : if VLIKELY (!is_full) {
256 [ + - ]: 479 : impl_->normal_queue->emplace_back(droppable, std::move(callback));
257 [ + + ]: 82 : } else if (overflow_policy == TaskOverflowPolicy::kReject) {
258 [ + - ]: 1 : return reject();
259 [ + + + - : 81 : } else if (impl_->strategy.load(std::memory_order_acquire) == kPopStrategy &&
+ + ]
260 : : overflow_policy != TaskOverflowPolicy::kBlock) {
261 [ + - + + ]: 4 : if (!drop_one_normal_task()) {
262 [ + - ]: 1 : return reject();
263 : : }
264 : :
265 [ + - ]: 3 : impl_->normal_queue->emplace_back(droppable, std::move(callback));
266 : 3 : is_full = false;
267 : :
268 : 3 : break;
269 : : }
270 [ + + + ]: 563 : }
271 : :
272 [ + + ]: 556 : if VUNLIKELY (is_full) {
273 [ + + - + : 77 : if (impl_->strategy.load(std::memory_order_acquire) == kOptimizationStrategy &&
- + ]
274 : : overflow_policy != TaskOverflowPolicy::kBlock) {
275 : : // LCOV_EXCL_START GCOVR_EXCL_START
276 : : if (++retry_cnt > 10) {
277 : : {
278 : : std::lock_guard lock(impl_->mtx);
279 : :
280 : : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
281 : : return reject();
282 : : }
283 : :
284 : : if VUNLIKELY (is_cancelled()) {
285 : : return false;
286 : : }
287 : :
288 : : if (!drop_one_normal_task()) {
289 : : return reject();
290 : : }
291 : :
292 : : impl_->normal_queue->emplace_back(droppable, std::move(callback));
293 : : is_full = false;
294 : : }
295 : :
296 : : CLOG_W("ThreadPool: Task is full, removed top data (%s).", impl_->name.c_str());
297 : : break;
298 : : }
299 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
300 : : }
301 : :
302 [ - + ]: 77 : if VUNLIKELY (is_cancelled()) {
303 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
304 : : }
305 : :
306 [ + - ]: 77 : std::this_thread::sleep_for(std::chrono::milliseconds(1));
307 : : }
308 : : } while (is_full);
309 [ + # ]: 431 : } else if (impl_->type == kLockfreeType) {
310 : 430 : bool notify_waiter = false;
311 : :
312 : : struct ProducerGuard final {
313 : 426 : explicit ProducerGuard(Impl& impl) noexcept : impl_ref(impl) {
314 : 426 : impl_ref.lockfree_producer_count.fetch_add(1U, std::memory_order_acq_rel);
315 : 426 : }
316 : :
317 : 433 : ~ProducerGuard() {
318 [ + + ]: 866 : if (impl_ref.lockfree_producer_count.fetch_sub(1U, std::memory_order_acq_rel) == 1U) {
319 : 240 : std::lock_guard lock(impl_ref.mtx);
320 : 240 : impl_ref.cv.notify_all();
321 : 240 : }
322 : 433 : }
323 : :
324 : : Impl& impl_ref;
325 : 430 : } producer_guard(*impl_);
326 : :
327 : 1306 : auto push_reserved_lockfree_task = [this, &reject, &is_cancelled, &callback]() -> bool {
328 [ - + ]: 436 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
329 : : release_lockfree_task(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
330 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
331 : : }
332 : :
333 [ - + ]: 435 : if VUNLIKELY (is_cancelled()) {
334 : : release_lockfree_task(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
335 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
336 : : }
337 : :
338 [ - + ]: 435 : if VUNLIKELY (!push_lockfree_task(std::move(callback))) {
339 : 0 : release_lockfree_task();
340 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
341 : : }
342 : :
343 : 435 : return true;
344 : 433 : };
345 : :
346 [ + - ]: 29 : do {
347 [ - + ]: 462 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
348 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
349 : : }
350 : :
351 [ - + ]: 454 : if VUNLIKELY (is_cancelled()) {
352 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
353 : : }
354 : :
355 [ + - ]: 455 : is_full = !reserve_lockfree_task(¬ify_waiter);
356 : :
357 [ + + ]: 467 : if VLIKELY (!is_full) {
358 [ + - - + ]: 436 : if VUNLIKELY (!push_reserved_lockfree_task()) {
359 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
360 : : }
361 : :
362 : 433 : break;
363 : : }
364 : :
365 [ + + ]: 31 : if (overflow_policy == TaskOverflowPolicy::kReject) {
366 [ + - ]: 1 : return reject();
367 : : }
368 : :
369 [ + + + - : 30 : if (impl_->strategy.load(std::memory_order_acquire) == kPopStrategy &&
+ + ]
370 : : overflow_policy != TaskOverflowPolicy::kBlock) {
371 [ + - - + ]: 1 : if (!drop_one_lockfree_task(true)) {
372 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
373 : : }
374 : :
375 [ + - - + ]: 1 : if VUNLIKELY (!push_reserved_lockfree_task()) {
376 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
377 : : }
378 : :
379 : 1 : is_full = false;
380 : 1 : break;
381 : : }
382 : :
383 [ + - ]: 29 : if VUNLIKELY (is_full) {
384 [ - + - - : 29 : if (impl_->strategy.load(std::memory_order_acquire) == kOptimizationStrategy &&
- + ]
385 : : overflow_policy != TaskOverflowPolicy::kBlock) {
386 : : // LCOV_EXCL_START GCOVR_EXCL_START
387 : : if (++retry_cnt > 10) {
388 : : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
389 : : return reject();
390 : : }
391 : :
392 : : if VUNLIKELY (is_cancelled()) {
393 : : return false;
394 : : }
395 : :
396 : : if (!drop_one_lockfree_task(true)) {
397 : : return reject();
398 : : }
399 : :
400 : : if VUNLIKELY (!push_reserved_lockfree_task()) {
401 : : return false;
402 : : }
403 : :
404 : : is_full = false;
405 : : CLOG_W("ThreadPool: Task is full, removed top data (%s).", impl_->name.c_str());
406 : : break;
407 : : }
408 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
409 : : }
410 : :
411 [ - + ]: 29 : if VUNLIKELY (is_cancelled()) {
412 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
413 : : }
414 : :
415 [ + - ]: 29 : std::this_thread::sleep_for(std::chrono::milliseconds(1));
416 : : }
417 : : } while (is_full);
418 : :
419 [ + + ]: 434 : if (notify_waiter) {
420 : : // Pair the empty-to-non-empty transition with the wait mutex so workers cannot miss the notify.
421 : : {
422 [ + - ]: 13 : std::lock_guard lock(impl_->mtx);
423 : 13 : }
424 : 13 : impl_->cv.notify_one();
425 : : }
426 [ + + ]: 436 : }
427 : :
428 [ + + ]: 915 : if (impl_->type != kLockfreeType) {
429 : 482 : impl_->cv.notify_one();
430 : : }
431 : :
432 : 915 : return !is_full;
433 : : }
434 : :
435 : 2 : size_t ThreadPool::get_task_count() const {
436 [ + + ]: 2 : if (impl_->type == kNormalType) {
437 [ + - ]: 1 : std::lock_guard lock(impl_->mtx);
438 : 1 : return impl_->normal_queue->size();
439 [ + - ]: 2 : } else if (impl_->type == kLockfreeType) {
440 : 2 : return impl_->lockfree_task_count.load(std::memory_order_acquire);
441 : : } else {
442 : : return 0U; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
443 : : }
444 : : }
445 : :
446 : 26 : bool ThreadPool::is_in_work_thread() const {
447 : 26 : return std::any_of(impl_->threads.begin(), impl_->threads.end(),
448 : 81 : [](const std::thread& thread) -> bool { return thread.get_id() == std::this_thread::get_id(); });
449 : : }
450 : :
451 : 889 : size_t ThreadPool::get_max_task_count() const { return kMaxTaskSize; }
452 : :
453 : 129 : bool ThreadPool::shutdown() {
454 : : {
455 [ + - ]: 129 : std::lock_guard lock(impl_->mtx);
456 : :
457 [ + + ]: 129 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
458 : 64 : return false;
459 : : }
460 : :
461 : 65 : impl_->quit_flag.store(true, std::memory_order_release);
462 [ + + ]: 129 : }
463 : :
464 [ + + ]: 65 : if (impl_->type == kLockfreeType) {
465 [ + - ]: 10 : std::unique_lock lock(impl_->mtx);
466 : 30 : impl_->cv.wait(lock, [this] { return impl_->lockfree_producer_count.load(std::memory_order_acquire) == 0U; });
467 : 10 : }
468 : :
469 : 65 : impl_->cv.notify_all();
470 : :
471 : 65 : const auto self_id = std::this_thread::get_id();
472 : :
473 [ + + ]: 190 : for (auto& thread : impl_->threads) {
474 [ + - ]: 125 : if (thread.joinable()) {
475 [ + + ]: 125 : if (thread.get_id() == self_id) {
476 [ + - ]: 1 : thread.detach();
477 : : } else {
478 [ + - ]: 124 : thread.join();
479 : : }
480 : : }
481 : : }
482 : :
483 : 65 : return true;
484 : : }
485 : :
486 : 67 : void ThreadPool::init() {
487 [ + + ]: 67 : if (impl_->type == kNormalType) {
488 : 57 : impl_->normal_queue.emplace();
489 [ + - ]: 10 : } else if (impl_->type == kLockfreeType) {
490 [ + - ]: 10 : const auto max_task_count = get_max_task_count(); // NOLINT(clang-analyzer-optin.cplusplus.VirtualCall)
491 [ + - ]: 10 : impl_->lockfree_queue.emplace(max_task_count);
492 : 10 : impl_->lockfree_task_count.store(0U, std::memory_order_release);
493 : : }
494 : :
495 [ + + ]: 67 : if VUNLIKELY (impl_->thread_count == 0) {
496 [ + - + - ]: 2 : VLOG_E("ThreadPool: Thread count is zero.");
497 : 1 : impl_->quit_flag.store(true, std::memory_order_release);
498 : 1 : return;
499 : : }
500 : :
501 : 66 : impl_->threads.reserve(impl_->thread_count);
502 : :
503 [ + + ]: 192 : for (size_t i = 0; i < impl_->thread_count; ++i) {
504 [ + + ]: 126 : if (impl_->type == kNormalType) {
505 : 104 : auto impl = impl_;
506 : 582 : std::thread thread([impl] {
507 : : for (;;) {
508 : 582 : Callback task;
509 : :
510 : : {
511 [ + - ]: 582 : std::unique_lock lock(impl->mtx);
512 : 583 : impl->cv.wait(lock, [impl] {
513 [ + + + + ]: 878 : return !impl->normal_queue->empty() || impl->quit_flag.load(std::memory_order_acquire);
514 : : });
515 : :
516 [ + + + - : 582 : if VUNLIKELY (impl->normal_queue->empty() && impl->quit_flag.load(std::memory_order_acquire)) {
+ + ]
517 : 103 : break;
518 : : }
519 : :
520 : 479 : task = std::move(std::get<1>(impl->normal_queue->front()));
521 : :
522 : 479 : impl->normal_queue->pop_front();
523 [ + + ]: 582 : }
524 : :
525 [ + - ]: 479 : if VLIKELY (task) {
526 [ + - ]: 479 : task();
527 : : }
528 [ + + ]: 1060 : }
529 [ + - ]: 207 : });
530 : :
531 [ + - ]: 104 : impl_->threads.emplace_back(std::move(thread));
532 [ + - ]: 126 : } else if (impl_->type == kLockfreeType) {
533 : 22 : auto impl = impl_;
534 : 482 : std::thread thread([impl] {
535 : : for (;;) {
536 : 482 : Impl::LockfreeTaskTuple task_tuple;
537 : :
538 : 482 : const bool has_task = impl->lockfree_queue->try_pop(task_tuple);
539 : :
540 [ + + ]: 482 : if (!has_task) {
541 [ + + + - : 68 : if VUNLIKELY (impl->quit_flag.load(std::memory_order_acquire) &&
+ + ]
542 : : impl->lockfree_task_count.load(std::memory_order_acquire) == 0U) {
543 [ + - ]: 44 : if (impl->lockfree_producer_count.load(std::memory_order_acquire) ==
544 : : 0U) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
545 : 22 : break;
546 : : }
547 : :
548 : : std::this_thread::yield(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
549 : 24 : continue;
550 : : }
551 : :
552 [ - + ]: 48 : if (impl->lockfree_task_count.load(std::memory_order_acquire) != 0U) {
553 : 0 : std::this_thread::yield();
554 : 0 : continue;
555 : : }
556 : :
557 [ + - ]: 24 : std::unique_lock lock(impl->mtx);
558 : 24 : impl->cv.wait(lock, [impl] {
559 [ + + + + ]: 168 : return impl->lockfree_task_count.load(std::memory_order_acquire) != 0U ||
560 : 110 : impl->quit_flag.load(std::memory_order_acquire);
561 : : });
562 : :
563 : 24 : continue;
564 : 24 : }
565 : :
566 : 436 : impl->lockfree_task_count.fetch_sub(1U, std::memory_order_acq_rel);
567 : :
568 : 436 : auto& task = std::get<0>(task_tuple);
569 : :
570 [ + - ]: 436 : if VLIKELY (task) {
571 [ + - ]: 436 : task();
572 : : }
573 [ + + + ]: 942 : }
574 [ + - ]: 44 : });
575 : :
576 [ + - ]: 22 : impl_->threads.emplace_back(std::move(thread));
577 : 22 : }
578 : : }
579 : : }
580 : :
581 : : } // namespace vlink
|
