Parallel Prefix / Sorting Project

Project Brief and API Contract

The Parallel Prefix / Sorting Project implements two classic data-parallel algorithms — the Blelloch exclusive prefix scan and the bitonic sorting network — and evaluates them in the work–span model. Building on the parallel prefix scan and parallel sorting foundations, the project treats every operation as a directed acyclic graph of primitive steps: work counts the total steps executed with one processor, while span counts the steps on the longest dependency chain with unlimited processors. Speedup, efficiency, and Amdahl-style limits all follow from these two numbers.

The public interface exposes two pure functions and one telemetry contract: parallel_scan(values) returns the exclusive prefix sums of an arbitrary-length array, and parallel_sort(values) returns ascending order. Both pad internally to the next power of two, replay every dependency level as a counted round, and report exact work and span counters alongside the result. Acceptance requires 100% parity with a sequential reference oracle on random and adversarial inputs, span bounds of $O(\log N)$ for the scan and $O(\log^2 N)$ for the sort, and a measured speedup bound of $\text{work} / \text{span}$ in line with Brent's theorem.

Parallel Engine Contract: the engines accept arbitrary integer arrays, return oracle-identical outputs, and guarantee work $O(N)$ for the scan and $O(N \log^2 N)$ for the sort with spans $O(\log N)$ and $O(\log^2 N)$ respectively. Success is measured in counted operations and dependency rounds — not wall-clock time — and any oracle mismatch is a hard failure.

Architecture and Data Flow

The system decomposes into a dispatch layer, two independent parallel engines, and a shared cost-accounting and verification pipeline. Data flows through the dispatcher into exactly one engine, and every primitive step performed by that engine is charged to the telemetry collector before the oracle renders a verdict.

The architecture comprises five interacting components:

Task Dispatcher & Validation: selects the engine for parallel_scan or parallel_sort, validates the input, and pads the array to the next power of two.
Blelloch Scan Engine: builds a reduce tree in the up-sweep, then propagates prefix seeds down the same tree in the down-sweep.
Bitonic Sort Engine: runs a fixed compare-exchange network of $\tfrac{1}{2}\log_2 N (\log_2 N + 1)$ stages, each of $N/2$ independent pair operations.
Work & Span Telemetry Collector: charges one work unit per primitive operation and one span unit per dependency level, yielding the critical-path length.
Reference Oracle & Test Harness: replays each request sequentially with an obviously correct loop and compares element-for-element before results are released.

Parallel Prefix / Sorting Project data flow

flowchart TD
    A[Client Scan / Sort Request] --> B[Dispatch & Validation]
    B --> C{Operation Type}
    C --> Exclusive Prefix Scan| D[Blelloch Scan Engine]
    C -->|Parallel Ordering| E[Bitonic Sort Engine]
    D --> F[Upsweep: Pairwise Reduce Tree]
    F --> G[Downsweep: Propagate Identity Seeds]
    E --> H[Compare-Exchange Network Stages]
    G --> I[Work and Span Telemetry Collector]
    H --> I
    I --> J[Compare against Sequential Oracle]
    J -->|Match| K[Return Results & Speedup Report]
    J -->|Mismatch| L[Raise Parity Verification Exception]

Core Implementations in Python, C++, and Java

The complete implementations below demonstrate the Blelloch scan and the bitonic network with explicit work and span counters. Each language panel is a complete, copy-paste-runnable program: it runs the sample array, asserts oracle parity, prints the counted costs, and finishes with a 200-trial randomized sweep over adversarial shapes (single elements, non-power-of-two lengths, duplicates, and negative values).

import random


def oracle_scan(values):
    """Sequential reference: obviously correct exclusive scan."""
    out, acc = [], 0
    for v in values:
        out.append(acc)
        acc += v
    return out


class Telemetry:
    def __init__(self):
        self.work = 0
        self.span = 0


