o wOi@sUddlZddlZddlZddlmZmZmZddlmZddlm Z ddl m Z ddl m Z mZmZmZmZmZmZmZmZmZmZdZdZd ZeZeGd d d Zed d d d Zeed<dZ eed<eGdddZ!dZ"eed<e!e"dZ#e!ed<GdddZ$Gdddee Z%eGdddZ&Gddde&Z'eGdd d Z(Gd!d"d"e)e Z*e*j+e*j,e*j-gZ.d#e*d$e/fd%d&Z0d'Z1eed(<eGd)d*d*Z2eGd+d,d,Z3eGd-d.d.Z4eee)e5e/dfZ6ed/d0Gd1d2d2Z7ed3Z8Gd4d5d5ee8Z9Gd6d7d7Z:Gd8d9d9e;ZGdd?d?e;Z@Gd@dAdAe;ZAeZBGdBdCdCe;ZCdDedEedFefdGdHZDdIeBd$eeeeffdJdKZEGdLdMdMej=ZFdS)NN)asdict dataclassfield)datetime)Enum)Template) AnyCallableDictGenericIterableListOptionalTupleTypeTypeVarUnionzS State: ${state} ; Num Restarts: ${num_restarts} Msg: ${msg} Replicas: ${replicas} z - Role: [${role}]: ${replicas} z~ - [${role}:${replica_id}] Timestamp: ${timestamp}; Exit Code: ${exit_code} State: ${state} Error Message: ${error_msg} c@sReZdZUdZeed<eed<eed<eedZe e e fed<e d dd Z d S) Resourcea Represents resource requirements for a ``Container``. Args: cpu: number of cpu cores (note: not hyper threads) gpu: number of gpus memMB: MB of ram capabilities: additional hardware specs (interpreted by scheduler) cpugpumemMBdefault_factory capabilitiesoriginalcKs*t|j}||t|j|j|j|dS)z Copies a resource and applies new capabilities. If the same capabilities are present in the original resource and as parameter, the one from parameter will be used. )rrrr)dictrupdaterrrr)rrres_capabilitiesrO/home/ubuntu/.local/lib/python3.10/site-packages/torchelastic/tsm/driver/api.pycopyFs  z Resource.copyN)rr)__name__ __module__ __qualname____doc__int__annotations__rrrr strr staticmethodr rrrrr5s  r)rrr NULL_RESOURCEallALLc@sdeZdZUdZeed<eZeed<e e dZ e ee fed<deddfddZd e ddfd d Zd S) ContaineraS Represents the specifications of the container that instances of ``Roles`` run on. Maps to the container abstraction that the underlying scheduler supports. This could be an actual container (e.g. Docker) or a physical instance depending on the scheduler. An ``image`` is a software bundle that is installed on a ``Container``. The container on the scheduler dictates what an image actually is. An image could be as simple as a tar-ball or map to a docker image. The scheduler typically knows how to "pull" the image given an image name (str), which could be a simple name (e.g. docker image) or a url (e.g. s3://path/my_image.tar). A ``Resource`` can be bound to a specific scheduler backend or ``SchedulerBackend.ALL`` (default) to specify that the same ``Resource`` is to be used for all schedulers. Usage: :: # define resource for all schedulers my_container = Container(image="pytorch/torch:1") .require(Resource(cpu=1, gpu=1, memMB=500)) .ports(tcp_store=8080, tensorboard=8081) # define resource for a specific scheduler my_container = Container(image="pytorch/torch:1") .require(Resource(cpu=1, gpu=1, memMB=500), "custom_scheduler") .ports(tcp_store=8080, tensorboard=8081) image resourcesrport_mapreturncC ||_|S)z> Sets resource requirements on the container. )r/)selfr/rrrrequireszContainer.requirekwargscKs|ji||S)z7 Adds a port mapping for the container )r0r)r3r5rrrportsszContainer.portsN)r!r"r#r$r'r&r*r/rrrr0r r%r4r6rrrrr-^s  r-z MISSING)r.NULL_CONTAINERc @s>eZdZdZdZdZdZedee de de de fd d Z d S) macrosap Defines macros that can be used with ``Role.entrypoint`` and ``Role.args``. The macros will be substituted at runtime to their actual values. Available macros: 1. ``img_root`` - root directory of the pulled image on the container 2. ``app_id`` - application id as assigned by the scheduler 3. ``replica_id`` - unique id for each instance of a replica of a Role, for instance a role with 3 replicas could have the 0, 1, 2 as replica ids. Note that when the container fails and is replaced, the new container will have the same ``replica_id`` as the one it is replacing. For instance if node 1 failed and was replaced by the scheduler the replacing node will also have ``replica_id=1``. Example: :: # runs: hello_world.py --app_id ${app_id} trainer = Role(name="trainer").runs("hello_world.py", "--app_id", macros.app_id) app = Application("train_app").of(trainer) app_handle = session.run(app, scheduler="local", cfg=RunConfig()) z ${img_root}z ${app_id}z ${replica_id}argsimg_rootapp_id replica_idcCs0g}|D]}t|j|||d}||q|S)N)r;r<r=)rsafe_substituteappend)r:r;r<r=args_subargsubrrr substitutes zmacros.substituteN) r!r"r#r$r;r<r=r(r r'rCrrrrr9s$r9c@seZdZdZdZdZdS) RetryPolicya Defines the retry policy for the ``Roles`` in the ``Application``. The policy defines the behavior when the role replica encounters a failure: 1. unsuccessful (non zero) exit code 2. hardware/host crashes 3. preemption 4. eviction .. note:: Not all retry policies are supported by all schedulers. However all schedulers must support ``RetryPolicy.APPLICATION``. Please refer to the scheduler's documentation for more information on the retry policies they support and behavior caveats (if any). 1. REPLICA: Replaces the replica instance. Surviving replicas are untouched. Use with ``ElasticRole`` to have torchelastic coordinate restarts and membership changes. Otherwise, it is up to the application to deal with failed replica departures and replacement replica admittance. 2. APPLICATION: Restarts the entire application. REPLICA APPLICATIONN)r!r"r#r$rErFrrrrrDsrDc@seZdZUdZeed<eZeed<ee dZ e eed<ee dZ eeefed<eZeed<dZeed <d Zeed <ejZeed <deded eddfddZdeddfddZdeddfddZd ed eddfddZdS)Rolea A set of nodes that perform a specific duty within the ``Application``. Examples: 1. Distributed data parallel app - made up of a single role (trainer). 2. App with parameter server - made up of multiple roles (trainer, ps). Usage: :: trainer = Role(name="trainer") .runs("my_trainer.py", "--arg", "foo", ENV_VAR="FOOBAR") .on(container) .replicas(4) Args: name: name of the role entrypoint: command (within the container) to invoke the role args: commandline arguments to the entrypoint cmd env: environment variable mappings container: container to run in replicas: number of container replicas to run max_retries: max number of retries before giving up retry_policy: retry behavior upon replica failures deployment_preference: hint to the scheduler on how to best deploy and manage replicas of this role name entrypointrr:env container num_replicasr max_retries retry_policyr5r1cOs,||_|jg|7_|ji||SN)rIr:rJrr3rIr:r5rrrruns sz Role.runscCr2rP)rK)r3rKrrronzRole.onreplicascCr2rP)rM)r3rUrrrrUrTz Role.replicascCs||_||_|SrP)rOrN)r3rOrNrrrwith_retry_policyszRole.with_retry_policyN)r!r"r#r$r'r&r7rIrlistr:r rrJr r8rKr-rMr%rNrDrFrOrRrSrUrVrrrrrGs     rGcs@eZdZdZdeffdd Zdedededdfd d ZZS) ElasticRoleac A ``Role`` for which the user provided ``entrypoint`` is executed with the torchelastic agent (in the container). Note that the torchelastic agent invokes multiple copies of ``entrypoint``. For more information about torchelastic see `torchelastic quickstart docs `__. .. important:: It is the responsibility of the user to ensure that the container's image includes torchelastic. Since TSM has no control over the build process of the image, it cannot automatically include torchelastic in the container's image. The following example launches 2 ``replicas`` (nodes) of an elastic ``my_train_script.py`` that is allowed to scale between 2 to 4 nodes. Each node runs 8 workers which are allowed to fail and restart a maximum of 3 times. .. warning:: ``replicas`` MUST BE an integer between (inclusive) ``nnodes``. That is, ``ElasticRole("trainer", nnodes="2:4").replicas(5)`` is invalid and will result in undefined behavior. :: elastic_trainer = ElasticRole("trainer", nproc_per_node=8, nnodes="2:4", max_restarts=3) .runs("my_train_script.py", "--script_arg", "foo", "--another_arg", "bar") .on(container) .replicas(2) # effectively runs: # python -m torchelastic.distributed.launch # --nproc_per_node 8 # --nnodes 2:4 # --max_restarts 3 # my_train_script.py --script_arg foo --another_arg bar rHc stj|dd|_|jddg7_g|_|dd|dtj|d||D]%\}}t |t rD|rC|jd |g7_q-|jd |t |g7_q-dS) NrHpythonz-mztorchelastic.distributed.launch rdzv_backendetcdrdzv_idrolez--) super__init__rIr:torchelastic_launch_args setdefaultr9r<items isinstanceboolr')r3rH launch_kwargsrAval __class__rrr`Cs   zElasticRole.__init__rIr:r5r1cOs`tj|s|tjstjtj|}|j|j7_|j|g|7_|j i||SrP) ospathisabs startswithr9r;joinr:rarJrrQrrrrRUs zElasticRole.runs)r!r"r#r$r'r`rR __classcell__rrrhrrXs$"rXc@sBeZdZUdZeed<eedZe e ed<de ddfddZ dS) Applicationz Represents a distributed application made up of multiple ``Roles``. Contains the necessary information for the driver to submit this app to the scheduler. rHrrolesr1cGs|jg|7_|SrP)rq)r3rqrrrofkszApplication.ofN) r!r"r#r$r'r&rrWrqr rGrrrrrrrp`s rpc@s:eZdZdZdZdZdZdZdZdZ dZ d e fd d Z d S) AppStatea State of the application. An application starts from an initial ``UNSUBMITTED`` state and moves through ``SUBMITTED``, ``PENDING``, ``RUNNING`` states finally reaching a terminal state: ``SUCCEEDED``,``FAILED``, ``CANCELLED``. If the scheduler supports preemption, the app moves from a ``RUNNING`` state to ``PENDING`` upon preemption. If the user stops the application, then the application state moves to ``STOPPED``, then to ``CANCELLED`` when the job is actually cancelled by the scheduler. 1. UNSUBMITTED - app has not been submitted to the scheduler yet 2. SUBMITTED - app has been successfully submitted to the scheduler 3. PENDING - app has been submitted to the scheduler pending allocation 4. RUNNING - app is running 5. SUCCEEDED - app has successfully completed 6. FAILED - app has unsuccessfully completed 7. CANCELLED - app was cancelled before completing rL @r1cC|jSrPrYr3rrr__str__zAppState.__str__N) r!r"r#r$ UNSUBMITTED SUBMITTEDPENDINGRUNNING SUCCEEDEDFAILED CANCELLEDr'r|rrrrrspsrsstater1cCs|tvSrP)_TERMINAL_STATES)rrrr is_terminalsrzNONEc@sheZdZUdZeed<eed<eed<dZe eed<dZ e eed<dZ e eed<d efd d Z dS) RoleReplicaStatusa The status of the replica during the job execution. Args: replica_id: The node rank, note: this is not a worker rank. state: The current state of the node. exit_code: `None`` if still running role: The role name end_time: Timestamp value if the node finished execution, None otherwise error_msg: Error message if any, None if job succeeded. r=rr^N exit_codeend_time error_msgr1cCs&ttj|j|j|j|j|j|jdS)z> Return human readable status representation. ) timestampr=rrr^r) r_REPLICA_FORMAT_TEMPLATErCrr=rrr^rr{rrrget_formatted_strsz#RoleReplicaStatus.get_formatted_str) r!r"r#r$r%r&rsr'rrrrrrrrrrs  rc@seZdZUdZeed<dZeed<dZe ed<e Z e ed<dZ e e ed <eed Zee eefed <d efd dZddZ dded ee eeffddZdded e fddZdS) AppStatusa The runtime status of the ``Application``. The scheduler can return an arbitrary text message (msg field). If any error occurs, scheduler can populate ``structured_error_msg`` with json response. ``replicas`` represent the statuses of the replicas in the job. If the job runs with multiple retries, the parameter will contain the statuses of the most recent retry. Note: if the previous retries failed, but the most recent retry succeeded or in progress, ``replicas`` will not contain ocurred errors. rr num_restartsmsgstructured_error_msgNui_urlrrUr1cCs t|jSrP)rrr{rrrr zAppStatus.is_terminalcCs@t|}|d}|tkrt|}nt}||d<tj|ddS)Nrrt)indent)rpoprjsonloadsdumps)r3app_status_dictrstructured_error_msg_parsedrrr__repr__s  zAppStatus.__repr__state_mask_filtercs2i}|jD]\}}fdd|D||<q|S)Ncs g|] }|jjBkr|qSr)rvalue.0replicarrr s z0AppStatus._get_role_replicas..)rUrc)r3rfilterred_replicasr^ role_replicasrrr_get_role_replicass   zAppStatus._get_role_replicascCstd}||}|D] \}}t|dkrq ddd|D}|ttj||d7}q ttj|j|j |j |dS)zJ Return a human readable representation of the AppStatus. rrcss|]}|VqdSrP)rrrrr s z.AppStatus.get_formatted_str..)r^rU)rrrrU) rrclenrnr_ROLE_REPLICA_FORMAT_TEMPLATErC_APP_STATUS_FORMAT_TEMPLATErrr)r3rrrr^filterred_role_replicas replicas_strrrrrs"     zAppStatus.get_formatted_str)r)r!r"r#r$rsr&rr%rr'rrrrrrrUr r rrerrrrrrrrrs"       rc@seZdZUdZdZeed<ejZ eed<dZ e ed<e Z eed<e Zeed<d Zeeed <eed Zeeed <eed Zeeeefed <d S)DescribeAppResponsea Response object returned by ``Scheduler.describe(app)`` API. Contains the status and description of the application as known by the scheduler. For some schedulers implementations this response object has necessary and sufficient information to recreate an ``Application`` object in the absence of the hosting ``Session``. For these types of schedulers, the user can re-``run()`` the attached application. Otherwise the user can only call non-creating methods (e.g. ``wait()``, ``status()``, etc). Since this class is a data class and contains many member variables we keep the usage simple and provide a no-args constructor and chose to access the member vars directly rather than provide accessors. If scheduler returns arbitrary message, the ``msg`` field should be populated. If scheduler returns a structured json, the ``structured_error_msg`` field should be populated. z r<rr)rrrNrrrqrU)r!r"r#r$r<r'r&rsr~rrr%rrrrrrrWrqr rGrrUr rrrrrrs     "rT)frozenc@sXeZdZUdZeedZeee fe d<dede fddZ ded e fd d Z d d Z dS) RunConfiga Additional run configs for the app. These are typically scheduler runtime configs/arguments that do not bind to ``Application`` nor the ``Scheduler``. For example a particular cluster (within the scheduler) the application should be submitted to. Since the same app can be launched into multiple types of clusters (devo, prod) the cluster id config does not bind to the app. Neither does this bind to the scheduler since the cluster can be partitioned by size of the instances (S, M, L) or by a preemption setting (e.g. on-demand vs spot). Since ``Session`` allows the application to be submitted to multiple schedulers, users who want to submit the same app into multiple schedulers from the same session can union all the ``RunConfig``s into a single object. The scheduler implementation will selectively read the configs it needs. This class is intended to be trivially serialized and passed around or saved hence only allow primitives as config values. Should the scheduler need more than simple primitives (e.g. list of str) it is up to the scheduler to document a way to encode thie value as a str and parse it (e.g. representing list of str as comma delimited str). Usage: :: # write config = RunConfig() config.set("run_as_user", prod") config.set("priority", 10) # read config.get("run_as_user") # "prod" config.get("priority") # 10 config.get("never_set") # None rcfgscfg_keycfg_valcCs||j|<dSrP)r)r3rrrrrsetUz RunConfig.setkeyr1cCs|j|dSrP)rget)r3rrrrrXrz RunConfig.getcCs |jSrP)rrr{rrrr[rzRunConfig.__repr__N)r!r"r#r$rrrr r' ConfigValuer&rrrrrrrr's * rTc@s4eZdZdZdedeegeffddZddZdS) AppDryRunInfoae Returned by ``Scheduler.submit_dryrun``. Represents the request that would have been made to the scheduler. The ``fmt_str()`` method of this object should return a pretty formatted string representation of the underlying request object such that ``print(info)`` yields a human readable representation of the underlying request. requestfmtcCs"||_||_d|_d|_d|_dSrP)r_fmt_app_cfg _scheduler)r3rrrrrr`ls   zAppDryRunInfo.__init__cCs ||jSrP)rrr{rrrr|s zAppDryRunInfo.__repr__N) r!r"r#r$rr r'r`rrrrrrbs rc @sReZdZdZddZ  ddedeeded efd d Zd e fd dZ ddZ dS)runoptsaD Holds the accepted scheduler run configuration keys, default value (if any), and help message string. These options are provided by the ``Scheduler`` and validated in ``Session.run`` against user provided ``RunConfig``. Allows ``None`` default values. Required opts must NOT have a non-None default. .. important:: This class has no accessors because it is intended to be constructed and returned by ``Scheduler.run_config_options`` and printed out as a "help" tool or as part of an exception msg. Usage: :: opts = runopts() opts.add("run_as_user", type_=str, help="user to run the job as") opts.add("cluster_id", type_=int, help="cluster to submit the job", required=True) opts.add("priority", type_=float, default=0.5, help="job priority") opts.add("preemptible", type_=bool, default=False, help="is the job preemptible") # invalid opts.add("illegal", default=10, required=True) opts.add("bad_type", type=str, default=10) opts.check(RunConfig) print(opts) cCs i|_dSrP)_optsr{rrrr`rzrunopts.__init__NFrtype_helpdefaultc Csp|r|durtd|d||dur-t||s-td|d|d|dt|jd ||||f|j|<dS) z Adds the ``config`` option with the given help string and ``default`` value (if any). If the ``default`` is not specified then this option is a required option. NzRequired option: z( must not specify default value. Given: zOption: , must be of type: . Given:  ()) ValueErrorrd TypeErrortyper!r)r3rrrrrequiredrrradds  z runopts.addconfigc Cst|j}|jD]G\}\}}}}||}|r)|dur)td|d|||durIt||sItd|d|jd|dt |jd |||durS| ||q |S) z Checks the given config against this ``runopts`` and sets default configs if not set. .. warning:: This method mutates the provided config! NzRequired run option: z, must be provided and not Nonez Run option: rz , but was: rr) rrr rrcrInvalidRunConfigExceptionrdr!rr) r3r resolved_cfgrrrr_helprgrrrresolves.     zrunopts.resolvec Cszi}|jD])\}\}}}}|rd|n|}d|ji}|r$d|d<n||d<||d<|||<qddl} | j|dd d S) N*rTrrrrrtP)rwidth)rrcr!pprintpformat) r3 pretty_optsrrrrrroptrrrrrs   zrunopts.__repr__)NF) r!r"r#r$r`r'rrrrrrrrrrrs   %rcs.eZdZdZdededdffdd ZZS)rz Raised when the supplied ``RunConfig`` does not satisfy the ``runopts``, either due to missing required configs or value type mismatch. invalid_reason run_configrcs t|d|d|dS)Nrz , Expected: r_r`)r3rrrrhrrr`s z"InvalidRunConfigException.__init__)r!r"r#r$r'rr`rorrrhrrs"rc@s,eZdZdZdefddZdededefdd Ze j d e defd d Z dedede fd dZ dedede fddZdefddZe j dedeefddZdefddZe j deddfddZdeddfddZ    d(dededed eed!eed"eedefd#d$Zded%eddfd&d'ZdS)) Schedulerz An interface abstracting functionalities of a scheduler. Implementors need only implement those methods annotated with ``@abc.abstractmethod``. session_namecC ||_dSrP)r)r3rrrrr`rzScheduler.__init__appcfgr1cCs|||}||S)z Submits the application to be run by the scheduler. Returns: The application id that uniquely identifies the submitted app. ) submit_dryrunschedule)r3rr dryrun_inforrrsubmit s  zScheduler.submitrcCt)ac Same as ``submit`` except that it takes an ``AppDryrunInfo``. Implementors are encouraged to implement this method rather than directly implementing ``submit`` since ``submit`` can be trivially implemented by: :: dryrun_info = self.submit_dryrun(app, cfg) return schedule(dryrun_info) NotImplementedErrorr3rrrrrszScheduler.schedulecCs*||}|||}||_||_|S)a Rather than submitting the request to run the app, returns the request object that would have been submitted to the underlying service. The type of the request object is scheduler dependent. This method can be used to dry-run an application. Please refer to the scheduler implementation's documentation regarding the actual return type. )run_optsr_submit_dryrunrr)r3rrrrrrrr&s  zScheduler.submit_dryruncCrrPr)r3rrrrrr5r}zScheduler._submit_dryruncCstS)z Returns the run configuration options expected by the scheduler. Basically a ``--help`` for the ``run`` API. )rr{rrrr8zScheduler.run_optsr<cCr)z Describes the specified application. Returns: Application description or ``None`` if the app does not exist. rr3r<rrrdescribe?zScheduler.describecCs||}|duS)zf Returns: ``True`` if the app exists (was submitted), ``False`` otherwise N)r)r3r<descrrrexistsIs zScheduler.existsNcCr)zo Kills the application. This method will only be called on an application that exists. rrrrr_cancel_existingQzScheduler._cancel_existingcCs||r ||dSdS)a; Cancels/kills the application. This method is idempotent within the same thread and is safe to call on the same application multiple times. However when called from multiple threads/processes on the same app the exact semantics of this method depends on the idempotency guarantees of the underlying scheduler API. .. note:: This method does not block for the application to reach a cancelled state. To ensure that the application reaches a terminal state use the ``wait`` API. N)rrrrrrcancelYs zScheduler.cancelr role_namekregexsinceuntilcCst|jjd)a Returns an iterator to the log lines of the ``k``th replica of the ``role``. The iterator ends end all qualifying log lines have been read. If the scheduler supports time-based cursors fetching log lines for custom time ranges, then the ``since``, ``until`` fields are honored, otherwise they are ignored. Not specifying ``since`` and ``until`` is equivalent to getting all available log lines. If the ``until`` is empty, then the iterator behaves like ``tail -f``, following the log output until the job reaches a terminal state. The exact definition of what constitutes a log is scheduler specific. Some schedulers may consider stderr or stdout as the log, others may read the logs from a log file. Behaviors and assumptions: 1. Produces an undefined-behavior if called on an app that does not exist The caller should check that the app exists using ``exists(app_id)`` prior to calling this method. 2. Is not stateful, calling this method twice with same parameters returns a new iterator. Prior iteration progress is lost. 3. Does not always support log-tailing. Not all schedulers support live log iteration (e.g. tailing logs while the app is running). Refer to the specific scheduler's documentation for the iterator's behavior. 4. Does not guarantee log retention. It is possible that by the time this method is called, the underlying scheduler may have purged the log records for this application. If so this method raises an arbitrary exception. 5. Only raises a ``StopIteration`` exception when the accessible log lines have been fully exhausted and the app has reached a final state. For instance, if the app gets stuck and does not produce any log lines, then the iterator blocks until the app eventually gets killed (either via timeout or manually) at which point it raises a ``StopIteration``. 6. Need not be supported by all schedulers. 7. Some schedulers may support line cursors by supporting ``__getitem__`` (e.g. ``iter[50]`` seeks to the 50th log line). Returns: An ``Iterator`` over log lines of the specified role replica Raises: NotImplementedError - if the scheduler does not support log iteration z+ does not support application log iteration)rrir#)r3r<rrrrrrrrlog_iterks; zScheduler.log_iter schedulercCs0|jD]}|jjtkrtd|jjdqdS)z Validates whether application is consistent with the scheduler. Raises: ValueError: if application is not compatible with scheduler zNo resources for container: z5. Did you forget to call container.require(resources)N)rqrKr/r*rr.)r3rrr^rrr _validates  zScheduler._validaterNNN)r!r"r#r$r'r`rprrabcabstractmethodrrrrrrrrrrrrr%rr rSchedulerBackendrrrrrrsF   ?rcs&eZdZdZdeffdd ZZS)MalformedAppHandleExceptionz6 Raised when APIs are given a bad app handle. app_handlecst|ddS)NzB is not of the form: :///rr3rrhrrr`sz$MalformedAppHandleException.__init__r!r"r#r$r'r`rorrrhrrsrcs*eZdZdZdedeffdd ZZS)SessionMismatchExceptiona Raised on session certain action APIs when the session_name on an app handle does not match the current session's name. Modify/update APIs raise this exception as modifying/updataing an application owned by a different session should not be allowed. rrcstd|d|ddS)Nz App handle: z is not owned by this session: zT. Please perform the action on the correct session or re-run the app on this sessionr)r3rrrhrrr`sz!SessionMismatchException.__init__rrrrhrrsrcs"eZdZdeffdd ZZS)UnknownSchedulerExceptionscheduler_backendctd|ddS)NzScheduler backend: zQ does not exist. Use session.scheduler_backends() to see all supported schedulersr)r3r rhrrr` z"UnknownSchedulerException.__init__)r!r"r#rr`rorrrhrr sr cs"eZdZdZdfdd ZZS)UnknownAppExceptionz Raised by ``Session`` APIs when either the application does not exist or the application is not owned by the session. r AppHandlecr )NzUnknown app = zp. Did you forget to call session.run()? Otherwise, the app may have already finished and purged by the schedulerrrrhrrr`r zUnknownAppException.__init__)rr)r!r"r#r$r`rorrrhrr sr r rr<cCs|d|d|S)Nz:///r)r rr<rrrmake_app_handlesrrcCsBddl}d}|||}|st||}|d|d|dfS)zX parses the app handle into ```(scheduler_backend, session_name, and app_id)``` rNz?(?P.+)://(?P.+)/(?P.+)r rr<)rematchr groupdict)rrpatternrgdrrrparse_app_handles rc@seZdZdZdefddZdefddZ  d-d ed ed e e de fd dZ e jdede fddZ  d-d ed ed e e defddZe jd ed ed e defddZdeeeffddZe jdeefddZe jde de efddZe jde de efddZe jdee effdd Ze jde dd fd!d"Ze jde de efd#d$Ze j % d.de d&ed'ed(e ed)e ed*e ede fd+d,Z!d S)/Sessiona Entrypoint and client-facing API for TSM. Has the methods for the user to define and act upon ``Applications``. The ``Session`` is stateful and represents a logical workspace of the user. It can be backed by a service (e.g. TSM server) for persistence or can be standalone with no persistence meaning that the ``Session`` lasts only during the duration of the hosting process (see the ``attach()`` API for instructions on re-parenting apps between sessions). rHcCrrP_name)r3rHrrrr`rzSession.__init__r1cCrz)z@ Returns: The name of this session. rr{rrrrHrz Session.namerNrrrcCs||||}||S)a Runs the given application in the specified mode. .. note:: sub-classes of ``Session`` should implement ``schedule`` method rather than overriding this method directly. Returns: An application handle that is used to call other action APIs on the app. Raises: AppNotReRunnableException: if the session/scheduler does not support re-running attached apps )dryrunr)r3rrrrrrrruns z Session.runrcCr)a Actually runs the application from the given dryrun info. Useful when one needs to overwrite a parameter in the scheduler request that is not configurable from one of the object APIs. .. warning:: Use sparingly since abusing this method to overwrite many parameters in the raw scheduler request may lead to your usage of TSM going out of compliance in the long term. This method is intended to unblock the user from experimenting with certain scheduler-specific features in the short term without having to wait until TSM exposes scheduler features in its APIs. .. note:: It is recommended that sub-classes of ``Session`` implement this method instead of directly implementing the ``run`` method. Usage: :: dryrun_info = session.dryrun(app, scheduler="default", cfg) # overwrite parameter "foo" to "bar" dryrun_info.request.foo = "bar" app_handle = session.submit(dryrun_info) rrrrrr1szSession.schedulecCs|js td|jd|jD]*}|jstd|jd|jdkr+td|jd|jtkr9td|jd q||||pBt}||_ |S) a Dry runs an app on the given scheduler with the provided run configs. Does not actually submit the app but rather returns what would have been submitted. The returned ``AppDryRunInfo`` is pretty formatted and can be printed or logged directly. Usage: :: dryrun_info = session.dryrun(app, scheduler="local", cfg) print(dryrun_info) zNo roles for app: z). Did you forget to call app.of(roles..)?zNo entrypoint for role: z:. Did you forget to call role.runs(entrypoint, args, env)?rz Non-positive replicas for role: z8. Did you forget to call role.replicas(positive_number)?zNo container for role: z+. Did you forget to call role.on(container)) rqrrHrIrMrKr8_dryrunrr)r3rrrr^rrrrrRs*       zSession.dryruncCr)z The actual dryrun logic. Implementors of ``Session`` should implement this method rather than ``dryrun``. r)r3rrrrrrrrzSession._dryruncCr)a, Returns the ``runopts`` for the supported scheduler backends. Usage: :: local_runopts = session.run_opts()["local"] print("local scheduler run options: {local_runopts}") Returns: A map of scheduler backend to its ``runopts`` rr{rrrrszSession.run_optscCr)z Returns a list of all supported scheduler backends. All session implementations must support a "default" scheduler backend and document what the default scheduler is. rr{rrrscheduler_backendsrzSession.scheduler_backendsrcCr)z Returns: The status of the application, or ``None`` if the app does not exist anymore (e.g. was stopped in the past and removed from the scheduler's backend). rrrrrstatusszSession.statuscCr)a Block waits (indefinitely) for the application to complete. Possible implementation: :: while(True): app_status = status(app) if app_status.is_terminal(): return sleep(10) Returns: The terminal status of the application, or ``None`` if the app does not exist anymore rrrrrwaitsz Session.waitcCr)z Returns the applications that were run with this session mapped by the app handle. The persistence of the session is implementation dependent. rr{rrrrWrz Session.listcCr)a Stops the application, effectively directing the scheduler to cancel the job. Does nothing if the app does not exist. .. note:: This method returns as soon as the cancel request has been submitted to the scheduler. The application will be in a ``RUNNING`` state until the scheduler actually terminates the job. If the scheduler successfully interrupts the job and terminates it the final state will be ``CANCELLED`` otherwise it will be ``FAILED``. Raises: SessionMismatchException: if the app handle does not belong to this session rrrrrstopsz Session.stopcCr)a Reconstructs the application (to the best extent) given the app handle. Note that the reconstructed application may not be the complete app as it was submitted via the run API. How much of the app can be reconstructed is scheduler dependent. Returns: Application or None if the app does not exist anymore or if the scheduler does not support describing the app handle rrrrrrs zSession.describerrrrrrcCr)a Returns an iterator over the log lines of the specified job container. .. note:: #. ``k`` is the node (host) id NOT the ``rank``. #. ``since`` and ``until`` need not always be honored (depends on scheduler). .. warning:: The semantics and guarantees of the returned iterator is highly scheduler dependent. See ``torchelastic.tsm.driver.api.Scheduler.log_iter`` for the high-level semantics of this log iterator. For this reason it is HIGHLY DISCOURAGED to use this method for generating output to pass to downstream functions/dependencies. This method DOES NOT guarantee that 100% of the log lines are returned. It is totally valid for this method to return no or partial log lines if the scheduler has already totally or partially purged log records for the application. Usage: :: app_handle = session.run(app, scheduler="local", cfg=RunConfig()) print("== trainer node 0 logs ==") for line in session.log_lines(app_handle, "trainer", k=0): print(line) Discouraged anti-pattern: :: # DO NOT DO THIS! # parses accuracy metric from log and reports it for this experiment run accuracy = -1 for line in session.log_lines(app_handle, "trainer", k=0): if matches_regex(line, "final model_accuracy:[0-9]*"): accuracy = parse_accuracy(line) break report(experiment_name, accuracy) Args: app_handle: application handle role_name: role within the app (e.g. trainer) k: k-th replica of the role to fetch the logs for regex: optional regex filter, returns all lines if left empty since: datetime based start cursor. If left empty begins from the first log line (start of job). until: datetime based end cursor. If left empty, follows the log output until the job completes and all log lines have been consumed. Returns: An iterator over the role k-th replica of the specified application. Raise: UnknownAppException: if the app does not exist in the scheduler SessionMismatchException: if the app handle does not belong to this session r)r3rrrrrrrrr log_linessCzSession.log_lines)rNr)"r!r"r#r$r'r`rHrprrrrrrrrrrrr rrr rrrrrWr rr%rr r!rrrrrs   # .    r)Grrrj dataclassesrrrrenumrstringrtypingrr r r r r rrrrrrrrr'rrr*r&r,r-r7r8r9rDrGrXrpr%rsrrrrrerrrrrfloatrrrrr ExceptionrABCrrrr rr rrrrrrrsx    4"  5+=B# "D7v 9