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