时间戳

首页 > 解决方案 > Python多处理，变量的预初始化

python - Python多处理，变量的预初始化

问题描述

我正在尝试使用多处理模块并行化我的代码。我正在编写的代码分两步工作。在第一步中，我初始化了一个类，该类计算并保存了几个变量，这些变量将在第二步中使用。在第二步中，程序使用先前初始化的变量执行计算。第一步的变量不会以任何方式修改。第一步的计算时间并不重要，但在第二步中，因为它按一定的顺序被调用了几百次。下面是构建的代码结构和 ist 输出的最小示例。

import numpy as np
import time
from multiprocessing import Pool

class test:
    def __init__(self):
        self.r = np.ones(10000000)


    def f(self,init):
        summed = 0
        for i in range(0,init):
            summed = summed + i
        return summed


if __name__ == "__main__":
    # first step 
    func = test()
    
    
    # second step
    # sequential
    start_time = time.time()
    for i in [1000000,1000000,1000000,1000000]:
        func.f(i)
    print('Sequential: ', time.time()-start_time)

    
    # parallel
    start_time = time.time()
    pool = Pool(processes=None)
    result = pool.starmap(func.f,[[1000000],[1000000],[1000000],[1000000]])
    print('Parallel: ', time.time()-start_time)

输出：
顺序：0.2673146724700928
并行：1.5638213157653809

据我了解，多处理变得更慢，因为必须将类 test 的变量 r 转移到所有工作进程。为了避免这种情况，我需要在开始 f 之前初始化每个工作人员的类。这可以通过多处理实现吗？还有其他工具可以做到这一点吗？

标签： pythonmultiprocessingpython-multiprocessing

解决方案

我已经通过使用多处理模块中的 Pipe 函数解决了这个问题。在第一步中，我现在可以初始化我的变量并设置多处理环境。然后我使用 Pipe 函数来传输输入数据。

对于“self.r = np.ones(100000000)”
并行管道：0.8008558750152588
并行 2：18.51273012161255

对于“self.r = np.ones(10000000)”
并行管道：0.71409010887146 并行 2：1.4551067352294922

import numpy as np
import time
import multiprocessing as mp


class Test:  # PEP8: `CamelCaseNames` for classes
    def __init__(self):
        self.r = np.ones(100000000)

    def f(self, init):
        summed = 0
        for i in range(init):
            summed = summed + i
        return summed


def my_function(value):
    func = Test()
    return func.f(value)


class Connection:
    def __init__(self):
        self.process = {}
        self.parent = {}
        self.child = {}

    def add(self, hub, process, parent_conn, child_conn):
        self.process[hub] = process
        self.parent[hub] = parent_conn
        self.child[hub] = child_conn


def multi_run(child_conn, func, i):
    while 1:
        init = child_conn.recv()
        data = func.f(init)
        child_conn.send(data)


if __name__ == "__main__":
    N_processes = 4

    func = Test()
    conn = Connection()
    # First step
    for i in range(N_processes):
        parent_conn, child_conn = mp.Pipe()
        process = mp.Process(target=multi_run, args=(child_conn, func, i))
        conn.add(i, process, parent_conn, child_conn)
        process.start()

    start_time = time.time()
    data = [[1000000, x] for x in range(30)]
    # Second step
    for i, j in data:
        conn.parent[j % N_processes].send(i)
    for i, j in data:
        conn.parent[j % N_processes].recv()
    print('Parallel piped:', time.time()-start_time)

    data = [[1000000] for x in range(30)]
    # parallel 2
    start_time = time.time()
    pool = mp.Pool(processes=None)
    result = pool.starmap(my_function, data)
    print('Parallel 2:', time.time()-start_time)

推荐阅读