RecursiveTask와 RecursiveAction으로 작업 나누기

앞 글에서는 큰 계산을 임계값 이하가 될 때까지 반으로 나누고, 한쪽은 fork()로 맡기고 다른 쪽은 직접 계산한 뒤 join()으로 합치는 Fork/Join 패턴을 의사 코드로 살펴봤습니다. 이 글에서는 그 패턴을 Java의 Fork/Join 프레임워크로 구현합니다. 코드는 길어 보여도 결국 어떻게 나눌지, 언제 그만 나눌지, 결과를 어떻게 합칠지 세 가지를 정하는 일입니다. 그중 나누는 방법이 틀리면 프레임워크가 아무리 잘 동작해도 답이 틀리므로, 나눈 범위가 입력 전체를 정확히 한 번씩 덮는지 계속 확인하며 진행합니다.

Fork/Join을 이루는 클래스

Fork/Join 프레임워크는 java.util.concurrent 패키지의 클래스 몇 개로 이뤄집니다. ForkJoinPool은 작업자 스레드와 작업 대기열을 관리하는 실행기이고, ForkJoinTask는 이 풀에서 실행되는 작업을 나타내는 추상 클래스입니다. 보통은 ForkJoinTask를 직접 상속하지 않고, 결과가 있는지에 따라 다음 두 하위 클래스 중 하나를 상속해 compute() 메서드를 구현합니다.

compute() 안에서 하위 작업을 다룰 때와 풀 밖에서 작업을 맡길 때는 다음 메서드를 씁니다.

RecursiveTask라는 이름에 '재귀'가 들어간 것은 compute()가 자기와 같은 타입의 작업을 만들어 다시 compute()하거나 fork()하기 때문입니다. 다만 일반 재귀 호출과 달리 fork()한 쪽은 다른 작업자가 동시에 실행할 수 있습니다.

RecursiveTask로 리스트의 합 구하기

앞 글과 같은 계산, 즉 1부터 8까지의 각 수에 10을 곱해 더하는 일을 RecursiveTask로 구현해 보겠습니다. 이번에는 값을 List<Integer>에 담아 두고, 작업마다 리스트의 어느 구간을 맡을지를 인덱스 범위로 나타냅니다. 범위는 시작 인덱스는 포함하고 끝 인덱스는 포함하지 않는 반열린 범위(half-open range) [from, to)로 표현합니다. 예를 들어 [0, 2)는 인덱스 0과 1을 뜻하고, 크기는 끝에서 시작을 뺀 2입니다.

import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public final class RecursiveListSumTask {
    private static final class SumTask extends RecursiveTask<Integer> {
        private final List<Integer> values;
        private final int from;
        private final int to;
        private final int threshold;

        private SumTask(List<Integer> values, int from, int to, int threshold) {
            this.values = values;
            this.from = from;
            this.to = to;
            this.threshold = threshold;
        }

        @Override
        protected Integer compute() {
            int size = to - from;
            if (size <= threshold) {
                int subtotal = 0;
                for (int index = from; index < to; index++) {
                    subtotal += values.get(index) * 10;
                }
                return subtotal;
            }

            int middle = from + size / 2;
            SumTask left = new SumTask(values, from, middle, threshold);
            SumTask right = new SumTask(values, middle, to, threshold);
            left.fork();
            int rightResult = right.compute();
            return left.join() + rightResult;
        }
    }

    public static void main(String[] args) {
        List<Integer> values = List.of(1, 2, 3, 4, 5, 6, 7, 8);
        try (ForkJoinPool pool = new ForkJoinPool(4)) {
            int result = pool.invoke(new SumTask(values, 0, values.size(), 2));
            System.out.println("sum=" + result);
        }
    }
}

sum=360

