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