1.2.1

下载 Notebook 下载完整案例

资源约束并行机调度问题¶

类别： 排程

使用 OptAgent 的 Python 接口描述变量、约束与目标。

问题与原始示例来源：Hexaly Code Templates。

问题描述¶

在 Resource-Constrained Parallel Machine Scheduling Problem 中，需要将一组任务分配到若干并行的相同互斥机器上。在同一台机器上，相邻任务之间存在与顺序相关的设置时间（setup times）。每个任务可以被分配可变数量的可再生资源。每个任务的处理时间是非线性的，取决于分配给该任务的资源数量。任何时刻的总资源消耗不得超过全局资源限制。目标是最小化完工时间（makespan），即最后一个任务的完成时间。

学习要点¶

• 添加 区间决策变量 来建模任务
• 添加 列表决策变量 来建模机器上任务的分配及其顺序
• 定义 符号 Lambda 函数 来建模互斥资源和累积资源约束

数据¶

我们提供随机生成的 Resource-Constrained Parallel Machine Scheduling Problem 实例，格式如下：

• 并行机器的数量。
• 任务的数量。
• 每个任务的类型。
• 每个任务的释放日期。
• 执行每个任务所需资源数量的下界。
• 执行每个任务所需资源数量的上界。
• 每个任务的基础持续时间（之后按分配的资源数量归一化）。
• 设置时间矩阵，指示同一台机器上两个相邻任务之间的最小转换时间。

建模思路¶

Resource-Constrained Parallel Machine Scheduling Problem 的 OptAgent 模型依赖三种类型的决策变量：

• 一个表示机器的 list 决策变量数组：list i 对应于分配到机器 i 的任务序列；
• 一个表示任务时间跨度的 interval 决策变量数组；
• 一个表示每个任务分配资源数量的整数决策变量数组。

我们将问题的约束定义如下。每个任务必须被分配到恰好一台机器，以确保唯一分配。每个任务关联的处理 interval 必须与其所需持续时间匹配，而该持续时间又取决于分配给它的资源数量。

互斥资源约束可以表述如下：对于任意 i，在位置 i+1 处理的任务必须在前一位置 i 上的任务结束后才能开始，二者之间再加上设置时间。为了建模此约束，我们定义了一个 符号 Lambda 函数 来表达两个相邻任务之间的关系。然后该函数在每台机器处理的所有任务上被应用于可变参数的 and 算子。注意，这些 and 表达式中的项数以及 list 的大小（每台机器上分配的任务数）在搜索过程中是变化的。

累积资源约束可以表述如下：对于每个时间槽 t，正在处理的任务所使用的资源数量不得超过总资源数。我们使用可变参数的 and 公式结合 符号 Lambda 函数，以确保可用资源数量在任何时刻都得到满足。这种写法集中表达了各时间段的资源约束；实际求解开销取决于实例规模与求解配置。

目标函数是最小化完工时间（makespan），即最后一个任务的完成时间。

Python 实现¶

# Copyright (c) Hexaly. Permission is hereby granted to use, copy,
# and modify this code for applications developed with Hexaly.
from pathlib import Path

from optagent import OptModel, solve

def main(instance_file, output_file=None, time_limit=60):
    model = OptModel()

    lines = Path(instance_file).read_text(encoding="utf-8").splitlines()

        # Number of resources
    nb_machines = int(lines[0])

        # Tasks data
    nb_tasks = int(lines[1])
    task_types = [int(i) for i in lines[2].split()]
    release_dates = [int(i) for i in lines[3].split()]

        # Resources data
    nb_resources = int(lines[4])
    min_resources = [int(i) for i in lines[5].split()]
    max_resources = [int(i) for i in lines[6].split()]

        # Durations data
    nb_duration_levels = max(max_resources)
    durations = []
    for line in lines[7:7 + nb_duration_levels]:
        durations.append([int(i) for i in line.split()])

        # Setup times between two consecutive tasks
    setup_data = [[int(i) for i in line.split()] for line in lines[7 + nb_duration_levels:]]
    setup = model.array(setup_data)

        # Horizon: trivial upper bound for the end times of the tasks
    H = max(release_dates) + sum(durations[0])

        # Interval decisions: time range of each task
    tasks = model.array([
        model.interval(release_dates[i], H)
        for i in range(nb_tasks)
    ])
        # Number of resources assigned to each task
    nb_assigned_resources = [
        model.int(min_resources[task_types[i]], max_resources[task_types[i]])
        for i in range(nb_tasks)
    ]

        # List decisions: sequence of tasks on each machine
    tasks_order = [model.list(nb_tasks) for m in range(nb_machines)]

        # Each task is scheduled on a machine
    model.constraint(model.partition(tasks_order))

        # Non-overlap constraints: the tasks assigned to the same machine are
        # scheduled one after the other, with sequence-dependent setup times
    for m, sequence in enumerate(tasks_order):
        no_overlap_lambda = model.lambda_function(
            lambda position: tasks[sequence[position // 1]].end()
            + setup[sequence[position // 1], sequence[(position + 1) // 1]]
            <= tasks[sequence[(position + 1) // 1]].start()
        )
        model.constraint(
            model.and_(model.range(0, model.count(sequence) - 1), no_overlap_lambda),
        )

        # The task duration depends on the number of assigned resources
    durations_array = model.array(durations)
    for i in range(nb_tasks):
        model.constraint(
            tasks[i].length() == durations_array[nb_assigned_resources[i] - 1, i],
        )

        # Makespan: time when all tasks have been processed
    makespan = model.max([tasks[i].end() for i in range(nb_tasks)])

        # Cumulative constraint: the number of assigned resources at any time
        # does not exceed the total number of available resources
    capacity_lambda = model.lambda_function(
        lambda t: model.sum(
            model.contains(tasks[i], t // 1) * nb_assigned_resources[i]
            for i in range(nb_tasks)
        ) <= nb_resources
    )
    model.constraint(model.and_(model.range(0, makespan), capacity_lambda))

        # Minimize the makespan
    model.minimize(makespan)
    solution = solve(model, time_limit_s=float(time_limit))
    if not solution.feasible:
        print(f"No feasible schedule found; Status = {solution.feasible}")
        return solution

        # Write the solution in a file with the following format:
        # - the total makespan
        # - for each task, the machine, the start and end times,
        #   the number of assigned resources */
    print(f"Makespan = {makespan.value}; Status = {solution.feasible}")
    if output_file is not None:
        lines = [str(makespan.value)]
        for m, sequence in enumerate(tasks_order):
            for task_id in sequence.value:
                lines.append(f"{m} {tasks[task_id].value.start()} {tasks[task_id].value.end()} {nb_assigned_resources[task_id].value}")
        Path(output_file).write_text("\n".join(lines) + "\n", encoding="utf-8")
    return solution

实例调用¶

下面使用仓库提供的资源约束并行机调度实例运行模型。

INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)

solution_rcpms = main(INSTANCE_DIR / "instance.txt", time_limit=1)
solution_rcpms.feasible