Memory False Sharing이란 무엇일까? 말 그대로 직역하면 메모리 거짓 공유이다. 이게 무엇을 뜻하는지 알아보자.

Cache Coherence

먼저 Cache Coherence를 알아야 한다. 멀티코어 환경에서 코어마다 cache가 각각 존재한다. 흔히 말하는 캐시의 개념으로 자주 사용하는 데이터들을 메모리보다 더 빠른 캐시에 저장함으로서 메모리에서 읽지 않고 바로 캐시에서 가져옴으로 성능적으로 큰 이득을 볼 수 있다. 이런 멀티코어 환경에서 각 코어들에 있는 cache들의 일관성을 유지하는 것을 Cache Coherence라고 말한다.
만약에 Core 1에서 메모리 주소 X에 있는 값을 읽기 위해 먼저 메모리에서 읽고 이를 Core 1 캐시에 저장하였다. 다음으로 Core 2에서 메모리 주소 X에 있는 값을 읽기 위해 메모리에서 이를 읽고 이를 Core 2의 캐시에 저장하였다. 만약 Core 1에서 add 연산으로 해당 변수를 원래 값인 1에서 5로 증가시켰다고 해보자. 그러면 Core 1의 캐시는 5로 업데이트 된다. 여기서 Core 2가 이 변수를 읽으면 무슨 값이 반환되어야 할까? 1일까 5일까?
Cache Coherence는 캐시에서 공유하고 있는 데이터의 값의 변경사항이 적시에 시스템 전체에 전파될 수 있도록 하는 원칙이다.
Cache Coherence는 다음 2가지가 필요하다.

• Write Propagation(쓰기 전파)
  어떠한 캐시에 데이터가 변경이 되면 이 cache line을 공유하고 있는 다른 캐시에도 이 변경사항이 전파되어야 한다.
• Transaction Serialization
  특정 메모리 주소로의 read/write은 모든 프로세서에게 같은 순서로 보여야 한다.

두번째의 Transaction Serialization은 다음 예를 보면 이해하기 쉽다.
Core 1,2,3,4 가 있을때 이들 모두 초기값이 0인 변수 S의 캐시된 복사본을 각 캐시에 가지고있다. 프로세서 P1은 이 S의 값을 10으로 변경한다. 그리고 프로세서 P2가 이어서 이 S의 값을 20으로 변경한다. 위의 Write Propagation를 보장한다면 P3와 P4가 이 변경사항을 볼 수 있다. 다만 프로세서 P3는 P2의 변경사항을 본 후, P1의 변경사항을 봐서 변수 S의 값으로 10을 반환받는다. 그리고 프로세서 P4는 원래의 순서에 따라 P1의 변경사항을 보고, P2의 변경사항을 그 다음으로 봐서 20을 반환받는다. 결국 프로세서 P3, P4는 캐시의 일관성을 보장할 수 없는 상태가 되었는데 이처럼 Write Propagation 하나만으로는 Cache Coherence가 보장이 안된다.
이를 위해 변수 S에 대한 Write는 반드시 순서가 지정이 되어야한다. Transaction Serialization이 보장이 된다면 S는 위의 예제에서 10으로 write하고 그리고 20으로 write 했기 때문에, 절대 변수 S에 대해 값 20으로 읽고 그다음 값 10으로 읽을 수가 없다. 반드시 값 10으로 읽고 그 다음 20으로 읽는다.

이처럼 Cache Coherence를 유지하기 위해서는 다른 프로세서에서 갱신한 캐시 값을 곧바로 반영을 하든 지연을 하든 해서 다른 프로세서에서 사용할 수 있도록 해주어야 한다. 캐시 일관성을 유지하기 위한 다양한 프로토콜들이 존재하며 대표적으로 MESI 프로토콜이 있다.

Cache Coherence에 대한 내용은 여기까지만 보도록 하고 cache line 이라는 것을 알아보자.

Cache Line

메인 메모리의 내용을 읽고 캐시에 이를 저장하는 과정에서 메모리를 읽을때에 이를 읽어들이는 최소 단위를 Cache Line이라고 한다. 메모리 I/O의 효율성을 위해서이며 spatial locality(공간 지역성)을 위해서이다. 보통의 cache line은 64byte 혹은 128byte로 이루어져 있으며 위에서 설명한 Cache Coherence도 cache line의 단위로 작동한다. 이렇게 cache line으로 읽어들인 데이터들로 캐시의 data block을 구성하게 된다.
또 cache line은 고정된 주소단위(보통은 64byte)로 접근하고 가져온다. 예를들면 다음과 같다.

