Better handle timeout in execute_together

Sometimes paramiko hangs on
remote status event wait in parallel calls
(when remote side already closed connection)

Author: penguinolog

doc/source/conf.py

@@ -16,7 +16,7 @@
 import pkg_resources
 
 release = pkg_resources.get_distribution("exec-helpers").version
-version = '.'.join(release.split('.')[:2])
+version = ".".join(release.split(".")[:2])
 
 # If extensions (or modules to document with autodoc) are in another directory,
 # add these directories to sys.path here. If the directory is relative to the
@@ -36,32 +36,32 @@
 # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
 # ones.
 extensions = [
-    'sphinx.ext.autodoc',
-    'sphinx.ext.doctest',
-    'sphinx.ext.coverage',
-    'sphinx.ext.viewcode',
+    "sphinx.ext.autodoc",
+    "sphinx.ext.doctest",
+    "sphinx.ext.coverage",
+    "sphinx.ext.viewcode",
 ]
 
 # Add any paths that contain templates here, relative to this directory.
-templates_path = ['_templates']
+templates_path = ["_templates"]
 
 # The suffix(es) of source filenames.
 # You can specify multiple suffix as a list of string:
 #
 # source_suffix = ['.rst', '.md']
-source_suffix = '.rst'
+source_suffix = ".rst"
 
 # The encoding of source files.
 #
 # source_encoding = 'utf-8-sig'
 
 # The master toctree document.
-master_doc = 'index'
+master_doc = "index"
 
 # General information about the project.
-project = 'Exec-helpers'
-copyright = '2018, Alexey Stepanov'
-author = 'Alexey Stepanov'
+project = "Exec-helpers"
+copyright = "2018, Alexey Stepanov"
+author = "Alexey Stepanov"
 
 # The version info for the project you're documenting, acts as replacement for
 # |version| and |release|, also used in various other places throughout the
@@ -112,7 +112,7 @@
 # show_authors = False
 
 # The name of the Pygments (syntax highlighting) style to use.
-pygments_style = 'sphinx'
+pygments_style = "sphinx"
 
 # A list of ignored prefixes for module index sorting.
 # modindex_common_prefix = []
@@ -129,14 +129,14 @@
 # The theme to use for HTML and HTML Help pages.  See the documentation for
 # a list of builtin themes.
 #
-html_theme = 'alabaster'
+html_theme = "alabaster"
 
 # Theme options are theme-specific and customize the look and feel of a theme
 # further.  For a list of options available for each theme, see the
 # documentation.
 #
 html_theme_options = {
-    'page_width': 'auto',
+    "page_width": "auto",
 }
 
 # Add any paths that contain custom themes here, relative to this directory.
@@ -165,7 +165,7 @@
 # Add any paths that contain custom static files (such as style sheets) here,
 # relative to this directory. They are copied after the builtin static files,
 # so a file named "default.css" will overwrite the builtin "default.css".
-html_static_path = ['_static']
+html_static_path = ["_static"]
 
 # Add any extra paths that contain custom files (such as robots.txt or
 # .htaccess) here, relative to this directory. These files are copied
@@ -245,35 +245,30 @@
 # html_search_scorer = 'scorer.js'
 
 # Output file base name for HTML help builder.
-htmlhelp_basename = 'exec_helpers_doc'
+htmlhelp_basename = "exec_helpers_doc"
 
 # -- Options for LaTeX output ---------------------------------------------
 
 latex_elements = {
-     # The paper size ('letterpaper' or 'a4paper').
-     #
-     # 'papersize': 'letterpaper',
-
-     # The font size ('10pt', '11pt' or '12pt').
-     #
-     # 'pointsize': '10pt',
-
-     # Additional stuff for the LaTeX preamble.
-     #
-     # 'preamble': '',
-
-     # Latex figure (float) alignment
-     #
-     # 'figure_align': 'htbp',
+    # The paper size ('letterpaper' or 'a4paper').
+    #
+    # 'papersize': 'letterpaper',
+    # The font size ('10pt', '11pt' or '12pt').
+    #
+    # 'pointsize': '10pt',
+    # Additional stuff for the LaTeX preamble.
+    #
+    # 'preamble': '',
+    # Latex figure (float) alignment
+    #
+    # 'figure_align': 'htbp',
 }
 
 # Grouping the document tree into LaTeX files. List of tuples
 # (source start file, target name, title,
 #  author, documentclass [howto, manual, or own class]).
 latex_documents = [
-    (master_doc, 'Exec-helpers.tex',
-     'Exec helpers Documentation',
-     'Alexey Stepanov', 'manual'),
+    (master_doc, "Exec-helpers.tex", "Exec helpers Documentation", "Alexey Stepanov", "manual"),
 ]
 
 # The name of an image file (relative to this directory) to place at the top of
