python - 使用启动方法“spawn”的 Python 多处理不起作用
问题描述
我编写了一个 Python 类来并行绘制 pylot。它在默认启动方法是 fork 的 Linux 上运行良好,但是当我在 Windows 上尝试它时遇到了问题(可以使用 spawn 启动方法在 Linux 上重现 - 请参见下面的代码)。我总是最终得到这个错误:
Traceback (most recent call last):
File "test.py", line 50, in <module>
test()
File "test.py", line 7, in test
asyncPlotter.saveLinePlotVec3("test")
File "test.py", line 41, in saveLinePlotVec3
args=(test, ))
File "test.py", line 34, in process
p.start()
File "C:\Users\adrian\AppData\Local\Programs\Python\Python37\lib\multiprocessing\process.py", line 112, in start
self._popen = self._Popen(self)
File "C:\Users\adrian\AppData\Local\Programs\Python\Python37\lib\multiprocessing\context.py", line 223, in _Popen
return _default_context.get_context().Process._Popen(process_obj)
File "C:\Users\adrian\AppData\Local\Programs\Python\Python37\lib\multiprocessing\context.py", line 322, in _Popen
return Popen(process_obj)
File "C:\Users\adrian\AppData\Local\Programs\Python\Python37\lib\multiprocessing\popen_spawn_win32.py", line 89, in __init__
reduction.dump(process_obj, to_child)
File "C:\Users\adrian\AppData\Local\Programs\Python\Python37\lib\multiprocessing\reduction.py", line 60, in dump
ForkingPickler(file, protocol).dump(obj)
TypeError: can't pickle weakref objects
C:\Python\MonteCarloTools>Traceback (most recent call last):
File "<string>", line 1, in <module>
File "C:\Users\adrian\AppData\Local\Programs\Python\Python37\lib\multiprocessing\spawn.py", line 99, in spawn_main
new_handle = reduction.steal_handle(parent_pid, pipe_handle)
File "C:\Users\adrian\AppData\Local\Programs\Python\Python37\lib\multiprocessing\reduction.py", line 82, in steal_handle
_winapi.PROCESS_DUP_HANDLE, False, source_pid)
OSError: [WinError 87] The parameter is incorrect
我希望有一种方法可以使此代码适用于 Windows。这里是 Linux 和 Windows 上可用的不同启动方法的链接:https ://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods
import multiprocessing as mp
def test():
manager = mp.Manager()
asyncPlotter = AsyncPlotter(manager.Value('i', 0))
asyncPlotter.saveLinePlotVec3("test")
asyncPlotter.saveLinePlotVec3("test")
asyncPlotter.join()
class AsyncPlotter():
def __init__(self, nc, processes=mp.cpu_count()):
self.nc = nc
self.pids = []
self.processes = processes
def linePlotVec3(self, nc, processes, test):
self.waitOnPool(nc, processes)
print(test)
nc.value -= 1
def waitOnPool(self, nc, processes):
while nc.value >= processes:
time.sleep(0.1)
nc.value += 1
def process(self, target, args):
ctx = mp.get_context('spawn')
p = ctx.Process(target=target, args=args)
p.start()
self.pids.append(p)
def saveLinePlotVec3(self, test):
self.process(target=self.linePlotVec3,
args=(self.nc, self.processes, test))
def join(self):
for p in self.pids:
p.join()
if __name__=='__main__':
test()
解决方案
使用spawn
start 方法时,Process
对象本身被腌制以供子进程使用。在您的代码中,target=target
参数是AsyncPlotter
. 看起来整个asyncPlotter
实例也必须被腌制才能工作,其中包括self.manager
显然不想被腌制。
简而言之,保持Manager
在AsyncPlotter
. 这适用于我的 macOS 系统:
def test():
manager = mp.Manager()
asyncPlotter = AsyncPlotter(manager.Value('i', 0))
...
此外,正如您在评论中指出的那样,asyncPlotter
重用时不起作用。我不知道细节,但看起来它与如何Value
跨进程共享对象有关。该test
功能需要像:
def test():
manager = mp.Manager()
nc = manager.Value('i', 0)
asyncPlotter1 = AsyncPlotter(nc)
asyncPlotter1.saveLinePlotVec3("test 1")
asyncPlotter2 = AsyncPlotter(nc)
asyncPlotter2.saveLinePlotVec3("test 2")
asyncPlotter1.join()
asyncPlotter2.join()
总而言之,您可能希望重组代码并使用进程池。它已经处理了AsyncPlotter
正在做的事情cpu_count
和并行执行:
from multiprocessing import Pool, set_start_method
from random import random
import time
def linePlotVec3(test):
time.sleep(random())
print("test", test)
if __name__ == "__main__":
set_start_method("spawn")
with Pool() as pool:
pool.map(linePlotVec3, range(20))
或者您可以使用 aProcessPoolExecutor
来做几乎相同的事情。此示例一次启动一个任务,而不是映射到列表:
from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp
import time
from random import random
def work(i):
r = random()
print("work", i, r)
time.sleep(r)
def main():
ctx = mp.get_context("spawn")
with ProcessPoolExecutor(mp_context=ctx) as pool:
for i in range(20):
pool.submit(work, i)
if __name__ == "__main__":
main()
推荐阅读
- php - 基本匿名用户评级系统 - 不工作
- asp.net-core-mvc - Blazor Hub 未跨整个 MVC 应用程序连接
- sql-server - 将 DTS 变量传递给带有 DTEXEC 的 SSIS 2008 包
- asp.net-core - 在 .Net core 3 和 blazor 中捆绑 css 和 js
- javascript - RXJS RetryWhen - 在尝试重试时继续下游错误
- javascript - 循环遍历对象的嵌套数组并用动态值替换空属性 - Javascript
- git - Git钩子在更新存储库后自动重命名文件
- angular - 使用 Azure AD 验证 Angular 8 客户端
- c# - 使用 AutoIT 进程后,如何将 Selenium Webdriver C# 重新聚焦到窗口?
- azure - 如何在 CosmosDB 中查询嵌套对象值