Branch data Line data Source code
1 : : /*
2 : : * Copyright 2025 Jinwoo Sung
3 : : *
4 : : * Licensed under the Apache License, Version 2.0 (the "License");
5 : : * you may not use this file except in compliance with the License.
6 : : * You may obtain a copy of the License at
7 : : *
8 : : * http://www.apache.org/licenses/LICENSE-2.0
9 : : *
10 : : * Unless required by applicable law or agreed to in writing, software
11 : : * distributed under the License is distributed on an "AS IS" BASIS,
12 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 : : * See the License for the specific language governing permissions and
14 : : * limitations under the License.
15 : : */
16 : :
17 : : #include "wirestead/wrapper/tcp_server/tcp_server.hpp"
18 : :
19 : : #include <atomic>
20 : : #include <boost/asio/executor_work_guard.hpp>
21 : : #include <boost/asio/io_context.hpp>
22 : : #include <boost/asio/steady_timer.hpp>
23 : : #include <chrono>
24 : : #include <mutex>
25 : : #include <shared_mutex>
26 : : #include <stdexcept>
27 : : #include <stop_token>
28 : : #include <thread>
29 : : #include <unordered_map>
30 : : #include <vector>
31 : :
32 : : #include "wirestead/base/common.hpp"
33 : : #include "wirestead/concurrency/io_thread_hook.hpp"
34 : : #include "wirestead/config/tcp_server_config.hpp"
35 : : #include "wirestead/factory/channel_factory.hpp"
36 : : #include "wirestead/transport/tcp_server/tcp_server.hpp"
37 : : #include "wirestead/wrapper/callback_guard.hpp"
38 : : #include "wirestead/wrapper/error_context_builder.hpp"
39 : :
40 : : namespace wirestead {
41 : : namespace wrapper {
42 : :
43 : : struct TcpServer::Impl : public std::enable_shared_from_this<Impl> {
44 : : mutable std::shared_mutex mutex_;
45 : : std::mutex bp_mutex_;
46 : : std::condition_variable bp_cv_;
47 : : uint16_t port_;
48 : : std::string bind_address_{"0.0.0.0"};
49 : : std::string tls_certificate_file_;
50 : : std::string tls_private_key_file_;
51 : : std::shared_ptr<interface::Channel> channel_;
52 : : std::shared_ptr<boost::asio::io_context> external_ioc_;
53 : : std::atomic<bool> use_external_context_{false};
54 : : std::atomic<bool> manage_external_context_{false};
55 : : std::jthread external_thread_;
56 : : std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>> work_guard_;
57 : :
58 : : std::vector<std::promise<bool>> pending_promises_;
59 : : std::atomic<bool> started_{false};
60 : : std::atomic<bool> is_listening_{false};
61 : 130 : std::shared_ptr<bool> alive_marker_{std::make_shared<bool>(true)};
62 : :
63 : : // Configuration
64 : : std::atomic<bool> auto_start_{false};
65 : : std::atomic<bool> shared_context_{false};
66 : : std::atomic<bool> port_retry_enabled_{false};
67 : : std::atomic<int> max_port_retries_{3};
68 : : std::atomic<int> port_retry_interval_ms_{1000};
69 : : std::atomic<int> idle_timeout_ms_{static_cast<int>(base::constants::DEFAULT_IDLE_TIMEOUT_MS)};
70 : : std::atomic<bool> client_limit_enabled_{false};
71 : : std::atomic<size_t> max_clients_{0};
72 : : std::atomic<size_t> backpressure_threshold_{base::constants::DEFAULT_BACKPRESSURE_THRESHOLD};
73 : : std::atomic<base::constants::BackpressureStrategy> backpressure_strategy_{
74 : : base::constants::BackpressureStrategy::Reliable};
75 : : std::atomic<bool> tcp_no_delay_{base::constants::DEFAULT_TCP_NO_DELAY};
76 : : std::atomic<bool> keep_alive_{base::constants::DEFAULT_KEEP_ALIVE};
77 : : std::atomic<size_t> send_buffer_size_{0};
78 : : std::atomic<size_t> receive_buffer_size_{0};
79 : : std::atomic<size_t> read_buffer_size_{base::constants::DEFAULT_READ_BUFFER_SIZE};
80 : :
81 : : ConnectionHandler on_client_connect_{nullptr};
82 : : ConnectionHandler on_disconnect_{nullptr};
83 : : // Shared snapshots: the io thread copies one out per received chunk, and a
84 : : // std::function copy allocates whenever the user handler outgrows its
85 : : // small-object buffer. See interface::SharedCallback.
86 : : interface::SharedCallback<MessageHandler> on_data_;
87 : : interface::SharedCallback<BatchMessageHandler> data_batch_handler_;
88 : : ErrorHandler on_error_{nullptr};
89 : : std::function<void(size_t)> on_backpressure_{nullptr};
90 : : FramerFactory framer_factory_{nullptr};
91 : : interface::SharedCallback<MessageHandler> on_message_;
92 : : interface::SharedCallback<BatchMessageHandler> message_batch_handler_;
93 : :
94 : : // Batching logic
95 : : std::vector<MessageContext> data_batch_queue_;
96 : : std::vector<MessageContext> message_batch_queue_;
97 : : std::unique_ptr<boost::asio::steady_timer> batch_timer_;
98 : : size_t max_batch_size_ = 100;
99 : 130 : std::chrono::milliseconds max_batch_latency_{1};
100 : :
101 : : std::unordered_map<ClientId, std::shared_ptr<framer::IFramer>> framers_;
102 : : // Cached transport pointer — set once in start(), avoids repeated dynamic_cast.
103 : : std::shared_ptr<transport::TcpServer> transport_cache_;
104 : :
105 : 124 : explicit Impl(uint16_t port)
106 : 248 : : port_(port),
107 : 124 : started_(false),
108 : 124 : is_listening_(false),
109 : 124 : auto_start_(false),
110 : 124 : port_retry_enabled_(false),
111 : 124 : max_port_retries_(3),
112 : 124 : port_retry_interval_ms_(1000),
113 : 124 : idle_timeout_ms_(static_cast<int>(base::constants::DEFAULT_IDLE_TIMEOUT_MS)),
114 : 124 : client_limit_enabled_(false),
115 : 124 : max_clients_(0),
116 : 124 : backpressure_threshold_(base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
117 : 744 : backpressure_strategy_(base::constants::BackpressureStrategy::Reliable) {}
118 : :
119 : 5 : Impl(uint16_t port, std::shared_ptr<boost::asio::io_context> external_ioc)
120 : 10 : : port_(port),
121 : 5 : external_ioc_(std::move(external_ioc)),
122 : 5 : use_external_context_(external_ioc_ != nullptr),
123 : 5 : manage_external_context_(false),
124 : 5 : started_(false),
125 : 5 : is_listening_(false),
126 : 5 : auto_start_(false),
127 : 5 : port_retry_enabled_(false),
128 : 5 : max_port_retries_(3),
129 : 5 : port_retry_interval_ms_(1000),
130 : 5 : idle_timeout_ms_(static_cast<int>(base::constants::DEFAULT_IDLE_TIMEOUT_MS)),
131 : 5 : client_limit_enabled_(false),
132 : 5 : max_clients_(0),
133 : 5 : backpressure_threshold_(base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
134 : 35 : backpressure_strategy_(base::constants::BackpressureStrategy::Reliable) {}
135 : :
136 : 1 : explicit Impl(std::shared_ptr<interface::Channel> channel)
137 : 2 : : port_(0),
138 : 1 : channel_(std::move(channel)),
139 : 1 : started_(false),
140 : 1 : is_listening_(false),
141 : 1 : auto_start_(false),
142 : 1 : port_retry_enabled_(false),
143 : 1 : max_port_retries_(3),
144 : 1 : port_retry_interval_ms_(1000),
145 : 1 : idle_timeout_ms_(static_cast<int>(base::constants::DEFAULT_IDLE_TIMEOUT_MS)),
146 : 1 : client_limit_enabled_(false),
147 : 1 : max_clients_(0),
148 : 1 : backpressure_threshold_(base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
149 : 7 : backpressure_strategy_(base::constants::BackpressureStrategy::Reliable) {
150 : 1 : transport_cache_ = std::dynamic_pointer_cast<transport::TcpServer>(channel_);
151 : : // #450: setup_internal_handlers() captures weak_from_this() - calling it
152 : : // from inside this constructor would capture an empty weak_ptr, since
153 : : // enable_shared_from_this isn't wired up until make_shared() finishes
154 : : // constructing the object. Deferred to TcpServer's own constructor,
155 : : // which runs after impl_ is a fully-formed shared_ptr<Impl>.
156 : 1 : }
157 : :
158 : 130 : ~Impl() {
159 : : try {
160 : 130 : stop();
161 : 0 : } catch (...) {
162 : 0 : }
163 : 130 : }
164 : :
165 : 454 : void fulfill_all_locked(bool value) {
166 [ + + ]: 578 : for (auto& p : pending_promises_) {
167 : : try {
168 : 124 : p.set_value(value);
169 : 0 : } catch (...) {
170 : 0 : }
171 : : }
172 : 454 : pending_promises_.clear();
173 : 454 : }
174 : :
175 : 2 : void flush_batches() {
176 : 2 : std::unique_lock<std::shared_mutex> lock(mutex_);
177 [ + - ]: 2 : if (!data_batch_queue_.empty()) {
178 : 2 : auto handler = data_batch_handler_;
179 : 2 : auto batch = std::move(data_batch_queue_);
180 : 2 : data_batch_queue_.clear();
181 [ + - ]: 2 : if (handler) {
182 : 2 : lock.unlock();
183 : 2 : detail::invoke_user_callback("tcp_server", "on_data_batch", handler, batch);
184 : 2 : lock.lock();
185 : : }
186 : 2 : }
187 [ + + ]: 2 : if (!message_batch_queue_.empty()) {
188 : 1 : auto handler = message_batch_handler_;
189 : 1 : auto batch = std::move(message_batch_queue_);
190 : 1 : message_batch_queue_.clear();
191 [ + - ]: 1 : if (handler) {
192 : 1 : lock.unlock();
193 : 1 : detail::invoke_user_callback("tcp_server", "on_message_batch", handler, batch);
194 : 1 : lock.lock();
195 : : }
196 : 1 : }
197 [ + - ]: 2 : if (batch_timer_) {
198 : 2 : batch_timer_->cancel();
199 : : }
200 : 2 : }
201 : :
202 : 4 : void schedule_batch_timer() {
203 [ - + ]: 4 : if (!batch_timer_) return;
204 : 4 : batch_timer_->expires_after(max_batch_latency_);
205 : 8 : batch_timer_->async_wait([this, weak_impl = weak_from_this(),
206 : 4 : weak_alive = std::weak_ptr<bool>(alive_marker_)](const boost::system::error_code& ec) {
207 [ + + ]: 4 : if (ec) return;
208 : 2 : auto impl_keepalive = weak_impl.lock();
209 [ - + ]: 2 : if (!impl_keepalive) return;
210 : 2 : auto alive = weak_alive.lock();
211 [ - + ]: 2 : if (!alive) return;
212 : 2 : flush_batches();
213 : 2 : });
214 : : }
215 : :
216 : 125 : std::future<bool> start() {
217 : 125 : std::unique_lock<std::shared_mutex> lock(mutex_);
218 [ + + ]: 125 : if (is_listening_.load()) {
219 : 1 : std::promise<bool> p;
220 : 1 : p.set_value(true);
221 : 1 : return p.get_future();
222 : 1 : }
223 : 124 : std::promise<bool> p;
224 : 124 : auto f = p.get_future();
225 : 124 : pending_promises_.push_back(std::move(p));
226 [ + + ]: 124 : if (started_.exchange(true)) return f;
227 : :
228 [ + + ]: 122 : if (!channel_) {
229 : 120 : config::TcpServerConfig config;
230 : 120 : config.bind_address = bind_address_;
231 : 120 : config.port = port_;
232 : 120 : config.enable_port_retry = port_retry_enabled_.load();
233 : 120 : config.max_port_retries = max_port_retries_.load();
234 : 120 : config.port_retry_interval_ms = port_retry_interval_ms_.load();
235 : 120 : config.idle_timeout_ms = idle_timeout_ms_.load();
236 : 120 : config.backpressure_threshold = backpressure_threshold_.load();
237 : 120 : config.backpressure_strategy = backpressure_strategy_.load();
238 : 120 : config.tcp_no_delay = tcp_no_delay_.load();
239 : 120 : config.keep_alive = keep_alive_.load();
240 : 120 : config.send_buffer_size = send_buffer_size_.load();
241 : 120 : config.receive_buffer_size = receive_buffer_size_.load();
242 : 120 : config.read_buffer_size = read_buffer_size_.load();
243 : 120 : config.use_shared_context = shared_context_.load();
244 : 120 : config.tls_certificate_file = tls_certificate_file_;
245 : 120 : config.tls_private_key_file = tls_private_key_file_;
246 : :
247 : 120 : channel_ = factory::ChannelFactory::create(config, external_ioc_);
248 : 120 : transport_cache_ = std::dynamic_pointer_cast<transport::TcpServer>(channel_);
249 : 120 : setup_internal_handlers();
250 : :
251 [ + + ]: 120 : if (client_limit_enabled_.load()) {
252 : 8 : auto transport_server = std::dynamic_pointer_cast<transport::TcpServer>(channel_);
253 [ + - ]: 8 : if (transport_server) {
254 : 16 : transport_server->set_client_limit(max_clients_.load());
255 : : }
256 : 8 : }
257 : 120 : }
258 : : // #506: take a local copy of the shared_ptr before unlocking. Calling
259 : : // through the raw channel_ member here would race a concurrent stop()'s
260 : : // channel_.reset() on the member itself (not just the pointee) - TSAN
261 : : // caught exactly this under TcpServerWrapperLifecycleTest.
262 : : // ConcurrentStartStop. A local copy holds its own reference, so it stays
263 : : // valid and race-free even if another thread resets the member.
264 : 122 : auto channel_copy = channel_;
265 : 122 : lock.unlock();
266 : 122 : channel_copy->start();
267 [ + + + + : 122 : if (use_external_context_.load() && manage_external_context_.load() && !external_thread_.joinable()) {
+ - + + ]
268 [ + - + - : 2 : if (external_ioc_ && external_ioc_->stopped()) {
+ + + + ]
269 : 1 : external_ioc_->restart();
270 : : }
271 : 4 : work_guard_ = std::make_unique<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>(
272 : 6 : boost::asio::make_work_guard(*external_ioc_));
273 : 4 : external_thread_ = std::jthread([ioc = external_ioc_](std::stop_token st) {
274 : 2 : wirestead::concurrency::run_io_thread_init();
275 : : try {
276 : 4 : std::stop_callback cb(st, [ioc] { ioc->stop(); });
277 : 2 : ioc->run();
278 : 2 : } catch (...) {
279 : 0 : }
280 : 4 : });
281 : : }
282 : 122 : return f;
283 : 125 : }
284 : :
285 : 258 : void stop() {
286 : 258 : bool should_join = false;
287 : : {
288 : 258 : std::unique_lock<std::shared_mutex> lock(mutex_);
289 [ + + ]: 258 : if (!started_.exchange(false)) {
290 : 136 : is_listening_.store(false);
291 : 136 : fulfill_all_locked(false);
292 : 136 : return;
293 : : }
294 : 122 : bp_cv_.notify_all();
295 [ + + ]: 122 : if (batch_timer_) {
296 : 121 : batch_timer_->cancel();
297 : 121 : batch_timer_.reset();
298 : : }
299 [ + + ]: 122 : if (channel_) {
300 : 121 : channel_->on_bytes(nullptr);
301 : 121 : channel_->on_state(nullptr);
302 : 121 : channel_->on_backpressure(nullptr);
303 : 121 : auto transport_server = std::dynamic_pointer_cast<transport::TcpServer>(channel_);
304 [ + + + - ]: 121 : if (transport_server) transport_server->request_stop();
305 : : // #506: same rationale as start() above - copy before unlocking so
306 : : // this call can't race a concurrent channel_.reset() on the member.
307 : 121 : auto channel_copy = channel_;
308 : 121 : lock.unlock();
309 : 121 : channel_copy->stop();
310 : 121 : lock.lock();
311 : 121 : }
312 [ + + + + : 122 : if (use_external_context_.load() && manage_external_context_.load()) {
+ + ]
313 [ + - ]: 2 : if (work_guard_) work_guard_.reset();
314 [ + - + - ]: 2 : if (external_ioc_) external_ioc_->stop();
315 : 2 : should_join = true;
316 : : }
317 : : // #444: transport_cache_ is a separate cached shared_ptr to the same
318 : : // transport object channel_ points at - without this, send_to()/
319 : : // broadcast()/max_clients() could still reach the stopped transport
320 : : // via transport_cache_ even after channel_.reset() below. framers_
321 : : // must also be cleared so a restart doesn't resume per-client framing
322 : : // state from stale ClientIds (mirrors UdsServer's existing behavior).
323 : 122 : transport_cache_.reset();
324 : 122 : framers_.clear();
325 : 122 : fulfill_all_locked(false);
326 : 122 : is_listening_.store(false);
327 : 258 : }
328 [ + + + - : 122 : if (should_join && external_thread_.joinable()) {
+ + ]
329 : : try {
330 [ + - ]: 2 : if (std::this_thread::get_id() != external_thread_.get_id()) {
331 : 2 : external_thread_.request_stop();
332 : 2 : external_thread_.join();
333 : : } else {
334 : 0 : external_thread_.detach();
335 : : }
336 : 0 : } catch (...) {
337 : 0 : }
338 : : }
339 : 122 : std::unique_lock<std::shared_mutex> lock(mutex_);
340 : 122 : channel_.reset();
341 : 122 : }
342 : :
343 : 4 : bool try_send_to(ClientId client_id, std::string_view data) {
344 : 4 : std::shared_lock<std::shared_mutex> lock(mutex_);
345 : 4 : const auto& ts = transport_cache_;
346 [ + + + - ]: 8 : return ts ? ts->try_send_to_client(client_id, data) : false;
347 : 4 : }
348 : :
349 : 1169 : bool try_broadcast(std::string_view data) {
350 : 1169 : std::shared_lock<std::shared_mutex> lock(mutex_);
351 : 1169 : const auto& ts = transport_cache_;
352 [ + + + - ]: 2338 : return ts ? ts->broadcast(data) : false;
353 : 1169 : }
354 : :
355 : 5 : bool send_to(ClientId client_id, std::string_view data) {
356 [ + + ]: 5 : if (backpressure_strategy_.load() == base::constants::BackpressureStrategy::Reliable)
357 : 4 : return send_to_blocking(client_id, data);
358 : 1 : return try_send_to(client_id, data);
359 : : }
360 : :
361 : 1165 : bool broadcast(std::string_view data) {
362 : : // In order to avoid Head-Of-Line blocking where a single slow client
363 : : // blocks the entire broadcast loop, we delegate to try_broadcast (async fan-out)
364 : : // even in Reliable mode. Transport-level backpressure will still protect the queues.
365 : 1165 : return try_broadcast(data);
366 : : }
367 : :
368 : : // channel_->on_backpressure()/session-level backpressure callbacks call bp_cv_.notify_all()
369 : : // from the transport's io_context thread without holding bp_mutex_ - a classic lost-wakeup
370 : : // race is possible: a waiter can check the predicate, find it still blocking, and be in the
371 : : // process of registering to wait when the notify fires. Poll with a bounded timeout instead
372 : : // of an unbounded wait() so a missed notify only costs a short delay rather than a
373 : : // permanent hang (see #427, #431).
374 : : //
375 : : // Returns false without sending instead of waiting if called from the
376 : : // channel's own io thread while backpressure is active for this client -
377 : : // e.g. a blocking send_to() called from inside an on_data/on_message
378 : : // callback. Clearing backpressure requires that same io thread to make
379 : : // progress, so blocking here would deadlock forever rather than
380 : : // eventually clear (#449).
381 : : // #509: the wait predicate (is_backpressure_active(client_id), tied to
382 : : // bp_high) and the transport's own hard queue-byte cap for that session
383 : : // (bp_limit) are different thresholds observed at different times, so a
384 : : // single write attempt can spuriously fail right after the wait exits.
385 : : // Bounded retry rather than unbounded, so a payload that can never fit
386 : : // still fails in bounded time.
387 : : static constexpr int kMaxBlockingSendAttempts = 5;
388 : :
389 : 5 : bool send_to_blocking(ClientId client_id, std::string_view data) {
390 [ + - ]: 5 : for (int attempt = 0; attempt < kMaxBlockingSendAttempts; ++attempt) {
391 : 5 : std::unique_lock<std::mutex> lock(bp_mutex_);
392 : 10 : auto predicate = [this, client_id]() {
393 : 10 : std::shared_lock<std::shared_mutex> rlock(mutex_);
394 : 10 : const auto& ts = transport_cache_;
395 [ + - + + : 20 : return !started_.load() || !ts || !ts->is_backpressure_active(client_id);
+ - + - ]
396 : 10 : };
397 [ + - - + : 5 : if (!predicate() && detail::in_data_callback()) return false;
- - - + ]
398 [ + - - + ]: 5 : while (!bp_cv_.wait_for(lock, std::chrono::milliseconds(50), predicate)) {
399 : : }
400 : 5 : lock.unlock();
401 : 5 : std::shared_lock<std::shared_mutex> rlock(mutex_);
402 : 5 : const auto& ts = transport_cache_;
403 [ + + ]: 5 : if (!ts) return false;
404 [ + - + - ]: 2 : if (ts->send_to_client(client_id, data)) return true;
405 : 10 : }
406 : 0 : return false;
407 : : }
408 : :
409 : 121 : void setup_internal_handlers() {
410 [ - + ]: 121 : if (!channel_) return;
411 : :
412 : 121 : batch_timer_ = std::make_unique<boost::asio::steady_timer>(channel_->get_executor());
413 : :
414 : 121 : std::weak_ptr<bool> weak_alive = alive_marker_;
415 : 121 : std::weak_ptr<Impl> weak_impl = weak_from_this();
416 : 121 : auto transport_server = std::dynamic_pointer_cast<transport::TcpServer>(channel_);
417 [ + + ]: 121 : if (transport_server) {
418 : 120 : transport_server->on_multi_connect([this, weak_impl, weak_alive](ClientId id, const std::string& info) {
419 : 130 : auto impl_keepalive = weak_impl.lock();
420 [ - + ]: 130 : if (!impl_keepalive) return;
421 : 130 : auto alive = weak_alive.lock();
422 [ - + ]: 130 : if (!alive) return;
423 : :
424 : 130 : ConnectionHandler handler;
425 : : {
426 : 130 : std::unique_lock<std::shared_mutex> lock(mutex_);
427 [ + + ]: 130 : if (framer_factory_) {
428 : 22 : auto framer = framer_factory_();
429 [ + - ]: 22 : if (framer) {
430 : 22 : auto shared_framer = std::shared_ptr<framer::IFramer>(std::move(framer));
431 : 22 : shared_framer->on_message([this, id](memory::ConstByteSpan msg) {
432 : : // #441: snapshot under a shared_lock (pure read), build the
433 : : // copy before taking the exclusive lock for queue mutation.
434 : : bool batch_mode;
435 : 1003 : interface::SharedCallback<MessageHandler> on_message_handler;
436 : : {
437 : 1003 : std::shared_lock<std::shared_mutex> lock(mutex_);
438 : 1003 : batch_mode = static_cast<bool>(message_batch_handler_);
439 : 1003 : on_message_handler = on_message_;
440 : 1003 : }
441 : :
442 [ + + ]: 1003 : if (batch_mode) {
443 : 3 : MessageContext ctx(id, memory::SafeDataBuffer(msg));
444 : 3 : interface::SharedCallback<BatchMessageHandler> flush_handler;
445 : 3 : std::vector<MessageContext> batch;
446 : : {
447 : 3 : std::unique_lock<std::shared_mutex> lock(mutex_);
448 : 3 : message_batch_queue_.emplace_back(std::move(ctx));
449 [ + + ]: 3 : if (message_batch_queue_.size() >= max_batch_size_) {
450 : 1 : flush_handler = message_batch_handler_;
451 : 1 : batch = std::move(message_batch_queue_);
452 : 1 : message_batch_queue_.clear();
453 [ + - ]: 2 : } else if (message_batch_queue_.size() == 1) {
454 : 2 : schedule_batch_timer();
455 : : }
456 : 3 : }
457 : 3 : detail::invoke_user_callback("tcp_server", "on_message_batch", flush_handler, batch);
458 : 3 : return;
459 : 3 : }
460 : :
461 : 1000 : detail::invoke_user_callback("tcp_server", "on_message", on_message_handler, MessageContext(id, msg));
462 : 1003 : });
463 : 22 : framers_[id] = std::move(shared_framer);
464 : 22 : }
465 : 22 : }
466 : 130 : handler = on_client_connect_;
467 : 130 : }
468 : 130 : detail::invoke_user_callback("tcp_server", "on_connect", handler, ConnectionContext(id, info));
469 : 130 : });
470 : 120 : transport_server->on_multi_data([this, weak_impl, weak_alive](ClientId id, memory::ConstByteSpan data_span) {
471 : 2466 : auto impl_keepalive = weak_impl.lock();
472 [ - + ]: 2466 : if (!impl_keepalive) return;
473 : 2466 : auto alive = weak_alive.lock();
474 [ - + ]: 2466 : if (!alive) return;
475 : :
476 : : // #449: everything below runs synchronously on this io thread -
477 : : // mark it so a blocking send_to()/broadcast() called from within
478 : : // one of these callbacks fails fast instead of deadlocking.
479 : 2466 : detail::CallbackGuard callback_guard;
480 : :
481 : : // #441: snapshot the handler/framer pointers under a shared_lock
482 : : // (not unique_lock) - this is a pure read, matching try_send's
483 : : // locking level so it no longer blocks concurrent sends even
484 : : // briefly.
485 : : bool batch_mode;
486 : 2466 : interface::SharedCallback<MessageHandler> handler;
487 : 2466 : std::shared_ptr<framer::IFramer> framer;
488 : : {
489 : 2466 : std::shared_lock<std::shared_mutex> lock(mutex_);
490 : 2466 : batch_mode = static_cast<bool>(data_batch_handler_);
491 : 2466 : handler = on_data_;
492 : 2466 : auto it = framers_.find(id);
493 [ + + ]: 2466 : if (it != framers_.end()) {
494 : 1002 : framer = it->second;
495 : : }
496 : 2466 : }
497 : :
498 [ + + ]: 2466 : if (batch_mode) {
499 : : // #441: build the copy before taking the exclusive lock, so the
500 : : // lock is only held for the queue mutation itself, not the
501 : : // allocation.
502 : 2 : MessageContext ctx(id, memory::SafeDataBuffer(data_span));
503 : 2 : interface::SharedCallback<BatchMessageHandler> flush_handler;
504 : 2 : std::vector<MessageContext> batch;
505 : : {
506 : 2 : std::unique_lock<std::shared_mutex> lock(mutex_);
507 : 2 : data_batch_queue_.emplace_back(std::move(ctx));
508 [ - + ]: 2 : if (data_batch_queue_.size() >= max_batch_size_) {
509 : 0 : flush_handler = data_batch_handler_;
510 : 0 : batch = std::move(data_batch_queue_);
511 : 0 : data_batch_queue_.clear();
512 [ + - ]: 2 : } else if (data_batch_queue_.size() == 1) {
513 : 2 : schedule_batch_timer();
514 : : }
515 : 2 : }
516 : 2 : detail::invoke_user_callback("tcp_server", "on_data_batch", flush_handler, batch);
517 : 2 : } else {
518 : 4928 : detail::invoke_user_callback("tcp_server", "on_data", handler, MessageContext(id, data_span));
519 : : }
520 : :
521 [ + + + - ]: 2466 : if (framer) framer->push_bytes(data_span);
522 : 2466 : });
523 : 120 : transport_server->on_multi_disconnect([this, weak_impl, weak_alive](ClientId id) {
524 : 118 : auto impl_keepalive = weak_impl.lock();
525 [ - + ]: 118 : if (!impl_keepalive) return;
526 : 118 : auto alive = weak_alive.lock();
527 [ - + ]: 118 : if (!alive) return;
528 : :
529 : 118 : ConnectionHandler handler;
530 : : {
531 : 118 : std::unique_lock<std::shared_mutex> lock(mutex_);
532 : 118 : framers_.erase(id);
533 : 118 : handler = on_disconnect_;
534 : 118 : }
535 : 118 : detail::invoke_user_callback("tcp_server", "on_disconnect", handler, ConnectionContext(id));
536 : 118 : });
537 : :
538 : 120 : transport_server->on_backpressure([this, weak_impl, weak_alive](size_t queued) {
539 : 64 : bp_cv_.notify_all();
540 : 64 : auto impl_keepalive = weak_impl.lock();
541 [ - + ]: 64 : if (!impl_keepalive) return;
542 : 64 : auto alive = weak_alive.lock();
543 [ - + ]: 64 : if (!alive) return;
544 : 64 : std::function<void(size_t)> handler;
545 : : {
546 : 64 : std::shared_lock<std::shared_mutex> lock(mutex_);
547 : 64 : handler = on_backpressure_;
548 : 64 : }
549 : 64 : detail::invoke_user_callback("tcp_server", "on_backpressure", handler, queued);
550 : 64 : });
551 : : }
552 : 121 : channel_->on_state([this, weak_impl, weak_alive](base::LinkState state) {
553 : 325 : auto impl_keepalive = weak_impl.lock();
554 [ - + ]: 325 : if (!impl_keepalive) return;
555 : 325 : auto alive = weak_alive.lock();
556 [ - + ]: 325 : if (!alive) return;
557 : :
558 [ + + ]: 325 : if (state == base::LinkState::Listening) {
559 : 192 : is_listening_.store(true);
560 : 192 : std::unique_lock<std::shared_mutex> lock(mutex_);
561 : 192 : fulfill_all_locked(true);
562 [ + + + - : 325 : } else if (state == base::LinkState::Error || state == base::LinkState::Closed ||
- + ]
563 : : state == base::LinkState::Idle) {
564 : 4 : ErrorHandler handler;
565 : 4 : is_listening_.store(false);
566 : : {
567 : 4 : std::unique_lock<std::shared_mutex> lock(mutex_);
568 : 4 : fulfill_all_locked(false);
569 [ + - ]: 4 : if (state == base::LinkState::Error) {
570 : 4 : handler = on_error_;
571 : : }
572 : 4 : }
573 [ + - ]: 4 : detail::invoke_user_callback("tcp_server", "on_error", handler,
574 [ + - + - : 8 : channel_ ? detail::build_error_context(*channel_, "Server error")
- - ]
575 : : : ErrorContext(ErrorCode::IoError, "Server error"));
576 : 4 : }
577 : 325 : });
578 : 121 : }
579 : :
580 : 1884 : RuntimeStats stats() const {
581 : 1884 : std::shared_lock<std::shared_mutex> lock(mutex_);
582 [ + - + - ]: 3768 : return channel_ ? channel_->stats() : RuntimeStats{};
583 : 1884 : }
584 : :
585 : 0 : void reset_stats() {
586 : 0 : std::shared_lock<std::shared_mutex> lock(mutex_);
587 [ # # # # ]: 0 : if (channel_) channel_->reset_stats();
588 : 0 : }
589 : : };
590 : :
591 : 124 : TcpServer::TcpServer(uint16_t port) : impl_(std::make_shared<Impl>(port)) {}
592 : 5 : TcpServer::TcpServer(uint16_t port, std::shared_ptr<boost::asio::io_context> ioc)
593 : 5 : : impl_(std::make_shared<Impl>(port, ioc)) {}
594 : 1 : TcpServer::TcpServer(std::shared_ptr<interface::Channel> ch) : impl_(std::make_shared<Impl>(ch)) {
595 : 1 : impl_->setup_internal_handlers();
596 : 1 : }
597 : 225 : TcpServer::~TcpServer() = default;
598 : :
599 : 0 : TcpServer::TcpServer(TcpServer&&) noexcept = default;
600 : 0 : TcpServer& TcpServer::operator=(TcpServer&&) noexcept = default;
601 : :
602 : 125 : std::future<bool> TcpServer::start() { return impl_->start(); }
603 : 128 : void TcpServer::stop() { impl_->stop(); }
604 : 54 : bool TcpServer::listening() const { return get_impl()->is_listening_.load(); }
605 : 1884 : RuntimeStats TcpServer::stats() const { return get_impl()->stats(); }
606 : 0 : void TcpServer::reset_stats() { impl_->reset_stats(); }
607 : :
608 : 1165 : bool TcpServer::broadcast(std::string_view data) { return impl_->broadcast(data); }
609 : 4 : bool TcpServer::try_broadcast(std::string_view data) { return impl_->try_broadcast(data); }
610 : 5 : bool TcpServer::send_to(ClientId client_id, std::string_view data) { return impl_->send_to(client_id, data); }
611 : 3 : bool TcpServer::try_send_to(ClientId client_id, std::string_view data) { return impl_->try_send_to(client_id, data); }
612 : :
613 : 1 : bool TcpServer::send_to_blocking(ClientId client_id, std::string_view data) {
614 : 1 : return impl_->send_to_blocking(client_id, data);
615 : : }
616 : :
617 : 6 : bool TcpServer::broadcast_line(std::string_view line) { return broadcast(std::string(line) + "\n"); }
618 : 2 : bool TcpServer::send_to_line(ClientId client_id, std::string_view line) {
619 : 6 : return send_to(client_id, std::string(line) + "\n");
620 : : }
621 : 6 : bool TcpServer::try_broadcast_line(std::string_view line) { return try_broadcast(std::string(line) + "\n"); }
622 : 2 : bool TcpServer::try_send_to_line(ClientId client_id, std::string_view line) {
623 : 6 : return try_send_to(client_id, std::string(line) + "\n");
624 : : }
625 : :
626 : 20 : TcpServer& TcpServer::on_connect(ConnectionHandler h) {
627 : 20 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
628 : 20 : impl_->on_client_connect_ = std::move(h);
629 : 20 : return *this;
630 : 20 : }
631 : 5 : TcpServer& TcpServer::on_disconnect(ConnectionHandler h) {
632 : 5 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
633 : 5 : impl_->on_disconnect_ = std::move(h);
634 : 5 : return *this;
635 : 5 : }
636 : 94 : TcpServer& TcpServer::on_data(MessageHandler h) {
637 : 94 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
638 : 94 : impl_->on_data_ = interface::share_callback(std::move(h));
639 : 94 : return *this;
640 : 94 : }
641 : 3 : TcpServer& TcpServer::on_data_batch(BatchMessageHandler h) {
642 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
643 : 3 : impl_->data_batch_handler_ = interface::share_callback(std::move(h));
644 : 3 : return *this;
645 : 3 : }
646 : 95 : TcpServer& TcpServer::on_error(ErrorHandler h) {
647 : 95 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
648 : 95 : impl_->on_error_ = std::move(h);
649 : 95 : return *this;
650 : 95 : }
651 : :
652 : 1 : TcpServer& TcpServer::on_backpressure(std::function<void(size_t)> h) {
653 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
654 : 1 : impl_->on_backpressure_ = std::move(h);
655 : 1 : return *this;
656 : 1 : }
657 : :
658 : 4 : TcpServer& TcpServer::framer(FramerFactory factory) {
659 : 4 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
660 : 4 : impl_->framer_factory_ = std::move(factory);
661 : 4 : return *this;
662 : 4 : }
663 : :
664 : 2 : TcpServer& TcpServer::on_message(MessageHandler handler) {
665 : 2 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
666 : 2 : impl_->on_message_ = interface::share_callback(std::move(handler));
667 : 2 : return *this;
668 : 2 : }
669 : :
670 : 3 : TcpServer& TcpServer::on_message_batch(BatchMessageHandler h) {
671 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
672 : 3 : impl_->message_batch_handler_ = interface::share_callback(std::move(h));
673 : 3 : return *this;
674 : 3 : }
675 : :
676 : 104 : size_t TcpServer::client_count() const {
677 : 104 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
678 : 104 : const auto& ts = get_impl()->transport_cache_;
679 [ + + + - ]: 208 : return ts ? ts->client_count() : 0;
680 : 104 : }
681 : :
682 : 5 : std::vector<ClientId> TcpServer::connected_clients() const {
683 : 5 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
684 : 5 : const auto& ts = get_impl()->transport_cache_;
685 [ + + + - ]: 10 : return ts ? ts->connected_clients() : std::vector<ClientId>();
686 : 5 : }
687 : :
688 : 27 : std::optional<RuntimeStats> TcpServer::client_stats(ClientId client_id) const {
689 : 27 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
690 : 27 : const auto& ts = get_impl()->transport_cache_;
691 [ + - + - ]: 54 : return ts ? ts->client_stats(client_id) : std::nullopt;
692 : 27 : }
693 : :
694 : 1 : TcpServer& TcpServer::auto_start(bool m) {
695 : 1 : impl_->auto_start_.store(m);
696 [ + - + - : 1 : if (impl_->auto_start_.load() && !impl_->started_.load()) start();
+ - + - ]
697 : 1 : return *this;
698 : : }
699 : :
700 : 1 : TcpServer& TcpServer::bind_address(const std::string& address) {
701 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
702 : 1 : impl_->bind_address_ = address;
703 : 1 : return *this;
704 : 1 : }
705 : :
706 : 2 : TcpServer& TcpServer::shared_context(bool use_shared) {
707 : 2 : impl_->shared_context_.store(use_shared);
708 : 2 : return *this;
709 : : }
710 : :
711 : 4 : TcpServer& TcpServer::port_retry(bool e, int m, int i) {
712 : 4 : impl_->port_retry_enabled_.store(e);
713 : 4 : impl_->max_port_retries_.store(m);
714 : 4 : impl_->port_retry_interval_ms_.store(i);
715 : 4 : return *this;
716 : : }
717 : :
718 : 2 : TcpServer& TcpServer::idle_timeout(std::chrono::milliseconds timeout) {
719 : 2 : impl_->idle_timeout_ms_.store(static_cast<int>(timeout.count()));
720 : 2 : return *this;
721 : : }
722 : :
723 : 15 : TcpServer& TcpServer::max_clients(size_t max) {
724 : 15 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
725 : 15 : impl_->max_clients_.store(max);
726 [ + + ]: 15 : if (max == 0) {
727 : 2 : impl_->client_limit_enabled_.store(false);
728 : : } else {
729 : 13 : impl_->client_limit_enabled_.store(true);
730 : : }
731 [ - + - - ]: 15 : if (impl_->transport_cache_) impl_->transport_cache_->set_client_limit(max);
732 : 15 : return *this;
733 : 15 : }
734 : :
735 : 100 : TcpServer& TcpServer::backpressure_threshold(size_t threshold) {
736 : 100 : impl_->backpressure_threshold_.store(threshold);
737 : 100 : return *this;
738 : : }
739 : :
740 : 5 : TcpServer& TcpServer::backpressure_strategy(base::constants::BackpressureStrategy strategy) {
741 : 5 : impl_->backpressure_strategy_.store(strategy);
742 : 5 : return *this;
743 : : }
744 : :
745 : 6 : size_t TcpServer::backpressure_threshold() const { return impl_->backpressure_threshold_.load(); }
746 : :
747 : 3 : base::constants::BackpressureStrategy TcpServer::backpressure_strategy() const {
748 : 3 : return impl_->backpressure_strategy_.load();
749 : : }
750 : :
751 : 1 : TcpServer& TcpServer::tcp_no_delay(bool enable) {
752 : 1 : impl_->tcp_no_delay_.store(enable);
753 : 1 : return *this;
754 : : }
755 : :
756 : 1 : TcpServer& TcpServer::keep_alive(bool enable) {
757 : 1 : impl_->keep_alive_.store(enable);
758 : 1 : return *this;
759 : : }
760 : :
761 : 6 : TcpServer& TcpServer::tls(const std::string& certificate_file, const std::string& private_key_file) {
762 : 6 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
763 : 6 : impl_->tls_certificate_file_ = certificate_file;
764 : 6 : impl_->tls_private_key_file_ = private_key_file;
765 : 6 : return *this;
766 : 6 : }
767 : :
768 : 3 : TcpServer& TcpServer::send_buffer_size(size_t bytes) {
769 : 3 : impl_->send_buffer_size_.store(bytes);
770 : 3 : return *this;
771 : : }
772 : :
773 : 1 : TcpServer& TcpServer::receive_buffer_size(size_t bytes) {
774 : 1 : impl_->receive_buffer_size_.store(bytes);
775 : 1 : return *this;
776 : : }
777 : :
778 : 3 : TcpServer& TcpServer::read_buffer_size(size_t bytes) {
779 : 3 : impl_->read_buffer_size_.store(bytes);
780 : 3 : return *this;
781 : : }
782 : :
783 : 3 : TcpServer& TcpServer::manage_external_context(bool m) {
784 : 3 : impl_->manage_external_context_.store(m);
785 : 3 : return *this;
786 : : }
787 : :
788 : 3 : TcpServer& TcpServer::batch_size(size_t size) {
789 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
790 : 3 : impl_->max_batch_size_ = size;
791 : 3 : return *this;
792 : 3 : }
793 : :
794 : 3 : TcpServer& TcpServer::batch_latency(std::chrono::milliseconds latency) {
795 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
796 : 3 : impl_->max_batch_latency_ = latency;
797 : 3 : return *this;
798 : 3 : }
799 : :
800 : : } // namespace wrapper
801 : : } // namespace wirestead
|
