diff --git a/vllm/distributed/kv_transfer/kv_connector/v1/nixl_connector.py b/vllm/distributed/kv_transfer/kv_connector/v1/nixl_connector.py index c2777b393..245ac7daf 100644 --- a/vllm/distributed/kv_transfer/kv_connector/v1/nixl_connector.py +++ b/vllm/distributed/kv_transfer/kv_connector/v1/nixl_connector.py @@ -926,6 +926,17 @@ class NixlConnectorWorker: else: self.use_host_buffer = self.kv_buffer_device == "cpu" + # reserve different cores for start_load_kv() from model_forward() + if self.device_type == "cpu": + numa_core_list = current_platform.discover_numa_topology() + # setup one last core in each numa for kv transfer. + rsv_cores_for_kv = [ + max(each_numa_core_list) for each_numa_core_list in numa_core_list + ] + + if rsv_cores_for_kv: + os.sched_setaffinity(0, rsv_cores_for_kv) + # support for oot platform which can't register nixl memory # type based on kv_buffer_device nixl_memory_type = current_platform.get_nixl_memory_type() diff --git a/vllm/platforms/cpu.py b/vllm/platforms/cpu.py index 46465a482..60180b272 100644 --- a/vllm/platforms/cpu.py +++ b/vllm/platforms/cpu.py @@ -213,6 +213,13 @@ class CpuPlatform(Platform): cache_config.cpu_kvcache_space_bytes = CpuPlatform.get_device_total_memory() + # reserve at least one core for nixl_connector under p/d case + if vllm_config.kv_transfer_config and ( + envs.VLLM_CPU_NUM_OF_RESERVED_CPU == 0 + or envs.VLLM_CPU_NUM_OF_RESERVED_CPU is None + ): + os.environ["VLLM_CPU_NUM_OF_RESERVED_CPU"] = "1" + parallel_config = vllm_config.parallel_config if ( parallel_config.world_size > 1 @@ -396,6 +403,60 @@ class CpuPlatform(Platform): return allowed_numa_nodes_list, logical_cpu_list + @classmethod + def discover_numa_topology(cls) -> list[list[int]]: + """ + Discover NUMA topology and keep the last physical core of each numa + into one core group list for nixl start_kv_load() + """ + SYS_NODE = "/sys/devices/system/node" + SYS_CPU = "/sys/devices/system/cpu" + + if not (os.path.exists(SYS_NODE) and os.path.exists(SYS_CPU)): + return [] + + core_rsv_for_kv = [] + for node in os.listdir(SYS_NODE): + if not node.startswith("node") or not node[4:].isdigit(): + continue + node_path = f"{SYS_NODE}/{node}" + + seen_phys = set() + for cpu in os.listdir(node_path): + if not cpu.startswith("cpu") or not cpu[3:].isdigit(): + continue + + cpu_id = int(cpu[3:]) + # thread_siblings based on cpu_id + path = f"{SYS_CPU}/cpu{cpu_id}/topology/thread_siblings_list" + + if os.path.exists(path): + try: + with open(path) as f: + s = f.read() + cpus: list[int] = [] + for part in s.strip().split(","): + if "-" in part: + a, b = map(int, part.split("-")) + cpus.extend(range(a, b + 1)) + else: + cpus.append(int(part)) + siblings = cpus if cpus else [cpu_id] + except (OSError, ValueError): + siblings = [cpu_id] + else: + siblings = [cpu_id] + + phys = min(siblings) + + if phys not in seen_phys: + seen_phys.add(phys) + + if len(seen_phys) > 0: + core_rsv_for_kv.append(list(seen_phys)) + + return core_rsv_for_kv + @classmethod def is_pin_memory_available(cls) -> bool: return False diff --git a/vllm/v1/worker/cpu_worker.py b/vllm/v1/worker/cpu_worker.py index 169696ca1..8ccd45bb0 100644 --- a/vllm/v1/worker/cpu_worker.py +++ b/vllm/v1/worker/cpu_worker.py @@ -54,6 +54,7 @@ class CPUWorker(Worker): def init_device(self): # Setup OpenMP threads affinity. omp_cpuids = envs.VLLM_CPU_OMP_THREADS_BIND + # Under numa binding some cores reserved for kv transfer in nixl_connector.py if omp_cpuids == "auto" and platform.system() == "Linux": cpu_arch = current_platform.get_cpu_architecture() if cpu_arch in (CpuArchEnum.POWERPC, CpuArchEnum.S390X):