RecursiveTask와 분할 경계
ForkJoinPool과 RecursiveTask의 실행 구조를 만들고 겹치는 분할 오답, 리프 크기, `fork`·`compute`·`join` 순서를 확인합니다.
Fork/Join 프레임워크는 ForkJoinPool이 작업자와 큐를 관리하고 ForkJoinTask가 분할 가능한 계산을 표현합니다.
값이 필요한 작업은 RecursiveTask<V>, 결과 없는 동작은 RecursiveAction을 상속합니다.
중요한 것은 class 이름이 아니라 분할이 전체 입력을 정확히 한 번 덮고, 리프 결과를 결합해 순차 기준과 같은 값을 만드는 것입니다.
midpoint를 양쪽 범위에 포함
양끝을 포함하는 범위 [from, to]를 [from, mid]와 [mid, to]로 나누면 중간점이 중복됩니다.
1~8에 10을 곱해 합칠 때 정상값은 360이지만 아래 출력은 wrong-sum=400입니다.
루트 작업의 중간점 4가 양쪽 리프에 반복해서 포함되기 때문입니다.
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);
}
}
}원칙은 반열린 범위 [from, to)를 일관되게 사용하거나, 양끝을 포함하는 범위라면 오른쪽을 mid + 1부터 시작하는 것입니다.
종료 조건도 같은 범위 정의로 계산해야 빈 범위, 무한 재귀, 중복 계산을 피할 수 있습니다.
완전한 RecursiveTask 구현
아래 구현은 List<Integer>의 반열린 인덱스 범위를 사용합니다.
범위 크기가 임계값 이하이면 직접 계산하고, 그보다 크면 정확히 둘로 나눕니다.
왼쪽 작업을 fork해 큐에 공개하고 오른쪽 작업은 현재 작업자가 계산한 뒤 왼쪽 작업을 join합니다.
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);
}
}
}pool.invoke(task)는 작업을 제출하고 결과가 완성될 때까지 기다립니다.
compute() 안에서 블로킹 I/O를 수행하면 제한된 작업자가 모두 대기할 수 있으므로 이 구조는 짧고 CPU 중심인 리프에 더 잘 맞습니다.
임계값은 정확성과 별개의 성능 매개변수
임계값이 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());
}
}
}
}이 확인 프로그램의 AtomicInteger는 계산 결과가 아니라 계측 카운터입니다.
핵심 결과는 각 작업이 반환해 결합합니다.
실측에서는 카운터 자체가 경합을 만들어 시간을 왜곡할 수 있으므로 구조 확인용 실행과 벤치마크 실행을 분리합니다.
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));
}
}fork·join 남용 방지
fork()는 하위 작업을 작업자 큐에 넣어 다른 작업자가 가져갈 기회를 줍니다.
join()은 완료되지 않았으면 기다리거나 도움 실행을 합니다.
현재 작업이 할 수 있는 일을 남겨 두지 않고 모든 하위를 fork한 뒤 차례로 join하면 지역성이 나빠지고 스케줄링이 늘 수 있습니다.
- 한 하위를
fork하고 다른 하위를 직접compute한다. join은 의존성이 실제로 필요한 지점까지 늦춘다.- 리프가 너무 작아지지 않도록 임계값을 데이터와 연산량에 맞춘다.
- 결합 연산은 가능하면 결합 법칙을 만족하고 순차 기준과 동일한 결과를 내야 한다.
- 작업이 외부 락을 오래 잡거나 블로킹 호출을 수행하지 않게 한다.
임계값을 CPU 코어 수만으로 계산할 수 있나요?
코어 수만으로는 부족합니다. 원소 하나의 계산 비용, 분할과 객체 생성 비용, 캐시 지역성, 입력 불균형이 모두 영향을 줍니다. 먼저 충분히 큰 리프를 선택하고 여러 대표 입력에서 결과가 같은지 확인한 뒤 벤치마크로 조정합니다. 원본 크기가 작으면 순차 경로를 택하는 전환점도 유용합니다.
연습 문제
반열린 인덱스 범위를 사용해 배열의 최댓값을 찾으세요.
빈 범위를 만들지 않고 임계값 이하에서는 직접 순회하며, 부모 작업은 Math.max로 결과를 결합해야 합니다.
해설 보기
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와 함께 모든 인덱스가 정확히 한 리프에 속하는 것입니다.
빈 배열에 대한 정책은 이 작업 밖에서 예외나 OptionalInt로 명시합니다.