Admit workers from the CPU inventory

// Uses Philox addressed sampling with explicit seed and draw identities.
// CPU comparison: ordinary sequential loop, one worker, and multiple workers.
// Run: aner run examples/parallel/cpu_inventory_simulation.aner --max-steps 500000000
// Standalone timings are one trial; run_cpu_cases.py repeats and varies execution order.
import aner.parallel
import aner.random

// Warehouse planning: replay independent demand years for one stocked SKU.
// Synthetic inputs represent days and integer stock units, not calibrated sales data.
// Parameters: years per task, days per year, supplier lead days, reorder point
// (units), order-up-to level (units), and explicit master demand seed.
fn simulate_inventory(parameters: List, iteration: Int64, seed: Int64) -> List {
    let years = cast<Int64>(parameters.get(0))
    let days_per_year = cast<Int64>(parameters.get(1))
    let lead_days = cast<Int64>(parameters.get(2))
    let reorder_point = cast<Int64>(parameters.get(3))
    let order_up_to = cast<Int64>(parameters.get(4))
    // The logical task identity controls demand. The injected executor seed is
    // intentionally unused, allowing a normal sequential loop to replay this task.
    let demand_seed = cast<Int64>(parameters.get(5)) + iteration
    var lost_units = 0
    var on_hand_unit_days = 0
    var demand_units = 0
    var order_count = 0
    var year = 0
    while year < years {
        var on_hand = order_up_to
        var on_order = 0
        let arrivals = Array<Int64>(lead_days, 0)
        var day = 0
        while day < days_per_year {
            // Orders placed after today's sales arrive at the start of day+lead.
            let slot = day % lead_days
            let arriving = arrivals.get(slot)
            arrivals.set(slot, 0)
            on_order = on_order - arriving
            on_hand = on_hand + arriving

            // A synthetic discrete demand histogram plus calendar effects.
            // Two distinct draw IDs per date also distinguish every replay year.
            let draw = 2 * (year * days_per_year + day)
            let u = random.uniform(seed: demand_seed, draw: draw)
            var demand = 12
            if u >= 0.15 { demand = 20 }
            if u >= 0.50 { demand = 30 }
            if u >= 0.80 { demand = 45 }
            if u >= 0.95 { demand = 70 }
            if day % 7 >= 5 { demand = demand + 8 }
            if day >= days_per_year - 35 { demand = demand + 10 }
            let promotion = random.uniform(seed: demand_seed, draw: draw + 1)
            if promotion < 0.04 { demand = demand + 40 }
            demand_units = demand_units + demand

            var served = demand
            if served > on_hand { served = on_hand }
            lost_units = lost_units + demand - served
            on_hand = on_hand - served
            on_hand_unit_days = on_hand_unit_days + on_hand

            // Inventory position includes every replenishment already in transit.
            // The ring slot is now empty; its new order is due lead_days later.
            let inventory_position = on_hand + on_order
            if inventory_position <= reorder_point {
                let quantity = order_up_to - inventory_position
                arrivals.set(slot, quantity)
                on_order = on_order + quantity
                order_count = order_count + 1
            }
            day = day + 1
        }
        year = year + 1
    }
    // Columns: mean lost-sales units/year, mean end-of-day stock units,
    // mean requested units/year, and mean replenishment orders/year.
    return List(Float64(lost_units) / Float64(years),
        Float64(on_hand_unit_days) / Float64(years * days_per_year),
        Float64(demand_units) / Float64(years),
        Float64(order_count) / Float64(years))
}

// Suggested harness inputs: 64 tasks and List(8, 365, 5, 180, 420, 6501).
// Keep lead_days > 0, years > 0, days_per_year >= 35, and
// 0 <= reorder_point < order_up_to when adapting the scenario.

// Change only these settings to try a larger batch or another CPU allocation.
let jobs = 64
let columns = 4
// 0 = automatic admitted CPU allocation; 1 = serial control.
let requested_workers = 0
// Runner rotates 0/1/2 to vary which method executes first.
let benchmark_order = 0
let parameters = List(32, 365, 5, 180, 420, 2026)

