From 0ca0507f9a28e50084e793a193322664ab751f2b Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Fri, 1 May 2026 16:38:32 +0800 Subject: [PATCH 1/5] fix ConcurrentLongHashMap throw ArrayIndexOutOfBoundsException When concurrent read write access the map, The key array and value array are not publish at the same time when shrink or expand. This fix encapsulate the key and value in the same field to avoid this happen --- .../collections/ConcurrentLongHashMap.java | 92 ++-- .../ConcurrentLongHashMapTest.java | 332 ++++++++++++-- .../ConcurrentLongHashMapBenchmark.java | 431 ++++++++++++++++++ 3 files changed, 787 insertions(+), 68 deletions(-) create mode 100644 microbenchmarks/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMapBenchmark.java diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMap.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMap.java index 20094a106fd..c664c324288 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMap.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMap.java @@ -310,11 +310,25 @@ public interface EntryProcessor { void accept(long key, V value); } - // A section is a portion of the hash map that is covered by a single + // A section is a portion of the hash map that is covered by a single lock. The keys, values + // and capacity arrays are bundled into an immutable Table snapshot so that readers always see + // a consistent (key, value, length) triple, eliminating the partial-publish race that the + // previous design had to paper over with Math.min(keys.length, values.length). @SuppressWarnings("serial") private static final class Section extends StampedLock { - private volatile long[] keys; - private volatile V[] values; + private static final class Table { + final long[] keys; + final V[] values; + final int capacity; + + Table(long[] keys, V[] values, int capacity) { + this.keys = keys; + this.values = values; + this.capacity = capacity; + } + } + + private volatile Table table; private volatile int capacity; private final int initCapacity; @@ -332,8 +346,7 @@ private static final class Section extends StampedLock { float expandFactor, float shrinkFactor) { this.capacity = alignToPowerOfTwo(capacity); this.initCapacity = this.capacity; - this.keys = new long[this.capacity]; - this.values = (V[]) new Object[this.capacity]; + this.table = new Table<>(new long[this.capacity], (V[]) new Object[this.capacity], this.capacity); this.size = 0; this.usedBuckets = 0; this.autoShrink = autoShrink; @@ -350,11 +363,10 @@ V get(long key, int 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); + Table table = this.table; + long[] keys = table.keys; + V[] values = table.values; + int bucket = signSafeMod(keyHash, table.capacity); try { while (true) { @@ -377,10 +389,10 @@ V get(long key, int keyHash) { stamp = readLock(); acquiredLock = true; - // update local variable - keys = this.keys; - values = this.values; - bucket = signSafeMod(keyHash, values.length); + table = this.table; + keys = table.keys; + values = table.values; + bucket = signSafeMod(keyHash, table.capacity); storedKey = keys[bucket]; storedValue = values[bucket]; } @@ -393,7 +405,7 @@ V get(long key, int keyHash) { } } - bucket = (bucket + 1) & (values.length - 1); + bucket = (bucket + 1) & (table.capacity - 1); } } finally { if (acquiredLock) { @@ -406,7 +418,10 @@ V put(long key, V value, int keyHash, boolean onlyIfAbsent, LongFunction valu int bucket = keyHash; long stamp = writeLock(); - int capacity = this.capacity; + Table table = this.table; + long[] keys = table.keys; + V[] values = table.values; + int capacity = table.capacity; // Remember where we find the first available spot int firstDeletedKey = -1; @@ -474,6 +489,9 @@ V put(long key, V value, int keyHash, boolean onlyIfAbsent, LongFunction valu private void cleanDeletedStatus(int startBucket) { // Cleanup all the buckets that were in `DeletedValue` state, // so that we can reduce unnecessary expansions + Table table = this.table; + V[] values = table.values; + int capacity = table.capacity; int lastBucket = signSafeMod(startBucket - 1, capacity); while (values[lastBucket] == DeletedValue) { values[lastBucket] = (V) EmptyValue; @@ -486,10 +504,13 @@ private void cleanDeletedStatus(int startBucket) { private V remove(long key, Object value, int keyHash) { int bucket = keyHash; long stamp = writeLock(); + Table table = this.table; + long[] keys = table.keys; + V[] values = table.values; + int capacity = table.capacity; try { while (true) { - int capacity = this.capacity; bucket = signSafeMod(bucket, capacity); long storedKey = keys[bucket]; @@ -551,7 +572,10 @@ int removeIf(LongObjectPredicate filter) { int removedCount = 0; try { // Go through all the buckets for this section - int capacity = this.capacity; + Table table = this.table; + long[] keys = table.keys; + V[] values = table.values; + int capacity = table.capacity; for (int bucket = 0; size > 0 && bucket < capacity; bucket++) { long storedKey = keys[bucket]; V storedValue = values[bucket]; @@ -605,8 +629,9 @@ void clear() { if (autoShrink && capacity > initCapacity) { shrinkToInitCapacity(); } else { - Arrays.fill(keys, 0); - Arrays.fill(values, EmptyValue); + Table table = this.table; + Arrays.fill(table.keys, 0); + Arrays.fill(table.values, EmptyValue); this.size = 0; this.usedBuckets = 0; } @@ -618,9 +643,10 @@ void clear() { public void forEach(EntryProcessor processor) { long stamp = tryOptimisticRead(); - int capacity = this.capacity; - long[] keys = this.keys; - V[] values = this.values; + Table table = this.table; + int capacity = table.capacity; + long[] keys = table.keys; + V[] values = table.values; boolean acquiredReadLock = false; @@ -632,9 +658,10 @@ public void forEach(EntryProcessor processor) { stamp = readLock(); acquiredReadLock = true; - capacity = this.capacity; - keys = this.keys; - values = this.values; + table = this.table; + capacity = table.capacity; + keys = table.keys; + values = table.values; } // Go through all the buckets for this section @@ -666,6 +693,9 @@ private void rehash(int newCapacity) { // Expand the hashmap long[] newKeys = new long[newCapacity]; V[] newValues = (V[]) new Object[newCapacity]; + Table table = this.table; + long[] keys = table.keys; + V[] values = table.values; // Re-hash table for (int i = 0; i < keys.length; i++) { @@ -676,11 +706,8 @@ private void rehash(int newCapacity) { } } - keys = newKeys; - values = newValues; + this.table = new Table<>(newKeys, newValues, newCapacity); usedBuckets = size; - // Capacity needs to be updated after the values, so that we won't see - // a capacity value bigger than the actual array size capacity = newCapacity; resizeThresholdUp = (int) (capacity * mapFillFactor); resizeThresholdBelow = (int) (capacity * mapIdleFactor); @@ -690,12 +717,9 @@ private void shrinkToInitCapacity() { long[] newKeys = new long[initCapacity]; V[] newValues = (V[]) new Object[initCapacity]; - keys = newKeys; - values = newValues; + table = new Table<>(newKeys, newValues, initCapacity); size = 0; usedBuckets = 0; - // Capacity needs to be updated after the values, so that we won't see - // a capacity value bigger than the actual array size capacity = initCapacity; resizeThresholdUp = (int) (capacity * mapFillFactor); resizeThresholdBelow = (int) (capacity * mapIdleFactor); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMapTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMapTest.java index b1f1b5437d4..7add2f1eb61 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMapTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMapTest.java @@ -33,6 +33,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Random; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CyclicBarrier; @@ -40,7 +41,9 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.function.LongFunction; import org.junit.Test; @@ -188,64 +191,325 @@ public void testExpandAndShrink() { assertTrue(map.capacity() == 8); } + /** + * Spins many readers against a section that is constantly expanding and shrinking. The + * stable key '1' is never removed, so every read must observe "v1"; volatile keys 2/3 may or + * may not be present at any instant. Any torn read or sentinel leak surfaces as an + * AssertionError or runtime exception captured in {@code ex}. + */ @Test - public void testConcurrentExpandAndShrinkAndGet() throws Throwable { + 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; - CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads); - Future future = null; - AtomicReference ex = new AtomicReference<>(); - for (int i = 0; i < readThreads; i++) { - executor.submit(() -> { - try { + CyclicBarrier barrier = new CyclicBarrier(readThreads + writeThreads); + AtomicReference ex = new AtomicReference<>(); + List> futures = new ArrayList<>(); + AtomicBoolean writerDone = new AtomicBoolean(false); + + try { + assertNull(map.put(1, "v1")); + + for (int i = 0; i < readThreads; i++) { + futures.add(executor.submit(() -> { barrier.await(); - } catch (Exception e) { - throw new RuntimeException(e); - } + try { + while (!writerDone.get()) { + assertEquals("v1", map.get(1)); + map.get(2); + map.get(3); + } + } catch (Throwable t) { + ex.compareAndSet(null, t); + } + return null; + })); + } + + futures.add(executor.submit(() -> { + barrier.await(); try { - map.get(1); - } catch (Exception e) { - ex.set(e); + for (int i = 0; i < n; i++) { + assertNull(map.put(2, "v2")); + assertNull(map.put(3, "v3")); + assertEquals(8, map.capacity()); + + assertTrue(map.remove(2, "v2")); + assertTrue(map.remove(3, "v3")); + assertEquals(4, map.capacity()); + } + } finally { + writerDone.set(true); } - }); + return null; + })); + + for (Future future : futures) { + future.get(60, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); } - assertNull(map.put(1, "v1")); - future = executor.submit(() -> { - try { - barrier.await(); - } catch (Exception e) { - throw new RuntimeException(e); + assertNull(ex.get()); + } + + /** + * Many concurrent writers all targeting the same section so {@code put}/{@code remove} race + * against {@code rehash} (both expand and shrink). Each writer owns a disjoint key range so + * the post-condition is deterministic. Readers concurrently look up every key written. + */ + @Test + public void testConcurrentMultiWriterExpandShrink() throws Throwable { + ConcurrentLongHashMap map = ConcurrentLongHashMap.newBuilder() + .expectedItems(4) + .concurrencyLevel(1) + .autoShrink(true) + .mapIdleFactor(0.20f) + .build(); + + final int writeThreads = 8; + final int readThreads = 8; + final int rounds = 200; + final int keysPerThread = 64; + + ExecutorService executor = Executors.newCachedThreadPool(); + CyclicBarrier barrier = new CyclicBarrier(writeThreads + readThreads); + AtomicReference ex = new AtomicReference<>(); + AtomicBoolean writersDone = new AtomicBoolean(false); + List> futures = new ArrayList<>(); + + try { + for (int t = 0; t < writeThreads; t++) { + final long base = (long) t * keysPerThread; + futures.add(executor.submit(() -> { + barrier.await(); + try { + for (int round = 0; round < rounds; round++) { + for (int k = 0; k < keysPerThread; k++) { + map.put(base + k, "v-" + (base + k)); + } + for (int k = 0; k < keysPerThread; k++) { + assertEquals("v-" + (base + k), map.get(base + k)); + } + for (int k = 0; k < keysPerThread; k++) { + assertEquals("v-" + (base + k), map.remove(base + k)); + } + for (int k = 0; k < keysPerThread; k++) { + assertNull(map.get(base + k)); + } + } + } catch (Throwable th) { + ex.compareAndSet(null, th); + } + return null; + })); + } + + for (int r = 0; r < readThreads; r++) { + futures.add(executor.submit(() -> { + barrier.await(); + try { + long total = (long) writeThreads * keysPerThread; + long key = 0; + while (!writersDone.get()) { + String v = map.get(key); + if (v != null && !v.equals("v-" + key)) { + throw new AssertionError("torn read for key " + key + ": " + v); + } + key = (key + 1) % total; + } + } catch (Throwable th) { + ex.compareAndSet(null, th); + } + return null; + })); + } + + for (int i = 0; i < writeThreads; i++) { + futures.get(i).get(120, TimeUnit.SECONDS); } + writersDone.set(true); + for (int i = writeThreads; i < futures.size(); i++) { + futures.get(i).get(60, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + } - for (int i = 0; i < n; i++) { - // expand hashmap - assertNull(map.put(2, "v2")); - assertNull(map.put(3, "v3")); - assertEquals(map.capacity(), 8); + assertNull(ex.get()); + assertEquals(0, map.size()); + } + + /** + * Differential test against {@link java.util.concurrent.ConcurrentHashMap}. Each thread owns + * a disjoint key partition (so any single-key sequence is linearizable), but every operation + * is mirrored onto both maps. Per-call return values must agree, and after the workload the + * two maps must contain exactly the same entries — including the reverse direction. + */ + @Test + public void testCorrectnessAgainstConcurrentHashMap() throws Throwable { + ConcurrentLongHashMap map = ConcurrentLongHashMap.newBuilder() + .expectedItems(8) + .concurrencyLevel(4) + .autoShrink(true) + .mapIdleFactor(0.20f) + .build(); + ConcurrentHashMap reference = new ConcurrentHashMap<>(); - // shrink hashmap - assertTrue(map.remove(2, "v2")); - assertTrue(map.remove(3, "v3")); - assertEquals(map.capacity(), 4); + final int nThreads = 8; + final int opsPerThread = 50_000; + final int keyRange = 2048; + + ExecutorService executor = Executors.newFixedThreadPool(nThreads); + CyclicBarrier barrier = new CyclicBarrier(nThreads); + List> futures = new ArrayList<>(); + + try { + for (int t = 0; t < nThreads; t++) { + final int threadId = t; + final long base = (long) threadId << 40; + futures.add(executor.submit(() -> { + Random rnd = new Random(threadId); + barrier.await(); + for (int i = 0; i < opsPerThread; i++) { + long key = base + rnd.nextInt(keyRange); + int op = rnd.nextInt(5); + String value = "v-" + threadId + "-" + i; + switch (op) { + case 0: + assertEquals("put differed at key=" + key, + reference.put(key, value), map.put(key, value)); + break; + case 1: + assertEquals("putIfAbsent differed at key=" + key, + reference.putIfAbsent(key, value), map.putIfAbsent(key, value)); + break; + case 2: + assertEquals("remove differed at key=" + key, + reference.remove(key), map.remove(key)); + break; + case 3: + assertEquals("get differed at key=" + key, + reference.get(key), map.get(key)); + break; + default: + assertEquals("containsKey differed at key=" + key, + reference.containsKey(key), map.containsKey(key)); + break; + } + } + return null; + })); } + + for (Future future : futures) { + future.get(120, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + } + + assertEquals("final size mismatch", reference.size(), map.size()); + for (Map.Entry e : reference.entrySet()) { + assertEquals("final value mismatch at key=" + e.getKey(), e.getValue(), map.get(e.getKey())); + } + AtomicLong observed = new AtomicLong(); + map.forEach((k, v) -> { + observed.incrementAndGet(); + assertEquals("orphan key in map: " + k, reference.get(k), v); }); + assertEquals(reference.size(), observed.get()); + } - future.get(); - assertTrue(ex.get() == null); - // shut down pool - executor.shutdown(); + /** + * forEach during concurrent writes is documented as not strongly thread-safe, but it must + * never throw, never expose {@code DeletedValue}/{@code EmptyValue} sentinels, and every + * observed (key, value) pair must be a legitimate pair that was written at some point. + */ + @Test + public void testForEachDuringWrites() throws Throwable { + ConcurrentLongHashMap map = ConcurrentLongHashMap.newBuilder() + .expectedItems(8) + .concurrencyLevel(1) + .autoShrink(true) + .mapIdleFactor(0.25f) + .build(); + + final int writers = 4; + final int keysPerWriter = 256; + final int writeRounds = 200; + final int forEachRounds = 100; + + ExecutorService executor = Executors.newCachedThreadPool(); + CyclicBarrier barrier = new CyclicBarrier(writers + 1); + AtomicReference ex = new AtomicReference<>(); + AtomicBoolean writersDone = new AtomicBoolean(false); + List> futures = new ArrayList<>(); + + try { + for (int t = 0; t < writers; t++) { + final long base = (long) t * keysPerWriter; + futures.add(executor.submit(() -> { + barrier.await(); + try { + for (int round = 0; round < writeRounds; round++) { + for (int k = 0; k < keysPerWriter; k++) { + map.put(base + k, "v-" + (base + k)); + } + for (int k = 0; k < keysPerWriter; k++) { + map.remove(base + k); + } + } + } catch (Throwable th) { + ex.compareAndSet(null, th); + } + return null; + })); + } + + futures.add(executor.submit(() -> { + barrier.await(); + try { + for (int round = 0; round < forEachRounds && !writersDone.get(); round++) { + AtomicInteger seen = new AtomicInteger(); + map.forEach((k, v) -> { + seen.incrementAndGet(); + String expected = "v-" + k; + if (!expected.equals(v)) { + throw new AssertionError("Inconsistent (k,v): (" + k + "," + v + ")"); + } + }); + long sz = map.size(); + assertTrue("size went negative: " + sz, sz >= 0); + assertTrue("size > universe: " + sz, + sz <= (long) writers * keysPerWriter); + } + } catch (Throwable th) { + ex.compareAndSet(null, th); + } + return null; + })); + + for (int i = 0; i < writers; i++) { + futures.get(i).get(120, TimeUnit.SECONDS); + } + writersDone.set(true); + futures.get(writers).get(60, TimeUnit.SECONDS); + } finally { + executor.shutdownNow(); + } + + assertNull(ex.get()); } @Test diff --git a/microbenchmarks/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMapBenchmark.java b/microbenchmarks/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMapBenchmark.java new file mode 100644 index 00000000000..03f115a8ac8 --- /dev/null +++ b/microbenchmarks/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMapBenchmark.java @@ -0,0 +1,431 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.bookkeeper.util.collections; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.bookkeeper.util.collections.ConcurrentLongHashMap.LongObjectPredicate; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Group; +import org.openjdk.jmh.annotations.GroupThreads; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.infra.Blackhole; + +/** + * Benchmarks for the long-keyed concurrent map. + * + *

