如何在Dask Distributed中实现简单FIFO调度避免Worker崩溃？

• 环境配置：多客户端（作为服务器）、1个调度器、1个配备3线程的Worker；客户端采用异步模式
• 任务提交代码：

processing_futures = await client.gather(client.compute(response, priority=100, 
                                                        resources={"GPU_RAM": 5}, # Worker总GPU内存为16
                                                        fifo_timeout='0ms'))

• 核心问题：当任务量超过1000个时，Worker会尝试一次性加载所有初始任务，导致大量中间结果占满内存引发崩溃；此任务属于易并行类型，但Dask默认处理逻辑导致内存过载
• 需求：实现任务逐个处理或按10个一组批量处理，而非一次性全部加载；确认是否可通过资源系统解决该问题，以及fifo_timeout='0ms'的作用是否符合预期

1. 优化Worker预取与内存控制 #

调整Worker行为，避免提前加载过多任务：

• 启动Worker时设置--prefetch-factor 1，让Worker仅预取与当前运行任务数量相等的任务（3线程Worker最多预取3个额外任务）
• 配置Worker内存阈值，触发溢写/资源释放：

  dask worker tcp://scheduler:8786 --nthreads 3 --resources GPU_RAM=16 --memory-target 0.6 --memory-spill 0.7 --memory-pause 0.8 --memory-terminate 0.9
  

  当内存占用达60%时开始将数据溢写到磁盘，避免直接触发内存过载。

2. 严格控制任务并发数 #

方式一：异步信号量限制并发提交 #

在客户端用asyncio.Semaphore控制同时提交的任务数（示例限制为10）：

import asyncio

sem = asyncio.Semaphore(10)

async def process_single_task(task):
    async with sem:
        future = await client.compute(task, priority=100, resources={"GPU_RAM":5}, fifo_timeout='0ms')
        return await future

# 假设response为包含所有任务的列表
processing_results = await asyncio.gather(*[process_single_task(t) for t in response])

方式二：分批次提交并等待完成 #

将任务按每组10个拆分，完成一组后再提交下一组：

batch_size = 10
all_results = []

for i in range(0, len(response), batch_size):
    batch = response[i:i+batch_size]
    futures = await client.compute(batch, priority=100, resources={"GPU_RAM":5}, fifo_timeout='0ms')
    batch_results = await client.gather(futures)
    all_results.extend(batch_results)

3. 强化资源调度与FIFO策略 #

• 确保Worker启动时声明正确的资源总量：--resources GPU_RAM=16，让调度器准确计算可同时承载的任务数（每个任务占5GPU内存，最多并行3个）
• 保持fifo_timeout='0ms'以禁用任务重排，确保任务严格按提交顺序执行，避免调度器提前派发过多任务
• 启动调度器时设置--worker-saturation 1，让调度器仅向Worker派发其当前可处理的任务数量，杜绝超额派发

4. 中间结果清理优化 #

若任务中间结果无需长期保留，处理完成后手动释放资源：

# 处理单个任务后释放相关资源
result = await future
del future
await client.cancel(future)

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