compute()는 먼저 맡은 범위의 크기 size = to - from을 구합니다. 크기가 임계값 이하이면 더 나누지 않고 반복문으로 직접 더해서 돌려주는데, 이렇게 직접 계산하는 작업이 리프입니다. 임계값보다 크면 가운데 인덱스 middle에서 둘로 나눕니다. 두 하위 범위 [from, middle)과 [middle, to)는 middle에서 맞닿아 있으므로 빠지거나 겹치는 인덱스가 없습니다. 임계값은 1 이상이어야 크기 1인 범위가 리프가 되어 재귀가 끝나며, 0이면 어떻게 되는지는 연습 문제 3에서 확인합니다.

나눈 다음의 세 줄이 앞 글에서 본 패턴입니다. left.fork()로 왼쪽 절반을 맡겨 다른 작업자가 가져갈 수 있게 하고, right.compute()로 오른쪽 절반을 지금 스레드에서 바로 계산합니다. 오른쪽 결과를 지역 변수 rightResult에 받아 둔 다음에야 left.join()으로 왼쪽 결과를 기다려 더합니다. main의 pool.invoke()는 루트 작업을 풀에 맡기고 최종 결과가 나올 때까지 기다리며, try-with-resources는 블록이 끝날 때 풀의 close()를 불러 작업자 스레드를 정리합니다.

임계값이 2이면 범위는 다음처럼 나뉩니다. 연결선은 어떤 범위가 어떤 범위로 나뉘었는지를 나타낼 뿐, 작업자가 실행한 순서를 나타내지는 않습니다.

루트 [0, 8)은 [0, 4)와 [4, 8)로 나뉘고, 각각은 다시 크기 2인 리프 둘로 나뉩니다. 리프 [0, 2)는 인덱스 0과 1에 저장된 값 1과 2로 (1 + 2) × 10 = 30을 돌려주고, 같은 방식으로 [2, 4)는 70, [4, 6)은 110, [6, 8)은 150을 돌려줍니다. 인덱스와 값이 1씩 차이 난다는 데 주의하세요. [0, 2)가 가리키는 것은 값 0과 1이 아니라 인덱스 0과 1입니다. 네 리프는 인덱스 0부터 7까지를 정확히 한 번씩 덮고, 부분합은 30 + 70 = 100과 110 + 150 = 260을 거쳐 360으로 합쳐집니다.

작업자 하나로 실행 순서 따라가기

앞 글에서 일반 스레드 풀은 이런 재귀에서 멈췄습니다. Fork/Join은 무엇이 다른지 보려고 new ForkJoinPool(4)를 new ForkJoinPool(1)로 바꿔 작업자를 하나만 남기고, 리프의 return subtotal; 바로 앞에 다음 줄을 넣어 실행해 보겠습니다.

System.out.println("leaf [" + from + ", " + to + ") on " + Thread.currentThread().getName());

leaf [6, 8) on ForkJoinPool-1-worker-1
leaf [4, 6) on ForkJoinPool-1-worker-1
leaf [2, 4) on ForkJoinPool-1-worker-1
leaf [0, 2) on ForkJoinPool-1-worker-1
sum=360

작업자가 하나뿐인데도 재귀가 끝까지 진행되어 sum=360이 나왔고, 네 리프를 모두 같은 작업자가 처리했습니다. 리프가 오른쪽부터 출력된 이유는 다음 표처럼 따라가 보면 알 수 있습니다. 대기열 칸은 fork()로 맡겼지만 아직 아무도 실행하지 않은 작업입니다.

join()을 부른 작업자는 기다리는 작업이 아직 자기 대기열에 있으면 잠들지 않고 그 작업을 직접 꺼내 실행합니다. 그래서 일반 스레드 풀처럼 기다리는 작업이 스레드를 붙잡아 아무도 진행하지 못하는 상황이 생기지 않습니다. 3단계와 5단계에서 보듯 이때 꺼내는 작업은 가장 최근에 fork()한 작업입니다. 작업자가 여러 개라면 그사이 놀고 있던 다른 작업자가 대기열에 남은 [0, 4) 같은 작업을 가져가 동시에 실행하는데, 이 과정은 작업 훔치기와 병렬 스트림에서 자세히 다룹니다.

더 깊이: 직접 만든 풀과 공용 풀

