Optimize aggregation buffer cleanup in CommandHandler

[gavincook]

When use pressure test to get data from redis, the redis data value is 16KB. It will cause the out of direct memory.
Saw the source code, I found:
Now the method decode in io.lettuce.core.protocol.CommandHandler will try to decode message received from redis server. When faced partial response message(most under BULK scene), will try to continue receive more and then decode more. While current buffer end with a complete message, it will discard the read bytes.
This behavior will keep read bytes always until input buffer end with a complete message. Under my pressure test sense, will always get partial message.

Reproduction setup

1. first set a small direct memory to make it easy to reach the direct memory limit, like 50M with -XX:MaxDirectMemorySize=20m
2. then set the logger level for io.lettuce.core.protocol to DEBUG
   Use below test code:

package io.lettuce.core.protocol;

import io.lettuce.core.RedisClient;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import java.util.concurrent.TimeUnit;

public class MemoryLeakTest {

    public static void main(String[] args) throws InterruptedException {
        RedisClient redisClient = RedisClient.create("redis://localhost:6379/0");
        StatefulRedisConnection<String, String> connection = redisClient.connect();
        RedisAsyncCommands<String, String> syncCommands = connection.async();

        syncCommands.set("key", "quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string,"
            + "quiet long string,quiet long string,quiet long string,quiet long string,quiet long string")
                    .await(5, TimeUnit.SECONDS);

        long startTime = System.currentTimeMillis();
        while (System.currentTimeMillis() - startTime <= 30_000) {
            syncCommands.get("key");
        }
        connection.close();
        redisClient.shutdown();
    }

}

Then the exception occur:

2018-10-31 10:40:51 [DEBUG] [lettuce-kqueueEventLoop-4-1] [channel=0x2ac75c7d, /127.0.0.1:64284 -> localhost/127.0.0.1:6379, chid=0x1] Unexpected exception during request: io.netty.util.internal.OutOfDirectMemoryError: failed to allocate 4194304 byte(s) of direct memory (used: 16845793, max: 20971520) (CommandHandler:216)

Environment

1. redis : 4.0.11
2. lettuce : 5.1.2-RELEASE

[gavincook]

When decode failure, means end with a partial message. Now will just return to continue receive more.
The code:

 if (!decode(ctx, buffer, command)) {
     return;
 }

We should discard the read bytes when end with a partial message.

 if (!decode(ctx, buffer, command)) {
      if (buffer.refCnt() != 0) {
           buffer.discardReadBytes();
      }
     return;
 }

Maybe we don't need discard read bytes each time when end with a partial message. We can optimize some, like :

1. monitor the buffer usage , when 1/3 left or less than a fixed size free, do discard
2. when face out of memory, try to do discard

[mp911de]

Thanks for the report. We had early discardReadBytes in CommandHandler (see #725). It's not exactly what you proposed, but touches a similar idea.

Discarding copies bytes around which can have a performance impact in terms of CPU time. This is a balance between memory cost and CPU cost. Right now, we optimize for CPU.

The approach of free percentage seems quite appealing to me. Do you want to give it a spin by submitting a PR that introduces conditional discard after decoding a command?

[mp911de]

Update: That's not exactly a memory leak. A leak causes non-reclaimable memory. This one is rather an optimization to memory usage.

[gavincook]

Sure, Let me try

[gavincook]

Here i pushed a pr on 916 , providing a buffer usage ratio to optimize buffer memory usage.

[mp911de]

I adjusted the ticket title to reflect the actual purpose.

[mp911de]

PR #916 is merged now. Thanks a lot.