并行 Python：将另一个模块中编写的函数传递给“提交”

Question

我正在使用 Parallel Python 模块 (pp)，并希望将作业提交给工作人员。但是，我要执行的函数在另一个模块中（用 Cython 编写），我不知道如何将函数名称导入新的 worker。建议的方法here，即在函数内部导入模块“walkerc”无法正常工作，因为walk本身是在walkerc中定义的，文件名为“walkerc.so”

import pp
from walkerc import walk
# Other stuff here
ser = pp.Server()
# Some more definitions
ser.submit(walk, (it, params))
ser.submit(walk, (1000, params), modules = ("walkerc",), globals = globals())

上面的两个语句都失败了，我得到以下错误：

Traceback（最近一次调用最后一次）：

文件“”，第 1 行，在 ser.submit(walk, (1000, params), modules = ("walkerc",), globals = globals())

文件“/usr/lib/python2.7/site-packages/pp.py”，第 458 行，在提交中 sfunc = self.__dumpsfunc((func, ) + depfuncs, 模块)

文件“/usr/lib/python2.7/site-packages/pp.py”，第 629 行，在 __dumpsfunc sources = [self.__get_source(func) for func in funcs]

文件“/usr/lib/python2.7/site-packages/pp.py”，第 696 行，在 __get_source sourcelines = inspect.getsourcelines(func)[0]

文件“/usr/lib/python2.7/inspect.py”，第 690 行，在 getsourcelines 中 行，lnum = findsource(object)

文件“/usr/lib/python2.7/inspect.py”，第 526 行，在 findsource 中 文件 = 获取文件（对象）

文件“/usr/lib/python2.7/inspect.py”，第 420 行，在 getfile '函数、回溯、框架或代码对象'.format(object))

TypeError: '' is not a module, class, method, 函数、回溯、框架或代码对象

函数 'walk' 本身在​​主程序中被正确导入，问题在于将其提交给新工作人员的过程。

如何正确指定函数名称“walk”？ 我不想在我调用它的同一个文件中定义“walk”，因为我已经在 Cython 中对其进行了修改并希望获得更好的性能。有其他选择吗？

Answer 1

尝试将您的 walk 函数重命名为其他名称，例如 mywalk。正如异常文本所暗示的，您的环境似乎有一个名为 walk 的内置函数，因此 inspect 模块会混淆。

我可以像这样在我的系统上成功传递我导入的walk 函数，这里没有冲突，也不需要更多，函数使用给定的参数执行：

import pp
from walkerc import walk

pps = pp.Server()
pps.submit(walk, args=(1,))

但是传递dir，这肯定是一个内置函数：

pps.submit(dir)

我得到了和你一样的错误：

回溯（最近一次通话最后）： 文件“parallel.py”，第 9 行，在 pps.submit(目录) ... 文件“.../lib/python2.7/inspect.py”，第 420 行，在 getfile '函数、回溯、框架或代码对象'.format(object)) TypeError：不是模块、类、方法、函数、回溯、框架或代码对象

以下讨论后更新：

所以这里的问题是 Python 将来自 C extensions 的成员视为内置。上面的代码适用于常规 Python 模块，但在从 C 扩展导入和传递函数时，我能够复制 OP 的错误。

因此，我将 C 扩展函数调用包装在一个普通的 Python 函数中，这样就可以了。请注意，现在 walk 函数导入已移至包装函数，因此它可以在调度时自己构造自己的上下文。

import pp

def walk(n):
    import walkerc
    return walkerc.walk(n)

def print_callback(result):
    print('callback: ', result)

pps = pp.Server()
job = pps.submit(walk, args=(1,), callback=print_callback)