예제는 new ForkJoinPool(4)로 병렬성(parallelism)이 4인 풀을 직접 만들었습니다. 병렬성은 풀이 동시에 활동하게 하려는 작업자 수의 목표로, 이 값을 프로세서 수보다 크게 정해도 프로그램은 동작합니다. 인수 없는 생성자 new ForkJoinPool()은 병렬성을 Runtime.getRuntime().availableProcessors()가 돌려주는 값으로 정합니다. 직접 만든 풀은 다 쓰면 닫아야 하는데, Java 19부터 ExecutorService에 들어온 close()를 ForkJoinPool도 구현하므로 예제처럼 try-with-resources로 닫을 수 있습니다.

JVM에는 따로 만들지 않아도 쓸 수 있는 ForkJoinPool.commonPool(), 즉 공용 풀이 하나 있습니다. ForkJoinPool의 Javadoc에 따르면 특정 풀에 명시적으로 맡기지 않은 ForkJoinTask는 모두 이 풀을 씁니다. 예를 들어 main에서 new SumTask(values, 0, values.size(), 2).invoke()를 부르면 루트 작업은 main 스레드에서 실행되고, 그 안에서 fork()한 하위 작업은 공용 풀로 갑니다. 공용 풀은 JVM 전체가 함께 쓰므로 shutdown()이나 close()를 불러도 종료되지 않습니다. 이 기계(사용 가능한 프로세서 2개)에서 공용 풀의 병렬성은 1이었는데, 이 값을 확인하는 방법과 여러 기능이 공용 풀을 함께 쓸 때 생기는 문제는 병렬 스트림의 결합 법칙과 성능 측정과 서버에서 공용 풀과 전용 풀 나눠 쓰기에서 다룹니다.

범위를 겹치게 나누면 생기는 오답

Fork/Join 코드에서 가장 흔한 실수는 범위를 나누는 지점에서 나옵니다. 다음 프로그램은 양 끝을 모두 포함하는 닫힌 범위 [from, to]로 1부터 8까지의 값을 직접 나타내는데, 오른쪽 하위 작업을 middle + 1이 아니라 middle부터 시작하게 만들었습니다. 리프 기준은 크기 5 이하입니다. 순차 기준은 360이지만 출력은 다릅니다.

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public final class OverlappingForkJoinSplitBug {
    private static final class SumTask extends RecursiveTask<Integer> {
        private final int from;
        private final int to;

        private SumTask(int from, int to) {
            this.from = from;
            this.to = to;
        }

        @Override
        protected Integer compute() {
            if (to - from + 1 <= 5) {
                int sum = 0;
                for (int value = from; value <= to; value++) {
                    sum += value * 10;
                }
                return sum;
            }
            int middle = (from + to) >>> 1;
            SumTask left = new SumTask(from, middle);
            SumTask right = new SumTask(middle, to);
            left.fork();
            return right.compute() + left.join();
        }
    }

    public static void main(String[] args) {
        try (ForkJoinPool pool = new ForkJoinPool(4)) {
            int result = pool.invoke(new SumTask(1, 8));
            System.out.println("wrong-sum=" + result);
        }
    }
}

wrong-sum=400

루트 [1, 8]은 크기가 8이라 임계값 5보다 크므로 middle = (1 + 8) >>> 1 = 4에서 나뉩니다(>>> 1은 부호 없는 오른쪽 시프트로 2로 나누는 연산입니다). 문제는 왼쪽 [1, 4]와 오른쪽 [4, 8]이 모두 4를 포함한다는 것입니다. 왼쪽은 크기 4, 오른쪽은 크기 5라서 둘 다 바로 계산되고, 왼쪽은 10 + 20 + 30 + 40 = 100을, 오른쪽은 40 + 50 + 60 + 70 + 80 = 300을 돌려줍니다. 4에 해당하는 40이 두 번 더해져서 360보다 40 큰 400이 나온 것입니다.

