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_session.hpp"
18 : :
19 : : #include <cstring>
20 : : #include <iostream>
21 : :
22 : : #include "wirestead/memory/memory_pool.hpp"
23 : : #include "wirestead/transport/base/bp_utils.hpp"
24 : : #include "wirestead/transport/tcp_server/boost_tcp_socket.hpp"
25 : :
26 : : namespace wirestead {
27 : : namespace transport {
28 : :
29 : 110 : TcpServerSession::TcpServerSession(net::io_context& ioc, tcp::socket sock, size_t backpressure_threshold,
30 : : int idle_timeout_ms, base::constants::BackpressureStrategy strategy,
31 : 110 : bool enable_memory_pool, size_t read_buffer_size)
32 : 110 : : ioc_(ioc),
33 : 110 : strand_(ioc.get_executor()),
34 : 110 : idle_timer_(ioc),
35 : 110 : socket_(std::make_unique<BoostTcpSocket>(std::move(sock))),
36 : 110 : enable_memory_pool_(enable_memory_pool),
37 : 110 : writing_(false),
38 : 110 : queue_bytes_(0),
39 : 110 : bp_strategy_(strategy),
40 : 110 : bp_high_(backpressure_threshold),
41 : 110 : idle_timeout_ms_(idle_timeout_ms),
42 : 110 : alive_(false),
43 : 440 : cleanup_done_(false) {
44 : 110 : bp_limit_ = std::min(std::max(bp_high_ * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
45 : : base::constants::MAX_BUFFER_SIZE);
46 [ + - ]: 110 : bp_low_ = bp_high_ > 1 ? bp_high_ / 2 : bp_high_;
47 [ - + ]: 110 : if (bp_low_ == 0) bp_low_ = 1;
48 : 110 : rx_.resize(
49 : 110 : std::clamp(read_buffer_size, base::constants::MIN_READ_BUFFER_SIZE, base::constants::MAX_READ_BUFFER_SIZE));
50 : 110 : }
51 : :
52 : 54 : TcpServerSession::TcpServerSession(net::io_context& ioc, std::unique_ptr<interface::TcpSocketInterface> socket,
53 : : size_t backpressure_threshold, int idle_timeout_ms,
54 : : base::constants::BackpressureStrategy strategy, bool enable_memory_pool,
55 : 54 : size_t read_buffer_size)
56 : 54 : : ioc_(ioc),
57 : 54 : strand_(ioc.get_executor()),
58 : 54 : idle_timer_(ioc),
59 : 54 : socket_(std::move(socket)),
60 : 54 : enable_memory_pool_(enable_memory_pool),
61 : 54 : writing_(false),
62 : 54 : queue_bytes_(0),
63 : 54 : bp_strategy_(strategy),
64 : 54 : bp_high_(backpressure_threshold),
65 : 54 : idle_timeout_ms_(idle_timeout_ms),
66 : 54 : alive_(false),
67 : 216 : cleanup_done_(false) {
68 : 54 : bp_limit_ = std::min(std::max(bp_high_ * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
69 : : base::constants::MAX_BUFFER_SIZE);
70 [ + - ]: 54 : bp_low_ = bp_high_ > 1 ? bp_high_ / 2 : bp_high_;
71 [ - + ]: 54 : if (bp_low_ == 0) bp_low_ = 1;
72 : 54 : rx_.resize(
73 : 54 : std::clamp(read_buffer_size, base::constants::MIN_READ_BUFFER_SIZE, base::constants::MAX_READ_BUFFER_SIZE));
74 : 54 : }
75 : :
76 : 164 : void TcpServerSession::start() {
77 [ - + ]: 164 : if (alive_.exchange(true)) return;
78 : 164 : auto self = shared_from_this();
79 : 164 : net::dispatch(strand_, [self] {
80 : 164 : self->reset_idle_timer();
81 : : // No-op on a plain socket, the TLS handshake on an encrypted one. Reading
82 : : // before it completes would hand the session ciphertext, so the first read
83 : : // waits on it - and a failed handshake closes rather than reads.
84 : 164 : self->socket_->async_handshake(net::bind_executor(self->strand_, [self](const boost::system::error_code& ec) {
85 [ + - - + : 164 : if (self->closing_ || !self->alive_) return;
- + ]
86 [ + + ]: 164 : if (ec) {
87 : 30 : WIRESTEAD_LOG_WARNING("tcp_server_session", "handshake", "Handshake failed: " + ec.message());
88 : 30 : self->do_close();
89 : 30 : return;
90 : : }
91 : 134 : self->reset_idle_timer();
92 : 134 : self->start_read();
93 : : }));
94 : 164 : });
95 : 164 : }
96 : :
97 : 60 : bool TcpServerSession::async_write_copy(memory::ConstByteSpan data) {
98 [ + - - + : 60 : if (!alive_ || closing_) {
- + ]
99 : 0 : stats_.record_failed_send();
100 : 0 : return false;
101 : : } // Don't queue writes if session is not alive
102 : :
103 : 60 : size_t size = data.size();
104 [ - + ]: 60 : if (size == 0) {
105 : 0 : stats_.record_failed_send();
106 : 0 : return false;
107 : : }
108 [ - + ]: 60 : if (size > base::constants::MAX_BUFFER_SIZE) {
109 : 0 : WIRESTEAD_LOG_ERROR("tcp_server_session", "write", "Write size exceeds maximum allowed");
110 : 0 : stats_.record_failed_send();
111 : 0 : return false;
112 : : }
113 : :
114 : : // Use memory pool for better performance (only for reasonable sizes)
115 [ + + + - ]: 60 : if (size <= base::constants::LARGE_BUFFER_THRESHOLD && enable_memory_pool_) { // Only use pool for buffers <= 64KB
116 : 7 : memory::PooledBuffer pooled_buffer(size, pool_);
117 [ + - + - ]: 7 : if (pooled_buffer.valid()) {
118 : : // Copy data to pooled buffer safely
119 : 7 : base::safe_memory::safe_memcpy(pooled_buffer.data(), data.data(), size);
120 [ + - - + ]: 7 : if (!queue_util::try_reserve_limit_bytes(write_reserve_mtx_, queue_bytes_, pending_bytes_, inflight_bytes_, size,
121 : : bp_limit_)) {
122 : 0 : stats_.record_failed_send();
123 : 0 : return false;
124 : : }
125 : 7 : stats_.record_accepted(size);
126 : 7 : net::post(strand_, [self = shared_from_this(), buf = std::move(pooled_buffer)]() mutable {
127 : 7 : const auto added = buf.size();
128 [ + - - + : 7 : if (!self->alive_ || self->closing_) { // Double-check in case session was closed
- + ]
129 : 0 : queue_util::release_reserved_limit_bytes(self->write_reserve_mtx_, self->inflight_bytes_, added);
130 : 0 : self->stats_.record_failed_send();
131 : 0 : return;
132 : : }
133 : 7 : self->route_enqueued_buffer(BufferVariant{std::move(buf)}, added);
134 : : });
135 : 7 : return true;
136 : : }
137 : 7 : }
138 : :
139 : : // Fallback to regular allocation for large buffers or pool exhaustion
140 [ + - + + ]: 53 : if (!queue_util::try_reserve_limit_bytes(write_reserve_mtx_, queue_bytes_, pending_bytes_, inflight_bytes_, size,
141 : : bp_limit_)) {
142 : 50 : stats_.record_failed_send();
143 : 50 : return false;
144 : : }
145 : 3 : std::vector<uint8_t> fallback(data.begin(), data.end());
146 : 3 : stats_.record_accepted(size);
147 : :
148 : 3 : net::post(strand_, [self = shared_from_this(), buf = std::move(fallback), size]() mutable {
149 [ + - - + : 3 : if (!self->alive_ || self->closing_) { // Double-check in case session was closed
- + ]
150 : 0 : queue_util::release_reserved_limit_bytes(self->write_reserve_mtx_, self->inflight_bytes_, size);
151 : 0 : self->stats_.record_failed_send();
152 : 0 : return;
153 : : }
154 : 3 : self->route_enqueued_buffer(BufferVariant{std::move(buf)}, size);
155 : : });
156 : 3 : return true;
157 : 3 : }
158 : :
159 : 24023 : bool TcpServerSession::async_write_move(std::vector<uint8_t>&& data) {
160 [ + - - + : 24023 : if (!alive_ || closing_) {
- + ]
161 : 0 : stats_.record_failed_send();
162 : 0 : return false;
163 : : }
164 : 24023 : const auto added = data.size();
165 [ - + ]: 24023 : if (added == 0) {
166 : 0 : stats_.record_failed_send();
167 : 0 : return false;
168 : : }
169 [ - + ]: 24023 : if (added > base::constants::MAX_BUFFER_SIZE) {
170 : 0 : WIRESTEAD_LOG_ERROR("tcp_server_session", "write", "Write size exceeds maximum allowed");
171 : 0 : stats_.record_failed_send();
172 : 0 : return false;
173 : : }
174 [ + + ]: 24023 : if (!queue_util::try_reserve_limit_bytes(write_reserve_mtx_, queue_bytes_, pending_bytes_, inflight_bytes_, added,
175 : : bp_limit_)) {
176 : 13517 : stats_.record_failed_send();
177 : 13517 : return false;
178 : : }
179 : 10506 : stats_.record_accepted(added);
180 : 10506 : net::post(strand_, [self = shared_from_this(), buf = std::move(data), added]() mutable {
181 [ + - - + : 10506 : if (!self->alive_ || self->closing_) {
- + ]
182 : 0 : queue_util::release_reserved_limit_bytes(self->write_reserve_mtx_, self->inflight_bytes_, added);
183 : 0 : self->stats_.record_failed_send();
184 : 0 : return;
185 : : }
186 : 10506 : self->route_enqueued_buffer(BufferVariant{std::move(buf)}, added);
187 : : });
188 : 10506 : return true;
189 : : }
190 : :
191 : 2 : bool TcpServerSession::async_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
192 [ + - + - : 2 : if (!alive_ || closing_ || !data || data->empty()) {
+ - - + -
+ ]
193 : 0 : stats_.record_failed_send();
194 : 0 : return false;
195 : : }
196 : 2 : const auto added = data->size();
197 [ - + ]: 2 : if (added > base::constants::MAX_BUFFER_SIZE) {
198 : 0 : WIRESTEAD_LOG_ERROR("tcp_server_session", "write", "Write size exceeds maximum allowed");
199 : 0 : stats_.record_failed_send();
200 : 0 : return false;
201 : : }
202 [ + + ]: 2 : if (!queue_util::try_reserve_limit_bytes(write_reserve_mtx_, queue_bytes_, pending_bytes_, inflight_bytes_, added,
203 : : bp_limit_)) {
204 : 1 : stats_.record_failed_send();
205 : 1 : return false;
206 : : }
207 : 1 : stats_.record_accepted(added);
208 : 1 : net::post(strand_, [self = shared_from_this(), buf = std::move(data), added]() mutable {
209 [ + - - + : 1 : if (!self->alive_ || self->closing_) {
- + ]
210 : 0 : queue_util::release_reserved_limit_bytes(self->write_reserve_mtx_, self->inflight_bytes_, added);
211 : 0 : self->stats_.record_failed_send();
212 : 0 : return;
213 : : }
214 : 1 : self->route_enqueued_buffer(BufferVariant{std::move(buf)}, added);
215 : : });
216 : 1 : return true;
217 : : }
218 : :
219 : 6 : bool TcpServerSession::async_try_write_copy(memory::ConstByteSpan data) {
220 [ + - - + : 6 : if (data.empty() || data.size() > base::constants::MAX_BUFFER_SIZE) {
- + ]
221 : 0 : stats_.record_failed_send();
222 : 0 : return false;
223 : : }
224 : 12 : return async_try_write_move(std::vector<uint8_t>(data.begin(), data.end()));
225 : : }
226 : :
227 : 10 : bool TcpServerSession::async_try_write_move(std::vector<uint8_t>&& data) {
228 [ + - - + : 10 : if (!alive_ || closing_) {
- + ]
229 : 0 : stats_.record_failed_send();
230 : 0 : return false;
231 : : }
232 : 10 : const auto added = data.size();
233 [ + - - + ]: 10 : if (added == 0 || added > base::constants::MAX_BUFFER_SIZE) {
234 : 0 : stats_.record_failed_send();
235 : 0 : return false;
236 : : }
237 : 3 : const auto reject_for_pressure = [this, added]() {
238 [ + + ]: 3 : if (bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
239 : 1 : stats_.record_dropped(1, added);
240 : : } else {
241 : 2 : stats_.record_failed_send();
242 : : }
243 : 13 : };
244 [ + + + - : 17 : if (backpressure_active_.load() || queue_bytes_ + added > bp_high_ ||
+ + ]
245 [ - + ]: 7 : queue_bytes_ + pending_bytes_ + added > bp_limit_) {
246 : 3 : reject_for_pressure();
247 : 3 : return false;
248 : : }
249 [ - + ]: 7 : if (!queue_util::try_reserve_write_bytes(queue_bytes_, pending_bytes_, backpressure_active_, added, bp_high_,
250 : : bp_limit_)) {
251 : 0 : reject_for_pressure();
252 : 0 : return false;
253 : : }
254 : 7 : stats_.record_accepted(added);
255 : :
256 : 7 : net::post(strand_, [self = shared_from_this(), buf = std::move(data), added]() mutable {
257 [ + - - + : 7 : if (!self->alive_ || self->closing_) {
- + ]
258 : 0 : queue_util::release_reserved_write_bytes(self->queue_bytes_, added);
259 : 0 : self->stats_.record_failed_send();
260 : 0 : return;
261 : : }
262 : :
263 : 7 : self->tx_.emplace_back(std::move(buf));
264 : 7 : self->observe_queue();
265 : 7 : self->report_backpressure(self->queue_bytes_);
266 [ - + ]: 7 : if (!self->writing_) self->do_write();
267 : : });
268 : 7 : return true;
269 : : }
270 : :
271 : 1321 : bool TcpServerSession::async_try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
272 [ + - + - : 1321 : if (!alive_ || closing_ || !data || data->empty()) {
+ - - + -
+ ]
273 : 0 : stats_.record_failed_send();
274 : 0 : return false;
275 : : }
276 : 1321 : const auto added = data->size();
277 [ - + ]: 1321 : if (added > base::constants::MAX_BUFFER_SIZE) {
278 : 0 : stats_.record_failed_send();
279 : 0 : return false;
280 : : }
281 : 225 : const auto reject_for_pressure = [this, added]() {
282 [ - + ]: 225 : if (bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
283 : 0 : stats_.record_dropped(1, added);
284 : : } else {
285 : 225 : stats_.record_failed_send();
286 : : }
287 : 1546 : };
288 [ + + + + : 2417 : if (backpressure_active_.load() || queue_bytes_ + added > bp_high_ ||
+ + ]
289 [ - + ]: 1096 : queue_bytes_ + pending_bytes_ + added > bp_limit_) {
290 : 225 : reject_for_pressure();
291 : 225 : return false;
292 : : }
293 [ - + ]: 1096 : if (!queue_util::try_reserve_write_bytes(queue_bytes_, pending_bytes_, backpressure_active_, added, bp_high_,
294 : : bp_limit_)) {
295 : 0 : reject_for_pressure();
296 : 0 : return false;
297 : : }
298 : 1096 : stats_.record_accepted(added);
299 : :
300 : 1096 : net::post(strand_, [self = shared_from_this(), buf = std::move(data), added]() mutable {
301 [ + - - + : 1096 : if (!self->alive_ || self->closing_) {
- + ]
302 : 0 : queue_util::release_reserved_write_bytes(self->queue_bytes_, added);
303 : 0 : self->stats_.record_failed_send();
304 : 0 : return;
305 : : }
306 : :
307 : 1096 : self->tx_.emplace_back(std::move(buf));
308 : 1096 : self->observe_queue();
309 : 1096 : self->report_backpressure(self->queue_bytes_);
310 [ + + ]: 1096 : if (!self->writing_) self->do_write();
311 : : });
312 : 1096 : return true;
313 : : }
314 : :
315 : 145 : void TcpServerSession::on_bytes(OnBytes cb) {
316 : 145 : auto self = shared_from_this();
317 : 145 : net::dispatch(strand_, [self, cb = std::move(cb)]() mutable {
318 [ + - - + : 145 : if (self->closing_.load() || self->cleanup_done_.load()) return;
- + ]
319 : 145 : self->on_bytes_ = std::move(cb);
320 : : });
321 : 145 : }
322 : 155 : void TcpServerSession::on_backpressure(OnBackpressure cb) {
323 : 155 : auto self = shared_from_this();
324 : 155 : net::dispatch(strand_, [self, cb = std::move(cb)]() mutable {
325 [ + + - + : 155 : if (self->closing_.load() || self->cleanup_done_.load()) return;
+ + ]
326 : 150 : self->on_bp_ = std::move(cb);
327 : : });
328 : 155 : }
329 : 147 : void TcpServerSession::on_close(OnClose cb) {
330 : 147 : auto self = shared_from_this();
331 : 147 : net::dispatch(strand_, [self, cb = std::move(cb)]() mutable {
332 [ + - - + : 147 : if (self->closing_.load() || self->cleanup_done_.load()) return;
- + ]
333 : 147 : self->on_close_ = std::move(cb);
334 : : });
335 : 147 : }
336 : :
337 : 1663 : bool TcpServerSession::alive() const { return alive_.load(); }
338 : :
339 : 179 : wrapper::RuntimeStats TcpServerSession::stats() const {
340 : 537 : return stats_.snapshot(queue_bytes_.load(std::memory_order_relaxed), pending_bytes_.load(std::memory_order_relaxed),
341 : 358 : backpressure_active_.load(std::memory_order_relaxed));
342 : : }
343 : :
344 : 0 : void TcpServerSession::reset_stats() {
345 : 0 : stats_.reset(queue_bytes_.load(std::memory_order_relaxed) + pending_bytes_.load(std::memory_order_relaxed));
346 : 0 : }
347 : :
348 : 33 : void TcpServerSession::stop() {
349 [ + + ]: 33 : if (closing_.exchange(true)) return;
350 : 24 : auto self = shared_from_this();
351 : 24 : net::post(strand_, [self] {
352 : : // Clear callbacks on the strand to block further user callbacks after stop.
353 : 11 : self->on_bytes_ = nullptr;
354 : 11 : self->on_bp_ = nullptr;
355 : 11 : self->on_close_ = nullptr;
356 : 11 : self->idle_timer_.cancel();
357 : 11 : self->do_close();
358 : 11 : });
359 : 24 : }
360 : :
361 : 2 : void TcpServerSession::cancel() {
362 : 2 : auto self = shared_from_this();
363 : 2 : net::dispatch(strand_, [self] {
364 : 2 : self->idle_timer_.cancel();
365 : 2 : boost::system::error_code ec;
366 : : // Cancelling the socket via close() causes ongoing operations to complete with operation_aborted.
367 : : // Unlike stop(), this does NOT set closing_ flag immediately, allowing the
368 : : // error handler to run normally and trigger do_close() via the error path.
369 [ + - ]: 2 : if (self->socket_) {
370 : 2 : self->socket_->close(ec);
371 : : }
372 : 2 : });
373 : 2 : }
374 : :
375 : 2852 : void TcpServerSession::start_read() {
376 : 2852 : auto self = shared_from_this();
377 : 8556 : socket_->async_read_some(
378 : 8556 : net::buffer(rx_.data(), rx_.size()), net::bind_executor(strand_, [self](auto ec, std::size_t n) {
379 [ + + - + : 2828 : if (self->closing_ || !self->alive_) return;
+ + ]
380 [ + + ]: 2821 : if (ec) {
381 : 101 : self->do_close();
382 : 101 : return;
383 : : }
384 : 2720 : self->reset_idle_timer();
385 [ + - ]: 2720 : if (n > 0) self->stats_.record_received(n);
386 [ + - ]: 2720 : if (self->on_bytes_) {
387 : : try {
388 : 2720 : self->on_bytes_(memory::ConstByteSpan(self->rx_.data(), n));
389 : 2 : } catch (const std::exception& e) {
390 : 6 : WIRESTEAD_LOG_ERROR("tcp_server_session", "on_bytes",
391 : : "Exception in on_bytes callback: " + std::string(e.what()));
392 : 2 : self->do_close();
393 : 2 : return;
394 : 0 : } catch (...) {
395 : 0 : WIRESTEAD_LOG_ERROR("tcp_server_session", "on_bytes", "Unknown exception in on_bytes callback");
396 : 0 : self->do_close();
397 : 0 : return;
398 : : }
399 : : }
400 : 2718 : self->start_read();
401 : : }));
402 : 2852 : }
403 : :
404 : 269 : void TcpServerSession::do_write() {
405 [ + + ]: 269 : if (tx_.empty()) {
406 : 93 : writing_ = false;
407 : 93 : return;
408 : : }
409 : 176 : writing_ = true;
410 : 176 : auto self = shared_from_this();
411 : :
412 : : // Drain several queued buffers into one scatter-gather write rather than one
413 : : // send syscall per message. The batch and its views stay alive for the whole
414 : : // operation because `writing_` keeps do_write() from re-entering.
415 : 176 : queue_util::take_gather_batch(tx_, current_write_batch_, current_write_views_);
416 : :
417 : 167 : auto on_write = [self](const boost::system::error_code& ec, std::size_t n) {
418 : : // Release the buffers immediately
419 : 167 : self->current_write_batch_.clear();
420 : :
421 [ + + - + : 167 : if (self->closing_ || !self->alive_) return;
+ + ]
422 [ + - ]: 165 : if (self->queue_bytes_ >= n) {
423 : 165 : self->queue_bytes_ -= n;
424 : : } else {
425 : 0 : self->queue_bytes_ = 0;
426 : : }
427 : 165 : self->report_backpressure(self->queue_bytes_);
428 : :
429 [ + + ]: 165 : if (ec) {
430 : 1 : self->do_close();
431 : 1 : return;
432 : : }
433 : 164 : self->stats_.record_sent(n);
434 : 164 : self->reset_idle_timer();
435 : 164 : self->do_write();
436 : 176 : };
437 : :
438 : 176 : socket_->async_write(current_write_views_, net::bind_executor(strand_, on_write));
439 : 176 : }
440 : :
441 : 146 : void TcpServerSession::do_close() {
442 [ - + ]: 146 : if (cleanup_done_.exchange(true)) return; // Ensures cleanup runs only once
443 : :
444 : 146 : alive_.store(false);
445 : 146 : closing_.store(true); // Redundant, but ensures consistency
446 : :
447 : : // Safely invoke on_close callback
448 : 146 : auto close_cb = std::move(on_close_);
449 : :
450 : 146 : WIRESTEAD_LOG_INFO("tcp_server_session", "disconnect", "Client disconnected");
451 : 146 : boost::system::error_code ec;
452 : 146 : socket_->shutdown(tcp::socket::shutdown_both, ec);
453 : 146 : socket_->close(ec);
454 : :
455 : : // Drain queued/pending writes and unconditionally clear backpressure,
456 : : // notifying any waiter directly - shares UdpChannel's terminal-drain
457 : : // helper (#434). Must run before on_bp_ is cleared below: otherwise a
458 : : // Reliable-mode caller blocked in send_to_blocking() for this client
459 : : // would never be woken up when the client disconnects via a read error
460 : : // or idle timeout (jwsung91/wirestead#452).
461 : : {
462 : 146 : auto f = bp_fields();
463 : 146 : queue_util::drain_and_clear_backpressure(f, on_bp_, [&]() {
464 : 146 : tx_.clear();
465 : 146 : queue_bytes_ = 0;
466 : 146 : pending_.clear();
467 : 146 : pending_bytes_ = 0;
468 : 146 : });
469 : : }
470 : :
471 : : // Clear all callbacks to prevent any further invocations
472 : 146 : on_bytes_ = nullptr;
473 : 146 : on_bp_ = nullptr;
474 : 146 : on_close_ = nullptr;
475 : 146 : idle_timer_.cancel();
476 : :
477 [ + + ]: 146 : if (close_cb) {
478 : : try {
479 : 133 : close_cb();
480 : 0 : } catch (const std::exception& e) {
481 : 0 : WIRESTEAD_LOG_ERROR("tcp_server_session", "on_close", "Exception in on_close callback: " + std::string(e.what()));
482 : 0 : } catch (...) {
483 : 0 : WIRESTEAD_LOG_ERROR("tcp_server_session", "on_close", "Unknown exception in on_close callback");
484 : 0 : }
485 : : }
486 : 146 : }
487 : :
488 : 11969 : queue_util::BackpressureFields TcpServerSession::bp_fields() {
489 : 11969 : return queue_util::BackpressureFields{queue_bytes_, pending_bytes_, backpressure_active_, bp_high_,
490 : 11969 : bp_low_, bp_limit_, bp_strategy_};
491 : : }
492 : :
493 : 10517 : void TcpServerSession::route_enqueued_buffer(BufferVariant&& buf, size_t added) {
494 : 10517 : auto f = bp_fields();
495 : 10517 : queue_util::DropAccounting dropped;
496 : 10517 : auto decision = queue_util::decide_enqueue(f, added, tx_, dropped);
497 [ + + + - ]: 10517 : if (dropped.any()) stats_.record_dropped(dropped.messages, dropped.bytes);
498 : :
499 [ - + ]: 10517 : if (decision == queue_util::EnqueueDecision::Rejected) {
500 : 0 : WIRESTEAD_LOG_ERROR("tcp_server_session", "write", "Queue limit exceeded, dropping message");
501 : : // #448: record as dropped so it's reflected in RuntimeStats instead of
502 : : // silently vanishing after being counted as accepted.
503 : 0 : stats_.record_dropped(1, added);
504 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
505 : 0 : report_backpressure(queue_bytes_ + added);
506 : 10479 : return;
507 : : }
508 [ + + ]: 10517 : if (decision == queue_util::EnqueueDecision::Pending) {
509 : 10479 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, pending_bytes_, inflight_bytes_, added);
510 : 10479 : pending_.emplace_back(std::move(buf));
511 : 10479 : observe_queue();
512 : 10479 : return;
513 : : }
514 : 38 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, queue_bytes_, inflight_bytes_, added);
515 : 38 : tx_.emplace_back(std::move(buf));
516 : 38 : observe_queue();
517 : 38 : report_backpressure(queue_bytes_);
518 [ + + + - ]: 38 : if (!writing_) do_write();
519 : : }
520 : :
521 : 12963 : void TcpServerSession::observe_queue() {
522 : 38889 : stats_.observe_queue(queue_bytes_.load(std::memory_order_relaxed) + pending_bytes_.load(std::memory_order_relaxed));
523 : 12963 : }
524 : :
525 : 1306 : void TcpServerSession::report_backpressure(size_t queued_bytes) {
526 [ + - - + : 1306 : if (closing_ || !alive_) return;
- + ]
527 : 1306 : observe_queue();
528 : 1306 : auto f = bp_fields();
529 : 1306 : queue_util::report_backpressure(
530 : 1306 : f, queued_bytes, on_bp_, stats_,
531 : 0 : [&]() -> size_t {
532 : 37 : const size_t moved = pending_bytes_.exchange(0);
533 [ + + ]: 39 : while (!pending_.empty()) {
534 : 2 : tx_.emplace_back(std::move(pending_.front()));
535 : 2 : pending_.pop_front();
536 : : }
537 : 37 : return moved;
538 : : },
539 : 1306 : [&]() {
540 : 37 : observe_queue();
541 [ - + ]: 37 : if (!writing_) do_write();
542 : 37 : });
543 : : }
544 : :
545 : 3182 : void TcpServerSession::reset_idle_timer() {
546 [ + + ]: 3182 : if (idle_timeout_ms_ <= 0) return;
547 : :
548 : : // Cancel any existing timer
549 : 3 : idle_timer_.cancel();
550 : :
551 : : // Reset timer
552 : 3 : idle_timer_.expires_after(std::chrono::milliseconds(idle_timeout_ms_));
553 : :
554 : 3 : auto self = shared_from_this();
555 : 3 : idle_timer_.async_wait(net::bind_executor(strand_, [self](const boost::system::error_code& ec) {
556 [ + + ]: 3 : if (ec == boost::asio::error::operation_aborted) return;
557 [ + - - + : 1 : if (!self->alive_ || self->closing_) return;
- + ]
558 : :
559 : 1 : WIRESTEAD_LOG_WARNING("tcp_server_session", "timeout", "Connection idle timeout expired, closing session");
560 : 1 : self->do_close();
561 : : }));
562 : 3 : }
563 : :
564 : : } // namespace transport
565 : : } // namespace wirestead
|
