From 3478a702e579b12d412ab17838f760422906722b Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Thu, 29 Feb 2024 20:43:31 +0000 Subject: [PATCH] Improve concurrent handling of result in WebAsyncManager 1. Use state transitions 2. Increase synchronized scope in setConcurrentResultAndDispatch See gh-32342 --- .../request/async/WebAsyncManager.java | 143 ++++++++++++------ 1 file changed, 97 insertions(+), 46 deletions(-) diff --git a/spring-web/src/main/java/org/springframework/web/context/request/async/WebAsyncManager.java b/spring-web/src/main/java/org/springframework/web/context/request/async/WebAsyncManager.java index 998be73e57..63a3268b66 100644 --- a/spring-web/src/main/java/org/springframework/web/context/request/async/WebAsyncManager.java +++ b/spring-web/src/main/java/org/springframework/web/context/request/async/WebAsyncManager.java @@ -22,6 +22,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.Callable; import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicReference; import javax.servlet.http.HttpServletRequest; @@ -99,12 +100,7 @@ public final class WebAsyncManager { @Nullable private volatile Object[] concurrentResultContext; - /* - * Whether the concurrentResult is an error. If such errors remain unhandled, some - * Servlet containers will call AsyncListener#onError at the end, after the ASYNC - * and/or the ERROR dispatch (Boot's case), and we need to ignore those. - */ - private volatile boolean errorHandlingInProgress; + private final AtomicReference state = new AtomicReference<>(State.NOT_STARTED); private final Map callableInterceptors = new LinkedHashMap<>(); @@ -257,6 +253,12 @@ public final class WebAsyncManager { * {@linkplain #getConcurrentResultContext() concurrentResultContext}. */ public void clearConcurrentResult() { + if (!this.state.compareAndSet(State.RESULT_SET, State.NOT_STARTED)) { + if (logger.isDebugEnabled()) { + logger.debug("Unexpected call to clear: [" + this.state.get() + "]"); + } + return; + } synchronized (WebAsyncManager.this) { this.concurrentResult = RESULT_NONE; this.concurrentResultContext = null; @@ -297,6 +299,11 @@ public final class WebAsyncManager { Assert.notNull(webAsyncTask, "WebAsyncTask must not be null"); Assert.state(this.asyncWebRequest != null, "AsyncWebRequest must not be null"); + if (!this.state.compareAndSet(State.NOT_STARTED, State.ASYNC_PROCESSING)) { + throw new IllegalStateException( + "Unexpected call to startCallableProcessing: [" + this.state.get() + "]"); + } + Long timeout = webAsyncTask.getTimeout(); if (timeout != null) { this.asyncWebRequest.setTimeout(timeout); @@ -320,7 +327,7 @@ public final class WebAsyncManager { this.asyncWebRequest.addTimeoutHandler(() -> { if (logger.isDebugEnabled()) { - logger.debug("Async request timeout for " + formatUri(this.asyncWebRequest)); + logger.debug("Servlet container timeout notification for " + formatUri(this.asyncWebRequest)); } Object result = interceptorChain.triggerAfterTimeout(this.asyncWebRequest, callable); if (result != CallableProcessingInterceptor.RESULT_NONE) { @@ -329,14 +336,12 @@ public final class WebAsyncManager { }); this.asyncWebRequest.addErrorHandler(ex -> { - if (!this.errorHandlingInProgress) { - if (logger.isDebugEnabled()) { - logger.debug("Async request error for " + formatUri(this.asyncWebRequest) + ": " + ex); - } - Object result = interceptorChain.triggerAfterError(this.asyncWebRequest, callable, ex); - result = (result != CallableProcessingInterceptor.RESULT_NONE ? result : ex); - setConcurrentResultAndDispatch(result); + if (logger.isDebugEnabled()) { + logger.debug("Servlet container error notification for " + formatUri(this.asyncWebRequest) + ": " + ex); } + Object result = interceptorChain.triggerAfterError(this.asyncWebRequest, callable, ex); + result = (result != CallableProcessingInterceptor.RESULT_NONE ? result : ex); + setConcurrentResultAndDispatch(result); }); this.asyncWebRequest.addCompletionHandler(() -> @@ -388,33 +393,40 @@ public final class WebAsyncManager { } private void setConcurrentResultAndDispatch(@Nullable Object result) { - synchronized (WebAsyncManager.this) { - if (this.concurrentResult != RESULT_NONE) { - return; - } - this.concurrentResult = result; - this.errorHandlingInProgress = (result instanceof Throwable); - } - Assert.state(this.asyncWebRequest != null, "AsyncWebRequest must not be null"); - if (this.asyncWebRequest.isAsyncComplete()) { - if (logger.isDebugEnabled()) { - logger.debug("Async result set but request already complete: " + formatUri(this.asyncWebRequest)); - } - return; - } - - if (result instanceof Exception) { - if (disconnectedClientHelper.checkAndLogClientDisconnectedException((Exception) result)) { + synchronized (WebAsyncManager.this) { + if (!this.state.compareAndSet(State.ASYNC_PROCESSING, State.RESULT_SET)) { + if (logger.isDebugEnabled()) { + logger.debug("Async result already set: " + + "[" + this.state.get() + "], ignored result: " + result + + " for " + formatUri(this.asyncWebRequest)); + } return; } - } - if (logger.isDebugEnabled()) { - logger.debug("Async " + (this.errorHandlingInProgress ? "error" : "result set") + - ", dispatch to " + formatUri(this.asyncWebRequest)); + this.concurrentResult = result; + if (logger.isDebugEnabled()) { + logger.debug("Async result set to: " + result + " for " + formatUri(this.asyncWebRequest)); + } + + if (result instanceof Exception) { + if (disconnectedClientHelper.checkAndLogClientDisconnectedException((Exception) result)) { + return; + } + } + + if (this.asyncWebRequest.isAsyncComplete()) { + if (logger.isDebugEnabled()) { + logger.debug("Async request already completed for " + formatUri(this.asyncWebRequest)); + } + return; + } + + if (logger.isDebugEnabled()) { + logger.debug("Performing async dispatch for " + formatUri(this.asyncWebRequest)); + } + this.asyncWebRequest.dispatch(); } - this.asyncWebRequest.dispatch(); } /** @@ -437,6 +449,11 @@ public final class WebAsyncManager { Assert.notNull(deferredResult, "DeferredResult must not be null"); Assert.state(this.asyncWebRequest != null, "AsyncWebRequest must not be null"); + if (!this.state.compareAndSet(State.NOT_STARTED, State.ASYNC_PROCESSING)) { + throw new IllegalStateException( + "Unexpected call to startDeferredResultProcessing: [" + this.state.get() + "]"); + } + Long timeout = deferredResult.getTimeoutValue(); if (timeout != null) { this.asyncWebRequest.setTimeout(timeout); @@ -450,6 +467,9 @@ public final class WebAsyncManager { final DeferredResultInterceptorChain interceptorChain = new DeferredResultInterceptorChain(interceptors); this.asyncWebRequest.addTimeoutHandler(() -> { + if (logger.isDebugEnabled()) { + logger.debug("Servlet container timeout notification for " + formatUri(this.asyncWebRequest)); + } try { interceptorChain.triggerAfterTimeout(this.asyncWebRequest, deferredResult); } @@ -459,16 +479,17 @@ public final class WebAsyncManager { }); this.asyncWebRequest.addErrorHandler(ex -> { - if (!this.errorHandlingInProgress) { - try { - if (!interceptorChain.triggerAfterError(this.asyncWebRequest, deferredResult, ex)) { - return; - } - deferredResult.setErrorResult(ex); - } - catch (Throwable interceptorEx) { - setConcurrentResultAndDispatch(interceptorEx); + if (logger.isDebugEnabled()) { + logger.debug("Servlet container error notification for " + formatUri(this.asyncWebRequest)); + } + try { + if (!interceptorChain.triggerAfterError(this.asyncWebRequest, deferredResult, ex)) { + return; } + deferredResult.setErrorResult(ex); + } + catch (Throwable interceptorEx) { + setConcurrentResultAndDispatch(interceptorEx); } }); @@ -494,10 +515,13 @@ public final class WebAsyncManager { synchronized (WebAsyncManager.this) { this.concurrentResult = RESULT_NONE; this.concurrentResultContext = processingContext; - this.errorHandlingInProgress = false; } Assert.state(this.asyncWebRequest != null, "AsyncWebRequest must not be null"); + if (logger.isDebugEnabled()) { + logger.debug("Started async request for " + formatUri(this.asyncWebRequest)); + } + this.asyncWebRequest.startAsync(); if (logger.isDebugEnabled()) { logger.debug("Started async request"); @@ -509,4 +533,31 @@ public final class WebAsyncManager { return (request != null ? request.getRequestURI() : "servlet container"); } + + /** + * Represents a state for {@link WebAsyncManager} to be in. + * 

+	 *        NOT_STARTED <------+
+	 *             |             |
+	 *             v             |
+	 *      ASYNC_PROCESSING     |
+	 *             |             |
+	 *             v             |
+	 *         RESULT_SET -------+
+	 *
+ * @since 5.3.33 + */ + private enum State { + + /** No async processing in progress. */ + NOT_STARTED, + + /** Async handling has started, but the result hasn't been set yet. */ + ASYNC_PROCESSING, + + /** The result is set, and an async dispatch was performed, unless there is a network error. */ + RESULT_SET + + } + }