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/wheel_timer.h"
25 : :
26 : : #include <atomic>
27 : : #include <limits>
28 : : #include <list>
29 : : #include <mutex>
30 : : #include <optional>
31 : : #include <thread>
32 : : #include <unordered_map>
33 : : #include <utility>
34 : : #include <vector>
35 : :
36 : : #include "./base/condition_variable.h"
37 : : #include "./base/logger.h"
38 : : #include "./base/memory_pool.h"
39 : : #include "./base/memory_resource.h"
40 : :
41 : : namespace vlink {
42 : :
43 : : // WheelTimer::Impl
44 : : struct WheelTimer::Impl final { // NOLINT(clang-analyzer-optin.performance.Padding)
45 : : // Handler
46 : : struct Handler final {
47 : : WheelTimer::Key key{-1};
48 : : uint32_t remaining_rounds{0};
49 : : WheelTimer::Callback callback;
50 : : uint32_t repeat_interval_ms{0};
51 : :
52 : 118 : Handler(WheelTimer::Key _key, uint32_t _rounds, WheelTimer::Callback&& _callback, uint32_t _repeat_ms = 0)
53 : 118 : : key(_key), remaining_rounds(_rounds), callback(std::move(_callback)), repeat_interval_ms(_repeat_ms) {}
54 : :
55 : : Handler(const Handler&) = default;
56 : :
57 : 28 : Handler(Handler&&) = default;
58 : :
59 : : Handler& operator=(const Handler&) = default;
60 : :
61 : : Handler& operator=(Handler&&) = default;
62 : : };
63 : :
64 : : alignas(64) std::atomic_bool stop_flag{false};
65 : : alignas(64) std::atomic_bool paused_flag{false};
66 : : alignas(64) std::atomic_bool is_running{false};
67 : :
68 : : std::atomic<uint32_t> catchup_limit{0};
69 : : std::atomic<WheelTimer::Key> next_key{1};
70 : :
71 : : uint32_t slots{0};
72 : : uint32_t interval_ms{5};
73 : :
74 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
75 : : std::optional<std::pmr::vector<std::pmr::list<Handler>>> wheels;
76 : : #else
77 : : std::optional<std::vector<std::list<Handler>>> wheels;
78 : : #endif
79 : :
80 : : uint32_t current_slot{0};
81 : :
82 : : std::thread worker_thread;
83 : :
84 : : std::mutex mtx;
85 : : std::mutex lifecycle_mtx;
86 : : ConditionVariable cv;
87 : :
88 : : std::unordered_map<WheelTimer::Key, std::pair<uint32_t, std::list<Handler>::iterator>> timer_index;
89 : :
90 : : void run();
91 : : };
92 : :
93 : : // WheelTimer
94 : 34 : WheelTimer::WheelTimer(uint32_t slots, uint32_t interval_ms) : impl_(MemoryResource::make_shared<Impl>()) {
95 [ + + + + : 34 : if VUNLIKELY (slots == 0 || interval_ms == 0) {
+ + ]
96 [ + - - + ]: 4 : VLOG_F("WheelTimer: Slots and interval_ms must be greater than 0.");
97 : : }
98 : :
99 [ + - ]: 32 : MemoryPool::global_instance();
100 : :
101 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
102 [ + - ]: 32 : impl_->wheels.emplace(&MemoryResource::global_instance());
103 : : #else
104 : : impl_->wheels.emplace();
105 : : #endif
106 : :
107 [ + - ]: 32 : impl_->slots = (slots == 0) ? 1U : slots;
108 [ + - ]: 32 : impl_->interval_ms = (interval_ms == 0) ? 1U : interval_ms;
109 [ + - ]: 32 : impl_->wheels->resize(impl_->slots);
110 : 34 : }
111 : :
112 : 32 : WheelTimer::~WheelTimer() {
113 : 32 : stop();
114 : 32 : impl_->wheels.reset();
115 : 32 : }
116 : :
117 : 32 : void WheelTimer::start() {
118 [ + - ]: 32 : std::lock_guard lifecycle_lock(impl_->lifecycle_mtx);
119 : :
120 : : {
121 [ + - ]: 32 : std::lock_guard lock(impl_->mtx);
122 : :
123 [ + + ]: 32 : if VUNLIKELY (impl_->is_running.load(std::memory_order_acquire)) {
124 [ + - + - ]: 2 : VLOG_W("WheelTimer: Timer is already running.");
125 : 1 : return;
126 : : }
127 [ + + ]: 32 : }
128 : :
129 [ - + ]: 31 : if (impl_->worker_thread.joinable()) {
130 : : impl_->worker_thread.join(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
131 : : }
132 : :
133 : : {
134 [ + - ]: 31 : std::lock_guard lock(impl_->mtx);
135 : 31 : impl_->stop_flag.store(false, std::memory_order_release);
136 : 31 : impl_->is_running.store(true, std::memory_order_release);
137 : 31 : }
138 : :
139 : : try {
140 : 31 : auto impl_copy = impl_;
141 [ + - ]: 62 : impl_->worker_thread = std::thread([impl_copy]() { impl_copy->run(); });
142 [ - - ]: 31 : } catch (std::exception&) {
143 : : // LCOV_EXCL_START GCOVR_EXCL_START
144 : : {
145 : : std::lock_guard lock(impl_->mtx);
146 : : impl_->is_running.store(false, std::memory_order_release);
147 : : }
148 : :
149 : : throw;
150 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
151 : 0 : }
152 [ + + ]: 32 : }
153 : :
154 : 64 : void WheelTimer::stop() {
155 : : {
156 [ + - ]: 64 : std::lock_guard lock(impl_->mtx);
157 : 64 : impl_->stop_flag.store(true, std::memory_order_release);
158 : 64 : }
159 : :
160 [ + - ]: 64 : wakeup();
161 : :
162 : : const bool called_from_worker =
163 [ + + + + ]: 64 : impl_->worker_thread.joinable() && impl_->worker_thread.get_id() == std::this_thread::get_id();
164 : :
165 [ + + ]: 64 : if (called_from_worker) {
166 [ + - ]: 1 : if (impl_->lifecycle_mtx.try_lock()) {
167 [ + - ]: 1 : if (impl_->worker_thread.joinable()) {
168 [ + - ]: 1 : impl_->worker_thread.detach();
169 : : }
170 : :
171 : 1 : impl_->lifecycle_mtx.unlock();
172 : : }
173 : :
174 : 1 : return;
175 : : }
176 : :
177 [ + - ]: 63 : std::lock_guard lifecycle_lock(impl_->lifecycle_mtx);
178 : :
179 [ + + ]: 63 : if (impl_->worker_thread.joinable()) {
180 [ + - ]: 30 : impl_->worker_thread.join();
181 : : } else {
182 [ + - ]: 33 : std::unique_lock lock(impl_->mtx);
183 : 66 : impl_->cv.wait(lock, [this]() { return !impl_->is_running.load(std::memory_order_acquire); });
184 : 33 : }
185 : 63 : }
186 : :
187 : 4 : void WheelTimer::pause() {
188 [ + - ]: 4 : std::lock_guard lock(impl_->mtx);
189 : 4 : impl_->paused_flag.store(true, std::memory_order_release);
190 : 4 : }
191 : :
192 : 3 : void WheelTimer::resume() {
193 [ + - ]: 3 : std::unique_lock lock(impl_->mtx);
194 : 3 : impl_->paused_flag.store(false, std::memory_order_release);
195 : :
196 [ + - ]: 3 : lock.unlock();
197 : :
198 [ + - ]: 3 : wakeup();
199 : 3 : }
200 : :
201 : 172 : void WheelTimer::wakeup() { impl_->cv.notify_one(); }
202 : :
203 : 6 : bool WheelTimer::is_running() const { return impl_->is_running.load(std::memory_order_acquire); }
204 : :
205 : 92 : WheelTimer::Key WheelTimer::add(uint32_t timeout_ms, Callback&& callback, uint32_t repeat_ms) {
206 [ + + ]: 92 : if VUNLIKELY (timeout_ms == 0) {
207 [ + - + - ]: 2 : VLOG_E("WheelTimer: Timeout must be greater than 0.");
208 : 1 : return -1;
209 : : }
210 : :
211 [ + + ]: 91 : if VUNLIKELY (!callback) {
212 [ + - + - ]: 2 : VLOG_E("WheelTimer: Callback must be non-empty.");
213 : 1 : return -1;
214 : : }
215 : :
216 [ + - ]: 90 : std::lock_guard lock(impl_->mtx);
217 : :
218 : 90 : uint32_t interval = impl_->interval_ms;
219 : 90 : uint32_t slots = impl_->slots;
220 : 90 : uint32_t current_slot = impl_->current_slot;
221 : :
222 : 90 : uint64_t ticks = (static_cast<uint64_t>(timeout_ms) + interval - 1) / interval;
223 : :
224 [ - + ]: 90 : if VUNLIKELY (ticks == 0) {
225 : : ticks = 1; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
226 : : }
227 : :
228 : 90 : uint64_t max_rounds = std::numeric_limits<uint32_t>::max();
229 : 90 : uint64_t rounds64 = ticks / slots;
230 : :
231 [ - + ]: 90 : if VUNLIKELY (rounds64 > max_rounds) {
232 : : VLOG_E("WheelTimer: Timeout too large (rounds overflow)."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
233 : : return -1; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
234 : : }
235 : :
236 : 90 : auto ticks_mod = static_cast<uint32_t>(ticks % slots);
237 : 90 : auto rounds = static_cast<uint32_t>(rounds64);
238 : 90 : uint32_t slot = (current_slot + ticks_mod) % slots;
239 : :
240 : 90 : WheelTimer::Key key = impl_->next_key.fetch_add(1, std::memory_order_relaxed);
241 : :
242 [ - + ]: 90 : if VUNLIKELY (key <= 0) {
243 : : impl_->next_key.store(1, std::memory_order_relaxed); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
244 : : key = impl_->next_key.fetch_add(1, std::memory_order_relaxed); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
245 : : }
246 : :
247 : 90 : int probe = 0;
248 : :
249 [ + - - + : 90 : while (impl_->timer_index.find(key) != impl_->timer_index.end() && probe < 8) {
- - - + ]
250 : : // LCOV_EXCL_START GCOVR_EXCL_START
251 : : key = impl_->next_key.fetch_add(1, std::memory_order_relaxed);
252 : :
253 : : if (key <= 0) {
254 : : impl_->next_key.store(1, std::memory_order_relaxed);
255 : : key = impl_->next_key.fetch_add(1, std::memory_order_relaxed);
256 : : }
257 : :
258 : : ++probe;
259 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
260 : : }
261 : :
262 [ + - - + ]: 90 : if VUNLIKELY (impl_->timer_index.find(key) != impl_->timer_index.end()) {
263 : : VLOG_E("WheelTimer: Failed to allocate a unique key."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
264 : : return -1; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
265 : : }
266 : :
267 : 90 : auto& slot_list = (*impl_->wheels)[slot];
268 [ + - ]: 90 : slot_list.emplace_back(key, rounds, std::move(callback), repeat_ms);
269 : :
270 [ + - ]: 90 : auto it = std::prev(slot_list.end());
271 [ + - ]: 90 : impl_->timer_index[key] = {slot, it};
272 : :
273 [ + - ]: 90 : wakeup();
274 : :
275 : 90 : return key;
276 : 90 : }
277 : :
278 : 16 : bool WheelTimer::remove(WheelTimer::Key key) {
279 : : {
280 [ + - ]: 16 : std::lock_guard lock(impl_->mtx);
281 : :
282 [ + - ]: 16 : auto it = impl_->timer_index.find(key);
283 : :
284 [ + + ]: 16 : if VUNLIKELY (it == impl_->timer_index.end()) {
285 : 2 : return false;
286 : : }
287 : :
288 : 14 : auto& slot_list = (*impl_->wheels)[it->second.first];
289 : 14 : slot_list.erase(it->second.second);
290 [ + - ]: 14 : impl_->timer_index.erase(it);
291 [ + + ]: 16 : }
292 : :
293 : 14 : wakeup();
294 : :
295 : 14 : return true;
296 : : }
297 : :
298 : 5 : uint32_t WheelTimer::get_remaining_time(Key key) const {
299 [ + - ]: 5 : std::lock_guard lock(impl_->mtx);
300 : :
301 [ + - ]: 5 : auto it = impl_->timer_index.find(key);
302 : :
303 [ + + ]: 5 : if (it == impl_->timer_index.end()) {
304 : 2 : return 0;
305 : : }
306 : :
307 : 3 : uint32_t slot = it->second.first;
308 : 3 : uint32_t current_slot = impl_->current_slot;
309 : 3 : uint32_t delta_slot = (slot + impl_->slots - current_slot) % impl_->slots;
310 : 3 : uint32_t rounds = it->second.second->remaining_rounds;
311 : :
312 : 3 : uint64_t total_ticks = static_cast<uint64_t>(rounds) * impl_->slots + delta_slot;
313 : 3 : uint64_t total_ms = total_ticks * impl_->interval_ms;
314 : :
315 [ - + ]: 3 : return (total_ms > std::numeric_limits<uint32_t>::max()) ? std::numeric_limits<uint32_t>::max()
316 : 3 : : static_cast<uint32_t>(total_ms);
317 : 5 : }
318 : :
319 : 3 : void WheelTimer::set_catchup_limit(uint32_t max_slots_to_catch_up) {
320 : 3 : impl_->catchup_limit.store(max_slots_to_catch_up, std::memory_order_relaxed);
321 : 3 : }
322 : :
323 : 31 : void WheelTimer::Impl::run() {
324 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
325 : : std::pmr::vector<std::pair<WheelTimer::Key, WheelTimer::Callback>> callbacks_to_execute(
326 [ + - ]: 31 : &MemoryResource::global_instance());
327 : : #else
328 : : std::vector<std::pair<WheelTimer::Key, WheelTimer::Callback>> callbacks_to_execute;
329 : : #endif
330 : :
331 : 31 : auto interval = std::chrono::milliseconds(interval_ms);
332 : :
333 : 31 : auto next_tick = std::chrono::steady_clock::now();
334 : :
335 : : for (;;) {
336 [ + - ]: 253 : std::unique_lock lock(mtx);
337 : :
338 [ + + ]: 253 : if VUNLIKELY (stop_flag.load(std::memory_order_acquire)) {
339 : 8 : break;
340 : : }
341 : :
342 [ + + + + : 249 : while (paused_flag.load(std::memory_order_acquire) && !stop_flag.load(std::memory_order_acquire)) {
+ + ]
343 : 4 : cv.wait(lock);
344 : : }
345 : :
346 [ + + ]: 245 : if VUNLIKELY (stop_flag.load(std::memory_order_acquire)) {
347 : 1 : break;
348 : : }
349 : :
350 : 244 : auto now = std::chrono::steady_clock::now();
351 : :
352 [ + - + + ]: 244 : if (now < next_tick) {
353 : 215 : cv.wait_until(lock, next_tick, [this]() -> bool {
354 [ + + + + ]: 437 : return stop_flag.load(std::memory_order_acquire) || paused_flag.load(std::memory_order_acquire);
355 : : });
356 : :
357 [ + + ]: 215 : if VUNLIKELY (stop_flag.load(std::memory_order_acquire)) {
358 : 22 : break;
359 : : }
360 : :
361 : 193 : now = std::chrono::steady_clock::now();
362 : : }
363 : :
364 : 222 : constexpr int64_t kStaleTickResetIntervals = 10;
365 : :
366 [ + - + - : 222 : if VUNLIKELY (next_tick + interval * kStaleTickResetIntervals < now) {
+ - - + ]
367 : : next_tick = now + interval; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
368 : : }
369 : :
370 : 222 : uint32_t advanced = 0;
371 : 222 : uint32_t catchup_limit_snapshot = catchup_limit.load(std::memory_order_relaxed);
372 : :
373 [ + - + + : 693 : while (now >= next_tick && !stop_flag.load(std::memory_order_acquire) &&
+ - + + ]
374 [ + + ]: 238 : !paused_flag.load(std::memory_order_acquire)) {
375 : 234 : auto& timers = (*wheels)[current_slot];
376 : :
377 [ + + ]: 335 : for (auto it = timers.begin(); it != timers.end();) {
378 [ + + ]: 101 : if VLIKELY (it->remaining_rounds > 0) {
379 : 6 : --(it->remaining_rounds);
380 : 6 : ++it;
381 : : } else {
382 [ + + ]: 95 : if (it->repeat_interval_ms > 0) {
383 [ + - ]: 28 : callbacks_to_execute.emplace_back(it->key, it->callback);
384 : :
385 : 28 : uint64_t repeat_ticks = (static_cast<uint64_t>(it->repeat_interval_ms) + interval_ms - 1) / interval_ms;
386 : :
387 [ - + ]: 28 : if VUNLIKELY (repeat_ticks == 0) {
388 : : repeat_ticks = 1; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
389 : : }
390 : :
391 : 28 : uint64_t rounds64 = repeat_ticks / slots;
392 : :
393 [ - + ]: 28 : if VUNLIKELY (rounds64 > std::numeric_limits<uint32_t>::max()) {
394 : : // LCOV_EXCL_START GCOVR_EXCL_START
395 : : VLOG_E("WheelTimer: Repeat interval too large.");
396 : :
397 : : timer_index.erase(it->key);
398 : : it = timers.erase(it);
399 : :
400 : : continue;
401 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
402 : 0 : }
403 : :
404 : 28 : auto repeat_ticks_mod = static_cast<uint32_t>(repeat_ticks % slots);
405 : 28 : auto new_rounds = static_cast<uint32_t>(rounds64);
406 : 28 : auto new_slot = (current_slot + repeat_ticks_mod) % slots;
407 : :
408 : 28 : Handler new_handler(it->key, new_rounds, std::move(it->callback), it->repeat_interval_ms);
409 : 28 : auto& new_list = (*wheels)[new_slot];
410 : :
411 [ + - ]: 28 : new_list.emplace_back(std::move(new_handler));
412 : :
413 [ + - ]: 28 : auto new_it = std::prev(new_list.end());
414 : :
415 [ + - ]: 28 : timer_index[it->key] = {new_slot, new_it};
416 : :
417 : 28 : it = timers.erase(it);
418 : 28 : } else {
419 [ + - ]: 67 : callbacks_to_execute.emplace_back(it->key, std::move(it->callback));
420 [ + - ]: 67 : timer_index.erase(it->key);
421 : 67 : it = timers.erase(it);
422 : : }
423 : : }
424 : : }
425 : :
426 : 234 : current_slot = (current_slot + 1) % slots;
427 : :
428 [ + - + - ]: 234 : next_tick += interval;
429 : :
430 [ + + ]: 234 : if (catchup_limit_snapshot > 0) {
431 [ + + ]: 45 : if (++advanced >= catchup_limit_snapshot) {
432 : 1 : break;
433 : : }
434 : : }
435 : : }
436 : :
437 : 222 : now = std::chrono::steady_clock::now();
438 : :
439 [ + - + - : 222 : if VUNLIKELY (next_tick + interval * kStaleTickResetIntervals < now) {
+ - - + ]
440 : : next_tick = now + interval; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
441 : : }
442 : :
443 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
444 [ + - ]: 222 : decltype(callbacks_to_execute) pending_callbacks(&MemoryResource::global_instance());
445 : : #else
446 : : decltype(callbacks_to_execute) pending_callbacks;
447 : : #endif
448 : :
449 : 222 : pending_callbacks.swap(callbacks_to_execute);
450 : :
451 [ + - ]: 222 : lock.unlock();
452 : :
453 [ + + ]: 317 : for (const auto& [key, callback] : pending_callbacks) {
454 [ + - ]: 95 : callback(key);
455 : : }
456 [ + + ]: 475 : }
457 : :
458 : : {
459 [ + - ]: 31 : std::lock_guard lock(mtx);
460 : 31 : is_running.store(false, std::memory_order_release);
461 : 31 : paused_flag.store(false, std::memory_order_release);
462 : 31 : current_slot = 0;
463 : :
464 : : #ifdef VLINK_ENABLE_BASE_MEMORY_RESOURCE
465 [ + - ]: 31 : wheels.emplace(&MemoryResource::global_instance());
466 : : #else
467 : : wheels.emplace();
468 : : #endif
469 : :
470 [ + - ]: 31 : wheels->resize(slots);
471 : :
472 : 31 : timer_index.clear();
473 : 31 : }
474 : :
475 : 31 : cv.notify_all();
476 : 31 : }
477 : :
478 : : } // namespace vlink
|
