如何在Dask SSHCluster中以编程方式禁用nanny进程？

问题背景 #

使用多主机Dask SSHCluster时，单主机启动多个Worker会导致内存占用过高甚至崩溃。尝试改为每个主机启动1个Worker，并在Worker内使用multiprocessing池：

manager = mp.Manager()
pool = mp.Pool()

或使用Dask Distributed的processes调度器：

delayed_values = [delayed(worker)(param) for param in params]
futures = compute(delayed_values, scheduler="processes", num_processes=n_processes)

均触发以下错误：

AssertionError: daemonic processes are not allowed to have children

问题根源是Dask默认通过Nanny进程管理Worker，Nanny创建的Worker是守护进程，而守护进程无法创建子进程。已知通过--no-nanny参数可以禁用Nanny，但需要代码层面的实现方式。

解决方案 #

在创建SSHCluster时，直接在worker_options中添加"nanny": False参数，即可禁用Nanny进程。

修改后的完整代码示例：

cluster = SSHCluster(
    ["localhost", "hostname"],
    connect_options={"known_hosts": None},
    worker_options={
        "n_workers": 1,  # 每个主机仅启动1个Worker
        "nanny": False   # 禁用Nanny进程
    },
    scheduler_options={"port": 0, "dashboard_address": ":8797"},
)

client = Client(cluster)

说明 #

禁用Nanny后，Worker进程不再是守护进程，此时就可以在Worker内部安全地创建multiprocessing子进程，或使用processes调度器执行任务，不会再触发守护进程相关的断言错误。

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