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
The following commit(s) were added to refs/heads/fix-reader by this push:
new 2cb635434 fix(services/sftp): avoid extra file open in positioned-read
probe
2cb635434 is described below
commit 2cb635434b8387c87530e618b47cec2110778c3e
Author: Erick Guan <[email protected]>
AuthorDate: Sun Jul 5 14:23:27 2026 +0800
fix(services/sftp): avoid extra file open in positioned-read probe
Restructure the positioned-read detection so the probe reuses the file
handle it opens instead of opening a separate file just for probing.
Key changes:
- The per-reader OnceCell now stores Option<Arc<SftpReaderHandle>>
instead of Arc<SftpReaderHandle>. This caches both outcomes:
Some(handle) when positioned reads work, None when unsupported.
Previously, returning Err from get_or_try_init left the cell
uninitialized, causing redundant file opens on every call.
- On first use, opens one file and probes positioned reads on it.
If supported, reuses that handle for subsequent reads. If not,
caches None service-wide via SftpCore::positioned_read_support
so all future readers skip the probe entirely.
- Removed the separate probe_positioned_read_support() method.
The probe is now folded into positioned_handle(), which is the
only caller.
---
core/services/sftp/src/reader.rs | 115 +++++++++++++++++++++++++--------------
1 file changed, 75 insertions(+), 40 deletions(-)
diff --git a/core/services/sftp/src/reader.rs b/core/services/sftp/src/reader.rs
index 227fa55b8..3ebc494ba 100644
--- a/core/services/sftp/src/reader.rs
+++ b/core/services/sftp/src/reader.rs
@@ -90,12 +90,18 @@ impl oio::ReadStream for SftpReadStream {
}
/// Reader returned by this backend.
+///
+/// On the first read, the reader opens a file handle and tries a positioned
+/// read (seek + read). If the server supports it, the handle is cached for the
+/// reader's lifetime and all subsequent reads go through the positioned path.
+/// If the server does not support positioned reads, the result is cached
+/// service-wide and the reader falls back to streaming.
pub struct SftpReader {
backend: SftpBackend,
path: String,
- /// Lazily opened positioned-read handle, cached per reader for the file's
- /// lifetime. Only set when the server supports positioned reads.
- positioned: Arc<OnceCell<Arc<SftpReaderHandle>>>,
+ /// Lazily resolved read strategy: `Some(handle)` when positioned reads
+ /// are supported, `None` when they are not.
+ positioned: Arc<OnceCell<Option<Arc<SftpReaderHandle>>>>,
}
impl SftpReader {
@@ -145,44 +151,67 @@ impl SftpReader {
Ok((client, file))
}
- /// Probe whether the server supports positioned reads (seek + read),
- /// caching the result service-wide so subsequent readers skip the probe.
- async fn probe_positioned_read_support(&self) -> Result<bool> {
- self.backend
- .core
- .positioned_read_support
- .get_or_try_init(|| async {
- // Open a temporary file just for the probe.
- let (client, file) = self.open_file().await?;
- let handle = SftpReaderHandle::new(client, file);
-
- match probe_positioned_read(&handle).await {
- Ok(supported) => Ok(supported),
- Err(err) if is_positioned_read_unsupported(&err) =>
Ok(false),
- Err(err) => Err(err),
- }
- })
- .await
- .copied()
- }
-
/// Return a cached positioned-read handle for this reader, or `None`
/// if the server does not support positioned reads.
+ ///
+ /// On the very first call across the entire service, this opens a file
+ /// and attempts a positioned read to detect support. The file handle is
+ /// reused for subsequent reads when supported. The detection result is
+ /// cached in `SftpCore::positioned_read_support` so all future readers
+ /// skip the probe entirely.
async fn positioned_handle(&self) -> Result<Option<Arc<SftpReaderHandle>>>
{
- if !self.probe_positioned_read_support().await? {
+ // Fast path: a previous reader already determined positioned reads
+ // are not supported — no need to open a file.
+ if self
+ .backend
+ .core
+ .positioned_read_support
+ .get()
+ .is_some_and(|&supported| !supported)
+ {
return Ok(None);
}
- // Positioned read is supported — lazily open or reuse the handle.
- let handle = self
+ // Resolve the per-reader strategy (probe + open handle in one shot).
+ let result = self
.positioned
.get_or_try_init(|| async {
let (client, file) = self.open_file().await?;
- Ok(Arc::new(SftpReaderHandle::new(client, file)))
+ let handle = Arc::new(SftpReaderHandle::new(client, file));
+
+ // If a previous reader already resolved support, respect that.
+ if let Some(&supported) =
self.backend.core.positioned_read_support.get() {
+ return Ok(if supported {
+ Some(handle)
+ } else {
+ // Unsupported — the handle we just opened is wasted,
+ // but the per-reader OnceCell is populated so we never
+ // re-enter this closure.
+ None
+ });
+ }
+
+ // First reader across the service — probe by trying a real
+ // positioned read on the file we already opened.
+ match probe_positioned_read(&handle).await {
+ Ok(()) => {
+ // Probe succeeded — positioned reads work. Cache the
+ // result service-wide so other readers skip the probe.
+ let _ =
self.backend.core.positioned_read_support.set(true).await;
+ Ok(Some(handle))
+ }
+ Err(err) if is_positioned_read_unsupported(&err) => {
+ // Positioned reads are not supported. Cache the
+ // result service-wide so other readers skip the probe.
+ let _ =
self.backend.core.positioned_read_support.set(false).await;
+ Ok(None)
+ }
+ Err(err) => Err(err),
+ }
})
.await?;
- Ok(Some(handle.clone()))
+ Ok(result.clone())
}
}
@@ -218,20 +247,26 @@ async fn read_positioned(handle: &SftpReaderHandle,
offset: u64, size: usize) ->
}
}
-/// Probe whether positioned reads work by reading one byte.
-///
-/// If the file has a known length, read at EOF (expecting an empty result).
-/// Otherwise, read at offset 0 (expecting at most one byte).
-async fn probe_positioned_read(handle: &SftpReaderHandle) -> Result<bool> {
+/// Probe whether positioned reads work by seeking to the end of the file and
+/// attempting a small read. A server that supports positioned reads will
+/// succeed (returning empty at EOF); a server that does not support seek
+/// will return an error we classify as "unsupported".
+async fn probe_positioned_read(handle: &SftpReaderHandle) -> Result<()> {
let mut file = handle.file.clone();
let len = file.metadata().await.map_err(parse_sftp_error)?.len();
- if let Some(len) = len {
- let buf = read_positioned(handle, len, 1).await?;
- Ok(buf.is_empty())
- } else {
- read_positioned(handle, 0, 1).await.map(|_| true)
- }
+ let probe_offset = len.unwrap_or(0);
+ file.seek(SeekFrom::Start(probe_offset))
+ .await
+ .map_err(new_std_io_error)?;
+
+ // A small read to confirm the data path works after seeking.
+ let _ = file
+ .read(1u32, BytesMut::with_capacity(1))
+ .await
+ .map_err(parse_sftp_error)?;
+
+ Ok(())
}
fn is_positioned_read_unsupported(err: &Error) -> bool {
