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