작업 훔치기와 병렬 스트림

앞 글의 RecursiveTask는 범위를 둘로 나눈 뒤 왼쪽은 fork()로 넘기고 오른쪽은 직접 compute()했습니다. 그런데 fork()로 넘긴 작업은 누가, 언제 실행할까요? 분할이 고르지 않으면 어떤 스레드는 일찍 끝나서 놀고, 어떤 스레드는 큰 덩어리를 붙잡고 오래 일하게 됩니다. 이 불균형은 누가 메울까요?

ForkJoinPool은 이 문제를 작업 훔치기(work stealing)로 풉니다. 이 글에서는 먼저 작업 훔치기가 어떻게 동작하는지 실행해서 확인합니다. 그다음 병렬 스트림이 바로 이 구조 위에서 돈다는 것을 보고, 그 때문에 생기는 주의점을 차례로 다룹니다. 원소를 처리하는 순서와 결과의 순서가 다르다는 것, 콜백에서 바깥 변수를 고치면 값이 사라진다는 것, 그리고 병렬 수집에 쓰는 Collector가 지켜야 할 조건입니다.

작업자마다 덱을 따로 두는 이유

가장 단순한 설계는 큐 하나에 작업을 모두 넣고 모든 작업자가 거기서 꺼내 가는 것입니다. 하지만 재귀 분할에서는 작은 작업이 아주 많이 생깁니다. 작업자들이 작업을 넣고 뺄 때마다 하나뿐인 큐를 두고 다투면, 계산보다 줄을 서는 데 시간을 더 쓰게 됩니다.

그래서 ForkJoinPool은 작업자 스레드마다 덱(deque)을 하나씩 줍니다. 덱은 양쪽 끝에서 넣고 뺄 수 있는 큐입니다. 작업자가 fork()한 작업은 자기 덱의 한쪽 끝에 쌓이고, 작업자는 일이 필요할 때 같은 끝에서 가장 최근에 넣은 작업부터 꺼냅니다. 스택처럼 나중에 넣은 것을 먼저 꺼내는(LIFO) 방식이고, API 문서는 이를 기본 모드인 '지역 스택 방식(locally stack-based mode)'이라고 부릅니다. 자기 덱이 빈 작업자는 다른 작업자의 덱에서 일을 가져오는데, 이때는 반대쪽 끝에 있는 가장 오래된 작업을 가져갑니다. 이것이 작업 훔치기입니다.

구체적인 예로 보겠습니다. 반열린 범위 [1, 65)의 합을 구하되, 범위를 절반이 아니라 1/3 지점에서 나누는 작업이 있다고 합시다. 작업자 A가 루트 작업을 받으면, 처음 몇 단계 동안 A의 덱은 다음처럼 쌓입니다.

이때 놀고 있던 작업자 B가 A의 덱에서 일을 훔친다면 가장 오래된 [1, 22)를 가져갑니다. 재귀 분할에서는 오래된 작업일수록 위쪽 단계에서 나온 큰 덩어리입니다. [1, 22)에는 원소가 21개 있어서 B는 이것을 다시 나눠 한동안 혼자 일할 수 있고, 그만큼 훔치러 오는 횟수가 줄어듭니다. 반대로 A는 방금 만든 작은 작업부터 처리하는데, 그 작업이 쓰는 데이터는 아직 캐시에 남아 있을 가능성이 큽니다. 무엇보다 주인과 도둑이 덱의 서로 다른 끝을 쓰므로 둘이 부딪치는 일이 드뭅니다.

생성자에 asyncMode를 true로 넘기면, 결과를 join하지 않는 작업에 대해서는 자기 덱에서도 먼저 넣은 작업부터(FIFO) 처리합니다. 결과를 기다리지 않는 이벤트성 작업만 처리하는 풀을 위한 설정이고, 이 글처럼 나누고 합치는 계산에는 기본 모드를 씁니다. 덧붙여, 도둑이 덱의 반대쪽 끝에서 가져간다는 것은 구현의 설계입니다. API는 어느 작업자가 어느 작업을 언제 실행할지 정하지 않습니다.

