From 40a535b14090280ffffba341a195d273986fcfa6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sun, 6 Apr 2014 07:47:55 +0300 Subject: [PATCH] INT-3352 RedisLockRegistry JIRA: https://jira.spring.io/browse/INT-3352 Provid a global lock registry, initially for the correlating message handlers. INT-3252 Polishing - Use a discrete key for each lock and use Redis EXPIRE - Use an internal DefaultLockRegistry to improve latency when threads are using the same registry instance. INT-3352 More Polish - Put localLock.unlock() in a finally block. - Unlock localLock if acquiring the RedisLock fails. INT-3352 Polishing - PR Comments - PR Comments - Improve equals() INT-3352 Add Aggregator Integration Test - move lock registry to the `util` package - Add an aggregator integration test INT-3352 Add Distributed Aggregator Test - Simulates a distributed environment with a shared message store and distinct lock registries. INT-3352 More PR Comments INT-3352 Add Multiple Registry Tests - Add tests to verify expected behavior when different registries are used. - Polish javadocs. --- .gitignore | 1 + build.gradle | 2 + .../redis/util/RedisLockRegistry.java | 485 ++++++++++++++++++ .../integration/redis/util/package-info.java | 4 + .../redis/rules/RedisAvailableTests.java | 4 +- .../AggregatorWithRedisLocksTests-context.xml | 42 ++ .../util/AggregatorWithRedisLocksTests.java | 226 ++++++++ .../redis/util/RedisLockRegistryTests.java | 431 ++++++++++++++++ 8 files changed, 1193 insertions(+), 2 deletions(-) create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/util/package-info.java create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests-context.xml create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests.java create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java diff --git a/.gitignore b/.gitignore index 4b939dc033..11fee3f1a0 100644 --- a/.gitignore +++ b/.gitignore @@ -28,3 +28,4 @@ target vf.gf.dmn-* /atlassian-ide-plugin.xml hostkey.ser +.springBeans diff --git a/build.gradle b/build.gradle index 98e4fc3dd7..e1df0e6656 100644 --- a/build.gradle +++ b/build.gradle @@ -96,6 +96,7 @@ subprojects { subproject -> saajApiVersion = '1.3.5' saajImplVersion = '1.3.23' servletApiVersion = '3.1.0' + slf4jVersion = "1.7.6" smackVersion = '3.2.1' springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '1.3.1.RELEASE' springDataMongoVersion = '1.1.1.RELEASE' @@ -453,6 +454,7 @@ project('spring-integration-redis') { exclude group: 'org.springframework' } testCompile "com.lambdaworks:lettuce:$lettuceVersion" + testCompile "org.slf4j:slf4j-log4j12:$slf4jVersion" } } 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 new file mode 100644 index 0000000000..40215f3f57 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java @@ -0,0 +1,485 @@ +/* + * 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.integration.redis.util; + +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.nio.ByteBuffer; +import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Date; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.data.redis.serializer.SerializationException; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.integration.util.DefaultLockRegistry; +import org.springframework.integration.util.LockRegistry; +import org.springframework.util.Assert; + +/** + * Implementation of {@link LockRegistry} providing a distributed lock using Redis. + * Locks are stored under the key {@code registryKey:lockKey}. Locks expire after + * (default 60) seconds. Threads unlocking an + * expired lock will get an {@link IllegalStateException}. This should be + * considered as a critical error because it is possible the protected + * resources were compromised. + *

+ * Locks are reentrant. + *

+ * However, locks are scoped by the registry; a lock from a different registry with the + * same key (even if the registry uses the same 'registryKey') are different + * locks, and the second cannot be acquired by the same thread while the first is + * locked. + *

+ * Note: This is not intended for low latency applications. It is intended + * for resource locking across multiple JVMs. + * When a lock is released by a remote system, waiting threads may take up to 100ms + * to acquire the lock. + * A more performant version would need to get notifications from the Redis stores + * of key changes. This is currently only available using the SYNC command. + *

+ * This limitation will usually not apply when a lock is released within this registry, + * unless another system takes the lock after the local lock is acquired here. + * A {@link DefaultLockRegistry} is used internally to achieve this optimization. + *

