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