Write fixes (#272)

* Added multi scp send test with timeout
* Updated clients to yield event loop in between writes.
* Moved eagain handling to base client
* Updated changelog
* Updated documentation

Author: pkittenis

Changelog.rst

@@ -1,6 +1,15 @@
 Change Log
 ============
 
+2.5.3
++++++
+
+Fixes
+-----
+
+* Sending files via ``scp_send`` or ``sftp_put`` with timeout set could timeout unexpectedly on opening remote file - #271.
+
+
 2.5.2
 +++++
 

doc/clients.rst

@@ -9,7 +9,7 @@ Below is a comparison of feature support for the two client types.
 Feature                          ssh2-python (libssh2)  ssh-python (libssh)
 ===============================  ====================== ======================
 Agent forwarding                  No                    Not yet implemented
-Proxying/tunnelling               Yes                   No
+Proxying/tunneling                Yes                   No
 Kerberos (GSS) authentication     Not supported         Yes
 Private key file authentication   Yes                   Yes
 Agent authentication              Yes                   Yes
@@ -25,4 +25,4 @@ Keep-alive functionality          Yes                   No
 
 The default client offers the most features, but lacks certain authentication mechanisms like GSS-API and certificate authentication. Both client types are based on C libraries and offer similar levels of performance.
 
-Users that need the authentication mechanisms not supported by the default client can ``from pssh.clients.ssh import ParallelSSHClient`` instead of the default client.
+Users that need the authentication mechanisms not supported by the default client can ``from pssh.clients.ssh import ParallelSSHClient, SSHClient`` instead of the default clients.

doc/installation.rst

@@ -36,7 +36,6 @@ Dependency         Minimum Version
 Building from Source
 ----------------------
 
-
 ``parallel-ssh`` is hosted on GitHub and the repository can be cloned with the following
 
 .. code-block:: shell
@@ -48,7 +47,7 @@ To install from source run:
 
 .. code-block:: shell
 
-  python setup.py install
+  pip install .
 
 Or for developing changes:
 
@@ -60,10 +59,14 @@ Or for developing changes:
 Python 2
 --------
 
-As of January 2021, Python 2 is no longer supported by the Python Software Foundation nor ``parallel-ssh`` - see `Sunset Python 2 <https://www.python.org/doc/sunset-python-2/>`_.
+As of January 1st 2021, Python 2 is no longer supported by the Python Software Foundation nor ``parallel-ssh`` - see `Sunset Python 2 <https://www.python.org/doc/sunset-python-2/>`_.
 
 Versions of ``parallel-ssh<=2.4.0`` will still work.
 
 Future releases are not guaranteed to be compatible or work at all with Python 2.
 
+Versions ``>=2.5.0`` may still work, no code changes have yet been made to stop Python 2 from working, though no support is offered if they do not.
+
+Vendors that distribute this package for Python 2 are advised to run integration tests themselves on versions ``>=2.5.0`` to confirm compatibility - no automated testing is performed on Python 2 by this project.
+
 If your company requires Python 2 support contact the author directly at the email address on Github commits to discuss rates.

pssh/clients/base/single.py

@@ -519,6 +519,25 @@ def run_command(self, command, sudo=False, user=None,
         host_out = self._make_host_output(channel, encoding, _timeout)
         return host_out
 
+    def _eagain_write_errcode(self, write_func, data, eagain, timeout=None):
+        data_len = len(data)
+        total_written = 0
+        while total_written < data_len:
+            rc, bytes_written = write_func(data[total_written:])
+            total_written += bytes_written
+            if rc == eagain:
+                self.poll(timeout=timeout)
+            sleep()
+
+    def _eagain_errcode(self, func, eagain, *args, **kwargs):
+        timeout = kwargs.pop('timeout', self.timeout)
+        with GTimeout(seconds=timeout, exception=Timeout):
+            ret = func(*args, **kwargs)
+            while ret == eagain:
+                self.poll()
+                ret = func(*args, **kwargs)
+            return ret
+
     def _eagain_write(self, write_func, data, timeout=None):
         raise NotImplementedError
 

pssh/clients/native/single.py

@@ -20,7 +20,7 @@
 from collections import deque
 from warnings import warn
 
-from gevent import sleep, spawn, get_hub, Timeout as GTimeout
+from gevent import sleep, spawn, get_hub
 from gevent.select import POLLIN, POLLOUT
 from ssh2.error_codes import LIBSSH2_ERROR_EAGAIN
 from ssh2.exceptions import SFTPHandleError, SFTPProtocolError, \
@@ -314,13 +314,7 @@ def close_channel(self, channel):
         self._eagain(channel.close)
 
     def _eagain(self, func, *args, **kwargs):
-        timeout = kwargs.pop('timeout', self.timeout)
-        with GTimeout(seconds=timeout, exception=Timeout):
-            ret = func(*args, **kwargs)
-            while ret == LIBSSH2_ERROR_EAGAIN:
-                self.poll()
-                ret = func(*args, **kwargs)
-            return ret
+        return self._eagain_errcode(func, LIBSSH2_ERROR_EAGAIN, *args, **kwargs)
 
     def _make_sftp(self):
         """Make SFTP client from open transport"""
@@ -707,18 +701,12 @@ def poll(self, timeout=None):
             events |= POLLOUT
         self._poll_socket(events, timeout=timeout)
 
-    def eagain_write(self, write_func, data, timeout=None):
+    def _eagain_write(self, write_func, data, timeout=None):
         """Write data with given write_func for an ssh2-python session while
         handling EAGAIN and resuming writes from last written byte on each call to
         write_func.
         """
-        data_len = len(data)
-        total_written = 0
-        while total_written < data_len:
-            rc, bytes_written = write_func(data[total_written:])
-            total_written += bytes_written
-            if rc == LIBSSH2_ERROR_EAGAIN:
-                self.poll(timeout=timeout)
+        return self._eagain_write_errcode(write_func, data, LIBSSH2_ERROR_EAGAIN, timeout=timeout)
 
-    def _eagain_write(self, write_func, data, timeout=None):
-        return self.eagain_write(write_func, data, timeout=timeout)
+    def eagain_write(self, write_func, data, timeout=None):
+        return self._eagain_write(write_func, data, timeout=timeout)

pssh/clients/ssh/single.py

@@ -304,7 +304,7 @@ def close_channel(self, channel):
         self._eagain(channel.close, timeout=self.timeout)
 
     def poll(self, timeout=None):
-        """ssh-python based co-operative gevent select on session socket."""
+        """ssh-python based co-operative gevent poll on session socket."""
         timeout = self.timeout if timeout is None else timeout
         directions = self.session.get_poll_flags()
         if directions == 0:
@@ -318,19 +318,7 @@ def poll(self, timeout=None):
 
     def _eagain(self, func, *args, **kwargs):
         """Run function given and handle EAGAIN for an ssh-python session"""
-        timeout = kwargs.pop('timeout', self.timeout)
-        with GTimeout(seconds=timeout, exception=Timeout):
-            ret = func(*args, **kwargs)
-            while ret == SSH_AGAIN:
-                self.poll(timeout=timeout)
-                ret = func(*args, **kwargs)
-            return ret
+        return self._eagain_errcode(func, SSH_AGAIN, *args, **kwargs)
 
     def _eagain_write(self, write_func, data, timeout=None):
-        data_len = len(data)
-        total_written = 0
-        while total_written < data_len:
-            rc, bytes_written = write_func(data[total_written:])
-            total_written += bytes_written
-            if rc == SSH_AGAIN:
-                self.poll(timeout=timeout)
+        return self._eagain_write_errcode(write_func, data, SSH_AGAIN, timeout=timeout)

tests/native/test_parallel_client.py

@@ -1582,6 +1582,53 @@ def test_scp_send_dir_recurse(self):
             except OSError:
                 pass
 
+    def test_scp_send_large_files_timeout(self):
+        hosts = ['127.0.0.1%s' % (i,) for i in range(1, 11)]
+        servers = [OpenSSHServer(host, port=self.port) for host in hosts]
+        for server in servers:
+            server.start_server()
+        client = ParallelSSHClient(
+            hosts, port=self.port, pkey=self.user_key, num_retries=1, timeout=1)
+        local_filename = 'test_file'
+        remote_filepath = 'file_copy'
+        copy_args = [{
+            'local_file': local_filename,
+            'remote_file': 'host_%s_%s' % (n, remote_filepath)}
+                     for n in range(len(hosts))]
+        remote_file_names = [arg['remote_file'] for arg in copy_args]
+        sha = sha256()
+        with open(local_filename, 'wb') as file_h:
+            for _ in range(5000):
+                data = os.urandom(1024)
+                file_h.write(data)
+                sha.update(data)
+        source_file_sha = sha.hexdigest()
+        sha = sha256()
+        cmds = client.scp_send('%(local_file)s', '%(remote_file)s', copy_args=copy_args)
+        try:
+            joinall(cmds, raise_error=True)
+        except Exception:
+            raise
+        else:
+            sleep(.2)
+            for remote_file_name in remote_file_names:
+                remote_file_abspath = os.path.expanduser('~/' + remote_file_name)
+                self.assertTrue(os.path.isfile(remote_file_abspath))
+                with open(remote_file_abspath, 'rb') as remote_fh:
+                    for data in remote_fh:
+                        sha.update(data)
+                remote_file_sha = sha.hexdigest()
+                sha = sha256()
+                self.assertEqual(source_file_sha, remote_file_sha)
+        finally:
+            try:
+                os.unlink(local_filename)
+                for remote_file_name in remote_file_names:
+                    remote_file_abspath = os.path.expanduser('~/' + remote_file_name)
+                    os.unlink(remote_file_abspath)
+            except OSError:
+                pass
+
     def test_scp_send(self):
         server2_host = '127.0.0.11'
         server3_host = '127.0.0.12'