• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2024 The Chromium Authors
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5from __future__ import annotations
6
7from abc import ABCMeta, abstractmethod
8import argparse
9import re
10from typing import Any, Dict, Hashable, List, Pattern, TYPE_CHECKING, Type, Union
11
12from immutabledict import immutabledict
13
14from crossbench import exception
15from crossbench import path as pth
16from crossbench.compat import StrEnum
17from crossbench.config import ConfigObject
18from crossbench.parse import NumberParser, ObjectParser
19if TYPE_CHECKING:
20  from crossbench.plt.base import Platform
21
22# Directory exposing info & controls for the frequency of all CPUs.
23_CPUS_DIR: pth.AnyPosixPath = pth.AnyPosixPath("/sys/devices/system/cpu")
24
25# Used to specify behavior for all CPUs.
26_WILDCARD_CONFIG_KEY = "*"
27
28# Matches the CPU names exposed by the system in _CPUS_DIR.
29_CPU_NAME_REGEX: Pattern[str] = re.compile("cpu[0-9]+$")
30
31
32class _ExtremeFrequency(StrEnum):
33  MAX = "max"
34  MIN = "min"
35
36
37if TYPE_CHECKING:
38  FrequencyType = Union[_ExtremeFrequency, int]
39
40
41class CPUFrequencyMap(ConfigObject, metaclass=ABCMeta):
42
43  @abstractmethod
44  def get_target_frequencies(
45      self, platform: Platform) -> immutabledict[pth.AnyPosixPath, int]:
46    raise NotImplementedError()
47
48  @property
49  @abstractmethod
50  def key(self) -> Hashable:
51    raise NotImplementedError()
52
53  @classmethod
54  def parse_dict(cls: Type[CPUFrequencyMap],
55                 config: Dict[str, Any]) -> CPUFrequencyMap:
56    if _WILDCARD_CONFIG_KEY in config:
57      return WildcardCPUFrequencyMap(config)
58
59    return ExplicitCPUFrequencyMap(config)
60
61  @classmethod
62  def parse_str(cls: Type[CPUFrequencyMap], value: str) -> CPUFrequencyMap:
63    return CPUFrequencyMap.parse_dict({_WILDCARD_CONFIG_KEY: value})
64
65  @classmethod
66  def _parse_frequency(cls, value: Any) -> FrequencyType:
67    if value == _ExtremeFrequency.MIN:
68      return _ExtremeFrequency.MIN
69
70    if value == _ExtremeFrequency.MAX:
71      return _ExtremeFrequency.MAX
72
73    try:
74      return NumberParser.positive_zero_int(value)
75    except argparse.ArgumentTypeError as e:
76      raise argparse.ArgumentTypeError(
77          f"Invalid value in CPU frequency map: {value}. Should "
78          "have been one of \"max\"|\"min\"|<int>|\"<int>\"") from e
79
80  def _get_target_frequency(self, platform: Platform, cpu_name: str,
81                            frequency: FrequencyType) -> int:
82    if not platform.exists(_CPUS_DIR):
83      # TODO(crbug.com/372862708): If different devices indeed use different
84      # dirs, consider making this configurable in the jSON.
85      raise FileNotFoundError(
86          f"{_CPUS_DIR} not found. Either {platform} does not support setting "
87          "CPU frequency or the CPUs are exposed in another path and that "
88          "requires extra support.")
89
90    cpu_dir: pth.AnyPosixPath = self._get_cpu_dir(cpu_name)
91    if not platform.is_dir(cpu_dir):
92      raise ValueError(f"Invalid CPU name: {cpu_name}.")
93
94    available_frequencies: List[int] = [
95        NumberParser.positive_zero_int(f)
96        for f in platform.cat(cpu_dir / "scaling_available_frequencies").rstrip(
97            "\n").rstrip(" ").split(" ")
98    ]
99    if frequency == _ExtremeFrequency.MIN:
100      return min(available_frequencies)
101    if frequency == _ExtremeFrequency.MAX:
102      return max(available_frequencies)
103    if frequency in available_frequencies:
104      assert isinstance(frequency, int)
105      return frequency
106    raise ValueError(f"Target frequency {frequency} for {cpu_name} "
107                     f"not allowed in {platform}. Available frequencies: "
108                     f"{available_frequencies}")
109
110  def _get_cpu_dir(self, cpu_name: str) -> pth.AnyPosixPath:
111    # Create new AnyPosixPath so pyfakefs is happy in tests.
112    return pth.AnyPosixPath(_CPUS_DIR / cpu_name / "cpufreq")
113
114
115class WildcardCPUFrequencyMap(CPUFrequencyMap):
116
117  def __init__(self, frequencies: Dict):
118    if len(frequencies) != 1:
119      raise argparse.ArgumentTypeError(
120          f"A wildcard ({_WILDCARD_CONFIG_KEY}) in "
121          "the CPU frequency map should be the only key.")
122
123    self._target_frequency = CPUFrequencyMap._parse_frequency(
124        list(frequencies.values())[0])
125
126  def get_target_frequencies(
127      self, platform: Platform) -> immutabledict[pth.AnyPosixPath, int]:
128    return immutabledict({
129        self._get_cpu_dir(p.name):
130            self._get_target_frequency(platform, p.name, self._target_frequency)
131        for p in platform.iterdir(_CPUS_DIR)
132        if _CPU_NAME_REGEX.match(p.name)
133    })
134
135  @property
136  def key(self) -> Hashable:
137    return self._target_frequency
138
139
140class ExplicitCPUFrequencyMap(CPUFrequencyMap):
141
142  def __init__(self, frequencies: Dict):
143    typed_map: Dict[str, FrequencyType] = {}
144    for k, v in frequencies.items():
145      with exception.annotate_argparsing(f"Parsing cpu frequency: {k}, {v}"):
146        typed_map[ObjectParser.non_empty_str(k)] = (
147            CPUFrequencyMap._parse_frequency(v))
148    self._frequencies: immutabledict[str, Union[_ExtremeFrequency,
149                                                int]] = immutabledict(typed_map)
150
151  def get_target_frequencies(
152      self, platform: Platform) -> immutabledict[pth.AnyPosixPath, int]:
153    return immutabledict({
154        self._get_cpu_dir(cpu_name):
155            self._get_target_frequency(platform, cpu_name, config_frequeny)
156        for cpu_name, config_frequeny in self._frequencies.items()
157    })
158
159  @property
160  def key(self) -> Hashable:
161    return self._frequencies
162