跳到主要内容
AI News HubLIVE
来源内容 · 翻译待补全6 分钟阅读

待翻译:How to Use Meta’s Rebalancer for Task Placement & Optimization

文章摘要

AI 服务暂时不可用,以下为来源摘要,待恢复后补全翻译:In this comprehensive coding tutorial, we explore Meta's Rebalancer assignment solver using its C++ core and Python bindings. Learn how to build datacenters, manage task placements across racks, handle constraints and goal priorities, and compare local search against the exact HiGHS solver using published benchmarks. The post How to Use Meta’s Rebalancer for Task Placement & Optimization appeared first on MarkTechPost.

来源MarkTechPost作者: Sana Hassan
待翻译:How to Use Meta’s Rebalancer for Task Placement & Optimization
报告错误

纠错通道尚未开通,可先复制下方文章信息留存。

查看更正说明
直接读正文

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]

展开要点与分析

文章情报

研究者进阶

要点

  • AI 服务暂时不可用,系统已先保留来源内容与降级元数据。
  • In this comprehensive coding tutorial, we explore Meta's Rebalancer assignment solver using its C++ core and Python bindings. Learn how to build datacenters, manage task placement…

要点与分析由自动化流程生成,可能有误,请结合原始来源核实。