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 2026/06/18 08:49:28 UTC
(opendal) branch main updated: refactor(core): lazy open positioned read handles (#7796)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new b94d02e03 refactor(core): lazy open positioned read handles (#7796)
b94d02e03 is described below
commit b94d02e03b994963ef9cf9f7993ef549644435e4
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Jun 18 16:49:22 2026 +0800
refactor(core): lazy open positioned read handles (#7796)
* Refine positioned read handle initialization
* Simplify positioned read handle initialization
---
core/core/src/raw/oio/read/position_read.rs | 87 ++++++++++++++++++++++-------
core/services/compfs/src/backend.rs | 85 ++++++++++++----------------
core/services/fs/Cargo.toml | 2 +-
core/services/fs/src/backend.rs | 77 ++++++++++---------------
core/services/hdfs-native/src/backend.rs | 65 ++++++++++-----------
core/services/hdfs/Cargo.toml | 2 +-
core/services/hdfs/src/backend.rs | 64 +++++++++++++--------
core/services/hdfs/src/core.rs | 17 +-----
core/services/hdfs/src/lib.rs | 1 -
core/services/hdfs/src/reader.rs | 79 --------------------------
core/services/monoiofs/src/backend.rs | 28 +++++-----
core/services/monoiofs/src/reader.rs | 4 +-
12 files changed, 227 insertions(+), 284 deletions(-)
diff --git a/core/core/src/raw/oio/read/position_read.rs b/core/core/src/raw/oio/read/position_read.rs
index 01196da82..e602689f2 100644
--- a/core/core/src/raw/oio/read/position_read.rs
+++ b/core/core/src/raw/oio/read/position_read.rs
@@ -18,6 +18,7 @@
use std::sync::Arc;
use futures::Future;
+use mea::once::OnceCell;
use crate::raw::*;
use crate::*;
@@ -26,26 +27,38 @@ const DEFAULT_POSITION_READ_MAX_BUF_SIZE: usize = 2 * 1024 * 1024;
/// PositionRead is used to implement [`oio::Read`] based on positioned reads.
///
-/// Services that implement [`PositionRead`] must support position-independent
-/// reads. `size` is the maximum number of bytes to read, and implementations may
-/// return fewer bytes. Returning an empty buffer means EOF.
+/// Services that implement [`PositionRead`] create a positioned read handle lazily
+/// and must support position-independent reads on that handle. `size` is the
+/// maximum number of bytes to read, and implementations may return fewer bytes.
+/// Returning an empty buffer means EOF.
pub trait PositionRead: Send + Sync + Unpin + 'static {
+ /// The opened positioned read handle.
+ type Handle: Send + Sync + Unpin + 'static;
+
+ /// Open the positioned read handle.
+ fn open(&self) -> impl Future<Output = Result<Self::Handle>> + MaybeSend;
+
/// Read up to `size` bytes from `offset`.
- fn read_at(&self, offset: u64, size: usize)
- -> impl Future<Output = Result<Buffer>> + MaybeSend;
+ fn read_at(
+ handle: &Self::Handle,
+ offset: u64,
+ size: usize,
+ ) -> impl Future<Output = Result<Buffer>> + MaybeSend;
}
/// PositionReader implements [`oio::Read`] based on [`PositionRead`].
pub struct PositionReader<R: PositionRead> {
- inner: Arc<R>,
+ reader: Arc<R>,
+ handle: Arc<OnceCell<R::Handle>>,
max_buf_size: usize,
}
impl<R: PositionRead> PositionReader<R> {
/// Create a new [`PositionReader`].
- pub fn new(inner: R) -> Self {
+ pub fn new(reader: R) -> Self {
Self {
- inner: Arc::new(inner),
+ reader: Arc::new(reader),
+ handle: Arc::new(OnceCell::new()),
max_buf_size: DEFAULT_POSITION_READ_MAX_BUF_SIZE,
}
}
@@ -67,13 +80,22 @@ impl<R: PositionRead> PositionReader<R> {
///
/// Panics if there are active streams that still share the inner reader.
pub fn into_inner(self) -> R {
- Arc::into_inner(self.inner).expect("position reader must not be shared")
+ Arc::into_inner(self.reader).expect("position reader must not be shared")
+ }
+
+ async fn handle(&self) -> Result<&R::Handle> {
+ self.handle.get_or_try_init(|| self.reader.open()).await
}
}
impl<R: PositionRead> oio::Read for PositionReader<R> {
async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
- let stream = PositionReadStream::new(self.inner.clone(), range, self.max_buf_size);
+ let stream = PositionReadStream::new(
+ self.reader.clone(),
+ self.handle.clone(),
+ range,
+ self.max_buf_size,
+ );
Ok((
RpRead::default(),
Box::new(stream) as Box<dyn oio::ReadStreamDyn>,
@@ -91,7 +113,8 @@ impl<R: PositionRead> oio::Read for PositionReader<R> {
while remaining > 0 {
let read_size = remaining.min(self.max_buf_size as u64) as usize;
- let buf = self.inner.read_at(offset, read_size).await?;
+ let handle = self.handle().await?;
+ let buf = R::read_at(handle, offset, read_size).await?;
check_position_read_size(read_size, buf.len())?;
if buf.is_empty() {
return Err(Error::new(
@@ -113,7 +136,8 @@ impl<R: PositionRead> oio::Read for PositionReader<R> {
}
struct PositionReadStream<R: PositionRead> {
- inner: Arc<R>,
+ reader: Arc<R>,
+ handle: Arc<OnceCell<R::Handle>>,
offset: u64,
remaining: Option<u64>,
max_buf_size: usize,
@@ -121,9 +145,15 @@ struct PositionReadStream<R: PositionRead> {
}
impl<R: PositionRead> PositionReadStream<R> {
- fn new(inner: Arc<R>, range: BytesRange, max_buf_size: usize) -> Self {
+ fn new(
+ reader: Arc<R>,
+ handle: Arc<OnceCell<R::Handle>>,
+ range: BytesRange,
+ max_buf_size: usize,
+ ) -> Self {
Self {
- inner,
+ reader,
+ handle,
offset: range.offset(),
remaining: range.size(),
max_buf_size,
@@ -143,7 +173,8 @@ impl<R: PositionRead> oio::ReadStream for PositionReadStream<R> {
.map(|remaining| remaining.min(self.max_buf_size as u64) as usize)
.unwrap_or(self.max_buf_size);
- let buf = self.inner.read_at(self.offset, read_size).await?;
+ let handle = self.handle.get_or_try_init(|| self.reader.open()).await?;
+ let buf = R::read_at(handle, self.offset, read_size).await?;
check_position_read_size(read_size, buf.len())?;
if buf.is_empty() {
self.done = true;
@@ -195,6 +226,7 @@ mod tests {
content: Bytes,
max_read: usize,
calls: Arc<Mutex<Vec<(u64, usize)>>>,
+ opens: Arc<Mutex<usize>>,
}
impl TestPositionRead {
@@ -203,21 +235,34 @@ mod tests {
content: Bytes::from_static(content),
max_read,
calls: Arc::default(),
+ opens: Arc::default(),
}
}
}
impl PositionRead for TestPositionRead {
- async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
- self.calls.lock().unwrap().push((offset, size));
+ type Handle = Self;
+
+ async fn open(&self) -> Result<Self::Handle> {
+ *self.opens.lock().unwrap() += 1;
+ Ok(Self {
+ content: self.content.clone(),
+ max_read: self.max_read,
+ calls: self.calls.clone(),
+ opens: self.opens.clone(),
+ })
+ }
+
+ async fn read_at(handle: &Self::Handle, offset: u64, size: usize) -> Result<Buffer> {
+ handle.calls.lock().unwrap().push((offset, size));
let offset = offset as usize;
- if offset >= self.content.len() {
+ if offset >= handle.content.len() {
return Ok(Buffer::new());
}
- let end = offset + size.min(self.max_read).min(self.content.len() - offset);
- Ok(Buffer::from(self.content.slice(offset..end)))
+ let end = offset + size.min(handle.max_read).min(handle.content.len() - offset);
+ Ok(Buffer::from(handle.content.slice(offset..end)))
}
}
@@ -225,12 +270,14 @@ mod tests {
async fn test_position_reader_read_handles_partial_reads() -> Result<()> {
let inner = TestPositionRead::new(b"0123456789", 2);
let calls = inner.calls.clone();
+ let opens = inner.opens.clone();
let reader = PositionReader::new(inner).with_max_buf_size(4);
let (_, buf) = reader.read(BytesRange::from(2..8)).await?;
assert_eq!(buf.to_vec(), b"234567");
assert_eq!(calls.lock().unwrap().as_slice(), &[(2, 4), (4, 4), (6, 2)]);
+ assert_eq!(*opens.lock().unwrap(), 1);
Ok(())
}
diff --git a/core/services/compfs/src/backend.rs b/core/services/compfs/src/backend.rs
index d0f70b41a..ea0f4c842 100644
--- a/core/services/compfs/src/backend.rs
+++ b/core/services/compfs/src/backend.rs
@@ -125,51 +125,30 @@ pub struct CompfsBackend {
/// Reader returned by this backend.
pub struct CompfsReader {
core: Arc<CompfsCore>,
- file: compio::fs::File,
+ path: PathBuf,
}
impl CompfsReader {
- fn new(core: Arc<CompfsCore>, file: compio::fs::File) -> Self {
- Self { core, file }
- }
-}
-
-impl oio::PositionRead for CompfsReader {
- async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
- if size == 0 {
- return Ok(Buffer::new());
- }
-
- let mut bs = self.core.buf_pool.get();
- bs.reserve(size);
-
- let file = self.file.clone();
- let (n, mut bs) = self
- .core
- .exec(move || async move {
- let (n, bs) = buf_try!(@try file.read_at(bs.slice(..size), offset).await);
- Ok((n, bs.into_inner()))
- })
- .await?;
-
- let frozen = bs.split_to(n).freeze();
- self.core.buf_pool.put(bs);
-
- Ok(CompfsBuffer::from(Buffer::from(frozen)).into())
+ fn new(core: Arc<CompfsCore>, path: PathBuf) -> Self {
+ Self { core, path }
}
}
-pub struct CompfsLazyReader {
+pub struct CompfsReaderHandle {
core: Arc<CompfsCore>,
- path: PathBuf,
+ file: compio::fs::File,
}
-impl CompfsLazyReader {
- fn new(core: Arc<CompfsCore>, path: PathBuf) -> Self {
- Self { core, path }
+impl CompfsReaderHandle {
+ fn new(core: Arc<CompfsCore>, file: compio::fs::File) -> Self {
+ Self { core, file }
}
+}
- async fn reader(&self) -> Result<oio::PositionReader<CompfsReader>> {
+impl oio::PositionRead for CompfsReader {
+ type Handle = CompfsReaderHandle;
+
+ async fn open(&self) -> Result<Self::Handle> {
let path = self.path.clone();
let file = self
.core
@@ -182,20 +161,30 @@ impl CompfsLazyReader {
})
.await?;
- Ok(oio::PositionReader::new(CompfsReader::new(
- self.core.clone(),
- file,
- )))
+ Ok(CompfsReaderHandle::new(self.core.clone(), file))
}
-}
-impl oio::Read for CompfsLazyReader {
- async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
- self.reader().await?.open(range).await
- }
+ async fn read_at(handle: &Self::Handle, offset: u64, size: usize) -> Result<Buffer> {
+ if size == 0 {
+ return Ok(Buffer::new());
+ }
- async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
- self.reader().await?.read(range).await
+ let mut bs = handle.core.buf_pool.get();
+ bs.reserve(size);
+
+ let file = handle.file.clone();
+ let (n, mut bs) = handle
+ .core
+ .exec(move || async move {
+ let (n, bs) = buf_try!(@try file.read_at(bs.slice(..size), offset).await);
+ Ok((n, bs.into_inner()))
+ })
+ .await?;
+
+ let frozen = bs.split_to(n).freeze();
+ handle.core.buf_pool.put(bs);
+
+ Ok(CompfsBuffer::from(Buffer::from(frozen)).into())
}
}
@@ -306,7 +295,7 @@ impl oio::List for CompfsLazyLister {
}
impl Service for CompfsBackend {
- type Reader = CompfsLazyReader;
+ type Reader = oio::PositionReader<CompfsReader>;
type Writer = CompfsLazyWriter;
type Lister = CompfsLazyLister;
type Deleter = oio::OneShotDeleter<CompfsDeleter>;
@@ -422,10 +411,10 @@ impl Service for CompfsBackend {
Ok(RpRename::default())
}
fn read(&self, _ctx: &OperationContext, path: &str, _: OpRead) -> Result<Self::Reader> {
- Ok(CompfsLazyReader::new(
+ Ok(oio::PositionReader::new(CompfsReader::new(
self.core.clone(),
self.core.prepare_path(path)?,
- ))
+ )))
}
fn write(&self, _ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
diff --git a/core/services/fs/Cargo.toml b/core/services/fs/Cargo.toml
index a6e69d27d..6e66f250d 100644
--- a/core/services/fs/Cargo.toml
+++ b/core/services/fs/Cargo.toml
@@ -37,7 +37,7 @@ opendal-core = { path = "../../core", version = "0.57.0", default-features = fal
"internal-tokio-rt",
] }
serde = { workspace = true, features = ["derive"] }
-tokio = { workspace = true, features = ["fs", "rt-multi-thread", "sync"] }
+tokio = { workspace = true, features = ["fs", "rt-multi-thread"] }
[target.'cfg(unix)'.dependencies]
xattr = "1"
diff --git a/core/services/fs/src/backend.rs b/core/services/fs/src/backend.rs
index 322537198..813422baa 100644
--- a/core/services/fs/src/backend.rs
+++ b/core/services/fs/src/backend.rs
@@ -180,10 +180,24 @@ pub struct FsBackend {
/// Reader returned by this backend.
pub struct FsReader {
core: Arc<FsCore>,
- file: Arc<File>,
+ path: String,
}
impl FsReader {
+ fn new(core: Arc<FsCore>, path: &str) -> Self {
+ Self {
+ core,
+ path: path.to_string(),
+ }
+ }
+}
+
+pub struct FsReaderHandle {
+ core: Arc<FsCore>,
+ file: Arc<File>,
+}
+
+impl FsReaderHandle {
fn new(core: Arc<FsCore>, file: File) -> Self {
Self {
core,
@@ -193,15 +207,22 @@ impl FsReader {
}
impl oio::PositionRead for FsReader {
- async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+ type Handle = FsReaderHandle;
+
+ async fn open(&self) -> Result<Self::Handle> {
+ let file = self.core.fs_open(&self.path).await?;
+ Ok(FsReaderHandle::new(self.core.clone(), file))
+ }
+
+ async fn read_at(handle: &Self::Handle, offset: u64, size: usize) -> Result<Buffer> {
if size == 0 {
return Ok(Buffer::new());
}
- let mut bs = self.core.buf_pool.get();
+ let mut bs = handle.core.buf_pool.get();
bs.resize(size, 0);
- let f = self.file.clone();
+ let f = handle.file.clone();
let (n, mut bs) = tokio::task::spawn_blocking(move || {
let n = read_at(&f, &mut bs, offset)?;
Ok::<_, Error>((n, bs))
@@ -210,51 +231,12 @@ impl oio::PositionRead for FsReader {
.map_err(new_task_join_error)??;
let frozen = bs.split_to(n).freeze();
- self.core.buf_pool.put(bs);
+ handle.core.buf_pool.put(bs);
Ok(Buffer::from(frozen))
}
}
-pub struct FsLazyReader {
- core: Arc<FsCore>,
- path: String,
- // Open the file once and reuse the same handle across every read operation
- reader: tokio::sync::OnceCell<oio::PositionReader<FsReader>>,
-}
-
-impl FsLazyReader {
- fn new(core: Arc<FsCore>, path: &str) -> Self {
- Self {
- core,
- path: path.to_string(),
- reader: tokio::sync::OnceCell::new(),
- }
- }
-
- async fn reader(&self) -> Result<&oio::PositionReader<FsReader>> {
- self.reader
- .get_or_try_init(|| async {
- let file = self.core.fs_open(&self.path).await?;
- Ok(oio::PositionReader::new(FsReader::new(
- self.core.clone(),
- file,
- )))
- })
- .await
- }
-}
-
-impl oio::Read for FsLazyReader {
- async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
- self.reader().await?.open(range).await
- }
-
- async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
- self.reader().await?.read(range).await
- }
-}
-
pub struct FsLazyWriter {
core: Arc<FsCore>,
executor: Executor,
@@ -343,7 +325,7 @@ impl oio::List for FsLazyLister {
}
impl Service for FsBackend {
- type Reader = FsLazyReader;
+ type Reader = oio::PositionReader<FsReader>;
type Writer = FsLazyWriter;
type Lister = FsLazyLister;
type Deleter = oio::OneShotDeleter<FsDeleter>;
@@ -373,7 +355,10 @@ impl Service for FsBackend {
}
fn read(&self, _ctx: &OperationContext, path: &str, _: OpRead) -> Result<Self::Reader> {
- Ok(FsLazyReader::new(self.core.clone(), path))
+ Ok(oio::PositionReader::new(FsReader::new(
+ self.core.clone(),
+ path,
+ )))
}
fn write(&self, ctx: &OperationContext, path: &str, op: OpWrite) -> Result<Self::Writer> {
diff --git a/core/services/hdfs-native/src/backend.rs b/core/services/hdfs-native/src/backend.rs
index 3c5249035..a7090a922 100644
--- a/core/services/hdfs-native/src/backend.rs
+++ b/core/services/hdfs-native/src/backend.rs
@@ -163,17 +163,38 @@ pub struct HdfsNativeBackend {
/// Reader returned by this backend.
pub struct HdfsNativeReader {
- file: hdfs_native::file::FileReader,
+ core: Arc<HdfsNativeCore>,
+ path: String,
}
impl HdfsNativeReader {
+ fn new(core: Arc<HdfsNativeCore>, path: &str) -> Self {
+ Self {
+ core,
+ path: path.to_string(),
+ }
+ }
+}
+
+pub struct HdfsNativeReaderHandle {
+ file: hdfs_native::file::FileReader,
+}
+
+impl HdfsNativeReaderHandle {
fn new(file: hdfs_native::file::FileReader) -> Self {
Self { file }
}
}
impl oio::PositionRead for HdfsNativeReader {
- async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+ type Handle = HdfsNativeReaderHandle;
+
+ async fn open(&self) -> Result<Self::Handle> {
+ let file = self.core.hdfs_open(&self.path).await?;
+ Ok(HdfsNativeReaderHandle::new(file))
+ }
+
+ async fn read_at(handle: &Self::Handle, offset: u64, size: usize) -> Result<Buffer> {
if size == 0 {
return Ok(Buffer::new());
}
@@ -182,13 +203,13 @@ impl oio::PositionRead for HdfsNativeReader {
return Ok(Buffer::new());
};
- let file_length = self.file.file_length();
+ let file_length = handle.file.file_length();
if offset >= file_length {
return Ok(Buffer::new());
}
let size = size.min(file_length - offset);
- let bytes = self
+ let bytes = handle
.file
.read_range(offset, size)
.await
@@ -198,35 +219,6 @@ impl oio::PositionRead for HdfsNativeReader {
}
}
-pub struct HdfsNativeLazyReader {
- core: Arc<HdfsNativeCore>,
- path: String,
-}
-
-impl HdfsNativeLazyReader {
- fn new(core: Arc<HdfsNativeCore>, path: &str) -> Self {
- Self {
- core,
- path: path.to_string(),
- }
- }
-
- async fn reader(&self) -> Result<oio::PositionReader<HdfsNativeReader>> {
- let file = self.core.hdfs_open(&self.path).await?;
- Ok(oio::PositionReader::new(HdfsNativeReader::new(file)))
- }
-}
-
-impl oio::Read for HdfsNativeLazyReader {
- async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
- self.reader().await?.open(range).await
- }
-
- async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
- self.reader().await?.read(range).await
- }
-}
-
pub struct HdfsNativeLazyWriter {
core: Arc<HdfsNativeCore>,
path: String,
@@ -306,7 +298,7 @@ impl oio::List for HdfsNativeLazyLister {
}
impl Service for HdfsNativeBackend {
- type Reader = HdfsNativeLazyReader;
+ type Reader = oio::PositionReader<HdfsNativeReader>;
type Writer = HdfsNativeLazyWriter;
type Lister = HdfsNativeLazyLister;
type Deleter = oio::OneShotDeleter<HdfsNativeDeleter>;
@@ -335,7 +327,10 @@ impl Service for HdfsNativeBackend {
Ok(RpStat::new(m))
}
fn read(&self, _ctx: &OperationContext, path: &str, _: OpRead) -> Result<Self::Reader> {
- Ok(HdfsNativeLazyReader::new(self.core.clone(), path))
+ Ok(oio::PositionReader::new(HdfsNativeReader::new(
+ self.core.clone(),
+ path,
+ )))
}
fn write(&self, _ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
diff --git a/core/services/hdfs/Cargo.toml b/core/services/hdfs/Cargo.toml
index b559f1b73..fe1b2bb30 100644
--- a/core/services/hdfs/Cargo.toml
+++ b/core/services/hdfs/Cargo.toml
@@ -38,4 +38,4 @@ futures = { workspace = true }
hdrs = { version = "0.3.2", features = ["async_file"] }
log = { workspace = true }
serde = { workspace = true, features = ["derive"] }
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["rt"] }
diff --git a/core/services/hdfs/src/backend.rs b/core/services/hdfs/src/backend.rs
index d5b9fbbb2..4191f7047 100644
--- a/core/services/hdfs/src/backend.rs
+++ b/core/services/hdfs/src/backend.rs
@@ -25,7 +25,6 @@ use super::config::HdfsConfig;
use super::core::HdfsCore;
use super::deleter::HdfsDeleter;
use super::lister::HdfsLister;
-use super::reader::HdfsReadStream;
use super::writer::HdfsWriter;
use opendal_core::raw::*;
use opendal_core::*;
@@ -193,29 +192,53 @@ pub struct HdfsBackend {
/// Reader returned by this backend.
pub struct HdfsReader {
- backend: HdfsBackend,
+ core: Arc<HdfsCore>,
path: String,
}
impl HdfsReader {
- fn new(backend: HdfsBackend, path: &str, _: OpRead) -> Self {
+ fn new(core: Arc<HdfsCore>, path: &str) -> Self {
Self {
- backend,
+ core,
path: path.to_string(),
}
}
}
-impl oio::StreamRead for HdfsReader {
- async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
- let backend = &self.backend;
- let path = self.path.as_str();
+pub struct HdfsReaderHandle {
+ file: Arc<hdrs::File>,
+}
+
+impl HdfsReaderHandle {
+ fn new(file: hdrs::File) -> Self {
+ Self { file: file.into() }
+ }
+}
- let f = backend.core.hdfs_read(path, range).await?;
- let rp = RpRead::default();
- let stream = HdfsReadStream::new(f, range.size().unwrap_or(u64::MAX) as _);
+impl oio::PositionRead for HdfsReader {
+ type Handle = HdfsReaderHandle;
- Ok((rp, Box::new(stream) as Box<dyn oio::ReadStreamDyn>))
+ async fn open(&self) -> Result<Self::Handle> {
+ let file = self.core.hdfs_open(&self.path).await?;
+ Ok(HdfsReaderHandle::new(file))
+ }
+
+ async fn read_at(handle: &Self::Handle, offset: u64, size: usize) -> Result<Buffer> {
+ if size == 0 {
+ return Ok(Buffer::new());
+ }
+
+ let file = handle.file.clone();
+ let buf = tokio::task::spawn_blocking(move || {
+ let mut buf = vec![0; size];
+ let n = file.read_at(&mut buf, offset).map_err(new_std_io_error)?;
+ buf.truncate(n);
+ Ok::<_, Error>(Buffer::from(buf))
+ })
+ .await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, "tokio task join failed").set_source(e))??;
+
+ Ok(buf)
}
}
@@ -270,7 +293,7 @@ impl oio::Write for HdfsLazyWriter {
}
impl Service for HdfsBackend {
- type Reader = oio::StreamReader<HdfsReader>;
+ type Reader = oio::PositionReader<HdfsReader>;
type Writer = HdfsLazyWriter;
type Lister = Option<HdfsLister>;
type Deleter = oio::OneShotDeleter<HdfsDeleter>;
@@ -298,16 +321,11 @@ impl Service for HdfsBackend {
let m = self.core.hdfs_stat(path)?;
Ok(RpStat::new(m))
}
- fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
- let output: oio::StreamReader<HdfsReader> = {
- Ok(oio::StreamReader::new(HdfsReader::new(
- self.clone(),
- path,
- args,
- )))
- }?;
-
- Ok(output)
+ fn read(&self, _ctx: &OperationContext, path: &str, _: OpRead) -> Result<Self::Reader> {
+ Ok(oio::PositionReader::new(HdfsReader::new(
+ self.core.clone(),
+ path,
+ )))
}
fn write(&self, _ctx: &OperationContext, path: &str, op: OpWrite) -> Result<Self::Writer> {
diff --git a/core/services/hdfs/src/core.rs b/core/services/hdfs/src/core.rs
index 1735b11bd..0d8767dca 100644
--- a/core/services/hdfs/src/core.rs
+++ b/core/services/hdfs/src/core.rs
@@ -17,7 +17,6 @@
use std::fmt::Debug;
use std::io;
-use std::io::SeekFrom;
use std::sync::Arc;
use opendal_core::raw::*;
@@ -67,25 +66,15 @@ impl HdfsCore {
Ok(m)
}
- pub async fn hdfs_read(&self, path: &str, range: BytesRange) -> Result<hdrs::AsyncFile> {
+ pub async fn hdfs_open(&self, path: &str) -> Result<hdrs::File> {
let p = build_rooted_abs_path(&self.root, path);
let client = self.client.clone();
- let mut f = client
- .open_file()
- .read(true)
- .async_open(&p)
+ let f = tokio::task::spawn_blocking(move || client.open_file().read(true).open(&p))
.await
+ .map_err(|e| Error::new(ErrorKind::Unexpected, "tokio task join failed").set_source(e))?
.map_err(new_std_io_error)?;
- if range.offset() != 0 {
- use futures::AsyncSeekExt;
-
- f.seek(SeekFrom::Start(range.offset()))
- .await
- .map_err(new_std_io_error)?;
- }
-
Ok(f)
}
diff --git a/core/services/hdfs/src/lib.rs b/core/services/hdfs/src/lib.rs
index c24c35538..bb070846b 100644
--- a/core/services/hdfs/src/lib.rs
+++ b/core/services/hdfs/src/lib.rs
@@ -28,7 +28,6 @@ mod config;
mod core;
mod deleter;
mod lister;
-mod reader;
mod writer;
pub use backend::HdfsBuilder as Hdfs;
diff --git a/core/services/hdfs/src/reader.rs b/core/services/hdfs/src/reader.rs
deleted file mode 100644
index 54bed2d0f..000000000
--- a/core/services/hdfs/src/reader.rs
+++ /dev/null
@@ -1,79 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-use bytes::BytesMut;
-use futures::AsyncReadExt;
-use hdrs::AsyncFile;
-use tokio::io::ReadBuf;
-
-use opendal_core::raw::*;
-use opendal_core::*;
-
-pub struct HdfsReadStream<F> {
- f: F,
- read: usize,
- size: usize,
- buf_size: usize,
- buf: BytesMut,
-}
-
-impl<F> HdfsReadStream<F> {
- pub fn new(f: F, size: usize) -> Self {
- Self {
- f,
- read: 0,
- size,
- // Use 2 MiB as default value.
- buf_size: 2 * 1024 * 1024,
- buf: BytesMut::new(),
- }
- }
-}
-
-impl oio::ReadStream for HdfsReadStream<AsyncFile> {
- async fn read(&mut self) -> Result<Buffer> {
- if self.read >= self.size {
- return Ok(Buffer::new());
- }
-
- let size = (self.size - self.read).min(self.buf_size);
- self.buf.reserve(size);
-
- let buf = &mut self.buf.spare_capacity_mut()[..size];
- let mut read_buf: ReadBuf = ReadBuf::uninit(buf);
-
- // SAFETY: Read at most `limit` bytes into `read_buf`.
- unsafe {
- read_buf.assume_init(size);
- }
-
- let n = self
- .f
- .read(read_buf.initialize_unfilled())
- .await
- .map_err(new_std_io_error)?;
- read_buf.advance(n);
- self.read += n;
-
- // Safety: We make sure that bs contains `n` more bytes.
- let filled = read_buf.filled().len();
- unsafe { self.buf.set_len(filled) }
-
- let frozen = self.buf.split().freeze();
- Ok(Buffer::from(frozen))
- }
-}
diff --git a/core/services/monoiofs/src/backend.rs b/core/services/monoiofs/src/backend.rs
index 14a9ddebb..127ac750d 100644
--- a/core/services/monoiofs/src/backend.rs
+++ b/core/services/monoiofs/src/backend.rs
@@ -92,29 +92,26 @@ pub struct MonoiofsBackend {
core: Arc<MonoiofsCore>,
}
-pub struct MonoiofsLazyReader {
+pub struct MonoiofsPositionReader {
core: Arc<MonoiofsCore>,
path: PathBuf,
}
-impl MonoiofsLazyReader {
+impl MonoiofsPositionReader {
fn new(core: Arc<MonoiofsCore>, path: PathBuf) -> Self {
Self { core, path }
}
-
- async fn reader(&self) -> Result<oio::PositionReader<MonoiofsReader>> {
- let reader = MonoiofsReader::new(self.core.clone(), self.path.clone()).await?;
- Ok(oio::PositionReader::new(reader))
- }
}
-impl oio::Read for MonoiofsLazyReader {
- async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
- self.reader().await?.open(range).await
+impl oio::PositionRead for MonoiofsPositionReader {
+ type Handle = MonoiofsReader;
+
+ async fn open(&self) -> Result<Self::Handle> {
+ MonoiofsReader::new(self.core.clone(), self.path.clone()).await
}
- async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
- self.reader().await?.read(range).await
+ async fn read_at(handle: &Self::Handle, offset: u64, size: usize) -> Result<Buffer> {
+ handle.read_at(offset, size).await
}
}
@@ -170,7 +167,7 @@ impl oio::Write for MonoiofsLazyWriter {
}
impl Service for MonoiofsBackend {
- type Reader = MonoiofsLazyReader;
+ type Reader = oio::PositionReader<MonoiofsPositionReader>;
type Writer = MonoiofsLazyWriter;
type Lister = ();
type Deleter = oio::OneShotDeleter<MonoiofsDeleter>;
@@ -207,7 +204,10 @@ impl Service for MonoiofsBackend {
}
fn read(&self, _ctx: &OperationContext, path: &str, _args: OpRead) -> Result<Self::Reader> {
let path = self.core.prepare_path(path)?;
- Ok(MonoiofsLazyReader::new(self.core.clone(), path))
+ Ok(oio::PositionReader::new(MonoiofsPositionReader::new(
+ self.core.clone(),
+ path,
+ )))
}
fn write(&self, _ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
diff --git a/core/services/monoiofs/src/reader.rs b/core/services/monoiofs/src/reader.rs
index 055a407c9..805b7280c 100644
--- a/core/services/monoiofs/src/reader.rs
+++ b/core/services/monoiofs/src/reader.rs
@@ -94,11 +94,11 @@ impl MonoiofsReader {
}
}
-impl oio::PositionRead for MonoiofsReader {
+impl MonoiofsReader {
/// Send read request to worker thread and wait for result. Actual
/// read happens in [`MonoiofsReader::worker_entrypoint`] running
/// on worker thread.
- async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
+ pub async fn read_at(&self, offset: u64, size: usize) -> Result<Buffer> {
if size == 0 {
return Ok(Buffer::new());
}