Source code for ax.core.runner

#!/usr/bin/env python3
# Copyright (c) Meta Platforms, Inc. and affiliates.
#
# This source code is licensed under the MIT license found in the
# LICENSE file in the root directory of this source tree.

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Any, Dict, Iterable, Optional, Set, TYPE_CHECKING

from ax.utils.common.base import Base
from ax.utils.common.serialization import SerializationMixin


if TYPE_CHECKING:  # pragma: no cover
    # import as module to make sphinx-autodoc-typehints happy
    from ax import core  # noqa F401


[docs]class Runner(Base, SerializationMixin, ABC):
    """Abstract base class for custom runner classes"""

    @property
    def staging_required(self) -> bool:
        """Whether the trial goes to staged or running state once deployed."""
        return False

[docs]    @abstractmethod
    def run(self, trial: core.base_trial.BaseTrial) -> Dict[str, Any]:
        """Deploys a trial based on custom runner subclass implementation.

        Args:
            trial: The trial to deploy.

        Returns:
            Dict of run metadata from the deployment process.
        """
        pass  # pragma: no cover

[docs]    def run_multiple(
        self, trials: Iterable[core.base_trial.BaseTrial]
    ) -> Dict[int, Dict[str, Any]]:
        """Runs a single evaluation for each of the given trials. Useful when deploying
        multiple trials at once is more efficient than deploying them one-by-one.
        Used in Ax ``Scheduler``.

        NOTE: By default simply loops over `run_trial`. Should be overwritten
        if deploying multiple trials in batch is preferable.

        Args:
            trials: Iterable of trials to be deployed, each containing arms with
                parameterizations to be evaluated. Can be a `Trial`
                if contains only one arm or a `BatchTrial` if contains
                multiple arms.

        Returns:
            Dict of trial index to the run metadata of that trial from the deployment
            process.
        """
        return {trial.index: self.run(trial=trial) for trial in trials}

[docs]    def poll_available_capacity(self) -> int:
        """Checks how much available capacity there is to schedule trial evaluations.
        Required for runners used with Ax ``Scheduler``.

        NOTE: This method might be difficult to implement in some systems. Returns -1
        if capacity of the system is "unlimited" or "unknown"
        (meaning that the ``Scheduler`` should be trying to schedule as many trials
        as is possible without violating scheduler settings). There is no need to
        artificially force this method to limit capacity; ``Scheduler`` has other
        limitations in place to limit number of trials running at once,
        like the ``SchedulerOptions.max_pending_trials`` setting, or
        more granular control in the form of the `max_parallelism`
        setting in each of the `GenerationStep`s of a `GenerationStrategy`).

        Returns:
            An integer, representing how many trials there is available capacity for;
            -1 if capacity is "unlimited" or not possible to know in advance.
        """
        return -1

[docs]    def poll_trial_status(
        self, trials: Iterable[core.base_trial.BaseTrial]
    ) -> Dict[core.base_trial.TrialStatus, Set[int]]:
        """Checks the status of any non-terminal trials and returns their
        indices as a mapping from TrialStatus to a list of indices. Required
        for runners used with Ax ``Scheduler``.

        NOTE: Does not need to handle waiting between polling calls while trials
        are running; this function should just perform a single poll.

        Args:
            trials: Trials to poll.

        Returns:
            A dictionary mapping TrialStatus to a list of trial indices that have
            the respective status at the time of the polling. This does not need to
            include trials that at the time of polling already have a terminal
            (ABANDONED, FAILED, COMPLETED) status (but it may).
        """
        raise NotImplementedError(
            f"{self.__class__.__name__} does not implement a `poll_trial_status` "
            "method."
        )

[docs]    def stop(
        self, trial: core.base_trial.BaseTrial, reason: Optional[str] = None
    ) -> Dict[str, Any]:
        """Stop a trial based on custom runner subclass implementation.

        Optional method.

        Args:
            trial: The trial to stop.
            reason: A message containing information why the trial is to be stopped.

        Returns:
            A dictionary of run metadata from the stopping process.
        """
        raise NotImplementedError(
            f"{self.__class__.__name__} does not implement a `stop` method."
        )

[docs]    def clone(self) -> Runner:
        """Create a copy of this Runner."""
        cls = type(self)
        # pyre-ignore[45]: Cannot instantiate abstract class `Runner`.
        return cls(
            **cls.deserialize_init_args(args=cls.serialize_init_args(obj=self)),
        )

    def __eq__(self, other: Runner) -> bool:
        same_class = self.__class__ == other.__class__
        same_init_args = self.serialize_init_args(
            obj=self
        ) == other.serialize_init_args(obj=other)
        return same_class and same_init_args