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/transport/udp/udp.hpp"
18 : :
19 : : #include <spdlog/fmt/fmt.h>
20 : :
21 : : #include <array>
22 : : #include <atomic>
23 : : #include <boost/asio.hpp>
24 : : #include <cstddef>
25 : : #include <deque>
26 : : #include <memory>
27 : : #include <mutex>
28 : : #include <optional>
29 : : #include <stdexcept>
30 : : #include <stop_token>
31 : : #include <thread>
32 : : #include <type_traits>
33 : : #include <variant>
34 : : #include <vector>
35 : :
36 : : #include "wirestead/base/common.hpp"
37 : : #include "wirestead/base/constants.hpp"
38 : : #include "wirestead/concurrency/io_thread_hook.hpp"
39 : : #include "wirestead/concurrency/thread_safe_state.hpp"
40 : : #include "wirestead/diagnostics/error_handler.hpp"
41 : : #include "wirestead/diagnostics/logger.hpp"
42 : : #include "wirestead/diagnostics/runtime_stats_counter.hpp"
43 : : #include "wirestead/memory/memory_pool.hpp"
44 : : #include "wirestead/transport/base/bp_state_machine.hpp"
45 : : #include "wirestead/transport/base/bp_utils.hpp"
46 : : #include "wirestead/transport/base/error_info_holder.hpp"
47 : :
48 : : namespace wirestead {
49 : : namespace transport {
50 : :
51 : : namespace net = boost::asio;
52 : : using udp = net::ip::udp;
53 : : using base::LinkState;
54 : : using concurrency::AtomicLinkState;
55 : :
56 : : struct UdpChannel::Impl {
57 : : std::unique_ptr<net::io_context> owned_ioc_;
58 : : net::io_context* ioc_;
59 : : bool owns_ioc_{true};
60 : : net::strand<net::io_context::executor_type> strand_;
61 : : std::unique_ptr<net::executor_work_guard<net::io_context::executor_type>> work_guard_;
62 : : std::jthread ioc_thread_;
63 : :
64 : : udp::socket socket_;
65 : : udp::endpoint local_endpoint_;
66 : : udp::endpoint recv_endpoint_;
67 : : std::optional<udp::endpoint> remote_endpoint_;
68 : :
69 : : using BufferVariant =
70 : : std::variant<memory::PooledBuffer, std::vector<uint8_t>, std::shared_ptr<const std::vector<uint8_t>>>;
71 : : struct TxItem {
72 : : BufferVariant buffer;
73 : : std::optional<udp::endpoint> destination;
74 : : };
75 : :
76 : : std::array<uint8_t, 65536> rx_{};
77 : : std::deque<TxItem> tx_;
78 : : std::deque<TxItem> pending_;
79 : : std::atomic<size_t> pending_bytes_{0};
80 : : bool writing_{false};
81 : : std::atomic<size_t> queue_bytes_{0};
82 : : // Bytes accepted by a plain async_write_* call but not yet routed onto the
83 : : // strand - reserved via try_reserve_limit_bytes() to close the
84 : : // accept-then-drop race (jwsung91/wirestead#517). inflight_bytes_ mutations
85 : : // and the queue_bytes_/pending_bytes_ increments that promote a
86 : : // reservation both go through write_reserve_mtx_ - see bp_utils.hpp.
87 : : std::atomic<size_t> inflight_bytes_{0};
88 : : std::mutex write_reserve_mtx_;
89 : : config::UdpConfig cfg_;
90 : : // #443: per-channel pool instead of the process-wide GlobalMemoryPool
91 : : // singleton - avoids cross-channel contention on the singleton's bucket
92 : : // mutexes. Capacity is much smaller than the old shared default since
93 : : // it's no longer amortized across every channel in the process.
94 : : // Prefill stays 0. This literal was written while MemoryPool discarded
95 : : // initial_pool_size, so 50 allocated nothing; #575 made the parameter real
96 : : // and turned it into ~1 MiB eagerly allocated per channel at construction.
97 : : // The pool fills as buffers are released.
98 : : memory::MemoryPool pool_{0, 200};
99 : : // Atomic rather than mutex-guarded: read both from the strand (report_backpressure,
100 : : // do_write) and from arbitrary caller threads (the async_try_write_* fast-fail
101 : : // prechecks) - a strand-post here would only protect the former, not the latter (#436).
102 : : std::atomic<base::constants::BackpressureStrategy> bp_strategy_{base::constants::BackpressureStrategy::Reliable};
103 : : size_t bp_high_;
104 : : size_t bp_low_;
105 : : size_t bp_limit_;
106 : : std::atomic<bool> backpressure_active_{false};
107 : : diagnostics::RuntimeStatsCounters stats_;
108 : :
109 : : std::atomic<bool> stop_requested_{false};
110 : : std::atomic<bool> stopping_{false};
111 : : std::atomic<bool> opened_{false};
112 : : std::atomic<bool> connected_{false};
113 : : bool started_{false};
114 : 77 : AtomicLinkState state_{LinkState::Idle};
115 : : std::atomic<bool> terminal_state_notified_{false};
116 : :
117 : : // Guards on_bytes_/on_bytes_from_/on_state_/on_bp_. Setters (called from
118 : : // any user thread) and the strand-confined read sites below both take
119 : : // this lock; readers copy the callback under lock then invoke the copy
120 : : // outside the lock, matching the pattern already used correctly by
121 : : // TcpClient/UdsClient/both servers (see #436).
122 : : mutable std::mutex callback_mtx_;
123 : : // Shared snapshots: the strand copies one out per received datagram, and a
124 : : // std::function copy allocates whenever the target outgrows its small-object
125 : : // buffer. See interface::SharedCallback.
126 : : interface::SharedCallback<OnBytes> on_bytes_;
127 : : interface::SharedCallback<UdpChannel::OnBytesFrom> on_bytes_from_;
128 : : interface::SharedCallback<OnState> on_state_;
129 : : interface::SharedCallback<OnBackpressure> on_bp_;
130 : :
131 : : ErrorInfoHolder error_info_holder_{"udp"};
132 : :
133 : 53 : explicit Impl(const config::UdpConfig& config)
134 : 53 : : owned_ioc_(std::make_unique<net::io_context>()),
135 : 53 : ioc_(owned_ioc_.get()),
136 : 53 : owns_ioc_(true),
137 : 53 : strand_(ioc_->get_executor()),
138 : 53 : socket_(strand_),
139 : 53 : cfg_(config),
140 : 53 : bp_strategy_(config.backpressure_strategy),
141 : 265 : bp_high_(config.backpressure_threshold) {
142 : 53 : init();
143 : 66 : }
144 : :
145 : 24 : Impl(const config::UdpConfig& config, net::io_context& external_ioc)
146 : 48 : : ioc_(&external_ioc),
147 : 24 : owns_ioc_(false),
148 : 24 : strand_(external_ioc.get_executor()),
149 : 24 : socket_(strand_),
150 : 24 : cfg_(config),
151 : 24 : bp_strategy_(config.backpressure_strategy),
152 : 144 : bp_high_(config.backpressure_threshold) {
153 : 24 : init();
154 : 24 : }
155 : :
156 : 77 : void init() {
157 : 77 : cfg_.validate_and_clamp();
158 : 77 : bp_high_ = cfg_.backpressure_threshold;
159 [ + - ]: 77 : bp_low_ = bp_high_ > 1 ? bp_high_ / 2 : bp_high_;
160 [ - + ]: 77 : if (bp_low_ == 0) bp_low_ = 1;
161 : 77 : bp_limit_ = std::min(std::max(bp_high_ * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
162 : : base::constants::MAX_BUFFER_SIZE);
163 [ - + ]: 77 : if (bp_limit_ < bp_high_) {
164 : 0 : bp_limit_ = bp_high_;
165 : : }
166 : 77 : set_remote_from_config();
167 : 76 : }
168 : :
169 : 76 : ~Impl() {
170 : : try {
171 : 76 : stop_requested_.store(true);
172 : 76 : stopping_.store(true);
173 [ + + + + : 76 : if (owns_ioc_ && work_guard_) {
+ + ]
174 : 51 : work_guard_.reset();
175 : : }
176 [ - + ]: 76 : if (ioc_thread_.joinable()) {
177 [ # # ]: 0 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
178 : 0 : ioc_thread_.detach();
179 : : } else {
180 : 0 : ioc_thread_.request_stop();
181 : 0 : ioc_thread_.join();
182 : : }
183 : : }
184 : 76 : perform_stop_cleanup();
185 : 0 : } catch (...) {
186 : 0 : }
187 : 76 : }
188 : :
189 : 74 : void open_socket(std::shared_ptr<UdpChannel> self) {
190 [ + - - + : 77 : if (stopping_.load() || stop_requested_.load()) return;
- + ]
191 : :
192 : 74 : boost::system::error_code ec;
193 : 74 : auto address = net::ip::make_address(cfg_.bind_address, ec);
194 [ - + ]: 74 : if (ec) {
195 : 0 : std::string msg = fmt::format("Invalid bind address: {}", cfg_.bind_address);
196 : 0 : WIRESTEAD_LOG_ERROR("udp", "bind", msg);
197 : 0 : transition_to(LinkState::Error, ec, "bind", msg);
198 : 0 : return;
199 : 0 : }
200 : :
201 : 74 : local_endpoint_ = udp::endpoint(address, cfg_.local_port);
202 : 74 : socket_.open(local_endpoint_.protocol(), ec);
203 [ - + ]: 74 : if (ec) {
204 : 0 : std::string msg = fmt::format("Socket open failed: {}", ec.message());
205 : 0 : WIRESTEAD_LOG_ERROR("udp", "open", msg);
206 : 0 : transition_to(LinkState::Error, ec, "open", msg);
207 : 0 : return;
208 : 0 : }
209 : :
210 [ + + ]: 74 : if (cfg_.reuse_address) {
211 : 2 : socket_.set_option(net::socket_base::reuse_address(true), ec);
212 [ - + ]: 2 : if (ec) {
213 : 0 : std::string msg = fmt::format("Failed to set reuse_address: {}", ec.message());
214 : 0 : WIRESTEAD_LOG_ERROR("udp", "open", msg);
215 : 0 : transition_to(LinkState::Error, ec, "open", msg);
216 : 0 : return;
217 : 0 : }
218 : : }
219 : :
220 [ + + ]: 74 : if (cfg_.enable_broadcast) {
221 : 1 : socket_.set_option(net::socket_base::broadcast(true), ec);
222 [ - + ]: 1 : if (ec) {
223 : 0 : std::string msg = fmt::format("Failed to set broadcast: {}", ec.message());
224 : 0 : WIRESTEAD_LOG_ERROR("udp", "open", msg);
225 : 0 : transition_to(LinkState::Error, ec, "open", msg);
226 : 0 : return;
227 : 0 : }
228 : : }
229 : :
230 : 74 : socket_.bind(local_endpoint_, ec);
231 [ + + ]: 74 : if (ec) {
232 : 3 : std::string msg = fmt::format("Bind failed: {}", ec.message());
233 : 3 : WIRESTEAD_LOG_ERROR("udp", "bind", msg);
234 : 3 : transition_to(LinkState::Error, ec, "bind", msg);
235 : 3 : return;
236 : 3 : }
237 : :
238 : : // After bind, which is what a group join attaches to. Fatal on failure:
239 : : // a receiver that silently did not join looks identical to a sensor that
240 : : // stopped sending, and that is the exact confusion this whole feature is
241 : : // meant to remove.
242 [ + + ]: 71 : if (cfg_.multicast_group) {
243 : 1 : const auto group = net::ip::make_address(*cfg_.multicast_group, ec);
244 [ + - ]: 1 : if (!ec) {
245 [ + - + - : 1 : if (group.is_v4() && cfg_.multicast_interface) {
+ - ]
246 : 1 : const auto iface = net::ip::make_address_v4(*cfg_.multicast_interface, ec);
247 [ + - + - : 1 : if (!ec) socket_.set_option(net::ip::multicast::join_group(group.to_v4(), iface), ec);
+ - ]
248 : : } else {
249 : 0 : socket_.set_option(net::ip::multicast::join_group(group), ec);
250 : : }
251 : : }
252 [ - + ]: 1 : if (ec) {
253 : 0 : std::string msg = fmt::format("Failed to join multicast group {}: {}", *cfg_.multicast_group, ec.message());
254 : 0 : WIRESTEAD_LOG_ERROR("udp", "multicast", msg);
255 : 0 : transition_to(LinkState::Error, ec, "multicast", msg);
256 : 0 : return;
257 : 0 : }
258 : 1 : WIRESTEAD_LOG_INFO("udp", "multicast", fmt::format("Joined multicast group {}", *cfg_.multicast_group));
259 : : }
260 : :
261 : : // Set large OS buffers for UDP to prevent drops unless explicitly configured.
262 : 71 : const int automatic_buf_size = std::max(static_cast<int>(bp_high_), 4 * 1024 * 1024);
263 : 71 : const int recv_buf_size =
264 [ - + ]: 71 : cfg_.receive_buffer_size > 0 ? static_cast<int>(cfg_.receive_buffer_size) : automatic_buf_size;
265 [ - + ]: 71 : const int send_buf_size = cfg_.send_buffer_size > 0 ? static_cast<int>(cfg_.send_buffer_size) : automatic_buf_size;
266 : :
267 : 71 : socket_.set_option(net::socket_base::receive_buffer_size(recv_buf_size), ec);
268 [ - + ]: 71 : if (ec) {
269 : 0 : WIRESTEAD_LOG_WARNING("udp", "open", fmt::format("Failed to set receive buffer size: {}", ec.message()));
270 : 0 : ec.clear();
271 : : }
272 : 71 : socket_.set_option(net::socket_base::send_buffer_size(send_buf_size), ec);
273 [ - + ]: 71 : if (ec) {
274 : 0 : WIRESTEAD_LOG_WARNING("udp", "open", fmt::format("Failed to set send buffer size: {}", ec.message()));
275 : 0 : ec.clear();
276 : : }
277 : :
278 : 71 : opened_.store(true);
279 [ + + ]: 71 : if (remote_endpoint_) {
280 : 24 : connected_.store(true);
281 : 24 : transition_to(LinkState::Connected);
282 : : } else {
283 : 47 : transition_to(LinkState::Listening);
284 : : }
285 : 71 : start_receive(self);
286 : : }
287 : :
288 : 127 : void start_receive(std::shared_ptr<UdpChannel> self) {
289 [ + - + - : 379 : if (stopping_.load() || stop_requested_.load() || state_.is_state(LinkState::Closed) ||
+ - ]
290 [ + + + - : 379 : state_.is_state(LinkState::Error) || !socket_.is_open()) {
- + + + ]
291 : 1 : return;
292 : : }
293 : :
294 : 126 : socket_.async_receive_from(net::buffer(rx_), recv_endpoint_,
295 : 252 : [self](const boost::system::error_code& ec, std::size_t bytes) {
296 : 104 : auto impl = self->get_impl();
297 : 104 : impl->handle_receive(self, ec, bytes);
298 : 104 : });
299 : : }
300 : :
301 : 104 : void handle_receive(std::shared_ptr<UdpChannel> self, const boost::system::error_code& ec, std::size_t bytes) {
302 [ + + ]: 104 : if (ec == boost::asio::error::operation_aborted) {
303 : 47 : return;
304 : : }
305 : :
306 [ + - + - : 114 : if (stopping_.load() || stop_requested_.load() || state_.is_state(LinkState::Closed) ||
+ - - + ]
307 : 114 : state_.is_state(LinkState::Error)) {
308 : 0 : return;
309 : : }
310 : :
311 [ + - - + : 114 : if (ec == boost::asio::error::message_size || bytes >= rx_.size()) {
- + ]
312 : 0 : WIRESTEAD_LOG_ERROR("udp", "receive", "Datagram truncated (buffer too small)");
313 : 0 : transition_to(LinkState::Error, ec, "receive", "Datagram truncated (buffer too small)");
314 : 0 : return;
315 : : }
316 : :
317 [ - + ]: 57 : if (ec) {
318 : 0 : std::string msg = fmt::format("Receive failed: {}", ec.message());
319 : 0 : WIRESTEAD_LOG_ERROR("udp", "receive", msg);
320 : 0 : transition_to(LinkState::Error, ec, "receive", msg);
321 : 0 : return;
322 : 0 : }
323 : :
324 [ + + ]: 57 : if (!remote_endpoint_) {
325 : 26 : remote_endpoint_ = recv_endpoint_;
326 : 26 : connected_.store(true);
327 : 26 : transition_to(LinkState::Connected);
328 : : }
329 : :
330 : : // #435: once a remote peer is configured or locked in (above), only
331 : : // deliver data from that exact sender through on_bytes() - the
332 : : // point-to-point API. Without this, any other host could send spoofed
333 : : // datagrams after the fact and have them treated as legitimate data.
334 : : // on_bytes_from() is intentionally NOT filtered here: it's the
335 : : // multi-sender API (used by the UdpServer wrapper), which tracks and
336 : : // trusts each sender as its own session by design.
337 [ + - + + ]: 57 : const bool from_established_remote = remote_endpoint_ && recv_endpoint_ == *remote_endpoint_;
338 : :
339 [ + - ]: 57 : if (bytes > 0) {
340 : 57 : stats_.record_received(bytes);
341 : 57 : interface::SharedCallback<OnBytes> on_bytes;
342 : 57 : interface::SharedCallback<UdpChannel::OnBytesFrom> on_bytes_from;
343 : : {
344 : 57 : std::lock_guard<std::mutex> lock(callback_mtx_);
345 : 57 : on_bytes = on_bytes_;
346 : 57 : on_bytes_from = on_bytes_from_;
347 : 57 : }
348 [ + + + + : 57 : if (on_bytes && from_established_remote) {
+ + ]
349 : : try {
350 : 62 : (*on_bytes)(memory::ConstByteSpan(rx_.data(), bytes));
351 : 0 : } catch (const std::exception& e) {
352 : 0 : std::string msg = fmt::format("Exception in bytes callback: {}", e.what());
353 : 0 : WIRESTEAD_LOG_ERROR("udp", "on_bytes", msg);
354 [ - - ]: 0 : if (cfg_.stop_on_callback_exception) {
355 : 0 : transition_to(LinkState::Error, {}, "on_bytes", msg);
356 : 0 : return;
357 : : }
358 : 0 : } catch (...) {
359 : 0 : WIRESTEAD_LOG_ERROR("udp", "on_bytes", "Unknown exception in bytes callback");
360 [ - - ]: 0 : if (cfg_.stop_on_callback_exception) {
361 : 0 : transition_to(LinkState::Error, {}, "on_bytes", "Unknown exception in bytes callback");
362 : 0 : return;
363 : : }
364 : 0 : }
365 : : }
366 : :
367 [ + + ]: 57 : if (on_bytes_from) {
368 : : try {
369 : 46 : (*on_bytes_from)(memory::ConstByteSpan(rx_.data(), bytes), recv_endpoint_);
370 : 1 : } catch (const std::exception& e) {
371 : 1 : std::string msg = fmt::format("Exception in bytes callback: {}", e.what());
372 : 1 : WIRESTEAD_LOG_ERROR("udp", "on_bytes_from", msg);
373 [ + - ]: 1 : if (cfg_.stop_on_callback_exception) {
374 : 1 : transition_to(LinkState::Error, {}, "on_bytes_from", msg);
375 : 1 : return;
376 : : }
377 : 2 : } catch (...) {
378 : 0 : WIRESTEAD_LOG_ERROR("udp", "on_bytes_from", "Unknown exception in bytes callback");
379 [ - - ]: 0 : if (cfg_.stop_on_callback_exception) {
380 : 0 : transition_to(LinkState::Error, {}, "on_bytes_from", "Unknown exception in bytes callback");
381 : 0 : return;
382 : : }
383 : 0 : }
384 : : }
385 : 58 : }
386 : :
387 : 56 : start_receive(self);
388 : : }
389 : :
390 : 340482 : queue_util::BackpressureFields bp_fields() {
391 : 340482 : return queue_util::BackpressureFields{queue_bytes_,
392 : 340482 : pending_bytes_,
393 : 340482 : backpressure_active_,
394 : 340482 : bp_high_,
395 : 340482 : bp_low_,
396 : 340482 : bp_limit_,
397 : 340482 : bp_strategy_.load(std::memory_order_relaxed)};
398 : : }
399 : :
400 : : // Drops all queued (tx_) and pending (pending_, reliable-mode overflow) writes and clears
401 : : // backpressure, notifying any waiter directly. Mirrors perform_stop_cleanup()'s approach
402 : : // rather than calling report_backpressure(), because report_backpressure() flushes pending_
403 : : // back into tx_ and can immediately re-arm backpressure_active_ if enough was queued there -
404 : : // which would leave a Reliable-mode sender blocked in send_blocking()'s bp_cv_ wait forever
405 : : // since nothing will ever call do_write() again once the channel has stopped/errored (#427).
406 : 7 : void drain_queue_and_clear_backpressure() {
407 : 7 : interface::SharedCallback<OnBackpressure> on_bp;
408 : : {
409 : 7 : std::lock_guard<std::mutex> lock(callback_mtx_);
410 : 7 : on_bp = on_bp_;
411 : 7 : }
412 : 7 : static const OnBackpressure kNoCallback;
413 : 7 : auto f = bp_fields();
414 [ + + + - ]: 7 : queue_util::drain_and_clear_backpressure(f, on_bp ? *on_bp : kNoCallback, [&]() {
415 : 7 : tx_.clear();
416 : 7 : queue_bytes_ = 0;
417 : 7 : pending_.clear();
418 : 7 : pending_bytes_ = 0;
419 : 7 : });
420 : 7 : }
421 : :
422 : 340240 : void do_write(std::shared_ptr<UdpChannel> self) {
423 [ + + + + : 340241 : if (writing_ || tx_.empty()) return;
+ + ]
424 [ + + + - : 129 : if (stop_requested_.load() || stopping_.load() || state_.is_state(LinkState::Closed) ||
+ - - + ]
425 : 129 : state_.is_state(LinkState::Error)) {
426 : 1 : writing_ = false;
427 : 1 : drain_queue_and_clear_backpressure();
428 : 1 : return;
429 : : }
430 : :
431 : 64 : auto current = std::move(tx_.front());
432 : 64 : tx_.pop_front();
433 : :
434 [ + + ]: 64 : const auto& dest_endpoint = current.destination ? current.destination : remote_endpoint_;
435 : :
436 [ - + ]: 64 : if (!dest_endpoint) {
437 : 0 : WIRESTEAD_LOG_WARNING("udp", "write", "Remote endpoint not set; dropping write request");
438 : 0 : writing_ = false;
439 : 0 : do_write(self); // Process next in queue
440 : 0 : return;
441 : : }
442 : :
443 : 64 : writing_ = true;
444 : :
445 [ + - ]: 64 : auto bytes_queued = std::visit(
446 : 64 : [](auto&& buf) -> size_t {
447 : : using Buffer = std::decay_t<decltype(buf)>;
448 : : if constexpr (std::is_same_v<Buffer, std::shared_ptr<const std::vector<uint8_t>>>) {
449 [ + - ]: 1 : return buf ? buf->size() : 0;
450 : : } else {
451 : 63 : return buf.size();
452 : : }
453 : : },
454 : : current.buffer);
455 : :
456 : 64 : auto on_write = [self, bytes_queued](const boost::system::error_code& ec, std::size_t bytes_written) {
457 : 64 : auto impl = self->get_impl();
458 [ + + ]: 64 : impl->queue_bytes_ = (impl->queue_bytes_ > bytes_queued) ? (impl->queue_bytes_ - bytes_queued) : 0;
459 : 64 : impl->report_backpressure(self, impl->queue_bytes_);
460 : :
461 [ - + ]: 64 : if (ec == boost::asio::error::operation_aborted) {
462 : 0 : impl->writing_ = false;
463 : 0 : return;
464 : : }
465 : :
466 [ + + + - : 126 : if (impl->stop_requested_.load() || impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) ||
+ - - + ]
467 : 126 : impl->state_.is_state(LinkState::Error)) {
468 : 2 : impl->writing_ = false;
469 : 2 : impl->drain_queue_and_clear_backpressure();
470 : 2 : return;
471 : : }
472 : :
473 [ + + ]: 62 : if (ec) {
474 : 4 : std::string msg = fmt::format("Send failed: {}", ec.message());
475 : 4 : WIRESTEAD_LOG_ERROR("udp", "write", msg);
476 : 4 : impl->transition_to(LinkState::Error, ec, "write", msg);
477 : 4 : impl->writing_ = false;
478 : : // do_write() will never run again to reach the "already in Error" cleanup above, so
479 : : // drain everything and clear backpressure here directly.
480 : 4 : impl->drain_queue_and_clear_backpressure();
481 : 4 : return;
482 : 4 : }
483 : :
484 : 58 : impl->stats_.record_sent(bytes_written);
485 : 58 : impl->writing_ = false;
486 : 58 : impl->do_write(self);
487 : 64 : };
488 : :
489 : 64 : std::visit(
490 [ + - ]: 128 : [&](auto&& buf) {
491 : : using T = std::decay_t<decltype(buf)>;
492 : :
493 : 152 : auto* data_ptr = [&]() {
494 : : if constexpr (std::is_same_v<T, std::shared_ptr<const std::vector<uint8_t>>>) {
495 : 1 : return buf->data();
496 : : } else {
497 : 63 : return buf.data();
498 : : }
499 : 64 : }();
500 : :
501 : 152 : auto size = [&]() {
502 : : if constexpr (std::is_same_v<T, std::shared_ptr<const std::vector<uint8_t>>>) {
503 : 1 : return buf->size();
504 : : } else {
505 : 63 : return buf.size();
506 : : }
507 : 64 : }();
508 : :
509 : 64 : socket_.async_send_to(
510 : 64 : net::buffer(data_ptr, size), *dest_endpoint,
511 : 256 : [buf_captured = std::move(buf), on_write = std::move(on_write)](
512 : 64 : const boost::system::error_code& ec, std::size_t bytes) mutable { on_write(ec, bytes); });
513 : 64 : },
514 : 64 : std::move(current.buffer));
515 : 64 : }
516 : :
517 : 205 : void close_socket() {
518 : 205 : boost::system::error_code ec;
519 : 205 : socket_.cancel(ec);
520 : 205 : socket_.close(ec);
521 : 205 : }
522 : :
523 : 247 : void notify_state() {
524 : 247 : interface::SharedCallback<OnState> on_state;
525 : : {
526 : 247 : std::lock_guard<std::mutex> lock(callback_mtx_);
527 : 247 : on_state = on_state_;
528 : 247 : }
529 [ + + ]: 247 : if (!on_state) return;
530 : : try {
531 : 205 : (*on_state)(state_.get());
532 : 3 : } catch (const std::exception& e) {
533 : 3 : WIRESTEAD_LOG_ERROR("udp", "on_state", fmt::format("Exception in state callback: {}", e.what()));
534 : 3 : } catch (...) {
535 : 0 : WIRESTEAD_LOG_ERROR("udp", "on_state", "Unknown exception in state callback");
536 : 0 : }
537 : 247 : }
538 : :
539 : 340234 : void report_backpressure(std::shared_ptr<UdpChannel> self, size_t queued_bytes) {
540 [ + + ]: 340234 : if (stop_requested_.load()) return;
541 : 340232 : observe_queue();
542 : :
543 : 340232 : interface::SharedCallback<OnBackpressure> on_bp;
544 : : {
545 : 340232 : std::lock_guard<std::mutex> lock(callback_mtx_);
546 : 340232 : on_bp = on_bp_;
547 : 340232 : }
548 : 340232 : static const OnBackpressure kNoCallback;
549 : :
550 : 340232 : auto f = bp_fields();
551 [ + + + - ]: 340345 : queue_util::report_backpressure(
552 : 113 : f, queued_bytes, on_bp ? *on_bp : kNoCallback, stats_,
553 : 0 : [&]() -> size_t {
554 : : // Flush pending_ → tx_
555 : 6 : const size_t moved = pending_bytes_.exchange(0);
556 [ + + ]: 18 : while (!pending_.empty()) {
557 : 12 : tx_.emplace_back(std::move(pending_.front()));
558 : 12 : pending_.pop_front();
559 : : }
560 : 6 : return moved;
561 : : },
562 : 340232 : [&]() {
563 : : // Post-flush sample: queue_bytes_ has already been updated by the
564 : : // time this kick runs, unlike the flush hook above (#434).
565 : 6 : observe_queue();
566 [ - + - - ]: 6 : if (!writing_) do_write(self);
567 : 6 : });
568 : 340232 : }
569 : :
570 : 680420 : void observe_queue() {
571 : 2041260 : stats_.observe_queue(queue_bytes_.load(std::memory_order_relaxed) + pending_bytes_.load(std::memory_order_relaxed));
572 : 680420 : }
573 : :
574 : 38 : bool enqueue_buffer(std::shared_ptr<UdpChannel> self, BufferVariant&& buffer, size_t size,
575 : : std::optional<udp::endpoint> dest = std::nullopt) {
576 [ + - + - : 76 : if (stopping_.load() || stop_requested_.load() || state_.is_state(LinkState::Closed) ||
+ - - + ]
577 : 76 : state_.is_state(LinkState::Error)) {
578 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, size);
579 : 0 : stats_.record_failed_send();
580 : 0 : return false;
581 : : }
582 : :
583 : 38 : auto f = bp_fields();
584 : 38 : queue_util::DropAccounting dropped;
585 : : // TxItem carries a destination endpoint alongside the BufferVariant that
586 : : // decide_enqueue()'s BestEffort trim needs to visit - project onto
587 : : // `.buffer` rather than the item itself (#434).
588 : : auto decision =
589 : 38 : queue_util::decide_enqueue(f, size, tx_, dropped, [](TxItem& item) -> BufferVariant& { return item.buffer; });
590 [ - + ]: 38 : if (dropped.any()) {
591 : 0 : stats_.record_dropped(dropped.messages, dropped.bytes);
592 : : }
593 : :
594 [ - + ]: 38 : if (decision == queue_util::EnqueueDecision::Rejected) {
595 : 0 : WIRESTEAD_LOG_ERROR("udp", "write",
596 : : fmt::format("Queue limit exceeded ({} bytes)", queue_bytes_ + pending_bytes_ + size));
597 : : // Always reporting here (rather than only for the non-Reliable-pending
598 : : // rejection path, as the pre-#434 code did) is a no-op in practice for
599 : : // the Reliable+pending case: queued_bytes is already >= bp_high_ or
600 : : // this rejection couldn't have happened, so report_backpressure()'s
601 : : // OFF-transition check (<= bp_low_) can't fire here either way.
602 : : // #448: record as dropped so it's reflected in RuntimeStats instead of
603 : : // silently vanishing after being counted as accepted.
604 : 0 : stats_.record_dropped(1, size);
605 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, size);
606 : 0 : report_backpressure(self, queue_bytes_ + size);
607 : 0 : return false;
608 : : }
609 : :
610 [ + + ]: 38 : if (decision == queue_util::EnqueueDecision::Pending) {
611 : 12 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, pending_bytes_, inflight_bytes_, size);
612 : 24 : pending_.push_back({std::move(buffer), dest});
613 : 12 : observe_queue();
614 : 12 : return true;
615 : : }
616 : :
617 : 26 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, queue_bytes_, inflight_bytes_, size);
618 : 52 : tx_.push_back({std::move(buffer), dest});
619 : 26 : observe_queue();
620 : 26 : report_backpressure(self, queue_bytes_);
621 : 26 : return true;
622 : 38 : }
623 : :
624 : 77 : void set_remote_from_config() {
625 [ + + - + : 77 : if (!cfg_.remote_address || !cfg_.remote_port) return;
+ + ]
626 : 25 : boost::system::error_code ec;
627 : 25 : auto addr = net::ip::make_address(*cfg_.remote_address, ec);
628 [ + + ]: 25 : if (ec) {
629 : 1 : throw std::runtime_error("Invalid remote address: " + *cfg_.remote_address);
630 : : }
631 : 24 : remote_endpoint_ = udp::endpoint(addr, *cfg_.remote_port);
632 : : }
633 : :
634 : 385 : void transition_to(LinkState target, const boost::system::error_code& ec = {}, std::string_view operation = {},
635 : : std::string_view msg = {}) {
636 [ - + ]: 385 : if (ec == net::error::operation_aborted) {
637 : 0 : return;
638 : : }
639 : :
640 : 385 : const auto current = state_.get();
641 [ + + + + ]: 385 : if ((current == LinkState::Closed || current == LinkState::Error) &&
642 [ - + - - ]: 138 : (target == LinkState::Closed || target == LinkState::Error)) {
643 : 138 : return;
644 : : }
645 : :
646 [ + + ]: 247 : if (target == LinkState::Error) {
647 [ - + + - ]: 8 : error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, operation,
648 [ - - - + : 16 : ec, msg.empty() ? std::string_view(ec.message()) : msg, static_cast<bool>(ec), 0);
- - ]
649 : : }
650 : :
651 [ + + + + ]: 247 : if (target == LinkState::Closed || target == LinkState::Error) {
652 [ - + ]: 76 : if (terminal_state_notified_.exchange(true)) {
653 : 0 : return;
654 : : }
655 [ - + ]: 171 : } else if (current == target) {
656 : 0 : return;
657 : : }
658 : :
659 : 247 : state_.set(target);
660 : 247 : notify_state();
661 : : }
662 : :
663 : 205 : void perform_stop_cleanup() {
664 : : try {
665 : 205 : close_socket();
666 : 205 : writing_ = false;
667 : 205 : interface::SharedCallback<OnBackpressure> on_bp;
668 : : {
669 : 205 : std::lock_guard<std::mutex> lock(callback_mtx_);
670 : 205 : on_bp = on_bp_;
671 : 205 : }
672 : 205 : static const OnBackpressure kNoCallback;
673 : 205 : auto f = bp_fields();
674 [ + + + - ]: 205 : queue_util::drain_and_clear_backpressure(f, on_bp ? *on_bp : kNoCallback, [&]() {
675 : 205 : tx_.clear();
676 : 205 : queue_bytes_ = 0;
677 : 205 : pending_.clear();
678 : 205 : pending_bytes_ = 0;
679 : 205 : });
680 : 205 : connected_.store(false);
681 : 205 : opened_.store(false);
682 [ + + + + : 205 : if (owns_ioc_ && work_guard_) {
+ + ]
683 : 102 : work_guard_->reset();
684 : : }
685 : 205 : transition_to(LinkState::Closed);
686 : : {
687 : 205 : std::lock_guard<std::mutex> lock(callback_mtx_);
688 : 205 : on_bytes_ = nullptr;
689 : 205 : on_state_ = nullptr;
690 : 205 : on_bp_ = nullptr;
691 : 205 : }
692 : 205 : } catch (...) {
693 : 0 : }
694 : 205 : }
695 : :
696 : 150 : void join_ioc_thread(bool allow_detach) {
697 [ + + + + : 150 : if (!owns_ioc_ || !ioc_thread_.joinable()) {
+ + ]
698 : 99 : return;
699 : : }
700 : :
701 [ - + ]: 51 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
702 [ # # ]: 0 : if (allow_detach) {
703 : 0 : ioc_thread_.detach();
704 : : }
705 : 0 : return;
706 : : }
707 : :
708 : : try {
709 : 51 : ioc_thread_.join();
710 : 0 : } catch (...) {
711 : 0 : }
712 : : }
713 : : };
714 : :
715 : 53 : std::shared_ptr<UdpChannel> UdpChannel::create(const config::UdpConfig& cfg) {
716 : 53 : return std::shared_ptr<UdpChannel>(new UdpChannel(cfg));
717 : : }
718 : :
719 : 24 : std::shared_ptr<UdpChannel> UdpChannel::create(const config::UdpConfig& cfg, net::io_context& ioc) {
720 : 24 : return std::shared_ptr<UdpChannel>(new UdpChannel(cfg, ioc));
721 : : }
722 : :
723 : 54 : UdpChannel::UdpChannel(const config::UdpConfig& cfg) : impl_(std::make_unique<Impl>(cfg)) {}
724 : :
725 : 24 : UdpChannel::UdpChannel(const config::UdpConfig& cfg, net::io_context& ioc) : impl_(std::make_unique<Impl>(cfg, ioc)) {}
726 : :
727 : 152 : UdpChannel::~UdpChannel() {
728 [ + - ]: 76 : if (impl_) {
729 : : // Cannot use shared_from_this in destructor. Use internal cleanup directly.
730 : 76 : impl_->stop_requested_.store(true);
731 : 76 : impl_->stopping_.store(true);
732 : 76 : impl_->perform_stop_cleanup();
733 : 76 : impl_->join_ioc_thread(true);
734 : : }
735 : 152 : }
736 : :
737 : 0 : UdpChannel::UdpChannel(UdpChannel&&) noexcept = default;
738 : 0 : UdpChannel& UdpChannel::operator=(UdpChannel&&) noexcept = default;
739 : :
740 : 74 : void UdpChannel::start() {
741 : 74 : auto impl = get_impl();
742 [ - + ]: 74 : if (impl->started_) return;
743 [ + - - + ]: 74 : if (!impl->cfg_.is_valid()) {
744 : 0 : throw std::runtime_error("Invalid UDP configuration");
745 : : }
746 : :
747 [ + + + - : 74 : if (impl->owns_ioc_ && impl->owned_ioc_ && impl->owned_ioc_->stopped()) {
+ - - + -
+ ]
748 : 0 : impl->owned_ioc_->restart();
749 : : }
750 : :
751 [ - + ]: 74 : if (impl->ioc_thread_.joinable()) {
752 : 0 : impl->join_ioc_thread(false);
753 : : }
754 : :
755 [ + + ]: 74 : if (impl->owns_ioc_) {
756 : : impl->work_guard_ =
757 : 51 : std::make_unique<net::executor_work_guard<net::io_context::executor_type>>(impl->ioc_->get_executor());
758 : : }
759 : :
760 : 74 : auto self = shared_from_this();
761 : 74 : net::dispatch(impl->strand_, [self]() {
762 : 74 : auto impl = self->get_impl();
763 : 74 : impl->stop_requested_.store(false);
764 : 74 : impl->stopping_.store(false);
765 : 74 : impl->terminal_state_notified_.store(false);
766 : 74 : impl->connected_.store(false);
767 : 74 : impl->opened_.store(false);
768 : 74 : impl->writing_ = false;
769 : 74 : impl->queue_bytes_ = 0;
770 : 74 : impl->backpressure_active_ = false;
771 : 74 : impl->state_.set(LinkState::Idle);
772 : :
773 : 74 : impl->transition_to(LinkState::Connecting);
774 : 74 : impl->open_socket(self);
775 : 74 : });
776 : :
777 [ + + ]: 74 : if (impl->owns_ioc_) {
778 : 102 : impl->ioc_thread_ = std::jthread([impl](std::stop_token st) {
779 : 51 : wirestead::concurrency::run_io_thread_init();
780 : : try {
781 : 51 : std::stop_callback cb(st, [impl] { impl->ioc_->stop(); });
782 : 51 : impl->ioc_->run();
783 : 51 : } catch (...) {
784 : 0 : }
785 : 102 : });
786 : : }
787 : :
788 : 74 : impl->started_ = true;
789 : 74 : }
790 : :
791 : 75 : void UdpChannel::stop() {
792 : 75 : auto impl = get_impl();
793 [ - + ]: 76 : if (impl->stop_requested_.exchange(true)) return;
794 : :
795 [ + + ]: 75 : if (!impl->started_) {
796 : 1 : impl->transition_to(LinkState::Closed);
797 : 1 : std::lock_guard<std::mutex> lock(impl->callback_mtx_);
798 : 1 : impl->on_bytes_ = nullptr;
799 : 1 : impl->on_state_ = nullptr;
800 : 1 : impl->on_bp_ = nullptr;
801 : 1 : return;
802 : 1 : }
803 : :
804 : 74 : impl->stopping_.store(true);
805 : 74 : auto self = shared_from_this();
806 : 127 : net::post(impl->strand_, [self]() { self->get_impl()->perform_stop_cleanup(); });
807 : :
808 : 74 : impl->join_ioc_thread(false);
809 : :
810 [ + + + - : 74 : if (impl->owns_ioc_ && impl->owned_ioc_) {
+ + ]
811 : 51 : impl->owned_ioc_->restart();
812 : : }
813 : :
814 : 74 : impl->started_ = false;
815 : 74 : }
816 : :
817 : 32 : bool UdpChannel::is_connected() const { return get_impl()->connected_.load(); }
818 : 19 : bool UdpChannel::is_backpressure_active() const { return get_impl()->backpressure_active_.load(); }
819 : 26 : wrapper::RuntimeStats UdpChannel::stats() const {
820 : 26 : auto impl = get_impl();
821 : 78 : return impl->stats_.snapshot(impl->queue_bytes_.load(std::memory_order_relaxed),
822 : : impl->pending_bytes_.load(std::memory_order_relaxed),
823 : 52 : impl->backpressure_active_.load(std::memory_order_relaxed));
824 : : }
825 : 1 : void UdpChannel::reset_stats() {
826 : 1 : auto impl = get_impl();
827 : 1 : impl->stats_.reset(impl->queue_bytes_.load(std::memory_order_relaxed) +
828 : 2 : impl->pending_bytes_.load(std::memory_order_relaxed));
829 : 1 : }
830 : :
831 : 92 : std::optional<diagnostics::ErrorInfo> UdpChannel::last_error_info() const {
832 : 92 : return get_impl()->error_info_holder_.last_error_info();
833 : : }
834 : :
835 : 49 : bool UdpChannel::async_write_copy(memory::ConstByteSpan data) {
836 : 49 : auto impl = get_impl();
837 [ + + ]: 49 : if (data.empty()) {
838 : 1 : impl->stats_.record_failed_send();
839 : 1 : return false;
840 : : }
841 [ + + ]: 48 : if (impl->stop_requested_.load()) {
842 : 2 : impl->stats_.record_failed_send();
843 : 2 : return false;
844 : : }
845 [ + - + - : 46 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
- + - + ]
846 : 0 : impl->stats_.record_failed_send();
847 : 0 : return false;
848 : : }
849 [ + + ]: 46 : if (!impl->remote_endpoint_) {
850 : 3 : impl->stats_.record_failed_send();
851 : 3 : return false;
852 : : }
853 : :
854 : 43 : size_t size = data.size();
855 [ - + ]: 43 : if (size > base::constants::MAX_BUFFER_SIZE) {
856 : 0 : WIRESTEAD_LOG_ERROR("udp", "write", "Write size exceeds maximum allowed");
857 : 0 : impl->stats_.record_failed_send();
858 : 0 : return false;
859 : : }
860 : :
861 [ + - + + ]: 43 : if (impl->cfg_.enable_memory_pool && size <= 65536) {
862 : 16 : memory::PooledBuffer pooled(size, impl->pool_);
863 [ + - + - ]: 16 : if (pooled.valid()) {
864 : 16 : base::safe_memory::safe_memcpy(pooled.data(), data.data(), size);
865 [ - + ]: 16 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queue_bytes_, impl->pending_bytes_,
866 : 16 : impl->inflight_bytes_, size, impl->bp_limit_)) {
867 : 0 : impl->stats_.record_failed_send();
868 : 0 : return false;
869 : : }
870 : 16 : impl->stats_.record_accepted(size);
871 : 16 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(pooled), size]() mutable {
872 : 16 : auto impl = self->get_impl();
873 [ + - - + ]: 16 : if (!impl->enqueue_buffer(self, std::move(buf), size)) return;
874 : 16 : impl->do_write(self);
875 : : });
876 : 16 : return true;
877 : : }
878 : 16 : }
879 : :
880 [ + + ]: 27 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queue_bytes_, impl->pending_bytes_,
881 : 27 : impl->inflight_bytes_, size, impl->bp_limit_)) {
882 : 12 : impl->stats_.record_failed_send();
883 : 12 : return false;
884 : : }
885 : 15 : std::vector<uint8_t> copy(data.begin(), data.end());
886 : 15 : impl->stats_.record_accepted(size);
887 : 15 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(copy), size]() mutable {
888 : 15 : auto impl = self->get_impl();
889 [ + - - + ]: 15 : if (!impl->enqueue_buffer(self, std::move(buf), size)) return;
890 : 15 : impl->do_write(self);
891 : : });
892 : 15 : return true;
893 : 15 : }
894 : :
895 : 4 : bool UdpChannel::async_write_move(std::vector<uint8_t>&& data) {
896 : 4 : auto impl = get_impl();
897 : 4 : auto size = data.size();
898 [ + + ]: 4 : if (size == 0) {
899 : 1 : impl->stats_.record_failed_send();
900 : 1 : return false;
901 : : }
902 [ + + ]: 3 : if (impl->stop_requested_.load()) {
903 : 1 : impl->stats_.record_failed_send();
904 : 1 : return false;
905 : : }
906 [ + - + - : 2 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
- + - + ]
907 : 0 : impl->stats_.record_failed_send();
908 : 0 : return false;
909 : : }
910 [ - + ]: 2 : if (!impl->remote_endpoint_) {
911 : 0 : impl->stats_.record_failed_send();
912 : 0 : return false;
913 : : }
914 : :
915 [ - + ]: 2 : if (size > impl->bp_limit_) {
916 : 0 : WIRESTEAD_LOG_ERROR("udp", "write", "Queue limit exceeded by single write");
917 : 0 : impl->transition_to(LinkState::Error, {}, "write", "Queue limit exceeded by single write");
918 : 0 : impl->stats_.record_failed_send();
919 : 0 : return false;
920 : : }
921 [ - + ]: 2 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queue_bytes_, impl->pending_bytes_,
922 : 2 : impl->inflight_bytes_, size, impl->bp_limit_)) {
923 : 0 : impl->stats_.record_failed_send();
924 : 0 : return false;
925 : : }
926 : 2 : impl->stats_.record_accepted(size);
927 : 2 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(data), size]() mutable {
928 : 2 : auto impl = self->get_impl();
929 [ + - - + ]: 2 : if (!impl->enqueue_buffer(self, std::move(buf), size)) return;
930 : 2 : impl->do_write(self);
931 : : });
932 : 2 : return true;
933 : : }
934 : :
935 : 4 : bool UdpChannel::async_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
936 : 4 : auto impl = get_impl();
937 [ + + + + : 4 : if (!data || data->empty()) {
+ + ]
938 : 2 : impl->stats_.record_failed_send();
939 : 2 : return false;
940 : : }
941 [ - + ]: 2 : if (impl->stop_requested_.load()) {
942 : 0 : impl->stats_.record_failed_send();
943 : 0 : return false;
944 : : }
945 [ + - + - : 2 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
- + - + ]
946 : 0 : impl->stats_.record_failed_send();
947 : 0 : return false;
948 : : }
949 [ + + ]: 2 : if (!impl->remote_endpoint_) {
950 : 1 : WIRESTEAD_LOG_WARNING("udp", "write", "Remote endpoint not set; dropping write request");
951 : 1 : impl->stats_.record_failed_send();
952 : 1 : return false;
953 : : }
954 : :
955 : 1 : auto size = data->size();
956 [ - + ]: 1 : if (size > impl->bp_limit_) {
957 : 0 : WIRESTEAD_LOG_ERROR("udp", "write", "Queue limit exceeded by single write");
958 : 0 : impl->transition_to(LinkState::Error, {}, "write", "Queue limit exceeded by single write");
959 : 0 : impl->stats_.record_failed_send();
960 : 0 : return false;
961 : : }
962 [ - + ]: 1 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queue_bytes_, impl->pending_bytes_,
963 : 1 : impl->inflight_bytes_, size, impl->bp_limit_)) {
964 : 0 : impl->stats_.record_failed_send();
965 : 0 : return false;
966 : : }
967 : 1 : impl->stats_.record_accepted(size);
968 : 1 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(data), size]() mutable {
969 : 1 : auto impl = self->get_impl();
970 [ + - - + ]: 1 : if (!impl->enqueue_buffer(self, std::move(buf), size)) return;
971 : 1 : impl->do_write(self);
972 : : });
973 : 1 : return true;
974 : : }
975 : :
976 : 414248 : bool UdpChannel::async_try_write_copy(memory::ConstByteSpan data) {
977 [ + - - + : 414248 : if (data.empty() || data.size() > base::constants::MAX_BUFFER_SIZE) {
- + ]
978 : 0 : get_impl()->stats_.record_failed_send();
979 : 0 : return false;
980 : : }
981 : 1242744 : return async_try_write_move(std::vector<uint8_t>(data.begin(), data.end()));
982 : : }
983 : :
984 : 414253 : bool UdpChannel::async_try_write_move(std::vector<uint8_t>&& data) {
985 : 414253 : auto impl = get_impl();
986 [ + - + - : 1242759 : if (impl->stop_requested_.load() || impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) ||
+ - ]
987 [ + - - + : 1242759 : impl->state_.is_state(LinkState::Error) || !impl->remote_endpoint_) {
- + ]
988 : 0 : impl->stats_.record_failed_send();
989 : 0 : return false;
990 : : }
991 : 414253 : const auto size = data.size();
992 [ + - - + ]: 414253 : if (size == 0 || size > base::constants::MAX_BUFFER_SIZE) {
993 : 0 : impl->stats_.record_failed_send();
994 : 0 : return false;
995 : : }
996 : 3 : const auto reject_for_pressure = [impl, size]() {
997 [ + + ]: 3 : if (impl->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
998 : 1 : impl->stats_.record_dropped(1, size);
999 : : } else {
1000 : 2 : impl->stats_.record_failed_send();
1001 : : }
1002 : 414256 : };
1003 [ + - + + : 828503 : if (impl->backpressure_active_.load() || impl->queue_bytes_ + size > impl->bp_high_ ||
+ + ]
1004 [ - + ]: 414250 : impl->queue_bytes_ + impl->pending_bytes_ + size > impl->bp_limit_) {
1005 : 3 : reject_for_pressure();
1006 : 3 : return false;
1007 : : }
1008 [ - + ]: 414250 : if (!queue_util::try_reserve_write_bytes(impl->queue_bytes_, impl->pending_bytes_, impl->backpressure_active_, size,
1009 : : impl->bp_high_, impl->bp_limit_)) {
1010 : 0 : reject_for_pressure();
1011 : 0 : return false;
1012 : : }
1013 : 414250 : impl->stats_.record_accepted(size);
1014 : :
1015 : 414250 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(data), size]() mutable {
1016 : 414248 : auto impl = self->get_impl();
1017 [ + - + - : 1094386 : if (impl->stop_requested_.load() || impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) ||
+ - ]
1018 [ + + - + : 1094386 : impl->state_.is_state(LinkState::Error) || !impl->remote_endpoint_) {
+ + ]
1019 : 74179 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, size);
1020 : 74179 : impl->stats_.record_failed_send();
1021 : 74179 : return;
1022 : : }
1023 : :
1024 : 340069 : impl->tx_.push_back({std::move(buf), std::nullopt});
1025 : 340069 : impl->observe_queue();
1026 : 340069 : impl->report_backpressure(self, impl->queue_bytes_);
1027 : 340069 : impl->do_write(self);
1028 : 340069 : });
1029 : 414250 : return true;
1030 : : }
1031 : :
1032 : 1 : bool UdpChannel::async_try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
1033 : 1 : auto impl = get_impl();
1034 [ + - + - : 3 : if (impl->stop_requested_.load() || impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) ||
+ - ]
1035 [ + - + - : 3 : impl->state_.is_state(LinkState::Error) || !impl->remote_endpoint_ || !data || data->empty()) {
+ - - + -
+ ]
1036 : 0 : impl->stats_.record_failed_send();
1037 : 0 : return false;
1038 : : }
1039 : 1 : const auto size = data->size();
1040 [ - + ]: 1 : if (size > base::constants::MAX_BUFFER_SIZE) {
1041 : 0 : impl->stats_.record_failed_send();
1042 : 0 : return false;
1043 : : }
1044 : 1 : const auto reject_for_pressure = [impl, size]() {
1045 [ - + ]: 1 : if (impl->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
1046 : 0 : impl->stats_.record_dropped(1, size);
1047 : : } else {
1048 : 1 : impl->stats_.record_failed_send();
1049 : : }
1050 : 2 : };
1051 [ + - - + : 1 : if (impl->backpressure_active_.load() || impl->queue_bytes_ + size > impl->bp_high_ ||
+ - ]
1052 [ # # ]: 0 : impl->queue_bytes_ + impl->pending_bytes_ + size > impl->bp_limit_) {
1053 : 1 : reject_for_pressure();
1054 : 1 : return false;
1055 : : }
1056 [ # # ]: 0 : if (!queue_util::try_reserve_write_bytes(impl->queue_bytes_, impl->pending_bytes_, impl->backpressure_active_, size,
1057 : : impl->bp_high_, impl->bp_limit_)) {
1058 : 0 : reject_for_pressure();
1059 : 0 : return false;
1060 : : }
1061 : 0 : impl->stats_.record_accepted(size);
1062 : :
1063 : 0 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(data), size]() mutable {
1064 : 0 : auto impl = self->get_impl();
1065 [ # # # # : 0 : if (impl->stop_requested_.load() || impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) ||
# # ]
1066 [ # # # # : 0 : impl->state_.is_state(LinkState::Error) || !impl->remote_endpoint_) {
# # ]
1067 : 0 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, size);
1068 : 0 : impl->stats_.record_failed_send();
1069 : 0 : return;
1070 : : }
1071 : :
1072 : 0 : impl->tx_.push_back({std::move(buf), std::nullopt});
1073 : 0 : impl->observe_queue();
1074 : 0 : impl->report_backpressure(self, impl->queue_bytes_);
1075 : 0 : impl->do_write(self);
1076 : 0 : });
1077 : 0 : return true;
1078 : : }
1079 : :
1080 : 409202 : void UdpChannel::on_bytes(OnBytes cb) {
1081 : 409202 : auto shared = interface::share_callback(std::move(cb));
1082 : 409202 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
1083 : 409202 : impl_->on_bytes_ = std::move(shared);
1084 : 409202 : }
1085 : :
1086 : 409261 : void UdpChannel::on_state(OnState cb) {
1087 : 409261 : auto shared = interface::share_callback(std::move(cb));
1088 : 409261 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
1089 : 409261 : impl_->on_state_ = std::move(shared);
1090 : 409261 : }
1091 : :
1092 : 409244 : void UdpChannel::on_backpressure(OnBackpressure cb) {
1093 : 409244 : auto shared = interface::share_callback(std::move(cb));
1094 : 409244 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
1095 : 409244 : impl_->on_bp_ = std::move(shared);
1096 : 409244 : }
1097 : :
1098 : 2 : void UdpChannel::set_backpressure_strategy(base::constants::BackpressureStrategy strategy) {
1099 : 2 : impl_->bp_strategy_.store(strategy, std::memory_order_relaxed);
1100 : 2 : }
1101 : :
1102 : 6 : bool UdpChannel::async_write_to(memory::ConstByteSpan data, const boost::asio::ip::udp::endpoint& destination) {
1103 : 6 : auto impl = get_impl();
1104 [ + + ]: 6 : if (data.empty()) {
1105 : 1 : impl->stats_.record_failed_send();
1106 : 1 : return false;
1107 : : }
1108 [ + + ]: 5 : if (impl->stop_requested_.load()) {
1109 : 1 : impl->stats_.record_failed_send();
1110 : 1 : return false;
1111 : : }
1112 [ + - + - : 4 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
- + - + ]
1113 : 0 : impl->stats_.record_failed_send();
1114 : 0 : return false;
1115 : : }
1116 : :
1117 : 4 : size_t size = data.size();
1118 [ - + ]: 4 : if (size > base::constants::MAX_BUFFER_SIZE) {
1119 : 0 : WIRESTEAD_LOG_ERROR("udp", "write_to", "Write size exceeds maximum allowed");
1120 : 0 : impl->stats_.record_failed_send();
1121 : 0 : return false;
1122 : : }
1123 [ + - + - ]: 4 : if (impl->cfg_.enable_memory_pool && size <= 65536) {
1124 : 4 : memory::PooledBuffer pooled(size, impl->pool_);
1125 [ + - + - ]: 4 : if (pooled.valid()) {
1126 : 4 : base::safe_memory::safe_memcpy(pooled.data(), data.data(), size);
1127 [ - + ]: 4 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queue_bytes_, impl->pending_bytes_,
1128 : 4 : impl->inflight_bytes_, size, impl->bp_limit_)) {
1129 : 0 : impl->stats_.record_failed_send();
1130 : 0 : return false;
1131 : : }
1132 : 4 : impl->stats_.record_accepted(size);
1133 : 4 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(pooled), size, destination]() mutable {
1134 : 4 : auto impl = self->get_impl();
1135 [ + - - + ]: 4 : if (!impl->enqueue_buffer(self, std::move(buf), size, destination)) return;
1136 : 4 : impl->do_write(self);
1137 : : });
1138 : 4 : return true;
1139 : : }
1140 : 4 : }
1141 : :
1142 [ # # ]: 0 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queue_bytes_, impl->pending_bytes_,
1143 : 0 : impl->inflight_bytes_, size, impl->bp_limit_)) {
1144 : 0 : impl->stats_.record_failed_send();
1145 : 0 : return false;
1146 : : }
1147 : 0 : std::vector<uint8_t> copy(data.begin(), data.end());
1148 : 0 : impl->stats_.record_accepted(size);
1149 : 0 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(copy), size, destination]() mutable {
1150 : 0 : auto impl = self->get_impl();
1151 [ # # # # ]: 0 : if (!impl->enqueue_buffer(self, std::move(buf), size, destination)) return;
1152 : 0 : impl->do_write(self);
1153 : : });
1154 : 0 : return true;
1155 : 0 : }
1156 : :
1157 : 78 : bool UdpChannel::async_try_write_to(memory::ConstByteSpan data, const boost::asio::ip::udp::endpoint& destination) {
1158 : 78 : auto impl = get_impl();
1159 [ + - + - : 234 : if (data.empty() || impl->stop_requested_.load() || impl->stopping_.load() ||
+ - ]
1160 [ + - - + : 234 : impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
- + ]
1161 : 0 : impl->stats_.record_failed_send();
1162 : 0 : return false;
1163 : : }
1164 : 78 : const auto size = data.size();
1165 [ - + ]: 78 : if (size > base::constants::MAX_BUFFER_SIZE) {
1166 : 0 : impl->stats_.record_failed_send();
1167 : 0 : return false;
1168 : : }
1169 : 0 : const auto reject_for_pressure = [impl, size]() {
1170 [ # # ]: 0 : if (impl->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
1171 : 0 : impl->stats_.record_dropped(1, size);
1172 : : } else {
1173 : 0 : impl->stats_.record_failed_send();
1174 : : }
1175 : 78 : };
1176 [ + - + - : 156 : if (impl->backpressure_active_.load() || impl->queue_bytes_ + size > impl->bp_high_ ||
- + ]
1177 [ - + ]: 78 : impl->queue_bytes_ + impl->pending_bytes_ + size > impl->bp_limit_) {
1178 : 0 : reject_for_pressure();
1179 : 0 : return false;
1180 : : }
1181 [ - + ]: 78 : if (!queue_util::try_reserve_write_bytes(impl->queue_bytes_, impl->pending_bytes_, impl->backpressure_active_, size,
1182 : : impl->bp_high_, impl->bp_limit_)) {
1183 : 0 : reject_for_pressure();
1184 : 0 : return false;
1185 : : }
1186 : :
1187 : 78 : std::vector<uint8_t> copy(data.begin(), data.end());
1188 : 78 : impl->stats_.record_accepted(size);
1189 : 78 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(copy), size, destination]() mutable {
1190 : 78 : auto impl = self->get_impl();
1191 [ + + + - : 153 : if (impl->stop_requested_.load() || impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) ||
+ - - + ]
1192 : 153 : impl->state_.is_state(LinkState::Error)) {
1193 : 3 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, size);
1194 : 3 : impl->stats_.record_failed_send();
1195 : 3 : return;
1196 : : }
1197 : :
1198 : 150 : impl->tx_.push_back({std::move(buf), destination});
1199 : 75 : impl->observe_queue();
1200 : 75 : impl->report_backpressure(self, impl->queue_bytes_);
1201 : 75 : impl->do_write(self);
1202 : 75 : });
1203 : 78 : return true;
1204 : 78 : }
1205 : :
1206 : : // Takes callback_mtx_ like every other setter here. It previously assigned
1207 : : // without the lock while the strand-confined read site took it, which raced
1208 : : // against a concurrent on_bytes_from() replacement.
1209 : 45 : void UdpChannel::on_bytes_from(OnBytesFrom cb) {
1210 : 45 : auto shared = interface::share_callback(std::move(cb));
1211 : 45 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
1212 : 45 : impl_->on_bytes_from_ = std::move(shared);
1213 : 45 : }
1214 : :
1215 : 0 : boost::asio::ip::udp::endpoint UdpChannel::local_endpoint() const { return get_impl()->local_endpoint_; }
1216 : :
1217 : 44 : boost::asio::any_io_executor UdpChannel::get_executor() { return get_impl()->ioc_->get_executor(); }
1218 : :
1219 : : } // namespace transport
1220 : : } // namespace wirestead
|
