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