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 <boost/asio/any_io_executor.hpp>
22 : : #include <boost/asio/io_context.hpp>
23 : : #include <boost/asio/strand.hpp>
24 : : #include <condition_variable>
25 : : #include <cstdint>
26 : : #include <functional>
27 : : #include <memory>
28 : : #include <mutex>
29 : : #include <string>
30 : : #include <string_view>
31 : : #include <thread>
32 : : #include <utility>
33 : : #include <vector>
34 : :
35 : : #include "wirestead/diagnostics/logger.hpp"
36 : :
37 : : // D-2: a blocking send must not wait for capacity while the calling thread
38 : : // executes any user callback, including callbacks from another channel.
39 : : // The common invocation below marks callback scope; capacity available sends
40 : : // still follow their ordinary acceptance rules. Existing receive-path guards
41 : : // remain compatible because this counter supports nested scopes.
42 : : namespace wirestead {
43 : : namespace wrapper {
44 : : class SendResult;
45 : : namespace detail {
46 : :
47 : : // A depth counter rather than a bool so that a nested/reentrant
48 : : // CallbackGuard on the same thread doesn't clear the flag out from under an
49 : : // outer guard that's still in scope: the inner guard's destructor would
50 : : // otherwise flip g_callback_depth back to "not in a callback" while the
51 : : // outer dispatch is still running, reopening the #449 deadlock this guard
52 : : // exists to prevent.
53 : : inline thread_local int g_callback_depth = 0;
54 : :
55 : : class CallbackGuard {
56 : : public:
57 : 8553 : CallbackGuard() { ++g_callback_depth; }
58 : 8553 : ~CallbackGuard() { --g_callback_depth; }
59 : : CallbackGuard(const CallbackGuard&) = delete;
60 : : CallbackGuard& operator=(const CallbackGuard&) = delete;
61 : : };
62 : :
63 : : // Historical internal name; covers every user-callback kind (D-2).
64 : 4211 : inline bool in_data_callback() { return g_callback_depth > 0; }
65 : :
66 : : // Capacity can only clear if the target I/O context keeps progressing. Test the
67 : : // inner context, not just strand membership: an ordinary task or sibling strand
68 : : // on the same context must not block its sole runner either.
69 : 2353 : inline bool executor_running_here(const boost::asio::any_io_executor& executor) {
70 : : using IoExecutor = boost::asio::io_context::executor_type;
71 [ + + ]: 2353 : if (auto* io = executor.target<IoExecutor>()) return io->running_in_this_thread();
72 [ + + ]: 1811 : if (auto* strand = executor.target<boost::asio::strand<IoExecutor>>())
73 : 1715 : return strand->get_inner_executor().running_in_this_thread();
74 [ + - ]: 96 : if (auto* strand = executor.target<boost::asio::strand<boost::asio::any_io_executor>>())
75 : 96 : return executor_running_here(strand->get_inner_executor());
76 : 0 : return false;
77 : : }
78 : :
79 : : // Admission gate for one wrapper object's user callbacks (D-1 in
80 : : // docs/communication_contract_v0.10_decisions.md).
81 : : //
82 : : // Shutdown complete means "no user callback of this object is running, and
83 : : // none from that run will start". Counting running callbacks alone cannot
84 : : // answer that: a callback path that has already passed its liveness check can
85 : : // register itself after a stop() has looked at the count. So admission, the
86 : : // count and the closed flag are all decided under one mutex, and stop()
87 : : // closes the gate before it waits.
88 : : //
89 : : // Each run carries a generation. A callback admitted for an earlier run is
90 : : // refused after a restart, so a leftover handler cannot be counted against -
91 : : // or delivered during - the new run.
92 : : // Testing seam: called at the point the admission race lives - after a
93 : : // handler has checked that its object is alive, before it registers itself -
94 : : // so a test can park a callback exactly there instead of waiting for that
95 : : // window to happen by chance. Nothing in the library ever sets it.
96 : : using PreAdmissionHook = void (*)();
97 : : inline std::atomic<PreAdmissionHook> g_pre_admission_hook{nullptr};
98 : : // Tests may park a TCP capacity waiter before registering its timed wait.
99 : : inline std::atomic<PreAdmissionHook> g_serial_capacity_wait_hook{nullptr};
100 : : inline std::atomic<void (*)(const SendResult&)> g_serial_capacity_wait_result_hook{nullptr};
101 : : inline std::atomic<void (*)(const SendResult&)> g_serial_send_result_hook{nullptr};
102 : : inline std::atomic<PreAdmissionHook> g_udp_capacity_wait_hook{nullptr};
103 : : inline std::atomic<void (*)(const SendResult&)> g_udp_capacity_wait_result_hook{nullptr};
104 : : inline std::atomic<void (*)(const SendResult&)> g_udp_send_result_hook{nullptr};
105 : : inline std::atomic<void (*)(const SendResult&)> g_udp_server_send_result_hook{nullptr};
106 : : inline std::atomic<PreAdmissionHook> g_uds_capacity_wait_hook{nullptr};
107 : : inline std::atomic<void (*)(const SendResult&)> g_uds_capacity_wait_result_hook{nullptr};
108 : : inline std::atomic<void (*)(const SendResult&)> g_uds_send_result_hook{nullptr};
109 : : inline std::atomic<PreAdmissionHook> g_tcp_capacity_wait_hook{nullptr};
110 : : // Observes the frozen internal outcome after a capacity wait has ended.
111 : : inline std::atomic<void (*)(const SendResult&)> g_tcp_capacity_wait_result_hook{nullptr};
112 : :
113 : : // Observes built-in TCP wrapper send outcomes at the bool boundary.
114 : : inline std::atomic<void (*)(const SendResult&)> g_tcp_send_result_hook{nullptr};
115 : :
116 : : class CallbackGate {
117 : : public:
118 : : // Held for the duration of one callback. `admitted()` false means the gate
119 : : // was closed, or the lease belongs to an earlier run: the caller must not
120 : : // invoke the user callback.
121 : : class Lease {
122 : : public:
123 : : Lease() = default;
124 [ + + ]: 15996 : Lease(CallbackGate* gate, bool admitted) : gate_(admitted ? gate : nullptr) {}
125 : 1 : Lease(Lease&& other) noexcept : gate_(other.gate_) { other.gate_ = nullptr; }
126 : : Lease& operator=(Lease&& other) noexcept {
127 : : if (this != &other) {
128 : : release();
129 : : gate_ = other.gate_;
130 : : other.gate_ = nullptr;
131 : : }
132 : : return *this;
133 : : }
134 : : Lease(const Lease&) = delete;
135 : : Lease& operator=(const Lease&) = delete;
136 : 15997 : ~Lease() { release(); }
137 : :
138 : 15996 : bool admitted() const { return gate_ != nullptr; }
139 : :
140 : : private:
141 : 15997 : void release() {
142 [ + + ]: 15997 : if (gate_ == nullptr) return;
143 : 15599 : CallbackGate* gate = gate_;
144 : 15599 : gate_ = nullptr;
145 : : {
146 : 15599 : std::lock_guard<std::mutex> lock(gate->mutex_);
147 : 15599 : --gate->running_;
148 : 15599 : auto it = std::find(gate->threads_.begin(), gate->threads_.end(), std::this_thread::get_id());
149 [ + - + - ]: 15599 : if (it != gate->threads_.end()) gate->threads_.erase(it);
150 : 15599 : }
151 : 15599 : gate->idle_.notify_all();
152 : : }
153 : :
154 : : CallbackGate* gate_ = nullptr;
155 : : };
156 : :
157 : : // Admission and registration in one step: a callback that is admitted is
158 : : // already counted, so no stop() can observe an empty gate and return while
159 : : // this callback is about to run.
160 : 15996 : Lease enter(uint64_t generation) {
161 [ + + + - ]: 15996 : if (auto hook = g_pre_admission_hook.load(std::memory_order_acquire)) hook();
162 : 15996 : std::lock_guard<std::mutex> lock(mutex_);
163 [ + + + + ]: 15996 : if (closed_ || generation != generation_) return Lease(this, false);
164 : 15599 : ++running_;
165 : 15599 : threads_.push_back(std::this_thread::get_id());
166 : 15599 : return Lease(this, true);
167 : 15996 : }
168 : :
169 : : // Stops admitting. Callbacks already admitted keep running; wait_until_idle()
170 : : // is what waits for them.
171 : 5954 : void close() {
172 : 5954 : std::lock_guard<std::mutex> lock(mutex_);
173 : 5954 : closed_ = true;
174 : 5954 : }
175 : :
176 : : // Opens the gate for a new run and returns that run's generation. Callbacks
177 : : // left over from the previous run are refused by enter().
178 : 2648 : uint64_t open_new_generation() {
179 : 2648 : std::lock_guard<std::mutex> lock(mutex_);
180 : 2648 : closed_ = false;
181 : 2648 : return ++generation_;
182 : 2648 : }
183 : :
184 : : uint64_t generation() const {
185 : : std::lock_guard<std::mutex> lock(mutex_);
186 : : return generation_;
187 : : }
188 : :
189 : : // True when this thread is running a callback of this object: waiting for
190 : : // the gate to drain from here would wait for itself.
191 : 5956 : bool active_on_this_thread() const {
192 : 5956 : std::lock_guard<std::mutex> lock(mutex_);
193 [ + - ]: 11912 : return std::find(threads_.begin(), threads_.end(), std::this_thread::get_id()) != threads_.end();
194 : 5956 : }
195 : :
196 : 214 : bool idle() const {
197 : 214 : std::lock_guard<std::mutex> lock(mutex_);
198 : 214 : return running_ == 0;
199 : 214 : }
200 : :
201 : 5787 : void wait_until_idle() {
202 : 5787 : std::unique_lock<std::mutex> lock(mutex_);
203 [ + - ]: 11575 : idle_.wait(lock, [this] { return running_ == 0; });
204 : 5787 : }
205 : :
206 : : private:
207 : : mutable std::mutex mutex_;
208 : : std::condition_variable idle_;
209 : : int running_ = 0;
210 : : bool closed_ = false;
211 : : uint64_t generation_ = 0;
212 : : std::vector<std::thread::id> threads_;
213 : : };
214 : :
215 : : // Invokes every wrapper callback under the D-2 nonwaiting-send guard and
216 : : // contains user exceptions so they cannot escape io_context::run(). The guard
217 : : // is restored before exception logging, including for nested callbacks.
218 : : template <typename Callback, typename... Args>
219 : 10621 : void invoke_user_callback(std::string_view component, std::string_view operation, const Callback& callback,
220 : : Args&&... args) {
221 [ + + ]: 10621 : if (!callback) return;
222 : : try {
223 : 5049 : CallbackGuard guard;
224 : 5049 : callback(std::forward<Args>(args)...);
225 : 5058 : } catch (const std::exception& e) {
226 : 12 : WIRESTEAD_LOG_ERROR(component, operation, "Uncaught exception in user callback: " + std::string(e.what()));
227 : 2 : } catch (...) {
228 : 1 : WIRESTEAD_LOG_ERROR(component, operation, "Uncaught non-standard exception in user callback");
229 : : }
230 : : }
231 : :
232 : : // Overload for a handler snapshotted as a shared pointer rather than copied
233 : : // out of its guarded member - see interface::SharedCallback. A null pointer
234 : : // means "not registered", exactly as an empty std::function does above.
235 : : template <typename Callback, typename... Args>
236 : 4635 : void invoke_user_callback(std::string_view component, std::string_view operation,
237 : : const std::shared_ptr<const Callback>& callback, Args&&... args) {
238 [ + + ]: 4635 : if (!callback) return;
239 : 4034 : invoke_user_callback(component, operation, *callback, std::forward<Args>(args)...);
240 : : }
241 : :
242 : : inline std::atomic<void (*)(const SendResult&)> g_tcp_server_send_result_hook{nullptr};
243 : : inline std::atomic<void (*)(const SendResult&)> g_uds_server_send_result_hook{nullptr};
244 : : inline std::atomic<void (*)()> g_tcp_server_capacity_wait_hook{nullptr};
245 : : inline std::atomic<void (*)(const SendResult&)> g_tcp_server_capacity_wait_result_hook{nullptr};
246 : : inline std::atomic<void (*)()> g_uds_server_capacity_wait_hook{nullptr};
247 : : inline std::atomic<void (*)(const SendResult&)> g_uds_server_capacity_wait_result_hook{nullptr};
248 : : } // namespace detail
249 : : } // namespace wrapper
250 : : } // namespace wirestead
|
