groovyx.gpars
[Groovy] Class GParsPool

java.lang.Object
  groovyx.gpars.GParsPool

class GParsPool
extends java.lang.Object

Enables a ParallelArray-based (from JSR-166y) DSL on collections. In general cases the Parallel Array implementation shows to be much faster (10 - 20 times) compared to the executor service implementation in GParsExecutorsPool. E.g.

 GParsPool.withPool(5) {final AtomicInteger result = new AtomicInteger(0)
 [1, 2, 3, 4, 5].eachParallel {result.addAndGet(it)}assertEquals 15, result}GParsPool.withPool(5) {final List result = [1, 2, 3, 4, 5].collectParallel {it * 2}assert ([2, 4, 6, 8, 10].equals(result))}GParsPool.withPool(5) {assert [1, 2, 3, 4, 5].everyParallel {it > 0}assert ![1, 2, 3, 4, 5].everyParallel {it > 1}}

Authors:

Vaclav Pech Date: Oct 23, 2008

currentPoolStack

private static final ThreadLocalPools currentPoolStack

Maps threads to their appropriate thread pools

defaultPoolSize

private static final int defaultPoolSize

Caches the default pool size.

GParsPool

GParsPool()

createDefaultUncaughtExceptionHandler

private static UncaughtExceptionHandler createDefaultUncaughtExceptionHandler()

createPool

private static java.lang.Object createPool()

Creates a new pool with the default size()

createPool

private static java.lang.Object createPool(int poolSize)

Creates a new pool with the given size()

createPool

private static java.lang.Object createPool(int poolSize, UncaughtExceptionHandler handler)

ensurePool

static java.lang.Object ensurePool(java.lang.Object pool, groovy.lang.Closure cl)

Just like withExistingPool() registers a thread pool, but doesn't install the GParsPoolUtil category. Used by ParallelEnhancer's Parallel mixins.

executeAsync

static java.util.List executeAsync(groovy.lang.Closure closures)

Starts multiple closures in separate threads, collecting Futures for their return values Reuses the pool defined by the surrounding withPool() call. If an exception is thrown from the closure when called on any of the collection's elements, it will be re-thrown in the calling thread when it calls the Future.get() method.

Returns:

Futures for the result values or exceptions of all closures

executeAsync

static java.util.List executeAsync(java.util.List closures)

Starts multiple closures in separate threads, collecting Futures for their return values Reuses the pool defined by the surrounding withPool() call. If an exception is thrown from the closure when called on any of the collection's elements, it will be re-thrown in the calling thread when it calls the Future.get() method.

Returns:

Futures for the result values or exceptions of all closures

executeAsyncAndWait

static java.util.List executeAsyncAndWait(groovy.lang.Closure closures)

Starts multiple closures in separate threads, collecting their return values Reuses the pool defined by the surrounding withPool() call. If an exception is thrown from the closure when called on any of the collection's elements, it will be re-thrown in the calling thread when it calls the Future.get() method.

throws:

AsyncException If any of the collection's elements causes the closure to throw an exception. The original exceptions will be stored in the AsyncException's concurrentExceptions field.

Returns:

The result values of all closures

executeAsyncAndWait

static java.util.List executeAsyncAndWait(java.util.List closures)

Starts multiple closures in separate threads, collecting their return values Reuses the pool defined by the surrounding withPool() call. If an exception is thrown from the closure when called on any of the collection's elements, it will be re-thrown in the calling thread when it calls the Future.get() method.

throws:

AsyncException If any of the collection's elements causes the closure to throw an exception. The original exceptions will be stored in the AsyncException's concurrentExceptions field.

Returns:

The result values of all closures

retrieveCurrentPool

protected static java.lang.Object retrieveCurrentPool()

Retrieves the pool assigned to the current thread.

runForkJoin

static T runForkJoin(AbstractForkJoinWorker rootWorker)

Starts a ForkJoin calculation with the supplied root worker and waits for the result.

Parameters:

rootWorker - The worker that calculates the root of the Fork/Join problem

Returns:

The result of the whole calculation

runForkJoin

static T runForkJoin(java.lang.Object args)

Starts a ForkJoin calculation with the supplied root worker and waits for the result.

Parameters:

rootWorker - The worker that calculates the root of the Fork/Join problem

Returns:

The result of the whole calculation

speculate

static java.lang.Object speculate(java.util.List alternatives)

Runs the supplied closures asynchronously and in parallel, returning the first result obtained and cancelling the other (slower) calculations. Typically used to run several different calculations in parallel, all of which are supposed to give the same result, but may last different amount of time each. If the system has enough threads available, the calculations can be test-run in parallel and the fastest result is then used, while the other results are cancelled or discarded.

Parameters:

