Parallel Algorithm Analysis

Why Parallel Algorithm Analysis Matters

Adding processors does not automatically divide runtime by their count. Some tasks depend on earlier results, some become ready together, and a processor may sit idle while work elsewhere remains unfinished. Parallel algorithm analysis separates the total computation from the dependency chain that processors cannot shorten.

The four central ideas are work, span, speedup, and Amdahl's law. Work counts all task time. Span measures the longest dependency path. Speedup compares an execution with a stated one-worker or sequential baseline. Amdahl models the limit imposed by a fixed serial fraction when the remaining portion scales ideally.

This lesson represents a computation as a weighted directed acyclic graph, computes its work and span, then simulates identical workers with an explicit deterministic greedy policy. Python, C++, and Java print the same completion batches, ready tasks, assignments, runtime, and exact rational metrics. A separate Amdahl calculation shows a fixed serial/parallel model alongside the DAG experiment.

The purpose is prediction and diagnosis: identify whether more processors could help, whether dependency structure is the bottleneck, and whether a proposed parallelization adds too much work. The same reasoning supports reductions, build pipelines, parallel loops, and task schedulers, provided the graph captures the required dependencies and the duration model is appropriate.

The simulator is sequential software describing an ideal schedule. It creates no threads and measures no hardware execution times. That separation lets you verify the reasoning before adding runtime overhead, communication, contention, or a specific parallel programming system.

Core Concepts and Terminology

A computation DAG is $G=(V,E)$, with $N=\lvert V\rvert$ tasks and $M=\lvert E\rvert$ precedence edges. Task $v$ has positive integer duration $d_v$. An edge $u\to v$ means v may start only after u finishes. Tasks are nonpreemptive: once started, a task runs to completion on one identical worker. The model has no communication delay, scheduling cost, failures, or resource contention.

Work is the sum of durations: $$ W=T_1=\sum_{v\in V}d_v. $$ Here $T_1$ is the time to execute this same DAG on one ideal worker. If the fastest sequential algorithm is different, its runtime is a different baseline; state which one a speedup uses.

Span is the longest weighted directed path: $$ S=T_\infty=\max_{\pi\text{ a directed path}}\sum_{v\in\pi}d_v. $$ With unlimited workers, every task can start as soon as its predecessors finish, so span is the ideal minimum completion time. It includes both source and final task durations. For an empty graph, define $W=S=0$ and completion time zero.

An earliest-start/finish dynamic program computes span in topological order: $$ \operatorname{ES}(v)=\max_{u\to v}\operatorname{EF}(u),\qquad \operatorname{EF}(v)=\operatorname{ES}(v)+d_v. $$ The maximum over no predecessors is zero. Then $S=\max_v\operatorname{EF}(v)$. Store an attaining predecessor to recover one critical path; multiple longest paths may exist. Earliest start assumes unlimited processors and can be earlier than actual start in a limited-worker schedule.

For $P\ge1$ workers and a legal makespan $T_P$, the work and span laws give $$ T_P\ge\max(W/P,S). $$ Integer task times allow the stronger integral lower bound $\max(\lceil W/P\rceil,S)$. Neither bound is generally an achievable runtime: task indivisibility and scheduling constraints can leave a gap.

Average available parallelism is $\mathcal P=W/S$ for a nonempty graph. It is a ratio, not an instantaneous count of ready tasks or a promise that exactly that many workers remain busy. Work-based speedup and efficiency are $$ \operatorname{Speedup}(P)=\frac W{T_P},\qquad \operatorname{Efficiency}(P)=\frac W{PT_P}. $$ Therefore speedup is at most $\min(P,W/S)$ in this model. These definitions and the scheduling bounds are treated in MIT's Races and Parallelism lecture. Empty-graph ratios such as $0/0$ are undefined; the program avoids printing them.

Amdahl's fixed-workload model. Let $\sigma\in[0,1]$ be the fraction of a one-processor execution that stays serial. If the remaining work divides perfectly among P workers with no added overhead, then $$ T_P^{\mathrm{Amdahl}}=T_1\left(\sigma+\frac{1-\sigma}{P}\right),\qquad \operatorname{Speedup}_{\mathrm{Amdahl}}(P)=\frac1{\sigma+(1-\sigma)/P}. $$ For $\sigma\gt0$, the limiting speedup as P grows is $1/\sigma$. For $\sigma=0$, the ideal formula gives P; for $\sigma=1$, it gives one. MIT's Amdahl discussion explains this fixed serial/parallel decomposition. It is a model of a particular workload decomposition, not a universal replacement for a DAG's scheduling analysis.

Do not substitute $S/W$ for $\sigma$ without justifying that decomposition. Span can include a different branch at different joins and includes one task from a parallel region; it is not automatically the sum of globally serial phases. The sample prints a separate $\sigma=1/4$ Amdahl family precisely so its assumptions remain explicit.

How Parallel Algorithm Analysis Works, Step by Step

