使用multiprocessing Pool.map生成导弹拦截轨迹时SQLite报错

问题背景 #

我构建了一个可根据输入状态参数输出导弹-目标最优拦截轨迹的模型，计划通过遍历一组状态参数批量生成多条拦截轨迹，并用multiprocessing实现多进程处理。但运行时仅生成一个Excel文件，且抛出SQLite表已存在的异常。

核心代码 #

轨迹优化函数CompOptimalTraj #

import openmdao.api as om
import numpy as np
import dymos as dm
import matplotlib.pyplot as plt
from MIssileODEdef import MissileODE
import pandas as pd


def CompOptimalTraj(States):

    # 拆分状态数组为单个状态变量
    X_m=States[0] 
    Y_m=States[1] 
    X_t=States[2] 
    Y_t=States[3] 
    theta=States[4] 
    phi_t =States[5] 
    phi_m=States[6] 
    V_m=States[7] 
    V_t=States[8] 
    A_t=States[9] 
    Counter = States[10]
    
    # 初始化问题与求解驱动
    p = om.Problem(model=om.Group())
    p.driver = om.pyOptSparseDriver()
    p.driver.options['optimizer'] = 'IPOPT'
    p.driver.declare_coloring()

    # 初始化单阶段轨迹
    traj = p.model.add_subsystem('traj', dm.Trajectory())
    phase = traj.add_phase('phase0', dm.Phase(
        ode_class=MissileODE, transcription=dm.GaussLobatto(num_segments=25, order=3)))

    phase.set_time_options(fix_initial=True, duration_bounds=(.5, 40))

    # 添加状态变量
    phase.add_state('X_m', fix_initial=True, fix_final=False,
                    rate_source='Xm_dot', units='m', lower=0, upper=10000)
    phase.add_state('Y_m', fix_initial=True, fix_final=False,
                    rate_source='Ym_dot', units='m', lower=0, upper=10000)
    phase.add_state('X_t', fix_initial=True, fix_final=False,
                    rate_source='Xt_dot', units='m', lower=0, upper=10000)
    phase.add_state('Y_t', fix_initial=True, fix_final=False,
                    rate_source='Yt_dot', units='m', lower=0, upper=10000)
    phase.add_state('theta', fix_initial=True, fix_final=False, lower=0,
                    upper=np.pi, rate_source='theta_dot', units='rad', targets=['theta'])
    phase.add_state('phi_t', fix_initial=True, fix_final=False, lower=0,
                    upper=np.pi, rate_source='phi_tdot', units='rad', targets=['phi_t'])
    phase.add_state('phi_m', fix_initial=True, fix_final=False, lower=0,
                    upper=np.pi, rate_source='phi_mdot', units='rad', targets=['phi_m'])
    phase.add_state('V_m', fix_initial=True, fix_final=False, lower=300,
                    upper=1200, rate_source='A_m', units='m/s', targets=['V_m'])
    phase.add_state('V_t', fix_initial=True, fix_final=False, lower=-600, upper=-300, rate_source='A_t', units='m/s', targets=['V_t'])

    # 添加控制变量与参数
    phase.add_control('A_m', continuity=True, rate_continuity=True,
                    units='m/(s**2)', lower=-50, upper=50, targets=['A_m'])
    phase.add_parameter('A_t', val=A_t, units='m/s**2',
                        opt=False, targets=['A_t'])

    # 添加边界约束与目标函数
    phase.add_boundary_constraint(
        'R =((X_t - X_m)**2 + (Y_t - Y_m)**2)**0.5', loc='final', equals=1)
    phase.add_objective('time', loc='final')

    p.model.linear_solver = om.DirectSolver()
    p.setup()

    # 设置初始值
    p['traj.phase0.t_initial'] = 0.0
    p['traj.phase0.t_duration'] = 40

    p.set_val('traj.phase0.states:X_m', phase.interp('X_m', ys=[X_m, 0.5]))
    p.set_val('traj.phase0.states:Y_m', phase.interp('Y_m', ys=[Y_m, 1]))
    p.set_val('traj.phase0.states:X_t', phase.interp('X_t', ys=[X_t, 1]))
    p.set_val('traj.phase0.states:Y_t', phase.interp('Y_t', ys=[Y_t, 0.2]))
    p.set_val('traj.phase0.states:theta',
            phase.interp('theta', ys=[theta, np.pi]))
    p.set_val('traj.phase0.states:V_t', phase.interp('V_t', ys=[V_t, -300]))
    p.set_val('traj.phase0.states:phi_t',
            phase.interp('phi_t', ys=[phi_t, np.pi]))
    p.set_val('traj.phase0.states:phi_m', phase.interp(
        'phi_m', ys=[phi_m,  np.pi]))
    p.set_val('traj.phase0.states:V_m', phase.interp('V_m', ys=[V_m, 1200]))

    p.set_val('traj.phase0.controls:A_m', phase.interp('A_m', [0, 0]))

    # 运行问题并记录结果
    dm.run_problem(p, run_driver=True, simulate = True, solution_record_file='dymos_solution'+str(Counter)+'.db',simulation_record_file='dymos_simulation'+str(Counter)+'.db')

    exp_out = traj.simulate()

    sol = om.CaseReader('dymos_solution'+str(Counter)+'.db').get_case('final')
    sim = om.CaseReader('dymos_simulation'+str(Counter)+'.db').get_case('final')

    t = sol.get_val('traj.phase0.timeseries.time')
    X_m = sol.get_val('traj.phase0.timeseries.states:X_m')
    Y_m = sol.get_val('traj.phase0.timeseries.states:Y_m')
    X_t = sol.get_val('traj.phase0.timeseries.states:X_t')
    Y_t = sol.get_val('traj.phase0.timeseries.states:Y_t')
    A_m = sol.get_val('traj.phase0.timeseries.controls:A_m')

    df = pd.DataFrame(list(zip(t,X_m,Y_m,X_t,Y_t,A_m)))
    
    return df.to_excel('MissileStateHistory'+str(Counter)+'.xlsx')