1
2
3
4
5
6
7
8
9
10
11

+---------+-------+-------+
| address | x1000 | int a |
+---------+-------+-------+
| address | x1004 | int b |
+---------+-------+-------+
| address | x1008 | int c |
+---------+-------+-------+
| address | x100B | int d |
+---------+-------+-------+
| ....... | ..... | ..... |
+---------+-------+-------+

위와 같은 메모리 구조가 있다고 할때 변수 a를 읽을때는 주소 x1000부터 cache line의 크기인 64byte만큼 가져오고, 변수 c를 읽을때에는 주소 x1008부터 64byte를 읽는게 아니다. 고정된 주소단위로 변수 c를 읽을때에도, write를 할때에도 주소 x1000으로 읽는다는 의미이다.

이제 cache line을 알았으니 다시 Memory False Sharing으로 돌아가자.

Memory False Sharing

Memory False Sharing은 동일한 cache line을 공유할때 Cache Coherence로 인해 성능이 느려지는 안티패턴을 의미한다. 위에서 본 예제로 다시 이해해 보자.

1
2
3
4
5
6
7
8
9
10
11

+---------+-------+-------+
| address | x1000 | int a |
+---------+-------+-------+
| address | x1004 | int b |
+---------+-------+-------+
| address | x1008 | int c |
+---------+-------+-------+
| address | x100B | int d |
+---------+-------+-------+
| ....... | ..... | ..... |
+---------+-------+-------+

메모리 구조가 위와 같을때 스레드 2개가 있고 스레드 1은 int 변수 a를 1씩 계속 더하는 일을 하고, 스레드 2는 int 변수 c를 1씩 계속 더하는 일을 한다고 해보자.

• Thread 1: while (true) { a++ }
• Thread 2: while (true) { c++ }

더하기를 시작하기 전 이미 해당 cache line이 캐시에 올라와있다면 CPU 캐시의 상태는 다음과 같을 것이다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

Core 1 Cache
+-------------+---------+----------------------+
| mem address | invalid | data block (64 byte) |
+-------------+---------+----------------------+
|    .....    |  .....  | .................... | 
+-------------+---------+----------------------+
|    x1000    |  false  | a | b | c | d | .... | 
+-------------+---------+----------------------+
|    .....    |  .....  | .................... | 
+-------------+---------+----------------------+

Core 2 Cache
+-------------+---------+----------------------+
| mem address | invalid | data block (64 byte) |
+-------------+---------+----------------------+
|    .....    |  .....  | .................... | 
+-------------+---------+----------------------+
|    x1000    |  false  | a | b | c | d | .... | 
+-------------+---------+----------------------+
|    .....    |  .....  | .................... | 
+-------------+---------+----------------------+

상황을 쉽게 이해하기 위해 두 스레드는 서로 다른 프로세서에서 실행되지만 시간상으로 볼 때 서로 사이좋게 번갈아 가며 add를 한다고 해보자.

1. Thread 1: a++
2. Thread 2: c++
3. Thread 1: a++
4. Thread 2: c++ 이런 순서대로 실행이 된다고 하자. 먼저 1번의 a++가 발생했을때는 Core 1의 cache에서 a에 해당하는 부분이 1을 증가시킨 값으로 write가 일어나게 된다. 하지만 여기서 문제가 발생한다. 바로 다음 2번이 실행되기를 원하지만 그 사이에는 많은 일이 발생한다. 1번을 실행하였을때 Core 1 Cache의 data block 값이 변하였고, Cache Coherence protocol에 의하여 2가 실행되기 전에 하드웨어 병목이 생긴다. MESI protocol에 의해 Core 2의 해당 cache line의 상태가 invalid 상태로 바뀌고 Core 2가 다시 데이터를 읽으려면 해당 cache line이 invalid 이기 때문에 Core 1에서 읽거나 해야한다.
   즉 cache line단위로 관리되기 때문에 Thread 2는 변수 a와는 전혀 상관이 없는 작업임에도 불구하고 변수 a에 대한 변경때문에 성능저하가 급격하게 나타나게 된다.
   두 변수 a와 c가 서로는 전혀 상관이 없는 데이터임에도 불구하고 같은 cache line에 있기때문에 CPU는 특정 변수가 변경될때마다 캐시 일관성을 맞추기 위해 작업을 하게된다. 이는 성능하락으로 이어진다.

