jamesfredley commented on code in PR #16786: URL: https://github.com/apache/grails-core/pull/16786#discussion_r4420924482
########## 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: With a ReadListener installed, Tomcat's CoyoteInputStream.read() calls isReady() while this method holds the lifecycle read lock. At EOF on a task thread that wait can hold Tomcat's socket lock. If completion already holds the socket lock, requestDestroyed() waits for the lifecycle write lock and both threads stop. Reproduced on Tomcat 11.0.27 during timeout completion. Do not hold the lifecycle lock across the read. ########## 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: Application code stores a LineNumberReader as a request attribute. While the request is live, (LineNumberReader) request.getAttribute('lines') throws ClassCastException because every attribute goes through guardStream() and a BufferedReader becomes a TaskReader. A reader backed by a StringReader also dies after completion even though it is not the servlet body. Guard only getInputStream, getReader, getOutputStream, and getWriter. ########## 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: A servlet that calls startAsync() and complete() before returning runs requestDestroyed() after the async callbacks, while isAsyncStarted() is still true. This branch registers the listener too late and returns without finishing the lifecycle. completed stays false, so later tasks keep the recycled request and lookup never detaches. Register the listener before the servlet returns, or finish this path when completion already happened. -- 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]
