use same node for SCAN/SSCAN/HSCAN during iteration. #230

redisson · Sep 4, 2015 · 8bdd6cf · 8bdd6cf
1 parent 0f663f0
commit 8bdd6cf
Show file tree

Hide file tree

Showing 13 changed files with 178 additions and 77 deletions.
diff --git a/src/main/java/org/redisson/CommandBatchExecutorService.java b/src/main/java/org/redisson/CommandBatchExecutorService.java
@@ -24,11 +24,11 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
-import org.redisson.client.WriteRedisConnectionException;
+import org.redisson.client.RedisClient;
-import org.redisson.client.RedisConnectionException;
 import org.redisson.client.RedisException;
 import org.redisson.client.RedisMovedException;
 import org.redisson.client.RedisTimeoutException;
+import org.redisson.client.WriteRedisConnectionException;
 import org.redisson.client.codec.Codec;
 import org.redisson.client.protocol.CommandData;
 import org.redisson.client.protocol.CommandsData;
@@ -102,7 +102,7 @@ public CommandBatchExecutorService(ConnectionManager connectionManager) {
 
     @Override
     protected <V, R> void async(boolean readOnlyMode, int slot, MultiDecoder<Object> messageDecoder,
-            Codec codec, RedisCommand<V> command, Object[] params, Promise<R> mainPromise, int attempt) {
+            Codec codec, RedisCommand<V> command, Object[] params, Promise<R> mainPromise, RedisClient client, int attempt) {
         if (executed) {
             throw new IllegalStateException("Batch already executed!");
         }

diff --git a/src/main/java/org/redisson/CommandExecutor.java b/src/main/java/org/redisson/CommandExecutor.java
@@ -18,6 +18,7 @@
 import java.util.Collection;
 import java.util.List;
 
+import org.redisson.client.RedisClient;
 import org.redisson.client.codec.Codec;
 import org.redisson.client.protocol.RedisCommand;
 import org.redisson.connection.ConnectionManager;
@@ -32,6 +33,8 @@
 //TODO ping support
 public interface CommandExecutor {
 
+    <T, R> R read(RedisClient client, String key, RedisCommand<T> command, Object ... params);
+
     <T, R> Future<R> evalWriteAllAsync(RedisCommand<T> command, SlotCallback<T, R> callback, String script, List<Object> keys, Object ... params);
 
     <R, T> Future<R> writeAllAsync(RedisCommand<T> command, SlotCallback<T, R> callback, Object ... params);

diff --git a/src/main/java/org/redisson/CommandExecutorService.java b/src/main/java/org/redisson/CommandExecutorService.java
@@ -26,12 +26,12 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.redisson.client.RedisClient;
 import org.redisson.client.RedisConnection;
-import org.redisson.client.WriteRedisConnectionException;
-import org.redisson.client.RedisConnectionException;
 import org.redisson.client.RedisException;
 import org.redisson.client.RedisMovedException;
 import org.redisson.client.RedisTimeoutException;
+import org.redisson.client.WriteRedisConnectionException;
 import org.redisson.client.codec.Codec;
 import org.redisson.client.protocol.CommandData;
 import org.redisson.client.protocol.RedisCommand;
@@ -98,7 +98,7 @@ public Promise<R> setFailure(Throwable cause) {
         };
 
         for (Integer slot : connectionManager.getEntries().keySet()) {
-            async(true, slot, null, connectionManager.getCodec(), command, params, promise, 0);
+            async(true, slot, null, connectionManager.getCodec(), command, params, promise, null, 0);
         }
         return mainPromise;
     }
@@ -135,7 +135,7 @@ public void operationComplete(Future<R> future) throws Exception {
         });
 
         Integer slot = slots.remove(0);
-        async(true, slot, null, connectionManager.getCodec(), command, params, attemptPromise, 0);
+        async(true, slot, null, connectionManager.getCodec(), command, params, attemptPromise, null, 0);
     }
 
     public <T> Future<Void> writeAllAsync(RedisCommand<T> command, Object ... params) {
@@ -172,7 +172,7 @@ public Promise<T> setFailure(Throwable cause) {
             }
         };
         for (Integer slot : connectionManager.getEntries().keySet()) {
-            async(readOnlyMode, slot, null, connectionManager.getCodec(), command, params, promise, 0);
+            async(readOnlyMode, slot, null, connectionManager.getCodec(), command, params, promise, null, 0);
         }
         return mainPromise;
     }