Compares two implementations: + *

    + *
  • {@code clhm} – {@link ConcurrentLongHashMap} (immutable Table snapshot, primitive long + * keys, zero allocation on the key path),
    • + *
  • {@code chm} – {@link java.util.concurrent.ConcurrentHashMap} as the JDK baseline.
    • + *
+ * + *

Workload mix: + *

    + *
  • {@link #getHit}/{@link #getMiss}/{@link #putRemove} – single-thread basics.
    • + *
  • {@link #concurrentGetHit} – read-only, 16 threads.
    • + *
  • {@link #concurrentMixedReader}/{@link #concurrentMixedWriter} – an asymmetric concurrent + * group: 12 readers + 4 writers operating on disjoint key partitions, so any concurrency + * bug (torn rehash, lost-update, partial-publish) shows up either as a JMH error or a + * reduced ops/sec number on the suspect implementation.
    • + *
  • {@link #concurrentExpandShrinkWriter} – starts the map at the smallest legal capacity and + * hammers a single section with put/remove so that every writer constantly forces a rehash + * (both expand and shrink). This is the highest-pressure rehash-vs-read benchmark.
    • + *
+ */ +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@BenchmarkMode(Mode.AverageTime) +@Fork(1) +@Warmup(iterations = 2, time = 5, timeUnit = TimeUnit.SECONDS) +@Measurement(iterations = 5, time = 5, timeUnit = TimeUnit.SECONDS) +public class ConcurrentLongHashMapBenchmark { + + /** + * Shared benchmark state for the steady-state benchmarks (the map is fully populated up + * front and the workload only mutates keys outside the resident set). + */ + @State(Scope.Benchmark) + public static class MapState { + @Param({"clhm", "chm"}) + private String implementation; + + @Param({"1024", "65536"}) + private int entries; + + private long[] presentKeys; + private long[] absentKeys; + private ConcurrentLongHashMap clhm; + private ConcurrentHashMap chm; + private AtomicLong writeKey; + + @Setup(Level.Trial) + public void setup() { + presentKeys = new long[entries]; + absentKeys = new long[entries]; + clhm = ConcurrentLongHashMap.newBuilder() + .expectedItems(entries) + .concurrencyLevel(16) + .build(); + chm = new ConcurrentHashMap<>(entries, 0.66f, 16); + + for (int i = 0; i < entries; i++) { + long key = i; + presentKeys[i] = key; + absentKeys[i] = key ^ Long.MIN_VALUE; + clhm.put(key, "value"); + chm.put(key, "value"); + } + + writeKey = new AtomicLong(1L << 48); + } + + String get(long key) { + return "clhm".equals(implementation) ? clhm.get(key) : chm.get(key); + } + + void put(long key, String value) { + if ("clhm".equals(implementation)) { + clhm.put(key, value); + } else { + chm.put(key, value); + } + } + + void remove(long key) { + if ("clhm".equals(implementation)) { + clhm.remove(key); + } else { + chm.remove(key); + } + } + + long nextWriteKey() { + return writeKey.getAndIncrement(); + } + } + + /** + * Per-thread cursor state. + */ + @State(Scope.Thread) + public static class CursorState { + private int index; + + int next(int length) { + int value = index; + index = value + 1; + return value & (length - 1); + } + } + + /** + * Independent key-stream state for concurrent benchmarks. Each writer thread owns a unique + * partition of the long key-space so the mutations don't trample each other and the steady + * state remains bounded; readers also walk a private cursor so they don't hot-spot a single + * bucket. + */ + @State(Scope.Thread) + public static class WriterState { + private static final AtomicLong NEXT_PARTITION = new AtomicLong(); + private long base; + private long offset; + // Keep a small per-writer working set so put + remove pair cleanly without unbounded growth. + private static final int WORKING_SET = 1024; + + @Setup(Level.Iteration) + public void setup() { + base = NEXT_PARTITION.getAndIncrement() << 40; + offset = 0; + } + + long nextKey() { + long k = base + (offset & (WORKING_SET - 1)); + offset++; + return k; + } + } + + @Benchmark + public void getHit(MapState map, CursorState cursor, Blackhole blackhole) { + blackhole.consume(map.get(map.presentKeys[cursor.next(map.presentKeys.length)])); + } + + @Benchmark + public void getMiss(MapState map, CursorState cursor, Blackhole blackhole) { + blackhole.consume(map.get(map.absentKeys[cursor.next(map.absentKeys.length)])); + } + + @Benchmark + public void putRemove(MapState map, Blackhole blackhole) { + long key = map.nextWriteKey(); + map.put(key, "value"); + blackhole.consume(map.get(key)); + map.remove(key); + } + + @Benchmark + @Threads(16) + public void concurrentGetHit(MapState map, CursorState cursor, Blackhole blackhole) { + blackhole.consume(map.get(map.presentKeys[cursor.next(map.presentKeys.length)])); + } + + /** Reader half of the asymmetric mixed-workload group: 12 reader threads. */ + @Benchmark + @Group("concurrentMixed") + @GroupThreads(12) + public void concurrentMixedReader(MapState map, CursorState cursor, Blackhole blackhole) { + blackhole.consume(map.get(map.presentKeys[cursor.next(map.presentKeys.length)])); + } + + /** Writer half of the asymmetric mixed-workload group: 4 writer threads. */ + @Benchmark + @Group("concurrentMixed") + @GroupThreads(4) + public void concurrentMixedWriter(MapState map, WriterState w, Blackhole blackhole) { + long key = w.nextKey(); + map.put(key, "value"); + blackhole.consume(map.get(key)); + map.remove(key); + } + + /** + * Holds an aggressively-shrinking, single-section map. Each writer thread's put/remove pair + * crosses the expand and shrink thresholds, so the rehash code path is exercised on nearly + * every operation. Reader threads chase the writers to surface read-vs-rehash races. + */ + @State(Scope.Benchmark) + public static class ChurningMapState { + @Param({"clhm", "chm"}) + private String implementation; + + private ConcurrentLongHashMap clhm; + private ConcurrentHashMap chm; + + @Setup(Level.Iteration) + public void setup() { + clhm = ConcurrentLongHashMap.newBuilder() + .expectedItems(2) + .concurrencyLevel(1) + .autoShrink(true) + .mapIdleFactor(0.25f) + .build(); + chm = new ConcurrentHashMap<>(4, 0.66f, 1); + } + + String get(long key) { + return "clhm".equals(implementation) ? clhm.get(key) : chm.get(key); + } + + String put(long key, String value) { + return "clhm".equals(implementation) ? clhm.put(key, value) : chm.put(key, value); + } + + String remove(long key) { + return "clhm".equals(implementation) ? clhm.remove(key) : chm.remove(key); + } + } + + /** Writer driving constant expand+shrink on a single section. */ + @Benchmark + @Group("concurrentExpandShrink") + @GroupThreads(4) + public void concurrentExpandShrinkWriter(ChurningMapState map, WriterState w, Blackhole bh) { + long k1 = w.nextKey(); + long k2 = w.nextKey(); + bh.consume(map.put(k1, "v")); + bh.consume(map.put(k2, "v")); + bh.consume(map.remove(k1)); + bh.consume(map.remove(k2)); + } + + /** Reader chasing the writers; reads must not throw or return torn values. */ + @Benchmark + @Group("concurrentExpandShrink") + @GroupThreads(4) + public void concurrentExpandShrinkReader(ChurningMapState map, WriterState w, Blackhole bh) { + // Read keys in the same partition as some writer; values are either "v" or null (or the + // special sentinels would surface as a class cast / NPE on a buggy impl). + bh.consume(map.get(w.nextKey())); + } + + // ------------------------------------------------------------------------------------------ + // Boxing-impact workload + // + // BookKeeper's actual usage of ConcurrentLongHashMap stores object values (CachedFileInfo, + // byte[] master keys, LedgerDescriptor, LedgerData, ...). The choice to keep a primitive-long + // map instead of switching to ConcurrentHashMap hinges on whether the long->Long + // autobox on every put/get/remove materially hurts throughput and GC pressure in practice. + // + // To make that visible to JMH we have to defeat the JDK's Long.valueOf cache (which short- + // circuits values in [-128, 127] to a shared instance). Keys here are seeded above the cache + // range and monotonically increase, so every CHM operation has to allocate a fresh boxed + // Long; the primitive-long map sees zero allocation on the key path. Run with `-prof gc` to + // see the alloc-rate divergence on top of the throughput numbers. + // ------------------------------------------------------------------------------------------ + + /** + * Steady-state map prepopulated with keys above the Long.valueOf cache range. Each operation + * exercises a hit (key already in the map), so the put/get/remove cost is measured against a + * non-trivial table without rehash noise. + */ + @State(Scope.Benchmark) + public static class BoxingMapState { + @Param({"clhm", "chm"}) + private String implementation; + + @Param({"1024", "65536"}) + private int entries; + + // Lock granularity. Override with `-p concurrency=...` to match CHM's bucket-level + // striping (default JDK CHM uses one synchronized monitor per bucket, so concurrency=1024 + // on a 1024-bucket map effectively gives the primitive map a comparable lock count). + @Param({"16"}) + private int concurrency; + + private long[] presentKeys; + private ConcurrentLongHashMap clhm; + private ConcurrentHashMap chm; + + @Setup(Level.Trial) + public void setup() { + presentKeys = new long[entries]; + int cl = Math.min(concurrency, entries); // builder requires expectedItems >= concurrencyLevel + clhm = ConcurrentLongHashMap.newBuilder() + .expectedItems(entries).concurrencyLevel(cl).build(); + chm = new ConcurrentHashMap<>(entries, 0.66f, cl); + + // Start the key space well above 127 so Long.valueOf cannot serve from its cache. + // Use an odd stride so consecutive keys land in different sections / cache lines. + final long base = 1L << 32; + for (int i = 0; i < entries; i++) { + long key = base + ((long) i) * 31L; + presentKeys[i] = key; + clhm.put(key, "value"); + chm.put(key, "value"); + } + } + + String get(long key) { + return "clhm".equals(implementation) ? clhm.get(key) : chm.get(key); + } + + String put(long key, String value) { + return "clhm".equals(implementation) ? clhm.put(key, value) : chm.put(key, value); + } + + String remove(long key) { + return "clhm".equals(implementation) ? clhm.remove(key) : chm.remove(key); + } + } + + /** Per-thread cursor that walks the prepopulated key array. */ + @State(Scope.Thread) + public static class BoxingCursor { + private int index; + + int next(int length) { + int v = index; + index = v + 1; + return v & (length - 1); + } + } + + /** + * Hot-path get on a key the JVM cannot serve from {@code Long.valueOf}'s cache. For the + * primitive-long maps this is a plain hash lookup; for CHM the caller-side autobox forces a + * fresh Long allocation per call. + */ + @Benchmark + public void boxingGetHit(BoxingMapState map, BoxingCursor cur, Blackhole bh) { + bh.consume(map.get(map.presentKeys[cur.next(map.presentKeys.length)])); + } + + /** Same as boxingGetHit but at 16-thread concurrency to amplify allocator-side contention. */ + @Benchmark + @Threads(16) + public void boxingConcurrentGetHit(BoxingMapState map, BoxingCursor cur, Blackhole bh) { + bh.consume(map.get(map.presentKeys[cur.next(map.presentKeys.length)])); + } + + /** + * Full put + get + remove cycle on cache-busting keys. CHM allocates 3 fresh Long boxes per + * iteration (one per call); the primitive-long maps allocate zero. {@code -prof gc} prints + * the alloc-rate delta directly. + */ + @Benchmark + public void boxingPutGetRemove(BoxingMapState map, BoxingCursor cur, Blackhole bh) { + long key = map.presentKeys[cur.next(map.presentKeys.length)]; + // Overwrite + re-read + remove + re-put so the steady-state size stays constant. + bh.consume(map.put(key, "value")); + bh.consume(map.get(key)); + bh.consume(map.remove(key)); + bh.consume(map.put(key, "value")); + } + + /** + * Concurrent put-then-remove churn: 16 threads independently allocate fresh keys (above the + * Long cache range) and immediately remove them. This is the workload most analogous to + * BookKeeper's hot ledger-descriptor / file-info eviction path. Run with {@code -prof gc} to + * measure the autobox allocation cost CHM pays here. + */ + @State(Scope.Thread) + public static class BoxingWriterState { + private static final AtomicLong NEXT_PARTITION = new AtomicLong(); + private long base; + private long offset; + + @Setup(Level.Iteration) + public void setup() { + // Each thread carves out a unique partition far above the Long cache range. + base = (1L << 40) | (NEXT_PARTITION.getAndIncrement() << 32); + offset = 0; + } + + long nextKey() { + // Stride of 31 keeps consecutive ops in different cache lines / sections. + return base + (offset++) * 31L; + } + } + + @Benchmark + @Threads(16) + public void boxingConcurrentPutRemove(BoxingMapState map, BoxingWriterState w, Blackhole bh) { + long key = w.nextKey(); + bh.consume(map.put(key, "value")); + bh.consume(map.remove(key)); + } + + // Reference imports kept reachable so static-analysis doesn't complain about unused symbols + // when we later extend this benchmark with predicate-based remove benchmarks. + @SuppressWarnings("unused") + private static final LongObjectPredicate KEEP_ALL = (k, v) -> false; +} From 21f33fbf82799abc58ef74fd6db54aae87b8b425 Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Fri, 1 May 2026 17:56:46 +0800 Subject: [PATCH 2/5] fix ConcurrentLongHashMap throw ArrayIndexOutOfBoundsException When concurrent read write access the map, The key array and value array are not publish at the same time when shrink or expand. This fix encapsulate the key and value in the same field to avoid this happen --- .../ConcurrentLongHashMapTest.java | 83 +++++++++++++++++++ 1 file changed, 83 insertions(+) diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMapTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMapTest.java index 7add2f1eb61..e6b4217bea0 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMapTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMapTest.java @@ -431,6 +431,89 @@ public void testCorrectnessAgainstConcurrentHashMap() throws Throwable { assertEquals(reference.size(), observed.get()); } + /** + * Cross-thread put-publish-then-read invariant: once a {@code put(k, v)} has returned and the + * writer has published k via a volatile counter, EVERY reader that observes that counter must + * see a non-null value for k. A failure here would mean a successful put was "lost" by the + * map's get path — the failure mode the Table-snapshot design exists to prevent. + * + *

The map starts at the smallest legal capacity with autoShrink enabled, so the rehash + * code path is exercised on virtually every put. This is the most aggressive workload for + * the rehash-vs-get race that the previous separate-volatile-arrays design couldn't survive. + */ + @Test + public void testNoLostGetAfterPublish() throws Throwable { + ConcurrentLongHashMap map = ConcurrentLongHashMap.newBuilder() + .expectedItems(2) + .concurrencyLevel(1) + .autoShrink(true) + .mapIdleFactor(0.25f) + .build(); + + final int totalKeys = 50_000; + final int readerThreads = 8; + + AtomicLong highestPublished = new AtomicLong(-1); + AtomicReference ex = new AtomicReference<>(); + + ExecutorService executor = Executors.newCachedThreadPool(); + CyclicBarrier barrier = new CyclicBarrier(readerThreads + 1); + List> futures = new ArrayList<>(); + + try { + // Writer: put then publish. The volatile-set on highestPublished establishes + // happens-before with any reader that observes the published value. + futures.add(executor.submit(() -> { + barrier.await(); + for (int i = 0; i < totalKeys; i++) { + assertNull(map.put(i, "v" + i)); + highestPublished.set(i); + } + return null; + })); + + // Readers: observe the published counter, then verify every key in [0, counter] is + // present with the expected value. The reader pulls the counter once per cycle and + // catches up to it before pulling again. + for (int r = 0; r < readerThreads; r++) { + futures.add(executor.submit(() -> { + barrier.await(); + try { + long lastChecked = -1; + while (lastChecked < totalKeys - 1) { + long target = highestPublished.get(); + while (lastChecked < target) { + lastChecked++; + String v = map.get(lastChecked); + if (v == null) { + throw new AssertionError( + "lost get for key " + lastChecked + + "; highestPublished=" + target); + } + if (!v.equals("v" + lastChecked)) { + throw new AssertionError( + "wrong value for key " + lastChecked + ": " + v); + } + } + } + } catch (Throwable t) { + ex.compareAndSet(null, t); + } + return null; + })); + } + + for (Future f : futures) { + f.get(120, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + } + + assertNull(ex.get()); + assertEquals(totalKeys, map.size()); + } + /** * forEach during concurrent writes is documented as not strongly thread-safe, but it must * never throw, never expose {@code DeletedValue}/{@code EmptyValue} sentinels, and every From 1f2d31116afa6c4e07ad196df5f1a732f5aa12cd Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Fri, 1 May 2026 17:57:54 +0800 Subject: [PATCH 3/5] fix ConcurrentLongHashMap throw ArrayIndexOutOfBoundsException When concurrent read write access the map, The key array and value array are not publish at the same time when shrink or expand. This fix encapsulate the key and value in the same field to avoid this happen --- .../util/collections/package-info.java | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) create mode 100644 microbenchmarks/src/main/java/org/apache/bookkeeper/util/collections/package-info.java diff --git a/microbenchmarks/src/main/java/org/apache/bookkeeper/util/collections/package-info.java b/microbenchmarks/src/main/java/org/apache/bookkeeper/util/collections/package-info.java new file mode 100644 index 00000000000..fa9f73f4b37 --- /dev/null +++ b/microbenchmarks/src/main/java/org/apache/bookkeeper/util/collections/package-info.java @@ -0,0 +1,19 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.bookkeeper.util.collections; \ No newline at end of file From 97af802fbd84ba75a46b3e2901b089a25f47330e Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Fri, 1 May 2026 18:53:55 +0800 Subject: [PATCH 4/5] fix ConcurrentLongHashMap throw ArrayIndexOutOfBoundsException When concurrent read write access the map, The key array and value array are not publish at the same time when shrink or expand. This fix encapsulate the key and value in the same field to avoid this happen --- .../bookkeeper/util/collections/ConcurrentLongHashMap.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMap.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMap.java index c664c324288..ba44631a0aa 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMap.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMap.java @@ -24,6 +24,7 @@ import static com.google.common.base.Preconditions.checkNotNull; import com.google.common.collect.Lists; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.Arrays; import java.util.List; import java.util.concurrent.locks.StampedLock; @@ -328,6 +329,8 @@ private static final class Table { } } + // Section is Serializable only by inheritance from StampedLock; never actually serialized. + @SuppressFBWarnings("SE_BAD_FIELD") private volatile Table table; private volatile int capacity; From bdc191eec2e7674e5f53ca2d59a7ce8d3919a81f Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Sat, 2 May 2026 14:45:10 +0800 Subject: [PATCH 5/5] fix ConcurrentLongHashMap throw ArrayIndexOutOfBoundsException When concurrent read write access the map, The key array and value array are not publish at the same time when shrink or expand. This fix encapsulate the key and value in the same field to avoid this happen --- .../collections/ConcurrentLongHashMap.java | 96 ++++++++----------- 1 file changed, 42 insertions(+), 54 deletions(-) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMap.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMap.java index ba44631a0aa..2b716664806 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMap.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongHashMap.java @@ -198,7 +198,7 @@ long getUsedBucketCount() { public long capacity() { long capacity = 0; for (Section s : sections) { - capacity += s.capacity; + capacity += s.table.capacity(); } return capacity; } @@ -317,23 +317,12 @@ public interface EntryProcessor { // previous design had to paper over with Math.min(keys.length, values.length). @SuppressWarnings("serial") private static final class Section extends StampedLock { - private static final class Table { - final long[] keys; - final V[] values; - final int capacity; - - Table(long[] keys, V[] values, int capacity) { - this.keys = keys; - this.values = values; - this.capacity = capacity; - } - } + private record Table(long[] keys, V[] values, int capacity) { } // Section is Serializable only by inheritance from StampedLock; never actually serialized. @SuppressFBWarnings("SE_BAD_FIELD") private volatile Table table; - private volatile int capacity; private final int initCapacity; private volatile int size; private int usedBuckets; @@ -347,9 +336,9 @@ private static final class Table { Section(int capacity, float mapFillFactor, float mapIdleFactor, boolean autoShrink, float expandFactor, float shrinkFactor) { - this.capacity = alignToPowerOfTwo(capacity); - this.initCapacity = this.capacity; - this.table = new Table<>(new long[this.capacity], (V[]) new Object[this.capacity], this.capacity); + int initial = alignToPowerOfTwo(capacity); + this.initCapacity = initial; + this.table = new Table<>(new long[initial], (V[]) new Object[initial], initial); this.size = 0; this.usedBuckets = 0; this.autoShrink = autoShrink; @@ -357,8 +346,8 @@ private static final class Table { this.mapIdleFactor = mapIdleFactor; this.expandFactor = expandFactor; this.shrinkFactor = shrinkFactor; - this.resizeThresholdUp = (int) (this.capacity * mapFillFactor); - this.resizeThresholdBelow = (int) (this.capacity * mapIdleFactor); + this.resizeThresholdUp = (int) (initial * mapFillFactor); + this.resizeThresholdBelow = (int) (initial * mapIdleFactor); } V get(long key, int keyHash) { @@ -367,9 +356,9 @@ V get(long key, int keyHash) { boolean acquiredLock = false; Table table = this.table; - long[] keys = table.keys; - V[] values = table.values; - int bucket = signSafeMod(keyHash, table.capacity); + long[] keys = table.keys(); + V[] values = table.values(); + int bucket = signSafeMod(keyHash, table.capacity()); try { while (true) { @@ -393,9 +382,9 @@ V get(long key, int keyHash) { acquiredLock = true; table = this.table; - keys = table.keys; - values = table.values; - bucket = signSafeMod(keyHash, table.capacity); + keys = table.keys(); + values = table.values(); + bucket = signSafeMod(keyHash, table.capacity()); storedKey = keys[bucket]; storedValue = values[bucket]; } @@ -408,7 +397,7 @@ V get(long key, int keyHash) { } } - bucket = (bucket + 1) & (table.capacity - 1); + bucket = (bucket + 1) & (table.capacity() - 1); } } finally { if (acquiredLock) { @@ -422,9 +411,9 @@ V put(long key, V value, int keyHash, boolean onlyIfAbsent, LongFunction valu long stamp = writeLock(); Table table = this.table; - long[] keys = table.keys; - V[] values = table.values; - int capacity = table.capacity; + long[] keys = table.keys(); + V[] values = table.values(); + int capacity = table.capacity(); // Remember where we find the first available spot int firstDeletedKey = -1; @@ -493,8 +482,8 @@ private void cleanDeletedStatus(int startBucket) { // Cleanup all the buckets that were in `DeletedValue` state, // so that we can reduce unnecessary expansions Table table = this.table; - V[] values = table.values; - int capacity = table.capacity; + V[] values = table.values(); + int capacity = table.capacity(); int lastBucket = signSafeMod(startBucket - 1, capacity); while (values[lastBucket] == DeletedValue) { values[lastBucket] = (V) EmptyValue; @@ -508,9 +497,9 @@ private V remove(long key, Object value, int keyHash) { int bucket = keyHash; long stamp = writeLock(); Table table = this.table; - long[] keys = table.keys; - V[] values = table.values; - int capacity = table.capacity; + long[] keys = table.keys(); + V[] values = table.values(); + int capacity = table.capacity(); try { while (true) { @@ -576,9 +565,9 @@ int removeIf(LongObjectPredicate filter) { try { // Go through all the buckets for this section Table table = this.table; - long[] keys = table.keys; - V[] values = table.values; - int capacity = table.capacity; + long[] keys = table.keys(); + V[] values = table.values(); + int capacity = table.capacity(); for (int bucket = 0; size > 0 && bucket < capacity; bucket++) { long storedKey = keys[bucket]; V storedValue = values[bucket]; @@ -610,6 +599,7 @@ int removeIf(LongObjectPredicate filter) { // so as to avoid frequent shrinking and expansion near initCapacity, // frequent shrinking and expansion, // additionally opened arrays will consume more memory and affect GC + int capacity = this.table.capacity(); int newCapacity = Math.max(alignToPowerOfTwo((int) (capacity / shrinkFactor)), initCapacity); int newResizeThresholdUp = (int) (newCapacity * mapFillFactor); if (newCapacity < capacity && newResizeThresholdUp > size) { @@ -629,12 +619,12 @@ void clear() { long stamp = writeLock(); try { - if (autoShrink && capacity > initCapacity) { + Table table = this.table; + if (autoShrink && table.capacity() > initCapacity) { shrinkToInitCapacity(); } else { - Table table = this.table; - Arrays.fill(table.keys, 0); - Arrays.fill(table.values, EmptyValue); + Arrays.fill(table.keys(), 0); + Arrays.fill(table.values(), EmptyValue); this.size = 0; this.usedBuckets = 0; } @@ -647,9 +637,9 @@ public void forEach(EntryProcessor processor) { long stamp = tryOptimisticRead(); Table table = this.table; - int capacity = table.capacity; - long[] keys = table.keys; - V[] values = table.values; + int capacity = table.capacity(); + long[] keys = table.keys(); + V[] values = table.values(); boolean acquiredReadLock = false; @@ -662,9 +652,9 @@ public void forEach(EntryProcessor processor) { acquiredReadLock = true; table = this.table; - capacity = table.capacity; - keys = table.keys; - values = table.values; + capacity = table.capacity(); + keys = table.keys(); + values = table.values(); } // Go through all the buckets for this section @@ -697,8 +687,8 @@ private void rehash(int newCapacity) { long[] newKeys = new long[newCapacity]; V[] newValues = (V[]) new Object[newCapacity]; Table table = this.table; - long[] keys = table.keys; - V[] values = table.values; + long[] keys = table.keys(); + V[] values = table.values(); // Re-hash table for (int i = 0; i < keys.length; i++) { @@ -711,9 +701,8 @@ private void rehash(int newCapacity) { this.table = new Table<>(newKeys, newValues, newCapacity); usedBuckets = size; - capacity = newCapacity; - resizeThresholdUp = (int) (capacity * mapFillFactor); - resizeThresholdBelow = (int) (capacity * mapIdleFactor); + resizeThresholdUp = (int) (newCapacity * mapFillFactor); + resizeThresholdBelow = (int) (newCapacity * mapIdleFactor); } private void shrinkToInitCapacity() { @@ -723,9 +712,8 @@ private void shrinkToInitCapacity() { table = new Table<>(newKeys, newValues, initCapacity); size = 0; usedBuckets = 0; - capacity = initCapacity; - resizeThresholdUp = (int) (capacity * mapFillFactor); - resizeThresholdBelow = (int) (capacity * mapIdleFactor); + resizeThresholdUp = (int) (initCapacity * mapFillFactor); + resizeThresholdBelow = (int) (initCapacity * mapIdleFactor); } private static void insertKeyValueNoLock(long[] keys, V[] values, long key, V value) {