You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2024/03/28 07:51:12 UTC
(opendal) 01/01: refactor: Impl into_futures_io_async_write for Writer
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch polish-writer
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit edd6b05dfe76bc1cc51803d350d6bc9147569df4
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Mar 28 15:50:46 2024 +0800
refactor: Impl into_futures_io_async_write for Writer
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/types/mod.rs | 1 +
core/src/types/writer.rs | 268 +++++++++++++------------------------
core/tests/behavior/async_write.rs | 109 ++-------------
3 files changed, 110 insertions(+), 268 deletions(-)
diff --git a/core/src/types/mod.rs b/core/src/types/mod.rs
index daf1a25b9d..da246e64cd 100644
--- a/core/src/types/mod.rs
+++ b/core/src/types/mod.rs
@@ -36,6 +36,7 @@ pub use blocking_reader::into_std_read::StdIoReader;
pub use blocking_reader::BlockingReader;
mod writer;
+pub use writer::into_futures_async_write::FuturesIoAsyncWriter;
pub use writer::BlockingWriter;
pub use writer::Writer;
diff --git a/core/src/types/writer.rs b/core/src/types/writer.rs
index 64c32667c0..74b1247a33 100644
--- a/core/src/types/writer.rs
+++ b/core/src/types/writer.rs
@@ -17,10 +17,6 @@
use std::io;
use std::pin::pin;
-use std::pin::Pin;
-use std::task::ready;
-use std::task::Context;
-use std::task::Poll;
use bytes::Buf;
use bytes::Bytes;
@@ -72,15 +68,13 @@ use crate::*;
/// - Services that doesn't support append will return [`ErrorKind::Unsupported`] error when
/// creating writer with `append` enabled.
pub struct Writer {
- state: State,
+ inner: oio::Writer,
}
impl Writer {
/// Create a new writer from an `oio::Writer`.
pub(crate) fn new(w: oio::Writer) -> Self {
- Writer {
- state: State::Idle(Some(w)),
- }
+ Writer { inner: w }
}
/// Create a new writer.
@@ -93,20 +87,14 @@ impl Writer {
pub(crate) async fn create(acc: FusedAccessor, path: &str, op: OpWrite) -> Result<Self> {
let (_, w) = acc.write(path, op).await?;
- Ok(Writer {
- state: State::Idle(Some(w)),
- })
+ Ok(Writer { inner: w })
}
/// Write into inner writer.
pub async fn write(&mut self, bs: impl Into<Bytes>) -> Result<()> {
- let State::Idle(Some(w)) = &mut self.state else {
- return Err(Error::new(ErrorKind::Unexpected, "writer must be valid"));
- };
-
let mut bs = bs.into();
while !bs.is_empty() {
- let n = w.write(bs.clone()).await?;
+ let n = self.inner.write(bs.clone()).await?;
bs.advance(n);
}
@@ -145,16 +133,12 @@ impl Writer {
S: futures::Stream<Item = Result<T>>,
T: Into<Bytes>,
{
- let State::Idle(Some(w)) = &mut self.state else {
- return Err(Error::new(ErrorKind::Unexpected, "writer must be valid"));
- };
-
let mut sink_from = pin!(sink_from);
let mut written = 0;
while let Some(bs) = sink_from.try_next().await? {
let mut bs = bs.into();
while !bs.is_empty() {
- let n = w.write(bs.clone()).await?;
+ let n = self.inner.write(bs.clone()).await?;
bs.advance(n);
written += n as u64;
}
@@ -162,45 +146,6 @@ impl Writer {
Ok(written)
}
- /// Copy into writer.
- ///
- /// copy will read data from given reader and write them into writer
- /// directly with only one constant in-memory buffer.
- ///
- /// # Notes
- ///
- /// - Copy doesn't support to be used with write concurrently.
- /// - Copy doesn't support to be used without content length now.
- ///
- /// # Examples
- ///
- /// ```no_run
- /// use bytes::Bytes;
- /// use futures::io::Cursor;
- /// use futures::stream;
- /// use futures::StreamExt;
- /// use opendal::Operator;
- /// use opendal::Result;
- ///
- /// async fn copy_example(op: Operator) -> Result<()> {
- /// let mut w = op.writer_with("path/to/file").await?;
- /// let reader = Cursor::new(vec![0; 4096]);
- /// w.copy(reader).await?;
- /// w.close().await?;
- /// Ok(())
- /// }
- /// ```
- pub async fn copy<R>(&mut self, read_from: R) -> Result<u64>
- where
- R: futures::AsyncRead,
- {
- futures::io::copy(read_from, self).await.map_err(|err| {
- Error::new(ErrorKind::Unexpected, "copy into writer failed")
- .with_operation("copy")
- .set_source(err)
- })
- }
-
/// Abort the writer and clean up all written data.
///
/// ## Notes
@@ -208,11 +153,7 @@ impl Writer {
/// Abort should only be called when the writer is not closed or
/// aborted, otherwise an unexpected error could be returned.
pub async fn abort(&mut self) -> Result<()> {
- let State::Idle(Some(w)) = &mut self.state else {
- return Err(Error::new(ErrorKind::Unexpected, "writer must be valid"));
- };
-
- w.abort().await
+ self.inner.abort().await
}
/// Close the writer and make sure all data have been committed.
@@ -222,133 +163,114 @@ impl Writer {
/// Close should only be called when the writer is not closed or
/// aborted, otherwise an unexpected error could be returned.
pub async fn close(&mut self) -> Result<()> {
- let State::Idle(Some(w)) = &mut self.state else {
- return Err(Error::new(ErrorKind::Unexpected, "writer must be valid"));
- };
+ self.inner.close().await
+ }
- w.close().await
+ /// Convert writer into [`FuturesIoAsyncWriter`] which implements [`futures::AsyncWrite`],
+ pub fn into_futures_io_async_write(self) -> FuturesIoAsyncWriter {
+ FuturesIoAsyncWriter::new(self.inner)
}
}
-enum State {
- Idle(Option<oio::Writer>),
- Writing(BoxedStaticFuture<(oio::Writer, Result<usize>)>),
- Closing(BoxedStaticFuture<(oio::Writer, Result<()>)>),
-}
+pub mod into_futures_async_write {
+ use std::io;
+ use std::pin::Pin;
+ use std::task::ready;
+ use std::task::Context;
+ use std::task::Poll;
-unsafe impl Sync for State {}
+ use bytes::Bytes;
+ use futures::AsyncWrite;
-impl futures::AsyncWrite for Writer {
- fn poll_write(
- mut self: Pin<&mut Self>,
- cx: &mut Context<'_>,
- buf: &[u8],
- ) -> Poll<io::Result<usize>> {
- match &mut self.state {
- State::Idle(w) => {
- let mut w = w.take().expect("writer must be valid");
- let bs = Bytes::copy_from_slice(buf);
- let fut = async move {
- let res = w.write(bs).await;
- (w, res)
- };
- self.state = State::Writing(Box::pin(fut));
- self.poll_write(cx, buf)
- }
- State::Writing(fut) => {
- let (w, res) = ready!(fut.as_mut().poll(cx));
- self.state = State::Idle(Some(w));
- Poll::Ready(res.map_err(format_std_io_error))
- }
- _ => Poll::Ready(Err(io::Error::new(
- io::ErrorKind::Interrupted,
- "another io operation is in progress",
- ))),
- }
+ use crate::raw::oio::Write;
+ use crate::raw::*;
+ use crate::*;
+
+ /// FuturesIoAsyncWriter is the adapter of [`AsyncWrite`] for [`Writer`].
+ ///
+ /// Users can use this adapter in cases where they need to use [`AsyncWrite`] related trait.
+ ///
+ /// FuturesIoAsyncWriter also implements [`Unpin`], [`Send`] and [`Sync`]
+ pub struct FuturesIoAsyncWriter {
+ state: State,
}
- /// Writer makes sure that every write is flushed.
- fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
- Poll::Ready(Ok(()))
+ enum State {
+ Idle(Option<oio::Writer>),
+ Writing(BoxedStaticFuture<(oio::Writer, Result<usize>)>),
+ Closing(BoxedStaticFuture<(oio::Writer, Result<()>)>),
}
- fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
- match &mut self.state {
- State::Idle(w) => {
- let mut w = w.take().expect("writer must be valid");
- let fut = async move {
- let res = w.close().await;
- (w, res)
- };
- self.state = State::Closing(Box::pin(fut));
- self.poll_close(cx)
- }
- State::Closing(fut) => {
- let (w, res) = ready!(fut.as_mut().poll(cx));
- self.state = State::Idle(Some(w));
- Poll::Ready(res.map_err(format_std_io_error))
+ /// # Safety
+ ///
+ /// FuturesReader only exposes `&mut self` to the outside world, so it's safe to be `Sync`.
+ unsafe impl Sync for State {}
+
+ impl FuturesIoAsyncWriter {
+ /// NOTE: don't allow users to create FuturesIoAsyncWriter directly.
+ #[inline]
+ pub fn new(r: oio::Writer) -> Self {
+ FuturesIoAsyncWriter {
+ state: State::Idle(Some(r)),
}
- _ => Poll::Ready(Err(io::Error::new(
- io::ErrorKind::Interrupted,
- "another io operation is in progress",
- ))),
}
}
-}
-impl tokio::io::AsyncWrite for Writer {
- fn poll_write(
- mut self: Pin<&mut Self>,
- cx: &mut Context<'_>,
- buf: &[u8],
- ) -> Poll<io::Result<usize>> {
- match &mut self.state {
- State::Idle(w) => {
- let mut w = w.take().expect("writer must be valid");
- let bs = Bytes::copy_from_slice(buf);
- let fut = async move {
- let res = w.write(bs).await;
- (w, res)
- };
- self.state = State::Writing(Box::pin(fut));
- self.poll_write(cx, buf)
+ impl AsyncWrite for FuturesIoAsyncWriter {
+ fn poll_write(
+ mut self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll<io::Result<usize>> {
+ match &mut self.state {
+ State::Idle(w) => {
+ let mut w = w.take().expect("writer must be valid");
+ let bs = Bytes::copy_from_slice(buf);
+ let fut = async move {
+ let res = w.write(bs).await;
+ (w, res)
+ };
+ self.state = State::Writing(Box::pin(fut));
+ self.poll_write(cx, buf)
+ }
+ State::Writing(fut) => {
+ let (w, res) = ready!(fut.as_mut().poll(cx));
+ self.state = State::Idle(Some(w));
+ Poll::Ready(res.map_err(format_std_io_error))
+ }
+ _ => Poll::Ready(Err(io::Error::new(
+ io::ErrorKind::Interrupted,
+ "another io operation is in progress",
+ ))),
}
- State::Writing(fut) => {
- let (w, res) = ready!(fut.as_mut().poll(cx));
- self.state = State::Idle(Some(w));
- Poll::Ready(res.map_err(format_std_io_error))
- }
- _ => Poll::Ready(Err(io::Error::new(
- io::ErrorKind::Interrupted,
- "another io operation is in progress",
- ))),
}
- }
- fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
- Poll::Ready(Ok(()))
- }
+ /// Writer makes sure that every write is flushed.
+ fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+ Poll::Ready(Ok(()))
+ }
- fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
- match &mut self.state {
- State::Idle(w) => {
- let mut w = w.take().expect("writer must be valid");
- let fut = async move {
- let res = w.close().await;
- (w, res)
- };
- self.state = State::Closing(Box::pin(fut));
- self.poll_shutdown(cx)
- }
- State::Closing(fut) => {
- let (w, res) = ready!(fut.as_mut().poll(cx));
- self.state = State::Idle(Some(w));
- Poll::Ready(res.map_err(format_std_io_error))
+ fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+ match &mut self.state {
+ State::Idle(w) => {
+ let mut w = w.take().expect("writer must be valid");
+ let fut = async move {
+ let res = w.close().await;
+ (w, res)
+ };
+ self.state = State::Closing(Box::pin(fut));
+ self.poll_close(cx)
+ }
+ State::Closing(fut) => {
+ let (w, res) = ready!(fut.as_mut().poll(cx));
+ self.state = State::Idle(Some(w));
+ Poll::Ready(res.map_err(format_std_io_error))
+ }
+ _ => Poll::Ready(Err(io::Error::new(
+ io::ErrorKind::Interrupted,
+ "another io operation is in progress",
+ ))),
}
- _ => Poll::Ready(Err(io::Error::new(
- io::ErrorKind::Interrupted,
- "another io operation is in progress",
- ))),
}
}
}
diff --git a/core/tests/behavior/async_write.rs b/core/tests/behavior/async_write.rs
index 79077b73a4..9bd71ad1ee 100644
--- a/core/tests/behavior/async_write.rs
+++ b/core/tests/behavior/async_write.rs
@@ -16,12 +16,10 @@
// under the License.
use anyhow::Result;
-use bytes::Buf;
-use bytes::Bytes;
use futures::io::BufReader;
use futures::io::Cursor;
-use futures::stream;
use futures::StreamExt;
+use futures::{stream, AsyncWriteExt};
use log::warn;
use sha2::Digest;
use sha2::Sha256;
@@ -45,8 +43,6 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
test_writer_write_with_concurrent,
test_writer_sink,
test_writer_sink_with_concurrent,
- test_writer_copy,
- test_writer_copy_with_concurrent,
test_writer_abort,
test_writer_abort_with_concurrent,
test_writer_futures_copy,
@@ -404,92 +400,6 @@ pub async fn test_writer_sink_with_concurrent(op: Operator) -> Result<()> {
Ok(())
}
-/// Reading data into writer
-pub async fn test_writer_copy(op: Operator) -> Result<()> {
- let cap = op.info().full_capability();
- if !(cap.write && cap.write_can_multi) {
- return Ok(());
- }
-
- let path = TEST_FIXTURE.new_file_path();
- let size = 5 * 1024 * 1024; // write file with 5 MiB
- let content_a = gen_fixed_bytes(size);
- let content_b = gen_fixed_bytes(size);
-
- let mut w = op.writer_with(&path).buffer(5 * 1024 * 1024).await?;
-
- let mut content = Bytes::from([content_a.clone(), content_b.clone()].concat());
- while !content.is_empty() {
- let reader = Cursor::new(content.clone());
- let n = w.copy(reader).await?;
- content.advance(n as usize);
- }
- w.close().await?;
-
- let meta = op.stat(&path).await.expect("stat must succeed");
- assert_eq!(meta.content_length(), (size * 2) as u64);
-
- let bs = op.read(&path).await?;
- assert_eq!(bs.len(), size * 2, "read size");
- assert_eq!(
- format!("{:x}", Sha256::digest(&bs[..size])),
- format!("{:x}", Sha256::digest(content_a)),
- "read content a"
- );
- assert_eq!(
- format!("{:x}", Sha256::digest(&bs[size..])),
- format!("{:x}", Sha256::digest(content_b)),
- "read content b"
- );
-
- Ok(())
-}
-
-/// Reading data into writer
-pub async fn test_writer_copy_with_concurrent(op: Operator) -> Result<()> {
- let cap = op.info().full_capability();
- if !(cap.write && cap.write_can_multi) {
- return Ok(());
- }
-
- let path = TEST_FIXTURE.new_file_path();
- let size = 5 * 1024 * 1024; // write file with 5 MiB
- let content_a = gen_fixed_bytes(size);
- let content_b = gen_fixed_bytes(size);
-
- let mut w = op
- .writer_with(&path)
- .buffer(5 * 1024 * 1024)
- .concurrent(4)
- .await?;
-
- let mut content = Bytes::from([content_a.clone(), content_b.clone()].concat());
- while !content.is_empty() {
- let reader = Cursor::new(content.clone());
- let n = w.copy(reader).await?;
- content.advance(n as usize);
- }
- w.close().await?;
-
- let meta = op.stat(&path).await.expect("stat must succeed");
- assert_eq!(meta.content_length(), (size * 2) as u64);
-
- let bs = op.read(&path).await?;
- assert_eq!(bs.len(), size * 2, "read size");
- assert_eq!(
- format!("{:x}", Sha256::digest(&bs[..size])),
- format!("{:x}", Sha256::digest(content_a)),
- "read content a"
- );
- assert_eq!(
- format!("{:x}", Sha256::digest(&bs[size..])),
- format!("{:x}", Sha256::digest(content_b)),
- "read content b"
- );
-
- Ok(())
-}
-
/// Copy data from reader to writer
pub async fn test_writer_futures_copy(op: Operator) -> Result<()> {
if !(op.info().full_capability().write_can_multi) {
@@ -500,7 +410,11 @@ pub async fn test_writer_futures_copy(op: Operator) -> Result<()> {
let (content, size): (Vec<u8>, usize) =
gen_bytes_with_range(10 * 1024 * 1024..20 * 1024 * 1024);
- let mut w = op.writer_with(&path).buffer(8 * 1024 * 1024).await?;
+ let mut w = op
+ .writer_with(&path)
+ .buffer(8 * 1024 * 1024)
+ .await?
+ .into_futures_io_async_write();
// Wrap a buf reader here to make sure content is read in 1MiB chunks.
let mut cursor = BufReader::with_capacity(1024 * 1024, Cursor::new(content.clone()));
@@ -535,12 +449,13 @@ pub async fn test_writer_futures_copy_with_concurrent(op: Operator) -> Result<()
.writer_with(&path)
.buffer(8 * 1024 * 1024)
.concurrent(4)
- .await?;
+ .await?
+ .into_futures_io_async_write();
// Wrap a buf reader here to make sure content is read in 1MiB chunks.
let mut cursor = BufReader::with_capacity(1024 * 1024, Cursor::new(content.clone()));
futures::io::copy_buf(&mut cursor, &mut w).await?;
- w.close().await?;
+ w.close().await.expect("close must succeed");
let meta = op.stat(&path).await.expect("stat must succeed");
assert_eq!(meta.content_length(), size as u64);
@@ -590,7 +505,11 @@ pub async fn test_writer_with_append(op: Operator) -> Result<()> {
let (content, size): (Vec<u8>, usize) =
gen_bytes_with_range(10 * 1024 * 1024..20 * 1024 * 1024);
- let mut a = op.writer_with(&path).append(true).await?;
+ let mut a = op
+ .writer_with(&path)
+ .append(true)
+ .await?
+ .into_futures_io_async_write();
// Wrap a buf reader here to make sure content is read in 1MiB chunks.
let mut cursor = BufReader::with_capacity(1024 * 1024, Cursor::new(content.clone()));