Skip to content

Commit 7f997f9

Browse files
jadamcrainjadamcrain
authored
Fix master task shutdown (#121)
* quick and dirty fix for infinite loop in state machine * use the type system to make it harder for this bug to occur * update tests and examples to target .NET 6
1 parent 056c6bb commit 7f997f9

File tree

11 files changed

+80
-54
lines changed

11 files changed

+80
-54
lines changed

‎Cargo.lock

Lines changed: 2 additions & 2 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎ffi/bindings/dotnet/examples/client/client.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
<PropertyGroup>
44
<OutputType>Exe</OutputType>
5-
<TargetFramework>netcoreapp3.1</TargetFramework>
5+
<TargetFramework>net6.0</TargetFramework>
66
<IsPublishable>False</IsPublishable>
77
</PropertyGroup>
88

‎ffi/bindings/dotnet/examples/server/server.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
<PropertyGroup>
44
<OutputType>Exe</OutputType>
5-
<TargetFramework>netcoreapp3.1</TargetFramework>
5+
<TargetFramework>net6.0</TargetFramework>
66
<IsPublishable>False</IsPublishable>
77
</PropertyGroup>
88

‎ffi/bindings/dotnet/rodbus-tests/rodbus-tests.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
<Project Sdk="Microsoft.NET.Sdk">
22

33
<PropertyGroup>
4-
<TargetFramework>netcoreapp3.1</TargetFramework>
4+
<TargetFramework>net6.0</TargetFramework>
55
<RootNamespace>rodbus_tests</RootNamespace>
66

77
<IsPackable>false</IsPackable>

‎rodbus/src/channel.rs

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
use crate::Shutdown;
2+
3+
/// wrap a Tokio receiver and only provide a recv() that returns a Result<T, Shutdown>
4+
/// that makes it harder to misuse.
5+
pub(crate) struct Receiver<T>(tokio::sync::mpsc::Receiver<T>);
6+
7+
impl<T> From<tokio::sync::mpsc::Receiver<T>> for Receiver<T> {
8+
fn from(value: tokio::sync::mpsc::Receiver<T>) -> Self {
9+
Self(value)
10+
}
11+
}
12+
13+
impl<T> Receiver<T> {
14+
pub(crate) async fn recv(&mut self) -> Result<T, Shutdown> {
15+
self.0.recv().await.ok_or(Shutdown)
16+
}
17+
}

‎rodbus/src/client/channel.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ impl Channel {
7373
let _ = crate::serial::client::SerialChannelTask::new(
7474
&path,
7575
serial_settings,
76-
rx,
76+
rx.into(),
7777
retry,
7878
decode,
7979
listener.unwrap_or_else(|| crate::client::NullListener::create()),

‎rodbus/src/client/task.rs

Lines changed: 52 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -25,17 +25,29 @@ pub(crate) enum SessionError {
2525
Shutdown,
2626
}
2727

28+
impl From<Shutdown> for SessionError {
29+
fn from(_: Shutdown) -> Self {
30+
SessionError::Shutdown
31+
}
32+
}
33+
2834
#[derive(Copy, Clone, Debug, PartialEq)]
2935
pub(crate) enum StateChange {
3036
Disable,
3137
Shutdown,
3238
}
3339

40+
impl From<Shutdown> for StateChange {
41+
fn from(_: Shutdown) -> Self {
42+
StateChange::Shutdown
43+
}
44+
}
45+
3446
impl std::fmt::Display for SessionError {
3547
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
3648
match self {
3749
SessionError::IoError(err) => {
38-
write!(f, "i/o error: {err}")
50+
write!(f, "I/O error: {err}")
3951
}
4052
SessionError::BadFrame => {
4153
write!(f, "Parser encountered a bad frame")
@@ -51,9 +63,9 @@ impl std::fmt::Display for SessionError {
5163
}
5264

5365
impl SessionError {
54-
pub(crate) fn from(err: &RequestError) -> Option<Self> {
66+
pub(crate) fn from_request_err(err: RequestError) -> Option<Self> {
5567
match err {
56-
RequestError::Io(x) => Some(SessionError::IoError(*x)),
68+
RequestError::Io(x) => Some(SessionError::IoError(x)),
5769
RequestError::BadFrame(_) => Some(SessionError::BadFrame),
5870
// all other errors don't kill the loop
5971
_ => None,
@@ -62,7 +74,7 @@ impl SessionError {
6274
}
6375

6476
pub(crate) struct ClientLoop {
65-
rx: tokio::sync::mpsc::Receiver<Command>,
77+
rx: crate::channel::Receiver<Command>,
6678
writer: FrameWriter,
6779
reader: FramedReader,
6880
tx_id: TxId,
@@ -72,7 +84,7 @@ pub(crate) struct ClientLoop {
7284

7385
impl ClientLoop {
7486
pub(crate) fn new(
75-
rx: tokio::sync::mpsc::Receiver<Command>,
87+
rx: crate::channel::Receiver<Command>,
7688
writer: FrameWriter,
7789
reader: FramedReader,
7890
decode: DecodeLevel,
@@ -118,32 +130,31 @@ impl ClientLoop {
118130

119131
pub(crate) async fn run(&mut self, io: &mut PhysLayer) -> SessionError {
120132
loop {
121-
tokio::select! {
122-
frame = self.reader.next_frame(io, self.decode) => {
123-
match frame {
124-
Ok(frame) => {
125-
tracing::warn!("Received unexpected frame while idle: {:?}", frame.header);
126-
}
127-
Err(err) => {
128-
if let Some(err) = SessionError::from(&err) {
129-
tracing::warn!("{}", err);
130-
return err;
131-
}
132-
}
133+
if let Err(err) = self.poll(io).await {
134+
tracing::warn!("ending session: {}", err);
135+
return err;
136+
}
137+
}
138+
}
139+
140+
pub(crate) async fn poll(&mut self, io: &mut PhysLayer) -> Result<(), SessionError> {
141+
tokio::select! {
142+
frame = self.reader.next_frame(io, self.decode) => {
143+
match frame {
144+
Ok(frame) => {
145+
tracing::warn!("Received unexpected frame while idle: {:?}", frame.header);
146+
Ok(())
133147
}
134-
}
135-
cmd = self.rx.recv() => {
136-
match cmd {
137-
// other side has closed the request channel
138-
None => return SessionError::Shutdown,
139-
Some(cmd) => {
140-
if let Err(err) = self.run_cmd(cmd, io).await {
141-
return err;
142-
}
143-
}
148+
Err(err) => match SessionError::from_request_err(err) {
149+
Some(err) => Err(err),
150+
None => Ok(()),
144151
}
145152
}
146153
}
154+
res = self.rx.recv() => {
155+
let cmd: Command = res?;
156+
self.run_cmd(cmd, io).await
157+
}
147158
}
148159
}
149160

@@ -166,7 +177,7 @@ impl ClientLoop {
166177

167178
// some request errors are a session error that will
168179
// bubble up and close the session
169-
if let Some(err) = SessionError::from(&err) {
180+
if let Some(err) = SessionError::from_request_err(err) {
170181
return Err(err);
171182
}
172183
}
@@ -240,21 +251,20 @@ impl ClientLoop {
240251
}
241252

242253
async fn fail_next_request(&mut self) -> Result<(), StateChange> {
243-
match self.rx.recv().await {
244-
None => return Err(StateChange::Disable),
245-
Some(cmd) => match cmd {
246-
Command::Request(mut req) => {
247-
req.details.fail(RequestError::NoConnection);
248-
}
249-
Command::Setting(x) => {
250-
self.change_setting(x);
251-
if !self.enabled {
252-
return Err(StateChange::Disable);
253-
}
254+
match self.rx.recv().await? {
255+
Command::Request(mut req) => {
256+
req.details.fail(RequestError::NoConnection);
257+
Ok(())
258+
}
259+
Command::Setting(x) => {
260+
self.change_setting(x);
261+
if self.enabled {
262+
Ok(())
263+
} else {
264+
Err(StateChange::Disable)
254265
}
255-
},
266+
}
256267
}
257-
Ok(())
258268
}
259269

260270
pub(crate) async fn fail_requests_for(
@@ -300,7 +310,7 @@ mod tests {
300310
let (tx, rx) = tokio::sync::mpsc::channel(16);
301311
let (mock, io_handle) = sfio_tokio_mock_io::mock();
302312
let mut client_loop = ClientLoop::new(
303-
rx,
313+
rx.into(),
304314
FrameWriter::tcp(),
305315
FramedReader::tcp(),
306316
DecodeLevel::default().application(AppDecodeLevel::DataValues),

‎rodbus/src/lib.rs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -167,6 +167,7 @@ pub mod constants;
167167
pub mod server;
168168

169169
// modules that are re-exported
170+
pub(crate) mod channel;
170171
pub(crate) mod decode;
171172
pub(crate) mod error;
172173
pub(crate) mod exception;

‎rodbus/src/serial/client.rs

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
use crate::common::phys::PhysLayer;
22
use crate::decode::DecodeLevel;
33
use crate::serial::SerialSettings;
4-
use tokio::sync::mpsc::Receiver;
54

65
use crate::client::message::Command;
76
use crate::client::task::{ClientLoop, SessionError, StateChange};
@@ -21,7 +20,7 @@ impl SerialChannelTask {
2120
pub(crate) fn new(
2221
path: &str,
2322
serial_settings: SerialSettings,
24-
rx: Receiver<Command>,
23+
rx: crate::channel::Receiver<Command>,
2524
retry: Box<dyn RetryStrategy>,
2625
decode: DecodeLevel,
2726
listener: Box<dyn Listener<PortState>>,

‎rodbus/src/tcp/client.rs

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@ use crate::error::Shutdown;
1111
use crate::retry::RetryStrategy;
1212

1313
use tokio::net::TcpStream;
14-
use tokio::sync::mpsc::Receiver;
1514

1615
pub(crate) fn spawn_tcp_channel(
1716
host: HostAddr,
@@ -37,7 +36,7 @@ pub(crate) fn create_tcp_channel(
3736
let task = async move {
3837
TcpChannelTask::new(
3938
host.clone(),
40-
rx,
39+
rx.into(),
4140
TcpTaskConnectionHandler::Tcp,
4241
connect_retry,
4342
decode,
@@ -81,7 +80,7 @@ pub(crate) struct TcpChannelTask {
8180
impl TcpChannelTask {
8281
pub(crate) fn new(
8382
host: HostAddr,
84-
rx: Receiver<Command>,
83+
rx: crate::channel::Receiver<Command>,
8584
connection_handler: TcpTaskConnectionHandler,
8685
connect_retry: Box<dyn RetryStrategy>,
8786
decode: DecodeLevel,

‎rodbus/src/tcp/tls/client.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,7 @@ pub(crate) fn create_tls_channel(
5454
let task = async move {
5555
TcpChannelTask::new(
5656
host.clone(),
57-
rx,
57+
rx.into(),
5858
TcpTaskConnectionHandler::Tls(tls_config),
5959
connect_retry,
6060
decode,

0 commit comments

Comments
 (0)