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/message_loop.h"
25 : :
26 : : #include <algorithm>
27 : : #include <atomic>
28 : : #include <deque>
29 : : #include <iterator>
30 : : #include <memory>
31 : : #include <mutex>
32 : : #include <optional>
33 : : #include <queue>
34 : : #include <string>
35 : : #include <thread>
36 : : #include <tuple>
37 : : #include <unordered_set>
38 : : #include <utility>
39 : : #include <vector>
40 : :
41 : : #include "./base/condition_variable.h"
42 : : #include "./base/logger.h"
43 : : #include "./base/memory_pool.h"
44 : : #include "./base/memory_resource.h"
45 : : #include "./base/mpmc_queue.h"
46 : : #include "./base/utils.h"
47 : :
48 : : #ifdef _WIN32
49 : : #include <Windows.h>
50 : : #undef max
51 : : #endif
52 : :
53 : : namespace vlink {
54 : :
55 : : static constexpr size_t kMaxTaskSize = 10000U;
56 : : static constexpr size_t kMaxTimerSize = 100U;
57 : : static constexpr uint32_t kMaxElapsedTime = 0U;
58 : : static constexpr int kMaxLockfreePushRetry = 32;
59 : : static constexpr int kResetDrainTimeoutMs = 1000;
60 : :
61 : : template <typename TypeT, typename TimeT, typename ReturnT>
62 : 11382 : static ReturnT get_current_time() noexcept {
63 : 11382 : const auto& duration = std::chrono::duration_cast<TimeT>(TypeT::now().time_since_epoch());
64 : :
65 : 11383 : return static_cast<ReturnT>(duration.count());
66 : : }
67 : :
68 : : // MessageLoopGlobal
69 : : struct MessageLoopGlobal final {
70 : : std::atomic<int> instance_index{0};
71 : :
72 : 1326 : static MessageLoopGlobal& get() {
73 : : static MessageLoopGlobal instance;
74 : :
75 : 1326 : return instance;
76 : : }
77 : :
78 : : private:
79 : : MessageLoopGlobal() = default;
80 : : };
81 : :
82 : : // MessageLoop::Impl
83 : : struct MessageLoop::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
84 : : using NormalTaskTuple = std::tuple<uint32_t, bool, MessageLoop::Callback>;
85 : : using LockfreeTaskTuple = std::tuple<uint32_t, MessageLoop::Callback>;
86 : : using PriorityTaskTuple = std::tuple<uint32_t, uint32_t, uint32_t, bool, MessageLoop::Callback>;
87 : :
88 : : struct RestoreDropPolicy final {
89 : : bool& value;
90 : : bool previous;
91 : :
92 : 89897 : ~RestoreDropPolicy() { value = previous; }
93 : : };
94 : :
95 : : struct PriorityCompare final {
96 : 198314 : bool operator()(const PriorityTaskTuple& lhs, const PriorityTaskTuple& rhs) const {
97 : 198314 : return priority_key(lhs) > priority_key(rhs);
98 : : }
99 : : };
100 : :
101 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
102 : : using NormalQueue = std::pmr::deque<NormalTaskTuple>;
103 : : using LockfreeQueue = MpmcQueue<LockfreeTaskTuple>;
104 : : using PriorityContainer = std::pmr::vector<PriorityTaskTuple>;
105 : : #else
106 : : using NormalQueue = std::deque<NormalTaskTuple>;
107 : : using LockfreeQueue = MpmcQueue<LockfreeTaskTuple>;
108 : : using PriorityContainer = std::vector<PriorityTaskTuple>;
109 : : #endif
110 : :
111 : : struct PriorityQueue final : std::priority_queue<PriorityTaskTuple, PriorityContainer, PriorityCompare> {
112 : : using Base = std::priority_queue<PriorityTaskTuple, PriorityContainer, PriorityCompare>;
113 : :
114 : : using Base::Base;
115 : :
116 : 4 : [[nodiscard]] PriorityContainer& container() noexcept { return this->c; }
117 : : };
118 : :
119 : 395928 : static uint64_t priority_key(const PriorityTaskTuple& task) {
120 : 395928 : return (static_cast<uint64_t>(std::get<0>(task)) << 32) | std::get<1>(task);
121 : : }
122 : :
123 : 1 : static bool priority_before(const PriorityTaskTuple& lhs, const PriorityTaskTuple& rhs) {
124 : 1 : return priority_key(lhs) < priority_key(rhs);
125 : : }
126 : :
127 : : std::atomic_bool is_running{false};
128 : : std::atomic_bool quit_flag{false};
129 : : std::atomic_bool force_quit_flag{false};
130 : : alignas(64) std::atomic_bool is_busy{false};
131 : : std::atomic_bool wakeup_pending{false};
132 : : std::atomic_bool manual_spin_waiting{false};
133 : : std::atomic_bool lockfree_needs_reset{false};
134 : : alignas(64) std::atomic_size_t lockfree_task_count{0U};
135 : : alignas(64) std::atomic_size_t lockfree_producer_count{0U};
136 : :
137 : : std::shared_ptr<MessageLoop::AliveState> alive_state{MemoryResource::make_shared<MessageLoop::AliveState>()};
138 : : std::atomic<std::thread::id> thread_id;
139 : :
140 : : #ifdef _WIN32
141 : : std::atomic<HANDLE> thread_handle{nullptr};
142 : : std::mutex thread_handle_mtx;
143 : : #endif
144 : :
145 : : std::string name;
146 : : MessageLoop::Type type{MessageLoop::kNormalType};
147 : : std::atomic<MessageLoop::Strategy> strategy{MessageLoop::kOptimizationStrategy};
148 : :
149 : : uint32_t task_seq{0};
150 : : size_t block_waiter_count{0U};
151 : : std::optional<NormalQueue> normal_queue;
152 : : std::optional<NormalQueue> normal_staging_queue;
153 : : std::optional<LockfreeQueue> lockfree_queue;
154 : : std::optional<PriorityQueue> priority_droppable_queue;
155 : : std::optional<PriorityQueue> priority_protected_queue;
156 : :
157 : : MessageLoop::Callback begin_callback;
158 : : MessageLoop::Callback end_callback;
159 : : MessageLoop::Callback idle_callback;
160 : :
161 : : std::thread thread;
162 : : std::unordered_set<Timer*> timer_set;
163 : : std::mutex spin_once_mtx;
164 : : std::mutex mtx;
165 : : ConditionVariable cv;
166 : : ConditionVariable capacity_cv;
167 : : bool current_task_droppable{true};
168 : : };
169 : :
170 : : // MessageLoop
171 : 1237 : MessageLoop::MessageLoop() : impl_(std::make_unique<Impl>()) {
172 : 1237 : impl_->name =
173 [ + - + - ]: 3711 : "MessageLoop_" + std::to_string(MessageLoopGlobal::get().instance_index.fetch_add(1, std::memory_order_relaxed));
174 : :
175 [ + - ]: 1237 : MemoryPool::global_instance();
176 : :
177 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
178 [ + - + - ]: 1237 : impl_->normal_queue.emplace(&MemoryResource::global_instance());
179 : : #else
180 : : impl_->normal_queue.emplace();
181 : : #endif
182 : 1237 : }
183 : :
184 : 89 : MessageLoop::MessageLoop(Type type) : impl_(std::make_unique<Impl>()) {
185 : 89 : impl_->name =
186 [ + - + - ]: 267 : "MessageLoop_" + std::to_string(MessageLoopGlobal::get().instance_index.fetch_add(1, std::memory_order_relaxed));
187 : 89 : impl_->type = type;
188 : :
189 [ + - ]: 89 : MemoryPool::global_instance();
190 : :
191 [ + + ]: 89 : if (impl_->type == kNormalType) {
192 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
193 [ + - + - ]: 40 : impl_->normal_queue.emplace(&MemoryResource::global_instance());
194 : : #else
195 : : impl_->normal_queue.emplace();
196 : : #endif
197 [ + + ]: 49 : } else if (impl_->type == kLockfreeType) {
198 [ + - ]: 21 : size_t max_task_size = get_max_task_count(); // NOLINT(clang-analyzer-optin.cplusplus.VirtualCall)
199 [ + - ]: 21 : impl_->lockfree_queue.emplace(max_task_size);
200 [ + - ]: 28 : } else if (impl_->type == kPriorityType) {
201 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
202 [ + - + - ]: 28 : impl_->priority_droppable_queue.emplace(&MemoryResource::global_instance());
203 [ + - + - ]: 28 : impl_->priority_protected_queue.emplace(&MemoryResource::global_instance());
204 : : #else
205 : : impl_->priority_droppable_queue.emplace();
206 : : impl_->priority_protected_queue.emplace();
207 : : #endif
208 : : }
209 : 89 : }
210 : :
211 : 1329 : MessageLoop::~MessageLoop() {
212 : : // NOLINTBEGIN
213 : : #ifdef _WIN32
214 : : if VUNLIKELY (Utils::is_terminating()) {
215 : : impl_->alive_state->alive.store(false, std::memory_order_release);
216 : : (void)impl_.release();
217 : : return;
218 : : }
219 : : #endif
220 : :
221 : 1326 : quit(!impl_->is_running.load(std::memory_order_acquire));
222 : :
223 : : {
224 : 1326 : std::lock_guard lock(impl_->alive_state->mtx);
225 : 1326 : impl_->alive_state->alive.store(false, std::memory_order_release);
226 : 1326 : }
227 : :
228 : : #ifdef _WIN32
229 : : if VUNLIKELY (impl_->is_running.load(std::memory_order_acquire)) {
230 : : (void)wait_for_quit(0, false);
231 : : }
232 : : #endif
233 : :
234 [ + + - + ]: 1326 : const bool self_destruct = impl_->thread.joinable() && impl_->thread.get_id() == std::this_thread::get_id();
235 : :
236 [ - + ]: 1326 : if VUNLIKELY (self_destruct) {
237 [ # # ]: 0 : CLOG_E(
238 : : "MessageLoop destroyed from its own loop thread(%s); destroying a loop from one of its own tasks is not "
239 : : "supported.",
240 : : impl_->name.c_str());
241 : : }
242 : :
243 [ + + ]: 1326 : if VUNLIKELY (impl_->is_running.load(std::memory_order_acquire)) {
244 [ + - ]: 40 : CLOG_W("MessageLoop is still running(%s).", impl_->name.c_str());
245 : :
246 [ + - ]: 20 : if VLIKELY (!self_destruct) {
247 : 20 : wait_for_quit(1000, false);
248 : : }
249 : : }
250 : :
251 [ + + ]: 1326 : if (impl_->thread.joinable()) {
252 : 801 : impl_->thread.join();
253 : : }
254 : :
255 : : #ifdef _WIN32
256 : : {
257 : : std::lock_guard lock(impl_->thread_handle_mtx);
258 : : HANDLE thread_handle = impl_->thread_handle.exchange(nullptr, std::memory_order_acq_rel);
259 : :
260 : : if (thread_handle != nullptr) {
261 : : ::CloseHandle(thread_handle);
262 : : }
263 : : }
264 : : #endif
265 : :
266 : 1326 : std::vector<Timer*> timers_to_delete;
267 : :
268 : : {
269 : 1326 : std::unique_lock lock(impl_->mtx);
270 : :
271 [ + + ]: 1327 : for (auto iter = impl_->timer_set.begin(); iter != impl_->timer_set.end();) {
272 : 1 : Timer* timer = *iter;
273 : 1 : timer->clear();
274 : :
275 [ + - ]: 1 : if (timer->is_once_type()) {
276 : 1 : timers_to_delete.emplace_back(timer);
277 : 1 : iter = impl_->timer_set.erase(iter);
278 : : } else {
279 : 0 : ++iter;
280 : : }
281 : : }
282 : 1326 : }
283 : :
284 [ + + ]: 1327 : for (auto* timer : timers_to_delete) {
285 : 1 : timer->~Timer();
286 : 1 : MemoryPool::global_instance().deallocate(timer, sizeof(Timer), alignof(Timer));
287 : : }
288 : : // NOLINTEND
289 : 1329 : }
290 : :
291 : 1043 : void MessageLoop::set_name(const std::string& name) { impl_->name = name; }
292 : :
293 : 2 : const std::string& MessageLoop::get_name() const { return impl_->name; }
294 : :
295 : 62 : MessageLoop::Type MessageLoop::get_type() const { return impl_->type; }
296 : :
297 : 4 : MessageLoop::Strategy MessageLoop::get_strategy() const { return impl_->strategy.load(std::memory_order_acquire); }
298 : :
299 : 174 : void MessageLoop::set_strategy(Strategy strategy) {
300 : : {
301 [ + - ]: 174 : std::lock_guard lock(impl_->mtx);
302 : 174 : impl_->strategy.store(strategy, std::memory_order_release);
303 : 174 : }
304 : :
305 : 174 : impl_->capacity_cv.notify_all();
306 : 174 : }
307 : :
308 : 3 : void MessageLoop::register_begin_handler(Callback&& callback) {
309 [ + + ]: 3 : if VUNLIKELY (impl_->is_running.load(std::memory_order_acquire)) {
310 [ + - + - : 2 : CLOG_E("MessageLoop is running and cannot be registered(%s).", impl_->name.c_str());
+ - ]
311 : 1 : return;
312 : : }
313 : :
314 : 2 : impl_->begin_callback = std::move(callback);
315 : : }
316 : :
317 : 3 : void MessageLoop::register_end_handler(Callback&& callback) {
318 [ + + ]: 3 : if VUNLIKELY (impl_->is_running.load(std::memory_order_acquire)) {
319 [ + - + - : 2 : CLOG_E("MessageLoop is running and cannot be registered(%s).", impl_->name.c_str());
+ - ]
320 : 1 : return;
321 : : }
322 : :
323 : 2 : impl_->end_callback = std::move(callback);
324 : : }
325 : :
326 : 9 : void MessageLoop::register_idle_handler(Callback&& callback) {
327 [ + + ]: 9 : if VUNLIKELY (impl_->is_running.load(std::memory_order_acquire)) {
328 [ + - + - : 2 : CLOG_E("MessageLoop is running and cannot be registered(%s).", impl_->name.c_str());
+ - ]
329 : 1 : return;
330 : : }
331 : :
332 : 8 : impl_->idle_callback = std::move(callback);
333 : : }
334 : :
335 : 3 : bool MessageLoop::run() {
336 : : #ifdef _WIN32
337 : : std::unique_lock thread_handle_lock(impl_->thread_handle_mtx);
338 : : #endif
339 : :
340 : 3 : bool expected = false;
341 : :
342 [ - + ]: 3 : if VUNLIKELY (!impl_->is_running.compare_exchange_strong(expected, true, std::memory_order_acq_rel,
343 : : std::memory_order_acquire)) {
344 : : CLOG_W("MessageLoop has already run(%s).", impl_->name.c_str()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
345 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
346 : : }
347 : :
348 : : #ifdef _WIN32
349 : : HANDLE old_thread_handle = impl_->thread_handle.exchange(nullptr, std::memory_order_acq_rel);
350 : :
351 : : if (old_thread_handle != nullptr) {
352 : : ::CloseHandle(old_thread_handle);
353 : : }
354 : :
355 : : HANDLE thread_handle = nullptr;
356 : :
357 : : if (::DuplicateHandle(::GetCurrentProcess(), ::GetCurrentThread(), ::GetCurrentProcess(), &thread_handle, SYNCHRONIZE,
358 : : FALSE, 0) != FALSE) {
359 : : impl_->thread_handle.store(thread_handle, std::memory_order_release);
360 : : }
361 : :
362 : : thread_handle_lock.unlock();
363 : : #endif
364 : :
365 [ - + - - : 3 : if (impl_->type == kLockfreeType && impl_->lockfree_needs_reset.exchange(false, std::memory_order_acq_rel)) {
- + ]
366 : : // LCOV_EXCL_START GCOVR_EXCL_START
367 : : std::lock_guard lock(impl_->mtx);
368 : : const auto max_task_count = get_max_task_count();
369 : : impl_->lockfree_queue.emplace(max_task_count);
370 : : impl_->lockfree_task_count.store(0U, std::memory_order_release);
371 : : }
372 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
373 : :
374 : 3 : impl_->quit_flag.store(false, std::memory_order_release);
375 : 3 : impl_->force_quit_flag.store(false, std::memory_order_release);
376 : :
377 [ + - ]: 3 : do_consume();
378 : :
379 : 3 : return true;
380 : : }
381 : :
382 : 809 : bool MessageLoop::async_run() {
383 : : #ifdef _WIN32
384 : : std::unique_lock thread_handle_lock(impl_->thread_handle_mtx);
385 : : #endif
386 : :
387 : 809 : bool expected = false;
388 : :
389 [ + + ]: 809 : if VUNLIKELY (!impl_->is_running.compare_exchange_strong(expected, true, std::memory_order_acq_rel,
390 : : std::memory_order_acquire)) {
391 [ + - + - : 6 : CLOG_W("MessageLoop has already run(%s).", impl_->name.c_str());
+ - ]
392 : 3 : return false;
393 : : }
394 : :
395 [ + + ]: 806 : if (impl_->thread.joinable()) {
396 [ + - ]: 5 : impl_->thread.join();
397 : : }
398 : :
399 : : #ifdef _WIN32
400 : : HANDLE old_thread_handle = impl_->thread_handle.exchange(nullptr, std::memory_order_acq_rel);
401 : :
402 : : if (old_thread_handle != nullptr) {
403 : : ::CloseHandle(old_thread_handle);
404 : : }
405 : : #endif
406 : :
407 [ + + + + : 806 : if (impl_->type == kLockfreeType && impl_->lockfree_needs_reset.exchange(false, std::memory_order_acq_rel)) {
+ + ]
408 [ + - ]: 1 : std::lock_guard lock(impl_->mtx);
409 [ + - ]: 1 : const auto max_task_count = get_max_task_count();
410 [ + - ]: 1 : impl_->lockfree_queue.emplace(max_task_count);
411 : 1 : impl_->lockfree_task_count.store(0U, std::memory_order_release);
412 : 1 : }
413 : :
414 : 806 : impl_->quit_flag.store(false, std::memory_order_release);
415 : 806 : impl_->force_quit_flag.store(false, std::memory_order_release);
416 : :
417 : : try {
418 [ + - ]: 1612 : impl_->thread = std::thread([this]() { do_consume(); });
419 : 0 : } catch (...) {
420 : : {
421 [ - - ]: 0 : std::lock_guard lock(impl_->mtx);
422 : 0 : impl_->quit_flag.store(true, std::memory_order_release);
423 : 0 : impl_->is_running.store(false, std::memory_order_release);
424 : 0 : }
425 : :
426 : 0 : impl_->cv.notify_all();
427 : 0 : impl_->capacity_cv.notify_all();
428 : 0 : throw;
429 : 0 : }
430 : :
431 : : #ifdef _WIN32
432 : : HANDLE thread_handle = nullptr;
433 : :
434 : : if (::DuplicateHandle(::GetCurrentProcess(), reinterpret_cast<HANDLE>(impl_->thread.native_handle()),
435 : : ::GetCurrentProcess(), &thread_handle, SYNCHRONIZE, FALSE, 0) != FALSE) {
436 : : HANDLE expected_thread_handle = nullptr;
437 : :
438 : : if (!impl_->thread_handle.compare_exchange_strong(expected_thread_handle, thread_handle, std::memory_order_acq_rel,
439 : : std::memory_order_acquire)) {
440 : : ::CloseHandle(thread_handle);
441 : : }
442 : : }
443 : :
444 : : thread_handle_lock.unlock();
445 : : #endif
446 : :
447 [ + - ]: 806 : if (!impl_->name.empty()) {
448 : 806 : Utils::set_thread_name(impl_->name, &impl_->thread);
449 : : }
450 : :
451 : 806 : return true;
452 : : }
453 : :
454 : 1 : bool MessageLoop::spin() { return run(); }
455 : :
456 : 27 : bool MessageLoop::spin_once(bool block) {
457 : 27 : std::unique_lock spin_once_lock(impl_->spin_once_mtx, std::try_to_lock);
458 : :
459 [ - + ]: 27 : if VUNLIKELY (!spin_once_lock.owns_lock()) {
460 : 0 : return false;
461 : : }
462 : :
463 [ + - + + : 27 : if VUNLIKELY (!MessageLoop::is_in_same_thread() &&
+ + + + ]
464 : : impl_->thread_id.load(std::memory_order_acquire) != std::thread::id()) {
465 [ + - + - : 2 : CLOG_E("MessageLoop spin_once called from different thread than run/async_run (%s).", impl_->name.c_str());
+ - ]
466 : 1 : return false;
467 : : }
468 : :
469 [ + + ]: 26 : if (impl_->type == kNormalType) {
470 [ + + ]: 18 : return process_normal_task(block, false);
471 [ + + ]: 8 : } else if (impl_->type == kLockfreeType) {
472 : : return process_lockfree_task(block); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
473 [ + - ]: 5 : } else if (impl_->type == kPriorityType) {
474 [ + - ]: 5 : return process_priority_task(block);
475 : : } else {
476 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
477 : : }
478 : 27 : }
479 : :
480 : 2930 : bool MessageLoop::quit(bool force) {
481 [ + + + + : 2930 : if VUNLIKELY (!force && (!impl_->is_running.load(std::memory_order_acquire) ||
+ + + + +
+ ]
482 : : impl_->quit_flag.load(std::memory_order_acquire))) {
483 : 16 : return false;
484 : : }
485 : :
486 : : {
487 [ + - ]: 2914 : std::lock_guard lock(impl_->mtx);
488 : 2914 : impl_->quit_flag.store(true, std::memory_order_release);
489 : 2914 : impl_->force_quit_flag.store(force, std::memory_order_release);
490 : 2914 : }
491 : :
492 [ + + ]: 2914 : if (impl_->type == kLockfreeType) {
493 [ + - ]: 35 : std::unique_lock lock(impl_->mtx);
494 : 107 : impl_->cv.wait(lock, [this] { return impl_->lockfree_producer_count.load(std::memory_order_acquire) == 0U; });
495 : 35 : }
496 : :
497 : 2914 : drop_pending_tasks();
498 : :
499 [ + + ]: 2914 : if (impl_->type == kLockfreeType) {
500 : 35 : impl_->lockfree_queue->notify_to_quit();
501 : 35 : impl_->lockfree_needs_reset.store(true, std::memory_order_release);
502 : : }
503 : :
504 : 2914 : impl_->cv.notify_all();
505 : 2914 : impl_->capacity_cv.notify_all();
506 : :
507 : 2914 : return true;
508 : : }
509 : :
510 : 1579 : bool MessageLoop::wait_for_quit(int ms, bool check) {
511 : : #ifdef _WIN32
512 : : {
513 : : std::lock_guard lock(impl_->thread_handle_mtx);
514 : : HANDLE thread_handle = impl_->thread_handle.load(std::memory_order_acquire);
515 : :
516 : : if (thread_handle != nullptr && ::WaitForSingleObject(thread_handle, 0) == WAIT_OBJECT_0) {
517 : : impl_->is_running.store(false, std::memory_order_release);
518 : : impl_->is_busy.store(false, std::memory_order_release);
519 : : impl_->cv.notify_all();
520 : : return true;
521 : : }
522 : : }
523 : : #endif
524 : :
525 [ + - ]: 1579 : std::unique_lock lock(impl_->mtx);
526 : :
527 [ + + + - : 1579 : if VUNLIKELY (check && is_in_same_thread()) {
+ + + + ]
528 [ + - + - : 2 : CLOG_E("MessageLoop wait_for_quit in work thread(%s).", impl_->name.c_str());
+ - ]
529 : 1 : return false;
530 : : }
531 : :
532 : 2304 : auto predicate = [this]() -> bool { return !impl_->is_running.load(std::memory_order_acquire); };
533 : :
534 [ + + ]: 1578 : if (ms == Timer::kInfinite) {
535 : 1243 : impl_->cv.wait(lock, std::move(predicate));
536 : 1243 : return true;
537 : : }
538 : :
539 [ + - ]: 335 : return impl_->cv.wait_for(lock, std::chrono::milliseconds(ms), std::move(predicate));
540 : 1579 : }
541 : :
542 : 83541 : bool MessageLoop::post_task(Callback&& callback) { return push_task(std::move(callback), kNoPriority); }
543 : :
544 : 20611 : TaskHandle MessageLoop::post_task_handle(Callback&& callback, const PostTaskOptions& options) {
545 [ + - ]: 20611 : auto handle = TaskHandle::make_task_handle(options.cancellation_token);
546 : :
547 [ + + ]: 20611 : if (handle.state() == TaskExecutionState::kCancelled) {
548 : 1 : return handle;
549 : : }
550 : :
551 [ + + + - : 20610 : if VUNLIKELY (impl_->type == kLockfreeType && options.drop_policy == TaskDropPolicy::kProtected) {
+ + ]
552 [ + - + - : 2 : CLOG_W("MessageLoop: TaskDropPolicy::kProtected is ignored by lock-free queues (%s).", impl_->name.c_str());
+ - ]
553 : : }
554 : :
555 [ + - ]: 20610 : auto tracked = TaskHandle::make_tracked_task(handle, std::move(callback));
556 : 20610 : const bool droppable = options.drop_policy == TaskDropPolicy::kDroppable;
557 : :
558 [ + - + + : 20610 : if VUNLIKELY (!push_task(std::move(tracked), kNoPriority, droppable, options.overflow_policy, &handle,
- + - + ]
559 : : options.cancellation_token.valid()) &&
560 : : !handle.is_done()) {
561 : : TaskHandle::mark_task_rejected(handle); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
562 : : }
563 : :
564 : 20610 : return handle;
565 : 20610 : }
566 : :
567 : 24 : bool MessageLoop::post_task_with_priority(Callback&& callback, uint16_t priority) {
568 [ + + ]: 24 : if VUNLIKELY (priority == kNoPriority) {
569 [ + - + - : 2 : CLOG_E("MessageLoop: Task priority cannot be zero (%s).", impl_->name.c_str());
+ - ]
570 : 1 : return false;
571 : : }
572 : :
573 [ + + ]: 23 : if VUNLIKELY (impl_->type != kPriorityType) {
574 [ + - + - : 6 : CLOG_E("MessageLoop: Task priority is not supported (%s).", impl_->name.c_str());
+ - ]
575 : 3 : return false;
576 : : }
577 : :
578 : 20 : return push_task(std::move(callback), priority);
579 : : }
580 : :
581 : 11 : TaskHandle MessageLoop::post_task_with_priority_handle(Callback&& callback, uint16_t priority,
582 : : const PostTaskOptions& options) {
583 [ + - ]: 11 : auto handle = TaskHandle::make_task_handle(options.cancellation_token);
584 : :
585 [ - + ]: 11 : if (handle.state() == TaskExecutionState::kCancelled) {
586 : : return handle; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
587 : : }
588 : :
589 [ + + ]: 11 : if VUNLIKELY (priority == kNoPriority) {
590 [ + - + - : 2 : CLOG_E("MessageLoop: Task priority cannot be zero (%s).", impl_->name.c_str());
+ - ]
591 [ + - ]: 1 : TaskHandle::mark_task_rejected(handle);
592 : 1 : return handle;
593 : : }
594 : :
595 [ + + ]: 10 : if VUNLIKELY (impl_->type != kPriorityType) {
596 [ + - + - : 2 : CLOG_E("MessageLoop: Task priority is not supported (%s).", impl_->name.c_str());
+ - ]
597 [ + - ]: 1 : TaskHandle::mark_task_rejected(handle);
598 : 1 : return handle;
599 : : }
600 : :
601 [ + - ]: 9 : auto tracked = TaskHandle::make_tracked_task(handle, std::move(callback));
602 : 9 : const bool droppable = options.drop_policy == TaskDropPolicy::kDroppable;
603 : :
604 [ + - + + : 9 : if VUNLIKELY (!push_task(std::move(tracked), priority, droppable, options.overflow_policy, &handle,
- + - + ]
605 : : options.cancellation_token.valid()) &&
606 : : !handle.is_done()) {
607 : : TaskHandle::mark_task_rejected(handle); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
608 : : }
609 : :
610 : 9 : return handle;
611 : 9 : }
612 : :
613 : 11046 : bool MessageLoop::wakeup() {
614 [ + + + + : 11046 : if VUNLIKELY (!impl_->is_running.load(std::memory_order_acquire) &&
+ + ]
615 : : !impl_->manual_spin_waiting.load(std::memory_order_acquire)) {
616 : 10131 : return false;
617 : : }
618 : :
619 : 915 : bool expected = false;
620 : :
621 [ + + ]: 915 : if (!impl_->wakeup_pending.compare_exchange_strong(expected, true, std::memory_order_acq_rel,
622 : : std::memory_order_acquire)) {
623 : 375 : return true;
624 : : }
625 : :
626 : : {
627 [ + - ]: 540 : std::lock_guard lock(impl_->mtx);
628 : 540 : }
629 : :
630 : 540 : impl_->cv.notify_all();
631 : :
632 : 540 : return true;
633 : : }
634 : :
635 : 323 : void MessageLoop::reset_lockfree_capacity() {
636 [ + + ]: 323 : if (impl_->type != kLockfreeType) {
637 : 322 : return;
638 : : }
639 : :
640 : 2 : std::vector<Callback> dropped_tasks;
641 [ + - ]: 2 : std::unique_lock lock(impl_->mtx);
642 : :
643 [ + + ]: 2 : if VUNLIKELY (impl_->is_running.load(std::memory_order_acquire)) {
644 [ + - + - : 2 : CLOG_E("MessageLoop: reset_lockfree_capacity called while running (%s).", impl_->name.c_str());
+ - ]
645 : 1 : return;
646 : : }
647 : :
648 [ + - ]: 1 : const bool drained = impl_->cv.wait_for(lock, std::chrono::milliseconds(kResetDrainTimeoutMs), [this] {
649 : 2 : return impl_->lockfree_producer_count.load(std::memory_order_acquire) == 0U;
650 : : });
651 : :
652 [ - + ]: 1 : if VUNLIKELY (!drained) {
653 [ # # # # : 0 : CLOG_E("MessageLoop: reset_lockfree_capacity skipped, producers still in flight (%s).", impl_->name.c_str());
# # ]
654 : 0 : return;
655 : : }
656 : :
657 [ + - ]: 1 : size_t max_task_size = get_max_task_count();
658 [ + - ]: 2 : dropped_tasks.reserve(impl_->lockfree_task_count.load(std::memory_order_acquire));
659 : 1 : Impl::LockfreeTaskTuple task;
660 : :
661 [ + + ]: 2 : while (impl_->lockfree_queue->try_pop(task)) {
662 [ + - ]: 1 : dropped_tasks.emplace_back(std::move(std::get<1>(task)));
663 : : }
664 : :
665 [ + - ]: 1 : impl_->lockfree_queue.emplace(max_task_size);
666 : 1 : impl_->lockfree_task_count.store(0U, std::memory_order_release);
667 : 1 : impl_->lockfree_needs_reset.store(false, std::memory_order_release);
668 [ + + + + ]: 3 : }
669 : :
670 : 691 : bool MessageLoop::is_running() const { return impl_->is_running.load(std::memory_order_acquire); }
671 : :
672 : 2265 : bool MessageLoop::is_ready_to_quit() const { return impl_->quit_flag.load(std::memory_order_acquire); }
673 : :
674 : 188 : bool MessageLoop::is_busy() const { return impl_->is_busy.load(std::memory_order_acquire); }
675 : :
676 : 60 : size_t MessageLoop::get_task_count() const {
677 [ + - ]: 60 : std::lock_guard lock(impl_->mtx);
678 : :
679 [ + + ]: 60 : if (impl_->type == kNormalType) {
680 : 51 : return impl_->normal_queue->size();
681 [ + + ]: 9 : } else if (impl_->type == kLockfreeType) {
682 : 8 : return impl_->lockfree_task_count.load(std::memory_order_acquire);
683 [ + - ]: 5 : } else if (impl_->type == kPriorityType) {
684 : 5 : return impl_->priority_droppable_queue->size() + impl_->priority_protected_queue->size();
685 : : }
686 : :
687 : : return 0; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
688 : 60 : }
689 : :
690 : 882 : bool MessageLoop::wait_for_idle(int ms, bool check) {
691 : : #ifdef _WIN32
692 : : {
693 : : std::lock_guard lock(impl_->thread_handle_mtx);
694 : : HANDLE thread_handle = impl_->thread_handle.load(std::memory_order_acquire);
695 : :
696 : : if (thread_handle != nullptr && ::WaitForSingleObject(thread_handle, 0) == WAIT_OBJECT_0) {
697 : : impl_->is_running.store(false, std::memory_order_release);
698 : : impl_->is_busy.store(false, std::memory_order_release);
699 : : impl_->cv.notify_all();
700 : : return true;
701 : : }
702 : : }
703 : : #endif
704 : :
705 [ + - ]: 882 : std::unique_lock lock(impl_->mtx);
706 : :
707 [ + + + - : 882 : if VUNLIKELY (check && is_in_same_thread()) {
+ + + + ]
708 [ + - + - : 2 : CLOG_E("MessageLoop wait_for_idle in work thread(%s).", impl_->name.c_str());
+ - ]
709 : 1 : return false;
710 : : }
711 : :
712 : 3196 : auto predicate = [this]() -> bool {
713 [ + + ]: 1094 : if (impl_->type == kNormalType) {
714 [ + + + + ]: 1063 : return !impl_->is_busy.load(std::memory_order_acquire) && impl_->normal_queue->empty();
715 [ + + ]: 31 : } else if (impl_->type == kLockfreeType) {
716 : 12 : const bool empty = impl_->lockfree_task_count.load(std::memory_order_acquire) == 0U;
717 [ + + + + ]: 12 : return !impl_->is_busy.load(std::memory_order_acquire) && empty;
718 [ + - ]: 19 : } else if (impl_->type == kPriorityType) {
719 [ + + + - ]: 19 : const bool empty = impl_->priority_droppable_queue->empty() && impl_->priority_protected_queue->empty();
720 [ + + + + ]: 19 : return !impl_->is_busy.load(std::memory_order_acquire) && empty;
721 : : }
722 : :
723 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
724 : 881 : };
725 : :
726 [ + + ]: 881 : if (ms == Timer::kInfinite) {
727 [ + - ]: 438 : impl_->cv.wait(lock, std::move(predicate));
728 : 438 : return true;
729 : : }
730 : :
731 [ + - ]: 443 : return impl_->cv.wait_for(lock, std::chrono::milliseconds(ms), std::move(predicate));
732 : 882 : }
733 : :
734 : 63080 : size_t MessageLoop::get_max_task_count() const { return kMaxTaskSize; }
735 : :
736 : 416 : size_t MessageLoop::get_max_timer_count() const { return kMaxTimerSize; }
737 : :
738 : 128412 : uint32_t MessageLoop::get_max_elapsed_time() const { return kMaxElapsedTime; }
739 : :
740 : 30164 : bool MessageLoop::is_in_same_thread() const {
741 : 30164 : return impl_->thread_id.load(std::memory_order_acquire) == std::this_thread::get_id();
742 : : }
743 : :
744 : 34 : std::shared_ptr<MessageLoop::AliveState> MessageLoop::get_alive_state() const { return impl_->alive_state; }
745 : :
746 : 23584 : bool MessageLoop::post_untracked_task(Callback&& callback, TaskOverflowPolicy overflow_policy,
747 : : TaskDropPolicy drop_policy) {
748 [ - + - - : 23584 : if VUNLIKELY (impl_->type == kLockfreeType && drop_policy == TaskDropPolicy::kProtected) {
- + ]
749 [ # # # # : 0 : CLOG_W("MessageLoop: TaskDropPolicy::kProtected is ignored by lock-free queues (%s).", impl_->name.c_str());
# # ]
750 : : }
751 : :
752 : 23583 : return push_task(std::move(callback), kNoPriority, drop_policy == TaskDropPolicy::kDroppable, overflow_policy);
753 : : }
754 : :
755 : 809 : void MessageLoop::on_begin() {
756 [ + + ]: 809 : if (impl_->begin_callback) {
757 : 2 : impl_->begin_callback();
758 : : }
759 : 809 : }
760 : :
761 : 809 : void MessageLoop::on_end() {
762 [ + + ]: 809 : if (impl_->end_callback) {
763 : 2 : impl_->end_callback();
764 : : }
765 : 809 : }
766 : :
767 : 18822 : void MessageLoop::on_idle() {
768 [ + + ]: 18822 : if (impl_->idle_callback) {
769 : 11 : impl_->idle_callback();
770 : : }
771 : 18821 : }
772 : :
773 : 70162 : bool MessageLoop::is_current_task_droppable() const { return impl_->current_task_droppable; }
774 : :
775 : 90332 : void MessageLoop::on_task_changed(Callback&& callback, uint32_t start_time) {
776 [ + + + - : 90332 : if (start_time > 0 && get_max_elapsed_time() > 0) {
+ + ]
777 : : uint32_t elapsed_time =
778 : 689 : get_current_time<std::chrono::steady_clock, std::chrono::milliseconds, uint32_t>() - start_time;
779 : :
780 [ + + ]: 689 : if VUNLIKELY (elapsed_time > get_max_elapsed_time()) {
781 : 1 : on_task_timeout(std::move(callback), elapsed_time);
782 : : } else {
783 : 688 : callback();
784 : : }
785 : : } else {
786 : 89643 : callback();
787 : : }
788 : 90329 : }
789 : :
790 : : // LCOV_EXCL_START GCOVR_EXCL_START
791 : : void MessageLoop::on_task_timeout(Callback&& callback, uint32_t elapsed_time) {
792 : : (void)callback;
793 : : CLOG_W("MessageLoop: Task timed out after %ums (%s).", elapsed_time, impl_->name.c_str());
794 : : }
795 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
796 : :
797 : 10003 : uint64_t MessageLoop::get_current_nano_time() {
798 : 10003 : return get_current_time<std::chrono::steady_clock, std::chrono::nanoseconds, uint64_t>();
799 : : }
800 : :
801 : 417 : bool MessageLoop::add_timer(Timer* timer) {
802 [ + - ]: 417 : std::lock_guard lock(impl_->mtx);
803 : :
804 [ - + ]: 417 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
805 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
806 : : }
807 : :
808 [ + - + + ]: 417 : if VUNLIKELY (impl_->timer_set.size() >= get_max_timer_count()) {
809 [ + - + - : 6 : CLOG_W("MessageLoop: Timer is full (%s).", impl_->name.c_str());
+ - ]
810 : 3 : return false;
811 : : }
812 : :
813 [ + - ]: 414 : return impl_->timer_set.emplace(timer).second;
814 : 417 : }
815 : :
816 : 555 : std::unique_lock<std::mutex> MessageLoop::lock_timers() { return std::unique_lock(impl_->mtx); }
817 : :
818 : 401 : bool MessageLoop::remove_timer(Timer* timer) {
819 [ + - ]: 401 : std::lock_guard lock(impl_->mtx);
820 : :
821 [ + - ]: 802 : return impl_->timer_set.erase(timer) != 0;
822 : 401 : }
823 : :
824 : 127764 : bool MessageLoop::push_task(Callback&& callback, uint16_t priority, bool droppable, TaskOverflowPolicy overflow_policy,
825 : : const TaskHandle* submit_handle, bool poll_cancellation) {
826 : 127764 : std::vector<Callback> dropped_tasks;
827 : :
828 : 174842 : auto is_cancelled = [submit_handle]() -> bool {
829 [ + + + + ]: 153814 : return submit_handle != nullptr && submit_handle->state() == TaskExecutionState::kCancelled;
830 : 127763 : };
831 : :
832 : 26 : auto reject = [submit_handle]() -> bool {
833 [ + + + - : 20 : if (submit_handle != nullptr && !submit_handle->is_done()) {
+ + ]
834 : 3 : TaskHandle::mark_task_rejected(*submit_handle);
835 : : }
836 : :
837 : 20 : return false;
838 : 127763 : };
839 : :
840 : 8754 : auto blocks_loop_thread = [this, overflow_policy]() -> bool {
841 [ + + ]: 44 : if ((overflow_policy == TaskOverflowPolicy::kBlock ||
842 [ + + + + ]: 8740 : impl_->strategy.load(std::memory_order_acquire) == kBlockStrategy) &&
843 [ + + ]: 4337 : MessageLoop::is_in_same_thread()) {
844 [ + - + - : 6 : CLOG_E("MessageLoop: blocking post from the loop thread would deadlock (%s).", impl_->name.c_str());
+ - ]
845 : 3 : return true;
846 : : }
847 : :
848 : 4367 : return false;
849 : 127763 : };
850 : :
851 : 26057 : auto block_for_capacity = [this, &is_cancelled, overflow_policy, poll_cancellation](auto&& has_capacity) -> bool {
852 [ + + + + : 4384 : if (overflow_policy != TaskOverflowPolicy::kBlock &&
+ + ]
853 : 29 : impl_->strategy.load(std::memory_order_acquire) != kBlockStrategy) {
854 : 22 : return false;
855 : : }
856 : :
857 [ + - ]: 4333 : std::unique_lock wait_lock(impl_->mtx);
858 : 4335 : ++impl_->block_waiter_count;
859 : :
860 : 49166 : auto capacity_available = [this, &is_cancelled, overflow_policy, &has_capacity] {
861 [ + - + - : 22414 : return impl_->quit_flag.load(std::memory_order_acquire) || is_cancelled() ||
+ + + + ]
862 [ + - + + ]: 10 : (overflow_policy != TaskOverflowPolicy::kBlock &&
863 [ + + + + ]: 22424 : impl_->strategy.load(std::memory_order_acquire) != kBlockStrategy) ||
864 [ + + + + ]: 22412 : has_capacity();
865 : : };
866 : :
867 [ + + ]: 4335 : if (poll_cancellation) {
868 [ + - ]: 29 : (void)impl_->capacity_cv.wait_for(wait_lock, std::chrono::milliseconds(1), capacity_available);
869 : : } else {
870 [ + - ]: 4306 : impl_->capacity_cv.wait(wait_lock, capacity_available);
871 : : }
872 : :
873 : 4335 : --impl_->block_waiter_count;
874 : :
875 : 4335 : return true;
876 : 132098 : };
877 : :
878 [ + + ]: 127763 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
879 [ + - ]: 5 : return reject();
880 : : }
881 : :
882 : 127759 : bool is_full = false;
883 : 127759 : bool should_notify = false;
884 : 127759 : int retry_cnt = 0;
885 : :
886 [ + + ]: 127759 : if (impl_->type == kNormalType) {
887 [ + + ]: 91592 : do {
888 : : {
889 [ + - ]: 91592 : std::lock_guard lock(impl_->mtx);
890 : :
891 [ + + ]: 91609 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
892 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
893 : : }
894 : :
895 [ + + ]: 91607 : if VUNLIKELY (is_cancelled()) {
896 : 1 : return false;
897 : : }
898 : :
899 [ + - ]: 91606 : is_full = impl_->normal_queue->size() >= get_max_task_count();
900 : :
901 [ + + ]: 91606 : if VLIKELY (!is_full) {
902 [ + - ]: 78575 : push_normal_task(std::move(callback), droppable);
903 [ + + ]: 13031 : } else if (overflow_policy == TaskOverflowPolicy::kReject) {
904 [ + - ]: 3 : return reject();
905 [ + + + + : 13028 : } else if (impl_->strategy.load(std::memory_order_acquire) == kPopStrategy &&
+ + ]
906 : : overflow_policy != TaskOverflowPolicy::kBlock) {
907 [ + - + + ]: 8684 : if (!drop_one_normal_task(dropped_tasks)) {
908 [ + - ]: 3 : return reject();
909 : : }
910 : :
911 [ + - ]: 8681 : push_normal_task(std::move(callback), droppable);
912 : :
913 : 8681 : is_full = false;
914 : : }
915 : :
916 [ + + ]: 87256 : if (!is_full &&
917 [ + + ]: 97335 : (impl_->is_running.load(std::memory_order_acquire) ||
918 [ + + + + ]: 188935 : impl_->manual_spin_waiting.load(std::memory_order_acquire)) &&
919 [ + + ]: 77178 : !impl_->wakeup_pending.load(std::memory_order_acquire)) {
920 : 13286 : impl_->wakeup_pending.store(true, std::memory_order_release);
921 : 13286 : should_notify = true;
922 : : }
923 [ + + ]: 91609 : }
924 : :
925 [ + + ]: 91596 : if VUNLIKELY (is_full) {
926 [ + - + + ]: 4344 : if VUNLIKELY (blocks_loop_thread()) {
927 [ + - ]: 1 : return reject();
928 : : }
929 : :
930 [ + - + + ]: 15540 : if (block_for_capacity([this] { return impl_->normal_queue->size() < get_max_task_count(); })) {
931 : 4330 : continue;
932 : : }
933 : :
934 [ + - + - : 11 : if (impl_->strategy.load(std::memory_order_acquire) == kOptimizationStrategy &&
+ - ]
935 : : overflow_policy != TaskOverflowPolicy::kBlock) {
936 [ + + ]: 11 : if (++retry_cnt > 10) {
937 : : {
938 [ + - ]: 1 : std::lock_guard lock(impl_->mtx);
939 : :
940 [ - + ]: 1 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
941 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
942 : : }
943 : :
944 [ - + ]: 1 : if VUNLIKELY (is_cancelled()) {
945 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
946 : : }
947 : :
948 [ + - - + ]: 1 : if (!drop_one_normal_task(dropped_tasks)) {
949 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
950 : : }
951 : :
952 [ + - ]: 1 : push_normal_task(std::move(callback), droppable);
953 : :
954 : 1 : is_full = false;
955 : :
956 [ - + ]: 2 : if ((impl_->is_running.load(std::memory_order_acquire) ||
957 [ + - - + ]: 2 : impl_->manual_spin_waiting.load(std::memory_order_acquire)) &&
958 [ # # ]: 0 : !impl_->wakeup_pending.load(std::memory_order_acquire)) {
959 : 0 : impl_->wakeup_pending.store(true, std::memory_order_release);
960 : 0 : should_notify = true;
961 : : }
962 [ + - ]: 1 : }
963 : :
964 [ + - + - : 2 : CLOG_W("MessageLoop: Task is full, removed top data (%s).", impl_->name.c_str());
+ - ]
965 : 1 : break;
966 : : }
967 : : }
968 : :
969 [ - + ]: 10 : if VUNLIKELY (is_cancelled()) {
970 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
971 : : }
972 : :
973 [ + - ]: 10 : std::this_thread::sleep_for(std::chrono::milliseconds(1));
974 : : }
975 : : } while (is_full);
976 : :
977 [ + + ]: 87253 : if (should_notify) {
978 : 13286 : impl_->cv.notify_all();
979 : : }
980 : :
981 : 87252 : return !is_full;
982 : :
983 [ + + ]: 40501 : } else if (impl_->type == kLockfreeType) {
984 : : struct ProducerGuard final {
985 : 10448 : explicit ProducerGuard(Impl& impl) noexcept : impl_ref(impl) {
986 : 10448 : impl_ref.lockfree_producer_count.fetch_add(1U, std::memory_order_acq_rel);
987 : 10448 : }
988 : :
989 : 10448 : ~ProducerGuard() {
990 [ + + ]: 20896 : if (impl_ref.lockfree_producer_count.fetch_sub(1U, std::memory_order_acq_rel) == 1U) {
991 : 10349 : std::lock_guard lock(impl_ref.mtx);
992 : 10349 : impl_ref.cv.notify_all();
993 : 10349 : }
994 : 10448 : }
995 : :
996 : : Impl& impl_ref;
997 : 10448 : } producer_guard(*impl_);
998 : :
999 : 31338 : auto push_reserved_lockfree_task = [this, &is_cancelled, &reject, &callback]() -> bool {
1000 [ - + ]: 10446 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
1001 : : release_lockfree_task(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1002 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1003 : : }
1004 : :
1005 [ - + ]: 10446 : if VUNLIKELY (is_cancelled()) {
1006 : : release_lockfree_task(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1007 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1008 : : }
1009 : :
1010 [ - + ]: 10446 : if VUNLIKELY (!push_lockfree_task(std::move(callback))) {
1011 : : release_lockfree_task(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1012 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1013 : : }
1014 : :
1015 : 10446 : return true;
1016 : 10448 : };
1017 : :
1018 [ + - ]: 11 : do {
1019 [ + + ]: 10459 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
1020 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1021 : : }
1022 : :
1023 [ - + ]: 10458 : if VUNLIKELY (is_cancelled()) {
1024 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1025 : : }
1026 : :
1027 [ + - ]: 10458 : is_full = !reserve_lockfree_task();
1028 : :
1029 [ + + ]: 10458 : if VLIKELY (!is_full) {
1030 [ + - - + ]: 10444 : if VUNLIKELY (!push_reserved_lockfree_task()) {
1031 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1032 : : }
1033 : :
1034 : 10444 : break;
1035 : : }
1036 : :
1037 [ - + ]: 14 : if (overflow_policy == TaskOverflowPolicy::kReject) {
1038 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1039 : : }
1040 : :
1041 [ + + + - : 14 : if (impl_->strategy.load(std::memory_order_acquire) == kPopStrategy &&
+ + ]
1042 : : overflow_policy != TaskOverflowPolicy::kBlock) {
1043 [ + - - + ]: 1 : if (!drop_one_lockfree_task(dropped_tasks, true)) {
1044 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1045 : : }
1046 : :
1047 [ + - - + ]: 1 : if VUNLIKELY (!push_reserved_lockfree_task()) {
1048 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1049 : : }
1050 : :
1051 : 1 : is_full = false;
1052 : :
1053 : 1 : break;
1054 : : }
1055 : :
1056 [ + - ]: 13 : if VUNLIKELY (is_full) {
1057 [ + - + + ]: 13 : if VUNLIKELY (blocks_loop_thread()) {
1058 [ + - ]: 1 : return reject();
1059 : : }
1060 : :
1061 [ + + + - : 12 : if (impl_->strategy.load(std::memory_order_acquire) == kOptimizationStrategy &&
+ + ]
1062 : : overflow_policy != TaskOverflowPolicy::kBlock) {
1063 [ + + ]: 11 : if (++retry_cnt > 10) {
1064 [ - + ]: 1 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
1065 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1066 : : }
1067 : :
1068 [ - + ]: 1 : if VUNLIKELY (is_cancelled()) {
1069 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1070 : : }
1071 : :
1072 [ + - - + ]: 1 : if (!drop_one_lockfree_task(dropped_tasks, true)) {
1073 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1074 : : }
1075 : :
1076 [ + - - + ]: 1 : if VUNLIKELY (!push_reserved_lockfree_task()) {
1077 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1078 : : }
1079 : :
1080 : 1 : is_full = false;
1081 [ + - + - : 2 : CLOG_W("MessageLoop: Task is full, removed top data (%s).", impl_->name.c_str());
+ - ]
1082 : 1 : break;
1083 : : }
1084 : : }
1085 : :
1086 [ - + ]: 11 : if VUNLIKELY (is_cancelled()) {
1087 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1088 : : }
1089 : :
1090 [ + - ]: 11 : std::this_thread::sleep_for(std::chrono::milliseconds(1));
1091 : : }
1092 : : } while (is_full);
1093 : :
1094 [ + - ]: 10446 : wakeup();
1095 : :
1096 : 10446 : return !is_full;
1097 [ + - ]: 40501 : } else if (impl_->type == kPriorityType) {
1098 [ + + ]: 30061 : do {
1099 : : {
1100 [ + - ]: 30066 : std::lock_guard lock(impl_->mtx);
1101 : :
1102 [ + + ]: 30066 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
1103 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1104 : : }
1105 : :
1106 [ - + ]: 30064 : if VUNLIKELY (is_cancelled()) {
1107 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1108 : : }
1109 : :
1110 : 30064 : is_full =
1111 [ + - ]: 30064 : impl_->priority_droppable_queue->size() + impl_->priority_protected_queue->size() >= get_max_task_count();
1112 : :
1113 [ + + ]: 30064 : if VLIKELY (!is_full) {
1114 [ + - ]: 30046 : push_priority_task(std::move(callback), priority, droppable);
1115 [ - + ]: 18 : } else if (overflow_policy == TaskOverflowPolicy::kReject) {
1116 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1117 [ + + + - : 18 : } else if (impl_->strategy.load(std::memory_order_acquire) == kPopStrategy &&
+ + ]
1118 : : overflow_policy != TaskOverflowPolicy::kBlock) {
1119 [ + - + + ]: 3 : if (!drop_one_priority_task(dropped_tasks)) {
1120 [ + - ]: 1 : return reject();
1121 : : }
1122 : :
1123 [ + - ]: 2 : push_priority_task(std::move(callback), priority, droppable);
1124 : :
1125 : 2 : is_full = false;
1126 : : }
1127 : :
1128 [ + + ]: 30048 : if (!is_full &&
1129 [ + + ]: 40064 : (impl_->is_running.load(std::memory_order_acquire) ||
1130 [ + + + + ]: 70127 : impl_->manual_spin_waiting.load(std::memory_order_acquire)) &&
1131 [ + + ]: 20033 : !impl_->wakeup_pending.load(std::memory_order_acquire)) {
1132 : 894 : impl_->wakeup_pending.store(true, std::memory_order_release);
1133 : 894 : should_notify = true;
1134 : : }
1135 [ + + ]: 30066 : }
1136 : :
1137 [ + + ]: 30063 : if VUNLIKELY (is_full) {
1138 [ + - + + ]: 15 : if VUNLIKELY (blocks_loop_thread()) {
1139 [ + - ]: 1 : return reject();
1140 : : }
1141 : :
1142 [ + - + + ]: 14 : if (block_for_capacity([this] {
1143 : 4 : return impl_->priority_droppable_queue->size() + impl_->priority_protected_queue->size() <
1144 : 4 : get_max_task_count();
1145 : : })) {
1146 : 3 : continue;
1147 : : }
1148 : :
1149 [ + - + - : 11 : if (impl_->strategy.load(std::memory_order_acquire) == kOptimizationStrategy &&
+ - ]
1150 : : overflow_policy != TaskOverflowPolicy::kBlock) {
1151 [ + + ]: 11 : if (++retry_cnt > 10) {
1152 : : {
1153 [ + - ]: 1 : std::lock_guard lock(impl_->mtx);
1154 : :
1155 [ - + ]: 1 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
1156 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1157 : : }
1158 : :
1159 [ - + ]: 1 : if VUNLIKELY (is_cancelled()) {
1160 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1161 : : }
1162 : :
1163 [ + - - + ]: 1 : if (!drop_one_priority_task(dropped_tasks)) {
1164 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1165 : : }
1166 : :
1167 [ + - ]: 1 : push_priority_task(std::move(callback), priority, droppable);
1168 : 1 : is_full = false;
1169 : :
1170 [ - + ]: 2 : if ((impl_->is_running.load(std::memory_order_acquire) ||
1171 [ + - - + ]: 2 : impl_->manual_spin_waiting.load(std::memory_order_acquire)) &&
1172 [ # # ]: 0 : !impl_->wakeup_pending.load(std::memory_order_acquire)) {
1173 : 0 : impl_->wakeup_pending.store(true, std::memory_order_release);
1174 : 0 : should_notify = true;
1175 : : }
1176 [ + - ]: 1 : }
1177 : :
1178 [ + - + - : 2 : CLOG_W("MessageLoop: Task is full, removed top data (%s).", impl_->name.c_str());
+ - ]
1179 : 1 : break;
1180 : : }
1181 : : }
1182 : :
1183 [ - + ]: 10 : if VUNLIKELY (is_cancelled()) {
1184 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1185 : : }
1186 : :
1187 [ + - ]: 10 : std::this_thread::sleep_for(std::chrono::milliseconds(1));
1188 : : }
1189 : : } while (is_full);
1190 : :
1191 [ + + ]: 30049 : if (should_notify) {
1192 : 894 : impl_->cv.notify_all();
1193 : : }
1194 : :
1195 : 30049 : return !is_full;
1196 : : }
1197 : :
1198 : : return reject(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1199 : 127766 : }
1200 : :
1201 : 8746 : bool MessageLoop::drop_one_normal_task(std::vector<Callback>& dropped) {
1202 [ + + ]: 195871 : for (auto iter = impl_->normal_queue->begin(); iter != impl_->normal_queue->end(); ++iter) {
1203 [ + + ]: 195807 : if (std::get<1>(*iter)) {
1204 [ + - ]: 8682 : dropped.emplace_back(std::move(std::get<2>(*iter)));
1205 [ + - ]: 8682 : impl_->normal_queue->erase(iter);
1206 : :
1207 : 8682 : return true;
1208 : : }
1209 : : }
1210 : :
1211 : 64 : return false;
1212 : : }
1213 : :
1214 : 2 : bool MessageLoop::drop_one_lockfree_task(std::vector<Callback>& dropped, bool keep_reserved) {
1215 : 2 : Impl::LockfreeTaskTuple task;
1216 : :
1217 [ - + ]: 2 : if (!impl_->lockfree_queue->try_pop<Impl::LockfreeQueue::kNoBehavior>(task)) {
1218 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1219 : : }
1220 : :
1221 [ + - ]: 2 : dropped.emplace_back(std::move(std::get<1>(task)));
1222 : :
1223 [ - + ]: 2 : if (!keep_reserved) {
1224 : : release_lockfree_task(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1225 : : }
1226 : :
1227 : 2 : return true;
1228 : 2 : }
1229 : :
1230 : 4 : bool MessageLoop::drop_one_priority_task(std::vector<Callback>& dropped) {
1231 : 4 : auto& container = impl_->priority_droppable_queue->container();
1232 : :
1233 [ + + ]: 4 : if (container.empty()) {
1234 : 1 : return false;
1235 : : }
1236 : :
1237 : 3 : auto victim = container.begin();
1238 : :
1239 [ + - + + ]: 4 : for (auto iter = std::next(container.begin()); iter != container.end(); ++iter) {
1240 : 1 : const uint32_t victim_rank = std::get<0>(*victim);
1241 : 1 : const uint32_t iter_rank = std::get<0>(*iter);
1242 : :
1243 [ - + - - : 1 : if (iter_rank > victim_rank || (iter_rank == victim_rank && std::get<1>(*iter) < std::get<1>(*victim))) {
- - + - ]
1244 : 1 : victim = iter;
1245 : : }
1246 : : }
1247 : :
1248 [ + - ]: 3 : dropped.emplace_back(std::move(std::get<4>(*victim)));
1249 [ + - ]: 3 : container.erase(victim);
1250 : :
1251 [ + - ]: 3 : std::make_heap(container.begin(), container.end(), Impl::PriorityCompare{});
1252 : :
1253 : 3 : return true;
1254 : : }
1255 : :
1256 : 10458 : bool MessageLoop::reserve_lockfree_task() {
1257 : 10458 : auto count = impl_->lockfree_task_count.load(std::memory_order_acquire);
1258 [ + - ]: 10458 : const auto max_count = get_max_task_count();
1259 : :
1260 [ + + ]: 10478 : while (count < max_count) {
1261 [ + + ]: 20928 : if (impl_->lockfree_task_count.compare_exchange_weak(count, count + 1U, std::memory_order_acq_rel,
1262 : : std::memory_order_acquire)) {
1263 : 10444 : return true;
1264 : : }
1265 : : }
1266 : :
1267 : 14 : return false;
1268 : : }
1269 : :
1270 : 10443 : void MessageLoop::release_lockfree_task() { impl_->lockfree_task_count.fetch_sub(1U, std::memory_order_acq_rel); }
1271 : :
1272 : 88605 : void MessageLoop::push_normal_task(Callback&& callback, bool droppable) {
1273 : 88605 : uint32_t start_time = 0;
1274 : :
1275 [ + - + + ]: 88605 : if (get_max_elapsed_time() > 0) {
1276 : 690 : start_time = get_current_time<std::chrono::steady_clock, std::chrono::milliseconds, uint32_t>();
1277 : : }
1278 : :
1279 [ + - ]: 88605 : impl_->normal_queue->emplace_back(start_time, droppable, std::move(callback));
1280 : 88605 : }
1281 : :
1282 : 10446 : bool MessageLoop::push_lockfree_task(Callback&& callback) {
1283 : 10446 : uint32_t start_time = 0;
1284 : :
1285 [ + - - + ]: 10446 : if (get_max_elapsed_time() > 0) {
1286 : 0 : start_time =
1287 : : get_current_time<std::chrono::steady_clock, std::chrono::milliseconds, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1288 : : uint32_t>(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1289 : : }
1290 : :
1291 [ + - ]: 10446 : for (int retry = 0; retry < kMaxLockfreePushRetry; ++retry) {
1292 [ + - ]: 10446 : if VLIKELY (impl_->lockfree_queue->try_push<Impl::LockfreeQueue::kNoBehavior>(
1293 : : std::forward_as_tuple(start_time, std::move(callback)))) {
1294 : 10446 : return true;
1295 : : }
1296 : :
1297 : : Utils::yield_cpu(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1298 : : }
1299 : :
1300 : : // LCOV_EXCL_START GCOVR_EXCL_START
1301 : : CLOG_E("MessageLoop: Failed to push lockfree task after %d retries (%s).", kMaxLockfreePushRetry,
1302 : : impl_->name.c_str());
1303 : : return false;
1304 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1305 : : }
1306 : :
1307 : 30050 : void MessageLoop::push_priority_task(Callback&& callback, uint16_t priority, bool droppable) {
1308 : 30050 : uint32_t start_time = 0;
1309 : :
1310 [ + - - + ]: 30050 : if (get_max_elapsed_time() > 0) {
1311 : 0 : start_time =
1312 : : get_current_time<std::chrono::steady_clock, std::chrono::milliseconds, // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1313 : : uint32_t>(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1314 : : }
1315 : :
1316 [ + + ]: 30050 : auto& queue = droppable ? impl_->priority_droppable_queue : impl_->priority_protected_queue;
1317 [ + + ]: 30050 : const uint16_t effective_priority = priority == kNoPriority ? static_cast<uint16_t>(kNormalPriority) : priority;
1318 [ + - ]: 60100 : queue->emplace(std::numeric_limits<uint16_t>::max() - effective_priority, impl_->task_seq, start_time, droppable,
1319 : 30050 : std::move(callback));
1320 : 30050 : ++impl_->task_seq;
1321 : 30050 : }
1322 : :
1323 : 809 : void MessageLoop::do_consume() {
1324 [ + - ]: 809 : std::unique_lock lock(impl_->mtx);
1325 : :
1326 : 809 : impl_->is_running.store(true, std::memory_order_release);
1327 : 809 : impl_->is_busy.store(true, std::memory_order_release);
1328 : 809 : impl_->thread_id.store(std::this_thread::get_id(), std::memory_order_release);
1329 : :
1330 : : #ifdef _WIN32
1331 : : HANDLE thread_handle = ::OpenThread(SYNCHRONIZE, FALSE, ::GetCurrentThreadId());
1332 : : HANDLE expected_thread_handle = nullptr;
1333 : :
1334 : : if (thread_handle != nullptr &&
1335 : : !impl_->thread_handle.compare_exchange_strong(expected_thread_handle, thread_handle, std::memory_order_acq_rel,
1336 : : std::memory_order_acquire)) {
1337 : : ::CloseHandle(thread_handle);
1338 : : }
1339 : : #endif
1340 : :
1341 [ + - ]: 809 : lock.unlock();
1342 : :
1343 [ + - ]: 809 : on_begin();
1344 : :
1345 [ + + ]: 809 : if (impl_->type == kNormalType) {
1346 [ + + ]: 781 : if (!impl_->normal_staging_queue.has_value()) {
1347 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
1348 [ + - + - ]: 777 : impl_->normal_staging_queue.emplace(&MemoryResource::global_instance());
1349 : : #else
1350 : : impl_->normal_staging_queue.emplace();
1351 : : #endif
1352 : : }
1353 : :
1354 [ + - + + ]: 17757 : while (process_normal_task(true, true)) {
1355 : : }
1356 [ + + ]: 28 : } else if (impl_->type == kLockfreeType) {
1357 [ + - + + ]: 114 : while (process_lockfree_task(true)) {
1358 : : }
1359 [ + - ]: 16 : } else if (impl_->type == kPriorityType) {
1360 [ + - + + ]: 926 : while (process_priority_task(true)) {
1361 : : }
1362 : : }
1363 : :
1364 [ + - ]: 809 : on_end();
1365 : :
1366 [ + - ]: 809 : lock.lock();
1367 : :
1368 : 808 : impl_->thread_id.store(std::thread::id(), std::memory_order_release);
1369 : 809 : impl_->is_running.store(false, std::memory_order_release);
1370 : 809 : impl_->is_busy.store(false, std::memory_order_release);
1371 : 808 : impl_->cv.notify_all();
1372 : :
1373 [ + - ]: 809 : lock.unlock();
1374 : 809 : }
1375 : :
1376 : 17775 : bool MessageLoop::process_normal_task(bool block, bool reuse_queue) {
1377 : 17775 : impl_->is_busy.store(true, std::memory_order_release);
1378 : :
1379 : 17775 : [[maybe_unused]] bool is_timeout = true;
1380 : 17775 : int64_t sleep_time = -1;
1381 : :
1382 : 17775 : std::vector<Callback> dropped_tasks;
1383 : :
1384 : 17774 : std::optional<Impl::NormalQueue> local_queue;
1385 : :
1386 [ + + ]: 17774 : if (!reuse_queue) {
1387 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
1388 [ + - + - ]: 18 : local_queue.emplace(&MemoryResource::global_instance());
1389 : : #else
1390 : : local_queue.emplace();
1391 : : #endif
1392 : : }
1393 : :
1394 [ + + + - : 17774 : auto& temp_queue = reuse_queue ? impl_->normal_staging_queue.value() : local_queue.value();
+ - ]
1395 : :
1396 [ + - ]: 17775 : std::unique_lock lock(impl_->mtx);
1397 : :
1398 : 17775 : impl_->task_seq = 0;
1399 [ + - ]: 17774 : temp_queue.swap(impl_->normal_queue.value());
1400 : 17774 : const bool notify_producers = impl_->block_waiter_count > 0U;
1401 : :
1402 [ + - ]: 17774 : lock.unlock();
1403 : :
1404 [ + + ]: 17775 : if VUNLIKELY (notify_producers) {
1405 : 3256 : impl_->capacity_cv.notify_all();
1406 : : }
1407 : :
1408 [ + + + + : 87631 : while (!temp_queue.empty() && !impl_->force_quit_flag.load(std::memory_order_acquire)) {
+ + ]
1409 : 69858 : auto&& [start_time, droppable, task] = std::move(const_cast<Impl::NormalTaskTuple&>(temp_queue.front()));
1410 : :
1411 : : {
1412 : 69858 : Impl::RestoreDropPolicy restore{impl_->current_task_droppable,
1413 [ + - ]: 69859 : std::exchange(impl_->current_task_droppable, droppable)};
1414 [ + + ]: 69859 : on_task_changed(std::move(task), start_time);
1415 : 69857 : }
1416 : :
1417 : 69857 : temp_queue.pop_front();
1418 : : }
1419 : :
1420 [ + - ]: 17774 : on_idle();
1421 : :
1422 [ + - ]: 17773 : lock.lock();
1423 : :
1424 [ + + ]: 17774 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
1425 [ + - ]: 781 : lock.unlock();
1426 [ + - ]: 781 : drop_pending_tasks();
1427 : 781 : temp_queue.clear();
1428 : 781 : return false;
1429 : : }
1430 : :
1431 [ + + ]: 16993 : if (block) {
1432 : 16977 : impl_->manual_spin_waiting.store(!impl_->is_running.load(std::memory_order_acquire), std::memory_order_release);
1433 : : }
1434 : :
1435 [ + + ]: 16993 : if (!impl_->timer_set.empty()) {
1436 [ + - ]: 9608 : process_timer_task(sleep_time, dropped_tasks);
1437 : : }
1438 : :
1439 : 16993 : impl_->is_busy.store(false, std::memory_order_release);
1440 : 16993 : impl_->cv.notify_all();
1441 : :
1442 [ + + ]: 16993 : if (block) {
1443 : 76371 : auto predicate = [this]() -> bool {
1444 [ + - ]: 45155 : return impl_->quit_flag.load(std::memory_order_acquire) || impl_->is_busy.load(std::memory_order_acquire) ||
1445 [ + + + + : 45155 : !impl_->normal_queue->empty() || impl_->wakeup_pending.load(std::memory_order_acquire);
+ + ]
1446 : 16977 : };
1447 : :
1448 [ + + ]: 16977 : if (sleep_time < 0) {
1449 : 7756 : impl_->cv.wait(lock, std::move(predicate));
1450 : 7756 : is_timeout = false;
1451 [ + + ]: 9221 : } else if (sleep_time > 0) {
1452 [ + - ]: 7907 : is_timeout = !impl_->cv.wait_for(lock, std::chrono::nanoseconds(sleep_time), std::move(predicate));
1453 : : }
1454 : :
1455 : 16977 : impl_->wakeup_pending.store(false, std::memory_order_release);
1456 : 16977 : impl_->manual_spin_waiting.store(false, std::memory_order_release);
1457 : : }
1458 : :
1459 : 16993 : return true;
1460 : 17777 : }
1461 : :
1462 : 117 : bool MessageLoop::process_lockfree_task(bool block) {
1463 : 117 : impl_->is_busy.store(true, std::memory_order_release);
1464 : :
1465 : 117 : [[maybe_unused]] bool is_timeout = true;
1466 : :
1467 : 117 : std::vector<Callback> dropped_tasks;
1468 : :
1469 [ + - ]: 557 : while (!impl_->force_quit_flag.load(std::memory_order_acquire)) {
1470 : 557 : Impl::LockfreeTaskTuple temp_task;
1471 : :
1472 [ + + ]: 557 : if (!impl_->lockfree_queue->try_pop<Impl::LockfreeQueue::kNoBehavior>(temp_task)) {
1473 : 117 : break;
1474 : : }
1475 : :
1476 [ + - ]: 440 : release_lockfree_task();
1477 : :
1478 : 440 : auto&& [start_time, task] = std::move(temp_task);
1479 : :
1480 [ + - ]: 440 : on_task_changed(std::move(task), start_time);
1481 [ + + ]: 557 : }
1482 : :
1483 [ + - ]: 117 : on_idle();
1484 : :
1485 [ + - ]: 117 : std::unique_lock lock(impl_->mtx);
1486 : :
1487 : 117 : int64_t sleep_time = -1;
1488 : :
1489 [ + + ]: 117 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
1490 [ + - ]: 12 : lock.unlock();
1491 [ + - ]: 12 : drop_pending_tasks();
1492 : 12 : return false;
1493 : : }
1494 : :
1495 [ + + ]: 105 : if (block) {
1496 : 103 : impl_->manual_spin_waiting.store(!impl_->is_running.load(std::memory_order_acquire), std::memory_order_release);
1497 : : }
1498 : :
1499 [ - + ]: 105 : if (!impl_->timer_set.empty()) {
1500 : : process_timer_task(sleep_time, dropped_tasks); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1501 : : }
1502 : :
1503 : 105 : impl_->is_busy.store(false, std::memory_order_release);
1504 : 105 : impl_->cv.notify_all();
1505 : :
1506 [ + + ]: 105 : if (block) {
1507 : 445 : auto predicate = [this]() -> bool {
1508 [ + - ]: 270 : return impl_->quit_flag.load(std::memory_order_acquire) || impl_->is_busy.load(std::memory_order_acquire) ||
1509 [ + + + + : 445 : impl_->lockfree_task_count.load(std::memory_order_acquire) != 0U ||
+ + ]
1510 : 187 : impl_->wakeup_pending.load(std::memory_order_acquire);
1511 : 103 : };
1512 : :
1513 [ + - ]: 103 : if (sleep_time < 0) {
1514 : 103 : impl_->cv.wait(lock, std::move(predicate));
1515 : 103 : is_timeout = false;
1516 : : } else if (sleep_time > 0) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1517 : : is_timeout = !impl_->cv.wait_for(lock, std::chrono::nanoseconds(sleep_time), // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1518 : : std::move(predicate)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1519 : : }
1520 : :
1521 : 103 : impl_->wakeup_pending.store(false, std::memory_order_release);
1522 : 103 : impl_->manual_spin_waiting.store(false, std::memory_order_release);
1523 : : }
1524 : :
1525 : 105 : return true;
1526 : 117 : }
1527 : :
1528 : 931 : bool MessageLoop::process_priority_task(bool block) {
1529 : 931 : impl_->is_busy.store(true, std::memory_order_release);
1530 : :
1531 : 931 : [[maybe_unused]] bool is_timeout = true;
1532 : 931 : int64_t sleep_time = -1;
1533 : :
1534 : 931 : std::vector<Callback> dropped_tasks;
1535 : :
1536 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
1537 [ + - + - ]: 931 : Impl::PriorityQueue temp_queue(&MemoryResource::global_instance());
1538 [ + - + - ]: 931 : Impl::PriorityQueue temp_protected_queue(&MemoryResource::global_instance());
1539 : : #else
1540 : : Impl::PriorityQueue temp_queue;
1541 : : Impl::PriorityQueue temp_protected_queue;
1542 : : #endif
1543 : :
1544 [ + - ]: 931 : std::unique_lock lock(impl_->mtx);
1545 : :
1546 : 931 : impl_->task_seq = 0;
1547 [ + - ]: 931 : temp_queue.swap(impl_->priority_droppable_queue.value());
1548 [ + - ]: 931 : temp_protected_queue.swap(impl_->priority_protected_queue.value());
1549 : 931 : const bool notify_producers = impl_->block_waiter_count > 0U;
1550 : :
1551 [ + - ]: 931 : lock.unlock();
1552 : :
1553 [ + + ]: 931 : if VUNLIKELY (notify_producers) {
1554 : 1 : impl_->capacity_cv.notify_all();
1555 : : }
1556 : :
1557 [ + + + + : 41011 : while ((!temp_queue.empty() || !temp_protected_queue.empty()) &&
+ + ]
1558 [ + - ]: 20040 : !impl_->force_quit_flag.load(std::memory_order_acquire)) {
1559 : : const bool drain_protected =
1560 [ + + ]: 30078 : temp_queue.empty() ||
1561 [ + + + - ]: 10038 : (!temp_protected_queue.empty() && !Impl::priority_before(temp_queue.top(), temp_protected_queue.top()));
1562 [ + + ]: 20040 : auto& selected_queue = drain_protected ? temp_protected_queue : temp_queue;
1563 : 20040 : auto&& [priority, seq, start_time, droppable, task] =
1564 : 20040 : std::move(const_cast<Impl::PriorityTaskTuple&>(selected_queue.top()));
1565 : :
1566 : : {
1567 : 20040 : Impl::RestoreDropPolicy restore{impl_->current_task_droppable,
1568 [ + - ]: 20040 : std::exchange(impl_->current_task_droppable, droppable)};
1569 [ + - ]: 20040 : on_task_changed(std::move(task), start_time);
1570 : 20040 : }
1571 : :
1572 [ + - ]: 20040 : selected_queue.pop();
1573 : : }
1574 : :
1575 [ + - ]: 931 : on_idle();
1576 : :
1577 [ + - ]: 931 : lock.lock();
1578 : :
1579 [ + + ]: 931 : if VUNLIKELY (impl_->quit_flag.load(std::memory_order_acquire)) {
1580 [ + - ]: 16 : lock.unlock();
1581 [ + - ]: 16 : drop_pending_tasks();
1582 : 16 : return false;
1583 : : }
1584 : :
1585 [ + + ]: 915 : if (block) {
1586 : 911 : impl_->manual_spin_waiting.store(!impl_->is_running.load(std::memory_order_acquire), std::memory_order_release);
1587 : : }
1588 : :
1589 [ + + ]: 915 : if (!impl_->timer_set.empty()) {
1590 [ + - ]: 4 : process_timer_task(sleep_time, dropped_tasks);
1591 : : }
1592 : :
1593 : 915 : impl_->is_busy.store(false, std::memory_order_release);
1594 : 915 : impl_->cv.notify_all();
1595 : :
1596 [ + + ]: 915 : if (block) {
1597 : 3814 : auto predicate = [this]() -> bool {
1598 [ + - ]: 1905 : return impl_->quit_flag.load(std::memory_order_acquire) || impl_->is_busy.load(std::memory_order_acquire) ||
1599 [ + + + + : 1960 : !impl_->priority_droppable_queue->empty() || !impl_->priority_protected_queue->empty() ||
+ + + + ]
1600 : 1015 : impl_->wakeup_pending.load(std::memory_order_acquire);
1601 : 911 : };
1602 : :
1603 [ + + ]: 911 : if (sleep_time < 0) {
1604 [ + - ]: 910 : impl_->cv.wait(lock, std::move(predicate));
1605 : 910 : is_timeout = false;
1606 [ + - ]: 1 : } else if (sleep_time > 0) {
1607 [ + - ]: 1 : is_timeout = !impl_->cv.wait_for(lock, std::chrono::nanoseconds(sleep_time), std::move(predicate));
1608 : : }
1609 : :
1610 : 911 : impl_->wakeup_pending.store(false, std::memory_order_release);
1611 : 911 : impl_->manual_spin_waiting.store(false, std::memory_order_release);
1612 : : }
1613 : :
1614 : 915 : return true;
1615 : 931 : }
1616 : :
1617 : 9612 : bool MessageLoop::process_timer_task(int64_t& next_sleep_time, std::vector<Callback>& dropped_tasks) {
1618 : 9612 : int64_t invoke_count = 0;
1619 : 9612 : int64_t remain_loop_count = 0;
1620 : 9612 : int64_t interval_time = 0;
1621 : 9612 : int64_t remain_time = 0;
1622 : 9612 : uint64_t processed_invoke_count = 0;
1623 : 9612 : bool has_erase = false;
1624 : 9612 : bool has_processed = false;
1625 : :
1626 : 9612 : next_sleep_time = -1;
1627 : :
1628 [ + + ]: 19391 : for (auto iter = impl_->timer_set.begin(); iter != impl_->timer_set.end();) {
1629 : 9778 : Timer* timer = *iter;
1630 : :
1631 [ + - + + ]: 9779 : if (!timer->is_active()) {
1632 : 372 : ++iter;
1633 : 372 : continue;
1634 : : }
1635 : :
1636 [ + - - + ]: 9405 : if VUNLIKELY (!timer->has_callback()) {
1637 : : timer->stop(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1638 : : ++iter; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1639 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1640 : : }
1641 : :
1642 [ + - ]: 9406 : const uint64_t timer_generation = timer->get_generation();
1643 : :
1644 [ + - ]: 9407 : const uint32_t interval = timer->get_interval();
1645 [ + + ]: 9407 : interval_time = interval == 0 ? Timer::kMinInterval : static_cast<uint64_t>(interval) * 1000'000U;
1646 : :
1647 [ + - ]: 9407 : uint64_t start_time = timer->get_start_time();
1648 [ + - ]: 9405 : uint64_t current_time = get_current_nano_time();
1649 : :
1650 [ - + ]: 9407 : if VUNLIKELY (current_time < start_time) {
1651 : : // LCOV_EXCL_START GCOVR_EXCL_START
1652 : : remain_time = static_cast<int64_t>((start_time - current_time) + static_cast<uint64_t>(interval_time) -
1653 : : Timer::kMinInterval);
1654 : :
1655 : : if (remain_time < 0) {
1656 : : remain_time = 0;
1657 : : }
1658 : :
1659 : : if (next_sleep_time < 0 || next_sleep_time > remain_time) {
1660 : : next_sleep_time = remain_time;
1661 : : }
1662 : :
1663 : : ++iter;
1664 : : continue;
1665 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1666 : : }
1667 : :
1668 [ + - ]: 9407 : processed_invoke_count = timer->get_invoke_count();
1669 : 9406 : invoke_count = (current_time - start_time + Timer::kMinInterval) / interval_time;
1670 : :
1671 : 9406 : remain_loop_count = invoke_count - processed_invoke_count;
1672 : :
1673 [ + + ]: 9406 : if (remain_loop_count > 0) {
1674 : 1410 : has_erase = false;
1675 : 1410 : bool capacity_blocked = false;
1676 : :
1677 [ + - ]: 1410 : auto alive_flag = timer->get_alive_flag();
1678 : 10680 : auto run_timer_callback = [this, timer, alive_flag, timer_generation]() {
1679 [ + + ]: 1337 : if VUNLIKELY (!alive_flag->load(std::memory_order_acquire)) {
1680 : 1 : return;
1681 : : }
1682 : :
1683 : : {
1684 [ + - ]: 1336 : std::lock_guard timer_lock(impl_->mtx);
1685 : :
1686 [ + - + - : 1336 : if VUNLIKELY (!alive_flag->load(std::memory_order_acquire) || impl_->timer_set.count(timer) == 0 ||
- + + - +
- + + +
+ ]
1687 : : timer->get_generation() != timer_generation) {
1688 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1689 : : }
1690 : :
1691 [ + - ]: 1333 : timer->begin_in_flight();
1692 [ + + ]: 1336 : }
1693 : :
1694 : 1333 : timer->run_callback();
1695 : 1333 : timer->end_in_flight();
1696 : 1410 : };
1697 : :
1698 [ + - + - : 1410 : for (int64_t i = 0; timer->get_remain_loop_count() != 0 && i < remain_loop_count; ++i) {
+ - + - ]
1699 [ + + ]: 1410 : if (impl_->type == kNormalType) {
1700 [ + - + + ]: 1409 : if VUNLIKELY (impl_->normal_queue->size() >= get_max_task_count()) {
1701 : : // LCOV_EXCL_START GCOVR_EXCL_START
1702 : : if (!drop_one_normal_task(dropped_tasks)) {
1703 : : CLOG_W("MessageLoop: Timer task is full and no task can be dropped (%s).", impl_->name.c_str());
1704 : : capacity_blocked = true;
1705 : : break;
1706 : : }
1707 : :
1708 : : CLOG_W("MessageLoop: Timer task is full, removed top data (%s).", impl_->name.c_str());
1709 : :
1710 : : std::this_thread::sleep_for(std::chrono::milliseconds(1));
1711 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1712 : : }
1713 : :
1714 [ + - + + ]: 1348 : if (timer->is_once_type()) {
1715 [ + - + - ]: 11 : push_normal_task(timer->take_callback());
1716 : : } else {
1717 [ + - + - ]: 1337 : push_normal_task(run_timer_callback);
1718 : : }
1719 : :
1720 : 1348 : ++processed_invoke_count;
1721 : 1348 : has_processed = true;
1722 [ - + ]: 1 : } else if (impl_->type == kLockfreeType) {
1723 : : // LCOV_EXCL_START GCOVR_EXCL_START
1724 : : if VUNLIKELY (!reserve_lockfree_task()) {
1725 : : if (!drop_one_lockfree_task(dropped_tasks, true)) {
1726 : : CLOG_W("MessageLoop: Timer task is full and no task can be dropped (%s).", impl_->name.c_str());
1727 : : capacity_blocked = true;
1728 : : break;
1729 : : }
1730 : :
1731 : : CLOG_W("MessageLoop: Timer task is full, removed top data (%s).", impl_->name.c_str());
1732 : : }
1733 : :
1734 : : bool pushed = false;
1735 : :
1736 : : if (timer->is_once_type()) {
1737 : : pushed = push_lockfree_task(timer->take_callback());
1738 : : } else {
1739 : : pushed = push_lockfree_task(run_timer_callback);
1740 : : }
1741 : :
1742 : : if VUNLIKELY (!pushed) {
1743 : : release_lockfree_task();
1744 : : capacity_blocked = true;
1745 : : break;
1746 : : }
1747 : :
1748 : : ++processed_invoke_count;
1749 : : has_processed = true;
1750 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1751 [ + - ]: 1 : } else if (impl_->type == kPriorityType) {
1752 [ + - - + ]: 1 : if VUNLIKELY (impl_->priority_droppable_queue->size() + impl_->priority_protected_queue->size() >=
1753 : : get_max_task_count()) {
1754 : : // LCOV_EXCL_START GCOVR_EXCL_START
1755 : : if (!drop_one_priority_task(dropped_tasks)) {
1756 : : CLOG_W("MessageLoop: Timer task is full and no droppable task exists (%s).", impl_->name.c_str());
1757 : : capacity_blocked = true;
1758 : : break;
1759 : : }
1760 : :
1761 : : CLOG_W("MessageLoop: Timer task is full, removed top data (%s).", impl_->name.c_str());
1762 : :
1763 : : std::this_thread::sleep_for(std::chrono::milliseconds(1));
1764 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
1765 : : }
1766 : :
1767 [ + - + - ]: 1 : if (timer->is_once_type()) {
1768 [ + - + - : 1 : push_priority_task(timer->take_callback(), timer->get_priority());
+ - ]
1769 : : } else {
1770 : : push_priority_task(run_timer_callback, timer->get_priority()); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1771 : : }
1772 : :
1773 : 1 : ++processed_invoke_count;
1774 : 1 : has_processed = true;
1775 : : }
1776 : :
1777 [ + - + + ]: 1349 : if (timer->is_once_type()) {
1778 [ + - ]: 12 : iter = impl_->timer_set.erase(iter);
1779 : 12 : timer->~Timer();
1780 [ + - ]: 12 : MemoryPool::global_instance().deallocate(timer, sizeof(Timer), alignof(Timer));
1781 : 12 : has_erase = true;
1782 : 12 : break;
1783 : : }
1784 : :
1785 [ + - ]: 1337 : timer->sub_remain_loop_count();
1786 : :
1787 [ + - + - ]: 1337 : if (!timer->is_strict()) {
1788 : 1337 : break;
1789 : : }
1790 : : }
1791 : :
1792 [ + + ]: 1410 : if VUNLIKELY (has_erase) {
1793 : 12 : continue;
1794 : : }
1795 : :
1796 [ + - + + ]: 1398 : if (timer->get_remain_loop_count() == 0) {
1797 [ + - ]: 13 : timer->set_invoke_count(processed_invoke_count);
1798 [ + - ]: 13 : timer->stop(false);
1799 : : } else {
1800 [ + + + - ]: 1385 : timer->set_invoke_count(capacity_blocked
1801 : : ? processed_invoke_count
1802 : : : static_cast<uint64_t>(invoke_count)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1803 : : }
1804 : :
1805 : 1398 : next_sleep_time = 0;
1806 : :
1807 [ + + + + ]: 1422 : } else {
1808 : 7996 : const auto elapsed_in_period =
1809 : 7996 : static_cast<int64_t>((current_time - start_time) % static_cast<uint64_t>(interval_time));
1810 : :
1811 : 7996 : remain_time = interval_time - elapsed_in_period;
1812 : :
1813 [ + + ]: 7996 : if (remain_time > static_cast<int64_t>(Timer::kMinInterval)) {
1814 : 7986 : remain_time -= static_cast<int64_t>(Timer::kMinInterval);
1815 : : }
1816 : :
1817 [ + + + - ]: 7996 : if (next_sleep_time < 0 || next_sleep_time > remain_time) {
1818 : 7996 : next_sleep_time = remain_time;
1819 : : }
1820 : : }
1821 : :
1822 : 9394 : ++iter;
1823 : : }
1824 : :
1825 : 9612 : return has_processed;
1826 : : }
1827 : :
1828 : 3723 : void MessageLoop::drop_pending_tasks() {
1829 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
1830 [ + - + - ]: 3723 : Impl::NormalQueue normal_queue(&MemoryResource::global_instance());
1831 [ + - + - ]: 3722 : Impl::PriorityQueue priority_droppable_queue(&MemoryResource::global_instance());
1832 [ + - + - ]: 3722 : Impl::PriorityQueue priority_protected_queue(&MemoryResource::global_instance());
1833 : : #else
1834 : : Impl::NormalQueue normal_queue;
1835 : : Impl::PriorityQueue priority_droppable_queue;
1836 : : Impl::PriorityQueue priority_protected_queue;
1837 : : #endif
1838 : :
1839 : 3722 : std::vector<Impl::LockfreeTaskTuple> lockfree_tasks;
1840 : :
1841 : : {
1842 [ + - ]: 3723 : std::lock_guard lock(impl_->mtx);
1843 : :
1844 [ + + ]: 3723 : if (impl_->type == kNormalType) {
1845 [ + - ]: 3613 : normal_queue.swap(impl_->normal_queue.value());
1846 [ + + ]: 110 : } else if (impl_->type == kLockfreeType) {
1847 : 47 : Impl::LockfreeTaskTuple task;
1848 [ + + ]: 10050 : while (impl_->lockfree_queue->try_pop<Impl::LockfreeQueue::kNoBehavior>(task)) {
1849 : : release_lockfree_task(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1850 : : lockfree_tasks.emplace_back(std::move(task)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
1851 : : }
1852 [ + - ]: 110 : } else if (impl_->type == kPriorityType) {
1853 [ + - ]: 63 : priority_droppable_queue.swap(impl_->priority_droppable_queue.value());
1854 [ + - ]: 63 : priority_protected_queue.swap(impl_->priority_protected_queue.value());
1855 : : }
1856 : 3723 : }
1857 : 3723 : }
1858 : :
1859 : : } // namespace vlink
|
