in src/sagemaker_training/process.py [0:0]
def check_error(cmd, error_classes, processes_per_host, cwd=None, capture_error=True, **kwargs):
"""Run a commmand, raising an exception if there is an error.
Args:
cmd ([str]): The command to be run.
error_classes (list): List of exception classes to watch and raise.
processes_per_host (int): Number of processes per host
capture_error (bool): Whether or not to include stderr in
the exception message (default: True). In either case,
stderr is streamed to the process's output.
**kwargs: Extra arguments that are passed to the subprocess.Popen constructor.
Returns:
subprocess.Popen: The process for the given command.
Raises:
ExecuteUserScriptError: If there is an exception raised when creating the process.
"""
error_classes = process_error_classes(error_classes)
if capture_error:
return_code, output, process = create(
cmd,
error_classes,
processes_per_host,
env=os.environ,
cwd=cwd or environment.code_dir,
capture_error=True,
**kwargs,
)
stderr = " ".join(output)
# remove duplicate while preserve order
stderr = "\n".join(list(dict.fromkeys(stderr.split("\n")))).strip()
else:
stderr = None
# remove extra quotes for subprocess.Popen
cmd[-1] = cmd[-1].strip('"')
process = subprocess.Popen(
cmd,
env=os.environ,
cwd=cwd or environment.code_dir,
stderr=stderr,
**kwargs,
)
return_code = process.wait()
if return_code:
extra_info = None
if return_code == 137:
extra_info = "OutOfMemory: Process killed by SIGKILL (signal 9)"
if stderr and any(path in stderr for path in SM_TRAINING_COMPILER_PATHS):
extra_info = "SMTrainingCompiler Error: Error occurred in SMTrainingCompiler."
# throw internal error classes first
internal_errors = [err for err in dir(errors) if isclass(getattr(errors, err))]
error_class = next(
(name for name in error_classes if name in internal_errors), "ExecuteUserScriptError"
)
error_class = getattr(errors, error_class)
# only replace ExecuteUserScriptError with custom library errors
if stderr and error_class == DEFAULT_ERROR_CLASS:
# find the first target error in stderr
error_name = next((str(name) for name in error_classes if str(name) in stderr), False)
if error_name:
if error_name == "SMTrainingCompilerConfigurationError":
error_class = type(
error_name,
(
errors.SMTrainingCompilerConfigurationError,
), # pylint: disable=protected-access
{},
)
raise error_class(stderr)
error_class = type(
error_name,
(errors._CalledProcessError,), # pylint: disable=protected-access
{},
)
raise error_class(
cmd=" ".join(cmd) if isinstance(cmd, list) else cmd,
return_code=return_code,
output=stderr,
info=extra_info,
)
return process