작업 훔치기를 실행으로 확인하기

다음 프로그램은 앞의 표처럼 범위를 1/3 지점에서 나눠 1부터 64까지의 합을 구합니다. 원소가 4개 이하인 작업은 더 나누지 않고 직접 더하는데, 이런 작업을 리프(leaf)라고 부릅니다. 리프를 계산할 때마다 그 리프를 실행한 스레드의 이름으로 개수를 세고, 마지막에는 풀이 집계한 훔치기 횟수 getStealCount()도 출력합니다.

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.atomic.LongAdder;

public final class WorkStealingDistribution {
    private static final class SumTask extends RecursiveTask<Long> {
        private final int from;
        private final int to;
        private final Map<String, LongAdder> leafCounts;

        private SumTask(int from, int to, Map<String, LongAdder> leafCounts) {
            this.from = from;
            this.to = to;
            this.leafCounts = leafCounts;
        }

        @Override
        protected Long compute() {
            if (to - from <= 4) {
                String worker = Thread.currentThread().getName();
                leafCounts.computeIfAbsent(worker, key -> new LongAdder()).increment();
                long sum = 0;
                for (int value = from; value < to; value++) {
                    sum += value;
                }
                return sum;
            }
            int middle = from + Math.max(1, (to - from) / 3); // 1/3 지점에서 고르지 않게 나눈다
            SumTask left = new SumTask(from, middle, leafCounts);
            SumTask right = new SumTask(middle, to, leafCounts);
            left.fork();
            return right.compute() + left.join();
        }
    }

    public static void main(String[] args) {
        Map<String, LongAdder> leafCounts = new ConcurrentHashMap<>();
        try (ForkJoinPool pool = new ForkJoinPool(4)) {
            long sum = pool.invoke(new SumTask(1, 65, leafCounts));
            System.out.println("sum=" + sum);
            System.out.println("steals=" + pool.getStealCount());
        }
        leafCounts.forEach((worker, leaves) -> System.out.println(worker + "=" + leaves.sum()));
    }
}

leafCounts는 계측용입니다. 여러 작업자가 동시에 갱신하므로 ConcurrentHashMap과 LongAdder를 썼지만, 정작 합계는 공유 변수가 아니라 각 작업의 반환값으로 모읍니다.

실행 결과(예):

sum=2080
steals=4
ForkJoinPool-1-worker-1=20
ForkJoinPool-1-worker-4=1
ForkJoinPool-1-worker-2=1
ForkJoinPool-1-worker-3=1

sum=2080은 1부터 64까지의 합(64 × 65 ÷ 2)이므로 몇 번을 실행해도 같습니다. 나머지 줄은 실행할 때마다 달라집니다. 프로세서가 2개인 환경에서 여덟 번 실행했더니 매번 작업자 하나가 리프를 18개에서 20개까지 맡고 나머지는 1개에서 4개씩 맡았으며, 어느 작업자가 많이 맡을지도 그때그때 달랐습니다. steals는 4에서 18 사이였습니다. 그래도 작업자별 개수를 더하면 언제나 23입니다. 리프의 개수는 범위를 나누는 규칙만으로 정해지고, 어느 스레드가 실행하느냐와는 관계가 없기 때문입니다.

일이 한 작업자에게 쏠린 것도 작업 훔치기의 성질을 보여 줍니다. 훔치기는 놀고 있는 작업자와 덱에서 기다리는 작업이 동시에 있을 때만 일어납니다. 이 계산은 아주 작아서, 다른 작업자들이 깨어나 덱을 살피는 사이에 루트를 받은 작업자가 대부분을 끝내 버립니다. 게다가 현재 구현에서는 left.join()에 도착했을 때 left가 아직 도둑맞지 않고 자기 덱에 남아 있으면, 그 작업자가 직접 꺼내 실행합니다. 즉 작업 훔치기는 일을 고르게 나눠 주는 배분 규칙이 아니라, 놀게 된 작업자가 스스로 일을 찾아가는 방식입니다.

