Planner scheduler and simulator

cuebot/scheduler-sim/.gitignore

@@ -0,0 +1,7 @@
+# Generated by setup.sh / simulate.py — never commit these.
+venv/
+opencue_proto/
+sim_hosts
+scheduler_sim.yaml
+resolve_local.so
+*.log

cuebot/scheduler-sim/BUILD.md

@@ -0,0 +1,78 @@
+# Building & running cuebot in this dev box (hard-won notes)
+
+This box has a toolchain trap. These are the exact steps that work.
+
+## The toolchain trap
+- The repo pins **Gradle 7.6.2** (cuebot/gradle/wrapper). It does NOT run on the
+  box's default **JDK 21** ("Unsupported class file major version 65" — bundled
+  ASM too old). The standalone **Gradle 8.14.3** in /opt runs on 21 but is too
+  new for the Spring Boot 2.2.1 plugin ("ArchivePublishArtifact").
+- Correct combo: **wrapper Gradle 7.6.2 + JDK 17**.
+
+## JDK 17 (with the proxy CA)
+A vanilla JDK 17 download can't fetch deps: the env's outbound proxy uses a TLS
+CA that vanilla cacerts don't trust (Gradle reports "plugin not found"). Fix:
+copy the managed JDK 21 truststore into the JDK 17.
+```
+# /tmp/jdk-17.0.2 was unpacked from openjdk-17.0.2_linux-x64; then:
+cp /usr/lib/jvm/java-21-openjdk-amd64/lib/security/cacerts /tmp/jdk-17.0.2/lib/security/cacerts
+```
+
+## Repos: drop the dead ones (build-time only, do NOT commit)
+cuebot/settings.gradle (pluginManagement) and build.gradle list `jcenter()` and
+`repo.spring.io/plugins-snapshot`, which are dead and break resolution on 7.6.2.
+Strip them before building:
+```
+# in cuebot/: remove the 'maven { url ".../plugins-snapshot" }' line and 'jcenter()' lines
+```
+
+## Postgres (must run as non-root; refuses root)
+```
+# run these as your own (non-root) user; postgres refuses root, no sudo needed
+PGBIN=/usr/lib/postgresql/16/bin
+rm -rf /tmp/pgdata && mkdir -p /tmp/pgdata /tmp/pgrun
+$PGBIN/initdb -D /tmp/pgdata -U cue --auth=trust
+$PGBIN/pg_ctl -D /tmp/pgdata -o "-p 5433 -k /tmp/pgrun -c listen_addresses=127.0.0.1" -l /tmp/pg.log start
+$PGBIN/psql -h127.0.0.1 -p5433 -Ucue -dpostgres -c "CREATE DATABASE cuebot;"
+# apply migrations in version order:
+cd cuebot/src/main/resources/conf/ddl/postgres/migrations
+for f in $(ls *.sql | sort -t_ -k1.2 -n); do $PGBIN/psql -h127.0.0.1 -p5433 -Ucue -dcuebot -v ON_ERROR_STOP=1 -q -f "$f"; done
+# base data: dept/services/config from seed_data.sql + scheduler-sim/sim_seed.sql
+```
+
+## Build / run cuebot (as your own user, JDK 17, wrapper 7.6.2)
+A dedicated gradle home /tmp/ghome-$USER holds the resolved deps. Build as the
+user that owns the checkout (a fresh `git clone` already is); no specific account
+is required. Remove cuebot/.gradle if you hit `checksums.lock (Permission denied)`.
+```
+cd cuebot && env \
+  CUEBOT_DB_URL="jdbc:postgresql://127.0.0.1:5433/cuebot" CUEBOT_DB_USER=cue CUEBOT_DB_PASSWORD= \
+  SCHEDULER_ENABLED=true SCHEDULER_INTERVAL_MS=250 SCHEDULER_RESERVATIONS_ENABLED=false \
+  ./gradlew bootRun -g /tmp/ghome-$USER -Dorg.gradle.java.home=/tmp/jdk-17.0.2 --console=plain >/tmp/cuebot.log 2>&1
+```
+gRPC serves on **8443**. Compile-only check: swap `bootRun` for `compileJava`
+(note `-Werror -Xlint:all` is on — warnings fail the build). Unit tests:
+`./gradlew test --tests "...SchedulerTests"`.
+
+## CRITICAL launch pattern (process management)
+Launch long-running procs (cuebot, pinger, fake_rqd) with the Bash tool's
+`run_in_background: true` and **NO inner `&`**. An inner `&` double-backgrounds
+and the JVM/Python gets SIGKILLed when the wrapper shell exits.
+After a cuebot restart, restart status_pinger.py too (its gRPC channel goes
+stale -> all ReportStatus fail -> hosts age to DOWN).
+
+## Reset the farm between runs
+```
+psql ... -c "DELETE FROM proc;"
+psql ... -c "UPDATE host SET int_cores_idle=int_cores,int_mem_idle=int_mem,int_gpus_idle=int_gpus,int_gpu_mem_idle=int_gpu_mem;"
+psql ... -c "UPDATE subscription SET int_cores=0,int_gpus=0;"
+# clear the job backlog (frames then layers) for the sim show:
+psql ... -c "DELETE FROM frame f USING job j WHERE f.pk_job=j.pk_job AND j.pk_show='10000000-0000-0000-0000-000000000003';"
+psql ... -c "DELETE FROM layer l USING job j WHERE l.pk_job=j.pk_job AND j.pk_show='10000000-0000-0000-0000-000000000003';"
+```
+
+## Gotcha: "unable to allocate additional memory"
+That is NOT a Postgres OOM. It's `trigger__verify_host_resources` raising when a
+booking pushes a host's int_*_idle below 0 (overbooking protection). Treat it as
+an overbooking/accounting signal, not a memory problem.
+```

cuebot/scheduler-sim/analysis/.gitignore

@@ -0,0 +1,4 @@
+# Generated benchmark outputs — never commit these.
+*.csv
+*.png
+__pycache__/

cuebot/scheduler-sim/analysis/README.md

@@ -0,0 +1,40 @@
+# scheduler-sim analysis tooling
+
+Ad-hoc benchmarking helpers for comparing scheduler modes (`--mode new|old|rust`)
+under `simulate.py`. **Samplers** record DB/CPU load *during* a run; **plotters**
+turn the recordings into graphs/tables *after*. The driver writes nothing to the
+repo — point everything at a scratch dir.
+
+All plotters read from `$SIM_BENCH_DIR` (default `/tmp/cmp2`) and expect files
+named `<tag>_dbstat.csv`, `<tag>_cpu.csv`, `<tag>_sim.log` per run.
+
+## Samplers (run in the background, alongside `simulate.py`)
+
+- `db_sampler.py <out.csv>` — every 2s, snapshots `pg_stat_database`
+  (commits/rollbacks, tuples read/written, deadlocks) + `pg_stat_activity`
+  (active backends, lock-waiters). Honors `SIM_PG_HOST`/`SIM_PG_PORT`.
+- `cpu_sampler.py <out.csv>` — every 3s, total CPU% + per-process cores for
+  cuebot (java), postgres, cue-scheduler, python (from `/proc`).
+
+```bash
+BENCH=/tmp/bench; mkdir -p $BENCH
+python analysis/db_sampler.py  $BENCH/new_dbstat.csv &
+python analysis/cpu_sampler.py $BENCH/new_cpu.csv &
+python simulate.py --mode new --feed 240 --stats 90 > $BENCH/new_sim.log 2>&1
+kill %1 %2
+```
+
+## Plotters / analyzers (run after; `SIM_BENCH_DIR=$BENCH`)
+
+- `make_graphs.py` — NEW vs RUST: 3 graphs (reads/s, writes/s, DB
+  health = lock-waiters + rollbacks). Tags: `new`, `rust`.
+- `make_graphs_ba.py` — before vs after: same 3 graphs. Tags: `before`, `after`.
+- `analyze_sweep.py` — prints a steady-state median table for the 2×2 grid
+  {new,rust}×{compress2,compress8}. Tags: `before`, `new8`, `rust2`, `rust8`.
+
+```bash
+SIM_BENCH_DIR=$BENCH python analysis/make_graphs.py     # writes g_*.png
+SIM_BENCH_DIR=$BENCH python analysis/analyze_sweep.py   # prints table
+```
+
+Generated `*.csv` / `*.png` are gitignored.

cuebot/scheduler-sim/analysis/analyze_sweep.py

@@ -0,0 +1,66 @@
+"""2x2 grid: {new,rust} x {compress2,compress8}. Steady-state medians (t90-180)."""
+import os
+import statistics as st
+CMP = os.environ.get("SIM_BENCH_DIR", "/tmp/cmp2")
+# logical tag -> file prefix
+CELLS=[("new  @2","before"),("new  @8","new8"),("rust @2","rust2"),("rust @8","rust8")]
+def sod(h):
+    a=list(map(int,h.split(":"))); return a[0]*3600+a[1]*60+a[2]
+def sim_t0(pre):
+    try:
+        for ln in open(f"{CMP}/{pre}_sim.log",errors="ignore"):
+            if "] util=" in ln:
+                return sod(ln[1:9])
+    except FileNotFoundError: return None
+    return None
+def med_sim(pre,key,lo,hi):
+    import re
+    t0=sim_t0(pre)
+    if t0 is None: return None
+    rx=re.compile(r"\[(\d\d:\d\d:\d\d)\].*?"+key+r"=\s*([\d.]+)")
+    v=[]
+    for ln in open(f"{CMP}/{pre}_sim.log",errors="ignore"):
+        m=rx.search(ln)
+        if m:
+            t=sod(m.group(1))-t0
+            if lo<=t<=hi: v.append(float(m.group(2)))
+    return st.median(v) if v else None
+def med_csv(pre,suffix,colidx,lo,hi,rate=False):
+    t0=sim_t0(pre)
+    if t0 is None: return None
+    try: rows=[ln.strip().split(",") for ln in open(f"{CMP}/{pre}_{suffix}.csv",errors="ignore")][1:]
+    except FileNotFoundError: return None
+    data=[]
+    for r in rows:
+        try: data.append((sod(r[0]),[float(x) for x in r[1:]]))
+        except: pass
+    if not data: return None
+    v=[]
+    if rate:
+        for i in range(1,len(data)):
+            (ta,a),(tb,b)=data[i-1],data[i]; dt=tb-ta
+            if dt<=0: continue
+            t=tb-t0
+            if lo<=t<=hi: v.append((b[colidx]-a[colidx])/dt)
+    else:
+        for ts,c in data:
+            t=ts-t0
+            if lo<=t<=hi and colidx < len(c): v.append(c[colidx])
+    return st.median(v) if v else None
+# dbstat cols (after ts): 0 commits,1 rollbacks,2 tup_ret,3 tup_fetch,4 ins,5 upd,6 del,7 deadlk,8 active,9 lockwait
+# cpu cols (after ts): 0 total_cpu%,1 cuebot,2 postgres,3 scheduler,4 python
+LO,HI=90,180
+print(f"{'cell':8} {'util%':>6} {'done/s':>7} {'orphan':>7} {'reads/s':>9} {'writes/s':>9} {'rollbk/s':>8} {'lockwt':>6} {'CPU%':>5} {'pg':>5} {'cuebot':>6} {'rust':>5} {'py':>5}")
+for name,pre in CELLS:
+    util=med_sim(pre,"util",LO,HI); done=med_sim(pre,"done/s",LO,HI); orp=med_sim(pre,"orphan",LO,HI)
+    def rd(): 
+        a=med_csv(pre,"dbstat",2,LO,HI,True); b=med_csv(pre,"dbstat",3,LO,HI,True)
+        return (a+b) if (a is not None and b is not None) else None
+    def wr():
+        xs=[med_csv(pre,"dbstat",i,LO,HI,True) for i in (4,5,6)]
+        return sum(x for x in xs if x is not None) if any(x is not None for x in xs) else None
+    reads=rd(); writes=wr(); rb=med_csv(pre,"dbstat",1,LO,HI,True); lw=med_csv(pre,"dbstat",9,LO,HI,False)
+    cpu=med_csv(pre,"cpu",0,LO,HI,False); pg=med_csv(pre,"cpu",2,LO,HI,False)
+    cb=med_csv(pre,"cpu",1,LO,HI,False); ru=med_csv(pre,"cpu",3,LO,HI,False); py=med_csv(pre,"cpu",4,LO,HI,False)
+    def f(x,d=0): return ("%.{}f".format(d)%x) if x is not None else "-"
+    print(f"{name:8} {f(util):>6} {f(done):>7} {f(orp):>7} {f(reads):>9} {f(writes):>9} {f(rb,1):>8} {f(lw,2):>6} {f(cpu):>5} {f(pg,2):>5} {f(cb,2):>6} {f(ru,2):>5} {f(py,2):>5}")

cuebot/scheduler-sim/analysis/cpu_sampler.py

@@ -0,0 +1,24 @@
+import time, os, sys, glob
+OUT=sys.argv[1]; CLK=os.sysconf("SC_CLK_TCK")
+def snap():
+    f=open("/proc/stat").readline().split()[1:]
+    idle=int(f[3])+int(f[4]); tot=sum(int(x) for x in f)
+    g={"cuebot":0,"postgres":0,"scheduler":0,"python":0}
+    for p in glob.glob("/proc/[0-9]*/stat"):
+        try:
+            d=open(p).read(); comm=d.split("(",1)[1].rsplit(")",1)[0]
+            r=d.rsplit(")",1)[1].split(); j=int(r[11])+int(r[12])
+            if comm=="java": g["cuebot"]+=j
+            elif comm.startswith("postgres"): g["postgres"]+=j
+            elif comm=="cue-scheduler": g["scheduler"]+=j
+            elif comm.startswith("python"): g["python"]+=j
+        except: pass
+    return tot,idle,g
+pt,pi,pg=snap(); pT=time.time()
+open(OUT,"w").write("ts,total_cpu_pct,cuebot_cores,postgres_cores,scheduler_cores,python_cores\n")
+while True:
+    time.sleep(3); t,i,g=snap(); now=time.time(); dt=now-pT; dtot=t-pt
+    cpu=100.0*(dtot-(i-pi))/dtot if dtot>0 else 0
+    c=lambda k:(g[k]-pg[k])/CLK/dt if dt>0 else 0
+    open(OUT,"a").write(f"{time.strftime('%H:%M:%S')},{cpu:.1f},{c('cuebot'):.2f},{c('postgres'):.2f},{c('scheduler'):.2f},{c('python'):.2f}\n")
+    pt,pi,pg,pT=t,i,g,now

cuebot/scheduler-sim/analysis/db_sampler.py

@@ -0,0 +1,20 @@
+import time, subprocess, sys, os
+OUT = sys.argv[1]
+_PORT = os.environ.get("SIM_PG_PORT", "5433")
+_HOST = os.environ.get("SIM_PG_HOST", "127.0.0.1")
+PSQL = ["psql","-tA","-h",_HOST,"-p",_PORT,"-U","cue","-d","cuebot","-c"]
+def q(sql): return subprocess.run(PSQL+[sql], capture_output=True, text=True).stdout.strip()
+with open(OUT,"w") as f:
+    f.write("ts,commits,rollbacks,tup_ret,tup_fetch,ins,upd,del,deadlocks,"
+            "blks_read,blks_hit,active,lockwait\n"); f.flush()
+    while True:
+        db = q("SELECT xact_commit||','||xact_rollback||','||tup_returned||','||tup_fetched"
+               "||','||tup_inserted||','||tup_updated||','||tup_deleted||','||deadlocks"
+               "||','||blks_read||','||blks_hit "
+               "FROM pg_stat_database WHERE datname='cuebot'")
+        act = q("SELECT count(*) FILTER (WHERE state='active')||','||"
+                "count(*) FILTER (WHERE wait_event_type='Lock') "
+                "FROM pg_stat_activity WHERE datname='cuebot' AND pid<>pg_backend_pid()")
+        if db and act:
+            f.write(f"{time.strftime('%H:%M:%S')},{db},{act}\n"); f.flush()
+        time.sleep(2)

cuebot/scheduler-sim/analysis/make_graphs.py

@@ -0,0 +1,90 @@
+"""NEW vs RUST -- 3 graphs: DB reads/s, DB writes/s, DB health (contention)."""
+import os
+import re
+import matplotlib; matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+
+# Directory holding <tag>_dbstat.csv and <tag>_sim.log produced by a run. Override
+# with SIM_BENCH_DIR; defaults to the legacy scratch dir used during development.
+CMP = os.environ.get("SIM_BENCH_DIR", "/tmp/cmp2")
+RUNS = [("new", "new (inline-dispose)", "#1f77b4"), ("rust", "rust", "#d62728")]
+WATCH = 185
+
+def sod(h):
+    a = list(map(int, h.split(":"))); return a[0]*3600 + a[1]*60 + a[2]
+
+def util_series(tag):
+    rx = re.compile(r"\[(\d\d:\d\d:\d\d)\]\s*util=\s*([\d.]+)%"); pts = []
+    try:
+        for ln in open(f"{CMP}/{tag}_sim.log", errors="ignore"):
+            m = rx.search(ln)
+            if m: pts.append((sod(m.group(1)), float(m.group(2))))
+    except FileNotFoundError: return []
+    if not pts: return []
+    t0 = pts[0][0]; return [(t-t0, v) for t, v in pts if 0 <= t-t0 <= WATCH]
+
+def rows(tag):
+    out = []
+    try: f = open(f"{CMP}/{tag}_dbstat.csv", errors="ignore")
+    except FileNotFoundError: return []
+    f.readline()
+    for ln in f:
+        p = ln.strip().split(",")
+        if len(p) < 11: continue
+        try: out.append((sod(p[0]), [int(x) for x in p[1:11]]))
+        except ValueError: continue
+    return out
+# cols after ts: 0 commits,1 rollbacks,2 tup_ret,3 tup_fetch,4 ins,5 upd,6 del,7 deadlocks,8 active,9 lockwait
+
+def rates(tag):
+    r = rows(tag)
+    o = {k: [] for k in ("reads","writes","rollbacks","deadlocks","lockwait","active")}
+    if not r: return o
+    t0 = r[0][0]
+    for i in range(1, len(r)):
+        (ta, a), (tb, b) = r[i-1], r[i]; dt = tb-ta
+        if dt <= 0: continue
+        rt = tb-t0
+        if not (0 <= rt <= WATCH+30): continue
+        o["reads"].append((rt, ((b[2]-a[2])+(b[3]-a[3]))/dt))
+        o["writes"].append((rt, ((b[4]-a[4])+(b[5]-a[5])+(b[6]-a[6]))/dt))
+        o["rollbacks"].append((rt, (b[1]-a[1])/dt))
+        o["deadlocks"].append((rt, (b[7]-a[7])/dt))
+        o["lockwait"].append((rt, b[9]))
+        o["active"].append((rt, b[8]))
+    return o
+
+R = {t: rates(t) for t, _, _ in RUNS}
+
+def line(title, fname, ylabel, key):
+    plt.figure(figsize=(10,5))
+    for tag, label, color in RUNS:
+        s = R[tag][key]
+        if s: plt.plot([p[0] for p in s], [p[1] for p in s], label=label, color=color, linewidth=1.9)
+    plt.title(title); plt.xlabel("seconds since measurement start"); plt.ylabel(ylabel)
+    plt.grid(True, alpha=0.3); plt.legend(); plt.tight_layout()
+    plt.savefig(f"{CMP}/{fname}", dpi=110); plt.close(); print("wrote", fname)
+
+line("Postgres read rate  (rows returned + fetched / s)", "g_reads.png", "rows read / s", "reads")
+line("Postgres write rate  (rows inserted + updated + deleted / s)", "g_writes.png", "rows written / s", "writes")
+
+# health: lock-waiting backends (left) + rollbacks/s (right), both modes
+fig, ax1 = plt.subplots(figsize=(10,5)); ax2 = ax1.twinx()
+for tag, label, color in RUNS:
+    lw, rb = R[tag]["lockwait"], R[tag]["rollbacks"]
+    if lw: ax1.plot([p[0] for p in lw], [p[1] for p in lw], color=color, ls="-", lw=1.9, label=f"{label}: lock-waiters")
+    if rb: ax2.plot([p[0] for p in rb], [p[1] for p in rb], color=color, ls="--", lw=1.5, label=f"{label}: rollbacks/s")
+ax1.set_xlabel("seconds since measurement start"); ax1.set_ylabel("lock-waiting backends (solid)")
+ax2.set_ylabel("rollbacks / s (dashed)"); ax1.grid(True, alpha=0.3)
+l1, b1 = ax1.get_legend_handles_labels(); l2, b2 = ax2.get_legend_handles_labels()
+ax1.legend(l1+l2, b1+b2, loc="upper left", fontsize=8)
+plt.title("DB health / contention  (lock-waiting backends + rollback rate)")
+plt.tight_layout(); plt.savefig(f"{CMP}/g_health.png", dpi=110); plt.close(); print("wrote g_health.png")
+
+print("\n=== summary (mean / peak) ===")
+def st(s): v=[x[1] for x in s]; return (sum(v)/len(v), max(v)) if v else (0,0)
+for tag, label, _ in RUNS:
+    u=st(util_series(tag)); rd=st(R[tag]["reads"]); wr=st(R[tag]["writes"])
+    rb=st(R[tag]["rollbacks"]); lw=st(R[tag]["lockwait"]); dl=st(R[tag]["deadlocks"]); ac=st(R[tag]["active"])
+    print(f"{label}: util {u[0]:.0f}/{u[1]:.0f}% | reads {rd[0]:.0f}/{rd[1]:.0f} | writes {wr[0]:.0f}/{wr[1]:.0f} "
+          f"| rollbacks/s {rb[0]:.1f}/{rb[1]:.1f} | lock-wait {lw[0]:.1f}/{lw[1]:.0f} | deadlk/s {dl[0]:.2f} | active {ac[0]:.1f}/{ac[1]:.0f}")