Branch data Line data Source code
1 : : /*
2 : : * Copyright 2025 Jinwoo Sung
3 : : *
4 : : * Licensed under the Apache License, Version 2.0 (the "License");
5 : : * you may not use this file except in compliance with the License.
6 : : * You may obtain a copy of the License at
7 : : *
8 : : * http://www.apache.org/licenses/LICENSE-2.0
9 : : *
10 : : * Unless required by applicable law or agreed to in writing, software
11 : : * distributed under the License is distributed on an "AS IS" BASIS,
12 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 : : * See the License for the specific language governing permissions and
14 : : * limitations under the License.
15 : : */
16 : :
17 : : #include "wirestead/wrapper/udp/udp.hpp"
18 : :
19 : : #include "wirestead/concurrency/io_thread_hook.hpp"
20 : :
21 : : #if defined(__GNUC__) || defined(__clang__)
22 : : #pragma GCC diagnostic ignored "-Wsign-conversion"
23 : : #pragma GCC diagnostic ignored "-Wconversion"
24 : : #endif
25 : :
26 : : #include <atomic>
27 : : #include <boost/asio/executor_work_guard.hpp>
28 : : #include <boost/asio/io_context.hpp>
29 : : #include <boost/asio/steady_timer.hpp>
30 : : #include <boost/asio/strand.hpp>
31 : : #include <chrono>
32 : : #include <iostream>
33 : : #include <mutex>
34 : : #include <shared_mutex>
35 : : #include <stdexcept>
36 : : #include <thread>
37 : : #include <vector>
38 : :
39 : : #include "wirestead/base/common.hpp"
40 : : #include "wirestead/config/validation.hpp"
41 : : #include "wirestead/factory/channel_factory.hpp"
42 : : #include "wirestead/interface/connection_channel.hpp"
43 : : #include "wirestead/transport/udp/detail/write_wait.hpp"
44 : : #include "wirestead/transport/udp/udp.hpp"
45 : : #include "wirestead/util/input_validator.hpp"
46 : : #include "wirestead/wrapper/bounded_receive.hpp"
47 : : #include "wirestead/wrapper/callback_guard.hpp"
48 : : #include "wirestead/wrapper/error_context_builder.hpp"
49 : : #include "wirestead/wrapper/lifecycle_events.hpp"
50 : : #include "wirestead/wrapper/send_retry.hpp"
51 : : #include "wirestead/wrapper/send_validation.hpp"
52 : :
53 : : namespace wirestead {
54 : : namespace wrapper {
55 : :
56 : : struct UdpClient::Impl : public std::enable_shared_from_this<Impl> {
57 : : mutable std::shared_mutex mutex_;
58 : : std::mutex bp_mutex_;
59 : : std::condition_variable bp_cv_;
60 : : // D-1: admission gate for this object's user callbacks. Admission, the
61 : : // running count and the closed flag are one decision, so a stop() cannot
62 : : // observe an empty gate while a callback is about to start, and a callback
63 : : // left over from a previous run is refused after a restart.
64 : : detail::CallbackGate callback_gate_;
65 : : std::mutex stop_finalize_mutex_;
66 : : bool stop_requested_ = false;
67 : : std::atomic<unsigned> stop_callers_{0};
68 : : std::atomic<uint64_t> callback_generation_{0};
69 : :
70 : : // True when this thread is one the target's shutdown needs: a callback of
71 : : // this object, or any thread currently running the external io_context this
72 : : // channel was built on (which a callback of another channel sharing it is).
73 : : // Such a caller requests the shutdown and returns; it cannot wait for work
74 : : // its own thread has to perform.
75 : 947 : bool shutdown_needs_this_thread() const {
76 [ + - + + ]: 947 : if (callback_gate_.active_on_this_thread()) return true;
77 [ + + - + : 939 : if (external_ioc && external_ioc->get_executor().running_in_this_thread()) return true;
+ + - + ]
78 : 939 : std::shared_lock<std::shared_mutex> lock(mutex_);
79 [ + + ]: 939 : if (!channel) return false;
80 : 763 : const auto executor = channel->get_executor();
81 : : using IoExecutor = boost::asio::io_context::executor_type;
82 [ + - + + ]: 763 : if (auto* io = executor.target<IoExecutor>()) return io->running_in_this_thread();
83 [ + - + + ]: 473 : if (auto* strand = executor.target<boost::asio::strand<IoExecutor>>())
84 : 470 : return strand->get_inner_executor().running_in_this_thread();
85 : 3 : return false;
86 : 939 : }
87 : :
88 : : config::UdpConfig cfg;
89 : : std::shared_ptr<interface::Channel> channel;
90 : : std::shared_ptr<boost::asio::io_context> external_ioc;
91 : : std::atomic<bool> use_external_context{false};
92 : : std::atomic<bool> manage_external_context{false};
93 : : std::thread external_thread;
94 : : std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>> work_guard;
95 : :
96 : : // Event handlers (Context based)
97 : : // Shared snapshots: the strand copies one out per received datagram, and a
98 : : // std::function copy allocates whenever the user handler outgrows its
99 : : // small-object buffer. See interface::SharedCallback.
100 : : interface::SharedCallback<MessageHandler> data_handler;
101 : : interface::SharedCallback<BatchMessageHandler> data_batch_handler_;
102 : : ConnectionHandler connect_handler{nullptr};
103 : : ConnectionHandler disconnect_handler{nullptr};
104 : : ErrorHandler error_handler{nullptr};
105 : : std::function<void(size_t)> bp_handler{nullptr};
106 : : interface::SharedCallback<MessageHandler> message_handler;
107 : : interface::SharedCallback<BatchMessageHandler> message_batch_handler_;
108 : :
109 : : std::shared_ptr<framer::IFramer> framer{nullptr};
110 : :
111 : : detail::LifecycleEvents lifecycle_events_;
112 : 22 : void require_stopped_config() const {
113 [ + + + - : 42 : if (started_.load() || stop_callers_.load() || (stop_requested_ && alive_marker) || detail::in_data_callback())
+ + + + -
+ + + ]
114 : 3 : throw std::logic_error("configuration requires completed stop");
115 : 19 : }
116 : : ReceiveLimits receive_limits_;
117 : : std::shared_ptr<detail::ReceiveBudget> receive_budget_{std::make_shared<detail::ReceiveBudget>(receive_limits_)};
118 : : std::shared_ptr<detail::ReceiveState> receive_state_{
119 : : std::make_shared<detail::ReceiveState>(receive_budget_->open_scope())};
120 : :
121 : : // Batching logic
122 : : detail::ReceiveBatch data_batch_queue_;
123 : : detail::ReceiveBatch message_batch_queue_;
124 : : std::unique_ptr<boost::asio::steady_timer> batch_timer_;
125 : : size_t max_batch_size_ = 100;
126 : 275 : std::chrono::milliseconds max_batch_latency_{1};
127 : :
128 : : std::atomic<bool> auto_start_{false};
129 : : std::vector<std::promise<bool>> pending_promises;
130 : : std::atomic<bool> started_{false};
131 : 275 : std::shared_ptr<bool> alive_marker{std::make_shared<bool>(true)};
132 : :
133 : : // False only for the dependency-injected-channel constructor below, where
134 : : // the caller owns the channel's identity and lifecycle (e.g. tests
135 : : // injecting a fake channel) - stop() must not discard and factory-rebuild
136 : : // a channel it didn't create itself.
137 : : bool factory_managed_channel_ = true;
138 : :
139 : 79 : explicit Impl(const config::UdpConfig& config) : cfg(config) {}
140 : 5 : Impl(const config::UdpConfig& config, std::shared_ptr<boost::asio::io_context> ioc)
141 [ + - + - : 5 : : cfg(config), external_ioc(std::move(ioc)), use_external_context(external_ioc != nullptr) {}
+ - + - +
- ]
142 : 191 : explicit Impl(std::shared_ptr<interface::Channel> ch) : channel(std::move(ch)), factory_managed_channel_(false) {
143 [ + + + + ]: 483 : if (!std::dynamic_pointer_cast<transport::UdpChannel>(channel) &&
144 [ + + + + : 292 : !std::dynamic_pointer_cast<interface::ConnectionChannel>(channel))
+ - ]
145 : 3 : throw std::invalid_argument("UdpClient requires its native transport or a ConnectionChannel");
146 : :
147 : : // #450: setup_internal_handlers() captures weak_from_this() - calling it
148 : : // from inside this constructor would capture an empty weak_ptr, since
149 : : // enable_shared_from_this isn't wired up until make_shared() finishes
150 : : // constructing the object. Deferred to UdpClient's own constructor,
151 : : // which runs after impl_ is a fully-formed shared_ptr<Impl>.
152 : 260 : }
153 : :
154 : 272 : ~Impl() {
155 : : try {
156 : 272 : stop();
157 : 0 : } catch (...) {
158 : 0 : }
159 : 272 : }
160 : :
161 : 1212 : void fulfill_all_locked(bool value) {
162 [ + + ]: 1436 : for (auto& promise : pending_promises) {
163 : : try {
164 : 224 : promise.set_value(value);
165 : 0 : } catch (...) {
166 : 0 : }
167 : : }
168 : 1212 : pending_promises.clear();
169 : 1212 : }
170 : :
171 : 34 : void flush_batches(uint64_t generation) {
172 : 34 : auto lease = callback_gate_.enter(generation);
173 [ - + ]: 34 : if (!lease.admitted()) return;
174 : : {
175 : 34 : std::shared_lock<std::shared_mutex> lock(mutex_);
176 [ + + + + : 34 : if (data_batch_queue_.empty() && message_batch_queue_.empty()) return;
+ + ]
177 : 34 : }
178 : 9 : std::unique_lock<std::shared_mutex> lock(mutex_);
179 [ + + ]: 9 : if (!data_batch_queue_.empty()) {
180 : 7 : auto handler = data_batch_handler_;
181 : 7 : auto batch = std::move(data_batch_queue_);
182 : 7 : data_batch_queue_.clear();
183 [ + - ]: 7 : if (handler) {
184 : 7 : lock.unlock();
185 : 7 : detail::invoke_user_callback("udp_client", "on_data_batch", handler, batch);
186 : 7 : lock.lock();
187 : : }
188 : 7 : }
189 [ + - + + ]: 9 : if (!callback_gate_.enter(generation).admitted()) return;
190 [ + + ]: 6 : if (!message_batch_queue_.empty()) {
191 : 3 : auto handler = message_batch_handler_;
192 : 3 : auto batch = std::move(message_batch_queue_);
193 : 3 : message_batch_queue_.clear();
194 [ + - ]: 3 : if (handler) {
195 : 3 : lock.unlock();
196 : 3 : detail::invoke_user_callback("udp_client", "on_message_batch", handler, batch);
197 : 3 : lock.lock();
198 : : }
199 : 3 : }
200 [ + - ]: 6 : if (batch_timer_) {
201 : 6 : batch_timer_->cancel();
202 : : }
203 : 37 : }
204 : :
205 : 18 : void schedule_batch_timer(uint64_t generation) {
206 [ - + ]: 18 : if (!batch_timer_) return;
207 : 18 : batch_timer_->expires_after(max_batch_latency_);
208 : 36 : batch_timer_->async_wait([this, generation, weak_impl = weak_from_this(),
209 : 18 : weak_alive = std::weak_ptr<bool>(alive_marker)](const boost::system::error_code& ec) {
210 [ + + ]: 9 : if (ec) return;
211 : 7 : auto impl_keepalive = weak_impl.lock();
212 [ - + ]: 7 : if (!impl_keepalive) return;
213 : 7 : auto alive = weak_alive.lock();
214 [ - + ]: 7 : if (!alive) return;
215 : 7 : flush_batches(generation);
216 : 7 : });
217 : : }
218 : :
219 : 282 : std::future<bool> start() {
220 : 282 : std::unique_lock<std::shared_mutex> lock(mutex_);
221 : 282 : stop_requested_ = false;
222 [ + + + - : 282 : if (channel && channel->is_connected()) {
+ + + + ]
223 : 58 : started_.store(true);
224 : 58 : std::promise<bool> p;
225 : 58 : p.set_value(true);
226 : 58 : return p.get_future();
227 : 58 : }
228 : :
229 : 224 : std::promise<bool> p;
230 : 224 : auto future = p.get_future();
231 : 224 : pending_promises.emplace_back(std::move(p));
232 : :
233 [ + + ]: 224 : if (started_.load()) {
234 : 1 : return future;
235 : : }
236 : :
237 [ + + + - ]: 223 : if (!alive_marker) alive_marker = std::make_shared<bool>(true);
238 [ + + ]: 223 : if (!channel) {
239 : 67 : channel = factory::ChannelFactory::create(cfg, external_ioc);
240 : : }
241 : 223 : setup_internal_handlers();
242 : 223 : started_.store(true);
243 : :
244 : 223 : auto channel_copy = channel;
245 : 223 : lock.unlock();
246 : 223 : channel_copy->start();
247 [ + + + + : 223 : if (use_external_context && manage_external_context && !external_thread.joinable()) {
+ - + + ]
248 [ + - - + : 1 : if (external_ioc->stopped()) external_ioc->restart();
- - ]
249 : 2 : work_guard = std::make_unique<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>(
250 : 3 : boost::asio::make_work_guard(*external_ioc));
251 : 2 : external_thread = std::thread([this, ioc = external_ioc]() {
252 : 1 : wirestead::concurrency::run_io_thread_init();
253 : : try {
254 : 1 : ioc->run();
255 : 0 : } catch (...) {
256 : 0 : }
257 : 2 : });
258 : : }
259 : 223 : return future;
260 : 282 : }
261 : :
262 : 947 : void stop() {
263 : 947 : stop_callers_.fetch_add(1);
264 : : struct StopCall {
265 : : std::atomic<unsigned>& callers;
266 : 947 : ~StopCall() { callers.fetch_sub(1); }
267 : 947 : } stop_call{stop_callers_};
268 : 947 : const bool request_only = shutdown_needs_this_thread();
269 : 947 : callback_gate_.close();
270 : 947 : std::shared_ptr<interface::Channel> channel_copy;
271 : : {
272 : 947 : std::unique_lock<std::shared_mutex> lock(mutex_);
273 : 947 : stop_requested_ = true;
274 [ + + + - ]: 947 : if (auto udp = std::dynamic_pointer_cast<transport::UdpChannel>(channel)) udp->cancel_write_waits();
275 [ + + + - ]: 947 : if (auto custom = std::dynamic_pointer_cast<interface::ConnectionChannel>(channel)) custom->cancel_write_waits();
276 : 947 : started_.store(false);
277 : :
278 : 947 : bp_cv_.notify_all();
279 : 947 : fulfill_all_locked(false);
280 : 947 : channel_copy = channel;
281 : 947 : }
282 [ + + + - ]: 947 : if (channel_copy) channel_copy->stop();
283 [ + + ]: 947 : if (request_only) return;
284 : :
285 : 925 : callback_gate_.wait_until_idle();
286 : 925 : std::lock_guard<std::mutex> finalize_lock(stop_finalize_mutex_);
287 : : {
288 : 925 : std::unique_lock<std::shared_mutex> lock(mutex_);
289 : 925 : alive_marker.reset();
290 [ + + ]: 925 : if (batch_timer_) {
291 : 278 : batch_timer_->cancel();
292 : 278 : batch_timer_.reset();
293 : : }
294 : :
295 [ + + ]: 925 : if (channel) {
296 : 749 : channel->on_bytes(nullptr);
297 : 749 : channel->on_state(nullptr);
298 : 749 : channel->on_backpressure(nullptr);
299 : : }
300 [ + + ]: 925 : if (factory_managed_channel_) channel.reset();
301 : 925 : data_batch_queue_.clear();
302 : 925 : message_batch_queue_.clear();
303 : 925 : receive_state_->reset(framer.get());
304 : 925 : }
305 [ + + + + : 925 : if (use_external_context && manage_external_context) {
+ + ]
306 : 7 : work_guard.reset();
307 [ + - + - ]: 7 : if (external_ioc) external_ioc->stop();
308 [ + + + - ]: 7 : if (external_thread.joinable()) external_thread.join();
309 : : }
310 : 969 : }
311 : :
312 : : // Caller holds mutex_. Native readiness remains part of transport admission.
313 : 827 : SendResult send_state(const std::shared_ptr<transport::UdpChannel>& udp, bool custom = false) {
314 [ + + ]: 1654 : if (stop_callers_.load() != 0) return SendResult::reject(SendRejection::Stopping);
315 [ + + ]: 826 : if (!started_.load()) {
316 [ + + ]: 74 : if (stop_requested_) {
317 [ - + ]: 35 : if (!callback_gate_.idle()) return SendResult::reject(SendRejection::Stopping);
318 [ + + ]: 35 : if (udp) {
319 : 32 : const auto state = udp->write_state();
320 [ + - + + : 32 : if (!state.accepted() && state.reason() == SendRejection::Stopping) return state;
+ + ]
321 : : }
322 : : }
323 : 62 : return SendResult::reject(SendRejection::NotStarted);
324 : : }
325 [ + + - + : 752 : if (!udp && !custom) return SendResult::reject(SendRejection::NotReady);
- + ]
326 : 752 : return SendResult::accept();
327 : : }
328 : :
329 : 664 : static SendResult finish_send(SendResult result) {
330 [ + + ]: 664 : if (auto hook = detail::g_udp_send_result_hook.load()) hook(result);
331 : 664 : return result;
332 : : }
333 : :
334 : : template <typename NativeWrite, typename CustomWrite>
335 : 183 : SendResult nonblocking_send(size_t size, bool best_effort_send, NativeWrite native_write, CustomWrite custom_write) {
336 : 183 : std::shared_lock<std::shared_mutex> lock(mutex_);
337 : 183 : auto udp = std::dynamic_pointer_cast<transport::UdpChannel>(channel);
338 [ + + ]: 183 : auto custom = udp ? nullptr : std::dynamic_pointer_cast<interface::ConnectionChannel>(channel);
339 : 183 : const auto result = [&]() -> SendResult {
340 [ + + + - : 183 : auto validation = detail::validate_payload_size(size, channel ? channel->write_queue_limit() : std::nullopt);
+ + + - +
+ + - ]
341 [ + + + + : 183 : if (!validation.accepted()) return validation;
+ + ]
342 [ + - + - : 122 : const auto state = send_state(udp, custom != nullptr);
+ - ]
343 [ + + + + : 122 : if (!state.accepted()) return state;
+ + ]
344 : : // Admission rechecks state and capacity together under the channel lock.
345 [ + + + - : 80 : auto admitted = custom ? custom_write(*custom) : native_write(*udp);
+ - + + +
- + - + +
+ - + - ]
346 [ + + + + : 80 : if (best_effort_send && !admitted.accepted() && admitted.reason() == SendRejection::WouldBlock)
+ + + + +
+ + + + +
+ + + + +
+ + + +
+ ]
347 : 9 : return SendResult::reject(SendRejection::QueueFull);
348 : 71 : return admitted;
349 : 183 : }();
350 : 183 : lock.unlock();
351 : 366 : return finish_send(result);
352 : 183 : }
353 : :
354 : 94 : SendResult try_send(std::string_view data, bool best_effort_send = false) {
355 : 94 : auto binary_view = base::safe_convert::string_to_bytes(data);
356 : 94 : memory::ConstByteSpan span(binary_view.first, binary_view.second);
357 : 94 : return nonblocking_send(
358 : 39 : data.size(), best_effort_send, [&](auto& udp) { return udp.try_write_copy(span); },
359 : 195 : [&](auto& channel) { return channel.async_try_write_copy_result(span); });
360 : : }
361 : :
362 : 40 : SendResult try_send_move(std::vector<uint8_t>&& data, bool best_effort_send = false) {
363 : 40 : return nonblocking_send(
364 : 12 : data.size(), best_effort_send, [&](auto& udp) { return udp.try_write_move(std::move(data)); },
365 : 45 : [&](auto& channel) { return channel.async_try_write_move_result(std::move(data)); });
366 : : }
367 : :
368 : 49 : SendResult try_send_shared(std::shared_ptr<const std::vector<uint8_t>> data, bool best_effort_send = false) {
369 [ + + ]: 92 : return nonblocking_send(
370 [ + - ]: 104 : data ? data->size() : 0, best_effort_send, [&](auto& udp) { return udp.try_write_shared(std::move(data)); },
371 : 54 : [&](auto& channel) { return channel.async_try_write_shared_result(std::move(data)); });
372 : : }
373 : :
374 : 142 : SendResult send(std::string_view data) {
375 [ + + ]: 142 : if (cfg.backpressure_strategy == base::constants::BackpressureStrategy::Reliable) return send_blocking(data);
376 : 13 : return try_send(data, true);
377 : : }
378 : :
379 : : struct ConnectionPin {
380 : : bool cannot_wait = false;
381 : : std::shared_ptr<interface::ConnectionChannel> custom;
382 : : interface::ConnectionChannel::Connection custom_wait;
383 : : std::shared_ptr<transport::UdpChannel> udp;
384 : : std::shared_ptr<transport::detail::UdpWriteWait> wait;
385 : : };
386 : :
387 : 111 : bool connection_matches(const ConnectionPin& pin) const {
388 [ + - + - : 111 : return !pin.udp || (pin.wait && pin.udp->write_connection() == pin.wait->sequence);
+ - + - ]
389 : : }
390 : :
391 : : // channel->on_backpressure() calls bp_cv_.notify_all() from the transport's io_context
392 : : // thread without holding bp_mutex_ (backpressure_active_ is a plain atomic on the transport
393 : : // side, not guarded by bp_mutex_ at all). That makes a classic lost-wakeup race possible: a
394 : : // waiter can check the predicate, find it still blocking, and be in the process of
395 : : // registering to wait when the notify fires - in the rare case that race is lost, an
396 : : // unbounded wait() would block forever. Poll with a bounded timeout instead so a missed
397 : : // notify only costs a short delay rather than a permanent hang (see #427, #431).
398 : : //
399 : : // Callback scopes cannot wait for capacity: they may be running on the
400 : : // executor needed to drain their own or another channel's queue (D-2).
401 : 402 : SendResult wait_for_backpressure_clear(std::unique_lock<std::mutex>& bp_lock, size_t payload_size,
402 : : uint64_t generation, const ConnectionPin& connection) {
403 [ + + ]: 402 : if (connection.custom_wait) {
404 : 290 : auto outcome = connection.custom_wait->poll_capacity();
405 [ + + ]: 290 : if (outcome) return *outcome;
406 [ + + ]: 118 : if (connection.cannot_wait) return SendResult::reject(SendRejection::WouldBlock);
407 [ + + + - ]: 37 : if (auto hook = detail::g_udp_capacity_wait_hook.load()) hook();
408 [ + - + + ]: 47 : while (!bp_cv_.wait_for(bp_lock, std::chrono::milliseconds(50), [&] {
409 : 84 : outcome = connection.custom_wait->poll_capacity();
410 : 84 : return outcome.has_value();
411 : : })) {
412 : : }
413 [ + + + - ]: 37 : if (auto hook = detail::g_udp_capacity_wait_result_hook.load()) hook(*outcome);
414 : 37 : return *outcome;
415 : : }
416 [ - + ]: 112 : if (!detail::payload_needs_capacity(payload_size)) return SendResult::accept();
417 : 112 : auto immediate = [this, payload_size, generation, &connection] {
418 : 112 : std::shared_lock<std::shared_mutex> lock(mutex_);
419 [ + + + - : 224 : return callback_generation_.load() != generation || !started_.load() || !channel || !channel->is_connected() ||
+ - + - ]
420 [ + - + - ]: 111 : !connection_matches(connection) ||
421 [ + - + - : 335 : !detail::payload_needs_capacity(payload_size, channel->write_queue_limit()) ||
+ - ]
422 [ + - + + ]: 223 : !channel->is_backpressure_active();
423 : 112 : };
424 : : // A bypass is not a completed capacity wait; final admission checks still apply.
425 [ + - + + ]: 112 : if (immediate()) return SendResult::accept();
426 [ - + ]: 24 : if (connection.cannot_wait) return SendResult::reject(SendRejection::WouldBlock);
427 [ + - + - ]: 24 : if (auto hook = detail::g_udp_capacity_wait_hook.load()) hook();
428 : 24 : std::optional<SendResult> outcome;
429 : 24 : auto released = [&] {
430 [ - + ]: 24 : if (outcome) return true;
431 : 24 : std::shared_lock<std::shared_mutex> lock(mutex_);
432 : : // The connection record retains the first terminal cause under the
433 : : // same transport lock used by stop, loss and admission.
434 : 24 : outcome = connection.udp->poll_write_wait(connection.wait);
435 : 24 : return outcome.has_value();
436 : 24 : };
437 [ + - - + ]: 24 : while (!bp_cv_.wait_for(bp_lock, std::chrono::milliseconds(50), released)) {
438 : : }
439 [ + - + - ]: 24 : if (auto hook = detail::g_udp_capacity_wait_result_hook.load()) hook(*outcome);
440 : 24 : return *outcome;
441 : : }
442 : :
443 : : // #509: high-water pressure and hard-limit reservations are different
444 : : // thresholds. Capacity can be refilled between a wait and admission, so
445 : : // retry transient native WouldBlock until admission or cancellation. Validation and
446 : : // terminal state failures return immediately.
447 : :
448 : : template <typename NativeWrite, typename CustomWrite>
449 : 481 : SendResult blocking_send(size_t size, NativeWrite native_write, CustomWrite custom_write) {
450 : : uint64_t generation;
451 : 481 : ConnectionPin connection;
452 : 481 : const auto result = [&]() -> SendResult {
453 : : {
454 : 481 : std::shared_lock<std::shared_mutex> lock(mutex_);
455 : : // Select the run and native connection together at entry. Validation
456 : : // precedes state and waiting, including the line delimiter and hard cap.
457 : 481 : generation = callback_generation_.load();
458 : 481 : connection.cannot_wait =
459 [ + + + + : 481 : detail::in_data_callback() || (channel && detail::executor_running_here(channel->get_executor()));
+ - + - +
+ + + - -
+ + + + +
- + - + +
+ + - - +
+ + + + -
+ - + + +
+ - - ]
460 : 481 : connection.udp = std::dynamic_pointer_cast<transport::UdpChannel>(channel);
461 [ + + + + : 481 : connection.custom = connection.udp ? nullptr : std::dynamic_pointer_cast<interface::ConnectionChannel>(channel);
+ + ]
462 [ + + + - : 481 : auto validation = detail::validate_payload_size(size, channel ? channel->write_queue_limit() : std::nullopt);
+ + + - +
+ + - ]
463 [ + + + + : 481 : if (!validation.accepted()) return validation;
+ + ]
464 [ + - + - : 414 : auto state = send_state(connection.udp, connection.custom != nullptr);
+ - ]
465 [ + + + + : 414 : if (!state.accepted()) return state;
+ + ]
466 [ + + + + : 382 : if (connection.custom) {
+ + ]
467 : 274 : auto captured = connection.custom->capture_write_connection();
468 [ + + + + : 274 : if (auto reason = std::get_if<SendRejection>(&captured)) return SendResult::reject(*reason);
+ + ]
469 : 253 : connection.custom_wait = std::get<interface::ConnectionChannel::Connection>(std::move(captured));
470 [ - + - - : 253 : if (!connection.custom_wait) throw std::logic_error("ConnectionChannel returned a null connection");
- + - - -
+ - - ]
471 : 274 : } else {
472 : 108 : state = connection.udp->write_state();
473 [ + + + + : 108 : if (!state.accepted()) return state;
+ + ]
474 : 100 : connection.wait = connection.udp->capture_write_wait();
475 [ - + - + : 100 : if (!connection.wait) return SendResult::reject(SendRejection::NotReady);
- + ]
476 : : }
477 : 481 : }
478 [ + + + + : 1046 : for (bool retry = false;; retry = true) {
+ + + + +
+ + + ]
479 [ + + + - : 402 : if (retry) detail::pause_send_retry(bp_cv_, bp_mutex_);
+ + + - +
+ + - ]
480 : 402 : std::unique_lock<std::mutex> bp_lock(bp_mutex_);
481 : 402 : const auto released = wait_for_backpressure_clear(bp_lock, size, generation, connection);
482 [ + + + + : 402 : if (!released.accepted()) return released; // Never overwrite the cause of release.
+ + ]
483 : 291 : bp_lock.unlock();
484 : 291 : std::shared_lock<std::shared_mutex> lock(mutex_);
485 [ + - + - : 291 : const auto state = send_state(connection.udp, connection.custom != nullptr);
+ - ]
486 [ - + - + : 291 : if (!state.accepted()) return state;
+ + ]
487 [ - + - + : 580 : if (callback_generation_.load() != generation) return SendResult::reject(SendRejection::NotReady);
- + ]
488 [ + + + - : 290 : const auto admitted = connection.custom_wait ? custom_write(*connection.custom_wait)
+ + + - +
+ + - ]
489 : 93 : : native_write(*connection.udp, connection.wait->sequence);
490 [ + + + + : 290 : if (admitted.accepted() || admitted.reason() != SendRejection::WouldBlock) return admitted;
+ + + + +
+ + + + +
+ + + + ]
491 : : // Only transient capacity refusal can be retried, never a terminal
492 : : // result. Callback callers must return without entering another wait.
493 [ + + + + : 73 : if (connection.cannot_wait) return admitted;
+ + ]
494 : : }
495 : 481 : }();
496 : 962 : return finish_send(result);
497 : 481 : }
498 : :
499 : 84 : SendResult send_move(std::vector<uint8_t>&& data) {
500 [ + + ]: 84 : if (cfg.backpressure_strategy != base::constants::BackpressureStrategy::Reliable)
501 : 13 : return try_send_move(std::move(data), true);
502 : 71 : return blocking_send(
503 : 16 : data.size(), [&](auto& udp, uint64_t sequence) { return udp.write_move(std::move(data), sequence); },
504 : 112 : [&](auto& connection) { return connection.write_move(std::move(data)); });
505 : : }
506 : :
507 : 96 : SendResult send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
508 [ + + ]: 96 : if (cfg.backpressure_strategy != base::constants::BackpressureStrategy::Reliable)
509 : 16 : return try_send_shared(std::move(data), true);
510 [ + + ]: 149 : return blocking_send(
511 [ + - ]: 153 : data ? data->size() : 0, [&](auto& udp, uint64_t sequence) { return udp.write_shared(data, sequence); },
512 : 121 : [&](auto& connection) { return connection.write_shared(data); });
513 : : }
514 : :
515 : 72 : SendResult send_line(std::string_view line) {
516 [ + + ]: 72 : if (cfg.backpressure_strategy == base::constants::BackpressureStrategy::Reliable) return send_line_blocking(line);
517 : 11 : return try_send_line(line, true);
518 : : }
519 : :
520 : 38 : SendResult try_send_line(std::string_view line, bool best_effort_send = false) {
521 : 114 : return try_send(std::string(line) + "\n", best_effort_send);
522 : : }
523 : :
524 : 330 : SendResult send_blocking(std::string_view data) {
525 : 330 : auto binary_view = base::safe_convert::string_to_bytes(data);
526 : 330 : memory::ConstByteSpan span(binary_view.first, binary_view.second);
527 : 330 : return blocking_send(
528 : 73 : data.size(), [&](auto& udp, uint64_t sequence) { return udp.write_copy(span, sequence); },
529 : 775 : [&](auto& connection) { return connection.write_copy(span); });
530 : : }
531 : :
532 : 381 : SendResult send_line_blocking(std::string_view line) { return send_blocking(std::string(line) + "\n"); }
533 : :
534 : 8 : RuntimeStats stats() const {
535 : 8 : std::shared_lock<std::shared_mutex> lock(mutex_);
536 [ + - + - ]: 16 : return channel ? channel->stats() : RuntimeStats{};
537 : 8 : }
538 : :
539 : 1 : void reset_stats() {
540 : 1 : std::shared_lock<std::shared_mutex> lock(mutex_);
541 : 1 : receive_budget_->reset_stats();
542 [ + - + - ]: 1 : if (channel) channel->reset_stats();
543 : 1 : }
544 : :
545 : 411 : void setup_internal_handlers() {
546 [ - + ]: 411 : if (!channel) return;
547 : :
548 : 411 : batch_timer_ = std::make_unique<boost::asio::steady_timer>(channel->get_executor());
549 : :
550 : 411 : std::weak_ptr<bool> weak_alive = alive_marker;
551 : 411 : std::weak_ptr<Impl> weak_impl = weak_from_this();
552 : 411 : receive_budget_->reset_stats();
553 : 411 : lifecycle_events_.reset();
554 : 411 : const auto generation = callback_gate_.open_new_generation();
555 : 411 : callback_generation_.store(generation);
556 : :
557 : 411 : channel->on_bytes([this, generation, weak_impl, weak_alive](memory::ConstByteSpan data) {
558 : 94 : auto impl_keepalive = weak_impl.lock();
559 [ - + ]: 94 : if (!impl_keepalive) return;
560 : 94 : auto alive = weak_alive.lock();
561 [ + + ]: 94 : if (!alive) return;
562 : 93 : auto lease = callback_gate_.enter(generation);
563 [ + + ]: 93 : if (!lease.admitted()) return;
564 : :
565 : : // #449: everything below runs synchronously on this io thread - mark
566 : : // it so a blocking send() called from within one of these callbacks
567 : : // fails fast instead of deadlocking.
568 : 92 : detail::CallbackGuard callback_guard;
569 : :
570 : : // #441: snapshot the handler/framer pointers under a shared_lock (not
571 : : // unique_lock) - this is a pure read, matching try_send's locking
572 : : // level so it no longer blocks concurrent sends even briefly.
573 : : bool batch_mode;
574 : 92 : interface::SharedCallback<MessageHandler> handler;
575 : 92 : std::shared_ptr<detail::ReceiveState> receive;
576 : 92 : std::shared_ptr<framer::IFramer> framer_to_push;
577 : : {
578 : 92 : std::shared_lock<std::shared_mutex> lock(mutex_);
579 : 92 : batch_mode = static_cast<bool>(data_batch_handler_);
580 : 92 : handler = data_handler;
581 : 92 : framer_to_push = framer;
582 : 92 : receive = receive_state_;
583 : 92 : }
584 : :
585 : : try {
586 : 92 : auto prepared = detail::prepare_receive(*receive, framer_to_push, 0, data, batch_mode);
587 [ + + ]: 88 : if (batch_mode) {
588 : : // #441: build the copy before taking the exclusive lock, so the
589 : : // lock is only held for the queue mutation itself, not the
590 : : // allocation.
591 : 16 : auto ctx = std::move(*prepared.raw);
592 : 16 : interface::SharedCallback<BatchMessageHandler> flush_handler;
593 : 16 : detail::ReceiveBatch batch;
594 : : {
595 : 16 : std::unique_lock<std::shared_mutex> lock(mutex_);
596 : 16 : data_batch_queue_.emplace_back(std::move(ctx));
597 [ + + ]: 16 : if (data_batch_queue_.size() >= max_batch_size_) {
598 : 4 : flush_handler = data_batch_handler_;
599 : 4 : batch = std::move(data_batch_queue_);
600 : 4 : data_batch_queue_.clear();
601 [ + - ]: 12 : } else if (data_batch_queue_.size() == 1) {
602 : 12 : schedule_batch_timer(generation);
603 : : }
604 : 16 : }
605 : 16 : detail::invoke_user_callback("udp_client", "on_data_batch", flush_handler, batch);
606 : 16 : } else {
607 : 144 : detail::invoke_user_callback("udp_client", "on_data", handler, MessageContext(0, data));
608 : : }
609 : :
610 : 88 : prepared.deliver();
611 : 92 : } catch (const detail::ReceiveOverflow& overflow) {
612 : 4 : receive->scope->overflow(data.size(), overflow.reason);
613 : 4 : } catch (const std::bad_alloc&) {
614 : 0 : receive->scope->overflow(data.size(), ReceiveOverflowReason::AllocationFailure);
615 : 0 : }
616 : 97 : });
617 : :
618 : 411 : channel->on_backpressure([this, generation, weak_impl, weak_alive](size_t queued) {
619 : 98 : auto impl_keepalive = weak_impl.lock();
620 [ - + ]: 98 : if (!impl_keepalive) return;
621 : 98 : auto alive = weak_alive.lock();
622 [ - + ]: 98 : if (!alive) return;
623 : 98 : auto lease = callback_gate_.enter(generation);
624 [ + + ]: 98 : if (!lease.admitted()) return;
625 : 74 : bp_cv_.notify_all();
626 : 74 : std::function<void(size_t)> handler;
627 : : {
628 : 74 : std::shared_lock<std::shared_mutex> lock(mutex_);
629 : 74 : handler = bp_handler;
630 : 74 : }
631 : 74 : detail::invoke_user_callback("udp_client", "on_backpressure", handler, queued);
632 : 146 : });
633 : :
634 : 411 : channel->on_state([this, generation, weak_impl, weak_alive](base::LinkState state) {
635 : 593 : auto impl_keepalive = weak_impl.lock();
636 [ - + ]: 593 : if (!impl_keepalive) return;
637 : 593 : auto alive = weak_alive.lock();
638 [ + + ]: 593 : if (!alive) return;
639 : 591 : auto lease = callback_gate_.enter(generation);
640 [ + + ]: 591 : if (!lease.admitted()) return;
641 [ + + + + ]: 441 : const bool ready = state == base::LinkState::Connected || state == base::LinkState::Listening;
642 : 441 : const auto events = lifecycle_events_.observe(state, ready);
643 : 441 : ConnectionHandler connected_handler, lost_handler;
644 : 441 : ErrorHandler terminal_handler;
645 : 441 : std::shared_ptr<interface::Channel> channel_snapshot;
646 [ + + ]: 441 : if (ready) {
647 : 228 : std::unique_lock<std::shared_mutex> lock(mutex_);
648 : 228 : fulfill_all_locked(true);
649 : 228 : connected_handler = connect_handler;
650 : 228 : } else {
651 [ + + + + : 213 : if (state == base::LinkState::Closed || state == base::LinkState::Idle || state == base::LinkState::Error) {
+ + ]
652 : 37 : std::unique_lock<std::shared_mutex> lock(mutex_);
653 : 37 : fulfill_all_locked(false);
654 : 37 : }
655 : : // A Connecting notification must not require the exclusive wrapper
656 : : // lock held off by a connection-pinned final admission on another thread.
657 : 213 : std::shared_lock<std::shared_mutex> lock(mutex_);
658 [ + + + - ]: 213 : if (events.disconnect) lost_handler = disconnect_handler;
659 [ + + ]: 213 : if (events.error) {
660 : 19 : terminal_handler = error_handler;
661 : 19 : channel_snapshot = channel;
662 : : }
663 : 213 : }
664 [ + + ]: 441 : if (events.disconnect) {
665 : 27 : flush_batches(generation);
666 : 27 : std::shared_lock<std::shared_mutex> read_lock(mutex_);
667 [ + + ]: 27 : if (framer) {
668 : 3 : read_lock.unlock();
669 : 3 : std::unique_lock<std::shared_mutex> lock(mutex_);
670 : 3 : receive_state_->reset(framer.get());
671 : 3 : }
672 : 27 : }
673 : 441 : auto notification_lease = callback_gate_.enter(generation);
674 [ + + ]: 441 : if (!notification_lease.admitted()) return;
675 : 880 : detail::invoke_user_callback("udp_client", "on_connect", connected_handler, ConnectionContext(0));
676 : 440 : detail::invoke_user_callback("udp_client", "on_disconnect", lost_handler, ConnectionContext(0));
677 : : // A loss callback may request stop; do not admit the following terminal
678 : : // error into the closed generation.
679 : 440 : auto error_lease = callback_gate_.enter(generation);
680 [ + + + + : 440 : if (error_lease.admitted() && terminal_handler)
+ + ]
681 [ + - ]: 15 : detail::invoke_user_callback("udp_client", "on_error", terminal_handler,
682 : 15 : channel_snapshot
683 [ + - + - : 30 : ? detail::build_error_context(*channel_snapshot, "Connection error")
- - ]
684 : : : ErrorContext(ErrorCode::IoError, "Connection error"));
685 : 902 : });
686 : 411 : }
687 : :
688 : 14 : void attach_framer_callback() {
689 [ - + ]: 14 : if (!framer) return;
690 : 14 : framer->on_message([this](memory::ConstByteSpan msg) {
691 : 111 : const auto generation = callback_generation_.load();
692 : 111 : auto message_lease = callback_gate_.enter(generation);
693 [ - + ]: 111 : if (!message_lease.admitted()) return;
694 : 111 : auto prepared = detail::take_prepared_message(0, msg);
695 : : // #441: snapshot under a shared_lock (pure read), build the copy
696 : : // before taking the exclusive lock for queue mutation.
697 : : bool batch_mode;
698 : 111 : interface::SharedCallback<MessageHandler> handler;
699 : : {
700 : 111 : std::shared_lock<std::shared_mutex> lock(mutex_);
701 : 111 : batch_mode = static_cast<bool>(message_batch_handler_);
702 : 111 : handler = message_handler;
703 : 111 : }
704 : :
705 [ + + ]: 111 : if (batch_mode) {
706 [ + - - - ]: 108 : auto ctx = prepared ? std::move(*prepared) : detail::retain_received(receive_state_->scope, 0, msg);
707 : 108 : interface::SharedCallback<BatchMessageHandler> flush_handler;
708 : 108 : detail::ReceiveBatch batch;
709 : : {
710 : 108 : std::unique_lock<std::shared_mutex> lock(mutex_);
711 : 108 : message_batch_queue_.emplace_back(std::move(ctx));
712 [ + + ]: 108 : if (message_batch_queue_.size() >= max_batch_size_) {
713 : 3 : flush_handler = message_batch_handler_;
714 : 3 : batch = std::move(message_batch_queue_);
715 : 3 : message_batch_queue_.clear();
716 [ + + ]: 105 : } else if (message_batch_queue_.size() == 1) {
717 : 6 : schedule_batch_timer(generation);
718 : : }
719 : 108 : }
720 : 108 : detail::invoke_user_callback("udp_client", "on_message_batch", flush_handler, batch);
721 : 108 : return;
722 : 108 : }
723 : :
724 : 3 : detail::invoke_user_callback("udp_client", "on_message", handler, MessageContext(0, msg));
725 : 327 : });
726 : : }
727 : :
728 : 26 : void set_framer(std::unique_ptr<framer::IFramer> f) {
729 : 26 : std::unique_lock<std::shared_mutex> lock(mutex_);
730 : 26 : auto receive = std::make_shared<detail::ReceiveState>(receive_state_->scope);
731 : 26 : framer = std::shared_ptr<framer::IFramer>(std::move(f));
732 : 26 : receive_state_ = std::move(receive);
733 [ + - + - : 26 : if (framer && (message_handler || message_batch_handler_)) attach_framer_callback();
- + - + -
- ]
734 : 26 : }
735 : :
736 : 6 : void on_message(MessageHandler handler) {
737 : 6 : std::unique_lock<std::shared_mutex> lock(mutex_);
738 : 6 : message_handler = interface::share_callback(std::move(handler));
739 [ + - + - ]: 6 : if (framer) attach_framer_callback();
740 : 6 : }
741 : :
742 : 8 : void on_message_batch(BatchMessageHandler handler) {
743 : 8 : std::unique_lock<std::shared_mutex> lock(mutex_);
744 : 8 : message_batch_handler_ = interface::share_callback(std::move(handler));
745 [ + - + - ]: 8 : if (framer) attach_framer_callback();
746 : 8 : }
747 : : };
748 : :
749 : 79 : UdpClient::UdpClient(const config::UdpConfig& cfg) : impl_(std::make_shared<Impl>(cfg)) {
750 : 79 : config::detail::validate(cfg);
751 : 80 : }
752 : 5 : UdpClient::UdpClient(const config::UdpConfig& cfg, std::shared_ptr<boost::asio::io_context> ioc)
753 : 5 : : impl_(std::make_shared<Impl>(cfg, ioc)) {
754 : 5 : config::detail::validate(cfg);
755 : 5 : }
756 : 191 : UdpClient::UdpClient(std::shared_ptr<interface::Channel> ch) : impl_(std::make_shared<Impl>(ch)) {
757 : 188 : impl_->setup_internal_handlers();
758 : 191 : }
759 : : // Stop on the destroying thread first. A callback in flight holds the Impl
760 : : // alive, and if its reference were the last one, ~Impl would run on the io
761 : : // thread, where the shutdown cannot join that thread (#613).
762 : 399 : UdpClient::~UdpClient() {
763 [ - + ]: 271 : if (!impl_) return;
764 : : try {
765 : 271 : impl_->stop();
766 : 0 : } catch (...) {
767 : 0 : }
768 : 399 : }
769 : :
770 : 0 : UdpClient::UdpClient(UdpClient&&) noexcept = default;
771 : : // The previous Impl is handed to a temporary, whose destructor stops it here.
772 : 0 : UdpClient& UdpClient::operator=(UdpClient&& other) noexcept {
773 [ # # ]: 0 : if (this != &other) {
774 : 0 : UdpClient previous(std::move(*this));
775 : 0 : impl_ = std::move(other.impl_);
776 : 0 : }
777 : 0 : return *this;
778 : : }
779 : :
780 : 8 : UdpClient& UdpClient::receive_limits(ReceiveLimits limits) {
781 : 8 : limits.validate();
782 : 7 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
783 [ + + + - : 19 : if (impl_->started_ || impl_->stop_callers_.load() || (impl_->stop_requested_ && impl_->alive_marker) ||
+ + + - -
+ + + ]
784 : 6 : detail::in_data_callback())
785 : 1 : throw std::logic_error("receive limits require completed stop");
786 : 6 : auto budget = std::make_shared<detail::ReceiveBudget>(limits);
787 : 6 : auto scope = budget->open_scope();
788 : 6 : auto receive = std::make_shared<detail::ReceiveState>(std::move(scope));
789 : 6 : impl_->receive_state_->reset(impl_->framer.get());
790 : 6 : impl_->receive_limits_ = limits;
791 : 6 : impl_->receive_budget_ = std::move(budget);
792 : 6 : impl_->receive_state_ = std::move(receive);
793 : 6 : return *this;
794 : 7 : }
795 : 30 : ReceiveMemoryStats UdpClient::receive_stats() const {
796 : 30 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
797 : 60 : return impl_->receive_budget_->stats();
798 : 30 : }
799 : :
800 : 282 : std::future<bool> UdpClient::start() { return impl_->start(); }
801 : 404 : void UdpClient::stop() { impl_->stop(); }
802 : 142 : SendResult UdpClient::send(std::string_view data) { return impl_->send(data); }
803 : 43 : SendResult UdpClient::try_send(std::string_view data) { return impl_->try_send(data); }
804 : 72 : SendResult UdpClient::send_line(std::string_view line) { return impl_->send_line(line); }
805 : 27 : SendResult UdpClient::try_send_line(std::string_view line) { return impl_->try_send_line(line); }
806 : 84 : SendResult UdpClient::send_move(std::vector<uint8_t>&& data) { return impl_->send_move(std::move(data)); }
807 : 27 : SendResult UdpClient::try_send_move(std::vector<uint8_t>&& data) { return impl_->try_send_move(std::move(data)); }
808 : 96 : SendResult UdpClient::send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
809 : 96 : return impl_->send_shared(std::move(data));
810 : : }
811 : 33 : SendResult UdpClient::try_send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
812 : 33 : return impl_->try_send_shared(std::move(data));
813 : : }
814 : 74 : SendResult UdpClient::send_blocking(std::string_view data) { return impl_->send_blocking(data); }
815 : 66 : SendResult UdpClient::send_line_blocking(std::string_view line) { return impl_->send_line_blocking(line); }
816 : 58 : bool UdpClient::connected() const {
817 : 58 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
818 [ + + + - : 116 : return impl_->channel && impl_->channel->is_connected();
+ + ]
819 : 58 : }
820 : 8 : RuntimeStats UdpClient::stats() const { return impl_->stats(); }
821 : 1 : void UdpClient::reset_stats() { impl_->reset_stats(); }
822 : :
823 : 36 : UdpClient& UdpClient::on_data(MessageHandler h) {
824 : 36 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
825 : 36 : impl_->data_handler = interface::share_callback(std::move(h));
826 : 36 : return *this;
827 : 36 : }
828 : 15 : UdpClient& UdpClient::on_data_batch(BatchMessageHandler h) {
829 : 15 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
830 : 15 : impl_->data_batch_handler_ = interface::share_callback(std::move(h));
831 : 15 : return *this;
832 : 15 : }
833 : 19 : UdpClient& UdpClient::on_connect(ConnectionHandler h) {
834 : 19 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
835 : 19 : impl_->connect_handler = std::move(h);
836 : 19 : return *this;
837 : 19 : }
838 : 18 : UdpClient& UdpClient::on_disconnect(ConnectionHandler h) {
839 : 18 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
840 : 18 : impl_->disconnect_handler = std::move(h);
841 : 18 : return *this;
842 : 18 : }
843 : 26 : UdpClient& UdpClient::on_error(ErrorHandler h) {
844 : 26 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
845 : 26 : impl_->error_handler = std::move(h);
846 : 26 : return *this;
847 : 26 : }
848 : :
849 : 4 : UdpClient& UdpClient::on_backpressure(std::function<void(size_t)> h) {
850 : 4 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
851 : 4 : impl_->bp_handler = std::move(h);
852 : 4 : return *this;
853 : 4 : }
854 : :
855 : 26 : UdpClient& UdpClient::framer(std::unique_ptr<framer::IFramer> f) {
856 : 26 : impl_->set_framer(std::move(f));
857 : 26 : return *this;
858 : : }
859 : 6 : UdpClient& UdpClient::on_message(MessageHandler h) {
860 : 6 : impl_->on_message(std::move(h));
861 : 6 : return *this;
862 : : }
863 : 8 : UdpClient& UdpClient::on_message_batch(BatchMessageHandler h) {
864 : 8 : impl_->on_message_batch(std::move(h));
865 : 8 : return *this;
866 : : }
867 : :
868 : 4 : UdpClient& UdpClient::auto_start(bool m) {
869 : 4 : impl_->auto_start_.store(m);
870 [ + + + - : 4 : if (impl_->auto_start_.load() && !impl_->started_.load()) start();
+ + + - ]
871 : 4 : return *this;
872 : : }
873 : :
874 : 17 : UdpClient& UdpClient::backpressure_threshold(size_t threshold) {
875 : 17 : config::detail::range(threshold, base::constants::MIN_BACKPRESSURE_THRESHOLD,
876 : : base::constants::MAX_BACKPRESSURE_THRESHOLD, "invalid backpressure_threshold");
877 : 15 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
878 : 15 : impl_->require_stopped_config();
879 : 13 : impl_->cfg.backpressure_threshold = threshold;
880 : 13 : return *this;
881 : 15 : }
882 : :
883 : 110 : UdpClient& UdpClient::backpressure_strategy(base::constants::BackpressureStrategy strategy) {
884 : 110 : config::detail::strategy(strategy);
885 : 109 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
886 : 109 : impl_->cfg.backpressure_strategy = strategy;
887 [ + + ]: 109 : if (impl_->channel) {
888 [ + - + + ]: 103 : if (auto* uc = dynamic_cast<transport::UdpChannel*>(impl_->channel.get())) {
889 : 66 : uc->set_backpressure_strategy(strategy);
890 : : }
891 : : }
892 : 109 : return *this;
893 : 109 : }
894 : :
895 : 7 : size_t UdpClient::backpressure_threshold() const {
896 : 7 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
897 : 14 : return impl_->cfg.backpressure_threshold;
898 : 7 : }
899 : :
900 : 3 : base::constants::BackpressureStrategy UdpClient::backpressure_strategy() const {
901 : 3 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
902 : 6 : return impl_->cfg.backpressure_strategy;
903 : 3 : }
904 : :
905 : 0 : UdpClient& UdpClient::send_buffer_size(size_t bytes) {
906 [ # # ]: 0 : if (bytes != 0)
907 : 0 : config::detail::range(bytes, base::constants::MIN_SOCKET_BUFFER_SIZE, base::constants::MAX_SOCKET_BUFFER_SIZE,
908 : : "invalid send_buffer_size");
909 : 0 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
910 : 0 : impl_->require_stopped_config();
911 : 0 : impl_->cfg.send_buffer_size = bytes;
912 : 0 : return *this;
913 : 0 : }
914 : :
915 : 0 : UdpClient& UdpClient::receive_buffer_size(size_t bytes) {
916 [ # # ]: 0 : if (bytes != 0)
917 : 0 : config::detail::range(bytes, base::constants::MIN_SOCKET_BUFFER_SIZE, base::constants::MAX_SOCKET_BUFFER_SIZE,
918 : : "invalid receive_buffer_size");
919 : 0 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
920 : 0 : impl_->require_stopped_config();
921 : 0 : impl_->cfg.receive_buffer_size = bytes;
922 : 0 : return *this;
923 : 0 : }
924 : :
925 : 7 : UdpClient& UdpClient::manage_external_context(bool m) {
926 : 7 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
927 : 7 : impl_->require_stopped_config();
928 : 6 : impl_->manage_external_context.store(m);
929 : 6 : return *this;
930 : 7 : }
931 : :
932 : 28 : UdpClient& UdpClient::batch_size(size_t size) {
933 [ + + ]: 28 : config::detail::require(size > 0, "batch size must be positive");
934 : 27 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
935 : 27 : impl_->max_batch_size_ = size;
936 : 27 : return *this;
937 : 27 : }
938 : :
939 : 26 : UdpClient& UdpClient::batch_latency(std::chrono::milliseconds latency) {
940 : 26 : config::detail::duration(latency, 0, std::numeric_limits<int>::max(), true, "invalid batch latency");
941 : 25 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
942 : 25 : impl_->max_batch_latency_ = latency;
943 : 25 : return *this;
944 : 25 : }
945 : :
946 : : } // namespace wrapper
947 : : } // namespace wirestead
|
