[PATCH 20/36] tuna: Dynamically find free CPU for test_cpu_exclusive
John Kacur <[email protected]> Fri, 10 Jul 2026 10:14:58 -0400
| Newsgroups | org.kernel.vger.linux-rt-users |
|---|---|
| Message-ID | <[email protected]> |
Update test_cpu_exclusive to dynamically find an available CPU based on the actual system configuration instead of hardcoding CPU ranges. The test now: - Scans all existing cpusets to find which CPUs are exclusively allocated - Reads actual system CPUs from /sys/fs/cgroup/cpuset.cpus.effective - Falls back to multiprocessing.cpu_count() if needed - Searches through actual available CPUs on the system - Finds the first non-exclusively-allocated CPU - Gracefully skips the test if all CPUs are exclusively allocated This works on systems with any number of CPUs (2, 4, 8, 64, 128+) and allows tests to run cleanly regardless of existing cpuset configuration. Assisted-by: Claude:claude-sonnet-4-5 Signed-off-by: John Kacur <[email protected]> --- tests/test_cpuset.py | 58 +++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 55 insertions(+), 3 deletions(-) diff --git a/tests/test_cpuset.py b/tests/test_cpuset.py index ca0a6b0c9ccd..92a3f833f371 100644 --- a/tests/test_cpuset.py +++ b/tests/test_cpuset.py @@ -200,11 +200,62 @@ class TestCpusetConfiguration(unittest.TestCase): def test_cpu_exclusive(self): """Test CPU exclusive (partition) setting""" + # Find a CPU that isn't already exclusively allocated to avoid conflicts + # Read all existing cpusets and their exclusive CPUs + used_cpus = set() + all_cpusets = cpuset.list_cpusets(recursive=True) + + for cs_name in all_cpusets: + cs_path = os.path.join('/sys/fs/cgroup', cs_name) + partition_file = os.path.join(cs_path, 'cpuset.cpus.partition') + + # Check if this cpuset has an isolated partition + if os.path.exists(partition_file): + with open(partition_file) as f: + partition = f.read().strip() + if partition.startswith('isolated'): + # This cpuset has exclusively allocated CPUs + cpus_file = os.path.join(cs_path, 'cpuset.cpus') + if os.path.exists(cpus_file): + with open(cpus_file) as f: + cpus_str = f.read().strip() + if cpus_str: + # Parse CPU list and add to used set + from tuna import tuna + try: + cpu_list = tuna.expand_cpulist(cpus_str) + used_cpus.update(cpu_list) + except: + pass + + # Get actual number of CPUs on this system + # Read from root cgroup's effective CPUs + root_cpus_file = '/sys/fs/cgroup/cpuset.cpus.effective' + try: + with open(root_cpus_file) as f: + root_cpus_str = f.read().strip() + from tuna import tuna + available_cpus = tuna.expand_cpulist(root_cpus_str) + except: + # Fallback: use multiprocessing to get CPU count + import multiprocessing + available_cpus = list(range(multiprocessing.cpu_count())) + + # Find first available CPU that isn't exclusively allocated + test_cpu = None + for cpu_num in available_cpus: + if cpu_num not in used_cpus: + test_cpu = str(cpu_num) + break + + if test_cpu is None: + self.skipTest("No available CPUs found for exclusive partition test (all CPUs are exclusively allocated)") + cs = cpuset.Cpuset('test_exclusive') self.test_cpusets.append(cs) cs.write_memnode('0') - cs.assign_cpus('0') + cs.assign_cpus(test_cpu) # Try to set exclusive (may not be supported on all kernels) cs.set_cpu_exclusive() @@ -213,8 +264,9 @@ class TestCpusetConfiguration(unittest.TestCase): if os.path.exists(partition_file): with open(partition_file) as f: partition = f.read().strip() - # Should be 'isolated' or 'root' (if not supported) - self.assertIn(partition, ['isolated', 'root', 'member']) + # Should be 'isolated', 'root', or 'member' + self.assertIn(partition, ['isolated', 'root', 'member'], + f"Unexpected partition state: {partition}") @unittest.skipUnless(os.geteuid() == 0, "Requires root permissions") -- 2.54.0