优化CPMpy中累积约束的性能：解决大规模任务调度问题

import cpmpy as cp
import logging
from typing import List

class CumulativeTestModel:
    def __init__(self, task_duration: int, nb_tasks: int, end_date: int):
        self.model: cp.Model = cp.Model()

        # 定义变量
        # objective: 最小化所需机器数量
        self.objective: cp.IntVar = cp.intvar(0, nb_tasks)
        # starts: 每个任务的开始时间
        starts: List[cp.IntVar] = [cp.intvar(0, end_date) for _ in range(nb_tasks)]
        # durations: 每个任务的持续时间
        durations: List[int] = [task_duration] * nb_tasks
        # ends: 每个任务的结束时间
        ends: List[cp.IntVar] = [cp.intvar(0, end_date) for _ in range(nb_tasks)]
        # demands: 每个任务对资源的占用量（此处为1，表示一台机器）
        demands: List[int] = [1] * nb_tasks

        # 添加 Cumulative 约束到模型
        # 确保在任何时间点，所有活动任务的累积需求（demands）不超过容量（capacity，即机器数量）
        self.model += cp.Cumulative(
            start=starts,
            duration=durations,
            end=ends,
            demand=demands,
            capacity=self.objective,
        )

        # 最小化目标变量（所需机器数量）
        self.model.minimize(self.objective)
        logging.info(f"Model created with {nb_tasks} tasks.")

    def run(self):
        # 使用 ortools 求解器
        solver = cp.model.SolverLookup.get("ortools", self.model)
        has_solution = solver.solve()

        if not has_solution:
            logging.info("No solution found.")
        else:
            logging.info(f"Solution found: {solver.status()} -> {self.objective.value()} (Time not explicitly logged here, but observed externally)")

if __name__ == "__main__":
    # 示例用法，观察不同任务数量下的性能
    logging.basicConfig(level=logging.INFO)
    print("--- 原始性能测试 ---")
    CumulativeTestModel(task_duration=10, nb_tasks=3, end_date=15).run()
    CumulativeTestModel(task_duration=10, nb_tasks=5, end_date=25).run()
    CumulativeTestModel(task_duration=10, nb_tasks=7, end_date=35).run()
    CumulativeTestModel(task_duration=10, nb_tasks=9, end_date=45).run()
    CumulativeTestModel(task_duration=10, nb_tasks=11, end_date=55).run()
    # CumulativeTestModel(task_duration=10, nb_tasks=13, end_date=65).run() # 在旧版本中会长时间运行或挂起
    # CumulativeTestModel(task_duration=10, nb_tasks=21, end_date=105).run() # 在旧版本中会长时间运行或挂起