Future.get을 부르는 시점과 병렬 합산

앞 글에서는 submit이 돌려준 Future로 결과를 받았습니다. 그런데 Future를 쓴다고 작업이 저절로 동시에 실행되지는 않습니다. 작업 하나를 맡기고 바로 get()으로 결과를 기다린 다음 둘째 작업을 맡기면, 첫 결과를 기다리는 동안 둘째 작업은 시작조차 하지 않습니다. 언제 get()을 부르느냐가 작업이 실제로 겹쳐 실행되는지를 정하는 셈입니다.

이 글에서는 같은 작업을 get()의 위치만 바꿔 실행해 걸린 시간을 비교하고, 숫자 범위를 둘로 나눠 동시에 더해 봅니다. 이어서 여러 결과를 하나의 기한 안에서 모으는 방법과, 일을 몇 조각으로 나눌지 정하는 기준을 살펴봅니다.

submit 바로 뒤에 get을 부르면

다음 프로그램은 작업자가 2개인 풀에 0.2초짜리 작업 두 개를 맡깁니다. 각 submit 바로 뒤에 .get()을 이어 붙였기 때문에, 첫 작업의 결과를 받은 다음에야 둘째 작업을 맡깁니다.

import java.util.concurrent.Executors;

public final class SequentialFutureWait {
    public static void main(String[] args) throws Exception {
        var executor = Executors.newFixedThreadPool(2);
        long start = System.nanoTime();
        try {
            int first = executor.submit(() -> {
                Thread.sleep(200);
                return 20;
            }).get();
            int second = executor.submit(() -> {
                Thread.sleep(200);
                return 22;
            }).get();
            System.out.println("sum=" + (first + second));
            System.out.println("millis=" + (System.nanoTime() - start) / 1_000_000);
        } finally {
            executor.shutdown();
        }
    }
}

sum=42
millis=436

합은 20 + 22 = 42이고, 걸린 시간은 400밀리초를 조금 넘습니다. 여러 번 실행하면 값은 매번 조금씩 다르지만 400보다 작아지지는 않습니다. main이 첫 get()에서 0.2초를 기다리는 동안 둘째 작업은 아직 맡겨지지도 않았고, 첫 결과를 받은 뒤에야 둘째 작업이 다시 0.2초를 자기 때문입니다. 작업자가 둘이어도 한 번에 하나만 일하는 셈입니다. 400을 넘는 나머지 시간은 작업을 맡기고 결과를 받고 출력하는 비용과 스케줄링 지연입니다.

이번에는 두 작업을 먼저 모두 맡긴 뒤에 결과를 받도록 순서만 바꿔 보겠습니다. 코드에서 달라진 곳은 get()을 부르는 위치뿐입니다.

import java.util.concurrent.Executors;

public final class ParallelFutureWait {
    public static void main(String[] args) throws Exception {
        var executor = Executors.newFixedThreadPool(2);
        long start = System.nanoTime();
        try {
            var first = executor.submit(() -> {
                Thread.sleep(200);
                return 20;
            });
            var second = executor.submit(() -> {
                Thread.sleep(200);
                return 22;
            });
            System.out.println("sum=" + (first.get() + second.get()));
            System.out.println("millis=" + (System.nanoTime() - start) / 1_000_000);
        } finally {
            executor.shutdown();
        }
    }
}

sum=42
millis=227

이번에는 200밀리초를 조금 넘는 시간에 끝났습니다. 값은 실행마다 조금씩 다르지만 앞 프로그램의 절반 정도라는 점은 같습니다. 두 작업이 get()보다 먼저 맡겨졌으므로 두 작업자가 동시에 0.2초씩 자고, main은 그동안 첫 get()에서 기다립니다. 첫 결과를 받을 즈음이면 둘째 작업도 거의 끝나 있어서 둘째 get()은 곧바로 돌아옵니다. 서로 기다릴 필요가 없는 작업이라면 먼저 모두 맡기고 결과는 나중에 모아야 실행 시간이 겹칩니다.

범위를 나눠 동시에 더하기

같은 방법으로 계산을 나눠 맡길 수 있습니다. 다음 ParallelRangeSum은 1부터 100까지의 합을 1부터 50까지와 51부터 100까지의 두 범위로 나누고, 각 범위의 합을 작업으로 맡깁니다. List.of의 인수인 두 submit이 모두 실행된 다음에 반복문이 결과를 모읍니다.

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

