Skip to content

rapl

zeus.device.cpu.rapl

RAPL CPUs.

  • RAPL (Running Average Power Limit): RAPL is a technology introduced by Intel that allows for power consumption monitoring and control at the processor and memory subsystem level. It provides mechanisms to enforce power limits and manage thermal conditions effectively.

  • Power Zone: A power zone in the context of RAPL refers to a logical grouping of components within the CPU or system that share a common power domain. Each power zone can be monitored and controlled independently. Typical power zones include the entire package, specific cores, and memory subsystems.

  • Package: The package refers to the physical CPU chip, which may contain multiple cores and integrated components. In RAPL, the package power domain encompasses the power consumption of all the cores and integrated units within the CPU package.

RaplWraparoundTracker

Monitor the wrapping around of RAPL counters.

This class acts as a lower level wrapper around a Python process that polls the wrapping of RAPL counters. This is primarily used by RAPLCPUs.

Warning

Since the monitor spawns a child process, it should not be instantiated as a global variable. Python puts a protection to prevent creating a process in global scope. Refer to the "Safe importing of main module" section in the Python documentation for more details.

Attributes:

Name Type Description
rapl_file_path str

File path of rapl file to track wraparounds for.
max_energy_uj float

Max value of rapl counter for rapl_file_path file. Used to determine the sleep period between polls
Source code in zeus/device/cpu/rapl.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class RaplWraparoundTracker:
    """Monitor the wrapping around of RAPL counters.

    This class acts as a lower level wrapper around a Python process that polls
    the wrapping of RAPL counters. This is primarily used by
    [`RAPLCPUs`][zeus.device.cpu.rapl.RAPLCPUs].

    !!! Warning
        Since the monitor spawns a child process, **it should not be instantiated as a global variable**.
        Python puts a protection to prevent creating a process in global scope.
        Refer to the "Safe importing of main module" section in the
        [Python documentation](https://docs.python.org/3/library/multiprocessing.html#the-spawn-and-forkserver-start-methods)
        for more details.

    Attributes:
        rapl_file_path (str): File path of rapl file to track wraparounds for.
        max_energy_uj (float): Max value of rapl counter for `rapl_file_path` file. Used to
            determine the sleep period between polls
    """

    def __init__(
        self,
        rapl_file_path: str,
        max_energy_uj: float,
    ) -> None:
        """Initialize the rapl monitor.

        Args:
            rapl_file_path: File path where the RAPL file is located
            max_energy_uj: Max energy range uj value
        """
        if not os.path.exists(rapl_file_path):
            raise ValueError(f"{rapl_file_path} is not a valid file path")

        # Set up logging.
        self.logger = get_logger(type(self).__name__)

        self.logger.info("Monitoring wrap around of %s", rapl_file_path)

        context = mp.get_context("spawn")
        self.wraparound_counter = context.Value("i", 0)
        # Spawn the power polling process.
        atexit.register(self._stop)
        self.process = context.Process(
            target=_polling_process,
            args=(rapl_file_path, max_energy_uj, self.wraparound_counter),
        )
        self.process.start()

    def _stop(self) -> None:
        """Stop monitoring power usage."""
        if self.process is not None:
            self.process.terminate()
            self.process.join(timeout=1.0)
            self.process.kill()
            self.process = None

    def get_num_wraparounds(self) -> int:
        """Get the number of wraparounds detected by the polling process."""
        with self.wraparound_counter.get_lock():
            return self.wraparound_counter.value

__init__ 

__init__(rapl_file_path, max_energy_uj)

Parameters:

Name Type Description Default
rapl_file_path str

File path where the RAPL file is located

 required
max_energy_uj float

Max energy range uj value

 required