어떻게 해결할 수 있을까?

어떻게하면 이를 해결할 수 있을까?
일종의 cache line size에 맞추어 padding을 넣어 서로 다른 cache line에 속하게할 수 있다. 예는 다음과 같다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

Core 1 Cache
+-------------+---------+----------------------+
| mem address | invalid | data block (64 byte) |
+-------------+---------+----------------------+
|    .....    |  .....  | .................... | 
+-------------+---------+----------------------+
|    x1000    |  false  | a |     padding      | 
+-------------+---------+----------------------+
|    .....    |  .....  | .................... | 
+-------------+---------+----------------------+

Core 2 Cache
+-------------+---------+----------------------+
| mem address | invalid | data block (64 byte) |
+-------------+---------+----------------------+
|    .....    |  .....  | .................... | 
+-------------+---------+----------------------+
|    x1040    |  false  | c |     padding      | 
+-------------+---------+----------------------+
|    .....    |  .....  | .................... | 
+-------------+---------+----------------------+

이처럼 변수뒤에 padding을 붙여줌으로서 서로 다른 cache line에 속하게 하면 위 같은 False Sharing 문제를 해결할 수 있다.
C++에서는 alignas 함수를 사용하여 padding을 넣어줄 수 있다.

1
2

alignas(64) int a = 0;
alignas(64) int c = 0;

자바도 이와 비슷한 방법으로 자바 8부터 @jdk.internal.vm.annotation.Contended 라는 어노테이션을 지원한다.
먼저 클래스 내부필드에 어노테이션을 적용하는 방법을 알아보자. 클래스 내부 필드에 이를 적용하게 되면 해당 필드는 앞뒤로 empty bytes로 패딩을 추가함으로서 object 안의 다른 필드들과 다른 cache line을 사용하도록 해준다.

1
2
3
4
5

public class Counter1 {
    @jdk.internal.vm.annotation.Contended
    public volatile long count1 = 0;
    public volatile long count2 = 0;
}

@Contended에는 group tag라는 것도 지원하는데 이 group tag는 필드단위에 적용되었을때에만 작동한다. Group은 서로 다른 모든 그룹과 독립된 cache line을 가지게 된다.

1
2
3
4
5
6
7
8
9
10

public class Counter1 {
    @jdk.internal.vm.annotation.Contended("group1")
    public volatile long count1 = 0;

    @jdk.internal.vm.annotation.Contended("group1");
    public volatile long count2 = 0;

    @jdk.internal.vm.annotation.Contended("group2");
    public volatile long count3 = 0;
}

위의 예처럼 group tag를 지정해주면, count1 변수와 count2 변수는 같은 그룹으로 지정이 되어있고 count3는 다른 그룹으로 지정되어있다.
이런 경우 count1과 count2는 count3과는 다른 cache line을 가지게 되며 count1과 count2는 그룹이 같으므로 같은 cache line으로 될 수 있다.

Contended 어노테이션은 클래스에도 적용할 수 있는데, 클래스에 적용하게되면 모든 field들이 같은 group tag를 가지는 것과 동일하다. 하지만 JVM 구현체에 따라서 다른 isolation 방법을 사용할 수 있다. 전체 object를 isolate 기준으로 할수도 있고 각 field 들을 isolate 기준으로 할수도 있다. (HotSpot JVM 기준으로는 class에 Contended 어노테이션이 적용되어있다면 모든 field 앞에 padding을 적용하는 것 같다. implementation in HotSpot JVM)

1
2
3
4
5

@jdk.internal.vm.annotation.Contended
public class Counter1 {
    public volatile long count1 = 0;
    public volatile long count2 = 0;
}

Contended 어노테이션은 이 용도에 맞게 각 object들이 서로 다른 스레드에서 접근하는 상황일때 사용하면 성능향상을 가져올 수 있을것이다.
실제 Contended 어노테이션은 ConcurrentHashMap 구현이나 ForkJoinPool.WorkQueue 등에서 사용하고 있다.

