ForkJoinPool 좀 보기!

최근에 4월 우아한 개발자 tech에서 virtual thread 관련 발표를 했었어용.

해당 기술 설명에 ForkJoinPool 이야기가 나왔는데, 나온 김에 정리하려고 합니다!

 

ForkJoinPool을 개발자가 직접적으로 다룰 일은 많지 않겠지만, 언급했던 virtual thread나 혹은 parallel stream에 default로 사용되기 때문에 개념적으로 알고 있는게 좋을거 같아요! 

 

사실 해당 개념에 대한 좋은 글은 널리고 널렸기 때문에 간략하게 제 공부용으로 포인트만 남기도록 하겠습니다 ㅎ.ㅎ

---

 

ForkJoinPool ? 

 

얘도 말 그대로 pool 이에요 ..

thread pool의 일종입니다. pool 안에 있는 각각의 thread가 하나의 worker가 되는거겠죠??

 

기존에 저희가 알던 일반 thread pool과는 어떤게 다를까요??

 

가장 큰 차이는 work-stealing과 worker마다 존재하는 덱이 아닐까 합니다.

 

덱?????????????????

- 큐는 한 방향에서 꺼내고 반대 방향에서는 넣는 구조를 가지고 있는데,

  덱은 큐의 소유자와 task를 스틸해가는 워커가 동시에 접근할 수 있도록,

  양 쪽 모두에서 꺼낼 수 있는 구조를 가진 자료구조에요!

 

ExecutorService

- ExecutorService 를 구현한 Class : ThreadPoolExecutor / ForkJoinPool 
                                                          :: thread pool을 관리하고, task를 수행하는 worker들을 관리.

 ThreadPoolExecutor는 Executors.newFixedThreadPool( params... ) ( 고정된 thread 개수를 가짐 )

                                       Executors.newCachedThreadPool() ( 어플리케이션의 요청 유입이 일정하지 않을 때 유용 )
                                       Executors.newScheduledThreadPool( params... ) ( 일정 주기를 반복하며 실행되는 task에 유용 ) 들을 통해 생성할 수 있다.

 

ForkJoinPool은 new 생성자를 통해서도 생성이 가능하지만. Executors.newWorkStealingPool( params.. )를 통해서도 생성이 가능하다. Executors 클래스를 통해 생성 시 asyncMode가 true로 생성된다.

 

 

** @param asyncMode

if true, establishes local first-in-first-out
* scheduling mode for forked tasks that are never joined. This
* mode may be more appropriate than default locally stack-based
* mode in applications in which worker threads only process
* event-style asynchronous tasks.  For default value, use {@code
* false}

 

ForkJoinPool은 work-stealing을 통해 task를 fork해요.

 

예를 들어, 임의의 worker A가 global queue를 통해서 작업 1을 가져왔어요.

해당 작업은 꽤나 큰 크기의 작업이라 A 전용 딕에 분할 ( fork ) 되어 들어가게 됩니다.

A가 자신의 일을 처리하고 있는 와중, 금방 본인의 일을 마친 woker B가 자신이 할 일이 없게 되면, 일이 남아 돌고있는 worker A의 일을 가져가다 본인이 처리하게 됩니다. ( work-stealing )

 

 

이런 work-stealing은 어디에 구현되어 있을까요?

> ForkJoinPool이 처리하는 ForkJoinTask 에 구현되어 있습니다.

ForkJoinPool은 Callable이나 Runnable 을 구현한 task도 처리가 가능하지만, 

ForkJoinTask를 구현한 RecursiveTask 혹은 RecursiveAction을 처리할 수도 있습니다.

 

 

ForkJoinPool은 스레드 풀에 분할-정복 알고리즘을 적용한 것과 같습니다. 실 활용 코드들을 보면 재귀함수를 사용한 것과 형태가 동일한 것이 많아요. 또한 재귀함수의 활용은 함수형 프로그래밍과 패러다임이 일치하죠!

 

이렇게 처리했을 때 장점이 뭘까요???

1. 스레드들의 처리량을 극대화 할 수 있다. 놀고있는 스레드가 없다는 의미에요.
2. task를 병렬처리 할 수 있다. 
   1. 하나의 task를 여러 worker가 처리하기 때문에 병렬처리를 할 수 있어요. ( 마냥 그렇다는 건 아니고 놀고 있는 스레드가 있어서 work-stealing을 하는 경우에 )

 

물론 단점도 있습니다.

 

ForkJoinPool은 별도의 설정이 없을 경우, commonForkJoinPool을 사용하게 되어있어요.

static 이라고 생각해도 될거 같아요. 위에서 언급한 parallel Stream을 사용할 경우, 내부적으로 ForkJoinPool을 사용한다고 했는데, 몇 개의 병렬 스트림을 생성해서 처리해도, 하나의 commonForkJoinPool을 공유해서 사용하게 된다는 의미입니다.

 

그래서 만약 병렬 스트림으로 처리하는 부분이 매우 많거나, 처리에 오랜 시간이 소요되는 task들이 병렬로 처리된다면, 하나의 스레드 풀에서 작업이 처리되기 때문에, 대기현상이 발생할 수 있어요. 대기 하는 task가 많아질 경우, 심하면 drop이 되겠죠 ㅠ-ㅠ...

이럴 경우, commonForkJoinPool이 아닌, customForkJoinPool을 new를 이용해서 생성하면 작업 별로 별개의 스레드 풀을 할당할 수 있어요.

 

그러나 이렇게 new를 이용해서 customForkJoinPool을 사용할 경우, 해당 풀을 더 이상 사용하지 않아도, 객체 참조가 해제되지 않아 GC로 적절하게 처리되지 않아서, OOM이 발생할 위험이 있습니다.

