You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by er...@apache.org on 2026/07/05 03:40:41 UTC
(opendal) 01/01: fix: reuse sftp read handles
This is an automated email from the ASF dual-hosted git repository.
erickguan pushed a commit to branch fix-reader
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 380729d923cf12e5ba93c8dcfd5feacdd0580e8b
Author: Erick Guan <29...@users.noreply.github.com>
AuthorDate: Sun Jul 5 11:40:14 2026 +0800
fix: reuse sftp read handles
---
.../docs/rfcs/7660_move_read_range_to_reader.md | 101 +++++----
core/core/src/types/context/read.rs | 2 +-
core/core/src/types/read/reader.rs | 236 +++++++++++++++++++++
core/services/sftp/src/backend.rs | 16 +-
core/services/sftp/src/reader.rs | 99 +++------
5 files changed, 333 insertions(+), 121 deletions(-)
diff --git a/core/core/src/docs/rfcs/7660_move_read_range_to_reader.md b/core/core/src/docs/rfcs/7660_move_read_range_to_reader.md
index 623906f60..1881e1780 100644
--- a/core/core/src/docs/rfcs/7660_move_read_range_to_reader.md
+++ b/core/core/src/docs/rfcs/7660_move_read_range_to_reader.md
@@ -6,19 +6,19 @@
# Summary
Change OpenDAL's raw read contract so byte ranges are selected by the raw reader
-instead of `Access::read`.
+instead of `Service::read`.
-Today `Access::read(path, OpRead { range, .. })` opens one concrete range
+Today `Service::read(path, OpRead { range, .. })` opens one concrete range
stream. This works well for HTTP object stores, where each range read is an
independent request. It is a poor fit for services such as `fs`, `sftp`,
`hdfs`, `hdfs-native`, and `monoiofs`, where the service may keep an opened
handle and execute multiple cursor-independent range reads against it. The
-current core reader planner still calls `Access::read` for every planned range,
+current core reader planner still calls `Service::read` for every planned range,
forcing those services to repeatedly open and close handles.
-This RFC keeps the `Access::read` operation, but changes its contract:
+This RFC keeps the `Service::read` operation, but changes its contract:
-- `Access::read(path, OpRead)` creates a reusable raw `oio::Read` for the path
+- `Service::read(path, OpRead)` creates a reusable raw `oio::Read` for the path
and read condition set.
- `oio::Read::open(range)` opens a range-scoped byte stream.
- `oio::Read::read(range)` reads one bounded planner range into one `Buffer`.
@@ -41,12 +41,12 @@ let parts = reader.fetch(vec![0..1024, 4096..8192]).await?;
However, the current implementation plans these public calls by generating new
raw range streams. `ReadGenerator::next_reader` and `ChunkedReader` both call
-`Access::read(path, args.with_range(range))` for every range. For `s3`, `gcs`,
+`Service::read(path, args.with_range(range))` for every range. For `s3`, `gcs`,
and `azblob`, this is natural because each range is a new HTTP request. For
handle based services, this means every range repeats open and seek work even
when the target data can be represented by a reusable handle.
-The root cause is that `Access::read` is modeled as a transport-level range
+The root cause is that `Service::read` is modeled as a transport-level range
request. OpenDAL needs a storage-neutral raw model:
- Object stores can implement a raw reader that issues an independent ranged
@@ -84,7 +84,7 @@ let ranges = reader.fetch(vec![0..1024, 4096..8192]).await?;
The difference is internal. OpenDAL creates one reusable raw reader for
`reader("path/to/file")`. After that, range operations are executed by this raw
-reader instead of reopening through `Access::read` for every range.
+reader instead of reopening through `Service::read` for every range.
For HTTP object stores, the raw reader stores the path and read arguments, then
sends a new ranged request for each range. Behavior and cost stay the same.
@@ -98,17 +98,18 @@ a shared seek cursor. It must use cursor-independent reads such as `pread` /
## Raw read contract
-`Access::read` remains the raw read operation:
+`Service::read` remains the raw read operation:
```rust,ignore
-pub trait Access: Send + Sync + Debug + Unpin + 'static {
+pub trait Service: Send + Sync + Debug + Unpin + 'static {
type Reader: oio::Read;
fn read(
&self,
+ ctx: &OperationContext,
path: &str,
args: OpRead,
- ) -> impl Future<Output = Result<(RpRead, Self::Reader)>> + MaybeSend;
+ ) -> Result<Self::Reader>;
}
```
@@ -117,7 +118,7 @@ The meaning changes:
- `path` and `OpRead` select a path and read condition set.
- The returned `Self::Reader` is a reusable raw read executor for that path and
condition set.
-- No range is selected by `Access::read`.
+- No range is selected by `Service::read`.
`OpRead` should no longer contain `BytesRange`:
@@ -136,7 +137,7 @@ pub struct OpRead {
`OpReader` remains an internal execution policy used by OpenDAL's public
`Reader`. It controls `concurrent`, `chunk`, `gap`, `prefetch`, and
-`content_length_hint`. It must not become part of the backend `Access::read`
+`content_length_hint`. It must not become part of the backend `Service::read`
contract.
`ReadOptions` and `ReaderOptions` should be converted differently:
@@ -277,10 +278,15 @@ does not require every service to optimize handle reuse immediately.
range, and execution policy:
```rust,ignore
-async fn read_inner(acc: Accessor, path: String, opts: ReadOptions) -> Result<Buffer> {
+async fn read_inner(
+ ctx: OperationContext,
+ srv: Servicer,
+ path: String,
+ opts: ReadOptions,
+) -> Result<Buffer> {
let (range, op_read, op_reader) = opts.into();
- let (rp, raw_reader) = acc.read(&path, op_read).await?;
- let reader = Reader::new(acc, path, raw_reader, op_reader, rp);
+ let raw_reader = srv.read(&ctx, &path, op_read)?;
+ let reader = Reader::new(ctx, srv, path, raw_reader, op_reader);
reader.read(range.to_range()).await
}
```
@@ -288,14 +294,19 @@ async fn read_inner(acc: Accessor, path: String, opts: ReadOptions) -> Result<Bu
`Operator::reader_options` should create the reusable raw reader once:
```rust,ignore
-async fn reader_inner(acc: Accessor, path: String, opts: ReaderOptions) -> Result<Reader> {
+async fn reader_inner(
+ ctx: OperationContext,
+ srv: Servicer,
+ path: String,
+ opts: ReaderOptions,
+) -> Result<Reader> {
let (op_read, op_reader) = opts.into();
- let (rp, raw_reader) = acc.read(&path, op_read).await?;
- Ok(Reader::new(acc, path, raw_reader, op_reader, rp))
+ let raw_reader = srv.read(&ctx, &path, op_read)?;
+ Ok(Reader::new(ctx, srv, path, raw_reader, op_reader))
}
```
-For services where `Access::read` currently performs remote I/O, the returned
+For services where `Service::read` currently performs remote I/O, the returned
raw reader should be lazy. This preserves the lazy reader behavior: creating
`op.reader(path)` should not require opening the underlying data request.
@@ -305,7 +316,7 @@ raw reader should be lazy. This preserves the lazy reader behavior: creating
```rust,ignore
pub struct ReadContext {
- acc: Accessor,
+ srv: Servicer,
path: String,
args: OpRead,
options: OpReader,
@@ -314,20 +325,19 @@ pub struct ReadContext {
}
```
-`acc`, `path`, and `args` are still useful for planning, especially when
+`srv`, `path`, and `args` are still useful for planning, especially when
OpenDAL needs `stat` to resolve unbounded ranges for chunked reads or
`AsyncSeek` adapters. Actual raw range I/O should go through `reader.open` or
-`reader.read`, not through repeated `acc.read` calls.
+`reader.read`, not through repeated `srv.read` calls.
`Reader::metadata()` remains cache-only. The cache is updated from `RpRead`
-returned by `Access::read`, `open`, or `read`. Lazy backends may return
-`RpRead::default()` from `Access::read` and fill metadata after the first raw
-range operation. Public `Reader::fetch` observes metadata from the planned raw
-`read` calls it executes.
+returned by `open` or `read`. Lazy backends may return `RpRead::default()` from
+the raw reader until the first raw range operation. Public `Reader::fetch`
+observes metadata from the planned raw `read` calls it executes.
## Layers
-Existing layers continue to wrap `Access::read`, but the wrapping point moves
+Existing layers continue to wrap `Service::read`, but the wrapping point moves
from a range stream to a reusable raw reader.
### Type eraser
@@ -335,11 +345,10 @@ from a range stream to a reusable raw reader.
The type eraser should erase raw readers using the chosen object-safe container:
```rust,ignore
-async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, oio::Reader)> {
+fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<oio::Reader> {
self.inner
- .read(path, args)
- .await
- .map(|(rp, r)| (rp, Box::new(r) as oio::Reader))
+ .read(ctx, path, args)
+ .map(|r| Box::new(r) as oio::Reader)
}
```
@@ -349,7 +358,7 @@ specific type alias.
### Complete layer
-`CompleteLayer` currently reads `args.range().size()` in `Access::read` and
+`CompleteLayer` previously read `args.range().size()` in `Service::read` and
wraps the returned range stream. After this RFC, range size is only known when
`open` or `read` is called.
@@ -371,13 +380,13 @@ every range operation:
### Retry layer
-`RetryLayer` currently stores `OpRead` and advances `args.range_mut()` after
+`RetryLayer` previously stored `OpRead` and advanced `args.range_mut()` after
each successful stream read. After this RFC, range progress must move from
`OpRead` into range operation wrappers.
The new retry model should be:
-- retry `Access::read(path, OpRead)` while creating the reusable raw reader;
+- retry `Service::read(path, OpRead)` while creating the reusable raw reader;
- retry `raw_reader.open(range)` when opening a range stream fails before bytes
are returned;
- wrap the returned `ReadStream` and track bytes already emitted;
@@ -391,7 +400,7 @@ for the same path and read condition set.
### Correctness check layer
`CorrectnessCheckLayer` can keep checking `read_with_*` capabilities in
-`Access::read`, because those options still belong to the path and read
+`Service::read`, because those options still belong to the path and read
condition set.
## Presign read
@@ -420,7 +429,7 @@ OpReader)` and discards `OpReader`.
Services like `s3`, `gcs`, `azblob`, `oss`, and `http` should return a raw
reader that stores `core`, `path`, and `OpRead`.
-Each `open(range)` sends the same request that the current `Access::read` sends
+Each `open(range)` sends the same request that the current `Service::read` sends
today. For example, S3's current logic:
```rust,ignore
@@ -466,8 +475,8 @@ public `Reader` keep the same user-facing behavior.
The raw API is breaking:
-- third-party `Access` implementations must remove range handling from
- `Access::read`;
+- third-party `Service` implementations must remove range handling from
+ `Service::read`;
- current range-scoped readers must become `ReadStream`;
- services must return reusable raw readers;
- layers must wrap reusable raw readers and range operation results.
@@ -477,7 +486,7 @@ Migration can be staged mechanically:
1. Introduce `ReadStream` and the new raw `Read` trait shape.
2. Move `BytesRange` out of `OpRead`.
3. Update type erasure and core read context.
-4. Convert object-store services by moving current `Access::read` request logic
+4. Convert object-store services by moving current `Service::read` request logic
into `open(range)`.
5. Convert memory-like services by slicing in `read(range)`.
6. Convert handle based services first with independent per-range opens, then
@@ -500,9 +509,9 @@ The implementation should be considered complete when:
# Drawbacks
-- This is a breaking raw API change. Third-party `Access` implementations must
+- This is a breaking raw API change. Third-party `Service` implementations must
be updated.
-- It touches many services because every `Access::read` implementation currently
+- It touches many services because every `Service::read` implementation currently
receives a range.
- The rename from range-scoped `oio::Read` to `oio::ReadStream` is large but
necessary to avoid overloading one trait with two meanings.
@@ -525,7 +534,7 @@ progress, and request sizing into core and layers. OpenDAL already has
`open(range)` for streaming and backpressure. The raw `read(range)` primitive
should instead express a complete, core-planned, bounded range read.
-## Add `Access::open_read_at`
+## Add `Service::open_read_at`
Rejected. It adds a parallel backend operation for an execution optimization.
OpenDAL's raw operation surface should stay centered on semantic operations:
@@ -552,10 +561,10 @@ ranges from it gives one stream two responsibilities and makes layer wrapping
harder. The reusable raw reader should be the place that knows how to execute
additional ranges.
-## Add `ReadPlan` beside `Access::read`
+## Add `ReadPlan` beside `Service::read`
Rejected. It improves execution planning but keeps the old range-scoped
-`Access::read` as the primary contract. This RFC fixes the raw contract instead
+`Service::read` as the primary contract. This RFC fixes the raw contract instead
of adding a second planning hook.
# Prior art
@@ -568,7 +577,7 @@ reduce extra `open` syscalls, and listed future `read_ranges` and native
`read_at` support.
The current raw contract did not fully realize that intent because range
-selection stayed in `Access::read`. This RFC keeps RFC-4382's public direction
+selection stayed in `Service::read`. This RFC keeps RFC-4382's public direction
and moves the raw contract to the place where handle reuse and batch range reads
can be implemented.
diff --git a/core/core/src/types/context/read.rs b/core/core/src/types/context/read.rs
index 0f8e64fb5..18cb028ca 100644
--- a/core/core/src/types/context/read.rs
+++ b/core/core/src/types/context/read.rs
@@ -35,7 +35,7 @@ pub struct ReadContext {
args: OpRead,
/// Options for the reader.
options: OpReader,
- /// Raw reader returned by [`Access::read`].
+ /// Raw reader returned by [`Service::read`].
reader: oio::Reader,
/// Complete object metadata observed from successful read opens.
metadata: OnceLock<Metadata>,
diff --git a/core/core/src/types/read/reader.rs b/core/core/src/types/read/reader.rs
index 9fcfa691a..3c2d67734 100644
--- a/core/core/src/types/read/reader.rs
+++ b/core/core/src/types/read/reader.rs
@@ -541,6 +541,8 @@ impl Reader {
mod tests {
use std::sync::Arc;
use std::sync::Mutex;
+ use std::sync::atomic::AtomicUsize;
+ use std::sync::atomic::Ordering;
use std::time::Duration;
use bytes::Bytes;
@@ -636,6 +638,240 @@ mod tests {
))
}
+ #[derive(Debug, Clone)]
+ struct CountingService {
+ content: Bytes,
+ service_reads: Arc<AtomicUsize>,
+ raw_reads: Arc<Mutex<Vec<BytesRange>>>,
+ }
+
+ impl CountingService {
+ fn new(content: Bytes) -> Self {
+ Self {
+ content,
+ service_reads: Arc::default(),
+ raw_reads: Arc::default(),
+ }
+ }
+ }
+
+ impl Service for CountingService {
+ type Reader = CountingReader;
+ type Writer = ();
+ type Lister = ();
+ type Deleter = ();
+ type Copier = ();
+
+ fn info(&self) -> ServiceInfo {
+ ServiceInfo::with_scheme("counting")
+ }
+
+ fn capability(&self) -> Capability {
+ Capability {
+ stat: true,
+ read: true,
+ ..Default::default()
+ }
+ }
+
+ async fn create_dir(
+ &self,
+ _: &OperationContext,
+ _: &str,
+ _: OpCreateDir,
+ ) -> Result<RpCreateDir> {
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "operation is not supported",
+ ))
+ }
+
+ async fn stat(&self, _: &OperationContext, _: &str, _: OpStat) -> Result<RpStat> {
+ Ok(RpStat::new(
+ Metadata::new(EntryMode::FILE).with_content_length(self.content.len() as u64),
+ ))
+ }
+
+ fn read(&self, _: &OperationContext, _: &str, _: OpRead) -> Result<Self::Reader> {
+ self.service_reads.fetch_add(1, Ordering::Relaxed);
+ Ok(CountingReader {
+ content: self.content.clone(),
+ raw_reads: self.raw_reads.clone(),
+ })
+ }
+
+ fn write(&self, _: &OperationContext, _: &str, _: OpWrite) -> Result<Self::Writer> {
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "operation is not supported",
+ ))
+ }
+
+ fn delete(&self, _: &OperationContext) -> Result<Self::Deleter> {
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "operation is not supported",
+ ))
+ }
+
+ fn list(&self, _: &OperationContext, _: &str, _: OpList) -> Result<Self::Lister> {
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "operation is not supported",
+ ))
+ }
+
+ fn copy(
+ &self,
+ _: &OperationContext,
+ _: &str,
+ _: &str,
+ _: OpCopy,
+ _: OpCopier,
+ ) -> Result<Self::Copier> {
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "operation is not supported",
+ ))
+ }
+
+ async fn rename(
+ &self,
+ _: &OperationContext,
+ _: &str,
+ _: &str,
+ _: OpRename,
+ ) -> Result<RpRename> {
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "operation is not supported",
+ ))
+ }
+
+ async fn presign(&self, _: &OperationContext, _: &str, _: OpPresign) -> Result<RpPresign> {
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "operation is not supported",
+ ))
+ }
+ }
+
+ #[derive(Debug)]
+ struct CountingReader {
+ content: Bytes,
+ raw_reads: Arc<Mutex<Vec<BytesRange>>>,
+ }
+
+ impl CountingReader {
+ fn slice(&self, range: BytesRange) -> Result<Bytes> {
+ let content_len = self.content.len();
+ let (start, end) = match range {
+ BytesRange::Range { offset, size } => {
+ let start = offset as usize;
+ let end = size.map_or(content_len, |size| start + size as usize);
+ (start, end)
+ }
+ BytesRange::Suffix { size } => {
+ let start = content_len.saturating_sub(size as usize);
+ (start, content_len)
+ }
+ };
+
+ if end > content_len {
+ return Err(Error::new(
+ ErrorKind::RangeNotSatisfied,
+ "range exceeds content length",
+ ));
+ }
+
+ Ok(self.content.slice(start..end))
+ }
+ }
+
+ impl oio::Read for CountingReader {
+ async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
+ self.raw_reads.lock().unwrap().push(range);
+ let content = self.slice(range)?;
+ Ok((
+ RpRead::new(
+ Metadata::new(EntryMode::FILE).with_content_length(self.content.len() as u64),
+ ),
+ Box::new(CountingReadStream { content }) as Box<dyn oio::ReadStreamDyn>,
+ ))
+ }
+
+ async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
+ self.raw_reads.lock().unwrap().push(range);
+ Ok((
+ RpRead::new(
+ Metadata::new(EntryMode::FILE).with_content_length(self.content.len() as u64),
+ ),
+ Buffer::from(self.slice(range)?),
+ ))
+ }
+ }
+
+ struct CountingReadStream {
+ content: Bytes,
+ }
+
+ impl oio::ReadStream for CountingReadStream {
+ async fn read(&mut self) -> Result<Buffer> {
+ Ok(Buffer::from(std::mem::take(&mut self.content)))
+ }
+ }
+
+ #[tokio::test]
+ async fn test_reader_reuses_raw_reader_for_planned_ranges() -> Result<()> {
+ let service = CountingService::new(Bytes::from_static(b"0123456789"));
+ let service_reads = service.service_reads.clone();
+ let raw_reads = service.raw_reads.clone();
+ let op = Operator::from_parts(OperationContext::default(), Arc::new(service));
+
+ let reader = op.reader_with("test_file").chunk(2).concurrent(2).await?;
+ assert_eq!(service_reads.load(Ordering::Relaxed), 1);
+
+ let buf = reader.read(0..6).await?;
+ assert_eq!(buf.to_bytes(), b"012345".as_slice());
+
+ let bufs = reader.fetch(vec![6..8, 8..10]).await?;
+ assert_eq!(bufs[0].to_bytes(), b"67".as_slice());
+ assert_eq!(bufs[1].to_bytes(), b"89".as_slice());
+
+ let mut async_reader = reader.clone().into_futures_async_read(0..10).await?;
+ let mut buf = [0; 3];
+ futures::AsyncReadExt::read_exact(&mut async_reader, &mut buf)
+ .await
+ .unwrap();
+ assert_eq!(&buf, b"012");
+ futures::AsyncSeekExt::seek(&mut async_reader, std::io::SeekFrom::Start(4))
+ .await
+ .unwrap();
+ let mut buf = [0; 2];
+ futures::AsyncReadExt::read_exact(&mut async_reader, &mut buf)
+ .await
+ .unwrap();
+ assert_eq!(&buf, b"45");
+
+ assert_eq!(service_reads.load(Ordering::Relaxed), 1);
+ assert_eq!(
+ *raw_reads.lock().unwrap(),
+ vec![
+ BytesRange::new(0, Some(2)),
+ BytesRange::new(2, Some(2)),
+ BytesRange::new(4, Some(2)),
+ BytesRange::new(6, Some(2)),
+ BytesRange::new(8, Some(2)),
+ BytesRange::new(0, Some(2)),
+ BytesRange::new(2, Some(2)),
+ BytesRange::new(4, Some(2)),
+ BytesRange::new(6, Some(2)),
+ ]
+ );
+
+ Ok(())
+ }
+
#[tokio::test]
async fn test_trait() -> Result<()> {
let op = Operator::via_iter(services::MEMORY_SCHEME, [])?;
diff --git a/core/services/sftp/src/backend.rs b/core/services/sftp/src/backend.rs
index 5b8cb8ab9..256cbd1da 100644
--- a/core/services/sftp/src/backend.rs
+++ b/core/services/sftp/src/backend.rs
@@ -203,7 +203,7 @@ pub struct SftpBackend {
}
impl Service for SftpBackend {
- type Reader = oio::StreamReader<SftpReader>;
+ type Reader = oio::PositionReader<SftpReader>;
type Writer = SftpLazyWriter;
type Lister = SftpLazyLister;
type Deleter = oio::OneShotDeleter<SftpDeleter>;
@@ -255,15 +255,11 @@ impl Service for SftpBackend {
Ok(RpStat::new(meta))
}
fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
- let output: oio::StreamReader<SftpReader> = {
- Ok(oio::StreamReader::new(SftpReader::new(
- self.clone(),
- path,
- args,
- )))
- }?;
-
- Ok(output)
+ Ok(oio::PositionReader::new(SftpReader::new(
+ self.clone(),
+ path,
+ args,
+ )))
}
fn write(&self, ctx: &OperationContext, path: &str, op: OpWrite) -> Result<Self::Writer> {
diff --git a/core/services/sftp/src/reader.rs b/core/services/sftp/src/reader.rs
index a2cf46881..39e97a053 100644
--- a/core/services/sftp/src/reader.rs
+++ b/core/services/sftp/src/reader.rs
@@ -22,7 +22,6 @@ use super::core::is_sftp_failure;
use super::core::parse_sftp_error;
use super::lister::SftpLister;
use super::writer::SftpWriter;
-use bytes::BytesMut;
use fastpool::bounded;
use opendal_core::raw::*;
use opendal_core::*;
@@ -30,59 +29,6 @@ use openssh_sftp_client::file::File;
use std::io::SeekFrom;
use tokio::io::AsyncSeekExt;
-pub struct SftpReadStream {
- /// Keep the connection alive while data stream is alive.
- _conn: bounded::Object<Manager>,
-
- file: File,
- chunk: usize,
- size: Option<usize>,
- read: usize,
- buf: BytesMut,
-}
-
-impl SftpReadStream {
- pub fn new(conn: bounded::Object<Manager>, file: File, size: Option<u64>) -> Self {
- Self {
- _conn: conn,
- file,
- size: size.map(|v| v as usize),
- chunk: 2 * 1024 * 1024,
- read: 0,
- buf: BytesMut::new(),
- }
- }
-}
-
-impl oio::ReadStream for SftpReadStream {
- async fn read(&mut self) -> Result<Buffer> {
- if self.read >= self.size.unwrap_or(usize::MAX) {
- return Ok(Buffer::new());
- }
-
- let size = if let Some(size) = self.size {
- (size - self.read).min(self.chunk)
- } else {
- self.chunk
- };
- self.buf.reserve(size);
-
- let Some(bytes) = self
- .file
- .read(size as u32, self.buf.split_off(0))
- .await
- .map_err(parse_sftp_error)?
- else {
- return Ok(Buffer::new());
- };
-
- self.read += bytes.len();
- self.buf = bytes;
- let bs = self.buf.split();
- Ok(Buffer::from(bs.freeze()))
- }
-}
-
/// Reader returned by this backend.
pub struct SftpReader {
backend: SftpBackend,
@@ -98,8 +44,22 @@ impl SftpReader {
}
}
-impl oio::StreamRead for SftpReader {
- async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
+pub struct SftpReaderHandle {
+ /// Keep the connection alive while the file handle is cached by `PositionReader`.
+ _conn: bounded::Object<Manager>,
+ file: File,
+}
+
+impl SftpReaderHandle {
+ pub(super) fn new(conn: bounded::Object<Manager>, file: File) -> Self {
+ Self { _conn: conn, file }
+ }
+}
+
+impl oio::PositionRead for SftpReader {
+ type Handle = SftpReaderHandle;
+
+ async fn open(&self) -> Result<Self::Handle> {
let backend = &self.backend;
let path = self.path.as_str();
@@ -110,21 +70,32 @@ impl oio::StreamRead for SftpReader {
let path = fs.canonicalize(path).await.map_err(parse_sftp_error)?;
- let mut f = client
+ let f = client
.open(path.as_path())
.await
.map_err(parse_sftp_error)?;
- if range.offset() != 0 {
- f.seek(SeekFrom::Start(range.offset()))
- .await
- .map_err(new_std_io_error)?;
+ Ok(SftpReaderHandle::new(client, f))
+ }
+
+ async fn read_at(handle: &Self::Handle, offset: u64, size: usize) -> Result<Buffer> {
+ if size == 0 {
+ return Ok(Buffer::new());
}
- let rp = RpRead::default();
- let stream = SftpReadStream::new(client, f, range.size());
+ let mut file = handle.file.clone();
+ file.seek(SeekFrom::Start(offset))
+ .await
+ .map_err(new_std_io_error)?;
- Ok((rp, Box::new(stream) as Box<dyn oio::ReadStreamDyn>))
+ match file
+ .read(size as u32, bytes::BytesMut::with_capacity(size))
+ .await
+ .map_err(parse_sftp_error)?
+ {
+ Some(bytes) => Ok(Buffer::from(bytes.freeze())),
+ None => Ok(Buffer::new()),
+ }
}
}