Heterogeneous CPU counts in Python

Readme.md

Heterogeneous CPU core count

This Python script uses only built-in stdlib functions and attempts to detect the CPU count of each type of CPU core. Many CPUs have only one core type, while others (especially for laptops) may have 2 or more CPU core types, sometimes called "big-little" or "super-performance-efficiency".

Usage

python cpu_hetero.py

Using HWLOC is simpler and one would expect more reliable. At the time of writing pyHWLOC supports Linux and Windows but not macOS yet.

uv run cpu_hwloc.py

where this is uv Python manager.

Examples of Heterogenous CPUs

• Snapdragon X2 Elite
• Intel Core Ultra
• Apple M1, M2, M3, M4, M5, M6
• AMD Ryzen AI Pro 300

cpu_hetero.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.9"
# dependencies = []
# ///

"""
Print heterogeneous CPU core counts.
For macOS, Linux, and Windows.

python3 cpu_hetero.py
"""

from __future__ import annotations

import platform
from collections import defaultdict
from pathlib import Path
from pprint import pprint


def macos() -> dict[str, int]:
    import subprocess

    n = int(subprocess.check_output(["sysctl", "-n", "hw.nperflevels"], text=True))
    counts: dict[str, int] = {}
    for i in range(n):
        count = int(subprocess.check_output(
            ["sysctl", "-n", f"hw.perflevel{i}.physicalcpu"], text=True
        ))
        name = subprocess.check_output(
            ["sysctl", "-n", f"hw.perflevel{i}.name"], text=True
        ).strip()
        counts[name or str(i)] = count

    return counts


def _expand(spec: str) -> list[int]:
    out: list[int] = []
    for part in spec.strip().split(","):
        if not part:
            continue
        if "-" in part:
            a, b = part.split("-", 1)
            out.extend(range(int(a), int(b) + 1))
        else:
            out.append(int(part))
    return out


def _physical(cpus: list[int]) -> int:
    seen = set()
    for cpu in cpus:
        base = Path(f"/sys/devices/system/cpu/cpu{cpu}/topology")
        sib = base / "core_cpus_list"
        if not sib.exists():
            sib = base / "thread_siblings_list"
        key = sib.read_text().strip() if sib.exists() else str(cpu)
        seen.add(key)
    return len(seen)


def _cpu_count(spec: str) -> int:
    cpus = _expand(spec)
    return _physical(cpus)


def linux() -> dict[str, int]:
    counts: dict[str, int] = {}
    names = {
        "cpu_core": "Performance",
        "cpu_atom": "Efficiency",
    }
    for path in sorted(Path("/sys/devices/system/cpu").glob("cpu_*/cpus")):
        kind = path.parent.name
        counts[names.get(kind, kind)] = _cpu_count(path.read_text())

    if counts:
        return counts

    # ARM / capacity-aware kernels
    by_cap: dict[int, list[int]] = defaultdict(list)
    for d in sorted(Path("/sys/devices/system/cpu").glob("cpu[0-9]*")):
        cap = d / "cpu_capacity"
        if not cap.exists():
            continue
        by_cap[int(cap.read_text())].append(int(d.name[3:]))

    if by_cap:
        return {str(i): _physical(by_cap[cap]) for i, cap in enumerate(sorted(by_cap, reverse=True))}

    online = Path("/sys/devices/system/cpu/online")
    if online.exists():
        return {"0": _physical(_expand(online.read_text()))}
    return {"0": 0}