new로 생성한 경우에, forkJoinPool.shutdown() 을 통해 직접 해제하여아 안전합니다.

 

스레드 풀 생성 시, 보통 param에 생성할 thread의 개수를 인자로 받는데, 스레드가 많으면 여러 일을 동시에 처리할 수 있어 처리율이 올라갈 것 같지만, 자바는 하나의 유저스레드 당 하나의 커널 스레드를 할당하기 때문에, 오히려 컨텍스트 스위칭에 리소스가 더 낭비될 수 있어요. 

 

개발자가 임의로 스레드 풀을 생성하여 사용할 때, 가장 best practice는 cpu core의 개수만큼 thread를 생성하는 것입니다.

( commonForkJoinPool의 스레드 수를 설정으로 변경하면 전체 성능 상 영향을 끼칠수 있어 사용하지 않는 것이 좋습니다. )

 

이때문에, virtual thread가 나오고 core수를 늘리기 위해 cpu의 병렬 처리를 통해 논리적 core수를 늘리는 등의 방법이 있는데, 이는 후에 작성하도록 하겠습니다 :)

 

그 다음 단점은,  일반 스레드 풀에 비해서 task를 분할해야 하고 후에 결과를 다시 Join해야 하는 작업으로 인해 subtask를 만들어내는 비용이 발생해 리소스가 더 많이 들 수 있어요.

 

 

제가 사용하고 있는 mac의 core가 8이기 때문에 ForkJoinPool 생성 시 thread 수를 8로 지정하였습니다.

총 소요 시간은 commonForkJoinPool이 3초, maybeIndividualForkJoinPool이 4초가 소요되었습니다.

task마다 스레드 풀을 구성해야 하니 어찌보면 당연한 결과 입니다.

 

 

commonForkJoinPool 의 실행결과 : 

 

하나의 commonPool에서 모든 task를 처리하는 것을 볼 수 있습니다.

 

 

 

 

maybeIndividualForkJoinPool 의 실행결과 : 

각기 다른 스레드 풀에서 처리되는 것을 볼 수 있습니다.

서로 상이한 스레드 풀의 워커들이 처리하는 것이 보이시죠??

 

 

 

 

그리고 생각을 해보면, 작업을 분할하고 다시 결과를 합쳐야하기 때문에, 분할 된 subtask들 사이에 연관관계 혹은 전후관계가 없는 것이 좋고, sorting이나 distinct 작업에는 적절하지 않습니다. 이는 parallel stream의 특성과도 유사할 수 있습니다. 

 

 

 

ForkJoinPool's get() vs join()

ForkJoinPool이 제공하는 메소드를 한 번 볼까요???

위 코드에서 보이는 바와 같이 submit() 이라는 인터페이스가 존재하는데, 이는 task를 thread pool에 맡기는 작업을 의미합니다.

실제 task가 처리되는 것은 아니에요!

 

task 처리와 관련된 메소드는 크게 get과 join 두 가지가 있습니다.

 

그냥 코드 살펴보다가 본건데, 둘의 가장 큰 차이가 뭘까요???

 

task 처리중 발생한 exception을 핸들링 할거냐 말거냐가 가장 큰 차이입니다.

 

보통 메소드 시그니처에 throw xxxException 이렇게 작성하면, 해당 메소드에서 발생한 Exception을 checkedException으로 취급해서 별도로 다룬다는 것을 명시하잖아요??

 

해당 메소드들에도 고렇게 작성이 되어있습니다!

 

join() :

 

join 입니다. 예외가 발생할 경우 repostException()을 던지는데, 해당 메소드를 타고 들어가면,

 

요렇게 나옵니다. 즉 uncheckedException을 발생시킬 것이고, 해당 예외는 따로 핸들링 되지 않은 채 그냥 예외로 타고타고 올라가게 되는 것이죠.

 

 

get() :

 

대문짝만하게 시그니처에 throws가 적혀있습니다. 두 가지 예외를 checkedException으로 사용하겠다는 의미입니다.

get을 사용할 경우 당연히 try - catch 구문으로 잡아줘야 합니다.

 

 

 

 

★☆ 적절한 사용 ★☆

• I/O burst 보다 CPU burst application 에 더 적합하다. 
  • I/O burst에서 성능 적 장점을 누리기가 어렵다.
• parallelStream은 처리하고자 하는 데이터의 양이 매우매우 클 때 유용하다. 
  • 처리할 데이터의 양이 많지 않은 경우 오히려 성능 상 불리하다.
• stream()의 distinct() 나 sorted() 는 내부적으로 처리하면서 공유 변수를 사용하고, 여러 스레드가 동시에 접근하지 못하도록 synchronized 처리가 되어있다. 그래서 이 경우 오히려 순차처리를 하는 것이 성능상 더 유리하다.
• ForkJoinPool에서는 큰 단위의 task를 일정한 단위의 subtask로 분할해서 처리하기 때문에, 분할에 유리한 자료구조에 적용하는 것이 좋다.
  • 균등하게 분할해야 하기 때문에 길이를 알 수 있는 자료구조가 좋다.
  • Array, ArrayList, HashSet, TreeSet, IntStream.range() : 좋음
  • LinkedList, Stream.interate() :  나쁨 
• 여러 스레드로 하나의 task를 분할해서 처리하기 때문에, 순서가 중요한 일이라던지, 혹은 분할된 task 끼리 전후관계 / 어떤 상태 값을 공유하는 상태인 경우에는 ForkJoinPool에서 처리하기에 적절하지 않다.