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_server/tcp_server.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 <unordered_map>
31 : : #include <vector>
32 : :
33 : : #include "wirestead/base/common.hpp"
34 : : #include "wirestead/concurrency/io_thread_hook.hpp"
35 : : #include "wirestead/config/tcp_server_config.hpp"
36 : : #include "wirestead/config/validation.hpp"
37 : : #include "wirestead/factory/channel_factory.hpp"
38 : : #include "wirestead/transport/tcp_server/tcp_server.hpp"
39 : : #include "wirestead/util/input_validator.hpp"
40 : : #include "wirestead/wrapper/callback_guard.hpp"
41 : : #include "wirestead/wrapper/error_context_builder.hpp"
42 : : #include "wirestead/wrapper/send_retry.hpp"
43 : : #include "wirestead/wrapper/send_validation.hpp"
44 : : #include "wirestead/wrapper/session_batch.hpp"
45 : :
46 : : namespace wirestead {
47 : : namespace wrapper {
48 : :
49 : : struct TcpServer::Impl : public std::enable_shared_from_this<Impl> {
50 : : mutable std::shared_mutex mutex_;
51 : : std::mutex bp_mutex_;
52 : : std::condition_variable bp_cv_;
53 : : // D-1: admission gate for this object's user callbacks, across every
54 : : // session. Admission and the running count are one decision under the gate's
55 : : // own lock, and a new run refuses callbacks left over from the previous one.
56 : : detail::CallbackGate callback_gate_;
57 : : std::mutex stop_finalize_mutex_;
58 : : bool injected_channel_ = false;
59 : : std::atomic<uint64_t> callback_generation_{0};
60 : : bool stop_requested_ = false;
61 : : std::atomic<unsigned> stop_callers_{0};
62 : :
63 : : // A thread the shutdown needs: one running a callback of this object, or one
64 : : // currently running the external io_context this server was built on - which
65 : : // a callback of another channel sharing that context is.
66 : 938 : bool shutdown_needs_this_thread() const {
67 [ + - + + ]: 938 : if (callback_gate_.active_on_this_thread()) return true;
68 [ + + - + : 934 : if (external_ioc_ && external_ioc_->get_executor().running_in_this_thread()) return true;
+ + - + ]
69 : 934 : std::shared_lock<std::shared_mutex> lock(mutex_);
70 [ + + ]: 934 : if (!channel_) return false;
71 : 540 : const auto executor = channel_->get_executor();
72 : : using IoExecutor = boost::asio::io_context::executor_type;
73 [ + - + + ]: 540 : if (auto* io = executor.target<IoExecutor>()) return io->running_in_this_thread();
74 [ + - + - ]: 529 : if (auto* strand = executor.target<boost::asio::strand<IoExecutor>>())
75 : 529 : return strand->get_inner_executor().running_in_this_thread();
76 : 0 : return false;
77 : 934 : }
78 : :
79 : : uint16_t port_;
80 : : std::string bind_address_{"0.0.0.0"};
81 : : std::string tls_certificate_file_;
82 : : std::string tls_private_key_file_;
83 : : std::shared_ptr<interface::Channel> channel_;
84 : : std::shared_ptr<boost::asio::io_context> external_ioc_;
85 : : std::atomic<bool> use_external_context_{false};
86 : : std::atomic<bool> manage_external_context_{false};
87 : : std::jthread external_thread_;
88 : : std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>> work_guard_;
89 : :
90 : : std::vector<std::promise<bool>> pending_promises_;
91 : : std::atomic<bool> started_{false};
92 : : std::atomic<bool> is_listening_{false};
93 : 284 : std::shared_ptr<bool> alive_marker_{std::make_shared<bool>(true)};
94 : :
95 : : // Configuration
96 : : std::atomic<bool> auto_start_{false};
97 : : std::atomic<bool> shared_context_{false};
98 : : std::atomic<bool> port_retry_enabled_{false};
99 : : std::atomic<int> max_port_retries_{3};
100 : : std::atomic<int> port_retry_interval_ms_{1000};
101 : : std::atomic<int> idle_timeout_ms_{static_cast<int>(base::constants::DEFAULT_IDLE_TIMEOUT_MS)};
102 : : std::atomic<bool> client_limit_enabled_{false};
103 : : std::atomic<size_t> max_clients_{0};
104 : : std::atomic<size_t> backpressure_threshold_{base::constants::DEFAULT_BACKPRESSURE_THRESHOLD};
105 : : std::atomic<base::constants::BackpressureStrategy> backpressure_strategy_{
106 : : base::constants::BackpressureStrategy::Reliable};
107 : : std::atomic<bool> tcp_no_delay_{base::constants::DEFAULT_TCP_NO_DELAY};
108 : : std::atomic<bool> keep_alive_{base::constants::DEFAULT_KEEP_ALIVE};
109 : : std::atomic<size_t> send_buffer_size_{0};
110 : : std::atomic<size_t> receive_buffer_size_{0};
111 : : std::atomic<size_t> read_buffer_size_{base::constants::DEFAULT_READ_BUFFER_SIZE};
112 : :
113 : : ConnectionHandler on_client_connect_{nullptr};
114 : : ConnectionHandler on_disconnect_{nullptr};
115 : : // Shared snapshots: the io thread copies one out per received chunk, and a
116 : : // std::function copy allocates whenever the user handler outgrows its
117 : : // small-object buffer. See interface::SharedCallback.
118 : : interface::SharedCallback<MessageHandler> on_data_;
119 : : interface::SharedCallback<BatchMessageHandler> data_batch_handler_;
120 : : ErrorHandler on_error_{nullptr};
121 : : std::function<void(size_t)> on_backpressure_{nullptr};
122 : : FramerFactory framer_factory_{nullptr};
123 : : interface::SharedCallback<MessageHandler> on_message_;
124 : : interface::SharedCallback<BatchMessageHandler> message_batch_handler_;
125 : :
126 : : // Batching logic
127 : : bool terminal_error_reported_ = false;
128 : 236 : void require_stopped_config() const {
129 [ + + + - : 468 : if (started_.load() || stop_callers_.load() || (stop_requested_ && alive_marker_) || detail::in_data_callback())
+ + + - -
+ + + ]
130 : 4 : throw std::logic_error("configuration requires completed stop");
131 : 232 : }
132 : : ReceiveLimits receive_limits_;
133 : : std::shared_ptr<detail::ReceiveBudget> receive_budget_{std::make_shared<detail::ReceiveBudget>(receive_limits_)};
134 : : std::unordered_map<ClientId, std::shared_ptr<detail::SessionBatch>> batches_;
135 : : size_t max_batch_size_ = 100;
136 : 284 : std::chrono::milliseconds max_batch_latency_{1};
137 : :
138 : : std::unordered_map<ClientId, std::shared_ptr<framer::IFramer>> framers_;
139 : : // Cached transport pointer — set once in start(), avoids repeated dynamic_cast.
140 : : std::shared_ptr<transport::TcpServer> transport_cache_;
141 : :
142 : 182 : explicit Impl(uint16_t port)
143 : 364 : : port_(port),
144 : 182 : started_(false),
145 : 182 : is_listening_(false),
146 : 182 : auto_start_(false),
147 : 182 : port_retry_enabled_(false),
148 : 182 : max_port_retries_(3),
149 : 182 : port_retry_interval_ms_(1000),
150 : 182 : idle_timeout_ms_(static_cast<int>(base::constants::DEFAULT_IDLE_TIMEOUT_MS)),
151 : 182 : client_limit_enabled_(false),
152 : 182 : max_clients_(0),
153 : 182 : backpressure_threshold_(base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
154 : 1092 : backpressure_strategy_(base::constants::BackpressureStrategy::Reliable) {}
155 : :
156 : 6 : Impl(uint16_t port, std::shared_ptr<boost::asio::io_context> external_ioc)
157 : 12 : : port_(port),
158 : 6 : external_ioc_(std::move(external_ioc)),
159 : 6 : use_external_context_(external_ioc_ != nullptr),
160 : 6 : manage_external_context_(false),
161 : 6 : started_(false),
162 : 6 : is_listening_(false),
163 : 6 : auto_start_(false),
164 : 6 : port_retry_enabled_(false),
165 : 6 : max_port_retries_(3),
166 : 6 : port_retry_interval_ms_(1000),
167 : 6 : idle_timeout_ms_(static_cast<int>(base::constants::DEFAULT_IDLE_TIMEOUT_MS)),
168 : 6 : client_limit_enabled_(false),
169 : 6 : max_clients_(0),
170 : 6 : backpressure_threshold_(base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
171 : 42 : backpressure_strategy_(base::constants::BackpressureStrategy::Reliable) {}
172 : :
173 : 96 : explicit Impl(std::shared_ptr<interface::Channel> channel)
174 : 192 : : port_(0),
175 : 96 : channel_(std::move(channel)),
176 : 96 : started_(false),
177 : 96 : is_listening_(false),
178 : 96 : auto_start_(false),
179 : 96 : port_retry_enabled_(false),
180 : 96 : max_port_retries_(3),
181 : 96 : port_retry_interval_ms_(1000),
182 : 96 : idle_timeout_ms_(static_cast<int>(base::constants::DEFAULT_IDLE_TIMEOUT_MS)),
183 : 96 : client_limit_enabled_(false),
184 : 96 : max_clients_(0),
185 : 96 : backpressure_threshold_(base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
186 : 672 : backpressure_strategy_(base::constants::BackpressureStrategy::Reliable) {
187 : 96 : injected_channel_ = true;
188 : 96 : transport_cache_ = std::dynamic_pointer_cast<transport::TcpServer>(channel_);
189 : : // #450: setup_internal_handlers() captures weak_from_this() - calling it
190 : : // from inside this constructor would capture an empty weak_ptr, since
191 : : // enable_shared_from_this isn't wired up until make_shared() finishes
192 : : // constructing the object. Deferred to TcpServer's own constructor,
193 : : // which runs after impl_ is a fully-formed shared_ptr<Impl>.
194 : 96 : }
195 : :
196 : 284 : ~Impl() {
197 : : try {
198 : 284 : stop();
199 : 0 : } catch (...) {
200 : 0 : }
201 : 284 : }
202 : :
203 : 1370 : void fulfill_all_locked(bool value) {
204 [ + + ]: 1652 : for (auto& p : pending_promises_) {
205 : : try {
206 : 282 : p.set_value(value);
207 : 0 : } catch (...) {
208 : 0 : }
209 : : }
210 : 1370 : pending_promises_.clear();
211 : 1370 : }
212 : :
213 : 216 : void flush_batches(uint64_t generation, ClientId id, const std::shared_ptr<detail::SessionBatch>& state) {
214 : 216 : auto lease = callback_gate_.enter(generation);
215 [ - + ]: 216 : if (!lease.admitted()) return;
216 : 216 : std::unique_lock<std::shared_mutex> lock(mutex_);
217 : 216 : auto found = batches_.find(id);
218 [ + - - + : 216 : if (found == batches_.end() || found->second != state) return;
- + ]
219 : 216 : state->scheduled = false;
220 : 216 : auto data = std::move(state->data);
221 : 216 : auto messages = std::move(state->messages);
222 : 216 : state->data.clear();
223 : 216 : state->messages.clear();
224 : 216 : auto data_handler = data_batch_handler_;
225 : 216 : auto message_handler = message_batch_handler_;
226 : 216 : lock.unlock();
227 [ + + + - ]: 216 : if (!data.empty()) detail::invoke_user_callback("tcp_server", "on_data_batch", data_handler, data);
228 : : // A callback may stop the server. Recheck admission before another callback.
229 : 216 : auto next = callback_gate_.enter(generation);
230 [ + - + + : 216 : if (next.admitted() && !messages.empty())
+ + ]
231 : 4 : detail::invoke_user_callback("tcp_server", "on_message_batch", message_handler, messages);
232 : 216 : }
233 : :
234 : : // Caller holds mutex_; do not postpone an existing deadline when the
235 : : // other queue receives its first item.
236 : 15 : void schedule_batch_timer(uint64_t generation, ClientId id) {
237 : 15 : auto found = batches_.find(id);
238 [ + - + + : 15 : if (found == batches_.end() || found->second->scheduled) return;
+ + ]
239 : 14 : auto state = found->second;
240 : 14 : state->scheduled = true;
241 : 14 : state->timer.expires_after(max_batch_latency_);
242 : 14 : state->timer.async_wait(
243 : 14 : [weak_impl = weak_from_this(), generation, id,
244 : : weak_state = std::weak_ptr<detail::SessionBatch>(state)](const boost::system::error_code& ec) {
245 [ + + ]: 11 : if (ec) return;
246 : 7 : auto self = weak_impl.lock();
247 : 7 : auto batch = weak_state.lock();
248 [ + - + - : 7 : if (self && batch) self->flush_batches(generation, id, batch);
+ - + - ]
249 : 7 : });
250 : 14 : }
251 : :
252 : 284 : std::future<bool> start() {
253 : 284 : std::unique_lock<std::shared_mutex> lock(mutex_);
254 : 284 : stop_requested_ = false;
255 [ + + ]: 284 : if (is_listening_.load()) {
256 : 2 : started_.store(true);
257 : 2 : std::promise<bool> p;
258 : 2 : p.set_value(true);
259 : 2 : return p.get_future();
260 : 2 : }
261 : 282 : std::promise<bool> p;
262 : 282 : auto f = p.get_future();
263 : 282 : pending_promises_.push_back(std::move(p));
264 [ - + ]: 282 : if (started_.exchange(true)) return f;
265 : :
266 [ + + + - ]: 282 : if (!alive_marker_) alive_marker_ = std::make_shared<bool>(true);
267 [ + + ]: 282 : if (!channel_) {
268 : 177 : config::TcpServerConfig config;
269 : 177 : config.bind_address = bind_address_;
270 : 177 : config.port = port_;
271 : 177 : config.enable_port_retry = port_retry_enabled_.load();
272 : 177 : config.max_port_retries = max_port_retries_.load();
273 : 177 : config.port_retry_interval_ms = port_retry_interval_ms_.load();
274 : 177 : config.idle_timeout_ms = idle_timeout_ms_.load();
275 : 177 : config.backpressure_threshold = backpressure_threshold_.load();
276 : 177 : config.backpressure_strategy = backpressure_strategy_.load();
277 : 177 : config.tcp_no_delay = tcp_no_delay_.load();
278 : 177 : config.keep_alive = keep_alive_.load();
279 : 177 : config.send_buffer_size = send_buffer_size_.load();
280 : 177 : config.receive_buffer_size = receive_buffer_size_.load();
281 : 177 : config.read_buffer_size = read_buffer_size_.load();
282 : 177 : config.use_shared_context = shared_context_.load();
283 : 177 : config.tls_certificate_file = tls_certificate_file_;
284 : 177 : config.tls_private_key_file = tls_private_key_file_;
285 : :
286 : 177 : channel_ = factory::ChannelFactory::create(config, external_ioc_);
287 : 177 : transport_cache_ = std::dynamic_pointer_cast<transport::TcpServer>(channel_);
288 : :
289 [ + + ]: 177 : if (client_limit_enabled_.load()) {
290 : 8 : auto transport_server = std::dynamic_pointer_cast<transport::TcpServer>(channel_);
291 [ + - ]: 8 : if (transport_server) {
292 : 16 : transport_server->set_client_limit(max_clients_.load());
293 : : }
294 : 8 : }
295 : 177 : }
296 : : // D-1: registering the handlers is what opens this run's generation and
297 : : // admits callbacks again - one step under the gate's lock, so a callback
298 : : // of the previous run can never be admitted into this one. It runs on
299 : : // every start, including on an injected channel whose object serves every
300 : : // run.
301 [ + - ]: 282 : if (channel_) {
302 : 282 : transport_cache_ = std::dynamic_pointer_cast<transport::TcpServer>(channel_);
303 : 282 : setup_internal_handlers();
304 : : }
305 : :
306 : : // #506: take a local copy of the shared_ptr before unlocking. Calling
307 : : // through the raw channel_ member here would race a concurrent stop()'s
308 : : // channel_.reset() on the member itself (not just the pointee) - TSAN
309 : : // caught exactly this under TcpServerWrapperLifecycleTest.
310 : : // ConcurrentStartStop. A local copy holds its own reference, so it stays
311 : : // valid and race-free even if another thread resets the member.
312 : 282 : auto channel_copy = channel_;
313 : 282 : lock.unlock();
314 : 282 : channel_copy->start();
315 [ + + + + : 282 : if (use_external_context_.load() && manage_external_context_.load() && !external_thread_.joinable()) {
+ - + + ]
316 [ + - + - : 2 : if (external_ioc_ && external_ioc_->stopped()) {
+ + + + ]
317 : 1 : external_ioc_->restart();
318 : : }
319 : 4 : work_guard_ = std::make_unique<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>(
320 : 6 : boost::asio::make_work_guard(*external_ioc_));
321 : 4 : external_thread_ = std::jthread([ioc = external_ioc_](std::stop_token st) {
322 : 2 : wirestead::concurrency::run_io_thread_init();
323 : : try {
324 : 2 : std::stop_callback cb(st, [ioc] { ioc->stop(); });
325 : 2 : ioc->run();
326 : 2 : } catch (...) {
327 : 0 : }
328 : 4 : });
329 : : }
330 : 282 : return f;
331 : 284 : }
332 : :
333 : 938 : void stop() {
334 : 938 : stop_callers_.fetch_add(1);
335 : : struct StopCall {
336 : : std::atomic<unsigned>& count;
337 : 938 : ~StopCall() { count.fetch_sub(1); }
338 : 938 : } stop_call{stop_callers_};
339 : 938 : const bool request_only = shutdown_needs_this_thread();
340 : 938 : callback_gate_.close();
341 : 938 : std::shared_ptr<interface::Channel> channel;
342 : : {
343 : 938 : std::unique_lock<std::shared_mutex> lock(mutex_);
344 : 938 : stop_requested_ = true;
345 [ + + + - ]: 938 : if (auto ts = transport_cache_) ts->cancel_target_waits();
346 : 938 : started_.store(false);
347 : 938 : is_listening_.store(false);
348 : 938 : bp_cv_.notify_all();
349 : 938 : fulfill_all_locked(false);
350 : 938 : channel = channel_;
351 : 938 : }
352 : : // Do not hold wrapper locks while the transport waits for callbacks.
353 [ + + + - ]: 938 : if (channel) channel->stop();
354 [ + + ]: 938 : if (request_only) return;
355 : :
356 : 905 : callback_gate_.wait_until_idle();
357 : : // All outside callers pass this lock, including those that found the
358 : : // channel already cleared. Joining and final state reset happen once.
359 : 905 : std::lock_guard<std::mutex> finalize_lock(stop_finalize_mutex_);
360 : : {
361 : 905 : std::unique_lock<std::shared_mutex> lock(mutex_);
362 : 905 : alive_marker_.reset();
363 : 905 : batches_.clear();
364 [ + + ]: 905 : if (channel_) {
365 : 504 : channel_->on_bytes(nullptr);
366 : 504 : channel_->on_state(nullptr);
367 : 504 : channel_->on_backpressure(nullptr);
368 : : }
369 [ + + ]: 905 : if (!injected_channel_) channel_.reset();
370 : 905 : transport_cache_.reset();
371 : 905 : framers_.clear();
372 : 905 : }
373 [ + + + + : 905 : if (use_external_context_.load() && manage_external_context_.load()) {
+ + ]
374 [ + + ]: 10 : if (work_guard_) work_guard_.reset();
375 [ + - + - ]: 10 : if (external_ioc_) external_ioc_->stop();
376 [ + + + - ]: 10 : if (external_thread_.joinable()) external_thread_.join();
377 : : }
378 : 971 : }
379 : :
380 : : // Caller holds mutex_. Final admission rechecks the native lifecycle.
381 : 1681 : SendResult send_state(const std::shared_ptr<transport::TcpServer>& ts) {
382 [ + + ]: 3362 : if (stop_callers_.load() != 0) return SendResult::reject(SendRejection::Stopping);
383 [ + + ]: 1680 : if (!started_.load()) {
384 [ + + ]: 44 : if (stop_requested_) {
385 [ - + ]: 26 : if (!callback_gate_.idle()) return SendResult::reject(SendRejection::Stopping);
386 [ + + ]: 26 : if (ts) {
387 : 15 : const auto state = ts->target_state();
388 [ + - + + : 15 : if (!state.accepted() && state.reason() == SendRejection::Stopping) return state;
+ + ]
389 : : }
390 : : }
391 : 33 : return SendResult::reject(SendRejection::NotStarted);
392 : : }
393 [ + + ]: 1636 : if (!ts) return SendResult::reject(SendRejection::NotReady);
394 : 1631 : return SendResult::accept();
395 : : }
396 : :
397 : 94 : SendResult try_send_to(ClientId client_id, std::string_view data, bool best_effort_send = false) {
398 : 94 : std::shared_lock<std::shared_mutex> lock(mutex_);
399 : 94 : const auto& ts = transport_cache_;
400 : 0 : const auto result = [&]() -> SendResult {
401 : : auto validation =
402 [ + + + - ]: 94 : detail::validate_payload_size(data.size(), ts ? ts->write_queue_limit(client_id) : std::nullopt);
403 [ + + ]: 94 : if (!validation.accepted()) return validation;
404 : 83 : const auto state = send_state(ts);
405 [ + + ]: 83 : if (!state.accepted()) return state;
406 : 114 : auto admitted = ts->write_target(
407 : 57 : client_id, memory::ConstByteSpan(reinterpret_cast<const uint8_t*>(data.data()), data.size()), true);
408 [ + + + + : 57 : if (best_effort_send && !admitted.accepted() && admitted.reason() == SendRejection::WouldBlock)
+ + + + ]
409 : 2 : return SendResult::reject(SendRejection::QueueFull);
410 : 55 : return admitted;
411 : 94 : }();
412 : 94 : lock.unlock();
413 [ + + + - ]: 94 : if (auto hook = detail::g_tcp_server_send_result_hook.load()) hook(result);
414 : 94 : return result;
415 : 94 : }
416 : :
417 : 1260 : FanoutResult try_broadcast(std::string_view data, bool append_newline = false) {
418 : 1260 : std::shared_lock<std::shared_mutex> lock(mutex_);
419 : 1260 : const auto& ts = transport_cache_;
420 [ + + + - : 1260 : return ts ? ts->broadcast_result({reinterpret_cast<const uint8_t*>(data.data()), data.size()}, send_state(ts),
+ - ]
421 : : append_newline)
422 : 1272 : : FanoutResult{};
423 : 1260 : }
424 : :
425 : 168 : SendResult send_to(ClientId client_id, std::string_view data) {
426 [ + + ]: 168 : if (backpressure_strategy_.load() == base::constants::BackpressureStrategy::Reliable)
427 : 150 : return send_to_blocking(client_id, data);
428 : 18 : return try_send_to(client_id, data, true);
429 : : }
430 : :
431 : 1210 : FanoutResult broadcast(std::string_view data) {
432 : : // In order to avoid Head-Of-Line blocking where a single slow client
433 : : // blocks the entire broadcast loop, we delegate to try_broadcast (async fan-out)
434 : : // even in Reliable mode. Transport-level backpressure will still protect the queues.
435 : 1210 : return try_broadcast(data);
436 : : }
437 : :
438 : : // channel_->on_backpressure()/session-level backpressure callbacks call bp_cv_.notify_all()
439 : : // from the transport's io_context thread without holding bp_mutex_ - a classic lost-wakeup
440 : : // race is possible: a waiter can check the predicate, find it still blocking, and be in the
441 : : // process of registering to wait when the notify fires. Poll with a bounded timeout instead
442 : : // of an unbounded wait() so a missed notify only costs a short delay rather than a
443 : : // permanent hang (see #427, #431).
444 : : //
445 : : // Returns false without sending instead of waiting if called from the
446 : : // channel's own io thread while backpressure is active for this client -
447 : : // e.g. a blocking send_to() called from inside an on_data/on_message
448 : : // callback. Clearing backpressure requires that same io thread to make
449 : : // progress, so blocking here would deadlock forever rather than
450 : : // eventually clear (#449).
451 : : // #509: the wait predicate (is_backpressure_active(client_id), tied to
452 : : // bp_high) and the transport's own hard queue-byte cap for that session
453 : : // (bp_limit) are different thresholds observed at different times, so a
454 : : // single write attempt can spuriously fail right after the wait exits.
455 : : // Retry capacity races until admission or cancellation. Oversized payloads
456 : : // are rejected by validation before entering this loop.
457 : :
458 : 251 : SendResult send_to_blocking(ClientId client_id, std::string_view data) {
459 : 251 : std::shared_ptr<transport::TcpServer> ts;
460 : 251 : std::shared_ptr<transport::TcpServerSession> session;
461 : 251 : uint64_t generation = 0;
462 : 251 : bool cannot_wait = false;
463 : 0 : const auto result = [&]() -> SendResult {
464 : : {
465 : 251 : std::shared_lock<std::shared_mutex> lock(mutex_);
466 : 251 : ts = transport_cache_;
467 : : auto validation =
468 [ + + + - ]: 251 : detail::validate_payload_size(data.size(), ts ? ts->write_queue_limit(client_id) : std::nullopt);
469 [ + + ]: 251 : if (!validation.accepted()) return validation;
470 : 231 : auto state = send_state(ts);
471 [ + + ]: 231 : if (!state.accepted()) return state;
472 : 212 : state = ts->target_state();
473 [ - + ]: 212 : if (!state.accepted()) return state;
474 : 212 : generation = callback_generation_.load();
475 [ + + + - : 212 : cannot_wait = detail::in_data_callback() || detail::executor_running_here(ts->get_executor());
+ - + + +
+ - - ]
476 : 212 : session = ts->capture_target(client_id);
477 [ + + ]: 212 : if (!session) return SendResult::reject(SendRejection::NotReady);
478 : 251 : }
479 : 213 : auto wait = [&]() -> SendResult {
480 : 213 : std::unique_lock<std::mutex> bp_lock(bp_mutex_);
481 : : {
482 : 213 : std::shared_lock<std::shared_mutex> lock(mutex_);
483 : : // No observed pressure means no capacity wait. Final admission still
484 : : // checks lifecycle and the exact session selected at entry.
485 [ + + + - : 637 : if (!started_.load() || callback_generation_.load() != generation ||
+ + ]
486 : 425 : ts->poll_target_wait(session).has_value())
487 : 98 : return SendResult::accept();
488 : 213 : }
489 [ + + ]: 115 : if (cannot_wait) return SendResult::reject(SendRejection::WouldBlock);
490 [ + + + - ]: 45 : if (auto hook = detail::g_tcp_server_capacity_wait_hook.load()) hook();
491 : 45 : std::optional<SendResult> outcome;
492 : 86 : auto released = [&] {
493 : : // The retained session owns its first terminal cause, even after
494 : : // removal from the map or a later server/wrapper restart.
495 : 86 : outcome = ts->poll_target_wait(session);
496 : 86 : return outcome.has_value();
497 : 45 : };
498 [ + - + + ]: 61 : while (!bp_cv_.wait_for(bp_lock, std::chrono::milliseconds(50), released)) {
499 : : }
500 [ + + + - ]: 45 : if (auto hook = detail::g_tcp_server_capacity_wait_result_hook.load()) hook(*outcome);
501 : 45 : return *outcome;
502 : 213 : };
503 : 201 : for (bool retry = false;; retry = true) {
504 [ + + + - ]: 213 : if (retry) detail::pause_send_retry(bp_cv_, bp_mutex_);
505 : 213 : const auto released = wait();
506 [ + + ]: 320 : if (!released.accepted()) return released;
507 : 119 : std::shared_lock<std::shared_mutex> lock(mutex_);
508 : 119 : const auto state = send_state(ts);
509 [ + + ]: 119 : if (!state.accepted()) return state;
510 [ + + ]: 228 : if (callback_generation_.load() != generation) return SendResult::reject(SendRejection::NotReady);
511 : 220 : const auto admitted = ts->write_target(
512 : 110 : client_id, memory::ConstByteSpan(reinterpret_cast<const uint8_t*>(data.data()), data.size()), false,
513 : : session);
514 [ + + + + : 110 : if (admitted.accepted() || admitted.reason() != SendRejection::WouldBlock) return admitted;
+ + ]
515 [ + + ]: 21 : if (cannot_wait) return admitted;
516 : 131 : }
517 : 251 : }();
518 [ + + + - ]: 251 : if (auto hook = detail::g_tcp_server_send_result_hook.load()) hook(result);
519 : 251 : return result;
520 : 251 : }
521 : :
522 : 378 : void setup_internal_handlers() {
523 [ - + ]: 378 : if (!channel_) return;
524 : :
525 : 378 : std::weak_ptr<bool> weak_alive = alive_marker_;
526 : 378 : std::weak_ptr<Impl> weak_impl = weak_from_this();
527 : : // Opening this run's generation and admitting again are one step under
528 : : // the gate's lock.
529 : 378 : receive_budget_->reset_stats();
530 : 378 : terminal_error_reported_ = false;
531 : 378 : const uint64_t generation = callback_gate_.open_new_generation();
532 : 378 : callback_generation_.store(generation);
533 : 378 : auto transport_server = std::dynamic_pointer_cast<transport::TcpServer>(channel_);
534 [ + + ]: 378 : if (transport_server) {
535 : 371 : transport_server->on_multi_connect([this, generation, weak_impl, weak_alive,
536 : : weak_native = std::weak_ptr<transport::TcpServer>(transport_server)](
537 : : ClientId id, const std::string& info) {
538 : 331 : auto impl_keepalive = weak_impl.lock();
539 [ - + ]: 331 : if (!impl_keepalive) return;
540 : 331 : auto alive = weak_alive.lock();
541 [ - + ]: 331 : if (!alive) return;
542 : 331 : auto lease = callback_gate_.enter(generation);
543 [ - + ]: 331 : if (!lease.admitted()) return;
544 : :
545 : 331 : auto native = weak_native.lock();
546 [ + - + - ]: 331 : auto executor = native ? native->client_executor(id) : std::nullopt;
547 [ - + ]: 331 : if (!executor) return;
548 : 331 : auto receive_scope = receive_budget_->open_scope();
549 [ + + ]: 331 : if (!receive_scope) {
550 : 1 : native->fail_receive(id);
551 : 1 : return;
552 : : }
553 : :
554 : 330 : ConnectionHandler handler;
555 : : {
556 : 330 : std::unique_lock<std::shared_mutex> lock(mutex_);
557 : 330 : batches_[id] = std::make_shared<detail::SessionBatch>(*executor, std::move(receive_scope));
558 [ + + ]: 330 : if (framer_factory_) {
559 : 58 : auto framer = framer_factory_();
560 [ + - ]: 58 : if (framer) {
561 : 58 : auto shared_framer = std::shared_ptr<framer::IFramer>(std::move(framer));
562 : 58 : shared_framer->on_message([this, id, generation](memory::ConstByteSpan msg) {
563 : 1021 : auto message_lease = callback_gate_.enter(generation);
564 [ + + ]: 1021 : if (!message_lease.admitted()) return;
565 : 1020 : auto prepared = detail::take_prepared_message(id, msg);
566 : : // #441: snapshot under a shared_lock (pure read), build the
567 : : // copy before taking the exclusive lock for queue mutation.
568 : : bool batch_mode;
569 : 1020 : interface::SharedCallback<MessageHandler> on_message_handler;
570 : : {
571 : 1020 : std::shared_lock<std::shared_mutex> lock(mutex_);
572 : 1020 : batch_mode = static_cast<bool>(message_batch_handler_);
573 : 1020 : on_message_handler = on_message_;
574 : 1020 : }
575 : :
576 [ + + ]: 1020 : if (batch_mode) {
577 : 12 : std::shared_ptr<detail::SessionBatch> receive;
578 : : {
579 : 12 : std::shared_lock<std::shared_mutex> lock(mutex_);
580 : 12 : auto state = batches_.find(id);
581 [ - + ]: 12 : if (state == batches_.end()) return;
582 : 12 : receive = state->second;
583 : 12 : }
584 [ + - - - ]: 12 : auto ctx = prepared ? std::move(*prepared) : detail::retain_received(receive->receive.scope, id, msg);
585 : 12 : interface::SharedCallback<BatchMessageHandler> flush_handler;
586 : 12 : detail::ReceiveBatch batch;
587 : : {
588 : 12 : std::unique_lock<std::shared_mutex> lock(mutex_);
589 : 12 : auto state = batches_.find(id);
590 [ + - - + : 12 : if (state == batches_.end() || callback_generation_ != generation) return;
- + ]
591 : 12 : state->second->messages.emplace_back(std::move(ctx));
592 [ + - + + ]: 12 : if (batches_.at(id)->messages.size() >= max_batch_size_) {
593 : 4 : flush_handler = message_batch_handler_;
594 : 4 : batch = std::move(batches_.at(id)->messages);
595 : 4 : batches_.at(id)->messages.clear();
596 [ + - + - ]: 8 : } else if (batches_.at(id)->messages.size() == 1) {
597 : 8 : schedule_batch_timer(generation, id);
598 : : }
599 : 12 : }
600 : 12 : detail::invoke_user_callback("tcp_server", "on_message_batch", flush_handler, batch);
601 : 12 : return;
602 : 12 : }
603 : :
604 : 1008 : detail::invoke_user_callback("tcp_server", "on_message", on_message_handler, MessageContext(id, msg));
605 : 1045 : });
606 : 58 : framers_[id] = std::move(shared_framer);
607 : 58 : }
608 : 58 : }
609 : 330 : handler = on_client_connect_;
610 : 330 : }
611 : 330 : detail::invoke_user_callback("tcp_server", "on_connect", handler, ConnectionContext(id, info));
612 : 336 : });
613 : 742 : transport_server->on_multi_data(
614 : 742 : [this, generation, weak_impl, weak_alive](ClientId id, memory::ConstByteSpan data_span) {
615 : 2652 : auto impl_keepalive = weak_impl.lock();
616 [ - + ]: 2652 : if (!impl_keepalive) return;
617 : 2652 : auto alive = weak_alive.lock();
618 [ - + ]: 2652 : if (!alive) return;
619 : :
620 : : // #449: everything below runs synchronously on this io thread -
621 : : // mark it so a blocking send_to()/broadcast() called from within
622 : : // one of these callbacks fails fast instead of deadlocking.
623 : 2652 : auto lease = callback_gate_.enter(generation);
624 [ - + ]: 2652 : if (!lease.admitted()) return;
625 : 2652 : detail::CallbackGuard callback_guard;
626 : :
627 : : // #441: snapshot the handler/framer pointers under a shared_lock
628 : : // (not unique_lock) - this is a pure read, matching try_send's
629 : : // locking level so it no longer blocks concurrent sends even
630 : : // briefly.
631 : : bool batch_mode;
632 : 2652 : std::shared_ptr<detail::SessionBatch> receive;
633 : 2652 : interface::SharedCallback<MessageHandler> handler;
634 : 2652 : std::shared_ptr<framer::IFramer> framer;
635 : : {
636 : 2652 : std::shared_lock<std::shared_mutex> lock(mutex_);
637 : 2652 : auto state = batches_.find(id);
638 [ - + ]: 2652 : if (state == batches_.end()) return;
639 : 2652 : receive = state->second;
640 : 2652 : batch_mode = static_cast<bool>(data_batch_handler_);
641 : 2652 : handler = on_data_;
642 : 2652 : auto it = framers_.find(id);
643 [ + + ]: 2652 : if (it != framers_.end()) {
644 : 1027 : framer = it->second;
645 : : }
646 : 2652 : }
647 : :
648 : : try {
649 : 2652 : auto prepared = detail::prepare_receive(receive->receive, framer, id, data_span, batch_mode);
650 [ + + ]: 2650 : if (batch_mode) {
651 : : // #441: build the copy before taking the exclusive lock, so the
652 : : // lock is only held for the queue mutation itself, not the
653 : : // allocation.
654 : 11 : auto ctx = std::move(*prepared.raw);
655 : 11 : interface::SharedCallback<BatchMessageHandler> flush_handler;
656 : 11 : detail::ReceiveBatch batch;
657 : : {
658 : 11 : std::unique_lock<std::shared_mutex> lock(mutex_);
659 : 11 : auto state = batches_.find(id);
660 [ + - - + : 11 : if (state == batches_.end() || callback_generation_ != generation) return;
- + ]
661 : 11 : state->second->data.emplace_back(std::move(ctx));
662 [ + - + + ]: 11 : if (batches_.at(id)->data.size() >= max_batch_size_) {
663 : 4 : flush_handler = data_batch_handler_;
664 : 4 : batch = std::move(batches_.at(id)->data);
665 : 4 : batches_.at(id)->data.clear();
666 [ + - + - ]: 7 : } else if (batches_.at(id)->data.size() == 1) {
667 : 7 : schedule_batch_timer(generation, id);
668 : : }
669 : 11 : }
670 : 11 : detail::invoke_user_callback("tcp_server", "on_data_batch", flush_handler, batch);
671 : 11 : } else {
672 : 5278 : detail::invoke_user_callback("tcp_server", "on_data", handler, MessageContext(id, data_span));
673 : : }
674 : :
675 : 2650 : prepared.deliver();
676 : 2652 : } catch (const detail::ReceiveOverflow& overflow) {
677 : 2 : receive->receive.scope->overflow(data_span.size(), overflow.reason);
678 : :
679 : : {
680 : 2 : std::unique_lock<std::shared_mutex> lock(mutex_);
681 : 2 : receive->data.clear();
682 : 2 : receive->messages.clear();
683 : 2 : receive->receive.reset(framer.get());
684 : 2 : }
685 [ + - + - ]: 2 : if (auto native = std::dynamic_pointer_cast<transport::TcpServer>(channel_)) native->fail_receive(id);
686 : :
687 : 2 : } catch (const std::bad_alloc&) {
688 : 0 : receive->receive.scope->overflow(data_span.size(), ReceiveOverflowReason::AllocationFailure);
689 : :
690 : : {
691 : 0 : std::unique_lock<std::shared_mutex> lock(mutex_);
692 : 0 : receive->data.clear();
693 : 0 : receive->messages.clear();
694 : 0 : receive->receive.reset(framer.get());
695 : 0 : }
696 [ - - - - ]: 0 : if (auto native = std::dynamic_pointer_cast<transport::TcpServer>(channel_)) native->fail_receive(id);
697 : 0 : }
698 : 2652 : });
699 : 371 : transport_server->on_multi_disconnect([this, generation, weak_impl, weak_alive](ClientId id) {
700 : 210 : auto impl_keepalive = weak_impl.lock();
701 [ - + ]: 210 : if (!impl_keepalive) return;
702 : 210 : auto alive = weak_alive.lock();
703 [ - + ]: 210 : if (!alive) return;
704 : 210 : auto lease = callback_gate_.enter(generation);
705 [ - + ]: 210 : if (!lease.admitted()) return;
706 : :
707 : 210 : std::shared_ptr<detail::SessionBatch> pending;
708 : : {
709 : 210 : std::shared_lock<std::shared_mutex> lock(mutex_);
710 : 210 : auto it = batches_.find(id);
711 [ + + ]: 210 : if (it != batches_.end()) pending = it->second;
712 : 210 : }
713 [ + + ]: 210 : if (!pending) return;
714 : 209 : flush_batches(generation, id, pending);
715 : 209 : auto after_flush = callback_gate_.enter(generation);
716 [ - + ]: 209 : if (!after_flush.admitted()) return;
717 : 209 : ConnectionHandler handler;
718 : : {
719 : 209 : std::unique_lock<std::shared_mutex> lock(mutex_);
720 : 209 : batches_.erase(id);
721 : 209 : framers_.erase(id);
722 : 209 : handler = on_disconnect_;
723 : 209 : }
724 : 209 : detail::invoke_user_callback("tcp_server", "on_disconnect", handler, ConnectionContext(id));
725 : 213 : });
726 : :
727 : 371 : transport_server->on_backpressure([this, generation, weak_impl, weak_alive](size_t queued) {
728 : 246 : bp_cv_.notify_all();
729 : 246 : auto impl_keepalive = weak_impl.lock();
730 [ - + ]: 246 : if (!impl_keepalive) return;
731 : 246 : auto alive = weak_alive.lock();
732 [ - + ]: 246 : if (!alive) return;
733 : 246 : auto lease = callback_gate_.enter(generation);
734 [ - + ]: 246 : if (!lease.admitted()) return;
735 : 246 : std::function<void(size_t)> handler;
736 : : {
737 : 246 : std::shared_lock<std::shared_mutex> lock(mutex_);
738 : 246 : handler = on_backpressure_;
739 : 246 : }
740 : 246 : detail::invoke_user_callback("tcp_server", "on_backpressure", handler, queued);
741 : 246 : });
742 : : }
743 : 378 : channel_->on_state([this, generation, weak_impl, weak_alive](base::LinkState state) {
744 : 757 : auto impl_keepalive = weak_impl.lock();
745 [ - + ]: 757 : if (!impl_keepalive) return;
746 : 757 : auto alive = weak_alive.lock();
747 [ + + ]: 757 : if (!alive) return;
748 : 756 : auto lease = callback_gate_.enter(generation);
749 [ + + ]: 756 : if (!lease.admitted()) return;
750 : :
751 [ + + ]: 749 : if (state == base::LinkState::Listening) {
752 : 428 : is_listening_.store(true);
753 : 428 : std::unique_lock<std::shared_mutex> lock(mutex_);
754 : 428 : terminal_error_reported_ = false;
755 : 428 : fulfill_all_locked(true);
756 [ + + + - : 749 : } else if (state == base::LinkState::Error || state == base::LinkState::Closed ||
- + ]
757 : : state == base::LinkState::Idle) {
758 : 4 : ErrorHandler handler;
759 : 4 : is_listening_.store(false);
760 : : {
761 : 4 : std::unique_lock<std::shared_mutex> lock(mutex_);
762 : 4 : fulfill_all_locked(false);
763 [ + - + - ]: 4 : if (state == base::LinkState::Error && !terminal_error_reported_) {
764 : 4 : terminal_error_reported_ = true;
765 : 4 : handler = on_error_;
766 : : }
767 : 4 : }
768 [ + - ]: 4 : detail::invoke_user_callback("tcp_server", "on_error", handler,
769 [ + - + - : 8 : channel_ ? detail::build_error_context(*channel_, "Server error")
- - ]
770 : : : ErrorContext(ErrorCode::IoError, "Server error"));
771 : 4 : }
772 : 772 : });
773 : 378 : }
774 : :
775 : 1985 : RuntimeStats stats() const {
776 : 1985 : std::shared_lock<std::shared_mutex> lock(mutex_);
777 [ + - + - ]: 3970 : return channel_ ? channel_->stats() : RuntimeStats{};
778 : 1985 : }
779 : :
780 : 3 : void reset_stats() {
781 : 3 : std::shared_lock<std::shared_mutex> lock(mutex_);
782 : 3 : receive_budget_->reset_stats();
783 [ + - + - ]: 3 : if (channel_) channel_->reset_stats();
784 : 3 : }
785 : : };
786 : :
787 : 182 : TcpServer::TcpServer(uint16_t port) : impl_(std::make_shared<Impl>(port)) {}
788 : 6 : TcpServer::TcpServer(uint16_t port, std::shared_ptr<boost::asio::io_context> ioc)
789 : 6 : : impl_(std::make_shared<Impl>(port, ioc)) {}
790 : 96 : TcpServer::TcpServer(std::shared_ptr<interface::Channel> ch) : impl_(std::make_shared<Impl>(ch)) {
791 : 96 : impl_->setup_internal_handlers();
792 : 96 : }
793 : : // Stop on the destroying thread first. A callback in flight holds the Impl
794 : : // alive, and if its reference were the last one, ~Impl would run on the io
795 : : // thread, where the shutdown cannot join that thread (#613).
796 : 422 : TcpServer::~TcpServer() {
797 [ - + ]: 284 : if (!impl_) return;
798 : : try {
799 : 284 : impl_->stop();
800 : 0 : } catch (...) {
801 : 0 : }
802 : 422 : }
803 : :
804 : 0 : TcpServer::TcpServer(TcpServer&&) noexcept = default;
805 : : // The previous Impl is handed to a temporary, whose destructor stops it here.
806 : 0 : TcpServer& TcpServer::operator=(TcpServer&& other) noexcept {
807 [ # # ]: 0 : if (this != &other) {
808 : 0 : TcpServer previous(std::move(*this));
809 : 0 : impl_ = std::move(other.impl_);
810 : 0 : }
811 : 0 : return *this;
812 : : }
813 : :
814 : 9 : TcpServer& TcpServer::receive_limits(ReceiveLimits limits) {
815 : 9 : limits.validate();
816 : 8 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
817 [ + + + - : 22 : if (impl_->started_ || impl_->stop_callers_.load() || (impl_->stop_requested_ && impl_->alive_marker_) ||
+ + + - -
+ + + ]
818 : 7 : detail::in_data_callback())
819 : 1 : throw std::logic_error("receive limits require completed stop");
820 : 7 : auto budget = std::make_shared<detail::ReceiveBudget>(limits);
821 : 7 : impl_->receive_limits_ = limits;
822 : 7 : impl_->receive_budget_ = std::move(budget);
823 : 7 : return *this;
824 : 8 : }
825 : 34 : ReceiveMemoryStats TcpServer::receive_stats() const {
826 : 34 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
827 : 68 : return impl_->receive_budget_->stats();
828 : 34 : }
829 : :
830 : 284 : std::future<bool> TcpServer::start() { return impl_->start(); }
831 : 370 : void TcpServer::stop() { impl_->stop(); }
832 : 94 : bool TcpServer::listening() const { return get_impl()->is_listening_.load(); }
833 : 1985 : RuntimeStats TcpServer::stats() const { return get_impl()->stats(); }
834 : 3 : void TcpServer::reset_stats() { impl_->reset_stats(); }
835 : :
836 : 1210 : FanoutResult TcpServer::broadcast(std::string_view data) { return impl_->broadcast(data); }
837 : 30 : FanoutResult TcpServer::try_broadcast(std::string_view data) { return impl_->try_broadcast(data); }
838 : 168 : SendResult TcpServer::send_to(ClientId client_id, std::string_view data) { return impl_->send_to(client_id, data); }
839 : 76 : SendResult TcpServer::try_send_to(ClientId client_id, std::string_view data) {
840 : 76 : return impl_->try_send_to(client_id, data);
841 : : }
842 : :
843 : 101 : SendResult TcpServer::send_to_blocking(ClientId client_id, std::string_view data) {
844 : 101 : return impl_->send_to_blocking(client_id, data);
845 : : }
846 : :
847 : 6 : FanoutResult TcpServer::broadcast_line(std::string_view line) { return impl_->try_broadcast(line, true); }
848 : 82 : SendResult TcpServer::send_to_line(ClientId client_id, std::string_view line) {
849 : 246 : return send_to(client_id, std::string(line) + "\n");
850 : : }
851 : 14 : FanoutResult TcpServer::try_broadcast_line(std::string_view line) { return impl_->try_broadcast(line, true); }
852 : 19 : SendResult TcpServer::try_send_to_line(ClientId client_id, std::string_view line) {
853 : 57 : return try_send_to(client_id, std::string(line) + "\n");
854 : : }
855 : :
856 : 148 : TcpServer& TcpServer::on_connect(ConnectionHandler h) {
857 : 148 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
858 : 148 : impl_->on_client_connect_ = std::move(h);
859 : 148 : return *this;
860 : 148 : }
861 : 15 : TcpServer& TcpServer::on_disconnect(ConnectionHandler h) {
862 : 15 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
863 : 15 : impl_->on_disconnect_ = std::move(h);
864 : 15 : return *this;
865 : 15 : }
866 : 107 : TcpServer& TcpServer::on_data(MessageHandler h) {
867 : 107 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
868 : 107 : impl_->on_data_ = interface::share_callback(std::move(h));
869 : 107 : return *this;
870 : 107 : }
871 : 10 : TcpServer& TcpServer::on_data_batch(BatchMessageHandler h) {
872 : 10 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
873 : 10 : impl_->data_batch_handler_ = interface::share_callback(std::move(h));
874 : 10 : return *this;
875 : 10 : }
876 : 104 : TcpServer& TcpServer::on_error(ErrorHandler h) {
877 : 104 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
878 : 104 : impl_->on_error_ = std::move(h);
879 : 104 : return *this;
880 : 104 : }
881 : :
882 : 31 : TcpServer& TcpServer::on_backpressure(std::function<void(size_t)> h) {
883 : 31 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
884 : 31 : impl_->on_backpressure_ = std::move(h);
885 : 31 : return *this;
886 : 31 : }
887 : :
888 : 26 : TcpServer& TcpServer::framer(FramerFactory factory) {
889 : 26 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
890 : 26 : impl_->framer_factory_ = std::move(factory);
891 : 26 : return *this;
892 : 26 : }
893 : :
894 : 10 : TcpServer& TcpServer::on_message(MessageHandler handler) {
895 : 10 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
896 : 10 : impl_->on_message_ = interface::share_callback(std::move(handler));
897 : 10 : return *this;
898 : 10 : }
899 : :
900 : 8 : TcpServer& TcpServer::on_message_batch(BatchMessageHandler h) {
901 : 8 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
902 : 8 : impl_->message_batch_handler_ = interface::share_callback(std::move(h));
903 : 8 : return *this;
904 : 8 : }
905 : :
906 : 189 : size_t TcpServer::client_count() const {
907 : 189 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
908 : 189 : const auto& ts = get_impl()->transport_cache_;
909 [ + + + - ]: 378 : return ts ? ts->client_count() : 0;
910 : 189 : }
911 : :
912 : 8 : std::vector<ClientId> TcpServer::connected_clients() const {
913 : 8 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
914 : 8 : const auto& ts = get_impl()->transport_cache_;
915 [ + + + - ]: 16 : return ts ? ts->connected_clients() : std::vector<ClientId>();
916 : 8 : }
917 : :
918 : 58 : std::optional<RuntimeStats> TcpServer::client_stats(ClientId client_id) const {
919 : 58 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
920 : 58 : const auto& ts = get_impl()->transport_cache_;
921 [ + - + - ]: 116 : return ts ? ts->client_stats(client_id) : std::nullopt;
922 : 58 : }
923 : :
924 : 1 : TcpServer& TcpServer::auto_start(bool m) {
925 : 1 : impl_->auto_start_.store(m);
926 [ + - + - : 1 : if (impl_->auto_start_.load() && !impl_->started_.load()) start();
+ - + - ]
927 : 1 : return *this;
928 : : }
929 : :
930 : 1 : TcpServer& TcpServer::bind_address(const std::string& address) {
931 [ + - - + : 1 : config::detail::require(util::InputValidator::is_valid_ipv4(address) || util::InputValidator::is_valid_ipv6(address),
- - - - +
- ]
932 : : "invalid bind address");
933 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
934 : 1 : impl_->require_stopped_config();
935 : 1 : impl_->bind_address_ = address;
936 : 1 : return *this;
937 : 1 : }
938 : :
939 : 2 : TcpServer& TcpServer::shared_context(bool use_shared) {
940 : 2 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
941 : 2 : impl_->require_stopped_config();
942 : 2 : impl_->shared_context_.store(use_shared);
943 : 2 : return *this;
944 : 2 : }
945 : :
946 : 4 : TcpServer& TcpServer::port_retry(bool e, int m, int i) {
947 : 4 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
948 : 4 : impl_->require_stopped_config();
949 : 4 : config::detail::range(m, 0, base::constants::MAX_RETRIES_LIMIT, "invalid port retry count");
950 : 4 : config::detail::range(i, base::constants::MIN_RETRY_INTERVAL_MS, base::constants::MAX_RETRY_INTERVAL_MS,
951 : : "invalid port retry interval");
952 : 4 : impl_->port_retry_enabled_.store(e);
953 : 4 : impl_->max_port_retries_.store(m);
954 : 4 : impl_->port_retry_interval_ms_.store(i);
955 : 4 : return *this;
956 : 4 : }
957 : :
958 : 4 : TcpServer& TcpServer::idle_timeout(std::chrono::milliseconds timeout) {
959 : 4 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
960 : 4 : impl_->require_stopped_config();
961 : 3 : config::detail::duration(timeout, base::constants::MIN_IDLE_TIMEOUT_MS, base::constants::MAX_IDLE_TIMEOUT_MS, true,
962 : : "invalid idle_timeout");
963 : 3 : impl_->idle_timeout_ms_.store(static_cast<int>(timeout.count()));
964 : 3 : return *this;
965 : 4 : }
966 : :
967 : 17 : TcpServer& TcpServer::max_clients(size_t max) {
968 : 17 : config::detail::range(max, 0, base::constants::MAX_MAX_CONNECTIONS, "invalid client limit");
969 : 16 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
970 : 16 : impl_->max_clients_.store(max);
971 [ + + ]: 16 : if (max == 0) {
972 : 2 : impl_->client_limit_enabled_.store(false);
973 : : } else {
974 : 14 : impl_->client_limit_enabled_.store(true);
975 : : }
976 [ + + + - ]: 16 : if (impl_->transport_cache_) impl_->transport_cache_->set_client_limit(max);
977 : 16 : return *this;
978 : 16 : }
979 : :
980 : 163 : TcpServer& TcpServer::backpressure_threshold(size_t threshold) {
981 : 163 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
982 : 163 : impl_->require_stopped_config();
983 : 162 : config::detail::range(threshold, base::constants::MIN_BACKPRESSURE_THRESHOLD,
984 : : base::constants::MAX_BACKPRESSURE_THRESHOLD, "invalid backpressure_threshold");
985 : 162 : impl_->backpressure_threshold_.store(threshold);
986 : 162 : return *this;
987 : 163 : }
988 : :
989 : 44 : TcpServer& TcpServer::backpressure_strategy(base::constants::BackpressureStrategy strategy) {
990 : 44 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
991 : 44 : impl_->require_stopped_config();
992 : 43 : config::detail::strategy(strategy);
993 : 43 : impl_->backpressure_strategy_.store(strategy);
994 : 43 : return *this;
995 : 44 : }
996 : :
997 : 10 : size_t TcpServer::backpressure_threshold() const { return impl_->backpressure_threshold_.load(); }
998 : :
999 : 3 : base::constants::BackpressureStrategy TcpServer::backpressure_strategy() const {
1000 : 3 : return impl_->backpressure_strategy_.load();
1001 : : }
1002 : :
1003 : 1 : TcpServer& TcpServer::tcp_no_delay(bool enable) {
1004 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1005 : 1 : impl_->require_stopped_config();
1006 : 1 : impl_->tcp_no_delay_.store(enable);
1007 : 1 : return *this;
1008 : 1 : }
1009 : :
1010 : 1 : TcpServer& TcpServer::keep_alive(bool enable) {
1011 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1012 : 1 : impl_->require_stopped_config();
1013 : 1 : impl_->keep_alive_.store(enable);
1014 : 1 : return *this;
1015 : 1 : }
1016 : :
1017 : 8 : TcpServer& TcpServer::tls(const std::string& certificate_file, const std::string& private_key_file) {
1018 [ + + ]: 8 : config::detail::require(certificate_file.empty() == private_key_file.empty(), "TLS requires certificate and key");
1019 : 6 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1020 : 6 : impl_->require_stopped_config();
1021 : 6 : impl_->tls_certificate_file_ = certificate_file;
1022 : 6 : impl_->tls_private_key_file_ = private_key_file;
1023 : 6 : return *this;
1024 : 6 : }
1025 : :
1026 : 3 : TcpServer& TcpServer::send_buffer_size(size_t bytes) {
1027 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1028 : 3 : impl_->require_stopped_config();
1029 [ + - ]: 3 : if (bytes != 0)
1030 : 3 : config::detail::range(bytes, base::constants::MIN_SOCKET_BUFFER_SIZE, base::constants::MAX_SOCKET_BUFFER_SIZE,
1031 : : "invalid send_buffer_size");
1032 : 3 : impl_->send_buffer_size_.store(bytes);
1033 : 3 : return *this;
1034 : 3 : }
1035 : :
1036 : 1 : TcpServer& TcpServer::receive_buffer_size(size_t bytes) {
1037 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1038 : 1 : impl_->require_stopped_config();
1039 [ + - ]: 1 : if (bytes != 0)
1040 : 1 : config::detail::range(bytes, base::constants::MIN_SOCKET_BUFFER_SIZE, base::constants::MAX_SOCKET_BUFFER_SIZE,
1041 : : "invalid receive_buffer_size");
1042 : 1 : impl_->receive_buffer_size_.store(bytes);
1043 : 1 : return *this;
1044 : 1 : }
1045 : :
1046 : 2 : TcpServer& TcpServer::read_buffer_size(size_t bytes) {
1047 : 2 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1048 : 2 : impl_->require_stopped_config();
1049 : 2 : config::detail::range(bytes, base::constants::MIN_READ_BUFFER_SIZE, base::constants::MAX_READ_BUFFER_SIZE,
1050 : : "invalid read_buffer_size");
1051 : 2 : impl_->read_buffer_size_.store(bytes);
1052 : 2 : return *this;
1053 : 2 : }
1054 : :
1055 : 4 : TcpServer& TcpServer::manage_external_context(bool m) {
1056 : 4 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1057 : 4 : impl_->require_stopped_config();
1058 : 3 : impl_->manage_external_context_.store(m);
1059 : 3 : return *this;
1060 : 4 : }
1061 : :
1062 : 23 : TcpServer& TcpServer::batch_size(size_t size) {
1063 [ + - ]: 23 : config::detail::require(size > 0, "batch size must be positive");
1064 : 23 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1065 : 23 : impl_->max_batch_size_ = size;
1066 : 23 : return *this;
1067 : 23 : }
1068 : :
1069 : 23 : TcpServer& TcpServer::batch_latency(std::chrono::milliseconds latency) {
1070 : 23 : config::detail::duration(latency, 0, std::numeric_limits<int>::max(), true, "invalid batch latency");
1071 : 23 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1072 : 23 : impl_->max_batch_latency_ = latency;
1073 : 23 : return *this;
1074 : 23 : }
1075 : :
1076 : : } // namespace wrapper
1077 : : } // namespace wirestead
|
