diff --git a/grails-async/core/src/main/groovy/org/grails/async/factory/PromiseFactoryBuilder.groovy b/grails-async/core/src/main/groovy/org/grails/async/factory/PromiseFactoryBuilder.groovy index 6960d24b165..6bdce0ad8dd 100644 --- a/grails-async/core/src/main/groovy/org/grails/async/factory/PromiseFactoryBuilder.groovy +++ b/grails-async/core/src/main/groovy/org/grails/async/factory/PromiseFactoryBuilder.groovy @@ -23,6 +23,7 @@ import grails.async.PromiseFactory import groovy.transform.CompileStatic import groovy.util.logging.Slf4j import org.grails.async.factory.future.CachedThreadPoolPromiseFactory +import org.grails.async.factory.future.VirtualThreadPromiseFactory /** * Constructs the default promise factory @@ -43,8 +44,14 @@ class PromiseFactoryBuilder { PromiseFactory promiseFactory if (promiseFactories.isEmpty()) { - log.debug('No PromiseFactory implementation found. Using default ExecutorService promise factory.') - promiseFactory = new CachedThreadPoolPromiseFactory() + if (System.getProperty('grails.async.promiseFactory') == 'virtual-thread') { + log.debug('No PromiseFactory implementation found. Using virtual thread promise factory.') + promiseFactory = new VirtualThreadPromiseFactory() + } + else { + log.debug('No PromiseFactory implementation found. Using default ExecutorService promise factory.') + promiseFactory = new CachedThreadPoolPromiseFactory() + } } else { promiseFactory = promiseFactories.first() diff --git a/grails-async/core/src/main/groovy/org/grails/async/factory/future/VirtualThreadPromiseFactory.groovy b/grails-async/core/src/main/groovy/org/grails/async/factory/future/VirtualThreadPromiseFactory.groovy new file mode 100644 index 00000000000..45e6eeef259 --- /dev/null +++ b/grails-async/core/src/main/groovy/org/grails/async/factory/future/VirtualThreadPromiseFactory.groovy @@ -0,0 +1,119 @@ +/* + * 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.async.factory.future + +import java.util.concurrent.Callable +import java.util.concurrent.ExecutorService +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit + +import groovy.transform.AutoFinal +import groovy.transform.CompileStatic + +import jakarta.annotation.PreDestroy + +import grails.async.Promise +import grails.async.PromiseList +import grails.async.factory.AbstractPromiseFactory +import org.grails.async.factory.BoundPromise + +/** + * PromiseFactory implementation backed by Java virtual threads. + * + * @since 8.0 + */ +@AutoFinal +@CompileStatic +class VirtualThreadPromiseFactory extends AbstractPromiseFactory implements Closeable { + + private final ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor() + + @Override + Promise createPromise(Class returnType) { + new BoundPromise(null) + } + + @Override + Promise createPromise() { + new BoundPromise(null) + } + + @Override + Promise createPromise(Closure... closures) { + if (closures.length == 1) { + Closure decoratedCallable = applyDecorators(closures[0], null) + FutureTaskPromise promise = new FutureTaskPromise(this, decoratedCallable as Callable) + executorService.execute(promise) + return promise + } + + def list = new PromiseList() + for (def closure : closures) { + list.add(createPromise(closure) as Promise) + } + list as Promise + } + + @Override + List waitAll(List> promises) { + promises.collect { Promise promise -> promise.get() } + } + + @Override + List waitAll(List> promises, long timeout, TimeUnit units) { + promises.collect { Promise promise -> promise.get(timeout, units) } + } + + @Override + Promise> onComplete(List> promises, Closure callable) { + // callable's return value is intentionally discarded: the resolved value of the + // returned Promise is the waited-on values themselves (matching Promise>), + // not whatever the T-typed callback happens to return. + def promise = new FutureTaskPromise>(this, { + def values = waitAll(promises) + callable.call(values) + return values + } as Callable>) + executorService.execute(promise) + promise + } + + @Override + Promise> onError(List> promises, Closure callable) { + def promise = new FutureTaskPromise>(this, { + try { + return waitAll(promises) + } + catch (Throwable e) { + callable.call(e) + return Collections. emptyList() + } + } as Callable>) + executorService.execute(promise) + promise + } + + @Override + @PreDestroy + void close() { + if (!executorService.isShutdown()) { + executorService.shutdown() + } + } +} diff --git a/grails-async/core/src/test/groovy/grails/async/VirtualThreadPromiseFactorySpec.groovy b/grails-async/core/src/test/groovy/grails/async/VirtualThreadPromiseFactorySpec.groovy new file mode 100644 index 00000000000..7d033719be8 --- /dev/null +++ b/grails-async/core/src/test/groovy/grails/async/VirtualThreadPromiseFactorySpec.groovy @@ -0,0 +1,167 @@ +/* + * 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 grails.async + +import java.util.concurrent.ExecutionException + +import org.grails.async.factory.PromiseFactoryBuilder +import org.grails.async.factory.SynchronousPromiseFactory +import org.grails.async.factory.future.VirtualThreadPromiseFactory +import spock.lang.Specification + +class VirtualThreadPromiseFactorySpec extends Specification { + + private String originalPromiseFactoryProperty + private PromiseFactory originalPromiseFactory + + def setup() { + originalPromiseFactoryProperty = System.getProperty('grails.async.promiseFactory') + originalPromiseFactory = Promises.promiseFactory + } + + def cleanup() { + if (originalPromiseFactoryProperty == null) { + System.clearProperty('grails.async.promiseFactory') + } + else { + System.setProperty('grails.async.promiseFactory', originalPromiseFactoryProperty) + } + Promises.promiseFactory = originalPromiseFactory + } + + void 'builder can opt in to virtual thread promise factory'() { + given: + System.setProperty('grails.async.promiseFactory', 'virtual-thread') + + when: + PromiseFactory factory = PromiseFactoryBuilder.build() + + then: + factory instanceof VirtualThreadPromiseFactory + + cleanup: + (factory as Closeable)?.close() + } + + void 'virtual thread factory executes promises'() { + given: + def factory = new VirtualThreadPromiseFactory() + + when: + Promise promise = factory.createPromise { 21 * 2 } + + then: + promise.get() == 42 + + cleanup: + factory.close() + } + + void 'multi-closure promises use the virtual thread executor'() { + given: + def factory = new VirtualThreadPromiseFactory() + // PromiseList normally delegates closure creation to the global factory. Using a + // synchronous factory makes the unfixed implementation run on this test thread, + // while the corrected implementation must still use this factory's virtual threads. + Promises.promiseFactory = new SynchronousPromiseFactory() + + when: + List result = (factory.createPromise( + { Thread.currentThread().isVirtual() }, + { Thread.currentThread().isVirtual() } + ) as Promise>).get() + + then: + result == [true, true] + + cleanup: + factory.close() + } + + void 'onComplete resolves to the waited values and invokes the callback for its side effect'() { + given: + def factory = new VirtualThreadPromiseFactory() + List> promises = [factory.createPromise { 1 }, factory.createPromise { 2 }] + List observed = null + + when: 'the returned promise is consumed as a statically-typed List, not just Object' + Promise> combined = factory.onComplete(promises) { List values -> + observed = values + 'a value that is not a List - the resolved value must not become this' + } + List result = combined.get() + + then: + result == [1, 2] + observed == [1, 2] + + cleanup: + factory.close() + } + + void 'onError resolves to the waited values without invoking the callback when nothing fails'() { + given: + def factory = new VirtualThreadPromiseFactory() + List> promises = [factory.createPromise { 1 }, factory.createPromise { 2 }] + boolean callbackInvoked = false + + when: + Promise> combined = factory.onError(promises) { callbackInvoked = true } + List result = combined.get() + + then: + result == [1, 2] + !callbackInvoked + + cleanup: + factory.close() + } + + void 'onError invokes the callback and resolves to an empty list when a promise fails'() { + given: + def factory = new VirtualThreadPromiseFactory() + List> promises = [factory.createPromise { throw new IllegalStateException('boom') }] + Throwable observed = null + + when: + Promise> combined = factory.onError(promises) { Throwable error -> observed = error } + List result = combined.get() + + then: + result == [] + observed instanceof ExecutionException + observed.cause instanceof IllegalStateException + observed.cause.message == 'boom' + + cleanup: + factory.close() + } + + void 'close is idempotent'() { + given: + def factory = new VirtualThreadPromiseFactory() + + when: + factory.close() + factory.close() + + then: + noExceptionThrown() + } +} diff --git a/grails-doc/src/en/guide/async/asyncPromises.adoc b/grails-doc/src/en/guide/async/asyncPromises.adoc index 4328a3661f2..c1da03412e3 100644 --- a/grails-doc/src/en/guide/async/asyncPromises.adoc +++ b/grails-doc/src/en/guide/async/asyncPromises.adoc @@ -93,7 +93,7 @@ def result = p.get(1,MINUTES) By default, the `Promises` static methods use an instance of `PromiseFactory`. This `PromiseFactory` interface has various implementations. The default implementation is link:{api}org/grails/async/factory/future/CachedThreadPoolPromiseFactory.html[CachedThreadPoolPromiseFactory] which uses a thread pool that will create threads as needed (the same as `java.util.concurrent.Executors.newCachedThreadPool()`) -However, the design of the Grails promises framework is such that you can swap out the underlying implementation for your own or one of the pre-supported implementations. For example to use RxJava 1.x simply add the RxJava dependency to `build.gradle`: +However, the design of the Grails promises framework is such that you can swap out the underlying implementation for your own or one of the pre-supported implementations. For example to use RxJava 3 simply add the RxJava dependency to `build.gradle`: [source,groovy,subs="attributes"] .build.gradle @@ -101,7 +101,11 @@ However, the design of the Grails promises framework is such that you can swap o runtimeOnly 'org.apache.grails.async:grails-async-rxjava3' ---- -With the above in place RxJava 1.x will be used to create `Promise` instances. +With the above in place RxJava 3 will be used to create `Promise` instances. + +NOTE: Since Grails 8, an opt-in Java 21 virtual-thread promise factory is also available. +To use it, set the JVM system property `grails.async.promiseFactory=virtual-thread`. +This implementation is selected only when no service-loaded PromiseFactory is available. The following table summarizes the available implementation and the dependency that should be added to activate them: @@ -129,7 +133,6 @@ The following table summarizes the available implementation and the dependency t You can also override the `grails.async.PromiseFactory` class used by `Promises` by setting the `promiseFactory` static field. - One common use case for this is unit testing, typically you do not want promises to execute asynchronously during unit tests, as this makes tests harder to write. For this purpose Grails ships with a `org.grails.async.factory.SynchronousPromiseFactory` instance that makes it easier to test promises: [source,groovy]