Building next generation of Concurrent Systems using Java’s CompletionService — Support for Callbacks and Event Driven Behavior
Concurrency is one of the hard topics in computer science. Building concurrent systems in real time is always a challenging task. I find great inspiration and genuine happiness while dealing with such kind of challenges and complexities. Perhaps, one reason being building these systems always test my skills and feeds my constant urge of learn and apply.
This is also one of those situations where the system designer have to balance quadrants of speed with high quality and reliability of system, all at same time. Balancing equations of this nature is a daunting task and success is not only matter of luck but also requires attention to details, of-course experience, and most important a desire to craft genuinely good software. The nature of problem requires dive deep at the lowest level possible ruling out all assumptions and filling knowledge gaps. Software Engineers dealing with such complexities needs to master the depth and breadth of situation as well as solution.
One of the many situations I came across while designing concurrent systems is when a blocked Executor needs to process variety of independent Tasks, and it simply cannot provides visibility in processing until all tasks are completed, either with success or failures. This is a well adapted design of concurrent components but from my perspective, this design can be further improved. The overall speed of this design entirely depends on the completion of slowest task. Even if the other fast running tasks completed quickly, the overall system is unable to process the results until the slowest one completes. At the very end Executor generally processes aggregated results received from all tasks.
However, Java’s CompletionService have taken a different approach to build concurrent systems of this nature with better capabilities. As always, the designers have adopted a very powerful decoupled design and clever use of data structures. On a high level the process starts like any other ExecutorService. Blocking Producer submit tasks and waits for entire process to complete. But at the same time Executor can consume arriving results by invoking take() without compromising the current process execution. Under the hood, CompletionService maintains a Queue as its data structure to keeps track of results of completed tasks, as they arrive. This internal Queue can be accessed by invoking take() at any given moment. Moreover the use of Queue helps maintains the order of results by their arrival order. The only caveat being once a particular result is read from Queue, the Producer is responsible to save those results. This caveat can be easily controlled in multiple ways.
The biggest advantage is no matter how much time the longest running task takes, the results of other faster completed tasks are available for consumption. This idea opens up possibilities to take concurrent systems to next level. In my opinion, this approach is elegantly providing a way to bake callback and Event Driven nature in ExecutorService based concurrent designs.
In order to learn and explore more about this capability, I tried to solved the following problem. Let’s quickly go over the details of Problem Statement we are going to solve.
Executor is a blocked Service whose job is to submit tasks and process results. i.e. Executor waits for all submitted tasks to complete processing.
Executor submits 3 independent Tasks all at same time ~ T(0)
Task1 completes at time = T(1), the fastest Task
Task2 completes at time=T(5) making it the longest running task
Task3 completes at time=T(3), the second fastest task
Timing of tasks is T(1)<T(3)<T(5)
Executor will return Aggregated result from Task1 + Task2+ Task3 at time=T(5) (+ some linear time)
Executor is continuously dispatching the result as they arrive at time=T(1), T(3), and T(5) to an external MessageDispatcher which can process those results even before Executor completes its job. (Please note, the MessageDispatcher is shown as Queue in fig. below, but in code it is just another java class for POC needs. But the idea is same.)
The following diagram further illustrates the problem statement.
Code Sample: Please find the code sample for this solution at my repo here.
Results: Following console of above code execution helps validate the theory:
Started Executor at : **** 00:35:43 PDT 2020
Executed Task 1. Completion Time = **** 00:35:43 PDT 2020
Message received at delegator 1 time **** 00:35:43 PDT 2020
Executed Task 3. Completion Time = **** 00:35:46 PDT 2020
Message received at delegator 3 time **** 00:35:46 PDT 2020
Executed Task 2 With Sleep of 5 sec. Completion Time = **** 00:35:48 PDT 2020
Message received at delegator 2 time **** 00:35:48 PDT 2020
Executor ended at time : **** 00:35:48 PDT 2020
Executor returning aggregatedResult= : 6
→ received result= : 6 at **** 00:35:48 PDT 2020
The approach here depends entirely on CompletionService and requires an Executor to carry out executions. JDK provides ExecutorCompletionService that wraps all the details and requires few lines of application code to make all of this work.
I hope you will find this approach valuable in your concurrency needs. Thanks for reading so far and spending your valuable time.
In my quest to keep getting better, I always encourage creative criticism and candid feedback. Please feel free to reach out.
