Using pyfmi with multiprocessing for simulation of

问题:

I am trying to simulate multiple Modelica FMUs in parallel using python/pyfmi and multiprocessing. However I am not able to return any pyfmi FMI objects from the subprocesses once the FMUs are initialized. It seems that pyfmi FMI objects (e.g. pyfmi.fmi.FMUModelCS2 or pyfmi.fmi.FMUState2) are not pickable. I also tried dill to pickle, which doesn't work for me eather. With dill the objects are picklable though, meaning no error, but somehow corrupted if I try to reload them afterwards. Does anyone have an idea of how to solve this issue? Thanks!

回答1:

The problem is that pyfmi.fmiFMUModelCS2 is a Cython class dependent on external libraries which makes it unpickable. So it is not possible unfortunately.

If you want to use multiprocessing the only way forward that I see is that you first create the processes and then load the FMUs into the separate processes. In this way you do not need to pickle the classes.

回答2:

I faced a similar problem when I created EstimationPy. I ended up creating a wrapper for running parallel simulation of the same FMU using multiple processes.

I suggest you to look at the implementation here https://github.com/lbl-srg/EstimationPy/blob/master/estimationpy/fmu_utils/fmu_pool.py

And to the example http://lbl-srg.github.io/EstimationPy/modules/examples/first_order.html#run-multiple-simulations

回答3:

The pathos module allows multiprocessing with a similar interface as the multiprocessing but relies on dill instead of pickle for serialisation. The Pool method works for parallel execution of model.simulate, provided that results are handled in memory:

n_core = 2
n_simulation = 10
# ====

import pyfmi
model = pyfmi.load_fmu(path_fmu)

def worker(*args):
    model.reset()
    print "================> %d" % args[0]
    return model.simulate(options=dict(result_handling="memory"))["y"]


from pathos.multiprocessing import Pool
pool = Pool(n_core)
out = pool.map(worker, range(n_simulation))
pool.close()
pool.join()

Note in the above snippet that it is necessary to handle results in memory : options=dict(result_handling="memory"). The default is to use temporary files which works for when the amount of simulations is small. However, the longer the queue, the higher the chance to get something like

Exception in thread Thread-27:
Traceback (most recent call last):
  File "/home/USER/anaconda2/lib/python2.7/threading.py", line 801, in __bootstrap_inner
    self.run()
  File "/home/USER/anaconda2/lib/python2.7/threading.py", line 754, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/home/USER/anaconda2/lib/python2.7/site-packages/multiprocess/pool.py", line 389, in _handle_results
    task = get()
  File "/home/USER/anaconda2/lib/python2.7/site-packages/dill/dill.py", line 260, in loads
    return load(file)
  File "/home/USER/anaconda2/lib/python2.7/site-packages/dill/dill.py", line 250, in load
    obj = pik.load()
  File "/home/USER/anaconda2/lib/python2.7/pickle.py", line 864, in load
    dispatch[key](self)
  File "/home/USER/anaconda2/lib/python2.7/pickle.py", line 1139, in load_reduce
    value = func(*args)
TypeError: __init__() takes exactly 2 arguments (1 given)

which I fail to grasp.