Source code in zeus/device/cpu/rapl.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def __init__(
    self,
    rapl_file_path: str,
    max_energy_uj: float,
) -> None:
    """Initialize the rapl monitor.

    Args:
        rapl_file_path: File path where the RAPL file is located
        max_energy_uj: Max energy range uj value
    """
    if not os.path.exists(rapl_file_path):
        raise ValueError(f"{rapl_file_path} is not a valid file path")

    # Set up logging.
    self.logger = get_logger(type(self).__name__)

    self.logger.info("Monitoring wrap around of %s", rapl_file_path)

    context = mp.get_context("spawn")
    self.wraparound_counter = context.Value("i", 0)
    # Spawn the power polling process.
    atexit.register(self._stop)
    self.process = context.Process(
        target=_polling_process,
        args=(rapl_file_path, max_energy_uj, self.wraparound_counter),
    )
    self.process.start()

_stop 

_stop()

Stop monitoring power usage.

Source code in zeus/device/cpu/rapl.py 
 97
 98
 99
100
101
102
103
def _stop(self) -> None:
    """Stop monitoring power usage."""
    if self.process is not None:
        self.process.terminate()
        self.process.join(timeout=1.0)
        self.process.kill()
        self.process = None

get_num_wraparounds 

get_num_wraparounds()

Get the number of wraparounds detected by the polling process.

Source code in zeus/device/cpu/rapl.py 
105
106
107
108
def get_num_wraparounds(self) -> int:
    """Get the number of wraparounds detected by the polling process."""
    with self.wraparound_counter.get_lock():
        return self.wraparound_counter.value

ZeusRAPLNotSupportedError

Bases: ZeusBaseCPUError

Zeus CPU exception class wrapper for RAPL not supported on CPU.

Source code in zeus/device/cpu/rapl.py 
143
144
145
146
147
148
class ZeusRAPLNotSupportedError(ZeusBaseCPUError):
    """Zeus CPU exception class wrapper for RAPL not supported on CPU."""

    def __init__(self, message: str) -> None:
        """Initialize Zeus Exception."""
        super().__init__(message)

__init__ 

__init__(message)
Source code in zeus/device/cpu/rapl.py 
146
147
148
def __init__(self, message: str) -> None:
    """Initialize Zeus Exception."""
    super().__init__(message)

ZeusRAPLFileInitError

Bases: ZeusBaseCPUError

Zeus CPU exception class wrapper for RAPL file initialization error on CPU.

Source code in zeus/device/cpu/rapl.py 
151
152
153
154
155
156
class ZeusRAPLFileInitError(ZeusBaseCPUError):
    """Zeus CPU exception class wrapper for RAPL file initialization error on CPU."""

    def __init__(self, message: str) -> None:
        """Initialize Zeus Exception."""
        super().__init__(message)

__init__ 

__init__(message)
Source code in zeus/device/cpu/rapl.py 
154
155
156
def __init__(self, message: str) -> None:
    """Initialize Zeus Exception."""
    super().__init__(message)

ZeusRAPLPermissionError

Bases: ZeusBaseCPUError

Zeus GPU exception that wraps No Permission to perform GPU operation.

Source code in zeus/device/cpu/rapl.py 
159
160
161
162
163
164
class ZeusRAPLPermissionError(ZeusBaseCPUError):
    """Zeus GPU exception that wraps No Permission to perform GPU operation."""

    def __init__(self, message: str) -> None:
        """Intialize the exception object."""
        super().__init__(message)

__init__ 

__init__(message)
Source code in zeus/device/cpu/rapl.py 
162
163
164
def __init__(self, message: str) -> None:
    """Intialize the exception object."""
    super().__init__(message)

RAPLFile

RAPL File class for each RAPL file.

This class defines the interface for interacting with a RAPL file for a package. A package can be a CPU or DRAM

