concurrent.futures.ProcessPoolExecutor 动态创建函数的限制
Restrictions on dynamically created functions with concurrent.futures.ProcessPoolExecutor
我正在尝试使用我在其他函数中动态创建的函数进行多处理。如果提供给 ProcessPoolExecutor 的函数是模块级的，我似乎可以 运行 这些：
def make_func(a):
def dynamic_func(i):
return i, i**2 + a
return dynamic_func
f_dyns = [make_func(a) for a in range(10)]
def loopfunc(i):
return f_dyns[i](i)
with concurrent.futures.ProcessPoolExecutor(3) as executor:
for i,r in executor.map(loopfunc, range(10)):
print(i,":",r)
输出：
0 : 0
1 : 2
2 : 6
3 : 12
4 : 20
5 : 30
6 : 42
7 : 56
8 : 72
9 : 90
但是，如果多处理是由 class 函数启动的，我将无法执行此操作：
class Test:
def __init__(self,myfunc):
self.f = myfunc
def loopfunc(self,i):
return self.f(i)
def run(self):
with concurrent.futures.ProcessPoolExecutor(3) as executor:
for i,r in executor.map(self.loopfunc, range(10)):
print(i,":",r)
o2 = Test(make_func(1))
o2.run()
输出：
Traceback (most recent call last):
File "/home/farmer/anaconda3/envs/general/lib/python3.6/multiprocessing/queues.py", line 234, in _feed
obj = _ForkingPickler.dumps(obj)
File "/home/farmer/anaconda3/envs/general/lib/python3.6/multiprocessing/reduction.py", line 51, in dumps
cls(buf, protocol).dump(obj)
AttributeError: Can't pickle local object 'make_func.<locals>.dynamic_func'
另一方面，如果我不在其中使用动态生成的函数，我可以 运行 在 class 函数上进行多处理。有什么办法解决这个问题吗？我尝试将动态生成的函数添加到 'globals' 字典中，但这似乎没有帮助：
def make_func_glob(a):
def dynamic_func(i):
return i, i**2 + a
globals()['my_func_{0}'.format(a)] = dynamic_func
make_func_glob(1)
print("test:", my_func_1(3))
o3 = Test(my_func_1)
o3.run()
输出：
test: (3, 10)
Traceback (most recent call last):
File "/home/farmer/anaconda3/envs/general/lib/python3.6/multiprocessing/queues.py", line 234, in _feed
obj = _ForkingPickler.dumps(obj)
File "/home/farmer/anaconda3/envs/general/lib/python3.6/multiprocessing/reduction.py", line 51, in dumps
cls(buf, protocol).dump(obj)
AttributeError: Can't pickle local object 'make_func_glob.<locals>.dynamic_func'
所以 python 仍然认为它是本地对象，即使我将它添加到 'globals' dict。像这样 'globals' 想法就可以了，我不需要任何花哨的东西。为了方便起见，我只是动态创建这些功能。如果它们成为全局对象，我会非常高兴。它们将始终由模块定义，其中只有一堆具有几乎相同的定义，因此以编程方式定义它们比手动将它们全部写出来更方便。所以我认为有可能以某种方式让 python 将它们识别为 "true" 函数，就像我通过 'exec' 定义它们一样。或者至少足够接近以至于我可以在我的并行代码中使用它们。
正如错误消息所暗示的那样，它更多地与酸洗有关，而不是动态生成的函数。来自 https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ProcessPoolExecutor
only picklable objects can be executed and returned.
从 https://docs.python.org/3/library/pickle.html#what-can-be-pickled-and-unpickled 开始，可以 pickle 的函数种类：
functions defined at the top level of a module (using def, not lambda)
这表明其他类型的函数不能被腌制。从问题的代码中跳出一个不符合此要求的函数：dynamic_func from...
def make_func(a):
def dynamic_func(i):
return i, i**2 + a
return dynamic_func
...你暗示这就是问题所在....
So I would have thought it was possible to somehow get python to recognise them as "true" functions
可以！您可以将 dynamic_func 放在顶层，并使用 partial 而不是闭包...
from functools import partial
def dynamic_func(a, i):
return i, i**2 + a
def make_func(a):
return partial(dynamic_func, a)
所以完整...
import concurrent.futures
from functools import partial
def dynamic_func(a, i):
return i, i**2 + a
def make_func(a):
return partial(dynamic_func, a)
f_dyns = [make_func(a) for a in range(10)]
def loopfunc(i):
return f_dyns[i](i)
class Test:
def __init__(self, myfunc):
self.f = myfunc
def loopfunc(self, i):
return self.f(i)
def run(self):
with concurrent.futures.ProcessPoolExecutor(3) as executor:
for i,r in executor.map(self.loopfunc, range(10)):
print(i,":",r)
o2 = Test(make_func(1))
o2.run()
但是...为什么没有 类 的原始形式有效，我不知道。根据我的理解，它会尝试 pickle 一个非顶级函数，所以我认为我的理解是有缺陷的。
答案不能解决所有情况。例如：
def _picklable_func(func, *args, **kwargs):
myfunc = partial(func, *args)
return myfunc
def invoke_func():
mod_fun_str = "oh.mymodule.myfunc"
mod_name, func_name = mod_fun_str.rsplit('.', 1)
mod = importlib.import_module(mod_name)
func = getattr(mod, func_name)
future = self.threadpool.submit(func, *args, **kwargs) # feature.result() always return None
#future = self.threadpool.submit(_picklable_func(func, *args, **kwargs), **kwargs) # this doesn't work too, 'myfunc' always returned None
if __name__=="__main__":
invoke_func()
问题是函数 invoke_func 中的 future.result() 仍然总是返回 None。
# myfunc in oh.mymodule module is
def myfunc():
return "hello"
我正在尝试使用我在其他函数中动态创建的函数进行多处理。如果提供给 ProcessPoolExecutor 的函数是模块级的，我似乎可以 运行 这些：
def make_func(a):
def dynamic_func(i):
return i, i**2 + a
return dynamic_func
f_dyns = [make_func(a) for a in range(10)]
def loopfunc(i):
return f_dyns[i](i)
with concurrent.futures.ProcessPoolExecutor(3) as executor:
for i,r in executor.map(loopfunc, range(10)):
print(i,":",r)
输出：
0 : 0
1 : 2
2 : 6
3 : 12
4 : 20
5 : 30
6 : 42
7 : 56
8 : 72
9 : 90
但是，如果多处理是由 class 函数启动的，我将无法执行此操作：
class Test:
def __init__(self,myfunc):
self.f = myfunc
def loopfunc(self,i):
return self.f(i)
def run(self):
with concurrent.futures.ProcessPoolExecutor(3) as executor:
for i,r in executor.map(self.loopfunc, range(10)):
print(i,":",r)
o2 = Test(make_func(1))
o2.run()
输出：
Traceback (most recent call last):
File "/home/farmer/anaconda3/envs/general/lib/python3.6/multiprocessing/queues.py", line 234, in _feed
obj = _ForkingPickler.dumps(obj)
File "/home/farmer/anaconda3/envs/general/lib/python3.6/multiprocessing/reduction.py", line 51, in dumps
cls(buf, protocol).dump(obj)
AttributeError: Can't pickle local object 'make_func.<locals>.dynamic_func'
另一方面，如果我不在其中使用动态生成的函数，我可以 运行 在 class 函数上进行多处理。有什么办法解决这个问题吗？我尝试将动态生成的函数添加到 'globals' 字典中，但这似乎没有帮助：
def make_func_glob(a):
def dynamic_func(i):
return i, i**2 + a
globals()['my_func_{0}'.format(a)] = dynamic_func
make_func_glob(1)
print("test:", my_func_1(3))
o3 = Test(my_func_1)
o3.run()
输出：
test: (3, 10)
Traceback (most recent call last):
File "/home/farmer/anaconda3/envs/general/lib/python3.6/multiprocessing/queues.py", line 234, in _feed
obj = _ForkingPickler.dumps(obj)
File "/home/farmer/anaconda3/envs/general/lib/python3.6/multiprocessing/reduction.py", line 51, in dumps
cls(buf, protocol).dump(obj)
AttributeError: Can't pickle local object 'make_func_glob.<locals>.dynamic_func'
所以 python 仍然认为它是本地对象，即使我将它添加到 'globals' dict。像这样 'globals' 想法就可以了，我不需要任何花哨的东西。为了方便起见，我只是动态创建这些功能。如果它们成为全局对象，我会非常高兴。它们将始终由模块定义，其中只有一堆具有几乎相同的定义，因此以编程方式定义它们比手动将它们全部写出来更方便。所以我认为有可能以某种方式让 python 将它们识别为 "true" 函数，就像我通过 'exec' 定义它们一样。或者至少足够接近以至于我可以在我的并行代码中使用它们。
正如错误消息所暗示的那样，它更多地与酸洗有关，而不是动态生成的函数。来自 https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ProcessPoolExecutor
only picklable objects can be executed and returned.
从 https://docs.python.org/3/library/pickle.html#what-can-be-pickled-and-unpickled 开始，可以 pickle 的函数种类：
functions defined at the top level of a module (using def, not lambda)
这表明其他类型的函数不能被腌制。从问题的代码中跳出一个不符合此要求的函数：dynamic_func from...
def make_func(a):
def dynamic_func(i):
return i, i**2 + a
return dynamic_func
...你暗示这就是问题所在....
So I would have thought it was possible to somehow get python to recognise them as "true" functions
可以！您可以将 dynamic_func 放在顶层，并使用 partial 而不是闭包...
from functools import partial
def dynamic_func(a, i):
return i, i**2 + a
def make_func(a):
return partial(dynamic_func, a)
所以完整...
import concurrent.futures
from functools import partial
def dynamic_func(a, i):
return i, i**2 + a
def make_func(a):
return partial(dynamic_func, a)
f_dyns = [make_func(a) for a in range(10)]
def loopfunc(i):
return f_dyns[i](i)
class Test:
def __init__(self, myfunc):
self.f = myfunc
def loopfunc(self, i):
return self.f(i)
def run(self):
with concurrent.futures.ProcessPoolExecutor(3) as executor:
for i,r in executor.map(self.loopfunc, range(10)):
print(i,":",r)
o2 = Test(make_func(1))
o2.run()
但是...为什么没有 类 的原始形式有效，我不知道。根据我的理解，它会尝试 pickle 一个非顶级函数，所以我认为我的理解是有缺陷的。
答案不能解决所有情况。例如：
def _picklable_func(func, *args, **kwargs):
myfunc = partial(func, *args)
return myfunc
def invoke_func():
mod_fun_str = "oh.mymodule.myfunc"
mod_name, func_name = mod_fun_str.rsplit('.', 1)
mod = importlib.import_module(mod_name)
func = getattr(mod, func_name)
future = self.threadpool.submit(func, *args, **kwargs) # feature.result() always return None
#future = self.threadpool.submit(_picklable_func(func, *args, **kwargs), **kwargs) # this doesn't work too, 'myfunc' always returned None
if __name__=="__main__":
invoke_func()
问题是函数 invoke_func 中的 future.result() 仍然总是返回 None。
# myfunc in oh.mymodule module is
def myfunc():
return "hello"