Skip to content

ExecutorService

Jump to bottom
Dominic Watson edited this page Jul 11, 2018 · 1 revision

ExecutorService

ExecutorService is the core Service in java.util.concurrent. If you've used CFThread, an ExecutorService is the closest conceptual match. It represents a configurable pool of threads that asynchronously run your Tasks. You can interact with it in the following ways:

  1. Submit-and-forget: You submit tasks, the service runs them asynchronously, and you ignore the results
  2. Submit-and-wait: You submit tasks, the service runs them asynchronously, and you wait for the results (with a configurable timeout)
  3. Invoke-multiple: You submit an array of tasks, the service runs them all asynchronously, and returns an array of Futures representing the results
  4. Invoke-any: You submit an array of tasks, the service runs them asynchronously and returns the first successful raw result

If you're familiar with CFThread, note that the "and-wait" idiom is similar to using thread action="join"

Under most circumstances, you'll need only one ExecutorService for your application. You can use all of the above submission techniques with that single instance, and you can submit heterogeneous tasks to that single instance. You could replace your current CFThread usage with a single ExecutorService with very little effort

Java APIs

http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ExecutorService.html

http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Future.html

In Action

To see the ExecutorService in action, with all scenarios discussed below, download CFConcurrent and run examples/ExecutorService/index.cfm

You may also be interested in

If you wish to decouple the execution of tasks from the processing of results, check out ExecutorCompletionService. If you're interested in creating periodic heartbeat-style Tasks, check out ScheduledThreadPoolExecutor.

Create the ExecutorService

In most cases, you'll create a single ExecutorService as an application-scoped variable in Application.cfc. Below shows how to do it in straight CF... If you use ColdSpring, WireBox, or DI/1, you'll know what to do. Note that this Application.cfc first inits the ExecutorService instance, then start()s the instance. Finally, onApplicationStop shuts down the instance.

component {

	this.name = "yourAwesomeApp";

	function onApplicationStart(){
		//a maxConcurrent of 0 will cause the service to default to the number of Available Processors + 1
		application.executorService = createObject("component", "cfconcurrent.ExecutorService")
			.init( serviceName = "executorServiceExample", maxConcurrent = 0, maxWorkQueueSize = 100000);
		application.executorService.setLoggingEnabled( true );
		application.executorService.start();
	}

	function onRequestStart(){
		if( structKeyExists(url, "reinit") and url.reinit eq "hotdawg" ){
			applicationStop();
			onApplicationStop();
		}
	}

	function onApplicationStop(){
		application.executorService.stop();
	}
}

Determining Appropriate maxConcurrent and workQueueSize Parameters

See Determining Configuration Options for guidance

Task Overview

As discussed on Page 1, we use Tasks for everything. These are CFCs with a result-returning call() method. Please read the documentation for Tasks for more detail.

