Use ExecutorCompletionService to compute results from 5 different datasources in parallel

Upasana | May 17, 2019 | 3 min read | 41 views

Hint: Interviewer must be looking for multi-threading. Fetch count from all sources in parallel. Bonus points: If machine has less than 5 cores (for 5 data sources), say 4 cores, prioritize so that first 4 cores get the highest priority.

At a higher level this is what we want to do:

Sequential version of Program

int count = getCount(ds1);
count = count + getCount(ds2);
count = count + getCount(ds3);
count = count + getCount(ds4);
count = count + getCount(ds5);

int getCount(DataSource ds) {
    //TODO: Execute the query and return the results
    return 0;
}

We can use ExecutorCompletionService to compute count operation in parallel and combine the results as they arrive.

CompletionService javadocs

A CompletionService that uses a supplied Executor to execute tasks. This class arranges that submitted tasks are, upon completion, placed on a queue accessible using take. The class is lightweight enough to be suitable for transient use when processing groups of tasks.

ExecutorCompletionService

Let’s create a placeholder for storing results of each computation.

Result.java

class Result {

    private int count;

    public Result() {
    }

    public Result(int count) {
        this.count = count;
    }

    public int getCount() {
        return count;
    }

    public void setCount(int count) {
        this.count = count;
    }
}

We can create a simple Callable implementation that accepts a datasource and runs a query to return the count of records.

DatasourceCounterTask.java

class DatasourceCounterTask implements Callable<Result> {

    private final DataSource dataSource;

    DatasourceCounterTask(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    @Override
    public Result call() throws Exception {
        return new Result(getCount(dataSource));
    }

    int getCount(DataSource ds) {
        //TODO: Execute the query and return the results
        return 0;
    }
}

Now comes the final part where we will use ExecutorCompletionService to run 5 different tasks in parallel and sum the results.

DatasourceCounter.java

import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.*;

public class DatasourceCounter {
    private int total = 0;

    void getTotalCount() throws ExecutionException, InterruptedException {
        Collection<Callable<Result>> solvers = new ArrayList<>();
        solvers.add(new DatasourceCounterTask(ds1));
        solvers.add(new DatasourceCounterTask(ds2));
        solvers.add(new DatasourceCounterTask(ds3));
        solvers.add(new DatasourceCounterTask(ds4));
        solvers.add(new DatasourceCounterTask(ds5));
        solve(Executors.newFixedThreadPool(5), solvers);
        System.out.println("total count = " + total);
    }

    void solve(Executor e, Collection<Callable<Result>> solvers) throws InterruptedException, ExecutionException {
        CompletionService<Result> ecs = new ExecutorCompletionService<>(e);
        for (Callable<Result> s : solvers)
            ecs.submit(s);
        int n = solvers.size();
        for (int i = 0; i < n; ++i) {
            Result r = ecs.take().get();    (1)
            if (r != null)
                use(r); (2)
        }
    }

     void use(Result result) {
        total += result.getCount();
    }
}

1	We submit the tasks to executor and wait for their results as they arrive.
2	We are summing up the results in a variable. this happens in the main thread so we do not need synchronization here.

That’s all.

The main benefit of using ExecutorCompletionService is that it makes results available to the caller as soon as they are ready, rather than to wait for entire set of tasks. That too without delving into low level semantics of inter thread communication.