How to propagate timeout from one FutureTask to another dependent one used by its Callable?

Question

I am in the following situation (perhaps I have overengineered the whole thing or I am in a complete deadlock, but cannot think to another way of doing that):

• Take one or more FutureTask that realize an asynchronous computation (listening of the network for multicast packets conveing statistics of different kinds), we will name it MulticastStatisticsProvider,
• Take another computation that depends on this first task to perform an extra computation (combining statistics with formulas to expose some synthetic informations), this computation is also asynchronous so defined in another FutureTask that we will call FormulaComputing,
• Problem: We want that FormulaComputing.get(timeout, timeUnit) propagates the timeout to its inner Callable when it calls MulticastStatisticsProvider.get(timeout, timeUnit) and cannot find a way to achieve this.

Hereunder, the state of code I achieve so far:

• This is the caller code.

  // This is the code creating the formula computing code.
  public FormulaComputing getFormulaComputing() {
    // Retrieve from another service a list of FutureTasks already
    // scheduled for execution in their own ThreadPool.
    List<MulticastStatisticsProvider> requiredTasks = service.getRequiredTasks();
    // Create the formulaComputing task and schedule it for execution
    FormulaComputing formulaComputing = new FormulaComputing(requiredTasks);
    threadPool.execute(formulaComputing);
    return formulaComputing;
  }
  
  // And then, from another caller
  getFormulaComputing().get(10, TimeUnit.SECONDS);
  

• This is the FormulaComputing code:

  public class FormulaComputing extends FutureTask<Object> {
    private long timeout;
    private TimeUnit timeUnit;
    private Map<String, Future<Map<String, ? extends AbstractSymposiumMessage>>> promises;
    private Closure<Object> callback;
  
    public FormulaComputing(List<MulticastStatisticsProvider> dependentTasks, Closure<Object> callback) {
      super(new Callable<Object>() {
        @Override
        public Object call() throws Exception {
          List<Object> results = new ArrayList<Object>();
          for (MulticastStatisticsProvider task : dependentTasks) {
            // Here is the problem, you cannot access field values in constructor: "Cannot refer to an instance method while explicitly invoking a constructor".
            results.add(task.get(getTimeout(), getTimeUnit()));
          }
          return callback.call(results.toArray());
        }
      });
    }
  
    @Override
    public Object get(long timeout, TimeUnit timeUnit) throws InterruptedException, ExecutionException, TimeoutException {
      this.timeout = timeout;
      this.timeUnit = timeUnit;
      return super.get(timeout, timeUnit);
    }
  }
  

I have imagined passing by introspection to alter private inner sync field of FutureTask with my hand-made Callable in my custom get method, but introspection and reflection are generally avoidable hacks.

Answer 1

If you use Guava, it looks like a good case for ListenableFutures:

List<ListenableFuture<Object>> requiredTasks = ...;

ListenableFuture<List<Object>> requiredTasksResult = Futures.allAsList(requiredTasks);

ListenableFuture<Object> resultFuture = Futures.transform(requiredTasksResult, new Function<List<Object>, Object>() {
    public Object apply(List<Object> results) {
        // Apply computing formula
    }
}, threadPool); // Function will be executed in threadPool

Object result = resultFuture.get(10, TimeUnit.SECONDS);

You can get ListenableFuture for FutureTask by submitting it to ListeningExecutorService, or by using ListenableFutureTask instead.