Source code in zeus/device/cpu/rapl.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
class RAPLFile:
    """RAPL File class for each RAPL file.

    This class defines the interface for interacting with a RAPL file for a package. A package can
    be a CPU or DRAM
    """

    def __init__(self, path: str) -> None:
        """Initialize RAPL file object. Each RAPL file object manages one energy_uj file."""
        self.path = path
        self.energy_uj_path = os.path.join(path, "energy_uj")
        try:
            with open(os.path.join(path, "name"), "r") as name_file:
                self.name: str = name_file.read().strip()
        except FileNotFoundError as err:
            raise ZeusRAPLFileInitError("Error reading package name") from err
        try:
            with open(self.energy_uj_path) as energy_file:
                self.last_energy = float(energy_file.read().strip())
        except FileNotFoundError as err:
            raise ZeusRAPLFileInitError("Error reading package energy") from err
        except PermissionError as err:
            raise cpu_common.ZeusCPUNoPermissionError(
                "Can't read file due to permission error"
            ) from err
        try:
            with open(
                os.path.join(path, "max_energy_range_uj"), "r"
            ) as max_energy_file:
                self.max_energy_range_uj = float(max_energy_file.read().strip())
        except FileNotFoundError as err:
            raise ZeusRAPLFileInitError(
                "Error reading package max energy range"
            ) from err

        self.wraparound_tracker = RaplWraparoundTracker(
            self.energy_uj_path, self.max_energy_range_uj
        )

    def __str__(self) -> str:
        """Return a string representation of the RAPL file object."""
        return f"RAPLFile(Path: {self.path}\nEnergy_uj_path: {self.energy_uj_path}\nName: {self.name}\
        \nLast_energy: {self.last_energy}\nMax_energy: {self.max_energy_range_uj})"

    def read(self) -> float:
        """Read the current energy value from the energy_uj file.

        Returns:
            The current energy value in millijoules.
        """
        with open(self.energy_uj_path) as energy_file:
            new_energy_uj = float(energy_file.read().strip())
        num_wraparounds = self.wraparound_tracker.get_num_wraparounds()
        return (new_energy_uj + num_wraparounds * self.max_energy_range_uj) / 1000.0

__init__ 

__init__(path)
Source code in zeus/device/cpu/rapl.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def __init__(self, path: str) -> None:
    """Initialize RAPL file object. Each RAPL file object manages one energy_uj file."""
    self.path = path
    self.energy_uj_path = os.path.join(path, "energy_uj")
    try:
        with open(os.path.join(path, "name"), "r") as name_file:
            self.name: str = name_file.read().strip()
    except FileNotFoundError as err:
        raise ZeusRAPLFileInitError("Error reading package name") from err
    try:
        with open(self.energy_uj_path) as energy_file:
            self.last_energy = float(energy_file.read().strip())
    except FileNotFoundError as err:
        raise ZeusRAPLFileInitError("Error reading package energy") from err
    except PermissionError as err:
        raise cpu_common.ZeusCPUNoPermissionError(
            "Can't read file due to permission error"
        ) from err
    try:
        with open(
            os.path.join(path, "max_energy_range_uj"), "r"
        ) as max_energy_file:
            self.max_energy_range_uj = float(max_energy_file.read().strip())
    except FileNotFoundError as err:
        raise ZeusRAPLFileInitError(
            "Error reading package max energy range"
        ) from err

    self.wraparound_tracker = RaplWraparoundTracker(
        self.energy_uj_path, self.max_energy_range_uj
    )

__str__ 

__str__()

Return a string representation of the RAPL file object.

Source code in zeus/device/cpu/rapl.py 
206
207
208
209
def __str__(self) -> str:
    """Return a string representation of the RAPL file object."""
    return f"RAPLFile(Path: {self.path}\nEnergy_uj_path: {self.energy_uj_path}\nName: {self.name}\
    \nLast_energy: {self.last_energy}\nMax_energy: {self.max_energy_range_uj})"

read 

read()

Read the current energy value from the energy_uj file.

Returns:

Type Description
float

The current energy value in millijoules.
Source code in zeus/device/cpu/rapl.py 
211
212
213
214
215
216
217
218
219
220
def read(self) -> float:
    """Read the current energy value from the energy_uj file.

    Returns:
        The current energy value in millijoules.
    """
    with open(self.energy_uj_path) as energy_file:
        new_energy_uj = float(energy_file.read().strip())
    num_wraparounds = self.wraparound_tracker.get_num_wraparounds()
    return (new_energy_uj + num_wraparounds * self.max_energy_range_uj) / 1000.0

