Buffered rw (#241)

* Added concurrent RW buffer class and tests
* Migrated ssh client to concurrent rw buffer. Added iteration support to reader. Updated ssh client line parsing.
* Added partial read timeout with and without join test and for updated API calls.
* Updated host output to make stdout/stderr dynamic properties.
* Deprecated reset generators function - now a no-op.
* Added read timeout and encoding attributes to host output.
* Updated changelog, documentation.

Author: pkittenis

Changelog.rst

@@ -1,6 +1,49 @@
 Change Log
 ============
 
+2.3.0
++++++
+
+Changes
+-------
+
+* ``SSHClient`` now starts buffering output from remote host, both standard output and standard error, when a command is run.
+* ``SSHClient.read_output``, ``SSHClient.read_stderr`` and iterating on stdout/stderr from ``HostOutput`` now read from the internal buffer rather than the SSH channel directly.
+* ``ParallelSSHClient.join`` no longer requires ``consume_output`` to be set in order to get exit codes without first reading output.
+* ``ParallelSSHClient.join`` with timeout no longer consumes output by default. It is now possible to use ``join`` with a timeout and capture output after ``join`` completes.
+* ``ParallelSSHClient.reset_output_generators`` is now a no-op and no longer required to be called after timeouts.
+* ``HostOutput.stdout`` and ``stderr`` are now dynamic properties.
+* Added ``HostOutput.read_timeout`` attribute. Can be used to see what read timeout was when ``run_command`` was called and to change timeout when next reading from ``HostOutput.stdout`` and ``stderr``.
+* Added ``HostOutput.encoding`` attribute for encoding used when ``run_command`` was called. Encoding can now be changed for when next reading output.
+* ``ParallelSSHClient.join`` with timeout no longer affects ``stdout`` or ``stderr`` read timeout set when ``run_command`` was called.
+* LibSSH clients under ``pssh.clients.ssh`` now allow output to be read as it becomes available without waiting for remote command to finish first.
+* Reading from output behaviour is now consistent across all client types - parallel and single clients under both ``pssh.clients.native`` and ``pssh.clients.ssh``.
+* ``ParallelSSHClient.join`` can now be called without arguments and defaults to last ran commands.
+* ``ParallelSSHClient.finished`` can now be called without arguments and defaults to last ran commands.
+
+
+This is now possible:
+
+.. code-block:: python
+
+   output = client.run_command(<..>)
+   client.join(output)
+   assert output[0].exit_code is not None
+
+As is this:
+
+.. code-block:: python
+
+   client.run_command(<..>, timeout=1)
+   client.join(output, timeout=1)
+   for line in output[0].stdout:
+       print(line)
+
+Output can be read after and has separate timeout from join.
+
+See `documentation for more examples on use of timeouts <https://parallel-ssh.readthedocs.io/en/latest/advanced.html#partial-output>`_.
+
+
 2.2.0
 +++++
 

doc/advanced.rst

@@ -188,7 +188,9 @@ See :py:mod:`HostConfig <pssh.config.HostConfig>` for all possible configuration
 Join and Output Timeouts
 **************************
 
-Clients have timeout functionality on reading output and ``client.join``. Join timeout is a timeout on all parallel commands in total and is separate from ``ParallelSSHClient(timeout=<..>)`` which is applied to SSH session operations individually.
+Clients have timeout functionality on reading output and ``client.join``.
+
+Join timeout is applied to all parallel commands in total and is separate from ``ParallelSSHClient(timeout=<..>)`` which is applied to SSH session operations individually.
 
 Timeout exceptions contain attributes for which commands have finished and which have not so client code can get output from any finished commands when handling timeouts.
 
@@ -207,18 +209,18 @@ The client will raise a ``Timeout`` exception if *all* remote commands have not
 
 .. code-block:: python
 
-   output = client.run_command(.., timeout=5)
+   output = client.run_command(.., read_timeout=5)
    for host_out in output:
        try:
            for line in host_out.stdout:
-	       pass
+	       print(line)
            for line in host_out.stderr:
-	       pass
+	       print(line)
        except Timeout:
            pass
 
 
-In the case of reading from output such as in the example above, timeout value is per output stream - meaning separate timeouts for stdout and stderr as well as separate timeout per host remote command.
+In the case of reading from output such as in the example above, timeout value is per output stream - meaning separate timeouts for stdout and stderr as well as separate timeout per host output.
 
 *New in 1.5.0*
 
@@ -253,6 +255,7 @@ In the above example, output is printed only for those commands which have compl
 
 Client code may choose to then join again only on the unfinished output if some commands have failed in order to gather remaining output.
 
+.. _partial-output:
 
 Reading Partial Output of Commands That Do Not Terminate
 ==========================================================
@@ -262,17 +265,16 @@ In some cases, such as when the remote command never terminates unless interrupt
 .. code-block:: python
 
    output = client.run_command(
-       'tail -f /var/log/messages', use_pty=True, timeout=1)
+       'while true; do echo a line; sleep .1; done', use_pty=True, read_timeout=1)
 