Reference

• https://en.wikipedia.org/wiki/False_sharing
• http://shumin.co.kr/comp-arch-false-sharing/
• https://www.baeldung.com/java-false-sharing-contended

이번 글은 자바 7에 도입된 Fork/Join Framework에 대한 내용입니다.

Fork/Join Framework

자바 7에는 Fork/Join Framework가 도입되었는데 이는 ExecutorService의 구현체로서 이를 활용하면 작업들을 멀티코어를 사용하도록 작업할 수 있습니다. 기본적으로 Fork/Join은 하나의 병렬화할 수 있는 작업을 재귀적으로 여러개의 작은 작업들로 분할하고 각 subtask들의 결과를 합쳐서 전체 결과를 반환합니다.
Fork/Join은 divide-and-conquer 알고리즘과 굉장히 비슷하다. 다만 Fork/Join Framework는 한가지 중요한 개념이 있는데 이상적으로는 worker thread가 노는경우가 없다. 왜냐하면 Fork/Join Framework에서는 work stealing이라는 기법을 사용해서 바쁜 worker thread로 부터 작업을 steal, 즉 작업을 훔쳐온다.
먼저 ForkJoin의 thread pool에 있는 모든 thread를 공정하게 분할한다. 각각의 스레드는 자신에게 할당된 task를 포함하는 double linked list를 참조하면서 작업이 끝날때마다 queue의 헤드에서 다른 task를 가져와서 처리한다. 다만 아무리 공정하게 태스크들을 분할한다고 해도 특정 한 스레드는 다른 스레드보다 자신에게 할당된 태스크들을 더 빠르게 처리 할 수 있는데, 이렇게 자신에게 주어진 태스크들을 다 처리해서 할일이 없어진 스레드는 다른 스레드의 queue의 tail에서 작업을 훔쳐(steal)온다. 모든 태스크가 다 끝날때까지 이 과정을 반복하여 스레드간의 작업부하를 균등하게 맞출 수 있다.

ForkJoinPool

java.util.concurrent.ForkJoinPool은 위에서 설명한 work stealing 방식으로 동작하는 ExecutorService의 구현체이다. 우리는 ForkJoinPool의 생성자로 작업에 사용할 processor number를 넘겨줌으로서 병렬화 레벨을 정할 수 있다. 기본값은 Runtime.getRunTime().availableProcessors() 결과로 결정된다. 또 다른 특징으로는 ExecutorService들의 구현체와는 다르게 ForkJoinPool은 모든 워커 스레드가 데몬스레드로 명시적으로 program을 exit할 때 shutdown을 호출할 필요가 없다. ForkJoinPool의 내부에서 worker thread를 등록하는 과정에서 daemon 스레드로 설정한다.

ForkJoinTask

java.util.concurrent.ForkJoinTask는 ForkJoinPool에서 실행되는 task의 abstract class이다. ForkJoinTask<V>는 Future<V>를 구현한다. ForkJoinTask는 일종의 light 한 스레드라고 생각하면 쉽다. 여러개의 task 들이 생성되면 이들은 ForkJoinPool의 설정된 스레드들에 의해 실행되게 된다.
RecursiveAction와 RecursiveTask<R>가 ForkJoinTask의 서브클래스들인데 이들 또한 abstract class들이다. 그래서 이들을 구현한 서브클래스를 만들어서 사용한다. RecursiveAction과 RecursiveTask<R>의 차이점은 RecursiveAction은 태스크가 생성하는 결과가 없을때 사용하고 결과가 있을때에는 RecursiveTask<R>을 사용한다. 두 클래스 모두 abstract method인 compute() 를 구현해야한다.

ForkJoinTask는 현재 실행상태를 확인하기 위한 몇가지 메서드를 제공한다.
isDone()은 태스크가 완료되었는지의 여부를 반환한다. isCompletedNormally()는 태스크가 cancellation이나 exception 없이 완료되었는지의 여부를 반환하고 이외에도 isCancelled(), isCompletedAbnormally() 등의 메서드를 제공한다.

RecursiveTask 활용

