Python进程没有为重用而清理

Question

嗨，

我偶然发现了processPoolExecutor的一个问题，在这个问题上，进程应该不能访问数据。让我解释一下：

我有一个类似于下面示例的情况：我得到了几个运行，每个运行以不同的参数开始。他们并行计算他们的东西，没有理由相互交互。现在，正如我所理解的，当一个进程分叉时，它会复制自己。子进程与其父进程具有相同的（内存）数据，但是如果它更改任何内容，它将在自己的副本上进行更改。如果我想让更改在子进程的生命周期内继续存在，我会调用队列、管道和其他IPC东西。

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import current_process, set_start_method

class Static:
    integer: int = 0

def inprocess(run: int) -> None:
    cp = current_process()
    # Print current state
    print(f"[{run:2d} {cp.pid} {cp.name}] int: {Static.integer}", flush=True)

    # Check value
    if Static.integer != 0:
        raise Exception(f"[{run:2d} {cp.pid} {cp.name}] Variable already set!")

    # Update value
    Static.integer = run + 1

def pooling():
    cp = current_process()
    # Get master's pid
    print(f"[{cp.pid} {cp.name}] Start")
    with ProcessPoolExecutor(max_workers=2) as executor:
        for i, _ in enumerate(executor.map(inprocess, range(4))):
            print(f"run #{i} finished", flush=True)

if __name__ == '__main__':
    set_start_method("fork")    # enforce fork
    pooling()

[1998 MainProcess] Start
[ 0 2020 Process-1] int: 0
[ 2 2020 Process-1] int: 1
[ 1 2021 Process-2] int: 0
[ 3 2021 Process-2] int: 2
run #0 finished
run #1 finished
concurrent.futures.process._RemoteTraceback:
"""
Traceback (most recent call last):
  File "/usr/lib/python3.6/concurrent/futures/process.py", line 175, in _process_worker
    r = call_item.fn(*call_item.args, **call_item.kwargs)
  File "/usr/lib/python3.6/concurrent/futures/process.py", line 153, in _process_chunk
    return [fn(*args) for args in chunk]
  File "/usr/lib/python3.6/concurrent/futures/process.py", line 153, in <listcomp>
    return [fn(*args) for args in chunk]
  File "<stdin>", line 14, in inprocess
Exception: [ 2 2020 Process-1] Variable already set!
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "<stdin>", line 29, in <module>
  File "<stdin>", line 24, in pooling
  File "/usr/lib/python3.6/concurrent/futures/process.py", line 366, in _chain_from_iterable_of_lists
    for element in iterable:
  File "/usr/lib/python3.6/concurrent/futures/_base.py", line 586, in result_iterator
    yield fs.pop().result()
  File "/usr/lib/python3.6/concurrent/futures/_base.py", line 425, in result
    return self.__get_result()
  File "/usr/lib/python3.6/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
Exception: [ 2 2020 Process-1] Variable already set!

我很感激任何帮助。

我找到了multiprocessing.pool.pool，在这里可以设置maxtasksperchild=1，这样工作进程的任务完成后就会被销毁。但是我不喜欢multiprocessing接口；processPoolExecutor使用起来更加舒适。此外，池的全部思想是节省进程设置时间，当每次运行后终止宿主进程时，这将被取消。

Answer 1

python中的全新进程不共享内存状态。但是processPoolExecutor重用流程实例。它毕竟是一个活动进程池。我假设这样做是为了防止一直弯腰和启动进程的OS开销。

您可以在其他分发技术中看到相同的行为，比如celery，如果您不小心，您可能会在执行之间泄漏全局状态。

我建议您更好地管理命名空间以封装数据。例如，使用您的示例，您可以将代码和数据封装在父类中，并在inprocess()中实例化，而不是将其存储在共享命名空间中，比如类中的静态字段或直接存储在模块中。这样，该对象最终将由垃圾回收器清理：

class State:
    def __init__(self):
        self.integer: int = 0

    def do_stuff():
        self.integer += 42

def use_global_function(state):
    state.integer -= 1664
    state.do_stuff()

def inprocess(run: int) -> None:
    cp = current_process()
    state = State()
    print(f"[{run:2d} {cp.pid} {cp.name}] int: {state.integer}", flush=True)
    if state.integer != 0:
        raise Exception(f"[{run:2d} {cp.pid} {cp.name}] Variable already set!")
    state.integer = run + 1
    state.do_stuff()
    use_global_function(state)