fix style of new commit

Author: roywei

src/sagemaker_training/mpi.py

@@ -13,7 +13,6 @@
 """This module contains functionality related to distributed training using
 MPI (Message Passing Interface)."""
 import argparse
-import inspect
 import logging
 import os
 import subprocess
@@ -24,22 +23,26 @@
 
 import gethostname
 from sagemaker_training import environment, errors, logging_config, process, timeout
-from inspect import isclass
+from inspect import getfile, isclass
 
 logger = logging_config.get_logger()
 logging.getLogger("paramiko").setLevel(logging.INFO)
 
 try:
     from smdistributed.modelparallel.backend import exceptions
+
     # list of exceptions SMMP wants training toolkit to catch and log
     exception_classes = [x for x in dir(exceptions) if isclass(getattr(exceptions, x))]
 except ImportError as e:
     logger.info("No exception classes found in smdistributed.modelparallel")
     exception_classes = []
 try:
     from smdistributed.modelparallel.torch import exceptions as torch_exceptions
+
     # list of torch exceptions SMMP wants training toolkit to catch and log
-    exception_classes += [x for x in dir(torch_exceptions) if isclass(getattr(torch_exceptions, x))]
+    exception_classes += [
+        x for x in dir(torch_exceptions) if isclass(getattr(torch_exceptions, x))
+    ]
 except ImportError as e:
     logger.info("No torch exception classes found in smdistributed.modelparallel")
 
@@ -49,7 +52,9 @@ class WorkerRunner(process.ProcessRunner):
     master execution to finish.
     """
 
-    def __init__(self, user_entry_point, args, env_vars, processes_per_host, master_hostname):
+    def __init__(
+        self, user_entry_point, args, env_vars, processes_per_host, master_hostname
+    ):
         """Initialize a WorkerRunner, which is responsible for preparing distributed
         training with MPI and waiting for MPI master execution to finish.
 
@@ -59,7 +64,9 @@ def __init__(self, user_entry_point, args, env_vars, processes_per_host, master_
             env_vars (dict(str,str)): A dictionary of environment variables.
             master_hostname (str): The master hostname.
         """
-        super(WorkerRunner, self).__init__(user_entry_point, args, env_vars, processes_per_host)
+        super(WorkerRunner, self).__init__(
+            user_entry_point, args, env_vars, processes_per_host
+        )
         self._master_hostname = str(master_hostname)
 
     def run(
@@ -77,7 +84,9 @@ def run(
             self._wait_master_to_start()
         logger.info("MPI Master online, creating SSH daemon.")
 
-        logger.info("Writing environment variables to /etc/environment for the MPI process.")
+        logger.info(
+            "Writing environment variables to /etc/environment for the MPI process."
+        )
         _write_env_vars_to_file()
 
         _start_sshd_daemon()
@@ -114,7 +123,9 @@ def _wait_orted_process_to_finish():  # type: () -> None
 def _orted_process():  # pylint: disable=inconsistent-return-statements
     """Wait a maximum of 5 minutes for orted process to start."""
     for _ in range(5 * 60):
-        procs = [p for p in psutil.process_iter(attrs=["name"]) if p.info["name"] == "orted"]
+        procs = [
+            p for p in psutil.process_iter(attrs=["name"]) if p.info["name"] == "orted"
+        ]
         if procs:
             logger.info("Process[es]: %s", procs)
             return procs
@@ -158,7 +169,9 @@ def __init__(
                 3600 seconds (ie. 1 hour).
         """
 
-        super(MasterRunner, self).__init__(user_entry_point, args, env_vars, processes_per_host)
+        super(MasterRunner, self).__init__(
+            user_entry_point, args, env_vars, processes_per_host
+        )
 
         self._master_hostname = master_hostname
         self._hosts = hosts
@@ -194,10 +207,14 @@ def _create_command(self):
         if self._processes_per_host == 1:
             host_list = self._hosts
         else:
-            host_list = ["%s:%s" % (host, self._processes_per_host) for host in self._hosts]
+            host_list = [
+                "%s:%s" % (host, self._processes_per_host) for host in self._hosts
+            ]
 
         msg = "Env Hosts: %s Hosts: %s process_per_hosts: %s num_processes: %s"
-        logger.info(msg, self._hosts, host_list, self._processes_per_host, num_processes)
+        logger.info(
+            msg, self._hosts, host_list, self._processes_per_host, num_processes
+        )
 
         overridden_known_options, additional_options = _parse_custom_mpi_options(
             self._custom_mpi_options
@@ -250,12 +267,16 @@ def _create_command(self):
             "-x",
             "PATH",
             "-x",
-            "LD_PRELOAD=%s" % inspect.getfile(gethostname),
+            "LD_PRELOAD=%s" % getfile(gethostname),
         ]
 
         command.extend(additional_options)
 
-        for credential in ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN"]:
+        for credential in [
+            "AWS_ACCESS_KEY_ID",
+            "AWS_SECRET_ACCESS_KEY",
+            "AWS_SESSION_TOKEN",
+        ]:
             if credential in os.environ:
                 command.extend(["-x", credential])
 
@@ -293,15 +314,19 @@ def run(self, wait=True, capture_error=False):
         if wait:
             process_spawned = process.check_error(
                 cmd,
-                exception_classes if training_env.is_modelparallel_enabled else errors.ExecuteUserScriptError,
+                exception_classes
+                if training_env.is_modelparallel_enabled
+                else errors.ExecuteUserScriptError,
                 self._processes_per_host,
                 capture_error=capture_error,
                 cwd=environment.code_dir,
             )
         else:
             _, _, process_spawned = process.create(
                 cmd,
-                exception_classes if training_env.is_modelparallel_enabled else errors.ExecuteUserScriptError,
+                exception_classes
+                if training_env.is_modelparallel_enabled
+                else errors.ExecuteUserScriptError,
                 self._processes_per_host,
                 capture_error=capture_error,
                 cwd=environment.code_dir,
@@ -311,7 +336,6 @@ def run(self, wait=True, capture_error=False):
         return process_spawned
 
 
-
 _SSH_DAEMON_NOT_FOUND_ERROR_MESSAGE = """
 SSH daemon not found, please install SSH to allow MPI to communicate different nodes in cluster.
 

src/sagemaker_training/params.py

@@ -59,4 +59,4 @@
 SMDATAPARALLEL_CUSTOM_MPI_OPTIONS = (
     "sagemaker_distributed_dataparallel_custom_mpi_options"
 )  # type: str
-SM_HP_MP_PARAMETERS = "SM_HP_MP_PARAMETERS" 
+SM_HP_MP_PARAMETERS = "SM_HP_MP_PARAMETERS"

src/sagemaker_training/process.py

@@ -90,7 +90,14 @@ async def watch(stream, error_classes, proc_per_host):
                 if err_line not in output:
                     output.append(err_line.strip(" :\n") + "\n")
             else:
-                if any(str(err) in err_line for err in (_PYTHON_ERRORS_ + error_classes if type(error_classes) == list else [error_classes])):
+                if any(
+                    str(err) in err_line
+                    for err in (
+                        _PYTHON_ERRORS_ + error_classes
+                        if isinstance(error_classes, list)
+                        else [error_classes]
+                    )
+                ):
                     # start logging error message if target exceptions found
                     start = True
                     output.append(err_line.strip(" :\n") + "\n")