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