fn require(condition: Bool, message: String) -> Unit {
    if !condition {
        let failure: Result<Bool, String> = Result.err(message)
        print(failure.unwrap())
    }
}

// NO PARALLELISM: finish one job before calling the same function for the next.
fn measure_sequential(parameters: List, jobs: Int64, columns: Int64) -> List {
    let started = Duration.monotonic()
    let values = Array<Float64>(jobs * columns, 0.0)
    for iteration in Range(jobs) {
        let row = simulate_inventory(parameters, iteration, 2026)
        for column in Range(columns) {
            values[iteration * columns + column] = cast<Float64>(row.get(column))
        }
    }
    let seconds = Duration.monotonic().subtract(started).total_seconds()
    return List(values, seconds)
}

// PARALLELISM: the task body is unchanged. The executor collects ordered rows.
// workers=1 is the control; workers=0 selects the machine's admitted allocation.
fn measure_repeat(parameters: List, jobs: Int64, workers: Int64) -> List {
    let started = Duration.monotonic()
    let results = parallel.repeat(iterations: jobs, task: simulate_inventory,
        parameters: parameters, execution: parallel.local(workers: workers), seed: 2026)
    let seconds = Duration.monotonic().subtract(started).total_seconds()
    return List(results, seconds)
}

// One untimed task exercises the model before the comparison.
let warmup = simulate_inventory(parameters, 0, 2026)
var sequential_run = List()
var one_worker_run = List()
var parallel_run = List()
for phase in Range(3) {
    let mode = (benchmark_order + phase) % 3
    if mode == 0 { sequential_run = measure_sequential(parameters, jobs, columns) }
    if mode == 1 { one_worker_run = measure_repeat(parameters, jobs, 1) }
    if mode == 2 { parallel_run = measure_repeat(parameters, jobs, requested_workers) }
}
let sequential_values = cast<Array<Float64>>(sequential_run.get(0))
let one_worker = cast<SimulationResults>(one_worker_run.get(0))
let concurrent = cast<SimulationResults>(parallel_run.get(0))
let sequential_seconds = cast<Float64>(sequential_run.get(1))
let one_worker_seconds = cast<Float64>(one_worker_run.get(1))
let parallel_seconds = cast<Float64>(parallel_run.get(1))

// Verification and printing happen after all timers stop.
require(one_worker.rows() == jobs && concurrent.rows() == jobs &&
    one_worker.cols() == columns && concurrent.cols() == columns &&
    one_worker.workers() == 1, "Incomplete results or incorrect serial allocation")
require(sequential_seconds > 0.0 && one_worker_seconds > 0.0 && parallel_seconds > 0.0,
    "Clock resolution is insufficient for this batch")
for iteration in Range(jobs) {
    require(one_worker.seed(iteration) == concurrent.seed(iteration), "Task seed mismatch")
    for column in Range(columns) {
        let expected = sequential_values[iteration * columns + column]
        require(expected == one_worker.value(iteration, column) &&
            expected == concurrent.value(iteration, column), "Output differs across execution methods")
    }
    require(concurrent.value(iteration, 0) >= 0.0 && concurrent.value(iteration, 1) >= 0.0 &&
        concurrent.value(iteration, 2) > 0.0 && concurrent.value(iteration, 0) <= concurrent.value(iteration, 2) &&
        concurrent.value(iteration, 3) > 0.0, "Invalid inventory metrics")
}
print("case=cpu_inventory_simulation")
print("jobs=" + String(jobs))
print("columns=" + String(columns))
print("workers=" + String(concurrent.workers()))
print("verified=true")
// Caller elapsed times include allocation, CPU context startup and result collection.
print("sequential_seconds=" + String(sequential_seconds))
print("one_worker_seconds=" + String(one_worker_seconds))
print("parallel_seconds=" + String(parallel_seconds))
print("speedup=" + String(sequential_seconds / parallel_seconds))
print("executor_speedup=" + String(one_worker.seconds() / concurrent.seconds()))
for column in Range(columns) {
    print("output_" + String(column) + "_mean=" + String(concurrent.mean(column)))
}
print("summary_json=" + concurrent.summary())
print("one_worker_summary_json=" + one_worker.summary())