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 : : #include "wirestead/wrapper/udp/udp_server.hpp"
18 : :
19 : : #include <spdlog/fmt/fmt.h>
20 : :
21 : : #include <boost/asio/executor_work_guard.hpp>
22 : : #include <boost/asio/io_context.hpp>
23 : : #include <boost/asio/steady_timer.hpp>
24 : : #include <iostream>
25 : : #include <mutex>
26 : : #include <shared_mutex>
27 : : #include <stop_token>
28 : : #include <thread>
29 : : #include <unordered_map>
30 : : #include <vector>
31 : :
32 : : #include "wirestead/base/common.hpp"
33 : : #include "wirestead/concurrency/io_thread_hook.hpp"
34 : : #include "wirestead/factory/channel_factory.hpp"
35 : : #include "wirestead/transport/udp/udp.hpp"
36 : : #include "wirestead/wrapper/callback_guard.hpp"
37 : : #include "wirestead/wrapper/error_context_builder.hpp"
38 : :
39 : : namespace wirestead {
40 : : namespace wrapper {
41 : :
42 : : namespace {
43 : : // std::hash<boost::asio::ip::udp::endpoint> is not available before Boost 1.74.
44 : : // Provide a portable hash by combining the raw address bytes and port.
45 : : struct UdpEndpointHash {
46 : 23 : std::size_t operator()(const boost::asio::ip::udp::endpoint& ep) const noexcept {
47 : 23 : std::size_t seed = 0;
48 : 138 : auto combine = [&](std::size_t v) { seed ^= v + 0x9e3779b9u + (seed << 6) + (seed >> 2); };
49 [ + - ]: 23 : if (ep.address().is_v4()) {
50 [ + + ]: 115 : for (auto byte : ep.address().to_v4().to_bytes()) {
51 : 92 : combine(std::hash<unsigned char>{}(byte));
52 : : }
53 : : } else {
54 [ # # ]: 0 : for (auto byte : ep.address().to_v6().to_bytes()) {
55 : 0 : combine(std::hash<unsigned char>{}(byte));
56 : : }
57 : : }
58 : 23 : combine(std::hash<unsigned short>{}(ep.port()));
59 : 23 : return seed;
60 : : }
61 : : };
62 : : } // namespace
63 : :
64 : : struct UdpServer::Impl : public std::enable_shared_from_this<Impl> {
65 : : config::UdpConfig cfg;
66 : : std::shared_ptr<transport::UdpChannel> channel;
67 : : std::shared_ptr<boost::asio::io_context> external_ioc;
68 : : std::atomic<bool> use_external_context{false};
69 : : std::atomic<bool> manage_external_context{false};
70 : : std::jthread external_thread;
71 : : std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>> work_guard;
72 : :
73 : : mutable std::shared_mutex mutex;
74 : : std::mutex bp_mutex_;
75 : : std::condition_variable bp_cv_;
76 : : std::vector<std::promise<bool>> pending_promises;
77 : : std::atomic<bool> started{false};
78 : : std::atomic<bool> is_listening{false};
79 : :
80 : : // Virtual Session Management
81 : : struct SessionEntry {
82 : : boost::asio::ip::udp::endpoint endpoint;
83 : : std::shared_ptr<framer::IFramer> framer;
84 : : std::chrono::steady_clock::time_point last_seen;
85 : : };
86 : : ClientId next_client_id{1};
87 : : std::unordered_map<boost::asio::ip::udp::endpoint, ClientId, UdpEndpointHash> endpoint_to_id;
88 : : std::unordered_map<ClientId, SessionEntry> sessions;
89 : 29 : std::chrono::milliseconds session_timeout{0}; // 0 = disabled
90 : : std::unique_ptr<boost::asio::steady_timer> reaper_timer;
91 : : std::atomic<bool> auto_start{false};
92 : : std::atomic<bool> client_limit_enabled{false};
93 : : std::atomic<size_t> max_clients_limit{0};
94 : :
95 : : ConnectionHandler on_connect{nullptr};
96 : : ConnectionHandler on_disconnect{nullptr};
97 : : // Shared snapshots: the strand copies one out per received datagram, and a
98 : : // std::function copy allocates whenever the user handler outgrows its
99 : : // small-object buffer. See interface::SharedCallback.
100 : : interface::SharedCallback<MessageHandler> on_data;
101 : : interface::SharedCallback<BatchMessageHandler> on_data_batch_;
102 : : ErrorHandler on_error{nullptr};
103 : : std::function<void(size_t)> bp_handler{nullptr};
104 : : FramerFactory framer_factory{nullptr};
105 : : interface::SharedCallback<MessageHandler> on_message;
106 : : interface::SharedCallback<BatchMessageHandler> on_message_batch_;
107 : :
108 : 29 : std::shared_ptr<bool> is_alive{std::make_shared<bool>(true)};
109 : :
110 : : // Batching logic
111 : : std::vector<MessageContext> data_batch_queue_;
112 : : std::vector<MessageContext> message_batch_queue_;
113 : : std::unique_ptr<boost::asio::steady_timer> batch_timer_;
114 : : size_t max_batch_size_ = 100;
115 : 29 : std::chrono::milliseconds max_batch_latency_{1};
116 : :
117 : 25 : explicit Impl(const config::UdpConfig& config) : cfg(config) {}
118 : 3 : Impl(const config::UdpConfig& config, std::shared_ptr<boost::asio::io_context> ioc)
119 [ + - + - ]: 3 : : cfg(config), external_ioc(std::move(ioc)), use_external_context(external_ioc != nullptr) {}
120 : 1 : explicit Impl(std::shared_ptr<interface::Channel> ch)
121 : 1 : : channel(std::dynamic_pointer_cast<transport::UdpChannel>(ch)) {
122 : : // #450: setup_internal_handlers() captures weak_from_this() - calling it
123 : : // from inside this constructor would capture an empty weak_ptr, since
124 : : // enable_shared_from_this isn't wired up until make_shared() finishes
125 : : // constructing the object. Deferred to UdpServer's own constructor,
126 : : // which runs after impl_ is a fully-formed shared_ptr<Impl>.
127 : 1 : }
128 : :
129 : 29 : ~Impl() {
130 : 29 : *is_alive = false;
131 : : try {
132 : 29 : stop();
133 : 0 : } catch (...) {
134 : 0 : }
135 : 29 : }
136 : :
137 : 103 : void fulfill_all_locked(bool value) {
138 [ + + ]: 124 : for (auto& promise : pending_promises) {
139 : : try {
140 : 21 : promise.set_value(value);
141 : 0 : } catch (...) {
142 : 0 : }
143 : : }
144 : 103 : pending_promises.clear();
145 : 103 : }
146 : :
147 : 2 : void flush_batches() {
148 : 2 : std::unique_lock<std::shared_mutex> lock(mutex);
149 [ + - ]: 2 : if (!data_batch_queue_.empty()) {
150 : 2 : auto handler = on_data_batch_;
151 : 2 : auto batch = std::move(data_batch_queue_);
152 : 2 : data_batch_queue_.clear();
153 [ + - ]: 2 : if (handler) {
154 : 2 : lock.unlock();
155 : 2 : detail::invoke_user_callback("udp_server", "on_data_batch", handler, batch);
156 : 2 : lock.lock();
157 : : }
158 : 2 : }
159 [ + + ]: 2 : if (!message_batch_queue_.empty()) {
160 : 1 : auto handler = on_message_batch_;
161 : 1 : auto batch = std::move(message_batch_queue_);
162 : 1 : message_batch_queue_.clear();
163 [ + - ]: 1 : if (handler) {
164 : 1 : lock.unlock();
165 : 1 : detail::invoke_user_callback("udp_server", "on_message_batch", handler, batch);
166 : 1 : lock.lock();
167 : : }
168 : 1 : }
169 [ + - ]: 2 : if (batch_timer_) {
170 : 2 : batch_timer_->cancel();
171 : : }
172 : 2 : }
173 : :
174 : 5 : void schedule_batch_timer() {
175 [ - + ]: 5 : if (!batch_timer_) return;
176 : 5 : batch_timer_->expires_after(max_batch_latency_);
177 : 10 : batch_timer_->async_wait([this, weak_impl = weak_from_this(),
178 : 5 : alive = std::weak_ptr<bool>(is_alive)](const boost::system::error_code& ec) {
179 : 5 : auto impl_keepalive = weak_impl.lock();
180 [ - + ]: 5 : if (!impl_keepalive) return;
181 : 5 : auto lock = alive.lock();
182 [ + - - + : 5 : if (!lock || !(*lock)) return;
- + ]
183 : :
184 [ + + ]: 5 : if (!ec) {
185 : 2 : flush_batches();
186 : : }
187 : 5 : });
188 : : }
189 : :
190 : 3 : void schedule_reaper() {
191 [ + - + - : 3 : if (!started.load() || !reaper_timer || session_timeout.count() <= 0) return;
- + - + ]
192 : :
193 : : // Run reaper at interval proportional to timeout (min 100ms, max 5s)
194 : : auto interval =
195 : 3 : std::max(std::chrono::milliseconds(100), std::min(std::chrono::milliseconds(5000), session_timeout / 2));
196 : :
197 : 3 : reaper_timer->expires_after(interval);
198 : 6 : reaper_timer->async_wait([this, weak_impl = weak_from_this(),
199 : 3 : alive = std::weak_ptr<bool>(is_alive)](const boost::system::error_code& ec) {
200 : 3 : auto impl_keepalive = weak_impl.lock();
201 [ - + ]: 3 : if (!impl_keepalive) return;
202 : 3 : auto lock = alive.lock();
203 [ + - - + : 3 : if (!lock || !(*lock)) return;
- + ]
204 : :
205 [ + + ]: 3 : if (!ec) {
206 : 2 : run_reaper();
207 : 2 : schedule_reaper();
208 : : }
209 : 3 : });
210 : : }
211 : :
212 : 2 : void run_reaper() {
213 [ - + ]: 2 : if (session_timeout.count() <= 0) return;
214 : :
215 : 2 : std::vector<std::pair<ClientId, std::string>> to_remove_with_info;
216 : 2 : auto now = std::chrono::steady_clock::now();
217 : :
218 : 2 : ConnectionHandler disconnect_handler;
219 : : {
220 : 2 : std::unique_lock<std::shared_mutex> lock(mutex);
221 [ + + ]: 4 : for (auto it = sessions.begin(); it != sessions.end();) {
222 [ + - + - : 2 : if (now - it->second.last_seen > session_timeout) {
+ + ]
223 : : std::string info =
224 : 1 : fmt::format("{}:{}", it->second.endpoint.address().to_string(), it->second.endpoint.port());
225 : 1 : endpoint_to_id.erase(it->second.endpoint);
226 : 1 : to_remove_with_info.push_back({it->first, info});
227 : 1 : it = sessions.erase(it);
228 : 1 : } else {
229 : 1 : ++it;
230 : : }
231 : : }
232 : 2 : disconnect_handler = on_disconnect;
233 : 2 : }
234 : :
235 : : // Call disconnect handlers outside the lock
236 [ + + ]: 3 : for (auto const& [id, info] : to_remove_with_info) {
237 : 1 : detail::invoke_user_callback("udp_server", "on_disconnect", disconnect_handler, ConnectionContext(id, info));
238 : : }
239 : 2 : }
240 : :
241 : 21 : void setup_internal_handlers() {
242 [ - + ]: 21 : if (!channel) return;
243 : :
244 : 21 : batch_timer_ = std::make_unique<boost::asio::steady_timer>(channel->get_executor());
245 : :
246 : 21 : std::weak_ptr<Impl> weak_impl = weak_from_this();
247 : :
248 : 21 : channel->on_bytes_from([this, weak_impl](memory::ConstByteSpan data, const boost::asio::ip::udp::endpoint& ep) {
249 : : // #450: keep Impl alive for the duration of this callback - on an
250 : : // externally-owned io_context, stop() doesn't join/wait for in-flight
251 : : // handlers, so a bare `this` could otherwise dangle.
252 : 20 : auto impl_keepalive = weak_impl.lock();
253 [ - + ]: 20 : if (!impl_keepalive) return;
254 : :
255 : : // #449: everything below runs synchronously on this io thread - mark
256 : : // it so a blocking send_to() called from within one of these
257 : : // callbacks fails fast instead of deadlocking.
258 : 20 : detail::CallbackGuard callback_guard;
259 : :
260 : 20 : ClientId client_id = 0;
261 : 20 : bool is_new = false;
262 : 20 : ConnectionHandler connect_handler_copy{nullptr};
263 : :
264 : : {
265 : 20 : std::unique_lock<std::shared_mutex> lock(mutex);
266 : 20 : auto it = endpoint_to_id.find(ep);
267 [ + + ]: 20 : if (it == endpoint_to_id.end()) {
268 [ + + + + : 21 : if (client_limit_enabled.load() && sessions.size() >= max_clients_limit.load()) {
+ + ]
269 : 1 : return;
270 : : }
271 : 18 : client_id = next_client_id++;
272 : 18 : endpoint_to_id[ep] = client_id;
273 : 18 : SessionEntry entry;
274 : 18 : entry.endpoint = ep;
275 : 18 : entry.last_seen = std::chrono::steady_clock::now();
276 : 18 : is_new = true;
277 : :
278 : : // Create framer for new session
279 [ + + ]: 18 : if (framer_factory) {
280 : 3 : auto framer = framer_factory();
281 [ + - ]: 3 : if (framer) {
282 : 3 : framer->on_message([this, client_id](memory::ConstByteSpan msg) {
283 : : // #441: snapshot under a shared_lock (pure read), build the
284 : : // copy before taking the exclusive lock for queue mutation.
285 : : bool batch_mode;
286 : 4 : interface::SharedCallback<MessageHandler> on_message_handler;
287 : : {
288 : 4 : std::shared_lock<std::shared_mutex> lock(mutex);
289 : 4 : batch_mode = static_cast<bool>(on_message_batch_);
290 : 4 : on_message_handler = on_message;
291 : 4 : }
292 : :
293 [ + + ]: 4 : if (batch_mode) {
294 : 3 : MessageContext ctx(client_id, memory::SafeDataBuffer(msg));
295 : 3 : interface::SharedCallback<BatchMessageHandler> flush_handler;
296 : 3 : std::vector<MessageContext> batch;
297 : : {
298 : 3 : std::unique_lock<std::shared_mutex> lock(mutex);
299 : 3 : message_batch_queue_.emplace_back(std::move(ctx));
300 [ + + ]: 3 : if (message_batch_queue_.size() >= max_batch_size_) {
301 : 1 : flush_handler = on_message_batch_;
302 : 1 : batch = std::move(message_batch_queue_);
303 : 1 : message_batch_queue_.clear();
304 [ + - ]: 2 : } else if (message_batch_queue_.size() == 1) {
305 : 2 : schedule_batch_timer();
306 : : }
307 : 3 : }
308 : 3 : detail::invoke_user_callback("udp_server", "on_message_batch", flush_handler, batch);
309 : 3 : return;
310 : 3 : }
311 : :
312 : 1 : detail::invoke_user_callback("udp_server", "on_message", on_message_handler,
313 : 2 : MessageContext(client_id, msg));
314 : 4 : });
315 : 3 : entry.framer = std::move(framer);
316 : : }
317 : 3 : }
318 : 18 : sessions[client_id] = std::move(entry);
319 : 18 : } else {
320 : 1 : client_id = it->second;
321 : 1 : sessions[client_id].last_seen = std::chrono::steady_clock::now();
322 : : }
323 : 19 : connect_handler_copy = on_connect;
324 : 20 : }
325 : :
326 [ + + ]: 19 : if (is_new) {
327 : 18 : detail::invoke_user_callback(
328 : : "udp_server", "on_connect", connect_handler_copy,
329 : 36 : ConnectionContext(client_id, fmt::format("{}:{}", ep.address().to_string(), ep.port())));
330 : : }
331 : :
332 : : {
333 : : // #441: snapshot the handler under a shared_lock (not unique_lock) -
334 : : // this is a pure read, matching try_send's locking level so it no
335 : : // longer blocks concurrent sends even briefly.
336 : : bool batch_mode;
337 : 19 : interface::SharedCallback<MessageHandler> data_handler_copy;
338 : : {
339 : 19 : std::shared_lock<std::shared_mutex> lock(mutex);
340 : 19 : batch_mode = static_cast<bool>(on_data_batch_);
341 : 19 : data_handler_copy = on_data;
342 : 19 : }
343 : :
344 [ + + ]: 19 : if (batch_mode) {
345 : : // #441: build the copy before taking the exclusive lock, so the
346 : : // lock is only held for the queue mutation itself, not the
347 : : // allocation.
348 : 4 : MessageContext ctx(client_id, memory::SafeDataBuffer(data));
349 : 4 : interface::SharedCallback<BatchMessageHandler> flush_handler;
350 : 4 : std::vector<MessageContext> batch;
351 : : {
352 : 4 : std::unique_lock<std::shared_mutex> lock(mutex);
353 : 4 : data_batch_queue_.emplace_back(std::move(ctx));
354 [ + + ]: 4 : if (data_batch_queue_.size() >= max_batch_size_) {
355 : 1 : flush_handler = on_data_batch_;
356 : 1 : batch = std::move(data_batch_queue_);
357 : 1 : data_batch_queue_.clear();
358 [ + - ]: 3 : } else if (data_batch_queue_.size() == 1) {
359 : 3 : schedule_batch_timer();
360 : : }
361 : 4 : }
362 : 4 : detail::invoke_user_callback("udp_server", "on_data_batch", flush_handler, batch);
363 : 4 : } else {
364 : 30 : detail::invoke_user_callback("udp_server", "on_data", data_handler_copy, MessageContext(client_id, data));
365 : : }
366 : 19 : }
367 : :
368 : : // Push to framer
369 : 19 : std::shared_ptr<framer::IFramer> target_framer;
370 : : {
371 : 19 : std::shared_lock<std::shared_mutex> lock(mutex);
372 : 19 : auto it = sessions.find(client_id);
373 [ + - ]: 19 : if (it != sessions.end()) {
374 : 19 : target_framer = it->second.framer;
375 : : }
376 : 19 : }
377 [ + + ]: 19 : if (target_framer) {
378 : 3 : target_framer->push_bytes(data);
379 : : }
380 : 22 : });
381 : :
382 : 21 : channel->on_backpressure([this, weak_impl](size_t queued) {
383 : 0 : bp_cv_.notify_all();
384 : 0 : auto impl_keepalive = weak_impl.lock();
385 [ # # ]: 0 : if (!impl_keepalive) return;
386 : 0 : std::function<void(size_t)> handler;
387 : : {
388 : 0 : std::shared_lock<std::shared_mutex> lock(mutex);
389 : 0 : handler = bp_handler;
390 : 0 : }
391 : 0 : detail::invoke_user_callback("udp_server", "on_backpressure", handler, queued);
392 : 0 : });
393 : :
394 : 21 : channel->on_state([this, weak_impl](base::LinkState state) {
395 : 77 : auto impl_keepalive = weak_impl.lock();
396 [ + + ]: 77 : if (!impl_keepalive) return;
397 : 74 : ErrorHandler error_handler_copy{nullptr};
398 [ + + + + ]: 74 : if (state == base::LinkState::Listening || state == base::LinkState::Connected) {
399 : 36 : is_listening.store(true);
400 : 36 : std::unique_lock<std::shared_mutex> lock(mutex);
401 : 36 : fulfill_all_locked(true);
402 [ + + + + : 74 : } else if (state == base::LinkState::Error || state == base::LinkState::Closed ||
- + ]
403 : : state == base::LinkState::Idle) {
404 : 18 : is_listening.store(false);
405 : 18 : std::unique_lock<std::shared_mutex> lock(mutex);
406 : 18 : fulfill_all_locked(false);
407 [ + + ]: 18 : if (state == base::LinkState::Error) {
408 : 1 : error_handler_copy = on_error;
409 : : }
410 : 18 : }
411 : :
412 [ + - ]: 74 : detail::invoke_user_callback("udp_server", "on_error", error_handler_copy,
413 [ + - + - : 148 : channel ? detail::build_error_context(*channel, "Server error")
- - ]
414 : : : ErrorContext(ErrorCode::IoError, "Server error"));
415 : 77 : });
416 : 21 : }
417 : :
418 : 21 : std::future<bool> start() {
419 : 21 : std::unique_lock<std::shared_mutex> lock(mutex);
420 [ - + ]: 21 : if (is_listening.load()) {
421 : 0 : std::promise<bool> p;
422 : 0 : p.set_value(true);
423 : 0 : return p.get_future();
424 : 0 : }
425 : :
426 : 21 : std::promise<bool> p;
427 : 21 : auto fut = p.get_future();
428 : 21 : pending_promises.emplace_back(std::move(p));
429 : :
430 [ - + ]: 21 : if (started.exchange(true)) {
431 : 0 : return fut;
432 : : }
433 : :
434 [ + + ]: 21 : if (!channel) {
435 : 20 : channel = std::dynamic_pointer_cast<transport::UdpChannel>(factory::ChannelFactory::create(cfg, external_ioc));
436 : 20 : setup_internal_handlers();
437 : : }
438 : :
439 : 21 : lock.unlock();
440 : 21 : channel->start();
441 : :
442 : 21 : lock.lock();
443 [ + + + - : 21 : if (use_external_context.load() && manage_external_context.load() && !external_thread.joinable()) {
+ - + + ]
444 [ + - - + : 1 : if (external_ioc->stopped()) external_ioc->restart();
- - ]
445 : 2 : work_guard = std::make_unique<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>(
446 : 3 : external_ioc->get_executor());
447 : 2 : external_thread = std::jthread([ioc = external_ioc](std::stop_token st) {
448 : 1 : wirestead::concurrency::run_io_thread_init();
449 : : try {
450 : 2 : std::stop_callback cb(st, [ioc] { ioc->stop(); });
451 : 1 : ioc->run();
452 : 1 : } catch (...) {
453 : 0 : }
454 : 2 : });
455 : : }
456 : :
457 [ + - + + : 21 : if (channel && session_timeout.count() > 0) {
+ + ]
458 : 1 : reaper_timer = std::make_unique<boost::asio::steady_timer>(channel->get_executor());
459 : 1 : schedule_reaper();
460 : : }
461 : :
462 : 21 : return fut;
463 : 21 : }
464 : :
465 : 49 : void stop() {
466 : 49 : bool should_join = false;
467 : : {
468 : 49 : std::unique_lock<std::shared_mutex> lock(mutex);
469 [ + + ]: 49 : if (!started.exchange(false)) {
470 : 28 : is_listening.store(false);
471 : 28 : fulfill_all_locked(false);
472 : 28 : return;
473 : : }
474 : 21 : bp_cv_.notify_all();
475 : :
476 [ + + ]: 21 : if (reaper_timer) {
477 : 1 : reaper_timer->cancel();
478 : 1 : reaper_timer.reset();
479 : : }
480 : :
481 [ + - ]: 21 : if (batch_timer_) {
482 : 21 : batch_timer_->cancel();
483 : 21 : batch_timer_.reset();
484 : : }
485 : :
486 [ + - ]: 21 : if (channel) {
487 : 21 : lock.unlock();
488 : 21 : channel->stop();
489 : : // Clear callbacks after stop() rather than before: the
490 : : // transport-level fix (#436) already synchronizes callback reads
491 : : // against these setters, but clearing after stop() means no
492 : : // in-flight handler can observe a null callback mid-shutdown in the
493 : : // first place - belt and braces once the underlying race is fixed
494 : : // at the source. Also now clears on_backpressure, which this path
495 : : // previously never did at all.
496 : 21 : channel->on_bytes_from(nullptr);
497 : 21 : channel->on_state(nullptr);
498 : 21 : channel->on_backpressure(nullptr);
499 : 21 : lock.lock();
500 : : }
501 : :
502 [ + + + - : 21 : if (use_external_context.load() && manage_external_context.load()) {
+ + ]
503 [ + - + - ]: 1 : if (external_ioc) external_ioc->stop();
504 : 1 : should_join = true;
505 : : }
506 : :
507 : 21 : is_listening.store(false);
508 : 21 : endpoint_to_id.clear();
509 : 21 : sessions.clear();
510 : 21 : next_client_id = 1;
511 : 21 : fulfill_all_locked(false);
512 : 49 : }
513 : :
514 [ + + + - : 21 : if (should_join && external_thread.joinable()) {
+ + ]
515 : : try {
516 [ + - ]: 1 : if (std::this_thread::get_id() != external_thread.get_id()) {
517 : 1 : external_thread.request_stop();
518 : 1 : external_thread.join();
519 : : } else {
520 : 0 : external_thread.detach();
521 : : }
522 : 0 : } catch (...) {
523 : 0 : }
524 : : }
525 : 21 : std::unique_lock<std::shared_mutex> lock(mutex);
526 : 21 : channel.reset();
527 : 21 : }
528 : :
529 : 5 : bool send_to(ClientId client_id, std::string_view data) {
530 [ + + ]: 5 : if (cfg.backpressure_strategy == base::constants::BackpressureStrategy::Reliable)
531 : 4 : return send_to_blocking(client_id, data);
532 : 1 : return try_send_to(client_id, data);
533 : : }
534 : :
535 : 39 : bool try_broadcast(std::string_view data) {
536 : 39 : std::shared_lock<std::shared_mutex> lock(mutex);
537 [ + + ]: 39 : if (!channel) return false;
538 : 38 : bool sent = false;
539 : 38 : auto bytes = base::safe_convert::string_to_bytes(data);
540 [ + + ]: 108 : for (const auto& [id, entry] : sessions) {
541 : 70 : sent |= channel->async_try_write_to(memory::ConstByteSpan(bytes.first, bytes.second), entry.endpoint);
542 : : }
543 : 38 : return sent;
544 : 39 : }
545 : :
546 : 36 : bool broadcast(std::string_view data) { return try_broadcast(data); }
547 : :
548 : : // channel->on_backpressure() calls bp_cv_.notify_all() from the transport's io_context
549 : : // thread without holding bp_mutex_ (backpressure_active_ is a plain atomic on the transport
550 : : // side, not guarded by bp_mutex_ at all). That makes a classic lost-wakeup race possible: a
551 : : // waiter can check the predicate, find it still blocking, and be in the process of
552 : : // registering to wait when the notify fires - in the rare case that race is lost, an
553 : : // unbounded wait() would block forever. Poll with a bounded timeout instead so a missed
554 : : // notify only costs a short delay rather than a permanent hang (see #427, #431).
555 : : //
556 : : // Returns false without sending instead of waiting if called from the
557 : : // channel's own io thread while backpressure is active - e.g. a blocking
558 : : // send_to() called from inside an on_data/on_message callback. Clearing
559 : : // backpressure requires that same io thread to make progress, so
560 : : // blocking here would deadlock forever rather than eventually clear
561 : : // (#449).
562 : : // #509: see identical rationale in wrapper/tcp_server/tcp_server.cc -
563 : : // bounded retry rather than a single attempt after the wait exits.
564 : : static constexpr int kMaxBlockingSendAttempts = 5;
565 : :
566 : 5 : bool send_to_blocking(ClientId client_id, std::string_view data) {
567 [ + + ]: 10 : for (int attempt = 0; attempt < kMaxBlockingSendAttempts; ++attempt) {
568 : 9 : std::unique_lock<std::mutex> bp_lock(bp_mutex_);
569 : 18 : auto predicate = [this] {
570 : 18 : std::shared_lock<std::shared_mutex> lock(mutex);
571 [ + + + - : 36 : return !started.load() || !channel || !channel->is_backpressure_active();
+ - + - ]
572 : 18 : };
573 [ + - - + : 9 : if (!predicate() && detail::in_data_callback()) return false;
- - - + ]
574 [ + - - + ]: 9 : while (!bp_cv_.wait_for(bp_lock, std::chrono::milliseconds(50), predicate)) {
575 : : }
576 : 9 : bp_lock.unlock();
577 [ + - + + ]: 9 : if (try_send_to(client_id, data)) return true;
578 : 9 : }
579 : 1 : return false;
580 : : }
581 : :
582 : 12 : bool try_send_to(ClientId client_id, std::string_view data) {
583 : 12 : std::shared_lock<std::shared_mutex> lock(mutex);
584 : 12 : auto it = sessions.find(client_id);
585 [ + + - + : 12 : if (it == sessions.end() || !channel) return false;
+ + ]
586 : :
587 : 7 : auto bytes = base::safe_convert::string_to_bytes(data);
588 : 7 : return channel->async_try_write_to(memory::ConstByteSpan(bytes.first, bytes.second), it->second.endpoint);
589 : 12 : }
590 : :
591 : 8 : RuntimeStats stats() const {
592 : 8 : std::shared_lock<std::shared_mutex> lock(mutex);
593 [ + - + - ]: 16 : return channel ? channel->stats() : RuntimeStats{};
594 : 8 : }
595 : :
596 : 0 : void reset_stats() {
597 : 0 : std::shared_lock<std::shared_mutex> lock(mutex);
598 [ # # # # ]: 0 : if (channel) channel->reset_stats();
599 : 0 : }
600 : : };
601 : :
602 : 4 : UdpServer::UdpServer(uint16_t port) {
603 : 4 : config::UdpConfig cfg;
604 : 4 : cfg.local_port = port;
605 : 4 : impl_ = std::make_shared<Impl>(cfg);
606 : 4 : }
607 : :
608 : 21 : UdpServer::UdpServer(const config::UdpConfig& cfg) : impl_(std::make_shared<Impl>(cfg)) {}
609 : :
610 : 3 : UdpServer::UdpServer(const config::UdpConfig& cfg, std::shared_ptr<boost::asio::io_context> ioc)
611 : 3 : : impl_(std::make_shared<Impl>(cfg, ioc)) {}
612 : :
613 : 1 : UdpServer::UdpServer(std::shared_ptr<interface::Channel> ch) : impl_(std::make_shared<Impl>(std::move(ch))) {
614 : 1 : impl_->setup_internal_handlers();
615 : 1 : }
616 : :
617 : 33 : UdpServer::~UdpServer() = default;
618 : :
619 : 0 : UdpServer::UdpServer(UdpServer&&) noexcept = default;
620 : 0 : UdpServer& UdpServer::operator=(UdpServer&&) noexcept = default;
621 : :
622 : 21 : std::future<bool> UdpServer::start() { return impl_->start(); }
623 : 20 : void UdpServer::stop() { impl_->stop(); }
624 : 8 : bool UdpServer::listening() const { return impl_->is_listening.load(); }
625 : 8 : RuntimeStats UdpServer::stats() const { return impl_->stats(); }
626 : 0 : void UdpServer::reset_stats() { impl_->reset_stats(); }
627 : :
628 : 36 : bool UdpServer::broadcast(std::string_view data) { return impl_->broadcast(data); }
629 : 3 : bool UdpServer::try_broadcast(std::string_view data) { return impl_->try_broadcast(data); }
630 : 5 : bool UdpServer::send_to(ClientId client_id, std::string_view data) { return impl_->send_to(client_id, data); }
631 : 2 : bool UdpServer::try_send_to(ClientId client_id, std::string_view data) { return impl_->try_send_to(client_id, data); }
632 : :
633 : 1 : bool UdpServer::send_to_blocking(ClientId client_id, std::string_view data) {
634 : 1 : return impl_->send_to_blocking(client_id, data);
635 : : }
636 : :
637 : 3 : bool UdpServer::broadcast_line(std::string_view line) { return broadcast(std::string(line) + "\n"); }
638 : 1 : bool UdpServer::send_to_line(ClientId client_id, std::string_view line) {
639 : 3 : return send_to(client_id, std::string(line) + "\n");
640 : : }
641 : 3 : bool UdpServer::try_broadcast_line(std::string_view line) { return try_broadcast(std::string(line) + "\n"); }
642 : 1 : bool UdpServer::try_send_to_line(ClientId client_id, std::string_view line) {
643 : 3 : return try_send_to(client_id, std::string(line) + "\n");
644 : : }
645 : :
646 : 6 : UdpServer& UdpServer::on_connect(ConnectionHandler h) {
647 : 6 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
648 : 6 : impl_->on_connect = std::move(h);
649 : 6 : return *this;
650 : 6 : }
651 : :
652 : 5 : UdpServer& UdpServer::on_disconnect(ConnectionHandler h) {
653 : 5 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
654 : 5 : impl_->on_disconnect = std::move(h);
655 : 5 : return *this;
656 : 5 : }
657 : :
658 : 8 : UdpServer& UdpServer::on_data(MessageHandler h) {
659 : 8 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
660 : 8 : impl_->on_data = interface::share_callback(std::move(h));
661 : 8 : return *this;
662 : 8 : }
663 : :
664 : 4 : UdpServer& UdpServer::on_data_batch(BatchMessageHandler h) {
665 : 4 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
666 : 4 : impl_->on_data_batch_ = interface::share_callback(std::move(h));
667 : 4 : return *this;
668 : 4 : }
669 : :
670 : 3 : UdpServer& UdpServer::on_error(ErrorHandler h) {
671 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
672 : 3 : impl_->on_error = std::move(h);
673 : 3 : return *this;
674 : 3 : }
675 : :
676 : 6 : UdpServer& UdpServer::framer(FramerFactory factory) {
677 : 6 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
678 : 6 : impl_->framer_factory = std::move(factory);
679 : 6 : return *this;
680 : 6 : }
681 : :
682 : 3 : UdpServer& UdpServer::on_message(MessageHandler h) {
683 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
684 : 3 : impl_->on_message = interface::share_callback(std::move(h));
685 : 3 : return *this;
686 : 3 : }
687 : :
688 : 3 : UdpServer& UdpServer::on_message_batch(BatchMessageHandler h) {
689 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
690 : 3 : impl_->on_message_batch_ = interface::share_callback(std::move(h));
691 : 3 : return *this;
692 : 3 : }
693 : :
694 : 17 : size_t UdpServer::client_count() const {
695 : 17 : std::shared_lock<std::shared_mutex> lock(impl_->mutex);
696 : 34 : return impl_->endpoint_to_id.size();
697 : 17 : }
698 : :
699 : 3 : std::vector<ClientId> UdpServer::connected_clients() const {
700 : 3 : std::shared_lock<std::shared_mutex> lock(impl_->mutex);
701 : 3 : std::vector<ClientId> ids;
702 : 3 : ids.reserve(impl_->sessions.size());
703 [ + + ]: 6 : for (const auto& [id, entry] : impl_->sessions) {
704 : 3 : ids.push_back(id);
705 : : }
706 : 6 : return ids;
707 : 3 : }
708 : :
709 : 1 : UdpServer& UdpServer::auto_start(bool m) {
710 : 1 : impl_->auto_start.store(m);
711 [ + - + - : 1 : if (impl_->auto_start.load() && !impl_->started.load()) {
+ - ]
712 : 1 : start();
713 : : }
714 : 1 : return *this;
715 : : }
716 : :
717 : 1 : UdpServer& UdpServer::bind_address(const std::string& address) {
718 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
719 : 1 : impl_->cfg.bind_address = address;
720 : 1 : return *this;
721 : 1 : }
722 : :
723 : 4 : UdpServer& UdpServer::idle_timeout(std::chrono::milliseconds timeout) {
724 : 4 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
725 : 4 : impl_->session_timeout = timeout;
726 [ - + ]: 4 : if (impl_->session_timeout.count() <= 0) {
727 [ # # ]: 0 : if (impl_->reaper_timer) {
728 : 0 : impl_->reaper_timer->cancel();
729 : 0 : impl_->reaper_timer.reset();
730 : : }
731 : 0 : return *this;
732 : : }
733 [ - + - - : 4 : if (impl_->started.load() && impl_->channel && !impl_->reaper_timer) {
- - - + ]
734 : 0 : impl_->reaper_timer = std::make_unique<boost::asio::steady_timer>(impl_->channel->get_executor());
735 : 0 : impl_->schedule_reaper();
736 : : }
737 : 4 : return *this;
738 : 4 : }
739 : :
740 : 2 : UdpServer& UdpServer::on_backpressure(std::function<void(size_t)> handler) {
741 : 2 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
742 : 2 : impl_->bp_handler = std::move(handler);
743 : 2 : return *this;
744 : 2 : }
745 : :
746 : 4 : UdpServer& UdpServer::max_clients(size_t max) {
747 [ + + ]: 4 : if (max == 0) {
748 : 1 : impl_->client_limit_enabled.store(false);
749 : 1 : impl_->max_clients_limit.store(0);
750 : : } else {
751 : 3 : impl_->client_limit_enabled.store(true);
752 : 3 : impl_->max_clients_limit.store(max);
753 : : }
754 : 4 : return *this;
755 : : }
756 : :
757 : 7 : UdpServer& UdpServer::backpressure_threshold(size_t threshold) {
758 : 7 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
759 : 7 : impl_->cfg.backpressure_threshold = threshold;
760 : 7 : return *this;
761 : 7 : }
762 : :
763 : 5 : UdpServer& UdpServer::backpressure_strategy(base::constants::BackpressureStrategy strategy) {
764 : 5 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
765 : 5 : impl_->cfg.backpressure_strategy = strategy;
766 [ - + ]: 5 : if (impl_->channel) {
767 : 0 : impl_->channel->set_backpressure_strategy(strategy);
768 : : }
769 : 5 : return *this;
770 : 5 : }
771 : :
772 : 3 : size_t UdpServer::backpressure_threshold() const {
773 : 3 : std::shared_lock<std::shared_mutex> lock(impl_->mutex);
774 : 6 : return impl_->cfg.backpressure_threshold;
775 : 3 : }
776 : :
777 : 3 : base::constants::BackpressureStrategy UdpServer::backpressure_strategy() const {
778 : 3 : std::shared_lock<std::shared_mutex> lock(impl_->mutex);
779 : 6 : return impl_->cfg.backpressure_strategy;
780 : 3 : }
781 : :
782 : 0 : UdpServer& UdpServer::send_buffer_size(size_t bytes) {
783 : 0 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
784 : 0 : impl_->cfg.send_buffer_size = bytes;
785 : 0 : return *this;
786 : 0 : }
787 : :
788 : 0 : UdpServer& UdpServer::receive_buffer_size(size_t bytes) {
789 : 0 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
790 : 0 : impl_->cfg.receive_buffer_size = bytes;
791 : 0 : return *this;
792 : 0 : }
793 : :
794 : 4 : UdpServer& UdpServer::manage_external_context(bool m) {
795 : 4 : impl_->manage_external_context.store(m);
796 : 4 : return *this;
797 : : }
798 : :
799 : 5 : UdpServer& UdpServer::batch_size(size_t size) {
800 : 5 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
801 : 5 : impl_->max_batch_size_ = size;
802 : 5 : return *this;
803 : 5 : }
804 : :
805 : 5 : UdpServer& UdpServer::batch_latency(std::chrono::milliseconds latency) {
806 : 5 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
807 : 5 : impl_->max_batch_latency_ = latency;
808 : 5 : return *this;
809 : 5 : }
810 : :
811 : : } // namespace wrapper
812 : : } // namespace wirestead
|