public final class ParallelRangeSum {
    static long sum(long start, long endInclusive) {
        long result = 0;
        for (long value = start; value <= endInclusive; value++) {
            result += value;
        }
        return result;
    }

    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        try {
            List<Future<Long>> futures = List.of(
                    executor.submit(() -> sum(1, 50)),
                    executor.submit(() -> sum(51, 100)));
            long total = 0;
            for (Future<Long> future : futures) {
                total += future.get();
            }
            System.out.println("total=" + total);
        } finally {
            executor.shutdown();
        }
    }
}

total=5050

단계 그림은 앞의 SequentialFutureWait와 이 프로그램이 첫 get()을 부르기 전에 작업을 몇 개 맡겼는지 비교한 것입니다. SequentialFutureWait는 첫 작업 하나만 맡긴 채 기다리고, ParallelRangeSum은 두 범위를 모두 맡긴 뒤에 기다립니다. 그림이 보여 주는 것은 호출 순서입니다. 두 프로그램은 하는 일이 다르므로 걸린 시간을 비교하는 그림이 아니고, 맡긴 순서만으로 작업자가 실제로 어떤 순서로 실행했는지도 알 수 없습니다.

결과 5050이 맞는지는 범위별로 검산할 수 있습니다. 1부터 50까지의 합은 50 × 51 ÷ 2 = 1275이고, 51부터 100까지의 합은 (51 + 100) × 50 ÷ 2 = 3775이므로 더하면 5050입니다. 첫 범위가 50에서 끝나고 둘째 범위가 51에서 시작하므로 빠지거나 겹치는 수가 없습니다. 만약 둘째 범위를 50부터 시작하면 50을 두 번 더해 5100이 됩니다. 이런 실수는 실행기를 어떻게 고르든 생기므로, 일을 나눌 때는 나눈 지점이 맞는지부터 테스트해야 합니다.

이 예제는 구조를 보여 주기 위한 것이라 계산이 아주 작습니다. 숫자 100개를 더하는 일은 작업을 맡기고 결과를 받는 비용보다 훨씬 싸므로, 실제로는 나누지 않고 한 스레드에서 더하는 편이 빠릅니다. 나눠서 이득이 나는지는 실제 입력 크기로 재 봐야 하며, 그 기준은 뒤의 '몇 조각으로 나눌까'에서 다룹니다.

여러 결과를 하나의 기한 안에서 기다리기

get()에 시간 제한을 주더라도 Future마다 1초씩 따로 주면, 결과 세 개를 모으는 데 최대 3초가 걸릴 수 있습니다. 화면 응답처럼 전체를 1초 안에 끝내야 한다면 기한을 한 번만 정하고, 각 get()에는 그 기한까지 남은 시간만 줘야 합니다. 다음 collect가 그렇게 동작합니다.

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public final class DeadlineFutureCollector {
    static <T> List<T> collect(List<Future<T>> futures, Duration budget) throws Exception {
        long deadline = System.nanoTime() + budget.toNanos();
        List<T> results = new ArrayList<>();
        try {
            for (Future<T> future : futures) {
                long remaining = deadline - System.nanoTime();
                if (remaining <= 0) throw new TimeoutException("batch deadline");
                results.add(future.get(remaining, TimeUnit.NANOSECONDS));
            }
            return List.copyOf(results);
        } catch (Exception e) {
            futures.forEach(future -> future.cancel(true));
            throw e;
        }
    }

    public static void main(String[] args) throws Exception {
        var executor = java.util.concurrent.Executors.newFixedThreadPool(2);
        try {
            var futures = List.of(executor.submit(() -> 7), executor.submit(() -> 8));
            System.out.println(collect(futures, Duration.ofSeconds(1)));
        } finally {
            executor.shutdownNow();
        }
    }
}

[7, 8]

collect는 들어오자마자 System.nanoTime()에 예산을 더해 기한을 한 번만 계산합니다. nanoTime()은 경과 시간을 재기 위한 값이라서 벽시계 시각과는 관계가 없습니다. 반복할 때마다 기한까지 남은 시간 remaining을 다시 구해, 남은 시간이 없으면 TimeoutException("batch deadline")을 직접 던지고 있으면 그만큼만 get()에 줍니다. 모든 결과를 받으면 List.copyOf로 바꿀 수 없는 복사본을 만들어 돌려줍니다.