1. Validate and canonicalize the DAG. Check positive duration bounds, edge endpoints, no self edges, and no duplicate directed edges. Sort a copy of the edges. Construct successor lists and indegrees without changing the inputs.
2. Find a deterministic topological order. Put zero-indegree tasks into a minimum-ID priority queue. Repeatedly pop the smallest ID, update successor earliest-start candidates, and release a successor when its indegree becomes zero. If fewer than N tasks are processed, reject a cycle rather than simulating a deadlock.
3. Compute work and span. Sum all durations. For each task in topological order, set finish to earliest start plus duration; propagate that finish to its successors using a maximum. For equal best predecessor finishes, choose the smaller predecessor ID. Choose the smallest-ID maximum-finish endpoint, then follow parent links backward and reverse them to recover one critical path.
4. Initialize the worker simulation separately. Restore original indegrees; the topological analysis already consumed a working copy. Ready roots enter a fresh minimum-ID queue. All worker slots begin free. Actual starts/finishes/worker IDs start at $-1$.
5. At an event time, finish the entire completion batch. Gather all running tasks finishing now, free their workers, sort completed IDs, and decrement the corresponding successor counts. Release a task only when all its predecessor completions have been processed. All simultaneous completions happen before dispatch at that time.
6. Dispatch deterministically and greedily. Scan free workers by increasing worker ID. Assign each the smallest ready task ID and record its start and finish. Continue until all workers are busy or the ready queue is empty. Dispatching a task does not release its successors: completion does.
7. Record and advance. Save the completed batch, ready queue before dispatch, started tasks, waiting queue after dispatch, and running tasks in worker order. If every task is completed, return now. Otherwise jump directly to the smallest running finish time and repeat. Positive durations guarantee strictly later completion events.

The program's priority queue is a ready-task policy, not a critical-path priority. Task IDs provide reproducibility but do not imply priority optimality. The scheduler is greedy in the work-conserving sense: it leaves no worker idle while a ready task waits. It can still make a suboptimal choice about which ready task to start.

The step diagram focuses on the sample's join at time nine. E and F complete together. G has two predecessor counts before the batch, one after E's completion, then zero after F's. G is queued once and starts on worker zero; worker one remains idle because no other task is ready. The final completion at time ten is recorded as a separate event.

flowchart TD
    A["Validate durations and edges; build graph"] --> B["Topological work/span analysis"]
    B --> C{"Cycle detected?"}
    C -->|Yes| D["Reject graph"]
    C -->|No| E["Initialize ready tasks and free workers"]
    E --> F{"Graph empty?"}
    F -->|Yes| L["Return makespan zero and empty events"]
    F -->|No| G["Complete all tasks finishing now; release successors"]
    G --> H["Assign smallest ready IDs to free workers in order"]
    H --> I["Record completion, ready, start and running state"]
    I --> J{"All tasks completed?"}
    J -->|Yes| K["Return schedule and metrics"]
    J -->|No| M["Jump to earliest running finish time"]
    M --> G

Worked Example with a Full Trace

Use task IDs zero through six, called A through G in the diagrams. Durations and all predecessor sets are:

Full unlimited-worker analysis. Kahn's minimum-ID order is $[0,1,2,3,4,5,6]$. The table gives every earliest-start/finish state and the chosen critical predecessor. Work accumulates by adding each row's duration. The parent selection uses earliest finish, not actual limited-worker finish.

Thus $W=17$, $S=10$, and the recovered path is $[0,1,4,6]$, or A–B–E–G. F could finish at time seven with unlimited workers, but that does not force the two-worker execution to finish F at seven.

Every event with two workers. Assign the smallest ready ID to the smallest free worker. Ready means ready immediately before dispatch; waiting means ready after dispatch. A task shown as running has been started but is still an unfinished predecessor. Worker labels are zero and one.

At time two, B, C, D all become ready, but only two workers are available. B and C start, leaving D waiting. C finishes at five, so D takes worker one from five to seven. B finishes at six and releases E because C has already completed. E runs on worker zero from six to nine. At seven, D's completion releases F, which runs on worker one until nine. G waits for both E and F, then runs from nine to ten.

All actual assignments, alongside their unlimited-worker earliest starts, are:

The schedule has $T_2=10$. Its integer lower bound is $\max(\lceil17/2\rceil,10)=10$, so this particular schedule is optimal. Its speedup is $17/10=1.7$, efficiency $17/20=0.85$, and idle worker-time is $2\cdot10-17=3$. Worker one is idle for two units at the start and one at the end; neither interval has a ready waiting task.

With one worker, the same DAG takes seventeen units. With two, three, four, or eight workers, the program reports ten: more capacity cannot shorten the A–B–E–G chain. This equality is a property of this sample; a different graph or priority can show different results.

A separate Amdahl comparison. For $\sigma=1/4$, ideal speedup is $4P/(P+3)$. The printed P values give:

The Amdahl values approach four under their ideal serial/parallel assumptions. They are not a fit to this DAG's span or a prediction of its schedule. All ratios are printed as reduced exact fractions so floating-point formatting cannot obscure cross-language agreement.

Implementation in Python, C++, and Java

Each panel is a complete program with the same weighted DAG, analysis tie rules, event-driven scheduler, Amdahl helper, and output. Run Python normally, compile C++ as C++17, or save the Java panel as Main.java and run Main. The programs simulate workers; they do not spawn actual parallel tasks.

Graph contract. Call run(durations,edges,processors). Durations are a flat integer sequence of length $0\le N\le64$, with each $1\le d_v\le1000$. Each directed edge is an ordered pair of task IDs in $[0,N)$. At most $4096$ edges may be supplied; self edges and duplicates are rejected. Require $1\le P\le64$. Node IDs need not already be topological. Inputs remain unchanged because sorting and indegree updates use copies.

Validation order is processors (PROCESSORS), excessive node count (SIZE), invalid durations (DURATION), excessive/malformed/out-of-range/self/duplicate edges (EDGE), then a genuine directed cycle (CYCLE). Python requires exact integer durations, endpoints, and processor count, rejecting booleans/floats/strings. Its edge rows may be lists or tuples of length two. C++ uses vector<int> durations and vector<pair<int,int>> edges; Java uses int[] and two-element int[][] rows. Non-null outer containers are expected; Java explicitly rejects a null or wrong-length edge row as EDGE.

The $64\cdot1000$ work bound keeps event times and counters small. Empty graphs return zero work/span/makespan, empty assignment arrays, and no events. The program's sample metrics operate on a nonempty graph; speedup, efficiency, and parallelism ratios are undefined for the empty graph and must not be formed as $0/0$.