스레드 풀을 이용하기위해 RecursiveTask<R>의 서브클래스를 만들어보자. parameter type R은 결과 형식을 의미한다. 우리는 RecursiveTask의 compute 메서드를 구현해야 한다.
protected abstract R compute();
compute 메서드는 태스크를 서브태스크로 분할하는 로직과 더이상 분할할 수 없을때 개별 서브태스크의 결과를 생산할 알고리즘을 정의한다. 따라서 대부분의 compute 메서드의 구현은 다음과 같다.

1
2
3
4
5
6
7
8

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

그렇다면 1부터 N까지의 합을 구하는 프로그램을 Fork/Join Framework를 사용하여 작성해보자. 코드는 다음과 같다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

import java.util.concurrent.RecursiveTask;

public class ForkJoinSumCalculator extends RecursiveTask<Long> {
    private final long[] numbers;
    private final int start;
    private final int end;
    private static final int THRESHOLD = 10_000;

    public ForkJoinSumCalculator(long[] numbers) {
        this(numbers, 0, numbers.length);
    }

    private ForkJoinSumCalculator(long[] numbers, int start, int end) {
        this.numbers = numbers;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        int size = end - start;
        if (size <= THRESHOLD) {
            return computeSequentially();
        }

        ForkJoinSumCalculator leftTask = new ForkJoinSumCalculator(
                numbers, start, start + size / 2);
        leftTask.fork();
        ForkJoinSumCalculator rightTask = new ForkJoinSumCalculator(
                numbers, start + size / 2, end);
        long rightResult = rightTask.compute();
        long leftResult = leftTask.join();
        return leftResult + rightResult;
    }

    private long computeSequentially() {
        long sum = 0;
        for (int i = start; i < end; i++) {
            sum += numbers[i];
        }

        return sum;
    }
}

그리고 다음과 같이 ForkJoinPool의 invoke 메서드를 사용해 실행시켜보자.

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.stream.LongStream;

public class Main {
    private static final long N = 30_000_000L;

    public static void main(String args[]) {
        long[] numbers = LongStream.rangeClosed(1, N).toArray();
        ForkJoinTask<Long> task = new ForkJoinSumCalculator(numbers);
        long sum = new ForkJoinPool().invoke(task);
        System.out.println(sum);
    }
}

Fork/Join을 사용할 때 왼쪽 작업과 오른쪽 작업에 모두 fork를 호출하는게 자연스러운것 처럼 보이지만 한쪽에는 fork를 호출하는 것 보다 compute를 호출하는게 더 효율적이다. 한 태스크에는 이 Fork/Join 스레드를 실행시킨 스레드를 재사용할 수 있으므로 불필요한 태스크를 다른 스레드에 할당하는 오버헤드를 피할 수 있다.
또 멀티코어에 Fork/Join을 사용하는게 무조건 순차처리보다 빠르지 않다. 각 서브태스크의 실행시간이 새로운 태스크를 forking하는데 드는 시간보다 충분히 길수록 좋다.

위의 예제에서는 덧셈을 수행할 숫자가 만개 이하면 분할을 더이상 하지 않고 계산했다. 그러면 현재는 태스크가 3천개가 생성되는데 어차피 코어의 수는 정해져있으므로 코어가 3개라면 각 코어마다 1천만개씩 덧셈을 수행하면 딱 알맞게 효율적으로 동작하지 않을까?
그렇지는 않다. 실제로는 코어 개수와 관계없이 적절하게 작은 크기로 분할된 많은 태스크를 forking 하는것이 바람직하다. 1천만개씩 덧셈을 수행하도록 한다고 해도 각 3개의 코어에서 이루어지는 작업이 동시에 끝나지는 않는다. 각 태스크에서 예상치못하게 지연이 생길 수 있어 작업완료시간이 크게 달라질 수 있다. 다만 Fork/Join Framework는 work-stealing 기법으로 idel한 스레드는 다른 스레드의 workQueue로 부터 작업을 훔쳐오기 때문에 모든 스레들에게 작업을 거의 공정하게 분할할 수 있다. 그러므로 태스크의 크기를 작게 나누어야 스레드 간의 작업부하 수준을 비슷하게 맞출 수 있다.

References

• https://howtodoinjava.com/java7/forkjoin-framework-tutorial-forkjoinpool-example/
• https://docs.oracle.com/javase/tutorial/essential/concurrency/forkjoin.html