From d40ca3536c946a413eb48f467e1e79b08f533e0d Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Tue, 8 Nov 2022 18:23:33 +0800 Subject: [PATCH 01/27] fix bug caused by optimistic locking` --- .../ConcurrentLongLongPairHashMap.java | 20 +++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java index c0ccad9b73d5b..8c073cd68db12 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java @@ -26,6 +26,8 @@ import java.util.Map; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.locks.StampedLock; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Concurrent hash map where both keys and values are composed of pairs of longs. @@ -38,6 +40,7 @@ *

Keys MUST be >= 0. */ public class ConcurrentLongLongPairHashMap { + private static final Logger logger = LoggerFactory.getLogger(ConcurrentLongLongPairHashMap.class); private static final long EmptyKey = -1L; private static final long DeletedKey = -2L; @@ -327,10 +330,19 @@ LongPair get(long key1, long key2, int keyHash) { try { while (true) { // First try optimistic locking - long storedKey1 = table[bucket]; - long storedKey2 = table[bucket + 1]; - long storedValue1 = table[bucket + 2]; - long storedValue2 = table[bucket + 3]; + long storedKey1 = 0; + long storedKey2 = 0; + long storedValue1 = 0; + long storedValue2 = 0; + try { + storedKey1 = table[bucket]; + storedKey2 = table[bucket + 1]; + storedValue1 = table[bucket + 2]; + storedValue2 = table[bucket + 3]; + } catch (ArrayIndexOutOfBoundsException e) { + // read a dirty capacity, cause ArrayIndexOutOfBoundsException + logger.error("Read a dirty capacity, fallback to Pessimistic Read. capacity:{}, bucket:{}", capacity, bucket); + } if (!acquiredLock && validate(stamp)) { // The values we have read are consistent From 5c7340307c9ccde4993ddfe6dd50b214e24ee11d Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 9 Nov 2022 09:49:38 +0800 Subject: [PATCH 02/27] change log level --- .../util/collections/ConcurrentLongLongPairHashMap.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java index 8c073cd68db12..5abfd0d0ea57c 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java @@ -341,7 +341,12 @@ LongPair get(long key1, long key2, int keyHash) { storedValue2 = table[bucket + 3]; } catch (ArrayIndexOutOfBoundsException e) { // read a dirty capacity, cause ArrayIndexOutOfBoundsException - logger.error("Read a dirty capacity, fallback to Pessimistic Read. capacity:{}, bucket:{}", capacity, bucket); + if (logger.isDebugEnabled()) { + logger.debug( + "Read a dirty capacity, fallback to Pessimistic Read. capacity:{}, bucket:{}, " + + "cause:{}", + capacity, bucket, e.getMessage()); + } } if (!acquiredLock && validate(stamp)) { From 6d38cf71564d126aa0f271fa002bb7fbbde7e03c Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 11 Nov 2022 11:36:45 +0800 Subject: [PATCH 03/27] throw ArrayIndexOutOfBoundsException if not caused by concurrent execution. --- .../common/util/collections/ConcurrentLongLongPairHashMap.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java index 5abfd0d0ea57c..5bc4b8663b644 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java @@ -341,6 +341,9 @@ LongPair get(long key1, long key2, int keyHash) { storedValue2 = table[bucket + 3]; } catch (ArrayIndexOutOfBoundsException e) { // read a dirty capacity, cause ArrayIndexOutOfBoundsException + if (validate(stamp)) { + throw e; + } if (logger.isDebugEnabled()) { logger.debug( "Read a dirty capacity, fallback to Pessimistic Read. capacity:{}, bucket:{}, " From 25f23e1c10144303764a716ef2cb805de3fb6b82 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 11 Nov 2022 11:40:33 +0800 Subject: [PATCH 04/27] change implementation --- .../ConcurrentLongLongPairHashMap.java | 33 ++++++------------- 1 file changed, 10 insertions(+), 23 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java index 5bc4b8663b644..dcb821b65b088 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java @@ -325,32 +325,18 @@ private static final class Section extends StampedLock { LongPair get(long key1, long key2, int keyHash) { long stamp = tryOptimisticRead(); boolean acquiredLock = false; - int bucket = signSafeMod(keyHash, capacity); + // add local variable here, so OutOfBound won't happen + long[] table = this.table; + // calculate table.length / 4 as capacity to avoid rehash changing capacity + int bucket = signSafeMod(keyHash, table.length / 4); try { while (true) { // First try optimistic locking - long storedKey1 = 0; - long storedKey2 = 0; - long storedValue1 = 0; - long storedValue2 = 0; - try { - storedKey1 = table[bucket]; - storedKey2 = table[bucket + 1]; - storedValue1 = table[bucket + 2]; - storedValue2 = table[bucket + 3]; - } catch (ArrayIndexOutOfBoundsException e) { - // read a dirty capacity, cause ArrayIndexOutOfBoundsException - if (validate(stamp)) { - throw e; - } - if (logger.isDebugEnabled()) { - logger.debug( - "Read a dirty capacity, fallback to Pessimistic Read. capacity:{}, bucket:{}, " - + "cause:{}", - capacity, bucket, e.getMessage()); - } - } + long storedKey1 = table[bucket]; + long storedKey2 = table[bucket + 1]; + long storedValue1 = table[bucket + 2]; + long storedValue2 = table[bucket + 3]; if (!acquiredLock && validate(stamp)) { // The values we have read are consistent @@ -365,7 +351,8 @@ LongPair get(long key1, long key2, int keyHash) { if (!acquiredLock) { stamp = readLock(); acquiredLock = true; - + // update local variable + table = this.table; bucket = signSafeMod(keyHash, capacity); storedKey1 = table[bucket]; storedKey2 = table[bucket + 1]; From 6a166b7459aec8d7e692ced6fb6a9123d0e7d863 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 14 Dec 2022 10:47:04 +0800 Subject: [PATCH 05/27] remove logger --- .../common/util/collections/ConcurrentLongLongPairHashMap.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java index dcb821b65b088..0329386917be9 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java @@ -40,7 +40,6 @@ *

Keys MUST be >= 0. */ public class ConcurrentLongLongPairHashMap { - private static final Logger logger = LoggerFactory.getLogger(ConcurrentLongLongPairHashMap.class); private static final long EmptyKey = -1L; private static final long DeletedKey = -2L; From daf6eaff1bd89b7fdb786bec3255a4ef8e127561 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 14 Dec 2022 10:47:49 +0800 Subject: [PATCH 06/27] remove dependency --- .../common/util/collections/ConcurrentLongLongPairHashMap.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java index 0329386917be9..33b5be27ca3a8 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java @@ -26,8 +26,6 @@ import java.util.Map; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.locks.StampedLock; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Concurrent hash map where both keys and values are composed of pairs of longs. From 4641ac570834cd3a60c747ed55344e625c45166f Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 14 Dec 2022 11:25:31 +0800 Subject: [PATCH 07/27] add test code. --- .../ConcurrentLongLongPairHashMap.java | 2 ++ .../ConcurrentLongLongPairHashMapTest.java | 33 +++++++++++++++++++ 2 files changed, 35 insertions(+) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java index 33b5be27ca3a8..ab10be84639fe 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java @@ -327,6 +327,8 @@ LongPair get(long key1, long key2, int keyHash) { // calculate table.length / 4 as capacity to avoid rehash changing capacity int bucket = signSafeMod(keyHash, table.length / 4); + System.out.println("key1:" + key1 + ",key2:" + key2 + ",bucket:" + bucket); + try { while (true) { // First try optimistic locking diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java index 8e74d285ffb9b..0aafd9319f9a1 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java @@ -173,6 +173,39 @@ public void testExpandAndShrink() { assertEquals(map.capacity(), 8); } + @Test + public void testExpandAndShrinkAndGet() { + ConcurrentLongLongPairHashMap map = ConcurrentLongLongPairHashMap.newBuilder() + .expectedItems(2) + .concurrencyLevel(1) + .autoShrink(true) + .mapIdleFactor(0.25f) + .build(); + ExecutorService service = Executors.newCachedThreadPool(); + + // expand hashmap + assertTrue(map.put(1, 1, 11, 11)); + assertTrue(map.put(2, 2, 22, 22)); + assertTrue(map.put(3, 3, 33, 33)); + assertEquals(map.capacity(), 8); + + service.submit(() -> { + while (true) { + map.get(1, 1); + } + }); + + service.submit(() -> { + // shrink hashmap + assertTrue(map.remove(1, 1, 11, 11)); + assertTrue(map.remove(2, 2, 22, 22)); + assertEquals(map.capacity(), 4); + }); + + // shut down pool + service.shutdown(); + } + @Test public void testExpandShrinkAndClear() { ConcurrentLongLongPairHashMap map = ConcurrentLongLongPairHashMap.newBuilder() From 1834c2f42c7dbe7b8ab4d4796e16f960696a67d5 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 14 Dec 2022 11:26:01 +0800 Subject: [PATCH 08/27] fix --- .../common/util/collections/ConcurrentLongLongPairHashMap.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java index ab10be84639fe..33b5be27ca3a8 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java @@ -327,8 +327,6 @@ LongPair get(long key1, long key2, int keyHash) { // calculate table.length / 4 as capacity to avoid rehash changing capacity int bucket = signSafeMod(keyHash, table.length / 4); - System.out.println("key1:" + key1 + ",key2:" + key2 + ",bucket:" + bucket); - try { while (true) { // First try optimistic locking From e896fb5158ad027d5771be7ca52bcee59418caa8 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 15 Dec 2022 15:22:47 +0800 Subject: [PATCH 09/27] stress test. --- .../ConcurrentLongLongPairHashMapTest.java | 58 +++++++++++++------ 1 file changed, 41 insertions(+), 17 deletions(-) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java index 0aafd9319f9a1..ce5f9bf8559c5 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java @@ -31,6 +31,7 @@ import java.util.List; import java.util.Map; import java.util.Random; +import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -174,36 +175,59 @@ public void testExpandAndShrink() { } @Test - public void testExpandAndShrinkAndGet() { + public void testExpandAndShrinkAndGet() throws Throwable { ConcurrentLongLongPairHashMap map = ConcurrentLongLongPairHashMap.newBuilder() .expectedItems(2) .concurrencyLevel(1) .autoShrink(true) .mapIdleFactor(0.25f) .build(); - ExecutorService service = Executors.newCachedThreadPool(); + assertEquals(map.capacity(), 4); - // expand hashmap - assertTrue(map.put(1, 1, 11, 11)); - assertTrue(map.put(2, 2, 22, 22)); - assertTrue(map.put(3, 3, 33, 33)); - assertEquals(map.capacity(), 8); + ExecutorService executor = Executors.newCachedThreadPool(); + final int readThreads = 16; + final int writeThreads = 1; + final int n = 1_000_000; + CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads); + Future future = null; + + for (int i = 0; i < readThreads; i++) { + executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + while (true) { + map.get(1, 1); + } + }); + } - service.submit(() -> { - while (true) { - map.get(1, 1); + assertTrue(map.put(1, 1, 11, 11)); + future = executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); } - }); - service.submit(() -> { - // shrink hashmap - assertTrue(map.remove(1, 1, 11, 11)); - assertTrue(map.remove(2, 2, 22, 22)); - assertEquals(map.capacity(), 4); + for (int i = 0; i < n; i++) { + // expand hashmap + assertTrue(map.put(2, 2, 22, 22)); + assertTrue(map.put(3, 3, 33, 33)); + assertEquals(map.capacity(), 8); + + // shrink hashmap + assertTrue(map.remove(2, 2, 22, 22)); + assertTrue(map.remove(3, 3, 33, 33)); + assertEquals(map.capacity(), 4); + } }); + future.get(); // shut down pool - service.shutdown(); + executor.shutdown(); } @Test From 4b5c43ef205adb349df8fce70bb95c099fbe8c0b Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 16 Dec 2022 14:26:36 +0800 Subject: [PATCH 10/27] fix ConcurrentOpenHashMap --- .../common/util/collections/ConcurrentOpenHashMap.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java index ea2e01768ac7e..42806f10fbb9a 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java @@ -332,7 +332,11 @@ private static final class Section extends StampedLock { V get(K key, int keyHash) { long stamp = tryOptimisticRead(); boolean acquiredLock = false; - int bucket = signSafeMod(keyHash, capacity); + + // add local variable here, so OutOfBound won't happen + Object[] table = this.table; + // calculate table.length / 2 as capacity to avoid rehash changing capacity + int bucket = signSafeMod(keyHash, table.length / 2); try { while (true) { @@ -354,6 +358,8 @@ V get(K key, int keyHash) { stamp = readLock(); acquiredLock = true; + // update local variable + table = this.table; bucket = signSafeMod(keyHash, capacity); storedKey = (K) table[bucket]; storedValue = (V) table[bucket + 1]; From fa5eaa8e533e2a1aeaffd7b9811de8f3f1f0f8ed Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 16 Dec 2022 16:17:26 +0800 Subject: [PATCH 11/27] fix ConcurrentLongPairSet --- .../common/util/collections/ConcurrentLongPairSet.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java index 2a1090503857c..d4f585bbb491e 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java @@ -331,7 +331,11 @@ private static final class Section extends StampedLock { boolean contains(long item1, long item2, int hash) { long stamp = tryOptimisticRead(); boolean acquiredLock = false; - int bucket = signSafeMod(hash, capacity); + + // add local variable here, so OutOfBound won't happen + long[] table = this.table; + // calculate table.length / 2 as capacity to avoid rehash changing capacity + int bucket = signSafeMod(hash, table.length / 2); try { while (true) { @@ -353,6 +357,8 @@ boolean contains(long item1, long item2, int hash) { stamp = readLock(); acquiredLock = true; + // update local variable + table = this.table; bucket = signSafeMod(hash, capacity); storedItem1 = table[bucket]; storedItem2 = table[bucket + 1]; From 15ead23c77c08c498be534fc688b6a2e6feea7bd Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 16 Dec 2022 16:50:11 +0800 Subject: [PATCH 12/27] fix ConcurrentOpenHashSet --- .../collections/ConcurrentOpenHashSet.java | 22 ++++++++----------- 1 file changed, 9 insertions(+), 13 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSet.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSet.java index cc8bc07b43095..5d3cfbced6427 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSet.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSet.java @@ -294,16 +294,16 @@ private static final class Section extends StampedLock { } boolean contains(V value, int keyHash) { - int bucket = keyHash; - long stamp = tryOptimisticRead(); boolean acquiredLock = false; + // add local variable here, so OutOfBound won't happen + V[] values = this.values; + // calculate table.length as capacity to avoid rehash changing capacity + int bucket = signSafeMod(keyHash, values.length); + try { while (true) { - int capacity = this.capacity; - bucket = signSafeMod(bucket, capacity); - // First try optimistic locking V storedValue = values[bucket]; @@ -321,15 +321,12 @@ boolean contains(V value, int keyHash) { stamp = readLock(); acquiredLock = true; + // update local variable + values = this.values; + bucket = signSafeMod(keyHash, capacity); storedValue = values[bucket]; } - if (capacity != this.capacity) { - // There has been a rehashing. We need to restart the search - bucket = keyHash; - continue; - } - if (value.equals(storedValue)) { return true; } else if (storedValue == EmptyValue) { @@ -337,8 +334,7 @@ boolean contains(V value, int keyHash) { return false; } } - - ++bucket; + bucket = (bucket + 1) & (values.length - 1); } } finally { if (acquiredLock) { From 4427cd901fbdbd67832a46b1bfff990fe2ac7725 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 16 Dec 2022 17:04:56 +0800 Subject: [PATCH 13/27] fix ConcurrentLongHashMap --- .../collections/ConcurrentLongHashMap.java | 25 +++++++++---------- 1 file changed, 12 insertions(+), 13 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java index 31b4cb7cbf152..aabb7d62a738b 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java @@ -306,16 +306,17 @@ private static final class Section extends StampedLock { } V get(long key, int keyHash) { - int bucket = keyHash; - long stamp = tryOptimisticRead(); boolean acquiredLock = false; + // add local variable here, so OutOfBound won't happen + long[] keys = this.keys; + V[] values = this.values; + // calculate table.length as capacity to avoid rehash changing capacity + int bucket = signSafeMod(keyHash, values.length); + try { while (true) { - int capacity = this.capacity; - bucket = signSafeMod(bucket, capacity); - // First try optimistic locking long storedKey = keys[bucket]; V storedValue = values[bucket]; @@ -333,16 +334,15 @@ V get(long key, int keyHash) { if (!acquiredLock) { stamp = readLock(); acquiredLock = true; + + // update local variable + keys = this.keys; + values = this.values; + bucket = signSafeMod(keyHash, capacity); storedKey = keys[bucket]; storedValue = values[bucket]; } - if (capacity != this.capacity) { - // There has been a rehashing. We need to restart the search - bucket = keyHash; - continue; - } - if (storedKey == key) { return storedValue != DeletedValue ? storedValue : null; } else if (storedValue == EmptyValue) { @@ -350,8 +350,7 @@ V get(long key, int keyHash) { return null; } } - - ++bucket; + bucket = (bucket + 1) & (values.length - 1); } } finally { if (acquiredLock) { From 91a86537eee454db974104c40de32108b6c9dac0 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 16 Dec 2022 18:08:07 +0800 Subject: [PATCH 14/27] add test code. --- .../ConcurrentLongHashMapTest.java | 56 ++++++++++++++++++ .../ConcurrentLongLongPairHashMapTest.java | 2 +- .../ConcurrentLongPairSetTest.java | 57 +++++++++++++++++++ .../ConcurrentOpenHashMapTest.java | 57 +++++++++++++++++++ .../ConcurrentOpenHashSetTest.java | 57 +++++++++++++++++++ 5 files changed, 228 insertions(+), 1 deletion(-) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java index a317fa63c0986..3311565d7a011 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java @@ -213,6 +213,62 @@ public void testExpandShrinkAndClear() { assertTrue(map.capacity() == initCapacity); } + @Test + public void testConcurrentExpandAndShrinkAndGet() throws Throwable { + ConcurrentLongHashMap map = ConcurrentLongHashMap.newBuilder() + .expectedItems(2) + .concurrencyLevel(1) + .autoShrink(true) + .mapIdleFactor(0.25f) + .build(); + assertEquals(map.capacity(), 4); + + ExecutorService executor = Executors.newCachedThreadPool(); + final int readThreads = 16; + final int writeThreads = 1; + final int n = 1_000_000; + CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads); + Future future = null; + + for (int i = 0; i < readThreads; i++) { + executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + while (true) { + map.get(1); + } + }); + } + + assertNull(map.put(1,"v1")); + future = executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + + for (int i = 0; i < n; i++) { + // expand hashmap + assertNull(map.put(2, "v2")); + assertNull(map.put(3, "v3")); + assertEquals(map.capacity(), 8); + + // shrink hashmap + assertTrue(map.remove(2, "v2")); + assertTrue(map.remove(3, "v3")); + assertEquals(map.capacity(), 4); + } + }); + + future.get(); + // shut down pool + executor.shutdown(); + } + @Test public void testRemove() { ConcurrentLongHashMap map = ConcurrentLongHashMap.newBuilder() diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java index ce5f9bf8559c5..d04f772e821f9 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java @@ -175,7 +175,7 @@ public void testExpandAndShrink() { } @Test - public void testExpandAndShrinkAndGet() throws Throwable { + public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ConcurrentLongLongPairHashMap map = ConcurrentLongLongPairHashMap.newBuilder() .expectedItems(2) .concurrencyLevel(1) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java index 7e947ae6e6aa3..a95360e3951d0 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java @@ -30,6 +30,7 @@ import java.util.List; import java.util.Random; import java.util.Set; +import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -211,6 +212,62 @@ public void testExpandShrinkAndClear() { assertTrue(map.capacity() == initCapacity); } + @Test + public void testConcurrentExpandAndShrinkAndGet() throws Throwable { + ConcurrentLongPairSet set = ConcurrentLongPairSet.newBuilder() + .expectedItems(2) + .concurrencyLevel(1) + .autoShrink(true) + .mapIdleFactor(0.25f) + .build(); + assertEquals(set.capacity(), 4); + + ExecutorService executor = Executors.newCachedThreadPool(); + final int readThreads = 16; + final int writeThreads = 1; + final int n = 1_000_000; + CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads); + Future future = null; + + for (int i = 0; i < readThreads; i++) { + executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + while (true) { + set.contains(1, 1); + } + }); + } + + assertTrue(set.add(1, 1)); + future = executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + + for (int i = 0; i < n; i++) { + // expand hashmap + assertTrue(set.add(2, 2)); + assertTrue(set.add(3, 3)); + assertEquals(set.capacity(), 8); + + // shrink hashmap + assertTrue(set.remove(2, 2)); + assertTrue(set.remove(3, 3)); + assertEquals(set.capacity(), 4); + } + }); + + future.get(); + // shut down pool + executor.shutdown(); + } + @Test public void testRemove() { diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java index 198a3f4c5c38b..076192f968cdb 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java @@ -32,6 +32,7 @@ import java.util.List; import java.util.Random; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -215,6 +216,62 @@ public void testExpandShrinkAndClear() { assertTrue(map.capacity() == initCapacity); } + @Test + public void testConcurrentExpandAndShrinkAndGet() throws Throwable { + ConcurrentOpenHashMap map = ConcurrentOpenHashMap.newBuilder() + .expectedItems(2) + .concurrencyLevel(1) + .autoShrink(true) + .mapIdleFactor(0.25f) + .build(); + assertEquals(map.capacity(), 4); + + ExecutorService executor = Executors.newCachedThreadPool(); + final int readThreads = 16; + final int writeThreads = 1; + final int n = 1_000_000; + CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads); + Future future = null; + + for (int i = 0; i < readThreads; i++) { + executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + while (true) { + map.get("k1"); + } + }); + } + + assertNull(map.put("k1","v1")); + future = executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + + for (int i = 0; i < n; i++) { + // expand hashmap + assertNull(map.put("k2", "v2")); + assertNull(map.put("k3", "v3")); + assertEquals(map.capacity(), 8); + + // shrink hashmap + assertTrue(map.remove("k2", "v2")); + assertTrue(map.remove("k3", "v3")); + assertEquals(map.capacity(), 4); + } + }); + + future.get(); + // shut down pool + executor.shutdown(); + } + @Test public void testRemove() { ConcurrentOpenHashMap map = diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java index 27c18abb8b347..ec09612ccc031 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java @@ -28,6 +28,7 @@ import java.util.Collections; import java.util.List; import java.util.Random; +import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -186,6 +187,62 @@ public void testExpandShrinkAndClear() { assertTrue(map.capacity() == initCapacity); } + @Test + public void testConcurrentExpandAndShrinkAndGet() throws Throwable { + ConcurrentOpenHashSet set = ConcurrentOpenHashSet.newBuilder() + .expectedItems(2) + .concurrencyLevel(1) + .autoShrink(true) + .mapIdleFactor(0.25f) + .build(); + assertEquals(set.capacity(), 4); + + ExecutorService executor = Executors.newCachedThreadPool(); + final int readThreads = 16; + final int writeThreads = 1; + final int n = 1_000_000; + CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads); + Future future = null; + + for (int i = 0; i < readThreads; i++) { + executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + while (true) { + set.contains("k1"); + } + }); + } + + assertTrue(set.add("k1")); + future = executor.submit(() -> { + try { + barrier.await(); + } catch (Exception e) { + throw new RuntimeException(e); + } + + for (int i = 0; i < n; i++) { + // expand hashmap + assertTrue(set.add("k2")); + assertTrue(set.add("k3")); + assertEquals(set.capacity(), 8); + + // shrink hashmap + assertTrue(set.remove("k2")); + assertTrue(set.remove("k3")); + assertEquals(set.capacity(), 4); + } + }); + + future.get(); + // shut down pool + executor.shutdown(); + } + @Test public void testRemove() { ConcurrentOpenHashSet set = From c693168a60554b459bc863c355a193dec4e0b6da Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 31 Aug 2023 17:00:01 +0800 Subject: [PATCH 15/27] update test code to reproduce bug. --- .../ConcurrentLongLongPairHashMapTest.java | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java index d04f772e821f9..0de3fdb5c84bf 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java @@ -35,6 +35,8 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicReference; + import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap.LongPair; import org.testng.annotations.Test; @@ -187,9 +189,10 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ExecutorService executor = Executors.newCachedThreadPool(); final int readThreads = 16; final int writeThreads = 1; - final int n = 1_000_000; + final int n = 1_000; CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads); Future future = null; + AtomicReference ex = new AtomicReference<>(); for (int i = 0; i < readThreads; i++) { executor.submit(() -> { @@ -199,7 +202,11 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { throw new RuntimeException(e); } while (true) { - map.get(1, 1); + try { + map.get(1, 1); + } catch (Exception e) { + ex.set(e); + } } }); } @@ -226,6 +233,7 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { }); future.get(); + assertTrue(ex.get() == null); // shut down pool executor.shutdown(); } From 0ab1800a3024ed1325d822c4dc6a3267927eca90 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Thu, 31 Aug 2023 17:27:34 +0800 Subject: [PATCH 16/27] update test code. --- .../util/collections/ConcurrentLongHashMapTest.java | 9 +++++++-- .../util/collections/ConcurrentLongPairSetTest.java | 11 +++++++++-- .../util/collections/ConcurrentOpenHashMapTest.java | 11 +++++++++-- .../util/collections/ConcurrentOpenHashSetTest.java | 11 +++++++++-- 4 files changed, 34 insertions(+), 8 deletions(-) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java index 3311565d7a011..b64714337ec3d 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java @@ -38,6 +38,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.LongFunction; import lombok.Cleanup; @@ -226,9 +227,10 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ExecutorService executor = Executors.newCachedThreadPool(); final int readThreads = 16; final int writeThreads = 1; - final int n = 1_000_000; + final int n = 1_000; CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads); Future future = null; + AtomicReference ex = new AtomicReference<>(); for (int i = 0; i < readThreads; i++) { executor.submit(() -> { @@ -237,8 +239,10 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { } catch (Exception e) { throw new RuntimeException(e); } - while (true) { + try { map.get(1); + } catch (Exception e) { + ex.set(e); } }); } @@ -265,6 +269,7 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { }); future.get(); + assertTrue(ex.get() == null); // shut down pool executor.shutdown(); } diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java index a95360e3951d0..bce2b8993835f 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java @@ -34,6 +34,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicReference; import lombok.Cleanup; import org.apache.pulsar.common.util.collections.ConcurrentLongPairSet.LongPair; @@ -225,9 +226,10 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ExecutorService executor = Executors.newCachedThreadPool(); final int readThreads = 16; final int writeThreads = 1; - final int n = 1_000_000; + final int n = 1_000; CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads); Future future = null; + AtomicReference ex = new AtomicReference<>(); for (int i = 0; i < readThreads; i++) { executor.submit(() -> { @@ -237,7 +239,11 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { throw new RuntimeException(e); } while (true) { - set.contains(1, 1); + try { + set.contains(1, 1); + } catch (Exception e) { + ex.set(e); + } } }); } @@ -264,6 +270,7 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { }); future.get(); + assertTrue(ex.get() == null); // shut down pool executor.shutdown(); } diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java index 076192f968cdb..c919469afec9d 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java @@ -38,6 +38,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import lombok.Cleanup; @@ -229,9 +230,10 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ExecutorService executor = Executors.newCachedThreadPool(); final int readThreads = 16; final int writeThreads = 1; - final int n = 1_000_000; + final int n = 1_000; CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads); Future future = null; + AtomicReference ex = new AtomicReference<>(); for (int i = 0; i < readThreads; i++) { executor.submit(() -> { @@ -241,7 +243,11 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { throw new RuntimeException(e); } while (true) { - map.get("k1"); + try { + map.get("k1"); + } catch (Exception e) { + ex.set(e); + } } }); } @@ -268,6 +274,7 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { }); future.get(); + assertTrue(ex.get() == null); // shut down pool executor.shutdown(); } diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java index ec09612ccc031..22ca7ad2776cc 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java @@ -32,6 +32,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicReference; import lombok.Cleanup; import org.testng.annotations.Test; @@ -200,9 +201,10 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ExecutorService executor = Executors.newCachedThreadPool(); final int readThreads = 16; final int writeThreads = 1; - final int n = 1_000_000; + final int n = 1_000; CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads); Future future = null; + AtomicReference ex = new AtomicReference<>(); for (int i = 0; i < readThreads; i++) { executor.submit(() -> { @@ -212,7 +214,11 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { throw new RuntimeException(e); } while (true) { - set.contains("k1"); + try { + set.contains("k1"); + } catch (Exception e) { + ex.set(e); + } } }); } @@ -239,6 +245,7 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { }); future.get(); + assertTrue(ex.get() == null); // shut down pool executor.shutdown(); } From 072079b890a01892d6ad45208f4fe4f98c9a8b57 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 1 Sep 2023 11:12:42 +0800 Subject: [PATCH 17/27] fix. --- .../pulsar/common/util/collections/ConcurrentLongHashMap.java | 2 +- .../common/util/collections/ConcurrentLongLongPairHashMap.java | 2 +- .../pulsar/common/util/collections/ConcurrentLongPairSet.java | 2 +- .../pulsar/common/util/collections/ConcurrentOpenHashMap.java | 2 +- .../pulsar/common/util/collections/ConcurrentOpenHashSet.java | 2 +- 5 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java index aabb7d62a738b..f5b47e7f1ab7a 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMap.java @@ -338,7 +338,7 @@ V get(long key, int keyHash) { // update local variable keys = this.keys; values = this.values; - bucket = signSafeMod(keyHash, capacity); + bucket = signSafeMod(keyHash, values.length); storedKey = keys[bucket]; storedValue = values[bucket]; } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java index 33b5be27ca3a8..c3d985d384cce 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java @@ -350,7 +350,7 @@ LongPair get(long key1, long key2, int keyHash) { acquiredLock = true; // update local variable table = this.table; - bucket = signSafeMod(keyHash, capacity); + bucket = signSafeMod(keyHash, table.length / 4); storedKey1 = table[bucket]; storedKey2 = table[bucket + 1]; storedValue1 = table[bucket + 2]; diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java index d4f585bbb491e..fabdb5087af4a 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java @@ -359,7 +359,7 @@ boolean contains(long item1, long item2, int hash) { // update local variable table = this.table; - bucket = signSafeMod(hash, capacity); + bucket = signSafeMod(hash, table.length / 2); storedItem1 = table[bucket]; storedItem2 = table[bucket + 1]; } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java index 42806f10fbb9a..6270f94b2bb7f 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java @@ -360,7 +360,7 @@ V get(K key, int keyHash) { // update local variable table = this.table; - bucket = signSafeMod(keyHash, capacity); + bucket = signSafeMod(keyHash, table.length / 2); storedKey = (K) table[bucket]; storedValue = (V) table[bucket + 1]; } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSet.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSet.java index 5d3cfbced6427..162cbed5a5ddc 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSet.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSet.java @@ -323,7 +323,7 @@ boolean contains(V value, int keyHash) { // update local variable values = this.values; - bucket = signSafeMod(keyHash, capacity); + bucket = signSafeMod(keyHash, values.length); storedValue = values[bucket]; } From 78b30e9d4b4e21630daaa8876796256a0664228b Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 1 Sep 2023 11:29:34 +0800 Subject: [PATCH 18/27] update test code. --- .../common/util/collections/ConcurrentOpenHashMapTest.java | 2 +- .../common/util/collections/ConcurrentOpenHashSetTest.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java index c919469afec9d..410d490b98faa 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java @@ -244,7 +244,7 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { } while (true) { try { - map.get("k1"); + map.get("k2"); } catch (Exception e) { ex.set(e); } diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java index 22ca7ad2776cc..6a40095ab0647 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java @@ -215,7 +215,7 @@ public void testConcurrentExpandAndShrinkAndGet() throws Throwable { } while (true) { try { - set.contains("k1"); + set.contains("k2"); } catch (Exception e) { ex.set(e); } From 9d6050f915f61a7e36abdcd0ec77cacf833c92d3 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 1 Sep 2023 11:55:09 +0800 Subject: [PATCH 19/27] extract magic number ITEM_SIZE --- .../ConcurrentLongLongPairHashMap.java | 31 +++++++++-------- .../collections/ConcurrentLongPairSet.java | 33 ++++++++++--------- .../collections/ConcurrentOpenHashMap.java | 31 +++++++++-------- 3 files changed, 52 insertions(+), 43 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java index c3d985d384cce..43279ffbd18c3 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java @@ -284,6 +284,9 @@ public Map asMap() { // A section is a portion of the hash map that is covered by a single @SuppressWarnings("serial") private static final class Section extends StampedLock { + // Each item take up 4 continuous array space. + private static final int ITEM_SIZE = 4; + // Keys and values are stored interleaved in the table array private volatile long[] table; @@ -306,7 +309,7 @@ private static final class Section extends StampedLock { float expandFactor, float shrinkFactor) { this.capacity = alignToPowerOfTwo(capacity); this.initCapacity = this.capacity; - this.table = new long[4 * this.capacity]; + this.table = new long[ITEM_SIZE * this.capacity]; this.size = 0; this.usedBuckets = 0; this.autoShrink = autoShrink; @@ -325,7 +328,7 @@ LongPair get(long key1, long key2, int keyHash) { // add local variable here, so OutOfBound won't happen long[] table = this.table; // calculate table.length / 4 as capacity to avoid rehash changing capacity - int bucket = signSafeMod(keyHash, table.length / 4); + int bucket = signSafeMod(keyHash, table.length / ITEM_SIZE); try { while (true) { @@ -350,7 +353,7 @@ LongPair get(long key1, long key2, int keyHash) { acquiredLock = true; // update local variable table = this.table; - bucket = signSafeMod(keyHash, table.length / 4); + bucket = signSafeMod(keyHash, table.length / ITEM_SIZE); storedKey1 = table[bucket]; storedKey2 = table[bucket + 1]; storedValue1 = table[bucket + 2]; @@ -365,7 +368,7 @@ LongPair get(long key1, long key2, int keyHash) { } } - bucket = (bucket + 4) & (table.length - 1); + bucket = (bucket + ITEM_SIZE) & (table.length - 1); } } finally { if (acquiredLock) { @@ -417,7 +420,7 @@ boolean put(long key1, long key2, long value1, long value2, int keyHash, boolean } } - bucket = (bucket + 4) & (table.length - 1); + bucket = (bucket + ITEM_SIZE) & (table.length - 1); } } finally { if (usedBuckets > resizeThresholdUp) { @@ -458,7 +461,7 @@ private boolean remove(long key1, long key2, long value1, long value2, int keyHa return false; } - bucket = (bucket + 4) & (table.length - 1); + bucket = (bucket + ITEM_SIZE) & (table.length - 1); } } finally { @@ -484,7 +487,7 @@ private boolean remove(long key1, long key2, long value1, long value2, int keyHa } private void cleanBucket(int bucket) { - int nextInArray = (bucket + 4) & (table.length - 1); + int nextInArray = (bucket + ITEM_SIZE) & (table.length - 1); if (table[nextInArray] == EmptyKey) { table[bucket] = EmptyKey; table[bucket + 1] = EmptyKey; @@ -493,7 +496,7 @@ private void cleanBucket(int bucket) { --usedBuckets; // Cleanup all the buckets that were in `DeletedKey` state, so that we can reduce unnecessary expansions - bucket = (bucket - 4) & (table.length - 1); + bucket = (bucket - ITEM_SIZE) & (table.length - 1); while (table[bucket] == DeletedKey) { table[bucket] = EmptyKey; table[bucket + 1] = EmptyKey; @@ -501,7 +504,7 @@ private void cleanBucket(int bucket) { table[bucket + 3] = ValueNotFound; --usedBuckets; - bucket = (bucket - 4) & (table.length - 1); + bucket = (bucket - ITEM_SIZE) & (table.length - 1); } } else { table[bucket] = DeletedKey; @@ -544,7 +547,7 @@ public void forEach(BiConsumerLongPair processor) { } // Go through all the buckets for this section - for (int bucket = 0; bucket < table.length; bucket += 4) { + for (int bucket = 0; bucket < table.length; bucket += ITEM_SIZE) { long storedKey1 = table[bucket]; long storedKey2 = table[bucket + 1]; long storedValue1 = table[bucket + 2]; @@ -573,11 +576,11 @@ public void forEach(BiConsumerLongPair processor) { } private void rehash(int newCapacity) { - long[] newTable = new long[4 * newCapacity]; + long[] newTable = new long[ITEM_SIZE * newCapacity]; Arrays.fill(newTable, EmptyKey); // Re-hash table - for (int i = 0; i < table.length; i += 4) { + for (int i = 0; i < table.length; i += ITEM_SIZE) { long storedKey1 = table[i]; long storedKey2 = table[i + 1]; long storedValue1 = table[i + 2]; @@ -597,7 +600,7 @@ private void rehash(int newCapacity) { } private void shrinkToInitCapacity() { - long[] newTable = new long[4 * initCapacity]; + long[] newTable = new long[ITEM_SIZE * initCapacity]; Arrays.fill(newTable, EmptyKey); table = newTable; @@ -626,7 +629,7 @@ private static void insertKeyValueNoLock(long[] table, int capacity, long key1, return; } - bucket = (bucket + 4) & (table.length - 1); + bucket = (bucket + ITEM_SIZE) & (table.length - 1); } } } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java index fabdb5087af4a..f7512fef7142c 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java @@ -294,6 +294,9 @@ public Set items(int numberOfItems, LongPairFunction longPairConverter // A section is a portion of the hash map that is covered by a single @SuppressWarnings("serial") private static final class Section extends StampedLock { + // Each item take up 2 continuous array space. + private static final int ITEM_SIZE = 2; + // Keys and values are stored interleaved in the table array private volatile long[] table; @@ -315,7 +318,7 @@ private static final class Section extends StampedLock { float expandFactor, float shrinkFactor) { this.capacity = alignToPowerOfTwo(capacity); this.initCapacity = this.capacity; - this.table = new long[2 * this.capacity]; + this.table = new long[ITEM_SIZE * this.capacity]; this.size = 0; this.usedBuckets = 0; this.autoShrink = autoShrink; @@ -335,7 +338,7 @@ boolean contains(long item1, long item2, int hash) { // add local variable here, so OutOfBound won't happen long[] table = this.table; // calculate table.length / 2 as capacity to avoid rehash changing capacity - int bucket = signSafeMod(hash, table.length / 2); + int bucket = signSafeMod(hash, table.length / ITEM_SIZE); try { while (true) { @@ -359,7 +362,7 @@ boolean contains(long item1, long item2, int hash) { // update local variable table = this.table; - bucket = signSafeMod(hash, table.length / 2); + bucket = signSafeMod(hash, table.length / ITEM_SIZE); storedItem1 = table[bucket]; storedItem2 = table[bucket + 1]; } @@ -372,7 +375,7 @@ boolean contains(long item1, long item2, int hash) { } } - bucket = (bucket + 2) & (table.length - 1); + bucket = (bucket + ITEM_SIZE) & (table.length - 1); } } finally { if (acquiredLock) { @@ -416,7 +419,7 @@ boolean add(long item1, long item2, long hash) { } } - bucket = (bucket + 2) & (table.length - 1); + bucket = (bucket + ITEM_SIZE) & (table.length - 1); } } finally { if (usedBuckets > resizeThresholdUp) { @@ -451,7 +454,7 @@ private boolean remove(long item1, long item2, int hash) { return false; } - bucket = (bucket + 2) & (table.length - 1); + bucket = (bucket + ITEM_SIZE) & (table.length - 1); } } finally { tryShrinkThenUnlock(stamp); @@ -465,7 +468,7 @@ private int removeIf(LongPairPredicate filter) { // Go through all the buckets for this section long stamp = writeLock(); try { - for (int bucket = 0; bucket < table.length; bucket += 2) { + for (int bucket = 0; bucket < table.length; bucket += ITEM_SIZE) { long storedItem1 = table[bucket]; long storedItem2 = table[bucket + 1]; if (storedItem1 != DeletedItem && storedItem1 != EmptyItem) { @@ -504,7 +507,7 @@ private void tryShrinkThenUnlock(long stamp) { } private void cleanBucket(int bucket) { - int nextInArray = (bucket + 2) & (table.length - 1); + int nextInArray = (bucket + ITEM_SIZE) & (table.length - 1); if (table[nextInArray] == EmptyItem) { table[bucket] = EmptyItem; table[bucket + 1] = EmptyItem; @@ -512,13 +515,13 @@ private void cleanBucket(int bucket) { // Cleanup all the buckets that were in `DeletedItem` state, // so that we can reduce unnecessary expansions - int lastBucket = (bucket - 2) & (table.length - 1); + int lastBucket = (bucket - ITEM_SIZE) & (table.length - 1); while (table[lastBucket] == DeletedItem) { table[lastBucket] = EmptyItem; table[lastBucket + 1] = EmptyItem; --usedBuckets; - lastBucket = (lastBucket - 2) & (table.length - 1); + lastBucket = (lastBucket - ITEM_SIZE) & (table.length - 1); } } else { table[bucket] = DeletedItem; @@ -548,7 +551,7 @@ public void forEach(LongPairConsumer processor) { // Go through all the buckets for this section. We try to renew the stamp only after a validation // error, otherwise we keep going with the same. long stamp = 0; - for (int bucket = 0; bucket < table.length; bucket += 2) { + for (int bucket = 0; bucket < table.length; bucket += ITEM_SIZE) { if (stamp == 0) { stamp = tryOptimisticRead(); } @@ -578,11 +581,11 @@ public void forEach(LongPairConsumer processor) { private void rehash(int newCapacity) { // Expand the hashmap - long[] newTable = new long[2 * newCapacity]; + long[] newTable = new long[ITEM_SIZE * newCapacity]; Arrays.fill(newTable, EmptyItem); // Re-hash table - for (int i = 0; i < table.length; i += 2) { + for (int i = 0; i < table.length; i += ITEM_SIZE) { long storedItem1 = table[i]; long storedItem2 = table[i + 1]; if (storedItem1 != EmptyItem && storedItem1 != DeletedItem) { @@ -601,7 +604,7 @@ private void rehash(int newCapacity) { private void shrinkToInitCapacity() { // Expand the hashmap - long[] newTable = new long[2 * initCapacity]; + long[] newTable = new long[ITEM_SIZE * initCapacity]; Arrays.fill(newTable, EmptyItem); table = newTable; @@ -627,7 +630,7 @@ private static void insertKeyValueNoLock(long[] table, int capacity, long item1, return; } - bucket = (bucket + 2) & (table.length - 1); + bucket = (bucket + ITEM_SIZE) & (table.length - 1); } } } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java index 6270f94b2bb7f..effec88fbeb58 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java @@ -296,6 +296,9 @@ public List values() { // A section is a portion of the hash map that is covered by a single @SuppressWarnings("serial") private static final class Section extends StampedLock { + // Each item take up 2 continuous array space. + private static final int ITEM_SIZE = 2; + // Keys and values are stored interleaved in the table array private volatile Object[] table; @@ -317,7 +320,7 @@ private static final class Section extends StampedLock { float expandFactor, float shrinkFactor) { this.capacity = alignToPowerOfTwo(capacity); this.initCapacity = this.capacity; - this.table = new Object[2 * this.capacity]; + this.table = new Object[ITEM_SIZE * this.capacity]; this.size = 0; this.usedBuckets = 0; this.autoShrink = autoShrink; @@ -336,7 +339,7 @@ V get(K key, int keyHash) { // add local variable here, so OutOfBound won't happen Object[] table = this.table; // calculate table.length / 2 as capacity to avoid rehash changing capacity - int bucket = signSafeMod(keyHash, table.length / 2); + int bucket = signSafeMod(keyHash, table.length / ITEM_SIZE); try { while (true) { @@ -360,7 +363,7 @@ V get(K key, int keyHash) { // update local variable table = this.table; - bucket = signSafeMod(keyHash, table.length / 2); + bucket = signSafeMod(keyHash, table.length / ITEM_SIZE); storedKey = (K) table[bucket]; storedValue = (V) table[bucket + 1]; } @@ -373,7 +376,7 @@ V get(K key, int keyHash) { } } - bucket = (bucket + 2) & (table.length - 1); + bucket = (bucket + ITEM_SIZE) & (table.length - 1); } } finally { if (acquiredLock) { @@ -426,7 +429,7 @@ V put(K key, V value, int keyHash, boolean onlyIfAbsent, Function valuePro } } - bucket = (bucket + 2) & (table.length - 1); + bucket = (bucket + ITEM_SIZE) & (table.length - 1); } } finally { if (usedBuckets > resizeThresholdUp) { @@ -455,7 +458,7 @@ private V remove(K key, Object value, int keyHash) { if (value == null || value.equals(storedValue)) { SIZE_UPDATER.decrementAndGet(this); - int nextInArray = (bucket + 2) & (table.length - 1); + int nextInArray = (bucket + ITEM_SIZE) & (table.length - 1); if (table[nextInArray] == EmptyKey) { table[bucket] = EmptyKey; table[bucket + 1] = null; @@ -463,13 +466,13 @@ private V remove(K key, Object value, int keyHash) { // Cleanup all the buckets that were in `DeletedKey` state, // so that we can reduce unnecessary expansions - int lastBucket = (bucket - 2) & (table.length - 1); + int lastBucket = (bucket - ITEM_SIZE) & (table.length - 1); while (table[lastBucket] == DeletedKey) { table[lastBucket] = EmptyKey; table[lastBucket + 1] = null; --usedBuckets; - lastBucket = (lastBucket - 2) & (table.length - 1); + lastBucket = (lastBucket - ITEM_SIZE) & (table.length - 1); } } else { table[bucket] = DeletedKey; @@ -485,7 +488,7 @@ private V remove(K key, Object value, int keyHash) { return null; } - bucket = (bucket + 2) & (table.length - 1); + bucket = (bucket + ITEM_SIZE) & (table.length - 1); } } finally { @@ -534,7 +537,7 @@ public void forEach(BiConsumer processor) { // Go through all the buckets for this section. We try to renew the stamp only after a validation // error, otherwise we keep going with the same. long stamp = 0; - for (int bucket = 0; bucket < table.length; bucket += 2) { + for (int bucket = 0; bucket < table.length; bucket += ITEM_SIZE) { if (stamp == 0) { stamp = tryOptimisticRead(); } @@ -564,10 +567,10 @@ public void forEach(BiConsumer processor) { private void rehash(int newCapacity) { // Expand the hashmap - Object[] newTable = new Object[2 * newCapacity]; + Object[] newTable = new Object[ITEM_SIZE * newCapacity]; // Re-hash table - for (int i = 0; i < table.length; i += 2) { + for (int i = 0; i < table.length; i += ITEM_SIZE) { K storedKey = (K) table[i]; V storedValue = (V) table[i + 1]; if (storedKey != EmptyKey && storedKey != DeletedKey) { @@ -583,7 +586,7 @@ private void rehash(int newCapacity) { } private void shrinkToInitCapacity() { - Object[] newTable = new Object[2 * initCapacity]; + Object[] newTable = new Object[ITEM_SIZE * initCapacity]; table = newTable; size = 0; @@ -608,7 +611,7 @@ private static void insertKeyValueNoLock(Object[] table, int capacity, K return; } - bucket = (bucket + 2) & (table.length - 1); + bucket = (bucket + ITEM_SIZE) & (table.length - 1); } } } From 79d557ec324c6106107ece459bc62ce1ff693872 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 1 Sep 2023 15:36:14 +0800 Subject: [PATCH 20/27] add some comments. --- .../common/util/collections/ConcurrentLongLongPairHashMap.java | 2 ++ .../pulsar/common/util/collections/ConcurrentLongPairSet.java | 2 ++ .../pulsar/common/util/collections/ConcurrentOpenHashMap.java | 2 ++ 3 files changed, 6 insertions(+) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java index 43279ffbd18c3..c3babbb8d1103 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java @@ -648,6 +648,8 @@ static final long hash(long key1, long key2) { } static final int signSafeMod(long n, int max) { + // as the ITEM_SIZE of Section is 4, so the index is the multiple of 4 + // that is to left shift 2 bits return (int) (n & (max - 1)) << 2; } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java index f7512fef7142c..f0f04c4edf904 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSet.java @@ -649,6 +649,8 @@ static final long hash(long key1, long key2) { } static final int signSafeMod(long n, int max) { + // as the ITEM_SIZE of Section is 2, so the index is the multiple of 2 + // that is to left shift 1 bit return (int) (n & (max - 1)) << 1; } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java index effec88fbeb58..1aa95d3090eb2 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMap.java @@ -627,6 +627,8 @@ static final long hash(K key) { } static final int signSafeMod(long n, int max) { + // as the ITEM_SIZE of Section is 2, so the index is the multiple of 2 + // that is to left shift 1 bit return (int) (n & (max - 1)) << 1; } From 346315008f64c910aaeb1d96a1192a02604d089b Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Wed, 13 Sep 2023 10:16:58 +0800 Subject: [PATCH 21/27] add invocationCount for test. --- .../common/util/collections/ConcurrentLongHashMapTest.java | 2 +- .../util/collections/ConcurrentLongLongPairHashMapTest.java | 2 +- .../common/util/collections/ConcurrentLongPairSetTest.java | 2 +- .../common/util/collections/ConcurrentOpenHashMapTest.java | 2 +- .../common/util/collections/ConcurrentOpenHashSetTest.java | 2 +- 5 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java index b64714337ec3d..61c847064dcec 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java @@ -214,7 +214,7 @@ public void testExpandShrinkAndClear() { assertTrue(map.capacity() == initCapacity); } - @Test + @Test(invocationCount = 10) public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ConcurrentLongHashMap map = ConcurrentLongHashMap.newBuilder() .expectedItems(2) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java index 0de3fdb5c84bf..80b43e35ffcad 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java @@ -176,7 +176,7 @@ public void testExpandAndShrink() { assertEquals(map.capacity(), 8); } - @Test + @Test(invocationCount = 10) public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ConcurrentLongLongPairHashMap map = ConcurrentLongLongPairHashMap.newBuilder() .expectedItems(2) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java index bce2b8993835f..2f5c25a0612f9 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java @@ -213,7 +213,7 @@ public void testExpandShrinkAndClear() { assertTrue(map.capacity() == initCapacity); } - @Test + @Test(invocationCount = 10) public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ConcurrentLongPairSet set = ConcurrentLongPairSet.newBuilder() .expectedItems(2) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java index 410d490b98faa..ce5d5b073e14e 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java @@ -217,7 +217,7 @@ public void testExpandShrinkAndClear() { assertTrue(map.capacity() == initCapacity); } - @Test + @Test(invocationCount = 10) public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ConcurrentOpenHashMap map = ConcurrentOpenHashMap.newBuilder() .expectedItems(2) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java index 6a40095ab0647..3f4223316077a 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java @@ -188,7 +188,7 @@ public void testExpandShrinkAndClear() { assertTrue(map.capacity() == initCapacity); } - @Test + @Test(invocationCount = 10) public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ConcurrentOpenHashSet set = ConcurrentOpenHashSet.newBuilder() .expectedItems(2) From dd4d14b8c1d892e7bb43b7338f54f29b79ca59ec Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 15 Sep 2023 12:11:09 +0800 Subject: [PATCH 22/27] add test timeout. --- .../common/util/collections/TripleLongPriorityQueueTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java index 01748e2b4ed97..7e92c6ec0c4d2 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java @@ -54,7 +54,7 @@ public void testQueue() { pq.close(); } - @Test + @Test(timeOut = 500000) public void testLargeQueue() { TripleLongPriorityQueue pq = new TripleLongPriorityQueue(); assertEquals(pq.size(), 0); From 7601d50de92541712999b20f6f7665ec2e5ca778 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 15 Sep 2023 12:55:59 +0800 Subject: [PATCH 23/27] add timeout. --- .../common/util/collections/TripleLongPriorityQueueTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java index 7e92c6ec0c4d2..bfab4f164114e 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java @@ -54,7 +54,7 @@ public void testQueue() { pq.close(); } - @Test(timeOut = 500000) + @Test(timeOut = 1000000) public void testLargeQueue() { TripleLongPriorityQueue pq = new TripleLongPriorityQueue(); assertEquals(pq.size(), 0); From 9dcf4d36b73c5033ea06f2ece7baa03cfbe9db7b Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 15 Sep 2023 13:04:41 +0800 Subject: [PATCH 24/27] add --- .../test/java/org/apache/pulsar/common/util/FutureUtilTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java index 6df4494edf886..07bcf70097619 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java @@ -181,7 +181,7 @@ public void testWaitForAny() { } } - @Test + @Test(timeOut = 1000000) public void testSequencer() { int concurrentNum = 1000; final ScheduledExecutorService executor = Executors.newScheduledThreadPool(concurrentNum); From c9b40ac364a13b998d35af298b914ede2e4db5db Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 15 Sep 2023 14:05:57 +0800 Subject: [PATCH 25/27] add timeout. --- .../common/util/collections/ConcurrentLongHashMapTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java index 61c847064dcec..8f99ae93e5db1 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java @@ -422,7 +422,7 @@ public void concurrentInsertionsAndReads() throws Throwable { assertEquals(map.size(), N * nThreads); } - @Test + @Test(timeOut = 1000000) public void stressConcurrentInsertionsAndReads() throws Throwable { ConcurrentLongHashMap map = ConcurrentLongHashMap.newBuilder() .expectedItems(4) From 33cf3f56120d3597ffa5bcdc08da7b78b36beab4 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 15 Sep 2023 15:08:57 +0800 Subject: [PATCH 26/27] remove invocationCount. --- .../java/org/apache/pulsar/common/util/FutureUtilTest.java | 1 - .../common/util/collections/ConcurrentLongHashMapTest.java | 3 +-- .../util/collections/ConcurrentLongLongPairHashMapTest.java | 2 +- .../common/util/collections/ConcurrentLongPairSetTest.java | 2 +- .../common/util/collections/ConcurrentOpenHashMapTest.java | 2 +- .../common/util/collections/ConcurrentOpenHashSetTest.java | 2 +- 6 files changed, 5 insertions(+), 7 deletions(-) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java index 07bcf70097619..7d44c187d7355 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java @@ -181,7 +181,6 @@ public void testWaitForAny() { } } - @Test(timeOut = 1000000) public void testSequencer() { int concurrentNum = 1000; final ScheduledExecutorService executor = Executors.newScheduledThreadPool(concurrentNum); diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java index 8f99ae93e5db1..e1f947ad8c4f6 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongHashMapTest.java @@ -214,7 +214,7 @@ public void testExpandShrinkAndClear() { assertTrue(map.capacity() == initCapacity); } - @Test(invocationCount = 10) + @Test public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ConcurrentLongHashMap map = ConcurrentLongHashMap.newBuilder() .expectedItems(2) @@ -422,7 +422,6 @@ public void concurrentInsertionsAndReads() throws Throwable { assertEquals(map.size(), N * nThreads); } - @Test(timeOut = 1000000) public void stressConcurrentInsertionsAndReads() throws Throwable { ConcurrentLongHashMap map = ConcurrentLongHashMap.newBuilder() .expectedItems(4) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java index 80b43e35ffcad..0de3fdb5c84bf 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMapTest.java @@ -176,7 +176,7 @@ public void testExpandAndShrink() { assertEquals(map.capacity(), 8); } - @Test(invocationCount = 10) + @Test public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ConcurrentLongLongPairHashMap map = ConcurrentLongLongPairHashMap.newBuilder() .expectedItems(2) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java index 2f5c25a0612f9..bce2b8993835f 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairSetTest.java @@ -213,7 +213,7 @@ public void testExpandShrinkAndClear() { assertTrue(map.capacity() == initCapacity); } - @Test(invocationCount = 10) + @Test public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ConcurrentLongPairSet set = ConcurrentLongPairSet.newBuilder() .expectedItems(2) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java index ce5d5b073e14e..410d490b98faa 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashMapTest.java @@ -217,7 +217,7 @@ public void testExpandShrinkAndClear() { assertTrue(map.capacity() == initCapacity); } - @Test(invocationCount = 10) + @Test public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ConcurrentOpenHashMap map = ConcurrentOpenHashMap.newBuilder() .expectedItems(2) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java index 3f4223316077a..6a40095ab0647 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenHashSetTest.java @@ -188,7 +188,7 @@ public void testExpandShrinkAndClear() { assertTrue(map.capacity() == initCapacity); } - @Test(invocationCount = 10) + @Test public void testConcurrentExpandAndShrinkAndGet() throws Throwable { ConcurrentOpenHashSet set = ConcurrentOpenHashSet.newBuilder() .expectedItems(2) From dd2fdf3be52e7a7de8614b9bc9d0e15950d5b4e5 Mon Sep 17 00:00:00 2001 From: thetumbled <843221020@qq.com> Date: Fri, 15 Sep 2023 16:11:39 +0800 Subject: [PATCH 27/27] remove timeout. --- .../common/util/collections/TripleLongPriorityQueueTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java index bfab4f164114e..01748e2b4ed97 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java @@ -54,7 +54,7 @@ public void testQueue() { pq.close(); } - @Test(timeOut = 1000000) + @Test public void testLargeQueue() { TripleLongPriorityQueue pq = new TripleLongPriorityQueue(); assertEquals(pq.size(), 0);