Java 102 : Do we really need Executor Service Abstraction ? Spoiler alert, Yes we do!

In the previous article, we ran into the shortcomings of the simple runnable thread abstraction for implementing multithreaded code in java.

We will explore the executor service abstraction from the concurrent package and see if it really solves our problems.

Just for a reminder! The problem at hand was computing the sum of each row of a given 2D list and returning it as a 1D list.

Input : [[1,2,3],             
         [4,5,6],
         [7,8,9]]
Output : [6,15,24]

let us try and solve this problem using the executor service and Runnable abstraction first, lets's see what our solution would look like when implementing the solution using these abstractions instead of bare thread and runnable abstractions

List<Integer> getEachRowSum(List<List<Integer>> input) {
    
    List<Integer> result = new ArrayList   
                          (Collections.nCopies(input.size(), 0));
    ExecutorService executor = Executors.newFixedThreadPool(10);
    for(Integer i = 0; i<input.size(); i++) {
        final Integer index = i;
        Runnable task = new Runnable() { 
            public void run() { 
                Integer rowSum = 0;
                for(Integer element: input.get(index)) {
                    rowSum+= element;
                }
                result.set(index, rowSum);
            }
        };
        // Submit means the task has been submitted,
        // to the worker threads inside thread pool for execution 
        executor.submit(task);
    }
        
    try {
        // Stop accepting any more tasks
        executor.shutdown();
        while (!executor.awaitTermination(24L, TimeUnit.HOURS)){
            // waiting for all tasks to complete
        }
    } catch(InterruptedException e) {
        // no nothing.
    }
    return result;
}

The above solution solves the problem we had of getting the row-wise sum for any particular list passed in, and we solve some issues with our previous implementation that we discussed in the last article like thrashing due to spawning a large number of threads …

But there are some issues which we couldn't solve
1. executor.awaitTermination(24L, TimeUnit.HOURS)we are using this condition just to make sure that the execution finishes on time, really not elegant, with no insight on if the task was completed or failed!
2. What if we want to return some value from the executing runnable or propagate a checked exception to the caller? it returns nothing 😓.
Instead, we should use the callable in place of the runnable interface.

what is Callable Interface?
1. The Callable interface is a functional interface similar to Runnable, but returns a value, ExecutorService takes in both Callable as well as Runnable tasks for execution.
2. It was designed to overcome the shortcomings of the Runnable interface, like not being able to return any value or propagate checked exceptions to the caller.

public interface Callable<V> {
     V call() throws Exception;

Now let's try and use this Callable interface to re-write our solution again.
In the first implementation using a runnable Interface, we were ignoring the future object that was being returned to us by the submit() function, since the future objects for runnable tasks do not contain any value (null).

// Not storing the future object 
executor.submit(task);

But for the callable tasks submitted to the executor, it returns a Future<T> object which contains the result of the task, or the exception wrapped inside the ExecutionException object if the task did not complete due to an exception.

What is the Future Interface now 😒?
1. The Future interface represents a result that will eventually be returned in the future. We can check if a future has been fed the result, if it's awaiting a result or if it has failed before we try to access it.
2. The executorService returns a result wrapped in the future interface for every task (it can be runnable or a callable) passed in as a task to the executor using the submit() function.
3. The result can only be retrieved using method get() when the computation has completed, blocking if necessary until it is ready. Cancellation is performed by the cancel() method.

Let's see how the same implementation would look like if we used the callable interface instead of runnable as tasks for the executorService.

public List<Integer> getEachRowSum(List<List<Integer>> input) {    
        List<Integer> result = new ArrayList<>    
                             (Collections.nCopies(input.size(), 0));
        // setting up the executor service
        ExecutorService executor = Executors.newFixedThreadPool(10);
       // setting up the results list
        List<Future<Integer>> futureResults = new ArrayList<>();
            
        for(Integer i = 0; i<input.size(); i++) {
            final Integer index = i;
            Callable<Integer> task = new Callable<Integer>() { 
                public Integer call() throws Exception { 
                    Integer rowSum = 0;
                    for(Integer element: input.get(index)) {
                        rowSum+= element;
                    }
                    return rowSum;
                }
            };
            // keeping track of the future object  
            futureResults.add(executor.submit(task));
        }
        
        try {
            // Getting the results after 
            // The get() call is a blocking call 
            for(int i=0;i<futureResults.size();i++) {
                while(!futureResults.get(i).isDone()) 
                     // print(the task is in progress)
                
                // The task execution is finished.  
                result.set(i, (futureResults.get(i).get()));
            }
            
            // Shutdown the executor release resources
            executor.shutdown();
            
        } catch(InterruptedException|ExecutionException e) {
            // do nothing
        }
        return result;
    }

This is almost all the essentials needed to be discussed for simple multi-threaded code in java, this looks elegant and works for the problem we have.

In addition, to all the problems we had in the previous implementation it also solves a lot of other problems like canceling the tasks using some timeout, returning the value if it gets computed within a specified time, and much more. We can dive deep into all this functionality in the official docs.

But there is another popular abstraction that is used to write multithreaded code in java inducted in Java 7, The ForkJoinPool ? what is that all about?
will write about it soon : /**WIP*/

leave me a clap if you found this article insightful 😄, Happy Reading !