程式扎記: [ Groovy Doc ] Cookbook Ex - Concurrency with Groovy

標籤

2015年1月14日 星期三

[ Groovy Doc ] Cookbook Ex - Concurrency with Groovy

Source From Here 
Using threads and AtomicInteger 
From Groovy you can use all of the normal concurrency facilities in Java and combine them with threads and closures as necessary. E.g. a (slightly modified) atomic counter from a Groovy example here
  1. import java.util.concurrent.atomic.AtomicInteger  
  2.   
  3. def counter = new AtomicInteger()  
  4.   
  5. synchronized out(message) {  
  6.     println(message)  
  7. }  
  8.   
  9. def th = Thread.start {  
  10.     for( i in 1..8 ) {  
  11.         sleep 30  
  12.         out "thread loop $i"  
  13.         counter.incrementAndGet()  
  14.     }  
  15. }  
  16.   
  17. for( j in 1..4 ) {  
  18.     sleep 50  
  19.     out "main loop $j"  
  20.     counter.incrementAndGet()  
  21. }  
  22.   
  23. th.join()  
  24.   
  25. assert counter.get() == 12  
Fibonacci with Executors 
A Groovy script to naively calculate the Fibonacci series inspired by the example here. Note: a version using memoizing will be much more efficient but this illustrates using Java's built-in concurrency primitives from Groovy. 
  1. import java.util.concurrent.*  
  2.   
  3. CUTOFF = 12    // not worth parallelizing for small n  
  4. THREADS = 100  
  5.   
  6. println "Calculating Fibonacci sequence in parallel..."  
  7. serialFib = { n -> (n < 2) ? n : serialFib(n-1) + serialFib(n-2) }  
  8. pool = Executors.newFixedThreadPool(THREADS)  
  9. defer = { c -> pool.submit(c as Callable) }  
  10. fib = { n ->  
  11.   if (n < CUTOFF) return serialFib(n)  
  12.   def left = defer{ fib(n-1) }  
  13.   def right = defer{ fib(n-2) }  
  14.   left.get() + right.get()  
  15. }  
  16.   
  17. (8..16).each{ n -> println "n=$n => ${fib(n)}" }  
  18. pool.shutdown()  
Which produces this: 
Calculating Fibonacci sequence in parallel...
n=8 => 21
n=9 => 34
n=10 => 55
n=11 => 89
n=12 => 144
n=13 => 233
n=14 => 377
n=15 => 610
n=16 => 987

If you want to convince yourself that some threads are actually being created, replace the last each line with: 
  1. showThreads = { println Thread.allStackTraces.keySet().join('\n') }  
  2. (8..16).each{ n -> if (n == 14) showThreads(); println "n=$n => ${fib(n)}" }  
which will add something like the following to your output: 
Thread[main,5,main]
Thread[pool-1-thread-1,5,main]
Thread[Signal Dispatcher,9,system]
Thread[Attach Listener,5,system]
Thread[pool-1-thread-5,5,main]
Thread[Reference Handler,10,system]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-2,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-6,5,main]
Thread[Finalizer,8,system]

Fibonacci with Functional Java 
If you wish to make use of the functionaljava (tested with 2.1.3) library , you could use a Groovy program such as this: 
  1. import fj.*  
  2. import fj.control.parallel.Strategy  
  3. import static fj.Function.curry as fcurry  
  4. import static fj.P1.curry as pcurry  
  5. import static fj.P1.fmap  
  6. import static fj.control.parallel.Actor.actor  
  7. import static fj.control.parallel.Promise.*  
  8. import static fj.data.List.range  
  9. import static java.util.concurrent.Executors.*  
  10.   
  11. CUTOFF  = 12   // not worth parallelizing for small n  
  12. START   = 8  
  13. END     = 16  
  14. THREADS = 4  
  15.   
  16. pool = newFixedThreadPool(THREADS)  
  17. su   = Strategy.executorStrategy(pool)  
  18. spi  = Strategy.executorStrategy(pool)  
  19. add  = fcurry({ a, b -> a + b } as F2)  
  20. nums = range(START, END + 1)  
  21.   
  22. println "Calculating Fibonacci sequence in parallel..."  
  23.   
  24. serialFib = { n -> n < 2 ? n : serialFib(n - 1) + serialFib(n - 2) }  
  25.   
  26. print = { results ->  
  27.   def n = START  
  28.   results.each{ println "n=${n++} => $it" }  
  29.   pool.shutdown()  
  30. } as Effect  
  31.   
  32. calc = { n ->  
  33.   n < CUTOFF ?  
  34.     promise(su, P.p(serialFib(n))) :  
  35.     calc.f(n - 1).bind(join(su, pcurry(calc).f(n - 2)), add)  
  36. } as F  
  37.   
  38. out = actor(su, print)  
  39. join(su, fmap(sequence(su)).f(spi.parMapList(calc).f(nums))).to(out)  
Using java.util.concurrent.Exchanger 
An example using Exchanger. We have two threads - one keeping evens and exchanging odd values with the other thread. Meanwhile the other thread is working in reverse fashion; keeping the odds and exchanging the evens. The algorithm here is dumb in that it relies on the same number of swaps for each side - which is fine for this example but would need to be altered for more general input values. 
  1. def x = new java.util.concurrent.Exchanger()  
  2. def first=1..20, second=21..40, evens, odds  
  3. def t1 = Thread.start{ odds = first.collect{ it % 2 != 0 ? it : x.exchange(it) } }  
  4. def t2 = Thread.start{ evens = second.collect{ it % 2 == 0 ? it : x.exchange(it) } }  
  5. [t1, t2]*.join()  
  6. println "evens: $evens"  
  7. println "odds: $odds"  
Which has the following output: 
evens: [2, 22, 4, 24, 6, 26, 8, 28, 10, 30, 12, 32, 14, 34, 16, 36, 18, 38, 20, 40]
odds: [1, 21, 3, 23, 5, 25, 7, 27, 9, 29, 11, 31, 13, 33, 15, 35, 17, 37, 19, 39]

Catching Exceptions with an Exception Handler 
When catching Exceptions with an exception handler in Groovy Scripts, there is potential for interaction with Groovy's runtime which catches exceptions and filters stacktraces. The best way to avoid this interaction is to create your own thread and call setDefaultUncaughtExceptionHandler directly on that thread instance as per below: 
  1. def th = Thread.start {  
  2.     println 'start'  
  3.     println Thread.currentThread()  
  4.     sleep 1000  
  5.     throw new NullPointerException()  
  6. }  
  7. th.setDefaultUncaughtExceptionHandler({t,ex ->  
  8.     println 'ignoring: ' + ex.class.name  
  9. } as Thread.UncaughtExceptionHandler)  
  10. th.join()  
More Information 
See also: 
Functional Programming with Groovy 
Multithreading with SwingBuilder

沒有留言:

張貼留言

網誌存檔

關於我自己

我的相片
Where there is a will, there is a way!