Use virtual threads for I/O bound operations

This commit introduces the use of virtual threads for short-lived
I/O bound operations. More in details virtual threads are used in
place of the usual thread pool for:
- PairingInboundStore: use a virtual thread pool to creare a CompletableFuture
  to await the completation of a remote command
- HttpClientFactory: use a virtual thread for HttpClient operations
- TimedInputStream: use a virtual thread pool to read the target input
  stream within a fixed amount of time.

Signed-off-by: Paolo Di Tommaso <[email protected]>

build.gradle

@@ -112,7 +112,7 @@ micronaut {
 //
 jib {
     from {
-        image = 'cr.seqera.io/public/nf-jdk:corretto-17.0.9-jemalloc'
+        image = 'cr.seqera.io/public/nf-jdk:corretto-21.0.1-jemalloc'
         platforms {
             platform { architecture = 'amd64'; os = 'linux' }
         }

src/main/groovy/io/seqera/wave/http/HttpClientFactory.groovy

@@ -25,7 +25,6 @@ import java.util.concurrent.Executors
 
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
-import io.seqera.wave.util.CustomThreadFactory
 /**
  * Java HttpClient factory
  *
@@ -35,7 +34,7 @@ import io.seqera.wave.util.CustomThreadFactory
 @CompileStatic
 class HttpClientFactory {
 
-    static private ExecutorService threadPool = Executors.newCachedThreadPool(new CustomThreadFactory("HttpClientThread"))
+    static private ExecutorService threadPool = Executors.newVirtualThreadPerTaskExecutor()
 
     static private Duration timeout = Duration.ofSeconds(20)
 

src/main/groovy/io/seqera/wave/service/data/future/AbstractFutureStore.groovy

@@ -20,6 +20,8 @@ package io.seqera.wave.service.data.future
 
 import java.time.Duration
 import java.util.concurrent.CompletableFuture
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
 import java.util.concurrent.TimeoutException
 
 import groovy.transform.CompileStatic
@@ -50,9 +52,12 @@ abstract class AbstractFutureStore<V> implements FutureStore<String,V> {
     @Value('${wave.pairing.channel.awaitTimeout:100ms}')
     private Duration pollInterval
 
+    private ExecutorService virtPool
+
     AbstractFutureStore(FutureHash<String> store, EncodingStrategy<V> encodingStrategy) {
         this.store = store
         this.encodingStrategy = encodingStrategy
+        this.virtPool = Executors.newVirtualThreadPerTaskExecutor()
     }
 
     abstract String prefix()
@@ -85,7 +90,7 @@ abstract class AbstractFutureStore<V> implements FutureStore<String,V> {
                 // sleep for a while
                 sleep(pollInterval.toMillis())
             }
-        })
+        }, virtPool) // <-- use a virtual thread pool
     }
 
     /**

src/main/groovy/io/seqera/wave/util/TimedInputStream.groovy

@@ -19,14 +19,17 @@
 package io.seqera.wave.util
 
 import java.time.Duration
-import java.util.concurrent.CompletableFuture
+import java.util.concurrent.Callable
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
 
 import groovy.transform.CompileStatic
 import io.seqera.wave.core.RoutePath
 import io.seqera.wave.exception.UnexpectedReadException
-
 /**
+ * An input stream filter that implements a timeout mechanism on the reading
+ * operation over the target stream.
  *
  * @author Paolo Di Tommaso <[email protected]>
  */
@@ -39,18 +42,21 @@ class TimedInputStream extends FilterInputStream {
 
     private final RoutePath route
 
+    private final ExecutorService executor
+
     private volatile boolean closed
 
     TimedInputStream(InputStream inputStream, Duration timeout, RoutePath route) {
         super(inputStream)
         this.target = inputStream
         this.timeoutMillis = (int)timeout.toMillis()
         this.route = route
+        this.executor = Executors.newVirtualThreadPerTaskExecutor()
     }
 
     @Override
     int read() throws IOException {
-        final result = CompletableFuture<Integer>.supplyAsync(() -> target.read())
+        final result = executor.submit((Callable<Integer>)(() -> target.read()))
         try {
             return result.get(timeoutMillis, TimeUnit.MILLISECONDS)
         }
@@ -62,7 +68,7 @@ class TimedInputStream extends FilterInputStream {
 
     @Override
     int read(byte[] b, int off, int len) throws IOException {
-        final result = CompletableFuture<Integer>.supplyAsync(() -> target.read(b,off,len))
+        final result = executor.submit((Callable<Integer>)(() -> target.read(b,off,len)))
         try {
             return result.get(timeoutMillis, TimeUnit.MILLISECONDS)
         }