exp05_authors_code.py
#!/usr/bin/env python3
"""Experiment 0.1c: the authors' own simulation, run headless.
Runs the cell classes from Zhang, Goldstein, Levin (2024) unmodified (vendored
under vendor/cell_research), with their thread-per-cell scheduling, their
frozen-cell mechanism and their per-swap snapshots, and scores the snapshots
with their delayed-gratification function copied verbatim, plus our two
metrics. Nothing of ours touches the dynamics. A check that our
reimplementation, not the authors' claim, is what produced the 0.1 numbers.
Usage: exp05_authors_code.py [--algo bubble] [--exps 30] [--n 100] [--frozen 0,1,2,3]
exp05_authors_code.py --selftest
"""
import argparse
import csv
import json
import os
import random
import sys
import threading
import time
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(HERE, "vendor", "cell_research"))
from modules.multithread.BubbleSortCell import BubbleSortCell # noqa: E402
from modules.multithread.CellGroup import CellGroup, GroupStatus # noqa: E402
from modules.multithread.InsertionSortCell import InsertionSortCell # noqa: E402
from modules.multithread.MultiThreadCell import CellStatus # noqa: E402
from modules.multithread.SelectionSortCell import SelectionSortCell # noqa: E402
from modules.multithread.StatusProbe import StatusProbe # noqa: E402
CELLS = {"bubble": BubbleSortCell, "insertion": InsertionSortCell, "selection": SelectionSortCell}
# --- the authors' analysis, verbatim from analysis/delay_gratification_analysis_for_not_move.py ---
def get_monotonicity(arr):
monotonicity_value = 0
prev = arr[0]
for i in range(1, len(arr)):
if arr[i] < prev:
monotonicity_value += 1
prev = arr[i]
return monotonicity_value
def dedup(arr):
res = []
for x in arr:
if len(res) == 0 or res[-1] != x:
res.append(x)
return res
def get_discrepency_arr(arr):
arr = dedup(arr)
temp = []
for i in range(1, len(arr)):
temp.append(arr[i] - arr[i - 1])
res = []
if len(temp) == 0:
return [0]
x = temp[0]
for i in range(1, len(temp)):
if temp[i - 1] * temp[i] > 0:
x += temp[i]
if temp[i - 1] * temp[i] <= 0:
res.append(x)
x = temp[i]
res.append(x)
return res
def get_first_pos(arr):
for i in range(len(arr)):
if arr[i] > 0:
return i
return len(arr)
def avg_wandering_range(arr):
d_arr = get_discrepency_arr(arr)
first_pos_index = get_first_pos(d_arr)
res = 0
n = 0
i = first_pos_index
if i >= len(d_arr) - 1:
return 0
while i < len(d_arr):
j = i + 1
if not (j < len(d_arr)) or d_arr[j] > 0:
return res
change = -(d_arr[j] + d_arr[i]) / d_arr[i]
res += change
n += 1
i += 2
if n == 0:
return 0
return res / n
def get_max_delay_gratification(steps):
monotonicity_changes = [get_monotonicity(step) for step in steps if len(step) > 0]
if len(monotonicity_changes) == 0:
return 0
return avg_wandering_range(monotonicity_changes)
# --- our metrics on their snapshots ---
def inversions(arr):
return sum(1 for i in range(len(arr)) for j in range(i + 1, len(arr)) if arr[i] > arr[j])
def dg_running_max(series):
total, best = 0.0, series[0]
for prev, cur in zip(series, series[1:]):
if cur > prev and prev < best:
total += cur - prev
best = max(best, cur)
return total
# --- the authors' driver, minus tkinter and numpy ---
def create_cells(value_list, lock, probe, cell_type, frozen):
left, right = (0, 1), (len(value_list) - 1, 1)
cells = []
for i in range(len(value_list)):
cells.append(CELLS[cell_type](i + 1, value_list[i], lock, (i, 1), cells, left, right, probe, disable_visualization=True))
group = CellGroup(cells, cells, 0, left, right, GroupStatus.ACTIVE, lock, 1000000000, 1000000000)
for c in cells:
c.group = group
for _ in range(frozen):
cells[random.randint(0, len(cells) - 1)].set_cell_to_freeze()
return cells, [group]
def no_cells_should_move(cells):
for c in cells:
if c.status == CellStatus.SLEEP:
return False
if c.status == CellStatus.ACTIVE and c.should_move():
return False
return True
def run_one(cell_type, n, frozen, seed, timeout=600):
random.seed(seed)
values = list(range(n))
random.shuffle(values)
lock = threading.Lock()
probe = StatusProbe()
cells, groups = create_cells(values, lock, probe, cell_type, frozen)
lock.acquire()
for c in cells:
if c.status != CellStatus.FREEZE:
c.start()
for g in groups:
g.start()
lock.release()
t0 = time.time()
while not no_cells_should_move(cells) and time.time() - t0 < timeout:
time.sleep(0.05)
timed_out = time.time() - t0 >= timeout
lock.acquire()
for c in cells:
c.status = CellStatus.INACTIVE
for g in groups:
g.status = GroupStatus.MERGED
lock.release()
for c in cells:
if c.is_alive():
c.join(5)
snaps = [[values], *[[s] for s in probe.sorting_steps]]
steps = [values] + list(probe.sorting_steps)
final = [c.value for c in cells]
n1 = n - 1
adj = [1 - get_monotonicity(s) / n1 for s in steps]
ken = [1 - inversions(s) / (n * n1 / 2) for s in steps]
return {
"seed": seed,
"swaps": probe.swap_count,
"final_adj_inv": get_monotonicity(final),
"final_inv": inversions(final),
"dg_paper": get_max_delay_gratification(steps),
"dg_adj": dg_running_max(adj),
"dg_kendall": dg_running_max(ken),
"timed_out": timed_out,
"seconds": round(time.time() - t0, 1),
}
def mean_ci(xs, rng, reps=1000):
m = sum(xs) / len(xs)
boots = sorted(sum(rng.choice(xs) for _ in xs) / len(xs) for _ in range(reps))
return m, boots[int(0.025 * reps)], boots[int(0.975 * reps)]
def selftest():
assert inversions([2, 0, 1]) == 2 and inversions(list(range(5))) == 0
# gains count while the start of the step is below the running max
assert dg_running_max([4, 2, 3, 5]) == 3
assert dg_running_max([1, 2, 3]) == 0
# one small run of the authors' bubble cells: must sort, and adjacent
# swaps never raise total inversions, so Kendall DG is 0 (P2)
r = run_one("bubble", 12, 0, seed=7, timeout=60)
assert not r["timed_out"], r
assert r["final_inv"] == 0 and r["dg_kendall"] == 0, r
print(f"selftest ok: bubble n=12 swaps={r['swaps']} {r['seconds']}s")
def main():
if "--selftest" in sys.argv[1:]:
return selftest()
ap = argparse.ArgumentParser()
ap.add_argument("--algo", default="bubble")
ap.add_argument("--exps", type=int, default=30)
ap.add_argument("--n", type=int, default=100)
ap.add_argument("--frozen", default="0,1,2,3")
ap.add_argument("--out", default=os.path.join(HERE, "results", "exp05"))
a = ap.parse_args()
os.makedirs(a.out, exist_ok=True)
summary = []
for nf in [int(x) for x in a.frozen.split(",")]:
rows = []
for e in range(a.exps):
r = run_one(a.algo, a.n, nf, seed=1000 * e + 7)
rows.append(r)
print(f"{a.algo} nf={nf} exp={e} swaps={r['swaps']} inv={r['final_inv']} "
f"dg_paper={r['dg_paper']:.3f} dg_adj={r['dg_adj']:.3f} dg_ken={r['dg_kendall']:.3f} "
f"{r['seconds']}s{' TIMEOUT' if r['timed_out'] else ''}", file=sys.stderr, flush=True)
with open(os.path.join(a.out, f"{a.algo}_nf{nf}_runs.json"), "w") as f:
json.dump(rows, f)
rng = random.Random(1)
row = {"algo": a.algo, "n": a.n, "n_frozen": nf, "exps": a.exps}
for key in ("swaps", "final_inv", "dg_paper", "dg_adj", "dg_kendall"):
row[key], row[key + "_lo"], row[key + "_hi"] = mean_ci([r[key] for r in rows], rng)
row["timeouts"] = sum(r["timed_out"] for r in rows)
summary.append(row)
print(f"SUMMARY {a.algo} nf={nf} dg_paper={row['dg_paper']:.3f} [{row['dg_paper_lo']:.3f},{row['dg_paper_hi']:.3f}] "
f"dg_ken={row['dg_kendall']:.4f} inv={row['final_inv']:.1f} timeouts={row['timeouts']}", file=sys.stderr, flush=True)
path = os.path.join(a.out, f"{a.algo}_summary.csv")
with open(path, "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=list(summary[0].keys()))
w.writeheader()
w.writerows(summary)
if __name__ == "__main__":
main()