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 <boost/asio/strand.hpp>
25 : : #include <iostream>
26 : : #include <mutex>
27 : : #include <shared_mutex>
28 : : #include <stop_token>
29 : : #include <thread>
30 : : #include <unordered_map>
31 : : #include <vector>
32 : :
33 : : #include "wirestead/base/common.hpp"
34 : : #include "wirestead/concurrency/io_thread_hook.hpp"
35 : : #include "wirestead/config/validation.hpp"
36 : : #include "wirestead/factory/channel_factory.hpp"
37 : : #include "wirestead/transport/base/stop_test_hook.hpp"
38 : : #include "wirestead/transport/udp/detail/write_wait.hpp"
39 : : #include "wirestead/transport/udp/udp.hpp"
40 : : #include "wirestead/util/input_validator.hpp"
41 : : #include "wirestead/wrapper/callback_guard.hpp"
42 : : #include "wirestead/wrapper/error_context_builder.hpp"
43 : : #include "wirestead/wrapper/send_retry.hpp"
44 : : #include "wirestead/wrapper/send_validation.hpp"
45 : : #include "wirestead/wrapper/session_batch.hpp"
46 : :
47 : : namespace wirestead {
48 : : namespace wrapper {
49 : :
50 : : namespace {
51 : 247 : std::chrono::steady_clock::time_point session_now() {
52 [ + + ]: 247 : if (auto hook = transport::detail::g_udp_session_clock_hook.load()) return hook();
53 : 173 : return std::chrono::steady_clock::now();
54 : : }
55 : : // std::hash<boost::asio::ip::udp::endpoint> is not available before Boost 1.74.
56 : : // Provide a portable hash by combining the raw address bytes and port.
57 : : struct UdpEndpointHash {
58 : 332 : std::size_t operator()(const boost::asio::ip::udp::endpoint& ep) const noexcept {
59 : 332 : std::size_t seed = 0;
60 : 1992 : auto combine = [&](std::size_t v) { seed ^= v + 0x9e3779b9u + (seed << 6) + (seed >> 2); };
61 [ + - ]: 332 : if (ep.address().is_v4()) {
62 [ + + ]: 1660 : for (auto byte : ep.address().to_v4().to_bytes()) {
63 : 1328 : combine(std::hash<unsigned char>{}(byte));
64 : : }
65 : : } else {
66 [ # # ]: 0 : for (auto byte : ep.address().to_v6().to_bytes()) {
67 : 0 : combine(std::hash<unsigned char>{}(byte));
68 : : }
69 : : }
70 : 332 : combine(std::hash<unsigned short>{}(ep.port()));
71 : 332 : return seed;
72 : : }
73 : : };
74 : : } // namespace
75 : :
76 : : struct UdpServer::Impl : public std::enable_shared_from_this<Impl> {
77 : : config::UdpConfig cfg;
78 : : std::shared_ptr<transport::UdpChannel> channel;
79 : : RuntimeStats stopped_stats_;
80 : : std::shared_ptr<boost::asio::io_context> external_ioc;
81 : : std::atomic<bool> use_external_context{false};
82 : : std::atomic<bool> manage_external_context{false};
83 : : std::jthread external_thread;
84 : : std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>> work_guard;
85 : :
86 : : mutable std::shared_mutex mutex;
87 : : std::mutex bp_mutex_;
88 : : std::condition_variable bp_cv_;
89 : : // D-1: admission gate for this object's user callbacks. Admission, the
90 : : // running count and the closed flag are one decision, so a stop() cannot
91 : : // observe an empty gate while a callback is about to start, and a callback
92 : : // left over from a previous run is refused after a restart.
93 : : detail::CallbackGate callback_gate_;
94 : : std::mutex stop_finalize_mutex_;
95 : : bool stop_requested_ = false;
96 : : std::atomic<unsigned> stop_callers_{0};
97 : : std::atomic<uint64_t> callback_generation_{0};
98 : :
99 : : // True when this thread is one the target's shutdown needs: a callback of
100 : : // this object, or any thread currently running the external io_context this
101 : : // channel was built on (which a callback of another channel sharing it is).
102 : : // Such a caller requests the shutdown and returns; it cannot wait for work
103 : : // its own thread has to perform.
104 : 547 : bool shutdown_needs_this_thread() const {
105 [ + - + + ]: 547 : if (callback_gate_.active_on_this_thread()) return true;
106 [ + + - + : 540 : if (external_ioc && external_ioc->get_executor().running_in_this_thread()) return true;
+ + - + ]
107 : 540 : std::shared_lock<std::shared_mutex> lock(mutex);
108 [ + + ]: 540 : if (!channel) return false;
109 : 371 : const auto executor = channel->get_executor();
110 : : using IoExecutor = boost::asio::io_context::executor_type;
111 [ + - - + ]: 371 : if (auto* io = executor.target<IoExecutor>()) return io->running_in_this_thread();
112 [ + - + - ]: 371 : if (auto* strand = executor.target<boost::asio::strand<IoExecutor>>())
113 : 371 : return strand->get_inner_executor().running_in_this_thread();
114 : 0 : return false;
115 : 540 : }
116 : :
117 : : std::vector<std::promise<bool>> pending_promises;
118 : : std::atomic<bool> started{false};
119 : : std::atomic<bool> is_listening{false};
120 : :
121 : : // Virtual Session Management
122 : : struct SessionEntry {
123 : : boost::asio::ip::udp::endpoint endpoint;
124 : : std::shared_ptr<transport::detail::UdpWriteWait> wait;
125 : : std::shared_ptr<framer::IFramer> framer;
126 : : std::chrono::steady_clock::time_point last_seen;
127 : : };
128 : : ClientId next_client_id{1};
129 : : std::unordered_map<boost::asio::ip::udp::endpoint, ClientId, UdpEndpointHash> endpoint_to_id;
130 : : std::unordered_map<ClientId, SessionEntry> sessions;
131 : 159 : std::chrono::milliseconds session_timeout{0}; // 0 = disabled
132 : : std::unique_ptr<boost::asio::steady_timer> reaper_timer;
133 : : std::atomic<bool> auto_start{false};
134 : : std::atomic<bool> client_limit_enabled{false};
135 : : std::atomic<size_t> max_clients_limit{0};
136 : :
137 : : ConnectionHandler on_connect{nullptr};
138 : : ConnectionHandler on_disconnect{nullptr};
139 : : ConnectionHandler on_session_expired{nullptr};
140 : : // Shared snapshots: the strand copies one out per received datagram, and a
141 : : // std::function copy allocates whenever the user handler outgrows its
142 : : // small-object buffer. See interface::SharedCallback.
143 : : interface::SharedCallback<MessageHandler> on_data;
144 : : interface::SharedCallback<BatchMessageHandler> on_data_batch_;
145 : : ErrorHandler on_error{nullptr};
146 : : std::function<void(size_t)> bp_handler{nullptr};
147 : : FramerFactory framer_factory{nullptr};
148 : : interface::SharedCallback<MessageHandler> on_message;
149 : : interface::SharedCallback<BatchMessageHandler> on_message_batch_;
150 : :
151 : : bool factory_managed_channel_ = true;
152 : 159 : std::shared_ptr<bool> is_alive{std::make_shared<bool>(true)};
153 : :
154 : : // Batching logic
155 : : bool terminal_error_reported_ = false;
156 : 65 : void require_stopped_config() const {
157 [ + + + - : 128 : if (started.load() || stop_callers_.load() || (stop_requested_ && is_alive) || detail::in_data_callback())
+ + + - -
+ + + ]
158 : 2 : throw std::logic_error("configuration requires completed stop");
159 : 63 : }
160 : : ReceiveLimits receive_limits_;
161 : : std::shared_ptr<detail::ReceiveBudget> receive_budget_{std::make_shared<detail::ReceiveBudget>(receive_limits_)};
162 : : std::unordered_map<ClientId, std::shared_ptr<detail::SessionBatch>> batches_;
163 : : size_t max_batch_size_ = 100;
164 : 159 : std::chrono::milliseconds max_batch_latency_{1};
165 : :
166 : 77 : explicit Impl(const config::UdpConfig& config) : cfg(config) {}
167 : 6 : Impl(const config::UdpConfig& config, std::shared_ptr<boost::asio::io_context> ioc)
168 [ + - + - : 6 : : cfg(config), external_ioc(std::move(ioc)), use_external_context(external_ioc != nullptr) {}
+ - ]
169 : 76 : explicit Impl(std::shared_ptr<interface::Channel> ch)
170 : 76 : : channel(std::dynamic_pointer_cast<transport::UdpChannel>(ch)), factory_managed_channel_(false) {
171 : : // #450: setup_internal_handlers() captures weak_from_this() - calling it
172 : : // from inside this constructor would capture an empty weak_ptr, since
173 : : // enable_shared_from_this isn't wired up until make_shared() finishes
174 : : // constructing the object. Deferred to UdpServer's own constructor,
175 : : // which runs after impl_ is a fully-formed shared_ptr<Impl>.
176 : 76 : }
177 : :
178 : 159 : ~Impl() {
179 : : try {
180 : 159 : stop();
181 : 0 : } catch (...) {
182 : 0 : }
183 : 159 : }
184 : :
185 : 862 : void fulfill_all_locked(bool value) {
186 [ + + ]: 1021 : for (auto& promise : pending_promises) {
187 : : try {
188 : 159 : promise.set_value(value);
189 : 0 : } catch (...) {
190 : 0 : }
191 : : }
192 : 862 : pending_promises.clear();
193 : 862 : }
194 : :
195 : 37 : void flush_batches(uint64_t generation, ClientId id, const std::shared_ptr<detail::SessionBatch>& state) {
196 : 37 : auto lease = callback_gate_.enter(generation);
197 [ + + ]: 37 : if (!lease.admitted()) return;
198 : 35 : std::unique_lock<std::shared_mutex> lock(mutex);
199 : 35 : auto found = batches_.find(id);
200 [ + - - + : 35 : if (found == batches_.end() || found->second != state) return;
- + ]
201 : 35 : state->scheduled = false;
202 : 35 : auto data = std::move(state->data);
203 : 35 : auto messages = std::move(state->messages);
204 : 35 : state->data.clear();
205 : 35 : state->messages.clear();
206 : 35 : auto data_handler = on_data_batch_;
207 : 35 : auto message_handler = on_message_batch_;
208 : 35 : lock.unlock();
209 [ + + + - ]: 35 : if (!data.empty()) detail::invoke_user_callback("udp_server", "on_data_batch", data_handler, data);
210 : : // A callback may stop the server. Recheck admission before another callback.
211 : 35 : auto next = callback_gate_.enter(generation);
212 [ + + + + : 35 : if (next.admitted() && !messages.empty())
+ + ]
213 : 5 : detail::invoke_user_callback("udp_server", "on_message_batch", message_handler, messages);
214 : 37 : }
215 : :
216 : : // Caller holds mutex; do not postpone an existing deadline when the
217 : : // other queue receives its first item.
218 : 20 : void schedule_batch_timer(uint64_t generation, ClientId id) {
219 : 20 : auto found = batches_.find(id);
220 [ + - + + : 20 : if (found == batches_.end() || found->second->scheduled) return;
+ + ]
221 : 19 : auto state = found->second;
222 : 19 : state->scheduled = true;
223 : 19 : state->timer.expires_after(max_batch_latency_);
224 : 19 : state->timer.async_wait(
225 : 19 : [weak_impl = weak_from_this(), generation, id,
226 : : weak_state = std::weak_ptr<detail::SessionBatch>(state)](const boost::system::error_code& ec) {
227 [ + + ]: 19 : if (ec) return;
228 : 13 : auto self = weak_impl.lock();
229 : 13 : auto batch = weak_state.lock();
230 [ + - + - : 13 : if (self && batch) self->flush_batches(generation, id, batch);
+ - + - ]
231 : 13 : });
232 : 19 : }
233 : :
234 : 57 : void schedule_reaper(uint64_t generation) {
235 [ + + + - : 57 : if (!started.load() || !reaper_timer || session_timeout.count() <= 0) return;
- + + + ]
236 : :
237 : : // Run reaper at interval proportional to timeout (min 100ms, max 5s)
238 : : auto interval =
239 : 56 : std::max(std::chrono::milliseconds(100), std::min(std::chrono::milliseconds(5000), session_timeout / 2));
240 : :
241 : 56 : reaper_timer->expires_after(interval);
242 : 112 : reaper_timer->async_wait([this, generation, weak_impl = weak_from_this(),
243 : 56 : alive = std::weak_ptr<bool>(is_alive)](const boost::system::error_code& ec) {
244 : 48 : auto impl_keepalive = weak_impl.lock();
245 [ - + ]: 48 : if (!impl_keepalive) return;
246 : 48 : auto lock = alive.lock();
247 [ + + - + : 48 : if (!lock || !(*lock)) return;
+ + ]
248 : :
249 [ + - ]: 39 : if (!ec) {
250 : 39 : auto lease = callback_gate_.enter(generation);
251 [ + + ]: 39 : if (!lease.admitted()) return;
252 : 30 : run_reaper();
253 : 30 : std::unique_lock<std::shared_mutex> guard(mutex);
254 : 30 : schedule_reaper(generation);
255 : 39 : }
256 : 66 : });
257 : : }
258 : :
259 : 30 : void run_reaper() {
260 : 30 : const auto generation = callback_generation_.load();
261 : 30 : const auto now = session_now();
262 : 30 : std::vector<std::pair<ClientId, std::shared_ptr<detail::SessionBatch>>> due;
263 : : {
264 : 30 : std::shared_lock<std::shared_mutex> lock(mutex);
265 [ - + ]: 30 : if (session_timeout.count() <= 0) return;
266 [ + + ]: 68 : for (const auto& [id, session] : sessions) {
267 [ + - + - : 38 : if (now - session.last_seen > session_timeout) {
+ + ]
268 : 24 : auto batch = batches_.find(id);
269 [ - + + - ]: 24 : due.emplace_back(id, batch == batches_.end() ? nullptr : batch->second);
270 : : }
271 : : }
272 : 30 : }
273 [ + + ]: 54 : for (const auto& [id, batch] : due) {
274 [ + - + - ]: 24 : if (batch) flush_batches(generation, id, batch);
275 : 24 : ConnectionHandler handler;
276 : 24 : std::string info;
277 : : {
278 : 24 : std::unique_lock<std::shared_mutex> lock(mutex);
279 [ + - - + : 24 : if (!started || callback_generation_ != generation) return;
- + ]
280 : 24 : auto it = sessions.find(id);
281 [ + - + - : 24 : if (it == sessions.end() || session_timeout.count() <= 0 || now - it->second.last_seen <= session_timeout)
+ - + - -
+ - + ]
282 : 0 : continue;
283 : 24 : info = fmt::format("{}:{}", it->second.endpoint.address().to_string(), it->second.endpoint.port());
284 [ + - + - ]: 24 : if (channel) channel->expire_session(it->second.wait);
285 : 24 : bp_cv_.notify_all();
286 : 24 : endpoint_to_id.erase(it->second.endpoint);
287 : 24 : batches_.erase(id);
288 : 24 : sessions.erase(it);
289 : 24 : handler = on_session_expired;
290 : 24 : }
291 : 24 : auto lease = callback_gate_.enter(generation);
292 [ + - ]: 24 : if (lease.admitted())
293 : 24 : detail::invoke_user_callback("udp_server", "on_session_expired", handler, ConnectionContext(id, info));
294 : 24 : }
295 : 30 : }
296 : :
297 : 235 : void setup_internal_handlers() {
298 [ - + ]: 235 : if (!channel) return;
299 : :
300 : 235 : std::weak_ptr<Impl> weak_impl = weak_from_this();
301 : 235 : receive_budget_->reset_stats();
302 : 235 : terminal_error_reported_ = false;
303 : 235 : const auto generation = callback_gate_.open_new_generation();
304 : 235 : callback_generation_.store(generation);
305 : :
306 : 470 : channel->on_bytes_from(
307 : 470 : [this, generation, weak_impl](memory::ConstByteSpan data, const boost::asio::ip::udp::endpoint& ep) {
308 : 219 : auto impl_keepalive = weak_impl.lock();
309 [ - + ]: 219 : if (!impl_keepalive) return;
310 : 219 : auto lease = callback_gate_.enter(generation);
311 [ - + ]: 219 : if (!lease.admitted()) return;
312 : 219 : detail::CallbackGuard callback_guard;
313 : 219 : ClientId client_id = 0;
314 : 219 : bool is_new = false, announced = false, batch_mode = false;
315 : 219 : std::shared_ptr<detail::SessionBatch> receive;
316 : 219 : std::shared_ptr<framer::IFramer> target_framer;
317 : 219 : ConnectionHandler connect_handler_copy;
318 : 219 : interface::SharedCallback<MessageHandler> data_handler_copy;
319 : : try {
320 : : {
321 : 219 : std::unique_lock<std::shared_mutex> lock(mutex);
322 [ + - - + : 219 : if (!started || generation != callback_generation_) return;
- + ]
323 : 219 : auto it = endpoint_to_id.find(ep);
324 [ + + ]: 219 : if (it == endpoint_to_id.end()) {
325 [ + + + + : 199 : if (client_limit_enabled && sessions.size() >= max_clients_limit) return;
+ + ]
326 : 197 : auto scope = receive_budget_->open_scope();
327 [ + + ]: 197 : if (!scope) return;
328 : 196 : client_id = next_client_id++;
329 : 196 : is_new = true;
330 : 196 : receive = std::make_shared<detail::SessionBatch>(channel->get_executor(), std::move(scope));
331 : 196 : endpoint_to_id.emplace(ep, client_id);
332 : 196 : batches_.emplace(client_id, receive);
333 : 196 : SessionEntry entry;
334 : 196 : entry.endpoint = ep;
335 : 196 : entry.wait = channel->capture_write_wait(false, true);
336 : 196 : entry.last_seen = session_now();
337 [ + + ]: 196 : if (framer_factory) {
338 : 38 : auto framer = framer_factory();
339 [ + - ]: 38 : if (framer) {
340 : 38 : framer->on_message([this, client_id, generation](memory::ConstByteSpan msg) {
341 : 49 : auto message_lease = callback_gate_.enter(generation);
342 [ - + ]: 49 : if (!message_lease.admitted()) return;
343 : 49 : auto prepared = detail::take_prepared_message(client_id, msg);
344 : : // #441: snapshot under a shared_lock (pure read), build the
345 : : // copy before taking the exclusive lock for queue mutation.
346 : : bool batch_mode;
347 : 49 : interface::SharedCallback<MessageHandler> on_message_handler;
348 : : {
349 : 49 : std::shared_lock<std::shared_mutex> lock(mutex);
350 : 49 : batch_mode = static_cast<bool>(on_message_batch_);
351 : 49 : on_message_handler = on_message;
352 : 49 : }
353 : :
354 [ + + ]: 49 : if (batch_mode) {
355 : 14 : std::shared_ptr<detail::SessionBatch> receive;
356 : : {
357 : 14 : std::shared_lock<std::shared_mutex> lock(mutex);
358 : 14 : auto state = batches_.find(client_id);
359 [ - + ]: 14 : if (state == batches_.end()) return;
360 : 14 : receive = state->second;
361 : 14 : }
362 [ + - ]: 14 : auto ctx = prepared ? std::move(*prepared)
363 : 14 : : detail::retain_received(receive->receive.scope, client_id, msg);
364 : 14 : interface::SharedCallback<BatchMessageHandler> flush_handler;
365 : 14 : detail::ReceiveBatch batch;
366 : : {
367 : 14 : std::unique_lock<std::shared_mutex> lock(mutex);
368 : 14 : auto state = batches_.find(client_id);
369 [ + - - + : 14 : if (state == batches_.end() || callback_generation_ != generation) return;
- + ]
370 : 14 : state->second->messages.emplace_back(std::move(ctx));
371 [ + - + + ]: 14 : if (batches_.at(client_id)->messages.size() >= max_batch_size_) {
372 : 4 : flush_handler = on_message_batch_;
373 : 4 : batch = std::move(batches_.at(client_id)->messages);
374 : 4 : batches_.at(client_id)->messages.clear();
375 [ + - + + ]: 10 : } else if (batches_.at(client_id)->messages.size() == 1) {
376 : 9 : schedule_batch_timer(generation, client_id);
377 : : }
378 : 14 : }
379 : 14 : detail::invoke_user_callback("udp_server", "on_message_batch", flush_handler, batch);
380 : 14 : return;
381 : 14 : }
382 : :
383 : 35 : detail::invoke_user_callback("udp_server", "on_message", on_message_handler,
384 : 70 : MessageContext(client_id, msg));
385 : 77 : });
386 : :
387 : 38 : entry.framer = std::move(framer);
388 : : }
389 : 38 : }
390 : 196 : sessions.emplace(client_id, std::move(entry));
391 : 197 : } else {
392 : 21 : client_id = it->second;
393 : 21 : receive = batches_.at(client_id);
394 : 21 : sessions.at(client_id).last_seen = session_now();
395 : : }
396 : 217 : auto& session = sessions.at(client_id);
397 : 217 : target_framer = session.framer;
398 [ + - + - : 217 : if (auto wait = session.wait; wait && wait->stats) wait->stats->counters.record_received(data.size());
+ - + - ]
399 : 217 : connect_handler_copy = on_connect;
400 : 217 : batch_mode = static_cast<bool>(on_data_batch_);
401 : 217 : data_handler_copy = on_data;
402 : 219 : }
403 : 217 : auto prepared = detail::prepare_receive(receive->receive, target_framer, client_id, data, batch_mode);
404 [ + + ]: 215 : if (is_new) {
405 : 196 : announced = true;
406 : 196 : detail::invoke_user_callback(
407 : : "udp_server", "on_connect", connect_handler_copy,
408 : 392 : ConnectionContext(client_id, fmt::format("{}:{}", ep.address().to_string(), ep.port())));
409 : : }
410 : 215 : auto after_connect = callback_gate_.enter(generation);
411 [ - + ]: 215 : if (!after_connect.admitted()) return;
412 [ + + ]: 215 : if (batch_mode) {
413 : 17 : auto ctx = std::move(*prepared.raw);
414 : 17 : interface::SharedCallback<BatchMessageHandler> flush_handler;
415 : 17 : detail::ReceiveBatch batch;
416 : : {
417 : 17 : std::unique_lock<std::shared_mutex> lock(mutex);
418 : 17 : auto state = batches_.find(client_id);
419 [ + - - + : 17 : if (state == batches_.end() || callback_generation_ != generation) return;
- + ]
420 : 17 : state->second->data.emplace_back(std::move(ctx));
421 [ + + ]: 17 : if (state->second->data.size() >= max_batch_size_) {
422 : 6 : flush_handler = on_data_batch_;
423 : 6 : batch = std::move(state->second->data);
424 : 6 : state->second->data.clear();
425 [ + - ]: 11 : } else if (state->second->data.size() == 1) {
426 : 11 : schedule_batch_timer(generation, client_id);
427 : : }
428 : 17 : }
429 : 17 : detail::invoke_user_callback("udp_server", "on_data_batch", flush_handler, batch);
430 : 17 : } else {
431 : 396 : detail::invoke_user_callback("udp_server", "on_data", data_handler_copy, MessageContext(client_id, data));
432 : : }
433 : 215 : auto after_data = callback_gate_.enter(generation);
434 [ + + + - ]: 215 : if (after_data.admitted()) prepared.deliver();
435 : 217 : } catch (const detail::ReceiveOverflow& overflow) {
436 : 2 : receive_budget_->overflow(overflow.reason, data.size());
437 [ - + - - ]: 2 : if (is_new && !announced) {
438 : 0 : std::unique_lock<std::shared_mutex> lock(mutex);
439 : 0 : endpoint_to_id.erase(ep);
440 : 0 : sessions.erase(client_id);
441 : 0 : batches_.erase(client_id);
442 : 0 : }
443 : 2 : } catch (const std::bad_alloc&) {
444 : 0 : receive_budget_->overflow(ReceiveOverflowReason::AllocationFailure, data.size());
445 [ - - - - ]: 0 : if (is_new && !announced) {
446 : 0 : std::unique_lock<std::shared_mutex> lock(mutex);
447 : 0 : endpoint_to_id.erase(ep);
448 : 0 : sessions.erase(client_id);
449 : 0 : batches_.erase(client_id);
450 : 0 : }
451 : 0 : }
452 : 231 : });
453 : :
454 : 235 : channel->on_backpressure([this, generation, weak_impl](size_t queued) {
455 : 130 : auto impl_keepalive = weak_impl.lock();
456 [ - + ]: 130 : if (!impl_keepalive) return;
457 : 130 : auto lease = callback_gate_.enter(generation);
458 [ + + ]: 130 : if (!lease.admitted()) return;
459 : 124 : bp_cv_.notify_all();
460 : 124 : std::function<void(size_t)> handler;
461 : : {
462 : 124 : std::shared_lock<std::shared_mutex> lock(mutex);
463 : 124 : handler = bp_handler;
464 : 124 : }
465 : 124 : detail::invoke_user_callback("udp_server", "on_backpressure", handler, queued);
466 : 136 : });
467 : :
468 : 235 : channel->on_state([this, generation, weak_impl](base::LinkState state) {
469 : 624 : auto impl_keepalive = weak_impl.lock();
470 [ - + ]: 624 : if (!impl_keepalive) return;
471 : 624 : auto lease = callback_gate_.enter(generation);
472 [ + + ]: 624 : if (!lease.admitted()) return;
473 : 473 : ErrorHandler error_handler_copy{nullptr};
474 [ + + + + ]: 473 : if (state == base::LinkState::Listening || state == base::LinkState::Connected) {
475 : 307 : is_listening.store(true);
476 : 307 : std::unique_lock<std::shared_mutex> lock(mutex);
477 : 307 : terminal_error_reported_ = false;
478 : 307 : fulfill_all_locked(true);
479 [ + + + + : 473 : } else if (state == base::LinkState::Error || state == base::LinkState::Closed ||
- + ]
480 : : state == base::LinkState::Idle) {
481 : 8 : is_listening.store(false);
482 : 8 : std::unique_lock<std::shared_mutex> lock(mutex);
483 : 8 : fulfill_all_locked(false);
484 [ + + + - ]: 8 : if (state == base::LinkState::Error && !terminal_error_reported_) {
485 : 7 : terminal_error_reported_ = true;
486 : 7 : error_handler_copy = on_error;
487 : : }
488 : 8 : }
489 : :
490 [ + - ]: 473 : detail::invoke_user_callback("udp_server", "on_error", error_handler_copy,
491 [ + - + - : 946 : channel ? detail::build_error_context(*channel, "Server error")
- - ]
492 : : : ErrorContext(ErrorCode::IoError, "Server error"));
493 : 775 : });
494 : 235 : }
495 : :
496 : 159 : std::future<bool> start() {
497 : 159 : std::unique_lock<std::shared_mutex> lock(mutex);
498 : 159 : stop_requested_ = false;
499 [ - + ]: 159 : if (is_listening.load()) {
500 : 0 : std::promise<bool> p;
501 : 0 : p.set_value(true);
502 : 0 : return p.get_future();
503 : 0 : }
504 : :
505 : 159 : std::promise<bool> p;
506 : 159 : auto fut = p.get_future();
507 : 159 : pending_promises.emplace_back(std::move(p));
508 : :
509 [ - + ]: 159 : if (started.exchange(true)) {
510 : 0 : return fut;
511 : : }
512 : :
513 [ + + + - ]: 159 : if (!is_alive) is_alive = std::make_shared<bool>(true);
514 [ + + ]: 159 : if (!channel) {
515 : 75 : channel = std::dynamic_pointer_cast<transport::UdpChannel>(factory::ChannelFactory::create(cfg, external_ioc));
516 : : }
517 : 159 : channel->reset_stats();
518 : 159 : stopped_stats_ = {};
519 : 159 : setup_internal_handlers();
520 : :
521 : 159 : auto channel_copy = channel;
522 : 159 : lock.unlock();
523 : 159 : channel_copy->start();
524 : :
525 : 159 : lock.lock();
526 [ - + ]: 159 : if (!started.load()) return fut;
527 [ + + + + : 159 : if (use_external_context.load() && manage_external_context.load() && !external_thread.joinable()) {
+ - + + ]
528 [ + - - + : 1 : if (external_ioc->stopped()) external_ioc->restart();
- - ]
529 : 2 : work_guard = std::make_unique<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>(
530 : 3 : external_ioc->get_executor());
531 : 2 : external_thread = std::jthread([ioc = external_ioc](std::stop_token st) {
532 : 1 : wirestead::concurrency::run_io_thread_init();
533 : : try {
534 : 1 : std::stop_callback cb(st, [ioc] { ioc->stop(); });
535 : 1 : ioc->run();
536 : 1 : } catch (...) {
537 : 0 : }
538 : 2 : });
539 : : }
540 : :
541 [ + - + + : 159 : if (channel && session_timeout.count() > 0) {
+ + ]
542 : 9 : reaper_timer = std::make_unique<boost::asio::steady_timer>(channel->get_executor());
543 : 18 : schedule_reaper(callback_generation_.load());
544 : : }
545 : :
546 : 159 : return fut;
547 : 159 : }
548 : :
549 : 547 : void stop() {
550 : 547 : stop_callers_.fetch_add(1);
551 : : struct StopCall {
552 : : std::atomic<unsigned>& callers;
553 : 547 : ~StopCall() { callers.fetch_sub(1); }
554 : 547 : } stop_call{stop_callers_};
555 : 547 : const bool request_only = shutdown_needs_this_thread();
556 : 547 : callback_gate_.close();
557 : 547 : std::shared_ptr<transport::UdpChannel> channel_copy;
558 : : {
559 : 547 : std::unique_lock<std::shared_mutex> lock(mutex);
560 : 547 : stop_requested_ = true;
561 [ + + + - ]: 547 : if (channel) channel->cancel_write_waits();
562 : 547 : started.store(false);
563 : 547 : is_listening.store(false);
564 : 547 : bp_cv_.notify_all();
565 : 547 : fulfill_all_locked(false);
566 : 547 : channel_copy = channel;
567 : 547 : }
568 [ + + + - ]: 547 : if (channel_copy) channel_copy->stop();
569 [ + + ]: 547 : if (request_only) return;
570 : :
571 : 535 : callback_gate_.wait_until_idle();
572 : 535 : std::lock_guard<std::mutex> finalize_lock(stop_finalize_mutex_);
573 : : {
574 : 535 : std::unique_lock<std::shared_mutex> lock(mutex);
575 : 535 : is_alive.reset();
576 : 535 : batches_.clear();
577 [ + + ]: 535 : if (reaper_timer) {
578 : 27 : reaper_timer->cancel();
579 : 27 : reaper_timer.reset();
580 : : }
581 [ + + ]: 535 : if (channel) {
582 : 366 : channel->on_bytes_from(nullptr);
583 : 366 : channel->on_state(nullptr);
584 : 366 : channel->on_backpressure(nullptr);
585 : : }
586 [ + + + - ]: 535 : if (channel) stopped_stats_ = channel->stats();
587 [ + + ]: 535 : if (factory_managed_channel_) channel.reset();
588 : 535 : endpoint_to_id.clear();
589 : 535 : sessions.clear();
590 : 535 : next_client_id = 1;
591 : 535 : }
592 [ + + + + : 535 : if (use_external_context && manage_external_context) {
+ + ]
593 : 7 : work_guard.reset();
594 [ + - + - ]: 7 : if (external_ioc) external_ioc->stop();
595 [ + + + - ]: 7 : if (external_thread.joinable()) external_thread.join();
596 : : }
597 : 559 : }
598 : :
599 : 124 : SendResult send_to(ClientId client_id, std::string_view data) {
600 [ + + ]: 124 : if (cfg.backpressure_strategy == base::constants::BackpressureStrategy::Reliable)
601 : 111 : return send_to_blocking(client_id, data);
602 : 13 : return try_send_to(client_id, data, true);
603 : : }
604 : :
605 : 80 : FanoutResult try_broadcast(std::string_view data, bool append_newline = false) {
606 : : // The shared wrapper lock fixes the virtual-session set for the whole
607 : : // nonblocking traversal, including each endpoint's admission decision.
608 : 80 : std::shared_lock<std::shared_mutex> lock(mutex);
609 : 80 : FanoutResult result;
610 : : const auto size =
611 : : append_newline
612 [ + + + + ]: 80 : ? (data.size() >= base::constants::MAX_BUFFER_SIZE ? base::constants::MAX_BUFFER_SIZE + 1 : data.size() + 1)
613 : 60 : : data.size();
614 [ + + + - ]: 80 : const auto validation = detail::validate_payload_size(size, channel ? channel->write_queue_limit() : std::nullopt);
615 [ + + + - ]: 80 : const auto state = validation.accepted() ? send_state() : validation;
616 : 80 : std::string line;
617 [ + + + + : 80 : if (append_newline && state.accepted() && !sessions.empty()) {
+ - + + ]
618 : 14 : line = std::string(data) + "\n";
619 : 14 : data = line;
620 : : }
621 : 80 : const auto bytes = base::safe_convert::string_to_bytes(data);
622 [ + + ]: 212 : for (const auto& [id, entry] : sessions) {
623 [ + + ]: 252 : result.add(state.accepted()
624 [ + - + - : 492 : ? channel->try_write_to({bytes.first, bytes.second}, entry.endpoint,
+ + - - ]
625 : 240 : entry.wait ? std::optional<uint64_t>(entry.wait->sequence) : std::nullopt,
626 : 120 : entry.wait)
627 : : : state);
628 : : }
629 : 160 : return result;
630 : 80 : }
631 : :
632 : 49 : FanoutResult broadcast(std::string_view data) { return try_broadcast(data); }
633 : :
634 : 1377 : static SendResult finish_send(SendResult result) {
635 [ + + ]: 1377 : if (auto hook = detail::g_udp_server_send_result_hook.load()) hook(result);
636 : 1377 : return result;
637 : : }
638 : :
639 : : // Caller holds mutex. The native call rechecks socket readiness at admission.
640 : 2623 : SendResult send_state() {
641 [ - + ]: 5246 : if (stop_callers_.load()) return SendResult::reject(SendRejection::Stopping);
642 [ + + ]: 2623 : if (!started.load()) {
643 [ + + ]: 21 : if (stop_requested_) {
644 [ - + ]: 4 : if (!callback_gate_.idle()) return SendResult::reject(SendRejection::Stopping);
645 [ + + ]: 4 : if (channel) {
646 : 3 : const auto state = channel->write_state(false);
647 [ + - - + : 3 : if (!state.accepted() && state.reason() == SendRejection::Stopping) return state;
- + ]
648 : : }
649 : : }
650 : 21 : return SendResult::reject(SendRejection::NotStarted);
651 : : }
652 [ - + ]: 2602 : if (!channel) return SendResult::reject(SendRejection::NotReady);
653 : 2602 : return channel->write_state(false);
654 : : }
655 : :
656 : 103 : SendResult try_send_to(ClientId client_id, std::string_view data, bool best_effort_send = false) {
657 : 0 : const auto result = [&]() -> SendResult {
658 : 103 : std::shared_lock<std::shared_mutex> lock(mutex);
659 : : auto validation =
660 [ + + + - ]: 103 : detail::validate_payload_size(data.size(), channel ? channel->write_queue_limit() : std::nullopt);
661 [ + + ]: 103 : if (!validation.accepted()) return validation;
662 : 97 : auto state = send_state();
663 [ + + ]: 97 : if (!state.accepted()) return state;
664 : 89 : auto it = sessions.find(client_id);
665 [ + + ]: 89 : if (it == sessions.end()) return SendResult::reject(SendRejection::NotReady);
666 : 85 : auto bytes = base::safe_convert::string_to_bytes(data);
667 [ + - + - ]: 255 : auto admitted = channel->try_write_to(
668 : 85 : {bytes.first, bytes.second}, it->second.endpoint,
669 : 170 : it->second.wait ? std::optional<uint64_t>(it->second.wait->sequence) : std::nullopt, it->second.wait);
670 [ + + + + : 85 : if (best_effort_send && !admitted.accepted() && admitted.reason() == SendRejection::WouldBlock)
+ - + + ]
671 : 5 : return SendResult::reject(SendRejection::QueueFull);
672 : 80 : return admitted;
673 : 206 : }();
674 : 206 : return finish_send(result);
675 : : }
676 : :
677 : 1274 : SendResult send_to_blocking(ClientId client_id, std::string_view data) {
678 : 0 : const auto result = [&]() -> SendResult {
679 : 1274 : std::shared_ptr<transport::UdpChannel> native;
680 : 1274 : std::shared_ptr<transport::detail::UdpWriteWait> wait;
681 : : uint64_t generation;
682 : 1274 : bool cannot_wait = false;
683 : : {
684 : 1274 : std::shared_lock<std::shared_mutex> lock(mutex);
685 : : auto validation =
686 [ + + + - ]: 1274 : detail::validate_payload_size(data.size(), channel ? channel->write_queue_limit() : std::nullopt);
687 [ + + ]: 1274 : if (!validation.accepted()) return validation;
688 : 1262 : auto state = send_state();
689 [ + + ]: 1262 : if (!state.accepted()) return state;
690 : 1254 : auto it = sessions.find(client_id);
691 [ + + - + : 1254 : if (it == sessions.end() || !it->second.wait) return SendResult::reject(SendRejection::NotReady);
+ + ]
692 : 1228 : native = channel;
693 [ + + + - : 1228 : cannot_wait = detail::in_data_callback() || detail::executor_running_here(native->get_executor());
+ - + + +
+ - - ]
694 : 1228 : wait = it->second.wait;
695 : 1228 : generation = callback_generation_.load();
696 : 1274 : }
697 : 1228 : for (bool retry = false;; retry = true) {
698 [ + + + - ]: 1266 : if (retry) detail::pause_send_retry(bp_cv_, bp_mutex_);
699 : 1266 : std::unique_lock<std::mutex> bp_lock(bp_mutex_);
700 : 1266 : auto outcome = native->poll_write_wait(wait);
701 [ + + ]: 1266 : if (!outcome) {
702 [ + + ]: 84 : if (cannot_wait) return SendResult::reject(SendRejection::WouldBlock);
703 [ + + + - ]: 17 : if (auto hook = detail::g_udp_capacity_wait_hook.load()) hook();
704 [ + - + + ]: 42 : while (!bp_cv_.wait_for(bp_lock, std::chrono::milliseconds(50), [&] {
705 : 80 : outcome = native->poll_write_wait(wait);
706 : 80 : return outcome.has_value();
707 : : })) {
708 : : }
709 [ + + + - ]: 17 : if (auto hook = detail::g_udp_capacity_wait_result_hook.load()) hook(*outcome);
710 : : }
711 [ + + ]: 1199 : if (!outcome->accepted()) return *outcome;
712 : 1192 : bp_lock.unlock();
713 : 1192 : std::shared_lock<std::shared_mutex> lock(mutex);
714 : 1192 : auto state = send_state();
715 [ - + ]: 1192 : if (!state.accepted()) return state;
716 : 1192 : auto it = sessions.find(client_id);
717 [ + - + - : 2384 : if (callback_generation_.load() != generation || it == sessions.end() || it->second.wait != wait)
- + - + ]
718 : 0 : return SendResult::reject(SendRejection::NotReady);
719 : 1192 : auto bytes = base::safe_convert::string_to_bytes(data);
720 : 1192 : const auto admitted = native->write_to({bytes.first, bytes.second}, it->second.endpoint, wait->sequence, wait);
721 [ + + + + : 1192 : if (admitted.accepted() || admitted.reason() != SendRejection::WouldBlock) return admitted;
+ + ]
722 [ + + ]: 39 : if (cannot_wait) return admitted;
723 : 2458 : }
724 : 2548 : }();
725 : 2548 : return finish_send(result);
726 : : }
727 : :
728 : 311 : RuntimeStats stats() const {
729 : 311 : std::shared_lock<std::shared_mutex> lock(mutex);
730 [ + + + - ]: 622 : return channel ? channel->stats() : stopped_stats_;
731 : 311 : }
732 : :
733 : 287 : std::optional<RuntimeStats> client_stats(ClientId id) const {
734 : 287 : std::shared_lock<std::shared_mutex> lock(mutex);
735 : 287 : auto it = sessions.find(id);
736 [ + + + - : 287 : if (!started.load() || !channel || it == sessions.end() || !it->second.wait) return std::nullopt;
+ + - + +
+ ]
737 : 268 : return channel->session_stats(it->second.wait);
738 : 287 : }
739 : :
740 : 16 : void reset_stats() {
741 : 16 : std::unique_lock<std::shared_mutex> lock(mutex);
742 : 16 : receive_budget_->reset_stats();
743 [ + + + - ]: 16 : if (channel) channel->reset_stats();
744 : 16 : const bool supported = stopped_stats_.send_accounting.has_value();
745 : 16 : stopped_stats_ = {};
746 [ + + ]: 16 : if (supported) stopped_stats_.send_accounting.emplace();
747 : 16 : }
748 : : };
749 : :
750 : 7 : UdpServer::UdpServer(uint16_t port) {
751 : 7 : config::UdpConfig cfg;
752 : 7 : cfg.local_port = port;
753 : 7 : impl_ = std::make_shared<Impl>(cfg);
754 : 7 : }
755 : :
756 : 70 : UdpServer::UdpServer(const config::UdpConfig& cfg) : impl_(std::make_shared<Impl>(cfg)) {
757 : 70 : config::detail::validate(cfg);
758 : 70 : }
759 : :
760 : 6 : UdpServer::UdpServer(const config::UdpConfig& cfg, std::shared_ptr<boost::asio::io_context> ioc)
761 : 6 : : impl_(std::make_shared<Impl>(cfg, ioc)) {
762 : 6 : config::detail::validate(cfg);
763 : 6 : }
764 : :
765 : 76 : UdpServer::UdpServer(std::shared_ptr<interface::Channel> ch) : impl_(std::make_shared<Impl>(std::move(ch))) {
766 : 76 : impl_->setup_internal_handlers();
767 : 76 : }
768 : :
769 : : // Stop on the destroying thread first. A callback in flight holds the Impl
770 : : // alive, and if its reference were the last one, ~Impl would run on the io
771 : : // thread, where the shutdown cannot join that thread (#613).
772 : 223 : UdpServer::~UdpServer() {
773 [ + + ]: 160 : if (!impl_) return;
774 : : try {
775 : 159 : impl_->stop();
776 : 0 : } catch (...) {
777 : 0 : }
778 : 224 : }
779 : :
780 : 1 : UdpServer::UdpServer(UdpServer&&) noexcept = default;
781 : : // The previous Impl is handed to a temporary, whose destructor stops it here.
782 : 1 : UdpServer& UdpServer::operator=(UdpServer&& other) noexcept {
783 [ + - ]: 1 : if (this != &other) {
784 : 1 : UdpServer previous(std::move(*this));
785 : 1 : impl_ = std::move(other.impl_);
786 : 1 : }
787 : 1 : return *this;
788 : : }
789 : :
790 : 9 : UdpServer& UdpServer::receive_limits(ReceiveLimits limits) {
791 : 9 : limits.validate();
792 : 8 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
793 [ + + + - : 22 : if (impl_->started || impl_->stop_callers_.load() || (impl_->stop_requested_ && impl_->is_alive) ||
+ + + - -
+ + + ]
794 : 7 : detail::in_data_callback())
795 : 1 : throw std::logic_error("receive limits require completed stop");
796 : 7 : auto budget = std::make_shared<detail::ReceiveBudget>(limits);
797 : 7 : impl_->receive_limits_ = limits;
798 : 7 : impl_->receive_budget_ = std::move(budget);
799 : 7 : return *this;
800 : 8 : }
801 : 31 : ReceiveMemoryStats UdpServer::receive_stats() const {
802 : 31 : std::shared_lock<std::shared_mutex> lock(impl_->mutex);
803 : 62 : return impl_->receive_budget_->stats();
804 : 31 : }
805 : :
806 : 159 : std::future<bool> UdpServer::start() { return impl_->start(); }
807 : 229 : void UdpServer::stop() { impl_->stop(); }
808 : 54 : bool UdpServer::listening() const { return impl_->is_listening.load(); }
809 : 311 : RuntimeStats UdpServer::stats() const { return impl_->stats(); }
810 : 287 : std::optional<RuntimeStats> UdpServer::client_stats(ClientId id) const { return impl_->client_stats(id); }
811 : 16 : void UdpServer::reset_stats() { impl_->reset_stats(); }
812 : :
813 : 49 : FanoutResult UdpServer::broadcast(std::string_view data) { return impl_->broadcast(data); }
814 : 11 : FanoutResult UdpServer::try_broadcast(std::string_view data) { return impl_->try_broadcast(data); }
815 : 124 : SendResult UdpServer::send_to(ClientId client_id, std::string_view data) { return impl_->send_to(client_id, data); }
816 : 90 : SendResult UdpServer::try_send_to(ClientId client_id, std::string_view data) {
817 : 90 : return impl_->try_send_to(client_id, data);
818 : : }
819 : :
820 : 1163 : SendResult UdpServer::send_to_blocking(ClientId client_id, std::string_view data) {
821 : 1163 : return impl_->send_to_blocking(client_id, data);
822 : : }
823 : :
824 : 10 : FanoutResult UdpServer::broadcast_line(std::string_view line) { return impl_->try_broadcast(line, true); }
825 : 56 : SendResult UdpServer::send_to_line(ClientId client_id, std::string_view line) {
826 : 168 : return send_to(client_id, std::string(line) + "\n");
827 : : }
828 : 10 : FanoutResult UdpServer::try_broadcast_line(std::string_view line) { return impl_->try_broadcast(line, true); }
829 : 6 : SendResult UdpServer::try_send_to_line(ClientId client_id, std::string_view line) {
830 : 18 : return try_send_to(client_id, std::string(line) + "\n");
831 : : }
832 : :
833 : 95 : UdpServer& UdpServer::on_connect(ConnectionHandler h) {
834 : 95 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
835 : 95 : impl_->on_connect = std::move(h);
836 : 95 : return *this;
837 : 95 : }
838 : :
839 : 6 : UdpServer& UdpServer::on_disconnect(ConnectionHandler h) {
840 : 6 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
841 : 6 : impl_->on_disconnect = std::move(h);
842 : 6 : return *this;
843 : 6 : }
844 : 13 : UdpServer& UdpServer::on_session_expired(ConnectionHandler h) {
845 : 13 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
846 : 13 : impl_->on_session_expired = std::move(h);
847 : 13 : return *this;
848 : 13 : }
849 : :
850 : 27 : UdpServer& UdpServer::on_data(MessageHandler h) {
851 : 27 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
852 : 27 : impl_->on_data = interface::share_callback(std::move(h));
853 : 27 : return *this;
854 : 27 : }
855 : :
856 : 14 : UdpServer& UdpServer::on_data_batch(BatchMessageHandler h) {
857 : 14 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
858 : 14 : impl_->on_data_batch_ = interface::share_callback(std::move(h));
859 : 14 : return *this;
860 : 14 : }
861 : :
862 : 4 : UdpServer& UdpServer::on_error(ErrorHandler h) {
863 : 4 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
864 : 4 : impl_->on_error = std::move(h);
865 : 4 : return *this;
866 : 4 : }
867 : :
868 : 27 : UdpServer& UdpServer::framer(FramerFactory factory) {
869 : 27 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
870 : 27 : impl_->framer_factory = std::move(factory);
871 : 27 : return *this;
872 : 27 : }
873 : :
874 : 9 : UdpServer& UdpServer::on_message(MessageHandler h) {
875 : 9 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
876 : 9 : impl_->on_message = interface::share_callback(std::move(h));
877 : 9 : return *this;
878 : 9 : }
879 : :
880 : 9 : UdpServer& UdpServer::on_message_batch(BatchMessageHandler h) {
881 : 9 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
882 : 9 : impl_->on_message_batch_ = interface::share_callback(std::move(h));
883 : 9 : return *this;
884 : 9 : }
885 : :
886 : 278 : size_t UdpServer::client_count() const {
887 : 278 : std::shared_lock<std::shared_mutex> lock(impl_->mutex);
888 : 556 : return impl_->endpoint_to_id.size();
889 : 278 : }
890 : :
891 : 73 : std::vector<ClientId> UdpServer::connected_clients() const {
892 : 73 : std::shared_lock<std::shared_mutex> lock(impl_->mutex);
893 : 73 : std::vector<ClientId> ids;
894 : 73 : ids.reserve(impl_->sessions.size());
895 [ + + ]: 167 : for (const auto& [id, entry] : impl_->sessions) {
896 : 94 : ids.push_back(id);
897 : : }
898 : 146 : return ids;
899 : 73 : }
900 : :
901 : 1 : UdpServer& UdpServer::auto_start(bool m) {
902 : 1 : impl_->auto_start.store(m);
903 [ + - + - : 1 : if (impl_->auto_start.load() && !impl_->started.load()) {
+ - ]
904 : 1 : start();
905 : : }
906 : 1 : return *this;
907 : : }
908 : :
909 : 1 : UdpServer& UdpServer::bind_address(const std::string& address) {
910 [ + - - + : 1 : config::detail::require(util::InputValidator::is_valid_ipv4(address) || util::InputValidator::is_valid_ipv6(address),
- - - - +
- ]
911 : : "invalid bind address");
912 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
913 : 1 : impl_->require_stopped_config();
914 : 1 : impl_->cfg.bind_address = address;
915 : 1 : return *this;
916 : 1 : }
917 : :
918 : 31 : UdpServer& UdpServer::idle_timeout(std::chrono::milliseconds timeout) {
919 : 31 : config::detail::duration(timeout, base::constants::MIN_IDLE_TIMEOUT_MS, base::constants::MAX_IDLE_TIMEOUT_MS, true,
920 : : "invalid idle_timeout");
921 : 31 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
922 : 31 : impl_->session_timeout = timeout;
923 [ + + ]: 31 : if (impl_->session_timeout.count() <= 0) {
924 [ - + ]: 1 : if (impl_->reaper_timer) {
925 : 0 : impl_->reaper_timer->cancel();
926 : 0 : impl_->reaper_timer.reset();
927 : : }
928 : 1 : return *this;
929 : : }
930 [ + + + - : 30 : if (impl_->started.load() && impl_->channel && !impl_->reaper_timer) {
+ - + + ]
931 : 18 : impl_->reaper_timer = std::make_unique<boost::asio::steady_timer>(impl_->channel->get_executor());
932 : 36 : impl_->schedule_reaper(impl_->callback_generation_.load());
933 : : }
934 : 30 : return *this;
935 : 31 : }
936 : :
937 : 39 : UdpServer& UdpServer::on_backpressure(std::function<void(size_t)> handler) {
938 : 39 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
939 : 39 : impl_->bp_handler = std::move(handler);
940 : 39 : return *this;
941 : 39 : }
942 : :
943 : 6 : UdpServer& UdpServer::max_clients(size_t max) {
944 : 6 : config::detail::range(max, 0, base::constants::MAX_MAX_CONNECTIONS, "invalid client limit");
945 [ + + ]: 5 : if (max == 0) {
946 : 1 : impl_->client_limit_enabled.store(false);
947 : 1 : impl_->max_clients_limit.store(0);
948 : : } else {
949 : 4 : impl_->client_limit_enabled.store(true);
950 : 4 : impl_->max_clients_limit.store(max);
951 : : }
952 : 5 : return *this;
953 : : }
954 : :
955 : 59 : UdpServer& UdpServer::backpressure_threshold(size_t threshold) {
956 : 59 : config::detail::range(threshold, base::constants::MIN_BACKPRESSURE_THRESHOLD,
957 : : base::constants::MAX_BACKPRESSURE_THRESHOLD, "invalid backpressure_threshold");
958 : 59 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
959 : 59 : impl_->require_stopped_config();
960 : 58 : impl_->cfg.backpressure_threshold = threshold;
961 : 58 : return *this;
962 : 59 : }
963 : :
964 : 25 : UdpServer& UdpServer::backpressure_strategy(base::constants::BackpressureStrategy strategy) {
965 : 25 : config::detail::strategy(strategy);
966 : 25 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
967 : 25 : impl_->cfg.backpressure_strategy = strategy;
968 [ + + ]: 25 : if (impl_->channel) {
969 : 20 : impl_->channel->set_backpressure_strategy(strategy);
970 : : }
971 : 25 : return *this;
972 : 25 : }
973 : :
974 : 5 : size_t UdpServer::backpressure_threshold() const {
975 : 5 : std::shared_lock<std::shared_mutex> lock(impl_->mutex);
976 : 10 : return impl_->cfg.backpressure_threshold;
977 : 5 : }
978 : :
979 : 3 : base::constants::BackpressureStrategy UdpServer::backpressure_strategy() const {
980 : 3 : std::shared_lock<std::shared_mutex> lock(impl_->mutex);
981 : 6 : return impl_->cfg.backpressure_strategy;
982 : 3 : }
983 : :
984 : 0 : UdpServer& UdpServer::send_buffer_size(size_t bytes) {
985 [ # # ]: 0 : if (bytes != 0)
986 : 0 : config::detail::range(bytes, base::constants::MIN_SOCKET_BUFFER_SIZE, base::constants::MAX_SOCKET_BUFFER_SIZE,
987 : : "invalid send_buffer_size");
988 : 0 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
989 : 0 : impl_->require_stopped_config();
990 : 0 : impl_->cfg.send_buffer_size = bytes;
991 : 0 : return *this;
992 : 0 : }
993 : :
994 : 0 : UdpServer& UdpServer::receive_buffer_size(size_t bytes) {
995 [ # # ]: 0 : if (bytes != 0)
996 : 0 : config::detail::range(bytes, base::constants::MIN_SOCKET_BUFFER_SIZE, base::constants::MAX_SOCKET_BUFFER_SIZE,
997 : : "invalid receive_buffer_size");
998 : 0 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
999 : 0 : impl_->require_stopped_config();
1000 : 0 : impl_->cfg.receive_buffer_size = bytes;
1001 : 0 : return *this;
1002 : 0 : }
1003 : :
1004 : 5 : UdpServer& UdpServer::manage_external_context(bool m) {
1005 : 5 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
1006 : 5 : impl_->require_stopped_config();
1007 : 4 : impl_->manage_external_context.store(m);
1008 : 4 : return *this;
1009 : 5 : }
1010 : :
1011 : 29 : UdpServer& UdpServer::batch_size(size_t size) {
1012 [ + - ]: 29 : config::detail::require(size > 0, "batch size must be positive");
1013 : 29 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
1014 : 29 : impl_->max_batch_size_ = size;
1015 : 29 : return *this;
1016 : 29 : }
1017 : :
1018 : 29 : UdpServer& UdpServer::batch_latency(std::chrono::milliseconds latency) {
1019 : 29 : config::detail::duration(latency, 0, std::numeric_limits<int>::max(), true, "invalid batch latency");
1020 : 29 : std::unique_lock<std::shared_mutex> lock(impl_->mutex);
1021 : 29 : impl_->max_batch_latency_ = latency;
1022 : 29 : return *this;
1023 : 29 : }
1024 : :
1025 : : } // namespace wrapper
1026 : : } // namespace wirestead
|
