Preview support for virtual threads (nextflow-io#3871)

This commit adds support for Java virtual threads. 

To enable this feature, the following variable should be declared 
in the launching environment: 

```
export NXF_ENABLE_VIRTUAL_THREADS
```

Note, this feature it's required the use of Java 19 or later. 

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

.github/workflows/build.yml

@@ -24,7 +24,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        java_version: [11, 19]
+        java_version: [11, 17, 19]
 
     steps:
       - name: Environment
@@ -92,7 +92,7 @@ jobs:
       any_changed: ${{ steps.changed-files.outputs.any_changed }}
 
   test:
-    if: ${{ !contains(github.event.head_commit.message, '[ci fast]') && needs.build.outputs.any_changed == 'true' }}
+    if: ${{ !contains(github.event.head_commit.message, '[ci fast]') }}
     needs: build
     runs-on: ubuntu-latest
     strategy:

build.gradle

@@ -61,7 +61,7 @@ allprojects {
 
     java {
         toolchain {
-            languageVersion = JavaLanguageVersion.of(17)
+            languageVersion = JavaLanguageVersion.of(19)
         }
     }
 
@@ -139,6 +139,7 @@ allprojects {
     if (JavaVersion.current() >= JavaVersion.VERSION_1_9) {
         tasks.withType(Test) {
             jvmArgs ([
+                    '--enable-preview',
                     '--add-opens=java.base/java.lang=ALL-UNNAMED',
                     '--add-opens=java.base/java.io=ALL-UNNAMED',
                     '--add-opens=java.base/java.nio=ALL-UNNAMED',
@@ -154,6 +155,7 @@ allprojects {
                     '--add-opens=java.base/sun.net.www.protocol.ftp=ALL-UNNAMED',
                     '--add-opens=java.base/sun.net.www.protocol.file=ALL-UNNAMED',
                     '--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED',
+                    '--add-opens=java.base/jdk.internal.vm=ALL-UNNAMED',
             ])
         }
     }

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.6-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.6.1-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

launch.sh

@@ -72,7 +72,8 @@ if [[ ! $JAVA_VER =~ $version_check ]]; then
     exit 1
 fi
 JVM_ARGS+=" -Dfile.encoding=UTF-8 -XX:+TieredCompilation -XX:TieredStopAtLevel=1"
-[[ $JAVA_VER =~ ^(9|10|11|12|13|14|15|16|17|18|19) ]] && JVM_ARGS+=" --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.nio.file.spi=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/sun.nio.fs=ALL-UNNAMED --add-opens=java.base/sun.net.www.protocol.http=ALL-UNNAMED --add-opens=java.base/sun.net.www.protocol.https=ALL-UNNAMED --add-opens=java.base/sun.net.www.protocol.ftp=ALL-UNNAMED --add-opens=java.base/sun.net.www.protocol.file=ALL-UNNAMED --add-opens=java.base/jdk.internal.misc=ALL-UNNAMED --add-opens=java.base/java.util.regex=ALL-UNNAMED"
+[[ $JAVA_VER =~ ^(9|10|11|12|13|14|15|16|17|18|19|20) ]] && JVM_ARGS+=" --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.nio.file.spi=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/sun.nio.fs=ALL-UNNAMED --add-opens=java.base/sun.net.www.protocol.http=ALL-UNNAMED --add-opens=java.base/sun.net.www.protocol.https=ALL-UNNAMED --add-opens=java.base/sun.net.www.protocol.ftp=ALL-UNNAMED --add-opens=java.base/sun.net.www.protocol.file=ALL-UNNAMED --add-opens=java.base/jdk.internal.misc=ALL-UNNAMED --add-opens=java.base/jdk.internal.vm=ALL-UNNAMED --add-opens=java.base/java.util.regex=ALL-UNNAMED"
+[[ $NXF_ENABLE_VIRTUAL_THREADS == 'true' ]] && JVM_ARGS+=" --enable-preview"
 
 ## flight recorded -- http://docs.oracle.com/javacomponents/jmc-5-4/jfr-runtime-guide/run.htm
 ##JVM_ARGS+=" -XX:+UnlockCommercialFeatures -XX:+FlightRecorder -XX:StartFlightRecording=duration=60s,filename=myrecording.jfr"

modules/nextflow/src/main/groovy/nextflow/Session.groovy

@@ -48,6 +48,7 @@ import nextflow.executor.ExecutorFactory
 import nextflow.extension.CH
 import nextflow.file.FileHelper
 import nextflow.file.FilePorter
+import nextflow.util.Threads
 import nextflow.util.ThreadPoolManager
 import nextflow.plugin.Plugins
 import nextflow.processor.ErrorStrategy
@@ -511,7 +512,9 @@ class Session implements ISession {
         registerSignalHandlers()
 
         // create tasks executor
-        execService = Executors.newFixedThreadPool(poolSize)
+        execService = Threads.useVirtual()
+                ? Executors.newVirtualThreadPerTaskExecutor()
+                : Executors.newFixedThreadPool(poolSize)
 
         // signal start to trace observers
         notifyFlowCreate()

modules/nextflow/src/main/groovy/nextflow/cli/CmdInfo.groovy

@@ -31,6 +31,7 @@ import nextflow.plugin.DefaultPlugins
 import nextflow.plugin.Plugins
 import nextflow.scm.AssetManager
 import nextflow.util.MemoryUnit
+import nextflow.util.Threads
 import org.yaml.snakeyaml.Yaml
 /**
  * CLI sub-command INFO
@@ -217,6 +218,8 @@ class CmdInfo extends CmdBase {
             final os = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean()
             result << BLANK << "Process: ${ManagementFactory.getRuntimeMXBean().getName()} " << getLocalAddress() << NEWLINE
             result << BLANK << "CPUs: ${os.availableProcessors} - Mem: ${totMem(os)} (${freeMem(os)}) - Swap: ${totSwap(os)} (${freeSwap(os)})"
+            if( Threads.useVirtual() )
+                result << " - Virtual threads ON"
         }
 
         if( level == 0  )

modules/nextflow/src/main/groovy/nextflow/file/DirWatcher.groovy

@@ -33,6 +33,8 @@ import java.nio.file.attribute.BasicFileAttributes
 
 import groovy.transform.CompileStatic
 import groovy.util.logging.Slf4j
+import nextflow.util.Threads
+
 /**
  * Watch the content of a directory for file system events
  *
@@ -133,7 +135,7 @@ class DirWatcher implements DirListener {
         this.watcher = base.getFileSystem().newWatchService()
         this.watchedPaths = new HashMap<WatchKey,Path>()
 
-        thread = Thread.startDaemon {
+        thread = Threads.start {
             try {
                 apply0()
             }

modules/nextflow/src/main/groovy/nextflow/file/FilePorter.groovy

@@ -25,9 +25,12 @@ import java.nio.file.attribute.BasicFileAttributes
 import java.util.concurrent.ExecutionException
 import java.util.concurrent.ExecutorService
 import java.util.concurrent.Future
+import java.util.concurrent.Semaphore
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.TimeoutException
 import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.locks.Lock
+import java.util.concurrent.locks.ReentrantLock
 
 import groovy.transform.Canonical
 import groovy.transform.CompileStatic
@@ -40,6 +43,7 @@ import nextflow.extension.FilesEx
 import nextflow.util.CacheHelper
 import nextflow.util.Duration
 import nextflow.util.ThreadPoolManager
+import nextflow.util.Threads
 
 /**
  * Move foreign (ie. remote) files to the staging work area
@@ -55,6 +59,8 @@ class FilePorter {
 
     static final private int MAX_RETRIES = 3
 
+    static final private int MAX_TRANSFERS = 50
+
     static final private Duration POLL_TIMEOUT = Duration.of('2sec')
 
     final Map<FileCopy,FileTransfer> stagingTransfers = new HashMap<>()
@@ -63,14 +69,26 @@ class FilePorter {
 
     private Duration pollTimeout
 
+    private final Semaphore semaphore
+
+    final private Lock sync
+
     final int maxRetries
 
-    final Session session
+    final int maxTransfers
+
+    final private Session session
 
     FilePorter( Session session ) {
         this.session = session
         maxRetries = session.config.navigate('filePorter.maxRetries') as Integer ?: MAX_RETRIES
         pollTimeout = session.config.navigate('filePorter.pollTimeout') as Duration ?: POLL_TIMEOUT
+        maxTransfers = session.config.navigate('filePorter.maxTransfers') as Integer ?: MAX_TRANSFERS
+        log.debug "File porter settings maxRetries=$maxRetries; maxTransfers=$maxTransfers; pollTimeout=$threadPool"
+        sync = new ReentrantLock()
+        // use a semaphore to cap the number of max transfer when using virtual thread
+        // when using platform threads the max transfers are limited by the thread pool itself
+        semaphore = Threads.useVirtual() ? new Semaphore(maxTransfers) : null
         threadPool = new ThreadPoolManager('FileTransfer')
                 .withConfig(session.config)
                 .createAndRegisterShutdownCallback(session)
@@ -87,26 +105,26 @@ class FilePorter {
     }
 
     protected FileTransfer createFileTransfer(Path source, Path target) {
-        return new FileTransfer(source, target, maxRetries)
-    }
-
-    protected Future submitForExecution(FileTransfer transfer) {
-        threadPool.submit(transfer)
+        return new FileTransfer(source, target, maxRetries, semaphore)
     }
 
     protected FileTransfer getOrSubmit(FileCopy copy) {
-        synchronized (this) {
+        sync.lock()
+        try {
             FileTransfer transfer = stagingTransfers.get(copy)
             if( transfer == null ) {
                 transfer = createFileTransfer(copy.source, copy.target)
-                transfer.result = submitForExecution(transfer)
+                transfer.result = threadPool.submit(transfer)
                 stagingTransfers.put(copy, transfer)
             }
             // increment the ref count
             transfer.refCount.incrementAndGet()
 
             return transfer
         }
+        finally {
+            sync.unlock()
+        }
     }
 
     protected void decOrRemove(FileTransfer action) {
@@ -258,12 +276,14 @@ class FilePorter {
          */
         final int maxRetries
 
+        final private Semaphore semaphore
         final AtomicInteger refCount
         volatile Future result
         private String message
         private int debugDelay
 
-        FileTransfer(Path foreignPath, Path stagePath, int maxRetries=0) {
+        FileTransfer(Path foreignPath, Path stagePath, int maxRetries, Semaphore semaphore) {
+            this.semaphore = semaphore
             this.source = foreignPath
             this.target = stagePath
             this.maxRetries = maxRetries
@@ -274,7 +294,15 @@ class FilePorter {
 
         @Override
         void run() throws Exception {
-            stageForeignFile(source, target)
+            if( semaphore )
+                semaphore.acquire()
+            try {
+                stageForeignFile(source, target)
+            }
+            finally {
+                if( semaphore )
+                    semaphore.release()
+            }
         }
 
         /**
@@ -331,19 +359,25 @@ class FilePorter {
         }
     }
 
-    synchronized protected FileCopy getCachePathFor(Path sourcePath, Path stageDir) {
-        final dirPath = stageDir.toUriString() // <-- use a string to avoid changes in the dir to alter the hashing
-        int i=0
-        while( true ) {
-            final uniq = List.of(sourcePath, dirPath, i++)
-            final hash = CacheHelper.hasher(uniq).hash().toString()
-            final targetPath = getCacheDir0(stageDir, hash).resolve(sourcePath.getName())
-            final result = new FileCopy(sourcePath, targetPath)
-            if( stagingTransfers.containsKey(result) )
-                return result
-            final exist = targetPath.exists()
-            if( !exist || checkPathIntegrity(sourcePath, targetPath) )
-                return result
+    protected FileCopy getCachePathFor(Path sourcePath, Path stageDir) {
+        sync.lock()
+        try {
+            final dirPath = stageDir.toUriString() // <-- use a string to avoid changes in the dir to alter the hashing
+            int i=0
+            while( true ) {
+                final uniq = List.of(sourcePath, dirPath, i++)
+                final hash = CacheHelper.hasher(uniq).hash().toString()
+                final targetPath = getCacheDir0(stageDir, hash).resolve(sourcePath.getName())
+                final result = new FileCopy(sourcePath, targetPath)
+                if( stagingTransfers.containsKey(result) )
+                    return result
+                final exist = targetPath.exists()
+                if( !exist || checkPathIntegrity(sourcePath, targetPath) )
+                    return result
+            }
+        }
+        finally {
+            sync.unlock()
         }
     }
 

modules/nextflow/src/main/groovy/nextflow/file/PathVisitor.groovy

@@ -35,6 +35,7 @@ import nextflow.Global
 import nextflow.Session
 import nextflow.extension.CH
 import nextflow.util.CustomThreadFactory
+import nextflow.util.Threads
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 /**
@@ -186,7 +187,10 @@ class PathVisitor {
     @Memoized
     @PackageScope
     static ExecutorService createExecutor(Session session) {
-        final result = Executors.newCachedThreadPool(new CustomThreadFactory('PathVisitor'))
+        final factory = new CustomThreadFactory('PathVisitor')
+        final result = Threads.useVirtual()
+                ? Executors.newThreadPerTaskExecutor(factory)
+                : Executors.newCachedThreadPool(factory)
         Global.onCleanup((it) -> result.shutdown())
         return result
     }

modules/nextflow/src/main/groovy/nextflow/processor/TaskPollingMonitor.groovy

@@ -30,6 +30,7 @@ import nextflow.Session
 import nextflow.executor.BatchCleanup
 import nextflow.executor.GridTaskHandler
 import nextflow.util.Duration
+import nextflow.util.Threads
 import nextflow.util.Throttle
 /**
  * Monitors the queued tasks waiting for their termination
@@ -287,7 +288,7 @@ class TaskPollingMonitor implements TaskMonitor {
         session.onShutdown { this.cleanup() }
 
         // launch the thread polling the queue
-        Thread.start('Task monitor') {
+        Threads.start('Task monitor') {
             try {
                 pollLoop()
             }
@@ -298,7 +299,7 @@ class TaskPollingMonitor implements TaskMonitor {
         }
 
         // launch daemon that submits tasks for execution
-        Thread.startDaemon('Task submitter', this.&submitLoop)
+        Threads.start('Task submitter', this.&submitLoop)
 
         return this
     }

modules/nextflow/src/main/groovy/nextflow/trace/AnsiLogObserver.groovy

@@ -21,6 +21,7 @@ import jline.TerminalFactory
 import nextflow.Session
 import nextflow.processor.TaskHandler
 import nextflow.util.Duration
+import nextflow.util.Threads
 import org.fusesource.jansi.Ansi
 import org.fusesource.jansi.AnsiConsole
 import static nextflow.util.LoggerHelper.isHashLogPrefix
@@ -405,7 +406,7 @@ class AnsiLogObserver implements TraceObserver {
         this.statsObserver = session.statsObserver
         this.startTimestamp = System.currentTimeMillis()
         AnsiConsole.systemInstall()
-        this.renderer = Thread.start('AnsiLogObserver', this.&render0)
+        this.renderer = Threads.start('AnsiLogObserver', this.&render0)
     }
 
     @Override

modules/nextflow/src/main/groovy/nextflow/util/CustomPoolFactory.groovy

@@ -15,8 +15,10 @@
  */
 
 package nextflow.util
+
 import groovy.transform.CompileStatic
 import groovy.transform.Memoized
+import groovy.util.logging.Slf4j
 import groovyx.gpars.scheduler.Pool
 import groovyx.gpars.scheduler.ResizeablePool
 import groovyx.gpars.util.PoolFactory
@@ -27,6 +29,7 @@ import groovyx.gpars.util.PoolFactory
  *
  * @author Paolo Di Tommaso <[email protected]>
  */
+@Slf4j
 @CompileStatic
 class CustomPoolFactory implements PoolFactory {
 
@@ -40,7 +43,8 @@ class CustomPoolFactory implements PoolFactory {
     @Override
     synchronized Pool createPool() {
 
-        def type = property(NXF_POOL_TYPE, 'default')
+        final defType = Threads.useVirtual() ? 'virtual' : 'default'
+        final type = property(NXF_POOL_TYPE, defType)
         switch (type) {
             case 'default':
                 int poolSize = Runtime.runtime.availableProcessors() +1
@@ -66,6 +70,10 @@ class CustomPoolFactory implements PoolFactory {
                 int size = property(NXF_MAX_THREADS, cpus+1) as int
                 return CustomThreadPool.unboundedPool(size)
 
+            case 'virtual':
+                log.debug "Creating virtual thread pool"
+                return CustomThreadPool.virtualPool()
+
             default:
                 throw new IllegalAccessException("Unknown thread pool type: `$type`")
         }

modules/nextflow/src/main/groovy/nextflow/util/CustomThreadPool.java

@@ -105,6 +105,10 @@ static Pool synchronousPool(int maxThreads) {
         );
     }
 
+    static Pool virtualPool() {
+        return new VirtualThreadPool( newDaemonThreadFactory() );
+    }
+
     /**
      * Creates the pool with specified executor.
      */