为什么并行处理后我的打印语句没有到达？

Question

我写了一个函数，运行一个随机模拟一系列化学反应，这里是函数的简要总结：

def stochastic_simulation(*args): # arguments are a series of arrays
    t = SimulationTimer()
    t.start()
    update molecule numbers for each species in model
    update current time of the system 
    t.stop()
    print(f"Accumulated time: {t.get_accumulated_time():0.10f} seconds")
return popul_num_all, tao_all, t.get_accumulated_time() # popul_num_all is an array of changing molecule numbers over time, tao_all is the evolution of time throughout the simulation. 

SimulationTimer 是一个定制的计时器：

class TimeError(Exception):
"""A custom exception used to report errors in use of Timer Class""" 
pass

class SimulationTimer:  
    def __init__(self):
        self._simulation_start_time = None
        self._simulation_stop_time = None 
        self.accumulated_elapsed_time = 0.0

    def start(self):
        """start a new timer"""
        if self._simulation_start_time is not None:    # attribute
            raise TimeError(f"Timer is running.\n Use .stop() to stop it")

        self._simulation_start_time = time.perf_counter()  
    def stop(self):
        """stop the time and report the elsaped time"""
        if self._simulation_start_time is None:
            raise TimeError(f"Timer is not running.\n Use .start() to start it.")

        self._simulation_stop_time = time.perf_counter()
        elapsed_simulation_time = self._simulation_stop_time - 
        self._simulation_start_time  
        self.accumulated_elapsed_time += elapsed_simulation_time

        self._simulation_start_time = None
        print(f"Elapsed time: {elapsed_simulation_time:0.10f} seconds")

    def get_accumulated_time(self):
        """ Return the elapsed time for later use"""
        return self.accumulated_elapsed_time

我正在尝试使用multiprocessing 库的pool 方法并行运行stochastic_simulation。 t.get_accumulated_time() 意味着并行运行我的所有模拟所需的时间。

if __name__ == '__main__':
    with Pool() as p:
        pool_results = p.map(stochastic_simulation, [start_state, LHS, stoch_rate, state_change_array])

        total_time = 0.0
        for tuple_results in pool_results:
            total_time += tuple_results[2] 
    print(f"Total time:\n{total_time}")

上面的代码是我运行并行模拟的方式，它有两个问题。首先，它实际上永远不会到达 print 语句，f"Total time:\n{total_time}" 永远不会被打印出来。其次，它永远不会返回运行所有并行模拟所花费的时间的一个组合度量t.get_accumulated_time()，而是为每个进程返回一个单独的时间。

我不确定这些问题是否与 scoping、class object 或 multiprocessing 有关。

编辑：

我对@9​​87654335@ 的输入都是数组。 描述以下系统：

1S + 0T + 0U --> 0S + 0T + 0U
2S + 0T + 0U --> 0S + 1T + 0U 
0S + 1T + 0U --> 2S + 0T + 0U
0S + 1T + 0U --> 0S + 0T + 1U   

1. initial_state 这是时间为零时每个分子种类的数量。在我的模型中有 4 个物种，所以它有 len = 3
2. LHS 是不同反应的反应物之间比率的二维数组。
3. stoch_rate是长度为4的数组每个反应的速率
4. state_change_array 是反应发生后分子的净变化。

有什么建议吗？

Answer 1

我能够使用以下代码获得一个最小的可重现示例：

import multiprocessing as mp
import time

def stochastic_simulation(*args): # arguments are a series of arrays
    print('args are:', *args)
    t = SimulationTimer()
    t.start()
    # update molecule numbers for each species in model
    # update current time of the system 
    t.stop()
    print(f"Accumulated time: {t.get_accumulated_time():0.10f} seconds")
    return t.get_accumulated_time() # deleted some returns

class TimeError(Exception):
    """A custom exception used to report errors in use of Timer Class""" 
    pass

class SimulationTimer: 
    accumulated_elapsed_time = 0.0  

    def __init__(self):
        self._simulation_start_time = None
        self._simulation_stop_time = None 

    def start(self):
        """start a new timer"""
        if self._simulation_start_time is not None:    # attribute
            raise TimeError(f"Timer is running.\n Use .stop() to stop it")

        self._simulation_start_time = time.perf_counter()  
        
    def stop(self):
        """stop the time and report the elsaped time"""
        if self._simulation_start_time is None:
            raise TimeError(f"Timer is not running.\n Use .start() to start it.")

        self._simulation_stop_time = time.perf_counter()
        elapsed_simulation_time = self._simulation_stop_time - self._simulation_start_time  
        self.accumulated_elapsed_time += elapsed_simulation_time

        self._simulation_start_time = None
        print(f"Elapsed time: {elapsed_simulation_time:0.10f} seconds")

    def get_accumulated_time(self):
        """ Return the elapsed time for later use"""
        return self.accumulated_elapsed_time

if __name__ == '__main__':
    with mp.Pool() as p:
        pool_results = p.map(stochastic_simulation, [1,2,3,4]) #changed inputs
        print(pool_results)
        total_time = 0.0
        for tuple_results in pool_results:
            total_time += tuple_results #removed the [2] index
    print(f"Total time:\n{total_time}")

输出：

args are: 1
Elapsed time: 0.0000015800 seconds
Accumulated time: 0.0000015800 seconds
args are: 2
Elapsed time: 0.0000011850 seconds
Accumulated time: 0.0000011850 seconds
args are: 3
Elapsed time: 0.0000007900 seconds
Accumulated time: 0.0000007900 seconds
args are: 4
Elapsed time: 0.0000007900 seconds
Accumulated time: 0.0000007900 seconds
[1.580000000001025e-06, 1.184999999986891e-06, 7.900000000005125e-07, 7.900000000005125e-07]
Total time:
4.344999999988941e-06

这让我相信你的输入可能会发生一些奇怪的事情......不确定你试图在计时器之间进行什么计算，但会尝试缩小代码停止的确切位置

另外，如果您想运行 4 次模拟，每个模拟有 4 个参数，您应该提供参数列表列表，即[[1,2,3,4], [4,3,2,1], [23,23,63,2], [6,2,1,5]]