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 <boost/asio/strand.hpp>
9 : : #include <shared_mutex>
10 : : #include <stop_token>
11 : : #include <thread>
12 : :
13 : : #include "wirestead/concurrency/io_thread_hook.hpp"
14 : : #include "wirestead/config/validation.hpp"
15 : : #include "wirestead/diagnostics/error_mapping.hpp"
16 : : #include "wirestead/factory/channel_factory.hpp"
17 : : #include "wirestead/framer/line_framer.hpp"
18 : : #include "wirestead/framer/packet_framer.hpp"
19 : : #include "wirestead/transport/uds/uds_server.hpp"
20 : : #include "wirestead/util/input_validator.hpp"
21 : : #include "wirestead/wrapper/callback_guard.hpp"
22 : : #include "wirestead/wrapper/error_context_builder.hpp"
23 : : #include "wirestead/wrapper/send_retry.hpp"
24 : : #include "wirestead/wrapper/send_validation.hpp"
25 : : #include "wirestead/wrapper/session_batch.hpp"
26 : :
27 : : namespace wirestead {
28 : : namespace wrapper {
29 : :
30 : : struct UdsServer::Impl : public std::enable_shared_from_this<Impl> {
31 : : std::string socket_path_;
32 : : std::shared_ptr<boost::asio::io_context> external_ioc_;
33 : : std::atomic<bool> use_external_context_{false};
34 : : std::atomic<bool> manage_external_context_{false};
35 : :
36 : : mutable std::shared_mutex mutex_;
37 : : std::mutex bp_mutex_;
38 : : std::condition_variable bp_cv_;
39 : : // D-1: admission gate for this object's user callbacks. Admission, the
40 : : // running count and the closed flag are one decision, so a stop() cannot
41 : : // observe an empty gate while a callback is about to start, and a callback
42 : : // left over from a previous run is refused after a restart.
43 : : detail::CallbackGate callback_gate_;
44 : : std::mutex stop_finalize_mutex_;
45 : : bool injected_channel_ = false;
46 : : std::atomic<uint64_t> callback_generation_{0};
47 : : bool stop_requested_ = false;
48 : : std::atomic<unsigned> stop_callers_{0};
49 : :
50 : : // True when this thread is one the target's shutdown needs: a callback of
51 : : // this object, or any thread currently running the external io_context this
52 : : // channel was built on (which a callback of another channel sharing it is).
53 : : // Such a caller requests the shutdown and returns; it cannot wait for work
54 : : // its own thread has to perform.
55 : 583 : bool shutdown_needs_this_thread() const {
56 [ + - + + ]: 583 : if (callback_gate_.active_on_this_thread()) return true;
57 [ + + - + : 579 : if (external_ioc_ && external_ioc_->get_executor().running_in_this_thread()) return true;
+ + - + ]
58 : 579 : std::shared_lock<std::shared_mutex> lock(mutex_);
59 [ + + ]: 579 : if (!server_) return false;
60 : 430 : const auto executor = server_->get_executor();
61 : : using IoExecutor = boost::asio::io_context::executor_type;
62 [ + - + + ]: 430 : if (auto* io = executor.target<IoExecutor>()) return io->running_in_this_thread();
63 [ + - + - ]: 419 : if (auto* strand = executor.target<boost::asio::strand<IoExecutor>>())
64 : 419 : return strand->get_inner_executor().running_in_this_thread();
65 : 0 : return false;
66 : 579 : }
67 : :
68 : : std::shared_ptr<interface::Channel> server_;
69 : : std::vector<std::promise<bool>> pending_promises_;
70 : : std::jthread external_thread_;
71 : : std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>> work_guard_;
72 : :
73 : : std::atomic<bool> started_{false};
74 : : std::atomic<bool> is_listening_{false};
75 : 170 : std::shared_ptr<bool> alive_marker_{std::make_shared<bool>(true)};
76 : :
77 : : // Configuration
78 : : std::atomic<bool> auto_start_{false};
79 : : std::atomic<int> idle_timeout_ms_{static_cast<int>(base::constants::DEFAULT_IDLE_TIMEOUT_MS)};
80 : : std::atomic<size_t> max_clients_{0};
81 : : std::atomic<bool> client_limit_enabled_{false};
82 : : std::atomic<int> socket_permissions_{-1};
83 : : std::atomic<size_t> backpressure_threshold_{base::constants::DEFAULT_BACKPRESSURE_THRESHOLD};
84 : : std::atomic<size_t> read_buffer_size_{base::constants::DEFAULT_READ_BUFFER_SIZE};
85 : : std::atomic<base::constants::BackpressureStrategy> backpressure_strategy_{
86 : : base::constants::BackpressureStrategy::Reliable};
87 : :
88 : : ConnectionHandler client_connect_handler_{nullptr};
89 : : ConnectionHandler client_disconnect_handler_{nullptr};
90 : : // Shared snapshots: the io thread copies one out per received chunk, and a
91 : : // std::function copy allocates whenever the user handler outgrows its
92 : : // small-object buffer. See interface::SharedCallback.
93 : : interface::SharedCallback<MessageHandler> data_handler_;
94 : : interface::SharedCallback<BatchMessageHandler> data_batch_handler_;
95 : : ErrorHandler error_handler_{nullptr};
96 : : std::function<void(size_t)> on_backpressure_{nullptr};
97 : : FramerFactory framer_factory_{nullptr};
98 : : interface::SharedCallback<MessageHandler> on_message_;
99 : : interface::SharedCallback<BatchMessageHandler> on_message_batch_;
100 : :
101 : : std::unordered_map<ClientId, std::shared_ptr<framer::IFramer>> framers_;
102 : :
103 : : // Batching logic
104 : : bool terminal_error_reported_ = false;
105 : 123 : void require_stopped_config() const {
106 [ + + + - : 242 : if (started_.load() || stop_callers_.load() || (stop_requested_ && alive_marker_) || detail::in_data_callback())
+ + + - -
+ + + ]
107 : 4 : throw std::logic_error("configuration requires completed stop");
108 : 119 : }
109 : : ReceiveLimits receive_limits_;
110 : : std::shared_ptr<detail::ReceiveBudget> receive_budget_{std::make_shared<detail::ReceiveBudget>(receive_limits_)};
111 : : std::unordered_map<ClientId, std::shared_ptr<detail::SessionBatch>> batches_;
112 : : size_t max_batch_size_ = 100;
113 : 170 : std::chrono::milliseconds max_batch_latency_{1};
114 : :
115 : 63 : explicit Impl(const std::string& socket_path) : socket_path_(socket_path) {}
116 : :
117 : 8 : Impl(const std::string& socket_path, std::shared_ptr<boost::asio::io_context> external_ioc)
118 : 8 : : socket_path_(socket_path),
119 : 8 : external_ioc_(std::move(external_ioc)),
120 : 8 : use_external_context_(external_ioc_ != nullptr),
121 : 16 : manage_external_context_(false) {}
122 : :
123 : 297 : explicit Impl(std::shared_ptr<interface::Channel> channel) : socket_path_(""), server_(std::move(channel)) {
124 : 99 : injected_channel_ = true;
125 : : // #450: setup_internal_handlers() captures weak_from_this() - calling it
126 : : // from inside this constructor would capture an empty weak_ptr, since
127 : : // enable_shared_from_this isn't wired up until make_shared() finishes
128 : : // constructing the object. Deferred to UdsServer's own constructor,
129 : : // which runs after impl_ is a fully-formed shared_ptr<Impl>.
130 : 99 : }
131 : :
132 : 170 : ~Impl() {
133 : : try {
134 : 170 : stop();
135 : 0 : } catch (...) {
136 : 0 : }
137 : 170 : }
138 : :
139 : 757 : void fulfill_all_locked(bool value) {
140 [ + + ]: 928 : for (auto& p : pending_promises_) {
141 : : try {
142 : 171 : p.set_value(value);
143 : 0 : } catch (...) {
144 : 0 : }
145 : : }
146 : 757 : pending_promises_.clear();
147 : 757 : }
148 : :
149 : 108 : void flush_batches(uint64_t generation, ClientId id, const std::shared_ptr<detail::SessionBatch>& state) {
150 : 108 : auto lease = callback_gate_.enter(generation);
151 [ - + ]: 108 : if (!lease.admitted()) return;
152 : 108 : std::unique_lock<std::shared_mutex> lock(mutex_);
153 : 108 : auto found = batches_.find(id);
154 [ + - - + : 108 : if (found == batches_.end() || found->second != state) return;
- + ]
155 : 108 : state->scheduled = false;
156 : 108 : auto data = std::move(state->data);
157 : 108 : auto messages = std::move(state->messages);
158 : 108 : state->data.clear();
159 : 108 : state->messages.clear();
160 : 108 : auto data_handler = data_batch_handler_;
161 : 108 : auto message_handler = on_message_batch_;
162 : 108 : lock.unlock();
163 [ + + + - ]: 108 : if (!data.empty()) detail::invoke_user_callback("uds_server", "on_data_batch", data_handler, data);
164 : : // A callback may stop the server. Recheck admission before another callback.
165 : 108 : auto next = callback_gate_.enter(generation);
166 [ + - + + : 108 : if (next.admitted() && !messages.empty())
+ + ]
167 : 4 : detail::invoke_user_callback("uds_server", "on_message_batch", message_handler, messages);
168 : 108 : }
169 : :
170 : : // Caller holds mutex_. Final admission rechecks the native lifecycle.
171 : 545 : SendResult send_state(const std::shared_ptr<transport::UdsServer>& ts) {
172 [ + + ]: 1090 : if (stop_callers_.load() != 0) return SendResult::reject(SendRejection::Stopping);
173 [ + + ]: 544 : if (!started_.load()) {
174 [ + + ]: 44 : if (stop_requested_) {
175 [ - + ]: 26 : if (!callback_gate_.idle()) return SendResult::reject(SendRejection::Stopping);
176 [ + - ]: 26 : if (ts) {
177 : 26 : const auto state = ts->target_state();
178 [ + - + + : 26 : if (!state.accepted() && state.reason() == SendRejection::Stopping) return state;
+ + ]
179 : : }
180 : : }
181 : 33 : return SendResult::reject(SendRejection::NotStarted);
182 : : }
183 [ + + ]: 500 : if (!ts) return SendResult::reject(SendRejection::NotReady);
184 : 495 : return SendResult::accept();
185 : : }
186 : :
187 : 94 : SendResult try_send_to(ClientId client_id, std::string_view data, bool best_effort_send = false) {
188 : 94 : std::shared_lock<std::shared_mutex> lock(mutex_);
189 : 94 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(server_);
190 : 0 : const auto result = [&]() -> SendResult {
191 : : auto validation =
192 [ + + + - ]: 94 : detail::validate_payload_size(data.size(), ts ? ts->write_queue_limit(client_id) : std::nullopt);
193 [ + + ]: 94 : if (!validation.accepted()) return validation;
194 : 83 : const auto state = send_state(ts);
195 [ + + ]: 83 : if (!state.accepted()) return state;
196 : 114 : auto admitted = ts->write_target(
197 : 57 : client_id, memory::ConstByteSpan(reinterpret_cast<const uint8_t*>(data.data()), data.size()), true);
198 [ + + + + : 57 : if (best_effort_send && !admitted.accepted() && admitted.reason() == SendRejection::WouldBlock)
+ + + + ]
199 : 2 : return SendResult::reject(SendRejection::QueueFull);
200 : 55 : return admitted;
201 : 94 : }();
202 : 94 : lock.unlock();
203 [ + + + - ]: 94 : if (auto hook = detail::g_uds_server_send_result_hook.load()) hook(result);
204 : 94 : return result;
205 : 94 : }
206 : :
207 : 128 : FanoutResult try_broadcast(std::string_view data, bool append_newline = false) {
208 : 128 : std::shared_lock<std::shared_mutex> lock(mutex_);
209 : 128 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(server_);
210 [ + + + - : 128 : return ts ? ts->broadcast_result({reinterpret_cast<const uint8_t*>(data.data()), data.size()}, send_state(ts),
+ - ]
211 : : append_newline)
212 : 140 : : FanoutResult{};
213 : 128 : }
214 : :
215 : 166 : SendResult send_to(ClientId client_id, std::string_view data) {
216 [ + + ]: 166 : if (backpressure_strategy_.load() == base::constants::BackpressureStrategy::Reliable)
217 : 148 : return send_to_blocking(client_id, data);
218 : 18 : return try_send_to(client_id, data, true);
219 : : }
220 : :
221 : 78 : FanoutResult broadcast(std::string_view data) { return try_broadcast(data); }
222 : :
223 : : // channel_->on_backpressure()/session-level backpressure callbacks call bp_cv_.notify_all()
224 : : // from the transport's io_context thread without holding bp_mutex_ - a classic lost-wakeup
225 : : // race is possible: a waiter can check the predicate, find it still blocking, and be in the
226 : : // process of registering to wait when the notify fires. Poll with a bounded timeout instead
227 : : // of an unbounded wait() so a missed notify only costs a short delay rather than a
228 : : // permanent hang (see #427, #431).
229 : : //
230 : : // Returns false without sending instead of waiting if called from the
231 : : // channel's own io thread while backpressure is active for this client -
232 : : // e.g. a blocking send_to() called from inside an on_data/on_message
233 : : // callback. Clearing backpressure requires that same io thread to make
234 : : // progress, so blocking here would deadlock forever rather than
235 : : // eventually clear (#449).
236 : : // #509: see identical rationale in wrapper/tcp_server/tcp_server.cc -
237 : : // capacity retry rather than a single attempt after the wait exits.
238 : :
239 : 249 : SendResult send_to_blocking(ClientId client_id, std::string_view data) {
240 : 249 : std::shared_ptr<transport::UdsServer> ts;
241 : 249 : std::shared_ptr<transport::UdsServerSession> session;
242 : 249 : uint64_t generation = 0;
243 : 249 : bool cannot_wait = false;
244 : 0 : const auto result = [&]() -> SendResult {
245 : : {
246 : 249 : std::shared_lock<std::shared_mutex> lock(mutex_);
247 : 249 : ts = std::dynamic_pointer_cast<transport::UdsServer>(server_);
248 : : auto validation =
249 [ + + + - ]: 249 : detail::validate_payload_size(data.size(), ts ? ts->write_queue_limit(client_id) : std::nullopt);
250 [ + + ]: 249 : if (!validation.accepted()) return validation;
251 : 229 : auto state = send_state(ts);
252 [ + + ]: 229 : if (!state.accepted()) return state;
253 : 210 : state = ts->target_state();
254 [ - + ]: 210 : if (!state.accepted()) return state;
255 : 210 : generation = callback_generation_.load();
256 [ + + + - : 210 : cannot_wait = detail::in_data_callback() || detail::executor_running_here(ts->get_executor());
+ - + + +
+ - - ]
257 : 210 : session = ts->capture_target(client_id);
258 [ + + ]: 210 : if (!session) return SendResult::reject(SendRejection::NotReady);
259 : 249 : }
260 : 211 : auto wait = [&]() -> SendResult {
261 : 211 : std::unique_lock<std::mutex> bp_lock(bp_mutex_);
262 : : {
263 : 211 : std::shared_lock<std::shared_mutex> lock(mutex_);
264 : : // No observed pressure means no capacity wait. Final admission still
265 : : // checks lifecycle and the exact session selected at entry.
266 [ + + + - : 631 : if (!started_.load() || callback_generation_.load() != generation ||
+ + ]
267 : 421 : ts->poll_target_wait(session).has_value())
268 : 96 : return SendResult::accept();
269 : 211 : }
270 [ + + ]: 115 : if (cannot_wait) return SendResult::reject(SendRejection::WouldBlock);
271 [ + + + - ]: 45 : if (auto hook = detail::g_uds_server_capacity_wait_hook.load()) hook();
272 : 45 : std::optional<SendResult> outcome;
273 : 86 : auto released = [&] {
274 : : // The retained session owns its first terminal cause, even after
275 : : // removal from the map or a later server/wrapper restart.
276 : 86 : outcome = ts->poll_target_wait(session);
277 : 86 : return outcome.has_value();
278 : 45 : };
279 [ + - + + ]: 61 : while (!bp_cv_.wait_for(bp_lock, std::chrono::milliseconds(50), released)) {
280 : : }
281 [ + + + - ]: 45 : if (auto hook = detail::g_uds_server_capacity_wait_result_hook.load()) hook(*outcome);
282 : 45 : return *outcome;
283 : 211 : };
284 : 199 : for (bool retry = false;; retry = true) {
285 [ + + + - ]: 211 : if (retry) detail::pause_send_retry(bp_cv_, bp_mutex_);
286 : 211 : const auto released = wait();
287 [ + + ]: 316 : if (!released.accepted()) return released;
288 : 117 : std::shared_lock<std::shared_mutex> lock(mutex_);
289 : 117 : const auto state = send_state(ts);
290 [ + + ]: 117 : if (!state.accepted()) return state;
291 [ + + ]: 224 : if (callback_generation_.load() != generation) return SendResult::reject(SendRejection::NotReady);
292 : 216 : const auto admitted = ts->write_target(
293 : 108 : client_id, memory::ConstByteSpan(reinterpret_cast<const uint8_t*>(data.data()), data.size()), false,
294 : : session);
295 [ + + + + : 108 : if (admitted.accepted() || admitted.reason() != SendRejection::WouldBlock) return admitted;
+ + ]
296 [ + + ]: 21 : if (cannot_wait) return admitted;
297 : 129 : }
298 : 249 : }();
299 [ + + + - ]: 249 : if (auto hook = detail::g_uds_server_send_result_hook.load()) hook(result);
300 : 249 : return result;
301 : 249 : }
302 : :
303 : : // Caller holds mutex_; do not postpone an existing deadline when the
304 : : // other queue receives its first item.
305 : 15 : void schedule_batch_timer(uint64_t generation, ClientId id) {
306 : 15 : auto found = batches_.find(id);
307 [ + - + + : 15 : if (found == batches_.end() || found->second->scheduled) return;
+ + ]
308 : 14 : auto state = found->second;
309 : 14 : state->scheduled = true;
310 : 14 : state->timer.expires_after(max_batch_latency_);
311 : 14 : state->timer.async_wait(
312 : 14 : [weak_impl = weak_from_this(), generation, id,
313 : : weak_state = std::weak_ptr<detail::SessionBatch>(state)](const boost::system::error_code& ec) {
314 [ + + ]: 11 : if (ec) return;
315 : 7 : auto self = weak_impl.lock();
316 : 7 : auto batch = weak_state.lock();
317 [ + - + - : 7 : if (self && batch) self->flush_batches(generation, id, batch);
+ - + - ]
318 : 7 : });
319 : 14 : }
320 : :
321 : 173 : std::future<bool> start() {
322 : 173 : std::unique_lock<std::shared_mutex> lock(mutex_);
323 : 173 : stop_requested_ = false;
324 [ + + ]: 173 : if (is_listening_.load()) {
325 : 2 : started_.store(true);
326 : 2 : std::promise<bool> p;
327 : 2 : p.set_value(true);
328 : 2 : return p.get_future();
329 : 2 : }
330 : 171 : std::promise<bool> p;
331 : 171 : auto f = p.get_future();
332 : 171 : pending_promises_.push_back(std::move(p));
333 [ - + ]: 171 : if (started_.exchange(true)) return f;
334 : :
335 [ + + + - ]: 171 : if (!alive_marker_) alive_marker_ = std::make_shared<bool>(true);
336 [ + + ]: 171 : if (!server_) {
337 : 63 : config::UdsServerConfig config;
338 : 63 : config.socket_path = socket_path_;
339 : 63 : config.idle_timeout_ms = idle_timeout_ms_.load();
340 : 63 : config.max_connections =
341 : 126 : static_cast<int>(std::min(max_clients_.load(), static_cast<size_t>(base::constants::MAX_MAX_CONNECTIONS)));
342 : 63 : config.backpressure_threshold = backpressure_threshold_.load();
343 : 63 : config.read_buffer_size = read_buffer_size_.load();
344 : 63 : config.backpressure_strategy = backpressure_strategy_.load();
345 : 63 : config.socket_permissions = socket_permissions_.load();
346 : 63 : server_ = factory::ChannelFactory::create(config, external_ioc_);
347 : 63 : }
348 : 171 : setup_internal_handlers();
349 : :
350 : 171 : auto channel_copy = server_;
351 : 171 : lock.unlock();
352 : 171 : channel_copy->start();
353 [ - + ]: 171 : if (!started_.load()) return f;
354 : :
355 [ + + + + : 171 : if (use_external_context_.load() && manage_external_context_.load() && !external_thread_.joinable()) {
+ - + + ]
356 [ + - + - : 4 : if (external_ioc_ && external_ioc_->stopped()) {
+ + + + ]
357 : 1 : external_ioc_->restart();
358 : : }
359 : 8 : work_guard_ = std::make_unique<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>(
360 : 12 : boost::asio::make_work_guard(*external_ioc_));
361 : 8 : external_thread_ = std::jthread([ioc = external_ioc_](std::stop_token st) {
362 : 4 : wirestead::concurrency::run_io_thread_init();
363 : : try {
364 : 4 : std::stop_callback cb(st, [ioc] { ioc->stop(); });
365 : 4 : ioc->run();
366 : 4 : } catch (...) {
367 : 0 : }
368 : 8 : });
369 : : }
370 : :
371 : 171 : return f;
372 : 173 : }
373 : :
374 : 583 : void stop() {
375 : 583 : stop_callers_.fetch_add(1);
376 : : struct StopCall {
377 : : std::atomic<unsigned>& count;
378 : 583 : ~StopCall() { count.fetch_sub(1); }
379 : 583 : } stop_call{stop_callers_};
380 : 583 : const bool request_only = shutdown_needs_this_thread();
381 : 583 : callback_gate_.close();
382 : 583 : std::shared_ptr<interface::Channel> channel;
383 : : {
384 : 583 : std::unique_lock<std::shared_mutex> lock(mutex_);
385 : 583 : stop_requested_ = true;
386 [ + + + - ]: 583 : if (auto ts = std::dynamic_pointer_cast<transport::UdsServer>(server_)) ts->cancel_target_waits();
387 : 583 : started_.store(false);
388 : 583 : is_listening_.store(false);
389 : 583 : bp_cv_.notify_all();
390 : 583 : fulfill_all_locked(false);
391 : 583 : channel = server_;
392 : 583 : }
393 : : // Do not hold wrapper locks while the transport waits for callbacks.
394 [ + + + - ]: 583 : if (channel) channel->stop();
395 [ + + ]: 583 : if (request_only) return;
396 : :
397 : 550 : callback_gate_.wait_until_idle();
398 : : // All outside callers pass this lock, including those that found the
399 : : // channel already cleared. Joining and final state reset happen once.
400 : 550 : std::lock_guard<std::mutex> finalize_lock(stop_finalize_mutex_);
401 : : {
402 : 550 : std::unique_lock<std::shared_mutex> lock(mutex_);
403 : 550 : alive_marker_.reset();
404 : 550 : batches_.clear();
405 [ + + ]: 550 : if (server_) {
406 : 399 : server_->on_bytes(nullptr);
407 : 399 : server_->on_state(nullptr);
408 : 399 : server_->on_backpressure(nullptr);
409 : : }
410 [ + + ]: 550 : if (!injected_channel_) server_.reset();
411 : 550 : framers_.clear();
412 : 550 : }
413 [ + + + + : 550 : if (use_external_context_.load() && manage_external_context_.load()) {
+ + ]
414 [ + + ]: 16 : if (work_guard_) work_guard_.reset();
415 [ + - + - ]: 16 : if (external_ioc_) external_ioc_->stop();
416 [ + + + - ]: 16 : if (external_thread_.joinable()) external_thread_.join();
417 : : }
418 : 616 : }
419 : :
420 : 270 : void setup_internal_handlers() {
421 [ - + ]: 270 : if (!server_) return;
422 : :
423 : 270 : std::weak_ptr<bool> weak_alive = alive_marker_;
424 : 270 : std::weak_ptr<Impl> weak_impl = weak_from_this();
425 : 270 : receive_budget_->reset_stats();
426 : 270 : terminal_error_reported_ = false;
427 : 270 : const auto generation = callback_gate_.open_new_generation();
428 : 270 : callback_generation_.store(generation);
429 : :
430 : 270 : auto transport_server = std::dynamic_pointer_cast<transport::UdsServer>(server_);
431 [ + + ]: 270 : if (transport_server) {
432 : 526 : transport_server->on_multi_connect(
433 : 526 : [this, generation, weak_impl, weak_alive,
434 : : weak_native = std::weak_ptr<transport::UdsServer>(transport_server)](ClientId id, const std::string& info) {
435 : 204 : auto impl_keepalive = weak_impl.lock();
436 [ - + ]: 204 : if (!impl_keepalive) return;
437 : 204 : auto alive = weak_alive.lock();
438 [ - + ]: 204 : if (!alive) return;
439 : 204 : auto lease = callback_gate_.enter(generation);
440 [ - + ]: 204 : if (!lease.admitted()) return;
441 : :
442 : 204 : auto native = weak_native.lock();
443 [ + - + - ]: 204 : auto executor = native ? native->client_executor(id) : std::nullopt;
444 [ - + ]: 204 : if (!executor) return;
445 : 204 : auto receive_scope = receive_budget_->open_scope();
446 [ + + ]: 204 : if (!receive_scope) {
447 : 1 : native->fail_receive(id);
448 : 1 : return;
449 : : }
450 : :
451 : 203 : ConnectionHandler handler;
452 : : {
453 : 203 : std::unique_lock<std::shared_mutex> lock(mutex_);
454 : 203 : batches_[id] = std::make_shared<detail::SessionBatch>(*executor, std::move(receive_scope));
455 [ + + ]: 203 : if (framer_factory_) {
456 : 40 : framers_[id] = framer_factory_();
457 : 40 : attach_framer_callback(id);
458 : : }
459 : 203 : handler = client_connect_handler_;
460 : 203 : }
461 : 203 : detail::invoke_user_callback("uds_server", "on_connect", handler, ConnectionContext(id, info));
462 : 209 : });
463 : 526 : transport_server->on_multi_data(
464 : 526 : [this, generation, weak_impl, weak_alive](ClientId id, memory::ConstByteSpan data_span) {
465 : 69 : auto impl_keepalive = weak_impl.lock();
466 [ - + ]: 69 : if (!impl_keepalive) return;
467 : 69 : auto alive = weak_alive.lock();
468 [ - + ]: 69 : if (!alive) return;
469 : 69 : auto lease = callback_gate_.enter(generation);
470 [ - + ]: 69 : if (!lease.admitted()) return;
471 : :
472 : : // #449: everything below runs synchronously on this io thread -
473 : : // mark it so a blocking send_to() called from within one of these
474 : : // callbacks fails fast instead of deadlocking.
475 : 69 : detail::CallbackGuard callback_guard;
476 : :
477 : : // #441: snapshot the handler/framer pointers under a shared_lock
478 : : // (not unique_lock) - this is a pure read, matching try_send's
479 : : // locking level so it no longer blocks concurrent sends even
480 : : // briefly.
481 : : bool batch_mode;
482 : 69 : std::shared_ptr<detail::SessionBatch> receive;
483 : 69 : interface::SharedCallback<MessageHandler> handler;
484 : 69 : std::shared_ptr<framer::IFramer> framer_to_push;
485 : : {
486 : 69 : std::shared_lock<std::shared_mutex> lock(mutex_);
487 : 69 : auto state = batches_.find(id);
488 [ - + ]: 69 : if (state == batches_.end()) return;
489 : 69 : receive = state->second;
490 : 69 : batch_mode = static_cast<bool>(data_batch_handler_);
491 : 69 : handler = data_handler_;
492 : 69 : auto it = framers_.find(id);
493 [ + + ]: 69 : if (it != framers_.end()) {
494 : 29 : framer_to_push = it->second;
495 : : }
496 : 69 : }
497 : :
498 : : try {
499 : 69 : auto prepared = detail::prepare_receive(receive->receive, framer_to_push, id, data_span, batch_mode);
500 [ + + ]: 67 : if (batch_mode) {
501 : : // #441: build the copy before taking the exclusive lock, so the
502 : : // lock is only held for the queue mutation itself, not the
503 : : // allocation.
504 : 11 : auto ctx = std::move(*prepared.raw);
505 : 11 : interface::SharedCallback<BatchMessageHandler> flush_handler;
506 : 11 : detail::ReceiveBatch batch;
507 : : {
508 : 11 : std::unique_lock<std::shared_mutex> lock(mutex_);
509 : 11 : auto state = batches_.find(id);
510 [ + - - + : 11 : if (state == batches_.end() || callback_generation_ != generation) return;
- + ]
511 : 11 : state->second->data.emplace_back(std::move(ctx));
512 [ + - + + ]: 11 : if (batches_.at(id)->data.size() >= max_batch_size_) {
513 : 4 : flush_handler = data_batch_handler_;
514 : 4 : batch = std::move(batches_.at(id)->data);
515 : 4 : batches_.at(id)->data.clear();
516 [ + - + - ]: 7 : } else if (batches_.at(id)->data.size() == 1) {
517 : 7 : schedule_batch_timer(generation, id);
518 : : }
519 : 11 : }
520 : 11 : detail::invoke_user_callback("uds_server", "on_data_batch", flush_handler, batch);
521 : 11 : } else {
522 : 112 : detail::invoke_user_callback("uds_server", "on_data", handler, MessageContext(id, data_span));
523 : : }
524 : :
525 : 67 : prepared.deliver();
526 : 69 : } catch (const detail::ReceiveOverflow& overflow) {
527 : 2 : receive->receive.scope->overflow(data_span.size(), overflow.reason);
528 : :
529 : : {
530 : 2 : std::unique_lock<std::shared_mutex> lock(mutex_);
531 : 2 : receive->data.clear();
532 : 2 : receive->messages.clear();
533 : 2 : receive->receive.reset(framer_to_push.get());
534 : 2 : }
535 [ + - + - ]: 2 : if (auto native = std::dynamic_pointer_cast<transport::UdsServer>(server_)) native->fail_receive(id);
536 : :
537 : 2 : } catch (const std::bad_alloc&) {
538 : 0 : receive->receive.scope->overflow(data_span.size(), ReceiveOverflowReason::AllocationFailure);
539 : :
540 : : {
541 : 0 : std::unique_lock<std::shared_mutex> lock(mutex_);
542 : 0 : receive->data.clear();
543 : 0 : receive->messages.clear();
544 : 0 : receive->receive.reset(framer_to_push.get());
545 : 0 : }
546 [ - - - - ]: 0 : if (auto native = std::dynamic_pointer_cast<transport::UdsServer>(server_)) native->fail_receive(id);
547 : 0 : }
548 : 69 : });
549 : 263 : transport_server->on_multi_disconnect([this, generation, weak_impl, weak_alive](ClientId id) {
550 : 102 : auto impl_keepalive = weak_impl.lock();
551 [ - + ]: 102 : if (!impl_keepalive) return;
552 : 102 : auto alive = weak_alive.lock();
553 [ - + ]: 102 : if (!alive) return;
554 : 102 : auto lease = callback_gate_.enter(generation);
555 [ - + ]: 102 : if (!lease.admitted()) return;
556 : :
557 : 102 : std::shared_ptr<detail::SessionBatch> pending;
558 : : {
559 : 102 : std::shared_lock<std::shared_mutex> lock(mutex_);
560 : 102 : auto it = batches_.find(id);
561 [ + + ]: 102 : if (it != batches_.end()) pending = it->second;
562 : 102 : }
563 [ + + ]: 102 : if (!pending) return;
564 : 101 : flush_batches(generation, id, pending);
565 : 101 : auto after_flush = callback_gate_.enter(generation);
566 [ - + ]: 101 : if (!after_flush.admitted()) return;
567 : 101 : ConnectionHandler handler;
568 : : {
569 : 101 : std::unique_lock<std::shared_mutex> lock(mutex_);
570 : 101 : batches_.erase(id);
571 : 101 : framers_.erase(id);
572 : 101 : handler = client_disconnect_handler_;
573 : 101 : }
574 : 101 : detail::invoke_user_callback("uds_server", "on_disconnect", handler, ConnectionContext(id));
575 : 105 : });
576 : :
577 : 263 : transport_server->on_backpressure([this, generation, weak_impl, weak_alive](size_t queued) {
578 : 164 : auto impl_keepalive = weak_impl.lock();
579 [ - + ]: 164 : if (!impl_keepalive) return;
580 : 164 : bp_cv_.notify_all();
581 : 164 : auto alive = weak_alive.lock();
582 [ - + ]: 164 : if (!alive) return;
583 : 164 : auto lease = callback_gate_.enter(generation);
584 [ + + ]: 164 : if (!lease.admitted()) return;
585 : 163 : std::function<void(size_t)> handler;
586 : : {
587 : 163 : std::shared_lock<std::shared_mutex> lock(mutex_);
588 : 163 : handler = on_backpressure_;
589 : 163 : }
590 : 163 : detail::invoke_user_callback("uds_server", "on_backpressure", handler, queued);
591 : 166 : });
592 : : }
593 : :
594 : 270 : server_->on_state([this, generation, weak_impl, weak_alive](base::LinkState state) {
595 : 176 : auto impl_keepalive = weak_impl.lock();
596 [ - + ]: 176 : if (!impl_keepalive) return;
597 : 176 : auto alive = weak_alive.lock();
598 [ + + ]: 176 : if (!alive) return;
599 : 175 : auto lease = callback_gate_.enter(generation);
600 [ + + ]: 175 : if (!lease.admitted()) return;
601 : :
602 [ + + ]: 174 : if (state == base::LinkState::Listening) {
603 : 170 : is_listening_.store(true);
604 : 170 : std::unique_lock<std::shared_mutex> lock(mutex_);
605 : 170 : terminal_error_reported_ = false;
606 : 170 : fulfill_all_locked(true);
607 [ - + - - : 174 : } else if (state == base::LinkState::Error || state == base::LinkState::Closed ||
- - ]
608 : : state == base::LinkState::Idle) {
609 : 4 : ErrorHandler handler;
610 : 4 : is_listening_.store(false);
611 : : {
612 : 4 : std::unique_lock<std::shared_mutex> lock(mutex_);
613 : 4 : fulfill_all_locked(false);
614 [ + - + - ]: 4 : if (state == base::LinkState::Error && !terminal_error_reported_) {
615 : 4 : terminal_error_reported_ = true;
616 : 4 : handler = error_handler_;
617 : : }
618 : 4 : }
619 [ + - ]: 4 : detail::invoke_user_callback("uds_server", "on_error", handler,
620 [ + - + - : 8 : server_ ? detail::build_error_context(*server_, "Server error")
- - ]
621 : : : ErrorContext(ErrorCode::IoError, "Server error"));
622 : 4 : }
623 : 179 : });
624 : 270 : }
625 : :
626 : 1974 : RuntimeStats stats() const {
627 : 1974 : std::shared_lock<std::shared_mutex> lock(mutex_);
628 [ + - + - ]: 3948 : return server_ ? server_->stats() : RuntimeStats{};
629 : 1974 : }
630 : :
631 : 3 : void reset_stats() {
632 : 3 : std::shared_lock<std::shared_mutex> lock(mutex_);
633 : 3 : receive_budget_->reset_stats();
634 [ + - + - ]: 3 : if (server_) server_->reset_stats();
635 : 3 : }
636 : :
637 : 40 : void attach_framer_callback(ClientId id) {
638 : 40 : auto it = framers_.find(id);
639 [ - + ]: 40 : if (it == framers_.end()) return;
640 : :
641 : 40 : const auto generation = callback_generation_.load();
642 : 40 : it->second->on_message([this, id, generation](memory::ConstByteSpan msg) {
643 : 23 : auto message_lease = callback_gate_.enter(generation);
644 [ + + ]: 23 : if (!message_lease.admitted()) return;
645 : 22 : auto prepared = detail::take_prepared_message(id, msg);
646 : : // #441: snapshot under a shared_lock (pure read), build the copy
647 : : // before taking the exclusive lock for queue mutation.
648 : : bool batch_mode;
649 : 22 : interface::SharedCallback<MessageHandler> handler;
650 : : {
651 : 22 : std::shared_lock<std::shared_mutex> lock(mutex_);
652 : 22 : batch_mode = static_cast<bool>(on_message_batch_);
653 : 22 : handler = on_message_;
654 : 22 : }
655 : :
656 [ + + ]: 22 : if (batch_mode) {
657 : 12 : std::shared_ptr<detail::SessionBatch> receive;
658 : : {
659 : 12 : std::shared_lock<std::shared_mutex> lock(mutex_);
660 : 12 : auto state = batches_.find(id);
661 [ - + ]: 12 : if (state == batches_.end()) return;
662 : 12 : receive = state->second;
663 : 12 : }
664 [ + - - - ]: 12 : auto ctx = prepared ? std::move(*prepared) : detail::retain_received(receive->receive.scope, id, msg);
665 : 12 : interface::SharedCallback<BatchMessageHandler> flush_handler;
666 : 12 : detail::ReceiveBatch batch;
667 : : {
668 : 12 : std::unique_lock<std::shared_mutex> lock(mutex_);
669 : 12 : auto state = batches_.find(id);
670 [ + - - + : 12 : if (state == batches_.end() || callback_generation_ != generation) return;
- + ]
671 : 12 : state->second->messages.emplace_back(std::move(ctx));
672 [ + - + + ]: 12 : if (batches_.at(id)->messages.size() >= max_batch_size_) {
673 : 4 : flush_handler = on_message_batch_;
674 : 4 : batch = std::move(batches_.at(id)->messages);
675 : 4 : batches_.at(id)->messages.clear();
676 [ + - + - ]: 8 : } else if (batches_.at(id)->messages.size() == 1) {
677 : 8 : schedule_batch_timer(generation, id);
678 : : }
679 : 12 : }
680 : 12 : detail::invoke_user_callback("uds_server", "on_message_batch", flush_handler, batch);
681 : 12 : return;
682 : 12 : }
683 : :
684 : 10 : detail::invoke_user_callback("uds_server", "on_message", handler, MessageContext(id, msg));
685 : 47 : });
686 : : }
687 : : };
688 : :
689 : 63 : UdsServer::UdsServer(const std::string& socket_path) : impl_(std::make_shared<Impl>(socket_path)) {
690 : 63 : config::detail::require(util::InputValidator::is_valid_uds_path(socket_path), "invalid UDS path");
691 : 64 : }
692 : :
693 : 8 : UdsServer::UdsServer(const std::string& socket_path, std::shared_ptr<boost::asio::io_context> external_ioc)
694 : 8 : : impl_(std::make_shared<Impl>(socket_path, std::move(external_ioc))) {
695 : 8 : config::detail::require(util::InputValidator::is_valid_uds_path(socket_path), "invalid UDS path");
696 : 8 : }
697 : :
698 : 99 : UdsServer::UdsServer(std::shared_ptr<interface::Channel> channel) : impl_(std::make_shared<Impl>(std::move(channel))) {
699 : 99 : impl_->setup_internal_handlers();
700 : 99 : }
701 : :
702 : : // Stop on the destroying thread first. A callback in flight holds the Impl
703 : : // alive, and if its reference were the last one, ~Impl would run on the io
704 : : // thread, where the shutdown cannot join that thread (#613).
705 : 210 : UdsServer::~UdsServer() {
706 [ - + ]: 169 : if (!impl_) return;
707 : : try {
708 : 169 : impl_->stop();
709 : 0 : } catch (...) {
710 : 0 : }
711 : 210 : }
712 : :
713 : 0 : UdsServer::UdsServer(UdsServer&&) noexcept = default;
714 : : // The previous Impl is handed to a temporary, whose destructor stops it here.
715 : 0 : UdsServer& UdsServer::operator=(UdsServer&& other) noexcept {
716 [ # # ]: 0 : if (this != &other) {
717 : 0 : UdsServer previous(std::move(*this));
718 : 0 : impl_ = std::move(other.impl_);
719 : 0 : }
720 : 0 : return *this;
721 : : }
722 : :
723 : 9 : UdsServer& UdsServer::receive_limits(ReceiveLimits limits) {
724 : 9 : limits.validate();
725 : 8 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
726 [ + + + - : 22 : if (impl_->started_ || impl_->stop_callers_.load() || (impl_->stop_requested_ && impl_->alive_marker_) ||
+ + + - -
+ + + ]
727 : 7 : detail::in_data_callback())
728 : 1 : throw std::logic_error("receive limits require completed stop");
729 : 7 : auto budget = std::make_shared<detail::ReceiveBudget>(limits);
730 : 7 : impl_->receive_limits_ = limits;
731 : 7 : impl_->receive_budget_ = std::move(budget);
732 : 7 : return *this;
733 : 8 : }
734 : 32 : ReceiveMemoryStats UdsServer::receive_stats() const {
735 : 32 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
736 : 64 : return impl_->receive_budget_->stats();
737 : 32 : }
738 : :
739 : 173 : std::future<bool> UdsServer::start() { return impl_->start(); }
740 : :
741 : 244 : void UdsServer::stop() { impl_->stop(); }
742 : :
743 : 55 : bool UdsServer::listening() const { return impl_->is_listening_.load(); }
744 : 1974 : RuntimeStats UdsServer::stats() const { return impl_->stats(); }
745 : 3 : void UdsServer::reset_stats() { impl_->reset_stats(); }
746 : :
747 : 78 : FanoutResult UdsServer::broadcast(std::string_view data) { return impl_->broadcast(data); }
748 : 30 : FanoutResult UdsServer::try_broadcast(std::string_view data) { return impl_->try_broadcast(data); }
749 : 166 : SendResult UdsServer::send_to(ClientId client_id, std::string_view data) { return impl_->send_to(client_id, data); }
750 : 76 : SendResult UdsServer::try_send_to(ClientId client_id, std::string_view data) {
751 : 76 : return impl_->try_send_to(client_id, data);
752 : : }
753 : :
754 : 101 : SendResult UdsServer::send_to_blocking(ClientId client_id, std::string_view data) {
755 : 101 : return impl_->send_to_blocking(client_id, data);
756 : : }
757 : :
758 : 6 : FanoutResult UdsServer::broadcast_line(std::string_view line) { return impl_->try_broadcast(line, true); }
759 : 82 : SendResult UdsServer::send_to_line(ClientId client_id, std::string_view line) {
760 : 246 : return send_to(client_id, std::string(line) + "\n");
761 : : }
762 : 14 : FanoutResult UdsServer::try_broadcast_line(std::string_view line) { return impl_->try_broadcast(line, true); }
763 : 19 : SendResult UdsServer::try_send_to_line(ClientId client_id, std::string_view line) {
764 : 57 : return try_send_to(client_id, std::string(line) + "\n");
765 : : }
766 : :
767 : 136 : UdsServer& UdsServer::on_connect(ConnectionHandler handler) {
768 : 136 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
769 : 136 : impl_->client_connect_handler_ = std::move(handler);
770 : 136 : return *this;
771 : 136 : }
772 : :
773 : 13 : UdsServer& UdsServer::on_disconnect(ConnectionHandler handler) {
774 : 13 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
775 : 13 : impl_->client_disconnect_handler_ = std::move(handler);
776 : 13 : return *this;
777 : 13 : }
778 : :
779 : 20 : UdsServer& UdsServer::on_data(MessageHandler handler) {
780 : 20 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
781 : 20 : impl_->data_handler_ = interface::share_callback(std::move(handler));
782 : 20 : return *this;
783 : 20 : }
784 : :
785 : 10 : UdsServer& UdsServer::on_data_batch(BatchMessageHandler handler) {
786 : 10 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
787 : 10 : impl_->data_batch_handler_ = interface::share_callback(std::move(handler));
788 : 10 : return *this;
789 : 10 : }
790 : :
791 : 21 : UdsServer& UdsServer::on_error(ErrorHandler handler) {
792 : 21 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
793 : 21 : impl_->error_handler_ = std::move(handler);
794 : 21 : return *this;
795 : 21 : }
796 : :
797 : 32 : UdsServer& UdsServer::on_backpressure(std::function<void(size_t)> handler) {
798 : 32 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
799 : 32 : impl_->on_backpressure_ = std::move(handler);
800 : 32 : return *this;
801 : 32 : }
802 : :
803 : 29 : UdsServer& UdsServer::framer(FramerFactory factory) {
804 : 29 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
805 : 29 : impl_->framer_factory_ = std::move(factory);
806 : 29 : return *this;
807 : 29 : }
808 : :
809 : 12 : UdsServer& UdsServer::on_message(MessageHandler handler) {
810 : 12 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
811 : 12 : impl_->on_message_ = interface::share_callback(std::move(handler));
812 : 12 : return *this;
813 : 12 : }
814 : :
815 : 8 : UdsServer& UdsServer::on_message_batch(BatchMessageHandler handler) {
816 : 8 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
817 : 8 : impl_->on_message_batch_ = interface::share_callback(std::move(handler));
818 : 8 : return *this;
819 : 8 : }
820 : :
821 : 100 : size_t UdsServer::client_count() const {
822 : 100 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
823 : 100 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(impl_->server_);
824 [ + + + - ]: 200 : return ts ? ts->client_count() : 0;
825 : 100 : }
826 : :
827 : 6 : std::vector<ClientId> UdsServer::connected_clients() const {
828 : 6 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
829 : 6 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(impl_->server_);
830 [ + + + - ]: 12 : return ts ? ts->connected_clients() : std::vector<ClientId>{};
831 : 6 : }
832 : :
833 : 28 : std::optional<RuntimeStats> UdsServer::client_stats(ClientId client_id) const {
834 : 28 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
835 : 28 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(impl_->server_);
836 [ + - + - ]: 56 : return ts ? ts->client_stats(client_id) : std::nullopt;
837 : 28 : }
838 : :
839 : 1 : UdsServer& UdsServer::auto_start(bool manage) {
840 : 1 : impl_->auto_start_.store(manage);
841 [ + - + - : 1 : if (impl_->auto_start_.load() && !impl_->started_.load()) {
+ - ]
842 : 1 : start();
843 : : }
844 : 1 : return *this;
845 : : }
846 : :
847 : 5 : UdsServer& UdsServer::idle_timeout(std::chrono::milliseconds timeout) {
848 : 5 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
849 : 5 : impl_->require_stopped_config();
850 : 4 : config::detail::duration(timeout, base::constants::MIN_IDLE_TIMEOUT_MS, base::constants::MAX_IDLE_TIMEOUT_MS, true,
851 : : "invalid idle_timeout");
852 : 4 : impl_->idle_timeout_ms_.store(static_cast<int>(timeout.count()));
853 : 4 : return *this;
854 : 5 : }
855 : :
856 : 7 : UdsServer& UdsServer::max_clients(size_t max) {
857 : 7 : config::detail::range(max, 0, base::constants::MAX_MAX_CONNECTIONS, "invalid client limit");
858 : 6 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
859 : 6 : impl_->max_clients_.store(max);
860 [ - + ]: 6 : if (max == 0) {
861 : 0 : impl_->client_limit_enabled_.store(false);
862 : : } else {
863 : 6 : impl_->client_limit_enabled_.store(true);
864 : : }
865 : 6 : auto ts = std::dynamic_pointer_cast<transport::UdsServer>(impl_->server_);
866 [ + + + - ]: 6 : if (ts) ts->set_client_limit(max);
867 : 6 : return *this;
868 : 6 : }
869 : :
870 : 0 : UdsServer& UdsServer::socket_permissions(int mode) {
871 : 0 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
872 : 0 : impl_->require_stopped_config();
873 [ # # # # : 0 : config::detail::require(mode == -1 || (mode >= 0 && mode <= 0777), "invalid socket permissions");
# # # # ]
874 : 0 : impl_->socket_permissions_.store(mode);
875 : 0 : return *this;
876 : 0 : }
877 : :
878 : 67 : UdsServer& UdsServer::backpressure_threshold(size_t threshold) {
879 : 67 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
880 : 67 : impl_->require_stopped_config();
881 : 66 : config::detail::range(threshold, base::constants::MIN_BACKPRESSURE_THRESHOLD,
882 : : base::constants::MAX_BACKPRESSURE_THRESHOLD, "invalid backpressure_threshold");
883 : 66 : impl_->backpressure_threshold_.store(threshold);
884 : 66 : return *this;
885 : 67 : }
886 : :
887 : 0 : UdsServer& UdsServer::read_buffer_size(size_t bytes) {
888 : 0 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
889 : 0 : impl_->require_stopped_config();
890 : 0 : config::detail::range(bytes, base::constants::MIN_READ_BUFFER_SIZE, base::constants::MAX_READ_BUFFER_SIZE,
891 : : "invalid read_buffer_size");
892 : 0 : impl_->read_buffer_size_.store(bytes);
893 : 0 : return *this;
894 : 0 : }
895 : :
896 : 44 : UdsServer& UdsServer::backpressure_strategy(base::constants::BackpressureStrategy strategy) {
897 : 44 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
898 : 44 : impl_->require_stopped_config();
899 : 43 : config::detail::strategy(strategy);
900 : 43 : impl_->backpressure_strategy_.store(strategy);
901 : 43 : return *this;
902 : 44 : }
903 : :
904 : 10 : size_t UdsServer::backpressure_threshold() const { return impl_->backpressure_threshold_.load(); }
905 : :
906 : 3 : base::constants::BackpressureStrategy UdsServer::backpressure_strategy() const {
907 : 3 : return impl_->backpressure_strategy_.load();
908 : : }
909 : :
910 : 7 : UdsServer& UdsServer::manage_external_context(bool manage) {
911 : 7 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
912 : 7 : impl_->require_stopped_config();
913 : 6 : impl_->manage_external_context_.store(manage);
914 : 6 : return *this;
915 : 7 : }
916 : :
917 : 23 : UdsServer& UdsServer::batch_size(size_t size) {
918 [ + - ]: 23 : config::detail::require(size > 0, "batch size must be positive");
919 : 23 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
920 : 23 : impl_->max_batch_size_ = size;
921 : 23 : return *this;
922 : 23 : }
923 : :
924 : 23 : UdsServer& UdsServer::batch_latency(std::chrono::milliseconds latency) {
925 : 23 : config::detail::duration(latency, 0, std::numeric_limits<int>::max(), true, "invalid batch latency");
926 : 23 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
927 : 23 : impl_->max_batch_latency_ = latency;
928 : 23 : return *this;
929 : 23 : }
930 : :
931 : : } // namespace wrapper
932 : : } // namespace wirestead
|
