from __future__ import annotations
__copyright__ = "Copyright 2022, automl.org"
__license__ = "3-clause BSD"
from abc import ABC, abstractmethod
from typing import Any, Iterator
import time
import traceback
import numpy as np
from ConfigSpace import Configuration
from smac.runhistory import StatusType, TrialInfo, TrialValue
from smac.scenario import Scenario
from smac.utils.logging import get_logger
logger = get_logger(__name__)
[docs]class AbstractRunner(ABC):
"""Interface class to handle the execution of SMAC configurations.
This interface defines how to interact with the SMBO loop.
The complexity of running a configuration as well as handling the results is abstracted to the
SMBO via an AbstractRunner.
From SMBO perspective, launching a configuration follows a
submit/collect scheme as follows:
1. A run is launched via ``submit_run()``
- ``submit_run`` internally calls ``run_wrapper()``, a method that contains common processing functions among
different runners.
- A class that implements AbstractRunner defines ``run()`` which is really the algorithm to
translate a ``TrialInfo`` to a ``TrialValue``, i.e. a configuration to an actual result.
2. A completed run is collected via ``iter_results()``, which iterates and consumes any finished runs, if any.
3. This interface also offers the method ``wait()`` as a mechanism to make sure we have enough
data in the next iteration to make a decision. For example, the intensifier might not be
able to select the next challenger until more results are available.
Parameters
----------
scenario : Scenario
required_arguments : list[str]
A list of required arguments, which are passed to the target function.
"""
def __init__(
self,
scenario: Scenario,
required_arguments: list[str] = [],
):
self._scenario = scenario
self._required_arguments = required_arguments
# The results are a FIFO structure, implemented via a list
# (because the Queue lock is not pickable). Finished runs are
# put in this list and collected via _process_pending_runs
self._results_queue: list[tuple[TrialInfo, TrialValue]] = []
self._crash_cost = scenario.crash_cost
self._supports_memory_limit = False
if isinstance(scenario.objectives, str):
objectives = [scenario.objectives]
else:
objectives = scenario.objectives
self._objectives = objectives
self._n_objectives = scenario.count_objectives()
# We need to exapdn crash cost if the user did not do it
if self._n_objectives > 1:
if not isinstance(scenario.crash_cost, list):
assert isinstance(scenario.crash_cost, float)
self._crash_cost = [scenario.crash_cost for _ in range(self._n_objectives)]
[docs] def run_wrapper(self, trial_info: TrialInfo) -> tuple[TrialInfo, TrialValue]:
"""Wrapper around run() to execute and check the execution of a given config.
This function encapsulates common
handling/processing, so that run() implementation is simplified.
Parameters
----------
trial_info : RunInfo
Object that contains enough information to execute a configuration run in isolation.
Returns
-------
info : TrialInfo
An object containing the configuration launched.
value : TrialValue
Contains information about the status/performance of config.
"""
start = time.time()
try:
status, cost, runtime, additional_info = self.run(
config=trial_info.config,
instance=trial_info.instance,
budget=trial_info.budget,
seed=trial_info.seed,
)
except Exception as e:
status = StatusType.CRASHED
cost = self._crash_cost
runtime = time.time() - start
# Add context information to the error message
exception_traceback = traceback.format_exc()
error_message = repr(e)
additional_info = {
"traceback": exception_traceback,
"error": error_message,
}
end = time.time()
# Catch NaN or inf
if not np.all(np.isfinite(cost)):
logger.warning(
"Target function returned infinity or nothing at all. Result is treated as CRASHED"
f" and cost is set to {self._crash_cost}."
)
if "traceback" in additional_info:
logger.warning(f"Traceback: {additional_info['traceback']}\n")
status = StatusType.CRASHED
if status == StatusType.CRASHED:
cost = self._crash_cost
trial_value = TrialValue(
status=status,
cost=cost,
time=runtime,
additional_info=additional_info,
starttime=start,
endtime=end,
)
return trial_info, trial_value
@property
def meta(self) -> dict[str, Any]:
"""Returns the meta-data of the created object."""
return {"name": self.__class__.__name__}
[docs] @abstractmethod
def submit_trial(self, trial_info: TrialInfo) -> None:
"""This function submits a configuration embedded in a TrialInfo object, and uses one of the workers to produce
a result (such result will eventually be available on the ``self._results_queue`` FIFO).
This interface method will be called by SMBO, with the expectation that a function will be executed by a worker.
What will be executed is dictated by ``trial_info``, and `how` it will be executed is decided via the child
class that implements a ``run`` method.
Because config submission can be a serial/parallel endeavor, it is expected to be implemented by a child class.
Parameters
----------
trial_info : TrialInfo
An object containing the configuration launched.
"""
raise NotImplementedError
[docs] @abstractmethod
def run(
self,
config: Configuration,
instance: str | None = None,
budget: float | None = None,
seed: int | None = None,
) -> tuple[StatusType, float | list[float], float, dict]:
"""Runs the target function with a configuration on a single instance-budget-seed
combination (aka trial).
Parameters
----------
config : Configuration
Configuration to be passed to the target function.
instance : str | None, defaults to None
The Problem instance.
budget : float | None, defaults to None
A positive, real-valued number representing an arbitrary limit to the target function
handled by the target function internally.
seed : int, defaults to None
Returns
-------
status : StatusType
Status of the trial.
cost : float | list[float]
Resulting cost(s) of the trial.
runtime : float
The time the target function took to run.
additional_info : dict
All further additional trial information.
"""
raise NotImplementedError
[docs] @abstractmethod
def iter_results(self) -> Iterator[tuple[TrialInfo, TrialValue]]:
"""This method returns any finished configuration, and returns a list with the
results of executing the configurations. This class keeps populating results
to ``self._results_queue`` until a call to ``get_finished`` trials is done. In this case,
the `self._results_queue` list is emptied and all trial values produced by running
`run` are returned.
Returns
-------
Iterator[tuple[TrialInfo, TrialValue]]:
A list of TrialInfo/TrialValue tuples, all of which have been finished.
"""
raise NotImplementedError
[docs] @abstractmethod
def wait(self) -> None:
"""The SMBO/intensifier might need to wait for trials to finish before making a decision."""
raise NotImplementedError
[docs] @abstractmethod
def is_running(self) -> bool:
"""Whether there are trials still running.
Generally, if the runner is serial, launching a trial instantly returns its result. On
parallel runners, there might be pending configurations to complete.
"""
raise NotImplementedError
[docs] @abstractmethod
def count_available_workers(self) -> int:
"""Returns the number of available workers."""
raise NotImplementedError