如何并行迭代SimPy模拟？

Question

我有一个SimPy模型，它返回一个随机结果，我想重复多次。每个复制都是独立的，因此为了使其更快，我想并行地运行它们。我已经尝试过Python的多处理、Pathos多处理和joblib 并行，但是每种方法都会得到相同的错误：TypeError: can't pickle generator objects。有没有办法避免这一错误，并并行运行仿真？

SimPy依赖于解释为这里的生成器，因此避免它们是不可能的。

Answer 1

这个错误很好地描述了这个问题。在您要发送到子进程的对象中，可能是函数参数中隐藏着一个生成器。能把这个生成器转换成列表吗？

例如，下面将引发您提到的错误：

from multiprocessing import Pool

def firstn(n):
    k = 0
    while k < n:
        yield k
        k += 1

if __name__ == "__main__":
    p = Pool(2)
    print(p.map(firstn, [1, 2, 3, 4]))

但这个方法很有效：

from multiprocessing import Pool

def firstn(n):
    k = 0
    while k < n:
        yield k
        k += 1

def wrapped(n):
    return list(firstn(n))

if __name__ == "__main__":
    p = Pool(2)
    print(p.map(wrapped, [1, 2, 3, 4]))

Answer 2

您需要在新流程中从零开始实例化环境，并注意只使用普通类型作为要在Pool中映射的参数。这里是一个经过重新处理的洗车示例(来自simpy文档)，它使用不同的种子和打印在每种情况下洗了多少辆车，进行了4次并行模拟。

import multiprocessing as mp
import simpy
import random


NUM_MACHINES = 2  # Number of machines in the carwash
WASHTIME = 5      # Minutes it takes to clean a car
T_INTER = 7       # Create a car every ~7 minutes
SIM_TIME = 20     # Simulation time in minutes


class Carwash(object):
    """A carwash has a limited number of machines (``NUM_MACHINES``) to
    clean cars in parallel.

    Cars have to request one of the machines. When they got one, they
    can start the washing processes and wait for it to finish (which
    takes ``washtime`` minutes).

    """
    def __init__(self, env, num_machines, washtime):
        self.env = env
        self.machine = simpy.Resource(env, num_machines)
        self.washtime = washtime

    def wash(self, car):
        """The washing processes. It takes a ``car`` processes and tries
        to clean it."""
        yield self.env.timeout(WASHTIME)


def car(env, name, cw):
    """The car process (each car has a ``name``) arrives at the carwash
    (``cw``) and requests a cleaning machine.

    It then starts the washing process, waits for it to finish and
    leaves to never come back ...

    """
    with cw.machine.request() as request:
        yield request
        yield env.process(cw.wash(name))


def setup(env, num_machines, washtime, t_inter):
    """Create a carwash, a number of initial cars and keep creating cars
    approx. every ``t_inter`` minutes."""
    # Create the carwash
    carwash = Carwash(env, num_machines, washtime)

    # Create 4 initial cars
    for i in range(4):
        env.process(car(env, 'Car %d' % i, carwash))

    # Create more cars while the simulation is running
    while True:
        yield env.timeout(random.randint(t_inter - 5, t_inter + 5))
        i += 1
        env.i = i
        env.process(car(env, 'Car %d' % i, carwash))


# additional wrapping function to be executed by the pool
def do_simulation_with_seed(rs):

    random.seed(rs)  # This influences only the specific process being run
    env = simpy.Environment()  # THE ENVIRONMENT IS CREATED HERE, IN THE CHILD PROCESS
    env.process(setup(env, NUM_MACHINES, WASHTIME, T_INTER))

    env.run(until=SIM_TIME)

    return env.i


if __name__ == '__main__':
    seeds = range(4)
    carwash_pool = mp.Pool(4)
    ncars_by_seed = carwash_pool.map(do_simulation_with_seed, seeds)
    for s, ncars in zip(seeds, ncars_by_seed):
        print('seed={} --> {} cars washed'.format(s, ncars))