작업을 나누고 합치는 Fork/Join 패턴

CPU를 오래 쓰는 계산이 있고 컴퓨터에 코어가 여러 개 있다면, 계산을 여러 코어에 나눠 맡기고 싶어집니다. 스레드 하나는 한 순간에 코어 하나에서만 실행되므로, 여러 코어를 쓰려면 일을 여러 스레드에 나눠 줘야 합니다. 이때 할 일은 세 가지입니다. 일을 서로 독립적인 조각으로 나누고, 조각들을 여러 스레드에서 실행하고, 조각마다 나온 부분 결과를 하나로 합칩니다. 이 가운데 하나라도 틀리면 프로그램은 빨라지기는커녕 틀린 답을 더 빨리 내놓습니다.

이 글에서는 1부터 8까지의 각 수에 10을 곱해 모두 더하는 작은 계산 하나를 여러 방법으로 풀어 봅니다. 먼저 한 스레드로 정답을 구해 두고, 스레드를 직접 만들어 나눠 본 다음, 스레드 풀에 맡겨 봅니다. 그 과정에서 결과를 너무 일찍 읽는 실수와, 일을 재귀적으로 나눌 때 일반 스레드 풀이 멈춰 버리는 문제를 차례로 만납니다. 두 번째 문제를 풀려고 만든 도구가 이 장의 주제인 Fork/Join 프레임워크입니다.

한 스레드로 정답부터 구하기

병렬 버전을 만들기 전에, 같은 입력을 한 스레드로 계산한 결과를 먼저 구해 둡니다. 이 결과를 순차 기준이라고 부르겠습니다. 병렬 버전이 맞았는지 판단하려면 비교할 정답이 있어야 하기 때문입니다. 다음 프로그램은 calculate로 각 수에 10을 곱하고 IntStream으로 모두 더합니다.

import java.util.stream.IntStream;

public final class SequentialHeavyJobBaseline {
    private static int calculate(int value) {
        return value * 10;
    }

    public static void main(String[] args) {
        int sum = IntStream.rangeClosed(1, 8).map(SequentialHeavyJobBaseline::calculate).sum();
        System.out.println("expected-sum=" + sum);
    }
}

expected-sum=360

10 + 20 + … + 80은 10 × (1 + 2 + … + 8) = 10 × 36 = 360입니다. 이 글의 병렬 버전들은 모두 이 360을 내야 하고, 범위를 어떻게 나누든 1부터 8까지의 수를 하나도 빠뜨리거나 두 번 세지 않아야 합니다.

실제 프로그램이라면 calculate는 값 하나에 수십 밀리초씩 걸리는 무거운 계산일 것입니다. 하지만 이 글의 예제는 결과가 맞는지 확인하는 것이 목적이라서, 계산을 가볍게 두고 항상 같은 답이 나오는 작은 입력을 씁니다. sleep을 넣어 무거운 작업을 흉내 내면 실행할 때마다 기다려야 할 뿐, 답이 맞는지 확인하는 데는 도움이 되지 않습니다. 속도를 재는 실험은 입력을 따로 키워서 해야 하며, 다음 글에서 한 번 해 봅니다.

스레드를 시작만 하고 결과를 읽으면

이제 범위를 1부터 4까지와 5부터 8까지로 나눠 두 스레드에 맡겨 보겠습니다. 다음 프로그램에는 흔한 실수가 하나 들어 있습니다. main이 두 작업자가 시작했다는 것만 확인하고, 계산이 끝나기를 기다리지 않은 채 결과를 더합니다.

import java.util.concurrent.CountDownLatch;

public final class UnjoinedWorkerResultBug {
    private static final class Task implements Runnable {
        private final int from;
        private final int to;
        private final CountDownLatch started;
        private final CountDownLatch release;
        private int result;

        private Task(int from, int to, CountDownLatch started, CountDownLatch release) {
            this.from = from;
            this.to = to;
            this.started = started;
            this.release = release;
        }

