From dc1e4b142e0503634c1f5aa6674e56d68be5d600 Mon Sep 17 00:00:00 2001 From: Claus Ibsen Date: Fri, 19 May 2023 11:13:28 +0200 Subject: [PATCH 1/3] CAMEL-19295: Experiment with sync LRUCache and test case that otherwise will OOME --- .../impl/DefaultEndpointRegistryTest.java | 54 +++++++++++++++++++ .../camel/support/DefaultLRUCacheFactory.java | 21 ++++---- 2 files changed, 65 insertions(+), 10 deletions(-) diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultEndpointRegistryTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultEndpointRegistryTest.java index 2d8cdc8e9864f..9ac3db65b5ed7 100644 --- a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultEndpointRegistryTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultEndpointRegistryTest.java @@ -16,9 +16,17 @@ */ package org.apache.camel.impl; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import org.apache.camel.ProducerTemplate; +import org.apache.camel.ServiceStatus; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.impl.engine.DefaultEndpointRegistry; +import org.apache.camel.impl.engine.SimpleCamelContext; import org.apache.camel.spi.EndpointRegistry; +import org.apache.camel.support.NormalizedUri; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -68,4 +76,50 @@ public void configure() throws Exception { assertTrue(reg.isStatic("file:error")); } + //Testing the issue https://issues.apache.org/jira/browse/CAMEL-19295 + @Test + public void testConcurrency() throws InterruptedException { + + SimpleCamelContext context = new SimpleCamelContext(); + context.start(); + + ProducerTemplate producerTemplate = context.createProducerTemplate(); + EndpointRegistry endpointRegistry = context.getEndpointRegistry(); + + int nThreads = 4; + ExecutorService executorService = Executors.newFixedThreadPool(nThreads); + int iterations = 500; + + for (int j = 0; j < iterations; j++) { + CountDownLatch allThreadCompletionSemaphore = new CountDownLatch(nThreads); + for (int i = 0; i < nThreads; i++) { + + executorService.submit(() -> { + + producerTemplate.requestBody("controlbus:route?routeId=route1&action=ACTION_STATUS&loggingLevel=off", null, + ServiceStatus.class); + producerTemplate.requestBody("controlbus:route?routeId=route2&action=ACTION_STATUS&loggingLevel=off", null, + ServiceStatus.class); + producerTemplate.requestBody("controlbus:route?routeId=route3&action=ACTION_STATUS&loggingLevel=off", null, + ServiceStatus.class); + producerTemplate.requestBody("controlbus:route?routeId=route4&action=ACTION_STATUS&loggingLevel=off", null, + ServiceStatus.class); + producerTemplate.requestBody("controlbus:route?routeId=route5&action=ACTION_STATUS&loggingLevel=off", null, + ServiceStatus.class); + + allThreadCompletionSemaphore.countDown(); + + }); + } + + allThreadCompletionSemaphore.await(); + + assertTrue(endpointRegistry.values().toArray() != null); + + } + + executorService.shutdown(); + + } + } diff --git a/core/camel-support/src/main/java/org/apache/camel/support/DefaultLRUCacheFactory.java b/core/camel-support/src/main/java/org/apache/camel/support/DefaultLRUCacheFactory.java index 995a1405a2dfb..eefc3acbe7a03 100644 --- a/core/camel-support/src/main/java/org/apache/camel/support/DefaultLRUCacheFactory.java +++ b/core/camel-support/src/main/java/org/apache/camel/support/DefaultLRUCacheFactory.java @@ -16,6 +16,7 @@ */ package org.apache.camel.support; +import java.util.Collections; import java.util.LinkedHashMap; import java.util.Map; import java.util.Objects; @@ -41,7 +42,7 @@ public class DefaultLRUCacheFactory extends LRUCacheFactory { @Override public Map createLRUCache(int maximumCacheSize) { LOG.trace("Creating LRUCache with maximumCacheSize: {}", maximumCacheSize); - return new SimpleLRUCache<>(maximumCacheSize); + return Collections.synchronizedMap(new SimpleLRUCache<>(maximumCacheSize)); } /** @@ -53,7 +54,7 @@ public Map createLRUCache(int maximumCacheSize) { @Override public Map createLRUCache(int maximumCacheSize, Consumer onEvict) { LOG.trace("Creating LRUCache with maximumCacheSize: {}", maximumCacheSize); - return new SimpleLRUCache<>(16, maximumCacheSize, onEvict); + return Collections.synchronizedMap(new SimpleLRUCache<>(16, maximumCacheSize, onEvict)); } /** @@ -67,7 +68,7 @@ public Map createLRUCache(int maximumCacheSize, Consumer onEvict @Override public Map createLRUCache(int initialCapacity, int maximumCacheSize) { LOG.trace("Creating LRUCache with initialCapacity: {}, maximumCacheSize: {}", initialCapacity, maximumCacheSize); - return new SimpleLRUCache<>(initialCapacity, maximumCacheSize); + return Collections.synchronizedMap(new SimpleLRUCache<>(initialCapacity, maximumCacheSize)); } /** @@ -83,7 +84,7 @@ public Map createLRUCache(int initialCapacity, int maximumCacheSize public Map createLRUCache(int initialCapacity, int maximumCacheSize, boolean stopOnEviction) { LOG.trace("Creating LRUCache with initialCapacity: {}, maximumCacheSize: {}, stopOnEviction: {}", initialCapacity, maximumCacheSize, stopOnEviction); - return new SimpleLRUCache<>(initialCapacity, maximumCacheSize, stopOnEviction); + return Collections.synchronizedMap(new SimpleLRUCache<>(initialCapacity, maximumCacheSize, stopOnEviction)); } /** @@ -96,20 +97,20 @@ public Map createLRUCache(int initialCapacity, int maximumCacheSize @Override public Map createLRUSoftCache(int maximumCacheSize) { LOG.trace("Creating LRUSoftCache with maximumCacheSize: {}", maximumCacheSize); - return new SimpleLRUCache<>(maximumCacheSize); + return Collections.synchronizedMap(new SimpleLRUCache<>(maximumCacheSize)); } @Override public Map createLRUSoftCache(int initialCapacity, int maximumCacheSize) { LOG.trace("Creating LRUCache with initialCapacity: {}, maximumCacheSize: {}", initialCapacity, maximumCacheSize); - return new SimpleLRUCache<>(initialCapacity, maximumCacheSize); + return Collections.synchronizedMap(new SimpleLRUCache<>(initialCapacity, maximumCacheSize)); } @Override public Map createLRUSoftCache(int initialCapacity, int maximumCacheSize, boolean stopOnEviction) { LOG.trace("Creating LRUCache with initialCapacity: {}, maximumCacheSize: {}, stopOnEviction: {}", initialCapacity, maximumCacheSize, stopOnEviction); - return new SimpleLRUCache<>(initialCapacity, maximumCacheSize, stopOnEviction); + return Collections.synchronizedMap(new SimpleLRUCache<>(initialCapacity, maximumCacheSize, stopOnEviction)); } /** @@ -122,20 +123,20 @@ public Map createLRUSoftCache(int initialCapacity, int maximumCache @Override public Map createLRUWeakCache(int maximumCacheSize) { LOG.trace("Creating LRUWeakCache with maximumCacheSize: {}", maximumCacheSize); - return new SimpleLRUCache<>(maximumCacheSize); + return Collections.synchronizedMap(new SimpleLRUCache<>(maximumCacheSize)); } @Override public Map createLRUWeakCache(int initialCapacity, int maximumCacheSize) { LOG.trace("Creating LRUCache with initialCapacity: {}, maximumCacheSize: {}", initialCapacity, maximumCacheSize); - return new SimpleLRUCache<>(initialCapacity, maximumCacheSize); + return Collections.synchronizedMap(new SimpleLRUCache<>(initialCapacity, maximumCacheSize)); } @Override public Map createLRUWeakCache(int initialCapacity, int maximumCacheSize, boolean stopOnEviction) { LOG.trace("Creating LRUCache with initialCapacity: {}, maximumCacheSize: {}, stopOnEviction: {}", initialCapacity, maximumCacheSize, stopOnEviction); - return new SimpleLRUCache<>(initialCapacity, maximumCacheSize, stopOnEviction); + return Collections.synchronizedMap(new SimpleLRUCache<>(initialCapacity, maximumCacheSize, stopOnEviction)); } private class SimpleLRUCache extends LinkedHashMap { From 734f27f3ae51e362903879a6cdefe8874791979b Mon Sep 17 00:00:00 2001 From: Nicolas Filotto Date: Tue, 23 May 2023 22:21:52 +0200 Subject: [PATCH 2/3] CAMEL-19295: Basic thread-safe LRU cache --- .../support/DefaultLRUCacheFactoryTest.java | 292 ++++++++++++++++++ .../camel/support/DefaultLRUCacheFactory.java | 272 ++++++++++++++-- 2 files changed, 545 insertions(+), 19 deletions(-) create mode 100644 core/camel-core/src/test/java/org/apache/camel/support/DefaultLRUCacheFactoryTest.java diff --git a/core/camel-core/src/test/java/org/apache/camel/support/DefaultLRUCacheFactoryTest.java b/core/camel-core/src/test/java/org/apache/camel/support/DefaultLRUCacheFactoryTest.java new file mode 100644 index 0000000000000..c9f7fdda9b421 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/support/DefaultLRUCacheFactoryTest.java @@ -0,0 +1,292 @@ +/* + * 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.camel.support; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * The test class for {@link DefaultLRUCacheFactory}. + */ +class DefaultLRUCacheFactoryTest { + + private final List consumed = new ArrayList<>(); + private final DefaultLRUCacheFactory.SimpleLRUCache map + = (DefaultLRUCacheFactory.SimpleLRUCache) new DefaultLRUCacheFactory(). createLRUCache(3, consumed::add); + + @Test + void forbiddenOperations() { + assertThrows(UnsupportedOperationException.class, () -> map.values().add("foo")); + assertThrows(UnsupportedOperationException.class, () -> map.keySet().add("foo")); + assertThrows(UnsupportedOperationException.class, () -> map.entrySet().add(Map.entry("x", "y"))); + } + + @Test + void setValue() { + assertNull(map.put("1", "One")); + assertEquals(1, map.size()); + assertEquals(1, map.getQueueSize()); + map.entrySet().iterator().next().setValue("bar"); + assertEquals(1, map.size()); + assertEquals(2, map.getQueueSize()); + } + + @Test + void queueSize() { + assertEquals(0, map.getQueueSize()); + map.put("1", "1"); + assertEquals(1, map.size()); + assertEquals(1, map.getQueueSize()); + map.put("1", "2"); + assertEquals(1, map.size()); + assertEquals(2, map.getQueueSize()); + map.put("1", "3"); + assertEquals(1, map.size()); + assertEquals(3, map.getQueueSize()); + map.put("1", "4"); + assertEquals(1, map.size()); + assertEquals(4, map.getQueueSize()); + map.put("1", "5"); + assertEquals(1, map.size()); + assertEquals(5, map.getQueueSize()); + map.put("1", "6"); + assertEquals(1, map.size()); + assertEquals(6, map.getQueueSize()); + map.put("1", "7"); + assertEquals(1, map.size()); + assertEquals(6, map.getQueueSize()); + map.put("1", "8"); + assertEquals(1, map.size()); + assertEquals(6, map.getQueueSize()); + } + + @Test + void put() { + assertEquals(0, map.size()); + assertNull(map.put("1", "One")); + assertEquals(1, map.size()); + assertNull(map.put("2", "Two")); + assertEquals(2, map.size()); + assertNull(map.put("3", "Three")); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertNull(map.put("4", "Four")); + assertEquals(3, map.size()); + assertEquals(1, consumed.size()); + assertFalse(map.containsKey("1")); + assertTrue(consumed.contains("One")); + assertEquals("Two", map.put("2", "Two v2")); + assertEquals(3, map.size()); + assertEquals(1, consumed.size()); + assertTrue(map.containsKey("2")); + assertEquals("Two v2", map.get("2")); + } + + @Test + void putIfAbsent() { + assertEquals(0, map.size()); + assertNull(map.putIfAbsent("1", "One")); + assertEquals(1, map.size()); + assertNull(map.putIfAbsent("2", "Two")); + assertEquals(2, map.size()); + assertNull(map.putIfAbsent("3", "Three")); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertNull(map.putIfAbsent("4", "Four")); + assertEquals(3, map.size()); + assertEquals(1, consumed.size()); + assertFalse(map.containsKey("1")); + assertTrue(consumed.contains("One")); + assertEquals("Two", map.putIfAbsent("2", "Two v2")); + assertEquals(3, map.size()); + assertEquals(1, consumed.size()); + assertTrue(map.containsKey("2")); + assertEquals("Two", map.get("2")); + assertNull(map.putIfAbsent("5", "Five")); + assertEquals(3, map.size()); + assertEquals(2, consumed.size()); + assertFalse(map.containsKey("2")); + assertTrue(consumed.contains("Two")); + } + + @Test + void computeIfAbsent() { + assertEquals(0, map.size()); + assertEquals("One", map.computeIfAbsent("1", k -> "One")); + assertEquals(1, map.size()); + assertEquals("Two", map.computeIfAbsent("2", k -> "Two")); + assertEquals(2, map.size()); + assertEquals("Three", map.computeIfAbsent("3", k -> "Three")); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertEquals("Four", map.computeIfAbsent("4", k -> "Four")); + assertEquals(3, map.size()); + assertEquals(1, consumed.size()); + assertFalse(map.containsKey("1")); + assertTrue(consumed.contains("One")); + assertNull(map.computeIfAbsent("1", k -> null)); + assertEquals(3, map.size()); + assertEquals(1, consumed.size()); + assertNull(map.computeIfAbsent("5", k -> null)); + assertEquals(3, map.size()); + assertEquals(1, consumed.size()); + assertEquals("Two", map.computeIfAbsent("2", k -> "Two v2")); + assertEquals(3, map.size()); + assertEquals(1, consumed.size()); + assertTrue(map.containsKey("2")); + assertEquals("Two", map.get("2")); + assertEquals("Five", map.computeIfAbsent("5", k -> "Five")); + assertEquals(3, map.size()); + assertEquals(2, consumed.size()); + assertFalse(map.containsKey("2")); + assertTrue(consumed.contains("Two")); + assertEquals("Five", map.computeIfAbsent("5", k -> null)); + assertEquals(3, map.size()); + assertEquals(2, consumed.size()); + } + + @Test + void computeIfPresent() { + assertEquals(0, map.size()); + map.putIfAbsent("1", "One"); + assertEquals(1, map.size()); + map.putIfAbsent("2", "Two"); + assertEquals(2, map.size()); + map.putIfAbsent("3", "Three"); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertNull(map.computeIfPresent("4", (k, v) -> "Four")); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertFalse(map.containsKey("4")); + assertEquals("One v2", map.computeIfPresent("1", (k, v) -> "One v2")); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertTrue(map.containsKey("1")); + assertEquals("One v2", map.get("1")); + assertNull(map.computeIfPresent("1", (k, v) -> null)); + assertEquals(2, map.size()); + assertEquals(0, consumed.size()); + assertFalse(map.containsKey("1")); + } + + @Test + void compute() { + assertEquals(0, map.size()); + assertEquals("One", map.compute("1", (k, v) -> "One")); + assertEquals(1, map.size()); + assertEquals("Two", map.compute("2", (k, v) -> "Two")); + assertEquals(2, map.size()); + assertEquals("Three", map.compute("3", (k, v) -> "Three")); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertEquals("Four", map.compute("4", (k, v) -> "Four")); + assertEquals(3, map.size()); + assertEquals(1, consumed.size()); + assertFalse(map.containsKey("1")); + assertTrue(consumed.contains("One")); + assertEquals("Two v2", map.compute("2", (k, v) -> "Two v2")); + assertEquals(3, map.size()); + assertEquals(1, consumed.size()); + assertTrue(map.containsKey("2")); + assertEquals("Two v2", map.get("2")); + assertNull(map.compute("2", (k, v) -> null)); + assertEquals(2, map.size()); + assertEquals(1, consumed.size()); + assertFalse(map.containsKey("2")); + } + + @Test + void merge() { + assertEquals(0, map.size()); + assertEquals("One", map.merge("1", "One", String::concat)); + assertEquals(1, map.size()); + assertEquals("Two", map.merge("2", "Two", String::concat)); + assertEquals(2, map.size()); + assertEquals("Three", map.merge("3", "Three", String::concat)); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertEquals("Four", map.merge("4", "Four", String::concat)); + assertEquals(3, map.size()); + assertEquals(1, consumed.size()); + assertFalse(map.containsKey("1")); + assertTrue(consumed.contains("One")); + assertEquals("TwoV2", map.merge("2", "V2", String::concat)); + assertEquals(3, map.size()); + assertEquals(1, consumed.size()); + assertNull(map.merge("2", "V2", (v1, v2) -> null)); + assertEquals(2, map.size()); + assertEquals(1, consumed.size()); + } + + @Test + void replace() { + assertEquals(0, map.size()); + assertNull(map.replace("1", "One")); + assertEquals(0, map.size()); + map.put("1", "One"); + assertEquals(1, map.size()); + map.put("2", "Two"); + assertEquals(2, map.size()); + map.put("3", "Three"); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertEquals("One", map.replace("1", "One v2")); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertEquals("Three", map.replace("3", "Three v2")); + assertEquals("Three v2", map.get("3")); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + } + + @Test + void replaceWithOldValue() { + assertEquals(0, map.size()); + map.put("1", "One"); + map.put("2", "Two"); + map.put("3", "Three"); + assertEquals(3, map.size()); + assertFalse(map.replace("1", "foo", "One")); + assertEquals(3, map.size()); + assertFalse(map.replace("1", "foo", "One v2")); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertEquals("One", map.get("1")); + assertTrue(map.replace("1", "One", "One v2")); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertEquals("One v2", map.get("1")); + assertFalse(map.replace("3", "foo", "Three v2")); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + assertTrue(map.replace("3", "Three", "Three v2")); + assertEquals("Three v2", map.get("3")); + assertEquals(3, map.size()); + assertEquals(0, consumed.size()); + } +} diff --git a/core/camel-support/src/main/java/org/apache/camel/support/DefaultLRUCacheFactory.java b/core/camel-support/src/main/java/org/apache/camel/support/DefaultLRUCacheFactory.java index eefc3acbe7a03..ff5d2ac0c6f64 100644 --- a/core/camel-support/src/main/java/org/apache/camel/support/DefaultLRUCacheFactory.java +++ b/core/camel-support/src/main/java/org/apache/camel/support/DefaultLRUCacheFactory.java @@ -20,7 +20,15 @@ import java.util.LinkedHashMap; import java.util.Map; import java.util.Objects; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.LongAdder; +import java.util.function.BiFunction; import java.util.function.Consumer; +import java.util.function.Function; import org.apache.camel.support.service.ServiceHelper; import org.slf4j.Logger; @@ -42,7 +50,7 @@ public class DefaultLRUCacheFactory extends LRUCacheFactory { @Override public Map createLRUCache(int maximumCacheSize) { LOG.trace("Creating LRUCache with maximumCacheSize: {}", maximumCacheSize); - return Collections.synchronizedMap(new SimpleLRUCache<>(maximumCacheSize)); + return new SimpleLRUCache<>(maximumCacheSize); } /** @@ -54,7 +62,7 @@ public Map createLRUCache(int maximumCacheSize) { @Override public Map createLRUCache(int maximumCacheSize, Consumer onEvict) { LOG.trace("Creating LRUCache with maximumCacheSize: {}", maximumCacheSize); - return Collections.synchronizedMap(new SimpleLRUCache<>(16, maximumCacheSize, onEvict)); + return new SimpleLRUCache<>(16, maximumCacheSize, onEvict); } /** @@ -68,7 +76,7 @@ public Map createLRUCache(int maximumCacheSize, Consumer onEvict @Override public Map createLRUCache(int initialCapacity, int maximumCacheSize) { LOG.trace("Creating LRUCache with initialCapacity: {}, maximumCacheSize: {}", initialCapacity, maximumCacheSize); - return Collections.synchronizedMap(new SimpleLRUCache<>(initialCapacity, maximumCacheSize)); + return new SimpleLRUCache<>(initialCapacity, maximumCacheSize); } /** @@ -84,7 +92,7 @@ public Map createLRUCache(int initialCapacity, int maximumCacheSize public Map createLRUCache(int initialCapacity, int maximumCacheSize, boolean stopOnEviction) { LOG.trace("Creating LRUCache with initialCapacity: {}, maximumCacheSize: {}, stopOnEviction: {}", initialCapacity, maximumCacheSize, stopOnEviction); - return Collections.synchronizedMap(new SimpleLRUCache<>(initialCapacity, maximumCacheSize, stopOnEviction)); + return new SimpleLRUCache<>(initialCapacity, maximumCacheSize, stopOnEviction); } /** @@ -97,20 +105,20 @@ public Map createLRUCache(int initialCapacity, int maximumCacheSize @Override public Map createLRUSoftCache(int maximumCacheSize) { LOG.trace("Creating LRUSoftCache with maximumCacheSize: {}", maximumCacheSize); - return Collections.synchronizedMap(new SimpleLRUCache<>(maximumCacheSize)); + return new SimpleLRUCache<>(maximumCacheSize); } @Override public Map createLRUSoftCache(int initialCapacity, int maximumCacheSize) { LOG.trace("Creating LRUCache with initialCapacity: {}, maximumCacheSize: {}", initialCapacity, maximumCacheSize); - return Collections.synchronizedMap(new SimpleLRUCache<>(initialCapacity, maximumCacheSize)); + return new SimpleLRUCache<>(initialCapacity, maximumCacheSize); } @Override public Map createLRUSoftCache(int initialCapacity, int maximumCacheSize, boolean stopOnEviction) { LOG.trace("Creating LRUCache with initialCapacity: {}, maximumCacheSize: {}, stopOnEviction: {}", initialCapacity, maximumCacheSize, stopOnEviction); - return Collections.synchronizedMap(new SimpleLRUCache<>(initialCapacity, maximumCacheSize, stopOnEviction)); + return new SimpleLRUCache<>(initialCapacity, maximumCacheSize, stopOnEviction); } /** @@ -123,27 +131,44 @@ public Map createLRUSoftCache(int initialCapacity, int maximumCache @Override public Map createLRUWeakCache(int maximumCacheSize) { LOG.trace("Creating LRUWeakCache with maximumCacheSize: {}", maximumCacheSize); - return Collections.synchronizedMap(new SimpleLRUCache<>(maximumCacheSize)); + return new SimpleLRUCache<>(maximumCacheSize); } @Override public Map createLRUWeakCache(int initialCapacity, int maximumCacheSize) { LOG.trace("Creating LRUCache with initialCapacity: {}, maximumCacheSize: {}", initialCapacity, maximumCacheSize); - return Collections.synchronizedMap(new SimpleLRUCache<>(initialCapacity, maximumCacheSize)); + return new SimpleLRUCache<>(initialCapacity, maximumCacheSize); } @Override public Map createLRUWeakCache(int initialCapacity, int maximumCacheSize, boolean stopOnEviction) { LOG.trace("Creating LRUCache with initialCapacity: {}, maximumCacheSize: {}, stopOnEviction: {}", initialCapacity, maximumCacheSize, stopOnEviction); - return Collections.synchronizedMap(new SimpleLRUCache<>(initialCapacity, maximumCacheSize, stopOnEviction)); + return new SimpleLRUCache<>(initialCapacity, maximumCacheSize, stopOnEviction); } - private class SimpleLRUCache extends LinkedHashMap { + class SimpleLRUCache extends ConcurrentHashMap { static final float DEFAULT_LOAD_FACTOR = 0.75f; - + /** + * The flag indicating that an eviction process is in progress. + */ + private final AtomicBoolean eviction = new AtomicBoolean(); + /** + * The maximum cache size. + */ private final int maximumCacheSize; + /** + * The last changes recorded. + */ + private final Queue> lastChanges = new ConcurrentLinkedQueue<>(); + /** + * The total amount of changes recorded. + */ + private final LongAdder totalChanges = new LongAdder(); + /** + * The function to call when an entry is evicted. + */ private final Consumer evict; public SimpleLRUCache(int maximumCacheSize) { @@ -160,21 +185,230 @@ public SimpleLRUCache(int initialCapacity, int maximumCacheSize, boolean stopOnE } public SimpleLRUCache(int initialCapacity, int maximumCacheSize, Consumer evicted) { - super(initialCapacity, DEFAULT_LOAD_FACTOR, true); + super(initialCapacity, DEFAULT_LOAD_FACTOR); this.maximumCacheSize = maximumCacheSize; this.evict = Objects.requireNonNull(evicted); } + /** + * Adds a new change in case the mapping function doesn't return {@code null}. + * + * @param context the context of the write operation + * @param mappingFunction the mapping function to apply. + * @return the result of the mapping function. + */ + private V addChange(OperationContext context, Function mappingFunction) { + K key = context.key; + V value = mappingFunction.apply(key); + if (value == null) { + return null; + } + lastChanges.add(Map.entry(key, value)); + totalChanges.increment(); + return value; + } + + @Override + public V put(K key, V value) { + if (key == null || value == null) { + throw new NullPointerException(); + } + try (OperationContext context = new OperationContext<>(this, key)) { + super.compute( + key, + (k, v) -> { + context.result = v; + return addChange(context, x -> value); + }); + return context.result; + } + } + + @Override + public V putIfAbsent(K key, V value) { + if (key == null || value == null) { + throw new NullPointerException(); + } + try (OperationContext context = new OperationContext<>(this, key)) { + super.compute( + key, + (k, v) -> { + context.result = v; + if (v != null) { + return v; + } + return addChange(context, x -> value); + }); + return context.result; + } + } + + @Override + public V computeIfAbsent(K key, Function mappingFunction) { + if (key == null || mappingFunction == null) { + throw new NullPointerException(); + } + try (OperationContext context = new OperationContext<>(this, key)) { + return super.computeIfAbsent(key, k -> addChange(context, mappingFunction)); + } + } + @Override - protected boolean removeEldestEntry(Map.Entry eldest) { - if (size() > maximumCacheSize) { - V value = eldest.getValue(); - evict.accept(value); - return true; + public V computeIfPresent(K key, BiFunction remappingFunction) { + if (key == null || remappingFunction == null) { + throw new NullPointerException(); + } + try (OperationContext context = new OperationContext<>(this, key)) { + return super.computeIfPresent(key, (k, v) -> addChange(context, x -> remappingFunction.apply(x, v))); } - return false; } + @Override + public V compute(K key, BiFunction remappingFunction) { + if (key == null || remappingFunction == null) { + throw new NullPointerException(); + } + try (OperationContext context = new OperationContext<>(this, key)) { + return super.compute(key, (k, v) -> addChange(context, x -> remappingFunction.apply(x, v))); + } + } + + @Override + public V merge(K key, V value, BiFunction remappingFunction) { + if (key == null || value == null || remappingFunction == null) { + throw new NullPointerException(); + } + try (OperationContext context = new OperationContext<>(this, key)) { + return super.compute( + key, + (k, oldValue) -> { + V newValue = (oldValue == null) ? value : remappingFunction.apply(oldValue, value); + return addChange(context, x -> newValue); + }); + } + } + + @Override + public boolean replace(K key, V oldValue, V newValue) { + if (key == null || oldValue == null || newValue == null) { + throw new NullPointerException(); + } + try (OperationContext context = new OperationContext<>(this, key)) { + super.computeIfPresent( + key, + (k, v) -> { + if (Objects.equals(oldValue, v)) { + context.result = addChange(context, x -> newValue); + return context.result; + } + return v; + }); + return context.result != null && Objects.equals(context.result, newValue); + } + } + + @Override + public V replace(K key, V value) { + if (key == null || value == null) { + throw new NullPointerException(); + } + try (OperationContext context = new OperationContext<>(this, key)) { + super.computeIfPresent( + key, + (k, v) -> { + context.result = v; + return addChange(context, x -> value); + }); + return context.result; + } + } + + @Override + public void putAll(Map m) { + for (Map.Entry e : m.entrySet()) { + put(e.getKey(), e.getValue()); + } + } + + @Override + public void replaceAll(BiFunction function) { + for (Map.Entry e : entrySet()) { + replace(e.getKey(), e.getValue(), function.apply(e.getKey(), e.getValue())); + } + } + + @Override + public Set> entrySet() { + return Collections.unmodifiableSet(super.entrySet()); + } + + /** + * @return the size of the queue of changes. + */ + int getQueueSize() { + return totalChanges.intValue(); + } + + /** + * Indicates whether an eviction is needed. An eviction can be triggered if the size of the map or the queue of + * changes exceeds the maximum allowed size which is respectively {@code maximumCacheSize} and + * {@code 2 * maximumCacheSize}. + * + * @return {@code true} if an eviction is needed, {@code false} otherwise. + */ + private boolean evictionNeeded() { + return size() > maximumCacheSize || getQueueSize() > 2 * maximumCacheSize; + } + + /** + * @return the oldest existing change. + */ + private Entry nextOldestChange() { + Entry oldest = lastChanges.poll(); + if (oldest != null) { + totalChanges.decrement(); + } + return oldest; + } + } + + /** + * The internal context of all write operations. + */ + private static class OperationContext implements AutoCloseable { + /** + * The result of the corresponding operation when applicable. + */ + V result; + /** + * The key against which the operation is made. + */ + final K key; + /** + * The underlying cache. + */ + private final SimpleLRUCache cache; + + OperationContext(SimpleLRUCache cache, K key) { + this.cache = cache; + this.key = key; + } + + @Override + public void close() { + if (cache.evictionNeeded() && cache.eviction.compareAndSet(false, true)) { + try { + while (cache.evictionNeeded()) { + Map.Entry oldest = cache.nextOldestChange(); + if (oldest != null && cache.remove(oldest.getKey(), oldest.getValue())) { + cache.evict.accept(oldest.getValue()); + } + } + } finally { + cache.eviction.set(false); + } + } + } } void doNothing(V value) { From 264b3c576442390578b0a593309bf7ba3e9444a4 Mon Sep 17 00:00:00 2001 From: Nicolas Filotto Date: Thu, 25 May 2023 11:25:23 +0200 Subject: [PATCH 3/3] CAMEL-19295: Fix regression in test DefaultProducerCacheTest --- .../java/org/apache/camel/impl/DefaultProducerCacheTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerCacheTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerCacheTest.java index 599d0a1bae5ce..9cdd6dd119e2d 100644 --- a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerCacheTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultProducerCacheTest.java @@ -189,8 +189,8 @@ public void testCacheEvictWhileInUse() throws Exception { // nothing has stopped yet even we have 3 producers and a cache limit of 2 assertEquals(0, stopCounter.get()); - // force evict p1 while its in use (eg simulate someone else grabbing it while evicting race condition) - cache.forceEvict(p1); + // force evict p2 while its in use (eg simulate someone else grabbing it while evicting race condition) + cache.forceEvict(p2); // and should still not be stopped assertEquals(0, stopCounter.get());