이 버그는 임계값에 따라 더 나쁘게 드러나기도 합니다. 같은 코드에서 리프 기준을 크기 1 이하로 바꾸면 나누는 도중에 범위 [4, 5]가 생기는데, 이 범위의 middle은 4라서 [4, 4]와 [4, 5]로 나뉩니다. 오른쪽이 부모와 똑같은 범위이므로 재귀가 끝나지 않고, 실제로 실행하면 pool.invoke()가 StackOverflowError를 던집니다. 같은 실수가 어떤 임계값에서는 조용히 틀린 답을, 어떤 임계값에서는 끝나지 않는 재귀를 만드는 셈입니다.

고치는 방법은 범위를 표현하는 규칙을 하나 정하고 끝까지 지키는 것입니다. 닫힌 범위를 쓴다면 오른쪽을 middle + 1부터 시작합니다. 반열린 범위를 쓴다면 앞 절처럼 [from, middle)과 [middle, to)로 나누는데, 이때는 middle이 두 범위에 모두 적혀 있어도 왼쪽은 middle을 포함하지 않으므로 겹치지 않습니다. 크기 계산과 종료 조건도 같은 규칙을 따라야 합니다. 닫힌 범위의 크기는 to - from + 1이고 반열린 범위의 크기는 to - from인데, 둘을 섞어 쓰면 원소가 빠지거나 겹치고, 빈 범위가 생기기도 합니다.

임계값은 답이 아니라 작업 수를 바꾼다

범위를 올바르게 나눴다면 임계값을 얼마로 정하든 답은 같아야 합니다. 임계값이 바꾸는 것은 작업이 몇 개로 쪼개지느냐입니다. 다음 프로그램은 1부터 16까지의 합을 임계값 1, 4, 16으로 각각 계산하면서 리프 작업의 수와 재귀의 최대 깊이를 함께 출력합니다. 깊이는 루트를 0으로 세고, 한 번 나눌 때마다 1씩 늘어납니다.

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.atomic.AtomicInteger;

public final class ForkJoinThresholdProbe {
    private record Result(long sum, int depth) {}

    private static final class ProbeTask extends RecursiveTask<Result> {
        private final int from;
        private final int to;
        private final int threshold;
        private final AtomicInteger leaves;
        private final int depth;

        private ProbeTask(int from, int to, int threshold, AtomicInteger leaves, int depth) {
            this.from = from;
            this.to = to;
            this.threshold = threshold;
            this.leaves = leaves;
            this.depth = depth;
        }

        @Override
        protected Result compute() {
            if (to - from <= threshold) {
                leaves.incrementAndGet();
                long sum = 0;
                for (int value = from; value < to; value++) {
                    sum += value;
                }
                return new Result(sum, depth);
            }
            int middle = from + (to - from) / 2;
            ProbeTask left = new ProbeTask(from, middle, threshold, leaves, depth + 1);
            ProbeTask right = new ProbeTask(middle, to, threshold, leaves, depth + 1);
            left.fork();
            Result second = right.compute();
            Result first = left.join();
            return new Result(first.sum() + second.sum(), Math.max(first.depth(), second.depth()));
        }
    }

    public static void main(String[] args) {
        for (int threshold : new int[] {1, 4, 16}) {
            AtomicInteger leaves = new AtomicInteger();
            try (ForkJoinPool pool = new ForkJoinPool(4)) {
                Result result = pool.invoke(new ProbeTask(1, 17, threshold, leaves, 0));
                System.out.printf(
                        "threshold=%d, sum=%d, leaves=%d, depth=%d%n",
                        threshold, result.sum(), leaves.get(), result.depth());
            }
        }
    }
}

threshold=1, sum=136, leaves=16, depth=4
threshold=4, sum=136, leaves=4, depth=2
threshold=16, sum=136, leaves=1, depth=0

이번 작업은 리스트 없이 값 자체를 범위로 다룹니다. 반열린 범위 [1, 17)은 값 1부터 16까지를 뜻하므로 합계는 16 × 17 ÷ 2 = 136입니다. ProbeTask는 합계와 함께 자기 아래에서 가장 깊은 리프의 깊이를 Result 레코드에 담아 돌려주고, 부모는 두 합을 더하고 두 깊이 중 큰 값을 고릅니다. 리프 수는 모든 작업이 함께 쓰는 AtomicInteger로 셉니다. 이 카운터는 분할 구조를 보려고 붙인 계측 도구일 뿐이고, 합계는 공유 변수 없이 반환값으로만 합쳐진다는 점을 눈여겨보세요.

