From 9dccf8aad60f27da73834efc8cb56a23350fa969 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 3 Mar 2021 11:26:08 -0800 Subject: [PATCH 1/2] Ensure read-lock is not continuously held on a section while iterating over concurrent maps --- .../collections/ConcurrentLongHashMap.java | 64 ++++++++-------- .../collections/ConcurrentLongPairSet.java | 75 ++++++++----------- .../collections/ConcurrentOpenHashMap.java | 63 +++++++--------- .../collections/ConcurrentOpenHashSet.java | 68 +++++++---------- 4 files changed, 116 insertions(+), 154 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 e4a8f2e738de3..f658f2f7cb9e7 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 @@ -153,14 +153,14 @@ private Section getSection(long hash) { } public void clear() { - for (Section s : sections) { - s.clear(); + for (int i = 0; i < sections.length; i++) { + sections[i].clear(); } } public void forEach(EntryProcessor processor) { - for (Section s : sections) { - s.forEach(processor); + for (int i = 0; i < sections.length; i++) { + sections[i].forEach(processor); } } @@ -393,46 +393,44 @@ void clear() { public void forEach(EntryProcessor processor) { long stamp = tryOptimisticRead(); + // We need to make sure that we read these 3 variables in a consistent way int capacity = this.capacity; long[] keys = this.keys; V[] values = this.values; - boolean acquiredReadLock = false; + // Validate no rehashing + if (!validate(stamp)) { + // Fallback to read lock + stamp = readLock(); - try { - - // Validate no rehashing - if (!validate(stamp)) { - // Fallback to read lock - stamp = readLock(); - acquiredReadLock = true; + capacity = this.capacity; + keys = this.keys; + values = this.values; + unlockRead(stamp); + } - capacity = this.capacity; - keys = this.keys; - values = this.values; + // 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. + for (int bucket = 0; bucket < capacity; bucket++) { + if (stamp == 0) { + stamp = tryOptimisticRead(); } - // Go through all the buckets for this section - for (int bucket = 0; bucket < capacity; bucket++) { - long storedKey = keys[bucket]; - V storedValue = values[bucket]; - - if (!acquiredReadLock && !validate(stamp)) { - // Fallback to acquiring read lock - stamp = readLock(); - acquiredReadLock = true; + long storedKey = keys[bucket]; + V storedValue = values[bucket]; - storedKey = keys[bucket]; - storedValue = values[bucket]; - } + if (!validate(stamp)) { + // Fallback to acquiring read lock + stamp = readLock(); - if (storedValue != DeletedValue && storedValue != EmptyValue) { - processor.accept(storedKey, storedValue); - } - } - } finally { - if (acquiredReadLock) { + storedKey = keys[bucket]; + storedValue = values[bucket]; unlockRead(stamp); + stamp = 0; + } + + if (storedValue != DeletedValue && storedValue != EmptyValue) { + processor.accept(storedKey, storedValue); } } } 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 8a7a59eb5c821..eaa827a8c470a 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 @@ -86,23 +86,23 @@ public ConcurrentLongPairSet(int expectedItems, int concurrencyLevel) { public long size() { long size = 0; - for (Section s : sections) { - size += s.size; + for (int i = 0; i < sections.length; i++) { + size += sections[i].size; } return size; } public long capacity() { long capacity = 0; - for (Section s : sections) { - capacity += s.capacity; + for (int i = 0; i < sections.length; i++) { + capacity += sections[i].capacity; } return capacity; } public boolean isEmpty() { - for (Section s : sections) { - if (s.size != 0) { + for (int i = 0; i < sections.length; i++) { + if (sections[i].size != 0) { return false; } } @@ -111,8 +111,8 @@ public boolean isEmpty() { long getUsedBucketCount() { long usedBucketCount = 0; - for (Section s : sections) { - usedBucketCount += s.usedBuckets; + for (int i = 0; i < sections.length; i++) { + usedBucketCount += sections[i].usedBuckets; } return usedBucketCount; } @@ -154,8 +154,8 @@ public void clear() { } public void forEach(LongPairConsumer processor) { - for (Section s : sections) { - s.forEach(processor); + for (int i = 0; i < sections.length; i++) { + sections[i].forEach(processor); } } @@ -169,8 +169,8 @@ public void forEach(LongPairConsumer processor) { */ public int removeIf(LongPairPredicate filter) { int removedValues = 0; - for (Section s : sections) { - removedValues += s.removeIf(filter); + for (int i = 0; i < sections.length; i++) { + removedValues += sections[i].removeIf(filter); } return removedValues; } @@ -194,8 +194,8 @@ public Set items(int numberOfItems) { @Override public Set items(int numberOfItems, LongPairFunction longPairConverter) { Set items = new HashSet<>(); - for (Section s : sections) { - s.forEach((item1, item2) -> { + for (int i = 0; i < sections.length; i++) { + sections[i].forEach((item1, item2) -> { if (items.size() < numberOfItems) { items.add(longPairConverter.apply(item1, item2)); } @@ -398,42 +398,31 @@ void clear() { } public void forEach(LongPairConsumer processor) { - long stamp = tryOptimisticRead(); - long[] table = this.table; - boolean acquiredReadLock = false; - - try { - // Validate no rehashing - if (!validate(stamp)) { - // Fallback to read lock - stamp = readLock(); - acquiredReadLock = true; - table = this.table; + // 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) { + if (stamp == 0) { + stamp = tryOptimisticRead(); } - // Go through all the buckets for this section - for (int bucket = 0; bucket < table.length; bucket += 2) { - long storedItem1 = table[bucket]; - long storedItem2 = table[bucket + 1]; - - if (!acquiredReadLock && !validate(stamp)) { - // Fallback to acquiring read lock - stamp = readLock(); - acquiredReadLock = true; + long storedItem1 = table[bucket]; + long storedItem2 = table[bucket + 1]; - storedItem1 = table[bucket]; - storedItem2 = table[bucket + 1]; - } + if (!validate(stamp)) { + // Fallback to acquiring read lock + stamp = readLock(); - if (storedItem1 != DeletedItem && storedItem1 != EmptyItem) { - processor.accept(storedItem1, storedItem2); - } - } - } finally { - if (acquiredReadLock) { + storedItem1 = table[bucket]; + storedItem2 = table[bucket + 1]; unlockRead(stamp); + stamp = 0; + } + + if (storedItem1 != DeletedItem && storedItem1 != EmptyItem) { + processor.accept(storedItem1, storedItem2); } } } 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 528f86291ee0d..1016f4f9ee492 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 @@ -20,7 +20,7 @@ import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkNotNull; -import com.google.common.collect.Lists; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; @@ -149,14 +149,14 @@ private Section getSection(long hash) { } public void clear() { - for (Section s : sections) { - s.clear(); + for (int i = 0; i < sections.length; i++) { + sections[i].clear(); } } public void forEach(BiConsumer processor) { - for (Section s : sections) { - s.forEach(processor); + for (int i = 0; i < sections.length; i++) { + sections[i].forEach(processor); } } @@ -164,13 +164,13 @@ public void forEach(BiConsumer processor) { * @return a new list of all keys (makes a copy) */ public List keys() { - List keys = Lists.newArrayList(); + List keys = new ArrayList<>((int) size()); forEach((key, value) -> keys.add(key)); return keys; } public List values() { - List values = Lists.newArrayList(); + List values = new ArrayList<>((int) size()); forEach((key, value) -> values.add(value)); return values; } @@ -354,42 +354,33 @@ void clear() { } public void forEach(BiConsumer processor) { - long stamp = tryOptimisticRead(); - + // Take a reference to the data table, if there is a rehashing event, we'll be + // simply iterating over a snapshot of the data. Object[] table = this.table; - boolean acquiredReadLock = false; - - try { - // Validate no rehashing - if (!validate(stamp)) { - // Fallback to read lock - stamp = readLock(); - acquiredReadLock = true; - table = this.table; + // 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) { + if (stamp == 0) { + stamp = tryOptimisticRead(); } - // Go through all the buckets for this section - for (int bucket = 0; bucket < table.length; bucket += 2) { - K storedKey = (K) table[bucket]; - V storedValue = (V) table[bucket + 1]; - - if (!acquiredReadLock && !validate(stamp)) { - // Fallback to acquiring read lock - stamp = readLock(); - acquiredReadLock = true; + K storedKey = (K) table[bucket]; + V storedValue = (V) table[bucket + 1]; - storedKey = (K) table[bucket]; - storedValue = (V) table[bucket + 1]; - } + if (!validate(stamp)) { + // Fallback to acquiring read lock + stamp = readLock(); - if (storedKey != DeletedKey && storedKey != EmptyKey) { - processor.accept(storedKey, storedValue); - } - } - } finally { - if (acquiredReadLock) { + storedKey = (K) table[bucket]; + storedValue = (V) table[bucket + 1]; unlockRead(stamp); + stamp = 0; + } + + if (storedKey != DeletedKey && storedKey != EmptyKey) { + processor.accept(storedKey, storedValue); } } } 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 d29b42aed6ef0..f5602181d1452 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 @@ -75,23 +75,23 @@ public ConcurrentOpenHashSet(int expectedItems, int concurrencyLevel) { public long size() { long size = 0; - for (Section s : sections) { - size += s.size; + for (int i = 0; i < sections.length; i++) { + size += sections[i].size; } return size; } public long capacity() { long capacity = 0; - for (Section s : sections) { - capacity += s.capacity; + for (int i = 0; i < sections.length; i++) { + capacity += sections[i].capacity; } return capacity; } public boolean isEmpty() { - for (Section s : sections) { - if (s.size != 0) { + for (int i = 0; i < sections.length; i++) { + if (sections[i].size != 0) { return false; } } @@ -124,14 +124,14 @@ private Section getSection(long hash) { } public void clear() { - for (Section s : sections) { - s.clear(); + for (int i = 0; i < sections.length; i++) { + sections[i].clear(); } } public void forEach(Consumer processor) { - for (Section s : sections) { - s.forEach(processor); + for (int i = 0; i < sections.length; i++) { + sections[i].forEach(processor); } } @@ -139,8 +139,8 @@ public int removeIf(Predicate filter) { checkNotNull(filter); int removedCount = 0; - for (Section s : sections) { - removedCount += s.removeIf(filter); + for (int i = 0; i < sections.length; i++) { + removedCount += sections[i].removeIf(filter); } return removedCount; @@ -371,44 +371,28 @@ private void cleanBucket(int bucket) { } public void forEach(Consumer processor) { - long stamp = tryOptimisticRead(); - - int capacity = this.capacity; V[] values = this.values; - boolean acquiredReadLock = false; + // 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 < capacity; bucket++) { + if (stamp == 0) { + stamp = tryOptimisticRead(); + } - try { + V storedValue = values[bucket]; - // Validate no rehashing if (!validate(stamp)) { - // Fallback to read lock + // Fallback to acquiring read lock stamp = readLock(); - acquiredReadLock = true; - - capacity = this.capacity; - values = this.values; + storedValue = values[bucket]; + unlockRead(stamp); + stamp = 0; } - // Go through all the buckets for this section - for (int bucket = 0; bucket < capacity; bucket++) { - V storedValue = values[bucket]; - - if (!acquiredReadLock && !validate(stamp)) { - // Fallback to acquiring read lock - stamp = readLock(); - acquiredReadLock = true; - - storedValue = values[bucket]; - } - - if (storedValue != DeletedValue && storedValue != EmptyValue) { - processor.accept(storedValue); - } - } - } finally { - if (acquiredReadLock) { - unlockRead(stamp); + if (storedValue != DeletedValue && storedValue != EmptyValue) { + processor.accept(storedValue); } } } From b7360a190dfd8fbcdfe9d5cb66872c324c48595d Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 3 Mar 2021 12:30:12 -0800 Subject: [PATCH 2/2] Added try/finally --- .../common/util/collections/ConcurrentLongHashMap.java | 10 +++++++--- .../common/util/collections/ConcurrentLongPairSet.java | 10 +++++++--- .../common/util/collections/ConcurrentOpenHashMap.java | 10 +++++++--- .../common/util/collections/ConcurrentOpenHashSet.java | 9 +++++++-- 4 files changed, 28 insertions(+), 11 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 f658f2f7cb9e7..cd285221bc862 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 @@ -423,9 +423,13 @@ public void forEach(EntryProcessor processor) { // Fallback to acquiring read lock stamp = readLock(); - storedKey = keys[bucket]; - storedValue = values[bucket]; - unlockRead(stamp); + try { + storedKey = keys[bucket]; + storedValue = values[bucket]; + } finally { + unlockRead(stamp); + } + stamp = 0; } 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 eaa827a8c470a..d537793a9dfbd 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 @@ -415,9 +415,13 @@ public void forEach(LongPairConsumer processor) { // Fallback to acquiring read lock stamp = readLock(); - storedItem1 = table[bucket]; - storedItem2 = table[bucket + 1]; - unlockRead(stamp); + try { + storedItem1 = table[bucket]; + storedItem2 = table[bucket + 1]; + } finally { + unlockRead(stamp); + } + stamp = 0; } 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 1016f4f9ee492..0388cd33aa7ca 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 @@ -373,9 +373,13 @@ public void forEach(BiConsumer processor) { // Fallback to acquiring read lock stamp = readLock(); - storedKey = (K) table[bucket]; - storedValue = (V) table[bucket + 1]; - unlockRead(stamp); + try { + storedKey = (K) table[bucket]; + storedValue = (V) table[bucket + 1]; + } finally { + unlockRead(stamp); + } + stamp = 0; } 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 f5602181d1452..c846642a8aca0 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 @@ -386,8 +386,13 @@ public void forEach(Consumer processor) { if (!validate(stamp)) { // Fallback to acquiring read lock stamp = readLock(); - storedValue = values[bucket]; - unlockRead(stamp); + + try { + storedValue = values[bucket]; + } finally { + unlockRead(stamp); + } + stamp = 0; }