DAG (방향성 비순환 그래프) 와 위상정렬

병렬처리를 작업할 때 필요한 개념에 대해서 공부하고 정리해봤다.

DAG (방향성 비순환 그래프)

순환하지 않고 한 방향으로만 흐르는 그래프.

주로 작업 간의 종속성과 실행 순서를 정의할 때 사용된다.

위상정렬

순서가 정해져 있는 일련의 작업을 차례대로 수행해야 할 때 사용할 수 있는 알고리즘

위상정렬을 왜 하는가 ?

순서를 어떻게 정할 것 인가?
의존성 그래프 자체는 그냥 "누가 누구를 기다리는지"에 대한 관계의 집합일 뿐이다. 그 자체로는 "그럼 뭐부터 실행해야 하는데?" 라는 질문에 답할 수 없다.

위상정렬은 "안전하게 실행 가능한 순서"를 기계적으로 계산해주는 알고리즘이다.
위상정렬이 하는 일은 크게 2가지가 있다.

1) 지금 뭘 실행해도 안전한가?
2) 동시에 실행해도 되는 게 뭔가? (병렬처리의 핵심)

동작원리

1. 진입차수가 0인 노드를 큐에 넣는다 -> 정의상 "아무것도 기다릴 필요 없는" 노드 = 지금 당장 병렬 실행 가능한 태스크들

2. 큐에서 하나 꺼내 처리(완료) 표시하고, 그 노드의 Dependents (후속 태스크)들의 진입차수를 1씩 감소시킴 (하나의 선행 조건이 사라졌으므로)

3. 감소시킨 결과 진입차수가 0이 된 노드가 있으면 다음 차례 큐에 추가

4. 큐가 빌 때까지 반복. 처리한 노드 수가 전체 노드 수보다 적게 끝나면 -> 진입차수가 절대 0이 되지 못하는 노드가 남아있다는 뜻 = 사이클 존재

예시 코드 (using Unitask)

using System;
using System.Collections.Generic;
using System.Linq;
using Cysharp.Threading.Tasks;

// ── 노드 정의 ──────────────────────────────────────────
public class TaskNode
{
    public string Id;
    public List<string> Dependents = new();   // 이 태스크 완료 시 풀어줄 후속 태스크들
    public int InDegree;                       // 남은 선행 의존성 개수
    public Func<UniTask> Execute;               // 실제 실행할 작업 (빌드, 다운로드 등)
}

// ── 이벤트 기반 DAG 실행기 ──────────────────────────────
public class ParallelDagExecutor
{
    private readonly Dictionary<string, TaskNode> _nodes = new();

    public void AddTask(string id, Func<UniTask> execute)
        => _nodes[id] = new TaskNode { Id = id, Execute = execute };

    public void AddDependency(string from, string dependsOn)
    {
        _nodes[dependsOn].Dependents.Add(from);
        _nodes[from].InDegree++;
    }

    public async UniTask RunAsync(CancellationToken ct = default)
    {
        var inDegree = _nodes.ToDictionary(n => n.Key, n => n.Value.InDegree);
        var totalCount = _nodes.Count;
        var completedCount = 0;
        var faulted = (Exception)null;

        // 전체 파이프라인 완료를 이벤트로 알리는 신호
        var pipelineDone = new UniTaskCompletionSource();

        // 진입차수 0인 노드 = 즉시 실행 가능한 태스크들
        var readyNodes = _nodes.Values.Where(n => inDegree[n.Id] == 0).ToList();

        if (readyNodes.Count == 0 && totalCount > 0)
            throw new InvalidOperationException("사이클 존재 또는 시작 노드 없음");

        foreach (var node in readyNodes)
            FireTask(node);
   		

        void FireTask(TaskNode node)
        {
            // fire-and-forget이지만 완료/예외를 이벤트로 처리
            RunNode(node, ct).Forget(ex =>
            {
                faulted = ex;
                pipelineDone.TrySetException(ex);
            });
        }

        async UniTask RunNode(TaskNode node, CancellationToken token)
        {
            await node.Execute();               // 실제 태스크 실행 (await 자체가 완료 이벤트 역할)
            token.ThrowIfCancellationRequested();

            completedCount++;

            // 후속 태스크들의 진입차수를 개별적으로 감소 → 0이 되는 즉시 실행
            foreach (var dependentId in node.Dependents)
            {
                if (--inDegree[dependentId] == 0)
                    FireTask(_nodes[dependentId]);
            }

            if (completedCount == totalCount)
                pipelineDone.TrySetResult();      // 마지막 태스크 완료 시 전체 완료 신호
        }

        await pipelineDone.Task;

        if (faulted != null)
            throw faulted;
    }
}