diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index f759bdc0a..62f34906a 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -53,6 +53,7 @@ import java.util.concurrent.CompletionStage; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import java.util.function.Supplier; import org.springframework.beans.BeanUtils; @@ -64,9 +65,9 @@ import org.springframework.data.redis.ExceptionTranslationStrategy; import org.springframework.data.redis.FallbackExceptionTranslationStrategy; import org.springframework.data.redis.connection.*; import org.springframework.data.redis.connection.convert.TransactionResultConverter; -import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware; import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceResultBuilder; import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceStatusResult; +import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware; import org.springframework.data.redis.core.RedisCommand; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -108,10 +109,12 @@ public class LettuceConnection extends AbstractRedisConnection { private boolean isMulti = false; private boolean isPipelined = false; private @Nullable List ppline; + private @Nullable PipeliningFlushState flushState; private final Queue> txResults = new LinkedList<>(); private volatile @Nullable LettuceSubscription subscription; /** flag indicating whether the connection needs to be dropped or not */ private boolean convertPipelineAndTxResults = true; + private PipeliningFlushPolicy pipeliningFlushPolicy = PipeliningFlushPolicy.flushEachCommand(); LettuceResult newLettuceResult(Future resultHolder) { return newLettuceResult(resultHolder, (val) -> val); @@ -518,6 +521,8 @@ public class LettuceConnection extends AbstractRedisConnection { if (!isPipelined) { isPipelined = true; ppline = new ArrayList<>(); + flushState = this.pipeliningFlushPolicy.newPipeline(); + flushState.onOpen(this.getOrCreateDedicatedConnection()); } } @@ -529,9 +534,11 @@ public class LettuceConnection extends AbstractRedisConnection { public List closePipeline() { if (!isPipelined) { - return Collections.emptyList(); } + + flushState.onClose(this.getOrCreateDedicatedConnection()); + flushState = null; isPipelined = false; List> futures = new ArrayList<>(ppline.size()); for (LettuceResult result : ppline) { @@ -863,6 +870,21 @@ public class LettuceConnection extends AbstractRedisConnection { this.convertPipelineAndTxResults = convertPipelineAndTxResults; } + /** + * Configures the flushing policy when using pipelining. + * + * @param pipeliningFlushPolicy the flushing policy to control when commands get written to the Redis connection. + * @see PipeliningFlushPolicy#flushEachCommand() + * @see #openPipeline() + * @see StatefulRedisConnection#flushCommands() + */ + public void setPipeliningFlushPolicy(PipeliningFlushPolicy pipeliningFlushPolicy) { + + Assert.notNull(pipeliningFlushPolicy, "PipeliningFlushingPolicy must not be null!"); + + this.pipeliningFlushPolicy = pipeliningFlushPolicy; + } + private void checkSubscription() { if (isSubscribed()) { throw new RedisSubscribedConnectionException( @@ -902,8 +924,14 @@ public class LettuceConnection extends AbstractRedisConnection { void pipeline(LettuceResult result) { if (isQueueing()) { + + if (flushState != null) { + flushState.onCommand(getOrCreateDedicatedConnection()); + } + transaction(result); } else { + flushState.onCommand(getOrCreateDedicatedConnection()); ppline.add(result); } } @@ -947,31 +975,43 @@ public class LettuceConnection extends AbstractRedisConnection { return (RedisAsyncCommands) getAsyncDedicatedConnection(); } + RedisClusterCommands getDedicatedConnection() { + + StatefulConnection connection = getOrCreateDedicatedConnection(); + + if (connection instanceof StatefulRedisConnection) { + return ((StatefulRedisConnection) connection).sync(); + } + if (connection instanceof StatefulRedisClusterConnection) { + return ((StatefulRedisClusterConnection) connection).sync(); + } + + throw new IllegalStateException( + String.format("%s is not a supported connection type.", connection.getClass().getName())); + } + protected RedisClusterAsyncCommands getAsyncDedicatedConnection() { + StatefulConnection connection = getOrCreateDedicatedConnection(); + + if (connection instanceof StatefulRedisConnection) { + return ((StatefulRedisConnection) connection).async(); + } + if (asyncDedicatedConn instanceof StatefulRedisClusterConnection) { + return ((StatefulRedisClusterConnection) connection).async(); + } + + throw new IllegalStateException( + String.format("%s is not a supported connection type.", connection.getClass().getName())); + } + + private StatefulConnection getOrCreateDedicatedConnection() { + if (asyncDedicatedConn == null) { asyncDedicatedConn = doGetAsyncDedicatedConnection(); } - if (asyncDedicatedConn instanceof StatefulRedisConnection) { - return ((StatefulRedisConnection) asyncDedicatedConn).async(); - } - if (asyncDedicatedConn instanceof StatefulRedisClusterConnection) { - return ((StatefulRedisClusterConnection) asyncDedicatedConn).async(); - } - - throw new IllegalStateException( - String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName())); - } - - private boolean customizedDatabaseIndex() { - return defaultDbIndex != dbIndex; - } - - private void potentiallySelectDatabase(int dbIndex) { - if (asyncDedicatedConn instanceof StatefulRedisConnection) { - ((StatefulRedisConnection) asyncDedicatedConn).sync().select(dbIndex); - } + return asyncDedicatedConn; } @SuppressWarnings("unchecked") @@ -979,23 +1019,6 @@ public class LettuceConnection extends AbstractRedisConnection { return (RedisCommands) getDedicatedConnection(); } - RedisClusterCommands getDedicatedConnection() { - - if (asyncDedicatedConn == null) { - asyncDedicatedConn = doGetAsyncDedicatedConnection(); - } - - if (asyncDedicatedConn instanceof StatefulRedisConnection) { - return ((StatefulRedisConnection) asyncDedicatedConn).sync(); - } - if (asyncDedicatedConn instanceof StatefulRedisClusterConnection) { - return ((StatefulRedisClusterConnection) asyncDedicatedConn).sync(); - } - - throw new IllegalStateException( - String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName())); - } - @SuppressWarnings("unchecked") protected StatefulConnection doGetAsyncDedicatedConnection() { @@ -1008,6 +1031,16 @@ public class LettuceConnection extends AbstractRedisConnection { return connection; } + private boolean customizedDatabaseIndex() { + return defaultDbIndex != dbIndex; + } + + private void potentiallySelectDatabase(int dbIndex) { + if (asyncDedicatedConn instanceof StatefulRedisConnection) { + ((StatefulRedisConnection) asyncDedicatedConn).sync().select(dbIndex); + } + } + io.lettuce.core.ScanCursor getScanCursor(long cursorId) { return io.lettuce.core.ScanCursor.of(Long.toString(cursorId)); } @@ -1319,4 +1352,163 @@ public class LettuceConnection extends AbstractRedisConnection { } } } + + /** + * Strategy interface to control pipelining flush behavior. Lettuce writes (flushes) each command individually to the + * Redis connection. Flushing behavior can be customized to optimize for performance. Flushing can be either stateless + * or stateful. An example for stateful flushing is size-based (buffer) flushing to flush after a configured number of + * commands. + * + * @see StatefulRedisConnection#setAutoFlushCommands(boolean) + * @see StatefulRedisConnection#flushCommands() + */ + public interface PipeliningFlushPolicy { + + /** + * Return a policy to flush after each command (default behavior). + * + * @return a policy to flush after each command. + */ + static PipeliningFlushPolicy flushEachCommand() { + return FlushEachCommand.INSTANCE; + } + + /** + * Return a policy to flush only if {@link #closePipeline()} is called. + * + * @return a policy to flush after each command. + */ + static PipeliningFlushPolicy flushOnClose() { + return FlushOnClose.INSTANCE; + } + + /** + * Return a policy to buffer commands and to flush once reaching the configured {@code bufferSize}. The buffer is + * recurring so a buffer size of e.g. {@code 2} will flush after 2, 4, 6, … commands. + * + * @param bufferSize the number of commands to buffer before flushing. Must be greater than zero. + * @return a policy to flush buffered commands to the Redis connection once the configured number of commands was + * issued. + */ + static PipeliningFlushPolicy buffered(int bufferSize) { + + Assert.isTrue(bufferSize > 0, "Buffer size must be greater than 0"); + return () -> new BufferedFlushing(bufferSize); + } + + PipeliningFlushState newPipeline(); + } + + /** + * State object associated with flushing of the currently ongoing pipeline. + */ + public interface PipeliningFlushState { + + /** + * Callback if the pipeline gets opened. + * + * @param connection + * @see #openPipeline() + */ + void onOpen(StatefulConnection connection); + + /** + * Callback for each issued Redis command. + * + * @param connection + * @see #pipeline(LettuceResult) + */ + void onCommand(StatefulConnection connection); + + /** + * Callback if the pipeline gets closed. + * + * @param connection + * @see #closePipeline() + */ + void onClose(StatefulConnection connection); + } + + /** + * Implementation to flush on each command. + */ + private enum FlushEachCommand implements PipeliningFlushPolicy, PipeliningFlushState { + + INSTANCE; + + @Override + public PipeliningFlushState newPipeline() { + return INSTANCE; + } + + @Override + public void onOpen(StatefulConnection connection) {} + + @Override + public void onCommand(StatefulConnection connection) {} + + @Override + public void onClose(StatefulConnection connection) {} + } + + /** + * Implementation to flush on closing the pipeline. + */ + private enum FlushOnClose implements PipeliningFlushPolicy, PipeliningFlushState { + + INSTANCE; + + @Override + public PipeliningFlushState newPipeline() { + return INSTANCE; + } + + @Override + public void onOpen(StatefulConnection connection) { + connection.setAutoFlushCommands(false); + } + + @Override + public void onCommand(StatefulConnection connection) { + + } + + @Override + public void onClose(StatefulConnection connection) { + connection.setAutoFlushCommands(true); + connection.flushCommands(); + } + } + + /** + * Pipeline state for buffered flushing. + */ + private static class BufferedFlushing implements PipeliningFlushState { + + private final AtomicLong commands = new AtomicLong(); + + private final int flushAfter; + + public BufferedFlushing(int flushAfter) { + this.flushAfter = flushAfter; + } + + @Override + public void onOpen(StatefulConnection connection) { + connection.setAutoFlushCommands(false); + } + + @Override + public void onCommand(StatefulConnection connection) { + if (commands.incrementAndGet() % flushAfter == 0) { + connection.flushCommands(); + } + } + + @Override + public void onClose(StatefulConnection connection) { + connection.setAutoFlushCommands(true); + connection.flushCommands(); + } + } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java index cbc724889..d3ee7c7c5 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java @@ -114,6 +114,7 @@ public class LettuceConnectionFactory private boolean convertPipelineAndTxResults = true; private RedisStandaloneConfiguration standaloneConfig = new RedisStandaloneConfiguration("localhost", 6379); + private PipeliningFlushPolicy pipeliningFlushPolicy = PipeliningFlushPolicy.flushEachCommand(); private @Nullable RedisConfiguration configuration; @@ -393,7 +394,10 @@ public class LettuceConnectionFactory @Nullable StatefulRedisConnection sharedConnection, LettuceConnectionProvider connectionProvider, long timeout, int database) { - return new LettuceConnection(sharedConnection, connectionProvider, timeout, database); + LettuceConnection connection = new LettuceConnection(sharedConnection, connectionProvider, timeout, database); + connection.setPipeliningFlushPolicy(this.pipeliningFlushPolicy); + + return connection; } /** @@ -414,8 +418,11 @@ public class LettuceConnectionFactory LettuceConnectionProvider connectionProvider, ClusterTopologyProvider topologyProvider, ClusterCommandExecutor clusterCommandExecutor, Duration commandTimeout) { - return new LettuceClusterConnection(sharedConnection, connectionProvider, topologyProvider, clusterCommandExecutor, - commandTimeout); + LettuceClusterConnection connection = new LettuceClusterConnection(sharedConnection, connectionProvider, + topologyProvider, clusterCommandExecutor, commandTimeout); + connection.setPipeliningFlushPolicy(this.pipeliningFlushPolicy); + + return connection; } /* @@ -552,6 +559,21 @@ public class LettuceConnectionFactory standaloneConfig.setPort(port); } + /** + * Configures the flushing policy when using pipelining. Defaults to {@link PipeliningFlushPolicy#flushEachCommand() + * flush on each command}. + * + * @param pipeliningFlushPolicy the flushing policy to control when commands get written to the Redis connection. + * @see LettuceConnection#openPipeline() + * @see StatefulRedisConnection#flushCommands() + */ + public void setPipeliningFlushPolicy(PipeliningFlushPolicy pipeliningFlushPolicy) { + + Assert.notNull(pipeliningFlushPolicy, "PipeliningFlushingPolicy must not be null!"); + + this.pipeliningFlushPolicy = pipeliningFlushPolicy; + } + /** * Returns the connection timeout (in milliseconds). * diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineFlushOnEndIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineFlushOnEndIntegrationTests.java new file mode 100644 index 000000000..952c61c3a --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineFlushOnEndIntegrationTests.java @@ -0,0 +1,41 @@ +/* + * Copyright 2011-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.lettuce; + +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.data.redis.test.util.RelaxedJUnit4ClassRunner; +import org.springframework.test.context.ContextConfiguration; + +/** + * Integration test of {@link LettuceConnection} pipeline functionality with + * {@link LettuceConnection.PipeliningFlushPolicy}. + * + * @author Mark Paluch + */ +@RunWith(RelaxedJUnit4ClassRunner.class) +@ContextConfiguration("LettuceConnectionPipelineFlushOnEndIntegrationTests-context.xml") +public class LettuceConnectionPipelineFlushOnEndIntegrationTests extends LettuceConnectionPipelineIntegrationTests { + + @Test + @Ignore("WATCH command is flushed during EXEC therefore we're not run commands between WATCH and EXEC") + public void testWatch() throws Exception { + + } + +} diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineTxFlushOnEndIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineTxFlushOnEndIntegrationTests.java new file mode 100644 index 000000000..909a20971 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineTxFlushOnEndIntegrationTests.java @@ -0,0 +1,29 @@ +/* + * Copyright 2013-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.lettuce; + +import org.springframework.test.context.ContextConfiguration; + +/** + * Integration test of {@link LettuceConnection} transactions within a pipeline + * {@link LettuceConnection.PipeliningFlushPolicy}. + * + * @author Mark Paluch + */ +@ContextConfiguration("LettuceConnectionPipelineFlushOnEndIntegrationTests-context.xml") +public class LettuceConnectionPipelineTxFlushOnEndIntegrationTests extends LettuceConnectionPipelineTxIntegrationTests { + +} diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/PipeliningFlushPolicyUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/PipeliningFlushPolicyUnitTests.java new file mode 100644 index 000000000..ea216ce1c --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/PipeliningFlushPolicyUnitTests.java @@ -0,0 +1,93 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.lettuce; + +import static org.mockito.Mockito.*; +import static org.springframework.data.redis.connection.lettuce.LettuceConnection.*; + +import io.lettuce.core.api.StatefulRedisConnection; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +/** + * Unit tests for {@link PipeliningFlushPolicy}. + */ +@RunWith(MockitoJUnitRunner.class) +public class PipeliningFlushPolicyUnitTests { + + @Mock StatefulRedisConnection connection; + + @Test // DATAREDIS-1011 + public void shouldFlushEachCommand() { + + PipeliningFlushPolicy policy = PipeliningFlushPolicy.flushEachCommand(); + + PipeliningFlushState state = policy.newPipeline(); + + state.onOpen(connection); + state.onCommand(connection); + state.onClose(connection); + + verifyNoInteractions(connection); + } + + @Test // DATAREDIS-1011 + public void shouldFlushOnClose() { + + PipeliningFlushPolicy policy = PipeliningFlushPolicy.flushOnClose(); + + PipeliningFlushState state = policy.newPipeline(); + + state.onOpen(connection); + + verify(connection).setAutoFlushCommands(false); + + state.onCommand(connection); + + verifyNoMoreInteractions(connection); + + state.onClose(connection); + + verify(connection).setAutoFlushCommands(true); + verify(connection).flushCommands(); + } + + @Test // DATAREDIS-1011 + public void shouldFlushOnBuffer() { + + PipeliningFlushPolicy policy = PipeliningFlushPolicy.buffered(2); + + PipeliningFlushState state = policy.newPipeline(); + + state.onOpen(connection); + + verify(connection).setAutoFlushCommands(false); + + state.onCommand(connection); + verifyNoMoreInteractions(connection); + + state.onCommand(connection); + verify(connection).flushCommands(); + + state.onClose(connection); + + verify(connection).setAutoFlushCommands(true); + verify(connection, times(2)).flushCommands(); + } +} diff --git a/src/test/resources/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineFlushOnEndIntegrationTests-context.xml b/src/test/resources/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineFlushOnEndIntegrationTests-context.xml new file mode 100644 index 000000000..27fd45a38 --- /dev/null +++ b/src/test/resources/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineFlushOnEndIntegrationTests-context.xml @@ -0,0 +1,27 @@ + + + + + + + + + + + + + + + + + + + +