임계값이 1이면 크기 1이 될 때까지 네 번 나눠 리프 16개가 생기고, 4이면 두 번 나눠 크기 4인 리프 4개가 생깁니다. 16이면 루트가 나뉘지 않고 혼자 전부 계산합니다. 세 경우 모두 합계는 136입니다. 리프가 많을수록 여러 작업자에게 나눠 줄 일이 많아지지만, 그만큼 작업 객체를 만들고 대기열에 넣고 기다리는 비용도 늘어납니다. 반대로 리프가 하나뿐이면 작업자가 몇 개든 한 스레드가 모든 일을 합니다.

그렇다면 임계값은 어떻게 정할까요? ForkJoinTask의 Javadoc은 아주 대략적인 기준으로, 작업 하나가 기본 연산을 100번보다 많이, 10,000번보다 적게 수행하게 하라고 안내합니다. 작업이 너무 크면 병렬로 처리해도 처리량이 늘지 않고, 너무 작으면 메모리와 작업 관리 비용이 계산을 압도하기 때문입니다. 이 기준에서 출발해 실제 데이터로 측정하며 조정하고, 입력 전체가 작다면 아예 나누지 않고 순차로 계산하는 분기를 두는 것도 좋습니다. 이때 앞의 AtomicInteger 같은 계측 코드는 모든 리프가 같은 변수를 갱신하면서 그 자체로 경쟁을 일으키므로, 구조를 확인하는 실행과 시간을 재는 실행은 따로 합니다.

더 깊이: 임계값에 따라 걸리는 시간

임계값이 실행 시간을 얼마나 바꾸는지 직접 재 보겠습니다. 다음 프로그램의 SumTask는 앞의 RecursiveListSumTask와 같은 구조인데, 리스트 대신 long 배열을 쓰고 합을 long으로 계산합니다. main은 0부터 99까지가 반복되는 값 1,000만 개를 채운 뒤, 임계값마다 같은 풀에서 열 번씩 실행해 가장 짧은 시간을 출력합니다. 처음 몇 번의 실행은 JIT 컴파일러가 코드를 최적화할 시간을 주는 예열 역할을 하고, 풀은 인수 없는 생성자로 만들어 병렬성이 사용 가능한 프로세서 수와 같습니다.

import java.util.Arrays;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public final class ThresholdTiming {
    private static final class SumTask extends RecursiveTask<Long> {
        private final long[] values;
        private final int from;
        private final int to;
        private final int threshold;

        private SumTask(long[] values, int from, int to, int threshold) {
            this.values = values;
            this.from = from;
            this.to = to;
            this.threshold = threshold;
        }

        @Override
        protected Long compute() {
            if (to - from <= threshold) {
                long sum = 0;
                for (int index = from; index < to; index++) {
                    sum += values[index];
                }
                return sum;
            }
            int middle = from + (to - from) / 2;
            SumTask left = new SumTask(values, from, middle, threshold);
            SumTask right = new SumTask(values, middle, to, threshold);
            left.fork();
            long rightSum = right.compute();
            return left.join() + rightSum;
        }
    }

    public static void main(String[] args) {
        long[] values = new long[10_000_000];
        Arrays.setAll(values, index -> index % 100);
        try (ForkJoinPool pool = new ForkJoinPool()) {
            System.out.println("parallelism=" + pool.getParallelism());
            for (int threshold : new int[] {1, 10, 100, 1_000, 10_000, 100_000, 1_000_000, values.length}) {
                long best = Long.MAX_VALUE;
                long sum = 0;
                for (int round = 0; round < 10; round++) {
                    long start = System.nanoTime();
                    sum = pool.invoke(new SumTask(values, 0, values.length, threshold));
                    best = Math.min(best, System.nanoTime() - start);
                }
                System.out.printf("threshold=%,d sum=%d best=%.1fms%n", threshold, sum, best / 1e6);
            }
        }
    }
}

