[PATCH 1/7] rteval: Add CoreSiblings class for CPU core topology queries
John Kacur <[email protected]> Tue, 21 Apr 2026 14:26:12 -0400
| Newsgroups | org.kernel.vger.linux-rt-users |
|---|---|
| Message-ID | <[email protected]> |
Add a new CoreSiblings class in rteval/sysinfo/coresiblings.py that provides a simple interface for querying which CPUs share physical cores. The class reads /sys/devices/system/cpu/cpu*/topology/thread_siblings_list to build a mapping of CPU core siblings and provides three methods: - share_core(cpu1, cpu2): Check if two CPUs share a physical core - get_siblings(cpu): Get all CPUs that share a core with a given CPU - get_core_groups(): Get all unique core sibling groups The implementation is generic and works with any SMT configuration: - No SMT (1 CPU per core) - 2-way SMT (typical Intel hyperthreading) - 4-way SMT (some IBM POWER processors) - 8-way SMT (IBM POWER8/9) This provides a foundation for cpuset management features where we need to understand CPU core topology for proper isolation. Assisted-by: Claude Sonnet 4.5 <[email protected]> Signed-off-by: John Kacur <[email protected]> --- rteval/sysinfo/coresiblings.py | 110 +++++++++++++++++++++++++++++++++ 1 file changed, 110 insertions(+) create mode 100644 rteval/sysinfo/coresiblings.py diff --git a/rteval/sysinfo/coresiblings.py b/rteval/sysinfo/coresiblings.py new file mode 100644 index 000000000000..3c97439d0a39 --- /dev/null +++ b/rteval/sysinfo/coresiblings.py @@ -0,0 +1,110 @@ +# -*- coding: utf-8 -*- +# SPDX-License-Identifier: GPL-2.0-or-later +# +# Copyright 2026 John Kacur <[email protected]> +# + +import os +from rteval.cpulist_utils import expand_cpulist + +class CoreSiblings: + """Query CPU core topology to determine which CPUs share physical cores""" + + def __init__(self, root="/"): + self.sysdir = os.path.join(root, 'sys', 'devices', 'system', 'cpu') + self.core_map = {} # Maps cpu -> set of sibling cpus + self._parse() + + def _parse(self): + """Parse thread_siblings_list for each CPU""" + for dirname in os.listdir(self.sysdir): + # Only parse cpu<integer> directories + if dirname.startswith('cpu') and os.path.isdir(os.path.join(self.sysdir, dirname)): + try: + cpu_id = int(dirname[3:]) + except ValueError: + continue + + siblings_file = os.path.join(self.sysdir, dirname, 'topology', 'thread_siblings_list') + if os.path.exists(siblings_file): + with open(siblings_file, 'r') as f: + siblings_str = f.read().strip() + # expand_cpulist returns a list of cpu numbers + siblings = set(expand_cpulist(siblings_str)) + self.core_map[cpu_id] = siblings + + def share_core(self, cpu1, cpu2): + """ + Check if two CPUs share the same physical core. + + Args: + cpu1: First CPU ID (int) + cpu2: Second CPU ID (int) + + Returns: + True if CPUs share a core, False otherwise + """ + if cpu1 not in self.core_map: + return False + return cpu2 in self.core_map[cpu1] + + def get_siblings(self, cpu): + """ + Get all CPUs that share a core with the given CPU. + + Args: + cpu: CPU ID (int) + + Returns: + Set of CPU IDs that share a core with cpu (includes cpu itself) + """ + return self.core_map.get(cpu, set()) + + def get_core_groups(self): + """ + Get all unique core sibling groups. + + Returns: + List of sets, where each set contains CPUs that share a core + """ + seen = set() + groups = [] + + for cpu, siblings in self.core_map.items(): + # Use frozenset as a hashable representation + group_key = frozenset(siblings) + if group_key not in seen: + seen.add(group_key) + groups.append(siblings) + + return groups + + +def unit_test(): + """Simple unit test""" + try: + cs = CoreSiblings() + + print("Core Sibling Groups:") + for i, group in enumerate(cs.get_core_groups()): + print(f" Core {i}: {sorted(group)}") + + # Test share_core with first two CPUs if they exist + if len(cs.core_map) >= 2: + cpus = sorted(cs.core_map.keys()) + cpu1, cpu2 = cpus[0], cpus[1] + print(f"\nDo CPU {cpu1} and CPU {cpu2} share a core? {cs.share_core(cpu1, cpu2)}") + print(f"CPU {cpu1} siblings: {sorted(cs.get_siblings(cpu1))}") + print(f"CPU {cpu2} siblings: {sorted(cs.get_siblings(cpu2))}") + + return 0 + except Exception as e: + print(f"** EXCEPTION: {e}") + import traceback + traceback.print_exc() + return 1 + + +if __name__ == '__main__': + import sys + sys.exit(unit_test()) -- 2.53.0