Dask并行仿真Worker卸载阶段CPU占用过高致数值异常求助

你的问题核心是Dask并行仿真末期CPU过载，伴随NaN计算错误，而串行执行正常，说明问题出在Dask的任务调度、结果收集阶段，或是Numba与Dask的并行机制冲突。以下是具体排查和解决建议：

1. 排查Numba与Dask的并行冲突 #

Numba的@jit装饰器如果启用了parallel=True，会在单任务内部开启多线程并行。若Dask Worker同时使用多线程，会导致逻辑核心被过度占用（20个Worker×Numba多线程），引发资源竞争，不仅会让CPU占用飙升，还可能因计算资源不足导致数值不稳定产生NaN。

• 解决：
  • 若Numba函数用了parallel=True，启动Dask Client时强制每个Worker只用1个线程：

    from dask.distributed import Client
    client = Client(n_workers=20, threads_per_worker=1)  # 匹配20颗物理核心，避免超线程竞争
    

  • 或者修改Numba装饰器为单线程：@njit(parallel=False)，让Dask负责多任务并行，Numba专注单任务编译加速。

2. 优化结果收集阶段的资源占用 #

仿真末期CPU飙升大概率是大量任务结果同时回传、序列化导致的。大NumPy数组的序列化/反序列化会消耗大量CPU资源，甚至可能因内存临时过载导致数据损坏出现NaN。

• 解决：
  • 分批收集结果：即使使用as_completed，也可以限制同时处理的任务数，避免一次性回传所有结果：

    from dask.distributed import as_completed
    
    futures_list = [client.compute(lazy) for lazy in lazy_results]
    results = []
    # 每次处理5个完成的任务，控制回传压力
    for batch in as_completed(futures_list, batch_size=5):
        for future in batch:
            results.append(future.result())
    

  • 本地文件暂存结果：让每个Worker将仿真结果写入本地文件，而非直接回传内存数组，最后再批量读取：

    import numpy as np
    def modelo_dinamico_with_save(...):
        result = modelo_dinamico(...)
        # 保存结果到Worker本地路径
        np.save(f"result_{i}.npy", result)
        return f"result_{i}.npy"  # 只回传文件名
    
    # 后续批量读取文件
    lazy_results = [dask.delayed(modelo_dinamico_with_save)(...) for ...]
    file_paths = dask.compute(*lazy_results)
    results = [np.load(fp) for fp in file_paths]
    

3. 定位NaN的产生源头 #

NaN并非必然在结果回传阶段产生，可能是某个任务在计算过程中因资源竞争（如CPU过载导致的计算中断）出现数值异常，但串行时资源充足不会触发。

• 解决：
  • 在modelo_dinamico函数末尾加入NaN检查，提前抛出错误定位问题任务：

    from numba import njit
    import numpy as np
    
    @njit
    def modelo_dinamico(...):
        # 原有计算逻辑
        ...
        # 检查结果是否含NaN
        if np.any(np.isnan(result_matrix)):
            raise ValueError(f"Task {task_id} generated NaN, input positions: {positions}")
        return result_matrix
    

  • 利用Dask Dashboard的Tasks面板，查看出错任务的详细日志，确认是特定输入导致的问题，还是随机的资源竞争问题。

4. 调整Dask Worker的资源配置 #

Windows环境下，Dask默认的Worker配置可能不匹配你的硬件（20颗双核处理器=40逻辑核心），导致任务调度效率低下，末期出现资源集中占用。

• 解决：
  • 限制Worker的内存上限，避免内存溢出触发虚拟内存交换（会大幅提升CPU占用）：

    client = Client(n_workers=20, threads_per_worker=1, memory_limit="4GB")  # 根据单任务内存占用调整
    

  • 关闭Dask的自适应调度：若启用了自适应，末期Worker可能频繁重启/调整，导致CPU波动，可在启动Client时关闭：

    client = Client(n_workers=20, threads_per_worker=1, adaptive=False)
    

5. 优化任务调用方式 #

你的两种调用方式本质都是批量提交任务，可尝试优化任务图的结构，减少不必要的数据传递：

• 解决：
  • 避免逐个传递Positions的8个元素，直接传递整个Positions[i]数组，减少参数传递的序列化开销：

    lazy_result = dask.delayed(modelo_dinamico)(Positions[i], npp, T_load, eff_data_sheet, fator_potencia_data_sheet, vel_data_sheet, i)
    

    同时修改modelo_dinamico函数接收数组参数，内部拆分使用。

内容的提问来源于stack exchange，提问作者nsantana