【Enhancement】优化主流程lookProcessChain方法 issue #3513

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<ResourceWrapper, ProcessorSlotChain> chainMap
-        = new HashMap<ResourceWrapper, ProcessorSlotChain>();
-
-    private static final Object LOCK = new Object();
+    private static final Map<ResourceWrapper, ProcessorSlotChain> CHAIN_MAP
+        = new ConcurrentHashMap<ResourceWrapper, ProcessorSlotChain>();
 
     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<Object> 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<ResourceWrapper, ProcessorSlotChain> newMap = new HashMap<ResourceWrapper, ProcessorSlotChain>(
-                        chainMap.size() + 1);
-                    newMap.putAll(chainMap);
-                    newMap.put(resourceWrapper, chain);
-                    chainMap = newMap;
+                    CHAIN_MAP.put(resourceWrapper, chain);
                 }
             }
         }
@@ -221,7 +215,9 @@ ProcessorSlot<Object> 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<ResourceWrapper, ProcessorSlotChain> getChainMap() {
-        return chainMap;
+        return CHAIN_MAP;
     }
 
     /**

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<Callable<ProcessorSlot<Object>>> tasks =
+            new ArrayList<Callable<ProcessorSlot<Object>>>(taskCount);
+        for (int i = 0; i < taskCount; i++) {
+            tasks.add(new Callable<ProcessorSlot<Object>>() {
+                @Override
+                public ProcessorSlot<Object> call() {
+                    return ctSph.lookProcessChain(resource);
+                }
+            });
+        }
+
+        List<ProcessorSlot<Object>> chains = invokeConcurrently(tasks);
+        ProcessorSlot<Object> expected = chains.get(0);
+        assertNotNull(expected);
+        for (ProcessorSlot<Object> 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<Callable<ProcessorSlot<Object>>> tasks =
+            new ArrayList<Callable<ProcessorSlot<Object>>>(taskCount);
+        for (int i = 0; i < taskCount; i++) {
+            final ResourceWrapper resource =
+                new StringResourceWrapper("concurrent-capacity-resource-" + i, EntryType.IN);
+            tasks.add(new Callable<ProcessorSlot<Object>>() {
+                @Override
+                public ProcessorSlot<Object> call() {
+                    return ctSph.lookProcessChain(resource);
+                }
+            });
+        }
+
+        int createdCount = 0;
+        for (ProcessorSlot<Object> 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 <T> List<T> invokeConcurrently(List<Callable<T>> 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<Future<T>> futures = new ArrayList<Future<T>>(taskCount);
+
+        try {
+            for (final Callable<T> task : tasks) {
+                futures.add(executor.submit(new Callable<T>() {
+                    @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<T> results = new ArrayList<T>(taskCount);
+            for (Future<T> 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();
     }
-}
+}