1. Introduction
The world of mainstream computing is changing rapidly these days. If you open the hood and look under the covers of your computer, you'll most likely see a dual-core processor there. Or a quad-core, if you're lucky enough. We all run our software on multi-processors. The code we write today and tomorrow will probably never run on a single processor system. Parallel hardware has become common-place. Not so with the software though, at least not yet. People still create single-threaded code, although it will never be able to leverage the full power of future hardware. Some experiment with low-level concurrency primitives, like threads, locks or synchronized blocks, however, it has become obvious that the common shared-memory multithreading causes more troubles than it solves. Low-level concurrency handling is usually hard to get right. And it's not much fun either. With such a radical change in hardware, software inevitably has to change dramatically too. Higher-level concurrency concepts like map/reduce, fork/join, actors or dataflow will provide natural abstractions for different types of problem domains while leveraging the multi-core hardware underneath.Meet GPars - an open-source concurrency library for Java and Groovy that aims to give you multiple high-level abstractions for writing concurrent code in Groovy - map/reduce, fork/join, asynchronous closures, actors, agents, dataflow concurrency and other concepts, which aim to make your Groovy code concurrent with little effort. With GPars your Groovy or Java code can easily utilize all the available processors on the target system. You can run multiple calculations at the same time, request network resources in parallel, safely solve hierarchical divide-and-conquer problems, perform functional style map/reduce collection processing or build your applications around the actor model.The project is open sourced under the Apache 2 License . If you're working on a commercial, open-source, educational or any other type of software project in Groovy, download the binaries or integrate them from the maven repository and get going. The way to witting highly concurrent Groovy code is wide open. Enjoy!1.1. Credits
This project could not have reached the point where we stand currently, without all the great help and contribution of many individuals, who have devoted their time, energy and expertise to make GPars a solid product. First, it is the people in the core team, who should be mentioned:- Václav Pech
- Dierk Koenig
- Alex Tkachman
- Russel Winder
- Paul King
- Jon Kerridge
- Hamlet d'Arcy
- Hans Dockter
- Guillaume Laforge
- Robert Fischer
- Johannes Link
- Graeme Rocher
- Alex Miller
- Jeff Gortatowsky
- Jiří Kropáček
2. Getting Started
Let's make several assumptions before we really start.- You know and love Groovy. Otherwise you'd hardly invest your valuable time into studying a Groovy concurrency library.
- If you don't want to use Groovy, you are prepared to pay the inevitable verbosity tax on using GPars from Java
- You target multi-core hardware with your code
- You use or want to use Groovy or Java to write concurrent code.
- You have at least some understanding that in concurrent code some things can happen at any time in any order and often more of them at the same time.
Brief overview
GPars aims to bring several useful concurrency abstractions to Java and Groovy developers. It's becoming obvious that dealing with concurrency on the thread/synchronized/lock level, as provided by the JVM, is way too low level to be safe and comfortable. Many high-level concepts, like actors or dataflow concurrency have been around for quite some time, since parallel computers had been in use in computer centers long before multi-core chips hit the hardware mainstream. Now, however, it's the time to adopt and test these abstractions for the mainstream software industry.The concepts available in GPars can be categorized into three main groups:- Code-level helpers - constructs that can be applied to small parts of the code-base such as individual algorithms or data structures without any major changes in the overall project architecture
- Parallel Collections
- Asynchronous Processing
- Fork/Join (Divide/Conquer)
- Architecture-level concepts - constructs that need to be taken into account when designing the project structure
- Actors
- Communicating Sequential Processes
- Dataflow Concurrency
- Shared Mutable State Protection - although about 95 of current use of shared mutable state can be avoided using proper abstractions, good abstractions are still necessary for the remaining 5% use cases, when shared mutable state can't be avoided
- Agents
- Software Transactional Memory (not implemented in GPars yet) would also belong to this group
2.1 Downloading and Installing
There are several ways to add GPars to your project. Either download and add all the jar files manually, specify a dependency in Maven, Ivy or Gradle build files or use Grape. If you're building a Grails or a Griffon application, you can leverage the appropriate plugins to fetch the jar files for you.Please visit the Integration page of the project for details.2.2 A Hello World Example
Once you got setup, try the following Groovy script to test that your setup is functional. For Java, see below.import static groovyx.gpars.actor.Actors.actor/** * A demo showing two cooperating actors. The decryptor decrypts received messages and replies them back. * The console actor sends a message to decrypt, prints out the reply and terminates both actors. * The main thread waits on both actors to finish using the join() method to prevent premature exit, * since both actors use the default actor group, which uses a daemon thread pool. * @author Dierk Koenig, Vaclav Pech */def decryptor = actor { loop { react {message -> if (message instanceof String) reply message.reverse() else stop() } } }def console = actor { decryptor.send 'lellarap si yvoorG' react { println 'Decrypted message: ' + it decryptor.send false } }[decryptor, console]*.join()
You should get a message "Decrypted message: Groovy is parallel" printed out on the console when you run the code.
GPars - a Java library
Although GPars has been primarily designed for the Groovy programming language, the solid technical foundation plus good performance characteristics make GPars a good Java library as well. Since most of GPars is written in Java, there is no extra performance penalty Java applications would pay when using GPars.For details please refer to the Java API section.To quick-test your integration through Java API, run the following Java actor code:
import groovyx.gpars.MessagingRunnable; import groovyx.gpars.actor.DynamicDispatchActor;public class StatelessActorDemo { public static void main(String[] args) throws InterruptedException { final MyStatelessActor actor = new MyStatelessActor(); actor.start(); actor.send("Hello"); actor.sendAndWait(10); actor.sendAndContinue(10.0, new MessagingRunnable<String>() { @Override protected void doRun(final String s) { System.out.println("Received a reply " + s); } }); } }class MyStatelessActor extends DynamicDispatchActor { public void onMessage(final String msg) { System.out.println("Received " + msg); replyIfExists("Thank you"); } public void onMessage(final Integer msg) { System.out.println("Received a number " + msg); replyIfExists("Thank you"); } public void onMessage(final Object msg) { System.out.println("Received an object " + msg); replyIfExists("Thank you"); } }
2.3 Code conventions
We follow certain conventions in the code samples. Understanding these may help you read and comprehend GPars code samples better.- The leftShift operator << has been overloaded on actors, agents and dataflow expressions (both variables and streams) to mean send a message or assign a value.
myActor << 'message'myAgent << {account -> account.add('5 USD')}myDataFlowVariable << 120332
- On actors and agents the default call() method has been also overloaded to mean send . So sending a message to an actor or agent may look like a regular method call.
myActor "message"myAgent {house -> house.repair()}
- The rightShift operator >> in GPars has the when bound meaning. So
myDataFlowVariable >> {value -> doSomethingWith(value)}
- GParsPool.withPool()
- GParsPool.withExistingPool()
- GParsExecutorsPool.withPool()
- GParsExecutorsPool.withExistingPool()
- Actors.actor()
- Actors.reactor()
- Actors.fairReactor()
- Actors.messageHandler()
- Actors.fairMessageHandler()
- Agent.agent()
- Agent.fairAgent()
- DataFlow.task()
- DataFlow.operator()
2.4 Getting Set-up in an IDE
Adding the GPars jar files to your project or defining the appropriate dependencies in pom.xml should be enough to get you started with GPars in your IDE.GPars DSL recognition
IntelliJ IDEA in both the free Community Edition and the commercial Ultimate Edition will recognize the GPars domain specific languages, complete methods like eachParallel() , reduce() or callAsync() and validate them. GPars uses the GroovyDSL mechanism, which teaches IntelliJ IDEA the DSLs as soon as the GPars jar file is added to the project.2.5 Applicability of concepts
Here you could find basic guide-lines helping you decide on which GPars abstraction to apply to your code at hands.- You're looking at a collection, which needs to be iterated or processed using one of the many beautiful Groovy collections method, like each() , collect() , find() and such. Proposing that processing each element of the collection is independent of the other items, using GPars parallel collections can be recommended.
- If you have a long-lasting calculation , which may safely run in the background, use the asynchronous invocation support in GPars. You can also benefit, if your long-calculating closures need to be passed around and yet you'd like them not to block the main application thread.
- You need to parallelize an algorithm at hand. You can identify sub-tasks and you're happy to explicitly express the options for parallelization. You create internally sequential tasks, each of which can run concurrently with the others, providing they all have a way to exchange data at some well-defined moments through communication channels with safe semantics. Use GPars dataflow tasks, variables and streams.
- You can't avoid shared mutable state. Multiple threads will be accessing shared data and (some of them) modifying the data. Traditional locking and synchronized approach feels too risky or unfamiliar. Go for agents, which will wrap your data and serialize all access to it.
- You're building a system with high concurrency demands. Tweaking a data structure here or task there won't cut it. You need to build the architecture from the ground up with concurrency in mind. Message-passing might be the way to go.
- Groovy CSP will give you highly deterministic and composable model for concurrent processes.
- If you're trying to solve a complex data-processing problem, consider GPars dataflow operator to build a data flow network.
- Actors will shine if you need to build a general-purpose, highly concurrent and scalable architecture.
2.6 What's new
Again, the new release, this time GPars 0.11, introduces a lot of gradual enhancements and improvements on top of the previous release.Check out the JIRA release notesProject changes
See the Breaking Changes listing for the list of breaking changes.
Parallel collections
- Added a speculate() method to concurrently run multiple functions each representing a different path to achieve the required goal, while harvesting the first calculated result
- Added the combine method into the map/reduce family of functions
- General speed-up of Parallel Array construction and execution
Fork / Join
- Nothing new, perhaps we've reached a maturity point here
Actors
- Speed-up of the actor messaging core
- Refactored Reactor and DynamicDispatchActor achieving a 15x speed-up
- Conditional and counting loop
- Changed the message timeout API
- Deprecated _AbstractPooledActor_
- Added DefaultActor as a new implementation of the stateful actor principle
- Removed support for replies and senders on messages
- Extended support for senders to actors themselves
- Removed the receive() from _DefaultActor_
Dataflow
- New specialized operators - selector , prioritySelector and _splitter_
- Added select and prioritySelect methods to select a value from multiple channels
- Dataflow tasks accept Callable as parameters on top of Closures
- Introduced dataflow queues
- Unified the interface for dataflow variables, streams and queues - _DataFlowChannel_
- Methods for atomic update to multiple operator's output channels
- Fixed issues with reusing the same parallel group in the DataFlowVariable whenBound() handlers
- Tasks can return values and can be joined
- DataFlowQueue's whenBound renamed to wheneverBound and whenNextBound renamed to whenBound to obey the DataFlowChannel interface contract
- DataFlow default parallel group has been made daemon
Agent
- Added listeners and validators
- Considerable speed-up
Other
- Java API - GPars can now be used from Java
- A stand-alone maven-based Java API demo application was added to show GPars integration and use from Java
Renaming hints
- The AbstractPooledActor class is now deprecated, use DefaultActor instead
- The DataFlowStream class has been renamed to _DataFlowQueue_
- The memoize method family has been renamed to gmemoize in order to avoid name collisions with memoize in Groovy 1.8
2.7 Java API - Using GPars from Java
Using GPars is very addictive, I guarantee. Once you get hooked you won't be able to code without it. May the world force you to write code in Java, you will still be able to benefit from most of GPars features.Java API specifics
Some parts of GPars are irrelevant in Java and it is better to use the underlying Java libraries directly:- Parallel Collection - use jsr-166y library's Parallel Array directly
- Fork/Join - use jsr-166y library's Fork/Join support directly
- Asynchronous functions - use Java executor services directly
GPars Closures in Java API
To overcome the lack of closures as a language element in Java and to avoid forcing users to use Groovy closures directly through the Java API, a few handy wrapper classes have been provided to help you define callbacks, actor body or dataflow tasks.- groovyx.gpars.MessagingRunnable - used for single-argument callbacks or actor body
- groovyx.gpars.ReactorMessagingRunnable - used for ReactiveActor body
- groovyx.gpars.DataFlowMessagingRunnable - used for dataflow operators' body
Actors
The DynamicDispatchActor as well as the ReactiveActor classes can be used just like in Groovy:import groovyx.gpars.MessagingRunnable; import groovyx.gpars.actor.DynamicDispatchActor; public class StatelessActorDemo { public static void main(String[] args) throws InterruptedException { final MyStatelessActor actor = new MyStatelessActor(); actor.start(); actor.send("Hello"); actor.sendAndWait(10); actor.sendAndContinue(10.0, new MessagingRunnable<String>() { @Override protected void doRun(final String s) { System.out.println("Received a reply " + s); } }); } } class MyStatelessActor extends DynamicDispatchActor { public void onMessage(final String msg) { System.out.println("Received " + msg); replyIfExists("Thank you"); } public void onMessage(final Integer msg) { System.out.println("Received a number " + msg); replyIfExists("Thank you"); } public void onMessage(final Object msg) { System.out.println("Received an object " + msg); replyIfExists("Thank you"); } }
import groovy.lang.Closure; import groovyx.gpars.ReactorMessagingRunnable; import groovyx.gpars.actor.Actor; import groovyx.gpars.actor.ReactiveActor;public class ReactorDemo { public static void main(final String[] args) throws InterruptedException { final Closure handler = new ReactorMessagingRunnable<Integer, Integer>() { @Override protected Integer doRun(final Integer integer) { return integer * 2; } }; final Actor actor = new ReactiveActor(handler); actor.start(); System.out.println("Result: " + actor.sendAndWait(1)); System.out.println("Result: " + actor.sendAndWait(2)); System.out.println("Result: " + actor.sendAndWait(3)); } }
Convenience factory methods
Obviously, all the essential factory methods to build actors quickly are available where you'd expect them.import groovy.lang.Closure; import groovyx.gpars.ReactorMessagingRunnable; import groovyx.gpars.actor.Actor; import groovyx.gpars.actor.Actors;public class ReactorDemo { public static void main(final String[] args) throws InterruptedException { final Closure handler = new ReactorMessagingRunnable<Integer, Integer>() { @Override protected Integer doRun(final Integer integer) { return integer * 2; } }; final Actor actor = Actors.reactor(handler); System.out.println("Result: " + actor.sendAndWait(1)); System.out.println("Result: " + actor.sendAndWait(2)); System.out.println("Result: " + actor.sendAndWait(3)); } }
Agents
import groovyx.gpars.MessagingRunnable; import groovyx.gpars.agent.Agent; public class AgentDemo { public static void main(final String[] args) throws InterruptedException { final Agent counter = new Agent<Integer>(0); counter.send(10); System.out.println("Current value: " + counter.getVal()); counter.send(new MessagingRunnable<Integer>() { @Override protected void doRun(final Integer integer) { counter.updateValue(integer + 1); } }); System.out.println("Current value: " + counter.getVal()); } }
Dataflow Concurrency
Both DataFlowVariables and DataFlowQueues can be used from Java without any hiccups. Just avoid the handy overloaded operators and go straight to the methods, like bind , whenBound_, _getVal and other. You may also continue using dataflow tasks passing to them instances of Runnable or Callable just like groovy Closure .import groovyx.gpars.MessagingRunnable; import groovyx.gpars.dataflow.DataFlowVariable; import groovyx.gpars.group.DefaultPGroup;import java.util.concurrent.Callable;public class DataFlowTaskDemo { public static void main(final String[] args) throws InterruptedException { final DefaultPGroup group = new DefaultPGroup(10); final DataFlowVariable a = new DataFlowVariable(); group.task(new Runnable() { public void run() { a.bind(10); } }); final DataFlowVariable result = group.task(new Callable() { public Object call() throws Exception { return (Integer)a.getVal() + 10; } }); result.whenBound(new MessagingRunnable<Integer>() { @Override protected void doRun(final Integer integer) { System.out.println("arguments = " + integer); } }); System.out.println("result = " + result.getVal()); } }
Dataflow operators
The sample below should illustrate the main differences between Groovy and Java API for dataflow operators.- Use the convenience factory methods accepting list of channels to create operators or selectors
- Use DataFlowMessagingRunnable to specify the operator body
- Call getOwningProcessor() to get hold of the operator from within the body in order to e.g. bind output values
import groovyx.gpars.DataFlowMessagingRunnable; import groovyx.gpars.dataflow.DataFlow; import groovyx.gpars.dataflow.DataFlowQueue; import groovyx.gpars.dataflow.operator.DataFlowProcessor;import java.util.Arrays; import java.util.List;public class DataflowOperatorDemo { public static void main(final String[] args) throws InterruptedException { final DataFlowQueue stream1 = new DataFlowQueue(); final DataFlowQueue stream2 = new DataFlowQueue(); final DataFlowQueue stream3 = new DataFlowQueue(); final DataFlowQueue stream4 = new DataFlowQueue(); final DataFlowProcessor op1 = DataFlow.selector(Arrays.asList(stream1), Arrays.asList(stream2), new DataFlowMessagingRunnable(1) { @Override protected void doRun(final Object[] objects) { getOwningProcessor().bindOutput(2*(Integer)objects[0]); } }); final List secondOperatorInput = Arrays.asList(stream2, stream3); final DataFlowProcessor op2 = DataFlow.operator(secondOperatorInput, Arrays.asList(stream4), new DataFlowMessagingRunnable(2) { @Override protected void doRun(final Object[] objects) { getOwningProcessor().bindOutput((Integer) objects[0] + (Integer) objects[1]); } }); stream1.bind(1); stream1.bind(2); stream1.bind(3); stream3.bind(100); stream3.bind(100); stream3.bind(100); System.out.println("Result: " + stream4.getVal()); System.out.println("Result: " + stream4.getVal()); System.out.println("Result: " + stream4.getVal()); op1.stop(); op2.stop(); } }
Performance
In general, GPars overhead is identical irrespective of whether you use it from Groovy or Java and tends to be very low. GPars actors, for example, can compete head-to-head with other JVM actor options, like Scala actors.Since Groovy code in general runs slower than Java code, mainly due to dynamic method invocation, you might consider writing your code in Java to improve performance. Typically numeric operations or frequent fine-grained method calls within a task or actor body may benefit from a rewrite into Java.Prerequisites
All the GPars integration rules apply to Java projects just like they do to Groovy projects. You only need to include the groovy distribution jar file in your project and all is clear to march ahead. You may also want to check out the sample Java Maven project to get tips on how to integrate GPars into a maven-based pure Java application - Sample Java Maven Project3. Data Parallelism
Focusing on data instead of processes helps a great deal to create robust concurrent programs. You as a programmer define your data together with functions that should be applied to it and then let the underlying machinery to process the data. Typically a set of concurrent tasks will be created and then they will be submitted to a thread pool for processing.In GPars the GParsPool and GParsExecutorsPool classes give you access to low-level data parallelism techniques. While the GParsPool class relies on the jsr-166y Fork/Join framework and so offers greater functionality and better performance, the GParsExecutorsPool uses good old Java executors and so is easier to setup in a managed or restricted environment.There are three fundamental domains covered by the GPars low-level data parallelism:- Processing collections concurrently
- Running functions (closures) asynchronously
- Performing Fork/Join (Divide/Conquer) algorithms
3.1 Parallel Collections
Dealing with data frequently involves manipulating collections. Lists, arrays, sets, maps, iterators, strings and lot of other data types can be viewed as collections of items. The common pattern to process such collections is to take elements sequentially, one-by-one, and make an action for each of the items in row.Take, for example, the min() function, which is supposed to return the smallest element of a collection. When you call the min() method on a collection of numbers, the caller thread will create an accumulator or so-far-the-smallest-value initialized to the minimum value of the given type, let say to zero. And then the thread will iterate through the elements of the collection and compare them with the value in the accumulator . Once all elements have been processed, the minimum value is stored in the accumulator .This algorithm, however simple, is totally wrong on multi-core hardware. Running the min() function on a dual-core chip can leverage at most 50% of the computing power of the chip. On a quad-core it would be only 25%. Correct, this algorithm effectively wastes 75% of the computing power of the chip.Tree-like structures proved to be more appropriate for parallel processing. The min() function in our example doesn't need to iterate through all the elements in row and compare their values with the accumulator . What it can do instead is relying on the multi-core nature of your hardware. A parallel_min() function could, for example, compare pairs (or tuples of certain size) of neighboring values in the collection and promote the smallest value from the tuple into a next round of comparison. Searching for minimum in different tuples can safely happen in parallel and so tuples in the same round can be processed by different cores at the same time without races or contention among threads.Meet Parallel Arrays
The jsr-166y library brings a very convenient abstraction called Parallel Arrays . GPars leverages the Parallel Arrays implementation in several ways. The GParsPool and GParsExecutorsPool classes provide parallel variants of the common Groovy iteration methods like each() , collect() , findAll() and such.def selfPortraits = images.findAllParallel{it.contains me}.collectParallel {it.resize()}
def smallestSelfPortrait = images.parallel.filter{it.contains me}.map{it.resize()}.min{it.sizeInMB}
3.1.1 GParsPool
Use of GParsPool - the JSR-166y based concurrent collection processorUsage of GParsPool
The GParsPool class enables a ParallelArray-based (from JSR-166y) concurrency DSL for collections and objects.Examples of use://summarize numbers concurrently GParsPool.withPool { final AtomicInteger result = new AtomicInteger(0) [1, 2, 3, 4, 5].eachParallel {result.addAndGet(it)} assertEquals 15, result } //multiply numbers asynchronously GParsPool.withPool { final List result = [1, 2, 3, 4, 5].collectParallel {it * 2} assert ([2, 4, 6, 8, 10].equals(result)) }
//check whether all elements within a collection meet certain criteria GParsPool.withPool(5) {ForkJoinPool pool -> assert [1, 2, 3, 4, 5].everyParallel {it > 0} assert ![1, 2, 3, 4, 5].everyParallel {it > 1} }
withPool(10) {...} withPool(20, exceptionHandler) {...}
withPool { assert [1, 2, 3, 4, 5].everyParallel {it > 0} assert ![1, 2, 3, 4, 5].everyParallel {it > 1} }
- eachParallel()
- eachWithIndexParallel()
- collectParallel()
- findAllParallel()
- findAnyParallel
- findParallel()
- everyParallel()
- anyParallel()
- grepParallel()
- groupByParallel()
- foldParallel()
- minParallel()
- maxParallel()
- sumParallel()
- splitParallel()
- countParallel()
- foldParallel()
Meta-class enhancer
As an alternative you can use the ParallelEnhancer class to enhance meta-classes of any classes or individual instances with the parallel methods.import groovyx.gpars.ParallelEnhancerdef list = [1, 2, 3, 4, 5, 6, 7, 8, 9]
ParallelEnhancer.enhanceInstance(list)
println list.collectParallel {it * 2 }def animals = ['dog', 'ant', 'cat', 'whale']
ParallelEnhancer.enhanceInstance animals
println (animals.anyParallel {it ==~ /ant/} ? 'Found an ant' : 'No ants found')
println (animals.everyParallel {it.contains('a')} ? 'All animals contain a' : 'Some animals can live without an a')
Exception handling
If an exception is thrown while processing any of the passed-in closures, the first exception gets re-thrown from the xxxParallel methods and the algorithm stops as soon as possible.The exception handling mechanism of GParsPool builds on the one built into the Fork/Join framework. Since Fork/Join algorithms are by nature hierarchical, once any part of the algorithm fails, there's usually little benefit from continuing the computation, since some branches of the algorithm will never return a result.Bear in mind that the GParsPool implementation doesn't give any guarantees about its behavior after a first unhandled exception occurs, beyond stopping the algorithm and re-throwing the first detected exception to the caller. This behavior, after all, is consistent with what the traditional sequential iteration methods do.
Transparently parallel collections
On top of adding new xxxParallel() methods, GPars can also let you change the semantics of the original iteration methods. For example, you may be passing a collection into a library method, which will process your collection in a sequential way, let say using the collect() method. By changing the semantics of the collect() method on your collection you can effectively parallelize the library sequential code.GParsPool.withPool { //The selectImportantNames() will process the name collections concurrently assert ['ALICE', 'JASON'] == selectImportantNames(['Joe', 'Alice', 'Dave', 'Jason'].makeTransparent()) }/** * A function implemented using standard sequential collect() and findAll() methods. */ def selectImportantNames(names) { names.collect {it.toUpperCase()}.findAll{it.size() > 4} }
/** * A function implemented using standard sequential collect() and findAll() methods. */ def selectImportantNames(names) { names.collect {it.toUpperCase()}.findAll{it.size() > 4} }def names = ['Joe', 'Alice', 'Dave', 'Jason'] ParallelEnhancer.enhanceInstance(names) //The selectImportantNames() will process the name collections concurrently assert ['ALICE', 'JASON'] == selectImportantNames(names.makeTransparent())
Dependency resolution
For the GParsPool class to work, the jsr166y-070108.jar must be on the classpath.<dependency> <groupId>org.coconut.forkjoin</groupId> <artifactId>jsr166y</artifactId> <version>070108</version> </dependency>
Avoid side-effects in functions
We have to warn you. Since the closures that are provided to the parallel methods like eachParallel() or collectParallel() may be run in parallel, you have to make sure that each of the closures is written in a thread-safe manner. The closures must hold no internal state, share data nor have side-effects beyond the boundaries the single element that they've been invoked on. Violations of these rules will open the door for race conditions and deadlocks, the most severe enemies of a modern multi-core programmer.Don't do this:def thumbnails = [] images.eachParallel {thumbnails << it.thumbnail} //Concurrently accessing a not-thread-safe collection of thumbnails, don't do this!
3.1.2 GParsExecutorsPool
Use of GParsExecutorsPool - the Java Executors' based concurrent collection processorUsage of GParsExecutorsPool
The GParsPool class enables a Java Executors-based concurrency DSL for collections and objects.The GParsExecutorsPool class can be used as a pure-JDK-based collection parallel processor. Unlike the GParsPool class, GParsExecutorsPool doesn't require jsr-166y jar file, but leverages the standard JDK executor services to parallelize closures processing a collections or an object iteratively. It needs to be states, however, that GParsPool performs typically much better than GParsExecutorsPool does.Examples of use://multiply numbers asynchronously GParsExecutorsPool.withPool { Collection<Future> result = [1, 2, 3, 4, 5].collectParallel{it * 10} assertEquals(new HashSet([10, 20, 30, 40, 50]), new HashSet((Collection)result*.get())) } //multiply numbers asynchronously using an asynchronous closure GParsExecutorsPool.withPool { def closure={it * 10} def asyncClosure=closure.async() Collection<Future> result = [1, 2, 3, 4, 5].collect(asyncClosure) assertEquals(new HashSet([10, 20, 30, 40, 50]), new HashSet((Collection)result*.get())) }
//find an element meeting specified criteria
GParsExecutorsPool.withPool(5) {ExecutorService service ->
service.submit({performLongCalculation()} as Runnable)
}
withPool(10) {...} withPool(20, threadFactory) {...}
withPool {
def result = [1, 2, 3, 4, 5].findParallel{Number number -> number > 2}
assert result in [3, 4, 5]
}
- eachParallel()
- eachWithIndexParallel()
- collectParallel()
- findAllParallel()
- findParallel()
- allParallel()
- anyParallel()
- grepParallel()
- groupByParallel()
Meta-class enhancer
As an alternative you can use the GParsExecutorsPoolEnhancer class to enhance meta-classes for any classes or individual instances with asynchronous methods.import groovyx.gpars.GParsExecutorsPoolEnhancerdef list = [1, 2, 3, 4, 5, 6, 7, 8, 9]
GParsExecutorsPoolEnhancer.enhanceInstance(list)
println list.collectParallel {it * 2 }def animals = ['dog', 'ant', 'cat', 'whale']
GParsExecutorsPoolEnhancer.enhanceInstance animals
println (animals.anyParallel {it ==~ /ant/} ? 'Found an ant' : 'No ants found')
println (animals.allParallel {it.contains('a')} ? 'All animals contain a' : 'Some animals can live without an a')
Exception handling
If exceptions are thrown while processing any of the passed-in closures, an instance of AsyncException wrapping all the original exceptions gets re-thrown from the xxxParallel methods.Avoid side-effects in functions
Once again we need to warn you about using closures with side-effects effecting objects beyond the scope of the single currently processed element or closures which keep state. Don't do that! It is dangerous to pass them to any of the xxxParallel() methods.3.1.3 Memoize
The memoize function enables caching of function's return values. Repeated calls to the memoized function with the same argument values will, instead of invoking the calculation encoded in the original function, retrieve the result value from an internal transparent cache. Provided the calculation is considerably slower than retrieving a cached value from the cache, this allows users to trade-off memory for performance. Checkout out the example, where we attempt to scan multiple websites for particular content:The memoize functionality of GPars has been contributed to Groovy in version 1.8 and if you run on Groovy 1.8 or later, it is recommended to use the Groovy functionality. Memoize in GPars is almost identical, except that it searches the memoize caches concurrently using the surrounding thread pool and so may give performance benefits in some scenarios.The GPars memoize functionality has been renamed to avoid future conflicts with the memoize functionality in Groovy. GPars now calls the methods with a preceding letter g , such as gmemoize().
Examples of use
GParsPool.withPool {
def urls = ['http://www.dzone.com', 'http://www.theserverside.com', 'http://www.infoq.com']
Closure download = {url ->
println "Downloading $url"
url.toURL().text.toUpperCase()
}
Closure cachingDownload = download.gmemoize() println 'Groovy sites today: ' + urls.findAllParallel {url -> cachingDownload(url).contains('GROOVY')}
println 'Grails sites today: ' + urls.findAllParallel {url -> cachingDownload(url).contains('GRAILS')}
println 'Griffon sites today: ' + urls.findAllParallel {url -> cachingDownload(url).contains('GRIFFON')}
println 'Gradle sites today: ' + urls.findAllParallel {url -> cachingDownload(url).contains('GRADLE')}
println 'Concurrency sites today: ' + urls.findAllParallel {url -> cachingDownload(url).contains('CONCURRENCY')}
println 'GPars sites today: ' + urls.findAllParallel {url -> cachingDownload(url).contains('GPARS')}
}
Fibonacci example
A purely functional, recursive implementation, following closely the definition of Fibonacci numbers is exponentially complex:Closure fib = {n -> n > 1 ? call(n - 1) + call(n - 2) : n}
Closure fib fib = {n -> n > 1 ? fib(n - 1) + fib(n - 2) : n}.gmemoize()
Available variants
memoize
The basic variant, which keeps values in the internal cache for the whole lifetime of the memoized function. Provides the best performance characteristics of all the variants.memoizeAtMost
Allows the user to set a hard limit on number of items cached. Once the limit has been reached, all subsequently added values will eliminate the oldest value from the cache using the LRU (Last Recently Used) strategy.So for our Fibonacci number example, we could safely reduce the cache size to two items:Closure fib fib = {n -> n > 1 ? fib(n - 1) + fib(n - 2) : n}.memoizeAtMost(2)
- Keep the memory footprint of the cache within defined boundaries
- Preserve desired performance characteristics of the function. Too large caches may take longer to retrieve the cached value than it would have taken to calculate the result directly.
memoizeAtLeast
Allows unlimited growth of the internal cache until the JVM's garbage collector decides to step in and evict SoftReferences, used by our implementation, from the memory. The single parameter value to the memoizeAtLeast() method specifies the minimum number of cached items that should be protected from gc eviction. The cache will never shrink below the specified number of entries. The cache ensures it only protects the most recently used items from eviction using the LRU (Last Recently Used) strategy.memoizeBetween
Combines memoizeAtLeast and memoizeAtMost and so allowing the cache to grow and shrink in the range between the two parameter values depending on available memory and the gc activity, yet the cache size will never exceed the upper size limit to preserve desired performance characteristics of the cache.3.2 Map-Reduce
The Parallel Collection Map/Reduce DSL gives GPars a more functional flavor. In general, the Map/Reduce DSL may be used for the same purpose as the xxxParallel() family methods and has very similar semantics. On the other hand, Map/Reduce can perform considerably faster, if you need to chain multiple methods to process a single collection in multiple steps:println 'Number of occurrences of the word GROOVY today: ' + urls.parallel
.map {it.toURL().text.toUpperCase()}
.filter {it.contains('GROOVY')}
.map{it.split()}
.map{it.findAll{word -> word.contains 'GROOVY'}.size()}
.sum()
- map()
- reduce()
- filter()
- size()
- sum()
- min()
- max()
- sort()
- groupBy()
- combine()
def myNumbers = (1..1000).parallel.filter{it % 2 == 0}.map{Math.sqrt it}.collection
Avoid side-effects in functions
Once again we need to warn you. To avoid nasty surprises, please, keep your closures, which you pass to the Map/Reduce functions, stateless and clean from side-effects.Availability
This feature is only available when using in the Fork/Join-based GParsPool , not in GParsExecutorsPool .Classical Example
A classical example, inspired by http://github.com/thevery, counting occurencies of words in a string:import static groovyx.gpars.GParsPool.withPooldef words = "This is just a plain text to count words in" print count(words)def count(arg) { withPool { return arg.parallel .map{[it, 1]} .groupBy{it[0]}.getParallel() .map {it.value=it.value.size();it} .sort{-it.value}.collection } }
def words = "This is just a plain text to count words in" print count(words)def count(arg) { withPool { return arg.parallel .map{[it, 1]} .combine(0) {sum, value -> sum + value}.getParallel() .sort{-it.value}.collection } }
Combine
The combine operation expects on its input a list of tuples (two-element lists) considered to be key-value pairs (such as [key1, value1, key2, value2, key1, value3, key3, value4 … ] ) with potentially repeating keys. When invoked, combine merges the values for identical keys using the provided accumulator function and produces a map mapping the original (unique) keys to their accumulated values. E.g. [a, b, c, d, a, e, c, f] will be combined into a : b+e, c : d+f, while the '+' operation on the values needs to be provided by the user as the accumulation closure. The accumulation function argument needs to specify a function to use for combining (accumulating) the values belonging to the same key. An initial accumulator value needs to be provided as well. Since the combine method processes items in parallel, the initial accumulator value will be reused multiple times. Thus the provided value must allow for reuse. It should be either a cloneable or immutable value or a closure returning a fresh initial accumulator each time requested. Good combinations of accumulator functions and reusable initial values include:accumulator = {List acc, value -> acc << value} initialValue = [] accumulator = {List acc, value -> acc << value} initialValue = {-> []} accumulator = {int sum, int value -> acc + value} initialValue = 0 accumulator = {int sum, int value -> sum + value} initialValue = {-> 0} accumulator = {ShoppingCart cart, Item value -> cart.addItem(value)} initialValue = {-> new ShoppingCart()}
3.3 Parallel Arrays
As an alternative, the efficient tree-based data structures defines in JSR-166y can be used directly. The parallelArray property on any collection or object will return a jsr166y.forkjoin.ParallelArray instance holding the elements of the original collection, which then can be manipulated through the jsr166y API. Please refer to the jsr166y documentation for the API details.groovyx.gpars.GParsPool.withPool { assert 15 == [1, 2, 3, 4, 5].parallelArray.reduce({a, b -> a + b} as Reducer, 0) //summarize assert 55 == [1, 2, 3, 4, 5].parallelArray.withMapping({it ** 2} as Mapper).reduce({a, b -> a + b} as Reducer, 0) //summarize squares assert 20 == [1, 2, 3, 4, 5].parallelArray.withFilter({it % 2 == 0} as Predicate) //summarize squares of even numbers .withMapping({it ** 2} as Mapper) .reduce({a, b -> a + b} as Reducer, 0) assert 'aa:bb:cc:dd:ee' == 'abcde'.parallelArray //concatenate duplicated characters with separator .withMapping({it * 2} as Mapper) .reduce({a, b -> "$a:$b"} as Reducer, "")
3.4 Asynchronous Invocation
Running long-lasting tasks in the background belongs to the activities, the need for which arises quite frequently. Your main thread of execution wants to initialize a few calculations, downloads, searches or such, however, the results may not be needed immediately. GPars gives the developers the tools to schedule the asynchronous activities for processing in the background and collect the results once they're needed.Usage of GParsPool and GParsExecutorsPool asynchronous processing facilities
Both GParsPool and GParsExecutorsPool provide almost identical services in this domain, although they leverage different underlying machinery, based on which of the two classes the user chooses.Closures enhancements
The following methods are added to closures inside the GPars(Executors)Pool.withPool() blocks:- async() - Creates an asynchronous variant of the supplied closure, which when invoked returns a future for the potential return value
- callAsync() - Calls a closure in a separate thread supplying the given arguments, returning a future for the potential return value,
GParsPool.withPool() { Closure longLastingCalculation = {calculate()} Closure fastCalculation = longLastingCalculation.async() //create a new closure, which starts the original closure on a thread pool Future result=fastCalculation() //returns almost immediately //do stuff while calculation performs … println result.get() }
GParsPool.withPool() { /** * The callAsync() method is an asynchronous variant of the default call() method to invoke a closure. * It will return a Future for the result value. */ assert 6 == {it * 2}.call(3) assert 6 == {it * 2}.callAsync(3).get() }
Timeouts
The callTimeoutAsync() methods, taking either a long value or a Duration instance, allow the user to have the calculation cancelled after a given time interval.{-> while(true) { Thread.sleep 1000 //Simulate a bit of interesting calculation if (Thread.currentThread().isInterrupted()) break; //We've been cancelled } }.callTimeoutAsync(2000)
Executor Service enhancements
The ExecutorService and jsr166y.forkjoin.ForkJoinPool class is enhanced with the << (leftShift) operator to submit tasks to the pool and return a Future for the result.Example:GParsExecutorsPool.withPool {ExecutorService executorService -> executorService << {println 'Inside parallel task'} }
Running functions (closures) in parallel
The GParsPool and GParsExecutorsPool classes also provide handy methods executeAsync() and executeAsyncAndWait() to easily run multiple closures asynchronously.Example:GParsPool.withPool { assertEquals([10, 20], GParsPool.executeAsyncAndWait({calculateA()}, {calculateB()})) //waits for results assertEquals([10, 20], GParsPool.executeAsync({calculateA()}, {calculateB()})*.get()) //returns Futures instead and doesn't wait for results to be calculated }
3.5 Parallel Speculations
With processor cores having become plentiful, some algorithms might benefit from brutal-force parallel duplication. Instead of deciding up-front about how to solve a problem, what algorithm to use or which location to connect to, you run all potential solutions in parallel.Parallel speculations
Imagine you need to perform a task like e.g. calculate an expensive function or read data from a file, database or internet. Luckily, you know of several good ways (e.g. functions or urls) to achieve your goal. However, they are not all equal. Although they return back the same (as far as your needs are concerned) result, they may all take different amount of time to complete and some of them may even fail (e.g. network issues). What's worse, no-one is going to tell you which path gives you the solution first nor which paths lead to no solution at all. Shall I run quick sort or merge sort on my list? Which url will work best? Is this service available at its primary location or should I use the backup one?GPars speculations give you the option to try all the available alternatives in parallel and so get the result from the fastest functional path, silently ignoring the slow or broken ones.This is what the speculate() methods on GParsPool and GParsExecutorsPool() can do.def numbers = … def quickSort = … def mergeSort = … def sortedNumbers = speculate(quickSort, mergeSort)
import static groovyx.gpars.GParsPool.speculate import static groovyx.gpars.GParsPool.withPooldef alternative1 = { 'http://www.dzone.com/links/index.html'.toURL().text }def alternative2 = { 'http://www.dzone.com/'.toURL().text }def alternative3 = { 'http://www.dzzzzzone.com/'.toURL().text //wrong url }def alternative4 = { 'http://dzone.com/'.toURL().text }withPool(4) { println speculate([alternative1, alternative2, alternative3, alternative4]).contains('groovy') }
Make sure the surrounding thread pool has enough threads to process all alternatives in parallel. The size of the pool should match the number of closures supplied.
Alternatives using dataflow variables and streams
In cases, when stopping unsuccessful alternatives is not needed, dataflow variables or streams may be used to obtain the result value from the winning speculation.Please refer to the Dataflow Concurrency section of the User Guide for details on Dataflow variables and streams.
import groovyx.gpars.dataflow.DataFlowQueue import static groovyx.gpars.dataflow.DataFlow.taskdef alternative1 = { 'http://www.dzone.com/links/index.html'.toURL().text }def alternative2 = { 'http://www.dzone.com/'.toURL().text }def alternative3 = { 'http://www.dzzzzzone.com/'.toURL().text //will fail due to wrong url }def alternative4 = { 'http://dzone.com/'.toURL().text }//Pick either one of the following, both will work: final def result = new DataFlowQueue() // final def result = new DataFlowVariable()[alternative1, alternative2, alternative3, alternative4].each {code -> task { try { result << code() } catch (ignore) { } //We deliberately ignore unsuccessful urls } }println result.val.contains('groovy')
3.6. Fork-Join
Fork/Join or Divide and Conquer is a very powerful abstraction to solve hierarchical problems.The abstraction
When talking about hierarchical problems, think about quick sort, merge sort, file system or general tree navigation and such.- Fork / Join algorithms essentially split a problem at hands into several smaller sub-problems and recursively apply the same algorithm to each of the sub-problems.
- Once the sub-problem is small enough, it is solved directly.
- The solutions of all sub-problems are combined to solve their parent problem, which in turn helps solve its own parent problem.
The GPars abstraction convenience layer
GPars can hide the complexities of dealing with threads, pools and recursive tasks from you, yet let you leverage the powerful Fork/Join implementation in jsr166y.import static groovyx.gpars.GParsPool.runForkJoin import static groovyx.gpars.GParsPool.withPoolwithPool() { println """Number of files: ${ runForkJoin(new File("./src")) {file -> long count = 0 file.eachFile { if (it.isDirectory()) { println "Forking a child task for $it" forkOffChild(it) //fork a child task } else { count++ } } return count + (childrenResults.sum(0)) //use results of children tasks to calculate and store own result } }""" }
def quicksort(numbers) { withPool { runForkJoin(0, numbers) {index, list -> def groups = list.groupBy {it <=> list[list.size().intdiv(2)]} if ((list.size() < 2) || (groups.size() == 1)) { return [index: index, list: list.clone()] } (-1..1).each {forkOffChild(it, groups[it] ?: [])} return [index: index, list: childrenResults.sort {it.index}.sum {it.list}] }.list } }
Alternative approach
Alternatively, the underlying mechanism of nested Fork/Join worker tasks can be used directly. Custom-tailored workers can eliminate the performance overhead associated with parameter spreading imposed when using the generic workers. Also, custom workers can be implemented in Java and so further increase the performance of the algorithm.public final class FileCounter extends AbstractForkJoinWorker<Long> { private final File file; def FileCounter(final File file) { this.file = file } @Override protected Long computeTask() { long count = 0; file.eachFile { if (it.isDirectory()) { println "Forking a thread for $it" forkOffChild(new FileCounter(it)) //fork a child task } else { count++ } } return count + ((childrenResults)?.sum() ?: 0) //use results of children tasks to calculate and store own result } }withPool(1) {pool -> //feel free to experiment with the number of fork/join threads in the pool println "Number of files: ${runForkJoin(new FileCounter(new File("..")))}" }
Fork / Join saves your resources
Fork/Join operations can be safely run with small number of threads thanks to internally using the TaskBarrier class to synchronize the threads. While a thread is blocked inside an algorithm waiting for its sub-problems to be calculated, the thread is silently returned to the pool to take on any of the available sub-problems from the task queue and process them. Although the algorithm creates as many tasks as there are sub-directories and tasks wait for the sub-directory tasks to complete, as few as one thread is enough to keep the computation going and eventually calculate a valid result.Mergesort example
import static groovyx.gpars.GParsPool.runForkJoin import static groovyx.gpars.GParsPool.withPool/** * Splits a list of numbers in half */ def split(List<Integer> list) { int listSize = list.size() int middleIndex = listSize / 2 def list1 = list[0..<middleIndex] def list2 = list[middleIndex..listSize - 1] return [list1, list2] }/** * Merges two sorted lists into one */ List<Integer> merge(List<Integer> a, List<Integer> b) { int i = 0, j = 0 final int newSize = a.size() + b.size() List<Integer> result = new ArrayList<Integer>(newSize) while ((i < a.size()) && (j < b.size())) { if (a[i] <= b[j]) result << a[i++] else result << b[j++] } if (i < a.size()) result.addAll(a[i..-1]) else result.addAll(b[j..-1]) return result }final def numbers = [1, 5, 2, 4, 3, 8, 6, 7, 3, 4, 5, 2, 2, 9, 8, 7, 6, 7, 8, 1, 4, 1, 7, 5, 8, 2, 3, 9, 5, 7, 4, 3]withPool(3) { //feel free to experiment with the number of fork/join threads in the pool println """Sorted numbers: ${ runForkJoin(numbers) {nums -> println "Thread ${Thread.currentThread().name[-1]}: Sorting $nums" switch (nums.size()) { case 0..1: return nums //store own result case 2: if (nums[0] <= nums[1]) return nums //store own result else return nums[-1..0] //store own result default: def splitList = split(nums) [splitList[0], splitList[1]].each {forkOffChild it} //fork a child task return merge(* childrenResults) //use results of children tasks to calculate and store own result } } }""" }
Mergesort example using a custom-tailored worker class
public final class SortWorker extends AbstractForkJoinWorker<List<Integer>> { private final List numbers def SortWorker(final List<Integer> numbers) { this.numbers = numbers.asImmutable() } /** * Splits a list of numbers in half */ def split(List<Integer> list) { int listSize = list.size() int middleIndex = listSize / 2 def list1 = list[0..<middleIndex] def list2 = list[middleIndex..listSize - 1] return [list1, list2] } /** * Merges two sorted lists into one */ List<Integer> merge(List<Integer> a, List<Integer> b) { int i = 0, j = 0 final int newSize = a.size() + b.size() List<Integer> result = new ArrayList<Integer>(newSize) while ((i < a.size()) && (j < b.size())) { if (a[i] <= b[j]) result << a[i++] else result << b[j++] } if (i < a.size()) result.addAll(a[i..-1]) else result.addAll(b[j..-1]) return result } /** * Sorts a small list or delegates to two children, if the list contains more than two elements. */ @Override protected List<Integer> computeTask() { println "Thread ${Thread.currentThread().name[-1]}: Sorting $numbers" switch (numbers.size()) { case 0..1: return numbers //store own result case 2: if (numbers[0] <= numbers[1]) return numbers //store own result else return numbers[-1..0] //store own result default: def splitList = split(numbers) [new SortWorker(splitList[0]), new SortWorker(splitList[1])].each{forkOffChild it} //fork a child task return merge(* childrenResults) //use results of children tasks to calculate and store own result } } }final def numbers = [1, 5, 2, 4, 3, 8, 6, 7, 3, 4, 5, 2, 2, 9, 8, 7, 6, 7, 8, 1, 4, 1, 7, 5, 8, 2, 3, 9, 5, 7, 4, 3]withPool(1) { //feel free to experiment with the number of fork/join threads in the pool println "Sorted numbers: ${runForkJoin(new SortWorker(numbers))}" }
Availability
This feature is only available when using in the Fork/Join-based GParsPool , not in GParsExecutorsPool .4. Groovy CSP
The CSP (Communicating Sequential Processes) abstraction builds on independent composable processes, which exchange messages in a synchronous manner. GPars leverages the JCSP library developed at the University of Kent, UK.Jon Kerridge, the author of the CSP implementation in GPars, provides exhaustive examples on of GroovyCSP use at his website:The GroovyCSP implementation leverages JCSP, a Java-based CSP library, which is licensed under GPL. Unlike the liberal Apache 2 license, which GPars uses, GPL is more restrictive on use in commercial software. Please make sure your application conforms to the GPL rules before enabling use of JCSP in your code.If the GPL license is too restrictive for your use, you might consider checking out the DataFlow Concurrency chapter of this User Guide to learn about tasks , selectors and operators , which may help you resolve concurrency issues in ways similar to the CSP approach. In fact the dataflow and CSP concepts stand very close to each other.
By default, without actively adding an explicit dependency on JCSP in your build file or downloading and including the JCSP jar file in your project, the standard commercial-software-friendly Apache 2 License terms apply to your project. GPars directly only depends on software licensed under licenses compatible with the Apache 2 License.
5. Actors
The actor support in gpars were inspired by the Actors library in Scala but have meanwhile gone beyond that.Actors allow for a messaging-based concurrency model, built from independent active objects that exchange messages and have no mutable shared state. Actors can help developers avoid issues like deadlocks, live-locks or starvation, so typical for shared memory, while leveraging the multi-core nature of today's hardware. A nice wrap-up of the key concepts behind actors was written recently by Ruben Vermeersch. Actors guarantee that always at most one thread processes the actor's body at a time and also under the covers the memory gets synchronized each time a thread gets assigned to an actor so the actor's state can be safely modified by code in the body without any other extra (synchronization or locking) effort . Ideally actor's code should never be invoked directly from outside so all the code of the actor class can only be executed by the thread handling the last received message and so all the actor's code is implicitly thread-safe . If any of the actor's methods is allowed to be called by other objects directly, the thread-safety guarantee for the actor's code and state are no longer valid .Types of actors
In general, you can find two types of actors in the wild - ones that hold implicit state and those, who don't. GPars gives you both options. Stateless actors, represented in GPars by the DynamicDispatchActor and the ReactiveActor classes, keep no track of what messages have arrived previously. You may thing of these as flat message handlers, which process messages as they come. Any state-based behavior has to be implemented by the user.The stateful actors, represented in GPars by the DefaultActor class (and previously also by the AbstractPooledActor class), allow the user to handle implicit state directly. After receiving a message the actor moves into a new state with different ways to handle future messages. To give you an example, a freshly started actor may only accept some types of messages, e.g. encrypted messages for decryption, only after it has received the encryption keys. The stateful actors allow to encode such dependencies directly in the structure of the message-handling code. Implicit state management, however, comes at a slight performance cost, mainly due to the lack of continuations support on JVM.Actor threading model
Since actors are detached from the system threads, a great number of actors can share a relatively small thread pool. This can go as far as having many concurrent actors that share a single pooled thread. This architecture allows to avoid some of the threading limitations of the JVM. In general, while the JVM can only give you a limited number of threads (typically around a couple of thousands), the number of actors is only limited by the available memory. If an actor has no work to do, it doesn't consume threads.Actor code is processed in chunks separated by quiet periods of waiting for new events (messages). This can be naturally modeled through continuations . As JVM doesn't support continuations directly, they have to be simulated in the actors frameworks, which has slight impact on organization of the actors' code. However, the benefits in most cases outweigh the difficulties.import groovyx.gpars.actor.Actor import groovyx.gpars.actor.DefaultActorclass GameMaster extends DefaultActor { int secretNum void afterStart() { secretNum = new Random().nextInt(10) } void act() { loop { react { int num -> if (num > secretNum) reply 'too large' else if (num < secretNum) reply 'too small' else { reply 'you win' terminate() } } } } }class Player extends DefaultActor { String name Actor server int myNum void act() { loop { myNum = new Random().nextInt(10) server.send myNum react { switch (it) { case 'too large': println "$name: $myNum was too large"; break case 'too small': println "$name: $myNum was too small"; break case 'you win': println "$name: I won $myNum"; terminate(); break } } } } }def master = new GameMaster().start() def player = new Player(name: 'Player', server: master).start()//this forces main thread to live until both actors stop [master, player]*.join()
Usage of Actors
Gpars provides consistent Actor APIs and DSLs. Actors in principal perform three specific operations - send messages, receive messages and create new actors. Although not specifically enforced by GPars messages should be immutable or at least follow the hands-off policy when the sender never touches the messages after the message has been sent off.Sending messages
Messages can be sent to actors using the send() method.def passiveActor = Actors.actor{ loop { react { msg -> println "Received: $msg"; } } } passiveActor.send 'Message 1' passiveActor << 'Message 2' //using the << operator passiveActor 'Message 3' //using the implicit call() method
def replyingActor = Actors.actor{ loop { react { msg -> println "Received: $msg"; reply "I've got $msg" } } } def reply1 = replyingActor.sendAndWait('Message 4') def reply2 = replyingActor.sendAndWait('Message 5', 10, TimeUnit.SECONDS) use (TimeCategory) { def reply3 = replyingActor.sendAndWait('Message 6', 10.seconds) }
friend.sendAndContinue 'I need money!', {money -> pocket money} println 'I can continue while my friend is collecting money for me'
Receiving messages
Non-blocking message retrieval
Calling the react() method, optionally with a timeout parameter, from within the actor's code will consume the next message from the actor's inbox, potentially waiting, if there is no message to be processed immediately.println 'Waiting for a gift' react {gift -> if (myWife.likes gift) reply 'Thank you!' }
Actors.actor { loop { println 'Waiting for a gift' react {gift -> if (myWife.likes gift) reply 'Thank you!' else { reply 'Try again, please' react {anotherGift -> if (myChildren.like gift) reply 'Thank you!' } println 'Never reached' } } println 'Never reached' } println 'Never reached' }
Sending replies
The reply/replyIfExists methods are not only defined on the actors themselves, but for AbstractPooledActor (not available in DefaultActor , DynamicDispatchActor nor ReactiveActor classes) also on the processed messages themselves upon their reception, which is particularly handy when handling multiple messages in a single call. In such cases reply() invoked on the actor sends a reply to authors of all the currently processed message (the last one), whereas reply() called on messages sends a reply to the author of the particular message only.See demo hereThe sender property
Messages upon retrieval offer the sender property to identify the originator of the message. The property is available inside the Actor's closure:react {tweet ->
if (isSpam(tweet)) ignoreTweetsFrom sender
sender.send 'Never write me again!'
}
Forwarding
When sending a message, a different actor can be specified as the sender so that potential replies to the message will be forwarded to the specified actor and not to the actual originator.def decryptor = Actors.actor { react {message -> reply message.reverse() // sender.send message.reverse() //An alternative way to send replies } }def console = Actors.actor { //This actor will print out decrypted messages, since the replies are forwarded to it react { println 'Decrypted message: ' + it } }decryptor.send 'lellarap si yvoorG', console //Specify an actor to send replies to console.join()
Creating Actors
Actors share a pool of threads, which are dynamically assigned to actors when the actors need to react to messages sent to them. The threads are returned to back the pool once a message has been processed and the actor is idle waiting for some more messages to arrive.For example, this is how you create an actor that prints out all messages that it receives.def console = Actors.actor { loop { react { println it } } }
final def decryptor = Actors.actor { loop { react {String message -> if ('stopService' == message) { println 'Stopping decryptor' stop() } else reply message.reverse() } } }Actors.actor { decryptor.send 'lellarap si yvoorG' react { println 'Decrypted message: ' + it decryptor.send 'stopService' } }.join()
def friend = Actors.actor { react { //this doesn't reply -> caller won't receive any answer in time println it //reply 'Hello' //uncomment this to answer conversation react { println it } } }def me = Actors.actor { friend.send('Hi') //wait for answer 1sec react(1000) {msg -> if (msg == Actor.TIMEOUT) { friend.send('I see, busy as usual. Never mind.') stop() } else { //continue conversation println "Thank you for $msg" } } }me.join()
Undelivered messages
Sometimes messages cannot be delivered to the target actor. When special action needs to be taken for undelivered messages, at actor termination all unprocessed messages from its queue have their onDeliveryError() method called. The onDeliveryError() method or closure defined on the message can, for example, send a notification back to the original sender of the message.final DefaultActor me me = Actors.actor { def message = 1 message.metaClass.onDeliveryError = {-> //send message back to the caller me << "Could not deliver $delegate" } def actor = Actors.actor { react { //wait 2sec in order next call in demo can be emitted Thread.sleep(2000) //stop actor after first message stop() } } actor << message actor << message react { //print whatever comes back println it }}me.join()
final DefaultActor me me = Actors.actor { def message1 = 1 def message2 = 2 def actor = Actors.actor { react { //wait 2sec in order next call in demo can be emitted Thread.sleep(2000) //stop actor after first message stop() } } me.metaClass.onDeliveryError = {msg -> //callback on actor inaccessibility println "Could not deliver message $msg" } actor << message1 actor << message2 actor.join()}me.join()
class MyActor extends DefaultActor { public void onDeliveryError(msg) { println "Could not deliver message $msg" } … }
Joining actors
Actors provide a join() method to allow callers to wait for the actor to terminate. A variant accepting a timeout is also available. The Groovy spread-dot operator comes in handy when joining multiple actors at a time.def master = new GameMaster().start() def player = new Player(name: 'Player', server: master).start()[master, player]*.join()
Conditional and counting loops
The loop() method allows for either a condition or a number of iterations to be specified, optionally accompanied with a closure to invoke once the loop finishes - After Loop Termination Code Handler .The following actor will loop three times to receive 3 messages and then prints out the maximum of the received messages.final Actor actor = Actors.actor { def candidates = [] def printResult = {-> println "The best offer is ${candidates.max()}"} loop(3, printResult) { react { candidates << it } } }actor 10 actor 30 actor 20 actor.join()
final Actor actor = Actors.actor { def candidates = [] final Closure printResult = {-> println "Reached best offer - ${candidates.max()}"} loop({-> candidates.max() < 30}, printResult) { react { candidates << it } } }actor 10 actor 20 actor 25 actor 31 actor 20 actor.join()
The After Loop Termination Code Handler can use actor's react{} but not loop() .
DefaultActor can be set to behave in a fair on non-fair (default) manner. Depending on the strategy chosen, the actor either makes the thread available to other actors sharing the same parallel group (fair), or keeps the thread fot itself until the message queue gets empty (non-fair). Generally, non-fair actors perform 2 - 3 times better than fair ones.Use either the fairActor() factory method or the actor's makeFair() method.
Custom schedulers
Actors leverage the standard JDK concurrency library by default. To provide a custom thread scheduler use the appropriate constructor parameter when creating a parallel group (PGroup class). The supplied scheduler will orchestrate threads in the group's thread pool.Please also see the numerous Actor Demos .5.1 Actors Principles
Actors share a pool of threads, which are dynamically assigned to actors when the actors need to react to messages sent to them. The threads are returned back to the pool once a message has been processed and the actor is idle waiting for some more messages to arrive. Actors become detached from the underlying threads and so a relatively small thread pool can serve potentially unlimited number of actors. Virtually unlimited scalability in number of actors is the main advantage of event-based actors , which are detached from the underlying physical threads.Here are some examples of how to use actors. This is how you create an actor that prints out all messages that it receives.import static groovyx.gpars.actor.Actors.*def console = actor { loop { react { println it } }
class CustomActor extends DefaultActor { @Override protected void act() { loop { react { println it } } } }def console=new CustomActor() console.start()
console.send('Message') console 'Message' console.sendAndWait 'Message' //Wait for a reply console.sendAndContinue 'Message', {reply -> println "I received reply: $reply"} //Forward the reply to a function
Creating an asynchronous service
import static groovyx.gpars.actor.Actors.*final def decryptor = actor { loop { react {String message-> reply message.reverse() } } }def console = actor { decryptor.send 'lellarap si yvoorG' react { println 'Decrypted message: ' + it } }console.join()
def friend = Actors.actor { react { //this doesn't reply -> caller won't receive any answer in time println it //reply 'Hello' //uncomment this to answer conversation react { println it } } }def me = Actors.actor { friend.send('Hi') //wait for answer 1sec react(1000) {msg -> if (msg == Actor.TIMEOUT) { friend.send('I see, busy as usual. Never mind.') stop() } else { //continue conversation println "Thank you for $msg" } } }me.join()
def friend = Actors.actor { react { //this doesn't reply -> caller won't receive any answer in time println it //reply 'Hello' //uncomment this to answer conversation react { println it } } }def me = Actors.actor { friend.send('Hi') delegate.metaClass.onTimeout = {-> friend.send('I see, busy as usual. Never mind.') stop() } //wait for answer 1sec react(1000) {msg -> if (msg != Actor.TIMEOUT) { //continue conversation println "Thank you for $msg" } } }me.join()
class MyActor extends DefaultActor { public void onTimeout() { … } protected void act() { … } }
Actors guarantee thread-safety for non-thread-safe code
Actors guarantee that always at most one thread processes the actor's body at a time and also under the covers the memory gets synchronized each time a thread gets assigned to an actor so the actor's state can be safely modified by code in the body without any other extra (synchronization or locking) effort .class MyCounterActor extends DefaultActor { private Integer counter = 0 protected void act() { loop { react { counter++ } } } }
Simple calculator
A little bit more realistic example of an event-driven actor that receives two numeric messages, sums them up and sends the result to the console actor.import groovyx.gpars.group.DefaultPGroup//not necessary, just showing that a single-threaded pool can still handle multiple actors def group = new DefaultPGroup(1);final def console = group.actor { loop { react { println 'Result: ' + it } } }final def calculator = group.actor { react {a -> react {b -> console.send(a + b) } } }calculator.send 2 calculator.send 3calculator.join() group.shutdown()
Concurrent Merge Sort Example
For comparison I'm also including a more involved example performing a concurrent merge sort of a list of integers using actors. You can see that thanks to flexibility of Groovy we came pretty close to the Scala model, although I still miss Scala pattern matching for message handling.import groovyx.gpars.group.DefaultPGroup import static groovyx.gpars.actor.Actors.actorClosure createMessageHandler(def parentActor) { return { react {List<Integer> message -> assert message != null switch (message.size()) { case 0..1: parentActor.send(message) break case 2: if (message[0] <= message[1]) parentActor.send(message) else parentActor.send(message[-1..0]) break default: def splitList = split(message) def child1 = actor(createMessageHandler(delegate)) def child2 = actor(createMessageHandler(delegate)) child1.send(splitList[0]) child2.send(splitList[1]) react {message1 -> react {message2 -> parentActor.send merge(message1, message2) } } } } } }def console = new DefaultPGroup(1).actor { react { println "Sorted array:t${it}" System.exit 0 } }def sorter = actor(createMessageHandler(console)) sorter.send([1, 5, 2, 4, 3, 8, 6, 7, 3, 9, 5, 3]) console.join()def split(List<Integer> list) { int listSize = list.size() int middleIndex = listSize / 2 def list1 = list[0..<middleIndex] def list2 = list[middleIndex..listSize - 1] return [list1, list2] }List<Integer> merge(List<Integer> a, List<Integer> b) { int i = 0, j = 0 final int newSize = a.size() + b.size() List<Integer> result = new ArrayList<Integer>(newSize) while ((i < a.size()) && (j < b.size())) { if (a[i] <= b[j]) result << a[i++] else result << b[j++] } if (i < a.size()) result.addAll(a[i..-1]) else result.addAll(b[j..-1]) return result }
Actor lifecycle methods
Each Actor can define lifecycle observing methods, which will be called whenever a certain lifecycle event occurs.- afterStart() - called right after the actor has been started.
- afterStop(List undeliveredMessages) - called right after the actor is stopped, passing in all the unprocessed messages from the queue.
- onInterrupt(InterruptedException e) - called when the actor's thread gets interrupted. Thread interruption will result in the stopping the actor in any case.
- onTimeout() - called when no messages are sent to the actor within the timeout specified for the currently blocking react method.
- onException(Throwable e) - called when an exception occurs in the actor's event handler. Actor will stop after return from this method.
class MyActor extends DefaultActor { public void afterStart() { … } public void onTimeout() { … } protected void act() { … } }
def myActor = actor { delegate.metaClass.onException = { log.error('Exception occurred', it) }… }
To help performance, you may consider using the silentStart() method instead of start() when starting a DynamicDispatchActor or a ReactiveActor . Calling silentStart() will by-pass some of the start-up machinery and as a result will also avoid calling the afterStart() method. Due to its stateful nature, DefaultActor cannot be started silently.
Pool management
_Actors_ can be organized into groups and as a default there's always an application-wide pooled actor group available. And just like the Actors abstract factory can be used to create actors in the default group, custom groups can be used as abstract factories to create new actors instances belonging to these groups.def myGroup = new DefaultPGroup()def actor1 = myGroup.actor {
…
}def actor2 = myGroup.actor {
…
}
def myGroup = new DefaultPGroup(10) //the pool will contain 10 threads
… (n+1 threads in the default pool after startup)Actors.defaultActorPGroup.resize 1 //use one-thread pool… (1 thread in the pool)Actors.defaultActorPGroup.resetDefaultSize()… (n+1 threads in the pool)Actors.defaultActorPGroup.shutdown()
def daemonGroup = new DefaultPGroup()def actor1 = daemonGroup.actor { … }def nonDaemonGroup = new NonDaemonPGroup()def actor2 = nonDaemonGroup.actor { … }class MyActor { def MyActor() { this.parallelGroup = nonDaemonGroup } void act() {...} }
def coreActors = new NonDaemonPGroup(5) //5 non-daemon threads pool def helperActors = new DefaultPGroup(1) //1 daemon thread pooldef priceCalculator = coreActors.actor { … }def paymentProcessor = coreActors.actor { … }def emailNotifier = helperActors.actor { … }def cleanupActor = helperActors.actor { … }//increase size of the core actor group coreActors.resize 6//shutdown the group's pool once you no longer need the group to release resources helperActors.shutdown()
Common trap: App terminates while actors do not receive messages
Most likely you're using daemon threads and pools, which is the default setting, and your main thread finishes. Calling actor.join() on any, some or all of your actors would block the main thread until the actor terminates and thus keep all your actors running. Alternatively use instances of NonDaemonPGroup and assign some of your actors to these groups.def nonDaemonGroup = new NonDaemonPGroup()
def myActor = nonDaemonGroup.actor {...}
def nonDaemonGroup = new NonDaemonPGroup()class MyActor extends DefaultActor { def MyActor() { this.parallelGroup = nonDaemonGroup } void act() {...} }def myActor = new MyActor()
5.2 Stateless Actors
Dynamic Dispatch Actor
The DynamicDispatchActor class is an actor allowing for an alternative structure of the message handling code. In general DynamicDispatchActor repeatedly scans for messages and dispatches arrived messages to one of the onMessage(message) methods defined on the actor. The DynamicDispatchActor leverages the Groovy dynamic method dispatch mechanism under the covers. Since, unlike DefaultActor descendants, a DynamicDispatchActor not ReactiveActor (discussed below) do not need to implicitly remember actor's state between subsequent message receptions, they provide much better performance characteristics, generally comparable to other actor frameworks, like e.g. Scala Actors.import groovyx.gpars.actor.Actors import groovyx.gpars.actor.DynamicDispatchActorfinal class MyActor extends DynamicDispatchActor { void onMessage(String message) { println 'Received string' } void onMessage(Integer message) { println 'Received integer' reply 'Thanks!' } void onMessage(Object message) { println 'Received object' sender.send 'Thanks!' } void onMessage(List message) { println 'Received list' stop() } }final def myActor = new MyActor().start()Actors.actor { myActor 1 myActor '' myActor 1.0 myActor(new ArrayList()) myActor.join() }.join()
final Actor myActor = new DynamicDispatchActor({ when {String msg -> println 'A String'; reply 'Thanks'} when {Double msg -> println 'A Double'; reply 'Thanks'} when {msg -> println 'A something ...'; reply 'What was that?';stop()} }) myActor.start() Actors.actor { myActor 'Hello' myActor 1.0d myActor 10 as BigDecimal myActor.join() }.join()
final class MyDDA extends DynamicDispatchActor { def MyDDA(final closure) { super(closure); } void onMessage(String message) { println 'Received string' } void onMessage(Integer message) { println 'Received integer' } void onMessage(Object message) { println 'Received object' } void onMessage(List message) { println 'Received list' stop() } }final def myActor = new MyDDA({ when {BigDecimal num -> println 'Received BigDecimal'} when {Float num -> println 'Got a float'} }).start() Actors.actor { myActor 'Hello' myActor 1.0f myActor 10 as BigDecimal myActor.send([]) myActor.join() }.join()
DynamicDispatchActor can be set to behave in a fair on non-fair (default) manner. Depending on the strategy chosen, the actor either makes the thread available to other actors sharing the same parallel group (fair), or keeps the thread fot itself until the message queue gets empty (non-fair). Generally, non-fair actors perform 2 - 3 times better than fair ones.Use either the fairMessageHandler() factory method or the actor's makeFair() method.
def fairActor = Actors.fairMessageHandler {...}
Reactive Actor
The ReactiveActor class, constructed typically by calling Actors.reactor() or _DefaultPGroup.reactor()_, allow for more event-driven like approach. When a reactive actor receives a message, the supplied block of code, which makes up the reactive actor's body, is run with the message as a parameter. The result returned from the code is sent in reply.final def group = new DefaultPGroup()final def doubler = group.reactor { 2 * it }group.actor { println 'Double of 10 = ' + doubler.sendAndWait(10) }group.actor { println 'Double of 20 = ' + doubler.sendAndWait(20) }group.actor { println 'Double of 30 = ' + doubler.sendAndWait(30) }for(i in (1..10)) { println "Double of $i = ${doubler.sendAndWait(i)}" }doubler.stop() doubler.join()
import groovyx.gpars.actor.Actor import groovyx.gpars.actor.Actorsfinal def doubler = Actors.reactor { 2 * it }Actor actor = Actors.actor { (1..10).each {doubler << it} int i = 0 loop { i += 1 if (i > 10) stop() else { react {message -> println "Double of $i = $message" } } } }actor.join() doubler.stop() doubler.join()
public class ReactiveActor extends DefaultActor { Closure body void act() { loop { react {message -> reply body(message) } } } }
ReactiveActor can be set to behave in a fair on non-fair (default) manner. Depending on the strategy chosen, the actor either makes the thread available to other actors sharing the same parallel group (fair), or keeps the thread fot itself until the message queue gets empty (non-fair). Generally, non-fair actors perform 2 - 3 times better than fair ones.Use either the fairReactor() factory method or the actor's makeFair() method.
def fairActor = Actors.fairReactor {...}
5.3 Tips and Tricks
Structuring actor's code
When extending the DefaultActor class, you can call any actor's methods from within the act() method and use the react() or loop() methods in them.class MyDemoActor extends DefaultActor { protected void act() { handleA() } private void handleA() { react {a -> handleB(a) } } private void handleB(int a) { react {b -> println a + b reply a + b } } }final def demoActor = new MyDemoActor() demoActor.start()Actors.actor { demoActor 10 demoActor 20 react { println "Result: $it" } }.join()
Actor demoActor = Actors.actor {
delegate.metaClass {
handleA = {->
react {a ->
handleB(a)
}
} handleB = {a ->
react {b ->
println a + b
reply a + b
}
}
} handleA()
}Actors.actor {
demoActor 10
demoActor 20
react {
println "Result: $it"
}
}.join()
Closure handleB = {a ->
react {b ->
println a + b
reply a + b
}
}Closure handleA = {->
react {a ->
handleB(a)
}
}Actor demoActor = Actors.actor {
handleA.delegate = delegate
handleB.delegate = delegate handleA()
}Actors.actor {
demoActor 10
demoActor 20
react {
println "Result: $it"
}
}.join()
Event-driven loops
When coding event-driven actors you have to have in mind that calls to react() and loop() methods have slightly different semantics. This becomes a bit of a challenge once you try to implement any types of loops in your actors. On the other hand, if you leverage the fact that react() only schedules a continuation and returns, you may call methods recursively without fear to fill up the stack. Look at the examples below, which respectively use the three described techniques for structuring actor's code.A subclass of DefaultActorclass MyLoopActor extends DefaultActor { protected void act() { outerLoop() } private void outerLoop() { react {a -> println 'Outer: ' + a if (a != 0) innerLoop() else println 'Done' } } private void innerLoop() { react {b -> println 'Inner ' + b if (b == 0) outerLoop() else innerLoop() } } }final def actor = new MyLoopActor().start() actor 10 actor 20 actor 0 actor 0 actor.join()
Actor actor = Actors.actor { delegate.metaClass { outerLoop = {-> react {a -> println 'Outer: ' + a if (a!=0) innerLoop() else println 'Done' } } innerLoop = {-> react {b -> println 'Inner ' + b if (b==0) outerLoop() else innerLoop() } } } outerLoop() }actor 10 actor 20 actor 0 actor 0 actor.join()
Closure innerLoopClosure outerLoop = {-> react {a -> println 'Outer: ' + a if (a!=0) innerLoop() else println 'Done' } }innerLoop = {-> react {b -> println 'Inner ' + b if (b==0) outerLoop() else innerLoop() } }Actor actor = Actors.actor { outerLoop.delegate = delegate innerLoop.delegate = delegate outerLoop() }actor 10 actor 20 actor 0 actor 0 actor.join()
class MyLoopingActor extends DefaultActor { protected void act() { loop { outerLoop() } } private void outerLoop() { react {a -> println 'Outer: ' + a if (a!=0) innerLoop() else println 'Done for now, but will loop again' } } private void innerLoop() { react {b -> println 'Inner ' + b if (b == 0) outerLoop() else innerLoop() } } }final def actor = new MyLoopingActor().start() actor 10 actor 20 actor 0 actor 0 actor 10 actor.stop() actor.join()
5.4 Classic Examples using Actors
A few examples on Actors use
Examples
- The Sieve of Eratosthenes
- Sleeping Barber
- Dining Philosophers
- Word Sort
- Load Balancer
The Sieve of Eratosthenes
Problem descriptionimport groovyx.gpars.actor.DynamicDispatchActor/** * Demonstrates concurrent implementation of the Sieve of Eratosthenes using actors * * In principle, the algorithm consists of concurrently run chained filters, * each of which detects whether the current number can be divided by a single prime number. * (generate nums 1, 2, 3, 4, 5, ...) -> (filter by mod 2) -> (filter by mod 3) -> (filter by mod 5) -> (filter by mod 7) -> (filter by mod 11) -> (caution! Primes falling out here) * The chain is built (grows) on the fly, whenever a new prime is found. */int requestedPrimeNumberBoundary = 1000final def firstFilter = new FilterActor(2).start()/** * Generating candidate numbers and sending them to the actor chain */ (2..requestedPrimeNumberBoundary).each { firstFilter it } firstFilter.sendAndWait 'Poisson'/** * Filter out numbers that can be divided by a single prime number */ final class FilterActor extends DynamicDispatchActor { private final int myPrime private def follower def FilterActor(final myPrime) { this.myPrime = myPrime; } /** * Try to divide the received number with the prime. If the number cannot be divided, send it along the chain. * If there's no-one to send it to, I'm the last in the chain, the number is a prime and so I will create and chain * a new actor responsible for filtering by this newly found prime number. */ def onMessage(int value) { if (value % myPrime != 0) { if (follower) follower value else { println "Found $value" follower = new FilterActor(value).start() } } } /** * Stop the actor on poisson reception */ def onMessage(def poisson) { if (follower) { def sender = sender follower.sendAndContinue(poisson, {this.stop(); sender?.send('Done')}) //Pass the poisson along and stop after a reply } else { //I am the last in the chain stop() reply 'Done' } } }
Sleeping Barber
Problem descriptionimport groovyx.gpars.group.DefaultPGroup import groovyx.gpars.actor.DefaultActor import groovyx.gpars.group.DefaultPGroup import groovyx.gpars.actor.Actorfinal def group = new DefaultPGroup()final def barber = group.actor { final def random = new Random() loop { react {message -> switch (message) { case Enter: message.customer.send new Start() println "Barber: Processing customer ${message.customer.name}" doTheWork(random) message.customer.send new Done() reply new Next() break case Wait: println "Barber: No customers. Going to have a sleep" break } } } }private def doTheWork(Random random) { Thread.sleep(random.nextInt(10) * 1000) }final Actor waitingRoomwaitingRoom = group.actor { final int capacity = 5 final List<Customer> waitingCustomers = [] boolean barberAsleep = true loop { react {message -> switch (message) { case Enter: if (waitingCustomers.size() == capacity) { reply new Full() } else { waitingCustomers << message.customer if (barberAsleep) { assert waitingCustomers.size() == 1 barberAsleep = false waitingRoom.send new Next() } else reply new Wait() } break case Next: if (waitingCustomers.size()>0) { def customer = waitingCustomers.remove(0) barber.send new Enter(customer:customer) } else { barber.send new Wait() barberAsleep = true } } } }}class Customer extends DefaultActor { String name Actor localBarbers void act() { localBarbers << new Enter(customer:this) loop { react {message -> switch (message) { case Full: println "Customer: $name: The waiting room is full. I am leaving." stop() break case Wait: println "Customer: $name: I will wait." break case Start: println "Customer: $name: I am now being served." break case Done: println "Customer: $name: I have been served." stop(); break } } } } }class Enter { Customer customer } class Full {} class Wait {} class Next {} class Start {} class Done {}def customers = [] customers << new Customer(name:'Joe', localBarbers:waitingRoom).start() customers << new Customer(name:'Dave', localBarbers:waitingRoom).start() customers << new Customer(name:'Alice', localBarbers:waitingRoom).start()sleep 15000 customers << new Customer(name: 'James', localBarbers: waitingRoom).start() sleep 5000 customers*.join() barber.stop() waitingRoom.stop()
Dining Philosophers
Problem descriptionimport groovyx.gpars.actor.DefaultActor import groovyx.gpars.actor.ActorsActors.defaultActorPGroup.resize 5final class Philosopher extends DefaultActor { private Random random = new Random() String name def forks = [] void act() { assert 2 == forks.size() loop { think() forks*.send new Take() def messages = [] react {a -> messages << [a, sender] react {b -> messages << [b, sender] if ([a, b].any {Rejected.isCase it}) { println "$name: tOops, can't get my forks! Giving up." final def accepted = messages.find {Accepted.isCase it[0]} if (accepted!=null) accepted[1].send new Finished() } else { eat() reply new Finished() } } } } } void think() { println "$name: tI'm thinking" Thread.sleep random.nextInt(5000) println "$name: tI'm done thinking" } void eat() { println "$name: tI'm EATING" Thread.sleep random.nextInt(2000) println "$name: tI'm done EATING" } }final class Fork extends DefaultActor { String name boolean available = true void act() { loop { react {message -> switch (message) { case Take: if (available) { available = false reply new Accepted() } else reply new Rejected() break case Finished: assert !available available = true break default: throw new IllegalStateException("Cannot process the message: $message") } } } } }final class Take {} final class Accepted {} final class Rejected {} final class Finished {}def forks = [ new Fork(name:'Fork 1'), new Fork(name:'Fork 2'), new Fork(name:'Fork 3'), new Fork(name:'Fork 4'), new Fork(name:'Fork 5') ]def philosophers = [ new Philosopher(name:'Joe', forks:[forks[0], forks[1]]), new Philosopher(name:'Dave', forks:[forks[1], forks[2]]), new Philosopher(name:'Alice', forks:[forks[2], forks[3]]), new Philosopher(name:'James', forks:[forks[3], forks[4]]), new Philosopher(name:'Phil', forks:[forks[4], forks[0]]), ]forks*.start() philosophers*.start()sleep 10000 forks*.stop() philosophers*.stop()
Word sort
Given a folder name, the script will sort words in all files in the folder. The SortMaster actor creates a given number of WordSortActors , splits among them the files to sort words in and collects the results.Inspired by Scala Concurrency blog post by Michael Galpin//Messages private final class FileToSort { String fileName } private final class SortResult { String fileName; List<String> words }//Worker actor final class WordSortActor extends DefaultActor { private List<String> sortedWords(String fileName) { parseFile(fileName).sort {it.toLowerCase()} } private List<String> parseFile(String fileName) { List<String> words = [] new File(fileName).splitEachLine(' ') {words.addAll(it)} return words } void act() { loop { react {message -> switch (message) { case FileToSort: println "Sorting file=${message.fileName} on thread ${Thread.currentThread().name}" reply new SortResult(fileName: message.fileName, words: sortedWords(message.fileName)) } } } } }//Master actor final class SortMaster extends DefaultActor { String docRoot = '/' int numActors = 1 List<List<String>> sorted = [] private CountDownLatch startupLatch = new CountDownLatch(1) private CountDownLatch doneLatch private void beginSorting() { int cnt = sendTasksToWorkers() doneLatch = new CountDownLatch(cnt) } private List createWorkers() { return (1..numActors).collect {new WordSortActor().start()} } private int sendTasksToWorkers() { List<Actor> workers = createWorkers() int cnt = 0 new File(docRoot).eachFile { workers[cnt % numActors] << new FileToSort(fileName: it) cnt += 1 } return cnt } public void waitUntilDone() { startupLatch.await() doneLatch.await() } void act() { beginSorting() startupLatch.countDown() loop { react { switch (it) { case SortResult: sorted << it.words doneLatch.countDown() println "Received results for file=${it.fileName}" } } } } }//start the actors to sort words def master = new SortMaster(docRoot: 'c:/tmp/Logs/', numActors: 5).start() master.waitUntilDone() println 'Done'File file = new File("c:/tmp/Logs/sorted_words.txt") file.withPrintWriter { printer -> master.sorted.each { printer.println it } }
Load Balancer
Demonstrates work balancing among adaptable set of workers. The load balancer receives tasks and queues them in a temporary task queue. When a worker finishes his assignment, it asks the load balancer for a new task.If the load balancer doesn't have any tasks available in the task queue, the worker is stopped. If the number of tasks in the task queue exceeds certain limit, a new worker is created to increase size of the worker pool.import groovyx.gpars.actor.Actor import groovyx.gpars.actor.DefaultActor/** * Demonstrates work balancing among adaptable set of workers. * The load balancer receives tasks and queues them in a temporary task queue. * When a worker finishes his assignment, it asks the load balancer for a new task. * If the load balancer doesn't have any tasks available in the task queue, the worker is stopped. * If the number of tasks in the task queue exceeds certain limit, a new worker is created * to increase size of the worker pool. */final class LoadBalancer extends DefaultActor { int workers = 0 List taskQueue = [] private static final QUEUE_SIZE_TRIGGER = 10 void act() { loop { react { message -> switch (message) { case NeedMoreWork: if (taskQueue.size() == 0) { println 'No more tasks in the task queue. Terminating the worker.' reply DemoWorker.EXIT workers -= 1 } else reply taskQueue.remove(0) break case WorkToDo: taskQueue << message if ((workers == 0) || (taskQueue.size() >= QUEUE_SIZE_TRIGGER)) { println 'Need more workers. Starting one.' workers += 1 new DemoWorker(this).start() } } println "Active workers=${workers}tTasks in queue=${taskQueue.size()}" } } } }final class DemoWorker extends DefaultActor { final static Object EXIT = new Object() private static final Random random = new Random() Actor balancer def DemoWorker(balancer) { this.balancer = balancer } void act() { loop { this.balancer << new NeedMoreWork() react { switch (it) { case WorkToDo: processMessage(it) break case EXIT: terminate() } } } } private void processMessage(message) { synchronized (random) { Thread.sleep random.nextInt(5000) } } } final class WorkToDo {} final class NeedMoreWork {}final Actor balancer = new LoadBalancer().start()//produce tasks for (i in 1..20) { Thread.sleep 100 balancer << new WorkToDo() }//produce tasks in a parallel thread Thread.start { for (i in 1..10) { Thread.sleep 1000 balancer << new WorkToDo() } }Thread.sleep 35000 //let the queues get empty balancer << new WorkToDo() balancer << new WorkToDo() Thread.sleep 10000balancer.stop() balancer.join()
6. Agent
The Agent class, which is a thread-safe non-blocking shared mutable state wrapper implementation inspired by Agents in Clojure.A lot of the concurrency problems disappear when you eliminate the need for Shared Mutable State with your architecture. Indeed, concepts like actors, CSP or dataflow concurrency avoid or isolate mutable state completely. In some cases, however, sharing mutable data is either inevitable or makes the design more natural and understandable. Think, for example, of a shopping cart in a typical e-commerce application, when multiple AJAX requests may hit the cart with read or write requests concurrently.
Introduction
In the Clojure programing language you can find a concept of Agents, the purpose of which is to protect mutable data that need to be shared across threads. Agents hide the data and protect it from direct access. Clients can only send commands (functions) to the agent. The commands will be serialized and processed against the data one-by-one in turn. With the commands being executed serially the commands do not need to care about concurrency and can assume the data is all theirs when run. Although implemented differently, GPars Agents, called Agent , fundamentally behave like actors. They accept messages and process them asynchronously. The messages, however, must be commands (functions or Groovy closures) and will be executed inside the agent. After reception the received function is run against the internal state of the Agent and the return value of the function is considered to be the new internal state of the Agent.Essentially, agents safe-guard mutable values by allowing only a single agent-managed thread to make modifications to them. The mutable values are not directly accessible from outside, but instead requests have to be sent to the agent and the agent guarantees to process the requests sequentially on behalf of the callers. Agents guarantee sequential execution of all requests and so consistency of the values.Schematically:agent = new Agent(0) //created a new Agent wrapping an integer with initial value 0 agent.send {increment()} //asynchronous send operation, sending the increment() function … //after some delay to process the message the internal Agent's state has been updated … assert agent.val== 1
Concepts
GPars provides an Agent class, which is a special-purpose thread-safe non-blocking implementation inspired by Agents in Clojure.An Agent wraps a reference to mutable state, held inside a single field, and accepts code (closures / commands) as messages, which can be sent to the Agent just like to any other actor using the '<<' operator, the send() methods or the implicit call() method. At some point after reception of a closure / command, the closure is invoked against the internal mutable field and can make changes to it. The closure is guaranteed to be run without intervention from other threads and so may freely alter the internal state of the Agent held in the internal <i>data</i> field.The whole update process is of the fire-and-forget type, since once the message (closure) is sent to the Agent, the caller thread can go off to do other things and come back later to check the current value with Agent.val or Agent.valAsync(closure).Basic rules
- When executed, the submitted commands obtain the agent's state ar a parameter.
- The submitted commands /closures can call any methods on the agent's state.
- Replacing the state object with a new one is also possible and is done using the updateValue() method.
- The return value of the submitted closure doesn't have a special meaning and is ignored.
- If the message sent to an Agent is not a closure, it is considered to be a new value for the internal reference field.
- The val property of an Agent will wait until all preceding commands in the agent's queue are consumed and then safely return the value of the Agent.
- The instantVal property will return an immediate snapshot of the internal agent's state.
- The valAsync() method will do the same without blocking the caller.
- All Agent instances share a default daemon thread pool. Setting the threadPool property of an Agent instance will allow it to use a different thread pool.
- Exceptions thrown by the commands can be collected using the errors property.
Examples
Shared list of members
The Agent wraps a list of members, who have been added to the jug. To add a new member a message (command to add a member) has to be sent to the jugMembers Agent.import groovyx.gpars.agent.Agent import java.util.concurrent.ExecutorService import java.util.concurrent.Executors/** * Create a new Agent wrapping a list of strings */ def jugMembers = new Agent<List<String>>(['Me']) //add MejugMembers.send {it.add 'James'} //add Jamesfinal Thread t1 = Thread.start { jugMembers.send {it.add 'Joe'} //add Joe }final Thread t2 = Thread.start { jugMembers << {it.add 'Dave'} //add Dave jugMembers {it.add 'Alice'} //add Alice (using the implicit call() method) }[t1, t2]*.join() println jugMembers.val jugMembers.valAsync {println "Current members: $it"}jugMembers.await()
Shared conference counting number of registrations
The Conference class allows registration and un-registration, however these methods can only be called from the commands sent to the conference Agent.import groovyx.gpars.agent.Agent/** * Conference stores number of registrations and allows parties to register and unregister. * It inherits from the Agent class and adds the register() and unregister() private methods, * which callers may use it the commands they submit to the Conference. */ class Conference extends Agent<Long> { def Conference() { super(0) } private def register(long num) { data += num } private def unregister(long num) { data -= num } }final Agent conference = new Conference() //new Conference created/** * Three external parties will try to register/unregister concurrently */final Thread t1 = Thread.start { conference << {register(10L)} //send a command to register 10 attendees }final Thread t2 = Thread.start { conference << {register(5L)} //send a command to register 5 attendees }final Thread t3 = Thread.start { conference << {unregister(3L)} //send a command to unregister 3 attendees }[t1, t2, t3]*.join()assert 12L == conference.val
Factory methods
Agent instances can also be created using the Agent.agent() factory method.def jugMembers = Agent.agent ['Me'] //add Me
Listeners and validators
Agents allow the user to add listeners and validators. While listeners will get notified each time the internal state changes, validators get a chance to reject a coming change by throwing an exception.final Agent counter = new Agent()counter.addListener {oldValue, newValue -> println "Changing value from $oldValue to $newValue"} counter.addListener {agent, oldValue, newValue -> println "Agent $agent changing value from $oldValue to $newValue"}counter.addValidator {oldValue, newValue -> if (oldValue > newValue) throw new IllegalArgumentException('Things can only go up in Groovy')} counter.addValidator {agent, oldValue, newValue -> if (oldValue == newValue) throw new IllegalArgumentException('Things never stay the same for $agent')}counter 10 counter 11 counter {updateValue 12} counter 10 //Will be rejected counter {updateValue it - 1} //Will be rejected counter {updateValue it} //Will be rejected counter {updateValue 11} //Will be rejected counter 12 //Will be rejected counter 20 counter.await()
assert counter.hasErrors() assert counter.errors.size() == 5
Validator gotchas
With Groovy being not very strict on data types and immutability, agent users should be aware of potential bumps on the road. If the submitted code modifies the state directly, validators will not be able to un-do the change in case of a validation rule violation. There are two possible solutions available:- Make sure you never change the supplied object representing current agent state
- Use custom copy strategy on the agent to allow the agent to create copies of the internal state
//Create an agent storing names, rejecting 'Joe' final Closure rejectJoeValidator = {oldValue, newValue -> if ('Joe' in newValue) throw new IllegalArgumentException('Joe is not allowed to enter our list.')}Agent agent = new Agent([]) agent.addValidator rejectJoeValidatoragent {it << 'Dave'} //Accepted agent {it << 'Joe'} //Erroneously accepted, since by-passes the validation mechanism println agent.val//Solution 1 - never alter the supplied state object agent = new Agent([]) agent.addValidator rejectJoeValidatoragent {updateValue(['Dave', * it])} //Accepted agent {updateValue(['Joe', * it])} //Rejected println agent.val//Solution 2 - use custom copy strategy on the agent agent = new Agent([], {it.clone()}) agent.addValidator rejectJoeValidatoragent {updateValue it << 'Dave'} //Accepted agent {updateValue it << 'Joe'} //Rejected, since 'it' is now just a copy of the internal agent's state println agent.val
Grouping
By default all Agent instances belong to the same group sharing its daemon thread pool.Custom groups can also create instances of Agent. These instances will belong to the group, which created them, and will share a thread pool. To create an Agent instance belonging to a group, call the agent() factory method on the group. This way you can organize and tune performance of agents.final def group = new NonDaemonPGroup(5) //create a group around a thread pool def jugMembers = group.agent(['Me']) //add Me
The default thread pool for agents contains daemon threads. Make sure that your custom thread pools either use daemon threads, too, which can be achieved either by using DefaultPGroup or by providing your own thread factory to a thread pool constructor, or in case your thread pools use non-daemon threads, such as when using the NonDaemonPGroup group class, make sure you shutdown the group or the thread pool explicitly by calling its shutdown() method, otherwise your applications will not exit.
Direct pool replacement
Alternatively, by calling the attachToThreadPool() method on an Agent instance a custom thread pool can be specified for it.def jugMembers = new Agent<List<String>>(['Me']) //add Mefinal ExecutorService pool = Executors.newFixedThreadPool(10) jugMembers.attachToThreadPool(new DefaultPool(pool))
Remember, like actors, a single Agent instance (aka agent) can never use more than one thread at a time
The shopping cart example
import groovyx.gpars.agent.Agentclass ShoppingCart { private def cartState = new Agent([:]) //----------------- public methods below here ---------------------------------- public void addItem(String product, int quantity) { cartState << {it[product] = quantity} //the << operator sends //a message to the Agent } public void removeItem(String product) { cartState << {it.remove(product)} } public Object listContent() { return cartState.val } public void clearItems() { cartState << performClear } public void increaseQuantity(String product, int quantityChange) { cartState << this.&changeQuantity.curry(product, quantityChange) } //----------------- private methods below here --------------------------------- private void changeQuantity(String product, int quantityChange, Map items) { items[product] = (items[product] ?: 0) + quantityChange } private Closure performClear = { it.clear() } } //----------------- script code below here ------------------------------------- final ShoppingCart cart = new ShoppingCart() cart.addItem 'Pilsner', 10 cart.addItem 'Budweisser', 5 cart.addItem 'Staropramen', 20cart.removeItem 'Budweisser' cart.addItem 'Budweisser', 15println "Contents ${cart.listContent()}"cart.increaseQuantity 'Budweisser', 3 println "Contents ${cart.listContent()}"cart.clearItems() println "Contents ${cart.listContent()}"
- Public methods may internally just send the required code off to the Agent, instead of executing the same functionality directly
public void addItem(String product, int quantity) { cartState[product]=quantity}
public void addItem(String product, int quantity) { cartState << {it[product] = quantity} }
public void clearItems() { cartState << performClear }private Closure performClear = { it.clear() }
The printer service example
Another example - a not thread-safe printer service shared by multiple threads. The printer needs to have the document and quality properties set before printing, so obviously a potential for race conditions if not guarded properly. Callers don't want to block until the printer is available, which the fire-and-forget nature of actors solves very elegantly.import groovyx.gpars.agent.Agent/** * A non-thread-safe service that slowly prints documents on at a time */ class PrinterService { String document String quality public void printDocument() { println "Printing $document in $quality quality" Thread.sleep 5000 println "Done printing $document" } }def printer = new Agent<PrinterService>(new PrinterService())final Thread thread1 = Thread.start { for (num in (1..3)) { final String text = "document $num" printer << {printerService -> printerService.document = text printerService.quality = 'High' printerService.printDocument() } Thread.sleep 200 } println 'Thread 1 is ready to do something else. All print tasks have been submitted' }final Thread thread2 = Thread.start { for (num in (1..4)) { final String text = "picture $num" printer << {printerService -> printerService.document = text printerService.quality = 'Medium' printerService.printDocument() } Thread.sleep 500 } println 'Thread 2 is ready to do something else. All print tasks have been submitted' }[thread1, thread2]*.join() printer.await()
Reading the value
To follow the clojure philosophy closely the Agent class gives reads higher priority than to writes. By using the instantVal property your read request will bypass the incoming message queue of the Agent and return the current snapshot of the internal state. The val property will wait in the message queue for processing, just like the non-blocking variant valAsync(Clojure cl) , which will invoke the provided closure with the internal state as a parameter.You have to bear in mind that the instantVal property might return although correct, but randomly looking results, since the internal state of the Agent at the time of instantVal execution is non-deterministic and depends on the messages that have been processed before the thread scheduler executes the body of instantVal .The await() method allows you to wait for processing all the messages submitted to the Agent before and so blocks the calling thread.State copy strategy
To avoid leaking the internal state the Agent class allows to specify a copy strategy as the second constructor argument. With the copy strategy specified, the internal state is processed by the copy strategy closure and the output value of the copy strategy value is returned to the caller instead of the actual internal state. This applies to _instantVal_, val as well as to valAsync() .Error handling
Exceptions thrown from within the submitted commands are stored inside the agent and can be obtained from the errors property. The property gets cleared once read.def jugMembers = new Agent<List>() assert jugMembers.errors.empty jugMembers.send {throw new IllegalStateException('test1')} jugMembers.send {throw new IllegalArgumentException('test2')} jugMembers.await() List errors = jugMembers.errors assertEquals(2, errors.size()) assert errors[0] instanceof IllegalStateException assertEquals 'test1', errors[0].message assert errors[1] instanceof IllegalArgumentException assertEquals 'test2', errors[1].message assert jugMembers.errors.empty
Fair and Non-fair agents
Agents can be either fair or non-fair. Fair agents give up the thread after processing each message, non-fair agents keep a thread until their message queue is empty. As a result, non-fair agents tend to perform better than fair ones. The default setting for all Agent instances is to be non-fair, however by calling its makeFair() method the instance can be made fair.def jugMembers = new Agent<List>(['Me']) //add Me
jugMembers.makeFair()
7. Dataflow Concurrency
Dataflow concurrency offers an alternative concurrency model, which is inherently safe and robust.Introduction
Check out the small example written in Groovy using GPars, which sums results of calculations performed by three concurrently run tasks:import static groovyx.gpars.dataflow.DataFlow.taskfinal def x = new DataFlowVariable() final def y = new DataFlowVariable() final def z = new DataFlowVariable()task { z << x.val + y.val }task { x << 10 }task { y << 5 }println "Result: ${z.val}"
import static groovyx.gpars.dataflow.DataFlow.taskfinal def df = new DataFlows()task { df.z = df.x + df.y }task { df.x = 10 }task { df.y = 5 }println "Result: ${df.z}"
The bind operation of dataflow variables silently accepts re-binding to a value, which is equal to an already bound value. Call bindUnique to reject equal values on already-bound variables.
Benefits
Here's what you gain by using Dataflow Concurrency (by Jonas Bonér ):- No race-conditions
- No live-locks
- Deterministic deadlocks
- Completely deterministic programs
- BEAUTIFUL code.
Concepts
Dataflow programming
Quoting Wikipedia
Operations (in Dataflow programs) consist of "black boxes" with inputs and outputs, all of which are always explicitly defined. They run as soon as all of their inputs become valid, as opposed to when the program encounters them. Whereas a traditional program essentially consists of a series of statements saying "do this, now do this", a dataflow program is more like a series of workers on an assembly line, who will do their assigned task as soon as the materials arrive. This is why dataflow languages are inherently parallel; the operations have no hidden state to keep track of, and the operations are all "ready" at the same time.Principles
With Dataflow Concurrency you can safely share variables across tasks. These variables (in Groovy instances of the DataFlowVariable class) can only be assigned (using the '<<' operator) a value once in their lifetime. The values of the variables, on the other hand, can be read multiple times (in Groovy through the val property), even before the value has been assigned. In such cases the reading task is suspended until the value is set by another task. So you can simply write your code for each task sequentially using Dataflow Variables and the underlying mechanics will make sure you get all the values you need in a thread-safe manner.In brief, you generally perform three operations with Dataflow variables:- Create a dataflow variable
- Wait for the variable to be bound (read it)
- Bind the variable (write to it)
- When the program encounters an unbound variable it waits for a value.
- It is not possible to change the value of a dataflow variable once it is bound.
- Dataflow variables makes it easy to create concurrent stream agents.
Dataflow Queues and Broadcasts
Before you go to check the samples of using Dataflow Variables, Tasks and Operators, you should know a bit about streams and queues to have a full picture of Dataflow Concurrency. Except for dataflow variables there are also the concepts of DataFlowQueues and DataFlowBroadcast that you can leverage in your code. You may think of them as thread-safe buffers or queues for message transfer among concurrent tasks or threads. Check out a typical producer-consumer demo:import static groovyx.gpars.dataflow.DataFlow.taskdef words = ['Groovy', 'fantastic', 'concurrency', 'fun', 'enjoy', 'safe', 'GPars', 'data', 'flow'] final def buffer = new DataFlowQueue()task { for (word in words) { buffer << word.toUpperCase() //add to the buffer } }task { while(true) println buffer.val //read from the buffer in a loop }
The DataFlowChannel interface combines two interfaces, each serving its purpose:You may prefer using these dedicated interfaces instead of the general DataFlowChannel interface, to better express the intended usage.
- DataFlowReadChannel holding all the methods necessary for reading values from a channel
- DataFlowWriteChannel holding all the methods necessary for writing values into a channel
Point-to-point communication
The DataFlowQueue class can be viewed as a point-to-point (1 to 1, many to 1) communication channel. It allows one or more producers send messages to one reader. If multiple readers read from the same DataFlowQueue , they will each consume different messages. Or to put it a different way, each message is consumed by exactly one reader. You can easily imagine a simple load-balancing scheme built around a shared DataFlowQueue with readers being added dynamically when the consumer part of your algorithm needs to scale up. This is also a useful default choice when connecting tasks or operators.Publish-subscribe communication
The DataFlowBroadcast class offers a publish-subscribe (1 to many, many to many) communication model. One or more producers write messages, while all registered readers will receive all the messages. Each message is thus consumed by all readers with a valid subscription at the moment when the message is being written to the channel. The readers subscribe by calling the createReadChannel() method.DataFlowWriteChannel broadcastStream = new DataFlowBroadcast()
DataFlowReadChannel stream1 = broadcastStream.createReadChannel()
DataFlowReadChannel stream2 = broadcastStream.createReadChannel()
broadcastStream << 'Message1'
broadcastStream << 'Message2'
broadcastStream << 'Message3'
assert stream1.val == stream2.val
assert stream1.val == stream2.val
assert stream1.val == stream2.val
DataFlowStream
The DataFlowStream class represents a deterministic dataflow channel. It is build around the concept of a functional queue and so provides a lock-free thread-safe implementation for message passing. Essentially, you may think of DataFlowStream as a 1 to many communication channel, since when a reader consumes a messages, other readers will still be able to read the message. Also, all messages arrive to all readers in the same order. Since DataFlowStream is implemented as a functional queue, its API requires that users traverse the values in the stream themselves. On the other hand DataFlowStream offers handy methods for value filtering or transformation together with interesting performance characteristics.The DataFlowStream class, unlike the other communication elements, does not implement the DataFlowChannel interface, since the semantics of its use is different. Use DataFlowStreamReadAdapter and DataFlowStreamWriteAdapter classes to wrap instances of the DataFlowChannel class in DataFlowReadChannel or DataFlowWriteChannel implementations.
import groovyx.gpars.dataflow.stream.DataFlowStream import groovyx.gpars.group.DefaultPGroup import groovyx.gpars.scheduler.ResizeablePool/** * Demonstrates concurrent implementation of the Sieve of Eratosthenes using dataflow tasks * * In principle, the algorithm consists of a concurrently run chained filters, * each of which detects whether the current number can be divided by a single prime number. * (generate nums 1, 2, 3, 4, 5, ...) -> (filter by mod 2) -> (filter by mod 3) -> (filter by mod 5) -> (filter by mod 7) -> (filter by mod 11) -> (caution! Primes falling out here) * The chain is built (grows) on the fly, whenever a new prime is found *//** * We need a resizeable thread pool, since tasks consume threads while waiting blocked for values at DataFlowQueue.val */ group = new DefaultPGroup(new ResizeablePool(true))final int requestedPrimeNumberCount = 100/** * Generating candidate numbers */ final DataFlowStream candidates = new DataFlowStream() group.task { candidates.generate(2, {it + 1}, {it < 1000}) }/** * Chain a new filter for a particular prime number to the end of the Sieve * @param inChannel The current end channel to consume * @param prime The prime number to divide future prime candidates with * @return A new channel ending the whole chain */ def filter(DataFlowStream inChannel, int prime) { inChannel.filter { number -> group.task { number % prime != 0 } } }/** * Consume Sieve output and add additional filters for all found primes */ def currentOutput = candidates requestedPrimeNumberCount.times { int prime = currentOutput.first println "Found: $prime" currentOutput = filter(currentOutput, prime) }
DataFlowStream Adapters
Since the DataFlowStream API as well as the semantics of its use are very different from the one defined by DataFlow(Read/Write)Channel , adapters have to be used in order to allow DataFlowStreams to be used with other dataflow elements. The DataFlowStreamReadAdapter class will wrap a DataFlowStream with necessary methods to read values, while the DataFlowStreamWriteAdapter class will provide write methods around the wrapped DataFlowStream .It is important to mention that the DataFlowStreamWriteAdapter is thread safe allowing multiple threads to add values to the wrapped DataFlowStream through the adapter. On the other hand, DataFlowStreamReadAdapter is designed to be used by a single thread.To minimize the overhead and stay in-line with the DataFlowStream semantics, the DataFlowStreamReadAdapter class is not thread-safe and should only be used from within a single thread. If multiple threads need to read from a DataFlowStream, they should each create their own wrapping DataFlowStreamReadAdapter .Thanks to the adapters DataFlowStream can be used for communication between operators or selectors, which expect DataFlow(Read/Write)Channels .
import groovyx.gpars.dataflow.DataFlowQueue import groovyx.gpars.dataflow.stream.DataFlowStream import groovyx.gpars.dataflow.stream.DataFlowStreamReadAdapter import groovyx.gpars.dataflow.stream.DataFlowStreamWriteAdapter import static groovyx.gpars.dataflow.DataFlow.selector import static groovyx.gpars.dataflow.DataFlow.operator/** * Demonstrates the use of DataFlowStreamAdapters to allow dataflow operators to use DataFlowStreams */final DataFlowStream a = new DataFlowStream() final DataFlowStream b = new DataFlowStream() def aw = new DataFlowStreamWriteAdapter(a) def bw = new DataFlowStreamWriteAdapter(b) def ar = new DataFlowStreamReadAdapter(a) def br = new DataFlowStreamReadAdapter(b)def result = new DataFlowQueue()def op1 = operator(ar, bw) { bindOutput it } def op2 = selector([br], [result]) { result << it }aw << 1 aw << 2 aw << 3 assert([1, 2, 3] == [result.val, result.val, result.val]) op1.stop() op2.stop() op1.join() op2.join()
import groovyx.gpars.dataflow.Select import groovyx.gpars.dataflow.stream.DataFlowStream import groovyx.gpars.dataflow.stream.DataFlowStreamReadAdapter import groovyx.gpars.dataflow.stream.DataFlowStreamWriteAdapter import static groovyx.gpars.dataflow.DataFlow.select import static groovyx.gpars.dataflow.DataFlow.task/** * Demonstrates the use of DataFlowStreamAdapters to allow dataflow select to select on DataFlowStreams */final DataFlowStream a = new DataFlowStream() final DataFlowStream b = new DataFlowStream() def aw = new DataFlowStreamWriteAdapter(a) def bw = new DataFlowStreamWriteAdapter(b) def ar = new DataFlowStreamReadAdapter(a) def br = new DataFlowStreamReadAdapter(b)final Select<?> select = select(ar, br) task { aw << 1 aw << 2 aw << 3 } assert 1 == select().value assert 2 == select().value assert 3 == select().value task { bw << 4 aw << 5 bw << 6 } def result = (1..3).collect{select()}.sort{it.value} assert result*.value == [4, 5, 6] assert result*.index == [1, 0, 1]
If you don't need any of the functional queue DataFlowStream-special functionality, like generation, filtering or mapping, you may consider using the DataFlowBroadcast class instead, which offers the publish-subscribe communication model through the DataFlowChannel interface.
Bind handlers
def a = new DataFlowVariable() a >> {println "The variable has just been bound to $it"} a.whenBound {println "Just to confirm that the variable has been really set to $it"} ...
def queue = new DataFlowQueue() queue.wheneverBound {println "A value $it arrived to the queue"}
Dataflow variables and broadcasts are one of several possible ways to implement Parallel Speculations . For details, please check out Parallel Speculations in the Parallel Collections section of the User Guide.
Further reading
Scala Dataflow library by Jonas BonérJVM concurrency presentation slides by Jonas BonérDataflow Concurrency library for Ruby7.1 Tasks
The Dataflow tasks give you an easy-to-grasp abstraction of mutually-independent logical tasks or threads, which can run concurrently and exchange data solely through Dataflow Variables, Queues, Broadcasts and Streams. DataFlow tasks with their easy-to-express mutual dependencies and inherently sequential body could also be used as a practical implementation of UML Activity Diagrams .Check out the examples.A simple mashup example
In the example we're downloading the front pages of three popular web sites, each in their own task, while in a separate task we're filtering out sites talking about Groovy today and forming the output. The output task synchronizes automatically with the three download tasks on the three Dataflow variables through which the content of each website is passed to the output task.import static groovyx.gpars.GParsPool.* import groovyx.gpars.dataflow.DataFlowVariable import static groovyx.gpars.dataflow.DataFlow.task /** * A simple mashup sample, downloads content of three websites * and checks how many of them refer to Groovy. */def dzone = new DataFlowVariable() def jroller = new DataFlowVariable() def theserverside = new DataFlowVariable()task { println 'Started downloading from DZone' dzone << 'http://www.dzone.com'.toURL().text println 'Done downloading from DZone' }task { println 'Started downloading from JRoller' jroller << 'http://www.jroller.com'.toURL().text println 'Done downloading from JRoller' }task { println 'Started downloading from TheServerSide' theserverside << 'http://www.theserverside.com'.toURL().text println 'Done downloading from TheServerSide' }task { withPool { println "Number of Groovy sites today: " + ([dzone, jroller, theserverside].findAllParallel { it.val.toUpperCase().contains 'GROOVY' }).size() } }.join()
Grouping tasks
Dataflow tasks can be organized into groups to allow for performance fine-tuning. Groups provide a handy task() factory method to create tasks attached to the groups.import groovyx.gpars.group.DefaultPGroupdef group = new DefaultPGroup()group.with { task { … } task { … } }
The default thread pool for dataflow tasks contains daemon threads, which means your application will exit as soon as the main thread finishes and won't wait for all tasks to complete. When grouping tasks, make sure that your custom thread pools either use daemon threads, too, which can be achieved by using DefaultPGroup or by providing your own thread factory to a thread pool constructor, or in case your thread pools use non-daemon threads, such as when using the NonDaemonPGroup group class, make sure you shutdown the group or the thread pool explicitly by calling its shutdown() method, otherwise your applications will not exit.
A mashup variant with methods
To avoid giving you wrong impression about structuring the Dataflow code, here's a rewrite of the mashup example, with a downloadPage() method performing the actual download in a separate task and returning a DataFlowVariable instance, so that the main application thread could eventually get hold of the downloaded content. Dataflow variables can obviously be passed around as parameters or return values.package groovyx.gpars.samples.dataflowimport static groovyx.gpars.GParsExecutorsPool.* import groovyx.gpars.dataflow.DataFlowVariable import static groovyx.gpars.dataflow.DataFlow.task /** * A simple mashup sample, downloads content of three websites and checks how many of them refer to Groovy. */ final List urls = ['http://www.dzone.com', 'http://www.jroller.com', 'http://www.theserverside.com']task { def pages = urls.collect { downloadPage(it) } withPool { println "Number of Groovy sites today: " + (pages.findAllParallel { it.val.toUpperCase().contains 'GROOVY' }).size() } }.join()def downloadPage(def url) { def page = new DataFlowVariable() task { println "Started downloading from $url" page << url.toURL().text println "Done downloading from $url" } return page }
A physical calculation example
Dataflow programs naturally scale with the number of processors. Up to a certain level, the more processors you have the faster the program runs. Check out, for example, the following script, which calculates parameters of a simple physical experiment and prints out the results. Each task performs its part of the calculation and may depend on values calculated by some other tasks as well as its result might be needed by some of the other tasks. With Dataflow Concurrency you can split the work between tasks or reorder the tasks themselves as you like and the dataflow mechanics will ensure the calculation will be accomplished correctly.import groovyx.gpars.dataflow.DataFlowVariable import static groovyx.gpars.dataflow.DataFlow.taskfinal def mass = new DataFlowVariable() final def radius = new DataFlowVariable() final def volume = new DataFlowVariable() final def density = new DataFlowVariable() final def acceleration = new DataFlowVariable() final def time = new DataFlowVariable() final def velocity = new DataFlowVariable() final def decelerationForce = new DataFlowVariable() final def deceleration = new DataFlowVariable() final def distance = new DataFlowVariable()def t = task { println """ Calculating distance required to stop a moving ball. ==================================================== The ball has a radius of ${radius.val} meters and is made of a material with ${density.val} kg/m3 density, which means that the ball has a volume of ${volume.val} m3 and a mass of ${mass.val} kg. The ball has been accelerating with ${acceleration.val} m/s2 from 0 for ${time.val} seconds and so reached a velocity of ${velocity.val} m/s.Given our ability to push the ball backwards with a force of ${decelerationForce.val} N (Newton), we can cause a deceleration of ${deceleration.val} m/s2 and so stop the ball at a distance of ${distance.val} m.======================================================================================================================= This example has been calculated asynchronously in multiple tasks using GPars DataFlow concurrency in Groovy. Author: ${author.val} """ System.exit 0 }task { mass << volume.val * density.val }task { volume << Math.PI * (radius.val ** 3) }task { radius << 2.5 density << 998.2071 //water acceleration << 9.80665 //free fall decelerationForce << 900 }task { println 'Enter your name:' def name = new InputStreamReader(System.in).readLine() author << (name?.trim()?.size()>0 ? name : 'anonymous') }task { time << 10 velocity << acceleration.val * time.val }task { deceleration << decelerationForce.val / mass.val }task { distance << deceleration.val * ((velocity.val/deceleration.val) ** 2) * 0.5 }t.join()
Deterministic deadlocks
If you happen to introduce a deadlock in your dependencies, the deadlock will occur each time you run the code. No randomness allowed. That's one of the benefits of Dataflow concurrency. Irrespective of the actual thread scheduling scheme, if you don't get a deadlock in tests, you won't get them in production.task { println a.val b << 'Hi there' }task { println b.val a << 'Hello man' }
DataFlows map
As a handy shortcut the DataFlows class can help you reduce the amount of code you have to write to leverage Dataflow variables.def df = new DataFlows()
df.x = 'value1'
assert df.x == 'value1'DataFlow.task {df.y = 'value2}assert df.y == 'value2'
Mixing DataFlows and Groovy with blocks
When inside a with block of a DataFlows instance, the dataflow variables stored inside the DataFlows instance can be accessed directly without the need to prefix them with the DataFlows instance identifier.new DataFlows().with {
x = 'value1'
assert x == 'value1' DataFlow.task {y = 'value2} assert y == 'value2'
}
Returning a value from a task
Typically dataflow tasks communicate through dataflow variables. On top of that, tasks can also return values, again through a dataflow variable. When you invoke the task() factory method, you get back an instance of DataFlowVariable, on which you can listen for the task's return value, just like when using any other DataFlowVariable.final DataFlowVariable t1 = task { return 10 } final DataFlowVariable t2 = task { return 20 } def results = [t1, t2]*.val println 'Both sub-tasks finished and returned values: ' + results
def task = task { println 'The task is running and calculating the return value' 30 } task >> {value -> println "The task finished and returned $value"}
task { final DataFlowVariable t1 = task { println 'First sub-task running.' } final DataFlowVariable t2 = task { println 'Second sub-task running' } [t1, t2]*.join() println 'Both sub-tasks finished' }.join()
7.2 Selects
Frequently a value needs to be obtained from one of several dataflow channels (variables, queues, broadcasts or streams). The Select class is suitable for such scenarios. Select can scan multiple dataflow channels and pick one channel from all the input channels, which currently have a value available for read. The value from that channels is read and returned to the caller together with the index of the originating channel. Picking the channel is either random, or based on channel priority, in which case channels with lower position index in the Select constructor have higher priority.Selecting a value from multiple channels
import groovyx.gpars.dataflow.DataFlowQueue import groovyx.gpars.dataflow.DataFlowVariable import static groovyx.gpars.dataflow.DataFlow.select import static groovyx.gpars.dataflow.DataFlow.task/** * Shows a basic use of Select, which monitors a set of input channels for values and makes these values * available on its output irrespective of their original input channel. * Note that dataflow variables and queues can be combined for Select. * * You might also consider checking out the prioritySelect method, which prioritizes values by the index of their input channel */ def a = new DataFlowVariable() def b = new DataFlowVariable() def c = new DataFlowQueue()task { sleep 3000 a << 10 }task { sleep 1000 b << 20 }task { sleep 5000 c << 30 }def select = select([a, b, c]) println "The fastest result is ${select().value}"
Note that the return type from select() is SelectResult , holding the value as well as the originating channel index.There are multiple ways to read values from a Select:
def sel = select(a, b, c, d) def result = sel.select() //Random selection def result = sel() //Random selection (a short-hand variant) def result = sel.select([true, true, false, true]) //Random selection with guards specified def result = sel([true, true, false, true]) //Random selection with guards specified (a short-hand variant) def result = sel.prioritySelect() //Priority selection def result = sel.prioritySelect([true, true, false, true]) //Priority selection with guards specifies
def handler = actor {...} def sel = select(a, b, c, d)sel.select(handler) //Random selection sel(handler) //Random selection (a short-hand variant) sel.select(handler, [true, true, false, true]) //Random selection with guards specified sel(handler, [true, true, false, true]) //Random selection with guards specified (a short-hand variant) sel.prioritySelect(handler) //Priority selection sel.prioritySelect(handler, [true, true, false, true]) //Priority selection with guards specifies
Guards
Guards allow the caller to omit some input channels from the selection. Guards are specified as a List of boolean flags passed to the select() or prioritySelect() methods.def sel = select(leaders, seniors, experts, juniors) def teamLead = sel([true, true, false, false]).value //Only 'leaders' and 'seniors' qualify for becoming a teamLead here
import groovyx.gpars.dataflow.DataFlowQueue import static groovyx.gpars.dataflow.DataFlow.select import static groovyx.gpars.dataflow.DataFlow.task/** * Demonstrates the ability to enable/disable channels during a value selection on a select by providing boolean guards. */ final DataFlowQueue operations = new DataFlowQueue() final DataFlowQueue numbers = new DataFlowQueue()def t = task { final def select = select(operations, numbers) 3.times { def instruction = select([true, false]).value def num1 = select([false, true]).value def num2 = select([false, true]).value final def formula = "$num1 $instruction $num2" println "$formula = ${new GroovyShell().evaluate(formula)}" } }task { operations << '+' operations << '+' operations << '*' }task { numbers << 10 numbers << 20 numbers << 30 numbers << 40 numbers << 50 numbers << 60 }t.join()
Priority Select
When certain channels should have precedence over others when selecting, the prioritySelect methods should be used instead./** * Shows a basic use of Priority Select, which monitors a set of input channels for values and makes these values * available on its output irrespective of their original input channel. * Note that dataflow variables, queues and broadcasts can be combined for Select. * Unlike plain select method call, the prioritySelect call gives precedence to input channels with lower index. * Available messages from high priority channels will be served before messages from lower-priority channels. * Messages received through a single input channel will have their mutual order preserved. * */ def critical = new DataFlowVariable() def ordinary = new DataFlowQueue() def whoCares = new DataFlowQueue()task { ordinary << 'All working fine' whoCares << 'I feel a bit tired' ordinary << 'We are on target' }task { ordinary << 'I have just started my work. Busy. Will come back later...' sleep 5000 ordinary << 'I am done for now' }task { whoCares << 'Huh, what is that noise' ordinary << 'Here I am to do some clean-up work' whoCares << 'I wonder whether unplugging this cable will eliminate that nasty sound.' critical << 'The server room goes on UPS!' whoCares << 'The sound has disappeared' }def select = select([critical, ordinary, whoCares]) println 'Starting to monitor our IT department' sleep 3000 10.times {println "Received: ${select.prioritySelect().value}"}
7.3 Operators
Dataflow Operators and Selectors provide a full Dataflow implementation with all the usual ceremony.Concepts
Full dataflow concurrency builds on the concept of channels connecting operators and selectors, which consume values coming through input channels, transform them into new values and output the new values into their output channels. While Operators wait for all input channels to have a value available for read before they start process them, Selectors are triggered by a value available on any of the input channels.operator(inputs: [a, b, c], outputs: [d]) {x, y, z ->
…
bindOutput 0, x + y + z
}
/** * CACHE * * Caches sites' contents. Accepts requests for url content, outputs the content. Outputs requests for download * if the site is not in cache yet. */ operator(inputs: [urlRequests], outputs: [downloadRequests, sites]) {request -> if (!request.content) { println "[Cache] Retrieving ${request.site}" def content = cache[request.site] if (content) { println "[Cache] Found in cache" bindOutput 1, [site: request.site, word:request.word, content: content] } else { def downloads = pendingDownloads[request.site] if (downloads != null) { println "[Cache] Awaiting download" downloads << request } else { pendingDownloads[request.site] = [] println "[Cache] Asking for download" bindOutput 0, request } } } else { println "[Cache] Caching ${request.site}" cache[request.site] = request.content bindOutput 1, request def downloads = pendingDownloads[request.site] if (downloads != null) { for (downloadRequest in downloads) { println "[Cache] Waking up" bindOutput 1, [site: downloadRequest.site, word:downloadRequest.word, content: request.content] } pendingDownloads.remove(request.site) } } }
op.metaClass.reportError = {Throwable e ->
//handle the exception
stop() //You can also stop the operator
}
Types of operators
There are specialized versions of operators serving specific purposes:- operator - the basic general-purpose operator
- selector - operator that is triggered by a value being available in any of its input channels
- prioritySelector - a selector that prefers delivering messages from lower-indexed input channels over higher-indexed ones
- splitter - a single-input operator copying its input values to all of its output channels
Chaining operators
Operators are typically combined into networks, when some operators consume output by other operators.operator(inputs:[a, b], outputs:[c, d]) {...}
splitter(c, [e, f])
selector(inputs:[e, d]: outputs:[]) {...}
def op1 = operator(inputs:[a, b], outputs:[c, d]) {...}
def sp1 = splitter(op1.outputs[0], [e, f]) //takes the first output of op1
selector(inputs:[sp1.outputs[0], op1.outputs[1]]: outputs:[]) {...} //takes the first output of sp1 and the second output of op1
Parallelize operators
By default an operator's body is processed by a single thread at a time. While this is a safe setting allowing the operator's body to be written in a non-thread-safe manner, once an operator becomes "hot" and data start to accumulate in the operator's input queues, you might consider allowing multiple threads to run the operator's body concurrently. Bear in mind that in such a case you need to avoid or protect shared resources from multi-threaded access. To enable multiple threads to run the operator's body concurrently, pass an extra maxForks parameter when creating an operator:def op = operator(inputs: [a, b, c], outputs: [d, e], maxForks: 2) {x, y, z ->
bindOutput 0, x + y + z
bindOutput 1, x * y * z
}
Please always make sure the group serving the operator holds enough threads to support all requested forks.The default group uses a resizeable thread pool as so will never run out of threads.def group = new DefaultPGroup(10) group.operator((inputs: [a, b, c], outputs: [d, e], maxForks: 5) {x, y, z -> ...}
Synchronizing the output
When enabling internal parallelization of an operator by setting the value for maxForks to a value greater than 1 it is important to remember that without explicit or implicit synchronization in the operators' body race-conditions may occur. Especially bear in mind that values written to multiple output channels are not guarantied to be written atomically in the same order to all the channelsoperator(inputs:[inputChannel], outputs:[a, b], maxForks:5) {msg ->
bindOutput 0, msg
bindOutput 1, msg
}
inputChannel << 1
inputChannel << 2
inputChannel << 3
inputChannel << 4
inputChannel << 5
a -> 1, 3, 2, 4, 5 b -> 2, 1, 3, 5, 4
def lock = new Object() operator(inputs:[inputChannel], outputs:[a, b], maxForks:5) {msg -> doStuffThatIsThreadSafe() synchronized(lock) { doSomethingThatMustNotBeAccessedByMultipleThreadsAtTheSameTime() bindOutput 0, msg bindOutput 1, 2*msg } }
operator(inputs:[inputChannel], outputs:[a, b], maxForks:5) {msg ->
doStuffThatIsThreadSafe()
bindAllOutputValuesAtomically msg, 2*msg
}
}
Using the bindAllOutputs or the bindAllOutputValues methods will not guarantee atomicity of writes across al the output channels when using internal parallelism. If preserving the order of messages in multiple output channels is not an issue, bindAllOutputs as well as bindAllOutputValues will provide better performance over the atomic variants.
Stopping operators
Dataflow operators and selectors can be stopped in two ways:- by calling the stop() method on all operators that need to be stopped
- by sending a poisson message.
def op1 = operator(inputs: [a, b, c], outputs: [d, e]) {x, y, z -> }def op2 = selector(inputs: [d], outputs: [f, out]) { }def op3 = prioritySelector(inputs: [e, f], outputs: [b]) {value, index -> }[op1, op2, op3]*.stop() //Stop all operators by calling the stop() method on them
op1.join()
op2.join()
op3.join()
def op1 = operator(inputs: [a, b, c], outputs: [d, e]) {x, y, z -> }def op2 = selector(inputs: [d], outputs: [f, out]) { }def op3 = prioritySelector(inputs: [e, f], outputs: [b]) {value, index -> }a << DataFlowPoisson.instance //Send the poissonop1.join()
op2.join()
op3.join()
Grouping operators
Dataflow operators can be organized into groups to allow for performance fine-tuning. Groups provide a handy operator() factory method to create tasks attached to the groups.import groovyx.gpars.group.DefaultPGroupdef group = new DefaultPGroup()group.with { operator(inputs: [a, b, c], outputs: [d]) {x, y, z -> … bindOutput 0, x + y + z } }
The default thread pool for dataflow operators contains daemon threads, which means your application will exit as soon as the main thread finishes and won't wait for all tasks to complete. When grouping operators, make sure that your custom thread pools either use daemon threads, too, which can be achieved by using DefaultPGroup or by providing your own thread factory to a thread pool constructor, or in case your thread pools use non-daemon threads, such as when using the NonDaemonPGroup group class, make sure you shutdown the group or the thread pool explicitly by calling its shutdown() method, otherwise your applications will not exit.
Selectors
Selector's body should be a closure consuming either one or two arguments.selector ([inputs : [a, b, c], outputs : [d, e]) {value -> .... }
selector ([inputs : [a, b, c], outputs : [d, e]) {value, index -> .... }
Priority Selector
When priorities need to be preserved among input channels, a DataFlowPrioritySelector should be used.prioritySelector[inputs : [a, b, c], outputs : [d, e]) {value, index -> … }
Join selector
A selector without a body closure specified will copy all incoming values to all of its output channels.def join = selector ([inputs : [programmers, analysis, managers], outputs : [employees, colleagues])
Internal parallelism
The maxForks attribute allowing for internal selectors parallelism is also available.selector ([inputs : [a, b, c], outputs : [d, e], maxForks : 5) {value -> .... }
Guards
Just like Selects , Selectors also allow the users to temporarily include/exclude individual input channels from selection. The guards input property can be used to set the initial mask on all input channels and the setGuards and setGuard methods are then available in the selector's body.import groovyx.gpars.dataflow.DataFlowQueue import static groovyx.gpars.dataflow.DataFlow.selector import static groovyx.gpars.dataflow.DataFlow.task/** * Demonstrates the ability to enable/disable channels during a value selection on a select by providing boolean guards. */ final DataFlowQueue operations = new DataFlowQueue() final DataFlowQueue numbers = new DataFlowQueue()def instruction def nums = []selector(inputs: [operations, numbers], outputs: [], guards: [true, false]) {value, index -> //initial guards is set here if (index == 0) { instruction = value setGuard(0, false) //setGuard() used here setGuard(1, true) } else nums << value if (nums.size() == 2) { setGuards([true, false]) //setGuards() used here final def formula = "${nums[0]} $instruction ${nums[1]}" println "$formula = ${new GroovyShell().evaluate(formula)}" nums.clear() } }task { operations << '+' operations << '+' operations << '*' }task { numbers << 10 numbers << 20 numbers << 30 numbers << 40 numbers << 50 numbers << 60 }
Avoid combining guards and maxForks greater than 1. Although the Selector is thread-safe and won't be damaged in any way, the guards are likely not to be set the way you expect. The multiple threads running selector's body concurrently will tend to over-write each-other's settings to the guards property.
7.4 Dataflow implementation
The Dataflow Concurrency in GPars builds on top of its actor support. All of the dataflow tasks share a thread pool and so the number threads created through DataFlow.task() factory method don't need to correspond to the number of physical threads required from the system. The PGroup.task() factory method can be used to attach the created task to a group. Since each group defines its own thread pool, you can easily organize tasks around different thread pools just like you do with actors.Combining actors and Dataflow Concurrency
The good news is that you can combine actors and Dataflow Concurrency in any way you feel fit for your particular problem at hands. You can freely you use Dataflow Variables from actors.final DataFlowVariable a = new DataFlowVariable()final Actor doubler = Actors.actor { react {message-> a << 2 * message } }final Actor fakingDoubler = actor { react { doubler.send it //send a number to the doubler println "Result ${a.val}" //wait for the result to be bound to 'a' } }fakingDoubler << 10
Using plain java threads
The DataFlowVariable as well as the DataFlowQueue classes can obviously be used from any thread of your application, not only from the tasks created by DataFlow.task() . Consider the following example:import groovyx.gpars.dataflow.DataFlowVariablefinal DataFlowVariable a = new DataFlowVariable<String>() final DataFlowVariable b = new DataFlowVariable<String>()Thread.start { println "Received: $a.val" Thread.sleep 2000 b << 'Thank you' }Thread.start { Thread.sleep 2000 a << 'An important message from the second thread' println "Reply: $b.val" }
7.5 Classic examples
The Sieve of Eratosthenes implementation using dataflow tasks
import groovyx.gpars.dataflow.DataFlowQueue import static groovyx.gpars.dataflow.DataFlow.task/** * Demonstrates concurrent implementation of the Sieve of Eratosthenes using dataflow tasks */final int requestedPrimeNumberCount = 1000final DataFlowQueue initialChannel = new DataFlowQueue()/** * Generating candidate numbers */ task { (2..10000).each { initialChannel << it } }/** * Chain a new filter for a particular prime number to the end of the Sieve * @param inChannel The current end channel to consume * @param prime The prime number to divide future prime candidates with * @return A new channel ending the whole chain */ def filter(inChannel, int prime) { def outChannel = new DataFlowQueue() task { while (true) { def number = inChannel.val if (number % prime != 0) { outChannel << number } } } return outChannel }/** * Consume Sieve output and add additional filters for all found primes */ def currentOutput = initialChannel requestedPrimeNumberCount.times { int prime = currentOutput.val println "Found: $prime" currentOutput = filter(currentOutput, prime) }
The Sieve of Eratosthenes implementation using a combination of dataflow tasks and operators
import groovyx.gpars.dataflow.DataFlowQueue import static groovyx.gpars.dataflow.DataFlow.operator import static groovyx.gpars.dataflow.DataFlow.task /** * Demonstrates concurrent implementation of the Sieve of Eratosthenes using dataflow tasks and operators */ final int requestedPrimeNumberCount = 100 final DataFlowQueue initialChannel = new DataFlowQueue() /** * Generating candidate numbers */ task { (2..1000).each { initialChannel << it } } /** * Chain a new filter for a particular prime number to the end of the Sieve * @param inChannel The current end channel to consume * @param prime The prime number to divide future prime candidates with * @return A new channel ending the whole chain */ def filter(inChannel, int prime) { def outChannel = new DataFlowQueue() operator([inputs: [inChannel], outputs: [outChannel]]) { if (it % prime != 0) { bindOutput it } } return outChannel } /** * Consume Sieve output and add additional filters for all found primes */ def currentOutput = initialChannel requestedPrimeNumberCount.times { int prime = currentOutput.val println "Found: $prime" currentOutput = filter(currentOutput, prime) }
8. Tips
General GPars Tips
Grouping
High-level concurrency concepts, like Agents, Actors or Dataflow tasks and operators can be grouped around shared thread pools. The PGroup class and its sub-classes represent convenient GPars wrappers around thread pools. Objects created using the group's factory methods will share the group's thread pool.def group1 = new DefaultPGroup() def group2 = new NonDaemonPGroup()group1.with { task {...} task {...} def op = operator(...) {...} def actor = actor{...} def anotherActor = group2.actor{...} //will belong to group2 def agent = safe(0) }
When customizing the thread pools for groups, consider using the existing GPars implementations - the DefaultPool or ResizeablePool classes. Or you may create your own implementation of the groovyx.gpars.scheduler.Pool interface to pass to the DefaultPGroup or NonDaemonPGroup constructors.
Java API
Most of GPars functionality can be used from Java just as well as from Groovy. Checkout the 2.6 Java API - Using GPars from Java section of the User Guide and experiment with the maven-based stand-alone Java demo application . Take GPars with you wherever you go!8.1. Performance
Your code in Groovy can be just as fast as code written in Java, Scala or any other programing language. This should not be surprising, since GPars is technically a solid tasty Java-made cake with a Groovy DSL cream on it.Unlike in Java, however, with GPars, as well as with other DSL-friendly languages, you are very likely to experience a useful kind of code speed-up for free, a speed-up coming from a better and cleaner design of your application. Coding with a concurrency DSL will give you smaller code-base with code using the concurrency primitives as language constructs. So it is much easier to build robust concurrent applications, identify potential bottle-necks or errors and eliminate them.While this whole User Guide is describing how to use Groovy and GPars to create beautiful and robust concurrent code, let's use this chapter to highlight a few places, where some code tuning or minor design compromises could give you interesting performance gains.Parallel Collections
Methods for parallel collection processing, like eachParallel() , collectParallel() and such use Parallel Array , an efficient tree-like data structure behind the scenes. This data structure has to be built from the original collection each time you call any of the parallel collection methods. Thus when chaining parallel method calls you might consider using the map/reduce API instead or resort to using the ParallelArray API directly, to avoid the Parallel Array creation overhead.GParsPool.withPool {
people.findAllParallel{it.isMale()}.collectParallel{it.name}.any{it == 'Joe'}
people.parallel.filter{it.isMale()}.map{it.name}.filter{it == 'Joe'}.size() > 0
people.parallelArray.withFilter({it.isMale()} as Predicate).withMapping({it.name} as Mapper).any{it == 'Joe'} != null
}
GParsPool.withPool(50) { … }
Actors
GPars actors are fast. DynamicDispatchActors and ReactiveActors are about twice as fast as the DefaultActors , since they don't have to maintain an implicit state between subsequent message arrivals. The DefaultActors are in fact on par in performance with actors in Scala , which you can hardly hear of as being slow.If top performance is what you're looking for, a good start is to identify the following patterns in your actor code:actor { loop { react {msg -> switch(msg) { case String:… case Integer:… } } } }
messageHandler { when{String msg -> ...} when{Integer msg -> ...} }
class MyHandler extends DynamicDispatchActor { public void handleMessage(String msg) { … } public void handleMessage(Integer msg) { … } }
Pool adjustment
GPars allows you to group actors around thread pools, giving you the freedom to organize actors any way you like. It is always worthwhile to experiment with the actor pool size and type. FJPool usually gives better characteristics that DefaultPool , but seems to be more sensitive to the number of threads in the pool. Sometimes using a ResizeablePool or ResizeableFJPool could help performance by automatic eliminating unneeded threads.def attackerGroup = new DefaultPGroup(new ResizeableFJPool(10)) def defenderGroup = new DefaultPGroup(new DefaultPool(5))def attacker = attackerGroup.actor {...} def defender = defenderGroup.messageHandler {...} ...