@@ -313,10 +308,7 @@
 
 # One entry per manual page. List of tuples
 # (source start file, name, description, authors, manual section).
-man_pages = [
-    (master_doc, 'exec-helpers', 'Exec helpers Documentation',
-     [author], 1)
-]
+man_pages = [(master_doc, "exec-helpers", "Exec helpers Documentation", [author], 1)]
 
 # If true, show URL addresses after external links.
 #
@@ -329,9 +321,15 @@
 # (source start file, target name, title, author,
 #  dir menu entry, description, category)
 texinfo_documents = [
-    (master_doc, 'exec-helpers', 'Exec helpers Documentation',
-     author, 'Exec-helpers', 'One line description of project.',
-     'Miscellaneous'),
+    (
+        master_doc,
+        "exec-helpers",
+        "Exec helpers Documentation",
+        author,
+        "Exec-helpers",
+        "One line description of project.",
+        "Miscellaneous",
+    ),
 ]
 
 # Documents to append as an appendix to all manuals.

exec_helpers/_ssh_base.py

@@ -709,6 +709,7 @@ def _execute_async(  # pylint: disable=arguments-differ
         get_pty: bool = False,
         width: int = 80,
         height: int = 24,
+        timeout: OptionalTimeoutT = None,
         **kwargs: typing.Any,
     ) -> SshExecuteAsyncResult:
         """Execute command in async mode and return channel with IO objects.
@@ -729,6 +730,8 @@ def _execute_async(  # pylint: disable=arguments-differ
         :type width: int
         :param height: PTY height
         :type height: int
+        :param timeout: timeout before stop execution with TimeoutError (will be set on channel)
+        :type timeout: typing.Union[int, float, None]
         :param kwargs: additional parameters for call.
         :type kwargs: typing.Any
         :return: Tuple with control interface and file-like objects for STDIN/STDERR/STDOUT
@@ -751,6 +754,8 @@ def _execute_async(  # pylint: disable=arguments-differ
         .. versionchanged:: 4.1.0 support chroot
         """
         chan: paramiko.Channel = self._ssh_transport.open_session()
+        if timeout is not None:
+            chan.settimeout(timeout)
 
         if get_pty:
             # Open PTY
@@ -1401,6 +1406,7 @@ def get_result(remote: SSHClientBase) -> exec_result.ExecResult:
 
             :param remote: SSH connection instance
             :return: execution result
+            :raises ExecHelperTimeoutError: Timeout exceeded
             """
             # pylint: disable=protected-access
             cmd_for_log: str = remote._mask_command(cmd=cmd, log_mask_re=log_mask_re)
@@ -1417,20 +1423,28 @@ def get_result(remote: SSHClientBase) -> exec_result.ExecResult:
                 log_mask_re=log_mask_re,
                 open_stdout=open_stdout,
                 open_stderr=open_stderr,
+                timeout=timeout,
                 **kwargs,
             )
             # pylint: enable=protected-access
 
-            async_result.interface.status_event.wait(timeout)
-            exit_code = async_result.interface.recv_exit_status()
+            done = async_result.interface.status_event.wait(timeout)
 
             res = exec_result.ExecResult(cmd=cmd_for_log, stdin=stdin, started=async_result.started)
             res.read_stdout(src=async_result.stdout)
             res.read_stderr(src=async_result.stderr)
-            res.exit_code = exit_code
+            if done:
+                res.exit_code = async_result.interface.recv_exit_status()
 
             async_result.interface.close()
-            return res
+            if done:
+                return res
+            async_result.interface.status_event.set()
+            result.set_timestamp()
+
+            wait_err_msg: str = _log_templates.CMD_WAIT_ERROR.format(result=res, timeout=timeout)
+            remote.logger.debug(wait_err_msg)
+            raise exceptions.ExecHelperTimeoutError(result=res, timeout=timeout)  # type: ignore
 
         prep_expected: typing.Sequence[ExitCodeT] = proc_enums.exit_codes_to_enums(expected)
         log_level: int = logging.INFO if verbose else logging.DEBUG
@@ -1443,14 +1457,15 @@ def get_result(remote: SSHClientBase) -> exec_result.ExecResult:
         errors: typing.Dict[typing.Tuple[str, int], exec_result.ExecResult] = {}
         raised_exceptions: typing.Dict[typing.Tuple[str, int], Exception] = {}
 
-        _, not_done = concurrent.futures.wait(list(futures.values()), timeout=timeout)
+        not_done: typing.Set[concurrent.futures.Future[exec_result.ExecResult]]
+        _done, not_done = concurrent.futures.wait(futures.values(), timeout=timeout)
 
         for fut in not_done:  # pragma: no cover
             fut.cancel()
 
         for remote, future in futures.items():
             try:
-                result = future.result()
+                result = future.result(timeout=0.1)
                 results[(remote.hostname, remote.port)] = result
                 if result.exit_code not in prep_expected:
                     errors[(remote.hostname, remote.port)] = result

exec_helpers/exceptions.py

@@ -304,7 +304,9 @@ def __init__(
 
         .. versionchanged:: 3.4.0 Expected is not optional, defaults os dependent
         """
