Close SSH channel after test (#3504)

This is paired with DFK behaviour introduced in PR #3503 which closes channels
in the DFK. This PR modifies a test which does not use the DFK to also perform
that shutdown.

With test parsl/tests/test_providers/test_local_provider.py, this reduces the
number of threads remaining at the end of execution from 3 to 2.

Author: benclifford

Diff for: parsl/channels/ssh/ssh.py

@@ -227,8 +227,20 @@ def pull_file(self, remote_source, local_dir):
 
     def close(self) -> None:
         if self._is_connected():
+            transport = self.ssh_client.get_transport()
             self.ssh_client.close()
 
+            # ssh_client.close calls transport.close, but transport.close does
+            # not always wait for the transport thread to be stopped. See impl
+            # of Transport.close in paramiko and issue
+            # https://github.com/paramiko/paramiko/issues/520
+            logger.debug("Waiting for transport thread to stop")
+            transport.join(30)
+            if transport.is_alive():
+                logger.warning("SSH transport thread did not shut down")
+            else:
+                logger.debug("SSH transport thread stopped")
+
     def isdir(self, path):
         """Return true if the path refers to an existing directory.
 

Diff for: parsl/tests/test_providers/test_local_provider.py

@@ -92,13 +92,17 @@ def test_ssh_channel():
                 # already exist, so create it here.
                 pathlib.Path('{}/known.hosts'.format(config_dir)).touch(mode=0o600)
                 script_dir = tempfile.mkdtemp()
-                p = LocalProvider(channel=SSHChannel('127.0.0.1', port=server_port,
-                                                     script_dir=remote_script_dir,
-                                                     host_keys_filename='{}/known.hosts'.format(config_dir),
-                                                     key_filename=priv_key),
-                                  launcher=SingleNodeLauncher(debug=False))
-                p.script_dir = script_dir
-                _run_tests(p)
+                channel = SSHChannel('127.0.0.1', port=server_port,
+                                     script_dir=remote_script_dir,
+                                     host_keys_filename='{}/known.hosts'.format(config_dir),
+                                     key_filename=priv_key)
+                try:
+                    p = LocalProvider(channel=channel,
+                                      launcher=SingleNodeLauncher(debug=False))
+                    p.script_dir = script_dir
+                    _run_tests(p)
+                finally:
+                    channel.close()
         finally:
             _stop_sshd(sshd_thread)