2

我最近为我的研究编写了一个带有类的程序，并且我试图将它并行化。当我将 Python 2.7 的 multiprocessing.Process 与 JoinableQueue 和托管数据一起使用时，我的程序最终会因已失效的进程而挂起。

import multiprocessing as mp
import traceback

class Paramfit(object):
    def __init__(self):
        pass

    def _calc_bond(self, index):
        # Calculate data

    def _calc_parallel(self, index):
        self._calc_bond(index)

    def run(self):
        for ts, force in itertools.izip(self.coortrj, self.forcevec):
        try:
            consumers = [mp.Process(target=self._calc_parallel,
                         args=(force,)) for i in range(nprocs)]
            for w in consumers:
                w.start()

            # Enqueue jobs
            for i in range(self.totalsites):
                self.tasks.put(i)

            # Add a poison pill for each consumer
            for i in range(nprocs):
                self.tasks.put(None)

            self.tasks.close()
            self.tasks.join()

    #       for w in consumers:
    #           w.join()
        except:
            traceback.print_exc()

_calc_parallel 调用其他一些类方法。

我什至尝试使用 multiprocessing.Pool 为此目的使用http://bytes.com/topic/python/answers/552476-why-cant-you-pickle-instancemethods上其他地方的 copy_reg 选项。

import multiprocessing as mp
import traceback

class Paramfit(object):
    def __init__(self):
        pass

    def _calc_bond(self, index):
        # Calculate data

    def _use_force(force):
        # Calculate data

    def _calc_parallel(self, index, force):
        self._calc_bond(index)
        self._use_force(force)

    def run(self):
        try:
            pool = mp.Pool(processes=nprocs, maxtasksperchild=2)
            args = itertools.izip(range(self.totalsites), itertools.repeat(force))
            pool.map_async(self._calc_parallel, args)
            pool.close()
            pool.join()
        except:
            traceback.print_exc()

但是，pool.map_async 似乎没有调用 self._calc_parallel。我知道在这两种情况下（进程和池），我都忽略了一些东西，但我不太清楚是什么。我通常处理超过 40,000 个元素。

谢谢您的帮助。

更新

在阅读了其他几篇文章后，我也尝试了 pathos.multiprocessing。

import pathos.multiprocessing as mp
class Paramfit(object):
    def __init__(self):
        pass

    def _calc_bond(self, index):
        # Calculate data

    def _use_force(force):
        # Calculate data

    def _calc_parallel(self, index, force):
        self._calc_bond(index)
        self._use_force(force)

    def run(self):
        try:
            pool = mp.ProcessingPool(nprocs)
            args = itertools.izip(range(self.totalsites), itertools.repeat(force))
            pool.amap(lambda x: self._calc_parallel(*x), args)
        except:
            traceback.print_exc()

而且，与我之前的尝试一样，这似乎也可以在不调用该方法的情况下快速完成。

更新 2

我决定修改代码，将我的庞然大物类拆分成更小、更易于管理的组件。但是，如果我使用 pathos.multiprocessing，我会遇到与之前发布的不同的情况（请参阅链接）。我的新代码现在有一个可用于计算的对象，然后通过其方法返回一个值。

import itertools
import pandas as pd
import pathos.multiprocessing as mp

class ForceData(object):
    def __init__(self, *args, **kwargs):
        # Setup data
        self.value = pd.DataFrame()
    def calculateBondData(self, index):
        # Calculation
        return self.value
    def calculateNonBondedData(self, index):
        # Calculation
        return self.value
    def calculateAll(self, index):
        # Because self.value is a pandas.DataFrame, changed internally
        self.calculateBondData(index)
        self.calculateNonBondedData(index)
        return self.value

class ForceMatrix(object):
    def __init__(self, *args, **kwargs):
        # Initialize data
        self._matrix = pd.DataFrame()
    def map(self, data):
        for value in data.get():
            for i, j in itertools.product(value.index, repeat=2):
                self._matrix.loc[[i], [j]] += value.values

def calculate(self, *args, **kwargs):
    # Setup initial information.
    fd = ForceData()
    matrix = ForceMatrix()
    pool = mp.ProcessingPool()
    data = pool.amap(fd.calculateAll, range(x))
    matrix.map(data, force)
    return matrix

我认为这是一个单独的功能func(dataobj, force)，但这似乎也无济于事。以目前的速度，我估计单个处理器上的完整计算需要 1000 多个小时，这对于应该更快的东西来说太长了。

更新 3 (4/30/15)