중간에 어떤 예외가 나든 catch는 모든 Future에 cancel(true)로 취소를 요청한 뒤 같은 예외를 다시 던집니다. 그래서 이 메서드는 결과를 전부 돌려주거나 예외를 던지거나 둘 중 하나이며, 앞에서 받아 둔 일부 결과만 돌려주는 일은 없습니다. 이미 끝난 Future에 대한 취소 요청은 아무 효과가 없고, 실행 중인 작업에 대한 요청도 작업이 인터럽트에 협력해야 효과가 있습니다. 취소를 요청한 뒤 작업이 실제로 끝났는지는 이 메서드가 확인하지 않습니다.

예산이 100밀리초일 때 남은 시간이 어떻게 줄어드는지 숫자로 따라가 보겠습니다. 다음 표는 첫 결과를 받고 정리하는 데 60밀리초가 걸리고, 둘째 결과는 그로부터 50밀리초 뒤에 준비된다고 가정한 것입니다.

둘째 get()은 새로 100밀리초를 받는 것이 아니라 남은 40밀리초만 받으므로, 110밀리초에 준비될 결과를 받지 못합니다. get()마다 100밀리초씩 따로 기다렸다면 이 결과를 받았겠지만, 대신 전체 대기 시간이 최대 200밀리초까지 늘어날 수 있습니다. 실제로 첫 작업은 60밀리초, 둘째 작업은 110밀리초를 자게 만들어 100밀리초 예산으로 실행해 보면, 100밀리초를 조금 넘긴 시점에 TimeoutException이 나고 둘째 작업은 취소됩니다. 이때 예외에는 메시지가 없는데, get()이 기다리다 시간 초과로 던진 것이기 때문입니다.

"batch deadline" 메시지는 반복을 시작하는 시점에 이미 기한이 지났을 때 나옵니다. 남은 시간을 먼저 검사하므로, 기한이 지난 뒤라면 다음 결과가 이미 준비돼 있어도 받지 않고 시간 초과로 처리합니다. 결과를 하나라도 더 건지는 것보다 기한을 지키는 쪽을 택한 설계입니다.

이 예산은 collect에 들어온 순간부터 계산되므로, 그 전에 작업을 맡기느라 쓴 시간은 빠집니다. 요청 전체를 하나의 기한으로 묶으려면 요청이 시작될 때 기한을 계산해 두고, 예산 대신 그 기한을 collect에 넘기도록 바꿔야 합니다. 또 기한은 get()이 기다리는 시간에만 적용되므로, 취소 처리와 실행기 종료까지 합친 실제 반환 시각은 예산보다 조금 늦을 수 있습니다.

이 예제는 짧은 양수 예산과 null이 아닌 결과를 가정합니다. 예산이 너무 커서 나노초로 나타낼 수 없으면 Duration.toNanos()가 ArithmeticException을 던지고, 작업이 null을 돌려주면 마지막 List.copyOf가 NullPointerException을 던집니다. null 결과가 어떻게 처리되는지는 연습 문제 4에서 따라가 봅니다.

몇 조각으로 나눌까

일을 너무 크게 나누면 가장 늦게 끝나는 조각이 전체 완료 시각을 정하고, 너무 잘게 나누면 작업을 맡기고 큐를 거치고 결과를 합치는 비용이 실제 계산보다 커집니다. CPU 계산이라면 Runtime.getRuntime().availableProcessors()가 알려 주는 프로세서 수를 출발점으로 삼고, 조각마다 일의 양이 비슷한지 확인합니다. 외부 서비스를 호출하는 일이 섞여 있다면 프로세서 수보다 상대 시스템의 연결 상한과 허용 요청률을 먼저 봅니다.

입력이 작을 때는 아예 나누지 않는 기준도 필요합니다. 측정해 보면 나누는 비용이 계산으로 아끼는 시간보다 큰 구간이 있는데, 그보다 작은 입력은 한 스레드에서 바로 계산하게 하면 간단한 요청이 큐와 스레드를 쓸데없이 거치지 않습니다. 일을 나누고 합치는 이런 계산은 병렬 스트림과 Fork/Join에서 더 편한 도구로 다시 만납니다.

