Branch data Line data Source code
1 : : /*
2 : : * Copyright 2025 Jinwoo Sung
3 : : *
4 : : * Licensed under the Apache License, Version 2.0 (the "License");
5 : : * you may not use this file except in compliance with the License.
6 : : * You may obtain a copy of the License at
7 : : *
8 : : * http://www.apache.org/licenses/LICENSE-2.0
9 : : *
10 : : * Unless required by applicable law or agreed to in writing, software
11 : : * distributed under the License is distributed on an "AS IS" BASIS,
12 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 : : * See the License for the specific language governing permissions and
14 : : * limitations under the License.
15 : : */
16 : :
17 : : #pragma once
18 : :
19 : : #include <algorithm>
20 : : #include <atomic>
21 : : #include <chrono>
22 : : #include <condition_variable>
23 : : #include <functional>
24 : : #include <mutex>
25 : : #include <shared_mutex>
26 : : #include <vector>
27 : :
28 : : #include "wirestead/base/common.hpp"
29 : :
30 : : namespace wirestead {
31 : : namespace concurrency {
32 : :
33 : : /**
34 : : * @brief Thread-safe state management class
35 : : *
36 : : * Provides thread-safe state management with read-write lock semantics.
37 : : * Multiple readers can access the state simultaneously, but only one writer
38 : : * can modify the state at a time.
39 : : */
40 : : template <typename StateType>
41 : : class ThreadSafeState {
42 : : public:
43 : : using State = StateType;
44 : : using StateCallback = std::function<void(const State&)>;
45 : : using StateCallbackHandle = size_t;
46 : :
47 : : // Constructors
48 : : explicit ThreadSafeState(const State& initial_state = State{});
49 : : ThreadSafeState(const ThreadSafeState&) = delete;
50 : : ThreadSafeState& operator=(const ThreadSafeState&) = delete;
51 : : ThreadSafeState(ThreadSafeState&&) = delete;
52 : : ThreadSafeState& operator=(ThreadSafeState&&) = delete;
53 : :
54 : : // State access methods
55 : : State state() const;
56 : : void set_state(const State& new_state);
57 : : void set_state(State&& new_state);
58 : :
59 : : // Atomic state operations
60 : : bool compare_and_set(const State& expected, const State& desired);
61 : : State exchange(const State& new_state);
62 : :
63 : : // State change notifications
64 : : StateCallbackHandle add_state_change_callback(StateCallback callback);
65 : : void remove_state_change_callback(StateCallbackHandle handle);
66 : : void clear_state_change_callbacks();
67 : :
68 : : // Wait for state change
69 : : void wait_for_state(const State& expected_state, std::chrono::milliseconds timeout = std::chrono::milliseconds(1000));
70 : : void wait_for_state_change(std::chrono::milliseconds timeout = std::chrono::milliseconds(1000));
71 : :
72 : : // Utility methods
73 : : bool is_state(const State& expected_state) const;
74 : : void notify_state_change();
75 : :
76 : : private:
77 : : mutable std::shared_mutex state_mutex_;
78 : : State state_;
79 : : // Monotonically increasing on every state change. wait_for_state_change()
80 : : // captures this before waiting and checks for it to differ from that
81 : : // baseline, so every concurrent waiter observes any given change - a
82 : : // shared "changed" bool that gets reset by whichever waiter wakes first
83 : : // would starve the others.
84 : : std::atomic<uint64_t> state_version_{0};
85 : :
86 : : struct CallbackInfo {
87 : : StateCallbackHandle handle;
88 : : StateCallback callback;
89 : : };
90 : : std::vector<CallbackInfo> callbacks_;
91 : : StateCallbackHandle next_handle_{1};
92 : : mutable std::mutex callbacks_mutex_;
93 : :
94 : : std::condition_variable_any state_cv_;
95 : :
96 : : void notify_callbacks(const State& new_state);
97 : : };
98 : :
99 : : /**
100 : : * @brief Thread-safe atomic state wrapper
101 : : *
102 : : * Lightweight wrapper for atomic state management when full thread-safe
103 : : * state management is not needed.
104 : : */
105 : : template <typename StateType>
106 : : class AtomicState {
107 : : public:
108 : : using State = StateType;
109 : :
110 : : explicit AtomicState(const State& initial_state = State{});
111 : :
112 : : State get() const noexcept;
113 : : void set(const State& new_state) noexcept;
114 : : void set(State&& new_state) noexcept;
115 : :
116 : : bool compare_and_set(const State& expected, const State& desired) noexcept;
117 : : State exchange(const State& new_state) noexcept;
118 : :
119 : : bool is_state(const State& expected_state) const noexcept;
120 : :
121 : : private:
122 : : std::atomic<State> state_;
123 : : };
124 : :
125 : : /**
126 : : * @brief Thread-safe counter with atomic operations
127 : : */
128 : : class ThreadSafeCounter {
129 : : public:
130 : : explicit ThreadSafeCounter(int64_t initial_value = 0);
131 : :
132 : : int64_t get() const noexcept;
133 : : int64_t increment() noexcept;
134 : : int64_t decrement() noexcept;
135 : : int64_t add(int64_t value) noexcept;
136 : : int64_t subtract(int64_t value) noexcept;
137 : :
138 : : bool compare_and_set(int64_t expected, int64_t desired) noexcept;
139 : : int64_t exchange(int64_t new_value) noexcept;
140 : :
141 : : void reset() noexcept;
142 : :
143 : : private:
144 : : std::atomic<int64_t> value_;
145 : : };
146 : :
147 : : /**
148 : : * @brief Thread-safe flag with atomic operations
149 : : */
150 : : class ThreadSafeFlag {
151 : : public:
152 : : explicit ThreadSafeFlag(bool initial_value = false);
153 : :
154 : : bool get() const noexcept;
155 : : void set(bool value = true) noexcept;
156 : : void clear() noexcept;
157 : :
158 : : bool test_and_set() noexcept;
159 : : bool compare_and_set(bool expected, bool desired) noexcept;
160 : :
161 : : void wait_for_true(std::chrono::milliseconds timeout = std::chrono::milliseconds(1000)) const;
162 : : void wait_for_false(std::chrono::milliseconds timeout = std::chrono::milliseconds(1000)) const;
163 : :
164 : : private:
165 : : std::atomic<bool> flag_;
166 : : mutable std::condition_variable cv_;
167 : : mutable std::mutex cv_mutex_;
168 : : };
169 : :
170 : : // Specialization for LinkState
171 : : using ThreadSafeLinkState = ThreadSafeState<base::LinkState>;
172 : : using AtomicLinkState = AtomicState<base::LinkState>;
173 : :
174 : : // Template implementations (must be in header for template instantiation)
175 : : template <typename StateType>
176 : : ThreadSafeState<StateType>::ThreadSafeState(const State& initial_state) : state_(initial_state) {}
177 : :
178 : : template <typename StateType>
179 : : StateType ThreadSafeState<StateType>::state() const {
180 : : std::shared_lock<std::shared_mutex> lock(state_mutex_);
181 : : return state_;
182 : : }
183 : :
184 : : template <typename StateType>
185 : : void ThreadSafeState<StateType>::set_state(const State& new_state) {
186 : : {
187 : : std::unique_lock<std::shared_mutex> lock(state_mutex_);
188 : : state_ = new_state;
189 : : state_version_.fetch_add(1, std::memory_order_relaxed);
190 : : }
191 : : notify_callbacks(new_state);
192 : : state_cv_.notify_all();
193 : : }
194 : :
195 : : template <typename StateType>
196 : : void ThreadSafeState<StateType>::set_state(State&& new_state) {
197 : : {
198 : : std::unique_lock<std::shared_mutex> lock(state_mutex_);
199 : : state_ = std::move(new_state);
200 : : state_version_.fetch_add(1, std::memory_order_relaxed);
201 : : }
202 : : notify_callbacks(state_);
203 : : state_cv_.notify_all();
204 : : }
205 : :
206 : : template <typename StateType>
207 : : bool ThreadSafeState<StateType>::compare_and_set(const State& expected, const State& desired) {
208 : : std::unique_lock<std::shared_mutex> lock(state_mutex_);
209 : : if (state_ == expected) {
210 : : state_ = desired;
211 : : state_version_.fetch_add(1, std::memory_order_relaxed);
212 : : lock.unlock();
213 : : notify_callbacks(desired);
214 : : state_cv_.notify_all();
215 : : return true;
216 : : }
217 : : return false;
218 : : }
219 : :
220 : : template <typename StateType>
221 : : StateType ThreadSafeState<StateType>::exchange(const State& new_state) {
222 : : State old_state;
223 : : {
224 : : std::unique_lock<std::shared_mutex> lock(state_mutex_);
225 : : old_state = state_;
226 : : state_ = new_state;
227 : : state_version_.fetch_add(1, std::memory_order_relaxed);
228 : : }
229 : : notify_callbacks(new_state);
230 : : state_cv_.notify_all();
231 : : return old_state;
232 : : }
233 : :
234 : : template <typename StateType>
235 : : typename ThreadSafeState<StateType>::StateCallbackHandle ThreadSafeState<StateType>::add_state_change_callback(
236 : : StateCallback callback) {
237 : : std::lock_guard<std::mutex> lock(callbacks_mutex_);
238 : : StateCallbackHandle handle = next_handle_++;
239 : : callbacks_.push_back({handle, std::move(callback)});
240 : : return handle;
241 : : }
242 : :
243 : : template <typename StateType>
244 : : void ThreadSafeState<StateType>::remove_state_change_callback(StateCallbackHandle handle) {
245 : : std::lock_guard<std::mutex> lock(callbacks_mutex_);
246 : : callbacks_.erase(std::remove_if(callbacks_.begin(), callbacks_.end(),
247 : : [handle](const CallbackInfo& info) { return info.handle == handle; }),
248 : : callbacks_.end());
249 : : }
250 : :
251 : : template <typename StateType>
252 : : void ThreadSafeState<StateType>::clear_state_change_callbacks() {
253 : : std::lock_guard<std::mutex> lock(callbacks_mutex_);
254 : : callbacks_.clear();
255 : : }
256 : :
257 : : template <typename StateType>
258 : : void ThreadSafeState<StateType>::wait_for_state(const State& expected_state, std::chrono::milliseconds timeout) {
259 : : std::unique_lock<std::shared_mutex> lock(state_mutex_);
260 : : state_cv_.wait_for(lock, timeout, [this, &expected_state] { return state_ == expected_state; });
261 : : }
262 : :
263 : : template <typename StateType>
264 : : void ThreadSafeState<StateType>::wait_for_state_change(std::chrono::milliseconds timeout) {
265 : : std::unique_lock<std::shared_mutex> lock(state_mutex_);
266 : : const uint64_t observed_version = state_version_.load(std::memory_order_relaxed);
267 : : state_cv_.wait_for(lock, timeout, [this, observed_version] {
268 : : return state_version_.load(std::memory_order_relaxed) != observed_version;
269 : : });
270 : : }
271 : :
272 : : template <typename StateType>
273 : : bool ThreadSafeState<StateType>::is_state(const State& expected_state) const {
274 : : std::shared_lock<std::shared_mutex> lock(state_mutex_);
275 : : return state_ == expected_state;
276 : : }
277 : :
278 : : template <typename StateType>
279 : : void ThreadSafeState<StateType>::notify_state_change() {
280 : : // Bump under state_mutex_ - wait_for_state_change() captures its baseline
281 : : // and enters the wait while holding this same lock, so incrementing
282 : : // without it would reopen a lost-wakeup window: a waiter that has already
283 : : // checked the predicate (and found it false) but hasn't yet started
284 : : // waiting would miss this notification and block for the full timeout.
285 : : {
286 : : std::unique_lock<std::shared_mutex> lock(state_mutex_);
287 : : state_version_.fetch_add(1, std::memory_order_relaxed);
288 : : }
289 : : state_cv_.notify_all();
290 : : }
291 : :
292 : : template <typename StateType>
293 : : void ThreadSafeState<StateType>::notify_callbacks(const State& new_state) {
294 : : std::lock_guard<std::mutex> lock(callbacks_mutex_);
295 : : for (const auto& info : callbacks_) {
296 : : try {
297 : : info.callback(new_state);
298 : : } catch (...) {
299 : : // Ignore callback exceptions to prevent state corruption
300 : : }
301 : : }
302 : : }
303 : :
304 : : // AtomicState template implementations
305 : : template <typename StateType>
306 : 477 : AtomicState<StateType>::AtomicState(const State& initial_state) : state_(initial_state) {}
307 : :
308 : : template <typename StateType>
309 : 2336 : StateType AtomicState<StateType>::get() const noexcept {
310 : 2336 : return state_.load();
311 : : }
312 : :
313 : : template <typename StateType>
314 : 841 : void AtomicState<StateType>::set(const State& new_state) noexcept {
315 : 841 : state_.store(new_state);
316 : 841 : }
317 : :
318 : : template <typename StateType>
319 : 1085 : void AtomicState<StateType>::set(State&& new_state) noexcept {
320 : 1085 : state_.store(new_state);
321 : 1085 : }
322 : :
323 : : template <typename StateType>
324 : : bool AtomicState<StateType>::compare_and_set(const State& expected, const State& desired) noexcept {
325 : : State expected_copy = expected;
326 : : return state_.compare_exchange_strong(expected_copy, desired);
327 : : }
328 : :
329 : : template <typename StateType>
330 : : StateType AtomicState<StateType>::exchange(const State& new_state) noexcept {
331 : : return state_.exchange(new_state);
332 : : }
333 : :
334 : : template <typename StateType>
335 : 1529225 : bool AtomicState<StateType>::is_state(const State& expected_state) const noexcept {
336 : 1529225 : return state_.load() == expected_state;
337 : : }
338 : :
339 : : // ThreadSafeCounter implementations
340 : : inline ThreadSafeCounter::ThreadSafeCounter(int64_t initial_value) : value_(initial_value) {}
341 : :
342 : : inline int64_t ThreadSafeCounter::get() const noexcept { return value_.load(); }
343 : :
344 : : inline int64_t ThreadSafeCounter::increment() noexcept { return value_.fetch_add(1) + 1; }
345 : :
346 : : inline int64_t ThreadSafeCounter::decrement() noexcept { return value_.fetch_sub(1) - 1; }
347 : :
348 : : inline int64_t ThreadSafeCounter::add(int64_t value) noexcept { return value_.fetch_add(value) + value; }
349 : :
350 : : inline int64_t ThreadSafeCounter::subtract(int64_t value) noexcept { return value_.fetch_sub(value) - value; }
351 : :
352 : : inline bool ThreadSafeCounter::compare_and_set(int64_t expected, int64_t desired) noexcept {
353 : : return value_.compare_exchange_strong(expected, desired);
354 : : }
355 : :
356 : : inline int64_t ThreadSafeCounter::exchange(int64_t new_value) noexcept { return value_.exchange(new_value); }
357 : :
358 : : inline void ThreadSafeCounter::reset() noexcept { value_.store(0); }
359 : :
360 : : // ThreadSafeFlag implementations
361 : : inline ThreadSafeFlag::ThreadSafeFlag(bool initial_value) : flag_(initial_value) {}
362 : :
363 : : inline bool ThreadSafeFlag::get() const noexcept { return flag_.load(); }
364 : :
365 : : inline void ThreadSafeFlag::set(bool value) noexcept {
366 : : flag_.store(value);
367 : : if (value) {
368 : : cv_.notify_all();
369 : : }
370 : : }
371 : :
372 : : inline void ThreadSafeFlag::clear() noexcept { flag_.store(false); }
373 : :
374 : : inline bool ThreadSafeFlag::test_and_set() noexcept { return flag_.exchange(true); }
375 : :
376 : : inline bool ThreadSafeFlag::compare_and_set(bool expected, bool desired) noexcept {
377 : : return flag_.compare_exchange_strong(expected, desired);
378 : : }
379 : :
380 : : inline void ThreadSafeFlag::wait_for_true(std::chrono::milliseconds timeout) const {
381 : : std::unique_lock<std::mutex> lock(cv_mutex_);
382 : : cv_.wait_for(lock, timeout, [this] { return flag_.load(); });
383 : : }
384 : :
385 : : inline void ThreadSafeFlag::wait_for_false(std::chrono::milliseconds timeout) const {
386 : : std::unique_lock<std::mutex> lock(cv_mutex_);
387 : : cv_.wait_for(lock, timeout, [this] { return !flag_.load(); });
388 : : }
389 : :
390 : : } // namespace concurrency
391 : : } // namespace wirestead
|
