Branch data Line data Source code
1 : : /*
2 : : * Copyright 2025 Jinwoo Sung
3 : : *
4 : : * Licensed under the Apache License, Version 2.0 (the "License");
5 : : * you may not use this file except in compliance with the License.
6 : : * You may obtain a copy of the License at
7 : : *
8 : : * http://www.apache.org/licenses/LICENSE-2.0
9 : : *
10 : : * Unless required by applicable law or agreed to in writing, software
11 : : * distributed under the License is distributed on an "AS IS" BASIS,
12 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 : : * See the License for the specific language governing permissions and
14 : : * limitations under the License.
15 : : */
16 : :
17 : : #include "wirestead/transport/tcp_server/tcp_server.hpp"
18 : :
19 : : #include <spdlog/fmt/fmt.h>
20 : :
21 : : #include <algorithm>
22 : : #include <atomic>
23 : : #include <boost/asio.hpp>
24 : : #include <condition_variable>
25 : : #include <future>
26 : : #include <iostream>
27 : : #include <mutex>
28 : : #include <stop_token>
29 : : #include <string_view>
30 : : #include <thread>
31 : : #include <unordered_map>
32 : :
33 : : #include "wirestead/concurrency/io_context_manager.hpp"
34 : : #include "wirestead/concurrency/io_thread_hook.hpp"
35 : : #include "wirestead/concurrency/thread_safe_state.hpp"
36 : : #include "wirestead/config/validation.hpp"
37 : : #include "wirestead/diagnostics/callback.hpp"
38 : : #include "wirestead/diagnostics/exceptions.hpp"
39 : : #include "wirestead/diagnostics/logger.hpp"
40 : : #include "wirestead/diagnostics/runtime_stats_counter.hpp"
41 : : #include "wirestead/interface/itcp_acceptor.hpp"
42 : : #include "wirestead/transport/base/error_info_holder.hpp"
43 : : #include "wirestead/transport/base/stop_test_hook.hpp"
44 : : #include "wirestead/transport/tcp_server/boost_tcp_acceptor.hpp"
45 : : #include "wirestead/transport/tcp_server/ssl_tcp_socket.hpp"
46 : : #include "wirestead/transport/tcp_server/tcp_server_session.hpp"
47 : : #include "wirestead/wrapper/send_validation.hpp"
48 : :
49 : : namespace wirestead {
50 : : namespace transport {
51 : :
52 : : namespace net = boost::asio;
53 : : using tcp = net::ip::tcp;
54 : :
55 : : struct TcpServer::Impl {
56 : : std::atomic<bool> stopping_{false};
57 : : // #503: stop() can call perform_cleanup() from two different paths - the
58 : : // dispatched-onto-the-io_context call, and (if that doesn't complete
59 : : // within the timeout below) a direct fallback call on the stopping
60 : : // thread. Under slow/instrumented builds (TSAN) the dispatched call can
61 : : // still be mid-flight when the timeout fires, so both could run
62 : : // perform_cleanup()'s body concurrently - including acceptor_->close(),
63 : : // racing with the io_context thread's own concurrent async_accept.
64 : : // Guards perform_cleanup() to run its body at most once per stop cycle,
65 : : // reset back to false in start() alongside stopping_.
66 : : std::atomic<bool> cleanup_started_{false};
67 : : std::atomic<ClientId> next_client_id_{0};
68 : :
69 : : std::unique_ptr<net::io_context> owned_ioc_;
70 : : bool owns_ioc_;
71 : : bool uses_shared_context_{false};
72 : : net::io_context& ioc_;
73 : : net::strand<net::io_context::executor_type> strand_;
74 : : std::atomic<uint64_t> generation_{0};
75 : : // start() may fail validation before dispatching any executor work.
76 : : bool run_dispatched_ = false;
77 : : std::unique_ptr<net::executor_work_guard<net::io_context::executor_type>> work_guard_;
78 : : std::jthread ioc_thread_;
79 : :
80 : : std::unique_ptr<interface::TcpAcceptorInterface> acceptor_;
81 : : config::TcpServerConfig cfg_;
82 : :
83 : 302 : concurrency::AtomicLinkState state_{base::LinkState::Idle};
84 : : // Shared snapshots for the handlers the io thread copies out per received
85 : : // chunk - a std::function copy allocates whenever the target outgrows its
86 : : // small-object buffer. See interface::SharedCallback. The connect/disconnect
87 : : // handlers below stay plain: they fire once per connection, not per chunk.
88 : : interface::SharedCallback<OnBytes> on_bytes_;
89 : : interface::SharedCallback<OnState> on_state_;
90 : : interface::SharedCallback<OnBackpressure> on_bp_;
91 : : MultiClientConnectHandler on_multi_connect_;
92 : : interface::SharedCallback<MultiClientDataHandler> on_multi_data_;
93 : : MultiClientDisconnectHandler on_multi_disconnect_;
94 : : diagnostics::RuntimeStatsCounters stats_;
95 : :
96 : : mutable std::mutex sessions_mutex_;
97 : : // D-1: shutdown requested and shutdown completed are separate states. The
98 : : // cleanup that tears the sessions down is what completes it, so waiting
99 : : // callers wait for its signal rather than for a lock.
100 : : std::mutex target_admission_mtx_;
101 : : std::mutex stop_mtx_;
102 : : std::condition_variable stop_cv_;
103 : : bool cleanup_done_ = false;
104 : : std::mutex join_mtx_;
105 : :
106 : 307 : void mark_cleanup_done() {
107 : 307 : detail::stop_test_hook(this, true);
108 : : {
109 : 307 : std::lock_guard<std::mutex> lock(stop_mtx_);
110 : 307 : cleanup_done_ = true;
111 : 307 : }
112 : 307 : stop_cv_.notify_all();
113 : 307 : }
114 : :
115 : : // Waits for the teardown that was dispatched onto the context, and nothing
116 : : // more: the contract's precondition is that the context's owner keeps it
117 : : // running, and running it here would execute unrelated handlers - another
118 : : // channel's user callbacks included - on the stopping thread.
119 : 845 : void wait_for_cleanup() {
120 : 845 : detail::stop_test_hook(this, false);
121 : 845 : std::unique_lock<std::mutex> lock(stop_mtx_);
122 : 1976 : stop_cv_.wait(lock, [this] { return cleanup_done_; });
123 : 845 : }
124 : : std::unordered_map<ClientId, std::shared_ptr<TcpServerSession>> sessions_;
125 : : // Stop removes targets immediately, but their counters remain visible until
126 : : // asynchronous cleanup transfers the final snapshot exactly once.
127 : : std::unordered_map<ClientId, std::shared_ptr<TcpServerSession>> retiring_sessions_;
128 : : wrapper::SendAccounting closed_send_accounting_;
129 : :
130 : 348 : void absorb_session(const std::shared_ptr<TcpServerSession>& session) {
131 : 348 : const auto snapshot = session->stats();
132 : 348 : stats_.absorb(snapshot);
133 : 348 : diagnostics::accumulate_send_accounting(closed_send_accounting_, *snapshot.send_accounting);
134 : 348 : }
135 : :
136 : : size_t max_clients_;
137 : : bool client_limit_enabled_;
138 : :
139 : : std::shared_ptr<TcpServerSession> current_session_;
140 : :
141 : : ErrorInfoHolder error_info_holder_{"tcp_server"};
142 : :
143 : 191 : explicit Impl(const config::TcpServerConfig& cfg, bool use_shared_context)
144 [ + + ]: 191 : : owned_ioc_(use_shared_context ? nullptr : std::make_unique<net::io_context>()),
145 : 191 : owns_ioc_(!use_shared_context),
146 : 191 : uses_shared_context_(use_shared_context),
147 [ + + + - : 191 : ioc_(use_shared_context ? concurrency::IoContextManager::instance().get_context() : *owned_ioc_),
+ - ]
148 : 191 : strand_(net::make_strand(ioc_)),
149 : 191 : cfg_(cfg),
150 : 191 : max_clients_(cfg.max_connections > 0 ? static_cast<size_t>(cfg.max_connections) : 0),
151 [ + - ]: 955 : client_limit_enabled_(cfg.max_connections > 0) {
152 : : try {
153 : 191 : acceptor_ = std::make_unique<BoostTcpAcceptor>(ioc_);
154 : 0 : } catch (const std::exception& e) {
155 : 0 : throw diagnostics::BuilderException("Failed to create TCP acceptor: " + std::string(e.what()), "tcp_server");
156 : 0 : }
157 : 191 : config::detail::validate(cfg_);
158 : 188 : max_clients_ = cfg_.max_connections > 0 ? static_cast<size_t>(cfg_.max_connections) : 0;
159 : 188 : client_limit_enabled_ = cfg_.max_connections > 0;
160 : 242 : }
161 : :
162 : 111 : Impl(const config::TcpServerConfig& cfg, std::unique_ptr<interface::TcpAcceptorInterface> acceptor,
163 : : net::io_context& ioc)
164 : 222 : : owns_ioc_(false),
165 : 111 : ioc_(ioc),
166 : 111 : strand_(net::make_strand(ioc_)),
167 : 111 : acceptor_(std::move(acceptor)),
168 : 111 : cfg_(cfg),
169 : 111 : max_clients_(cfg.max_connections > 0 ? static_cast<size_t>(cfg.max_connections) : 0),
170 [ + - + - ]: 444 : client_limit_enabled_(cfg.max_connections > 0) {
171 [ + + ]: 111 : if (!acceptor_) {
172 : 7 : throw diagnostics::BuilderException("Failed to create TCP acceptor", "tcp_server");
173 : : }
174 : 110 : config::detail::validate(cfg_);
175 : 108 : max_clients_ = cfg_.max_connections > 0 ? static_cast<size_t>(cfg_.max_connections) : 0;
176 : 108 : client_limit_enabled_ = cfg_.max_connections > 0;
177 : 162 : }
178 : :
179 : 296 : ~Impl() {
180 : : try {
181 : 296 : stopping_.store(true);
182 [ - + ]: 296 : if (ioc_thread_.joinable()) {
183 [ # # ]: 0 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
184 : 0 : ioc_thread_.detach();
185 : : } else {
186 : 0 : ioc_thread_.request_stop();
187 : 0 : ioc_thread_.join();
188 : : }
189 : : }
190 : 296 : perform_cleanup();
191 : 0 : } catch (...) {
192 : 0 : }
193 : 296 : }
194 : :
195 : 805 : void notify_state() {
196 [ + + ]: 805 : if (stopping_.load()) return;
197 : 788 : interface::SharedCallback<OnState> cb;
198 : : {
199 : 788 : std::lock_guard<std::mutex> lock(sessions_mutex_);
200 : 788 : cb = on_state_;
201 : 788 : }
202 : 788 : diagnostics::invoke_callback("tcp_server", "on_state", cb, state_.get());
203 : 788 : }
204 : :
205 : : // One context for the whole server, shared by every accepted connection.
206 : : // Built once in start() so a bad certificate fails there rather than on the
207 : : // first client.
208 : : #ifdef WIRESTEAD_TLS_ENABLED
209 : : std::shared_ptr<boost::asio::ssl::context> ssl_context_;
210 : : #endif
211 : :
212 : : // Builds the ssl::context, or reports why it cannot. Called from start(), so
213 : : // a certificate problem surfaces as a start failure with a message rather
214 : : // than as connections that mysteriously drop at handshake.
215 : 305 : std::string init_tls() {
216 : : // Half a TLS config is a request for TLS, not a request for plaintext. An
217 : : // empty environment variable or a typo would otherwise leave tls_enabled()
218 : : // false and bring the server up unencrypted without saying anything, which
219 : : // is the one outcome this whole path exists to prevent.
220 [ - + ]: 305 : if (!cfg_.tls_certificate_file.empty() != !cfg_.tls_private_key_file.empty()) {
221 : 0 : return cfg_.tls_certificate_file.empty() ? "TLS needs a certificate as well as a private key"
222 : 0 : : "TLS needs a private key as well as a certificate";
223 : : }
224 [ + + ]: 305 : if (!cfg_.tls_enabled()) return {};
225 : : #ifndef WIRESTEAD_TLS_ENABLED
226 : : return "TLS was configured but this build has WIRESTEAD_ENABLE_TLS=OFF";
227 : : #else
228 : : namespace ssl = boost::asio::ssl;
229 : : try {
230 : 6 : auto ctx = std::make_shared<ssl::context>(ssl::context::tls_server);
231 : : // Anything below TLS 1.2 is broken in ways nobody should opt into by
232 : : // accident, so the floor is not configurable.
233 : 6 : ctx->set_options(ssl::context::default_workarounds | ssl::context::no_sslv2 | ssl::context::no_sslv3 |
234 : : ssl::context::no_tlsv1 | ssl::context::no_tlsv1_1 | ssl::context::single_dh_use);
235 : 6 : ctx->use_certificate_chain_file(cfg_.tls_certificate_file);
236 : 5 : ctx->use_private_key_file(cfg_.tls_private_key_file, ssl::context::pem);
237 : 5 : ssl_context_ = std::move(ctx);
238 : 5 : return {};
239 : 7 : } catch (const std::exception& e) {
240 : 3 : return std::string("Failed to load TLS certificate or key: ") + e.what();
241 : 1 : }
242 : : #endif
243 : : }
244 : :
245 : 348 : std::shared_ptr<TcpServerSession> make_session(tcp::socket sock) {
246 : : #ifdef WIRESTEAD_TLS_ENABLED
247 [ + + ]: 348 : if (ssl_context_) {
248 : : return std::make_shared<TcpServerSession>(
249 : 68 : ioc_, std::make_unique<SslTcpSocket>(std::move(sock), ssl_context_), cfg_.backpressure_threshold,
250 : 68 : cfg_.idle_timeout_ms, cfg_.backpressure_strategy, cfg_.enable_memory_pool, cfg_.read_buffer_size);
251 : : }
252 : : #endif
253 : 314 : return std::make_shared<TcpServerSession>(ioc_, std::move(sock), cfg_.backpressure_threshold, cfg_.idle_timeout_ms,
254 : 314 : cfg_.backpressure_strategy, cfg_.enable_memory_pool,
255 : 314 : cfg_.read_buffer_size);
256 : : }
257 : :
258 : 348 : void apply_accepted_socket_options(tcp::socket& sock) {
259 : 348 : boost::system::error_code ec;
260 : :
261 [ + - ]: 348 : if (cfg_.tcp_no_delay) {
262 : 348 : sock.set_option(tcp::no_delay(true), ec);
263 [ - + ]: 348 : if (ec) {
264 : 0 : WIRESTEAD_LOG_WARNING("tcp_server", "socket_options",
265 : : fmt::format("Failed to set TCP_NODELAY: {}", ec.message()));
266 : 0 : ec.clear();
267 : : }
268 : : }
269 : :
270 [ - + ]: 348 : if (cfg_.keep_alive) {
271 : 0 : sock.set_option(net::socket_base::keep_alive(true), ec);
272 [ # # ]: 0 : if (ec) {
273 : 0 : WIRESTEAD_LOG_WARNING("tcp_server", "socket_options",
274 : : fmt::format("Failed to set keep_alive: {}", ec.message()));
275 : 0 : ec.clear();
276 : : }
277 : : }
278 : :
279 [ + + ]: 348 : if (cfg_.send_buffer_size > 0) {
280 : 4 : sock.set_option(net::socket_base::send_buffer_size(static_cast<int>(cfg_.send_buffer_size)), ec);
281 [ - + ]: 4 : if (ec) {
282 : 0 : WIRESTEAD_LOG_WARNING("tcp_server", "socket_options",
283 : : fmt::format("Failed to set send buffer size: {}", ec.message()));
284 : 0 : ec.clear();
285 : : }
286 : : }
287 : :
288 [ - + ]: 348 : if (cfg_.receive_buffer_size > 0) {
289 : 0 : sock.set_option(net::socket_base::receive_buffer_size(static_cast<int>(cfg_.receive_buffer_size)), ec);
290 [ # # ]: 0 : if (ec) {
291 : 0 : WIRESTEAD_LOG_WARNING("tcp_server", "socket_options",
292 : : fmt::format("Failed to set receive buffer size: {}", ec.message()));
293 : 0 : ec.clear();
294 : : }
295 : : }
296 : 348 : }
297 : :
298 : 304 : void attempt_port_binding(std::shared_ptr<TcpServer> self, int retry_count) {
299 [ - + ]: 308 : if (stopping_.load()) return;
300 : 304 : boost::system::error_code ec;
301 : :
302 : 304 : auto address = net::ip::make_address(cfg_.bind_address, ec);
303 [ - + ]: 304 : if (ec) {
304 : 0 : std::string msg = fmt::format("Invalid bind address: {}, {}", cfg_.bind_address, ec.message());
305 : 0 : WIRESTEAD_LOG_ERROR("tcp_server", "bind", msg);
306 : 0 : error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONFIGURATION, "bind",
307 : : ec, msg, false, static_cast<uint32_t>(retry_count));
308 : 0 : state_.set(base::LinkState::Error);
309 : 0 : notify_state();
310 : 0 : return;
311 : 0 : }
312 : :
313 [ + - + + ]: 304 : if (!acceptor_->is_open()) {
314 [ - + + - ]: 303 : acceptor_->open(address.is_v6() ? tcp::v6() : tcp::v4(), ec);
315 [ + + ]: 303 : if (ec) {
316 : 1 : std::string msg = fmt::format("Failed to open acceptor: {}", ec.message());
317 : 1 : WIRESTEAD_LOG_ERROR("tcp_server", "open", msg);
318 : 1 : error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::SYSTEM, "open", ec,
319 : : msg, false, static_cast<uint32_t>(retry_count));
320 : 1 : state_.set(base::LinkState::Error);
321 : 1 : notify_state();
322 : 1 : return;
323 : 1 : }
324 : : }
325 : :
326 : 303 : acceptor_->bind(tcp::endpoint(address, cfg_.port), ec);
327 [ + + ]: 303 : if (ec) {
328 [ + + + - ]: 2 : if (cfg_.enable_port_retry && retry_count < cfg_.max_port_retries) {
329 : 1 : auto timer = std::make_shared<net::steady_timer>(strand_);
330 : 1 : timer->expires_after(std::chrono::milliseconds(cfg_.port_retry_interval_ms));
331 : 1 : timer->async_wait(
332 : 2 : [self, retry_count, timer, generation = generation_.load()](const boost::system::error_code& timer_ec) {
333 [ + - ]: 1 : if (!timer_ec) {
334 : 1 : auto* timer_impl = self->get_impl();
335 [ + - + - : 2 : if (!timer_impl->stopping_.load() && timer_impl->generation_.load() == generation) {
+ - ]
336 : 1 : timer_impl->attempt_port_binding(self, retry_count + 1);
337 : : }
338 : : }
339 : 1 : });
340 : 1 : return;
341 : 1 : } else {
342 : 1 : std::string msg = fmt::format("Failed to bind to port {}: {}", cfg_.port, ec.message());
343 : 1 : WIRESTEAD_LOG_ERROR("tcp_server", "bind", msg);
344 : 1 : error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "bind",
345 : : ec, msg, false, static_cast<uint32_t>(retry_count));
346 : 1 : state_.set(base::LinkState::Error);
347 : 1 : notify_state();
348 : 1 : return;
349 : 1 : }
350 : : }
351 : :
352 : 301 : acceptor_->listen(boost::asio::socket_base::max_listen_connections, ec);
353 [ + + ]: 301 : if (ec) {
354 : 1 : std::string msg = fmt::format("Failed to listen on port {}: {}", cfg_.port, ec.message());
355 : 1 : WIRESTEAD_LOG_ERROR("tcp_server", "listen", msg);
356 : 1 : error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "listen",
357 : : ec, msg, false, static_cast<uint32_t>(retry_count));
358 : 1 : state_.set(base::LinkState::Error);
359 : 1 : notify_state();
360 : 1 : return;
361 : 1 : }
362 : :
363 : 300 : state_.set(base::LinkState::Listening);
364 : 300 : notify_state();
365 : 300 : do_accept(self);
366 : : }
367 : :
368 : 2422 : void do_accept(std::shared_ptr<TcpServer> self) {
369 [ + + + - : 2422 : if (stopping_.load() || !acceptor_ || !acceptor_->is_open()) return;
- + + + ]
370 : :
371 : 4806 : acceptor_->async_accept([self, generation = generation_.load()](auto ec, tcp::socket sock) {
372 : 2403 : net::dispatch(self->get_impl()->strand_, [self, generation, ec, sock = std::move(sock)]() mutable {
373 : 2403 : auto* accept_impl = self->get_impl();
374 [ + + - + : 4526 : if (accept_impl->stopping_.load() || accept_impl->generation_.load() != generation) {
+ + ]
375 : 2055 : return;
376 : : }
377 [ + + ]: 2123 : if (ec) {
378 [ + - ]: 1 : if (ec != boost::asio::error::operation_aborted) {
379 : 2 : accept_impl->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR,
380 : 1 : diagnostics::ErrorCategory::CONNECTION, "accept", ec,
381 : : fmt::format("Accept failed: {}", ec.message()), true, 0);
382 : : // Keep Listening while the accept operation is retried.
383 : : }
384 [ + - + - : 1 : if (!accept_impl->state_.is_state(base::LinkState::Closed) && !accept_impl->stopping_.load()) {
+ - ]
385 : 1 : auto timer = std::make_shared<net::steady_timer>(accept_impl->strand_);
386 : 1 : timer->expires_after(std::chrono::milliseconds(100));
387 : 1 : timer->async_wait([self, timer, generation](const boost::system::error_code&) {
388 : 0 : auto* retry_impl = self->get_impl();
389 [ # # # # : 0 : if (!retry_impl->stopping_.load() && retry_impl->generation_.load() == generation) {
# # ]
390 : 0 : retry_impl->do_accept(self);
391 : : }
392 : : });
393 : 1 : }
394 : 1 : return;
395 : : }
396 : :
397 : 2122 : boost::system::error_code ep_ec;
398 : 2122 : auto rep = sock.remote_endpoint(ep_ec);
399 : 2122 : std::string client_info = "unknown";
400 [ + - ]: 2122 : if (!ep_ec) {
401 : 2122 : client_info = fmt::format("{}:{}", rep.address().to_string(), rep.port());
402 : : }
403 : :
404 [ + - ]: 2122 : if (accept_impl->client_limit_enabled_) {
405 : : bool over_limit;
406 : : {
407 : 2122 : std::lock_guard<std::mutex> lock(accept_impl->sessions_mutex_);
408 : 2122 : over_limit = accept_impl->sessions_.size() >= accept_impl->max_clients_;
409 : 2122 : }
410 [ + + ]: 2122 : if (over_limit) {
411 : : // #437: accept and immediately close over-limit connections
412 : : // instead of pausing the accept loop entirely - a client whose
413 : : // TCP handshake already completed while paused would otherwise
414 : : // sit connected but silent until a slot frees up.
415 : 1774 : boost::system::error_code close_ec;
416 : 1774 : sock.close(close_ec);
417 : 1774 : accept_impl->do_accept(self);
418 : 1774 : return;
419 : : }
420 : : }
421 : :
422 : 348 : accept_impl->apply_accepted_socket_options(sock);
423 : :
424 : : // The session only talks to TcpSocketInterface, so TLS is a matter of
425 : : // which implementation it gets wrapped in here. Everything downstream -
426 : : // reads, writes, backpressure, stats - is identical either way.
427 : 348 : auto new_session = accept_impl->make_session(std::move(sock));
428 : :
429 : 348 : ClientId client_id = accept_impl->next_client_id_.fetch_add(1);
430 : :
431 : 348 : std::weak_ptr<TcpServer> weak_self = self;
432 : :
433 : 348 : new_session->on_bytes([weak_self, client_id](memory::ConstByteSpan data) {
434 : 2666 : auto shared_self = weak_self.lock();
435 [ - + ]: 2666 : if (!shared_self) return;
436 : 2666 : auto* bytes_impl = shared_self->get_impl();
437 : :
438 : 2666 : interface::SharedCallback<OnBytes> cb;
439 : 2666 : interface::SharedCallback<MultiClientDataHandler> multi_cb;
440 : : {
441 : 2666 : std::lock_guard<std::mutex> lock(bytes_impl->sessions_mutex_);
442 : 2666 : cb = bytes_impl->on_bytes_;
443 : 2666 : multi_cb = bytes_impl->on_multi_data_;
444 : 2666 : }
445 : 2666 : diagnostics::invoke_callback("tcp_server", "on_bytes", cb, data);
446 [ + + ]: 2666 : if (multi_cb) {
447 : 2655 : diagnostics::invoke_callback("tcp_server", "on_multi_data", multi_cb, client_id, data);
448 : : }
449 : 2666 : });
450 : :
451 : 348 : interface::SharedCallback<OnBackpressure> bp_cb;
452 : : {
453 : 348 : std::lock_guard<std::mutex> lock(accept_impl->sessions_mutex_);
454 : 348 : bp_cb = accept_impl->on_bp_;
455 : 348 : }
456 [ + + + - : 348 : if (bp_cb) new_session->on_backpressure(*bp_cb);
+ - ]
457 : :
458 : 348 : new_session->on_close([weak_self, client_id, new_session, generation] {
459 : 229 : auto shared_self = weak_self.lock();
460 [ - + ]: 229 : if (!shared_self) return;
461 : 229 : auto* notify_impl = shared_self->get_impl();
462 [ + + - + : 229 : if (notify_impl->stopping_ || notify_impl->generation_ != generation) return;
+ + ]
463 : 213 : MultiClientDisconnectHandler disconnect_cb;
464 : : {
465 : 213 : std::lock_guard<std::mutex> lock(notify_impl->sessions_mutex_);
466 : 213 : disconnect_cb = notify_impl->on_multi_disconnect_;
467 : 213 : }
468 : 213 : diagnostics::invoke_callback("tcp_server", "on_disconnect", disconnect_cb, client_id);
469 : 213 : net::dispatch(shared_self->get_impl()->strand_, [shared_self, client_id, new_session, generation] {
470 : 213 : auto* close_impl = shared_self->get_impl();
471 [ + + - + : 422 : if (close_impl->stopping_.load() || close_impl->generation_.load() != generation) return;
+ + ]
472 : :
473 : 209 : bool was_current = false;
474 : : {
475 : 209 : std::lock_guard<std::mutex> lock(close_impl->sessions_mutex_);
476 : : // Carry the session's totals over to the server before it goes away,
477 : : // so stats() keeps reporting what this connection did. Tied to the
478 : : // erase below, which makes it exactly once even if on_close re-fires.
479 : 209 : auto it = close_impl->sessions_.find(client_id);
480 [ + - + - : 209 : if (it != close_impl->sessions_.end() && it->second) {
+ - ]
481 : 209 : close_impl->absorb_session(it->second);
482 : : }
483 : 209 : close_impl->sessions_.erase(client_id);
484 : 209 : was_current = (close_impl->current_session_ == new_session);
485 [ + + ]: 209 : if (was_current) {
486 [ + + ]: 173 : if (!close_impl->sessions_.empty())
487 : 5 : close_impl->current_session_ = close_impl->sessions_.begin()->second;
488 : : else
489 : 168 : close_impl->current_session_.reset();
490 : : }
491 : 209 : }
492 [ + + ]: 209 : if (was_current) {
493 : 173 : close_impl->state_.set(base::LinkState::Listening);
494 : 173 : close_impl->notify_state();
495 : : }
496 : : });
497 : 229 : });
498 : :
499 : : {
500 : 348 : std::lock_guard<std::mutex> lock(accept_impl->sessions_mutex_);
501 : 348 : accept_impl->sessions_.emplace(client_id, new_session);
502 : 348 : accept_impl->current_session_ = new_session;
503 : 348 : auto connect_cb = accept_impl->on_multi_connect_;
504 : : // Publish alive admission and queue notification before releasing the
505 : : // map lock: any sender observing this session queues behind connect.
506 : 696 : new_session->start_with_notification(
507 : 696 : [self, generation, client_id, client_info, connect_cb = std::move(connect_cb)] {
508 : 347 : auto* impl = self->get_impl();
509 [ + + - + : 347 : if (impl->stopping_ || impl->generation_ != generation) return;
+ + ]
510 : 344 : diagnostics::invoke_callback("tcp_server", "on_connect", connect_cb, client_id, client_info);
511 : 344 : net::post(impl->strand_, [self, generation, client_id] {
512 : 344 : auto* state_impl = self->get_impl();
513 [ + + - + : 344 : if (state_impl->stopping_ || state_impl->generation_ != generation) return;
+ + ]
514 : : {
515 : 329 : std::lock_guard<std::mutex> lock(state_impl->sessions_mutex_);
516 [ + - + + ]: 329 : if (state_impl->sessions_.find(client_id) == state_impl->sessions_.end()) return;
517 : 329 : }
518 : 328 : state_impl->state_.set(base::LinkState::Connected);
519 : 328 : state_impl->notify_state();
520 : : });
521 : : });
522 : 348 : }
523 : :
524 : 348 : accept_impl->do_accept(self);
525 : 2122 : });
526 : 2403 : });
527 : : }
528 : :
529 : 307 : void finish_cleanup() {
530 : 307 : state_.set(base::LinkState::Closed);
531 [ + + ]: 307 : if (work_guard_) work_guard_->reset();
532 : 307 : mark_cleanup_done();
533 : 307 : }
534 : :
535 : 603 : void perform_cleanup(std::shared_ptr<TcpServer> self = {}) {
536 [ + + ]: 799 : if (cleanup_started_.exchange(true)) return;
537 : 307 : boost::system::error_code ec;
538 [ + - + - ]: 307 : if (acceptor_) acceptor_->close(ec);
539 : 307 : std::vector<std::pair<ClientId, std::shared_ptr<TcpServerSession>>> sessions;
540 : : {
541 : 307 : std::lock_guard<std::mutex> lock(sessions_mutex_);
542 : 307 : retiring_sessions_.swap(sessions_);
543 [ + - + + ]: 446 : for (const auto& entry : retiring_sessions_) sessions.push_back(entry);
544 : 307 : current_session_.reset();
545 : 307 : }
546 [ + + + - ]: 307 : if (auto hook = detail::g_server_sessions_retiring_hook.load()) hook();
547 [ + + ]: 307 : if (sessions.empty()) {
548 : 196 : finish_cleanup();
549 : 196 : return;
550 : : }
551 : : // Completion includes every session's callback body and cleanup. Each
552 : : // session owns its outstanding I/O; final state changes are serialized
553 : : // with accept/retry handlers on the server's management strand.
554 : 111 : auto remaining = std::make_shared<size_t>(sessions.size());
555 [ + + ]: 250 : for (auto& [client_id, session] : sessions) {
556 : 139 : session->async_stop([this, self, remaining, client_id] {
557 : 139 : net::post(strand_, [this, self, remaining, client_id] {
558 : : {
559 : 139 : std::lock_guard<std::mutex> lock(sessions_mutex_);
560 : 139 : const auto it = retiring_sessions_.find(client_id);
561 [ + - ]: 139 : if (it != retiring_sessions_.end()) {
562 : 139 : absorb_session(it->second);
563 : 139 : retiring_sessions_.erase(it);
564 : : }
565 : 139 : }
566 [ + + ]: 139 : if (--*remaining == 0) finish_cleanup();
567 : 139 : });
568 : 139 : });
569 : : }
570 : 307 : }
571 : :
572 : 879 : void stop(std::shared_ptr<TcpServer> self) {
573 : 879 : const bool on_executor = ioc_.get_executor().running_in_this_thread();
574 : : bool first;
575 : : {
576 : 879 : std::lock_guard<std::mutex> lock(target_admission_mtx_);
577 : 879 : first = !stopping_.exchange(true);
578 [ + + ]: 879 : if (first) {
579 : 307 : std::lock_guard<std::mutex> sessions_lock(sessions_mutex_);
580 [ + + ]: 443 : for (auto& entry : sessions_)
581 [ + - + - ]: 136 : if (entry.second) entry.second->request_stop();
582 : 307 : }
583 : 879 : }
584 [ + + ]: 879 : if (first) {
585 : : {
586 : 307 : std::lock_guard<std::mutex> lock(sessions_mutex_);
587 : 307 : on_bytes_ = nullptr;
588 : 307 : on_state_ = nullptr;
589 : 307 : on_bp_ = nullptr;
590 : 307 : on_multi_connect_ = nullptr;
591 : 307 : on_multi_data_ = nullptr;
592 : 307 : on_multi_disconnect_ = nullptr;
593 : 307 : }
594 [ + + ]: 307 : if (!run_dispatched_) {
595 : 3 : perform_cleanup(self);
596 : : } else {
597 : 608 : net::post(strand_, [this, self] { perform_cleanup(self); });
598 : : }
599 : : }
600 [ + + ]: 879 : if (on_executor) return;
601 : 845 : wait_for_cleanup();
602 : 845 : std::lock_guard<std::mutex> join_lock(join_mtx_);
603 : 845 : join_owned_thread();
604 : 845 : }
605 : :
606 : 845 : void join_owned_thread() {
607 [ + + ]: 845 : if (!owns_ioc_) return;
608 [ + + ]: 382 : if (work_guard_) work_guard_->reset();
609 [ + + ]: 382 : if (!ioc_thread_.joinable()) return;
610 [ - + ]: 184 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
611 : 0 : ioc_thread_.detach();
612 : : } else {
613 : 184 : ioc_thread_.join();
614 : : }
615 : : }
616 : : };
617 : :
618 : 191 : std::shared_ptr<TcpServer> TcpServer::create(const config::TcpServerConfig& cfg, bool use_shared_context) {
619 : 191 : return std::shared_ptr<TcpServer>(new TcpServer(cfg, use_shared_context));
620 : : }
621 : :
622 : 111 : std::shared_ptr<TcpServer> TcpServer::create(const config::TcpServerConfig& cfg,
623 : : std::unique_ptr<interface::TcpAcceptorInterface> acceptor,
624 : : net::io_context& ioc) {
625 : 114 : return std::shared_ptr<TcpServer>(new TcpServer(cfg, std::move(acceptor), ioc));
626 : : }
627 : :
628 : 191 : TcpServer::TcpServer(const config::TcpServerConfig& cfg, bool use_shared_context)
629 : 194 : : impl_(std::make_unique<Impl>(cfg, use_shared_context)) {}
630 : :
631 : 111 : TcpServer::TcpServer(const config::TcpServerConfig& cfg, std::unique_ptr<interface::TcpAcceptorInterface> acceptor,
632 : 111 : net::io_context& ioc)
633 : 114 : : impl_(std::make_unique<Impl>(cfg, std::move(acceptor), ioc)) {}
634 : :
635 : 592 : TcpServer::~TcpServer() {
636 [ + - ]: 296 : if (impl_) {
637 : : // Pass nullptr to stop() to indicate we are in destructor and cannot use shared_from_this
638 : 296 : impl_->stop(nullptr);
639 : : }
640 : 592 : }
641 : :
642 : 0 : TcpServer::TcpServer(TcpServer&&) noexcept = default;
643 : 0 : TcpServer& TcpServer::operator=(TcpServer&&) noexcept = default;
644 : :
645 : 305 : void TcpServer::start() {
646 : 305 : auto impl = get_impl();
647 : 305 : auto current = impl->state_.get();
648 [ + - + - : 305 : if (current == base::LinkState::Listening || current == base::LinkState::Connected ||
- + ]
649 : : current == base::LinkState::Connecting) {
650 : 1 : return;
651 : : }
652 : 305 : const auto generation = impl->generation_.fetch_add(1) + 1;
653 : 305 : impl->run_dispatched_ = false;
654 : 305 : impl->stopping_.store(false);
655 : 305 : impl->cleanup_started_.store(false);
656 : : {
657 : 305 : std::lock_guard<std::mutex> lock(impl->stop_mtx_);
658 : 305 : impl->cleanup_done_ = false;
659 : 305 : }
660 : : // Restart contract (#444): stats() resets on restart. The server-level
661 : : // counters now outlive the sessions that fed them, so clearing them here is
662 : : // what keeps that promise - before absorption they were empty and a restart
663 : : // zeroed the aggregate for free.
664 : : {
665 : 305 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
666 : 305 : impl->stats_.reset(0);
667 : 305 : impl->closed_send_accounting_ = {};
668 : 305 : }
669 : :
670 : : // Load the certificate before binding. A server asked for TLS that cannot
671 : : // provide it must not come up in plaintext instead - that is the failure mode
672 : : // where everything looks healthy and nothing is encrypted.
673 [ + - + + ]: 305 : if (const auto tls_error = impl->init_tls(); !tls_error.empty()) {
674 : 1 : WIRESTEAD_LOG_ERROR("tcp_server", "start", tls_error);
675 : 1 : impl->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONFIGURATION,
676 : : "start", {}, tls_error, false, 0);
677 : 1 : impl->state_.set(base::LinkState::Error);
678 : 1 : impl->notify_state();
679 : 1 : return;
680 : 305 : }
681 : :
682 [ + + ]: 304 : if (impl->uses_shared_context_) {
683 : 2 : auto& manager = concurrency::IoContextManager::instance();
684 [ + - - + ]: 2 : if (!manager.is_running()) {
685 : 0 : manager.start();
686 : : }
687 : : }
688 : :
689 [ - + ]: 304 : if (!impl->acceptor_) {
690 : 0 : impl->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::SYSTEM, "start",
691 : : {}, "No acceptor available", false, 0);
692 : 0 : impl->state_.set(base::LinkState::Error);
693 : 0 : impl->notify_state();
694 : 0 : return;
695 : : }
696 : :
697 [ + + + - : 304 : if (impl->owns_ioc_ && !impl->ioc_thread_.joinable()) {
+ + ]
698 [ + - - + ]: 184 : if (impl->ioc_.stopped()) {
699 : 0 : impl->ioc_.restart();
700 : : }
701 : : impl->work_guard_ =
702 : 184 : std::make_unique<net::executor_work_guard<net::io_context::executor_type>>(impl->ioc_.get_executor());
703 : 368 : impl->ioc_thread_ = std::jthread([impl](std::stop_token st) {
704 : 184 : wirestead::concurrency::run_io_thread_init();
705 : : try {
706 : 184 : std::stop_callback cb(st, [impl] { impl->ioc_.stop(); });
707 : 184 : impl->ioc_.run();
708 : 184 : } catch (...) {
709 : 0 : }
710 : 368 : });
711 : : }
712 : 304 : impl->run_dispatched_ = true;
713 : 304 : auto self = shared_from_this();
714 : 304 : net::dispatch(impl->strand_, [self, generation] {
715 : 304 : auto* impl = self->get_impl();
716 [ + + - + : 607 : if (impl->stopping_.load() || impl->generation_.load() != generation) return;
+ + ]
717 : 303 : impl->attempt_port_binding(self, 0);
718 : : });
719 : 304 : }
720 : :
721 : 583 : void TcpServer::stop() { impl_->stop(shared_from_this()); }
722 : :
723 : 0 : void TcpServer::request_stop() {
724 : 0 : auto impl = get_impl();
725 [ # # ]: 0 : if (impl->stopping_.load()) return;
726 : 0 : auto self = shared_from_this();
727 : 0 : net::post(impl->strand_, [self, generation = impl->generation_.load()] {
728 [ # # ]: 0 : if (self->get_impl()->generation_.load() == generation) self->stop();
729 : 0 : });
730 : 0 : }
731 : :
732 : 1 : bool TcpServer::is_connected() const {
733 : 1 : auto impl = get_impl();
734 : 1 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
735 [ + - + - : 2 : return impl->current_session_ && impl->current_session_->alive();
+ - ]
736 : 1 : }
737 : :
738 : 1 : bool TcpServer::is_backpressure_active() const { return false; }
739 : :
740 : 209 : bool TcpServer::is_backpressure_active(ClientId client_id) const {
741 : 209 : auto impl = get_impl();
742 : 209 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
743 : 209 : auto it = impl->sessions_.find(client_id);
744 [ + - + - : 209 : if (it != impl->sessions_.end() && it->second) {
+ - ]
745 : 209 : return it->second->is_backpressure_active();
746 : : }
747 : 0 : return false;
748 : 209 : }
749 : :
750 : 353 : std::optional<size_t> TcpServer::write_queue_limit(ClientId client_id) const {
751 : 353 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
752 : 353 : auto it = impl_->sessions_.find(client_id);
753 [ + + - + : 353 : if (it == impl_->sessions_.end() || !it->second) return std::nullopt;
+ + ]
754 : 298 : return it->second->write_queue_limit();
755 : 353 : }
756 : :
757 : 624 : boost::asio::any_io_executor TcpServer::get_executor() { return impl_->strand_; }
758 : :
759 : 2040 : wrapper::RuntimeStats TcpServer::stats() const {
760 : 2040 : auto impl = get_impl();
761 : : // Snapshot retained totals and live sessions under the same lock as the
762 : : // disconnect transfer; otherwise a retiring session can disappear from both.
763 : 2040 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
764 : 2040 : auto aggregate = impl->stats_.snapshot(0, 0, false);
765 : 2040 : aggregate.send_accounting = impl->closed_send_accounting_;
766 [ + + ]: 6120 : for (const auto* sessions : {&impl->sessions_, &impl->retiring_sessions_}) {
767 [ + + ]: 4290 : for (const auto& entry : *sessions) {
768 [ - + ]: 210 : if (!entry.second) continue;
769 : 210 : const auto session_stats = entry.second->stats();
770 : 210 : diagnostics::accumulate_send_accounting(*aggregate.send_accounting, *session_stats.send_accounting);
771 : 210 : aggregate.bytes_accepted += session_stats.bytes_accepted;
772 : 210 : aggregate.messages_accepted += session_stats.messages_accepted;
773 : 210 : aggregate.bytes_sent += session_stats.bytes_sent;
774 : 210 : aggregate.messages_sent += session_stats.messages_sent;
775 : 210 : aggregate.bytes_received += session_stats.bytes_received;
776 : 210 : aggregate.messages_received += session_stats.messages_received;
777 : 210 : aggregate.failed_sends += session_stats.failed_sends;
778 : 210 : aggregate.dropped_messages += session_stats.dropped_messages;
779 : 210 : aggregate.dropped_bytes += session_stats.dropped_bytes;
780 : 210 : aggregate.backpressure_events += session_stats.backpressure_events;
781 : : // Retiring sessions retain cumulative totals, not live queue gauges.
782 [ + + ]: 210 : if (sessions == &impl->sessions_) {
783 : 198 : aggregate.queued_bytes += session_stats.queued_bytes;
784 : 198 : aggregate.pending_bytes += session_stats.pending_bytes;
785 [ + - - + ]: 198 : aggregate.backpressure_active = aggregate.backpressure_active || session_stats.backpressure_active;
786 : : }
787 : : // Peak, not a total: summing per-session peaks would report a depth no
788 : : // session ever reached, because the peaks need not have been simultaneous.
789 : 210 : aggregate.max_queued_bytes = std::max(aggregate.max_queued_bytes, session_stats.max_queued_bytes);
790 : : }
791 : : }
792 : 4080 : return aggregate;
793 : 2040 : }
794 : :
795 : 3 : void TcpServer::reset_stats() {
796 : 3 : auto impl = get_impl();
797 : 3 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
798 : 3 : impl->stats_.reset(0);
799 : 3 : impl->closed_send_accounting_ = {};
800 [ + + ]: 9 : for (const auto* sessions : {&impl->sessions_, &impl->retiring_sessions_}) {
801 [ + + ]: 11 : for (const auto& entry : *sessions) {
802 [ + - + - ]: 5 : if (entry.second) entry.second->reset_stats();
803 : : }
804 : : }
805 : 3 : }
806 : :
807 : 5 : std::optional<diagnostics::ErrorInfo> TcpServer::last_error_info() const {
808 : 5 : return get_impl()->error_info_holder_.last_error_info();
809 : : }
810 : :
811 : 2 : bool TcpServer::async_write_copy(memory::ConstByteSpan data) {
812 : 2 : auto impl = get_impl();
813 [ - + ]: 2 : if (impl->stopping_.load()) {
814 : 0 : impl->stats_.record_failed_send();
815 : 0 : return false;
816 : : }
817 : 2 : std::shared_ptr<TcpServerSession> session;
818 : : {
819 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
820 : 2 : session = impl->current_session_;
821 : 2 : }
822 : :
823 [ + + + - : 2 : if (session && session->alive()) {
+ - + + ]
824 : 1 : return session->async_write_copy(data);
825 : : }
826 : 1 : impl->stats_.record_failed_send();
827 : 1 : return false;
828 : 2 : }
829 : :
830 : 2 : bool TcpServer::async_write_move(std::vector<uint8_t>&& data) {
831 : 2 : auto impl = get_impl();
832 [ - + ]: 2 : if (impl->stopping_.load()) {
833 : 0 : impl->stats_.record_failed_send();
834 : 0 : return false;
835 : : }
836 : 2 : std::shared_ptr<TcpServerSession> session;
837 : : {
838 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
839 : 2 : session = impl->current_session_;
840 : 2 : }
841 : :
842 [ + + + - : 2 : if (session && session->alive()) {
+ - + + ]
843 : 1 : return session->async_write_move(std::move(data));
844 : : }
845 : 1 : impl->stats_.record_failed_send();
846 : 1 : return false;
847 : 2 : }
848 : :
849 : 2 : bool TcpServer::async_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
850 : 2 : auto impl = get_impl();
851 [ + - + - : 2 : if (impl->stopping_.load() || !data || data->empty()) {
- + - + ]
852 : 0 : impl->stats_.record_failed_send();
853 : 0 : return false;
854 : : }
855 : 2 : std::shared_ptr<TcpServerSession> session;
856 : : {
857 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
858 : 2 : session = impl->current_session_;
859 : 2 : }
860 : :
861 [ + + + - : 2 : if (session && session->alive()) {
+ - + + ]
862 : 1 : return session->async_write_shared(std::move(data));
863 : : }
864 : 1 : impl->stats_.record_failed_send();
865 : 1 : return false;
866 : 2 : }
867 : :
868 : 2 : bool TcpServer::async_try_write_copy(memory::ConstByteSpan data) {
869 : 2 : auto impl = get_impl();
870 [ - + ]: 2 : if (impl->stopping_.load()) {
871 : 0 : impl->stats_.record_failed_send();
872 : 0 : return false;
873 : : }
874 : 2 : std::shared_ptr<TcpServerSession> session;
875 : : {
876 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
877 : 2 : session = impl->current_session_;
878 : 2 : }
879 : :
880 [ + + + - : 2 : if (session && session->alive()) {
+ - + + ]
881 : 1 : return session->async_try_write_copy(data);
882 : : }
883 : 1 : impl->stats_.record_failed_send();
884 : 1 : return false;
885 : 2 : }
886 : :
887 : 2 : bool TcpServer::async_try_write_move(std::vector<uint8_t>&& data) {
888 : 2 : auto impl = get_impl();
889 [ - + ]: 2 : if (impl->stopping_.load()) {
890 : 0 : impl->stats_.record_failed_send();
891 : 0 : return false;
892 : : }
893 : 2 : std::shared_ptr<TcpServerSession> session;
894 : : {
895 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
896 : 2 : session = impl->current_session_;
897 : 2 : }
898 : :
899 [ + + + - : 2 : if (session && session->alive()) {
+ - + + ]
900 : 1 : return session->async_try_write_move(std::move(data));
901 : : }
902 : 1 : impl->stats_.record_failed_send();
903 : 1 : return false;
904 : 2 : }
905 : :
906 : 2 : bool TcpServer::async_try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
907 : 2 : auto impl = get_impl();
908 [ + - + - : 2 : if (impl->stopping_.load() || !data || data->empty()) {
- + - + ]
909 : 0 : impl->stats_.record_failed_send();
910 : 0 : return false;
911 : : }
912 : 2 : std::shared_ptr<TcpServerSession> session;
913 : : {
914 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
915 : 2 : session = impl->current_session_;
916 : 2 : }
917 : :
918 [ + + + - : 2 : if (session && session->alive()) {
+ - + + ]
919 : 1 : return session->async_try_write_shared(std::move(data));
920 : : }
921 : 1 : impl->stats_.record_failed_send();
922 : 1 : return false;
923 : 2 : }
924 : :
925 : : // Each setter builds the shared snapshot before taking the lock, so the
926 : : // allocation stays outside the critical section the io thread contends on.
927 : 496 : void TcpServer::on_bytes(OnBytes cb) {
928 : 496 : auto shared = interface::share_callback(std::move(cb));
929 : 496 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
930 : 496 : impl_->on_bytes_ = std::move(shared);
931 : 496 : }
932 : 876 : void TcpServer::on_state(OnState cb) {
933 : 876 : auto shared = interface::share_callback(std::move(cb));
934 : 876 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
935 : 876 : impl_->on_state_ = std::move(shared);
936 : 876 : }
937 : 868 : void TcpServer::on_backpressure(OnBackpressure cb) {
938 : 868 : auto impl = get_impl();
939 : 868 : auto shared = interface::share_callback(std::move(cb));
940 : 868 : std::shared_ptr<TcpServerSession> session;
941 : 868 : interface::SharedCallback<OnBackpressure> bp_cb;
942 : : {
943 : 868 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
944 : 868 : impl->on_bp_ = std::move(shared);
945 : 868 : bp_cb = impl->on_bp_;
946 : 868 : session = impl->current_session_;
947 : 868 : }
948 : :
949 [ - + - - : 868 : if (session) session->on_backpressure(bp_cb ? *bp_cb : OnBackpressure{});
- - - - ]
950 : 868 : }
951 : :
952 : 2 : bool TcpServer::broadcast(std::string_view message) {
953 : 2 : auto shared_data = std::make_shared<const std::vector<uint8_t>>(message.begin(), message.end());
954 : 2 : auto impl = get_impl();
955 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
956 : 2 : bool sent = false;
957 : 2 : bool attempted = false;
958 [ + + ]: 3 : for (auto& entry : impl->sessions_) {
959 : 1 : auto& session = entry.second;
960 [ + - + - : 1 : if (session && session->alive()) {
+ - + - ]
961 : 1 : attempted = true;
962 [ + - + - ]: 1 : if (session->async_try_write_shared(shared_data)) sent = true;
963 : : }
964 : : }
965 [ + + ]: 2 : if (!attempted) impl->stats_.record_failed_send();
966 : 2 : return sent;
967 : 2 : }
968 : :
969 : 2 : bool TcpServer::broadcast(memory::ConstByteSpan data) {
970 : 2 : auto shared_data = std::make_shared<const std::vector<uint8_t>>(data.begin(), data.end());
971 : 2 : auto impl = get_impl();
972 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
973 : 2 : bool sent = false;
974 : 2 : bool attempted = false;
975 [ + + ]: 3 : for (auto& entry : impl->sessions_) {
976 : 1 : auto& session = entry.second;
977 [ + - + - : 1 : if (session && session->alive()) {
+ - + - ]
978 : 1 : attempted = true;
979 [ + - + - ]: 1 : if (session->async_try_write_shared(shared_data)) sent = true;
980 : : }
981 : : }
982 [ + + ]: 2 : if (!attempted) impl->stats_.record_failed_send();
983 : 2 : return sent;
984 : 2 : }
985 : :
986 : 1248 : wrapper::FanoutResult TcpServer::broadcast_result(memory::ConstByteSpan data, wrapper::SendResult wrapper_state,
987 : : bool append_newline) {
988 : 1248 : std::vector<std::pair<ClientId, std::shared_ptr<TcpServerSession>>> targets;
989 : : {
990 : 1248 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
991 : 1248 : targets.reserve(impl_->sessions_.size());
992 [ + + ]: 2693 : for (const auto& [id, session] : impl_->sessions_) {
993 [ + - + - : 1445 : if (session && session->alive()) targets.emplace_back(id, session);
+ - + - +
- ]
994 : : }
995 : 1248 : }
996 [ + + + - ]: 1248 : if (auto hook = detail::g_tcp_fanout_snapshot_hook.load()) hook();
997 : 1248 : wrapper::FanoutResult result;
998 : 1248 : std::shared_ptr<const std::vector<uint8_t>> shared_data;
999 [ + + ]: 2693 : for (const auto& [id, session] : targets) {
1000 : : const auto size =
1001 : : append_newline
1002 [ + + + - ]: 1445 : ? (data.size() >= base::constants::MAX_BUFFER_SIZE ? base::constants::MAX_BUFFER_SIZE + 1 : data.size() + 1)
1003 : 1419 : : data.size();
1004 : 1445 : const auto validation = wrapper::detail::validate_payload_size(size, session->write_queue_limit());
1005 [ + + ]: 1445 : if (!validation.accepted()) {
1006 : 80 : result.add(validation);
1007 : 80 : continue;
1008 : : }
1009 [ - + ]: 1365 : if (!wrapper_state.accepted()) {
1010 : 0 : result.add(wrapper_state);
1011 : 0 : continue;
1012 : : }
1013 [ + + ]: 1365 : if (!shared_data) {
1014 : 1195 : auto payload = std::make_shared<std::vector<uint8_t>>();
1015 [ + - + - ]: 1195 : if (!data.empty()) payload->assign(data.begin(), data.end());
1016 [ + + + - ]: 1195 : if (append_newline) payload->push_back('\n');
1017 : 1195 : shared_data = std::move(payload);
1018 : 1195 : }
1019 : 1365 : std::lock_guard<std::mutex> admission_lock(impl_->target_admission_mtx_);
1020 : 1365 : const auto state = target_state();
1021 [ - + ]: 1365 : if (!state.accepted()) {
1022 : 0 : result.add(state);
1023 : 0 : continue;
1024 : : }
1025 : 1365 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
1026 : 1365 : const auto it = impl_->sessions_.find(id);
1027 [ + - - + : 1365 : if (it == impl_->sessions_.end() || it->second != session) {
- + ]
1028 : 0 : result.add(wrapper::SendResult::reject(wrapper::SendRejection::NotReady));
1029 : 0 : continue;
1030 : : }
1031 : 1365 : result.add(session->try_write_shared(shared_data));
1032 : 1365 : }
1033 : 2496 : return result;
1034 : 1248 : }
1035 : :
1036 : 114 : bool TcpServer::send_to_client(ClientId client_id, std::string_view message) {
1037 : 342 : return send_to_client(client_id,
1038 : 342 : memory::ConstByteSpan(reinterpret_cast<const uint8_t*>(message.data()), message.size()));
1039 : : }
1040 : :
1041 : 122 : bool TcpServer::send_to_client(ClientId client_id, memory::ConstByteSpan data) {
1042 : 122 : const auto result = write_target(client_id, data, false);
1043 [ + + + - ]: 122 : if (auto hook = detail::g_tcp_server_write_result_hook.load()) hook(result);
1044 : 244 : return result.accepted();
1045 : : }
1046 : :
1047 : 19 : bool TcpServer::try_send_to_client(ClientId client_id, std::string_view message) {
1048 : 57 : return try_send_to_client(client_id,
1049 : 57 : memory::ConstByteSpan(reinterpret_cast<const uint8_t*>(message.data()), message.size()));
1050 : : }
1051 : :
1052 : 27 : bool TcpServer::try_send_to_client(ClientId client_id, memory::ConstByteSpan data) {
1053 : 27 : const auto result = write_target(client_id, data, true);
1054 [ + + + - ]: 27 : if (auto hook = detail::g_tcp_server_write_result_hook.load()) hook(result);
1055 : 54 : return result.accepted();
1056 : : }
1057 : :
1058 : 2454 : wrapper::SendResult TcpServer::target_state() const {
1059 [ + + ]: 2454 : if (impl_->stopping_.load()) {
1060 : 23 : std::lock_guard<std::mutex> lock(impl_->stop_mtx_);
1061 [ + + ]: 23 : return wrapper::SendResult::reject(impl_->cleanup_done_ ? wrapper::SendRejection::NotStarted
1062 : 23 : : wrapper::SendRejection::Stopping);
1063 : 23 : }
1064 [ + + ]: 4862 : if (impl_->generation_.load() == 0) return wrapper::SendResult::reject(wrapper::SendRejection::NotStarted);
1065 : 2427 : return wrapper::SendResult::accept();
1066 : : }
1067 : :
1068 : 3 : void TcpServer::fail_receive(ClientId id) {
1069 : 3 : auto session = capture_target(id);
1070 [ - + ]: 3 : if (!session) return;
1071 : 6 : net::dispatch(session->strand_, [session] { session->do_close(); });
1072 : 3 : }
1073 : :
1074 : 331 : std::optional<boost::asio::any_io_executor> TcpServer::client_executor(ClientId client_id) const {
1075 : 331 : auto session = capture_target(client_id);
1076 [ - + ]: 331 : if (!session) return std::nullopt;
1077 : 331 : return session->strand_;
1078 : 331 : }
1079 : :
1080 : 546 : std::shared_ptr<TcpServerSession> TcpServer::capture_target(ClientId client_id) const {
1081 : 546 : std::lock_guard<std::mutex> admission_lock(impl_->target_admission_mtx_);
1082 [ + - - + ]: 546 : if (!target_state().accepted()) return {};
1083 : 546 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
1084 : 546 : auto it = impl_->sessions_.find(client_id);
1085 [ + + ]: 546 : return it == impl_->sessions_.end() ? nullptr : it->second;
1086 : 546 : }
1087 : :
1088 : 298 : std::optional<wrapper::SendResult> TcpServer::poll_target_wait(const std::shared_ptr<TcpServerSession>& session) const {
1089 [ - + ]: 298 : if (!session) return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
1090 : 298 : return session->poll_write_wait();
1091 : : }
1092 : :
1093 : 318 : void TcpServer::cancel_target_waits() {
1094 : 318 : std::lock_guard<std::mutex> admission_lock(impl_->target_admission_mtx_);
1095 : 318 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
1096 [ + + ]: 448 : for (auto& entry : impl_->sessions_)
1097 [ + - + - ]: 130 : if (entry.second) entry.second->cancel_write_wait();
1098 : 318 : }
1099 : :
1100 : 316 : wrapper::SendResult TcpServer::write_target(ClientId client_id, memory::ConstByteSpan data, bool try_only,
1101 : : const std::shared_ptr<TcpServerSession>& expected) {
1102 [ + + ]: 316 : if (expected) {
1103 [ + + + - ]: 110 : if (auto hook = detail::g_tcp_server_pinned_write_hook.load()) hook();
1104 : : }
1105 : 316 : std::lock_guard<std::mutex> admission_lock(impl_->target_admission_mtx_);
1106 : 316 : const auto state = target_state();
1107 [ + + ]: 316 : if (!state.accepted()) {
1108 : 12 : impl_->stats_.record_failed_send();
1109 : 12 : return state;
1110 : : }
1111 : 304 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
1112 : 304 : auto it = impl_->sessions_.find(client_id);
1113 [ + + + - : 304 : if (it == impl_->sessions_.end() || !it->second || (expected && it->second != expected)) {
+ + - + +
+ ]
1114 : 25 : impl_->stats_.record_failed_send();
1115 : 25 : return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
1116 : : }
1117 [ + + + - : 279 : return try_only ? it->second->try_write_copy(data) : it->second->write_copy(data);
+ - ]
1118 : 316 : }
1119 : :
1120 : 516 : size_t TcpServer::client_count() const {
1121 : 516 : auto impl = get_impl();
1122 : 516 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
1123 : 516 : size_t alive = 0;
1124 [ + + ]: 978 : for (const auto& entry : impl->sessions_)
1125 [ + - + - : 462 : if (entry.second && entry.second->alive()) ++alive;
+ + + + ]
1126 : 516 : return alive;
1127 : 516 : }
1128 : :
1129 : 70 : std::optional<wrapper::RuntimeStats> TcpServer::client_stats(ClientId client_id) const {
1130 : 70 : auto impl = get_impl();
1131 : 70 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
1132 : 70 : auto it = impl->sessions_.find(client_id);
1133 [ + + - + : 70 : if (it == impl->sessions_.end() || !it->second) return std::nullopt;
+ + ]
1134 : 63 : return it->second->stats();
1135 : 70 : }
1136 : :
1137 : 92 : std::vector<ClientId> TcpServer::connected_clients() const {
1138 : 92 : auto impl = get_impl();
1139 : 92 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
1140 : 92 : std::vector<ClientId> connected_clients;
1141 : 92 : connected_clients.reserve(impl->sessions_.size());
1142 [ + + ]: 189 : for (const auto& entry : impl->sessions_)
1143 [ + - + - : 97 : if (entry.second && entry.second->alive()) connected_clients.push_back(entry.first);
+ + + + +
- ]
1144 : 184 : return connected_clients;
1145 : 92 : }
1146 : :
1147 : 376 : void TcpServer::on_multi_connect(MultiClientConnectHandler h) {
1148 : 376 : std::lock_guard<std::mutex> l(impl_->sessions_mutex_);
1149 : 376 : impl_->on_multi_connect_ = std::move(h);
1150 : 376 : }
1151 : 375 : void TcpServer::on_multi_data(MultiClientDataHandler h) {
1152 : 375 : auto shared = interface::share_callback(std::move(h));
1153 : 375 : std::lock_guard<std::mutex> l(impl_->sessions_mutex_);
1154 : 375 : impl_->on_multi_data_ = std::move(shared);
1155 : 375 : }
1156 : 372 : void TcpServer::on_multi_disconnect(MultiClientDisconnectHandler h) {
1157 : 372 : std::lock_guard<std::mutex> l(impl_->sessions_mutex_);
1158 : 372 : impl_->on_multi_disconnect_ = std::move(h);
1159 : 372 : }
1160 : :
1161 : 9 : void TcpServer::set_client_limit(size_t max) {
1162 : 9 : config::detail::range(max, 0, base::constants::MAX_MAX_CONNECTIONS, "invalid client limit");
1163 : 9 : auto impl = get_impl();
1164 [ - + ]: 9 : if (max > base::constants::MAX_MAX_CONNECTIONS) {
1165 : 0 : max = base::constants::MAX_MAX_CONNECTIONS;
1166 : : }
1167 : 9 : std::lock_guard<std::mutex> l(impl->sessions_mutex_);
1168 : 9 : impl->max_clients_ = max;
1169 [ - + ]: 9 : if (max == 0) {
1170 : 0 : impl->client_limit_enabled_ = false;
1171 : : } else {
1172 : 9 : impl->client_limit_enabled_ = true;
1173 : : }
1174 : 9 : }
1175 : :
1176 : 21 : base::LinkState TcpServer::state() const { return get_impl()->state_.get(); }
1177 : :
1178 : : } // namespace transport
1179 : : } // namespace wirestead
|
