Solve several streaming, DAG, and DP tasks

Problem Statement

Modern machine learning feature and inference pipelines process continuous data streams through directed acyclic graph (DAG) execution plans. You are tasked with implementing two core optimization routines for an execution engine in a class PipelineOptimizer.

Part 1: DAG Latency and Peak Parallelism

Given n tasks labeled 0 to n - 1, a list of directed edges where [u, v] indicates task u must complete before task v can begin, and an array durations where durations[i] is the execution time of task i:

• Tasks begin at the earliest possible non-negative time step once all predecessor tasks finish.
• Multiple independent tasks can run concurrently without hardware limits.
• Return [min_latency, max_parallelism], where min_latency is the total time to finish all tasks and max_parallelism is the maximum number of tasks running simultaneously at any point in time. If a task finishes at time t and another starts at time t, they are not considered simultaneously running at time t.

Part 2: Streaming Checkpoint Placement via Sliding-Window DP

The pipeline processes a streaming sequence of operations represented by an array costs of length m. To satisfy memory limits, in every contiguous sliding window of max_window consecutive operations, at least one checkpoint must be placed.

• Placing a checkpoint at operation i costs costs[i].
• A valid placement ensures that for every window of size max_window (i.e. costs[k : k + max_window] for all 0 <= k <= m - max_window), at least one chosen index falls within that window.
• Return the minimum total cost required to place checkpoints across the stream.

Interface

class PipelineOptimizer:
    def analyze_dag(
        self, n: int, edges: list[list[int]], durations: list[int]
    ) -> list[int]:
        pass

    def min_checkpoint_cost(self, costs: list[int], max_window: int) -> int:
        pass

Examples

Example 1

optimizer = PipelineOptimizer()
res1 = optimizer.analyze_dag(
    n=4,
    edges=[[0, 1], [0, 2], [1, 3], [2, 3]],
    durations=[3, 2, 4, 1]
)
# Output: [8, 2]

Explanation: Task 0 runs during [0, 3). Tasks 1 and 2 start at 3: Task 1 runs during [3, 5) and Task 2 runs during [3, 7). Both run concurrently during [3, 5), giving peak parallelism of 2. Task 3 starts at max(5, 7) = 7 and runs during [7, 8). Total latency is 8.

Example 2

optimizer = PipelineOptimizer()
res2 = optimizer.min_checkpoint_cost(
    costs=[3, 2, 4, 1, 5, 2],
    max_window=3
)
# Output: 3

Explanation: Placing checkpoints at index 1 (cost = 2) and index 3 (cost = 1) covers all windows: [0, 2] contains index 1; [1, 3], [2, 4], and [3, 5] contain index 3. Total cost is 2 + 1 = 3.

Constraints

• 1 <= n <= 2 * 10^4
• 0 <= len(edges) <= 5 * 10^4; the graph is guaranteed to be a valid DAG.
• 1 <= durations[i] <= 10^6
• 1 <= m == len(costs) <= 10^5
• 1 <= max_window <= m
• 1 <= costs[i] <= 10^9