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 <cstdio>
26 : : #include <cstring>
27 : : #include <future>
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/diagnostics/logger.hpp"
40 : : #include "wirestead/diagnostics/runtime_stats_counter.hpp"
41 : : #include "wirestead/interface/iuds_acceptor.hpp"
42 : : #include "wirestead/transport/base/error_info_holder.hpp"
43 : : #include "wirestead/transport/uds/boost_uds_acceptor.hpp"
44 : : #include "wirestead/transport/uds/uds_server_session.hpp"
45 : :
46 : : #if !defined(WIRESTEAD_PLATFORM_WINDOWS)
47 : : #include <sys/socket.h>
48 : : #include <sys/stat.h>
49 : : #include <sys/un.h>
50 : : #include <unistd.h>
51 : : #endif
52 : :
53 : : namespace wirestead {
54 : : namespace transport {
55 : :
56 : : namespace net = boost::asio;
57 : : using uds = net::local::stream_protocol;
58 : :
59 : : namespace {
60 : :
61 : : // #438: an existing path at the configured socket location should only ever
62 : : // be silently removed if it's a genuinely stale (no longer listened-on)
63 : : // socket file - not a regular file/directory (misconfiguration) and not a
64 : : // socket another live process is still listening on (which would otherwise
65 : : // be silently hijacked instead of failing with a clear error). Returns a
66 : : // non-empty reason string if bind should be refused; empty if it's safe to
67 : : // remove the path (or there's nothing there) and proceed.
68 : 32 : std::string existing_uds_path_blocks_bind(const std::string& path) {
69 : : #if defined(WIRESTEAD_PLATFORM_WINDOWS)
70 : : (void)path;
71 : : return {};
72 : : #else
73 : 32 : struct stat st {};
74 [ + + ]: 32 : if (::stat(path.c_str(), &st) != 0) {
75 : 29 : return {}; // nothing exists at this path - nothing to check
76 : : }
77 [ + + ]: 3 : if (!S_ISSOCK(st.st_mode)) {
78 : 2 : return "existing path is not a socket file";
79 : : }
80 : :
81 : 2 : int probe_fd = ::socket(AF_UNIX, SOCK_STREAM, 0);
82 [ - + ]: 2 : if (probe_fd < 0) {
83 : 0 : return {}; // can't probe - fall back to prior (remove-and-proceed) behavior
84 : : }
85 : 2 : struct sockaddr_un addr {};
86 : 2 : addr.sun_family = AF_UNIX;
87 : 2 : std::strncpy(addr.sun_path, path.c_str(), sizeof(addr.sun_path) - 1);
88 : 2 : int rc = ::connect(probe_fd, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr));
89 : 2 : ::close(probe_fd);
90 [ + + ]: 2 : if (rc == 0) {
91 : 2 : return "another process is already listening on this socket path";
92 : : }
93 : 1 : return {}; // stale socket (nothing accepted the probe connect) - safe to remove
94 : : #endif
95 : : }
96 : :
97 : : } // namespace
98 : :
99 : : struct UdsServer::Impl {
100 : : std::unique_ptr<net::io_context> owned_ioc_;
101 : : net::io_context* ioc_ = nullptr;
102 : : std::unique_ptr<net::executor_work_guard<net::io_context::executor_type>> work_guard_;
103 : : std::jthread ioc_thread_;
104 : : bool owns_ioc_ = true;
105 : :
106 : : std::atomic<bool> stopping_{false};
107 : : std::atomic<ClientId> next_client_id_{0};
108 : : // #438: only this instance's own successful bind() may remove the socket
109 : : // file on cleanup. Without this, a second UdsServer pointed at the same
110 : : // path whose start() failed before ever binding (e.g. because a live
111 : : // listener already owns that path) would still delete the first server's
112 : : // socket file the moment stop()/the destructor ran - the exact hijack
113 : : // #438 is about, just reached via stop() instead of start().
114 : : std::atomic<bool> bound_{false};
115 : :
116 : : std::unique_ptr<interface::UdsAcceptorInterface> acceptor_;
117 : : config::UdsServerConfig cfg_;
118 : :
119 : 34 : concurrency::AtomicLinkState state_{base::LinkState::Idle};
120 : : // Shared snapshots for the handlers the io thread copies out per received
121 : : // chunk - a std::function copy allocates whenever the target outgrows its
122 : : // small-object buffer. See interface::SharedCallback. The connect/disconnect
123 : : // handlers stay plain: they fire once per connection, not per chunk.
124 : : interface::SharedCallback<OnBytes> on_bytes_;
125 : : interface::SharedCallback<OnState> on_state_;
126 : : interface::SharedCallback<OnBackpressure> on_bp_;
127 : : MultiClientConnectHandler on_multi_connect_;
128 : : interface::SharedCallback<MultiClientDataHandler> on_multi_data_;
129 : : MultiClientDisconnectHandler on_multi_disconnect_;
130 : : diagnostics::RuntimeStatsCounters stats_;
131 : :
132 : : mutable std::mutex sessions_mutex_;
133 : : std::unordered_map<ClientId, std::shared_ptr<UdsServerSession>> sessions_;
134 : :
135 : : ErrorInfoHolder error_info_holder_{"uds_server"};
136 : :
137 : 34 : Impl(const config::UdsServerConfig& cfg, net::io_context* ioc_ptr)
138 [ + + ]: 34 : : owned_ioc_(ioc_ptr ? nullptr : std::make_unique<net::io_context>()),
139 [ + + ]: 34 : ioc_(ioc_ptr ? ioc_ptr : owned_ioc_.get()),
140 : 34 : owns_ioc_(!ioc_ptr),
141 : 68 : cfg_(cfg) {
142 : 34 : cfg_.validate_and_clamp();
143 : 34 : acceptor_ = std::make_unique<BoostUdsAcceptor>(*ioc_);
144 : 34 : }
145 : 34 : ~Impl() {
146 : 34 : stopping_ = true;
147 [ - + ]: 34 : if (work_guard_) {
148 : 0 : work_guard_.reset();
149 : : }
150 [ + - + + ]: 34 : if (ioc_ && owns_ioc_) {
151 [ - + ]: 23 : if (ioc_thread_.joinable()) {
152 [ # # ]: 0 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
153 : 0 : ioc_thread_.detach();
154 : : } else {
155 : 0 : ioc_thread_.request_stop();
156 : 0 : ioc_thread_.join();
157 : : }
158 : : }
159 : : }
160 : : // UDS Cleanup: socket file should be removed, but only if this instance
161 : : // actually bound it (#438).
162 [ - + ]: 34 : if (bound_.load()) {
163 : 0 : std::remove(cfg_.socket_path.c_str());
164 : : }
165 : 34 : }
166 : : void do_accept(std::shared_ptr<UdsServer> self);
167 : : void notify_state();
168 : :
169 : 34 : void perform_cleanup() {
170 : : try {
171 : 34 : boost::system::error_code ec;
172 [ + - ]: 34 : if (acceptor_) {
173 : 34 : acceptor_->close(ec);
174 : : }
175 : :
176 : 34 : std::vector<std::shared_ptr<UdsServerSession>> sessions_to_stop;
177 : : {
178 : 34 : std::lock_guard<std::mutex> lock(sessions_mutex_);
179 [ + + ]: 44 : for (auto& pair : sessions_) {
180 : 10 : sessions_to_stop.push_back(pair.second);
181 : : }
182 : 34 : sessions_.clear();
183 : 34 : }
184 : :
185 [ + + ]: 44 : for (auto& session : sessions_to_stop) {
186 [ + - ]: 10 : if (session) {
187 : 10 : session->stop();
188 : : }
189 : : }
190 : :
191 [ + + ]: 34 : if (bound_.exchange(false)) {
192 : 27 : std::remove(cfg_.socket_path.c_str());
193 : : }
194 : :
195 : 34 : state_.set(base::LinkState::Idle);
196 : 34 : notify_state();
197 : 34 : } catch (...) {
198 : 0 : }
199 : 34 : }
200 : :
201 : 34 : void stop(std::shared_ptr<UdsServer> self) {
202 [ - + ]: 34 : if (stopping_.exchange(true)) {
203 : 0 : return;
204 : : }
205 : :
206 : : {
207 : 34 : std::lock_guard<std::mutex> lock(sessions_mutex_);
208 : 34 : on_bytes_ = nullptr;
209 : 34 : on_state_ = nullptr;
210 : 34 : on_bp_ = nullptr;
211 : 34 : on_multi_connect_ = nullptr;
212 : 34 : on_multi_data_ = nullptr;
213 : 34 : on_multi_disconnect_ = nullptr;
214 : 34 : }
215 : :
216 [ - + ]: 34 : if (ioc_->get_executor().running_in_this_thread()) {
217 : 0 : perform_cleanup();
218 [ # # ]: 0 : if (owns_ioc_) {
219 : 0 : work_guard_.reset();
220 : 0 : ioc_->stop();
221 : : }
222 : 0 : return;
223 : : }
224 : :
225 [ + + + + ]: 34 : bool has_active_ioc = owns_ioc_ || !ioc_->stopped();
226 : :
227 [ + + + - : 34 : if (has_active_ioc && self) {
+ + ]
228 : 28 : auto cleanup_promise = std::make_shared<std::promise<void>>();
229 : 28 : auto cleanup_future = cleanup_promise->get_future();
230 : :
231 : 28 : std::weak_ptr<UdsServer> weak_self = self;
232 : 28 : net::dispatch(*ioc_, [weak_self, cleanup_promise]() {
233 [ + - ]: 23 : if (auto shared_self = weak_self.lock()) {
234 : 23 : auto* cleanup_impl = shared_self->get_impl();
235 : 23 : cleanup_impl->perform_cleanup();
236 : 23 : }
237 : 23 : cleanup_promise->set_value();
238 : 23 : });
239 : :
240 [ + - + + ]: 28 : if (cleanup_future.wait_for(std::chrono::seconds(2)) == std::future_status::timeout) {
241 : 5 : perform_cleanup();
242 : : }
243 : 28 : } else {
244 : 6 : perform_cleanup();
245 : : }
246 : :
247 [ + + ]: 34 : if (owns_ioc_) {
248 : 23 : work_guard_.reset();
249 : 23 : ioc_->stop();
250 : : }
251 : :
252 [ + + + + : 34 : if (owns_ioc_ && ioc_thread_.joinable()) {
+ + ]
253 [ + - ]: 19 : if (std::this_thread::get_id() != ioc_thread_.get_id()) {
254 : 19 : ioc_thread_.join();
255 : : } else {
256 : 0 : ioc_thread_.detach();
257 : : }
258 : 19 : ioc_->restart();
259 : : }
260 : : }
261 : : };
262 : :
263 : 23 : std::shared_ptr<UdsServer> UdsServer::create(const config::UdsServerConfig& cfg) {
264 : 23 : return std::shared_ptr<UdsServer>(new UdsServer(cfg));
265 : : }
266 : :
267 : 11 : std::shared_ptr<UdsServer> UdsServer::create(const config::UdsServerConfig& cfg,
268 : : std::unique_ptr<interface::UdsAcceptorInterface> acceptor,
269 : : net::io_context& ioc) {
270 : 11 : return std::shared_ptr<UdsServer>(new UdsServer(cfg, std::move(acceptor), ioc));
271 : : }
272 : :
273 : 23 : UdsServer::UdsServer(const config::UdsServerConfig& cfg) : impl_(std::make_unique<Impl>(cfg, nullptr)) {}
274 : 11 : UdsServer::UdsServer(const config::UdsServerConfig& cfg, std::unique_ptr<interface::UdsAcceptorInterface> acceptor,
275 : 11 : net::io_context& ioc)
276 : 11 : : impl_(std::make_unique<Impl>(cfg, &ioc)) {
277 : 11 : impl_->acceptor_ = std::move(acceptor);
278 : 11 : }
279 : :
280 : 68 : UdsServer::~UdsServer() {
281 [ + - - + : 34 : if (impl_ && impl_->state_.get() != base::LinkState::Idle) {
- + ]
282 : 0 : impl_->stop(nullptr);
283 : : }
284 : 68 : }
285 : :
286 : 0 : UdsServer::UdsServer(UdsServer&&) noexcept = default;
287 : 0 : UdsServer& UdsServer::operator=(UdsServer&&) noexcept = default;
288 : :
289 : 33 : void UdsServer::start() {
290 [ - + ]: 39 : if (impl_->state_.get() == base::LinkState::Listening) return;
291 : :
292 : 33 : impl_->stopping_ = false;
293 : : // Restart contract (#444): stats() resets on restart. The server-level
294 : : // counters now outlive the sessions that fed them, so clearing them here is
295 : : // what keeps that promise - before absorption they were empty and a restart
296 : : // zeroed the aggregate for free.
297 : 33 : impl_->stats_.reset(0);
298 : :
299 [ + - + + ]: 33 : if (!impl_->cfg_.is_valid()) {
300 : 1 : WIRESTEAD_LOG_ERROR("uds_server", "start", "Invalid UDS server configuration or socket path");
301 : 1 : impl_->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONFIGURATION,
302 : : "start", {}, "Invalid UDS server configuration or socket path", false, 0);
303 : 1 : impl_->state_.set(base::LinkState::Error);
304 : 1 : impl_->notify_state();
305 : 1 : return;
306 : : }
307 : :
308 : : // #438: only remove an existing path at this location if it's actually a
309 : : // stale socket - never a regular file/directory (misconfiguration), and
310 : : // never a socket another live process is still listening on (that would
311 : : // otherwise be silently hijacked instead of failing loudly).
312 : 32 : std::string block_reason = existing_uds_path_blocks_bind(impl_->cfg_.socket_path);
313 [ + + ]: 32 : if (!block_reason.empty()) {
314 : 2 : std::string msg = fmt::format("Refusing to bind {}: {}", impl_->cfg_.socket_path, block_reason);
315 : 2 : WIRESTEAD_LOG_ERROR("uds_server", "start", msg);
316 : 4 : impl_->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION,
317 : 2 : "start", make_error_code(boost::system::errc::address_in_use), msg, false,
318 : : 0);
319 : 2 : impl_->state_.set(base::LinkState::Error);
320 : 2 : impl_->notify_state();
321 : 2 : return;
322 : 2 : }
323 : 30 : std::remove(impl_->cfg_.socket_path.c_str());
324 : :
325 : 30 : boost::system::error_code ec;
326 : 30 : impl_->acceptor_->open(uds(), ec);
327 [ - + ]: 30 : if (ec) {
328 : 0 : std::string msg = fmt::format("Failed to open acceptor: {}", ec.message());
329 : 0 : WIRESTEAD_LOG_ERROR("uds_server", "start", msg);
330 : 0 : impl_->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::SYSTEM, "open",
331 : : ec, msg, false, 0);
332 : 0 : impl_->state_.set(base::LinkState::Error);
333 : 0 : impl_->notify_state();
334 : 0 : return;
335 : 0 : }
336 : :
337 : 30 : uds::endpoint endpoint;
338 : : try {
339 : 30 : endpoint = uds::endpoint(impl_->cfg_.socket_path);
340 : 0 : } catch (const std::exception& e) {
341 : 0 : std::string msg = fmt::format("Invalid UDS endpoint: {}", e.what());
342 : 0 : WIRESTEAD_LOG_ERROR("uds_server", "start", msg);
343 : 0 : impl_->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONFIGURATION,
344 : 0 : "start", make_error_code(boost::system::errc::filename_too_long), msg, false,
345 : : 0);
346 : 0 : impl_->state_.set(base::LinkState::Error);
347 : 0 : impl_->notify_state();
348 : 0 : return;
349 : 0 : }
350 : :
351 : 30 : impl_->acceptor_->bind(endpoint, ec);
352 [ + + ]: 30 : if (ec) {
353 : 3 : std::string msg = fmt::format("Failed to bind to {}: {}", impl_->cfg_.socket_path, ec.message());
354 : 3 : WIRESTEAD_LOG_ERROR("uds_server", "start", msg);
355 : 3 : impl_->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION,
356 : : "bind", ec, msg, false, 0);
357 : 3 : impl_->state_.set(base::LinkState::Error);
358 : 3 : impl_->notify_state();
359 : 3 : return;
360 : 3 : }
361 : 27 : impl_->bound_.store(true);
362 : :
363 : 27 : impl_->acceptor_->listen(net::socket_base::max_listen_connections, ec);
364 [ - + ]: 27 : if (ec) {
365 : 0 : std::string msg = fmt::format("Failed to listen: {}", ec.message());
366 : 0 : WIRESTEAD_LOG_ERROR("uds_server", "start", msg);
367 : 0 : impl_->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION,
368 : : "listen", ec, msg, false, 0);
369 : 0 : impl_->state_.set(base::LinkState::Error);
370 : 0 : impl_->notify_state();
371 : 0 : return;
372 : 0 : }
373 : :
374 : : // #438: restrict local access to the socket file if requested. Best-effort
375 : : // - a chmod failure is logged but does not fail startup, since the socket
376 : : // is already bound and listening at this point.
377 [ + + ]: 27 : if (impl_->cfg_.socket_permissions != -1) {
378 : : #if !defined(WIRESTEAD_PLATFORM_WINDOWS)
379 [ - + ]: 1 : if (::chmod(impl_->cfg_.socket_path.c_str(), static_cast<mode_t>(impl_->cfg_.socket_permissions)) != 0) {
380 : 0 : WIRESTEAD_LOG_WARNING("uds_server", "start",
381 : : fmt::format("Failed to chmod socket {}: {}", impl_->cfg_.socket_path, strerror(errno)));
382 : : }
383 : : #else
384 : : WIRESTEAD_LOG_WARNING("uds_server", "start", "socket_permissions is ignored on Windows");
385 : : #endif
386 : : }
387 : :
388 : 27 : impl_->state_.set(base::LinkState::Listening);
389 : 27 : impl_->notify_state();
390 : :
391 [ + + + - : 27 : if (impl_->owns_ioc_ && !impl_->ioc_thread_.joinable()) {
+ + ]
392 [ + - - + ]: 19 : if (impl_->ioc_->stopped()) {
393 : 0 : impl_->ioc_->restart();
394 : : }
395 : 19 : impl_->work_guard_ =
396 : 38 : std::make_unique<net::executor_work_guard<net::io_context::executor_type>>(net::make_work_guard(*impl_->ioc_));
397 : 38 : impl_->ioc_thread_ = std::jthread([impl = impl_.get()](std::stop_token st) {
398 : 19 : wirestead::concurrency::run_io_thread_init();
399 : : try {
400 : 19 : std::stop_callback cb(st, [impl] { impl->ioc_->stop(); });
401 : 19 : impl->ioc_->run();
402 : 19 : } catch (...) {
403 : 0 : }
404 : 38 : });
405 : : }
406 : :
407 : 54 : net::post(impl_->ioc_->get_executor(), [self = shared_from_this()]() { self->impl_->do_accept(self); });
408 : 32 : }
409 : :
410 : 34 : void UdsServer::stop() { impl_->stop(shared_from_this()); }
411 : :
412 : 3 : bool UdsServer::is_connected() const { return impl_->state_.get() == base::LinkState::Listening; }
413 : 0 : bool UdsServer::is_backpressure_active() const { return false; }
414 : :
415 : 0 : bool UdsServer::is_backpressure_active(ClientId client_id) const {
416 : 0 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
417 : 0 : auto it = impl_->sessions_.find(client_id);
418 [ # # # # : 0 : if (it != impl_->sessions_.end() && it->second) {
# # ]
419 : 0 : return it->second->is_backpressure_active();
420 : : }
421 : 0 : return false;
422 : 0 : }
423 : :
424 : 17 : boost::asio::any_io_executor UdsServer::get_executor() { return impl_->ioc_->get_executor(); }
425 : :
426 : 1879 : wrapper::RuntimeStats UdsServer::stats() const {
427 : 1879 : auto aggregate = impl_->stats_.snapshot(0, 0, false);
428 : 1879 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
429 [ + + ]: 1885 : for (const auto& pair : impl_->sessions_) {
430 [ - + ]: 6 : if (!pair.second) continue;
431 : 6 : const auto session_stats = pair.second->stats();
432 : 6 : aggregate.bytes_accepted += session_stats.bytes_accepted;
433 : 6 : aggregate.messages_accepted += session_stats.messages_accepted;
434 : 6 : aggregate.bytes_sent += session_stats.bytes_sent;
435 : 6 : aggregate.messages_sent += session_stats.messages_sent;
436 : 6 : aggregate.bytes_received += session_stats.bytes_received;
437 : 6 : aggregate.messages_received += session_stats.messages_received;
438 : 6 : aggregate.failed_sends += session_stats.failed_sends;
439 : 6 : aggregate.dropped_messages += session_stats.dropped_messages;
440 : 6 : aggregate.dropped_bytes += session_stats.dropped_bytes;
441 : 6 : aggregate.backpressure_events += session_stats.backpressure_events;
442 : 6 : aggregate.queued_bytes += session_stats.queued_bytes;
443 : 6 : aggregate.pending_bytes += session_stats.pending_bytes;
444 : : // Peak, not a total: summing per-session peaks would report a depth no
445 : : // session ever reached, because the peaks need not have been simultaneous.
446 : 6 : aggregate.max_queued_bytes = std::max(aggregate.max_queued_bytes, session_stats.max_queued_bytes);
447 [ + - - + ]: 6 : aggregate.backpressure_active = aggregate.backpressure_active || session_stats.backpressure_active;
448 : : }
449 : 3758 : return aggregate;
450 : 1879 : }
451 : :
452 : 0 : void UdsServer::reset_stats() {
453 : 0 : impl_->stats_.reset(0);
454 : 0 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
455 [ # # ]: 0 : for (const auto& pair : impl_->sessions_) {
456 [ # # # # ]: 0 : if (pair.second) pair.second->reset_stats();
457 : : }
458 : 0 : }
459 : :
460 : 3 : std::optional<diagnostics::ErrorInfo> UdsServer::last_error_info() const {
461 : 3 : return impl_->error_info_holder_.last_error_info();
462 : : }
463 : :
464 : 1 : bool UdsServer::async_write_copy(memory::ConstByteSpan data) {
465 : 1 : auto shared_data = std::make_shared<const std::vector<uint8_t>>(data.begin(), data.end());
466 : 2 : return async_write_shared(shared_data);
467 : 1 : }
468 : :
469 : 1 : bool UdsServer::async_write_move(std::vector<uint8_t>&& data) {
470 : 1 : auto shared_data = std::make_shared<const std::vector<uint8_t>>(std::move(data));
471 : 2 : return async_write_shared(shared_data);
472 : 1 : }
473 : :
474 : 3 : bool UdsServer::async_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
475 [ + - + - : 3 : if (impl_->stopping_.load() || !data || data->empty()) {
- + - + ]
476 : 0 : impl_->stats_.record_failed_send();
477 : 0 : return false;
478 : : }
479 : 3 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
480 : 3 : bool sent = false;
481 : 3 : bool attempted = false;
482 [ - + ]: 3 : for (auto& pair : impl_->sessions_) {
483 [ # # # # : 0 : if (pair.second && pair.second->alive() && pair.second->async_write_shared(data)) {
# # # # #
# # # # #
# # ]
484 : 0 : attempted = true;
485 : 0 : sent = true;
486 [ # # # # : 0 : } else if (pair.second && pair.second->alive()) {
# # # # ]
487 : 0 : attempted = true;
488 : : }
489 : : }
490 [ + - ]: 3 : if (!attempted) impl_->stats_.record_failed_send();
491 : 3 : return sent;
492 : 3 : }
493 : :
494 : 0 : bool UdsServer::async_try_write_copy(memory::ConstByteSpan data) {
495 : 0 : auto shared_data = std::make_shared<const std::vector<uint8_t>>(data.begin(), data.end());
496 : 0 : return async_try_write_shared(shared_data);
497 : 0 : }
498 : :
499 : 0 : bool UdsServer::async_try_write_move(std::vector<uint8_t>&& data) {
500 : 0 : auto shared_data = std::make_shared<const std::vector<uint8_t>>(std::move(data));
501 : 0 : return async_try_write_shared(shared_data);
502 : 0 : }
503 : :
504 : 39 : bool UdsServer::async_try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
505 [ + - + - : 39 : if (impl_->stopping_.load() || !data || data->empty()) {
- + - + ]
506 : 0 : impl_->stats_.record_failed_send();
507 : 0 : return false;
508 : : }
509 : 39 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
510 : 39 : bool sent = false;
511 : 39 : bool attempted = false;
512 [ + + ]: 108 : for (auto& pair : impl_->sessions_) {
513 [ + - + - : 69 : if (pair.second && pair.second->alive()) {
+ - + - ]
514 : 69 : attempted = true;
515 [ + - + - ]: 69 : if (pair.second->async_try_write_shared(data)) sent = true;
516 : : }
517 : : }
518 [ + + ]: 39 : if (!attempted) impl_->stats_.record_failed_send();
519 : 39 : return sent;
520 : 39 : }
521 : :
522 : 17 : void UdsServer::on_bytes(OnBytes cb) {
523 : 17 : auto shared = interface::share_callback(std::move(cb));
524 : 17 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
525 : 17 : impl_->on_bytes_ = std::move(shared);
526 : 17 : }
527 : :
528 : 37 : void UdsServer::on_state(OnState cb) {
529 : 37 : auto shared = interface::share_callback(std::move(cb));
530 : 37 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
531 : 37 : impl_->on_state_ = std::move(shared);
532 : 37 : }
533 : :
534 : 34 : void UdsServer::on_backpressure(OnBackpressure cb) {
535 : 34 : auto shared = interface::share_callback(std::move(cb));
536 : 34 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
537 : 34 : impl_->on_bp_ = std::move(shared);
538 : 34 : }
539 : :
540 : 38 : bool UdsServer::broadcast(std::string_view message) {
541 : : auto data =
542 : 0 : std::make_shared<const std::vector<uint8_t>>(reinterpret_cast<const uint8_t*>(message.data()),
543 : 38 : reinterpret_cast<const uint8_t*>(message.data()) + message.size());
544 : 76 : return async_try_write_shared(data);
545 : 38 : }
546 : :
547 : 1 : bool UdsServer::broadcast(memory::ConstByteSpan data) {
548 : 1 : auto shared_data = std::make_shared<const std::vector<uint8_t>>(data.begin(), data.end());
549 : 2 : return async_try_write_shared(shared_data);
550 : 1 : }
551 : :
552 : 0 : bool UdsServer::send_to_client(ClientId client_id, std::string_view message) {
553 : 0 : return send_to_client(client_id,
554 : 0 : memory::ConstByteSpan(reinterpret_cast<const uint8_t*>(message.data()), message.size()));
555 : : }
556 : :
557 : 0 : bool UdsServer::send_to_client(ClientId client_id, memory::ConstByteSpan data) {
558 : 0 : std::shared_ptr<UdsServerSession> session;
559 : : {
560 : 0 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
561 : 0 : auto it = impl_->sessions_.find(client_id);
562 [ # # ]: 0 : if (it != impl_->sessions_.end()) session = it->second;
563 : 0 : }
564 [ # # ]: 0 : if (session) {
565 : 0 : return session->async_write_copy(data);
566 : : }
567 : 0 : impl_->stats_.record_failed_send();
568 : 0 : return false;
569 : 0 : }
570 : :
571 : 2 : bool UdsServer::try_send_to_client(ClientId client_id, std::string_view message) {
572 : 6 : return try_send_to_client(client_id,
573 : 6 : memory::ConstByteSpan(reinterpret_cast<const uint8_t*>(message.data()), message.size()));
574 : : }
575 : :
576 : 2 : bool UdsServer::try_send_to_client(ClientId client_id, memory::ConstByteSpan data) {
577 : 2 : std::shared_ptr<UdsServerSession> session;
578 : : {
579 : 2 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
580 : 2 : auto it = impl_->sessions_.find(client_id);
581 [ + - ]: 2 : if (it != impl_->sessions_.end()) session = it->second;
582 : 2 : }
583 [ + - ]: 2 : if (session) {
584 : 2 : return session->async_try_write_copy(data);
585 : : }
586 : 0 : impl_->stats_.record_failed_send();
587 : 0 : return false;
588 : 2 : }
589 : :
590 : 21 : size_t UdsServer::client_count() const {
591 : 21 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
592 : 42 : return impl_->sessions_.size();
593 : 21 : }
594 : :
595 : 2 : std::vector<ClientId> UdsServer::connected_clients() const {
596 : 2 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
597 : 2 : std::vector<ClientId> ids;
598 [ + - + + ]: 4 : for (const auto& pair : impl_->sessions_) ids.push_back(pair.first);
599 : 4 : return ids;
600 : 2 : }
601 : :
602 : 5 : std::optional<wrapper::RuntimeStats> UdsServer::client_stats(ClientId client_id) const {
603 : 5 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
604 : 5 : auto it = impl_->sessions_.find(client_id);
605 [ + + - + : 5 : if (it == impl_->sessions_.end() || !it->second) return std::nullopt;
+ + ]
606 : 3 : return it->second->stats();
607 : 5 : }
608 : :
609 : 0 : void UdsServer::set_client_limit(size_t max_clients) {
610 : 0 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
611 : 0 : impl_->cfg_.max_connections =
612 : 0 : static_cast<int>(std::min(max_clients, static_cast<size_t>(base::constants::MAX_MAX_CONNECTIONS)));
613 : 0 : }
614 : :
615 : 17 : void UdsServer::on_multi_connect(MultiClientConnectHandler handler) {
616 : 17 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
617 : 17 : impl_->on_multi_connect_ = std::move(handler);
618 : 17 : }
619 : :
620 : 17 : void UdsServer::on_multi_data(MultiClientDataHandler handler) {
621 : 17 : auto shared = interface::share_callback(std::move(handler));
622 : 17 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
623 : 17 : impl_->on_multi_data_ = std::move(shared);
624 : 17 : }
625 : :
626 : 17 : void UdsServer::on_multi_disconnect(MultiClientDisconnectHandler handler) {
627 : 17 : std::lock_guard<std::mutex> lock(impl_->sessions_mutex_);
628 : 17 : impl_->on_multi_disconnect_ = std::move(handler);
629 : 17 : }
630 : :
631 : 14 : base::LinkState UdsServer::state() const { return impl_->state_.get(); }
632 : :
633 : 51 : void UdsServer::Impl::do_accept(std::shared_ptr<UdsServer> self) {
634 : 51 : acceptor_->async_accept([self](const boost::system::error_code& ec, uds::socket socket) {
635 [ + - + + : 47 : if (!self || self->impl_->stopping_) return;
+ + ]
636 : :
637 [ + + ]: 24 : if (!ec) {
638 : : ClientId client_id;
639 : : {
640 : 23 : std::lock_guard<std::mutex> lock(self->impl_->sessions_mutex_);
641 [ + + - + ]: 31 : if (self->impl_->cfg_.max_connections > 0 &&
642 [ - + ]: 8 : self->impl_->sessions_.size() >= static_cast<size_t>(self->impl_->cfg_.max_connections)) {
643 : 0 : boost::system::error_code ignored;
644 : 0 : socket.close(ignored);
645 : 0 : auto* impl = self->impl_.get();
646 : 0 : impl->do_accept(self);
647 : 0 : return;
648 : : }
649 : 23 : client_id = self->impl_->next_client_id_++;
650 : 23 : }
651 : :
652 : : auto session = std::make_shared<UdsServerSession>(
653 : 23 : *self->impl_->ioc_, std::move(socket), self->impl_->cfg_.backpressure_threshold,
654 : 23 : self->impl_->cfg_.idle_timeout_ms, self->impl_->cfg_.backpressure_strategy,
655 : 46 : self->impl_->cfg_.enable_memory_pool, self->impl_->cfg_.read_buffer_size);
656 : :
657 : 23 : std::weak_ptr<UdsServer> weak_self = self;
658 : 23 : session->on_bytes([weak_self, client_id](memory::ConstByteSpan data) {
659 : 19 : auto s = weak_self.lock();
660 [ - + ]: 19 : if (!s) return;
661 : 19 : interface::SharedCallback<MultiClientDataHandler> data_handler;
662 : 19 : interface::SharedCallback<OnBytes> bytes_handler;
663 : : {
664 : 19 : std::lock_guard<std::mutex> lock(s->impl_->sessions_mutex_);
665 : 19 : data_handler = s->impl_->on_multi_data_;
666 : 19 : bytes_handler = s->impl_->on_bytes_;
667 : 19 : }
668 [ + + + - ]: 19 : if (data_handler) (*data_handler)(client_id, data);
669 [ - + - - ]: 19 : if (bytes_handler) (*bytes_handler)(data);
670 : 19 : });
671 : :
672 : 23 : session->on_close([weak_self, client_id]() {
673 : 22 : auto s = weak_self.lock();
674 [ + - + + : 22 : if (!s || s->impl_->stopping_) return;
+ + ]
675 : :
676 : 13 : MultiClientDisconnectHandler disconnect_handler;
677 : : {
678 : 13 : std::lock_guard<std::mutex> lock(s->impl_->sessions_mutex_);
679 [ - + ]: 13 : if (s->impl_->stopping_) return; // Double check inside lock
680 : : // Carry the session's totals over to the server before it goes away,
681 : : // so stats() keeps reporting what this connection did. Tied to the
682 : : // erase below, which makes it exactly once even if on_close re-fires.
683 : 13 : auto it = s->impl_->sessions_.find(client_id);
684 [ + - + - : 13 : if (it != s->impl_->sessions_.end() && it->second) {
+ - ]
685 : 13 : s->impl_->stats_.absorb(it->second->stats());
686 : : }
687 : 13 : s->impl_->sessions_.erase(client_id);
688 : 13 : disconnect_handler = s->impl_->on_multi_disconnect_;
689 : 13 : }
690 [ + + + - ]: 13 : if (disconnect_handler) disconnect_handler(client_id);
691 : 22 : });
692 : :
693 : : // alive_ must be true before the session enters sessions_, so that
694 : : // broadcast() callers who observe client_count() >= 1 are guaranteed
695 : : // to pass the alive() check inside async_try_write_shared().
696 : 23 : session->start();
697 : :
698 : : {
699 : 23 : std::lock_guard<std::mutex> lock(self->impl_->sessions_mutex_);
700 : 23 : self->impl_->sessions_[client_id] = session;
701 : 23 : }
702 : :
703 : 23 : MultiClientConnectHandler connect_handler;
704 : : {
705 : 23 : std::lock_guard<std::mutex> lock(self->impl_->sessions_mutex_);
706 : 23 : connect_handler = self->impl_->on_multi_connect_;
707 : 23 : }
708 [ + + + - : 55 : if (connect_handler) connect_handler(client_id, "UDS Client");
+ - ]
709 : :
710 : : // Continue accepting
711 : 23 : auto* impl = self->impl_.get();
712 : 23 : impl->do_accept(self);
713 : 23 : } else {
714 : 1 : auto* impl = self->impl_.get();
715 [ - + ]: 1 : if (impl->stopping_.load()) return;
716 : :
717 : : // Log only real errors, not operation_aborted
718 [ + - ]: 1 : if (ec != boost::asio::error::operation_aborted) {
719 : 1 : std::string msg = fmt::format("Accept failed: {}", ec.message());
720 : 1 : WIRESTEAD_LOG_ERROR("uds_server", "accept", msg);
721 : 1 : impl->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION,
722 : : "accept", ec, msg, true, 0);
723 : 1 : impl->state_.set(base::LinkState::Error);
724 : 1 : impl->notify_state();
725 : 1 : }
726 : :
727 [ + - ]: 1 : if (!impl->stopping_.load()) {
728 : 1 : auto timer = std::make_shared<net::steady_timer>(*impl->ioc_);
729 : 1 : timer->expires_after(std::chrono::milliseconds(100));
730 : 1 : timer->async_wait([self, timer](const boost::system::error_code&) {
731 : 1 : auto* retry_impl = self->impl_.get();
732 [ + - ]: 1 : if (!retry_impl->stopping_.load()) {
733 : 1 : retry_impl->do_accept(self);
734 : : }
735 : 1 : });
736 : 1 : }
737 : : }
738 : : });
739 : 51 : }
740 : :
741 : 68 : void UdsServer::Impl::notify_state() {
742 : 68 : interface::SharedCallback<OnState> cb;
743 : : {
744 : 68 : std::lock_guard<std::mutex> lock(sessions_mutex_);
745 : 68 : cb = on_state_;
746 : 68 : }
747 [ + + + - ]: 68 : if (cb) (*cb)(state_.get());
748 : 68 : }
749 : :
750 : : } // namespace transport
751 : : } // namespace wirestead
|
