[pbs-devel] [PATCH proxmox-backup 04/12] tokio 1.0: AsyncRead/Seek with ReadBuf

["Fabian Grünbichler" <f.gruenbichler@proxmox.com>]

Signed-off-by: Fabian Grünbichler <f.gruenbichler@proxmox.com>
---
 src/backup/async_index_reader.rs   | 23 +++++++-------
 src/tools/async_io.rs              | 50 ++----------------------------
 src/tools/wrapped_reader_stream.rs |  8 +++--
 3 files changed, 19 insertions(+), 62 deletions(-)

diff --git a/src/backup/async_index_reader.rs b/src/backup/async_index_reader.rs
index f6a72099..2a04282c 100644
--- a/src/backup/async_index_reader.rs
+++ b/src/backup/async_index_reader.rs
@@ -6,7 +6,7 @@ use std::io::SeekFrom;
 use anyhow::Error;
 use futures::future::FutureExt;
 use futures::ready;
-use tokio::io::{AsyncRead, AsyncSeek};
+use tokio::io::{AsyncRead, AsyncSeek, ReadBuf};
 
 use proxmox::sys::error::io_err_other;
 use proxmox::io_format_err;
@@ -71,8 +71,8 @@ where
     fn poll_read(
         self: Pin<&mut Self>,
         cx: &mut Context,
-        buf: &mut [u8],
-    ) -> Poll<tokio::io::Result<usize>> {
+        buf: &mut ReadBuf,
+    ) -> Poll<tokio::io::Result<()>> {
         let this = Pin::get_mut(self);
         loop {
             match &mut this.state {
@@ -86,12 +86,12 @@ where
                     } else {
                         match this.index.chunk_from_offset(this.position) {
                             Some(res) => res,
-                            None => return Poll::Ready(Ok(0))
+                            None => return Poll::Ready(Ok(()))
                         }
                     };
 
                     if idx >= this.index.index_count() {
-                        return Poll::Ready(Ok(0));
+                        return Poll::Ready(Ok(()));
                     }
 
                     let info = this
@@ -142,13 +142,13 @@ where
                 AsyncIndexReaderState::HaveData => {
                     let offset = this.current_chunk_offset as usize;
                     let len = this.read_buffer.len();
-                    let n = if len - offset < buf.len() {
+                    let n = if len - offset < buf.remaining() {
                         len - offset
                     } else {
-                        buf.len()
+                        buf.remaining()
                     };
 
-                    buf[0..n].copy_from_slice(&this.read_buffer[offset..(offset + n)]);
+                    buf.put_slice(&this.read_buffer[offset..(offset + n)]);
                     this.position += n as u64;
 
                     if offset + n == len {
@@ -158,7 +158,7 @@ where
                         this.state = AsyncIndexReaderState::HaveData;
                     }
 
-                    return Poll::Ready(Ok(n));
+                    return Poll::Ready(Ok(()));
                 }
             }
         }
@@ -172,9 +172,8 @@ where
 {
     fn start_seek(
         self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
         pos: SeekFrom,
-    ) -> Poll<tokio::io::Result<()>> {
+    ) -> tokio::io::Result<()> {
         let this = Pin::get_mut(self);
         this.seek_to_pos = match pos {
             SeekFrom::Start(offset) => {
@@ -187,7 +186,7 @@ where
                 this.position as i64 + offset
             }
         };
-        Poll::Ready(Ok(()))
+        Ok(())
     }
 
     fn poll_complete(
diff --git a/src/tools/async_io.rs b/src/tools/async_io.rs
index 4e4107c0..3a5a6c9a 100644
--- a/src/tools/async_io.rs
+++ b/src/tools/async_io.rs
@@ -1,13 +1,12 @@
 //! Generic AsyncRead/AsyncWrite utilities.
 
 use std::io;
-use std::mem::MaybeUninit;
 use std::os::unix::io::{AsRawFd, RawFd};
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use futures::stream::{Stream, TryStream};
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::net::TcpListener;
 use hyper::client::connect::Connection;
 
@@ -20,8 +19,8 @@ impl<L: AsyncRead + Unpin, R: AsyncRead + Unpin> AsyncRead for EitherStream<L, R
     fn poll_read(
         self: Pin<&mut Self>,
         cx: &mut Context,
-        buf: &mut [u8],
-    ) -> Poll<Result<usize, io::Error>> {
+        buf: &mut ReadBuf,
+    ) -> Poll<Result<(), io::Error>> {
         match self.get_mut() {
             EitherStream::Left(ref mut s) => {
                 Pin::new(s).poll_read(cx, buf)
@@ -31,31 +30,6 @@ impl<L: AsyncRead + Unpin, R: AsyncRead + Unpin> AsyncRead for EitherStream<L, R
             }
         }
     }
-
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
-        match *self {
-            EitherStream::Left(ref s) => s.prepare_uninitialized_buffer(buf),
-            EitherStream::Right(ref s) => s.prepare_uninitialized_buffer(buf),
-        }
-    }
-
-    fn poll_read_buf<B>(
-        self: Pin<&mut Self>,
-        cx: &mut Context,
-        buf: &mut B,
-    ) -> Poll<Result<usize, io::Error>>
-    where
-        B: bytes::BufMut,
-    {
-        match self.get_mut() {
-            EitherStream::Left(ref mut s) => {
-                Pin::new(s).poll_read_buf(cx, buf)
-            }
-            EitherStream::Right(ref mut s) => {
-                Pin::new(s).poll_read_buf(cx, buf)
-            }
-        }
-    }
 }
 
 impl<L: AsyncWrite + Unpin, R: AsyncWrite + Unpin> AsyncWrite for EitherStream<L, R> {
@@ -95,24 +69,6 @@ impl<L: AsyncWrite + Unpin, R: AsyncWrite + Unpin> AsyncWrite for EitherStream<L
             }
         }
     }
-
-    fn poll_write_buf<B>(
-        self: Pin<&mut Self>,
-        cx: &mut Context,
-        buf: &mut B,
-    ) -> Poll<Result<usize, io::Error>>
-    where
-        B: bytes::Buf,
-    {
-        match self.get_mut() {
-            EitherStream::Left(ref mut s) => {
-                Pin::new(s).poll_write_buf(cx, buf)
-            }
-            EitherStream::Right(ref mut s) => {
-                Pin::new(s).poll_write_buf(cx, buf)
-            }
-        }
-    }
 }
 
 // we need this for crate::client::http_client:
diff --git a/src/tools/wrapped_reader_stream.rs b/src/tools/wrapped_reader_stream.rs
index 0294cc21..4b01b072 100644
--- a/src/tools/wrapped_reader_stream.rs
+++ b/src/tools/wrapped_reader_stream.rs
@@ -3,7 +3,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::sync::mpsc::Receiver;
 
-use tokio::io::AsyncRead;
+use tokio::io::{AsyncRead, ReadBuf};
 use futures::ready;
 use futures::stream::Stream;
 
@@ -69,8 +69,10 @@ impl<R: AsyncRead + Unpin> Stream for AsyncReaderStream<R> {
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         let this = self.get_mut();
-        match ready!(Pin::new(&mut this.reader).poll_read(cx, &mut this.buffer)) {
-            Ok(n) => {
+        let mut read_buf = ReadBuf::new(&mut this.buffer);
+        match ready!(Pin::new(&mut this.reader).poll_read(cx, &mut read_buf)) {
+            Ok(()) => {
+                let n = read_buf.filled().len();
                 if n == 0 {
                     // EOF
                     Poll::Ready(None)
-- 
2.20.1