Python - 动态使用multiprocessing模块

Question

我正在尝试开发一个包装器，用来动态使用 multiprocessing 模块。我有很多函数分散在不同的模块里，需要好好管理这些函数。我需要能够把来自任何模块的函数和它的参数传递给我的包装器。因为这些数据和函数在运行时才会知道，所以它们是依赖用户的。

下面是我想要实现的一个例子：

import sys
from multiprocessing import Process, Queue, cpu_count

def dyn():
    pass

class mp():
    def __init__(self, data, func, n_procs = cpu_count()):
        self.data    = data
        self.q       = Queue()
        self.n_procs = n_procs

        # Replace module-level 'dyn' function with the provided function
        setattr(sys.modules[__name__], 'dyn', func)
        # Calling dyn(...) at this point will produce the same output as
        # calling func(...)

    def _worker(self, *items):
        data = []
        for item in items:
            data.append(dyn(item))
        self.q.put(data)

    def compute(self):
        for item in self.data:
            Process(target=getattr(self, '_worker'), args=item).start()

    def items(self):
        queue_count = self.n_procs
        while queue_count > 0:
            queue_count -= 1
            yield self.q.get()

if __name__ == '__main__':  
    def work(x):
        return x ** 2

    # Create workers
    data = [range(10)] * cpu_count()
    workers = mp(data, work)

    # Make the workers work
    workers.compute()

    # Get the data from the workers
    workers_data = []
    for item in workers.items():
        workers_data.append(item)
    print workers_data

在这个例子中，输出应该是这个格式：

[[0, 1, 4, 9, 16, 25, 36, 49, 64, 81] * n_procs]

如果你尝试运行这段代码，你会遇到一个异常，提示传递给 dyn 的参数太多了。我认为问题在于 dyn 在这个实例中被覆盖了，但当调用 Process 时，这些更改就不再存在了。

我该如何解决这个问题呢？

注意 - 这段代码需要在Windows上运行。我使用的是Python 2.7。

更新

根据我收到的评论，我决定做一些“麻烦”的事情。下面是我工作的解决方案：

import sys, re, uuid, os
from cStringIO import StringIO
from multiprocessing import Process, Queue, cpu_count

class mp():
    def __init__(self, data, func, n_procs = cpu_count()):
        self.data    = data
        self.q       = Queue()
        self.n_procs = n_procs
        self.module  = 'm' + str(uuid.uuid1()).replace('-', '')
        self.file    = self.module + '.py'

        # Build external module
        self.__func_to_module(func)

    def __func_to_module(self, func):   
        with open(self.file, 'wb') as f:
            for line in StringIO(func):
                if 'def' in line:
                    f.write(re.sub(r'def .*\(', 'def work(', line))
                else:
                    f.write(line)

    def _worker(self, q, module, *items):
        exec('from %s import work' % module)
        data = []
        for item in items[0]:
            data.append(work(item))
        q.put(data)

    def compute(self):
        for item in self.data:
            Process(target=getattr(self, '_worker'),
                args=(self.q, self.module, item)).start()

    def items(self):
        queue_count = self.n_procs
        while queue_count > 0:
            queue_count -= 1
            yield self.q.get()
        os.remove(self.file)
        os.remove(self.file + 'c')

if __name__ == '__main__':  
    func = '''def func(x):
        return x ** 2'''

    # Create workers
    data = [range(10)] * cpu_count()
    workers = mp(data, func)

    # Make the workers work
    workers.compute()

    # Get the data from the workers
    workers_data = []
    for item in workers.items():
        workers_data.append(item)
    print workers_data

Answer 1

在Windows系统上，每当启动一个新的进程时，模块会被重新加载，这样就会丢失对“dyn”的定义。不过，你可以通过队列传递一个函数，或者通过参数传递给进程的目标函数。

def _worker(*items, func=None, q=None):  
    #note that this had better be a function not a method
    data = []
    for item in items:
        data.append(func(item))
    q.put(data)

#...
Process(target=_worker, args=item, kwargs={'func':dyn, 'q':q})