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 <future>
25 : : #include <iostream>
26 : : #include <mutex>
27 : : #include <stop_token>
28 : : #include <string_view>
29 : : #include <thread>
30 : : #include <unordered_map>
31 : :
32 : : #include "wirestead/concurrency/io_context_manager.hpp"
33 : : #include "wirestead/concurrency/io_thread_hook.hpp"
34 : : #include "wirestead/concurrency/thread_safe_state.hpp"
35 : : #include "wirestead/diagnostics/exceptions.hpp"
36 : : #include "wirestead/diagnostics/logger.hpp"
37 : : #include "wirestead/diagnostics/runtime_stats_counter.hpp"
38 : : #include "wirestead/interface/itcp_acceptor.hpp"
39 : : #include "wirestead/transport/base/error_info_holder.hpp"
40 : : #include "wirestead/transport/tcp_server/boost_tcp_acceptor.hpp"
41 : : #include "wirestead/transport/tcp_server/ssl_tcp_socket.hpp"
42 : : #include "wirestead/transport/tcp_server/tcp_server_session.hpp"
43 : :
44 : : namespace wirestead {
45 : : namespace transport {
46 : :
47 : : namespace net = boost::asio;
48 : : using tcp = net::ip::tcp;
49 : :
50 : : struct TcpServer::Impl {
51 : : std::atomic<bool> stopping_{false};
52 : : // #503: stop() can call perform_cleanup() from two different paths - the
53 : : // dispatched-onto-the-io_context call, and (if that doesn't complete
54 : : // within the timeout below) a direct fallback call on the stopping
55 : : // thread. Under slow/instrumented builds (TSAN) the dispatched call can
56 : : // still be mid-flight when the timeout fires, so both could run
57 : : // perform_cleanup()'s body concurrently - including acceptor_->close(),
58 : : // racing with the io_context thread's own concurrent async_accept.
59 : : // Guards perform_cleanup() to run its body at most once per stop cycle,
60 : : // reset back to false in start() alongside stopping_.
61 : : std::atomic<bool> cleanup_started_{false};
62 : : std::atomic<ClientId> next_client_id_{0};
63 : :
64 : : std::unique_ptr<net::io_context> owned_ioc_;
65 : : bool owns_ioc_;
66 : : bool uses_shared_context_{false};
67 : : net::io_context& ioc_;
68 : : std::unique_ptr<net::executor_work_guard<net::io_context::executor_type>> work_guard_;
69 : : std::jthread ioc_thread_;
70 : :
71 : : std::unique_ptr<interface::TcpAcceptorInterface> acceptor_;
72 : : config::TcpServerConfig cfg_;
73 : :
74 : 141 : concurrency::AtomicLinkState state_{base::LinkState::Idle};
75 : : // Shared snapshots for the handlers the io thread copies out per received
76 : : // chunk - a std::function copy allocates whenever the target outgrows its
77 : : // small-object buffer. See interface::SharedCallback. The connect/disconnect
78 : : // handlers below stay plain: they fire once per connection, not per chunk.
79 : : interface::SharedCallback<OnBytes> on_bytes_;
80 : : interface::SharedCallback<OnState> on_state_;
81 : : interface::SharedCallback<OnBackpressure> on_bp_;
82 : : MultiClientConnectHandler on_multi_connect_;
83 : : interface::SharedCallback<MultiClientDataHandler> on_multi_data_;
84 : : MultiClientDisconnectHandler on_multi_disconnect_;
85 : : diagnostics::RuntimeStatsCounters stats_;
86 : :
87 : : mutable std::mutex sessions_mutex_;
88 : : std::unordered_map<ClientId, std::shared_ptr<TcpServerSession>> sessions_;
89 : :
90 : : size_t max_clients_;
91 : : bool client_limit_enabled_;
92 : :
93 : : std::shared_ptr<TcpServerSession> current_session_;
94 : :
95 : : ErrorInfoHolder error_info_holder_{"tcp_server"};
96 : :
97 : 132 : explicit Impl(const config::TcpServerConfig& cfg, bool use_shared_context)
98 [ + + ]: 132 : : owned_ioc_(use_shared_context ? nullptr : std::make_unique<net::io_context>()),
99 : 132 : owns_ioc_(!use_shared_context),
100 : 132 : uses_shared_context_(use_shared_context),
101 [ + + + - : 132 : ioc_(use_shared_context ? concurrency::IoContextManager::instance().get_context() : *owned_ioc_),
+ - ]
102 : 132 : cfg_(cfg),
103 : 132 : max_clients_(cfg.max_connections > 0 ? static_cast<size_t>(cfg.max_connections) : 0),
104 [ + - ]: 660 : client_limit_enabled_(cfg.max_connections > 0) {
105 : : try {
106 : 132 : acceptor_ = std::make_unique<BoostTcpAcceptor>(ioc_);
107 : 0 : } catch (const std::exception& e) {
108 : 0 : throw diagnostics::BuilderException("Failed to create TCP acceptor: " + std::string(e.what()), "tcp_server");
109 : 0 : }
110 : 132 : cfg_.validate_and_clamp();
111 : 132 : max_clients_ = cfg_.max_connections > 0 ? static_cast<size_t>(cfg_.max_connections) : 0;
112 : 132 : client_limit_enabled_ = cfg_.max_connections > 0;
113 : 132 : }
114 : :
115 : 9 : Impl(const config::TcpServerConfig& cfg, std::unique_ptr<interface::TcpAcceptorInterface> acceptor,
116 : : net::io_context& ioc)
117 : 18 : : owns_ioc_(false),
118 : 9 : ioc_(ioc),
119 : 9 : acceptor_(std::move(acceptor)),
120 : 9 : cfg_(cfg),
121 : 9 : max_clients_(cfg.max_connections > 0 ? static_cast<size_t>(cfg.max_connections) : 0),
122 [ + - ]: 36 : client_limit_enabled_(cfg.max_connections > 0) {
123 [ + + ]: 9 : if (!acceptor_) {
124 : 7 : throw diagnostics::BuilderException("Failed to create TCP acceptor", "tcp_server");
125 : : }
126 : 8 : cfg_.validate_and_clamp();
127 : 8 : max_clients_ = cfg_.max_connections > 0 ? static_cast<size_t>(cfg_.max_connections) : 0;
128 : 8 : client_limit_enabled_ = cfg_.max_connections > 0;
129 : 23 : }
130 : :
131 : 135 : ~Impl() {
132 : : try {
133 : 135 : stopping_.store(true);
134 [ - + ]: 135 : if (ioc_thread_.joinable()) {
135 [ # # ]: 0 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
136 : 0 : ioc_thread_.detach();
137 : : } else {
138 : 0 : ioc_thread_.request_stop();
139 : 0 : ioc_thread_.join();
140 : : }
141 : : }
142 : 135 : perform_cleanup();
143 : 0 : } catch (...) {
144 : 0 : }
145 : 135 : }
146 : :
147 : 506 : void notify_state() {
148 [ + + ]: 506 : if (stopping_.load()) return;
149 : 351 : interface::SharedCallback<OnState> cb;
150 : : try {
151 : : {
152 : 351 : std::lock_guard<std::mutex> lock(sessions_mutex_);
153 : 351 : cb = on_state_;
154 : 351 : }
155 [ + + ]: 351 : if (cb) {
156 : 330 : (*cb)(state_.get());
157 : : }
158 : 1 : } catch (...) {
159 : 1 : }
160 : 351 : }
161 : :
162 : : // One context for the whole server, shared by every accepted connection.
163 : : // Built once in start() so a bad certificate fails there rather than on the
164 : : // first client.
165 : : #ifdef WIRESTEAD_TLS_ENABLED
166 : : std::shared_ptr<boost::asio::ssl::context> ssl_context_;
167 : : #endif
168 : :
169 : : // Builds the ssl::context, or reports why it cannot. Called from start(), so
170 : : // a certificate problem surfaces as a start failure with a message rather
171 : : // than as connections that mysteriously drop at handshake.
172 : 139 : std::string init_tls() {
173 : : // Half a TLS config is a request for TLS, not a request for plaintext. An
174 : : // empty environment variable or a typo would otherwise leave tls_enabled()
175 : : // false and bring the server up unencrypted without saying anything, which
176 : : // is the one outcome this whole path exists to prevent.
177 [ + + ]: 139 : if (!cfg_.tls_certificate_file.empty() != !cfg_.tls_private_key_file.empty()) {
178 : 2 : return cfg_.tls_certificate_file.empty() ? "TLS needs a certificate as well as a private key"
179 : 4 : : "TLS needs a private key as well as a certificate";
180 : : }
181 [ + + ]: 137 : if (!cfg_.tls_enabled()) return {};
182 : : #ifndef WIRESTEAD_TLS_ENABLED
183 : : return "TLS was configured but this build has WIRESTEAD_ENABLE_TLS=OFF";
184 : : #else
185 : : namespace ssl = boost::asio::ssl;
186 : : try {
187 : 4 : auto ctx = std::make_shared<ssl::context>(ssl::context::tls_server);
188 : : // Anything below TLS 1.2 is broken in ways nobody should opt into by
189 : : // accident, so the floor is not configurable.
190 : 4 : ctx->set_options(ssl::context::default_workarounds | ssl::context::no_sslv2 | ssl::context::no_sslv3 |
191 : : ssl::context::no_tlsv1 | ssl::context::no_tlsv1_1 | ssl::context::single_dh_use);
192 : 4 : ctx->use_certificate_chain_file(cfg_.tls_certificate_file);
193 : 3 : ctx->use_private_key_file(cfg_.tls_private_key_file, ssl::context::pem);
194 : 3 : ssl_context_ = std::move(ctx);
195 : 3 : return {};
196 : 5 : } catch (const std::exception& e) {
197 : 3 : return std::string("Failed to load TLS certificate or key: ") + e.what();
198 : 1 : }
199 : : #endif
200 : : }
201 : :
202 : 140 : std::shared_ptr<TcpServerSession> make_session(tcp::socket sock) {
203 : : #ifdef WIRESTEAD_TLS_ENABLED
204 [ + + ]: 140 : if (ssl_context_) {
205 : : return std::make_shared<TcpServerSession>(
206 : 64 : ioc_, std::make_unique<SslTcpSocket>(std::move(sock), ssl_context_), cfg_.backpressure_threshold,
207 : 64 : cfg_.idle_timeout_ms, cfg_.backpressure_strategy, cfg_.enable_memory_pool, cfg_.read_buffer_size);
208 : : }
209 : : #endif
210 : 108 : return std::make_shared<TcpServerSession>(ioc_, std::move(sock), cfg_.backpressure_threshold, cfg_.idle_timeout_ms,
211 : 108 : cfg_.backpressure_strategy, cfg_.enable_memory_pool,
212 : 108 : cfg_.read_buffer_size);
213 : : }
214 : :
215 : 140 : void apply_accepted_socket_options(tcp::socket& sock) {
216 : 140 : boost::system::error_code ec;
217 : :
218 [ + - ]: 140 : if (cfg_.tcp_no_delay) {
219 : 140 : sock.set_option(tcp::no_delay(true), ec);
220 [ - + ]: 140 : if (ec) {
221 : 0 : WIRESTEAD_LOG_WARNING("tcp_server", "socket_options",
222 : : fmt::format("Failed to set TCP_NODELAY: {}", ec.message()));
223 : 0 : ec.clear();
224 : : }
225 : : }
226 : :
227 [ - + ]: 140 : if (cfg_.keep_alive) {
228 : 0 : sock.set_option(net::socket_base::keep_alive(true), ec);
229 [ # # ]: 0 : if (ec) {
230 : 0 : WIRESTEAD_LOG_WARNING("tcp_server", "socket_options",
231 : : fmt::format("Failed to set keep_alive: {}", ec.message()));
232 : 0 : ec.clear();
233 : : }
234 : : }
235 : :
236 [ + + ]: 140 : if (cfg_.send_buffer_size > 0) {
237 : 4 : sock.set_option(net::socket_base::send_buffer_size(static_cast<int>(cfg_.send_buffer_size)), ec);
238 [ - + ]: 4 : if (ec) {
239 : 0 : WIRESTEAD_LOG_WARNING("tcp_server", "socket_options",
240 : : fmt::format("Failed to set send buffer size: {}", ec.message()));
241 : 0 : ec.clear();
242 : : }
243 : : }
244 : :
245 [ - + ]: 140 : if (cfg_.receive_buffer_size > 0) {
246 : 0 : sock.set_option(net::socket_base::receive_buffer_size(static_cast<int>(cfg_.receive_buffer_size)), ec);
247 [ # # ]: 0 : if (ec) {
248 : 0 : WIRESTEAD_LOG_WARNING("tcp_server", "socket_options",
249 : : fmt::format("Failed to set receive buffer size: {}", ec.message()));
250 : 0 : ec.clear();
251 : : }
252 : : }
253 : 140 : }
254 : :
255 : 133 : void attempt_port_binding(std::shared_ptr<TcpServer> self, int retry_count) {
256 [ - + ]: 140 : if (stopping_.load()) return;
257 : 133 : boost::system::error_code ec;
258 : :
259 : 133 : auto address = net::ip::make_address(cfg_.bind_address, ec);
260 [ + + ]: 133 : if (ec) {
261 : 2 : std::string msg = fmt::format("Invalid bind address: {}, {}", cfg_.bind_address, ec.message());
262 : 2 : WIRESTEAD_LOG_ERROR("tcp_server", "bind", msg);
263 : 2 : error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONFIGURATION, "bind",
264 : : ec, msg, false, static_cast<uint32_t>(retry_count));
265 : 2 : state_.set(base::LinkState::Error);
266 : 2 : notify_state();
267 : 2 : return;
268 : 2 : }
269 : :
270 [ + - + + ]: 131 : if (!acceptor_->is_open()) {
271 [ - + + - ]: 129 : acceptor_->open(address.is_v6() ? tcp::v6() : tcp::v4(), ec);
272 [ + + ]: 129 : if (ec) {
273 : 1 : std::string msg = fmt::format("Failed to open acceptor: {}", ec.message());
274 : 1 : WIRESTEAD_LOG_ERROR("tcp_server", "open", msg);
275 : 1 : error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::SYSTEM, "open", ec,
276 : : msg, false, static_cast<uint32_t>(retry_count));
277 : 1 : state_.set(base::LinkState::Error);
278 : 1 : notify_state();
279 : 1 : return;
280 : 1 : }
281 : : }
282 : :
283 : 130 : acceptor_->bind(tcp::endpoint(address, cfg_.port), ec);
284 [ + + ]: 130 : if (ec) {
285 [ + + + - ]: 3 : if (cfg_.enable_port_retry && retry_count < cfg_.max_port_retries) {
286 : 2 : auto timer = std::make_shared<net::steady_timer>(ioc_);
287 : 2 : timer->expires_after(std::chrono::milliseconds(cfg_.port_retry_interval_ms));
288 : 2 : timer->async_wait([self, retry_count, timer](const boost::system::error_code& timer_ec) {
289 [ + - ]: 2 : if (!timer_ec) {
290 : 2 : auto* timer_impl = self->get_impl();
291 [ + - ]: 2 : if (!timer_impl->stopping_.load()) {
292 : 2 : timer_impl->attempt_port_binding(self, retry_count + 1);
293 : : }
294 : : }
295 : 2 : });
296 : 2 : return;
297 : 2 : } else {
298 : 1 : std::string msg = fmt::format("Failed to bind to port {}: {}", cfg_.port, ec.message());
299 : 1 : WIRESTEAD_LOG_ERROR("tcp_server", "bind", msg);
300 : 1 : error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "bind",
301 : : ec, msg, false, static_cast<uint32_t>(retry_count));
302 : 1 : state_.set(base::LinkState::Error);
303 : 1 : notify_state();
304 : 1 : return;
305 : 1 : }
306 : : }
307 : :
308 : 127 : acceptor_->listen(boost::asio::socket_base::max_listen_connections, ec);
309 [ + + ]: 127 : if (ec) {
310 : 1 : std::string msg = fmt::format("Failed to listen on port {}: {}", cfg_.port, ec.message());
311 : 1 : WIRESTEAD_LOG_ERROR("tcp_server", "listen", msg);
312 : 1 : error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, "listen",
313 : : ec, msg, false, static_cast<uint32_t>(retry_count));
314 : 1 : state_.set(base::LinkState::Error);
315 : 1 : notify_state();
316 : 1 : return;
317 : 1 : }
318 : :
319 : 126 : state_.set(base::LinkState::Listening);
320 : 126 : notify_state();
321 : 126 : do_accept(self);
322 : : }
323 : :
324 : 2057 : void do_accept(std::shared_ptr<TcpServer> self) {
325 [ + + + - : 2057 : if (stopping_.load() || !acceptor_ || !acceptor_->is_open()) return;
- + + + ]
326 : :
327 : 2042 : acceptor_->async_accept([self](auto ec, tcp::socket sock) {
328 : 2040 : auto* accept_impl = self->get_impl();
329 [ + + ]: 2040 : if (accept_impl->stopping_.load()) {
330 : 1900 : return;
331 : : }
332 [ + + ]: 1932 : if (ec) {
333 [ + - ]: 1 : if (ec != boost::asio::error::operation_aborted) {
334 : 1 : accept_impl->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR,
335 : : diagnostics::ErrorCategory::CONNECTION, "accept", ec,
336 : : fmt::format("Accept failed: {}", ec.message()), true, 0);
337 : 1 : accept_impl->state_.set(base::LinkState::Error);
338 : 1 : accept_impl->notify_state();
339 : : }
340 [ + - + - : 1 : if (!accept_impl->state_.is_state(base::LinkState::Closed) && !accept_impl->stopping_.load()) {
+ - ]
341 : 1 : auto timer = std::make_shared<net::steady_timer>(accept_impl->ioc_);
342 : 1 : timer->expires_after(std::chrono::milliseconds(100));
343 : 1 : timer->async_wait([self, timer](const boost::system::error_code&) {
344 : 0 : auto* retry_impl = self->get_impl();
345 [ # # ]: 0 : if (!retry_impl->stopping_.load()) {
346 : 0 : retry_impl->do_accept(self);
347 : : }
348 : : });
349 : 1 : }
350 : 1 : return;
351 : : }
352 : :
353 : 1931 : boost::system::error_code ep_ec;
354 : 1931 : auto rep = sock.remote_endpoint(ep_ec);
355 : 1931 : std::string client_info = "unknown";
356 [ + - ]: 1931 : if (!ep_ec) {
357 : 1931 : client_info = fmt::format("{}:{}", rep.address().to_string(), rep.port());
358 : : }
359 : :
360 [ + - ]: 1931 : if (accept_impl->client_limit_enabled_) {
361 : : bool over_limit;
362 : : {
363 : 1931 : std::lock_guard<std::mutex> lock(accept_impl->sessions_mutex_);
364 : 1931 : over_limit = accept_impl->sessions_.size() >= accept_impl->max_clients_;
365 : 1931 : }
366 [ + + ]: 1931 : if (over_limit) {
367 : : // #437: accept and immediately close over-limit connections
368 : : // instead of pausing the accept loop entirely - a client whose
369 : : // TCP handshake already completed while paused would otherwise
370 : : // sit connected but silent until a slot frees up.
371 : 1791 : boost::system::error_code close_ec;
372 : 1791 : sock.close(close_ec);
373 : 1791 : accept_impl->do_accept(self);
374 : 1791 : return;
375 : : }
376 : : }
377 : :
378 : 140 : accept_impl->apply_accepted_socket_options(sock);
379 : :
380 : : // The session only talks to TcpSocketInterface, so TLS is a matter of
381 : : // which implementation it gets wrapped in here. Everything downstream -
382 : : // reads, writes, backpressure, stats - is identical either way.
383 : 140 : auto new_session = accept_impl->make_session(std::move(sock));
384 : :
385 : 140 : ClientId client_id = accept_impl->next_client_id_.fetch_add(1);
386 : :
387 : 140 : std::weak_ptr<TcpServer> weak_self = self;
388 : :
389 : 140 : new_session->on_bytes([weak_self, client_id](memory::ConstByteSpan data) {
390 : 2473 : auto shared_self = weak_self.lock();
391 [ - + ]: 2473 : if (!shared_self) return;
392 : 2473 : auto* bytes_impl = shared_self->get_impl();
393 : :
394 : 2473 : interface::SharedCallback<OnBytes> cb;
395 : 2473 : interface::SharedCallback<MultiClientDataHandler> multi_cb;
396 : : {
397 : 2473 : std::lock_guard<std::mutex> lock(bytes_impl->sessions_mutex_);
398 : 2473 : cb = bytes_impl->on_bytes_;
399 : 2473 : multi_cb = bytes_impl->on_multi_data_;
400 : 2473 : }
401 [ + + + - ]: 2473 : if (cb) (*cb)(data);
402 [ + + ]: 2473 : if (multi_cb) {
403 : 2466 : (*multi_cb)(client_id, data);
404 : : }
405 : 2473 : });
406 : :
407 : 140 : interface::SharedCallback<OnBackpressure> bp_cb;
408 : : {
409 : 140 : std::lock_guard<std::mutex> lock(accept_impl->sessions_mutex_);
410 : 140 : bp_cb = accept_impl->on_bp_;
411 : 140 : }
412 [ + + + - : 140 : if (bp_cb) new_session->on_backpressure(*bp_cb);
+ - ]
413 : :
414 : 140 : new_session->on_close([weak_self, client_id, new_session] {
415 : 128 : auto shared_self = weak_self.lock();
416 [ - + ]: 128 : if (!shared_self) return;
417 : 128 : auto* close_impl = shared_self->get_impl();
418 [ + + ]: 128 : if (close_impl->stopping_.load()) return;
419 : :
420 : 119 : MultiClientDisconnectHandler disconnect_cb;
421 : : {
422 : 119 : std::lock_guard<std::mutex> lock(close_impl->sessions_mutex_);
423 : 119 : disconnect_cb = close_impl->on_multi_disconnect_;
424 : 119 : }
425 [ + + + - ]: 119 : if (disconnect_cb) disconnect_cb(client_id);
426 : :
427 : 119 : bool was_current = false;
428 : : {
429 : 119 : std::lock_guard<std::mutex> lock(close_impl->sessions_mutex_);
430 : : // Carry the session's totals over to the server before it goes away,
431 : : // so stats() keeps reporting what this connection did. Tied to the
432 : : // erase below, which makes it exactly once even if on_close re-fires.
433 : 119 : auto it = close_impl->sessions_.find(client_id);
434 [ + - + - : 119 : if (it != close_impl->sessions_.end() && it->second) {
+ - ]
435 : 119 : close_impl->stats_.absorb(it->second->stats());
436 : : }
437 : 119 : close_impl->sessions_.erase(client_id);
438 : 119 : was_current = (close_impl->current_session_ == new_session);
439 [ + + ]: 119 : if (was_current) {
440 [ + + ]: 91 : if (!close_impl->sessions_.empty())
441 : 4 : close_impl->current_session_ = close_impl->sessions_.begin()->second;
442 : : else
443 : 87 : close_impl->current_session_.reset();
444 : : }
445 : 119 : }
446 [ + + ]: 119 : if (was_current) {
447 : 91 : close_impl->state_.set(base::LinkState::Listening);
448 : 91 : close_impl->notify_state();
449 : : }
450 : 128 : });
451 : :
452 : : // alive_ must be true before the session enters sessions_, so that
453 : : // broadcast() callers who observe client_count() >= 1 are guaranteed
454 : : // to pass the alive() check inside async_try_write_shared().
455 : 140 : new_session->start();
456 : :
457 : : {
458 : 140 : std::lock_guard<std::mutex> lock(accept_impl->sessions_mutex_);
459 : 140 : accept_impl->sessions_.emplace(client_id, new_session);
460 : 140 : accept_impl->current_session_ = new_session;
461 : 140 : }
462 : :
463 : 140 : MultiClientConnectHandler connect_cb;
464 : : {
465 : 140 : std::lock_guard<std::mutex> lock(accept_impl->sessions_mutex_);
466 : 140 : connect_cb = accept_impl->on_multi_connect_;
467 : 140 : }
468 [ + + + - ]: 140 : if (connect_cb) connect_cb(client_id, client_info);
469 : :
470 : 140 : accept_impl->state_.set(base::LinkState::Connected);
471 : 140 : accept_impl->notify_state();
472 : 140 : accept_impl->do_accept(self);
473 : 1931 : });
474 : : }
475 : :
476 : 275 : void perform_cleanup() {
477 [ + + ]: 275 : if (cleanup_started_.exchange(true)) return;
478 : : try {
479 : 140 : boost::system::error_code ec;
480 [ + - + - : 140 : if (acceptor_ && acceptor_->is_open()) {
+ + + + ]
481 : 128 : acceptor_->close(ec);
482 : : }
483 : :
484 : 140 : std::vector<std::shared_ptr<TcpServerSession>> sessions_copy;
485 : : {
486 : 140 : std::lock_guard<std::mutex> lock(sessions_mutex_);
487 : 140 : sessions_copy.reserve(sessions_.size());
488 [ + + ]: 161 : for (auto& kv : sessions_) {
489 : 21 : sessions_copy.push_back(kv.second);
490 : : }
491 : 140 : sessions_.clear();
492 : 140 : current_session_.reset();
493 : 140 : }
494 : :
495 [ + + ]: 161 : for (auto& session : sessions_copy) {
496 [ + - ]: 21 : if (session) {
497 : 21 : session->stop();
498 : : }
499 : : }
500 : :
501 : 140 : state_.set(base::LinkState::Closed);
502 : 140 : notify_state();
503 : 140 : } catch (...) {
504 : 0 : }
505 : : }
506 : :
507 : 260 : void stop(std::shared_ptr<TcpServer> self) {
508 [ + + ]: 260 : if (stopping_.exchange(true)) {
509 : 120 : return;
510 : : }
511 : :
512 : : {
513 : 140 : std::lock_guard<std::mutex> lock(sessions_mutex_);
514 : 140 : on_bytes_ = nullptr;
515 : 140 : on_state_ = nullptr;
516 : 140 : on_bp_ = nullptr;
517 : 140 : on_multi_connect_ = nullptr;
518 : 140 : on_multi_data_ = nullptr;
519 : 140 : on_multi_disconnect_ = nullptr;
520 : 140 : }
521 : :
522 [ - + ]: 140 : if (ioc_.get_executor().running_in_this_thread()) {
523 : 0 : perform_cleanup();
524 [ # # ]: 0 : if (owns_ioc_) ioc_.stop();
525 : 0 : return;
526 : : }
527 : :
528 : : // #503: previously gated on has_active_ioc (owns_ioc_ || a running
529 : : // shared IoContextManager) before deciding to dispatch perform_cleanup()
530 : : // onto ioc_ vs. calling it directly - but a "managed external context"
531 : : // (owns_ioc_ == false, since some other owner constructed the
532 : : // io_context, yet that owner's own thread may still be actively
533 : : // running it, e.g. the wrapper layer's own io_context+thread) fell
534 : : // through to the direct-call branch, racing acceptor_->close() against
535 : : // that thread's concurrent async_accept(). Whenever we have a valid
536 : : // self to dispatch through, always prefer dispatching onto ioc_ (with
537 : : // the same timeout-based direct-call fallback for the case where
538 : : // nothing is actually pumping it) - a stale, unserviced io_context only
539 : : // costs one extra harmless wait before falling back to the exact same
540 : : // direct call as before; self is only null when called from the
541 : : // destructor, where shared_from_this() isn't available and a direct
542 : : // call is the only option.
543 [ + - ]: 140 : if (self) {
544 : 140 : auto cleanup_promise = std::make_shared<std::promise<void>>();
545 : 140 : auto cleanup_future = cleanup_promise->get_future();
546 : :
547 : 140 : std::weak_ptr<TcpServer> weak_self = self;
548 : 140 : net::dispatch(ioc_, [weak_self, cleanup_promise]() {
549 [ + - ]: 132 : if (auto shared_self = weak_self.lock()) {
550 : 132 : auto* cleanup_impl = shared_self->get_impl();
551 : 132 : cleanup_impl->perform_cleanup();
552 : 132 : }
553 : 132 : cleanup_promise->set_value();
554 : 132 : });
555 : :
556 [ + - + + ]: 140 : if (cleanup_future.wait_for(std::chrono::seconds(2)) == std::future_status::timeout) {
557 : 8 : perform_cleanup();
558 : : }
559 : 140 : } else {
560 : 0 : perform_cleanup();
561 : : }
562 : :
563 [ + + ]: 140 : if (owns_ioc_) {
564 [ + + ]: 130 : if (work_guard_) work_guard_->reset();
565 [ + + ]: 130 : if (ioc_thread_.joinable()) {
566 [ - + ]: 126 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
567 : 0 : ioc_thread_.detach();
568 : : } else {
569 : 126 : ioc_thread_.request_stop();
570 : 126 : ioc_thread_.join();
571 : : }
572 : 126 : ioc_.restart();
573 : : }
574 : : }
575 : : }
576 : : };
577 : :
578 : 132 : std::shared_ptr<TcpServer> TcpServer::create(const config::TcpServerConfig& cfg, bool use_shared_context) {
579 : 132 : return std::shared_ptr<TcpServer>(new TcpServer(cfg, use_shared_context));
580 : : }
581 : :
582 : 9 : std::shared_ptr<TcpServer> TcpServer::create(const config::TcpServerConfig& cfg,
583 : : std::unique_ptr<interface::TcpAcceptorInterface> acceptor,
584 : : net::io_context& ioc) {
585 : 10 : return std::shared_ptr<TcpServer>(new TcpServer(cfg, std::move(acceptor), ioc));
586 : : }
587 : :
588 : 132 : TcpServer::TcpServer(const config::TcpServerConfig& cfg, bool use_shared_context)
589 : 132 : : impl_(std::make_unique<Impl>(cfg, use_shared_context)) {}
590 : :
591 : 9 : TcpServer::TcpServer(const config::TcpServerConfig& cfg, std::unique_ptr<interface::TcpAcceptorInterface> acceptor,
592 : 9 : net::io_context& ioc)
593 : 10 : : impl_(std::make_unique<Impl>(cfg, std::move(acceptor), ioc)) {}
594 : :
595 : 270 : TcpServer::~TcpServer() {
596 [ + - - + : 135 : if (impl_ && !impl_->state_.is_state(base::LinkState::Closed)) {
- + ]
597 : : // Pass nullptr to stop() to indicate we are in destructor and cannot use shared_from_this
598 : 0 : impl_->stop(nullptr);
599 : : }
600 : 270 : }
601 : :
602 : 0 : TcpServer::TcpServer(TcpServer&&) noexcept = default;
603 : 0 : TcpServer& TcpServer::operator=(TcpServer&&) noexcept = default;
604 : :
605 : 140 : void TcpServer::start() {
606 : 140 : auto impl = get_impl();
607 : 140 : auto current = impl->state_.get();
608 [ + + + - : 140 : if (current == base::LinkState::Listening || current == base::LinkState::Connected ||
- + ]
609 : : current == base::LinkState::Connecting) {
610 : 4 : return;
611 : : }
612 : 139 : impl->stopping_.store(false);
613 : 139 : impl->cleanup_started_.store(false);
614 : : // Restart contract (#444): stats() resets on restart. The server-level
615 : : // counters now outlive the sessions that fed them, so clearing them here is
616 : : // what keeps that promise - before absorption they were empty and a restart
617 : : // zeroed the aggregate for free.
618 : 139 : impl->stats_.reset(0);
619 : :
620 : : // Load the certificate before binding. A server asked for TLS that cannot
621 : : // provide it must not come up in plaintext instead - that is the failure mode
622 : : // where everything looks healthy and nothing is encrypted.
623 [ + - + + ]: 139 : if (const auto tls_error = impl->init_tls(); !tls_error.empty()) {
624 : 3 : WIRESTEAD_LOG_ERROR("tcp_server", "start", tls_error);
625 : 3 : impl->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONFIGURATION,
626 : : "start", {}, tls_error, false, 0);
627 : 3 : impl->state_.set(base::LinkState::Error);
628 : 3 : impl->notify_state();
629 : 3 : return;
630 : 139 : }
631 : :
632 [ + + ]: 136 : if (impl->uses_shared_context_) {
633 : 2 : auto& manager = concurrency::IoContextManager::instance();
634 [ + - - + ]: 2 : if (!manager.is_running()) {
635 : 0 : manager.start();
636 : : }
637 : : }
638 : :
639 [ - + ]: 136 : if (!impl->acceptor_) {
640 : 0 : impl->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::SYSTEM, "start",
641 : : {}, "No acceptor available", false, 0);
642 : 0 : impl->state_.set(base::LinkState::Error);
643 : 0 : impl->notify_state();
644 : 0 : return;
645 : : }
646 : :
647 [ + + + - : 136 : if (impl->owns_ioc_ && !impl->ioc_thread_.joinable()) {
+ + ]
648 [ + - - + ]: 126 : if (impl->ioc_.stopped()) {
649 : 0 : impl->ioc_.restart();
650 : : }
651 : : impl->work_guard_ =
652 : 126 : std::make_unique<net::executor_work_guard<net::io_context::executor_type>>(impl->ioc_.get_executor());
653 : 252 : impl->ioc_thread_ = std::jthread([impl](std::stop_token st) {
654 : 126 : wirestead::concurrency::run_io_thread_init();
655 : : try {
656 : 252 : std::stop_callback cb(st, [impl] { impl->ioc_.stop(); });
657 : 126 : impl->ioc_.run();
658 : 126 : } catch (...) {
659 : 0 : }
660 : 252 : });
661 : : }
662 : 136 : auto self = shared_from_this();
663 [ - + ]: 136 : if (impl->ioc_.get_executor().running_in_this_thread()) {
664 [ # # ]: 0 : if (!impl->stopping_.load()) {
665 : 0 : impl->attempt_port_binding(self, 0);
666 : : }
667 : : } else {
668 : 136 : net::dispatch(impl->ioc_, [self] {
669 : 136 : auto impl = self->get_impl();
670 [ + + ]: 136 : if (impl->stopping_.load()) return;
671 : 131 : impl->attempt_port_binding(self, 0);
672 : : });
673 : : }
674 : 136 : }
675 : :
676 : 260 : void TcpServer::stop() { impl_->stop(shared_from_this()); }
677 : :
678 : 120 : void TcpServer::request_stop() {
679 : 120 : auto impl = get_impl();
680 [ - + ]: 120 : if (impl->stopping_.load()) return;
681 : 120 : auto self = shared_from_this();
682 : 237 : net::post(impl->ioc_, [self] { self->stop(); });
683 : 120 : }
684 : :
685 : 1 : bool TcpServer::is_connected() const {
686 : 1 : auto impl = get_impl();
687 : 1 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
688 [ + - + - : 2 : return impl->current_session_ && impl->current_session_->alive();
+ - ]
689 : 1 : }
690 : :
691 : 1 : bool TcpServer::is_backpressure_active() const { return false; }
692 : :
693 : 5 : bool TcpServer::is_backpressure_active(ClientId client_id) const {
694 : 5 : auto impl = get_impl();
695 : 5 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
696 : 5 : auto it = impl->sessions_.find(client_id);
697 [ + - + - : 5 : if (it != impl->sessions_.end() && it->second) {
+ - ]
698 : 5 : return it->second->is_backpressure_active();
699 : : }
700 : 0 : return false;
701 : 5 : }
702 : :
703 : 120 : boost::asio::any_io_executor TcpServer::get_executor() { return impl_->ioc_.get_executor(); }
704 : :
705 : 1886 : wrapper::RuntimeStats TcpServer::stats() const {
706 : 1886 : auto impl = get_impl();
707 : 1886 : auto aggregate = impl->stats_.snapshot(0, 0, false);
708 : 1886 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
709 [ + + ]: 1899 : for (const auto& entry : impl->sessions_) {
710 [ - + ]: 13 : if (!entry.second) continue;
711 : 13 : const auto session_stats = entry.second->stats();
712 : 13 : aggregate.bytes_accepted += session_stats.bytes_accepted;
713 : 13 : aggregate.messages_accepted += session_stats.messages_accepted;
714 : 13 : aggregate.bytes_sent += session_stats.bytes_sent;
715 : 13 : aggregate.messages_sent += session_stats.messages_sent;
716 : 13 : aggregate.bytes_received += session_stats.bytes_received;
717 : 13 : aggregate.messages_received += session_stats.messages_received;
718 : 13 : aggregate.failed_sends += session_stats.failed_sends;
719 : 13 : aggregate.dropped_messages += session_stats.dropped_messages;
720 : 13 : aggregate.dropped_bytes += session_stats.dropped_bytes;
721 : 13 : aggregate.backpressure_events += session_stats.backpressure_events;
722 : 13 : aggregate.queued_bytes += session_stats.queued_bytes;
723 : 13 : aggregate.pending_bytes += session_stats.pending_bytes;
724 : : // Peak, not a total: summing per-session peaks would report a depth no
725 : : // session ever reached, because the peaks need not have been simultaneous.
726 : 13 : aggregate.max_queued_bytes = std::max(aggregate.max_queued_bytes, session_stats.max_queued_bytes);
727 [ + + + + ]: 13 : aggregate.backpressure_active = aggregate.backpressure_active || session_stats.backpressure_active;
728 : : }
729 : 3772 : return aggregate;
730 : 1886 : }
731 : :
732 : 0 : void TcpServer::reset_stats() {
733 : 0 : auto impl = get_impl();
734 : 0 : impl->stats_.reset(0);
735 : 0 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
736 [ # # ]: 0 : for (const auto& entry : impl->sessions_) {
737 [ # # # # ]: 0 : if (entry.second) entry.second->reset_stats();
738 : : }
739 : 0 : }
740 : :
741 : 5 : std::optional<diagnostics::ErrorInfo> TcpServer::last_error_info() const {
742 : 5 : return get_impl()->error_info_holder_.last_error_info();
743 : : }
744 : :
745 : 2 : bool TcpServer::async_write_copy(memory::ConstByteSpan data) {
746 : 2 : auto impl = get_impl();
747 [ - + ]: 2 : if (impl->stopping_.load()) {
748 : 0 : impl->stats_.record_failed_send();
749 : 0 : return false;
750 : : }
751 : 2 : std::shared_ptr<TcpServerSession> session;
752 : : {
753 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
754 : 2 : session = impl->current_session_;
755 : 2 : }
756 : :
757 [ + + + - : 2 : if (session && session->alive()) {
+ - + + ]
758 : 1 : return session->async_write_copy(data);
759 : : }
760 : 1 : impl->stats_.record_failed_send();
761 : 1 : return false;
762 : 2 : }
763 : :
764 : 2 : bool TcpServer::async_write_move(std::vector<uint8_t>&& data) {
765 : 2 : auto impl = get_impl();
766 [ - + ]: 2 : if (impl->stopping_.load()) {
767 : 0 : impl->stats_.record_failed_send();
768 : 0 : return false;
769 : : }
770 : 2 : std::shared_ptr<TcpServerSession> session;
771 : : {
772 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
773 : 2 : session = impl->current_session_;
774 : 2 : }
775 : :
776 [ + + + - : 2 : if (session && session->alive()) {
+ - + + ]
777 : 1 : return session->async_write_move(std::move(data));
778 : : }
779 : 1 : impl->stats_.record_failed_send();
780 : 1 : return false;
781 : 2 : }
782 : :
783 : 2 : bool TcpServer::async_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
784 : 2 : auto impl = get_impl();
785 [ + - + - : 2 : if (impl->stopping_.load() || !data || data->empty()) {
- + - + ]
786 : 0 : impl->stats_.record_failed_send();
787 : 0 : return false;
788 : : }
789 : 2 : std::shared_ptr<TcpServerSession> session;
790 : : {
791 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
792 : 2 : session = impl->current_session_;
793 : 2 : }
794 : :
795 [ + + + - : 2 : if (session && session->alive()) {
+ - + + ]
796 : 1 : return session->async_write_shared(std::move(data));
797 : : }
798 : 1 : impl->stats_.record_failed_send();
799 : 1 : return false;
800 : 2 : }
801 : :
802 : 2 : bool TcpServer::async_try_write_copy(memory::ConstByteSpan data) {
803 : 2 : auto impl = get_impl();
804 [ - + ]: 2 : if (impl->stopping_.load()) {
805 : 0 : impl->stats_.record_failed_send();
806 : 0 : return false;
807 : : }
808 : 2 : std::shared_ptr<TcpServerSession> session;
809 : : {
810 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
811 : 2 : session = impl->current_session_;
812 : 2 : }
813 : :
814 [ + + + - : 2 : if (session && session->alive()) {
+ - + + ]
815 : 1 : return session->async_try_write_copy(data);
816 : : }
817 : 1 : impl->stats_.record_failed_send();
818 : 1 : return false;
819 : 2 : }
820 : :
821 : 2 : bool TcpServer::async_try_write_move(std::vector<uint8_t>&& data) {
822 : 2 : auto impl = get_impl();
823 [ - + ]: 2 : if (impl->stopping_.load()) {
824 : 0 : impl->stats_.record_failed_send();
825 : 0 : return false;
826 : : }
827 : 2 : std::shared_ptr<TcpServerSession> session;
828 : : {
829 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
830 : 2 : session = impl->current_session_;
831 : 2 : }
832 : :
833 [ + + + - : 2 : if (session && session->alive()) {
+ - + + ]
834 : 1 : return session->async_try_write_move(std::move(data));
835 : : }
836 : 1 : impl->stats_.record_failed_send();
837 : 1 : return false;
838 : 2 : }
839 : :
840 : 2 : bool TcpServer::async_try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
841 : 2 : auto impl = get_impl();
842 [ + - + - : 2 : if (impl->stopping_.load() || !data || data->empty()) {
- + - + ]
843 : 0 : impl->stats_.record_failed_send();
844 : 0 : return false;
845 : : }
846 : 2 : std::shared_ptr<TcpServerSession> session;
847 : : {
848 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
849 : 2 : session = impl->current_session_;
850 : 2 : }
851 : :
852 [ + + + - : 2 : if (session && session->alive()) {
+ - + + ]
853 : 1 : return session->async_try_write_shared(std::move(data));
854 : : }
855 : 1 : impl->stats_.record_failed_send();
856 : 1 : return false;
857 : 2 : }
858 : :
859 : : // Each setter builds the shared snapshot before taking the lock, so the
860 : : // allocation stays outside the critical section the io thread contends on.
861 : 122 : void TcpServer::on_bytes(OnBytes cb) {
862 : 122 : auto shared = interface::share_callback(std::move(cb));
863 : 122 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
864 : 122 : impl_->on_bytes_ = std::move(shared);
865 : 122 : }
866 : 247 : void TcpServer::on_state(OnState cb) {
867 : 247 : auto shared = interface::share_callback(std::move(cb));
868 : 247 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
869 : 247 : impl_->on_state_ = std::move(shared);
870 : 247 : }
871 : 241 : void TcpServer::on_backpressure(OnBackpressure cb) {
872 : 241 : auto impl = get_impl();
873 : 241 : auto shared = interface::share_callback(std::move(cb));
874 : 241 : std::shared_ptr<TcpServerSession> session;
875 : 241 : interface::SharedCallback<OnBackpressure> bp_cb;
876 : : {
877 : 241 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
878 : 241 : impl->on_bp_ = std::move(shared);
879 : 241 : bp_cb = impl->on_bp_;
880 : 241 : session = impl->current_session_;
881 : 241 : }
882 : :
883 [ + + - + : 241 : if (session) session->on_backpressure(bp_cb ? *bp_cb : OnBackpressure{});
- - + - ]
884 : 241 : }
885 : :
886 : 1167 : bool TcpServer::broadcast(std::string_view message) {
887 : 1167 : auto shared_data = std::make_shared<const std::vector<uint8_t>>(message.begin(), message.end());
888 : 1167 : auto impl = get_impl();
889 : 1167 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
890 : 1167 : bool sent = false;
891 : 1167 : bool attempted = false;
892 [ + + ]: 2485 : for (auto& entry : impl->sessions_) {
893 : 1318 : auto& session = entry.second;
894 [ + - + - : 1318 : if (session && session->alive()) {
+ - + - ]
895 : 1318 : attempted = true;
896 [ + - + + ]: 1318 : if (session->async_try_write_shared(shared_data)) sent = true;
897 : : }
898 : : }
899 [ + + ]: 1167 : if (!attempted) impl->stats_.record_failed_send();
900 : 1167 : return sent;
901 : 1167 : }
902 : :
903 : 2 : bool TcpServer::broadcast(memory::ConstByteSpan data) {
904 : 2 : auto shared_data = std::make_shared<const std::vector<uint8_t>>(data.begin(), data.end());
905 : 2 : auto impl = get_impl();
906 : 2 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
907 : 2 : bool sent = false;
908 : 2 : bool attempted = false;
909 [ + + ]: 3 : for (auto& entry : impl->sessions_) {
910 : 1 : auto& session = entry.second;
911 [ + - + - : 1 : if (session && session->alive()) {
+ - + - ]
912 : 1 : attempted = true;
913 [ + - + - ]: 1 : if (session->async_try_write_shared(shared_data)) sent = true;
914 : : }
915 : : }
916 [ + + ]: 2 : if (!attempted) impl->stats_.record_failed_send();
917 : 2 : return sent;
918 : 2 : }
919 : :
920 : 54 : bool TcpServer::send_to_client(ClientId client_id, std::string_view message) {
921 : 162 : return send_to_client(client_id,
922 : 162 : memory::ConstByteSpan(reinterpret_cast<const uint8_t*>(message.data()), message.size()));
923 : : }
924 : :
925 : 55 : bool TcpServer::send_to_client(ClientId client_id, memory::ConstByteSpan data) {
926 : 55 : auto impl = get_impl();
927 : 55 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
928 : 55 : auto it = impl->sessions_.find(client_id);
929 [ + + + - : 55 : if (it != impl->sessions_.end() && it->second && it->second->alive()) {
+ - + - +
+ ]
930 : 54 : return it->second->async_write_copy(data);
931 : : }
932 : 1 : impl->stats_.record_failed_send();
933 : 1 : return false;
934 : 55 : }
935 : :
936 : 5 : bool TcpServer::try_send_to_client(ClientId client_id, std::string_view message) {
937 : 15 : return try_send_to_client(client_id,
938 : 15 : memory::ConstByteSpan(reinterpret_cast<const uint8_t*>(message.data()), message.size()));
939 : : }
940 : :
941 : 6 : bool TcpServer::try_send_to_client(ClientId client_id, memory::ConstByteSpan data) {
942 : 6 : auto impl = get_impl();
943 : 6 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
944 : 6 : auto it = impl->sessions_.find(client_id);
945 [ + + + - : 6 : if (it != impl->sessions_.end() && it->second && it->second->alive()) {
+ - + - +
+ ]
946 : 4 : return it->second->async_try_write_copy(data);
947 : : }
948 : 2 : impl->stats_.record_failed_send();
949 : 2 : return false;
950 : 6 : }
951 : :
952 : 105 : size_t TcpServer::client_count() const {
953 : 105 : auto impl = get_impl();
954 : 105 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
955 : 105 : size_t alive = 0;
956 [ + + ]: 251 : for (const auto& entry : impl->sessions_)
957 [ + - + - : 146 : if (entry.second && entry.second->alive()) ++alive;
+ - + - ]
958 : 105 : return alive;
959 : 105 : }
960 : :
961 : 27 : std::optional<wrapper::RuntimeStats> TcpServer::client_stats(ClientId client_id) const {
962 : 27 : auto impl = get_impl();
963 : 27 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
964 : 27 : auto it = impl->sessions_.find(client_id);
965 [ + + - + : 27 : if (it == impl->sessions_.end() || !it->second) return std::nullopt;
+ + ]
966 : 24 : return it->second->stats();
967 : 27 : }
968 : :
969 : 6 : std::vector<ClientId> TcpServer::connected_clients() const {
970 : 6 : auto impl = get_impl();
971 : 6 : std::lock_guard<std::mutex> lock(impl->sessions_mutex_);
972 : 6 : std::vector<ClientId> connected_clients;
973 : 6 : connected_clients.reserve(impl->sessions_.size());
974 [ + + ]: 14 : for (const auto& entry : impl->sessions_)
975 [ + - + - : 8 : if (entry.second && entry.second->alive()) connected_clients.push_back(entry.first);
+ - + - +
- ]
976 : 12 : return connected_clients;
977 : 6 : }
978 : :
979 : 121 : void TcpServer::on_multi_connect(MultiClientConnectHandler h) {
980 : 121 : std::lock_guard<std::mutex> l(impl_->sessions_mutex_);
981 : 121 : impl_->on_multi_connect_ = std::move(h);
982 : 121 : }
983 : 120 : void TcpServer::on_multi_data(MultiClientDataHandler h) {
984 : 120 : auto shared = interface::share_callback(std::move(h));
985 : 120 : std::lock_guard<std::mutex> l(impl_->sessions_mutex_);
986 : 120 : impl_->on_multi_data_ = std::move(shared);
987 : 120 : }
988 : 120 : void TcpServer::on_multi_disconnect(MultiClientDisconnectHandler h) {
989 : 120 : std::lock_guard<std::mutex> l(impl_->sessions_mutex_);
990 : 120 : impl_->on_multi_disconnect_ = std::move(h);
991 : 120 : }
992 : :
993 : 8 : void TcpServer::set_client_limit(size_t max) {
994 : 8 : auto impl = get_impl();
995 [ - + ]: 8 : if (max > base::constants::MAX_MAX_CONNECTIONS) {
996 : 0 : max = base::constants::MAX_MAX_CONNECTIONS;
997 : : }
998 : 8 : std::lock_guard<std::mutex> l(impl->sessions_mutex_);
999 : 8 : impl->max_clients_ = max;
1000 [ - + ]: 8 : if (max == 0) {
1001 : 0 : impl->client_limit_enabled_ = false;
1002 : : } else {
1003 : 8 : impl->client_limit_enabled_ = true;
1004 : : }
1005 : 8 : }
1006 : :
1007 : 21 : base::LinkState TcpServer::state() const { return get_impl()->state_.get(); }
1008 : :
1009 : : } // namespace transport
1010 : : } // namespace wirestead
|