        @Override
        public void run() {
            started.countDown();
            try {
                release.await();
            } catch (InterruptedException exception) {
                Thread.currentThread().interrupt();
                return;
            }
            for (int value = from; value <= to; value++) {
                result += value * 10;
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch started = new CountDownLatch(2);
        CountDownLatch release = new CountDownLatch(1);
        Task left = new Task(1, 4, started, release);
        Task right = new Task(5, 8, started, release);
        Thread first = new Thread(left, "left-worker");
        Thread second = new Thread(right, "right-worker");
        first.start();
        second.start();
        started.await();

        System.out.println("wrong-sum=" + (left.result + right.result));
        release.countDown();
        first.join();
        second.join();
    }
}

wrong-sum=0

이 프로그램은 실수의 결과를 매번 똑같이 보여 주려고 CountDownLatch 두 개를 씁니다. CountDownLatch는 정해 둔 횟수만큼 countDown()이 불릴 때까지 await()를 부른 스레드를 기다리게 하는 도구입니다. started는 개수가 2이므로, 두 작업자가 run()에 들어와 countDown()을 한 번씩 부르면 main의 started.await()가 풀립니다. 반대로 작업자들은 release.await()에서 멈춰 있다가 main이 release.countDown()을 불러야 계산을 시작합니다. 그래서 실행 순서가 다음처럼 고정됩니다.

main이 결과를 읽는 2단계에서 작업자들은 아직 계산을 시작하지도 않았으므로 두 result 필드는 초깃값 0입니다. 3단계부터의 release.countDown()과 두 join()은 작업자들이 계산을 마치고 프로그램이 정상적으로 끝나게 해 줄 뿐, 이미 출력한 값을 고치지는 못합니다.

래치가 없는 실제 코드에서는 결과가 이렇게 일정하지 않습니다. 작업자가 얼마나 빨리 실행되느냐에 따라 0이 나올 수도 있고, 100이나 260, 계산 도중의 값, 운이 좋으면 360이 나올 수도 있습니다. 대부분의 실행에서 우연히 맞는 답이 나오면 테스트도 통과해 버리므로 오히려 더 찾기 어려운 버그가 됩니다. 이 예제의 래치는 그 가운데 가장 나쁜 경우를 매번 재현하려고 넣은 장치입니다.

원인은 start()가 스레드를 시작시킬 뿐 끝날 때까지 기다려 주지 않는다는 데 있습니다. 작업자가 시작했다는 사실은 작업자가 끝났는지에 대해 아무것도 알려 주지 않습니다. result에 volatile을 붙이면 해결될 것 같지만 그렇지 않습니다. volatile은 다른 스레드가 쓴 최신 값을 읽게 해 주지만, 계산이 끝나기 전에 읽으면 그 최신 값이 0이거나 계산 도중의 값일 뿐입니다. 필요한 것은 계산이 끝난 시점을 아는 것이고, 결과는 반드시 그 뒤에 읽어야 합니다.

join으로 끝나기를 기다린 뒤 더하기

고치는 방법은 join으로 스레드가 끝날 때까지 기다리기에서 본 그대로입니다. 두 스레드를 모두 시작한 다음, 각 스레드에 join()을 불러 끝날 때까지 기다리고 나서 결과를 더합니다. 다음 JoinedRangeWorkers는 래치를 걷어 내고 그렇게 고친 버전입니다.

public final class JoinedRangeWorkers {
    private static final class RangeTask implements Runnable {
        private final int from;
        private final int to;
        private int result;

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

