Branch data Line data Source code
1 : : #include "wirestead/wrapper/uds_server/uds_server.hpp"
2 : :
3 : : #include <algorithm>
4 : : #include <atomic>
5 : : #include <boost/asio/executor_work_guard.hpp>
6 : : #include <boost/asio/io_context.hpp>
7 : : #include <boost/asio/steady_timer.hpp>
8 : : #include <shared_mutex>
9 : : #include <stop_token>
10 : : #include <thread>
11 : :
12 : : #include "wirestead/concurrency/io_thread_hook.hpp"
13 : : #include "wirestead/diagnostics/error_mapping.hpp"
14 : : #include "wirestead/factory/channel_factory.hpp"
15 : : #include "wirestead/framer/line_framer.hpp"
16 : : #include "wirestead/framer/packet_framer.hpp"
17 : : #include "wirestead/transport/uds/uds_server.hpp"
18 : : #include "wirestead/wrapper/callback_guard.hpp"
19 : : #include "wirestead/wrapper/error_context_builder.hpp"
20 : :
21 : : namespace wirestead {
22 : : namespace wrapper {
23 : :
24 : : struct UdsServer::Impl : public std::enable_shared_from_this<Impl> {
25 : : std::string socket_path_;
26 : : std::shared_ptr<boost::asio::io_context> external_ioc_;
27 : : std::atomic<bool> use_external_context_{false};
28 : : std::atomic<bool> manage_external_context_{false};
29 : :
30 : : mutable std::shared_mutex mutex_;
31 : : std::mutex bp_mutex_;
32 : : std::condition_variable bp_cv_;
33 : : std::shared_ptr<interface::Channel> server_;
34 : : std::vector<std::promise<bool>> pending_promises_;
35 : : std::jthread external_thread_;
36 : : std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>> work_guard_;
37 : :
38 : : std::atomic<bool> started_{false};
39 : : std::atomic<bool> is_listening_{false};
40 : 26 : std::shared_ptr<bool> alive_marker_{std::make_shared<bool>(true)};
41 : :
42 : : // Configuration
43 : : std::atomic<bool> auto_start_{false};
44 : : std::atomic<int> idle_timeout_ms_{static_cast<int>(base::constants::DEFAULT_IDLE_TIMEOUT_MS)};
45 : : std::atomic<size_t> max_clients_{0};
46 : : std::atomic<bool> client_limit_enabled_{false};
47 : : std::atomic<int> socket_permissions_{-1};
48 : : std::atomic<size_t> backpressure_threshold_{base::constants::DEFAULT_BACKPRESSURE_THRESHOLD};
49 : : std::atomic<size_t> read_buffer_size_{base::constants::DEFAULT_READ_BUFFER_SIZE};
50 : : std::atomic<base::constants::BackpressureStrategy> backpressure_strategy_{
51 : : base::constants::BackpressureStrategy::Reliable};
52 : :
53 : : ConnectionHandler client_connect_handler_{nullptr};
54 : : ConnectionHandler client_disconnect_handler_{nullptr};
55 : : // Shared snapshots: the io thread copies one out per received chunk, and a
56 : : // std::function copy allocates whenever the user handler outgrows its
57 : : // small-object buffer. See interface::SharedCallback.
58 : : interface::SharedCallback<MessageHandler> data_handler_;
59 : : interface::SharedCallback<BatchMessageHandler> data_batch_handler_;
60 : : ErrorHandler error_handler_{nullptr};
61 : : std::function<void(size_t)> on_backpressure_{nullptr};
62 : : FramerFactory framer_factory_{nullptr};
63 : : interface::SharedCallback<MessageHandler> on_message_;
64 : : interface::SharedCallback<BatchMessageHandler> on_message_batch_;
65 : :
66 : : std::unordered_map<ClientId, std::shared_ptr<framer::IFramer>> framers_;
67 : :
68 : : // Batching logic
69 : : std::vector<MessageContext> data_batch_queue_;
70 : : std::vector<MessageContext> message_batch_queue_;
71 : : std::unique_ptr<boost::asio::steady_timer> batch_timer_;
72 : : size_t max_batch_size_ = 100;
73 : 26 : std::chrono::milliseconds max_batch_latency_{1};
74 : :
75 : 16 : explicit Impl(const std::string& socket_path) : socket_path_(socket_path) {}
76 : :
77 : 6 : Impl(const std::string& socket_path, std::shared_ptr<boost::asio::io_context> external_ioc)
78 : 6 : : socket_path_(socket_path),
79 : 6 : external_ioc_(std::move(external_ioc)),
80 : 6 : use_external_context_(external_ioc_ != nullptr),
81 : 12 : manage_external_context_(false) {}
82 : :
83 : 12 : explicit Impl(std::shared_ptr<interface::Channel> channel) : socket_path_(""), server_(std::move(channel)) {
84 : : // #450: setup_internal_handlers() captures weak_from_this() - calling it
85 : : // from inside this constructor would capture an empty weak_ptr, since
86 : : // enable_shared_from_this isn't wired up until make_shared() finishes
87 : : // constructing the object. Deferred to UdsServer's own constructor,
88 : : // which runs after impl_ is a fully-formed shared_ptr<Impl>.
89 : 4 : }
90 : :
91 : 26 : ~Impl() {
92 : : try {
93 : 26 : stop();
94 : 0 : } catch (...) {
95 : 0 : }
96 : 26 : }
97 : :
98 : 63 : void fulfill_all_locked(bool value) {
99 [ + + ]: 81 : for (auto& p : pending_promises_) {
100 : : try {
101 : 18 : p.set_value(value);
102 : 0 : } catch (...) {
103 : 0 : }
104 : : }
105 : 63 : pending_promises_.clear();
106 : 63 : }
107 : :
108 : 2 : void flush_batches() {
109 : 2 : std::unique_lock<std::shared_mutex> lock(mutex_);
110 [ + - ]: 2 : if (!data_batch_queue_.empty()) {
111 : 2 : auto handler = data_batch_handler_;
112 : 2 : auto batch = std::move(data_batch_queue_);
113 : 2 : data_batch_queue_.clear();
114 [ + - ]: 2 : if (handler) {
115 : 2 : lock.unlock();
116 : 2 : detail::invoke_user_callback("uds_server", "on_data_batch", handler, batch);
117 : 2 : lock.lock();
118 : : }
119 : 2 : }
120 [ + + ]: 2 : if (!message_batch_queue_.empty()) {
121 : 1 : auto handler = on_message_batch_;
122 : 1 : auto batch = std::move(message_batch_queue_);
123 : 1 : message_batch_queue_.clear();
124 [ + - ]: 1 : if (handler) {
125 : 1 : lock.unlock();
126 : 1 : detail::invoke_user_callback("uds_server", "on_message_batch", handler, batch);
127 : 1 : lock.lock();
128 : : }
129 : 1 : }
130 [ + - ]: 2 : if (batch_timer_) {
131 : 2 : batch_timer_->cancel();
132 : : }
133 : 2 : }
134 : :
135 : 4 : bool try_send_to(ClientId client_id, std::string_view data) {
136 : 4 : std::shared_lock<std::shared_mutex> lock(mutex_);
137 : 4 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(server_);
138 [ + + + - ]: 8 : return ts ? ts->try_send_to_client(client_id, data) : false;
139 : 4 : }
140 : :
141 : 41 : bool try_broadcast(std::string_view data) {
142 : 41 : std::shared_lock<std::shared_mutex> lock(mutex_);
143 : 41 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(server_);
144 [ + + + - ]: 82 : return ts ? ts->broadcast(data) : false;
145 : 41 : }
146 : :
147 : 3 : bool send_to(ClientId client_id, std::string_view data) {
148 [ + + ]: 3 : if (backpressure_strategy_.load() == base::constants::BackpressureStrategy::Reliable)
149 : 2 : return send_to_blocking(client_id, data);
150 : 1 : return try_send_to(client_id, data);
151 : : }
152 : :
153 : 37 : bool broadcast(std::string_view data) { return try_broadcast(data); }
154 : :
155 : : // channel_->on_backpressure()/session-level backpressure callbacks call bp_cv_.notify_all()
156 : : // from the transport's io_context thread without holding bp_mutex_ - a classic lost-wakeup
157 : : // race is possible: a waiter can check the predicate, find it still blocking, and be in the
158 : : // process of registering to wait when the notify fires. Poll with a bounded timeout instead
159 : : // of an unbounded wait() so a missed notify only costs a short delay rather than a
160 : : // permanent hang (see #427, #431).
161 : : //
162 : : // Returns false without sending instead of waiting if called from the
163 : : // channel's own io thread while backpressure is active for this client -
164 : : // e.g. a blocking send_to() called from inside an on_data/on_message
165 : : // callback. Clearing backpressure requires that same io thread to make
166 : : // progress, so blocking here would deadlock forever rather than
167 : : // eventually clear (#449).
168 : : // #509: see identical rationale in wrapper/tcp_server/tcp_server.cc -
169 : : // bounded retry rather than a single attempt after the wait exits.
170 : : static constexpr int kMaxBlockingSendAttempts = 5;
171 : :
172 : 3 : bool send_to_blocking(ClientId client_id, std::string_view data) {
173 [ + - ]: 3 : for (int attempt = 0; attempt < kMaxBlockingSendAttempts; ++attempt) {
174 : 3 : std::unique_lock<std::mutex> lock(bp_mutex_);
175 : 6 : auto predicate = [this, client_id]() {
176 : 6 : std::shared_lock<std::shared_mutex> rlock(mutex_);
177 : 6 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(server_);
178 [ + - - + : 12 : return !started_.load() || !ts || !ts->is_backpressure_active(client_id);
- - - - ]
179 : 6 : };
180 [ + - - + : 3 : if (!predicate() && detail::in_data_callback()) return false;
- - - + ]
181 [ + - - + ]: 3 : while (!bp_cv_.wait_for(lock, std::chrono::milliseconds(50), predicate)) {
182 : : }
183 : 3 : lock.unlock();
184 : 3 : std::shared_lock<std::shared_mutex> rlock(mutex_);
185 : 3 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(server_);
186 [ + - ]: 3 : if (!ts) return false;
187 [ # # # # ]: 0 : if (ts->send_to_client(client_id, data)) return true;
188 : 9 : }
189 : 0 : return false;
190 : : }
191 : :
192 : 4 : void schedule_batch_timer() {
193 [ - + ]: 4 : if (!batch_timer_) return;
194 : 4 : batch_timer_->expires_after(max_batch_latency_);
195 : 8 : batch_timer_->async_wait([this, weak_impl = weak_from_this(),
196 : 4 : weak_alive = std::weak_ptr<bool>(alive_marker_)](const boost::system::error_code& ec) {
197 [ + + ]: 4 : if (ec) return;
198 : 2 : auto impl_keepalive = weak_impl.lock();
199 [ - + ]: 2 : if (!impl_keepalive) return;
200 : 2 : auto alive = weak_alive.lock();
201 [ - + ]: 2 : if (!alive) return;
202 : 2 : flush_batches();
203 : 2 : });
204 : : }
205 : :
206 : 19 : std::future<bool> start() {
207 : 19 : std::unique_lock<std::shared_mutex> lock(mutex_);
208 [ + + ]: 19 : if (is_listening_.load()) {
209 : 1 : std::promise<bool> p;
210 : 1 : p.set_value(true);
211 : 1 : return p.get_future();
212 : 1 : }
213 : 18 : std::promise<bool> p;
214 : 18 : auto f = p.get_future();
215 : 18 : pending_promises_.push_back(std::move(p));
216 [ - + ]: 18 : if (started_.exchange(true)) return f;
217 : :
218 [ + + ]: 18 : if (!server_) {
219 : 14 : config::UdsServerConfig config;
220 : 14 : config.socket_path = socket_path_;
221 : 14 : config.idle_timeout_ms = idle_timeout_ms_.load();
222 : 14 : config.max_connections =
223 : 28 : static_cast<int>(std::min(max_clients_.load(), static_cast<size_t>(base::constants::MAX_MAX_CONNECTIONS)));
224 : 14 : config.backpressure_threshold = backpressure_threshold_.load();
225 : 14 : config.read_buffer_size = read_buffer_size_.load();
226 : 14 : config.backpressure_strategy = backpressure_strategy_.load();
227 : 14 : config.socket_permissions = socket_permissions_.load();
228 : 14 : server_ = factory::ChannelFactory::create(config, external_ioc_);
229 : 14 : setup_internal_handlers();
230 : 14 : }
231 : :
232 : 18 : lock.unlock();
233 : 18 : server_->start();
234 : :
235 [ + + + - : 18 : if (use_external_context_.load() && manage_external_context_.load() && !external_thread_.joinable()) {
+ - + + ]
236 [ + - + - : 4 : if (external_ioc_ && external_ioc_->stopped()) {
+ + + + ]
237 : 1 : external_ioc_->restart();
238 : : }
239 : 8 : work_guard_ = std::make_unique<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>(
240 : 12 : boost::asio::make_work_guard(*external_ioc_));
241 : 8 : external_thread_ = std::jthread([ioc = external_ioc_](std::stop_token st) {
242 : 4 : wirestead::concurrency::run_io_thread_init();
243 : : try {
244 : 8 : std::stop_callback cb(st, [ioc] { ioc->stop(); });
245 : 4 : ioc->run();
246 : 4 : } catch (...) {
247 : 0 : }
248 : 8 : });
249 : : }
250 : :
251 : 18 : return f;
252 : 19 : }
253 : :
254 : 44 : void stop() {
255 : 44 : bool should_join = false;
256 : : {
257 : 44 : std::unique_lock<std::shared_mutex> lock(mutex_);
258 [ + + ]: 44 : if (!started_.exchange(false)) {
259 : 26 : bp_cv_.notify_all();
260 : 26 : is_listening_.store(false);
261 : 26 : fulfill_all_locked(false);
262 : 26 : return;
263 : : }
264 : 18 : bp_cv_.notify_all();
265 : :
266 [ + - ]: 18 : if (batch_timer_) {
267 : 18 : batch_timer_->cancel();
268 : 18 : batch_timer_.reset();
269 : : }
270 : :
271 [ + - ]: 18 : if (server_) {
272 : 18 : server_->on_bytes(nullptr);
273 : 18 : server_->on_state(nullptr);
274 : 18 : server_->on_backpressure(nullptr);
275 : 18 : lock.unlock();
276 : 18 : server_->stop();
277 : 18 : lock.lock();
278 : : }
279 : :
280 [ + + + - : 18 : if (use_external_context_.load() && manage_external_context_.load()) {
+ + ]
281 [ + - ]: 4 : if (work_guard_) work_guard_.reset();
282 [ + - + - ]: 4 : if (external_ioc_) external_ioc_->stop();
283 : 4 : should_join = true;
284 : : }
285 : :
286 : 18 : is_listening_.store(false);
287 : 18 : framers_.clear();
288 : 18 : fulfill_all_locked(false);
289 : 44 : }
290 : :
291 [ + + + - : 18 : if (should_join && external_thread_.joinable()) {
+ + ]
292 : : try {
293 [ + - ]: 4 : if (std::this_thread::get_id() != external_thread_.get_id()) {
294 : 4 : external_thread_.request_stop();
295 : 4 : external_thread_.join();
296 : : } else {
297 : 0 : external_thread_.detach();
298 : : }
299 : 0 : } catch (...) {
300 : 0 : }
301 : : }
302 : 18 : std::unique_lock<std::shared_mutex> lock(mutex_);
303 : 18 : server_.reset();
304 : 18 : }
305 : :
306 : 18 : void setup_internal_handlers() {
307 [ - + ]: 18 : if (!server_) return;
308 : :
309 : 18 : batch_timer_ = std::make_unique<boost::asio::steady_timer>(server_->get_executor());
310 : :
311 : 18 : std::weak_ptr<bool> weak_alive = alive_marker_;
312 : 18 : std::weak_ptr<Impl> weak_impl = weak_from_this();
313 : :
314 : 18 : auto transport_server = std::dynamic_pointer_cast<transport::UdsServer>(server_);
315 [ + + ]: 18 : if (transport_server) {
316 : 17 : transport_server->on_multi_connect([this, weak_impl, weak_alive](ClientId id, const std::string& info) {
317 : 16 : auto impl_keepalive = weak_impl.lock();
318 [ - + ]: 16 : if (!impl_keepalive) return;
319 : 16 : auto alive = weak_alive.lock();
320 [ - + ]: 16 : if (!alive) return;
321 : :
322 : 16 : ConnectionHandler handler;
323 : : {
324 : 16 : std::unique_lock<std::shared_mutex> lock(mutex_);
325 [ + + ]: 16 : if (framer_factory_) {
326 : 4 : framers_[id] = framer_factory_();
327 : 4 : attach_framer_callback(id);
328 : : }
329 : 16 : handler = client_connect_handler_;
330 : 16 : }
331 : 16 : detail::invoke_user_callback("uds_server", "on_connect", handler, ConnectionContext(id, info));
332 : 16 : });
333 : 17 : transport_server->on_multi_data([this, weak_impl, weak_alive](ClientId id, memory::ConstByteSpan data_span) {
334 : 10 : auto impl_keepalive = weak_impl.lock();
335 [ - + ]: 10 : if (!impl_keepalive) return;
336 : 10 : auto alive = weak_alive.lock();
337 [ - + ]: 10 : if (!alive) return;
338 : :
339 : : // #449: everything below runs synchronously on this io thread -
340 : : // mark it so a blocking send_to() called from within one of these
341 : : // callbacks fails fast instead of deadlocking.
342 : 10 : detail::CallbackGuard callback_guard;
343 : :
344 : : // #441: snapshot the handler/framer pointers under a shared_lock
345 : : // (not unique_lock) - this is a pure read, matching try_send's
346 : : // locking level so it no longer blocks concurrent sends even
347 : : // briefly.
348 : : bool batch_mode;
349 : 10 : interface::SharedCallback<MessageHandler> handler;
350 : 10 : std::shared_ptr<framer::IFramer> framer_to_push;
351 : : {
352 : 10 : std::shared_lock<std::shared_mutex> lock(mutex_);
353 : 10 : batch_mode = static_cast<bool>(data_batch_handler_);
354 : 10 : handler = data_handler_;
355 : 10 : auto it = framers_.find(id);
356 [ + + ]: 10 : if (it != framers_.end()) {
357 : 4 : framer_to_push = it->second;
358 : : }
359 : 10 : }
360 : :
361 [ + + ]: 10 : if (batch_mode) {
362 : : // #441: build the copy before taking the exclusive lock, so the
363 : : // lock is only held for the queue mutation itself, not the
364 : : // allocation.
365 : 2 : MessageContext ctx(id, memory::SafeDataBuffer(data_span));
366 : 2 : interface::SharedCallback<BatchMessageHandler> flush_handler;
367 : 2 : std::vector<MessageContext> batch;
368 : : {
369 : 2 : std::unique_lock<std::shared_mutex> lock(mutex_);
370 : 2 : data_batch_queue_.emplace_back(std::move(ctx));
371 [ - + ]: 2 : if (data_batch_queue_.size() >= max_batch_size_) {
372 : 0 : flush_handler = data_batch_handler_;
373 : 0 : batch = std::move(data_batch_queue_);
374 : 0 : data_batch_queue_.clear();
375 [ + - ]: 2 : } else if (data_batch_queue_.size() == 1) {
376 : 2 : schedule_batch_timer();
377 : : }
378 : 2 : }
379 : 2 : detail::invoke_user_callback("uds_server", "on_data_batch", flush_handler, batch);
380 : 2 : } else {
381 : 16 : detail::invoke_user_callback("uds_server", "on_data", handler, MessageContext(id, data_span));
382 : : }
383 : :
384 [ + + + - ]: 10 : if (framer_to_push) framer_to_push->push_bytes(data_span);
385 : 10 : });
386 : 17 : transport_server->on_multi_disconnect([this, weak_impl, weak_alive](ClientId id) {
387 : 10 : auto impl_keepalive = weak_impl.lock();
388 [ - + ]: 10 : if (!impl_keepalive) return;
389 : 10 : auto alive = weak_alive.lock();
390 [ - + ]: 10 : if (!alive) return;
391 : :
392 : 10 : ConnectionHandler handler;
393 : : {
394 : 10 : std::unique_lock<std::shared_mutex> lock(mutex_);
395 : 10 : framers_.erase(id);
396 : 10 : handler = client_disconnect_handler_;
397 : 10 : }
398 : 10 : detail::invoke_user_callback("uds_server", "on_disconnect", handler, ConnectionContext(id));
399 : 10 : });
400 : :
401 : 17 : transport_server->on_backpressure([this, weak_impl, weak_alive](size_t queued) {
402 : 0 : bp_cv_.notify_all();
403 : 0 : auto impl_keepalive = weak_impl.lock();
404 [ # # ]: 0 : if (!impl_keepalive) return;
405 : 0 : auto alive = weak_alive.lock();
406 [ # # ]: 0 : if (!alive) return;
407 : 0 : std::function<void(size_t)> handler;
408 : : {
409 : 0 : std::shared_lock<std::shared_mutex> lock(mutex_);
410 : 0 : handler = on_backpressure_;
411 : 0 : }
412 : 0 : detail::invoke_user_callback("uds_server", "on_backpressure", handler, queued);
413 : 0 : });
414 : : }
415 : :
416 : 18 : server_->on_state([this, weak_impl, weak_alive](base::LinkState state) {
417 : 19 : auto impl_keepalive = weak_impl.lock();
418 [ - + ]: 19 : if (!impl_keepalive) return;
419 : 19 : auto alive = weak_alive.lock();
420 [ - + ]: 19 : if (!alive) return;
421 : :
422 [ + + ]: 19 : if (state == base::LinkState::Listening) {
423 : 17 : is_listening_.store(true);
424 : 17 : std::unique_lock<std::shared_mutex> lock(mutex_);
425 : 17 : fulfill_all_locked(true);
426 [ - + - - : 19 : } else if (state == base::LinkState::Error || state == base::LinkState::Closed ||
- - ]
427 : : state == base::LinkState::Idle) {
428 : 2 : ErrorHandler handler;
429 : 2 : is_listening_.store(false);
430 : : {
431 : 2 : std::unique_lock<std::shared_mutex> lock(mutex_);
432 : 2 : fulfill_all_locked(false);
433 [ + - ]: 2 : if (state == base::LinkState::Error) {
434 : 2 : handler = error_handler_;
435 : : }
436 : 2 : }
437 [ + - ]: 2 : detail::invoke_user_callback("uds_server", "on_error", handler,
438 [ + - + - : 4 : server_ ? detail::build_error_context(*server_, "Server error")
- - ]
439 : : : ErrorContext(ErrorCode::IoError, "Server error"));
440 : 2 : }
441 : 19 : });
442 : 18 : }
443 : :
444 : 1879 : RuntimeStats stats() const {
445 : 1879 : std::shared_lock<std::shared_mutex> lock(mutex_);
446 [ + - + - ]: 3758 : return server_ ? server_->stats() : RuntimeStats{};
447 : 1879 : }
448 : :
449 : 0 : void reset_stats() {
450 : 0 : std::shared_lock<std::shared_mutex> lock(mutex_);
451 [ # # # # ]: 0 : if (server_) server_->reset_stats();
452 : 0 : }
453 : :
454 : 4 : void attach_framer_callback(ClientId id) {
455 : 4 : auto it = framers_.find(id);
456 [ - + ]: 4 : if (it == framers_.end()) return;
457 : :
458 : 4 : it->second->on_message([this, id](memory::ConstByteSpan msg) {
459 : : // #441: snapshot under a shared_lock (pure read), build the copy
460 : : // before taking the exclusive lock for queue mutation.
461 : : bool batch_mode;
462 : 5 : interface::SharedCallback<MessageHandler> handler;
463 : : {
464 : 5 : std::shared_lock<std::shared_mutex> lock(mutex_);
465 : 5 : batch_mode = static_cast<bool>(on_message_batch_);
466 : 5 : handler = on_message_;
467 : 5 : }
468 : :
469 [ + + ]: 5 : if (batch_mode) {
470 : 3 : MessageContext ctx(id, memory::SafeDataBuffer(msg));
471 : 3 : interface::SharedCallback<BatchMessageHandler> flush_handler;
472 : 3 : std::vector<MessageContext> batch;
473 : : {
474 : 3 : std::unique_lock<std::shared_mutex> lock(mutex_);
475 : 3 : message_batch_queue_.emplace_back(std::move(ctx));
476 [ + + ]: 3 : if (message_batch_queue_.size() >= max_batch_size_) {
477 : 1 : flush_handler = on_message_batch_;
478 : 1 : batch = std::move(message_batch_queue_);
479 : 1 : message_batch_queue_.clear();
480 [ + - ]: 2 : } else if (message_batch_queue_.size() == 1) {
481 : 2 : schedule_batch_timer();
482 : : }
483 : 3 : }
484 : 3 : detail::invoke_user_callback("uds_server", "on_message_batch", flush_handler, batch);
485 : 3 : return;
486 : 3 : }
487 : :
488 : 2 : detail::invoke_user_callback("uds_server", "on_message", handler, MessageContext(id, msg));
489 : 5 : });
490 : : }
491 : : };
492 : :
493 : 16 : UdsServer::UdsServer(const std::string& socket_path) : impl_(std::make_shared<Impl>(socket_path)) {}
494 : :
495 : 6 : UdsServer::UdsServer(const std::string& socket_path, std::shared_ptr<boost::asio::io_context> external_ioc)
496 : 6 : : impl_(std::make_shared<Impl>(socket_path, std::move(external_ioc))) {}
497 : :
498 : 4 : UdsServer::UdsServer(std::shared_ptr<interface::Channel> channel) : impl_(std::make_shared<Impl>(std::move(channel))) {
499 : 4 : impl_->setup_internal_handlers();
500 : 4 : }
501 : :
502 : 36 : UdsServer::~UdsServer() = default;
503 : :
504 : 0 : UdsServer::UdsServer(UdsServer&&) noexcept = default;
505 : 0 : UdsServer& UdsServer::operator=(UdsServer&&) noexcept = default;
506 : :
507 : 19 : std::future<bool> UdsServer::start() { return impl_->start(); }
508 : :
509 : 18 : void UdsServer::stop() { impl_->stop(); }
510 : :
511 : 19 : bool UdsServer::listening() const { return impl_->is_listening_.load(); }
512 : 1879 : RuntimeStats UdsServer::stats() const { return impl_->stats(); }
513 : 0 : void UdsServer::reset_stats() { impl_->reset_stats(); }
514 : :
515 : 37 : bool UdsServer::broadcast(std::string_view data) { return impl_->broadcast(data); }
516 : 4 : bool UdsServer::try_broadcast(std::string_view data) { return impl_->try_broadcast(data); }
517 : 3 : bool UdsServer::send_to(ClientId client_id, std::string_view data) { return impl_->send_to(client_id, data); }
518 : 3 : bool UdsServer::try_send_to(ClientId client_id, std::string_view data) { return impl_->try_send_to(client_id, data); }
519 : :
520 : 1 : bool UdsServer::send_to_blocking(ClientId client_id, std::string_view data) {
521 : 1 : return impl_->send_to_blocking(client_id, data);
522 : : }
523 : :
524 : 6 : bool UdsServer::broadcast_line(std::string_view line) { return broadcast(std::string(line) + "\n"); }
525 : 2 : bool UdsServer::send_to_line(ClientId client_id, std::string_view line) {
526 : 6 : return send_to(client_id, std::string(line) + "\n");
527 : : }
528 : 6 : bool UdsServer::try_broadcast_line(std::string_view line) { return try_broadcast(std::string(line) + "\n"); }
529 : 2 : bool UdsServer::try_send_to_line(ClientId client_id, std::string_view line) {
530 : 6 : return try_send_to(client_id, std::string(line) + "\n");
531 : : }
532 : :
533 : 8 : UdsServer& UdsServer::on_connect(ConnectionHandler handler) {
534 : 8 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
535 : 8 : impl_->client_connect_handler_ = std::move(handler);
536 : 8 : return *this;
537 : 8 : }
538 : :
539 : 3 : UdsServer& UdsServer::on_disconnect(ConnectionHandler handler) {
540 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
541 : 3 : impl_->client_disconnect_handler_ = std::move(handler);
542 : 3 : return *this;
543 : 3 : }
544 : :
545 : 7 : UdsServer& UdsServer::on_data(MessageHandler handler) {
546 : 7 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
547 : 7 : impl_->data_handler_ = interface::share_callback(std::move(handler));
548 : 7 : return *this;
549 : 7 : }
550 : :
551 : 3 : UdsServer& UdsServer::on_data_batch(BatchMessageHandler handler) {
552 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
553 : 3 : impl_->data_batch_handler_ = interface::share_callback(std::move(handler));
554 : 3 : return *this;
555 : 3 : }
556 : :
557 : 10 : UdsServer& UdsServer::on_error(ErrorHandler handler) {
558 : 10 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
559 : 10 : impl_->error_handler_ = std::move(handler);
560 : 10 : return *this;
561 : 10 : }
562 : :
563 : 2 : UdsServer& UdsServer::on_backpressure(std::function<void(size_t)> handler) {
564 : 2 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
565 : 2 : impl_->on_backpressure_ = std::move(handler);
566 : 2 : return *this;
567 : 2 : }
568 : :
569 : 7 : UdsServer& UdsServer::framer(FramerFactory factory) {
570 : 7 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
571 : 7 : impl_->framer_factory_ = std::move(factory);
572 : 7 : return *this;
573 : 7 : }
574 : :
575 : 4 : UdsServer& UdsServer::on_message(MessageHandler handler) {
576 : 4 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
577 : 4 : impl_->on_message_ = interface::share_callback(std::move(handler));
578 : 4 : return *this;
579 : 4 : }
580 : :
581 : 3 : UdsServer& UdsServer::on_message_batch(BatchMessageHandler handler) {
582 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
583 : 3 : impl_->on_message_batch_ = interface::share_callback(std::move(handler));
584 : 3 : return *this;
585 : 3 : }
586 : :
587 : 20 : size_t UdsServer::client_count() const {
588 : 20 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
589 : 20 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(impl_->server_);
590 [ + + + - ]: 40 : return ts ? ts->client_count() : 0;
591 : 20 : }
592 : :
593 : 3 : std::vector<ClientId> UdsServer::connected_clients() const {
594 : 3 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
595 : 3 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(impl_->server_);
596 [ + + + - ]: 6 : return ts ? ts->connected_clients() : std::vector<ClientId>{};
597 : 3 : }
598 : :
599 : 5 : std::optional<RuntimeStats> UdsServer::client_stats(ClientId client_id) const {
600 : 5 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
601 : 5 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(impl_->server_);
602 [ + - + - ]: 10 : return ts ? ts->client_stats(client_id) : std::nullopt;
603 : 5 : }
604 : :
605 : 1 : UdsServer& UdsServer::auto_start(bool manage) {
606 : 1 : impl_->auto_start_.store(manage);
607 [ + - + - : 1 : if (impl_->auto_start_.load() && !impl_->started_.load()) {
+ - ]
608 : 1 : start();
609 : : }
610 : 1 : return *this;
611 : : }
612 : :
613 : 3 : UdsServer& UdsServer::idle_timeout(std::chrono::milliseconds timeout) {
614 : 3 : impl_->idle_timeout_ms_.store(static_cast<int>(timeout.count()));
615 : 3 : return *this;
616 : : }
617 : :
618 : 5 : UdsServer& UdsServer::max_clients(size_t max) {
619 : 5 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
620 : 5 : impl_->max_clients_.store(max);
621 [ - + ]: 5 : if (max == 0) {
622 : 0 : impl_->client_limit_enabled_.store(false);
623 : : } else {
624 : 5 : impl_->client_limit_enabled_.store(true);
625 : : }
626 : 5 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(impl_->server_);
627 [ - + - - ]: 5 : if (ts) ts->set_client_limit(max);
628 : 5 : return *this;
629 : 5 : }
630 : :
631 : 0 : UdsServer& UdsServer::socket_permissions(int mode) {
632 : 0 : impl_->socket_permissions_.store(mode);
633 : 0 : return *this;
634 : : }
635 : :
636 : 13 : UdsServer& UdsServer::backpressure_threshold(size_t threshold) {
637 : 13 : impl_->backpressure_threshold_.store(threshold);
638 : 13 : return *this;
639 : : }
640 : :
641 : 0 : UdsServer& UdsServer::read_buffer_size(size_t bytes) {
642 : 0 : impl_->read_buffer_size_.store(bytes);
643 : 0 : return *this;
644 : : }
645 : :
646 : 5 : UdsServer& UdsServer::backpressure_strategy(base::constants::BackpressureStrategy strategy) {
647 : 5 : impl_->backpressure_strategy_.store(strategy);
648 : 5 : return *this;
649 : : }
650 : :
651 : 6 : size_t UdsServer::backpressure_threshold() const { return impl_->backpressure_threshold_.load(); }
652 : :
653 : 3 : base::constants::BackpressureStrategy UdsServer::backpressure_strategy() const {
654 : 3 : return impl_->backpressure_strategy_.load();
655 : : }
656 : :
657 : 6 : UdsServer& UdsServer::manage_external_context(bool manage) {
658 : 6 : impl_->manage_external_context_.store(manage);
659 : 6 : return *this;
660 : : }
661 : :
662 : 3 : UdsServer& UdsServer::batch_size(size_t size) {
663 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
664 : 3 : impl_->max_batch_size_ = size;
665 : 3 : return *this;
666 : 3 : }
667 : :
668 : 3 : UdsServer& UdsServer::batch_latency(std::chrono::milliseconds latency) {
669 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
670 : 3 : impl_->max_batch_latency_ = latency;
671 : 3 : return *this;
672 : 3 : }
673 : :
674 : : } // namespace wrapper
675 : : } // namespace wirestead
|
