wehub-resource-sync
1075 lines
39 KiB
Python
1075 lines
39 KiB
Python
import copy
|
|
import json
|
|
import logging
|
|
import os
|
|
import platform
|
|
import re
|
|
import time
|
|
import uuid
|
|
from numbers import Number
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
|
|
|
|
import ray
|
|
import ray.cloudpickle as cloudpickle
|
|
from ray._common.utils import binary_to_hex, hex_to_binary
|
|
from ray.air.constants import (
|
|
EXPR_ERROR_FILE,
|
|
EXPR_ERROR_PICKLE_FILE,
|
|
TRAINING_ITERATION,
|
|
)
|
|
from ray.exceptions import RayActorError, RayTaskError
|
|
from ray.train._internal.checkpoint_manager import _CheckpointManager
|
|
from ray.train._internal.session import _FutureTrainingResult, _TrainingResult
|
|
from ray.train._internal.storage import StorageContext, _exists_at_fs_path
|
|
from ray.train.constants import RAY_TRAIN_COUNT_PREEMPTION_AS_FAILURE
|
|
from ray.tune import Checkpoint, CheckpointConfig
|
|
from ray.tune.error import TuneError
|
|
from ray.tune.execution.placement_groups import (
|
|
PlacementGroupFactory,
|
|
resource_dict_to_pg_factory,
|
|
)
|
|
|
|
# NOTE(rkn): We import ray.tune.registry here instead of importing the names we
|
|
# need because there are cyclic imports that may cause specific names to not
|
|
# have been defined yet. See https://github.com/ray-project/ray/issues/1716.
|
|
from ray.tune.registry import get_trainable_cls, validate_trainable
|
|
from ray.tune.result import (
|
|
DEBUG_METRICS,
|
|
DONE,
|
|
NODE_IP,
|
|
PID,
|
|
STDERR_FILE,
|
|
STDOUT_FILE,
|
|
TRIAL_ID,
|
|
TRIAL_INFO,
|
|
)
|
|
from ray.tune.trainable.metadata import _TrainingRunMetadata
|
|
from ray.tune.utils import date_str, flatten_dict
|
|
from ray.tune.utils.serialization import (
|
|
TuneFunctionEncoder,
|
|
_loads_with_cloudpickle,
|
|
)
|
|
from ray.util import log_once
|
|
from ray.util.annotations import Deprecated, DeveloperAPI
|
|
|
|
DEBUG_PRINT_INTERVAL = 5
|
|
_DEFAULT_WIN_MAX_PATH_LENGTH = 260
|
|
TRIAL_STATE_FILENAME = "trial_metadata.json"
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class _Location:
|
|
"""Describes the location at which Trial is placed to run."""
|
|
|
|
def __init__(self, hostname=None, pid=None):
|
|
self.hostname = hostname
|
|
self.pid = pid
|
|
|
|
def __str__(self):
|
|
if not self.pid:
|
|
return ""
|
|
elif self.hostname == platform.node():
|
|
return "pid={}".format(self.pid)
|
|
else:
|
|
return "{}:{}".format(self.hostname, self.pid)
|
|
|
|
|
|
@DeveloperAPI
|
|
class ExportFormat:
|
|
"""Describes the format to import/export the trial Trainable.
|
|
|
|
This may correspond to different file formats based on the
|
|
Trainable implementation.
|
|
"""
|
|
|
|
CHECKPOINT = "checkpoint"
|
|
MODEL = "model"
|
|
ONNX = "onnx"
|
|
H5 = "h5"
|
|
|
|
@staticmethod
|
|
def validate(formats: List[str]):
|
|
"""Validates formats.
|
|
|
|
Args:
|
|
formats: List of export format strings; each entry is normalized in
|
|
place and checked against the supported set.
|
|
|
|
Raises:
|
|
ValueError: if the format is unknown.
|
|
"""
|
|
for i in range(len(formats)):
|
|
formats[i] = formats[i].strip().lower()
|
|
if formats[i] not in [
|
|
ExportFormat.CHECKPOINT,
|
|
ExportFormat.MODEL,
|
|
ExportFormat.ONNX,
|
|
ExportFormat.H5,
|
|
]:
|
|
raise TuneError("Unsupported import/export format: " + formats[i])
|
|
|
|
|
|
class _TrialInfo:
|
|
"""Serializable struct for holding information for a Trial.
|
|
|
|
Attributes:
|
|
trial_name: String name of the current trial.
|
|
trial_id: trial_id of the trial
|
|
trial_resources: resources used by trial.
|
|
"""
|
|
|
|
def __init__(self, trial: "Trial"):
|
|
"""Initialize ``_TrialInfo`` from a ``Trial``.
|
|
|
|
Args:
|
|
trial: The ``Trial`` whose identifying information should be
|
|
captured into this serializable struct.
|
|
"""
|
|
self._trial_name = str(trial)
|
|
self._trial_id = trial.trial_id
|
|
self._trial_resources = trial.placement_group_factory
|
|
self._experiment_name = trial.experiment_dir_name
|
|
|
|
@property
|
|
def experiment_name(self):
|
|
return self._experiment_name
|
|
|
|
@property
|
|
def trial_name(self):
|
|
return self._trial_name
|
|
|
|
@property
|
|
def trial_id(self):
|
|
return self._trial_id
|
|
|
|
@property
|
|
def trial_resources(self) -> PlacementGroupFactory:
|
|
return self._trial_resources
|
|
|
|
@trial_resources.setter
|
|
def trial_resources(self, new_resources: PlacementGroupFactory):
|
|
self._trial_resources = new_resources
|
|
|
|
|
|
class _TemporaryTrialState:
|
|
"""Temporary trial state.
|
|
|
|
Values saved here should not be restored on resume.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.location = _Location()
|
|
|
|
self.ray_actor: Optional[ray.actor.ActorHandle] = None
|
|
|
|
self.saving_to: Optional[_FutureTrainingResult] = None
|
|
self.restoring_from: Optional[_TrainingResult] = None
|
|
|
|
self.num_restore_failures: int = 0
|
|
|
|
def __getstate__(self):
|
|
return {}
|
|
|
|
|
|
def _get_max_path_length() -> int:
|
|
if hasattr(os, "pathconf"):
|
|
return os.pathconf("/", "PC_PATH_MAX")
|
|
# Windows
|
|
return _DEFAULT_WIN_MAX_PATH_LENGTH
|
|
|
|
|
|
def _create_unique_logdir_name(root: str, relative_logdir: str) -> str:
|
|
candidate = Path(root).expanduser().joinpath(relative_logdir)
|
|
if candidate.exists():
|
|
relative_logdir_old = relative_logdir
|
|
relative_logdir += "_" + uuid.uuid4().hex[:4]
|
|
logger.info(
|
|
f"Creating a new dirname {relative_logdir} because "
|
|
f"trial dirname '{relative_logdir_old}' already exists."
|
|
)
|
|
return relative_logdir
|
|
|
|
|
|
def _get_trainable_kwargs(trial: "Trial") -> Dict[str, Any]:
|
|
trial.init_local_path()
|
|
|
|
trial_config = copy.deepcopy(trial.config)
|
|
trial_config[TRIAL_INFO] = _TrialInfo(trial)
|
|
stdout_file, stderr_file = trial.log_to_file
|
|
trial_config[STDOUT_FILE] = stdout_file
|
|
trial_config[STDERR_FILE] = stderr_file
|
|
|
|
assert trial.storage.trial_dir_name
|
|
|
|
kwargs = {
|
|
"config": trial_config,
|
|
"storage": trial.storage,
|
|
}
|
|
|
|
return kwargs
|
|
|
|
|
|
@DeveloperAPI
|
|
class Trial:
|
|
"""A trial object holds the state for one model training run.
|
|
|
|
Trials are themselves managed by the TrialRunner class, which implements
|
|
the event loop for submitting trial runs to a Ray cluster.
|
|
|
|
Trials start in the PENDING state, and transition to RUNNING once started.
|
|
On error, it transitions to ERROR, otherwise TERMINATED on success.
|
|
|
|
There are resources allocated to each trial. These should be specified
|
|
using ``PlacementGroupFactory``.
|
|
|
|
Attributes:
|
|
trainable_name: Name of the trainable object to be executed.
|
|
config: Provided configuration dictionary with evaluated params.
|
|
trial_id: Unique identifier for the trial.
|
|
path: Path where results for this trial are stored. Can be on
|
|
the local node or on cloud storage.
|
|
local_path: Path on the local disk where results are stored.
|
|
remote_path: Path on cloud storage where results are stored,
|
|
or None if not set.
|
|
relative_logdir: Directory of the trial relative to its
|
|
experiment directory.
|
|
evaluated_params: Evaluated parameters by search algorithm,
|
|
experiment_tag: Identifying trial name to show in the console
|
|
status: One of PENDING, RUNNING, PAUSED, TERMINATED, ERROR/
|
|
error_file: Path to the errors that this trial has raised.
|
|
|
|
"""
|
|
|
|
_nonjson_fields = [
|
|
"results",
|
|
"extra_arg",
|
|
"placement_group_factory",
|
|
"_resources",
|
|
"_default_placement_group_factory",
|
|
]
|
|
|
|
PENDING = "PENDING"
|
|
RUNNING = "RUNNING"
|
|
PAUSED = "PAUSED"
|
|
TERMINATED = "TERMINATED"
|
|
ERROR = "ERROR"
|
|
|
|
def __init__(
|
|
self,
|
|
trainable_name: str,
|
|
*,
|
|
config: Optional[Dict] = None,
|
|
trial_id: Optional[str] = None,
|
|
storage: Optional[StorageContext] = None,
|
|
evaluated_params: Optional[Dict] = None,
|
|
experiment_tag: str = "",
|
|
placement_group_factory: Optional[PlacementGroupFactory] = None,
|
|
stopping_criterion: Optional[Dict[str, float]] = None,
|
|
checkpoint_config: Optional[CheckpointConfig] = None,
|
|
export_formats: Optional[List[str]] = None,
|
|
restore_path: Optional[str] = None,
|
|
trial_name_creator: Optional[Callable[["Trial"], str]] = None,
|
|
trial_dirname_creator: Optional[Callable[["Trial"], str]] = None,
|
|
log_to_file: Union[Optional[str], Tuple[Optional[str], Optional[str]]] = None,
|
|
max_failures: int = 0,
|
|
stub: bool = False,
|
|
_setup_default_resource: bool = True,
|
|
):
|
|
"""Initialize a new trial.
|
|
|
|
The args here take the same meaning as the command line flags defined
|
|
in ray.tune.experiment.config_parser.
|
|
|
|
Args:
|
|
trainable_name: Name of the registered trainable to execute.
|
|
config: Hyperparameter configuration for this trial.
|
|
trial_id: Unique identifier for this trial. Auto-generated if not
|
|
provided.
|
|
storage: ``StorageContext`` describing where trial results and
|
|
checkpoints are persisted.
|
|
evaluated_params: Parameters chosen by the search algorithm, used
|
|
for display and configuration export.
|
|
experiment_tag: Identifying trial name to show in the console.
|
|
placement_group_factory: Resource specification for the trial as a
|
|
``PlacementGroupFactory`` (or dict that will be converted to one).
|
|
stopping_criterion: Mapping of metric name to threshold value that
|
|
stops the trial once exceeded.
|
|
checkpoint_config: ``CheckpointConfig`` controlling how trial
|
|
checkpoints are saved and rotated.
|
|
export_formats: List of formats (e.g. ``["model"]``) to export at
|
|
the end of the trial.
|
|
restore_path: Path to a checkpoint directory used to seed the
|
|
trial's initial state.
|
|
trial_name_creator: Optional callable that returns a display name
|
|
for the trial.
|
|
trial_dirname_creator: Optional callable that returns the trial
|
|
directory name relative to the experiment directory.
|
|
log_to_file: Either a single path or a ``(stdout, stderr)`` pair of
|
|
paths to which the trial's stdout/stderr should be written.
|
|
max_failures: Number of times to retry the trial from its latest
|
|
checkpoint before giving up.
|
|
stub: If True, skip trainable validation. Used when building
|
|
``Trial`` objects from on-disk checkpoints for inspection only.
|
|
_setup_default_resource: Whether to set up default resources.
|
|
When initializing trials from checkpoints, this field is set to false,
|
|
so that setting up default resources can be delayed till after
|
|
``trial.config`` is loaded from checkpoints.
|
|
"""
|
|
# If this is set, trainables are not validated or looked up.
|
|
# This can be used e.g. to initialize Trial objects from checkpoints
|
|
# without loading the trainable first.
|
|
self.stub = stub
|
|
|
|
if not self.stub:
|
|
validate_trainable(trainable_name)
|
|
# Trial config
|
|
self.trainable_name = trainable_name
|
|
self.trial_id = Trial.generate_id() if trial_id is None else trial_id
|
|
|
|
self.temporary_state = _TemporaryTrialState()
|
|
self.run_metadata = _TrainingRunMetadata()
|
|
|
|
# Create a copy, since `init_local_path` updates the context with the
|
|
# generated trial dirname.
|
|
self.storage = copy.copy(storage)
|
|
|
|
self.config = config or {}
|
|
# Save a copy of the original unresolved config so that we can swap
|
|
# out and update any reference config values after restoration.
|
|
self.__unresolved_config = self.config
|
|
|
|
# Parameters that Tune varies across searches.
|
|
self.evaluated_params = evaluated_params or {}
|
|
self.experiment_tag = experiment_tag
|
|
self.stopping_criterion = stopping_criterion or {}
|
|
|
|
self._setup_default_resource = _setup_default_resource
|
|
|
|
if placement_group_factory and not isinstance(
|
|
placement_group_factory, PlacementGroupFactory
|
|
):
|
|
placement_group_factory = resource_dict_to_pg_factory(
|
|
placement_group_factory
|
|
)
|
|
|
|
self._default_placement_group_factory = placement_group_factory
|
|
# Will be created in create_placement_group_factory().
|
|
self.placement_group_factory = None
|
|
|
|
self.log_to_file = log_to_file
|
|
# Make sure `stdout_file, stderr_file = Trial.log_to_file` works
|
|
if (
|
|
not self.log_to_file
|
|
or not isinstance(self.log_to_file, Sequence)
|
|
or not len(self.log_to_file) == 2
|
|
):
|
|
self.log_to_file = (None, None)
|
|
|
|
self.max_failures = max_failures
|
|
|
|
# Local trial state that is updated during the run
|
|
self._default_result_or_future: Union[ray.ObjectRef, dict, None] = None
|
|
|
|
self.export_formats = export_formats
|
|
self.status = Trial.PENDING
|
|
self.relative_logdir = None
|
|
|
|
self.trial_name_creator = trial_name_creator
|
|
self.trial_dirname_creator = trial_dirname_creator
|
|
self.custom_trial_name = None
|
|
self.custom_dirname = None
|
|
|
|
# Checkpoint config
|
|
checkpoint_config = checkpoint_config or CheckpointConfig()
|
|
|
|
self.run_metadata.checkpoint_manager = _CheckpointManager(
|
|
checkpoint_config=checkpoint_config
|
|
)
|
|
|
|
# Restoration fields
|
|
self.restore_path = restore_path
|
|
self._restore_checkpoint_result: Optional[_TrainingResult] = None
|
|
if restore_path:
|
|
# tune.run(restore) passes in a path without metrics.
|
|
self._restore_checkpoint_result = _TrainingResult(
|
|
checkpoint=Checkpoint.from_directory(restore_path), metrics={}
|
|
)
|
|
|
|
if trial_name_creator:
|
|
self.custom_trial_name = trial_name_creator(self)
|
|
|
|
if trial_dirname_creator:
|
|
self.custom_dirname = trial_dirname_creator(self)
|
|
if os.path.sep in self.custom_dirname:
|
|
raise ValueError(
|
|
f"Trial dirname must not contain '/'. Got {self.custom_dirname}"
|
|
)
|
|
|
|
self._state_json = None
|
|
|
|
def create_placement_group_factory(self):
|
|
"""Compute placement group factory if needed.
|
|
|
|
Note: this must be called after all the placeholders in
|
|
self.config are resolved.
|
|
"""
|
|
trainable_cls = self.get_trainable_cls()
|
|
if not trainable_cls or not self._setup_default_resource:
|
|
# Create placement group factory using default resources.
|
|
self.placement_group_factory = (
|
|
self._default_placement_group_factory or resource_dict_to_pg_factory()
|
|
)
|
|
return
|
|
|
|
default_resources = trainable_cls.default_resource_request(self.config)
|
|
|
|
# If Trainable returns resources, do not allow manual override via
|
|
# `resources_per_trial` by the user.
|
|
if default_resources and self._default_placement_group_factory:
|
|
raise TuneError(
|
|
"Resources for {} have been automatically set to {} "
|
|
"by its `default_resource_request()` method. Please "
|
|
"clear the `resources_per_trial` option.".format(
|
|
trainable_cls, default_resources
|
|
)
|
|
)
|
|
|
|
if default_resources and not isinstance(
|
|
default_resources, PlacementGroupFactory
|
|
):
|
|
default_resources = resource_dict_to_pg_factory(default_resources)
|
|
|
|
self.placement_group_factory = (
|
|
# default_resource_request
|
|
default_resources
|
|
# resources_per_trial
|
|
or self._default_placement_group_factory
|
|
# cpu=1
|
|
or resource_dict_to_pg_factory()
|
|
)
|
|
|
|
def _get_default_result_or_future(self) -> Optional[dict]:
|
|
"""Calls ray.get on self._default_result_or_future and assigns back.
|
|
|
|
Returns None in case of exceptions.
|
|
Will also set the trial location if runner is set.
|
|
"""
|
|
if self._default_result_or_future and isinstance(
|
|
self._default_result_or_future, ray.ObjectRef
|
|
):
|
|
try:
|
|
self._default_result_or_future = ray.get(self._default_result_or_future)
|
|
except RayActorError: # error during initialization
|
|
self._default_result_or_future = None
|
|
if self._default_result_or_future and self.temporary_state.ray_actor:
|
|
self.set_location(
|
|
_Location(
|
|
self._default_result_or_future.get(NODE_IP),
|
|
self._default_result_or_future.get(PID),
|
|
)
|
|
)
|
|
return self._default_result_or_future
|
|
|
|
def resolve_config_placeholders(self, placeholder_resolvers: Dict[Tuple, Any]):
|
|
from ray.tune.impl.placeholder import resolve_placeholders
|
|
|
|
# Make a copy of the unresolved config before resolve it.
|
|
self.config = copy.deepcopy(self.__unresolved_config)
|
|
resolve_placeholders(self.config, placeholder_resolvers)
|
|
|
|
@property
|
|
def last_result(self) -> dict:
|
|
# The logic in here is as follows:
|
|
# 1. If the trial has reported at least once, last_result would have
|
|
# been set and therefore would not be empty. We can just return it.
|
|
# 2. If the trial has not reported at least once but we have the
|
|
# future for the default results dict, (obtained through
|
|
# Trainable.get_auto_filled_metrics), we get that future
|
|
# and return it.
|
|
# 3. In the worst case where we have nothing, we just set the
|
|
# trial_id and return that.
|
|
result = self.run_metadata.last_result
|
|
if not {k for k in result if k != TRIAL_ID}:
|
|
self._get_default_result_or_future()
|
|
result = self._default_result_or_future or result
|
|
result.setdefault(TRIAL_ID, self.trial_id)
|
|
return result
|
|
|
|
@property
|
|
def metric_analysis(self):
|
|
return self.run_metadata.metric_analysis
|
|
|
|
@property
|
|
def metric_n_steps(self):
|
|
return self.run_metadata.metric_n_steps
|
|
|
|
def get_ray_actor_ip(self) -> Optional[str]:
|
|
if self.temporary_state.location.hostname:
|
|
return self.temporary_state.location.hostname
|
|
|
|
if not self.temporary_state.ray_actor:
|
|
return None
|
|
|
|
hostname, pid = ray.get(
|
|
self.temporary_state.ray_actor.get_current_ip_pid.remote()
|
|
)
|
|
self.temporary_state.location = _Location(hostname, pid)
|
|
return self.temporary_state.location.hostname
|
|
|
|
@property
|
|
@Deprecated("Replaced by `local_experiment_path`")
|
|
def local_dir(self):
|
|
return self.local_experiment_path
|
|
|
|
@property
|
|
def experiment_dir_name(self):
|
|
return self.storage.experiment_dir_name
|
|
|
|
@property
|
|
def remote_experiment_path(self) -> str:
|
|
return self.storage.experiment_fs_path
|
|
|
|
@property
|
|
def local_experiment_path(self) -> str:
|
|
return self.storage.experiment_driver_staging_path
|
|
|
|
@property
|
|
@Deprecated("Replaced by `local_path`")
|
|
def logdir(self) -> Optional[str]:
|
|
# TODO(justinvyu): [Deprecated] Remove in 2.11.
|
|
raise DeprecationWarning("Use `local_path` instead of `logdir`.")
|
|
|
|
@property
|
|
def local_path(self) -> Optional[str]:
|
|
return self.storage.trial_driver_staging_path
|
|
|
|
@property
|
|
def path(self) -> Optional[str]:
|
|
return self.storage.trial_fs_path
|
|
|
|
@property
|
|
def has_reported_at_least_once(self) -> bool:
|
|
return bool(self.run_metadata.last_result)
|
|
|
|
@property
|
|
def node_ip(self):
|
|
return self.temporary_state.location.hostname
|
|
|
|
@property
|
|
def checkpoint_at_end(self):
|
|
config = self.run_metadata.checkpoint_manager.checkpoint_config
|
|
return config.checkpoint_at_end
|
|
|
|
@property
|
|
def checkpoint_freq(self):
|
|
config = self.run_metadata.checkpoint_manager.checkpoint_config
|
|
return config.checkpoint_frequency
|
|
|
|
@property
|
|
def latest_checkpoint_result(self) -> Optional[_TrainingResult]:
|
|
# NOTE: Fallback to the checkpoint passed in from `tune.run(restore)`
|
|
# if the trial hasn't saved any checkpoints itself yet.
|
|
return (
|
|
self.run_metadata.checkpoint_manager.latest_checkpoint_result
|
|
or self._restore_checkpoint_result
|
|
)
|
|
|
|
@property
|
|
def checkpoint(self) -> Optional[Checkpoint]:
|
|
"""Returns the most recent checkpoint if one has been saved."""
|
|
return (
|
|
self.latest_checkpoint_result.checkpoint
|
|
if self.latest_checkpoint_result
|
|
else None
|
|
)
|
|
|
|
@classmethod
|
|
def generate_id(cls):
|
|
return str(uuid.uuid4().hex)[:8]
|
|
|
|
def reset(self) -> "Trial":
|
|
# If there is `default_resource_request` associated with the trainable,
|
|
# clear `resources` and `placement_group_factory`.
|
|
# This is mainly relevant for RLlib tuning jobs, where we save users
|
|
# of the trouble to specify the resources themselves by having some
|
|
# default resources for popular RLlib algorithms.
|
|
trainable_cls = self.get_trainable_cls()
|
|
clear_resources = trainable_cls and trainable_cls.default_resource_request(
|
|
self.config
|
|
)
|
|
placement_group_factory = (
|
|
self.placement_group_factory if not clear_resources else None
|
|
)
|
|
|
|
checkpoint_config = self.run_metadata.checkpoint_manager.checkpoint_config
|
|
return Trial(
|
|
self.trainable_name,
|
|
config=self.config,
|
|
trial_id=None,
|
|
evaluated_params=self.evaluated_params,
|
|
experiment_tag=self.experiment_tag,
|
|
placement_group_factory=placement_group_factory,
|
|
stopping_criterion=self.stopping_criterion,
|
|
checkpoint_config=checkpoint_config,
|
|
export_formats=self.export_formats,
|
|
restore_path=self.restore_path,
|
|
trial_name_creator=self.trial_name_creator,
|
|
trial_dirname_creator=self.trial_dirname_creator,
|
|
log_to_file=self.log_to_file,
|
|
max_failures=self.max_failures,
|
|
storage=self.storage,
|
|
)
|
|
|
|
@Deprecated("Replaced by `init_local_path()`")
|
|
def init_logdir(self):
|
|
# TODO(justinvyu): [Deprecated] Remove in 2.11.
|
|
raise DeprecationWarning("Use `init_local_path` instead of `init_logdir`.")
|
|
|
|
def init_local_path(self):
|
|
"""Init logdir."""
|
|
if not self.relative_logdir:
|
|
self.relative_logdir = _create_unique_logdir_name(
|
|
str(self.local_experiment_path), self._generate_dirname()
|
|
)
|
|
# Populate the storage context with the trial dir name we just generated.
|
|
self.storage.trial_dir_name = self.relative_logdir
|
|
|
|
assert self.local_path
|
|
logdir_path = Path(self.local_path)
|
|
max_path_length = _get_max_path_length()
|
|
if len(str(logdir_path)) >= max_path_length:
|
|
logger.warning(
|
|
f"The path to the trial log directory is too long "
|
|
f"(max length: {max_path_length}. "
|
|
f"Consider using `trial_dirname_creator` to shorten the path. "
|
|
f"Path: {logdir_path}"
|
|
)
|
|
logdir_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
self.invalidate_json_state()
|
|
|
|
def update_resources(self, resources: Union[dict, PlacementGroupFactory]):
|
|
"""EXPERIMENTAL: Updates the resource requirements.
|
|
|
|
Should only be called when the trial is not running.
|
|
|
|
Args:
|
|
resources: New resource requirements for the trial; a ``dict``
|
|
will be converted into a ``PlacementGroupFactory``.
|
|
|
|
Raises:
|
|
ValueError: if trial status is running.
|
|
"""
|
|
if self.status is Trial.RUNNING:
|
|
raise ValueError("Cannot update resources while Trial is running.")
|
|
|
|
placement_group_factory = resources
|
|
if isinstance(resources, dict):
|
|
placement_group_factory = resource_dict_to_pg_factory(resources)
|
|
|
|
self.placement_group_factory = placement_group_factory
|
|
|
|
self.invalidate_json_state()
|
|
|
|
def set_ray_actor(self, ray_actor):
|
|
self.temporary_state.ray_actor = ray_actor
|
|
if ray_actor:
|
|
# Do not block here, the result will be gotten when last_result
|
|
# property is accessed
|
|
self._default_result_or_future = ray_actor.get_auto_filled_metrics.remote(
|
|
debug_metrics_only=True
|
|
)
|
|
|
|
def set_location(self, location):
|
|
"""Sets the location of the trial."""
|
|
self.temporary_state.location = location
|
|
|
|
def set_status(self, status):
|
|
"""Sets the status of the trial."""
|
|
self.status = status
|
|
if status == Trial.RUNNING:
|
|
if self.run_metadata.start_time is None:
|
|
self.run_metadata.start_time = time.time()
|
|
self.invalidate_json_state()
|
|
|
|
def set_config(self, config):
|
|
self.config = config
|
|
self.invalidate_json_state()
|
|
|
|
def set_experiment_tag(self, experiment_tag):
|
|
self.experiment_tag = experiment_tag
|
|
self.invalidate_json_state()
|
|
|
|
def set_storage(self, new_storage: StorageContext):
|
|
"""Updates the storage context of the trial.
|
|
|
|
If the `storage_path` or `experiment_dir_name` has changed, then this setter
|
|
also updates the paths of all checkpoints tracked by the checkpoint manager.
|
|
This enables restoration from a checkpoint if the user moves the directory.
|
|
"""
|
|
original_storage = self.storage
|
|
|
|
checkpoint_manager = self.run_metadata.checkpoint_manager
|
|
|
|
for checkpoint_result in checkpoint_manager.best_checkpoint_results:
|
|
checkpoint_result.checkpoint = Checkpoint(
|
|
path=checkpoint_result.checkpoint.path.replace(
|
|
original_storage.trial_fs_path, new_storage.trial_fs_path, 1
|
|
),
|
|
filesystem=new_storage.storage_filesystem,
|
|
)
|
|
latest_checkpoint_result = checkpoint_manager.latest_checkpoint_result
|
|
if latest_checkpoint_result:
|
|
latest_checkpoint_result.checkpoint = Checkpoint(
|
|
path=latest_checkpoint_result.checkpoint.path.replace(
|
|
original_storage.trial_fs_path, new_storage.trial_fs_path, 1
|
|
),
|
|
filesystem=new_storage.storage_filesystem,
|
|
)
|
|
|
|
self.storage = new_storage
|
|
self.invalidate_json_state()
|
|
|
|
@property
|
|
def num_failures(self):
|
|
return self.run_metadata.num_failures
|
|
|
|
@property
|
|
def num_failures_after_restore(self):
|
|
return self.run_metadata.num_failures_after_restore
|
|
|
|
@property
|
|
def error_file(self):
|
|
if not self.local_path or not self.run_metadata.error_filename:
|
|
return None
|
|
return Path(self.local_path, self.run_metadata.error_filename).as_posix()
|
|
|
|
@property
|
|
def pickled_error_file(self):
|
|
if not self.local_path or not self.run_metadata.pickled_error_filename:
|
|
return None
|
|
return Path(
|
|
self.local_path, self.run_metadata.pickled_error_filename
|
|
).as_posix()
|
|
|
|
def get_pickled_error(self) -> Optional[Exception]:
|
|
"""Returns the pickled error object if it exists in storage.
|
|
|
|
This is a pickled version of the latest error that the trial encountered.
|
|
"""
|
|
error_filename = self.run_metadata.pickled_error_filename
|
|
if error_filename is None:
|
|
return None
|
|
|
|
fs = self.storage.storage_filesystem
|
|
pickled_error_fs_path = Path(
|
|
self.storage.trial_fs_path, error_filename
|
|
).as_posix()
|
|
|
|
if _exists_at_fs_path(fs=fs, fs_path=pickled_error_fs_path):
|
|
with fs.open_input_stream(pickled_error_fs_path) as f:
|
|
return cloudpickle.loads(f.readall())
|
|
return None
|
|
|
|
def get_error(self) -> Optional[TuneError]:
|
|
"""Returns the error text file trace as a TuneError object
|
|
if it exists in storage.
|
|
|
|
This is a text trace of the latest error that the trial encountered,
|
|
which is used in the case that the error is not picklable.
|
|
"""
|
|
error_filename = self.run_metadata.error_filename
|
|
if error_filename is None:
|
|
return None
|
|
|
|
fs = self.storage.storage_filesystem
|
|
txt_error_fs_path = Path(self.storage.trial_fs_path, error_filename).as_posix()
|
|
|
|
if _exists_at_fs_path(fs=fs, fs_path=txt_error_fs_path):
|
|
with fs.open_input_stream(txt_error_fs_path) as f:
|
|
return f.readall().decode()
|
|
return None
|
|
|
|
def _handle_restore_error(self, exc: Exception):
|
|
# For Restoration errors, we only increment the restore failure count
|
|
# if the number of failures exceeds the restore retry limit.
|
|
if self.temporary_state.num_restore_failures >= int(
|
|
os.environ.get("TUNE_RESTORE_RETRY_NUM", 0)
|
|
):
|
|
self.run_metadata.num_failures += 1
|
|
else:
|
|
self.temporary_state.num_restore_failures += 1
|
|
|
|
def _handle_ray_actor_error(self, exc: RayActorError):
|
|
count_preemption_errors = bool(
|
|
int(os.environ.get(RAY_TRAIN_COUNT_PREEMPTION_AS_FAILURE, "0"))
|
|
)
|
|
if not exc.preempted or count_preemption_errors:
|
|
# Only count non-preempted actor errors as failures.
|
|
self.run_metadata.num_failures += 1
|
|
|
|
def _handle_ray_task_error(self, exc: RayTaskError):
|
|
cause = exc.as_instanceof_cause()
|
|
if isinstance(cause, RayActorError):
|
|
# Handle the RayActorError directly (ex: Ray Train worker actor errors)
|
|
return self._handle_ray_actor_error(cause)
|
|
|
|
# Increment failures for all user errors (which get raised as RayTaskError)
|
|
self.run_metadata.num_failures += 1
|
|
|
|
def handle_error(
|
|
self, exc: Optional[Union[TuneError, RayTaskError, RayActorError]] = None
|
|
):
|
|
if self.is_restoring:
|
|
self._handle_restore_error(exc)
|
|
elif isinstance(exc, RayActorError):
|
|
self._handle_ray_actor_error(exc)
|
|
elif isinstance(exc, RayTaskError):
|
|
self._handle_ray_task_error(exc)
|
|
else:
|
|
self.run_metadata.num_failures += 1
|
|
|
|
if self.local_path:
|
|
self.run_metadata.error_filename = EXPR_ERROR_FILE
|
|
if isinstance(exc, (RayTaskError, RayActorError)):
|
|
# Piping through the actual error to result grid.
|
|
self.run_metadata.pickled_error_filename = EXPR_ERROR_PICKLE_FILE
|
|
with open(self.pickled_error_file, "wb") as f:
|
|
cloudpickle.dump(exc, f)
|
|
with open(self.error_file, "a+") as f:
|
|
f.write(
|
|
"Failure # {} (occurred at {})\n".format(
|
|
self.run_metadata.num_failures, date_str()
|
|
)
|
|
)
|
|
f.write(str(exc) + "\n")
|
|
self.run_metadata.invalidate_cache()
|
|
|
|
def should_stop(self, result):
|
|
"""Whether the given result meets this trial's stopping criteria."""
|
|
if result.get(DONE):
|
|
return True
|
|
|
|
for criterion, stop_value in self.stopping_criterion.items():
|
|
if isinstance(criterion, dict):
|
|
raise ValueError(
|
|
"Stopping criteria is now flattened by default. "
|
|
"Use forward slashes to nest values `key1/key2/key3`."
|
|
)
|
|
elif criterion not in result:
|
|
if log_once("tune_trial_stop_criterion_not_found"):
|
|
logger.warning(
|
|
f"Stopping criterion '{criterion}' not found in result dict! "
|
|
f"Available keys are {list(result.keys())}. If '{criterion}' is"
|
|
" never reported, the run will continue until training is "
|
|
"finished."
|
|
)
|
|
elif result[criterion] >= stop_value:
|
|
return True
|
|
return False
|
|
|
|
def should_checkpoint(self):
|
|
"""Whether this trial is due for checkpointing."""
|
|
result = self.last_result or {}
|
|
if result.get(DONE) and self.checkpoint_at_end:
|
|
return True
|
|
return (
|
|
self.checkpoint_freq
|
|
and result.get(TRAINING_ITERATION, 0) % self.checkpoint_freq == 0
|
|
)
|
|
|
|
def has_checkpoint(self) -> bool:
|
|
return self.checkpoint is not None
|
|
|
|
def on_checkpoint(self, checkpoint_result: _TrainingResult):
|
|
"""Hook for handling checkpoints taken by the Trainable.
|
|
|
|
Args:
|
|
checkpoint_result: Training result containing the checkpoint taken.
|
|
"""
|
|
self.run_metadata.checkpoint_manager.register_checkpoint(checkpoint_result)
|
|
# Update the checkpoint index to keep the checkpoint index in sync.
|
|
# This index will get restored when the trial is restored and will
|
|
# be passed to the Trainable as the starting checkpoint index.
|
|
self.storage._update_checkpoint_index(checkpoint_result.metrics)
|
|
|
|
self.invalidate_json_state()
|
|
self.run_metadata.invalidate_cache()
|
|
|
|
def on_restore(self):
|
|
"""Handles restoration completion."""
|
|
assert self.is_restoring
|
|
self.run_metadata.last_result = self.temporary_state.restoring_from.metrics
|
|
self.run_metadata.last_result.setdefault("config", self.config)
|
|
self.temporary_state.restoring_from = None
|
|
self.temporary_state.num_restore_failures = 0
|
|
|
|
def should_recover(self):
|
|
"""Returns whether the trial qualifies for retrying.
|
|
|
|
`num_failures` should represent the number of times the trial has
|
|
failed *up to the moment this method is called.* If we've failed
|
|
5 times and `max_failures=5`, then we should recover, since
|
|
we only pass the limit on the 6th failure.
|
|
|
|
Note this may return true even when there is no checkpoint, either because
|
|
`self.checkpoint_freq` is `0` or because the trial failed before
|
|
a checkpoint has been made.
|
|
"""
|
|
return (
|
|
self.run_metadata.num_failures <= self.max_failures or self.max_failures < 0
|
|
)
|
|
|
|
def update_last_result(self, result):
|
|
if self.experiment_tag:
|
|
result.update(experiment_tag=self.experiment_tag)
|
|
|
|
self.set_location(_Location(result.get(NODE_IP), result.get(PID)))
|
|
self.run_metadata.last_result = result
|
|
self.run_metadata.last_result_time = time.time()
|
|
|
|
metric_result = self.last_result.copy()
|
|
for remove_metric in DEBUG_METRICS:
|
|
metric_result.pop(remove_metric, None)
|
|
|
|
for metric, value in flatten_dict(metric_result).items():
|
|
if isinstance(value, Number):
|
|
self.run_metadata.update_metric(
|
|
metric, value, step=result.get("training_iteration")
|
|
)
|
|
|
|
def get_trainable_cls(self):
|
|
if self.stub:
|
|
return None
|
|
return get_trainable_cls(self.trainable_name)
|
|
|
|
def is_finished(self):
|
|
return self.status in [Trial.ERROR, Trial.TERMINATED]
|
|
|
|
@property
|
|
def is_restoring(self):
|
|
return self.temporary_state.restoring_from is not None
|
|
|
|
@property
|
|
def is_saving(self):
|
|
return self.temporary_state.saving_to is not None
|
|
|
|
def __repr__(self):
|
|
return self._trainable_name(include_trial_id=True)
|
|
|
|
def __str__(self):
|
|
return self._trainable_name(include_trial_id=True)
|
|
|
|
def _trainable_name(self, include_trial_id=False):
|
|
"""Combines ``env`` with ``trainable_name`` and ``trial_id``.
|
|
|
|
Can be overridden with a custom string creator.
|
|
"""
|
|
if self.custom_trial_name:
|
|
return self.custom_trial_name
|
|
|
|
if "env" in self.config:
|
|
env = self.config["env"]
|
|
if isinstance(env, type):
|
|
env = env.__name__
|
|
identifier = "{}_{}".format(self.trainable_name, env)
|
|
else:
|
|
identifier = self.trainable_name
|
|
if include_trial_id:
|
|
identifier += "_" + self.trial_id
|
|
return identifier.replace("/", "_")
|
|
|
|
def _generate_dirname(self):
|
|
if self.custom_dirname:
|
|
generated_dirname = self.custom_dirname
|
|
else:
|
|
MAX_LEN_IDENTIFIER = int(os.environ.get("TUNE_MAX_LEN_IDENTIFIER", "130"))
|
|
generated_dirname = f"{str(self)}_{self.experiment_tag}"
|
|
generated_dirname = generated_dirname[:MAX_LEN_IDENTIFIER]
|
|
generated_dirname += f"_{date_str()}"
|
|
# This is the file path used by rsync. ['/', '(', ')'] are not allowed.
|
|
return re.sub("[/()]", "_", generated_dirname)
|
|
|
|
def invalidate_json_state(self):
|
|
self._state_json = None
|
|
|
|
def get_json_state(self) -> Tuple[str, str]:
|
|
if self._state_json is None:
|
|
state = self.__getstate__()
|
|
state.pop("run_metadata", None)
|
|
self._state_json = json.dumps(state, indent=2, cls=TuneFunctionEncoder)
|
|
|
|
runtime_metadata_json = self.run_metadata.get_json_state()
|
|
|
|
return self._state_json, runtime_metadata_json
|
|
|
|
@classmethod
|
|
def from_json_state(cls, json_state: str, stub: bool = False) -> "Trial":
|
|
state = _loads_with_cloudpickle(json_state)
|
|
|
|
new_trial = Trial(
|
|
state["trainable_name"],
|
|
stub=stub,
|
|
_setup_default_resource=False,
|
|
)
|
|
|
|
new_trial.__setstate__(state)
|
|
|
|
return new_trial
|
|
|
|
def restore_run_metadata(self, run_metadata: str):
|
|
self.run_metadata = _TrainingRunMetadata.from_json_state(run_metadata)
|
|
|
|
@classmethod
|
|
def from_directory(
|
|
cls, path: Union[str, os.PathLike], stub: bool = False
|
|
) -> "Trial":
|
|
metadata_path = Path(path, TRIAL_STATE_FILENAME)
|
|
if not metadata_path.exists():
|
|
raise FileNotFoundError(
|
|
f"Can't restore trial from path: File `{metadata_path}` not found."
|
|
)
|
|
|
|
json_state = metadata_path.read_text()
|
|
return cls.from_json_state(json_state, stub=stub)
|
|
|
|
def __getstate__(self):
|
|
"""Memento generator for Trial.
|
|
|
|
Sets RUNNING trials to PENDING.
|
|
Note this can only occur if the trial holds a PERSISTENT checkpoint.
|
|
"""
|
|
state = self.__dict__.copy()
|
|
|
|
for key in self._nonjson_fields:
|
|
state[key] = binary_to_hex(cloudpickle.dumps(state.get(key)))
|
|
|
|
state.pop("temporary_state", None)
|
|
|
|
state["_state_json"] = None
|
|
state["_default_result_or_future"] = None
|
|
|
|
return state
|
|
|
|
def __setstate__(self, state):
|
|
if state["status"] == Trial.RUNNING:
|
|
state["status"] = Trial.PENDING
|
|
for key in self._nonjson_fields:
|
|
if key in state:
|
|
state[key] = cloudpickle.loads(hex_to_binary(state[key]))
|
|
|
|
# Ensure that stub doesn't get overridden
|
|
stub = state.pop("stub", True)
|
|
self.__dict__.update(state)
|
|
self.stub = stub or getattr(self, "stub", False)
|
|
|
|
if not self.stub:
|
|
validate_trainable(self.trainable_name)
|
|
|
|
self.temporary_state = _TemporaryTrialState()
|
|
|
|
assert self.placement_group_factory
|