Ported combined FT test to Nima and enhancements to Async

nima/fault-tolerance/src/main/java/io/helidon/nima/faulttolerance/Async.java

@@ -16,9 +16,13 @@
 
 package io.helidon.nima.faulttolerance;
 
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
 import java.util.function.Supplier;
 
+import io.helidon.common.LazyValue;
+
 /**
  * Runs synchronous suppliers asynchronously using virtual threads. Includes
  * convenient static method to avoid creating instances of this class.
@@ -55,4 +59,53 @@ static Async create() {
     static <T> CompletableFuture<T> invokeStatic(Supplier<T> supplier) {
         return create().invoke(supplier);
     }
+
+    /**
+     * A new builder to build a customized {@link Async} instance.
+     * @return a new builder
+     */
+    static Builder builder() {
+        return new Builder();
+    }
+
+    /**
+     * Fluent API Builder for {@link Async}.
+     */
+    class Builder implements io.helidon.common.Builder<Builder, Async> {
+        private LazyValue<? extends ExecutorService> executor;
+
+        private Builder() {
+        }
+
+        @Override
+        public Async build() {
+            return new AsyncImpl(this);
+        }
+
+        /**
+         * Configure executor service to use for executing tasks asynchronously.
+         *
+         * @param executor executor service supplier
+         * @return updated builder instance
+         */
+        public Builder executor(Supplier<? extends ExecutorService> executor) {
+            this.executor = LazyValue.create(Objects.requireNonNull(executor));
+            return this;
+        }
+
+        /**
+         * Configure executor service to use for executing tasks asynchronously.
+         *
+         * @param executor executor service
+         * @return updated builder instance
+         */
+        public Builder executor(ExecutorService executor) {
+            this.executor = LazyValue.create(Objects.requireNonNull(executor));
+            return this;
+        }
+
+        LazyValue<? extends ExecutorService> executor() {
+            return executor;
+        }
+    }
 }

nima/fault-tolerance/src/main/java/io/helidon/nima/faulttolerance/AsyncImpl.java

@@ -17,17 +17,33 @@
 package io.helidon.nima.faulttolerance;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.function.Supplier;
 