작업자별 숫자는 실행한 리프의 수이지 훔친 횟수가 아닙니다. 훔친 횟수는 getStealCount()가 알려 주는데, API 문서는 이 값을 '작업을 제출한 스레드가 아닌 다른 스레드가 실행을 마친 작업 수의 추정치'라고 설명합니다. 리프가 아닌 중간 작업도 훔칠 수 있고, 중간 작업 하나를 훔치면 그 아래의 리프들도 함께 따라가므로 두 숫자는 맞아떨어지지 않습니다. 이런 분포와 횟수는 스케줄러와 머신에 따라 달라지므로 테스트에서 특정 값을 기대해서는 안 됩니다. 테스트할 것은 언제나 같아야 하는 값, 여기서는 2080입니다. 물론 합계가 맞았다고 분할에 겹치거나 빠진 범위가 없다는 증명이 되지는 않으므로, 분할 코드는 앞 글에서처럼 반열린 범위 규칙으로 따로 확인해야 합니다.

병렬 스트림은 같은 구조 위에서 돈다

parallelStream()으로 만들거나 parallel()을 붙인 스트림도 내부에서는 같은 일을 합니다. 스트림 라이브러리는 원본을 Spliterator(나눌 수 있는 반복자)로 조각내고, 조각마다 Fork/Join 작업을 만들어 작업자들에게 맡긴 뒤, 조각별 결과를 합칩니다. 앞 글에서 직접 짠 분할과 fork, join, 결합을 라이브러리가 대신 해 주는 셈입니다.

어느 풀에서 실행되는지는 Stream API가 정하지 않습니다. 현재 JDK 구현은 보통 ForkJoinPool.commonPool(), 즉 프로세스 전체가 함께 쓰는 공용 풀을 쓰고, ForkJoinPool의 작업자 스레드 안에서 최종 연산을 호출하면 그 풀을 씁니다. 이 차이와 공용 풀을 쓸 때의 주의점은 다음 글에서 다룹니다. 여기서 기억할 것은 앞의 예제에서 본 것처럼 어떤 원소를 어느 스레드가 언제 처리할지 정해져 있지 않다는 점입니다. 아래의 주의점은 모두 이 사실에서 나옵니다.

처리 순서와 결과 순서는 다르다

스트림의 원본이 원소를 내놓는 순서를 만남 순서(encounter order)라고 합니다. List는 인덱스 순서가 만남 순서이며, 원소가 크기순으로 정렬돼 있다는 뜻은 아닙니다. 다음 프로그램은 만남 순서가 5, 1, 4, 2, 3인 목록 하나로 병렬 스트림 세 개를 따로 만들고, 최종 연산마다 이 순서를 어떻게 다루는지 비교합니다.

import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

public final class ParallelEncounterOrder {
    public static void main(String[] args) {
        List<Integer> source = List.of(5, 1, 4, 2, 3);

        List<Integer> collected = source.parallelStream().map(value -> value * 10).toList();

        ConcurrentLinkedQueue<Integer> insertionOrder = new ConcurrentLinkedQueue<>();
        source.parallelStream().forEach(insertionOrder::add);

        StringBuilder orderedText = new StringBuilder();
        source.parallelStream().forEachOrdered(value -> orderedText.append(value).append(' '));

        System.out.println("collected=" + collected);
        System.out.println("forEach-size=" + insertionOrder.size());
        System.out.println("forEach-order=" + insertionOrder);
        System.out.println("ordered=" + orderedText.toString().trim());
    }
}

세 스트림은 같은 원본에서 따로 만들었으므로 서로의 실행에 영향을 주지 않습니다. 병렬 실행에서 일이 처리되는 순서는 실행마다 달라질 수 있으므로, 아래는 한 번 실행한 예입니다.

