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/timer.h"
25 : :
26 : : #include <atomic>
27 : : #include <memory>
28 : : #include <mutex>
29 : : #include <new>
30 : : #include <utility>
31 : :
32 : : #include "./base/condition_variable.h"
33 : : #include "./base/logger.h"
34 : : #include "./base/memory_pool.h"
35 : : #include "./base/memory_resource.h"
36 : : #include "./base/message_loop.h"
37 : : #include "./base/utils.h"
38 : :
39 : : namespace vlink {
40 : :
41 : : // Timer::Impl
42 : : struct Timer::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
43 : : std::atomic_bool is_busy{false};
44 : : std::atomic<uint32_t> in_flight_count{0};
45 : : std::atomic<uint64_t> start_time{0};
46 : : std::atomic<uint64_t> generation{0};
47 : : std::atomic<int32_t> remain_loop_count{Timer::kInfinite};
48 : : std::atomic<uint64_t> invoke_count{0};
49 : :
50 : : std::atomic<int32_t> loop_count{Timer::kInfinite};
51 : : std::atomic<uint32_t> interval{1000U};
52 : : std::atomic<uint16_t> priority{MessageLoop::kTimerPriority};
53 : : std::atomic<MessageLoop*> message_loop{nullptr};
54 : : std::atomic_bool is_strict{false};
55 : :
56 : : bool is_once_type{false};
57 : : bool has_pending_callback{false};
58 : :
59 : : Timer::Callback callback{nullptr};
60 : : Timer::Callback pending_callback{nullptr};
61 : :
62 : : std::mutex mtx;
63 : : std::recursive_mutex recursive_mtx;
64 : : ConditionVariable cv;
65 : :
66 : : std::shared_ptr<std::atomic_bool> alive_flag{MemoryResource::make_shared<std::atomic_bool>(true)};
67 : : };
68 : :
69 : : // Timer
70 [ + - ]: 556 : Timer::Timer() : impl_(std::make_unique<Impl>()) { MemoryPool::global_instance(); }
71 : :
72 : 2 : Timer::Timer(MessageLoop* message_loop) : impl_(std::make_unique<Impl>()) {
73 [ + - ]: 2 : MemoryPool::global_instance();
74 [ + - ]: 2 : attach(message_loop);
75 : 2 : }
76 : :
77 : 31 : Timer::Timer(MessageLoop* message_loop, uint32_t interval_ms, int32_t loop_count, Callback&& callback)
78 : 31 : : impl_(std::make_unique<Impl>()) {
79 : 31 : impl_->interval.store(interval_ms, std::memory_order_relaxed);
80 : 31 : impl_->loop_count.store(loop_count, std::memory_order_relaxed);
81 : 31 : impl_->remain_loop_count.store(loop_count, std::memory_order_relaxed);
82 : 31 : impl_->callback = std::move(callback);
83 : :
84 [ + - ]: 31 : MemoryPool::global_instance();
85 : :
86 [ + - ]: 31 : attach(message_loop);
87 : 31 : }
88 : :
89 : 20 : Timer::Timer(uint32_t interval_ms, int32_t loop_count, Callback&& callback) : impl_(std::make_unique<Impl>()) {
90 : 20 : impl_->interval.store(interval_ms, std::memory_order_relaxed);
91 : 20 : impl_->loop_count.store(loop_count, std::memory_order_relaxed);
92 : 20 : impl_->remain_loop_count.store(loop_count, std::memory_order_relaxed);
93 : 20 : impl_->callback = std::move(callback);
94 : :
95 [ + - ]: 20 : MemoryPool::global_instance();
96 : 20 : }
97 : :
98 : 609 : Timer::~Timer() {
99 : 609 : impl_->alive_flag->store(false, std::memory_order_release);
100 : :
101 : : #ifdef _WIN32
102 : : if (Utils::is_terminating()) {
103 : : (void)impl_.release();
104 : : return;
105 : : }
106 : : #endif
107 : :
108 [ + + ]: 609 : if (!impl_->is_once_type) {
109 : 595 : detach();
110 : 595 : wait_for_idle();
111 : : }
112 : 609 : }
113 : :
114 : 15 : bool Timer::call_once(MessageLoop* message_loop, uint32_t interval_ms, Callback&& callback, uint16_t priority) {
115 [ + + ]: 15 : if VUNLIKELY (!callback) {
116 [ + - + - : 2 : VLOG_E("Timer: Callback is null for call_once.");
+ - ]
117 : 1 : return false;
118 : : }
119 : :
120 : 14 : auto& pool = MemoryPool::global_instance();
121 : 14 : void* mem = pool.allocate(sizeof(Timer), alignof(Timer));
122 : :
123 [ - + ]: 14 : if VUNLIKELY (!mem) {
124 : : VLOG_E("Timer: MemoryPool allocate failed for call_once."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
125 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
126 : : }
127 : :
128 [ + - ]: 14 : auto* timer = new (mem) Timer(interval_ms, 1, std::move(callback));
129 : :
130 : 14 : timer->impl_->is_once_type = true;
131 : :
132 [ + + ]: 14 : if (priority > 0) {
133 : 1 : timer->set_priority(priority);
134 : : }
135 : :
136 [ + + ]: 14 : if (timer->attach(message_loop)) {
137 [ + - ]: 13 : timer->start();
138 : 13 : return true;
139 : : }
140 : :
141 : : // LCOV_EXCL_START GCOVR_EXCL_START
142 : : timer->~Timer();
143 : : pool.deallocate(mem, sizeof(Timer), alignof(Timer));
144 : :
145 : : return false;
146 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
147 : : }
148 : :
149 : 20191 : bool Timer::is_active() const { return impl_->start_time.load(std::memory_order_acquire) != 0; }
150 : :
151 : 1340 : bool Timer::is_strict() const { return impl_->is_strict.load(std::memory_order_relaxed); }
152 : :
153 : 18830 : uint32_t Timer::get_interval() const { return impl_->interval.load(std::memory_order_relaxed); }
154 : :
155 : 14 : int32_t Timer::get_loop_count() const { return impl_->loop_count.load(std::memory_order_relaxed); }
156 : :
157 : 5620 : int32_t Timer::get_remain_loop_count() const { return impl_->remain_loop_count.load(std::memory_order_relaxed); }
158 : :
159 : 18817 : uint64_t Timer::get_invoke_count() const { return impl_->invoke_count.load(std::memory_order_relaxed); }
160 : :
161 : 4 : uint16_t Timer::get_priority() const { return impl_->priority.load(std::memory_order_relaxed); }
162 : :
163 : 11 : MessageLoop* Timer::get_message_loop() const { return impl_->message_loop.load(std::memory_order_acquire); }
164 : :
165 : 420 : bool Timer::attach(MessageLoop* message_loop) {
166 [ + + ]: 420 : if VUNLIKELY (!message_loop) {
167 [ + - + - : 2 : VLOG_F("Timer: MessageLoop is null.");
- + ]
168 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
169 : : }
170 : :
171 : 419 : MessageLoop* old_message_loop = impl_->message_loop.load(std::memory_order_acquire);
172 : :
173 [ + + ]: 419 : if (old_message_loop == message_loop) {
174 : 2 : return true;
175 : : }
176 : :
177 [ + + ]: 417 : if (old_message_loop) {
178 : 1 : stop();
179 : 1 : impl_->message_loop.store(nullptr, std::memory_order_release);
180 : 1 : old_message_loop->remove_timer(this);
181 : : }
182 : :
183 [ + + ]: 417 : if VUNLIKELY (!message_loop->add_timer(this)) {
184 : 3 : return false;
185 : : }
186 : :
187 : 414 : impl_->message_loop.store(message_loop, std::memory_order_release);
188 : 414 : return true;
189 : : }
190 : :
191 : 611 : bool Timer::detach() {
192 [ - + ]: 611 : if VUNLIKELY (impl_->is_once_type) {
193 : : return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
194 : : }
195 : :
196 : 611 : MessageLoop* message_loop = impl_->message_loop.load(std::memory_order_acquire);
197 : :
198 [ + + ]: 611 : if (message_loop) {
199 : 400 : stop();
200 : 400 : impl_->message_loop.store(nullptr, std::memory_order_release);
201 : :
202 : 400 : return message_loop->remove_timer(this);
203 : : }
204 : :
205 : 211 : return false;
206 : : }
207 : :
208 : 296 : void Timer::start(Callback&& callback) {
209 [ + + ]: 296 : if (callback) {
210 : 143 : set_callback(std::move(callback));
211 : : }
212 : :
213 [ + + + - : 591 : if (!is_active() && impl_->remain_loop_count.load(std::memory_order_relaxed) != 0) {
+ + ]
214 : 295 : force_to_start();
215 : : }
216 : 296 : }
217 : :
218 : 262 : void Timer::restart() { force_to_start(true); }
219 : :
220 : 891 : void Timer::stop() { stop(true); }
221 : :
222 : 2 : void Timer::set_strict(bool strict) { impl_->is_strict.store(strict, std::memory_order_relaxed); }
223 : :
224 : 782 : void Timer::set_interval(uint32_t interval_ms) {
225 : 782 : uint32_t old_interval = impl_->interval.exchange(interval_ms, std::memory_order_acq_rel);
226 : :
227 [ + + ]: 782 : if (old_interval == interval_ms) {
228 : 540 : return;
229 : : }
230 : :
231 [ + + ]: 242 : uint64_t interval_nano = interval_ms == 0 ? kMinInterval : static_cast<uint64_t>(interval_ms) * 1000'000U;
232 : :
233 : 242 : uint64_t start_snapshot = impl_->start_time.load(std::memory_order_acquire);
234 : :
235 [ + + ]: 242 : if (start_snapshot != 0) {
236 : 42 : uint64_t now_ns = MessageLoop::get_current_nano_time();
237 : :
238 [ + - ]: 42 : if VLIKELY (now_ns >= start_snapshot) {
239 : 42 : impl_->invoke_count.store((now_ns - start_snapshot) / interval_nano, std::memory_order_relaxed);
240 : : }
241 : :
242 : 42 : MessageLoop* message_loop = impl_->message_loop.load(std::memory_order_acquire);
243 : :
244 [ + - ]: 42 : if (message_loop) {
245 : 42 : message_loop->wakeup();
246 : : }
247 : : }
248 : : }
249 : :
250 : 208 : void Timer::set_loop_count(int32_t loop_count) {
251 : 208 : int32_t old_loop_count = impl_->loop_count.exchange(loop_count, std::memory_order_acq_rel);
252 : :
253 [ + + ]: 208 : if (old_loop_count == loop_count) {
254 : 204 : return;
255 : : }
256 : :
257 : 4 : impl_->remain_loop_count.store(loop_count, std::memory_order_relaxed);
258 : :
259 [ + + ]: 4 : if (is_active()) {
260 : 1 : MessageLoop* message_loop = impl_->message_loop.load(std::memory_order_acquire);
261 : :
262 [ + - ]: 1 : if (message_loop) {
263 : 1 : message_loop->wakeup();
264 : : }
265 : : }
266 : : }
267 : :
268 : 272 : void Timer::set_callback(Callback&& callback) {
269 : 272 : Callback previous;
270 [ + - ]: 272 : std::lock_guard lock(impl_->recursive_mtx);
271 : :
272 [ + + ]: 272 : if VUNLIKELY (impl_->is_busy.load(std::memory_order_acquire)) {
273 : 1 : previous = std::move(impl_->pending_callback);
274 : 1 : impl_->pending_callback = std::move(callback);
275 : 1 : impl_->has_pending_callback = true;
276 : 1 : return;
277 : : }
278 : :
279 : 271 : previous = std::move(impl_->callback);
280 : 271 : impl_->callback = std::move(callback);
281 [ + + + + ]: 273 : }
282 : :
283 : 2 : void Timer::set_priority(uint16_t priority) { impl_->priority.store(priority, std::memory_order_relaxed); }
284 : :
285 : 1333 : void Timer::run_callback() {
286 : 1333 : Callback previous;
287 : :
288 : : {
289 [ + - ]: 1333 : std::lock_guard lock(impl_->mtx);
290 : :
291 : : {
292 [ + - ]: 1333 : std::lock_guard callback_lock(impl_->recursive_mtx);
293 : 1333 : impl_->is_busy.store(true, std::memory_order_release);
294 : 1333 : }
295 : :
296 [ + - ]: 1333 : if VLIKELY (impl_->callback) {
297 [ + - ]: 1333 : impl_->callback();
298 : : }
299 : :
300 : : {
301 [ + - ]: 1333 : std::lock_guard callback_lock(impl_->recursive_mtx);
302 : :
303 [ + + ]: 1333 : if VUNLIKELY (impl_->has_pending_callback) {
304 : 1 : previous = std::move(impl_->callback);
305 : 1 : impl_->callback = std::move(impl_->pending_callback);
306 : 1 : impl_->has_pending_callback = false;
307 : : }
308 : :
309 : 1333 : impl_->is_busy.store(false, std::memory_order_release);
310 : 1333 : }
311 : 1333 : }
312 : :
313 : 1333 : impl_->cv.notify_all();
314 : 1333 : }
315 : :
316 : 1333 : void Timer::begin_in_flight() { impl_->in_flight_count.fetch_add(1, std::memory_order_acq_rel); }
317 : :
318 : 1333 : void Timer::end_in_flight() {
319 [ + - ]: 1333 : std::lock_guard lock(impl_->mtx);
320 : :
321 [ + - ]: 2666 : if (impl_->in_flight_count.fetch_sub(1, std::memory_order_acq_rel) == 1U) {
322 : 1333 : impl_->cv.notify_all();
323 : : }
324 : 1333 : }
325 : :
326 : 595 : void Timer::wait_for_idle() {
327 [ + - ]: 595 : std::unique_lock lock(impl_->mtx);
328 : 595 : impl_->cv.wait(lock, [this]() -> bool {
329 [ + - ]: 1190 : return !impl_->is_busy.load(std::memory_order_acquire) &&
330 [ + - ]: 1785 : impl_->in_flight_count.load(std::memory_order_acquire) == 0;
331 : : });
332 : 595 : }
333 : :
334 : 904 : void Timer::stop(bool invalidate_pending) {
335 : 904 : impl_->start_time.store(0, std::memory_order_release);
336 : :
337 [ + + ]: 904 : if (invalidate_pending) {
338 : 891 : impl_->invoke_count.store(0, std::memory_order_relaxed);
339 : 891 : impl_->generation.fetch_add(1, std::memory_order_acq_rel);
340 : : }
341 : 904 : }
342 : :
343 : 1 : void Timer::clear() { impl_->message_loop.store(nullptr, std::memory_order_release); }
344 : :
345 : 557 : void Timer::force_to_start(bool reset_loop_count) {
346 [ + + ]: 557 : if VUNLIKELY (!has_callback()) {
347 [ - + ]: 1 : if (reset_loop_count) {
348 : 0 : impl_->remain_loop_count.store(impl_->loop_count.load(std::memory_order_relaxed), std::memory_order_relaxed);
349 : : }
350 : :
351 [ + - + - : 2 : VLOG_E("Timer: Callback is not set.");
+ - ]
352 : 1 : return;
353 : : }
354 : :
355 : 556 : MessageLoop* message_loop = impl_->message_loop.load(std::memory_order_acquire);
356 : :
357 : : {
358 : 556 : std::unique_lock<std::mutex> lock;
359 : :
360 [ + + ]: 556 : if VLIKELY (message_loop) {
361 [ + - ]: 555 : lock = message_loop->lock_timers();
362 : : }
363 : :
364 [ + + ]: 556 : if (reset_loop_count) {
365 : 524 : impl_->remain_loop_count.store(impl_->loop_count.load(std::memory_order_relaxed), std::memory_order_relaxed);
366 : : }
367 : :
368 [ + - ]: 556 : impl_->start_time.store(MessageLoop::get_current_nano_time(), std::memory_order_release);
369 : 556 : impl_->invoke_count.store(0, std::memory_order_relaxed);
370 : 556 : impl_->generation.fetch_add(1, std::memory_order_acq_rel);
371 : 556 : }
372 : :
373 [ + + ]: 556 : if VLIKELY (message_loop) {
374 : 555 : message_loop->wakeup();
375 : : } else {
376 [ + - + - : 2 : VLOG_E("Timer: MessageLoop is not attached.");
+ - ]
377 : : }
378 : : }
379 : :
380 : 0 : void Timer::set_remain_loop_count(int32_t loop_count) const {
381 : 0 : impl_->remain_loop_count.store(loop_count, std::memory_order_relaxed);
382 : 0 : }
383 : :
384 : 1337 : void Timer::sub_remain_loop_count() const {
385 [ + + ]: 2674 : if (impl_->remain_loop_count.load(std::memory_order_relaxed) <= 0) {
386 : 1316 : return;
387 : : }
388 : :
389 : 21 : impl_->remain_loop_count.fetch_sub(1, std::memory_order_relaxed);
390 : : }
391 : :
392 : 1398 : void Timer::set_invoke_count(uint64_t invoke_count) const {
393 : 1398 : impl_->invoke_count.store(invoke_count, std::memory_order_relaxed);
394 : 1398 : }
395 : :
396 : 18813 : uint64_t Timer::get_start_time() const { return impl_->start_time.load(std::memory_order_acquire); }
397 : :
398 : 21486 : uint64_t Timer::get_generation() const { return impl_->generation.load(std::memory_order_acquire); }
399 : :
400 : 2699 : bool Timer::is_once_type() const { return impl_->is_once_type; }
401 : :
402 : 9963 : bool Timer::has_callback() const {
403 [ + - ]: 9963 : std::lock_guard lock(impl_->recursive_mtx);
404 : 19926 : return impl_->callback != nullptr;
405 : 9963 : }
406 : :
407 : 12 : Timer::Callback Timer::take_callback() {
408 [ + - ]: 12 : std::lock_guard lock(impl_->recursive_mtx);
409 [ + - ]: 24 : return std::exchange(impl_->callback, nullptr);
410 : 12 : }
411 : :
412 : 1410 : std::shared_ptr<std::atomic_bool> Timer::get_alive_flag() const { return impl_->alive_flag; }
413 : :
414 : : } // namespace vlink
|