+import io.helidon.common.LazyValue;
+
+/**
+ * Implementation of {@code Async}. If no executor specified in builder, then it will
+ * use {@link Executors#newVirtualThreadPerTaskExecutor}. Note that this default executor
+ * is not configurable using Helidon's config.
+ */
 class AsyncImpl implements Async {
+    private final LazyValue<? extends ExecutorService> executor;
+
+    AsyncImpl() {
+        this.executor = LazyValue.create(Executors.newVirtualThreadPerTaskExecutor());
+    }
 
-    private AsyncImpl() {
+    AsyncImpl(Builder builder) {
+        this.executor = builder.executor() != null ? builder.executor()
+                : LazyValue.create(Executors.newVirtualThreadPerTaskExecutor());
     }
 
     @Override
     public <T> CompletableFuture<T> invoke(Supplier<T> supplier) {
         CompletableFuture<T> result = new CompletableFuture<>();
-        Thread.ofVirtual().start(() -> {
+        executor.get().submit(() -> {
             try {
                 T t = supplier.get();
                 result.complete(t);
@@ -38,6 +54,9 @@ public <T> CompletableFuture<T> invoke(Supplier<T> supplier) {
         return result;
     }
 
+    /**
+     * Default {@code Async} instance that uses {@link Executors#newVirtualThreadPerTaskExecutor}.
+     */
     static final class DefaultAsyncInstance {
         private static final Async INSTANCE = new AsyncImpl();
 

nima/fault-tolerance/src/main/java/io/helidon/nima/faulttolerance/Bulkhead.java

@@ -16,12 +16,10 @@
 
 package io.helidon.nima.faulttolerance;
 
-import java.util.Objects;
-import java.util.concurrent.ExecutorService;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.function.Supplier;
 
-import io.helidon.common.LazyValue;
-
 /**
  * Bulkhead protects a resource that cannot serve unlimited parallel
  * requests.
@@ -83,17 +81,41 @@ interface Stats {
         long waitingQueueSize();
     }
 
+    /**
+     * A Bulkhead listener for queueing operations.
+     */
+    interface QueueListener {
+
+        /**
+         * Called right before blocking on the internal semaphore's queue.
+         *
+         * @param supplier the supplier to be enqueued
+         * @param <T> type of value returned by supplier
+         */
+        default <T> void enqueueing(Supplier<? extends T> supplier) {
+        }
+
+        /**
+         * Called after semaphore is acquired and before supplier is called.
+         *
+         * @param supplier the supplier to execute
+         * @param <T> type of value returned by supplier
+         */
+        default <T> void dequeued(Supplier<? extends T> supplier) {
+        }
+    }
+
     /**
      * Fluent API builder for {@link io.helidon.nima.faulttolerance.Bulkhead}.
      */
     class Builder implements io.helidon.common.Builder<Builder, Bulkhead> {
         private static final int DEFAULT_LIMIT = 10;
         private static final int DEFAULT_QUEUE_LENGTH = 10;
 
-        private LazyValue<? extends ExecutorService> executor = FaultTolerance.executor();
         private int limit = DEFAULT_LIMIT;
         private int queueLength = DEFAULT_QUEUE_LENGTH;
         private String name = "Bulkhead-" + System.identityHashCode(this);
+        private List<QueueListener> listeners = new ArrayList<>();
 
         private Builder() {
         }
@@ -103,16 +125,6 @@ public Bulkhead build() {
             return new BulkheadImpl(this);
         }
 
-        /**
-         * Configure executor service to use for executing tasks asynchronously.
-         *
-         * @param executor executor service supplier
-         * @return updated builder instance
-         */
-        public Builder executor(Supplier<? extends ExecutorService> executor) {
-            this.executor = LazyValue.create(Objects.requireNonNull(executor));
-            return this;
-        }
 
         /**
          * Maximal number of parallel requests going through this bulkhead.
@@ -150,6 +162,17 @@ public Builder name(String name) {
             return this;
         }
 
+        /**
+         * Add a queue listener to this bulkhead.
+         *
+         * @param listener a queue listener
+         * @return updated builder instance
+         */
+        public Builder addQueueListener(QueueListener listener) {
+            listeners.add(listener);
+            return this;
+        }
+
         int limit() {
             return limit;
         }
@@ -158,12 +181,12 @@ int queueLength() {
             return queueLength;
         }
 
-        LazyValue<? extends ExecutorService> executor() {
-            return executor;
-        }
-
         String name() {
             return name;
         }
+
+        List<QueueListener> queueListeners() {
+            return listeners;
+        }
     }
 }

nima/fault-tolerance/src/main/java/io/helidon/nima/faulttolerance/BulkheadImpl.java

@@ -17,6 +17,7 @@
 package io.helidon.nima.faulttolerance;
 
 import java.lang.System.Logger.Level;
+import java.util.List;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -32,11 +33,13 @@ class BulkheadImpl implements Bulkhead {
     private final AtomicLong callsAccepted = new AtomicLong(0L);
     private final AtomicLong callsRejected = new AtomicLong(0L);
     private final AtomicInteger enqueued = new AtomicInteger();
+    private final List<QueueListener> listeners;
 
     BulkheadImpl(Builder builder) {
         this.inProgress = new Semaphore(builder.limit(), true);
         this.name = builder.name();
         this.maxQueue = builder.queueLength();
+        this.listeners = builder.queueListeners();
     }
 
     @Override
@@ -46,7 +49,7 @@ public String name() {
 
     @Override
     public <T> T invoke(Supplier<? extends T> supplier) {
-
+        // execute immediately if semaphore can be acquired
         if (inProgress.tryAcquire()) {
             if (LOGGER.isLoggable(Level.DEBUG)) {
                 LOGGER.log(Level.DEBUG, name + " invoke immediate " + supplier);
@@ -62,7 +65,11 @@ public <T> T invoke(Supplier<? extends T> supplier) {
         }
         try {
             // block current thread until permit available
+            listeners.forEach(l -> l.enqueueing(supplier));
             inProgress.acquire();
+
+            // unblocked so we can proceed with execution
+            listeners.forEach(l -> l.dequeued(supplier));
             enqueued.decrementAndGet();
             if (LOGGER.isLoggable(Level.DEBUG)) {
                 LOGGER.log(Level.DEBUG, name + " invoking " + supplier);

nima/fault-tolerance/src/main/java/io/helidon/nima/faulttolerance/TimeoutImpl.java

@@ -16,7 +16,6 @@
 
 package io.helidon.nima.faulttolerance;
 
-import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
@@ -35,10 +34,8 @@ class TimeoutImpl implements Timeout {
     private final LazyValue<? extends ScheduledExecutorService> executor;
     private final boolean currentThread;
     private final String name;
-    private final Duration timeout;
 
     TimeoutImpl(Builder builder) {
-        this.timeout = builder.timeout();
         this.timeoutMillis = builder.timeout().toMillis();
         this.executor = builder.executor();
         this.currentThread = builder.currentThread();
@@ -60,7 +57,11 @@ public <T> T invoke(Supplier<? extends T> supplier) {
             } catch (InterruptedException e) {
                 throw new TimeoutException("Call interrupted", e);
             } catch (ExecutionException e) {
-                throw new TimeoutException("Asynchronous execution error", e.getCause());
+                // Map java.util.concurrent.TimeoutException to Nima's TimeoutException
+                if (e.getCause() instanceof java.util.concurrent.TimeoutException) {
+                   throw new TimeoutException("Timeout reached", e.getCause().getCause());
+                }
+                throw new RuntimeException("Asynchronous execution error", e.getCause());
             }
         } else {
             Thread thisThread = Thread.currentThread();

nima/fault-tolerance/src/test/java/io/helidon/nima/faulttolerance/AsyncTest.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright (c) 2022 Oracle and/or its affiliates.
+ *
+ * Licensed 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
+ *
+ *     http://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 io.helidon.nima.faulttolerance;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.CoreMatchers.is;
+
+class AsyncTest {
+    private static final long WAIT_TIMEOUT_MILLIS = 2000;
+
+    @Test
+    void testDefaultExecutorCreate() {
+        Thread thread = testAsync(Async.create());
+        assertThat(thread.isVirtual(), is(true));
+    }
+
+    @Test
+    void testDefaultExecutorBuilder(){
+        Async async = Async.builder().build();
+        Thread thread = testAsync(async);
+        assertThat(thread.isVirtual(), is(true));
+    }
+
+    @Test
+    void testCustomExecutorBuilder() {
+        Async async = Async.builder()
+                .executor(FaultTolerance.executor())        // platform thread executor
+                .build();
+        Thread thread = testAsync(async);
+        assertThat(thread.isVirtual(), is(false));
+    }
+
+    private Thread testAsync(Async async) {
+        try {
+            CompletableFuture<Thread> cf = new CompletableFuture<>();
+            async.invoke(() -> {
+                cf.complete(Thread.currentThread());
+                return null;
+            });
+            return cf.get(WAIT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}