Analysis output. analyze returns successor lists, original indegrees, canonical topological order, earliest starts/finishes, chosen parents, one critical path, work, and span. For equal longest predecessor finishes, the smaller predecessor ID wins. For equal span endpoints, the smallest ID wins. These choices affect which equally valid critical path is displayed, not its length.

Simulation output. Every event contains its time, sorted completed IDs, sorted ready IDs before dispatch, IDs started in worker-assignment order, waiting IDs after dispatch, and running IDs in worker order. Per-node actual starts, finishes, and assigned workers are also returned. The notation u@w->f in standard output means task u is assigned to worker w and finishes at time f. All intervals are half open, so a worker completing a task at time t can begin another at the same t.

Exact Amdahl helper. amdahl(s,d,P) represents serial fraction $s/d$ and returns reduced speedup $$ \frac{dP}{sP+d-s}. $$ Require integers $1\le d\le10^6$, $0\le s\le d$, and $1\le P\le64$; violations give AMDAHL. C++/Java use 64-bit intermediate products, while Python uses arbitrary-precision integers. The helper treats fractions zero and one correctly and is independent of the graph simulator. The generic fraction formatter is internal and expects a positive denominator in all program uses.

from dataclasses import dataclass
from heapq import heapify, heappop, heappush
from math import gcd


@dataclass
class Analysis:
    durations: list
    successors: list
    indegrees: list
    topo: list
    earliest: list
    finish: list
    parent: list
    critical: list
    work: int
    span: int


@dataclass
class Event:
    time: int
    completed: list
    ready: list
    started: list
    waiting: list
    running: list


@dataclass
class Result:
    analysis: Analysis
    starts: list
    finishes: list
    workers: list
    events: list
    makespan: int


def analyze(durations, edges):
    n = len(durations)
    if n > 64:
        raise ValueError("SIZE")
    if any(type(d) is not int or not 1 <= d <= 1000 for d in durations):
        raise ValueError("DURATION")
    if len(edges) > 4096:
        raise ValueError("EDGE")
    for e in edges:
        if (not isinstance(e, (list, tuple)) or len(e) != 2
                or any(type(v) is not int or not 0 <= v < n for v in e)
                or e[0] == e[1]):
            raise ValueError("EDGE")
    ordered = sorted((u, v) for u, v in edges)
    if any(ordered[i] == ordered[i - 1] for i in range(1, len(ordered))):
        raise ValueError("EDGE")
    successors = [[] for _ in durations]
    indegrees = [0] * n
    for u, v in ordered:
        successors[u].append(v)
        indegrees[v] += 1
    remaining = indegrees.copy()
    ready = [u for u in range(n) if remaining[u] == 0]
    heapify(ready)
    topo, earliest, finish, parent = [], [0] * n, [0] * n, [-1] * n
    while ready:
        u = heappop(ready)
        topo.append(u)
        finish[u] = earliest[u] + durations[u]
        for v in successors[u]:
            if (finish[u] > earliest[v] or
                    (finish[u] == earliest[v] and (parent[v] == -1 or u < parent[v]))):
                earliest[v], parent[v] = finish[u], u
            remaining[v] -= 1
            if remaining[v] == 0:
                heappush(ready, v)
    if len(topo) != n:
        raise ValueError("CYCLE")
    span = max(finish, default=0)
    critical = []
    if n:
        u = next(i for i in range(n) if finish[i] == span)
        while u != -1:
            critical.append(u)
            u = parent[u]
        critical.reverse()
    return Analysis(durations.copy(), successors, indegrees, topo, earliest,
                    finish, parent, critical, sum(durations), span)


def run(durations, edges, processors):
    if type(processors) is not int or not 1 <= processors <= 64:
        raise ValueError("PROCESSORS")
    a = analyze(durations, edges)
    n = len(durations)
    starts, finishes, workers = [-1] * n, [-1] * n, [-1] * n
    if n == 0:
        return Result(a, starts, finishes, workers, [], 0)
    remaining = a.indegrees.copy()
    ready = [u for u in range(n) if remaining[u] == 0]
    heapify(ready)
    running = [-1] * processors
    events = []
    now = done = 0
    while True:
        completed = []
        for w, u in enumerate(running):
            if u != -1 and finishes[u] == now:
                completed.append(u)
                running[w] = -1
        completed.sort()
        for u in completed:
            done += 1
            for v in a.successors[u]:
                remaining[v] -= 1
                if remaining[v] == 0:
                    heappush(ready, v)
        before, started = sorted(ready), []
        for w in range(processors):
            if running[w] == -1 and ready:
                u = heappop(ready)
                starts[u], finishes[u], workers[u] = now, now + durations[u], w
                running[w] = u
                started.append(u)
        active = [u for u in running if u != -1]
        events.append(Event(now, completed, before, started, sorted(ready), active))
        if done == n:
            break
        assert active
        now = min(finishes[u] for u in active)
    return Result(a, starts, finishes, workers, events, now)


def fraction(numerator, denominator):
    g = gcd(numerator, denominator)
    return f"{numerator // g}/{denominator // g}"


def amdahl(serial_numerator, denominator, processors):
    if (any(type(v) is not int for v in (serial_numerator, denominator, processors))
            or not 1 <= denominator <= 1000000 or not 0 <= serial_numerator <= denominator
            or not 1 <= processors <= 64):
        raise ValueError("AMDAHL")
    return fraction(denominator * processors,
                    serial_numerator * processors + denominator - serial_numerator)


def compact(values):
    return '[' + ','.join(map(str, values)) + ']'


def assignments(ids, result):
    return '[' + ','.join(f"{u}@{result.workers[u]}->{result.finishes[u]}" for u in ids) + ']'


