diff --git a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/CtSph.java b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/CtSph.java index 88a76f750d..c756c349a3 100755 --- a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/CtSph.java +++ b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/CtSph.java @@ -16,8 +16,8 @@ package com.alibaba.csp.sentinel; import java.lang.reflect.Method; -import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import com.alibaba.csp.sentinel.log.RecordLog; import com.alibaba.csp.sentinel.context.Context; @@ -48,10 +48,8 @@ public class CtSph implements Sph { * Same resource({@link ResourceWrapper#equals(Object)}) will share the same * {@link ProcessorSlotChain}, no matter in which {@link Context}. */ - private static volatile Map chainMap - = new HashMap(); - - private static final Object LOCK = new Object(); + private static final Map CHAIN_MAP + = new ConcurrentHashMap(); private AsyncEntry asyncEntryWithNoChain(ResourceWrapper resourceWrapper, Context context) { AsyncEntry entry = new AsyncEntry(resourceWrapper, null, context); @@ -192,22 +190,18 @@ public Entry entry(ResourceWrapper resourceWrapper, int count, Object... args) t * @return {@link ProcessorSlotChain} of the resource */ ProcessorSlot lookProcessChain(ResourceWrapper resourceWrapper) { - ProcessorSlotChain chain = chainMap.get(resourceWrapper); + ProcessorSlotChain chain = CHAIN_MAP.get(resourceWrapper); if (chain == null) { - synchronized (LOCK) { - chain = chainMap.get(resourceWrapper); + synchronized (CHAIN_MAP) { + chain = CHAIN_MAP.get(resourceWrapper); if (chain == null) { // Entry size limit. - if (chainMap.size() >= Constants.MAX_SLOT_CHAIN_SIZE) { + if (CHAIN_MAP.size() >= Constants.MAX_SLOT_CHAIN_SIZE) { return null; } chain = SlotChainProvider.newSlotChain(); - Map newMap = new HashMap( - chainMap.size() + 1); - newMap.putAll(chainMap); - newMap.put(resourceWrapper, chain); - chainMap = newMap; + CHAIN_MAP.put(resourceWrapper, chain); } } } @@ -221,7 +215,9 @@ ProcessorSlot lookProcessChain(ResourceWrapper resourceWrapper) { * @since 0.2.0 */ public static int entrySize() { - return chainMap.size(); + synchronized (CHAIN_MAP) { + return CHAIN_MAP.size(); + } } /** @@ -230,7 +226,9 @@ public static int entrySize() { * @since 0.2.0 */ static void resetChainMap() { - chainMap.clear(); + synchronized (CHAIN_MAP) { + CHAIN_MAP.clear(); + } } /** @@ -239,7 +237,7 @@ static void resetChainMap() { * @since 0.2.0 */ static Map getChainMap() { - return chainMap; + return CHAIN_MAP; } /** diff --git a/sentinel-core/src/test/java/com/alibaba/csp/sentinel/CtSphTest.java b/sentinel-core/src/test/java/com/alibaba/csp/sentinel/CtSphTest.java index 1d910ee76a..f88cb87ef8 100644 --- a/sentinel-core/src/test/java/com/alibaba/csp/sentinel/CtSphTest.java +++ b/sentinel-core/src/test/java/com/alibaba/csp/sentinel/CtSphTest.java @@ -1,5 +1,16 @@ package com.alibaba.csp.sentinel; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentMap; +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.concurrent.TimeoutException; + import com.alibaba.csp.sentinel.context.Context; import com.alibaba.csp.sentinel.context.ContextTestUtil; import com.alibaba.csp.sentinel.context.ContextUtil; @@ -269,6 +280,99 @@ public void testLookUpSlotChain() { assertNull(ctSph.lookProcessChain(r2)); } + @Test + public void testChainMapSupportsConcurrentAccess() { + assertTrue("Chain map should support lock-free concurrent reads", + CtSph.getChainMap() instanceof ConcurrentMap); + } + + @Test + public void testLookUpSameSlotChainConcurrently() throws Exception { + final int taskCount = 32; + final ResourceWrapper resource = new StringResourceWrapper("concurrent-resource", EntryType.IN); + List>> tasks = + new ArrayList>>(taskCount); + for (int i = 0; i < taskCount; i++) { + tasks.add(new Callable>() { + @Override + public ProcessorSlot call() { + return ctSph.lookProcessChain(resource); + } + }); + } + + List> chains = invokeConcurrently(tasks); + ProcessorSlot expected = chains.get(0); + assertNotNull(expected); + for (ProcessorSlot chain : chains) { + assertSame("Same resource should share one slot chain", expected, chain); + } + assertEquals(1, CtSph.entrySize()); + } + + @Test + public void testLookUpSlotChainAtCapacityConcurrently() throws Exception { + fillResources(Constants.MAX_SLOT_CHAIN_SIZE - 1); + final int taskCount = 16; + List>> tasks = + new ArrayList>>(taskCount); + for (int i = 0; i < taskCount; i++) { + final ResourceWrapper resource = + new StringResourceWrapper("concurrent-capacity-resource-" + i, EntryType.IN); + tasks.add(new Callable>() { + @Override + public ProcessorSlot call() { + return ctSph.lookProcessChain(resource); + } + }); + } + + int createdCount = 0; + for (ProcessorSlot chain : invokeConcurrently(tasks)) { + if (chain != null) { + createdCount++; + } + } + assertEquals("Only one resource should be created at the capacity boundary", 1, createdCount); + assertEquals(Constants.MAX_SLOT_CHAIN_SIZE, CtSph.entrySize()); + } + + private List invokeConcurrently(List> tasks) throws Exception { + final int taskCount = tasks.size(); + final ExecutorService executor = Executors.newFixedThreadPool(taskCount); + final CountDownLatch ready = new CountDownLatch(taskCount); + final CountDownLatch start = new CountDownLatch(1); + final List> futures = new ArrayList>(taskCount); + + try { + for (final Callable task : tasks) { + futures.add(executor.submit(new Callable() { + @Override + public T call() throws Exception { + ready.countDown(); + if (!start.await(5, TimeUnit.SECONDS)) { + throw new TimeoutException("Timed out waiting for concurrent test start"); + } + return task.call(); + } + })); + } + + assertTrue("Concurrent tasks were not ready in time", ready.await(5, TimeUnit.SECONDS)); + start.countDown(); + + List results = new ArrayList(taskCount); + for (Future future : futures) { + results.add(future.get(10, TimeUnit.SECONDS)); + } + return results; + } finally { + start.countDown(); + executor.shutdownNow(); + assertTrue("Executor did not terminate in time", executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + private void fillFullContext() { for (int i = 0; i < Constants.MAX_CONTEXT_NAME_SIZE; i++) { ContextUtil.enter("test-context-" + i); @@ -277,7 +381,11 @@ private void fillFullContext() { } private void fillFullResources() { - for (int i = 0; i < Constants.MAX_SLOT_CHAIN_SIZE; i++) { + fillResources(Constants.MAX_SLOT_CHAIN_SIZE); + } + + private void fillResources(int count) { + for (int i = 0; i < count; i++) { ResourceWrapper resourceWrapper = new StringResourceWrapper("test-resource-" + i, EntryType.IN); CtSph.getChainMap().put(resourceWrapper, SlotChainProvider.newSlotChain()); } @@ -362,4 +470,4 @@ public void tearDown() throws Exception { ContextTestUtil.resetContextMap(); CtSph.resetChainMap(); } -} \ No newline at end of file +}