이 튜토리얼에서는 다음을 구현합니다. Rebalancer, Meta의 C++ 코어와 Python 바인딩을 갖춘 할당 솔버입니다. 객체, 컨테이너, 차원, 스코프, 파티션을 설명하고, 제약 조건과 목표를 추가한 다음, 로컬 서치 또는 정확한 혼합 정수 솔버 HiGHS를 선택합니다. 작은 데이터센터를 구축하고, 사용 중인 호스트를 비우고, 장난감 모델, 상호 배타적인 호스트, 합성 클러스터, 그리고 공개된 60개의 OR-Library 인스턴스에서 두 솔버를 비교합니다. 모든 계획은 솔버의 플래그가 아니라 반환된 할당을 다시 세어 판단합니다. 이 빌드에서 24회의 솔브 중 solved 플래그는 여전히 메모리가 용량을 초과한 상태로 남은 14회의 솔브에서 True를 나타냈습니다. 로컬 서치에서 HARD는 오류를 일으키지 않고 위반 상태의 시작을 반환했으며, 철자가 잘못된 spec 키는 조용히 무시되었고, enum 이름은 거부되었습니다. 시간 측정은 macOS wheel이 설치된 Apple M4에서 수행되었으며, Colab의 Linux wheel에서는 시간이 다를 수 있습니다.
Copy Codeimport 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'}"
rebalancer 1.0.4를 설치하고 Python 버전과 하드웨어를 출력합니다. 모든 시간 측정이 이에 의존하기 때문입니다. 바인딩은 enum 이름을 거부하고 알 수 없는 spec 키를 조용히 무시하므로, 노트북은 하나의 표에서 정수를 전달하고 솔브 전에 각 spec을 확인합니다. 그런 다음 게이트가 두 솔버로 두 작업 모델을 솔브하며, 각각 하나의 작업을 이동시키고 로컬 서치로 0.8 ms, HiGHS로 0.9 ms가 걸립니다. 이 셀은 또한 이 빌드가 기본 XPRESS 솔버와 Gurobi 패키지를 거부함을 확인합니다.
Copy CodeREADME_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']} <- 0 means no time, not 'no limit'")
claim("README model: recounted excess reaches 0 GB", sum(ex1.values()) == 0)
claim("README model: a 2-2 start is left alone", mb == 0)
claim("README model: solveTime=0 makes no moves", mz == 0)
return (f"3-1 start -> {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()
README 예제, 즉 20 GB 호스트 두 대에서 10 GB 작업 네 개를 풀고 한 호스트에 작업 세 개가 있는 상태를 솔브하고, 솔버를 신뢰하는 대신 결과를 다시 셉니다. 로컬 서치는 0.3 ms에 하나의 작업을 이동시키고, 재계산 결과 초과분이 10 GB에서 0으로 줄어듭니다. 솔버 자체의 brokenCount도 1에서 0으로 떨어지고 endReason은 OPTIMAL을 나타냅니다. 대조군은 플래그에 대한 요점을 보여줍니다. solveTime=0으로 설정하면 솔버는 아무것도 이동시키지 않고 여전히 용량을 10 GB 초과한 상태로 두면서 solved=True를 보고합니다.
Copy CodeHOSTS = [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 group limit leaves rack duplicates", cr["rack"] > 0)
return (f"violations CPU/mem/rack {c0['cpu']:.0f}/{c0['mem']:.0f}/{c0['rack']} -> {c1['cpu']:.0f}/"
f"{c1['mem']:.0f}/{c1['rack']}, {moved(before, sol['assignment'])} tasks moved, {ms:.1f} ms; "
f"without the rule {cc['rack']} rack duplicates, with a RELATIVE limit {cr['rack']}")
datacenter_step()
네 개의 랙에 여덟 개의 호스트를 구축하고, 각각 네 개의 작업을 가진 네 개의 잡을 만들며, host0이 CPU 200%와 랙 중복 12개인 상태에서 시작합니다. CPU 및 메모리 용량, 랙당 잡당 작업 하나 규칙, CPU 균형 목표를 추가합니다. 솔버는 13개의 작업을 이동시켜 위반 없이 호스트당 두 개를 남기며, 세 번의 새로운 빌드에서 중간 솔브 시간은 2.5 ms였습니다. 랙 규칙이 없으면 재계산에서 5개의 랙 중복이 발견되고, 이에 대한 RELATIVE 제한 역시 5로 끝나므로 여기서는 아무것도 강제하지 않습니다.
Copy Code@section("4. What 'solved' means: the constraint policies under local search and HiGHS, checked by recount")
def policy_step():
cases = {"F (fixable)": ({f"t{i}": 10.0 for i in range(4)}, {"host0": ["t0", "t1", "t2"], "host1": ["t3"]}),
"I (infeasible)": ({f"t{i}": 20.0 for i in range(3)}, {"host0": ["t0", "t1", "t2"], "host1": []})}
caps = {"host0": 20.0, "host1": 20.0}
def build(sizes, start, policy, solver_spec):
p = new_problem(start)
p.add_object_dimension("memory", sizes)
p.add_container_dimension("memory", {}, default_value=20.0)
p.add_constraint(spec("capacitySpec", name="memory_capacity", scope="host", dimension="memory"),
policy=POLICY[policy])
p.add_solver(solver_spec)
return p
rows, stable, solved_but_broken, cells = [], [], 0, 0
print(f"{'case':<15}{'policy':<9}{'solver':<14}{'outcome':<36}{'moved':>6}{'broken':>9}{'solved':>8}"
f"{'brokenVal':>11}{'excess GB':>11}")
for case, (sizes, start) in cases.items():
for policy in ("DEFAULT", "HARD", "SOFT"):
for solver_name, solver_spec, reps in [("local search", local_search(1), 3), ("HiGHS", highs(2), 1)]:
outs = []
for _ in range(reps):
cells += 1
try:
sol, _ = solve(build(sizes, start, policy, solver_spec))
ex = sum(excess(sol["assignment"], sizes, caps).values())
fc, ic = sol["finalConstraint"], sol["initialConstraint"]
solved_but_broken += fc["solved"] and ex > 0
outs.append({"out": "returned, " + ends(sol), "moved": moved(where(start), sol["assignment"]),
"broken": f"{ic['brokenCount']}->{fc['brokenCount']}", "solved": fc["solved"],
"bv": fc["brokenVal"], "ex": ex, "sol": sol, "sizes": sizes})
except Exception as e:
outs.append({"out": "raised: " + clean(e, 30, typed=False), "moved": "-", "broken": "-",
"solved": "-",
"bv": None, "ex": None})
o = outs[0]
if reps > 1:
stable.append(len({(x["out"], x["moved"], x["broken"], x["bv"]) for x in outs}) == 1)
bv = "-" if o["bv"] is None else f"{o['bv']:g}"
exs = "-" if o["ex"] is None else f"{o['ex']:.0f}"
print(f"{case:<15}{policy:<9}{solver_name:<14}{o['out']:<36}{o['moved']!s:>6}{o['broken']:>9}"
f"{o['solved']!s:>8}{bv:>11}{exs:>11}")
rows.append((case, policy, solver_name, o))
print(f"\nlocal-search repeats (3 per row) identical in {sum(stable)} of {len(stable)} rows")
print(f"cells where solved=True although the recount still finds excess memory: {solved_but_broken} of {cells}")
# What does brokenVal measure? Three candidate formulas, tested on the first returned result of every row above and
# on four starts read before any move: two hosts over capacity, and three starts with hosts of different sizes.
probes = [({"host0": ["t0", "t1", "t2", "t3"], "host1": ["t4", "t5", "t6"], "host2": [], "host3": []},
{"host0": 20.0, "host1": 20.0, "host2": 20.0, "host3": 20.0}),
({"host0": ["t0", "t1", "t2"], "host1": []}, {"host0": 20.0, "host1": 40.0}),
({"host0": ["t0", "t1", "t2", "t3"], "host1": ["t4", "t5", "t6", "t7"]}, {"host0": 20.0, "host1": 30.0}),
({"host0": ["t0", "t1", "t2"], "host1": ["t3", "t4", "t5", "t6"], "host2": ["t7"]},
{"host0": 20.0, "host1": 30.0, "host2": 100.0})]
samples = [(o["sol"]["assignment"], o["sizes"], caps, o["bv"]) for *_, o in rows if o["bv"] is not None]
for start, pcaps in probes:
sizes = {t: 10.0 for objs in start.values() for t in objs}
p = new_problem(start)
p.add_object_dimension("memory", sizes)
p.add_container_dimension("memory", pcaps, default_value=0.0)
p.add_constraint(spec("capacitySpec", name="memory_capacity", scope="host", dimension="memory"))
p.add_solver(local_search(0))
probe, _ = solve(p)
bv = probe["initialConstraint"]["brokenVal"]
samples.append((where(start), sizes, pcaps, bv))
ex = excess(where(start), sizes, pcaps)
print(f"start {[f'{load(where(start), sizes)[h]:.0f}/{c:.0f}' for h, c in pcaps.items()]} GB (load/capacity): "
f"excess {sum(ex.values()):.0f} GB, brokenVal {bv:.4g}")
cands = {"total excess in GB": lambda e, c: sum(e.values()),
"sum over hosts of excess / that host's capacity": lambda e, c: sum(e[h] / c[h] for h in c),
"total excess / mean capacity of all hosts": lambda e, c: sum(e.values()) / (sum(c.values()) / len(c))}
match = [name for name, f in cands.items()
if all(abs(f(excess(a, s, c), c) - bv) < 1e-9 for a, s, c, bv in samples)]
print(f"formula that matches brokenVal in all {len(samples)} samples: {match or 'NONE'}")
hard_ls_f = next(o for c, p_, s, o in rows if c.startswith("F") and p_ == "HARD" and s == "local search")
hard_h_i = next(o for c, p_, s, o in rows if c.startswith("I") and p_ == "HARD" and s == "HiGHS")
claim("policies: HARD + local search leaves the fixable start unchanged without raising",
hard_ls_f["moved"] == 0 and hard_ls_f["ex"] and hard_ls_f["ex"] > 0)
claim("policies: HARD + HiGHS raises 'infeasible' on the infeasible start",
hard_h_i["out"].startswith("raised") and "infeasible" in hard_h_i["out"])
claim("policies: solved=True appears with excess remaining", solved_but_broken > 0)
return (f"solved=True while excess remained in {solved_but_broken} of {cells} solves; HARD + local search moved "
f"{hard_ls_f['moved']} tasks on the fixable start, no error; HARD + HiGHS on the infeasible start "
f"{hard_h_i['out']}; brokenVal = {', '.join(match) or 'no candidate'}")
policy_step()
수정 가능한 시작 F와 불가능한 시작 I를 각 제약 정책과 두 솔버에서 테스트하며 모든 결과를 다시 셉니다. 로컬 서치에서 HARD는 두 시작을 그대로 반환하며 OPTIMAL과 solved=True를 나타내는 반면, HiGHS는 F를 수정하고 I에 대해 infeasible 오류를 일으킵니다. 24회의 솔브 중 solved는 여전히 메모리가 용량을 초과한 상태로 남은 14회의 솔브에서 True를 나타냈습니다. 15개 샘플 모두에서 솔버의 brokenVal은 총 초과분을 평균 호스트 용량으로 나눈 값과 같습니다. OPTIMAL을 신뢰하기보다 brokenCount를 읽거나 재계산하십시오.
Copy Code@section("5. Goal tuples: a lower tuple always wins, weights only trade off inside one tuple")
def tuple_step():
# Part A: one task would like to move (affinity 1) and every move costs 1000 (MinimizeMovement on task_count).
def affinity_vs_movement(aff_w, move_w, move_tuple):
p = new_problem({"host0": ["t0", "t1"], "host1": []})
p.add_goal(spec("assignmentAffinitiesSpec", name="t0_prefers_host1", scope="host",
affinities=[{"objectName": "t0", "scopeItemName": "host1", "affinity": 1.0}]),
weight=aff_w, tuple_pos=0)
p.add_goal(spec("minimizeMovementSpec", name="moves", scope="host", dimension="task_count",
doNotNormalize=True), weight=move_w, tuple_pos=move_tuple)
p.add_solver(local_search(2))
sol, _ = solve(p)
return sol["assignment"]["t0"] == "host1", goals(sol)
print("Part A, no constraints: affinity for t0 on host1 against a movement cost")
A = {}
for label, args in [("same tuple, affinity weight 1, movement weight 1000", (1.0, 1000.0, 0)),
("same tuple, affinity weight 1000, movement weight 1", (1000.0, 1.0, 0)),
("affinity in tuple 0, movement weight 1000 in tuple 1", (1.0, 1000.0, 1))]:
A[label] = affinity_vs_movement(*args)
print(f" {label:<55} t0 moved: {str(A[label][0]):<5} goals per tuple {A[label][1]}")
# Part B: a broken capacity constraint becomes a 'fix it' goal in tuple tuple_pos_if_broken (default 0).
sizes = {f"t{i}": 10.0 for i in range(4)}
def fixit(start, move_tuple, fix_tuple=None):
p = new_problem(start)
p.add_object_dimension("mem", sizes)
p.add_container_dimension("mem", {}, default_value=20.0)
kw = {} if fix_tuple is None else {"tuple_pos_if_broken": fix_tuple}
p.add_constraint(spec("capacitySpec", name="mem_capacity", scope="host", dimension="mem"), **kw)
p.add_goal(spec("minimizeMovementSpec", name="moves", scope="host", dimension="mem", doNotNormalize=True),
weight=1000.0, tuple_pos=move_tuple)
p.add_solver(local_search(2))
sol, _ = solve(p)
return sol
broken3 = {"host0": ["t0", "t1", "t2"], "host1": ["t3"]}
B = {"X1 movement tuple 0, fix-it tuple 0 (default)": fixit(broken3, 0),
"X2 movement tuple 0, fix-it tuple 1": fixit(broken3, 0, 1),
"X3 movement tuple 1, fix-it tuple 0": fixit(broken3, 1, 0),
"X4 feasible 2-2 start, movement tuple 0": fixit({"host0": ["t0", "t1"], "host1": ["t2", "t3"]}, 0)}
print("\nPart B, broken capacity (30 GB on a 20 GB host) plus a movement goal of weight 1000 on the moved GB")
EX = {k: sum(excess(sol["assignment"], sizes, {"host0": 20.0, "host1": 20.0}).values()) for k, sol in B.items()}
for label, sol in B.items():
ex = EX[label]
print(f" {label:<48} moves {len(sol['movesSummary'])}, recounted excess {ex:.0f} GB, brokenCount "
f"{sol['finalConstraint']['brokenCount']}, goals per tuple {goals(sol)}")
x2, x3 = B["X2 movement tuple 0, fix-it tuple 1"], B["X3 movement tuple 1, fix-it tuple 0"]
# The fix-it value against 10000 (invalidState) + 100 (invalidCost) x brokenVal, with brokenVal recounted as the total
# excess over the 20 GB mean capacity (step 4), on X2 and on a second start with 40 GB on host0 (X2b).
x2b = fixit({"host0": ["t0", "t1", "t2", "t3"], "host1": []}, 0, 1)
fix_rows = []
for label, sol, start_excess in (("X2", x2, EX["X2 movement tuple 0, fix-it tuple 1"]),
("X2b", x2b, sum(excess(x2b["assignment"], sizes, {"host0": 20.0, "host1": 20.0}).values()))):
bv = start_excess / 20.0
fix_rows.append(abs(goals(sol)[-1] - (10000 + 100 * bv)) < 1e-6 and len(sol["movesSummary"]) == 0)
print(f" {label}: {start_excess:.0f} GB over, no move; tuple-1 value {goals(sol)[-1]:g} vs 10000 + 100 x {bv:g} = "
f"{10000 + 100 * bv:g}")
gb_moved = 10.0 * moved(where(broken3), x3["assignment"])
move_rule = abs(goals(x3)[-1] - 1000.0 * gb_moved) < 1e-6
print(f" fix-it value = 10000 + 100 x brokenVal on both starts: {'confirmed' if all(fix_rows) else 'not confirmed'}")
print(f" X3's tuple-1 value {goals(x3)[-1]:g} vs weight 1000 x {gb_moved:g} GB moved (recounted): "
f"{'confirmed' if move_rule else 'not confirmed'}")
a = list(A.values())
claim("tuples: weight 1 vs 1000 in one tuple keeps t0 home; swapping weights moves it", not a[0][0] and a[1][0])
claim("tuples: a tuple-0 goal beats any weight in tuple 1", a[2][0])
claim("tuples: fix-it in a later tuple than movement (X2) leaves 10 GB of excess, no move",
EX["X2 movement tuple 0, fix-it tuple 1"] > 0 and len(x2["movesSummary"]) == 0)
claim("tuples: fix-it in an earlier tuple than movement (X3) removes the excess",
EX["X3 movement tuple 1, fix-it tuple 0"] == 0)
claim("tuples: the fix-it value is 10000 + 100 x the recounted brokenVal on two starts", all(fix_rows))
return (f"t0 moves with weights 1 vs 1000: {a[0][0]}, 1000 vs 1: {a[1][0]}, affinity one tuple above: {a[2][0]}; "
f"fix-it below movement: {len(x2['movesSummary'])} moves, brokenCount {x2['finalConstraint']['brokenCount']}; "
f"fix-it above: {len(x3['movesSummary'])} move, brokenCount {x3['finalConstraint']['brokenCount']}")
tuple_step()
낮은 목표 튜플이 항상 먼저 결정되며, 가중치는 하나의 튜플 내에서만 상충 관계를 이룹니다. 둘 다 튜플 0에 있을 때, 어피니티의 가중치 1은 이동의 가중치 1000에게 지고, 가중치를 바꾸면 t0이 이동합니다. 어피니티가 튜플 0에 있고 이동이 튜플 1에 있을 때에도 t0은 여전히 이동합니다. 이동보다 나중 튜플에 있는 고치기(fix-it) 목표는 이동 없이 10 GB 초과 상태를 남기는 반면, 순서를 뒤집으면 한 번의 이동으로 이를 복구합니다. 고치기 값은 10000에 100을 곱한 brokenVal을 더한 값이며, 두 시작에서 확인되었습니다.
Copy Code@section("6. Draining a busy host under a move budget, and what a misspelled key does")
def drain_step():
before = where(DC_START)
builds = [solve(datacenter(local_search(3), drain="host1")) for _ in range(3)]
sol0, dt0 = builds[0]
c = dc_check(sol0["assignment"], drain="host1")
r = len(sol0["movesSummary"])
left0 = c["per_host"][1]
print(f"drain host1 (starts with {len(DC_START['host1'])} jobC tasks), no budget: host1 holds {left0}, "
f"recounted CPU/mem/rack violations {c['cpu']:.0f}/{c['mem']:.0f}/{c['rack']}, tasks per host "
f"{c['per_host']}")
print(f" {r} moves ({moved(before, sol0['assignment'])} tasks changed host), endReason {ends(sol0)}, brokenCount "
f"{sol0['initialConstraint']['brokenCount']} -> {sol0['finalConstraint']['brokenCount']}, "
f"{dt0 * 1000:.1f} ms")
print(f" three fresh builds of this model: {[len(b['movesSummary']) for b, _ in builds]} moves, host1 left with "
f"{[dc_check(b['assignment'], drain='host1')['per_host'][1] for b, _ in builds]} tasks")
print(f" rack0 is now host0 alone: {c['per_host'][0]} tasks = {c['per_host'][0] * TASK_CPU:.0f} of "
f"{HOST_CPU:.0f} CPU, one per job")
budget_rows = {}
for k in (r - 1, max(1, r // 2)):
sol, _ = solve(datacenter(local_search(3, stopAfterMoves=k), drain="host1"))
ck = dc_check(sol["assignment"], drain="host1")
budget_rows[k] = (sol, ck)
print(f"stopAfterMoves={k:<3} -> {len(sol['movesSummary'])} moves, endReason {ends(sol)}, host1 still holds "
f"{ck['per_host'][1]}, recounted violations {ck['cpu']:.0f}/{ck['mem']:.0f}/{ck['rack']}, brokenCount "
f"{sol['finalConstraint']['brokenCount']}, solved={sol['finalConstraint']['solved']}")
typo = local_search(3)
typo["localSearchSolverSpec"]["stopAfterMove"] = max(1, r // 2) # bypasses spec(), as a hand-written dict would
sol_t, _ = solve(datacenter(typo, drain="host1"))
same = len(sol_t["movesSummary"]) > max(1, r // 2) and ends(sol_t) == "UNABLE_TO_FIND_MORE_MOVES"
print(f"misspelled 'stopAfterMove={max(1, r // 2)}' -> {len(sol_t['movesSummary'])} moves, endReason "
f"{ends(sol_t)}: {'ran to the end; the budget was ignored without an error' if same else 'stopped early'}")
try:
spec("localSearchSolverSpec", stopAfterMove=3)
except KeyError as e:
print(f"spec() catches it before the solve: {e.args[0]}")
half = budget_rows[max(1, r // 2)][0]
claim("drain: unbudgeted run empties host1 with no recounted violation",
left0 == 0 and c["cpu"] == c["mem"] == c["rack"] == 0)
claim("drain: stopAfterMoves stops early with endReason HIT_MOVE_LIMIT", "HIT_MOVE_LIMIT" in ends(half))
claim("drain: a misspelled budget key is ignored silently", same)
return (f"host1 emptied in {r} moves; stopAfterMoves={max(1, r // 2)} -> {ends(half)}, host1 holds "
f"{budget_rows[max(1, r // 2)][1]['per_host'][1]}; misspelled key ignored: {same}")
drain_step()
ToFree 제약 조건 하에서 host1을 비워 host0만 rack0에 CPU 16 중 16 상태로 남깁니다. 예산이 없을 때, 로컬 서치는 위반 없이 13번의 이동으로 host1을 비우며, 세 번의 새로운 빌드가 이에 일치합니다. stopAfterMoves 예산은 오류가 아니라 부분 계획을 제공합니다: 12번의 이동에서는 하나의 작업이 여전히 host1에 남아 있고, 6번의 이동에서는 재계산이 CPU 4와 16 GB 초과분 및 랙 중복 6개를 발견하지만 solved는 True입니다. 철자가 잘못된 키인 stopAfterMove는 조용히 무시되며, 솔브는 전체 13번의 이동을 실행합니다.
Copy Code@section("7. Exact (HiGHS) against local search: balance formulas on the datacenter, and mutually exclusive hosts")
def exact_step():
if not HIGHS_OK:
raise RuntimeError("HiGHS failed the gate in step 1")
before = where(DC_START)
ERR = {}
def row(label, make, builds):
sols = []
for _ in range(builds):
try:
sols.append(solve(make()))
except Exception as e:
ERR[label] = clean(e, 60)
print(f" {label:<44} raised {ERR[label]}")
return None
sol = sols[0][0]
c = dc_check(sol["assignment"])
ms = statistics.median(dt for _, dt in sols) * 1000
print(f" {label:<44} CPU spread {c['umax'] - c['umin']:.2f}, violations {c['cpu']:.0f}/{c['mem']:.0f}/"
f"{c['rack']}, moved {moved(before, sol['assignment']):>2}, {ends(sol):<26} median {ms:7.1f} ms "
f"over {builds}")
return sol, c
print("Datacenter model of step 3; spread = max - min CPU utilisation, recounted (objectives of different "
"formulas are not comparable)")
R = {}
R["LINEAR local"] = row("LINEAR, local search", lambda: datacenter(local_search(2)), 3)
R["LINEAR highs"] = row("LINEAR, HiGHS", lambda: datacenter(highs(6)), 2)
R["LINEAR highs+move"] = row("LINEAR, HiGHS, movement goal in tuple 1",
lambda: datacenter(highs(6), movement_tuple=1), 2)
R["SQUARES local"] = row("SQUARES, local search", lambda: datacenter(local_search(2), formula="SQUARES"), 3)
R["SQUARES highs"] = row("SQUARES, HiGHS", lambda: datacenter(highs(6), formula="SQUARES"), 2)
R["IDEAL local"] = row("IDEAL, local search", lambda: datacenter(local_search(2), formula="IDEAL"), 1)
R["IDEAL highs"] = row("IDEAL, HiGHS", lambda: datacenter(highs(3), formula="IDEAL"), 1)
if R["IDEAL local"]:
print(f" IDEAL's own objective at that spread: {goals(R['IDEAL local'][0])} (IDEAL sums each host's squared "
f"utilisation: 8 hosts x 0.5^2 = {8 * 0.5 ** 2:g})")
# ExclusiveScopeItems: three hosts that exclude each other pairwise; one task on each.
print("\nExclusiveScopeItems: host1, host2, host3 pairwise exclusive, one task on each (a broken start)")
pairs = [("host1", "host2"), ("host1", "host3"), ("host2", "host3")]
conflicts = [{"scopeItem": "host1", "conflictingScopeItemsWithOverlap": [{"conflictingScopeItem": "host2"},
{"conflictingScopeItem": "host3"}]},
{"scopeItem": "host2", "conflictingScopeItemsWithOverlap": [{"conflictingScopeItem": "host3"}]}]
def exclusive(start, solver_spec):
p = new_problem(start)
p.add_constraint(spec("exclusiveScopeItemsSpec", name="exclusive_hosts", scope="host",
dimension="task_count", conflictInfoList=conflicts))
p.add_solver(solver_spec)
sol, dt = solve(p)
used = set(sol["assignment"].values())
return sol, dt, sorted(used), sum(a in used and b in used for a, b in pairs)
three = {"host1": ["task1"], "host2": ["task2"], "host3": ["task3"]}
two = {"host1": ["task1", "task2"], "host2": [], "host3": ["task3"]} # the start after one move
ex_res = {}
for label, solver_spec in [("local search", local_search(2)), ("HiGHS", highs(6))]:
sol, dt, used, bad = exclusive(three, solver_spec)
ex_res[label] = bad
print(f" {label:<13} hosts in use {used}, conflicting pairs both in use {bad} of 3, brokenCount "
f"{sol['finalConstraint']['brokenCount']}, solved={sol['finalConstraint']['solved']}, {ends(sol)}, "
f"{dt * 1000:.1f} ms")
bv3 = exclusive(three, local_search(0))[0]["initialConstraint"]["brokenVal"]
after_one = [] # all six single moves from the three-host start
for task, src in (("task1", "host1"), ("task2", "host2"), ("task3", "host3")):
for dst in ("host1", "host2", "host3"):
if dst != src:
state = {h: [t for t in objs if t != task] for h, objs in three.items()}
state[dst] = state[dst] + [task]
after_one.append(exclusive(state, local_search(0))[0]["initialConstraint"]["brokenVal"])
sol2, _, used2, bad2 = exclusive(two, local_search(2))
print(f" why local search stops: the solver's brokenVal is {bv3:g} with three hosts in use (3 conflicting pairs); "
f"after each of the {len(after_one)} possible single moves (2 hosts, 1 pair) it is {sorted(set(after_one))}, so "
f"{'no single move lowers it' if min(after_one) >= bv3 else 'some single move lowers it'}")
print(f" control, local search started from the two-host state: {len(sol2['movesSummary'])} move(s), hosts in use "
f"{used2}, conflicting pairs {bad2}, {ends(sol2)}")
lin_l, lin_h, lin_hm = R["LINEAR local"], R["LINEAR highs"], R["LINEAR highs+move"]
claim("exact: IDEAL balance raises a HiGHS error", R["IDEAL highs"] is None and "HiGHS Error" in ERR.get("IDEAL, HiGHS", ""))
if lin_h and lin_hm:
claim("exact: a tuple-1 movement goal cuts HiGHS's moves without losing balance",
moved(before, lin_hm[0]["assignment"]) < moved(before, lin_h[0]["assignment"])
and lin_hm[1]["umax"] - lin_hm[1]["umin"] == lin_h[1]["umax"] - lin_h[1]["umin"])
claim("exact: HiGHS clears the exclusivity conflicts", ex_res.get("HiGHS") == 0)
claim("exact: local search leaves exclusivity conflicts from the three-host start", ex_res.get("local search", 0) > 0)
claim("exact: no single move from the three-host start lowers the solver's brokenVal", min(after_one) >= bv3)
claim("exact: from the two-host state local search clears them", bad2 == 0)
return (f"LINEAR balance: local search moved {moved(before, lin_l[0]['assignment'])} tasks, HiGHS "
f"{moved(before, lin_h[0]['assignment'])}, HiGHS with a movement tuple {moved(before, lin_hm[0]['assignment'])}"
f"; IDEAL under HiGHS raised: {R['IDEAL highs'] is None}; exclusivity conflicts left: local search "
f"{ex_res.get('local search')} (0 from a two-host start), HiGHS {ex_res.get('HiGHS')}")
exact_step()
우리는 재계산된 CPU 분산으로 판단하여 균형 측면에서 HiGHS를 지역 탐색(local search)과 비교합니다. 수식 간 목적값은 비교할 수 없기 때문입니다. LINEAR 수식에서는 두 방법 모두 분산 0에 도달하지만, 지역 탐색은 13개의 작업을 중앙값 1.8 ms에 이동한 반면 HiGHS는 16개를 7.8 ms에 이동했습니다. IDEAL은 지역 탐색에서는 균형을 이루지만 HiGHS에서는 오류를 발생시킵니다. 서로 배타적인 세 호스트에서는 단일 이동으로는 brokenVal이 낮아지지 않아 지역 탐색이 세 호스트가 모두 사용 중이고 충돌하는 모든 쌍이 여전히 깨진 상태로 멈춥니다. HiGHS는 5.9 ms에 충돌을 해소합니다.
Copy CodeCHILD = r'''
import sys, json, time, math, random, resource, hashlib, collections
from rebalancer import ProblemSolver
mode, N, M, cap, seed = sys.argv[1], int(sys.argv[2]), int(sys.argv[3]), int(sys.argv[4]), int(sys.argv[5])
p = ProblemSolver(service_name="rebalancer_tutorial", service_scope="scale")
p.set_log_level("ERROR")
single_swap = [{"singleMoveTypeSpec": {}}, {"swapMoveTypeSpec": {}}]
if mode == "scale":
rng = random.Random(seed)
cpu = [rng.randint(1, 8) for _ in range(N)]
mem = [rng.randint(1, 32) for _ in range(N)]
start = [min(M - 1, int(M * rng.random() ** 2)) for _ in range(N)] # skewed: low-index containers overloaded
cpu_cap, mem_cap = max(8, math.ceil(sum(cpu) / (0.6 * M))), max(32, math.ceil(sum(mem) / (0.6 * M)))
objs, conts = [f"o{i}" for i in range(N)], [f"c{j}" for j in range(M)]
assign = {c: [] for c in conts}
for i, j in enumerate(start):
assign[conts[j]].append(objs[i])
p.set_object_name("obj").set_container_name("cont").set_assignment(assign)
p.add_object_dimension("cpu", dict(zip(objs, map(float, cpu))))
p.add_object_dimension("mem", dict(zip(objs, map(float, mem))))
p.add_container_dimension("cpu", {}, default_value=float(cpu_cap))
p.add_container_dimension("mem", {}, default_value=float(mem_cap))
p.add_constraint({"capacitySpec": {"name": "cpu_cap", "scope": "cont", "dimension": "cpu", "bound": 1}})
p.add_constraint({"capacitySpec": {"name": "mem_cap", "scope": "cont", "dimension": "mem", "bound": 1}})
p.add_goal({"balanceSpec": {"name": "cpu_balance", "scope": "cont", "dimension": "cpu", "formula": 1}})
p.add_solver({"localSearchSolverSpec": {"moveTypeList": single_swap, "solveTime": cap, "randomSeed": seed}})
else: # the step-3 datacenter with HiGHS, to see whether tied optima come back identical in fresh processes
hosts = [f"host{i}" for i in range(8)]
job_of = {f"job{j}_t{k}": f"job{j}" for j in "ABCD" for k in range(4)}
assign = {h: [] for h in hosts}
assign["host0"] = [t for t, j in job_of.items() if j in ("jobA", "jobB")]
assign["host1"] = [t for t, j in job_of.items() if j == "jobC"]
assign["host2"] = [t for t, j in job_of.items() if j == "jobD"]
p.set_object_name("task").set_container_name("host").set_assignment(assign)
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", {h: f"rack{i // 2}" for i, h in enumerate(hosts)})
for d, v, c in (("cpu", 4.0, 16.0), ("mem", 16.0, 64.0)):
p.add_object_dimension(d, {t: v for t in job_of})
p.add_container_dimension(d, {}, default_value=c)
p.add_constraint({"capacitySpec": {"name": d + "_capacity", "scope": "host", "dimension": d, "bound": 1}})
p.add_constraint({"groupCountSpec": {"name": "one_task_per_job_per_rack", "scope": "rack", "partitionName": "job",
"bound": 1, "limit": {"type": 1, "globalLimit": 1.0}}})
p.add_goal({"balanceSpec": {"name": "cpu_balance", "scope": "host", "dimension": "cpu", "formula": 1}})
p.add_solver({"optimalSolverSpec": {"solverPackage": 2, "solveTime": cap, "suppressLogs": True}})
t = time.perf_counter()
sol = p.solve()
dt = time.perf_counter() - t
a = sol["assignment"]
out = {"wall": dt, "end": [r["endReason"] for r in sol["solverSummaries"]], "objective": sol["finalGlobalObjective"]["goals"][0]["value"],
"broken": [sol["initialConstraint"]["brokenCount"], sol["finalConstraint"]["brokenCount"]],
"hash": hashlib.sha256(json.dumps(sorted(a.items())).encode()).hexdigest()[:12],
"rss_mb": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / (1024 ** 2 if sys.platform == "darwin" else 1024)}
if mode == "scale":
lc, lm = collections.Counter(), collections.Counter()
for i, o in enumerate(objs):
lc[a[o]] += cpu[i]; lm[a[o]] += mem[i]
out["over_cap"] = sum(lc[c] > cpu_cap or lm[c] > mem_cap for c in conts)
out["over_cap_start"] = sum(sum(cpu[i] for i, j in enumerate(start) if j == k) > cpu_cap or
sum(mem[i] for i, j in enumerate(start) if j == k) > mem_cap for k in range(M))
out["moved"] = sum(a[objs[i]] != conts[j] for i, j in enumerate(start))
out["spread"] = (max(lc[c] for c in conts) - min(lc[c] for c in conts)) / cpu_cap
print(json.dumps(out))
'''
@section("8. Scale and repeatability: synthetic overloaded clusters, one fresh process per run")
def scale_step():
child = os.path.join(tempfile.mkdtemp(), "rebalancer_child.py")
open(child, "w").write(CHILD)
def run(mode, n, m, cap):
r = subprocess.run([sys.executable, "-P", child, mode, str(n), str(m), str(cap), "42"], capture_output=True,
text=True, timeout=cap + 120)
if r.returncode != 0:
raise RuntimeError(f"child exited {r.returncode}: {r.stderr.strip()[-200:]}")
out = json.loads(r.stdout.strip().splitlines()[-1])
SOLVE_SECONDS[CURRENT[0]] += out["wall"]
return out
print("objects x containers, cap | solve s | endReason | over-cap containers start -> end | brokenCount "
"| moved | CPU spread | peak RSS")
rows = []
for n, m, cap, reps in [(1000, 100, 10, 3), (1000, 1000, 15, 1), (10000, 100, 25, 1)]:
for _ in range(reps):
o = run("scale", n, m, cap)
rows.append(((n, m, cap), o))
e = "/".join(END.get(x, str(x)) for x in o["end"])
print(f"{n:>6} x {m:<5} {cap:>3} s | {o['wall']:7.2f} | {e:<20} | {o['over_cap_start']:>12} -> "
f"{o['over_cap']:<4} | {o['broken'][0]} -> {o['broken'][1]} | {o['moved']:>5} | "
f"{o['spread']:10.3f} | {o['rss_mb']:5.0f} MB")
first = [o for k, o in rows if k == (1000, 100, 10)]
print(f"\n1000 x 100 in 3 fresh processes: {len({round(o['objective'], 9) for o in first})} distinct objective(s), "
f"{len({o['hash'] for o in first})} distinct assignment(s)")
hs = []
if not HIGHS_OK:
print("datacenter with HiGHS: skipped, HiGHS failed the step-1 gate")
else:
hs = [run("highs", 0, 0, 5) for _ in range(3)]
print(f"datacenter with HiGHS in 3 fresh processes: objectives {[round(o['objective'], 6) for o in hs]}, "
f"{len({o['hash'] for o in hs})} distinct assignment(s), endReason "
f"{sorted({END.get(x, str(x)) for o in hs for x in o['end']})}")
big = rows[-1][1]
claim("scale: 1000 x 100 has no over-capacity container after the solve, in every process",
all(o["over_cap"] == 0 for o in first))
return (f"1,000 x 100: {statistics.median(o['wall'] for o in first):.2f} s median, {first[0]['over_cap_start']} -> "
f"{max(o['over_cap'] for o in first)} over-capacity containers, {len({o['hash'] for o in first})} assignment(s) "
f"in 3 processes; 10,000 x 100: {'/'.join(END.get(x, str(x)) for x in big['end'])} at {rows[-1][0][2]} s with "
f"{big['over_cap_start']} -> {big['over_cap']} over capacity; "
+ (f"HiGHS tie: {len({o['hash'] for o in hs})} assignments in {len(hs)} processes, "
f"{len({round(o['objective'], 9) for o in hs})} objective value(s)" if hs else "HiGHS skipped"))
scale_step()
합성 클러스터를 시간 제한 하에서 각각 새 프로세스로 실행합니다. 1,000개 객체, 100개 컨테이너 클러스터는 세 프로세스에서 각각 1.22 s, 1.27 s, 1.35 s가 걸렸고, 매번 동일한 할당을 반환했으며 13개의 과부하 컨테이너를 모두 해소했습니다. 1,000 x 1,000 클러스터는 3.55 s가 걸렸습니다. 100개 컨테이너에 10,000개 객체인 경우 25 s 제한에 도달하여 HIT_TIME_LIMIT가 반환되므로, endReason과 재계산을 통해 완료된 계획과 부분 계획을 구별합니다. 데이터센터에 대한 세 번의 HiGHS 실행은 세 가지 서로 다른 할당으로 목적값 0에 도달했으므로, 우리는 할당이 아닌 목적값을 비교합니다.
Copy CodeORLIB = "https://people.brunel.ac.uk/~mastjjb/jeb/orlib/"
def fetch(url, tries=4):
for k in range(tries):
try:
with urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": "rebalancer-tutorial"}),
timeout=60) as r:
return r.read()
except Exception as e:
if k == tries - 1:
raise
print(f" retry {k + 1} for {url.rsplit('/', 1)[-1]}: {clean(e, 50)}")
time.sleep(2 * 2 ** k)
def parse_gap(text):
tok = text.split()
P, i, out = int(tok[0]), 1, []
for _ in range(P):
m, n = int(tok[i]), int(tok[i + 1])
i += 2
profit = [[int(tok[i + r * n + j]) for j in range(n)] for r in range(m)]
i += m * n
weight = [[int(tok[i + r * n + j]) for j in range(n)] for r in range(m)]
i += m * n
cap = [int(tok[i + r]) for r in range(m)]
i += m
out.append((m, n, profit, weight, cap))
assert i == len(tok), "every token of the file must be consumed"
return out
# Published optima (maximization), OR-Library gapinfo.html; the cell re-reads the page and checks every value.
GAP_OPTIMA = {"gap1": [336, 327, 339, 341, 326], "gap2": [434, 436, 420, 419, 428], "gap3": [580, 564, 573, 570, 564],
"gap4": [656, 644, 673, 647, 664], "gap5": [563, 558, 564, 568, 559], "gap6": [761, 759, 758, 752, 747],
"gap7": [942, 949, 968, 945, 951], "gap8": [1133, 1134, 1141, 1117, 1127],
"gap9": [709, 717, 712, 723, 706], "gap10": [958, 963, 960, 947, 947],
"gap11": [1139, 1178, 1195, 1171, 1171], "gap12": [1451, 1449, 1433, 1447, 1446]}
def gap_problem(inst, solver_spec):
m, n, profit, weight, cap = inst
agents, jobs = [f"agent{i}" for i in range(m)], [f"job{j}" for j in range(n)]
used, start = [0] * m, {a: [] for a in agents}
for j in range(n): # first fit, least-loaded fallback; the solver repairs any overload
fits = [i for i in range(m) if used[i] + weight[i][j] <= cap[i]]
i = fits[0] if fits else min(range(m), key=lambda k: used[k] - cap[k])
start[agents[i]].append(jobs[j])
used[i] += weight[i][j]
p = new_problem(start, objects="job", containers="agent")
# A job's resource use depends on which agent it sits on: a dynamic object dimension keyed by the agent scope.
p.add_dynamic_object_dimension("resource", "agent",
{agents[i]: {jobs[j]: float(weight[i][j]) for j in range(n)} for i in range(m)},
default_value=0.0)
p.add_container_dimension("resource", {agents[i]: float(cap[i]) for i in range(m)}, default_value=0.0)
p.add_constraint(spec("capacitySpec", name="agent_capacity", scope="agent", dimension="resource",
bound=MAX_BOUND))
p.add_goal(spec("assignmentAffinitiesSpec", name="profit", scope="agent",
affinities=[{"objectName": jobs[j], "scopeItemName": agents[i], "affinity": float(profit[i][j])}
for i in range(m) for j in range(n)]), weight=1.0)
p.add_solver(solver_spec)
return p, agents, jobs
def gap_recount(inst, assignment, agents, jobs):
m, n, profit, weight, cap = inst
idx = {a: i for i, a in enumerate(agents)}
use = [0] * m
for j, job in enumerate(jobs):
use[idx[assignment[job]]] += weight[idx[assignment[job]]][j]
total = sum(profit[idx[assignment[job]]][j] for j, job in enumerate(jobs))
return total, all(use[i] <= cap[i] for i in range(m)) and len(assignment) == n
@section("9. Generalised assignment: 60 OR-Library instances against their published optima")
def gap_step():
page = fetch(ORLIB + "gapinfo.html").decode("latin-1")
published = {name: int(v) for name, v in re.findall(r"(c\d+-\d)\s+(\d+)", re.sub(r"<[^>]+>", " ", page))}
data, digest, nbytes = {}, hashlib.sha256(), 0
for g in GAP_OPTIMA:
raw = fetch(ORLIB + f"files/{g}.txt")
digest.update(raw)
nbytes += len(raw)
data[g] = parse_gap(raw.decode())
pinned = digest.hexdigest()[:12] == "54513b5b96f6" # the files this notebook was verified with
print(f"gap1..gap12: {nbytes / 1024:.0f} KB, sha256 of the 12 files {digest.hexdigest()[:12]} (same as the verified "
f"copy: {pinned}), {sum(len(v) for v in data.values())} instances")
if not pinned:
raise RuntimeError("the OR-Library GAP files differ from the ones this notebook was verified with")
# The page names each instance c<agents><jobs>-<k> (c515-1 is 5 agents, 15 jobs, problem 1).
names = {g: [f"c{m}{n}-{k + 1}" for k, (m, n, *_) in enumerate(v)] for g, v in data.items()}
same = all(published.get(nm) == GAP_OPTIMA[g][k] for g in data for k, nm in enumerate(names[g]))
print(f"gapinfo.html: {len(published)} published optima, matched by instance name to the table in this notebook: "
f"{same}")
if not same:
raise RuntimeError("the optima on gapinfo.html differ from the embedded table")
guard = {"HiGHS": 90.0, "local": 120.0} # seconds; a solve starts only while its cap still fits
print(f"\n{'group':<6}{'size':>7} | {'HiGHS = optimum':>15}{'slowest':>9} | local search = optimum "
f"(seed 1/2/3, best){'':>2} mean shortfall")
tot = collections.Counter()
shortfalls, relative, disagree, worst = [], [], 0, 0.0
for g, insts in data.items():
hit_h, slow, hit, group_short, group_h = 0, 0.0, [0, 0, 0, 0], [], 0
for k, inst in enumerate(insts):
opt = GAP_OPTIMA[g][k]
sol = None
if HIGHS_OK and guard["HiGHS"] >= 10:
p, agents, jobs = gap_problem(inst, highs(10))
try:
sol, dt = solve(p)
except Exception as e: # a HiGHS failure costs this instance's exact run, not the whole step
print(f" {g}#{k + 1}: HiGHS raised {clean(e, 60)}")
tot["h_err"] += 1
if sol is not None:
guard["HiGHS"] -= dt
tot["h_sec"] += dt
val, feas = gap_recount(inst, sol["assignment"], agents, jobs)
slow, worst = max(slow, dt), max(worst, dt)
tot["h_run"] += 1
group_h += 1
tot["h_feas"] += feas
tot["h_optimal_end"] += ends(sol) == "OPTIMAL"
hit_h += feas and val == opt
if feas and val > opt:
raise RuntimeError(f"{g}#{k + 1}: HiGHS profit {val} exceeds the published optimum {opt}")
vals = []
for seed in (1, 2, 3):
if guard["local"] < 3:
continue
p, agents, jobs = gap_problem(inst, local_search(3, seed=seed, moves=("single", "swap", "triple", "kl")))
sol, dt = solve(p)
guard["local"] -= dt
tot["l_sec"] += dt
val, feas = gap_recount(inst, sol["assignment"], agents, jobs)
if val > opt and feas:
raise RuntimeError(f"{g}#{k + 1}: local-search profit {val} exceeds the published optimum {opt}")
tot["l_run"] += 1
tot["l_feas"] += feas
vals.append(val if feas else None)
hit[seed - 1] += feas and val == opt
best = max((v for v in vals if v is not None), default=None)
if len(vals) == 3 and best is not None:
hit[3] += best == opt
shortfalls.append(opt - best)
group_short.append(opt - best)
relative.append((opt - best) / opt)
disagree += len(set(vals)) > 1
tot["h_hit"] += hit_h
for j in range(4):
tot[f"l_hit{j}"] += hit[j]
m, n = insts[0][0], insts[0][1]
gs = f"{statistics.mean(group_short):6.1f}" if group_short else " -"
print(f"{g:<6}{f'{m}x{n}':>7} | {hit_h:>11} of {group_h}{slow:>8.2f}s | {hit[0]:>4} {hit[1]:>4} {hit[2]:>4} {hit[3]:>6}"
f"{'':>20}{gs}")
if not HIGHS_OK:
print("\nHiGHS: skipped, it failed the step-1 gate")
else:
print(f"\nHiGHS: {tot['h_hit']} of {tot['h_run']} equal the published optimum; endReason OPTIMAL "
f"{tot['h_optimal_end']} of {tot['h_run']}; recounted feasible {tot['h_feas']} of {tot['h_run']}; raised "
f"{tot['h_err']}; slowest {worst:.2f} s, {tot['h_sec']:.1f} s in total")
mean_short = statistics.mean(shortfalls) if shortfalls else float("nan")
print(f"local search (single, swap, triple loop, Kernighan-Lin; first-fit start; 3 s cap): seed 1/2/3 hit "
f"{tot['l_hit0']}/{tot['l_hit1']}/{tot['l_hit2']}, best of three {tot['l_hit3']} of {len(shortfalls)}; "
f"best-of-three shortfall mean {mean_short:.1f} "
f"({100 * statistics.mean(relative) if relative else float('nan'):.2f}% of the optimum), "
f"max {max(shortfalls, default=0)}; seeds disagree on {disagree} "
f"instances; recounted feasible {tot['l_feas']} of {tot['l_run']}; {tot['l_sec']:.1f} s in total")
if (HIGHS_OK and tot["h_run"] < 60) or len(shortfalls) < 60:
print(f"PARTIAL: {tot['h_run']} HiGHS and {len(shortfalls)} local-search instances completed (time guard or "
f"errors); the 60-of-60 comparison does not apply")
claim("GAP: all 60 HiGHS runs ran and equal the published optimum", tot["h_hit"] == tot["h_run"] == 60)
claim("GAP: all 180 local-search answers ran and are feasible by recount", tot["l_feas"] == tot["l_run"] == 180)
return ((f"HiGHS {tot['h_hit']}/{tot['h_run']} published optima (slowest {worst:.2f} s)" if HIGHS_OK else
"HiGHS skipped") + f"; local search best-of-3 {tot['l_hit3']}/{len(shortfalls)}, best-of-3 shortfall "
f"mean {mean_short:.1f}")
gap_step()
우리는 공개된 최적해를 가진 OR-Library의 60개 일반화 할당 인스턴스에 대해 솔버를 검증하며, 실행 시점에 다운로드하고 체크섬을 확인합니다. HiGHS는 60개 최적해 전부와 일치했고, 모든 답이 실행 가능했으며, 가장 느린 경우 1.31 s였습니다. 첫 적합(first-fit) 시작점에서 출발한 지역 탐색은 세 시드 중 최선의 시드를 사용해도 60개 중 4개에서만 최적해에 도달했고, 인스턴스별 최선의 답은 평균 7.7 이익 단위(1.07%) 부족했습니다. 아홉 번의 실행에서 평균 부족분은 7.6에서 8.1 사이였고, 고정된 시드조차 원인을 규명하지 못한 이유로 실행마다 달라졌습니다.
Copy Codedef parse_binpack(text, scale=1):
tok = text.split()
name, C, n, best = tok[1], float(tok[2]), int(tok[3]), int(tok[4])
sizes = [float(x) for x in tok[5:5 + n]]
assert len(sizes) == n, "the file holds fewer sizes than it declares"
S_int = [round(s * scale) for s in sizes]
assert abs(sum(S_int) - scale * sum(sizes)) < 1e-6, "scaled sizes must stay exact"
return name, round(C * scale), S_int, best
def bin_problem(sizes, C, start, solver_spec, minimise):
p = new_problem(start, objects="item", containers="bin")
items = [f"item{i}" for i in range(len(sizes))]
p.add_object_dimension("size", dict(zip(items, map(float, sizes))))
p.add_object_dimension("count", {it: 1.0 for it in items})
p.add_container_dimension("size", {}, default_value=float(C))
p.add_constraint(spec("capacitySpec", name="bin_capacity", scope="bin", dimension="size", bound=MAX_BOUND))
if minimise:
p.add_goal(spec("minimizeContainersSpec", name="fewest_bins", scope="bin", dimension="count"), weight=1.0)
p.add_solver(solver_spec)
return p
def pack(sizes, C, K=None):
"""First-fit decreasing; with K given, first fit into K bins with least-loaded fallback (may overflow)."""
order = sorted(range(len(sizes)), key=lambda i: -sizes[i])
loads, bins = [], []
if K:
loads, bins = [0] * K, [[] for _ in range(K)]
for i in order:
fits = [b for b in range(len(loads)) if loads[b] + sizes[i] <= C]
if fits:
b = fits[0]
elif K:
b = min(range(K), key=lambda q: loads[q])
else:
loads.append(0)
bins.append([])
b = len(loads) - 1
loads[b] += sizes[i]
bins[b].append(f"item{i}")
return {f"bin{b}": bins[b] for b in range(len(bins))}
def bins_used(assignment, sizes, C):
L = collections.Counter()
for it, b in assignment.items():
L[b] += sizes[int(it[4:])]
return sum(1 for v in L.values() if v > 0), all(v <= C for v in L.values()) and len(assignment) == len(sizes)
@section("10. Bin packing: where local search with MinimizeContainers stops, measured against the L1 bound")
def binpack_step():
rows = []
for fname, scale in (("binpack1.txt", 1), ("binpack5.txt", 10)):
raw = fetch(ORLIB + "files/" + fname)
if hashlib.sha256(raw).hexdigest()[:12] != {"binpack1.txt": "891fe4b3be86", "binpack5.txt": "33cb019a3f05"}[fname]:
raise RuntimeError(f"{fname} differs from the copy this notebook was verified with")
name, C, sizes, best = parse_binpack(raw.decode(), scale)
L1 = -(-sum(sizes) // C)
ffd = pack(sizes, C)
print(f"{name} ({fname}, {len(raw) / 1024:.1f} KB, sha256 {hashlib.sha256(raw).hexdigest()[:12]} as verified): "
f"{len(sizes)} items, capacity {C}"
f"{' (sizes x10)' if scale > 1 else ''}; L1 lower bound {L1} (file's best known {best}); first-fit "
f"decreasing {len(ffd)} bins")
sol, dt = solve(bin_problem(sizes, C, ffd, local_search(5, seed=1, moves=("single", "swap", "triple", "kl")),
True))
used, ok = bins_used(sol["assignment"], sizes, C)
print(f" A MinimizeContainers, local search from FFD: {used} bins (gap to L1 {used - L1}), feasible {ok}, "
f"{ends(sol)}, {dt:.2f} s")
startK = pack(sizes, C, K=L1)
solB, dtB = solve(bin_problem(sizes, C, startK, local_search(5, seed=1, moves=("single", "swap", "triple",
"kl")), False))
usedB, okB = bins_used(solB["assignment"], sizes, C)
print(f" B exactly L1 = {L1} bins, capacity only, local search: feasible {okB} ({usedB} bins used), "
f"{ends(solB)}, {dtB:.2f} s")
okC = "skipped"
if not HIGHS_OK:
print(" C HiGHS: skipped, it failed the step-1 gate")
else:
solC, dtC = solve(bin_problem(sizes, C, startK, highs(10), False))
usedC, okC = bins_used(solC["assignment"], sizes, C)
print(f" C exactly L1 = {L1} bins, capacity only, HiGHS: feasible {okC}, {ends(solC)}, {dtC:.2f} s "
f"(a time limit is 'not found in 10 s', not a proof of infeasibility)")
rows.append((name, L1, len(ffd), used, okB, okC, ok))
claim("bin packing: MinimizeContainers local search stays above L1, feasibly, on at least one instance",
any(u > L and okA for _, L, _, u, _, _, okA in rows))
claim("bin packing: capacity-only local search with exactly L1 bins packs u120_00 (optimal by the bound)",
rows[0][4] is True)
return "; ".join(f"{n}: L1 {L}, FFD {f}, MinimizeContainers {u}, fixed-L1 local feasible {b}, HiGHS {c}"
for n, L, f, u, b, c, _ in rows)
binpack_step()
두 개의 OR-Library 인스턴스에서 MinimizeContainers로 빈 패킹을 테스트하고, 가장 잘 알려진 빈 개수인 48과 20과 같은 L1 하한과 비교 측정합니다. 첫 적합 감소(first-fit decreasing)에서 출발한 지역 탐색은 어느 쪽에서도 빈을 하나도 줄이지 못했습니다. 용량만으로 정확히 L1개의 빈을 요청하면 u120_00을 0.02 s에 패킹하는데, 이는 하한과 일치하므로 최적입니다. t60_00에서는 지역 탐색이 20-빈 패킹을 찾지 못했고, HiGHS도 어느 인스턴스에서든 10 s 내에 찾지 못했는데, 이는 비실현성의 증명이 아니라 시간 초과입니다.
Copy Codebanner("SUMMARY")
for name, res in RESULTS.items():
print(f" {name:<100s} {res}")
print("\nClaims, each decided in this run by a recount, an endReason or a raised error, as the claim says:")
for step, text, held in CLAIMS:
print(f" [{'held' if held else 'DID NOT HOLD'}] step {step}: {text}")
failed = [k for k, v in RESULTS.items() if str(v).startswith("SKIPPED")]
if not HIGHS_OK:
print("\nHiGHS failed the step-1 gate: step 7 did not run, and the HiGHS parts of steps 8, 9 and 10 were skipped")
print(f"\n{sum(h for *_, h in CLAIMS)} of {len(CLAIMS)} claims held; steps failed or skipped: {failed or 'none'}")
print(f"seconds inside solve() per step: {dict((k, round(v, 2)) for k, v in sorted(SOLVE_SECONDS.items(), key=lambda kv: int(kv[0])))}"
f", total {sum(SOLVE_SECONDS.values()):.1f} s")
print(f"run on: {HW}; Python {platform.python_version()}; rebalancer {VERSION}")
print("not tested here: production scale and the paper's or Meta's latency and volume figures; XPRESS and Gurobi "
"(absent from the wheel); thread control; GAP classes a-d; public cluster traces.")
요약은 모든 주장을 재계산, endReason 또는 발생한 오류를 통한 검증 방식과 단계별 솔버 소요 초와 함께 나열합니다. 이번 실행에서 28개 주장이 모두 유지되었고 어떤 단계도 실패하지 않았으며, 솔버 내부에서 90.1 s가 소요되었습니다. 여기서 테스트되지 않은 것은 프로덕션 규모, 논문 또는 Meta의 지연 시간과 처리량 수치, 그리고 XPRESS 및 Gurobi 백엔드입니다.
Rebalancer는 배치를 읽기 쉬운 모델로 변환하고, 지역 탐색 또는 정확한 HiGHS 솔버로 해를 구합니다. 이 모델들에서 Rebalancer는 8개 호스트 클러스터를 랙에 걸쳐 재분산하고, 바쁜 호스트를 비웠으며, HiGHS를 통해 공개된 60개 OR-Library 최적해를 모두 일치시켰고, M4에서 1,000개 객체, 100개 컨테이너 클러스터를 약 1.3 s에 복구했습니다. 이 실행은 함정들도 보여주었습니다: solved와 OPTIMAL이 위반된 제약 조건과 공존할 수 있고, HARD는 솔버에 따라 다르며, 잘못 쓴 키는 무시되고, 열거형 이름은 거부되며, IDEAL은 HiGHS에서 실패하고, 목표 튜플의 순서가 깨진 제약 조건이 고쳐지는지를 결정합니다. 열거형은 정수로 전달하고, spec 키를 확인하고, 모든 결과를 재계산하고, 할당이 아닌 목적값을 비교하십시오.
다음을 확인해 보십시오: FULL CODES here. 이 프로젝트의 모든 크레딧은 해당 연구자에게 돌아갑니다. 또한 저희를 Twitter 에서 팔로우하고 저희 150k+ML SubReddit 에 가입하고 our Newsletter를 구독하는 것도 잊지 마세요. 잠깐! 텔레그램을 사용하시나요? 이제 텔레그램에서도 저희와 함께하실 수 있습니다.
이 글 How to Use Meta’s Rebalancer for Task Placement & Optimization 은 다음에서 처음 게시되었습니다: MarkTechPost.