How do I return interim results from Callable?

Question

I have Callable object executed using ExecutorService.

How to return interim results from this callable?

I know there is javax.swing.SwingWorker#publish(results) for Swing but I don't use Swing.

Answer 1

There are a couple of ways of doing this. You could do it with a callback or you could do it with a queue.

Here's an example of doing it with a callback:

public static interface Callback<T> {
    public void on(T event);
}

Then, an implementation of the callback that does something with your in progress events:

final Callback<String> callback = new Callback<String>() {
    public void on(String event) {
        System.out.println(event);
    }
};

Now you can use the callback in your pool:

Future<String> submit = pool.submit(new Callable<String>() {
    public String call() throws Exception {
        for(int i = 0; i < 10; i++) {
            callback.on("process " + i);
        }
        return "done";
    }
});

Answer 2

You can pass, let's say, an AtomicInteger to your class (the one that will be submitted by the executor) inside that class you increment it's value and from the calling thread you check it's value

Something like this:

public class LongComputation {

private AtomicInteger progress = new AtomicInteger(0);

public static void main(String[] args) throws InterruptedException,
        ExecutionException {

    AtomicInteger progress = new AtomicInteger(0);
    LongComputation computation = new LongComputation(progress);

    ExecutorService executor = Executors.newFixedThreadPool(2);
    Future<Integer> result = executor.submit(() -> computation.compute());
    executor.shutdown();

    while (!result.isDone()) {
        System.out.printf("Progress...%d%%%n", progress.intValue());
        TimeUnit.MILLISECONDS.sleep(100);
    }

    System.out.printf("Result=%d%n", result.get());
}

public LongComputation(AtomicInteger progress) {
    this.progress = progress;
}

public int compute() throws InterruptedException {

    for (int i = 0; i < 100; i++) {
        TimeUnit.MILLISECONDS.sleep(100);
        progress.incrementAndGet();
    }

    return 1_000_000;
}

}

Answer 3

It is not clear what an "interim result" really is. The interfaces used in the concurrency package simply do not define this, but assume methods that resemble more or less pure functions.

Hence, instead this:

interim     = compute something
finalresult = compute something else

do something like this:

interim = compute something
final1 = new Pair( interim, fork(new Future() { compute something else }) )

(Pseudocode, thought to convey the idea, not compileable code)

---

EDIT The idea is: instead of running a single monolithic block of computations (that happens to reach a state where some "interim results" are available) break it up so that the first task returns the former "interim" result and, at the same time, forks a second task that computes the final result. Of course, a handle to this task must be delivered to the caller so that it eventually can get the final result. Usually, this is done with the Future interface.

Answer 4

What you're looking for is java.util.concurrent.Future.

A Future represents the result of an asynchronous computation. Methods are provided to check if the computation is complete, to wait for its completion, and to retrieve the result of the computation. 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. Additional methods are provided to determine if the task completed normally or was cancelled. Once a computation has completed, the computation cannot be cancelled. If you would like to use a Future for the sake of cancellability but not provide a usable result, you can declare types of the form Future and return null as a result of the underlying task.

Answer 5

You would have to roll your own API with something like Observer/Observerable if you want to publish intermediate results as a push. A simpler thing would be to just poll for current state through some self defined method.