Last active
July 11, 2025 02:27
-
-
Save chen3feng/6ca10c96212f01391466409c8e78f82b to your computer and use it in GitHub Desktop.
Get CPU and Memory Quota
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import os | |
| import glob | |
| import platform | |
| import subprocess | |
| def get_physical_cpu_cores(): | |
| plat = platform.system() | |
| if plat == 'Linux': | |
| return get_physical_cpu_cores_linux() | |
| elif plat == 'Darwin': | |
| return get_physical_cpu_cores_mac() | |
| elif plat == 'Windows': | |
| return get_physical_cpu_cores_windows() | |
| else: | |
| return os.cpu_count() or 1 | |
| def get_physical_cpu_cores_linux(): | |
| try: | |
| output = subprocess.check_output(['lscpu'], encoding='utf-8') | |
| cores = None | |
| sockets = None | |
| for line in output.splitlines(): | |
| if line.startswith('Core(s) per socket:'): | |
| cores = int(line.split(':')[1].strip()) | |
| elif line.startswith('Socket(s):'): | |
| sockets = int(line.split(':')[1].strip()) | |
| if cores is not None and sockets is not None: | |
| return cores * sockets | |
| except Exception: | |
| pass | |
| # fallback to logical cpu count | |
| return os.cpu_count() or 1 | |
| def get_physical_cpu_cores_mac(): | |
| try: | |
| output = subprocess.check_output(['sysctl', 'hw.physicalcpu'], encoding='utf-8') | |
| return int(output.strip().split(': ')[1]) | |
| except Exception: | |
| return os.cpu_count() or 1 | |
| def get_physical_cpu_cores_windows(): | |
| try: | |
| output = subprocess.check_output( | |
| ['wmic', 'cpu', 'get', 'NumberOfCores'], | |
| encoding='utf-8', stderr=subprocess.DEVNULL | |
| ) | |
| lines = output.strip().splitlines() | |
| cores = 0 | |
| for line in lines[1:]: | |
| line = line.strip() | |
| if line.isdigit(): | |
| cores += int(line) | |
| return cores if cores > 0 else (os.cpu_count() or 1) | |
| except Exception: | |
| return os.cpu_count() or 1 | |
| def get_cpu_quota(): | |
| if platform.system() != 'Linux': | |
| # macOS and Windows: no cgroup, return physical cores | |
| return float(get_physical_cpu_cores()) | |
| # Linux: check cgroup v2 cpu.max | |
| cgroup2_cpu_max = '/sys/fs/cgroup/cpu.max' | |
| if os.path.isfile(cgroup2_cpu_max): | |
| line = read_first_line(cgroup2_cpu_max) | |
| if line: | |
| quota, period = line.split() | |
| if quota == 'max': | |
| return float(get_physical_cpu_cores()) | |
| try: | |
| return float(quota) / float(period) | |
| except Exception: | |
| return float(get_physical_cpu_cores()) | |
| # Linux: check cgroup v1 cpu.cfs_quota_us and cpu.cfs_period_us | |
| cgroup_cpu_paths = glob.glob('/sys/fs/cgroup/cpu*/cpu.cfs_quota_us') | |
| for quota_path in cgroup_cpu_paths: | |
| period_path = quota_path.replace('cpu.cfs_quota_us', 'cpu.cfs_period_us') | |
| if os.path.isfile(quota_path) and os.path.isfile(period_path): | |
| quota_str = read_first_line(quota_path) | |
| period_str = read_first_line(period_path) | |
| try: | |
| quota = int(quota_str) | |
| period = int(period_str) | |
| if quota > 0 and period > 0: | |
| return float(quota) / float(period) | |
| elif quota == -1: | |
| return float(get_physical_cpu_cores()) | |
| except Exception: | |
| continue | |
| return float(get_physical_cpu_cores()) | |
| def get_memory_quota(): | |
| """ | |
| Obtain memory quota in bytes. | |
| return memory quota if run in container otherwise return physical memory size | |
| return None on error | |
| """ | |
| if platform.system() == 'Linux': | |
| cgroup_v2_mem_max = '/sys/fs/cgroup/memory.max' | |
| cgroup_v1_mem_limit = '/sys/fs/cgroup/memory/memory.limit_in_bytes' | |
| if os.path.isfile(cgroup_v2_mem_max): | |
| val = read_first_line(cgroup_v2_mem_max) | |
| if val != 'max': | |
| try: | |
| return int(val) | |
| except Exception: | |
| pass | |
| elif os.path.isfile(cgroup_v1_mem_limit): | |
| val = read_first_line(cgroup_v1_mem_limit) | |
| try: | |
| val_int = int(val) | |
| if val_int < 1 << 50: | |
| return val_int | |
| except Exception: | |
| pass | |
| return get_total_memory() | |
| def get_total_memory(): | |
| system = platform.system() | |
| if system == 'Linux': | |
| return get_total_memory_linux() | |
| elif system == 'Darwin': | |
| return get_total_memory_mac() | |
| elif system == 'Windows': | |
| return get_total_memory_windows() | |
| else: | |
| return None | |
| def get_total_memory_linux(): | |
| try: | |
| with open('/proc/meminfo', 'r') as f: | |
| for line in f: | |
| if line.startswith('MemTotal:'): | |
| parts = line.split() | |
| # MemTotal is in kB | |
| return int(parts[1]) * 1024 | |
| except Exception: | |
| pass | |
| return None | |
| def get_total_memory_mac(): | |
| try: | |
| output = subprocess.check_output(['sysctl', 'hw.memsize'], encoding='utf-8') | |
| # output like: hw.memsize: 17179869184 | |
| return output.strip().split(': ')[1] | |
| except Exception: | |
| pass | |
| return None | |
| def get_total_memory_windows(): | |
| try: | |
| import ctypes | |
| class MEMORYSTATUSEX(ctypes.Structure): | |
| _fields_ = [ | |
| ('dwLength', ctypes.c_ulong), | |
| ('dwMemoryLoad', ctypes.c_ulong), | |
| ('ullTotalPhys', ctypes.c_ulonglong), | |
| ('ullAvailPhys', ctypes.c_ulonglong), | |
| ('ullTotalPageFile', ctypes.c_ulonglong), | |
| ('ullAvailPageFile', ctypes.c_ulonglong), | |
| ('ullTotalVirtual', ctypes.c_ulonglong), | |
| ('ullAvailVirtual', ctypes.c_ulonglong), | |
| ('ullAvailExtendedVirtual', ctypes.c_ulonglong), | |
| ] | |
| stat = MEMORYSTATUSEX() | |
| stat.dwLength = ctypes.sizeof(MEMORYSTATUSEX) | |
| ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(stat)) | |
| return int(stat.ullTotalPhys) | |
| except Exception: | |
| pass | |
| return None | |
| def read_first_line(path): | |
| try: | |
| with open(path, 'r') as f: | |
| return f.readline().strip() | |
| except Exception: | |
| return None | |
| def main(): | |
| print(f"PHY Cores: {get_physical_cpu_cores()}") | |
| print(f"CPU Quota: {get_cpu_quota()}") | |
| print(f"MEM Size: {get_total_memory()}") | |
| print(f"MEM Quota: {get_memory_quota()}") | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment