[pbs-devel] [PATCH proxmox-backup 04/12] tokio 1.0: AsyncRead/Seek with ReadBuf
Fabian Grünbichler
f.gruenbichler at proxmox.com
Tue Jan 12 14:58:18 CET 2021
Signed-off-by: Fabian Grünbichler <f.gruenbichler at 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
More information about the pbs-devel
mailing list