Branch data Line data Source code
1 : : /*
2 : : * Copyright 2025 Jinwoo Sung
3 : : *
4 : : * Licensed under the Apache License, Version 2.0 (the "License");
5 : : * you may not use this file except in compliance with the License.
6 : : * You may obtain a copy of the License at
7 : : *
8 : : * http://www.apache.org/licenses/LICENSE-2.0
9 : : *
10 : : * Unless required by applicable law or agreed to in writing, software
11 : : * distributed under the License is distributed on an "AS IS" BASIS,
12 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 : : * See the License for the specific language governing permissions and
14 : : * limitations under the License.
15 : : */
16 : :
17 : : #include "wirestead/transport/serial/serial.hpp"
18 : :
19 : : #include <spdlog/fmt/fmt.h>
20 : :
21 : : #include <atomic>
22 : : #include <boost/asio.hpp>
23 : : #include <chrono>
24 : : #include <cstddef>
25 : : #include <deque>
26 : : #include <memory>
27 : : #include <mutex>
28 : : #include <optional>
29 : : #include <stop_token>
30 : : #include <string>
31 : : #include <thread>
32 : : #include <variant>
33 : : #include <vector>
34 : :
35 : : #include "wirestead/base/common.hpp"
36 : : #include "wirestead/base/constants.hpp"
37 : : #include "wirestead/concurrency/io_context_manager.hpp"
38 : : #include "wirestead/concurrency/io_thread_hook.hpp"
39 : : #include "wirestead/concurrency/thread_safe_state.hpp"
40 : : #include "wirestead/diagnostics/error_handler.hpp"
41 : : #include "wirestead/diagnostics/logger.hpp"
42 : : #include "wirestead/diagnostics/runtime_stats_counter.hpp"
43 : : #include "wirestead/interface/iserial_port.hpp"
44 : : #include "wirestead/memory/memory_pool.hpp"
45 : : #include "wirestead/transport/base/bp_state_machine.hpp"
46 : : #include "wirestead/transport/base/bp_utils.hpp"
47 : : #include "wirestead/transport/base/error_info_holder.hpp"
48 : : #include "wirestead/transport/serial/boost_serial_port.hpp"
49 : :
50 : : namespace wirestead {
51 : : namespace transport {
52 : :
53 : : namespace net = boost::asio;
54 : : using base::LinkState;
55 : : using concurrency::AtomicLinkState;
56 : :
57 : : using BufferVariant =
58 : : std::variant<memory::PooledBuffer, std::vector<uint8_t>, std::shared_ptr<const std::vector<uint8_t>>>;
59 : :
60 : : struct Serial::Impl {
61 : : bool started_ = false;
62 : : std::atomic<bool> stopping_{false};
63 : : std::unique_ptr<net::io_context> owned_ioc_;
64 : : net::io_context& ioc_;
65 : : bool owns_ioc_{false};
66 : : bool uses_shared_context_{false};
67 : : net::strand<net::io_context::executor_type> strand_;
68 : : std::unique_ptr<net::executor_work_guard<net::io_context::executor_type>> work_guard_;
69 : : std::jthread ioc_thread_;
70 : :
71 : : std::unique_ptr<interface::SerialPortInterface> port_;
72 : : config::SerialConfig cfg_;
73 : : // #443: per-channel pool instead of the process-wide GlobalMemoryPool
74 : : // singleton - avoids cross-channel contention on the singleton's bucket
75 : : // mutexes. Capacity is much smaller than the old shared default since
76 : : // it's no longer amortized across every channel in the process.
77 : : // Prefill stays 0. This literal was written while MemoryPool discarded
78 : : // initial_pool_size, so 50 allocated nothing; #575 made the parameter real
79 : : // and turned it into ~1 MiB eagerly allocated per channel at construction.
80 : : // The pool fills as buffers are released.
81 : : memory::MemoryPool pool_{0, 200};
82 : : net::steady_timer retry_timer_;
83 : : // Watchdog for cfg_.rx_idle_timeout_ms: armed on connect, pushed back by
84 : : // every read that carries bytes, so it only ever fires on silence.
85 : : net::steady_timer rx_idle_timer_;
86 : :
87 : : std::vector<uint8_t> rx_;
88 : : std::deque<BufferVariant> tx_;
89 : : std::deque<BufferVariant> pending_;
90 : : std::atomic<size_t> pending_bytes_{0};
91 : : // Buffers handed to the in-flight gather write; current_write_views_
92 : : // points into the batch, so neither is touched while a write is in flight.
93 : : std::vector<BufferVariant> current_write_batch_;
94 : : std::vector<net::const_buffer> current_write_views_;
95 : : bool writing_ = false;
96 : : std::atomic<size_t> queued_bytes_{0};
97 : : // Bytes accepted by a plain async_write_* call but not yet routed onto the
98 : : // strand - reserved via try_reserve_limit_bytes() to close the
99 : : // accept-then-drop race (jwsung91/wirestead#517). inflight_bytes_ mutations
100 : : // and the queued_bytes_/pending_bytes_ increments that promote a
101 : : // reservation both go through write_reserve_mtx_ - see bp_utils.hpp.
102 : : std::atomic<size_t> inflight_bytes_{0};
103 : : std::mutex write_reserve_mtx_;
104 : : // Atomic rather than mutex-guarded: read both from the strand and from
105 : : // arbitrary caller threads (async_try_write_* fast-fail prechecks) - a
106 : : // strand-post/dispatch here would only protect the former (#436).
107 : : std::atomic<base::constants::BackpressureStrategy> bp_strategy_{base::constants::BackpressureStrategy::Reliable};
108 : : // Mirrors cfg_.retry_interval_ms but is the one actually read by
109 : : // schedule_retry(); set_retry_interval() writes only this atomic rather
110 : : // than mutating cfg_ directly, so the read/write pair for this specific
111 : : // field doesn't need a mutex (#436).
112 : : std::atomic<unsigned> retry_interval_ms_;
113 : : size_t bp_high_;
114 : : size_t bp_limit_;
115 : : size_t bp_low_;
116 : : std::atomic<bool> backpressure_active_{false};
117 : : diagnostics::RuntimeStatsCounters stats_;
118 : :
119 : : // Guards on_bytes_/on_state_/on_bp_. Setters (called from any user thread)
120 : : // and the strand-confined read sites both take this lock; readers copy
121 : : // the callback under lock then invoke the copy outside the lock (#436).
122 : : mutable std::mutex callback_mtx_;
123 : : // Shared snapshots: the strand copies one out per received chunk, and a
124 : : // std::function copy allocates whenever the target outgrows its small-object
125 : : // buffer. See interface::SharedCallback.
126 : : interface::SharedCallback<OnBytes> on_bytes_;
127 : : interface::SharedCallback<OnState> on_state_;
128 : : interface::SharedCallback<OnBackpressure> on_bp_;
129 : :
130 : : std::atomic<bool> opened_{false};
131 : 49 : AtomicLinkState state_{LinkState::Idle};
132 : :
133 : : ErrorInfoHolder error_info_holder_{"serial"};
134 : :
135 : 25 : void observe_queue() {
136 : 25 : stats_.observe_queue(queued_bytes_.load(std::memory_order_relaxed) +
137 : 50 : pending_bytes_.load(std::memory_order_relaxed));
138 : 25 : }
139 : :
140 : 23 : queue_util::BackpressureFields bp_fields() {
141 : 23 : return queue_util::BackpressureFields{queued_bytes_,
142 : 23 : pending_bytes_,
143 : 23 : backpressure_active_,
144 : 23 : bp_high_,
145 : 23 : bp_low_,
146 : 23 : bp_limit_,
147 : 23 : bp_strategy_.load(std::memory_order_relaxed)};
148 : : }
149 : :
150 : : // Shared decide_enqueue()/route dispatch used by all 4 async_write_* call
151 : : // sites (async_write_copy's pool+fallback paths, async_write_move,
152 : : // async_write_shared) (#434). Unlike every other transport, a rejection
153 : : // caused by the tx_/BestEffort-trim overflow check is treated as a FATAL
154 : : // transport error here rather than reject-and-continue - this is a
155 : : // pre-existing, deliberate Serial-specific behavior the design plan
156 : : // flagged as needing a product decision, not something to silently unify
157 : : // away. A rejection caused by the separate Reliable-pending_-overflow
158 : : // check (queue+pending+added > bp_limit_ while backpressure is already
159 : : // active) still just drops and continues, matching everyone else -
160 : : // distinguished here since decide_enqueue()'s Rejected result doesn't
161 : : // otherwise tell the two apart.
162 : : //
163 : : // Also normalizes an internal inconsistency found while migrating: only
164 : : // async_write_copy's *fallback* path (and async_write_move/_shared)
165 : : // triggered the fatal-error behavior on tx_ overflow before this change -
166 : : // the *pooled*-buffer path in async_write_copy did not, for no documented
167 : : // reason. All 4 call sites now behave identically.
168 : 9 : void route_enqueued_buffer(std::shared_ptr<Serial> self, BufferVariant&& buf, size_t added) {
169 : : const bool reliable_pending_active =
170 [ + + - + ]: 14 : bp_strategy_.load(std::memory_order_relaxed) == base::constants::BackpressureStrategy::Reliable &&
171 : 5 : backpressure_active_.load(std::memory_order_relaxed);
172 : :
173 : 9 : auto f = bp_fields();
174 : 9 : queue_util::DropAccounting dropped;
175 : 9 : auto decision = queue_util::decide_enqueue(f, added, tx_, dropped);
176 [ + + + - ]: 9 : if (dropped.any()) stats_.record_dropped(dropped.messages, dropped.bytes);
177 : :
178 [ - + ]: 9 : if (decision == queue_util::EnqueueDecision::Rejected) {
179 : 0 : WIRESTEAD_LOG_ERROR("serial", "write", "Queue limit exceeded, dropping message");
180 : : // #448: record as dropped so it's reflected in RuntimeStats instead of
181 : : // silently vanishing after being counted as accepted.
182 : 0 : stats_.record_dropped(1, added);
183 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
184 [ # # ]: 0 : if (reliable_pending_active) {
185 : 0 : return;
186 : : }
187 : 0 : report_backpressure(queued_bytes_ + added);
188 : 0 : tx_.clear();
189 : 0 : queued_bytes_ = 0;
190 : 0 : writing_ = false;
191 : 0 : state_.set(LinkState::Error);
192 : 0 : notify_state();
193 : 0 : handle_error(self, "write_queue_overflow", make_error_code(boost::system::errc::no_buffer_space));
194 : 0 : return;
195 : : }
196 [ - + ]: 9 : if (decision == queue_util::EnqueueDecision::Pending) {
197 : 0 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, pending_bytes_, inflight_bytes_, added);
198 : 0 : pending_.emplace_back(std::move(buf));
199 : 0 : observe_queue();
200 : 0 : return;
201 : : }
202 : 9 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, queued_bytes_, inflight_bytes_, added);
203 : 9 : tx_.emplace_back(std::move(buf));
204 : 9 : observe_queue();
205 : 9 : report_backpressure(queued_bytes_);
206 [ + + + - ]: 9 : if (!writing_) do_write(self);
207 : : }
208 : :
209 : 12 : explicit Impl(const config::SerialConfig& cfg, bool use_shared_context)
210 [ + + ]: 12 : : owned_ioc_(use_shared_context ? nullptr : std::make_unique<net::io_context>()),
211 [ + + + - : 12 : ioc_(use_shared_context ? concurrency::IoContextManager::instance().get_context() : *owned_ioc_),
+ - ]
212 : 12 : owns_ioc_(!use_shared_context),
213 : 12 : uses_shared_context_(use_shared_context),
214 : 12 : strand_(ioc_.get_executor()),
215 : 12 : cfg_(cfg),
216 : 12 : retry_timer_(ioc_),
217 : 12 : rx_idle_timer_(ioc_),
218 : 12 : bp_strategy_(cfg.backpressure_strategy),
219 : 12 : retry_interval_ms_(cfg.retry_interval_ms),
220 : 72 : bp_high_(cfg.backpressure_threshold) {
221 : 12 : init();
222 : 12 : port_ = std::make_unique<BoostSerialPort>(ioc_);
223 : 12 : }
224 : :
225 : 37 : Impl(const config::SerialConfig& cfg, std::unique_ptr<interface::SerialPortInterface> port, net::io_context& ioc)
226 : 74 : : ioc_(ioc),
227 : 37 : owns_ioc_(false),
228 : 37 : strand_(ioc.get_executor()),
229 : 37 : port_(std::move(port)),
230 : 37 : cfg_(cfg),
231 : 37 : retry_timer_(ioc),
232 : 37 : rx_idle_timer_(ioc),
233 : 37 : bp_strategy_(cfg.backpressure_strategy),
234 : 37 : retry_interval_ms_(cfg.retry_interval_ms),
235 : 222 : bp_high_(cfg.backpressure_threshold) {
236 : 37 : init();
237 : 37 : }
238 : :
239 : 49 : void init() {
240 : 49 : cfg_.validate_and_clamp();
241 : 49 : bp_high_ = cfg_.backpressure_threshold;
242 : 49 : bp_limit_ = std::min(std::max(bp_high_ * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
243 : : base::constants::MAX_BUFFER_SIZE);
244 [ + - ]: 49 : bp_low_ = bp_high_ > 1 ? bp_high_ / 2 : bp_high_;
245 [ - + ]: 49 : if (bp_low_ == 0) bp_low_ = 1;
246 : 49 : rx_.resize(cfg_.read_chunk);
247 : 49 : }
248 : :
249 : 44 : ~Impl() {
250 : : try {
251 : 44 : stopping_.store(true);
252 [ - + ]: 44 : if (ioc_thread_.joinable()) {
253 [ # # ]: 0 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
254 : 0 : ioc_thread_.detach();
255 : : } else {
256 : 0 : ioc_thread_.request_stop();
257 : 0 : ioc_thread_.join();
258 : : }
259 : : }
260 : 44 : perform_cleanup();
261 : 0 : } catch (...) {
262 : 0 : }
263 : 44 : }
264 : :
265 : 54 : void open_and_configure(std::shared_ptr<Serial> self) {
266 : 54 : boost::system::error_code ec;
267 : 54 : port_->open(cfg_.device, ec);
268 [ + + ]: 54 : if (ec) {
269 : 18 : WIRESTEAD_LOG_ERROR("serial", "open", fmt::format("Failed to open device: {} - {}", cfg_.device, ec.message()));
270 : 18 : handle_error(self, "open", ec);
271 : 23 : return;
272 : : }
273 : :
274 : 36 : port_->set_option(net::serial_port_base::baud_rate(cfg_.baud_rate), ec);
275 [ + + ]: 36 : if (ec) {
276 : 1 : WIRESTEAD_LOG_ERROR("serial", "configure", fmt::format("Failed baud rate: {}", ec.message()));
277 : 1 : handle_error(self, "baud_rate", ec);
278 : 1 : return;
279 : : }
280 : :
281 : 35 : port_->set_option(net::serial_port_base::character_size(cfg_.char_size), ec);
282 [ + + ]: 35 : if (ec) {
283 : 1 : WIRESTEAD_LOG_ERROR("serial", "configure", fmt::format("Failed char size: {}", ec.message()));
284 : 1 : handle_error(self, "char_size", ec);
285 : 1 : return;
286 : : }
287 : :
288 : : using sb = net::serial_port_base::stop_bits;
289 [ - + + - : 34 : port_->set_option(sb(cfg_.stop_bits == 2 ? sb::two : sb::one), ec);
+ - ]
290 [ + + ]: 34 : if (ec) {
291 : 1 : WIRESTEAD_LOG_ERROR("serial", "configure", fmt::format("Failed stop bits: {}", ec.message()));
292 : 1 : handle_error(self, "stop_bits", ec);
293 : 1 : return;
294 : : }
295 : :
296 : : using pa = net::serial_port_base::parity;
297 : 33 : pa::type p = pa::none;
298 [ - + ]: 33 : if (cfg_.parity == config::SerialConfig::Parity::Even)
299 : 0 : p = pa::even;
300 [ + + ]: 33 : else if (cfg_.parity == config::SerialConfig::Parity::Odd)
301 : 2 : p = pa::odd;
302 : 33 : port_->set_option(pa(p), ec);
303 [ + + ]: 33 : if (ec) {
304 : 1 : WIRESTEAD_LOG_ERROR("serial", "configure", fmt::format("Failed parity: {}", ec.message()));
305 : 1 : handle_error(self, "parity", ec);
306 : 1 : return;
307 : : }
308 : :
309 : : using fc = net::serial_port_base::flow_control;
310 : 32 : fc::type f = fc::none;
311 [ - + ]: 32 : if (cfg_.flow == config::SerialConfig::Flow::Software)
312 : 0 : f = fc::software;
313 [ + + ]: 32 : else if (cfg_.flow == config::SerialConfig::Flow::Hardware)
314 : 1 : f = fc::hardware;
315 : 32 : port_->set_option(fc(f), ec);
316 [ + + ]: 32 : if (ec) {
317 : 1 : WIRESTEAD_LOG_ERROR("serial", "configure", fmt::format("Failed flow control: {}", ec.message()));
318 : 1 : handle_error(self, "flow_control", ec);
319 : 1 : return;
320 : : }
321 : :
322 : : // RS-485 before the first read and before anything is written: the very
323 : : // first frame out has to be sent with the direction pin already under the
324 : : // driver's control, or it goes out with the transceiver still in receive.
325 : : //
326 : : // Unlike low_latency, a refusal here is worth a warning rather than a
327 : : // debug line. Asking for RS-485 and silently running in plain UART mode
328 : : // means every write collides on a shared bus, which presents as garbage
329 : : // from the device rather than as a configuration problem.
330 [ + + + + ]: 33 : if (cfg_.rs485.enabled &&
331 [ + - + + ]: 2 : !port_->set_rs485(cfg_.rs485.rts_on_send, cfg_.rs485.rx_during_tx, cfg_.rs485.delay_rts_before_send_ms,
332 : : cfg_.rs485.delay_rts_after_send_ms)) {
333 : 1 : WIRESTEAD_LOG_WARNING("serial", "configure",
334 : : fmt::format("RS-485 mode was requested but {} does not support it; the adapter must switch "
335 : : "direction in hardware or the bus will collide",
336 : : cfg_.device));
337 : : }
338 : :
339 [ + + - + : 31 : if ((cfg_.dtr || cfg_.rts) && !port_->set_modem_lines(cfg_.dtr, cfg_.rts)) {
+ - - + -
+ ]
340 : 0 : WIRESTEAD_LOG_WARNING("serial", "configure", fmt::format("Could not set DTR/RTS on {}", cfg_.device));
341 : : }
342 : :
343 : : // Best effort, after the line settings and before the first read: a driver
344 : : // without a latency timer just says no, and the port is fine either way.
345 [ + + + - : 31 : if (cfg_.low_latency && !port_->set_low_latency()) {
+ + + + ]
346 : 6 : WIRESTEAD_LOG_DEBUG("serial", "configure",
347 : : fmt::format("Low-latency mode unavailable on {}, using the driver default", cfg_.device));
348 : : }
349 : :
350 : 31 : WIRESTEAD_LOG_INFO("serial", "connect", fmt::format("Device opened: {}", cfg_.device));
351 : 31 : start_read(self);
352 : 31 : reset_rx_idle_timer(self);
353 : :
354 : 31 : opened_.store(true);
355 : 31 : state_.set(LinkState::Connected);
356 : 31 : notify_state();
357 : 31 : do_write(self);
358 : : }
359 : :
360 : 41 : void start_read(std::shared_ptr<Serial> self) {
361 : 123 : port_->async_read_some(
362 : 123 : net::buffer(rx_.data(), rx_.size()), net::bind_executor(strand_, [self](auto ec, std::size_t n) {
363 : 40 : auto impl = self->get_impl();
364 [ + + ]: 40 : if (ec) {
365 : 27 : impl->handle_error(self, "read", ec);
366 : 30 : return;
367 : : }
368 [ + - ]: 13 : if (n > 0) {
369 : 13 : impl->stats_.record_received(n);
370 : 13 : impl->reset_rx_idle_timer(self);
371 : : }
372 : 13 : interface::SharedCallback<OnBytes> on_bytes;
373 : : {
374 : 13 : std::lock_guard<std::mutex> lock(impl->callback_mtx_);
375 : 13 : on_bytes = impl->on_bytes_;
376 : 13 : }
377 [ + + ]: 13 : if (on_bytes) {
378 : : try {
379 : 3 : (*on_bytes)(memory::ConstByteSpan(impl->rx_.data(), n));
380 : 5 : } catch (const std::exception& e) {
381 : 2 : std::string msg = fmt::format("Exception in callback: {}", e.what());
382 : 2 : WIRESTEAD_LOG_ERROR("serial", "on_bytes", msg);
383 [ + + ]: 2 : if (impl->cfg_.stop_on_callback_exception) {
384 : 1 : impl->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR,
385 : : diagnostics::ErrorCategory::COMMUNICATION, "on_bytes", {}, msg,
386 : : false, 0);
387 : 1 : impl->opened_.store(false);
388 : 1 : impl->close_port();
389 : 1 : impl->state_.set(LinkState::Error);
390 : 1 : impl->notify_state();
391 : 1 : return;
392 : : }
393 : 1 : impl->handle_error(self, "on_bytes_callback", make_error_code(boost::system::errc::io_error));
394 : 1 : return;
395 : 4 : } catch (...) {
396 [ - + ]: 1 : if (impl->cfg_.stop_on_callback_exception) {
397 : 0 : impl->error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR,
398 : : diagnostics::ErrorCategory::COMMUNICATION, "on_bytes", {},
399 : : "Unknown exception in callback", false, 0);
400 : 0 : impl->opened_.store(false);
401 : 0 : impl->close_port();
402 : 0 : impl->state_.set(LinkState::Error);
403 : 0 : impl->notify_state();
404 : 0 : return;
405 : : }
406 : 1 : impl->handle_error(self, "on_bytes_callback", make_error_code(boost::system::errc::io_error));
407 : 1 : return;
408 : : }
409 : : }
410 : 10 : impl->start_read(self);
411 : 13 : }));
412 : 41 : }
413 : :
414 : 40 : void do_write(std::shared_ptr<Serial> self) {
415 [ + - + + : 40 : if (stopping_.load() || tx_.empty()) {
+ + ]
416 : 32 : writing_ = false;
417 : 32 : return;
418 : : }
419 : 8 : writing_ = true;
420 : :
421 : : // Drain several queued buffers into one scatter-gather write rather than
422 : : // one write syscall per message. `writing_` keeps do_write() from
423 : : // re-entering, so the batch and its views stay put for the whole operation.
424 : 8 : queue_util::take_gather_batch(tx_, current_write_batch_, current_write_views_);
425 : :
426 : 8 : auto on_write = [self](const boost::system::error_code& ec, std::size_t n) {
427 : 8 : auto impl = self->get_impl();
428 : 8 : impl->current_write_batch_.clear();
429 : :
430 [ + + ]: 8 : if (impl->queued_bytes_ >= n) {
431 : 4 : impl->queued_bytes_ -= n;
432 : : } else {
433 : 4 : impl->queued_bytes_ = 0;
434 : : }
435 : 8 : impl->report_backpressure(impl->queued_bytes_);
436 : :
437 [ + + ]: 8 : if (impl->stopping_.load()) {
438 : 4 : impl->writing_ = false;
439 : 4 : return;
440 : : }
441 : :
442 [ + + ]: 4 : if (ec) {
443 : 2 : impl->handle_error(self, "write", ec);
444 : 2 : return;
445 : : }
446 : 2 : impl->stats_.record_sent(n);
447 : 2 : impl->do_write(self);
448 : 8 : };
449 : :
450 : 8 : port_->async_write(current_write_views_, net::bind_executor(strand_, on_write));
451 : 8 : }
452 : :
453 : 113 : void perform_cleanup() {
454 : : try {
455 : 113 : retry_timer_.cancel();
456 : 113 : close_port();
457 : 113 : tx_.clear();
458 : 113 : queued_bytes_ = 0;
459 : 113 : pending_.clear();
460 : 113 : pending_bytes_ = 0;
461 : 113 : writing_ = false;
462 : 113 : report_backpressure(queued_bytes_);
463 : 113 : opened_.store(false);
464 : 113 : state_.set(LinkState::Closed);
465 : 113 : notify_state();
466 : 0 : } catch (...) {
467 : 0 : }
468 : 113 : }
469 : :
470 : 57 : void handle_error(std::shared_ptr<Serial> self, const char* where, const boost::system::error_code& ec) {
471 [ - + ]: 57 : if (ec == boost::asio::error::eof) {
472 [ # # # # ]: 0 : if (self) start_read(self);
473 : 0 : return;
474 : : }
475 : :
476 [ + + ]: 57 : if (stopping_.load()) {
477 : 23 : perform_cleanup();
478 : 23 : return;
479 : : }
480 : :
481 [ + + ]: 34 : if (ec == boost::asio::error::operation_aborted) {
482 [ + + ]: 4 : if (state_.is_state(LinkState::Error)) return;
483 : : // Connecting means a reopen is already scheduled and the port was closed
484 : : // on purpose - the read this aborts is the one that closure cancelled.
485 : : // Cleaning up here would cancel that retry and report Closed instead.
486 [ + - ]: 3 : if (state_.is_state(LinkState::Connecting)) return;
487 : 0 : perform_cleanup();
488 : 0 : return;
489 : : }
490 : :
491 : 30 : bool retryable = cfg_.reopen_on_error;
492 : 30 : diagnostics::error_reporting::report_connection_error("serial", where, ec, retryable);
493 : :
494 : 30 : WIRESTEAD_LOG_ERROR("serial", where, fmt::format("Error: {}", ec.message()));
495 : 30 : error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, where, ec,
496 : 60 : ec.message(), retryable, 0);
497 : :
498 [ + + ]: 30 : if (cfg_.reopen_on_error) {
499 : 16 : opened_.store(false);
500 : 16 : close_port();
501 : 16 : state_.set(LinkState::Connecting);
502 : 16 : notify_state();
503 [ + - + - ]: 16 : if (self) schedule_retry(self, where, ec);
504 : : } else {
505 : 14 : opened_.store(false);
506 : 14 : close_port();
507 : 14 : state_.set(LinkState::Error);
508 : 14 : notify_state();
509 : : }
510 : : }
511 : :
512 : 16 : void schedule_retry(std::shared_ptr<Serial> self, const char* where, const boost::system::error_code& ec) {
513 : : (void)ec;
514 : 16 : WIRESTEAD_LOG_INFO("serial", "retry", fmt::format("Scheduling retry at {}", where));
515 [ - + ]: 16 : if (stopping_.load()) return;
516 : 32 : retry_timer_.expires_after(std::chrono::milliseconds(retry_interval_ms_.load()));
517 : 16 : retry_timer_.async_wait([self](auto e) {
518 [ + + + - : 12 : if (!e && self && !self->get_impl()->stopping_.load()) self->get_impl()->open_and_configure(self);
+ - + + +
- ]
519 : 12 : });
520 : : }
521 : :
522 : : // Rearmed by every read that carried bytes, so the deadline always measures
523 : : // silence rather than time since connect. Writes deliberately do not rearm
524 : : // it: a driver polling a mute device would otherwise keep it alive forever.
525 : 44 : void reset_rx_idle_timer(std::shared_ptr<Serial> self) {
526 : 44 : const unsigned timeout_ms = cfg_.rx_idle_timeout_ms;
527 [ + + - + : 44 : if (timeout_ms == 0 || stopping_.load()) return;
+ + ]
528 : :
529 : 15 : rx_idle_timer_.expires_after(std::chrono::milliseconds(timeout_ms));
530 : 15 : rx_idle_timer_.async_wait(net::bind_executor(strand_, [self, timeout_ms](const boost::system::error_code& e) {
531 [ + + ]: 15 : if (e) return; // rearmed or cancelled
532 : 3 : auto impl = self->get_impl();
533 [ + - - + : 3 : if (impl->stopping_.load() || !impl->opened_.load()) return;
- + ]
534 : :
535 : 3 : WIRESTEAD_LOG_WARNING("serial", "rx_idle_timeout",
536 : : fmt::format("No data received for {}ms on {}", timeout_ms, impl->cfg_.device));
537 : : // Same path a read error takes, so reopen_on_error decides what happens
538 : : // next and the reconnect/backoff plumbing is not duplicated here.
539 : 3 : impl->handle_error(self, "rx_idle_timeout", make_error_code(boost::asio::error::timed_out));
540 : : }));
541 : : }
542 : :
543 : 144 : void close_port() {
544 : 144 : rx_idle_timer_.cancel();
545 : 144 : boost::system::error_code ec;
546 [ + - + - : 144 : if (port_ && port_->is_open()) {
+ + + + ]
547 : 36 : port_->close(ec);
548 : : }
549 : 144 : }
550 : :
551 : 220 : void notify_state() {
552 [ + + ]: 243 : if (stopping_.load()) return;
553 : 107 : interface::SharedCallback<OnState> on_state;
554 : : {
555 : 107 : std::lock_guard<std::mutex> lock(callback_mtx_);
556 : 107 : on_state = on_state_;
557 : 107 : }
558 [ + + ]: 107 : if (!on_state) return;
559 : : try {
560 : 84 : (*on_state)(state_.get());
561 : 0 : } catch (...) {
562 : 0 : }
563 : 107 : }
564 : :
565 : : // Unlike every other migrated transport, this deliberately keeps taking no
566 : : // `self` parameter and passes an empty kick_write hook to the shared
567 : : // state machine below: perform_cleanup() (reachable from ~Impl(), where
568 : : // shared_from_this() cannot be used at all) calls this with no self
569 : : // available, and the corresponding do_write() kick after a backpressure
570 : : // OFF transition is still issued manually at every async_write_* call
571 : : // site exactly as before (#434 - normalizing this into the shared hook
572 : : // was judged not worth the self-availability hazard in the cleanup path).
573 : 131 : void report_backpressure(size_t qb) {
574 [ + + ]: 131 : if (stopping_.load()) return;
575 : 14 : observe_queue();
576 : :
577 : 14 : interface::SharedCallback<OnBackpressure> on_bp;
578 : : {
579 : 14 : std::lock_guard<std::mutex> lock(callback_mtx_);
580 : 14 : on_bp = on_bp_;
581 : 14 : }
582 : 14 : static const OnBackpressure kNoCallback;
583 : :
584 : 14 : auto f = bp_fields();
585 [ + + + - ]: 26 : queue_util::report_backpressure(
586 : 12 : f, qb, on_bp ? *on_bp : kNoCallback, stats_,
587 : 0 : [&]() -> size_t {
588 : 1 : const size_t moved = pending_bytes_.exchange(0);
589 [ - + ]: 1 : while (!pending_.empty()) {
590 : 0 : tx_.emplace_back(std::move(pending_.front()));
591 : 0 : pending_.pop_front();
592 : : }
593 : 1 : return moved;
594 : : },
595 : 15 : [&]() { observe_queue(); });
596 : 14 : }
597 : : };
598 : :
599 : 12 : std::shared_ptr<Serial> Serial::create(const config::SerialConfig& cfg, bool use_shared_context) {
600 : 12 : return std::shared_ptr<Serial>(new Serial(cfg, use_shared_context));
601 : : }
602 : :
603 : 2 : std::shared_ptr<Serial> Serial::create(const config::SerialConfig& cfg, net::io_context& ioc) {
604 : 2 : return std::shared_ptr<Serial>(new Serial(cfg, std::make_unique<BoostSerialPort>(ioc), ioc));
605 : : }
606 : :
607 : 35 : std::shared_ptr<Serial> Serial::create(const config::SerialConfig& cfg,
608 : : std::unique_ptr<interface::SerialPortInterface> port, net::io_context& ioc) {
609 : 35 : return std::shared_ptr<Serial>(new Serial(cfg, std::move(port), ioc));
610 : : }
611 : :
612 : 12 : Serial::Serial(const config::SerialConfig& cfg, bool use_shared_context)
613 : 12 : : impl_(std::make_unique<Impl>(cfg, use_shared_context)) {}
614 : :
615 : 37 : Serial::Serial(const config::SerialConfig& cfg, std::unique_ptr<interface::SerialPortInterface> port,
616 : 37 : net::io_context& ioc)
617 : 37 : : impl_(std::make_unique<Impl>(cfg, std::move(port), ioc)) {}
618 : :
619 : 88 : Serial::~Serial() {
620 [ + - - + : 44 : if (impl_ && impl_->started_ && !impl_->state_.is_state(LinkState::Closed)) {
- - - + ]
621 : : // In destructor, stop without shared_from_this
622 : 0 : impl_->stopping_.store(true);
623 : 0 : impl_->perform_cleanup();
624 [ # # # # : 0 : if (impl_->owns_ioc_ && impl_->ioc_thread_.joinable()) {
# # ]
625 : 0 : impl_->ioc_thread_.join();
626 : : }
627 : : }
628 : 88 : }
629 : :
630 : 0 : Serial::Serial(Serial&&) noexcept = default;
631 : 0 : Serial& Serial::operator=(Serial&&) noexcept = default;
632 : :
633 : 47 : void Serial::start() {
634 : 47 : auto impl = get_impl();
635 [ - + ]: 47 : if (impl->started_) return;
636 : 47 : impl->stopping_.store(false);
637 : 47 : WIRESTEAD_LOG_INFO("serial", "start", fmt::format("Starting device: {}", impl->cfg_.device));
638 [ + + ]: 47 : if (impl->uses_shared_context_) {
639 : 2 : auto& manager = concurrency::IoContextManager::instance();
640 [ + - - + : 2 : if (!manager.is_running()) manager.start();
- - ]
641 [ + - - + : 2 : if (impl->ioc_.stopped()) impl->ioc_.restart();
- - ]
642 : : }
643 : : impl->work_guard_ =
644 : 47 : std::make_unique<net::executor_work_guard<net::io_context::executor_type>>(impl->ioc_.get_executor());
645 [ + + ]: 47 : if (impl->owns_ioc_) {
646 : 20 : impl->ioc_thread_ = std::jthread([impl](std::stop_token st) {
647 : 10 : wirestead::concurrency::run_io_thread_init();
648 : : try {
649 : 10 : std::stop_callback cb(st, [impl] { impl->ioc_.stop(); });
650 : 10 : impl->ioc_.run();
651 : 10 : } catch (...) {
652 : 0 : }
653 : 20 : });
654 : : }
655 : 47 : auto self = shared_from_this();
656 : 47 : net::post(impl->strand_, [self] {
657 : 46 : auto impl = self->get_impl();
658 [ + + ]: 46 : if (!impl->stopping_.load()) {
659 : 45 : impl->state_.set(LinkState::Connecting);
660 : 45 : impl->notify_state();
661 : 45 : impl->open_and_configure(self);
662 : : }
663 : 46 : });
664 : 47 : impl->started_ = true;
665 : 47 : }
666 : :
667 : 48 : void Serial::stop() {
668 : 48 : auto impl = get_impl();
669 [ + + ]: 48 : if (!impl->started_) {
670 : 1 : impl->state_.set(LinkState::Closed);
671 : 1 : return;
672 : : }
673 : :
674 [ - + ]: 47 : if (impl->stopping_.exchange(true)) return;
675 : :
676 : 47 : auto self = shared_from_this();
677 : 47 : net::post(impl->strand_, [self] {
678 : 46 : auto impl = self->get_impl();
679 : 46 : impl->perform_cleanup();
680 [ + + ]: 46 : if (impl->owns_ioc_) impl->ioc_.stop();
681 : 46 : });
682 : :
683 [ + + + - : 47 : if (impl->owns_ioc_ && impl->ioc_thread_.joinable()) {
+ + ]
684 : 10 : impl->ioc_thread_.join();
685 : 10 : impl->ioc_.restart();
686 : : }
687 : 47 : impl->started_ = false;
688 : 47 : }
689 : :
690 : 17 : bool Serial::is_connected() const { return get_impl()->opened_.load(); }
691 : 7 : bool Serial::is_backpressure_active() const { return get_impl()->backpressure_active_.load(); }
692 : 8 : wrapper::RuntimeStats Serial::stats() const {
693 : 8 : auto impl = get_impl();
694 : 24 : return impl->stats_.snapshot(impl->queued_bytes_.load(std::memory_order_relaxed),
695 : : impl->pending_bytes_.load(std::memory_order_relaxed),
696 : 16 : impl->backpressure_active_.load(std::memory_order_relaxed));
697 : : }
698 : 1 : void Serial::reset_stats() {
699 : 1 : auto impl = get_impl();
700 : 1 : impl->stats_.reset(impl->queued_bytes_.load(std::memory_order_relaxed) +
701 : 2 : impl->pending_bytes_.load(std::memory_order_relaxed));
702 : 1 : }
703 : :
704 : 19 : std::optional<diagnostics::ErrorInfo> Serial::last_error_info() const {
705 : 19 : return get_impl()->error_info_holder_.last_error_info();
706 : : }
707 : :
708 : 15 : boost::asio::any_io_executor Serial::get_executor() { return impl_->strand_; }
709 : :
710 : 3 : bool Serial::async_write_copy(memory::ConstByteSpan data) {
711 : 3 : auto impl = get_impl();
712 [ + - + - : 3 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
- + - + ]
713 : 0 : impl->stats_.record_failed_send();
714 : 0 : return false;
715 : : }
716 : :
717 : 3 : size_t n = data.size();
718 [ - + ]: 3 : if (n == 0) {
719 : 0 : impl->stats_.record_failed_send();
720 : 0 : return false;
721 : : }
722 [ - + ]: 3 : if (n > base::constants::MAX_BUFFER_SIZE) {
723 : 0 : WIRESTEAD_LOG_ERROR("serial", "write", "Write size exceeds maximum");
724 : 0 : impl->stats_.record_failed_send();
725 : 0 : return false;
726 : : }
727 : :
728 [ + + + - ]: 3 : if (n <= 65536 && impl->cfg_.enable_memory_pool) {
729 : 2 : memory::PooledBuffer pooled(n, impl->pool_);
730 [ + - + - ]: 2 : if (pooled.valid()) {
731 : 2 : base::safe_memory::safe_memcpy(pooled.data(), data.data(), n);
732 [ - + ]: 2 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queued_bytes_, impl->pending_bytes_,
733 : 2 : impl->inflight_bytes_, n, impl->bp_limit_)) {
734 : 0 : impl->stats_.record_failed_send();
735 : 0 : return false;
736 : : }
737 : 2 : impl->stats_.record_accepted(n);
738 : 2 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(pooled)]() mutable {
739 : 2 : auto impl = self->get_impl();
740 : 2 : const auto added = buf.size();
741 : 2 : impl->route_enqueued_buffer(self, BufferVariant{std::move(buf)}, added);
742 : 2 : });
743 : 2 : return true;
744 : : }
745 : 2 : }
746 : :
747 [ + - ]: 1 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queued_bytes_, impl->pending_bytes_,
748 : 1 : impl->inflight_bytes_, n, impl->bp_limit_)) {
749 : 1 : impl->stats_.record_failed_send();
750 : 1 : return false;
751 : : }
752 : 0 : std::vector<uint8_t> fallback(data.begin(), data.end());
753 : 0 : impl->stats_.record_accepted(n);
754 : 0 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(fallback)]() mutable {
755 : 0 : auto impl = self->get_impl();
756 : 0 : const auto added = buf.size();
757 : 0 : impl->route_enqueued_buffer(self, BufferVariant{std::move(buf)}, added);
758 : 0 : });
759 : 0 : return true;
760 : 0 : }
761 : :
762 : 7 : bool Serial::async_write_move(std::vector<uint8_t>&& data) {
763 : 7 : auto impl = get_impl();
764 [ + - + - : 7 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
- + - + ]
765 : 0 : impl->stats_.record_failed_send();
766 : 0 : return false;
767 : : }
768 : 7 : const auto added = data.size();
769 [ - + ]: 7 : if (added == 0) {
770 : 0 : impl->stats_.record_failed_send();
771 : 0 : return false;
772 : : }
773 [ + + ]: 7 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queued_bytes_, impl->pending_bytes_,
774 : 7 : impl->inflight_bytes_, added, impl->bp_limit_)) {
775 : 1 : impl->stats_.record_failed_send();
776 : 1 : return false;
777 : : }
778 : 6 : impl->stats_.record_accepted(added);
779 : 6 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(data), added]() mutable {
780 : 6 : auto impl = self->get_impl();
781 : 6 : impl->route_enqueued_buffer(self, BufferVariant{std::move(buf)}, added);
782 : 6 : });
783 : 6 : return true;
784 : : }
785 : :
786 : 2 : bool Serial::async_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
787 : 2 : auto impl = get_impl();
788 [ + - + - : 2 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
- + - + ]
789 : 0 : impl->stats_.record_failed_send();
790 : 0 : return false;
791 : : }
792 [ + - - + : 2 : if (!data || data->empty()) {
- + ]
793 : 0 : impl->stats_.record_failed_send();
794 : 0 : return false;
795 : : }
796 : 2 : const auto added = data->size();
797 [ + + ]: 2 : if (!queue_util::try_reserve_limit_bytes(impl->write_reserve_mtx_, impl->queued_bytes_, impl->pending_bytes_,
798 : 2 : impl->inflight_bytes_, added, impl->bp_limit_)) {
799 : 1 : impl->stats_.record_failed_send();
800 : 1 : return false;
801 : : }
802 : 1 : impl->stats_.record_accepted(added);
803 : 1 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(data), added]() mutable {
804 : 1 : auto impl = self->get_impl();
805 : 1 : impl->route_enqueued_buffer(self, BufferVariant{std::move(buf)}, added);
806 : 1 : });
807 : 1 : return true;
808 : : }
809 : :
810 : 1 : bool Serial::async_try_write_copy(memory::ConstByteSpan data) {
811 [ + - - + : 1 : if (data.empty() || data.size() > base::constants::MAX_BUFFER_SIZE) {
- + ]
812 : 0 : get_impl()->stats_.record_failed_send();
813 : 0 : return false;
814 : : }
815 : 3 : return async_try_write_move(std::vector<uint8_t>(data.begin(), data.end()));
816 : : }
817 : :
818 : 4 : bool Serial::async_try_write_move(std::vector<uint8_t>&& data) {
819 : 4 : auto impl = get_impl();
820 [ + - + - : 4 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
- + - + ]
821 : 0 : impl->stats_.record_failed_send();
822 : 0 : return false;
823 : : }
824 : 4 : const auto added = data.size();
825 [ + - - + ]: 4 : if (added == 0 || added > base::constants::MAX_BUFFER_SIZE) {
826 : 0 : impl->stats_.record_failed_send();
827 : 0 : return false;
828 : : }
829 : 3 : const auto reject_for_pressure = [impl, added]() {
830 [ + + ]: 3 : if (impl->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
831 : 1 : impl->stats_.record_dropped(1, added);
832 : : } else {
833 : 2 : impl->stats_.record_failed_send();
834 : : }
835 : 7 : };
836 [ + + + - : 5 : if (impl->backpressure_active_.load() || impl->queued_bytes_ + added > impl->bp_high_ ||
+ + ]
837 [ - + ]: 1 : impl->queued_bytes_ + impl->pending_bytes_ + added > impl->bp_limit_) {
838 : 3 : reject_for_pressure();
839 : 3 : return false;
840 : : }
841 [ - + ]: 1 : if (!queue_util::try_reserve_write_bytes(impl->queued_bytes_, impl->pending_bytes_, impl->backpressure_active_, added,
842 : : impl->bp_high_, impl->bp_limit_)) {
843 : 0 : reject_for_pressure();
844 : 0 : return false;
845 : : }
846 : 1 : impl->stats_.record_accepted(added);
847 : :
848 : 1 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(data), added]() mutable {
849 : 1 : auto impl = self->get_impl();
850 [ + - + - : 1 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
- + - + ]
851 : 0 : queue_util::release_reserved_write_bytes(impl->queued_bytes_, added);
852 : 0 : impl->stats_.record_failed_send();
853 : 0 : return;
854 : : }
855 : :
856 : 1 : impl->tx_.emplace_back(std::move(buf));
857 : 1 : impl->observe_queue();
858 : 1 : impl->report_backpressure(impl->queued_bytes_);
859 [ - + - - ]: 1 : if (!impl->writing_) impl->do_write(self);
860 : : });
861 : 1 : return true;
862 : : }
863 : :
864 : 1 : bool Serial::async_try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
865 : 1 : auto impl = get_impl();
866 [ + - + - ]: 2 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error) ||
867 [ + - + - : 2 : !data || data->empty()) {
- + - + ]
868 : 0 : impl->stats_.record_failed_send();
869 : 0 : return false;
870 : : }
871 : 1 : const auto added = data->size();
872 [ - + ]: 1 : if (added > base::constants::MAX_BUFFER_SIZE) {
873 : 0 : impl->stats_.record_failed_send();
874 : 0 : return false;
875 : : }
876 : 1 : const auto reject_for_pressure = [impl, added]() {
877 [ - + ]: 1 : if (impl->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
878 : 0 : impl->stats_.record_dropped(1, added);
879 : : } else {
880 : 1 : impl->stats_.record_failed_send();
881 : : }
882 : 2 : };
883 [ - + - - : 1 : if (impl->backpressure_active_.load() || impl->queued_bytes_ + added > impl->bp_high_ ||
+ - ]
884 [ # # ]: 0 : impl->queued_bytes_ + impl->pending_bytes_ + added > impl->bp_limit_) {
885 : 1 : reject_for_pressure();
886 : 1 : return false;
887 : : }
888 [ # # ]: 0 : if (!queue_util::try_reserve_write_bytes(impl->queued_bytes_, impl->pending_bytes_, impl->backpressure_active_, added,
889 : : impl->bp_high_, impl->bp_limit_)) {
890 : 0 : reject_for_pressure();
891 : 0 : return false;
892 : : }
893 : 0 : impl->stats_.record_accepted(added);
894 : :
895 : 0 : net::post(impl->strand_, [self = shared_from_this(), buf = std::move(data), added]() mutable {
896 : 0 : auto impl = self->get_impl();
897 [ # # # # : 0 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
# # # # ]
898 : 0 : queue_util::release_reserved_write_bytes(impl->queued_bytes_, added);
899 : 0 : impl->stats_.record_failed_send();
900 : 0 : return;
901 : : }
902 : :
903 : 0 : impl->tx_.emplace_back(std::move(buf));
904 : 0 : impl->observe_queue();
905 : 0 : impl->report_backpressure(impl->queued_bytes_);
906 [ # # # # ]: 0 : if (!impl->writing_) impl->do_write(self);
907 : : });
908 : 0 : return true;
909 : : }
910 : :
911 : 34 : void Serial::on_bytes(OnBytes cb) {
912 : 34 : auto shared = interface::share_callback(std::move(cb));
913 : 34 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
914 : 34 : impl_->on_bytes_ = std::move(shared);
915 : 34 : }
916 : 51 : void Serial::on_state(OnState cb) {
917 : 51 : auto shared = interface::share_callback(std::move(cb));
918 : 51 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
919 : 51 : impl_->on_state_ = std::move(shared);
920 : 51 : }
921 : 37 : void Serial::on_backpressure(OnBackpressure cb) {
922 : 37 : auto shared = interface::share_callback(std::move(cb));
923 : 37 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
924 : 37 : impl_->on_bp_ = std::move(shared);
925 : 37 : }
926 : :
927 : 1 : void Serial::set_backpressure_strategy(base::constants::BackpressureStrategy strategy) {
928 : 1 : get_impl()->bp_strategy_.store(strategy, std::memory_order_relaxed);
929 : 1 : }
930 : :
931 : 0 : void Serial::set_retry_interval(unsigned interval_ms) {
932 [ # # ]: 0 : if (interval_ms < base::constants::MIN_RETRY_INTERVAL_MS) {
933 : 0 : interval_ms = base::constants::MIN_RETRY_INTERVAL_MS;
934 [ # # ]: 0 : } else if (interval_ms > base::constants::MAX_RETRY_INTERVAL_MS) {
935 : 0 : interval_ms = base::constants::MAX_RETRY_INTERVAL_MS;
936 : : }
937 : 0 : get_impl()->retry_interval_ms_.store(interval_ms, std::memory_order_relaxed);
938 : 0 : }
939 : :
940 : : } // namespace transport
941 : : } // namespace wirestead
|