collected=[50, 10, 40, 20, 30]
forEach-size=5
forEach-order=[1, 5, 4, 3, 2]
ordered=5 1 4 2 3

첫 줄의 toList()는 각 값에 10을 곱한 결과를 원본과 같은 순서로 돌려줍니다. map은 여러 스레드에서 아무 순서로나 실행됐을 수 있지만, 조각별 결과를 합칠 때 왼쪽 조각의 결과를 오른쪽 조각의 결과 앞에 두기 때문에 최종 목록은 만남 순서를 따릅니다. 그래서 이 줄은 언제 실행해도 같습니다.

둘째와 셋째 줄은 forEach가 큐에 넣은 값입니다. 다섯 값이 모두 들어갔다는 것은 언제나 같지만, 들어간 순서는 실행마다 다릅니다. 이 환경에서 여덟 번 실행하자 [1, 5, 4, 3, 2], [4, 3, 2, 1, 5], [4, 3, 2, 5, 1]이 섞여 나왔습니다. API 문서는 forEach의 동작이 명시적으로 비결정적이라고 밝힙니다. 병렬 파이프라인에서는 만남 순서를 지키지 않으며, 각 원소에 대한 동작을 라이브러리가 고른 아무 시점에 아무 스레드에서나 실행할 수 있습니다. 순서를 지키려면 병렬의 이점을 포기해야 하기 때문입니다. 이 큐의 순서는 add가 호출된 순서일 뿐, 각 작업자가 일을 시작하거나 끝낸 시각을 잰 기록도 아니라는 점도 기억해 두세요.

마지막 줄의 forEachOrdered는 원소를 만남 순서대로 하나씩 처리하므로 5 1 4 2 3이 나옵니다. forEachOrdered의 API 문서에 따르면 한 원소에 대한 동작은 다음 원소에 대한 동작보다 먼저 일어나고(happens-before), 그 결과가 다음 동작에 보입니다. 다만 각 동작이 어느 스레드에서 실행될지는 여전히 라이브러리가 고릅니다. 스레드 안전하지 않은 StringBuilder를 여기서 써도 되는 것은 이 보장 덕분이며, 같은 코드를 forEach로 바꾸면 데이터 경합이 됩니다. 그 대신 앞 원소의 동작이 끝나야 다음 원소로 넘어갈 수 있으므로, 이 단계에서는 병렬 처리의 이점이 대부분 사라집니다.

순서가 필요 없다면 unordered()로 알려 주기

만남 순서를 지키는 데는 비용이 듭니다. 예를 들어 limit(3)은 아무 원소 세 개가 아니라 만남 순서상 처음 세 개를 돌려줘야 합니다. 그래서 병렬로 실행하면 앞쪽 조각의 결과가 나올 때까지 뒤쪽 조각이 만든 결과를 붙잡아 두거나 버려야 합니다. distinct()도 순서가 있는 스트림에서는 같은 값 가운데 처음 나온 것을 남겨야 하므로 비슷한 조정이 필요합니다.

unordered()는 이 의무를 풀어 줍니다. 원소를 뒤섞는 연산이 아니라, 이후 단계에서 만남 순서를 지키지 않아도 된다고 라이브러리에 알리는 표시입니다. 그러면 list.parallelStream().unordered().limit(3)은 앞의 세 개 대신 아무 세 개나 돌려줄 수 있고, 그만큼 빨라질 여지가 생깁니다. API 문서도 순서가 있는 병렬 파이프라인에서 limit()가 꽤 비쌀 수 있다며, 상황이 허락하면 unordered()로 순서 제약을 없애 보라고 권합니다. 반대로 결과의 순서나 어떤 원소가 뽑혔는지가 호출한 쪽에 의미가 있다면 쓰지 않아야 합니다. 뒤에서 볼 동시 수집도 순서 제약이 없을 때 일어납니다.

