🦊 chap7. 병렬 데이터 처리와 성능

chap7. 병렬 데이터 처리와 성능

7.1 병렬 스트림

컬렉션에 parallelStream 을 호출하면 병렬 스트림이 생성된다.

이는 각각의 스레드에서 처리할 수 있도록 스트림 요소를 여러 청크로 분할한 스트림이다.

숫자 n 을 인수로 받아서 1부터 n까지의 모든 숫자의 합계를 반환하는 메서드를 구현해보자.

public long sequentialSum(long n) {
	return Stream.iterate(1L, i -> i + 1)
								.limit(n)
								.reduce(0L, Long::sum);
}

전통적인 JAVA 에서는 반복문으로 구현할 수 있다.

public long iterativeSum(long n) {
	long result = 0;
	for (long i = 1L; i <= n; i++) {
		result += i;
	}
	return result;
}

n 이 커진다면 이 연산을 병렬로 처리하는 것이 좋을 것이다.

7.1.1 순차 스트림을 병렬 스트림으로 변환하기

순차 스트림에 parallel 메서드를 호출하면 기존의 함수형 리듀싱 연산이 병렬로 처리된다.

public long parallelSum(long n) {
	return Stream.iterate(1L, i -> i + 1)
								.limit(n)
								.parallel()
								.reduce(0L, Long::sum);
}

순차 스트림에 parallel 을 호출해도 스트림 자체는 아무 변화도 일어나지 않지만,

내부적으로는 이후 연산이 병렬로 수행해야 함을 의미하는 boolean 플래그가 설정된다.

반대로 sequential 로 병렬 스트림을 순차 스트림을 바꿀 수 있는데 이 두 메서드를 이용해서 제어할 수 있다.

stream.parallel()
			.filter(...)
			.sequential()
			.map(...)
			.parallel()
			.reduce(...)

parallel 과 sequential 중 마지막에 호출된 메서드가 전체 파이프라인에 영향을 미친다.

여기에서는 parallel 이므로 파이프라인은 전체적으로 병렬 실행된다.

7.1.2 스트림 성능 측정

병렬화를 이용하면 순차나 반복에 비해 성능이 좋아질 것이라 추측했지만

소프트웨어 공학에서 추측은 위험한 방법이다.

항상 측정을 하라.

자바 마이크로벤치마크 하니스 (JMH) 라이브러리는 간단하고 어노테이션 기반 방식을 지원하며 안정적으로 자바 가상 머신 (JVM)을 대상으로 하는 벤치마크를 구현할 수 있다.

@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Fork(2, jvmArgs={"-Xms4G", "-Xmx4G"})
public class ParallelStreamBenchmark {
	private static final long N = 10_000_000L;

	@Benchmark
	public long sequentialSum() {
		return Stream.iterate(1L, i -> i + 1).limit(N)
								.retduce(0L, Long::sum);
	}

	@TearDown(Level.Invocation)
	public void tearDown() {
		System.gc();
	}
}

벤치마크가 가능한 가비지 컬렉터의 영향을 받지 않도록 힙의 크기를 충분하게 설정했고

벤치마크가 끝날 때마다 가비지 컬렉터가 실행되도록 강제했다.

이 코드를 실행할 때 JMH 명령은 핫스팟이 코드를 최적화 할 수 있도록 20번을 실행하며

벤치마크를 준비한 다음 20번을 더 실행해 최종 결과를 계산한다.

즉 20 + 20 회 반복 실행한다.

sequentialSum -> 121.843
iterativeSum -> 3.273
parallelSum -> 603.059

병렬 버전이 쿼드 코어 CPU를 활용하지 못하고 순차 버전에 비해 5배나 느렸다.

• 반복 결과로 박싱된 객체가 만들어지기 때문에 숫자를 더하려면 언박싱이 필요하다.
• 반복 작업은 병렬로 수행할 수 있는 독립 단위로 나누기가 어렵다.

2번째의 이유로 스트림을 병렬로 처리할 수 있도록 청크로 분할할 수 없었다.

순차처리 방식과 크게 다른 점이 없어 스레드를 할당하는 오버헤드만 증가하게 되었다.

더 특화된 메서드 사용

LongStream.rangeClosed 는 iterate 에 비해 다음과 같은 장점이 있다.

• 기본형 long 을 직접 사용하므로 박싱, 언박싱 오버헤드가 사라진다.
• 쉽게 청크로 분할할 수 있는 숫자 범위를 생산한다.

@Benchmark
public long rangedSum() {
	return LongStream.rangeClosed(1, N)
									.reduce(0L, Long::sum);
}

rangedSum -> 5.315

