在python的并行进程中运行类方法答案

【问题标题】：Run a class method in parallel processes in python在python的并行进程中运行类方法
【发布时间】：2017-08-24 10:59:38
【问题描述】：

我正在尝试并未能并行运行一个巨大的循环。循环正是特定类的一种方法，在循环内部我称它为另一种方法。它确实有效，但由于某种原因，列表中只有一个进程，并且输出（参见代码）始终为“Worker 0”。要么没有创建进程，要么没有并行运行。结构如下：

main.py

from my_class.py import MyClass

def main():
    class_object = MyClass()
    class_object.method()

if __name__ == '__main__':
    main()

my_class.py

from multiprocessing import Process

MyClass(object):
    def __init__(self):
        # do something

    def _method(self, worker_num, n_workers, amount, job, data):
        for i, val in enumerate(job):
            print('Worker %d' % worker_num)
            self.another_method(val, data)

    def another_method(self):
        # do something to the data

    def method(self):
        # definitions of data and job_size go here

        n_workers = 16
        chunk = job_size // n_workers
        resid = job_size - chunk * n_workers

        workers = []
        for worker_num in range(n_workers):
            st = worker_num * chunk
            amount = chunk if worker_num != n_workers - 1 else chunk + resid
            worker = Process(target=self._method, args=[worker_num, n_workers, amount, job[st:st+amount], data])
            worker.start()
            workers.append(worker)

        for worker in workers:
            worker.join()

        return data

我已经阅读了一些关于需要主模块可导入的子进程的内容，但我不知道在我的情况下如何做到这一点。

【问题讨论】：

除了缺少job_size, job, data 我正在运行您的示例并按预期工作。输出 Worker 0 到 15。看来你在重新发明轮子，你知道Process Pools
@stovfl 是的，它确实按顺序工作（尽管它可能会弄乱顺序），但我想在多个内核上并行使用它以加快计算速度。我不太了解进程池，谢谢。
我不认为你对工人的创造是错误的。正如您所说，创建并运行了一个工人。因此，您的类方法已正确转移给工作人员。但是，我通常会在启动所有工作人员之前创建它们。所以我有一个小循环创建工人，然后一个循环启动工人。
@stovfl 首先，感谢您的帮助，我尝试打印 pid，它们实际上是不同的，但仍然只有一个内核在使用。所以问题是，我可以对 Process 对象使用多个内核还是需要使用 Pools？

标签： python python-3.x multiprocessing

【解决方案1】：

问题：...但仍然只有一个内核在使用。所以问题是，我可以对 Process 对象使用多个内核吗

这不取决于 Process 使用哪个 CPU 的 Python 解释器。
相关：on-what-cpu-cores-are-my-python-processes-running

用下面的代码扩展你的def _method(...，看看实际发生了什么：

注意：getpidcore(pid) 是 Distribution 依赖，可能FAIL！

def getpidcore(pid):
    with open('/proc/{}/stat'.format(pid), 'rb') as fh:
        core = int(fh.read().split()[-14])
        return core

class MyClass(object): 
    ...
    def _method(self, worker_num, n_workers, amount, job, data):
        for i, val in enumerate(job):
            core = getpidcore(os.getpid())
            print('core:{} pid:{} Worker({})'.format(core, os.getpid(), (worker_num, n_workers, amount, job)))

输出：

core:1 pid:7623 Worker((0, 16, 1, [1]))
core:1 pid:7625 Worker((2, 16, 1, [3]))
core:0 pid:7624 Worker((1, 16, 1, [2]))
core:1 pid:7626 Worker((3, 16, 1, [4]))
core:1 pid:7628 Worker((5, 16, 1, [6]))
core:0 pid:7627 Worker((4, 16, 1, [5]))

使用 Python 测试：Linux 上的 3.4.2

【讨论】：

我无法在我的案例中重现这种行为，并放弃了支持 mpi4py。还是谢谢。