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 4a521c2baaf186bd0d813913e91d5a56d05250a8 Author: Richard Zowalla <[email protected]> AuthorDate: Tue Sep 8 21:20:49 2026 +0200 TOMEE-4703 capture thread context eagerly on the submitting thread ApplicationThreadContextProvider stored a live reference to the submitting thread's ThreadContext and copied it later, in begin(), on the thread running the task. A ThreadContext is thread confined, so that copy raced with its owner (TOMEE-4699). Capture an immutable snapshot in currentContext() instead, and build a fresh ThreadContext from it per begin(). The snapshot cannot hold a pre-built ThreadContext: enter() writes into its argument and rejects a context that is already entered, so such a snapshot could never be applied twice, let alone concurrently. The capture leaves out context data bound to the capturing thread's in-flight invocation. InvocationContext belongs to the interceptor chain the caller is still inside, and its context data is an unsynchronized map that BaseContext.getContextData() hands to application code; DestroyContext refers back to the captured context and would keep it alive for the life of the capture. Both are re-established on the thread the context is applied to. Also formats the context data outside the map's monitor, so application hashCode() implementations no longer run under a lock taken on every invocation. --- .../org/apache/openejb/core/ThreadContext.java | 110 +++++++++++- .../impl/ApplicationThreadContextProvider.java | 21 ++- .../openejb/threads/ThreadContextCaptureTest.java | 184 +++++++++++++++++++++ 3 files changed, 301 insertions(+), 14 deletions(-) diff --git a/container/openejb-core/src/main/java/org/apache/openejb/core/ThreadContext.java b/container/openejb-core/src/main/java/org/apache/openejb/core/ThreadContext.java index 5ef55eb8a8..9e270f24bc 100644 --- a/container/openejb-core/src/main/java/org/apache/openejb/core/ThreadContext.java +++ b/container/openejb-core/src/main/java/org/apache/openejb/core/ThreadContext.java @@ -25,6 +25,7 @@ import org.apache.openejb.util.Logger; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; @@ -151,13 +152,104 @@ public class ThreadContext { this.currentOperation = operation; } + /** + * Copies the given context. A {@link ThreadContext} is thread confined, so this must only be + * called on the thread that owns <code>that</code> - otherwise the copy races with its owner. + * To hand a context to another thread use {@link #capture()} instead. + */ public ThreadContext(final ThreadContext that) { this.beanContext = that.beanContext; this.primaryKey = that.primaryKey; - this.data.putAll(that.data); + synchronized (that.data) { + this.data.putAll(that.data); + } this.oldClassLoader = that.oldClassLoader; } + /** + * Captures the calling thread's context, if any, as an immutable value that can safely be handed + * to other threads. This has to be called on the thread owning the context, see + * {@link Capture} for the reasoning. + * + * @return an immutable capture of the current thread's context, or <code>null</code> if this + * thread has no context entered + */ + public static Capture capture() { + final ThreadContext current = threadStorage.get(); + return current == null ? null : new Capture(current); + } + + /** + * An immutable capture of the propagatable state of a {@link ThreadContext}, taken on the thread + * owning that context. + * <p> + * A capture carries the same state the {@link ThreadContext#ThreadContext(ThreadContext) copy + * constructor} carries - bean context, primary key and the context data - and deliberately does + * not carry per-thread lifecycle state such as the class loader to restore, the entered flag or + * the current operation. + * <p> + * Instances are immutable and can therefore be applied to any number of threads, including + * concurrently. {@link #newThreadContext()} hands every caller its own mutable + * {@link ThreadContext}, because {@link ThreadContext#enter(ThreadContext)} mutates the context + * it is given and rejects a context that is already entered. + */ + public static final class Capture { + + /** + * Context data bound to the capturing thread's in-flight invocation, by class name so that + * this class does not have to see the types. These are deliberately not propagated: + * <ul> + * <li><code>InvocationContext</code> belongs to the interceptor chain the capturing thread + * is still inside. It is single use, and its context data is an unsynchronized map that + * {@link BaseContext#getContextData()} hands straight to application code.</li> + * <li><code>DestroyContext</code> refers back to the captured context, so propagating it + * would keep that context alive for as long as the capture lives. The listener that + * created it creates another one on the thread the context is applied to.</li> + * </ul> + */ + private static final Set<String> NON_PROPAGATED = Set.of( + "jakarta.interceptor.InvocationContext", + "org.apache.openejb.cdi.RequestScopedThreadContextListener$DestroyContext"); + + private final BeanContext beanContext; + private final Object primaryKey; + private final Map<Class, Object> data; + + private Capture(final ThreadContext that) { + this.beanContext = that.beanContext; + this.primaryKey = that.primaryKey; + + final Map<Class, Object> copy = new HashMap<>(); + synchronized (that.data) { + for (final Map.Entry<Class, Object> entry : that.data.entrySet()) { + if (!NON_PROPAGATED.contains(entry.getKey().getName())) { + copy.put(entry.getKey(), entry.getValue()); + } + } + } + this.data = Collections.unmodifiableMap(copy); + } + + /** + * @return a new, mutable {@link ThreadContext} holding the captured state, ready to be + * passed to {@link ThreadContext#enter(ThreadContext)} by the calling thread + */ + public ThreadContext newThreadContext() { + final ThreadContext context = new ThreadContext(beanContext, primaryKey); + context.data.putAll(data); + return context; + } + + @Override + public String toString() { + return "ThreadContext.Capture{" + + "beanContext=" + beanContext.getId() + + ", primaryKey=" + primaryKey + + ", data=" + dataToString(data) + + '}'; + } + } + public BeanContext getBeanContext() { return beanContext; } @@ -226,8 +318,7 @@ public class ThreadContext { return "ThreadContext{" + "beanContext=" + beanContext.getId() + ", primaryKey=" + primaryKey + - ", data(" + data.size() + - ")=" + dataToString(data) + + ", data=" + dataToString(data) + ", oldClassLoader=" + oldClassLoader + ", currentOperation=" + currentOperation + ", invokedInterface=" + invokedInterface + @@ -236,10 +327,17 @@ public class ThreadContext { '}'; } - private String dataToString(final Map<Class, Object> data) { - return data.entrySet().stream() + private static String dataToString(final Map<Class, Object> data) { + // data may be a synchronized map, iterating it needs the map's monitor - see TOMEE-4699. + // Copy under the monitor and format outside it: the values are application objects and + // their hashCode() must not run while holding a lock taken on every invocation. + final Map<Class, Object> copy; + synchronized (data) { + copy = new HashMap<>(data); + } + + return "(" + copy.size() + ")=" + copy.entrySet().stream() .map(entry -> entry.getKey() + "=" + (entry.getValue() == null ? "null" : entry.getValue().hashCode())) .collect(Collectors.joining(", ")); - } } diff --git a/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ApplicationThreadContextProvider.java b/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ApplicationThreadContextProvider.java index 600b42c4d1..eab46c4dc8 100755 --- a/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ApplicationThreadContextProvider.java +++ b/container/openejb-core/src/main/java/org/apache/openejb/threads/impl/ApplicationThreadContextProvider.java @@ -39,7 +39,9 @@ public class ApplicationThreadContextProvider implements ThreadContextProvider, return clearedContext(props); } - return new ApplicationThreadContextSnapshot(appContext.getId(), ThreadContext.getThreadContext()); + // capture eagerly, on the thread owning the ThreadContext - see TOMEE-4699. A ThreadContext is + // thread confined, reading it later from the thread running the task races with its owner. + return new ApplicationThreadContextSnapshot(appContext.getId(), ThreadContext.capture()); } @Override @@ -54,11 +56,11 @@ public class ApplicationThreadContextProvider implements ThreadContextProvider, public static class ApplicationThreadContextSnapshot implements ThreadContextSnapshot, Serializable { private final Object appId; - private final ThreadContext threadContext; + private final ThreadContext.Capture capturedThreadContext; - public ApplicationThreadContextSnapshot(final Object appId, final ThreadContext threadContext) { + public ApplicationThreadContextSnapshot(final Object appId, final ThreadContext.Capture capturedThreadContext) { this.appId = appId; - this.threadContext = threadContext; + this.capturedThreadContext = capturedThreadContext; } @Override @@ -71,9 +73,12 @@ public class ApplicationThreadContextProvider implements ThreadContextProvider, final ClassLoader oldCl = Thread.currentThread().getContextClassLoader(); Thread.currentThread().setContextClassLoader(appContext.getClassLoader()); - // Don't touch ThreadContext if it is already correct or none was captured - boolean changeThreadContext = threadContext != null && threadContext != ThreadContext.getThreadContext(); - ThreadContext oldThreadContext = changeThreadContext ? ThreadContext.enter(new ThreadContext(threadContext)) : null; + // Don't touch ThreadContext if none was captured. If one was captured, always enter a fresh + // copy of it: this snapshot may be applied to any number of threads, including concurrently, + // so the capture itself must never be entered - ThreadContext.enter mutates its argument. + final boolean changeThreadContext = capturedThreadContext != null; + final ThreadContext oldThreadContext = + changeThreadContext ? ThreadContext.enter(capturedThreadContext.newThreadContext()) : null; return new ApplicationThreadContextRestorer(oldCl, oldThreadContext, changeThreadContext); } @@ -81,7 +86,7 @@ public class ApplicationThreadContextProvider implements ThreadContextProvider, public String toString() { return "ApplicationThreadContextSnapshot@" + System.identityHashCode(this) + "{appId=" + appId + - "{threadContext=" + threadContext + + "{capturedThreadContext=" + capturedThreadContext + '}'; } 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 new file mode 100644 index 0000000000..b32ccfc482 --- /dev/null +++ b/container/openejb-core/src/test/java/org/apache/openejb/threads/ThreadContextCaptureTest.java @@ -0,0 +1,184 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.openejb.threads; + +import org.apache.openejb.core.ThreadContext; +import org.apache.openejb.jee.EnterpriseBean; +import org.apache.openejb.jee.SingletonBean; +import org.apache.openejb.junit.ApplicationComposer; +import org.apache.openejb.testing.Module; +import org.apache.openejb.threads.impl.ContextServiceImpl; +import org.junit.Test; +import org.junit.runner.RunWith; + +import jakarta.annotation.Resource; +import jakarta.ejb.EJB; +import jakarta.ejb.Singleton; +import jakarta.enterprise.concurrent.ContextService; +import jakarta.enterprise.concurrent.ManagedExecutorService; +import jakarta.interceptor.InvocationContext; +import java.net.URL; +import java.net.URLClassLoader; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +/** + * TOMEE-4699: the ThreadContext of the submitting thread must be captured by value, on the submitting + * thread, and not read again later from the thread running the task. + */ +@RunWith(ApplicationComposer.class) +public class ThreadContextCaptureTest { + @Module + public EnterpriseBean bean() { + return new SingletonBean(CaptureFacade.class).localBean(); + } + + @EJB + private CaptureFacade facade; + + @Test + public void contextIsCapturedWhenTheContextualProxyIsCreated() throws Exception { + facade.checkCaptureTime(); + } + + @Test + public void submittingWhileMutatingTheCallerContextDoesNotFail() throws Exception { + facade.submitWhileMutating(); + } + + @Test + public void inlineExecutionLeavesTheCallersClassLoaderInPlace() throws Exception { + facade.checkInlineClassLoader(); + } + + @Test + public void theCallersInvocationContextIsNotPropagated() throws Exception { + facade.checkInvocationContextNotPropagated(); + } + + public static class BeforeCapture { + } + + public static class AfterCapture { + } + + public static class Churn { + } + + @Singleton + public static class CaptureFacade { + @Resource + private ContextService contextService; + + @Resource + private ManagedExecutorService executorService; + + public void checkCaptureTime() throws Exception { + final ThreadContext caller = ThreadContext.getThreadContext(); + assertNotNull(caller); + + caller.set(BeforeCapture.class, new BeforeCapture()); + + final Callable<Object[]> contextual = contextService.contextualCallable(() -> { + final ThreadContext taskContext = ThreadContext.getThreadContext(); + assertNotNull(taskContext); + return new Object[]{taskContext.get(BeforeCapture.class), taskContext.get(AfterCapture.class)}; + }); + + // the caller keeps mutating its own, thread confined ThreadContext after the capture + caller.set(AfterCapture.class, new AfterCapture()); + + final ExecutorService plain = Executors.newSingleThreadExecutor(); + try { + final Object[] seen = plain.submit(contextual).get(1, TimeUnit.MINUTES); + assertNotNull("state present at capture time must be propagated", seen[0]); + assertNull("state added after the capture must not leak into the task", seen[1]); + } finally { + plain.shutdownNow(); + caller.remove(BeforeCapture.class); + caller.remove(AfterCapture.class); + } + } + + public void checkInlineClassLoader() { + final Thread thread = Thread.currentThread(); + final ClassLoader original = thread.getContextClassLoader(); + + // take the snapshot under the real thread context class loader - the application is + // resolved from it, so a foreign loader here would yield a cleared snapshot instead + final ContextServiceImpl impl = ContextServiceImpl.class.cast(contextService); + final ContextServiceImpl.Snapshot snapshot = impl.snapshot(null); + + // a loader that is neither the application's nor the bean's, so that a restore to + // either of those is visible here + final ClassLoader marker = new URLClassLoader(new URL[0], original); + thread.setContextClassLoader(marker); + try { + impl.exit(impl.enter(snapshot)); + + assertSame("applying and restoring a context must give the thread back as it was", + marker, thread.getContextClassLoader()); + } finally { + thread.setContextClassLoader(original); + } + } + + public void checkInvocationContextNotPropagated() throws Exception { + final ThreadContext caller = ThreadContext.getThreadContext(); + assertNotNull(caller); + // the interceptor stack put this there on the way into this method + assertNotNull("precondition: the caller is inside an invocation", + caller.get(InvocationContext.class)); + + final Future<InvocationContext> seen = executorService.submit( + () -> ThreadContext.getThreadContext().get(InvocationContext.class)); + + assertNull("the caller's in-flight InvocationContext must not be handed to the task", + seen.get(1, TimeUnit.MINUTES)); + } + + public void submitWhileMutating() throws Exception { + final ThreadContext caller = ThreadContext.getThreadContext(); + assertNotNull(caller); + + final List<Future<Boolean>> futures = new ArrayList<>(); + for (int i = 0; i < 200; i++) { + futures.add(executorService.submit(() -> ThreadContext.getThreadContext() != null)); + + // this is what an interceptor stack does to the caller's context while the tasks start + for (int j = 0; j < 200; j++) { + caller.set(Churn.class, new Churn()); + caller.remove(Churn.class); + } + } + + for (final Future<Boolean> future : futures) { + assertTrue(future.get(1, TimeUnit.MINUTES)); + } + } + } +}
