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 d70fa1d996bbc..4bcae075a55f8 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-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 995a1405a2dfb..f52d7e60b45a8 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,10 +16,19 @@ */ package org.apache.camel.support; +import java.util.Collections; 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; @@ -138,11 +147,28 @@ public Map createLRUWeakCache(int initialCapacity, int maximumCache 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) { @@ -159,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 + 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))); + } + } + @Override - protected boolean removeEldestEntry(Map.Entry eldest) { - if (size() > maximumCacheSize) { - V value = eldest.getValue(); - evict.accept(value); - return true; + 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); + }); } - return false; } + @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()) { + 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) {