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 <cctype>
25 : : #include <iostream>
26 : : #include <mutex>
27 : : #include <optional>
28 : : #include <shared_mutex>
29 : : #include <stdexcept>
30 : : #include <thread>
31 : : #include <vector>
32 : :
33 : : #include "wirestead/base/common.hpp"
34 : : #include "wirestead/base/constants.hpp"
35 : : #include "wirestead/concurrency/io_thread_hook.hpp"
36 : : #include "wirestead/factory/channel_factory.hpp"
37 : : #include "wirestead/transport/serial/serial.hpp"
38 : : #include "wirestead/wrapper/callback_guard.hpp"
39 : : #include "wirestead/wrapper/error_context_builder.hpp"
40 : :
41 : : namespace wirestead {
42 : : namespace wrapper {
43 : :
44 : : namespace {
45 : 38 : std::string to_lower(std::string s) {
46 : 216 : std::transform(s.begin(), s.end(), s.begin(), [](unsigned char c) { return std::tolower(c); });
47 : 38 : return s;
48 : : }
49 : : } // namespace
50 : :
51 : : struct Serial::Impl : public std::enable_shared_from_this<Impl> {
52 : : mutable std::shared_mutex mutex_;
53 : : std::mutex bp_mutex_;
54 : : std::condition_variable bp_cv_;
55 : : std::string device;
56 : : uint32_t baud_rate;
57 : : std::shared_ptr<interface::Channel> channel;
58 : : std::shared_ptr<boost::asio::io_context> external_ioc;
59 : : std::atomic<bool> use_external_context{false};
60 : : std::atomic<bool> manage_external_context{false};
61 : : std::thread external_thread;
62 : : std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>> work_guard_;
63 : :
64 : : std::vector<std::promise<bool>> pending_promises_;
65 : : std::atomic<bool> started_{false};
66 : 51 : std::shared_ptr<bool> alive_marker_{std::make_shared<bool>(true)};
67 : :
68 : : // Event handlers (Context based)
69 : : // Shared snapshots: the strand copies one out per received chunk, and a
70 : : // std::function copy allocates whenever the user handler outgrows its
71 : : // small-object buffer. See interface::SharedCallback.
72 : : interface::SharedCallback<MessageHandler> data_handler;
73 : : interface::SharedCallback<BatchMessageHandler> data_batch_handler_;
74 : : ConnectionHandler connect_handler{nullptr};
75 : : ConnectionHandler disconnect_handler{nullptr};
76 : : ErrorHandler error_handler{nullptr};
77 : : std::function<void(size_t)> bp_handler{nullptr};
78 : : interface::SharedCallback<MessageHandler> message_handler;
79 : : interface::SharedCallback<BatchMessageHandler> message_batch_handler_;
80 : :
81 : : std::shared_ptr<framer::IFramer> framer{nullptr};
82 : :
83 : : // Batching logic
84 : : std::vector<MessageContext> data_batch_queue_;
85 : : std::vector<MessageContext> message_batch_queue_;
86 : : std::unique_ptr<boost::asio::steady_timer> batch_timer_;
87 : : size_t max_batch_size_ = 100;
88 : 51 : std::chrono::milliseconds max_batch_latency_{1};
89 : :
90 : : // Configuration
91 : : std::atomic<bool> auto_start_ = false;
92 : : std::atomic<bool> shared_context_{false};
93 : : int data_bits = 8;
94 : : int stop_bits = 1;
95 : : std::string parity = "none";
96 : : std::string flow_control = "none";
97 : : bool reopen_on_error = true;
98 : : size_t read_chunk = base::constants::DEFAULT_READ_BUFFER_SIZE;
99 : : bool low_latency = true;
100 : : config::SerialConfig::Rs485 rs485{};
101 : : std::optional<bool> dtr;
102 : : std::optional<bool> rts;
103 : 51 : std::chrono::milliseconds rx_idle_timeout{0};
104 : : std::chrono::milliseconds retry_interval{base::constants::DEFAULT_RETRY_INTERVAL_MS};
105 : : size_t backpressure_threshold = base::constants::DEFAULT_BACKPRESSURE_THRESHOLD;
106 : : base::constants::BackpressureStrategy backpressure_strategy = base::constants::BackpressureStrategy::Reliable;
107 : :
108 : : // False only for the dependency-injected-channel constructor below, where
109 : : // the caller owns the channel's identity and lifecycle (e.g. tests
110 : : // injecting a fake channel) - stop() must not discard and factory-rebuild
111 : : // a channel it didn't create itself.
112 : : bool factory_managed_channel_ = true;
113 : :
114 : 170 : Impl(const std::string& dev, uint32_t baud) : device(dev), baud_rate(baud) {}
115 : 2 : Impl(const std::string& dev, uint32_t baud, std::shared_ptr<boost::asio::io_context> ioc)
116 [ + - + - : 10 : : device(dev), baud_rate(baud), external_ioc(std::move(ioc)), use_external_context(external_ioc != nullptr) {}
+ - + - ]
117 : 75 : explicit Impl(std::shared_ptr<interface::Channel> ch) : channel(std::move(ch)), factory_managed_channel_(false) {
118 : : // #450: setup_internal_handlers() captures weak_from_this() - calling it
119 : : // from inside this constructor would capture an empty weak_ptr, since
120 : : // enable_shared_from_this isn't wired up until make_shared() finishes
121 : : // constructing the object. Deferred to Serial's own constructor, which
122 : : // runs after impl_ is a fully-formed shared_ptr<Impl> (already called
123 : : // there for this constructor, so nothing to add there).
124 : 15 : }
125 : :
126 : 51 : ~Impl() {
127 : : try {
128 : 51 : stop();
129 : 0 : } catch (...) {
130 : 0 : }
131 : 51 : }
132 : :
133 : 88 : void fulfill_all_locked(bool value) {
134 [ + + ]: 109 : for (auto& promise : pending_promises_) {
135 : : try {
136 : 21 : promise.set_value(value);
137 : 0 : } catch (...) {
138 : 0 : }
139 : : }
140 : 88 : pending_promises_.clear();
141 : 88 : }
142 : :
143 : 1 : void flush_batches() {
144 : 1 : std::unique_lock<std::shared_mutex> lock(mutex_);
145 [ + - ]: 1 : if (!data_batch_queue_.empty()) {
146 : 1 : auto handler = data_batch_handler_;
147 : 1 : auto batch = std::move(data_batch_queue_);
148 : 1 : data_batch_queue_.clear();
149 [ + - ]: 1 : if (handler) {
150 : 1 : lock.unlock();
151 : 1 : detail::invoke_user_callback("serial", "on_data_batch", handler, batch);
152 : 1 : lock.lock();
153 : : }
154 : 1 : }
155 [ + - ]: 1 : if (!message_batch_queue_.empty()) {
156 : 1 : auto handler = message_batch_handler_;
157 : 1 : auto batch = std::move(message_batch_queue_);
158 : 1 : message_batch_queue_.clear();
159 [ + - ]: 1 : if (handler) {
160 : 1 : lock.unlock();
161 : 1 : detail::invoke_user_callback("serial", "on_message_batch", handler, batch);
162 : 1 : lock.lock();
163 : : }
164 : 1 : }
165 [ + - ]: 1 : if (batch_timer_) {
166 : 1 : batch_timer_->cancel();
167 : : }
168 : 1 : }
169 : :
170 : 5 : void schedule_batch_timer() {
171 [ - + ]: 5 : if (!batch_timer_) return;
172 : 5 : batch_timer_->expires_after(max_batch_latency_);
173 : 10 : batch_timer_->async_wait([this, weak_impl = weak_from_this(),
174 : 5 : weak_alive = std::weak_ptr<bool>(alive_marker_)](const boost::system::error_code& ec) {
175 [ + + ]: 2 : if (ec) return;
176 : 1 : auto impl_keepalive = weak_impl.lock();
177 [ - + ]: 1 : if (!impl_keepalive) return;
178 : 1 : auto alive = weak_alive.lock();
179 [ - + ]: 1 : if (!alive) return;
180 : 1 : flush_batches();
181 : 1 : });
182 : : }
183 : :
184 : 26 : std::future<bool> start() {
185 : 26 : std::unique_lock<std::shared_mutex> lock(mutex_);
186 [ + + + - : 26 : if (channel && channel->is_connected()) {
+ + + + ]
187 : 5 : started_.store(true);
188 : 5 : std::promise<bool> p;
189 : 5 : p.set_value(true);
190 : 5 : return p.get_future();
191 : 5 : }
192 : :
193 : 21 : std::promise<bool> p;
194 : 21 : auto future = p.get_future();
195 : 21 : pending_promises_.emplace_back(std::move(p));
196 : :
197 [ + + ]: 21 : if (started_.load()) {
198 : 1 : return future;
199 : : }
200 : :
201 [ + + ]: 20 : if (!channel) {
202 : 13 : channel = factory::ChannelFactory::create(build_config_locked(), external_ioc);
203 : 13 : setup_internal_handlers();
204 : : }
205 : 20 : started_.store(true);
206 : :
207 : 20 : lock.unlock();
208 : 20 : channel->start();
209 [ + + + - : 20 : if (use_external_context && manage_external_context && !external_thread.joinable()) {
+ - + + ]
210 [ + - + - : 1 : if (external_ioc && external_ioc->stopped()) {
- + - + ]
211 : 0 : external_ioc->restart();
212 : : }
213 : 2 : work_guard_ = std::make_unique<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>>(
214 : 3 : boost::asio::make_work_guard(*external_ioc));
215 : 2 : external_thread = std::thread([ioc = external_ioc]() {
216 : 1 : wirestead::concurrency::run_io_thread_init();
217 : : try {
218 : 1 : ioc->run();
219 : 0 : } catch (...) {
220 : 0 : }
221 : 2 : });
222 : : }
223 : :
224 : 20 : return future;
225 : 26 : }
226 : :
227 : 72 : void stop() {
228 : 72 : std::unique_lock<std::shared_mutex> lock(mutex_);
229 [ + + ]: 72 : if (!started_.load()) {
230 : 48 : fulfill_all_locked(false);
231 : 48 : return;
232 : : }
233 : 24 : started_.store(false);
234 : 24 : bp_cv_.notify_all();
235 : :
236 [ + - ]: 24 : if (batch_timer_) {
237 : 24 : batch_timer_->cancel();
238 : 24 : batch_timer_.reset();
239 : : }
240 : :
241 [ + - ]: 24 : if (channel) {
242 : 24 : lock.unlock();
243 : 24 : channel->stop();
244 : : // Clear callbacks after stop() rather than before: the transport-level
245 : : // fix (#436) already synchronizes callback reads against these
246 : : // setters, but clearing after stop() means no in-flight handler can
247 : : // observe a null callback mid-shutdown in the first place - belt and
248 : : // braces once the underlying race is fixed at the source.
249 : 24 : channel->on_bytes(nullptr);
250 : 24 : channel->on_state(nullptr);
251 : 24 : channel->on_backpressure(nullptr);
252 : 24 : lock.lock();
253 [ + + ]: 24 : if (factory_managed_channel_) {
254 : : // Fully release the channel rather than reusing it: start()'s
255 : : // `if (!channel)` guard is what re-runs setup_internal_handlers() and
256 : : // rebuilds config from the (possibly changed) staged fields. Reusing
257 : : // a stopped channel left every handler nulled forever - including
258 : : // the one that fulfills the start() future - so a restart would
259 : : // hang (jwsung91/wirestead#444). Injected channels (factory_managed_
260 : : // channel_ == false) are exempt: the caller owns that channel's
261 : : // identity, so we must not discard and factory-rebuild it.
262 : 13 : channel.reset();
263 : : }
264 : : }
265 : :
266 [ + + ]: 24 : if (work_guard_) {
267 : 1 : work_guard_.reset();
268 : : }
269 : :
270 [ + + + - : 24 : if (use_external_context && manage_external_context && external_thread.joinable()) {
+ - + + ]
271 [ + - + - ]: 1 : if (external_ioc) external_ioc->stop();
272 [ + - ]: 1 : if (std::this_thread::get_id() != external_thread.get_id()) {
273 : 1 : lock.unlock();
274 : 1 : external_thread.join();
275 : 1 : lock.lock();
276 : : } else {
277 : 0 : external_thread.detach();
278 : : }
279 : : }
280 : :
281 : 24 : fulfill_all_locked(false);
282 : :
283 [ + + + - ]: 24 : if (framer) framer->reset();
284 : 72 : }
285 : :
286 : 5 : bool try_send(std::string_view data) {
287 : 5 : std::shared_lock<std::shared_mutex> lock(mutex_);
288 [ + - + - : 5 : if (channel && channel->is_connected()) {
+ - + - ]
289 : 5 : auto binary_view = base::safe_convert::string_to_bytes(data);
290 : 5 : return channel->async_try_write_copy(memory::ConstByteSpan(binary_view.first, binary_view.second));
291 : : }
292 : 0 : return false;
293 : 5 : }
294 : :
295 : 3 : bool try_send_move(std::vector<uint8_t>&& data) {
296 : 3 : std::shared_lock<std::shared_mutex> lock(mutex_);
297 [ + - + - : 3 : if (channel && channel->is_connected()) {
+ - + - ]
298 : 3 : return channel->async_try_write_move(std::move(data));
299 : : }
300 : 0 : return false;
301 : 3 : }
302 : :
303 : 5 : bool try_send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
304 [ + + + + : 5 : if (!data || data->empty()) return false;
+ + ]
305 : 3 : std::shared_lock<std::shared_mutex> lock(mutex_);
306 [ + - + - : 3 : if (channel && channel->is_connected()) {
+ - + - ]
307 : 3 : return channel->async_try_write_shared(std::move(data));
308 : : }
309 : 0 : return false;
310 : 3 : }
311 : :
312 : 7 : bool send(std::string_view data) {
313 [ + + ]: 7 : if (backpressure_strategy == base::constants::BackpressureStrategy::Reliable) return send_blocking(data);
314 : 1 : return try_send(data);
315 : : }
316 : :
317 : : // channel->on_backpressure() calls bp_cv_.notify_all() from the transport's io_context
318 : : // thread without holding bp_mutex_ (backpressure_active_ is a plain atomic on the transport
319 : : // side, not guarded by bp_mutex_ at all). That makes a classic lost-wakeup race possible: a
320 : : // waiter can check the predicate, find it still blocking, and be in the process of
321 : : // registering to wait when the notify fires - in the rare case that race is lost, an
322 : : // unbounded wait() would block forever. Poll with a bounded timeout instead so a missed
323 : : // notify only costs a short delay rather than a permanent hang (see #427, #431).
324 : : // Returns false instead of waiting if called from the channel's own io
325 : : // thread while backpressure is active - e.g. a blocking send() called
326 : : // from inside an on_data/on_message callback. Clearing backpressure
327 : : // requires that same io thread to make progress, so blocking here would
328 : : // deadlock forever rather than eventually clear (#449).
329 : 13 : bool wait_for_backpressure_clear(std::unique_lock<std::mutex>& bp_lock) {
330 : 17 : auto predicate = [this] {
331 : 17 : std::shared_lock<std::shared_mutex> lock(mutex_);
332 [ + + + - : 34 : return !started_.load() || !channel || !channel->is_connected() || !channel->is_backpressure_active();
+ - + + +
- + + ]
333 : 17 : };
334 [ + - + + ]: 13 : if (predicate()) return true;
335 [ - + ]: 1 : if (detail::in_data_callback()) return false;
336 [ + - + + ]: 2 : while (!bp_cv_.wait_for(bp_lock, std::chrono::milliseconds(50), predicate)) {
337 : : }
338 : 1 : return true;
339 : : }
340 : :
341 : : // #509: wait_for_backpressure_clear()'s condition and the transport's own
342 : : // hard queue-byte cap are different thresholds observed at different
343 : : // times, so a single write attempt can spuriously fail right after the
344 : : // wait exits. Bounded retry rather than unbounded, so a payload that can
345 : : // never fit still fails in bounded time.
346 : : static constexpr int kMaxBlockingSendAttempts = 5;
347 : :
348 : 2 : bool send_move(std::vector<uint8_t>&& data) {
349 [ + + ]: 2 : if (backpressure_strategy == base::constants::BackpressureStrategy::Reliable) {
350 [ + - ]: 1 : for (int attempt = 0; attempt < kMaxBlockingSendAttempts; ++attempt) {
351 : 1 : std::unique_lock<std::mutex> bp_lock(bp_mutex_);
352 [ + - - + ]: 1 : if (!wait_for_backpressure_clear(bp_lock)) return false;
353 : 1 : bp_lock.unlock();
354 : 1 : std::shared_lock<std::shared_mutex> lock(mutex_);
355 [ + - + - : 1 : if (!started_.load() || !channel || !channel->is_connected()) return false;
+ - - + -
+ ]
356 [ + - + - ]: 1 : if (channel->async_write_move(std::move(data))) return true;
357 : 2 : }
358 : 0 : return false;
359 : : }
360 : 1 : return try_send_move(std::move(data));
361 : : }
362 : :
363 : 2 : bool send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
364 [ + - - + : 2 : if (!data || data->empty()) return false;
- + ]
365 [ + + ]: 2 : if (backpressure_strategy == base::constants::BackpressureStrategy::Reliable) {
366 [ + - ]: 1 : for (int attempt = 0; attempt < kMaxBlockingSendAttempts; ++attempt) {
367 : 1 : std::unique_lock<std::mutex> bp_lock(bp_mutex_);
368 [ + - - + ]: 1 : if (!wait_for_backpressure_clear(bp_lock)) return false;
369 : 1 : bp_lock.unlock();
370 : 1 : std::shared_lock<std::shared_mutex> lock(mutex_);
371 [ + - + - : 1 : if (!started_.load() || !channel || !channel->is_connected()) return false;
+ - - + -
+ ]
372 [ + - + - ]: 1 : if (channel->async_write_shared(data)) return true;
373 : 2 : }
374 : 0 : return false;
375 : : }
376 : 1 : return try_send_shared(std::move(data));
377 : : }
378 : :
379 : 3 : bool send_line(std::string_view line) {
380 [ + + ]: 3 : if (backpressure_strategy == base::constants::BackpressureStrategy::Reliable) return send_line_blocking(line);
381 : 1 : return try_send_line(line);
382 : : }
383 : :
384 : 6 : bool try_send_line(std::string_view line) { return try_send(std::string(line) + "\n"); }
385 : :
386 : 10 : bool send_blocking(std::string_view data) {
387 : 10 : auto binary_view = base::safe_convert::string_to_bytes(data);
388 : 10 : memory::ConstByteSpan span(binary_view.first, binary_view.second);
389 [ + - ]: 11 : for (int attempt = 0; attempt < kMaxBlockingSendAttempts; ++attempt) {
390 : 11 : std::unique_lock<std::mutex> bp_lock(bp_mutex_);
391 [ + - - + ]: 11 : if (!wait_for_backpressure_clear(bp_lock)) return false;
392 : 11 : bp_lock.unlock();
393 : 11 : std::shared_lock<std::shared_mutex> lock(mutex_);
394 [ + + + - : 11 : if (!started_.load() || !channel || !channel->is_connected()) return false;
+ - + + +
+ ]
395 : : // #509: previously discarded async_write_copy()'s result and always
396 : : // returned true here, silently reporting success even when the write
397 : : // was actually rejected - now correctly forwards/retries like the
398 : : // other transports.
399 [ + - + + ]: 6 : if (channel->async_write_copy(span)) return true;
400 : 21 : }
401 : 0 : return false;
402 : : }
403 : :
404 : 9 : bool send_line_blocking(std::string_view line) { return send_blocking(std::string(line) + "\n"); }
405 : :
406 : 2 : RuntimeStats stats() const {
407 : 2 : std::shared_lock<std::shared_mutex> lock(mutex_);
408 [ + - + - ]: 4 : return channel ? channel->stats() : RuntimeStats{};
409 : 2 : }
410 : :
411 : 0 : void reset_stats() {
412 : 0 : std::shared_lock<std::shared_mutex> lock(mutex_);
413 [ # # # # ]: 0 : if (channel) channel->reset_stats();
414 : 0 : }
415 : :
416 : 28 : void setup_internal_handlers() {
417 [ - + ]: 28 : if (!channel) return;
418 : :
419 : 28 : batch_timer_ = std::make_unique<boost::asio::steady_timer>(channel->get_executor());
420 : :
421 : 28 : std::weak_ptr<bool> weak_alive = alive_marker_;
422 : 28 : std::weak_ptr<Impl> weak_impl = weak_from_this();
423 : :
424 : 28 : channel->on_bytes([this, weak_impl, weak_alive](memory::ConstByteSpan data) {
425 : 5 : auto impl_keepalive = weak_impl.lock();
426 [ - + ]: 5 : if (!impl_keepalive) return;
427 : 5 : auto alive = weak_alive.lock();
428 [ - + ]: 5 : if (!alive) return;
429 : :
430 : : // #449: everything below runs synchronously on this io thread - mark
431 : : // it so a blocking send() called from within one of these callbacks
432 : : // fails fast instead of deadlocking.
433 : 5 : detail::CallbackGuard callback_guard;
434 : :
435 : : // #441: snapshot the handler/framer pointers under a shared_lock (not
436 : : // unique_lock) - this is a pure read, matching try_send's locking
437 : : // level so it no longer blocks concurrent sends even briefly.
438 : : bool batch_mode;
439 : 5 : interface::SharedCallback<MessageHandler> handler;
440 : 5 : std::shared_ptr<framer::IFramer> framer_to_push;
441 : : {
442 : 5 : std::shared_lock<std::shared_mutex> lock(mutex_);
443 : 5 : batch_mode = static_cast<bool>(data_batch_handler_);
444 : 5 : handler = data_handler;
445 : 5 : framer_to_push = framer;
446 : 5 : }
447 : :
448 [ + + ]: 5 : if (batch_mode) {
449 : : // #441: build the copy before taking the exclusive lock, so the
450 : : // lock is only held for the queue mutation itself, not the
451 : : // allocation.
452 : 4 : MessageContext ctx(0, memory::SafeDataBuffer(data));
453 : 4 : interface::SharedCallback<BatchMessageHandler> flush_handler;
454 : 4 : std::vector<MessageContext> batch;
455 : : {
456 : 4 : std::unique_lock<std::shared_mutex> lock(mutex_);
457 : 4 : data_batch_queue_.emplace_back(std::move(ctx));
458 [ + + ]: 4 : if (data_batch_queue_.size() >= max_batch_size_) {
459 : 1 : flush_handler = data_batch_handler_;
460 : 1 : batch = std::move(data_batch_queue_);
461 : 1 : data_batch_queue_.clear();
462 [ + - ]: 3 : } else if (data_batch_queue_.size() == 1) {
463 : 3 : schedule_batch_timer();
464 : : }
465 : 4 : }
466 : 4 : detail::invoke_user_callback("serial", "on_data_batch", flush_handler, batch);
467 : 4 : } else {
468 : 2 : detail::invoke_user_callback("serial", "on_data", handler, MessageContext(0, data));
469 : : }
470 : :
471 [ + + + - ]: 5 : if (framer_to_push) framer_to_push->push_bytes(data);
472 : 5 : });
473 : :
474 : 28 : channel->on_backpressure([this, weak_impl, weak_alive](size_t queued) {
475 : 1 : bp_cv_.notify_all();
476 : 1 : auto impl_keepalive = weak_impl.lock();
477 [ - + ]: 1 : if (!impl_keepalive) return;
478 : 1 : auto alive = weak_alive.lock();
479 [ - + ]: 1 : if (!alive) return;
480 : 1 : std::function<void(size_t)> handler;
481 : : {
482 : 1 : std::shared_lock<std::shared_mutex> lock(mutex_);
483 : 1 : handler = bp_handler;
484 : 1 : }
485 : 1 : detail::invoke_user_callback("serial", "on_backpressure", handler, queued);
486 : 1 : });
487 : :
488 : 28 : channel->on_state([this, weak_impl, weak_alive](base::LinkState state) {
489 : 33 : auto impl_keepalive = weak_impl.lock();
490 [ - + ]: 33 : if (!impl_keepalive) return;
491 : 33 : auto alive = weak_alive.lock();
492 [ - + ]: 33 : if (!alive) return;
493 : :
494 [ + + + ]: 33 : switch (state) {
495 : 7 : case base::LinkState::Connected: {
496 : 7 : ConnectionHandler handler;
497 : : {
498 : 7 : std::unique_lock<std::shared_mutex> lock(mutex_);
499 : 7 : fulfill_all_locked(true);
500 : 7 : handler = connect_handler;
501 : 7 : }
502 : 7 : detail::invoke_user_callback("serial", "on_connect", handler, ConnectionContext(0));
503 : 7 : break;
504 : 7 : }
505 : 9 : case base::LinkState::Closed:
506 : : case base::LinkState::Error:
507 : : case base::LinkState::Idle: {
508 : 9 : ConnectionHandler disconnect_handler_snapshot;
509 : 9 : ErrorHandler error_handler_snapshot;
510 : : {
511 : 9 : std::unique_lock<std::shared_mutex> lock(mutex_);
512 : 9 : fulfill_all_locked(false);
513 [ + + ]: 9 : if (state == base::LinkState::Error) {
514 : 8 : error_handler_snapshot = error_handler;
515 : : } else {
516 : 1 : disconnect_handler_snapshot = disconnect_handler;
517 : : }
518 : 9 : }
519 : 9 : detail::invoke_user_callback("serial", "on_disconnect", disconnect_handler_snapshot, ConnectionContext(0));
520 [ + - ]: 9 : detail::invoke_user_callback("serial", "on_error", error_handler_snapshot,
521 [ + - + - : 18 : channel ? detail::build_error_context(*channel, "Connection error")
- - ]
522 : : : ErrorContext(ErrorCode::IoError, "Connection error"));
523 : 9 : break;
524 : 9 : }
525 : 17 : default:
526 : 17 : break;
527 : : }
528 : 33 : });
529 : 28 : }
530 : :
531 : 5 : void attach_framer_callback() {
532 [ - + ]: 5 : if (!framer) return;
533 : 5 : framer->on_message([this](memory::ConstByteSpan msg) {
534 : : // #441: snapshot under a shared_lock (pure read), build the copy
535 : : // before taking the exclusive lock for queue mutation.
536 : : bool batch_mode;
537 : 3 : interface::SharedCallback<MessageHandler> handler;
538 : : {
539 : 3 : std::shared_lock<std::shared_mutex> lock(mutex_);
540 : 3 : batch_mode = static_cast<bool>(message_batch_handler_);
541 : 3 : handler = message_handler;
542 : 3 : }
543 : :
544 [ + - ]: 3 : if (batch_mode) {
545 : 3 : MessageContext ctx(0, memory::SafeDataBuffer(msg));
546 : 3 : interface::SharedCallback<BatchMessageHandler> flush_handler;
547 : 3 : std::vector<MessageContext> batch;
548 : : {
549 : 3 : std::unique_lock<std::shared_mutex> lock(mutex_);
550 : 3 : message_batch_queue_.emplace_back(std::move(ctx));
551 [ + + ]: 3 : if (message_batch_queue_.size() >= max_batch_size_) {
552 : 1 : flush_handler = message_batch_handler_;
553 : 1 : batch = std::move(message_batch_queue_);
554 : 1 : message_batch_queue_.clear();
555 [ + - ]: 2 : } else if (message_batch_queue_.size() == 1) {
556 : 2 : schedule_batch_timer();
557 : : }
558 : 3 : }
559 : 3 : detail::invoke_user_callback("serial", "on_message_batch", flush_handler, batch);
560 : 3 : return;
561 : 3 : }
562 : :
563 : 0 : detail::invoke_user_callback("serial", "on_message", handler, MessageContext(0, msg));
564 : 3 : });
565 : : }
566 : :
567 : 4 : void set_framer(std::unique_ptr<framer::IFramer> f) {
568 : 4 : std::unique_lock<std::shared_mutex> lock(mutex_);
569 : 4 : framer = std::shared_ptr<framer::IFramer>(std::move(f));
570 [ + - + - : 4 : if (framer && (message_handler || message_batch_handler_)) attach_framer_callback();
- + - + -
- ]
571 : 4 : }
572 : :
573 : 2 : void on_message(MessageHandler handler) {
574 : 2 : std::unique_lock<std::shared_mutex> lock(mutex_);
575 : 2 : message_handler = interface::share_callback(std::move(handler));
576 [ + - + - ]: 2 : if (framer) attach_framer_callback();
577 : 2 : }
578 : :
579 : 3 : void on_message_batch(BatchMessageHandler handler) {
580 : 3 : std::unique_lock<std::shared_mutex> lock(mutex_);
581 : 3 : message_batch_handler_ = interface::share_callback(std::move(handler));
582 [ + - + - ]: 3 : if (framer) attach_framer_callback();
583 : 3 : }
584 : :
585 : 19 : config::SerialConfig build_config_locked() const {
586 : 19 : config::SerialConfig config;
587 : 19 : config.device = device;
588 : 19 : config.baud_rate = baud_rate;
589 : 19 : config.char_size = static_cast<unsigned int>(data_bits);
590 : 19 : config.stop_bits = static_cast<unsigned int>(stop_bits);
591 : 19 : std::string p = to_lower(parity);
592 [ + - + + ]: 19 : if (p == "even")
593 : 2 : config.parity = config::SerialConfig::Parity::Even;
594 [ + - - + ]: 17 : else if (p == "odd")
595 : 0 : config.parity = config::SerialConfig::Parity::Odd;
596 : : else
597 : 17 : config.parity = config::SerialConfig::Parity::None;
598 : :
599 : 19 : std::string f = to_lower(flow_control);
600 [ + - - + ]: 19 : if (f == "software")
601 : 0 : config.flow = config::SerialConfig::Flow::Software;
602 [ + - + + ]: 19 : else if (f == "hardware")
603 : 2 : config.flow = config::SerialConfig::Flow::Hardware;
604 : : else
605 : 17 : config.flow = config::SerialConfig::Flow::None;
606 : :
607 : 19 : config.retry_interval_ms = static_cast<unsigned int>(retry_interval.count());
608 : 19 : config.reopen_on_error = reopen_on_error;
609 : 19 : config.read_chunk = read_chunk;
610 : 19 : config.low_latency = low_latency;
611 : 19 : config.rs485 = rs485;
612 : 19 : config.dtr = dtr;
613 : 19 : config.rts = rts;
614 : 19 : config.rx_idle_timeout_ms = static_cast<unsigned>(rx_idle_timeout.count());
615 : 19 : config.backpressure_threshold = backpressure_threshold;
616 : 19 : config.backpressure_strategy = backpressure_strategy;
617 : 19 : config.use_shared_context = shared_context_.load();
618 : 38 : return config;
619 : 19 : }
620 : : };
621 : :
622 : 34 : Serial::Serial(const std::string& d, uint32_t b) : impl_(std::make_shared<Impl>(d, b)) {}
623 : 2 : Serial::Serial(const std::string& d, uint32_t b, std::shared_ptr<boost::asio::io_context> i)
624 : 2 : : impl_(std::make_shared<Impl>(d, b, i)) {}
625 : 15 : Serial::Serial(std::shared_ptr<interface::Channel> ch) : impl_(std::make_shared<Impl>(ch)) {
626 : 15 : impl_->setup_internal_handlers();
627 : 15 : }
628 : 72 : Serial::~Serial() = default;
629 : :
630 : 0 : Serial::Serial(Serial&&) noexcept = default;
631 : 0 : Serial& Serial::operator=(Serial&&) noexcept = default;
632 : :
633 : 26 : std::future<bool> Serial::start() { return impl_->start(); }
634 : 21 : void Serial::stop() { impl_->stop(); }
635 : 7 : bool Serial::send(std::string_view data) { return impl_->send(data); }
636 : 2 : bool Serial::try_send(std::string_view data) { return impl_->try_send(data); }
637 : 3 : bool Serial::send_line(std::string_view line) { return impl_->send_line(line); }
638 : 1 : bool Serial::try_send_line(std::string_view line) { return impl_->try_send_line(line); }
639 : 2 : bool Serial::send_move(std::vector<uint8_t>&& data) { return impl_->send_move(std::move(data)); }
640 : 2 : bool Serial::try_send_move(std::vector<uint8_t>&& data) { return impl_->try_send_move(std::move(data)); }
641 : 2 : bool Serial::send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
642 : 2 : return impl_->send_shared(std::move(data));
643 : : }
644 : 4 : bool Serial::try_send_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
645 : 4 : return impl_->try_send_shared(std::move(data));
646 : : }
647 : 1 : bool Serial::send_blocking(std::string_view data) { return impl_->send_blocking(data); }
648 : 1 : bool Serial::send_line_blocking(std::string_view line) { return impl_->send_line_blocking(line); }
649 : 8 : bool Serial::connected() const {
650 : 8 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
651 [ + + + - : 16 : return impl_->channel && impl_->channel->is_connected();
+ + ]
652 : 8 : }
653 : 2 : RuntimeStats Serial::stats() const { return impl_->stats(); }
654 : 0 : void Serial::reset_stats() { impl_->reset_stats(); }
655 : :
656 : 17 : Serial& Serial::on_data(MessageHandler h) {
657 : 17 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
658 : 17 : impl_->data_handler = interface::share_callback(std::move(h));
659 : 17 : return *this;
660 : 17 : }
661 : 3 : Serial& Serial::on_data_batch(BatchMessageHandler h) {
662 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
663 : 3 : impl_->data_batch_handler_ = interface::share_callback(std::move(h));
664 : 3 : return *this;
665 : 3 : }
666 : 4 : Serial& Serial::on_connect(ConnectionHandler h) {
667 : 4 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
668 : 4 : impl_->connect_handler = std::move(h);
669 : 4 : return *this;
670 : 4 : }
671 : 3 : Serial& Serial::on_disconnect(ConnectionHandler h) {
672 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
673 : 3 : impl_->disconnect_handler = std::move(h);
674 : 3 : return *this;
675 : 3 : }
676 : 23 : Serial& Serial::on_error(ErrorHandler h) {
677 : 23 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
678 : 23 : impl_->error_handler = std::move(h);
679 : 23 : return *this;
680 : 23 : }
681 : :
682 : 2 : Serial& Serial::on_backpressure(std::function<void(size_t)> h) {
683 : 2 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
684 : 2 : impl_->bp_handler = std::move(h);
685 : 2 : return *this;
686 : 2 : }
687 : :
688 : 4 : Serial& Serial::framer(std::unique_ptr<framer::IFramer> f) {
689 : 4 : impl_->set_framer(std::move(f));
690 : 4 : return *this;
691 : : }
692 : 2 : Serial& Serial::on_message(MessageHandler h) {
693 : 2 : impl_->on_message(std::move(h));
694 : 2 : return *this;
695 : : }
696 : 3 : Serial& Serial::on_message_batch(BatchMessageHandler h) {
697 : 3 : impl_->on_message_batch(std::move(h));
698 : 3 : return *this;
699 : : }
700 : :
701 : 2 : Serial& Serial::auto_start(bool m) {
702 : 2 : impl_->auto_start_.store(m);
703 [ + - + - : 2 : if (impl_->auto_start_.load() && !impl_->started_.load()) start();
+ - + - ]
704 : 2 : return *this;
705 : : }
706 : :
707 : 2 : Serial& Serial::shared_context(bool use_shared) {
708 : 2 : impl_->shared_context_.store(use_shared);
709 : 2 : return *this;
710 : : }
711 : :
712 : 1 : Serial& Serial::baud_rate(uint32_t b) {
713 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
714 : 1 : impl_->baud_rate = b;
715 : 1 : return *this;
716 : 1 : }
717 : 9 : Serial& Serial::data_bits(int d) {
718 : 9 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
719 : 9 : impl_->data_bits = d;
720 : 9 : return *this;
721 : 9 : }
722 : 7 : Serial& Serial::stop_bits(int s) {
723 : 7 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
724 : 7 : impl_->stop_bits = s;
725 : 7 : return *this;
726 : 7 : }
727 : 12 : Serial& Serial::parity(const std::string& p) {
728 : 12 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
729 : 12 : impl_->parity = p;
730 : 12 : return *this;
731 : 12 : }
732 : 11 : Serial& Serial::flow_control(const std::string& f) {
733 : 11 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
734 : 11 : impl_->flow_control = f;
735 : 11 : return *this;
736 : 11 : }
737 : 3 : Serial& Serial::read_chunk(size_t bytes) {
738 : 3 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
739 : 3 : impl_->read_chunk = bytes;
740 : 3 : return *this;
741 : 3 : }
742 : :
743 : 1 : Serial& Serial::low_latency(bool enable) {
744 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
745 : 1 : impl_->low_latency = enable;
746 : 1 : return *this;
747 : 1 : }
748 : :
749 : 1 : Serial& Serial::rs485(bool rts_on_send, bool rx_during_tx, unsigned delay_before_ms, unsigned delay_after_ms) {
750 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
751 : 1 : impl_->rs485.enabled = true;
752 : 1 : impl_->rs485.rts_on_send = rts_on_send;
753 : 1 : impl_->rs485.rx_during_tx = rx_during_tx;
754 : 1 : impl_->rs485.delay_rts_before_send_ms = delay_before_ms;
755 : 1 : impl_->rs485.delay_rts_after_send_ms = delay_after_ms;
756 : 1 : return *this;
757 : 1 : }
758 : :
759 : 1 : Serial& Serial::dtr(bool assert_line) {
760 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
761 : 1 : impl_->dtr = assert_line;
762 : 1 : return *this;
763 : 1 : }
764 : :
765 : 1 : Serial& Serial::rts(bool assert_line) {
766 : 1 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
767 : 1 : impl_->rts = assert_line;
768 : 1 : return *this;
769 : 1 : }
770 : :
771 : 2 : Serial& Serial::rx_idle_timeout(std::chrono::milliseconds timeout) {
772 : 2 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
773 : 2 : impl_->rx_idle_timeout = timeout;
774 : 2 : return *this;
775 : 2 : }
776 : :
777 : 7 : Serial& Serial::reopen_on_error(bool enable) {
778 : 7 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
779 : 7 : impl_->reopen_on_error = enable;
780 : 7 : return *this;
781 : 7 : }
782 : 6 : Serial& Serial::retry_interval(std::chrono::milliseconds i) {
783 : 6 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
784 : 6 : impl_->retry_interval = i;
785 [ - + ]: 6 : if (impl_->channel) {
786 : 0 : auto ts = std::dynamic_pointer_cast<transport::Serial>(impl_->channel);
787 [ # # # # ]: 0 : if (ts) ts->set_retry_interval(static_cast<unsigned int>(i.count()));
788 : 0 : }
789 : 6 : return *this;
790 : 6 : }
791 : :
792 : 23 : Serial& Serial::backpressure_threshold(size_t threshold) {
793 : 23 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
794 : 23 : impl_->backpressure_threshold = threshold;
795 : 23 : return *this;
796 : 23 : }
797 : :
798 : 5 : Serial& Serial::backpressure_strategy(base::constants::BackpressureStrategy strategy) {
799 : 5 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
800 : 5 : impl_->backpressure_strategy = strategy;
801 [ + + ]: 5 : if (impl_->channel) {
802 : 2 : auto ts = std::dynamic_pointer_cast<transport::Serial>(impl_->channel);
803 [ - + - - ]: 2 : if (ts) ts->set_backpressure_strategy(strategy);
804 : 2 : }
805 : 5 : return *this;
806 : 5 : }
807 : :
808 : 3 : size_t Serial::backpressure_threshold() const {
809 : 3 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
810 : 6 : return impl_->backpressure_threshold;
811 : 3 : }
812 : :
813 : 3 : base::constants::BackpressureStrategy Serial::backpressure_strategy() const {
814 : 3 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
815 : 6 : return impl_->backpressure_strategy;
816 : 3 : }
817 : :
818 : 6 : config::SerialConfig Serial::build_config() const {
819 : 6 : std::shared_lock<std::shared_mutex> lock(impl_->mutex_);
820 : 12 : return impl_->build_config_locked();
821 : 6 : }
822 : :
823 : 2 : Serial& Serial::manage_external_context(bool m) {
824 : 2 : impl_->manage_external_context.store(m);
825 : 2 : return *this;
826 : : }
827 : :
828 : 2 : Serial& Serial::batch_size(size_t size) {
829 : 2 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
830 : 2 : impl_->max_batch_size_ = size;
831 : 2 : return *this;
832 : 2 : }
833 : :
834 : 2 : Serial& Serial::batch_latency(std::chrono::milliseconds latency) {
835 : 2 : std::unique_lock<std::shared_mutex> lock(impl_->mutex_);
836 : 2 : impl_->max_batch_latency_ = latency;
837 : 2 : return *this;
838 : 2 : }
839 : :
840 : : } // namespace wrapper
841 : : } // namespace wirestead
|
