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 
 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
109
110
111
112
113
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.
        logger.info("RaplWraparoundTracker is 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), daemon=True
        )
        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 
 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
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.
    logger.info("RaplWraparoundTracker is 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), daemon=True
    )
    self.process.start()

_stop 

_stop()

Stop monitoring power usage.

Source code in zeus/device/cpu/rapl.py 
102
103
104
105
106
107
108
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 
110
111
112
113
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 
157
158
159
160
161
162
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 
160
161
162
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 
165
166
167
168
169
170
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 
168
169
170
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 
173
174
175
176
177
178
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 
176
177
178
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 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
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
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; RAPL requires root privileges. "
                "Refer to https://ml.energy/zeus/getting_started/#system-privileges for more details."
            ) 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 | None = None
        self._wraparound_tracker_init_lock = threading.Lock()

    @property
    def wraparound_tracker(self) -> RaplWraparoundTracker:
        """Return the RaplWraparoundTracker, creating it thread-safely on first access."""
        if self._wraparound_tracker is None:
            with self._wraparound_tracker_init_lock:
                if self._wraparound_tracker is None:
                    self._wraparound_tracker = RaplWraparoundTracker(self.energy_uj_path, self.max_energy_range_uj)
        return self._wraparound_tracker

    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

wraparound_tracker property 

wraparound_tracker

Return the RaplWraparoundTracker, creating it thread-safely on first access.

__init__ 

__init__(path)
Source code in zeus/device/cpu/rapl.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
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; RAPL requires root privileges. "
            "Refer to https://ml.energy/zeus/getting_started/#system-privileges for more details."
        ) 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 | None = None
    self._wraparound_tracker_init_lock = threading.Lock()

__str__ 

__str__()

Return a string representation of the RAPL file object.

Source code in zeus/device/cpu/rapl.py 
296
297
298
299
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 
301
302
303
304
305
306
307
308
309
310
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 
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
390
391
392
393
394
395
396
397
398
399
400
401
402
403
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 get_total_energy_consumption(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 supports_get_dram_energy_consumption(self) -> bool:
        """Returns True if the specified CPU powerzone supports retrieving the subpackage energy consumption."""
        return self.dram is not None

    def get_power_limits(self) -> CpuDramPowerLimits:
        """Returns the power limits of the CPU package and DRAM zones."""
        return CpuDramPowerLimits(
            cpu=_read_zone_power_limits(self.path),
            dram=None if self.dram is None else _read_zone_power_limits(self.dram.path),
        )

__init__ 

__init__(cpu_index, rapl_dir)
Source code in zeus/device/cpu/rapl.py 
360
361
362
363
364
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()

get_total_energy_consumption 

get_total_energy_consumption()

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

Source code in zeus/device/cpu/rapl.py 
386
387
388
389
390
391
392
def get_total_energy_consumption(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)

supports_get_dram_energy_consumption 

supports_get_dram_energy_consumption()

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

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

get_power_limits 

get_power_limits()

Returns the power limits of the CPU package and DRAM zones.

Source code in zeus/device/cpu/rapl.py 
398
399
400
401
402
403
def get_power_limits(self) -> CpuDramPowerLimits:
    """Returns the power limits of the CPU package and DRAM zones."""
    return CpuDramPowerLimits(
        cpu=_read_zone_power_limits(self.path),
        dram=None if self.dram is None else _read_zone_power_limits(self.dram.path),
    )

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 
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
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, client: ZeusdClient) -> None:
        """Initialize the Intel CPU with a specified index.

        Args:
            cpu_index: Index of the CPU.
            client: ZeusdClient connected to the daemon.
        """
        self.cpu_index = cpu_index
        self._client = client
        require_capabilities(client, read_cpu=True, cpu_ids=[cpu_index])
        idx = client.cpu_ids.index(cpu_index)
        self.dram_available = client.dram_available[idx]

    def get_total_energy_consumption(self) -> CpuDramMeasurement:
        """Returns the total energy consumption of the specified powerzone. Units: mJ."""
        result = self._client.get_cpu_energy([self.cpu_index], cpu=True, dram=True)
        entry = result[self.cpu_index]
        if entry.cpu_energy_uj is None:
            raise ZeusdError(f"CPU {self.cpu_index}: cpu_energy_uj is null in response")
        cpu_mj = entry.cpu_energy_uj / 1000

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

        return CpuDramMeasurement(cpu_mj=cpu_mj, dram_mj=dram_mj)

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

    def get_power_limits(self) -> CpuDramPowerLimits:
        """Returns the power limits of the CPU package and DRAM zones."""
        return self._client.get_cpu_power_limit([self.cpu_index])[self.cpu_index]

__init__ 

__init__(cpu_index, client)

Parameters:

Name Type Description Default
cpu_index int

Index of the CPU.

 required
client ZeusdClient

ZeusdClient connected to the daemon.

 required
