Method invokeAnyImpl

output/java_guava/1.4.19/MoreExecutors.java:643–711

An implementation of ExecutorService#invokeAny for ListeningExecutorService implementations.

(
    ListeningExecutorService executorService,
    Collection<? extends Callable<T>> tasks,
    boolean timed,
    long nanos)

   */

  @GwtIncompatible
  static <T> T invokeAnyImpl(
    ListeningExecutorService executorService,
    Collection<? extends Callable<T>> tasks,
    boolean timed,
    long nanos) throws InterruptedException, ExecutionException, TimeoutException {
    checkNotNull(executorService);
    int ntasks = tasks.size();
    checkArgument(ntasks > 0);
    List<Future<T>> futures = Lists.newArrayListWithCapacity(ntasks);
    BlockingQueue<Future<T>> futureQueue = Queues.newLinkedBlockingQueue();

    // For efficiency, especially in executors with limited
    // parallelism, check to see if previously submitted tasks are
    // done before submitting more of them. This interleaving
    // plus the exception mechanics account for messiness of main
    // loop.
    try {
      // Record exceptions so that if we fail to obtain any
      // result, we can throw the last exception we got.
      ExecutionException ee = null;
      long lastTime = timed ? System.nanoTime() : 0;
      Iterator<? extends Callable<T>> it = tasks.iterator();
      futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
      --ntasks;
      int active = 1;
      while (true) {
        Future<T> f = futureQueue.poll();
        if (f == null) {
          if (ntasks > 0) {
            --ntasks;
            futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
            ++active;
          } else if (active == 0) {
            break;
          } else if (timed) {
                   f = futureQueue.poll(nanos, TimeUnit.NANOSECONDS);
                   if (f == null) {
                     throw new TimeoutException();
                   }

                   long now = System.nanoTime();
                   nanos -= now - lastTime;
                   lastTime = now;
          } else {
                   f = futureQueue.take();
          }
        }
        if (f != null) {
          --active;
          try {
            return f.get();
          } catch (ExecutionException eex) {
            ee = eex;
          } catch (RuntimeException rex) {
            ee = new ExecutionException(rex);
          }
        }