AI 服務暫時不可用，以下為來源正文，待恢復後補全翻譯。
In this tutorial, we implement the Rebalancer, Meta’s assignment solver, with a C++ core and a Python binding. We describe objects, containers, dimensions, scopes, and partitions, add constraints and goals, then choose local search or the exact mixed-integer solver HiGHS. We build a small datacenter, drain a busy host, and compare the two solvers on toy models, mutually exclusive hosts, synthetic clusters, and 60 published OR-Library instances. We judge every plan by recounting the returned assignment rather than by the solver’s flags. In this build, across 24 solves, the solved flag read True in 14 solves that still left memory over capacity. HARD under local search returned a violating start without raising; misspelled spec keys were ignored silently, and enum names are rejected. The timings come from an Apple M4 with the macOS wheel, and Colab’s Linux wheel may time differently. Copy CodeCopiedUse a different Browser import 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) ", 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: 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: 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: accepted") except Exception as e: print(f" {label: {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()) {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']} {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() We solve the README example, four 10 GB tasks on two 20 GB hosts with three tasks on one host, and recount the result instead of trusting the solver. Local search moves one task in 0.3 ms, and the recount drops the excess from 10 GB to 0. The solver’s own brokenCount also falls from 1 to 0 and endReason reads OPTIMAL. A control makes the point about the flag: with solveTime=0 the solver moves nothing, still leaves 10 GB over capacity, and reports solved=True. Copy CodeCopiedUse a different Browser HOSTS = [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 gro [truncated for AI cost control]