RAPLCPU

Bases: CPU

Control a single CPU that supports RAPL.

Source code in zeus/device/cpu/rapl.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
class RAPLCPU(cpu_common.CPU):
    """Control a single CPU that supports RAPL."""

    def __init__(self, cpu_index: int, rapl_dir: str) -> None:
        """Initialize the Intel CPU with a specified index."""
        super().__init__(cpu_index)
        self.rapl_dir = rapl_dir
        self._get_powerzone()

    _exception_map = {
        FileNotFoundError: cpu_common.ZeusCPUNotFoundError,
        PermissionError: cpu_common.ZeusCPUNoPermissionError,
        OSError: cpu_common.ZeusCPUInitError,
    }

    def _get_powerzone(self) -> None:
        self.path = os.path.join(self.rapl_dir, f"intel-rapl:{self.cpu_index}")
        self.rapl_file: RAPLFile = RAPLFile(self.path)
        self.dram: RAPLFile | None = None
        for dir in os.listdir(self.path):
            if "intel-rapl" in dir:
                try:
                    rapl_file = RAPLFile(os.path.join(self.path, dir))
                except ZeusRAPLFileInitError as err:
                    warnings.warn(
                        f"Failed to initialize subpackage {err}", stacklevel=1
                    )
                    continue
                if rapl_file.name == "dram":
                    self.dram = rapl_file

    def getTotalEnergyConsumption(self) -> CpuDramMeasurement:
        """Returns the total energy consumption of the specified powerzone. Units: mJ."""
        cpu_mj = self.rapl_file.read()
        dram_mj = None
        if self.dram is not None:
            dram_mj = self.dram.read()
        return CpuDramMeasurement(cpu_mj=cpu_mj, dram_mj=dram_mj)

    def supportsGetDramEnergyConsumption(self) -> bool:
        """Returns True if the specified CPU powerzone supports retrieving the subpackage energy consumption."""
        return self.dram is not None

__init__ 

__init__(cpu_index, rapl_dir)
Source code in zeus/device/cpu/rapl.py 
226
227
228
229
230
def __init__(self, cpu_index: int, rapl_dir: str) -> None:
    """Initialize the Intel CPU with a specified index."""
    super().__init__(cpu_index)
    self.rapl_dir = rapl_dir
    self._get_powerzone()

getTotalEnergyConsumption 

getTotalEnergyConsumption()

Returns the total energy consumption of the specified powerzone. Units: mJ.

Source code in zeus/device/cpu/rapl.py 
254
255
256
257
258
259
260
def getTotalEnergyConsumption(self) -> CpuDramMeasurement:
    """Returns the total energy consumption of the specified powerzone. Units: mJ."""
    cpu_mj = self.rapl_file.read()
    dram_mj = None
    if self.dram is not None:
        dram_mj = self.dram.read()
    return CpuDramMeasurement(cpu_mj=cpu_mj, dram_mj=dram_mj)

supportsGetDramEnergyConsumption 

supportsGetDramEnergyConsumption()

Returns True if the specified CPU powerzone supports retrieving the subpackage energy consumption.

Source code in zeus/device/cpu/rapl.py 
262
263
264
def supportsGetDramEnergyConsumption(self) -> bool:
    """Returns True if the specified CPU powerzone supports retrieving the subpackage energy consumption."""
    return self.dram is not None

ZeusdRAPLCPU

Bases: RAPLCPU

A RAPLCPU that interfaces with RAPL via zeusd.

The parent RAPLCPU class requires root privileges to interface with RAPL. ZeusdRAPLCPU (this class) overrides RAPLCPU's methods so that they instead send requests to the Zeus daemon, which will interface with RAPL on behalf of ZeusdRAPLCPU. As a result, ZeusdRAPLCPU does not need root privileges to monitor CPU and DRAM energy consumption.

