itsscb/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2025-10-01 12:20:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

io: read during write in copy_bidirectional and copy (#6532)

Browse Source
This commit is contained in:
Armillus 2024-06-05 00:29:28 +02:00 committed by GitHub
parent 49609d073f
commit 3f397ccded
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 30 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
tokio/src/io/util/copy.rs
View File

@ -96,12 +96,9 @@ impl CopyBuffer {
// Keep track of task budget // Keep track of task budget
let coop = ready!(crate::runtime::coop::poll_proceed(cx)); let coop = ready!(crate::runtime::coop::poll_proceed(cx));
loop { loop {
// If our buffer is empty, then we need to read some data to // If there is some space left in our buffer, then we try to read some
// continue. // data to continue, thus maximizing the chances of a large write.
if self.pos == self.cap && !self.read_done { if self.cap < self.buf.len() && !self.read_done {
self.pos = 0;
self.cap = 0;
match self.poll_fill_buf(cx, reader.as_mut()) { match self.poll_fill_buf(cx, reader.as_mut()) {
Poll::Ready(Ok(())) => { Poll::Ready(Ok(())) => {
#[cfg(any( #[cfg(any(
@ -131,25 +128,29 @@ impl CopyBuffer {
return Poll::Ready(Err(err)); return Poll::Ready(Err(err));
} }
Poll::Pending => { Poll::Pending => {
// Try flushing when the reader has no progress to avoid deadlock // Ignore pending reads when our buffer is not empty, because
// when the reader depends on buffered writer. // we can try to write data immediately.
if self.need_flush { if self.pos == self.cap {
ready!(writer.as_mut().poll_flush(cx))?; // Try flushing when the reader has no progress to avoid deadlock
#[cfg(any( // when the reader depends on buffered writer.
feature = "fs", if self.need_flush {
feature = "io-std", ready!(writer.as_mut().poll_flush(cx))?;
feature = "net", #[cfg(any(
feature = "process", feature = "fs",
feature = "rt", feature = "io-std",
feature = "signal", feature = "net",
feature = "sync", feature = "process",
feature = "time", feature = "rt",
))] feature = "signal",
coop.made_progress(); feature = "sync",
self.need_flush = false; feature = "time",
} ))]
coop.made_progress();
self.need_flush = false;
}
return Poll::Pending; return Poll::Pending;
}
} }
} }
} }
@ -188,9 +189,13 @@ impl CopyBuffer {
"writer returned length larger than input slice" "writer returned length larger than input slice"
); );
// All data has been written, the buffer can be considered empty again
self.pos = 0;
self.cap = 0;
// If we've written all the data and we've seen EOF, flush out the // If we've written all the data and we've seen EOF, flush out the
// data and finish the transfer. // data and finish the transfer.
if self.pos == self.cap && self.read_done { if self.read_done {
ready!(writer.as_mut().poll_flush(cx))?; ready!(writer.as_mut().poll_flush(cx))?;
#[cfg(any( #[cfg(any(
feature = "fs", feature = "fs",