        @Override
        public void run() {
            int local = 0;
            for (int value = from; value <= to; value++) {
                local += value * 10;
            }
            result = local;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        RangeTask left = new RangeTask(1, 4);
        RangeTask right = new RangeTask(5, 8);
        Thread first = new Thread(left, "range-1-4");
        Thread second = new Thread(right, "range-5-8");

        first.start();
        second.start();
        first.join();
        second.join();

        int sum = left.result + right.result;
        System.out.println("joined-sum=" + sum);
    }
}

joined-sum=360

첫 작업자는 (1 + 2 + 3 + 4) × 10 = 100을, 둘째 작업자는 (5 + 6 + 7 + 8) × 10 = 260을 자기 result 필드에 저장합니다. 단계 그림처럼 main은 첫 join()이 돌아오면 왼쪽 작업자의 기록을, 둘째 join()이 돌아오면 오른쪽 작업자의 기록을 확인한 셈이고, 두 join()이 모두 돌아온 뒤에야 100 + 260 = 360을 계산합니다. 어느 작업자가 먼저 끝나는지는 실행할 때마다 다를 수 있지만 결과는 달라지지 않습니다. 오른쪽 작업자가 먼저 끝났다면 둘째 join()이 기다리지 않고 바로 돌아올 뿐입니다.

join() 뒤에 읽은 값을 믿을 수 있는 것은 Java 메모리 모델이 그렇게 정해 두었기 때문입니다. JLS 17.4.5는 한 스레드의 모든 동작이, 다른 스레드가 그 스레드에 대한 join()에서 정상적으로 돌아오기보다 먼저 일어난다(happens-before)고 정합니다. 그래서 join()에서 돌아온 main에게는 작업자가 result에 쓴 값이 보입니다. 이 규칙은 기록이 보이는 조건을 정할 뿐이고, 작업자들이 끝나는 순서를 정하지는 않습니다.

각 작업자가 자기 필드에만 쓴다는 점도 중요합니다. 두 작업자가 같은 int sum 변수에 sum += …을 하면 읽기, 더하기, 쓰기 세 단계가 서로 끼어들어 한쪽의 갱신이 사라질 수 있습니다(가시성·원자성·임계 영역). 반면 부분 결과를 작업자마다 따로 두고 마지막에 한 스레드에서 합치면, 여러 스레드가 함께 고치는 변수가 아예 없으므로 락도 필요 없습니다. 또 run()은 지역 변수 local에 합을 모은 뒤 마지막에 한 번만 result에 쓰므로, 앞의 버그 예제와 달리 계산 도중의 값이 필드에 드러나지 않습니다. 이 '각자 계산하고 마지막에 합친다'는 구조는 뒤의 Fork/Join에서도 그대로 쓰입니다.

스레드 풀에 맡기고 Future로 결과 받기

작업마다 Thread를 새로 만들면 스레드를 만들고 없애는 비용이 들고, 스레드 객체를 일일이 들고 있다가 join()해야 합니다. ExecutorService로 작업 실행하기에서 본 스레드 풀을 쓰면 미리 만들어 둔 작업자 스레드를 재사용하고, 작업을 Callable로 맡긴 뒤 결과를 Future로 받을 수 있습니다. 다음 프로그램은 같은 두 범위를 작업자가 2개인 고정 크기 스레드 풀에 맡깁니다.

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public final class PooledRangeResults {
    private static Callable<Integer> range(int from, int to) {
        return () -> {
            int subtotal = 0;
            for (int value = from; value <= to; value++) {
                subtotal += value * 10;
            }
            return subtotal;
        };
    }

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        try (ExecutorService pool = Executors.newFixedThreadPool(2)) {
            List<Future<Integer>> futures = new ArrayList<>();
            futures.add(pool.submit(range(1, 4)));
            futures.add(pool.submit(range(5, 8)));

            int sum = 0;
            for (Future<Integer> future : futures) {
                sum += future.get();
            }
            System.out.println("pooled-sum=" + sum);
        }
    }
}

pooled-sum=360

range는 범위의 부분합을 돌려주는 Callable<Integer>를 만듭니다. submit은 작업을 풀에 맡기고 곧바로 Future를 돌려주며, 두 작업을 모두 맡긴 다음에 반복문이 future.get()으로 결과를 하나씩 받습니다. get()은 작업이 끝날 때까지 기다렸다가 결과를 돌려주므로, 앞 절에서 join()과 필드가 나눠 하던 일을 한 번에 해 줍니다. submit 바로 뒤에서 get()을 부르면 작업이 하나씩 실행되는 이유는 Future.get을 부르는 시점과 병렬 합산에서 다뤘습니다.

블록을 벗어날 때 try-with-resources가 ExecutorService의 close()를 부릅니다. Java 19에서 추가된 이 메서드는 새 작업을 더 받지 않고, 이미 맡긴 작업이 모두 끝날 때까지 기다린 다음 돌아옵니다. 그래서 풀을 따로 종료하는 코드 없이도 프로그램이 깔끔하게 끝납니다.

결과를 기다리는 방법을 지금까지 두 가지 봤고, 다음 글에서 ForkJoinTask.join()이 하나 더 나옵니다. 세 방법은 작업이 예외로 끝났을 때 하는 일이 서로 다릅니다.

Thread.join()은 작업자가 예외로 끝나도 아무 일 없었다는 듯 돌아오기 때문에, 결과 필드가 왜 비어 있는지 호출한 쪽에서는 알 수 없습니다. 기다리는 도중 인터럽트를 받았을 때도 차이가 있습니다. Thread.join()과 Future.get()은 InterruptedException을 던지지만, ForkJoinTask.join()은 인터럽트로 중단되지 않고 계속 기다립니다.

재귀로 나누면 스레드 풀이 멈추는 이유

지금까지는 범위를 미리 두 조각으로 나눴습니다. 조각 수를 미리 정하는 방식은 단순하지만 한계가 있습니다. 조각마다 걸리는 시간이 다르면 가장 늦은 조각이 끝날 때까지 나머지 스레드는 놀게 되고, 입력 크기나 코어 수가 바뀔 때마다 몇 조각으로 나눌지 다시 정해야 합니다. 그래서 큰 계산은 흔히 '충분히 작아질 때까지 반으로 나눈다'는 규칙으로 나눕니다. 작업이 크면 자기를 두 하위 작업으로 나누고, 하위 작업도 크면 다시 나누는 재귀 구조입니다. 문제를 같은 모양의 작은 문제로 쪼개 풀고 그 답을 합치는 이런 방식을 분할 정복(divide and conquer)이라고 합니다.

그런데 이 재귀를 일반 스레드 풀에서 그대로 하면 문제가 생깁니다. 다음 프로그램은 범위의 크기가 2 이하가 될 때까지 반으로 나누고, 나눈 두 하위 작업을 풀에 맡긴 다음 get()으로 결과를 기다립니다. 풀의 작업자는 2개이고, 결과를 영원히 기다리지 않도록 main은 1초까지만 기다립니다.

import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public final class RecursiveSubmitStall {
    // Executors.newFixedThreadPool(2)와 같은 구성: 작업자 2개, 크기 제한이 없는 큐
    private static final ThreadPoolExecutor POOL =
            new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());

