COLDBOX-1332 #resolve

Virtual Thread executors support

Author: lmajano

server-adobe@2023.json

@@ -20,7 +20,7 @@
     },
     "JVM":{
         "heapSize":"1024",
-		"javaVersion":"openjdk11_jre"
+        "javaVersion":"openjdk21_jre"
     },
     "cfconfig":{
         "file":".cfconfig.json"

system/async/AsyncManager.cfc

@@ -64,17 +64,21 @@ component accessors="true" singleton {
 	 * scheduled executor then actually execute scheduled tasks.
 	 *
 	 * Types of Executors:
+	 * - cached : An unbounded pool where the number of threads will grow according to the tasks it needs to service. The threads are killed by a default 60 second timeout if not used and the pool shrinks back
 	 * - fixed : By default it will build one with 20 threads on it. Great for multiple task execution and worker processing
+	 * - fork_join : A pool that uses the ForkJoinPool.commonPool() by default, it is great for parallel tasks and recursive tasks
 	 * - single : A great way to control that submitted tasks will execute in the order of submission: FIFO
-	 * - cached : An unbounded pool where the number of threads will grow according to the tasks it needs to service. The threads are killed by a default 60 second timeout if not used and the pool shrinks back
 	 * - scheduled : A pool to use for scheduled tasks that can run one time or periodically
+	 * - work_stealing : A pool that allows you to run parallel tasks, it will use the ForkJoinPool.commonPool() by default
+	 * - virtual : A pool that uses the VirtualThreadPerTaskExecutor, it is great for IO bound tasks and can run many tasks in parallel without blocking, requires Java 19+.
 	 *
 	 * @see            https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
 	 * @name           The name of the executor used for registration
-	 * @type           The type of executor to build fixed, cached, single, scheduled
+	 * @type           The type of executor to build fixed, cached, single, scheduled, work_stealing, virtual
 	 * @threads        How many threads to assign to the thread scheduler, default is 20
 	 * @debug          Add output debugging
 	 * @loadAppContext Load the CFML App contexts or not, disable if not used
+	 * @parallelism	The number of parallel tasks to run at the same time, default is 0 which means no parallelism, only applies to work stealing executors
 	 *
 	 * @return The ColdBox Schedule class to work with the schedule: coldbox.system.async.executors.Executor
 	 */
@@ -83,7 +87,8 @@ component accessors="true" singleton {
 		type                   = "fixed",
 		numeric threads        = this.$executors.DEFAULT_THREADS,
 		boolean debug          = false,
-		boolean loadAppContext = true
+		boolean loadAppContext = true,
+		numeric parallelism = 0
 	){
 		// Build it if not found
 		if ( !variables.executors.keyExists( arguments.name ) ) {
@@ -100,7 +105,7 @@ component accessors="true" singleton {
 	 *
 	 * @see            https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/Executors.html
 	 *
-	 * @type           Available types are: fixed, cached, single, scheduled, {WireBoxID}
+	 * @type           Available types are: fixed, fork_join, cached, single, scheduled, work_stealing, virtual, {WireBoxID}
 	 * @threads        The number of threads to seed the executor with, if it allows it
 	 * @debug          Add output debugging
 	 * @loadAppContext Load the CFML App contexts or not, disable if not used
@@ -111,33 +116,46 @@ component accessors="true" singleton {
 		required type,
 		numeric threads,
 		boolean debug          = false,
-		boolean loadAppContext = true
+		boolean loadAppContext = true,
+		numeric parallelism = 0
 	){
 		// Factory to build the right executor
 		switch ( arguments.type ) {
+			case "cached": {
+				arguments.executor = this.$executors.newCachedThreadPool();
+				return new executors.Executor( argumentCollection = arguments );
+			}
 			case "fixed": {
 				arguments.executor = this.$executors.newFixedThreadPool( arguments.threads );
 				return new executors.Executor( argumentCollection = arguments );
 			}
-			case "cached": {
-				arguments.executor = this.$executors.newCachedThreadPool();
+			case "fork_join": {
+				arguments.executor = this.$executors.newForkJoinPool( arguments.threads );
 				return new executors.Executor( argumentCollection = arguments );
 			}
+			case "scheduled": {
+				arguments.executor = this.$executors.newScheduledThreadPool( arguments.threads );
+				return new executors.ScheduledExecutor( argumentCollection = arguments );
+			}
 			case "single": {
 				arguments.executor = this.$executors.newFixedThreadPool( 1 );
 				return new executors.Executor( argumentCollection = arguments );
 			}
-			case "scheduled": {
-				arguments.executor = this.$executors.newScheduledThreadPool( arguments.threads );
-				return new executors.ScheduledExecutor( argumentCollection = arguments );
+			case "work_stealing": {
+				arguments.executor = this.$executors.newWorkStealingPoolExecutor( arguments.parallelism );
+				return new executors.Executor( argumentCollection = arguments );
+			}
+			case "virtual" : {
+				arguments.executor = this.$executors.newVirtualThreadExecutor();
+				return new executors.Executor( argumentCollection = arguments );
 			}
 			default: {
 			}
 		}
 		throw(
 			type    = "InvalidExecutorType",
 			message = "The executor you requested :#arguments.type# does not exist.",
-			detail  = "Valid executors are: fixed, cached, single, scheduled"
+			detail  = "Valid executors are: fixed, fork_join, cached, single, scheduled, virtual, work_stealing, {WireBoxID}"
 		);
 	}
 

system/async/executors/Executor.cfc

@@ -32,6 +32,29 @@ component accessors="true" singleton {
 	// Prepare the static time unit class
 	this.$timeUnit = new coldbox.system.async.time.TimeUnit();
 
+	variables.features = {
+		"ScheduledThreadPoolExecutor": {
+			"pool": true,
+			"taskMethods": true,
+			"isTerminating": true
+		},
+		"ThreadPoolExecutor": {
+			"pool": true,
+			"taskMethods": true,
+			"isTerminating": true
+		},
+		"ForkJoinPool": {
+			"pool": false,
+			"taskMethods": false,
+			"isTerminating": true
+		},
+		"ThreadPerTaskExecutor": {
+			"pool": false,
+			"taskMethods": false,
+			"isTerminating": false
+		}
+	};
+
 	/**
 	 * Constructor
 	 *
@@ -90,6 +113,21 @@ component accessors="true" singleton {
 	 * Executor Utility Methods *
 	 ****************************************************************/
 
+	 /**
+	  * Checks if the executor has a feature based on the type and feature name.
+
+	  * @feature The feature to check for, e.g., "hasPool", "hasTaskMethods", "hasIsTerminating"
+	 *
+	 * @return  true if the feature exists for the type, false otherwise
+	  */
+	 private boolean function hasFeature( required string feature ){
+		var classType = variables.native.getClass().getSimpleName();
+		if( !variables.features.keyExists( classType ) ){
+			return false;
+		}
+		return variables.features[ classType ][ feature ];
+	}
+
 	/**
 	 * Returns true if all tasks have completed following shut down.
 	 */
@@ -102,7 +140,10 @@ component accessors="true" singleton {
 	 * not completely terminated.
 	 */
 	boolean function isTerminating(){
-		return variables.native.isTerminating();
+		if( hasFeature( "isTerminating" ) ){
+			return variables.native.isTerminating();
+		}
+		return false;
 	}
 
 	/**
@@ -210,49 +251,49 @@ component accessors="true" singleton {
 	 * Returns the approximate number of threads that are actively executing tasks.
 	 */
 	numeric function getActiveCount(){
-		return variables.native.getActiveCount();
+		return hasFeature( "taskMethods") ? variables.native.getActiveCount() : 0;
 	}
 
 	/**
 	 * Returns the approximate total number of tasks that have ever been scheduled for execution.
 	 */
 	numeric function getTaskCount(){
-		return variables.native.getTaskCount();
+		return hasFeature( "taskMethods") ? variables.native.getTaskCount() : 0;
 	}
 
 	/**
 	 * Returns the approximate total number of tasks that have completed execution.
 	 */
 	numeric function getCompletedTaskCount(){
-		return variables.native.getCompletedTaskCount();
+		return hasFeature( "taskMethods") ? variables.native.getCompletedTaskCount() : 0;
 	}
 
 	/**
 	 * Returns the core number of threads.
 	 */
 	numeric function getCorePoolSize(){
-		return variables.native.getCorePoolSize();
+		return hasFeature( "pool") ? variables.native.getCorePoolSize() : 0;
 	}
 
 	/**
 	 * Returns the largest number of threads that have ever simultaneously been in the pool.
 	 */
 	numeric function getLargestPoolSize(){
-		return variables.native.getLargestPoolSize();
+		return hasFeature( "pool") ? variables.native.getLargestPoolSize() : 0;
 	}
 
 	/**
 	 * Returns the maximum allowed number of threads.
 	 */
 	numeric function getMaximumPoolSize(){
-		return variables.native.getMaximumPoolSize();
+		return hasFeature( "pool") ? variables.native.getMaximumPoolSize() : 0;
 	}
 
 	/**
 	 * Returns the current number of threads in the pool.
 	 */
 	numeric function getPoolSize(){
-		return variables.native.getPoolSize();
+		return hasFeature( "pool") ? variables.native.getPoolSize() : 0;
 	}
 
 	/**
@@ -273,7 +314,8 @@ component accessors="true" singleton {
 			"activeCount"        : getActiveCount(),
 			"isTerminated"       : isTerminated(),
 			"isTerminating"      : isTerminating(),
-			"isShutdown"         : isShutdown()
+			"isShutdown"         : isShutdown(),
+			"type" : variables.native.getClass().getName()
 		};
 	}
 

system/async/executors/ExecutorBuilder.cfc

@@ -1,12 +1,15 @@
 /**
  * Static class to build executors from Java:
  *
+ * - CachedThreadPool
  * - FixedThreadPool
+ * - ForkJoinPool
  * - SingleThreadPool
- * - CachedThreadPool
  * - ScheduledThreadPool
+ * - WorkStealingPool
+ * - VirtualThreadPool
  *
- * @see https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html
+ * @see https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/Executors.html
  */
 component singleton {
 
@@ -57,4 +60,49 @@ component singleton {
 		return variables.jExecutors.newScheduledThreadPool( javacast( "int", arguments.corePoolSize ) );
 	}
 
+	/**
+	 * Create a virtual thread executor that can be used to run tasks in a virtual thread context.
+	 * @see https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/VirtualThreadExecutor.html
+	 * @return VirtualThreadExecutor: The newly created virtual thread executor
+	 */
+	function newVirtualThreadExecutor(){
+		return variables.jExecutors.newVirtualThreadPerTaskExecutor();
+	}
+
+	/**
+	 * Create a work stealing pool executor that can be used to run tasks in a work-stealing context.
+	 * @see https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/WorkStealingPoolExecutor.html
+	 * @return WorkStealingPoolExecutor: The newly created work stealing pool executor
+	 */
+	function newWorkStealingPoolExecutor( numeric parallelism = 0 ){
+		if ( arguments.parallelism > 0 ) {
+			return variables.jExecutors.newWorkStealingPool( javacast( "int", arguments.parallelism ) );
+		}
+		// If no parallelism is specified, use the default behavior
+		// which is to use the number of available processors.
+		// This is similar to the default behavior of the Java Executors class.
+		// This will create a work-stealing pool with a parallelism level
+		// equal to the number of available processors.
+		// This is useful for tasks that can benefit from parallel execution.
+		return variables.jExecutors.newWorkStealingPool();
+	}
+
+	/**
+	 * New ForkJoinPool executor.
+	 *
+	 * @see https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/concurrent/ForkJoinPool.html
+	 *
+	 * @param maxThreads The maximum number of threads to use in the pool, defaults to 20.
+	 *
+	 * @return ForkJoinPool: The newly created ForkJoinPool
+	 */
+	function newForkJoinPool( numeric maxThreads = this.DEFAULT_THREADS ) {
+		if( maxThreads > 0 ) {
+			return createObject( "java", "java.util.concurrent.ForkJoinPool" ).init( javacast( "int", arguments.maxThreads ) );
+		}
+		// If no maxThreads is specified, use the default behavior
+		// which is to use the number of available processors.
+		return createObject( "java", "java.util.concurrent.ForkJoinPool" ).commonPool();
+	}
+
 }

tests/specs/async/ExecutorServicesSpec.cfc

@@ -15,28 +15,56 @@ component extends="BaseAsyncSpec" {
 					var executor = asyncManager.newExecutor( "unitTest" );
 					expect( executor.getName() ).toBe( "unitTest" );
 					expect( executor.getCorePoolSize() ).toBe( 20 );
+					expect( executor.getStats() ).toBeStruct();
 				} );
 				it( "can create the default fixed executor with custom threads", function(){
 					var executor = asyncManager.newExecutor( name: "unitTest", threads: 100 );
 					expect( executor.getName() ).toBe( "unitTest" );
 					expect( executor.getCorePoolSize() ).toBe( 100 );
+					expect( executor.getStats() ).toBeStruct();
 				} );
 				it( "can create the a single executor", function(){
 					var executor = asyncManager.newExecutor( name: "unitTest", type: "single" );
 					expect( executor.getName() ).toBe( "unitTest" );
 					expect( executor.getCorePoolSize() ).toBe( 1 );
+					expect( executor.getStats() ).toBeStruct();
 				} );
 				it( "can create the a cached executor", function(){
 					var executor = asyncManager.newExecutor( name: "unitTest", type: "cached" );
 					expect( executor.getName() ).toBe( "unitTest" );
 					expect( executor.getPoolSize() ).toBe( 0 );
+					expect( executor.getStats() ).toBeStruct();
+				} );
+				it( "can create a fork_join executor", function(){
+					var executor = asyncManager.newExecutor( name: "fork_join", type: "fork_join" );
+					expect( executor.getName() ).toBe( "fork_join" );
+					expect( executor.getPoolSize() ).toBe( 0 );
+					expect( executor.getStats() ).toBeStruct();
+				} );
+				it( "can create a work_stealing executor", function(){
+					var executor = asyncManager.newExecutor( name: "work_stealing", type: "work_stealing" );
+					expect( executor.getName() ).toBe( "work_stealing" );
+					expect( executor.getPoolSize() ).toBe( 0 );
+					expect( executor.getStats() ).toBeStruct();
 				} );
 				it( "can create a scheduled executor", function(){
 					var executor = asyncManager.newExecutor( name: "unitTest", type: "scheduled" );
 					expect( executor.getName() ).toBe( "unitTest" );
 					expect( executor.getCorePoolSize() ).toBe( 20 );
 					expect( executor.getNative().toString() ).toInclude( "ScheduledThreadPoolExecutor" );
-				} );
+					expect( executor.getStats() ).toBeStruct();
+				} );
+				// Skip on Adobe as their dumb reflection does not support virtual threads
+				it(
+					title: "can create a virtual thread executor",
+					skip: ( server.keyExists( "coldfusion" ) && server.coldfusion.productName.findNoCase( "ColdFusion" ) ),
+					body: function(){
+						var executor = asyncManager.newExecutor( name: "virtual", type: "virtual" );
+						expect( executor.getName() ).toBe( "virtual" );
+						expect( executor.getPoolSize() ).toBe( 0 );
+						expect( executor.getStats() ).toBeStruct();
+					}
+				);
 				it( "can throw an exception if the wrong type is passed", function(){
 					expect( function(){
 						asyncManager.newExecutor( name: "unitTest", type: "bogus" );
@@ -117,7 +145,7 @@ component extends="BaseAsyncSpec" {
 					var executor1 = asyncManager.newExecutor( "unitTest1" );
 					var statusMap = asyncManager.getExecutorStatusMap( "unitTest1" );
 
-					// debug( statusMap );
+					debug( statusMap );
 					expect( statusMap.isShutdown ).toBeFalse();
 				} );
 			} );