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()) We install rebalancer 1.0.4 and print the Python version and hardware, because every timing depends on them. The binding rejects enum names and silently ignores unknown spec keys, so the notebook passes integers from one table and checks each spec before solving. A gate then solves a two-task model with both solvers, each moving one task, in 0.8 ms with local search and 0.9 ms with HiGHS. The cell also confirms that this build refuses the default XPRESS solver and the Gurobi package. Copy CodeCopiedUse a different Browser README_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']} {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"])