See here for details on system privileges required.

Source code in zeus/device/cpu/rapl.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
class ZeusdRAPLCPU(RAPLCPU):
    """A RAPLCPU that interfaces with RAPL via zeusd.

    The parent RAPLCPU class requires root privileges to interface with RAPL.
    ZeusdRAPLCPU (this class) overrides RAPLCPU's methods so that they instead send requests
    to the Zeus daemon, which will interface with RAPL on behalf of ZeusdRAPLCPU. As a result,
    ZeusdRAPLCPU does not need root privileges to monitor CPU and DRAM energy consumption.

    See [here](https://ml.energy/zeus/getting_started/#system-privileges)
    for details on system privileges required.
    """

    def __init__(
        self,
        cpu_index: int,
        zeusd_sock_path: str = "/var/run/zeusd.sock",
    ) -> None:
        """Initialize the Intel CPU with a specified index."""
        self.cpu_index = cpu_index

        self._client = httpx.Client(transport=httpx.HTTPTransport(uds=zeusd_sock_path))
        self._url_prefix = f"http://zeusd/cpu/{cpu_index}"

        self.dram_available = self._supportsGetDramEnergyConsumption()

    def _supportsGetDramEnergyConsumption(self) -> bool:
        """Calls zeusd to return if the specified CPU supports DRAM energy monitoring."""
        resp = self._client.get(
            self._url_prefix + "/supports_dram_energy",
        )
        if resp.status_code != 200:
            raise ZeusdError(
                f"Failed to get whether DRAM energy is supported: {resp.text}"
            )
        data = resp.json()
        dram_available = data.get("dram_available")
        if dram_available is None:
            raise ZeusdError("Failed to get whether DRAM energy is supported.")
        return dram_available

    def getTotalEnergyConsumption(self) -> CpuDramMeasurement:
        """Returns the total energy consumption of the specified powerzone. Units: mJ."""
        resp = self._client.post(
            self._url_prefix + "/get_index_energy",
            json={
                "cpu": True,
                "dram": True,
            },
        )
        if resp.status_code != 200:
            raise ZeusdError(f"Failed to get total energy consumption: {resp.text}")

        data = resp.json()
        cpu_mj = data["cpu_energy_uj"] / 1000

        dram_mj = None
        dram_uj = data.get("dram_energy_uj")
        if dram_uj is None:
            if self.dram_available:
                raise ZeusdError(
                    "DRAM energy should be available but no measurement was found"
                )
        else:
            dram_mj = dram_uj / 1000

        return CpuDramMeasurement(cpu_mj=cpu_mj, dram_mj=dram_mj)

    def supportsGetDramEnergyConsumption(self) -> bool:
        """Returns True if the specified CPU powerzone supports retrieving the subpackage energy consumption."""
        return self.dram_available

__init__ 

__init__(cpu_index, zeusd_sock_path='/var/run/zeusd.sock')
Source code in zeus/device/cpu/rapl.py 
279
280
281
282
283
284
285
286
287
288
289
290
def __init__(
    self,
    cpu_index: int,
    zeusd_sock_path: str = "/var/run/zeusd.sock",
) -> None:
    """Initialize the Intel CPU with a specified index."""
    self.cpu_index = cpu_index

    self._client = httpx.Client(transport=httpx.HTTPTransport(uds=zeusd_sock_path))
    self._url_prefix = f"http://zeusd/cpu/{cpu_index}"

    self.dram_available = self._supportsGetDramEnergyConsumption()

_supportsGetDramEnergyConsumption 

_supportsGetDramEnergyConsumption()

Calls zeusd to return if the specified CPU supports DRAM energy monitoring.

Source code in zeus/device/cpu/rapl.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def _supportsGetDramEnergyConsumption(self) -> bool:
    """Calls zeusd to return if the specified CPU supports DRAM energy monitoring."""
    resp = self._client.get(
        self._url_prefix + "/supports_dram_energy",
    )
    if resp.status_code != 200:
        raise ZeusdError(
            f"Failed to get whether DRAM energy is supported: {resp.text}"
        )
    data = resp.json()
    dram_available = data.get("dram_available")
    if dram_available is None:
        raise ZeusdError("Failed to get whether DRAM energy is supported.")
    return dram_available

