Fun with Guava's ListeningExecutorService

As a follow up to my previous blog post, I decided to rewrite the code sample using the more advance Google Guava ListeningExecutorService and ListenableFuture API, so here it is:

 import com.google.common.util.concurrent.Futures;  
 import com.google.common.util.concurrent.ListenableFuture;  
 import com.google.common.util.concurrent.ListeningExecutorService;  
 import com.google.common.util.concurrent.MoreExecutors;  
 import java.util.ArrayList;  
 import java.util.List;  
 import java.util.Random;  
 import java.util.Scanner;  
 import java.util.concurrent.Callable;  
 import java.util.concurrent.ExecutionException;  
 import java.util.concurrent.Executors;  

 class ListenableFutureExample {  
   public static void main(String[] args) {  
     Scanner in = new Scanner(System.in);  
     final int nThreads = in.nextInt();  
     final int n = in.nextInt();  
     System.out.println("Using " + nThreads + " threads");  
     ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(nThreads));  
     try {  
       try {  
         testSomeWorkers(service, n);  
       } catch (InterruptedException | ExecutionException e) {  
         e.printStackTrace();  
       }  
       try {  
         testJobCanceling(service, n);  
       } catch (InterruptedException | ExecutionException e) {  
         e.printStackTrace();  
       }  
     } finally {  
       // necessary or the thread pool will keep the JVM up and running!  
       service.shutdown();  
     }  
   }  

   public static void testSomeWorkers(ListeningExecutorService service, int n) throws InterruptedException, ExecutionException {  
     // using the Guava's utility method allAsList will return all the results in a future list  
     // enormously simplifying the code:  
     ListenableFuture<List<Integer>> ret = Futures.successfulAsList(addSomeTasks(service, n));  
     // the call to get() is now the blocking piece of code  
     System.out.println("Values returned from computations: " + ret.get());  
     System.out.println("All done.");  
   }  

   public static void testJobCanceling(ListeningExecutorService service, int n) throws InterruptedException, ExecutionException {  
     List<ListenableFuture<Integer>> tasks = addSomeTasks(service, n);  
     ListenableFuture<List<Integer>> ret = Futures.allAsList(tasks);  
     Thread.sleep(1000);  
     System.out.println("Actually nevermind!");  
     ret.cancel(true);  
     // let's see how many tasks were actually cancelled by asking the original futures:  
     List<Integer> completed = new ArrayList<>();  
     for (ListenableFuture<Integer> f : tasks) if (!f.isCancelled()) completed.add(f.get());  
     System.out.println("There were " + (n - completed.size()) + " cancelled tasks and " + completed.size() + " completed tasks: " + completed);  
     System.out.println("All done.");  
   }  

   private static List<ListenableFuture<Integer>> addSomeTasks(ListeningExecutorService service, int howMany) {  
     System.out.println("Enqueuing " + howMany + " tasks...");  
     List<ListenableFuture<Integer>> ret = new ArrayList<>();  
     for (int i = 1; i <= howMany; i++) {  
       final int n = i;  
       ret.add(service.submit(new Callable<Integer>() {  
         @Override  
         public Integer call() {  
           try {  
             try {  
               System.out.println("Task " + n + ": Doing some very important work...");  
               Thread.sleep(200 + rnd.nextInt(200));  
             } catch (InterruptedException e) {  
               System.out.println("Task " + n + " interrupted while doing very important work");  
               return null;  
             }  
             try {  
               System.out.println("Task " + n + ": Doing more important work...");  
               Thread.sleep(200 + rnd.nextInt(200));  
             } catch (InterruptedException e) {  
               System.out.println("Task " + n + " interrupted while doing important work");  
               return null;  
             }  
             try {  
               System.out.println("Task " + n + ": Doing slightly less important work...");  
               Thread.sleep(200 + rnd.nextInt(200));  
             } catch (InterruptedException e) {  
               System.out.println("Task " + n + " interrupted while doing slightly less important work");  
               return null;  
             }  
             int ret = rnd.nextInt();  
             System.out.println("Task " + n + ": about to return " + ret);  
             return ret;  
           } finally {  
             System.out.println("Task " + n + ": cleaning up");  
           }  
         }  
       }));  
     }  
     return ret;  
   }  
   private final static Random rnd = new Random();  
 }  

The Guava API has the advantage to allow one to register a callback to a ListenableFuture and apply transformations to futures, resulting in a chain of non-blocking operations, very much like Scala's Futures. Non-blocking concurrency will greatly limit resource usage, if used properly since there will be no idle threads blocking while waiting on the results of other threads' computations.

