[PATCH 1/6] rteval: Add cpuset module for cgroup v2 management

John Kacur
2026-04-17T19:51:34Z

rteval: Add cpuset module for cgroup v2 management

Add a new cpuset.py module that provides comprehensive cpuset
management using Linux cgroup v2 interface. This module includes:

- Cpuset class for creating and manipulating CPU sets
- CpusetsInit class for initialization and capability detection
- TaskMigrate class for moving processes between cpusets
- Context manager support for automatic cleanup
- Full cgroup v2 support including cpuset.cpus.partition

The module modernizes cpuset management for cgroup v2, replacing
older cgroup v1 interfaces while maintaining API compatibility
where possible.

Signed-off-by: John Kacur <jkacur@redhat.com>
---
rteval/cpuset.py | 350 +++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 350 insertions(+)
create mode 100755 rteval/cpuset.py

diff --git a/rteval/cpuset.py b/rteval/cpuset.py
new file mode 100755
index 000000000000..2fdbabf40125
--- /dev/null
+++ b/rteval/cpuset.py
@@ -0,0 +1,350 @@
+#!/usr/bin/python3
+# Copyright 2026 John Kacur <jkacur@redhat.com>
+""" Module for creating and using cpusets """
+
+import errno
+import os
+from glob import glob
+
+
+class Cpuset:
+ """ Class for manipulating cpusets """
+
+ mpath = '/sys/fs/cgroup'
+
+ def __init__(self, name=None):
+ self.cpuset_name = None
+ self._cpuset_path = None
+ self.create_cpuset(name)
+
+ @property
+ def cpuset_path(self):
+ """ Return the cpuset path, for example /sys/fs/cgroup/rteval_0 """
+ return self._cpuset_path
+
+ def __str__(self):
+ return f'cpuset_name={self.cpuset_name}, cpuset_path={self.cpuset_path}'
+
+ def create_cpuset(self, name=None):
+ """ Create a cpuset (cgroup) below /sys/fs/cgroup """
+ if name is None:
+ raise ValueError("cpuset name cannot be None")
+ self.cpuset_name = name
+ path = os.path.join(Cpuset.mpath, self.cpuset_name)
+ if not os.path.exists(path):
+ os.mkdir(path)
+ self._cpuset_path = path
+ # Enable cpuset controller in parent if needed
+ self._enable_controllers()
+
+ def _enable_controllers(self):
+ """ Enable cpuset controller in parent cgroup """
+ # Get parent path
+ parent_path = os.path.dirname(self._cpuset_path)
+ if parent_path == '/sys/fs':
+ parent_path = '/sys/fs/cgroup'
+
+ subtree_control = os.path.join(parent_path, 'cgroup.subtree_control')
+ try:
+ # Read current controllers
+ with open(subtree_control, 'r', encoding='utf-8') as f:
+ current = f.read().strip()
+
+ # Enable cpuset if not already enabled
+ # Controllers are space-separated, so split and check
+ if 'cpuset' not in current.split():
+ with open(subtree_control, 'w', encoding='utf-8') as f:
+ f.write('+cpuset')
+ except (OSError, PermissionError):
+ # May fail if already enabled or no permissions
+ pass
+
+ def assign_cpus(self, cpu_str):
+ """
+ Assign cpus in list-format to a cpuset
+ Note: In cgroup v2, you must call write_memnode() before assign_cpus()
+ """
+ path = os.path.join(self._cpuset_path, "cpuset.cpus")
+ with open(path, 'w', encoding='utf-8') as f:
+ f.write(cpu_str)
+
+ def write_pid(self, pid):
+ """
+ Place a pid in a cpuset
+ Returns True if successful, False if the process cannot be moved
+ Raises OSError for unexpected errors
+ """
+ path = os.path.join(self._cpuset_path, "cgroup.procs")
+ pid_str = str(pid)
+ try:
+ with open(path, 'w', encoding='utf-8') as f:
+ f.write(pid_str)
+ f.flush()
+ return True
+ except OSError as err:
+ # EINVAL: Invalid argument (process can't be moved, or invalid PID format)
+ # ESRCH: No such process (process died between read and write)
+ # EBUSY: Device or resource busy (process can't be moved)
+ # EACCES: Permission denied (some kernel threads can't be moved)
+ if err.errno in (errno.EINVAL, errno.ESRCH, errno.EBUSY, errno.EACCES):
+ return False
+ else:
+ raise
+
+ def write_memnode(self, memnode):
+ """ Place memnode in a cpuset """
+ path = os.path.join(self._cpuset_path, "cpuset.mems")
+ memnode_str = str(memnode)
+ with open(path, 'w', encoding='utf-8') as f:
+ f.write(memnode_str)
+
+ def write_cpu_exclusive(self, cpu_exclusive):
+ """
+ Set CPU partition type (cgroup v2)
+ In v2, cpu_exclusive is replaced by cpuset.cpus.partition
+
+ Args:
+ cpu_exclusive: True/1/'1' for 'isolated', False/0/'0' for 'member'
+ """
+ path = os.path.join(self._cpuset_path, "cpuset.cpus.partition")
+ if not os.path.exists(path):
+ # cpuset.cpus.partition may not exist in all kernels
+ return
+
+ # Convert to boolean: handles bool, int, or string input
+ if isinstance(cpu_exclusive, str):
+ is_exclusive = cpu_exclusive not in ('0', '', 'false', 'False')
+ else:
+ is_exclusive = bool(cpu_exclusive)
+
+ partition_type = 'isolated' if is_exclusive else 'member'
+ try:
+ with open(path, 'w', encoding='utf-8') as f:
+ f.write(partition_type)
+ except OSError:
+ # Partition may not be supported or may fail
+ pass
+
+ def set_cpu_exclusive(self):
+ """ Turn cpu_exclusive on (set partition to isolated) """
+ self.write_cpu_exclusive(True)
+
+ def unset_cpu_exclusive(self):
+ """ Turn cpu_exclusive off (set partition to member) """
+ self.write_cpu_exclusive(False)
+
+ def write_memory_migrate(self, memory_migrate):
+ """
+ Memory migrate is not available in cgroup v2
+ This is kept for API compatibility but does nothing
+ """
+ pass
+
+ def set_memory_migrate(self):
+ """ Memory_migrate not available in cgroup v2 (no-op for compatibility) """
+ pass
+
+ def unset_memory_migrate(self):
+ """ Memory migrate not available in cgroup v2 (no-op for compatibility) """
+ pass
+
+ def destroy(self):
+ """
+ Remove this cpuset (cgroup directory)
+ The cpuset must be empty (no processes) before it can be removed.
+ Raises OSError if the directory is not empty or doesn't exist.
+ """
+ if self._cpuset_path and os.path.exists(self._cpuset_path):
+ os.rmdir(self._cpuset_path)
+
+ def __enter__(self):
+ """
+ Context manager entry point
+ Allows usage: with Cpuset('name') as cpuset: ...
+ """
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ """
+ Context manager exit point
+ Automatically destroys the cpuset when exiting the with block
+ """
+ self.destroy()
+ return False # Don't suppress exceptions
+
+ def get_tasks(self):
+ """ Get the tasks (PIDs) currently in a cpuset """
+ path = os.path.join(self.cpuset_path, "cgroup.procs")
+ with open(path, 'r', encoding='utf-8') as f:
+ return [line.strip() for line in f if line.strip()]
+
+
+class CpusetsInit(Cpuset):
+ """
+ CpusetsInit initializes cpusets (cgroup v2)
+ It verifies that cgroup v2 is mounted and cpuset controller is available
+ This class represents the root cgroup /sys/fs/cgroup
+ """
+
+ mpath = '/sys/fs/cgroup'
+
+ def __init__(self):
+ self._supported = self._cpuset_supported()
+ self._numa_nodes = self._get_numa_nodes() if self._supported else 0
+ if self._supported:
+ # Set attributes directly for the root cgroup
+ self.cpuset_name = None
+ self._cpuset_path = Cpuset.mpath
+
+ @property
+ def supported(self):
+ """ Return True if cpuset is supported """
+ return self._supported
+
+ @property
+ def numa_nodes(self):
+ """ Return the number of numa nodes """
+ return self._numa_nodes
+
+ @staticmethod
+ def _cpuset_supported():
+ """
+ Return True if cpusets are supported
+ For cgroup v2, check if:
+ 1. cgroup2 filesystem is available
+ 2. /sys/fs/cgroup is mounted
+ 3. cpuset controller is available
+ """
+ # Check if cgroup2 is in /proc/filesystems
+ cgroup2_found = False
+ dpath = '/proc/filesystems'
+ try:
+ with open(dpath, encoding='utf-8') as fp:
+ for line in fp:
+ elems = line.strip().split()
+ if len(elems) == 2 and elems[1] == "cgroup2":
+ cgroup2_found = True
+ break
+ except OSError:
+ return False
+
+ if not cgroup2_found:
+ return False
+
+ # Check if /sys/fs/cgroup is mounted and has cpuset controller
+ cgroup_path = '/sys/fs/cgroup'
+ controllers_file = os.path.join(cgroup_path, 'cgroup.controllers')
+
+ if not os.path.exists(controllers_file):
+ return False
+
+ try:
+ with open(controllers_file, encoding='utf-8') as f:
+ controllers = f.read().strip().split()
+ return 'cpuset' in controllers
+ except OSError:
+ return False
+
+ @staticmethod
+ def _get_numa_nodes():
+ """
+ return the number of numa nodes
+ used by init to set _numa_nodes
+ Users of this class should use the "numa_nodes" method
+ """
+ return len(glob('/sys/devices/system/node/node*'))
+
+
+class TaskMigrate:
+ """ A class for migrating a set of pids from one cpuset to another """
+
+ def __init__(self, cpuset_from, cpuset_to):
+ self.cpuset_from = cpuset_from
+ self.cpuset_to = cpuset_to
+ # memory_migrate is not available in cgroup v2, but keep for API compat
+ self.cpuset_to.set_memory_migrate()
+ self.migrated = 0
+ self.failed = 0
+
+ def migrate(self):
+ """ Do the migration, only attempt on pids where this is allowed """
+ self.migrated = 0
+ self.failed = 0
+ tasks = self.cpuset_from.get_tasks()
+ for task in tasks:
+ if self.cpuset_to.write_pid(task):
+ self.migrated += 1
+ else:
+ self.failed += 1
+ return self.migrated, self.failed
+
+
+if __name__ == '__main__':
+
+ cpusets_init = CpusetsInit()
+ print(f'cpusets_init.supported = {cpusets_init.supported}')
+ if cpusets_init.supported:
+ print(f'cpusets_init.numa_nodes = {cpusets_init.numa_nodes}')
+ print(f'cpusets_init.cpuset_path = {cpusets_init.cpuset_path}')
+
+ # Creating and manipulating cgroups requires root permissions
+ print("\nTo test cpuset creation, run as root:")
+ print(" sudo python3 cpuset.py")
+
+ try:
+ # Example 1: Using context manager (recommended)
+ print("\n" + "="*60)
+ print("Example 1: Context Manager (automatic cleanup)")
+ print("="*60)
+ with Cpuset('rteval_demo') as cpuset:
+ print(f"Created cpuset: {cpuset.cpuset_path}")
+ cpuset.write_memnode('0')
+ cpuset.assign_cpus('0-3')
+ print(f"Configured cpuset with CPUs 0-3")
+ print("Context manager exited - cpuset automatically destroyed!")
+
+ # Example 2: Manual cleanup (traditional approach)
+ print("\n" + "="*60)
+ print("Example 2: Manual Cleanup (traditional)")
+ print("="*60)
+ print("Creating cpuset0...")
+ cpuset0 = Cpuset('rteval_0')
+ print("Creating cpuset1...")
+ cpuset1 = Cpuset('rteval_1')
+
+ # In cgroup v2, must set mems before cpus
+ print("Setting memory nodes...")
+ cpuset0.write_memnode('0')
+ cpuset1.write_memnode('0')
+
+ print("Assigning CPUs to cpuset0 (0-4,8-11)...")
+ cpuset0.assign_cpus('0-4,8-11')
+ print("Assigning CPUs to cpuset1 (5-7)...")
+ cpuset1.assign_cpus('5-7')
+
+ print("Setting cpu_exclusive on cpuset1...")
+ cpuset1.set_cpu_exclusive()
+
+ print("Migrating tasks from root to cpuset0...")
+ tm = TaskMigrate(cpusets_init, cpuset0)
+ migrated, failed = tm.migrate()
+
+ print(f"\nMigration complete: {migrated} moved, {failed} skipped")
+ print(f"Tasks now in cpuset0: {len(cpuset0.get_tasks())}")
+
+ # Cleanup: move processes back to root before removing cgroups
+ print("\nCleaning up: moving processes back to root cgroup...")
+ tm_back = TaskMigrate(cpuset0, cpusets_init)
+ moved_back, _ = tm_back.migrate()
+ print(f"Moved {moved_back} processes back to root")
+
+ cpuset0.destroy()
+ cpuset1.destroy()
+ print("Manual cleanup complete")
+ except PermissionError as e:
+ print(f"\nPermission denied: {e}")
+ print("Run as root to create and manipulate cgroups")
+ except Exception as e:
+ print(f"\nError at step: {e}")
+ import traceback
+ traceback.print_exc()
--
2.53.0