多进程调用代码 #

import numpy as np
import pandas as pd
from MissileOPT import CompOptimalTraj
from multiprocessing import Pool
  
xl = pd.read_excel('Trial5StateParamGen.xlsx')
StateParams = xl.to_numpy()

if __name__ == '__main__':
    with Pool(processes=3) as pool:
       pool.map(CompOptimalTraj,StateParams)

报错信息 #

Traceback (most recent call last):
  File "C:\Users\Name\anaconda3\lib\multiprocessing\pool.py", line 125, in worker
    result = (True, func(*args, **kwds))
  File "C:\Users\Name\anaconda3\lib\multiprocessing\pool.py", line 48, in mapstar
    return list(map(*args))
  File "c:\Users\Name\OneDrive - University of Bath\Desktop\Uni stuff\5th year uni\FYP\Code\OC\OC_algos_Trial5-Multiprocessing\MissileOPT.py", line 112, in CompOptimalTraj
    dm.run_problem(p, run_driver=True, simulate=True)
  File "C:\Users\Name\anaconda3\lib\site-packages\dymos\run_problem.py", line 79, in run_problem
    problem.final_setup()
  File "C:\Users\Name\anaconda3\lib\site-packages\openmdao\utils\hooks.py", line 131, in execute_hooks
    ret = f(*args, **kwargs)
  File "C:\Users\Name\anaconda3\lib\site-packages\openmdao\core\problem.py", line 1030, in final_setup
    self._setup_recording()
  File "C:\Users\Name\anaconda3\lib\site-packages\openmdao\core\problem.py", line 767, in _setup_recording
    self._rec_mgr.startup(self, self.comm)
  File "C:\Users\Name\anaconda3\lib\site-packages\openmdao\recorders\recording_manager.py", line 75, in startup
    recorder.startup(recording_requester, comm)
  File "C:\Users\Name\anaconda3\lib\site-packages\openmdao\recorders\sqlite_recorder.py", line 332, in startup
    self._initialize_database(comm)
  File "C:\Users\Name\anaconda3\lib\site-packages\openmdao\recorders\sqlite_recorder.py", line 238, in _initialize_database
    c.execute("CREATE TABLE global_iterations(id INTEGER PRIMARY KEY, ")
sqlite3.OperationalError: table global_iterations already exists

问题原因 #

OpenMDAO默认启用全局SQLite记录器，多进程运行时，多个进程会同时尝试初始化同一个默认数据库文件（通常是cases.sqlite），导致表重复创建的冲突。即使你指定了solution_record_file和simulation_record_file，OpenMDAO的全局记录器仍在后台运行，引发资源竞争。

解决方法 #

方法1：直接禁用全局记录器 #

在每个进程的问题初始化后，显式关闭全局记录功能：

# 在p = om.Problem(model=om.Group())之后添加以下代码
p.recording_options['record_viewer_data'] = False
p.driver.recording_options['record_viewer_data'] = False
# 清除所有默认记录器
p._rec_mgr.clear_recorders()

方法2：为每个进程分配独立记录文件 #

如果需要保留记录功能，为每个进程创建唯一的记录文件，避免冲突：

# 在创建Problem后添加以下代码
from openmdao.recorders.sqlite_recorder import SqliteRecorder

# 用Counter生成唯一文件名
recorder = SqliteRecorder(f'process_recorder_{Counter}.db')
p.add_recorder(recorder)
# 清除默认记录器，只保留自定义的
p._rec_mgr.clear_recorders()
p._rec_mgr.add_recorder(recorder)

方法3：修改dm.run_problem调用参数 #

在调用dm.run_problem时，添加record_viewer_data=False参数，禁用默认的viewer数据记录：

dm.run_problem(p, run_driver=True, simulate=True, 
               solution_record_file=f'dymos_solution{Counter}.db',
               simulation_record_file=f'dymos_simulation{Counter}.db',
               record_viewer_data=False)

额外注意事项 #

• 确保所有生成的文件（DB、Excel）都通过Counter保持唯一，避免进程间文件覆盖
• 多进程运行时尽量减少共享文件系统的操作，必要时可对文件操作加锁，但优先推荐每个进程使用独立文件

内容的提问来源于stack exchange，提问作者S0yboi