diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java index b46f21c1fa..6db6fdde59 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.redis.util; import java.net.InetAddress; @@ -22,11 +23,12 @@ import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.Date; -import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.Set; +import java.util.WeakHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; @@ -93,7 +95,9 @@ public final class RedisLockRegistry implements LockRegistry { private final RedisTemplate redisTemplate; - private final ThreadLocal> threadLocks = new ThreadLocal>(); + private final ThreadLocal> weakThreadLocks = new ThreadLocal>(); + + private final ThreadLocal> hardThreadLocks = new ThreadLocal>(); private final long expireAfter; @@ -101,6 +105,8 @@ public final class RedisLockRegistry implements LockRegistry { private final LockSerializer lockSerializer = new LockSerializer(); + private boolean useWeakReferences = false; + static { String host; try { @@ -156,48 +162,120 @@ public final class RedisLockRegistry implements LockRegistry { this.localRegistry = localRegistry; } + /** + * Change the state of thread local weak references storage for unlocked locks. + * Thread local weak references are used for lock obtaining optimization - + * thread will get same {@link RedisLock} object for certain key before actual + * locking and after unlocking (if variable still exists). + *

While is switched off (by default) every {@link RedisLockRegistry#obtain} call will provide + * different {@link RedisLock} objects for same unlocked key. + * @param useWeakReferences set to true for switch thread local weak references storage on, false by default + * @since 4.0.7 + */ + public void setUseWeakReferences(boolean useWeakReferences) { + this.useWeakReferences = useWeakReferences; + } + + /** + * Weak referenced locks, lock is kept here when actual lock is NOT gained. + * Used for obtaining same lock object within same thread and key. + * To avoid memory leaks lock objects without actual lock are kept as weak references. + * After gaining the actual lock, lock object moves from weak reference to hard reference and vise a versa. + */ + private Collection getWeakThreadLocks() { + Set locks = this.weakThreadLocks.get(); + if (locks == null) { + locks = Collections.newSetFromMap(new WeakHashMap()); + this.weakThreadLocks.set(locks); + } + return locks; + } + + /** + * Hard referenced locks, lock is kept here when actual lock is gained. + */ + private Collection getHardThreadLocks() { + List locks = this.hardThreadLocks.get(); + if (locks == null) { + locks = new LinkedList(); + this.hardThreadLocks.set(locks); + } + return locks; + } + + private RedisLock findLock(Collection locks, Object key) { + if (locks != null) { + for (RedisLock lock : locks) { + if (lock.getLockKey().equals(key)) { + return lock; + } + } + } + return null; + } + + private void toHardThreadStorage(RedisLock lock) { + if (this.weakThreadLocks.get() != null) { + this.weakThreadLocks.get().remove(lock); + } + + getHardThreadLocks().add(lock); + + //clean up + if (this.weakThreadLocks.get() != null && this.weakThreadLocks.get().isEmpty()) { + this.weakThreadLocks.remove(); + } + } + + private void toWeakThreadStorage(RedisLock lock) { + //to avoid collection creation on existence check use direct fields + if (this.hardThreadLocks.get() != null) { + getHardThreadLocks().remove(lock); + } + + if (this.useWeakReferences) { + getWeakThreadLocks().add(lock); + } + + //clean up + if (this.hardThreadLocks.get() != null && this.hardThreadLocks.get().isEmpty()) { + this.hardThreadLocks.remove(); + } + } + @Override public Lock obtain(Object lockKey) { Assert.isInstanceOf(String.class, lockKey); - List locks = this.threadLocks.get(); - if (locks == null) { - locks = new LinkedList(); - this.threadLocks.set(locks); - } - RedisLock lock = null; - for (RedisLock alock : locks) { - if (alock.getLockKey().equals(lockKey)) { - lock = alock; - break; - } - } + + //try to find the lock within hard references + RedisLock lock = findLock(this.hardThreadLocks.get(), lockKey); + /* * If the lock is locked, check that it matches what's in the store. * If it doesn't, the lock must have expired. */ if (lock != null && lock.thread != null) { - RedisLock lockInStore = RedisLockRegistry.this.redisTemplate - .boundValueOps(this.registryKey + ":" + lockKey).get(); + RedisLock lockInStore = this.redisTemplate.boundValueOps(this.registryKey + ":" + lockKey).get(); if (lockInStore == null || !lock.equals(lockInStore)) { - removeLockFromThreadLocal(locks, lock); + getHardThreadLocks().remove(lock); lock = null; } } - if (lock == null) { - lock = new RedisLock((String) lockKey); - locks.add(lock); - } - return lock; - } - private void removeLockFromThreadLocal(List locks, RedisLock lock) { - Iterator iterator = locks.iterator(); - while (iterator.hasNext()) { - if (iterator.next().equals(lock)) { - iterator.remove(); - break; + if (lock == null) { + //try to find the lock within weak references + lock = findLock(this.weakThreadLocks.get(), lockKey); + + if (lock == null) { + lock = new RedisLock((String) lockKey); + + if (this.useWeakReferences) { + getWeakThreadLocks().add(lock); + } } } + + return lock; } public Collection listLocks() { @@ -312,48 +390,59 @@ public final class RedisLockRegistry implements LockRegistry { this.reLock++; return true; } + + toHardThreadStorage(this); + /* * Set these now so they will be persisted if successful. */ this.lockedAt = System.currentTimeMillis(); this.threadName = currentThread.getName(); - Boolean success = RedisLockRegistry.this.redisTemplate.execute(new SessionCallback() { + Boolean success = false; + try { + success = RedisLockRegistry.this.redisTemplate.execute(new SessionCallback() { - @SuppressWarnings({"unchecked", "rawtypes"}) - @Override - public Boolean execute(RedisOperations ops) throws DataAccessException { - String key = constructLockKey(); + @SuppressWarnings({"unchecked", "rawtypes"}) + @Override + public Boolean execute(RedisOperations ops) throws DataAccessException { + String key = constructLockKey(); - ops.watch(key); //monitor key + ops.watch(key); //monitor key - ops.multi(); //transaction start + if (ops.opsForValue().get(key) != null) { + ops.unwatch(); //key already exists, stop monitoring + return false; + } - //can't rely on operations result inside transaction, execution is delayed till `exec()` - ops.opsForValue().setIfAbsent(key, RedisLock.this); + ops.multi(); //transaction start - //set expire on key if exists - ops.expire(key, RedisLockRegistry.this.expireAfter, TimeUnit.MILLISECONDS); + //set the value and expire + ops.opsForValue() + .set(key, RedisLock.this, RedisLockRegistry.this.expireAfter, TimeUnit.MILLISECONDS); - //exec will contain all operations result or null - if execution has been aborted due to 'watch' - List result = ops.exec(); + //exec will contain all operations result or null - if execution has been aborted due to 'watch' + return ops.exec() != null; + } - //check 'setIfAbsent' result (first in list) - return (result != null) && (!result.isEmpty()) && (Boolean.TRUE.equals(result.get(0))); + }); + + } finally { + + if (!success) { + this.lockedAt = 0; + this.threadName = null; + toWeakThreadStorage(this); + } + else { + this.thread = currentThread; + if (logger.isDebugEnabled()) { + logger.debug("New lock; " + this.toString()); + } } - }); + } - if (!success) { - this.lockedAt = 0; - this.threadName = null; - } - else { - this.thread = currentThread; - if (logger.isDebugEnabled()) { - logger.debug("New lock; " + this.toString()); - } - } return success; } @@ -388,22 +477,20 @@ public final class RedisLockRegistry implements LockRegistry { } throw new IllegalStateException("Lock is owned by " + this.thread.getName() + "; " + this.toString()); } + try { if (this.reLock-- <= 0) { - List locks = RedisLockRegistry.this.threadLocks.get(); - if (locks != null) { - removeLockFromThreadLocal(locks, this); - if (locks.size() == 0) { // last lock for this thread - RedisLockRegistry.this.threadLocks.remove(); + try { + this.assertLockInRedisIsUnchanged(); + RedisLockRegistry.this.redisTemplate.delete(constructLockKey()); + if (logger.isDebugEnabled()) { + logger.debug("Released lock; " + this.toString()); } + } finally { + this.thread = null; + this.reLock = 0; + toWeakThreadStorage(this); } - this.assertLockInRedisIsUnchanged(); - RedisLockRegistry.this.redisTemplate.delete(constructLockKey()); - if (logger.isDebugEnabled()) { - logger.debug("Released lock; " + this.toString()); - } - this.thread = null; - this.reLock = 0; } } finally { @@ -501,8 +588,8 @@ public final class RedisLockRegistry implements LockRegistry { int keyLength = t.lockKey.length(); int threadNameLength = t.threadName.length(); byte[] value = new byte[1 + hostLength + - 1 + keyLength + - 1 + threadNameLength + 8]; + 1 + keyLength + + 1 + threadNameLength + 8]; ByteBuffer buff = ByteBuffer.wrap(value); buff.put((byte) hostLength) .put(t.lockHost) diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java index 1f501ecf30..829a60a754 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.redis.util; import static org.hamcrest.Matchers.containsString; @@ -50,6 +51,7 @@ import org.springframework.integration.test.util.TestUtils; /** * @author Gary Russell + * @author Konstantin Yakimov * @since 4.0 * */ @@ -79,13 +81,13 @@ public class RedisLockRegistryTests extends RedisAvailableTests { Lock lock = registry.obtain("foo"); lock.lock(); try { - assertNotNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNotNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } finally { lock.unlock(); } } - assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } @Test @@ -96,18 +98,18 @@ public class RedisLockRegistryTests extends RedisAvailableTests { Lock lock = registry.obtain("foo"); lock.lockInterruptibly(); try { - assertNotNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNotNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } finally { lock.unlock(); } } - assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } @Test @RedisAvailable - public void testRentrantLock() throws Exception { + public void testReentrantLock() throws Exception { RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); for (int i = 0; i < 10; i++) { Lock lock1 = registry.obtain("foo"); @@ -127,12 +129,12 @@ public class RedisLockRegistryTests extends RedisAvailableTests { lock1.unlock(); } } - assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } @Test @RedisAvailable - public void testRentrantLockInterruptibly() throws Exception { + public void testReentrantLockInterruptibly() throws Exception { RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); for (int i = 0; i < 10; i++) { Lock lock1 = registry.obtain("foo"); @@ -152,7 +154,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { lock1.unlock(); } } - assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } @Test @@ -177,7 +179,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { lock1.unlock(); } } - assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } @Test @@ -210,7 +212,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { Object ise = result.get(10, TimeUnit.SECONDS); assertThat(ise, instanceOf(IllegalStateException.class)); assertThat(((Exception) ise).getMessage(), containsString("Lock is not locked")); - assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } @Test @@ -223,7 +225,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { final CountDownLatch latch2 = new CountDownLatch(1); final CountDownLatch latch3 = new CountDownLatch(1); lock1.lockInterruptibly(); - assertNotNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNotNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); Executors.newSingleThreadExecutor().execute(new Runnable() { @Override @@ -232,7 +234,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { try { latch1.countDown(); lock2.lockInterruptibly(); - assertNotNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNotNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); latch2.await(10, TimeUnit.SECONDS); locked.set(true); } @@ -251,7 +253,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { latch2.countDown(); assertTrue(latch3.await(10, TimeUnit.SECONDS)); assertTrue(locked.get()); - assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } @Test @@ -265,7 +267,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { final CountDownLatch latch2 = new CountDownLatch(1); final CountDownLatch latch3 = new CountDownLatch(1); lock1.lockInterruptibly(); - assertNotNull(TestUtils.getPropertyValue(registry1, "threadLocks", ThreadLocal.class).get()); + assertNotNull(TestUtils.getPropertyValue(registry1, "hardThreadLocks", ThreadLocal.class).get()); Executors.newSingleThreadExecutor().execute(new Runnable() { @Override @@ -274,7 +276,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { try { latch1.countDown(); lock2.lockInterruptibly(); - assertNotNull(TestUtils.getPropertyValue(registry2, "threadLocks", ThreadLocal.class).get()); + assertNotNull(TestUtils.getPropertyValue(registry2, "hardThreadLocks", ThreadLocal.class).get()); latch2.await(10, TimeUnit.SECONDS); locked.set(true); } @@ -293,8 +295,8 @@ public class RedisLockRegistryTests extends RedisAvailableTests { latch2.countDown(); assertTrue(latch3.await(10, TimeUnit.SECONDS)); assertTrue(locked.get()); - assertNull(TestUtils.getPropertyValue(registry1, "threadLocks", ThreadLocal.class).get()); - assertNull(TestUtils.getPropertyValue(registry2, "threadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry1, "hardThreadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry2, "hardThreadLocks", ThreadLocal.class).get()); } @Test @@ -325,7 +327,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { Object ise = result.get(10, TimeUnit.SECONDS); assertThat(ise, instanceOf(IllegalStateException.class)); assertThat(((Exception) ise).getMessage(), containsString("Lock is owned by")); - assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } @Test @@ -343,8 +345,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { foo.unlock(); bar.unlock(); baz.unlock(); - System.out.println(locks.iterator().next()); - assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } @Test @@ -353,7 +354,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests", 1000); Lock foo = registry.obtain("foo"); foo.lockInterruptibly(); - this.waitForExpire("foo"); + waitForExpire("foo"); try { foo.unlock(); fail("Expected exception"); @@ -361,7 +362,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { catch (IllegalStateException e) { assertThat(e.getMessage(), containsString("Lock was released due to expiration")); } - assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } @Test @@ -370,7 +371,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests", 1000); Lock foo1 = registry.obtain("foo"); foo1.lockInterruptibly(); - this.waitForExpire("foo"); + waitForExpire("foo"); Lock foo2 = registry.obtain("foo"); assertNotSame(foo1, foo2); foo2.lockInterruptibly(); @@ -383,7 +384,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { assertThat(e.getMessage(), containsString("lock in store:")); } foo2.unlock(); - assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); } @Test @@ -391,6 +392,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { public void testEquals() throws Exception { RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); RedisLockRegistry registry1 = new RedisLockRegistry(connectionFactory, "rlrTests"); + registry1.setUseWeakReferences(true); RedisLockRegistry registry2 = new RedisLockRegistry(connectionFactory, "rlrTests"); RedisLockRegistry registry3 = new RedisLockRegistry(connectionFactory, "rlrTests2"); Lock lock1 = registry1.obtain("foo"); @@ -419,6 +421,64 @@ public class RedisLockRegistryTests extends RedisAvailableTests { lock2.unlock(); } + @Test + @RedisAvailable + public void testThreadLocalListLeaks() { + RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests", 1000); + registry.setUseWeakReferences(true); + + for (int i = 0; i < 10; i++) { + registry.obtain("foo" + i); + } + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); + + for (int i = 0; i < 10; i++) { + Lock lock = registry.obtain("foo" + i); + lock.lock(); + } + assertEquals(10, + ((Collection) TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()).size()); + assertNull(TestUtils.getPropertyValue(registry, "weakThreadLocks", ThreadLocal.class).get()); + + for (int i = 0; i < 10; i++) { + Lock lock = registry.obtain("foo" + i); + assertNotNull(TestUtils.getPropertyValue(lock, "thread", Thread.class)); + lock.unlock(); + } + assertNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get()); + } + + @Test + @RedisAvailable + public void testExpireNotChanged() throws Exception { + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + final RedisLockRegistry registry = new RedisLockRegistry(connectionFactory, "rlrTests", 10000); + Lock lock = registry.obtain("foo"); + lock.lock(); + + Long expire = getExpire(registry, "foo"); + + Future result = Executors.newSingleThreadExecutor().submit(new Callable() { + + @Override + public Object call() throws Exception { + Lock lock2 = registry.obtain("foo"); + assertFalse(lock2.tryLock()); + return null; + } + + }); + result.get(); + assertEquals(expire, getExpire(registry, "foo")); + lock.unlock(); + } + + private Long getExpire(RedisLockRegistry registry, String lockKey) { + RedisTemplate template = this.createTemplate(); + String registryKey = TestUtils.getPropertyValue(registry, "registryKey", String.class); + return template.getExpire(registryKey + ":" + lockKey); + } + private void waitForExpire(String key) throws Exception { RedisTemplate template = this.createTemplate(); int n = 0; diff --git a/spring-integration-redis/src/test/resources/log4j.properties b/spring-integration-redis/src/test/resources/log4j.properties index 7ebfb6af9b..388b5d9c31 100644 --- a/spring-integration-redis/src/test/resources/log4j.properties +++ b/spring-integration-redis/src/test/resources/log4j.properties @@ -5,4 +5,4 @@ log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - <%m>%n log4j.category.org.springframework.integration=WARN -log4j.category.org.springframework.integration.redis=DEBUG +log4j.category.org.springframework.integration.redis=INFO