在本教程中，我们实现 Rebalancer，即 Meta 的分配求解器，它拥有 C++ 核心和 Python 绑定。我们描述对象、容器、维度、作用域和分区，添加约束和目标，然后选择局部搜索或精确的混合整数求解器 HiGHS。我们构建一个小型数据中心，排空一台繁忙的主机，并在玩具模型、互斥主机、合成集群以及 60 个已发布的 OR-Library 实例上比较这两种求解器。我们通过重新统计返回的分配结果来评判每个方案，而不是依赖求解器的标志位。在此构建版本中，在 24 次求解中，有 14 次求解的 solved 标志读取为 True，但仍然存在内存超出容量的问题。局部搜索下的 HARD 返回了违规的起始状态且未抛出异常；拼错规格键会被静默忽略，而枚举名称则会被拒绝。计时数据来自配备 macOS wheel 的 Apple M4，Colab 的 Linux wheel 计时可能有所不同。
Copy Codeimport os
import re
import sys
import json
import math
import time
import random
import hashlib
import platform
import tempfile
import statistics
import traceback
import subprocess
import collections
import urllib.request
RESULTS = {} # step -> headline (or the failure), saved to results.json after every step
CLAIMS = [] # (step, claim, held) rows for the ledger in the last cell
SOLVE_SECONDS = collections.Counter() # step -> seconds spent inside solve()
CURRENT = ["1"]
def banner(title):
print("\n" + "=" * 78)
print(title)
print("=" * 78)
def section(name):
def wrap(fn):
def run(*a, **kw):
CURRENT[0] = name.split(".")[0]
banner(name)
t0 = time.time()
try:
out = fn(*a, **kw)
RESULTS[name] = out if isinstance(out, str) else "ok"
return out
except Exception as e:
RESULTS[name] = f"SKIPPED / FAILED -> {type(e).__name__}: {e}"
print(f"\n[!] {name} did not complete: {type(e).__name__}: {e}")
traceback.print_exc(limit=3)
return None
finally:
print(f"\n[{time.time() - t0:.1f} s]")
json.dump({"results": RESULTS, "claims": CLAIMS, "solve_seconds": SOLVE_SECONDS},
open("results.json", "w"), indent=1)
return run
return wrap
def claim(text, held):
CLAIMS.append((CURRENT[0], text, bool(held)))
return held
banner("1. Install rebalancer 1.0.4, check the build, and solve one small model with both solvers")
assert (3, 12) <= sys.version_info[:2] <= (3, 14), "the 1.0.4 wheels exist for Python 3.12, 3.13 and 3.14 only"
assert (sys.platform.startswith("linux") and platform.machine() == "x86_64") or \
(sys.platform == "darwin" and platform.machine() == "arm64"), f"no rebalancer 1.0.4 wheel for {sys.platform} {platform.machine()}"
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "--retries", "5", "--timeout", "120",
"rebalancer==1.0.4"], check=True)
import importlib.metadata as md
import rebalancer.specs as S
from rebalancer import ProblemSolver
# The binding takes every enum as its Thrift integer. The type stubs advertise the names ("HARD", "HIGHS"), but a name
# raises '"" is not a valid int', and an out-of-range integer is accepted silently, so the integers live in one table.
POLICY = {"DEFAULT": 1, "HARD": 2, "SOFT": 3}
LIMIT = {"ABSOLUTE": 1, "RELATIVE": 2}
FORMULA = {"LINEAR": 1, "SQUARES": 2, "MAX": 3, "IDEAL": 4}
PACKAGE = {"XPRESS": 0, "GUROBI": 1, "HIGHS": 2}
MAX_BOUND = 1
END = {1: "OPTIMAL", 2: "HIT_TIME_LIMIT", 3: "HIT_MOVE_LIMIT", 4: "HIT_STOP_CONSTRAINT", 5: "HIT_PLATEAU_TIME",
6: "NO_SOLUTION_FOUND", 7: "HIT_STAGE_MOVE_LIMIT", 8: "HIT_EXCEPTION", 9: "UNABLE_TO_FIND_MORE_MOVES",
10: "HIT_MIN_CYCLE_OBJECTIVE_IMPROVEMENT", 11: "UNABLE_TO_FIND_IMPROVING_MOVES"}
NESTED = {"limit": "Limit", "conflictInfoList": "ScopeItemConflictInfo",
"conflictingScopeItemsWithOverlap": "ConflictingScopeItemInfo"}
def check_keys(name, fields, where):
cls = getattr(S, name)
unknown = set(fields) - (cls.__required_keys__ | cls.__optional_keys__)
if unknown:
raise KeyError(f"{where} has no field {sorted(unknown)}; the binding would ignore it silently")
for key, value in fields.items():
sub = "AssignmentAffinity" if (where, key) == ("assignmentAffinitiesSpec", "affinities") else NESTED.get(key)
for item in (value if isinstance(value, list) else [value]):
if sub and isinstance(item, dict):
check_keys(sub, item, f"{where}.{key}")
def spec(arm, **fields):
"""One union arm, e.g. spec("capacitySpec", name=..., scope=..., dimension=...).
The binding ignores unknown keys without a warning, so the keys are checked against the TypedDicts in rebalancer.specs:
the arm's own keys and the nested limit, affinities and conflict lists used in this notebook."""
check_keys(arm[0].upper() + arm[1:], fields, arm)
return {arm: fields}
MOVES = {"single": {"singleMoveTypeSpec": {}}, "swap": {"swapMoveTypeSpec": {}},
"triple": {"tripleLoopMoveTypeSpec": {}}, "kl": {"klSearchMoveTypeSpec": {}}}
def local_search(seconds=2, seed=0, moves=("single", "swap"), **extra):
return spec("localSearchSolverSpec", moveTypeList=[MOVES[m] for m in moves], solveTime=int(seconds),
randomSeed=seed, **extra)
def highs(seconds=5):
# solverPackage 2 is HiGHS, the only MIP solver compiled into the pip wheel. suppressLogs removes a banner that the
# optimal path otherwise prints to stdout.
return spec("optimalSolverSpec", solverPackage=PACKAGE["HIGHS"], solveTime=int(seconds), suppressLogs=True)
def new_problem(assignment, objects="task", containers="host"):
p = ProblemSolver(service_name="rebalancer_tutorial", service_scope="marktechpost")
p.set_log_level("ERROR") # silences ~50 log lines per solve; never "OFF", which exits the process
p.set_object_name(objects).set_container_name(containers) # names first, then the assignment, then dimensions
p.set_assignment(assignment) # every container must be listed, empty ones too
return p
def solve(p):
t = time.perf_counter()
sol = p.solve()
dt = time.perf_counter() - t
SOLVE_SECONDS[CURRENT[0]] += dt
return sol, dt
# Independent checks: everything below reads only the returned assignment, never the solver's own verdict.
def where(container_to_objects):
return {o: c for c, objs in container_to_objects.items() for o in objs}
def by_container(object_to_container, containers):
out = {c: [] for c in containers}
for o, c in sorted(object_to_container.items()):
out[c].append(o)
return out
def load(object_to_container, values):
out = collections.Counter()
for o, c in object_to_container.items():
out[c] += values.get(o, 0.0)
return out
def excess(object_to_container, values, caps):
L = load(object_to_container, values)
return {c: max(0.0, L[c] - caps[c]) for c in caps}
def moved(before, after):
return sum(before[o] != after[o] for o in before)
def ends(sol):
return "/".join(END.get(r["endReason"], str(r["endReason"])) for r in sol["solverSummaries"])
def goals(sol):
return [round(g["value"], 6) for g in sol["finalGlobalObjective"]["goals"]]
def clean(e, n=90, typed=True):
msg = re.sub(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", "<id>", str(e)).strip()[:n]
return f"{type(e).__name__}: {msg}" if typed else msg
def cpu_name():
if os.path.exists("/proc/cpuinfo"):
for line in open("/proc/cpuinfo"):
if line.startswith("model name"):
return line.split(":", 1)[1].strip()
if sys.platform == "darwin":
r = subprocess.run(["sysctl", "-n", "machdep.cpu.brand_string"], capture_output=True, text=True)
if r.stdout.strip():
return r.stdout.strip()
return platform.processor() or platform.machine()
def ram_gb():
if os.path.exists("/proc/meminfo"):
kb = next(int(l.split()[1]) for l in open("/proc/meminfo") if l.startswith("MemTotal"))
return kb / 1024 ** 2
r = subprocess.run(["sysctl", "-n", "hw.memsize"], capture_output=True, text=True)
return int(r.stdout) / 1024 ** 3 if r.stdout.strip() else float("nan")
VERSION = md.version("rebalancer")
HW = f"{cpu_name()}, {os.cpu_count()} logical CPUs, {ram_gb():.0f} GB RAM"
print(f"rebalancer {VERSION} python {platform.python_version()} {platform.system()} {platform.machine()}")
print(f"hardware: {HW}")
assert VERSION == "1.0.4", "this notebook is written against rebalancer 1.0.4; restart the runtime after installing"
def two_tasks(solver_spec, policy=None):
# Two 10 GB tasks on host0, two hosts of 15 GB: the start breaks capacity and one task has to move.
p = new_problem({"host0": ["t0", "t1"], "host1": []})
p.add_object_dimension("memory", {"t0": 10.0, "t1": 10.0})
p.add_container_dimension("memory", {}, default_value=15.0)
kw = {} if policy is None else {"policy": policy}
p.add_constraint(spec("capacitySpec", name="memory_capacity", scope="host", dimension="memory"), **kw)
p.add_solver(solver_spec)
return p
GATE = {}
for label, make in [("local search", lambda: two_tasks(local_search(2))),
("HiGHS (solverPackage=2)", lambda: two_tasks(highs(2)))]:
try:
sol, dt = solve(make())
per_host = load(sol["assignment"], {"t0": 10.0, "t1": 10.0})
GATE[label] = sol
print(f"{label:<28} -> host loads {dict(sorted(per_host.items()))} GB, endReason {ends(sol)}, "
f"{dt * 1000:.1f} ms")
except Exception as e:
GATE[label] = None
print(f"{label:<28} -> FAILED {clean(e)}")
print("\nWhat the build refuses (each is caught here; the message is the binding's own):")
for label, make in [("OptimalSolverSpec() default", lambda: two_tasks(spec("optimalSolverSpec", suppressLogs=True))),
("solverPackage=1 (Gurobi)", lambda: two_tasks(spec("optimalSolverSpec", solverPackage=1,
suppressLogs=True))),
('policy="HARD" (a name)', lambda: two_tasks(local_search(2), policy="HARD")),
('solverPackage="HIGHS"', lambda: two_tasks({"optimalSolverSpec": {"solverPackage": "HIGHS"}}))]:
try:
solve(make())
print(f" {label:<28} -> accepted")
except Exception as e:
print(f" {label:<28} -> {clean(e, 60)}")
try:
spec("localSearchSolverSpec", timeLimitMs=500)
except KeyError as e:
print(f" spec(..., timeLimitMs=500) -> {e.args[0]}")
def gate_ok(sol): # solved means OPTIMAL and, by recount, no host above its 15 GB
return (sol is not None and ends(sol) == "OPTIMAL" and len(sol["assignment"]) == 2
and max(load(sol["assignment"], {"t0": 10.0, "t1": 10.0}).values()) <= 15.0)
HIGHS_OK = gate_ok(GATE["HiGHS (solverPackage=2)"])
if not gate_ok(GATE["local search"]):
raise RuntimeError("local search did not solve the two-task model; nothing below can run on this runtime")
print(f"\nGATE: local search solved; HiGHS {'solved' if HIGHS_OK else 'FAILED (steps that need it will say so)'}")
RESULTS["1. Install, check the build, solve one small model with both solvers"] = f"rebalancer {VERSION}, Python {platform.python_version()}, {HW}; HiGHS {'ok' if HIGHS_OK else 'failed'}"
我们安装 rebalancer 1.0.4 并打印 Python 版本和硬件信息，因为所有计时都取决于它们。该绑定拒绝枚举名称并静默忽略未知的规格键，因此笔记本从一个表中传入整数并在求解前检查每个规格。一个门槛测试随后用两个求解器求解一个双任务模型，各自移动一个任务，局部搜索用时 0.8 ms，HiGHS 用时 0.9 ms。该单元还确认此构建版本拒绝默认的 XPRESS 求解器和 Gurobi 包。
Copy CodeREADME_TASKS = {f"task{i}": 10.0 for i in range(4)}
README_CAPS = {"host0": 20.0, "host1": 20.0}
def readme_model(start, solver_spec):
p = new_problem(start)
p.add_object_dimension("memory", README_TASKS)
p.add_container_dimension("memory", {}, default_value=20.0)
# No limit given: the capacity limit is RELATIVE 1.0, i.e. 100% of each host's memory dimension.
p.add_constraint(spec("capacitySpec", name="memory_capacity", scope="host", dimension="memory"))
p.add_solver(solver_spec)
return p
@section("2. The README model: four 10 GB tasks, two 20 GB hosts, and a recount instead of the solver's verdict")
def readme_step():
start = {"host0": ["task0", "task1", "task2"], "host1": ["task3"]}
sol, dt = solve(readme_model(start, local_search(2)))
before, after = where(start), sol["assignment"]
ex0, ex1 = excess(before, README_TASKS, README_CAPS), excess(after, README_TASKS, README_CAPS)
l0, l1 = load(before, README_TASKS), load(after, README_TASKS)
ic, fc = sol["initialConstraint"], sol["finalConstraint"]
print(f"start: {start}")
print(f"final: {by_container(after, README_CAPS)}")
print(f"host loads {l0['host0']:.0f}/{l0['host1']:.0f} GB -> {l1['host0']:.0f}/{l1['host1']:.0f} GB "
f"against 20 GB; recounted excess {sum(ex0.values()):.0f} -> {sum(ex1.values()):.0f} GB")
print(f"solver's own report: brokenCount {ic['brokenCount']} -> {fc['brokenCount']}, brokenVal "
f"{ic['brokenVal']:g} -> {fc['brokenVal']:g}, solved={fc['solved']}")
print(f"tasks moved (diff against the start): {moved(before, after)} movesSummary entries: "
f"{len(sol['movesSummary'])} endReason {ends(sol)} solve {dt * 1000:.1f} ms")
print(f"result keys: {', '.join(sorted(sol))}")
balanced = {"host0": ["task0", "task1"], "host1": ["task2", "task3"]}
sol_b, _ = solve(readme_model(balanced, local_search(2)))
sol_z, _ = solve(readme_model(start, local_search(0)))
mb, mz = moved(where(balanced), sol_b["assignment"]), moved(before, sol_z["assignment"])
exz = sum(excess(sol_z["assignment"], README_TASKS, README_CAPS).values())
print(f"\ncontrol, start already 2-2: {mb} tasks moved, endReason {ends(sol_b)}")
print(f"control, solveTime=0 on the 3-1 start: {mz} tasks moved, endReason {ends(sol_z)}, recounted excess "
f"{exz:.0f} GB, solved={sol_z['finalConstraint']['solved']} <- 0 means no time, not 'no limit'")
claim("README model: recounted excess reaches 0 GB", sum(ex1.values()) == 0)
claim("README model: a 2-2 start is left alone", mb == 0)
claim("README model: solveTime=0 makes no moves", mz == 0)
return (f"3-1 start -> {l1['host0']:.0f}/{l1['host1']:.0f} GB, excess {sum(ex0.values()):.0f} -> "
f"{sum(ex1.values()):.0f} GB, {moved(before, after)} task moved; solveTime=0 moved {mz}")
readme_step()
我们求解 README 示例：两个 20 GB 主机上的四个 10 GB 任务，其中一台主机上有三个任务，并重新统计结果而不是相信求解器。局部搜索在 0.3 ms 内移动一个任务，重新统计将超出量从 10 GB 降至 0。求解器自身的 brokenCount 也从 1 降至 0，endReason 读取为 OPTIMAL。一个对照实验说明了该标志位的问题：当 solveTime=0 时，求解器不移动任何任务，仍然留下 10 GB 超出容量，却报告 solved=True。
Copy CodeHOSTS = [f"host{i}" for i in range(8)]
RACK = {h: f"rack{i // 2}" for i, h in enumerate(HOSTS)} # Python add_scope maps container -> scope item
JOB_OF = {f"job{j}_t{k}": f"job{j}" for j in "ABCD" for k in range(4)}
TASK_CPU, TASK_MEM, HOST_CPU, HOST_MEM = 4.0, 16.0, 16.0, 64.0
DC_START = {h: [] for h in HOSTS}
DC_START["host0"] = [t for t, j in JOB_OF.items() if j in ("jobA", "jobB")]
DC_START["host1"] = [t for t, j in JOB_OF.items() if j == "jobC"]
DC_START["host2"] = [t for t, j in JOB_OF.items() if j == "jobD"]
def datacenter(solver_spec, start=DC_START, rack_rule=True, formula="LINEAR", drain=None, movement_tuple=None,
limit_type="ABSOLUTE"):
p = new_problem(start)
p.add_partition("job", {j: [t for t in JOB_OF if JOB_OF[t] == j] for j in sorted(set(JOB_OF.values()))})
p.add_scope("rack", RACK)
p.add_object_dimension("cpu", {t: TASK_CPU for t in JOB_OF})
p.add_object_dimension("mem", {t: TASK_MEM for t in JOB_OF})
p.add_container_dimension("cpu", {}, default_value=HOST_CPU)
p.add_container_dimension("mem", {}, default_value=HOST_MEM)
p.add_constraint(spec("capacitySpec", name="cpu_capacity", scope="host", dimension="cpu", bound=MAX_BOUND))
p.add_constraint(spec("capacitySpec", name="mem_capacity", scope="host", dimension="mem", bound=MAX_BOUND))
if rack_rule: # at most one task of each job per rack, as an ABSOLUTE limit (step 3 shows what RELATIVE does)
p.add_constraint(spec("groupCountSpec", name="one_task_per_job_per_rack", scope="rack", partitionName="job",
bound=MAX_BOUND, limit={"type": LIMIT[limit_type], "globalLimit": 1.0}))
if drain:
p.add_constraint(spec("toFreeSpec", name=f"drain_{drain}", containers=[drain]))
p.add_goal(spec("balanceSpec", name="cpu_balance", scope="host", dimension="cpu", formula=FORMULA[formula]),
weight=1.0, tuple_pos=0)
if movement_tuple is not None:
p.add_goal(spec("minimizeMovementSpec", name="tasks_moved", scope="host", dimension="task_count",
doNotNormalize=True), weight=1.0, tuple_pos=movement_tuple)
p.add_solver(solver_spec)
return p
def dc_check(o2c, drain=None):
cpu, mem = load(o2c, {t: TASK_CPU for t in JOB_OF}), load(o2c, {t: TASK_MEM for t in JOB_OF})
per_rack_job = collections.Counter((RACK[o2c[t]], JOB_OF[t]) for t in JOB_OF)
live = [h for h in HOSTS if h != drain]
util = [cpu[h] / HOST_CPU for h in live]
return {"cpu": sum(max(0.0, cpu[h] - HOST_CPU) for h in HOSTS),
"mem": sum(max(0.0, mem[h] - HOST_MEM) for h in HOSTS),
"rack": sum(n - 1 for n in per_rack_job.values() if n > 1),
"per_host": [sum(1 for t in JOB_OF if o2c[t] == h) for h in HOSTS],
"umax": max(util), "umin": min(util), "complete": sorted(o2c) == sorted(JOB_OF)}
@section("3. A small datacenter: racks, jobs, CPU and memory capacity, one task per job per rack, CPU balance")
def datacenter_step():
before = where(DC_START)
c0 = dc_check(before)
print(f"start: tasks per host {c0['per_host']} (host0 holds jobA and jobB: {c0['umax']:.0%} CPU)")
print(f"start, recounted: CPU excess {c0['cpu']:.0f}, memory excess {c0['mem']:.0f}, "
f"extra tasks of a job sharing a rack {c0['rack']}")
runs = []
for _ in range(3): # three fresh builds of the same model
sol, dt = solve(datacenter(local_search(3)))
runs.append((sol, dt))
sol = runs[0][0]
c1 = dc_check(sol["assignment"])
print(f"\nfinal (build 1): tasks per host {c1['per_host']}, CPU use {c1['umin']:.0%}..{c1['umax']:.0%}")
print(f"final, recounted: CPU excess {c1['cpu']:.0f}, memory excess {c1['mem']:.0f}, rack duplicates "
f"{c1['rack']}, every task placed once: {c1['complete']}")
print(f"tasks moved {moved(before, sol['assignment'])}, endReason {ends(sol)}, balance objective "
f"{goals(sol)}, solver brokenCount {sol['initialConstraint']['brokenCount']} -> "
f"{sol['finalConstraint']['brokenCount']}")
same = [goals(s) == goals(sol) and moved(before, s["assignment"]) == moved(before, sol["assignment"])
for s, _ in runs]
ms = statistics.median(dt for _, dt in runs) * 1000
print(f"3 fresh builds: same objective and move count in {sum(same)} of 3; median solve {ms:.1f} ms")
sol_c, _ = solve(datacenter(local_search(3), rack_rule=False))
cc = dc_check(sol_c["assignment"])
print(f"\ncontrol without the rack rule: rack duplicates {cc['rack']}, tasks per host {cc['per_host']} "
f"(the checker sees the rule only when the solver is not enforcing it)")
sol_r, _ = solve(datacenter(local_search(3), limit_type="RELATIVE"))
cr = dc_check(sol_r["assignment"])
print(f"control, the same rule with limit type RELATIVE: brokenCount at the start {sol_r['initialConstraint']['brokenCount']}"
f" (ABSOLUTE: {sol['initialConstraint']['brokenCount']}), rack duplicates after the solve {cr['rack']}")
ok = c1["cpu"] == c1["mem"] == c1["rack"] == 0 and c1["complete"]
claim("datacenter: every constraint holds by recount", ok)
claim("datacenter: removing the rack rule brings back rack duplicates", cc["rack"] > 0)
claim("datacenter: a RELATIVE group limit leaves rack duplicates", cr["rack"] > 0)
return (f"violations CPU/mem/rack {c0['cpu']:.0f}/{c0['mem']:.0f}/{c0['rack']} -> {c1['cpu']:.0f}/"
f"{c1['mem']:.0f}/{c1['rack']}, {moved(before, sol['assignment'])} tasks moved, {ms:.1f} ms; "
f"without the rule {cc['rack']} rack duplicates, with a RELATIVE limit {cr['rack']}")
datacenter_step()
我们构建四个机架中的八台主机，四个作业各含四个任务，并从 host0 处于 200% CPU 和 12 个机架重复项开始。我们添加 CPU 和内存容量约束、每作业每机架一个任务的规则，以及一个 CPU 均衡目标。求解器移动 13 个任务，使每台主机留下两个任务且无违规，三次全新构建的中位求解时间为 2.5 ms。没有机架规则时，重新统计发现 5 个机架重复项，而对其施加 RELATIVE 限制同样以 5 结束，因此该限制在此不起作用。
Copy Code@section("4. What 'solved' means: the constraint policies under local search and HiGHS, checked by recount")
def policy_step():
cases = {"F (fixable)": ({f"t{i}": 10.0 for i in range(4)}, {"host0": ["t0", "t1", "t2"], "host1": ["t3"]}),
"I (infeasible)": ({f"t{i}": 20.0 for i in range(3)}, {"host0": ["t0", "t1", "t2"], "host1": []})}
caps = {"host0": 20.0, "host1": 20.0}
def build(sizes, start, policy, solver_spec):
p = new_problem(start)
p.add_object_dimension("memory", sizes)
p.add_container_dimension("memory", {}, default_value=20.0)
p.add_constraint(spec("capacitySpec", name="memory_capacity", scope="host", dimension="memory"),
policy=POLICY[policy])
p.add_solver(solver_spec)
return p
rows, stable, solved_but_broken, cells = [], [], 0, 0
print(f"{'case':<15}{'policy':<9}{'solver':<14}{'outcome':<36}{'moved':>6}{'broken':>9}{'solved':>8}"
f"{'brokenVal':>11}{'excess GB':>11}")
for case, (sizes, start) in cases.items():
for policy in ("DEFAULT", "HARD", "SOFT"):
for solver_name, solver_spec, reps in [("local search", local_search(1), 3), ("HiGHS", highs(2), 1)]:
outs = []
for _ in range(reps):
cells += 1
try:
sol, _ = solve(build(sizes, start, policy, solver_spec))
ex = sum(excess(sol["assignment"], sizes, caps).values())
fc, ic = sol["finalConstraint"], sol["initialConstraint"]
solved_but_broken += fc["solved"] and ex > 0
outs.append({"out": "returned, " + ends(sol), "moved": moved(where(start), sol["assignment"]),
"broken": f"{ic['brokenCount']}->{fc['brokenCount']}", "solved": fc["solved"],
"bv": fc["brokenVal"], "ex": ex, "sol": sol, "sizes": sizes})
except Exception as e:
outs.append({"out": "raised: " + clean(e, 30, typed=False), "moved": "-", "broken": "-",
"solved": "-",
"bv": None, "ex": None})
o = outs[0]
if reps > 1:
stable.append(len({(x["out"], x["moved"], x["broken"], x["bv"]) for x in outs}) == 1)
bv = "-" if o["bv"] is None else f"{o['bv']:g}"
exs = "-" if o["ex"] is None else f"{o['ex']:.0f}"
print(f"{case:<15}{policy:<9}{solver_name:<14}{o['out']:<36}{o['moved']!s:>6}{o['broken']:>9}"
f"{o['solved']!s:>8}{bv:>11}{exs:>11}")
rows.append((case, policy, solver_name, o))
print(f"\nlocal-search repeats (3 per row) identical in {sum(stable)} of {len(stable)} rows")
print(f"cells where solved=True although the recount still finds excess memory: {solved_but_broken} of {cells}")
# What does brokenVal measure? Three candidate formulas, tested on the first returned result of every row above and
# on four starts read before any move: two hosts over capacity, and three starts with hosts of different sizes.
probes = [({"host0": ["t0", "t1", "t2", "t3"], "host1": ["t4", "t5", "t6"], "host2": [], "host3": []},
{"host0": 20.0, "host1": 20.0, "host2": 20.0, "host3": 20.0}),
({"host0": ["t0", "t1", "t2"], "host1": []}, {"host0": 20.0, "host1": 40.0}),
({"host0": ["t0", "t1", "t2", "t3"], "host1": ["t4", "t5", "t6", "t7"]}, {"host0": 20.0, "host1": 30.0}),
({"host0": ["t0", "t1", "t2"], "host1": ["t3", "t4", "t5", "t6"], "host2": ["t7"]},
{"host0": 20.0, "host1": 30.0, "host2": 100.0})]
samples = [(o["sol"]["assignment"], o["sizes"], caps, o["bv"]) for *_, o in rows if o["bv"] is not None]
for start, pcaps in probes:
sizes = {t: 10.0 for objs in start.values() for t in objs}
p = new_problem(start)
p.add_object_dimension("memory", sizes)
p.add_container_dimension("memory", pcaps, default_value=0.0)
p.add_constraint(spec("capacitySpec", name="memory_capacity", scope="host", dimension="memory"))
p.add_solver(local_search(0))
probe, _ = solve(p)
bv = probe["initialConstraint"]["brokenVal"]
samples.append((where(start), sizes, pcaps, bv))
ex = excess(where(start), sizes, pcaps)
print(f"start {[f'{load(where(start), sizes)[h]:.0f}/{c:.0f}' for h, c in pcaps.items()]} GB (load/capacity): "
f"excess {sum(ex.values()):.0f} GB, brokenVal {bv:.4g}")
cands = {"total excess in GB": lambda e, c: sum(e.values()),
"sum over hosts of excess / that host's capacity": lambda e, c: sum(e[h] / c[h] for h in c),
"total excess / mean capacity of all hosts": lambda e, c: sum(e.values()) / (sum(c.values()) / len(c))}
match = [name for name, f in cands.items()
if all(abs(f(excess(a, s, c), c) - bv) < 1e-9 for a, s, c, bv in samples)]
print(f"formula that matches brokenVal in all {len(samples)} samples: {match or 'NONE'}")
hard_ls_f = next(o for c, p_, s, o in rows if c.startswith("F") and p_ == "HARD" and s == "local search")
hard_h_i = next(o for c, p_, s, o in rows if c.startswith("I") and p_ == "HARD" and s == "HiGHS")
claim("policies: HARD + local search leaves the fixable start unchanged without raising",
hard_ls_f["moved"] == 0 and hard_ls_f["ex"] and hard_ls_f["ex"] > 0)
claim("policies: HARD + HiGHS raises 'infeasible' on the infeasible start",
hard_h_i["out"].startswith("raised") and "infeasible" in hard_h_i["out"])
claim("policies: solved=True appears with excess remaining", solved_but_broken > 0)
return (f"solved=True while excess remained in {solved_but_broken} of {cells} solves; HARD + local search moved "
f"{hard_ls_f['moved']} tasks on the fixable start, no error; HARD + HiGHS on the infeasible start "
f"{hard_h_i['out']}; brokenVal = {', '.join(match) or 'no candidate'}")
policy_step()
我们在每种约束策略和两个求解器下测试一个可修复的起始状态 F 和一个不可行的起始状态 I，并重新统计每个结果。局部搜索下的 HARD 对两个起始状态都原样返回，并报告 OPTIMAL 和 solved=True，而 HiGHS 修复了 F 并对 I 抛出不可行错误。在这 24 次求解中，有 14 次求解的 solved 读取为 True，但仍然存在内存超出容量的问题。在全部 15 个样本中，求解器的 brokenVal 都等于总超出量除以平均主机容量。请读取 brokenCount 或重新统计，而不要相信 OPTIMAL。
Copy Code@section("5. Goal tuples: a lower tuple always wins, weights only trade off inside one tuple")
def tuple_step():
# Part A: one task would like to move (affinity 1) and every move costs 1000 (MinimizeMovement on task_count).
def affinity_vs_movement(aff_w, move_w, move_tuple):
p = new_problem({"host0": ["t0", "t1"], "host1": []})
p.add_goal(spec("assignmentAffinitiesSpec", name="t0_prefers_host1", scope="host",
affinities=[{"objectName": "t0", "scopeItemName": "host1", "affinity": 1.0}]),
weight=aff_w, tuple_pos=0)
p.add_goal(spec("minimizeMovementSpec", name="moves", scope="host", dimension="task_count",
doNotNormalize=True), weight=move_w, tuple_pos=move_tuple)
p.add_solver(local_search(2))
sol, _ = solve(p)
return sol["assignment"]["t0"] == "host1", goals(sol)
print("Part A, no constraints: affinity for t0 on host1 against a movement cost")
A = {}
for label, args in [("same tuple, affinity weight 1, movement weight 1000", (1.0, 1000.0, 0)),
("same tuple, affinity weight 1000, movement weight 1", (1000.0, 1.0, 0)),
("affinity in tuple 0, movement weight 1000 in tuple 1", (1.0, 1000.0, 1))]:
A[label] = affinity_vs_movement(*args)
print(f" {label:<55} t0 moved: {str(A[label][0]):<5} goals per tuple {A[label][1]}")
# Part B: a broken capacity constraint becomes a 'fix it' goal in tuple tuple_pos_if_broken (default 0).
sizes = {f"t{i}": 10.0 for i in range(4)}
def fixit(start, move_tuple, fix_tuple=None):
p = new_problem(start)
p.add_object_dimension("mem", sizes)
p.add_container_dimension("mem", {}, default_value=20.0)
kw = {} if fix_tuple is None else {"tuple_pos_if_broken": fix_tuple}
p.add_constraint(spec("capacitySpec", name="mem_capacity", scope="host", dimension="mem"), **kw)
p.add_goal(spec("minimizeMovementSpec", name="moves", scope="host", dimension="mem", doNotNormalize=True),
weight=1000.0, tuple_pos=move_tuple)
p.add_solver(local_search(2))
sol, _ = solve(p)
return sol
broken3 = {"host0": ["t0", "t1", "t2"], "host1": ["t3"]}
B = {"X1 movement tuple 0, fix-it tuple 0 (default)": fixit(broken3, 0),
"X2 movement tuple 0, fix-it tuple 1": fixit(broken3, 0, 1),
"X3 movement tuple 1, fix-it tuple 0": fixit(broken3, 1, 0),
"X4 feasible 2-2 start, movement tuple 0": fixit({"host0": ["t0", "t1"], "host1": ["t2", "t3"]}, 0)}
print("\nPart B, broken capacity (30 GB on a 20 GB host) plus a movement goal of weight 1000 on the moved GB")
EX = {k: sum(excess(sol["assignment"], sizes, {"host0": 20.0, "host1": 20.0}).values()) for k, sol in B.items()}
for label, sol in B.items():
ex = EX[label]
print(f" {label:<48} moves {len(sol['movesSummary'])}, recounted excess {ex:.0f} GB, brokenCount "
f"{sol['finalConstraint']['brokenCount']}, goals per tuple {goals(sol)}")
x2, x3 = B["X2 movement tuple 0, fix-it tuple 1"], B["X3 movement tuple 1, fix-it tuple 0"]
# The fix-it value against 10000 (invalidState) + 100 (invalidCost) x brokenVal, with brokenVal recounted as the total
# excess over the 20 GB mean capacity (step 4), on X2 and on a second start with 40 GB on host0 (X2b).
x2b = fixit({"host0": ["t0", "t1", "t2", "t3"], "host1": []}, 0, 1)
fix_rows = []
for label, sol, start_excess in (("X2", x2, EX["X2 movement tuple 0, fix-it tuple 1"]),
("X2b", x2b, sum(excess(x2b["assignment"], sizes, {"host0": 20.0, "host1": 20.0}).values()))):
bv = start_excess / 20.0
fix_rows.append(abs(goals(sol)[-1] - (10000 + 100 * bv)) < 1e-6 and len(sol["movesSummary"]) == 0)
print(f" {label}: {start_excess:.0f} GB over, no move; tuple-1 value {goals(sol)[-1]:g} vs 10000 + 100 x {bv:g} = "
f"{10000 + 100 * bv:g}")
gb_moved = 10.0 * moved(where(broken3), x3["assignment"])
move_rule = abs(goals(x3)[-1] - 1000.0 * gb_moved) < 1e-6
print(f" fix-it value = 10000 + 100 x brokenVal on both starts: {'confirmed' if all(fix_rows) else 'not confirmed'}")
print(f" X3's tuple-1 value {goals(x3)[-1]:g} vs weight 1000 x {gb_moved:g} GB moved (recounted): "
f"{'confirmed' if move_rule else 'not confirmed'}")
a = list(A.values())
claim("tuples: weight 1 vs 1000 in one tuple keeps t0 home; swapping weights moves it", not a[0][0] and a[1][0])
claim("tuples: a tuple-0 goal beats any weight in tuple 1", a[2][0])
claim("tuples: fix-it in a later tuple than movement (X2) leaves 10 GB of excess, no move",
EX["X2 movement tuple 0, fix-it tuple 1"] > 0 and len(x2["movesSummary"]) == 0)
claim("tuples: fix-it in an earlier tuple than movement (X3) removes the excess",
EX["X3 movement tuple 1, fix-it tuple 0"] == 0)
claim("tuples: the fix-it value is 10000 + 100 x the recounted brokenVal on two starts", all(fix_rows))
return (f"t0 moves with weights 1 vs 1000: {a[0][0]}, 1000 vs 1: {a[1][0]}, affinity one tuple above: {a[2][0]}; "
f"fix-it below movement: {len(x2['movesSummary'])} moves, brokenCount {x2['finalConstraint']['brokenCount']}; "
f"fix-it above: {len(x3['movesSummary'])} move, brokenCount {x3['finalConstraint']['brokenCount']}")
tuple_step()
较低的目标元组总是优先决定，而权重只在一个元组内部进行权衡。当两者都在元组 0 中时，亲和性的权重 1 输给移动的权重 1000，而交换权重则会使 t0 移动。当亲和性在元组 0、移动在元组 1 时，t0 仍然移动。当修复目标所在元组晚于移动目标时，会留下 10 GB 超出且没有移动，而相反的顺序则用一次移动将其修复。修复目标的值为 10000 加上 100 乘以 brokenVal，已在两个起始状态上验证。
Copy Code@section("6. Draining a busy host under a move budget, and what a misspelled key does")
def drain_step():
before = where(DC_START)
builds = [solve(datacenter(local_search(3), drain="host1")) for _ in range(3)]
sol0, dt0 = builds[0]
c = dc_check(sol0["assignment"], drain="host1")
r = len(sol0["movesSummary"])
left0 = c["per_host"][1]
print(f"drain host1 (starts with {len(DC_START['host1'])} jobC tasks), no budget: host1 holds {left0}, "
f"recounted CPU/mem/rack violations {c['cpu']:.0f}/{c['mem']:.0f}/{c['rack']}, tasks per host "
f"{c['per_host']}")
print(f" {r} moves ({moved(before, sol0['assignment'])} tasks changed host), endReason {ends(sol0)}, brokenCount "
f"{sol0['initialConstraint']['brokenCount']} -> {sol0['finalConstraint']['brokenCount']}, "
f"{dt0 * 1000:.1f} ms")
print(f" three fresh builds of this model: {[len(b['movesSummary']) for b, _ in builds]} moves, host1 left with "
f"{[dc_check(b['assignment'], drain='host1')['per_host'][1] for b, _ in builds]} tasks")
print(f" rack0 is now host0 alone: {c['per_host'][0]} tasks = {c['per_host'][0] * TASK_CPU:.0f} of "
f"{HOST_CPU:.0f} CPU, one per job")
budget_rows = {}
for k in (r - 1, max(1, r // 2)):
sol, _ = solve(datacenter(local_search(3, stopAfterMoves=k), drain="host1"))
ck = dc_check(sol["assignment"], drain="host1")
budget_rows[k] = (sol, ck)
print(f"stopAfterMoves={k:<3} -> {len(sol['movesSummary'])} moves, endReason {ends(sol)}, host1 still holds "
f"{ck['per_host'][1]}, recounted violations {ck['cpu']:.0f}/{ck['mem']:.0f}/{ck['rack']}, brokenCount "
f"{sol['finalConstraint']['brokenCount']}, solved={sol['finalConstraint']['solved']}")
typo = local_search(3)
typo["localSearchSolverSpec"]["stopAfterMove"] = max(1, r // 2) # bypasses spec(), as a hand-written dict would
sol_t, _ = solve(datacenter(typo, drain="host1"))
same = len(sol_t["movesSummary"]) > max(1, r // 2) and ends(sol_t) == "UNABLE_TO_FIND_MORE_MOVES"
print(f"misspelled 'stopAfterMove={max(1, r // 2)}' -> {len(sol_t['movesSummary'])} moves, endReason "
f"{ends(sol_t)}: {'ran to the end; the budget was ignored without an error' if same else 'stopped early'}")
try:
spec("localSearchSolverSpec", stopAfterMove=3)
except KeyError as e:
print(f"spec() catches it before the solve: {e.args[0]}")
half = budget_rows[max(1, r // 2)][0]
claim("drain: unbudgeted run empties host1 with no recounted violation",
left0 == 0 and c["cpu"] == c["mem"] == c["rack"] == 0)
claim("drain: stopAfterMoves stops early with endReason HIT_MOVE_LIMIT", "HIT_MOVE_LIMIT" in ends(half))
claim("drain: a misspelled budget key is ignored silently", same)
return (f"host1 emptied in {r} moves; stopAfterMoves={max(1, r // 2)} -> {ends(half)}, host1 holds "
f"{budget_rows[max(1, r // 2)][1]['per_host'][1]}; misspelled key ignored: {same}")
drain_step()
我们在 ToFree 约束下排空 host1，使 host0 独自留在 rack0 中，CPU 使用为 16 中的 16。在没有预算的情况下，局部搜索用 13 次移动清空 host1 且无违规，三次全新构建结果一致。stopAfterMoves 预算给出的是部分方案而非错误：在 12 次移动时仍有一个任务留在 host1 上，在 6 次移动时重新统计发现 4 CPU 和 16 GB 超出以及 6 个机架重复项，然而 solved 为 True。一个拼错的键 stopAfterMove 被静默忽略，求解运行了全部 13 次移动。
Copy Code@section("7. Exact (HiGHS) against local search: balance formulas on the datacenter, and mutually exclusive hosts")
def exact_step():
if not HIGHS_OK:
raise RuntimeError("HiGHS failed the gate in step 1")
before = where(DC_START)
ERR = {}
def row(label, make, builds):
sols = []
for _ in range(builds):
try:
sols.append(solve(make()))
except Exception as e:
ERR[label] = clean(e, 60)
print(f" {label:<44} raised {ERR[label]}")
return None
sol = sols[0][0]
c = dc_check(sol["assignment"])
ms = statistics.median(dt for _, dt in sols) * 1000
print(f" {label:<44} CPU spread {c['umax'] - c['umin']:.2f}, violations {c['cpu']:.0f}/{c['mem']:.0f}/"
f"{c['rack']}, moved {moved(before, sol['assignment']):>2}, {ends(sol):<26} median {ms:7.1f} ms "
f"over {builds}")
return sol, c
print("Datacenter model of step 3; spread = max - min CPU utilisation, recounted (objectives of different "
"formulas are not comparable)")
R = {}
R["LINEAR local"] = row("LINEAR, local search", lambda: datacenter(local_search(2)), 3)
R["LINEAR highs"] = row("LINEAR, HiGHS", lambda: datacenter(highs(6)), 2)
R["LINEAR highs+move"] = row("LINEAR, HiGHS, movement goal in tuple 1",
lambda: datacenter(highs(6), movement_tuple=1), 2)
R["SQUARES local"] = row("SQUARES, local search", lambda: datacenter(local_search(2), formula="SQUARES"), 3)
R["SQUARES highs"] = row("SQUARES, HiGHS", lambda: datacenter(highs(6), formula="SQUARES"), 2)
R["IDEAL local"] = row("IDEAL, local search", lambda: datacenter(local_search(2), formula="IDEAL"), 1)
R["IDEAL highs"] = row("IDEAL, HiGHS", lambda: datacenter(highs(3), formula="IDEAL"), 1)
if R["IDEAL local"]:
print(f" IDEAL's own objective at that spread: {goals(R['IDEAL local'][0])} (IDEAL sums each host's squared "
f"utilisation: 8 hosts x 0.5^2 = {8 * 0.5 ** 2:g})")
# ExclusiveScopeItems: three hosts that exclude each other pairwise; one task on each.
print("\nExclusiveScopeItems: host1, host2, host3 pairwise exclusive, one task on each (a broken start)")
pairs = [("host1", "host2"), ("host1", "host3"), ("host2", "host3")]
conflicts = [{"scopeItem": "host1", "conflictingScopeItemsWithOverlap": [{"conflictingScopeItem": "host2"},
{"conflictingScopeItem": "host3"}]},
{"scopeItem": "host2", "conflictingScopeItemsWithOverlap": [{"conflictingScopeItem": "host3"}]}]
def exclusive(start, solver_spec):
p = new_problem(start)
p.add_constraint(spec("exclusiveScopeItemsSpec", name="exclusive_hosts", scope="host",
dimension="task_count", conflictInfoList=conflicts))
p.add_solver(solver_spec)
sol, dt = solve(p)
used = set(sol["assignment"].values())
return sol, dt, sorted(used), sum(a in used and b in used for a, b in pairs)
three = {"host1": ["task1"], "host2": ["task2"], "host3": ["task3"]}
two = {"host1": ["task1", "task2"], "host2": [], "host3": ["task3"]} # the start after one move
ex_res = {}
for label, solver_spec in [("local search", local_search(2)), ("HiGHS", highs(6))]:
sol, dt, used, bad = exclusive(three, solver_spec)
ex_res[label] = bad
print(f" {label:<13} hosts in use {used}, conflicting pairs both in use {bad} of 3, brokenCount "
f"{sol['finalConstraint']['brokenCount']}, solved={sol['finalConstraint']['solved']}, {ends(sol)}, "
f"{dt * 1000:.1f} ms")
bv3 = exclusive(three, local_search(0))[0]["initialConstraint"]["brokenVal"]
after_one = [] # all six single moves from the three-host start
for task, src in (("task1", "host1"), ("task2", "host2"), ("task3", "host3")):
for dst in ("host1", "host2", "host3"):
if dst != src:
state = {h: [t for t in objs if t != task] for h, objs in three.items()}
state[dst] = state[dst] + [task]
after_one.append(exclusive(state, local_search(0))[0]["initialConstraint"]["brokenVal"])
sol2, _, used2, bad2 = exclusive(two, local_search(2))
print(f" why local search stops: the solver's brokenVal is {bv3:g} with three hosts in use (3 conflicting pairs); "
f"after each of the {len(after_one)} possible single moves (2 hosts, 1 pair) it is {sorted(set(after_one))}, so "
f"{'no single move lowers it' if min(after_one) >= bv3 else 'some single move lowers it'}")
print(f" control, local search started from the two-host state: {len(sol2['movesSummary'])} move(s), hosts in use "
f"{used2}, conflicting pairs {bad2}, {ends(sol2)}")
lin_l, lin_h, lin_hm = R["LINEAR local"], R["LINEAR highs"], R["LINEAR highs+move"]
claim("exact: IDEAL balance raises a HiGHS error", R["IDEAL highs"] is None and "HiGHS Error" in ERR.get("IDEAL, HiGHS", ""))
if lin_h and lin_hm:
claim("exact: a tuple-1 movement goal cuts HiGHS's moves without losing balance",
moved(before, lin_hm[0]["assignment"]) < moved(before, lin_h[0]["assignment"])
and lin_hm[1]["umax"] - lin_hm[1]["umin"] == lin_h[1]["umax"] - lin_h[1]["umin"])
claim("exact: HiGHS clears the exclusivity conflicts", ex_res.get("HiGHS") == 0)
claim("exact: local search leaves exclusivity conflicts from the three-host start", ex_res.get("local search", 0) > 0)
claim("exact: no single move from the three-host start lowers the solver's brokenVal", min(after_one) >= bv3)
claim("exact: from the two-host state local search clears them", bad2 == 0)
return (f"LINEAR balance: local search moved {moved(before, lin_l[0]['assignment'])} tasks, HiGHS "
f"{moved(before, lin_h[0]['assignment'])}, HiGHS with a movement tuple {moved(before, lin_hm[0]['assignment'])}"
f"; IDEAL under HiGHS raised: {R['IDEAL highs'] is None}; exclusivity conflicts left: local search "
f"{ex_res.get('local search')} (0 from a two-host start), HiGHS {ex_res.get('HiGHS')}")
exact_step()
我们将 HiGHS 与局部搜索在负载均衡上进行比较，均衡以重新统计的 CPU 分散度来衡量，因为不同公式之间的目标值不可比。在 LINEAR 公式下，两者都达到零分散度，但局部搜索以中位数 1.8 ms 移动了 13 个任务，而 HiGHS 在 7.8 ms 中移动了 16 个。IDEAL 在局部搜索下可以均衡，但在 HiGHS 下会引发错误。在三个互斥主机上，局部搜索停止时三个主机都在使用中，且每一对冲突仍然存在，因为没有任何单一移动能降低 brokenVal。HiGHS 在 5.9 ms 中清除了冲突。
复制代码CHILD = r'''
import sys, json, time, math, random, resource, hashlib, collections
from rebalancer import ProblemSolver
mode, N, M, cap, seed = sys.argv[1], int(sys.argv[2]), int(sys.argv[3]), int(sys.argv[4]), int(sys.argv[5])
p = ProblemSolver(service_name="rebalancer_tutorial", service_scope="scale")
p.set_log_level("ERROR")
single_swap = [{"singleMoveTypeSpec": {}}, {"swapMoveTypeSpec": {}}]
if mode == "scale":
rng = random.Random(seed)
cpu = [rng.randint(1, 8) for _ in range(N)]
mem = [rng.randint(1, 32) for _ in range(N)]
start = [min(M - 1, int(M * rng.random() ** 2)) for _ in range(N)] # skewed: low-index containers overloaded
cpu_cap, mem_cap = max(8, math.ceil(sum(cpu) / (0.6 * M))), max(32, math.ceil(sum(mem) / (0.6 * M)))
objs, conts = [f"o{i}" for i in range(N)], [f"c{j}" for j in range(M)]
assign = {c: [] for c in conts}
for i, j in enumerate(start):
assign[conts[j]].append(objs[i])
p.set_object_name("obj").set_container_name("cont").set_assignment(assign)
p.add_object_dimension("cpu", dict(zip(objs, map(float, cpu))))
p.add_object_dimension("mem", dict(zip(objs, map(float, mem))))
p.add_container_dimension("cpu", {}, default_value=float(cpu_cap))
p.add_container_dimension("mem", {}, default_value=float(mem_cap))
p.add_constraint({"capacitySpec": {"name": "cpu_cap", "scope": "cont", "dimension": "cpu", "bound": 1}})
p.add_constraint({"capacitySpec": {"name": "mem_cap", "scope": "cont", "dimension": "mem", "bound": 1}})
p.add_goal({"balanceSpec": {"name": "cpu_balance", "scope": "cont", "dimension": "cpu", "formula": 1}})
p.add_solver({"localSearchSolverSpec": {"moveTypeList": single_swap, "solveTime": cap, "randomSeed": seed}})
else: # the step-3 datacenter with HiGHS, to see whether tied optima come back identical in fresh processes
hosts = [f"host{i}" for i in range(8)]
job_of = {f"job{j}_t{k}": f"job{j}" for j in "ABCD" for k in range(4)}
assign = {h: [] for h in hosts}
assign["host0"] = [t for t, j in job_of.items() if j in ("jobA", "jobB")]
assign["host1"] = [t for t, j in job_of.items() if j == "jobC"]
assign["host2"] = [t for t, j in job_of.items() if j == "jobD"]
p.set_object_name("task").set_container_name("host").set_assignment(assign)
p.add_partition("job", {j: [t for t in job_of if job_of[t] == j] for j in sorted(set(job_of.values()))})
p.add_scope("rack", {h: f"rack{i // 2}" for i, h in enumerate(hosts)})
for d, v, c in (("cpu", 4.0, 16.0), ("mem", 16.0, 64.0)):
p.add_object_dimension(d, {t: v for t in job_of})
p.add_container_dimension(d, {}, default_value=c)
p.add_constraint({"capacitySpec": {"name": d + "_capacity", "scope": "host", "dimension": d, "bound": 1}})
p.add_constraint({"groupCountSpec": {"name": "one_task_per_job_per_rack", "scope": "rack", "partitionName": "job",
"bound": 1, "limit": {"type": 1, "globalLimit": 1.0}}})
p.add_goal({"balanceSpec": {"name": "cpu_balance", "scope": "host", "dimension": "cpu", "formula": 1}})
p.add_solver({"optimalSolverSpec": {"solverPackage": 2, "solveTime": cap, "suppressLogs": True}})
t = time.perf_counter()
sol = p.solve()
dt = time.perf_counter() - t
a = sol["assignment"]
out = {"wall": dt, "end": [r["endReason"] for r in sol["solverSummaries"]], "objective": sol["finalGlobalObjective"]["goals"][0]["value"],
"broken": [sol["initialConstraint"]["brokenCount"], sol["finalConstraint"]["brokenCount"]],
"hash": hashlib.sha256(json.dumps(sorted(a.items())).encode()).hexdigest()[:12],
"rss_mb": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / (1024 ** 2 if sys.platform == "darwin" else 1024)}
if mode == "scale":
lc, lm = collections.Counter(), collections.Counter()
for i, o in enumerate(objs):
lc[a[o]] += cpu[i]; lm[a[o]] += mem[i]
out["over_cap"] = sum(lc[c] > cpu_cap or lm[c] > mem_cap for c in conts)
out["over_cap_start"] = sum(sum(cpu[i] for i, j in enumerate(start) if j == k) > cpu_cap or
sum(mem[i] for i, j in enumerate(start) if j == k) > mem_cap for k in range(M))
out["moved"] = sum(a[objs[i]] != conts[j] for i, j in enumerate(start))
out["spread"] = (max(lc[c] for c in conts) - min(lc[c] for c in conts)) / cpu_cap
print(json.dumps(out))
'''
@section("8. Scale and repeatability: synthetic overloaded clusters, one fresh process per run")
def scale_step():
child = os.path.join(tempfile.mkdtemp(), "rebalancer_child.py")
open(child, "w").write(CHILD)
def run(mode, n, m, cap):
r = subprocess.run([sys.executable, "-P", child, mode, str(n), str(m), str(cap), "42"], capture_output=True,
text=True, timeout=cap + 120)
if r.returncode != 0:
raise RuntimeError(f"child exited {r.returncode}: {r.stderr.strip()[-200:]}")
out = json.loads(r.stdout.strip().splitlines()[-1])
SOLVE_SECONDS[CURRENT[0]] += out["wall"]
return out
print("objects x containers, cap | solve s | endReason | over-cap containers start -> end | brokenCount "
"| moved | CPU spread | peak RSS")
rows = []
for n, m, cap, reps in [(1000, 100, 10, 3), (1000, 1000, 15, 1), (10000, 100, 25, 1)]:
for _ in range(reps):
o = run("scale", n, m, cap)
rows.append(((n, m, cap), o))
e = "/".join(END.get(x, str(x)) for x in o["end"])
print(f"{n:>6} x {m:<5} {cap:>3} s | {o['wall']:7.2f} | {e:<20} | {o['over_cap_start']:>12} -> "
f"{o['over_cap']:<4} | {o['broken'][0]} -> {o['broken'][1]} | {o['moved']:>5} | "
f"{o['spread']:10.3f} | {o['rss_mb']:5.0f} MB")
first = [o for k, o in rows if k == (1000, 100, 10)]
print(f"\n1000 x 100 in 3 fresh processes: {len({round(o['objective'], 9) for o in first})} distinct objective(s), "
f"{len({o['hash'] for o in first})} distinct assignment(s)")
hs = []
if not HIGHS_OK:
print("datacenter with HiGHS: skipped, HiGHS failed the step-1 gate")
else:
hs = [run("highs", 0, 0, 5) for _ in range(3)]
print(f"datacenter with HiGHS in 3 fresh processes: objectives {[round(o['objective'], 6) for o in hs]}, "
f"{len({o['hash'] for o in hs})} distinct assignment(s), endReason "
f"{sorted({END.get(x, str(x)) for o in hs for x in o['end']})}")
big = rows[-1][1]
claim("scale: 1000 x 100 has no over-capacity container after the solve, in every process",
all(o["over_cap"] == 0 for o in first))
return (f"1,000 x 100: {statistics.median(o['wall'] for o in first):.2f} s median, {first[0]['over_cap_start']} -> "
f"{max(o['over_cap'] for o in first)} over-capacity containers, {len({o['hash'] for o in first})} assignment(s) "
f"in 3 processes; 10,000 x 100: {'/'.join(END.get(x, str(x)) for x in big['end'])} at {rows[-1][0][2]} s with "
f"{big['over_cap_start']} -> {big['over_cap']} over capacity; "
+ (f"HiGHS tie: {len({o['hash'] for o in hs})} assignments in {len(hs)} processes, "
f"{len({round(o['objective'], 9) for o in hs})} objective value(s)" if hs else "HiGHS skipped"))
scale_step()
我们在时间上限下运行合成集群，每次都在全新进程中运行。1,000 对象、100 容器的集群在三个进程中分别耗时 1.22、1.27 和 1.35 s，每次都返回相同的分配结果，并清除了全部 13 个过载容器。一个 1,000 乘 1,000 的集群耗时 3.55 s。在 100 个容器上放 10,000 个对象时，25 s 上限被触及并返回 HIT_TIME_LIMIT，因此 endReason 和重新统计可以区分完成的计划与部分完成的计划。对同一数据中心运行三次 HiGHS，目标值均为 0 但分配结果各不相同，因此我们比较目标值而非分配结果。
复制代码ORLIB = "https://people.brunel.ac.uk/~mastjjb/jeb/orlib/"
def fetch(url, tries=4):
for k in range(tries):
try:
with urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": "rebalancer-tutorial"}),
timeout=60) as r:
return r.read()
except Exception as e:
if k == tries - 1:
raise
print(f" retry {k + 1} for {url.rsplit('/', 1)[-1]}: {clean(e, 50)}")
time.sleep(2 * 2 ** k)
def parse_gap(text):
tok = text.split()
P, i, out = int(tok[0]), 1, []
for _ in range(P):
m, n = int(tok[i]), int(tok[i + 1])
i += 2
profit = [[int(tok[i + r * n + j]) for j in range(n)] for r in range(m)]
i += m * n
weight = [[int(tok[i + r * n + j]) for j in range(n)] for r in range(m)]
i += m * n
cap = [int(tok[i + r]) for r in range(m)]
i += m
out.append((m, n, profit, weight, cap))
assert i == len(tok), "every token of the file must be consumed"
return out
# Published optima (maximization), OR-Library gapinfo.html; the cell re-reads the page and checks every value.
GAP_OPTIMA = {"gap1": [336, 327, 339, 341, 326], "gap2": [434, 436, 420, 419, 428], "gap3": [580, 564, 573, 570, 564],
"gap4": [656, 644, 673, 647, 664], "gap5": [563, 558, 564, 568, 559], "gap6": [761, 759, 758, 752, 747],
"gap7": [942, 949, 968, 945, 951], "gap8": [1133, 1134, 1141, 1117, 1127],
"gap9": [709, 717, 712, 723, 706], "gap10": [958, 963, 960, 947, 947],
"gap11": [1139, 1178, 1195, 1171, 1171], "gap12": [1451, 1449, 1433, 1447, 1446]}
def gap_problem(inst, solver_spec):
m, n, profit, weight, cap = inst
agents, jobs = [f"agent{i}" for i in range(m)], [f"job{j}" for j in range(n)]
used, start = [0] * m, {a: [] for a in agents}
for j in range(n): # first fit, least-loaded fallback; the solver repairs any overload
fits = [i for i in range(m) if used[i] + weight[i][j] <= cap[i]]
i = fits[0] if fits else min(range(m), key=lambda k: used[k] - cap[k])
start[agents[i]].append(jobs[j])
used[i] += weight[i][j]
p = new_problem(start, objects="job", containers="agent")
# A job's resource use depends on which agent it sits on: a dynamic object dimension keyed by the agent scope.
p.add_dynamic_object_dimension("resource", "agent",
{agents[i]: {jobs[j]: float(weight[i][j]) for j in range(n)} for i in range(m)},
default_value=0.0)
p.add_container_dimension("resource", {agents[i]: float(cap[i]) for i in range(m)}, default_value=0.0)
p.add_constraint(spec("capacitySpec", name="agent_capacity", scope="agent", dimension="resource",
bound=MAX_BOUND))
p.add_goal(spec("assignmentAffinitiesSpec", name="profit", scope="agent",
affinities=[{"objectName": jobs[j], "scopeItemName": agents[i], "affinity": float(profit[i][j])}
for i in range(m) for j in range(n)]), weight=1.0)
p.add_solver(solver_spec)
return p, agents, jobs
def gap_recount(inst, assignment, agents, jobs):
m, n, profit, weight, cap = inst
idx = {a: i for i, a in enumerate(agents)}
use = [0] * m
for j, job in enumerate(jobs):
use[idx[assignment[job]]] += weight[idx[assignment[job]]][j]
total = sum(profit[idx[assignment[job]]][j] for j, job in enumerate(jobs))
return total, all(use[i] <= cap[i] for i in range(m)) and len(assignment) == n
@section("9. Generalised assignment: 60 OR-Library instances against their published optima")
def gap_step():
page = fetch(ORLIB + "gapinfo.html").decode("latin-1")
published = {name: int(v) for name, v in re.findall(r"(c\d+-\d)\s+(\d+)", re.sub(r"<[^>]+>", " ", page))}
data, digest, nbytes = {}, hashlib.sha256(), 0
for g in GAP_OPTIMA:
raw = fetch(ORLIB + f"files/{g}.txt")
digest.update(raw)
nbytes += len(raw)
data[g] = parse_gap(raw.decode())
pinned = digest.hexdigest()[:12] == "54513b5b96f6" # the files this notebook was verified with
print(f"gap1..gap12: {nbytes / 1024:.0f} KB, sha256 of the 12 files {digest.hexdigest()[:12]} (same as the verified "
f"copy: {pinned}), {sum(len(v) for v in data.values())} instances")
if not pinned:
raise RuntimeError("the OR-Library GAP files differ from the ones this notebook was verified with")
# The page names each instance c<agents><jobs>-<k> (c515-1 is 5 agents, 15 jobs, problem 1).
names = {g: [f"c{m}{n}-{k + 1}" for k, (m, n, *_) in enumerate(v)] for g, v in data.items()}
same = all(published.get(nm) == GAP_OPTIMA[g][k] for g in data for k, nm in enumerate(names[g]))
print(f"gapinfo.html: {len(published)} published optima, matched by instance name to the table in this notebook: "
f"{same}")
if not same:
raise RuntimeError("the optima on gapinfo.html differ from the embedded table")
guard = {"HiGHS": 90.0, "local": 120.0} # seconds; a solve starts only while its cap still fits
print(f"\n{'group':<6}{'size':>7} | {'HiGHS = optimum':>15}{'slowest':>9} | local search = optimum "
f"(seed 1/2/3, best){'':>2} mean shortfall")
tot = collections.Counter()
shortfalls, relative, disagree, worst = [], [], 0, 0.0
for g, insts in data.items():
hit_h, slow, hit, group_short, group_h = 0, 0.0, [0, 0, 0, 0], [], 0
for k, inst in enumerate(insts):
opt = GAP_OPTIMA[g][k]
sol = None
if HIGHS_OK and guard["HiGHS"] >= 10:
p, agents, jobs = gap_problem(inst, highs(10))
try:
sol, dt = solve(p)
except Exception as e: # a HiGHS failure costs this instance's exact run, not the whole step
print(f" {g}#{k + 1}: HiGHS raised {clean(e, 60)}")
tot["h_err"] += 1
if sol is not None:
guard["HiGHS"] -= dt
tot["h_sec"] += dt
val, feas = gap_recount(inst, sol["assignment"], agents, jobs)
slow, worst = max(slow, dt), max(worst, dt)
tot["h_run"] += 1
group_h += 1
tot["h_feas"] += feas
tot["h_optimal_end"] += ends(sol) == "OPTIMAL"
hit_h += feas and val == opt
if feas and val > opt:
raise RuntimeError(f"{g}#{k + 1}: HiGHS profit {val} exceeds the published optimum {opt}")
vals = []
for seed in (1, 2, 3):
if guard["local"] < 3:
continue
p, agents, jobs = gap_problem(inst, local_search(3, seed=seed, moves=("single", "swap", "triple", "kl")))
sol, dt = solve(p)
guard["local"] -= dt
tot["l_sec"] += dt
val, feas = gap_recount(inst, sol["assignment"], agents, jobs)
if val > opt and feas:
raise RuntimeError(f"{g}#{k + 1}: local-search profit {val} exceeds the published optimum {opt}")
tot["l_run"] += 1
tot["l_feas"] += feas
vals.append(val if feas else None)
hit[seed - 1] += feas and val == opt
best = max((v for v in vals if v is not None), default=None)
if len(vals) == 3 and best is not None:
hit[3] += best == opt
shortfalls.append(opt - best)
group_short.append(opt - best)
relative.append((opt - best) / opt)
disagree += len(set(vals)) > 1
tot["h_hit"] += hit_h
for j in range(4):
tot[f"l_hit{j}"] += hit[j]
m, n = insts[0][0], insts[0][1]
gs = f"{statistics.mean(group_short):6.1f}" if group_short else " -"
print(f"{g:<6}{f'{m}x{n}':>7} | {hit_h:>11} of {group_h}{slow:>8.2f}s | {hit[0]:>4} {hit[1]:>4} {hit[2]:>4} {hit[3]:>6}"
f"{'':>20}{gs}")
if not HIGHS_OK:
print("\nHiGHS: skipped, it failed the step-1 gate")
else:
print(f"\nHiGHS: {tot['h_hit']} of {tot['h_run']} equal the published optimum; endReason OPTIMAL "
f"{tot['h_optimal_end']} of {tot['h_run']}; recounted feasible {tot['h_feas']} of {tot['h_run']}; raised "
f"{tot['h_err']}; slowest {worst:.2f} s, {tot['h_sec']:.1f} s in total")
mean_short = statistics.mean(shortfalls) if shortfalls else float("nan")
print(f"local search (single, swap, triple loop, Kernighan-Lin; first-fit start; 3 s cap): seed 1/2/3 hit "
f"{tot['l_hit0']}/{tot['l_hit1']}/{tot['l_hit2']}, best of three {tot['l_hit3']} of {len(shortfalls)}; "
f"best-of-three shortfall mean {mean_short:.1f} "
f"({100 * statistics.mean(relative) if relative else float('nan'):.2f}% of the optimum), "
f"max {max(shortfalls, default=0)}; seeds disagree on {disagree} "
f"instances; recounted feasible {tot['l_feas']} of {tot['l_run']}; {tot['l_sec']:.1f} s in total")
if (HIGHS_OK and tot["h_run"] < 60) or len(shortfalls) < 60:
print(f"PARTIAL: {tot['h_run']} HiGHS and {len(shortfalls)} local-search instances completed (time guard or "
f"errors); the 60-of-60 comparison does not apply")
claim("GAP: all 60 HiGHS runs ran and equal the published optimum", tot["h_hit"] == tot["h_run"] == 60)
claim("GAP: all 180 local-search answers ran and are feasible by recount", tot["l_feas"] == tot["l_run"] == 180)
return ((f"HiGHS {tot['h_hit']}/{tot['h_run']} published optima (slowest {worst:.2f} s)" if HIGHS_OK else
"HiGHS skipped") + f"; local search best-of-3 {tot['l_hit3']}/{len(shortfalls)}, best-of-3 shortfall "
f"mean {mean_short:.1f}")
gap_step()
我们针对 60 个 OR-Library 广义指派实例检验求解器，这些实例都有已发表的最优解，并在运行时下载和校验校验和。HiGHS 匹配了全部 60 个最优解，每个答案都可行，最慢的一个耗时 1.31 s。即使使用三个种子中最好的一个，从首次适配起始点出发的局部搜索在 60 个实例中也只有 4 个达到最优，其每个实例的最佳答案平均比最优少 7.7 个利润单位（1.07%）。在九次运行中，平均差距在 7.6 到 8.1 之间波动，即使是固定种子在不同运行间也存在差异，原因我们未能查明。
复制代码def parse_binpack(text, scale=1):
tok = text.split()
name, C, n, best = tok[1], float(tok[2]), int(tok[3]), int(tok[4])
sizes = [float(x) for x in tok[5:5 + n]]
assert len(sizes) == n, "the file holds fewer sizes than it declares"
S_int = [round(s * scale) for s in sizes]
assert abs(sum(S_int) - scale * sum(sizes)) < 1e-6, "scaled sizes must stay exact"
return name, round(C * scale), S_int, best
def bin_problem(sizes, C, start, solver_spec, minimise):
p = new_problem(start, objects="item", containers="bin")
items = [f"item{i}" for i in range(len(sizes))]
p.add_object_dimension("size", dict(zip(items, map(float, sizes))))
p.add_object_dimension("count", {it: 1.0 for it in items})
p.add_container_dimension("size", {}, default_value=float(C))
p.add_constraint(spec("capacitySpec", name="bin_capacity", scope="bin", dimension="size", bound=MAX_BOUND))
if minimise:
p.add_goal(spec("minimizeContainersSpec", name="fewest_bins", scope="bin", dimension="count"), weight=1.0)
p.add_solver(solver_spec)
return p
def pack(sizes, C, K=None):
"""First-fit decreasing; with K given, first fit into K bins with least-loaded fallback (may overflow)."""
order = sorted(range(len(sizes)), key=lambda i: -sizes[i])
loads, bins = [], []
if K:
loads, bins = [0] * K, [[] for _ in range(K)]
for i in order:
fits = [b for b in range(len(loads)) if loads[b] + sizes[i] <= C]
if fits:
b = fits[0]
elif K:
b = min(range(K), key=lambda q: loads[q])
else:
loads.append(0)
bins.append([])
b = len(loads) - 1
loads[b] += sizes[i]
bins[b].append(f"item{i}")
return {f"bin{b}": bins[b] for b in range(len(bins))}
def bins_used(assignment, sizes, C):
L = collections.Counter()
for it, b in assignment.items():
L[b] += sizes[int(it[4:])]
return sum(1 for v in L.values() if v > 0), all(v <= C for v in L.values()) and len(assignment) == len(sizes)
@section("10. Bin packing: where local search with MinimizeContainers stops, measured against the L1 bound")
def binpack_step():
rows = []
for fname, scale in (("binpack1.txt", 1), ("binpack5.txt", 10)):
raw = fetch(ORLIB + "files/" + fname)
if hashlib.sha256(raw).hexdigest()[:12] != {"binpack1.txt": "891fe4b3be86", "binpack5.txt": "33cb019a3f05"}[fname]:
raise RuntimeError(f"{fname} differs from the copy this notebook was verified with")
name, C, sizes, best = parse_binpack(raw.decode(), scale)
L1 = -(-sum(sizes) // C)
ffd = pack(sizes, C)
print(f"{name} ({fname}, {len(raw) / 1024:.1f} KB, sha256 {hashlib.sha256(raw).hexdigest()[:12]} as verified): "
f"{len(sizes)} items, capacity {C}"
f"{' (sizes x10)' if scale > 1 else ''}; L1 lower bound {L1} (file's best known {best}); first-fit "
f"decreasing {len(ffd)} bins")
sol, dt = solve(bin_problem(sizes, C, ffd, local_search(5, seed=1, moves=("single", "swap", "triple", "kl")),
True))
used, ok = bins_used(sol["assignment"], sizes, C)
print(f" A MinimizeContainers, local search from FFD: {used} bins (gap to L1 {used - L1}), feasible {ok}, "
f"{ends(sol)}, {dt:.2f} s")
startK = pack(sizes, C, K=L1)
solB, dtB = solve(bin_problem(sizes, C, startK, local_search(5, seed=1, moves=("single", "swap", "triple",
"kl")), False))
usedB, okB = bins_used(solB["assignment"], sizes, C)
print(f" B exactly L1 = {L1} bins, capacity only, local search: feasible {okB} ({usedB} bins used), "
f"{ends(solB)}, {dtB:.2f} s")
okC = "skipped"
if not HIGHS_OK:
print(" C HiGHS: skipped, it failed the step-1 gate")
else:
solC, dtC = solve(bin_problem(sizes, C, startK, highs(10), False))
usedC, okC = bins_used(solC["assignment"], sizes, C)
print(f" C exactly L1 = {L1} bins, capacity only, HiGHS: feasible {okC}, {ends(solC)}, {dtC:.2f} s "
f"(a time limit is 'not found in 10 s', not a proof of infeasibility)")
rows.append((name, L1, len(ffd), used, okB, okC, ok))
claim("bin packing: MinimizeContainers local search stays above L1, feasibly, on at least one instance",
any(u > L and okA for _, L, _, u, _, _, okA in rows))
claim("bin packing: capacity-only local search with exactly L1 bins packs u120_00 (optimal by the bound)",
rows[0][4] is True)
return "; ".join(f"{n}: L1 {L}, FFD {f}, MinimizeContainers {u}, fixed-L1 local feasible {b}, HiGHS {c}"
for n, L, f, u, b, c, _ in rows)
binpack_step()
我们在两个 OR-Library 实例上用 MinimizeContainers 测试装箱问题，并与 L1 下界进行比较，该下界等于它们已知的最佳箱数 48 和 20。从首次适配递减出发的局部搜索在两个实例上都没有减少任何箱数。仅凭容量要求恰好 L1 个箱，就能在 0.02 s 内装下 u120_00，达到下界，因而是最优的。在 t60_00 上，局部搜索找不到 20 箱的装法，而 HiGHS 在两个实例上 10 s 内也找不到，这是超时，而不是不可行的证明。
复制代码banner("SUMMARY")
for name, res in RESULTS.items():
print(f" {name:<100s} {res}")
print("\nClaims, each decided in this run by a recount, an endReason or a raised error, as the claim says:")
for step, text, held in CLAIMS:
print(f" [{'held' if held else 'DID NOT HOLD'}] step {step}: {text}")
failed = [k for k, v in RESULTS.items() if str(v).startswith("SKIPPED")]
if not HIGHS_OK:
print("\nHiGHS failed the step-1 gate: step 7 did not run, and the HiGHS parts of steps 8, 9 and 10 were skipped")
print(f"\n{sum(h for *_, h in CLAIMS)} of {len(CLAIMS)} claims held; steps failed or skipped: {failed or 'none'}")
print(f"seconds inside solve() per step: {dict((k, round(v, 2)) for k, v in sorted(SOLVE_SECONDS.items(), key=lambda kv: int(kv[0])))}"
f", total {sum(SOLVE_SECONDS.values()):.1f} s")
print(f"run on: {HW}; Python {platform.python_version()}; rebalancer {VERSION}")
print("not tested here: production scale and the paper's or Meta's latency and volume figures; XPRESS and Gurobi "
"(absent from the wheel); thread control; GAP classes a-d; public cluster traces.")
摘要列出了每一项声明及其验证方式，即通过重新统计、endReason 或引发的错误，以及每步的求解器秒数。在这次运行中，全部 28 项声明都成立，没有任何步骤失败，求解器内共花费 90.1 s。此处未测试的包括生产规模、论文或 Meta 的延迟与吞吐量数据，以及 XPRESS 和 Gurobi 后端。
Rebalancer 将放置问题转化为可读的模型，由局部搜索或精确求解器 HiGHS 求解。在这些模型上，它跨机架重新均衡了一个八主机集群，排空了一个繁忙的主机，通过 HiGHS 匹配了全部 60 个已发表的 OR-Library 最优解，并在 M4 上用约 1.3 s 修复了一个 1,000 对象、100 容器的集群。这次运行还暴露了陷阱：solved 和 OPTIMAL 可能与被违反的约束并存，HARD 因求解器而异，拼写错误的键会被忽略，枚举名称会被拒绝，IDEAL 在 HiGHS 下失败，而目标元组的顺序决定了被违反的约束是否会被修复。将枚举作为整数传入，检查 spec 键，重新统计每一个结果，并比较目标值而非分配结果。
查看 FULL CODES here。所有功劳归功于该项目的研究者。此外，欢迎关注我们的 Twitter ，别忘了加入我们的 150k+ML SubReddit 并订阅 our Newsletter。等等！你在用 telegram 吗？ 现在你也可以在 telegram 上加入我们。
这篇文章 How to Use Meta’s Rebalancer for Task Placement & Optimization 首发于 MarkTechPost.