@Benchmark
public long parallelRangedSum() {
	return LongStream.rangeClosed(1, N)
									.parallel()
									.reduce(0L, Long::sum);
}

parallelRangedSum -> 2.677

기존의 iterate 팩토리 메서드로 생성한 순차 버전에 비해 스트림 처리 속도가 빨라졌고

순차 실행보다 빠른 성능을 갖는 병렬 리듀싱을 만들었다.

하지만 병렬화가 완전 공짜는 아니라는 사실을 기억해라.

스트림을 재귀적으로 분할해야 하고, 서브스트림을 서로 다른 스레드의 리듀싱 연산으로 할당하고, 이것들을 하나의 값으로 합쳐야 한다.

7.1.3 병렬 스트림의 올바른 사용법

병렬 스트림을 잘못 사용하면서 발생하는 많은 문제는 공유된 상태를 바꾸는 알고리즘을 사용하기 때문에 일어난다.

public long sideEffectSum(long n) {
	Accumulator accumulator = new Accumulator();
	LongStream.rangeClosed(1, n).forEach(accumulator::add);
	return accumulator.total;
}

public class Accumulator {
	public long total = 0;
	public void add(long value) { total += value; }
}

이 코드는 본질적으로 순차 실행할 수 있도록 구현되어 있으므로 병렬로 실행하면 참사가 일어난다.

동기화로 문제를 해결하다보면 결국 병렬화라는 특성이 없어진다.

7.1.4 병렬 스트림 효과적으로 사용하기

• 확신이 서지 않으면 직접 측정하라.

병렬 스트림이 항상 순차 스트림보다 빠른 것은 아니다.

적절한 벤치마크로 직접 성능을 측정하는 것이 바람직하다.

• 박싱을 주의하라.

자동 박싱과 언박싱은 성능을 크게 저하시킬 수 있는 요소이다.

자바 8은 박싱 동작을 피할 수 있도록 기본형 특화 스트림을 제공한다.

• 병렬 스트림에서 성능이 떨어지는 연산이 있다.

limit 이나 findFirst 처럼 요소의 순서에 의존하는 연산을 병렬 스트림에서 수행하면 비싼 비용을 치러야 한다.

대신 findAny 는 요소의 순서와 상관없이 연산하므로 findFirst 보다 성능이 좋다.

• 스트림에서 수행하는 전체 파이프라인 연산 비용을 고려하라.

• 소량의 데이터에서는 병렬화 과정에서 생기는 부가 비용을 상쇄할 만큼 이득이 없다.

• 스트림을 구성하는 자료구조가 적절한지 확인하라.

ArrayList 를 LinkedList 보다 효율적으로 분할할 수 있다.

range 팩토리 메서드로 만든 기본형 스트림도 쉽게 분할할 수 있다.

• 스트림의 특성과 파이프라인 중간 연산이 스트림의 특성을 어떻게 바꾸는지에 따라 분해 과정 성능이 달라질 수 있다.

• 최종 연산의 병합 과정 비용이 비싸다면 병렬 스트림으로 얻은 성능 이익이 서브스트림의 부분 결과를 합치는 과정에서 상쇄될 수 있다.

7.2 포크/조인 프레임워크

포크/조인 프레임워크는 병렬화할 수 있는 작업을 재귀적으로 작은 작업으로 분할한 다음

서브테스크 각각의 결과를 합쳐 전체 결과를 만들도록 설계되었다.

7.2.1 RecursiveTask 활용

스레드 풀을 이용하기 위해 RecursiveTask<R> 의 서브클래스를 만들어야 한다.

RecursiveTask 를 정의하기 위해 추상 메서드 compute 를 구현해야 한다.

protected abstract R compute();

compute 는 태스트를 서브태스크로 분할하는 로직과

더 이상 분할할 수 없을 때 개별 서브태스트의 결과를 생산할 알고리즘을 정의한다.

대부분의 compute 메서드 구현은 다음과 같은 의사코드를 유지한다.

if (태스트가 충분히 작거나 더 이상 분할할 수 없으면) {
	순차적으로 태스트 계산
} else {
	태스크를 두 서브태스크로 분할
	태스크가 다시 서브태스크로 분할되도록 이 메서드를 재귀적으로 호출함
	모든 서브태스크 연산이 완료될 때까지 기다림
	각 서브태스크의 결과를 합침
}

이 알고리즘은 분할정복 알고리즘의 병렬화 버전이다.

7.2.2 포크/조인 프레임워크를 제대로 사용하는 방법

• join 메서드를 태스크에 호출하면 태스크가 생산하는 결과가 준비될 때까지 호출자를 블록시킨다.

따라서 두 서브태스크가 모두 시작된 다음에 join 을 호출해야 한다.