-   # Read as many lines of output as server has sent before the timeout
+   # Read as many lines of output as hosts have sent before the timeout
    stdout = []
    for host_out in output:
        try:
            for line in host_out.stdout:
                stdout.append(line)
        except Timeout:
-           # This allows client code to continue to read output after timeout
-           client.reset_output_generators(host_out, timeout=1)
+           pass
 
    # Closing channel which has PTY has the effect of terminating
    # any running processes started on that channel.
@@ -281,18 +283,20 @@ In some cases, such as when the remote command never terminates unless interrupt
    # Join is not strictly needed here as channel has already been closed and
    # command has finished, but is safe to use regardless.
    client.join(output)
+   # Can now read output up to when the channel was closed without blocking.
+   rest_of_stdout = list(output[0].stdout)
 
 Without a PTY, a ``join`` call with a timeout will complete with timeout exception raised but the remote process will be left running as per SSH protocol specifications.
 
-Furthermore, once reading output has timed out, it is necessary to restart the output generators as by Python design they only iterate once. This is done by ``client.reset_output_generators`` in the above example.
+.. note::
 
-Generator reset is also performed automatically by calls to ``join`` and does not need to be done manually when ``join`` is used after output reading.
+   Read timeout may be changed after ``run_command`` has been called by changing ``HostOutput.read_timeout`` for that particular host output.
 
 .. note::
 
-   ``join`` with a timeout forces output to be consumed as otherwise the pending output will keep the channel open and make it appear as if command has not yet finished.
+   When output from commands is not needed, it is best to use ``client.join(consume_output=True)`` so that output buffers are consumed automatically.
 
-   To capture output when using ``join`` with a timeout, gather output first before calling ``join``, making use of output timeout as well, and/or make use of :ref:`host logger` functionality.
+   If output is not read or automatically consumed by ``join`` output buffers will continually grow, resulting in increasing memory consumption while the client is running, though memory use rises very slowly.
 
 
 Per-Host Configuration
@@ -320,19 +324,7 @@ In the above example, the client is configured to connect to hostname ``localhos
 
 When using ``host_config``, the number of ``HostConfig`` entries must match the number of hosts in ``client.hosts``. An exception is raised on client initialisation if not.
 
-
-.. note::
-
-   Currently only ``port``, ``user``, ``password`` and ``private_key`` ``HostConfig`` values are used.
-
-
-.. note::
-
-   Proxy host configuration is currently per ``ParallelSSHClient`` and cannot yet be provided via per-host configuration.
-   Multiple clients can be used to make use of multiple proxy hosts.
-
-   This feature will be provided in future releases.
-
+As of `2.2.0`, proxy configuration can also be provided in ``HostConfig``.
 
 .. _per-host-cmds:
 
@@ -586,7 +578,7 @@ For example, to copy the local files ``['local_file_1', 'local_file_2']`` as rem
 
 The client will copy ``local_file_1`` to ``host1`` as ``remote_file_1`` and ``local_file_2`` to ``host2`` as ``remote_file_2``.
 