콜백에서 바깥 변수를 바꾸면 값이 사라진다

병렬 스트림에서 가장 흔한 실수는 콜백 안에서 바깥의 변수를 고치는 것입니다. count[0]++ 같은 증가는 한 번에 일어나는 것처럼 보이지만, 실제로는 값을 읽고, 1을 더하고, 다시 쓰는 세 단계입니다. 두 스레드가 같은 값을 읽은 뒤 각자 쓰면 한쪽이 쓴 값을 다른 쪽이 덮어써서 증가 하나가 사라집니다. 이를 갱신 손실(lost update)이라고 합니다.

다음 프로그램은 이 상황을 매번 재현하려고 배리어를 끼워 넣었습니다. CyclicBarrier(2)는 두 스레드가 모두 await()에 도착할 때까지 먼저 온 쪽을 기다리게 합니다. 그래서 두 콜백은 반드시 둘 다 값을 읽은 뒤에야 쓰기 단계로 넘어갑니다.

import java.util.List;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ForkJoinPool;

public final class ParallelSharedCounterBug {
    public static void main(String[] args) throws Exception {
        int[] value = {0};
        CyclicBarrier bothRead = new CyclicBarrier(2);

        try (ForkJoinPool pool = new ForkJoinPool(2)) {
            pool.submit(() -> List.of(10, 20).parallelStream().forEach(ignored -> {
                int observed = value[0];       // 1. 읽기
                await(bothRead);               // 두 콜백이 모두 읽을 때까지 기다린다
                value[0] = observed + 1;       // 2. 증가한 값 쓰기
            })).get();
        }

        System.out.println("wrong-count=" + value[0]);
    }

    private static void await(CyclicBarrier barrier) {
        try {
            barrier.await();
        } catch (InterruptedException exception) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(exception);
        } catch (BrokenBarrierException exception) {
            throw new IllegalStateException(exception);
        }
    }
}

출력은 두 콜백이 모두 끝난 뒤 value[0]에 남은 값입니다. submit(...).get()은 작업이 끝날 때까지 기다리므로, 이 줄이 찍힐 때는 두 쓰기가 모두 끝난 상태입니다.

wrong-count=1

원소가 두 개이니 2가 나와야 하지만 1이 나옵니다. 두 콜백을 A, B라고 하고 상태를 따라가 보면 이유가 보입니다.

두 콜백이 각자 읽어 둔 0에 1을 더해 같은 자리에 1을 썼으므로, 어느 쪽이 먼저 쓰든 결과는 1입니다. 배리어는 이 순서를 강제로 만드는 실험 장치일 뿐 실제 코드에 넣을 것은 아닙니다. 또 이 예제는 두 원소가 서로 다른 작업자에게 나뉘어 동시에 실행돼야 끝납니다. 한 스레드가 두 원소를 차례로 처리하면 첫 콜백이 배리어에서 영원히 기다리기 때문에, 작업자가 둘인 전용 풀 안에서 실행했습니다.

배리어가 없으면 어떨까요? 이 환경에서 IntStream.range(0, n).parallel().forEach(i -> count[0]++)를 n을 바꿔 가며 다섯 번씩 실행해 봤습니다. n이 100만일 때는 다섯 번 모두 정확했고, 1000만일 때는 한 번만 7500000이 나왔으며, 1억일 때는 3200만에서 7200만 사이의 값이 매번 다르게 나왔습니다. 이처럼 경쟁은 타이밍에 따라 일어나기도 하고 일어나지 않기도 합니다. 그래서 작은 입력으로 테스트를 통과했다고 안전한 코드라고 할 수 없고, 틀릴 때 얼마나 틀릴지도 예측할 수 없습니다.

공유 변수 대신 조각별 결과를 합치기

