python中数组中类存储的多处理成员函数

Question

我主要使用 C、C++ 进行编程，最近将一个项目转换为 python。除非我无法轻松地转换多处理。

在示例中，我有一个填充了一个球类的数组，该类具有一个名为 update 的成员函数，其中传入了 3 个变量。

就是这样。它存储在一个称为球的数组中。 我已经浏览了足够多的帖子文档和视频，但没有找到任何涵盖这几个方面的内容，但没有展示如何处理传入的变量。

理想情况下，我会创建一个流程拉动，让它在它们之间拆分工作。 我需要检索对象并更新原始进程空间中的对象。

不确定，但看起来可能更容易强制它输出一个元组，然后使用所有数据更新类，然后编写另一个函数来更新类。

我们非常感谢您反馈在 python 中执行此操作的最佳方法。我也欣赏性能胜过做某事的容易程度。毕竟，这就是这样做的意义所在。 提前致谢。

class Ball:
          
    def __init__(self,x,y,vx,vy,c):
        self.x=x
        self.y=y
        self.vx=vx
        self.vy=vy
        self.color=c
        return
    @classmethod
    def update(self,w,h,t):
        time = float(t)/float(1000000)
        #print(time)
        xp = float(self.vx)*float(time)
        yp= float(self.vy)*float(time)
        self.x += xp
        self.y += yp
        #print (str(xp) +"," +str(yp))
        if self.x<32:
            self.vx = 0 - self.vx
            self.x += (32-self.x)
        if self.y<32:
            self.vy = 0 - self.vy
            self.y += (32-self.y)
        if self.x+32>w:
            self.vx = 0 - self.vx
            self.x -= (self.x+32)-w
        if self.y+32>h:
            self.vy = 0 - self.vy
            self.y -= (self.y+32)-h
        return

类通过以下方法更新

def play_u(self):
    t = self.gt.elapsed_time()
    self.gt.set_timer()
    for i in self.balls:
        i.update(self.width,self.height,t)
    return

Answer 1

这里有一个关于如何调用 update 来处理多个具有相同参数的 Ball 对象的想法。这里我使用multiprocessing.pool.Pool 类。

由于 Python 将 Ball 对象从主进程序列化/反序列化到池中将执行任务的进程，因此对对象的任何修改都不会反映在“存在”的对象副本中在主要过程中（如您所见）。但这并不妨碍update 返回一个已更新属性的列表（或元组），这些属性已被主进程用于更新其对象副本。

class Ball:
    # If this is a class constant, then it can and should stay here:
    radius = 32

    def __init__(self, x, y, vx, vy, c):
        self.x = x
        self.y = y
        self.vx = vx
        self.vy = vy
        self.color = c
        return

    def update(self, w, h, t):
        time = float(t) / 1000000.0
        #print(time)
        xp = float(self.vx) * float(time)
        yp = float(self.vy) * float(time)
        self.x += xp
        self.y += yp
        #print (str(xp) +"," +str(yp))
        if self.x < 32:
            self.vx = 0 - self.vx
            self.x += (32 - self.x)
        if self.y < 32:
            self.vy = 0 - self.vy
            self.y += (32 - self.y)
        if self.x + 32 > w:
            self.vx = 0 - self.vx
            self.x -= (self.x + 32) - w
        if self.y + 32 > h:
            self.vy = 0 - self.vy
            self.y -= (self.y + 32) - h
        # Return tuple of attributes that have changed
        # (Not used by serial benchmark)
        return (self.x, self.y, self.vx, self.vy)

    def __repr__(self):
        """
        Return internal dictionary of attributes as a string
        """
        return str(self.__dict__)

def prepare_benchmark():
    balls = [Ball(1, 2, 3, 4, 5) for _ in range(1000)]
    arg_list = (3.0, 4.0, 1.0)
    return balls, arg_list

def serial(balls, arg_list):
    for ball in balls:
        ball.update(*arg_list)

def parallel_updater(arg_list, ball):
    return ball.update(*arg_list)

def parallel(pool, balls, arg_list):
    from functools import partial

    worker = partial(parallel_updater, arg_list)
    results = pool.map(worker, balls)
    for idx, result in enumerate(results):
        ball = balls[idx]
        # unpack:
        ball.x, ball.y, ball.vx, ball.vy = result

def parallel2(pool, balls, arg_list):
    results = [pool.apply_async(ball.update, args=arg_list) for ball in balls]
    for idx, result in enumerate(results):
        ball = balls[idx]
        # unpack:
        ball.x, ball.y, ball.vx, ball.vy = result.get()

def main():
    import time

    # Serial performance:
    balls, arg_list = prepare_benchmark()
    t = time.perf_counter()
    serial(balls, arg_list)
    elapsed = time.perf_counter() - t
    print(balls[0])
    print('Serial elapsed time:', elapsed)

    print()
    print('-'*80)
    print()

    # Parallel performance using map
    # We won't even include the time it takes to create the pool
    from multiprocessing import Pool
    pool = Pool() # pool size is 8 on my desktop
    balls, arg_list = prepare_benchmark()
    t = time.perf_counter()
    parallel(pool, balls, arg_list)
    elapsed = time.perf_counter() - t
    print(balls[0])
    print('Parallel elapsed time:', elapsed)

    print()
    print('-'*80)
    print()

    # Parallel performance using apply_async
    balls, arg_list = prepare_benchmark()
    t = time.perf_counter()
    parallel2(pool, balls, arg_list)
    elapsed = time.perf_counter() - t
    print(balls[0])
    print('Parallel2 elapsed time:', elapsed)


    pool.close()
    pool.join()


# Required for windows
if __name__ == '__main__':
    main()

打印：

{'x': -29.0, 'y': -28.0, 'vx': 3, 'vy': 4, 'color': 5}
Serial elapsed time: 0.0018328999999999984

--------------------------------------------------------------------------------

{'x': -29.0, 'y': -28.0, 'vx': 3, 'vy': 4, 'color': 5}
Parallel elapsed time: 0.236945

--------------------------------------------------------------------------------

{'x': -29.0, 'y': -28.0, 'vx': 3, 'vy': 4, 'color': 5}
Parallel2 elapsed time: 0.1460790000000000

我对所有内容都使用了无意义的参数，但是您可以看到，当您拥有像 @ 这样微不足道的工作函数时，无法通过并行处理 1,000 个调用来补偿处理主进程对象的序列化/反序列化和更新的开销987654329@.

请注意，使用方法 apply_async 的基准 Parallel2 在这种情况下实际上比使用方法 map 的基准 Parallel 性能更高，这有点令人惊讶。我的猜测是，这部分是由于必须使用方法functools.partial 以arg_list 的形式将额外的、不变的w、h 和t 参数传递给工作函数@987654337 @，它提供了所需的附加函数调用。因此，基准 Parallel 必须为每次球更新进行总共两个函数调用。