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/wrapper/serial/serial.hpp"
18 : :
19 : : #include <algorithm>
20 : : #include <atomic>
21 : : #include <boost/asio/executor_work_guard.hpp>
22 : : #include <boost/asio/io_context.hpp>
23 : : #include <boost/asio/steady_timer.hpp>
24 : : #include <boost/asio/strand.hpp>
25 : : #include <cctype>
26 : : #include <iostream>
27 : : #include <mutex>
28 : : #include <optional>
29 : : #include <shared_mutex>
30 : : #include <stdexcept>
31 : : #include <thread>
32 : : #include <vector>
33 : :
34 : : #include "wirestead/base/common.hpp"
35 : : #include "wirestead/base/constants.hpp"
36 : : #include "wirestead/concurrency/io_thread_hook.hpp"
37 : : #include "wirestead/config/validation.hpp"
38 : : #include "wirestead/factory/channel_factory.hpp"
39 : : #include "wirestead/interface/connection_channel.hpp"
40 : : #include "wirestead/transport/serial/detail/write_wait.hpp"
41 : : #include "wirestead/transport/serial/serial.hpp"
42 : : #include "wirestead/util/input_validator.hpp"
43 : : #include "wirestead/wrapper/bounded_receive.hpp"
44 : : #include "wirestead/wrapper/callback_guard.hpp"
45 : : #include "wirestead/wrapper/error_context_builder.hpp"
46 : : #include "wirestead/wrapper/lifecycle_events.hpp"
47 : : #include "wirestead/wrapper/send_retry.hpp"
48 : : #include "wirestead/wrapper/send_validation.hpp"
49 : :
50 : : namespace wirestead {
51 : : namespace wrapper {
52 : :
53 : : namespace {
54 : 60 : std::string to_lower(std::string s) {
55 : 312 : std::transform(s.begin(), s.end(), s.begin(), [](unsigned char c) { return std::tolower(c); });
56 : 60 : return s;
57 : : }
58 : : } // namespace
59 : :
60 : : struct Serial::Impl : public std::enable_shared_from_this<Impl> {
61 : : mutable std::shared_mutex mutex_;
62 : : std::mutex bp_mutex_;
63 : : std::condition_variable bp_cv_;
64 : : // D-1: admission gate for this object's user callbacks. Admission, the
65 : : // running count and the closed flag are one decision, so a stop() cannot
66 : : // observe an empty gate while a callback is about to start, and a callback
67 : : // left over from a previous run is refused after a restart.
68 : : detail::CallbackGate callback_gate_;
69 : : std::mutex stop_finalize_mutex_;
70 : : bool stop_requested_ = false;
71 : : std::atomic<unsigned> stop_callers_{0};
72 : : std::atomic<uint64_t> callback_generation_{0};
73 : :
74 : : // True when this thread is one the target's shutdown needs: a callback of
75 : : // this object, or any thread currently running the external io_context this
76 : : // channel was built on (which a callback of another channel sharing it is).
77 : : // Such a caller requests the shutdown and returns; it cannot wait for work
78 : : // its own thread has to perform.
79 : 865 : bool shutdown_needs_this_thread() const {
80 [ + - + + ]: 865 : if (callback_gate_.active_on_this_thread()) return true;
81 [ + + - + : 851 : if (external_ioc && external_ioc->get_executor().running_in_this_thread()) return true;
+ + - + ]
82 : 851 : std::shared_lock<std::shared_mutex> lock(mutex_);
83 [ + + ]: 851 : if (!channel) return false;
84 : 761 : const auto executor = channel->get_executor();
85 : : using IoExecutor = boost::asio::io_context::executor_type;
86 [ + - + + ]: 761 : if (auto* io = executor.target<IoExecutor>()) return io->running_in_this_thread();
87 [ + - + + ]: 462 : if (auto* strand = executor.target<boost::asio::strand<IoExecutor>>())
88 : 459 : return strand->get_inner_executor().running_in_this_thread();
89 : 3 : return false;
90 : 851 : }
91 : :
92 : : std::string device;
93 : : uint32_t baud_rate;
94 : : std::shared_ptr<interface::Channel> channel;
95 : : std::shared_ptr<boost::asio::io_context> external_ioc;
96 : : std::atomic<bool> use_external_context{false};
97 : : std::atomic<bool> manage_external_context{false};
98 : : std::thread external_thread;
99 : : std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>> work_guard_;
100 : :
101 : : std::vector<std::promise<bool>> pending_promises_;
102 : : std::atomic<bool> started_{false};
103 : 257 : std::shared_ptr<bool> alive_marker_{std::make_shared<bool>(true)};
104 : :
105 : : // Event handlers (Context based)
106 : : // Shared snapshots: the strand copies one out per received chunk, and a
107 : : // std::function copy allocates whenever the user handler outgrows its
108 : : // small-object buffer. See interface::SharedCallback.
109 : : interface::SharedCallback<MessageHandler> data_handler;
110 : : interface::SharedCallback<BatchMessageHandler> data_batch_handler_;
111 : : ConnectionHandler connect_handler{nullptr};
112 : : ConnectionHandler disconnect_handler{nullptr};
113 : : ErrorHandler error_handler{nullptr};
114 : : std::function<void(size_t)> bp_handler{nullptr};
115 : : interface::SharedCallback<MessageHandler> message_handler;
116 : : interface::SharedCallback<BatchMessageHandler> message_batch_handler_;
117 : :
118 : : std::shared_ptr<framer::IFramer> framer{nullptr};
119 : :
120 : : detail::LifecycleEvents lifecycle_events_;
121 : 94 : void require_stopped_config() const {
122 [ + + + - : 186 : if (started_.load() || stop_callers_.load() || (stop_requested_ && alive_marker_) || detail::in_data_callback())
+ + + + -
+ + + ]
123 : 3 : throw std::logic_error("configuration requires completed stop");
124 : 91 : }
125 : : ReceiveLimits receive_limits_;
126 : : std::shared_ptr<detail::ReceiveBudget> receive_budget_{std::make_shared<detail::ReceiveBudget>(receive_limits_)};
127 : : std::shared_ptr<detail::ReceiveState> receive_state_{
128 : : std::make_shared<detail::ReceiveState>(receive_budget_->open_scope())};
129 : : // Batching logic
130 : : detail::ReceiveBatch data_batch_queue_;
131 : : detail::ReceiveBatch message_batch_queue_;
132 : : std::unique_ptr<boost::asio::steady_timer> batch_timer_;
133 : : size_t max_batch_size_ = 100;
134 : 257 : std::chrono::milliseconds max_batch_latency_{1};
135 : :
136 : : // Configuration
137 : : std::atomic<bool> auto_start_ = false;
138 : : std::atomic<bool> shared_context_{false};
139 : : int data_bits = 8;
140 : : int stop_bits = 1;
141 : : std::string parity = "none";
142 : : std::string flow_control = "none";
143 : : bool reopen_on_error = true;
144 : : size_t read_chunk = base::constants::DEFAULT_READ_BUFFER_SIZE;
145 : : bool low_latency = true;
146 : : config::SerialConfig::Rs485 rs485{};
147 : : std::optional<bool> dtr;
148 : : bool keep_dtr_on_close = false;
149 : : std::optional<bool> rts;
150 : 257 : std::chrono::milliseconds rx_idle_timeout{0};
151 : : std::chrono::milliseconds retry_interval{base::constants::DEFAULT_RETRY_INTERVAL_MS};
152 : : size_t backpressure_threshold = base::constants::DEFAULT_BACKPRESSURE_THRESHOLD;
153 : : base::constants::BackpressureStrategy backpressure_strategy = base::constants::BackpressureStrategy::Reliable;
154 : :
155 : : // False only for the dependency-injected-channel constructor below, where
156 : : // the caller owns the channel's identity and lifecycle (e.g. tests
157 : : // injecting a fake channel) - stop() must not discard and factory-rebuild
158 : : // a channel it didn't create itself.
159 : : bool factory_managed_channel_ = true;
160 : :
161 : 200 : Impl(const std::string& dev, uint32_t baud) : device(dev), baud_rate(baud) {}
162 : 3 : Impl(const std::string& dev, uint32_t baud, std::shared_ptr<boost::asio::io_context> ioc)
163 [ + - + - : 15 : : device(dev), baud_rate(baud), external_ioc(std::move(ioc)), use_external_context(external_ioc != nullptr) {}
+ - + - +
- + - +
- ]
164 : 1070 : explicit Impl(std::shared_ptr<interface::Channel> ch) : channel(std::move(ch)), factory_managed_channel_(false) {
165 [ + + + + ]: 531 : if (!std::dynamic_pointer_cast<transport::Serial>(channel) &&
166 [ + + + + : 317 : !std::dynamic_pointer_cast<interface::ConnectionChannel>(channel))
+ - ]
167 : 3 : throw std::invalid_argument("Serial requires its native transport or a ConnectionChannel");
168 : :
169 : : // #450: setup_internal_handlers() captures weak_from_this() - calling it
170 : : // from inside this constructor would capture an empty weak_ptr, since
171 : : // enable_shared_from_this isn't wired up until make_shared() finishes
172 : : // constructing the object. Deferred to Serial's own constructor, which
173 : : // runs after impl_ is a fully-formed shared_ptr<Impl> (already called
174 : : // there for this constructor, so nothing to add there).
175 : 289 : }
176 : :
177 : 254 : ~Impl() {
178 : : try {
179 : 254 : stop();
180 : 0 : } catch (...) {
181 : 0 : }
182 : 254 : }
183 : :
184 : 1096 : void fulfill_all_locked(bool value) {
185 [ + + ]: 1293 : for (auto& promise : pending_promises_) {
186 : : try {
187 : 197 : promise.set_value(value);
188 : 0 : } catch (...) {
189 : 0 : }
190 : : }
191 : 1096 : pending_promises_.clear();
192 : 1096 : }
193 : :
194 : 45 : void flush_batches(uint64_t generation) {
195 : 45 : auto lease = callback_gate_.enter(generation);
196 [ - + ]: 45 : if (!lease.admitted()) return;
197 : : {
198 : 45 : std::shared_lock<std::shared_mutex> lock(mutex_);
199 [ + + + + : 45 : if (data_batch_queue_.empty() && message_batch_queue_.empty()) return;
+ + ]
200 : 45 : }
201 : 7 : std::unique_lock<std::shared_mutex> lock(mutex_);
202 [ + + ]: 7 : if (!data_batch_queue_.empty()) {
203 : 6 : auto handler = data_batch_handler_;
204 : 6 : auto batch = std::move(data_batch_queue_);
205 : 6 : data_batch_queue_.clear();
206 [ + - ]: 6 : if (handler) {
207 : 6 : lock.unlock();
208 : 6 : detail::invoke_user_callback("serial", "on_data_batch", handler, batch);
209 : 6 : lock.lock();
210 : : }
211 : 6 : }
212 [ + - + + ]: 7 : if (!callback_gate_.enter(generation).admitted()) return;
213 [ + + ]: 4 : if (!message_batch_queue_.empty()) {
214 : 2 : auto handler = message_batch_handler_;
215 : 2 : auto batch = std::move(message_batch_queue_);
216 : 2 : message_batch_queue_.clear();
217 [ + - ]: 2 : if (handler) {
218 : 2 : lock.unlock();
219 : 2 : detail::invoke_user_callback("serial", "on_message_batch", handler, batch);
220 : 2 : lock.lock();
221 : : }
222 : 2 : }
223 [ + - ]: 4 : if (batch_timer_) {
224 : 4 : batch_timer_->cancel();
225 : : }
226 : 48 : }
227 : :
228 : 15 : void schedule_batch_timer(uint64_t generation) {
229 [ - + ]: 15 : if (!batch_timer_) return;
230 : 15 : batch_timer_->expires_after(max_batch_latency_);
231 : 30 : batch_timer_->async_wait([this, generation, weak_impl = weak_from_this(),
232 : 15 : weak_alive = std::weak_ptr<bool>(alive_marker_)](const boost::system::error_code& ec) {
233 [ + + ]: 7 : if (ec) return;
234 : 5 : auto impl_keepalive = weak_impl.lock();
235 [ - + ]: 5 : if (!impl_keepalive) return;
236 : 5 : auto alive = weak_alive.lock();
237 [ - + ]: 5 : if (!alive) return;
238 : 5 : flush_batches(generation);
239 : 5 : });
240 : : }
241 : :
242 : 255 : std::future<bool> start() {
243 : 255 : std::unique_lock<std::shared_mutex> lock(mutex_);
244 : 255 : stop_requested_ = false;
245 [ + + + - : 255 : if (channel && channel->is_connected()) {
+ + + + ]
246 : 58 : started_.store(true);
247 : 58 : std::promise<bool> p;
248 : 58 : p.set_value(true);
249 : 58 : return p.get_future();
250 : 58 : }
251 : :
252 : 197 : std::promise<bool> p;
253 : 197 : auto future = p.get_future();
254 : 197 : pending_promises_.emplace_back(std::move(p));
255 : :
256 [ + + ]: 197 : if (started_.load()) {
257 : 1 : return future;
258 : : }
259 : :
260 [ + + + - ]: 196 : if (!alive_marker_) alive_marker_ = std::make_shared<bool>(true);
261 [ + + ]: 196 : if (!channel) {
262 : 20 : channel = factory::ChannelFactory::create(build_config_locked(), external_ioc);
263 : : }
264 : 196 : setup_internal_handlers();
265 : 196 : started_.store(true);
266 : :
267 : 196 : auto channel_copy = channel;
268 : 196 : lock.unlock();
269 : 196 : channel_copy->start();
270 [ + + + - : 196 : if (use_external_context && manage_external_context && !external_thread.joinable()) {
+ - + + ]
271 [ + - + - : 3 : if (external_ioc && external_ioc->stopped()) {
+ + + + ]
272 : 1 : external_ioc->restart();
273 : : }
274 : 6 : work_guard_ = std::make_unique<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>(
275 : 9 : boost::asio::make_work_guard(*external_ioc));
276 : 6 : external_thread = std::thread([ioc = external_ioc]() {
277 : 3 : wirestead::concurrency::run_io_thread_init();
278 : : try {
279 : 3 : ioc->run();
280 : 0 : } catch (...) {
281 : 0 : }
282 : 6 : });
283 : : }
284 : :
285 : 196 : return future;
286 : 255 : }
287 : :
288 : 865 : void stop() {
289 : 865 : stop_callers_.fetch_add(1);
290 : : struct StopCall {
291 : : std::atomic<unsigned>& callers;
292 : 865 : ~StopCall() { callers.fetch_sub(1); }
293 : 865 : } stop_call{stop_callers_};
294 : 865 : const bool request_only = shutdown_needs_this_thread();
295 : 865 : callback_gate_.close();
296 : 865 : std::shared_ptr<interface::Channel> channel_copy;
297 : : {
298 : 865 : std::unique_lock<std::shared_mutex> lock(mutex_);
299 : 865 : stop_requested_ = true;
300 [ + + + - ]: 865 : if (auto serial = std::dynamic_pointer_cast<transport::Serial>(channel)) serial->cancel_write_waits();
301 [ + + + - ]: 865 : if (auto custom = std::dynamic_pointer_cast<interface::ConnectionChannel>(channel)) custom->cancel_write_waits();
302 : 865 : started_.store(false);
303 : :
304 : 865 : bp_cv_.notify_all();
305 : 865 : fulfill_all_locked(false);
306 : 865 : channel_copy = channel;
307 : 865 : }
308 [ + + + - ]: 865 : if (channel_copy) channel_copy->stop();
309 [ + + ]: 865 : if (request_only) return;
310 : :
311 : 837 : callback_gate_.wait_until_idle();
312 : 837 : std::lock_guard<std::mutex> finalize_lock(stop_finalize_mutex_);
313 : : {
314 : 837 : std::unique_lock<std::shared_mutex> lock(mutex_);
315 : 837 : alive_marker_.reset();
316 [ + + ]: 837 : if (batch_timer_) {
317 : 252 : batch_timer_->cancel();
318 : 252 : batch_timer_.reset();
319 : : }
320 : :
321 [ + + ]: 837 : if (channel) {
322 : 745 : channel->on_bytes(nullptr);
323 : 745 : channel->on_state(nullptr);
324 : 745 : channel->on_backpressure(nullptr);
325 : : }
326 [ + + ]: 837 : if (factory_managed_channel_) channel.reset();
327 : 837 : data_batch_queue_.clear();
328 : 837 : message_batch_queue_.clear();
329 : 837 : receive_state_->reset(framer.get());
330 : 837 : }
331 [ + + + - : 837 : if (use_external_context && manage_external_context) {
+ + ]
332 : 12 : work_guard_.reset();
333 [ + - + - ]: 12 : if (external_ioc) external_ioc->stop();
334 [ + + + - ]: 12 : if (external_thread.joinable()) external_thread.join();
335 : : }
336 : 893 : }
337 : :
338 : : // Caller holds mutex_. Native readiness remains part of transport admission.
339 : 756 : SendResult send_state(const std::shared_ptr<transport::Serial>& serial, bool custom = false) {
340 [ + + ]: 1512 : if (stop_callers_.load() != 0) return SendResult::reject(SendRejection::Stopping);
341 [ + + ]: 755 : if (!started_.load()) {
342 [ + + ]: 78 : if (stop_requested_) {
343 [ - + ]: 41 : if (!callback_gate_.idle()) return SendResult::reject(SendRejection::Stopping);
344 [ + + ]: 41 : if (serial) {
345 : 38 : const auto state = serial->write_state();
346 [ + - + + : 38 : if (!state.accepted() && state.reason() == SendRejection::Stopping) return state;
+ + ]
347 : : }
348 : : }
349 : 66 : return SendResult::reject(SendRejection::NotStarted);
350 : : }
351 [ + + - + : 677 : if (!serial && !custom) return SendResult::reject(SendRejection::NotReady);
- + ]
352 : 677 : return SendResult::accept();
353 : : }
354 : :
355 : 633 : static SendResult finish_send(SendResult result) {
356 [ + + ]: 633 : if (auto hook = detail::g_serial_send_result_hook.load()) hook(result);
357 : 633 : return result;
358 : : }
359 : :
360 : : template <typename NativeWrite, typename CustomWrite>
361 : 181 : SendResult nonblocking_send(size_t size, bool best_effort_send, NativeWrite native_write, CustomWrite custom_write) {
362 : 181 : std::shared_lock<std::shared_mutex> lock(mutex_);
363 : 181 : auto serial = std::dynamic_pointer_cast<transport::Serial>(channel);
364 [ + + ]: 181 : auto custom = serial ? nullptr : std::dynamic_pointer_cast<interface::ConnectionChannel>(channel);
365 : 181 : const auto result = [&]() -> SendResult {
366 [ + + + - : 181 : auto validation = detail::validate_payload_size(size, channel ? channel->write_queue_limit() : std::nullopt);
+ + + - +
+ + - ]
367 [ + + + + : 181 : if (!validation.accepted()) return validation;
+ + ]
368 [ + - + - : 120 : const auto state = send_state(serial, custom != nullptr);
+ - ]
369 [ + + + + : 120 : if (!state.accepted()) return state;
+ + ]
370 : : // Admission rechecks state and capacity together under the channel lock.
371 [ + + + - : 78 : auto admitted = custom ? custom_write(*custom) : native_write(*serial);
+ - + + +
- + - + +
+ - + - ]
372 [ + + + + : 78 : if (best_effort_send && !admitted.accepted() && admitted.reason() == SendRejection::WouldBlock)
+ + + + +
+ + + + +
+ + + + +
+ + + +
+ ]
373 : 9 : return SendResult::reject(SendRejection::QueueFull);
374 : 69 : return admitted;
375 : 181 : }();
376 : 181 : lock.unlock();
377 : 362 : return finish_send(result);
378 : 181 : }
379 : :
380 : 92 : SendResult try_send(std::string_view data, bool best_effort_send = false) {
381 : 92 : auto binary_view = base::safe_convert::string_to_bytes(data);
382 : 92 : memory::ConstByteSpan span(binary_view.first, binary_view.second);
383 : 92 : return nonblocking_send(
384 : 37 : data.size(), best_effort_send, [&](auto& serial) { return serial.try_write_copy(span); },
385 : 191 : [&](auto& channel) { return channel.async_try_write_copy_result(span); });
386 : : }
387 : :
388 : 40 : SendResult try_send_move(std::vector<uint8_t>&& data, bool best_effort_send = false) {
389 : 40 : return nonblocking_send(
390 : 12 : data.size(), best_effort_send, [&](auto& serial) { return serial.try_write_move(std::move(data)); },
391 : 45 : [&](auto& channel) { return channel.async_try_write_move_result(std::move(data)); });
392 : : }
393 : :
394 : 49 : SendResult try_send_shared(std::shared_ptr<const std::vector<uint8_t>> data, bool best_effort_send = false) {
395 [ + + ]: 92 : return nonblocking_send(
396 : 92 : data ? data->size() : 0, best_effort_send,
397 : 12 : [&](auto& serial) { return serial.try_write_shared(std::move(data)); },
398 : 54 : [&](auto& channel) { return channel.async_try_write_shared_result(std::move(data)); });
399 : : }
400 : :
401 : 102 : SendResult send(std::string_view data) {
402 [ + + ]: 102 : if (backpressure_strategy == base::constants::BackpressureStrategy::Reliable) return send_blocking(data);
403 : 13 : return try_send(data, true);
404 : : }
405 : :
406 : : struct ConnectionPin {
407 : : bool cannot_wait = false;
408 : : std::shared_ptr<interface::ConnectionChannel> custom;
409 : : interface::ConnectionChannel::Connection custom_wait;
410 : : std::shared_ptr<transport::Serial> serial;
411 : : std::shared_ptr<transport::detail::SerialWriteWait> wait;
412 : : };
413 : :
414 : 83 : bool connection_matches(const ConnectionPin& pin) const {
415 [ + - + - : 83 : return !pin.serial || (pin.wait && pin.serial->write_connection() == pin.wait->sequence);
+ - + - ]
416 : : }
417 : :
418 : : // channel->on_backpressure() calls bp_cv_.notify_all() from the transport's io_context
419 : : // thread without holding bp_mutex_ (backpressure_active_ is a plain atomic on the transport
420 : : // side, not guarded by bp_mutex_ at all). That makes a classic lost-wakeup race possible: a
421 : : // waiter can check the predicate, find it still blocking, and be in the process of
422 : : // registering to wait when the notify fires - in the rare case that race is lost, an
423 : : // unbounded wait() would block forever. Poll with a bounded timeout instead so a missed
424 : : // notify only costs a short delay rather than a permanent hang (see #427, #431).
425 : : //
426 : : // Callback scopes cannot wait for capacity: they may be running on the
427 : : // executor needed to drain their own or another channel's queue (D-2).
428 : 374 : SendResult wait_for_backpressure_clear(std::unique_lock<std::mutex>& bp_lock, size_t payload_size,
429 : : uint64_t generation, const ConnectionPin& connection) {
430 [ + + ]: 374 : if (connection.custom_wait) {
431 : 290 : auto outcome = connection.custom_wait->poll_capacity();
432 [ + + ]: 290 : if (outcome) return *outcome;
433 [ + + ]: 118 : if (connection.cannot_wait) return SendResult::reject(SendRejection::WouldBlock);
434 [ + + + - ]: 37 : if (auto hook = detail::g_serial_capacity_wait_hook.load()) hook();
435 [ + - + + ]: 46 : while (!bp_cv_.wait_for(bp_lock, std::chrono::milliseconds(50), [&] {
436 : 80 : outcome = connection.custom_wait->poll_capacity();
437 : 80 : return outcome.has_value();
438 : : })) {
439 : : }
440 [ + + + - ]: 37 : if (auto hook = detail::g_serial_capacity_wait_result_hook.load()) hook(*outcome);
441 : 37 : return *outcome;
442 : : }
443 [ - + ]: 84 : if (!detail::payload_needs_capacity(payload_size)) return SendResult::accept();
444 : 84 : auto immediate = [this, payload_size, generation, &connection] {
445 : 84 : std::shared_lock<std::shared_mutex> lock(mutex_);
446 [ + + + - : 168 : return callback_generation_.load() != generation || !started_.load() || !channel || !channel->is_connected() ||
+ - + - ]
447 [ + - + - ]: 83 : !connection_matches(connection) ||
448 [ + - + - : 251 : !detail::payload_needs_capacity(payload_size, channel->write_queue_limit()) ||
+ - ]
449 [ + - + + ]: 167 : !channel->is_backpressure_active();
450 : 84 : };
451 : : // A bypass is not a completed capacity wait; final admission checks still apply.
452 [ + - + + ]: 84 : if (immediate()) return SendResult::accept();
453 [ + + ]: 42 : if (connection.cannot_wait) return SendResult::reject(SendRejection::WouldBlock);
454 [ + - + - ]: 36 : if (auto hook = detail::g_serial_capacity_wait_hook.load()) hook();
455 : 36 : std::optional<SendResult> outcome;
456 : 36 : auto released = [&] {
457 [ - + ]: 36 : if (outcome) return true;
458 : 36 : std::shared_lock<std::shared_mutex> lock(mutex_);
459 : : // The connection record retains the first terminal cause under the
460 : : // same transport lock used by stop, loss and admission.
461 : 36 : outcome = connection.serial->poll_write_wait(connection.wait);
462 : 36 : return outcome.has_value();
463 : 36 : };
464 [ + - - + ]: 36 : while (!bp_cv_.wait_for(bp_lock, std::chrono::milliseconds(50), released)) {
465 : : }
466 [ + - + - ]: 36 : if (auto hook = detail::g_serial_capacity_wait_result_hook.load()) hook(*outcome);
467 : 36 : return *outcome;
468 : : }
469 : :
470 : : // #509: high-water pressure and hard-limit reservations are different
471 : : // thresholds. Capacity can be refilled between a wait and admission, so
472 : : // retry transient native WouldBlock until admission or cancellation. Validation and
473 : : // terminal state failures return immediately.
474 : :
475 : : template <typename NativeWrite, typename CustomWrite>
476 : 452 : SendResult blocking_send(size_t size, NativeWrite native_write, CustomWrite custom_write) {
477 : : uint64_t generation;
478 : 452 : ConnectionPin connection;
479 : 452 : const auto result = [&]() -> SendResult {
480 : : {
481 : 452 : std::shared_lock<std::shared_mutex> lock(mutex_);
482 : : // Select the run and native connection together at entry. Validation
483 : : // precedes state and waiting, including the line delimiter and hard cap.
484 : 452 : generation = callback_generation_.load();
485 : 452 : connection.cannot_wait =
486 [ + + + + : 452 : detail::in_data_callback() || (channel && detail::executor_running_here(channel->get_executor()));
+ - + - +
+ + + - -
+ + + + +
- + - + +
+ + - - +
+ + + + -
+ - + + +
+ - - ]
487 : 452 : connection.serial = std::dynamic_pointer_cast<transport::Serial>(channel);
488 : 452 : connection.custom =
489 [ + + + + : 904 : connection.serial ? nullptr : std::dynamic_pointer_cast<interface::ConnectionChannel>(channel);
+ + ]
490 [ + + + - : 452 : auto validation = detail::validate_payload_size(size, channel ? channel->write_queue_limit() : std::nullopt);
+ + + - +
+ + - ]
491 [ + + + + : 452 : if (!validation.accepted()) return validation;
+ + ]
492 [ + - + - : 385 : auto state = send_state(connection.serial, connection.custom != nullptr);
+ - ]
493 [ + + + + : 385 : if (!state.accepted()) return state;
+ + ]
494 [ + + + + : 355 : if (connection.custom) {
+ + ]
495 : 274 : auto captured = connection.custom->capture_write_connection();
496 [ + + + + : 274 : if (auto reason = std::get_if<SendRejection>(&captured)) return SendResult::reject(*reason);
+ + ]
497 : 253 : connection.custom_wait = std::get<interface::ConnectionChannel::Connection>(std::move(captured));
498 [ - + - - : 253 : if (!connection.custom_wait) throw std::logic_error("ConnectionChannel returned a null connection");
- + - - -
+ - - ]
499 : 274 : } else {
500 : 81 : state = connection.serial->write_state();
501 [ + + + + : 81 : if (!state.accepted()) return state;
+ + ]
502 : 72 : connection.wait = connection.serial->capture_write_wait();
503 [ - + - + : 72 : if (!connection.wait) return SendResult::reject(SendRejection::NotReady);
- + ]
504 : : }
505 : 452 : }
506 [ + + + + : 950 : for (bool retry = false;; retry = true) {
+ + + + +
+ + + ]
507 [ + + + - : 374 : if (retry) detail::pause_send_retry(bp_cv_, bp_mutex_);
+ + + - +
+ + - ]
508 : 374 : std::unique_lock<std::mutex> bp_lock(bp_mutex_);
509 : 374 : const auto released = wait_for_backpressure_clear(bp_lock, size, generation, connection);
510 [ + + + + : 374 : if (!released.accepted()) return released; // Never overwrite the cause of release.
+ + ]
511 : 251 : bp_lock.unlock();
512 : 251 : std::shared_lock<std::shared_mutex> lock(mutex_);
513 [ + - + - : 251 : const auto state = send_state(connection.serial, connection.custom != nullptr);
+ - ]
514 [ + + + + : 251 : if (!state.accepted()) return state;
+ + ]
515 [ - + - + : 488 : if (callback_generation_.load() != generation) return SendResult::reject(SendRejection::NotReady);
- + ]
516 [ + + + - : 244 : const auto admitted = connection.custom_wait ? custom_write(*connection.custom_wait)
+ + + - +
+ + - ]
517 : 47 : : native_write(*connection.serial, connection.wait->sequence);
518 [ + + + + : 244 : if (admitted.accepted() || admitted.reason() != SendRejection::WouldBlock) return admitted;
+ + + + +
+ + + + +
+ + + + ]
519 : : // Only transient capacity refusal can be retried, never a terminal
520 : : // result. Callback callers must return without entering another wait.
521 [ + + + + : 73 : if (connection.cannot_wait) return admitted;
+ + ]
522 : : }
523 : 452 : }();
524 : 904 : return finish_send(result);
525 : 452 : }
526 : :
527 : 87 : SendResult send_move(std::vector<uint8_t>&& data) {
528 [ + + ]: 87 : if (backpressure_strategy != base::constants::BackpressureStrategy::Reliable)
529 : 13 : return try_send_move(std::move(data), true);
530 : 74 : return blocking_send(
531 : 16 : data.size(), [&](auto& serial, uint64_t sequence) { return serial.write_move(std::move(data), sequence); },
532 : 115 : [&](auto& connection) { return connection.write_move(std::move(data)); });
533 : : }
534 : :
535 : 99 : SendResult send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
536 [ + + ]: 99 : if (backpressure_strategy != base::constants::BackpressureStrategy::Reliable)
537 : 16 : return try_send_shared(std::move(data), true);
538 [ + + ]: 155 : return blocking_send(
539 [ + - ]: 159 : data ? data->size() : 0, [&](auto& serial, uint64_t sequence) { return serial.write_shared(data, sequence); },
540 : 124 : [&](auto& connection) { return connection.write_shared(data); });
541 : : }
542 : :
543 : 73 : SendResult send_line(std::string_view line) {
544 [ + + ]: 73 : if (backpressure_strategy == base::constants::BackpressureStrategy::Reliable) return send_line_blocking(line);
545 : 11 : return try_send_line(line, true);
546 : : }
547 : :
548 : 38 : SendResult try_send_line(std::string_view line, bool best_effort_send = false) {
549 : 114 : return try_send(std::string(line) + "\n", best_effort_send);
550 : : }
551 : :
552 : 295 : SendResult send_blocking(std::string_view data) {
553 : 295 : auto binary_view = base::safe_convert::string_to_bytes(data);
554 : 295 : memory::ConstByteSpan span(binary_view.first, binary_view.second);
555 : 295 : return blocking_send(
556 : 27 : data.size(), [&](auto& serial, uint64_t sequence) { return serial.write_copy(span, sequence); },
557 : 705 : [&](auto& connection) { return connection.write_copy(span); });
558 : : }
559 : :
560 : 396 : SendResult send_line_blocking(std::string_view line) { return send_blocking(std::string(line) + "\n"); }
561 : :
562 : 2 : RuntimeStats stats() const {
563 : 2 : std::shared_lock<std::shared_mutex> lock(mutex_);
564 [ + - + - ]: 4 : return channel ? channel->stats() : RuntimeStats{};
565 : 2 : }
566 : :
567 : 1 : void reset_stats() {
568 : 1 : std::shared_lock<std::shared_mutex> lock(mutex_);
569 : 1 : receive_budget_->reset_stats();
570 [ + - + - ]: 1 : if (channel) channel->reset_stats();
571 : 1 : }
572 : :
573 : 407 : void setup_internal_handlers() {
574 [ - + ]: 407 : if (!channel) return;
575 : :
576 : 407 : batch_timer_ = std::make_unique<boost::asio::steady_timer>(channel->get_executor());
577 : :
578 : 407 : std::weak_ptr<bool> weak_alive = alive_marker_;
579 : 407 : std::weak_ptr<Impl> weak_impl = weak_from_this();
580 : 407 : receive_budget_->reset_stats();
581 : 407 : lifecycle_events_.reset();
582 : 407 : const auto generation = callback_gate_.open_new_generation();
583 : 407 : callback_generation_.store(generation);
584 : :
585 : 407 : channel->on_bytes([this, generation, weak_impl, weak_alive](memory::ConstByteSpan data) {
586 : 48 : auto impl_keepalive = weak_impl.lock();
587 [ - + ]: 48 : if (!impl_keepalive) return;
588 : 48 : auto alive = weak_alive.lock();
589 [ + + ]: 48 : if (!alive) return;
590 : 47 : auto lease = callback_gate_.enter(generation);
591 [ + + ]: 47 : if (!lease.admitted()) return;
592 : :
593 : : // #449: everything below runs synchronously on this io thread - mark
594 : : // it so a blocking send() called from within one of these callbacks
595 : : // fails fast instead of deadlocking.
596 : 46 : detail::CallbackGuard callback_guard;
597 : :
598 : : // #441: snapshot the handler/framer pointers under a shared_lock (not
599 : : // unique_lock) - this is a pure read, matching try_send's locking
600 : : // level so it no longer blocks concurrent sends even briefly.
601 : : bool batch_mode;
602 : 46 : interface::SharedCallback<MessageHandler> handler;
603 : 46 : std::shared_ptr<detail::ReceiveState> receive;
604 : 46 : std::shared_ptr<framer::IFramer> framer_to_push;
605 : : {
606 : 46 : std::shared_lock<std::shared_mutex> lock(mutex_);
607 : 46 : batch_mode = static_cast<bool>(data_batch_handler_);
608 : 46 : handler = data_handler;
609 : 46 : framer_to_push = framer;
610 : 46 : receive = receive_state_;
611 : 46 : }
612 : :
613 : : try {
614 : 46 : auto prepared = detail::prepare_receive(*receive, framer_to_push, 0, data, batch_mode);
615 [ + + ]: 41 : if (batch_mode) {
616 : : // #441: build the copy before taking the exclusive lock, so the
617 : : // lock is only held for the queue mutation itself, not the
618 : : // allocation.
619 : 15 : auto ctx = std::move(*prepared.raw);
620 : 15 : interface::SharedCallback<BatchMessageHandler> flush_handler;
621 : 15 : detail::ReceiveBatch batch;
622 : : {
623 : 15 : std::unique_lock<std::shared_mutex> lock(mutex_);
624 : 15 : data_batch_queue_.emplace_back(std::move(ctx));
625 [ + + ]: 15 : if (data_batch_queue_.size() >= max_batch_size_) {
626 : 4 : flush_handler = data_batch_handler_;
627 : 4 : batch = std::move(data_batch_queue_);
628 : 4 : data_batch_queue_.clear();
629 [ + - ]: 11 : } else if (data_batch_queue_.size() == 1) {
630 : 11 : schedule_batch_timer(generation);
631 : : }
632 : 15 : }
633 : 15 : detail::invoke_user_callback("serial", "on_data_batch", flush_handler, batch);
634 : 15 : } else {
635 : 52 : detail::invoke_user_callback("serial", "on_data", handler, MessageContext(0, data));
636 : : }
637 : :
638 : 41 : prepared.deliver();
639 : 46 : } catch (const detail::ReceiveOverflow& overflow) {
640 : 5 : receive->scope->overflow(data.size(), overflow.reason);
641 : :
642 : : {
643 : 5 : std::unique_lock<std::shared_mutex> lock(mutex_);
644 : 5 : data_batch_queue_.clear();
645 : 5 : message_batch_queue_.clear();
646 : 5 : receive->reset(framer_to_push.get());
647 : 5 : }
648 [ + - + - ]: 5 : if (auto native = std::dynamic_pointer_cast<transport::Serial>(channel)) native->fail_receive();
649 : :
650 : 5 : } catch (const std::bad_alloc&) {
651 : 0 : receive->scope->overflow(data.size(), ReceiveOverflowReason::AllocationFailure);
652 : :
653 : : {
654 : 0 : std::unique_lock<std::shared_mutex> lock(mutex_);
655 : 0 : data_batch_queue_.clear();
656 : 0 : message_batch_queue_.clear();
657 : 0 : receive->reset(framer_to_push.get());
658 : 0 : }
659 [ - - - - ]: 0 : if (auto native = std::dynamic_pointer_cast<transport::Serial>(channel)) native->fail_receive();
660 : 0 : }
661 : 51 : });
662 : :
663 : 407 : channel->on_backpressure([this, generation, weak_impl, weak_alive](size_t queued) {
664 : 83 : auto impl_keepalive = weak_impl.lock();
665 [ - + ]: 83 : if (!impl_keepalive) return;
666 : 83 : auto alive = weak_alive.lock();
667 [ + + ]: 83 : if (!alive) return;
668 : 82 : auto lease = callback_gate_.enter(generation);
669 [ + + ]: 82 : if (!lease.admitted()) return;
670 : 81 : bp_cv_.notify_all();
671 : 81 : std::function<void(size_t)> handler;
672 : : {
673 : 81 : std::shared_lock<std::shared_mutex> lock(mutex_);
674 : 81 : handler = bp_handler;
675 : 81 : }
676 : 81 : detail::invoke_user_callback("serial", "on_backpressure", handler, queued);
677 : 86 : });
678 : :
679 : 407 : channel->on_state([this, generation, weak_impl, weak_alive](base::LinkState state) {
680 : 411 : auto impl_keepalive = weak_impl.lock();
681 [ - + ]: 411 : if (!impl_keepalive) return;
682 : 411 : auto alive = weak_alive.lock();
683 [ + + ]: 411 : if (!alive) return;
684 : 408 : auto lease = callback_gate_.enter(generation);
685 [ + + ]: 408 : if (!lease.admitted()) return;
686 : 406 : const bool ready = state == base::LinkState::Connected;
687 : 406 : const auto events = lifecycle_events_.observe(state, ready);
688 : 406 : ConnectionHandler connected_handler, lost_handler;
689 : 406 : ErrorHandler terminal_handler;
690 : 406 : std::shared_ptr<interface::Channel> channel_snapshot;
691 [ + + ]: 406 : if (ready) {
692 : 202 : std::unique_lock<std::shared_mutex> lock(mutex_);
693 : 202 : receive_state_->reset(framer.get());
694 : 202 : data_batch_queue_.clear();
695 : 202 : message_batch_queue_.clear();
696 : 202 : fulfill_all_locked(true);
697 : 202 : connected_handler = connect_handler;
698 : 202 : } else {
699 [ + + + + : 204 : if (state == base::LinkState::Closed || state == base::LinkState::Idle || state == base::LinkState::Error) {
+ + ]
700 : 29 : std::unique_lock<std::shared_mutex> lock(mutex_);
701 : 29 : fulfill_all_locked(false);
702 : 29 : }
703 : : // A Connecting notification must not require the exclusive wrapper
704 : : // lock held off by a connection-pinned final admission on another thread.
705 : 204 : std::shared_lock<std::shared_mutex> lock(mutex_);
706 [ + + + - ]: 204 : if (events.disconnect) lost_handler = disconnect_handler;
707 [ + + ]: 204 : if (events.error) {
708 : 23 : terminal_handler = error_handler;
709 : 23 : channel_snapshot = channel;
710 : : }
711 : 204 : }
712 [ + + ]: 406 : if (events.disconnect) {
713 : 40 : flush_batches(generation);
714 : 40 : std::shared_lock<std::shared_mutex> read_lock(mutex_);
715 [ + + ]: 40 : if (framer) {
716 : 8 : read_lock.unlock();
717 : 8 : std::unique_lock<std::shared_mutex> lock(mutex_);
718 : 8 : receive_state_->reset(framer.get());
719 : 8 : }
720 : 40 : }
721 : 406 : auto notification_lease = callback_gate_.enter(generation);
722 [ + + ]: 406 : if (!notification_lease.admitted()) return;
723 : 810 : detail::invoke_user_callback("serial", "on_connect", connected_handler, ConnectionContext(0));
724 : 405 : detail::invoke_user_callback("serial", "on_disconnect", lost_handler, ConnectionContext(0));
725 : : // A loss callback may request stop; do not admit the following terminal
726 : : // error into the closed generation.
727 : 405 : auto error_lease = callback_gate_.enter(generation);
728 [ + + + + : 405 : if (error_lease.admitted() && terminal_handler)
+ + ]
729 [ + - ]: 16 : detail::invoke_user_callback("serial", "on_error", terminal_handler,
730 : 16 : channel_snapshot
731 [ + - + - : 32 : ? detail::build_error_context(*channel_snapshot, "Connection error")
- - ]
732 : : : ErrorContext(ErrorCode::IoError, "Connection error"));
733 : 425 : });
734 : 407 : }
735 : :
736 : 10 : void attach_framer_callback() {
737 [ - + ]: 10 : if (!framer) return;
738 : 10 : framer->on_message([this](memory::ConstByteSpan msg) {
739 : 7 : const auto generation = callback_generation_.load();
740 : 7 : auto message_lease = callback_gate_.enter(generation);
741 [ - + ]: 7 : if (!message_lease.admitted()) return;
742 : 7 : auto prepared = detail::take_prepared_message(0, msg);
743 : : // #441: snapshot under a shared_lock (pure read), build the copy
744 : : // before taking the exclusive lock for queue mutation.
745 : : bool batch_mode;
746 : 7 : interface::SharedCallback<MessageHandler> handler;
747 : : {
748 : 7 : std::shared_lock<std::shared_mutex> lock(mutex_);
749 : 7 : batch_mode = static_cast<bool>(message_batch_handler_);
750 : 7 : handler = message_handler;
751 : 7 : }
752 : :
753 [ + + ]: 7 : if (batch_mode) {
754 [ + - - - ]: 6 : auto ctx = prepared ? std::move(*prepared) : detail::retain_received(receive_state_->scope, 0, msg);
755 : 6 : interface::SharedCallback<BatchMessageHandler> flush_handler;
756 : 6 : detail::ReceiveBatch batch;
757 : : {
758 : 6 : std::unique_lock<std::shared_mutex> lock(mutex_);
759 : 6 : message_batch_queue_.emplace_back(std::move(ctx));
760 [ + + ]: 6 : if (message_batch_queue_.size() >= max_batch_size_) {
761 : 2 : flush_handler = message_batch_handler_;
762 : 2 : batch = std::move(message_batch_queue_);
763 : 2 : message_batch_queue_.clear();
764 [ + - ]: 4 : } else if (message_batch_queue_.size() == 1) {
765 : 4 : schedule_batch_timer(generation);
766 : : }
767 : 6 : }
768 : 6 : detail::invoke_user_callback("serial", "on_message_batch", flush_handler, batch);
769 : 6 : return;
770 : 6 : }
771 : :
772 : 1 : detail::invoke_user_callback("serial", "on_message", handler, MessageContext(0, msg));
773 : 19 : });
774 : : }
775 : :
776 : 22 : void set_framer(std::unique_ptr<framer::IFramer> f) {
777 : 22 : std::unique_lock<std::shared_mutex> lock(mutex_);
778 : 22 : auto receive = std::make_shared<detail::ReceiveState>(receive_state_->scope);
779 : 22 : framer = std::shared_ptr<framer::IFramer>(std::move(f));
780 : 22 : receive_state_ = std::move(receive);
781 [ + - + - : 22 : if (framer && (message_handler || message_batch_handler_)) attach_framer_callback();
- + - + -
- ]
782 : 22 : }
783 : :
784 : 4 : void on_message(MessageHandler handler) {
785 : 4 : std::unique_lock<std::shared_mutex> lock(mutex_);
786 : 4 : message_handler = interface::share_callback(std::move(handler));
787 [ + - + - ]: 4 : if (framer) attach_framer_callback();
788 : 4 : }
789 : :
790 : 6 : void on_message_batch(BatchMessageHandler handler) {
791 : 6 : std::unique_lock<std::shared_mutex> lock(mutex_);
792 : 6 : message_batch_handler_ = interface::share_callback(std::move(handler));
793 [ + - + - ]: 6 : if (framer) attach_framer_callback();
794 : 6 : }
795 : :
796 : 30 : config::SerialConfig build_config_locked() const {
797 : 30 : config::SerialConfig config;
798 : 30 : config.device = device;
799 : 30 : config.baud_rate = baud_rate;
800 : 30 : config.char_size = static_cast<unsigned int>(data_bits);
801 : 30 : config.stop_bits = static_cast<unsigned int>(stop_bits);
802 : 30 : std::string p = to_lower(parity);
803 [ + - + + ]: 30 : if (p == "even")
804 : 3 : config.parity = config::SerialConfig::Parity::Even;
805 [ + - - + ]: 27 : else if (p == "odd")
806 : 0 : config.parity = config::SerialConfig::Parity::Odd;
807 : : else
808 : 27 : config.parity = config::SerialConfig::Parity::None;
809 : :
810 : 30 : std::string f = to_lower(flow_control);
811 [ + - - + ]: 30 : if (f == "software")
812 : 0 : config.flow = config::SerialConfig::Flow::Software;
813 [ + - + + ]: 30 : else if (f == "hardware")
814 : 3 : config.flow = config::SerialConfig::Flow::Hardware;
815 : : else
816 : 27 : config.flow = config::SerialConfig::Flow::None;
817 : :
818 : 30 : config.retry_interval_ms = static_cast<unsigned int>(retry_interval.count());
819 : 30 : config.reopen_on_error = reopen_on_error;
820 : 30 : config.read_chunk = read_chunk;
821 : 30 : config.low_latency = low_latency;
822 : 30 : config.rs485 = rs485;
823 : 30 : config.dtr = dtr;
824 : 30 : config.keep_dtr_on_close = keep_dtr_on_close;
825 : 30 : config.rts = rts;
826 : 30 : config.rx_idle_timeout_ms = static_cast<unsigned>(rx_idle_timeout.count());
827 : 30 : config.backpressure_threshold = backpressure_threshold;
828 : 30 : config.backpressure_strategy = backpressure_strategy;
829 : 30 : config.use_shared_context = shared_context_.load();
830 : 60 : return config;
831 : 30 : }
832 : : };
833 : :
834 : 40 : Serial::Serial(const std::string& d, uint32_t b) : impl_(std::make_shared<Impl>(d, b)) {
835 : 40 : config::detail::require(util::InputValidator::is_valid_device_path(d), "invalid serial device");
836 : 39 : config::detail::range(b, base::constants::MIN_BAUD_RATE, base::constants::MAX_BAUD_RATE, "invalid baud rate");
837 : 41 : }
838 : 3 : Serial::Serial(const std::string& d, uint32_t b, std::shared_ptr<boost::asio::io_context> i)
839 : 3 : : impl_(std::make_shared<Impl>(d, b, i)) {
840 : 3 : config::detail::require(util::InputValidator::is_valid_device_path(d), "invalid serial device");
841 : 3 : config::detail::range(b, base::constants::MIN_BAUD_RATE, base::constants::MAX_BAUD_RATE, "invalid baud rate");
842 : 3 : }
843 : 214 : Serial::Serial(std::shared_ptr<interface::Channel> ch) : impl_(std::make_shared<Impl>(ch)) {
844 : 211 : impl_->setup_internal_handlers();
845 : 214 : }
846 : : // Stop on the destroying thread first. A callback in flight holds the Impl
847 : : // alive, and if its reference were the last one, ~Impl would run on the io
848 : : // thread, where the shutdown cannot join that thread (#613).
849 : 366 : Serial::~Serial() {
850 [ - + ]: 253 : if (!impl_) return;
851 : : try {
852 : 253 : impl_->stop();
853 : 0 : } catch (...) {
854 : 0 : }
855 : 366 : }
856 : :
857 : 0 : Serial::Serial(Serial&&) noexcept = default;
858 : : // The previous Impl is handed to a temporary, whose destructor stops it here.
859 : 0 : Serial& Serial::operator=(Serial&& other) noexcept {
860 [ # # ]: 0 : if (this != &other) {
861 : 0 : Serial previous(std::move(*this));
862 : 0 : impl_ = std::move(other.impl_);
863 : 0 : }
864 : 0 : return *this;
865 : : }
866 : :
867 : 9 : Serial& Serial::receive_limits(ReceiveLimits limits) {
868 : 9 : limits.validate();
869 : 8 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
870 [ + + + - : 22 : if (impl_->started_ || impl_->stop_callers_.load() || (impl_->stop_requested_ && impl_->alive_marker_) ||
+ + + - -
+ + + ]
871 : 7 : detail::in_data_callback())
872 : 1 : throw std::logic_error("receive limits require completed stop");
873 : 7 : auto budget = std::make_shared<detail::ReceiveBudget>(limits);
874 : 7 : auto scope = budget->open_scope();
875 : 7 : auto receive = std::make_shared<detail::ReceiveState>(std::move(scope));
876 : 7 : impl_->receive_state_->reset(impl_->framer.get());
877 : 7 : impl_->receive_limits_ = limits;
878 : 7 : impl_->receive_budget_ = std::move(budget);
879 : 7 : impl_->receive_state_ = std::move(receive);
880 : 7 : return *this;
881 : 8 : }
882 : 54 : ReceiveMemoryStats Serial::receive_stats() const {
883 : 54 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
884 : 108 : return impl_->receive_budget_->stats();
885 : 54 : }
886 : :
887 : 255 : std::future<bool> Serial::start() { return impl_->start(); }
888 : 358 : void Serial::stop() { impl_->stop(); }
889 : 102 : SendResult Serial::send(std::string_view data) { return impl_->send(data); }
890 : 41 : SendResult Serial::try_send(std::string_view data) { return impl_->try_send(data); }
891 : 73 : SendResult Serial::send_line(std::string_view line) { return impl_->send_line(line); }
892 : 27 : SendResult Serial::try_send_line(std::string_view line) { return impl_->try_send_line(line); }
893 : 87 : SendResult Serial::send_move(std::vector<uint8_t>&& data) { return impl_->send_move(std::move(data)); }
894 : 27 : SendResult Serial::try_send_move(std::vector<uint8_t>&& data) { return impl_->try_send_move(std::move(data)); }
895 : 99 : SendResult Serial::send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
896 : 99 : return impl_->send_shared(std::move(data));
897 : : }
898 : 33 : SendResult Serial::try_send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
899 : 33 : return impl_->try_send_shared(std::move(data));
900 : : }
901 : 74 : SendResult Serial::send_blocking(std::string_view data) { return impl_->send_blocking(data); }
902 : 70 : SendResult Serial::send_line_blocking(std::string_view line) { return impl_->send_line_blocking(line); }
903 : 8 : bool Serial::connected() const {
904 : 8 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
905 [ + + + - : 16 : return impl_->channel && impl_->channel->is_connected();
+ + ]
906 : 8 : }
907 : 2 : RuntimeStats Serial::stats() const { return impl_->stats(); }
908 : 1 : void Serial::reset_stats() { impl_->reset_stats(); }
909 : :
910 : 38 : Serial& Serial::on_data(MessageHandler h) {
911 : 38 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
912 : 38 : impl_->data_handler = interface::share_callback(std::move(h));
913 : 38 : return *this;
914 : 38 : }
915 : 15 : Serial& Serial::on_data_batch(BatchMessageHandler h) {
916 : 15 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
917 : 15 : impl_->data_batch_handler_ = interface::share_callback(std::move(h));
918 : 15 : return *this;
919 : 15 : }
920 : 70 : Serial& Serial::on_connect(ConnectionHandler h) {
921 : 70 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
922 : 70 : impl_->connect_handler = std::move(h);
923 : 70 : return *this;
924 : 70 : }
925 : 18 : Serial& Serial::on_disconnect(ConnectionHandler h) {
926 : 18 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
927 : 18 : impl_->disconnect_handler = std::move(h);
928 : 18 : return *this;
929 : 18 : }
930 : 38 : Serial& Serial::on_error(ErrorHandler h) {
931 : 38 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
932 : 38 : impl_->error_handler = std::move(h);
933 : 38 : return *this;
934 : 38 : }
935 : :
936 : 6 : Serial& Serial::on_backpressure(std::function<void(size_t)> h) {
937 : 6 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
938 : 6 : impl_->bp_handler = std::move(h);
939 : 6 : return *this;
940 : 6 : }
941 : :
942 : 22 : Serial& Serial::framer(std::unique_ptr<framer::IFramer> f) {
943 : 22 : impl_->set_framer(std::move(f));
944 : 22 : return *this;
945 : : }
946 : 4 : Serial& Serial::on_message(MessageHandler h) {
947 : 4 : impl_->on_message(std::move(h));
948 : 4 : return *this;
949 : : }
950 : 6 : Serial& Serial::on_message_batch(BatchMessageHandler h) {
951 : 6 : impl_->on_message_batch(std::move(h));
952 : 6 : return *this;
953 : : }
954 : :
955 : 2 : Serial& Serial::auto_start(bool m) {
956 : 2 : impl_->auto_start_.store(m);
957 [ + - + - : 2 : if (impl_->auto_start_.load() && !impl_->started_.load()) start();
+ - + - ]
958 : 2 : return *this;
959 : : }
960 : :
961 : 3 : Serial& Serial::shared_context(bool use_shared) {
962 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
963 : 3 : impl_->require_stopped_config();
964 : 3 : impl_->shared_context_.store(use_shared);
965 : 3 : return *this;
966 : 3 : }
967 : :
968 : 1 : Serial& Serial::baud_rate(uint32_t b) {
969 : 1 : config::detail::range(b, base::constants::MIN_BAUD_RATE, base::constants::MAX_BAUD_RATE, "invalid baud_rate");
970 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
971 : 1 : impl_->require_stopped_config();
972 : 1 : impl_->baud_rate = b;
973 : 1 : return *this;
974 : 1 : }
975 : 9 : Serial& Serial::data_bits(int d) {
976 : 9 : config::detail::range(d, base::constants::MIN_DATA_BITS, base::constants::MAX_DATA_BITS, "invalid data_bits");
977 : 8 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
978 : 8 : impl_->require_stopped_config();
979 : 8 : impl_->data_bits = d;
980 : 8 : return *this;
981 : 8 : }
982 : 8 : Serial& Serial::stop_bits(int s) {
983 : 8 : config::detail::range(s, base::constants::MIN_STOP_BITS, base::constants::MAX_STOP_BITS, "invalid stop_bits");
984 : 7 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
985 : 7 : impl_->require_stopped_config();
986 : 7 : impl_->stop_bits = s;
987 : 7 : return *this;
988 : 7 : }
989 : 11 : Serial& Serial::parity(const std::string& p) {
990 : 11 : config::detail::text_option(p, {"none", "even", "odd"}, "invalid parity");
991 : 10 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
992 : 10 : impl_->require_stopped_config();
993 : 10 : impl_->parity = p;
994 : 10 : return *this;
995 : 10 : }
996 : 11 : Serial& Serial::flow_control(const std::string& f) {
997 : 11 : config::detail::text_option(f, {"none", "software", "hardware"}, "invalid flow control");
998 : 10 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
999 : 10 : impl_->require_stopped_config();
1000 : 10 : impl_->flow_control = f;
1001 : 10 : return *this;
1002 : 10 : }
1003 : 1 : Serial& Serial::read_chunk(size_t bytes) {
1004 : 1 : config::detail::range(bytes, base::constants::MIN_READ_BUFFER_SIZE, base::constants::MAX_READ_BUFFER_SIZE,
1005 : : "invalid read_chunk");
1006 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1007 : 1 : impl_->require_stopped_config();
1008 : 1 : impl_->read_chunk = bytes;
1009 : 1 : return *this;
1010 : 1 : }
1011 : :
1012 : 1 : Serial& Serial::low_latency(bool enable) {
1013 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1014 : 1 : impl_->require_stopped_config();
1015 : 1 : impl_->low_latency = enable;
1016 : 1 : return *this;
1017 : 1 : }
1018 : :
1019 : 1 : Serial& Serial::rs485(bool rts_on_send, bool rx_during_tx, unsigned delay_before_ms, unsigned delay_after_ms) {
1020 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1021 : 1 : impl_->require_stopped_config();
1022 : 1 : impl_->rs485.enabled = true;
1023 : 1 : impl_->rs485.rts_on_send = rts_on_send;
1024 : 1 : impl_->rs485.rx_during_tx = rx_during_tx;
1025 : 1 : impl_->rs485.delay_rts_before_send_ms = delay_before_ms;
1026 : 1 : impl_->rs485.delay_rts_after_send_ms = delay_after_ms;
1027 : 1 : return *this;
1028 : 1 : }
1029 : :
1030 : 5 : Serial& Serial::dtr(bool assert_line) {
1031 : 5 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1032 : 5 : impl_->require_stopped_config();
1033 [ + - + + : 5 : config::detail::require(assert_line || !impl_->keep_dtr_on_close, "keep_dtr_on_close conflicts with dtr(false)");
+ + ]
1034 : 4 : impl_->dtr = assert_line;
1035 : 4 : return *this;
1036 : 5 : }
1037 : :
1038 : 6 : Serial& Serial::keep_dtr_on_close(bool enable) {
1039 : 6 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1040 : 6 : impl_->require_stopped_config();
1041 [ + + + + : 6 : config::detail::require(!enable || impl_->dtr != false, "keep_dtr_on_close conflicts with dtr(false)");
+ + ]
1042 : 3 : impl_->keep_dtr_on_close = enable;
1043 : 3 : return *this;
1044 : 6 : }
1045 : :
1046 : 1 : Serial& Serial::rts(bool assert_line) {
1047 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1048 : 1 : impl_->require_stopped_config();
1049 : 1 : impl_->rts = assert_line;
1050 : 1 : return *this;
1051 : 1 : }
1052 : :
1053 : 2 : Serial& Serial::rx_idle_timeout(std::chrono::milliseconds timeout) {
1054 : 2 : config::detail::duration(timeout, base::constants::MIN_IDLE_TIMEOUT_MS, base::constants::MAX_IDLE_TIMEOUT_MS, true,
1055 : : "invalid rx_idle_timeout");
1056 : 2 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1057 : 2 : impl_->require_stopped_config();
1058 : 2 : impl_->rx_idle_timeout = timeout;
1059 : 2 : return *this;
1060 : 2 : }
1061 : :
1062 : 7 : Serial& Serial::reopen_on_error(bool enable) {
1063 : 7 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1064 : 7 : impl_->require_stopped_config();
1065 : 7 : impl_->reopen_on_error = enable;
1066 : 7 : return *this;
1067 : 7 : }
1068 : 9 : Serial& Serial::retry_interval(std::chrono::milliseconds i) {
1069 : 9 : config::detail::duration(i, base::constants::MIN_RETRY_INTERVAL_MS, base::constants::MAX_RETRY_INTERVAL_MS, false,
1070 : : "invalid retry_interval");
1071 : 7 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1072 : 7 : impl_->retry_interval = i;
1073 [ + + ]: 7 : if (impl_->channel) {
1074 : 1 : auto ts = std::dynamic_pointer_cast<transport::Serial>(impl_->channel);
1075 [ - + - - ]: 1 : if (ts) ts->set_retry_interval(static_cast<unsigned int>(i.count()));
1076 : 1 : }
1077 : 7 : return *this;
1078 : 7 : }
1079 : :
1080 : 29 : Serial& Serial::backpressure_threshold(size_t threshold) {
1081 : 29 : config::detail::range(threshold, base::constants::MIN_BACKPRESSURE_THRESHOLD,
1082 : : base::constants::MAX_BACKPRESSURE_THRESHOLD, "invalid backpressure_threshold");
1083 : 27 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1084 : 27 : impl_->require_stopped_config();
1085 : 25 : impl_->backpressure_threshold = threshold;
1086 : 25 : return *this;
1087 : 27 : }
1088 : :
1089 : 77 : Serial& Serial::backpressure_strategy(base::constants::BackpressureStrategy strategy) {
1090 : 77 : config::detail::strategy(strategy);
1091 : 76 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1092 : 76 : impl_->backpressure_strategy = strategy;
1093 [ + + ]: 76 : if (impl_->channel) {
1094 : 73 : auto ts = std::dynamic_pointer_cast<transport::Serial>(impl_->channel);
1095 [ + + + - ]: 73 : if (ts) ts->set_backpressure_strategy(strategy);
1096 : 73 : }
1097 : 76 : return *this;
1098 : 76 : }
1099 : :
1100 : 7 : size_t Serial::backpressure_threshold() const {
1101 : 7 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
1102 : 14 : return impl_->backpressure_threshold;
1103 : 7 : }
1104 : :
1105 : 3 : base::constants::BackpressureStrategy Serial::backpressure_strategy() const {
1106 : 3 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
1107 : 6 : return impl_->backpressure_strategy;
1108 : 3 : }
1109 : :
1110 : 10 : config::SerialConfig Serial::build_config() const {
1111 : 10 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
1112 : 20 : return impl_->build_config_locked();
1113 : 10 : }
1114 : :
1115 : 4 : Serial& Serial::manage_external_context(bool m) {
1116 : 4 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1117 : 4 : impl_->require_stopped_config();
1118 : 3 : impl_->manage_external_context.store(m);
1119 : 3 : return *this;
1120 : 4 : }
1121 : :
1122 : 26 : Serial& Serial::batch_size(size_t size) {
1123 [ + + ]: 26 : config::detail::require(size > 0, "batch size must be positive");
1124 : 25 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1125 : 25 : impl_->max_batch_size_ = size;
1126 : 25 : return *this;
1127 : 25 : }
1128 : :
1129 : 24 : Serial& Serial::batch_latency(std::chrono::milliseconds latency) {
1130 : 24 : config::detail::duration(latency, 0, std::numeric_limits<int>::max(), true, "invalid batch latency");
1131 : 23 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
1132 : 23 : impl_->max_batch_latency_ = latency;
1133 : 23 : return *this;
1134 : 23 : }
1135 : :
1136 : : } // namespace wrapper
1137 : : } // namespace wirestead
|
