hk2166 commented on code in PR #16786: URL: https://github.com/apache/grails-core/pull/16786#discussion_r4423675321
########## grails-async/plugin/src/main/groovy/org/grails/plugins/web/async/TaskStreams.groovy: ########## @@ -0,0 +1,643 @@ +/* + * 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 + * + * https://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.grails.plugins.web.async + +import groovy.transform.CompileStatic +import groovy.transform.PackageScope + +import jakarta.servlet.ReadListener +import jakarta.servlet.ServletInputStream +import jakarta.servlet.ServletOutputStream +import jakarta.servlet.WriteListener + +/** + * The guarded body streams and readers of a task's request and the guarded output of its response: each calls the + * live one under the read lock of the request's {@link TaskRequestLifecycle}, until the request completes, and + * then throws an {@link IllegalStateException}. + */ +@PackageScope +@CompileStatic +final class TaskStreams { + + private TaskStreams() { + } + + /** + * Guards what a live request or response returned. + * + * @param lifecycle the lifecycle of the request + * @param live a ServletInputStream, BufferedReader, ServletOutputStream or PrintWriter + * @param what what returned it, for the failure + * @return the guarded one, attached to the lifecycle + */ + static Object guard(TaskRequestLifecycle lifecycle, Object live, String what) { + Object guarded + TaskRequestLifecycle.Detachable detachable + if (live instanceof ServletInputStream) { + TaskInputStream stream = new TaskInputStream(lifecycle, (ServletInputStream) live, what) + guarded = stream + detachable = stream + } + else if (live instanceof BufferedReader) { + TaskReader reader = new TaskReader(lifecycle, (BufferedReader) live, what) + guarded = reader + detachable = reader + } + else if (live instanceof ServletOutputStream) { + TaskOutputStream stream = new TaskOutputStream(lifecycle, (ServletOutputStream) live, what) + guarded = stream + detachable = stream + } + else { + TaskWriter writer = new TaskWriter(lifecycle, (PrintWriter) live, what) + guarded = new TaskPrintWriter(writer) + detachable = writer + } + lifecycle.attach(detachable) + return guarded + } + + private static final class TaskInputStream extends ServletInputStream implements TaskRequestLifecycle.Detachable { + + private final TaskRequestLifecycle lifecycle + private final String what + private volatile ServletInputStream live + + private TaskInputStream(TaskRequestLifecycle lifecycle, ServletInputStream live, String what) { + this.lifecycle = lifecycle + this.live = live + this.what = what + } + + @Override + void detach() { + live = null + } + + private ServletInputStream enter() { + return lifecycle.enter(live, "$what body".toString()) + } + + // After a call made without the lock, which the request completing may have overtaken: whatever the call did + // or threw, it reports that the request has completed + private void checkLive() { + if (lifecycle.completed) { + throw TaskRequestLifecycle.completedFailure("$what body".toString()) + } + } + + @Override + int read() throws IOException { Review Comment: Fixed in 6ed3af27ec. A read of the body no longer holds the lifecycle lock once a read listener has been set through a task's stream. In that mode `CoyoteInputStream.read()` asks `isReady()`, which can take the connection's lock. Such reads, and `getPart`/`getParts` (which read the body), are checked before and after instead. If completion overtook the call, the call throws the completed `IllegalStateException`, whatever it returned. Without a read listener, a blocking read still holds the lock. Completion therefore waits for a read in progress before Tomcat recycles the `InputBuffer` and swallows the rest of the body on the same socket. A review I ran before pushing pointed out that a read still running then could otherwise pick up the next request's bytes on a reused connection. The same reasoning is applied to everything that runs into Tomcat's own locks: - **Without the lock:** `isReady()` on the body, `setReadListener`/`setWriteListener`, and `startAsync`. `startAsync` waits for the synchronized `AsyncStateMachine`, which `asyncPostProcess` holds while firing `onComplete`/`requestDestroyed`. - **Under the lock:** writes, flushes and closes of the output, and output `isReady()`. These only raise `CLIENT_FLUSH`/`CLOSE`/`NB_WRITE_INTEREST`, which take no lock the completing thread holds. - **A backstop:** completion waits at most 60 seconds for a call in flight, then detaches and logs a warning. That covers a call that waits for a monitor the completing thread holds, such as a failing flush reporting `asyncError`. 60 seconds is as long as Tomcat lets a blocking read or write wait by default, and `grails.async.requestCompletionWait` changes it. Tests (the first three fail on 4694264b0c): - `once a read listener was set through a task stream, #call does not hold off the request completing`, for reading the body and the parts. - `startAsync does not hold off the request completing...`. - `a read of the body a task is making when the request completes finishes first`. - `a write a task is making to the response when the request completes finishes first`. - `completing the request gives up waiting, after the wait the listener was given, for a call that waits for what the completing thread holds`. ########## grails-async/plugin/src/main/groovy/org/grails/plugins/web/async/TaskRequestLifecycle.groovy: ########## @@ -0,0 +1,289 @@ +/* + * 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 + * + * https://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.grails.plugins.web.async + +import java.lang.ref.WeakReference +import java.util.concurrent.locks.ReentrantReadWriteLock +import java.util.function.Supplier + +import groovy.transform.CompileStatic +import groovy.transform.PackageScope + +import jakarta.servlet.AsyncEvent +import jakarta.servlet.AsyncListener +import jakarta.servlet.ServletRequest +import jakarta.servlet.ServletRequestWrapper + +import org.springframework.web.context.request.RequestAttributes + +import org.grails.web.servlet.mvc.GrailsWebRequest + +/** + * The life of a request as the tasks started during it see it: live until the request completes, then detached. + * + * <p>Every object a task is given that reaches into the request, its {@link TaskRequest}, {@link TaskResponse} and + * the streams, readers and writers they return, calls the live object only while it holds the read lock and the + * request has not completed. Completing takes the write lock, so a call a task is making finishes first, and then + * releases the live objects, so every later call throws an {@link IllegalStateException}. Nothing is read from the + * request as it completes.</p> + * + * <p>The request completes when the container is done with it, as {@link TaskRequestListener} tells, after any + * forward, include or asynchronous processing within it. Without the listener, such as in a unit test, it completes + * with the web request the first of its tasks was started during.</p> + */ +@PackageScope +@CompileStatic +final class TaskRequestLifecycle implements AsyncListener { + + private static final String LIFECYCLE = TaskRequestLifecycle.name + '.LIFECYCLE' + private static final String COMPLETION_CALLBACK = TaskRequestLifecycle.name + '.COMPLETION' + + private static final int PURGE_THRESHOLD = 32 + + private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock() + private final Object monitor = new Object() + // What calls into the request, held weakly, so what a finished task dropped is not kept until the request + // completes. Created with the first task, so a request without tasks allocates nothing more + private List<WeakReference<Detachable>> attached + private int purgeAt = PURGE_THRESHOLD + // Whether the container tells the listener when it is done with the request + private final boolean observed + private volatile boolean completed + + private TaskRequestLifecycle(boolean observed) { + this.observed = observed + } + + /** + * The lifecycle of the request of a web request, shared by every task started during the request, whichever + * request of it, such as the request of a forward, each was given. It has to be called while the web request + * is the current one, when the task is submitted. Once the container is done with the request, it is the + * completed lifecycle, so a task started then, such as during an error page rendered after that, runs without + * the request. + * + * @param webRequest the web request a task is started during + * @return the lifecycle of its request + * @throws IllegalStateException when the container has already recycled the request + */ + static TaskRequestLifecycle of(GrailsWebRequest webRequest) { + // Kept on the request the container created, which every wrapper of it shows + ServletRequest underlying = underlying(webRequest.currentRequest) + Object existing = underlying.getAttribute(LIFECYCLE) + if (existing instanceof TaskRequestLifecycle && + (((TaskRequestLifecycle) existing).observed || !((TaskRequestLifecycle) existing).completed)) { + return (TaskRequestLifecycle) existing + } + // Without the listener, the request completes with the web request the first task was started during + synchronized (TaskRequestLifecycle) { + existing = underlying.getAttribute(LIFECYCLE) + if (existing instanceof TaskRequestLifecycle && !((TaskRequestLifecycle) existing).completed) { + return (TaskRequestLifecycle) existing + } + TaskRequestLifecycle lifecycle = new TaskRequestLifecycle(false) + if (webRequest.isActive()) { + underlying.setAttribute(LIFECYCLE, lifecycle) + webRequest.registerDestructionCallback(COMPLETION_CALLBACK, lifecycle::complete, RequestAttributes.SCOPE_REQUEST) + } + else { + lifecycle.complete() + } + return lifecycle + } + } + + /** + * Gives a request the container will tell {@link TaskRequestListener} it is done with its lifecycle, before + * anything processes it, so the attribute holding it is one an include keeps. + * + * @param request the request the container created + */ + static void initialized(ServletRequest request) { + if (!(request.getAttribute(LIFECYCLE) instanceof TaskRequestLifecycle)) { + request.setAttribute(LIFECYCLE, new TaskRequestLifecycle(true)) + } + } + + /** + * Completes the lifecycle of a request the container is done with, or, while the request is still processed + * asynchronously, once that processing completes. + * + * @param request the request the container created + */ + static void destroyed(ServletRequest request) { + Object lifecycle = request.getAttribute(LIFECYCLE) + if (!(lifecycle instanceof TaskRequestLifecycle)) { + return + } + try { + if (request.isAsyncStarted()) { Review Comment: Fixed in 6ed3af27ec. Tomcat in `MUST_COMPLETE` reports `isAsyncStarted()` as true from `fireOnComplete`. Jetty 12 (ee11) fires `requestDestroyed` at the end of every dispatch, async or not (`ServletChannel.dispatch`). So `requestDestroyed` with async started can mean "done" on one container and "still going" on the other. The lifecycle no longer relies on that call alone: - **Listening early:** the first task started once async processing has begun registers the lifecycle as an `AsyncListener`. Tomcat calls `onComplete` before it fires `requestDestroyed`, so this covers a servlet or controller that completes async before returning, such as a fast `ctx.start { ...; ctx.complete() }`. - **Fallback:** if no task was started after async began, `requestDestroyed` with async started still adds the listener for Jetty. Meanwhile every call first checks that the request still holds the lifecycle. Tomcat clears the attributes when it recycles the request, or the facade throws, so the next call detaches the tasks and throws the completed `IllegalStateException`. `lookup()` returns `null` from then on. Tests (all fail on 4694264b0c): - `a task started once asynchronous processing has begun is detached as it completes, even when it completes before the dispatch ends`. - `a request whose asynchronous processing completed before the dispatch ended completes once the container recycles it, which #recycling`, for cleared attributes and for a throwing facade. ########## grails-async/plugin/src/main/groovy/org/grails/plugins/web/async/TaskGuard.groovy: ########## @@ -0,0 +1,171 @@ +/* + * 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 + * + * https://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.grails.plugins.web.async + +import java.lang.reflect.InvocationHandler +import java.lang.reflect.InvocationTargetException +import java.lang.reflect.Method +import java.lang.reflect.Proxy + +import groovy.transform.CompileStatic +import groovy.transform.PackageScope + +import jakarta.servlet.ServletInputStream +import jakarta.servlet.ServletOutputStream + +/** + * What a {@link TaskRequest} or {@link TaskResponse} delegates to: the live request or response, called under the + * read lock of the request's {@link TaskRequestLifecycle}, until the request completes, and then nothing, so every + * call throws an {@link IllegalStateException}. + * + * <p>What a call returns that reads or writes the live object later is guarded the same way: the body's stream or + * reader and the response's stream or writer. The parameters, headers, attribute names, locales, trailer fields and + * parts are returned as a copy, since the container reuses what backs them for the requests that follow. Anything + * else, such as an attribute's value, is returned as it is.</p> + */ +@PackageScope +@CompileStatic +final class TaskGuard implements InvocationHandler, TaskRequestLifecycle.Detachable { + + // What the container answers from what it reuses for the requests that follow, so a task keeps a copy + private static final Set<String> SNAPSHOTS = Set.of('getParameterMap', 'getParameterNames', 'getHeaderNames', + 'getHeaders', 'getAttributeNames', 'getLocales', 'getTrailerFields', 'getParts') + + private final TaskRequestLifecycle lifecycle + private final String what + // The guarded stream, reader or writer of each the live object returned, so a task asking again gets the same + private final Map<Object, Object> guarded = new IdentityHashMap<>() + private volatile Object live + + private TaskGuard(TaskRequestLifecycle lifecycle, Object live, String what) { + this.lifecycle = lifecycle + this.live = live + this.what = what + } + + /** + * A guard of a live object, attached to the lifecycle of its request. + * + * @param lifecycle the lifecycle of the request + * @param live the live request or response + * @param what what it is, for the failure + * @return the guard + */ + static TaskGuard of(TaskRequestLifecycle lifecycle, Object live, String what) { + TaskGuard guard = new TaskGuard(lifecycle, live, what) + lifecycle.attach(guard) + return guard + } + + /** + * The guarded view of the live object, through the one interface given. + * + * @param type the interface, HttpServletRequest or HttpServletResponse + * @return a proxy calling the live object while the request is live + */ + <T> T proxy(Class<T> type) { + return type.cast(Proxy.newProxyInstance(type.classLoader, [type] as Class<?>[], this)) + } + + /** + * @return the live object, or null once the request has completed + */ + Object getLive() { + return live + } + + /** + * Has the guard call another live object, as a container does when it wraps the request for a forward and + * unwraps it after. + * + * @param replacement the object to call from now on + */ + void replace(Object replacement) { + lifecycle.enter(live, what) + try { + live = replacement + } + finally { + lifecycle.exit() + } + } + + @Override + void detach() { + live = null + synchronized (guarded) { + guarded.clear() + } + } + + @Override + Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + if (method.declaringClass == Object) { + switch (method.name) { + case 'equals': + return proxy.is(args[0]) + case 'hashCode': + return System.identityHashCode(proxy) + default: + return live == null ? "The $what of a completed request".toString() : "The $what of a task".toString() + } + } + Object target = lifecycle.enter(live, what) + try { + return guard(method, method.invoke(target, args)) + } + catch (InvocationTargetException e) { + throw e.targetException + } + finally { + lifecycle.exit() + } + } + + private Object guard(Method method, Object result) { Review Comment: Fixed in 6ed3af27ec. Only what `getInputStream`, `getReader`, `getOutputStream` and `getWriter` return is guarded now, chosen by method name. An attribute value is returned exactly as the request holds it, so `(LineNumberReader) request.getAttribute('lines')` works. A reader backed by a `StringReader` stays the application's own after the request completes. Test: `an attribute value a task reads is the object the request holds, even a reader`. It casts a `LineNumberReader` attribute inside a task and reads from it after completion; it fails on 4694264b0c with the `ClassCastException`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
