resources/libraries/python/MLRsearch/measurement_database.py

   1 # Copyright (c) 2023 Cisco and/or its affiliates.
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at:
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 # See the License for the specific language governing permissions and
  12 # limitations under the License.
  13
  14 """Module defining MeasurementDatabase class."""
  15
  16 from dataclasses import dataclass
  17 from typing import Dict, Tuple
  18
  19 from .discrete_load import DiscreteLoad
  20 from .discrete_result import DiscreteResult
  21 from .load_stats import LoadStats
  22 from .relevant_bounds import RelevantBounds
  23 from .target_spec import TargetSpec
  24 from .trimmed_stat import TrimmedStat
  25
  26
  27 @dataclass
  28 class MeasurementDatabase:
  29     """Structure holding measurement results for multiple durations and loads.
  30
  31     Several utility methods are added, accomplishing tasks useful for MLRsearch.
  32
  33     While TargetStats can decide when a single load is a lower bound (or upper),
  34     it does not deal with loss inversion (higher load with less load).
  35
  36     This class introduces the concept of relevant bounds.
  37     Relevant upper bound is simply the lowest load classified as an upper bound.
  38     But relevant lower bound is only chosen from lower bound loads
  39     strictly smaller than the relevant upper bound.
  40     This way any higher loads with good results are ignored,
  41     so relevant bound give conservative estimate of SUT true performance.
  42     """
  43
  44     targets: Tuple[TargetSpec] = None
  45     """Targets to track stats for."""
  46     load_to_stats: Dict[DiscreteLoad, LoadStats] = None
  47     """Mapping from loads to stats."""
  48
  49     def __post_init__(self) -> None:
  50         """Check and sort initial values.
  51
  52         If no stats yet, initialize empty ones.
  53
  54         :raises ValueError: If there are no targets.
  55         """
  56         if not self.targets:
  57             raise ValueError(f"Database needs targets: {self.targets!r}")
  58         if not self.load_to_stats:
  59             self.load_to_stats = {}
  60         self._sort()
  61
  62     def _sort(self) -> None:
  63         """Sort keys from low to high load."""
  64         self.load_to_stats = dict(sorted(self.load_to_stats.items()))
  65
  66     def __getitem__(self, key: DiscreteLoad) -> LoadStats:
  67         """Allow access to stats as if self was load_to_stats.
  68
  69         This also accepts LoadStats as key, so callers do not need
  70         to care about hashability.
  71
  72         :param key: The load to get stats for.
  73         :type key: DiscreteLoad
  74         :returns: Stats for the given load.
  75         :rtype LoadStats:
  76         """
  77         return self.load_to_stats[key.hashable()]
  78
  79     def add(self, result: DiscreteResult) -> None:
  80         """Incorporate given trial measurement result.
  81
  82         :param result: Measurement result to add to the database.
  83         :type result: DiscreteResult
  84         """
  85         discrete_load = result.discrete_load.hashable()
  86         if not discrete_load.is_round:
  87             raise ValueError(f"Not round load: {discrete_load!r}")
  88         if discrete_load not in self.load_to_stats:
  89             self.load_to_stats[discrete_load] = LoadStats.new_empty(
  90                 load=discrete_load,
  91                 targets=self.targets,
  92             )
  93             self._sort()
  94         self.load_to_stats[discrete_load].add(result)
  95
  96     def get_relevant_bounds(self, target: TargetSpec) -> RelevantBounds:
  97         """Return None or a valid trimmed stat, for the two relevant bounds.
  98
  99         A load is valid only if both optimistic and pessimistic estimates agree.
 100
 101         If some value is not available, None is returned instead.
 102         The returned stats are trimmed to the argument target.
 103
 104         The implementation starts from low loads
 105         and the search stops at lowest upper bound,
 106         thus conforming to the conservative definition of relevant bounds.
 107
 108         :param target: Target to classify loads when finding bounds.
 109         :type target: TargetSpec
 110         :returns: Relevant lower bound, relevant upper bound.
 111         :rtype: RelevantBounds
 112         """
 113         lower_bound, upper_bound = None, None
 114         for load_stats in self.load_to_stats.values():
 115             opt, pes = load_stats.estimates(target)
 116             if opt != pes:
 117                 continue
 118             if not opt:
 119                 upper_bound = load_stats
 120                 break
 121             lower_bound = load_stats
 122         if lower_bound:
 123             lower_bound = TrimmedStat.for_target(lower_bound, target)
 124         if upper_bound:
 125             upper_bound = TrimmedStat.for_target(upper_bound, target)
 126         return RelevantBounds(clo=lower_bound, chi=upper_bound)