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/uds/uds_server_session.hpp"
18 : :
19 : : #include "wirestead/diagnostics/callback.hpp"
20 : : #include "wirestead/transport/base/bp_utils.hpp"
21 : : #include "wirestead/transport/base/stop_test_hook.hpp"
22 : : #include "wirestead/transport/uds/boost_uds_socket.hpp"
23 : :
24 : : namespace wirestead {
25 : : namespace transport {
26 : :
27 : 221 : UdsServerSession::UdsServerSession(net::io_context& ioc, uds::socket sock, size_t backpressure_threshold,
28 : : int idle_timeout_ms, base::constants::BackpressureStrategy strategy,
29 : 221 : bool enable_memory_pool, size_t read_buffer_size)
30 : 221 : : ioc_(ioc),
31 : 221 : strand_(net::make_strand(ioc_)),
32 : 221 : idle_timer_(ioc),
33 : 221 : socket_(std::make_unique<BoostUdsSocket>(std::move(sock))),
34 : 221 : enable_memory_pool_(enable_memory_pool),
35 : 221 : bp_strategy_(strategy),
36 : 221 : bp_high_(backpressure_threshold),
37 [ + - ]: 221 : bp_low_(backpressure_threshold > 1 ? backpressure_threshold / 2 : backpressure_threshold),
38 : 221 : bp_limit_(std::min(std::max(backpressure_threshold * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
39 : : base::constants::MAX_BUFFER_SIZE)),
40 : 663 : idle_timeout_ms_(idle_timeout_ms) {
41 : 221 : rx_.resize(
42 : 221 : std::clamp(read_buffer_size, base::constants::MIN_READ_BUFFER_SIZE, base::constants::MAX_READ_BUFFER_SIZE));
43 : 221 : }
44 : :
45 : 161 : UdsServerSession::UdsServerSession(net::io_context& ioc, std::unique_ptr<interface::UdsSocketInterface> socket,
46 : : size_t backpressure_threshold, int idle_timeout_ms,
47 : : base::constants::BackpressureStrategy strategy, bool enable_memory_pool,
48 : 161 : size_t read_buffer_size)
49 : 161 : : ioc_(ioc),
50 : 161 : strand_(net::make_strand(ioc_)),
51 : 161 : idle_timer_(ioc),
52 : 161 : socket_(std::move(socket)),
53 : 161 : enable_memory_pool_(enable_memory_pool),
54 : 161 : bp_strategy_(strategy),
55 : 161 : bp_high_(backpressure_threshold),
56 [ + - ]: 161 : bp_low_(backpressure_threshold > 1 ? backpressure_threshold / 2 : backpressure_threshold),
57 : 161 : bp_limit_(std::min(std::max(backpressure_threshold * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
58 : : base::constants::MAX_BUFFER_SIZE)),
59 : 483 : idle_timeout_ms_(idle_timeout_ms) {
60 : 161 : rx_.resize(
61 : 161 : std::clamp(read_buffer_size, base::constants::MIN_READ_BUFFER_SIZE, base::constants::MAX_READ_BUFFER_SIZE));
62 : 161 : }
63 : :
64 : 160 : void UdsServerSession::start() { start_with_notification({}); }
65 : :
66 : 381 : void UdsServerSession::start_with_notification(std::function<void()> notify) {
67 : 381 : alive_ = true;
68 : 381 : net::post(strand_, [self = shared_from_this(), notify = std::move(notify)] {
69 [ + + - + : 381 : if (self->closing_ || !self->alive_) return;
+ + ]
70 : 347 : diagnostics::invoke_callback("uds_server_session", "on_connect", notify);
71 [ + + - + : 347 : if (self->closing_ || !self->alive_) return;
+ + ]
72 : 342 : self->reset_idle_timer();
73 : 342 : self->start_read();
74 : : });
75 : 381 : }
76 : :
77 : 216 : void UdsServerSession::stop() { async_stop({}); }
78 : :
79 : 331 : void UdsServerSession::async_stop(std::function<void()> completion) {
80 : 331 : std::lock_guard<std::mutex> admission_lock(submission_mtx_);
81 : 331 : send_accounting_.end(Ledger::Cause::ExplicitStop);
82 : 331 : closing_.store(true);
83 [ + + ]: 331 : if (!wait_ended_by_) wait_ended_by_ = wrapper::SendRejection::NotReady;
84 : 331 : net::post(strand_, [self = shared_from_this(), completion = std::move(completion)] {
85 : 331 : self->on_bytes_ = nullptr;
86 : 331 : self->on_bp_ = nullptr;
87 : 331 : self->on_close_ = nullptr;
88 : 331 : self->idle_timer_.cancel();
89 : 331 : self->do_close();
90 [ + + ]: 331 : if (completion) completion();
91 : 331 : });
92 : 331 : }
93 : :
94 : 296 : std::optional<wrapper::SendResult> UdsServerSession::poll_write_wait() const {
95 : 296 : std::lock_guard<std::mutex> lock(submission_mtx_);
96 [ + + ]: 296 : if (wait_ended_by_) return wrapper::SendResult::reject(*wait_ended_by_);
97 [ + - - + : 254 : if (!alive_ || closing_) return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
- + ]
98 [ + + ]: 254 : if (!backpressure_active_) return wrapper::SendResult::accept();
99 : 156 : return std::nullopt;
100 : 296 : }
101 : :
102 : 114 : void UdsServerSession::request_stop() {
103 : 114 : std::lock_guard<std::mutex> lock(submission_mtx_);
104 : 114 : send_accounting_.end(Ledger::Cause::ExplicitStop);
105 : 114 : closing_ = true;
106 [ + + ]: 114 : if (!wait_ended_by_) wait_ended_by_ = wrapper::SendRejection::CancelledWhileWaiting;
107 : 114 : }
108 : :
109 : 108 : void UdsServerSession::cancel_write_wait() {
110 : 108 : std::lock_guard<std::mutex> lock(submission_mtx_);
111 [ + + ]: 108 : if (!wait_ended_by_) wait_ended_by_ = wrapper::SendRejection::CancelledWhileWaiting;
112 : 108 : }
113 : :
114 : 760 : bool UdsServerSession::alive() const { return alive_.load(); }
115 : :
116 : 935 : wrapper::RuntimeStats UdsServerSession::stats() const {
117 : : auto snapshot =
118 : 2805 : stats_.snapshot(queue_bytes_.load(std::memory_order_relaxed), pending_bytes_.load(std::memory_order_relaxed),
119 : 935 : backpressure_active_.load(std::memory_order_relaxed));
120 : 935 : snapshot.send_accounting = send_accounting_.snapshot();
121 : 935 : return snapshot;
122 : : }
123 : :
124 : 26 : void UdsServerSession::reset_stats() {
125 : 26 : std::lock_guard<std::mutex> lock(submission_mtx_);
126 : 26 : send_accounting_.reset();
127 : 78 : stats_.reset(queue_bytes_.load(std::memory_order_relaxed) + pending_bytes_.load(std::memory_order_relaxed));
128 : 26 : }
129 : :
130 : 890 : bool UdsServerSession::async_write_copy(memory::ConstByteSpan data) {
131 : 890 : const auto result = write_copy(data);
132 [ + + + - ]: 890 : if (auto hook = detail::g_uds_session_write_result_hook.load()) hook(result);
133 : 1780 : return result.accepted();
134 : : }
135 : :
136 : 1059 : wrapper::SendResult UdsServerSession::write_copy(memory::ConstByteSpan data) {
137 : 1059 : std::lock_guard<std::mutex> admission_lock(submission_mtx_);
138 [ + + + - ]: 1059 : if (auto hook = detail::g_uds_session_write_admission_hook.load()) hook();
139 [ + + + + : 1059 : if (!alive_ || closing_) {
+ + ]
140 : 31 : stats_.record_failed_send();
141 : 31 : return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
142 : : } // Don't queue writes if session is not alive
143 : :
144 : 1028 : size_t size = data.size();
145 [ + + ]: 1028 : if (size == 0) {
146 : 6 : stats_.record_failed_send();
147 : 6 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
148 : : }
149 [ + + ]: 1022 : if (size > base::constants::MAX_BUFFER_SIZE) {
150 : 1 : WIRESTEAD_LOG_ERROR("uds_server_session", "write", "Write size exceeds maximum allowed");
151 : 1 : stats_.record_failed_send();
152 : 1 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
153 : : }
154 : :
155 : : // Use memory pool for better performance (only for reasonable sizes)
156 [ + + + + ]: 1021 : if (size <= base::constants::LARGE_BUFFER_THRESHOLD && enable_memory_pool_) { // Only use pool for buffers <= 64KB
157 : 571 : memory::PooledBuffer pooled_buffer(size, pool_);
158 [ + - + - ]: 571 : if (pooled_buffer.valid()) {
159 : : // Copy data to pooled buffer safely
160 : 571 : base::safe_memory::safe_memcpy(pooled_buffer.data(), data.data(), size);
161 [ + - + + ]: 571 : if (!queue_util::try_reserve_limit_bytes(write_reserve_mtx_, queue_bytes_, pending_bytes_, inflight_bytes_, size,
162 : : bp_limit_)) {
163 : 25 : stats_.record_failed_send();
164 : 25 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
165 : : }
166 : 546 : Ledger::Admission admission(send_accounting_, size);
167 : 546 : stats_.record_accepted(size);
168 : 546 : net::post(strand_,
169 : 546 : [self = shared_from_this(), buf = std::move(pooled_buffer), request = admission.request()]() mutable {
170 : 546 : const auto added = buf.size();
171 [ + + + + : 546 : if (!self->alive_ || self->closing_) { // Double-check in case session was closed
+ + ]
172 : 5 : queue_util::release_reserved_limit_bytes(self->write_reserve_mtx_, self->inflight_bytes_, added);
173 : 5 : self->stats_.record_failed_send();
174 : 5 : return;
175 : : }
176 : 541 : self->route_enqueued_buffer(TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
177 : : });
178 : 546 : admission.commit();
179 : 546 : return wrapper::SendResult::accept();
180 : 546 : }
181 : 571 : }
182 : :
183 : : // Fallback to regular allocation for large buffers or pool exhaustion
184 [ + - + + ]: 450 : if (!queue_util::try_reserve_limit_bytes(write_reserve_mtx_, queue_bytes_, pending_bytes_, inflight_bytes_, size,
185 : : bp_limit_)) {
186 : 4 : stats_.record_failed_send();
187 : 4 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
188 : : }
189 : 446 : std::vector<uint8_t> fallback(data.begin(), data.end());
190 : 446 : Ledger::Admission admission(send_accounting_, size);
191 : 446 : stats_.record_accepted(size);
192 : :
193 : 446 : net::post(strand_,
194 : 446 : [self = shared_from_this(), buf = std::move(fallback), size, request = admission.request()]() mutable {
195 [ + + + + : 446 : if (!self->alive_ || self->closing_) { // Double-check in case session was closed
+ + ]
196 : 5 : queue_util::release_reserved_limit_bytes(self->write_reserve_mtx_, self->inflight_bytes_, size);
197 : 5 : self->stats_.record_failed_send();
198 : 5 : return;
199 : : }
200 : 441 : self->route_enqueued_buffer(TrackedBuffer{BufferVariant{std::move(buf)}, request}, size);
201 : : });
202 : 446 : admission.commit();
203 : 446 : return wrapper::SendResult::accept();
204 : 1059 : }
205 : :
206 : 512 : bool UdsServerSession::async_write_move(std::vector<uint8_t>&& data) {
207 : 512 : const auto result = write_move(std::move(data));
208 [ + + + - ]: 512 : if (auto hook = detail::g_uds_session_write_result_hook.load()) hook(result);
209 : 1024 : return result.accepted();
210 : : }
211 : :
212 : 512 : wrapper::SendResult UdsServerSession::write_move(std::vector<uint8_t>&& data) {
213 : 512 : std::lock_guard<std::mutex> admission_lock(submission_mtx_);
214 [ + + + - ]: 512 : if (auto hook = detail::g_uds_session_write_admission_hook.load()) hook();
215 [ + + + + : 512 : if (!alive_ || closing_) {
+ + ]
216 : 14 : stats_.record_failed_send();
217 : 14 : return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
218 : : }
219 : 498 : const auto added = data.size();
220 [ + + ]: 498 : if (added == 0) {
221 : 5 : stats_.record_failed_send();
222 : 5 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
223 : : }
224 [ + + ]: 493 : if (added > base::constants::MAX_BUFFER_SIZE) {
225 : 1 : WIRESTEAD_LOG_ERROR("uds_server_session", "write", "Write size exceeds maximum allowed");
226 : 1 : stats_.record_failed_send();
227 : 1 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
228 : : }
229 [ + - + + ]: 492 : if (!queue_util::try_reserve_limit_bytes(write_reserve_mtx_, queue_bytes_, pending_bytes_, inflight_bytes_, added,
230 : : bp_limit_)) {
231 : 5 : stats_.record_failed_send();
232 : 5 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
233 : : }
234 : 487 : Ledger::Admission admission(send_accounting_, added);
235 : 487 : stats_.record_accepted(added);
236 : 487 : net::post(strand_,
237 : 487 : [self = shared_from_this(), buf = std::move(data), added, request = admission.request()]() mutable {
238 [ + + + + : 487 : if (!self->alive_ || self->closing_) {
+ + ]
239 : 19 : queue_util::release_reserved_limit_bytes(self->write_reserve_mtx_, self->inflight_bytes_, added);
240 : 19 : self->stats_.record_failed_send();
241 : 19 : return;
242 : : }
243 : 468 : self->route_enqueued_buffer(TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
244 : : });
245 : 487 : admission.commit();
246 : 487 : return wrapper::SendResult::accept();
247 : 512 : }
248 : :
249 : 464 : bool UdsServerSession::async_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
250 : 464 : const auto result = write_shared(std::move(data));
251 [ + + + - ]: 464 : if (auto hook = detail::g_uds_session_write_result_hook.load()) hook(result);
252 : 928 : return result.accepted();
253 : : }
254 : :
255 : 464 : wrapper::SendResult UdsServerSession::write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
256 : 464 : std::lock_guard<std::mutex> admission_lock(submission_mtx_);
257 [ + + + - ]: 464 : if (auto hook = detail::g_uds_session_write_admission_hook.load()) hook();
258 [ + + + + : 464 : if (!alive_ || closing_) {
+ + ]
259 : 10 : stats_.record_failed_send();
260 : 10 : return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
261 : : }
262 [ + + + + : 454 : if (!data || data->empty()) {
+ + ]
263 : 6 : stats_.record_failed_send();
264 : 6 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
265 : : }
266 : 448 : const auto added = data->size();
267 [ + + ]: 448 : if (added > base::constants::MAX_BUFFER_SIZE) {
268 : 1 : WIRESTEAD_LOG_ERROR("uds_server_session", "write", "Write size exceeds maximum allowed");
269 : 1 : stats_.record_failed_send();
270 : 1 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
271 : : }
272 [ + - + + ]: 447 : if (!queue_util::try_reserve_limit_bytes(write_reserve_mtx_, queue_bytes_, pending_bytes_, inflight_bytes_, added,
273 : : bp_limit_)) {
274 : 8 : stats_.record_failed_send();
275 : 8 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
276 : : }
277 : 439 : Ledger::Admission admission(send_accounting_, added);
278 : 439 : stats_.record_accepted(added);
279 : 439 : net::post(strand_,
280 : 439 : [self = shared_from_this(), buf = std::move(data), added, request = admission.request()]() mutable {
281 [ + + + + : 439 : if (!self->alive_ || self->closing_) {
+ + ]
282 : 7 : queue_util::release_reserved_limit_bytes(self->write_reserve_mtx_, self->inflight_bytes_, added);
283 : 7 : self->stats_.record_failed_send();
284 : 7 : return;
285 : : }
286 : 432 : self->route_enqueued_buffer(TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
287 : : });
288 : 439 : admission.commit();
289 : 439 : return wrapper::SendResult::accept();
290 : 464 : }
291 : :
292 : 456 : bool UdsServerSession::async_try_write_copy(memory::ConstByteSpan data) {
293 : 456 : const auto result = try_write_copy(data);
294 [ + + + - ]: 456 : if (auto hook = detail::g_uds_session_write_result_hook.load()) hook(result);
295 : 912 : return result.accepted();
296 : : }
297 : :
298 : 512 : wrapper::SendResult UdsServerSession::try_write_copy(memory::ConstByteSpan data) {
299 [ + + + + : 512 : if (data.empty() || data.size() > base::constants::MAX_BUFFER_SIZE) {
+ + ]
300 : 6 : stats_.record_failed_send();
301 [ + + ]: 6 : return wrapper::SendResult::reject(data.empty() ? wrapper::SendRejection::InvalidArgument
302 : 6 : : wrapper::SendRejection::TooLarge);
303 : : }
304 : 1012 : return try_write_move(std::vector<uint8_t>(data.begin(), data.end()));
305 : : }
306 : :
307 : 470 : bool UdsServerSession::async_try_write_move(std::vector<uint8_t>&& data) {
308 : 470 : const auto result = try_write_move(std::move(data));
309 [ + + + - ]: 470 : if (auto hook = detail::g_uds_session_write_result_hook.load()) hook(result);
310 : 940 : return result.accepted();
311 : : }
312 : :
313 : 976 : wrapper::SendResult UdsServerSession::try_write_move(std::vector<uint8_t>&& data) {
314 : 976 : std::lock_guard<std::mutex> admission_lock(submission_mtx_);
315 [ + + + - ]: 976 : if (auto hook = detail::g_uds_session_write_admission_hook.load()) hook();
316 [ + + + + : 976 : if (!alive_ || closing_) {
+ + ]
317 : 16 : stats_.record_failed_send();
318 : 16 : return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
319 : : }
320 : 960 : const auto added = data.size();
321 [ + + + + ]: 960 : if (added == 0 || added > base::constants::MAX_BUFFER_SIZE) {
322 : 6 : stats_.record_failed_send();
323 [ + + ]: 6 : return wrapper::SendResult::reject(added == 0 ? wrapper::SendRejection::InvalidArgument
324 : 6 : : wrapper::SendRejection::TooLarge);
325 : : }
326 : 22 : const auto reject_for_pressure = [this, added]() {
327 [ + + ]: 22 : if (bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
328 : 14 : stats_.record_dropped(1, added);
329 : : } else {
330 : 8 : stats_.record_failed_send();
331 : : }
332 : 976 : };
333 [ + + + + : 1886 : if (backpressure_active_.load() || queue_bytes_ + added > bp_high_ ||
+ + ]
334 [ - + ]: 932 : queue_bytes_ + pending_bytes_ + added > bp_limit_) {
335 : 22 : reject_for_pressure();
336 : 22 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
337 : : }
338 [ - + ]: 932 : if (!queue_util::try_reserve_write_bytes(write_reserve_mtx_, inflight_bytes_, queue_bytes_, pending_bytes_,
339 : 932 : backpressure_active_, added, bp_high_, bp_limit_)) {
340 : 0 : reject_for_pressure();
341 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
342 : : }
343 : 932 : Ledger::Admission admission(send_accounting_, added);
344 : 932 : stats_.record_accepted(added);
345 : :
346 : 932 : net::post(strand_,
347 : 932 : [self = shared_from_this(), buf = std::move(data), added, request = admission.request()]() mutable {
348 [ + + + + : 932 : if (!self->alive_ || self->closing_) {
+ + ]
349 : 26 : queue_util::release_reserved_write_bytes(self->queue_bytes_, added);
350 : 26 : self->stats_.record_failed_send();
351 : 26 : return;
352 : : }
353 : :
354 : 906 : self->tx_.push_back(TrackedBuffer{BufferVariant{std::move(buf)}, request});
355 : 906 : self->observe_queue();
356 : 906 : self->report_backpressure(self->queue_bytes_);
357 [ + + ]: 906 : if (!self->writing_) self->do_write();
358 : : });
359 : 932 : admission.commit();
360 : 932 : return wrapper::SendResult::accept();
361 : 976 : }
362 : :
363 : 460 : bool UdsServerSession::async_try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
364 : 460 : const auto result = try_write_shared(std::move(data));
365 [ + + + - ]: 460 : if (auto hook = detail::g_uds_session_write_result_hook.load()) hook(result);
366 : 920 : return result.accepted();
367 : : }
368 : :
369 : 571 : wrapper::SendResult UdsServerSession::try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
370 : 571 : std::lock_guard<std::mutex> admission_lock(submission_mtx_);
371 [ + + + - ]: 571 : if (auto hook = detail::g_uds_session_write_admission_hook.load()) hook();
372 [ + + + + : 571 : if (!alive_ || closing_) {
+ + ]
373 : 10 : stats_.record_failed_send();
374 : 10 : return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
375 : : }
376 [ + - + + : 561 : if (!data || data->empty()) {
+ + ]
377 : 5 : stats_.record_failed_send();
378 : 5 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
379 : : }
380 : 556 : const auto added = data->size();
381 [ + + ]: 556 : if (added > base::constants::MAX_BUFFER_SIZE) {
382 : 1 : stats_.record_failed_send();
383 : 1 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
384 : : }
385 : 17 : const auto reject_for_pressure = [this, added]() {
386 [ + + ]: 17 : if (bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
387 : 7 : stats_.record_dropped(1, added);
388 : : } else {
389 : 10 : stats_.record_failed_send();
390 : : }
391 : 572 : };
392 [ + + + + : 1094 : if (backpressure_active_.load() || queue_bytes_ + added > bp_high_ ||
+ + ]
393 [ - + ]: 539 : queue_bytes_ + pending_bytes_ + added > bp_limit_) {
394 : 16 : reject_for_pressure();
395 : 16 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
396 : : }
397 [ + + ]: 539 : if (!queue_util::try_reserve_write_bytes(write_reserve_mtx_, inflight_bytes_, queue_bytes_, pending_bytes_,
398 : 539 : backpressure_active_, added, bp_high_, bp_limit_)) {
399 : 1 : reject_for_pressure();
400 : 1 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
401 : : }
402 : 538 : Ledger::Admission admission(send_accounting_, added);
403 : 538 : stats_.record_accepted(added);
404 : :
405 : 538 : net::post(strand_,
406 : 538 : [self = shared_from_this(), buf = std::move(data), added, request = admission.request()]() mutable {
407 [ + + + + : 538 : if (!self->alive_ || self->closing_) {
+ + ]
408 : 17 : queue_util::release_reserved_write_bytes(self->queue_bytes_, added);
409 : 17 : self->stats_.record_failed_send();
410 : 17 : return;
411 : : }
412 : :
413 : 521 : self->tx_.push_back(TrackedBuffer{BufferVariant{std::move(buf)}, request});
414 : 521 : self->observe_queue();
415 : 521 : self->report_backpressure(self->queue_bytes_);
416 [ + + ]: 521 : if (!self->writing_) self->do_write();
417 : : });
418 : 538 : admission.commit();
419 : 538 : return wrapper::SendResult::accept();
420 : 571 : }
421 : :
422 : 221 : void UdsServerSession::on_bytes(OnBytes cb) {
423 : 221 : auto self = shared_from_this();
424 : 221 : net::dispatch(strand_, [self, cb = std::move(cb)]() mutable {
425 [ - + ]: 221 : if (self->closing_.load()) return;
426 : 221 : self->on_bytes_ = std::move(cb);
427 : : });
428 : 221 : }
429 : 226 : void UdsServerSession::on_backpressure(OnBackpressure cb) {
430 : 226 : auto self = shared_from_this();
431 : 226 : net::dispatch(strand_, [self, cb = std::move(cb)]() mutable {
432 [ - + ]: 226 : if (self->closing_.load()) return;
433 : 226 : self->on_bp_ = std::move(cb);
434 : : });
435 : 226 : }
436 : 224 : void UdsServerSession::on_close(OnClose cb) {
437 : 224 : auto self = shared_from_this();
438 : 224 : net::dispatch(strand_, [self, cb = std::move(cb)]() mutable {
439 [ - + ]: 224 : if (self->closing_.load()) return;
440 : 224 : self->on_close_ = std::move(cb);
441 : : });
442 : 224 : }
443 : :
444 : 416 : void UdsServerSession::start_read() {
445 [ + + - + : 416 : if (closing_ || !alive_) return;
+ + ]
446 : 830 : socket_->async_read_some(net::buffer(rx_.data(), rx_.size()),
447 : 830 : [self = shared_from_this()](const boost::system::error_code& ec, size_t bytes) {
448 : 412 : net::dispatch(self->strand_, [self, ec, bytes] {
449 [ + + - + : 412 : if (self->closing_ || !self->alive_) return;
+ + ]
450 [ + + ]: 193 : if (ec) {
451 : 111 : self->do_close();
452 : 111 : return;
453 : : }
454 [ + - ]: 82 : if (bytes > 0) self->stats_.record_received(bytes);
455 : 82 : diagnostics::invoke_callback("uds_server_session", "on_bytes", self->on_bytes_,
456 : 82 : memory::ConstByteSpan(self->rx_.data(), bytes));
457 [ + + - + : 82 : if (self->closing_ || !self->alive_) return;
+ + ]
458 : 74 : self->reset_idle_timer();
459 : 74 : self->start_read();
460 : : });
461 : 412 : });
462 : : }
463 : :
464 : 663 : void UdsServerSession::do_write() {
465 : : // Serialize the actual handoff with stop, not only the enqueue handler.
466 : 663 : std::unique_lock<std::mutex> lock(submission_mtx_);
467 [ + - + - : 663 : if (closing_ || !alive_ || tx_.empty() || writing_) return;
+ + - + +
+ ]
468 : 544 : writing_ = true;
469 : 544 : const size_t bytes_to_write = queue_util::take_gather_batch(tx_, current_write_batch_, current_write_views_, payload);
470 : : // Admission still fences stop/loss; amortize the ledger lock over the batch.
471 : 6984 : send_accounting_.begin_batch(current_write_batch_, [](const auto& item) { return item.request; });
472 : 544 : auto self = shared_from_this();
473 : : // Initiation may write inline; never hold admission across the syscall.
474 : : // The batch members are strand-owned and the completion posts back to it.
475 : 544 : lock.unlock();
476 : : try {
477 : 565 : socket_->async_write(current_write_views_, [self, bytes_to_write](const boost::system::error_code& ec, size_t n) {
478 : : // The interface erases associated executors. Post explicitly, including
479 : : // for endpoints that complete inline during initiation.
480 : 529 : net::post(self->strand_, [self, bytes_to_write, ec, n] {
481 [ + + + + ]: 529 : const bool failed = ec || n != bytes_to_write;
482 : : {
483 : 529 : std::lock_guard<std::mutex> completion_lock(self->submission_mtx_);
484 [ + + - + : 529 : if (self->closing_ || !self->alive_) {
+ + ]
485 : 100 : self->current_write_batch_.clear();
486 : 100 : return;
487 : : }
488 : 429 : self->send_accounting_.complete_batch(
489 : 3521 : self->current_write_batch_, std::min(n, bytes_to_write), [](const auto& item) { return item.request; },
490 : 3092 : [](const auto& item) {
491 : 6184 : return std::visit([](const auto& b) { return queue_util::variant_buffer_size(b); }, item.buffer);
492 : : });
493 : 429 : self->current_write_batch_.clear();
494 [ + + ]: 429 : if (failed) {
495 : 38 : self->send_accounting_.end(Ledger::Cause::ConnectionLoss);
496 : 38 : self->closing_ = true;
497 [ + - ]: 38 : if (!self->wait_ended_by_) self->wait_ended_by_ = wrapper::SendRejection::NotReady;
498 : : }
499 : 529 : }
500 [ + + ]: 429 : if (failed) {
501 : 38 : self->do_close();
502 : 38 : return;
503 : : }
504 : 391 : queue_util::release_reserved_write_bytes(self->queue_bytes_, bytes_to_write);
505 : 391 : self->stats_.record_sent(n);
506 : : // Keep writing_ set while callbacks move pending data or enqueue writes.
507 : 391 : self->report_backpressure(self->queue_bytes_);
508 : 391 : self->writing_ = false;
509 : 391 : self->do_write();
510 : : });
511 : 529 : });
512 : 7 : } catch (...) {
513 : 7 : lock.lock();
514 : 7 : send_accounting_.end(Ledger::Cause::ConnectionLoss);
515 : 7 : closing_ = true;
516 [ + - ]: 7 : if (!wait_ended_by_) wait_ended_by_ = wrapper::SendRejection::NotReady;
517 : 7 : lock.unlock();
518 : 7 : do_close();
519 : 7 : }
520 : 663 : }
521 : :
522 : 492 : void UdsServerSession::do_close() {
523 [ + + ]: 492 : if (cleanup_done_) return;
524 : 382 : cleanup_done_ = true;
525 : : {
526 : 382 : std::lock_guard<std::mutex> lock(submission_mtx_);
527 : 382 : send_accounting_.end(Ledger::Cause::ConnectionLoss);
528 [ + + ]: 382 : if (!wait_ended_by_) wait_ended_by_ = wrapper::SendRejection::NotReady;
529 : 382 : closing_ = true;
530 : 382 : alive_ = false;
531 : 382 : }
532 : 382 : auto close_cb = std::move(on_close_);
533 : :
534 : 382 : boost::system::error_code ec;
535 : 382 : idle_timer_.cancel();
536 : 382 : socket_->close(ec);
537 : :
538 : : // Drain queued/pending writes and unconditionally clear backpressure,
539 : : // notifying any waiter directly - shares UdpChannel's terminal-drain
540 : : // helper (#434). Must run before on_bp_ is cleared below: otherwise a
541 : : // Reliable-mode caller blocked in send_to_blocking() for this client
542 : : // would never be woken up when the client disconnects via a read error
543 : : // or idle timeout (jwsung91/wirestead#452).
544 : 382 : writing_ = false;
545 : : {
546 : 382 : auto f = bp_fields();
547 : 382 : queue_util::drain_and_clear_backpressure(f, on_bp_, [&]() {
548 : 382 : tx_.clear();
549 : : // In-flight write buffers stay owned until its completion runs.
550 : 382 : queue_bytes_ = 0;
551 : 382 : pending_.clear();
552 : 382 : pending_bytes_ = 0;
553 : 382 : });
554 : : }
555 : :
556 : 382 : on_bytes_ = nullptr;
557 : 382 : on_bp_ = nullptr;
558 : 382 : on_close_ = nullptr;
559 [ + + ]: 382 : if (close_cb) {
560 : 111 : diagnostics::invoke_callback("uds_server_session", "on_close", close_cb);
561 : : }
562 : 382 : }
563 : :
564 : 5942 : queue_util::BackpressureFields UdsServerSession::bp_fields() {
565 : 5942 : return queue_util::BackpressureFields{queue_bytes_, pending_bytes_, backpressure_active_, bp_high_,
566 : 5942 : bp_low_, bp_limit_, bp_strategy_, &write_reserve_mtx_};
567 : : }
568 : :
569 : 1882 : void UdsServerSession::route_enqueued_buffer(TrackedBuffer&& buf, size_t added) {
570 : 1882 : auto f = bp_fields();
571 : 1882 : queue_util::DropAccounting dropped;
572 : 1882 : auto decision = queue_util::decide_enqueue(f, added, tx_, dropped, payload, [this](const TrackedBuffer& item) {
573 : : send_accounting_.discard(item.request, Ledger::Cause::QueuePressure);
574 : : });
575 [ - + - - ]: 1882 : if (dropped.any()) stats_.record_dropped(dropped.messages, dropped.bytes);
576 : :
577 [ - + ]: 1882 : if (decision == queue_util::EnqueueDecision::Rejected) {
578 : 0 : send_accounting_.discard(buf.request, Ledger::Cause::QueuePressure);
579 : 0 : WIRESTEAD_LOG_ERROR("uds_server_session", "write", "Queue limit exceeded, dropping message");
580 : : // #448: record as dropped so it's reflected in RuntimeStats instead of
581 : : // silently vanishing after being counted as accepted.
582 : 0 : stats_.record_dropped(1, added);
583 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
584 : 0 : report_backpressure(queue_bytes_ + added);
585 : 22 : return;
586 : : }
587 [ + + ]: 1882 : if (decision == queue_util::EnqueueDecision::Pending) {
588 : 22 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, pending_bytes_, inflight_bytes_, added);
589 : 22 : pending_.emplace_back(std::move(buf));
590 : 22 : observe_queue();
591 : 22 : return;
592 : : }
593 : 1860 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, queue_bytes_, inflight_bytes_, added);
594 : 1860 : tx_.emplace_back(std::move(buf));
595 : 1860 : observe_queue();
596 : 1860 : report_backpressure(queue_bytes_);
597 [ + + + - ]: 1860 : if (!writing_) do_write();
598 : : }
599 : :
600 : 7035 : void UdsServerSession::observe_queue() {
601 : 21105 : stats_.observe_queue(queue_bytes_.load(std::memory_order_relaxed) + pending_bytes_.load(std::memory_order_relaxed));
602 : 7035 : }
603 : :
604 : 3678 : void UdsServerSession::report_backpressure(size_t queued_bytes) {
605 [ + - - + : 3678 : if (closing_ || !alive_) return;
- + ]
606 : 3678 : observe_queue();
607 : 3678 : auto f = bp_fields();
608 : 3678 : queue_util::report_backpressure(
609 : 3678 : f, queued_bytes, on_bp_, stats_,
610 : 0 : [&]() -> size_t {
611 : 48 : const size_t moved = pending_bytes_.exchange(0);
612 [ + + ]: 49 : while (!pending_.empty()) {
613 : 1 : tx_.emplace_back(std::move(pending_.front()));
614 : 1 : pending_.pop_front();
615 : : }
616 : 48 : return moved;
617 : : },
618 : 3678 : [&]() {
619 : 48 : observe_queue();
620 [ - + ]: 48 : if (!writing_) do_write();
621 : 48 : });
622 : : }
623 : :
624 : 416 : void UdsServerSession::reset_idle_timer() {
625 [ + + ]: 416 : if (idle_timeout_ms_ <= 0) return;
626 : :
627 : 9 : idle_timer_.cancel();
628 : 9 : idle_timer_.expires_after(std::chrono::milliseconds(idle_timeout_ms_));
629 : :
630 : 9 : auto self = shared_from_this();
631 : 9 : idle_timer_.async_wait(net::bind_executor(strand_, [self](const boost::system::error_code& ec) {
632 [ + + ]: 9 : if (ec == boost::asio::error::operation_aborted) return;
633 [ + - - + : 2 : if (!self->alive_ || self->closing_) return;
- + ]
634 : 2 : WIRESTEAD_LOG_WARNING("uds_server_session", "timeout", "Connection idle timeout expired, closing session");
635 : 2 : self->do_close();
636 : : }));
637 : 9 : }
638 : :
639 : : } // namespace transport
640 : : } // namespace wirestead
|
