From 28fbd0b6740c05564acb6e0bc22626751854ca60 Mon Sep 17 00:00:00 2001 From: TwistedRiCen <16397953+TwistedRiCen@users.noreply.github.com> Date: Fri, 4 Sep 2026 22:52:46 +0800 Subject: [PATCH] fix: isolate processor slot chain links --- .../AbstractLinkedProcessorSlot.java | 38 +++ .../slotchain/DefaultProcessorSlotChain.java | 19 +- .../slotchain/ProcessorSlotContext.java | 63 +++++ .../DefaultProcessorSlotChainTest.java | 267 ++++++++++++++++++ .../slots/DefaultSlotChainBuilderTest.java | 40 ++- 5 files changed, 410 insertions(+), 17 deletions(-) create mode 100644 sentinel-core/src/main/java/com/alibaba/csp/sentinel/slotchain/ProcessorSlotContext.java create mode 100644 sentinel-core/src/test/java/com/alibaba/csp/sentinel/slotchain/DefaultProcessorSlotChainTest.java diff --git a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slotchain/AbstractLinkedProcessorSlot.java b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slotchain/AbstractLinkedProcessorSlot.java index 2da213a21b..c882a40704 100755 --- a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slotchain/AbstractLinkedProcessorSlot.java +++ b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slotchain/AbstractLinkedProcessorSlot.java @@ -23,11 +23,14 @@ */ public abstract class AbstractLinkedProcessorSlot implements ProcessorSlot { + private static final ThreadLocal CHAIN_CONTEXT = new ThreadLocal<>(); + private AbstractLinkedProcessorSlot next = null; @Override public void fireEntry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args) throws Throwable { + AbstractLinkedProcessorSlot next = getNext(); if (next != null) { next.transformEntry(context, resourceWrapper, obj, count, prioritized, args); } @@ -42,12 +45,17 @@ void transformEntry(Context context, ResourceWrapper resourceWrapper, Object o, @Override public void fireExit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) { + AbstractLinkedProcessorSlot next = getNext(); if (next != null) { next.exit(context, resourceWrapper, count, args); } } public AbstractLinkedProcessorSlot getNext() { + ChainContext context = CHAIN_CONTEXT.get(); + if (context != null && context.source == this) { + return context.next; + } return next; } @@ -55,4 +63,34 @@ public void setNext(AbstractLinkedProcessorSlot next) { this.next = next; } + static ChainContext setChainContext(AbstractLinkedProcessorSlot source, + AbstractLinkedProcessorSlot next) { + ChainContext previous = CHAIN_CONTEXT.get(); + CHAIN_CONTEXT.set(new ChainContext(source, next)); + return previous; + } + + static void restoreChainContext(ChainContext previous) { + if (previous == null) { + CHAIN_CONTEXT.remove(); + } else { + CHAIN_CONTEXT.set(previous); + } + } + + static boolean hasActiveChainContext() { + return CHAIN_CONTEXT.get() != null; + } + + static final class ChainContext { + + private final AbstractLinkedProcessorSlot source; + private final AbstractLinkedProcessorSlot next; + + private ChainContext(AbstractLinkedProcessorSlot source, AbstractLinkedProcessorSlot next) { + this.source = source; + this.next = next; + } + } + } diff --git a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slotchain/DefaultProcessorSlotChain.java b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slotchain/DefaultProcessorSlotChain.java index 5906d1792b..c274644224 100755 --- a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slotchain/DefaultProcessorSlotChain.java +++ b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slotchain/DefaultProcessorSlotChain.java @@ -41,17 +41,19 @@ public void exit(Context context, ResourceWrapper resourceWrapper, int count, Ob @Override public void addFirst(AbstractLinkedProcessorSlot protocolProcessor) { - protocolProcessor.setNext(first.getNext()); - first.setNext(protocolProcessor); + AbstractLinkedProcessorSlot context = wrap(protocolProcessor); + context.setNext(first.getNext()); + first.setNext(context); if (end == first) { - end = protocolProcessor; + end = context; } } @Override public void addLast(AbstractLinkedProcessorSlot protocolProcessor) { - end.setNext(protocolProcessor); - end = protocolProcessor; + AbstractLinkedProcessorSlot context = wrap(protocolProcessor); + end.setNext(context); + end = context; } /** @@ -80,4 +82,11 @@ public void exit(Context context, ResourceWrapper resourceWrapper, int count, Ob first.exit(context, resourceWrapper, count, args); } + private AbstractLinkedProcessorSlot wrap(AbstractLinkedProcessorSlot processor) { + if (processor instanceof ProcessorSlotContext) { + return processor; + } + return new ProcessorSlotContext<>(processor); + } + } diff --git a/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slotchain/ProcessorSlotContext.java b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slotchain/ProcessorSlotContext.java new file mode 100644 index 0000000000..cd115eb02d --- /dev/null +++ b/sentinel-core/src/main/java/com/alibaba/csp/sentinel/slotchain/ProcessorSlotContext.java @@ -0,0 +1,63 @@ +/* + * Copyright 1999-2018 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.csp.sentinel.slotchain; + +import com.alibaba.csp.sentinel.context.Context; + +/** + * A chain-local node that delegates processing to a {@link ProcessorSlot} while owning an independent + * {@code next} reference. + * + * @param type of the entry parameter + */ +public final class ProcessorSlotContext extends AbstractLinkedProcessorSlot { + + private final AbstractLinkedProcessorSlot delegate; + + ProcessorSlotContext(AbstractLinkedProcessorSlot delegate) { + this.delegate = delegate; + } + + @Override + public void entry(Context context, ResourceWrapper resourceWrapper, T param, int count, boolean prioritized, + Object... args) throws Throwable { + ChainContext previous = setChainContext(delegate, getNext()); + try { + delegate.entry(context, resourceWrapper, param, count, prioritized, args); + } finally { + restoreChainContext(previous); + } + } + + @Override + public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) { + ChainContext previous = setChainContext(delegate, getNext()); + try { + delegate.exit(context, resourceWrapper, count, args); + } finally { + restoreChainContext(previous); + } + } + + /** + * Get the SPI-managed slot that performs the actual processing. + * + * @return delegate slot + */ + public AbstractLinkedProcessorSlot getDelegate() { + return delegate; + } +} diff --git a/sentinel-core/src/test/java/com/alibaba/csp/sentinel/slotchain/DefaultProcessorSlotChainTest.java b/sentinel-core/src/test/java/com/alibaba/csp/sentinel/slotchain/DefaultProcessorSlotChainTest.java new file mode 100644 index 0000000000..c2a9771b7c --- /dev/null +++ b/sentinel-core/src/test/java/com/alibaba/csp/sentinel/slotchain/DefaultProcessorSlotChainTest.java @@ -0,0 +1,267 @@ +/* + * Copyright 1999-2018 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.csp.sentinel.slotchain; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; + +import com.alibaba.csp.sentinel.context.Context; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.fail; + +/** + * Test cases for {@link DefaultProcessorSlotChain}. + */ +public class DefaultProcessorSlotChainTest { + + @Test + public void testSharedPredecessorDoesNotOverwritePrototypeSuccessor() throws Throwable { + List invocations = new ArrayList<>(); + RecordingSlot singletonB = new RecordingSlot("B", invocations); + RecordingSlot prototypeA1 = new RecordingSlot("A1", invocations); + RecordingSlot prototypeA2 = new RecordingSlot("A2", invocations); + + ProcessorSlotChain chain1 = new DefaultProcessorSlotChain(); + chain1.addLast(singletonB); + chain1.addLast(prototypeA1); + + ProcessorSlotChain chain2 = new DefaultProcessorSlotChain(); + chain2.addLast(singletonB); + chain2.addLast(prototypeA2); + + chain1.entry(null, null, null, 1, false); + + assertEquals(Arrays.asList("B", "A1"), invocations); + + invocations.clear(); + chain1.exit(null, null, 1); + assertEquals(Arrays.asList("B-exit", "A1-exit"), invocations); + + ProcessorSlotContext contextB1 = (ProcessorSlotContext) chain1.getNext(); + ProcessorSlotContext contextB2 = (ProcessorSlotContext) chain2.getNext(); + assertNotSame(contextB1, contextB2); + assertSame(singletonB, contextB1.getDelegate()); + assertSame(singletonB, contextB2.getDelegate()); + } + + @Test + public void testScopedContextSupportsNestedChains() throws Throwable { + List invocations = new ArrayList<>(); + + ProcessorSlotChain nested = new DefaultProcessorSlotChain(); + nested.addLast(new RecordingSlot("nested-B", invocations)); + nested.addLast(new RecordingSlot("nested-A", invocations)); + + ProcessorSlotChain outer = new DefaultProcessorSlotChain(); + outer.addLast(new NestedSlot("outer-B", invocations, nested)); + outer.addLast(new RecordingSlot("outer-A", invocations)); + + outer.entry(null, null, null, 1, false); + + assertEquals(Arrays.asList("outer-B", "nested-B", "nested-A", "outer-A"), invocations); + assertFalse(AbstractLinkedProcessorSlot.hasActiveChainContext()); + } + + @Test + public void testScopedContextSupportsRepeatedFireEntry() throws Throwable { + List invocations = new ArrayList<>(); + ProcessorSlotChain chain = new DefaultProcessorSlotChain(); + chain.addLast(new DoubleFireSlot("B", invocations)); + chain.addLast(new RecordingSlot("A", invocations)); + + chain.entry(null, null, null, 1, false); + + assertEquals(Arrays.asList("B", "A", "A"), invocations); + assertFalse(AbstractLinkedProcessorSlot.hasActiveChainContext()); + } + + @Test + public void testScopedContextIsRestoredAfterException() throws Throwable { + ProcessorSlotChain chain = new DefaultProcessorSlotChain(); + chain.addLast(new ThrowingSlot()); + + try { + chain.entry(null, null, null, 1, false); + fail("Should throw the expected exception"); + } catch (IllegalStateException expected) { + assertEquals("expected", expected.getMessage()); + } + assertFalse(AbstractLinkedProcessorSlot.hasActiveChainContext()); + + List invocations = new ArrayList<>(); + RecordingSlot directB = new RecordingSlot("direct-B", invocations); + directB.setNext(new RecordingSlot("direct-A", invocations)); + directB.entry(null, null, null, 1, false); + assertEquals(Arrays.asList("direct-B", "direct-A"), invocations); + } + + @Test + public void testConcurrentChainsKeepIndependentSuccessors() throws Exception { + int chainCount = 32; + int invocationCount = 100; + AtomicInteger sharedCount = new AtomicInteger(); + CountingSlot singletonB = new CountingSlot(sharedCount); + List prototypeCounts = new ArrayList<>(chainCount); + List chains = new ArrayList<>(chainCount); + + for (int i = 0; i < chainCount; i++) { + AtomicInteger prototypeCount = new AtomicInteger(); + ProcessorSlotChain chain = new DefaultProcessorSlotChain(); + chain.addLast(singletonB); + chain.addLast(new CountingSlot(prototypeCount)); + prototypeCounts.add(prototypeCount); + chains.add(chain); + } + + ExecutorService executor = Executors.newFixedThreadPool(8); + List> futures = new ArrayList<>(chainCount); + try { + for (final ProcessorSlotChain chain : chains) { + futures.add(executor.submit(new Runnable() { + @Override + public void run() { + try { + for (int i = 0; i < invocationCount; i++) { + chain.entry(null, null, null, 1, false); + } + } catch (Throwable t) { + throw new AssertionError(t); + } + if (AbstractLinkedProcessorSlot.hasActiveChainContext()) { + throw new AssertionError("Chain context leaked from invocation"); + } + } + })); + } + for (Future future : futures) { + future.get(); + } + } finally { + executor.shutdownNow(); + } + + assertEquals(chainCount * invocationCount, sharedCount.get()); + for (AtomicInteger prototypeCount : prototypeCounts) { + assertEquals(invocationCount, prototypeCount.get()); + } + } + + private static class RecordingSlot extends AbstractLinkedProcessorSlot { + + private final String name; + private final List invocations; + + private RecordingSlot(String name, List invocations) { + this.name = name; + this.invocations = invocations; + } + + @Override + public void entry(Context context, ResourceWrapper resourceWrapper, Object param, int count, + boolean prioritized, Object... args) throws Throwable { + invocations.add(name); + fireEntry(context, resourceWrapper, param, count, prioritized, args); + } + + @Override + public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) { + invocations.add(name + "-exit"); + fireExit(context, resourceWrapper, count, args); + } + + protected void recordEntry() { + invocations.add(name); + } + } + + private static class NestedSlot extends RecordingSlot { + + private final ProcessorSlotChain nested; + + private NestedSlot(String name, List invocations, ProcessorSlotChain nested) { + super(name, invocations); + this.nested = nested; + } + + @Override + public void entry(Context context, ResourceWrapper resourceWrapper, Object param, int count, + boolean prioritized, Object... args) throws Throwable { + recordEntry(); + nested.entry(context, resourceWrapper, param, count, prioritized, args); + fireEntry(context, resourceWrapper, param, count, prioritized, args); + } + } + + private static class DoubleFireSlot extends RecordingSlot { + + private DoubleFireSlot(String name, List invocations) { + super(name, invocations); + } + + @Override + public void entry(Context context, ResourceWrapper resourceWrapper, Object param, int count, + boolean prioritized, Object... args) throws Throwable { + recordEntry(); + fireEntry(context, resourceWrapper, param, count, prioritized, args); + fireEntry(context, resourceWrapper, param, count, prioritized, args); + } + } + + private static class ThrowingSlot extends AbstractLinkedProcessorSlot { + + @Override + public void entry(Context context, ResourceWrapper resourceWrapper, Object param, int count, + boolean prioritized, Object... args) { + throw new IllegalStateException("expected"); + } + + @Override + public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) { + // No-op. + } + } + + private static class CountingSlot extends AbstractLinkedProcessorSlot { + + private final AtomicInteger counter; + + private CountingSlot(AtomicInteger counter) { + this.counter = counter; + } + + @Override + public void entry(Context context, ResourceWrapper resourceWrapper, Object param, int count, + boolean prioritized, Object... args) throws Throwable { + counter.incrementAndGet(); + fireEntry(context, resourceWrapper, param, count, prioritized, args); + } + + @Override + public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) { + fireExit(context, resourceWrapper, count, args); + } + } +} diff --git a/sentinel-core/src/test/java/com/alibaba/csp/sentinel/slots/DefaultSlotChainBuilderTest.java b/sentinel-core/src/test/java/com/alibaba/csp/sentinel/slots/DefaultSlotChainBuilderTest.java index 9527dc3990..eff05a8193 100644 --- a/sentinel-core/src/test/java/com/alibaba/csp/sentinel/slots/DefaultSlotChainBuilderTest.java +++ b/sentinel-core/src/test/java/com/alibaba/csp/sentinel/slots/DefaultSlotChainBuilderTest.java @@ -17,6 +17,7 @@ import com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot; import com.alibaba.csp.sentinel.slotchain.ProcessorSlotChain; +import com.alibaba.csp.sentinel.slotchain.ProcessorSlotContext; import com.alibaba.csp.sentinel.slots.block.authority.AuthoritySlot; import com.alibaba.csp.sentinel.slots.block.degrade.DefaultCircuitBreakerSlot; import com.alibaba.csp.sentinel.slots.block.degrade.DegradeSlot; @@ -45,34 +46,35 @@ public void testBuild() { // Verify the order of slot AbstractLinkedProcessorSlot next = slotChain.getNext(); - assertTrue(next instanceof NodeSelectorSlot); + ProcessorSlotContext nodeSelectorContext = assertContext(next, NodeSelectorSlot.class); // Store the first NodeSelectorSlot instance - NodeSelectorSlot nodeSelectorSlot = (NodeSelectorSlot) next; + NodeSelectorSlot nodeSelectorSlot = (NodeSelectorSlot) nodeSelectorContext.getDelegate(); next = next.getNext(); - assertTrue(next instanceof ClusterBuilderSlot); + assertContext(next, ClusterBuilderSlot.class); next = next.getNext(); - assertTrue(next instanceof LogSlot); + ProcessorSlotContext logContext = assertContext(next, LogSlot.class); + LogSlot logSlot = (LogSlot) logContext.getDelegate(); next = next.getNext(); - assertTrue(next instanceof StatisticSlot); + assertContext(next, StatisticSlot.class); next = next.getNext(); - assertTrue(next instanceof AuthoritySlot); + assertContext(next, AuthoritySlot.class); next = next.getNext(); - assertTrue(next instanceof SystemSlot); + assertContext(next, SystemSlot.class); next = next.getNext(); - assertTrue(next instanceof FlowSlot); + assertContext(next, FlowSlot.class); next = next.getNext(); - assertTrue(next instanceof DefaultCircuitBreakerSlot); + assertContext(next, DefaultCircuitBreakerSlot.class); next = next.getNext(); - assertTrue(next instanceof DegradeSlot); + assertContext(next, DegradeSlot.class); next = next.getNext(); assertNull(next); @@ -84,10 +86,24 @@ public void testBuild() { assertNotSame(slotChain, slotChain2); next = slotChain2.getNext(); - assertTrue(next instanceof NodeSelectorSlot); + ProcessorSlotContext nodeSelectorContext2 = assertContext(next, NodeSelectorSlot.class); + assertNotSame(nodeSelectorContext, nodeSelectorContext2); // Store the second NodeSelectorSlot instance - NodeSelectorSlot nodeSelectorSlot2 = (NodeSelectorSlot) next; + NodeSelectorSlot nodeSelectorSlot2 = (NodeSelectorSlot) nodeSelectorContext2.getDelegate(); // Verify the two NodeSelectorSlot instances are different assertNotSame(nodeSelectorSlot, nodeSelectorSlot2); + + next = next.getNext().getNext(); + ProcessorSlotContext logContext2 = assertContext(next, LogSlot.class); + assertNotSame(logContext, logContext2); + assertSame(logSlot, logContext2.getDelegate()); + } + + private ProcessorSlotContext assertContext(AbstractLinkedProcessorSlot slot, + Class delegateClass) { + assertTrue(slot instanceof ProcessorSlotContext); + ProcessorSlotContext context = (ProcessorSlotContext) slot; + assertTrue(delegateClass.isInstance(context.getDelegate())); + return context; } }