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/tcp_client/tcp_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/transport/base/stop_test_hook.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 : : #ifdef WIRESTEAD_TLS_ENABLED
35 : : #include <boost/asio/ssl.hpp>
36 : : #endif
37 : : #include <condition_variable>
38 : : #include <cstdint>
39 : : #include <cstring>
40 : : #include <deque>
41 : : #include <iostream>
42 : : #include <memory>
43 : : #include <mutex>
44 : : #include <optional>
45 : : #include <stop_token>
46 : : #include <string>
47 : : #include <thread>
48 : : #include <type_traits>
49 : : #include <variant>
50 : : #include <vector>
51 : :
52 : : #if defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__)
53 : : #include <sys/socket.h>
54 : : #include <sys/types.h>
55 : : #endif
56 : :
57 : : #include "wirestead/base/constants.hpp"
58 : : #include "wirestead/concurrency/io_context_manager.hpp"
59 : : #include "wirestead/concurrency/thread_safe_state.hpp"
60 : : #include "wirestead/diagnostics/error_handler.hpp"
61 : : #include "wirestead/diagnostics/error_mapping.hpp"
62 : : #include "wirestead/diagnostics/logger.hpp"
63 : : #include "wirestead/diagnostics/runtime_stats_counter.hpp"
64 : : #include "wirestead/diagnostics/send_accounting.hpp"
65 : : #include "wirestead/memory/memory_pool.hpp"
66 : : #include "wirestead/transport/base/bp_state_machine.hpp"
67 : : #include "wirestead/transport/base/bp_utils.hpp"
68 : : #include "wirestead/transport/base/error_info_holder.hpp"
69 : : #include "wirestead/transport/tcp_client/detail/reconnect_decider.hpp"
70 : : #include "wirestead/transport/tcp_client/detail/write_wait.hpp"
71 : :
72 : : namespace wirestead {
73 : : namespace transport {
74 : :
75 : : namespace net = boost::asio;
76 : : using tcp = net::ip::tcp;
77 : :
78 : : using base::LinkState;
79 : : using concurrency::AtomicLinkState;
80 : : using config::TcpClientConfig;
81 : : using interface::Channel;
82 : :
83 : : struct TcpClient::Impl {
84 : : using Ledger = diagnostics::SendAccountingLedger;
85 : : struct TrackedBuffer {
86 : : BufferVariant buffer;
87 : : Ledger::Request request;
88 : : };
89 : : struct BufferProjection {
90 : : template <typename T>
91 : 8128 : decltype(auto) operator()(T& item) const {
92 : 8128 : return (item.buffer);
93 : : }
94 : : };
95 : : Ledger send_accounting_;
96 : : // Members moved from TcpClient
97 : : std::shared_ptr<net::io_context> owned_ioc_;
98 : : net::io_context* ioc_ = nullptr;
99 : : net::strand<net::io_context::executor_type> strand_;
100 : : // Internal write lambdas have the default allocator and no associated executor.
101 : : // Cache post's properties once; never execute inline under submission_mtx_.
102 : : const net::strand<net::io_context::executor_type> write_executor_;
103 : : std::unique_ptr<net::executor_work_guard<net::io_context::executor_type>> work_guard_;
104 : : std::jthread ioc_thread_;
105 : : std::atomic<uint64_t> lifecycle_seq_{0};
106 : : std::atomic<uint64_t> stop_seq_{0};
107 : : std::atomic<uint64_t> current_seq_{0};
108 : : // Changes when a usable connection ends, independently of start/stop runs.
109 : : std::atomic<uint64_t> connection_seq_{0};
110 : : std::shared_ptr<detail::TcpWriteWait> write_wait_;
111 : : tcp::resolver resolver_;
112 : : tcp::socket socket_;
113 : :
114 : : #ifdef WIRESTEAD_TLS_ENABLED
115 : : // The stream borrows socket_ rather than owning it, so connect, socket
116 : : // options, cancel and close all keep operating on socket_ exactly as they do
117 : : // without TLS. Only reads, writes and shutdown route through here, and only
118 : : // while a connection is up - it is rebuilt per connection because a TLS
119 : : // session cannot outlive the socket it negotiated on.
120 : : std::shared_ptr<boost::asio::ssl::context> ssl_context_;
121 : : std::optional<boost::asio::ssl::stream<tcp::socket&>> tls_;
122 : : // Whether IO should currently be routed through tls_, tracked separately
123 : : // from tls_.has_value() because the stream now outlives the connection - see
124 : : // close_socket(). Strand-confined, like the stream itself.
125 : : bool tls_engaged_ = false;
126 : : #endif
127 : :
128 : : // True when this connection is encrypted. Reads to it happen on the strand.
129 : 2313 : bool tls_active() const {
130 : : #ifdef WIRESTEAD_TLS_ENABLED
131 : 2313 : return tls_engaged_;
132 : : #else
133 : : return false;
134 : : #endif
135 : : }
136 : : // Guards the mutable subset of cfg_ (retry_interval_ms, max_retries,
137 : : // connection_timeout_ms, idle_timeout_ms, idle_timeout_action) and
138 : : // reconnect_policy_ below - the fields that have runtime setters
139 : : // (set_retry_interval() etc.) reachable from any user thread while the
140 : : // strand concurrently reads them for reconnect/idle-timeout decisions.
141 : : // Fields with no runtime setter (tcp_no_delay, keep_alive, buffer sizes)
142 : : // are set once at construction and read only at connect time, so they
143 : : // don't need this lock (#436).
144 : : mutable std::mutex cfg_mtx_;
145 : : TcpClientConfig cfg_;
146 : : // #443: per-channel pool instead of the process-wide GlobalMemoryPool
147 : : // singleton - avoids cross-channel contention on the singleton's bucket
148 : : // mutexes. Capacity is much smaller than the old shared default (400/2000)
149 : : // since it's no longer amortized across every channel in the process.
150 : : // Prefill stays 0. This literal was written while MemoryPool discarded
151 : : // initial_pool_size, so 50 allocated nothing; #575 made the parameter real
152 : : // and turned it into ~1 MiB eagerly allocated per channel at construction.
153 : : // The pool fills as buffers are released.
154 : : memory::MemoryPool pool_{0, 200};
155 : : net::steady_timer retry_timer_;
156 : : net::steady_timer connect_timer_;
157 : : net::steady_timer idle_timer_;
158 : : bool owns_ioc_ = true;
159 : : // Orders caller-side write admission and strand submission before stop.
160 : : // Never run strand handlers/user callbacks while holding this mutex.
161 : : std::mutex submission_mtx_;
162 : : std::atomic<bool> stop_requested_{false};
163 : : std::atomic<bool> stopping_{false};
164 : : std::atomic<bool> terminal_state_notified_{false};
165 : : std::atomic<bool> reconnect_pending_{false};
166 : :
167 : : // Per-connection storage, sized from cfg_.read_buffer_size. A cancelled
168 : : // read retains its buffer even if a replacement connection is ready.
169 : : std::shared_ptr<std::vector<uint8_t>> rx_;
170 : : std::deque<TrackedBuffer> tx_;
171 : : std::deque<TrackedBuffer> pending_;
172 : : std::atomic<size_t> pending_bytes_{0};
173 : : // Each operation owns its buffers until its completion handler returns,
174 : : // even if a replacement connection has already started writing.
175 : : struct WriteBatch {
176 : : std::vector<TrackedBuffer> buffers;
177 : : std::vector<net::const_buffer> views;
178 : : size_t bytes = 0;
179 : : };
180 : : std::shared_ptr<WriteBatch> active_write_;
181 : : bool writing_ = false;
182 : : std::atomic<size_t> queue_bytes_{0};
183 : : // Bytes accepted by a plain async_write_* call but not yet routed onto the
184 : : // strand - reserved via try_reserve_limit_bytes() to close the
185 : : // accept-then-drop race (jwsung91/wirestead#517). inflight_bytes_ mutations
186 : : // and the queue_bytes_/pending_bytes_ increments that promote a
187 : : // reservation both go through write_reserve_mtx_ - see bp_utils.hpp.
188 : : std::atomic<size_t> inflight_bytes_{0};
189 : : std::mutex write_reserve_mtx_;
190 : : // D-1: shutdown *requested* and shutdown *completed* are separate states.
191 : : // The cleanup handler that runs on the strand is what completes it, and
192 : : // waiting callers wait for that signal rather than for a mutex - a mutex
193 : : // only says another caller was here, not that the teardown ran.
194 : : std::mutex stop_mtx_;
195 : : std::condition_variable stop_cv_;
196 : : bool cleanup_done_ = false;
197 : : std::atomic<bool> cleanup_started_{false};
198 : : // Strand-confined. Cancellation does not mean its completion handler has
199 : : // run; keep the run closed until those handlers have left the old buffers.
200 : : size_t pending_io_ = 0;
201 : : bool cleanup_finished_ = false;
202 : : struct IoCompletion {
203 : : Impl* impl;
204 : 4275 : ~IoCompletion() {
205 [ + + + + ]: 4275 : if (impl->cleanup_finished_ && impl->pending_io_ == 1) {
206 [ + + ]: 476 : if (auto hook = detail::g_tcp_io_completion_hook.load()) hook();
207 : : }
208 [ + + + + : 4275 : if (--impl->pending_io_ == 0 && impl->cleanup_finished_) impl->mark_cleanup_done();
+ + ]
209 : 4275 : }
210 : : };
211 : :
212 : : // Serializes the join itself: joining a thread from two callers is not
213 : : // allowed, and the second caller must not return before the first has.
214 : : std::mutex join_mtx_;
215 : :
216 : 722 : void mark_cleanup_done() {
217 : 722 : detail::stop_test_hook(this, true);
218 : : {
219 : 722 : std::lock_guard<std::mutex> lock(stop_mtx_);
220 : 722 : cleanup_done_ = true;
221 : 722 : }
222 : 722 : stop_cv_.notify_all();
223 : 722 : }
224 : :
225 : : // Waits for the teardown that was posted to the strand. The contract's
226 : : // precondition for an externally run io_context is that the caller keeps it
227 : : // running, so this waits and nothing more: running the caller's executor
228 : : // here would execute other channels' handlers on the stopping thread.
229 : 1459 : void wait_for_cleanup() {
230 : 1459 : detail::stop_test_hook(this, false);
231 : 1459 : std::unique_lock<std::mutex> lock(stop_mtx_);
232 : 3576 : stop_cv_.wait(lock, [this] { return cleanup_done_; });
233 : 1459 : }
234 : :
235 : : // Called with submission_mtx_ held. Explicit stop and readiness publication
236 : : // use that mutex; cleanup completion has its own condition-variable mutex.
237 : 6471 : std::optional<wrapper::SendRejection> admission_rejection(
238 : : std::optional<uint64_t> expected_connection = std::nullopt) {
239 [ + + ]: 6471 : if (stop_requested_.load()) {
240 : 151 : std::lock_guard<std::mutex> lock(stop_mtx_);
241 [ + + ]: 151 : return cleanup_done_ ? wrapper::SendRejection::NotStarted : wrapper::SendRejection::Stopping;
242 : 151 : }
243 [ + + ]: 12640 : if (current_seq_.load() == 0) return wrapper::SendRejection::NotStarted;
244 [ + + + - : 20549 : if ((expected_connection && *expected_connection != connection_seq_.load()) || state_.is_state(LinkState::Closed) ||
+ - ]
245 [ + + + - : 18747 : state_.is_state(LinkState::Error) || !ioc_ || !connected_.load()) {
+ + + + ]
246 : 104 : return wrapper::SendRejection::NotReady;
247 : : }
248 : 6147 : return std::nullopt;
249 : : }
250 : :
251 : : // Atomic rather than mutex-guarded: read both from the strand and from
252 : : // arbitrary caller threads (async_try_write_* fast-fail prechecks) (#436).
253 : : std::atomic<base::constants::BackpressureStrategy> bp_strategy_{base::constants::BackpressureStrategy::Reliable};
254 : : size_t bp_high_;
255 : : size_t bp_low_;
256 : : size_t bp_limit_;
257 : : std::atomic<bool> backpressure_active_{false};
258 : : diagnostics::RuntimeStatsCounters stats_;
259 : : unsigned first_retry_interval_ms_ = 100;
260 : :
261 : : // Shared snapshots rather than plain std::functions: the io thread copies
262 : : // one out per received chunk, and a std::function copy allocates whenever
263 : : // the target outgrows its small-object buffer. See interface::SharedCallback.
264 : : interface::SharedCallback<OnBytes> on_bytes_;
265 : : interface::SharedCallback<OnState> on_state_;
266 : : interface::SharedCallback<OnBackpressure> on_bp_;
267 : : mutable std::mutex callback_mtx_;
268 : : std::atomic<bool> connected_{false};
269 : 464 : AtomicLinkState state_{LinkState::Idle};
270 : : int retry_attempts_ = 0;
271 : : uint32_t reconnect_attempt_count_{0};
272 : : std::optional<ReconnectPolicy> reconnect_policy_;
273 : :
274 : : ErrorInfoHolder error_info_holder_{"tcp_client"};
275 : :
276 : 464 : Impl(const TcpClientConfig& cfg, net::io_context* ioc_ptr)
277 [ + + + - ]: 464 : : owned_ioc_(ioc_ptr ? nullptr : std::make_shared<net::io_context>()),
278 [ + + ]: 464 : ioc_(ioc_ptr ? ioc_ptr : owned_ioc_.get()),
279 : 464 : strand_(net::make_strand(*ioc_)),
280 : 464 : write_executor_(net::prefer(net::require(strand_, net::execution::blocking.never),
281 : : net::execution::relationship.fork,
282 : 464 : net::execution::allocator(std::allocator<void>{}))),
283 : 464 : resolver_(strand_),
284 : 464 : socket_(strand_),
285 : 464 : cfg_(cfg),
286 : 464 : retry_timer_(strand_),
287 : 464 : connect_timer_(strand_),
288 : 464 : idle_timer_(strand_),
289 : 464 : owns_ioc_(!ioc_ptr),
290 : 464 : bp_strategy_(cfg.backpressure_strategy),
291 : 3248 : bp_high_(cfg.backpressure_threshold) {
292 : 464 : init();
293 : 489 : }
294 : :
295 : 2210 : void mark_disconnected() {
296 : 2210 : std::lock_guard<std::mutex> lock(submission_mtx_);
297 : 2210 : mark_disconnected_locked();
298 : 2210 : }
299 : :
300 : : // Admission, write completion and explicit stop use the same terminal boundary.
301 : 2237 : void mark_disconnected_locked() {
302 [ + + ]: 2237 : if (connected_.exchange(false)) {
303 : 505 : send_accounting_.end(Ledger::Cause::ConnectionLoss);
304 [ + - ]: 505 : if (write_wait_) write_wait_->end(wrapper::SendRejection::NotReady);
305 : 505 : connection_seq_.fetch_add(1);
306 : : }
307 : 2237 : }
308 : :
309 : : void discard_connection_writes();
310 : :
311 : 464 : void init() {
312 : 464 : connected_ = false;
313 : 464 : writing_ = false;
314 : 464 : queue_bytes_ = 0;
315 : 464 : pending_bytes_ = 0;
316 : 464 : config::detail::validate(cfg_);
317 : 463 : rx_ = std::make_shared<std::vector<uint8_t>>(cfg_.read_buffer_size);
318 : 463 : recalculate_backpressure_bounds();
319 : 463 : first_retry_interval_ms_ = std::min(first_retry_interval_ms_, cfg_.retry_interval_ms);
320 : 463 : }
321 : :
322 : : void do_resolve_connect(std::shared_ptr<TcpClient> self, uint64_t seq);
323 : : void schedule_retry(std::shared_ptr<TcpClient> self, uint64_t seq);
324 : : void start_read(std::shared_ptr<TcpClient> self, uint64_t seq);
325 : : void do_write(std::shared_ptr<TcpClient> self, uint64_t seq);
326 : : void handle_close(std::shared_ptr<TcpClient> self, uint64_t seq, const boost::system::error_code& ec = {});
327 : : void handle_idle_timeout(std::shared_ptr<TcpClient> self, uint64_t seq);
328 : : void transition_to(LinkState next, const boost::system::error_code& ec = {});
329 : : void perform_stop_cleanup();
330 : : void reset_start_state();
331 : : void join_ioc_thread(bool allow_detach);
332 : : void close_socket();
333 : : void recalculate_backpressure_bounds();
334 : : void report_backpressure(std::shared_ptr<TcpClient> self, size_t queued_bytes);
335 : : void observe_queue();
336 : : // Shared decide_enqueue()/route dispatch used by all 3 async_write_* variants (#434).
337 : : // Every caller reserves added bytes in inflight_bytes_ before posting.
338 : : void route_enqueued_buffer(std::shared_ptr<TcpClient> self, TrackedBuffer&& buf, size_t added);
339 : : queue_util::BackpressureFields bp_fields();
340 : : void notify_state();
341 : : void reset_io_objects();
342 : : void apply_socket_options();
343 : : void handshake_then(std::shared_ptr<TcpClient> self, uint64_t seq, std::function<void()> next);
344 : : void finish_connect(std::shared_ptr<TcpClient> self, uint64_t seq);
345 : : void reset_idle_timer(std::shared_ptr<TcpClient> self, uint64_t seq);
346 : : void cancel_idle_timer();
347 : : void record_error(diagnostics::ErrorLevel lvl, diagnostics::ErrorCategory cat, std::string_view operation,
348 : : const boost::system::error_code& ec, std::string_view msg, bool retryable, uint32_t retry_count);
349 : : };
350 : :
351 : 148 : std::shared_ptr<TcpClient> TcpClient::create(const TcpClientConfig& cfg) {
352 : 148 : return std::shared_ptr<TcpClient>(new TcpClient(cfg));
353 : : }
354 : :
355 : 316 : std::shared_ptr<TcpClient> TcpClient::create(const TcpClientConfig& cfg, boost::asio::io_context& ioc) {
356 : 316 : return std::shared_ptr<TcpClient>(new TcpClient(cfg, ioc));
357 : : }
358 : :
359 : 149 : TcpClient::TcpClient(const TcpClientConfig& cfg) : impl_(std::make_unique<Impl>(cfg, nullptr)) {}
360 : 316 : TcpClient::TcpClient(const TcpClientConfig& cfg, boost::asio::io_context& ioc)
361 : 316 : : impl_(std::make_unique<Impl>(cfg, &ioc)) {}
362 : :
363 : 927 : TcpClient::~TcpClient() {
364 : : // #446: null after being moved-from - the move ctor/assignment are
365 : : // defaulted, and destroying a moved-from instance must not dereference
366 : : // a null impl_ (matches TcpServer/Serial/UdpChannel/UdsServer's
367 : : // destructors, which already guard this way).
368 [ + + ]: 464 : if (!impl_) return;
369 : 463 : stop();
370 : 463 : impl_->join_ioc_thread(true);
371 : :
372 : 463 : impl_->on_bytes_ = nullptr;
373 : 463 : impl_->on_state_ = nullptr;
374 : 463 : impl_->on_bp_ = nullptr;
375 : 929 : }
376 : :
377 : 1 : TcpClient::TcpClient(TcpClient&&) noexcept = default;
378 : 0 : TcpClient& TcpClient::operator=(TcpClient&&) noexcept = default;
379 : :
380 : 5 : std::optional<diagnostics::ErrorInfo> TcpClient::last_error_info() const {
381 : 5 : return impl_->error_info_holder_.last_error_info();
382 : : }
383 : :
384 : 700 : void TcpClient::start() {
385 : 700 : auto current_state = impl_->state_.get();
386 [ + - - + ]: 700 : if (current_state == LinkState::Connecting || current_state == LinkState::Connected) {
387 : 0 : WIRESTEAD_LOG_DEBUG("tcp_client", "start", "Start called while already active, ignoring");
388 : 0 : return;
389 : : }
390 : :
391 [ - + ]: 700 : if (!impl_->ioc_) {
392 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "start", "io_context is null");
393 : : }
394 : :
395 : 700 : impl_->recalculate_backpressure_bounds();
396 : :
397 [ + - + - : 700 : if (impl_->ioc_ && impl_->ioc_->stopped()) {
+ + + + ]
398 : 21 : WIRESTEAD_LOG_DEBUG("tcp_client", "start", "io_context stopped; restarting before start");
399 : 21 : impl_->ioc_->restart();
400 : : }
401 : :
402 [ - + ]: 700 : if (impl_->ioc_thread_.joinable()) {
403 : 0 : impl_->join_ioc_thread(false);
404 : : }
405 : :
406 : 700 : const auto seq = impl_->lifecycle_seq_.fetch_add(1) + 1;
407 : 700 : impl_->current_seq_.store(seq);
408 : 700 : impl_->reset_start_state();
409 : :
410 [ + + + - : 700 : if (impl_->owns_ioc_ && impl_->ioc_) {
+ + ]
411 : 163 : impl_->work_guard_ =
412 : 326 : std::make_unique<net::executor_work_guard<net::io_context::executor_type>>(impl_->ioc_->get_executor());
413 : 326 : impl_->ioc_thread_ = std::jthread([ioc = impl_->owned_ioc_](std::stop_token st) {
414 : 163 : wirestead::concurrency::run_io_thread_init();
415 : : try {
416 : 163 : std::stop_callback cb(st, [ioc] { ioc->stop(); });
417 : 163 : ioc->run();
418 : 163 : } catch (const std::exception& e) {
419 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "io_context", fmt::format("IO context error: {}", e.what()));
420 : 0 : diagnostics::error_reporting::report_system_error("tcp_client", "io_context",
421 : 0 : fmt::format("Exception in IO context: {}", e.what()));
422 : 0 : }
423 : 326 : });
424 : : }
425 : :
426 : 700 : auto weak_self = weak_from_this();
427 [ + - ]: 700 : if (impl_->ioc_) {
428 : 700 : net::dispatch(impl_->strand_, [weak_self, seq] {
429 [ + - ]: 700 : if (auto self = weak_self.lock()) {
430 [ + + ]: 1400 : if (seq <= self->impl_->stop_seq_.load()) {
431 : 205 : return;
432 : : }
433 : 495 : self->impl_->mark_disconnected();
434 : 495 : self->impl_->reset_io_objects();
435 : 495 : self->impl_->transition_to(LinkState::Connecting);
436 : 495 : self->impl_->do_resolve_connect(self, seq);
437 : 700 : }
438 : : });
439 : : } else {
440 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "start", "io_context is null");
441 : : }
442 : 700 : }
443 : :
444 : 1524 : void TcpClient::stop() {
445 : 1524 : const bool on_executor = impl_->ioc_->get_executor().running_in_this_thread();
446 : : bool first;
447 : : {
448 : 1524 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
449 : 1524 : first = !impl_->stop_requested_.exchange(true);
450 [ + + ]: 1524 : if (first) {
451 : 722 : impl_->send_accounting_.end(Impl::Ledger::Cause::ExplicitStop);
452 [ + + ]: 722 : if (impl_->write_wait_) impl_->write_wait_->end(wrapper::SendRejection::CancelledWhileWaiting);
453 : 722 : impl_->stopping_.store(true);
454 : 1444 : impl_->stop_seq_.store(impl_->current_seq_.load());
455 : : }
456 : 1524 : }
457 [ + + ]: 1524 : if (first) {
458 : : // Only a never-started transport has no executor work to serialize with.
459 [ + + ]: 1444 : if (impl_->current_seq_.load() == 0) {
460 : 23 : impl_->perform_stop_cleanup();
461 : : } else {
462 : : // A waiting destructor keeps Impl alive until this handler completes.
463 : : // Ordinary requests retain the transport across request-only returns.
464 : 699 : auto self = weak_from_this().lock();
465 : 699 : auto* impl = impl_.get();
466 : 1398 : net::post(impl_->strand_, [self, impl] { impl->perform_stop_cleanup(); });
467 : 699 : }
468 : : }
469 [ + + ]: 1524 : if (on_executor) return;
470 : :
471 : : // External executors must keep progressing, as required by D-1. Never
472 : : // poll them here or infer completion from stopped()/elapsed time.
473 : 1459 : impl_->wait_for_cleanup();
474 : 1459 : std::lock_guard<std::mutex> join_lock(impl_->join_mtx_);
475 : 1459 : impl_->join_ioc_thread(false);
476 : 1459 : }
477 : :
478 : 3672 : bool TcpClient::is_connected() const { return get_impl()->connected_.load(); }
479 : 1989 : bool TcpClient::is_backpressure_active() const { return get_impl()->backpressure_active_.load(); }
480 : 4122 : std::optional<size_t> TcpClient::write_queue_limit() const { return impl_->bp_limit_; }
481 : 852 : wrapper::RuntimeStats TcpClient::stats() const {
482 : 3408 : auto result = impl_->stats_.snapshot(impl_->queue_bytes_.load(std::memory_order_relaxed),
483 : 852 : impl_->pending_bytes_.load(std::memory_order_relaxed),
484 : 852 : impl_->backpressure_active_.load(std::memory_order_relaxed));
485 : 852 : result.send_accounting = impl_->send_accounting_.snapshot();
486 : 852 : return result;
487 : : }
488 : 15 : void TcpClient::reset_stats() {
489 : 15 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
490 : 15 : impl_->send_accounting_.reset();
491 : 15 : impl_->stats_.reset(impl_->queue_bytes_.load(std::memory_order_relaxed) +
492 : 30 : impl_->pending_bytes_.load(std::memory_order_relaxed));
493 : 15 : }
494 : :
495 : 4 : void TcpClient::fail_receive() {
496 : 4 : const auto connection = write_connection();
497 [ - + ]: 4 : if (!connection) return;
498 : 4 : const auto run = impl_->current_seq_.load();
499 : 4 : net::dispatch(impl_->strand_, [self = shared_from_this(), connection, run] {
500 [ + - + - : 8 : if (run != self->impl_->current_seq_.load() || self->write_connection() != connection) return;
- + - + ]
501 : 4 : self->impl_->handle_close(self, run, make_error_code(boost::system::errc::no_buffer_space));
502 : : });
503 : : }
504 : :
505 : 10792 : boost::asio::any_io_executor TcpClient::get_executor() { return impl_->strand_; }
506 : :
507 : 43 : std::shared_ptr<detail::TcpWriteWait> TcpClient::capture_write_wait() const {
508 : 43 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
509 [ + - - + : 43 : if (impl_->stop_requested_.load() || !impl_->connected_.load()) return {};
- + ]
510 : 43 : return impl_->write_wait_;
511 : 43 : }
512 : :
513 : 36 : std::optional<wrapper::SendResult> TcpClient::poll_write_wait(const std::shared_ptr<detail::TcpWriteWait>& wait) const {
514 : 36 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
515 [ - + ]: 36 : if (!wait) return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
516 [ + + ]: 36 : if (wait->ended_by) return wrapper::SendResult::reject(*wait->ended_by);
517 [ + - - + : 24 : if (wait->sequence != impl_->connection_seq_.load() || !impl_->connected_.load())
- + ]
518 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
519 [ + - ]: 12 : if (!impl_->backpressure_active_.load()) return wrapper::SendResult::accept();
520 : 0 : return std::nullopt;
521 : 36 : }
522 : :
523 : 451 : void TcpClient::cancel_write_waits() {
524 : 451 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
525 [ + + ]: 451 : if (impl_->write_wait_) impl_->write_wait_->end(wrapper::SendRejection::CancelledWhileWaiting);
526 : 451 : }
527 : :
528 : 1888 : std::optional<uint64_t> TcpClient::write_connection() const {
529 : 1888 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
530 [ + - - + : 1888 : if (impl_->stop_requested_.load() || !impl_->connected_.load()) return std::nullopt;
- + ]
531 : 3776 : return impl_->connection_seq_.load();
532 : 1888 : }
533 : :
534 : 1873 : wrapper::SendResult TcpClient::write_state() {
535 : 1873 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
536 [ + - + + ]: 1873 : if (auto reason = impl_->admission_rejection()) return wrapper::SendResult::reject(*reason);
537 : 1827 : return wrapper::SendResult::accept();
538 : 1873 : }
539 : :
540 : 152 : wrapper::SendResult TcpClient::async_write_copy_result(memory::ConstByteSpan data) {
541 : 152 : const auto result = write_copy(data, std::nullopt);
542 [ + + + - ]: 152 : if (auto hook = detail::g_tcp_write_result_hook.load()) hook(result);
543 : 152 : return result;
544 : : }
545 : :
546 : 1422 : wrapper::SendResult TcpClient::write_copy(memory::ConstByteSpan data, std::optional<uint64_t> expected_connection) {
547 [ + + ]: 1422 : if (expected_connection) {
548 [ + + + - ]: 1270 : if (auto hook = detail::g_tcp_pinned_write_hook.load()) hook();
549 : : }
550 : 1422 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
551 : 1422 : const auto seq = impl_->current_seq_.load();
552 : 1422 : const auto connection = impl_->connection_seq_.load();
553 [ + - + + ]: 1422 : if (auto reason = impl_->admission_rejection(expected_connection)) {
554 : 49 : impl_->stats_.record_failed_send();
555 : 49 : return wrapper::SendResult::reject(*reason);
556 : : }
557 [ + + + - ]: 1373 : if (auto hook = detail::g_tcp_write_admission_hook.load()) hook();
558 : :
559 : 1373 : size_t size = data.size();
560 [ + + ]: 1373 : if (size == 0) {
561 : 7 : WIRESTEAD_LOG_WARNING("tcp_client", "async_write_copy", "Ignoring zero-length write");
562 : 7 : impl_->stats_.record_failed_send();
563 : 7 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
564 : : }
565 : :
566 [ + + ]: 1366 : if (size > base::constants::MAX_BUFFER_SIZE) {
567 : 5 : WIRESTEAD_LOG_ERROR("tcp_client", "async_write_copy",
568 : : fmt::format("Write size exceeds maximum allowed ({} bytes)", size));
569 : 5 : impl_->stats_.record_failed_send();
570 : 5 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
571 : : }
572 : :
573 [ + + + + : 1361 : if (size <= 65536 && impl_->cfg_.enable_memory_pool) {
+ + ]
574 : : try {
575 : 1319 : memory::PooledBuffer pooled_buffer(size, impl_->pool_);
576 [ + - + - ]: 1319 : if (pooled_buffer.valid()) {
577 : 1319 : base::safe_memory::safe_memcpy(pooled_buffer.data(), data.data(), size);
578 : 1319 : const auto added = pooled_buffer.size();
579 : :
580 [ + - + + ]: 1319 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
581 : 1319 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
582 : 3 : impl_->stats_.record_failed_send();
583 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
584 : : }
585 : 1316 : impl_->stats_.record_accepted(added);
586 : 1316 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
587 : 1316 : const auto request = admission.request();
588 : 1316 : impl_->write_executor_.execute([self = shared_from_this(), buf = std::move(pooled_buffer), added, seq,
589 : : connection, request]() mutable {
590 [ - + ]: 2632 : if (seq != self->impl_->current_seq_.load()) return;
591 [ + + ]: 2632 : if (connection != self->impl_->connection_seq_.load()) {
592 : 1 : self->impl_->stats_.record_dropped(1, added);
593 : 1 : queue_util::release_reserved_limit_bytes(self->impl_->write_reserve_mtx_, self->impl_->inflight_bytes_,
594 : : added);
595 : 1 : return;
596 : : }
597 : 1315 : self->impl_->route_enqueued_buffer(self, Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
598 : : });
599 : 1316 : admission.commit();
600 : 1316 : return wrapper::SendResult::accept();
601 : 1316 : }
602 : 1319 : } catch (const std::exception& e) {
603 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "async_write_copy",
604 : : fmt::format("Failed to acquire pooled buffer: {}", e.what()));
605 : 0 : }
606 : : }
607 : :
608 : 42 : std::vector<uint8_t> fallback(data.begin(), data.end());
609 : 42 : const auto added = fallback.size();
610 : :
611 [ + - + + ]: 42 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
612 : 42 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
613 : 2 : impl_->stats_.record_failed_send();
614 : 2 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
615 : : }
616 : 40 : impl_->stats_.record_accepted(added);
617 : 40 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
618 : 40 : const auto request = admission.request();
619 : :
620 : 40 : impl_->write_executor_.execute([self = shared_from_this(), buf = std::move(fallback), added, seq, connection,
621 : : request]() mutable {
622 [ - + ]: 80 : if (seq != self->impl_->current_seq_.load()) return;
623 [ + + ]: 80 : if (connection != self->impl_->connection_seq_.load()) {
624 : 1 : self->impl_->stats_.record_dropped(1, added);
625 : 1 : queue_util::release_reserved_limit_bytes(self->impl_->write_reserve_mtx_, self->impl_->inflight_bytes_, added);
626 : 1 : return;
627 : : }
628 : 39 : self->impl_->route_enqueued_buffer(self, Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
629 : : });
630 : 40 : admission.commit();
631 : 40 : return wrapper::SendResult::accept();
632 : 1422 : }
633 : :
634 : 197 : wrapper::SendResult TcpClient::async_write_move_result(std::vector<uint8_t>&& data) {
635 : 197 : const auto result = write_move(std::move(data), std::nullopt);
636 [ + + + - ]: 197 : if (auto hook = detail::g_tcp_write_result_hook.load()) hook(result);
637 : 197 : return result;
638 : : }
639 : :
640 : 725 : wrapper::SendResult TcpClient::write_move(std::vector<uint8_t>&& data, std::optional<uint64_t> expected_connection) {
641 [ + + ]: 725 : if (expected_connection) {
642 [ + + + - ]: 528 : if (auto hook = detail::g_tcp_pinned_write_hook.load()) hook();
643 : : }
644 : 725 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
645 : 725 : const auto seq = impl_->current_seq_.load();
646 : 725 : const auto connection = impl_->connection_seq_.load();
647 [ + - + + ]: 725 : if (auto reason = impl_->admission_rejection(expected_connection)) {
648 : 47 : impl_->stats_.record_failed_send();
649 : 47 : return wrapper::SendResult::reject(*reason);
650 : : }
651 [ + + + - ]: 678 : if (auto hook = detail::g_tcp_write_admission_hook.load()) hook();
652 : 678 : const auto size = data.size();
653 [ + + ]: 678 : if (size == 0) {
654 : 6 : WIRESTEAD_LOG_WARNING("tcp_client", "async_write_move", "Ignoring zero-length write");
655 : 6 : impl_->stats_.record_failed_send();
656 : 6 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
657 : : }
658 [ + + ]: 672 : if (size > base::constants::MAX_BUFFER_SIZE) {
659 : 5 : WIRESTEAD_LOG_ERROR("tcp_client", "async_write_move",
660 : : fmt::format("Write size exceeds maximum allowed ({} bytes)", size));
661 : 5 : impl_->stats_.record_failed_send();
662 : 5 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
663 : : }
664 : :
665 : 667 : const auto added = size;
666 : :
667 [ + - + + ]: 667 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
668 : 667 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
669 : 17 : impl_->stats_.record_failed_send();
670 : 17 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
671 : : }
672 : 650 : impl_->stats_.record_accepted(added);
673 : 650 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
674 : 650 : const auto request = admission.request();
675 : 650 : impl_->write_executor_.execute([self = shared_from_this(), buf = std::move(data), added, seq, connection,
676 : : request]() mutable {
677 [ - + ]: 1300 : if (seq != self->impl_->current_seq_.load()) return;
678 [ + + ]: 1300 : if (connection != self->impl_->connection_seq_.load()) {
679 : 1 : self->impl_->stats_.record_dropped(1, added);
680 : 1 : queue_util::release_reserved_limit_bytes(self->impl_->write_reserve_mtx_, self->impl_->inflight_bytes_, added);
681 : 1 : return;
682 : : }
683 : 649 : self->impl_->route_enqueued_buffer(self, Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
684 : : });
685 : 650 : admission.commit();
686 : 650 : return wrapper::SendResult::accept();
687 : 725 : }
688 : :
689 : 102 : wrapper::SendResult TcpClient::async_write_shared_result(std::shared_ptr<const std::vector<uint8_t>> data) {
690 : 102 : const auto result = write_shared(std::move(data), std::nullopt);
691 [ + + + - ]: 102 : if (auto hook = detail::g_tcp_write_result_hook.load()) hook(result);
692 : 102 : return result;
693 : : }
694 : :
695 : 106 : wrapper::SendResult TcpClient::write_shared(std::shared_ptr<const std::vector<uint8_t>> data,
696 : : std::optional<uint64_t> expected_connection) {
697 [ + + ]: 106 : if (expected_connection) {
698 [ + + + - ]: 4 : if (auto hook = detail::g_tcp_pinned_write_hook.load()) hook();
699 : : }
700 : 106 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
701 : 106 : const auto seq = impl_->current_seq_.load();
702 : 106 : const auto connection = impl_->connection_seq_.load();
703 [ + - + + ]: 106 : if (auto reason = impl_->admission_rejection(expected_connection)) {
704 : 41 : impl_->stats_.record_failed_send();
705 : 41 : return wrapper::SendResult::reject(*reason);
706 : : }
707 [ + + + - ]: 65 : if (auto hook = detail::g_tcp_write_admission_hook.load()) hook();
708 [ + + + + : 65 : if (!data || data->empty()) {
+ + ]
709 : 7 : WIRESTEAD_LOG_WARNING("tcp_client", "async_write_shared", "Ignoring empty shared buffer");
710 : 7 : impl_->stats_.record_failed_send();
711 : 7 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
712 : : }
713 : 58 : const auto size = data->size();
714 [ + + ]: 58 : if (size > base::constants::MAX_BUFFER_SIZE) {
715 : 5 : WIRESTEAD_LOG_ERROR("tcp_client", "async_write_shared",
716 : : fmt::format("Write size exceeds maximum allowed ({} bytes)", size));
717 : 5 : impl_->stats_.record_failed_send();
718 : 5 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
719 : : }
720 : :
721 : 53 : const auto added = size;
722 : :
723 [ + - + + ]: 53 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
724 : 53 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
725 : 5 : impl_->stats_.record_failed_send();
726 : 5 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
727 : : }
728 : 48 : impl_->stats_.record_accepted(added);
729 : 48 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
730 : 48 : const auto request = admission.request();
731 : 48 : impl_->write_executor_.execute([self = shared_from_this(), buf = std::move(data), added, seq, connection,
732 : : request]() mutable {
733 [ - + ]: 96 : if (seq != self->impl_->current_seq_.load()) return;
734 [ + + ]: 96 : if (connection != self->impl_->connection_seq_.load()) {
735 : 1 : self->impl_->stats_.record_dropped(1, added);
736 : 1 : queue_util::release_reserved_limit_bytes(self->impl_->write_reserve_mtx_, self->impl_->inflight_bytes_, added);
737 : 1 : return;
738 : : }
739 : 47 : self->impl_->route_enqueued_buffer(self, Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
740 : : });
741 : 48 : admission.commit();
742 : 48 : return wrapper::SendResult::accept();
743 : 106 : }
744 : :
745 : 97 : wrapper::SendResult TcpClient::async_try_write_copy_result(memory::ConstByteSpan data) {
746 : 97 : const auto result = try_write_copy(data);
747 [ + + + - ]: 97 : if (auto hook = detail::g_tcp_write_result_hook.load()) hook(result);
748 : 97 : return result;
749 : : }
750 : :
751 : 2134 : wrapper::SendResult TcpClient::try_write_copy(memory::ConstByteSpan data) {
752 [ + + ]: 2134 : if (data.empty()) {
753 : 5 : impl_->stats_.record_failed_send();
754 : 5 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
755 : : }
756 [ + + ]: 2129 : if (data.size() > base::constants::MAX_BUFFER_SIZE) {
757 : 4 : impl_->stats_.record_failed_send();
758 : 4 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
759 : : }
760 : 4250 : return try_write_move(std::vector<uint8_t>(data.begin(), data.end()));
761 : : }
762 : :
763 : 108 : wrapper::SendResult TcpClient::async_try_write_move_result(std::vector<uint8_t>&& data) {
764 : 108 : const auto result = try_write_move(std::move(data));
765 [ + + + - ]: 108 : if (auto hook = detail::g_tcp_write_result_hook.load()) hook(result);
766 : 108 : return result;
767 : : }
768 : :
769 : 2245 : wrapper::SendResult TcpClient::try_write_move(std::vector<uint8_t>&& data) {
770 : 2245 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
771 : 2245 : const auto seq = impl_->current_seq_.load();
772 : 2245 : const auto connection = impl_->connection_seq_.load();
773 [ + - + + ]: 2245 : if (auto reason = impl_->admission_rejection()) {
774 : 96 : impl_->stats_.record_failed_send();
775 : 96 : return wrapper::SendResult::reject(*reason);
776 : : }
777 [ + + + - ]: 2149 : if (auto hook = detail::g_tcp_write_admission_hook.load()) hook();
778 : 2149 : const auto added = data.size();
779 [ + + + + ]: 2149 : if (added == 0 || added > base::constants::MAX_BUFFER_SIZE) {
780 : 9 : impl_->stats_.record_failed_send();
781 [ + + ]: 9 : return wrapper::SendResult::reject(added == 0 ? wrapper::SendRejection::InvalidArgument
782 : 9 : : wrapper::SendRejection::TooLarge);
783 : : }
784 : 27 : const auto reject_for_pressure = [this, added]() {
785 [ + + ]: 27 : if (impl_->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
786 : 18 : impl_->stats_.record_dropped(1, added);
787 : : } else {
788 : 9 : impl_->stats_.record_failed_send();
789 : : }
790 : 2167 : };
791 [ + + + + : 4253 : if (impl_->backpressure_active_.load() || impl_->queue_bytes_ + added > impl_->bp_high_ ||
+ + ]
792 [ - + ]: 2113 : impl_->queue_bytes_ + impl_->pending_bytes_ + added > impl_->bp_limit_) {
793 : 27 : reject_for_pressure();
794 : 27 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
795 : : }
796 [ + - - + ]: 2113 : if (!queue_util::try_reserve_write_bytes(impl_->write_reserve_mtx_, impl_->inflight_bytes_, impl_->queue_bytes_,
797 : 2113 : impl_->pending_bytes_, impl_->backpressure_active_, added, impl_->bp_high_,
798 : 2113 : impl_->bp_limit_)) {
799 : 0 : reject_for_pressure();
800 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
801 : : }
802 : 2113 : impl_->stats_.record_accepted(added);
803 : 2113 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
804 : 2113 : const auto request = admission.request();
805 : :
806 : 2113 : impl_->write_executor_.execute(
807 : 2113 : [self = shared_from_this(), buf = std::move(data), added, seq, connection, request]() mutable {
808 [ - + ]: 4226 : if (seq != self->impl_->current_seq_.load()) return;
809 [ + + ]: 4226 : if (connection != self->impl_->connection_seq_.load()) {
810 : 2 : self->impl_->stats_.record_dropped(1, added);
811 : : // The old connection drain already removed this queue reservation.
812 : 2 : return;
813 : : }
814 : 2111 : auto impl = self->impl_.get();
815 [ + + + - : 4185 : if (impl->stop_requested_.load() || impl->state_.is_state(LinkState::Closed) ||
- + ]
816 : 4185 : impl->state_.is_state(LinkState::Error)) {
817 : 37 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, added);
818 : 37 : impl->stats_.record_failed_send();
819 : 37 : return;
820 : : }
821 : :
822 : 2074 : impl->tx_.push_back(Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request});
823 : 2074 : impl->observe_queue();
824 : 2074 : impl->report_backpressure(self, impl->queue_bytes_);
825 [ + + + - ]: 2118 : if (!impl->writing_) impl->do_write(self, impl->current_seq_.load());
826 : : });
827 : 2113 : admission.commit();
828 : 2113 : return wrapper::SendResult::accept();
829 : 2245 : }
830 : :
831 : 97 : wrapper::SendResult TcpClient::async_try_write_shared_result(std::shared_ptr<const std::vector<uint8_t>> data) {
832 : 97 : const auto result = try_write_shared(std::move(data));
833 [ + + + - ]: 97 : if (auto hook = detail::g_tcp_write_result_hook.load()) hook(result);
834 : 97 : return result;
835 : : }
836 : :
837 : 109 : wrapper::SendResult TcpClient::try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
838 : 109 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
839 : 109 : const auto seq = impl_->current_seq_.load();
840 : 109 : const auto connection = impl_->connection_seq_.load();
841 [ + - + + : 109 : if (!data || data->empty()) {
+ + ]
842 : 5 : impl_->stats_.record_failed_send();
843 : 5 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
844 : : }
845 [ + + ]: 104 : if (data->size() > base::constants::MAX_BUFFER_SIZE) {
846 : 4 : impl_->stats_.record_failed_send();
847 : 4 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
848 : : }
849 : 100 : const auto added = data->size();
850 : 8 : const auto reject_for_pressure = [this, added]() {
851 [ + + ]: 8 : if (impl_->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
852 : 4 : impl_->stats_.record_dropped(1, added);
853 : : } else {
854 : 4 : impl_->stats_.record_failed_send();
855 : : }
856 : 108 : };
857 [ + - + + ]: 100 : if (auto reason = impl_->admission_rejection()) {
858 : 45 : impl_->stats_.record_failed_send();
859 : 45 : return wrapper::SendResult::reject(*reason);
860 : : }
861 [ + + + - ]: 55 : if (auto hook = detail::g_tcp_write_admission_hook.load()) hook();
862 [ + - + + : 102 : if (impl_->backpressure_active_.load() || impl_->queue_bytes_ + added > impl_->bp_high_ ||
+ + ]
863 [ - + ]: 47 : impl_->queue_bytes_ + impl_->pending_bytes_ + added > impl_->bp_limit_) {
864 : 8 : reject_for_pressure();
865 : 8 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
866 : : }
867 [ + - - + ]: 47 : if (!queue_util::try_reserve_write_bytes(impl_->write_reserve_mtx_, impl_->inflight_bytes_, impl_->queue_bytes_,
868 : 47 : impl_->pending_bytes_, impl_->backpressure_active_, added, impl_->bp_high_,
869 : 47 : impl_->bp_limit_)) {
870 : 0 : reject_for_pressure();
871 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
872 : : }
873 : 47 : impl_->stats_.record_accepted(added);
874 : 47 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
875 : 47 : const auto request = admission.request();
876 : :
877 : 47 : impl_->write_executor_.execute(
878 : 47 : [self = shared_from_this(), buf = std::move(data), added, seq, connection, request]() mutable {
879 [ - + ]: 94 : if (seq != self->impl_->current_seq_.load()) return;
880 [ + + ]: 94 : if (connection != self->impl_->connection_seq_.load()) {
881 : 1 : self->impl_->stats_.record_dropped(1, added);
882 : : // The old connection drain already removed this queue reservation.
883 : 1 : return;
884 : : }
885 : 46 : auto impl = self->impl_.get();
886 [ + + + - : 76 : if (impl->stop_requested_.load() || impl->state_.is_state(LinkState::Closed) ||
- + ]
887 : 76 : impl->state_.is_state(LinkState::Error)) {
888 : 16 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, added);
889 : 16 : impl->stats_.record_failed_send();
890 : 16 : return;
891 : : }
892 : :
893 : 30 : impl->tx_.push_back(Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request});
894 : 30 : impl->observe_queue();
895 : 30 : impl->report_backpressure(self, impl->queue_bytes_);
896 [ + + + - ]: 45 : if (!impl->writing_) impl->do_write(self, impl->current_seq_.load());
897 : : });
898 : 47 : admission.commit();
899 : 47 : return wrapper::SendResult::accept();
900 : 109 : }
901 : :
902 : : // Each setter builds the shared snapshot before taking the lock, so the
903 : : // allocation stays outside the critical section the io thread contends on.
904 : 783 : void TcpClient::on_bytes(OnBytes cb) {
905 : 783 : auto shared = interface::share_callback(std::move(cb));
906 : 783 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
907 : 783 : impl_->on_bytes_ = std::move(shared);
908 : 783 : }
909 : 827 : void TcpClient::on_state(OnState cb) {
910 : 827 : auto shared = interface::share_callback(std::move(cb));
911 : 827 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
912 : 827 : impl_->on_state_ = std::move(shared);
913 : 827 : }
914 : 770 : void TcpClient::on_backpressure(OnBackpressure cb) {
915 : 770 : auto shared = interface::share_callback(std::move(cb));
916 : 770 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
917 : 770 : impl_->on_bp_ = std::move(shared);
918 : 770 : }
919 : 25 : void TcpClient::set_backpressure_strategy(base::constants::BackpressureStrategy strategy) {
920 : 25 : config::detail::strategy(strategy);
921 : 25 : impl_->bp_strategy_.store(strategy, std::memory_order_relaxed);
922 : 25 : }
923 : :
924 : 3 : void TcpClient::set_retry_interval(unsigned interval_ms) {
925 : 3 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
926 : 3 : auto candidate = impl_->cfg_;
927 : 3 : candidate.retry_interval_ms = interval_ms;
928 : 3 : config::detail::validate(candidate);
929 : 2 : impl_->cfg_.retry_interval_ms = interval_ms;
930 : 4 : }
931 : 2 : void TcpClient::set_max_retries(int max_retries) {
932 : 2 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
933 : 2 : auto candidate = impl_->cfg_;
934 : 2 : candidate.max_retries = max_retries;
935 : 2 : config::detail::validate(candidate);
936 : 2 : impl_->cfg_.max_retries = max_retries;
937 : 2 : }
938 : 2 : void TcpClient::set_connection_timeout(unsigned timeout_ms) {
939 : 2 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
940 : 2 : auto candidate = impl_->cfg_;
941 : 2 : candidate.connection_timeout_ms = timeout_ms;
942 : 2 : config::detail::validate(candidate);
943 : 2 : impl_->cfg_.connection_timeout_ms = timeout_ms;
944 : 2 : }
945 : 1 : void TcpClient::set_idle_timeout(unsigned timeout_ms) {
946 : 1 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
947 : 1 : auto candidate = impl_->cfg_;
948 : 1 : candidate.idle_timeout_ms = timeout_ms;
949 : 1 : config::detail::validate(candidate);
950 : 1 : impl_->cfg_.idle_timeout_ms = timeout_ms;
951 : 1 : }
952 : 1 : void TcpClient::set_idle_timeout_action(IdleTimeoutAction action) {
953 [ - + - - : 1 : config::detail::require(action == IdleTimeoutAction::Close || action == IdleTimeoutAction::Reconnect,
+ - ]
954 : : "invalid idle timeout action");
955 : 1 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
956 : 1 : impl_->cfg_.idle_timeout_action = action;
957 : 1 : }
958 : 8 : void TcpClient::set_reconnect_policy(ReconnectPolicy policy) {
959 : 8 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
960 [ + + ]: 8 : if (policy) {
961 : 7 : impl_->reconnect_policy_ = std::move(policy);
962 : : } else {
963 : 1 : impl_->reconnect_policy_ = std::nullopt;
964 : : }
965 : 8 : }
966 : :
967 : : // Impl methods implementation
968 : :
969 : 534 : void TcpClient::Impl::apply_socket_options() {
970 : 534 : boost::system::error_code ec;
971 : :
972 [ + - ]: 534 : if (cfg_.tcp_no_delay) {
973 : 534 : socket_.set_option(tcp::no_delay(true), ec);
974 [ - + ]: 534 : if (ec) {
975 : 0 : WIRESTEAD_LOG_WARNING("tcp_client", "socket_options", fmt::format("Failed to set TCP_NODELAY: {}", ec.message()));
976 : 0 : ec.clear();
977 : : }
978 : : }
979 : :
980 [ - + ]: 534 : if (cfg_.keep_alive) {
981 : 0 : socket_.set_option(net::socket_base::keep_alive(true), ec);
982 [ # # ]: 0 : if (ec) {
983 : 0 : WIRESTEAD_LOG_WARNING("tcp_client", "socket_options", fmt::format("Failed to set keep_alive: {}", ec.message()));
984 : 0 : ec.clear();
985 : : }
986 : : }
987 : :
988 [ + + ]: 534 : if (cfg_.send_buffer_size > 0) {
989 : 192 : socket_.set_option(net::socket_base::send_buffer_size(static_cast<int>(cfg_.send_buffer_size)), ec);
990 [ - + ]: 192 : if (ec) {
991 : 0 : WIRESTEAD_LOG_WARNING("tcp_client", "socket_options",
992 : : fmt::format("Failed to set send buffer size: {}", ec.message()));
993 : 0 : ec.clear();
994 : : }
995 : : }
996 : :
997 [ - + ]: 534 : if (cfg_.receive_buffer_size > 0) {
998 : 0 : socket_.set_option(net::socket_base::receive_buffer_size(static_cast<int>(cfg_.receive_buffer_size)), ec);
999 [ # # ]: 0 : if (ec) {
1000 : 0 : WIRESTEAD_LOG_WARNING("tcp_client", "socket_options",
1001 : : fmt::format("Failed to set receive buffer size: {}", ec.message()));
1002 : 0 : ec.clear();
1003 : : }
1004 : : }
1005 : 534 : }
1006 : :
1007 : 598 : void TcpClient::Impl::do_resolve_connect(std::shared_ptr<TcpClient> self, uint64_t seq) {
1008 : 598 : ++pending_io_;
1009 : 598 : resolver_.async_resolve(
1010 : 1196 : cfg_.host, fmt::format("{}", cfg_.port), [self, seq](auto ec, tcp::resolver::results_type results) {
1011 : 598 : IoCompletion completion{self->impl_.get()};
1012 [ + + - + : 1195 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) {
+ + ]
1013 : 1 : return;
1014 : : }
1015 [ + + - + : 597 : if (self->impl_->stop_requested_.load() || self->impl_->stopping_.load()) {
+ + ]
1016 : 4 : return;
1017 : : }
1018 [ + + ]: 593 : if (ec) {
1019 : : bool has_policy;
1020 : : {
1021 : 1 : std::lock_guard<std::mutex> lock(self->impl_->cfg_mtx_);
1022 : 1 : has_policy = self->impl_->reconnect_policy_.has_value();
1023 : 1 : }
1024 : 1 : uint32_t current_attempts =
1025 [ - + ]: 1 : has_policy ? self->impl_->reconnect_attempt_count_ : static_cast<uint32_t>(self->impl_->retry_attempts_);
1026 : 2 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "resolve",
1027 : : ec, fmt::format("Resolution failed: {}", ec.message()),
1028 : 1 : diagnostics::is_retryable_tcp_connect_error(ec), current_attempts);
1029 : 1 : self->impl_->schedule_retry(self, seq);
1030 : 1 : return;
1031 : : }
1032 : : unsigned connection_timeout_ms;
1033 : : {
1034 : 592 : std::lock_guard<std::mutex> lock(self->impl_->cfg_mtx_);
1035 : 592 : connection_timeout_ms = self->impl_->cfg_.connection_timeout_ms;
1036 : 592 : }
1037 : 592 : self->impl_->connect_timer_.expires_after(std::chrono::milliseconds(connection_timeout_ms));
1038 : 592 : ++self->impl_->pending_io_;
1039 : 592 : self->impl_->connect_timer_.async_wait([self, seq](const boost::system::error_code& timer_ec) {
1040 : 592 : IoCompletion completion{self->impl_.get()};
1041 [ + + - + : 594 : if (timer_ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) {
+ + ]
1042 : 590 : return;
1043 : : }
1044 [ + - + - : 2 : if (!timer_ec && !self->impl_->stop_requested_.load() && !self->impl_->stopping_.load()) {
+ - + - ]
1045 : : bool has_policy;
1046 : : unsigned timeout_ms;
1047 : : {
1048 : 2 : std::lock_guard<std::mutex> lock(self->impl_->cfg_mtx_);
1049 : 2 : has_policy = self->impl_->reconnect_policy_.has_value();
1050 : 2 : timeout_ms = self->impl_->cfg_.connection_timeout_ms;
1051 : 2 : }
1052 : 2 : WIRESTEAD_LOG_ERROR("tcp_client", "connect_timeout",
1053 : : fmt::format("Connection timed out after {}ms", timeout_ms));
1054 [ - + ]: 2 : uint32_t current_attempts = has_policy ? self->impl_->reconnect_attempt_count_
1055 : 2 : : static_cast<uint32_t>(self->impl_->retry_attempts_);
1056 : 4 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "connect",
1057 : : boost::asio::error::timed_out, "Connection timed out",
1058 : 2 : diagnostics::is_retryable_tcp_connect_error(boost::asio::error::timed_out),
1059 : : current_attempts);
1060 : 2 : self->impl_->handle_close(self, seq, boost::asio::error::timed_out);
1061 : : }
1062 : 592 : });
1063 : :
1064 : 592 : ++self->impl_->pending_io_;
1065 : 592 : net::async_connect(self->impl_->socket_, results, [self, seq](auto ec2, const auto&) {
1066 : 592 : IoCompletion completion{self->impl_.get()};
1067 [ + + - + : 1180 : if (ec2 == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) {
+ + ]
1068 : 4 : return;
1069 : : }
1070 [ + - - + : 588 : if (self->impl_->stop_requested_.load() || self->impl_->stopping_.load()) {
- + ]
1071 : 0 : self->impl_->close_socket();
1072 : 0 : self->impl_->connect_timer_.cancel();
1073 : 0 : return;
1074 : : }
1075 [ + + ]: 588 : if (ec2) {
1076 : 54 : self->impl_->connect_timer_.cancel();
1077 : : bool has_policy;
1078 : : {
1079 : 54 : std::lock_guard<std::mutex> lock(self->impl_->cfg_mtx_);
1080 : 54 : has_policy = self->impl_->reconnect_policy_.has_value();
1081 : 54 : }
1082 [ + + ]: 54 : uint32_t current_attempts = has_policy ? self->impl_->reconnect_attempt_count_
1083 : 27 : : static_cast<uint32_t>(self->impl_->retry_attempts_);
1084 : 108 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "connect",
1085 : : ec2, fmt::format("Connection failed: {}", ec2.message()),
1086 : 54 : diagnostics::is_retryable_tcp_connect_error(ec2), current_attempts);
1087 : 54 : self->impl_->schedule_retry(self, seq);
1088 : 54 : return;
1089 : : }
1090 : 534 : self->impl_->connect_timer_.cancel();
1091 : 534 : self->impl_->retry_attempts_ = 0;
1092 : 534 : self->impl_->reconnect_attempt_count_ = 0;
1093 : :
1094 : : #if defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__)
1095 : : int yes = 1;
1096 : : (void)::setsockopt(static_cast<int>(self->impl_->socket_.native_handle()), SOL_SOCKET, SO_NOSIGPIPE, &yes,
1097 : : static_cast<socklen_t>(sizeof(yes)));
1098 : : #endif
1099 : :
1100 : 534 : self->impl_->apply_socket_options();
1101 : :
1102 : : // TCP is up; TLS still has to prove who answered. Reading before the
1103 : : // handshake completes would hand the session ciphertext, and a failed
1104 : : // handshake must not be reported as a working connection - so the
1105 : : // rest of the connect path waits behind it. Plaintext runs the
1106 : : // continuation immediately, which is what it did before TLS existed.
1107 : 1039 : self->impl_->handshake_then(self, seq, [self, seq] { self->impl_->finish_connect(self, seq); });
1108 : 592 : });
1109 : 598 : });
1110 : 598 : }
1111 : :
1112 : : // Sets up the TLS stream if configured, runs the handshake, then hands control
1113 : : // back. Without TLS - or in a build without it - `next` runs straight away and
1114 : : // the connect path is byte for byte what it was.
1115 : 534 : void TcpClient::Impl::handshake_then(std::shared_ptr<TcpClient> self, uint64_t seq, std::function<void()> next) {
1116 : : #ifdef WIRESTEAD_TLS_ENABLED
1117 : 534 : bool want_tls = false;
1118 : 534 : std::string ca_file;
1119 : : {
1120 : 534 : std::lock_guard<std::mutex> lock(cfg_mtx_);
1121 : 534 : want_tls = cfg_.tls_enabled;
1122 : 534 : ca_file = cfg_.tls_ca_file;
1123 : 534 : }
1124 : :
1125 [ + + ]: 534 : if (want_tls) {
1126 : : namespace ssl = boost::asio::ssl;
1127 : : try {
1128 [ + + ]: 30 : if (!ssl_context_) {
1129 : 2 : auto ctx = std::make_shared<ssl::context>(ssl::context::tls_client);
1130 : 2 : ctx->set_options(ssl::context::default_workarounds | ssl::context::no_sslv2 | ssl::context::no_sslv3 |
1131 : : ssl::context::no_tlsv1 | ssl::context::no_tlsv1_1);
1132 [ + + ]: 2 : if (ca_file.empty()) {
1133 : 1 : ctx->set_default_verify_paths();
1134 : : } else {
1135 : 1 : ctx->load_verify_file(ca_file);
1136 : : }
1137 : : // Not optional. An unverified TLS connection encrypts traffic to
1138 : : // whoever answered, which is what an attacker in the middle wants.
1139 : 2 : ctx->set_verify_mode(ssl::verify_peer);
1140 : 2 : ssl_context_ = std::move(ctx);
1141 : 2 : }
1142 : 30 : tls_.emplace(socket_, *ssl_context_);
1143 : 30 : tls_engaged_ = true;
1144 : 30 : tls_->set_verify_callback(ssl::host_name_verification(cfg_.host));
1145 : : // SNI, and the name the certificate is checked against.
1146 : 30 : ::SSL_set_tlsext_host_name(tls_->native_handle(), cfg_.host.c_str());
1147 : 0 : } catch (const std::exception& e) {
1148 : 0 : const std::string msg = std::string("TLS setup failed: ") + e.what();
1149 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "handshake", msg);
1150 : 0 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "handshake",
1151 : : boost::asio::error::invalid_argument, msg, false, 0);
1152 : 0 : tls_.reset();
1153 : 0 : tls_engaged_ = false;
1154 : 0 : handle_close(self, seq, boost::asio::error::invalid_argument);
1155 : 0 : return;
1156 : 0 : }
1157 : :
1158 : 30 : ++pending_io_;
1159 : 60 : tls_->async_handshake(ssl::stream_base::client,
1160 : 60 : net::bind_executor(strand_, [this, self, seq, next](const boost::system::error_code& ec) {
1161 : 30 : IoCompletion completion{this};
1162 [ + - + - : 60 : if (seq != current_seq_.load() || stop_requested_.load() || stopping_.load()) return;
- + - + ]
1163 [ + + ]: 30 : if (ec) {
1164 : 29 : WIRESTEAD_LOG_ERROR("tcp_client", "handshake", "TLS handshake failed: " + ec.message());
1165 : 29 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION,
1166 : 58 : "handshake", ec, "TLS handshake failed: " + ec.message(), false, 0);
1167 : : // Not tls_.reset(): this runs from inside the
1168 : : // stream's own completion handler, and
1169 : : // close_socket() below explains why the stream
1170 : : // has to outlive its operations.
1171 : 29 : tls_engaged_ = false;
1172 : 29 : handle_close(self, seq, ec);
1173 : 29 : return;
1174 : : }
1175 : 1 : next();
1176 : 30 : }));
1177 : 30 : return;
1178 : : }
1179 : : #else
1180 : : (void)self;
1181 : : (void)seq;
1182 : : #endif
1183 : 504 : next();
1184 : 534 : }
1185 : :
1186 : 505 : void TcpClient::Impl::finish_connect(std::shared_ptr<TcpClient> self, uint64_t seq) {
1187 : : // Set here rather than at TCP connect: with TLS, a socket whose handshake
1188 : : // has not finished is not a usable connection, and connected() is what
1189 : : // callers poll before sending. Reporting true for a peer that failed
1190 : : // verification would be worse than useless.
1191 : : {
1192 : 505 : std::lock_guard<std::mutex> lock(submission_mtx_);
1193 [ + - + - : 1010 : if (seq != current_seq_.load() || stop_requested_.load() || stopping_.load()) return;
- + - + ]
1194 : 1010 : write_wait_ = std::make_shared<detail::TcpWriteWait>(connection_seq_.load());
1195 : 505 : connected_.store(true);
1196 : 505 : }
1197 : 505 : rx_ = std::make_shared<std::vector<uint8_t>>(cfg_.read_buffer_size);
1198 : 505 : const auto connection = connection_seq_.load();
1199 : 505 : transition_to(LinkState::Connected);
1200 : 505 : boost::system::error_code ep_ec;
1201 : 505 : auto rep = socket_.remote_endpoint(ep_ec);
1202 [ + - ]: 505 : if (!ep_ec) {
1203 : 505 : WIRESTEAD_LOG_INFO("tcp_client", "connect",
1204 : : fmt::format("Connected to {}:{}", rep.address().to_string(), rep.port()));
1205 : : }
1206 : 505 : start_read(self, seq);
1207 : 505 : reset_idle_timer(self, seq);
1208 : 505 : net::post(strand_, [self, seq, connection]() {
1209 [ + - + + : 1515 : if (seq != self->impl_->current_seq_.load() || connection != self->impl_->connection_seq_.load()) return;
+ + ]
1210 [ + + + - ]: 496 : if (!self->impl_->writing_) self->impl_->do_write(self, seq);
1211 : : });
1212 : : }
1213 : :
1214 : 169 : void TcpClient::Impl::schedule_retry(std::shared_ptr<TcpClient> self, uint64_t seq) {
1215 : 169 : mark_disconnected();
1216 [ + - - + : 169 : if (stop_requested_.load() || stopping_.load()) {
- + ]
1217 : 31 : return;
1218 : : }
1219 : :
1220 : : // Prevent double scheduling of reconnect
1221 [ - + ]: 169 : if (reconnect_pending_.exchange(true)) {
1222 : 0 : return;
1223 : : }
1224 : :
1225 : 169 : std::optional<diagnostics::ErrorInfo> last_err = error_info_holder_.last_error_info();
1226 : :
1227 [ - + ]: 169 : if (!last_err) {
1228 : 0 : last_err = diagnostics::ErrorInfo(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION,
1229 : : "tcp_client", "schedule_retry", "Unknown error",
1230 : 0 : make_error_code(boost::asio::error::not_connected), true);
1231 : : }
1232 : :
1233 : : // Snapshot once rather than locking repeatedly for each read below -
1234 : : // cfg_/reconnect_policy_ can change concurrently via set_retry_interval()
1235 : : // etc. from any user thread while this runs on the strand (#436).
1236 : 169 : TcpClientConfig cfg_snapshot;
1237 : 169 : std::optional<ReconnectPolicy> reconnect_policy_snapshot;
1238 : : {
1239 : 169 : std::lock_guard<std::mutex> lock(cfg_mtx_);
1240 : 169 : cfg_snapshot = cfg_;
1241 : 169 : reconnect_policy_snapshot = reconnect_policy_;
1242 : 169 : }
1243 : :
1244 : : // Determine current attempt count based on active mode
1245 : : uint32_t current_attempts =
1246 [ + + ]: 169 : reconnect_policy_snapshot ? reconnect_attempt_count_ : static_cast<uint32_t>(retry_attempts_);
1247 : :
1248 : 169 : auto decision = detail::decide_reconnect(cfg_snapshot, *last_err, current_attempts, reconnect_policy_snapshot);
1249 : :
1250 [ + + ]: 169 : if (!decision.should_retry) {
1251 : 31 : WIRESTEAD_LOG_INFO("tcp_client", "retry", "Reconnect stopped by policy/config");
1252 : 31 : transition_to(LinkState::Error);
1253 : 31 : reconnect_pending_.store(false);
1254 : 31 : return;
1255 : : }
1256 : :
1257 : : // The decider returns a base delay for both policy and fallback paths.
1258 : 138 : std::chrono::milliseconds delay = decision.delay.value_or(std::chrono::milliseconds(cfg_snapshot.retry_interval_ms));
1259 [ + + ]: 138 : if (reconnect_policy_snapshot) {
1260 : 25 : reconnect_attempt_count_++;
1261 : : } else {
1262 : : // Preserve existing "fast first retry" behavior for non-policy mode.
1263 : 113 : ++retry_attempts_;
1264 [ + + ]: 113 : if (retry_attempts_ == 1) {
1265 : 103 : delay = std::chrono::milliseconds(first_retry_interval_ms_);
1266 : : }
1267 : : }
1268 : :
1269 : 138 : transition_to(LinkState::Connecting);
1270 : :
1271 : 138 : WIRESTEAD_LOG_INFO("tcp_client", "retry",
1272 : : fmt::format("Scheduling retry in {:.3f}s", static_cast<double>(delay.count()) / 1000.0));
1273 : :
1274 : 138 : retry_timer_.expires_after(delay);
1275 : 138 : ++pending_io_;
1276 : 138 : retry_timer_.async_wait([self, seq](const boost::system::error_code& ec) {
1277 : 138 : IoCompletion completion{self->impl_.get()};
1278 : : // Clear pending flag regardless of result (fired or aborted)
1279 : 138 : self->impl_->reconnect_pending_.store(false);
1280 : :
1281 [ + + - + : 241 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) {
+ + ]
1282 : 35 : return;
1283 : : }
1284 [ + - + - : 103 : if (!ec && !self->impl_->stop_requested_.load() && !self->impl_->stopping_.load())
+ - + - ]
1285 : 103 : self->impl_->do_resolve_connect(self, seq);
1286 : 138 : });
1287 : 231 : }
1288 : :
1289 : 722 : void TcpClient::Impl::start_read(std::shared_ptr<TcpClient> self, uint64_t seq) {
1290 : 722 : const auto connection = connection_seq_.load();
1291 : 722 : auto buffer = rx_;
1292 : 722 : ++pending_io_;
1293 : 722 : auto on_read = [self, seq, connection, buffer](auto ec, std::size_t n) {
1294 : 722 : IoCompletion completion{self->impl_.get()};
1295 [ + + + - : 1324 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load() ||
+ + ]
1296 [ + + ]: 602 : connection != self->impl_->connection_seq_.load()) {
1297 : 451 : return;
1298 : : }
1299 [ + + ]: 271 : if (self->impl_->stop_requested_.load()) {
1300 : 9 : return;
1301 : : }
1302 [ + + ]: 262 : if (ec) {
1303 : 45 : self->impl_->handle_close(self, seq, ec);
1304 : 45 : return;
1305 : : }
1306 [ + - ]: 217 : if (n > 0) {
1307 : 217 : self->impl_->reset_idle_timer(self, seq);
1308 : : }
1309 : 217 : interface::SharedCallback<OnBytes> on_bytes;
1310 : : {
1311 : 217 : std::lock_guard<std::mutex> lock(self->impl_->callback_mtx_);
1312 : 217 : on_bytes = self->impl_->on_bytes_;
1313 : 217 : }
1314 : :
1315 : 217 : self->impl_->stats_.record_received(n);
1316 : :
1317 : 217 : diagnostics::invoke_callback("tcp_client", "on_bytes", on_bytes, memory::ConstByteSpan(buffer->data(), n));
1318 : 217 : self->impl_->start_read(self, seq);
1319 : 1444 : };
1320 : : #ifdef WIRESTEAD_TLS_ENABLED
1321 [ + + ]: 722 : if (tls_active()) {
1322 : 2 : tls_->async_read_some(net::buffer(buffer->data(), buffer->size()), std::move(on_read));
1323 : 2 : return;
1324 : : }
1325 : : #endif
1326 : 720 : socket_.async_read_some(net::buffer(buffer->data(), buffer->size()), std::move(on_read));
1327 : 724 : }
1328 : :
1329 : 3333 : void TcpClient::Impl::do_write(std::shared_ptr<TcpClient> self, uint64_t seq) {
1330 [ + + ]: 3333 : if (stop_requested_.load()) {
1331 : 24 : tx_.clear();
1332 : 24 : queue_bytes_ = 0;
1333 : 24 : pending_.clear();
1334 : 24 : pending_bytes_ = 0;
1335 : 24 : writing_ = false;
1336 : 24 : report_backpressure(self, queue_bytes_);
1337 : 1742 : return;
1338 : : }
1339 : :
1340 [ - + ]: 3309 : if (!connected_.load()) {
1341 : 0 : writing_ = false;
1342 : 0 : return;
1343 : : }
1344 : :
1345 [ + + + - : 3309 : if (tx_.empty() || state_.is_state(LinkState::Closed) || state_.is_state(LinkState::Error)) {
- + + + ]
1346 : 1709 : writing_ = false;
1347 : 1709 : return;
1348 : : }
1349 : 1600 : writing_ = true;
1350 : :
1351 : : // The ledger, not the admission mutex, fences this handoff against stop and
1352 : : // loss: taking the admission mutex here per batch contends with every producer.
1353 : 1600 : auto batch = std::make_shared<WriteBatch>();
1354 : 1600 : const auto queued_bytes = queue_util::take_gather_batch(tx_, batch->buffers, batch->views, BufferProjection{});
1355 : 1600 : batch->bytes = queued_bytes;
1356 [ + + + + ]: 3198 : if (stop_requested_.load() ||
1357 [ + - - + ]: 9722 : !send_accounting_.begin_batch(batch->buffers, [](const auto& item) { return item.request; })) {
1358 : : // Stop/loss already terminated these; leave them to its queue cleanup.
1359 : 2 : queue_util::return_gather_batch(tx_, batch->buffers);
1360 : 2 : writing_ = false;
1361 : 2 : return;
1362 : : }
1363 : 1598 : active_write_ = batch;
1364 : 1598 : const auto connection = connection_seq_.load();
1365 : :
1366 : 1598 : ++pending_io_;
1367 : 1591 : auto on_write = [self, queued_bytes, seq, connection, batch](auto ec, std::size_t bytes_written) {
1368 : 1591 : IoCompletion completion{self->impl_.get()};
1369 : : bool current_connection;
1370 : : {
1371 : : // Only a failure must close admission; success needs just the ledger.
1372 : 1591 : std::unique_lock<std::mutex> admission_lock(self->impl_->submission_mtx_, std::defer_lock);
1373 [ + + + - ]: 1591 : if (ec) admission_lock.lock();
1374 : 1591 : current_connection = connection == self->impl_->connection_seq_.load();
1375 : 1591 : self->impl_->send_accounting_.complete_batch(
1376 : 5646 : batch->buffers, bytes_written, [](const auto& item) { return item.request; },
1377 : 4055 : [](const auto& item) {
1378 : 8110 : return std::visit([](const auto& b) { return queue_util::variant_buffer_size(b); }, item.buffer);
1379 : : });
1380 : : // Preserve the completed prefix and close admission before a competing
1381 : : // stop can reclassify the rest of this connection as an explicit stop.
1382 [ + + + + : 1618 : if (current_connection && ec && ec != net::error::operation_aborted && seq == self->impl_->current_seq_.load()) {
+ - + - +
+ ]
1383 : 27 : self->impl_->mark_disconnected_locked();
1384 : : }
1385 : 1591 : }
1386 [ + + ]: 1591 : if (!current_connection) {
1387 : : // Release pooled buffers while the captured client still owns its pool.
1388 : 63 : batch->buffers.clear();
1389 : 63 : return;
1390 : : }
1391 : 1528 : self->impl_->active_write_.reset();
1392 [ + - - + : 3056 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) {
- + ]
1393 : 0 : batch->buffers.clear();
1394 : 0 : queue_util::release_reserved_write_bytes(self->impl_->queue_bytes_, queued_bytes);
1395 : 0 : self->impl_->report_backpressure(self, self->impl_->queue_bytes_);
1396 : 0 : self->impl_->writing_ = false;
1397 : 0 : return;
1398 : : }
1399 : :
1400 [ + + ]: 1528 : if (ec) {
1401 : 27 : queue_util::return_gather_batch(self->impl_->tx_, batch->buffers);
1402 : :
1403 : 27 : WIRESTEAD_LOG_ERROR("tcp_client", "do_write", fmt::format("Write failed: {}", ec.message()));
1404 : 27 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::COMMUNICATION, "write", ec,
1405 : : fmt::format("Write failed: {}", ec.message()), false, 0);
1406 : 27 : self->impl_->writing_ = false;
1407 : 27 : self->impl_->handle_close(self, seq, ec);
1408 : 27 : return;
1409 : : }
1410 : :
1411 : 1501 : batch->buffers.clear();
1412 : 1501 : self->impl_->stats_.record_sent(bytes_written);
1413 [ + - ]: 1501 : if (bytes_written > 0) {
1414 : 1501 : self->impl_->reset_idle_timer(self, seq);
1415 : : }
1416 : 1501 : queue_util::release_reserved_write_bytes(self->impl_->queue_bytes_, queued_bytes);
1417 : 1501 : self->impl_->report_backpressure(self, self->impl_->queue_bytes_);
1418 : :
1419 [ + + + - : 3001 : if (self->impl_->stop_requested_.load() || self->impl_->state_.is_state(LinkState::Closed) ||
- + ]
1420 : 3001 : self->impl_->state_.is_state(LinkState::Error)) {
1421 : 1 : self->impl_->writing_ = false;
1422 : 1 : return;
1423 : : }
1424 : :
1425 : 1500 : self->impl_->do_write(self, seq);
1426 : 3189 : };
1427 : :
1428 : : try {
1429 [ + + - + ]: 1598 : if (auto hook = detail::g_tcp_write_initiation_hook.load()) hook();
1430 : : #ifdef WIRESTEAD_TLS_ENABLED
1431 [ + + ]: 1591 : if (tls_active()) {
1432 : 1 : net::async_write(*tls_, batch->views, on_write);
1433 : : } else
1434 : : #endif
1435 : : {
1436 : 1590 : net::async_write(socket_, batch->views, on_write);
1437 : : }
1438 : 7 : } catch (...) {
1439 : : // No completion handler owns this pending-I/O slot after failed initiation.
1440 : : // The accepted request crossed the local-write boundary; connection cleanup
1441 : : // terminates it as an abort and discards requests still awaiting handoff.
1442 : 7 : --pending_io_;
1443 : 7 : mark_disconnected();
1444 : 7 : handle_close(self, seq, make_error_code(net::error::no_buffer_space));
1445 : 7 : return;
1446 : 7 : }
1447 [ + + + - ]: 1591 : if (auto hook = detail::g_tcp_write_started_hook.load()) hook();
1448 : 1607 : }
1449 : :
1450 : 114 : void TcpClient::Impl::handle_close(std::shared_ptr<TcpClient> self, uint64_t seq, const boost::system::error_code& ec) {
1451 [ + - - + : 228 : if (ec == net::error::operation_aborted || seq != current_seq_.load()) {
- + ]
1452 : 0 : return;
1453 : : }
1454 : 114 : mark_disconnected();
1455 : 114 : discard_connection_writes();
1456 : 114 : WIRESTEAD_LOG_INFO("tcp_client", "handle_close", fmt::format("Closing connection. Error: {}", ec.message()));
1457 [ + - ]: 114 : if (ec) {
1458 : : bool has_policy;
1459 : : {
1460 : 114 : std::lock_guard<std::mutex> lock(cfg_mtx_);
1461 : 114 : has_policy = reconnect_policy_.has_value();
1462 : 114 : }
1463 : 114 : const bool retryable = diagnostics::is_retryable_tcp_connect_error(ec);
1464 [ + + ]: 114 : const uint32_t current_attempts = has_policy ? reconnect_attempt_count_ : static_cast<uint32_t>(retry_attempts_);
1465 : :
1466 : 114 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "handle_close", ec,
1467 : 228 : fmt::format("Connection closed with error: {}", ec.message()), retryable, current_attempts);
1468 : : }
1469 : 114 : writing_ = false;
1470 : 114 : cancel_idle_timer();
1471 : 114 : connect_timer_.cancel();
1472 : 114 : close_socket();
1473 [ + + + - : 114 : if (stop_requested_.load() || stopping_.load() || state_.is_state(LinkState::Closed)) {
- + + + ]
1474 : 1 : transition_to(LinkState::Closed, ec);
1475 : 1 : return;
1476 : : }
1477 : 113 : transition_to(LinkState::Connecting, ec);
1478 : 113 : schedule_retry(self, seq);
1479 : : }
1480 : :
1481 : 3 : void TcpClient::Impl::handle_idle_timeout(std::shared_ptr<TcpClient> self, uint64_t seq) {
1482 [ + - + - : 6 : if (seq != current_seq_.load() || stop_requested_.load() || stopping_.load() || !connected_.load()) {
+ - - + -
+ ]
1483 : 2 : return;
1484 : : }
1485 : :
1486 : 3 : mark_disconnected();
1487 : 3 : discard_connection_writes();
1488 : : IdleTimeoutAction idle_timeout_action;
1489 : : unsigned idle_timeout_ms;
1490 : : bool has_policy;
1491 : : {
1492 : 3 : std::lock_guard<std::mutex> lock(cfg_mtx_);
1493 : 3 : idle_timeout_action = cfg_.idle_timeout_action;
1494 : 3 : idle_timeout_ms = cfg_.idle_timeout_ms;
1495 : 3 : has_policy = reconnect_policy_.has_value();
1496 : 3 : }
1497 : :
1498 : 3 : const auto ec = make_error_code(boost::asio::error::timed_out);
1499 : 3 : const bool should_reconnect = idle_timeout_action == IdleTimeoutAction::Reconnect;
1500 [ - + ]: 3 : const uint32_t current_attempts = has_policy ? reconnect_attempt_count_ : static_cast<uint32_t>(retry_attempts_);
1501 : :
1502 [ + - + - : 3 : WIRESTEAD_LOG_WARNING("tcp_client", "idle_timeout",
+ - + - +
+ + - +
- ]
1503 : : fmt::format("Idle timeout expired after {}ms; {}", idle_timeout_ms,
1504 : : should_reconnect ? "scheduling reconnect" : "closing connection"));
1505 : 3 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "idle_timeout", ec,
1506 : : "Idle timeout expired", should_reconnect, current_attempts);
1507 : :
1508 : 3 : writing_ = false;
1509 : 3 : cancel_idle_timer();
1510 : 3 : connect_timer_.cancel();
1511 : 3 : close_socket();
1512 : :
1513 [ + + ]: 3 : if (!should_reconnect) {
1514 : 2 : transition_to(LinkState::Closed, ec);
1515 : 2 : return;
1516 : : }
1517 : :
1518 : 1 : transition_to(LinkState::Connecting, ec);
1519 : 1 : schedule_retry(self, seq);
1520 : : }
1521 : :
1522 : 117 : void TcpClient::Impl::discard_connection_writes() {
1523 : 117 : size_t messages = tx_.size() + pending_.size();
1524 : 117 : size_t bytes = 0;
1525 [ + + ]: 159 : for (const auto& buffer : tx_)
1526 : 84 : bytes += std::visit([](const auto& b) { return queue_util::variant_buffer_size(b); }, buffer.buffer);
1527 [ + + ]: 118 : for (const auto& buffer : pending_)
1528 : 2 : bytes += std::visit([](const auto& b) { return queue_util::variant_buffer_size(b); }, buffer.buffer);
1529 [ + + ]: 117 : if (active_write_) {
1530 : 20 : messages += active_write_->buffers.size();
1531 : 20 : bytes += active_write_->bytes;
1532 : : // Its completion handler retains the buffers; it must not mutate new state.
1533 : 20 : active_write_.reset();
1534 : : }
1535 : 117 : tx_.clear();
1536 : 117 : pending_.clear();
1537 : 117 : queue_bytes_ = 0;
1538 : 117 : pending_bytes_ = 0;
1539 : 117 : writing_ = false;
1540 : 117 : backpressure_active_ = false;
1541 [ + + ]: 117 : if (messages) stats_.record_dropped(messages, bytes);
1542 : 117 : }
1543 : :
1544 : 1334 : void TcpClient::Impl::close_socket() {
1545 : 1334 : boost::system::error_code ec;
1546 : : #ifdef WIRESTEAD_TLS_ENABLED
1547 : : // One SSL_shutdown writes close_notify and returns; a second would wait for
1548 : : // the peer's, which is the block measured at 8 s on the server side.
1549 [ + + + - ]: 1334 : if (tls_) ::SSL_shutdown(tls_->native_handle());
1550 : : // Deliberately no tls_.reset() here. socket_.cancel()/close() do not retract
1551 : : // a boost::asio::ssl::detail::io_op continuation that is already queued on
1552 : : // the strand: the strand runs it afterwards, and it calls back into the
1553 : : // engine's BIO. Destroying the stream at this point therefore left that
1554 : : // continuation dereferencing freed memory - a SIGSEGV inside BIO_ctrl,
1555 : : // reproducible under parallel load and seen intermittently in CI.
1556 : : //
1557 : : // The stream is instead destroyed by the next tls_.emplace(), which runs on
1558 : : // the strand after any pending continuation, or by ~Impl once the io thread
1559 : : // has been joined. It cannot be moved out to a holder either - a pending
1560 : : // operation holds the stream's original address.
1561 : 1334 : tls_engaged_ = false;
1562 : : #endif
1563 : 1334 : socket_.shutdown(tcp::socket::shutdown_both, ec);
1564 : 1334 : socket_.close(ec);
1565 : 1334 : }
1566 : :
1567 : 1163 : void TcpClient::Impl::recalculate_backpressure_bounds() {
1568 : 1163 : bp_high_ = cfg_.backpressure_threshold;
1569 [ + - ]: 1163 : bp_low_ = bp_high_ > 1 ? bp_high_ / 2 : bp_high_;
1570 [ - + ]: 1163 : if (bp_low_ == 0) {
1571 : 0 : bp_low_ = 1;
1572 : : }
1573 : 1163 : bp_limit_ = std::min(std::max(bp_high_ * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
1574 : : base::constants::MAX_BUFFER_SIZE);
1575 [ - + ]: 1163 : if (bp_limit_ < bp_high_) {
1576 : 0 : bp_limit_ = bp_high_;
1577 : : }
1578 : 1163 : backpressure_active_ = false;
1579 : 1163 : }
1580 : :
1581 : 7566 : queue_util::BackpressureFields TcpClient::Impl::bp_fields() {
1582 : 7566 : return queue_util::BackpressureFields{queue_bytes_,
1583 : 7566 : pending_bytes_,
1584 : 7566 : backpressure_active_,
1585 : 7566 : bp_high_,
1586 : 7566 : bp_low_,
1587 : 7566 : bp_limit_,
1588 : 7566 : bp_strategy_.load(std::memory_order_relaxed),
1589 : 7566 : &write_reserve_mtx_};
1590 : : }
1591 : :
1592 : 2050 : void TcpClient::Impl::route_enqueued_buffer(std::shared_ptr<TcpClient> self, TrackedBuffer&& buf, size_t added) {
1593 [ + + + - : 2050 : if (stop_requested_.load() || state_.is_state(LinkState::Closed) || state_.is_state(LinkState::Error)) {
- + + + ]
1594 : 53 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
1595 : 53 : stats_.record_failed_send();
1596 : 77 : return;
1597 : : }
1598 : :
1599 : 1997 : auto f = bp_fields();
1600 : 1997 : queue_util::DropAccounting dropped;
1601 : 1997 : auto decision = queue_util::decide_enqueue(f, added, tx_, dropped, BufferProjection{}, [this](const auto& removed) {
1602 : : send_accounting_.discard(removed.request, Ledger::Cause::QueuePressure);
1603 : : });
1604 [ - + - - ]: 1997 : if (dropped.any()) stats_.record_dropped(dropped.messages, dropped.bytes);
1605 : :
1606 [ - + ]: 1997 : if (decision == queue_util::EnqueueDecision::Rejected) {
1607 : 0 : send_accounting_.discard(buf.request, Ledger::Cause::QueuePressure);
1608 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "write", fmt::format("Queue limit exceeded ({} bytes)", queue_bytes_ + added));
1609 : 0 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::COMMUNICATION, "write",
1610 : : boost::asio::error::no_buffer_space, "Queue limit exceeded", false, 0);
1611 : : // #448: this path used to leave RuntimeStats showing the message as
1612 : : // accepted (from the caller-thread pre-check) with no corresponding
1613 : : // sent/dropped/queued accounting - record it as dropped so it's at
1614 : : // least observable.
1615 : 0 : stats_.record_dropped(1, added);
1616 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
1617 : 0 : report_backpressure(self, queue_bytes_ + added);
1618 : 0 : return;
1619 : : }
1620 [ + + ]: 1997 : if (decision == queue_util::EnqueueDecision::Pending) {
1621 : 24 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, pending_bytes_, inflight_bytes_, added);
1622 : 24 : pending_.emplace_back(std::move(buf));
1623 : 24 : observe_queue();
1624 : 24 : return;
1625 : : }
1626 : 1973 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, queue_bytes_, inflight_bytes_, added);
1627 : 1973 : tx_.emplace_back(std::move(buf));
1628 : 1973 : observe_queue();
1629 : 1973 : report_backpressure(self, queue_bytes_);
1630 [ + + + - ]: 3311 : if (!writing_) do_write(self, current_seq_.load());
1631 : : }
1632 : :
1633 : 9703 : void TcpClient::Impl::observe_queue() {
1634 : 29109 : stats_.observe_queue(queue_bytes_.load(std::memory_order_relaxed) + pending_bytes_.load(std::memory_order_relaxed));
1635 : 9703 : }
1636 : :
1637 : 5602 : void TcpClient::Impl::report_backpressure(std::shared_ptr<TcpClient> self, size_t queued_bytes) {
1638 [ + + - + : 5602 : if (stop_requested_.load() || stopping_.load()) return;
+ + ]
1639 : 5569 : observe_queue();
1640 : :
1641 : 5569 : interface::SharedCallback<OnBackpressure> on_bp;
1642 : : {
1643 : 5569 : std::lock_guard<std::mutex> lock(callback_mtx_);
1644 : 5569 : on_bp = on_bp_;
1645 : 5569 : }
1646 : 5569 : static const OnBackpressure kNoCallback;
1647 : :
1648 : 5569 : auto f = bp_fields();
1649 [ + + + - ]: 10772 : queue_util::report_backpressure(
1650 : 5203 : f, queued_bytes, on_bp ? *on_bp : kNoCallback, stats_,
1651 : 0 : [&]() -> size_t {
1652 : 33 : const size_t moved = pending_bytes_.exchange(0);
1653 [ + + ]: 56 : while (!pending_.empty()) {
1654 : 23 : tx_.emplace_back(std::move(pending_.front()));
1655 : 23 : pending_.pop_front();
1656 : : }
1657 : 33 : return moved;
1658 : : },
1659 : 5569 : [&]() {
1660 : 33 : observe_queue();
1661 [ - + - - ]: 33 : if (!writing_) do_write(self, current_seq_.load());
1662 : 33 : });
1663 : 5569 : }
1664 : :
1665 : 2008 : void TcpClient::Impl::transition_to(LinkState next, const boost::system::error_code& ec) {
1666 [ - + ]: 2008 : if (ec == net::error::operation_aborted) {
1667 : 0 : return;
1668 : : }
1669 : :
1670 : 2008 : const auto current = state_.get();
1671 [ + + + + ]: 2008 : const bool retrying_same_state = (next == LinkState::Connecting && current == LinkState::Connecting);
1672 [ + + + + ]: 2008 : if ((current == LinkState::Closed || current == LinkState::Error) &&
1673 [ - + - - ]: 34 : (next == LinkState::Closed || next == LinkState::Error)) {
1674 : 34 : return;
1675 : : }
1676 : :
1677 [ + + + + ]: 1974 : if (next == LinkState::Closed || next == LinkState::Error) {
1678 [ - + ]: 722 : if (terminal_state_notified_.exchange(true)) {
1679 : 0 : return;
1680 : : }
1681 [ + + - + ]: 1252 : } else if (current == next && !retrying_same_state) {
1682 : 0 : return;
1683 : : }
1684 : :
1685 : 1974 : state_.set(next);
1686 : 1974 : notify_state();
1687 : : }
1688 : :
1689 : 722 : void TcpClient::Impl::perform_stop_cleanup() {
1690 [ - + ]: 722 : if (cleanup_started_.exchange(true)) return;
1691 : : // Signals completion on every exit, including the error paths below.
1692 : : struct CompletionSignal {
1693 : : Impl* impl;
1694 : 722 : ~CompletionSignal() {
1695 : 722 : impl->cleanup_finished_ = true;
1696 [ + + ]: 722 : if (impl->pending_io_ == 0) impl->mark_cleanup_done();
1697 : 722 : }
1698 : 722 : } completion_signal{this};
1699 : : try {
1700 : 722 : retry_timer_.cancel();
1701 : 722 : connect_timer_.cancel();
1702 : 722 : cancel_idle_timer();
1703 : 722 : resolver_.cancel();
1704 : 722 : boost::system::error_code ec_cancel;
1705 : 722 : socket_.cancel(ec_cancel);
1706 : 722 : close_socket();
1707 : 722 : tx_.clear();
1708 : 722 : queue_bytes_ = 0;
1709 : 722 : pending_.clear();
1710 : 722 : pending_bytes_ = 0;
1711 : 722 : writing_ = false;
1712 : 722 : active_write_.reset();
1713 : 722 : mark_disconnected();
1714 : : // Deliberately does NOT fire on_bp_ here, unlike UDP/server sessions'
1715 : : // terminal drain (#434): this is an explicit, tested contract
1716 : : // (ContractComplianceTest.TcpClient_Backpressure_Contract) - a
1717 : : // Reliable-mode caller blocked in send_blocking() is woken instead via
1718 : : // the wrapper's own bp_cv_.notify_all()/is_connected() check, not a
1719 : : // relief callback. Don't "fix" this to match the other transports
1720 : : // without updating that contract test first.
1721 : 722 : backpressure_active_ = false;
1722 : :
1723 [ + + + + : 722 : if (owns_ioc_ && work_guard_) {
+ + ]
1724 : 163 : work_guard_->reset();
1725 : : }
1726 : 722 : transition_to(LinkState::Closed);
1727 : 0 : } catch (const std::exception& e) {
1728 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "stop_cleanup", fmt::format("Cleanup error: {}", e.what()));
1729 : 0 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::SYSTEM, "stop_cleanup", {},
1730 : 0 : fmt::format("Cleanup error: {}", e.what()), false, 0);
1731 : 0 : diagnostics::error_reporting::report_system_error("tcp_client", "stop_cleanup",
1732 : 0 : fmt::format("Exception in stop cleanup: {}", e.what()));
1733 : 0 : } catch (...) {
1734 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "stop_cleanup", "Unknown error in stop cleanup");
1735 : 0 : diagnostics::error_reporting::report_system_error("tcp_client", "stop_cleanup", "Unknown error in stop cleanup");
1736 : 0 : }
1737 : 722 : }
1738 : :
1739 : 700 : void TcpClient::Impl::reset_start_state() {
1740 : 700 : cleanup_started_.store(false);
1741 : 700 : cleanup_finished_ = false;
1742 : 700 : inflight_bytes_.store(0);
1743 : : {
1744 : 700 : std::lock_guard<std::mutex> lock(stop_mtx_);
1745 : 700 : cleanup_done_ = false;
1746 : 700 : }
1747 : 700 : stop_requested_.store(false);
1748 : 700 : stopping_.store(false);
1749 : 700 : terminal_state_notified_.store(false);
1750 : 700 : reconnect_pending_.store(false);
1751 : 700 : retry_attempts_ = 0;
1752 : 700 : reconnect_attempt_count_ = 0;
1753 : 700 : mark_disconnected();
1754 : 700 : writing_ = false;
1755 : 700 : queue_bytes_ = 0;
1756 : 700 : pending_.clear();
1757 : 700 : pending_bytes_ = 0;
1758 : 700 : backpressure_active_ = false;
1759 : 700 : state_.set(LinkState::Idle);
1760 : 700 : }
1761 : :
1762 : 1922 : void TcpClient::Impl::join_ioc_thread(bool allow_detach) {
1763 [ + + + + : 1922 : if (!owns_ioc_ || !ioc_thread_.joinable()) {
+ + ]
1764 : 1759 : return;
1765 : : }
1766 : :
1767 [ - + ]: 163 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
1768 [ # # ]: 0 : if (allow_detach) {
1769 : 0 : ioc_thread_.detach();
1770 : : }
1771 : 0 : return;
1772 : : }
1773 : :
1774 : : try {
1775 : 163 : ioc_thread_.join();
1776 : 0 : } catch (const std::exception& e) {
1777 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "join", "Join failed: " + std::string(e.what()));
1778 : 0 : } catch (...) {
1779 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "join", "Join failed with unknown error");
1780 : 0 : }
1781 : : }
1782 : :
1783 : 1974 : void TcpClient::Impl::notify_state() {
1784 [ + + - + : 2426 : if (stop_requested_.load() || stopping_.load()) return;
+ + ]
1785 : :
1786 : 1284 : interface::SharedCallback<OnState> on_state;
1787 : : {
1788 : 1284 : std::lock_guard<std::mutex> lock(callback_mtx_);
1789 : 1284 : on_state = on_state_;
1790 : 1284 : }
1791 [ + + ]: 1284 : if (!on_state) return;
1792 : :
1793 : : try {
1794 : 832 : (*on_state)(state_.get());
1795 : 5 : } catch (const std::exception& e) {
1796 : 12 : WIRESTEAD_LOG_ERROR("tcp_client", "on_state", "Exception in state callback: " + std::string(e.what()));
1797 : 5 : } catch (...) {
1798 : 1 : WIRESTEAD_LOG_ERROR("tcp_client", "on_state", "Unknown exception in state callback");
1799 : 1 : }
1800 : 1284 : }
1801 : :
1802 : 230 : void TcpClient::Impl::record_error(diagnostics::ErrorLevel lvl, diagnostics::ErrorCategory cat,
1803 : : std::string_view operation, const boost::system::error_code& ec,
1804 : : std::string_view msg, bool retryable, uint32_t retry_count) {
1805 : 230 : error_info_holder_.record_error(lvl, cat, operation, ec, msg, retryable, retry_count);
1806 : 230 : }
1807 : :
1808 : 495 : void TcpClient::Impl::reset_io_objects() {
1809 : : try {
1810 : 495 : boost::system::error_code ec_cancel;
1811 : 495 : socket_.cancel(ec_cancel);
1812 : 495 : close_socket();
1813 : 495 : socket_ = tcp::socket(strand_);
1814 : 495 : resolver_.cancel();
1815 : 495 : resolver_ = tcp::resolver(strand_);
1816 : 495 : retry_timer_ = net::steady_timer(strand_);
1817 : 495 : connect_timer_ = net::steady_timer(strand_);
1818 : 495 : idle_timer_ = net::steady_timer(strand_);
1819 : 495 : tx_.clear();
1820 : 495 : queue_bytes_ = 0;
1821 : 495 : pending_.clear();
1822 : 495 : pending_bytes_ = 0;
1823 : 495 : writing_ = false;
1824 : 495 : backpressure_active_ = false;
1825 : 0 : } catch (const std::exception& e) {
1826 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "reset_io_objects", fmt::format("Reset error: {}", e.what()));
1827 : 0 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::SYSTEM, "reset_io_objects", {},
1828 : 0 : fmt::format("Reset error: {}", e.what()), false, 0);
1829 : 0 : diagnostics::error_reporting::report_system_error(
1830 : 0 : "tcp_client", "reset_io_objects", fmt::format("Exception while resetting io objects: {}", e.what()));
1831 : 0 : } catch (...) {
1832 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "reset_io_objects", "Unknown reset error");
1833 : 0 : diagnostics::error_reporting::report_system_error("tcp_client", "reset_io_objects",
1834 : : "Unknown error while resetting io objects");
1835 : 0 : }
1836 : 495 : }
1837 : :
1838 : 2223 : void TcpClient::Impl::reset_idle_timer(std::shared_ptr<TcpClient> self, uint64_t seq) {
1839 : : unsigned idle_timeout_ms;
1840 : : {
1841 : 2223 : std::lock_guard<std::mutex> lock(cfg_mtx_);
1842 : 2223 : idle_timeout_ms = cfg_.idle_timeout_ms;
1843 : 2223 : }
1844 [ + + + - : 2223 : if (idle_timeout_ms == 0 || !connected_.load() || stop_requested_.load() || stopping_.load()) {
+ - - + +
+ ]
1845 : 2211 : return;
1846 : : }
1847 : :
1848 : 12 : idle_timer_.cancel();
1849 : 12 : idle_timer_.expires_after(std::chrono::milliseconds(idle_timeout_ms));
1850 : 12 : const auto connection = connection_seq_.load();
1851 : 12 : ++pending_io_;
1852 : 12 : idle_timer_.async_wait([self, seq, connection](const boost::system::error_code& ec) {
1853 : 12 : IoCompletion completion{self->impl_.get()};
1854 [ + + + - : 18 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load() ||
+ + ]
1855 [ - + ]: 6 : connection != self->impl_->connection_seq_.load()) {
1856 : 9 : return;
1857 : : }
1858 [ + - ]: 3 : if (!ec) {
1859 : 3 : self->impl_->handle_idle_timeout(self, seq);
1860 : : }
1861 : 12 : });
1862 : : }
1863 : :
1864 : 839 : void TcpClient::Impl::cancel_idle_timer() { idle_timer_.cancel(); }
1865 : :
1866 : : } // namespace transport
1867 : : } // namespace wirestead
|