-Each item in ``copy_args`` list should be a dictionary as shown above. Number of ``copy_args`` must match length of ``client.hosts`` if provided or exception will be raised.
+Each item in ``copy_args`` list should be a dictionary as shown above. Number of items in ``copy_args`` must match length of ``client.hosts`` if provided or exception will be raised.
 
 ``copy_remote_file``, ``scp_send`` and ``scp_recv`` may all be used in the same manner to configure remote and local file names per host.
 
@@ -605,6 +597,7 @@ If wanting to copy a file from a single remote host and retain the original file
 
    client = SSHClient('localhost')
    client.copy_remote_file('remote_filename', 'local_filename')
+   client.scp_recv('remote_filename', 'local_filename')
 
 .. seealso::
 
@@ -653,6 +646,7 @@ Hosts list can be modified in place. A call to ``run_command`` will create new c
 
    client.hosts = ['otherhost']
    print(client.run_command('exit 0'))
+       <..>
        host='otherhost'
        exit_code=None
        <..>

doc/quickstart.rst

@@ -117,23 +117,16 @@ Standard output, aka ``stdout``, for a given :py:class:`HostOutput <pssh.output.
       <line by line output>
       <..>
 
-There is nothing special needed to ensure output is available.
-
-Please note that retrieving all of a command's standard output by definition requires that the command has completed.
-
-Iterating over ``stdout`` for any host *to completion* will therefor *only complete* when that host's command has completed unless interrupted.
+Iterating over ``stdout`` will only end when the remote command has finished unless interrupted.
 
 The ``timeout`` keyword argument to ``run_command`` may be used to cause output generators to timeout if no output is received after the given number of seconds - see `join and output timeouts <advanced.html#join-and-output-timeouts>`_.
 
-``stdout`` is a generator. Iterating over it will consume the remote standard output stream via the network as it becomes available. To retrieve all of stdout can wrap it with list, per below.
+``stdout`` is a generator. To retrieve all of stdout can wrap it with list, per below.
 
 .. code-block:: python
 
    stdout = list(host_out.stdout)
 
-.. warning::
-
-   This will store the entirety of stdout into memory.
 
 All hosts iteration
 -------------------

pssh/clients/base/parallel.py

@@ -21,6 +21,7 @@
 
 import gevent.pool
 
+from warnings import warn
 from gevent import joinall, spawn, Timeout as GTimeout
 from gevent.hub import Hub
 
@@ -123,27 +124,16 @@ def _get_output_from_greenlet(self, cmd, raise_error=False):
                 ex = Timeout()
             if raise_error:
                 raise ex
-            return HostOutput(host, None, None, None, None,
-                              None, exception=ex)
+            return HostOutput(host, None, None, None,
+                              exception=ex)
 
     def get_last_output(self, cmds=None, timeout=None,
                         return_list=True):
-        """Get output for last commands executed by ``run_command``
+        """Get output for last commands executed by ``run_command``.
 
         :param cmds: Commands to get output for. Defaults to ``client.cmds``
         :type cmds: list(:py:class:`gevent.Greenlet`)
-        :param greenlet_timeout: (Optional) Greenlet timeout setting.
-          Defaults to no timeout. If set, this function will raise
-          :py:class:`gevent.Timeout` after ``greenlet_timeout`` seconds
-          if no result is available from greenlets.
-          In some cases, such as when using proxy hosts, connection timeout
-          is controlled by proxy server and getting result from greenlets may
-          hang indefinitely if remote server is unavailable. Use this setting
-          to avoid blocking in such circumstances.
-          Note that ``gevent.Timeout`` is a special class that inherits from
-          ``BaseException`` and thus **can not be caught** by
-          ``stop_on_errors=False``.
-        :type greenlet_timeout: float
+        :param timeout: No-op - to be removed.
         :param return_list: No-op - list of ``HostOutput`` always returned.
           Parameter kept for backwards compatibility - to be removed in future
           releases.