그렇지 않으면 각각의 서브태스크가 서로 다른 태스크가 끝나기를 기다리며 성능이 하락한다.

• RecursiveTask 내에서는 ForkJoinPool 의 invoke 메서드를 사용하지 말아야 한다.

대신 compute 나 fork 를 직접 호출하라.

순차 코드에서 병렬 계산을 시작할 때만 invoke 를 사용한다.

• 서브태스크에 fork 메서드를 호출해서 ForkJoinPool 의 일정을 조절할 수 있다.

한쪽 작업에는 fork 보다 compute 를 호출하는 것이 효율적이다.

두 서브태스크의 한 태스크에는 같은 스레드를 재사용할 수 있기 때문에 불필요한 태스크를 할당하는 오버헤드를 피할 수 있다.

• 포크/조인 프레임워크를 이용하는 병렬 계산은 디버깅이 어렵다.

보통 stack trace 로 문제가 일어난 과정을 쉽게 확인할 수 있지만, 프레임워크에서는 fork 라 불리는 다른 스레드에서 compute 를 호출하기 때문에 도움이 되지 않는다.

• 멀티코어에 무조건 프레임워크를 사용하는 것이 순차 처리보다 빠를 거라는 생각을 버려라.

7.2.3 작업 훔치기

작업 훔치기 기법에서는 ForkJoinPool 의 모든 스레드를 거의 공정하게 분할한다.

각각의 스레드는 할당된 태스크를 포함하는 더블 링크드 리스트를 참조하면서 작업이 끝날 때마다 큐의 헤드에서 다른 태스크를 가져와서 작업한다.

한 스레드는 다른 스레드보다 자신에게 할당된 태스크를 더 빨리 처리할 수 있다.

할일이 없어진 스레드는 유휴 상태로 바뀌는 것이 아니라 다른 스레드 큐의 꼬리에서 작업을 훔쳐온다.

작업자의 큐에 있는 태스크를 두 개의 서브 태스크로 분할했을 때,

둘 중 하나의 태스크를 다른 유휴 작업자가 훔쳐갈 수 있다.

주어진 태스크를 순차 실행할 단계가 될 때까지 이 과정을 재귀적으로 반복한다.

7.3 Spliterator 인터페이스

자바 8은 Spliterator 라는 인터페이스를 제공한다.

이는 분할할 수 있는 반복자라는 의미이다.

public interface Spliterator<T> {
	boolean tryAdvance(Consumer<? super ?> action);
	Spliterator<T> trySplit();
	long estimateSize();
	int characteristics();
}

• T는 Spliterator 에서 탐색하는 요소의 형식을 가리킨다.
• tryAdvance 메서드는 요소를 소비하며 탐색할 요소가 남아있으면 true 를 반환한다.
• trySplit. 메서드는 일부 요소를 분할해서 두 번째 Spliterator 를 생성한다.
• Spliterator 에서는 estimateSize 메서드로 탐색할 요소 수 정보를 제공할 수 있다.

7.3.1 분할 과정

스트림을 여러 스트림으로 분할하는 과정은 재귀적으로 일어난다.

trySplit 의 결과가 null 이 될 때까지 이 과정을 반복한다.

Spliterator 특성

Spliterator 는 characteristics 라는 추상 메서드도 정의한다.

이는 자체의 특성 집합을 포함하는 int 를 반환한다.

• ORDERED : 정해진 순서가 있어 유의해야 한다.
• DISTINCT : x.equals(y) 는 항상 false 이다.
• SORTED : 미리 정의된 정렬 순서를 따른다.
• SIZED : 크기가 알려진 소스로 Spliterator 를 생성해서 estimatedSize() 는 정확한 값을 반환한다.
• NON-NULL : 탐색할 모든 요소는 null 이 아니다.
• IMMUTABLE : 모든 소스는 불변이다.
• CONCURRENT : 동기화 없이 여러 스레드에서 소스를 동시에 고칠 수 있다.
• SUBSIZED : 자신 및 모든 하위 Spliterator 는 SIZED 이다.

7.3.2 커스텀 Spliterator 구현하기

문자열의 단어 수를 계산하는 단순한 메서드를 구현해보자.

다음은 반복 버전이다.

public int countWordsIteratively(String s) {
	int counter = 0;
	boolean lastSpace = true;
	for (char c : s.toCharArray()) {
		if (Character.isWhitespace(c)) {
			lastSpace = true;
		} else {
			if (lastSpace) counter++;
			lastSpace = false;
		}
	}
}

함수형으로 단어 수를 세는 메서드 재구현하기

우선 String 을 스트림으로 변환해야 한다.

Stream<Character> stream = IntStream.range(0, SENTENCE.length())
																			.mapToObj(SENTENCE::charAt);