@@ -196,10 +196,22 @@ public <T, R> R read(String key, Codec codec, RedisCommand<T> command, Object ..
         return get(res);
     }
 
+    public <T, R> R read(RedisClient client, String key, RedisCommand<T> command, Object ... params) {
+        Future<R> res = readAsync(client, key, connectionManager.getCodec(), command, params);
+        return get(res);
+    }
+
+    public <T, R> Future<R> readAsync(RedisClient client, String key, Codec codec, RedisCommand<T> command, Object ... params) {
+        Promise<R> mainPromise = connectionManager.newPromise();
+        int slot = connectionManager.calcSlot(key);
+        async(true, slot, null, codec, command, params, mainPromise, client, 0);
+        return mainPromise;
+    }
+
     public <T, R> Future<R> readAsync(String key, Codec codec, RedisCommand<T> command, Object ... params) {
         Promise<R> mainPromise = connectionManager.newPromise();
         int slot = connectionManager.calcSlot(key);
-        async(true, slot, null, codec, command, params, mainPromise, 0);
+        async(true, slot, null, codec, command, params, mainPromise, null, 0);
         return mainPromise;
     }
 
@@ -210,7 +222,7 @@ public <T, R> R write(Integer slot, Codec codec, RedisCommand<T> command, Object
 
     public <T, R> Future<R> writeAsync(Integer slot, Codec codec, RedisCommand<T> command, Object ... params) {
         Promise<R> mainPromise = connectionManager.newPromise();
-        async(false, slot, null, codec, command, params, mainPromise, 0);
+        async(false, slot, null, codec, command, params, mainPromise, null, 0);
         return mainPromise;
     }
 
@@ -328,7 +340,7 @@ public Promise<T> setFailure(Throwable cause) {
         args.addAll(keys);
         args.addAll(Arrays.asList(params));
         for (Integer slot : connectionManager.getEntries().keySet()) {
-            async(readOnlyMode, slot, null, connectionManager.getCodec(), command, args.toArray(), promise, 0);
+            async(readOnlyMode, slot, null, connectionManager.getCodec(), command, args.toArray(), promise, null, 0);
         }
         return mainPromise;
     }
@@ -341,7 +353,7 @@ private <T, R> Future<R> evalAsync(boolean readOnlyMode, String key, Codec codec
         args.addAll(keys);
         args.addAll(Arrays.asList(params));
         int slot = connectionManager.calcSlot(key);
-        async(readOnlyMode, slot, null, codec, evalCommandType, args.toArray(), mainPromise, 0);
+        async(readOnlyMode, slot, null, codec, evalCommandType, args.toArray(), mainPromise, null, 0);
         return mainPromise;
     }
 
@@ -371,12 +383,12 @@ public <T, R> R write(String key, Codec codec, RedisCommand<T> command, Object .
     public <T, R> Future<R> writeAsync(String key, Codec codec, RedisCommand<T> command, Object ... params) {
         Promise<R> mainPromise = connectionManager.newPromise();
         int slot = connectionManager.calcSlot(key);
-        async(false, slot, null, codec, command, params, mainPromise, 0);
+        async(false, slot, null, codec, command, params, mainPromise, null, 0);
         return mainPromise;
     }
 
     protected <V, R> void async(final boolean readOnlyMode, final int slot, final MultiDecoder<Object> messageDecoder, final Codec codec, final RedisCommand<V> command,
-                            final Object[] params, final Promise<R> mainPromise, final int attempt) {
+                            final Object[] params, final Promise<R> mainPromise, final RedisClient client, final int attempt) {
         if (!connectionManager.getShutdownLatch().acquire()) {
             mainPromise.setFailure(new IllegalStateException("Redisson is shutdown"));
             return;
@@ -400,18 +412,22 @@ public void run(Timeout timeout) throws Exception {
                 }
 
                 int count = attempt + 1;
-                async(readOnlyMode, slot, messageDecoder, codec, command, params, mainPromise, count);
+                async(readOnlyMode, slot, messageDecoder, codec, command, params, mainPromise, client, count);
             }
         };
 
         try {
-            org.redisson.client.RedisConnection connection;
+            RedisConnection connection;
             if (readOnlyMode) {
-                connection = connectionManager.connectionReadOp(slot);
+                if (client != null) {
+                    connection = connectionManager.connectionReadOp(slot, client);
+                } else {
+                    connection = connectionManager.connectionReadOp(slot);
+                }
             } else {
                 connection = connectionManager.connectionWriteOp(slot);
             }
-            log.debug("getting connection for command {} via slot {} using {}", command, slot, connection.getRedisClient().getAddr());
+            log.debug("getting connection for command {} from slot {} using node {}", command, slot, connection.getRedisClient().getAddr());
             ChannelFuture future = connection.send(new CommandData<V, R>(attemptPromise, messageDecoder, codec, command, params));
 
             ex.set(new RedisTimeoutException());
@@ -449,12 +465,16 @@ public void operationComplete(Future<R> future) throws Exception {
                 if (future.cause() instanceof RedisMovedException) {
                     RedisMovedException ex = (RedisMovedException)future.cause();
                     connectionManager.getTimer().newTimeout(retryTimerTask, connectionManager.getConfig().getRetryInterval(), TimeUnit.MILLISECONDS);
-                    async(readOnlyMode, ex.getSlot(), messageDecoder, codec, command, params, mainPromise, attempt);
+                    async(readOnlyMode, ex.getSlot(), messageDecoder, codec, command, params, mainPromise, client, attempt);
                     return;
                 }
 
                 if (future.isSuccess()) {
-                    mainPromise.setSuccess(future.getNow());
+                    R res = future.getNow();
+                    if (res instanceof RedisClientResult) {
+                        ((RedisClientResult)res).setRedisClient(client);
+                    }
+                    mainPromise.setSuccess(res);
                 } else {
                     mainPromise.setFailure(future.cause());
                 }

diff --git a/src/main/java/org/redisson/RedisClientResult.java b/src/main/java/org/redisson/RedisClientResult.java
@@ -0,0 +1,11 @@
+package org.redisson;
+
+import org.redisson.client.RedisClient;
+
+public interface RedisClientResult {
+
+    void setRedisClient(RedisClient client);
+
+    RedisClient getRedisClient();
+
+}
diff --git a/src/main/java/org/redisson/RedissonMap.java b/src/main/java/org/redisson/RedissonMap.java
@@ -27,6 +27,7 @@
 import java.util.NoSuchElementException;
 import java.util.Set;
 
+import org.redisson.client.RedisClient;
 import org.redisson.client.codec.StringCodec;
 import org.redisson.client.protocol.RedisCommand;
 import org.redisson.client.protocol.RedisCommand.ValueType;
@@ -277,15 +278,16 @@ public long fastRemove(K ... keys) {
         return get(fastRemoveAsync(keys));
     }
 
-    private MapScanResult<Object, V> scanIterator(long startPos) {
+    private MapScanResult<Object, V> scanIterator(RedisClient client, long startPos) {
-        return commandExecutor.read(getName(), RedisCommands.HSCAN, getName(), startPos);
+        return commandExecutor.read(client, getName(), RedisCommands.HSCAN, getName(), startPos);
     }
 
     private Iterator<Map.Entry<K, V>> iterator() {
         return new Iterator<Map.Entry<K, V>>() {
 
             private Iterator<Map.Entry<K, V>> iter;
             private long iterPos = 0;
+            private RedisClient client;
 
             private boolean removeExecuted;
             private Map.Entry<K,V> value;
@@ -294,7 +296,10 @@ private Iterator<Map.Entry<K, V>> iterator() {
             public boolean hasNext() {
                 if (iter == null
                         || (!iter.hasNext() && iterPos != 0)) {
-                    MapScanResult<Object, V> res = scanIterator(iterPos);
+                    MapScanResult<Object, V> res = scanIterator(client, iterPos);
+                    if (iter == null) {
+                        client = res.getRedisClient();
+                    }
                     iter = ((Map<K, V>)res.getMap()).entrySet().iterator();
                     iterPos = res.getPos();
                 }

diff --git a/src/main/java/org/redisson/RedissonSet.java b/src/main/java/org/redisson/RedissonSet.java
@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.NoSuchElementException;
 
+import org.redisson.client.RedisClient;
 import org.redisson.client.protocol.RedisCommand;
 import org.redisson.client.protocol.RedisCommands;
 import org.redisson.client.protocol.convertor.BooleanReplayConvertor;
@@ -68,15 +69,16 @@ public Future<Boolean> containsAsync(Object o) {
         return commandExecutor.readAsync(getName(), RedisCommands.SISMEMBER, getName(), o);
     }
 
-    private ListScanResult<V> scanIterator(long startPos) {
+    private ListScanResult<V> scanIterator(RedisClient client, long startPos) {
-        return commandExecutor.read(getName(), RedisCommands.SSCAN, getName(), startPos);
+        return commandExecutor.read(client, getName(), RedisCommands.SSCAN, getName(), startPos);
     }
 
     @Override
     public Iterator<V> iterator() {
         return new Iterator<V>() {
 
             private Iterator<V> iter;
+            private RedisClient client;
             private Long iterPos;
 
             private boolean removeExecuted;
@@ -85,11 +87,12 @@ public Iterator<V> iterator() {
             @Override
             public boolean hasNext() {
                 if (iter == null) {
-                    ListScanResult<V> res = scanIterator(0);
+                    ListScanResult<V> res = scanIterator(null, 0);
+                    client = res.getRedisClient();
                     iter = res.getValues().iterator();
                     iterPos = res.getPos();
                 } else if (!iter.hasNext() && iterPos != 0) {
-                    ListScanResult<V> res = scanIterator(iterPos);
+                    ListScanResult<V> res = scanIterator(client, iterPos);
                     iter = res.getValues().iterator();
                     iterPos = res.getPos();
                 }

diff --git a/src/main/java/org/redisson/client/protocol/decoder/ListScanResult.java b/src/main/java/org/redisson/client/protocol/decoder/ListScanResult.java
@@ -17,10 +17,14 @@
 
 import java.util.List;
 
-public class ListScanResult<V> {
+import org.redisson.RedisClientResult;
+import org.redisson.client.RedisClient;
+
+public class ListScanResult<V> implements RedisClientResult {
 
     private final Long pos;
     private final List<V> values;
+    private RedisClient client;
 
     public ListScanResult(Long pos, List<V> values) {
         this.pos = pos;
@@ -35,4 +39,13 @@ public List<V> getValues() {
         return values;
     }
 
+    @Override
+    public void setRedisClient(RedisClient client) {
+        this.client = client;
+    }
+
+    public RedisClient getRedisClient() {
+        return client;
+    }
+
 }
diff --git a/src/main/java/org/redisson/client/protocol/decoder/MapScanResult.java b/src/main/java/org/redisson/client/protocol/decoder/MapScanResult.java
@@ -17,10 +17,14 @@
 
 import java.util.Map;
 
-public class MapScanResult<K, V> {
+import org.redisson.RedisClientResult;
+import org.redisson.client.RedisClient;
+
+public class MapScanResult<K, V> implements RedisClientResult {
 
     private final Long pos;
     private final Map<K, V> values;
+    private RedisClient client;
 
     public MapScanResult(Long pos, Map<K, V> values) {
         super();
@@ -36,4 +40,13 @@ public Map<K, V> getMap() {
         return values;
     }
 
+    @Override
+    public void setRedisClient(RedisClient client) {
+        this.client = client;
+    }
+
+    public RedisClient getRedisClient() {
+        return client;
+    }
+
 }