How to know which callable finished upon pool method in CompletionService in java

Question

Let say you are writing a code like following:

CompletionService<T> completion = new ExecutorCompletionService<>(new ExecutorService());

for(Callable callable : callableList) {
    completion.submit(callable);
}

// Do something else

while(true) {
    Future<T> future = completion.poll();

    // At this point, is there a way to find out which callable is returning this future?
}

If you see the code above, is there a way to find out which callable is returned at the time I call completion.poll and have future in return?

I can extend Callable to have some kind of ID to do that, but I want to find out if it is possible by java itself generically.

Answer 1

TL;DR: Although there may be some intrusive way of doing so (i.e. reflection), the Future returned by the ExecutionCompletionService.poll() method does not expose the Callable that was completed.

---

Looking at the JDK 9 source for ExecutionCompletionService.submit(Callable), the submitted Callable is wrapped in a RunnableFuture (whose actual type is FutureTask if we look into newTaskFor(Callable)):

public Future<V> submit(Callable<V> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<V> f = newTaskFor(task);
    executor.execute(new QueueingFuture<V>(f, completionQueue));
    return f;
}

private RunnableFuture<V> newTaskFor(Callable<V> task) {
    if (aes == null)
        return new FutureTask<V>(task);
    else
        return aes.newTaskFor(task);
}

Even if aes.newTaskFor(task) is called, where aes is an AbstractExecutorService, the result is a FutureTask:

public abstract class AbstractExecutorService implements ExecutorService {

    // ...

    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }
}

If we look into the QueueingFuture inner class, we see the following definition:

private static class QueueingFuture<V> extends FutureTask<Void> {
    QueueingFuture(RunnableFuture<V> task,
                   BlockingQueue<Future<V>> completionQueue) {
        super(task, null);
        this.task = task;
        this.completionQueue = completionQueue;
    }
    private final Future<V> task;
    private final BlockingQueue<Future<V>> completionQueue;
    protected void done() { completionQueue.add(task); }
}

The completionQueue passed to the newly created QueueingFuture in the submit method is a BlockingQueue<Future<T>> that stores the stores the Future objects that correspond to the completed Callable objects supplied to submit. Stated differently, once a Callable submitted to ExecutorCompletionService using the submit method is finished executing, the Future that corresponds to that submitted Callable is enqueued in completionQueue. The ExecutionCompletionService.poll() method simply delegates to a poll on the completionQueue:

public Future<V> poll() {
    return completionQueue.poll();
}

Thus, the only way we could obtain a reference to the original Callable is to obtain it from the Future object returned from ExecutionCompletionService.poll() (your original question). But, looking at the Future interface, there does not exist a method that exposes any Callable:

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

Even the FutureTask (the implementation type returned from ExecutionCompletionService.poll()) does not expose its internal Callable:

public class FutureTask<V> implements RunnableFuture<V> {

    private Callable<V> callable;

    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        // ...
    }
}

If FutureTask had some method exposing the underlying Callable, then we may be able to make an explicit cast to FutureTask, but since it does not, that route would not solve the problem at hand.

The issue here is that we are trying to combine synchronous code with an asynchronous ExecutionCompletionService. One way to solve this issue in a synchronized manner would be to return the ID of the Callable when it completes so that calling Future.get() on the Future returned from ExecutionCompletionService.poll() returns the ID (as you mentioned). Then the ID can be paired to its original Callable. An asynchronous approach would be to register a callback with the original Callable that is called when the Callable completes.

Hopefully the explanation was more helpful than a simple no.