diff --git a/src/main/python/systemds/scuro/drsearch/hyperparameter_tuner.py b/src/main/python/systemds/scuro/drsearch/hyperparameter_tuner.py index 14658714d4b..20cdb0bd40a 100644 --- a/src/main/python/systemds/scuro/drsearch/hyperparameter_tuner.py +++ b/src/main/python/systemds/scuro/drsearch/hyperparameter_tuner.py @@ -24,7 +24,8 @@ import os import numpy as np import logging -from dataclasses import dataclass +from dataclasses import dataclass, field +from types import SimpleNamespace import time import copy from joblib import Parallel, delayed @@ -42,7 +43,12 @@ SplitAwareDAGEvaluator, dag_requires_fitting, ) -from systemds.scuro.drsearch.task import PerformanceMeasure +from systemds.scuro.drsearch.task import INFERENCE_TIME, PerformanceMeasure +from systemds.scuro.drsearch.objectives import ( + pareto_front, + rank_by_pareto, + validate_objectives, +) import pickle import torch from systemds.scuro.utils.checkpointing import CheckpointManager @@ -53,7 +59,6 @@ trial_node_signatures, ) from systemds.scuro.drsearch.process_cache import HPOExecutionCache -from systemds.scuro.drsearch.ranking import tie_broken from systemds.scuro.drsearch.operator_registry import is_expensive_representation _CUDA_OOM_ERRORS = tuple( @@ -373,12 +378,25 @@ class HyperparamResult: task_name: str dag: Any mm_opt: bool = False + # Non-dominated (params, scores) in multi-objective mode, best first. + pareto_results: List[Tuple[Dict[str, Any], Any]] = field(default_factory=list) + + +def _val_scores(scores) -> Dict[str, float]: + """Mean validation scores of one evaluated config, in any of the forms + the tuner stores them.""" + if isinstance(scores, HyperparamEvaluationScores): + return scores.all_val_scores + if isinstance(scores, (list, tuple)) and isinstance(scores[1], PerformanceMeasure): + return scores[1].average_scores + return {} class HyperparamResults: - def __init__(self, tasks, modalities): + def __init__(self, tasks, modalities, k=None): self.tasks = tasks self.modalities = modalities + self.k = k self.results = {} for task in tasks: self.results[task.model.name] = { @@ -402,17 +420,20 @@ def setup_mm(self, optimize_unimodal): self.results[task.model.name] = {"mm_results": []} def get_k_best_dags(self, modality, task): - results = self.results[task.model.name][modality.modality_id] + """One DAG per tuned representation with its best params, or in + multi-objective mode one per Pareto config (at most k per representation).""" + configs = [] + for result in self.results[task.model.name][modality.modality_id]: + front = [params for params, _ in result.pareto_results][: self.k] + configs.extend((result, params) for params in front or [result.best_params]) dags = [] - for result in results: - dag_with_best_params = copy.deepcopy(result.dag) - for node in dag_with_best_params.nodes: + for result, params in configs: + dag_with_params = copy.deepcopy(result.dag) + for node in dag_with_params.nodes: if node.operation is not None: - node.parameters = _apply_trial_params_to_node( - node, result.best_params - ) - dags.append(dag_with_best_params) - return results, dags + node.parameters = _apply_trial_params_to_node(node, params) + dags.append(dag_with_params) + return [result for result, _ in configs], dags def get_k_best_results(self, modality, task, performance_metric_name): results, dags = self.get_k_best_dags(modality, task) @@ -452,10 +473,23 @@ def __init__( wandb_tags: Optional[List[str]] = None, enable_checkpointing: bool = False, representation_cache_size: int = 128, + objectives=None, ): + """ + :param objectives: [(name, "max" | "min"), ...] read from the validation + scores, e.g. [("accuracy", "max"), (INFERENCE_TIME, "min")]. The first + one replaces scoring_metric/maximize_metric. Defaults to + [(scoring_metric, "max" if maximize_metric else "min")]. + """ + if objectives is None: + objectives = [(scoring_metric, "max" if maximize_metric else "min")] + self.objectives = validate_objectives(objectives) + self.is_multi_objective = len(self.objectives) > 1 + scoring_metric, direction = self.objectives[0] + maximize_metric = direction == "max" self.tasks = tasks self.unimodal_optimization_results = optimization_results - self.optimization_results = HyperparamResults(tasks, modalities) + self.optimization_results = HyperparamResults(tasks, modalities, k) self.n_jobs = n_jobs self.scoring_metric = scoring_metric self.maximize_metric = maximize_metric @@ -490,9 +524,13 @@ def __init__( self._wandb_run = None self.enable_checkpointing = enable_checkpointing + # Shared intermediates (e.g. mel spectrograms) would make the reusing + # representation look faster than it is, so they are off when time is optimized. + timed = any(name == INFERENCE_TIME for name, _ in self.objectives) self.execution_cache = HPOExecutionCache( representation_entries=representation_cache_size, should_cache_representation=self._should_cache_representation, + intermediate_bytes=0 if timed else None, ) def _should_cache_representation(self, operation: Any) -> bool: @@ -725,13 +763,17 @@ def get_score(result): score = np.mean(score) return score - best_params, best_score = all_results[0][0], get_score(all_results[0]) - for params, score in all_results[1:]: + pareto_results = ( + self._pareto_results(all_results) if self.is_multi_objective else [] + ) + candidates = pareto_results or all_results + best_params, best_score = candidates[0][0], get_score(candidates[0]) + for params, score in candidates[1:]: candidate_score = get_score((params, score)) if self._is_better(candidate_score, best_score): best_params, best_score = params, candidate_score - if hyperparams and best_params != baseline_params: + if hyperparams and best_params != baseline_params and not pareto_results: baseline_folds = self._score_value_list(baseline_raw_scores[1]) candidate_folds = next(s for p, s in all_results if p == best_params)[1] @@ -762,10 +804,18 @@ def get_score(result): task_name=task.model.name, dag=dag, mm_opt=mm_opt, + pareto_results=pareto_results, ) return best_result + def _pareto_results(self, all_results): + candidates = [ + SimpleNamespace(params=params, scores=scores, val_score=_val_scores(scores)) + for params, scores in all_results + ] + return [(c.params, c.scores) for c in pareto_front(candidates, self.objectives)] + def _leaf_input_stats(self, node: RepresentationNode, dag) -> Any: if dag is None or len(node.inputs) != 1: return None @@ -1043,7 +1093,6 @@ def _search_best_configs( tied_params: Optional[Dict[str, str]] = None, ) -> List[Tuple[Dict[str, Any], Any]]: import optuna - from optuna.trial import TrialState optuna.logging.set_verbosity( optuna.logging.INFO if self.debug else optuna.logging.WARNING @@ -1069,7 +1118,10 @@ def _search_best_configs( if budget <= 0: return all_results - direction = "maximize" if self.maximize_metric else "minimize" + directions = [ + "maximize" if direction == "max" else "minimize" + for _, direction in self.objectives + ] sampler = ( optuna.samplers.TPESampler(seed=self.random_state) if self.optuna_sampler == "tpe" @@ -1077,7 +1129,7 @@ def _search_best_configs( ) study = optuna.create_study( - direction=direction, + directions=directions, sampler=sampler, study_name=f"{task.model.name}-{rep_name}"[:64], ) @@ -1105,7 +1157,7 @@ def _search_best_configs( }, } wandb_cb = WeightsAndBiasesCallback( - metric_name=self.scoring_metric, + metric_name=[name for name, _ in self.objectives], wandb_kwargs=wandb_kwargs["wandb_kwargs"], ) except ImportError: @@ -1158,10 +1210,21 @@ def objective(trial: optuna.Trial) -> float: val_folds = evaluation_scores.val_fold_scores.get(self.scoring_metric, []) if val_folds is not None and len(val_folds) > 1: lam = 0.5 - robust_val = np.mean(val_folds) - lam * np.std(val_folds, ddof=1) - return robust_val - - return val_score + val_score = np.mean(val_folds) - lam * np.std(val_folds, ddof=1) + if not self.is_multi_objective: + return val_score + + values = tuple( + ( + val_score + if name == self.scoring_metric + else float(evaluation_scores.all_val_scores.get(name, np.nan)) + ) + for name, _ in self.objectives + ) + if np.isnan(values).any(): + raise optuna.TrialPruned() + return values callbacks = [c for c in [wandb_cb] if c is not None] n_jobs = 1 if self.n_jobs == 0 else max(1, abs(self.n_jobs)) @@ -1179,17 +1242,6 @@ def objective(trial: optuna.Trial) -> float: wandb.run.finish() all_results.extend(trial_results) - - for trial in study.trials: - if trial.state != TrialState.COMPLETE: - continue - config = trial.params - key = self._config_key(config) - if key in seen: - all_results.append(seen[key]) - else: - pass - return all_results def _get_cached_modalities_for_task(self, task, modality_ids): @@ -1261,6 +1313,8 @@ def _evaluate_dag_once( on_cpu=False, node_cache=None, ): + # Task.run keeps per-run state on the task; trials may run in threads. + task = copy.copy(task) dag_copy = copy.deepcopy(dag) for node_id in node_order: node = dag_copy.get_node_by_id(node_id) @@ -1288,7 +1342,9 @@ def _evaluate_dag_once( node_cache=node_cache, execution_cache=self.execution_cache, ) - return params, task.run(modified_modality.data) + return params, task.run( + modified_modality.data, modified_modality.transform_time + ) def _evaluate_dag_trial( self, @@ -1381,24 +1437,10 @@ def tune_multimodal_representations( self.resume_from_checkpoint() self.optimization_results.setup_mm(optimize_unimodal) for task in self.tasks: - - def _get_metric_value(result): - score = result.val_score - if isinstance(score, PerformanceMeasure): - score = score.average_scores - if isinstance(score, dict): - return score.get( - self.scoring_metric, - float("-inf") if self.maximize_metric else float("inf"), - ) - return score - - candidates = optimization_results[task.model.name] - best_results = sorted( - [candidates[i] for i in tie_broken(candidates)], - key=_get_metric_value, - reverse=self.maximize_metric, - )[:k] + best_results, _ = rank_by_pareto( + optimization_results[task.model.name], self.objectives + ) + best_results = best_results[:k] best_optimization_results = best_results skip_remaining = self._checkpoint_manager.skip_remaining_by_key.get( task.model.name, 0 diff --git a/src/main/python/systemds/scuro/drsearch/multimodal_ga_optimizer.py b/src/main/python/systemds/scuro/drsearch/multimodal_ga_optimizer.py index f754116de93..0743307d5aa 100644 --- a/src/main/python/systemds/scuro/drsearch/multimodal_ga_optimizer.py +++ b/src/main/python/systemds/scuro/drsearch/multimodal_ga_optimizer.py @@ -60,6 +60,12 @@ sample_stats, typecheck, ) +from systemds.scuro.drsearch.objectives import ( + dominates, + objective_vector, + retime_front, + validate_objectives, +) from systemds.scuro.drsearch.operator_registry import Registry from systemds.scuro.drsearch.representation_dag import ( RepresentationDAGBuilder, @@ -67,6 +73,7 @@ compose_dag_leaves, ) from systemds.scuro.drsearch.task import Task +from systemds.scuro.drsearch.test_set_evaluation import measure_inference_time from systemds.scuro.drsearch.split_aware_evaluator import ( SplitAwareDAGEvaluator, dag_requires_fitting, @@ -317,7 +324,7 @@ def _evaluate_genome_body( fused = AggregatedRepresentation().transform(fused) t0 = time.time() - scores = task.run(fused.data) + scores = task.run(fused.data, fused.transform_time) task_time = time.time() - t0 total = time.time() - start @@ -441,17 +448,7 @@ def __init__( self.maximize_metric = maximize_metric if objectives is not None: - if len(objectives) < 1: - raise ValueError( - "objectives must contain at least one (name, direction) pair" - ) - for name, direction in objectives: - if direction not in ("max", "min"): - raise ValueError( - f"objective direction must be 'max' or 'min', got " - f"{direction!r} for objective {name!r}" - ) - self.objective_specs: List[ObjectiveSpec] = list(objectives) + self.objective_specs: List[ObjectiveSpec] = validate_objectives(objectives) self.metric_name = self.objective_specs[0][0] self.maximize_metric = self.objective_specs[0][1] == "max" else: @@ -942,11 +939,7 @@ def _record_evaluation( def _dominates(self, a: Tuple[float, ...], b: Tuple[float, ...]) -> bool: """True if objective tuple `a` Pareto-dominates `b`, direction-aware.""" - wa = [w * v for w, v in zip(self._objective_weights, a)] - wb = [w * v for w, v in zip(self._objective_weights, b)] - return all(x >= y for x, y in zip(wa, wb)) and any( - x > y for x, y in zip(wa, wb) - ) + return dominates(a, b, self.objective_specs) def _update_hall_of_fame(self, task_name: str, result: FusionSearchResult) -> None: timing = { @@ -986,6 +979,27 @@ def _update_hall_of_fame(self, task_name: str, result: FusionSearchResult) -> No def get_hall_of_fame(self, task_name: str) -> List[FusionSearchResult]: return list(self.hall_of_fame.get(task_name, [])) + def retime_hall_of_fame( + self, task: Task, repeats: int = 3 + ) -> List[FusionSearchResult]: + """Re-time the Pareto front serially (see ``retime_front``) and rebuild + the multi-objective hall of fame from the re-timed front.""" + task_name = task.model.name + front = retime_front( + self.optimization_results.get(task_name, []), + self.objective_specs, + lambda result: measure_inference_time( + self.compose_end_to_end_dag(result.dag, task), self.modalities, task + ), + repeats, + ) + if self.is_multi_objective: + self.hall_of_fame[task_name] = list(front) + self._hof_fitness[task_name] = [ + objective_vector(result, self.objective_specs) for result in front + ] + return front + def _extract_k_best(self, unimodal_results) -> Dict[str, Dict[str, List[Any]]]: k_best = {} for task in self.tasks: diff --git a/src/main/python/systemds/scuro/drsearch/multimodal_optimizer.py b/src/main/python/systemds/scuro/drsearch/multimodal_optimizer.py index a4f6a701efe..3bd201f0c7b 100644 --- a/src/main/python/systemds/scuro/drsearch/multimodal_optimizer.py +++ b/src/main/python/systemds/scuro/drsearch/multimodal_optimizer.py @@ -46,6 +46,7 @@ order_invariant, sample_stats, ) +from systemds.scuro.drsearch.objectives import pareto_front, validate_objectives from systemds.scuro.drsearch.operator_registry import Registry from systemds.scuro.utils.schema_helpers import get_shape @@ -105,7 +106,9 @@ def _evaluate_dag_worker(dag_pickle, task_pickle, modalities_pickle, debug=False final_representation = agg_operator.transform(final_representation) eval_start = time.time() - scores = task.run(final_representation.data) + scores = task.run( + final_representation.data, final_representation.transform_time + ) eval_time = time.time() - eval_start total_time = time.time() - start_time @@ -140,8 +143,12 @@ def __init__( checkpoint_dir: Optional[str] = None, resume: bool = False, algebra: Optional[AlgebraConfig] = None, + objectives=None, ): self.modalities = modalities + self.objectives = ( + [(metric, "max")] if objectives is None else validate_objectives(objectives) + ) self.tasks = tasks self.k = k self.debug = debug @@ -181,6 +188,10 @@ def resume_from_checkpoint(self): results, _, _ = loaded self.optimization_results = results + def get_pareto_front(self, task_name: str) -> List["OptimizationResult"]: + """Non-dominated results for ``task_name`` under ``self.objectives``.""" + return pareto_front(self.optimization_results[task_name], self.objectives) + def _ensure_results_initialized(self) -> Dict[str, List["OptimizationResult"]]: if not isinstance(self.optimization_results, dict): self.optimization_results = {} @@ -527,7 +538,9 @@ def _evaluate_dag(self, dag: RepresentationDag, task: Task) -> "OptimizationResu fused_representation = agg_operator.transform(fused_representation) eval_start = time.time() - scores = task.run(fused_representation.data) + scores = task.run( + fused_representation.data, fused_representation.transform_time + ) eval_time = time.time() - eval_start total_time = time.time() - start_time diff --git a/src/main/python/systemds/scuro/drsearch/node_executor.py b/src/main/python/systemds/scuro/drsearch/node_executor.py index 7c4f051b274..7a35135b4f8 100644 --- a/src/main/python/systemds/scuro/drsearch/node_executor.py +++ b/src/main/python/systemds/scuro/drsearch/node_executor.py @@ -22,6 +22,7 @@ import multiprocessing as mp import os import time +from contextlib import contextmanager from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Union @@ -93,6 +94,58 @@ def _lazy_leaf_budget_fraction() -> float: return float(os.environ.get("SCURO_LAZY_LEAF_BUDGET_FRACTION", "0.5")) +def _decoded_leaf_fits_cpu_budget(modality: Any, data: LazyFileSequence) -> bool: + estimate = getattr(modality.data_loader, "estimate_decoded_bytes", None) + if not callable(estimate): + return False + metadata = modality.metadata if len(modality.metadata or []) == len(data) else None + decoded_bytes = estimate(metadata) + budget = cpu_memory_budget_bytes(modality.modality_type.name) + return ( + bool(decoded_bytes) and decoded_bytes <= budget * _lazy_leaf_budget_fraction() + ) + + +@contextmanager +def decoded_lazy_leaves(modalities: List[Any]): + """Load leaves and decode lazy ones in threads, like the executor's leaf load. + + The load and decode time is added to each leaf's transform_time. Leaves that + do not fit the CPU budget stay lazy and decode inside the transform. Data, + metadata and transform_time are restored on exit. + """ + restore = [] + try: + for modality in modalities: + loader = getattr(modality, "data_loader", None) + if loader is None or loader.chunk_size is not None: + continue + restore.append( + (modality, modality._data, modality.metadata, modality.transform_time) + ) + start = time.perf_counter() + if not modality.has_data(): + modality.extract_raw_data() + data = modality.data + if ( + isinstance(data, LazyFileSequence) + and _materialize_lazy_leaves() + and _decoded_leaf_fits_cpu_budget(modality, data) + ): + arrays = data.materialize() + for array in arrays: + if hasattr(array, "flags"): + array.flags.writeable = False + modality._data = arrays + modality.transform_time += time.perf_counter() - start + yield + finally: + for modality, data, metadata, transform_time in reversed(restore): + modality._data = data + modality.metadata = metadata + modality.transform_time = transform_time + + def _keep_worker_models() -> bool: return os.environ.get("SCURO_KEEP_WORKER_MODELS", "1") == "1" @@ -273,10 +326,10 @@ def _run_node_op(): } -def _task_data(modality, aggregation): +def _task_input(modality, aggregation): if aggregation is None: - return modality.data - return aggregation.operation(params=aggregation.parameters).transform(modality).data + return modality + return aggregation.operation(params=aggregation.parameters).transform(modality) def _execute_task_batch_worker(task: Any, items: List[tuple], gpu_id: int): @@ -286,7 +339,8 @@ def _execute_task_batch_worker(task: Any, items: List[tuple], gpu_id: int): datas, failed = [], {} for node_id, modality, aggregation in items: try: - datas.append((node_id, _task_data(modality, aggregation))) + task_input = _task_input(modality, aggregation) + datas.append((node_id, task_input.data, task_input.transform_time)) except Exception as e: failed[node_id] = f"{type(e).__name__}: {e}" @@ -294,13 +348,17 @@ def _execute_task_batch_worker(task: Any, items: List[tuple], gpu_id: int): cores = sorted(allowed) os.sched_setaffinity(0, {cores[os.getpid() % len(cores)]}) try: - scores = task.run_many([data for _, data in datas], device) + scores = task.run_many( + [data for _, data, _ in datas], + device, + [seconds for _, _, seconds in datas], + ) finally: os.sched_setaffinity(0, allowed) cleanup_gpu(gpu_id) share = (time.perf_counter() - start_time) / max(1, len(datas)) results = {} - for i, ((node_id, _), result) in enumerate(zip(datas, scores)): + for i, ((node_id, _, _), result) in enumerate(zip(datas, scores)): if isinstance(result, Exception): failed[node_id] = f"{type(result).__name__}: {result}" continue @@ -337,7 +395,8 @@ def _execute_task_worker( def _run_task(): start = time.perf_counter() - scores = task.run(_task_data(modality, aggregation)) + task_input = _task_input(modality, aggregation) + scores = task.run(task_input.data, task_input.transform_time) end = time.perf_counter() return scores, end - start @@ -416,12 +475,17 @@ def _run(): def _load_leaf_worker(modality: Any, materialize: bool = False) -> Dict[str, Any]: + # load_seconds covers only reading/decoding, so it can be charged to the + # leaf like the timed load in apply_representations. + start = time.perf_counter() if hasattr(modality, "extract_raw_data") and not modality.has_data(): modality.extract_raw_data() + load_seconds = time.perf_counter() - start data = modality.data if materialize and isinstance(data, LazyFileSequence): arrays = data.materialize() + decode_seconds = time.perf_counter() - start shared, shm_name, shm_bytes = move_arrays_to_shared_memory(arrays) if shared is not None: return { @@ -429,6 +493,7 @@ def _load_leaf_worker(modality: Any, materialize: bool = False) -> Dict[str, Any "metadata": modality.metadata, "shm_name": shm_name, "shm_bytes": shm_bytes, + "load_seconds": decode_seconds, } if sum(getattr(a, "nbytes", 0) for a in arrays) < SHARED_MEMORY_MIN_BYTES: return { @@ -436,6 +501,7 @@ def _load_leaf_worker(modality: Any, materialize: bool = False) -> Dict[str, Any "metadata": modality.metadata, "shm_name": None, "materialized": True, + "load_seconds": decode_seconds, } resident_bytes = 0 @@ -450,7 +516,12 @@ def _load_leaf_worker(modality: Any, materialize: bool = False) -> Dict[str, Any if wrapped is not None: data = wrapped - return {"data": data, "metadata": modality.metadata, "shm_name": shm_name} + return { + "data": data, + "metadata": modality.metadata, + "shm_name": shm_name, + "load_seconds": load_seconds, + } def _retained_gpu_bytes() -> int: @@ -660,6 +731,7 @@ def __init__( self._leaf_shm_owners: Dict[str, Any] = {} self._leaf_original_data: Dict[str, Any] = {} self._leaf_unshared_originals: List[tuple] = [] + self._leaf_load_seconds: Dict[int, float] = {} self._task_results: Dict[str, ResultEntry] = {} self._fused_shapes: Dict[tuple, tuple] = {} self._fused_ready_since: Dict[str, float] = {} @@ -776,6 +848,8 @@ def _load_leaf_modalities(self) -> None: ) modality._data = jr.value["data"] modality.metadata = jr.value["metadata"] + modality.transform_time += jr.value["load_seconds"] + self._leaf_load_seconds[id(modality)] = jr.value["load_seconds"] if jr.value.get("materialized"): self._leaf_unshared_originals.append((modality, original)) shm_name = jr.value.get("shm_name") @@ -806,12 +880,16 @@ def _cleanup_leaf_shared_memory(self) -> None: if callable(close): close() modality._data = self._leaf_original_data.pop(shm_name, None) + modality.transform_time -= self._leaf_load_seconds.pop(id(modality), 0) unlink_shm(shm_name) self._leaf_shm_names = [] self._leaf_shm_owners = {} for modality, original in self._leaf_unshared_originals: modality._data = original + modality.transform_time -= self._leaf_load_seconds.pop(id(modality), 0) self._leaf_unshared_originals = [] + # Leaves whose loaded data stays attached keep their load time. + self._leaf_load_seconds = {} def _submit_node( self, node_id: str, allow_gpu_worker_for_cpu: bool = False diff --git a/src/main/python/systemds/scuro/drsearch/objectives.py b/src/main/python/systemds/scuro/drsearch/objectives.py new file mode 100644 index 00000000000..e47ff927f43 --- /dev/null +++ b/src/main/python/systemds/scuro/drsearch/objectives.py @@ -0,0 +1,237 @@ +# ------------------------------------------------------------- +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +# ------------------------------------------------------------- +"""Objectives and Pareto ranking shared by all optimization stages. + +An objective is a ``(name, direction)`` pair with direction ``"max"`` or +``"min"``. Values are read from ``entry.val_score[name]``, so any metric of +the task and ``INFERENCE_TIME`` can be used, e.g. +``[("accuracy", "max"), (INFERENCE_TIME, "min")]``. +""" + +import math +from typing import Callable, Dict, List, Sequence, Tuple + +import numpy as np + +from systemds.scuro.drsearch.ranking import tie_broken +from systemds.scuro.drsearch.task import INFERENCE_TIME, PerformanceMeasure + +Objective = Tuple[str, str] + +# The value measured during the search, kept after a serial re-timing pass. +SEARCH_INFERENCE_TIME = "search_" + INFERENCE_TIME + +__all__ = [ + "INFERENCE_TIME", + "Objective", + "crowding_distance", + "dominates", + "non_dominated_sort", + "objective_vector", + "pareto_front", + "rank_by_pareto", + "retime_front", + "SEARCH_INFERENCE_TIME", + "validate_objectives", +] + +# Search-cost attributes some result types carry next to val_score. +_ATTRIBUTE_OBJECTIVES = ("runtime", "task_time", "representation_time") + + +def validate_objectives(objectives) -> List[Objective]: + objectives = [tuple(objective) for objective in objectives] + if not objectives: + raise ValueError("objectives must contain at least one (name, direction) pair") + for name, direction in objectives: + if direction not in ("max", "min"): + raise ValueError( + f"objective direction must be 'max' or 'min', got " + f"{direction!r} for objective {name!r}" + ) + return objectives + + +def objective_vector(entry, objectives: Sequence[Objective]) -> Tuple[float, ...]: + """Objective values of ``entry``; NaN where a value is missing.""" + scores = getattr(entry, "val_score", None) + if isinstance(scores, PerformanceMeasure): + scores = scores.average_scores + values = [] + for name, _ in objectives: + if isinstance(scores, dict) and name in scores: + value = scores[name] + elif name in _ATTRIBUTE_OBJECTIVES: + value = getattr(entry, name, math.nan) + else: + value = math.nan + try: + values.append(float(value)) + except (TypeError, ValueError): + values.append(math.nan) + return tuple(values) + + +def _oriented(vector, objectives): + """Flip minimized objectives so that larger is better everywhere.""" + return [ + v if direction == "max" else -v for v, (_, direction) in zip(vector, objectives) + ] + + +def dominates(a, b, objectives: Sequence[Objective]) -> bool: + """True if objective vector ``a`` Pareto-dominates ``b``.""" + a, b = _oriented(a, objectives), _oriented(b, objectives) + return all(x >= y for x, y in zip(a, b)) and any(x > y for x, y in zip(a, b)) + + +def non_dominated_sort(vectors, objectives: Sequence[Objective]) -> List[List[int]]: + """Indices of ``vectors`` grouped into fronts, the Pareto front first.""" + n = len(vectors) + dominated = [[] for _ in range(n)] + dominator_count = [0] * n + for i in range(n): + for j in range(i + 1, n): + if dominates(vectors[i], vectors[j], objectives): + dominated[i].append(j) + dominator_count[j] += 1 + elif dominates(vectors[j], vectors[i], objectives): + dominated[j].append(i) + dominator_count[i] += 1 + + fronts = [] + current = [i for i in range(n) if dominator_count[i] == 0] + while current: + fronts.append(current) + following = [] + for i in current: + for j in dominated[i]: + dominator_count[j] -= 1 + if dominator_count[j] == 0: + following.append(j) + current = sorted(following) + return fronts + + +def crowding_distance(vectors, front: Sequence[int]) -> Dict[int, float]: + """NSGA-II crowding distance of each index in ``front``; boundary points get inf.""" + distance = {i: 0.0 for i in front} + if len(front) <= 2: + return {i: math.inf for i in front} + for m in range(len(vectors[front[0]])): + ordered = sorted(front, key=lambda i: vectors[i][m]) + low, high = vectors[ordered[0]][m], vectors[ordered[-1]][m] + distance[ordered[0]] = distance[ordered[-1]] = math.inf + if high == low: + continue + for prev, cur, nxt in zip(ordered, ordered[1:], ordered[2:]): + distance[cur] += (vectors[nxt][m] - vectors[prev][m]) / (high - low) + return distance + + +def _valid(entries, objectives): + """(index, entry, vector) for entries with a finite value for every objective.""" + rows = [] + for index, entry in enumerate(entries): + if entry is None: + continue + vector = objective_vector(entry, objectives) + if all(math.isfinite(v) for v in vector): + rows.append((index, entry, vector)) + return rows + + +def _ranked(entries, objectives): + """(front rank, original index, entry) in rank_by_pareto order.""" + objectives = validate_objectives(objectives) + rows = _valid(list(entries), objectives) + if not rows: + return [] + indices, valid_entries, vectors = zip(*rows) + tie_position = {local: pos for pos, local in enumerate(tie_broken(valid_entries))} + + keys = {} + for front_rank, front in enumerate(non_dominated_sort(vectors, objectives)): + crowding = ( + crowding_distance(vectors, front) + if len(objectives) > 1 + else dict.fromkeys(front, 0.0) + ) + for local in front: + keys[local] = (front_rank, -crowding[local], tie_position[local]) + + return [ + (keys[local][0], indices[local], valid_entries[local]) + for local in sorted(keys, key=keys.get) + ] + + +def rank_by_pareto(entries, objectives: Sequence[Objective]): + """Order entries by Pareto front, then by crowding distance (more isolated + first), then by ``tie_broken``. Entries missing an objective are dropped. + + With a single objective this is a sort by that objective. Returns + ``(sorted_entries, sorted_indices)`` like the other ``rank_by_*`` functions. + """ + ranked = _ranked(entries, objectives) + return [entry for _, _, entry in ranked], [index for _, index, _ in ranked] + + +def pareto_front(entries, objectives: Sequence[Objective]) -> list: + """The non-dominated entries, in ``rank_by_pareto`` order.""" + return [entry for rank, _, entry in _ranked(entries, objectives) if rank == 0] + + +def _val_scores(entry) -> dict: + scores = entry.val_score + return scores.average_scores if isinstance(scores, PerformanceMeasure) else scores + + +def retime_front( + entries, + objectives: Sequence[Objective], + measure: Callable[[object], float], + repeats: int = 3, +) -> list: + """Re-time the Pareto front serially and return the front after re-timing. + + Timings taken while other candidates run in parallel are too noisy to rank + near-identical pipelines. ``measure(entry)`` re-measures the entry's + ``INFERENCE_TIME`` alone; it is called ``repeats`` times per front member and + the median replaces the search value, which is kept under + ``SEARCH_INFERENCE_TIME``. Entries that join the front after re-timing are + re-timed as well, until every front member has been measured serially. + """ + if INFERENCE_TIME not in (name for name, _ in objectives): + return pareto_front(entries, objectives) + retimed = set() + while True: + front = pareto_front(entries, objectives) + pending = [entry for entry in front if id(entry) not in retimed] + if not pending: + return front + for entry in pending: + scores = _val_scores(entry) + scores.setdefault(SEARCH_INFERENCE_TIME, scores[INFERENCE_TIME]) + scores[INFERENCE_TIME] = float( + np.median([measure(entry) for _ in range(max(1, repeats))]) + ) + retimed.add(id(entry)) diff --git a/src/main/python/systemds/scuro/drsearch/split_aware_evaluator.py b/src/main/python/systemds/scuro/drsearch/split_aware_evaluator.py index 9e47f1a960b..2ad2a9d7225 100644 --- a/src/main/python/systemds/scuro/drsearch/split_aware_evaluator.py +++ b/src/main/python/systemds/scuro/drsearch/split_aware_evaluator.py @@ -21,7 +21,7 @@ import copy import os import time -from dataclasses import dataclass +from dataclasses import dataclass, replace import numpy as np @@ -53,6 +53,7 @@ class SplitEvaluationResult: scores: list representation_time: float task_time: float + inference_cost: float _REFIT_CACHE = None @@ -111,16 +112,17 @@ def materialize(modality) -> None: if loader is None: raise ValueError("Split-aware evaluation requires materialized data") + start = time.perf_counter() if getattr(loader, "chunk_size", None) is None: modality.extract_raw_data() - return - - full_loader = copy.copy(loader) - full_loader.chunk_size = None - full_loader.reset() - data, metadata = full_loader.load() - modality._data = data - modality.metadata = metadata + else: + full_loader = copy.copy(loader) + full_loader.chunk_size = None + full_loader.reset() + data, metadata = full_loader.load() + modality._data = data + modality.metadata = metadata + modality.transform_time += time.perf_counter() - start @staticmethod def _wrap(transformed, instance_ids) -> IndexedModality: @@ -160,6 +162,21 @@ def _apply_stateless(self, operation, inputs, node): return self._wrap(transformed, inputs.instance_ids) + @staticmethod + def _timed(transform, partition_input) -> IndexedModality: + start = time.perf_counter() + result = transform(partition_input) + elapsed = time.perf_counter() - start + inputs = ( + partition_input + if isinstance(partition_input, tuple) + else (partition_input,) + ) + return replace( + result, + transform_seconds=elapsed + sum(item.transform_seconds for item in inputs), + ) + @staticmethod def _align_partition_shapes(result): if len(result) < 2 or not all(isinstance(r, IndexedModality) for r in result): @@ -292,16 +309,19 @@ def execute_node(node_id): fit_state = operation.fit(fit_input, labels=labels) aggregation = pushdown_aggregation_for_node(node.parameters) result = tuple( - operation.transform_fitted( + self._timed( + lambda p: operation.transform_fitted( + p, fit_state, aggregation=aggregation + ), partition, - fit_state, - aggregation=aggregation, ) for partition in input_partitions ) else: result = tuple( - self._apply_stateless(operation, partition, node) + self._timed( + lambda p: self._apply_stateless(operation, p, node), partition + ) for partition in input_partitions ) result = self._align_partition_shapes(result) @@ -341,7 +361,11 @@ def _indexed_modalities(self): full = [] for modality in self.modalities: self.materialize(modality) - full.append(IndexedModality(modality, range(len(modality.data)))) + full.append( + IndexedModality( + modality, range(len(modality.data)), modality.transform_time + ) + ) self._full = tuple(full) return self._full @@ -367,20 +391,27 @@ def evaluate(self) -> SplitEvaluationResult: test_scores = PerformanceMeasure("test", self.task.performance_measures) self.task.training_time = [] self.task.inference_time = [] + self.task.val_inference_time = [] training_work = TrainingWorkAccumulator() representation_time = 0.0 task_time = 0.0 + transform_seconds = 0.0 + n_transformed = 0 for split in self.task.evaluation_splits: train = tuple(item.subset(split.fit_indices) for item in full) evaluation = tuple(item.subset(split.evaluation_indices) for item in full) test = tuple(item.subset(self.task.test_indices) for item in full) start = time.perf_counter() - transformed_train, transformed_evaluation, transformed_test = ( - self._execute_partitions(train, evaluation, test) - ) + transformed = self._execute_partitions(train, evaluation, test) representation_time += time.perf_counter() - start + transformed_train, transformed_evaluation, transformed_test = transformed + fold_transform_seconds = sum(p.transform_seconds for p in transformed) + fold_instances = sum(len(p) for p in transformed) + transform_seconds += fold_transform_seconds + n_transformed += fold_instances + representation_per_instance = fold_transform_seconds / fold_instances model = self.task.create_model() train_X = self._partition_data(transformed_train) @@ -400,7 +431,9 @@ def evaluate(self) -> SplitEvaluationResult: evaluation_y, ) self.task.training_time.append(time.perf_counter() - train_start) + val_start = time.perf_counter() val_score = model.test(evaluation_X, evaluation_y) + self.task.val_inference_time.append(time.perf_counter() - val_start) test_start = time.perf_counter() test_score = model.test(np.asarray(test_X), test_y) self.task.inference_time.append(time.perf_counter() - test_start) @@ -409,6 +442,13 @@ def evaluate(self) -> SplitEvaluationResult: train_scores.add_scores(train_score[0]) val_scores.add_scores(val_score[0]) test_scores.add_scores(test_score[0]) + val_scores.add_time_per_instance( + representation_per_instance + + self.task.val_inference_time[-1] / len(evaluation_X) + ) + test_scores.add_time_per_instance( + representation_per_instance + self.task.inference_time[-1] / len(test_X) + ) training_work.add(model) @@ -416,6 +456,9 @@ def evaluate(self) -> SplitEvaluationResult: model.clean_up() self.task.training_work = training_work + inference_timing = self.task.inference_cost_timing( + transform_seconds / n_transformed + ) self.task.last_run_timing = { "train_time_per_fold_s": list(self.task.training_time), "test_inference_time_per_fold_s": list(self.task.inference_time), @@ -430,6 +473,7 @@ def evaluate(self) -> SplitEvaluationResult: else 0.0 ), "n_test_instances": len(self.task.test_indices), + **inference_timing, **training_work.as_dict(), } @@ -441,4 +485,5 @@ def evaluate(self) -> SplitEvaluationResult: ], representation_time=representation_time, task_time=task_time, + inference_cost=inference_timing["inference_cost"], ) diff --git a/src/main/python/systemds/scuro/drsearch/task.py b/src/main/python/systemds/scuro/drsearch/task.py index 6cc1ec37b5c..fbc9b9e4bba 100644 --- a/src/main/python/systemds/scuro/drsearch/task.py +++ b/src/main/python/systemds/scuro/drsearch/task.py @@ -46,6 +46,9 @@ ) +INFERENCE_TIME = "inference_time_per_instance_s" + + def _as_features(X): X = np.asarray(X, dtype=np.float32) return X.reshape(len(X), -1) @@ -58,6 +61,7 @@ def __init__(self, name, metrics, higher_is_better=True): self.metrics = metrics self.higher_is_better = higher_is_better self.scores = {} + self.time_per_instance = [] if isinstance(metrics, list): for metric in metrics: @@ -72,25 +76,33 @@ def add_scores(self, scores): else: self.scores[self.metrics].append(scores[self.metrics]) + def add_time_per_instance(self, seconds): + """Representation + model inference seconds for one instance in one fold.""" + self.time_per_instance.append(float(seconds)) + + def _values(self): + values = dict(self.scores) + if self.time_per_instance: + values[INFERENCE_TIME] = self.time_per_instance + return values + def compute_averages(self): - self.average_scores = {} - if isinstance(self.metrics, list): - for metric in self.metrics: - self.average_scores[metric] = np.mean(self.scores[metric]) - else: - self.average_scores[self.metrics] = np.mean(self.scores[self.metrics]) + self.average_scores = { + metric: np.mean(values) for metric, values in self._values().items() + } return self def fold_scores(self): return { - metric: [float(v) for v in values] for metric, values in self.scores.items() + metric: [float(v) for v in values] + for metric, values in self._values().items() } def score_stds(self): """Per-metric standard deviation across folds (0.0 for a single fold).""" return { metric: (float(np.std(values, ddof=1)) if len(values) > 1 else 0.0) - for metric, values in self.scores.items() + for metric, values in self._values().items() } @@ -130,6 +142,7 @@ def __init__( self.cv_seed = get_global_seed() if cv_seed is None else cv_seed self.measure_performance = measure_performance self.inference_time = [] + self.val_inference_time = [] self.training_time = [] self.last_run_timing = {} self.training_work = TrainingWorkAccumulator() @@ -461,16 +474,19 @@ def get_train_test_split(self, data): return X_train, y_train, X_test, y_test - def run(self, data): + def run(self, data, representation_time=0.0): """ The run method needs to be implemented by every task class It handles the training and validation procedures for the specific task :param data: The aligned data used in the prediction process + :param representation_time: time spent computing data, used for the + predicted test inference cost :return: the validation accuracy """ self.validate_input_memory(data) self._reset_params() model = self.create_model() + representation_time_per_instance = representation_time / len(data) test_X = self._gather_by_indices(data, self.test_indices) test_y = self._gather_by_indices(self.labels, self.test_indices) @@ -485,6 +501,12 @@ def run(self, data): val_y = self._gather_by_indices(self.labels, fold_val_indices) self._run_fold(model, train_X, train_y, val_X, val_y, test_X, test_y) + self.record_time_per_instance( + representation_time_per_instance, + representation_time_per_instance, + len(val_X), + len(test_X), + ) if hasattr(model, "clean_up"): model.clean_up() del model @@ -499,6 +521,7 @@ def run(self, data): float(np.mean(self.inference_time)) if self.inference_time else 0.0 ), "n_test_instances": len(self.test_indices) if self.test_indices else 0, + **self.inference_cost_timing(representation_time_per_instance), **self.training_work.as_dict(), } return [ @@ -507,7 +530,7 @@ def run(self, data): self.test_scores.compute_averages(), ] - def run_many(self, datas, device): + def run_many(self, datas, device, representation_times=None): model = self.create_model() rows = list( dict.fromkeys(list(self.train_indices) + list(self.test_indices or [])) @@ -533,23 +556,28 @@ def run_many(self, datas, device): seed = getattr(model, "seed", None) seed = get_train_seed(get_global_seed() if seed is None else seed) - outputs = {} + outputs, inference_per_instance = {}, {} start = time.perf_counter() for width, group in by_width.items(): slots = candidate_slots(model, width, len(rows), self.kfold) for i in range(0, len(group), slots): chunk = group[i : i + slots] - fitted = fit_folds( + fitted, per_row_s = fit_folds( model, [X for _, X in chunk], labels, splits, seed, device ) outputs.update(zip([c for c, _ in chunk], fitted)) + inference_per_instance.update((c, per_row_s) for c, _ in chunk) per_fit_s = (time.perf_counter() - start) / max(1, len(outputs) * self.kfold) + if representation_times is None: + representation_times = [0.0] * len(datas) for c, out in outputs.items(): if not np.isfinite(out).all(): results[c] = ValueError("training diverged: non-finite model output") continue self._reset_params() + representation_per_instance = representation_times[c] / len(datas[c]) + time_per_instance = representation_per_instance + inference_per_instance[c] for k, (fit, val) in enumerate(splits): for measure, idx in ( (self.train_scores, fit), @@ -559,11 +587,16 @@ def run_many(self, datas, device): measure.add_scores( model.scores_from_output(out[k][idx], labels[idx])[0] ) + self.val_scores.add_time_per_instance(time_per_instance) + if test: + self.test_scores.add_time_per_instance(time_per_instance) self.last_run_timing = { "train_time_per_fold_s": [per_fit_s] * self.kfold, "train_time_mean_s": per_fit_s, "n_test_instances": len(test), "fused_batch_candidates": len(outputs), + "representation_time_per_instance_s": representation_per_instance, + "val_inference_time_per_instance_s": inference_per_instance[c], } results[c] = [ self.train_scores.compute_averages(), @@ -657,8 +690,37 @@ def _measure_single_sample_latency( "inference_latency_source": source, } + def inference_cost_timing(self, representation_time_per_instance): + """Predicted test runtime: per-instance representation time plus + per-instance validation inference time, scaled by n_test.""" + n_val = sum(len(fold) for fold in self.cv_val_indices) + inference_per_instance = sum(self.val_inference_time) / n_val + n_test = 0 if self.test_indices is None else len(self.test_indices) + return { + "representation_time_per_instance_s": representation_time_per_instance, + "val_inference_time_per_instance_s": inference_per_instance, + "inference_cost": ( + representation_time_per_instance + inference_per_instance + ) + * n_test, + } + + def record_time_per_instance( + self, val_representation_time, test_representation_time, n_val, n_test + ): + """Add the last fold's per-instance time (representation + model + inference) to the val and test measures.""" + self.val_scores.add_time_per_instance( + val_representation_time + self.val_inference_time[-1] / n_val + ) + if n_test: + self.test_scores.add_time_per_instance( + test_representation_time + self.inference_time[-1] / n_test + ) + def _reset_params(self): self.inference_time = [] + self.val_inference_time = [] self.training_time = [] self.training_work = TrainingWorkAccumulator() self.train_scores = PerformanceMeasure("train", self.performance_measures) @@ -666,16 +728,16 @@ def _reset_params(self): self.test_scores = PerformanceMeasure("test", self.performance_measures) def _run_fold(self, model, train_X, train_y, val_X, val_y, test_X, test_y): - train_start = time.time() + train_start = time.perf_counter() train_score = model.fit(train_X, train_y, val_X, val_y) - train_end = time.time() - self.training_time.append(train_end - train_start) + self.training_time.append(time.perf_counter() - train_start) self.train_scores.add_scores(train_score[0]) + val_start = time.perf_counter() val_score = model.test(val_X, val_y) - test_start = time.time() + self.val_inference_time.append(time.perf_counter() - val_start) + test_start = time.perf_counter() test_score = model.test(np.asarray(test_X), test_y) - test_end = time.time() - self.inference_time.append(test_end - test_start) + self.inference_time.append(time.perf_counter() - test_start) self.val_scores.add_scores(val_score[0]) self.test_scores.add_scores(test_score[0]) # No-op for the non-neural models that share this machinery. diff --git a/src/main/python/systemds/scuro/drsearch/test_set_evaluation.py b/src/main/python/systemds/scuro/drsearch/test_set_evaluation.py index 4479727e9d1..d52c7251948 100644 --- a/src/main/python/systemds/scuro/drsearch/test_set_evaluation.py +++ b/src/main/python/systemds/scuro/drsearch/test_set_evaluation.py @@ -20,15 +20,21 @@ # ------------------------------------------------------------- from __future__ import annotations +import copy import time from typing import Any, Dict, List from systemds.scuro.modality.modality import Modality -from systemds.scuro.drsearch.representation_dag import RepresentationDag +from systemds.scuro.drsearch.node_executor import decoded_lazy_leaves +from systemds.scuro.drsearch.representation_dag import ( + RepresentationDag, + pushdown_aggregation, +) from systemds.scuro.drsearch.split_aware_evaluator import ( SplitAwareDAGEvaluator, dag_requires_fitting, ) +from systemds.scuro.drsearch.task import INFERENCE_TIME from systemds.scuro.representations.aggregated_representation import ( AggregatedRepresentation, ) @@ -51,6 +57,28 @@ def _match_expected_dim(modality, task): return modality +def measure_inference_time( + dag: RepresentationDag, modalities: List[Modality], task +) -> float: + """Validation inference time per instance of ``dag``, measured alone. + + ``dag`` must start from the raw ``modalities`` (for a fused result, use the + optimizer's end-to-end DAG). The DAG and the task run in this process, one + after the other, so the value is free of contention from parallel workers. + """ + (dag,) = pushdown_aggregation([copy.deepcopy(dag)]) + task = copy.copy(task) + with decoded_lazy_leaves(modalities): + if dag_requires_fitting(dag, modalities): + scores = SplitAwareDAGEvaluator(dag, modalities, task).evaluate().scores + else: + output = _match_expected_dim( + _unwrap(dag.execute(modalities, task, enable_cache=False)), task + ) + scores = task.run(output.data, output.transform_time) + return float(scores[1].average_scores[INFERENCE_TIME]) + + def measure_representation_time_on_test_set( dag: RepresentationDag, modalities: List[Modality], @@ -67,14 +95,16 @@ def measure_representation_time_on_test_set( if dag_requires_fitting(dag): if task is None: raise ValueError("A task is required to fit the representation DAG") - transformed = SplitAwareDAGEvaluator(dag, modalities, task).fit_transform( - task.train_indices, test_indices - ) + with decoded_lazy_leaves(modalities): + transformed = SplitAwareDAGEvaluator( + dag, modalities, task + ).fit_transform(task.train_indices, test_indices) output = _match_expected_dim(transformed.modality, task) else: - output = _match_expected_dim( - _unwrap(dag.execute(subsets, task, enable_cache=False)), task - ) + with decoded_lazy_leaves(subsets): + output = _match_expected_dim( + _unwrap(dag.execute(subsets, task, enable_cache=False)), task + ) timings.append(time.perf_counter() - t0) timings.sort() @@ -115,15 +145,16 @@ def measure_test_set_application( if full_data is None: t0 = time.perf_counter() - if dag_requires_fitting(dag): - transformed = SplitAwareDAGEvaluator(dag, modalities, task).fit_transform( - task.train_indices, range(len(modalities[0].data)) - ) - output = _match_expected_dim(transformed.modality, task) - else: - output = _match_expected_dim( - _unwrap(dag.execute(modalities, task, enable_cache=False)), task - ) + with decoded_lazy_leaves(modalities): + if dag_requires_fitting(dag): + transformed = SplitAwareDAGEvaluator( + dag, modalities, task + ).fit_transform(task.train_indices, range(len(modalities[0].data))) + output = _match_expected_dim(transformed.modality, task) + else: + output = _match_expected_dim( + _unwrap(dag.execute(modalities, task, enable_cache=False)), task + ) record["full_representation_time_s"] = time.perf_counter() - t0 full_data = None if output is None else output.data diff --git a/src/main/python/systemds/scuro/drsearch/trial_result_cache.py b/src/main/python/systemds/scuro/drsearch/trial_result_cache.py index 9298dbe46c2..6fa2f349d45 100644 --- a/src/main/python/systemds/scuro/drsearch/trial_result_cache.py +++ b/src/main/python/systemds/scuro/drsearch/trial_result_cache.py @@ -58,7 +58,9 @@ def _nbytes(value: Any) -> int: def _clone(value: Any) -> Any: if isinstance(value, IndexedModality): - return IndexedModality(_clone(value.modality), value.instance_ids) + return IndexedModality( + _clone(value.modality), value.instance_ids, value.transform_seconds + ) if isinstance(value, tuple): return tuple(_clone(item) for item in value) if hasattr(value, "_data"): diff --git a/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py b/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py index b097cda4138..64624ee10e6 100644 --- a/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py +++ b/src/main/python/systemds/scuro/drsearch/unimodal_optimizer.py @@ -30,24 +30,31 @@ from functools import lru_cache from weakref import WeakValueDictionary +from systemds.scuro.dataloader.base_loader import LazyFileSequence from systemds.scuro.modality.type import ModalityType from systemds.scuro.drsearch.node_executor import ( NodeExecutor, ResultEntry, - _lazy_leaf_budget_fraction, + decoded_lazy_leaves, ) -from systemds.scuro.dataloader.base_loader import LazyFileSequence -from systemds.scuro.utils.memory_utility import cpu_memory_budget_bytes from systemds.scuro.drsearch.split_aware_evaluator import ( SplitAwareDAGEvaluator, dag_requires_fitting, ) + +from systemds.scuro.drsearch.test_set_evaluation import measure_inference_time from systemds.scuro.drsearch.trial_result_cache import TrialNodeCache, TrialResultCache from systemds.scuro.representations.representation import ( FitRequirement, RepresentationStats, ) from systemds.scuro.drsearch.ranking import rank_by_tradeoff, rank_by_robustness +from systemds.scuro.drsearch.objectives import ( + pareto_front, + rank_by_pareto, + retime_front, + validate_objectives, +) from systemds.scuro.drsearch.task import PerformanceMeasure from systemds.scuro.representations.concatenation import Concatenation from systemds.scuro.representations.hadamard import Hadamard @@ -144,6 +151,7 @@ def __init__( result_path=None, k=2, metric_name="accuracy", + objectives=None, checkpoint_every: Optional[int] = 1, resume: bool = False, max_num_workers: int = -1, @@ -187,7 +195,7 @@ def __init__( self.operator_registry = Registry() self.operator_performance = UnimodalResults( - modalities, tasks, debug, True, k, self.metric_name + modalities, tasks, debug, True, k, self.metric_name, objectives=objectives ) self._tasks_require_same_dims = True self.expected_dimensions = tasks[0].expected_dim @@ -1088,7 +1096,7 @@ def _summarize_windows(self, builder, node_id): ) -SELECTION_STRATEGIES = ("performance", "robustness", "tradeoff") +SELECTION_STRATEGIES = ("performance", "robustness", "tradeoff", "pareto") class UnimodalResults: @@ -1104,6 +1112,7 @@ def __init__( tradeoff_weights=(0.5, 0.5), robustness_std_weight=1.0, robustness_one_se_parsimony=True, + objectives=None, ): self.modality_ids = [modality.modality_id for modality in modalities] self.task_names = [task.model.name for task in tasks] @@ -1113,6 +1122,13 @@ def __init__( self.store_cache = store_cache self.k = k self.metric_name = metric_name + self.objectives = ( + [(metric_name, "max")] + if objectives is None + else validate_objectives(objectives) + ) + if len(self.objectives) > 1 and selection_strategy == "performance": + selection_strategy = "pareto" self.set_selection_strategy( selection_strategy, tradeoff_weights=tradeoff_weights, @@ -1242,6 +1258,8 @@ def with_selection_strategy(self, selection_strategy, **strategy_kwargs): def _rank_entries(self, task_results, performance_metric_name): strategy = getattr(self, "selection_strategy", "performance") + if strategy == "pareto": + return rank_by_pareto(task_results, self.objectives) if strategy == "robustness": return rank_by_robustness( task_results, @@ -1296,8 +1314,7 @@ def get_k_best_results( task_cache = None if not task_cache: cache = [] - lazy_data = self._decode_lazy_leaf_once(modality, results) - try: + with decoded_lazy_leaves([modality]): for result in results: if dag_requires_fitting(result.dag): cache.append( @@ -1305,9 +1322,6 @@ def get_k_best_results( ) else: cache.append(result.dag.execute([modality])) - finally: - if lazy_data is not None: - modality._data = lazy_data elif isinstance(task_cache, list): cache = task_cache @@ -1334,6 +1348,22 @@ def get_k_best_results( return results, cache + def get_pareto_front(self, modality, task, retime_repeats: int = 0): + """Non-dominated results of ``modality`` for ``task`` under ``self.objectives``. + + With ``retime_repeats`` > 0 the front is re-timed serially first (see + ``retime_front``), which updates the re-timed results in place. + """ + results = self.results[modality.modality_id][task.model.name] + if retime_repeats <= 0: + return pareto_front(results, self.objectives) + return retime_front( + results, + self.objectives, + lambda entry: measure_inference_time(entry.dag, [modality], task), + retime_repeats, + ) + def _fit_k_best_leaf(self, dag, modality, task): supervised = any( _fit_requirement(node) is FitRequirement.SUPERVISED for node in dag.nodes @@ -1370,16 +1400,6 @@ def _decode_lazy_leaf_once(modality, results): metadata = ( modality.metadata if len(modality.metadata or []) == len(data) else None ) - decoded_bytes = estimate(metadata) if callable(estimate) else None - budget = cpu_memory_budget_bytes(modality.modality_type.name) - if not decoded_bytes or decoded_bytes > budget * _lazy_leaf_budget_fraction(): - return None - arrays = data.materialize() - for array in arrays: - if hasattr(array, "flags"): - array.flags.writeable = False - modality._data = arrays - return data def get_k_most_robust_results( self, diff --git a/src/main/python/systemds/scuro/modality/indexed.py b/src/main/python/systemds/scuro/modality/indexed.py index a45eb0a0a7b..6815e004d52 100644 --- a/src/main/python/systemds/scuro/modality/indexed.py +++ b/src/main/python/systemds/scuro/modality/indexed.py @@ -25,10 +25,15 @@ @dataclass(frozen=True) class IndexedModality: - """A modality partition paired with its global instance IDs.""" + """A modality partition paired with its global instance IDs. + + ``transform_seconds`` is the time spent transforming this partition, + summed over every node up to it (fitting excluded). + """ modality: Modality instance_ids: tuple + transform_seconds: float = 0.0 def __post_init__(self): instance_ids = tuple(int(i) for i in self.instance_ids) @@ -72,4 +77,5 @@ def subset(self, instance_ids) -> "IndexedModality": return IndexedModality( self.modality.subset(local_indices), requested_ids, + self.transform_seconds * len(requested_ids) / max(1, len(self)), ) diff --git a/src/main/python/systemds/scuro/modality/unimodal_modality.py b/src/main/python/systemds/scuro/modality/unimodal_modality.py index df2c3adb733..8b37da3a56e 100644 --- a/src/main/python/systemds/scuro/modality/unimodal_modality.py +++ b/src/main/python/systemds/scuro/modality/unimodal_modality.py @@ -37,6 +37,14 @@ ) from systemds.scuro.utils.identifier import Identifier +_NO_CHUNK = object() + + +def _timed_call(transform): + start = time.perf_counter() + result = transform() + return result, time.perf_counter() - start + class UnimodalModality(Modality): @@ -274,21 +282,28 @@ def _apply_transform_units(self, units, parallel): padding_per_representation[representation_key] = False original_lengths_per_representation[representation_key] = [] - start = ( - time.time() - ) # TODO: should be repalced in unimodal_representation.transform + # Each key is charged its own transform time plus the shared chunk loading, + # i.e. what it would cost if it ran alone. + transform_seconds = dict.fromkeys( + transformed_modalities_per_representation, 0.0 + ) + load_seconds = 0.0 with ( ThreadPoolExecutor(max_workers=len(units)) if parallel and len(units) > 1 else nullcontext() ) as executor: time_s = time.time() - chunks = ( + chunks = iter( (None,) if not self.data_loader.is_chunked and self.has_data() else self.iter_raw_data_chunks(reset=True) ) - for _ in chunks: + while True: + load_start = time.perf_counter() + if next(chunks, _NO_CHUNK) is _NO_CHUNK: + break + load_seconds += time.perf_counter() - load_start representations_futures = {} serial_transforms = [] for keys, _, transform in units: @@ -298,11 +313,13 @@ def _apply_transform_units(self, units, parallel): if not pending_keys: continue if executor is None: - serial_transforms.append((pending_keys, transform)) - else: - representations_futures[executor.submit(transform)] = ( - pending_keys + serial_transforms.append( + (pending_keys, partial(_timed_call, transform)) ) + else: + representations_futures[ + executor.submit(_timed_call, transform) + ] = pending_keys completed = ( ( (representations_futures[future], future.result) @@ -313,12 +330,13 @@ def _apply_transform_units(self, units, parallel): ) for pending_keys, transform in completed: try: - transformed_chunks = transform() + transformed_chunks, seconds = transform() except Exception as e: for representation_key in pending_keys: failed_representations[representation_key] = e continue for representation_key in pending_keys: + transform_seconds[representation_key] += seconds transformed_chunk = transformed_chunks.get(representation_key) if transformed_chunk is None: failed_representations[representation_key] = ValueError( @@ -357,7 +375,9 @@ def _apply_transform_units(self, units, parallel): ) transformed_modalities_per_representation[ representation_key - ].transform_time += (time.time() - start) + ].transform_time += ( + transform_seconds[representation_key] + load_seconds + ) transformed_modalities_per_representation[ representation_key ].self_contained = representation.self_contained diff --git a/src/main/python/systemds/scuro/models/fused_mlp.py b/src/main/python/systemds/scuro/models/fused_mlp.py index 7688f98b8a4..8cdfb75faf5 100644 --- a/src/main/python/systemds/scuro/models/fused_mlp.py +++ b/src/main/python/systemds/scuro/models/fused_mlp.py @@ -46,6 +46,7 @@ import math import os +import time import numpy as np import torch @@ -164,7 +165,8 @@ def fit_folds(model, features, labels, splits, seed, device, eval_elements=2**27 features: C arrays of shape rows x d (one d for all), labels: rows (x O), splits: K (fit_positions, val_positions) into the rows. Returns the - eval-mode outputs, C x K x rows x O, as a numpy array. + eval-mode outputs, C x K x rows x O, as a numpy array, and the seconds the + eval pass spent per row and model (an amortized inference estimate). """ n_candidates, n_folds = len(features), len(splits) n_rows, dim = features[0].shape @@ -247,9 +249,13 @@ def noise(width): [dim] + [layer[1].shape[-1] for layer in net.layers if layer[0] == "linear"] ) chunk = max(1, eval_elements // (groups * widest)) + eval_start = time.perf_counter() with torch.no_grad(): for start in range(0, n_rows, chunk): xb = (X[cand, start : start + chunk] - mu) / sd outputs.append(net.forward(xb).cpu()) + eval_seconds = time.perf_counter() - eval_start out = torch.cat(outputs, dim=1)[: n_candidates * n_folds].numpy() - return out.reshape(n_candidates, n_folds, n_rows, -1) + return out.reshape(n_candidates, n_folds, n_rows, -1), eval_seconds / ( + groups * n_rows + ) diff --git a/src/main/python/tests/scuro/test_chunked_leaf_execution.py b/src/main/python/tests/scuro/test_chunked_leaf_execution.py index 91bf0a2f2bf..13a5635db45 100644 --- a/src/main/python/tests/scuro/test_chunked_leaf_execution.py +++ b/src/main/python/tests/scuro/test_chunked_leaf_execution.py @@ -153,7 +153,7 @@ def estimate_peak_memory_bytes(self, input_stats): def get_output_stats(self, input_stats): return RepresentationStats(input_stats.num_instances, (1,)) - def run(self, data): + def run(self, data, representation_time=0.0): count = float(len(data)) scores = [] for split in ("train", "val", "test"): diff --git a/src/main/python/tests/scuro/test_hp_tuner.py b/src/main/python/tests/scuro/test_hp_tuner.py index 1e0bad3991e..0c56858d658 100644 --- a/src/main/python/tests/scuro/test_hp_tuner.py +++ b/src/main/python/tests/scuro/test_hp_tuner.py @@ -51,6 +51,7 @@ from systemds.scuro.modality.type import ModalityType from systemds.scuro.drsearch.hyperparameter_tuner import HyperparameterTuner +from systemds.scuro.drsearch.objectives import INFERENCE_TIME, dominates from systemds.scuro.drsearch.representation_dag import CSEAwareDAGBuilder from systemds.scuro.representations.bert import Bert from systemds.scuro.representations.clip import CLIPText @@ -93,6 +94,49 @@ def test_hp_tuner_for_text_modality(self): ) self.run_hp_for_modality([text]) + def test_multi_objective_hp_tuning_keeps_a_pareto_front(self): + objectives = [("accuracy", "max"), (INFERENCE_TIME, "min")] + text_data, text_md = ModalityRandomDataGenerator().create_text_data( + self.num_instances + ) + text = UnimodalModality( + TestDataLoader( + self.indices, None, ModalityType.TEXT, text_data, str, text_md + ) + ) + with patch.object( + Registry, "_representations", {ModalityType.TEXT: [BoW, W2V]} + ), patch.object(Registry, "_fusion_operators", []): + unimodal_optimizer = UnimodalOptimizer( + [text], self.tasks, False, k=2, max_num_workers=1, objectives=objectives + ) + unimodal_optimizer.optimize() + hp = HyperparameterTuner( + [text], + self.tasks, + unimodal_optimizer.operator_performance, + k=2, + n_jobs=1, + objectives=objectives, + ) + hp.tune_unimodal_representations(max_eval_per_rep=4) + + for task in self.tasks: + results = hp.optimization_results.results[task.model.name][text.modality_id] + self.assertTrue(results) + for result in results: + front = result.pareto_results + self.assertTrue(front) + self.assertIn(result.best_params, [params for params, _ in front]) + vectors = [ + tuple(scores.all_val_scores[name] for name, _ in objectives) + for _, scores in front + ] + for a in vectors: + self.assertFalse(any(dominates(b, a, objectives) for b in vectors)) + _, dags = hp.optimization_results.get_k_best_dags(text, task) + self.assertLessEqual(len(dags), 2 * len(results)) + # TODO: Add once the final multimodal optimizer is implemented # def test_multimodal_hp_tuning(self): # audio_data, audio_md = ModalityRandomDataGenerator().create_audio_data( diff --git a/src/main/python/tests/scuro/test_hpo_cache.py b/src/main/python/tests/scuro/test_hpo_cache.py index 67240adf4f0..617c9b990d5 100644 --- a/src/main/python/tests/scuro/test_hpo_cache.py +++ b/src/main/python/tests/scuro/test_hpo_cache.py @@ -128,7 +128,7 @@ def test_hpo_reuses_unchanged_dag_prefix(self): dag = builder.build(scaling_id) task = SimpleNamespace( model=SimpleNamespace(name="cache_task"), - run=lambda data: [float(np.mean(data))] * 3, + run=lambda data, representation_time=0.0: [float(np.mean(data))] * 3, ) tuner = HyperparameterTuner( [modality], diff --git a/src/main/python/tests/scuro/test_inference_cost.py b/src/main/python/tests/scuro/test_inference_cost.py new file mode 100644 index 00000000000..9c876507298 --- /dev/null +++ b/src/main/python/tests/scuro/test_inference_cost.py @@ -0,0 +1,189 @@ +# ------------------------------------------------------------- +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +# ------------------------------------------------------------- +import time +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +import numpy as np + +from systemds.scuro.drsearch.objectives import SEARCH_INFERENCE_TIME +from systemds.scuro.drsearch.representation_dag import RepresentationDAGBuilder +from systemds.scuro.drsearch.split_aware_evaluator import SplitAwareDAGEvaluator +from systemds.scuro.drsearch.task import INFERENCE_TIME +from systemds.scuro.drsearch.trial_result_cache import _clone +from systemds.scuro.drsearch.unimodal_optimizer import UnimodalResults +from systemds.scuro.modality.indexed import IndexedModality +from systemds.scuro.modality.type import ModalityType +from systemds.scuro.modality.unimodal_modality import UnimodalModality +from systemds.scuro.representations.bow import BoW +from systemds.scuro.representations.tfidf import TfIdf +from tests.scuro.data_generator import ( + ModalityRandomDataGenerator, + TestDataLoader, + TestTask, +) + +NUM_INSTANCES = 20 + + +def _text_modality(): + data, metadata = ModalityRandomDataGenerator().create_text_data(NUM_INSTANCES, 10) + return UnimodalModality( + TestDataLoader( + np.arange(NUM_INSTANCES), None, ModalityType.TEXT, data, str, metadata + ) + ) + + +def _tfidf_dag(modality): + builder = RepresentationDAGBuilder() + return builder.build( + builder.create_operation_node( + TfIdf, [builder.create_leaf_node(modality.modality_id)], {} + ) + ) + + +def _slowed(method, seconds): + def wrapper(*args, **kwargs): + time.sleep(seconds) + return method(*args, **kwargs) + + return wrapper + + +class TestInferenceCost(unittest.TestCase): + def test_split_path_records_time_per_instance(self): + text = _text_modality() + task = TestTask("InferenceCostTask", "Test1", NUM_INSTANCES) + + evaluation = SplitAwareDAGEvaluator(_tfidf_dag(text), text, task).evaluate() + + val, test = evaluation.scores[1], evaluation.scores[2] + self.assertEqual(len(val.time_per_instance), task.kfold) + self.assertEqual(len(test.time_per_instance), task.kfold) + self.assertAlmostEqual( + val.average_scores[INFERENCE_TIME], np.mean(val.time_per_instance) + ) + self.assertIn(INFERENCE_TIME, val.fold_scores()) + self.assertIn(INFERENCE_TIME, val.score_stds()) + + timing = task.last_run_timing + self.assertGreater(timing["representation_time_per_instance_s"], 0.0) + self.assertGreater(timing["val_inference_time_per_instance_s"], 0.0) + self.assertAlmostEqual( + evaluation.inference_cost, + ( + timing["representation_time_per_instance_s"] + + timing["val_inference_time_per_instance_s"] + ) + * len(task.test_indices), + ) + + def test_split_path_excludes_fit_time(self): + text = _text_modality() + task = TestTask("InferenceCostTask", "Test1", NUM_INSTANCES) + + with patch.object(TfIdf, "fit", _slowed(TfIdf.fit, 0.2)): + evaluation = SplitAwareDAGEvaluator(_tfidf_dag(text), text, task).evaluate() + + n_transformed = task.kfold * (len(task.train_indices) + len(task.test_indices)) + transform_total = ( + task.last_run_timing["representation_time_per_instance_s"] * n_transformed + ) + self.assertGreaterEqual(evaluation.representation_time, 0.2 * task.kfold) + self.assertLess(transform_total, 0.2) + + def test_split_path_charges_leaf_loading(self): + text = _text_modality() + task = TestTask("InferenceCostTask", "Test1", NUM_INSTANCES) + loader = text.data_loader + + with patch.object(loader, "load", _slowed(loader.load, 0.2)): + SplitAwareDAGEvaluator(_tfidf_dag(text), text, task).evaluate() + + self.assertGreaterEqual( + task.last_run_timing["representation_time_per_instance_s"], + 0.2 / NUM_INSTANCES, + ) + + def test_unimodal_front_is_retimed_serially(self): + text = _text_modality() + task = TestTask("InferenceCostTask", "Test1", NUM_INSTANCES) + results = UnimodalResults( + [text], [task], objectives=[("accuracy", "max"), (INFERENCE_TIME, "min")] + ) + entry = SimpleNamespace( + dag=_tfidf_dag(text), val_score={"accuracy": 0.5, INFERENCE_TIME: 100.0} + ) + results.results[text.modality_id][task.model.name].append(entry) + + loader = text.data_loader + with patch.object(loader, "load", _slowed(loader.load, 0.2)): + front = results.get_pareto_front(text, task, retime_repeats=1) + + self.assertEqual(front, [entry]) + self.assertEqual(entry.val_score[SEARCH_INFERENCE_TIME], 100.0) + self.assertGreaterEqual(entry.val_score[INFERENCE_TIME], 0.2 / NUM_INSTANCES) + self.assertLess(entry.val_score[INFERENCE_TIME], 100.0) + self.assertFalse(text.has_data()) + + def test_task_run_records_time_per_instance(self): + task = TestTask("InferenceCostTask", "Test1", NUM_INSTANCES) + data = list(np.random.default_rng(0).random((NUM_INSTANCES, 4))) + + _, val, test = task.run(data, representation_time=2.0) + + representation_per_instance = 2.0 / NUM_INSTANCES + self.assertEqual(len(val.time_per_instance), task.kfold) + for fold, seconds in enumerate(val.time_per_instance): + self.assertAlmostEqual( + seconds, + representation_per_instance + + task.val_inference_time[fold] / len(task.cv_val_indices[fold]), + ) + self.assertGreater(test.average_scores[INFERENCE_TIME], 0.1) + + def test_subset_prorates_and_cache_clone_keeps_transform_seconds(self): + text = _text_modality() + text.extract_raw_data() + full = IndexedModality(text, range(NUM_INSTANCES), 2.0) + + part = full.subset(range(5)) + + self.assertAlmostEqual(part.transform_seconds, 0.5) + self.assertAlmostEqual(_clone(part).transform_seconds, 0.5) + + def test_leaf_batch_charges_each_representation_its_own_time(self): + text = _text_modality() + text.extract_raw_data() + + slow, fast = TfIdf(), BoW() + with patch.object(TfIdf, "transform", _slowed(TfIdf.transform, 0.3)): + results = text.apply_representations([slow, fast]) + + self.assertGreaterEqual(results[slow.name].transform_time, 0.3) + self.assertLess(results[fast.name].transform_time, 0.3) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/main/python/tests/scuro/test_multimodal_ga_optimizer.py b/src/main/python/tests/scuro/test_multimodal_ga_optimizer.py index 0affa8acaf2..bef5d9f27b0 100644 --- a/src/main/python/tests/scuro/test_multimodal_ga_optimizer.py +++ b/src/main/python/tests/scuro/test_multimodal_ga_optimizer.py @@ -41,6 +41,7 @@ _objective_value, _task_uses_gpu, ) +from systemds.scuro.drsearch.task import INFERENCE_TIME from systemds.scuro.drsearch.operator_registry import Registry, register_fusion_operator from systemds.scuro.modality.type import ModalityType from systemds.scuro.representations.average import Average @@ -572,6 +573,62 @@ def body(*_a, acc=accuracy, rt=runtime, **_kw): } self.assertEqual(front, {(0.9, 10.0), (0.5, 1.0)}) + def test_retime_hall_of_fame_rebuilds_the_front_from_serial_times(self): + optimizer, _, task = _make_optimizer( + objectives=[("accuracy", "max"), (INFERENCE_TIME, "min")] + ) + optimizer.fusion_operators = [Concatenation] + name = task.model.name + + # (accuracy, search time, serial time); "noisy_fast" dominates both + # others during the search, "hidden" only reaches the front once + # "steady" has been re-timed. + points = { + "noisy_fast": (0.9, 1.0, 3.0), + "steady": (0.85, 2.0, 2.6), + "hidden": (0.8, 2.5, 1.5), + } + serial = {} + for label, (accuracy, seconds, serial_seconds) in points.items(): + genome = optimizer._random_genome(name) + + def body(dag, *_a, acc=accuracy, sec=seconds, **_kw): + return (acc, sec), { + "train_score": {}, + "val_score": {"accuracy": acc, INFERENCE_TIME: sec}, + "test_score": {}, + "runtime": 0.0, + "task_time": 0.0, + "representation_time": 0.0, + } + + with patch(f"{MODULE}._evaluate_genome_body", side_effect=body): + optimizer._evaluate_genome(genome, task) + serial[id(optimizer.optimization_results[name][-1].dag)] = ( + label, + serial_seconds, + ) + self.assertEqual(len(optimizer.optimization_results[name]), 3) + self.assertEqual(len(optimizer.get_hall_of_fame(name)), 1) + + with patch.object( + optimizer, "compose_end_to_end_dag", side_effect=lambda dag, _task: dag + ), patch( + f"{MODULE}.measure_inference_time", + side_effect=lambda dag, _mods, _task: serial[id(dag)][1], + ): + optimizer.retime_hall_of_fame(task, repeats=1) + + hof = optimizer.get_hall_of_fame(name) + self.assertEqual( + {serial[id(r.dag)][0]: r.val_score[INFERENCE_TIME] for r in hof}, + {"noisy_fast": 3.0, "steady": 2.6, "hidden": 1.5}, + ) + self.assertEqual( + sorted(optimizer._hof_fitness[name]), + [(0.8, 1.5), (0.85, 2.6), (0.9, 3.0)], + ) + def test_failed_evaluations_never_enter_the_hall_of_fame(self): optimizer, _, task = _make_optimizer() optimizer.fusion_operators = [Concatenation] diff --git a/src/main/python/tests/scuro/test_unimodal_optimizer.py b/src/main/python/tests/scuro/test_unimodal_optimizer.py index 27c19c5e1cf..68ea1197d45 100644 --- a/src/main/python/tests/scuro/test_unimodal_optimizer.py +++ b/src/main/python/tests/scuro/test_unimodal_optimizer.py @@ -20,6 +20,8 @@ # ------------------------------------------------------------- +import os +import time import unittest from types import SimpleNamespace @@ -27,6 +29,7 @@ from systemds.scuro.representations.color_histogram import ColorHistogram from systemds.scuro.drsearch.operator_registry import Registry, load_builtin_operators from systemds.scuro.drsearch.node_executor import ResultEntry +from systemds.scuro.drsearch.task import INFERENCE_TIME from systemds.scuro.drsearch.unimodal_optimizer import ( UnimodalOptimizer, UnimodalResults, @@ -232,6 +235,7 @@ def k_best(decoded_bytes): def decode(name): decoded.append(name) + time.sleep(0.001) return images[int(name)].copy() lazy = LazyFileSequence([str(i) for i in self.indices], decode) @@ -248,8 +252,12 @@ def decode(name): results.results[image.modality_id][task.model.name].append( SimpleNamespace(dag=dag, val_score={"accuracy": accuracy}) ) - _, cache = results.get_k_best_results(image, task, "accuracy") + with patch.dict(os.environ, {"SCURO_DECODE_THREADS": "1"}): + _, cache = results.get_k_best_results(image, task, "accuracy") self.assertIs(image.data, lazy) + self.assertEqual(image.transform_time, 0) + for c in cache: + self.assertGreaterEqual(c.transform_time, 0.001 * self.num_instances) return len(decoded), [np.asarray(c.data) for c in cache] lazy_decodes, lazy_outputs = k_best(None) @@ -924,3 +932,25 @@ def test_views_share_the_candidate_pool_and_leave_the_source_alone(self): def test_unknown_strategy_is_rejected(self): with self.assertRaises(ValueError): self.results.with_selection_strategy("cheapest") + + def test_two_objectives_select_from_the_pareto_front(self): + seconds = {"noisy_slow": 2.0, "stable_fast": 0.5, "weak_fastest": 0.6} + for name, entry in self.entries.items(): + entry.val_score[INFERENCE_TIME] = seconds[name] + results = UnimodalResults( + [self.modality], + [self.task], + k=3, + objectives=[("accuracy", "max"), (INFERENCE_TIME, "min")], + ) + results.results["m1"]["t1"] = list(self.entries.values()) + + self.assertEqual(results.selection_strategy, "pareto") + ranked, _ = results.get_k_best_results( + self.modality, self.task, "accuracy", cache_needed=False + ) + self.assertIs(ranked[-1], self.entries["weak_fastest"]) + self.assertEqual( + results.get_pareto_front(self.modality, self.task), + ranked[:2], + )