def windows() -> dict[str, int]:
    import ctypes
    from ctypes import wintypes

    RelationProcessorCore = 0
    ERROR_INSUFFICIENT_BUFFER = 122
    LTP_PC_SMT = 0x1

    class GROUP_AFFINITY(ctypes.Structure):
        _fields_ = [
            ("Mask", ctypes.c_uint64),
            ("Group", wintypes.WORD),
            ("Reserved", wintypes.WORD * 3),
        ]

    class PROCESSOR_RELATIONSHIP(ctypes.Structure):
        _fields_ = [
            ("Flags", ctypes.c_ubyte),
            ("EfficiencyClass", ctypes.c_ubyte),
            ("Reserved", ctypes.c_ubyte * 20),
            ("GroupCount", wintypes.WORD),
            ("GroupMask", GROUP_AFFINITY * 1),
        ]

    class SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX(ctypes.Structure):
        _fields_ = [
            ("Relationship", wintypes.DWORD),
            ("Size", wintypes.DWORD),
            ("Processor", PROCESSOR_RELATIONSHIP),
        ]

    windll = getattr(ctypes, "WinDLL")
    get_last_error = getattr(ctypes, "get_last_error")
    win_error = getattr(ctypes, "WinError")

    k32 = windll("kernel32", use_last_error=True)
    fn = k32.GetLogicalProcessorInformationEx
    fn.argtypes = [wintypes.DWORD, ctypes.c_void_p, ctypes.POINTER(wintypes.DWORD)]
    fn.restype = wintypes.BOOL

    size = wintypes.DWORD(0)
    if fn(RelationProcessorCore, None, ctypes.byref(size)):
        raise OSError("unexpected success with null buffer")
    if get_last_error() != ERROR_INSUFFICIENT_BUFFER:
        raise win_error()

    buf = ctypes.create_string_buffer(size.value)
    if not fn(RelationProcessorCore, buf, ctypes.byref(size)):
        raise win_error()

    by_class = defaultdict(lambda: {"physical": 0, "logical": 0, "smt": 0})
    offset = 0
    while offset < size.value:
        info = SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX.from_buffer(buf, offset)
        proc = info.Processor
        group_mask_offset = (
            offset
            + SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX.Processor.offset
            + PROCESSOR_RELATIONSHIP.GroupMask.offset
        )
        bits = sum(
            bin(GROUP_AFFINITY.from_buffer(
                buf, group_mask_offset + group * ctypes.sizeof(GROUP_AFFINITY)
            ).Mask).count("1")
            for group in range(proc.GroupCount)
        )
        rec = by_class[proc.EfficiencyClass]
        rec["physical"] += 1
        rec["logical"] += bits
        if proc.Flags & LTP_PC_SMT:
            rec["smt"] += 1
        offset += info.Size

    classes = sorted(by_class)
    return {str(cls): by_class[cls]["physical"] for cls in classes}


if __name__ == "__main__":
    osname = platform.system()
    if osname == "Darwin":
        pprint(macos())
    elif osname == "Linux":
        pprint(linux())
    elif osname == "Windows":
        pprint(windows())
    else:
        raise SystemExit(f"Unsupported OS: {osname}")

cpu_hwloc.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "pyhwloc",
# ]
# ///

"""
Print heterogeneous CPU kinds using pyhwloc.

https://github.com/open-mpi/pyhwloc/
"""

from pyhwloc import Topology
from pyhwloc.hwobject import ObjType


def label(i: int, n: int) -> str:
    if n == 1:
        return "Homogeneous"
    if i == 0:
        return "Efficiency"
    if i == n - 1:
        return "Performance"
    return f"Mid kind {i}"


with Topology() as topo:
    kinds = topo.get_cpukinds()
    n = kinds.n_kinds()
    print(f"CPU kinds: {n}")
    print(f"Total cores: {topo.n_cores()}")

    for i in range(n):
        cpuset, efficiency, infos = kinds.get_info(i)
        n_pu = cpuset.weight()
        n_core = sum(
            1
            for core in topo.iter_objs_by_type(ObjType.CORE)
            if core.cpuset is not None and core.cpuset.isincluded(cpuset)
        )
        extra = " ".join(f"{k}={v}" for k, v in infos.items())
        print(
            f"{label(i, n)} [kind {i}, efficiency={efficiency}]: "
            f"{n_core} physical / {n_pu} logical  {cpuset}"
            + (f"  {extra}" if extra else "")
        )