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.hpp"
18 : :
19 : : #include <spdlog/fmt/fmt.h>
20 : :
21 : : #include <algorithm>
22 : : #include <atomic>
23 : : #include <boost/asio.hpp>
24 : : #include <cerrno>
25 : : #include <condition_variable>
26 : : #include <cstdio>
27 : : #include <cstring>
28 : : #include <mutex>
29 : : #include <stop_token>
30 : : #include <string_view>
31 : : #include <thread>
32 : : #include <unordered_map>
33 : :
34 : : #include "wirestead/base/platform.hpp"
35 : : #include "wirestead/builder/auto_initializer.hpp"
36 : : #include "wirestead/concurrency/io_context_manager.hpp"
37 : : #include "wirestead/concurrency/io_thread_hook.hpp"
38 : : #include "wirestead/concurrency/thread_safe_state.hpp"
39 : : #include "wirestead/config/validation.hpp"
40 : : #include "wirestead/diagnostics/callback.hpp"
41 : : #include "wirestead/diagnostics/logger.hpp"
42 : : #include "wirestead/diagnostics/runtime_stats_counter.hpp"
43 : : #include "wirestead/interface/iuds_acceptor.hpp"
44 : : #include "wirestead/transport/base/error_info_holder.hpp"
45 : : #include "wirestead/transport/base/stop_test_hook.hpp"
46 : : #include "wirestead/transport/uds/boost_uds_acceptor.hpp"
47 : : #include "wirestead/transport/uds/uds_server_session.hpp"
48 : : #include "wirestead/wrapper/send_validation.hpp"
49 : :
50 : : #if !defined(WIRESTEAD_PLATFORM_WINDOWS)
51 : : #include <sys/socket.h>
52 : : #include <sys/stat.h>
53 : : #include <sys/un.h>
54 : : #include <unistd.h>
55 : : #endif
56 : :
57 : : namespace wirestead {
58 : : namespace transport {
59 : :
60 : : namespace net = boost::asio;
61 : : using uds = net::local::stream_protocol;
62 : :
63 : : namespace {
64 : :
65 : : // #438: an existing path at the configured socket location should only ever
66 : : // be silently removed if it's a genuinely stale (no longer listened-on)
67 : : // socket file - not a regular file/directory (misconfiguration) and not a
68 : : // socket another live process is still listening on (which would otherwise
69 : : // be silently hijacked instead of failing with a clear error). Returns a
70 : : // non-empty reason string if bind should be refused; empty if it's safe to
71 : : // remove the path (or there's nothing there) and proceed.
72 : 194 : std::string existing_uds_path_blocks_bind(const std::string& path) {
73 : : #if defined(WIRESTEAD_PLATFORM_WINDOWS)
74 : : (void)path;
75 : : return {};
76 : : #else
77 : 194 : struct stat st {};
78 [ + + ]: 194 : if (::stat(path.c_str(), &st) != 0) {
79 : 191 : return {}; // nothing exists at this path - nothing to check
80 : : }
81 [ + + ]: 3 : if (!S_ISSOCK(st.st_mode)) {
82 : 2 : return "existing path is not a socket file";
83 : : }
84 : :
85 : 2 : int probe_fd = ::socket(AF_UNIX, SOCK_STREAM, 0);
86 [ - + ]: 2 : if (probe_fd < 0) {
87 : 0 : return {}; // can't probe - fall back to prior (remove-and-proceed) behavior
88 : : }
89 : 2 : struct sockaddr_un addr {};
90 : 2 : addr.sun_family = AF_UNIX;
91 : 2 : std::strncpy(addr.sun_path, path.c_str(), sizeof(addr.sun_path) - 1);
92 : 2 : int rc = ::connect(probe_fd, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr));
93 : 2 : ::close(probe_fd);
94 [ + + ]: 2 : if (rc == 0) {
95 : 2 : return "another process is already listening on this socket path";
96 : : }
97 : 1 : return {}; // stale socket (nothing accepted the probe connect) - safe to remove
98 : : #endif
99 : : }
100 : :
101 : : } // namespace
102 : :
103 : : struct UdsServer::Impl {
104 : : std::unique_ptr<net::io_context> owned_ioc_;
105 : : net::io_context* ioc_ = nullptr;
106 : : net::strand<net::io_context::executor_type> strand_;
107 : : std::atomic<uint64_t> generation_{0};
108 : : bool run_dispatched_ = false;
109 : : std::atomic<bool> cleanup_started_{false};
110 : : std::unique_ptr<net::executor_work_guard<net::io_context::executor_type>> work_guard_;
111 : : std::jthread ioc_thread_;
112 : : bool owns_ioc_ = true;
113 : :
114 : : std::atomic<bool> stopping_{false};
115 : : std::atomic<ClientId> next_client_id_{0};
116 : : std::mutex target_admission_mtx_;
117 : : std::mutex stop_mtx_;
118 : : std::condition_variable stop_cv_;
119 : : bool cleanup_done_ = false;
120 : : std::mutex join_mtx_;
121 : :
122 : 198 : void mark_cleanup_done() {
123 : 198 : detail::stop_test_hook(this, true);
124 : : {
125 : 198 : std::lock_guard<std::mutex> lock(stop_mtx_);
126 : 198 : cleanup_done_ = true;
127 : 198 : }
128 : 198 : stop_cv_.notify_all();
129 : 198 : }
130 : :
131 : : // Waits for the teardown that was dispatched onto the context, and nothing
132 : : // more: the contract's precondition is that the context's owner keeps it
133 : : // running, and running it here would execute unrelated handlers - another
134 : : // channel's user callbacks included - on the stopping thread.
135 : 625 : void wait_for_cleanup() {
136 : 625 : detail::stop_test_hook(this, false);
137 : 625 : std::unique_lock<std::mutex> lock(stop_mtx_);
138 : 1420 : stop_cv_.wait(lock, [this] { return cleanup_done_; });
139 : 625 : }
140 : :
141 : : // #438: only this instance's own successful bind() may remove the socket
142 : : // file on cleanup. Without this, a second UdsServer pointed at the same
143 : : // path whose start() failed before ever binding (e.g. because a live
144 : : // listener already owns that path) would still delete the first server's
145 : : // socket file the moment stop()/the destructor ran - the exact hijack
146 : : // #438 is about, just reached via stop() instead of start().
147 : : std::atomic<bool> bound_{false};
148 : :
149 : : std::unique_ptr<interface::UdsAcceptorInterface> acceptor_;
150 : : config::UdsServerConfig cfg_;
151 : :
152 : 191 : concurrency::AtomicLinkState state_{base::LinkState::Idle};
153 : : // Shared snapshots for the handlers the io thread copies out per received
154 : : // chunk - a std::function copy allocates whenever the target outgrows its
155 : : // small-object buffer. See interface::SharedCallback. The connect/disconnect
156 : : // handlers stay plain: they fire once per connection, not per chunk.
157 : : interface::SharedCallback<OnBytes> on_bytes_;
158 : : interface::SharedCallback<OnState> on_state_;
159 : : interface::SharedCallback<OnBackpressure> on_bp_;
160 : : MultiClientConnectHandler on_multi_connect_;
161 : : interface::SharedCallback<MultiClientDataHandler> on_multi_data_;
162 : : MultiClientDisconnectHandler on_multi_disconnect_;
163 : : diagnostics::RuntimeStatsCounters stats_;
164 : :
165 : : mutable std::mutex sessions_mutex_;
166 : : std::unordered_map<ClientId, std::shared_ptr<UdsServerSession>> sessions_;
167 : : // Stop removes targets immediately, but their counters remain visible until
168 : : // asynchronous cleanup transfers the final snapshot exactly once.
169 : : std::unordered_map<ClientId, std::shared_ptr<UdsServerSession>> retiring_sessions_;
170 : : wrapper::SendAccounting closed_send_accounting_;
171 : :
172 : 221 : void absorb_session(const std::shared_ptr<UdsServerSession>& session) {
173 : 221 : const auto snapshot = session->stats();
174 : 221 : stats_.absorb(snapshot);
175 : 221 : diagnostics::accumulate_send_accounting(closed_send_accounting_, *snapshot.send_accounting);
176 : 221 : }
177 : :
178 : : ErrorInfoHolder error_info_holder_{"uds_server"};
179 : :
180 : 191 : Impl(const config::UdsServerConfig& cfg, net::io_context* ioc_ptr)
181 [ + + ]: 191 : : owned_ioc_(ioc_ptr ? nullptr : std::make_unique<net::io_context>()),
182 [ + + ]: 191 : ioc_(ioc_ptr ? ioc_ptr : owned_ioc_.get()),
183 : 191 : strand_(net::make_strand(*ioc_)),
184 : 191 : owns_ioc_(!ioc_ptr),
185 : 382 : cfg_(cfg) {
186 : 191 : config::detail::validate(cfg_);
187 : 187 : acceptor_ = std::make_unique<BoostUdsAcceptor>(*ioc_);
188 : 251 : }
189 : 187 : ~Impl() {
190 : 187 : stopping_ = true;
191 [ - + ]: 187 : if (work_guard_) {
192 : 0 : work_guard_.reset();
193 : : }
194 [ + - + + ]: 187 : if (ioc_ && owns_ioc_) {
195 [ - + ]: 70 : if (ioc_thread_.joinable()) {
196 [ # # ]: 0 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
197 : 0 : ioc_thread_.detach();
198 : : } else {
199 : 0 : ioc_thread_.request_stop();
200 : 0 : ioc_thread_.join();
201 : : }
202 : : }
203 : : }
204 : : // UDS Cleanup: socket file should be removed, but only if this instance
205 : : // actually bound it (#438).
206 [ - + ]: 187 : if (bound_.load()) {
207 : 0 : std::remove(cfg_.socket_path.c_str());
208 : : }
209 : 187 : }
210 : : void do_accept(std::shared_ptr<UdsServer> self, uint64_t generation);
211 : : void notify_state();
212 : :
213 : 198 : void finish_cleanup() {
214 [ + + ]: 198 : if (bound_.exchange(false)) std::remove(cfg_.socket_path.c_str());
215 : 198 : state_.set(base::LinkState::Idle);
216 [ + + ]: 198 : if (owns_ioc_) work_guard_.reset();
217 : 198 : mark_cleanup_done();
218 : 198 : }
219 : :
220 : 198 : void perform_cleanup(std::shared_ptr<UdsServer> self = {}) {
221 [ - + ]: 308 : if (cleanup_started_.exchange(true)) return;
222 : 198 : boost::system::error_code ec;
223 [ + - + - ]: 198 : if (acceptor_) acceptor_->close(ec);
224 : 198 : std::vector<std::pair<ClientId, std::shared_ptr<UdsServerSession>>> sessions;
225 : : {
226 : 198 : std::lock_guard<std::mutex> lock(sessions_mutex_);
227 : 198 : retiring_sessions_.swap(sessions_);
228 [ + - + + ]: 313 : for (const auto& entry : retiring_sessions_) sessions.push_back(entry);
229 : 198 : }
230 [ + + + - ]: 198 : if (auto hook = detail::g_server_sessions_retiring_hook.load()) hook();
231 [ + + ]: 198 : if (sessions.empty()) {
232 : 110 : finish_cleanup();
233 : 110 : return;
234 : : }
235 : : // Completion includes every session's callback body and cleanup. Each
236 : : // session owns its outstanding I/O; final state changes are serialized
237 : : // with accept/retry handlers on the server's management strand.
238 : 88 : auto remaining = std::make_shared<size_t>(sessions.size());
239 [ + + ]: 203 : for (auto& [client_id, session] : sessions) {
240 : 115 : session->async_stop([this, self, remaining, client_id] {
241 : 115 : net::post(strand_, [this, self, remaining, client_id] {
242 : : {
243 : 115 : std::lock_guard<std::mutex> lock(sessions_mutex_);
244 : 115 : const auto it = retiring_sessions_.find(client_id);
245 [ + - ]: 115 : if (it != retiring_sessions_.end()) {
246 : 115 : absorb_session(it->second);
247 : 115 : retiring_sessions_.erase(it);
248 : : }
249 : 115 : }
250 [ + + ]: 115 : if (--*remaining == 0) finish_cleanup();
251 : 115 : });
252 : 115 : });
253 : : }
254 : 198 : }
255 : :
256 : 659 : void stop(std::shared_ptr<UdsServer> self) {
257 : 659 : const bool on_executor = ioc_->get_executor().running_in_this_thread();
258 : : bool first;
259 : : {
260 : 659 : std::lock_guard<std::mutex> lock(target_admission_mtx_);
261 : 659 : first = !stopping_.exchange(true);
262 [ + + ]: 659 : if (first) {
263 : 198 : std::lock_guard<std::mutex> sessions_lock(sessions_mutex_);
264 [ + + ]: 312 : for (auto& entry : sessions_)
265 [ + - + - ]: 114 : if (entry.second) entry.second->request_stop();
266 : 198 : }
267 : 659 : }
268 [ + + ]: 659 : if (first) {
269 : : {
270 : 198 : std::lock_guard<std::mutex> lock(sessions_mutex_);
271 : 198 : on_bytes_ = nullptr;
272 : 198 : on_state_ = nullptr;
273 : 198 : on_bp_ = nullptr;
274 : 198 : on_multi_connect_ = nullptr;
275 : 198 : on_multi_data_ = nullptr;
276 : 198 : on_multi_disconnect_ = nullptr;
277 : 198 : }
278 [ + + ]: 198 : if (!run_dispatched_) {
279 : 9 : perform_cleanup(self);
280 : : } else {
281 : 378 : net::post(strand_, [this, self] { perform_cleanup(self); });
282 : : }
283 : : }
284 [ + + ]: 659 : if (on_executor) return;
285 : 625 : wait_for_cleanup();
286 : 625 : std::lock_guard<std::mutex> join_lock(join_mtx_);
287 : 625 : join_owned_thread();
288 : 625 : }
289 : :
290 : 625 : void join_owned_thread() {
291 [ + + + + : 625 : if (owns_ioc_ && ioc_thread_.joinable()) ioc_thread_.join();
+ + ]
292 : 625 : }
293 : : };
294 : :
295 : 73 : std::shared_ptr<UdsServer> UdsServer::create(const config::UdsServerConfig& cfg) {
296 : 73 : return std::shared_ptr<UdsServer>(new UdsServer(cfg));
297 : : }
298 : :
299 : 118 : std::shared_ptr<UdsServer> UdsServer::create(const config::UdsServerConfig& cfg,
300 : : std::unique_ptr<interface::UdsAcceptorInterface> acceptor,
301 : : net::io_context& ioc) {
302 : 119 : return std::shared_ptr<UdsServer>(new UdsServer(cfg, std::move(acceptor), ioc));
303 : : }
304 : :
305 : 76 : UdsServer::UdsServer(const config::UdsServerConfig& cfg) : impl_(std::make_unique<Impl>(cfg, nullptr)) {}
306 : 118 : UdsServer::UdsServer(const config::UdsServerConfig& cfg, std::unique_ptr<interface::UdsAcceptorInterface> acceptor,
307 : 118 : net::io_context& ioc)
308 : 118 : : impl_(std::make_unique<Impl>(cfg, &ioc)) {
309 : 117 : impl_->acceptor_ = std::move(acceptor);
310 : 119 : }
311 : :
312 : 374 : UdsServer::~UdsServer() {
313 [ + - ]: 187 : if (impl_) {
314 : 187 : impl_->stop(nullptr);
315 : : }
316 : 374 : }
317 : :
318 : 0 : UdsServer::UdsServer(UdsServer&&) noexcept = default;
319 : 0 : UdsServer& UdsServer::operator=(UdsServer&&) noexcept = default;
320 : :
321 : 194 : void UdsServer::start() {
322 [ - + ]: 199 : if (impl_->state_.get() == base::LinkState::Listening) return;
323 : :
324 : 194 : const auto generation = impl_->generation_.fetch_add(1) + 1;
325 : 194 : impl_->stopping_ = false;
326 : 194 : impl_->run_dispatched_ = false;
327 : 194 : impl_->cleanup_started_ = false;
328 : : {
329 : 194 : std::lock_guard<std::mutex> lock(impl_->stop_mtx_);
330 : 194 : impl_->cleanup_done_ = false;
331 : 194 : }
332 : : // Restart contract (#444): stats() resets on restart. The server-level
333 : : // counters now outlive the sessions that fed them, so clearing them here is
334 : : // what keeps that promise - before absorption they were empty and a restart
335 : : // zeroed the aggregate for free.
336 : : {
337 : 194 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
338 : 194 : impl_->stats_.reset(0);
339 : 194 : impl_->closed_send_accounting_ = {};
340 : 194 : }
341 : :
342 [ + - - + ]: 194 : if (!impl_->cfg_.is_valid()) {
343 : 0 : WIRESTEAD_LOG_ERROR("uds_server", "start", "Invalid UDS server configuration or socket path");
344 : 0 : impl_->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONFIGURATION,
345 : : "start", {}, "Invalid UDS server configuration or socket path", false, 0);
346 : 0 : impl_->state_.set(base::LinkState::Error);
347 : 0 : impl_->notify_state();
348 : 0 : return;
349 : : }
350 : :
351 : : // #438: only remove an existing path at this location if it's actually a
352 : : // stale socket - never a regular file/directory (misconfiguration), and
353 : : // never a socket another live process is still listening on (that would
354 : : // otherwise be silently hijacked instead of failing loudly).
355 : 194 : std::string block_reason = existing_uds_path_blocks_bind(impl_->cfg_.socket_path);
356 [ + + ]: 194 : if (!block_reason.empty()) {
357 : 2 : std::string msg = fmt::format("Refusing to bind {}: {}", impl_->cfg_.socket_path, block_reason);
358 : 2 : WIRESTEAD_LOG_ERROR("uds_server", "start", msg);
359 : 4 : impl_->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION,
360 : 2 : "start", make_error_code(boost::system::errc::address_in_use), msg, false,
361 : : 0);
362 : 2 : impl_->state_.set(base::LinkState::Error);
363 : 2 : impl_->notify_state();
364 : 2 : return;
365 : 2 : }
366 : 192 : std::remove(impl_->cfg_.socket_path.c_str());
367 : :
368 : 192 : boost::system::error_code ec;
369 : 192 : impl_->acceptor_->open(uds(), ec);
370 [ - + ]: 192 : if (ec) {
371 : 0 : std::string msg = fmt::format("Failed to open acceptor: {}", ec.message());
372 : 0 : WIRESTEAD_LOG_ERROR("uds_server", "start", msg);
373 : 0 : impl_->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::SYSTEM, "open",
374 : : ec, msg, false, 0);
375 : 0 : impl_->state_.set(base::LinkState::Error);
376 : 0 : impl_->notify_state();
377 : 0 : return;
378 : 0 : }
379 : :
380 : 192 : uds::endpoint endpoint;
381 : : try {
382 : 192 : endpoint = uds::endpoint(impl_->cfg_.socket_path);
383 : 0 : } catch (const std::exception& e) {
384 : 0 : std::string msg = fmt::format("Invalid UDS endpoint: {}", e.what());
385 : 0 : WIRESTEAD_LOG_ERROR("uds_server", "start", msg);
386 : 0 : impl_->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONFIGURATION,
387 : 0 : "start", make_error_code(boost::system::errc::filename_too_long), msg, false,
388 : : 0);
389 : 0 : impl_->state_.set(base::LinkState::Error);
390 : 0 : impl_->notify_state();
391 : 0 : return;
392 : 0 : }
393 : :
394 : 192 : impl_->acceptor_->bind(endpoint, ec);
395 [ + + ]: 192 : if (ec) {
396 : 3 : std::string msg = fmt::format("Failed to bind to {}: {}", impl_->cfg_.socket_path, ec.message());
397 : 3 : WIRESTEAD_LOG_ERROR("uds_server", "start", msg);
398 : 3 : impl_->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION,
399 : : "bind", ec, msg, false, 0);
400 : 3 : impl_->state_.set(base::LinkState::Error);
401 : 3 : impl_->notify_state();
402 : 3 : return;
403 : 3 : }
404 : 189 : impl_->bound_.store(true);
405 : :
406 : 189 : impl_->acceptor_->listen(net::socket_base::max_listen_connections, ec);
407 [ - + ]: 189 : if (ec) {
408 : 0 : std::string msg = fmt::format("Failed to listen: {}", ec.message());
409 : 0 : WIRESTEAD_LOG_ERROR("uds_server", "start", msg);
410 : 0 : impl_->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION,
411 : : "listen", ec, msg, false, 0);
412 : 0 : impl_->state_.set(base::LinkState::Error);
413 : 0 : impl_->notify_state();
414 : 0 : return;
415 : 0 : }
416 : :
417 : : // #438: restrict local access to the socket file if requested. Best-effort
418 : : // - a chmod failure is logged but does not fail startup, since the socket
419 : : // is already bound and listening at this point.
420 [ + + ]: 189 : if (impl_->cfg_.socket_permissions != -1) {
421 : : #if !defined(WIRESTEAD_PLATFORM_WINDOWS)
422 [ - + ]: 1 : if (::chmod(impl_->cfg_.socket_path.c_str(), static_cast<mode_t>(impl_->cfg_.socket_permissions)) != 0) {
423 : 0 : WIRESTEAD_LOG_WARNING("uds_server", "start",
424 : : fmt::format("Failed to chmod socket {}: {}", impl_->cfg_.socket_path, strerror(errno)));
425 : : }
426 : : #else
427 : : WIRESTEAD_LOG_WARNING("uds_server", "start", "socket_permissions is ignored on Windows");
428 : : #endif
429 : : }
430 : :
431 : 189 : impl_->state_.set(base::LinkState::Listening);
432 : 189 : impl_->notify_state();
433 [ - + ]: 189 : if (impl_->stopping_) return;
434 : :
435 [ + + + - : 189 : if (impl_->owns_ioc_ && !impl_->ioc_thread_.joinable()) {
+ + ]
436 [ + - - + ]: 67 : if (impl_->ioc_->stopped()) {
437 : 0 : impl_->ioc_->restart();
438 : : }
439 : 67 : impl_->work_guard_ =
440 : 134 : std::make_unique<net::executor_work_guard<net::io_context::executor_type>>(net::make_work_guard(*impl_->ioc_));
441 : 134 : impl_->ioc_thread_ = std::jthread([impl = impl_.get()](std::stop_token st) {
442 : 67 : wirestead::concurrency::run_io_thread_init();
443 : : try {
444 : 67 : std::stop_callback cb(st, [impl] { impl->ioc_->stop(); });
445 : 67 : impl->ioc_->run();
446 : 67 : } catch (...) {
447 : 0 : }
448 : 134 : });
449 : : }
450 : :
451 : 189 : impl_->run_dispatched_ = true;
452 : 378 : net::post(impl_->strand_, [self = shared_from_this(), generation]() { self->impl_->do_accept(self, generation); });
453 : 194 : }
454 : :
455 : 472 : void UdsServer::stop() { impl_->stop(weak_from_this().lock()); }
456 : :
457 : 3 : bool UdsServer::is_connected() const { return impl_->state_.get() == base::LinkState::Listening; }
458 : 0 : bool UdsServer::is_backpressure_active() const { return false; }
459 : :
460 : 128 : bool UdsServer::is_backpressure_active(ClientId client_id) const {
461 : 128 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
462 : 128 : auto it = impl_->sessions_.find(client_id);
463 [ + - + - : 128 : if (it != impl_->sessions_.end() && it->second) {
+ - ]
464 : 128 : return it->second->is_backpressure_active();
465 : : }
466 : 0 : return false;
467 : 128 : }
468 : :
469 : 364 : std::optional<size_t> UdsServer::write_queue_limit(ClientId client_id) const {
470 : 364 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
471 : 364 : auto it = impl_->sessions_.find(client_id);
472 [ + + - + : 364 : if (it == impl_->sessions_.end() || !it->second) return std::nullopt;
+ + ]
473 : 298 : return it->second->write_queue_limit();
474 : 364 : }
475 : :
476 : 514 : boost::asio::any_io_executor UdsServer::get_executor() { return impl_->strand_; }
477 : :
478 : 2027 : wrapper::RuntimeStats UdsServer::stats() const {
479 : : // Snapshot retained totals and live sessions under the same lock as the
480 : : // disconnect transfer; otherwise a retiring session can disappear from both.
481 : 2027 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
482 : 2027 : auto aggregate = impl_->stats_.snapshot(0, 0, false);
483 : 2027 : aggregate.send_accounting = impl_->closed_send_accounting_;
484 [ + + ]: 6081 : for (const auto* sessions : {&impl_->sessions_, &impl_->retiring_sessions_}) {
485 [ + + ]: 4247 : for (const auto& pair : *sessions) {
486 [ - + ]: 193 : if (!pair.second) continue;
487 : 193 : const auto session_stats = pair.second->stats();
488 : 193 : diagnostics::accumulate_send_accounting(*aggregate.send_accounting, *session_stats.send_accounting);
489 : 193 : aggregate.bytes_accepted += session_stats.bytes_accepted;
490 : 193 : aggregate.messages_accepted += session_stats.messages_accepted;
491 : 193 : aggregate.bytes_sent += session_stats.bytes_sent;
492 : 193 : aggregate.messages_sent += session_stats.messages_sent;
493 : 193 : aggregate.bytes_received += session_stats.bytes_received;
494 : 193 : aggregate.messages_received += session_stats.messages_received;
495 : 193 : aggregate.failed_sends += session_stats.failed_sends;
496 : 193 : aggregate.dropped_messages += session_stats.dropped_messages;
497 : 193 : aggregate.dropped_bytes += session_stats.dropped_bytes;
498 : 193 : aggregate.backpressure_events += session_stats.backpressure_events;
499 : : // Retiring sessions retain cumulative totals, not live queue gauges.
500 [ + + ]: 193 : if (sessions == &impl_->sessions_) {
501 : 181 : aggregate.queued_bytes += session_stats.queued_bytes;
502 : 181 : aggregate.pending_bytes += session_stats.pending_bytes;
503 [ + - - + ]: 181 : aggregate.backpressure_active = aggregate.backpressure_active || session_stats.backpressure_active;
504 : : }
505 : : // Peak, not a total: summing per-session peaks would report a depth no
506 : : // session ever reached, because the peaks need not have been simultaneous.
507 : 193 : aggregate.max_queued_bytes = std::max(aggregate.max_queued_bytes, session_stats.max_queued_bytes);
508 : : }
509 : : }
510 : 4054 : return aggregate;
511 : 2027 : }
512 : :
513 : 3 : void UdsServer::reset_stats() {
514 : 3 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
515 : 3 : impl_->stats_.reset(0);
516 : 3 : impl_->closed_send_accounting_ = {};
517 [ + + ]: 9 : for (const auto* sessions : {&impl_->sessions_, &impl_->retiring_sessions_}) {
518 [ + + ]: 11 : for (const auto& entry : *sessions) {
519 [ + - + - ]: 5 : if (entry.second) entry.second->reset_stats();
520 : : }
521 : : }
522 : 3 : }
523 : :
524 : 5 : std::optional<diagnostics::ErrorInfo> UdsServer::last_error_info() const {
525 : 5 : return impl_->error_info_holder_.last_error_info();
526 : : }
527 : :
528 : 1 : bool UdsServer::async_write_copy(memory::ConstByteSpan data) {
529 : 1 : auto shared_data = std::make_shared<const std::vector<uint8_t>>(data.begin(), data.end());
530 : 2 : return async_write_shared(shared_data);
531 : 1 : }
532 : :
533 : 5 : bool UdsServer::async_write_move(std::vector<uint8_t>&& data) {
534 : 5 : auto shared_data = std::make_shared<std::vector<uint8_t>>(std::move(data));
535 : 5 : const bool accepted = async_write_shared(shared_data);
536 : : // Rejected session admissions retain no reference. With no accepting target,
537 : : // ownership still belongs to the caller; restore it before returning false.
538 [ + + ]: 5 : if (!accepted) data = std::move(*shared_data);
539 : 5 : return accepted;
540 : 5 : }
541 : :
542 : 7 : bool UdsServer::async_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
543 [ + + + - : 7 : if (impl_->stopping_.load() || !data || data->empty()) {
- + + + ]
544 : 1 : impl_->stats_.record_failed_send();
545 : 1 : return false;
546 : : }
547 : 6 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
548 : 6 : bool sent = false;
549 : 6 : bool attempted = false;
550 [ + + ]: 10 : for (auto& pair : impl_->sessions_) {
551 [ + - + - : 4 : if (pair.second && pair.second->alive() && pair.second->async_write_shared(data)) {
+ - + - +
+ + - + +
- - ]
552 : 1 : attempted = true;
553 : 1 : sent = true;
554 [ + - + - : 3 : } else if (pair.second && pair.second->alive()) {
+ - + - ]
555 : 3 : attempted = true;
556 : : }
557 : : }
558 [ + + ]: 6 : if (!attempted) impl_->stats_.record_failed_send();
559 : 6 : return sent;
560 : 6 : }
561 : :
562 : 0 : bool UdsServer::async_try_write_copy(memory::ConstByteSpan data) {
563 : 0 : auto shared_data = std::make_shared<const std::vector<uint8_t>>(data.begin(), data.end());
564 : 0 : return async_try_write_shared(shared_data);
565 : 0 : }
566 : :
567 : 4 : bool UdsServer::async_try_write_move(std::vector<uint8_t>&& data) {
568 : 4 : auto shared_data = std::make_shared<std::vector<uint8_t>>(std::move(data));
569 : 4 : const bool accepted = async_try_write_shared(shared_data);
570 : : // Rejected session admissions retain no reference. With no accepting target,
571 : : // ownership still belongs to the caller; restore it before returning false.
572 [ + + ]: 4 : if (!accepted) data = std::move(*shared_data);
573 : 4 : return accepted;
574 : 4 : }
575 : :
576 : 6 : bool UdsServer::async_try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
577 [ + + + - : 6 : if (impl_->stopping_.load() || !data || data->empty()) {
- + + + ]
578 : 1 : impl_->stats_.record_failed_send();
579 : 1 : return false;
580 : : }
581 : 5 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
582 : 5 : bool sent = false;
583 : 5 : bool attempted = false;
584 [ + + ]: 9 : for (auto& pair : impl_->sessions_) {
585 [ + - + - : 4 : if (pair.second && pair.second->alive()) {
+ - + - ]
586 : 4 : attempted = true;
587 [ + - + + ]: 4 : if (pair.second->async_try_write_shared(data)) sent = true;
588 : : }
589 : : }
590 [ + + ]: 5 : if (!attempted) impl_->stats_.record_failed_send();
591 : 5 : return sent;
592 : 5 : }
593 : :
594 : 389 : void UdsServer::on_bytes(OnBytes cb) {
595 : 389 : auto shared = interface::share_callback(std::move(cb));
596 : 389 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
597 : 389 : impl_->on_bytes_ = std::move(shared);
598 : 389 : }
599 : :
600 : 661 : void UdsServer::on_state(OnState cb) {
601 : 661 : auto shared = interface::share_callback(std::move(cb));
602 : 661 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
603 : 661 : impl_->on_state_ = std::move(shared);
604 : 661 : }
605 : :
606 : 654 : void UdsServer::on_backpressure(OnBackpressure cb) {
607 : 654 : auto shared = interface::share_callback(std::move(cb));
608 : 654 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
609 : 654 : impl_->on_bp_ = std::move(shared);
610 : 654 : }
611 : :
612 : 1 : bool UdsServer::broadcast(std::string_view message) {
613 : : auto data =
614 : 0 : std::make_shared<const std::vector<uint8_t>>(reinterpret_cast<const uint8_t*>(message.data()),
615 : 1 : reinterpret_cast<const uint8_t*>(message.data()) + message.size());
616 : 2 : return async_try_write_shared(data);
617 : 1 : }
618 : :
619 : 1 : bool UdsServer::broadcast(memory::ConstByteSpan data) {
620 : 1 : auto shared_data = std::make_shared<const std::vector<uint8_t>>(data.begin(), data.end());
621 : 2 : return async_try_write_shared(shared_data);
622 : 1 : }
623 : :
624 : 116 : wrapper::FanoutResult UdsServer::broadcast_result(memory::ConstByteSpan data, wrapper::SendResult wrapper_state,
625 : : bool append_newline) {
626 : 116 : std::vector<std::pair<ClientId, std::shared_ptr<UdsServerSession>>> targets;
627 : : {
628 : 116 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
629 : 116 : targets.reserve(impl_->sessions_.size());
630 [ + + ]: 307 : for (const auto& [id, session] : impl_->sessions_) {
631 [ + - + - : 191 : if (session && session->alive()) targets.emplace_back(id, session);
+ - + - +
- ]
632 : : }
633 : 116 : }
634 [ + + + - ]: 116 : if (auto hook = detail::g_uds_fanout_snapshot_hook.load()) hook();
635 : 116 : wrapper::FanoutResult result;
636 : 116 : std::shared_ptr<const std::vector<uint8_t>> shared_data;
637 [ + + ]: 307 : for (const auto& [id, session] : targets) {
638 : : const auto size =
639 : : append_newline
640 [ + + + - ]: 191 : ? (data.size() >= base::constants::MAX_BUFFER_SIZE ? base::constants::MAX_BUFFER_SIZE + 1 : data.size() + 1)
641 : 165 : : data.size();
642 : 191 : const auto validation = wrapper::detail::validate_payload_size(size, session->write_queue_limit());
643 [ + + ]: 191 : if (!validation.accepted()) {
644 : 80 : result.add(validation);
645 : 80 : continue;
646 : : }
647 [ - + ]: 111 : if (!wrapper_state.accepted()) {
648 : 0 : result.add(wrapper_state);
649 : 0 : continue;
650 : : }
651 [ + + ]: 111 : if (!shared_data) {
652 : 63 : auto payload = std::make_shared<std::vector<uint8_t>>();
653 [ + - + - ]: 63 : if (!data.empty()) payload->assign(data.begin(), data.end());
654 [ + + + - ]: 63 : if (append_newline) payload->push_back('\n');
655 : 63 : shared_data = std::move(payload);
656 : 63 : }
657 : 111 : std::lock_guard<std::mutex> admission_lock(impl_->target_admission_mtx_);
658 : 111 : const auto state = target_state();
659 [ - + ]: 111 : if (!state.accepted()) {
660 : 0 : result.add(state);
661 : 0 : continue;
662 : : }
663 : 111 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
664 : 111 : const auto it = impl_->sessions_.find(id);
665 [ + - - + : 111 : if (it == impl_->sessions_.end() || it->second != session) {
- + ]
666 : 0 : result.add(wrapper::SendResult::reject(wrapper::SendRejection::NotReady));
667 : 0 : continue;
668 : : }
669 : 111 : result.add(session->try_write_shared(shared_data));
670 : 111 : }
671 : 232 : return result;
672 : 116 : }
673 : :
674 : 64 : bool UdsServer::send_to_client(ClientId client_id, std::string_view message) {
675 : 192 : return send_to_client(client_id,
676 : 192 : memory::ConstByteSpan(reinterpret_cast<const uint8_t*>(message.data()), message.size()));
677 : : }
678 : :
679 : 71 : bool UdsServer::send_to_client(ClientId client_id, memory::ConstByteSpan data) {
680 : 71 : const auto result = write_target(client_id, data, false);
681 [ + + + - ]: 71 : if (auto hook = detail::g_uds_server_write_result_hook.load()) hook(result);
682 : 142 : return result.accepted();
683 : : }
684 : :
685 : 16 : bool UdsServer::try_send_to_client(ClientId client_id, std::string_view message) {
686 : 48 : return try_send_to_client(client_id,
687 : 48 : memory::ConstByteSpan(reinterpret_cast<const uint8_t*>(message.data()), message.size()));
688 : : }
689 : :
690 : 23 : bool UdsServer::try_send_to_client(ClientId client_id, memory::ConstByteSpan data) {
691 : 23 : const auto result = write_target(client_id, data, true);
692 [ + - + - ]: 23 : if (auto hook = detail::g_uds_server_write_result_hook.load()) hook(result);
693 : 46 : return result.accepted();
694 : : }
695 : :
696 : 1023 : wrapper::SendResult UdsServer::target_state() const {
697 [ + + ]: 1023 : if (impl_->stopping_.load()) {
698 : 34 : std::lock_guard<std::mutex> lock(impl_->stop_mtx_);
699 [ + + ]: 34 : return wrapper::SendResult::reject(impl_->cleanup_done_ ? wrapper::SendRejection::NotStarted
700 : 34 : : wrapper::SendRejection::Stopping);
701 : 34 : }
702 [ + + ]: 1978 : if (impl_->generation_.load() == 0) return wrapper::SendResult::reject(wrapper::SendRejection::NotStarted);
703 : 985 : return wrapper::SendResult::accept();
704 : : }
705 : :
706 : 3 : void UdsServer::fail_receive(ClientId id) {
707 : 3 : auto session = capture_target(id);
708 [ - + ]: 3 : if (!session) return;
709 : 6 : net::dispatch(session->strand_, [session] { session->do_close(); });
710 : 3 : }
711 : :
712 : 204 : std::optional<boost::asio::any_io_executor> UdsServer::client_executor(ClientId client_id) const {
713 : 204 : auto session = capture_target(client_id);
714 [ - + ]: 204 : if (!session) return std::nullopt;
715 : 204 : return session->strand_;
716 : 204 : }
717 : :
718 : 417 : std::shared_ptr<UdsServerSession> UdsServer::capture_target(ClientId client_id) const {
719 : 417 : std::lock_guard<std::mutex> admission_lock(impl_->target_admission_mtx_);
720 [ + - - + ]: 417 : if (!target_state().accepted()) return {};
721 : 417 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
722 : 417 : auto it = impl_->sessions_.find(client_id);
723 [ + + ]: 417 : return it == impl_->sessions_.end() ? nullptr : it->second;
724 : 417 : }
725 : :
726 : 296 : std::optional<wrapper::SendResult> UdsServer::poll_target_wait(const std::shared_ptr<UdsServerSession>& session) const {
727 [ - + ]: 296 : if (!session) return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
728 : 296 : return session->poll_write_wait();
729 : : }
730 : :
731 : 423 : void UdsServer::cancel_target_waits() {
732 : 423 : std::lock_guard<std::mutex> admission_lock(impl_->target_admission_mtx_);
733 : 423 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
734 [ + + ]: 531 : for (auto& entry : impl_->sessions_)
735 [ + - + - ]: 108 : if (entry.second) entry.second->cancel_write_wait();
736 : 423 : }
737 : :
738 : 259 : wrapper::SendResult UdsServer::write_target(ClientId client_id, memory::ConstByteSpan data, bool try_only,
739 : : const std::shared_ptr<UdsServerSession>& expected) {
740 [ + + ]: 259 : if (expected) {
741 [ + + + - ]: 108 : if (auto hook = detail::g_uds_server_pinned_write_hook.load()) hook();
742 : : }
743 : 259 : std::lock_guard<std::mutex> admission_lock(impl_->target_admission_mtx_);
744 : 259 : const auto state = target_state();
745 [ + + ]: 259 : if (!state.accepted()) {
746 : 12 : impl_->stats_.record_failed_send();
747 : 12 : return state;
748 : : }
749 : 247 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
750 : 247 : auto it = impl_->sessions_.find(client_id);
751 [ + + + - : 247 : if (it == impl_->sessions_.end() || !it->second || (expected && it->second != expected)) {
+ + - + +
+ ]
752 : 22 : impl_->stats_.record_failed_send();
753 : 22 : return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
754 : : }
755 [ + + + - : 225 : return try_only ? it->second->try_write_copy(data) : it->second->write_copy(data);
+ - ]
756 : 259 : }
757 : :
758 : 537 : size_t UdsServer::client_count() const {
759 : 537 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
760 : 537 : size_t alive = 0;
761 [ + + ]: 835 : for (const auto& [id, session] : impl_->sessions_) {
762 [ + - + - : 298 : if (session && session->alive()) ++alive;
+ + + + ]
763 : : }
764 : 537 : return alive;
765 : 537 : }
766 : :
767 : 90 : std::vector<ClientId> UdsServer::connected_clients() const {
768 : 90 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
769 : 90 : std::vector<ClientId> ids;
770 [ + + ]: 183 : for (const auto& [id, session] : impl_->sessions_) {
771 [ + - + - : 93 : if (session && session->alive()) ids.push_back(id);
+ + + + +
- ]
772 : : }
773 : 180 : return ids;
774 : 90 : }
775 : :
776 : 40 : std::optional<wrapper::RuntimeStats> UdsServer::client_stats(ClientId client_id) const {
777 : 40 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
778 : 40 : auto it = impl_->sessions_.find(client_id);
779 [ + + - + : 40 : if (it == impl_->sessions_.end() || !it->second) return std::nullopt;
+ + ]
780 : 34 : return it->second->stats();
781 : 40 : }
782 : :
783 : 1 : void UdsServer::set_client_limit(size_t max_clients) {
784 : 1 : config::detail::range(max_clients, 0, base::constants::MAX_MAX_CONNECTIONS, "invalid client limit");
785 : 1 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
786 : 2 : impl_->cfg_.max_connections =
787 : 1 : static_cast<int>(std::min(max_clients, static_cast<size_t>(base::constants::MAX_MAX_CONNECTIONS)));
788 : 1 : }
789 : :
790 : 267 : void UdsServer::on_multi_connect(MultiClientConnectHandler handler) {
791 : 267 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
792 : 267 : impl_->on_multi_connect_ = std::move(handler);
793 : 267 : }
794 : :
795 : 267 : void UdsServer::on_multi_data(MultiClientDataHandler handler) {
796 : 267 : auto shared = interface::share_callback(std::move(handler));
797 : 267 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
798 : 267 : impl_->on_multi_data_ = std::move(shared);
799 : 267 : }
800 : :
801 : 264 : void UdsServer::on_multi_disconnect(MultiClientDisconnectHandler handler) {
802 : 264 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
803 : 264 : impl_->on_multi_disconnect_ = std::move(handler);
804 : 264 : }
805 : :
806 : 16 : base::LinkState UdsServer::state() const { return impl_->state_.get(); }
807 : :
808 : 411 : void UdsServer::Impl::do_accept(std::shared_ptr<UdsServer> self, uint64_t generation) {
809 [ + + - + : 411 : if (stopping_ || generation != generation_) return;
+ + ]
810 : 407 : acceptor_->async_accept([self, generation](const boost::system::error_code& ec, uds::socket socket) {
811 : 403 : net::dispatch(self->impl_->strand_, [self, generation, ec, socket = std::move(socket)]() mutable {
812 [ + + - + : 403 : if (self->impl_->stopping_ || generation != self->impl_->generation_) return;
+ + ]
813 : :
814 [ + + ]: 222 : if (!ec) {
815 : : ClientId client_id;
816 : : {
817 : 221 : std::lock_guard<std::mutex> lock(self->impl_->sessions_mutex_);
818 [ + + - + ]: 374 : if (self->impl_->cfg_.max_connections > 0 &&
819 [ - + ]: 153 : self->impl_->sessions_.size() >= static_cast<size_t>(self->impl_->cfg_.max_connections)) {
820 : 0 : boost::system::error_code ignored;
821 : 0 : socket.close(ignored);
822 : 0 : auto* impl = self->impl_.get();
823 : 0 : impl->do_accept(self, generation);
824 : 0 : return;
825 : : }
826 : 221 : client_id = self->impl_->next_client_id_++;
827 : 221 : }
828 : :
829 : : auto session = std::make_shared<UdsServerSession>(
830 : 221 : *self->impl_->ioc_, std::move(socket), self->impl_->cfg_.backpressure_threshold,
831 : 221 : self->impl_->cfg_.idle_timeout_ms, self->impl_->cfg_.backpressure_strategy,
832 : 442 : self->impl_->cfg_.enable_memory_pool, self->impl_->cfg_.read_buffer_size);
833 : :
834 : 221 : std::weak_ptr<UdsServer> weak_self = self;
835 : 221 : session->on_bytes([weak_self, client_id, generation](memory::ConstByteSpan data) {
836 : 82 : auto s = weak_self.lock();
837 [ + - + + : 82 : if (!s || s->impl_->stopping_ || generation != s->impl_->generation_) return;
- + + + ]
838 : 81 : interface::SharedCallback<MultiClientDataHandler> data_handler;
839 : 81 : interface::SharedCallback<OnBytes> bytes_handler;
840 : : {
841 : 81 : std::lock_guard<std::mutex> lock(s->impl_->sessions_mutex_);
842 : 81 : data_handler = s->impl_->on_multi_data_;
843 : 81 : bytes_handler = s->impl_->on_bytes_;
844 : 81 : }
845 : 81 : diagnostics::invoke_callback("uds_server", "on_multi_data", data_handler, client_id, data);
846 : 81 : diagnostics::invoke_callback("uds_server", "on_bytes", bytes_handler, data);
847 : 82 : });
848 : :
849 : : // Forward each session's pressure transitions through the current
850 : : // server handler, just like data. The snapshot permits replacement
851 : : // after connect and invokes user code outside sessions_mutex_.
852 : 221 : session->on_backpressure([weak_self, generation](size_t queued) {
853 : 167 : auto s = weak_self.lock();
854 [ + - + - : 167 : if (!s || s->impl_->stopping_ || generation != s->impl_->generation_) return;
- + - + ]
855 : 167 : interface::SharedCallback<OnBackpressure> handler;
856 : : {
857 : 167 : std::lock_guard<std::mutex> lock(s->impl_->sessions_mutex_);
858 : 167 : handler = s->impl_->on_bp_;
859 : 167 : }
860 : 167 : diagnostics::invoke_callback("uds_server", "on_backpressure", handler, queued);
861 : 167 : });
862 : :
863 : 221 : session->on_close([weak_self, client_id, generation]() {
864 : 108 : auto s = weak_self.lock();
865 [ + - + + : 108 : if (!s || s->impl_->stopping_ || generation != s->impl_->generation_) return;
- + + + ]
866 : 107 : MultiClientDisconnectHandler disconnect_handler;
867 : : {
868 : 107 : std::lock_guard<std::mutex> lock(s->impl_->sessions_mutex_);
869 : 107 : disconnect_handler = s->impl_->on_multi_disconnect_;
870 : 107 : }
871 : : // Keep the closing session visible to stop until notification returns,
872 : : // so native shutdown also waits for this session-strand callback.
873 : 107 : diagnostics::invoke_callback("uds_server", "on_disconnect", disconnect_handler, client_id);
874 : : {
875 : 107 : std::lock_guard<std::mutex> lock(s->impl_->sessions_mutex_);
876 [ + + - + : 107 : if (s->impl_->stopping_ || generation != s->impl_->generation_) return;
+ + ]
877 : 106 : auto it = s->impl_->sessions_.find(client_id);
878 [ + - + - : 106 : if (it != s->impl_->sessions_.end() && it->second) s->impl_->absorb_session(it->second);
+ - + - ]
879 : 106 : s->impl_->sessions_.erase(client_id);
880 : 107 : }
881 : 109 : });
882 : :
883 : : {
884 : 221 : std::lock_guard<std::mutex> lock(self->impl_->sessions_mutex_);
885 : 221 : self->impl_->sessions_[client_id] = session;
886 : 221 : auto connect_handler = self->impl_->on_multi_connect_;
887 : : // Admission becomes visible only after connect is queued on the
888 : : // session strand, ahead of reads, writes and pressure callbacks.
889 : 221 : session->start_with_notification([self, client_id, generation, connect_handler = std::move(connect_handler)] {
890 [ + + - + : 220 : if (self->impl_->stopping_ || self->impl_->generation_ != generation) return;
+ + ]
891 : 219 : diagnostics::invoke_callback("uds_server", "on_connect", connect_handler, client_id, "UDS Client");
892 : : });
893 : 221 : }
894 : :
895 : : // Continue accepting
896 : 221 : auto* impl = self->impl_.get();
897 : 221 : impl->do_accept(self, generation);
898 : 221 : } else {
899 : 1 : auto* impl = self->impl_.get();
900 [ - + ]: 1 : if (impl->stopping_.load()) return;
901 : :
902 : : // Log only real errors, not operation_aborted
903 [ + - ]: 1 : if (ec != boost::asio::error::operation_aborted) {
904 : 1 : std::string msg = fmt::format("Accept failed: {}", ec.message());
905 : 1 : WIRESTEAD_LOG_ERROR("uds_server", "accept", msg);
906 : 2 : impl->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION,
907 : 1 : "accept", ec, msg, true, 0);
908 : : // Keep Listening while the accept operation is retried.
909 : 1 : }
910 : :
911 [ + - ]: 1 : if (!impl->stopping_.load()) {
912 : 1 : auto timer = std::make_shared<net::steady_timer>(impl->strand_);
913 : 1 : timer->expires_after(std::chrono::milliseconds(100));
914 : 1 : timer->async_wait([self, timer, generation](const boost::system::error_code& ec) {
915 : 1 : auto* retry_impl = self->impl_.get();
916 [ + - + - : 1 : if (!ec && !retry_impl->stopping_.load() && generation == retry_impl->generation_) {
+ - + - ]
917 : 1 : retry_impl->do_accept(self, generation);
918 : : }
919 : 1 : });
920 : 1 : }
921 : : }
922 : : });
923 : 403 : });
924 : : }
925 : :
926 : 194 : void UdsServer::Impl::notify_state() {
927 : 194 : interface::SharedCallback<OnState> cb;
928 : : {
929 : 194 : std::lock_guard<std::mutex> lock(sessions_mutex_);
930 : 194 : cb = on_state_;
931 : 194 : }
932 : 194 : diagnostics::invoke_callback("uds_server", "on_state", cb, state_.get());
933 : 194 : }
934 : :
935 : : } // namespace transport
936 : : } // namespace wirestead
|
