#!/usr/bin/env python3
"""Deterministic queue scheduling experiment. Python standard library only.

Run: python3 queue_simulation.py --out results
Times are simulated integer milliseconds, never wall-clock benchmarks.
No network requests, cloud resources, random numbers or third-party packages.
"""

import argparse
import csv
import heapq
import json
import math
from collections import Counter, deque
from pathlib import Path

WORKERS = 8
TENANTS = 20
POLICIES = ("fifo", "round_robin", "capped_round_robin")
SCENARIOS = {"equal_jobs": 100, "slow_bulk_jobs": 500}


def workload(bulk_duration):
    jobs = [dict(id=i, tenant=0, arrival_ms=0, service_ms=bulk_duration)
            for i in range(400)]
    for batch in range(4):
        for tenant in range(1, TENANTS):
            jobs.append(dict(id=len(jobs), tenant=tenant,
                             arrival_ms=200 + 400 * batch, service_ms=100))
    return sorted(jobs, key=lambda j: (j["arrival_ms"], j["id"]))


def percentile(values, p):
    """Nearest-rank quantile, with no interpolation."""
    return sorted(values)[math.ceil(p * len(values)) - 1]


def simulate(jobs, policy, workers=WORKERS):
    assert policy in POLICIES
    assert jobs and workers > 0
    assert len({j["id"] for j in jobs}) == len(jobs)
    assert all(j["service_ms"] > 0 and j["arrival_ms"] >= 0 for j in jobs)
    pending = sorted((dict(j) for j in jobs), key=lambda j: (j["arrival_ms"], j["id"]))
    queues = {t: deque() for t in sorted({j["tenant"] for j in jobs})}
    tenant_order = list(queues)
    global_queue = deque()
    active = Counter()
    running = []
    finished = []
    cursor = 0
    next_job = 0
    now = pending[0]["arrival_ms"]
    max_active = Counter()

    while len(finished) < len(jobs):
        # Complete all jobs first, admit all arrivals, then dispatch. This tie
        # order is fixed for all policies and both scenarios.
        while running and running[0][0] <= now:
            end, _, row = heapq.heappop(running)
            active[row["tenant"]] -= 1
            row["finish_ms"] = end
            finished.append(row)
        while next_job < len(pending) and pending[next_job]["arrival_ms"] <= now:
            row = pending[next_job]
            (global_queue if policy == "fifo" else queues[row["tenant"]]).append(row)
            next_job += 1

        while len(running) < workers:
            row = None
            if policy == "fifo":
                if global_queue:
                    row = global_queue.popleft()
            else:
                for _ in tenant_order:
                    tenant = tenant_order[cursor]
                    cursor = (cursor + 1) % len(tenant_order)
                    eligible = policy != "capped_round_robin" or active[tenant] < 2
                    if queues[tenant] and eligible:
                        row = queues[tenant].popleft()
                        break
            if row is None:
                break
            row["start_ms"] = now
            row["wait_ms"] = now - row["arrival_ms"]
            active[row["tenant"]] += 1
            max_active[row["tenant"]] = max(max_active[row["tenant"]], active[row["tenant"]])
            heapq.heappush(running, (now + row["service_ms"], row["id"], row))

        if len(finished) == len(jobs):
            break
        events = ([running[0][0]] if running else [])
        if next_job < len(pending):
            events.append(pending[next_job]["arrival_ms"])
        assert events, "Pending work cannot be stranded without an event."
        now = min(events)

    finished.sort(key=lambda j: j["id"])
    assert len(finished) == len(jobs) == len({j["id"] for j in finished})
    assert all(j["wait_ms"] >= 0 and j["finish_ms"] - j["start_ms"] == j["service_ms"] for j in finished)
    # Independent sweep verifies capacity and the hard per-tenant cap.
    events = sorted((time, change, j["tenant"]) for j in finished
                    for time, change in [(j["start_ms"], 1), (j["finish_ms"], -1)])
    in_flight = Counter()
    for _, change, tenant in events:
        in_flight[tenant] += change
        assert 0 <= sum(in_flight.values()) <= workers
        if policy == "capped_round_robin":
            assert in_flight[tenant] <= 2
    quiet = [j for j in finished if j["tenant"] != 0]
    bulk = [j for j in finished if j["tenant"] == 0]
    end = max(j["finish_ms"] for j in finished)
    summary = dict(
        policy=policy, jobs=len(finished), quiet_jobs=len(quiet), bulk_jobs=len(bulk),
        quiet_p50_wait_ms=percentile([j["wait_ms"] for j in quiet], .5),
        quiet_p95_wait_ms=percentile([j["wait_ms"] for j in quiet], .95),
        quiet_max_wait_ms=max(j["wait_ms"] for j in quiet),
        quiet_p95_completion_ms=percentile([j["finish_ms"]-j["arrival_ms"] for j in quiet], .95),
        bulk_finished_ms=max(j["finish_ms"] for j in bulk),
        all_finished_ms=end,
        worker_utilization_pct=round(100 * sum(j["service_ms"] for j in finished) / (workers * end), 2),
        max_bulk_in_flight=max_active[0],
    )
    return summary, finished