출력의 첫 줄은 풀의 병렬성이고, 이어서 임계값마다 합계와 가장 짧은 시간이 한 줄씩 나옵니다. 합계는 어느 임계값에서나 sum=495000000이어야 하지만, 시간은 기계와 그때의 부하에 따라 크게 달라지므로 여기에 정해진 값을 적을 수 없습니다. 이 글을 쓴 실습 기계는 사용 가능한 프로세서가 2개였고(첫 줄이 parallelism=2), 다른 프로그램과 CPU를 함께 쓰는 환경이었습니다. 프로그램을 아홉 번 실행했는데, 다음 표는 그중 기계가 유난히 바빴던 두 번을 뺀 일곱 번의 측정 범위입니다. 빠진 두 번은 거의 모든 값이 몇 배씩 느렸습니다.

숫자는 실행마다 흔들려도 아홉 번 모두 같았던 경향이 있습니다. 임계값 1과 10은 언제나 가장 느린 두 개였고, 임계값 1은 같은 실행의 가장 빠른 임계값보다 약 24배에서 90배 느렸습니다. 리프 하나가 덧셈을 한 번만 하고 끝나니, 작업 객체를 만들고 맡기고 기다리는 비용이 계산 비용을 압도한 것입니다. 반면 1,000 이상의 임계값은 서로 비슷한 범위에 있어서, 좋은 임계값은 한 점이 아니라 꽤 넓은 구간이라는 것을 알 수 있습니다.

나누지 않은 경우(리프 1개)와 비교하면 병렬화의 이득은 크지 않았습니다. 각 실행에서 1,000 이상 가운데 가장 빠른 임계값을 골라도 나누지 않은 경우보다 빨라야 2배 남짓이었고, 거의 차이가 없는 실행도 있었습니다. 기계가 바빴던 한 번은 오히려 나누지 않은 쪽이 두 배 넘게 빨랐습니다. 프로세서가 2개이니 기대할 수 있는 최대치가 2배 정도인 데다, 원소 하나에 덧셈 한 번뿐인 계산은 값을 메모리에서 읽어 오는 속도에 묶이기 쉽기 때문입니다.

원소마다 계산이 무겁고 코어가 많을수록 이득이 커지므로 자기 기계에서 직접 실행해 비교해 보세요. 믿을 만한 측정에 필요한 도구와 방법은 JIT 예열과 벤치마크에서 다룹니다.

결과가 없는 작업은 RecursiveAction으로

작업이 값을 돌려줄 필요 없이 일만 하면 된다면 RecursiveAction을 상속합니다. 대표적인 예가 배열을 제자리에서 바꾸는 작업입니다. 다음 프로그램은 배열의 모든 원소를 두 배로 만듭니다.

import java.util.Arrays;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

public final class DisjointArrayTransformAction {
    private static final class DoubleAction extends RecursiveAction {
        private final int[] values;
        private final int from;
        private final int to;

        private DoubleAction(int[] values, int from, int to) {
            this.values = values;
            this.from = from;
            this.to = to;
        }

        @Override
        protected void compute() {
            if (to - from <= 2) {
                for (int index = from; index < to; index++) {
                    values[index] *= 2;
                }
                return;
            }
            int middle = from + (to - from) / 2;
            invokeAll(new DoubleAction(values, from, middle), new DoubleAction(values, middle, to));
        }
    }

    public static void main(String[] args) {
        int[] values = {1, 2, 3, 4, 5, 6, 7, 8};
        try (ForkJoinPool pool = new ForkJoinPool(4)) {
            pool.invoke(new DoubleAction(values, 0, values.length));
        }
        System.out.println(Arrays.toString(values));
    }
}

[2, 4, 6, 8, 10, 12, 14, 16]

