Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Update stream impl for Framed to return None after Err #4166

Merged
merged 8 commits into from Oct 26, 2021
Merged

Update stream impl for Framed to return None after Err #4166

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
3ae1591
Update stream impl for Framed to return `None` after `Err`
Oct 12, 2021
b098fea
Add framed stream test
Oct 13, 2021
83f88bd
Move stream to the paused state after returning None
Oct 13, 2021
0d504bd
Try fixing diagram formatting
Oct 13, 2021
0577753
Give fresh diagram a try
Oct 19, 2021
73d6d3f
Convert test to `tokio-test`
Oct 22, 2021
0154048
Merge branch 'master' into update-decoder
Oct 25, 2021
d387593
Update tokio-util/tests/framed_stream.rs
Darksonn Oct 26, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
1 change: 1 addition & 0 deletions tokio-util/src/codec/framed.rs
View file
Open in desktop
Expand Up @@ -106,6 +106,7 @@ where
eof: false,
is_readable: false,
buffer: BytesMut::with_capacity(capacity),
has_errored: false
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There are some rustfmt failures:

Diff in /home/runner/work/tokio/tokio/tokio-util/src/codec/framed.rs at line 106:
                         eof: false,
                         is_readable: false,
                         buffer: BytesMut::with_capacity(capacity),
-                        has_errored: false
+                        has_errored: false,
                     },
                     write: WriteFrame::default(),
                 },
Diff in /home/runner/work/tokio/tokio/tokio-util/src/codec/framed_read.rs at line 51:
                     eof: false,
                     is_readable: false,
                     buffer: BytesMut::with_capacity(capacity),
-                    has_errored: false
+                    has_errored: false,
                 },
             },
         }
Diff in /home/runner/work/tokio/tokio/tokio-util/src/codec/framed.rs at line 106:
                         eof: false,
                         is_readable: false,
                         buffer: BytesMut::with_capacity(capacity),
-                        has_errored: false
+                        has_errored: false,
                     },
                     write: WriteFrame::default(),
                 },
Diff in /home/runner/work/tokio/tokio/tokio-util/src/codec/framed_read.rs at line 51:
                     eof: false,
                     is_readable: false,
                     buffer: BytesMut::with_capacity(capacity),
-                    has_errored: false
+                    has_errored: false,
                 },
             },
         }

You can fix them by running

rustfmt --edition 2018 $(git ls-files '*.rs')

},
write: WriteFrame::default(),
},
Expand Down
66 changes: 41 additions & 25 deletions tokio-util/src/codec/framed_impl.rs
View file
Open in desktop
Expand Up @@ -31,6 +31,7 @@ pub(crate) struct ReadFrame {
pub(crate) eof: bool,
pub(crate) is_readable: bool,
pub(crate) buffer: BytesMut,
pub(crate) has_errored: bool,
}