해결책은 콜백 밖의 가변 상태를 여러 작업자가 함께 고치지 않는 것입니다. 개수가 필요하면 count(), 합계가 필요하면 sum()이나 결합 법칙을 만족하는 reduce, 여러 값을 모아야 하면 collect를 씁니다. 예를 들어 List.of(10, 20).parallelStream().count()는 언제나 2를 돌려줍니다. 이런 연산은 조각마다 자기 결과를 따로 만들고 마지막에 합치므로, 여러 스레드가 같은 변수를 건드릴 일이 없습니다. 앞 글의 RecursiveTask가 공유 누산기 대신 반환값으로 합계를 모은 것과 같은 구조입니다.

카운터를 AtomicInteger로 바꾸면 데이터 경합은 사라지고 값도 맞게 나옵니다. 하지만 모든 원소가 변수 하나를 두고 줄을 서므로 병렬로 처리하는 의미가 줄어들고, 결과를 콜백의 부수 효과로 모으는 구조도 그대로 남습니다. LongAdder는 갱신을 여러 칸에 나눠 경쟁을 줄이지만, 이것도 축약 연산을 쓸 수 없을 때 고려할 차선책입니다.

병렬 수집과 Collector의 특성

collect에 넘기는 Collector는 Collectors로 List·Map 만들고 그룹으로 묶기에서 본 네 함수로 이뤄집니다. 공급자(supplier)가 빈 결과 컨테이너를 만들고, 누산 함수(accumulator)가 원소를 하나씩 담고, 결합 함수(combiner)가 두 컨테이너를 합치고, 마무리 함수(finisher)가 최종 타입으로 바꿉니다. 병렬로 수집할 때는 보통 조각마다 공급자로 컨테이너를 따로 만들어 채운 뒤, 결합 함수로 둘씩 합쳐 올라갑니다. 컨테이너가 조각마다 따로 있으므로 누산 함수는 스레드 안전할 필요가 없습니다.

Collector는 여기에 특성(characteristics)이라는 추가 정보를 붙일 수 있습니다. 라이브러리는 이 정보를 믿고 수집 방법을 고르므로, 실제 구현과 맞지 않는 특성을 붙이면 결과가 틀어집니다.

Stream.collect의 API 문서에 따르면 스트림이 병렬이고, 수집기가 CONCURRENT이며, 스트림에 순서가 없거나 수집기가 UNORDERED이면 동시 수집(concurrent reduction)이 일어납니다. 이때는 컨테이너를 하나만 만들어 모든 스레드가 함께 채웁니다. 그러니 일반 HashMap을 쓰는 수집기에 CONCURRENT만 붙이면 데이터 경합을 라이브러리 안으로 숨기는 셈입니다.

다음 수집기는 단어별 개수를 셉니다. 컨테이너는 ConcurrentHashMap이고 개수는 LongAdder에 쌓으므로 동시에 갱신해도 안전합니다. 그래서 CONCURRENT를 선언할 수 있고, 개수는 입력 순서와 상관없으므로 UNORDERED도 선언합니다. 마무리 함수는 LongAdder의 합계를 꺼내 변경할 수 없는 Map<String, Long>으로 옮깁니다.

import java.util.EnumSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public final class ConcurrentFrequencyCollector {
    private static final class Frequencies
            implements Collector<String, ConcurrentHashMap<String, LongAdder>, Map<String, Long>> {

        @Override
        public Supplier<ConcurrentHashMap<String, LongAdder>> supplier() {
            return ConcurrentHashMap::new;
        }

        @Override
        public BiConsumer<ConcurrentHashMap<String, LongAdder>, String> accumulator() {
            return (map, word) -> map.computeIfAbsent(word, key -> new LongAdder()).increment();
        }

        @Override
        public BinaryOperator<ConcurrentHashMap<String, LongAdder>> combiner() {
            return (left, right) -> {
                right.forEach((word, count) ->
                        left.computeIfAbsent(word, key -> new LongAdder()).add(count.sum()));
                return left;
            };
        }

        @Override
        public Function<ConcurrentHashMap<String, LongAdder>, Map<String, Long>> finisher() {
            return mutable -> mutable.entrySet().stream()
                    .collect(Collectors.toUnmodifiableMap(
                            Map.Entry::getKey, entry -> entry.getValue().sum()));
        }

        @Override
        public Set<Characteristics> characteristics() {
            return EnumSet.of(Characteristics.CONCURRENT, Characteristics.UNORDERED);
        }
    }

    public static void main(String[] args) {
        Map<String, Long> counts = Stream.of("java", "stream", "java", "fork", "stream", "java")
                .parallel()
                .collect(new Frequencies());
        System.out.println(counts);
    }
}

