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 dd70f7db4aed367260460bacd7b01884f622f804 Author: Erick Guan <[email protected]> AuthorDate: Sun Jul 5 12:55:44 2026 +0800 Fix reader --- 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 ++++++--------- 4 files changed, 278 insertions(+), 75 deletions(-) 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()), + } } }