getTotalEnergyConsumption 

getTotalEnergyConsumption()

Returns the total energy consumption of the specified powerzone. Units: mJ.

Source code in zeus/device/cpu/rapl.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
def getTotalEnergyConsumption(self) -> CpuDramMeasurement:
    """Returns the total energy consumption of the specified powerzone. Units: mJ."""
    resp = self._client.post(
        self._url_prefix + "/get_index_energy",
        json={
            "cpu": True,
            "dram": True,
        },
    )
    if resp.status_code != 200:
        raise ZeusdError(f"Failed to get total energy consumption: {resp.text}")

    data = resp.json()
    cpu_mj = data["cpu_energy_uj"] / 1000

    dram_mj = None
    dram_uj = data.get("dram_energy_uj")
    if dram_uj is None:
        if self.dram_available:
            raise ZeusdError(
                "DRAM energy should be available but no measurement was found"
            )
    else:
        dram_mj = dram_uj / 1000

    return CpuDramMeasurement(cpu_mj=cpu_mj, dram_mj=dram_mj)

supportsGetDramEnergyConsumption 

supportsGetDramEnergyConsumption()

Returns True if the specified CPU powerzone supports retrieving the subpackage energy consumption.

Source code in zeus/device/cpu/rapl.py 
334
335
336
def supportsGetDramEnergyConsumption(self) -> bool:
    """Returns True if the specified CPU powerzone supports retrieving the subpackage energy consumption."""
    return self.dram_available

RAPLCPUs

Bases: CPUs

RAPL CPU Manager object, containing individual RAPLCPU objects, abstracting RAPL calls and handling related exceptions.

Source code in zeus/device/cpu/rapl.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
class RAPLCPUs(cpu_common.CPUs):
    """RAPL CPU Manager object, containing individual RAPLCPU objects, abstracting RAPL calls and handling related exceptions."""

    def __init__(self) -> None:
        """Instantiates IntelCPUs object, setting up tracking for specified Intel CPUs."""
        if not rapl_is_available():
            raise ZeusRAPLNotSupportedError("RAPL is not supported on this CPU.")

        self.rapl_dir = RAPL_DIR if os.path.exists(RAPL_DIR) else CONTAINER_RAPL_DIR
        self._init_cpus()

    @property
    def cpus(self) -> Sequence[RAPLCPU]:
        """Returns a list of CPU objects being tracked."""
        return self._cpus

    def _init_cpus(self) -> None:
        """Initialize all Intel CPUs."""
        self._cpus = []

        cpu_indices = []

        def sort_key(dir):
            return int(dir.split(":")[1])

        for dir in sorted(glob(f"{self.rapl_dir}/intel-rapl:*"), key=sort_key):
            parts = dir.split(":")
            if len(parts) > 1 and parts[1].isdigit():
                cpu_indices.append(int(parts[1]))

        # If `ZEUSD_SOCK_PATH` is set, always use ZeusdRAPLCPU
        if (sock_path := os.environ.get("ZEUSD_SOCK_PATH")) is not None:
            if not Path(sock_path).exists():
                raise ZeusdError(
                    f"ZEUSD_SOCK_PATH points to non-existent file: {sock_path}"
                )
            if not Path(sock_path).is_socket():
                raise ZeusdError(f"ZEUSD_SOCK_PATH is not a socket: {sock_path}")
            if not os.access(sock_path, os.W_OK):
                raise ZeusdError(f"ZEUSD_SOCK_PATH is not writable: {sock_path}")
            self._cpus = [
                ZeusdRAPLCPU(cpu_index, sock_path) for cpu_index in cpu_indices
            ]
        else:
            self._cpus = [
                RAPLCPU(cpu_index, self.rapl_dir) for cpu_index in cpu_indices
            ]

    def __del__(self) -> None:
        """Shuts down the Intel CPU monitoring."""
        pass

