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..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 @@ -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,48 @@ 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]; + long storedKey = keys[bucket]; + V storedValue = values[bucket]; - if (!acquiredReadLock && !validate(stamp)) { - // Fallback to acquiring read lock - stamp = readLock(); - acquiredReadLock = true; + if (!validate(stamp)) { + // Fallback to acquiring read lock + stamp = readLock(); + try { storedKey = keys[bucket]; storedValue = values[bucket]; + } finally { + unlockRead(stamp); } - if (storedValue != DeletedValue && storedValue != EmptyValue) { - processor.accept(storedKey, storedValue); - } + stamp = 0; } - } finally { - if (acquiredReadLock) { - unlockRead(stamp); + + 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..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 @@ -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,35 @@ 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]; + long storedItem1 = table[bucket]; + long storedItem2 = table[bucket + 1]; - if (!acquiredReadLock && !validate(stamp)) { - // Fallback to acquiring read lock - stamp = readLock(); - acquiredReadLock = true; + if (!validate(stamp)) { + // Fallback to acquiring read lock + stamp = readLock(); + try { storedItem1 = table[bucket]; storedItem2 = table[bucket + 1]; + } finally { + unlockRead(stamp); } - if (storedItem1 != DeletedItem && storedItem1 != EmptyItem) { - processor.accept(storedItem1, storedItem2); - } + stamp = 0; } - } finally { - if (acquiredReadLock) { - unlockRead(stamp); + + 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..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 @@ -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,37 @@ 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]; + K storedKey = (K) table[bucket]; + V storedValue = (V) table[bucket + 1]; - if (!acquiredReadLock && !validate(stamp)) { - // Fallback to acquiring read lock - stamp = readLock(); - acquiredReadLock = true; + if (!validate(stamp)) { + // Fallback to acquiring read lock + stamp = readLock(); + try { storedKey = (K) table[bucket]; storedValue = (V) table[bucket + 1]; + } finally { + unlockRead(stamp); } - if (storedKey != DeletedKey && storedKey != EmptyKey) { - processor.accept(storedKey, storedValue); - } + stamp = 0; } - } finally { - if (acquiredReadLock) { - unlockRead(stamp); + + 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..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 @@ -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,33 @@ 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; - } - - // 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; + try { storedValue = values[bucket]; + } finally { + unlockRead(stamp); } - if (storedValue != DeletedValue && storedValue != EmptyValue) { - processor.accept(storedValue); - } + stamp = 0; } - } finally { - if (acquiredReadLock) { - unlockRead(stamp); + + if (storedValue != DeletedValue && storedValue != EmptyValue) { + processor.accept(storedValue); } } }