嗨,我正在尝试使用多处理来加速我的代码。但是,apply_async对我不起作用。我尝试做一个简单的例子,例如:
from multiprocessing.pool import Pool
t = [0, 1, 2, 3, 4, 5]
def cube(x):
t[x] = x**3
pool = Pool(processes=4)
for i in range(6):
pool.apply_async(cube, args=(i, ))
for x in t:
print(x)
它并没有像我期望的那样真正改变t
。
我的真实代码是这样的:
from multiprocessing.pool import Pool
def func(a, b, c, d):
#some calculations
#save result to files
#no return value
lt = #list of possible value of a
#set values to b, c, d
p = Pool()
for i in lt:
p.apply_async(func, args=(i, b, c, d, ))
这里的问题在哪里?
谢谢!
更新:感谢评论和答案,现在我明白为什么我的简单示例不起作用了。但是,我的真实代码仍然有问题。我已经检查了我的func
不依赖于任何全局变量,因此它似乎与我的示例代码不是同一个问题。
按照建议,我在func
中添加了一个返回值,现在我的代码是:
f = Flux("reactor")
d = Detector("Ge")
mv = arange(-6, 1.5, 0.5)
p = Pool()
lt = ["uee", "dee"]
for i in lt:
re = p.apply_async(res, args=(i, d, f, mv, ))
print(re.get())
p.close()
p.join()
现在我收到以下错误:
Traceback (most recent call last):
File "/Users/Shu/Documents/Programming/Python/Research/debug.py", line 35, in <module>
print(re.get())
File "/usr/local/Cellar/python3/3.6.0/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/pool.py", line 608, in get
raise self._value
File "/usr/local/Cellar/python3/3.6.0/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/pool.py", line 385, in _handle_tasks
put(task)
File "/usr/local/Cellar/python3/3.6.0/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/connection.py", line 206, in send
self._send_bytes(_ForkingPickler.dumps(obj))
File "/usr/local/Cellar/python3/3.6.0/Frameworks/Python.framework/Versions/3.6/lib/python3.6/multiprocessing/reduction.py", line 51, in dumps
cls(buf, protocol).dump(obj)
AttributeError: Can't pickle local object 'Flux.__init__.<locals>.<lambda>'
编辑:您提供的第一个示例不起作用,原因很简单:进程不共享内存。因此,更改t[x] = x**3
不会应用于父进程,t
列表的值保持不变。
您需要实际从计算中返回值,并从中构建一个新列表。
def cube(x):
return x**3
t = [0, 1, 2, 3, 4, 5]
p = Pool()
t = p.map(cube, t)
print(t)
如果,正如你在第二个例子中声称的那样,结果不应该被返回,而是独立地存储在文件中,而这并没有发生,我建议检查你的函数的返回值,看看函数本身是否引发了异常。
我建议您获得实际结果,看看会发生什么:
p = Pool()
for i in lt:
res = p.apply_async(func, args=(i, b, c, d, ))
print(res.get()) # this will raise an exception if it happens within func
p.close() # do not accept any more tasks
p.join() # wait for the completion of all scheduled jobs
函数退出太早,请尝试在脚本末尾添加以下代码:
import time
time.sleep(3)