由于@MikeMcKerns 有用的见解，我可能已经确定了一个可能的解决方案。在 iMac（四核）或集群的 16 核节点上，我发现对于没有键的粗粒度（CG）系统，双倍itertools.imap似乎是我最好的解决方案（1000 个 CG 站点）时钟每个轨迹帧大约 5.2 秒。当我进入一个包含一些键细节的系统（3000 个代表水的 CG 站点）时，我发现在 iMac 上（使用 1 个核心），itertools.imap然后是pathos.ThreadingPool.uimap（4 个线程）以大约 85 秒/帧的速度运行；如果我按照@MikeMcKerns 的评论中的建议尝试进程池（4 核 x 2）/线程池（4 线程），计算时间增加了 2.5 倍。在 16 核集群（32 pp/16 tp）上，这个 CG 系统运行速度也很慢（大约 160 秒/帧）。在 iMac（1 核/4 线程）上具有 42,778 个站点和众多键的 CG 系统可能会以大约 58 分钟/帧的速度运行。我还没有在集群的 16 核节点上测试这个大型系统，但我不确定是否使用进程池/线程池来进一步加快速度。

例子：

# For a CG system with no bond details
for i in range(nframes):
    data1 = itertools.imap(func1, range(nsites))
    data2 = itertools.imap(func2, data1)
    for values in data2:
        func3(values)

# For a system with bond details
import pathos.multiprocessing as mp

tpool = mp.ThreadingPool(mp.cpu_count())
for i in range(nframes):
    data1 = itertools.imap(func1, range(nsites))
    data2 = tpool.uimap(func2, data1)
    for values in data2:
        func3(values)

# Seems to be the slowest in the bunch on iMac and possibly on 16-cores of a node.
ppool = mp.ProcessingPool(mp.cpu_count() * 2)
tpool = mp.ThreadingPool(mp.cpu_count())
for i in range(nframes):
    data1 = ppool.uimap(func1, range(nsites))
    data2 = tpool.uimap(func2, data1)
    for values in data2:
        func3(values)

我怀疑系统越大，我从多处理中获得的好处就越多。我知道大型 CG 系统（42,778 个站点）大约需要 0.08 秒/站点，而 0.02 秒/站点（3000 个 CG 站点）或 0.05 秒/站点（1000 个站点无键）。

在我努力减少计算时间的过程中，我发现了可以减少一些计算的区域（例如，global变量和算法更改），但是如果我可以通过全面的多线程处理进一步减少它，那就太好了。

4

1 回答 1

2

如果您使用的是 python 2.7，您的选择相当有限

这是一个使用池参数调用对象方法的简短示例。

第一个问题是只有在模块顶层定义的函数才能被pickle。基于 Unix 的系统有办法绕过这个限制，但不应该依赖这个。因此，您必须定义一个函数，该函数接受您想要的对象和调用相关方法所需的参数。

例如：

def do_square(args):
    squarer, x = args # unpack args
    return squarer.square(x)

该类Squarer可能如下所示：

class Squarer(object):
    def square(self, x):
        return x * x

现在并行应用平方函数。

if __name__ == "__main__":
    # all pool work must be done inside the if statement otherwise a recursive 
    # cycle of Pool spawning will be created.

    pool = Pool()
    sq = Squarer()
    # create args as a sequence of tuples
    args = [(sq, i) for i in range(10)]
    print pool.map(do_square, args)

    pool.close()
    pool.join()

请注意，平方器不是有状态的。它接收数据、处理数据并返回结果。这是因为孩子的状态与父母的状态是分开的。除非使用多处理提供的队列、管道或其他共享状态类，否则其中一个的更改不会反映在另一个中。一般来说，最好从子进程返回计算数据，然后在父进程中处理该数据，而不是尝试将数据存储在子进程可以访问的某个共享变量中。

有状态平方器不与多处理一起使用的示例：

class StatefulSquarer(object):

    def __init__(self):
        self.results = []

    def square(self, x):
        self.results.append(x * x)

if __name__ == "__main__":

    print("without pool")
    sq = StatefulSquarer()
    map(do_square, [(sq, i) for i in range(10)])
    print(sq.results)

    print("with pool")
    pool = Pool()
    sq = StatefulSquarer()
    pool.map(do_square, [(sq, i) for i in range(10)])
    print(sq.results)

    pool.close()
    pool.join()

如果您想完成这项工作，那么更好的解决方案将是：

for result in pool.map(do_square, [(sq, i) for i in range(10)]):
    sq.results.append(result)

如果你的类很大，但不可变怎么办。每次在 map 中启动新任务时，都必须将这个巨大的对象复制到子进程中。但是，您可以通过仅将其复制到子进程一次来节省时间。

from multiprocessing import Pool

def child_init(sq_):
    global sq
    sq = sq_

def do_square(x):
    return sq.square(x)

class Squarer(object):
    def square(self, x):
        return x * x

if __name__ == "__main__":
    sq = Squarer()
    pool = Pool(initializer=child_init, initargs=(sq,))

    print(pool.map(do_square, range(10)))

    pool.close()
    pool.join()
于 2015-04-23T13:33:13.490 回答