Spring Integration - ThreadPoolExecutor with udp-inbound-channel-adapter

최근 *** 프로젝트와 관련하여 여러 사이트에서 Out Of Memory Error 가 발생했다.

heap space 에러가 가장 많은데 사실 단순한 프로그래밍 실수라면 dump 파일 분석으로 쉽게 처리할 수 있는 것이 OOME 이다.

이번에 dump 파일을 분석 해 보니 아래와 같이 "java.util.concurrent.ThreadPoolExecutor"가 전체 

heap space 의 90% 정도를 차지하고 있었다. 

해당 프로젝트는 엄청나게 많은 양의 패킷을 모두 받아 특정 정보가 포함되어 있는 패킷들을 선별하여 모니터링 할 수 있는데 

Spring Integration 의 udp-inbound-channel-adapter 를 이용하여 패킷을 수신하도록 개발 되어 있다. 

 - udp-inbound-channel-adapter는 https://kimyhcj.tistory.com/114 에서 확인 가능하다.

문제를 해결하기 위해서 먼저 Task 가 무엇인지 간단하게 살펴본다.

Spring은 TaskExecutor interface와 TaskScheduler Interface로 Task의 비동기 시행과 스케쥴링에 대한 추상화를 제공한다. 

(JDK 1.3부터 추가된 Timer와 Quartz를 이용해서 사용할 수 도 있다.)

Spring의 TaskExecutor Interface는 java.util.concurrent.Excecutor 와 같다. 우리는 concurrent 패키지를 이용하여 자체적인 Thread Pool을 만들지 않고 

안정적으로 Thread를 Pooling 한다. 

TaskExcecutor를 구현하는 구현체는 SimpleAsyncTaskExecutor, SyncTaskExecutor, ConcurrentTaskExcecutor 등 많다.

필요에 따라 선택해서 사용하면 된다. (https://docs.spring.io/spring/docs/3.1.x/javadoc-api/org/springframework/core/task/ 참고)

udp-inbound-channel-adapter 로 들어온 패킷들은 각각의 task 를 통해 별도의 thread 에서 처리 된다. 

이 프로젝트에서는 무작위로 오는 패킷을 계속 수신하고 패킷이 완성 되더라도 5초간 대기하고 있다가 

(추가로 들어오는 패킷이 없는지 대기) 완성형 패킷인 경우 별도의 처리를 위한 Class 로 넘기고 task는 종료가 되는 로직을 가지고 있다. 

그리고 task executor 는 아래와 같이 설정되어 있었다.

<taskLexecutor id="testExecutor" pool-size="200" queue-capacity="50" rejection-policy="CALLER_RUNS" />

총 200개의 thread pool 을 가지고 있고 각각의 thread는 주어진 task 를 처리하게 된다. 

thread 들이 모두 바쁜 경우 대기 큐로 들어가게 되며 최대 50개 까지 큐에 쌓이도록 설정되어 있다.

thread 가 모두 바쁘고 대기 큐도 모두 가득차 있는 경우는 설정되어 있는 rejection-policy 에 따라 동작한다.

default 는 AbortPolicy (예외 발생 시킴) 이고 위에서 설정한 CallerRunsPolicy, DiscardPolicy, DiscardOldestPolicy 로 4가지 정책을 지원한다.

https://docs.oracle.com/javase/6/docs/api/java/util/concurrent/ThreadPoolExecutor.html

다시 문제로 돌아가보면 모든 패킷을 수신하여 처리하는 task 는 바쁘고 (또는 모두 처리 했는데 5초간 대기하면서 pool로 반환하지 않았고) 대기 큐도 가득차 있는 상태에서 Caller Runs 정책에 의하여 execute 자체를 호출하는 스레드가 작업을 수행하도록 되면서 제한된 heap space에서 overflow가 발생한것으로 예상할 수 있다.

이 문제를 해결하기 위해서는 다양한 방식으로 접근해야 할 필요가 있다.

1. 패킷 완성 후 대기시간 (5초)이 필요한지 여부 판단
 - 프로젝트 성격 상 당연히 필요하다.
 - 다만, 5초는 너무 길다. 
 - 패킷량, 완성형 패킷의 완성되는 속도 등을 측정하여 적당한 시간을 찾아 설정할 수 있도록 해야 한다. 

2. pool-size와 queue-capacity를 늘리고 heap space 도 늘린다.
 - dump 파일을 분석해 보면 하나의 task 가 가지고 있는 byte size를 알 수 있다. 
 - 정해진 프로토콜에 의해 전달되는 패킷이기 때문에 패킷량과 task 의 pool-size 를 계산해 보면 heap space를 얼마나 늘려야 할지 계산이 나온다. 
 - 다만, heap space를 늘리게 되면 full gc 시에 문제를 일으킬 가능성이 높아지기 때문에 (stop the world) 이 부분은 조심스럽게 접근해야 한다. 

3. rejection-policy를 DiscardPolicy 또는 DiscardOldestPolicy 이용
 - 모든 패킷을 처리해야 하는 상황이 아닌 경우 과감하게 버리는 것도 방법이 될 수 있다. 
 - 위에서 말한 1번 2번 방법을 모두 진행했음에도 해결이 되지 않을 경우 최후의 수단으로 해결 가능한 방법으로 보인다.