-        exceptions_str: str = "\n\t".join(f"{host}:{port} - {exc} " for (host, port), exc in exceptions.items())
+        exceptions_str: str = "\n\t".join(
+            f"{host}:{port} - {str(exc) if str(exc) else repr(exc)}" for (host, port), exc in exceptions.items()
+        )
         message: str = _message or f"Command {command!r} during execution raised exceptions: \n\t{exceptions_str}"
         super().__init__(
             command=command,

test/async_api/test_subprocess_special.py

@@ -122,13 +122,17 @@ class MockParameters(typing.NamedTuple):
     "TimeoutError": dict(
         stdout=(),
         command_parameters=CommandParameters(),
-        mock_parameters=MockParameters(ec=(asyncio.TimeoutError(), -9),),
+        mock_parameters=MockParameters(
+            ec=(asyncio.TimeoutError(), -9),
+        ),
         expect_exc=exec_helpers.ExecHelperTimeoutError,
     ),
     "TimeoutError_no_kill": dict(
         stdout=(),
         command_parameters=CommandParameters(),
-        mock_parameters=MockParameters(ec=(asyncio.TimeoutError(), None),),
+        mock_parameters=MockParameters(
+            ec=(asyncio.TimeoutError(), None),
+        ),
         expect_exc=exec_helpers.ExecHelperNoKillError,
     ),
     "stdin_closed_PIPE_windows": dict(
@@ -238,7 +242,8 @@ def create_subprocess_shell(mocker, monkeypatch, run_parameters):
     mocker.patch("psutil.Process")
 
     def create_mock(
-        stdout: typing.Optional[typing.Tuple] = None, **kwargs,
+        stdout: typing.Optional[typing.Tuple] = None,
+        **kwargs,
     ):
         """Parametrized code."""
         proc = asynctest.CoroutineMock()

test/test_ssh_client_execute.py

@@ -464,13 +464,15 @@ def test_009_execute_together(ssh, ssh2, execute_async, exec_result, run_paramet
                     log_mask_re=None,
                     open_stdout=run_parameters["open_stdout"],
                     open_stderr=run_parameters["open_stderr"],
+                    timeout=default_timeout,
                 ),
                 mock.call(
                     command,
                     stdin=run_parameters["stdin"],
                     log_mask_re=None,
                     open_stdout=run_parameters["open_stdout"],
                     open_stderr=run_parameters["open_stderr"],
+                    timeout=default_timeout,
                 ),
             )
         )
@@ -503,13 +505,15 @@ def test_010_execute_together_expected(ssh, ssh2, execute_async, exec_result, ru
                 log_mask_re=None,
                 open_stdout=run_parameters["open_stdout"],
                 open_stderr=run_parameters["open_stderr"],
+                timeout=default_timeout,
             ),
             mock.call(
                 command,
                 stdin=run_parameters.get("stdin", None),
                 log_mask_re=None,
                 open_stdout=run_parameters["open_stdout"],
                 open_stderr=run_parameters["open_stderr"],
+                timeout=default_timeout,
             ),
         )
     )

test/test_ssh_client_execute_special.py

@@ -240,8 +240,22 @@ def get_patched_execute_async_retval() -> SshExecuteAsyncResult:
     results = exec_helpers.SSHClient.execute_together(remotes=remotes, command=cmd)
     execute_async.assert_has_calls(
         (
-            mock.call(decoded_cmd, stdin=None, log_mask_re=None, open_stdout=True, open_stderr=True),
-            mock.call(decoded_cmd, stdin=None, log_mask_re=None, open_stdout=True, open_stderr=True),
+            mock.call(
+                decoded_cmd,
+                stdin=None,
+                log_mask_re=None,
+                open_stdout=True,
+                open_stderr=True,
+                timeout=default_timeout,
+            ),
+            mock.call(
+                decoded_cmd,
+                stdin=None,
+                log_mask_re=None,
+                open_stdout=True,
+                open_stderr=True,
+                timeout=default_timeout,
+            ),
         )
     )
     for result in results.values():