마무리 함수는 Collectors.toUnmodifiableMap으로 결과를 만드는데, 이 Map은 반복 순서가 정해져 있지 않습니다. 그래서 단어가 찍히는 순서는 실행마다 다를 수 있고, 아래는 한 번 실행한 예입니다.

{java=3, stream=2, fork=1}

공급자, 누산 함수, 마무리 함수를 차례로 거치며 상태가 어떻게 바뀌는지는 아래 단계별 그림으로 확인할 수 있습니다.

개수는 언제나 java=3, stream=2, fork=1이지만 키가 출력되는 순서는 실행할 때마다 바뀝니다. 이 환경에서 여덟 번 실행하자 {fork=1, java=3, stream=2}, {stream=2, fork=1, java=3}을 비롯해 여러 순서가 나왔습니다. 이것은 병렬 처리 때문이 아닙니다. toUnmodifiableMap이 만드는 변경 불가 Map은 순회 순서를 정하지 않으며, 현재 구현은 JVM을 실행할 때마다 순서가 달라지기도 합니다. 그러니 결과를 검사할 때는 출력된 문자열이 아니라 키별 값을 비교해야 합니다.

마무리 함수가 누산에 쓰던 ConcurrentHashMap<String, LongAdder>와 다른 새 Map을 만들기 때문에 이 수집기는 IDENTITY_FINISH를 선언할 수 없습니다. 단계별 그림의 마지막 단계가 바로 이 복사입니다. 또 CONCURRENT는 같은 컨테이너를 동시에 갱신해도 된다는 허락일 뿐, 작업자 수나 실행 순서를 정하는 지시가 아닙니다.

표준 라이브러리에 같은 일을 하는 수집기가 있다면 그것을 먼저 씁니다. 단어별 개수라면 Collectors.groupingByConcurrent(Function.identity(), Collectors.counting())이 ConcurrentMap<String, Long>을 돌려주며, 이 수집기도 CONCURRENT와 UNORDERED 특성을 가집니다. 연습 문제 1에서 직접 써 봅니다.

더 깊이: 동시 수집에서는 결합 함수가 호출되지 않는다

동시 수집에서는 컨테이너가 하나뿐이므로 합칠 상대가 없습니다. 위 수집기와 같은 함수들에 호출 횟수를 세는 코드를 넣어 병렬 스트림에서 실행해 보면, 이 환경에서 공급자는 1번, 결합 함수는 0번 호출됐습니다. 그런데 UNORDERED 없이 CONCURRENT만 선언하고 순서가 있는 병렬 스트림에 쓰면 동시 수집의 조건이 깨져 보통의 병렬 수집으로 돌아갑니다. 이때는 공급자가 4번, 결합 함수가 3번 호출됐고, 스트림에 unordered()를 붙이자 다시 1번과 0번이 됐습니다. 정확한 횟수는 조각 수에 따라 달라지지만, 어느 경로를 타느냐는 앞의 규칙대로 정해집니다.

이 실험은 결합 함수를 대충 구현해도 된다는 뜻이 아닙니다. 특성이나 스트림의 순서가 바뀌는 순간 그 함수가 실제로 호출되기 때문입니다. Collector.Characteristics의 API 문서도 UNORDERED가 아닌 CONCURRENT 수집기는 순서가 없는 원본에 적용할 때만 동시에 평가해야 한다고 적고 있습니다.

