vllm.utils.cpu_resource_utils ¶

@cache
def _synthesize_cpu_list() -> list[LogicalCPUInfo]:
    """Synthesize a flat CPU list: each logical CPU is its own core on
    NUMA node 0.  Used when lscpu output is unavailable or unparsable
    (e.g. macOS, RISC-V)."""
    cpu_count = os.cpu_count()
    assert cpu_count
    return [LogicalCPUInfo(i, i, 0) for i in range(cpu_count)]

@cache
def get_cgroup_memory_limit() -> tuple[int | None, int | None]:
    """Return (limit, usage) in bytes from cgroup, or (None, None).

    Supports both cgroup v2 (unified) and v1. Returns (None, None) when
    not running under a constrained cgroup (e.g. bare metal, or limit
    reported as `max`/an unrealistically large value).
    """
    if sys.platform != "linux":
        return None, None

    # cgroup v2 unified hierarchy
    v2_limit = _read_int_file("/sys/fs/cgroup/memory.max")
    if v2_limit is not None:
        v2_usage = _read_int_file("/sys/fs/cgroup/memory.current")
        return v2_limit, v2_usage

    # cgroup v1
    v1_limit = _read_int_file("/sys/fs/cgroup/memory/memory.limit_in_bytes")
    if v1_limit is not None:
        # cgroup v1 reports a huge sentinel (close to PAGE_COUNTER_MAX)
        # when unlimited. Treat absurdly large values as "no limit".
        if v1_limit >= (1 << 62):
            return None, None
        v1_usage = _read_int_file("/sys/fs/cgroup/memory/memory.usage_in_bytes")
        return v1_limit, v1_usage

    return None, None

def parse_id_list(raw_str: str) -> list[int]:
    """Parses strings like '0-2,4,7-8' into [0, 1, 2, 4, 7, 8]"""
    result: list[int] = []
    if not raw_str:
        return result

    for part in raw_str.split(","):
        if "-" in part:
            start, end = map(int, part.split("-"))
            result.extend(range(start, end + 1))
        else:
            result.append(int(part))
    return sorted(list(set(result)))