Branch data Line data Source code
1 : : /*
2 : : * Copyright 2025 Jinwoo Sung
3 : : *
4 : : * Licensed under the Apache License, Version 2.0 (the "License");
5 : : * you may not use this file except in compliance with the License.
6 : : * You may obtain a copy of the License at
7 : : *
8 : : * http://www.apache.org/licenses/LICENSE-2.0
9 : : *
10 : : * Unless required by applicable law or agreed to in writing, software
11 : : * distributed under the License is distributed on an "AS IS" BASIS,
12 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 : : * See the License for the specific language governing permissions and
14 : : * limitations under the License.
15 : : */
16 : :
17 : : #include "wirestead/transport/uds/uds_client.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 : : #endif
24 : :
25 : : #include <spdlog/fmt/fmt.h>
26 : :
27 : : #include <algorithm>
28 : : #include <array>
29 : : #include <atomic>
30 : : #include <boost/asio.hpp>
31 : : #include <deque>
32 : : #include <memory>
33 : : #include <mutex>
34 : : #include <stop_token>
35 : : #include <thread>
36 : :
37 : : #include "wirestead/base/constants.hpp"
38 : : #include "wirestead/concurrency/thread_safe_state.hpp"
39 : : #include "wirestead/diagnostics/error_handler.hpp"
40 : : #include "wirestead/diagnostics/error_mapping.hpp"
41 : : #include "wirestead/diagnostics/logger.hpp"
42 : : #include "wirestead/diagnostics/runtime_stats_counter.hpp"
43 : : #include "wirestead/memory/memory_pool.hpp"
44 : : #include "wirestead/transport/base/bp_state_machine.hpp"
45 : : #include "wirestead/transport/base/bp_utils.hpp"
46 : : #include "wirestead/transport/base/error_info_holder.hpp"
47 : : #include "wirestead/transport/uds/boost_uds_socket.hpp"
48 : : #include "wirestead/transport/uds/detail/reconnect_decider.hpp"
49 : :
50 : : namespace wirestead {
51 : : namespace transport {
52 : :
53 : : namespace net = boost::asio;
54 : : using uds = net::local::stream_protocol;
55 : :
56 : : using base::LinkState;
57 : : using concurrency::AtomicLinkState;
58 : : using config::UdsClientConfig;
59 : : using interface::Channel;
60 : :
61 : : struct UdsClient::Impl {
62 : : std::shared_ptr<net::io_context> owned_ioc_;
63 : : net::io_context* ioc_ = nullptr;
64 : : net::strand<net::io_context::executor_type> strand_;
65 : : std::unique_ptr<net::executor_work_guard<net::io_context::executor_type>> work_guard_;
66 : : std::jthread ioc_thread_;
67 : : std::atomic<uint64_t> current_seq_{0};
68 : : std::unique_ptr<interface::UdsSocketInterface> socket_;
69 : : // Guards the mutable subset of cfg_ (retry_interval_ms, etc.) and
70 : : // reconnect_policy_ below - see the identical rationale in
71 : : // transport/tcp_client/tcp_client.cc (#436).
72 : : mutable std::mutex cfg_mtx_;
73 : : UdsClientConfig cfg_;
74 : : // #443: UDS never actually pooled - async_write_copy always heap-allocated
75 : : // a fresh std::vector, despite a dead PooledBuffer std::visit branch
76 : : // suggesting otherwise. Give it a real per-channel pool like every other
77 : : // transport, instead of the process-wide GlobalMemoryPool singleton.
78 : : // Prefill stays 0. This literal was written while MemoryPool discarded
79 : : // initial_pool_size, so 50 allocated nothing; #575 made the parameter real
80 : : // and turned it into ~1 MiB eagerly allocated per channel at construction.
81 : : // The pool fills as buffers are released.
82 : : memory::MemoryPool pool_{0, 200};
83 : : net::steady_timer retry_timer_;
84 : : net::steady_timer connect_timer_;
85 : : bool owns_ioc_ = true;
86 : : std::atomic<bool> stop_requested_{false};
87 : : std::atomic<bool> stopping_{false};
88 : :
89 : : // Sized from cfg_.read_buffer_size in init() rather than being a fixed
90 : : // std::array, so a bulk-transfer workload can trade memory for fewer read
91 : : // completions and callback dispatches.
92 : : std::vector<uint8_t> rx_;
93 : : std::deque<BufferVariant> tx_;
94 : : std::deque<BufferVariant> pending_;
95 : : std::atomic<size_t> pending_bytes_{0};
96 : : // Buffers handed to the in-flight gather write; current_write_views_
97 : : // points into the batch, so neither is touched while a write is in flight.
98 : : std::vector<BufferVariant> current_write_batch_;
99 : : std::vector<net::const_buffer> current_write_views_;
100 : : bool writing_ = false;
101 : : std::atomic<size_t> queue_bytes_{0};
102 : : // Bytes accepted by a plain async_write_* call but not yet routed onto the
103 : : // strand - reserved via try_reserve_limit_bytes() to close the
104 : : // accept-then-drop race (jwsung91/wirestead#517). inflight_bytes_ mutations
105 : : // and the queue_bytes_/pending_bytes_ increments that promote a
106 : : // reservation both go through write_reserve_mtx_ - see bp_utils.hpp.
107 : : std::atomic<size_t> inflight_bytes_{0};
108 : : std::mutex write_reserve_mtx_;
109 : : // Atomic rather than mutex-guarded: read both from the strand and from
110 : : // arbitrary caller threads (async_try_write_* fast-fail prechecks) (#436).
111 : : std::atomic<base::constants::BackpressureStrategy> bp_strategy_{base::constants::BackpressureStrategy::Reliable};
112 : : size_t bp_high_;
113 : : size_t bp_low_;
114 : : size_t bp_limit_;
115 : : std::atomic<bool> backpressure_active_{false};
116 : : diagnostics::RuntimeStatsCounters stats_;
117 : :
118 : : // Shared snapshots: the io thread copies one out per received chunk, and a
119 : : // std::function copy allocates whenever the target outgrows its small-object
120 : : // buffer. See interface::SharedCallback.
121 : : interface::SharedCallback<OnBytes> on_bytes_;
122 : : interface::SharedCallback<OnState> on_state_;
123 : : interface::SharedCallback<OnBackpressure> on_bp_;
124 : : mutable std::mutex callback_mtx_;
125 : : std::atomic<bool> connected_{false};
126 : 39 : AtomicLinkState state_{LinkState::Idle};
127 : : int retry_attempts_ = 0;
128 : : uint32_t reconnect_attempt_count_{0};
129 : : std::optional<ReconnectPolicy> reconnect_policy_;
130 : :
131 : : ErrorInfoHolder error_info_holder_{"uds_client"};
132 : :
133 : 39 : Impl(const UdsClientConfig& cfg, net::io_context* ioc_ptr,
134 : : std::unique_ptr<interface::UdsSocketInterface> socket = nullptr)
135 [ + + ]: 39 : : owned_ioc_(ioc_ptr ? nullptr : std::make_shared<net::io_context>()),
136 [ + + ]: 39 : ioc_(ioc_ptr ? ioc_ptr : owned_ioc_.get()),
137 : 39 : strand_(net::make_strand(*ioc_)),
138 : 39 : socket_(std::move(socket)),
139 : 39 : cfg_(cfg),
140 : 39 : retry_timer_(strand_),
141 : 39 : connect_timer_(strand_),
142 : 39 : owns_ioc_(!ioc_ptr),
143 : 39 : bp_strategy_(cfg.backpressure_strategy),
144 : 234 : bp_high_(cfg.backpressure_threshold) {
145 [ + + ]: 39 : if (!socket_) {
146 : 18 : socket_ = std::make_unique<BoostUdsSocket>(uds::socket(strand_));
147 : : }
148 : 39 : init();
149 : 39 : }
150 : :
151 : 39 : void init() {
152 : 39 : connected_ = false;
153 : 39 : writing_ = false;
154 : 39 : queue_bytes_ = 0;
155 : 39 : pending_bytes_ = 0;
156 : 39 : cfg_.validate_and_clamp();
157 : 39 : rx_.resize(cfg_.read_buffer_size);
158 : 39 : recalculate_backpressure_bounds();
159 : 39 : }
160 : :
161 : : void do_connect(std::shared_ptr<UdsClient> self, uint64_t seq);
162 : : void schedule_retry(std::shared_ptr<UdsClient> self, uint64_t seq);
163 : : void start_read(std::shared_ptr<UdsClient> self, uint64_t seq);
164 : : void do_write(std::shared_ptr<UdsClient> self, uint64_t seq);
165 : : void handle_close(std::shared_ptr<UdsClient> self, uint64_t seq, const boost::system::error_code& ec = {});
166 : : void transition_to(LinkState next, const boost::system::error_code& ec = {});
167 : : void perform_stop_cleanup(uint64_t seq);
168 : : void close_socket();
169 : : void recalculate_backpressure_bounds();
170 : : void report_backpressure(std::shared_ptr<UdsClient> self, size_t queued_bytes);
171 : : void observe_queue();
172 : : // Shared decide_enqueue()/route dispatch used by both async_write_* variants (#434).
173 : : void route_enqueued_buffer(std::shared_ptr<UdsClient> self, BufferVariant&& buf, size_t added);
174 : : queue_util::BackpressureFields bp_fields();
175 : : void record_error(diagnostics::ErrorLevel lvl, diagnostics::ErrorCategory cat, std::string_view operation,
176 : : const boost::system::error_code& ec, std::string_view msg, bool retryable, uint32_t retry_count);
177 : :
178 : 39 : ~Impl() {
179 : 39 : stop_requested_ = true;
180 : 39 : stopping_ = true;
181 : :
182 : 39 : retry_timer_.cancel();
183 : 39 : connect_timer_.cancel();
184 : 39 : close_socket();
185 : :
186 [ - + ]: 39 : if (work_guard_) {
187 : 0 : work_guard_.reset();
188 : : }
189 : :
190 [ + - + + : 39 : if (ioc_ && owns_ioc_ && ioc_thread_.joinable()) {
- + - + ]
191 [ # # ]: 0 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
192 : 0 : ioc_thread_.detach();
193 : : } else {
194 : 0 : ioc_thread_.request_stop();
195 : 0 : ioc_thread_.join();
196 : : }
197 : : }
198 : 39 : }
199 : : };
200 : :
201 : 13 : std::shared_ptr<UdsClient> UdsClient::create(const UdsClientConfig& cfg) {
202 : 13 : return std::shared_ptr<UdsClient>(new UdsClient(cfg));
203 : : }
204 : :
205 : 5 : std::shared_ptr<UdsClient> UdsClient::create(const UdsClientConfig& cfg, boost::asio::io_context& ioc) {
206 : 5 : return std::shared_ptr<UdsClient>(new UdsClient(cfg, ioc));
207 : : }
208 : :
209 : 21 : std::shared_ptr<UdsClient> UdsClient::create(const UdsClientConfig& cfg,
210 : : std::unique_ptr<interface::UdsSocketInterface> socket,
211 : : boost::asio::io_context& ioc) {
212 : 21 : return std::shared_ptr<UdsClient>(new UdsClient(cfg, std::move(socket), ioc));
213 : : }
214 : :
215 : 13 : UdsClient::UdsClient(const UdsClientConfig& cfg) : impl_(std::make_unique<Impl>(cfg, nullptr)) {}
216 : 5 : UdsClient::UdsClient(const UdsClientConfig& cfg, boost::asio::io_context& ioc)
217 : 5 : : impl_(std::make_unique<Impl>(cfg, &ioc)) {}
218 : :
219 : 21 : UdsClient::UdsClient(const UdsClientConfig& cfg, std::unique_ptr<interface::UdsSocketInterface> socket,
220 : 21 : boost::asio::io_context& ioc)
221 : 21 : : impl_(std::make_unique<Impl>(cfg, &ioc, std::move(socket))) {}
222 : :
223 : 79 : UdsClient::~UdsClient() {
224 : : // #446: null after being moved-from - the move ctor/assignment are
225 : : // defaulted, and destroying a moved-from instance must not dereference
226 : : // a null impl_ (matches TcpServer/Serial/UdpChannel/UdsServer's
227 : : // destructors, which already guard this way).
228 [ + + ]: 40 : if (!impl_) return;
229 : 39 : stop();
230 : :
231 [ + + - + : 39 : if (impl_->owns_ioc_ && impl_->ioc_thread_.joinable()) {
- + ]
232 [ # # ]: 0 : if (std::this_thread::get_id() != impl_->ioc_thread_.get_id()) {
233 : 0 : impl_->ioc_thread_.join();
234 : : } else {
235 : 0 : impl_->ioc_thread_.detach();
236 : : }
237 : : }
238 : 81 : }
239 : :
240 : 1 : UdsClient::UdsClient(UdsClient&&) noexcept = default;
241 : 0 : UdsClient& UdsClient::operator=(UdsClient&&) noexcept = default;
242 : :
243 : 34 : void UdsClient::start() {
244 : 34 : auto current_state = impl_->state_.get();
245 [ + + - + ]: 34 : if (current_state == LinkState::Connecting || current_state == LinkState::Connected) {
246 : 1 : return;
247 : : }
248 : :
249 : 33 : impl_->recalculate_backpressure_bounds();
250 : 33 : impl_->stop_requested_ = false;
251 : 33 : impl_->stopping_ = false;
252 : 33 : impl_->current_seq_++;
253 : 33 : uint64_t seq = impl_->current_seq_.load();
254 : :
255 [ + + + - : 33 : if (impl_->owns_ioc_ && !impl_->ioc_thread_.joinable()) {
+ + ]
256 [ - + ]: 12 : if (impl_->ioc_->stopped()) {
257 : 0 : impl_->ioc_->restart();
258 : : }
259 : 12 : impl_->work_guard_ =
260 : 24 : std::make_unique<net::executor_work_guard<net::io_context::executor_type>>(net::make_work_guard(*impl_->ioc_));
261 : 24 : impl_->ioc_thread_ = std::jthread([ioc = impl_->owned_ioc_](std::stop_token st) {
262 : 12 : wirestead::concurrency::run_io_thread_init();
263 : : try {
264 : 12 : std::stop_callback cb(st, [ioc] { ioc->stop(); });
265 : 12 : ioc->run();
266 : 12 : } catch (...) {
267 : 0 : }
268 : 24 : });
269 : : }
270 : :
271 : 33 : net::post(impl_->strand_, [self = shared_from_this(), seq]() {
272 : 33 : self->impl_->transition_to(LinkState::Connecting);
273 : 33 : self->impl_->do_connect(self, seq);
274 : 33 : });
275 : : }
276 : :
277 : 77 : void UdsClient::stop() {
278 : 77 : bool already_stopping = impl_->stopping_.exchange(true);
279 [ + + ]: 77 : if (already_stopping) return;
280 : :
281 : 39 : impl_->stop_requested_ = true;
282 : 39 : impl_->connected_ = false;
283 : 39 : const auto seq = impl_->current_seq_.fetch_add(1) + 1;
284 : :
285 : : // Release work guard and allow io_context to run out of work
286 [ + - ]: 39 : if (impl_->ioc_) {
287 [ + + ]: 39 : if (auto self = weak_from_this().lock()) {
288 : 71 : net::post(impl_->strand_, [self, seq]() { self->impl_->perform_stop_cleanup(seq); });
289 : : } else {
290 : 1 : impl_->perform_stop_cleanup(seq);
291 : 39 : }
292 : : } else {
293 : 0 : impl_->perform_stop_cleanup(seq);
294 : : }
295 : :
296 [ + + + + : 39 : if (impl_->owns_ioc_ && impl_->ioc_thread_.joinable()) {
+ + ]
297 [ - + ]: 12 : if (std::this_thread::get_id() == impl_->ioc_thread_.get_id()) {
298 : 0 : impl_->ioc_thread_.detach();
299 : : } else {
300 : 12 : impl_->ioc_thread_.join();
301 : : }
302 : : }
303 : :
304 : : // Transition to Idle state (lock-free or safe call)
305 : 39 : impl_->state_.set(LinkState::Idle);
306 : : }
307 : :
308 : 51 : bool UdsClient::is_connected() const { return impl_->connected_.load(); }
309 : 17 : bool UdsClient::is_backpressure_active() const { return impl_->backpressure_active_.load(); }
310 : 6 : wrapper::RuntimeStats UdsClient::stats() const {
311 : 24 : return impl_->stats_.snapshot(impl_->queue_bytes_.load(std::memory_order_relaxed),
312 : 6 : impl_->pending_bytes_.load(std::memory_order_relaxed),
313 : 12 : impl_->backpressure_active_.load(std::memory_order_relaxed));
314 : : }
315 : 0 : void UdsClient::reset_stats() {
316 : 0 : impl_->stats_.reset(impl_->queue_bytes_.load(std::memory_order_relaxed) +
317 : 0 : impl_->pending_bytes_.load(std::memory_order_relaxed));
318 : 0 : }
319 : :
320 : 16 : boost::asio::any_io_executor UdsClient::get_executor() { return impl_->strand_; }
321 : :
322 : 17 : bool UdsClient::async_write_copy(memory::ConstByteSpan data) {
323 : 17 : size_t size = data.size();
324 [ + - + - : 17 : if (impl_->cfg_.enable_memory_pool && size > 0 && size <= 65536) {
+ - + - ]
325 [ + + - + : 17 : if (!impl_->connected_.load() || impl_->stop_requested_.load()) {
+ + ]
326 : 1 : impl_->stats_.record_failed_send();
327 : 17 : return false;
328 : : }
329 : 16 : memory::PooledBuffer pooled(size, impl_->pool_);
330 [ + - + - ]: 16 : if (pooled.valid()) {
331 : 16 : base::safe_memory::safe_memcpy(pooled.data(), data.data(), size);
332 [ + - - + ]: 16 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
333 : 16 : impl_->inflight_bytes_, size, impl_->bp_limit_)) {
334 : 0 : impl_->stats_.record_failed_send();
335 : 0 : return false;
336 : : }
337 : 16 : impl_->stats_.record_accepted(size);
338 : 16 : net::post(impl_->strand_, [this, self = shared_from_this(), buf = std::move(pooled)]() mutable {
339 : 16 : size_t added = buf.size();
340 : 16 : impl_->route_enqueued_buffer(self, BufferVariant{std::move(buf)}, added);
341 : 16 : });
342 : 16 : return true;
343 : : }
344 : 16 : }
345 : :
346 : 0 : std::vector<uint8_t> vec(data.begin(), data.end());
347 : 0 : return async_write_move(std::move(vec));
348 : 0 : }
349 : :
350 : 7 : bool UdsClient::async_write_move(std::vector<uint8_t>&& data) {
351 [ + + - + : 7 : if (!impl_->connected_.load() || impl_->stop_requested_.load()) {
+ + ]
352 : 1 : impl_->stats_.record_failed_send();
353 : 1 : return false;
354 : : }
355 [ - + ]: 6 : if (data.empty()) {
356 : 0 : impl_->stats_.record_failed_send();
357 : 0 : return false;
358 : : }
359 : 6 : const auto added = data.size();
360 [ + + ]: 6 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
361 : 6 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
362 : 1 : impl_->stats_.record_failed_send();
363 : 1 : return false;
364 : : }
365 : 5 : impl_->stats_.record_accepted(added);
366 : 5 : net::post(impl_->strand_, [this, self = shared_from_this(), data = std::move(data), added]() mutable {
367 : 5 : impl_->route_enqueued_buffer(self, BufferVariant{std::move(data)}, added);
368 : 5 : });
369 : 5 : return true;
370 : : }
371 : :
372 : 3 : bool UdsClient::async_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
373 [ + + + - : 3 : if (!impl_->connected_.load() || impl_->stop_requested_.load() || !data || data->empty()) {
+ - - + +
+ ]
374 : 2 : impl_->stats_.record_failed_send();
375 : 2 : return false;
376 : : }
377 : 1 : const auto added = data->size();
378 [ - + ]: 1 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
379 : 1 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
380 : 0 : impl_->stats_.record_failed_send();
381 : 0 : return false;
382 : : }
383 : 1 : impl_->stats_.record_accepted(added);
384 : 1 : net::post(impl_->strand_, [this, self = shared_from_this(), data = std::move(data), added]() mutable {
385 : 1 : impl_->route_enqueued_buffer(self, BufferVariant{std::move(data)}, added);
386 : 1 : });
387 : 1 : return true;
388 : : }
389 : :
390 : 1 : bool UdsClient::async_try_write_copy(memory::ConstByteSpan data) {
391 [ + - - + : 1 : if (data.empty() || data.size() > base::constants::MAX_BUFFER_SIZE) {
- + ]
392 : 0 : impl_->stats_.record_failed_send();
393 : 0 : return false;
394 : : }
395 : 3 : return async_try_write_move(std::vector<uint8_t>(data.begin(), data.end()));
396 : : }
397 : :
398 : 4 : bool UdsClient::async_try_write_move(std::vector<uint8_t>&& data) {
399 [ + - - + : 4 : if (!impl_->connected_.load() || impl_->stop_requested_.load()) {
- + ]
400 : 0 : impl_->stats_.record_failed_send();
401 : 0 : return false;
402 : : }
403 : 4 : const auto added = data.size();
404 [ + - - + ]: 4 : if (added == 0 || added > base::constants::MAX_BUFFER_SIZE) {
405 : 0 : impl_->stats_.record_failed_send();
406 : 0 : return false;
407 : : }
408 : 3 : const auto reject_for_pressure = [this, added]() {
409 [ + + ]: 3 : if (impl_->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
410 : 1 : impl_->stats_.record_dropped(1, added);
411 : : } else {
412 : 2 : impl_->stats_.record_failed_send();
413 : : }
414 : 7 : };
415 [ + + + - : 5 : if (impl_->backpressure_active_.load() || impl_->queue_bytes_ + added > impl_->bp_high_ ||
+ + ]
416 [ - + ]: 1 : impl_->queue_bytes_ + impl_->pending_bytes_ + added > impl_->bp_limit_) {
417 : 3 : reject_for_pressure();
418 : 3 : return false;
419 : : }
420 [ - + ]: 1 : if (!queue_util::try_reserve_write_bytes(impl_->queue_bytes_, impl_->pending_bytes_, impl_->backpressure_active_,
421 : 1 : added, impl_->bp_high_, impl_->bp_limit_)) {
422 : 0 : reject_for_pressure();
423 : 0 : return false;
424 : : }
425 : 1 : impl_->stats_.record_accepted(added);
426 : :
427 : 1 : net::post(impl_->strand_, [this, self = shared_from_this(), data = std::move(data), added]() mutable {
428 [ + - - + : 1 : if (!impl_->connected_.load() || impl_->stop_requested_.load()) {
- + ]
429 : 0 : queue_util::release_reserved_write_bytes(impl_->queue_bytes_, added);
430 : 0 : impl_->stats_.record_failed_send();
431 : 0 : return;
432 : : }
433 : :
434 : 1 : impl_->tx_.emplace_back(std::move(data));
435 : 1 : impl_->observe_queue();
436 : 1 : impl_->report_backpressure(self, impl_->queue_bytes_);
437 [ - + - - ]: 1 : if (!impl_->writing_) impl_->do_write(self, impl_->current_seq_.load());
438 : : });
439 : 1 : return true;
440 : : }
441 : :
442 : 1 : bool UdsClient::async_try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
443 [ + - + - : 1 : if (!impl_->connected_.load() || impl_->stop_requested_.load() || !data || data->empty()) {
+ - - + -
+ ]
444 : 0 : impl_->stats_.record_failed_send();
445 : 0 : return false;
446 : : }
447 : 1 : const auto added = data->size();
448 [ - + ]: 1 : if (added > base::constants::MAX_BUFFER_SIZE) {
449 : 0 : impl_->stats_.record_failed_send();
450 : 0 : return false;
451 : : }
452 : 1 : const auto reject_for_pressure = [this, added]() {
453 [ - + ]: 1 : if (impl_->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
454 : 0 : impl_->stats_.record_dropped(1, added);
455 : : } else {
456 : 1 : impl_->stats_.record_failed_send();
457 : : }
458 : 2 : };
459 [ - + - - : 1 : if (impl_->backpressure_active_.load() || impl_->queue_bytes_ + added > impl_->bp_high_ ||
+ - ]
460 [ # # ]: 0 : impl_->queue_bytes_ + impl_->pending_bytes_ + added > impl_->bp_limit_) {
461 : 1 : reject_for_pressure();
462 : 1 : return false;
463 : : }
464 [ # # ]: 0 : if (!queue_util::try_reserve_write_bytes(impl_->queue_bytes_, impl_->pending_bytes_, impl_->backpressure_active_,
465 : 0 : added, impl_->bp_high_, impl_->bp_limit_)) {
466 : 0 : reject_for_pressure();
467 : 0 : return false;
468 : : }
469 : 0 : impl_->stats_.record_accepted(added);
470 : :
471 : 0 : net::post(impl_->strand_, [this, self = shared_from_this(), data = std::move(data), added]() mutable {
472 [ # # # # : 0 : if (!impl_->connected_.load() || impl_->stop_requested_.load()) {
# # ]
473 : 0 : queue_util::release_reserved_write_bytes(impl_->queue_bytes_, added);
474 : 0 : impl_->stats_.record_failed_send();
475 : 0 : return;
476 : : }
477 : :
478 : 0 : impl_->tx_.emplace_back(std::move(data));
479 : 0 : impl_->observe_queue();
480 : 0 : impl_->report_backpressure(self, impl_->queue_bytes_);
481 [ # # # # ]: 0 : if (!impl_->writing_) impl_->do_write(self, impl_->current_seq_.load());
482 : : });
483 : 0 : return true;
484 : : }
485 : :
486 : 33 : void UdsClient::on_bytes(OnBytes cb) {
487 : 33 : auto shared = interface::share_callback(std::move(cb));
488 : 33 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
489 : 33 : impl_->on_bytes_ = std::move(shared);
490 : 33 : }
491 : :
492 : 38 : void UdsClient::on_state(OnState cb) {
493 : 38 : auto shared = interface::share_callback(std::move(cb));
494 : 38 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
495 : 38 : impl_->on_state_ = std::move(shared);
496 : 38 : }
497 : :
498 : 37 : void UdsClient::on_backpressure(OnBackpressure cb) {
499 : 37 : auto shared = interface::share_callback(std::move(cb));
500 : 37 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
501 : 37 : impl_->on_bp_ = std::move(shared);
502 : 37 : }
503 : :
504 : 0 : void UdsClient::set_backpressure_strategy(base::constants::BackpressureStrategy strategy) {
505 : 0 : impl_->bp_strategy_.store(strategy, std::memory_order_relaxed);
506 : 0 : }
507 : :
508 : 0 : void UdsClient::set_retry_interval(unsigned interval_ms) {
509 : 0 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
510 : 0 : impl_->cfg_.retry_interval_ms = interval_ms;
511 : 0 : impl_->cfg_.validate_and_clamp();
512 : 0 : }
513 : :
514 : 0 : void UdsClient::set_reconnect_policy(ReconnectPolicy policy) {
515 : 0 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
516 [ # # ]: 0 : if (policy) {
517 : 0 : impl_->reconnect_policy_ = std::move(policy);
518 : : } else {
519 : 0 : impl_->reconnect_policy_ = std::nullopt;
520 : : }
521 : 0 : }
522 : :
523 : 7 : std::optional<diagnostics::ErrorInfo> UdsClient::last_error_info() const {
524 : 7 : return impl_->error_info_holder_.last_error_info();
525 : : }
526 : :
527 : 34 : void UdsClient::Impl::do_connect(std::shared_ptr<UdsClient> self, uint64_t seq) {
528 [ + - + - : 68 : if (stop_requested_.load() || stopping_.load() || seq != current_seq_.load()) {
- + - + ]
529 : 1 : return;
530 : : }
531 : :
532 [ + - + + ]: 34 : if (!cfg_.is_valid()) {
533 : 3 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONFIGURATION, "connect",
534 : 1 : make_error_code(boost::system::errc::invalid_argument), "Invalid UDS socket path", false,
535 : 1 : self->impl_->reconnect_attempt_count_);
536 : 1 : self->impl_->transition_to(LinkState::Error);
537 : 1 : return;
538 : : }
539 : :
540 : 33 : std::shared_ptr<uds::endpoint> endpoint;
541 : : try {
542 : 33 : endpoint = std::make_shared<uds::endpoint>(cfg_.socket_path);
543 : 0 : } catch (const std::exception& e) {
544 : 0 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONFIGURATION, "connect",
545 : 0 : make_error_code(boost::system::errc::filename_too_long),
546 : 0 : "Invalid UDS endpoint: " + std::string(e.what()), false,
547 : 0 : self->impl_->reconnect_attempt_count_);
548 : 0 : self->impl_->transition_to(LinkState::Error);
549 : 0 : return;
550 : 0 : }
551 : :
552 : : unsigned connection_timeout_ms;
553 : : {
554 : 33 : std::lock_guard<std::mutex> lock(cfg_mtx_);
555 : 33 : connection_timeout_ms = cfg_.connection_timeout_ms;
556 : 33 : }
557 : 33 : connect_timer_.expires_after(std::chrono::milliseconds(connection_timeout_ms));
558 : 33 : connect_timer_.async_wait(net::bind_executor(strand_, [self, seq](const boost::system::error_code& ec) {
559 [ + + - + : 34 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) return;
+ + ]
560 [ + - ]: 1 : if (!ec) {
561 : 2 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "connect",
562 : : boost::asio::error::timed_out, "Connection timed out", true,
563 : 1 : self->impl_->reconnect_attempt_count_);
564 : 1 : self->impl_->handle_close(self, seq, boost::asio::error::timed_out);
565 : : }
566 : : }));
567 : :
568 : 33 : socket_->async_connect(*endpoint, net::bind_executor(strand_, [self, seq,
569 : : endpoint](const boost::system::error_code& ec) {
570 : 31 : self->impl_->connect_timer_.cancel();
571 [ + - - + : 62 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) return;
- + ]
572 [ + - - + : 31 : if (self->impl_->stop_requested_.load() || self->impl_->stopping_.load()) {
- + ]
573 : 0 : self->impl_->close_socket();
574 : 0 : return;
575 : : }
576 [ + + ]: 31 : if (ec) {
577 : 10 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "connect", ec,
578 : 10 : "Connect failed: " + ec.message(), diagnostics::is_retryable_uds_connect_error(ec),
579 : 5 : self->impl_->reconnect_attempt_count_);
580 : 5 : self->impl_->schedule_retry(self, seq);
581 : 5 : return;
582 : : }
583 : :
584 : 26 : self->impl_->connected_ = true;
585 : 26 : self->impl_->reconnect_attempt_count_ = 0;
586 : 26 : self->impl_->retry_attempts_ = 0;
587 : 26 : self->impl_->transition_to(LinkState::Connected);
588 : 26 : self->impl_->start_read(self, seq);
589 : 26 : self->impl_->writing_ = false; // Force reset
590 : 26 : self->impl_->do_write(self, seq);
591 : : }));
592 : 33 : }
593 : :
594 : 8 : void UdsClient::Impl::schedule_retry(std::shared_ptr<UdsClient> self, uint64_t seq) {
595 : 8 : transition_to(LinkState::Error);
596 : :
597 : : // Snapshot once rather than locking repeatedly - cfg_/reconnect_policy_
598 : : // can change concurrently via set_retry_interval() etc. from any user
599 : : // thread while this runs on the strand (#436).
600 : 8 : UdsClientConfig cfg_snapshot;
601 : 8 : std::optional<ReconnectPolicy> reconnect_policy_snapshot;
602 : : {
603 : 8 : std::lock_guard<std::mutex> lock(cfg_mtx_);
604 : 8 : cfg_snapshot = cfg_;
605 : 8 : reconnect_policy_snapshot = reconnect_policy_;
606 : 8 : }
607 : :
608 : : diagnostics::ErrorInfo dummy_err(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "uds_client",
609 : 8 : "connect", "Retry pending", boost::system::error_code(), true);
610 : : auto decision =
611 : 8 : detail::decide_reconnect_uds(cfg_snapshot, dummy_err, reconnect_attempt_count_, reconnect_policy_snapshot);
612 : :
613 [ + + + + : 8 : if (!decision.should_retry || stop_requested_.load() || stopping_.load()) {
- + + + ]
614 : 2 : transition_to(LinkState::Idle);
615 : 2 : return;
616 : : }
617 : :
618 : 6 : reconnect_attempt_count_++;
619 : 6 : retry_timer_.expires_after(decision.delay.value_or(std::chrono::milliseconds(cfg_snapshot.retry_interval_ms)));
620 : 6 : retry_timer_.async_wait(net::bind_executor(strand_, [self, seq](const boost::system::error_code& ec) {
621 [ + + - + : 7 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) return;
+ + ]
622 : 1 : self->impl_->do_connect(self, seq);
623 : : }));
624 : 12 : }
625 : :
626 : 38 : void UdsClient::Impl::start_read(std::shared_ptr<UdsClient> self, uint64_t seq) {
627 [ + - + - : 76 : if (stop_requested_.load() || stopping_.load() || seq != current_seq_.load()) {
- + - + ]
628 : 0 : return;
629 : : }
630 : :
631 : 76 : socket_->async_read_some(net::buffer(rx_.data(), rx_.size()),
632 : 76 : net::bind_executor(strand_, [self, seq](const boost::system::error_code& ec, size_t bytes) {
633 [ + + - + : 43 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) return;
+ + ]
634 [ + - - + : 14 : if (self->impl_->stop_requested_.load() || self->impl_->stopping_.load()) return;
- + ]
635 [ + + ]: 14 : if (ec) {
636 : 2 : self->impl_->handle_close(self, seq, ec);
637 : 2 : return;
638 : : }
639 : :
640 : 12 : interface::SharedCallback<OnBytes> cb;
641 : : {
642 : 12 : std::lock_guard<std::mutex> lock(self->impl_->callback_mtx_);
643 : 12 : cb = self->impl_->on_bytes_;
644 : 12 : }
645 [ + - + - ]: 12 : if (bytes > 0) self->impl_->stats_.record_received(bytes);
646 [ + - + - ]: 12 : if (cb) (*cb)(memory::ConstByteSpan(self->impl_->rx_.data(), bytes));
647 : 12 : self->impl_->start_read(self, seq);
648 : 12 : }));
649 : : }
650 : :
651 : 46 : void UdsClient::Impl::do_write(std::shared_ptr<UdsClient> self, uint64_t seq) {
652 [ + - + - : 92 : if (stop_requested_.load() || stopping_.load() || seq != current_seq_.load()) {
- + - + ]
653 : 0 : tx_.clear();
654 : 0 : pending_.clear();
655 : 0 : queue_bytes_ = 0;
656 : 0 : pending_bytes_ = 0;
657 : 0 : current_write_batch_.clear();
658 : 0 : writing_ = false;
659 : 0 : return;
660 : : }
661 : :
662 [ + + - + : 46 : if (tx_.empty() || writing_) return;
+ + ]
663 : 19 : writing_ = true;
664 : : // Drain several queued buffers into one scatter-gather write rather than one
665 : : // send syscall per message. `writing_` keeps do_write() from re-entering, so
666 : : // the batch and its views stay put for the whole operation.
667 : 19 : const size_t bytes_to_write = queue_util::take_gather_batch(tx_, current_write_batch_, current_write_views_);
668 : :
669 : 57 : socket_->async_write(
670 : 19 : current_write_views_,
671 : 38 : net::bind_executor(strand_, [self, seq, bytes_to_write](const boost::system::error_code& ec, size_t written) {
672 [ + + - + : 33 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) return;
+ + ]
673 [ + - - + : 15 : if (self->impl_->stop_requested_.load() || self->impl_->stopping_.load()) {
- + ]
674 : 0 : self->impl_->current_write_batch_.clear();
675 : 0 : self->impl_->writing_ = false;
676 : 0 : return;
677 : : }
678 : 15 : self->impl_->writing_ = false;
679 : 15 : self->impl_->current_write_batch_.clear();
680 : 15 : self->impl_->queue_bytes_ =
681 [ + - ]: 15 : (self->impl_->queue_bytes_ >= bytes_to_write) ? (self->impl_->queue_bytes_ - bytes_to_write) : 0;
682 : 15 : self->impl_->report_backpressure(self, self->impl_->queue_bytes_);
683 : :
684 [ - + ]: 15 : if (ec) {
685 : 0 : self->impl_->handle_close(self, seq, ec);
686 : 0 : return;
687 : : }
688 : 15 : self->impl_->stats_.record_sent(written);
689 [ + + + - ]: 15 : if (!self->impl_->tx_.empty()) self->impl_->do_write(self, seq);
690 : : }));
691 : : }
692 : :
693 : 3 : void UdsClient::Impl::handle_close(std::shared_ptr<UdsClient> self, uint64_t seq, const boost::system::error_code&) {
694 : 3 : connected_ = false;
695 : 3 : close_socket();
696 : 3 : retry_timer_.cancel();
697 : 3 : connect_timer_.cancel();
698 : :
699 [ - + ]: 3 : if (stop_requested_) {
700 : 0 : transition_to(LinkState::Idle);
701 : : } else {
702 : 3 : schedule_retry(self, seq);
703 : : }
704 : 3 : }
705 : :
706 : 70 : void UdsClient::Impl::transition_to(LinkState next, const boost::system::error_code&) {
707 : 70 : state_.set(next);
708 : 70 : interface::SharedCallback<OnState> cb;
709 : : {
710 : 70 : std::lock_guard<std::mutex> lock(callback_mtx_);
711 : 70 : cb = on_state_;
712 : 70 : }
713 [ + + + - ]: 70 : if (cb) (*cb)(next);
714 : 70 : }
715 : :
716 : 34 : void UdsClient::Impl::perform_stop_cleanup(uint64_t seq) {
717 [ - + ]: 68 : if (seq != current_seq_.load()) {
718 : 0 : return;
719 : : }
720 : :
721 : 34 : retry_timer_.cancel();
722 : 34 : connect_timer_.cancel();
723 : 34 : close_socket();
724 : 34 : tx_.clear();
725 : 34 : queue_bytes_ = 0;
726 : 34 : pending_.clear();
727 : 34 : pending_bytes_ = 0;
728 : 34 : current_write_batch_.clear();
729 : 34 : writing_ = false;
730 : 34 : connected_.store(false);
731 : 34 : backpressure_active_.store(false);
732 : :
733 [ + + + + : 34 : if (owns_ioc_ && work_guard_) {
+ + ]
734 : 12 : work_guard_.reset();
735 : : }
736 : :
737 : 34 : state_.set(LinkState::Idle);
738 : : }
739 : :
740 : 76 : void UdsClient::Impl::close_socket() {
741 : 76 : boost::system::error_code ec;
742 : 76 : socket_->close(ec);
743 : 76 : }
744 : :
745 : 72 : void UdsClient::Impl::recalculate_backpressure_bounds() {
746 : 72 : bp_high_ = cfg_.backpressure_threshold;
747 [ + - ]: 72 : bp_low_ = bp_high_ > 1 ? bp_high_ / 2 : bp_high_;
748 [ - + ]: 72 : if (bp_low_ == 0) bp_low_ = 1;
749 : 72 : bp_limit_ = std::min(std::max(bp_high_ * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
750 : : base::constants::MAX_BUFFER_SIZE);
751 [ - + ]: 72 : if (bp_limit_ < bp_high_) bp_limit_ = bp_high_;
752 : 72 : backpressure_active_ = false;
753 : 72 : }
754 : :
755 : 60 : queue_util::BackpressureFields UdsClient::Impl::bp_fields() {
756 : 60 : return queue_util::BackpressureFields{queue_bytes_,
757 : 60 : pending_bytes_,
758 : 60 : backpressure_active_,
759 : 60 : bp_high_,
760 : 60 : bp_low_,
761 : 60 : bp_limit_,
762 : 60 : bp_strategy_.load(std::memory_order_relaxed)};
763 : : }
764 : :
765 : 22 : void UdsClient::Impl::route_enqueued_buffer(std::shared_ptr<UdsClient> self, BufferVariant&& buf, size_t added) {
766 : 22 : auto f = bp_fields();
767 : 22 : queue_util::DropAccounting dropped;
768 : 22 : auto decision = queue_util::decide_enqueue(f, added, tx_, dropped);
769 [ + + + - ]: 22 : if (dropped.any()) stats_.record_dropped(dropped.messages, dropped.bytes);
770 : :
771 [ - + ]: 22 : if (decision == queue_util::EnqueueDecision::Rejected) {
772 : 0 : WIRESTEAD_LOG_ERROR("uds_client", "write", fmt::format("Queue limit exceeded ({} bytes)", queue_bytes_ + added));
773 : : // #448: record as dropped so it's reflected in RuntimeStats instead of
774 : : // silently vanishing after being counted as accepted.
775 : 0 : stats_.record_dropped(1, added);
776 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
777 : 0 : report_backpressure(self, queue_bytes_ + added);
778 : 0 : return;
779 : : }
780 [ - + ]: 22 : if (decision == queue_util::EnqueueDecision::Pending) {
781 : 0 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, pending_bytes_, inflight_bytes_, added);
782 : 0 : pending_.emplace_back(std::move(buf));
783 : 0 : observe_queue();
784 : 0 : return;
785 : : }
786 : 22 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, queue_bytes_, inflight_bytes_, added);
787 : 22 : tx_.emplace_back(std::move(buf));
788 : 22 : observe_queue();
789 : 22 : report_backpressure(self, queue_bytes_);
790 [ + + + - ]: 40 : if (!writing_) do_write(self, current_seq_.load());
791 : : }
792 : :
793 : 62 : void UdsClient::Impl::observe_queue() {
794 : 186 : stats_.observe_queue(queue_bytes_.load(std::memory_order_relaxed) + pending_bytes_.load(std::memory_order_relaxed));
795 : 62 : }
796 : :
797 : 38 : void UdsClient::Impl::report_backpressure(std::shared_ptr<UdsClient> self, size_t queued_bytes) {
798 [ + - - + : 38 : if (stop_requested_.load() || stopping_.load()) return;
- + ]
799 : 38 : observe_queue();
800 : :
801 : 38 : interface::SharedCallback<OnBackpressure> on_bp;
802 : : {
803 : 38 : std::lock_guard<std::mutex> lock(callback_mtx_);
804 : 38 : on_bp = on_bp_;
805 : 38 : }
806 : 38 : static const OnBackpressure kNoCallback;
807 : :
808 : 38 : auto f = bp_fields();
809 [ + + + - ]: 70 : queue_util::report_backpressure(
810 : 32 : f, queued_bytes, on_bp ? *on_bp : kNoCallback, stats_,
811 : 0 : [&]() -> size_t {
812 : 1 : const size_t moved = pending_bytes_.exchange(0);
813 [ - + ]: 1 : while (!pending_.empty()) {
814 : 0 : tx_.emplace_back(std::move(pending_.front()));
815 : 0 : pending_.pop_front();
816 : : }
817 : 1 : return moved;
818 : : },
819 : 38 : [&]() {
820 : 1 : observe_queue();
821 [ + - + - ]: 2 : if (!writing_) do_write(self, current_seq_.load());
822 : 1 : });
823 : 38 : }
824 : :
825 : 7 : void UdsClient::Impl::record_error(diagnostics::ErrorLevel lvl, diagnostics::ErrorCategory cat,
826 : : std::string_view operation, const boost::system::error_code& ec,
827 : : std::string_view msg, bool retryable, uint32_t retry_count) {
828 : 7 : error_info_holder_.record_error(lvl, cat, operation, ec, msg, retryable, retry_count);
829 : 7 : }
830 : :
831 : : } // namespace transport
832 : : } // namespace wirestead
|