def main():
    durations = [2, 4, 3, 2, 3, 2, 1]
    edges = [(0, 1), (0, 2), (0, 3), (1, 4), (2, 4), (2, 5), (3, 5), (4, 6), (5, 6)]
    r = run(durations, edges, 2)
    a = r.analysis
    print("durations=" + compact(durations) + " processors=2")
    print(f"work={a.work} span={a.span} parallelism={fraction(a.work, a.span)}")
    print("topo=" + compact(a.topo) + " critical=" + compact(a.critical))
    print("earliest=" + compact(a.earliest) + " earliest_finish=" + compact(a.finish))
    for e in r.events:
        print(f"t={e.time} completed={compact(e.completed)} ready={compact(e.ready)} "
              f"start={assignments(e.started, r)} waiting={compact(e.waiting)} running={assignments(e.running, r)}")
    lower = max((a.work + 1) // 2, a.span)
    print(f"makespan={r.makespan} lower_bound={lower} greedy_upper={fraction(a.work + a.span, 2)} "
          f"speedup={fraction(a.work, r.makespan)} efficiency={fraction(a.work, 2 * r.makespan)}")
    for p in (1, 2, 3, 4, 8):
        s = run(durations, edges, p)
        print(f"p={p} dag_time={s.makespan} dag_speedup={fraction(a.work, s.makespan)} "
              f"amdahl_serial=1/4 amdahl_speedup={amdahl(1, 4, p)}")


if __name__ == "__main__":
    main()

#include <algorithm>
#include <cassert>
#include <functional>
#include <iostream>
#include <numeric>
#include <queue>
#include <stdexcept>
#include <string>
#include <utility>
#include <vector>
using namespace std;
using Edge=pair<int,int>;
struct Analysis {
    vector<int> durations;
    vector<vector<int>> successors;
    vector<int> indegrees,topo,earliest,finish,parent,critical;
    int work=0,span=0;
};
struct Event { int time;vector<int> completed,ready,started,waiting,running; };
struct Result { Analysis analysis;vector<int> starts,finishes,workers;vector<Event> events;int makespan=0; };
using Ready=priority_queue<int,vector<int>,greater<int>>;
vector<int> sorted_ready(Ready ready) {
    vector<int> out;while(!ready.empty()) { out.push_back(ready.top());ready.pop(); }return out;
}
Analysis analyze(const vector<int>& durations,const vector<Edge>& edges) {
    if(durations.size()>64)throw invalid_argument("SIZE");
    for(int d:durations)if(d<1 || d>1000)throw invalid_argument("DURATION");
    if(edges.size()>4096)throw invalid_argument("EDGE");
    int n=static_cast<int>(durations.size());
    for(auto [u,v]:edges)if(u<0 || u>=n || v<0 || v>=n || u==v)throw invalid_argument("EDGE");
    vector<Edge> ordered=edges;sort(ordered.begin(),ordered.end());
    if(adjacent_find(ordered.begin(),ordered.end())!=ordered.end())throw invalid_argument("EDGE");
    Analysis a;a.durations=durations;a.successors.resize(n);a.indegrees.assign(n,0);
    a.earliest.assign(n,0);a.finish.assign(n,0);a.parent.assign(n,-1);
    for(auto [u,v]:ordered) { a.successors[u].push_back(v);++a.indegrees[v]; }
    vector<int> remaining=a.indegrees;Ready ready;
    for(int u=0;u<n;++u)if(remaining[u]==0)ready.push(u);
    while(!ready.empty()) {
        int u=ready.top();ready.pop();a.topo.push_back(u);a.finish[u]=a.earliest[u]+durations[u];
        for(int v:a.successors[u]) {
            if(a.finish[u]>a.earliest[v] || (a.finish[u]==a.earliest[v] && (a.parent[v]==-1 || u<a.parent[v]))) {
                a.earliest[v]=a.finish[u];a.parent[v]=u;
            }
            if(--remaining[v]==0)ready.push(v);
        }
    }
    if(static_cast<int>(a.topo.size())!=n)throw invalid_argument("CYCLE");
    a.work=accumulate(durations.begin(),durations.end(),0);
    if(n) {
        int u=static_cast<int>(max_element(a.finish.begin(),a.finish.end())-a.finish.begin());a.span=a.finish[u];
        while(u!=-1) { a.critical.push_back(u);u=a.parent[u]; }
        reverse(a.critical.begin(),a.critical.end());
    }
    return a;
}
Result run(const vector<int>& durations,const vector<Edge>& edges,int processors) {
    if(processors<1 || processors>64)throw invalid_argument("PROCESSORS");
    Result r;r.analysis=analyze(durations,edges);const Analysis& a=r.analysis;
    int n=static_cast<int>(durations.size());
    r.starts.assign(n,-1);r.finishes.assign(n,-1);r.workers.assign(n,-1);
    if(n==0)return r;
    vector<int> remaining=a.indegrees,running(processors,-1);Ready ready;
    for(int u=0;u<n;++u)if(remaining[u]==0)ready.push(u);
    int now=0,done=0;
    while(true) {
        vector<int> completed;
        for(int w=0;w<processors;++w) {
            int u=running[w];
            if(u!=-1 && r.finishes[u]==now) { completed.push_back(u);running[w]=-1; }
        }
        sort(completed.begin(),completed.end());
        for(int u:completed) { ++done;for(int v:a.successors[u])if(--remaining[v]==0)ready.push(v); }
        vector<int> before=sorted_ready(ready),started;
        for(int w=0;w<processors;++w)if(running[w]==-1 && !ready.empty()) {
            int u=ready.top();ready.pop();r.starts[u]=now;r.finishes[u]=now+durations[u];r.workers[u]=w;
            running[w]=u;started.push_back(u);
        }
        vector<int> active;for(int u:running)if(u!=-1)active.push_back(u);
        r.events.push_back(Event{now,completed,before,started,sorted_ready(ready),active});
        if(done==n)break;
        assert(!active.empty());
        now=r.finishes[active.front()];for(int u:active)now=min(now,r.finishes[u]);
    }
    r.makespan=now;return r;
}
string fraction(long long numerator,long long denominator) {
    long long g=gcd(numerator,denominator);return to_string(numerator/g)+"/"+to_string(denominator/g);
}
string amdahl(int serial_numerator,int denominator,int processors) {
    if(denominator<1 || denominator>1000000 || serial_numerator<0 || serial_numerator>denominator || processors<1 || processors>64)
        throw invalid_argument("AMDAHL");
    return fraction(1LL*denominator*processors,1LL*serial_numerator*processors+denominator-serial_numerator);
}
string compact(const vector<int>& values) {
    string s="[";for(size_t i=0;i<values.size();++i) { if(i)s+=',';s+=to_string(values[i]); }return s+"]";
}
string assignments(const vector<int>& ids,const Result& r) {
    string s="[";for(size_t i=0;i<ids.size();++i) {
        if(i)s+=',';
        int u=ids[i];s+=to_string(u)+"@"+to_string(r.workers[u])+"->"+to_string(r.finishes[u]);
    }return s+"]";
}
int main() {
    vector<int> durations{2,4,3,2,3,2,1};
    vector<Edge> edges{ {0,1},{0,2},{0,3},{1,4},{2,4},{2,5},{3,5},{4,6},{5,6} };
    Result r=run(durations,edges,2);const Analysis& a=r.analysis;
    cout<<"durations="<<compact(durations)<<" processors=2\nwork="<<a.work<<" span="<<a.span<<" parallelism="<<fraction(a.work,a.span)
        <<"\ntopo="<<compact(a.topo)<<" critical="<<compact(a.critical)<<"\nearliest="<<compact(a.earliest)<<" earliest_finish="<<compact(a.finish)<<'\n';
    for(const Event&e:r.events)cout<<"t="<<e.time<<" completed="<<compact(e.completed)<<" ready="<<compact(e.ready)
        <<" start="<<assignments(e.started,r)<<" waiting="<<compact(e.waiting)<<" running="<<assignments(e.running,r)<<'\n';
    cout<<"makespan="<<r.makespan<<" lower_bound="<<max((a.work+1)/2,a.span)<<" greedy_upper="<<fraction(a.work+a.span,2)
        <<" speedup="<<fraction(a.work,r.makespan)<<" efficiency="<<fraction(a.work,2*r.makespan)<<'\n';
    for(int p:vector<int>{1,2,3,4,8}) {
        Result s=run(durations,edges,p);
        cout<<"p="<<p<<" dag_time="<<s.makespan<<" dag_speedup="<<fraction(a.work,s.makespan)
            <<" amdahl_serial=1/4 amdahl_speedup="<<amdahl(1,4,p)<<'\n';
    }
    return 0;
}

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.PriorityQueue;

public class Main {
    static final class Analysis {
        final int[] durations,indegrees,earliest,finish,parent;
        final List<List<Integer>> successors=new ArrayList<>();
        final List<Integer> topo=new ArrayList<>(),critical=new ArrayList<>();
        int work,span;
        Analysis(int[] d) {
            durations=d.clone();int n=d.length;indegrees=new int[n];earliest=new int[n];finish=new int[n];parent=new int[n];Arrays.fill(parent,-1);
            for(int i=0;i<n;++i)successors.add(new ArrayList<>());
        }
    }
    static final class Event {
        final int time;
        final List<Integer> completed,ready,started,waiting,running;
        Event(int t,List<Integer> c,List<Integer> r,List<Integer> s,List<Integer> w,List<Integer> a) {
            time=t;completed=c;ready=r;started=s;waiting=w;running=a;
        }
    }
    static final class Result {
        final Analysis analysis;
        final int[] starts,finishes,workers;
        final List<Event> events=new ArrayList<>();
        int makespan;
        Result(Analysis a) {
            analysis=a;int n=a.durations.length;starts=new int[n];finishes=new int[n];workers=new int[n];
            Arrays.fill(starts,-1);Arrays.fill(finishes,-1);Arrays.fill(workers,-1);
        }
    }
    static List<Integer> sortedReady(PriorityQueue<Integer> ready) {
        List<Integer> out=new ArrayList<>(ready);out.sort(Integer::compare);return out;
    }
    static Analysis analyze(int[] durations,int[][] edges) {
        int n=durations.length;if(n>64)throw new IllegalArgumentException("SIZE");
        for(int d:durations)if(d<1 || d>1000)throw new IllegalArgumentException("DURATION");
        if(edges.length>4096)throw new IllegalArgumentException("EDGE");
        int[][] ordered=new int[edges.length][];
        for(int i=0;i<edges.length;++i) {
            int[] e=edges[i];
            if(e==null || e.length!=2 || e[0]<0 || e[0]>=n || e[1]<0 || e[1]>=n || e[0]==e[1])throw new IllegalArgumentException("EDGE");
            ordered[i]=e.clone();
        }
        Arrays.sort(ordered,Comparator.<int[]>comparingInt(e->e[0]).thenComparingInt(e->e[1]));
        for(int i=1;i<ordered.length;++i)if(Arrays.equals(ordered[i-1],ordered[i]))throw new IllegalArgumentException("EDGE");
        Analysis a=new Analysis(durations);
        for(int[] e:ordered) { a.successors.get(e[0]).add(e[1]);++a.indegrees[e[1]]; }
        int[] remaining=a.indegrees.clone();PriorityQueue<Integer> ready=new PriorityQueue<>();
        for(int u=0;u<n;++u)if(remaining[u]==0)ready.add(u);
        while(!ready.isEmpty()) {
            int u=ready.remove();a.topo.add(u);a.finish[u]=a.earliest[u]+durations[u];
            for(int v:a.successors.get(u)) {
                if(a.finish[u]>a.earliest[v] || (a.finish[u]==a.earliest[v] && (a.parent[v]==-1 || u<a.parent[v]))) {
                    a.earliest[v]=a.finish[u];a.parent[v]=u;
                }
                if(--remaining[v]==0)ready.add(v);
            }
        }
        if(a.topo.size()!=n)throw new IllegalArgumentException("CYCLE");
        for(int d:durations)a.work+=d;
        if(n!=0) {
            int last=0;for(int i=1;i<n;++i)if(a.finish[i]>a.finish[last])last=i;
            a.span=a.finish[last];
            for(int u=last;u!=-1;u=a.parent[u])a.critical.add(u);
            java.util.Collections.reverse(a.critical);
        }
        return a;
    }
    static Result run(int[] durations,int[][] edges,int processors) {
        if(processors<1 || processors>64)throw new IllegalArgumentException("PROCESSORS");
        Result r=new Result(analyze(durations,edges));Analysis a=r.analysis;int n=durations.length;if(n==0)return r;
        int[] remaining=a.indegrees.clone(),running=new int[processors];Arrays.fill(running,-1);
        PriorityQueue<Integer> ready=new PriorityQueue<>();for(int u=0;u<n;++u)if(remaining[u]==0)ready.add(u);
        int now=0,done=0;
        while(true) {
            List<Integer> completed=new ArrayList<>();
            for(int w=0;w<processors;++w) {
                int u=running[w];if(u!=-1 && r.finishes[u]==now) { completed.add(u);running[w]=-1; }
            }
            completed.sort(Integer::compare);
            for(int u:completed) { ++done;for(int v:a.successors.get(u))if(--remaining[v]==0)ready.add(v); }
            List<Integer> before=sortedReady(ready),started=new ArrayList<>();
            for(int w=0;w<processors;++w)if(running[w]==-1 && !ready.isEmpty()) {
                int u=ready.remove();r.starts[u]=now;r.finishes[u]=now+durations[u];r.workers[u]=w;running[w]=u;started.add(u);
            }
            List<Integer> active=new ArrayList<>();for(int u:running)if(u!=-1)active.add(u);
            r.events.add(new Event(now,completed,before,started,sortedReady(ready),active));
            if(done==n)break;
            if(active.isEmpty())throw new AssertionError("progress");
            now=r.finishes[active.get(0)];for(int u:active)now=Math.min(now,r.finishes[u]);
        }
        r.makespan=now;return r;
    }
    static long gcd(long a,long b) { while(b!=0) { long t=a%b;a=b;b=t; }return a; }
    static String fraction(long n,long d) { long g=gcd(n,d);return (n/g)+"/"+(d/g); }
    static String amdahl(int serialNumerator,int denominator,int processors) {
        if(denominator<1 || denominator>1000000 || serialNumerator<0 || serialNumerator>denominator || processors<1 || processors>64)
            throw new IllegalArgumentException("AMDAHL");
        return fraction(1L*denominator*processors,1L*serialNumerator*processors+denominator-serialNumerator);
    }
    static String compact(int[] a) { return Arrays.toString(a).replace(" ",""); }
    static String compact(List<Integer> a) { return a.toString().replace(" ",""); }
    static String assignments(List<Integer> ids,Result r) {
        StringBuilder s=new StringBuilder("[");for(int u:ids) { if(s.length()>1)s.append(',');s.append(u).append('@').append(r.workers[u]).append("->").append(r.finishes[u]); }return s.append(']').toString();
    }
    public static void main(String[]args) {
        int[] durations={2,4,3,2,3,2,1};
        int[][] edges={ {0,1},{0,2},{0,3},{1,4},{2,4},{2,5},{3,5},{4,6},{5,6} };
        Result r=run(durations,edges,2);Analysis a=r.analysis;
        System.out.println("durations="+compact(durations)+" processors=2");
        System.out.println("work="+a.work+" span="+a.span+" parallelism="+fraction(a.work,a.span));
        System.out.println("topo="+compact(a.topo)+" critical="+compact(a.critical));
        System.out.println("earliest="+compact(a.earliest)+" earliest_finish="+compact(a.finish));
        for(Event e:r.events)System.out.println("t="+e.time+" completed="+compact(e.completed)+" ready="+compact(e.ready)
            +" start="+assignments(e.started,r)+" waiting="+compact(e.waiting)+" running="+assignments(e.running,r));
        System.out.println("makespan="+r.makespan+" lower_bound="+Math.max((a.work+1)/2,a.span)+" greedy_upper="+fraction(a.work+a.span,2)
            +" speedup="+fraction(a.work,r.makespan)+" efficiency="+fraction(a.work,2*r.makespan));
        for(int p:new int[]{1,2,3,4,8}) {
            Result s=run(durations,edges,p);
            System.out.println("p="+p+" dag_time="+s.makespan+" dag_speedup="+fraction(a.work,s.makespan)+" amdahl_serial=1/4 amdahl_speedup="+amdahl(1,4,p));
        }
    }
}

Expected standard output, identical in all three languages:

• durations=[2,4,3,2,3,2,1] processors=2
• work=17 span=10 parallelism=17/10
• topo=[0,1,2,3,4,5,6] critical=[0,1,4,6]
• earliest=[0,2,2,2,6,5,9] earliest_finish=[2,6,5,4,9,7,10]
• t=0 completed=[] ready=[0] start=[0@0->2] waiting=[] running=[0@0->2]
• t=2 completed=[0] ready=[1,2,3] start=[1@0->6,2@1->5] waiting=[3] running=[1@0->6,2@1->5]
• t=5 completed=[2] ready=[3] start=[3@1->7] waiting=[] running=[1@0->6,3@1->7]
• t=6 completed=[1] ready=[4] start=[4@0->9] waiting=[] running=[4@0->9,3@1->7]
• t=7 completed=[3] ready=[5] start=[5@1->9] waiting=[] running=[4@0->9,5@1->9]
• t=9 completed=[4,5] ready=[6] start=[6@0->10] waiting=[] running=[6@0->10]
• t=10 completed=[6] ready=[] start=[] waiting=[] running=[]
• makespan=10 lower_bound=10 greedy_upper=27/2 speedup=17/10 efficiency=17/20
• p=1 dag_time=17 dag_speedup=1/1 amdahl_serial=1/4 amdahl_speedup=1/1
• p=2 dag_time=10 dag_speedup=17/10 amdahl_serial=1/4 amdahl_speedup=8/5
• p=3 dag_time=10 dag_speedup=17/10 amdahl_serial=1/4 amdahl_speedup=2/1
• p=4 dag_time=10 dag_speedup=17/10 amdahl_serial=1/4 amdahl_speedup=16/7
• p=8 dag_time=10 dag_speedup=17/10 amdahl_serial=1/4 amdahl_speedup=32/11

Complexity, Correctness, and Invariants

Laws and greedy guarantees. Any legal P-worker execution needs $T_P\ge W/P$ by processor-time capacity and $T_P\ge S$ by precedence. A work-conserving greedy list schedule on identical workers with zero overhead also satisfies the refined upper bound $$ T_P\le\frac WP+\left(1-\frac1P\right)S =\frac{W+(P-1)S}{P} \le\frac WP+S. $$ The sample prints $27/2$ for the refined two-worker upper bound, while its actual makespan is ten. The upper bound is a guarantee, not a runtime formula. The work/span scheduling treatment in CMU's parallel-computing notes explains the ideal greedy principle and why slackness matters.

One way to see the refined bound is to trace backward from the final finishing task, choosing the predecessor that completed last in the actual schedule. This produces a valid dependency chain of total duration $L\le S$. Before each chain task begins, any gap after its chosen predecessor finished occurs while that task is ready. Greediness forces all P workers to be busy throughout those gaps. During the chain's own execution, at least one worker is busy. Thus $W\ge P(T_P-L)+L$, which rearranges to $T_P\le W/P+(1-1/P)L$ and hence the bound above. The chain used in this proof need not equal the critical path returned by the unlimited-worker analysis.

Combining upper and lower bounds gives a constant-factor guarantee; the refined form is at most $(2-1/P)$ times optimal for this model. It does not prove a particular priority order optimal. When $P\ll W/S$, the additive span term is small relative to $W/P$, so ideal greedy speedup can approach linear scaling. When $P\gg W/S$, span limits further benefit. Large parallelism is useful only if the algorithm also avoids excessive work compared with its sequential competitor.

Correctness of work/span analysis. The ready heap contains exactly the unprocessed tasks with no unprocessed predecessor. Removing a root of that remaining graph preserves topological validity. When a task is processed, every predecessor finish is final; the maximum yields the earliest feasible start. Induction along topological order proves every computed ES/EF and the maximum span. Parent links attain the selected maxima, so their recovered chain has total duration S. Summing each duration once proves work. Processing fewer than N vertices means a cycle remains and no valid topological execution exists.

Correctness of the simulation. Each task is in exactly one state: not started, running on one worker, or completed. A not-started task's remaining predecessor count equals the number of its predecessors not yet completed, including running ones. Only a transition to zero enqueues it, and it is removed exactly once at dispatch. A dispatched task occupies one worker until its recorded finish, so no worker intervals overlap and no task is started twice.

At an event, finish all tasks ending now before dispatch. Every new start therefore has all predecessor finishes at or before its start. The scheduler fills every free worker while the ready queue is nonempty. Positive durations make the next running finish strictly later; acyclicity guarantees progress until every task completes. These invariants prove a legal deterministic greedy schedule. They do not establish race freedom for a real program unless the supplied DAG captures all required data dependencies.

The analyzer's CPU cost differs from W. Let N count tasks and M edges. Sorting the copied edges for validation/canonicalization costs $O(M\log(M+1))$. Graph construction and the dynamic program scan all edges; the minimum-ID topological heap inserts/removes each node once, costing $O(N\log(N+1))$. The general work/span analysis can be linear with an ordinary queue, but its canonical minimum-ID order here costs a heap factor.

The event simulator has at most $N+1$ events because every event after time zero completes at least one task. It scans P worker slots and copies/sorts ready-state snapshots at each event. Each node is dispatched once and each edge is released once. Including analysis, a safe total host-runtime bound is $$ O\bigl(M\log(M+1)+N^2\log(N+1)+NP+N+M+P\bigr). $$ Snapshot storage can reach $O(N^2)$; graph/analysis/assignments use $O(N+M)$, and running-worker storage $O(P)$. Overall host space is $O(N^2+N+M+P)$. These are simulator implementation bounds. They are not the ideal task computation's $W$ work or S span, and software simulation time is not $T_P$.

Best/average/worst qualifications. There is no task-order probability distribution, so no universal average makespan is asserted. For a fixed graph, W and S are fixed; scheduling priority and processor count affect actual makespan. Chains have $T_P=W=S$ for all P. Independent unit tasks have $T_P=\lceil N/P\rceil$, span one, and work N. The host analyzer still reads the graph in both cases; simultaneous completions can reduce its number of recorded events. The Amdahl helper performs bounded integer arithmetic and Euclidean fraction reduction, not parallel execution.

Variants, Pitfalls, and Optimizations

Do not replace bounds with equalities. $\max(W/P,S)$ is a lower bound and $W/P+(1-1/P)S$ a greedy upper bound. Neither is generally the actual runtime. Even indivisible independent tasks of durations $[2,2,2]$ on two workers need four units, while the integral work/span lower bound is three. The graph's task granularity matters.

Greedy priority can change makespan. In the diagram's separate example $[2,2,3]$, smallest-ID scheduling takes five units; starting the three-unit task first permits four. Both dispatch whenever a task can run. A longest-ready-task or bottom-level priority can be useful to explore, but a heuristic needs its own evaluation and does not solve every precedence-constrained scheduling instance optimally.

Distinguish graph time from worker waiting. ES/EF describe unlimited workers. The sample's D could start at time two but actually starts at five; F's earliest start is five but actual start is seven. Replacing ES with actual start during the span dynamic program would make a supposed graph invariant depend on a particular P-worker schedule.

Use all predecessors, and complete before releasing. Counting a task as finished when dispatched can start a child too early. Requiring only one of several parents can also violate a join. Duplicate edges can corrupt indegree bookkeeping if the builder counts them differently from the releaser. This code rejects duplicates and processes all same-time completions before applying ready-ID priority.

Weight the graph in consistent units. Longest path by edge count is span only for an appropriate unit-task convention. Node count alone misses durations, communication, or weighted costs. If communication is essential, model it as explicit tasks/dependencies or a justified edge-delay extension. Do not mix microseconds, instruction counts, and bytes in one path sum.

Keep the baseline explicit. Work-based speedup compares this DAG's one-worker cost W with its P-worker cost. An algorithm with twice the sequential work may scale nicely yet still lose to a better serial algorithm. Work efficiency compares the parallel algorithm's W with the best relevant sequential complexity. Real measured speedup can exceed P because of changed memory behavior or baselines; that lies outside the fixed-cost model's work law.

Amdahl needs a fixed problem and serial fraction. The formula assumes the serial duration stays unchanged, the parallel remainder divides perfectly, and added overhead is zero. The fraction must come from the same one-processor baseline. Simply setting $\sigma=S/W$ generally double-counts a portion of the critical path as globally serial. Scaling the problem as P grows asks a different question; distinguish fixed-size strong scaling from scaled-workload experiments.

Granularity trades overhead against parallelism. Splitting a large task can expose more ready work and reduce indivisibility, but creates scheduling, allocation, and synchronization costs. Keeping tasks too coarse can hide available parallelism; making them too fine can inflate W. Represent essential spawn/join work or measure it when moving beyond the ideal DAG.

Real resources add constraints. Memory bandwidth, caches, NUMA placement, network transfers, locks, and worker heterogeneity can make durations depend on concurrent activity. A deterministic zero-overhead simulator cannot predict those interactions. Profiling actual code should separate task work, waiting, scheduler activity, and communication, then compare with the graph's structural lower bounds.

A valid graph is an assumption about the program. Missing a read/write dependency can produce a fast simulated schedule that is unsafe to execute. Adding an unnecessary dependency preserves safety but reduces modeled parallelism. Validate both directions of the graph construction: every required ordering is present, and artificial serialization has a reason. Nonpreemptive tasks here also have no internal race or resource model.

Handle empty graphs and ties deliberately. Empty W/S and speedup ratios are undefined, even though empty makespan is zero. Equal predecessor finishes can give several critical paths. IDs are only deterministic tie-breakers, so relabeling independent tasks can change the priority schedule while leaving work/span unchanged. Different topological orders are valid; compare canonical outputs only when the tie conventions match.

Practice Problems and Real-World Use

Use independent checks for the graph analysis and scheduling logic. A predecessor-based memoized DFS can compute longest weighted paths independently of the implementation's Kahn traversal, while a unit-time reference simulator can verify its event jumps. Compare the entire schedule, not just final runtime: every completion batch, ready set, dispatch, waiting set, running set, and assignment interval should agree.

Check that every task starts and completes exactly once, no worker's intervals overlap, every dependency satisfies finish[u] <= start[v], interval length equals duration, and both work/span lower bounds hold. With one worker, makespan must equal W; with at least N workers, it must equal S. Check the greedy upper bound too, including cases where it is loose or attained. Reordering the same input edge list must not alter the canonical result.

All three implementations passed $3575$ shared inputs per language: $2414$ graph workloads and $1161$ Amdahl cases. The graph suite has $2398$ valid cases and $16$ rejected cases; the Amdahl suite includes $8$ rejected cases. It covers exhaustive forward-edge DAGs through four vertices, randomized relabeled DAGs, chains, independent tasks, fan-in/fan-out, equal-time joins, empty graphs, maximum node/duration/processor limits, invalid edges/cycles, and the separate granularity/priority counterexamples. Independent DFS and integer-tick oracles match every analysis field, assignment, event, and makespan. Exact rational arithmetic checks the Amdahl helper, including serial fractions zero/one and large denominators. Additional checks cover $29$ Python type/edge-shape rejections and $75$ input-edge permutation comparisons. C++ ran with address and undefined-behavior sanitizers; all sample output matches byte for byte.

For algorithm design, start by writing a dependency decomposition and estimating work/span before tuning worker count. Then improve the expensive term: reduce redundant work, shorten a long dependency chain, or expose enough tasks to keep workers busy. Compare ideal improvements with measured scheduler and memory costs.

Applications include a build graph with compilation/link dependencies, a numerical reduction tree, a parallel prefix primitive, or a data-processing pipeline. Their real cost models differ, but each benefits from separating computation volume, mandatory order, and resource-limited execution.