빠른 것보다 먼저 맞아야 합니다. 부분합이 int 범위를 넘을 수 있다면 부분합과 전체 합을 long으로 계산하고, long의 범위마저 넘을 수 있는 입력은 따로 처리해야 합니다. 부동소수점 합은 더하는 묶음이 달라지면 결과의 마지막 자리가 달라질 수 있습니다. 예를 들어 (0.1 + 0.2) + 0.3은 0.6000000000000001이고 0.1 + (0.2 + 0.3)은 0.6이므로, 나눠서 더한 결과는 허용 오차를 두고 비교해야 합니다. 그래서 병렬 버전은 먼저 순차 버전과 여러 입력에서 결과가 같은지 확인한 다음에 처리 시간을 비교합니다.

더 깊이: 끝값이 Long.MAX_VALUE라면

ParallelRangeSum의 sum은 1부터 100처럼 작은 양수 범위를 가정합니다. 끝값이 Long.MAX_VALUE이면 value <= endInclusive가 언제나 참이라서, 마지막 value++에서 값이 Long.MIN_VALUE로 넘어간 뒤에도 반복이 계속됩니다. 즉 반복문이 끝나지 않습니다. 모든 long 범위를 다뤄야 한다면 끝값을 처리한 직후 반복을 빠져나오도록 반복 방식을 바꾸고, Math.addExact처럼 오버플로가 나면 예외를 던지는 덧셈으로 부분합을 계산해야 합니다.

정리

• Future를 쓴다고 작업이 동시에 실행되지는 않습니다. submit 바로 뒤에 get()을 부르면 작업이 하나씩 실행됩니다.
• 서로 기다릴 필요가 없는 작업은 먼저 모두 맡기고, Future를 모아 두었다가 나중에 결과를 받습니다.
• 일을 나눌 때는 나눈 지점에서 값이 빠지거나 겹치지 않는지부터 확인합니다. 작은 입력은 나누지 않는 편이 빠를 수 있습니다.
• 여러 결과를 기다릴 때는 기한을 한 번 정하고, 각 get()에 남은 시간만 줍니다. 실패하면 남은 작업에 취소를 요청하되, 취소 요청이 작업 종료를 확인해 주지는 않습니다.
• 작업 하나가 실패했을 때 전체를 실패로 볼지, 나머지 결과라도 쓸지는 기능에 맞게 정합니다. collect는 전체를 실패로 보는 방식입니다.

연습 문제

1. 1부터 1,000까지를 두 범위로 나눠 각 범위의 홀수 개수와 짝수 개수를 Callable로 계산하고, 두 작업을 모두 맡긴 뒤 결과를 합치세요.
2. 1번 풀이의 대상 범위를 2부터 1,000까지로 바꿔 2부터 500까지와 501부터 1,000까지로 나누면, 각 부분과 최종 결과의 홀수·짝수 개수는 얼마인가요? 전체 개수 999와도 맞는지 확인하세요.
3. 작업을 맡기는 데 이미 30밀리초를 쓴 뒤 collect를 100밀리초 예산으로 호출했습니다. 첫 결과를 받고 정리하는 데 60밀리초가 걸렸고, 둘째 결과는 그로부터 50밀리초 뒤에 준비됩니다. 둘째 get()에는 시간이 얼마나 주어지며, 결과를 받을 수 있나요? 앞의 30밀리초는 이 기한에 포함되나요?
4. 예산은 넉넉하고, 두 Future가 각각 7과 null을 돌려주고 정상적으로 끝났습니다. collect는 [7, null]을 돌려주나요? 어디서 어떤 예외가 나는지, 7만 담은 목록이라도 돌려주는지 설명하세요.

import java.util.concurrent.Executors;

public final class ParallelParitySolution {
    record Counts(int odd, int even) {
        Counts plus(Counts other) { return new Counts(odd + other.odd, even + other.even); }
    }

    static Counts count(int start, int end) {
        int odd = 0, even = 0;
        for (int n = start; n <= end; n++) {
            if ((n & 1) == 0) {
                even++;
            } else {
                odd++;
            }
        }
        return new Counts(odd, even);
    }

    public static void main(String[] args) throws Exception {
        var executor = Executors.newFixedThreadPool(2);
        try {
            var left = executor.submit(() -> count(1, 500));
            var right = executor.submit(() -> count(501, 1_000));
            System.out.println(left.get().plus(right.get()));
        } finally {
            executor.shutdown();
        }
    }
}

Counts[odd=500, even=500]

다음 글

Future 취소와 시간 초과, 실패 구분하기에서는 이 글에서 요청만 하고 넘어간 취소가 실제로 어떻게 동작하는지, 작업이 인터럽트에 협력해야 하는 이유와 함께 살펴봅니다.