io: remove Poll from the AsyncSeek::start_seek return value (#2885)

This commit is contained in:
Zahari Dichev
2020-10-08 10:56:01 +02:00
committed by GitHub
parent d94ab62c54
commit 43bd11bf2f
4 changed files with 49 additions and 66 deletions
+2 -6
View File
@@ -125,12 +125,8 @@ where
L: AsyncSeek, L: AsyncSeek,
R: AsyncSeek, R: AsyncSeek,
{ {
fn start_seek( fn start_seek(self: Pin<&mut Self>, position: SeekFrom) -> Result<()> {
self: Pin<&mut Self>, delegate_call!(self.start_seek(position))
cx: &mut Context<'_>,
position: SeekFrom,
) -> Poll<Result<()>> {
delegate_call!(self.start_seek(cx, position))
} }
fn poll_complete(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<u64>> { fn poll_complete(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<u64>> {
+20 -25
View File
@@ -86,6 +86,8 @@ pub struct File {
/// error is observed while performing a read, it is saved until the next /// error is observed while performing a read, it is saved until the next
/// write / flush call. /// write / flush call.
last_write_err: Option<io::ErrorKind>, last_write_err: Option<io::ErrorKind>,
pos: u64,
} }
#[derive(Debug)] #[derive(Debug)]
@@ -199,6 +201,7 @@ impl File {
std: Arc::new(std), std: Arc::new(std),
state: State::Idle(Some(Buf::with_capacity(0))), state: State::Idle(Some(Buf::with_capacity(0))),
last_write_err: None, last_write_err: None,
pos: 0,
} }
} }
@@ -332,7 +335,9 @@ impl File {
self.state = Idle(Some(buf)); self.state = Idle(Some(buf));
match op { match op {
Operation::Seek(res) => res.map(|_| ()), Operation::Seek(res) => res.map(|pos| {
self.pos = pos;
}),
_ => unreachable!(), _ => unreachable!(),
} }
} }
@@ -524,9 +529,12 @@ impl AsyncRead for File {
self.last_write_err = Some(e.kind()); self.last_write_err = Some(e.kind());
self.state = Idle(Some(buf)); self.state = Idle(Some(buf));
} }
Operation::Seek(_) => { Operation::Seek(result) => {
assert!(buf.is_empty()); assert!(buf.is_empty());
self.state = Idle(Some(buf)); self.state = Idle(Some(buf));
if let Ok(pos) = result {
self.pos = pos;
}
continue; continue;
} }
} }
@@ -537,13 +545,10 @@ impl AsyncRead for File {
} }
impl AsyncSeek for File { impl AsyncSeek for File {
fn start_seek( fn start_seek(mut self: Pin<&mut Self>, mut pos: SeekFrom) -> io::Result<()> {
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
mut pos: SeekFrom,
) -> Poll<io::Result<()>> {
loop { loop {
match self.state { match self.state {
Busy(_) => panic!("must wait for poll_complete before calling start_seek"),
Idle(ref mut buf_cell) => { Idle(ref mut buf_cell) => {
let mut buf = buf_cell.take().unwrap(); let mut buf = buf_cell.take().unwrap();
@@ -562,22 +567,7 @@ impl AsyncSeek for File {
let res = (&*std).seek(pos); let res = (&*std).seek(pos);
(Operation::Seek(res), buf) (Operation::Seek(res), buf)
})); }));
return Ok(());
return Ready(Ok(()));
}
Busy(ref mut rx) => {
let (op, buf) = ready!(Pin::new(rx).poll(cx))?;
self.state = Idle(Some(buf));
match op {
Operation::Read(_) => {}
Operation::Write(Err(e)) => {
assert!(self.last_write_err.is_none());
self.last_write_err = Some(e.kind());
}
Operation::Write(_) => {}
Operation::Seek(_) => {}
}
} }
} }
} }
@@ -586,7 +576,7 @@ impl AsyncSeek for File {
fn poll_complete(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> { fn poll_complete(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
loop { loop {
match self.state { match self.state {
Idle(_) => panic!("must call start_seek before calling poll_complete"), Idle(_) => return Poll::Ready(Ok(self.pos)),
Busy(ref mut rx) => { Busy(ref mut rx) => {
let (op, buf) = ready!(Pin::new(rx).poll(cx))?; let (op, buf) = ready!(Pin::new(rx).poll(cx))?;
self.state = Idle(Some(buf)); self.state = Idle(Some(buf));
@@ -598,7 +588,12 @@ impl AsyncSeek for File {
self.last_write_err = Some(e.kind()); self.last_write_err = Some(e.kind());
} }
Operation::Write(_) => {} Operation::Write(_) => {}
Operation::Seek(res) => return Ready(res), Operation::Seek(res) => {
if let Ok(pos) = res {
self.pos = pos;
}
return Ready(res);
}
} }
} }
} }
+17 -28
View File
@@ -23,36 +23,33 @@ pub trait AsyncSeek {
/// ///
/// If this function returns successfully, then the job has been submitted. /// If this function returns successfully, then the job has been submitted.
/// To find out when it completes, call `poll_complete`. /// To find out when it completes, call `poll_complete`.
fn start_seek( ///
self: Pin<&mut Self>, /// # Errors
cx: &mut Context<'_>, ///
position: SeekFrom, /// This function can return [`io::ErrorKind::Other`] in case there is
) -> Poll<io::Result<()>>; /// another seek in progress. To avoid this, it is advisable that any call
/// to `start_seek` is preceded by a call to `poll_complete` to ensure all
/// pending seeks have completed.
fn start_seek(self: Pin<&mut Self>, position: SeekFrom) -> io::Result<()>;
/// Waits for a seek operation to complete. /// Waits for a seek operation to complete.
/// ///
/// If the seek operation completed successfully, /// If the seek operation completed successfully,
/// this method returns the new position from the start of the stream. /// this method returns the new position from the start of the stream.
/// That position can be used later with [`SeekFrom::Start`]. /// That position can be used later with [`SeekFrom::Start`]. Repeatedly
/// calling this function without calling `start_seek` might return the
/// same result.
/// ///
/// # Errors /// # Errors
/// ///
/// Seeking to a negative offset is considered an error. /// Seeking to a negative offset is considered an error.
///
/// # Panics
///
/// Calling this method without calling `start_seek` first is an error.
fn poll_complete(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>>; fn poll_complete(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>>;
} }
macro_rules! deref_async_seek { macro_rules! deref_async_seek {
() => { () => {
fn start_seek( fn start_seek(mut self: Pin<&mut Self>, pos: SeekFrom) -> io::Result<()> {
mut self: Pin<&mut Self>, Pin::new(&mut **self).start_seek(pos)
cx: &mut Context<'_>,
pos: SeekFrom,
) -> Poll<io::Result<()>> {
Pin::new(&mut **self).start_seek(cx, pos)
} }
fn poll_complete(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> { fn poll_complete(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
@@ -74,12 +71,8 @@ where
P: DerefMut + Unpin, P: DerefMut + Unpin,
P::Target: AsyncSeek, P::Target: AsyncSeek,
{ {
fn start_seek( fn start_seek(self: Pin<&mut Self>, pos: SeekFrom) -> io::Result<()> {
self: Pin<&mut Self>, self.get_mut().as_mut().start_seek(pos)
cx: &mut Context<'_>,
pos: SeekFrom,
) -> Poll<io::Result<()>> {
self.get_mut().as_mut().start_seek(cx, pos)
} }
fn poll_complete(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> { fn poll_complete(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<u64>> {
@@ -88,12 +81,8 @@ where
} }
impl<T: AsRef<[u8]> + Unpin> AsyncSeek for io::Cursor<T> { impl<T: AsRef<[u8]> + Unpin> AsyncSeek for io::Cursor<T> {
fn start_seek( fn start_seek(mut self: Pin<&mut Self>, pos: SeekFrom) -> io::Result<()> {
mut self: Pin<&mut Self>, io::Seek::seek(&mut *self, pos).map(drop)
_: &mut Context<'_>,
pos: SeekFrom,
) -> Poll<io::Result<()>> {
Poll::Ready(io::Seek::seek(&mut *self, pos).map(drop))
} }
fn poll_complete(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<u64>> { fn poll_complete(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<u64>> {
Poll::Ready(Ok(self.get_mut().position())) Poll::Ready(Ok(self.get_mut().position()))
+10 -7
View File
@@ -40,14 +40,17 @@ where
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let me = self.project(); let me = self.project();
match me.pos { match me.pos {
Some(pos) => match Pin::new(&mut *me.seek).start_seek(cx, *pos) { Some(pos) => {
Poll::Ready(Ok(())) => { // ensure no seek in progress
*me.pos = None; ready!(Pin::new(&mut *me.seek).poll_complete(cx))?;
Pin::new(&mut *me.seek).poll_complete(cx) match Pin::new(&mut *me.seek).start_seek(*pos) {
Ok(()) => {
*me.pos = None;
Pin::new(&mut *me.seek).poll_complete(cx)
}
Err(e) => Poll::Ready(Err(e)),
} }
Poll::Ready(Err(e)) => Poll::Ready(Err(e)), }
Poll::Pending => Poll::Pending,
},
None => Pin::new(&mut *me.seek).poll_complete(cx), None => Pin::new(&mut *me.seek).poll_complete(cx),
} }
} }