Как да използвате Rebalancer на Meta за разполагане и оптимизиране на задачи

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) <= 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'}"