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