I assume at this point you know what a Task is. Briefly, let's review some typical scenarios:

  1. Use CFFeed to process RSS feeds
  2. Upload a file and "do stuff" with it.... virus scan, create thumbnails (if it's an image), parse metadata, etc
  3. Accept a directory path and zip all the files in that directory
  4. "Log Stuff" at the end of every request
  5. Accept a PDF upload, turn each page into separate image thumbnails (lucky you)
  6. Accept a directory path, and compute the total file size of all files in all subdirectories
  7. And so on...

When thinking about Tasks, think: "Could I break this into multiple steps that would benefit from concurrent execution".

And, when thinking about Tasks, it's imperative that you limit shared mutability. In other words, limit how much each Task needs to know about other Tasks. The Best Case is when a Task only knows about the work it needs to do, and nothing else.

Using the ExecutorService instance

By now you should have a Task in mind, and presumably it's modeled in a Task CFC. Now, let's process that Task asynchronously, using the submission techniques described at the start of this page.

Submit-and-forget

You'll create an instance of the task, and then submit it to the service. That's it.

task = new HelloTask( args );
application.executorService.submit( task );

Submit-and-wait

You'll create an instance of the task and submit it to the service. This will return a Future, which is a kind of promise that your Task will complete eventually and return a result. You'll call Future.get(), which waits for your Task to complete and return a result. get() will return the result of your call() method:

task = new HelloTask( args );
future = application.executorService.submit( task );
callResultWithNoTimeout = future.get();

In this scenario, if HelloTask.call() returns a CF Struct, then callResultWithNoTimeout will represent that struct upon completion.

Submit-and-wait With Timeout

You can specify a timeout, such that if the task does not return in a certain amount of time, execution is cancelled. In that case, Future.get() will return an exception, which you should wrap in try/catch; otherwise, the get() call will throw an exception.

task = new HelloTask( args );
future = application.executorService.submit( task );
callResultWithTimeout = future.get( 200, application.executorService.getObjectFactory().MILLISECONDS );

In this example, if the task does not complete in 200 milliseconds, execution is cancelled

Invoke Multiple

Often, you have groups of tasks, and you wish to have them run asynchronously, and then you post-process the results. Perhaps this is a computation. Perhaps it's the RSS example above, where all "fetch" Tasks are run asynchronously, and then after all complete you post-process them in the same request. This is your typical start(); join(); idiom, which typically looks like this:

for( some loop construct ){
  thread name="mythread_#i#" action="run"{
      //do stuff
  }
}

thread action="join";

Using CFConcurrent, you'll represent each unit of work as a separate Task, and you'll submit the group of Tasks to the ExecutorService via invokeAll():

tasks = [];
//DirectorySizeTask computes the size of all its subdirectories and returns a struct with several keys, one of which is directorySize
for( directory in directories ){
	arrayAppend( tasks, new DirectorySizeTask( directory ) );
}

futures = application.executorService.invokeAll( tasks );
for( future in futures ){
	totalDirectorySize += future.get().directorySize;
}

Invoke Any

Perhaps you will find a use for this. Unlike the above examples, this does not return a Future but instead returns the result of the first completed Task's Future.get() method. In other words, it directly returns the results of the call method:

tasks = [];
for( i = 1; i <= 10; i++ ){
	arrayAppend( tasks, new HelloTask( args ) );
}

result = application.executorService.invokeAny( tasks );
//since HelloTask.call() returns a struct, result will be the returned struct from the first completed task

Future.get() and Exceptions

In all of the cases above where we use Future.get() -- which represents the result of the call() method, we normally are dealing with the expected result of the call method... perhaps a simple value, perhaps a struct, perhaps an object. However, if the call() method throws an error, or if the call method is cancelled as a result of a timeout, it will return an exception. When this happens, Future.get() will throw that exception. This is absolutely the behavior you want, but you do need to be prepared for it.

  1. In your call() method, wrap the entire thing in try/catch to minimize chances of throwing an error
  2. If you're using Future.get( someTimeout ), know that if that timeout is tripped, the get() call will cause an exception.

When using get() with a timeout, you'll want to wrap it in try/catch. In the example below, we give it only 15 milliseconds to complete all tasks, an unrealistic number designed to ensure that at least some of the tasks throw a CancelledException. In this case, some tasks will have completed, and some will have been cancelled.

Correct programs will expect this behavior and will be written to account for it. Thus, the code that processes the results will need to be responsive to errors.

The good news here is that this is trivial to unit test. Your unit tests needn't know at all about concurrency, Tasks, or the execution environment. In fact, if you have a "ResultsProcessor" object or chunk of code, all it needs to know about is an array of results... some of those results will be the expected object/struct/whatever, and some will be an exception, and that scenario is simple to construct in a test.

futures = application.executorService.invokeAll( tasks, 15, application.executorService.getObjectFactory().MILLISECONDS );
invokeAllResultsWithTimeout = [];
for( future in futures ){
	try{
		arrayAppend( invokeAllResultsWithTimeout, future.get() );
	} catch( any e ){
		arrayAppend( invokeAllResultsWithTimeout, duplicate(e) );
	}
}

Clone this wiki locally