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