정리

• ForkJoinPool은 작업자마다 덱을 두고, 자기 덱에서는 최근에 넣은 작업부터 처리합니다. 일이 없는 작업자는 다른 작업자 덱의 가장 오래된 작업, 즉 큰 덩어리를 훔칩니다.
• 작업 훔치기는 일을 고르게 나눠 주는 규칙이 아니라 놀게 된 작업자가 일을 찾아가는 방식입니다. 작업자별 분포와 훔친 횟수는 실행마다 달라지므로 테스트에서 기대하지 않습니다.
• 병렬 스트림은 원본을 Spliterator로 나눠 Fork/Join 작업으로 처리합니다. 어느 원소를 어느 스레드가 언제 처리할지는 정해져 있지 않습니다.
• toList() 같은 결과는 만남 순서를 지키지만 forEach의 실행 순서는 비결정적입니다. forEachOrdered는 순서와 가시성을 보장하는 대신 병렬의 이점을 줄이고, 순서가 필요 없으면 unordered()로 제약을 풉니다.
• 콜백에서 공유 변수를 고치면 갱신이 사라집니다. count, sum, reduce, collect처럼 조각별 결과를 합치는 연산을 씁니다.
• CONCURRENT는 컨테이너와 누산 함수가 실제로 동시 갱신에 안전할 때만 선언합니다. 병렬 스트림에 CONCURRENT 수집기를 쓰고 스트림이나 수집기 한쪽이 순서를 따지지 않으면, 컨테이너 하나를 함께 채우는 동시 수집이 일어납니다.

연습 문제

1. 태그별 개수를 병렬로 세기

여러 문서의 태그를 병렬 스트림으로 펼쳐 태그별 개수를 세세요. 공유 HashMap을 직접 고치지 말고 표준 동시 수집기를 쓰며, 결과의 키 순서는 따지지 않습니다.

import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Function;
import java.util.stream.Collectors;

public final class ParallelTagFrequencySolution {
    private record Document(String title, List<String> tags) {}

    public static void main(String[] args) {
        List<Document> documents = List.of(
                new Document("stream", List.of("java", "api")),
                new Document("pool", List.of("java", "concurrency")),
                new Document("collector", List.of("api", "java")));

        ConcurrentMap<String, Long> counts = documents.parallelStream()
                .flatMap(document -> document.tags().stream())
                .unordered()
                .collect(Collectors.groupingByConcurrent(Function.identity(), Collectors.counting()));

        System.out.println("java=" + counts.get("java"));
        System.out.println("api=" + counts.get("api"));
        System.out.println("concurrency=" + counts.get("concurrency"));
    }
}

java=3
api=2
concurrency=1

2. 태그 등장 횟수와 문서 수 구분하기

연습 문제 1의 첫 문서 태그를 java, api, java로 바꾸고, 태그가 없는 문서 하나를 더 추가합니다.

1. 태그별 등장 횟수는 어떻게 되나요?
2. java 태그가 들어 있는 문서는 몇 개인가요?
3. flatMap(document -> document.tags().stream().distinct())로 바꾸면 무엇을 세게 되나요?

3. ArrayList에 결과를 모으는 코드 고치기

다음 코드는 무엇이 문제이고, 어떻게 고쳐야 할까요? source는 정수가 담긴 List<Integer>라고 가정합니다.

List<Integer> result = new ArrayList<>();
source.parallelStream().map(value -> value * 10).forEach(result::add);

다음 글

병렬 스트림의 결합 법칙과 성능 측정에서는 이 글에서 쓴 reduce와 collect가 병렬에서도 맞는 답을 내려면 연산이 어떤 법칙을 지켜야 하는지 확인합니다. 이어서 공용 풀을 함께 쓸 때의 주의점과, 병렬이 정말 빠른지 재는 방법을 다룹니다.