This is an automated email from the ASF dual-hosted git repository. rzo1 pushed a commit to branch TOMEE-4703 in repository https://gitbox.apache.org/repos/asf/tomee.git
commit fa40d15924fee3c3a028fb39fb81f99159aa7456 Author: Richard Zowalla <[email protected]> AuthorDate: Tue Sep 8 21:25:52 2026 +0200 TOMEE-4703 fix context capture in currentContextExecutor and per-task CUTask.Context Two defects adjacent to the eager capture, both about a context being taken or held in the wrong place. currentContextExecutor() was "command -> contextualRunnable(command).run()", so the context was captured inside execute(), on whatever thread submitted a command. ContextService specifies an executor "that runs tasks on the same thread from which execute is invoked but with context that is captured from the thread that invokes currentContextExecutor". Capture the snapshot when the executor is created and reuse it, via a CUTask constructor that takes a snapshot that has already been captured. CUTask held one CUTask.Context per task, created in the constructor, but a contextual proxy runs its task more than once and may run it on several threads at the same time. Concurrent invocations then shared one Context: they raced on its "previous" field and its exit task list, and threw "Can't enter a context twice" whenever the callers were inside a context of their own. Create the Context per invocation instead. --- .../openejb/threads/impl/ContextServiceImpl.java | 6 +- .../apache/openejb/threads/task/CURunnable.java | 5 ++ .../org/apache/openejb/threads/task/CUTask.java | 23 +++++--- .../openejb/threads/ThreadContextCaptureTest.java | 69 ++++++++++++++++++++++ 4 files changed, 95 insertions(+), 8 deletions(-) diff --git a/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ContextServiceImpl.java b/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ContextServiceImpl.java index a10dd44016..2460fe5cbb 100644 --- a/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ContextServiceImpl.java +++ b/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ContextServiceImpl.java @@ -26,6 +26,7 @@ import jakarta.enterprise.concurrent.spi.ThreadContextSnapshot; import org.apache.openejb.OpenEJBRuntimeException; import org.apache.openejb.resource.thread.ManagedExecutorServiceImplFactory; import org.apache.openejb.threads.future.CUCompletableFuture; +import org.apache.openejb.threads.task.CURunnable; import org.apache.openejb.threads.task.CUTask; import org.apache.openejb.util.LogCategory; import org.apache.openejb.util.Logger; @@ -149,7 +150,10 @@ public class ContextServiceImpl implements ContextService, Serializable { @Override public Executor currentContextExecutor() { - return command -> contextualRunnable(command).run(); + // "context that is captured from the thread that invokes currentContextExecutor" - the + // capture belongs here, not in execute(), which runs on whatever thread submits a command + final Snapshot snapshot = snapshot(null); + return command -> new CURunnable(command, this, snapshot).run(); } @Override diff --git a/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CURunnable.java b/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CURunnable.java index a47b1bc836..ed04883b13 100644 --- a/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CURunnable.java +++ b/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CURunnable.java @@ -28,6 +28,11 @@ public class CURunnable extends CUTask<Void> implements Runnable { super(task, ContextServiceImplFactory.newPropagateEverythingContextService()); delegate = task; } + public CURunnable(final Runnable task, final ContextServiceImpl contextService, final ContextServiceImpl.Snapshot snapshot) { + super(task, contextService, snapshot); + this.delegate = task; + } + public CURunnable(final Runnable task, final ContextServiceImpl contextService) { super(task, contextService); delegate = task; diff --git a/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CUTask.java b/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CUTask.java index c73813a262..30e9fbfd0f 100644 --- a/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CUTask.java +++ b/container/openejb-core/src/main/java/org/apache/openejb/threads/task/CUTask.java @@ -44,18 +44,24 @@ public abstract class CUTask<T> extends ManagedTaskListenerTask implements Compa protected final ContextServiceImpl contextService; private final ContextServiceImpl.Snapshot snapshot; private final Object[] containerListenerStates; - private final Context initialContext; public CUTask(final Object task, final ContextServiceImpl contextService) { - this(task, contextService, null); + this(task, contextService, (Map<String, String>) null); } public CUTask(final Object task, final ContextServiceImpl contextService, Map<String, String> props) { + this(task, contextService, contextService.snapshot(props)); + } + + /** + * Uses a context snapshot that was captured earlier, on the thread the context has to be taken + * from. {@link ContextService#currentContextExecutor()} needs this: it has to capture when the + * executor is created, not when a task is handed to it. + */ + public CUTask(final Object task, final ContextServiceImpl contextService, final ContextServiceImpl.Snapshot snapshot) { super(task); this.contextService = contextService; - - snapshot = contextService.snapshot(props); - initialContext = new Context(); + this.snapshot = snapshot; if (CONTAINER_LISTENERS.length > 0) { containerListenerStates = new Object[CONTAINER_LISTENERS.length]; for (int i = 0; i < CONTAINER_LISTENERS.length; i++) { @@ -67,7 +73,10 @@ public abstract class CUTask<T> extends ManagedTaskListenerTask implements Compa } protected T invoke(final Callable<T> call) throws Exception { - initialContext.enter(); + // one per invocation, not one per task: a contextual proxy runs its task more than once, + // and may run it on several threads at the same time + final Context invocationContext = new Context(); + invocationContext.enter(); final Object[] oldStates; if (CONTAINER_LISTENERS.length > 0) { oldStates = new Object[CONTAINER_LISTENERS.length]; @@ -105,7 +114,7 @@ public abstract class CUTask<T> extends ManagedTaskListenerTask implements Compa if (contextService != null && state != null) { contextService.exit(state); } - initialContext.exit(); + invocationContext.exit(); } } } diff --git a/container/openejb-core/src/test/java/org/apache/openejb/threads/ThreadContextCaptureTest.java b/container/openejb-core/src/test/java/org/apache/openejb/threads/ThreadContextCaptureTest.java index b32ccfc482..aacd0fc27d 100644 --- a/container/openejb-core/src/test/java/org/apache/openejb/threads/ThreadContextCaptureTest.java +++ b/container/openejb-core/src/test/java/org/apache/openejb/threads/ThreadContextCaptureTest.java @@ -36,6 +36,8 @@ import java.net.URLClassLoader; import java.util.ArrayList; import java.util.List; import java.util.concurrent.Callable; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -80,6 +82,16 @@ public class ThreadContextCaptureTest { facade.checkInvocationContextNotPropagated(); } + @Test + public void currentContextExecutorCapturesWhereItWasCreated() throws Exception { + facade.checkCurrentContextExecutorCaptureTime(); + } + + @Test + public void aContextualProxyCanRunOnSeveralThreadsAtOnce() throws Exception { + facade.checkContextualProxyIsReusable(); + } + public static class BeforeCapture { } @@ -161,6 +173,63 @@ public class ThreadContextCaptureTest { seen.get(1, TimeUnit.MINUTES)); } + public void checkCurrentContextExecutorCaptureTime() throws Exception { + final ThreadContext caller = ThreadContext.getThreadContext(); + assertNotNull(caller); + + caller.set(BeforeCapture.class, new BeforeCapture()); + final Executor executor = contextService.currentContextExecutor(); + caller.set(AfterCapture.class, new AfterCapture()); + + try { + final Object[] seen = new Object[2]; + executor.execute(() -> { + final ThreadContext taskContext = ThreadContext.getThreadContext(); + seen[0] = taskContext.get(BeforeCapture.class); + seen[1] = taskContext.get(AfterCapture.class); + }); + + assertNotNull("state present when the executor was created must be propagated", seen[0]); + assertNull("state added afterwards must not be, the capture belongs to" + + " currentContextExecutor(), not to execute()", seen[1]); + } finally { + caller.remove(BeforeCapture.class); + caller.remove(AfterCapture.class); + } + } + + public void checkContextualProxyIsReusable() throws Exception { + final int threads = 2; + final int rounds = 200; + final Callable<Boolean> contextual = contextService.contextualCallable( + () -> ThreadContext.getThreadContext() != null); + + // one proxy, called by two threads at the same time, over and over. The callers are + // managed tasks themselves, so each already has a CUTask context of its own - which is + // what makes a task scoped CUTask.Context observable. The SPI requires a snapshot to be + // applicable "to any number of threads, including concurrently". + final CyclicBarrier barrier = new CyclicBarrier(threads); + final List<Future<Boolean>> futures = new ArrayList<>(); + for (int i = 0; i < threads; i++) { + futures.add(executorService.submit(() -> { + for (int round = 0; round < rounds; round++) { + barrier.await(1, TimeUnit.MINUTES); + if (!contextual.call()) { + return false; + } + } + return true; + })); + } + + for (final Future<Boolean> future : futures) { + assertTrue(future.get(1, TimeUnit.MINUTES)); + } + + // and again afterwards, sequentially + assertTrue(contextual.call()); + } + public void submitWhileMutating() throws Exception { final ThreadContext caller = ThreadContext.getThreadContext(); assertNotNull(caller);
