您可以利用多进程来在子进程中运行模拟,并利用复制-写入(copy-on-write)的分支优势,仅在开始时解封/处理数据一次:
import multiprocessing
import pickle
mp = multiprocessing.get_context('fork')
data = pickle.load(open(DATA_ROOT + pickle_name, 'rb'))
def _run_simulation(_):
run_simulation()
with mp.Pool() as pool:
pool.map(_run_simulation, range(num_simulations))
如果您想对每个模拟运行进行参数化,可以按照以下方式操作:
import multiprocessing
import pickle
mp = multiprocessing.get_context('fork')
data = pickle.load(open(DATA_ROOT + pickle_name, 'rb'))
with mp.Pool() as pool:
simulations = ('arg for simulation run', 'arg for another simulation run')
pool.map(run_simulation, simulations)
这样,run_simulation
函数将会接收到来自simulations
元组的值,这可以允许每个模拟使用不同的参数运行,甚至只是为了记录/保存目的分配每次运行一个ID编号或名称。
整个方法都依赖于fork的可用性。有关在Python的内置multiprocessing库中使用fork的更多信息,请参见有关上下文和启动方法的文档。您可能还希望考虑使用forkserver
多进程上下文(使用mp = multiprocessing.get_context('fork')
)出于文档中所述的原因。
如果您不想并行运行模拟,则可以调整此方法以适用于该需求。关键的是,为了只处理数据一次,您必须在处理数据的进程或其子进程之一中调用run_simulation
。
例如,如果您想编辑run_simulation
执行的操作,然后再次运行它,您可以使用类似以下代码的方式进行:
main.py
:
import multiprocessing
from multiprocessing.connection import Connection
import pickle
from data import load_data
load_data()
mp = multiprocessing.get_context('fork')
def load_and_run_simulation(result_pipe: Connection) -> None:
from simulation import run_simulation
try:
run_simulation()
except Exception as ex:
result_pipe.send(ex)
else:
result_pipe.send(None)
def run_simulation_in_child_process() -> None:
result_pipe_output, result_pipe_input = mp.Pipe(duplex=False)
proc = mp.Process(
target=load_and_run_simulation,
args=(result_pipe_input,)
)
print('Starting simulation')
proc.start()
try:
result = result_pipe_output.recv()
if isinstance(result, Exception):
raise result
proc.join()
except KeyboardInterrupt:
print("Caught 'KeyboardInterrupt'; terminating simulation")
proc.terminate()
print('Simulation finished')
if __name__ == '__main__':
while True:
choice = input('\n'.join((
'What would you like to do?',
'1) Run simulation',
'2) Exit\n',
)))
if choice.strip() == '1':
run_simulation_in_child_process()
elif choice.strip() == '2':
exit()
else:
print(f'Invalid option: {choice!r}')
data.py
:
from functools import lru_cache
@lru_cache
def load_data():
with open(DATA_ROOT + pickle_name, 'rb') as f:
return pickle.load(f)
simulation.py
:
from data import load_data
data = load_data()
def run_simulation():
...
上述三个文件应该都在同一个目录下,和一个可以为空的
__init__.py
文件一起。
main.py
文件可以根据您的喜好进行重命名,并且是此程序的主要入口点。您可以直接运行
simulation.py
,但这将导致长时间加载/处理数据,这是您最初遇到的问题。当
main.py
正在运行时,可以编辑
simulation.py
文件,因为每次从
main.py
运行模拟时,它会重新加载。
对于macOS用户:在macOS上分叉可能有点错误,这就是为什么Python默认使用
spawn
方法在macOS上进行多进程处理,但仍然支持
fork
和
forkserver
。如果遇到崩溃或与多进程相关的问题,请尝试将
OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES
添加到您的环境中。有关更多详细信息,请参见
https://dev59.com/KVUL5IYBdhLWcg3wjIpQ#52230415。
shared_memory
将信息存储为字节缓冲区。如果您不想共享数组,那么您可能需要重新序列化数据以保存在其中。 - thsheanumpy
,但它使用了什么呢?你需要在运行之间保存什么庞大的数据结构?你不能将整个 Python 对象保存到某种共享内存空间中,如果你尝试这样做,会严重破坏解释器的内存管理。但是根据你的数据实际情况,你可能能够共享一些东西,我们只有在了解数据的情况下才能知道它是什么。 - Blckknght