perf: add busyLoop mode

Author: okg-cxf

src/main/java/io/lettuce/core/AutoBatchFlushOptions.java

@@ -2,6 +2,8 @@
 
 import java.io.Serializable;
 
+import io.lettuce.core.internal.LettuceAssert;
+
 /**
  * Options for command timeouts. These options configure how and whether commands time out once they were dispatched. Command
  * timeout begins:
@@ -24,16 +26,26 @@ public class AutoBatchFlushOptions implements Serializable {
 
     public static final int DEFAULT_BATCH_SIZE = 8;
 
+    public static final boolean DEFAULT_USE_BUSY_LOOP = false;
+
+    public static final long DEFAULT_BUSY_LOOP_DELAY_IN_NANOS = 400;
+
     private final boolean enableAutoBatchFlush;
 
     private final int writeSpinCount;
 
     private final int batchSize;
 
+    private final boolean busyLoop;
+
+    private final long busyLoopDelayInNanos;
+
     public AutoBatchFlushOptions(AutoBatchFlushOptions.Builder builder) {
         this.enableAutoBatchFlush = builder.enableAutoBatchFlush;
         this.writeSpinCount = builder.writeSpinCount;
         this.batchSize = builder.batchSize;
+        this.busyLoop = builder.busyLoop;
+        this.busyLoopDelayInNanos = builder.busyLoopDelayInNanos;
     }
 
     /**
@@ -61,6 +73,10 @@ public static class Builder {
 
         private int batchSize = DEFAULT_BATCH_SIZE;
 
+        private boolean busyLoop = DEFAULT_USE_BUSY_LOOP;
+
+        private long busyLoopDelayInNanos = DEFAULT_BUSY_LOOP_DELAY_IN_NANOS;
+
         /**
          * Enable auto batch flush.
          *
@@ -79,6 +95,8 @@ public Builder enableAutoBatchFlush(boolean enableAutoBatchFlush) {
          * @return {@code this}
          */
         public Builder writeSpinCount(int writeSpinCount) {
+            LettuceAssert.isPositive(writeSpinCount, "Write spin count must be greater 0");
+
             this.writeSpinCount = writeSpinCount;
             return this;
         }
@@ -90,10 +108,24 @@ public Builder writeSpinCount(int writeSpinCount) {
          * @return {@code this}
          */
         public Builder batchSize(int batchSize) {
+            LettuceAssert.isPositive(batchSize, "Batch size must be greater 0");
+
             this.batchSize = batchSize;
             return this;
         }
 
+        public Builder busyLoop(boolean busyLoop) {
+            this.busyLoop = busyLoop;
+            return this;
+        }
+
+        public Builder busyLoopDelayInNanos(long busyLoopDelayInNanos) {
+            LettuceAssert.isNonNegative(busyLoopDelayInNanos, "Busy loop delay must be greater 0");
+
+            this.busyLoopDelayInNanos = busyLoopDelayInNanos;
+            return this;
+        }
+
         /**
          * Create a new instance of {@link AutoBatchFlushOptions}.
          *
@@ -126,4 +158,12 @@ public int getBatchSize() {
         return batchSize;
     }
 
+    public boolean isBusyLoop() {
+        return busyLoop;
+    }
+
+    public long getBusyLoopDelayInNanos() {
+        return busyLoopDelayInNanos;
+    }
+
 }

src/main/java/io/lettuce/core/ContextualChannel.java

@@ -2,6 +2,8 @@
 
 import java.net.SocketAddress;
 
+import javax.annotation.Nonnull;
+
 import io.lettuce.core.context.ConnectionContext;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.Channel;
@@ -15,7 +17,6 @@
 import io.netty.channel.EventLoop;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
-import org.jetbrains.annotations.NotNull;
 
 /**
  * @author chenxiaofan
@@ -250,7 +251,7 @@ public <T> boolean hasAttr(AttributeKey<T> attributeKey) {
     }
 
     @Override
-    public int compareTo(@NotNull Channel o) {
+    public int compareTo(@Nonnull Channel o) {
         return this == o ? 0 : this.id().compareTo(o.id());
     }
 

src/main/java/io/lettuce/core/internal/LettuceAssert.java

@@ -237,9 +237,15 @@ public static void assertState(boolean condition, Supplier<String> messageSuppli
         }
     }
 
-    public static void isPositive(int writeSpinCount, String writeSpinCountIsNotPositive) {
-        if (writeSpinCount <= 0) {
-            throw new IllegalArgumentException(writeSpinCountIsNotPositive);
+    public static void isPositive(int digit, String message) {
+        if (digit <= 0) {
+            throw new IllegalArgumentException(message);
+        }
+    }
+
+    public static void isNonNegative(long digit, String message) {
+        if (digit < 0) {
+            throw new IllegalArgumentException(message);
         }
     }
 

src/main/java/io/lettuce/core/protocol/DefaultBatchFlushEndpoint.java

@@ -169,6 +169,10 @@ protected static void cancelCommandOnEndpointClose(RedisCommand<?, ?, ?> cmd) {
 
     private final int batchSize;
 
+    private final boolean busyLoop;
+
+    private final long busyLoopDelayInNanos;
+
     /**
      * Create a new {@link BatchFlushEndpoint}.
      *
@@ -197,6 +201,8 @@ protected DefaultBatchFlushEndpoint(ClientOptions clientOptions, ClientResources
         this.callbackOnClose = callbackOnClose;
         this.writeSpinCount = clientOptions.getAutoBatchFlushOptions().getWriteSpinCount();
         this.batchSize = clientOptions.getAutoBatchFlushOptions().getBatchSize();
+        this.busyLoop = clientOptions.getAutoBatchFlushOptions().isBusyLoop();
+        this.busyLoopDelayInNanos = clientOptions.getAutoBatchFlushOptions().getBusyLoopDelayInNanos();
     }
 
     @Override
@@ -607,6 +613,10 @@ private void scheduleSendJobOnConnected(final ContextualChannel chan) {
     }
 
     private void scheduleSendJobIfNeeded(final ContextualChannel chan) {
+        if (busyLoop) {
+            return;
+        }
+
         final EventLoop eventLoop = chan.eventLoop();
         if (eventLoop.inEventLoop()) {
             scheduleSendJobInEventLoopIfNeeded(chan);
@@ -669,6 +679,13 @@ private void loopSend0(final BatchFlushEndPointContext batchFlushEndPointContext
             return;
         }
 
+        if (busyLoop) {
+            // Don't use chan.eventLoop().execute(), otherwise performance will drop, since the event loop
+            // thread will trap within a certain time period.
+            chan.eventLoop().schedule(() -> loopSend(chan), busyLoopDelayInNanos, TimeUnit.NANOSECONDS);
+            return;
+        }
+
         if (firstCall) {
             // // Don't setUnsafe here because loopSend0() may result in a delayed loopSend() call.
             batchFlushEndPointContext.hasOngoingSendLoop.exitSafe();
@@ -683,7 +700,7 @@ private void loopSend0(final BatchFlushEndPointContext batchFlushEndPointContext
     private int pollBatch(final BatchFlushEndPointContext batchFlushEndPointContext, ContextualChannel chan) {
         int count = 0;
         for (; count < batchSize; count++) {
-            final RedisCommand<?, ?, ?> cmd = this.taskQueue.poll(); // relaxed poll is faster and we wil retry later anyway.
+            final RedisCommand<?, ?, ?> cmd = this.taskQueue.poll();
             if (cmd == null) {
                 break;
             }