Use the block size determined by asyncssh if nothing is specified

sshfs/file.py

@@ -25,32 +25,41 @@ def __init__(
         self.mode = mode
         self.max_requests = max_requests or _MAX_SFTP_REQUESTS
 
-        if block_size is None:
-            # "The OpenSSH SFTP server will close the connection
-            # if it receives a message larger than 256 KB, and
-            # limits read requests to returning no more than
-            # 64 KB."
-            #
-            # We are going to use the maximum block_size possible
-            # with a 16KB margin (so instead of sending 256 KB data,
-            # we'll send 240 KB + headers for write requests)
-
-            if self.readable():
-                block_size = READ_BLOCK_SIZE
-            else:
-                block_size = WRITE_BLOCK_SIZE
-
         # The blocksize is often used with constructs like
         # shutil.copyfileobj(src, dst, length=file.blocksize) and since we are
         # using pipelining, we are going to reflect the total size rather than
         # a size of chunk to our limits.
-        self.blocksize = block_size * self.max_requests
+        self.blocksize = (
+            None if block_size is None else block_size * self.max_requests
+        )
 
         self.kwargs = kwargs
 
         self._file = sync(self.loop, self._open_file)
         self._closed = False
 
+    def _determine_block_size(self, channel):
+        # Use the limits reported by the server (limits@openssh.com) to
+        # get the best performance. A zero max_packet_len means the
+        # server never reported limits and the read/write lengths are
+        # asyncssh's synthesized 16 KiB floors -- fall through to the
+        # larger defaults below instead of degrading to them.
+        limits = getattr(channel, "limits", None)
+        if limits and limits.max_packet_len:
+            if self.readable():
+                return limits.max_read_len
+            return limits.max_write_len
+
+        # "The OpenSSH SFTP server will close the connection
+        # if it receives a message larger than 256 KB, and
+        # limits read requests to returning no more than
+        # 64 KB."
+        #
+        # We are going to use the maximum block_size possible
+        # with a 16KB margin (so instead of sending 256 KB data,
+        # we'll send 240 KB + headers for write requests)
+        return READ_BLOCK_SIZE if self.readable() else WRITE_BLOCK_SIZE
+
     @wrap_exceptions
     async def _open_file(self):
         # TODO: this needs to keep a reference to the
@@ -60,6 +69,10 @@ async def _open_file(self):
         # it's operations but the pool it thinking this
         # channel is freed.
         async with self.fs._pool.get() as channel:
+            if self.blocksize is None:
+                self.blocksize = (
+                    self._determine_block_size(channel) * self.max_requests
+                )
             return await channel.open(
                 self.path,
                 self.mode,
@@ -80,9 +93,6 @@ async def _open_file(self):
     def readable(self):
         return "r" in self.mode or "+" in self.mode
 
-    def seekable(self):
-        return "r" in self.mode or "w" in self.mode
-
     def seekable(self):
         return True
 

tests/test_sshfs.py

@@ -6,13 +6,16 @@
 from concurrent import futures
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
+from types import SimpleNamespace
 
 import fsspec
 import pytest
 from asyncssh.sftp import SFTPAttrs, SFTPFailure
 from importlib_metadata import entry_points
 
 from sshfs import SSHFileSystem
+from sshfs.file import SSHFile
+from sshfs.utils import READ_BLOCK_SIZE, WRITE_BLOCK_SIZE
 
 _STATIC = (Path(__file__).parent / "static").resolve()
 USERS = {"user": _STATIC / "user.key"}
@@ -336,6 +339,50 @@ def test_exceptions(fs, remote_dir):
         fs.makedirs(remote_dir + "/dir/a/b/c")
 
 
+def test_open_block_size(fs, remote_dir):
+    # mockssh (paramiko) does not implement limits@openssh.com, so the
+    # tuned defaults must survive asyncssh's synthesized 16 KiB floors.
+    fs.touch(remote_dir + "/a.txt")
+    with fs.open(remote_dir + "/a.txt", "rb") as file:
+        assert file.blocksize == READ_BLOCK_SIZE * file.max_requests
+    with fs.open(remote_dir + "/b.txt", "wb") as file:
+        assert file.blocksize == WRITE_BLOCK_SIZE * file.max_requests
+    # An explicit block_size always wins.
+    with fs.open(remote_dir + "/c.txt", "wb", block_size=4096) as file:
+        assert file.blocksize == 4096 * file.max_requests
+
+
+def test_determine_block_size():
+    reader = SimpleNamespace(readable=lambda: True)
+    writer = SimpleNamespace(readable=lambda: False)
+    determine = SSHFile._determine_block_size
+
+    # The server reported its limits: use them.
+    reported = SimpleNamespace(
+        limits=SimpleNamespace(
+            max_packet_len=262144,
+            max_read_len=261120,
+            max_write_len=131072,
+        )
+    )
+    assert determine(reader, reported) == 261120
+    assert determine(writer, reported) == 131072
+
+    # No limits@openssh.com support: asyncssh synthesizes 16 KiB
+    # read/write floors with max_packet_len == 0; keep the defaults.
+    synthesized = SimpleNamespace(
+        limits=SimpleNamespace(
+            max_packet_len=0, max_read_len=16384, max_write_len=16384
+        )
+    )
+    assert determine(reader, synthesized) == READ_BLOCK_SIZE
+    assert determine(writer, synthesized) == WRITE_BLOCK_SIZE
+
+    # asyncssh without the limits API at all.
+    assert determine(reader, SimpleNamespace()) == READ_BLOCK_SIZE
+    assert determine(writer, SimpleNamespace()) == WRITE_BLOCK_SIZE
+
+
 def test_open_rw(fs, remote_dir):
     data = b"dvc.org"