Guard Map.clear from a write-back removal listener (fixes #872)

If the removal listener re-inserts the entry during a clear, then the ConcurrentHashMap iterator will return the new entry. This can cause an infinite loop. Ideally the key is removed once, which allows for a write back to refresh the cache without double invalidations.
ben-manes · Feb 22, 2023 · 6f9e9b8 · 6f9e9b8
1 parent 354693b
commit 6f9e9b8
Show file tree

Hide file tree

Showing 6 changed files with 72 additions and 19 deletions.
diff --git a/caffeine/src/main/java/com/github/benmanes/caffeine/cache/BoundedLocalCache.java b/caffeine/src/main/java/com/github/benmanes/caffeine/cache/BoundedLocalCache.java
@@ -46,9 +46,11 @@
 import java.time.Duration;
 import java.util.AbstractCollection;
 import java.util.AbstractSet;
+import java.util.ArrayDeque;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.Deque;
 import java.util.HashMap;
 import java.util.IdentityHashMap;
 import java.util.Iterator;
@@ -2033,6 +2035,7 @@ public long estimatedSize() {
 
   @Override
   public void clear() {
+    Deque<Node<K, V>> entries;
     evictionLock.lock();
     try {
       // Discard all pending reads
@@ -2050,23 +2053,27 @@ public void clear() {
         pacer.cancel();
       }
 
-      // Discard all entries
-      int threshold = (WRITE_BUFFER_MAX / 2);
+      // Discard all entries, falling back to one-by-one to avoid excessive lock hold times
       long now = expirationTicker().read();
-      for (var node : data.values()) {
-        if (writeBuffer.size() >= threshold) {
-          // Fallback to one-by-one to avoid excessive lock hold times
-          break;
-        }
-        removeNode(node, now);
+      int threshold = (WRITE_BUFFER_MAX / 2);
+      entries = new ArrayDeque<>(data.values());
+      while (!entries.isEmpty() && (writeBuffer.size() < threshold)) {
+        removeNode(entries.poll(), now);
       }
     } finally {
       evictionLock.unlock();
     }
 
-    // Remove any stragglers, such as if released early to more aggressively flush incoming writes
-    for (Object key : keySet()) {
-      remove(key);
+    // Remove any stragglers if released early to more aggressively flush incoming writes
+    while (!entries.isEmpty()) {
+      var node = entries.poll();
+      var key = node.getKey();
+      if (key != null) {
+        remove(key);
+      }
+    }
+    if (collectKeys()) {
+      cleanUp();
     }
   }
 

diff --git a/caffeine/src/main/java/com/github/benmanes/caffeine/cache/UnboundedLocalCache.java b/caffeine/src/main/java/com/github/benmanes/caffeine/cache/UnboundedLocalCache.java
@@ -35,6 +35,7 @@
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -403,7 +404,7 @@ public void clear() {
       data.clear();
       return;
     }
-    for (K key : data.keySet()) {
+    for (K key : List.copyOf(data.keySet())) {
       remove(key);
     }
   }

diff --git a/caffeine/src/test/java/com/github/benmanes/caffeine/cache/BoundedLocalCacheTest.java b/caffeine/src/test/java/com/github/benmanes/caffeine/cache/BoundedLocalCacheTest.java
@@ -163,6 +163,29 @@ public void clear_pendingWrites(BoundedLocalCache<Int, Int> cache, CacheContext
     assertThat(cache.writeBuffer).isEmpty();
   }
 
+  @Test(dataProvider = "caches")
+  @CacheSpec(implementation = Implementation.Caffeine, loader = Loader.IDENTITY,
+      population = Population.FULL, removalListener = Listener.MOCKITO)
+  public void clear_pendingWrites_reload(BoundedLocalCache<Int, Int> cache, CacheContext context) {
+    var populate = new boolean[] { true };
+    Answer<?> fillWriteBuffer = invocation -> {
+      while (populate[0] && cache.writeBuffer.offer(() -> {})) {
+        // ignored
+      }
+      var loadingCache = (LoadingCache<?, ?>) context.cache();
+      loadingCache.refresh(invocation.getArgument(0));
+      populate[0] = false;
+      return null;
+    };
+    doAnswer(fillWriteBuffer)
+      .when(context.removalListener())
+      .onRemoval(any(), any(), any());
+
+    cache.clear();
+    assertThat(cache.writeBuffer).isEmpty();
+    assertThat(cache).hasSize(context.initialSize());
+  }
+
   /* --------------- Maintenance --------------- */
 
   @Test

diff --git a/caffeine/src/test/java/com/github/benmanes/caffeine/cache/CacheTest.java b/caffeine/src/test/java/com/github/benmanes/caffeine/cache/CacheTest.java
@@ -25,6 +25,8 @@
 import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth.assertWithMessage;
 import static java.util.function.Function.identity;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
 import static uk.org.lidalia.slf4jext.Level.ERROR;
 import static uk.org.lidalia.slf4jext.Level.WARN;
 
@@ -44,6 +46,7 @@
 import java.util.stream.IntStream;
 
 import org.apache.commons.lang3.tuple.Triple;
+import org.mockito.stubbing.Answer;
 import org.testng.annotations.Listeners;
 import org.testng.annotations.Test;
 
@@ -75,6 +78,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSetMultimap;
 import com.google.common.collect.Iterables;
+import com.google.common.collect.Maps;
 import com.google.common.primitives.Ints;
 import com.google.common.testing.EqualsTester;
 import com.google.common.testing.NullPointerTester;
@@ -809,12 +813,30 @@ public void invalidateAll_null(Cache<Int, Int> cache, CacheContext context) {
   @CacheSpec(population = Population.FULL, compute = Compute.SYNC,
       executorFailure = ExecutorFailure.IGNORED, executor = CacheExecutor.REJECTING,
       removalListener = Listener.CONSUMING)
-  public void removalListener_rejected(Cache<Int, Int> cache, CacheContext context) {
+  public void invalidateAll_removalListener_rejected(Cache<Int, Int> cache, CacheContext context) {
     cache.invalidateAll();
     assertThat(context).removalNotifications().withCause(EXPLICIT)
         .contains(context.original()).exclusively();
   }
 
+  @CheckNoStats
+  @Test(dataProvider = "caches")
+  @CacheSpec(removalListener = Listener.MOCKITO)
+  public void invalidateAll_removalListener_writeback(Cache<Int, Int> cache, CacheContext context) {
+    Answer<?> writeback = invocation -> {
+      cache.put(invocation.getArgument(0), invocation.getArgument(0));
+      return null;
+    };
+    doAnswer(writeback)
+        .when(context.removalListener())
+        .onRemoval(any(), any(), any());
+
+    cache.invalidateAll();
+    assertThat(context).removalNotifications().withCause(EXPLICIT)
+        .contains(context.original()).exclusively();
+    assertThat(cache).containsExactlyEntriesIn(Maps.asMap(context.original().keySet(), key -> key));
+  }
+
   /* --------------- cleanup --------------- */
 
   @CacheSpec

diff --git a/caffeine/src/test/java/com/github/benmanes/caffeine/cache/issues/Issue859Test.java b/caffeine/src/test/java/com/github/benmanes/caffeine/cache/issues/Issue859Test.java
@@ -45,7 +45,7 @@ public final class Issue859Test {
   private static final int NUMBER_OF_KEYS = 10;
 
   @Test
-  void scheduleIfPendingWrites() {
+  public void scheduleIfPendingWrites() {
     var runs = new ArrayList<TestRun>();
     for (int i = 1; i <= NUMBER_OF_RUNS; i++) {
       runs.add(runTest());

diff --git a/gradle/dependencies.gradle b/gradle/dependencies.gradle
@@ -99,7 +99,7 @@ ext {
     jacoco: '0.8.7',
     jmh: '0.6.8',
     jmhReport: '0.9.0',
-    nexusPublish: '1.1.0',
+    nexusPublish: '1.2.0',
     nullaway: '1.5.0',
     pmd: '6.54.0',
     semanticVersioning: '1.1.0',
@@ -108,7 +108,7 @@ ext {
     spotbugs: '4.7.3',
     spotbugsContrib: '7.4.7',
     spotbugsPlugin: '5.0.13',
-    versions: '0.45.0',
+    versions: '0.46.0',
   ]
   platformVersions = [
     asm: '9.4',
@@ -258,15 +258,15 @@ ext {
   ]
   restrictions = [
     'com.beust:jcommander': '1.82',
-    'com.fasterxml.jackson:jackson-bom': '2.14.1',
+    'com.fasterxml.jackson:jackson-bom': '2.14.2',
     'com.google.protobuf:protobuf-java': '3.21.8',
     'com.thoughtworks.xstream:xstream': '1.4.20',
     'net.sourceforge.nekohtml:nekohtml': '1.9.22',
     'org.apache.bcel:bcel': '6.6.1',
     'org.apache.commons:commons-text': '1.10.0',
-    'org.apache.httpcomponents:httpclient': '4.5.13',
+    'org.apache.httpcomponents:httpclient': '4.5.14',
     'org.bouncycastle:bcprov-jdk15on': '1.70',
-    'org.jsoup:jsoup': '1.15.3',
+    'org.jsoup:jsoup': '1.15.4',
     'org.yaml:snakeyaml': '1.33',
     'xerces:xercesImpl': '2.12.2',
   ]