+ * {@link Condition}s are not supported. + * + * @author Gary Russell + * @since 4.0 + * + */ +public final class RedisLockRegistry implements LockRegistry { + + private static final Log logger = LogFactory.getLog(LockRegistry.class); + + private static final byte[] hostName; + + private static final long DEFAULT_EXPIRE_AFTER = 60000; + + private final String registryKey; + + private final RedisTemplate redisTemplate; + + private final ThreadLocal> threadLocks = new ThreadLocal>(); + + private final long expireAfter; + + private final LockRegistry localRegistry = new DefaultLockRegistry(); + + static { + String host; + try { + host = InetAddress.getLocalHost().getHostName(); + } + catch (UnknownHostException e) { + host = "unknownHost"; + } + hostName = host.getBytes(); + } + + /** + * Constructs a lock registry with the default (60 second) lock expiration. + * @param connectionFactory The connection factory. + * @param registryKey The key prefix for locks. + */ + public RedisLockRegistry(RedisConnectionFactory connectionFactory, String registryKey) { + this(connectionFactory, registryKey, DEFAULT_EXPIRE_AFTER); + } + + /** + * Constructs a lock registry with the supplied lock expiration. + * @param connectionFactory The connection factory. + * @param registryKey The key prefix for locks. + * @param expireAfter The expiration in milliseconds. + */ + public RedisLockRegistry(RedisConnectionFactory connectionFactory, String registryKey, long expireAfter) { + Assert.notNull(connectionFactory, "'connectionFactory' cannot be null"); + Assert.notNull(registryKey, "'registryKey' cannot be null"); + this.redisTemplate = new RedisTemplate(); + this.redisTemplate.setConnectionFactory(connectionFactory); + this.redisTemplate.setKeySerializer(new StringRedisSerializer()); + this.redisTemplate.setValueSerializer(new LockSerializer()); + this.redisTemplate.afterPropertiesSet(); + this.registryKey = registryKey; + this.expireAfter = expireAfter; + } + + @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; + } + } + /* + * 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(); + if (lockInStore == null || !lock.equals(lockInStore)) { + removeLockFromThreadLocal(locks, 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; + } + } + } + + public Collection listLocks() { + Set keys = this.redisTemplate.keys(this.registryKey + ":*"); + List locks = new ArrayList(keys.size()); + for (String key : keys) { + RedisLock lock = this.redisTemplate.boundValueOps(key).get(); + if (lock != null) { + locks.add(lock); + } + } + return locks; + } + + private class RedisLock implements Lock { + + private final String lockKey; + + private long lockedAt; + + private Thread thread; + + private String threadName; + + private byte[] lockHost; + + private int reLock; + + private RedisLock(String lockKey) { + this.lockKey = lockKey; + this.lockHost = RedisLockRegistry.hostName; + } + + private String getLockKey() { + return lockKey; + } + + @Override + public void lock() { + Lock localLock = RedisLockRegistry.this.localRegistry.obtain(lockKey); + localLock.lock(); + try { + while (true) { + try { + while (!this.obtainLock()) { + Thread.sleep(100); + } + break; + } + catch (InterruptedException e) { + /* + * This method must be uninterruptible so catch and ignore + * interrupts and only break out of the while loop when + * we get the lock. + */ + } + } + } + catch (Exception e) { + localLock.unlock(); + throw new RuntimeException(e); + } + } + + @Override + public void lockInterruptibly() throws InterruptedException { + Lock localLock = RedisLockRegistry.this.localRegistry.obtain(lockKey); + localLock.lockInterruptibly(); + try { + while (!this.obtainLock()) { + Thread.sleep(100); + } + } + catch (InterruptedException ie) { + localLock.unlock(); + throw ie; + } + catch (Exception e) { + localLock.unlock(); + throw new RuntimeException(e); + } + } + + @Override + public boolean tryLock() { + Lock localLock = RedisLockRegistry.this.localRegistry.obtain(lockKey); + try { + if (!localLock.tryLock()) { + return false; + } + boolean obtainedLock = this.obtainLock(); + if (!obtainedLock) { + localLock.unlock(); + } + return obtainedLock; + } + catch (Exception e) { + localLock.unlock(); + throw new RuntimeException(e); + } + } + + private boolean obtainLock() { + Thread currentThread = Thread.currentThread(); + if (currentThread.equals(this.thread)) { + this.reLock++; + return true; + } + /* + * Set these now so they will be persisted if successful. + */ + this.lockedAt = System.currentTimeMillis(); + this.threadName = currentThread.getName(); + Boolean success = RedisLockRegistry.this.redisTemplate.boundValueOps( + constructLockKey()).setIfAbsent(this); + if (!success) { + this.lockedAt = 0; + this.threadName = null; + } + else { + this.thread = currentThread; + RedisLockRegistry.this.redisTemplate.expire(constructLockKey(), + RedisLockRegistry.this.expireAfter, TimeUnit.MILLISECONDS); + if (logger.isDebugEnabled()) { + logger.debug("New lock; " + this.toString()); + } + } + return success; + } + + @Override + public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { + Lock localLock = RedisLockRegistry.this.localRegistry.obtain(lockKey); + if (!localLock.tryLock(time, unit)) { + return false; + } + try { + long expire = System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(time, unit); + boolean acquired = false; + while (!(acquired = this.obtainLock()) && System.currentTimeMillis() < expire) { + Thread.sleep(100); + } + if (!acquired) { + localLock.unlock(); + } + return acquired; + } + catch (Exception e) { + localLock.unlock(); + throw new RuntimeException(e); + } + } + + @Override + public void unlock() { + if (!Thread.currentThread().equals(this.thread)) { + if (this.thread == null) { + throw new IllegalStateException("Lock is not locked; " + this.toString()); + } + 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(); + } + } + this.assertLockInRedisIsUnchanged(); + RedisLockRegistry.this.redisTemplate.delete(constructLockKey()); + if (logger.isDebugEnabled()) { + logger.debug("Released lock; " + this.toString()); + } + this.thread = null; + this.reLock = 0; + } + } + finally { + Lock localLock = RedisLockRegistry.this.localRegistry.obtain(lockKey); + localLock.unlock(); + } + } + + private void assertLockInRedisIsUnchanged() { + RedisLock lockInStore = RedisLockRegistry.this.redisTemplate.boundValueOps( + constructLockKey()).get(); + if (lockInStore == null || !this.equals(lockInStore)) { + throw new IllegalStateException("Lock was released due to expiration; " + this.toString() + + (lockInStore == null ? "" : "; lock in store: " + lockInStore.toString())); + } + } + + private String constructLockKey() { + return RedisLockRegistry.this.registryKey + ":" + this.lockKey; + } + + @Override + public Condition newCondition() { + throw new UnsupportedOperationException("Conditions are not supported"); + } + + @Override + public String toString() { + SimpleDateFormat dateFormat = new SimpleDateFormat("YYYY-MM-dd@HH:mm:ss.SSS"); + return "RedisLock [lockKey=" + constructLockKey() + + ",lockedAt=" + dateFormat.format(new Date(this.lockedAt)) + + ", thread=" + this.threadName + + ", lockHost=" + new String(this.lockHost) + + "]"; + } + + @Override + public int hashCode() { + final int prime = 31; + int result = 1; + result = prime * result + getOuterType().hashCode(); + result = prime * result + Arrays.hashCode(lockHost); + result = prime * result + ((lockKey == null) ? 0 : lockKey.hashCode()); + result = prime * result + (int) (lockedAt ^ (lockedAt >>> 32)); + result = prime * result + ((threadName == null) ? 0 : threadName.hashCode()); + return result; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null) { + return false; + } + if (getClass() != obj.getClass()) { + return false; + } + RedisLock other = (RedisLock) obj; + if (!getOuterType().equals(other.getOuterType())) { + return false; + } + if (!Arrays.equals(lockHost, other.lockHost)) { + return false; + } + if (!lockKey.equals(other.lockKey)) { + return false; + } + if (lockedAt != other.lockedAt) { + return false; + } + if (threadName == null) { + if (other.threadName != null) { + return false; + } + } + else if (!threadName.equals(other.threadName)) { + return false; + } + return true; + } + + private RedisLockRegistry getOuterType() { + return RedisLockRegistry.this; + } + + } + + private class LockSerializer implements RedisSerializer { + + @Override + public byte[] serialize(RedisLock t) throws SerializationException { + int hostLength = t.lockHost.length; + int keyLength = t.lockKey.length(); + int threadNameLength = t.threadName.length(); + byte[] value = new byte[1 + hostLength + + 1 + keyLength + + 1 + threadNameLength + 8]; + ByteBuffer buff = ByteBuffer.wrap(value); + buff.put((byte) hostLength) + .put(t.lockHost) + .put((byte) keyLength) + .put(t.lockKey.getBytes()) + .put((byte) threadNameLength) + .put(t.threadName.getBytes()) + .putLong(t.lockedAt); + return value; + } + + @Override + public RedisLock deserialize(byte[] bytes) throws SerializationException { + if (bytes == null) { + return null; + } + ByteBuffer buff = ByteBuffer.wrap(bytes); + byte[] host = new byte[buff.get()]; + buff.get(host); + byte[] lockKey = new byte[buff.get()]; + buff.get(lockKey); + byte[] threadName = new byte[buff.get()]; + buff.get(threadName); + long lockedAt = buff.getLong(); + RedisLock lock = new RedisLock(new String(lockKey)); + lock.lockedAt = lockedAt; + lock.lockHost = host; + lock.threadName = new String(threadName); + return lock; + } + + } + +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/package-info.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/package-info.java new file mode 100644 index 0000000000..768b316bd2 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/util/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides utility classes. + */ +package org.springframework.integration.redis.util; diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java index 96d301d9b2..6a82c43e63 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-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. @@ -44,7 +44,7 @@ public class RedisAvailableTests { @Rule public RedisAvailableRule redisAvailableRule = new RedisAvailableRule(); - protected RedisConnectionFactory getConnectionFactoryForTest(){ + protected RedisConnectionFactory getConnectionFactoryForTest() { LettuceConnectionFactory connectionFactory = RedisAvailableRule.connectionFactoryResource.get(); RedisTemplate rt = new RedisTemplate(); rt.setConnectionFactory(connectionFactory); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests-context.xml new file mode 100644 index 0000000000..4b141100ab --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests-context.xml @@ -0,0 +1,42 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests.java new file mode 100644 index 0000000000..63c8216faf --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/AggregatorWithRedisLocksTests.java @@ -0,0 +1,226 @@ +/* + * 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.integration.redis.util; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.aggregator.ReleaseStrategy; +import org.springframework.integration.redis.rules.RedisAvailable; +import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.integration.store.MessageGroup; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 4.0 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class AggregatorWithRedisLocksTests extends RedisAvailableTests { + + @Autowired + private LatchingReleaseStrategy releaseStrategy; + + @Autowired + private MessageChannel in; + + @Autowired + private MessageChannel in2; + + @Autowired + private PollableChannel out; + + private volatile Exception exception; + + private RedisTemplate template; + + @Before + @After + public void setup() { + this.template = this.createTemplate(); + Set keys = template.keys("aggregatorWithRedisLocksTests:*"); + for (String key : keys) { + template.delete(key); + } + } + + @Test + @RedisAvailable + public void testLockSingleGroup() throws Exception { + this.releaseStrategy.reset(1); + Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 1)); + Executors.newSingleThreadExecutor().execute(asyncSend("bar", 2, 1)); + assertTrue(this.releaseStrategy.latch2.await(10, TimeUnit.SECONDS)); + assertEquals(1, this.template.keys("aggregatorWithRedisLocksTests:*").size()); + this.releaseStrategy.latch1.countDown(); + assertNotNull(this.out.receive(10000)); + assertEquals(1, this.releaseStrategy.maxCallers.get()); + this.assertNoLocksAfterTest(); + assertNull("Unexpected exception:" + (this.exception != null ? this.exception.toString() : ""), this.exception); + } + + @Test + @RedisAvailable + public void testLockThreeGroups() throws Exception { + this.releaseStrategy.reset(3); + Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 1)); + Executors.newSingleThreadExecutor().execute(asyncSend("bar", 2, 1)); + Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 2)); + Executors.newSingleThreadExecutor().execute(asyncSend("bar", 2, 2)); + Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 3)); + Executors.newSingleThreadExecutor().execute(asyncSend("bar", 2, 3)); + assertTrue(this.releaseStrategy.latch2.await(10, TimeUnit.SECONDS)); + assertEquals(3, this.template.keys("aggregatorWithRedisLocksTests:*").size()); + this.releaseStrategy.latch1.countDown(); + this.releaseStrategy.latch1.countDown(); + this.releaseStrategy.latch1.countDown(); + assertNotNull(this.out.receive(10000)); + assertNotNull(this.out.receive(10000)); + assertNotNull(this.out.receive(10000)); + assertEquals(3, this.releaseStrategy.maxCallers.get()); + this.assertNoLocksAfterTest(); + assertNull("Unexpected exception:" + (this.exception != null ? this.exception.toString() : ""), this.exception); + } + + @Test + @RedisAvailable + public void testDistributedAggregator() throws Exception { + this.releaseStrategy.reset(1); + Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 1)); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + try { + in2.send(new GenericMessage("bar", stubHeaders(2, 2, 1))); + } + catch (Exception e) { + e.printStackTrace(); + exception = e; + } + } + }); + assertTrue(this.releaseStrategy.latch2.await(10, TimeUnit.SECONDS)); + assertEquals(1, this.template.keys("aggregatorWithRedisLocksTests:*").size()); + this.releaseStrategy.latch1.countDown(); + assertNotNull(this.out.receive(10000)); + assertEquals(1, this.releaseStrategy.maxCallers.get()); + this.assertNoLocksAfterTest(); + assertNull("Unexpected exception:" + (this.exception != null ? this.exception.toString() : ""), this.exception); + } + + private void assertNoLocksAfterTest() throws Exception { + int n = 0; + while (n++ < 100 && this.template.keys("aggregatorWithRedisLocksTests:*").size() > 0) { + Thread.sleep(100); + } + assertEquals(0, this.template.keys("aggregatorWithRedisLocksTests:*").size()); + } + + private Runnable asyncSend(final String payload, final int sequence, final int correlation) { + return new Runnable() { + + @Override + public void run() { + try { + in.send(new GenericMessage(payload, stubHeaders(sequence, 2, correlation))); + } + catch (Exception e) { + e.printStackTrace(); + exception = e; + } + } + }; + } + + private Map stubHeaders(int sequenceNumber, int sequenceSize, int correlationId) { + Map headers = new HashMap(); + headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber); + headers.put(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize); + headers.put(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId); + return headers; + } + + private RedisTemplate createTemplate() { + RedisTemplate template = new RedisTemplate(); + template.setConnectionFactory(this.getConnectionFactoryForTest()); + template.setKeySerializer(new StringRedisSerializer()); + template.afterPropertiesSet(); + return template; + } + + public static class LatchingReleaseStrategy implements ReleaseStrategy { + + private volatile CountDownLatch latch1; + + private volatile CountDownLatch latch2; + + private volatile AtomicInteger callers; + + private volatile AtomicInteger maxCallers; + + @Override + public boolean canRelease(MessageGroup group) { + synchronized(this) { + this.callers.incrementAndGet(); + this.maxCallers.set(Math.max(this.maxCallers.get(), this.callers.get())); + } + this.latch2.countDown(); + try { + this.latch1.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + this.callers.decrementAndGet(); + return group.size() > 1; + } + + public void reset(int expectedConcurrency) { + this.latch1 = new CountDownLatch(expectedConcurrency); + this.latch2 = new CountDownLatch(expectedConcurrency); + this.callers = new AtomicInteger(); + this.maxCallers = new AtomicInteger(); + } + } +} 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 new file mode 100644 index 0000000000..6868accd87 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java @@ -0,0 +1,431 @@ +/* + * 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.integration.redis.util; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.Collection; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Lock; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.integration.redis.rules.RedisAvailable; +import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.integration.test.util.TestUtils; + +/** + * @author Gary Russell + * @since 4.0 + * + */ +public class RedisLockRegistryTests extends RedisAvailableTests { + + @Before + @After + public void shutDown() { + RedisTemplate template = this.createTemplate(); + template.delete("rlrTests"); + template.delete("rlrTests2"); + } + + private RedisTemplate createTemplate() { + RedisTemplate template = new RedisTemplate(); + template.setConnectionFactory(this.getConnectionFactoryForTest()); + template.setKeySerializer(new StringRedisSerializer()); + template.afterPropertiesSet(); + return template; + } + + @Test + @RedisAvailable + public void testLock() throws Exception { + RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); + for (int i = 0; i < 10; i++) { + Lock lock = registry.obtain("foo"); + lock.lock(); + try { + assertNotNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + } + finally { + lock.unlock(); + } + } + assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + } + + @Test + @RedisAvailable + public void testLockInterruptibly() throws Exception { + RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); + for (int i = 0; i < 10; i++) { + Lock lock = registry.obtain("foo"); + lock.lockInterruptibly(); + try { + assertNotNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + } + finally { + lock.unlock(); + } + } + assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + } + + @Test + @RedisAvailable + public void testRentrantLock() throws Exception { + RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); + for (int i = 0; i < 10; i++) { + Lock lock1 = registry.obtain("foo"); + lock1.lock(); + try { + Lock lock2 = registry.obtain("foo"); + assertSame(lock1, lock2); + lock2.lock(); + try { + + } + finally { + lock2.unlock(); + } + } + finally { + lock1.unlock(); + } + } + assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + } + + @Test + @RedisAvailable + public void testRentrantLockInterruptibly() throws Exception { + RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); + for (int i = 0; i < 10; i++) { + Lock lock1 = registry.obtain("foo"); + lock1.lockInterruptibly(); + try { + Lock lock2 = registry.obtain("foo"); + assertSame(lock1, lock2); + lock2.lockInterruptibly(); + try { + + } + finally { + lock2.unlock(); + } + } + finally { + lock1.unlock(); + } + } + assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + } + + @Test + @RedisAvailable + public void testTwoLocks() throws Exception { + RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); + for (int i = 0; i < 10; i++) { + Lock lock1 = registry.obtain("foo"); + lock1.lockInterruptibly(); + try { + Lock lock2 = registry.obtain("bar"); + assertNotSame(lock1, lock2); + lock2.lockInterruptibly(); + try { + + } + finally { + lock2.unlock(); + } + } + finally { + lock1.unlock(); + } + } + assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + } + + @Test + @RedisAvailable + public void testTwoThreadsSecondFailsToGetLock() throws Exception { + final RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); + final Lock lock1 = registry.obtain("foo"); + lock1.lockInterruptibly(); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch = new CountDownLatch(1); + Future result = Executors.newSingleThreadExecutor().submit(new Callable() { + + @Override + public Object call() throws Exception { + Lock lock2 = registry.obtain("foo"); + locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS)); + latch.countDown(); + try { + lock2.unlock(); + } + catch (IllegalStateException ise) { + return ise; + } + return null; + } + }); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock1.unlock(); + 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()); + } + + @Test + @RedisAvailable + public void testTwoThreads() throws Exception { + final RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); + final Lock lock1 = registry.obtain("foo"); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + final CountDownLatch latch3 = new CountDownLatch(1); + lock1.lockInterruptibly(); + assertNotNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + Lock lock2 = registry.obtain("foo"); + try { + latch1.countDown(); + lock2.lockInterruptibly(); + assertNotNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + latch2.await(10, TimeUnit.SECONDS); + locked.set(true); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + finally { + lock2.unlock(); + latch3.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock1.unlock(); + latch2.countDown(); + assertTrue(latch3.await(10, TimeUnit.SECONDS)); + assertTrue(locked.get()); + assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + } + + @Test + @RedisAvailable + public void testTwoThreadsDifferentRegistries() throws Exception { + final RedisLockRegistry registry1 = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); + final RedisLockRegistry registry2 = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); + final Lock lock1 = registry1.obtain("foo"); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + final CountDownLatch latch3 = new CountDownLatch(1); + lock1.lockInterruptibly(); + assertNotNull(TestUtils.getPropertyValue(registry1, "threadLocks", ThreadLocal.class).get()); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + Lock lock2 = registry2.obtain("foo"); + try { + latch1.countDown(); + lock2.lockInterruptibly(); + assertNotNull(TestUtils.getPropertyValue(registry2, "threadLocks", ThreadLocal.class).get()); + latch2.await(10, TimeUnit.SECONDS); + locked.set(true); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + finally { + lock2.unlock(); + latch3.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock1.unlock(); + 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()); + } + + @Test + @RedisAvailable + public void testTwoThreadsWrongOneUnlocks() throws Exception { + final RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); + final Lock lock = registry.obtain("foo"); + lock.lockInterruptibly(); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch = new CountDownLatch(1); + Future result = Executors.newSingleThreadExecutor().submit(new Callable() { + + @Override + public Object call() throws Exception { + try { + lock.unlock(); + } + catch (IllegalStateException ise) { + latch.countDown(); + return ise; + } + return null; + } + }); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock.unlock(); + 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()); + } + + @Test + @RedisAvailable + public void testList() throws Exception { + RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests"); + Lock foo = registry.obtain("foo"); + foo.lockInterruptibly(); + Lock bar = registry.obtain("bar"); + bar.lockInterruptibly(); + Lock baz = registry.obtain("baz"); + baz.lockInterruptibly(); + Collection locks = registry.listLocks(); + assertEquals(3, locks.size()); + foo.unlock(); + bar.unlock(); + baz.unlock(); + System.out.println(locks.iterator().next()); + assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + } + + @Test + @RedisAvailable + public void testExpireNoLockInStore() throws Exception { + RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests", 1000); + Lock foo = registry.obtain("foo"); + foo.lockInterruptibly(); + this.waitForExpire("foo"); + try { + foo.unlock(); + fail("Expected exception"); + } + catch (IllegalStateException e) { + assertThat(e.getMessage(), containsString("Lock was released due to expiration")); + } + assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + } + + @Test + @RedisAvailable + public void testExpireNewLockInStore() throws Exception { + RedisLockRegistry registry = new RedisLockRegistry(this.getConnectionFactoryForTest(), "rlrTests", 1000); + Lock foo1 = registry.obtain("foo"); + foo1.lockInterruptibly(); + this.waitForExpire("foo"); + Lock foo2 = registry.obtain("foo"); + assertNotSame(foo1, foo2); + foo2.lockInterruptibly(); + try { + foo1.unlock(); + fail("Expected exception"); + } + catch (IllegalStateException e) { + assertThat(e.getMessage(), containsString("Lock was released due to expiration")); + assertThat(e.getMessage(), containsString("lock in store:")); + } + foo2.unlock(); + assertNull(TestUtils.getPropertyValue(registry, "threadLocks", ThreadLocal.class).get()); + } + + @Test + @RedisAvailable + public void testEquals() throws Exception { + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); + RedisLockRegistry registry1 = new RedisLockRegistry(connectionFactory, "rlrTests"); + RedisLockRegistry registry2 = new RedisLockRegistry(connectionFactory, "rlrTests"); + RedisLockRegistry registry3 = new RedisLockRegistry(connectionFactory, "rlrTests2"); + Lock lock1 = registry1.obtain("foo"); + Lock lock2 = registry1.obtain("foo"); + assertEquals(lock1, lock2); + lock1.lock(); + lock2.lock(); + assertEquals(lock1, lock2); + lock1.unlock(); + lock2.unlock(); + assertEquals(lock1, lock2); + + lock1 = registry1.obtain("foo"); + lock2 = registry2.obtain("foo"); + assertNotEquals(lock1, lock2); + lock1.lock(); + assertFalse(lock2.tryLock()); + lock1.unlock(); + + lock1 = registry1.obtain("foo"); + lock2 = registry3.obtain("foo"); + assertNotEquals(lock1, lock2); + lock1.lock(); + lock2.lock(); + lock1.unlock(); + lock2.unlock(); + } + + private void waitForExpire(String key) throws Exception { + RedisTemplate template = this.createTemplate(); + int n = 0; + while (n++ < 100 && template.keys("rlrTests:" + key).size() > 0) { + Thread.sleep(100); + } + assertTrue(key + " key did not expire", n < 100); + } + +}