class ParallelScan:
    """Blelloch exclusive scan with work and span counters."""

    def __init__(self, telemetry):
        self.t = telemetry

    def scan(self, values):
        n = len(values)
        if n == 0:
            return []
        size = 1
        while size < n:
            size *= 2
        a = list(values) + [0] * (size - n)

        d = 1                                    # UP-SWEEP: reduce tree
        while d < size:
            for i in range(0, size, 2 * d):
                a[i + 2 * d - 1] += a[i + d - 1]
                self.t.work += 1
            self.t.span += 1
            d *= 2

        a[size - 1] = 0                          # identity seed
        d = size // 2                            # DOWN-SWEEP: scan tree
        while d >= 1:
            for i in range(0, size, 2 * d):
                left = a[i + d - 1]
                prefix = a[i + 2 * d - 1]
                a[i + d - 1] = prefix
                a[i + 2 * d - 1] = prefix + left
                self.t.work += 1
            self.t.span += 1
            d //= 2

        return a[\:n]


class ParallelSort:
    """Bitonic sorting network: O(N log^2 N) work, O(log^2 N) span."""

    def __init__(self, telemetry):
        self.t = telemetry

    def sort(self, values):
        n = len(values)
        if n < 2:
            return list(values)
        size = 1
        while size < n:
            size *= 2
        a = list(values) + [float("inf")] * (size - n)

        k = 2
        while k <= size:
            j = k // 2
            while j >= 1:
                for i in range(size):
                    partner = i ^ j
                    if partner > i:
                        ascending = (i & k) == 0
                        if (ascending and a[partner] < a[i]) or (
                            not ascending and a[partner] > a[i]
                        ):
                            a[i], a[partner] = a[partner], a[i]
                        self.t.work += 1
                self.t.span += 1
                j //= 2
            k *= 2
        return a[\:n]