Just for illustration purposes, an example of using Guava's future transformation capabilities to implement fully non-blocking asynchronous computations follows. Enjoy!

 import com.google.common.base.Function;  
 import com.google.common.base.Optional;  
 import com.google.common.util.concurrent.Futures;  
 import com.google.common.util.concurrent.ListenableFuture;  
 import com.google.common.util.concurrent.ListeningExecutorService;  
 import com.google.common.util.concurrent.MoreExecutors;  
 import java.math.BigInteger;  
 import java.util.ArrayList;  
 import java.util.List;  
 import java.util.Random;  
 import java.util.concurrent.Callable;  
 import java.util.concurrent.ExecutionException;  
 import java.util.concurrent.Executors;  

 class ListenableFutureChain {  
   public static void main(String[] args) {  
     ListenableFutureChain chain = new ListenableFutureChain(4);  
     // let's find a few 512 bit prime numbers for our awesome encryption algorithm!  
     ListenableFuture<List<BigInteger>> probablePrimes = chain.findSomePrimeNumbers(20, 512);  
     // now finally do something with the future prime list  
     try {  
       // WARNING: this call is blocking, for illustration purposes only.  
       // It's recommended to design so that you don't need to do this,  
       // as in the function findSomePrimeNumbers()  
       for (BigInteger i : probablePrimes.get()) System.out.println(i);  
     } catch (InterruptedException | ExecutionException e) {  
       e.printStackTrace();  
     }  
     // remember to call this or the executor service will keep the JVM "awake"!  
     chain.dispose();  
   }  

   public ListenableFutureChain(final int nThreads) {  
     executorService = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(nThreads));  
   }  

   public void dispose() {  
     executorService.shutdown();  
   }  

   private final ListeningExecutorService executorService;  
   private final Random random = new Random();  

   public ListenableFuture<List<BigInteger>> findSomePrimeNumbers(final int nAttempts, final int nBits) {  
     List<ListenableFuture<Optional<BigInteger>>> probablePrimes = new ArrayList<>(nAttempts);  
     for (int i = 0; i < nAttempts; i++) {  
       // submit a task for execution and retrieve the ListenableFuture  
       ListenableFuture<BigInteger> probablePrimeFuture = executorService.submit(new Callable<BigInteger>() {  
         @Override  
         public BigInteger call() throws Exception {  
           // I'm going to find a probable prime number  
           return BigInteger.probablePrime(nBits, random);  
         }  
       });  
       // transform the previous ListenableFuture using a function; returns another ListenableFuture (non blocking operation)  
       ListenableFuture<Optional<BigInteger>> primeOrNot = Futures.transform(probablePrimeFuture, new Function<BigInteger, Optional<BigInteger>>() {  
         @Override  
         public Optional<BigInteger> apply(BigInteger p) {  
           // I'm going to return only the probable primes that are actually prime  
           if (isPrime(p)) return Optional.of(p);  
           return Optional.absent();  
         }  
       }, executorService);  
       // add the second future to a list  
       probablePrimes.add(primeOrNot);  
     }  
     // transform the list of futures to a future of list (Guava magic!), only retain successful futures (again, non blocking!)  
     ListenableFuture<List<Optional<BigInteger>>> primesOrNoValues = Futures.successfulAsList(probablePrimes);  
     // transform the future list to a future list containing only the prime numbers in question and return this future (still non blocking)  
     return Futures.transform(primesOrNoValues, new Function<List<Optional<BigInteger>>, List<BigInteger>>() {  
       @Override  
       public List<BigInteger> apply(List<Optional<BigInteger>> primes) {  
         List<BigInteger> ret = new ArrayList<>(primes.size());  
         //  
         for (Optional<BigInteger> optional : primes)  
           if (optional != null && optional.isPresent()) ret.add(optional.get());  
         return ret;  
       }  
     }, executorService);  
     // Note that this whole function is non blocking; you can tell by the fact that there's no InterruptedException being thrown anywhere.  
   }  

   private boolean isPrime(final BigInteger p) {  
     // TODO: do some fancy primality test! (note that this would take a while in real life)  
     // let's just return true or false randomly for now... ;)  
     return random.nextBoolean();  
   }  
 }  

Fun with java's ExecutorService

It's surprisingly difficult to find a decent example of the java Executor framework that explains some of the API's "gotchas". It took me a while way back then when I was trying to figure it out, so I decided to post a hopefully useful example. In the code below, I use an executor service to execute some work in parallel. This is the easiest way to exploit simple parallelism in java 7 (java 8 introduces parallel streams which make things even simpler). The executor service is created as a fixed thread pool using a convenient static factory method available in the Executors class. It's very instructive to look at the javadoc for this and other static factory methods available in this class, and even the source code if you are feeling adventurous, to understand how the thread pools are created and what additional options are available to customize them.

