diff --git a/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/ExperimentManagerFactory.java b/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/ExperimentManagerFactory.java index 8897f53..64e2ef3 100644 --- a/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/ExperimentManagerFactory.java +++ b/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/ExperimentManagerFactory.java @@ -26,9 +26,9 @@ public class ExperimentManagerFactory { private static final ExperimentManagerFactory INSTANCE = new ExperimentManagerFactory(); - private ExperimentManager experimentManager; - private ConfigurationProvider experimentConfigurationProvider; - private TreatmentAllocator treatmentAllocator; + private volatile ExperimentManager experimentManager; + private volatile ConfigurationProvider experimentConfigurationProvider; + private volatile TreatmentAllocator treatmentAllocator; public static ExperimentManagerFactory getInstance() { return INSTANCE; @@ -39,6 +39,9 @@ public ExperimentManager provideExperimentManager() { return experimentManager; } synchronized (this) { + if (experimentManager != null) { + return experimentManager; + } ConfigurationProvider provider = provideExperimentConfigurationProvider(); TreatmentAllocator allocator = provideTreatmentAllocator(); ExperimentHandler handler = new ExperimentHandler(); @@ -52,6 +55,9 @@ public ConfigurationProvider provideExperimentConfigura return experimentConfigurationProvider; } synchronized (this) { + if (experimentConfigurationProvider != null) { + return experimentConfigurationProvider; + } LocalCacheRegistry localCacheRegistry = DefaultLocalCacheRegistryFactory.getInstance().getDefaultLocalCacheRegistrySingleton(); Dao cacheDao = new LocalCacheDao(localCacheRegistry); experimentConfigurationProvider = new ExperimentConfigurationProvider(cacheDao); @@ -64,7 +70,9 @@ public TreatmentAllocator provideTreatmentAllocator() { return treatmentAllocator; } synchronized (this) { - treatmentAllocator = new TreatmentAllocatorOnRandom(); + if (treatmentAllocator == null) { + treatmentAllocator = new TreatmentAllocatorOnRandom(); + } return treatmentAllocator; } } diff --git a/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/ExtractorRegistryFactory.java b/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/ExtractorRegistryFactory.java index a095a5d..a39f133 100644 --- a/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/ExtractorRegistryFactory.java +++ b/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/ExtractorRegistryFactory.java @@ -18,7 +18,7 @@ public class ExtractorRegistryFactory { private static final ExtractorRegistryFactory INSTANCE = new ExtractorRegistryFactory(); - private ExtractorRegistry extractorRegistry; + private volatile ExtractorRegistry extractorRegistry; private ExtractorRegistryFactory() { } @@ -32,7 +32,9 @@ public ExtractorRegistry getSingleton() { return extractorRegistry; } synchronized (this) { - extractorRegistry = new ExtractorRegistry(new HashMap<>()); + if (extractorRegistry == null) { + extractorRegistry = new ExtractorRegistry(new HashMap<>()); + } return extractorRegistry; } } diff --git a/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/LocalCacheRegistryFactory.java b/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/LocalCacheRegistryFactory.java index c2a978d..6e21c6c 100644 --- a/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/LocalCacheRegistryFactory.java +++ b/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/LocalCacheRegistryFactory.java @@ -24,13 +24,16 @@ */ public abstract class LocalCacheRegistryFactory { - LocalCacheRegistry defaultLocalCacheRegistry; + volatile LocalCacheRegistry defaultLocalCacheRegistry; public LocalCacheRegistry getDefaultLocalCacheRegistrySingleton() { if (defaultLocalCacheRegistry != null) { return defaultLocalCacheRegistry; } synchronized (this) { + if (defaultLocalCacheRegistry != null) { + return defaultLocalCacheRegistry; + } boolean useDefaultLocalCacheBuilderConfig = true; ConcurrentMap> cacheMap = new ConcurrentHashMap<>(); Map localCacheBuilderMap = getLocalCacheBuilderMap(); diff --git a/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/TransformerRegistryFactory.java b/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/TransformerRegistryFactory.java index 0b519be..50a99f5 100644 --- a/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/TransformerRegistryFactory.java +++ b/java/src/main/java/com/amazon/demanddriventrafficevaluator/factory/TransformerRegistryFactory.java @@ -18,7 +18,7 @@ public class TransformerRegistryFactory { private static final TransformerRegistryFactory INSTANCE = new TransformerRegistryFactory(); - private TransformerRegistry transformerRegistry; + private volatile TransformerRegistry transformerRegistry; private TransformerRegistryFactory() { } @@ -32,7 +32,9 @@ public TransformerRegistry getSingleton() { return transformerRegistry; } synchronized (this) { - transformerRegistry = new TransformerRegistry(new HashMap<>()); + if (transformerRegistry == null) { + transformerRegistry = new TransformerRegistry(new HashMap<>()); + } return transformerRegistry; } } diff --git a/java/src/main/java/com/amazon/demanddriventrafficevaluator/modelfeature/Registry.java b/java/src/main/java/com/amazon/demanddriventrafficevaluator/modelfeature/Registry.java index dfba617..52d5476 100644 --- a/java/src/main/java/com/amazon/demanddriventrafficevaluator/modelfeature/Registry.java +++ b/java/src/main/java/com/amazon/demanddriventrafficevaluator/modelfeature/Registry.java @@ -3,6 +3,8 @@ package com.amazon.demanddriventrafficevaluator.modelfeature; +import java.util.Collections; +import java.util.HashMap; import java.util.Map; /** @@ -12,15 +14,33 @@ * objects of a specific type. It uses a Map to store objects, with string keys representing * the names of the registered objects. *

+ *

+ * Thread safety: the registry is filled by a setup task on a scheduler thread while request + * threads read it. The records are held as an immutable snapshot in a volatile field. Each + * {@code register} call is synchronized, copies the current snapshot, adds the new entry and + * publishes the copy, so a reader always sees a complete map and never a map being changed. + * Reads take no lock: one volatile read and one lookup in the snapshot (plus a second lookup + * only when the name is missing). Registration is expected to happen a handful of times at + * startup, so the copy on each write is cheap. + *

* * @param The type of objects stored in this registry. */ public abstract class Registry { - private final Map records; + private volatile Map records; + /** + * Creates a registry holding a snapshot of the given records. + *

+ * The map is copied; later changes to it are not seen by the registry. + *

+ * + * @param records The initial records. + * @throws NullPointerException if {@code records} is null. + */ protected Registry(Map records) { - this.records = records; + this.records = snapshot(new HashMap<>(records)); } /** @@ -43,16 +63,17 @@ protected Registry(Map records) { * @throws IllegalArgumentException if an object with the same name is already registered * or if the object cannot be instantiated. */ - public void register(String name, Class object) { + public synchronized void register(String name, Class object) { if (records.containsKey(name)) { throw new IllegalArgumentException(getRegistryType() + " with name " + name + " already registered"); - } else { - try { - records.put(name, object.getDeclaredConstructor().newInstance()); - } catch (Exception e) { - throw new IllegalArgumentException("Failed to register " + getRegistryType() + " with " + name, e); - } } + T instance; + try { + instance = object.getDeclaredConstructor().newInstance(); + } catch (Exception e) { + throw new IllegalArgumentException("Failed to register " + getRegistryType() + " with " + name, e); + } + publish(name, instance); } /** @@ -66,12 +87,11 @@ public void register(String name, Class object) { * @param object The object to register. * @throws IllegalArgumentException if an object with the same name is already registered. */ - public void register(String name, T object) { + public synchronized void register(String name, T object) { if (records.containsKey(name)) { throw new IllegalArgumentException(getRegistryType() + " with name " + name + " already registered"); - } else { - records.put(name, object); } + publish(name, object); } /** @@ -82,15 +102,22 @@ public void register(String name, T object) { * @throws IllegalArgumentException if no object is registered with the given name. */ public T get(String name) { - if (records.containsKey(name)) { - return records.get(name); - } else { + Map current = records; + T object = current.get(name); + // containsKey only on a miss, so a name registered with a null value still returns null + if (object == null && !current.containsKey(name)) { throw new IllegalArgumentException(getRegistryType() + " with name " + name + " not registered"); } + return object; } /** * Returns a map of all registered objects. + *

+ * The map is an unmodifiable snapshot of the registry at the time of the call; objects + * registered later are not added to it, so call this method again to see them. The + * registered objects themselves are not copied. + *

* * @return An unmodifiable map of all objects in the registry, where keys are object names * and values are the registered objects. @@ -98,4 +125,17 @@ public T get(String name) { public Map getRecords() { return records; } + + // Callers hold the lock: copy, add, then publish the copy through the volatile field. + private void publish(String name, T object) { + Map next = new HashMap<>(records); + next.put(name, object); + records = snapshot(next); + } + + // Collections.unmodifiableMap over a private HashMap, not Map.copyOf: Map.copyOf rejects null + // keys and values (HashMap and this class accepted them) and randomizes iteration order per JVM. + private static Map snapshot(Map map) { + return Collections.unmodifiableMap(map); + } } diff --git a/java/src/test/java/com/amazon/demanddriventrafficevaluator/factory/LazySingletonConcurrencyTest.java b/java/src/test/java/com/amazon/demanddriventrafficevaluator/factory/LazySingletonConcurrencyTest.java new file mode 100644 index 0000000..378f803 --- /dev/null +++ b/java/src/test/java/com/amazon/demanddriventrafficevaluator/factory/LazySingletonConcurrencyTest.java @@ -0,0 +1,107 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package com.amazon.demanddriventrafficevaluator.factory; + +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Constructor; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; +import java.util.function.Supplier; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; + +/** + * Concurrent first-call tests for the lazy singletons (issue #16): every caller + * racing on a fresh factory must receive the same instance. + */ +class LazySingletonConcurrencyTest { + + private static final int THREADS = 16; + private static final int ROUNDS = 50; + + @Test + void extractorRegistryFactory_concurrentFirstCalls_returnSameInstance() throws Exception { + assertSameInstanceUnderRace(() -> newViaPrivateConstructor(ExtractorRegistryFactory.class), + ExtractorRegistryFactory::getSingleton); + } + + @Test + void transformerRegistryFactory_concurrentFirstCalls_returnSameInstance() throws Exception { + assertSameInstanceUnderRace(() -> newViaPrivateConstructor(TransformerRegistryFactory.class), + TransformerRegistryFactory::getSingleton); + } + + @Test + void localCacheRegistryFactory_concurrentFirstCalls_returnSameInstance() throws Exception { + assertSameInstanceUnderRace(DefaultLocalCacheRegistryFactory::new, + LocalCacheRegistryFactory::getDefaultLocalCacheRegistrySingleton); + } + + @Test + void experimentManagerFactory_concurrentFirstCalls_returnSameExperimentManager() throws Exception { + assertSameInstanceUnderRace(ExperimentManagerFactory::new, + ExperimentManagerFactory::provideExperimentManager); + } + + @Test + void experimentManagerFactory_concurrentFirstCalls_returnSameConfigurationProvider() throws Exception { + assertSameInstanceUnderRace(ExperimentManagerFactory::new, + ExperimentManagerFactory::provideExperimentConfigurationProvider); + } + + @Test + void experimentManagerFactory_concurrentFirstCalls_returnSameTreatmentAllocator() throws Exception { + assertSameInstanceUnderRace(ExperimentManagerFactory::new, + ExperimentManagerFactory::provideTreatmentAllocator); + } + + private static F newViaPrivateConstructor(Class type) { + try { + Constructor constructor = type.getDeclaredConstructor(); + constructor.setAccessible(true); + return constructor.newInstance(); + } catch (ReflectiveOperationException e) { + throw new IllegalStateException(e); + } + } + + private static void assertSameInstanceUnderRace(Supplier freshFactory, Function getter) + throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(THREADS); + try { + for (int round = 0; round < ROUNDS; round++) { + F factory = freshFactory.get(); + CountDownLatch ready = new CountDownLatch(THREADS); + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(); + for (int i = 0; i < THREADS; i++) { + futures.add(executor.submit(() -> { + ready.countDown(); + start.await(); + return getter.apply(factory); + })); + } + ready.await(); + start.countDown(); + T first = futures.get(0).get(10, TimeUnit.SECONDS); + assertNotNull(first); + for (Future future : futures) { + assertSame(first, future.get(10, TimeUnit.SECONDS), + "concurrent first callers received different instances in round " + round); + } + } + } finally { + executor.shutdownNow(); + executor.awaitTermination(10, TimeUnit.SECONDS); + } + } +} diff --git a/java/src/test/java/com/amazon/demanddriventrafficevaluator/modelfeature/RegistryTest.java b/java/src/test/java/com/amazon/demanddriventrafficevaluator/modelfeature/RegistryTest.java new file mode 100644 index 0000000..590bcae --- /dev/null +++ b/java/src/test/java/com/amazon/demanddriventrafficevaluator/modelfeature/RegistryTest.java @@ -0,0 +1,193 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package com.amazon.demanddriventrafficevaluator.modelfeature; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +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.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Thread-safety contract of {@link Registry} (issue #16): records are published as immutable + * snapshots, so readers never see a map that a register() call is changing. + */ +class RegistryTest { + + private static final int REGISTRATIONS = 1_000; + private static final int READERS = 4; + + private static final class TestRegistry extends Registry { + TestRegistry(Map records) { + super(records); + } + + @Override + protected String getRegistryType() { + return "Test"; + } + } + + @Test + void getRecords_isUnmodifiable() { + TestRegistry registry = new TestRegistry(new HashMap<>(Map.of("a", "A"))); + + Map records = registry.getRecords(); + + assertThrows(UnsupportedOperationException.class, () -> records.put("b", "B")); + assertThrows(UnsupportedOperationException.class, () -> records.remove("a")); + assertThrows(UnsupportedOperationException.class, records::clear); + } + + @Test + void constructor_copiesMap_laterChangesToCallerMapAreNotSeen() { + Map source = new HashMap<>(Map.of("a", "A")); + TestRegistry registry = new TestRegistry(source); + + source.put("b", "B"); + + assertFalse(registry.getRecords().containsKey("b")); + assertThrows(IllegalArgumentException.class, () -> registry.get("b")); + } + + @Test + void register_doesNotChangeEarlierSnapshot_andIsVisibleInNewSnapshot() { + TestRegistry registry = new TestRegistry(new HashMap<>(Map.of("a", "A"))); + Map before = registry.getRecords(); + + registry.register("b", "B"); + + assertEquals(Map.of("a", "A"), before); + assertEquals(Map.of("a", "A", "b", "B"), registry.getRecords()); + assertEquals("B", registry.get("b")); + } + + @Test + void register_whileIteratingRecords_doesNotThrowConcurrentModification() { + TestRegistry registry = new TestRegistry(new HashMap<>(Map.of("a", "A", "b", "B"))); + + // Single-threaded, deterministic: a live HashMap fails fast here. + assertDoesNotThrow(() -> { + int i = 0; + for (String key : registry.getRecords().keySet()) { + registry.register("new-" + key + i++, "X"); + } + }); + assertEquals(4, registry.getRecords().size()); + } + + @Test + void get_nameRegisteredWithNullValue_returnsNull() { + TestRegistry registry = new TestRegistry(new HashMap<>()); + + registry.register("n", (Object) null); + + assertNull(registry.get("n")); + assertThrows(IllegalArgumentException.class, () -> registry.register("n", "N")); + } + + @Test + void get_unknownName_throwsWithSameMessage() { + TestRegistry registry = new TestRegistry(new HashMap<>()); + + IllegalArgumentException e = assertThrows(IllegalArgumentException.class, () -> registry.get("x")); + assertEquals("Test with name x not registered", e.getMessage()); + } + + @Test + void register_duplicateName_throwsAndKeepsOriginal() { + TestRegistry registry = new TestRegistry(new HashMap<>(Map.of("a", "A"))); + + IllegalArgumentException e = assertThrows(IllegalArgumentException.class, () -> registry.register("a", "Z")); + assertEquals("Test with name a already registered", e.getMessage()); + assertEquals("A", registry.get("a")); + } + + @Test + void registerWhileReadersGetAndIterate_readersNeverFail() throws Exception { + Object seed = new Object(); + TestRegistry registry = new TestRegistry(new HashMap<>(Map.of("seed", seed))); + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); + AtomicBoolean writerDone = new AtomicBoolean(); + // Snapshots seen by readers with a size strictly between the start and the end, i.e. reads + // that overlapped the writer. Without one the test proves nothing, so it must be > 0. + AtomicInteger overlappingReads = new AtomicInteger(); + CountDownLatch start = new CountDownLatch(1); + + Thread writer = new Thread(() -> { + try { + start.await(); + for (int i = 0; i < REGISTRATIONS; i++) { + registry.register("k" + i, "v" + i); + if (i == REGISTRATIONS / 2) { + // Hold the registry half full until a reader has seen it, so readers overlap + // the writer even when they are scheduled late. + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (overlappingReads.get() == 0 && System.nanoTime() < deadline) { + Thread.onSpinWait(); + } + } + } + } catch (Throwable t) { + failures.add(t); + } finally { + writerDone.set(true); + } + }); + List readers = new ArrayList<>(); + for (int r = 0; r < READERS; r++) { + readers.add(new Thread(() -> { + try { + start.await(); + while (!writerDone.get()) { + assertSame(seed, registry.get("seed")); + Map snapshot = registry.getRecords(); + int seen = 0; + for (Map.Entry entry : snapshot.entrySet()) { + if (entry.getValue() != null) { + seen++; + } + } + assertEquals(snapshot.size(), seen); + if (seen > 1 && seen < REGISTRATIONS + 1) { + overlappingReads.incrementAndGet(); + } + } + } catch (Throwable t) { + failures.add(t); + } + })); + } + writer.start(); + readers.forEach(Thread::start); + start.countDown(); + writer.join(TimeUnit.SECONDS.toMillis(60)); + assertFalse(writer.isAlive(), "writer did not finish"); + for (Thread reader : readers) { + reader.join(TimeUnit.SECONDS.toMillis(60)); + assertFalse(reader.isAlive(), "reader did not finish"); + } + + assertTrue(failures.isEmpty(), () -> "reader/writer failures: " + failures); + assertTrue(overlappingReads.get() > 0, "no read overlapped the writer"); + assertEquals(REGISTRATIONS + 1, registry.getRecords().size()); + for (int i = 0; i < REGISTRATIONS; i++) { + assertEquals("v" + i, registry.get("k" + i)); + } + } +} diff --git a/java/src/test/java/com/amazon/demanddriventrafficevaluator/modelfeature/extractor/ExtractorRegistryTest.java b/java/src/test/java/com/amazon/demanddriventrafficevaluator/modelfeature/extractor/ExtractorRegistryTest.java index 062f000..766b339 100644 --- a/java/src/test/java/com/amazon/demanddriventrafficevaluator/modelfeature/extractor/ExtractorRegistryTest.java +++ b/java/src/test/java/com/amazon/demanddriventrafficevaluator/modelfeature/extractor/ExtractorRegistryTest.java @@ -19,7 +19,6 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; -import static org.junit.jupiter.api.Assertions.assertTrue; @ExtendWith(MockitoExtension.class) class ExtractorRegistryTest { @@ -47,9 +46,9 @@ void setUp() { void testConstructor() { // Test that registry is created with correct extractors assertNotNull(registry); - assertEquals(2, mockExtractors.size()); - assertTrue(mockExtractors.containsKey("extractor1")); - assertTrue(mockExtractors.containsKey("extractor2")); + assertEquals(2, registry.getRecords().size()); + assertEquals(mockExtractor1, registry.get("extractor1")); + assertEquals(mockExtractor2, registry.get("extractor2")); } @Test @@ -74,8 +73,8 @@ void testRegisterClass() { registry.register("jsonExtractor", JsonExtractor.class); - assertEquals(1, mockExtractors.size()); - assertEquals(JsonExtractor.class, mockExtractors.get("jsonExtractor").getClass()); + assertEquals(1, registry.getRecords().size()); + assertEquals(JsonExtractor.class, registry.get("jsonExtractor").getClass()); } @Test @@ -97,8 +96,8 @@ void testRegisterObject() { registry.register("jsonExtractor", new JsonExtractor()); - assertEquals(1, mockExtractors.size()); - assertEquals(JsonExtractor.class, mockExtractors.get("jsonExtractor").getClass()); + assertEquals(1, registry.getRecords().size()); + assertEquals(JsonExtractor.class, registry.get("jsonExtractor").getClass()); } @Test