如何使用 Meta 的 Rebalancer 進行任務部署與最佳化

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 Browserimport 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 SOLVESECONDS = 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.

printexc(limit=3) return None finally: print(f"\n[{time.time() - t0:.1f} s]") json.dump({"results": RESULTS, "claims": CLAIMS, "solveseconds": SOLVESECONDS}, 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.versioninfo[: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() == "x8664") 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} MAXBOUND = 1 END = {1: "OPTIMAL", 2: "HITTIMELIMIT", 3: "HITMOVELIMIT", 4: "HITSTOPCONSTRAINT", 5: "HITPLATEAUTIME", 6: "NOSOLUTIONFOUND", 7: "HITSTAGEMOVELIMIT", 8: "HITEXCEPTION", 9: "UNABLETOFINDMOREMOVES", 10: "HITMINCYCLEOBJECTIVEIMPROVEMENT", 11: "UNABLETOFINDIMPROVINGMOVES"} NESTED = {"limit": "Limit", "conflictInfoList": "ScopeItemConflictInfo", "conflictingScopeItemsWithOverlap": "ConflictingScopeItemInfo"} def checkkeys(name, fields, where): cls = getattr(S, name) unknown = set(fields) - (cls.

requiredkeys | cls.optionalkeys) 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): checkkeys(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.""" checkkeys(arm[0].

upper() + arm[1:], fields, arm) return {arm: fields} MOVES = {"single": {"singleMoveTypeSpec": {}}, "swap": {"swapMoveTypeSpec": {}}, "triple": {"tripleLoopMoveTypeSpec": {}}, "kl": {"klSearchMoveTypeSpec": {}}} def localsearch(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 newproblem(assignment, objects="task", containers="host"): p = ProblemSolver(servicename="rebalancertutorial", servicescope="marktechpost") p.

setloglevel("ERROR") # silences ~50 log lines per solve; never "OFF", which exits the process p.setobjectname(objects).setcontainername(containers) # names first, then the assignment, then dimensions p.

setassignment(assignment) # every container must be listed, empty ones too return p def solve(p): t = time.perfcounter() sol = p.solve() dt = time.

perfcounter() - t SOLVESECONDS[CURRENT[0]] += dt return sol, dt # Independent checks: everything below reads only the returned assignment, never the solver's own verdict.def where(containertoobjects): return {o: c for c, objs in containertoobjects.

items() for o in objs} def bycontainer(objecttocontainer, containers): out = {c: [] for c in containers} for o, c in sorted(objecttocontainer.items()): out[c].append(o) return out def load(objecttocontainer, values): out = collections.Counter() for o, c in objecttocontainer.

items(): out[c] += values.get(o, 0.0) return out def excess(objecttocontainer, values, caps): L = load(objecttocontainer, 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}", "", str(e)).

strip()[:n] return f"{type(e).__name__}: {msg}" if typed else msg def cpuname(): 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.brandstring"], captureoutput=True, text=True) if r.stdout.strip(): return r.stdout.strip() return platform.processor() or platform.machine() def ramgb(): 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"], captureoutput=True, text=True) return int(r.stdout) / 1024 ** 3 if r.stdout.strip() else float("nan") VERSION = md.version("rebalancer") HW = f"{cpuname()}, {os.cpucount()} logical CPUs, {ramgb():.

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 tw