cpus property 

cpus

Returns a list of CPU objects being tracked.

__init__ 

__init__()
Source code in zeus/device/cpu/rapl.py 
342
343
344
345
346
347
348
def __init__(self) -> None:
    """Instantiates IntelCPUs object, setting up tracking for specified Intel CPUs."""
    if not rapl_is_available():
        raise ZeusRAPLNotSupportedError("RAPL is not supported on this CPU.")

    self.rapl_dir = RAPL_DIR if os.path.exists(RAPL_DIR) else CONTAINER_RAPL_DIR
    self._init_cpus()

_init_cpus 

_init_cpus()

Initialize all Intel CPUs.

Source code in zeus/device/cpu/rapl.py 
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
def _init_cpus(self) -> None:
    """Initialize all Intel CPUs."""
    self._cpus = []

    cpu_indices = []

    def sort_key(dir):
        return int(dir.split(":")[1])

    for dir in sorted(glob(f"{self.rapl_dir}/intel-rapl:*"), key=sort_key):
        parts = dir.split(":")
        if len(parts) > 1 and parts[1].isdigit():
            cpu_indices.append(int(parts[1]))

    # If `ZEUSD_SOCK_PATH` is set, always use ZeusdRAPLCPU
    if (sock_path := os.environ.get("ZEUSD_SOCK_PATH")) is not None:
        if not Path(sock_path).exists():
            raise ZeusdError(
                f"ZEUSD_SOCK_PATH points to non-existent file: {sock_path}"
            )
        if not Path(sock_path).is_socket():
            raise ZeusdError(f"ZEUSD_SOCK_PATH is not a socket: {sock_path}")
        if not os.access(sock_path, os.W_OK):
            raise ZeusdError(f"ZEUSD_SOCK_PATH is not writable: {sock_path}")
        self._cpus = [
            ZeusdRAPLCPU(cpu_index, sock_path) for cpu_index in cpu_indices
        ]
    else:
        self._cpus = [
            RAPLCPU(cpu_index, self.rapl_dir) for cpu_index in cpu_indices
        ]

__del__ 

__del__()

Shuts down the Intel CPU monitoring.

Source code in zeus/device/cpu/rapl.py 
387
388
389
def __del__(self) -> None:
    """Shuts down the Intel CPU monitoring."""
    pass

_polling_process 

_polling_process(
    rapl_file_path, max_energy_uj, wraparound_counter
)

Check for wraparounds in the specified rapl file.

Source code in zeus/device/cpu/rapl.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def _polling_process(
    rapl_file_path: str, max_energy_uj: float, wraparound_counter: Synchronized[int]
) -> None:
    """Check for wraparounds in the specified rapl file."""
    try:
        with open(rapl_file_path) as rapl_file:
            last_energy_uj = float(rapl_file.read().strip())
        while True:
            sleep_time = 1.0
            with open(rapl_file_path, "r") as rapl_file:
                energy_uj = float(rapl_file.read().strip())
            if max_energy_uj - energy_uj < RAPL_COUNTER_MAX_INCREASE:
                sleep_time = 0.1
            if energy_uj < last_energy_uj:
                with wraparound_counter.get_lock():
                    wraparound_counter.value += 1
            last_energy_uj = energy_uj
            time.sleep(sleep_time)
    except KeyboardInterrupt:
        return

rapl_is_available cached 

rapl_is_available()

Check if RAPL is available.

Source code in zeus/device/cpu/rapl.py 
133
134
135
136
137
138
139
140
@lru_cache(maxsize=1)
def rapl_is_available() -> bool:
    """Check if RAPL is available."""
    if not os.path.exists(RAPL_DIR) and not os.path.exists(CONTAINER_RAPL_DIR):
        logger.info("RAPL is not supported on this CPU.")
        return False
    logger.info("RAPL is available.")
    return True