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/transport/base/bp_utils.hpp"
20 : : #include "wirestead/transport/uds/boost_uds_socket.hpp"
21 : :
22 : : namespace wirestead {
23 : : namespace transport {
24 : :
25 : 23 : UdsServerSession::UdsServerSession(net::io_context& ioc, uds::socket sock, size_t backpressure_threshold,
26 : : int idle_timeout_ms, base::constants::BackpressureStrategy strategy,
27 : 23 : bool enable_memory_pool, size_t read_buffer_size)
28 : 23 : : ioc_(ioc),
29 : 23 : strand_(net::make_strand(ioc_)),
30 : 23 : idle_timer_(ioc),
31 : 23 : socket_(std::make_unique<BoostUdsSocket>(std::move(sock))),
32 : 23 : enable_memory_pool_(enable_memory_pool),
33 : 23 : bp_strategy_(strategy),
34 : 23 : bp_high_(backpressure_threshold),
35 [ + - ]: 23 : bp_low_(backpressure_threshold > 1 ? backpressure_threshold / 2 : backpressure_threshold),
36 : 23 : bp_limit_(std::min(std::max(backpressure_threshold * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
37 : : base::constants::MAX_BUFFER_SIZE)),
38 : 69 : idle_timeout_ms_(idle_timeout_ms) {
39 : 23 : rx_.resize(
40 : 23 : std::clamp(read_buffer_size, base::constants::MIN_READ_BUFFER_SIZE, base::constants::MAX_READ_BUFFER_SIZE));
41 : 23 : }
42 : :
43 : 11 : UdsServerSession::UdsServerSession(net::io_context& ioc, std::unique_ptr<interface::UdsSocketInterface> socket,
44 : : size_t backpressure_threshold, int idle_timeout_ms,
45 : : base::constants::BackpressureStrategy strategy, bool enable_memory_pool,
46 : 11 : size_t read_buffer_size)
47 : 11 : : ioc_(ioc),
48 : 11 : strand_(net::make_strand(ioc_)),
49 : 11 : idle_timer_(ioc),
50 : 11 : socket_(std::move(socket)),
51 : 11 : enable_memory_pool_(enable_memory_pool),
52 : 11 : bp_strategy_(strategy),
53 : 11 : bp_high_(backpressure_threshold),
54 [ + - ]: 11 : bp_low_(backpressure_threshold > 1 ? backpressure_threshold / 2 : backpressure_threshold),
55 : 11 : bp_limit_(std::min(std::max(backpressure_threshold * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
56 : : base::constants::MAX_BUFFER_SIZE)),
57 : 33 : idle_timeout_ms_(idle_timeout_ms) {
58 : 11 : rx_.resize(
59 : 11 : std::clamp(read_buffer_size, base::constants::MIN_READ_BUFFER_SIZE, base::constants::MAX_READ_BUFFER_SIZE));
60 : 11 : }
61 : :
62 : 33 : void UdsServerSession::start() {
63 : 33 : alive_ = true;
64 : 33 : net::dispatch(strand_, [self = shared_from_this()]() {
65 : 33 : self->reset_idle_timer();
66 : 33 : self->start_read();
67 : 33 : });
68 : 33 : }
69 : :
70 : 18 : void UdsServerSession::stop() {
71 [ + + ]: 18 : if (closing_.exchange(true)) return;
72 : 16 : net::post(strand_, [this, self = shared_from_this()]() {
73 : 15 : on_bytes_ = nullptr;
74 : 15 : on_bp_ = nullptr;
75 : 15 : do_close();
76 : 15 : });
77 : : }
78 : :
79 : 75 : bool UdsServerSession::alive() const { return alive_.load(); }
80 : :
81 : 28 : wrapper::RuntimeStats UdsServerSession::stats() const {
82 : 84 : return stats_.snapshot(queue_bytes_.load(std::memory_order_relaxed), pending_bytes_.load(std::memory_order_relaxed),
83 : 56 : backpressure_active_.load(std::memory_order_relaxed));
84 : : }
85 : :
86 : 0 : void UdsServerSession::reset_stats() {
87 : 0 : stats_.reset(queue_bytes_.load(std::memory_order_relaxed) + pending_bytes_.load(std::memory_order_relaxed));
88 : 0 : }
89 : :
90 : 2 : bool UdsServerSession::async_write_copy(memory::ConstByteSpan data) {
91 : 2 : size_t size = data.size();
92 [ + - + - : 2 : if (enable_memory_pool_ && size > 0 && size <= 65536) {
+ + ]
93 [ + - - + : 1 : if (!alive_ || closing_) {
- + ]
94 : 0 : stats_.record_failed_send();
95 : 1 : return false;
96 : : }
97 : 1 : memory::PooledBuffer pooled(size, pool_);
98 [ + - + - ]: 1 : if (pooled.valid()) {
99 : 1 : base::safe_memory::safe_memcpy(pooled.data(), data.data(), size);
100 [ + - - + ]: 1 : if (!queue_util::try_reserve_limit_bytes(write_reserve_mtx_, queue_bytes_, pending_bytes_, inflight_bytes_, size,
101 : : bp_limit_)) {
102 : 0 : stats_.record_failed_send();
103 : 0 : return false;
104 : : }
105 : 1 : stats_.record_accepted(size);
106 : 1 : net::post(strand_, [this, self = shared_from_this(), buf = std::move(pooled)]() mutable {
107 : 1 : size_t added = buf.size();
108 [ - + ]: 1 : if (!alive_) {
109 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
110 : 0 : stats_.record_failed_send();
111 : 0 : return;
112 : : }
113 : 1 : route_enqueued_buffer(BufferVariant{std::move(buf)}, added);
114 : : });
115 : 1 : return true;
116 : : }
117 : 1 : }
118 : :
119 : 1 : std::vector<uint8_t> vec(data.begin(), data.end());
120 : 1 : return async_write_move(std::move(vec));
121 : 1 : }
122 : :
123 : 5 : bool UdsServerSession::async_write_move(std::vector<uint8_t>&& data) {
124 [ + + - + : 5 : if (!alive_ || closing_) {
+ + ]
125 : 1 : stats_.record_failed_send();
126 : 1 : return false;
127 : : }
128 [ - + ]: 4 : if (data.empty()) {
129 : 0 : stats_.record_failed_send();
130 : 0 : return false;
131 : : }
132 : 4 : const auto added = data.size();
133 [ - + ]: 4 : if (!queue_util::try_reserve_limit_bytes(write_reserve_mtx_, queue_bytes_, pending_bytes_, inflight_bytes_, added,
134 : : bp_limit_)) {
135 : 0 : stats_.record_failed_send();
136 : 0 : return false;
137 : : }
138 : 4 : stats_.record_accepted(added);
139 : 4 : net::post(strand_, [this, self = shared_from_this(), data = std::move(data), added]() mutable {
140 [ - + ]: 4 : if (!alive_) {
141 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
142 : 0 : stats_.record_failed_send();
143 : 0 : return;
144 : : }
145 : 4 : route_enqueued_buffer(BufferVariant{std::move(data)}, added);
146 : : });
147 : 4 : return true;
148 : : }
149 : :
150 : 4 : bool UdsServerSession::async_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
151 [ + + + - : 4 : if (!alive_ || closing_ || !data || data->empty()) {
+ - - + +
+ ]
152 : 2 : stats_.record_failed_send();
153 : 2 : return false;
154 : : }
155 : 2 : const auto added = data->size();
156 [ - + ]: 2 : if (!queue_util::try_reserve_limit_bytes(write_reserve_mtx_, queue_bytes_, pending_bytes_, inflight_bytes_, added,
157 : : bp_limit_)) {
158 : 0 : stats_.record_failed_send();
159 : 0 : return false;
160 : : }
161 : 2 : stats_.record_accepted(added);
162 : 2 : net::post(strand_, [this, self = shared_from_this(), data = std::move(data), added]() mutable {
163 [ - + ]: 2 : if (!alive_) {
164 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
165 : 0 : stats_.record_failed_send();
166 : 0 : return;
167 : : }
168 : 2 : route_enqueued_buffer(BufferVariant{std::move(data)}, added);
169 : : });
170 : 2 : return true;
171 : : }
172 : :
173 : 3 : bool UdsServerSession::async_try_write_copy(memory::ConstByteSpan data) {
174 [ + - - + : 3 : if (data.empty() || data.size() > base::constants::MAX_BUFFER_SIZE) {
- + ]
175 : 0 : stats_.record_failed_send();
176 : 0 : return false;
177 : : }
178 : 6 : return async_try_write_move(std::vector<uint8_t>(data.begin(), data.end()));
179 : : }
180 : :
181 : 6 : bool UdsServerSession::async_try_write_move(std::vector<uint8_t>&& data) {
182 [ + - - + : 6 : if (!alive_ || closing_) {
- + ]
183 : 0 : stats_.record_failed_send();
184 : 0 : return false;
185 : : }
186 : 6 : const auto added = data.size();
187 [ + - - + ]: 6 : if (added == 0 || added > base::constants::MAX_BUFFER_SIZE) {
188 : 0 : stats_.record_failed_send();
189 : 0 : return false;
190 : : }
191 : 3 : const auto reject_for_pressure = [this, added]() {
192 [ + + ]: 3 : if (bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
193 : 1 : stats_.record_dropped(1, added);
194 : : } else {
195 : 2 : stats_.record_failed_send();
196 : : }
197 : 9 : };
198 [ + + + - : 9 : if (backpressure_active_.load() || queue_bytes_ + added > bp_high_ ||
+ + ]
199 [ - + ]: 3 : queue_bytes_ + pending_bytes_ + added > bp_limit_) {
200 : 3 : reject_for_pressure();
201 : 3 : return false;
202 : : }
203 [ - + ]: 3 : if (!queue_util::try_reserve_write_bytes(queue_bytes_, pending_bytes_, backpressure_active_, added, bp_high_,
204 : : bp_limit_)) {
205 : 0 : reject_for_pressure();
206 : 0 : return false;
207 : : }
208 : 3 : stats_.record_accepted(added);
209 : :
210 : 3 : net::post(strand_, [this, self = shared_from_this(), data = std::move(data), added]() mutable {
211 [ + - - + : 3 : if (!alive_ || closing_) {
- + ]
212 : 0 : queue_util::release_reserved_write_bytes(queue_bytes_, added);
213 : 0 : stats_.record_failed_send();
214 : 0 : return;
215 : : }
216 : :
217 : 3 : tx_.emplace_back(std::move(data));
218 : 3 : observe_queue();
219 : 3 : report_backpressure(queue_bytes_);
220 [ - + ]: 3 : if (!writing_) do_write();
221 : : });
222 : 3 : return true;
223 : : }
224 : :
225 : 70 : bool UdsServerSession::async_try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
226 [ + - + - : 70 : if (!alive_ || closing_ || !data || data->empty()) {
+ - - + -
+ ]
227 : 0 : stats_.record_failed_send();
228 : 0 : return false;
229 : : }
230 : 70 : const auto added = data->size();
231 [ - + ]: 70 : if (added > base::constants::MAX_BUFFER_SIZE) {
232 : 0 : stats_.record_failed_send();
233 : 0 : return false;
234 : : }
235 : 1 : const auto reject_for_pressure = [this, added]() {
236 [ - + ]: 1 : if (bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
237 : 0 : stats_.record_dropped(1, added);
238 : : } else {
239 : 1 : stats_.record_failed_send();
240 : : }
241 : 71 : };
242 [ + + + - : 139 : if (backpressure_active_.load() || queue_bytes_ + added > bp_high_ ||
+ + ]
243 [ - + ]: 69 : queue_bytes_ + pending_bytes_ + added > bp_limit_) {
244 : 1 : reject_for_pressure();
245 : 1 : return false;
246 : : }
247 [ - + ]: 69 : if (!queue_util::try_reserve_write_bytes(queue_bytes_, pending_bytes_, backpressure_active_, added, bp_high_,
248 : : bp_limit_)) {
249 : 0 : reject_for_pressure();
250 : 0 : return false;
251 : : }
252 : 69 : stats_.record_accepted(added);
253 : :
254 : 69 : net::post(strand_, [this, self = shared_from_this(), data = std::move(data), added]() mutable {
255 [ + + - + : 69 : if (!alive_ || closing_) {
+ + ]
256 : 4 : queue_util::release_reserved_write_bytes(queue_bytes_, added);
257 : 4 : stats_.record_failed_send();
258 : 4 : return;
259 : : }
260 : :
261 : 65 : tx_.emplace_back(std::move(data));
262 : 65 : observe_queue();
263 : 65 : report_backpressure(queue_bytes_);
264 [ + + ]: 65 : if (!writing_) do_write();
265 : : });
266 : 69 : return true;
267 : : }
268 : :
269 : : // Dispatched onto the strand rather than assigned directly: these setters
270 : : // may be called from any user thread (e.g. UdsServer::on_backpressure()
271 : : // forwarding to an already-accepted session), while the strand-confined
272 : : // read sites below access the same fields with no other synchronization.
273 : : // Matches the pattern already used correctly by TcpServerSession (#436).
274 : 23 : void UdsServerSession::on_bytes(OnBytes cb) {
275 : 23 : auto self = shared_from_this();
276 : 23 : net::dispatch(strand_, [self, cb = std::move(cb)]() mutable {
277 [ - + ]: 23 : if (self->closing_.load()) return;
278 : 23 : self->on_bytes_ = std::move(cb);
279 : : });
280 : 23 : }
281 : 5 : void UdsServerSession::on_backpressure(OnBackpressure cb) {
282 : 5 : auto self = shared_from_this();
283 : 5 : net::dispatch(strand_, [self, cb = std::move(cb)]() mutable {
284 [ - + ]: 5 : if (self->closing_.load()) return;
285 : 5 : self->on_bp_ = std::move(cb);
286 : : });
287 : 5 : }
288 : 26 : void UdsServerSession::on_close(OnClose cb) {
289 : 26 : auto self = shared_from_this();
290 : 26 : net::dispatch(strand_, [self, cb = std::move(cb)]() mutable {
291 [ - + ]: 26 : if (self->closing_.load()) return;
292 : 26 : self->on_close_ = std::move(cb);
293 : : });
294 : 26 : }
295 : :
296 : 52 : void UdsServerSession::start_read() {
297 : 156 : socket_->async_read_some(
298 : 104 : net::buffer(rx_.data(), rx_.size()),
299 : 104 : net::bind_executor(strand_, [this, self = shared_from_this()](const boost::system::error_code& ec, size_t bytes) {
300 [ + + - + : 41 : if (closing_ || !alive_) return;
+ + ]
301 [ + + ]: 32 : if (ec) {
302 : 13 : do_close();
303 : 13 : return;
304 : : }
305 [ + - ]: 19 : if (bytes > 0) stats_.record_received(bytes);
306 [ + - + - ]: 19 : if (on_bytes_) on_bytes_(memory::ConstByteSpan(rx_.data(), bytes));
307 : 19 : reset_idle_timer();
308 : 19 : start_read();
309 : : }));
310 : 52 : }
311 : :
312 : 27 : void UdsServerSession::do_write() {
313 [ + + - + : 27 : if (tx_.empty() || writing_) return;
+ + ]
314 : 26 : writing_ = true;
315 : : // Drain several queued buffers into one scatter-gather write rather than one
316 : : // send syscall per message. `writing_` keeps do_write() from re-entering.
317 : 26 : const size_t bytes_to_write = queue_util::take_gather_batch(tx_, current_write_batch_, current_write_views_);
318 : :
319 : 52 : socket_->async_write(current_write_views_,
320 : 52 : net::bind_executor(strand_, [this, self = shared_from_this(), bytes_to_write](
321 : : const boost::system::error_code& ec, size_t written) {
322 [ + + - + : 25 : if (closing_ || !alive_) return;
+ + ]
323 : 22 : writing_ = false;
324 : 22 : current_write_batch_.clear();
325 [ + - ]: 22 : queue_bytes_ = (queue_bytes_ >= bytes_to_write) ? (queue_bytes_ - bytes_to_write) : 0;
326 : 22 : report_backpressure(queue_bytes_);
327 : :
328 [ + + ]: 22 : if (ec) {
329 : 3 : do_close();
330 : 3 : return;
331 : : }
332 : 19 : stats_.record_sent(written);
333 [ + + ]: 19 : if (!tx_.empty()) do_write();
334 : : }));
335 : : }
336 : :
337 : 33 : void UdsServerSession::do_close() {
338 [ + + - + : 33 : if (!closing_.exchange(true) && !alive_) return;
- + ]
339 : 33 : alive_ = false;
340 : 33 : auto close_cb = std::move(on_close_);
341 : :
342 : 33 : boost::system::error_code ec;
343 : 33 : socket_->close(ec);
344 : :
345 : : // Drain queued/pending writes and unconditionally clear backpressure,
346 : : // notifying any waiter directly - shares UdpChannel's terminal-drain
347 : : // helper (#434). Must run before on_bp_ is cleared below: otherwise a
348 : : // Reliable-mode caller blocked in send_to_blocking() for this client
349 : : // would never be woken up when the client disconnects via a read error
350 : : // or idle timeout (jwsung91/wirestead#452).
351 : 33 : writing_ = false;
352 : : {
353 : 33 : auto f = bp_fields();
354 : 33 : queue_util::drain_and_clear_backpressure(f, on_bp_, [&]() {
355 : 33 : tx_.clear();
356 : 33 : current_write_batch_.clear();
357 : 33 : queue_bytes_ = 0;
358 : 33 : pending_.clear();
359 : 33 : pending_bytes_ = 0;
360 : 33 : });
361 : : }
362 : :
363 : 33 : on_bytes_ = nullptr;
364 : 33 : on_bp_ = nullptr;
365 : 33 : on_close_ = nullptr;
366 [ + + ]: 33 : if (close_cb) {
367 : : try {
368 : 25 : close_cb();
369 : 0 : } catch (...) {
370 : 0 : }
371 : : }
372 : 33 : }
373 : :
374 : 137 : queue_util::BackpressureFields UdsServerSession::bp_fields() {
375 : 137 : return queue_util::BackpressureFields{queue_bytes_, pending_bytes_, backpressure_active_, bp_high_,
376 : 137 : bp_low_, bp_limit_, bp_strategy_};
377 : : }
378 : :
379 : 7 : void UdsServerSession::route_enqueued_buffer(BufferVariant&& buf, size_t added) {
380 : 7 : auto f = bp_fields();
381 : 7 : queue_util::DropAccounting dropped;
382 : 7 : auto decision = queue_util::decide_enqueue(f, added, tx_, dropped);
383 [ - + - - ]: 7 : if (dropped.any()) stats_.record_dropped(dropped.messages, dropped.bytes);
384 : :
385 [ - + ]: 7 : if (decision == queue_util::EnqueueDecision::Rejected) {
386 : 0 : WIRESTEAD_LOG_ERROR("uds_server_session", "write", "Queue limit exceeded, dropping message");
387 : : // #448: record as dropped so it's reflected in RuntimeStats instead of
388 : : // silently vanishing after being counted as accepted.
389 : 0 : stats_.record_dropped(1, added);
390 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
391 : 0 : report_backpressure(queue_bytes_ + added);
392 : 0 : return;
393 : : }
394 [ - + ]: 7 : if (decision == queue_util::EnqueueDecision::Pending) {
395 : 0 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, pending_bytes_, inflight_bytes_, added);
396 : 0 : pending_.emplace_back(std::move(buf));
397 : 0 : observe_queue();
398 : 0 : return;
399 : : }
400 : 7 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, queue_bytes_, inflight_bytes_, added);
401 : 7 : tx_.emplace_back(std::move(buf));
402 : 7 : observe_queue();
403 : 7 : report_backpressure(queue_bytes_);
404 [ + - + - ]: 7 : if (!writing_) do_write();
405 : : }
406 : :
407 : 173 : void UdsServerSession::observe_queue() {
408 : 519 : stats_.observe_queue(queue_bytes_.load(std::memory_order_relaxed) + pending_bytes_.load(std::memory_order_relaxed));
409 : 173 : }
410 : :
411 : 97 : void UdsServerSession::report_backpressure(size_t queued_bytes) {
412 [ + - - + : 97 : if (closing_ || !alive_) return;
- + ]
413 : 97 : observe_queue();
414 : 97 : auto f = bp_fields();
415 : 97 : queue_util::report_backpressure(
416 : 97 : f, queued_bytes, on_bp_, stats_,
417 : 0 : [&]() -> size_t {
418 : 1 : const size_t moved = pending_bytes_.exchange(0);
419 [ - + ]: 1 : while (!pending_.empty()) {
420 : 0 : tx_.emplace_back(std::move(pending_.front()));
421 : 0 : pending_.pop_front();
422 : : }
423 : 1 : return moved;
424 : : },
425 : 97 : [&]() {
426 : 1 : observe_queue();
427 [ + - ]: 1 : if (!writing_) do_write();
428 : 1 : });
429 : : }
430 : :
431 : 52 : void UdsServerSession::reset_idle_timer() {
432 [ + + ]: 52 : if (idle_timeout_ms_ <= 0) return;
433 : :
434 : 9 : idle_timer_.cancel();
435 : 9 : idle_timer_.expires_after(std::chrono::milliseconds(idle_timeout_ms_));
436 : :
437 : 9 : auto self = shared_from_this();
438 : 9 : idle_timer_.async_wait(net::bind_executor(strand_, [self](const boost::system::error_code& ec) {
439 [ + + ]: 8 : if (ec == boost::asio::error::operation_aborted) return;
440 [ + - - + : 2 : if (!self->alive_ || self->closing_) return;
- + ]
441 : 2 : WIRESTEAD_LOG_WARNING("uds_server_session", "timeout", "Connection idle timeout expired, closing session");
442 : 2 : self->do_close();
443 : : }));
444 : 9 : }
445 : :
446 : : } // namespace transport
447 : : } // namespace wirestead
|
