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

Posted 2023-02-16

【中文标题】在 Python 中使用 ProcessPoolExecutor 的运行调用数不正确

【英文标题】：Incorrect number of running calls with ProcessPoolExecutor in Python

【发布时间】：2019-10-28 10:35:54

【问题描述】：

在 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 个正在运行的调用；预期）：

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

【问题讨论】：

可能的情况是进程启动（这需要时间）和状态之间存在竞争条件，而使用线程时速度会快得多。我的意思是：一旦第一个返回 True，它就可以再次为 False。状态快照不是原子的。

@Jean-FrançoisFabre 我尝试在两者之间使用time.sleep(3)，但没有区别。

在轮询运行状态更改数字之前使用sleep。我之前有 1 个，现在我有 5 个......

如果您在调用中添加一些打印，您会看到只有 2 个进程正在运行。运行状态可能不可靠。

@Jean-FrançoisFabre 所以你认为Future.running 方法坏了？

【参考方案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。

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

【讨论】：

感谢示例代码证明只有 2 个调用正在运行。但是对我来说，它看起来像是 Future.running 方法实现中的一个错误，它返回不一致的状态。如果您在检查未来状态（如您的示例中）之前睡眠 5 秒钟，您可以确定所有期货都处于挂起状态，除了 2 应该处于运行状态。所以Future.running 方法不应该返回 3、5 或任何不同于 2 的数字。否则为什么要首先公开这样一个无用的方法？我在 Python 错误跟踪器中提交了一个错误：bugs.python.org/issue37276。

我不认为它没用。关于错误报告的好电话。我对结果很感兴趣。代码至少可以说是不平凡的。

我发现了一个更严重的错误：***.com/questions/56609847