if __name__ == "__main__":
    sample = [3, 1, 4, 1, 5, 9, 2, 6]

    t_scan = Telemetry()
    scan_result = ParallelScan(t_scan).scan(sample)
    assert scan_result == oracle_scan(sample), "Scan output mismatch!"

    t_sort = Telemetry()
    sort_result = ParallelSort(t_sort).sort(sample)
    assert sort_result == sorted(sample), "Sort output mismatch!"

    print("Python Parallel Engine: all oracle checks passed.")
    print("Scan:", scan_result)
    print("Sort:", sort_result)
    print("Scan work:", t_scan.work, "span:", t_scan.span,
          "speedup:", t_scan.work // t_scan.span)
    print("Sort work:", t_sort.work, "span:", t_sort.span,
          "speedup:", t_sort.work // t_sort.span)

    rng = random.Random(7)
    for _ in range(200):
        data = [rng.randint(-50, 50) for _ in range(rng.randint(1, 64))]
        assert ParallelScan(Telemetry()).scan(data) == oracle_scan(data)
        assert ParallelSort(Telemetry()).sort(data) == sorted(data)
    print("Randomized oracle sweep: 200 trials passed.")

#include <iostream>
#include <vector>
#include <algorithm>
#include <cassert>
#include <limits>
#include <random>

struct Telemetry {
    long long work = 0;
    long long span = 0;
};

std::vector<long long> oracle_scan(const std::vector<long long>& values) {
    std::vector<long long> out;
    out.reserve(values.size());
    long long acc = 0;
    for (long long v : values) {
        out.push_back(acc);
        acc += v;
    }
    return out;
}

class ParallelScan {
public:
    explicit ParallelScan(Telemetry& telemetry) : t(telemetry) {}

    std::vector<long long> scan(const std::vector<long long>& values) {
        size_t n = values.size();
        if (n == 0) return {};
        size_t size = 1;
        while (size < n) size *= 2;
        std::vector<long long> a(values);
        a.resize(size, 0);

        for (size_t d = 1; d < size; d *= 2) {          // UP-SWEEP
            for (size_t i = 0; i < size; i += 2 * d) {
                a[i + 2 * d - 1] += a[i + d - 1];
                t.work++;
            }
            t.span++;
        }

        a[size - 1] = 0;                                 // identity seed
        for (size_t d = size / 2; d >= 1; d /= 2) {      // DOWN-SWEEP
            for (size_t i = 0; i < size; i += 2 * d) {
                long long left = a[i + d - 1];
                long long prefix = a[i + 2 * d - 1];
                a[i + d - 1] = prefix;
                a[i + 2 * d - 1] = prefix + left;
                t.work++;
            }
            t.span++;
        }
        a.resize(n);
        return a;
    }

private:
    Telemetry& t;
};

class ParallelSort {
public:
    explicit ParallelSort(Telemetry& telemetry) : t(telemetry) {}

    std::vector<long long> sort(std::vector<long long> a) const {
        size_t n = a.size();
        if (n < 2) return a;
        size_t size = 1;
        while (size < n) size *= 2;
        a.resize(size, std::numeric_limits<long long>::max());

        for (size_t k = 2; k <= size; k *= 2) {
            for (size_t j = k / 2; j >= 1; j /= 2) {
                for (size_t i = 0; i < size; i++) {
                    size_t partner = i ^ j;
                    if (partner > i) {
                        bool ascending = (i & k) == 0;
                        if ((ascending && a[partner] < a[i]) ||
                            (!ascending && a[partner] > a[i])) {
                            std::swap(a[i], a[partner]);
                        }
                        t.work++;
                    }
                }
                t.span++;
            }
        }
        a.resize(n);
        return a;
    }

private:
    Telemetry& t;
};

int main() {
    std::vector<long long> sample = {3, 1, 4, 1, 5, 9, 2, 6};

    Telemetry tScan;
    std::vector<long long> scanResult = ParallelScan(tScan).scan(sample);
    assert(scanResult == oracle_scan(sample));

    Telemetry tSort;
    std::vector<long long> sortResult = ParallelSort(tSort).sort(sample);
    std::vector<long long> expected = sample;
    std::sort(expected.begin(), expected.end());
    assert(sortResult == expected);

    std::cout << "C++ Parallel Engine: all oracle checks passed.\n";
    std::cout << "Scan work: " << tScan.work << " span: " << tScan.span
              << " speedup: " << tScan.work / tScan.span << "\n";
    std::cout << "Sort work: " << tSort.work << " span: " << tSort.span
              << " speedup: " << tSort.work / tSort.span << "\n";

    std::mt19937 rng(7);
    std::uniform_int_distribution<int> value(-50, 50), length(1, 64);
    for (int trial = 0; trial < 200; trial++) {
        std::vector<long long> data(length(rng));
        for (auto& v : data) v = value(rng);

        Telemetry ts;
        assert(ParallelScan(ts).scan(data) == oracle_scan(data));

        std::vector<long long> sortedData = data;
        std::sort(sortedData.begin(), sortedData.end());
        Telemetry tp;
        assert(ParallelSort(tp).sort(data) == sortedData);
    }
    std::cout << "Randomized oracle sweep: 200 trials passed.\n";
    return 0;
}

import java.util.*;

public class Main {

    public static class Telemetry {
        public long work = 0;
        public long span = 0;
    }

    public static long[] oracleScan(long[] values) {
        long[] out = new long[values.length];
        long acc = 0;
        for (int i = 0; i < values.length; i++) {
            out[i] = acc;
            acc += values[i];
        }
        return out;
    }

    public static long[] parallelScan(long[] values, Telemetry t) {
        int n = values.length;
        if (n == 0) return new long[0];
        int size = 1;
        while (size < n) size *= 2;
        long[] a = Arrays.copyOf(values, size);

        for (int d = 1; d < size; d *= 2) {            // UP-SWEEP
            for (int i = 0; i < size; i += 2 * d) {
                a[i + 2 * d - 1] += a[i + d - 1];
                t.work++;
            }
            t.span++;
        }

        a[size - 1] = 0;                               // identity seed
        for (int d = size / 2; d >= 1; d /= 2) {       // DOWN-SWEEP
            for (int i = 0; i < size; i += 2 * d) {
                long left = a[i + d - 1];
                long prefix = a[i + 2 * d - 1];
                a[i + d - 1] = prefix;
                a[i + 2 * d - 1] = prefix + left;
                t.work++;
            }
            t.span++;
        }
        return Arrays.copyOf(a, n);
    }

    public static long[] parallelSort(long[] values, Telemetry t) {
        int n = values.length;
        if (n < 2) return values.clone();
        int size = 1;
        while (size < n) size *= 2;
        long[] a = Arrays.copyOf(values, size);
        Arrays.fill(a, n, size, Long.MAX_VALUE);

        for (int k = 2; k <= size; k *= 2) {
            for (int j = k / 2; j >= 1; j /= 2) {
                for (int i = 0; i < size; i++) {
                    int partner = i ^ j;
                    if (partner > i) {
                        boolean ascending = (i & k) == 0;
                        if ((ascending && a[partner] < a[i])
 || (!ascending && a[partner] > a[i])) {
                            long tmp = a[i];
                            a[i] = a[partner];
                            a[partner] = tmp;
                        }
                        t.work++;
                    }
                }
                t.span++;
            }
        }
        return Arrays.copyOf(a, n);
    }

    public static void main(String[] args) {
        long[] sample = {3, 1, 4, 1, 5, 9, 2, 6};

        Telemetry tScan = new Telemetry();
        long[] scanResult = parallelScan(sample, tScan);
        if (!Arrays.equals(scanResult, oracleScan(sample))) {
            throw new AssertionError("Scan output mismatch!");
        }

        Telemetry tSort = new Telemetry();
        long[] sortResult = parallelSort(sample, tSort);
        long[] expected = sample.clone();
        Arrays.sort(expected);
        if (!Arrays.equals(sortResult, expected)) {
            throw new AssertionError("Sort output mismatch!");
        }

        System.out.println("Java Parallel Engine: all oracle checks passed.");
        System.out.println("Scan work: " + tScan.work + " span: " + tScan.span
                + " speedup: " + (tScan.work / tScan.span));
        System.out.println("Sort work: " + tSort.work + " span: " + tSort.span
                + " speedup: " + (tSort.work / tSort.span));

        Random rng = new Random(7);
        for (int trial = 0; trial < 200; trial++) {
            int len = 1 + rng.nextInt(64);
            long[] data = new long[len];
            for (int i = 0; i < len; i++) {
                data[i] = rng.nextInt(101) - 50;
            }
            long[] sortedData = data.clone();
            Arrays.sort(sortedData);
            if (!Arrays.equals(parallelScan(data, new Telemetry()), oracleScan(data))
 || !Arrays.equals(parallelSort(data, new Telemetry()), sortedData)) {
                throw new AssertionError("Randomized sweep failed at trial " + trial);
            }
        }
        System.out.println("Randomized oracle sweep: 200 trials passed.");
    }
}

Visual Walkthrough of the Engine

To visualize the engine, consider an exclusive scan over the sample array $[3, 1, 4, 1, 5, 9, 2, 6]$ with $N = 8$. The first diagram traces one end-to-end run: the up-sweep folds the array into a reduce tree, and the down-sweep reuses that same tree to scatter prefix seeds back to the leaves, with the internal state highlighted at every level.

The second view contrasts the two cost axes that determine everything above: total work, the runtime with a single processor, against span, the dependency chain that caps speedup no matter how many processors are available.

Correctness Testing Against a Reference Oracle

Verification replays every request against a deliberately slow, obviously correct sequential oracle: a plain accumulator loop for the scan and the standard library sort for ordering. The harness sweeps two hundred randomized trials per run over adversarial shapes — single elements, non-power-of-two lengths, duplicates, negatives, and already-sorted or reverse-sorted inputs — and treats any element-wise divergence as a hard failure, since parallel routines must be deterministic regardless of how operations are scheduled.

Testing Lab

Find the Bug

Scenario: A student's Blelloch scan passes the trivial $N = 1$ test but fails oracle parity on every trial with $N \geq 2$, with outputs that look uniformly "too large."

Diagnostic Trace:

Oracle Input: sample array $[3, 1, 4, 1, 5, 9, 2, 6]$; the oracle expects the exclusive scan $[0, 3, 4, 8, 9, 14, 23, 25]$.
Behavior Observed: the engine returns $[31, 34, 35, 39, 40, 45, 54, 56]$ — sorted-looking, plausible, and wrong.
Audit Inspection: subtracting the oracle element-wise yields a constant offset of $31$ on every position — exactly the array's total sum. Inspection of the engine shows the identity seed assignment a[size - 1] = 0 between the up-sweep and down-sweep loops was deleted during a refactor.
Failure Point: with the root slot still holding the reduce-tree total instead of the identity, every prefix propagated down the scan tree inherits the full sum $31$, so each output is the true exclusive prefix plus the grand total.

Remediation: reinsert the identity seed before the down-sweep, and add a standing assertion that scan_output[N-1] + input[N-1] equals the up-sweep root total — a cheap invariant that catches any future seed or propagation regression.

Benchmarks and Complexity Report

The engine replays each workload at power-of-two sizes ($N = 2^{10}$ to $2^{16}$) and compares the counted work and span against theory. Because every primitive operation and dependency level is counted exactly, the measured columns match the closed-form bounds to the operation, and the ratio $\text{work} / \text{span}$ bounds the best achievable speedup by Brent's theorem before a single real thread is spawned.

Operation / Workload	Input Size ($N$)	Theoretical Work	Theoretical Span	Measured Work	Measured Span	Work / Span Speedup
Blelloch Exclusive Scan	$1{,}024$	$2N - 2 = 2{,}046$	$2\lceil \log_2 N \rceil = 20$	$2{,}046$	$20$	$\approx 102\times$
Blelloch Exclusive Scan	$65{,}536$	$131{,}070$	$32$	$131{,}070$	$32$	$\approx 4{,}096\times$
Bitonic Sort	$1{,}024$	$\frac{N}{2} \cdot \frac{\log_2 N (\log_2 N + 1)}{2} = 28{,}160$	$\frac{\log_2 N (\log_2 N + 1)}{2} = 55$	$28{,}160$	$55$	$\approx 512\times$
Bitonic Sort	$65{,}536$	$4{,}456{,}448$	$136$	$4{,}456{,}448$	$136$	$\approx 32{,}768\times$
Sequential Baseline Scan	$65{,}536$	$N - 1 = 65{,}535$	$N - 1 = 65{,}535$	$65{,}535$	$65{,}535$	$1\times$

Extensions and Stretch Goals

To evolve the simulated PRAM engine toward production parallel systems, the following milestones trade counted abstractions for real hardware, real schedulers, and harder workloads.

Extension Roadmap & Engineering Targets:

Tier 1 (Real Thread Pools): map each dependency level onto std::async / ForkJoinPool / concurrent.futures with Brent-style round scheduling, and reconcile measured wall-clock speedup against the work/span bound.
Tier 2 (GPU Kernels): port the bitonic network and Blelloch scan to CUDA or OpenCL with coalesced memory access and shared-memory tree buffers, profiling occupancy against the $O(\log N)$ critical path.
Tier 3 (Work-Efficient Sorting): replace bitonic sort with a parallel radix sort composed from the scan primitive, restoring $O(N)$-like work while keeping $O(\log N)$-depth passes and deterministic, reproducible outputs for floating-point reductions.

Cookie Consent