compute()의 반환 타입이 void라는 점만 빼면 구조는 같습니다. 크기 2 이하인 범위는 반복문으로 직접 두 배로 만들고, 그보다 크면 반으로 나눈 두 하위 작업을 invokeAll에 넘깁니다. invokeAll(a, b)는 두 작업을 실행하고 둘 다 끝날 때까지 기다리는 메서드라서, 합칠 결과가 없을 때는 fork()와 join()을 직접 쓰는 것보다 간단합니다. 하위 작업 중 하나가 예외로 끝나면 invokeAll이 그 예외를 다시 던집니다.

여러 작업자가 같은 배열을 동시에 고치는데도 안전한 이유는 각 리프가 맡는 인덱스 범위가 서로 겹치지 않기 때문입니다. [0, 2)를 맡은 리프는 인덱스 0과 1만, [2, 4)를 맡은 리프는 2와 3만 고치므로 두 작업자가 같은 원소를 건드리는 일이 없습니다. 반대로 여러 리프가 같은 원소를 고치거나 하나의 ArrayList에 함께 add하면 데이터 경합(data race)이 생깁니다. main은 pool.invoke()가 돌아온 뒤에 배열을 출력합니다. pool.invoke()는 모든 하위 작업이 끝난 뒤에야 돌아오고, Future.get()과 마찬가지로 작업이 끝나기를 기다린 쪽에는 작업이 한 기록이 보이므로 바뀐 값이 빠짐없이 출력됩니다.

fork와 join을 쓸 때 지킬 규칙

앞의 예제들은 모두 fork(), compute(), join() 순서를 지켰습니다. 이 순서가 왜 중요한지는 순서를 바꿔 보면 드러납니다. 다음은 자주 보이는 잘못된 코드입니다.

left.fork();
return left.join() + right.compute(); // 왼쪽이 끝나야 오른쪽 계산을 시작한다

Java는 +의 왼쪽 피연산자를 먼저 계산하므로 left.join()이 먼저 실행되고, 현재 작업자는 왼쪽 결과가 나올 때까지 오른쪽 계산을 시작하지 못합니다. 그사이 다른 작업자가 left를 가져가지 않았다면 앞의 표에서 본 것처럼 현재 작업자가 left를 직접 실행하고, 그것이 끝난 뒤에야 right를 계산합니다. 결국 두 절반을 한 스레드가 차례로 계산하는 셈입니다. 답은 여전히 맞기 때문에 테스트로는 드러나지 않고 느려지기만 합니다. 이 기계에서 리프 32,768개짜리 계산을 두 순서로 비교해 보니, 예열이 끝난 뒤에는 잘못된 순서가 매번 더 오래 걸렸습니다.

이 밖에도 다음 규칙을 지키면 Fork/Join 코드가 맞고 빠르게 동작합니다.

• 하위 작업 하나는 fork()하고, 다른 하나는 직접 compute()합니다.
• join()은 결과가 정말 필요한 지점까지 미룹니다. 위의 잘못된 코드처럼 join()을 먼저 부르면 병렬성이 사라집니다.
• 여러 작업을 fork()했다면 가장 나중에 fork()한 작업부터 join()합니다. Javadoc도 a.fork(); b.fork(); b.join(); a.join();이 a를 먼저 join하는 것보다 훨씬 효율적일 가능성이 높다고 설명합니다. 작업자는 가장 최근에 맡긴 작업부터 직접 꺼내 실행하기 때문입니다.
• 결과를 합치는 연산은 순차 계산과 같은 답을 내야 합니다. 덧셈이나 최댓값처럼 묶는 순서를 바꿔도 결과가 같은 연산, 즉 결합 법칙을 만족하는 연산을 씁니다. 자세한 내용은 병렬 스트림의 결합 법칙과 성능 측정에서 다룹니다.
• compute() 안에서 블로킹 I/O를 하거나 락을 오래 잡지 않습니다. 풀이 관리하지 않는 기다림은 작업자를 묶어 둡니다.
• 작업끼리 서로를 join()하는 순환이 생기면 안 됩니다. Javadoc은 완료를 기다리는 관계가 순환하지 않을 때, 즉 계산을 방향성 비순환 그래프(DAG)로 나타낼 수 있을 때만 join()이 적합하며, 그렇지 않으면 작업들이 서로를 기다리는 교착 상태에 빠질 수 있다고 경고합니다.

