共享内存

内容

共享内存¶

异步调度器接受任何 concurrent.futures.Executor 实例。这包括 Python 标准库中定义的 ThreadPoolExecutor 和 ProcessPoolExecutor 实例，以及任何第三方库中的子类。Dask 还定义了自己的 SynchronousExecutor，它只是简单地在主线程上运行函数（对调试有用）。

完整的 dask get 函数分别存在于 dask.threaded.get、dask.multiprocessing.get 和 dask.get 中。

政策¶

异步调度器维护着索引数据结构，这些结构显示了哪些任务依赖于哪些数据，哪些数据是可用的，哪些数据在等待哪些任务完成之前无法释放，以及哪些任务当前正在运行。它可以在常数时间内相对于总任务数和可用任务数更新这些索引结构。这些索引结构使得dask异步调度器能够在单台机器上扩展到非常多的任务。

为了保持内存占用小，我们选择将准备运行的任务保存在一个后进先出的栈中，这样最新可用的任务会优先执行。这鼓励在开始新的任务链之前完成相关的任务链。这也可以在常数时间内查询。

性能¶

简而言之 线程调度器的开销大致如下：

每个任务有200微秒的开销
10微秒启动时间（如果你希望每次都创建一个新的 ThreadPoolExecutor）
任务数量的常数缩放
每个任务的依赖项数量的线性缩放

调度器引入了开销。这种开销实际上限制了我们并行性的粒度。下面我们测量了在不同应用函数（线程化、同步、多进程）和不同负载类型（极度并行、密集通信）下异步调度器的开销。

我们可以做的最快/最简单的测试是使用 IPython 的 timeit 魔法：

In [1]: import dask.array as da

In [2]: x = da.ones(1000, chunks=(2,)).sum()

In [3]: len(x.dask)
Out[3]: 1168

In [4]: %timeit x.compute()
33 ms +- 337 us per loop (mean +- std. dev. of 7 runs, 10 loops each)

因此，每个任务大约需要 90 微秒。其中大约 100 毫秒来自开销：

In [5]: x = da.ones(1000, chunks=(1000,)).sum()

In [6]: %timeit x.compute()
1.23 ms +- 36.5 us per loop (mean +- std. dev. of 7 runs, 1,000 loops each)

每次启动 ThreadPoolExecutor 都会有一些开销。这可以通过使用全局或上下文池来缓解：

>>> from concurrent.futures import ThreadPoolExecutor
>>> pool = ThreadPoolExecutor()
>>> dask.config.set(pool=pool)  # set global ThreadPoolExecutor

or

>>> with dask.config.set(pool=pool)  # use ThreadPoolExecutor throughout with block
...     ...