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 <chrono>
31 : : #include <iostream>
32 : : #include <mutex>
33 : : #include <shared_mutex>
34 : : #include <stdexcept>
35 : : #include <thread>
36 : : #include <vector>
37 : :
38 : : #include "wirestead/base/common.hpp"
39 : : #include "wirestead/factory/channel_factory.hpp"
40 : : #include "wirestead/transport/udp/udp.hpp"
41 : : #include "wirestead/wrapper/callback_guard.hpp"
42 : : #include "wirestead/wrapper/error_context_builder.hpp"
43 : :
44 : : namespace wirestead {
45 : : namespace wrapper {
46 : :
47 : : struct UdpClient::Impl : public std::enable_shared_from_this<Impl> {
48 : : mutable std::shared_mutex mutex_;
49 : : std::mutex bp_mutex_;
50 : : std::condition_variable bp_cv_;
51 : : config::UdpConfig cfg;
52 : : std::shared_ptr<interface::Channel> channel;
53 : : std::shared_ptr<boost::asio::io_context> external_ioc;
54 : : std::atomic<bool> use_external_context{false};
55 : : std::atomic<bool> manage_external_context{false};
56 : : std::thread external_thread;
57 : : std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>> work_guard;
58 : :
59 : : // Event handlers (Context based)
60 : : // Shared snapshots: the strand copies one out per received datagram, and a
61 : : // std::function copy allocates whenever the user handler outgrows its
62 : : // small-object buffer. See interface::SharedCallback.
63 : : interface::SharedCallback<MessageHandler> data_handler;
64 : : interface::SharedCallback<BatchMessageHandler> data_batch_handler_;
65 : : ConnectionHandler connect_handler{nullptr};
66 : : ConnectionHandler disconnect_handler{nullptr};
67 : : ErrorHandler error_handler{nullptr};
68 : : std::function<void(size_t)> bp_handler{nullptr};
69 : : interface::SharedCallback<MessageHandler> message_handler;
70 : : interface::SharedCallback<BatchMessageHandler> message_batch_handler_;
71 : :
72 : : std::shared_ptr<framer::IFramer> framer{nullptr};
73 : :
74 : : // Batching logic
75 : : std::vector<MessageContext> data_batch_queue_;
76 : : std::vector<MessageContext> message_batch_queue_;
77 : : std::unique_ptr<boost::asio::steady_timer> batch_timer_;
78 : : size_t max_batch_size_ = 100;
79 : 49 : std::chrono::milliseconds max_batch_latency_{1};
80 : :
81 : : std::atomic<bool> auto_start_{false};
82 : : std::vector<std::promise<bool>> pending_promises;
83 : : std::atomic<bool> started_{false};
84 : 49 : std::shared_ptr<bool> alive_marker{std::make_shared<bool>(true)};
85 : :
86 : : // False only for the dependency-injected-channel constructor below, where
87 : : // the caller owns the channel's identity and lifecycle (e.g. tests
88 : : // injecting a fake channel) - stop() must not discard and factory-rebuild
89 : : // a channel it didn't create itself.
90 : : bool factory_managed_channel_ = true;
91 : :
92 : 29 : explicit Impl(const config::UdpConfig& config) : cfg(config) {}
93 : 5 : Impl(const config::UdpConfig& config, std::shared_ptr<boost::asio::io_context> ioc)
94 [ + - + - ]: 5 : : cfg(config), external_ioc(std::move(ioc)), use_external_context(external_ioc != nullptr) {}
95 : 15 : explicit Impl(std::shared_ptr<interface::Channel> ch) : channel(std::move(ch)), factory_managed_channel_(false) {
96 : : // #450: setup_internal_handlers() captures weak_from_this() - calling it
97 : : // from inside this constructor would capture an empty weak_ptr, since
98 : : // enable_shared_from_this isn't wired up until make_shared() finishes
99 : : // constructing the object. Deferred to UdpClient's own constructor,
100 : : // which runs after impl_ is a fully-formed shared_ptr<Impl>.
101 : 15 : }
102 : :
103 : 49 : ~Impl() {
104 : : try {
105 : 49 : stop();
106 : 0 : } catch (...) {
107 : 0 : }
108 : 49 : }
109 : :
110 : 113 : void fulfill_all_locked(bool value) {
111 [ + + ]: 140 : for (auto& promise : pending_promises) {
112 : : try {
113 : 27 : promise.set_value(value);
114 : 0 : } catch (...) {
115 : 0 : }
116 : : }
117 : 113 : pending_promises.clear();
118 : 113 : }
119 : :
120 : 1 : void flush_batches() {
121 : 1 : std::unique_lock<std::shared_mutex> lock(mutex_);
122 [ + - ]: 1 : if (!data_batch_queue_.empty()) {
123 : 1 : auto handler = data_batch_handler_;
124 : 1 : auto batch = std::move(data_batch_queue_);
125 : 1 : data_batch_queue_.clear();
126 [ + - ]: 1 : if (handler) {
127 : 1 : lock.unlock();
128 : 1 : detail::invoke_user_callback("udp_client", "on_data_batch", handler, batch);
129 : 1 : lock.lock();
130 : : }
131 : 1 : }
132 [ + - ]: 1 : if (!message_batch_queue_.empty()) {
133 : 1 : auto handler = message_batch_handler_;
134 : 1 : auto batch = std::move(message_batch_queue_);
135 : 1 : message_batch_queue_.clear();
136 [ + - ]: 1 : if (handler) {
137 : 1 : lock.unlock();
138 : 1 : detail::invoke_user_callback("udp_client", "on_message_batch", handler, batch);
139 : 1 : lock.lock();
140 : : }
141 : 1 : }
142 [ + - ]: 1 : if (batch_timer_) {
143 : 1 : batch_timer_->cancel();
144 : : }
145 : 1 : }
146 : :
147 : 6 : void schedule_batch_timer() {
148 [ - + ]: 6 : if (!batch_timer_) return;
149 : 6 : batch_timer_->expires_after(max_batch_latency_);
150 : 12 : batch_timer_->async_wait([this, weak_impl = weak_from_this(),
151 : 6 : weak_alive = std::weak_ptr<bool>(alive_marker)](const boost::system::error_code& ec) {
152 [ + + ]: 2 : if (ec) return;
153 : 1 : auto impl_keepalive = weak_impl.lock();
154 [ - + ]: 1 : if (!impl_keepalive) return;
155 : 1 : auto alive = weak_alive.lock();
156 [ - + ]: 1 : if (!alive) return;
157 : 1 : flush_batches();
158 : 1 : });
159 : : }
160 : :
161 : 32 : std::future<bool> start() {
162 : 32 : std::unique_lock<std::shared_mutex> lock(mutex_);
163 [ + + + - : 32 : if (channel && channel->is_connected()) {
+ + + + ]
164 : 5 : started_.store(true);
165 : 5 : std::promise<bool> p;
166 : 5 : p.set_value(true);
167 : 5 : return p.get_future();
168 : 5 : }
169 : :
170 : 27 : std::promise<bool> p;
171 : 27 : auto future = p.get_future();
172 : 27 : pending_promises.emplace_back(std::move(p));
173 : :
174 [ + + ]: 27 : if (started_.load()) {
175 : 1 : return future;
176 : : }
177 : :
178 [ + + ]: 26 : if (!channel) {
179 : 20 : channel = factory::ChannelFactory::create(cfg, external_ioc);
180 : 20 : setup_internal_handlers();
181 : : }
182 : 26 : started_.store(true);
183 : :
184 : 26 : lock.unlock();
185 : 26 : channel->start();
186 [ + + + + : 26 : if (use_external_context && manage_external_context && !external_thread.joinable()) {
+ - + + ]
187 [ + - - + : 1 : if (external_ioc->stopped()) external_ioc->restart();
- - ]
188 : 2 : work_guard = std::make_unique<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>(
189 : 3 : boost::asio::make_work_guard(*external_ioc));
190 : 2 : external_thread = std::thread([this, ioc = external_ioc]() {
191 : 1 : wirestead::concurrency::run_io_thread_init();
192 : : try {
193 : 1 : ioc->run();
194 : 0 : } catch (...) {
195 : 0 : }
196 : 2 : });
197 : : }
198 : 26 : return future;
199 : 32 : }
200 : :
201 : 74 : void stop() {
202 : 74 : std::unique_lock<std::shared_mutex> lock(mutex_);
203 [ + + ]: 74 : if (!started_.load()) {
204 : 44 : fulfill_all_locked(false);
205 : 44 : return;
206 : : }
207 : 30 : started_.store(false);
208 : 30 : bp_cv_.notify_all();
209 : :
210 [ + - ]: 30 : if (batch_timer_) {
211 : 30 : batch_timer_->cancel();
212 : 30 : batch_timer_.reset();
213 : : }
214 : :
215 [ + - ]: 30 : if (channel) {
216 : 30 : lock.unlock();
217 : 30 : channel->stop();
218 : : // Clear callbacks after stop() rather than before: the transport-level
219 : : // fix (#436) already synchronizes callback reads against these
220 : : // setters, but clearing after stop() means no in-flight handler can
221 : : // observe a null callback mid-shutdown in the first place - belt and
222 : : // braces once the underlying race is fixed at the source.
223 : 30 : channel->on_bytes(nullptr);
224 : 30 : channel->on_state(nullptr);
225 : 30 : channel->on_backpressure(nullptr);
226 : 30 : lock.lock();
227 [ + + ]: 30 : if (factory_managed_channel_) {
228 : : // Fully release the channel rather than reusing it: start()'s
229 : : // `if (!channel)` guard is what re-runs setup_internal_handlers() and
230 : : // rebuilds config from the (possibly changed) staged fields. Reusing
231 : : // a stopped channel left every handler nulled forever - including
232 : : // the one that fulfills the start() future - so a restart would
233 : : // hang (jwsung91/wirestead#444). Injected channels (factory_managed_
234 : : // channel_ == false) are exempt: the caller owns that channel's
235 : : // identity, so we must not discard and factory-rebuild it.
236 : 20 : channel.reset();
237 : : }
238 : : }
239 : :
240 [ + + ]: 30 : if (work_guard) {
241 : 1 : work_guard.reset();
242 : : }
243 : :
244 [ + + + + : 30 : if (use_external_context && manage_external_context && external_thread.joinable()) {
+ - + + ]
245 [ + - ]: 1 : if (external_ioc) {
246 : 1 : external_ioc->stop();
247 : : }
248 [ + - ]: 1 : if (std::this_thread::get_id() != external_thread.get_id()) {
249 : 1 : lock.unlock();
250 : 1 : external_thread.join();
251 : 1 : lock.lock();
252 : : } else {
253 : 0 : external_thread.detach();
254 : : }
255 : : }
256 : :
257 : 30 : fulfill_all_locked(false);
258 : :
259 [ + + ]: 30 : if (framer) {
260 : 2 : framer->reset();
261 : : }
262 : 74 : }
263 : :
264 : 13 : bool send(std::string_view data) {
265 [ + + ]: 13 : if (cfg.backpressure_strategy == base::constants::BackpressureStrategy::Reliable) return send_blocking(data);
266 : 1 : return try_send(data);
267 : : }
268 : :
269 : : // #509: wait_for_backpressure_clear()'s condition and the transport's own
270 : : // hard queue-byte cap are different thresholds observed at different
271 : : // times, so a single write attempt can spuriously fail right after the
272 : : // wait exits. Bounded retry rather than unbounded, so a payload that can
273 : : // never fit still fails in bounded time. If a single write is rejected
274 : : // because the payload alone exceeds the transport's cap, UdpChannel also
275 : : // transitions to LinkState::Error (unlike TCP/UDS) - wait_for_backpressure_
276 : : // clear()'s own is_connected() check catches that and ends the retry loop
277 : : // after one real attempt, so this doesn't retry into a broken channel.
278 : : static constexpr int kMaxBlockingSendAttempts = 5;
279 : :
280 : 2 : bool send_move(std::vector<uint8_t>&& data) {
281 [ + + ]: 2 : if (cfg.backpressure_strategy == base::constants::BackpressureStrategy::Reliable) {
282 [ + - ]: 1 : for (int attempt = 0; attempt < kMaxBlockingSendAttempts; ++attempt) {
283 : 1 : std::unique_lock<std::mutex> bp_lock(bp_mutex_);
284 [ + - - + ]: 1 : if (!wait_for_backpressure_clear(bp_lock)) return false;
285 : 1 : bp_lock.unlock();
286 : 1 : std::shared_lock<std::shared_mutex> lock(mutex_);
287 [ + - + - : 1 : if (!started_.load() || !channel || !channel->is_connected()) return false;
+ - - + -
+ ]
288 [ + - + - ]: 1 : if (channel->async_write_move(std::move(data))) return true;
289 : 2 : }
290 : 0 : return false;
291 : : }
292 : 1 : return try_send_move(std::move(data));
293 : : }
294 : :
295 : 2 : bool send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
296 [ + - - + : 2 : if (!data || data->empty()) return false;
- + ]
297 [ + + ]: 2 : if (cfg.backpressure_strategy == base::constants::BackpressureStrategy::Reliable) {
298 [ + - ]: 1 : for (int attempt = 0; attempt < kMaxBlockingSendAttempts; ++attempt) {
299 : 1 : std::unique_lock<std::mutex> bp_lock(bp_mutex_);
300 [ + - - + ]: 1 : if (!wait_for_backpressure_clear(bp_lock)) return false;
301 : 1 : bp_lock.unlock();
302 : 1 : std::shared_lock<std::shared_mutex> lock(mutex_);
303 [ + - + - : 1 : if (!started_.load() || !channel || !channel->is_connected()) return false;
+ - - + -
+ ]
304 [ + - + - ]: 1 : if (channel->async_write_shared(data)) return true;
305 : 2 : }
306 : 0 : return false;
307 : : }
308 : 1 : return try_send_shared(std::move(data));
309 : : }
310 : :
311 : 5 : bool send_line(std::string_view line) {
312 [ + + ]: 5 : if (cfg.backpressure_strategy == base::constants::BackpressureStrategy::Reliable) return send_line_blocking(line);
313 : 1 : return try_send_line(line);
314 : : }
315 : :
316 : 6 : bool try_send_line(std::string_view line) { return try_send(std::string(line) + "\n"); }
317 : :
318 : 20 : bool send_blocking(std::string_view data) {
319 : 20 : auto binary_view = base::safe_convert::string_to_bytes(data);
320 : 20 : memory::ConstByteSpan span(binary_view.first, binary_view.second);
321 [ + - ]: 21 : for (int attempt = 0; attempt < kMaxBlockingSendAttempts; ++attempt) {
322 : 21 : std::unique_lock<std::mutex> bp_lock(bp_mutex_);
323 [ + - - + ]: 21 : if (!wait_for_backpressure_clear(bp_lock)) return false;
324 : 21 : bp_lock.unlock();
325 : 21 : std::shared_lock<std::shared_mutex> lock(mutex_);
326 [ + + + - : 21 : if (!started_.load() || !channel || !channel->is_connected()) return false;
+ - - + +
+ ]
327 [ + - + + ]: 15 : if (channel->async_write_copy(span)) return true;
328 : 41 : }
329 : 0 : return false;
330 : : }
331 : :
332 : : // channel->on_backpressure() calls bp_cv_.notify_all() from the transport's io_context
333 : : // thread without holding bp_mutex_ (backpressure_active_ is a plain atomic on the transport
334 : : // side, not guarded by bp_mutex_ at all). That makes a classic lost-wakeup race possible: a
335 : : // waiter can check the predicate, find it still blocking, and be in the process of
336 : : // registering to wait when the notify fires - in the rare case that race is lost, an
337 : : // unbounded wait() would block forever. Poll with a bounded timeout instead so a missed
338 : : // notify only costs a short delay rather than a permanent hang (see #427).
339 : : //
340 : : // Returns false instead of waiting if called from the channel's own io
341 : : // thread while backpressure is active - e.g. a blocking send() called
342 : : // from inside an on_data/on_message callback. Clearing backpressure
343 : : // requires that same io thread to make progress, so blocking here would
344 : : // deadlock forever rather than eventually clear (#449).
345 : 23 : bool wait_for_backpressure_clear(std::unique_lock<std::mutex>& bp_lock) {
346 : 27 : auto predicate = [this] {
347 : 27 : std::shared_lock<std::shared_mutex> lock(mutex_);
348 [ + + + - : 54 : return !started_.load() || !channel || !channel->is_connected() || !channel->is_backpressure_active();
+ - + - +
- + + ]
349 : 27 : };
350 [ + - + + ]: 23 : if (predicate()) return true;
351 [ - + ]: 1 : if (detail::in_data_callback()) return false;
352 [ + - + + ]: 2 : while (!bp_cv_.wait_for(bp_lock, std::chrono::milliseconds(50), predicate)) {
353 : : }
354 : 1 : return true;
355 : : }
356 : :
357 : 5 : bool try_send(std::string_view data) {
358 : 5 : std::shared_lock<std::shared_mutex> lock(mutex_);
359 [ + - + - : 5 : if (channel && channel->is_connected()) {
+ - + - ]
360 : 5 : auto binary_view = base::safe_convert::string_to_bytes(data);
361 : 5 : return channel->async_try_write_copy(memory::ConstByteSpan(binary_view.first, binary_view.second));
362 : : }
363 : 0 : return false;
364 : 5 : }
365 : :
366 : 3 : bool try_send_move(std::vector<uint8_t>&& data) {
367 : 3 : std::shared_lock<std::shared_mutex> lock(mutex_);
368 [ + - + - : 3 : if (channel && channel->is_connected()) {
+ - + - ]
369 : 3 : return channel->async_try_write_move(std::move(data));
370 : : }
371 : 0 : return false;
372 : 3 : }
373 : :
374 : 5 : bool try_send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
375 [ + + + + : 5 : if (!data || data->empty()) return false;
+ + ]
376 : 3 : std::shared_lock<std::shared_mutex> lock(mutex_);
377 [ + - + - : 3 : if (channel && channel->is_connected()) {
+ - + - ]
378 : 3 : return channel->async_try_write_shared(std::move(data));
379 : : }
380 : 0 : return false;
381 : 3 : }
382 : :
383 : 12 : bool send_line_blocking(std::string_view line) { return send_blocking(std::string(line) + "\n"); }
384 : :
385 : 8 : RuntimeStats stats() const {
386 : 8 : std::shared_lock<std::shared_mutex> lock(mutex_);
387 [ + - + - ]: 16 : return channel ? channel->stats() : RuntimeStats{};
388 : 8 : }
389 : :
390 : 0 : void reset_stats() {
391 : 0 : std::shared_lock<std::shared_mutex> lock(mutex_);
392 [ # # # # ]: 0 : if (channel) channel->reset_stats();
393 : 0 : }
394 : :
395 : 35 : void setup_internal_handlers() {
396 [ - + ]: 35 : if (!channel) return;
397 : :
398 : 35 : batch_timer_ = std::make_unique<boost::asio::steady_timer>(channel->get_executor());
399 : :
400 : 35 : std::weak_ptr<bool> weak_alive = alive_marker;
401 : 35 : std::weak_ptr<Impl> weak_impl = weak_from_this();
402 : :
403 : 35 : channel->on_bytes([this, weak_impl, weak_alive](memory::ConstByteSpan data) {
404 : 13 : auto impl_keepalive = weak_impl.lock();
405 [ - + ]: 13 : if (!impl_keepalive) return;
406 : 13 : auto alive = weak_alive.lock();
407 [ - + ]: 13 : if (!alive) return;
408 : :
409 : : // #449: everything below runs synchronously on this io thread - mark
410 : : // it so a blocking send() called from within one of these callbacks
411 : : // fails fast instead of deadlocking.
412 : 13 : detail::CallbackGuard callback_guard;
413 : :
414 : : // #441: snapshot the handler/framer pointers under a shared_lock (not
415 : : // unique_lock) - this is a pure read, matching try_send's locking
416 : : // level so it no longer blocks concurrent sends even briefly.
417 : : bool batch_mode;
418 : 13 : interface::SharedCallback<MessageHandler> handler;
419 : 13 : std::shared_ptr<framer::IFramer> framer_to_push;
420 : : {
421 : 13 : std::shared_lock<std::shared_mutex> lock(mutex_);
422 : 13 : batch_mode = static_cast<bool>(data_batch_handler_);
423 : 13 : handler = data_handler;
424 : 13 : framer_to_push = framer;
425 : 13 : }
426 : :
427 [ + + ]: 13 : if (batch_mode) {
428 : : // #441: build the copy before taking the exclusive lock, so the
429 : : // lock is only held for the queue mutation itself, not the
430 : : // allocation.
431 : 4 : MessageContext ctx(0, memory::SafeDataBuffer(data));
432 : 4 : interface::SharedCallback<BatchMessageHandler> flush_handler;
433 : 4 : std::vector<MessageContext> batch;
434 : : {
435 : 4 : std::unique_lock<std::shared_mutex> lock(mutex_);
436 : 4 : data_batch_queue_.emplace_back(std::move(ctx));
437 [ + + ]: 4 : if (data_batch_queue_.size() >= max_batch_size_) {
438 : 1 : flush_handler = data_batch_handler_;
439 : 1 : batch = std::move(data_batch_queue_);
440 : 1 : data_batch_queue_.clear();
441 [ + - ]: 3 : } else if (data_batch_queue_.size() == 1) {
442 : 3 : schedule_batch_timer();
443 : : }
444 : 4 : }
445 : 4 : detail::invoke_user_callback("udp_client", "on_data_batch", flush_handler, batch);
446 : 4 : } else {
447 : 18 : detail::invoke_user_callback("udp_client", "on_data", handler, MessageContext(0, data));
448 : : }
449 : :
450 [ + + + - ]: 13 : if (framer_to_push) framer_to_push->push_bytes(data);
451 : 13 : });
452 : :
453 : 35 : channel->on_backpressure([this, weak_impl, weak_alive](size_t queued) {
454 : 1 : bp_cv_.notify_all();
455 : 1 : auto impl_keepalive = weak_impl.lock();
456 [ - + ]: 1 : if (!impl_keepalive) return;
457 : 1 : auto alive = weak_alive.lock();
458 [ - + ]: 1 : if (!alive) return;
459 : 1 : std::function<void(size_t)> handler;
460 : : {
461 : 1 : std::shared_lock<std::shared_mutex> lock(mutex_);
462 : 1 : handler = bp_handler;
463 : 1 : }
464 : 1 : detail::invoke_user_callback("udp_client", "on_backpressure", handler, queued);
465 : 1 : });
466 : :
467 : 35 : channel->on_state([this, weak_impl, weak_alive](base::LinkState state) {
468 : 63 : auto impl_keepalive = weak_impl.lock();
469 [ + + ]: 63 : if (!impl_keepalive) return;
470 : 57 : auto alive = weak_alive.lock();
471 [ - + ]: 57 : if (!alive) return;
472 : :
473 [ + + + ]: 57 : switch (state) {
474 : 21 : case base::LinkState::Connected:
475 : : case base::LinkState::Listening: {
476 : 21 : ConnectionHandler handler;
477 : : {
478 : 21 : std::unique_lock<std::shared_mutex> lock(mutex_);
479 : 21 : fulfill_all_locked(true);
480 : 21 : handler = connect_handler;
481 : 21 : }
482 : 21 : detail::invoke_user_callback("udp_client", "on_connect", handler, ConnectionContext(0));
483 : 21 : break;
484 : 21 : }
485 : 18 : case base::LinkState::Closed:
486 : : case base::LinkState::Error:
487 : : case base::LinkState::Idle: {
488 : 18 : ConnectionHandler disconnect_handler_snapshot;
489 : 18 : ErrorHandler error_handler_snapshot;
490 : : {
491 : 18 : std::unique_lock<std::shared_mutex> lock(mutex_);
492 : 18 : fulfill_all_locked(false);
493 [ + + ]: 18 : if (state == base::LinkState::Error) {
494 : 3 : error_handler_snapshot = error_handler;
495 : : } else {
496 : 15 : disconnect_handler_snapshot = disconnect_handler;
497 : : }
498 : 18 : }
499 : 18 : detail::invoke_user_callback("udp_client", "on_disconnect", disconnect_handler_snapshot,
500 : 36 : ConnectionContext(0));
501 : 18 : detail::invoke_user_callback("udp_client", "on_error", error_handler_snapshot,
502 : 36 : detail::build_error_context(*channel, "Connection error"));
503 : 18 : break;
504 : 18 : }
505 : 18 : default:
506 : 18 : break;
507 : : }
508 : 63 : });
509 : 35 : }
510 : :
511 : 7 : void attach_framer_callback() {
512 [ - + ]: 7 : if (!framer) return;
513 : 7 : framer->on_message([this](memory::ConstByteSpan msg) {
514 : : // #441: snapshot under a shared_lock (pure read), build the copy
515 : : // before taking the exclusive lock for queue mutation.
516 : : bool batch_mode;
517 : 103 : interface::SharedCallback<MessageHandler> handler;
518 : : {
519 : 103 : std::shared_lock<std::shared_mutex> lock(mutex_);
520 : 103 : batch_mode = static_cast<bool>(message_batch_handler_);
521 : 103 : handler = message_handler;
522 : 103 : }
523 : :
524 [ + - ]: 103 : if (batch_mode) {
525 : 103 : MessageContext ctx(0, memory::SafeDataBuffer(msg));
526 : 103 : interface::SharedCallback<BatchMessageHandler> flush_handler;
527 : 103 : std::vector<MessageContext> batch;
528 : : {
529 : 103 : std::unique_lock<std::shared_mutex> lock(mutex_);
530 : 103 : message_batch_queue_.emplace_back(std::move(ctx));
531 [ + + ]: 103 : if (message_batch_queue_.size() >= max_batch_size_) {
532 : 2 : flush_handler = message_batch_handler_;
533 : 2 : batch = std::move(message_batch_queue_);
534 : 2 : message_batch_queue_.clear();
535 [ + + ]: 101 : } else if (message_batch_queue_.size() == 1) {
536 : 3 : schedule_batch_timer();
537 : : }
538 : 103 : }
539 : 103 : detail::invoke_user_callback("udp_client", "on_message_batch", flush_handler, batch);
540 : 103 : return;
541 : 103 : }
542 : :
543 : 0 : detail::invoke_user_callback("udp_client", "on_message", handler, MessageContext(0, msg));
544 : 103 : });
545 : : }
546 : :
547 : 7 : void set_framer(std::unique_ptr<framer::IFramer> f) {
548 : 7 : std::unique_lock<std::shared_mutex> lock(mutex_);
549 : 7 : framer = std::shared_ptr<framer::IFramer>(std::move(f));
550 [ + - + - : 7 : if (framer && (message_handler || message_batch_handler_)) attach_framer_callback();
- + - + -
- ]
551 : 7 : }
552 : :
553 : 3 : void on_message(MessageHandler handler) {
554 : 3 : std::unique_lock<std::shared_mutex> lock(mutex_);
555 : 3 : message_handler = interface::share_callback(std::move(handler));
556 [ + - + - ]: 3 : if (framer) attach_framer_callback();
557 : 3 : }
558 : :
559 : 4 : void on_message_batch(BatchMessageHandler handler) {
560 : 4 : std::unique_lock<std::shared_mutex> lock(mutex_);
561 : 4 : message_batch_handler_ = interface::share_callback(std::move(handler));
562 [ + - + - ]: 4 : if (framer) attach_framer_callback();
563 : 4 : }
564 : : };
565 : :
566 : 29 : UdpClient::UdpClient(const config::UdpConfig& cfg) : impl_(std::make_shared<Impl>(cfg)) {}
567 : 5 : UdpClient::UdpClient(const config::UdpConfig& cfg, std::shared_ptr<boost::asio::io_context> ioc)
568 : 5 : : impl_(std::make_shared<Impl>(cfg, ioc)) {}
569 : 15 : UdpClient::UdpClient(std::shared_ptr<interface::Channel> ch) : impl_(std::make_shared<Impl>(ch)) {
570 : 15 : impl_->setup_internal_handlers();
571 : 15 : }
572 : 56 : UdpClient::~UdpClient() = default;
573 : :
574 : 0 : UdpClient::UdpClient(UdpClient&&) noexcept = default;
575 : 0 : UdpClient& UdpClient::operator=(UdpClient&&) noexcept = default;
576 : :
577 : 32 : std::future<bool> UdpClient::start() { return impl_->start(); }
578 : 25 : void UdpClient::stop() { impl_->stop(); }
579 : 13 : bool UdpClient::send(std::string_view data) { return impl_->send(data); }
580 : 2 : bool UdpClient::try_send(std::string_view data) { return impl_->try_send(data); }
581 : 5 : bool UdpClient::send_line(std::string_view line) { return impl_->send_line(line); }
582 : 1 : bool UdpClient::try_send_line(std::string_view line) { return impl_->try_send_line(line); }
583 : 2 : bool UdpClient::send_move(std::vector<uint8_t>&& data) { return impl_->send_move(std::move(data)); }
584 : 2 : bool UdpClient::try_send_move(std::vector<uint8_t>&& data) { return impl_->try_send_move(std::move(data)); }
585 : 2 : bool UdpClient::send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
586 : 2 : return impl_->send_shared(std::move(data));
587 : : }
588 : 4 : bool UdpClient::try_send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
589 : 4 : return impl_->try_send_shared(std::move(data));
590 : : }
591 : 4 : bool UdpClient::send_blocking(std::string_view data) { return impl_->send_blocking(data); }
592 : 0 : bool UdpClient::send_line_blocking(std::string_view line) { return impl_->send_line_blocking(line); }
593 : 11 : bool UdpClient::connected() const {
594 : 11 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
595 [ + + + - : 22 : return impl_->channel && impl_->channel->is_connected();
+ + ]
596 : 11 : }
597 : 8 : RuntimeStats UdpClient::stats() const { return impl_->stats(); }
598 : 0 : void UdpClient::reset_stats() { impl_->reset_stats(); }
599 : :
600 : 13 : UdpClient& UdpClient::on_data(MessageHandler h) {
601 : 13 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
602 : 13 : impl_->data_handler = interface::share_callback(std::move(h));
603 : 13 : return *this;
604 : 13 : }
605 : 3 : UdpClient& UdpClient::on_data_batch(BatchMessageHandler h) {
606 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
607 : 3 : impl_->data_batch_handler_ = interface::share_callback(std::move(h));
608 : 3 : return *this;
609 : 3 : }
610 : 6 : UdpClient& UdpClient::on_connect(ConnectionHandler h) {
611 : 6 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
612 : 6 : impl_->connect_handler = std::move(h);
613 : 6 : return *this;
614 : 6 : }
615 : 4 : UdpClient& UdpClient::on_disconnect(ConnectionHandler h) {
616 : 4 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
617 : 4 : impl_->disconnect_handler = std::move(h);
618 : 4 : return *this;
619 : 4 : }
620 : 7 : UdpClient& UdpClient::on_error(ErrorHandler h) {
621 : 7 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
622 : 7 : impl_->error_handler = std::move(h);
623 : 7 : return *this;
624 : 7 : }
625 : :
626 : 3 : UdpClient& UdpClient::on_backpressure(std::function<void(size_t)> h) {
627 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
628 : 3 : impl_->bp_handler = std::move(h);
629 : 3 : return *this;
630 : 3 : }
631 : :
632 : 7 : UdpClient& UdpClient::framer(std::unique_ptr<framer::IFramer> f) {
633 : 7 : impl_->set_framer(std::move(f));
634 : 7 : return *this;
635 : : }
636 : 3 : UdpClient& UdpClient::on_message(MessageHandler h) {
637 : 3 : impl_->on_message(std::move(h));
638 : 3 : return *this;
639 : : }
640 : 4 : UdpClient& UdpClient::on_message_batch(BatchMessageHandler h) {
641 : 4 : impl_->on_message_batch(std::move(h));
642 : 4 : return *this;
643 : : }
644 : :
645 : 4 : UdpClient& UdpClient::auto_start(bool m) {
646 : 4 : impl_->auto_start_.store(m);
647 [ + + + - : 4 : if (impl_->auto_start_.load() && !impl_->started_.load()) start();
+ + + - ]
648 : 4 : return *this;
649 : : }
650 : :
651 : 10 : UdpClient& UdpClient::backpressure_threshold(size_t threshold) {
652 : 10 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
653 : 10 : impl_->cfg.backpressure_threshold = threshold;
654 : 10 : return *this;
655 : 10 : }
656 : :
657 : 8 : UdpClient& UdpClient::backpressure_strategy(base::constants::BackpressureStrategy strategy) {
658 : 8 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
659 : 8 : impl_->cfg.backpressure_strategy = strategy;
660 [ + + ]: 8 : if (impl_->channel) {
661 [ + - - + ]: 2 : if (auto* uc = dynamic_cast<transport::UdpChannel*>(impl_->channel.get())) {
662 : 0 : uc->set_backpressure_strategy(strategy);
663 : : }
664 : : }
665 : 8 : return *this;
666 : 8 : }
667 : :
668 : 3 : size_t UdpClient::backpressure_threshold() const {
669 : 3 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
670 : 6 : return impl_->cfg.backpressure_threshold;
671 : 3 : }
672 : :
673 : 3 : base::constants::BackpressureStrategy UdpClient::backpressure_strategy() const {
674 : 3 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
675 : 6 : return impl_->cfg.backpressure_strategy;
676 : 3 : }
677 : :
678 : 0 : UdpClient& UdpClient::send_buffer_size(size_t bytes) {
679 : 0 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
680 : 0 : impl_->cfg.send_buffer_size = bytes;
681 : 0 : return *this;
682 : 0 : }
683 : :
684 : 0 : UdpClient& UdpClient::receive_buffer_size(size_t bytes) {
685 : 0 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
686 : 0 : impl_->cfg.receive_buffer_size = bytes;
687 : 0 : return *this;
688 : 0 : }
689 : :
690 : 6 : UdpClient& UdpClient::manage_external_context(bool m) {
691 : 6 : impl_->manage_external_context.store(m);
692 : 6 : return *this;
693 : : }
694 : :
695 : 3 : UdpClient& UdpClient::batch_size(size_t size) {
696 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
697 : 3 : impl_->max_batch_size_ = size;
698 : 3 : return *this;
699 : 3 : }
700 : :
701 : 3 : UdpClient& UdpClient::batch_latency(std::chrono::milliseconds latency) {
702 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
703 : 3 : impl_->max_batch_latency_ = latency;
704 : 3 : return *this;
705 : 3 : }
706 : :
707 : : } // namespace wrapper
708 : : } // namespace wirestead
|
