Dav*_*e J 1 python python-3.x concurrent.futures python-multiprocessing
你好呀,
我偶然发现了一个问题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()
Run Code Online (Sandbox Code Playgroud)
[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!
Run Code Online (Sandbox Code Playgroud)
max_workers=1由于进程被重复使用,因此也可以使用 重现此行为。启动方法对错误没有影响(尽管"fork"
似乎只使用多个进程)。
总结一下:我希望进程中的每次新运行都包含所有以前的数据,但没有来自任何其他运行的新数据。那可能吗?我将如何实现它?为什么上面的方法没有完全做到这一点?
我很感激任何帮助。
我发现multiprocessing.pool.Pool可以在哪里设置maxtasksperchild=1,以便工作进程在其任务完成时被销毁。但我不喜欢这个
multiprocessing界面;使用起来更ProcessPoolExecutor舒服。此外,池的整体想法是节省进程设置时间,当每次运行后终止托管进程时,该时间将被忽略。
python 中的全新进程不共享内存状态。但是ProcessPoolExecutor重用流程实例。毕竟它是一个活动进程池。我认为这样做是为了防止一直弯腰和启动进程的操作系统开销。
您会在其他分发技术(例如 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)
Run Code Online (Sandbox Code Playgroud)
| 归档时间: |
|
| 查看次数: |
1491 次 |
| 最近记录: |