From db611028dac0aff6eae7ea828afe50d27dacb156 Mon Sep 17 00:00:00 2001 From: Unseok Kim Date: Fri, 12 Nov 2021 06:36:27 +0900 Subject: [PATCH] GH-3672: Clean up Jdbc & ZK LockRegistry caches Fixes https://github.com/spring-projects/spring-integration/issues/3672 * Clean up `JdbcLockRegistry`, `ZookeeperLockRegistry` cache automatically * setCapacity(int capacity) to cacheCapacity(int capacity) * field rename `capacity`to `cacheCapacity`, add static --- .../jdbc/lock/JdbcLockRegistry.java | 49 +++-- .../jdbc/lock/JdbcLockRegistryTests.java | 175 ++++++++++++++++- .../redis/util/RedisLockRegistry.java | 10 +- .../redis/util/RedisLockRegistryTests.java | 15 +- .../zookeeper/lock/ZookeeperLockRegistry.java | 49 +++-- .../zookeeper/lock/ZkLockRegistryTests.java | 183 +++++++++++++++++- src/reference/asciidoc/jdbc.adoc | 3 + src/reference/asciidoc/redis.adoc | 2 +- src/reference/asciidoc/zookeeper.adoc | 3 + 9 files changed, 447 insertions(+), 42 deletions(-) diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java index 1a50141350..152242af2d 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java @@ -17,10 +17,9 @@ package org.springframework.integration.jdbc.lock; import java.time.Duration; -import java.util.Iterator; +import java.util.LinkedHashMap; import java.util.Map; import java.util.Map.Entry; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; @@ -54,6 +53,7 @@ import org.springframework.util.Assert; * @author Stefan Vassilev * @author Olivier Hubaut * @author Fran Aranda + * @author Unseok Kim * * @since 4.3 */ @@ -61,12 +61,24 @@ public class JdbcLockRegistry implements ExpirableLockRegistry, RenewableLockReg private static final int DEFAULT_IDLE = 100; - private final Map locks = new ConcurrentHashMap<>(); + private static final int DEFAULT_CAPACITY = 100_000; + + private final Map locks = + new LinkedHashMap(16, 0.75F, true) { + + @Override + protected boolean removeEldestEntry(Entry eldest) { + return size() > JdbcLockRegistry.this.cacheCapacity; + } + + }; private final LockRepository client; private Duration idleBetweenTries = Duration.ofMillis(DEFAULT_IDLE); + private int cacheCapacity = DEFAULT_CAPACITY; + public JdbcLockRegistry(LockRepository client) { this.client = client; } @@ -82,11 +94,22 @@ public class JdbcLockRegistry implements ExpirableLockRegistry, RenewableLockReg this.idleBetweenTries = idleBetweenTries; } + /** + * Set the capacity of cached locks. + * @param cacheCapacity The capacity of cached lock, (default 100_000). + * @since 5.5.6 + */ + public void setCacheCapacity(int cacheCapacity) { + this.cacheCapacity = cacheCapacity; + } + @Override public Lock obtain(Object lockKey) { Assert.isInstanceOf(String.class, lockKey); String path = pathFor((String) lockKey); - return this.locks.computeIfAbsent(path, (key) -> new JdbcLock(this.client, this.idleBetweenTries, key)); + synchronized (this.locks) { + return this.locks.computeIfAbsent(path, key -> new JdbcLock(this.client, this.idleBetweenTries, key)); + } } private String pathFor(String input) { @@ -95,14 +118,13 @@ public class JdbcLockRegistry implements ExpirableLockRegistry, RenewableLockReg @Override public void expireUnusedOlderThan(long age) { - Iterator> iterator = this.locks.entrySet().iterator(); long now = System.currentTimeMillis(); - while (iterator.hasNext()) { - Entry entry = iterator.next(); - JdbcLock lock = entry.getValue(); - if (now - lock.getLastUsed() > age && !lock.isAcquiredInThisProcess()) { - iterator.remove(); - } + synchronized (this.locks) { + this.locks.entrySet() + .removeIf(entry -> { + JdbcLock lock = entry.getValue(); + return now - lock.getLastUsed() > age && !lock.isAcquiredInThisProcess(); + }); } } @@ -110,7 +132,10 @@ public class JdbcLockRegistry implements ExpirableLockRegistry, RenewableLockReg public void renewLock(Object lockKey) { Assert.isInstanceOf(String.class, lockKey); String path = pathFor((String) lockKey); - JdbcLock jdbcLock = this.locks.get(path); + JdbcLock jdbcLock; + synchronized (this.locks) { + jdbcLock = this.locks.get(path); + } if (jdbcLock == null) { throw new IllegalStateException("Could not found mutex at " + path); } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java index 94910b9e92..a1ea3cffc2 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/lock/JdbcLockRegistryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-2021 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. @@ -20,8 +20,12 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import java.util.Map; +import java.util.Queue; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.Lock; @@ -35,6 +39,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.util.UUIDConverter; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; @@ -43,6 +48,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; * @author Artem Bilan * @author Stefan Vassilev * @author Alexandre Strubel + * @author Unseok Kim * * @since 4.3 */ @@ -312,4 +318,171 @@ public class JdbcLockRegistryTests { .isThrownBy(() -> registry.renewLock("foo")); } + @Test + public void concurrentObtainCapacityTest() throws InterruptedException { + final int KEY_CNT = 500; + final int CAPACITY_CNT = 179; + final int THREAD_CNT = 4; + + final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT); + registry.setCacheCapacity(CAPACITY_CNT); + final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT); + + for (int i = 0; i < KEY_CNT; i++) { + int finalI = i; + executorService.submit(() -> { + countDownLatch.countDown(); + try { + countDownLatch.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + String keyId = "foo:" + finalI; + Lock obtain = registry.obtain(keyId); + obtain.lock(); + obtain.unlock(); + }); + } + executorService.shutdown(); + executorService.awaitTermination(5, TimeUnit.SECONDS); + + //capacity limit test + assertThat(getRegistryLocks(registry)).hasSize(CAPACITY_CNT); + + + registry.expireUnusedOlderThan(-1000); + assertThat(getRegistryLocks(registry)).isEmpty(); + } + + @Test + public void concurrentObtainRemoveOrderTest() throws InterruptedException { + final int THREAD_CNT = 2; + final int DUMMY_LOCK_CNT = 3; + + final int CAPACITY_CNT = THREAD_CNT; + + final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT); + registry.setCacheCapacity(CAPACITY_CNT); + final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT); + final Queue remainLockCheckQueue = new LinkedBlockingQueue<>(); + + //Removed due to capcity limit + for (int i = 0; i < DUMMY_LOCK_CNT; i++) { + Lock obtainLock0 = registry.obtain("foo:" + i); + obtainLock0.lock(); + obtainLock0.unlock(); + } + + for (int i = DUMMY_LOCK_CNT; i < THREAD_CNT + DUMMY_LOCK_CNT; i++) { + int finalI = i; + executorService.submit(() -> { + countDownLatch.countDown(); + try { + countDownLatch.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + String keyId = "foo:" + finalI; + remainLockCheckQueue.offer(toUUID(keyId)); + Lock obtain = registry.obtain(keyId); + obtain.lock(); + obtain.unlock(); + }); + } + + executorService.shutdown(); + executorService.awaitTermination(5, TimeUnit.SECONDS); + + assertThat(getRegistryLocks(registry)).containsKeys( + remainLockCheckQueue.toArray(new String[remainLockCheckQueue.size()])); + } + + @Test + public void concurrentObtainAccessRemoveOrderTest() throws InterruptedException { + final int THREAD_CNT = 2; + final int DUMMY_LOCK_CNT = 3; + + final int CAPACITY_CNT = THREAD_CNT + 1; + final String REMAIN_DUMMY_LOCK_KEY = "foo:1"; + + final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT); + registry.setCacheCapacity(CAPACITY_CNT); + final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT); + final Queue remainLockCheckQueue = new LinkedBlockingQueue<>(); + + //Removed due to capcity limit + for (int i = 0; i < DUMMY_LOCK_CNT; i++) { + Lock obtainLock0 = registry.obtain("foo:" + i); + obtainLock0.lock(); + obtainLock0.unlock(); + } + + Lock obtainLock0 = registry.obtain(REMAIN_DUMMY_LOCK_KEY); + obtainLock0.lock(); + obtainLock0.unlock(); + remainLockCheckQueue.offer(toUUID(REMAIN_DUMMY_LOCK_KEY)); + + for (int i = DUMMY_LOCK_CNT; i < THREAD_CNT + DUMMY_LOCK_CNT; i++) { + int finalI = i; + executorService.submit(() -> { + countDownLatch.countDown(); + try { + countDownLatch.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + String keyId = "foo:" + finalI; + remainLockCheckQueue.offer(toUUID(keyId)); + Lock obtain = registry.obtain(keyId); + obtain.lock(); + obtain.unlock(); + }); + } + + executorService.shutdown(); + executorService.awaitTermination(5, TimeUnit.SECONDS); + + assertThat(getRegistryLocks(registry)).containsKeys( + remainLockCheckQueue.toArray(new String[remainLockCheckQueue.size()])); + } + + @Test + public void setCapacityTest() { + final int CAPACITY_CNT = 4; + registry.setCacheCapacity(CAPACITY_CNT); + + registry.obtain("foo:1"); + registry.obtain("foo:2"); + registry.obtain("foo:3"); + + //capacity 4->3 + registry.setCacheCapacity(CAPACITY_CNT - 1); + + registry.obtain("foo:4"); + + assertThat(getRegistryLocks(registry)).hasSize(3); + assertThat(getRegistryLocks(registry)).containsKeys(toUUID("foo:2"), + toUUID("foo:3"), + toUUID("foo:4")); + + //capacity 3->4 + registry.setCacheCapacity(CAPACITY_CNT); + registry.obtain("foo:5"); + assertThat(getRegistryLocks(registry)).hasSize(4); + assertThat(getRegistryLocks(registry)).containsKeys(toUUID("foo:3"), + toUUID("foo:4"), + toUUID("foo:5")); + } + + @SuppressWarnings("unchecked") + private static Map getRegistryLocks(JdbcLockRegistry registry) { + return TestUtils.getPropertyValue(registry, "locks", Map.class); + } + + private static String toUUID(String key) { + return UUIDConverter.getUUID(key).toString(); + } } 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 56ee7faeba..719f387c97 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 @@ -99,7 +99,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl @Override protected boolean removeEldestEntry(Entry eldest) { - return size() > RedisLockRegistry.this.capacity; + return size() > RedisLockRegistry.this.cacheCapacity; } }; @@ -114,7 +114,7 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl private final long expireAfter; - private int capacity = DEFAULT_CAPACITY; + private int cacheCapacity = DEFAULT_CAPACITY; /** * An {@link ExecutorService} to call {@link StringRedisTemplate#delete} in @@ -168,11 +168,11 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl /** * Set the capacity of cached locks. - * @param capacity The capacity of cached lock, (default 100_000). + * @param cacheCapacity The capacity of cached lock, (default 100_000). * @since 5.5.6 */ - public void setCapacity(int capacity) { - this.capacity = capacity; + public void setCacheCapacity(int cacheCapacity) { + this.cacheCapacity = cacheCapacity; } @Override 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 6ca5cf0e53..dfe3d25859 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 @@ -449,7 +449,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT); final RedisConnectionFactory connectionFactory = getConnectionFactoryForTest(); final RedisLockRegistry registry = new RedisLockRegistry(connectionFactory, this.registryKey, 10000); - registry.setCapacity(CAPACITY_CNT); + registry.setCacheCapacity(CAPACITY_CNT); final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT); for (int i = 0; i < KEY_CNT; i++) { @@ -490,7 +490,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT); final RedisConnectionFactory connectionFactory = getConnectionFactoryForTest(); final RedisLockRegistry registry = new RedisLockRegistry(connectionFactory, this.registryKey, 10000); - registry.setCapacity(CAPACITY_CNT); + registry.setCacheCapacity(CAPACITY_CNT); final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT); final Queue remainLockCheckQueue = new LinkedBlockingQueue<>(); @@ -538,7 +538,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT); final RedisConnectionFactory connectionFactory = getConnectionFactoryForTest(); final RedisLockRegistry registry = new RedisLockRegistry(connectionFactory, this.registryKey, 10000); - registry.setCapacity(CAPACITY_CNT); + registry.setCacheCapacity(CAPACITY_CNT); final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT); final Queue remainLockCheckQueue = new LinkedBlockingQueue<>(); @@ -585,14 +585,14 @@ public class RedisLockRegistryTests extends RedisAvailableTests { final int CAPACITY_CNT = 4; final RedisConnectionFactory connectionFactory = getConnectionFactoryForTest(); final RedisLockRegistry registry = new RedisLockRegistry(connectionFactory, this.registryKey, 10000); - registry.setCapacity(CAPACITY_CNT); + registry.setCacheCapacity(CAPACITY_CNT); registry.obtain("foo:1"); registry.obtain("foo:2"); registry.obtain("foo:3"); //capacity 4->3 - registry.setCapacity(CAPACITY_CNT - 1); + registry.setCacheCapacity(CAPACITY_CNT - 1); registry.obtain("foo:4"); @@ -600,7 +600,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { assertThat(getRedisLockRegistryLocks(registry)).containsKeys("foo:2", "foo:3", "foo:4"); //capacity 3->4 - registry.setCapacity(CAPACITY_CNT); + registry.setCacheCapacity(CAPACITY_CNT); registry.obtain("foo:5"); assertThat(TestUtils.getPropertyValue(registry, "locks", Map.class).size()).isEqualTo(4); assertThat(getRedisLockRegistryLocks(registry)).containsKeys("foo:3", "foo:4", "foo:5"); @@ -636,8 +636,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests { } @SuppressWarnings("unchecked") - private Map getRedisLockRegistryLocks(RedisLockRegistry registry) { + private static Map getRedisLockRegistryLocks(RedisLockRegistry registry) { return TestUtils.getPropertyValue(registry, "locks", Map.class); } - } diff --git a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java index 450c557b1c..4ad038824d 100644 --- a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java +++ b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2021 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. @@ -16,10 +16,9 @@ package org.springframework.integration.zookeeper.lock; -import java.util.Iterator; +import java.util.LinkedHashMap; import java.util.Map; import java.util.Map.Entry; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -44,6 +43,7 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Artem Bilan * @author Vedran Pavic + * @author Unseok Kim * * @since 4.2 * @@ -56,7 +56,17 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableB private final KeyToPathStrategy keyToPath; - private final Map locks = new ConcurrentHashMap<>(); + private static final int DEFAULT_CAPACITY = 30_000; + + private final Map locks = + new LinkedHashMap(16, 0.75F, true) { + + @Override + protected boolean removeEldestEntry(Entry eldest) { + return size() > ZookeeperLockRegistry.this.cacheCapacity; + } + + }; private final boolean trackingTime; @@ -71,6 +81,8 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableB private boolean mutexTaskExecutorExplicitlySet; + private int cacheCapacity = DEFAULT_CAPACITY; + /** * Construct a lock registry using the default {@link KeyToPathStrategy} which * simple appends the key to '/SpringIntegration-LockRegistry/'. @@ -120,11 +132,23 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableB this.mutexTaskExecutorExplicitlySet = true; } + /** + * Set the capacity of cached locks. + * @param cacheCapacity The capacity of cached lock, (default 30_000). + * @since 5.5.6 + */ + public void setCacheCapacity(int cacheCapacity) { + this.cacheCapacity = cacheCapacity; + } + @Override public Lock obtain(Object lockKey) { Assert.isInstanceOf(String.class, lockKey); String path = this.keyToPath.pathFor((String) lockKey); - ZkLock lock = this.locks.computeIfAbsent(path, p -> new ZkLock(this.client, this.mutexTaskExecutor, p)); + ZkLock lock; + synchronized (this.locks) { + lock = this.locks.computeIfAbsent(path, p -> new ZkLock(this.client, this.mutexTaskExecutor, p)); + } if (this.trackingTime) { lock.setLastUsed(System.currentTimeMillis()); } @@ -143,15 +167,14 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableB if (!this.trackingTime) { throw new IllegalStateException("Ths KeyToPathStrategy is bounded; expiry is not supported"); } - Iterator> iterator = this.locks.entrySet().iterator(); + long now = System.currentTimeMillis(); - while (iterator.hasNext()) { - Entry entry = iterator.next(); - ZkLock lock = entry.getValue(); - if (now - lock.getLastUsed() > age - && !lock.isAcquiredInThisProcess()) { - iterator.remove(); - } + synchronized (this.locks) { + this.locks.entrySet() + .removeIf(entry -> { + ZkLock lock = entry.getValue(); + return now - lock.getLastUsed() > age && !lock.isAcquiredInThisProcess(); + }); } } diff --git a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java index 8964f7effd..b2415a6c7a 100644 --- a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java +++ b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-2021 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. @@ -20,9 +20,12 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.fail; import java.util.Map; +import java.util.Queue; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.Lock; @@ -35,7 +38,8 @@ import org.springframework.messaging.MessagingException; /** * @author Gary Russell - * @author Artem Bilan\ + * @author Artem Bilan + * @author Unseok Kim * * @since 4.2 * @@ -336,4 +340,179 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport { registry.destroy(); } + @Test + public void concurrentObtainCapacityTest() throws InterruptedException { + final int KEY_CNT = 50; + final int CAPACITY_CNT = 17; + final int THREAD_CNT = 4; + + final CountDownLatch maincountDownLatch = new CountDownLatch(KEY_CNT); + final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT); + final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + registry.setCacheCapacity(CAPACITY_CNT); + final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT); + + for (int i = 0; i < KEY_CNT; i++) { + int finalI = i; + executorService.submit(() -> { + countDownLatch.countDown(); + try { + countDownLatch.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + String keyId = "foo:" + finalI; + Lock obtain = registry.obtain(keyId); + maincountDownLatch.countDown(); + obtain.lock(); + obtain.unlock(); + }); + } + executorService.shutdown(); + maincountDownLatch.await(); + executorService.awaitTermination(5, TimeUnit.SECONDS); + + //capacity limit test + assertThat(getRegistryLocks(registry)).hasSize(CAPACITY_CNT); + + + registry.expireUnusedOlderThan(-1000); + assertThat(getRegistryLocks(registry)).isEmpty(); + registry.destroy(); + } + + @Test + public void concurrentObtainRemoveOrderTest() throws InterruptedException { + final int THREAD_CNT = 2; + final int DUMMY_LOCK_CNT = 3; + + final int CAPACITY_CNT = THREAD_CNT; + + final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT); + final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + registry.setCacheCapacity(CAPACITY_CNT); + final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT); + final Queue remainLockCheckQueue = new LinkedBlockingQueue<>(); + + //Removed due to capcity limit + for (int i = 0; i < DUMMY_LOCK_CNT; i++) { + Lock obtainLock0 = registry.obtain("foo:" + i); + obtainLock0.lock(); + obtainLock0.unlock(); + } + + for (int i = DUMMY_LOCK_CNT; i < THREAD_CNT + DUMMY_LOCK_CNT; i++) { + int finalI = i; + executorService.submit(() -> { + countDownLatch.countDown(); + try { + countDownLatch.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + String keyId = "foo:" + finalI; + remainLockCheckQueue.offer(toKey(keyId)); + Lock obtain = registry.obtain(keyId); + obtain.lock(); + obtain.unlock(); + }); + } + + executorService.shutdown(); + executorService.awaitTermination(5, TimeUnit.SECONDS); + + assertThat(getRegistryLocks(registry)).containsKeys( + remainLockCheckQueue.toArray(new String[remainLockCheckQueue.size()])); + registry.destroy(); + } + + @Test + public void concurrentObtainAccessRemoveOrderTest() throws InterruptedException { + final int THREAD_CNT = 2; + final int DUMMY_LOCK_CNT = 3; + + final int CAPACITY_CNT = THREAD_CNT + 1; + final String REMAIN_DUMMY_LOCK_KEY = "foo:1"; + + final CountDownLatch countDownLatch = new CountDownLatch(THREAD_CNT); + final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + registry.setCacheCapacity(CAPACITY_CNT); + final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT); + final Queue remainLockCheckQueue = new LinkedBlockingQueue<>(); + + //Removed due to capcity limit + for (int i = 0; i < DUMMY_LOCK_CNT; i++) { + Lock obtainLock0 = registry.obtain("foo:" + i); + obtainLock0.lock(); + obtainLock0.unlock(); + } + + Lock obtainLock0 = registry.obtain(REMAIN_DUMMY_LOCK_KEY); + obtainLock0.lock(); + obtainLock0.unlock(); + remainLockCheckQueue.offer(toKey(REMAIN_DUMMY_LOCK_KEY)); + + for (int i = DUMMY_LOCK_CNT; i < THREAD_CNT + DUMMY_LOCK_CNT; i++) { + int finalI = i; + executorService.submit(() -> { + countDownLatch.countDown(); + try { + countDownLatch.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + String keyId = "foo:" + finalI; + remainLockCheckQueue.offer(toKey(keyId)); + Lock obtain = registry.obtain(keyId); + obtain.lock(); + obtain.unlock(); + }); + } + + executorService.shutdown(); + executorService.awaitTermination(5, TimeUnit.SECONDS); + + assertThat(getRegistryLocks(registry)).containsKeys( + remainLockCheckQueue.toArray(new String[remainLockCheckQueue.size()])); + registry.destroy(); + } + + @Test + public void setCapacityTest() { + final int CAPACITY_CNT = 4; + final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + registry.setCacheCapacity(CAPACITY_CNT); + + registry.obtain("foo:1"); + registry.obtain("foo:2"); + registry.obtain("foo:3"); + + //capacity 4->3 + registry.setCacheCapacity(CAPACITY_CNT - 1); + + registry.obtain("foo:4"); + + assertThat(getRegistryLocks(registry)).hasSize(3); + assertThat(getRegistryLocks(registry)).containsKeys(toKey("foo:2"), toKey("foo:3"), toKey("foo:4")); + + //capacity 3->4 + registry.setCacheCapacity(CAPACITY_CNT); + registry.obtain("foo:5"); + assertThat(getRegistryLocks(registry)).hasSize(4); + assertThat(getRegistryLocks(registry)).containsKeys(toKey("foo:3"), toKey("foo:4"), toKey("foo:5")); + registry.destroy(); + } + + @SuppressWarnings("unchecked") + private static Map getRegistryLocks(ZookeeperLockRegistry registry) { + return TestUtils.getPropertyValue(registry, "locks", Map.class); + } + + private static String toKey(String path) { + final String DEFAULT_ROOT = "/SpringIntegration-LockRegistry"; + return DEFAULT_ROOT + "/" + path; + } } diff --git a/src/reference/asciidoc/jdbc.adoc b/src/reference/asciidoc/jdbc.adoc index 2104626051..a27a093eba 100644 --- a/src/reference/asciidoc/jdbc.adoc +++ b/src/reference/asciidoc/jdbc.adoc @@ -1091,6 +1091,9 @@ So the time to live can be highly reduce and deployments can retake a lost lock NOTE: The lock renewal can be done only if the lock is held by the current thread. +String with version 5.5.6, the `JdbcLockRegistry` is support automatically clean up cache for JdbcLock in `JdbcLockRegistry.locks` via `JdbcLockRegistry.setCacheCapacity()`. +See its JavaDocs for more information. + [[jdbc-metadata-store]] === JDBC Metadata Store diff --git a/src/reference/asciidoc/redis.adoc b/src/reference/asciidoc/redis.adoc index 6b7bf80c4c..fb86df726b 100644 --- a/src/reference/asciidoc/redis.adoc +++ b/src/reference/asciidoc/redis.adoc @@ -884,5 +884,5 @@ You should set the expiry at a large enough value to prevent this condition, but Starting with version 5.0, the `RedisLockRegistry` implements `ExpirableLockRegistry`, which removes locks last acquired more than `age` ago and that are not currently locked. -String with version 5.5.6, the `RedisLockRegistry` is support automatically clean up cache for redisLocks in `RedisLockRegistry.locks` via `RedisLockRegistry.setCapacity()`. +String with version 5.5.6, the `RedisLockRegistry` is support automatically clean up cache for redisLocks in `RedisLockRegistry.locks` via `RedisLockRegistry.setCacheCapacity()`. See its JavaDocs for more information. \ No newline at end of file diff --git a/src/reference/asciidoc/zookeeper.adoc b/src/reference/asciidoc/zookeeper.adoc index 8441ad103e..d4129f9ff7 100644 --- a/src/reference/asciidoc/zookeeper.adoc +++ b/src/reference/asciidoc/zookeeper.adoc @@ -84,6 +84,9 @@ public interface KeyToPathStrategy { If the strategy returns `true` from `isBounded`, unused locks do not need to be harvested. For unbounded strategies (such as the default), you need to periodically invoke `expireUnusedOlderThan(long age)` to remove old unused locks from memory. +String with version 5.5.6, the `ZookeeperLockRegistry` is support automatically clean up cache for ZkLock in `ZookeeperLockRegistry.locks` via `ZookeeperLockRegistry.setCacheCapacity()`. +See its JavaDocs for more information. + [[zk-leadership]] === Zookeeper Leadership Event Handling