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 <condition_variable>
25 : : #include <cstddef>
26 : : #include <deque>
27 : : #include <memory>
28 : : #include <mutex>
29 : : #include <optional>
30 : : #include <stdexcept>
31 : : #include <stop_token>
32 : : #include <thread>
33 : : #include <type_traits>
34 : : #include <variant>
35 : : #include <vector>
36 : :
37 : : #include "wirestead/base/common.hpp"
38 : : #include "wirestead/base/constants.hpp"
39 : : #include "wirestead/concurrency/io_thread_hook.hpp"
40 : : #include "wirestead/concurrency/thread_safe_state.hpp"
41 : : #include "wirestead/config/validation.hpp"
42 : : #include "wirestead/diagnostics/callback.hpp"
43 : : #include "wirestead/diagnostics/error_handler.hpp"
44 : : #include "wirestead/diagnostics/logger.hpp"
45 : : #include "wirestead/diagnostics/runtime_stats_counter.hpp"
46 : : #include "wirestead/diagnostics/send_accounting.hpp"
47 : : #include "wirestead/memory/memory_pool.hpp"
48 : : #include "wirestead/transport/base/bp_state_machine.hpp"
49 : : #include "wirestead/transport/base/bp_utils.hpp"
50 : : #include "wirestead/transport/base/error_info_holder.hpp"
51 : : #include "wirestead/transport/base/stop_test_hook.hpp"
52 : : #include "wirestead/transport/udp/detail/write_wait.hpp"
53 : : #include "wirestead/wrapper/send_validation.hpp"
54 : :
55 : : namespace wirestead {
56 : : namespace transport {
57 : :
58 : : namespace net = boost::asio;
59 : : using udp = net::ip::udp;
60 : : using base::LinkState;
61 : : using concurrency::AtomicLinkState;
62 : :
63 : : struct UdpChannel::Impl {
64 : : std::unique_ptr<net::io_context> owned_ioc_;
65 : : net::io_context* ioc_;
66 : : bool owns_ioc_{true};
67 : : net::strand<net::io_context::executor_type> strand_;
68 : : std::unique_ptr<net::executor_work_guard<net::io_context::executor_type>> work_guard_;
69 : : std::jthread ioc_thread_;
70 : :
71 : : udp::socket socket_;
72 : : udp::endpoint local_endpoint_;
73 : : udp::endpoint recv_endpoint_;
74 : : std::optional<udp::endpoint> remote_endpoint_;
75 : :
76 : : using BufferVariant =
77 : : std::variant<memory::PooledBuffer, std::vector<uint8_t>, std::shared_ptr<const std::vector<uint8_t>>>;
78 : : using Ledger = diagnostics::SendAccountingLedger;
79 : : Ledger send_accounting_;
80 : : struct TxItem {
81 : : BufferVariant buffer;
82 : : std::optional<udp::endpoint> destination;
83 : : Ledger::Request request;
84 : : std::shared_ptr<detail::UdpWriteWait> session{};
85 : : };
86 : 112 : static size_t item_size(const TxItem& item) {
87 : 224 : return std::visit([](const auto& b) { return queue_util::variant_buffer_size(b); }, item.buffer);
88 : : }
89 : 262252 : static void release_session_bytes(const TxItem& item, bool pending) {
90 [ + + + - : 262252 : if (item.session && item.session->stats)
+ + ]
91 [ + + ]: 79 : queue_util::release_reserved_write_bytes(pending ? item.session->stats->pending : item.session->stats->queued,
92 : : item_size(item));
93 : 262252 : }
94 : 24 : void discard_session_queue(const std::shared_ptr<detail::UdpWriteWait>& session) {
95 : 48 : const auto erase = [&](auto& queue, auto& bytes, bool pending) {
96 [ + + ]: 69 : for (auto it = queue.begin(); it != queue.end();) {
97 [ + + ]: 21 : if (it->session == session) {
98 : 15 : queue_util::release_reserved_write_bytes(bytes, item_size(*it));
99 : 15 : release_session_bytes(*it, pending);
100 : 15 : it = queue.erase(it);
101 : : } else
102 : 6 : ++it;
103 : : }
104 : 48 : };
105 : 24 : erase(tx_, queue_bytes_, false);
106 : 24 : erase(pending_, pending_bytes_, true);
107 : 24 : }
108 : :
109 : : std::array<uint8_t, 65536> rx_{};
110 : : std::deque<TxItem> tx_;
111 : : std::deque<TxItem> pending_;
112 : : std::atomic<size_t> pending_bytes_{0};
113 : : bool writing_{false};
114 : : std::atomic<size_t> queue_bytes_{0};
115 : : // Bytes accepted by a plain async_write_* call but not yet routed onto the
116 : : // strand - reserved via try_reserve_limit_bytes() to close the
117 : : // accept-then-drop race (jwsung91/wirestead#517). inflight_bytes_ mutations
118 : : // and the queue_bytes_/pending_bytes_ increments that promote a
119 : : // reservation both go through write_reserve_mtx_ - see bp_utils.hpp.
120 : : std::atomic<size_t> inflight_bytes_{0};
121 : : std::mutex write_reserve_mtx_;
122 : : config::UdpConfig cfg_;
123 : : // #443: per-channel pool instead of the process-wide GlobalMemoryPool
124 : : // singleton - avoids cross-channel contention on the singleton's bucket
125 : : // mutexes. Capacity is much smaller than the old shared default since
126 : : // it's no longer amortized across every channel in the process.
127 : : // Prefill stays 0. This literal was written while MemoryPool discarded
128 : : // initial_pool_size, so 50 allocated nothing; #575 made the parameter real
129 : : // and turned it into ~1 MiB eagerly allocated per channel at construction.
130 : : // The pool fills as buffers are released.
131 : : memory::MemoryPool pool_{0, 200};
132 : : // Atomic rather than mutex-guarded: read both from the strand (report_backpressure,
133 : : // do_write) and from arbitrary caller threads (the async_try_write_* fast-fail
134 : : // prechecks) - a strand-post here would only protect the former, not the latter (#436).
135 : : std::atomic<base::constants::BackpressureStrategy> bp_strategy_{base::constants::BackpressureStrategy::Reliable};
136 : : size_t bp_high_;
137 : : size_t bp_low_;
138 : : size_t bp_limit_;
139 : : std::atomic<bool> backpressure_active_{false};
140 : : diagnostics::RuntimeStatsCounters stats_;
141 : :
142 : : std::atomic<bool> stop_requested_{false};
143 : : std::atomic<bool> stopping_{false};
144 : : std::atomic<bool> opened_{false};
145 : : std::atomic<bool> connected_{false};
146 : : std::atomic<bool> started_{false};
147 : 479 : AtomicLinkState state_{LinkState::Idle};
148 : : std::atomic<bool> terminal_state_notified_{false};
149 : :
150 : : std::atomic<uint64_t> generation_{0};
151 : : std::mutex submission_mtx_;
152 : : std::mutex stop_mtx_;
153 : : std::condition_variable stop_cv_;
154 : : bool cleanup_done_ = false;
155 : : bool cleanup_started_ = false;
156 : : bool cleanup_finished_ = false;
157 : : size_t pending_io_ = 0; // Strand-confined.
158 : : std::mutex join_mtx_;
159 : : static inline thread_local Impl* active_cleanup_ = nullptr;
160 : :
161 : : std::vector<std::weak_ptr<detail::UdpWriteWait>> write_waits_;
162 : : // Caller holds submission_mtx_. A UDP run cannot change its selected remote.
163 : 272544 : std::optional<wrapper::SendRejection> admission_rejection(bool require_remote,
164 : : std::optional<uint64_t> expected_run = std::nullopt) {
165 [ + + ]: 272544 : if (stop_requested_.load()) {
166 : 66 : std::lock_guard<std::mutex> lock(stop_mtx_);
167 [ + + ]: 66 : return cleanup_done_ ? wrapper::SendRejection::NotStarted : wrapper::SendRejection::Stopping;
168 : 66 : }
169 [ + + ]: 272478 : if (!started_) return wrapper::SendRejection::NotStarted;
170 [ + + + + : 817257 : if ((expected_run && *expected_run != generation_) || !opened_ || state_.is_state(LinkState::Closed) ||
+ - + - ]
171 [ + + + + : 817257 : state_.is_state(LinkState::Error) || (require_remote && !remote_endpoint_))
+ + + + ]
172 : 88 : return wrapper::SendRejection::NotReady;
173 : 272360 : return std::nullopt;
174 : : }
175 : 1930 : void end_write_waits(wrapper::SendRejection reason) {
176 [ + + ]: 2788 : for (auto& weak : write_waits_)
177 [ + + ]: 858 : if (auto wait = weak.lock()) wait->end(reason);
178 : 2788 : std::erase_if(write_waits_, [](const auto& weak) { return weak.expired(); });
179 : 1930 : }
180 : :
181 : : // Caller holds submission_mtx_. For UDP, loss means terminal local socket
182 : : // failure; there is no peer connection or delivery acknowledgement.
183 : 559 : void fail_writes_locked() {
184 : 559 : send_accounting_.end(Ledger::Cause::ConnectionLoss);
185 : 559 : opened_ = false;
186 : 559 : connected_ = false;
187 : 559 : end_write_waits(wrapper::SendRejection::NotReady);
188 : 559 : }
189 : :
190 : 521 : void mark_cleanup_done() {
191 : 521 : detail::stop_test_hook(this, true);
192 [ + + ]: 521 : if (owns_ioc_) work_guard_.reset();
193 : 521 : started_ = false;
194 : : {
195 : 521 : std::lock_guard<std::mutex> lock(stop_mtx_);
196 : 521 : cleanup_done_ = true;
197 : 521 : }
198 : 521 : stop_cv_.notify_all();
199 : 521 : }
200 : :
201 : : struct IoCompletion {
202 : : Impl* impl;
203 : 8886 : ~IoCompletion() {
204 [ + + + + ]: 8886 : if (impl->cleanup_finished_ && impl->pending_io_ == 1) {
205 [ + + ]: 466 : if (auto hook = detail::g_udp_io_completion_hook.load()) hook();
206 : : }
207 [ + + + + : 8886 : if (--impl->pending_io_ == 0 && impl->cleanup_finished_) impl->mark_cleanup_done();
+ + ]
208 : 8886 : }
209 : : };
210 : :
211 : : // Guards on_bytes_/on_bytes_from_/on_state_/on_bp_. Setters (called from
212 : : // any user thread) and the strand-confined read sites below both take
213 : : // this lock; readers copy the callback under lock then invoke the copy
214 : : // outside the lock, matching the pattern already used correctly by
215 : : // TcpClient/UdsClient/both servers (see #436).
216 : : mutable std::mutex callback_mtx_;
217 : : // Shared snapshots: the strand copies one out per received datagram, and a
218 : : // std::function copy allocates whenever the target outgrows its small-object
219 : : // buffer. See interface::SharedCallback.
220 : : interface::SharedCallback<OnBytes> on_bytes_;
221 : : interface::SharedCallback<UdpChannel::OnBytesFrom> on_bytes_from_;
222 : : interface::SharedCallback<OnState> on_state_;
223 : : interface::SharedCallback<OnBackpressure> on_bp_;
224 : :
225 : : ErrorInfoHolder error_info_holder_{"udp"};
226 : :
227 : 161 : explicit Impl(const config::UdpConfig& config)
228 : 161 : : owned_ioc_(std::make_unique<net::io_context>()),
229 : 161 : ioc_(owned_ioc_.get()),
230 : 161 : owns_ioc_(true),
231 : 161 : strand_(ioc_->get_executor()),
232 : 161 : socket_(strand_),
233 : 161 : cfg_(config),
234 : 161 : bp_strategy_(config.backpressure_strategy),
235 : 805 : bp_high_(config.backpressure_threshold) {
236 : 161 : init();
237 : 209 : }
238 : :
239 : 318 : Impl(const config::UdpConfig& config, net::io_context& external_ioc)
240 : 636 : : ioc_(&external_ioc),
241 : 318 : owns_ioc_(false),
242 : 318 : strand_(external_ioc.get_executor()),
243 : 318 : socket_(strand_),
244 : 318 : cfg_(config),
245 : 318 : bp_strategy_(config.backpressure_strategy),
246 : 1908 : bp_high_(config.backpressure_threshold) {
247 : 318 : init();
248 : 334 : }
249 : :
250 : 479 : void init() {
251 : 479 : config::detail::validate(cfg_);
252 : 475 : bp_high_ = cfg_.backpressure_threshold;
253 [ + - ]: 475 : bp_low_ = bp_high_ > 1 ? bp_high_ / 2 : bp_high_;
254 [ - + ]: 475 : if (bp_low_ == 0) bp_low_ = 1;
255 : 475 : bp_limit_ = std::min(std::max(bp_high_ * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
256 : : base::constants::MAX_BUFFER_SIZE);
257 [ - + ]: 475 : if (bp_limit_ < bp_high_) {
258 : 0 : bp_limit_ = bp_high_;
259 : : }
260 : 475 : set_remote_from_config();
261 : 475 : }
262 : :
263 : 475 : ~Impl() {
264 : : try {
265 : 475 : stop_requested_.store(true);
266 : 475 : stopping_.store(true);
267 [ + + - + : 475 : if (owns_ioc_ && work_guard_) {
- + ]
268 : 0 : work_guard_.reset();
269 : : }
270 [ - + ]: 475 : if (ioc_thread_.joinable()) {
271 [ # # ]: 0 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
272 : 0 : ioc_thread_.detach();
273 : : } else {
274 : 0 : ioc_thread_.request_stop();
275 : 0 : ioc_thread_.join();
276 : : }
277 : : }
278 : 475 : perform_stop_cleanup();
279 : 0 : } catch (...) {
280 : 0 : }
281 : 475 : }
282 : :
283 : 505 : void open_socket(std::shared_ptr<UdpChannel> self) {
284 [ + - - + : 514 : if (stopping_.load() || stop_requested_.load()) return;
- + ]
285 : :
286 : 505 : boost::system::error_code ec;
287 : 505 : auto address = net::ip::make_address(cfg_.bind_address, ec);
288 [ - + ]: 505 : if (ec) {
289 : 0 : std::string msg = fmt::format("Invalid bind address: {}", cfg_.bind_address);
290 : 0 : WIRESTEAD_LOG_ERROR("udp", "bind", msg);
291 : 0 : transition_to(LinkState::Error, ec, "bind", msg);
292 : 0 : return;
293 : 0 : }
294 : :
295 : 505 : local_endpoint_ = udp::endpoint(address, cfg_.local_port);
296 : 505 : socket_.open(local_endpoint_.protocol(), ec);
297 [ - + ]: 505 : if (ec) {
298 : 0 : std::string msg = fmt::format("Socket open failed: {}", ec.message());
299 : 0 : WIRESTEAD_LOG_ERROR("udp", "open", msg);
300 : 0 : transition_to(LinkState::Error, ec, "open", msg);
301 : 0 : return;
302 : 0 : }
303 : :
304 [ + + ]: 505 : if (cfg_.reuse_address) {
305 : 2 : socket_.set_option(net::socket_base::reuse_address(true), ec);
306 [ - + ]: 2 : if (ec) {
307 : 0 : std::string msg = fmt::format("Failed to set reuse_address: {}", ec.message());
308 : 0 : WIRESTEAD_LOG_ERROR("udp", "open", msg);
309 : 0 : transition_to(LinkState::Error, ec, "open", msg);
310 : 0 : return;
311 : 0 : }
312 : : }
313 : :
314 [ + + ]: 505 : if (cfg_.enable_broadcast) {
315 : 1 : socket_.set_option(net::socket_base::broadcast(true), ec);
316 [ - + ]: 1 : if (ec) {
317 : 0 : std::string msg = fmt::format("Failed to set broadcast: {}", ec.message());
318 : 0 : WIRESTEAD_LOG_ERROR("udp", "open", msg);
319 : 0 : transition_to(LinkState::Error, ec, "open", msg);
320 : 0 : return;
321 : 0 : }
322 : : }
323 : :
324 : 505 : socket_.bind(local_endpoint_, ec);
325 [ + + ]: 505 : if (ec) {
326 : 5 : std::string msg = fmt::format("Bind failed: {}", ec.message());
327 : 5 : WIRESTEAD_LOG_ERROR("udp", "bind", msg);
328 : 5 : transition_to(LinkState::Error, ec, "bind", msg);
329 : 5 : return;
330 : 5 : }
331 : :
332 : : // After bind, which is what a group join attaches to. Fatal on failure:
333 : : // a receiver that silently did not join looks identical to a sensor that
334 : : // stopped sending, and that is the exact confusion this whole feature is
335 : : // meant to remove.
336 [ + + ]: 500 : if (cfg_.multicast_group) {
337 : 1 : const auto group = net::ip::make_address(*cfg_.multicast_group, ec);
338 [ + - ]: 1 : if (!ec) {
339 [ + - + - : 1 : if (group.is_v4() && cfg_.multicast_interface) {
+ - ]
340 : 1 : const auto iface = net::ip::make_address_v4(*cfg_.multicast_interface, ec);
341 [ + - + - : 1 : if (!ec) socket_.set_option(net::ip::multicast::join_group(group.to_v4(), iface), ec);
+ - ]
342 : : } else {
343 : 0 : socket_.set_option(net::ip::multicast::join_group(group), ec);
344 : : }
345 : : }
346 [ - + ]: 1 : if (ec) {
347 : 0 : std::string msg = fmt::format("Failed to join multicast group {}: {}", *cfg_.multicast_group, ec.message());
348 : 0 : WIRESTEAD_LOG_ERROR("udp", "multicast", msg);
349 : 0 : transition_to(LinkState::Error, ec, "multicast", msg);
350 : 0 : return;
351 : 0 : }
352 : 1 : WIRESTEAD_LOG_INFO("udp", "multicast", fmt::format("Joined multicast group {}", *cfg_.multicast_group));
353 : : }
354 : :
355 : : // Set large OS buffers for UDP to prevent drops unless explicitly configured.
356 : 500 : const int automatic_buf_size = std::max(static_cast<int>(bp_high_), 4 * 1024 * 1024);
357 : 500 : const int recv_buf_size =
358 [ - + ]: 500 : cfg_.receive_buffer_size > 0 ? static_cast<int>(cfg_.receive_buffer_size) : automatic_buf_size;
359 [ - + ]: 500 : const int send_buf_size = cfg_.send_buffer_size > 0 ? static_cast<int>(cfg_.send_buffer_size) : automatic_buf_size;
360 : :
361 : 500 : socket_.set_option(net::socket_base::receive_buffer_size(recv_buf_size), ec);
362 [ - + ]: 500 : if (ec) {
363 : 0 : WIRESTEAD_LOG_WARNING("udp", "open", fmt::format("Failed to set receive buffer size: {}", ec.message()));
364 : 0 : ec.clear();
365 : : }
366 : 500 : socket_.set_option(net::socket_base::send_buffer_size(send_buf_size), ec);
367 [ - + ]: 500 : if (ec) {
368 : 0 : WIRESTEAD_LOG_WARNING("udp", "open", fmt::format("Failed to set send buffer size: {}", ec.message()));
369 : 0 : ec.clear();
370 : : }
371 : :
372 : : {
373 : 500 : std::lock_guard<std::mutex> lock(submission_mtx_);
374 [ + + ]: 500 : if (stop_requested_) return;
375 : 496 : opened_.store(true);
376 : 496 : connected_.store(remote_endpoint_.has_value());
377 : 500 : }
378 [ + + ]: 496 : if (remote_endpoint_) {
379 : 263 : transition_to(LinkState::Connected);
380 : : } else {
381 : 233 : transition_to(LinkState::Listening);
382 : : }
383 : 496 : start_receive(self);
384 : : }
385 : :
386 : 4588 : void start_receive(std::shared_ptr<UdpChannel> self) {
387 [ + - + - : 13726 : if (stopping_.load() || stop_requested_.load() || state_.is_state(LinkState::Closed) ||
+ - ]
388 [ + + + - : 13726 : state_.is_state(LinkState::Error) || !socket_.is_open()) {
- + + + ]
389 : 19 : return;
390 : : }
391 : :
392 : 4569 : ++pending_io_;
393 : 4569 : socket_.async_receive_from(net::buffer(rx_), recv_endpoint_,
394 : 9138 : [self](const boost::system::error_code& ec, std::size_t bytes) {
395 : 4569 : auto impl = self->get_impl();
396 : 4569 : IoCompletion completed{impl};
397 : 4569 : impl->handle_receive(self, ec, bytes);
398 : 4569 : });
399 : : }
400 : :
401 : 4569 : void handle_receive(std::shared_ptr<UdpChannel> self, boost::system::error_code ec, std::size_t bytes) {
402 [ + + ]: 4569 : if (ec == boost::asio::error::operation_aborted) {
403 : 464 : return;
404 : : }
405 : :
406 [ + + + - : 8207 : if (stopping_.load() || stop_requested_.load() || state_.is_state(LinkState::Closed) ||
+ - - + ]
407 : 8207 : state_.is_state(LinkState::Error)) {
408 : 3 : return;
409 : : }
410 : :
411 [ + + ]: 4102 : if (auto hook = detail::g_udp_receive_result_hook.load()) hook(ec);
412 : :
413 [ + - - + : 8204 : if (ec == boost::asio::error::message_size || bytes >= rx_.size()) {
- + ]
414 : 0 : WIRESTEAD_LOG_ERROR("udp", "receive", "Datagram truncated (buffer too small)");
415 : 0 : transition_to(LinkState::Error, ec, "receive", "Datagram truncated (buffer too small)");
416 : 0 : return;
417 : : }
418 : :
419 [ + + ]: 4102 : if (ec) {
420 : 10 : std::string msg = fmt::format("Receive failed: {}", ec.message());
421 : 10 : WIRESTEAD_LOG_ERROR("udp", "receive", msg);
422 : 10 : transition_to(LinkState::Error, ec, "receive", msg);
423 : 10 : return;
424 : 10 : }
425 : :
426 : 4092 : bool learned_remote = false;
427 : : {
428 : 4092 : std::lock_guard<std::mutex> lock(submission_mtx_);
429 [ + + ]: 4092 : if (!remote_endpoint_) {
430 : 170 : remote_endpoint_ = recv_endpoint_;
431 : 170 : connected_.store(true);
432 : 170 : learned_remote = true;
433 : : }
434 : 4092 : }
435 [ + + + - ]: 4092 : if (learned_remote) transition_to(LinkState::Connected);
436 : :
437 : : // #435: once a remote peer is configured or locked in (above), only
438 : : // deliver data from that exact sender through on_bytes() - the
439 : : // point-to-point API. Without this, any other host could send spoofed
440 : : // datagrams after the fact and have them treated as legitimate data.
441 : : // on_bytes_from() is intentionally NOT filtered here: it's the
442 : : // multi-sender API (used by the UdpServer wrapper), which tracks and
443 : : // trusts each sender as its own session by design.
444 [ + - + + ]: 4092 : const bool from_established_remote = remote_endpoint_ && recv_endpoint_ == *remote_endpoint_;
445 : :
446 [ + - ]: 4092 : if (bytes > 0) {
447 : 4092 : stats_.record_received(bytes);
448 : 4092 : interface::SharedCallback<OnBytes> on_bytes;
449 : 4092 : interface::SharedCallback<UdpChannel::OnBytesFrom> on_bytes_from;
450 : : {
451 : 4092 : std::lock_guard<std::mutex> lock(callback_mtx_);
452 : 4092 : on_bytes = on_bytes_;
453 : 4092 : on_bytes_from = on_bytes_from_;
454 : 4092 : }
455 [ + + ]: 7961 : if (on_bytes && from_established_remote &&
456 [ + + + + : 11831 : !diagnostics::invoke_callback("udp", "on_bytes", on_bytes, memory::ConstByteSpan(rx_.data(), bytes)) &&
- + ]
457 : 2 : cfg_.stop_on_callback_exception) {
458 : 0 : self->stop();
459 : 0 : return;
460 : : }
461 : 4092 : if (!diagnostics::invoke_callback("udp", "on_bytes_from", on_bytes_from, memory::ConstByteSpan(rx_.data(), bytes),
462 [ + + - + ]: 8186 : recv_endpoint_) &&
463 : 2 : cfg_.stop_on_callback_exception) {
464 : 0 : self->stop();
465 : 0 : return;
466 : : }
467 : 4092 : }
468 : :
469 : 4092 : start_receive(self);
470 : : }
471 : :
472 : 272299 : queue_util::BackpressureFields bp_fields() {
473 : 272299 : return queue_util::BackpressureFields{queue_bytes_,
474 : 272299 : pending_bytes_,
475 : 272299 : backpressure_active_,
476 : 272299 : bp_high_,
477 : 272299 : bp_low_,
478 : 272299 : bp_limit_,
479 : 272299 : bp_strategy_.load(std::memory_order_relaxed),
480 : 272299 : &write_reserve_mtx_};
481 : : }
482 : :
483 : : // Drops all queued (tx_) and pending (pending_, reliable-mode overflow) writes and clears
484 : : // backpressure, notifying any waiter directly. Mirrors perform_stop_cleanup()'s approach
485 : : // rather than calling report_backpressure(), because report_backpressure() flushes pending_
486 : : // back into tx_ and can immediately re-arm backpressure_active_ if enough was queued there -
487 : : // which would leave a Reliable-mode sender blocked in send_blocking()'s bp_cv_ wait forever
488 : : // since nothing will ever call do_write() again once the channel has stopped/errored (#427).
489 : 615 : void drain_queue_and_clear_backpressure() {
490 : 615 : interface::SharedCallback<OnBackpressure> on_bp;
491 : : {
492 : 615 : std::lock_guard<std::mutex> lock(callback_mtx_);
493 : 615 : on_bp = on_bp_;
494 : 615 : }
495 : 615 : static const OnBackpressure kNoCallback;
496 : 615 : auto f = bp_fields();
497 [ + + + - ]: 615 : queue_util::drain_and_clear_backpressure(f, on_bp ? *on_bp : kNoCallback, [&]() {
498 [ + - + + ]: 262845 : for (const auto& item : tx_) release_session_bytes(item, false);
499 [ + - + + ]: 622 : for (const auto& item : pending_) release_session_bytes(item, true);
500 : 615 : tx_.clear();
501 : 615 : queue_bytes_ = 0;
502 : 615 : pending_.clear();
503 : 615 : pending_bytes_ = 0;
504 : 615 : });
505 : 615 : }
506 : :
507 : 270830 : void do_write(std::shared_ptr<UdpChannel> self) {
508 [ + + + + : 270844 : if (writing_ || tx_.empty()) return;
+ + ]
509 [ + + + - : 8661 : if (stop_requested_.load() || stopping_.load() || state_.is_state(LinkState::Closed) ||
+ - - + ]
510 : 8661 : state_.is_state(LinkState::Error)) {
511 : 1 : writing_ = false;
512 : 1 : drain_queue_and_clear_backpressure();
513 : 1 : return;
514 : : }
515 : :
516 [ + - - + : 4330 : if (stop_requested_.load() || !opened_.load()) return;
- + ]
517 [ + - + - : 4330 : while (!tx_.empty() && !send_accounting_.contains(tx_.front().request)) {
- + - + ]
518 : 0 : queue_util::release_reserved_write_bytes(queue_bytes_, item_size(tx_.front()));
519 : 0 : release_session_bytes(tx_.front(), false);
520 : 0 : tx_.pop_front();
521 : : }
522 [ - + ]: 4330 : if (tx_.empty()) {
523 : 0 : report_backpressure(self, queue_bytes_);
524 : 0 : return;
525 : : }
526 : 4330 : auto current = std::move(tx_.front());
527 : 4330 : tx_.pop_front();
528 : :
529 [ + + ]: 4330 : const auto& dest_endpoint = current.destination ? current.destination : remote_endpoint_;
530 : :
531 [ - + ]: 4330 : if (!dest_endpoint) {
532 : 0 : WIRESTEAD_LOG_WARNING("udp", "write", "Remote endpoint not set; dropping write request");
533 : 0 : send_accounting_.discard(current.request, Ledger::Cause::ConnectionLoss);
534 : 0 : const auto bytes = std::visit([](const auto& b) { return queue_util::variant_buffer_size(b); }, current.buffer);
535 : 0 : queue_util::release_reserved_write_bytes(queue_bytes_, bytes);
536 : 0 : release_session_bytes(current, false);
537 : 0 : writing_ = false;
538 : 0 : do_write(self); // Process next in queue
539 : 0 : return;
540 : : }
541 : :
542 : : // The ledger, not the admission mutex, fences handoff against stop, loss
543 : : // and session expiry: begin() fails for a request they already terminated.
544 : : // Taking the admission mutex per datagram contends with every producer.
545 [ + - - + ]: 4330 : if (!send_accounting_.begin(current.request)) {
546 : 0 : queue_util::release_reserved_write_bytes(queue_bytes_, item_size(current));
547 : 0 : release_session_bytes(current, false);
548 : 0 : do_write(self);
549 : 0 : return;
550 : : }
551 : 4330 : writing_ = true;
552 : :
553 [ + - ]: 4330 : auto bytes_queued = std::visit(
554 : 4330 : [](auto&& buf) -> size_t {
555 : : using Buffer = std::decay_t<decltype(buf)>;
556 : : if constexpr (std::is_same_v<Buffer, std::shared_ptr<const std::vector<uint8_t>>>) {
557 [ + - ]: 35 : return buf ? buf->size() : 0;
558 : : } else {
559 : 4295 : return buf.size();
560 : : }
561 : : },
562 : : current.buffer);
563 : :
564 : 4330 : const auto request = current.request;
565 : 4330 : const auto generation = generation_.load();
566 : 4317 : auto on_write = [self, bytes_queued, request, generation, session = current.session](boost::system::error_code ec,
567 : : std::size_t bytes_written) {
568 : 4317 : auto impl = self->get_impl();
569 : : {
570 [ + + - + : 4317 : if (!ec && bytes_written != bytes_queued) ec = net::error::message_size;
- + ]
571 : : // Only a failure must close admission; success needs just the ledger.
572 : 4317 : std::unique_lock<std::mutex> lock(impl->submission_mtx_, std::defer_lock);
573 [ + + + - ]: 4317 : if (ec) lock.lock();
574 [ - + ]: 4317 : if (generation != impl->generation_) return;
575 : 4317 : impl->send_accounting_.complete(request, bytes_written);
576 [ + + + - : 4317 : if (ec && !impl->stop_requested_) {
+ + ]
577 : 25 : impl->fail_writes_locked();
578 [ - + ]: 25 : if (ec == net::error::operation_aborted) ec = net::error::connection_aborted;
579 : : }
580 : 4317 : }
581 : 4317 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, bytes_queued);
582 [ + + + - : 4317 : if (session && session->stats) queue_util::release_reserved_write_bytes(session->stats->queued, bytes_queued);
+ + ]
583 : 4317 : impl->report_backpressure(self, impl->queue_bytes_);
584 : :
585 [ + + + - : 8566 : if (impl->stop_requested_.load() || impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) ||
+ - - + ]
586 : 8566 : impl->state_.is_state(LinkState::Error)) {
587 : 68 : impl->writing_ = false;
588 : 68 : impl->drain_queue_and_clear_backpressure();
589 : 68 : return;
590 : : }
591 : :
592 [ + + ]: 4249 : if (ec) {
593 : 25 : std::string msg = fmt::format("Send failed: {}", ec.message());
594 : 25 : WIRESTEAD_LOG_ERROR("udp", "write", msg);
595 : 25 : impl->transition_to(LinkState::Error, ec, "write", msg);
596 : 25 : impl->writing_ = false;
597 : : // do_write() will never run again to reach the "already in Error" cleanup above, so
598 : : // drain everything and clear backpressure here directly.
599 : 25 : impl->drain_queue_and_clear_backpressure();
600 : 25 : return;
601 : 25 : }
602 : :
603 : 4224 : impl->stats_.record_sent(bytes_written);
604 [ + + + - : 4224 : if (session && session->stats) session->stats->counters.record_sent(bytes_written);
+ + ]
605 : 4224 : impl->writing_ = false;
606 : 4224 : impl->do_write(self);
607 : 4330 : };
608 : :
609 : 4330 : ++pending_io_;
610 : : try {
611 [ + + - + ]: 4330 : if (auto hook = detail::g_udp_write_initiation_hook.load()) hook();
612 : 4317 : std::visit(
613 [ + - ]: 8634 : [&](auto&& buf) {
614 : : using T = std::decay_t<decltype(buf)>;
615 : :
616 : 12481 : auto* data_ptr = [&]() {
617 : : if constexpr (std::is_same_v<T, std::shared_ptr<const std::vector<uint8_t>>>) {
618 : 33 : return buf->data();
619 : : } else {
620 : 4284 : return buf.data();
621 : : }
622 : 4317 : }();
623 : :
624 : 12481 : auto size = [&]() {
625 : : if constexpr (std::is_same_v<T, std::shared_ptr<const std::vector<uint8_t>>>) {
626 : 33 : return buf->size();
627 : : } else {
628 : 4284 : return buf.size();
629 : : }
630 : 4317 : }();
631 : :
632 : 4317 : socket_.async_send_to(net::buffer(data_ptr, size), *dest_endpoint,
633 : 12951 : [self, buf_captured = std::move(buf), on_write = std::move(on_write)](
634 : : const boost::system::error_code& ec, std::size_t bytes) mutable {
635 : 12951 : auto finish = [self, buffer = std::move(buf_captured),
636 : 4317 : on_write = std::move(on_write), ec, bytes]() mutable {
637 : 4317 : IoCompletion completed{self->get_impl()};
638 : : // Release pooled storage before signalling I/O completion.
639 : 4317 : auto owned_buffer = std::move(buffer);
640 : : (void)owned_buffer;
641 [ - + - - : 4317 : if (auto hook = detail::g_udp_write_completion_hook.load()) hook();
- + - - -
+ - - ]
642 : 4317 : on_write(ec, bytes);
643 : 4317 : };
644 [ - + + + : 4317 : if (auto hook = detail::g_udp_defer_write_completion_hook.load()) {
+ + ]
645 [ - - + - : 9 : auto pending = std::make_shared<decltype(finish)>(std::move(finish));
+ - ]
646 : 18 : hook([pending] { (*pending)(); });
647 : 9 : } else {
648 : 4308 : finish();
649 : : }
650 : 4317 : });
651 : 4317 : },
652 : 4317 : std::move(current.buffer));
653 : 13 : } catch (...) {
654 [ + + + - : 13 : if (current.session && current.session->stats)
+ + ]
655 : 3 : queue_util::release_reserved_write_bytes(current.session->stats->queued, bytes_queued);
656 : 13 : --pending_io_;
657 : : {
658 : 13 : std::lock_guard<std::mutex> lock(submission_mtx_);
659 : 13 : fail_writes_locked();
660 : 13 : }
661 : 13 : writing_ = false;
662 : 13 : transition_to(LinkState::Error, make_error_code(boost::system::errc::no_buffer_space), "write",
663 : : "Failed to initiate datagram write");
664 : 13 : return;
665 : 13 : }
666 [ + + + - ]: 4317 : if (auto hook = detail::g_udp_write_started_hook.load()) hook();
667 : 4343 : }
668 : :
669 : 521 : void close_socket() {
670 : 521 : boost::system::error_code ec;
671 : 521 : socket_.cancel(ec);
672 : 521 : socket_.close(ec);
673 : 521 : }
674 : :
675 : 1692 : void notify_state() {
676 : 1692 : interface::SharedCallback<OnState> on_state;
677 : : {
678 : 1692 : std::lock_guard<std::mutex> lock(callback_mtx_);
679 : 1692 : on_state = on_state_;
680 : 1692 : }
681 [ + + ]: 1692 : if (!on_state) return;
682 : : try {
683 : 1591 : (*on_state)(state_.get());
684 : 7 : } catch (const std::exception& e) {
685 : 6 : WIRESTEAD_LOG_ERROR("udp", "on_state", fmt::format("Exception in state callback: {}", e.what()));
686 : 7 : } catch (...) {
687 : 1 : WIRESTEAD_LOG_ERROR("udp", "on_state", "Unknown exception in state callback");
688 : 1 : }
689 : 1692 : }
690 : :
691 : 270846 : void report_backpressure(std::shared_ptr<UdpChannel> self, size_t queued_bytes) {
692 [ + + ]: 270846 : if (stop_requested_.load()) return;
693 : 270771 : observe_queue();
694 : :
695 : 270771 : interface::SharedCallback<OnBackpressure> on_bp;
696 : : {
697 : 270771 : std::lock_guard<std::mutex> lock(callback_mtx_);
698 : 270771 : on_bp = on_bp_;
699 : 270771 : }
700 : 270771 : static const OnBackpressure kNoCallback;
701 : :
702 : 270771 : auto f = bp_fields();
703 [ + + + - ]: 271423 : queue_util::report_backpressure(
704 : 652 : f, queued_bytes, on_bp ? *on_bp : kNoCallback, stats_,
705 : 0 : [&]() -> size_t {
706 : : // Flush pending_ → tx_
707 : 123 : const size_t moved = pending_bytes_.exchange(0);
708 [ + + ]: 190 : while (!pending_.empty()) {
709 : 67 : auto& item = pending_.front();
710 [ + + + - : 67 : if (item.session && item.session->stats) {
+ + ]
711 : 9 : queue_util::release_reserved_write_bytes(item.session->stats->pending, item_size(item));
712 : 9 : item.session->stats->queued.fetch_add(item_size(item));
713 : 9 : item.session->stats->observe_queue();
714 : : }
715 : 67 : tx_.emplace_back(std::move(item));
716 : 67 : pending_.pop_front();
717 : : }
718 : 123 : return moved;
719 : : },
720 : 270771 : [&]() {
721 : : // Post-flush sample: queue_bytes_ has already been updated by the
722 : : // time this kick runs, unlike the flush hook above (#434).
723 : 123 : observe_queue();
724 [ - + - - ]: 123 : if (!writing_) do_write(self);
725 : 123 : });
726 : 270771 : }
727 : :
728 : 537476 : void observe_queue() {
729 : 1612428 : stats_.observe_queue(queue_bytes_.load(std::memory_order_relaxed) + pending_bytes_.load(std::memory_order_relaxed));
730 : 537476 : }
731 : :
732 : 1496 : bool enqueue_buffer(std::shared_ptr<UdpChannel> self, BufferVariant&& buffer, size_t size, Ledger::Request request,
733 : : std::optional<udp::endpoint> dest = std::nullopt,
734 : : std::shared_ptr<detail::UdpWriteWait> session = {}) {
735 [ + + + + : 1918 : if (stopping_.load() || stop_requested_.load() || state_.is_state(LinkState::Closed) ||
+ - + + ]
736 : 1918 : state_.is_state(LinkState::Error)) {
737 : 1104 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, size);
738 : 1104 : stats_.record_failed_send();
739 [ + + + - : 1104 : if (session && session->stats) session->stats->counters.record_failed_send();
+ + ]
740 : 1104 : return false;
741 : : }
742 [ + - - + ]: 392 : if (!send_accounting_.contains(request)) {
743 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, size);
744 : 0 : report_backpressure(self, queue_bytes_);
745 : 0 : return false;
746 : : }
747 : :
748 : 392 : auto f = bp_fields();
749 : 392 : queue_util::DropAccounting dropped;
750 : : // TxItem carries a destination endpoint alongside the BufferVariant that
751 : : // decide_enqueue()'s BestEffort trim needs to visit - project onto
752 : : // `.buffer` rather than the item itself (#434).
753 : 392 : auto decision = queue_util::decide_enqueue(
754 : 392 : f, size, tx_, dropped, [](TxItem& item) -> BufferVariant& { return item.buffer; },
755 : : [this](const TxItem& item) {
756 : : send_accounting_.discard(item.request, Ledger::Cause::QueuePressure);
757 : : release_session_bytes(item, false);
758 : : if (item.session && item.session->stats) item.session->stats->counters.record_dropped(1, item_size(item));
759 : : });
760 [ - + ]: 392 : if (dropped.any()) {
761 : 0 : stats_.record_dropped(dropped.messages, dropped.bytes);
762 : : }
763 : :
764 [ - + ]: 392 : if (decision == queue_util::EnqueueDecision::Rejected) {
765 : 0 : send_accounting_.discard(request, Ledger::Cause::QueuePressure);
766 : 0 : WIRESTEAD_LOG_ERROR("udp", "write",
767 : : fmt::format("Queue limit exceeded ({} bytes)", queue_bytes_ + pending_bytes_ + size));
768 : : // Always reporting here (rather than only for the non-Reliable-pending
769 : : // rejection path, as the pre-#434 code did) is a no-op in practice for
770 : : // the Reliable+pending case: queued_bytes is already >= bp_high_ or
771 : : // this rejection couldn't have happened, so report_backpressure()'s
772 : : // OFF-transition check (<= bp_low_) can't fire here either way.
773 : : // #448: record as dropped so it's reflected in RuntimeStats instead of
774 : : // silently vanishing after being counted as accepted.
775 : 0 : stats_.record_dropped(1, size);
776 [ # # # # : 0 : if (session && session->stats) session->stats->counters.record_dropped(1, size);
# # # # ]
777 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, size);
778 : 0 : report_backpressure(self, queue_bytes_ + size);
779 : 0 : return false;
780 : : }
781 : :
782 [ + + ]: 392 : if (decision == queue_util::EnqueueDecision::Pending) {
783 : 77 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, pending_bytes_, inflight_bytes_, size);
784 [ + + + - : 77 : if (session && session->stats) {
+ + ]
785 : 19 : session->stats->pending.fetch_add(size);
786 : 19 : session->stats->observe_queue();
787 : : }
788 : 154 : pending_.push_back({std::move(buffer), dest, request, std::move(session)});
789 : 77 : observe_queue();
790 : 77 : return true;
791 : : }
792 : :
793 : 315 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, queue_bytes_, inflight_bytes_, size);
794 [ + + + - : 315 : if (session && session->stats) {
+ + ]
795 : 111 : session->stats->queued.fetch_add(size);
796 : 111 : session->stats->observe_queue();
797 : : }
798 : 630 : tx_.push_back({std::move(buffer), dest, request, std::move(session)});
799 : 315 : observe_queue();
800 : 315 : report_backpressure(self, queue_bytes_);
801 : 315 : return true;
802 : 392 : }
803 : :
804 : 983 : void set_remote_from_config() {
805 [ + + - + : 983 : if (!cfg_.remote_address || !cfg_.remote_port) return;
+ + ]
806 : 498 : boost::system::error_code ec;
807 : 498 : auto addr = net::ip::make_address(*cfg_.remote_address, ec);
808 [ - + ]: 498 : if (ec) {
809 : 0 : throw std::runtime_error("Invalid remote address: " + *cfg_.remote_address);
810 : : }
811 : 498 : remote_endpoint_ = udp::endpoint(addr, *cfg_.remote_port);
812 : : }
813 : :
814 : 1746 : void transition_to(LinkState target, const boost::system::error_code& ec = {}, std::string_view operation = {},
815 : : std::string_view msg = {}) {
816 [ - + ]: 1746 : if (ec == net::error::operation_aborted) {
817 : 0 : return;
818 : : }
819 : :
820 : 1746 : const auto current = state_.get();
821 [ + - + + ]: 1746 : if ((current == LinkState::Closed || current == LinkState::Error) &&
822 [ - + - - ]: 54 : (target == LinkState::Closed || target == LinkState::Error)) {
823 : 54 : return;
824 : : }
825 : :
826 [ + + ]: 1692 : if (target == LinkState::Error) {
827 [ - + + - ]: 54 : error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, operation,
828 [ - - - + : 108 : ec, msg.empty() ? std::string_view(ec.message()) : msg, static_cast<bool>(ec), 0);
- - ]
829 : : }
830 : :
831 [ + + + + ]: 1692 : if (target == LinkState::Closed || target == LinkState::Error) {
832 [ - + ]: 521 : if (terminal_state_notified_.exchange(true)) {
833 : 0 : return;
834 : : }
835 [ - + ]: 1171 : } else if (current == target) {
836 : 0 : return;
837 : : }
838 : :
839 : : {
840 : 1692 : std::lock_guard<std::mutex> lock(submission_mtx_);
841 [ + + + + : 1692 : if (target == LinkState::Closed || target == LinkState::Error) fail_writes_locked();
+ - ]
842 : 1692 : state_.set(target);
843 : 1692 : }
844 [ + + + + ]: 1692 : if (target == LinkState::Closed || target == LinkState::Error) drain_queue_and_clear_backpressure();
845 : 1692 : notify_state();
846 : : }
847 : :
848 : 996 : void perform_stop_cleanup() {
849 [ + + ]: 996 : if (cleanup_started_) return;
850 : 521 : cleanup_started_ = true;
851 : : struct CompletionSignal {
852 : : Impl* impl;
853 : : Impl* previous;
854 : 521 : ~CompletionSignal() {
855 : 521 : impl->cleanup_finished_ = true;
856 [ + + ]: 521 : if (impl->pending_io_ == 0) impl->mark_cleanup_done();
857 : 521 : active_cleanup_ = previous;
858 : 521 : }
859 : 521 : } completion{this, active_cleanup_};
860 : 521 : active_cleanup_ = this;
861 : : try {
862 : 521 : close_socket();
863 : 521 : writing_ = false;
864 : 521 : interface::SharedCallback<OnBackpressure> on_bp;
865 : : {
866 : 521 : std::lock_guard<std::mutex> lock(callback_mtx_);
867 : 521 : on_bp = on_bp_;
868 : 521 : }
869 : 521 : static const OnBackpressure kNoCallback;
870 : 521 : auto f = bp_fields();
871 [ + + + - ]: 521 : queue_util::drain_and_clear_backpressure(f, on_bp ? *on_bp : kNoCallback, [&]() {
872 [ - - - + ]: 521 : for (const auto& item : tx_) release_session_bytes(item, false);
873 [ - - - + ]: 521 : for (const auto& item : pending_) release_session_bytes(item, true);
874 : 521 : tx_.clear();
875 : 521 : queue_bytes_ = 0;
876 : 521 : pending_.clear();
877 : 521 : pending_bytes_ = 0;
878 : 521 : });
879 : 521 : connected_.store(false);
880 : 521 : opened_.store(false);
881 : 521 : transition_to(LinkState::Closed);
882 : : {
883 : 521 : std::lock_guard<std::mutex> lock(callback_mtx_);
884 : 521 : on_bytes_ = nullptr;
885 : 521 : on_bytes_from_ = nullptr;
886 : 521 : on_state_ = nullptr;
887 : 521 : on_bp_ = nullptr;
888 : 521 : }
889 : 521 : } catch (...) {
890 : 0 : }
891 : 521 : }
892 : :
893 : 1490 : void join_ioc_thread(bool allow_detach) {
894 [ + + + + : 1490 : if (!owns_ioc_ || !ioc_thread_.joinable()) {
+ + ]
895 : 1331 : return;
896 : : }
897 : :
898 [ - + ]: 159 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
899 [ # # ]: 0 : if (allow_detach) {
900 : 0 : ioc_thread_.detach();
901 : : }
902 : 0 : return;
903 : : }
904 : :
905 : : try {
906 : 159 : ioc_thread_.join();
907 : 0 : } catch (...) {
908 : 0 : }
909 : : }
910 : : };
911 : :
912 : 161 : std::shared_ptr<UdpChannel> UdpChannel::create(const config::UdpConfig& cfg) {
913 : 161 : return std::shared_ptr<UdpChannel>(new UdpChannel(cfg));
914 : : }
915 : :
916 : 318 : std::shared_ptr<UdpChannel> UdpChannel::create(const config::UdpConfig& cfg, net::io_context& ioc) {
917 : 318 : return std::shared_ptr<UdpChannel>(new UdpChannel(cfg, ioc));
918 : : }
919 : :
920 : 164 : UdpChannel::UdpChannel(const config::UdpConfig& cfg) : impl_(std::make_unique<Impl>(cfg)) {}
921 : :
922 : 319 : UdpChannel::UdpChannel(const config::UdpConfig& cfg, net::io_context& ioc) : impl_(std::make_unique<Impl>(cfg, ioc)) {}
923 : :
924 : 950 : UdpChannel::~UdpChannel() {
925 [ + - ]: 475 : if (impl_) stop();
926 : 950 : }
927 : :
928 : 0 : UdpChannel::UdpChannel(UdpChannel&&) noexcept = default;
929 : 0 : UdpChannel& UdpChannel::operator=(UdpChannel&&) noexcept = default;
930 : :
931 : 508 : void UdpChannel::start() {
932 : 508 : auto impl = get_impl();
933 [ - + ]: 508 : if (impl->started_) return;
934 [ - + - - ]: 508 : if (!impl->cfg_.is_valid()) throw std::runtime_error("Invalid UDP configuration");
935 [ - + ]: 508 : if (impl->ioc_thread_.joinable()) impl->join_ioc_thread(false);
936 [ + + + + : 508 : if (impl->owns_ioc_ && impl->ioc_->stopped()) impl->ioc_->restart();
+ + ]
937 : :
938 : 508 : const auto generation = impl->generation_.fetch_add(1) + 1;
939 : : {
940 : 508 : std::lock_guard<std::mutex> lock(impl->submission_mtx_);
941 : 508 : impl->remote_endpoint_.reset();
942 : 508 : impl->set_remote_from_config();
943 : 508 : impl->inflight_bytes_ = 0;
944 : 508 : }
945 : : {
946 : 508 : std::lock_guard<std::mutex> lock(impl->stop_mtx_);
947 : 508 : impl->cleanup_done_ = false;
948 : 508 : }
949 : 508 : impl->cleanup_started_ = false;
950 : 508 : impl->cleanup_finished_ = false;
951 : 508 : impl->stop_requested_ = false;
952 : 508 : impl->stopping_ = false;
953 : 508 : impl->terminal_state_notified_ = false;
954 : 508 : impl->connected_ = false;
955 : 508 : impl->opened_ = false;
956 : 508 : impl->writing_ = false;
957 : 508 : impl->queue_bytes_ = 0;
958 : 508 : impl->pending_bytes_ = 0;
959 : 508 : impl->backpressure_active_ = false;
960 : 508 : impl->state_.set(LinkState::Idle);
961 : 508 : impl->started_ = true;
962 : :
963 [ + + ]: 508 : if (impl->owns_ioc_) {
964 : : impl->work_guard_ =
965 : 159 : std::make_unique<net::executor_work_guard<net::io_context::executor_type>>(impl->ioc_->get_executor());
966 : 318 : impl->ioc_thread_ = std::jthread([impl](std::stop_token st) {
967 : 159 : wirestead::concurrency::run_io_thread_init();
968 : : try {
969 : 159 : std::stop_callback cb(st, [impl] { impl->ioc_->stop(); });
970 : 159 : impl->ioc_->run();
971 : 159 : } catch (...) {
972 : 0 : }
973 : 318 : });
974 : : }
975 : 508 : net::post(impl->strand_, [self = shared_from_this(), generation] {
976 : 508 : auto impl = self->get_impl();
977 [ + + - + : 508 : if (impl->stop_requested_ || generation != impl->generation_) return;
+ + ]
978 : 505 : impl->transition_to(LinkState::Connecting);
979 : 505 : impl->open_socket(self);
980 : : });
981 : : }
982 : :
983 : 1568 : void UdpChannel::stop() {
984 : 1568 : auto impl = get_impl();
985 [ + + ]: 1645 : if (Impl::active_cleanup_ == impl) return;
986 : 1567 : const bool on_executor = impl->ioc_->get_executor().running_in_this_thread();
987 : 1567 : bool inline_cleanup = false;
988 : : {
989 : : // Accepted writes are posted before this cleanup request.
990 : 1567 : std::lock_guard<std::mutex> lock(impl->submission_mtx_);
991 [ + + ]: 1567 : if (!impl->stop_requested_.exchange(true)) {
992 : 521 : impl->send_accounting_.end(Impl::Ledger::Cause::ExplicitStop);
993 : 521 : impl->end_write_waits(wrapper::SendRejection::CancelledWhileWaiting);
994 : 521 : impl->stopping_ = true;
995 [ + + ]: 1042 : if (impl->generation_.load() == 0) {
996 : 13 : inline_cleanup = true;
997 : : } else {
998 : 508 : auto self = weak_from_this().lock();
999 : 1016 : net::post(impl->strand_, [self, impl] { impl->perform_stop_cleanup(); });
1000 : 508 : }
1001 : : }
1002 : 1567 : }
1003 [ + + + - ]: 1567 : if (inline_cleanup) impl->perform_stop_cleanup();
1004 [ + + ]: 1567 : if (on_executor) return;
1005 : 1490 : detail::stop_test_hook(impl, false);
1006 : : {
1007 : 1490 : std::unique_lock<std::mutex> lock(impl->stop_mtx_);
1008 : 3446 : impl->stop_cv_.wait(lock, [impl] { return impl->cleanup_done_; });
1009 : 1490 : }
1010 : 1490 : std::lock_guard<std::mutex> join_lock(impl->join_mtx_);
1011 : 1490 : impl->join_ioc_thread(false);
1012 : 1490 : }
1013 : :
1014 : 487 : bool UdpChannel::is_connected() const { return get_impl()->connected_.load(); }
1015 : 157 : bool UdpChannel::is_backpressure_active() const { return get_impl()->backpressure_active_.load(); }
1016 : 1902 : std::optional<size_t> UdpChannel::write_queue_limit() const { return get_impl()->bp_limit_; }
1017 : 1941 : wrapper::RuntimeStats UdpChannel::stats() const {
1018 : 1941 : auto impl = get_impl();
1019 : 5823 : auto result = impl->stats_.snapshot(impl->queue_bytes_.load(std::memory_order_relaxed),
1020 : : impl->pending_bytes_.load(std::memory_order_relaxed),
1021 : 1941 : impl->backpressure_active_.load(std::memory_order_relaxed));
1022 : 1941 : result.send_accounting = impl->send_accounting_.snapshot();
1023 : 1941 : return result;
1024 : : }
1025 : 214 : void UdpChannel::reset_stats() {
1026 : 214 : auto impl = get_impl();
1027 : 214 : std::lock_guard<std::mutex> lock(impl->submission_mtx_);
1028 : 214 : impl->send_accounting_.reset();
1029 [ + + ]: 242 : for (const auto& weak : impl->write_waits_) {
1030 [ + + + - : 28 : if (auto wait = weak.lock(); wait && wait->stats)
+ + ]
1031 : 70 : wait->stats->counters.reset(wait->stats->queued.load() + wait->stats->pending.load());
1032 : : }
1033 : 214 : impl->stats_.reset(impl->queue_bytes_.load(std::memory_order_relaxed) +
1034 : 428 : impl->pending_bytes_.load(std::memory_order_relaxed));
1035 : 214 : }
1036 : :
1037 : 483 : std::optional<diagnostics::ErrorInfo> UdpChannel::last_error_info() const {
1038 : 483 : return get_impl()->error_info_holder_.last_error_info();
1039 : : }
1040 : :
1041 : 296 : std::shared_ptr<detail::UdpWriteWait> UdpChannel::capture_write_wait(bool require_remote, bool track_session) {
1042 : 296 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
1043 [ + - - + ]: 296 : if (impl_->admission_rejection(require_remote)) return {};
1044 : : auto wait = std::make_shared<detail::UdpWriteWait>(
1045 : 592 : detail::UdpWriteWait{impl_->generation_.load(), require_remote, std::nullopt, {}});
1046 [ + + + - ]: 296 : if (track_session) wait->stats = std::make_shared<detail::UdpSessionStats>();
1047 : 352 : std::erase_if(impl_->write_waits_, [](const auto& weak) { return weak.expired(); });
1048 : 296 : impl_->write_waits_.push_back(wait);
1049 : 296 : return wait;
1050 : 296 : }
1051 : 268 : wrapper::RuntimeStats UdpChannel::session_stats(const std::shared_ptr<detail::UdpWriteWait>& wait) const {
1052 [ + - - + : 268 : if (!wait || !wait->stats) return {};
- + ]
1053 : 268 : auto impl = get_impl();
1054 : 1072 : auto snapshot = wait->stats->counters.snapshot(wait->stats->queued.load(), wait->stats->pending.load(),
1055 : 268 : impl->backpressure_active_.load());
1056 : : // Pressure is shared by every destination on this socket, not a peer queue policy.
1057 : 268 : snapshot.backpressure_events = impl->stats_.backpressure_events.load();
1058 : 268 : snapshot.send_accounting = impl->send_accounting_.snapshot(wait->stats->accounting);
1059 : 268 : return snapshot;
1060 : : }
1061 : :
1062 : 24 : void UdpChannel::expire_session(const std::shared_ptr<detail::UdpWriteWait>& wait) {
1063 [ - + ]: 24 : if (!wait) return;
1064 : 24 : auto impl = get_impl();
1065 : 24 : std::lock_guard<std::mutex> lock(impl->submission_mtx_);
1066 : 24 : wait->end(wrapper::SendRejection::NotReady);
1067 [ - + ]: 24 : if (!wait->stats) return;
1068 : 24 : impl->send_accounting_.discard_waiting(wait->stats->accounting, Impl::Ledger::Cause::SessionExpiry);
1069 : 24 : const auto generation = impl->generation_.load();
1070 : 24 : net::post(impl->strand_, [self = shared_from_this(), wait, generation] {
1071 : 24 : auto impl = self->get_impl();
1072 [ - + ]: 24 : if (generation != impl->generation_) return;
1073 : 24 : impl->discard_session_queue(wait);
1074 : 24 : impl->report_backpressure(self, impl->queue_bytes_);
1075 : 24 : impl->do_write(self);
1076 : : });
1077 : 24 : }
1078 : :
1079 : 1370 : std::optional<wrapper::SendResult> UdpChannel::poll_write_wait(const std::shared_ptr<detail::UdpWriteWait>& wait) {
1080 : 1370 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
1081 [ - + ]: 1370 : if (!wait) return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
1082 [ + + ]: 1370 : if (wait->ended_by) return wrapper::SendResult::reject(*wait->ended_by);
1083 [ + - - + ]: 1345 : if (auto reason = impl_->admission_rejection(wait->require_remote, wait->sequence))
1084 : 0 : return wrapper::SendResult::reject(*reason);
1085 [ + + ]: 1345 : if (!impl_->backpressure_active_) return wrapper::SendResult::accept();
1086 : 147 : return std::nullopt;
1087 : 1370 : }
1088 : 0 : void UdpChannel::end_write_wait(const std::shared_ptr<detail::UdpWriteWait>& wait, wrapper::SendRejection reason) {
1089 : 0 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
1090 [ # # ]: 0 : if (wait) wait->end(reason);
1091 : 0 : }
1092 : 850 : void UdpChannel::cancel_write_waits() {
1093 : 850 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
1094 : 850 : impl_->end_write_waits(wrapper::SendRejection::CancelledWhileWaiting);
1095 : 850 : }
1096 : 111 : std::optional<uint64_t> UdpChannel::write_connection() const {
1097 : 111 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
1098 [ + - - + ]: 111 : if (impl_->admission_rejection(true)) return std::nullopt;
1099 : 222 : return impl_->generation_.load();
1100 : 111 : }
1101 : 2745 : wrapper::SendResult UdpChannel::write_state(bool require_remote) {
1102 : 2745 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
1103 [ + - + + ]: 2745 : if (auto reason = impl_->admission_rejection(require_remote)) return wrapper::SendResult::reject(*reason);
1104 : 2699 : return wrapper::SendResult::accept();
1105 : 2745 : }
1106 : :
1107 : 121 : wrapper::SendResult UdpChannel::async_write_copy_result(memory::ConstByteSpan data) {
1108 : 121 : const auto result = write_copy(data);
1109 [ + + + - ]: 121 : if (auto hook = detail::g_udp_write_result_hook.load()) hook(result);
1110 : 121 : return result;
1111 : : }
1112 : :
1113 : 194 : wrapper::SendResult UdpChannel::write_copy(memory::ConstByteSpan data, std::optional<uint64_t> expected_run) {
1114 [ + + ]: 194 : if (expected_run) {
1115 [ + + + - ]: 73 : if (auto hook = detail::g_udp_pinned_write_hook.load()) hook();
1116 : : }
1117 : 194 : auto impl = get_impl();
1118 : 194 : std::lock_guard<std::mutex> submission_lock(impl->submission_mtx_);
1119 : 194 : const auto generation = impl->generation_.load();
1120 : 194 : const size_t size = data.size();
1121 : 194 : const auto validation = wrapper::detail::validate_payload_size(size);
1122 [ + + ]: 194 : if (!validation.accepted()) {
1123 : 5 : impl->stats_.record_failed_send();
1124 : 5 : return validation;
1125 : : }
1126 [ + - + + ]: 189 : if (auto reason = impl->admission_rejection(true, expected_run)) {
1127 : 26 : impl->stats_.record_failed_send();
1128 : 26 : return wrapper::SendResult::reject(*reason);
1129 : : }
1130 [ + + + + ]: 163 : if (impl->cfg_.enable_memory_pool && size <= 65536) {
1131 : 107 : memory::PooledBuffer pooled(size, impl->pool_);
1132 [ + - + - ]: 107 : if (pooled.valid()) {
1133 : 107 : base::safe_memory::safe_memcpy(pooled.data(), data.data(), size);
1134 [ + + ]: 107 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queue_bytes_, impl->pending_bytes_,
1135 : 107 : impl->inflight_bytes_, size, impl->bp_limit_)) {
1136 : 3 : impl->stats_.record_failed_send();
1137 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1138 : : }
1139 : 104 : impl->stats_.record_accepted(size);
1140 : 104 : Impl::Ledger::Admission admission(impl->send_accounting_, size);
1141 : 104 : const auto request = admission.request();
1142 : 104 : net::post(impl->strand_,
1143 : 104 : [self = shared_from_this(), generation, buf = std::move(pooled), size, request]() mutable {
1144 : 104 : auto impl = self->get_impl();
1145 [ - + ]: 104 : if (generation != impl->generation_) return;
1146 [ + - + + ]: 104 : if (!impl->enqueue_buffer(self, std::move(buf), size, request)) return;
1147 : 84 : impl->do_write(self);
1148 : : });
1149 : 104 : admission.commit();
1150 : 104 : return wrapper::SendResult::accept();
1151 : 104 : }
1152 : 107 : }
1153 : :
1154 [ + + ]: 56 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queue_bytes_, impl->pending_bytes_,
1155 : 56 : impl->inflight_bytes_, size, impl->bp_limit_)) {
1156 : 14 : impl->stats_.record_failed_send();
1157 : 14 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1158 : : }
1159 : 42 : std::vector<uint8_t> copy(data.begin(), data.end());
1160 : 42 : impl->stats_.record_accepted(size);
1161 : 42 : Impl::Ledger::Admission admission(impl->send_accounting_, size);
1162 : 42 : const auto request = admission.request();
1163 : 42 : net::post(impl->strand_, [self = shared_from_this(), generation, buf = std::move(copy), size, request]() mutable {
1164 : 42 : auto impl = self->get_impl();
1165 [ - + ]: 42 : if (generation != impl->generation_) return;
1166 [ + - + + ]: 42 : if (!impl->enqueue_buffer(self, std::move(buf), size, request)) return;
1167 : 32 : impl->do_write(self);
1168 : : });
1169 : 42 : admission.commit();
1170 : 42 : return wrapper::SendResult::accept();
1171 : 194 : }
1172 : :
1173 : 69 : wrapper::SendResult UdpChannel::async_write_move_result(std::vector<uint8_t>&& data) {
1174 : 69 : const auto result = write_move(std::move(data));
1175 [ + + + - ]: 69 : if (auto hook = detail::g_udp_write_result_hook.load()) hook(result);
1176 : 69 : return result;
1177 : : }
1178 : :
1179 : 85 : wrapper::SendResult UdpChannel::write_move(std::vector<uint8_t>&& data, std::optional<uint64_t> expected_run) {
1180 [ + + ]: 85 : if (expected_run) {
1181 [ + + + - ]: 16 : if (auto hook = detail::g_udp_pinned_write_hook.load()) hook();
1182 : : }
1183 : 85 : auto impl = get_impl();
1184 : 85 : std::lock_guard<std::mutex> submission_lock(impl->submission_mtx_);
1185 : 85 : const auto generation = impl->generation_.load();
1186 : 85 : const size_t size = data.size();
1187 : 85 : const auto validation = wrapper::detail::validate_payload_size(size);
1188 [ + + ]: 85 : if (!validation.accepted()) {
1189 : 4 : impl->stats_.record_failed_send();
1190 : 4 : return validation;
1191 : : }
1192 [ + - + + ]: 81 : if (auto reason = impl->admission_rejection(true, expected_run)) {
1193 : 16 : impl->stats_.record_failed_send();
1194 : 16 : return wrapper::SendResult::reject(*reason);
1195 : : }
1196 [ + + ]: 65 : if (size > impl->bp_limit_) {
1197 : 1 : WIRESTEAD_LOG_ERROR("udp", "write", "Queue limit exceeded by single write");
1198 : : // Keep error callbacks ordered with the same run's cleanup.
1199 : 1 : net::post(impl->strand_, [self = shared_from_this(), generation] {
1200 : 1 : auto impl = self->get_impl();
1201 [ + - - + : 1 : if (generation != impl->generation_ || impl->stop_requested_) return;
- + ]
1202 : 1 : impl->transition_to(LinkState::Error, {}, "write", "Queue limit exceeded by single write");
1203 : : });
1204 : 1 : impl->stats_.record_failed_send();
1205 : 1 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
1206 : : }
1207 [ + + ]: 64 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queue_bytes_, impl->pending_bytes_,
1208 : 64 : impl->inflight_bytes_, size, impl->bp_limit_)) {
1209 : 16 : impl->stats_.record_failed_send();
1210 : 16 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1211 : : }
1212 : 48 : impl->stats_.record_accepted(size);
1213 : 48 : Impl::Ledger::Admission admission(impl->send_accounting_, size);
1214 : 48 : const auto request = admission.request();
1215 : 48 : net::post(impl->strand_, [self = shared_from_this(), generation, buf = std::move(data), size, request]() mutable {
1216 : 48 : auto impl = self->get_impl();
1217 [ - + ]: 48 : if (generation != impl->generation_) return;
1218 [ + - + + ]: 48 : if (!impl->enqueue_buffer(self, std::move(buf), size, request)) return;
1219 : 28 : impl->do_write(self);
1220 : : });
1221 : 48 : admission.commit();
1222 : 48 : return wrapper::SendResult::accept();
1223 : 85 : }
1224 : :
1225 : 53 : wrapper::SendResult UdpChannel::async_write_shared_result(std::shared_ptr<const std::vector<uint8_t>> data) {
1226 : 53 : const auto result = write_shared(std::move(data));
1227 [ + + + - ]: 53 : if (auto hook = detail::g_udp_write_result_hook.load()) hook(result);
1228 : 53 : return result;
1229 : : }
1230 : :
1231 : 57 : wrapper::SendResult UdpChannel::write_shared(std::shared_ptr<const std::vector<uint8_t>> data,
1232 : : std::optional<uint64_t> expected_run) {
1233 [ + + ]: 57 : if (expected_run) {
1234 [ + + + - ]: 4 : if (auto hook = detail::g_udp_pinned_write_hook.load()) hook();
1235 : : }
1236 : 57 : auto impl = get_impl();
1237 : 57 : std::lock_guard<std::mutex> submission_lock(impl->submission_mtx_);
1238 : 57 : const auto generation = impl->generation_.load();
1239 [ + + ]: 57 : const size_t size = data ? data->size() : 0;
1240 : 57 : const auto validation = wrapper::detail::validate_payload_size(size);
1241 [ + + ]: 57 : if (!validation.accepted()) {
1242 : 5 : impl->stats_.record_failed_send();
1243 : 5 : return validation;
1244 : : }
1245 [ + - + + ]: 52 : if (auto reason = impl->admission_rejection(true, expected_run)) {
1246 : 16 : impl->stats_.record_failed_send();
1247 : 16 : return wrapper::SendResult::reject(*reason);
1248 : : }
1249 [ - + ]: 36 : if (size > impl->bp_limit_) {
1250 : 0 : WIRESTEAD_LOG_ERROR("udp", "write", "Queue limit exceeded by single write");
1251 : : // Keep error callbacks ordered with the same run's cleanup.
1252 : 0 : net::post(impl->strand_, [self = shared_from_this(), generation] {
1253 : 0 : auto impl = self->get_impl();
1254 [ # # # # : 0 : if (generation != impl->generation_ || impl->stop_requested_) return;
# # ]
1255 : 0 : impl->transition_to(LinkState::Error, {}, "write", "Queue limit exceeded by single write");
1256 : : });
1257 : 0 : impl->stats_.record_failed_send();
1258 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
1259 : : }
1260 [ + + ]: 36 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queue_bytes_, impl->pending_bytes_,
1261 : 36 : impl->inflight_bytes_, size, impl->bp_limit_)) {
1262 : 4 : impl->stats_.record_failed_send();
1263 : 4 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1264 : : }
1265 : 32 : impl->stats_.record_accepted(size);
1266 : 32 : Impl::Ledger::Admission admission(impl->send_accounting_, size);
1267 : 32 : const auto request = admission.request();
1268 : 32 : net::post(impl->strand_, [self = shared_from_this(), generation, buf = std::move(data), size, request]() mutable {
1269 : 32 : auto impl = self->get_impl();
1270 [ - + ]: 32 : if (generation != impl->generation_) return;
1271 [ + - + + ]: 32 : if (!impl->enqueue_buffer(self, std::move(buf), size, request)) return;
1272 : 20 : impl->do_write(self);
1273 : : });
1274 : 32 : admission.commit();
1275 : 32 : return wrapper::SendResult::accept();
1276 : 57 : }
1277 : :
1278 : 265980 : wrapper::SendResult UdpChannel::async_try_write_copy_result(memory::ConstByteSpan data) {
1279 : 265980 : const auto result = try_write_copy(data);
1280 [ + + + - ]: 265980 : if (auto hook = detail::g_udp_write_result_hook.load()) hook(result);
1281 : 265980 : return result;
1282 : : }
1283 : :
1284 : 266019 : wrapper::SendResult UdpChannel::try_write_copy(memory::ConstByteSpan data, std::optional<uint64_t> expected_run) {
1285 : 266019 : const auto validation = wrapper::detail::validate_payload_size(data.size());
1286 [ + + ]: 266019 : if (!validation.accepted()) {
1287 : 3 : get_impl()->stats_.record_failed_send();
1288 : 3 : return validation;
1289 : : }
1290 : 532032 : return try_write_move(std::vector<uint8_t>(data.begin(), data.end()), expected_run);
1291 : : }
1292 : :
1293 : 83 : wrapper::SendResult UdpChannel::async_try_write_move_result(std::vector<uint8_t>&& data) {
1294 : 83 : const auto result = try_write_move(std::move(data));
1295 [ + + + - ]: 83 : if (auto hook = detail::g_udp_write_result_hook.load()) hook(result);
1296 : 83 : return result;
1297 : : }
1298 : :
1299 : 266111 : wrapper::SendResult UdpChannel::try_write_move(std::vector<uint8_t>&& data, std::optional<uint64_t> expected_run) {
1300 [ - + ]: 266111 : if (expected_run) {
1301 [ # # # # ]: 0 : if (auto hook = detail::g_udp_pinned_write_hook.load()) hook();
1302 : : }
1303 : 266111 : auto impl = get_impl();
1304 : 266111 : std::lock_guard<std::mutex> submission_lock(impl->submission_mtx_);
1305 : 266111 : const auto generation = impl->generation_.load();
1306 : 266111 : const size_t size = data.size();
1307 : 266111 : const auto validation = wrapper::detail::validate_payload_size(size);
1308 [ + + ]: 266111 : if (!validation.accepted()) {
1309 : 3 : impl->stats_.record_failed_send();
1310 : 3 : return validation;
1311 : : }
1312 [ + - + + ]: 266108 : if (auto reason = impl->admission_rejection(true, expected_run)) {
1313 : 46 : impl->stats_.record_failed_send();
1314 : 46 : return wrapper::SendResult::reject(*reason);
1315 : : }
1316 : 18 : const auto reject_for_pressure = [impl, size]() {
1317 [ + + ]: 18 : if (impl->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
1318 : 11 : impl->stats_.record_dropped(1, size);
1319 : : } else {
1320 : 7 : impl->stats_.record_failed_send();
1321 : : }
1322 : 266080 : };
1323 [ + - + + : 532106 : if (impl->backpressure_active_.load() || impl->queue_bytes_ + size > impl->bp_high_ ||
+ + ]
1324 [ - + ]: 266044 : impl->queue_bytes_ + impl->pending_bytes_ + size > impl->bp_limit_) {
1325 : 18 : reject_for_pressure();
1326 : 18 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1327 : : }
1328 [ - + ]: 266044 : if (!queue_util::try_reserve_write_bytes(impl->write_reserve_mtx_, impl->inflight_bytes_, impl->queue_bytes_,
1329 : 266044 : impl->pending_bytes_, impl->backpressure_active_, size, impl->bp_high_,
1330 : : impl->bp_limit_)) {
1331 : 0 : reject_for_pressure();
1332 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1333 : : }
1334 : 266044 : impl->stats_.record_accepted(size);
1335 : 266044 : Impl::Ledger::Admission admission(impl->send_accounting_, size);
1336 : 266044 : const auto request = admission.request();
1337 : :
1338 : 266044 : net::post(impl->strand_, [self = shared_from_this(), generation, buf = std::move(data), size, request]() mutable {
1339 : 266044 : auto impl = self->get_impl();
1340 [ - + ]: 266044 : if (generation != impl->generation_) return;
1341 [ + - + - : 798074 : if (impl->stop_requested_.load() || impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) ||
+ + ]
1342 [ + + - + : 798074 : impl->state_.is_state(LinkState::Error) || !impl->remote_endpoint_) {
+ + ]
1343 : 39 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, size);
1344 : 39 : impl->stats_.record_failed_send();
1345 : 39 : return;
1346 : : }
1347 : :
1348 : 266005 : impl->tx_.push_back({std::move(buf), std::nullopt, request});
1349 : 266005 : impl->observe_queue();
1350 : 266005 : impl->report_backpressure(self, impl->queue_bytes_);
1351 : 266005 : impl->do_write(self);
1352 : 266005 : });
1353 : 266044 : admission.commit();
1354 : 266044 : return wrapper::SendResult::accept();
1355 : 266111 : }
1356 : :
1357 : 49 : wrapper::SendResult UdpChannel::async_try_write_shared_result(std::shared_ptr<const std::vector<uint8_t>> data) {
1358 : 49 : const auto result = try_write_shared(std::move(data));
1359 [ + + + - ]: 49 : if (auto hook = detail::g_udp_write_result_hook.load()) hook(result);
1360 : 49 : return result;
1361 : : }
1362 : :
1363 : 61 : wrapper::SendResult UdpChannel::try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data,
1364 : : std::optional<uint64_t> expected_run) {
1365 [ - + ]: 61 : if (expected_run) {
1366 [ # # # # ]: 0 : if (auto hook = detail::g_udp_pinned_write_hook.load()) hook();
1367 : : }
1368 : 61 : auto impl = get_impl();
1369 : 61 : std::lock_guard<std::mutex> submission_lock(impl->submission_mtx_);
1370 : 61 : const auto generation = impl->generation_.load();
1371 [ + - ]: 61 : const size_t size = data ? data->size() : 0;
1372 : 61 : const auto validation = wrapper::detail::validate_payload_size(size);
1373 [ + + ]: 61 : if (!validation.accepted()) {
1374 : 3 : impl->stats_.record_failed_send();
1375 : 3 : return validation;
1376 : : }
1377 [ + - + + ]: 58 : if (auto reason = impl->admission_rejection(true, expected_run)) {
1378 : 19 : impl->stats_.record_failed_send();
1379 : 19 : return wrapper::SendResult::reject(*reason);
1380 : : }
1381 : 7 : const auto reject_for_pressure = [impl, size]() {
1382 [ + + ]: 7 : if (impl->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
1383 : 4 : impl->stats_.record_dropped(1, size);
1384 : : } else {
1385 : 3 : impl->stats_.record_failed_send();
1386 : : }
1387 : 46 : };
1388 [ + - + + : 71 : if (impl->backpressure_active_.load() || impl->queue_bytes_ + size > impl->bp_high_ ||
+ + ]
1389 [ - + ]: 32 : impl->queue_bytes_ + impl->pending_bytes_ + size > impl->bp_limit_) {
1390 : 7 : reject_for_pressure();
1391 : 7 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1392 : : }
1393 [ - + ]: 32 : if (!queue_util::try_reserve_write_bytes(impl->write_reserve_mtx_, impl->inflight_bytes_, impl->queue_bytes_,
1394 : 32 : impl->pending_bytes_, impl->backpressure_active_, size, impl->bp_high_,
1395 : : impl->bp_limit_)) {
1396 : 0 : reject_for_pressure();
1397 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1398 : : }
1399 : 32 : impl->stats_.record_accepted(size);
1400 : 32 : Impl::Ledger::Admission admission(impl->send_accounting_, size);
1401 : 32 : const auto request = admission.request();
1402 : :
1403 : 32 : net::post(impl->strand_, [self = shared_from_this(), generation, buf = std::move(data), size, request]() mutable {
1404 : 32 : auto impl = self->get_impl();
1405 [ - + ]: 32 : if (generation != impl->generation_) return;
1406 [ + - + - : 80 : if (impl->stop_requested_.load() || impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) ||
+ + ]
1407 [ + + - + : 80 : impl->state_.is_state(LinkState::Error) || !impl->remote_endpoint_) {
+ + ]
1408 : 13 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, size);
1409 : 13 : impl->stats_.record_failed_send();
1410 : 13 : return;
1411 : : }
1412 : :
1413 : 19 : impl->tx_.push_back({std::move(buf), std::nullopt, request});
1414 : 19 : impl->observe_queue();
1415 : 19 : impl->report_backpressure(self, impl->queue_bytes_);
1416 : 19 : impl->do_write(self);
1417 : 19 : });
1418 : 32 : admission.commit();
1419 : 32 : return wrapper::SendResult::accept();
1420 : 61 : }
1421 : :
1422 : 388063 : void UdpChannel::on_bytes(OnBytes cb) {
1423 : 388063 : auto shared = interface::share_callback(std::move(cb));
1424 : 388063 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
1425 : 388063 : impl_->on_bytes_ = std::move(shared);
1426 : 388063 : }
1427 : :
1428 : 388811 : void UdpChannel::on_state(OnState cb) {
1429 : 388811 : auto shared = interface::share_callback(std::move(cb));
1430 : 388811 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
1431 : 388811 : impl_->on_state_ = std::move(shared);
1432 : 388811 : }
1433 : :
1434 : 388663 : void UdpChannel::on_backpressure(OnBackpressure cb) {
1435 : 388663 : auto shared = interface::share_callback(std::move(cb));
1436 : 388663 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
1437 : 388663 : impl_->on_bp_ = std::move(shared);
1438 : 388663 : }
1439 : :
1440 : 91 : void UdpChannel::set_backpressure_strategy(base::constants::BackpressureStrategy strategy) {
1441 : 91 : config::detail::strategy(strategy);
1442 : 91 : impl_->bp_strategy_.store(strategy, std::memory_order_relaxed);
1443 : 91 : }
1444 : :
1445 : 133 : bool UdpChannel::async_write_to(memory::ConstByteSpan data, const boost::asio::ip::udp::endpoint& destination) {
1446 : 133 : const auto result = write_to(data, destination);
1447 [ + + + - ]: 133 : if (auto hook = detail::g_udp_write_result_hook.load()) hook(result);
1448 : 266 : return result.accepted();
1449 : : }
1450 : :
1451 : 1325 : wrapper::SendResult UdpChannel::write_to(memory::ConstByteSpan data, const boost::asio::ip::udp::endpoint& destination,
1452 : : std::optional<uint64_t> expected_run,
1453 : : std::shared_ptr<detail::UdpWriteWait> session) {
1454 [ + + ]: 1325 : if (expected_run) {
1455 [ + + + - ]: 1192 : if (auto hook = detail::g_udp_pinned_write_hook.load()) hook();
1456 : : }
1457 : 1325 : auto impl = get_impl();
1458 : 1325 : std::lock_guard<std::mutex> submission_lock(impl->submission_mtx_);
1459 : 1325 : const auto generation = impl->generation_.load();
1460 : 1325 : const size_t size = data.size();
1461 : 1325 : const auto validation = wrapper::detail::validate_payload_size(size);
1462 [ + + ]: 1325 : if (!validation.accepted()) {
1463 : 3 : impl->stats_.record_failed_send();
1464 [ - + - - : 3 : if (session && session->stats) session->stats->counters.record_failed_send();
- + ]
1465 : 3 : return validation;
1466 : : }
1467 [ + - + + ]: 1322 : if (auto reason = impl->admission_rejection(false, expected_run)) {
1468 : 11 : impl->stats_.record_failed_send();
1469 [ + + + - : 11 : if (session && session->stats) session->stats->counters.record_failed_send();
+ + ]
1470 : 11 : return wrapper::SendResult::reject(*reason);
1471 : : }
1472 [ + + + - : 1311 : if (session && (session->ended_by || session->sequence != impl->generation_)) {
- + - + ]
1473 : 0 : impl->stats_.record_failed_send();
1474 [ # # ]: 0 : if (session->stats) session->stats->counters.record_failed_send();
1475 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
1476 : : }
1477 [ + + + + ]: 1311 : if (impl->cfg_.enable_memory_pool && size <= 65536) {
1478 : 1226 : memory::PooledBuffer pooled(size, impl->pool_);
1479 [ + - + - ]: 1226 : if (pooled.valid()) {
1480 : 1226 : base::safe_memory::safe_memcpy(pooled.data(), data.data(), size);
1481 [ + + ]: 1226 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queue_bytes_, impl->pending_bytes_,
1482 : 1226 : impl->inflight_bytes_, size, impl->bp_limit_)) {
1483 : 26 : impl->stats_.record_failed_send();
1484 [ + - + - : 26 : if (session && session->stats) session->stats->counters.record_failed_send();
+ - ]
1485 : 26 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1486 : : }
1487 : 1200 : impl->stats_.record_accepted(size);
1488 [ + + + - : 1200 : if (session && session->stats) session->stats->counters.record_accepted(size);
+ + ]
1489 : : Impl::Ledger::Admission admission(
1490 : 1200 : impl->send_accounting_, size,
1491 [ + + + - : 1200 : session && session->stats ? session->stats->accounting : Impl::Ledger::GroupHandle{});
+ - ]
1492 : 1200 : const auto request = admission.request();
1493 : 1200 : net::post(impl->strand_, [self = shared_from_this(), generation, buf = std::move(pooled), size, destination,
1494 : : request, session]() mutable {
1495 : 1200 : auto impl = self->get_impl();
1496 [ - + ]: 1200 : if (generation != impl->generation_) return;
1497 [ + - + + ]: 1200 : if (!impl->enqueue_buffer(self, std::move(buf), size, request, destination, session)) return;
1498 : 172 : impl->do_write(self);
1499 : : });
1500 : 1200 : admission.commit();
1501 : 1200 : return wrapper::SendResult::accept();
1502 : 1200 : }
1503 : 1226 : }
1504 : :
1505 [ + + ]: 85 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queue_bytes_, impl->pending_bytes_,
1506 : 85 : impl->inflight_bytes_, size, impl->bp_limit_)) {
1507 : 15 : impl->stats_.record_failed_send();
1508 [ + + + - : 15 : if (session && session->stats) session->stats->counters.record_failed_send();
+ + ]
1509 : 15 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1510 : : }
1511 : 70 : std::vector<uint8_t> copy(data.begin(), data.end());
1512 : 70 : impl->stats_.record_accepted(size);
1513 [ + + + - : 70 : if (session && session->stats) session->stats->counters.record_accepted(size);
+ + ]
1514 : : Impl::Ledger::Admission admission(
1515 : 70 : impl->send_accounting_, size,
1516 [ + + + - : 70 : session && session->stats ? session->stats->accounting : Impl::Ledger::GroupHandle{});
+ - ]
1517 : 70 : const auto request = admission.request();
1518 : 70 : net::post(impl->strand_, [self = shared_from_this(), generation, buf = std::move(copy), size, destination, request,
1519 : : session]() mutable {
1520 : 70 : auto impl = self->get_impl();
1521 [ - + ]: 70 : if (generation != impl->generation_) return;
1522 [ + - + + ]: 70 : if (!impl->enqueue_buffer(self, std::move(buf), size, request, destination, session)) return;
1523 : 56 : impl->do_write(self);
1524 : : });
1525 : 70 : admission.commit();
1526 : 70 : return wrapper::SendResult::accept();
1527 : 1325 : }
1528 : :
1529 : 33 : bool UdpChannel::async_try_write_to(memory::ConstByteSpan data, const boost::asio::ip::udp::endpoint& destination) {
1530 : 33 : const auto result = try_write_to(data, destination);
1531 [ - + - - ]: 33 : if (auto hook = detail::g_udp_write_result_hook.load()) hook(result);
1532 : 66 : return result.accepted();
1533 : : }
1534 : :
1535 : 238 : wrapper::SendResult UdpChannel::try_write_to(memory::ConstByteSpan data,
1536 : : const boost::asio::ip::udp::endpoint& destination,
1537 : : std::optional<uint64_t> expected_run,
1538 : : std::shared_ptr<detail::UdpWriteWait> session) {
1539 [ + + ]: 238 : if (expected_run) {
1540 [ + + + - ]: 205 : if (auto hook = detail::g_udp_pinned_write_hook.load()) hook();
1541 : : }
1542 : 238 : auto impl = get_impl();
1543 : 238 : std::lock_guard<std::mutex> submission_lock(impl->submission_mtx_);
1544 : 238 : const auto generation = impl->generation_.load();
1545 : 238 : const size_t size = data.size();
1546 : 238 : const auto validation = wrapper::detail::validate_payload_size(size);
1547 [ + + ]: 238 : if (!validation.accepted()) {
1548 : 1 : impl->stats_.record_failed_send();
1549 [ - + - - : 1 : if (session && session->stats) session->stats->counters.record_failed_send();
- + ]
1550 : 1 : return validation;
1551 : : }
1552 [ + - + + ]: 237 : if (auto reason = impl->admission_rejection(false, expected_run)) {
1553 : 4 : impl->stats_.record_failed_send();
1554 [ + + + - : 4 : if (session && session->stats) session->stats->counters.record_failed_send();
+ + ]
1555 : 4 : return wrapper::SendResult::reject(*reason);
1556 : : }
1557 [ + + + - : 233 : if (session && (session->ended_by || session->sequence != impl->generation_)) {
- + - + ]
1558 : 0 : impl->stats_.record_failed_send();
1559 [ # # ]: 0 : if (session->stats) session->stats->counters.record_failed_send();
1560 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
1561 : : }
1562 : 42 : const auto reject_for_pressure = [impl, size, session]() {
1563 [ + + ]: 42 : if (impl->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
1564 : 22 : impl->stats_.record_dropped(1, size);
1565 [ + + + - : 22 : if (session && session->stats) session->stats->counters.record_dropped(1, size);
+ + ]
1566 : : } else {
1567 : 20 : impl->stats_.record_failed_send();
1568 [ + - + - : 20 : if (session && session->stats) session->stats->counters.record_failed_send();
+ - ]
1569 : : }
1570 : 233 : };
1571 [ + + + + : 424 : if (impl->backpressure_active_.load() || impl->queue_bytes_ + size > impl->bp_high_ ||
+ + ]
1572 [ - + ]: 191 : impl->queue_bytes_ + impl->pending_bytes_ + size > impl->bp_limit_) {
1573 : 42 : reject_for_pressure();
1574 : 42 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1575 : : }
1576 [ - + ]: 191 : if (!queue_util::try_reserve_write_bytes(impl->write_reserve_mtx_, impl->inflight_bytes_, impl->queue_bytes_,
1577 : 191 : impl->pending_bytes_, impl->backpressure_active_, size, impl->bp_high_,
1578 : : impl->bp_limit_)) {
1579 : 0 : reject_for_pressure();
1580 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1581 : : }
1582 : :
1583 : 191 : std::vector<uint8_t> copy(data.begin(), data.end());
1584 : 191 : impl->stats_.record_accepted(size);
1585 [ + + + - : 191 : if (session && session->stats) session->stats->counters.record_accepted(size);
+ + ]
1586 : : Impl::Ledger::Admission admission(
1587 : 191 : impl->send_accounting_, size,
1588 [ + + + - : 191 : session && session->stats ? session->stats->accounting : Impl::Ledger::GroupHandle{});
+ - ]
1589 : 191 : const auto request = admission.request();
1590 [ + + + - : 191 : if (session && session->stats) {
+ + ]
1591 : 163 : session->stats->queued.fetch_add(size);
1592 : 163 : session->stats->observe_queue();
1593 : : }
1594 : 191 : net::post(impl->strand_, [self = shared_from_this(), generation, buf = std::move(copy), size, destination, request,
1595 : : session]() mutable {
1596 : 191 : auto impl = self->get_impl();
1597 [ - + ]: 191 : if (generation != impl->generation_) return;
1598 [ + + + - : 362 : if (impl->stop_requested_.load() || impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) ||
+ - + + ]
1599 : 362 : impl->state_.is_state(LinkState::Error)) {
1600 : 25 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, size);
1601 [ + + + - : 25 : if (session && session->stats) queue_util::release_reserved_write_bytes(session->stats->queued, size);
+ + ]
1602 : 25 : impl->stats_.record_failed_send();
1603 [ + + + - : 25 : if (session && session->stats) session->stats->counters.record_failed_send();
+ + ]
1604 : 25 : return;
1605 : : }
1606 : :
1607 [ - + ]: 166 : if (!impl->send_accounting_.contains(request)) {
1608 : 0 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, size);
1609 [ # # # # : 0 : if (session && session->stats) queue_util::release_reserved_write_bytes(session->stats->queued, size);
# # ]
1610 : 0 : impl->report_backpressure(self, impl->queue_bytes_);
1611 : 0 : return;
1612 : : }
1613 : :
1614 : 332 : impl->tx_.push_back({std::move(buf), destination, request, session});
1615 : 166 : impl->observe_queue();
1616 : 166 : impl->report_backpressure(self, impl->queue_bytes_);
1617 : 166 : impl->do_write(self);
1618 : 166 : });
1619 : 191 : admission.commit();
1620 : 191 : return wrapper::SendResult::accept();
1621 : 238 : }
1622 : :
1623 : : // Takes callback_mtx_ like every other setter here. It previously assigned
1624 : : // without the lock while the strand-confined read site took it, which raced
1625 : : // against a concurrent on_bytes_from() replacement.
1626 : 605 : void UdpChannel::on_bytes_from(OnBytesFrom cb) {
1627 : 605 : auto shared = interface::share_callback(std::move(cb));
1628 : 605 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
1629 : 605 : impl_->on_bytes_from_ = std::move(shared);
1630 : 605 : }
1631 : :
1632 : 33 : boost::asio::ip::udp::endpoint UdpChannel::local_endpoint() const { return get_impl()->local_endpoint_; }
1633 : :
1634 : 2624 : boost::asio::any_io_executor UdpChannel::get_executor() { return get_impl()->strand_; }
1635 : :
1636 : : } // namespace transport
1637 : : } // namespace wirestead
|