The testSomeWorkers() method creates a few tasks (instances of Callable), invokes them using the executor service and retrieves their results using the corresponding Futures. The helper method addSomeTasks() simply calls the submit() method on the executor service to submit the n tasks and retrieve the corresponding future. The future will return the value returned by the Callable once the execution is over.

The testJobCanceling() method works pretty much like testSomeWorkers() but it shows how the tasks can be cancelled by using the Future.cancel() method. The cancel() method's only argument specifies whether or not the thread in which the job is running should be interrupted. In the Callable.call() function, invoking Thread.sleep() can potentially throw an InterruptedException. This will happen when we call cancel() on the future associated with this callable and thus the exception is handled by returning immediately. It's a common idiom in java to handle interruptions in this fashion to achieve some control over concurrent workers.

Enjoy the code:

 import java.util.*;  
 import java.util.concurrent.*;  
 class ExecutorServiceExample {  
   public static void main(String[] args) {  
     Scanner in = new Scanner(System.in);  
     final int nThreads = in.nextInt();  
     final int n = in.nextInt();  
     System.out.println("Using " + nThreads + " threads");  
     ExecutorService service = Executors.newFixedThreadPool(nThreads);  
     try {  
       testSomeWorkers(service, n);  
     } catch (InterruptedException | ExecutionException e) {  
       e.printStackTrace();  
     }  
     try {  
       testJobCanceling(service, n);  
     } catch (InterruptedException | ExecutionException e) {  
       e.printStackTrace();  
     }  
     // necessary or the thread pool will keep the JVM up and running!  
     service.shutdown();  
   }  
   public static void testSomeWorkers(ExecutorService service, int n) throws InterruptedException, ExecutionException {  
     // create and invoke some "tasks" on this executor service  
     Collection<Future<Integer>> taskFutures = addSomeTasks(service, n);  
     System.out.println("Waiting for all tasks to complete...");  
     List<Integer> ret = new ArrayList<>();  
     // retrieve the result of the tasks' computation  
     for (Future<Integer> f : taskFutures) ret.add(f.get());  
     System.out.println("Values returned from computations: " + ret);  
     System.out.println("All done.");  
   }  
   public static void testJobCanceling(ExecutorService service, int n) throws InterruptedException, ExecutionException {  
     Collection<Future<Integer>> taskFutures = addSomeTasks(service, n);  
     Thread.sleep(1000);  
     System.out.println("Actually nevermind!");  
     List<Integer> completed = new ArrayList<>();  
     List<Future<Integer>> cancelled = new ArrayList<>();  
     // try to cancel the tasks that are running  
     for (Future<Integer> f : taskFutures) {  
       // if successfully cancel add to the cancelled list  
       if (f.cancel(true)) cancelled.add(f);  
         // otherwise get the result  
       else completed.add(f.get());  
     }  
     System.out.println("" + cancelled.size() + " tasks were successfully cancelled");  
     if (!completed.isEmpty()) System.out.println("Values returned from computations: " + completed);  
     System.out.println("All done.");  
   }  
   private static Collection<Future<Integer>> addSomeTasks(ExecutorService service, int howMany) {  
     System.out.println("Enqueuing " + howMany + " tasks...");  
     List<Future<Integer>> ret = new ArrayList<>();  
     for (int i = 0; i < howMany; i++) {  
       final int n = i;  
       ret.add(service.submit(new Callable<Integer>() {  
         @Override  
         public Integer call() {  
           try {  
             try {  
               System.out.println("Task " + n + ": Doing some very important work...");  
               Thread.sleep(200 + rnd.nextInt(200));  
             } catch (InterruptedException e) {  
               System.out.println("Task " + n + " interrupted while doing very important work");  
               return null;  
             }  
             try {  
               System.out.println("Task " + n + ": Doing more important work...");  
               Thread.sleep(200 + rnd.nextInt(200));  
             } catch (InterruptedException e) {  
               System.out.println("Task " + n + " interrupted while doing important work");  
               return null;  
             }  
             try {  
               System.out.println("Task " + n + ": Doing slightly less important work...");  
               Thread.sleep(200 + rnd.nextInt(200));  
             } catch (InterruptedException e) {  
               System.out.println("Task " + n + " interrupted while doing slightly less important work");  
               return null;  
             }  
             return rnd.nextInt();  
           } finally {  
             System.out.println("Cleaning up after task " + n);  
           }  
         }  
       }));  
     }  
     return ret;  
   }  
   private final static Random rnd = new Random();  
 }