Tasks

Tasks

Imagine a scenario where you need to scour dozens of directories and delete files of a certain age. Or scrape dozens of web pages, parse the results, and "do stuff" with those results. Or simply flush logging to the database at the end of a web request.

Those are tasks. Tasks are fine-grained units of work that can typically be done independent of other work happening concurrently in the system. The less they know about their operating environment -- the less state they share and manage -- the easier they will be to program and test.

Now, imagine you want to create many of these tasks and have them executed concurrently. You want control over the number of max concurrent tasks allowed to run at once... perhaps you know that your server(s) can handle dozens of concurrent Tasks because most of the work is IO, not CPU, intensive. Perhaps you want to post-process the results of those tasks, rather than simply throw them into the machine and hope they work. Perhaps you want to submit a group of tasks for concurrent processing, and then post-process that group of tasks when all of them are finished. Or perhaps you have a heartbeat style task that you wish to run every second, whose purpose is to find work to do and schedule other, separate concurrently running Tasks.

These are the problems the Java Concurrency Framework Solves. CFConcurrent helps you do this in 100% CFML.

In CFConcurrent, Tasks are ColdFusion Components. You can initialize then with any data you want, just like any other CFC. The only restriction is that the method that does the work must be called call(). This is not a restriction imposed by CFConcurrent; rather, it's how you implement the Callable interface, which is necessary to pass your CFC instance to a Java Concurrency Framework Executor Service for processing. It is entirely appropriate that your Task CFCs are simply wrappers around other objects in your system, and call() merely delegates work to those components.

Tasks must not be singletons; they must be transient. In other words, every task you submit to the JCF must be a new object.

To bridge the gap between CF and Java, CFConcurrent will create a proxy for your CFC -- using JavaLoader on CF9 or native on CF10 -- which enables your CFC to be passed to a Java object that expects an interface.

Java APIs

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

Structure of a Task

Here's a simple CFC that adheres to the Callable interface, written in a style which I prefer, which is to maintain a Structure of data and return that structure from the call() method. You might also to create a Result object rather than a struct. Here's HelloTask.cfc:

component{
	results = { created = now(), createTS = getTickCount(), error={} };

	function init( id ){
		structAppend( results, arguments );
		return this;
	}

	function call(){
		try{
			results.message = "Hidey ho!";
		} catch( any e ){
			writeLog("OH NOES!!!!! #e.message#; #e.detail#");
			results.error = e;
		}
		results.endTS = getTickCount();
		return results;
	}
}

I strongly recommend wrapping the contents of your call() in a try block. If an error occurs, store it in the result that you return.

All of your Task objects will follow the same format as above, whenever the JCF expects a Callable.

When do I need to use run() instead of call()?

In the JCF JavaDocs, some methods will require a Runnable instead of a Callable. Runnable objects will implement a run() method, and that method will not return a result (i.e. it is void). The JCF will require a Runnable in cases where it does not make sense to have a result-bearing Task. Examples are the fire-and-forget ExecutorService.execute( Runnable ) method and the ScheduledExecutorService.scheduleWithFixedDelay( Runnable ) and ScheduledExecutorService.scheduleAtFixedRate( Runnable ) methods.

When working with CFConcurrent, method signatures and documentation will indicate when a Runnable is required. When working with the underlying Java objects directly, the JavaDocs will tell you whether a Callable or Runnable is required.

Futures

In the JavaDocs for Executor Services, you'll see that methods accepting a Callable, such as submit(), return a Future (http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Future.html).

A Future represents the result of your task's execution. You fetch that result with Future.get().

Look up above at that sample Task. It returns a struct named results, containing several fields.

The call to Future.get() will return that results struct to you.

Putting it all together

For now, we'll ignore the couple of lines of code it takes to construct the service. It's enough to know that the service variable here is an ExecutorService. Brace yourself, this is tough:

task = new HelloTask( "someID" );
future = application.executorService.submit( task );
result = future.get();

That's it. Notice how this looks like any method call you've ever made. The difference is that when you submit() your task to the executorService, it's running concurrently with 0, 1, 10, 100, who-knows-how-many other similar tasks.

This is important: future.get() will block until your task's call() method is executed and returns a result. When you submit your task to the service, it might sit around awhile waiting to execute. Sometimes this will not be important to you. Sometimes it will, and at that time, you'll be interested in using a ExecutorCompletionService.

Now, let's imagine you want to run that task, but if it doesn't complete in a certain amount of time, you want it cancelled. Here's how:

task = new HelloTask( url.sleepTime );
future = application.executorService.submit( task );
result = future.get( 100, application.executorService.getObjectFactory().MILLISECONDS );

In this case, we give it 100ms to execute. If it doesn't finish in that time, then the call will cancel and our result will contain an exception

Conclusion

Tasks are CFCs. They implement a call() or a run() method. call() methods return results; run() methods do not return a result.

You pass a task instance to methods of Executor Service objects. They return a Future, and you retrieve the results with Future.get(). It can accept a timeout such that it throws an exception if the timeout occurs before the result is available. Otherwise, the result will be whatever you returned from your call() method.