Merge branch '1.10' into 1.11

.gitignore

@@ -7,3 +7,4 @@ test.log
 docs/
 !sites/docs
 _build
+.coverage

.travis.yml

@@ -10,7 +10,7 @@ install:
   - pip install -r dev-requirements.txt
 script:
   # Main tests, with coverage!
-  - coverage run --source=paramiko test.py --verbose
+  - invoke coverage
   # Ensure documentation & invoke pipeline run OK.
   # Run 'docs' first since its objects.inv is referred to by 'www'.
   # Also force warnings to be errors since most of them tend to be actual

paramiko/sftp_client.py

@@ -736,7 +736,7 @@ class SFTPClient (BaseSFTP):
                     self._convert_status(msg)
                 return t, msg
             if fileobj is not type(None):
-                fileobj._async_response(t, msg)
+                fileobj._async_response(t, msg, num)
             if waitfor is None:
                 # just doing a single check
                 break

paramiko/sftp_file.py

@@ -20,6 +20,8 @@
 L{SFTPFile}
 """
 
+from __future__ import with_statement
+
 from binascii import hexlify
 from collections import deque
 import socket
@@ -53,7 +55,8 @@ class SFTPFile (BufferedFile):
         self._prefetching = False
         self._prefetch_done = False
         self._prefetch_data = {}
-        self._prefetch_reads = []
+        self._prefetch_extents = {}
+        self._prefetch_lock = threading.Lock()
         self._saved_exception = None
         self._reqs = deque()
 
@@ -91,7 +94,7 @@ class SFTPFile (BufferedFile):
             pass
 
     def _data_in_prefetch_requests(self, offset, size):
-        k = [i for i in self._prefetch_reads if i[0] <= offset]
+        k = [x for x in self._prefetch_extents.values() if x[0] <= offset]
         if len(k) == 0:
             return False
         k.sort(lambda x, y: cmp(x[0], y[0]))
@@ -447,7 +450,6 @@ class SFTPFile (BufferedFile):
     def _start_prefetch(self, chunks):
         self._prefetching = True
         self._prefetch_done = False
-        self._prefetch_reads.extend(chunks)
 
         t = threading.Thread(target=self._prefetch_thread, args=(chunks,))
         t.setDaemon(True)
@@ -457,9 +459,11 @@ class SFTPFile (BufferedFile):
         # do these read requests in a temporary thread because there may be
         # a lot of them, so it may block.
         for offset, length in chunks:
-            self.sftp._async_request(self, CMD_READ, self.handle, long(offset), int(length))
+            with self._prefetch_lock:
+                num = self.sftp._async_request(self, CMD_READ, self.handle, long(offset), int(length))
+                self._prefetch_extents[num] = (offset, length)
 
-    def _async_response(self, t, msg):
+    def _async_response(self, t, msg, num):
         if t == CMD_STATUS:
             # save exception and re-raise it on next file operation
             try:
@@ -470,10 +474,12 @@ class SFTPFile (BufferedFile):
         if t != CMD_DATA:
             raise SFTPError('Expected data')
         data = msg.get_string()
-        offset, length = self._prefetch_reads.pop(0)
+        with self._prefetch_lock:
-        self._prefetch_data[offset] = data
+            offset, length = self._prefetch_extents[num]
-        if len(self._prefetch_reads) == 0:
+            self._prefetch_data[offset] = data
-            self._prefetch_done = True
+            del self._prefetch_extents[num]
+            if len(self._prefetch_extents) == 0:
+                self._prefetch_done = True
     
     def _check_exception(self):
         "if there's a saved exception, raise & clear it"

sites/www/changelog.rst

@@ -2,7 +2,10 @@
 Changelog
 =========
 
-* :bug:`193` (and its attentant PRs :issue:`230` & :issue:`253`): Fix SSH agent
+* :bug:`34` (PR :issue:`35`) Fix SFTP prefetching incompatibility with some
+  SFTP servers regarding request/response ordering. Thanks to Richard
+  Kettlewell for catch & patch.
+* :bug:`193` (and its attentant PRs :issue:`230` & :issue:`253`) Fix SSH agent
   problems present on Windows. Thanks to David Hobbs for initial report and to
   Aarni Koskela & Olle Lundberg for the patches.
 * :release:`1.11.3 <2014-01-08>`

tasks.py

@@ -1,7 +1,7 @@
 from os.path import join
 
-from invoke import Collection
+from invoke import Collection, ctask as task
-from invocations import docs as _docs, testing
+from invocations import docs as _docs
 
 
 d = 'sites'
@@ -20,4 +20,15 @@ www = Collection.from_module(_docs, name='www', config={
     'sphinx.target': join(path, '_build'),
 })
 
-ns = Collection(testing.test, docs=docs, www=www)
+
+# Until we move to spec-based testing
+@task
+def test(ctx):
+    ctx.run("python test.py --verbose")
+
+@task
+def coverage(ctx):
+    ctx.run("coverage run --source=paramiko test.py --verbose")
+
+
+ns = Collection(test, coverage, docs=docs, www=www)