Compare sequential and parallel image processing

// CPU comparison: ordinary sequential loop, one worker, and multiple workers.
// Run: aner run examples/parallel/cpu_image_blur.aner --max-steps 500000000
// Standalone timings are one trial; run_cpu_cases.py repeats and varies execution order.
import aner.parallel
// Batch preprocessing: one independent grayscale camera image per task.
// The fixture is synthetic: a lighting gradient, a bright central object and
// reproducible sensor texture. Each image varies exposure; no file I/O is timed.
import aner.math

fn process_image(parameters: List, iteration: Int64, seed: Int64) -> List {
    let width = cast<Int64>(parameters.get(0))
    let height = cast<Int64>(parameters.get(1))
    let passes = cast<Int64>(parameters.get(2))
    let pixels = width * height
    var source = Array<Float64>(pixels, 0.0)
    var destination = Array<Float64>(pixels, 0.0)
    let exposure = 0.85 + Float64(iteration % 15) * 0.02

    // Both buffers start with the same boundary pixels. The blur updates only
    // interior pixels, so every pass preserves the original one-pixel border.
    var y = 0
    while y < height {
        var x = 0
        while x < width {
            let texture = Float64((x * 17 + y * 13 + iteration * 3) % 23) - 11.0
            var intensity = 20.0 + 140.0 * Float64(x) / Float64(width - 1) + texture
            if x >= width / 4 && x < 3 * width / 4 &&
               y >= height / 4 && y < 3 * height / 4 {
                intensity = intensity + 65.0
            }
            intensity = intensity * exposure
            if intensity < 0.0 { intensity = 0.0 }
            if intensity > 255.0 { intensity = 255.0 }
            let index = y * width + x
            source.set(index, intensity)
            destination.set(index, intensity)
            x = x + 1
        }
        y = y + 1
    }

    var pass = 0
    while pass < passes {
        y = 1
        while y < height - 1 {
            var x = 1
            while x < width - 1 {
                let index = y * width + x
                let neighborhood = source.get(index - width - 1) + source.get(index - width) +
                    source.get(index - width + 1) + source.get(index - 1) + source.get(index) +
                    source.get(index + 1) + source.get(index + width - 1) +
                    source.get(index + width) + source.get(index + width + 1)
                destination.set(index, neighborhood / 9.0)
                x = x + 1
            }
            y = y + 1
        }
        // Read the old image throughout one pass; swap after the full pass.
        let previous = source
        source = destination
        destination = previous
        pass = pass + 1
    }

    var pixel_sum = 0.0
    var square_sum = 0.0
    var index = 0
    while index < pixels {
        let value = source.get(index)
        pixel_sum = pixel_sum + value
        square_sum = square_sum + value * value
        index = index + 1
    }
    // Deterministic output columns: mean gray level, RMS gray level, center pixel.
    return List(pixel_sum / Float64(pixels), math.sqrt(square_sum / Float64(pixels)),
        source.get((height / 2) * width + width / 2))
}

// Change only these settings to try a larger batch or another CPU allocation.
let jobs = 64
let columns = 3
// 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(64, 64, 3)

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 = process_image(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: process_image,
        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 = process_image(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, 0) <= 255.0 &&
        concurrent.value(iteration, 1) >= concurrent.value(iteration, 0) &&
        concurrent.value(iteration, 2) >= 0.0 && concurrent.value(iteration, 2) <= 255.0, "Invalid grayscale metrics")
}
print("case=cpu_image_blur")
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())