    private static int sum(int from, int to) throws Exception {
        if (to - from + 1 <= 2) {
            int subtotal = 0;
            for (int value = from; value <= to; value++) {
                subtotal += value * 10;
            }
            return subtotal;
        }
        int middle = (from + to) / 2;
        Future<Integer> left = POOL.submit(() -> sum(from, middle));
        Future<Integer> right = POOL.submit(() -> sum(middle + 1, to));
        return left.get() + right.get();
    }

    public static void main(String[] args) throws Exception {
        Future<Integer> root = POOL.submit(() -> sum(1, 8));
        try {
            System.out.println("sum=" + root.get(1, TimeUnit.SECONDS));
        } catch (TimeoutException exception) {
            System.out.println(
                    "timeout: busy=" + POOL.getActiveCount() + ", queued=" + POOL.getQueue().size());
        } finally {
            POOL.shutdownNow();
        }
    }
}

timeout: busy=2, queued=3

1초가 지나도 루트 작업이 끝나지 않았습니다. busy는 작업을 실행 중인 스레드 수이고, queued는 큐에서 차례를 기다리는 작업 수입니다. 작업자 2개가 모두 바쁜데 큐에는 작업 3개가 남아 있다는 뜻입니다. 무슨 일이 있었는지 차례로 따라가 보겠습니다.

3단계에서 두 작업자는 모두 get() 안에서 기다리고 있습니다. [1, 2]가 실행돼야 [1, 4]가 끝나고 [1, 4]가 끝나야 루트가 끝나는데, [1, 2]를 실행할 작업자가 남아 있지 않습니다. 기다리는 작업이 스레드를 붙잡고 있어서, 그 기다림을 풀어 줄 작업이 실행되지 못하는 것입니다. 이렇게 풀의 스레드가 모두 다른 작업의 결과를 기다리느라 멈추는 상황을 스레드 기아 교착 상태(thread starvation deadlock)라고 부르며, 저절로는 풀리지 않습니다. 이 예제는 마지막에 shutdownNow()가 기다리는 작업자들을 인터럽트하기 때문에 프로그램이 끝납니다.

작업자를 늘리면 될 것 같지만 근본적인 해결책이 아닙니다. 이 기계에서 실험해 보면, 작업자를 4개로 늘렸을 때 1부터 8까지는 끝났지만 범위를 1부터 16까지로 늘리자 다시 멈췄습니다. 결과를 기다리는 작업, 즉 재귀 트리에서 리프가 아닌 작업의 수는 입력이 커질수록 늘어나고, 그 작업들이 동시에 스레드를 하나씩 붙잡을 수 있기 때문입니다. 게다가 작업자 수가 아슬아슬할 때는 스레드가 만들어지는 타이밍에 따라 끝나기도 하고 멈추기도 해서 테스트로 잡아내기도 어렵습니다. 작업자 3개로 1부터 8까지를 계산하면 이 기계에서는 끝나는 실행과 멈추는 실행이 섞여 나왔습니다.

Fork/Join은 바로 이 문제를 풀려고 만든 프레임워크입니다. Fork/Join의 작업자는 하위 작업의 결과를 기다려야 할 때 그냥 잠들지 않고, 아직 처리되지 않은 작업을 찾아 대신 실행합니다. 기다리는 바로 그 하위 작업이 아직 자기 대기열에 남아 있으면 직접 꺼내 실행하고, 다른 작업자가 이미 가져갔다면 그동안 처리할 수 있는 다른 작업을 찾아 돕다가 할 일이 없을 때만 기다립니다. 그래서 같은 모양의 재귀를 작업자가 하나뿐인 풀에서 실행해도 끝까지 진행되며, 다음 글에서 이를 직접 확인합니다.

더 깊이: 풀이 아는 기다림과 모르는 기다림

ForkJoinPool의 Javadoc은 일부 작업이 다른 작업의 join을 기다리느라 멈춰 있더라도, 풀이 내부 작업자 스레드를 추가하거나 멈추거나 다시 깨워서 활동하는 스레드 수를 충분히 유지하려고 한다고 설명합니다. join()은 풀이 직접 관리하는 기다림이라서, 기다리는 작업자에게 다른 일을 시키거나 필요하면 예비 스레드를 투입할 수 있다는 뜻입니다. 하지만 같은 문서는 블로킹 I/O나 풀이 관리하지 않는 다른 동기화 앞에서는 이런 조정을 보장하지 않는다고 덧붙입니다. Fork/Join 작업 안에서 다른 실행기의 Future.get()이나 소켓 읽기처럼 풀이 모르는 방식으로 기다리면, 앞의 고정 스레드 풀과 같은 이유로 작업자가 묶일 수 있습니다. 그래서 ForkJoinTask 문서는 나눌 수 있는 작업 안에서 블로킹 I/O를 하지 말라고 권하고, 꼭 기다려야 하는 경우를 위해 풀에 기다림을 알리는 ForkJoinPool.ManagedBlocker를 제공합니다.

Fork/Join 패턴: 나누고, 한쪽은 맡기고, 결과를 합친다

Fork/Join으로 작성하는 계산은 대개 다음 모양을 따릅니다. 실제 Java 코드는 다음 글에서 작성하고, 여기서는 흐름만 의사 코드로 보겠습니다.

compute(작업):
    작업이 임계값 이하로 작으면
        직접 계산해서 결과를 돌려준다
    그렇지 않으면
        겹치지 않는 두 하위 작업 left, right로 나눈다
        left.fork()                    // 다른 작업자가 가져갈 수 있게 맡긴다
        rightResult = right.compute()  // 나머지 절반은 지금 스레드가 계산한다
        leftResult = left.join()       // 맡긴 절반이 끝나기를 기다린다
        leftResult와 rightResult를 합쳐 돌려준다

포크(fork)는 하위 작업을 대기열에 넣어 다른 작업자가 가져갈 수 있게 하는 것이고, 조인(join)은 그 하위 작업이 끝나기를 기다려 결과를 받는 것입니다. 그 사이에 현재 스레드는 놀지 않고 나머지 절반을 직접 계산합니다. 이름 그대로 작업이 갈라졌다가(fork) 다시 합쳐지는(join) 흐름이며, 재귀가 깊어질수록 갈라진 작업이 여러 작업자에게 퍼져 동시에 계산됩니다. 정리하면 큰 작업은 나누고, 충분히 작은 작업은 직접 계산하고, 하위 결과는 기다렸다가 합치는 세 단계의 반복입니다.

두 하위 작업을 모두 fork()할 수도 있지만, 그러면 현재 스레드는 할 일 없이 두 결과를 기다리게 됩니다. 한쪽을 직접 계산하면 작업 하나를 대기열에 넣고 꺼내는 비용을 아끼고, 현재 스레드도 계산에 참여합니다. 다만 앞에서 본 것처럼 Fork/Join에서는 join()에서 기다리는 작업자도 다른 일을 돕기 때문에, 둘 다 fork()하는 방식이 언제나 느린 것은 아닙니다. 어느 쪽이 빠른지는 계산의 크기와 나누는 방식에 따라 달라지므로 측정해서 정합니다.

이 패턴이 맞는 답을 내려면 세 가지 조건이 필요합니다. 첫째, 나눈 하위 작업들이 입력 전체를 빠짐없이, 겹치지 않게 덮어야 합니다. 둘째, 리프 작업(더 나누지 않고 직접 계산하는 작업)끼리 공유 변수에 기대지 않고 독립적으로 계산할 수 있어야 합니다. 셋째, 결과를 합치는 연산이 순차 계산과 같은 답을 내야 합니다. 합계라면 덧셈이 그런 연산입니다. 프레임워크는 이 조건들을 검사해 주지 않으며, 나누는 코드와 언제 그만 나눌지도 프로그래머가 직접 정합니다.

언제 그만 나눌지를 정하는 기준을 임계값(threshold)이라고 합니다. 작업 크기가 임계값 이하이면 더 나누지 않고 직접 계산합니다. 임계값이 너무 크면 하위 작업이 몇 개 생기지 않아 일이 없는 작업자가 생기고, 너무 작으면 작업 객체를 만들고 맡기고 기다리는 비용이 정작 계산보다 커집니다. 임계값에 따라 실행 시간이 얼마나 달라지는지는 다음 글에서 측정해 봅니다. 작업자마다 자기 대기열을 두고, 일이 떨어진 작업자가 다른 작업자의 대기열에서 작업을 가져오는 방식(작업 훔치기)은 작업 훔치기와 병렬 스트림에서 자세히 다룹니다.

어떤 방법을 고를까

지금까지 본 방법은 각자 잘 맞는 일이 다릅니다. 다음 기준에서 출발하되, 실제 선택은 측정으로 확인합니다.

• 작은 컬렉션을 가볍게 변환하는 일은 순차 반복문이나 순차 Stream으로 충분합니다. 나누고 합치는 비용을 감당할 만큼 계산이 크지 않기 때문입니다.
• 성격이 서로 다른 독립 작업 몇 개, 예를 들어 서로 다른 두 서비스를 호출하는 일은 ExecutorService와 Future로 표현하는 편이 명확합니다.
• 크고, 재귀적으로 나눌 수 있고, 주로 CPU를 쓰는 계산에는 Fork/Join이 알맞습니다.
• 파일이나 네트워크처럼 기다리는 시간이 긴 작업은 Fork/Join 풀, 특히 JVM 전체가 함께 쓰는 공용 풀(common pool)에 넣지 말고 별도의 실행기에서 실행합니다. 자세한 기준은 서버에서 공용 풀과 전용 풀 나눠 쓰기에서 다룹니다.
• 어느 방법을 쓰든 공유 변수를 함께 고치는 대신 작업마다 지역 결과를 만들고 마지막에 명시적으로 합칩니다.

정리

• 병렬 계산은 나누기, 동시에 실행하기, 합치기의 세 단계로 이뤄집니다. 병렬 버전은 먼저 순차 기준과 같은 답을 내야 합니다.
• start()는 스레드가 끝나기를 기다려 주지 않습니다. 결과는 Thread.join(), Future.get(), ForkJoinTask.join()처럼 완료를 기다리는 호출 뒤에 읽습니다.
• 작업마다 부분 결과를 따로 두고 마지막에 한 스레드에서 합치면 공유 변수와 락이 필요 없습니다.
• 일반 스레드 풀에서 작업이 하위 작업을 맡기고 get()으로 기다리면, 작업자가 모두 기다리느라 멈출 수 있습니다. Fork/Join은 기다리는 작업자가 남은 작업을 대신 처리하게 해서 재귀 분할이 멈추지 않게 합니다.
• Fork/Join 패턴은 '임계값 이하면 직접 계산하고, 아니면 둘로 나눠 한쪽은 fork(), 다른 쪽은 compute(), 마지막에 join()'으로 요약됩니다.

연습 문제

1. 1부터 12까지를 네 구간으로 나눠 각 구간의 제곱합을 Callable로 계산하세요. 공유 누산기를 쓰지 말고, 모든 Future가 끝난 뒤 합계 650을 출력해야 합니다.
2. 1번 풀이에서 마지막 구간의 시작만 9로 바꾸면 합계는 얼마가 되나요? 그와 별개로, 마지막 구간은 그대로 두고 세 번째 구간의 끝만 8로 바꾸면 얼마가 되나요? 합계를 확인하는 것 말고 어떤 조건을 확인해야 이런 실수를 막을 수 있나요?
3. RecursiveSubmitStall의 리프 조건을 to - from + 1 <= 4로 바꾸면 작업자 2개로 끝까지 실행될까요? 작업자가 1개라면 어떨까요? 앞의 단계 표처럼 따라가며 설명하세요.

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public final class PooledSquareSumSolution {
    private static Callable<Integer> squares(int from, int to) {
        return () -> {
            int subtotal = 0;
            for (int value = from; value <= to; value++) {
                subtotal += value * value;
            }
            return subtotal;
        };
    }

    public static void main(String[] args) throws Exception {
        try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
            List<Future<Integer>> parts =
                    List.of(
                            pool.submit(squares(1, 3)),
                            pool.submit(squares(4, 6)),
                            pool.submit(squares(7, 9)),
                            pool.submit(squares(10, 12)));
            int total = 0;
            for (Future<Integer> part : parts) {
                total += part.get();
            }
            System.out.println("square-sum=" + total);
        }
    }
}

square-sum=650

다음 글

RecursiveTask와 RecursiveAction으로 작업 나누기에서는 이 글의 패턴을 RecursiveTask로 직접 구현합니다. 범위를 겹치게 나눴을 때 생기는 오답, 임계값에 따라 달라지는 작업 수와 실행 시간, 결과가 없는 작업에 쓰는 RecursiveAction도 함께 살펴봅니다.