在 Python 中使用 ProcessPoolExecutor 的运行调用数不正确

Question

在 Python 的 concurrent.futures 标准模块中，为什么 ProcessPoolExecutor 中的运行调用数是 max_workers + 1 而不是像 ThreadPoolExecutor 中的 max_workers？仅当提交的调用数量严格大于池工作进程的数量时才会发生这种情况。

以下 Python 代码 sn-p 向 ProcessPoolExecutor 中的 2 个工作人员提交 8 个调用：

import concurrent.futures
import time


def call():
    while True:
        time.sleep(1)


if __name__ == "__main__":
    with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
        futures = [executor.submit(call) for _ in range(8)]
        time.sleep(5)

        for future in futures:
            print(future.running())

打印这个（3 个正在运行的调用；意外，因为有 2 个工作人员）：

正确
真的
真的
假的
假的
假的
假的
假的

在使用 ThreadPoolExecutor 时会打印此信息（2 个正在运行的调用；预期）：

正确
真的
假的
假的
假的
假的
假的
假的

Answer 1

好吧，我不会太相信这种running() 方法。似乎并不能真正反映实际的运行状态。

确保进程状态的最佳方法是让它们打印/更新某些内容。我选择使用multiprocessing.Manager().dict() 对象创建共享字典。

即使在多处理环境中，此进程同步对象也可以从任何进程安全地查阅/更新，并且具有共享状态。

每次启动进程时，更新共享字典，以 PID 作为键，True 作为值。在退出时设置False。

import concurrent.futures
import multiprocessing
import time,os


def call(shared_dict):
    shared_dict[os.getpid()] = True
    print("start",shared_dict)
    time.sleep(10)
    shared_dict[os.getpid()] = False
    print("end",shared_dict)


if __name__ == "__main__":

    with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
        shared_dict = multiprocessing.Manager().dict()
        futures = [executor.submit(call,shared_dict) for _ in range(8)]
        time.sleep(5)
        for future in futures:
            print(future.running())

这是我得到的输出：

start {3076: True}
start {9968: True, 3076: True}
True
True
True
True
True
False
False
False
end {9968: True, 3076: False}
start {9968: True, 3076: True}
end {9968: False, 3076: True}
start {9968: True, 3076: True}
end {9968: True, 3076: False}
start {9968: True, 3076: True}
end {9968: False, 3076: True}
start {9968: True, 3076: True}
end {9968: True, 3076: False}
start {9968: True, 3076: True}
end {9968: False, 3076: True}
start {9968: True, 3076: True}
end {9968: True, 3076: False}
end {9968: False, 3076: False}

如您所见，我有 5 个正在运行的进程。而我的字典清楚地表明了这一点

• 同时运行的进程不超过 2 个
• 进程在开始时只创建一次，然后重复用于执行进一步的调用（毕竟这是一个池）

我们来看看极简的documentation：

running() 如果调用当前正在执行且无法取消，则返回 True。

这似乎反映了一种与取消Future 对象未来执行的可能性相关的状态（因为它尚未正确初始化/连接到通信队列并且仍然是时候取消它）而不是实际的“进程本身的运行状态。

这可能就是source code 中这条评论在set_running_or_notify_cancel 定义下的含义：

将未来标记为正在运行或处理任何取消通知。

如果未来已被取消（cancel() 被调用并返回 True），那么任何等待未来完成的线程（尽管调用 as_completed() 或 wait()）都会收到通知并返回 False。

如果未取消未来，则将其置于运行状态（未来对 running() 的调用将返回 True）并返回 True。

我们再次了解到，最好让子流程进行协作，公布他们的状态，而不是试图使用记录不明确的方法来敲诈它。