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