From c38c5f3c09f679534d79a197ae7fa804b85fa869 Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Fri, 16 May 2014 12:39:19 +0200 Subject: [PATCH] DATAREDIS-300 - Fix inconsistency with connection open() and close(). Redis connections have been left open when calling RedisTemplate.execute as the connection might have been bound to a thread, released but never been unbound. We now only bind and unbind connections in case of enabled transaction support, while just getting and releasing it for all other purposes. Original pull request: #69. --- .../data/redis/core/RedisConnectionUtils.java | 108 +++++++++++++----- .../data/redis/core/RedisTemplate.java | 15 ++- .../core/MultithreadedRedisTemplateTests.java | 92 +++++++++++++++ 3 files changed, 187 insertions(+), 28 deletions(-) create mode 100644 src/test/java/org/springframework/data/redis/core/MultithreadedRedisTemplateTests.java diff --git a/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java b/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java index c4f08aab5..b3db62fd4 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java +++ b/src/main/java/org/springframework/data/redis/core/RedisConnectionUtils.java @@ -147,36 +147,18 @@ public abstract class RedisConnectionUtils { return conn; } - private static void potentiallyRegisterTransactionSynchronisation(final RedisConnectionHolder connHolder, - RedisConnectionFactory factory) { + private static void potentiallyRegisterTransactionSynchronisation(RedisConnectionHolder connHolder, + final RedisConnectionFactory factory) { if (isActualNonReadonlyTransactionActive()) { - final RedisConnection connection = connHolder.getConnection(); - if (!connHolder.isTransactionSyncronisationActive()) { connHolder.setTransactionSyncronisationActive(true); - connection.multi(); + + RedisConnection conn = connHolder.getConnection(); + conn.multi(); - TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { - @Override - public void afterCompletion(int status) { - - switch (status) { - - case TransactionSynchronization.STATUS_COMMITTED: - connection.exec(); - break; - - case TransactionSynchronization.STATUS_ROLLED_BACK: - case TransactionSynchronization.STATUS_UNKNOWN: - default: - connection.discard(); - } - - connHolder.setTransactionSyncronisationActive(false); - } - }); + TransactionSynchronizationManager.registerSynchronization(new RedisTransactionSynchronizer(connHolder, conn)); } } } @@ -202,9 +184,20 @@ public abstract class RedisConnectionUtils { * @param factory the Redis factory that the connection was created with */ public static void releaseConnection(RedisConnection conn, RedisConnectionFactory factory) { + if (conn == null) { return; } + + RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager.getResource(factory); + + if (connHolder != null && connHolder.isTransactionSyncronisationActive()) { + if (log.isDebugEnabled()) { + log.debug("Redis Connection will be closed when transaction finished."); + } + return; + } + // Only release non-transactional/non-bound connections. if (!isConnectionTransactional(conn, factory)) { if (log.isDebugEnabled()) { @@ -220,12 +213,24 @@ public abstract class RedisConnectionUtils { * @param factory Redis factory */ public static void unbindConnection(RedisConnectionFactory factory) { + RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager .unbindResourceIfPossible(factory); + if (connHolder != null) { - RedisConnection connection = connHolder.getConnection(); - connection.close(); + if (connHolder.isTransactionSyncronisationActive()) { + if (log.isDebugEnabled()) { + log.debug("Redis Connection will be closed when outer transaction finished."); + } + } else { + if (log.isDebugEnabled()) { + log.debug("Closing bound connection."); + } + RedisConnection connection = connHolder.getConnection(); + connection.close(); + } } + } /** @@ -242,9 +247,60 @@ public abstract class RedisConnectionUtils { } RedisConnectionHolder connHolder = (RedisConnectionHolder) TransactionSynchronizationManager .getResource(connFactory); + return (connHolder != null && conn == connHolder.getConnection()); } + /** + * A {@link TransactionSynchronizationAdapter} that makes sure that the associated RedisConnection is released after the transaction completes. + * + * @author Christoph Strobl + * @author Thomas Darimont + */ + private static class RedisTransactionSynchronizer extends TransactionSynchronizationAdapter { + + private final RedisConnectionHolder connHolder; + private final RedisConnection connection; + + /** + * Creates a new {@link RedisTransactionSynchronizer}. + * + * @param connHolder + * @param connection + */ + private RedisTransactionSynchronizer(RedisConnectionHolder connHolder, RedisConnection connection) { + + this.connHolder = connHolder; + this.connection = connection; + } + + @Override + public void afterCompletion(int status) { + + try { + switch (status) { + + case TransactionSynchronization.STATUS_COMMITTED: + connection.exec(); + break; + + case TransactionSynchronization.STATUS_ROLLED_BACK: + case TransactionSynchronization.STATUS_UNKNOWN: + default: + connection.discard(); + } + } finally { + + if (log.isDebugEnabled()) { + log.debug("Closing bound connection after transaction completed with " + status); + } + + connHolder.setTransactionSyncronisationActive(false); + connection.close(); + } + } + } + /** * @author Christoph Strobl * @since 1.3 diff --git a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java index 6eb3c2ac3..fe7673a51 100644 --- a/src/main/java/org/springframework/data/redis/core/RedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/RedisTemplate.java @@ -169,7 +169,13 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation RedisConnectionFactory factory = getConnectionFactory(); RedisConnection conn = null; try { - conn = RedisConnectionUtils.bindConnection(factory, enableTransactionSupport); + + if (enableTransactionSupport) { + // only bind resources in case of potential transaction synchronization + conn = RedisConnectionUtils.bindConnection(factory, enableTransactionSupport); + } else { + conn = RedisConnectionUtils.getConnection(factory); + } boolean existingConnection = TransactionSynchronizationManager.hasResource(factory); @@ -191,7 +197,12 @@ public class RedisTemplate extends RedisAccessor implements RedisOperation // TODO: any other connection processing? return postProcessResult(result, connToUse, existingConnection); } finally { - RedisConnectionUtils.releaseConnection(conn, factory); + + if (enableTransactionSupport) { + RedisConnectionUtils.unbindConnection(factory); + } else { + RedisConnectionUtils.releaseConnection(conn, factory); + } } } diff --git a/src/test/java/org/springframework/data/redis/core/MultithreadedRedisTemplateTests.java b/src/test/java/org/springframework/data/redis/core/MultithreadedRedisTemplateTests.java new file mode 100644 index 000000000..6990cc501 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/core/MultithreadedRedisTemplateTests.java @@ -0,0 +1,92 @@ +/* + * Copyright 2014 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 + * + * http://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.core; + +import static org.junit.Assert.*; + +import java.util.Arrays; +import java.util.Collection; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; +import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; +import org.springframework.data.redis.connection.srp.SrpConnectionFactory; + +/** + * @author Artem Bilian + * @author Christoph Strobl + */ +@RunWith(Parameterized.class) +public class MultithreadedRedisTemplateTests { + + private RedisConnectionFactory factory; + + public MultithreadedRedisTemplateTests(RedisConnectionFactory factory) { + this.factory = factory; + } + + @Parameters + public static Collection testParams() { + + JedisConnectionFactory jedis = new JedisConnectionFactory(); + jedis.setPort(6379); + jedis.afterPropertiesSet(); + + LettuceConnectionFactory lettuce = new LettuceConnectionFactory(); + lettuce.setPort(6379); + lettuce.afterPropertiesSet(); + + SrpConnectionFactory srp = new SrpConnectionFactory(); + srp.setPort(6479); + srp.afterPropertiesSet(); + + return Arrays.asList(new Object[][] { { jedis }, { lettuce }, { srp } }); + } + + /** + * @see DATAREDIS-300 + */ + @Test + public void assertResouresAreReleasedProperlyWhenSharingRedisTemplate() throws InterruptedException { + + final RedisTemplate template = new RedisTemplate(); + template.setConnectionFactory(factory); + template.afterPropertiesSet(); + + ExecutorService executor = Executors.newCachedThreadPool(); + + for (int i = 0; i < 9; i++) { + executor.execute(new Runnable() { + + @Override + public void run() { + template.boundValueOps("foo").get(); + } + }); + } + + executor.shutdown(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + +}