sysHAX智能调度算法：如何实现AI推理任务在CPU和GPU之间的动态分配

sysHAX智能调度算法：如何实现AI推理任务在CPU和GPU之间的动态分配

【免费下载链接】sysHAXsysHAX Heterogeneous collaborative acceleration runtime项目地址: https://gitcode.com/openeuler/sysHAX

前往项目官网免费下载：https://ar.openeuler.org/ar/

sysHAX作为一款Heterogeneous collaborative acceleration runtime，其核心价值在于通过智能调度算法实现AI推理任务在CPU和GPU之间的动态分配，从而最大化利用硬件资源，提升整体系统性能。本文将深入解析这一调度机制的工作原理和实现方式。

智能调度的核心架构

sysHAX的调度系统建立在多层次的硬件架构基础上，通过协调CPU和GPU的计算资源，实现任务的高效分配。

图1：sysHAX的CPU+GPU异构计算架构示意图

从整体架构来看，调度系统主要由以下几个核心模块组成：

• 调度器（src/core/scheduler.py）：负责决策任务分配策略
• 执行器（src/core/runner.py）：处理具体设备上的任务执行
• 监控器（src/core/monitor.py）：实时收集硬件资源使用情况
• 指标收集器（src/core/metrics.py）：统计性能数据并反馈给调度器

动态分配的决策机制

sysHAX调度算法的核心在于其动态决策机制，该机制基于实时硬件状态和任务特性做出智能分配。调度器会综合考虑以下因素：

1. 硬件资源使用情况

监控器持续收集CPU和GPU的关键指标，包括：

• 缓存使用率（GPU_CACHE_usage和CPU_CACHE_usage）
• 运行中请求数量
• 等待队列长度
• 吞吐量（tokens/s）

这些数据为调度决策提供了基础依据，确保任务分配符合当前硬件负载状况。

2. 任务特性分析

调度器会分析任务的特性，如是否包含num_decode_tokens参数，从而决定是否适合在CPU或GPU上执行。例如，当检测到任务包含num_decode_tokens时，会优先调度到CPU执行：

if "num_decode_tokens" in task_data: decision["device"] = "CPU" Logger.debug("任务包含num_decode_tokens，强制调度到CPU")

3. 动态调度策略

调度器实现了多种调度策略，根据不同场景智能选择：

• 负载均衡策略：当GPU或CPU负载过高时，自动将任务分配到负载较低的设备
• 吞吐量优先策略：比较GPU和CPU的吞吐量（tokens/s），优先选择吞吐量更高的设备
• 资源限制策略：当某一设备达到最大并发量时，自动将任务分配到另一设备

以下是调度决策的核心逻辑片段：

if decision["device"] == "GPU" and self.gpu_running_num < self.gpu_max_batch: scheduled["GPU"] += 1 task = asyncio.create_task(self._execute_task(decision["device"], task_data)) Logger.debug(f"任务分配到GPU执行") elif decision["device"] == "CPU" and self.cpu_running_num < self.cpu_max_batch: Logger.debug("自动开启CPU侧prefill任务的num_decode_tokens=1以启用部分解码卸载") scheduled["CPU"] += 1 task = asyncio.create_task(self._execute_task(decision["device"], task_data)) Logger.debug(f"任务分配到CPU执行")

任务执行流程

一旦调度决策确定，任务将被分配到相应的设备执行。执行器（Runner）模块负责处理具体的任务执行逻辑：

图2：sysHAX任务部署与执行流程

执行流程主要包括：

1. 根据设备类型选择对应的服务URL：

   service_url = self.v1_chat_gpu if device == "GPU" else self.v1_chat_cpu

2. 执行任务并收集性能指标：

   async for chunk in self.metrics_service.stream_with_metrics(gen, device=device): yield chunk

3. 处理任务结果或错误：

   except Exception as e: Logger.error(f"{device}任务执行失败: {e}", exc_info=True) await self.resubmit_task(task_data, resubmit_task_data)

性能优化技术

sysHAX调度系统还集成了多种性能优化技术，进一步提升异构计算的效率：

1. 自动PD解耦

当开启auto_pd_offload功能时，系统会自动将CPU侧不适合执行的prefill任务转移到GPU执行：

if self.syshax_config.auto_pd_offload and decision["device"] == "CPU": # CPU侧不适合执行prefill任务，当开启auto_pd_offload会自动进行PD解耦 decision["device"] = "GPU"

2. 动态批处理

调度器会根据设备负载动态调整批处理大小，确保资源利用率最大化：

CPU_MAX_BATCH_SIZE = self.syshax_config.cpu_max_batch_size # 是否将任务转移到CPU if self.gpu_running_num + self.gpu_scheduled_running_num >= GPU_MAX_BATCH_SIZE: use_cpu = True elif self.cpu_running_num + self.gpu_scheduled_running_num >= CPU_MAX_BATCH_SIZE: use_cpu = False

3. 实时监控与自适应调整

监控器持续跟踪系统状态，调度器根据反馈实时调整策略：

monitor_data["gpu_cache_usage"] = self._parse_metrics(monitor_text, RE_GPU_CACHE, float) monitor_data["cpu_cache_usage"] = self._parse_metrics(monitor_text, RE_CPU_CACHE, float)

实际应用效果

通过这些智能调度策略，sysHAX能够显著提升AI推理任务的执行效率。具体表现为：

• 资源利用率提升：通过动态分配，CPU和GPU资源得到充分利用
• 响应时间优化：根据任务特性和硬件状态分配任务，减少等待时间
• 系统吞吐量提高：优先将任务分配到吞吐量更高的设备，提升整体处理能力

图3：sysHAX系统架构与性能监控界面

总结

sysHAX的智能调度算法通过综合考虑硬件资源、任务特性和实时性能数据，实现了AI推理任务在CPU和GPU之间的动态分配。这一机制不仅最大化了资源利用率，还显著提升了系统的整体性能和响应速度。

通过src/core/scheduler.py、src/core/runner.py等核心模块的协同工作，sysHAX为异构计算环境下的AI推理任务提供了高效、智能的调度解决方案。无论是高负载还是低负载场景，系统都能自动调整策略，确保最佳性能表现。

如果您想体验这一强大的调度系统，可以通过以下命令获取源码：

git clone https://gitcode.com/openeuler/sysHAX

探索sysHAX，开启智能异构计算之旅！🚀

【免费下载链接】sysHAXsysHAX Heterogeneous collaborative acceleration runtime项目地址: https://gitcode.com/openeuler/sysHAX