Optimize memory：Support shrinking in ConcurrentLongLongPairHashMap

bookkeeper-server/src/main/java/org/apache/bookkeeper/util/collections/ConcurrentLongLongPairHashMap.java

@@ -48,6 +48,7 @@ public class ConcurrentLongLongPairHashMap {
     private static final long ValueNotFound = -1L;
 
     private static final float MapFillFactor = 0.66f;
+    private static final float MapIdleFactor = 0.25f;
 
     private static final int DefaultExpectedItems = 256;
     private static final int DefaultConcurrencyLevel = 16;
@@ -228,14 +229,16 @@ private static final class Section extends StampedLock {
         private volatile int capacity;
         private volatile int size;
         private int usedBuckets;
-        private int resizeThreshold;
+        private int resizeThresholdUp;
+        private int resizeThresholdBelow;
 
         Section(int capacity) {
             this.capacity = alignToPowerOfTwo(capacity);
             this.table = new long[4 * this.capacity];
             this.size = 0;
             this.usedBuckets = 0;
-            this.resizeThreshold = (int) (this.capacity * MapFillFactor);
+            this.resizeThresholdUp = (int) (this.capacity * MapFillFactor);
+            this.resizeThresholdBelow = (int) (this.capacity * MapIdleFactor);
             Arrays.fill(table, EmptyKey);
         }
 
@@ -336,9 +339,10 @@ boolean put(long key1, long key2, long value1, long value2, int keyHash, boolean
                     bucket = (bucket + 4) & (table.length - 1);
                 }
             } finally {
-                if (usedBuckets > resizeThreshold) {
+                if (usedBuckets > resizeThresholdUp) {
                     try {
-                        rehash();
+                        // Expand the hashmap
+                        rehash(capacity * 2);
                     } finally {
                         unlockWrite(stamp);
                     }
@@ -376,7 +380,16 @@ private boolean remove(long key1, long key2, long value1, long value2, int keyHa
                 }
 
             } finally {
-                unlockWrite(stamp);
+                if (size < resizeThresholdBelow) {
+                    try {
+                        // shrink the hashmap
+                        rehash(capacity / 2);
+                    } finally {
+                        unlockWrite(stamp);
+                    }
+                } else {
+                    unlockWrite(stamp);
+                }
             }
         }
 
@@ -388,6 +401,18 @@ private void cleanBucket(int bucket) {
                 table[bucket + 2] = ValueNotFound;
                 table[bucket + 3] = ValueNotFound;
                 --usedBuckets;
+
+                // Reduce unnecessary rehash
+                bucket = (bucket - 4) & (table.length - 1);
+                while (table[bucket] == DeletedKey) {
+                    table[bucket] = EmptyKey;
+                    table[bucket + 1] = EmptyKey;
+                    table[bucket + 2] = ValueNotFound;
+                    table[bucket + 3] = ValueNotFound;
+                    --usedBuckets;
+
+                    bucket = (bucket - 4) & (table.length - 1);
+                }
             } else {
                 table[bucket] = DeletedKey;
                 table[bucket + 1] = DeletedKey;
@@ -453,9 +478,7 @@ public void forEach(BiConsumerLongPair processor) {
             }
         }
 
-        private void rehash() {
-            // Expand the hashmap
-            int newCapacity = capacity * 2;
+        private void rehash(int newCapacity) {
             long[] newTable = new long[4 * newCapacity];
             Arrays.fill(newTable, EmptyKey);
 
@@ -475,7 +498,8 @@ private void rehash() {
             // 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;
-            resizeThreshold = (int) (capacity * MapFillFactor);
+            resizeThresholdUp = (int) (capacity * MapFillFactor);
+            resizeThresholdBelow = (int) (capacity * MapIdleFactor);
         }
 
         private static void insertKeyValueNoLock(long[] table, int capacity, long key1, long key2, long value1,

bookkeeper-server/src/test/java/org/apache/bookkeeper/util/collections/ConcurrentLongLongPairHashMapTest.java

@@ -113,6 +113,22 @@ public void testRemove() {
         assertTrue(map.isEmpty());
     }
 
+    @Test
+    public void testExpandAndShrink() {
+        ConcurrentLongLongPairHashMap map = new ConcurrentLongLongPairHashMap(2, 1);
+        assertTrue(map.put(1, 1, 11, 11));
+        assertTrue(map.put(2, 2, 22, 22));
+        assertTrue(map.put(3, 3, 33, 33));
+
+        // expand hashmap
+        assertTrue(map.capacity() == 8);
+
+        assertTrue(map.remove(1, 1, 11, 11));
+        assertFalse(map.remove(2, 2, 22, 22));
+        // shrink hashmap
+        assertTrue(map.capacity() == 4);
+    }
+
     @Test
     public void testNegativeUsedBucketCount() {
         ConcurrentLongLongPairHashMap map = new ConcurrentLongLongPairHashMap(16, 1);