Memory leak on connection loss

[kichik]

There seems to be a memory leak on reconnection. This seems to be related to a very specific way of the connection going down because I don't always see it and I only see it together with the state exception.

I'm using Netty 4.0.25 with pooled buffers -Dio.netty.allocator.type=pooled
To get the stack I use -Dio.netty.leakDetectionLevel=advanced
Latest lettuce 3.0.2.

2015-01-30 16:21:17,932 WARN io.netty.channel.DefaultChannelPipeline: An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
com.lambdaworks.redis.RedisException: Invalid first byte: 111
        at com.lambdaworks.redis.protocol.RedisStateMachine.readReplyType(RedisStateMachine.java:200)
        at com.lambdaworks.redis.protocol.RedisStateMachine.decode(RedisStateMachine.java:98)
        at com.lambdaworks.redis.protocol.RedisStateMachine.decode(RedisStateMachine.java:61)
        at com.lambdaworks.redis.pubsub.PubSubCommandHandler.decode(PubSubCommandHandler.java:55)
        at com.lambdaworks.redis.protocol.CommandHandler.channelRead(CommandHandler.java:89)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
        at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
        at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
        at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
        at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
        at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:130)
        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
        at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:116)
        at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
        at java.lang.Thread.run(Thread.java:744)
2015-01-30 16:21:17,935 ERROR io.netty.util.ResourceLeakDetector: LEAK: ByteBuf.release() was not called before it's garbage-collected.
Recent access records: 0
Created at:
        io.netty.buffer.PooledByteBufAllocator.newHeapBuffer(PooledByteBufAllocator.java:240)
        io.netty.buffer.AbstractByteBufAllocator.heapBuffer(AbstractByteBufAllocator.java:136)
        io.netty.buffer.AbstractByteBufAllocator.heapBuffer(AbstractByteBufAllocator.java:122)
        com.lambdaworks.redis.protocol.CommandHandler.channelRegistered(CommandHandler.java:59)
        io.netty.channel.AbstractChannelHandlerContext.invokeChannelRegistered(AbstractChannelHandlerContext.java:133)
        io.netty.channel.AbstractChannelHandlerContext.fireChannelRegistered(AbstractChannelHandlerContext.java:119)
        io.netty.channel.ChannelInboundHandlerAdapter.channelRegistered(ChannelInboundHandlerAdapter.java:42)
        io.netty.channel.AbstractChannelHandlerContext.invokeChannelRegistered(AbstractChannelHandlerContext.java:133)
        io.netty.channel.AbstractChannelHandlerContext.fireChannelRegistered(AbstractChannelHandlerContext.java:119)
        io.netty.channel.ChannelInboundHandlerAdapter.channelRegistered(ChannelInboundHandlerAdapter.java:42)
        io.netty.channel.AbstractChannelHandlerContext.invokeChannelRegistered(AbstractChannelHandlerContext.java:133)
        io.netty.channel.AbstractChannelHandlerContext.fireChannelRegistered(AbstractChannelHandlerContext.java:119)
        io.netty.channel.ChannelInboundHandlerAdapter.channelRegistered(ChannelInboundHandlerAdapter.java:42)
        io.netty.channel.AbstractChannelHandlerContext.invokeChannelRegistered(AbstractChannelHandlerContext.java:133)
        io.netty.channel.AbstractChannelHandlerContext.fireChannelRegistered(AbstractChannelHandlerContext.java:119)
        io.netty.channel.ChannelInitializer.channelRegistered(ChannelInitializer.java:71)
        io.netty.channel.AbstractChannelHandlerContext.invokeChannelRegistered(AbstractChannelHandlerContext.java:133)
        io.netty.channel.AbstractChannelHandlerContext.fireChannelRegistered(AbstractChannelHandlerContext.java:119)
        io.netty.channel.DefaultChannelPipeline.fireChannelRegistered(DefaultChannelPipeline.java:733)
        io.netty.channel.AbstractChannel$AbstractUnsafe.register0(AbstractChannel.java:449)
        io.netty.channel.AbstractChannel$AbstractUnsafe.access$100(AbstractChannel.java:377)
        io.netty.channel.AbstractChannel$AbstractUnsafe$1.run(AbstractChannel.java:423)
        io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:380)
        io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:357)
        io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:116)
        io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
        java.lang.Thread.run(Thread.java:744)

I'll share more information when I find it.

[mp911de]

I pushed 37c9423 to branch issue-26-memory-leak-fix. Can you give it a try? Deallocation happens using netty the same way resources are allocated.

[mp911de]

You can set the log level to DEBUG for the logger com.lambdaworks.redis.protocol (or even com.lambdaworks) to inspect the raw data. ASCII 111 is o which looks like some data was either in the buffer or the reply was not expected from redis. Guess, this will need further investigation.

[kichik]

I will give it a shot. Sadly I don't have a good way to reproduce yet.
The fix looks good and makes sense. The allocation/release is now symmetrical.
Other invalid bytes I had were 'l', 'o', 'u', 'C'.
This might be related to the way we wrap the connection in SSL.

[mp911de]

hm. Looks like a feature request :)

[kichik]

SSL wrapping? That might have actually helped avoid all of these issues. Lettuce would have known the connection is not really up and it's only the SSL tunnel that's up.

[mp911de]

Yep. Created issue #28

[kichik]

Another thing that could have helped avoid this is a more thorough handshake that verified Lettuce is talking to Redis on the other side before beginning to dispatch commands. Maybe a ping/pong sequence first.

[kichik]

No luck reproducing yet.

[kichik]

Still not able to reproduce the exact issue, but I was able to confirm this fix is good. Instead of bad bytes, I got plain old connection reset by peer. When Redis goes down and Lettuce reconnects close() is never called. I've also added assert buffer == null at the top of of channelRegistered() and without this fix it fires multiple times on reconnection.

[mp911de]

Setting this one to resolved. The SSL issue is addressed with issue #28

[mp911de]

3.0.3.Final released

[kichik]

Thanks! :)