This is an automated email from the ASF dual-hosted git repository.
terrymanu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new dcd2221dded Replace copy-on-write sets in exclusive operator contexts
(#40340)
dcd2221dded is described below
commit dcd2221dded94320479e2021f47cb74cd0c08729
Author: Liang Zhang <[email protected]>
AuthorDate: Thu Oct 22 19:35:46 2026 +0800
Replace copy-on-write sets in exclusive operator contexts (#40340)
---
.../exclusive/ClusterExclusiveOperatorContext.java | 5 ++-
.../ClusterExclusiveOperatorContextTest.java | 40 +++++++++++++++++++
.../StandaloneExclusiveOperatorContext.java | 6 +--
.../StandaloneExclusiveOperatorContextTest.java | 45 ++++++++++++++++++++++
4 files changed, 91 insertions(+), 5 deletions(-)
diff --git
a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContext.java
b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContext.java
index 3078be5cfdd..a942f215037 100644
---
a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContext.java
+++
b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContext.java
@@ -28,7 +28,8 @@ import org.apache.shardingsphere.mode.retry.RetryExecutor;
import java.util.Collection;
import java.util.Optional;
-import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -41,7 +42,7 @@ public final class ClusterExclusiveOperatorContext implements
ExclusiveOperatorC
private final ClusterPersistRepository repository;
- private final Collection<String> exclusiveOperationKeys = new
CopyOnWriteArraySet<>();
+ private final Set<String> exclusiveOperationKeys =
ConcurrentHashMap.newKeySet();
@Override
public Optional<ExclusiveLockHandle> start(final String operationKey,
final long timeoutMillis) {
diff --git
a/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContextTest.java
b/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContextTest.java
index 69ba6171d96..35fe6520cfe 100644
---
a/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContextTest.java
+++
b/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/exclusive/ClusterExclusiveOperatorContextTest.java
@@ -28,12 +28,15 @@ import
org.apache.shardingsphere.mode.repository.cluster.listener.DataChangedEve
import org.apache.shardingsphere.mode.repository.cluster.lock.DistributedLock;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -72,6 +75,43 @@ class ClusterExclusiveOperatorContextTest {
}
}
+ @Test
+ void assertStartWithConcurrentSameKey() throws InterruptedException,
ExecutionException, TimeoutException {
+ SharedState sharedState = new SharedState();
+ ClusterExclusiveOperatorContext context = new
ClusterExclusiveOperatorContext(new
FreshLockClusterPersistRepository(sharedState, true, true));
+ CountDownLatch startGate = new CountDownLatch(1);
+ CountDownLatch attemptsFinished = new CountDownLatch(2);
+ ExecutorService executorService = Executors.newFixedThreadPool(2);
+ Collection<Future<Boolean>> actualResults = new ArrayList<>(2);
+ try {
+ for (int i = 0; i < 2; i++) {
+ actualResults.add(executorService.submit(() -> {
+ assertTrue(startGate.await(5L, TimeUnit.SECONDS));
+ Optional<ExclusiveLockHandle> actualLockHandle =
context.start("foo_operation", 0L);
+ try {
+ attemptsFinished.countDown();
+ assertTrue(attemptsFinished.await(5L,
TimeUnit.SECONDS));
+ return actualLockHandle.isPresent();
+ } finally {
+ actualLockHandle.ifPresent(ExclusiveLockHandle::close);
+ }
+ }));
+ }
+ startGate.countDown();
+ int actualAcquiredCount = 0;
+ for (Future<Boolean> each : actualResults) {
+ if (each.get(5L, TimeUnit.SECONDS)) {
+ actualAcquiredCount++;
+ }
+ }
+ assertThat(actualAcquiredCount, is(1));
+ assertThat(sharedState.tryLockCallCount.get(), is(1));
+ } finally {
+ startGate.countDown();
+ executorService.shutdownNow();
+ }
+ }
+
@Test
void assertStartWaitsUntilOperationKeyReleased() throws
InterruptedException, ExecutionException, TimeoutException {
SharedState sharedState = new SharedState();
diff --git
a/mode/type/standalone/core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContext.java
b/mode/type/standalone/core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContext.java
index c82770cc023..ad2af0802b7 100644
---
a/mode/type/standalone/core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContext.java
+++
b/mode/type/standalone/core/src/main/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContext.java
@@ -21,16 +21,16 @@ import
org.apache.shardingsphere.mode.exclusive.ExclusiveLockHandle;
import org.apache.shardingsphere.mode.exclusive.ExclusiveOperatorContext;
import org.apache.shardingsphere.mode.retry.RetryExecutor;
-import java.util.Collection;
import java.util.Optional;
-import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
/**
* Standalone lock context.
*/
public final class StandaloneExclusiveOperatorContext implements
ExclusiveOperatorContext {
- private final Collection<String> exclusiveOperationKeys = new
CopyOnWriteArraySet<>();
+ private final Set<String> exclusiveOperationKeys =
ConcurrentHashMap.newKeySet();
@Override
public Optional<ExclusiveLockHandle> start(final String operationKey,
final long timeoutMillis) {
diff --git
a/mode/type/standalone/core/src/test/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContextTest.java
b/mode/type/standalone/core/src/test/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContextTest.java
index 24b51aba163..3076fc8bb97 100644
---
a/mode/type/standalone/core/src/test/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContextTest.java
+++
b/mode/type/standalone/core/src/test/java/org/apache/shardingsphere/mode/manager/standalone/exclusive/StandaloneExclusiveOperatorContextTest.java
@@ -20,9 +20,19 @@ package
org.apache.shardingsphere.mode.manager.standalone.exclusive;
import org.apache.shardingsphere.mode.exclusive.ExclusiveLockHandle;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.not;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -43,6 +53,41 @@ class StandaloneExclusiveOperatorContextTest {
firstHandle.close();
}
+ @Test
+ void assertStartWithConcurrentSameKey() throws InterruptedException,
ExecutionException, TimeoutException {
+ StandaloneExclusiveOperatorContext context = new
StandaloneExclusiveOperatorContext();
+ CountDownLatch startGate = new CountDownLatch(1);
+ CountDownLatch attemptsFinished = new CountDownLatch(2);
+ ExecutorService executorService = Executors.newFixedThreadPool(2);
+ Collection<Future<Boolean>> actualResults = new ArrayList<>(2);
+ try {
+ for (int i = 0; i < 2; i++) {
+ actualResults.add(executorService.submit(() -> {
+ assertTrue(startGate.await(5L, TimeUnit.SECONDS));
+ Optional<ExclusiveLockHandle> actualLockHandle =
context.start("foo_operation", 0L);
+ try {
+ attemptsFinished.countDown();
+ assertTrue(attemptsFinished.await(5L,
TimeUnit.SECONDS));
+ return actualLockHandle.isPresent();
+ } finally {
+ actualLockHandle.ifPresent(ExclusiveLockHandle::close);
+ }
+ }));
+ }
+ startGate.countDown();
+ int actualAcquiredCount = 0;
+ for (Future<Boolean> each : actualResults) {
+ if (each.get(5L, TimeUnit.SECONDS)) {
+ actualAcquiredCount++;
+ }
+ }
+ assertThat(actualAcquiredCount, is(1));
+ } finally {
+ startGate.countDown();
+ executorService.shutdownNow();
+ }
+ }
+
@Test
void assertCloseReleasesOperationKey() {
StandaloneExclusiveOperatorContext context = new
StandaloneExclusiveOperatorContext();
