- class torchx.schedulers.ray_scheduler.RayScheduler(session_name: str)¶
RayScheduler is a TorchX scheduling interface to Ray. The job def workers will be launched as Ray actors
The job environment is specified by the TorchX workspace. Any files in the workspace will be present in the Ray job unless specified in
.torchxignore. Python dependencies will be read from the
requirements.txtfile located at the root of the workspace unless it’s overridden via
usage: [cluster_config_file=CLUSTER_CONFIG_FILE],[cluster_name=CLUSTER_NAME],[dashboard_address=DASHBOARD_ADDRESS],[requirements=REQUIREMENTS] optional arguments: cluster_config_file=CLUSTER_CONFIG_FILE (str, None) Use CLUSTER_CONFIG_FILE to access or create the Ray cluster. cluster_name=CLUSTER_NAME (str, None) Override the configured cluster name. dashboard_address=DASHBOARD_ADDRESS (str, 127.0.0.1:8265) Use ray status to get the dashboard address you will submit jobs against requirements=REQUIREMENTS (str, None) Path to requirements.txt
Partial support. Ray only supports a single log stream so only a dummy “ray/0” combined log role is supported. Tailing and time seeking are not supported.
Partial support. RayScheduler will return job status but does not provide the complete original AppSpec.
Workspaces / Patching
Partial support. Multi role jobs are not supported.
- describe(app_id: str) Optional[DescribeAppResponse] ¶
Describes the specified application.
AppDef description or
Noneif the app does not exist.
- list() List[ListAppResponse] ¶
For apps launched on the scheduler, this API returns a list of ListAppResponse objects each of which have app id and its status. Note: This API is in prototype phase and is subject to change.
- log_iter(app_id: str, role_name: Optional[str] = None, k: int = 0, regex: Optional[str] = None, since: Optional[datetime] = None, until: Optional[datetime] = None, should_tail: bool = False, streams: Optional[Stream] = None) Iterable[str] ¶
Returns an iterator to the log lines of the
k``th replica of the ``role. The iterator ends when all qualifying log lines have been read.
If the scheduler supports time-based cursors fetching log lines for custom time ranges, then the
untilfields are honored, otherwise they are ignored. Not specifying
untilis equivalent to getting all available log lines. If the
untilis 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:
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.
Is not stateful, calling this method twice with same parameters returns a new iterator. Prior iteration progress is lost.
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.
- 3.1 If the scheduler supports log-tailing, it should be controlled
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.
should_tailis True, the method only raises a
StopIterationexception 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
should_tailis False, the method raises
StopIterationwhen there are no more logs.
Need not be supported by all schedulers.
Some schedulers may support line cursors by supporting
iterseeks to the 50th log line).
- Whitespace is preserved, each new line should include
support interactive progress bars the returned lines don’t need to include
\nbut should then be printed without a newline to correctly handle
- Whitespace is preserved, each new line should include
streams – The IO output streams to select. One of: combined, stdout, stderr. If the selected stream isn’t supported by the scheduler it will throw an ValueError.
Iteratorover log lines of the specified role replica
NotImplementedError – if the scheduler does not support log iteration
- schedule(dryrun_info: AppDryRunInfo[RayJob]) str ¶
submitexcept that it takes an
AppDryRunInfo. Implementors are encouraged to implement this method rather than directly implementing
submitcan be trivially implemented by:
dryrun_info = self.submit_dryrun(app, cfg) return schedule(dryrun_info)
- torchx.schedulers.ray_scheduler.has_ray() bool ¶
Indicates whether Ray is installed in the current Python environment.
- torchx.schedulers.ray_scheduler.serialize(actors: List[RayActor], dirpath: str, output_filename: str = 'actors.json') None ¶
- class torchx.schedulers.ray_scheduler.RayJob(app_id: str, working_dir: str, cluster_config_file: ~typing.Optional[str] = None, cluster_name: ~typing.Optional[str] = None, dashboard_address: ~typing.Optional[str] = None, requirements: ~typing.Optional[str] = None, actors: ~typing.List[~torchx.schedulers.ray.ray_common.RayActor] = <factory>)¶
Represents a job that should be run on a Ray cluster.
app_id (str) – The unique ID of the application (a.k.a. job).
cluster_config_file (Optional[str]) – The Ray cluster configuration file.
cluster_name (Optional[str]) – The cluster name to use.
dashboard_address (Optional[str]) – The existing dashboard IP address to connect to
working_dir (str) – The working directory to copy to the cluster
requirements (Optional[str]) – The libraries to install on the cluster per requirements.txt
actors (List[torchx.schedulers.ray.ray_common.RayActor]) – The Ray actors which represent the job to be run. This attribute is dumped to a JSON file and copied to the cluster where ray_main.py uses it to initiate the job.