def self_check():
    # Independently calculable fixtures cover FIFO order, idle time, round-robin
    # fairness, cap enforcement and quantile definition.
    assert percentile(list(range(1, 77)), .95) == 73
    fixture = [dict(id=i, tenant=0 if i < 3 else 1, arrival_ms=0, service_ms=10)
               for i in range(4)]
    _, fifo = simulate(fixture, "fifo", workers=1)
    _, fair = simulate(fixture, "round_robin", workers=1)
    assert fifo[3]["start_ms"] == 30
    assert fair[3]["start_ms"] == 10
    idle = [dict(id=0, tenant=0, arrival_ms=0, service_ms=10),
            dict(id=1, tenant=1, arrival_ms=100, service_ms=10)]
    for policy in POLICIES:
        _, rows = simulate(idle, policy, workers=8)
        assert [r["wait_ms"] for r in rows] == [0, 0]


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--out", type=Path, default=Path("results"))
    args = parser.parse_args()
    self_check()
    args.out.mkdir(parents=True, exist_ok=True)
    result = dict(
        experiment="Dreamtsoft SaaS queue simulation", version=1,
        measurement_date="2026-09-05", time_unit="simulated milliseconds",
        assumptions=dict(workers=WORKERS, tenants=TENANTS, bulk_jobs=400,
            quiet_tenants=19, jobs_per_quiet_tenant=4, quiet_arrivals_ms=[200, 600, 1000, 1400],
            quiet_service_ms=100, bulk_service_ms=SCENARIOS, cap_per_tenant=2,
            quiet_percentile_denominator=76, p95_nearest_rank=73,
            ordering="finish, arrive, dispatch; arrival ties by ascending job ID; tenants in ascending ID order",
            preemption=False, randomness=False, exclusions=0),
        limits=["Synthetic workload, not production traffic or a vendor benchmark.",
                "No network, storage, locks, retries, failures, polling or scheduler overhead.",
                "All work is admitted; queues are unbounded in this model.",
                "Round robin is a reference algorithm, not an implementation of any managed queue.",
                "The hard cap does not borrow otherwise idle slots."],
        runs=[])
    traces = []
    for scenario, duration in SCENARIOS.items():
        for policy in POLICIES:
            summary, rows = simulate(workload(duration), policy)
            repeated, repeated_rows = simulate(workload(duration), policy)
            assert summary == repeated and rows == repeated_rows
            result["runs"].append(dict(scenario=scenario, **summary))
            traces.extend(dict(scenario=scenario, policy=policy, **j) for j in rows)
    (args.out / "results.json").write_text(json.dumps(result, indent=2) + "\n")
    with (args.out / "job-traces.csv").open("w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=list(traces[0]))
        writer.writeheader()
        writer.writerows(traces)
    print("scenario | policy | quiet p95 wait ms | bulk finish ms | utilization %")
    for r in result["runs"]:
        print(f'{r["scenario"]} | {r["policy"]} | {r["quiet_p95_wait_ms"]} | {r["bulk_finished_ms"]} | {r["worker_utilization_pct"]}')


if __name__ == "__main__":
    main()