并行机调度问题 (PMS)¶

类别： 排程

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

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

问题描述¶

并行机调度问题 (PMS) 是调度文献中的经典问题。它由在一组并行机上调度任务组成。在最基本的表述中，所有机器都是相同的（即任务可以在任何机器上调度），并且每个任务只能由一台机器处理。

目标是找到一个使 makespan（所有任务的最大完成时间）最小的调度方案。这种并行机调度问题 (PMS) 的变体也被称为 P || Cmax，其中 P 表示存在相同的并行机，而 Cmax 表示优化准则是 makespan。

建模要点¶

添加 set 决策变量建模任务到资源的分配
定义 lambda 函数来计算每台机器的 makespan

数据¶

为了说明此示例，我们从文献中选取了一组实例。并行机调度 (PMS) 实例的格式如下：

第一行：
任务数量；
机器数量；
第二行：与每个任务相关联的长度。

建模思路¶

并行机调度 (PMS) 的 OptAgent 模型使用 set 决策变量。对每台机器，我们定义一个 set 变量表示分配给该机器的任务集合。我们将 set 变量约束为构成一个 partition，确保每个任务恰好在一台机器上调度。

我们使用对 set 的可变参数 sum 算子和一个 lambda 函数来计算一台机器的 makespan，该函数返回与任何任务索引相关联的长度。请注意，求和中的项数在搜索过程中会动态变化，set 的大小也会变化。

我们使用 max 算子从每台机器各自的 makespan 中提取全局 makespan。该表达式定义了搜索过程中需要最小化的目标函数。

Python 实现¶

from pathlib import Path

from optagent import OptModel, solve


def read_instance(filename):
    lines = Path(filename).read_text(encoding="utf-8").splitlines()
    header = lines[0].split()
    n_tasks = int(header[2])
    n_machines = int(header[3])
    task_lengths = [int(value) for value in lines[1].split() if value != "0"]
    if len(task_lengths) != n_tasks:
        raise ValueError("task length count does not match the instance header")
    return n_tasks, n_machines, task_lengths


def main(input_file, output_file=None, time_limit=5):
    n_tasks, n_machines, task_lengths = read_instance(input_file)
    makespan_lb = sum(task_lengths) // n_machines
    model = OptModel()

    # machine_tasks[k] contains exactly the tasks assigned to machine k.
    machine_tasks = [
        model.set(n_tasks)
        for k in range(n_machines)
    ]
    model.constraint(model.partition(machine_tasks))

    lengths = model.array(task_lengths)
    lengths_lambda = model.lambda_function(lambda task: lengths[task // 1])
    machine_makespan = [
        model.sum(tasks, lengths_lambda) for tasks in machine_tasks
    ]
    makespan = model.max(machine_makespan)
    model.minimize(makespan)

    solution = solve(
        model,
        time_limit_s=float(time_limit),
        objective_threshold={0: makespan_lb},
    )
    if not solution.feasible:
        print(f"No feasible schedule found; Status = {solution.feasible}")
        return solution

    lines = []
    for machine in range(n_machines):
        items = machine_tasks[machine].value
        line = (
            f"Makespan machine {machine}: {machine_makespan[machine].value} | "
            f"Items: {' '.join(map(str, items))}"
        )
        print(line)
        lines.append(line)
    if output_file is not None:
        Path(output_file).write_text("\n".join(lines) + "\n", encoding="utf-8")
    return solution

本地运行¶

在 notebook 所在目录执行以下 cell，即可调用并行机调度实例。

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

solution_pms = main(INSTANCE_DIR / "p_cmax-n120-m3.txt", time_limit=1)