Dispatch Mutiny scheduled operations using Vert.x timers when possible

Some Mutiny operations such as delaying an item, performing a retry, or generating streams
from periodic ticks all require a ScheduledExecutorService.

This introduces a wrapper ScheduledExecutorService that dispatches some of the operations
to Vert.x timers when called from a Vert.x event-loop, offering a more natural threading
mental model as well as avoiding thread hops between worker threads and event-loops.

The implementation only tunes Mutiny with that Vert.x-aware wrapper ScheduledExecutorService
from the quarkus-vertx extension because non-server applications might not have Vert.x available.

We also introduce an artificial build item in the Mutiny extension to ensure that the Vert.x build
items always execute after those from Mutiny, or else we could have cases where Mutiny re-installs
a non-Vert.x aware scheduler in its own configuration.

Fixes: #50918

Author: jponge

extensions/mutiny/deployment/src/main/java/io/quarkus/mutiny/deployment/MutinyProcessor.java

@@ -16,12 +16,13 @@ public class MutinyProcessor {
 
     @BuildStep
     @Record(ExecutionTime.RUNTIME_INIT)
-    public void runtimeInit(ExecutorBuildItem executorBuildItem,
+    public MutinyRuntimeInitBuildItem runtimeInit(ExecutorBuildItem executorBuildItem,
             MutinyInfrastructure recorder,
             ShutdownContextBuildItem shutdownContext,
             Optional<ContextHandlerBuildItem> contextHandler) {
         ContextHandler<Object> handler = contextHandler.map(ContextHandlerBuildItem::contextHandler).orElse(null);
         recorder.configureMutinyInfrastructure(executorBuildItem.getExecutorProxy(), shutdownContext, handler);
+        return new MutinyRuntimeInitBuildItem();
     }
 
     @BuildStep

extensions/mutiny/deployment/src/main/java/io/quarkus/mutiny/deployment/MutinyRuntimeInitBuildItem.java

@@ -0,0 +1,10 @@
+package io.quarkus.mutiny.deployment;
+
+import io.quarkus.builder.item.SimpleBuildItem;
+
+/**
+ * Marker build item to detect when Mutiny has been initialized at runtime.
+ */
+public final class MutinyRuntimeInitBuildItem extends SimpleBuildItem {
+
+}

extensions/vertx/deployment/src/main/java/io/quarkus/vertx/core/deployment/VertxCoreProcessor.java

@@ -52,6 +52,7 @@
 import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
 import io.quarkus.deployment.logging.LogCleanupFilterBuildItem;
 import io.quarkus.gizmo.Gizmo;
+import io.quarkus.mutiny.deployment.MutinyRuntimeInitBuildItem;
 import io.quarkus.netty.deployment.EventLoopSupplierBuildItem;
 import io.quarkus.vertx.VertxOptionsCustomizer;
 import io.quarkus.vertx.core.runtime.VertxCoreRecorder;
@@ -234,7 +235,11 @@ CoreVertxBuildItem build(
             List<VertxOptionsConsumerBuildItem> vertxOptionsConsumers,
             BuildProducer<SyntheticBeanBuildItem> syntheticBeans,
             BuildProducer<EventLoopSupplierBuildItem> eventLoops,
-            ExecutorBuildItem executorBuildItem) {
+            ExecutorBuildItem executorBuildItem,
+            MutinyRuntimeInitBuildItem mutinyRuntimeInitBuildItem) {
+
+        // Override the Mutiny infrastructure ScheduledExecutorService to dispatch scheduled operations to a Vert.x timer
+        recorder.wrapMainExecutorForMutiny(executorBuildItem.getExecutorProxy());
 
         Collections.sort(vertxOptionsConsumers);
         List<Consumer<VertxOptions>> consumers = new ArrayList<>(vertxOptionsConsumers.size());

extensions/vertx/deployment/src/test/java/io/quarkus/vertx/ScheduledTasksTest.java

@@ -0,0 +1,153 @@
+package io.quarkus.vertx;
+
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.everyItem;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+import static org.hamcrest.Matchers.not;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Uni;
+import io.smallrye.mutiny.helpers.test.UniAssertSubscriber;
+import io.smallrye.mutiny.subscription.Cancellable;
+import io.vertx.core.Vertx;
+
+public class ScheduledTasksTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest config = new QuarkusUnitTest()
+            .setArchiveProducer(() -> ShrinkWrap
+                    .create(JavaArchive.class).addClasses(MyBean.class));
+
+    @Inject
+    Vertx vertx;
+
+    @Inject
+    MyBean myBean;
+
+    @Test
+    void uniFromRegularThread() {
+        String hello = myBean.delayedHello().await().atMost(Duration.ofSeconds(5));
+        assertThat(hello, containsString("Hello!"));
+        assertThat(hello, containsString("executor-thread-"));
+    }
+
+    @Test
+    void uniFromEventLoop() {
+        AtomicReference<String> result = new AtomicReference<>();
+        vertx.runOnContext(v -> {
+            myBean.delayedHello().subscribe().with(
+                    result::set,
+                    err -> result.set(err.getMessage()));
+        });
+        await().atMost(Duration.ofSeconds(5)).until(() -> result.get() != null);
+        assertThat(result.get(), containsString("Hello!"));
+        assertThat(result.get(), containsString("vert.x-eventloop-thread-"));
+    }
+
+    @Test
+    void multiFromRegularThread() {
+        LongAdder counter = new LongAdder();
+        AtomicBoolean cancelled = new AtomicBoolean();
+        ArrayList<String> items = new ArrayList<>();
+        Cancellable cancellable = myBean.ticks()
+                .onCancellation().invoke(() -> cancelled.set(true))
+                .subscribe().with(tick -> {
+                    items.add(tick);
+                    counter.increment();
+                });
+        long start = System.currentTimeMillis();
+        await().atMost(Duration.ofSeconds(5)).untilAdder(counter, greaterThan(3L));
+        cancellable.cancel();
+        await().atMost(Duration.ofSeconds(1)).untilTrue(cancelled);
+        long duration = System.currentTimeMillis() - start;
+        assertThat((long) items.size(), lessThanOrEqualTo(duration / 100L + 1L));
+        assertThat(items.get(0), containsString("Hello executor-thread-"));
+    }
+
+    @Test
+    void multiFromEventLoop() {
+        LongAdder counter = new LongAdder();
+        AtomicBoolean cancelled = new AtomicBoolean();
+        ArrayList<String> items = new ArrayList<>();
+        AtomicReference<Cancellable> cancellable = new AtomicReference<>();
+        vertx.runOnContext(v -> {
+            cancellable.set(myBean.ticks()
+                    .onCancellation().invoke(() -> cancelled.set(true))
+                    .subscribe().with(tick -> {
+                        items.add(tick);
+                        counter.increment();
+                    }));
+        });
+        long start = System.currentTimeMillis();
+        await().atMost(Duration.ofSeconds(5)).untilAdder(counter, greaterThan(3L));
+        cancellable.get().cancel();
+        await().atMost(Duration.ofSeconds(1)).untilTrue(cancelled);
+        long duration = System.currentTimeMillis() - start;
+        assertThat((long) items.size(), lessThanOrEqualTo(duration / 100L + 1L));
+        assertThat(items.get(0), containsString("Hello vert.x-eventloop-thread-"));
+    }
+
+    @Test
+    void retryFromRegularThread() {
+        List<String> threadTraces = new ArrayList<>();
+        UniAssertSubscriber<String> sub = UniAssertSubscriber.create();
+        myBean.backoff(threadTraces).subscribe().withSubscriber(sub);
+        sub.awaitFailure().assertFailedWith(IOException.class, "boom");
+        assertThat(threadTraces, hasSize(6));
+        assertThat(threadTraces, allOf((everyItem(not(containsString("vert.x-eventloop-thread-"))))));
+    }
+
+    @Test
+    void retryFromEventLoop() {
+        List<String> threadTraces = new ArrayList<>();
+        UniAssertSubscriber<String> sub = UniAssertSubscriber.create();
+        vertx.runOnContext(v -> myBean.backoff(threadTraces).subscribe().withSubscriber(sub));
+        sub.awaitFailure().assertFailedWith(IOException.class, "boom");
+        assertThat(threadTraces, hasSize(6));
+        assertThat(threadTraces, allOf(everyItem(containsString("vert.x-eventloop-thread-"))));
+    }
+
+    @ApplicationScoped
+    static class MyBean {
+
+        public Uni<String> delayedHello() {
+            return Uni.createFrom().item("Hello!")
+                    .onItem().delayIt().by(Duration.ofSeconds(1L))
+                    .onItem().transform(s -> s + " :: " + Thread.currentThread().getName());
+        }
+
+        public Multi<String> ticks() {
+            return Multi.createFrom().ticks()
+                    .every(Duration.ofMillis(100))
+                    .onItem().transform(tick -> "Hello " + Thread.currentThread().getName());
+        }
+
+        public Uni<String> backoff(List<String> threadTraces) {
+            return Uni.createFrom().<String> failure(new IOException("boom"))
+                    .onSubscription().invoke(() -> threadTraces.add(Thread.currentThread().getName()))
+                    .onFailure(IOException.class).retry().withBackOff(Duration.ofMillis(10)).atMost(5L);
+        }
+    }
+}

extensions/vertx/runtime/src/main/java/io/quarkus/vertx/core/runtime/VertxCoreRecorder.java

@@ -8,11 +8,20 @@
 import static io.vertx.core.file.impl.FileResolverImpl.CACHE_DIR_BASE_PROP_NAME;
 
 import java.io.File;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Random;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -39,6 +48,7 @@
 import io.quarkus.vertx.runtime.VertxCurrentContextFactory;
 import io.quarkus.vertx.runtime.jackson.QuarkusJacksonFactory;
 import io.smallrye.common.cpu.ProcessorInfo;
+import io.smallrye.mutiny.infrastructure.Infrastructure;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Context;
 import io.vertx.core.Handler;
@@ -755,4 +765,9 @@ private static void deleteDirectory(File directory) {
         }
         directory.delete();
     }
+
+    public void wrapMainExecutorForMutiny(ScheduledExecutorService service) {
+        VertxTimerAwareScheduledExecutorService wrapper = new VertxTimerAwareScheduledExecutorService(service);
+        Infrastructure.setDefaultExecutor(wrapper, false);
+    }
 }