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 : :
38 : : namespace vlink {
39 : :
40 : : // Timer::Impl
41 : : struct Timer::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
42 : : std::atomic_bool is_busy{false};
43 : : std::atomic<uint32_t> in_flight_count{0};
44 : : std::atomic<uint64_t> start_time{0};
45 : : std::atomic<uint64_t> generation{0};
46 : : std::atomic<int32_t> remain_loop_count{Timer::kInfinite};
47 : : std::atomic<uint64_t> invoke_count{0};
48 : :
49 : : std::atomic<int32_t> loop_count{Timer::kInfinite};
50 : : std::atomic<uint32_t> interval{1000U};
51 : : std::atomic<uint16_t> priority{MessageLoop::kTimerPriority};
52 : : std::atomic<MessageLoop*> message_loop{nullptr};
53 : : std::atomic_bool is_strict{false};
54 : :
55 : : bool is_once_type{false};
56 : :
57 : : Timer::Callback callback{nullptr};
58 : :
59 : : std::mutex mtx;
60 : : std::recursive_mutex recursive_mtx;
61 : : ConditionVariable cv;
62 : :
63 : : std::shared_ptr<std::atomic_bool> alive_flag{MemoryResource::make_shared<std::atomic_bool>(true)};
64 : : };
65 : :
66 : : // Timer
67 [ + - ]: 336 : Timer::Timer() : impl_(std::make_unique<Impl>()) { MemoryPool::global_instance(); }
68 : :
69 : 1 : Timer::Timer(MessageLoop* message_loop) : impl_(std::make_unique<Impl>()) {
70 [ + - ]: 1 : MemoryPool::global_instance();
71 [ + - ]: 1 : attach(message_loop);
72 : 1 : }
73 : :
74 : 26 : Timer::Timer(MessageLoop* message_loop, uint32_t interval_ms, int32_t loop_count, Callback&& callback)
75 : 26 : : impl_(std::make_unique<Impl>()) {
76 : 26 : impl_->interval.store(interval_ms, std::memory_order_relaxed);
77 : 26 : impl_->loop_count.store(loop_count, std::memory_order_relaxed);
78 : 26 : impl_->remain_loop_count.store(loop_count, std::memory_order_relaxed);
79 : 26 : impl_->callback = std::move(callback);
80 : :
81 [ + - ]: 26 : MemoryPool::global_instance();
82 : :
83 [ + - ]: 26 : attach(message_loop);
84 : 26 : }
85 : :
86 : 20 : Timer::Timer(uint32_t interval_ms, int32_t loop_count, Callback&& callback) : impl_(std::make_unique<Impl>()) {
87 : 20 : impl_->interval.store(interval_ms, std::memory_order_relaxed);
88 : 20 : impl_->loop_count.store(loop_count, std::memory_order_relaxed);
89 : 20 : impl_->remain_loop_count.store(loop_count, std::memory_order_relaxed);
90 : 20 : impl_->callback = std::move(callback);
91 : :
92 [ + - ]: 20 : MemoryPool::global_instance();
93 : 20 : }
94 : :
95 : 383 : Timer::~Timer() {
96 : 383 : impl_->alive_flag->store(false, std::memory_order_release);
97 : :
98 : 383 : MessageLoop* message_loop = impl_->message_loop.load(std::memory_order_acquire);
99 : :
100 [ + + + + : 383 : if (message_loop && !impl_->is_once_type) {
+ + ]
101 [ + + + - ]: 231 : const bool should_wait = message_loop->is_running() && !message_loop->is_in_same_thread();
102 : :
103 : 231 : detach();
104 : :
105 [ + + ]: 231 : if (should_wait) {
106 : 100 : wait_for_idle();
107 : : }
108 : : }
109 : 383 : }
110 : :
111 : 15 : bool Timer::call_once(MessageLoop* message_loop, uint32_t interval_ms, Callback&& callback, uint16_t priority) {
112 [ + + ]: 15 : if VUNLIKELY (!callback) {
113 [ + - + - ]: 2 : VLOG_E("Timer: Callback is null for call_once.");
114 : 1 : return false;
115 : : }
116 : :
117 : 14 : auto& pool = MemoryPool::global_instance();
118 : 14 : void* mem = pool.allocate(sizeof(Timer), alignof(Timer));
119 : :
120 [ - + ]: 14 : if VUNLIKELY (!mem) {
121 : : VLOG_E("Timer: MemoryPool allocate failed for call_once."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
122 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
123 : : }
124 : :
125 [ + - ]: 14 : auto* timer = new (mem) Timer(interval_ms, 1, std::move(callback));
126 : :
127 : 14 : timer->impl_->is_once_type = true;
128 : :
129 [ + + ]: 14 : if (priority > 0) {
130 : 1 : timer->set_priority(priority);
131 : : }
132 : :
133 [ + + ]: 14 : if (timer->attach(message_loop)) {
134 [ + - ]: 13 : timer->start();
135 : 13 : return true;
136 : : }
137 : :
138 : : // LCOV_EXCL_START GCOVR_EXCL_START
139 : : timer->~Timer();
140 : : pool.deallocate(mem, sizeof(Timer), alignof(Timer));
141 : :
142 : : return false;
143 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
144 : : }
145 : :
146 : 7362 : bool Timer::is_active() const { return impl_->start_time.load(std::memory_order_acquire) != 0; }
147 : :
148 : 958 : bool Timer::is_strict() const { return impl_->is_strict.load(std::memory_order_relaxed); }
149 : :
150 : 13940 : uint32_t Timer::get_interval() const { return impl_->interval.load(std::memory_order_relaxed); }
151 : :
152 : 14 : int32_t Timer::get_loop_count() const { return impl_->loop_count.load(std::memory_order_relaxed); }
153 : :
154 : 3848 : int32_t Timer::get_remain_loop_count() const { return impl_->remain_loop_count.load(std::memory_order_relaxed); }
155 : :
156 : 6966 : uint64_t Timer::get_invoke_count() const { return impl_->invoke_count.load(std::memory_order_relaxed); }
157 : :
158 : 4 : uint16_t Timer::get_priority() const { return impl_->priority.load(std::memory_order_relaxed); }
159 : :
160 : 11 : MessageLoop* Timer::get_message_loop() const { return impl_->message_loop.load(std::memory_order_acquire); }
161 : :
162 : 267 : bool Timer::attach(MessageLoop* message_loop) {
163 [ + + ]: 267 : if VUNLIKELY (!message_loop) {
164 [ + - - + ]: 2 : VLOG_F("Timer: MessageLoop is null.");
165 : : return false; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
166 : : }
167 : :
168 : 266 : MessageLoop* old_message_loop = impl_->message_loop.load(std::memory_order_acquire);
169 : :
170 [ + + ]: 266 : if (old_message_loop == message_loop) {
171 : 2 : return true;
172 : : }
173 : :
174 [ + + ]: 264 : if (old_message_loop) {
175 : 1 : stop();
176 : 1 : impl_->message_loop.store(nullptr, std::memory_order_release);
177 : 1 : old_message_loop->remove_timer(this);
178 : : }
179 : :
180 [ + + ]: 264 : if VUNLIKELY (!message_loop->add_timer(this)) {
181 : 3 : return false;
182 : : }
183 : :
184 : 261 : impl_->message_loop.store(message_loop, std::memory_order_release);
185 : 261 : return true;
186 : : }
187 : :
188 : 246 : bool Timer::detach() {
189 [ - + ]: 246 : if VUNLIKELY (impl_->is_once_type) {
190 : : return true; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
191 : : }
192 : :
193 : 246 : MessageLoop* message_loop = impl_->message_loop.load(std::memory_order_acquire);
194 : :
195 [ + + ]: 246 : if (message_loop) {
196 : 245 : stop();
197 : 245 : impl_->message_loop.store(nullptr, std::memory_order_release);
198 : :
199 : 245 : return message_loop->remove_timer(this);
200 : : }
201 : :
202 : 1 : return false;
203 : : }
204 : :
205 : 144 : void Timer::start(Callback&& callback) {
206 [ + + ]: 144 : if (callback) {
207 [ + - ]: 108 : std::lock_guard lock(impl_->recursive_mtx);
208 : 108 : impl_->callback = std::move(callback);
209 : 108 : }
210 : :
211 [ + - + - : 288 : if (!is_active() && impl_->remain_loop_count.load(std::memory_order_relaxed) != 0) {
+ - ]
212 : 144 : force_to_start();
213 : : }
214 : 144 : }
215 : :
216 : 64 : void Timer::restart() {
217 : 128 : impl_->remain_loop_count.store(impl_->loop_count.load(std::memory_order_relaxed), std::memory_order_relaxed);
218 : 64 : force_to_start();
219 : 64 : }
220 : :
221 : 513 : void Timer::stop() { stop(true); }
222 : :
223 : 523 : void Timer::stop(bool invalidate_pending) {
224 : 523 : impl_->start_time.store(0, std::memory_order_release);
225 : 523 : impl_->invoke_count.store(0, std::memory_order_relaxed);
226 : :
227 [ + + ]: 523 : if (invalidate_pending) {
228 : 513 : impl_->generation.fetch_add(1, std::memory_order_acq_rel);
229 : : }
230 : 523 : }
231 : :
232 : 2 : void Timer::set_strict(bool strict) { impl_->is_strict.store(strict, std::memory_order_relaxed); }
233 : :
234 : 665 : void Timer::set_interval(uint32_t interval_ms) {
235 : 665 : uint32_t old_interval = impl_->interval.exchange(interval_ms, std::memory_order_acq_rel);
236 : :
237 [ + + ]: 665 : if (old_interval == interval_ms) {
238 : 535 : return;
239 : : }
240 : :
241 [ + + ]: 130 : uint64_t interval_nano = interval_ms == 0 ? kMinInterval : static_cast<uint64_t>(interval_ms) * 1000'000U;
242 : :
243 : 130 : uint64_t start_snapshot = impl_->start_time.load(std::memory_order_acquire);
244 : :
245 [ + + ]: 130 : if (start_snapshot != 0) {
246 : 43 : uint64_t now_ns = MessageLoop::get_current_nano_time();
247 : :
248 [ + - ]: 43 : if VLIKELY (now_ns >= start_snapshot) {
249 : 43 : impl_->invoke_count.store((now_ns - start_snapshot) / interval_nano, std::memory_order_relaxed);
250 : : }
251 : :
252 : 43 : MessageLoop* message_loop = impl_->message_loop.load(std::memory_order_acquire);
253 : :
254 [ + - ]: 43 : if (message_loop) {
255 : 43 : message_loop->wakeup();
256 : : }
257 : : }
258 : : }
259 : :
260 : 91 : void Timer::set_loop_count(int32_t loop_count) {
261 : 91 : int32_t old_loop_count = impl_->loop_count.exchange(loop_count, std::memory_order_acq_rel);
262 : :
263 [ + + ]: 91 : if (old_loop_count == loop_count) {
264 : 87 : return;
265 : : }
266 : :
267 : 4 : impl_->remain_loop_count.store(loop_count, std::memory_order_relaxed);
268 : :
269 [ + + ]: 4 : if (is_active()) {
270 : 1 : MessageLoop* message_loop = impl_->message_loop.load(std::memory_order_acquire);
271 : :
272 [ + - ]: 1 : if (message_loop) {
273 : 1 : message_loop->wakeup();
274 : : }
275 : : }
276 : : }
277 : :
278 : 16 : void Timer::set_callback(Callback&& callback) {
279 [ + - ]: 16 : std::lock_guard lock(impl_->recursive_mtx);
280 : 16 : impl_->callback = std::move(callback);
281 : 16 : }
282 : :
283 : 952 : void Timer::run_callback() {
284 : : {
285 [ + - ]: 952 : std::lock_guard recursive_lock(impl_->recursive_mtx);
286 [ + - ]: 952 : std::lock_guard lock(impl_->mtx);
287 : :
288 : 952 : impl_->is_busy.store(true, std::memory_order_release);
289 : :
290 [ + - ]: 952 : if VLIKELY (impl_->callback) {
291 [ + - ]: 952 : impl_->callback();
292 : : }
293 : :
294 : 952 : impl_->is_busy.store(false, std::memory_order_release);
295 : 952 : }
296 : :
297 : 952 : impl_->cv.notify_all();
298 : 952 : }
299 : :
300 : 952 : void Timer::begin_in_flight() { impl_->in_flight_count.fetch_add(1, std::memory_order_acq_rel); }
301 : :
302 : 952 : void Timer::end_in_flight() {
303 [ + - ]: 1904 : if (impl_->in_flight_count.fetch_sub(1, std::memory_order_acq_rel) == 1U) {
304 [ + - ]: 952 : std::lock_guard lock(impl_->mtx);
305 : 952 : impl_->cv.notify_all();
306 : 952 : }
307 : 952 : }
308 : :
309 : 100 : void Timer::wait_for_idle() {
310 [ + - ]: 100 : std::unique_lock lock(impl_->mtx);
311 : 100 : impl_->cv.wait(lock, [this]() -> bool {
312 [ + - ]: 200 : return !impl_->is_busy.load(std::memory_order_acquire) &&
313 [ + - ]: 300 : impl_->in_flight_count.load(std::memory_order_acquire) == 0;
314 : : });
315 : 100 : }
316 : :
317 : 3 : void Timer::clear() { impl_->message_loop.store(nullptr, std::memory_order_release); }
318 : :
319 : 208 : void Timer::force_to_start() {
320 [ + + ]: 208 : if VUNLIKELY (!has_callback()) {
321 [ + - + - ]: 2 : VLOG_E("Timer: Callback is not set.");
322 : 1 : return;
323 : : }
324 : :
325 : 207 : impl_->start_time.store(MessageLoop::get_current_nano_time(), std::memory_order_release);
326 : 207 : impl_->invoke_count.store(0, std::memory_order_relaxed);
327 : 207 : impl_->generation.fetch_add(1, std::memory_order_acq_rel);
328 : :
329 : 207 : MessageLoop* message_loop = impl_->message_loop.load(std::memory_order_acquire);
330 : :
331 [ + + ]: 207 : if VLIKELY (message_loop) {
332 : 206 : message_loop->wakeup();
333 : : } else {
334 [ + - + - ]: 2 : VLOG_E("Timer: MessageLoop is not attached.");
335 : : }
336 : : }
337 : :
338 : 0 : void Timer::set_remain_loop_count(int32_t loop_count) const {
339 : 0 : impl_->remain_loop_count.store(loop_count, std::memory_order_relaxed);
340 : 0 : }
341 : :
342 : 955 : void Timer::sub_remain_loop_count() const {
343 [ + + ]: 1910 : if (impl_->remain_loop_count.load(std::memory_order_relaxed) <= 0) {
344 : 937 : return;
345 : : }
346 : :
347 : 18 : impl_->remain_loop_count.fetch_sub(1, std::memory_order_relaxed);
348 : : }
349 : :
350 : 945 : void Timer::set_invoke_count(uint64_t invoke_count) const {
351 : 945 : impl_->invoke_count.store(invoke_count, std::memory_order_relaxed);
352 : 945 : }
353 : :
354 : 2 : void Timer::set_priority(uint16_t priority) { impl_->priority.store(priority, std::memory_order_relaxed); }
355 : :
356 : 6964 : uint64_t Timer::get_start_time() const { return impl_->start_time.load(std::memory_order_acquire); }
357 : :
358 : 8874 : uint64_t Timer::get_generation() const { return impl_->generation.load(std::memory_order_acquire); }
359 : :
360 : 1937 : bool Timer::is_once_type() const { return impl_->is_once_type; }
361 : :
362 : 3690 : bool Timer::has_callback() const {
363 [ + - ]: 3690 : std::lock_guard lock(impl_->recursive_mtx);
364 : 7380 : return impl_->callback != nullptr;
365 : 3690 : }
366 : :
367 : 12 : Timer::Callback Timer::take_callback() {
368 [ + - ]: 12 : std::lock_guard lock(impl_->recursive_mtx);
369 [ + - ]: 24 : return std::exchange(impl_->callback, nullptr);
370 : 12 : }
371 : :
372 : 967 : std::shared_ptr<std::atomic_bool> Timer::get_alive_flag() const { return impl_->alive_flag; }
373 : :
374 : : } // namespace vlink
|