지금까지 발견한 단어 수를 계산하는 int 변수와

마지막 문자가 공백이었는지 여부를 기억하는 boolean 변수가 필요하다.

이들을 캡슐화하는 새로운 클래스 WordCounter 를 만들어야 한다.

class WordCounter {
	private final int counter;
	private final boolean lastSpace;

	public WordCounter(int counter, boolean lastSpace) {
		this.counter = counter;
		this.lastSpace = lastSpace;
	}

	public WordCounter accumulate(Character c) {
		if (Character.isWhitespace(c)) {
			return lastSpace ? this : new WordCounter(counter, true);
		} else {
			return lastSpace ? new WordCounter(counter + 1, false) : this;
		}
	}

	public WordCounter combine(WordCounter wordCounter) {
		return new WordCounter(counter + wordCounter.counter, wordCounter.lastSpace);
	}

	public int getCounter() {
		return counter;
	}
}

스트림을 탐색하면서 새로운 문자를 찾을 때마다 accumulate 메서드를 호출한다.

새로운 비공백 문자를 탐색한 후 마지막 문자가 공백이면 counter 를 증가시킨다.

combine 은 문자열 서브 스트림을 처리한 WordCounter 결과를 합친다.

이제 스트림의 리듀싱 연산을 직관적으로 구현할 수 있다.

private int countWords(Stream<Character> stream) {
	WordCounter wordCounter = stream.reduce(new WordCounter(0, true),
																					WordCounter::accumulate,
																					WordCounter::combine);
	return wordCounter.getCounter();
}

-> Found 19 words

WordCounter 병렬로 수행하기

countWords(stream.parallel())

-> Found 25 words

병렬 스트림은 원하는 결과가 나오지 않는다.

원래 문자열을 임의의 위치에서 둘로 나누다보니 예상치 못하게 하나의 단어를 둘로 계산하는 상황이 발생할 수 있다.

단어가 끝나는 위치에서만 분할하는 방법으로 이 문제를 해결할 수 있다.

그러려면 단어 끝에서 문자열을 분할하는 문자 Spliterator 가 필요하다.

class WordCounterSpliterator implements Spliterator<Character> {
	private final String string;
	private int currentChar = 0;
	public WordCounterSpliterator(String string) {
		this.string = string;
	}

	// 현재 문자를 소비하고 소비할 문자가 남아있으면 true 를 리턴
	@Override
	public boolean tryAdvance(Consumer<? super Character> action) {
		action.accept(string.charAt(currentChar++));
		return currentChar < string.length();
	}

	@Override
	public Spliterator<Character> trySplit() {
		int currentSize = string.length() - currentChar;
		if (currentSize < 10) return null;
		
		for (int splitPos = currentSize / 2 + currentChar; splitPost < string.length(); splitPos++) {
			if (Character.isWhitespace(string.charAt(splitPos))) {
				Spliterator<Character> spliterator =
						new WordCounterSpliterator(string.substring(currentChar, splitPos));
				currentChar = splitPost;
				return spliterator;
			}
		}
		return null;
	}
	
	@Override
	public long estimateSize() {
		return string.length() - currentChar;
	}

	@Override
	public int characteristics() {
		return ORDERED + SIZED + SUBSIZED + NON-NULL + IMMUTABLE;
	}
}

• tryAdvance 는 현재 인덱스의 문자를 Consumer 에 제공한 후 인덱스를 증가한다.

새로운 커서 위치가 전체 문자열 길이보다 작으면 true 를 반환하는데

이는 반복 탐색할 문자가 남아있다는 뜻이다.

• trySplit 은 반복될 자료구조를 분할하는 로직을 포함, 가장 중요한 메서드이다.

분할 동작을 중단할 한계를 설정해야 한다.

실전 어플리케이션에서는 너무 많은 태스크를 만들지 않도록 더 높은 한계값을 설정해야 한다.

남은 문자 수가 한계값 이하면 null 을 반환하여 분할을 중지한다.

분할이 필요한 경우 파싱할 문자열 청크 중간 위치를 기준으로 분할하도록 지시한다.

• 탐색해야 할 요소의 개수 estimatedSize 는 파싱할 문자열 전체 길이와 반복중인 위치의 차 이다.

• characteristic 메서드는 이 Spliterator 의 특성을 포함한다.

WordCounterSpliterator 활용

이제 이를 병렬 스트림에 사용할 수 있다.

Spliterator<Charater> spliterator = new WordCounterSpliterator(SENTENCE);
Stream<Character> stream = StreamSupport.stream(spliterator, true);

-> Found 19 words

이제 기대하는 출력 결과가 나왔다.