정리

• ForkJoinPool은 작업자와 대기열을 관리하고, 계산은 RecursiveTask<V>(결과 있음)나 RecursiveAction(결과 없음)의 compute()에 작성합니다.
• compute()는 임계값 이하면 직접 계산하고, 아니면 둘로 나눠 left.fork(), right.compute(), left.join() 순서로 진행합니다. join()을 먼저 부르면 답은 맞아도 병렬성이 사라집니다.
• join()을 부른 작업자는 기다리는 작업이 자기 대기열에 남아 있으면 직접 실행하므로, 작업자가 하나뿐이어도 재귀가 끝까지 진행됩니다.
• 범위는 한 가지 규칙으로 표현합니다. 반열린 범위 [from, middle)과 [middle, to)는 겹치지 않고, 닫힌 범위라면 오른쪽을 middle + 1부터 시작합니다.
• 임계값은 답을 바꾸지 않고 작업 수만 바꿉니다. 너무 작으면 작업 관리 비용이 계산을 압도하므로, Javadoc의 '기본 연산 100번에서 10,000번' 기준에서 출발해 측정으로 조정합니다.
• RecursiveAction은 리프마다 겹치지 않는 영역을 고칠 때 안전합니다.

연습 문제

1. 반열린 인덱스 범위를 사용해 int 배열의 최댓값을 구하는 RecursiveTask<Integer>를 작성하세요. 빈 범위를 만들지 않아야 하고, 크기가 임계값 3 이하인 범위는 직접 순회하며, 부모 작업은 Math.max로 두 결과를 합칩니다. 배열 {4, 17, -3, 29, 8, 11, 2}에서 max=29가 나와야 합니다.
2. 1번 풀이의 배열을 {-4, -17, -3, -29, -8, -11, -2}로 바꾸면 리프 범위와 각 리프의 최댓값, 최종 결과는 무엇인가요? 리프에서 최댓값의 초깃값을 values[from] 대신 0으로 두면 어떻게 되나요? 빈 배열을 넘기면 어떻게 되나요?
3. ForkJoinThresholdProbe와 같은 규칙으로 값 1부터 5까지, 즉 반열린 범위 [1, 6)을 나눈다고 합시다. 임계값이 2일 때와 1일 때 각각 리프 수, 최대 깊이, 합계를 구하세요. 임계값이 0이면 어떻게 되나요?

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public final class ForkJoinMaximumSolution {
    private static final class MaxTask extends RecursiveTask<Integer> {
        private final int[] values;
        private final int from;
        private final int to;

        private MaxTask(int[] values, int from, int to) {
            this.values = values;
            this.from = from;
            this.to = to;
        }

        @Override
        protected Integer compute() {
            if (to - from <= 3) {
                int maximum = values[from];
                for (int index = from + 1; index < to; index++) {
                    maximum = Math.max(maximum, values[index]);
                }
                return maximum;
            }
            int middle = from + (to - from) / 2;
            MaxTask left = new MaxTask(values, from, middle);
            MaxTask right = new MaxTask(values, middle, to);
            left.fork();
            return Math.max(right.compute(), left.join());
        }
    }

    public static void main(String[] args) {
        int[] values = {4, 17, -3, 29, 8, 11, 2};
        try (ForkJoinPool pool = new ForkJoinPool(3)) {
            int maximum = pool.invoke(new MaxTask(values, 0, values.length));
            System.out.println("max=" + maximum);
        }
    }
}

max=29

다음 글

작업 훔치기와 병렬 스트림에서는 이 글에서 대기열 칸으로만 보여 준 작업자별 대기열과, 일이 없는 작업자가 다른 작업자의 작업을 가져오는 작업 훔치기를 살펴봅니다. 이어서 같은 Fork/Join 구조 위에서 동작하는 병렬 Stream의 처리 순서와 주의할 점을 다룹니다.