OR-Tools CP-SAT Solver多Worker搜索冻结与Worker数量优化问询

问题背景 #

在Docker容器中通过Airflow运行标准作业车间调度任务，机器配置为Intel Xeon Gold 6230 CPU、300GB+内存。遇到两个核心问题：

1. 运算操作数约1000时，求解器因内存错误（返回码-9）崩溃，增加内存无法解决；
2. 设置大量Worker（≥100）可正常求解，但Worker数量越多，在找到指定数量解或触发定时器调用StopSearch后，越容易出现任务冻结，持续心跳无退出，日志显示已执行StopSearch但无响应。

使用的ObjectiveEarlyStopping回调类代码如下：

class ObjectiveEarlyStopping(cp_model.CpSolverSolutionCallback):
    def __init__(self, solution_count_limit=GlobalVariables.SOLUTION_COUNT_LIMIT,
                 max_execution_time=GlobalVariables.MIN_EXECUTION_WORKTIME):
        super(ObjectiveEarlyStopping, self).__init__()
        self._solution_count = 0
        self._solution_limit = max(2, solution_count_limit)
        self._max_execution_time = max_execution_time

        self._logger = LocalLogger().logger
        self._timer = None
        self._no_improvement_timer_limit = GlobalVariables.SOLUTION_IMPROVEMENT_TIMEOUT_SEC
        self._total_execution_time = 0

    def on_solution_callback(self):
        self._solution_count += 1
        self._logger.info(f"Feasible solution #{self._solution_count} found.")

        if self._solution_count >= self._solution_limit:
            self._logger.debug(f"Stopping search after {self._solution_count} solutions")
            self._stop_timer()
            self._timer = None
            self._logger.info(f"Stopping search __3")
            super().StopSearch()
            self._logger.info(f"Stopping search __4")
        else:
            self._reset_timer()

    def _stop_timer(self):
        if self._timer:
            self._timer.cancel()

    def _reset_timer(self):
        self._total_execution_time += self._no_improvement_timer_limit
        self._stop_timer()
        self._timer = Timer(self._no_improvement_timer_limit, self.StopSearch)
        self._timer.start()

    def StopSearch(self):
        self._logger.debug(f"{self._no_improvement_timer_limit} seconds without improvement")
        self._timer = None

        if self._solution_count >= 2 or self._total_execution_time > self._max_execution_time:
            self._logger.info(f"Stopping search __1")
            super().StopSearch()
            self._logger.info(f"Stopping search __2")
        else:
            self._logger.debug("Not enough solutions, continue search")
            self._reset_timer()

日志关键片段：

[2024-04-12, 13:46:49 UTC] {solution_callback.py:46} DEBUG - 20 seconds without improvement
[2024-04-12, 13:46:49 UTC] {solution_callback.py:50} INFO - Stopping search __1
[2024-04-12, 13:46:49 UTC] {solution_callback.py:52} INFO - Stopping search __2
[后续持续输出Airflow心跳日志，任务无退出]

问题解答 #

1. 如何避免求解器退出时的冻结问题？ #

（1）移除自定义Timer，改用求解器原生参数控制停止

自定义threading.Timer运行在独立线程中，调用StopSearch存在线程安全问题，易导致求解器无法响应停止信号。直接使用OR-Tools CP-SAT原生参数控制停止条件，稳定性更高：

from ortools.sat.python import cp_model

solver = cp_model.CpSolver()
# 设置最大运行时间（秒）
solver.parameters.max_time_in_seconds = GlobalVariables.MIN_EXECUTION_WORKTIME
# 设置可行解数量上限
solver.parameters.solution_limit = max(2, GlobalVariables.SOLUTION_COUNT_LIMIT)
# 设置无改进超时（针对目标优化场景）
solver.parameters.no_solution_improvement_limit = GlobalVariables.SOLUTION_IMPROVEMENT_TIMEOUT_SEC

（2）修复回调方法命名冲突

自定义回调类中的StopSearch方法与父类重名，导致调用逻辑混乱。重命名自定义方法避免覆盖：

def _trigger_stop_search(self):
    self._logger.debug(f"{self._no_improvement_timer_limit} seconds without improvement")
    self._timer = None

    if self._solution_count >= 2 or self._total_execution_time > self._max_execution_time:
        self._logger.info(f"Stopping search __1")
        super().StopSearch()
        self._logger.info(f"Stopping search __2")
    else:
        self._logger.debug("Not enough solutions, continue search")
        self._reset_timer()

# 修改_reset_timer中的调用
self._timer = Timer(self._no_improvement_timer_limit, self._trigger_stop_search)

（3）添加任务超时兜底机制

确保Airflow任务能强制退出，避免无限挂起：

• Airflow任务级超时：在DAG定义中设置execution_timeout：

  from datetime import timedelta
  
  task = PythonOperator(
      task_id="production_shceduling_optimizer",
      python_callable=your_solver_function,
      execution_timeout=timedelta(minutes=30),
      # 其他参数
  )
  

• 进程级隔离：将求解器逻辑放在子进程中运行，超时强制终止：

  import multiprocessing
  from multiprocessing import Process
  
  def run_solver():
      # 求解器逻辑代码
      pass
  
  p = Process(target=run_solver)
  p.start()
  p.join(timeout=1800)  # 30分钟超时
  if p.is_alive():
      p.terminate()
      p.join()
      raise Exception("Solver timed out and was terminated")
  

（4）升级OR-Tools版本并简化回调逻辑

旧版本OR-Tools可能存在多线程停止的bug，升级到最新稳定版。同时减少回调中的IO操作（如频繁打日志），避免阻塞求解器停止流程。

2. 如何根据输入参数合理选择Worker数量？ #

这里的Worker对应OR-Tools的num_search_workers（并行搜索线程数），需结合硬件、问题规模综合选择：

• 硬件上限约束：Intel Xeon Gold 6230为20物理核、40逻辑线程，线程数不要超过40，否则会引发严重CPU上下文切换，降低效率且增加冲突概率。建议初始设置为20（物理核数）或40（逻辑线程数）。
• 问题规模适配：
  • 操作数<500：设置为10-20（物理核数的1/2），减少内存开销和线程冲突；
  • 操作数500-1500：设置为20-40，平衡并行效率与稳定性；
  • 操作数>1500：尝试40，但需监控内存/CPU使用率，若出现波动则下调至20。
• 内存错误排查：返回码-9通常是系统OOM Killer触发的终止，单线程崩溃大概率是模型设计问题：
  1. 优化模型：减少冗余变量，用紧凑约束替代离散约束；
  2. 设置求解器内存上限：solver.parameters.max_memory_in_mb = 200000（根据机器内存调整）；
  3. 监控内存使用：用top或docker stats确认是否真的内存不足。
• 动态调优：针对不同操作数测试不同Worker数量的求解时间、内存占用、冻结概率，形成映射表：

  操作数范围	推荐Worker数量
  <500	10-20
  500-1500	20-40
  >1500	30-40

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