Source code in zeus/device/cpu/rapl.py 
418
419
420
421
422
423
424
425
426
427
428
429
def __init__(self, cpu_index: int, client: ZeusdClient) -> None:
    """Initialize the Intel CPU with a specified index.

    Args:
        cpu_index: Index of the CPU.
        client: ZeusdClient connected to the daemon.
    """
    self.cpu_index = cpu_index
    self._client = client
    require_capabilities(client, read_cpu=True, cpu_ids=[cpu_index])
    idx = client.cpu_ids.index(cpu_index)
    self.dram_available = client.dram_available[idx]

get_total_energy_consumption 

get_total_energy_consumption()

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

Source code in zeus/device/cpu/rapl.py 
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
def get_total_energy_consumption(self) -> CpuDramMeasurement:
    """Returns the total energy consumption of the specified powerzone. Units: mJ."""
    result = self._client.get_cpu_energy([self.cpu_index], cpu=True, dram=True)
    entry = result[self.cpu_index]
    if entry.cpu_energy_uj is None:
        raise ZeusdError(f"CPU {self.cpu_index}: cpu_energy_uj is null in response")
    cpu_mj = entry.cpu_energy_uj / 1000

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

    return CpuDramMeasurement(cpu_mj=cpu_mj, dram_mj=dram_mj)

supports_get_dram_energy_consumption 

supports_get_dram_energy_consumption()

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

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

get_power_limits 

get_power_limits()

Returns the power limits of the CPU package and DRAM zones.

Source code in zeus/device/cpu/rapl.py 
452
453
454
def get_power_limits(self) -> CpuDramPowerLimits:
    """Returns the power limits of the CPU package and DRAM zones."""
    return self._client.get_cpu_power_limit([self.cpu_index])[self.cpu_index]

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 
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
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 env vars are set, use ZeusdRAPLCPU backed by a shared client.
        config = ZeusdConfig.from_env()
        if config is not None:
            try:
                client = ZeusdClient(config)
                self._cpus = [ZeusdRAPLCPU(cpu_index, client) for cpu_index in cpu_indices]
            except ZeusBaseError as e:
                raise cpu_common.ZeusCPUInitError(str(e)) from e
        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 
460
461
462
463
464
465
466
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 
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
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 env vars are set, use ZeusdRAPLCPU backed by a shared client.
    config = ZeusdConfig.from_env()
    if config is not None:
        try:
            client = ZeusdClient(config)
            self._cpus = [ZeusdRAPLCPU(cpu_index, client) for cpu_index in cpu_indices]
        except ZeusBaseError as e:
            raise cpu_common.ZeusCPUInitError(str(e)) from e
    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 
498
499
500
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 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
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
                    logger.debug(
                        "RAPL wraparound detected for %s; counter now: %d",
                        rapl_file_path,
                        wraparound_counter.value,
                    )
            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 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
@lru_cache(maxsize=1)
def rapl_is_available() -> bool:
    """Check if RAPL is available."""
    if os.path.exists(RAPL_DIR):
        logger.info("RAPL directory (%s) is available.", RAPL_DIR)
        return True
    if os.path.exists(CONTAINER_RAPL_DIR):
        logger.info(
            "You are likely in a container, and RAPL directory (%s) is available.",
            CONTAINER_RAPL_DIR,
        )
        return True
    logger.info("RAPL is not supported on this CPU.")
    return False

get_current_rapl_zone_id 

get_current_rapl_zone_id(pid='current')

Return the RAPL package zone ID for the socket the given process runs on.

RAPL exposes one top-level power zone per CPU package (socket) under /sys/class/powercap/intel-rapl:*. This helper maps a running process to the RAPL zone ID of the package it is currently executing on, so that zone ID can be passed as a cpu_indices entry to only measure that socket.

The physical package ID reported by sysfs topology is an arbitrary platform identifier and is not guaranteed to equal the RAPL zone ID, so the mapping is resolved via RAPL's own zone naming:

  1. Read the logical CPU the process last ran on from /proc/{pid}/stat.
  2. Read that CPU's physical package ID N from sysfs topology.
  3. Read that CPU's die ID D from sysfs topology (if available).
  4. Find the RAPL zone whose name file reads package-N or package-N-die-D and return its zone ID.

Parameters:

Name Type Description Default
pid int | Literal['current']

The process ID to look up, or "current" for the current process.

 'current'

Note

Linux schedulers can preempt and reschedule processes to different CPUs. To prevent this from happening during monitoring, use taskset to pin processes to specific CPUs.

