小编Jef*_*eff的帖子

想象一下,我有一个拥有10名工人和40个核心总数的dask网格.这是一个共享网格,所以我不想用我的工作完全浸透它.我有1000个任务要做,我想一次提交(并且主动运行)最多20个任务.

具体来说,

from time import sleep
from random import random

def inc(x):
    from random import random
    sleep(random() * 2)
    return x + 1

def double(x):
    from random import random
    sleep(random())
    return 2 * x

>>> from distributed import Executor
>>> e = Executor('127.0.0.1:8786')
>>> e
<Executor: scheduler=127.0.0.1:8786 workers=10 threads=40>

如果我设置了一个队列系统

>>> from queue import Queue
>>> input_q = Queue()
>>> remote_q = e.scatter(input_q)
>>> inc_q = e.map(inc, remote_q)
>>> double_q = e.map(double, inc_q)

这将工作,但是,这将把我的所有任务转储到网格,使其饱和.理想情况下我可以:

e.scatter(input_q, max_submit=20)

看来这里的文档示例允许我使用maxsize …