ResourceLeakDetector reportLeak 오류가 발생

[snkim]

안녕하세요~

netty를 사용하여 서버 구현 중 ByteToMessageDecoder 를 이용한 frame decoder 구현 시 문제가 생겼습니다.

테스트 시 다음과 같은 오류 로그가 출력되는데 아직 실력이 부족하여 원인 파악을 하지 못하고 있습니다.

[오류 로그]

1월 27, 2014 12:00:57 오전 io.netty.util.ResourceLeakDetector reportLeak
심각: LEAK: ByteBuf.release() was not called before it's garbage-collected. Enable advanced leak reporting to find out where the leak occurred. To enable advanced leak reporting, specify the JVM option '-Dio.netty.leakDetectionLevel=advanced' or call ResourceLeakDetector.setLevel()

혹시나 제가 netty의 구조, 사용법을 잘못이해하고 있는것은 아닌지 하여 메시지의 구조와 처리방식, 소스코드를 올려봅니다.

먼저 메시지의 구조는 아래와 같습니다.

 | 헤더(62byte) | Data (가변) |

 

헤더의 최초 5바이트는 5바이트를 제외한 메시지 전체 길이를 나타냅니다.

또한 클라이언트에서는 메시지 전송 시 4KByte 씩 조각내어 전송을 합니다. (테스트 클라이언트는 header write, body write하도록 구현)

전송되는 메시지의 평균 크기는 40KByte ~ 60KByte 사이입니다.

일단 클라이언트에서는 하나의 메시지를 보내고 컨넥션을 끊을 수도 있고, 컨넥션을 유지하고 계속 메시지만 보낼 수도 있는 상황(일종의 스트리밍) 입니다.

frame decoder는 ReplayingDecoder를 참조하여 State를 만들어서 swtich, case를 이용하여 상태 별 처리를 하도록 하였습니다.

테스트 클라이언트는 컨넥션 생성 이후 62바이트의 헤더와 40KByte 정도의 데이터를 무한루프로 전송을 합니다.

위의 오류 로그는 160개의 메시지 전송 후 발생하였습니다.

--> 42,929 * 160 = 6,868,640 byte

혹시 session을 유지하고 데이터를 주고 받는 경우 pipeline을 통하여 수신되는 ByteBuf는 무한히 확장이 되는 것인가요? 아니면 읽어 들인 byte만큼 비워지며 유지가 되는지요?

channelRead 호출 시 ByteBuf의 내용을 internal buffer에 기록하고 ByteBuf는 release 되며 internal buffer는 callDecode가 호출되고 난 이후에 읽을 수 없는상황에 release 되는 것으로 보입니다.

제가 이해하고 있는게 맞는지요? 현제 제가 구현한 코드를 보자면 지속적으로 스트림 데이터가 유입되고 있는 상황이라 channelRead 호출 시 생성되는 internal buffer가 release 되지는 않을 것으로 보입니다.

ByteBuf의 처리를 잘못하여 위의 오류가 발생하는지 아니면 다른 이유로 발생하는지 아니면 로직 구현을 잘못하여 발생하는지 도저히 갈피를 못잡고 있습니다.

도움을 부탁드립니다.

감사합니다.

[소스 코드]

protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer)
   throws Exception {
  while(true) {
   switch(stat()) {
    case READ_INITIAL:
     if(buffer.readableBytes() < HEADER_CHECK_LENGTH) 
      return null;

     
     String length = buffer.readBytes(HEADER_CHECK_LENGTH).toString(Charset.defaultCharset());
     // 전문 길이 체크 전 마킹
     buffer.resetReaderIndex();
     
     try {
      messageLength = Integer.parseInt(length);
     } catch (NumberFormatException nfe) {
      checkpoint(Stat.INVALID_MESSAGE);
      continue;
     }
     
     checkpoint(Stat.READ_HEADER);
    case READ_HEADER:
     if(buffer.readableBytes() < HEADER_LENGTH)
      return null;
     
     byte[] headerData = new byte[HEADER_LENGTH + HEADER_CHECK_LENGTH];
     buffer.readBytes(headerData);

     offset = HEADER_LENGTH;
     checkpoint(Stat.READ_DATA_INITIAL);
    case READ_DATA_INITIAL:
     if(messageLength < FRAME_SIZE) // 4096 바이트
      checkpoint(Stat.READ_DATA_SHORT);
     else
      checkpoint(Stat.READ_DATA_LONG);

     continue;
    case READ_DATA_SHORT:
     if(buffer.readableBytes() < messageLength)
      return null;
     else {
      byte[] dataByte = new byte[messageLength - HEADER_LENGTH];
      buffer.readBytes(dataByte);
      checkpoint(Stat.COMPLETE);
      continue;      
     }
    case READ_DATA_LONG:
     byte[] dataByte;
     if(messageLength - offset > FRAME_SIZE) {
      if(buffer.readableBytes() < FRAME_SIZE)
       return null;
      else {
       dataByte = new byte[FRAME_SIZE];
       buffer.readBytes(dataByte);
       offset += FRAME_SIZE;
       chunkedCount++;
       return null;
      }
     } else {
      if(buffer.readableBytes() < (messageLength - offset))
       return null;
      else {
       dataByte = new byte[messageLength - offset];
       buffer.readBytes(dataByte);
       offset += messageLength - offset;
       chunkedCount++;
       checkpoint(Stat.COMPLETE);
       continue;
      }
     }
    case COMPLETE:
    System.out.println("COMPLETE :: Reader Index:" + buffer.readerIndex());
     // TODO 데이터부 마지막 | 확인하고 에러 처리할 것, 
     offset = 0;
     return new AJMSMessageFrame(); // 실제 패킷은 버리고 빈 객체를 생성하여 최종 handler에 전달
     
    case INVALID_MESSAGE: // 구현할 것
     return null;
   }
  }
 }

["이희승 (Trustin Lee)"]

> String length = buffer.readBytes(HEADER_CHECK_LENGTH).toString(Charset.defaultCharset());

readBytes(int) 에서 리턴한 버퍼를 릴리즈하지 않으셨네요. 저라면 readSlice(int) 를 이용하거나 하겠습니다.  그럼 릴리즈할 필요도 없고 메모리 할당도 적게 하게 됩니다.

구체적으로 어디서 leak 이 발생했는지 확인하려면 로그 메시지에 적힌 대로 JVM 기동시 '-Dio.netty.leakDetectionLevel=advanced' 옵션을 주시면 됩니다.

"Enable advanced leak reporting to find out where the leak occurred. To enable advanced leak reporting, specify the JVM option '-Dio.netty.leakDetectionLevel=advanced' or call ResourceLeakDetector.setLevel()"

--
Google 그룹스 'Netty Korean User Group' 그룹에 가입했으므로 본 메일이 전송되었습니다.
이 그룹에서 탈퇴하고 더 이상 이메일을 받지 않으려면 netty-ko+u...@googlegroups.com에 이메일을 보내세요.
더 많은 옵션을 보려면 https://groups.google.com /groups/opt_out을(를) 방문하세요.

-- 
https://twitter.com/trustin
https://twitter.com/trustin_ko
https://twitter.com/netty_project

["이희승 (Trustin Lee)"]

구체적인 누수 문제 해결 방법은 다음 페이지에 정리되어 있습니다:

http://netty.io/wiki/reference-counted-objects.html

>> <https://groups.google.com/groups/opt_out%EC%9D%84%28%EB%A5%BC>) 방문
>> 하세요.

[snkim]

이희승님 말씀대로 참고하여 해결하였습니다.

대단히 감사 드립니다~

아래 두 코드 때문에 문제가 생겼었네요.

-- 코드

Object decoded = decode(ctx, buffer);
  if (decoded != null) {
   out.add(decoded);
   decoded = null;
   reset();
  }

마지막 decoded 객체를 다음 핸들러로 전달하는 과정에서 decoded 객체를 날려버리는 코드가 있었는데,

앞서 작성한 코드 중 decoded 객체를 리턴할 때 생성한 객체의 생성 시기가 문제가 된 것 같습니다.

-- 코드