Source code for torch.distributed.elastic.multiprocessing.errors
#!/usr/bin/env python3# Copyright (c) Facebook, Inc. and its affiliates.# All rights reserved.## This source code is licensed under the BSD-style license found in the# LICENSE file in the root directory of this source tree."""Each host in a distributed PyTorch job runs with a single TorchElastic agent,and multiple workers (as children processes of the TorchElastic agent).Since the workers are user-provided (your PyTorch script/job), TorchElastichas a way to propagate errors on the trainers through the agent and up to thescheduler, which ultimately informs the end-user about the state of the joband applies any retry policies.TorchElastic categorizes errors into 3 categories:+----------------+----------------+--------------------------------------------------------------+| Category | Sub-Category | Description |+================+================+==============================================================+| User Error | Input Error | invalid inputs to TorchElastic APIs (e.g. min > max nodes) || +----------------+--------------------------------------------------------------+| | Worker Failure | any failures on the worker child process |+----------------+----------------+--------------------------------------------------------------+| Platform Error | n/a | failures caused by the agent |+----------------+----------------+--------------------------------------------------------------+| Infra Error | n/a | failures outside the domain of the agent and workers || | | (e.g. host failures) |+----------------+----------------+--------------------------------------------------------------+All errors other than "Worker Failure" are either raised canonically from theagent process or implicitly or explicitly crash the agent process. So thestandard language (python) provided exception handling strategies apply.Worker Failures are special because the exception/failure originates on a differentprocess from the agent so the error needs to be propagated inter-process(e.g. the agent cannot simply ``try-catch`` an exception raised on the worker process).TorchElastic agents use :func:`torch.distributed.elastic.multiprocessing.start_processes`to launch the workers which has a simple file based inter-process error propagationbuilt-in.Any function or binary entrypoint decorated with :func:`record`will write uncaught exceptions (with the trace information) to a file specified by theenvironment variable ``TORCHELASTIC_ERROR_FILE``. The parent process (e.g. agent)sets this env var on each child it launches, then aggregates the error files for allchildren, and propagates the one with the **smallest** timestamp (e.g. the **first** error)."""importjsonimportosimportsignalimportsocketimporttimeimportwarningsfromdataclassesimportdataclass,fieldfromdatetimeimportdatetimefromfunctoolsimportwrapsfromstringimportTemplatefromtypingimportAny,Callable,Dict,List,Optional,Tuple,TypeVarfromtorch.distributed.elastic.utils.loggingimportget_loggerfrom.error_handlerimportErrorHandler# noqa: F401from.handlersimportget_error_handler# noqa: F401log=get_logger()JSON=Dict_EMPTY_ERROR_DATA={"message":"<NONE>"}_NOT_AVAILABLE="<N/A>"T=TypeVar("T")
[docs]@dataclassclassProcessFailure:""" Represents the failed process result. When the worker process fails, it may record failure root cause into the file. Tries to read the failure timestamp from the provided ``error_file``, if the ``error_file`` does not exist, the timestamp is the current timestamp (seconds since epoch). The ``message`` field is a concise explanation of the failure. If the error file exists then the message is obtained from the error file. Otherwise one is generated based on the failure signature. .. note:: It is assumed that the ``error_file`` is written by ``torch.distributed.elastic.multiprocessing.errors.error_handler.ErrorHandler``. Otherwise the behavior is undefined. """local_rank:intpid:intexitcode:interror_file:strerror_file_data:JSON=field(init=False)message:str=field(init=False)timestamp:int=field(init=False)def__post_init__(self):self.error_file_data=_EMPTY_ERROR_DATAifos.path.isfile(self.error_file):try:withopen(self.error_file,"r")asfp:self.error_file_data=json.load(fp)log.debug(f"User process failed with error data: {json.dumps(self.error_file_data,indent=2)}")self.message,self.timestamp=self._get_error_data(self.error_file_data)exceptException:log.exception(f"Failed to parse reply file: {self.error_file}")raiseelse:self._set_no_reply_file()# make up an informative message if not already presentifnotself.message:# signals typically do not generate an error file messageifself.exitcode<0:self.message=(f"Signal {-self.exitcode} ({self.signal_name()})"f" received by PID {self.pid}")else:self.message="To enable traceback see: https://pytorch.org/docs/stable/elastic/errors.html"def_get_error_data(self,error_file_data:Dict[str,Any])->Tuple[str,int]:message=error_file_data["message"]ifisinstance(message,str):timestamp=int(error_file_data.get("timestamp",0))else:timestamp=int(message["extraInfo"]["timestamp"])return(message,timestamp)def_set_no_reply_file(self):self.error_file=_NOT_AVAILABLEself.error_file_data=_EMPTY_ERROR_DATAself.message=""self.timestamp=int(time.time())defsignal_name(self)->str:ifself.exitcode<0:returnsignal.Signals(-self.exitcode).nameelse:return_NOT_AVAILABLEdeftimestamp_isoformat(self):""" Returns timestamp in ISO format (YYYY-MM-DD_HH:MM:SS) """returndatetime.fromtimestamp(self.timestamp).isoformat(sep="_")
GlobalRank=int_FAILURE_FORMAT_TEMPLATE="""[${idx}]: time : ${time} host : ${hostname} rank : ${rank} (local_rank: ${local_rank}) exitcode : ${exitcode} (pid: ${pid}) error_file: ${error_file} traceback : ${message}"""# extra new lines before and after are intentional_MSG_FORMAT_TEMPLATE="""${boarder}${title}${section}Failures:${other_failures}${section}Root Cause (first observed failure):${root_failure}${boarder}"""
[docs]classChildFailedError(Exception):""" Special exception type that can be raised from a function annotated with the ``@record`` decorator to have the child process' (root exception) propagate up the stack as-is (e.g. without being wrapped in the parent's traceback). Useful in cases where the parent is a simple nanny process and the child (worker) processes are actually doing meaningful compute. In this case, errors typically occur on the child process as the parent is not doing anything non-trivial, and child errors should be propagated to the scheduler for accurate root cause diagnostics. .. note:: The propagation relies on error files rather than exception handling to support both function and binary launches. Example: :: # process tree on a host (container) 0: scheduler-init-process: |- 1: torchelastic_agent: |- 2: trainer_0 (ok) |- 3: trainer_1 (fail) -> error.json |- ... |- n+2: trainer_n (ok) |- n+3: other processes |- ... In the example above, trainer 1's failure (written into error.json) is the root cause and should be reported to the scheduler's init process. The torchelastic agent raises a ``ChildFailedError("trainer", {1: "trainer_1/error.json"})`` upon detecting trainer 1's failure which would propagate the contents of trainer 1's error file to the scheduler's init process. """def__init__(self,name:str,failures:Dict[GlobalRank,ProcessFailure]):self.name=nameself.failures=failuresassert(self.failures)# does not make sense to create a ChildFaileError with no failuressuper().__init__(self.format_msg())defget_first_failure(self)->Tuple[GlobalRank,ProcessFailure]:rank=min(self.failures.keys(),key=lambdar:self.failures[r].timestamp)returnrank,self.failures[rank]defformat_msg(self,boarder_delim="=",section_delim="-"):title=f"{self.name} FAILED"root_rank,root_failure=self.get_first_failure()root_failure_fmt:str=""other_failures_fmt:List[str]=[]width=len(title)foridx,(rank,failure)inenumerate(self.failures.items()):fmt,w=self._format_failure(idx,rank,failure)width=max(width,w)ifrank==root_rank:root_failure_fmt=fmtelse:other_failures_fmt.append(fmt)# upper boundary on widthwidth=min(width,60)returnTemplate(_MSG_FORMAT_TEMPLATE).substitute(boarder=boarder_delim*width,title=title,section=section_delim*width,root_failure=root_failure_fmt,other_failures="\n".join(other_failures_fmtor[" <NO_OTHER_FAILURES>"]),)def_format_failure(self,idx:int,rank:int,failure:ProcessFailure)->Tuple[str,int]:# failure.message is either a str (when the failure does not generate a traceback - e.g. signals)# or a dict (json) of the form# {"message": $ERROR_MSG, "extraInfo": {"py_callstack": $TRACEBACK, timestamp: $TS}}# so the display logic is:# 1. if failure.message is not a dict (it is a str) just show it as is# 2. else try to get the traceback (py_callstack)# 3. if the traceback is not there, use the message# 4. if the message is not there show <N/A>msg=failure.messageifisinstance(failure.message,dict):msg=(failure.message.get("extraInfo",{}).get("py_callstack",failure.message.get("message","<N/A>")).replace("\n","\n ")# to properly indent the traceback)fmt=Template(_FAILURE_FORMAT_TEMPLATE).substitute(idx=idx,time=failure.timestamp_isoformat(),hostname=socket.getfqdn(),rank=rank,local_rank=failure.local_rank,exitcode=failure.exitcode,pid=failure.pid,error_file=failure.error_file,message=msg,)width=0forlineinfmt.split("\n"):width=max(width,len(line))returnfmt,width
[docs]defrecord(fn:Callable[...,T],error_handler:Optional[ErrorHandler]=None)->Callable[...,T]:""" Syntactic sugar to record errors/exceptions that happened in the decorated function using the provided ``error_handler``. Using this decorator is equivalent to: :: error_handler = get_error_handler() error_handler.initialize() try: foobar() except ChildFailedError as e: _, failure = e.get_first_failure() error_handler.dump_error_file(failure.error_file, failure.exitcode) raise except Exception as e: error_handler.record(e) raise .. important:: use this decorator once per process at the top level method, typically this is the main method. Example :: @record def main(): pass if __name__=="__main__": main() """ifnoterror_handler:error_handler=get_error_handler()defwrap(f):@wraps(f)defwrapper(*args,**kwargs):asserterror_handlerisnotNone# assertion for mypy type checkererror_handler.initialize()try:returnf(*args,**kwargs)exceptChildFailedErrorase:rank,failure=e.get_first_failure()iffailure.error_file!=_NOT_AVAILABLE:error_handler.dump_error_file(failure.error_file,failure.exitcode)else:log.info((f"local_rank {rank} FAILED with no error file."f" Decorate your entrypoint fn with @record for traceback info."f" See: https://pytorch.org/docs/stable/elastic/errors.html"))raiseexceptExceptionase:error_handler.record_exception(e)raisereturnwrapperreturnwrap(fn)
Docs
Access comprehensive developer documentation for PyTorch
To analyze traffic and optimize your experience, we serve cookies on this site. By clicking or navigating, you agree to allow our usage of cookies. As the current maintainers of this site, Facebook’s Cookies Policy applies. Learn more, including about available controls: Cookies Policy.