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 : :
21 : : #if defined(__GNUC__) || defined(__clang__)
22 : : #pragma GCC diagnostic ignored "-Wsign-conversion"
23 : : #endif
24 : :
25 : : #include <spdlog/fmt/fmt.h>
26 : :
27 : : #include <algorithm>
28 : : #include <array>
29 : : #include <atomic>
30 : : #include <boost/asio.hpp>
31 : : #ifdef WIRESTEAD_TLS_ENABLED
32 : : #include <boost/asio/ssl.hpp>
33 : : #endif
34 : : #include <cstdint>
35 : : #include <cstring>
36 : : #include <deque>
37 : : #include <iostream>
38 : : #include <memory>
39 : : #include <mutex>
40 : : #include <optional>
41 : : #include <stop_token>
42 : : #include <string>
43 : : #include <thread>
44 : : #include <type_traits>
45 : : #include <variant>
46 : : #include <vector>
47 : :
48 : : #if defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__)
49 : : #include <sys/socket.h>
50 : : #include <sys/types.h>
51 : : #endif
52 : :
53 : : #include "wirestead/base/constants.hpp"
54 : : #include "wirestead/concurrency/io_context_manager.hpp"
55 : : #include "wirestead/concurrency/thread_safe_state.hpp"
56 : : #include "wirestead/diagnostics/error_handler.hpp"
57 : : #include "wirestead/diagnostics/error_mapping.hpp"
58 : : #include "wirestead/diagnostics/logger.hpp"
59 : : #include "wirestead/diagnostics/runtime_stats_counter.hpp"
60 : : #include "wirestead/memory/memory_pool.hpp"
61 : : #include "wirestead/transport/base/bp_state_machine.hpp"
62 : : #include "wirestead/transport/base/bp_utils.hpp"
63 : : #include "wirestead/transport/base/error_info_holder.hpp"
64 : : #include "wirestead/transport/tcp_client/detail/reconnect_decider.hpp"
65 : :
66 : : namespace wirestead {
67 : : namespace transport {
68 : :
69 : : namespace net = boost::asio;
70 : : using tcp = net::ip::tcp;
71 : :
72 : : using base::LinkState;
73 : : using concurrency::AtomicLinkState;
74 : : using config::TcpClientConfig;
75 : : using interface::Channel;
76 : :
77 : : struct TcpClient::Impl {
78 : : // Members moved from TcpClient
79 : : std::shared_ptr<net::io_context> owned_ioc_;
80 : : net::io_context* ioc_ = nullptr;
81 : : net::strand<net::io_context::executor_type> strand_;
82 : : std::unique_ptr<net::executor_work_guard<net::io_context::executor_type>> work_guard_;
83 : : std::jthread ioc_thread_;
84 : : std::atomic<uint64_t> lifecycle_seq_{0};
85 : : std::atomic<uint64_t> stop_seq_{0};
86 : : std::atomic<uint64_t> current_seq_{0};
87 : : tcp::resolver resolver_;
88 : : tcp::socket socket_;
89 : :
90 : : #ifdef WIRESTEAD_TLS_ENABLED
91 : : // The stream borrows socket_ rather than owning it, so connect, socket
92 : : // options, cancel and close all keep operating on socket_ exactly as they do
93 : : // without TLS. Only reads, writes and shutdown route through here, and only
94 : : // while a connection is up - it is rebuilt per connection because a TLS
95 : : // session cannot outlive the socket it negotiated on.
96 : : std::shared_ptr<boost::asio::ssl::context> ssl_context_;
97 : : std::optional<boost::asio::ssl::stream<tcp::socket&>> tls_;
98 : : // Whether IO should currently be routed through tls_, tracked separately
99 : : // from tls_.has_value() because the stream now outlives the connection - see
100 : : // close_socket(). Strand-confined, like the stream itself.
101 : : bool tls_engaged_ = false;
102 : : #endif
103 : :
104 : : // True when this connection is encrypted. Reads to it happen on the strand.
105 : 1527 : bool tls_active() const {
106 : : #ifdef WIRESTEAD_TLS_ENABLED
107 : 1527 : return tls_engaged_;
108 : : #else
109 : : return false;
110 : : #endif
111 : : }
112 : : // Guards the mutable subset of cfg_ (retry_interval_ms, max_retries,
113 : : // connection_timeout_ms, idle_timeout_ms, idle_timeout_action) and
114 : : // reconnect_policy_ below - the fields that have runtime setters
115 : : // (set_retry_interval() etc.) reachable from any user thread while the
116 : : // strand concurrently reads them for reconnect/idle-timeout decisions.
117 : : // Fields with no runtime setter (tcp_no_delay, keep_alive, buffer sizes)
118 : : // are set once at construction and read only at connect time, so they
119 : : // don't need this lock (#436).
120 : : mutable std::mutex cfg_mtx_;
121 : : TcpClientConfig cfg_;
122 : : // #443: per-channel pool instead of the process-wide GlobalMemoryPool
123 : : // singleton - avoids cross-channel contention on the singleton's bucket
124 : : // mutexes. Capacity is much smaller than the old shared default (400/2000)
125 : : // since it's no longer amortized across every channel in the process.
126 : : // Prefill stays 0. This literal was written while MemoryPool discarded
127 : : // initial_pool_size, so 50 allocated nothing; #575 made the parameter real
128 : : // and turned it into ~1 MiB eagerly allocated per channel at construction.
129 : : // The pool fills as buffers are released.
130 : : memory::MemoryPool pool_{0, 200};
131 : : net::steady_timer retry_timer_;
132 : : net::steady_timer connect_timer_;
133 : : net::steady_timer idle_timer_;
134 : : bool owns_ioc_ = true;
135 : : std::atomic<bool> stop_requested_{false};
136 : : std::atomic<bool> stopping_{false};
137 : : std::atomic<bool> terminal_state_notified_{false};
138 : : std::atomic<bool> reconnect_pending_{false};
139 : :
140 : : // Sized from cfg_.read_buffer_size in init() rather than being a fixed
141 : : // std::array, so a bulk-transfer workload can trade memory for fewer read
142 : : // completions and callback dispatches.
143 : : std::vector<uint8_t> rx_;
144 : : std::deque<BufferVariant> tx_;
145 : : std::deque<BufferVariant> pending_;
146 : : std::atomic<size_t> pending_bytes_{0};
147 : : // Buffers handed to the in-flight write. Several at a time rather than one:
148 : : // a backlog of queued messages used to cost one send syscall each. `views_`
149 : : // points into `current_write_batch_`, so both stay untouched for the whole
150 : : // async_write - `writing_` is what guarantees that.
151 : : std::vector<BufferVariant> current_write_batch_;
152 : : std::vector<net::const_buffer> current_write_views_;
153 : : bool writing_ = false;
154 : : std::atomic<size_t> queue_bytes_{0};
155 : : // Bytes accepted by a plain async_write_* call but not yet routed onto the
156 : : // strand - reserved via try_reserve_limit_bytes() to close the
157 : : // accept-then-drop race (jwsung91/wirestead#517). inflight_bytes_ mutations
158 : : // and the queue_bytes_/pending_bytes_ increments that promote a
159 : : // reservation both go through write_reserve_mtx_ - see bp_utils.hpp.
160 : : std::atomic<size_t> inflight_bytes_{0};
161 : : std::mutex write_reserve_mtx_;
162 : : // Atomic rather than mutex-guarded: read both from the strand and from
163 : : // arbitrary caller threads (async_try_write_* fast-fail prechecks) (#436).
164 : : std::atomic<base::constants::BackpressureStrategy> bp_strategy_{base::constants::BackpressureStrategy::Reliable};
165 : : size_t bp_high_;
166 : : size_t bp_low_;
167 : : size_t bp_limit_;
168 : : std::atomic<bool> backpressure_active_{false};
169 : : diagnostics::RuntimeStatsCounters stats_;
170 : : unsigned first_retry_interval_ms_ = 100;
171 : :
172 : : // Shared snapshots rather than plain std::functions: the io thread copies
173 : : // one out per received chunk, and a std::function copy allocates whenever
174 : : // the target outgrows its small-object buffer. See interface::SharedCallback.
175 : : interface::SharedCallback<OnBytes> on_bytes_;
176 : : interface::SharedCallback<OnState> on_state_;
177 : : interface::SharedCallback<OnBackpressure> on_bp_;
178 : : mutable std::mutex callback_mtx_;
179 : : std::atomic<bool> connected_{false};
180 : 137 : AtomicLinkState state_{LinkState::Idle};
181 : : int retry_attempts_ = 0;
182 : : uint32_t reconnect_attempt_count_{0};
183 : : std::optional<ReconnectPolicy> reconnect_policy_;
184 : :
185 : : ErrorInfoHolder error_info_holder_{"tcp_client"};
186 : :
187 : 137 : Impl(const TcpClientConfig& cfg, net::io_context* ioc_ptr)
188 [ + + ]: 137 : : owned_ioc_(ioc_ptr ? nullptr : std::make_shared<net::io_context>()),
189 [ + + ]: 137 : ioc_(ioc_ptr ? ioc_ptr : owned_ioc_.get()),
190 : 137 : strand_(net::make_strand(*ioc_)),
191 : 137 : resolver_(strand_),
192 : 137 : socket_(strand_),
193 : 137 : cfg_(cfg),
194 : 137 : retry_timer_(strand_),
195 : 137 : connect_timer_(strand_),
196 : 137 : idle_timer_(strand_),
197 : 137 : owns_ioc_(!ioc_ptr),
198 : 137 : bp_strategy_(cfg.backpressure_strategy),
199 : 822 : bp_high_(cfg.backpressure_threshold) {
200 : 137 : init();
201 : 137 : }
202 : :
203 : 137 : void init() {
204 : 137 : connected_ = false;
205 : 137 : writing_ = false;
206 : 137 : queue_bytes_ = 0;
207 : 137 : pending_bytes_ = 0;
208 : 137 : cfg_.validate_and_clamp();
209 : 137 : rx_.resize(cfg_.read_buffer_size);
210 : 137 : recalculate_backpressure_bounds();
211 : 137 : first_retry_interval_ms_ = std::min(first_retry_interval_ms_, cfg_.retry_interval_ms);
212 : 137 : }
213 : :
214 : : void do_resolve_connect(std::shared_ptr<TcpClient> self, uint64_t seq);
215 : : void schedule_retry(std::shared_ptr<TcpClient> self, uint64_t seq);
216 : : void start_read(std::shared_ptr<TcpClient> self, uint64_t seq);
217 : : void do_write(std::shared_ptr<TcpClient> self, uint64_t seq);
218 : : void handle_close(std::shared_ptr<TcpClient> self, uint64_t seq, const boost::system::error_code& ec = {});
219 : : void handle_idle_timeout(std::shared_ptr<TcpClient> self, uint64_t seq);
220 : : void transition_to(LinkState next, const boost::system::error_code& ec = {});
221 : : void perform_stop_cleanup();
222 : : void reset_start_state();
223 : : void join_ioc_thread(bool allow_detach);
224 : : void close_socket();
225 : : void recalculate_backpressure_bounds();
226 : : void report_backpressure(std::shared_ptr<TcpClient> self, size_t queued_bytes);
227 : : void observe_queue();
228 : : // Shared decide_enqueue()/route dispatch used by all 3 async_write_* variants (#434).
229 : : // `reserved` tells this whether the caller reserved `added` bytes into
230 : : // inflight_bytes_ via try_reserve_limit_bytes() - only Reliable-strategy
231 : : // sends do (jwsung91/wirestead#517); BestEffort's plain path has no
232 : : // precheck and relies entirely on decide_enqueue()'s own keep-latest trim.
233 : : void route_enqueued_buffer(std::shared_ptr<TcpClient> self, BufferVariant&& buf, size_t added, bool reserved);
234 : : queue_util::BackpressureFields bp_fields();
235 : : void notify_state();
236 : : void reset_io_objects();
237 : : void apply_socket_options();
238 : : void handshake_then(std::shared_ptr<TcpClient> self, uint64_t seq, std::function<void()> next);
239 : : void finish_connect(std::shared_ptr<TcpClient> self, uint64_t seq);
240 : : void reset_idle_timer(std::shared_ptr<TcpClient> self, uint64_t seq);
241 : : void cancel_idle_timer();
242 : : void record_error(diagnostics::ErrorLevel lvl, diagnostics::ErrorCategory cat, std::string_view operation,
243 : : const boost::system::error_code& ec, std::string_view msg, bool retryable, uint32_t retry_count);
244 : : };
245 : :
246 : 100 : std::shared_ptr<TcpClient> TcpClient::create(const TcpClientConfig& cfg) {
247 : 100 : return std::shared_ptr<TcpClient>(new TcpClient(cfg));
248 : : }
249 : :
250 : 37 : std::shared_ptr<TcpClient> TcpClient::create(const TcpClientConfig& cfg, boost::asio::io_context& ioc) {
251 : 37 : return std::shared_ptr<TcpClient>(new TcpClient(cfg, ioc));
252 : : }
253 : :
254 : 100 : TcpClient::TcpClient(const TcpClientConfig& cfg) : impl_(std::make_unique<Impl>(cfg, nullptr)) {}
255 : 37 : TcpClient::TcpClient(const TcpClientConfig& cfg, boost::asio::io_context& ioc)
256 : 37 : : impl_(std::make_unique<Impl>(cfg, &ioc)) {}
257 : :
258 : 273 : TcpClient::~TcpClient() {
259 : : // #446: null after being moved-from - the move ctor/assignment are
260 : : // defaulted, and destroying a moved-from instance must not dereference
261 : : // a null impl_ (matches TcpServer/Serial/UdpChannel/UdsServer's
262 : : // destructors, which already guard this way).
263 [ + + ]: 137 : if (!impl_) return;
264 : 136 : stop();
265 : 136 : impl_->join_ioc_thread(true);
266 : :
267 : 136 : impl_->on_bytes_ = nullptr;
268 : 136 : impl_->on_state_ = nullptr;
269 : 136 : impl_->on_bp_ = nullptr;
270 : 275 : }
271 : :
272 : 1 : TcpClient::TcpClient(TcpClient&&) noexcept = default;
273 : 0 : TcpClient& TcpClient::operator=(TcpClient&&) noexcept = default;
274 : :
275 : 2 : std::optional<diagnostics::ErrorInfo> TcpClient::last_error_info() const {
276 : 2 : return impl_->error_info_holder_.last_error_info();
277 : : }
278 : :
279 : 148 : void TcpClient::start() {
280 : 148 : auto current_state = impl_->state_.get();
281 [ + - - + ]: 148 : if (current_state == LinkState::Connecting || current_state == LinkState::Connected) {
282 : 0 : WIRESTEAD_LOG_DEBUG("tcp_client", "start", "Start called while already active, ignoring");
283 : 0 : return;
284 : : }
285 : :
286 [ - + ]: 148 : if (!impl_->ioc_) {
287 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "start", "io_context is null");
288 : : }
289 : :
290 : 148 : impl_->recalculate_backpressure_bounds();
291 : :
292 [ + - + - : 148 : if (impl_->ioc_ && impl_->ioc_->stopped()) {
+ + + + ]
293 : 21 : WIRESTEAD_LOG_DEBUG("tcp_client", "start", "io_context stopped; restarting before start");
294 : 21 : impl_->ioc_->restart();
295 : : }
296 : :
297 [ - + ]: 148 : if (impl_->ioc_thread_.joinable()) {
298 : 0 : impl_->join_ioc_thread(false);
299 : : }
300 : :
301 : 148 : const auto seq = impl_->lifecycle_seq_.fetch_add(1) + 1;
302 : 148 : impl_->current_seq_.store(seq);
303 : :
304 [ + + + - : 148 : if (impl_->owns_ioc_ && impl_->ioc_) {
+ + ]
305 : 116 : impl_->work_guard_ =
306 : 232 : std::make_unique<net::executor_work_guard<net::io_context::executor_type>>(impl_->ioc_->get_executor());
307 : 232 : impl_->ioc_thread_ = std::jthread([ioc = impl_->owned_ioc_](std::stop_token st) {
308 : 116 : wirestead::concurrency::run_io_thread_init();
309 : : try {
310 : 116 : std::stop_callback cb(st, [ioc] { ioc->stop(); });
311 : 116 : ioc->run();
312 : 116 : } catch (const std::exception& e) {
313 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "io_context", fmt::format("IO context error: {}", e.what()));
314 : 0 : diagnostics::error_reporting::report_system_error("tcp_client", "io_context",
315 : 0 : fmt::format("Exception in IO context: {}", e.what()));
316 : 0 : }
317 : 232 : });
318 : : }
319 : :
320 : 148 : auto weak_self = weak_from_this();
321 [ + - ]: 148 : if (impl_->ioc_) {
322 : 148 : net::dispatch(impl_->strand_, [weak_self, seq] {
323 [ + - ]: 145 : if (auto self = weak_self.lock()) {
324 [ + + ]: 290 : if (seq <= self->impl_->stop_seq_.load()) {
325 : 2 : return;
326 : : }
327 : 143 : self->impl_->reset_start_state();
328 : 143 : self->impl_->connected_.store(false);
329 : 143 : self->impl_->reset_io_objects();
330 : 143 : self->impl_->transition_to(LinkState::Connecting);
331 : 143 : self->impl_->do_resolve_connect(self, seq);
332 : 145 : }
333 : : });
334 : : } else {
335 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "start", "io_context is null");
336 : : }
337 : 148 : }
338 : :
339 : 298 : void TcpClient::stop() {
340 [ + + ]: 298 : if (impl_->stop_requested_.exchange(true)) {
341 : 142 : return;
342 : : }
343 : :
344 : 156 : impl_->stopping_.store(true);
345 : 312 : impl_->stop_seq_.store(impl_->current_seq_.load());
346 [ - + ]: 156 : if (!impl_->ioc_) {
347 : 0 : return;
348 : : }
349 : :
350 : : // Post via a raw Impl* rather than weak_from_this().lock(): when stop()
351 : : // runs from ~TcpClient(), the shared_ptr use count is already 0, so that
352 : : // lock() is guaranteed null (standard shared_ptr/enable_shared_from_this
353 : : // behavior during destruction) and perform_stop_cleanup() - which resets
354 : : // work_guard_ - would never be posted, leaving join_ioc_thread() below
355 : : // blocked forever with no work_guard reset to let io_context::run()
356 : : // return. impl_ itself stays alive until after join_ioc_thread() returns
357 : : // (~TcpClient() doesn't destroy it until its body finishes), so capturing
358 : : // the raw Impl* is safe in both the destructor and non-destructor paths.
359 : 156 : Impl* impl_ptr = impl_.get();
360 : 280 : net::post(impl_->strand_, [impl_ptr]() { impl_ptr->perform_stop_cleanup(); });
361 : :
362 : 156 : impl_->join_ioc_thread(false);
363 : : }
364 : :
365 : 5588 : bool TcpClient::is_connected() const { return get_impl()->connected_.load(); }
366 : 1751 : bool TcpClient::is_backpressure_active() const { return get_impl()->backpressure_active_.load(); }
367 : 9 : wrapper::RuntimeStats TcpClient::stats() const {
368 : 36 : return impl_->stats_.snapshot(impl_->queue_bytes_.load(std::memory_order_relaxed),
369 : 9 : impl_->pending_bytes_.load(std::memory_order_relaxed),
370 : 18 : impl_->backpressure_active_.load(std::memory_order_relaxed));
371 : : }
372 : 0 : void TcpClient::reset_stats() {
373 : 0 : impl_->stats_.reset(impl_->queue_bytes_.load(std::memory_order_relaxed) +
374 : 0 : impl_->pending_bytes_.load(std::memory_order_relaxed));
375 : 0 : }
376 : :
377 : 92 : boost::asio::any_io_executor TcpClient::get_executor() { return impl_->socket_.get_executor(); }
378 : :
379 : 1255 : bool TcpClient::async_write_copy(memory::ConstByteSpan data) {
380 [ + - + - ]: 3763 : if (impl_->stop_requested_.load() || impl_->state_.is_state(LinkState::Closed) ||
381 [ + + - + : 3763 : impl_->state_.is_state(LinkState::Error) || !impl_->ioc_) {
+ + ]
382 : 1 : impl_->stats_.record_failed_send();
383 : 1 : return false;
384 : : }
385 : :
386 : 1254 : size_t size = data.size();
387 [ + + ]: 1254 : if (size == 0) {
388 : 2 : WIRESTEAD_LOG_WARNING("tcp_client", "async_write_copy", "Ignoring zero-length write");
389 : 2 : impl_->stats_.record_failed_send();
390 : 2 : return false;
391 : : }
392 : :
393 [ + + ]: 1252 : if (size > base::constants::MAX_BUFFER_SIZE) {
394 : 1 : WIRESTEAD_LOG_ERROR("tcp_client", "async_write_copy",
395 : : fmt::format("Write size exceeds maximum allowed ({} bytes)", size));
396 : 1 : impl_->stats_.record_failed_send();
397 : 1 : return false;
398 : : }
399 : :
400 [ + + + - : 1251 : if (size <= 65536 && impl_->cfg_.enable_memory_pool) {
+ + ]
401 : : try {
402 : 1249 : memory::PooledBuffer pooled_buffer(size, impl_->pool_);
403 [ + - + - ]: 1249 : if (pooled_buffer.valid()) {
404 : 1249 : base::safe_memory::safe_memcpy(pooled_buffer.data(), data.data(), size);
405 : 1249 : const auto added = pooled_buffer.size();
406 : 1249 : const bool reliable = impl_->bp_strategy_ == base::constants::BackpressureStrategy::Reliable;
407 [ + + - + ]: 2487 : if (reliable &&
408 [ + - - + ]: 1238 : !queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
409 : 1238 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
410 : 0 : impl_->stats_.record_failed_send();
411 : 0 : return false;
412 : : }
413 : 1249 : impl_->stats_.record_accepted(added);
414 : 1249 : net::dispatch(impl_->strand_,
415 : 2498 : [self = shared_from_this(), buf = std::move(pooled_buffer), added, reliable]() mutable {
416 : 1248 : self->impl_->route_enqueued_buffer(self, BufferVariant{std::move(buf)}, added, reliable);
417 : 1248 : });
418 : 1249 : return true;
419 : : }
420 : 1249 : } catch (const std::exception& e) {
421 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "async_write_copy",
422 : : fmt::format("Failed to acquire pooled buffer: {}", e.what()));
423 : 0 : }
424 : : }
425 : :
426 : 2 : std::vector<uint8_t> fallback(data.begin(), data.end());
427 : 2 : const auto added = fallback.size();
428 : 2 : const bool reliable = impl_->bp_strategy_ == base::constants::BackpressureStrategy::Reliable;
429 [ + - + + ]: 4 : if (reliable &&
430 [ + - + + ]: 2 : !queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
431 : 2 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
432 : 1 : impl_->stats_.record_failed_send();
433 : 1 : return false;
434 : : }
435 : 1 : impl_->stats_.record_accepted(added);
436 : :
437 : 1 : net::dispatch(impl_->strand_, [self = shared_from_this(), buf = std::move(fallback), added, reliable]() mutable {
438 : 1 : self->impl_->route_enqueued_buffer(self, BufferVariant{std::move(buf)}, added, reliable);
439 : 1 : });
440 : 1 : return true;
441 : 2 : }
442 : :
443 : 519 : bool TcpClient::async_write_move(std::vector<uint8_t>&& data) {
444 [ + - + - ]: 1555 : if (impl_->stop_requested_.load() || impl_->state_.is_state(LinkState::Closed) ||
445 [ + + - + : 1555 : impl_->state_.is_state(LinkState::Error) || !impl_->ioc_) {
+ + ]
446 : 1 : impl_->stats_.record_failed_send();
447 : 1 : return false;
448 : : }
449 : 518 : const auto size = data.size();
450 [ + + ]: 518 : if (size == 0) {
451 : 1 : WIRESTEAD_LOG_WARNING("tcp_client", "async_write_move", "Ignoring zero-length write");
452 : 1 : impl_->stats_.record_failed_send();
453 : 1 : return false;
454 : : }
455 [ + + ]: 517 : if (size > base::constants::MAX_BUFFER_SIZE) {
456 : 1 : WIRESTEAD_LOG_ERROR("tcp_client", "async_write_move",
457 : : fmt::format("Write size exceeds maximum allowed ({} bytes)", size));
458 : 1 : impl_->stats_.record_failed_send();
459 : 1 : return false;
460 : : }
461 : :
462 : 516 : const auto added = size;
463 : 516 : const bool reliable = impl_->bp_strategy_ == base::constants::BackpressureStrategy::Reliable;
464 [ + + - + ]: 1031 : if (reliable &&
465 [ + - - + ]: 515 : !queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
466 : 515 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
467 : 0 : impl_->stats_.record_failed_send();
468 : 0 : return false;
469 : : }
470 : 516 : impl_->stats_.record_accepted(added);
471 : 516 : net::dispatch(impl_->strand_, [self = shared_from_this(), buf = std::move(data), added, reliable]() mutable {
472 : 516 : self->impl_->route_enqueued_buffer(self, BufferVariant{std::move(buf)}, added, reliable);
473 : 516 : });
474 : 516 : return true;
475 : : }
476 : :
477 : 6 : bool TcpClient::async_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
478 [ + - + - ]: 16 : if (impl_->stop_requested_.load() || impl_->state_.is_state(LinkState::Closed) ||
479 [ + + - + : 16 : impl_->state_.is_state(LinkState::Error) || !impl_->ioc_) {
+ + ]
480 : 1 : impl_->stats_.record_failed_send();
481 : 1 : return false;
482 : : }
483 [ + + + + : 5 : if (!data || data->empty()) {
+ + ]
484 : 2 : WIRESTEAD_LOG_WARNING("tcp_client", "async_write_shared", "Ignoring empty shared buffer");
485 : 2 : impl_->stats_.record_failed_send();
486 : 2 : return false;
487 : : }
488 : 3 : const auto size = data->size();
489 [ + + ]: 3 : if (size > base::constants::MAX_BUFFER_SIZE) {
490 : 1 : WIRESTEAD_LOG_ERROR("tcp_client", "async_write_shared",
491 : : fmt::format("Write size exceeds maximum allowed ({} bytes)", size));
492 : 1 : impl_->stats_.record_failed_send();
493 : 1 : return false;
494 : : }
495 : :
496 : 2 : const auto added = size;
497 : 2 : const bool reliable = impl_->bp_strategy_ == base::constants::BackpressureStrategy::Reliable;
498 [ + - - + ]: 4 : if (reliable &&
499 [ + - - + ]: 2 : !queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queue_bytes_, impl_->pending_bytes_,
500 : 2 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
501 : 0 : impl_->stats_.record_failed_send();
502 : 0 : return false;
503 : : }
504 : 2 : impl_->stats_.record_accepted(added);
505 : 2 : net::dispatch(impl_->strand_, [self = shared_from_this(), buf = std::move(data), added, reliable]() mutable {
506 : 2 : self->impl_->route_enqueued_buffer(self, BufferVariant{std::move(buf)}, added, reliable);
507 : 2 : });
508 : 2 : return true;
509 : : }
510 : :
511 : 2001 : bool TcpClient::async_try_write_copy(memory::ConstByteSpan data) {
512 [ - + ]: 2001 : if (data.empty()) {
513 : 0 : impl_->stats_.record_failed_send();
514 : 0 : return false;
515 : : }
516 [ - + ]: 2001 : if (data.size() > base::constants::MAX_BUFFER_SIZE) {
517 : 0 : impl_->stats_.record_failed_send();
518 : 0 : return false;
519 : : }
520 : 6003 : return async_try_write_move(std::vector<uint8_t>(data.begin(), data.end()));
521 : : }
522 : :
523 : 2004 : bool TcpClient::async_try_write_move(std::vector<uint8_t>&& data) {
524 [ + - + - ]: 6012 : if (impl_->stop_requested_.load() || impl_->state_.is_state(LinkState::Closed) ||
525 [ + - - + : 6012 : impl_->state_.is_state(LinkState::Error) || !impl_->ioc_) {
- + ]
526 : 0 : impl_->stats_.record_failed_send();
527 : 0 : return false;
528 : : }
529 : 2004 : const auto added = data.size();
530 [ + - - + ]: 2004 : if (added == 0 || added > base::constants::MAX_BUFFER_SIZE) {
531 : 0 : impl_->stats_.record_failed_send();
532 : 0 : return false;
533 : : }
534 : 513 : const auto reject_for_pressure = [this, added]() {
535 [ + + ]: 513 : if (impl_->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
536 : 1 : impl_->stats_.record_dropped(1, added);
537 : : } else {
538 : 512 : impl_->stats_.record_failed_send();
539 : : }
540 : 2517 : };
541 [ + + + - : 3495 : if (impl_->backpressure_active_.load() || impl_->queue_bytes_ + added > impl_->bp_high_ ||
+ + ]
542 [ - + ]: 1491 : impl_->queue_bytes_ + impl_->pending_bytes_ + added > impl_->bp_limit_) {
543 : 513 : reject_for_pressure();
544 : 513 : return false;
545 : : }
546 [ - + ]: 1491 : if (!queue_util::try_reserve_write_bytes(impl_->queue_bytes_, impl_->pending_bytes_, impl_->backpressure_active_,
547 : 1491 : added, impl_->bp_high_, impl_->bp_limit_)) {
548 : 0 : reject_for_pressure();
549 : 0 : return false;
550 : : }
551 : 1491 : impl_->stats_.record_accepted(added);
552 : :
553 : 1491 : net::dispatch(impl_->strand_, [self = shared_from_this(), buf = std::move(data), added]() mutable {
554 : 1491 : auto impl = self->impl_.get();
555 [ + - + - : 2982 : if (impl->stop_requested_.load() || impl->state_.is_state(LinkState::Closed) ||
- + ]
556 : 2982 : impl->state_.is_state(LinkState::Error)) {
557 : 0 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, added);
558 : 0 : impl->stats_.record_failed_send();
559 : 0 : return;
560 : : }
561 : :
562 : 1491 : impl->tx_.emplace_back(std::move(buf));
563 : 1491 : impl->observe_queue();
564 : 1491 : impl->report_backpressure(self, impl->queue_bytes_);
565 [ + + + - ]: 1493 : if (!impl->writing_) impl->do_write(self, impl->current_seq_.load());
566 : : });
567 : 1491 : return true;
568 : : }
569 : :
570 : 1 : bool TcpClient::async_try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
571 [ + - - + : 1 : if (!data || data->empty()) {
- + ]
572 : 0 : impl_->stats_.record_failed_send();
573 : 0 : return false;
574 : : }
575 [ - + ]: 1 : if (data->size() > base::constants::MAX_BUFFER_SIZE) {
576 : 0 : impl_->stats_.record_failed_send();
577 : 0 : return false;
578 : : }
579 : 1 : const auto added = data->size();
580 : 1 : const auto reject_for_pressure = [this, added]() {
581 [ - + ]: 1 : if (impl_->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
582 : 0 : impl_->stats_.record_dropped(1, added);
583 : : } else {
584 : 1 : impl_->stats_.record_failed_send();
585 : : }
586 : 2 : };
587 [ + - + - ]: 3 : if (impl_->stop_requested_.load() || impl_->state_.is_state(LinkState::Closed) ||
588 [ + - - + : 3 : impl_->state_.is_state(LinkState::Error) || !impl_->ioc_) {
- + ]
589 : 0 : impl_->stats_.record_failed_send();
590 : 0 : return false;
591 : : }
592 [ - + - - : 1 : if (impl_->backpressure_active_.load() || impl_->queue_bytes_ + added > impl_->bp_high_ ||
+ - ]
593 [ # # ]: 0 : impl_->queue_bytes_ + impl_->pending_bytes_ + added > impl_->bp_limit_) {
594 : 1 : reject_for_pressure();
595 : 1 : return false;
596 : : }
597 [ # # ]: 0 : if (!queue_util::try_reserve_write_bytes(impl_->queue_bytes_, impl_->pending_bytes_, impl_->backpressure_active_,
598 : 0 : added, impl_->bp_high_, impl_->bp_limit_)) {
599 : 0 : reject_for_pressure();
600 : 0 : return false;
601 : : }
602 : 0 : impl_->stats_.record_accepted(added);
603 : :
604 : 0 : net::dispatch(impl_->strand_, [self = shared_from_this(), buf = std::move(data), added]() mutable {
605 : 0 : auto impl = self->impl_.get();
606 [ # # # # : 0 : if (impl->stop_requested_.load() || impl->state_.is_state(LinkState::Closed) ||
# # ]
607 : 0 : impl->state_.is_state(LinkState::Error)) {
608 : 0 : queue_util::release_reserved_write_bytes(impl->queue_bytes_, added);
609 : 0 : impl->stats_.record_failed_send();
610 : 0 : return;
611 : : }
612 : :
613 : 0 : impl->tx_.emplace_back(std::move(buf));
614 : 0 : impl->observe_queue();
615 : 0 : impl->report_backpressure(self, impl->queue_bytes_);
616 [ # # # # ]: 0 : if (!impl->writing_) impl->do_write(self, impl->current_seq_.load());
617 : : });
618 : 0 : return true;
619 : : }
620 : :
621 : : // Each setter builds the shared snapshot before taking the lock, so the
622 : : // allocation stays outside the critical section the io thread contends on.
623 : 189 : void TcpClient::on_bytes(OnBytes cb) {
624 : 189 : auto shared = interface::share_callback(std::move(cb));
625 : 189 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
626 : 189 : impl_->on_bytes_ = std::move(shared);
627 : 189 : }
628 : 218 : void TcpClient::on_state(OnState cb) {
629 : 218 : auto shared = interface::share_callback(std::move(cb));
630 : 218 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
631 : 218 : impl_->on_state_ = std::move(shared);
632 : 218 : }
633 : 198 : void TcpClient::on_backpressure(OnBackpressure cb) {
634 : 198 : auto shared = interface::share_callback(std::move(cb));
635 : 198 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
636 : 198 : impl_->on_bp_ = std::move(shared);
637 : 198 : }
638 : 1 : void TcpClient::set_backpressure_strategy(base::constants::BackpressureStrategy strategy) {
639 : 1 : impl_->bp_strategy_.store(strategy, std::memory_order_relaxed);
640 : 1 : }
641 : :
642 : 3 : void TcpClient::set_retry_interval(unsigned interval_ms) {
643 : 3 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
644 : 3 : impl_->cfg_.retry_interval_ms = interval_ms;
645 : 3 : impl_->cfg_.validate_and_clamp();
646 : 3 : }
647 : 2 : void TcpClient::set_max_retries(int max_retries) {
648 : 2 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
649 : 2 : impl_->cfg_.max_retries = max_retries;
650 : 2 : impl_->cfg_.validate_and_clamp();
651 : 2 : }
652 : 2 : void TcpClient::set_connection_timeout(unsigned timeout_ms) {
653 : 2 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
654 : 2 : impl_->cfg_.connection_timeout_ms = timeout_ms;
655 : 2 : impl_->cfg_.validate_and_clamp();
656 : 2 : }
657 : 1 : void TcpClient::set_idle_timeout(unsigned timeout_ms) {
658 : 1 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
659 : 1 : impl_->cfg_.idle_timeout_ms = timeout_ms;
660 : 1 : impl_->cfg_.validate_and_clamp();
661 : 1 : }
662 : 1 : void TcpClient::set_idle_timeout_action(IdleTimeoutAction action) {
663 : 1 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
664 : 1 : impl_->cfg_.idle_timeout_action = action;
665 : 1 : }
666 : 8 : void TcpClient::set_reconnect_policy(ReconnectPolicy policy) {
667 : 8 : std::lock_guard<std::mutex> lock(impl_->cfg_mtx_);
668 [ + + ]: 8 : if (policy) {
669 : 7 : impl_->reconnect_policy_ = std::move(policy);
670 : : } else {
671 : 1 : impl_->reconnect_policy_ = std::nullopt;
672 : : }
673 : 8 : }
674 : :
675 : : // Impl methods implementation
676 : :
677 : 148 : void TcpClient::Impl::apply_socket_options() {
678 : 148 : boost::system::error_code ec;
679 : :
680 [ + - ]: 148 : if (cfg_.tcp_no_delay) {
681 : 148 : socket_.set_option(tcp::no_delay(true), ec);
682 [ - + ]: 148 : if (ec) {
683 : 0 : WIRESTEAD_LOG_WARNING("tcp_client", "socket_options", fmt::format("Failed to set TCP_NODELAY: {}", ec.message()));
684 : 0 : ec.clear();
685 : : }
686 : : }
687 : :
688 [ - + ]: 148 : if (cfg_.keep_alive) {
689 : 0 : socket_.set_option(net::socket_base::keep_alive(true), ec);
690 [ # # ]: 0 : if (ec) {
691 : 0 : WIRESTEAD_LOG_WARNING("tcp_client", "socket_options", fmt::format("Failed to set keep_alive: {}", ec.message()));
692 : 0 : ec.clear();
693 : : }
694 : : }
695 : :
696 [ - + ]: 148 : if (cfg_.send_buffer_size > 0) {
697 : 0 : socket_.set_option(net::socket_base::send_buffer_size(static_cast<int>(cfg_.send_buffer_size)), ec);
698 [ # # ]: 0 : if (ec) {
699 : 0 : WIRESTEAD_LOG_WARNING("tcp_client", "socket_options",
700 : : fmt::format("Failed to set send buffer size: {}", ec.message()));
701 : 0 : ec.clear();
702 : : }
703 : : }
704 : :
705 [ - + ]: 148 : if (cfg_.receive_buffer_size > 0) {
706 : 0 : socket_.set_option(net::socket_base::receive_buffer_size(static_cast<int>(cfg_.receive_buffer_size)), ec);
707 [ # # ]: 0 : if (ec) {
708 : 0 : WIRESTEAD_LOG_WARNING("tcp_client", "socket_options",
709 : : fmt::format("Failed to set receive buffer size: {}", ec.message()));
710 : 0 : ec.clear();
711 : : }
712 : : }
713 : 148 : }
714 : :
715 : 211 : void TcpClient::Impl::do_resolve_connect(std::shared_ptr<TcpClient> self, uint64_t seq) {
716 : 211 : resolver_.async_resolve(
717 : 422 : cfg_.host, fmt::format("{}", cfg_.port), [self, seq](auto ec, tcp::resolver::results_type results) {
718 [ + - - + : 416 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) {
- + ]
719 : 3 : return;
720 : : }
721 [ + + - + : 208 : if (self->impl_->stop_requested_.load() || self->impl_->stopping_.load()) {
+ + ]
722 : 2 : return;
723 : : }
724 [ + + ]: 206 : if (ec) {
725 : : bool has_policy;
726 : : {
727 : 1 : std::lock_guard<std::mutex> lock(self->impl_->cfg_mtx_);
728 : 1 : has_policy = self->impl_->reconnect_policy_.has_value();
729 : 1 : }
730 : 1 : uint32_t current_attempts =
731 [ - + ]: 1 : has_policy ? self->impl_->reconnect_attempt_count_ : static_cast<uint32_t>(self->impl_->retry_attempts_);
732 : 2 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "resolve",
733 : : ec, fmt::format("Resolution failed: {}", ec.message()),
734 : 1 : diagnostics::is_retryable_tcp_connect_error(ec), current_attempts);
735 : 1 : self->impl_->schedule_retry(self, seq);
736 : 1 : return;
737 : : }
738 : : unsigned connection_timeout_ms;
739 : : {
740 : 205 : std::lock_guard<std::mutex> lock(self->impl_->cfg_mtx_);
741 : 205 : connection_timeout_ms = self->impl_->cfg_.connection_timeout_ms;
742 : 205 : }
743 : 205 : self->impl_->connect_timer_.expires_after(std::chrono::milliseconds(connection_timeout_ms));
744 : 205 : self->impl_->connect_timer_.async_wait([self, seq](const boost::system::error_code& timer_ec) {
745 [ + + - + : 206 : if (timer_ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) {
+ + ]
746 : 202 : return;
747 : : }
748 [ + - + - : 2 : if (!timer_ec && !self->impl_->stop_requested_.load() && !self->impl_->stopping_.load()) {
+ - + - ]
749 : : bool has_policy;
750 : : unsigned timeout_ms;
751 : : {
752 : 2 : std::lock_guard<std::mutex> lock(self->impl_->cfg_mtx_);
753 : 2 : has_policy = self->impl_->reconnect_policy_.has_value();
754 : 2 : timeout_ms = self->impl_->cfg_.connection_timeout_ms;
755 : 2 : }
756 : 2 : WIRESTEAD_LOG_ERROR("tcp_client", "connect_timeout",
757 : : fmt::format("Connection timed out after {}ms", timeout_ms));
758 [ - + ]: 2 : uint32_t current_attempts = has_policy ? self->impl_->reconnect_attempt_count_
759 : 2 : : static_cast<uint32_t>(self->impl_->retry_attempts_);
760 : 4 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "connect",
761 : : boost::asio::error::timed_out, "Connection timed out",
762 : 2 : diagnostics::is_retryable_tcp_connect_error(boost::asio::error::timed_out),
763 : : current_attempts);
764 : 2 : self->impl_->handle_close(self, seq, boost::asio::error::timed_out);
765 : : }
766 : : });
767 : :
768 : 205 : net::async_connect(self->impl_->socket_, results, [self, seq](auto ec2, const auto&) {
769 [ + + - + : 408 : if (ec2 == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) {
+ + ]
770 : 2 : return;
771 : : }
772 [ + - - + : 203 : if (self->impl_->stop_requested_.load() || self->impl_->stopping_.load()) {
- + ]
773 : 0 : self->impl_->close_socket();
774 : 0 : self->impl_->connect_timer_.cancel();
775 : 0 : return;
776 : : }
777 [ + + ]: 203 : if (ec2) {
778 : 55 : self->impl_->connect_timer_.cancel();
779 : : bool has_policy;
780 : : {
781 : 55 : std::lock_guard<std::mutex> lock(self->impl_->cfg_mtx_);
782 : 55 : has_policy = self->impl_->reconnect_policy_.has_value();
783 : 55 : }
784 [ + + ]: 55 : uint32_t current_attempts = has_policy ? self->impl_->reconnect_attempt_count_
785 : 28 : : static_cast<uint32_t>(self->impl_->retry_attempts_);
786 : 110 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "connect",
787 : : ec2, fmt::format("Connection failed: {}", ec2.message()),
788 : 55 : diagnostics::is_retryable_tcp_connect_error(ec2), current_attempts);
789 : 55 : self->impl_->schedule_retry(self, seq);
790 : 55 : return;
791 : : }
792 : 148 : self->impl_->connect_timer_.cancel();
793 : 148 : self->impl_->retry_attempts_ = 0;
794 : 148 : self->impl_->reconnect_attempt_count_ = 0;
795 : :
796 : : #if defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__)
797 : : int yes = 1;
798 : : (void)::setsockopt(static_cast<int>(self->impl_->socket_.native_handle()), SOL_SOCKET, SO_NOSIGPIPE, &yes,
799 : : static_cast<socklen_t>(sizeof(yes)));
800 : : #endif
801 : :
802 : 148 : self->impl_->apply_socket_options();
803 : :
804 : : // TCP is up; TLS still has to prove who answered. Reading before the
805 : : // handshake completes would hand the session ciphertext, and a failed
806 : : // handshake must not be reported as a working connection - so the
807 : : // rest of the connect path waits behind it. Plaintext runs the
808 : : // continuation immediately, which is what it did before TLS existed.
809 : 267 : self->impl_->handshake_then(self, seq, [self, seq] { self->impl_->finish_connect(self, seq); });
810 : : });
811 : : });
812 : 211 : }
813 : :
814 : : // Sets up the TLS stream if configured, runs the handshake, then hands control
815 : : // back. Without TLS - or in a build without it - `next` runs straight away and
816 : : // the connect path is byte for byte what it was.
817 : 148 : void TcpClient::Impl::handshake_then(std::shared_ptr<TcpClient> self, uint64_t seq, std::function<void()> next) {
818 : : #ifdef WIRESTEAD_TLS_ENABLED
819 : 148 : bool want_tls = false;
820 : 148 : std::string ca_file;
821 : : {
822 : 148 : std::lock_guard<std::mutex> lock(cfg_mtx_);
823 : 148 : want_tls = cfg_.tls_enabled;
824 : 148 : ca_file = cfg_.tls_ca_file;
825 : 148 : }
826 : :
827 [ + + ]: 148 : if (want_tls) {
828 : : namespace ssl = boost::asio::ssl;
829 : : try {
830 [ + + ]: 30 : if (!ssl_context_) {
831 : 2 : auto ctx = std::make_shared<ssl::context>(ssl::context::tls_client);
832 : 2 : ctx->set_options(ssl::context::default_workarounds | ssl::context::no_sslv2 | ssl::context::no_sslv3 |
833 : : ssl::context::no_tlsv1 | ssl::context::no_tlsv1_1);
834 [ + + ]: 2 : if (ca_file.empty()) {
835 : 1 : ctx->set_default_verify_paths();
836 : : } else {
837 : 1 : ctx->load_verify_file(ca_file);
838 : : }
839 : : // Not optional. An unverified TLS connection encrypts traffic to
840 : : // whoever answered, which is what an attacker in the middle wants.
841 : 2 : ctx->set_verify_mode(ssl::verify_peer);
842 : 2 : ssl_context_ = std::move(ctx);
843 : 2 : }
844 : 30 : tls_.emplace(socket_, *ssl_context_);
845 : 30 : tls_engaged_ = true;
846 : 30 : tls_->set_verify_callback(ssl::host_name_verification(cfg_.host));
847 : : // SNI, and the name the certificate is checked against.
848 : 30 : ::SSL_set_tlsext_host_name(tls_->native_handle(), cfg_.host.c_str());
849 : 0 : } catch (const std::exception& e) {
850 : 0 : const std::string msg = std::string("TLS setup failed: ") + e.what();
851 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "handshake", msg);
852 : 0 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "handshake",
853 : : boost::asio::error::invalid_argument, msg, false, 0);
854 : 0 : tls_.reset();
855 : 0 : tls_engaged_ = false;
856 : 0 : handle_close(self, seq, boost::asio::error::invalid_argument);
857 : 0 : return;
858 : 0 : }
859 : :
860 : 60 : tls_->async_handshake(ssl::stream_base::client,
861 : 60 : net::bind_executor(strand_, [this, self, seq, next](const boost::system::error_code& ec) {
862 [ + - + - : 60 : if (seq != current_seq_.load() || stop_requested_.load() || stopping_.load()) return;
- + - + ]
863 [ + + ]: 30 : if (ec) {
864 : 29 : WIRESTEAD_LOG_ERROR("tcp_client", "handshake", "TLS handshake failed: " + ec.message());
865 : 29 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION,
866 : 58 : "handshake", ec, "TLS handshake failed: " + ec.message(), false, 0);
867 : : // Not tls_.reset(): this runs from inside the
868 : : // stream's own completion handler, and
869 : : // close_socket() below explains why the stream
870 : : // has to outlive its operations.
871 : 29 : tls_engaged_ = false;
872 : 29 : handle_close(self, seq, ec);
873 : 29 : return;
874 : : }
875 : 1 : next();
876 : : }));
877 : 30 : return;
878 : : }
879 : : #else
880 : : (void)self;
881 : : (void)seq;
882 : : #endif
883 : 118 : next();
884 : 148 : }
885 : :
886 : 119 : void TcpClient::Impl::finish_connect(std::shared_ptr<TcpClient> self, uint64_t seq) {
887 : : // Set here rather than at TCP connect: with TLS, a socket whose handshake
888 : : // has not finished is not a usable connection, and connected() is what
889 : : // callers poll before sending. Reporting true for a peer that failed
890 : : // verification would be worse than useless.
891 : 119 : connected_.store(true);
892 : 119 : transition_to(LinkState::Connected);
893 : 119 : boost::system::error_code ep_ec;
894 : 119 : auto rep = socket_.remote_endpoint(ep_ec);
895 [ + - ]: 119 : if (!ep_ec) {
896 : 119 : WIRESTEAD_LOG_INFO("tcp_client", "connect",
897 : : fmt::format("Connected to {}:{}", rep.address().to_string(), rep.port()));
898 : : }
899 : 119 : start_read(self, seq);
900 : 119 : reset_idle_timer(self, seq);
901 : 119 : net::post(strand_, [self, seq]() {
902 : 118 : self->impl_->writing_ = false;
903 : 118 : self->impl_->do_write(self, seq);
904 : 118 : });
905 : 119 : }
906 : :
907 : 99 : void TcpClient::Impl::schedule_retry(std::shared_ptr<TcpClient> self, uint64_t seq) {
908 : 99 : connected_.store(false);
909 [ + - - + : 99 : if (stop_requested_.load() || stopping_.load()) {
- + ]
910 : 9 : return;
911 : : }
912 : :
913 : : // Prevent double scheduling of reconnect
914 [ - + ]: 99 : if (reconnect_pending_.exchange(true)) {
915 : 0 : return;
916 : : }
917 : :
918 : 99 : std::optional<diagnostics::ErrorInfo> last_err = error_info_holder_.last_error_info();
919 : :
920 [ - + ]: 99 : if (!last_err) {
921 : 0 : last_err = diagnostics::ErrorInfo(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION,
922 : : "tcp_client", "schedule_retry", "Unknown error",
923 : 0 : make_error_code(boost::asio::error::not_connected), true);
924 : : }
925 : :
926 : : // Snapshot once rather than locking repeatedly for each read below -
927 : : // cfg_/reconnect_policy_ can change concurrently via set_retry_interval()
928 : : // etc. from any user thread while this runs on the strand (#436).
929 : 99 : TcpClientConfig cfg_snapshot;
930 : 99 : std::optional<ReconnectPolicy> reconnect_policy_snapshot;
931 : : {
932 : 99 : std::lock_guard<std::mutex> lock(cfg_mtx_);
933 : 99 : cfg_snapshot = cfg_;
934 : 99 : reconnect_policy_snapshot = reconnect_policy_;
935 : 99 : }
936 : :
937 : : // Determine current attempt count based on active mode
938 : : uint32_t current_attempts =
939 [ + + ]: 99 : reconnect_policy_snapshot ? reconnect_attempt_count_ : static_cast<uint32_t>(retry_attempts_);
940 : :
941 : 99 : auto decision = detail::decide_reconnect(cfg_snapshot, *last_err, current_attempts, reconnect_policy_snapshot);
942 : :
943 [ + + ]: 99 : if (!decision.should_retry) {
944 : 9 : WIRESTEAD_LOG_INFO("tcp_client", "retry", "Reconnect stopped by policy/config");
945 : 9 : transition_to(LinkState::Error);
946 : 9 : reconnect_pending_.store(false);
947 : 9 : return;
948 : : }
949 : :
950 : : // The decider returns a base delay for both policy and fallback paths.
951 : 90 : std::chrono::milliseconds delay = decision.delay.value_or(std::chrono::milliseconds(cfg_snapshot.retry_interval_ms));
952 [ + + ]: 90 : if (reconnect_policy_snapshot) {
953 : 25 : reconnect_attempt_count_++;
954 : : } else {
955 : : // Preserve existing "fast first retry" behavior for non-policy mode.
956 : 65 : ++retry_attempts_;
957 [ + + ]: 65 : if (retry_attempts_ == 1) {
958 : 56 : delay = std::chrono::milliseconds(first_retry_interval_ms_);
959 : : }
960 : : }
961 : :
962 : 90 : transition_to(LinkState::Connecting);
963 : :
964 : 90 : WIRESTEAD_LOG_INFO("tcp_client", "retry",
965 : : fmt::format("Scheduling retry in {:.3f}s", static_cast<double>(delay.count()) / 1000.0));
966 : :
967 : 90 : retry_timer_.expires_after(delay);
968 : 90 : retry_timer_.async_wait([self, seq](const boost::system::error_code& ec) {
969 : : // Clear pending flag regardless of result (fired or aborted)
970 : 81 : self->impl_->reconnect_pending_.store(false);
971 : :
972 [ + + - + : 149 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) {
+ + ]
973 : 13 : return;
974 : : }
975 [ + - + - : 68 : if (!ec && !self->impl_->stop_requested_.load() && !self->impl_->stopping_.load())
+ - + - ]
976 : 68 : self->impl_->do_resolve_connect(self, seq);
977 : : });
978 : 117 : }
979 : :
980 : 234 : void TcpClient::Impl::start_read(std::shared_ptr<TcpClient> self, uint64_t seq) {
981 : 229 : auto on_read = [self, seq](auto ec, std::size_t n) {
982 [ + + - + : 358 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) {
+ + ]
983 : 114 : return;
984 : : }
985 [ + + ]: 129 : if (self->impl_->stop_requested_.load()) {
986 : 2 : return;
987 : : }
988 [ + + ]: 127 : if (ec) {
989 : 10 : self->impl_->handle_close(self, seq, ec);
990 : 10 : return;
991 : : }
992 [ + - ]: 117 : if (n > 0) {
993 : 117 : self->impl_->reset_idle_timer(self, seq);
994 : : }
995 : 117 : interface::SharedCallback<OnBytes> on_bytes;
996 : : {
997 : 117 : std::lock_guard<std::mutex> lock(self->impl_->callback_mtx_);
998 : 117 : on_bytes = self->impl_->on_bytes_;
999 : 117 : }
1000 : :
1001 : 117 : self->impl_->stats_.record_received(n);
1002 : :
1003 [ + - ]: 117 : if (on_bytes) {
1004 : : try {
1005 : 117 : (*on_bytes)(memory::ConstByteSpan(self->impl_->rx_.data(), n));
1006 : 2 : } catch (const std::exception& e) {
1007 : 1 : WIRESTEAD_LOG_ERROR("tcp_client", "on_bytes", fmt::format("Exception in on_bytes callback: {}", e.what()));
1008 : 2 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::COMMUNICATION, "on_bytes",
1009 : : boost::asio::error::connection_aborted,
1010 : 1 : fmt::format("Exception in on_bytes: {}", e.what()), false, 0);
1011 : 1 : self->impl_->handle_close(self, seq, make_error_code(boost::asio::error::connection_aborted));
1012 : 1 : return;
1013 : 1 : } catch (...) {
1014 : 1 : WIRESTEAD_LOG_ERROR("tcp_client", "on_bytes", "Unknown exception in on_bytes callback");
1015 : 1 : self->impl_->handle_close(self, seq, make_error_code(boost::asio::error::connection_aborted));
1016 : 1 : return;
1017 : : }
1018 : : }
1019 : 115 : self->impl_->start_read(self, seq);
1020 : 351 : };
1021 : : #ifdef WIRESTEAD_TLS_ENABLED
1022 [ + + ]: 234 : if (tls_active()) {
1023 : 2 : tls_->async_read_some(net::buffer(rx_.data(), rx_.size()), std::move(on_read));
1024 : 2 : return;
1025 : : }
1026 : : #endif
1027 : 232 : socket_.async_read_some(net::buffer(rx_.data(), rx_.size()), std::move(on_read));
1028 : 234 : }
1029 : :
1030 : 2570 : void TcpClient::Impl::do_write(std::shared_ptr<TcpClient> self, uint64_t seq) {
1031 [ + + ]: 2570 : if (stop_requested_.load()) {
1032 : 9 : tx_.clear();
1033 : 9 : queue_bytes_ = 0;
1034 : 9 : pending_.clear();
1035 : 9 : pending_bytes_ = 0;
1036 : 9 : writing_ = false;
1037 : 9 : report_backpressure(self, queue_bytes_);
1038 : 1278 : return;
1039 : : }
1040 : :
1041 [ + + ]: 2561 : if (!connected_.load()) {
1042 : 11 : writing_ = false;
1043 : 11 : return;
1044 : : }
1045 : :
1046 [ + + + - : 2550 : if (tx_.empty() || state_.is_state(LinkState::Closed) || state_.is_state(LinkState::Error)) {
- + + + ]
1047 : 1257 : writing_ = false;
1048 : 1257 : return;
1049 : : }
1050 : 1293 : writing_ = true;
1051 : :
1052 : 1293 : const auto queued_bytes = queue_util::take_gather_batch(tx_, current_write_batch_, current_write_views_);
1053 : :
1054 : 1293 : auto on_write = [self, queued_bytes, seq](auto ec, std::size_t bytes_written) {
1055 [ + - - + : 2586 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) {
- + ]
1056 : 0 : self->impl_->current_write_batch_.clear();
1057 : 0 : self->impl_->queue_bytes_ =
1058 [ # # ]: 0 : (self->impl_->queue_bytes_ > queued_bytes) ? (self->impl_->queue_bytes_ - queued_bytes) : 0;
1059 : 0 : self->impl_->report_backpressure(self, self->impl_->queue_bytes_);
1060 : 0 : self->impl_->writing_ = false;
1061 : 0 : return;
1062 : : }
1063 : :
1064 [ - + ]: 1293 : if (ec) {
1065 : 0 : queue_util::return_gather_batch(self->impl_->tx_, self->impl_->current_write_batch_);
1066 : :
1067 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "do_write", fmt::format("Write failed: {}", ec.message()));
1068 : 0 : self->impl_->record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::COMMUNICATION, "write", ec,
1069 : : fmt::format("Write failed: {}", ec.message()), false, 0);
1070 : 0 : self->impl_->writing_ = false;
1071 : 0 : self->impl_->handle_close(self, seq, ec);
1072 : 0 : return;
1073 : : }
1074 : :
1075 : 1293 : self->impl_->current_write_batch_.clear();
1076 : 1293 : self->impl_->stats_.record_sent(bytes_written);
1077 [ + - ]: 1293 : if (bytes_written > 0) {
1078 : 1293 : self->impl_->reset_idle_timer(self, seq);
1079 : : }
1080 : 1293 : self->impl_->queue_bytes_ =
1081 [ + + ]: 1293 : (self->impl_->queue_bytes_ > queued_bytes) ? (self->impl_->queue_bytes_ - queued_bytes) : 0;
1082 : 1293 : self->impl_->report_backpressure(self, self->impl_->queue_bytes_);
1083 : :
1084 [ + - + - : 2586 : if (self->impl_->stop_requested_.load() || self->impl_->state_.is_state(LinkState::Closed) ||
- + ]
1085 : 2586 : self->impl_->state_.is_state(LinkState::Error)) {
1086 : 0 : self->impl_->writing_ = false;
1087 : 0 : return;
1088 : : }
1089 : :
1090 : 1293 : self->impl_->do_write(self, seq);
1091 : 1293 : };
1092 : :
1093 : : #ifdef WIRESTEAD_TLS_ENABLED
1094 [ + + ]: 1293 : if (tls_active()) {
1095 : 1 : net::async_write(*tls_, current_write_views_, on_write);
1096 : 1 : return;
1097 : : }
1098 : : #endif
1099 : 1292 : net::async_write(socket_, current_write_views_, on_write);
1100 : 1293 : }
1101 : :
1102 : 43 : void TcpClient::Impl::handle_close(std::shared_ptr<TcpClient> self, uint64_t seq, const boost::system::error_code& ec) {
1103 [ + - - + : 86 : if (ec == net::error::operation_aborted || seq != current_seq_.load()) {
- + ]
1104 : 0 : return;
1105 : : }
1106 : 43 : WIRESTEAD_LOG_INFO("tcp_client", "handle_close", fmt::format("Closing connection. Error: {}", ec.message()));
1107 [ + - ]: 43 : if (ec) {
1108 : : bool has_policy;
1109 : : {
1110 : 43 : std::lock_guard<std::mutex> lock(cfg_mtx_);
1111 : 43 : has_policy = reconnect_policy_.has_value();
1112 : 43 : }
1113 : 43 : const bool retryable = diagnostics::is_retryable_tcp_connect_error(ec);
1114 [ + + ]: 43 : const uint32_t current_attempts = has_policy ? reconnect_attempt_count_ : static_cast<uint32_t>(retry_attempts_);
1115 : :
1116 : 43 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "handle_close", ec,
1117 : 86 : fmt::format("Connection closed with error: {}", ec.message()), retryable, current_attempts);
1118 : : }
1119 : 43 : connected_.store(false);
1120 : 43 : writing_ = false;
1121 : 43 : cancel_idle_timer();
1122 : 43 : connect_timer_.cancel();
1123 : 43 : close_socket();
1124 [ + + + - : 43 : if (stop_requested_.load() || stopping_.load() || state_.is_state(LinkState::Closed)) {
- + + + ]
1125 : 1 : transition_to(LinkState::Closed, ec);
1126 : 1 : return;
1127 : : }
1128 : 42 : transition_to(LinkState::Connecting, ec);
1129 : 42 : schedule_retry(self, seq);
1130 : : }
1131 : :
1132 : 3 : void TcpClient::Impl::handle_idle_timeout(std::shared_ptr<TcpClient> self, uint64_t seq) {
1133 [ + - + - : 6 : if (seq != current_seq_.load() || stop_requested_.load() || stopping_.load() || !connected_.load()) {
+ - - + -
+ ]
1134 : 2 : return;
1135 : : }
1136 : :
1137 : : IdleTimeoutAction idle_timeout_action;
1138 : : unsigned idle_timeout_ms;
1139 : : bool has_policy;
1140 : : {
1141 : 3 : std::lock_guard<std::mutex> lock(cfg_mtx_);
1142 : 3 : idle_timeout_action = cfg_.idle_timeout_action;
1143 : 3 : idle_timeout_ms = cfg_.idle_timeout_ms;
1144 : 3 : has_policy = reconnect_policy_.has_value();
1145 : 3 : }
1146 : :
1147 : 3 : const auto ec = make_error_code(boost::asio::error::timed_out);
1148 : 3 : const bool should_reconnect = idle_timeout_action == IdleTimeoutAction::Reconnect;
1149 [ - + ]: 3 : const uint32_t current_attempts = has_policy ? reconnect_attempt_count_ : static_cast<uint32_t>(retry_attempts_);
1150 : :
1151 [ + - + - : 3 : WIRESTEAD_LOG_WARNING("tcp_client", "idle_timeout",
+ - + - +
+ + - +
- ]
1152 : : fmt::format("Idle timeout expired after {}ms; {}", idle_timeout_ms,
1153 : : should_reconnect ? "scheduling reconnect" : "closing connection"));
1154 : 3 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "idle_timeout", ec,
1155 : : "Idle timeout expired", should_reconnect, current_attempts);
1156 : :
1157 : 3 : connected_.store(false);
1158 : 3 : writing_ = false;
1159 : 3 : cancel_idle_timer();
1160 : 3 : connect_timer_.cancel();
1161 : 3 : close_socket();
1162 : :
1163 [ + + ]: 3 : if (!should_reconnect) {
1164 : 2 : transition_to(LinkState::Closed, ec);
1165 : 2 : return;
1166 : : }
1167 : :
1168 : 1 : transition_to(LinkState::Connecting, ec);
1169 : 1 : schedule_retry(self, seq);
1170 : : }
1171 : :
1172 : 313 : void TcpClient::Impl::close_socket() {
1173 : 313 : boost::system::error_code ec;
1174 : : #ifdef WIRESTEAD_TLS_ENABLED
1175 : : // One SSL_shutdown writes close_notify and returns; a second would wait for
1176 : : // the peer's, which is the block measured at 8 s on the server side.
1177 [ + + + - ]: 313 : if (tls_) ::SSL_shutdown(tls_->native_handle());
1178 : : // Deliberately no tls_.reset() here. socket_.cancel()/close() do not retract
1179 : : // a boost::asio::ssl::detail::io_op continuation that is already queued on
1180 : : // the strand: the strand runs it afterwards, and it calls back into the
1181 : : // engine's BIO. Destroying the stream at this point therefore left that
1182 : : // continuation dereferencing freed memory - a SIGSEGV inside BIO_ctrl,
1183 : : // reproducible under parallel load and seen intermittently in CI.
1184 : : //
1185 : : // The stream is instead destroyed by the next tls_.emplace(), which runs on
1186 : : // the strand after any pending continuation, or by ~Impl once the io thread
1187 : : // has been joined. It cannot be moved out to a holder either - a pending
1188 : : // operation holds the stream's original address.
1189 : 313 : tls_engaged_ = false;
1190 : : #endif
1191 : 313 : socket_.shutdown(tcp::socket::shutdown_both, ec);
1192 : 313 : socket_.close(ec);
1193 : 313 : }
1194 : :
1195 : 285 : void TcpClient::Impl::recalculate_backpressure_bounds() {
1196 : 285 : bp_high_ = cfg_.backpressure_threshold;
1197 [ + - ]: 285 : bp_low_ = bp_high_ > 1 ? bp_high_ / 2 : bp_high_;
1198 [ - + ]: 285 : if (bp_low_ == 0) {
1199 : 0 : bp_low_ = 1;
1200 : : }
1201 : 285 : bp_limit_ = std::min(std::max(bp_high_ * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
1202 : : base::constants::MAX_BUFFER_SIZE);
1203 [ - + ]: 285 : if (bp_limit_ < bp_high_) {
1204 : 0 : bp_limit_ = bp_high_;
1205 : : }
1206 : 285 : backpressure_active_ = false;
1207 : 285 : }
1208 : :
1209 : 6316 : queue_util::BackpressureFields TcpClient::Impl::bp_fields() {
1210 : 6316 : return queue_util::BackpressureFields{queue_bytes_,
1211 : 6316 : pending_bytes_,
1212 : 6316 : backpressure_active_,
1213 : 6316 : bp_high_,
1214 : 6316 : bp_low_,
1215 : 6316 : bp_limit_,
1216 : 6316 : bp_strategy_.load(std::memory_order_relaxed)};
1217 : : }
1218 : :
1219 : 1767 : void TcpClient::Impl::route_enqueued_buffer(std::shared_ptr<TcpClient> self, BufferVariant&& buf, size_t added,
1220 : : bool reserved) {
1221 [ + + + - : 1767 : if (stop_requested_.load() || state_.is_state(LinkState::Closed) || state_.is_state(LinkState::Error)) {
- + + + ]
1222 [ + - + - ]: 1 : if (reserved) queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
1223 : 1 : stats_.record_failed_send();
1224 : 1 : return;
1225 : : }
1226 : :
1227 : 1766 : auto f = bp_fields();
1228 : 1766 : queue_util::DropAccounting dropped;
1229 : 1766 : auto decision = queue_util::decide_enqueue(f, added, tx_, dropped);
1230 [ + + + - ]: 1766 : if (dropped.any()) stats_.record_dropped(dropped.messages, dropped.bytes);
1231 : :
1232 [ - + ]: 1766 : if (decision == queue_util::EnqueueDecision::Rejected) {
1233 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "write", fmt::format("Queue limit exceeded ({} bytes)", queue_bytes_ + added));
1234 : 0 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::COMMUNICATION, "write",
1235 : : boost::asio::error::no_buffer_space, "Queue limit exceeded", false, 0);
1236 : : // #448: this path used to leave RuntimeStats showing the message as
1237 : : // accepted (from the caller-thread pre-check) with no corresponding
1238 : : // sent/dropped/queued accounting - record it as dropped so it's at
1239 : : // least observable.
1240 : 0 : stats_.record_dropped(1, added);
1241 [ # # # # ]: 0 : if (reserved) queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
1242 : 0 : report_backpressure(self, queue_bytes_ + added);
1243 : 0 : return;
1244 : : }
1245 [ - + ]: 1766 : if (decision == queue_util::EnqueueDecision::Pending) {
1246 [ # # ]: 0 : if (reserved) {
1247 : 0 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, pending_bytes_, inflight_bytes_, added);
1248 : : } else {
1249 : 0 : queue_util::commit_unreserved_limit_bytes(write_reserve_mtx_, pending_bytes_, added);
1250 : : }
1251 : 0 : pending_.emplace_back(std::move(buf));
1252 : 0 : observe_queue();
1253 : 0 : return;
1254 : : }
1255 [ + + ]: 1766 : if (reserved) {
1256 : 1754 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, queue_bytes_, inflight_bytes_, added);
1257 : : } else {
1258 : 12 : queue_util::commit_unreserved_limit_bytes(write_reserve_mtx_, queue_bytes_, added);
1259 : : }
1260 : 1766 : tx_.emplace_back(std::move(buf));
1261 : 1766 : observe_queue();
1262 : 1766 : report_backpressure(self, queue_bytes_);
1263 [ + + + - ]: 2923 : if (!writing_) do_write(self, current_seq_.load());
1264 : : }
1265 : :
1266 : 7810 : void TcpClient::Impl::observe_queue() {
1267 : 23430 : stats_.observe_queue(queue_bytes_.load(std::memory_order_relaxed) + pending_bytes_.load(std::memory_order_relaxed));
1268 : 7810 : }
1269 : :
1270 : 4559 : void TcpClient::Impl::report_backpressure(std::shared_ptr<TcpClient> self, size_t queued_bytes) {
1271 [ + + - + : 4559 : if (stop_requested_.load() || stopping_.load()) return;
+ + ]
1272 : 4550 : observe_queue();
1273 : :
1274 : 4550 : interface::SharedCallback<OnBackpressure> on_bp;
1275 : : {
1276 : 4550 : std::lock_guard<std::mutex> lock(callback_mtx_);
1277 : 4550 : on_bp = on_bp_;
1278 : 4550 : }
1279 : 4550 : static const OnBackpressure kNoCallback;
1280 : :
1281 : 4550 : auto f = bp_fields();
1282 [ + + + - ]: 9097 : queue_util::report_backpressure(
1283 : 4547 : f, queued_bytes, on_bp ? *on_bp : kNoCallback, stats_,
1284 : 0 : [&]() -> size_t {
1285 : 3 : const size_t moved = pending_bytes_.exchange(0);
1286 [ - + ]: 3 : while (!pending_.empty()) {
1287 : 0 : tx_.emplace_back(std::move(pending_.front()));
1288 : 0 : pending_.pop_front();
1289 : : }
1290 : 3 : return moved;
1291 : : },
1292 : 4550 : [&]() {
1293 : 3 : observe_queue();
1294 [ - + - - ]: 3 : if (!writing_) do_write(self, current_seq_.load());
1295 : 3 : });
1296 : 4550 : }
1297 : :
1298 : 531 : void TcpClient::Impl::transition_to(LinkState next, const boost::system::error_code& ec) {
1299 [ - + ]: 531 : if (ec == net::error::operation_aborted) {
1300 : 0 : return;
1301 : : }
1302 : :
1303 : 531 : const auto current = state_.get();
1304 [ + + + + ]: 531 : const bool retrying_same_state = (next == LinkState::Connecting && current == LinkState::Connecting);
1305 [ + + + + ]: 531 : if ((current == LinkState::Closed || current == LinkState::Error) &&
1306 [ - + - - ]: 7 : (next == LinkState::Closed || next == LinkState::Error)) {
1307 : 7 : return;
1308 : : }
1309 : :
1310 [ + + + + ]: 524 : if (next == LinkState::Closed || next == LinkState::Error) {
1311 [ - + ]: 129 : if (terminal_state_notified_.exchange(true)) {
1312 : 0 : return;
1313 : : }
1314 [ + + - + ]: 395 : } else if (current == next && !retrying_same_state) {
1315 : 0 : return;
1316 : : }
1317 : :
1318 : 524 : state_.set(next);
1319 : 524 : notify_state();
1320 : : }
1321 : :
1322 : 124 : void TcpClient::Impl::perform_stop_cleanup() {
1323 : : try {
1324 : 124 : retry_timer_.cancel();
1325 : 124 : connect_timer_.cancel();
1326 : 124 : cancel_idle_timer();
1327 : 124 : resolver_.cancel();
1328 : 124 : boost::system::error_code ec_cancel;
1329 : 124 : socket_.cancel(ec_cancel);
1330 : 124 : close_socket();
1331 : 124 : tx_.clear();
1332 : 124 : queue_bytes_ = 0;
1333 : 124 : pending_.clear();
1334 : 124 : pending_bytes_ = 0;
1335 : 124 : writing_ = false;
1336 : 124 : connected_.store(false);
1337 : : // Deliberately does NOT fire on_bp_ here, unlike UDP/server sessions'
1338 : : // terminal drain (#434): this is an explicit, tested contract
1339 : : // (ContractComplianceTest.TcpClient_Backpressure_Contract) - a
1340 : : // Reliable-mode caller blocked in send_blocking() is woken instead via
1341 : : // the wrapper's own bp_cv_.notify_all()/is_connected() check, not a
1342 : : // relief callback. Don't "fix" this to match the other transports
1343 : : // without updating that contract test first.
1344 : 124 : backpressure_active_ = false;
1345 : :
1346 [ + + + - : 124 : if (owns_ioc_ && work_guard_) {
+ + ]
1347 : 116 : work_guard_->reset();
1348 : : }
1349 : 124 : transition_to(LinkState::Closed);
1350 : 0 : } catch (const std::exception& e) {
1351 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "stop_cleanup", fmt::format("Cleanup error: {}", e.what()));
1352 : 0 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::SYSTEM, "stop_cleanup", {},
1353 : 0 : fmt::format("Cleanup error: {}", e.what()), false, 0);
1354 : 0 : diagnostics::error_reporting::report_system_error("tcp_client", "stop_cleanup",
1355 : 0 : fmt::format("Exception in stop cleanup: {}", e.what()));
1356 : 0 : } catch (...) {
1357 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "stop_cleanup", "Unknown error in stop cleanup");
1358 : 0 : diagnostics::error_reporting::report_system_error("tcp_client", "stop_cleanup", "Unknown error in stop cleanup");
1359 : 0 : }
1360 : 124 : }
1361 : :
1362 : 143 : void TcpClient::Impl::reset_start_state() {
1363 : 143 : stop_requested_.store(false);
1364 : 143 : stopping_.store(false);
1365 : 143 : terminal_state_notified_.store(false);
1366 : 143 : reconnect_pending_.store(false);
1367 : 143 : retry_attempts_ = 0;
1368 : 143 : reconnect_attempt_count_ = 0;
1369 : 143 : connected_.store(false);
1370 : 143 : writing_ = false;
1371 : 143 : queue_bytes_ = 0;
1372 : 143 : pending_.clear();
1373 : 143 : pending_bytes_ = 0;
1374 : 143 : backpressure_active_ = false;
1375 : 143 : state_.set(LinkState::Idle);
1376 : 143 : }
1377 : :
1378 : 292 : void TcpClient::Impl::join_ioc_thread(bool allow_detach) {
1379 [ + + + + : 292 : if (!owns_ioc_ || !ioc_thread_.joinable()) {
+ + ]
1380 : 175 : return;
1381 : : }
1382 : :
1383 [ + + ]: 117 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
1384 [ - + ]: 1 : if (allow_detach) {
1385 : 0 : ioc_thread_.detach();
1386 : : }
1387 : 1 : return;
1388 : : }
1389 : :
1390 : : try {
1391 : 116 : ioc_thread_.join();
1392 : 0 : } catch (const std::exception& e) {
1393 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "join", "Join failed: " + std::string(e.what()));
1394 : 0 : } catch (...) {
1395 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "join", "Join failed with unknown error");
1396 : 0 : }
1397 : : }
1398 : :
1399 : 524 : void TcpClient::Impl::notify_state() {
1400 [ + + - + : 577 : if (stop_requested_.load() || stopping_.load()) return;
+ + ]
1401 : :
1402 : 406 : interface::SharedCallback<OnState> on_state;
1403 : : {
1404 : 406 : std::lock_guard<std::mutex> lock(callback_mtx_);
1405 : 406 : on_state = on_state_;
1406 : 406 : }
1407 [ + + ]: 406 : if (!on_state) return;
1408 : :
1409 : : try {
1410 : 353 : (*on_state)(state_.get());
1411 : 2 : } catch (const std::exception& e) {
1412 : 6 : WIRESTEAD_LOG_ERROR("tcp_client", "on_state", "Exception in state callback: " + std::string(e.what()));
1413 : 2 : } catch (...) {
1414 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "on_state", "Unknown exception in state callback");
1415 : 0 : }
1416 : 406 : }
1417 : :
1418 : 134 : void TcpClient::Impl::record_error(diagnostics::ErrorLevel lvl, diagnostics::ErrorCategory cat,
1419 : : std::string_view operation, const boost::system::error_code& ec,
1420 : : std::string_view msg, bool retryable, uint32_t retry_count) {
1421 : 134 : error_info_holder_.record_error(lvl, cat, operation, ec, msg, retryable, retry_count);
1422 : 134 : }
1423 : :
1424 : 143 : void TcpClient::Impl::reset_io_objects() {
1425 : : try {
1426 : 143 : boost::system::error_code ec_cancel;
1427 : 143 : socket_.cancel(ec_cancel);
1428 : 143 : close_socket();
1429 : 143 : socket_ = tcp::socket(strand_);
1430 : 143 : resolver_.cancel();
1431 : 143 : resolver_ = tcp::resolver(strand_);
1432 : 143 : retry_timer_ = net::steady_timer(strand_);
1433 : 143 : connect_timer_ = net::steady_timer(strand_);
1434 : 143 : idle_timer_ = net::steady_timer(strand_);
1435 : 143 : tx_.clear();
1436 : 143 : queue_bytes_ = 0;
1437 : 143 : pending_.clear();
1438 : 143 : pending_bytes_ = 0;
1439 : 143 : writing_ = false;
1440 : 143 : backpressure_active_ = false;
1441 : 0 : } catch (const std::exception& e) {
1442 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "reset_io_objects", fmt::format("Reset error: {}", e.what()));
1443 : 0 : record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::SYSTEM, "reset_io_objects", {},
1444 : 0 : fmt::format("Reset error: {}", e.what()), false, 0);
1445 : 0 : diagnostics::error_reporting::report_system_error(
1446 : 0 : "tcp_client", "reset_io_objects", fmt::format("Exception while resetting io objects: {}", e.what()));
1447 : 0 : } catch (...) {
1448 : 0 : WIRESTEAD_LOG_ERROR("tcp_client", "reset_io_objects", "Unknown reset error");
1449 : 0 : diagnostics::error_reporting::report_system_error("tcp_client", "reset_io_objects",
1450 : : "Unknown error while resetting io objects");
1451 : 0 : }
1452 : 143 : }
1453 : :
1454 : 1529 : void TcpClient::Impl::reset_idle_timer(std::shared_ptr<TcpClient> self, uint64_t seq) {
1455 : : unsigned idle_timeout_ms;
1456 : : {
1457 : 1529 : std::lock_guard<std::mutex> lock(cfg_mtx_);
1458 : 1529 : idle_timeout_ms = cfg_.idle_timeout_ms;
1459 : 1529 : }
1460 [ + + + - : 1529 : if (idle_timeout_ms == 0 || !connected_.load() || stop_requested_.load() || stopping_.load()) {
+ + - + +
+ ]
1461 : 1518 : return;
1462 : : }
1463 : :
1464 : 11 : idle_timer_.cancel();
1465 : 11 : idle_timer_.expires_after(std::chrono::milliseconds(idle_timeout_ms));
1466 : 11 : idle_timer_.async_wait([self, seq](const boost::system::error_code& ec) {
1467 [ + + - + : 14 : if (ec == net::error::operation_aborted || seq != self->impl_->current_seq_.load()) {
+ + ]
1468 : 8 : return;
1469 : : }
1470 [ + - ]: 3 : if (!ec) {
1471 : 3 : self->impl_->handle_idle_timeout(self, seq);
1472 : : }
1473 : : });
1474 : : }
1475 : :
1476 : 170 : void TcpClient::Impl::cancel_idle_timer() { idle_timer_.cancel(); }
1477 : :
1478 : : } // namespace transport
1479 : : } // namespace wirestead
|
