[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