Source code in zeus/device/cpu/rapl.py 
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
221
222
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
def get_current_rapl_zone_id(pid: int | Literal["current"] = "current") -> int:
    """Return the RAPL package zone ID for the socket the given process runs on.

    RAPL exposes one top-level power zone per CPU package (socket) under
    `/sys/class/powercap/intel-rapl:*`. This helper maps a running process to the
    RAPL zone ID of the package it is currently executing on, so that zone ID can
    be passed as a `cpu_indices` entry to only measure that socket.

    The physical package ID reported by sysfs topology is an arbitrary platform
    identifier and is not guaranteed to equal the RAPL zone ID, so the mapping is
    resolved via RAPL's own zone naming:

    1. Read the logical CPU the process last ran on from `/proc/{pid}/stat`.
    2. Read that CPU's physical package ID `N` from sysfs topology.
    3. Read that CPU's die ID `D` from sysfs topology (if available).
    4. Find the RAPL zone whose `name` file reads `package-N` or `package-N-die-D`
       and return its zone ID.

    Args:
        pid: The process ID to look up, or "current" for the current process.

    !!! Note
        Linux schedulers can preempt and reschedule processes to different CPUs. To
        prevent this from happening during monitoring, use `taskset` to pin
        processes to specific CPUs.
    """
    if pid == "current":
        pid = os.getpid()

    # The logical CPU the process last executed on is the 39th field of
    # /proc/{pid}/stat. The second field (comm) can contain spaces and
    # parentheses, so split off everything up to the last ")" first; the
    # processor field is then at index 36 of the remaining fields.
    with open(f"/proc/{pid}/stat") as stat_file:
        cpu_core = int(stat_file.read().rsplit(")", 1)[1].split()[36])

    topology_dir = f"/sys/devices/system/cpu/cpu{cpu_core}/topology"
    with open(f"{topology_dir}/physical_package_id") as phys_package_file:
        package_id = int(phys_package_file.read().strip())

    # RAPL zone names are either "package-N" or "package-N-die-D", so accept both.
    # die_id may not exist on all kernels/platforms.
    acceptable_names = {f"package-{package_id}"}
    try:
        with open(f"{topology_dir}/die_id") as die_id_file:
            die_id = int(die_id_file.read().strip())
        acceptable_names.add(f"package-{package_id}-die-{die_id}")
    except (OSError, ValueError):
        pass

    rapl_dir = RAPL_DIR if os.path.exists(RAPL_DIR) else CONTAINER_RAPL_DIR
    for zone_dir in glob(f"{rapl_dir}/intel-rapl:[0-9]*"):
        # Only consider top-level package zones (intel-rapl:N), not sub-zones like
        # intel-rapl:N:M (e.g., core, dram).
        base = os.path.basename(zone_dir)
        if base.count(":") != 1:
            continue
        try:
            with open(os.path.join(zone_dir, "name")) as name_file:
                name = name_file.read().strip()
        except OSError:
            continue
        if name in acceptable_names:
            return int(base.split(":")[1])

    raise ZeusRAPLNotSupportedError(
        f"Could not find a RAPL package zone matching physical package ID {package_id} "
        f"(process {pid} last ran on CPU {cpu_core}) under {rapl_dir}."
    )

_read_zone_power_limits 

_read_zone_power_limits(zone_dir)

Read the power limit state of a RAPL powercap zone.

The kernel numbers constraints from zero without gaps, so reading stops at the first index without a constraint_K_name file. A constraint attribute the kernel has no value for fails with ENODATA (e.g., the time window of peak_power) and is read as None.

Raises:

Type Description
OSError

If a power limit file cannot be read.
ValueError

If a power limit file has an unexpected value.
Source code in zeus/device/cpu/rapl.py 
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
def _read_zone_power_limits(zone_dir: str) -> ZonePowerLimits:
    """Read the power limit state of a RAPL powercap zone.

    The kernel numbers constraints from zero without gaps, so reading stops at the first index without a `constraint_K_name` file.
    A constraint attribute the kernel has no value for fails with `ENODATA` (e.g., the time window of `peak_power`) and is read as None.

    Raises:
        OSError: If a power limit file cannot be read.
        ValueError: If a power limit file has an unexpected value.
    """

    def read(file: str) -> str:
        with open(os.path.join(zone_dir, file)) as f:
            return f.read().strip()

    def read_optional_int(file: str) -> int | None:
        try:
            return int(read(file))
        except OSError as e:
            if e.errno == errno.ENODATA:
                return None
            raise

    enabled = read("enabled")
    if enabled not in ("0", "1"):
        raise ValueError(f"Unexpected value {enabled!r} in {os.path.join(zone_dir, 'enabled')}")

    constraints = []
    index = 0
    while os.path.exists(os.path.join(zone_dir, f"constraint_{index}_name")):
        max_power_uw = read_optional_int(f"constraint_{index}_max_power_uw")
        constraints.append(
            PowerLimitConstraint(
                name=read(f"constraint_{index}_name"),
                power_limit_mw=int(read(f"constraint_{index}_power_limit_uw")) // 1000,
                max_power_mw=None if max_power_uw is None else max_power_uw // 1000,
                time_window_us=read_optional_int(f"constraint_{index}_time_window_us"),
            )
        )
        index += 1

    return ZonePowerLimits(enabled=enabled == "1", constraints=constraints)