问题

我尝试将 a ProcessPoolExecutor 与递归调用一起使用，但它不起作用。我在下面创建了一个最小的示例

from concurrent.futures import ProcessPoolExecutor
from time import sleep

executor = ProcessPoolExecutor()
i = 3

def test():
    global i
    print("top")
    if i == 0:
        print("bar")
        return
    else:
        print("foo")
        i -= 1
        p = executor.submit(test)
        print(p.result())
        print(i)

test()

我期望发生的是 i 递减，在进程池中进行递归调用，然后我们等待结果。这种情况会递归发生，直到 i 达到 0，此时调用会级联回来。然而，这并没有发生。我看到以下输出

top
foo
top
foo

此时它挂起

可能的问题

竞争条件

我认为这可能是一个竞争条件并且 i 每次都不会减少，但这不应该这很重要，因为该函数应该被调用多少次

全局状态

我通过使递归调用以某种随机概率发生来消除对全局状态的依赖。两次调用后它仍然挂起。

编辑：我使用的是 Ubuntu 23

在理解 ProcessPoolExecutor 在 Python 中的工作方式时遇到了常见问题。问题不在于竞争条件或直接与全局状态相关，而在于 进程彼此隔离 。

当使用 ProcessPoolExecutor 时，是在创建进程池。每个进程都有自己独立的内存空间。这意味着在一个进程中对全局变量 i 所做的任何更改在其他进程中都不可见。

让我们分解一下代码中发生的事情：

1. 初始调用： 从 test() 调用开始， i 初始化为 3。
2. 第一次递归： test() 被调用，打印 "top" 和 "foo"， i 在 当前进程的内存中 递减为 2，并使用 executor.submit(test) 提交一个新任务给进程池。
3. 第二个进程： 进程池启动一个新进程来处理 executor.submit(test) 。 这个新进程有它自己的 i 副本，它从 3 开始，而不是 2 。它打印 "top" 和 "foo"，并且它版本的 i 递减为 2。然后它提交另一个任务给进程池。
4. 挂起： 此时，有两个进程都在等待 p.result() 。但是，这两个进程提交的任务都依赖于 i 达到 0。由于每个进程都有自己的 i 副本，并且它们都在递减自己的本地副本，因此该条件永远不会满足，导致挂起。

解决方案：

为了解决这个问题，需要一种在进程之间共享 i 的值的方法。可以通过多种方式实现此目的：

• 共享内存（ multiprocessing.Value 或 multiprocessing.Array ）： 这些对象允许在进程之间共享内存中的值或数组。
• 队列（ multiprocessing.Queue ）： 队列提供了一种在进程之间安全地传递消息（因此是数据）的方式。可以使用队列在工作进程和主进程之间传递 i 的更新值。
• 其他 IPC 机制： Python 的 multiprocessing 模块提供了其他进程间通信 (IPC) 机制，例如管道和管理器，这些机制可用于更复杂的数据共享场景。

示例（使用 multiprocessing.Value ）：

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Value
from time import sleep

i = Value('i', 3)  # 使用共享内存创建 'i'

def test():
    global i
    print("top")
    if i.value == 0:
        print("bar")
        return
    else:
        print("foo")
        with i.get_lock():  # 保护对共享变量的访问
            i.value -= 1
        p = executor.submit(test)
        print(p.result())
        print(i.value)

if __name__ == '__main__':  # 保护入口点
    with ProcessPoolExecutor() as executor:
        test()

在这个版本中，我们使用 multiprocessing.Value 创建了一个共享的 i 。 with i.get_lock(): 块确保对 i 的递减操作是原子的，防止了竞争条件。

请记住，多处理引入了额外的复杂性。在处理进程池和共享数据时，必须谨慎管理同步和通信以避免错误并确保正确性。