Fork/Join 문제 분할

병렬 처리는 parallel()이라는 한 단어가 아니라 작업을 독립 단위로 나누고, 실행 자원을 배정하고, 모든 부분 결과를 합치는 설계입니다. 먼저 순차 실행의 기준값을 확보한 뒤 직접 스레드와 풀을 사용해 보면 Fork/Join이 해결하는 문제가 보입니다. CPU 코어가 여러 개여도 작업 사이의 의존성이나 합산 규칙이 틀리면 빠르기는커녕 오답을 더 빨리 만듭니다.

작업자 완료 전 결과 오판

아래 예제는 두 작업자가 시작됐다는 사실만 확인하고 결과를 합칩니다. release를 열기 전에는 작업자가 값을 쓸 수 없도록 만들어 오답을 재현하며, 출력은 wrong-sum=0입니다. start()는 완료를 보장하지 않으며, 다른 스레드가 쓴 값을 안전하게 읽게 하는 완료 경계도 아닙니다.

lab/UnjoinedWorkerResultBug.java

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();
    }
}

문제는 result에 volatile을 붙이지 않은 데 있지 않습니다. 값 하나의 가시성을 높여도 계산이 끝났다는 시점을 모르면 완성되지 않은 중간 상태를 읽습니다. 원칙은 완료를 기다린 다음 결과를 모은다입니다. Thread.join(), Future.get(), ForkJoinTask.join()은 각각 이러한 완료 경계를 제공합니다.

순차 기준선 실행

병렬 후보를 평가하기 전 동일한 입력과 계산으로 순차 기준을 만듭니다. 총합, 입력 순서, 예외 정책을 이 단계에서 고정해야 병렬 구현의 정확성을 비교할 수 있습니다.

src/SequentialHeavyJobBaseline.java

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);
    }
}

출력은 360입니다. 실제 고비용 작업이 1초의 sleep을 포함한다고 해도 이 교재의 검증 예제는 불필요하게 기다리지 않습니다. 성능 실험에서는 계산량을 별도 매개변수로 두고, 정확성 예제에서는 결과가 결정적인 작은 입력을 사용합니다.

직접 스레드의 분할·`join` 책임

범위 1~~8을 1~~4와 5~8로 나눠 서로 독립적으로 계산합니다. 각 작업은 자기 필드에만 쓰므로 공유 가변 누산기가 없습니다. main은 두 스레드를 join한 뒤 결과를 더합니다.

src/JoinedRangeWorkers.java

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);
    }
}

join()은 단순히 기다릴 뿐 아니라 작업자가 완료하기 전에 수행한 동작이 join 이후의 main 스레드에 보이는 happens-before 관계를 만듭니다. 반대로 여러 작업자가 같은 int sum에 +=를 수행하면 읽기·수정·쓰기 과정이 겹쳐 갱신이 손실될 수 있습니다. 부분 결과를 격리하고 마지막에 합치는 구조가 동기화 범위를 작게 유지합니다.

스레드 풀: 작업과 작업자 생명주기 분리

작업마다 플랫폼 스레드를 직접 만드는 대신 ExecutorService에 Callable<Integer>를 제출하면 작업자 재사용과 대기 큐를 풀이 관리합니다. 호출자는 Future에서 결과를 가져와 join 단계로 사용합니다.

src/PooledRangeResults.java

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);
        }
    }
}

Future.get()은 완료까지 기다리고 작업 예외를 ExecutionException으로 전달합니다. 운영 코드에서는 인터럽트를 삼키지 않고 상위 취소 정책으로 연결해야 합니다. 풀 크기는 작업 개수와 같게 고정하는 공식이 아니며 CPU 중심 여부, 블로킹 비율, 동시 요청 수를 함께 봐야 합니다.

Fork·Execute·Join 패턴 해석

Fork/Join은 큰 작업을 임계값보다 작아질 때까지 재귀적으로 분할합니다. 리프는 직접 계산하고 부모는 두 결과를 결합합니다. 분할 가능한가, 리프 계산이 독립적인가, 결합 연산이 올바른가가 핵심 규칙입니다.

분할(fork): 범위를 겹치지 않는 하위 범위로 나눈다.
실행: 임계값 이하의 리프를 현재 작업자가 계산한다.
결합(join): 하위 결과가 완료될 때까지 기다린 뒤 결합한다.

한쪽을 fork()한 뒤 다른 쪽을 현재 스레드가 compute()하고 마지막에 join하면 호출자도 일을 합니다. 양쪽을 모두 fork하고 즉시 기다리면 큐 이동과 대기만 늘 수 있습니다. 임계값이 너무 크면 병렬성이 부족하고 너무 작으면 작업 생성·스케줄링·join 비용이 계산보다 커집니다.

선택 기준

작은 컬렉션의 값싼 변환은 순차 반복문이나 순차 Stream을 기준으로 삼습니다.
서로 성격이 다른 소수의 독립 작업은 ExecutorService와 Future로 표현하는 편이 명확합니다.
균일하고 재귀 분할 가능한 CPU 계산은 Fork/Join 후보입니다.
요청 처리 안의 블로킹 I/O를 공용 풀에 섞는 것은 별도 자원 정책이 필요합니다.
어느 방식이든 공유 가변 상태 대신 작업-지역 결과와 명시적 결합을 우선합니다.

직접 만든 스레드 수를 CPU 코어 수에 맞추면 항상 가장 빠른가요?

아닙니다. CPU 중심 계산은 코어 수를 출발점으로 삼을 수 있지만 분할 비용, 메모리 대역폭, 캐시 지역성, 다른 프로세스와 애플리케이션 작업의 경쟁이 결과를 바꿉니다. 블로킹 작업은 대기 비율을 고려하되 무제한 스레드가 답은 아닙니다. 동일한 운영 조건에서 예열과 반복 측정을 거쳐야 합니다.

연습 문제

1~12를 네 구간으로 나눠 각 구간이 제곱합을 반환하게 하세요. 공유 누산기를 쓰지 말고 모든 Future가 끝난 뒤 650을 출력해야 합니다.

해설 보기

exercise/PooledSquareSumSolution.java

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);
        }
    }
}

종료 기준은 출력값만이 아닙니다. 각 작업이 자신의 지역 부분합만 변경하고, 호출자가 get() 이후에 결과를 결합하며, 풀이 닫혀 프로세스가 끝나는지까지 확인합니다.