From 42ad8d190da0cd003cef007f15f7ab018c8b0ae6 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Thu, 13 Feb 2020 16:25:23 +0100 Subject: [PATCH] DATAREDIS-1011 - Allow configuration of Lettuce pipelining flush behavior. We now allow customization of flushing when using pipelining with Lettuce. Lettuce flushes each command by default. To optimize for performance, flushing behavior can be customized using a PipeliningFlushPolicy. Flush each command, flush on close and flush after n commands are the bundled implementations that can be configured on LettuceConnectionFactory. Original Pull Request: #511 --- .../connection/lettuce/LettuceConnection.java | 268 +++++++++++++++--- .../lettuce/LettuceConnectionFactory.java | 28 +- ...ionPipelineFlushOnEndIntegrationTests.java | 41 +++ ...nPipelineTxFlushOnEndIntegrationTests.java | 29 ++ .../PipeliningFlushPolicyUnitTests.java | 93 ++++++ ...lineFlushOnEndIntegrationTests-context.xml | 27 ++ 6 files changed, 445 insertions(+), 41 deletions(-) create mode 100644 src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineFlushOnEndIntegrationTests.java create mode 100644 src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineTxFlushOnEndIntegrationTests.java create mode 100644 src/test/java/org/springframework/data/redis/connection/lettuce/PipeliningFlushPolicyUnitTests.java create mode 100644 src/test/resources/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineFlushOnEndIntegrationTests-context.xml 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 @@ + + + + + + + + + + + + + + + + + + + +