pub(crate) struct WriteFrame {
Expand All @@ -49,6 +50,7 @@ impl Default for ReadFrame {
eof: false,
is_readable: false,
buffer: BytesMut::with_capacity(INITIAL_CAPACITY),
has_errored: false,
}
}
}
Expand All @@ -72,6 +74,7 @@ impl From<BytesMut> for ReadFrame {
buffer,
is_readable: size > 0,
eof: false,
has_errored: false,
}
}
}
Expand Down Expand Up @@ -126,30 +129,37 @@ where
//
// The initial state is `reading`.
//
// | state | eof | is_readable |
// |---------|-------|-------------|
// | reading | false | false |
// | framing | false | true |
// | pausing | true | true |
// | paused | true | false |
//
// `decode_eof`
// returns `Some` read 0 bytes
// │ │ │ │
// │ ▼ │ ▼
// ┌───────┐ `decode_eof` ┌──────┐
// ┌──read 0 bytes──▶│pausing│─returns `None`─▶│paused│──┐
// │ └───────┘ └──────┘ │
// pending read┐ │ ┌──────┐ │ ▲ │
// │ │ │ │ │ │ │ │
// │ ▼ │ │ `decode` returns `Some`│ pending read
// │ ╔═══════╗ ┌───────┐◀─┘ │
// └──║reading║─read n>0 bytes─▶│framing│ │
// ╚═══════╝ └───────┘◀──────read n>0 bytes┘
// ▲ │
// │ │
// └─`decode` returns `None`─┘
// | state | eof | is_readable | has_errored |
// |---------|-------|-------------|-------------|
// | reading | false | false | false |
// | framing | false | true | false |
// | pausing | true | true | false |
// | paused | true | false | false |
// | errored | <any> | <any> | true | `decode_eof` returns Err
// ┌───────────────────────────────────────┐
// `decode_eof` │ │
// returns Ok(`Some`) │ read 0 bytes │
// │ │ ┌───────┘ │ │ │
// │ ▼ │ │ ▼ │
// ┌───────────┐ `decode_eof` ┌──────┐ │
// ┌──read 0 bytes──▶│ pausing │─returns Ok(`None`)─▶│paused│───────┐ │
// │ └───────────┘ └──────┘ │ ▼
// pending read┐ │ ┌──────┐ │ ▲ │ ┌─────────┐
// │ │ │ │ │ │ │ │ │ errored │
// │ ▼ │ │ `decode` returns `Some` │ pending read └─────────┘
// │ ╔═══════╗ ┌───────────┐◀─┘ │ ▲
// └──║reading║─read n>0 bytes─▶│ framing │ │ │
// ╚═══════╝ └───────────┘◀──────read n>0 bytes┘ │
// ▲ │ │ │
// │ │ │ `decode` returns Err │
// └─`decode` returns `None`──┘ └────────────────────────────────────────────────┘
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This picture is kinda broken. I think you need to adjust some spaces to make stuff line up.

Copy link
Contributor Author

@bIgBV bIgBV Oct 12, 2021 •
edited

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was afraid this might happen. It looks fine on my machine.
image

Let me try fixing it.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You might need to use a different character for the arrows.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The diagram is still pretty broken.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I just copied the characters being used currently for the arrows. Also, it seems to render fine in the diff on my end as well
image

I wonder if I could clean it up by using plain old ascii instead of the unicode characters.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

that might indeed be the best way.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry for the late reply, I was traveling over the weekend. I'll push a new revision tonight.

loop {
// Return `None` if we have encountered an error from the underlying decoder
// See: https://github.com/tokio-rs/tokio/issues/3976
if state.has_errored {
return Poll::Ready(None);
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently the stream is not fused when you reach EOF normally. Maybe we should put it back into the paused state after returning None?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not quite sure what that means. If we move to the paused state, that means that the errored state is no longer terminal. I thought we wanted to return None to the stream after the underlying decoder errored out?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure, but do we want to keep returning none if they keep trying to read from it? See #3272 for context.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah I see what you mean now. Yeah, it makes sense to move it to the paused state after returning None.


// Repeatedly call `decode` or `decode_eof` while the buffer is "readable",
// i.e. it _might_ contain data consumable as a frame or closing frame.
// Both signal that there is no such data by returning `None`.
Expand All @@ -165,7 +175,10 @@ where
// pausing or framing
if state.eof {
// pausing
let frame = pinned.codec.decode_eof(&mut state.buffer)?;
let frame = pinned.codec.decode_eof(&mut state.buffer).map_err(|err| {
state.has_errored = true;
err
})?;
if frame.is_none() {
state.is_readable = false; // prepare pausing -> paused
}
Expand All @@ -176,7 +189,10 @@ where
// framing
trace!("attempting to decode a frame");

if let Some(frame) = pinned.codec.decode(&mut state.buffer)? {
if let Some(frame) = pinned.codec.decode(&mut state.buffer).map_err(|op| {
state.has_errored = true;
op
})? {
trace!("frame decoded from buffer");
// implicit framing -> framing
return Poll::Ready(Some(Ok(frame)));
Expand Down
1 change: 1 addition & 0 deletions tokio-util/src/codec/framed_read.rs
View file
Open in desktop
Expand Up @@ -51,6 +51,7 @@ where
eof: false,
is_readable: false,
buffer: BytesMut::with_capacity(capacity),
has_errored: false
},
},
}
Expand Down