redis#906 - use buffer usage ratio to optimize memory usage

gavincook · Nov 20, 2018 · 95cfaa8 · 95cfaa8
1 parent 5f27184
commit 95cfaa8
Show file tree

Hide file tree

Showing 2 changed files with 42 additions and 3 deletions.
diff --git a/src/main/java/io/lettuce/core/ClientOptions.java b/src/main/java/io/lettuce/core/ClientOptions.java
@@ -36,6 +36,7 @@ public class ClientOptions implements Serializable {
     public static final SocketOptions DEFAULT_SOCKET_OPTIONS = SocketOptions.create();
     public static final SslOptions DEFAULT_SSL_OPTIONS = SslOptions.create();
     public static final TimeoutOptions DEFAULT_TIMEOUT_OPTIONS = TimeoutOptions.create();
+    public static final int DEFAULT_BUFFER_USAGE_RATIO = 3;
 
     private final boolean pingBeforeActivateConnection;
     private final boolean autoReconnect;
@@ -46,6 +47,7 @@ public class ClientOptions implements Serializable {
     private final SocketOptions socketOptions;
     private final SslOptions sslOptions;
     private final TimeoutOptions timeoutOptions;
+    private final int bufferUsageRatio;
     private final Builder builder;
 
     protected ClientOptions(Builder builder) {
@@ -58,6 +60,7 @@ protected ClientOptions(Builder builder) {
         this.socketOptions = builder.socketOptions;
         this.sslOptions = builder.sslOptions;
         this.timeoutOptions = builder.timeoutOptions;
+        this.bufferUsageRatio = builder.bufferUsageRatio;
         this.builder = builder;
     }
 
@@ -71,6 +74,7 @@ protected ClientOptions(ClientOptions original) {
         this.socketOptions = original.getSocketOptions();
         this.sslOptions = original.getSslOptions();
         this.timeoutOptions = original.getTimeoutOptions();
+        this.bufferUsageRatio = original.getBufferUsageRatio();
         this.builder = original.builder;
     }
 
@@ -116,6 +120,7 @@ public static class Builder {
         private SocketOptions socketOptions = DEFAULT_SOCKET_OPTIONS;
         private SslOptions sslOptions = DEFAULT_SSL_OPTIONS;
         private TimeoutOptions timeoutOptions = DEFAULT_TIMEOUT_OPTIONS;
+        private int bufferUsageRatio = DEFAULT_BUFFER_USAGE_RATIO;
 
         protected Builder() {
         }
@@ -238,6 +243,20 @@ public Builder timeoutOptions(TimeoutOptions timeoutOptions) {
             return this;
         }
 
+        /**
+         * Sets the buffer usage ratio to {@link io.lettuce.core.protocol.CommandHandler} for determine when to discard
+         * read bytes.See {@link #DEFAULT_BUFFER_USAGE_RATIO}.
+         *
+         * @param bufferUsageRatio must greater than 0
+         * @return {@code this}
+         * @since 5.2
+         */
+        public Builder bufferUsageRatio(int bufferUsageRatio) {
+            LettuceAssert.isTrue(bufferUsageRatio > 0, "BufferUsageRatio must grater than 0");
+            this.bufferUsageRatio = bufferUsageRatio;
+            return this;
+        }
+
         /**
          * Create a new instance of {@link ClientOptions}.
          *
@@ -354,6 +373,18 @@ public TimeoutOptions getTimeoutOptions() {
         return timeoutOptions;
     }
 
+    /**
+     * Buffer usage ratio for {@link io.lettuce.core.protocol.CommandHandler}. Discard operation for read bytes occur
+     * When buffer usage reach {@code bufferUsageRatio / bufferUsageRatio + 1}. For example, sets 3 to bufferUsageRatio,
+     * means buffer usage reach 75 percentage, discard operation occur.
+     *
+     * @return the buffer usage ratio.
+     * @since 5.2
+     */
+    public int getBufferUsageRatio() {
+        return bufferUsageRatio;
+    }
+
     /**
      * Behavior of connections in disconnected state.
      */

diff --git a/src/main/java/io/lettuce/core/protocol/CommandHandler.java b/src/main/java/io/lettuce/core/protocol/CommandHandler.java
@@ -588,6 +588,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Interrup
 
             try {
                 if (!decode(ctx, buffer, command)) {
+                    discardReadBytesIfNecessary(buffer);
                     return;
                 }
             } catch (Exception e) {
@@ -614,9 +615,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Interrup
             afterDecode(ctx, command);
         }
 
-        if (buffer.refCnt() != 0) {
-            buffer.discardReadBytes();
-        }
+        discardReadBytesIfNecessary(buffer);
     }
 
     /**
@@ -840,6 +839,15 @@ private static long nanoTime() {
         return System.nanoTime();
     }
 
+    private void discardReadBytesIfNecessary(ByteBuf buffer) {
+        int bufferUsageRatio = clientOptions.getBufferUsageRatio();
+        if (buffer.writerIndex() * (bufferUsageRatio + 1) >= buffer.capacity() * bufferUsageRatio) {
+            if (buffer.refCnt() != 0) {
+                buffer.discardReadBytes();
+            }
+        }
+    }
+
     public enum LifecycleState {
         NOT_CONNECTED, REGISTERED, CONNECTED, ACTIVATING, ACTIVE, DISCONNECTED, DEACTIVATING, DEACTIVATED, CLOSED,
     }