alternatives - All the functions to invoke in parallel

Returns:

The fastest result obtained

speculate

static java.lang.Object speculate(groovy.lang.Closure alternatives)

Runs the supplied closures asynchronously and in parallel, returning the first result obtained and cancelling the other (slower) calculations. Typically used to run several different calculations in parallel, all of which are supposed to give the same result, but may last different amount of time each. If the system has enough threads available, the calculations can be test-run in parallel and the fastest result is then used, while the other results are cancelled or discarded.

Parameters:

alternatives - All the functions to invoke in parallel

Returns:

The fastest result obtained

withExistingPool

static java.lang.Object withExistingPool(java.lang.Object pool, groovy.lang.Closure cl)

Reuses an instance of ForkJoinPool, binds it to the current thread, enables the ParallelArray DSL and runs the supplied closure. Within the supplied code block the ForkJoinPool is available as the only parameter, collections have been enhanced with the eachParallel(), collectParallel() and other methods from the GParsPoolUtil category class. E.g. calling images.eachParallel{processImage(it}} will call the potentially long-lasting processImage() operation on each image in the images collection in parallel. Be sure to synchronize all modifiable state shared by the asynchronously running closures.

 GParsPool.withExistingPool(anotherPool) {GParsPool pool ->
     def result = Collections.synchronizedSet(new HashSet())
     [1, 2, 3, 4, 5].eachParallel {Number number -> result.add(number * 10)}*     assertEquals(new HashSet([10, 20, 30, 40, 50]), result)
}*  

Parameters:

pool - The thread pool to use, the pool will not be shutdown after this method returns

withPool

static java.lang.Object withPool(groovy.lang.Closure cl)

Creates a new instance of ForkJoinPool, binds it to the current thread, enables the ParallelArray DSL and runs the supplied closure. Within the supplied code block the ForkJoinPool is available as the only parameter, collections have been enhanced with the eachParallel(), collectParallel() and other methods from the GParsPoolUtil category class. E.g. calling images.eachParallel{processImage(it}} will call the potentially long-lasting processImage() operation on each image in the images collection in parallel. Be sure to synchronize all modifiable state shared by the asynchronously running closures.

 GParsPool.withPool {GParsPool pool ->
     def result = Collections.synchronizedSet(new HashSet())
     [1, 2, 3, 4, 5].eachParallel {Number number -> result.add(number * 10)}*     assertEquals(new HashSet([10, 20, 30, 40, 50]), result)
}* 

Parameters:

cl - The block of code to invoke with the DSL enabled

withPool

static java.lang.Object withPool(int numberOfThreads, groovy.lang.Closure cl)

Creates a new instance of ForkJoinPool, binds it to the current thread, enables the ParallelArray DSL and runs the supplied closure. Within the supplied code block the ForkJoinPool is available as the only parameter, collections have been enhanced with the eachParallel(), collectParallel() and other methods from the GParsPoolUtil category class. E.g. calling images.eachParallel{processImage(it}} will call the potentially long-lasting processImage() operation on each image in the images collection in parallel. Be sure to synchronize all modifiable state shared by the asynchronously running closures.

 GParsPool.withPool(5) {GParsPool pool ->
     def result = Collections.synchronizedSet(new HashSet())
     [1, 2, 3, 4, 5].eachParallel {Number number -> result.add(number * 10)}*     assertEquals(new HashSet([10, 20, 30, 40, 50]), result)
}* 

Parameters:

numberOfThreads - Number of threads in the newly created thread pool

cl - The block of code to invoke with the DSL enabled

withPool

static java.lang.Object withPool(int numberOfThreads, UncaughtExceptionHandler handler, groovy.lang.Closure cl)

Creates a new instance of ForkJoinPool, binds it to the current thread, enables the ParallelArray DSL and runs the supplied closure. Within the supplied code block the ForkJoinPool is available as the only parameter, collections have been enhanced with the eachParallel(), collectParallel() and other methods from the GParsPoolUtil category class. E.g. calling images.eachParallel{processImage(it}} will call the potentially long-lasting processImage() operation on each image in the images collection in parallel. Be sure to synchronize all modifiable state shared by the asynchronously running closures.

 GParsPool.withPool(5, handler) {GParsPool pool ->
     def result = Collections.synchronizedSet(new HashSet())
     [1, 2, 3, 4, 5].eachParallel {Number number -> result.add(number * 10)}*     assertEquals(new HashSet([10, 20, 30, 40, 50]), result)
}* 

Parameters:

numberOfThreads - Number of threads in the newly created thread pool

handler - Handler for uncaught exceptions raised in code performed by the pooled threads

cl - The block of code to invoke with the DSL enabled

Copyright © 2008–2010 Václav Pech. All Rights Reserved.