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 <condition_variable>
25 : : #include <cstddef>
26 : : #include <deque>
27 : : #include <memory>
28 : : #include <mutex>
29 : : #include <optional>
30 : : #include <stop_token>
31 : : #include <string>
32 : : #include <string_view>
33 : : #include <thread>
34 : : #include <variant>
35 : : #include <vector>
36 : :
37 : : #include "wirestead/base/common.hpp"
38 : : #include "wirestead/base/constants.hpp"
39 : : #include "wirestead/concurrency/io_context_manager.hpp"
40 : : #include "wirestead/concurrency/io_thread_hook.hpp"
41 : : #include "wirestead/concurrency/thread_safe_state.hpp"
42 : : #include "wirestead/config/validation.hpp"
43 : : #include "wirestead/diagnostics/callback.hpp"
44 : : #include "wirestead/diagnostics/error_handler.hpp"
45 : : #include "wirestead/diagnostics/logger.hpp"
46 : : #include "wirestead/diagnostics/runtime_stats_counter.hpp"
47 : : #include "wirestead/diagnostics/send_accounting.hpp"
48 : : #include "wirestead/interface/iserial_port.hpp"
49 : : #include "wirestead/memory/memory_pool.hpp"
50 : : #include "wirestead/transport/base/bp_state_machine.hpp"
51 : : #include "wirestead/transport/base/bp_utils.hpp"
52 : : #include "wirestead/transport/base/error_info_holder.hpp"
53 : : #include "wirestead/transport/base/stop_test_hook.hpp"
54 : : #include "wirestead/transport/serial/boost_serial_port.hpp"
55 : : #include "wirestead/transport/serial/detail/write_wait.hpp"
56 : :
57 : : namespace wirestead {
58 : : namespace transport {
59 : :
60 : : namespace net = boost::asio;
61 : : using base::LinkState;
62 : : using concurrency::AtomicLinkState;
63 : :
64 : : using BufferVariant =
65 : : std::variant<memory::PooledBuffer, std::vector<uint8_t>, std::shared_ptr<const std::vector<uint8_t>>>;
66 : :
67 : : struct Serial::Impl {
68 : : std::atomic<bool> started_{false};
69 : : std::atomic<bool> stopping_{false};
70 : : std::unique_ptr<net::io_context> owned_ioc_;
71 : : net::io_context& ioc_;
72 : : bool owns_ioc_{false};
73 : : bool uses_shared_context_{false};
74 : : net::strand<net::io_context::executor_type> strand_;
75 : : std::unique_ptr<net::executor_work_guard<net::io_context::executor_type>> work_guard_;
76 : : std::jthread ioc_thread_;
77 : :
78 : : std::unique_ptr<interface::SerialPortInterface> port_;
79 : : config::SerialConfig cfg_;
80 : : // #443: per-channel pool instead of the process-wide GlobalMemoryPool
81 : : // singleton - avoids cross-channel contention on the singleton's bucket
82 : : // mutexes. Capacity is much smaller than the old shared default since
83 : : // it's no longer amortized across every channel in the process.
84 : : // Prefill stays 0. This literal was written while MemoryPool discarded
85 : : // initial_pool_size, so 50 allocated nothing; #575 made the parameter real
86 : : // and turned it into ~1 MiB eagerly allocated per channel at construction.
87 : : // The pool fills as buffers are released.
88 : : memory::MemoryPool pool_{0, 200};
89 : : net::steady_timer retry_timer_;
90 : : // Watchdog for cfg_.rx_idle_timeout_ms: armed on connect, pushed back by
91 : : // every read that carries bytes, so it only ever fires on silence.
92 : : net::steady_timer rx_idle_timer_;
93 : :
94 : : std::shared_ptr<std::vector<uint8_t>> rx_;
95 : : using Ledger = diagnostics::SendAccountingLedger;
96 : : struct TrackedBuffer {
97 : : BufferVariant buffer;
98 : : Ledger::Request request;
99 : : };
100 : : struct BufferProjection {
101 : 650 : const BufferVariant& operator()(const TrackedBuffer& item) const { return item.buffer; }
102 : : };
103 : : Ledger send_accounting_;
104 : : std::deque<TrackedBuffer> tx_;
105 : : std::deque<TrackedBuffer> pending_;
106 : : std::atomic<size_t> pending_bytes_{0};
107 : : // Completions retain their own buffers across device reopen.
108 : : struct WriteBatch {
109 : : std::vector<TrackedBuffer> buffers;
110 : : std::vector<net::const_buffer> views;
111 : : size_t bytes = 0;
112 : : };
113 : : std::shared_ptr<WriteBatch> active_write_;
114 : : bool writing_ = false;
115 : : std::atomic<size_t> queued_bytes_{0};
116 : : // Bytes accepted by a plain async_write_* call but not yet routed onto the
117 : : // strand - reserved via try_reserve_limit_bytes() to close the
118 : : // accept-then-drop race (jwsung91/wirestead#517). inflight_bytes_ mutations
119 : : // and the queued_bytes_/pending_bytes_ increments that promote a
120 : : // reservation both go through write_reserve_mtx_ - see bp_utils.hpp.
121 : : std::atomic<size_t> inflight_bytes_{0};
122 : : std::mutex write_reserve_mtx_;
123 : : std::mutex stop_mtx_;
124 : : std::condition_variable stop_cv_;
125 : : bool cleanup_done_ = false;
126 : : bool cleanup_started_ = false;
127 : : bool cleanup_finished_ = false;
128 : : size_t pending_io_ = 0; // Strand-confined, including handler destruction.
129 : : std::mutex join_mtx_;
130 : : std::mutex submission_mtx_;
131 : : std::atomic<uint64_t> generation_{0};
132 : : std::atomic<uint64_t> connection_seq_{0};
133 : : std::shared_ptr<detail::SerialWriteWait> write_wait_;
134 : :
135 : : // Caller holds submission_mtx_: state, connection and admission are one decision.
136 : 776 : std::optional<wrapper::SendRejection> admission_rejection(
137 : : std::optional<uint64_t> expected_connection = std::nullopt) {
138 [ + + ]: 776 : if (stopping_.load()) {
139 : 62 : std::lock_guard<std::mutex> lock(stop_mtx_);
140 [ + + ]: 62 : return cleanup_done_ ? wrapper::SendRejection::NotStarted : wrapper::SendRejection::Stopping;
141 : 62 : }
142 [ + + ]: 1428 : if (generation_.load() == 0) return wrapper::SendRejection::NotStarted;
143 [ + + + - : 2105 : if ((expected_connection && *expected_connection != connection_seq_.load()) || state_.is_state(LinkState::Closed) ||
+ - ]
144 [ + + + + : 2058 : state_.is_state(LinkState::Error) || !opened_.load()) {
+ + ]
145 : 51 : return wrapper::SendRejection::NotReady;
146 : : }
147 : 637 : return std::nullopt;
148 : : }
149 : :
150 : 346 : void mark_cleanup_done() {
151 : : // A discarded fake completion can also release the last tracked operation.
152 : : // Only now is it safe to release the gather-write storage.
153 : 346 : active_write_.reset();
154 : 346 : detail::stop_test_hook(this, true);
155 : 346 : work_guard_.reset();
156 : 346 : started_ = false;
157 : : {
158 : 346 : std::lock_guard<std::mutex> lock(stop_mtx_);
159 : 346 : cleanup_done_ = true;
160 : 346 : }
161 : 346 : stop_cv_.notify_all();
162 : 346 : }
163 : :
164 : : // The serial interface erases executor associations into std::function.
165 : : // Explicit dispatch preserves serialization. The lifetime also accounts for
166 : : // test ports that discard an operation instead of invoking its handler.
167 : : template <typename Handler>
168 : 763 : auto track_io(std::shared_ptr<Serial> self, Handler handler, bool defer = false) {
169 : 763 : ++pending_io_;
170 : 2289 : std::shared_ptr<void> lifetime(nullptr, [self](void*) {
171 : 2289 : net::dispatch(self->impl_->strand_, [self] {
172 : 763 : auto* impl = self->impl_.get();
173 [ + + + + : 763 : if (impl->cleanup_finished_ && impl->pending_io_ == 1) {
+ + + - +
+ + + + +
+ + ]
174 [ - + - + : 187 : if (auto hook = detail::g_serial_io_completion_hook.load()) hook();
- + + + ]
175 : : }
176 [ + + + - : 763 : if (--impl->pending_io_ == 0 && impl->cleanup_finished_) impl->mark_cleanup_done();
+ + + + +
- + + + +
+ + + + +
+ + + +
+ ]
177 : : });
178 : : });
179 : 1511 : return [self, lifetime, handler = std::move(handler), defer](auto... args) {
180 : 1496 : auto completion = [lifetime, handler, args...]() mutable { handler(args...); };
181 : : // Write initiation holds submission_mtx_; injected interfaces may invoke
182 : : // the callback inline. Queue it before acquiring that mutex again.
183 [ - + - + : 748 : if (defer)
+ - - + ]
184 : 274 : net::post(self->impl_->strand_, std::move(completion));
185 : : else
186 : 474 : net::dispatch(self->impl_->strand_, std::move(completion));
187 : 2274 : };
188 : 763 : }
189 : :
190 : : // Atomic rather than mutex-guarded: read both from the strand and from
191 : : // arbitrary caller threads (async_try_write_* fast-fail prechecks) - a
192 : : // strand-post/dispatch here would only protect the former (#436).
193 : : std::atomic<base::constants::BackpressureStrategy> bp_strategy_{base::constants::BackpressureStrategy::Reliable};
194 : : // Mirrors cfg_.retry_interval_ms but is the one actually read by
195 : : // schedule_retry(); set_retry_interval() writes only this atomic rather
196 : : // than mutating cfg_ directly, so the read/write pair for this specific
197 : : // field doesn't need a mutex (#436).
198 : : std::atomic<unsigned> retry_interval_ms_;
199 : : size_t bp_high_;
200 : : size_t bp_limit_;
201 : : size_t bp_low_;
202 : : std::atomic<bool> backpressure_active_{false};
203 : : diagnostics::RuntimeStatsCounters stats_;
204 : :
205 : : // Guards on_bytes_/on_state_/on_bp_. Setters (called from any user thread)
206 : : // and the strand-confined read sites both take this lock; readers copy
207 : : // the callback under lock then invoke the copy outside the lock (#436).
208 : : mutable std::mutex callback_mtx_;
209 : : // Shared snapshots: the strand copies one out per received chunk, and a
210 : : // std::function copy allocates whenever the target outgrows its small-object
211 : : // buffer. See interface::SharedCallback.
212 : : interface::SharedCallback<OnBytes> on_bytes_;
213 : : interface::SharedCallback<OnState> on_state_;
214 : : interface::SharedCallback<OnBackpressure> on_bp_;
215 : :
216 : : std::atomic<bool> opened_{false};
217 : 329 : AtomicLinkState state_{LinkState::Idle};
218 : :
219 : : ErrorInfoHolder error_info_holder_{"serial"};
220 : :
221 : 939 : void observe_queue() {
222 : 939 : stats_.observe_queue(queued_bytes_.load(std::memory_order_relaxed) +
223 : 1878 : pending_bytes_.load(std::memory_order_relaxed));
224 : 939 : }
225 : :
226 : 776 : queue_util::BackpressureFields bp_fields() {
227 : 776 : return queue_util::BackpressureFields{queued_bytes_,
228 : 776 : pending_bytes_,
229 : 776 : backpressure_active_,
230 : 776 : bp_high_,
231 : 776 : bp_low_,
232 : 776 : bp_limit_,
233 : 776 : bp_strategy_.load(std::memory_order_relaxed),
234 : 776 : &write_reserve_mtx_};
235 : : }
236 : :
237 : : // Shared decide_enqueue()/route dispatch used by all 4 async_write_* call
238 : : // sites (async_write_copy's pool+fallback paths, async_write_move,
239 : : // async_write_shared) (#434). Unlike every other transport, a rejection
240 : : // caused by the tx_/BestEffort-trim overflow check is treated as a FATAL
241 : : // transport error here rather than reject-and-continue - this is a
242 : : // pre-existing, deliberate Serial-specific behavior the design plan
243 : : // flagged as needing a product decision, not something to silently unify
244 : : // away. A rejection caused by the separate Reliable-pending_-overflow
245 : : // check (queue+pending+added > bp_limit_ while backpressure is already
246 : : // active) still just drops and continues, matching everyone else -
247 : : // distinguished here since decide_enqueue()'s Rejected result doesn't
248 : : // otherwise tell the two apart.
249 : : //
250 : : // Also normalizes an internal inconsistency found while migrating: only
251 : : // async_write_copy's *fallback* path (and async_write_move/_shared)
252 : : // triggered the fatal-error behavior on tx_ overflow before this change -
253 : : // the *pooled*-buffer path in async_write_copy did not, for no documented
254 : : // reason. All 4 call sites now behave identically.
255 : 298 : void route_enqueued_buffer(std::shared_ptr<Serial> self, TrackedBuffer&& buf, size_t added) {
256 [ + + ]: 298 : if (stopping_) {
257 : 22 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
258 : 45 : return;
259 : : }
260 : : const bool reliable_pending_active =
261 [ + + + + ]: 498 : bp_strategy_.load(std::memory_order_relaxed) == base::constants::BackpressureStrategy::Reliable &&
262 : 222 : backpressure_active_.load(std::memory_order_relaxed);
263 : :
264 : 276 : auto f = bp_fields();
265 : 276 : queue_util::DropAccounting dropped;
266 : 276 : auto decision = queue_util::decide_enqueue(
267 : 276 : f, added, tx_, dropped, BufferProjection{},
268 : : [this](const TrackedBuffer& item) { send_accounting_.discard(item.request, Ledger::Cause::QueuePressure); });
269 [ - + - - ]: 276 : if (dropped.any()) stats_.record_dropped(dropped.messages, dropped.bytes);
270 : :
271 [ - + ]: 276 : if (decision == queue_util::EnqueueDecision::Rejected) {
272 : 0 : send_accounting_.discard(buf.request, Ledger::Cause::QueuePressure);
273 : 0 : WIRESTEAD_LOG_ERROR("serial", "write", "Queue limit exceeded, dropping message");
274 : : // #448: record as dropped so it's reflected in RuntimeStats instead of
275 : : // silently vanishing after being counted as accepted.
276 : 0 : stats_.record_dropped(1, added);
277 : 0 : queue_util::release_reserved_limit_bytes(write_reserve_mtx_, inflight_bytes_, added);
278 [ # # ]: 0 : if (reliable_pending_active) {
279 : 0 : return;
280 : : }
281 : 0 : report_backpressure(queued_bytes_ + added);
282 : 0 : mark_disconnected();
283 : 0 : discard_connection_writes();
284 : 0 : state_.set(LinkState::Error);
285 : 0 : notify_state();
286 : 0 : handle_error(self, "write_queue_overflow", make_error_code(boost::system::errc::no_buffer_space));
287 : 0 : return;
288 : : }
289 [ + + ]: 276 : if (decision == queue_util::EnqueueDecision::Pending) {
290 : 23 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, pending_bytes_, inflight_bytes_, added);
291 : 23 : pending_.emplace_back(std::move(buf));
292 : 23 : observe_queue();
293 : 23 : return;
294 : : }
295 : 253 : queue_util::commit_reserved_limit_bytes(write_reserve_mtx_, queued_bytes_, inflight_bytes_, added);
296 : 253 : tx_.emplace_back(std::move(buf));
297 : 253 : observe_queue();
298 : 253 : report_backpressure(queued_bytes_);
299 [ + + + - ]: 253 : if (!writing_) do_write(self);
300 : : }
301 : :
302 : 26 : explicit Impl(const config::SerialConfig& cfg, bool use_shared_context)
303 [ + + ]: 26 : : owned_ioc_(use_shared_context ? nullptr : std::make_unique<net::io_context>()),
304 [ + + + - : 26 : ioc_(use_shared_context ? concurrency::IoContextManager::instance().get_context() : *owned_ioc_),
+ - ]
305 : 26 : owns_ioc_(!use_shared_context),
306 : 26 : uses_shared_context_(use_shared_context),
307 : 26 : strand_(ioc_.get_executor()),
308 : 26 : cfg_(cfg),
309 : 26 : retry_timer_(ioc_),
310 : 26 : rx_idle_timer_(ioc_),
311 : 26 : bp_strategy_(cfg.backpressure_strategy),
312 : 26 : retry_interval_ms_(cfg.retry_interval_ms),
313 : 156 : bp_high_(cfg.backpressure_threshold) {
314 : 26 : init();
315 : 25 : port_ = std::make_unique<BoostSerialPort>(ioc_);
316 : 45 : }
317 : :
318 : 303 : Impl(const config::SerialConfig& cfg, std::unique_ptr<interface::SerialPortInterface> port, net::io_context& ioc)
319 : 606 : : ioc_(ioc),
320 : 303 : owns_ioc_(false),
321 : 303 : strand_(ioc.get_executor()),
322 : 303 : port_(std::move(port)),
323 : 303 : cfg_(cfg),
324 : 303 : retry_timer_(ioc),
325 : 303 : rx_idle_timer_(ioc),
326 : 303 : bp_strategy_(cfg.backpressure_strategy),
327 : 303 : retry_interval_ms_(cfg.retry_interval_ms),
328 : 1818 : bp_high_(cfg.backpressure_threshold) {
329 : 303 : init();
330 : 322 : }
331 : :
332 : 809 : void mark_disconnected() {
333 : 809 : std::lock_guard<std::mutex> lock(submission_mtx_);
334 : 809 : mark_disconnected_locked();
335 : 809 : }
336 : 838 : void mark_disconnected_locked() {
337 [ + + ]: 838 : if (opened_.exchange(false)) {
338 : 361 : send_accounting_.end(Ledger::Cause::ConnectionLoss);
339 [ + - ]: 361 : if (write_wait_) write_wait_->end(wrapper::SendRejection::NotReady);
340 : 361 : connection_seq_.fetch_add(1);
341 : : }
342 : 838 : }
343 : :
344 : 802 : void discard_connection_writes() {
345 : 802 : size_t messages = tx_.size() + pending_.size();
346 : 802 : size_t bytes = 0;
347 [ + + ]: 938 : for (const auto& buffer : tx_)
348 : 272 : bytes += std::visit([](const auto& b) { return queue_util::variant_buffer_size(b); }, buffer.buffer);
349 [ + + ]: 823 : for (const auto& buffer : pending_)
350 : 42 : bytes += std::visit([](const auto& b) { return queue_util::variant_buffer_size(b); }, buffer.buffer);
351 [ + + ]: 802 : if (active_write_) {
352 : 159 : messages += active_write_->buffers.size();
353 : 159 : bytes += active_write_->bytes;
354 : : // Its completion handler retains the buffers; it must not mutate new state.
355 : 159 : active_write_.reset();
356 : : }
357 : 802 : tx_.clear();
358 : 802 : pending_.clear();
359 : 802 : queued_bytes_ = 0;
360 : 802 : pending_bytes_ = 0;
361 : 802 : writing_ = false;
362 : 802 : backpressure_active_ = false;
363 [ + + ]: 802 : if (messages) stats_.record_dropped(messages, bytes);
364 : 802 : }
365 : :
366 : 329 : void init() {
367 : 329 : config::detail::validate(cfg_);
368 : 327 : bp_high_ = cfg_.backpressure_threshold;
369 : 327 : bp_limit_ = std::min(std::max(bp_high_ * 4, base::constants::DEFAULT_BACKPRESSURE_THRESHOLD),
370 : : base::constants::MAX_BUFFER_SIZE);
371 [ + - ]: 327 : bp_low_ = bp_high_ > 1 ? bp_high_ / 2 : bp_high_;
372 [ - + ]: 327 : if (bp_low_ == 0) bp_low_ = 1;
373 : 327 : rx_ = std::make_shared<std::vector<uint8_t>>(cfg_.read_chunk);
374 : 327 : }
375 : :
376 : 327 : ~Impl() {
377 : : try {
378 : 327 : stopping_.store(true);
379 [ - + ]: 327 : if (ioc_thread_.joinable()) {
380 [ # # ]: 0 : if (std::this_thread::get_id() == ioc_thread_.get_id()) {
381 : 0 : ioc_thread_.detach();
382 : : } else {
383 : 0 : ioc_thread_.request_stop();
384 : 0 : ioc_thread_.join();
385 : : }
386 : : }
387 : 327 : perform_cleanup();
388 : 0 : } catch (...) {
389 : 0 : }
390 : 327 : }
391 : :
392 : 383 : void open_and_configure(std::shared_ptr<Serial> self) {
393 [ - + ]: 405 : if (stopping_) return;
394 : 383 : boost::system::error_code ec;
395 : 383 : port_->open(cfg_.device, ec);
396 [ + + ]: 383 : if (ec) {
397 : 16 : WIRESTEAD_LOG_ERROR("serial", "open", fmt::format("Failed to open device: {} - {}", cfg_.device, ec.message()));
398 : 16 : handle_error(self, "open", ec);
399 : 16 : return;
400 : : }
401 : :
402 : 367 : port_->set_option(net::serial_port_base::baud_rate(cfg_.baud_rate), ec);
403 [ + + ]: 367 : if (ec) {
404 : 1 : WIRESTEAD_LOG_ERROR("serial", "configure", fmt::format("Failed baud rate: {}", ec.message()));
405 : 1 : handle_error(self, "baud_rate", ec);
406 : 1 : return;
407 : : }
408 : :
409 : 366 : port_->set_option(net::serial_port_base::character_size(cfg_.char_size), ec);
410 [ + + ]: 366 : if (ec) {
411 : 1 : WIRESTEAD_LOG_ERROR("serial", "configure", fmt::format("Failed char size: {}", ec.message()));
412 : 1 : handle_error(self, "char_size", ec);
413 : 1 : return;
414 : : }
415 : :
416 : : using sb = net::serial_port_base::stop_bits;
417 [ - + + - : 365 : port_->set_option(sb(cfg_.stop_bits == 2 ? sb::two : sb::one), ec);
+ - ]
418 [ + + ]: 365 : if (ec) {
419 : 1 : WIRESTEAD_LOG_ERROR("serial", "configure", fmt::format("Failed stop bits: {}", ec.message()));
420 : 1 : handle_error(self, "stop_bits", ec);
421 : 1 : return;
422 : : }
423 : :
424 : : using pa = net::serial_port_base::parity;
425 : 364 : pa::type p = pa::none;
426 [ - + ]: 364 : if (cfg_.parity == config::SerialConfig::Parity::Even)
427 : 0 : p = pa::even;
428 [ + + ]: 364 : else if (cfg_.parity == config::SerialConfig::Parity::Odd)
429 : 2 : p = pa::odd;
430 : 364 : port_->set_option(pa(p), ec);
431 [ + + ]: 364 : if (ec) {
432 : 1 : WIRESTEAD_LOG_ERROR("serial", "configure", fmt::format("Failed parity: {}", ec.message()));
433 : 1 : handle_error(self, "parity", ec);
434 : 1 : return;
435 : : }
436 : :
437 : : using fc = net::serial_port_base::flow_control;
438 : 363 : fc::type f = fc::none;
439 [ - + ]: 363 : if (cfg_.flow == config::SerialConfig::Flow::Software)
440 : 0 : f = fc::software;
441 [ + + ]: 363 : else if (cfg_.flow == config::SerialConfig::Flow::Hardware)
442 : 1 : f = fc::hardware;
443 : 363 : port_->set_option(fc(f), ec);
444 [ + + ]: 363 : if (ec) {
445 : 1 : WIRESTEAD_LOG_ERROR("serial", "configure", fmt::format("Failed flow control: {}", ec.message()));
446 : 1 : handle_error(self, "flow_control", ec);
447 : 1 : return;
448 : : }
449 : :
450 : : // RS-485 before the first read and before anything is written: the very
451 : : // first frame out has to be sent with the direction pin already under the
452 : : // driver's control, or it goes out with the transceiver still in receive.
453 : : //
454 : : // Unlike low_latency, a refusal here is worth a warning rather than a
455 : : // debug line. Asking for RS-485 and silently running in plain UART mode
456 : : // means every write collides on a shared bus, which presents as garbage
457 : : // from the device rather than as a configuration problem.
458 [ + + + + ]: 364 : if (cfg_.rs485.enabled &&
459 [ + - + + ]: 2 : !port_->set_rs485(cfg_.rs485.rts_on_send, cfg_.rs485.rx_during_tx, cfg_.rs485.delay_rts_before_send_ms,
460 : : cfg_.rs485.delay_rts_after_send_ms)) {
461 : 1 : WIRESTEAD_LOG_WARNING("serial", "configure",
462 : : fmt::format("RS-485 mode was requested but {} does not support it; the adapter must switch "
463 : : "direction in hardware or the bus will collide",
464 : : cfg_.device));
465 : : }
466 : :
467 [ + + - + : 362 : if ((cfg_.dtr || cfg_.rts) && !port_->set_modem_lines(cfg_.dtr, cfg_.rts)) {
+ - - + -
+ ]
468 : 0 : WIRESTEAD_LOG_WARNING("serial", "configure", fmt::format("Could not set DTR/RTS on {}", cfg_.device));
469 : : }
470 : :
471 [ + + + - : 362 : if (cfg_.keep_dtr_on_close && !port_->keep_dtr_on_close()) {
+ - + + ]
472 : 1 : handle_error(self, "keep_dtr_on_close", make_error_code(boost::system::errc::operation_not_supported));
473 : 1 : return;
474 : : }
475 : :
476 : : // Best effort, after the line settings and before the first read: a driver
477 : : // without a latency timer just says no, and the port is fine either way.
478 [ + + + - : 361 : if (cfg_.low_latency && !port_->set_low_latency()) {
+ + + + ]
479 : 219 : WIRESTEAD_LOG_DEBUG("serial", "configure",
480 : : fmt::format("Low-latency mode unavailable on {}, using the driver default", cfg_.device));
481 : : }
482 : :
483 : 361 : WIRESTEAD_LOG_INFO("serial", "connect", fmt::format("Device opened: {}", cfg_.device));
484 : : {
485 : 361 : std::lock_guard<std::mutex> lock(submission_mtx_);
486 [ - + ]: 361 : if (stopping_) return;
487 : 361 : const auto connection = connection_seq_.fetch_add(1) + 1;
488 : 361 : write_wait_ = std::make_shared<detail::SerialWriteWait>(connection);
489 : 361 : opened_.store(true);
490 : 361 : state_.set(LinkState::Connected);
491 : 361 : }
492 : 361 : rx_ = std::make_shared<std::vector<uint8_t>>(cfg_.read_chunk);
493 : 361 : start_read(self);
494 : 361 : reset_rx_idle_timer(self);
495 : 361 : notify_state();
496 : 361 : do_write(self);
497 : : }
498 : :
499 : 411 : void start_read(std::shared_ptr<Serial> self) {
500 [ + + + + : 411 : if (stopping_ || !opened_) return;
+ + ]
501 : 788 : const auto connection = connection_seq_.load();
502 : 394 : auto buffer = rx_;
503 : 1182 : port_->async_read_some(
504 : 1182 : net::buffer(*buffer), track_io(self, [self, connection, buffer](auto ec, std::size_t n) {
505 : 394 : auto impl = self->get_impl();
506 [ + + + + : 467 : if (impl->stopping_ || connection != impl->connection_seq_) return;
+ + ]
507 [ + + ]: 116 : if (ec) {
508 : 72 : impl->handle_error(self, "read", ec);
509 : 72 : return;
510 : : }
511 [ + - ]: 44 : if (n > 0) {
512 : 44 : impl->stats_.record_received(n);
513 : 44 : impl->reset_rx_idle_timer(self);
514 : : }
515 : 44 : interface::SharedCallback<OnBytes> on_bytes;
516 : : {
517 : 44 : std::lock_guard<std::mutex> lock(impl->callback_mtx_);
518 : 44 : on_bytes = impl->on_bytes_;
519 : 44 : }
520 [ + + + + ]: 49 : if (!diagnostics::invoke_callback("serial", "on_bytes", on_bytes, memory::ConstByteSpan(buffer->data(), n)) &&
521 : 5 : impl->cfg_.stop_on_callback_exception) {
522 : 1 : self->stop(); // Explicit opt-in stops quietly, without recursive error notification.
523 : 1 : return;
524 : : }
525 : 43 : impl->start_read(self);
526 : 44 : }));
527 : 394 : }
528 : :
529 : 727 : void do_write(std::shared_ptr<Serial> self) {
530 [ + + + - : 727 : if (stopping_.load() || !opened_.load() || tx_.empty() || writing_) return;
+ + - + +
+ ]
531 : 289 : std::unique_lock<std::mutex> submission_lock(submission_mtx_);
532 [ + - - + : 289 : if (stopping_.load() || !opened_.load()) return;
- + ]
533 : 289 : writing_ = true;
534 : 289 : auto batch = std::make_shared<WriteBatch>();
535 : 289 : batch->bytes = queue_util::take_gather_batch(tx_, batch->buffers, batch->views, BufferProjection{});
536 : : // Admission still fences stop/loss; amortize the ledger lock over the batch.
537 : 939 : send_accounting_.begin_batch(batch->buffers, [](const auto& item) { return item.request; });
538 : 289 : active_write_ = batch;
539 : 289 : const auto connection = connection_seq_.load();
540 : : auto completion = track_io(
541 : : self,
542 : 289 : [self, connection, batch](boost::system::error_code ec, size_t written) {
543 : 274 : auto* impl = self->get_impl();
544 : : bool current_connection;
545 : : {
546 : 274 : std::lock_guard<std::mutex> lock(impl->submission_mtx_);
547 : 274 : current_connection = connection == impl->connection_seq_.load();
548 : 548 : impl->send_accounting_.complete_batch(
549 : 584 : batch->buffers, written, [](const auto& item) { return item.request; },
550 : 310 : [](const auto& item) {
551 : 620 : return std::visit([](const auto& b) { return queue_util::variant_buffer_size(b); }, item.buffer);
552 : : });
553 : : // A short composed write without an error is still a terminal failure.
554 [ + + + + : 274 : if (!ec && written < batch->bytes) ec = net::error::connection_reset;
+ + ]
555 [ + + + + : 274 : if (current_connection && ec) impl->mark_disconnected_locked();
+ + + - ]
556 : 274 : }
557 [ + + ]: 274 : if (!current_connection) {
558 : 144 : batch->buffers.clear();
559 : 144 : return;
560 : : }
561 : 130 : impl->active_write_.reset();
562 : 130 : impl->writing_ = false;
563 [ - + ]: 130 : if (impl->stopping_.load()) {
564 : 0 : batch->buffers.clear();
565 : 0 : return;
566 : : }
567 [ + + ]: 130 : if (ec) {
568 : 29 : queue_util::return_gather_batch(impl->tx_, batch->buffers);
569 : 29 : impl->handle_error(self, "write", ec);
570 : 29 : return;
571 : : }
572 : 101 : batch->buffers.clear();
573 : 101 : queue_util::release_reserved_write_bytes(impl->queued_bytes_, batch->bytes);
574 : 101 : impl->stats_.record_sent(written);
575 : 101 : impl->report_backpressure(impl->queued_bytes_);
576 : 101 : impl->do_write(self);
577 : : },
578 : 289 : true);
579 : : // Initiation may write inline; never hold admission across the syscall.
580 : 289 : submission_lock.unlock();
581 : : try {
582 : 296 : port_->async_write(batch->views, std::move(completion));
583 : 7 : } catch (...) {
584 : 7 : mark_disconnected();
585 : 7 : const auto ec = make_error_code(boost::system::errc::no_buffer_space);
586 : 7 : handle_error(self, "write", ec);
587 : 7 : }
588 : 289 : }
589 : :
590 : 673 : void perform_cleanup() {
591 : : try {
592 : 673 : retry_timer_.cancel();
593 : 673 : mark_disconnected();
594 : 673 : discard_connection_writes();
595 : 673 : close_port();
596 : 673 : tx_.clear();
597 : 673 : queued_bytes_ = 0;
598 : 673 : pending_.clear();
599 : 673 : pending_bytes_ = 0;
600 : 673 : writing_ = false;
601 : 673 : report_backpressure(queued_bytes_);
602 : 673 : opened_.store(false);
603 : 673 : state_.set(LinkState::Closed);
604 : 673 : notify_state();
605 : 0 : } catch (...) {
606 : 0 : }
607 : 673 : }
608 : :
609 : 346 : void perform_stop_cleanup() {
610 [ - + ]: 346 : if (cleanup_started_) return;
611 : 346 : cleanup_started_ = true;
612 : 346 : perform_cleanup();
613 : 346 : backpressure_active_ = false;
614 : 346 : cleanup_finished_ = true;
615 [ + + ]: 346 : if (pending_io_ == 0) mark_cleanup_done();
616 : : }
617 : :
618 : 137 : void handle_error(std::shared_ptr<Serial> self, const char* where, const boost::system::error_code& ec) {
619 [ + + ]: 137 : if (stopping_) return;
620 [ + + + + : 136 : if (ec == boost::asio::error::eof && std::string_view(where) == "read") {
+ + ]
621 [ + - + - ]: 7 : if (self) start_read(self);
622 : 7 : return;
623 : : }
624 : :
625 [ - + ]: 129 : if (ec == boost::asio::error::operation_aborted) {
626 [ # # ]: 0 : if (state_.is_state(LinkState::Error)) return;
627 : : // Connecting means a reopen is already scheduled and the port was closed
628 : : // on purpose - the read this aborts is the one that closure cancelled.
629 : : // Cleaning up here would cancel that retry and report Closed instead.
630 [ # # ]: 0 : if (state_.is_state(LinkState::Connecting)) return;
631 : 0 : perform_cleanup();
632 : 0 : return;
633 : : }
634 : :
635 [ + + + + ]: 129 : const bool retryable = cfg_.reopen_on_error && std::string_view(where) != "keep_dtr_on_close";
636 : 129 : diagnostics::error_reporting::report_connection_error("serial", where, ec, retryable);
637 : :
638 : 129 : WIRESTEAD_LOG_ERROR("serial", where, fmt::format("Error: {}", ec.message()));
639 : 129 : error_info_holder_.record_error(diagnostics::ErrorLevel::ERROR, diagnostics::ErrorCategory::CONNECTION, where, ec,
640 : 258 : ec.message(), retryable, 0);
641 : :
642 [ + + ]: 129 : if (retryable) {
643 : 65 : mark_disconnected();
644 : 65 : discard_connection_writes();
645 : 65 : close_port();
646 : 65 : state_.set(LinkState::Connecting);
647 : 65 : notify_state();
648 [ + - + - ]: 65 : if (self) schedule_retry(self, where, ec);
649 : : } else {
650 : 64 : mark_disconnected();
651 : 64 : discard_connection_writes();
652 : 64 : close_port();
653 : 64 : state_.set(LinkState::Error);
654 : 64 : notify_state();
655 : : }
656 : : }
657 : :
658 : 65 : void schedule_retry(std::shared_ptr<Serial> self, const char* where, const boost::system::error_code& ec) {
659 : : (void)ec;
660 : 65 : WIRESTEAD_LOG_INFO("serial", "retry", fmt::format("Scheduling retry at {}", where));
661 [ - + ]: 65 : if (stopping_.load()) return;
662 : 130 : retry_timer_.expires_after(std::chrono::milliseconds(retry_interval_ms_.load()));
663 : 65 : retry_timer_.async_wait(track_io(self, [self](auto e) {
664 [ + + + - : 65 : if (!e && !self->get_impl()->stopping_.load()) self->get_impl()->open_and_configure(self);
+ + + - ]
665 : 65 : }));
666 : : }
667 : :
668 : : // Rearmed by every read that carried bytes, so the deadline always measures
669 : : // silence rather than time since connect. Writes deliberately do not rearm
670 : : // it: a driver polling a mute device would otherwise keep it alive forever.
671 : 405 : void reset_rx_idle_timer(std::shared_ptr<Serial> self) {
672 : 405 : const unsigned timeout_ms = cfg_.rx_idle_timeout_ms;
673 [ + + - + : 405 : if (timeout_ms == 0 || stopping_.load()) return;
+ + ]
674 : :
675 : 15 : const auto connection = connection_seq_.load();
676 : 15 : rx_idle_timer_.expires_after(std::chrono::milliseconds(timeout_ms));
677 : 15 : rx_idle_timer_.async_wait(track_io(self, [self, timeout_ms, connection](const boost::system::error_code& e) {
678 [ + + ]: 15 : if (e) return; // rearmed or cancelled
679 : 2 : auto impl = self->get_impl();
680 [ + - + - : 2 : if (impl->stopping_.load() || !impl->opened_.load() || connection != impl->connection_seq_) return;
- + - + ]
681 : :
682 : 2 : WIRESTEAD_LOG_WARNING("serial", "rx_idle_timeout",
683 : : fmt::format("No data received for {}ms on {}", timeout_ms, impl->cfg_.device));
684 : : // Same path a read error takes, so reopen_on_error decides what happens
685 : : // next and the reconnect/backoff plumbing is not duplicated here.
686 : 2 : impl->handle_error(self, "rx_idle_timeout", make_error_code(boost::asio::error::timed_out));
687 : : }));
688 : : }
689 : :
690 : 802 : void close_port() {
691 : 802 : rx_idle_timer_.cancel();
692 : 802 : boost::system::error_code ec;
693 [ + - + - : 802 : if (port_ && port_->is_open()) {
+ + + + ]
694 : 367 : port_->close(ec);
695 : : }
696 : 802 : }
697 : :
698 : 1493 : void notify_state() {
699 [ + + ]: 1919 : if (stopping_.load()) return;
700 : 820 : interface::SharedCallback<OnState> on_state;
701 : : {
702 : 820 : std::lock_guard<std::mutex> lock(callback_mtx_);
703 : 820 : on_state = on_state_;
704 : 820 : }
705 [ + + ]: 820 : if (!on_state) return;
706 : 394 : diagnostics::invoke_callback("serial", "on_state", on_state, state_.get());
707 : 820 : }
708 : :
709 : : // Unlike every other migrated transport, this deliberately keeps taking no
710 : : // `self` parameter and passes an empty kick_write hook to the shared
711 : : // state machine below: perform_cleanup() (reachable from ~Impl(), where
712 : : // shared_from_this() cannot be used at all) calls this with no self
713 : : // available, and the corresponding do_write() kick after a backpressure
714 : : // OFF transition is still issued manually at every async_write_* call
715 : : // site exactly as before (#434 - normalizing this into the shared hook
716 : : // was judged not worth the self-availability hazard in the cleanup path).
717 : 1176 : void report_backpressure(size_t qb) {
718 [ + + ]: 1176 : if (stopping_.load()) return;
719 : 500 : observe_queue();
720 : :
721 : 500 : interface::SharedCallback<OnBackpressure> on_bp;
722 : : {
723 : 500 : std::lock_guard<std::mutex> lock(callback_mtx_);
724 : 500 : on_bp = on_bp_;
725 : 500 : }
726 : 500 : static const OnBackpressure kNoCallback;
727 : :
728 : 500 : auto f = bp_fields();
729 [ + + + - ]: 607 : queue_util::report_backpressure(
730 : 107 : f, qb, on_bp ? *on_bp : kNoCallback, stats_,
731 : 0 : [&]() -> size_t {
732 : 14 : const size_t moved = pending_bytes_.exchange(0);
733 [ + + ]: 16 : while (!pending_.empty()) {
734 : 2 : tx_.emplace_back(std::move(pending_.front()));
735 : 2 : pending_.pop_front();
736 : : }
737 : 14 : return moved;
738 : : },
739 : 514 : [&]() { observe_queue(); });
740 : 500 : }
741 : : };
742 : :
743 : 26 : std::shared_ptr<Serial> Serial::create(const config::SerialConfig& cfg, bool use_shared_context) {
744 : 26 : return std::shared_ptr<Serial>(new Serial(cfg, use_shared_context));
745 : : }
746 : :
747 : 15 : std::shared_ptr<Serial> Serial::create(const config::SerialConfig& cfg, net::io_context& ioc) {
748 : 15 : return std::shared_ptr<Serial>(new Serial(cfg, std::make_unique<BoostSerialPort>(ioc), ioc));
749 : : }
750 : :
751 : 288 : std::shared_ptr<Serial> Serial::create(const config::SerialConfig& cfg,
752 : : std::unique_ptr<interface::SerialPortInterface> port, net::io_context& ioc) {
753 : 289 : return std::shared_ptr<Serial>(new Serial(cfg, std::move(port), ioc));
754 : : }
755 : :
756 : 26 : Serial::Serial(const config::SerialConfig& cfg, bool use_shared_context)
757 : 27 : : impl_(std::make_unique<Impl>(cfg, use_shared_context)) {}
758 : :
759 : 303 : Serial::Serial(const config::SerialConfig& cfg, std::unique_ptr<interface::SerialPortInterface> port,
760 : 303 : net::io_context& ioc)
761 : 304 : : impl_(std::make_unique<Impl>(cfg, std::move(port), ioc)) {}
762 : :
763 : 654 : Serial::~Serial() {
764 [ + - ]: 327 : if (impl_) stop();
765 : 654 : }
766 : :
767 : 0 : Serial::Serial(Serial&&) noexcept = default;
768 : 0 : Serial& Serial::operator=(Serial&&) noexcept = default;
769 : :
770 : 331 : void Serial::start() {
771 : 331 : auto impl = get_impl();
772 [ - + ]: 331 : if (impl->started_) return;
773 [ - + ]: 331 : if (impl->ioc_thread_.joinable()) impl->ioc_thread_.join();
774 [ + + + + : 331 : if (impl->owns_ioc_ && impl->ioc_.stopped()) impl->ioc_.restart();
+ + ]
775 [ + + ]: 331 : if (impl->uses_shared_context_) {
776 : 4 : auto& manager = concurrency::IoContextManager::instance();
777 [ + + ]: 4 : if (!manager.is_running()) manager.start();
778 : : }
779 : 331 : const auto generation = impl->generation_.fetch_add(1) + 1;
780 : : {
781 : 331 : std::lock_guard<std::mutex> lock(impl->stop_mtx_);
782 : 331 : impl->cleanup_done_ = false;
783 : 331 : }
784 : 331 : impl->cleanup_started_ = false;
785 : 331 : impl->cleanup_finished_ = false;
786 : 331 : impl->stopping_ = false;
787 : 331 : impl->opened_ = false;
788 : 331 : impl->backpressure_active_ = false;
789 : 331 : impl->inflight_bytes_ = 0;
790 : 331 : impl->state_.set(LinkState::Idle);
791 : 331 : impl->started_ = true;
792 : : impl->work_guard_ =
793 : 331 : std::make_unique<net::executor_work_guard<net::io_context::executor_type>>(impl->ioc_.get_executor());
794 [ + + ]: 331 : if (impl->owns_ioc_) {
795 : 42 : impl->ioc_thread_ = std::jthread([impl](std::stop_token st) {
796 : 21 : wirestead::concurrency::run_io_thread_init();
797 : : try {
798 : 21 : std::stop_callback cb(st, [impl] { impl->ioc_.stop(); });
799 : 21 : impl->ioc_.run();
800 : 21 : } catch (...) {
801 : 0 : }
802 : 42 : });
803 : : }
804 : 331 : net::post(impl->strand_, [self = shared_from_this(), generation] {
805 : 331 : auto impl = self->get_impl();
806 [ + - + + : 331 : if (generation != impl->generation_ || impl->stopping_) return;
+ + ]
807 : 330 : impl->state_.set(LinkState::Connecting);
808 : 330 : impl->notify_state();
809 : 330 : impl->open_and_configure(self);
810 : : });
811 : : }
812 : :
813 : 1044 : void Serial::stop() {
814 : 1044 : auto impl = get_impl();
815 : 1044 : const bool on_executor = impl->ioc_.get_executor().running_in_this_thread();
816 : : bool first;
817 : : {
818 : 1044 : std::lock_guard<std::mutex> lock(impl->submission_mtx_);
819 : 1044 : first = !impl->stopping_.exchange(true);
820 [ + + + - ]: 1044 : if (first) impl->send_accounting_.end(Impl::Ledger::Cause::ExplicitStop);
821 [ + + + + : 1044 : if (first && impl->write_wait_) impl->write_wait_->end(wrapper::SendRejection::CancelledWhileWaiting);
+ + ]
822 : 1044 : }
823 [ + + ]: 1044 : if (first) {
824 [ + + ]: 692 : if (impl->generation_.load() == 0)
825 : 15 : impl->perform_stop_cleanup();
826 : : else {
827 : 331 : auto self = weak_from_this().lock();
828 : 662 : net::post(impl->strand_, [self, impl] { impl->perform_stop_cleanup(); });
829 : 331 : }
830 : : }
831 [ + + ]: 1044 : if (on_executor) return;
832 : 975 : detail::stop_test_hook(impl, false);
833 : : {
834 : 975 : std::unique_lock<std::mutex> lock(impl->stop_mtx_);
835 : 2235 : impl->stop_cv_.wait(lock, [impl] { return impl->cleanup_done_; });
836 : 975 : }
837 : 975 : std::lock_guard<std::mutex> join_lock(impl->join_mtx_);
838 [ + + + - ]: 975 : if (impl->ioc_thread_.joinable()) impl->ioc_thread_.join();
839 : 975 : }
840 : :
841 : 854 : bool Serial::is_connected() const { return get_impl()->opened_.load(); }
842 : 211 : bool Serial::is_backpressure_active() const { return get_impl()->backpressure_active_.load(); }
843 : 396 : std::optional<size_t> Serial::write_queue_limit() const { return get_impl()->bp_limit_; }
844 : 744 : wrapper::RuntimeStats Serial::stats() const {
845 : 744 : auto impl = get_impl();
846 : 2232 : auto result = impl->stats_.snapshot(impl->queued_bytes_.load(std::memory_order_relaxed),
847 : : impl->pending_bytes_.load(std::memory_order_relaxed),
848 : 744 : impl->backpressure_active_.load(std::memory_order_relaxed));
849 : 744 : result.send_accounting = impl->send_accounting_.snapshot();
850 : 744 : return result;
851 : : }
852 : 23 : void Serial::reset_stats() {
853 : 23 : auto impl = get_impl();
854 : 23 : std::lock_guard<std::mutex> lock(impl->submission_mtx_);
855 : 23 : impl->send_accounting_.reset();
856 : 23 : impl->stats_.reset(impl->queued_bytes_.load(std::memory_order_relaxed) +
857 : 46 : impl->pending_bytes_.load(std::memory_order_relaxed));
858 : 23 : }
859 : :
860 : 23 : std::optional<diagnostics::ErrorInfo> Serial::last_error_info() const {
861 : 23 : return get_impl()->error_info_holder_.last_error_info();
862 : : }
863 : :
864 : 5 : void Serial::fail_receive() {
865 : 5 : const auto connection = write_connection();
866 [ - + ]: 5 : if (!connection) return;
867 : 5 : const auto run = impl_->generation_.load();
868 : 5 : net::dispatch(impl_->strand_, [self = shared_from_this(), connection, run] {
869 [ + - + - : 10 : if (run != self->impl_->generation_.load() || self->write_connection() != connection) return;
- + - + ]
870 : 5 : self->impl_->handle_error(self, "receive_limit", make_error_code(boost::system::errc::no_buffer_space));
871 : : });
872 : : }
873 : :
874 : 941 : boost::asio::any_io_executor Serial::get_executor() { return impl_->strand_; }
875 : :
876 : 72 : std::shared_ptr<detail::SerialWriteWait> Serial::capture_write_wait() const {
877 : 72 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
878 [ + - - + : 72 : if (impl_->stopping_.load() || !impl_->opened_.load()) return {};
- + ]
879 : 72 : return impl_->write_wait_;
880 : 72 : }
881 : :
882 : 36 : std::optional<wrapper::SendResult> Serial::poll_write_wait(const std::shared_ptr<detail::SerialWriteWait>& wait) const {
883 : 36 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
884 [ - + ]: 36 : if (!wait) return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
885 [ + + ]: 36 : if (wait->ended_by) return wrapper::SendResult::reject(*wait->ended_by);
886 [ + - - + : 24 : if (wait->sequence != impl_->connection_seq_.load() || !impl_->opened_.load())
- + ]
887 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::NotReady);
888 [ + - ]: 12 : if (!impl_->backpressure_active_.load()) return wrapper::SendResult::accept();
889 : 0 : return std::nullopt;
890 : 36 : }
891 : :
892 : 467 : void Serial::cancel_write_waits() {
893 : 467 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
894 [ + + ]: 467 : if (impl_->write_wait_) impl_->write_wait_->end(wrapper::SendRejection::CancelledWhileWaiting);
895 : 467 : }
896 : :
897 : 93 : std::optional<uint64_t> Serial::write_connection() const {
898 : 93 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
899 [ + - - + : 93 : if (impl_->stopping_.load() || !impl_->opened_.load()) return std::nullopt;
- + ]
900 : 186 : return impl_->connection_seq_.load();
901 : 93 : }
902 : :
903 : 119 : wrapper::SendResult Serial::write_state() {
904 : 119 : std::lock_guard<std::mutex> lock(impl_->submission_mtx_);
905 [ + - + + ]: 119 : if (auto reason = impl_->admission_rejection()) return wrapper::SendResult::reject(*reason);
906 : 72 : return wrapper::SendResult::accept();
907 : 119 : }
908 : :
909 : 107 : wrapper::SendResult Serial::async_write_copy_result(memory::ConstByteSpan data) {
910 : 107 : const auto result = write_copy(data, std::nullopt);
911 [ + + + - ]: 107 : if (auto hook = detail::g_serial_write_result_hook.load()) hook(result);
912 : 107 : return result;
913 : : }
914 : :
915 : 134 : wrapper::SendResult Serial::write_copy(memory::ConstByteSpan data, std::optional<uint64_t> expected_connection) {
916 [ + + ]: 134 : if (expected_connection) {
917 [ + + + - ]: 27 : if (auto hook = detail::g_serial_pinned_write_hook.load()) hook();
918 : : }
919 : 134 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
920 : 134 : const auto seq = impl_->generation_.load();
921 : 134 : const auto connection = impl_->connection_seq_.load();
922 [ + - + + ]: 134 : if (auto reason = impl_->admission_rejection(expected_connection)) {
923 : 15 : impl_->stats_.record_failed_send();
924 : 15 : return wrapper::SendResult::reject(*reason);
925 : : }
926 [ - + - - ]: 119 : if (auto hook = detail::g_serial_write_admission_hook.load()) hook();
927 : :
928 : 119 : size_t size = data.size();
929 [ + + ]: 119 : if (size == 0) {
930 : 4 : WIRESTEAD_LOG_WARNING("serial", "async_write_copy", "Ignoring zero-length write");
931 : 4 : impl_->stats_.record_failed_send();
932 : 4 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
933 : : }
934 : :
935 [ - + ]: 115 : if (size > base::constants::MAX_BUFFER_SIZE) {
936 : 0 : WIRESTEAD_LOG_ERROR("serial", "async_write_copy",
937 : : fmt::format("Write size exceeds maximum allowed ({} bytes)", size));
938 : 0 : impl_->stats_.record_failed_send();
939 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
940 : : }
941 : :
942 [ + + + + : 115 : if (size <= 65536 && impl_->cfg_.enable_memory_pool) {
+ + ]
943 : : try {
944 : 71 : memory::PooledBuffer pooled_buffer(size, impl_->pool_);
945 [ + - + - ]: 71 : if (pooled_buffer.valid()) {
946 : 71 : base::safe_memory::safe_memcpy(pooled_buffer.data(), data.data(), size);
947 : 71 : const auto added = pooled_buffer.size();
948 [ + - + + ]: 71 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queued_bytes_, impl_->pending_bytes_,
949 : 71 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
950 : 3 : impl_->stats_.record_failed_send();
951 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
952 : : }
953 : 68 : impl_->stats_.record_accepted(added);
954 : 68 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
955 : 68 : const auto request = admission.request();
956 : 68 : net::post(impl_->strand_, [self = shared_from_this(), buf = std::move(pooled_buffer), added, seq, connection,
957 : : request]() mutable {
958 [ - + ]: 136 : if (seq != self->impl_->generation_.load()) return;
959 [ + + ]: 136 : if (connection != self->impl_->connection_seq_.load()) {
960 : 3 : self->impl_->stats_.record_dropped(1, added);
961 : 3 : queue_util::release_reserved_limit_bytes(self->impl_->write_reserve_mtx_, self->impl_->inflight_bytes_,
962 : : added);
963 : 3 : return;
964 : : }
965 : 65 : self->impl_->route_enqueued_buffer(self, Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
966 : : });
967 : 68 : admission.commit();
968 : 68 : return wrapper::SendResult::accept();
969 : 68 : }
970 : 71 : } catch (const std::exception& e) {
971 : 0 : WIRESTEAD_LOG_ERROR("serial", "async_write_copy", fmt::format("Failed to acquire pooled buffer: {}", e.what()));
972 : 0 : }
973 : : }
974 : :
975 : 44 : std::vector<uint8_t> fallback(data.begin(), data.end());
976 : 44 : const auto added = fallback.size();
977 [ + - + + ]: 44 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queued_bytes_, impl_->pending_bytes_,
978 : 44 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
979 : 3 : impl_->stats_.record_failed_send();
980 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
981 : : }
982 : 41 : impl_->stats_.record_accepted(added);
983 : 41 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
984 : 41 : const auto request = admission.request();
985 : :
986 : 41 : net::post(impl_->strand_, [self = shared_from_this(), buf = std::move(fallback), added, seq, connection,
987 : : request]() mutable {
988 [ - + ]: 82 : if (seq != self->impl_->generation_.load()) return;
989 [ + + ]: 82 : if (connection != self->impl_->connection_seq_.load()) {
990 : 3 : self->impl_->stats_.record_dropped(1, added);
991 : 3 : queue_util::release_reserved_limit_bytes(self->impl_->write_reserve_mtx_, self->impl_->inflight_bytes_, added);
992 : 3 : return;
993 : : }
994 : 38 : self->impl_->route_enqueued_buffer(self, Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
995 : : });
996 : 41 : admission.commit();
997 : 41 : return wrapper::SendResult::accept();
998 : 134 : }
999 : :
1000 : 167 : wrapper::SendResult Serial::async_write_move_result(std::vector<uint8_t>&& data) {
1001 : 167 : const auto result = write_move(std::move(data), std::nullopt);
1002 [ + + + - ]: 167 : if (auto hook = detail::g_serial_write_result_hook.load()) hook(result);
1003 : 167 : return result;
1004 : : }
1005 : :
1006 : 183 : wrapper::SendResult Serial::write_move(std::vector<uint8_t>&& data, std::optional<uint64_t> expected_connection) {
1007 [ + + ]: 183 : if (expected_connection) {
1008 [ + + + - ]: 16 : if (auto hook = detail::g_serial_pinned_write_hook.load()) hook();
1009 : : }
1010 : 183 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
1011 : 183 : const auto seq = impl_->generation_.load();
1012 : 183 : const auto connection = impl_->connection_seq_.load();
1013 [ + - + + ]: 183 : if (auto reason = impl_->admission_rejection(expected_connection)) {
1014 : 12 : impl_->stats_.record_failed_send();
1015 : 12 : return wrapper::SendResult::reject(*reason);
1016 : : }
1017 [ - + - - ]: 171 : if (auto hook = detail::g_serial_write_admission_hook.load()) hook();
1018 : 171 : const auto size = data.size();
1019 [ + + ]: 171 : if (size == 0) {
1020 : 3 : WIRESTEAD_LOG_WARNING("serial", "async_write_move", "Ignoring zero-length write");
1021 : 3 : impl_->stats_.record_failed_send();
1022 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
1023 : : }
1024 [ - + ]: 168 : if (size > base::constants::MAX_BUFFER_SIZE) {
1025 : 0 : WIRESTEAD_LOG_ERROR("serial", "async_write_move",
1026 : : fmt::format("Write size exceeds maximum allowed ({} bytes)", size));
1027 : 0 : impl_->stats_.record_failed_send();
1028 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
1029 : : }
1030 : :
1031 : 168 : const auto added = size;
1032 [ + - + + ]: 168 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queued_bytes_, impl_->pending_bytes_,
1033 : 168 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
1034 : 17 : impl_->stats_.record_failed_send();
1035 : 17 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1036 : : }
1037 : 151 : impl_->stats_.record_accepted(added);
1038 : 151 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
1039 : 151 : const auto request = admission.request();
1040 : 151 : net::post(impl_->strand_, [self = shared_from_this(), buf = std::move(data), added, seq, connection,
1041 : : request]() mutable {
1042 [ - + ]: 302 : if (seq != self->impl_->generation_.load()) return;
1043 [ + + ]: 302 : if (connection != self->impl_->connection_seq_.load()) {
1044 : 5 : self->impl_->stats_.record_dropped(1, added);
1045 : 5 : queue_util::release_reserved_limit_bytes(self->impl_->write_reserve_mtx_, self->impl_->inflight_bytes_, added);
1046 : 5 : return;
1047 : : }
1048 : 146 : self->impl_->route_enqueued_buffer(self, Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
1049 : : });
1050 : 151 : admission.commit();
1051 : 151 : return wrapper::SendResult::accept();
1052 : 183 : }
1053 : :
1054 : 69 : wrapper::SendResult Serial::async_write_shared_result(std::shared_ptr<const std::vector<uint8_t>> data) {
1055 : 69 : const auto result = write_shared(std::move(data), std::nullopt);
1056 [ + + + - ]: 69 : if (auto hook = detail::g_serial_write_result_hook.load()) hook(result);
1057 : 69 : return result;
1058 : : }
1059 : :
1060 : 73 : wrapper::SendResult Serial::write_shared(std::shared_ptr<const std::vector<uint8_t>> data,
1061 : : std::optional<uint64_t> expected_connection) {
1062 [ + + ]: 73 : if (expected_connection) {
1063 [ + + + - ]: 4 : if (auto hook = detail::g_serial_pinned_write_hook.load()) hook();
1064 : : }
1065 : 73 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
1066 : 73 : const auto seq = impl_->generation_.load();
1067 : 73 : const auto connection = impl_->connection_seq_.load();
1068 [ + - + + ]: 73 : if (auto reason = impl_->admission_rejection(expected_connection)) {
1069 : 11 : impl_->stats_.record_failed_send();
1070 : 11 : return wrapper::SendResult::reject(*reason);
1071 : : }
1072 [ - + - - ]: 62 : if (auto hook = detail::g_serial_write_admission_hook.load()) hook();
1073 [ + - + + : 62 : if (!data || data->empty()) {
+ + ]
1074 : 3 : WIRESTEAD_LOG_WARNING("serial", "async_write_shared", "Ignoring empty shared buffer");
1075 : 3 : impl_->stats_.record_failed_send();
1076 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
1077 : : }
1078 : 59 : const auto size = data->size();
1079 [ - + ]: 59 : if (size > base::constants::MAX_BUFFER_SIZE) {
1080 : 0 : WIRESTEAD_LOG_ERROR("serial", "async_write_shared",
1081 : : fmt::format("Write size exceeds maximum allowed ({} bytes)", size));
1082 : 0 : impl_->stats_.record_failed_send();
1083 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
1084 : : }
1085 : :
1086 : 59 : const auto added = size;
1087 [ + - + + ]: 59 : if (!queue_util::try_reserve_limit_bytes(impl_->write_reserve_mtx_, impl_->queued_bytes_, impl_->pending_bytes_,
1088 : 59 : impl_->inflight_bytes_, added, impl_->bp_limit_)) {
1089 : 5 : impl_->stats_.record_failed_send();
1090 : 5 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1091 : : }
1092 : 54 : impl_->stats_.record_accepted(added);
1093 : 54 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
1094 : 54 : const auto request = admission.request();
1095 : 54 : net::post(impl_->strand_, [self = shared_from_this(), buf = std::move(data), added, seq, connection,
1096 : : request]() mutable {
1097 [ - + ]: 108 : if (seq != self->impl_->generation_.load()) return;
1098 [ + + ]: 108 : if (connection != self->impl_->connection_seq_.load()) {
1099 : 5 : self->impl_->stats_.record_dropped(1, added);
1100 : 5 : queue_util::release_reserved_limit_bytes(self->impl_->write_reserve_mtx_, self->impl_->inflight_bytes_, added);
1101 : 5 : return;
1102 : : }
1103 : 49 : self->impl_->route_enqueued_buffer(self, Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request}, added);
1104 : : });
1105 : 54 : admission.commit();
1106 : 54 : return wrapper::SendResult::accept();
1107 : 73 : }
1108 : :
1109 : 68 : wrapper::SendResult Serial::async_try_write_copy_result(memory::ConstByteSpan data) {
1110 : 68 : const auto result = try_write_copy(data);
1111 [ + + + - ]: 68 : if (auto hook = detail::g_serial_write_result_hook.load()) hook(result);
1112 : 68 : return result;
1113 : : }
1114 : :
1115 : 105 : wrapper::SendResult Serial::try_write_copy(memory::ConstByteSpan data) {
1116 [ + + ]: 105 : if (data.empty()) {
1117 : 3 : impl_->stats_.record_failed_send();
1118 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
1119 : : }
1120 [ - + ]: 102 : if (data.size() > base::constants::MAX_BUFFER_SIZE) {
1121 : 0 : impl_->stats_.record_failed_send();
1122 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
1123 : : }
1124 : 204 : return try_write_move(std::vector<uint8_t>(data.begin(), data.end()));
1125 : : }
1126 : :
1127 : 76 : wrapper::SendResult Serial::async_try_write_move_result(std::vector<uint8_t>&& data) {
1128 : 76 : const auto result = try_write_move(std::move(data));
1129 [ + + + - ]: 76 : if (auto hook = detail::g_serial_write_result_hook.load()) hook(result);
1130 : 76 : return result;
1131 : : }
1132 : :
1133 : 190 : wrapper::SendResult Serial::try_write_move(std::vector<uint8_t>&& data) {
1134 : 190 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
1135 : 190 : const auto seq = impl_->generation_.load();
1136 : 190 : const auto connection = impl_->connection_seq_.load();
1137 [ + - + + ]: 190 : if (auto reason = impl_->admission_rejection()) {
1138 : 38 : impl_->stats_.record_failed_send();
1139 : 38 : return wrapper::SendResult::reject(*reason);
1140 : : }
1141 [ - + - - ]: 152 : if (auto hook = detail::g_serial_write_admission_hook.load()) hook();
1142 : 152 : const auto added = data.size();
1143 [ + + - + ]: 152 : if (added == 0 || added > base::constants::MAX_BUFFER_SIZE) {
1144 : 3 : impl_->stats_.record_failed_send();
1145 [ + - ]: 3 : return wrapper::SendResult::reject(added == 0 ? wrapper::SendRejection::InvalidArgument
1146 : 3 : : wrapper::SendRejection::TooLarge);
1147 : : }
1148 : 18 : const auto reject_for_pressure = [this, added]() {
1149 [ + + ]: 18 : if (impl_->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
1150 : 11 : impl_->stats_.record_dropped(1, added);
1151 : : } else {
1152 : 7 : impl_->stats_.record_failed_send();
1153 : : }
1154 : 167 : };
1155 [ + + + + : 280 : if (impl_->backpressure_active_.load() || impl_->queued_bytes_ + added > impl_->bp_high_ ||
+ + ]
1156 [ - + ]: 131 : impl_->queued_bytes_ + impl_->pending_bytes_ + added > impl_->bp_limit_) {
1157 : 18 : reject_for_pressure();
1158 : 18 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1159 : : }
1160 [ + - - + ]: 131 : if (!queue_util::try_reserve_write_bytes(impl_->write_reserve_mtx_, impl_->inflight_bytes_, impl_->queued_bytes_,
1161 : 131 : impl_->pending_bytes_, impl_->backpressure_active_, added, impl_->bp_high_,
1162 : 131 : impl_->bp_limit_)) {
1163 : 0 : reject_for_pressure();
1164 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1165 : : }
1166 : 131 : impl_->stats_.record_accepted(added);
1167 : 131 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
1168 : 131 : const auto request = admission.request();
1169 : :
1170 : 131 : net::post(impl_->strand_, [self = shared_from_this(), buf = std::move(data), added, seq, connection,
1171 : : request]() mutable {
1172 [ - + ]: 262 : if (seq != self->impl_->generation_.load()) return;
1173 [ + + ]: 262 : if (connection != self->impl_->connection_seq_.load()) {
1174 : 10 : self->impl_->stats_.record_dropped(1, added);
1175 : : // The old connection drain already removed this queue reservation.
1176 : 10 : return;
1177 : : }
1178 : 121 : auto impl = self->impl_.get();
1179 [ + + + - : 121 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
- + + + ]
1180 : 18 : queue_util::release_reserved_write_bytes(impl->queued_bytes_, added);
1181 : 18 : impl->stats_.record_failed_send();
1182 : 18 : return;
1183 : : }
1184 : :
1185 : 103 : impl->tx_.push_back(Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request});
1186 : 103 : impl->observe_queue();
1187 : 103 : impl->report_backpressure(impl->queued_bytes_);
1188 [ + + + - ]: 103 : if (!impl->writing_) impl->do_write(self);
1189 : : });
1190 : 131 : admission.commit();
1191 : 131 : return wrapper::SendResult::accept();
1192 : 190 : }
1193 : :
1194 : 68 : wrapper::SendResult Serial::async_try_write_shared_result(std::shared_ptr<const std::vector<uint8_t>> data) {
1195 : 68 : const auto result = try_write_shared(std::move(data));
1196 [ + + + - ]: 68 : if (auto hook = detail::g_serial_write_result_hook.load()) hook(result);
1197 : 68 : return result;
1198 : : }
1199 : :
1200 : 80 : wrapper::SendResult Serial::try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) {
1201 : 80 : std::lock_guard<std::mutex> submission_lock(impl_->submission_mtx_);
1202 : 80 : const auto seq = impl_->generation_.load();
1203 : 80 : const auto connection = impl_->connection_seq_.load();
1204 [ + - + + : 80 : if (!data || data->empty()) {
+ + ]
1205 : 3 : impl_->stats_.record_failed_send();
1206 : 3 : return wrapper::SendResult::reject(wrapper::SendRejection::InvalidArgument);
1207 : : }
1208 [ - + ]: 77 : if (data->size() > base::constants::MAX_BUFFER_SIZE) {
1209 : 0 : impl_->stats_.record_failed_send();
1210 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::TooLarge);
1211 : : }
1212 : 77 : const auto added = data->size();
1213 : 7 : const auto reject_for_pressure = [this, added]() {
1214 [ + + ]: 7 : if (impl_->bp_strategy_ == base::constants::BackpressureStrategy::BestEffort) {
1215 : 4 : impl_->stats_.record_dropped(1, added);
1216 : : } else {
1217 : 3 : impl_->stats_.record_failed_send();
1218 : : }
1219 : 84 : };
1220 [ + - + + ]: 77 : if (auto reason = impl_->admission_rejection()) {
1221 : 16 : impl_->stats_.record_failed_send();
1222 : 16 : return wrapper::SendResult::reject(*reason);
1223 : : }
1224 [ - + - - ]: 61 : if (auto hook = detail::g_serial_write_admission_hook.load()) hook();
1225 [ + + + + : 115 : if (impl_->backpressure_active_.load() || impl_->queued_bytes_ + added > impl_->bp_high_ ||
+ + ]
1226 [ - + ]: 54 : impl_->queued_bytes_ + impl_->pending_bytes_ + added > impl_->bp_limit_) {
1227 : 7 : reject_for_pressure();
1228 : 7 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1229 : : }
1230 [ + - - + ]: 54 : if (!queue_util::try_reserve_write_bytes(impl_->write_reserve_mtx_, impl_->inflight_bytes_, impl_->queued_bytes_,
1231 : 54 : impl_->pending_bytes_, impl_->backpressure_active_, added, impl_->bp_high_,
1232 : 54 : impl_->bp_limit_)) {
1233 : 0 : reject_for_pressure();
1234 : 0 : return wrapper::SendResult::reject(wrapper::SendRejection::WouldBlock);
1235 : : }
1236 : 54 : impl_->stats_.record_accepted(added);
1237 : 54 : Impl::Ledger::Admission admission(impl_->send_accounting_, added);
1238 : 54 : const auto request = admission.request();
1239 : :
1240 : 54 : net::post(impl_->strand_, [self = shared_from_this(), buf = std::move(data), added, seq, connection,
1241 : : request]() mutable {
1242 [ - + ]: 108 : if (seq != self->impl_->generation_.load()) return;
1243 [ + + ]: 108 : if (connection != self->impl_->connection_seq_.load()) {
1244 : 5 : self->impl_->stats_.record_dropped(1, added);
1245 : : // The old connection drain already removed this queue reservation.
1246 : 5 : return;
1247 : : }
1248 : 49 : auto impl = self->impl_.get();
1249 [ + + + - : 49 : if (impl->stopping_.load() || impl->state_.is_state(LinkState::Closed) || impl->state_.is_state(LinkState::Error)) {
- + + + ]
1250 : 3 : queue_util::release_reserved_write_bytes(impl->queued_bytes_, added);
1251 : 3 : impl->stats_.record_failed_send();
1252 : 3 : return;
1253 : : }
1254 : :
1255 : 46 : impl->tx_.push_back(Impl::TrackedBuffer{BufferVariant{std::move(buf)}, request});
1256 : 46 : impl->observe_queue();
1257 : 46 : impl->report_backpressure(impl->queued_bytes_);
1258 [ + + + - ]: 46 : if (!impl->writing_) impl->do_write(self);
1259 : : });
1260 : 54 : admission.commit();
1261 : 54 : return wrapper::SendResult::accept();
1262 : 80 : }
1263 : :
1264 : 699 : void Serial::on_bytes(OnBytes cb) {
1265 : 699 : auto shared = interface::share_callback(std::move(cb));
1266 : 699 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
1267 : 699 : impl_->on_bytes_ = std::move(shared);
1268 : 699 : }
1269 : 721 : void Serial::on_state(OnState cb) {
1270 : 721 : auto shared = interface::share_callback(std::move(cb));
1271 : 721 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
1272 : 721 : impl_->on_state_ = std::move(shared);
1273 : 721 : }
1274 : 700 : void Serial::on_backpressure(OnBackpressure cb) {
1275 : 700 : auto shared = interface::share_callback(std::move(cb));
1276 : 700 : std::lock_guard<std::mutex> lock(impl_->callback_mtx_);
1277 : 700 : impl_->on_bp_ = std::move(shared);
1278 : 700 : }
1279 : :
1280 : 37 : void Serial::set_backpressure_strategy(base::constants::BackpressureStrategy strategy) {
1281 : 37 : config::detail::strategy(strategy);
1282 : 37 : get_impl()->bp_strategy_.store(strategy, std::memory_order_relaxed);
1283 : 37 : }
1284 : :
1285 : 0 : void Serial::set_retry_interval(unsigned interval_ms) {
1286 : 0 : config::detail::range(interval_ms, base::constants::MIN_RETRY_INTERVAL_MS, base::constants::MAX_RETRY_INTERVAL_MS,
1287 : : "invalid retry interval");
1288 : 0 : get_impl()->retry_interval_ms_.store(interval_ms, std::memory_order_relaxed);
1289 : 0 : }
1290 : :
1291 : : } // namespace transport
1292 : : } // namespace wirestead
|