@@ -161,32 +151,9 @@ def get_last_output(self, cmds=None, timeout=None,
     def reset_output_generators(self, host_out, timeout=None,
                                 client=None, channel=None,
                                 encoding='utf-8'):
-        """Reset output generators for host output.
-
-        :param host_out: Host output
-        :type host_out: :py:class:`pssh.output.HostOutput`
-        :param client: (Optional) SSH client
-        :type client: :py:class:`pssh.ssh2_client.SSHClient`
-        :param channel: (Optional) SSH channel
-        :type channel: :py:class:`ssh2.channel.Channel`
-        :param timeout: (Optional) Timeout setting
-        :type timeout: int
-        :param encoding: (Optional) Encoding to use for output. Must be valid
-          `Python codec <https://docs.python.org/library/codecs.html>`_
-        :type encoding: str
-
-        :rtype: tuple(stdout, stderr)
-        """
-        channel = host_out.channel if channel is None else channel
-        client = host_out.client if client is None else client
-        stdout = client.read_output_buffer(
-            client.read_output(channel, timeout=timeout), encoding=encoding)
-        stderr = client.read_output_buffer(
-            client.read_stderr(channel, timeout=timeout),
-            prefix='\t[err]', encoding=encoding)
-        host_out.stdout = stdout
-        host_out.stderr = stderr
-        return stdout, stderr
+        """No-op - to be removed."""
+        warn("This function is a no-op and deprecated. "
+             "It will be removed in future releases")
 
     def _get_host_config_values(self, host_i, host):
         if self.host_config is None:
@@ -251,7 +218,7 @@ def _consume_output(self, stdout, stderr):
         for line in stderr:
             pass
 
-    def join(self, output, consume_output=False, timeout=None,
+    def join(self, output=None, consume_output=False, timeout=None,
              encoding='utf-8'):
         """Wait until all remote commands in output have finished.
         Does *not* block other commands from running in parallel.
@@ -264,9 +231,7 @@ def join(self, output, consume_output=False, timeout=None,
           output on call to ``join`` when host logger has been enabled.
         :type consume_output: bool
         :param timeout: Timeout in seconds if **all** remote commands are not
-          yet finished. Note that use of timeout forces ``consume_output=True``
-          otherwise the channel output pending to be consumed always results
-          in the channel not being finished.
+          yet finished.
           This function's timeout is for all commands in total and will therefor
           be affected by pool size and total number of concurrent commands in
           self.pool.
@@ -281,7 +246,9 @@ def join(self, output, consume_output=False, timeout=None,
           reached with commands still running.
 
         :rtype: ``None``"""
-        if not isinstance(output, list):
+        if output is None:
+            output = self.get_last_output()
+        elif not isinstance(output, list):
             raise ValueError("Unexpected output object type")
         cmds = [self.pool.spawn(self._join, host_out, timeout=timeout,
                                 consume_output=consume_output, encoding=encoding)
@@ -305,21 +272,24 @@ def _join(self, host_out, consume_output=False, timeout=None,
         client = host_out.client
         if client is None:
             return
-        stdout, stderr = self.reset_output_generators(
-            host_out, channel=channel, timeout=timeout,
-            encoding=encoding)
         client.wait_finished(channel, timeout=timeout)
         if consume_output:
-            self._consume_output(stdout, stderr)
+            self._consume_output(host_out.stdout, host_out.stderr)
         return host_out
 
-    def finished(self, output):
-        """Check if commands have finished without blocking
+    def finished(self, output=None):
+        """Check if commands have finished without blocking.
 
         :param output: As returned by
-          :py:func:`pssh.pssh_client.ParallelSSHClient.get_output`
+          :py:func:`pssh.pssh_client.ParallelSSHClient.get_last_output`
+        :type output: list
+
         :rtype: bool
         """
+        if output is None:
+            output = self.get_last_output()
+            if output is None:
+                return True
         for host_out in output:
             chan = host_out.channel
             if host_out.client and not host_out.client.finished(chan):