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 : : #include "wirestead/config/validation.hpp"
21 : : #include "wirestead/diagnostics/callback.hpp"
22 : : #include "wirestead/diagnostics/send_accounting.hpp"
23 : :
24 : : #if defined(__GNUC__) || defined(__clang__)
25 : : #pragma GCC diagnostic ignored "-Wsign-conversion"
26 : : #endif
27 : :
28 : : #include <spdlog/fmt/fmt.h>
29 : :
30 : : #include <algorithm>
31 : : #include <array>
32 : : #include <atomic>
33 : : #include <boost/asio.hpp>
34 : : #include <condition_variable>
35 : : #include <deque>
36 : : #include <memory>
37 : : #include <mutex>
38 : : #include <stop_token>
39 : : #include <thread>
40 : :
41 : : #include "wirestead/base/constants.hpp"
42 : : #include "wirestead/concurrency/thread_safe_state.hpp"
43 : : #include "wirestead/diagnostics/error_handler.hpp"
44 : : #include "wirestead/diagnostics/error_mapping.hpp"
45 : : #include "wirestead/diagnostics/logger.hpp"
46 : : #include "wirestead/diagnostics/runtime_stats_counter.hpp"
47 : : #include "wirestead/memory/memory_pool.hpp"
48 : : #include "wirestead/transport/base/bp_state_machine.hpp"
49 : : #include "wirestead/transport/base/bp_utils.hpp"
50 : : #include "wirestead/transport/base/error_info_holder.hpp"
51 : : #include "wirestead/transport/base/stop_test_hook.hpp"
52 : : #include "wirestead/transport/uds/boost_uds_socket.hpp"
53 : : #include "wirestead/transport/uds/detail/reconnect_decider.hpp"
54 : : #include "wirestead/transport/uds/detail/write_wait.hpp"
55 : :
56 : : namespace wirestead {
57 : : namespace transport {
58 : :
59 : : namespace net = boost::asio;
60 : : using uds = net::local::stream_protocol;
61 : :
62 : : using base::LinkState;
63 : : using concurrency::AtomicLinkState;
64 : : using config::UdsClientConfig;
65 : : using interface::Channel;
66 : :
67 : : struct UdsClient::Impl {
68 : : std::shared_ptr<net::io_context> owned_ioc_;
69 : : net::io_context* ioc_ = nullptr;
70 : : net::strand<net::io_context::executor_type> strand_;
71 : : std::unique_ptr<net::executor_work_guard<net::io_context::executor_type>> work_guard_;
72 : : std::jthread ioc_thread_;
73 : : std::atomic<uint64_t> current_seq_{0};
74 : : std::atomic<uint64_t> connection_seq_{0};
75 : : std::shared_ptr<detail::UdsWriteWait> write_wait_;
76 : : std::unique_ptr<interface::UdsSocketInterface> socket_;
77 : : // Guards the mutable subset of cfg_ (retry_interval_ms, etc.) and
78 : : // reconnect_policy_ below - see the identical rationale in
79 : : // transport/tcp_client/tcp_client.cc (#436).
80 : : mutable std::mutex cfg_mtx_;
81 : : UdsClientConfig cfg_;
82 : : // #443: UDS never actually pooled - async_write_copy always heap-allocated
83 : : // a fresh std::vector, despite a dead PooledBuffer std::visit branch
84 : : // suggesting otherwise. Give it a real per-channel pool like every other
85 : : // transport, instead of the process-wide GlobalMemoryPool singleton.
86 : : // Prefill stays 0. This literal was written while MemoryPool discarded
87 : : // initial_pool_size, so 50 allocated nothing; #575 made the parameter real
88 : : // and turned it into ~1 MiB eagerly allocated per channel at construction.
89 : : // The pool fills as buffers are released.
90 : : memory::MemoryPool pool_{0, 200};
91 : : net::steady_timer retry_timer_;
92 : : net::steady_timer connect_timer_;
93 : : bool owns_ioc_ = true;
94 : : std::atomic<bool> stop_requested_{false};
95 : : std::atomic<bool> stopping_{false};
96 : :
97 : : std::mutex stop_mtx_;
98 : : std::condition_variable stop_cv_;
99 : : bool cleanup_done_ = false;
100 : : bool cleanup_finished_ = false;
101 : : size_t pending_io_ = 0; // Strand-confined, including completion destruction.
102 : : std::mutex join_mtx_;
103 : : // Order accepted write submissions before the cleanup post. A caller
104 : : // that passed a fast precheck must not post old work after completion.
105 : : std::mutex submission_mtx_;
106 : :
107 : : // Caller holds submission_mtx_: state, connection and admission are one decision.
108 : 882 : std::optional<wrapper::SendRejection> admission_rejection(
109 : : std::optional<uint64_t> expected_connection = std::nullopt) {
110 [ + + ]: 882 : if (stop_requested_.load()) {
111 : 62 : std::lock_guard<std::mutex> lock(stop_mtx_);
112 [ + + ]: 62 : return cleanup_done_ ? wrapper::SendRejection::NotStarted : wrapper::SendRejection::Stopping;
113 : 62 : }
114 [ + + ]: 1640 : if (current_seq_.load() == 0) return wrapper::SendRejection::NotStarted;
115 [ + + + - : 2470 : if ((expected_connection && *expected_connection != connection_seq_.load()) || state_.is_state(LinkState::Closed) ||
+ - ]
116 [ + + + - : 2370 : state_.is_state(LinkState::Error) || !ioc_ || !connected_.load()) {
+ + + + ]
117 : 50 : return wrapper::SendRejection::NotReady;
118 : : }
119 : 742 : return std::nullopt;
120 : : }
121 : :
122 : 355 : void mark_cleanup_done() {
123 : 355 : detail::stop_test_hook(this, true);
124 [ + + ]: 355 : if (owns_ioc_) work_guard_.reset();
125 : : {
126 : 355 : std::lock_guard<std::mutex> lock(stop_mtx_);
127 : 355 : cleanup_done_ = true;
128 : 355 : }
129 : 355 : stop_cv_.notify_all();
130 : 355 : }
131 : :
132 : : // The socket interface erases executor associations into std::function.
133 : : // Explicit dispatch preserves serialization. The lifetime also accounts for
134 : : // test sockets that discard an operation instead of invoking its handler.
135 : : template <typename Handler>
136 : 1623 : auto track_io(std::shared_ptr<UdsClient> self, Handler handler) {
137 : 1623 : ++pending_io_;
138 : 4869 : std::shared_ptr<void> lifetime(nullptr, [self](void*) {
139 : 4869 : net::dispatch(self->impl_->strand_, [self] {
140 : 1623 : auto* impl = self->impl_.get();
141 [ + + + + : 1623 : if (impl->cleanup_finished_ && impl->pending_io_ == 1) {
+ + + + +
+ + - - +
- - + + +
+ ]
142 [ - + + + : 167 : if (auto hook = detail::g_uds_io_completion_hook.load()) hook();
- + - - -
+ ]
143 : : }
144 [ + + + + : 1623 : if (--impl->pending_io_ == 0 && impl->cleanup_finished_) impl->mark_cleanup_done();
+ + + + +
+ + + + +
+ - + + -
+ - - - +
+ + + + +
+ ]
145 : : });
146 : : });
147 : 4823 : return [self, lifetime, handler = std::move(handler)](auto... args) {
148 : 3200 : net::dispatch(self->impl_->strand_, [lifetime, handler, args...]() mutable { handler(args...); });
149 : 3246 : };
150 : 1623 : }
151 : :
152 : : // Sized from cfg_.read_buffer_size in init() rather than being a fixed
153 : : // std::array, so a bulk-transfer workload can trade memory for fewer read
154 : : // completions and callback dispatches.
155 : : std::shared_ptr<std::vector<uint8_t>> rx_;
156 : : using Ledger = diagnostics::SendAccountingLedger;
157 : : struct TrackedBuffer {
158 : : BufferVariant buffer;
159 : : Ledger::Request request;
160 : : };
161 : : struct BufferProjection {
162 : 750 : const BufferVariant& operator()(const TrackedBuffer& item) const { return item.buffer; }
163 : : };
164 : : Ledger send_accounting_;
165 : : std::deque<TrackedBuffer> tx_;
166 : : std::deque<TrackedBuffer> pending_;
167 : : std::atomic<size_t> pending_bytes_{0};
168 : : // Each completion retains its gather buffers across loss/reconnect. A stale
169 : : // completion may release its batch but must not mutate the new connection.
170 : : struct WriteBatch {
171 : : std::vector<TrackedBuffer> buffers;
172 : : std::vector<net::const_buffer> views;
173 : : size_t bytes = 0;
174 : : };
175 : : std::shared_ptr<WriteBatch> active_write_;
176 : : bool writing_ = false;
177 : : std::atomic<size_t> queue_bytes_{0};
178 : : // Bytes accepted by a plain async_write_* call but not yet routed onto the
179 : : // strand - reserved via try_reserve_limit_bytes() to close the
180 : : // accept-then-drop race (jwsung91/wirestead#517). inflight_bytes_ mutations
181 : : // and the queue_bytes_/pending_bytes_ increments that promote a
182 : : // reservation both go through write_reserve_mtx_ - see bp_utils.hpp.
183 : : std::atomic<size_t> inflight_bytes_{0};
184 : : std::mutex write_reserve_mtx_;
185 : : // Atomic rather than mutex-guarded: read both from the strand and from
186 : : // arbitrary caller threads (async_try_write_* fast-fail prechecks) (#436).
187 : : std::atomic<base::constants::BackpressureStrategy> bp_strategy_{base::constants::BackpressureStrategy::Reliable};
188 : : size_t bp_high_;
189 : : size_t bp_low_;
190 : : size_t bp_limit_;
191 : : std::atomic<bool> backpressure_active_{false};
192 : : diagnostics::RuntimeStatsCounters stats_;
193 : :
194 : : // Shared snapshots: the io thread copies one out per received chunk, and a
195 : : // std::function copy allocates whenever the target outgrows its small-object
196 : : // buffer. See interface::SharedCallback.
197 : : interface::SharedCallback<OnBytes> on_bytes_;
198 : : interface::SharedCallback<OnState> on_state_;
199 : : interface::SharedCallback<OnBackpressure> on_bp_;
200 : : mutable std::mutex callback_mtx_;
201 : : std::atomic<bool> connected_{false};
202 : 343 : AtomicLinkState state_{LinkState::Idle};
203 : : int retry_attempts_ = 0;
204 : : uint32_t reconnect_attempt_count_{0};
205 : : std::optional<ReconnectPolicy> reconnect_policy_;
206 : :
207 : : ErrorInfoHolder error_info_holder_{"uds_client"};
208 : :
209 : 343 : Impl(const UdsClientConfig& cfg, net::io_context* ioc_ptr,
210 : : std::unique_ptr<interface::UdsSocketInterface> socket = nullptr)
211 [ + + ]: 343 : : owned_ioc_(ioc_ptr ? nullptr : std::make_shared<net::io_context>()),
212 [ + + ]: 343 : ioc_(ioc_ptr ? ioc_ptr : owned_ioc_.get()),
213 : 343 : strand_(net::make_strand(*ioc_)),
214 : 343 : socket_(std::move(socket)),
215 : 343 : cfg_(cfg),
216 : 343 : retry_timer_(strand_),
217 : 343 : connect_timer_(strand_),
218 : 343 : owns_ioc_(!ioc_ptr),
219 : 343 : bp_strategy_(cfg.backpressure_strategy),
220 : 2058 : bp_high_(cfg.backpressure_threshold) {
221 [ + + ]: 343 : if (!socket_) {
222 : 168 : socket_ = std::make_unique<BoostUdsSocket>(uds::socket(strand_));
223 : : }
224 : 343 : init();
225 : 383 : }
226 : :
227 : 128 : void mark_disconnected() {
228 : 128 : std::lock_guard<std::mutex> lock(submission_mtx_);
229 : 128 : mark_disconnected_locked();
230 : 128 : }
231 : 216 : void mark_disconnected_locked() {
232 [ + + ]: 216 : if (connected_.exchange(false)) {
233 : 120 : send_accounting_.end(Ledger::Cause::ConnectionLoss);
234 [ + - ]: 120 : if (write_wait_) write_wait_->end(wrapper::SendRejection::NotReady);
235 : 120 : connection_seq_.fetch_add(1);
236 : : }
237 : 216 : }
238 : :
239 : : void discard_connection_writes();
240 : :
241 : 343 : void init() {
242 : 343 : connected_ = false;
243 : 343 : writing_ = false;
244 : 343 : queue_bytes_ = 0;
245 : 343 : pending_bytes_ = 0;
246 : 343 : config::detail::validate(cfg_);
247 : 341 : rx_ = std::make_shared<std::vector<uint8_t>>(cfg_.read_buffer_size);
248 : 341 : recalculate_backpressure_bounds();
249 : 341 : }
250 : :
251 : : void do_connect(std::shared_ptr<UdsClient> self, uint64_t seq);
252 : : void schedule_retry(std::shared_ptr<UdsClient> self, uint64_t seq);
253 : : void start_read(std::shared_ptr<UdsClient> self, uint64_t seq);
254 : : void do_write(std::shared_ptr<UdsClient> self, uint64_t seq);
255 : : void handle_close(std::shared_ptr<UdsClient> self, uint64_t seq, const boost::system::error_code& ec = {});
256 : : void transition_to(LinkState next, const boost::system::error_code& ec = {});
257 : : void perform_stop_cleanup();
258 : : void close_socket();
259 : : void recalculate_backpressure_bounds();
260 : : void report_backpressure(std::shared_ptr<UdsClient> self, size_t queued_bytes);
261 : : void observe_queue();
262 : : // Shared decide_enqueue()/route dispatch used by both async_write_* variants (#434).
263 : : void route_enqueued_buffer(std::shared_ptr<UdsClient> self, TrackedBuffer&& buf, size_t added);
264 : : queue_util::BackpressureFields bp_fields();
265 : : void record_error(diagnostics::ErrorLevel lvl, diagnostics::ErrorCategory cat, std::string_view operation,
266 : : const boost::system::error_code& ec, std::string_view msg, bool retryable, uint32_t retry_count);
267 : :
268 : 341 : ~Impl() {
269 : 341 : stop_requested_ = true;
270 : 341 : stopping_ = true;
271 : :
272 : 341 : retry_timer_.cancel();
273 : 341 : connect_timer_.cancel();
274 : 341 : close_socket();
275 : :
276 [ - + ]: 341 : if (work_guard_) {
277 : 0 : work_guard_.reset();
278 : : }
279 : :
280 [ + - + + : 341 : if (ioc_ && owns_ioc_ && ioc_thread_.joinable()) {
- + - + ]
281 [ # # ]: 0 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
282 : 0 : ioc_thread_.detach();
283 : : } else {
284 : 0 : ioc_thread_.request_stop();
285 : 0 : ioc_thread_.join();
286 : : }
287 : : }
288 : 341 : }
289 : : };
290 : :
291 : 64 : std::shared_ptr<UdsClient> UdsClient::create(const UdsClientConfig& cfg) {
292 : 64 : return std::shared_ptr<UdsClient>(new UdsClient(cfg));
293 : : }
294 : :
295 : 104 : std::shared_ptr<UdsClient> UdsClient::create(const UdsClientConfig& cfg, boost::asio::io_context& ioc) {
296 : 104 : return std::shared_ptr<UdsClient>(new UdsClient(cfg, ioc));
297 : : }
298 : :
299 : 175 : std::shared_ptr<UdsClient> UdsClient::create(const UdsClientConfig& cfg,
300 : : std::unique_ptr<interface::UdsSocketInterface> socket,
301 : : boost::asio::io_context& ioc) {
302 : 176 : return std::shared_ptr<UdsClient>(new UdsClient(cfg, std::move(socket), ioc));
303 : : }
304 : :
305 : 65 : UdsClient::UdsClient(const UdsClientConfig& cfg) : impl_(std::make_unique<Impl>(cfg, nullptr)) {}
306 : 104 : UdsClient::UdsClient(const UdsClientConfig& cfg, boost::asio::io_context& ioc)
307 : 104 : : impl_(std::make_unique<Impl>(cfg, &ioc)) {}
308 : :
309 : 175 : UdsClient::UdsClient(const UdsClientConfig& cfg, std::unique_ptr<interface::UdsSocketInterface> socket,
310 : 175 : boost::asio::io_context& ioc)
311 : 176 : : impl_(std::make_unique<Impl>(cfg, &ioc, std::move(socket))) {}
312 : :
313 : 683 : UdsClient::~UdsClient() {
314 : : // #446: null after being moved-from - the move ctor/assignment are
315 : : // defaulted, and destroying a moved-from instance must not dereference
316 : : // a null impl_ (matches TcpServer/Serial/UdpChannel/UdsServer's
317 : : // destructors, which already guard this way).
318 [ + + ]: 342 : if (!impl_) return;
319 : 341 : stop();
320 : :
321 [ + + - + : 341 : if (impl_->owns_ioc_ && impl_->ioc_thread_.joinable()) {
- + ]
322 [ # # ]: 0 : if (std::this_thread::get_id() != impl_->ioc_thread_.get_id()) {
323 : 0 : impl_->ioc_thread_.join();
324 : : } else {
325 : 0 : impl_->ioc_thread_.detach();
326 : : }
327 : : }
328 : 685 : }
329 : :
330 : 1 : UdsClient::UdsClient(UdsClient&&) noexcept = default;
331 : 0 : UdsClient& UdsClient::operator=(UdsClient&&) noexcept = default;
332 : :
333 : 340 : void UdsClient::start() {
334 : 340 : auto current_state = impl_->state_.get();
335 [ + + - + ]: 340 : if (current_state == LinkState::Connecting || current_state == LinkState::Connected) {
336 : 1 : return;
337 : : }
338 : :
339 : 339 : impl_->recalculate_backpressure_bounds();
340 : 339 : impl_->stop_requested_ = false;
341 : 339 : impl_->stopping_ = false;
342 : : {
343 : 339 : std::lock_guard<std::mutex> lock(impl_->stop_mtx_);
344 : 339 : impl_->cleanup_done_ = false;
345 : 339 : }
346 : 339 : impl_->cleanup_finished_ = false;
347 : 339 : impl_->inflight_bytes_ = 0;
348 : 339 : impl_->active_write_.reset();
349 : 339 : impl_->current_seq_++;
350 : 339 : uint64_t seq = impl_->current_seq_.load();
351 : :
352 [ + + + - : 339 : if (impl_->owns_ioc_ && !impl_->ioc_thread_.joinable()) {
+ + ]
353 [ - + ]: 62 : if (impl_->ioc_->stopped()) {
354 : 0 : impl_->ioc_->restart();
355 : : }
356 : 62 : impl_->work_guard_ =
357 : 124 : std::make_unique<net::executor_work_guard<net::io_context::executor_type>>(net::make_work_guard(*impl_->ioc_));
358 : 124 : impl_->ioc_thread_ = std::jthread([ioc = impl_->owned_ioc_](std::stop_token st) {
359 : 62 : wirestead::concurrency::run_io_thread_init();
360 : : try {
361 : 62 : std::stop_callback cb(st, [ioc] { ioc->stop(); });
362 : 62 : ioc->run();
363 : 62 : } catch (...) {
364 : 0 : }
365 : 124 : });
366 : : }
367 : :
368 : 339 : net::post(impl_->strand_, [self = shared_from_this(), seq]() {
369 [ + - - + : 339 : if (self->impl_->stop_requested_ || seq != self->impl_->current_seq_) return;
- + ]
370 : 339 : self->impl_->transition_to(LinkState::Connecting);
371 : 339 : self->impl_->do_connect(self, seq);
372 : : });
373 : : }
374 : :
375 : 994 : void UdsClient::stop() {
376 : 994 : const bool on_executor = impl_->ioc_->get_executor().running_in_this_thread();
377 : : {
378 : 994 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
379 [ + + ]: 994 : if (!impl_->stop_requested_.exchange(true)) {
380 : 355 : impl_->send_accounting_.end(Impl::Ledger::Cause::ExplicitStop);
381 [ + + ]: 355 : if (impl_->write_wait_) impl_->write_wait_->end(wrapper::SendRejection::CancelledWhileWaiting);
382 : 355 : impl_->stopping_ = true;
383 : 355 : impl_->connected_ = false;
384 [ + + ]: 710 : if (impl_->current_seq_.load() == 0) {
385 : 16 : impl_->perform_stop_cleanup();
386 : : } else {
387 : 339 : auto self = weak_from_this().lock();
388 : 339 : auto* impl = impl_.get();
389 : 678 : net::post(impl_->strand_, [self, impl] { impl->perform_stop_cleanup(); });
390 : 339 : }
391 : : }
392 : 994 : }
393 [ + + ]: 994 : if (on_executor) return;
394 : :
395 : 935 : detail::stop_test_hook(impl_.get(), false);
396 : : {
397 : 935 : std::unique_lock<std::mutex> lock(impl_->stop_mtx_);
398 : 2159 : impl_->stop_cv_.wait(lock, [this] { return impl_->cleanup_done_; });
399 : 935 : }
400 : 935 : std::lock_guard<std::mutex> join_lock(impl_->join_mtx_);
401 [ + + + + : 935 : if (impl_->owns_ioc_ && impl_->ioc_thread_.joinable()) impl_->ioc_thread_.join();
+ + + - ]
402 : 935 : }
403 : :
404 : 848 : bool UdsClient::is_connected() const { return impl_->connected_.load(); }
405 : 294 : bool UdsClient::is_backpressure_active() const { return impl_->backpressure_active_.load(); }
406 : 418 : std::optional<size_t> UdsClient::write_queue_limit() const { return impl_->bp_limit_; }
407 : 733 : wrapper::RuntimeStats UdsClient::stats() const {
408 : 2932 : auto result = impl_->stats_.snapshot(impl_->queue_bytes_.load(std::memory_order_relaxed),
409 : 733 : impl_->pending_bytes_.load(std::memory_order_relaxed),
410 : 733 : impl_->backpressure_active_.load(std::memory_order_relaxed));
411 : 733 : result.send_accounting = impl_->send_accounting_.snapshot();
412 : 733 : return result;
413 : : }
414 : 22 : void UdsClient::reset_stats() {
415 : 22 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
416 : 22 : impl_->send_accounting_.reset();
417 : 22 : impl_->stats_.reset(impl_->queue_bytes_.load(std::memory_order_relaxed) +
418 : 44 : impl_->pending_bytes_.load(std::memory_order_relaxed));
419 : 22 : }
420 : :
421 : 4 : void UdsClient::fail_receive() {
422 : 4 : const auto connection = write_connection();
423 [ - + ]: 4 : if (!connection) return;
424 : 4 : const auto run = impl_->current_seq_.load();
425 : 4 : net::dispatch(impl_->strand_, [self = shared_from_this(), connection, run] {
426 [ + - + - : 8 : if (run != self->impl_->current_seq_.load() || self->write_connection() != connection) return;
- + - + ]
427 : 4 : self->impl_->handle_close(self, run, make_error_code(boost::system::errc::no_buffer_space));
428 : : });
429 : : }
430 : :
431 : 802 : boost::asio::any_io_executor UdsClient::get_executor() { return impl_->strand_; }
432 : :
433 : 43 : std::shared_ptr<detail::UdsWriteWait> UdsClient::capture_write_wait() const {
434 : 43 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
435 [ + - - + : 43 : if (impl_->stop_requested_.load() || !impl_->connected_.load()) return {};
- + ]
436 : 43 : return impl_->write_wait_;
437 : 43 : }
438 : :
439 : 36 : std::optional<wrapper::SendResult> UdsClient::poll_write_wait(const std::shared_ptr<detail::UdsWriteWait>& wait) const {
440 : 36 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
441 [ - + ]: 36 : if (!wait) return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
442 [ + + ]: 36 : if (wait->ended_by) return wrapper::SendResult::reject(*wait->ended_by);
443 [ + - - + : 24 : if (wait->sequence != impl_->connection_seq_.load() || !impl_->connected_.load())
- + ]
444 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
445 [ + - ]: 12 : if (!impl_->backpressure_active_.load()) return wrapper::SendResult::accept();
446 : 0 : return std::nullopt;
447 : 36 : }
448 : :
449 : 429 : void UdsClient::cancel_write_waits() {
450 : 429 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
451 [ + + ]: 429 : if (impl_->write_wait_) impl_->write_wait_->end(wrapper::SendRejection::CancelledWhileWaiting);
452 : 429 : }
453 : :
454 : 186 : std::optional<uint64_t> UdsClient::write_connection() const {
455 : 186 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
456 [ + - - + : 186 : if (impl_->stop_requested_.load() || !impl_->connected_.load()) return std::nullopt;
- + ]
457 : 372 : return impl_->connection_seq_.load();
458 : 186 : }
459 : :
460 : 171 : wrapper::SendResult UdsClient::write_state() {
461 : 171 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
462 [ + - + + ]: 171 : if (auto reason = impl_->admission_rejection()) return wrapper::SendResult::reject(*reason);
463 : 125 : return wrapper::SendResult::accept();
464 : 171 : }
465 : :
466 : 110 : wrapper::SendResult UdsClient::async_write_copy_result(memory::ConstByteSpan data) {
467 : 110 : const auto result = write_copy(data, std::nullopt);
468 [ + + + - ]: 110 : if (auto hook = detail::g_uds_write_result_hook.load()) hook(result);
469 : 110 : return result;
470 : : }
471 : :
472 : 189 : wrapper::SendResult UdsClient::write_copy(memory::ConstByteSpan data, std::optional<uint64_t> expected_connection) {
473 [ + + ]: 189 : if (expected_connection) {
474 [ + + + - ]: 79 : if (auto hook = detail::g_uds_pinned_write_hook.load()) hook();
475 : : }
476 : 189 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
477 : 189 : const auto seq = impl_->current_seq_.load();
478 : 189 : const auto connection = impl_->connection_seq_.load();
479 [ + - + + ]: 189 : if (auto reason = impl_->admission_rejection(expected_connection)) {
480 : 15 : impl_->stats_.record_failed_send();
481 : 15 : return wrapper::SendResult::reject(*reason);
482 : : }
483 [ - + - - ]: 174 : if (auto hook = detail::g_uds_write_admission_hook.load()) hook();
484 : :
485 : 174 : size_t size = data.size();
486 [ + + ]: 174 : if (size == 0) {
487 : 4 : WIRESTEAD_LOG_WARNING("uds_client", "async_write_copy", "Ignoring zero-length write");
488 : 4 : impl_->stats_.record_failed_send();
489 : 4 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
490 : : }
491 : :
492 [ - + ]: 170 : if (size > base::constants::MAX_BUFFER_SIZE) {
493 : 0 : WIRESTEAD_LOG_ERROR("uds_client", "async_write_copy",
494 : : fmt::format("Write size exceeds maximum allowed ({} bytes)", size));
495 : 0 : impl_->stats_.record_failed_send();
496 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
497 : : }
498 : :
499 [ + + + + : 170 : if (size <= 65536 && impl_->cfg_.enable_memory_pool) {
+ + ]
500 : : try {
501 : 127 : memory::PooledBuffer pooled_buffer(size, impl_->pool_);
502 [ + - + - ]: 127 : if (pooled_buffer.valid()) {
503 : 127 : base::safe_memory::safe_memcpy(pooled_buffer.data(), data.data(), size);
504 : 127 : const auto added = pooled_buffer.size();
505 [ + - + + ]: 127 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
506 : 127 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
507 : 3 : impl_->stats_.record_failed_send();
508 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
509 : : }
510 : 124 : impl_->stats_.record_accepted(added);
511 : 124 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
512 : 124 : const auto request = admission.request();
513 : 124 : net::post(impl_->strand_, [self = shared_from_this(), buf = std::move(pooled_buffer), added, seq, connection,
514 : : request]() mutable {
515 [ - + ]: 248 : if (seq != self->impl_->current_seq_.load()) return;
516 [ + + ]: 248 : if (connection != self->impl_->connection_seq_.load()) {
517 : 3 : self->impl_->stats_.record_dropped(1, added);
518 : 3 : queue_util::release_reserved_limit_bytes(self->impl_->write_reserve_mtx_, self->impl_->inflight_bytes_,
519 : : added);
520 : 3 : return;
521 : : }
522 : 121 : self->impl_->route_enqueued_buffer(self, Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
523 : : });
524 : 124 : admission.commit();
525 : 124 : return wrapper::SendResult::accept();
526 : 124 : }
527 : 127 : } catch (const std::exception& e) {
528 : 0 : WIRESTEAD_LOG_ERROR("uds_client", "async_write_copy",
529 : : fmt::format("Failed to acquire pooled buffer: {}", e.what()));
530 : 0 : }
531 : : }
532 : :
533 : 43 : std::vector<uint8_t> fallback(data.begin(), data.end());
534 : 43 : const auto added = fallback.size();
535 [ + - + + ]: 43 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
536 : 43 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
537 : 2 : impl_->stats_.record_failed_send();
538 : 2 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
539 : : }
540 : 41 : impl_->stats_.record_accepted(added);
541 : 41 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
542 : 41 : const auto request = admission.request();
543 : :
544 : 41 : net::post(impl_->strand_, [self = shared_from_this(), buf = std::move(fallback), added, seq, connection,
545 : : request]() mutable {
546 [ - + ]: 82 : if (seq != self->impl_->current_seq_.load()) return;
547 [ + + ]: 82 : if (connection != self->impl_->connection_seq_.load()) {
548 : 3 : self->impl_->stats_.record_dropped(1, added);
549 : 3 : queue_util::release_reserved_limit_bytes(self->impl_->write_reserve_mtx_, self->impl_->inflight_bytes_, added);
550 : 3 : return;
551 : : }
552 : 38 : self->impl_->route_enqueued_buffer(self, Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
553 : : });
554 : 41 : admission.commit();
555 : 41 : return wrapper::SendResult::accept();
556 : 189 : }
557 : :
558 : 164 : wrapper::SendResult UdsClient::async_write_move_result(std::vector<uint8_t>&& data) {
559 : 164 : const auto result = write_move(std::move(data), std::nullopt);
560 [ + + + - ]: 164 : if (auto hook = detail::g_uds_write_result_hook.load()) hook(result);
561 : 164 : return result;
562 : : }
563 : :
564 : 181 : wrapper::SendResult UdsClient::write_move(std::vector<uint8_t>&& data, std::optional<uint64_t> expected_connection) {
565 [ + + ]: 181 : if (expected_connection) {
566 [ + + + - ]: 17 : if (auto hook = detail::g_uds_pinned_write_hook.load()) hook();
567 : : }
568 : 181 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
569 : 181 : const auto seq = impl_->current_seq_.load();
570 : 181 : const auto connection = impl_->connection_seq_.load();
571 [ + - + + ]: 181 : if (auto reason = impl_->admission_rejection(expected_connection)) {
572 : 12 : impl_->stats_.record_failed_send();
573 : 12 : return wrapper::SendResult::reject(*reason);
574 : : }
575 [ - + - - ]: 169 : if (auto hook = detail::g_uds_write_admission_hook.load()) hook();
576 : 169 : const auto size = data.size();
577 [ + + ]: 169 : if (size == 0) {
578 : 3 : WIRESTEAD_LOG_WARNING("uds_client", "async_write_move", "Ignoring zero-length write");
579 : 3 : impl_->stats_.record_failed_send();
580 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
581 : : }
582 [ + + ]: 166 : if (size > base::constants::MAX_BUFFER_SIZE) {
583 : 1 : WIRESTEAD_LOG_ERROR("uds_client", "async_write_move",
584 : : fmt::format("Write size exceeds maximum allowed ({} bytes)", size));
585 : 1 : impl_->stats_.record_failed_send();
586 : 1 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
587 : : }
588 : :
589 : 165 : const auto added = size;
590 [ + - + + ]: 165 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
591 : 165 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
592 : 16 : impl_->stats_.record_failed_send();
593 : 16 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
594 : : }
595 : 149 : impl_->stats_.record_accepted(added);
596 : 149 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
597 : 149 : const auto request = admission.request();
598 : 149 : net::post(impl_->strand_, [self = shared_from_this(), buf = std::move(data), added, seq, connection,
599 : : request]() mutable {
600 [ - + ]: 298 : if (seq != self->impl_->current_seq_.load()) return;
601 [ + + ]: 298 : if (connection != self->impl_->connection_seq_.load()) {
602 : 5 : self->impl_->stats_.record_dropped(1, added);
603 : 5 : queue_util::release_reserved_limit_bytes(self->impl_->write_reserve_mtx_, self->impl_->inflight_bytes_, added);
604 : 5 : return;
605 : : }
606 : 144 : self->impl_->route_enqueued_buffer(self, Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
607 : : });
608 : 149 : admission.commit();
609 : 149 : return wrapper::SendResult::accept();
610 : 181 : }
611 : :
612 : 70 : wrapper::SendResult UdsClient::async_write_shared_result(std::shared_ptr<const std::vector<uint8_t>> data) {
613 : 70 : const auto result = write_shared(std::move(data), std::nullopt);
614 [ + + + - ]: 70 : if (auto hook = detail::g_uds_write_result_hook.load()) hook(result);
615 : 70 : return result;
616 : : }
617 : :
618 : 74 : wrapper::SendResult UdsClient::write_shared(std::shared_ptr<const std::vector<uint8_t>> data,
619 : : std::optional<uint64_t> expected_connection) {
620 [ + + ]: 74 : if (expected_connection) {
621 [ + + + - ]: 4 : if (auto hook = detail::g_uds_pinned_write_hook.load()) hook();
622 : : }
623 : 74 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
624 : 74 : const auto seq = impl_->current_seq_.load();
625 : 74 : const auto connection = impl_->connection_seq_.load();
626 [ + - + + ]: 74 : if (auto reason = impl_->admission_rejection(expected_connection)) {
627 : 13 : impl_->stats_.record_failed_send();
628 : 13 : return wrapper::SendResult::reject(*reason);
629 : : }
630 [ - + - - ]: 61 : if (auto hook = detail::g_uds_write_admission_hook.load()) hook();
631 [ + - + + : 61 : if (!data || data->empty()) {
+ + ]
632 : 3 : WIRESTEAD_LOG_WARNING("uds_client", "async_write_shared", "Ignoring empty shared buffer");
633 : 3 : impl_->stats_.record_failed_send();
634 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
635 : : }
636 : 58 : const auto size = data->size();
637 [ - + ]: 58 : if (size > base::constants::MAX_BUFFER_SIZE) {
638 : 0 : WIRESTEAD_LOG_ERROR("uds_client", "async_write_shared",
639 : : fmt::format("Write size exceeds maximum allowed ({} bytes)", size));
640 : 0 : impl_->stats_.record_failed_send();
641 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
642 : : }
643 : :
644 : 58 : const auto added = size;
645 [ + - + + ]: 58 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
646 : 58 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
647 : 4 : impl_->stats_.record_failed_send();
648 : 4 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
649 : : }
650 : 54 : impl_->stats_.record_accepted(added);
651 : 54 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
652 : 54 : const auto request = admission.request();
653 : 54 : net::post(impl_->strand_, [self = shared_from_this(), buf = std::move(data), added, seq, connection,
654 : : request]() mutable {
655 [ - + ]: 108 : if (seq != self->impl_->current_seq_.load()) return;
656 [ + + ]: 108 : if (connection != self->impl_->connection_seq_.load()) {
657 : 5 : self->impl_->stats_.record_dropped(1, added);
658 : 5 : queue_util::release_reserved_limit_bytes(self->impl_->write_reserve_mtx_, self->impl_->inflight_bytes_, added);
659 : 5 : return;
660 : : }
661 : 49 : self->impl_->route_enqueued_buffer(self, Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
662 : : });
663 : 54 : admission.commit();
664 : 54 : return wrapper::SendResult::accept();
665 : 74 : }
666 : :
667 : 68 : wrapper::SendResult UdsClient::async_try_write_copy_result(memory::ConstByteSpan data) {
668 : 68 : const auto result = try_write_copy(data);
669 [ + + + - ]: 68 : if (auto hook = detail::g_uds_write_result_hook.load()) hook(result);
670 : 68 : return result;
671 : : }
672 : :
673 : 105 : wrapper::SendResult UdsClient::try_write_copy(memory::ConstByteSpan data) {
674 [ + + ]: 105 : if (data.empty()) {
675 : 3 : impl_->stats_.record_failed_send();
676 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
677 : : }
678 [ - + ]: 102 : if (data.size() > base::constants::MAX_BUFFER_SIZE) {
679 : 0 : impl_->stats_.record_failed_send();
680 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
681 : : }
682 : 204 : return try_write_move(std::vector<uint8_t>(data.begin(), data.end()));
683 : : }
684 : :
685 : 76 : wrapper::SendResult UdsClient::async_try_write_move_result(std::vector<uint8_t>&& data) {
686 : 76 : const auto result = try_write_move(std::move(data));
687 [ + + + - ]: 76 : if (auto hook = detail::g_uds_write_result_hook.load()) hook(result);
688 : 76 : return result;
689 : : }
690 : :
691 : 190 : wrapper::SendResult UdsClient::try_write_move(std::vector<uint8_t>&& data) {
692 : 190 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
693 : 190 : const auto seq = impl_->current_seq_.load();
694 : 190 : const auto connection = impl_->connection_seq_.load();
695 [ + - + + ]: 190 : if (auto reason = impl_->admission_rejection()) {
696 : 38 : impl_->stats_.record_failed_send();
697 : 38 : return wrapper::SendResult::reject(*reason);
698 : : }
699 [ - + - - ]: 152 : if (auto hook = detail::g_uds_write_admission_hook.load()) hook();
700 : 152 : const auto added = data.size();
701 [ + + - + ]: 152 : if (added == 0 || added > base::constants::MAX_BUFFER_SIZE) {
702 : 3 : impl_->stats_.record_failed_send();
703 [ + - ]: 3 : return wrapper::SendResult::reject(added == 0 ? wrapper::SendRejection::InvalidArgument
704 : 3 : : wrapper::SendRejection::TooLarge);
705 : : }
706 : 18 : const auto reject_for_pressure = [this, added]() {
707 [ + + ]: 18 : if (impl_->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
708 : 11 : impl_->stats_.record_dropped(1, added);
709 : : } else {
710 : 7 : impl_->stats_.record_failed_send();
711 : : }
712 : 167 : };
713 [ + + + + : 280 : if (impl_->backpressure_active_.load() || impl_->queue_bytes_ + added > impl_->bp_high_ ||
+ + ]
714 [ - + ]: 131 : impl_->queue_bytes_ + impl_->pending_bytes_ + added > impl_->bp_limit_) {
715 : 18 : reject_for_pressure();
716 : 18 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
717 : : }
718 [ + - - + ]: 131 : if (!queue_util::try_reserve_write_bytes(impl_->write_reserve_mtx_, impl_->inflight_bytes_, impl_->queue_bytes_,
719 : 131 : impl_->pending_bytes_, impl_->backpressure_active_, added, impl_->bp_high_,
720 : 131 : impl_->bp_limit_)) {
721 : 0 : reject_for_pressure();
722 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
723 : : }
724 : 131 : impl_->stats_.record_accepted(added);
725 : 131 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
726 : 131 : const auto request = admission.request();
727 : :
728 : 131 : net::post(impl_->strand_,
729 : 262 : [self = shared_from_this(), buf = std::move(data), added, seq, connection, request]() mutable {
730 [ - + ]: 262 : if (seq != self->impl_->current_seq_.load()) return;
731 [ + + ]: 262 : if (connection != self->impl_->connection_seq_.load()) {
732 : 10 : self->impl_->stats_.record_dropped(1, added);
733 : : // The old connection drain already removed this queue reservation.
734 : 10 : return;
735 : : }
736 : 121 : auto impl = self->impl_.get();
737 [ + + + - : 224 : if (impl->stop_requested_.load() || impl->state_.is_state(LinkState::Closed) ||
- + ]
738 : 224 : impl->state_.is_state(LinkState::Error)) {
739 : 18 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, added);
740 : 18 : impl->stats_.record_failed_send();
741 : 18 : return;
742 : : }
743 : :
744 : 103 : impl->tx_.push_back(Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request});
745 : 103 : impl->observe_queue();
746 : 103 : impl->report_backpressure(self, impl->queue_bytes_);
747 [ + + + - ]: 171 : if (!impl->writing_) impl->do_write(self, impl->current_seq_.load());
748 : : });
749 : 131 : admission.commit();
750 : 131 : return wrapper::SendResult::accept();
751 : 190 : }
752 : :
753 : 68 : wrapper::SendResult UdsClient::async_try_write_shared_result(std::shared_ptr<const std::vector<uint8_t>> data) {
754 : 68 : const auto result = try_write_shared(std::move(data));
755 [ + + + - ]: 68 : if (auto hook = detail::g_uds_write_result_hook.load()) hook(result);
756 : 68 : return result;
757 : : }
758 : :
759 : 80 : wrapper::SendResult UdsClient::try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
760 : 80 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
761 : 80 : const auto seq = impl_->current_seq_.load();
762 : 80 : const auto connection = impl_->connection_seq_.load();
763 [ + - + + : 80 : if (!data || data->empty()) {
+ + ]
764 : 3 : impl_->stats_.record_failed_send();
765 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
766 : : }
767 [ - + ]: 77 : if (data->size() > base::constants::MAX_BUFFER_SIZE) {
768 : 0 : impl_->stats_.record_failed_send();
769 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
770 : : }
771 : 77 : const auto added = data->size();
772 : 7 : const auto reject_for_pressure = [this, added]() {
773 [ + + ]: 7 : if (impl_->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
774 : 4 : impl_->stats_.record_dropped(1, added);
775 : : } else {
776 : 3 : impl_->stats_.record_failed_send();
777 : : }
778 : 84 : };
779 [ + - + + ]: 77 : if (auto reason = impl_->admission_rejection()) {
780 : 16 : impl_->stats_.record_failed_send();
781 : 16 : return wrapper::SendResult::reject(*reason);
782 : : }
783 [ - + - - ]: 61 : if (auto hook = detail::g_uds_write_admission_hook.load()) hook();
784 [ + + + + : 115 : if (impl_->backpressure_active_.load() || impl_->queue_bytes_ + added > impl_->bp_high_ ||
+ + ]
785 [ - + ]: 54 : impl_->queue_bytes_ + impl_->pending_bytes_ + added > impl_->bp_limit_) {
786 : 7 : reject_for_pressure();
787 : 7 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
788 : : }
789 [ + - - + ]: 54 : if (!queue_util::try_reserve_write_bytes(impl_->write_reserve_mtx_, impl_->inflight_bytes_, impl_->queue_bytes_,
790 : 54 : impl_->pending_bytes_, impl_->backpressure_active_, added, impl_->bp_high_,
791 : 54 : impl_->bp_limit_)) {
792 : 0 : reject_for_pressure();
793 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
794 : : }
795 : 54 : impl_->stats_.record_accepted(added);
796 : 54 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
797 : 54 : const auto request = admission.request();
798 : :
799 : 54 : net::post(impl_->strand_,
800 : 108 : [self = shared_from_this(), buf = std::move(data), added, seq, connection, request]() mutable {
801 [ - + ]: 108 : if (seq != self->impl_->current_seq_.load()) return;
802 [ + + ]: 108 : if (connection != self->impl_->connection_seq_.load()) {
803 : 5 : self->impl_->stats_.record_dropped(1, added);
804 : : // The old connection drain already removed this queue reservation.
805 : 5 : return;
806 : : }
807 : 49 : auto impl = self->impl_.get();
808 [ + + + - : 94 : if (impl->stop_requested_.load() || impl->state_.is_state(LinkState::Closed) ||
- + ]
809 : 94 : impl->state_.is_state(LinkState::Error)) {
810 : 4 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, added);
811 : 4 : impl->stats_.record_failed_send();
812 : 4 : return;
813 : : }
814 : :
815 : 45 : impl->tx_.push_back(Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request});
816 : 45 : impl->observe_queue();
817 : 45 : impl->report_backpressure(self, impl->queue_bytes_);
818 [ + + + - ]: 73 : if (!impl->writing_) impl->do_write(self, impl->current_seq_.load());
819 : : });
820 : 54 : admission.commit();
821 : 54 : return wrapper::SendResult::accept();
822 : 80 : }
823 : :
824 : 683 : void UdsClient::on_bytes(OnBytes cb) {
825 : 683 : auto shared = interface::share_callback(std::move(cb));
826 : 683 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
827 : 683 : impl_->on_bytes_ = std::move(shared);
828 : 683 : }
829 : :
830 : 688 : void UdsClient::on_state(OnState cb) {
831 : 688 : auto shared = interface::share_callback(std::move(cb));
832 : 688 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
833 : 688 : impl_->on_state_ = std::move(shared);
834 : 688 : }
835 : :
836 : 686 : void UdsClient::on_backpressure(OnBackpressure cb) {
837 : 686 : auto shared = interface::share_callback(std::move(cb));
838 : 686 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
839 : 686 : impl_->on_bp_ = std::move(shared);
840 : 686 : }
841 : :
842 : 36 : void UdsClient::set_backpressure_strategy(base::constants::BackpressureStrategy strategy) {
843 : 36 : config::detail::strategy(strategy);
844 : 36 : impl_->bp_strategy_.store(strategy, std::memory_order_relaxed);
845 : 36 : }
846 : :
847 : 0 : void UdsClient::set_retry_interval(unsigned interval_ms) {
848 : 0 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
849 : 0 : auto candidate = impl_->cfg_;
850 : 0 : candidate.retry_interval_ms = interval_ms;
851 : 0 : config::detail::validate(candidate);
852 : 0 : impl_->cfg_.retry_interval_ms = interval_ms;
853 : 0 : }
854 : :
855 : 0 : void UdsClient::set_reconnect_policy(ReconnectPolicy policy) {
856 : 0 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
857 [ # # ]: 0 : if (policy) {
858 : 0 : impl_->reconnect_policy_ = std::move(policy);
859 : : } else {
860 : 0 : impl_->reconnect_policy_ = std::nullopt;
861 : : }
862 : 0 : }
863 : :
864 : 9 : std::optional<diagnostics::ErrorInfo> UdsClient::last_error_info() const {
865 : 9 : return impl_->error_info_holder_.last_error_info();
866 : : }
867 : :
868 : 393 : void UdsClient::Impl::do_connect(std::shared_ptr<UdsClient> self, uint64_t seq) {
869 [ + - + - : 786 : if (stop_requested_.load() || stopping_.load() || seq != current_seq_.load()) {
- + - + ]
870 : 0 : return;
871 : : }
872 : :
873 [ + - - + ]: 393 : if (!cfg_.is_valid()) {
874 : 0 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONFIGURATION, "connect",
875 : 0 : make_error_code(boost::system::errc::invalid_argument), "Invalid UDS socket path", false,
876 : 0 : self->impl_->reconnect_attempt_count_);
877 : 0 : self->impl_->transition_to(LinkState::Error);
878 : 0 : return;
879 : : }
880 : :
881 : 393 : std::shared_ptr<uds::endpoint> endpoint;
882 : : try {
883 : 393 : endpoint = std::make_shared<uds::endpoint>(cfg_.socket_path);
884 : 0 : } catch (const std::exception& e) {
885 : 0 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONFIGURATION, "connect",
886 : 0 : make_error_code(boost::system::errc::filename_too_long),
887 : 0 : "Invalid UDS endpoint: " + std::string(e.what()), false,
888 : 0 : self->impl_->reconnect_attempt_count_);
889 : 0 : self->impl_->transition_to(LinkState::Error);
890 : 0 : return;
891 : 0 : }
892 : :
893 : : unsigned connection_timeout_ms;
894 : : {
895 : 393 : std::lock_guard<std::mutex> lock(cfg_mtx_);
896 : 393 : connection_timeout_ms = cfg_.connection_timeout_ms;
897 : 393 : }
898 : 393 : connect_timer_.expires_after(std::chrono::milliseconds(connection_timeout_ms));
899 : 393 : connect_timer_.async_wait(track_io(self, [self, seq](const boost::system::error_code& ec) {
900 [ + + - + : 394 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) return;
+ + ]
901 [ + - ]: 1 : if (!ec) {
902 : 2 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "connect",
903 : : boost::asio::error::timed_out, "Connection timed out", true,
904 : 1 : self->impl_->reconnect_attempt_count_);
905 : 1 : self->impl_->handle_close(self, seq, boost::asio::error::timed_out);
906 : : }
907 : : }));
908 : :
909 : 393 : socket_->async_connect(*endpoint, track_io(self, [self, seq, endpoint](const boost::system::error_code& ec) {
910 : 391 : self->impl_->connect_timer_.cancel();
911 [ + - - + : 782 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) return;
- + ]
912 [ + + - + : 391 : if (self->impl_->stop_requested_.load() || self->impl_->stopping_.load()) {
+ + ]
913 : 1 : self->impl_->close_socket();
914 : 1 : return;
915 : : }
916 [ + + ]: 390 : if (ec) {
917 : 22 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "connect", ec,
918 : 22 : "Connect failed: " + ec.message(), diagnostics::is_retryable_uds_connect_error(ec),
919 : 11 : self->impl_->reconnect_attempt_count_);
920 : 11 : self->impl_->schedule_retry(self, seq);
921 : 11 : return;
922 : : }
923 : :
924 : : {
925 : 379 : std::lock_guard<std::mutex> lock(self->impl_->submission_mtx_);
926 [ - + ]: 379 : if (self->impl_->stop_requested_.load()) return;
927 : 379 : const auto connection = self->impl_->connection_seq_.fetch_add(1) + 1;
928 : 379 : self->impl_->write_wait_ = std::make_shared<detail::UdsWriteWait>(connection);
929 : 379 : self->impl_->connected_ = true;
930 : 379 : }
931 : 379 : self->impl_->rx_ = std::make_shared<std::vector<uint8_t>>(self->impl_->cfg_.read_buffer_size);
932 : 379 : self->impl_->reconnect_attempt_count_ = 0;
933 : 379 : self->impl_->retry_attempts_ = 0;
934 : 379 : self->impl_->transition_to(LinkState::Connected);
935 : 379 : self->impl_->start_read(self, seq);
936 : 379 : self->impl_->writing_ = false; // Force reset
937 : 379 : self->impl_->do_write(self, seq);
938 : : }));
939 : 393 : }
940 : :
941 : 132 : void UdsClient::Impl::schedule_retry(std::shared_ptr<UdsClient> self, uint64_t seq) {
942 [ + + - + : 193 : if (stop_requested_.load() || stopping_.load()) return;
+ + ]
943 : :
944 : : // Snapshot once rather than locking repeatedly - cfg_/reconnect_policy_
945 : : // can change concurrently via set_retry_interval() etc. from any user
946 : : // thread while this runs on the strand (#436).
947 : 131 : UdsClientConfig cfg_snapshot;
948 : 131 : std::optional<ReconnectPolicy> reconnect_policy_snapshot;
949 : : {
950 : 131 : std::lock_guard<std::mutex> lock(cfg_mtx_);
951 : 131 : cfg_snapshot = cfg_;
952 : 131 : reconnect_policy_snapshot = reconnect_policy_;
953 : 131 : }
954 : :
955 : : diagnostics::ErrorInfo dummy_err(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "uds_client",
956 : 131 : "connect", "Retry pending", boost::system::error_code(), true);
957 : : auto decision =
958 : 131 : detail::decide_reconnect_uds(cfg_snapshot, dummy_err, reconnect_attempt_count_, reconnect_policy_snapshot);
959 : :
960 [ + + ]: 131 : if (!decision.should_retry) {
961 : 61 : transition_to(LinkState::Error);
962 : 61 : return;
963 : : }
964 : :
965 : 70 : transition_to(LinkState::Connecting);
966 [ + - - + : 70 : if (stop_requested_.load() || stopping_.load()) return;
- + ]
967 : 70 : reconnect_attempt_count_++;
968 : 70 : retry_timer_.expires_after(decision.delay.value_or(std::chrono::milliseconds(cfg_snapshot.retry_interval_ms)));
969 : 70 : retry_timer_.async_wait(track_io(self, [self, seq](const boost::system::error_code& ec) {
970 [ + + - + : 124 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) return;
+ + ]
971 : 54 : self->impl_->do_connect(self, seq);
972 : : }));
973 : 253 : }
974 : :
975 : 443 : void UdsClient::Impl::start_read(std::shared_ptr<UdsClient> self, uint64_t seq) {
976 [ + + + - : 879 : if (stop_requested_.load() || seq != current_seq_.load() || !connected_.load()) return;
+ + + + ]
977 : 432 : const auto connection = connection_seq_.load();
978 : 432 : auto buffer = rx_;
979 : 1296 : socket_->async_read_some(
980 : 864 : net::buffer(*buffer),
981 : 864 : track_io(self, [self, seq, connection, buffer](const boost::system::error_code& ec, size_t bytes) {
982 : 426 : auto* impl = self->impl_.get();
983 [ + - ]: 730 : if (ec == net::error::operation_aborted || seq != impl->current_seq_.load() ||
984 [ + + + + : 730 : connection != impl->connection_seq_.load() || impl->stop_requested_.load())
+ + + + ]
985 : 362 : return;
986 [ + + ]: 127 : if (ec) {
987 : 63 : impl->handle_close(self, seq, ec);
988 : 63 : return;
989 : : }
990 : 64 : interface::SharedCallback<OnBytes> cb;
991 : : {
992 : 64 : std::lock_guard<std::mutex> lock(impl->callback_mtx_);
993 : 64 : cb = impl->on_bytes_;
994 : 64 : }
995 [ + - + - ]: 64 : if (bytes > 0) impl->stats_.record_received(bytes);
996 : 64 : diagnostics::invoke_callback("uds_client", "on_bytes", cb, memory::ConstByteSpan(buffer->data(), bytes));
997 : 64 : impl->start_read(self, seq);
998 : 64 : }));
999 : 432 : }
1000 : :
1001 : 861 : void UdsClient::Impl::do_write(std::shared_ptr<UdsClient> self, uint64_t seq) {
1002 [ + + + - : 1722 : if (stop_requested_.load() || seq != current_seq_.load() || !connected_.load() || tx_.empty() || writing_) return;
+ - + + -
+ + + ]
1003 : : // The ledger, not the admission mutex, fences this handoff against stop and
1004 : : // loss: taking the admission mutex here per batch contends with every producer.
1005 : 337 : auto batch = std::make_shared<WriteBatch>();
1006 : 337 : batch->bytes = queue_util::take_gather_batch(tx_, batch->buffers, batch->views, BufferProjection{});
1007 [ + - + + ]: 1085 : if (!send_accounting_.begin_batch(batch->buffers, [](const auto& item) { return item.request; })) {
1008 : : // Stop/loss already terminated these; leave them to its queue cleanup.
1009 : 2 : queue_util::return_gather_batch(tx_, batch->buffers);
1010 : 2 : return;
1011 : : }
1012 : 335 : writing_ = true;
1013 : 335 : active_write_ = batch;
1014 : 335 : const auto connection = connection_seq_.load();
1015 : 335 : auto completion = track_io(self, [self, seq, connection, batch](boost::system::error_code ec, size_t written) {
1016 : 320 : auto* impl = self->impl_.get();
1017 : : bool current_connection;
1018 : : {
1019 : : // A short composed write without an error is still a terminal failure.
1020 [ + + + + : 320 : if (!ec && written < batch->bytes) ec = net::error::connection_reset;
+ + ]
1021 : : // Only a failure must close admission; success needs just the ledger.
1022 : 320 : std::unique_lock<std::mutex> lock(impl->submission_mtx_, std::defer_lock);
1023 [ + + + - ]: 320 : if (ec) lock.lock();
1024 : 320 : current_connection = connection == impl->connection_seq_.load();
1025 : 640 : impl->send_accounting_.complete_batch(
1026 : 678 : batch->buffers, written, [](const auto& item) { return item.request; },
1027 : 358 : [](const auto& item) {
1028 : 716 : return std::visit([](const auto& b) { return queue_util::variant_buffer_size(b); }, item.buffer);
1029 : : });
1030 [ + + + + : 320 : if (current_connection && ec) impl->mark_disconnected_locked();
+ + + - ]
1031 : 320 : }
1032 [ + + ]: 320 : if (!current_connection) {
1033 : 45 : batch->buffers.clear();
1034 : 45 : return;
1035 : : }
1036 : 275 : impl->active_write_.reset();
1037 : 275 : impl->writing_ = false;
1038 [ + + - + : 475 : if (impl->stop_requested_.load() || seq != impl->current_seq_.load()) {
+ + ]
1039 : 75 : batch->buffers.clear();
1040 : 75 : return;
1041 : : }
1042 [ + + ]: 200 : if (ec) {
1043 : 46 : queue_util::return_gather_batch(impl->tx_, batch->buffers);
1044 : 46 : impl->handle_close(self, seq, ec);
1045 : 46 : return;
1046 : : }
1047 : 154 : batch->buffers.clear();
1048 : 154 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, batch->bytes);
1049 : 154 : impl->stats_.record_sent(written);
1050 : 154 : impl->report_backpressure(self, impl->queue_bytes_);
1051 : 154 : impl->do_write(self, seq);
1052 : 335 : });
1053 : : try {
1054 : 342 : socket_->async_write(batch->views, std::move(completion));
1055 : 7 : } catch (...) {
1056 : 7 : mark_disconnected();
1057 : 7 : const auto ec = make_error_code(boost::system::errc::no_buffer_space);
1058 : 7 : handle_close(self, seq, ec);
1059 : 7 : }
1060 : 337 : }
1061 : :
1062 : 121 : void UdsClient::Impl::discard_connection_writes() {
1063 : 121 : size_t messages = tx_.size() + pending_.size();
1064 : 121 : size_t bytes = 0;
1065 [ + + ]: 247 : for (const auto& buffer : tx_)
1066 : 252 : bytes += std::visit([](const auto& b) { return queue_util::variant_buffer_size(b); }, buffer.buffer);
1067 [ - + ]: 121 : for (const auto& buffer : pending_)
1068 : 0 : bytes += std::visit([](const auto& b) { return queue_util::variant_buffer_size(b); }, buffer.buffer);
1069 [ + + ]: 121 : if (active_write_) {
1070 : 52 : messages += active_write_->buffers.size();
1071 : 52 : bytes += active_write_->bytes;
1072 : : // Its completion handler retains the buffers; it must not mutate new state.
1073 : 52 : active_write_.reset();
1074 : : }
1075 : 121 : tx_.clear();
1076 : 121 : pending_.clear();
1077 : 121 : queue_bytes_ = 0;
1078 : 121 : pending_bytes_ = 0;
1079 : 121 : writing_ = false;
1080 : 121 : backpressure_active_ = false;
1081 [ + + ]: 121 : if (messages) stats_.record_dropped(messages, bytes);
1082 : 121 : }
1083 : :
1084 : 121 : void UdsClient::Impl::handle_close(std::shared_ptr<UdsClient> self, uint64_t seq, const boost::system::error_code& ec) {
1085 [ + - - + : 242 : if (ec == net::error::operation_aborted || seq != current_seq_.load()) return;
- + ]
1086 [ + - + + : 121 : if (ec && ec != net::error::timed_out)
+ + ]
1087 : 120 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "connection", ec,
1088 : 240 : "Connection closed: " + ec.message(), true, reconnect_attempt_count_);
1089 : 121 : mark_disconnected();
1090 : 121 : discard_connection_writes();
1091 : 121 : close_socket();
1092 : 121 : retry_timer_.cancel();
1093 : 121 : connect_timer_.cancel();
1094 : :
1095 [ - + ]: 121 : if (stop_requested_) {
1096 : 0 : transition_to(LinkState::Idle);
1097 : : } else {
1098 : 121 : schedule_retry(self, seq);
1099 : : }
1100 : : }
1101 : :
1102 : 849 : void UdsClient::Impl::transition_to(LinkState next, const boost::system::error_code&) {
1103 : 849 : state_.set(next);
1104 : 849 : interface::SharedCallback<OnState> cb;
1105 : : {
1106 : 849 : std::lock_guard<std::mutex> lock(callback_mtx_);
1107 : 849 : cb = on_state_;
1108 : 849 : }
1109 : 849 : diagnostics::invoke_callback("uds_client", "on_state", cb, next);
1110 : 849 : }
1111 : :
1112 : 355 : void UdsClient::Impl::perform_stop_cleanup() {
1113 : 355 : retry_timer_.cancel();
1114 : 355 : connect_timer_.cancel();
1115 : 355 : close_socket();
1116 : 355 : tx_.clear();
1117 : 355 : queue_bytes_ = 0;
1118 : 355 : pending_.clear();
1119 : 355 : pending_bytes_ = 0;
1120 : 355 : writing_ = false;
1121 : 355 : connected_.store(false);
1122 : 355 : backpressure_active_.store(false);
1123 : :
1124 : 355 : state_.set(LinkState::Idle);
1125 : 355 : cleanup_finished_ = true;
1126 [ + + ]: 355 : if (pending_io_ == 0) mark_cleanup_done();
1127 : 355 : }
1128 : :
1129 : 818 : void UdsClient::Impl::close_socket() {
1130 : 818 : boost::system::error_code ec;
1131 : 818 : socket_->close(ec);
1132 : 818 : }
1133 : :
1134 : 680 : void UdsClient::Impl::recalculate_backpressure_bounds() {
1135 : 680 : bp_high_ = cfg_.backpressure_threshold;
1136 [ + - ]: 680 : bp_low_ = bp_high_ > 1 ? bp_high_ / 2 : bp_high_;
1137 [ - + ]: 680 : if (bp_low_ == 0) bp_low_ = 1;
1138 : 680 : bp_limit_ = std::min(std::max(bp_high_ * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
1139 : : base::constants::MAX_BUFFER_SIZE);
1140 [ - + ]: 680 : if (bp_limit_ < bp_high_) bp_limit_ = bp_high_;
1141 : 680 : backpressure_active_ = false;
1142 : 680 : }
1143 : :
1144 : 922 : queue_util::BackpressureFields UdsClient::Impl::bp_fields() {
1145 : 922 : return queue_util::BackpressureFields{queue_bytes_,
1146 : 922 : pending_bytes_,
1147 : 922 : backpressure_active_,
1148 : 922 : bp_high_,
1149 : 922 : bp_low_,
1150 : 922 : bp_limit_,
1151 : 922 : bp_strategy_.load(std::memory_order_relaxed),
1152 : 922 : &write_reserve_mtx_};
1153 : : }
1154 : :
1155 : 352 : void UdsClient::Impl::route_enqueued_buffer(std::shared_ptr<UdsClient> self, TrackedBuffer&& buf, size_t added) {
1156 [ + + - + : 352 : if (stop_requested_ || stopping_) {
+ + ]
1157 : 29 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
1158 : 54 : return;
1159 : : }
1160 : 323 : auto f = bp_fields();
1161 : 323 : queue_util::DropAccounting dropped;
1162 : 323 : auto decision = queue_util::decide_enqueue(
1163 : 323 : f, added, tx_, dropped, BufferProjection{},
1164 : : [this](const TrackedBuffer& item) { send_accounting_.discard(item.request, Ledger::Cause::QueuePressure); });
1165 [ - + - - ]: 323 : if (dropped.any()) stats_.record_dropped(dropped.messages, dropped.bytes);
1166 : :
1167 [ - + ]: 323 : if (decision == queue_util::EnqueueDecision::Rejected) {
1168 : 0 : send_accounting_.discard(buf.request, Ledger::Cause::QueuePressure);
1169 : 0 : WIRESTEAD_LOG_ERROR("uds_client", "write", fmt::format("Queue limit exceeded ({} bytes)", queue_bytes_ + added));
1170 : : // #448: record as dropped so it's reflected in RuntimeStats instead of
1171 : : // silently vanishing after being counted as accepted.
1172 : 0 : stats_.record_dropped(1, added);
1173 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
1174 : 0 : report_backpressure(self, queue_bytes_ + added);
1175 : 0 : return;
1176 : : }
1177 [ + + ]: 323 : if (decision == queue_util::EnqueueDecision::Pending) {
1178 : 25 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, pending_bytes_, inflight_bytes_, added);
1179 : 25 : pending_.emplace_back(std::move(buf));
1180 : 25 : observe_queue();
1181 : 25 : return;
1182 : : }
1183 : 298 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, queue_bytes_, inflight_bytes_, added);
1184 : 298 : tx_.emplace_back(std::move(buf));
1185 : 298 : observe_queue();
1186 : 298 : report_backpressure(self, queue_bytes_);
1187 [ + + + - ]: 516 : if (!writing_) do_write(self, current_seq_.load());
1188 : : }
1189 : :
1190 : 1084 : void UdsClient::Impl::observe_queue() {
1191 : 3252 : stats_.observe_queue(queue_bytes_.load(std::memory_order_relaxed) + pending_bytes_.load(std::memory_order_relaxed));
1192 : 1084 : }
1193 : :
1194 : 600 : void UdsClient::Impl::report_backpressure(std::shared_ptr<UdsClient> self, size_t queued_bytes) {
1195 [ + + - + : 600 : if (stop_requested_.load() || stopping_.load()) return;
+ + ]
1196 : 599 : observe_queue();
1197 : :
1198 : 599 : interface::SharedCallback<OnBackpressure> on_bp;
1199 : : {
1200 : 599 : std::lock_guard<std::mutex> lock(callback_mtx_);
1201 : 599 : on_bp = on_bp_;
1202 : 599 : }
1203 : 599 : static const OnBackpressure kNoCallback;
1204 : :
1205 : 599 : auto f = bp_fields();
1206 [ + + + - ]: 809 : queue_util::report_backpressure(
1207 : 210 : f, queued_bytes, on_bp ? *on_bp : kNoCallback, stats_,
1208 : 0 : [&]() -> size_t {
1209 : 14 : const size_t moved = pending_bytes_.exchange(0);
1210 [ + + ]: 18 : while (!pending_.empty()) {
1211 : 4 : tx_.emplace_back(std::move(pending_.front()));
1212 : 4 : pending_.pop_front();
1213 : : }
1214 : 14 : return moved;
1215 : : },
1216 : 599 : [&]() {
1217 : 14 : observe_queue();
1218 [ + - + - ]: 28 : if (!writing_) do_write(self, current_seq_.load());
1219 : 14 : });
1220 : 599 : }
1221 : :
1222 : 132 : void UdsClient::Impl::record_error(diagnostics::ErrorLevel lvl, diagnostics::ErrorCategory cat,
1223 : : std::string_view operation, const boost::system::error_code& ec,
1224 : : std::string_view msg, bool retryable, uint32_t retry_count) {
1225 : 132 : error_info_holder_.record_error(lvl, cat, operation, ec, msg, retryable, retry_count);
1226 : 132 : }
1227 : :
1228 : : } // namespace transport
1229 : : } // namespace wirestead
|
