001/*
002 * Copyright (C) 2007 The Guava Authors
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 * http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016
017package com.google.common.util.concurrent;
018
019import static com.google.common.base.Preconditions.checkArgument;
020import static com.google.common.base.Preconditions.checkNotNull;
021
022import com.google.common.annotations.Beta;
023import com.google.common.annotations.GwtCompatible;
024import com.google.common.annotations.GwtIncompatible;
025import com.google.common.annotations.VisibleForTesting;
026import com.google.common.base.Supplier;
027import com.google.common.base.Throwables;
028import com.google.common.collect.Lists;
029import com.google.common.collect.Queues;
030import com.google.common.util.concurrent.ForwardingListenableFuture.SimpleForwardingListenableFuture;
031
032import java.lang.reflect.InvocationTargetException;
033import java.util.Collection;
034import java.util.Collections;
035import java.util.Iterator;
036import java.util.List;
037import java.util.concurrent.BlockingQueue;
038import java.util.concurrent.Callable;
039import java.util.concurrent.Delayed;
040import java.util.concurrent.ExecutionException;
041import java.util.concurrent.Executor;
042import java.util.concurrent.ExecutorService;
043import java.util.concurrent.Executors;
044import java.util.concurrent.Future;
045import java.util.concurrent.RejectedExecutionException;
046import java.util.concurrent.ScheduledExecutorService;
047import java.util.concurrent.ScheduledFuture;
048import java.util.concurrent.ScheduledThreadPoolExecutor;
049import java.util.concurrent.ThreadFactory;
050import java.util.concurrent.ThreadPoolExecutor;
051import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
052import java.util.concurrent.TimeUnit;
053import java.util.concurrent.TimeoutException;
054
055import javax.annotation.concurrent.GuardedBy;
056
057/**
058 * Factory and utility methods for {@link java.util.concurrent.Executor}, {@link
059 * ExecutorService}, and {@link ThreadFactory}.
060 *
061 * @author Eric Fellheimer
062 * @author Kyle Littlefield
063 * @author Justin Mahoney
064 * @since 3.0
065 */
066@GwtCompatible(emulated = true)
067public final class MoreExecutors {
068  private MoreExecutors() {}
069
070  /**
071   * Converts the given ThreadPoolExecutor into an ExecutorService that exits
072   * when the application is complete.  It does so by using daemon threads and
073   * adding a shutdown hook to wait for their completion.
074   *
075   * <p>This is mainly for fixed thread pools.
076   * See {@link Executors#newFixedThreadPool(int)}.
077   *
078   * @param executor the executor to modify to make sure it exits when the
079   *        application is finished
080   * @param terminationTimeout how long to wait for the executor to
081   *        finish before terminating the JVM
082   * @param timeUnit unit of time for the time parameter
083   * @return an unmodifiable version of the input which will not hang the JVM
084   */
085  @Beta
086  @GwtIncompatible("TODO")
087  public static ExecutorService getExitingExecutorService(
088      ThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
089    return new Application()
090        .getExitingExecutorService(executor, terminationTimeout, timeUnit);
091  }
092
093  /**
094   * Converts the given ScheduledThreadPoolExecutor into a
095   * ScheduledExecutorService that exits when the application is complete.  It
096   * does so by using daemon threads and adding a shutdown hook to wait for
097   * their completion.
098   *
099   * <p>This is mainly for fixed thread pools.
100   * See {@link Executors#newScheduledThreadPool(int)}.
101   *
102   * @param executor the executor to modify to make sure it exits when the
103   *        application is finished
104   * @param terminationTimeout how long to wait for the executor to
105   *        finish before terminating the JVM
106   * @param timeUnit unit of time for the time parameter
107   * @return an unmodifiable version of the input which will not hang the JVM
108   */
109  @Beta
110  @GwtIncompatible("TODO")
111  public static ScheduledExecutorService getExitingScheduledExecutorService(
112      ScheduledThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
113    return new Application()
114        .getExitingScheduledExecutorService(executor, terminationTimeout, timeUnit);
115  }
116
117  /**
118   * Add a shutdown hook to wait for thread completion in the given
119   * {@link ExecutorService service}.  This is useful if the given service uses
120   * daemon threads, and we want to keep the JVM from exiting immediately on
121   * shutdown, instead giving these daemon threads a chance to terminate
122   * normally.
123   * @param service ExecutorService which uses daemon threads
124   * @param terminationTimeout how long to wait for the executor to finish
125   *        before terminating the JVM
126   * @param timeUnit unit of time for the time parameter
127   */
128  @Beta
129  @GwtIncompatible("TODO")
130  public static void addDelayedShutdownHook(
131      ExecutorService service, long terminationTimeout, TimeUnit timeUnit) {
132    new Application()
133        .addDelayedShutdownHook(service, terminationTimeout, timeUnit);
134  }
135
136  /**
137   * Converts the given ThreadPoolExecutor into an ExecutorService that exits
138   * when the application is complete.  It does so by using daemon threads and
139   * adding a shutdown hook to wait for their completion.
140   *
141   * <p>This method waits 120 seconds before continuing with JVM termination,
142   * even if the executor has not finished its work.
143   *
144   * <p>This is mainly for fixed thread pools.
145   * See {@link Executors#newFixedThreadPool(int)}.
146   *
147   * @param executor the executor to modify to make sure it exits when the
148   *        application is finished
149   * @return an unmodifiable version of the input which will not hang the JVM
150   */
151  @Beta
152  @GwtIncompatible("concurrency")
153  public static ExecutorService getExitingExecutorService(ThreadPoolExecutor executor) {
154    return new Application().getExitingExecutorService(executor);
155  }
156
157  /**
158   * Converts the given ThreadPoolExecutor into a ScheduledExecutorService that
159   * exits when the application is complete.  It does so by using daemon threads
160   * and adding a shutdown hook to wait for their completion.
161   *
162   * <p>This method waits 120 seconds before continuing with JVM termination,
163   * even if the executor has not finished its work.
164   *
165   * <p>This is mainly for fixed thread pools.
166   * See {@link Executors#newScheduledThreadPool(int)}.
167   *
168   * @param executor the executor to modify to make sure it exits when the
169   *        application is finished
170   * @return an unmodifiable version of the input which will not hang the JVM
171   */
172  @Beta
173  @GwtIncompatible("TODO")
174  public static ScheduledExecutorService getExitingScheduledExecutorService(
175      ScheduledThreadPoolExecutor executor) {
176    return new Application().getExitingScheduledExecutorService(executor);
177  }
178
179  /** Represents the current application to register shutdown hooks. */
180  @GwtIncompatible("TODO")
181  @VisibleForTesting
182  static class Application {
183
184    final ExecutorService getExitingExecutorService(
185        ThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
186      useDaemonThreadFactory(executor);
187      ExecutorService service = Executors.unconfigurableExecutorService(executor);
188      addDelayedShutdownHook(service, terminationTimeout, timeUnit);
189      return service;
190    }
191
192    final ScheduledExecutorService getExitingScheduledExecutorService(
193        ScheduledThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
194      useDaemonThreadFactory(executor);
195      ScheduledExecutorService service = Executors.unconfigurableScheduledExecutorService(executor);
196      addDelayedShutdownHook(service, terminationTimeout, timeUnit);
197      return service;
198    }
199
200    final void addDelayedShutdownHook(
201        final ExecutorService service, final long terminationTimeout, final TimeUnit timeUnit) {
202      checkNotNull(service);
203      checkNotNull(timeUnit);
204      addShutdownHook(MoreExecutors.newThread("DelayedShutdownHook-for-" + service, new Runnable() {
205        @Override
206        public void run() {
207          try {
208            // We'd like to log progress and failures that may arise in the
209            // following code, but unfortunately the behavior of logging
210            // is undefined in shutdown hooks.
211            // This is because the logging code installs a shutdown hook of its
212            // own. See Cleaner class inside {@link LogManager}.
213            service.shutdown();
214            service.awaitTermination(terminationTimeout, timeUnit);
215          } catch (InterruptedException ignored) {
216            // We're shutting down anyway, so just ignore.
217          }
218        }
219      }));
220    }
221
222    final ExecutorService getExitingExecutorService(ThreadPoolExecutor executor) {
223      return getExitingExecutorService(executor, 120, TimeUnit.SECONDS);
224    }
225
226    final ScheduledExecutorService getExitingScheduledExecutorService(
227        ScheduledThreadPoolExecutor executor) {
228      return getExitingScheduledExecutorService(executor, 120, TimeUnit.SECONDS);
229    }
230
231    @VisibleForTesting void addShutdownHook(Thread hook) {
232      Runtime.getRuntime().addShutdownHook(hook);
233    }
234  }
235
236  @GwtIncompatible("TODO")
237  private static void useDaemonThreadFactory(ThreadPoolExecutor executor) {
238    executor.setThreadFactory(new ThreadFactoryBuilder()
239        .setDaemon(true)
240        .setThreadFactory(executor.getThreadFactory())
241        .build());
242  }
243
244  /**
245   * Creates an executor service that runs each task in the thread
246   * that invokes {@code execute/submit}, as in {@link CallerRunsPolicy}.  This
247   * applies both to individually submitted tasks and to collections of tasks
248   * submitted via {@code invokeAll} or {@code invokeAny}.  In the latter case,
249   * tasks will run serially on the calling thread.  Tasks are run to
250   * completion before a {@code Future} is returned to the caller (unless the
251   * executor has been shutdown).
252   *
253   * <p>Although all tasks are immediately executed in the thread that
254   * submitted the task, this {@code ExecutorService} imposes a small
255   * locking overhead on each task submission in order to implement shutdown
256   * and termination behavior.
257   *
258   * <p>The implementation deviates from the {@code ExecutorService}
259   * specification with regards to the {@code shutdownNow} method.  First,
260   * "best-effort" with regards to canceling running tasks is implemented
261   * as "no-effort".  No interrupts or other attempts are made to stop
262   * threads executing tasks.  Second, the returned list will always be empty,
263   * as any submitted task is considered to have started execution.
264   * This applies also to tasks given to {@code invokeAll} or {@code invokeAny}
265   * which are pending serial execution, even the subset of the tasks that
266   * have not yet started execution.  It is unclear from the
267   * {@code ExecutorService} specification if these should be included, and
268   * it's much easier to implement the interpretation that they not be.
269   * Finally, a call to {@code shutdown} or {@code shutdownNow} may result
270   * in concurrent calls to {@code invokeAll/invokeAny} throwing
271   * RejectedExecutionException, although a subset of the tasks may already
272   * have been executed.
273   *
274   * @since 10.0 (<a href="https://github.com/google/guava/wiki/Compatibility"
275   *        >mostly source-compatible</a> since 3.0)
276   * @deprecated Use {@link #directExecutor()} if you only require an {@link Executor} and
277   *     {@link #newDirectExecutorService()} if you need a {@link ListeningExecutorService}. This
278   *     method will be removed in August 2016.
279   */
280  @Deprecated
281  @GwtIncompatible("TODO")
282  public static ListeningExecutorService sameThreadExecutor() {
283    return new DirectExecutorService();
284  }
285
286  // See sameThreadExecutor javadoc for behavioral notes.
287  @GwtIncompatible("TODO")
288  private static final class DirectExecutorService
289      extends AbstractListeningExecutorService {
290    /**
291     * Lock used whenever accessing the state variables
292     * (runningTasks, shutdown) of the executor
293     */
294    private final Object lock = new Object();
295
296    /*
297     * Conceptually, these two variables describe the executor being in
298     * one of three states:
299     *   - Active: shutdown == false
300     *   - Shutdown: runningTasks > 0 and shutdown == true
301     *   - Terminated: runningTasks == 0 and shutdown == true
302     */
303    @GuardedBy("lock") private int runningTasks = 0;
304    @GuardedBy("lock") private boolean shutdown = false;
305
306    @Override
307    public void execute(Runnable command) {
308      startTask();
309      try {
310        command.run();
311      } finally {
312        endTask();
313      }
314    }
315
316    @Override
317    public boolean isShutdown() {
318      synchronized (lock) {
319        return shutdown;
320      }
321    }
322
323    @Override
324    public void shutdown() {
325      synchronized (lock) {
326        shutdown = true;
327        if (runningTasks == 0) {
328          lock.notifyAll();
329        }
330      }
331    }
332
333    // See sameThreadExecutor javadoc for unusual behavior of this method.
334    @Override
335    public List<Runnable> shutdownNow() {
336      shutdown();
337      return Collections.emptyList();
338    }
339
340    @Override
341    public boolean isTerminated() {
342      synchronized (lock) {
343        return shutdown && runningTasks == 0;
344      }
345    }
346
347    @Override
348    public boolean awaitTermination(long timeout, TimeUnit unit)
349        throws InterruptedException {
350      long nanos = unit.toNanos(timeout);
351      synchronized (lock) {
352        for (;;) {
353          if (shutdown && runningTasks == 0) {
354            return true;
355          } else if (nanos <= 0) {
356            return false;
357          } else {
358            long now = System.nanoTime();
359            TimeUnit.NANOSECONDS.timedWait(lock, nanos);
360            nanos -= System.nanoTime() - now;  // subtract the actual time we waited
361          }
362        }
363      }
364    }
365
366    /**
367     * Checks if the executor has been shut down and increments the running
368     * task count.
369     *
370     * @throws RejectedExecutionException if the executor has been previously
371     *         shutdown
372     */
373    private void startTask() {
374      synchronized (lock) {
375        if (shutdown) {
376          throw new RejectedExecutionException("Executor already shutdown");
377        }
378        runningTasks++;
379      }
380    }
381
382    /**
383     * Decrements the running task count.
384     */
385    private void endTask() {
386      synchronized (lock) {
387        int numRunning = --runningTasks;
388        if (numRunning == 0) {
389          lock.notifyAll();
390        }
391      }
392    }
393  }
394
395  /**
396   * Creates an executor service that runs each task in the thread
397   * that invokes {@code execute/submit}, as in {@link CallerRunsPolicy}  This
398   * applies both to individually submitted tasks and to collections of tasks
399   * submitted via {@code invokeAll} or {@code invokeAny}.  In the latter case,
400   * tasks will run serially on the calling thread.  Tasks are run to
401   * completion before a {@code Future} is returned to the caller (unless the
402   * executor has been shutdown).
403   *
404   * <p>Although all tasks are immediately executed in the thread that
405   * submitted the task, this {@code ExecutorService} imposes a small
406   * locking overhead on each task submission in order to implement shutdown
407   * and termination behavior.
408   *
409   * <p>The implementation deviates from the {@code ExecutorService}
410   * specification with regards to the {@code shutdownNow} method.  First,
411   * "best-effort" with regards to canceling running tasks is implemented
412   * as "no-effort".  No interrupts or other attempts are made to stop
413   * threads executing tasks.  Second, the returned list will always be empty,
414   * as any submitted task is considered to have started execution.
415   * This applies also to tasks given to {@code invokeAll} or {@code invokeAny}
416   * which are pending serial execution, even the subset of the tasks that
417   * have not yet started execution.  It is unclear from the
418   * {@code ExecutorService} specification if these should be included, and
419   * it's much easier to implement the interpretation that they not be.
420   * Finally, a call to {@code shutdown} or {@code shutdownNow} may result
421   * in concurrent calls to {@code invokeAll/invokeAny} throwing
422   * RejectedExecutionException, although a subset of the tasks may already
423   * have been executed.
424   *
425   * @since 18.0 (present as MoreExecutors.sameThreadExecutor() since 10.0)
426   */
427  @GwtIncompatible("TODO")
428  public static ListeningExecutorService newDirectExecutorService() {
429    return new DirectExecutorService();
430  }
431
432  /**
433   * Returns an {@link Executor} that runs each task in the thread that invokes
434   * {@link Executor#execute execute}, as in {@link CallerRunsPolicy}.
435   *
436   * <p>This instance is equivalent to: <pre>   {@code
437   *   final class DirectExecutor implements Executor {
438   *     public void execute(Runnable r) {
439   *       r.run();
440   *     }
441   *   }}</pre>
442   *
443   * <p>This should be preferred to {@link #newDirectExecutorService()} because the implementing the
444   * {@link ExecutorService} subinterface necessitates significant performance overhead.
445   *
446   * @since 18.0
447   */
448  public static Executor directExecutor() {
449    return DirectExecutor.INSTANCE;
450  }
451
452  /** See {@link #directExecutor} for behavioral notes. */
453  private enum DirectExecutor implements Executor {
454    INSTANCE;
455    @Override public void execute(Runnable command) {
456      command.run();
457    }
458
459    @Override public String toString() {
460      return "MoreExecutors.directExecutor()";
461    }
462  }
463
464  /**
465   * Creates an {@link ExecutorService} whose {@code submit} and {@code
466   * invokeAll} methods submit {@link ListenableFutureTask} instances to the
467   * given delegate executor. Those methods, as well as {@code execute} and
468   * {@code invokeAny}, are implemented in terms of calls to {@code
469   * delegate.execute}. All other methods are forwarded unchanged to the
470   * delegate. This implies that the returned {@code ListeningExecutorService}
471   * never calls the delegate's {@code submit}, {@code invokeAll}, and {@code
472   * invokeAny} methods, so any special handling of tasks must be implemented in
473   * the delegate's {@code execute} method or by wrapping the returned {@code
474   * ListeningExecutorService}.
475   *
476   * <p>If the delegate executor was already an instance of {@code
477   * ListeningExecutorService}, it is returned untouched, and the rest of this
478   * documentation does not apply.
479   *
480   * @since 10.0
481   */
482  @GwtIncompatible("TODO")
483  public static ListeningExecutorService listeningDecorator(
484      ExecutorService delegate) {
485    return (delegate instanceof ListeningExecutorService)
486        ? (ListeningExecutorService) delegate
487        : (delegate instanceof ScheduledExecutorService)
488        ? new ScheduledListeningDecorator((ScheduledExecutorService) delegate)
489        : new ListeningDecorator(delegate);
490  }
491
492  /**
493   * Creates a {@link ScheduledExecutorService} whose {@code submit} and {@code
494   * invokeAll} methods submit {@link ListenableFutureTask} instances to the
495   * given delegate executor. Those methods, as well as {@code execute} and
496   * {@code invokeAny}, are implemented in terms of calls to {@code
497   * delegate.execute}. All other methods are forwarded unchanged to the
498   * delegate. This implies that the returned {@code
499   * ListeningScheduledExecutorService} never calls the delegate's {@code
500   * submit}, {@code invokeAll}, and {@code invokeAny} methods, so any special
501   * handling of tasks must be implemented in the delegate's {@code execute}
502   * method or by wrapping the returned {@code
503   * ListeningScheduledExecutorService}.
504   *
505   * <p>If the delegate executor was already an instance of {@code
506   * ListeningScheduledExecutorService}, it is returned untouched, and the rest
507   * of this documentation does not apply.
508   *
509   * @since 10.0
510   */
511  @GwtIncompatible("TODO")
512  public static ListeningScheduledExecutorService listeningDecorator(
513      ScheduledExecutorService delegate) {
514    return (delegate instanceof ListeningScheduledExecutorService)
515        ? (ListeningScheduledExecutorService) delegate
516        : new ScheduledListeningDecorator(delegate);
517  }
518
519  @GwtIncompatible("TODO")
520  private static class ListeningDecorator
521      extends AbstractListeningExecutorService {
522    private final ExecutorService delegate;
523
524    ListeningDecorator(ExecutorService delegate) {
525      this.delegate = checkNotNull(delegate);
526    }
527
528    @Override
529    public final boolean awaitTermination(long timeout, TimeUnit unit)
530        throws InterruptedException {
531      return delegate.awaitTermination(timeout, unit);
532    }
533
534    @Override
535    public final boolean isShutdown() {
536      return delegate.isShutdown();
537    }
538
539    @Override
540    public final boolean isTerminated() {
541      return delegate.isTerminated();
542    }
543
544    @Override
545    public final void shutdown() {
546      delegate.shutdown();
547    }
548
549    @Override
550    public final List<Runnable> shutdownNow() {
551      return delegate.shutdownNow();
552    }
553
554    @Override
555    public final void execute(Runnable command) {
556      delegate.execute(command);
557    }
558  }
559
560  @GwtIncompatible("TODO")
561  private static final class ScheduledListeningDecorator
562      extends ListeningDecorator implements ListeningScheduledExecutorService {
563    @SuppressWarnings("hiding")
564    final ScheduledExecutorService delegate;
565
566    ScheduledListeningDecorator(ScheduledExecutorService delegate) {
567      super(delegate);
568      this.delegate = checkNotNull(delegate);
569    }
570
571    @Override
572    public ListenableScheduledFuture<?> schedule(
573        Runnable command, long delay, TimeUnit unit) {
574      TrustedListenableFutureTask<Void> task =
575          TrustedListenableFutureTask.create(command, null);
576      ScheduledFuture<?> scheduled = delegate.schedule(task, delay, unit);
577      return new ListenableScheduledTask<Void>(task, scheduled);
578    }
579
580    @Override
581    public <V> ListenableScheduledFuture<V> schedule(
582        Callable<V> callable, long delay, TimeUnit unit) {
583      TrustedListenableFutureTask<V> task = TrustedListenableFutureTask.create(callable);
584      ScheduledFuture<?> scheduled = delegate.schedule(task, delay, unit);
585      return new ListenableScheduledTask<V>(task, scheduled);
586    }
587
588    @Override
589    public ListenableScheduledFuture<?> scheduleAtFixedRate(
590        Runnable command, long initialDelay, long period, TimeUnit unit) {
591      NeverSuccessfulListenableFutureTask task =
592          new NeverSuccessfulListenableFutureTask(command);
593      ScheduledFuture<?> scheduled =
594          delegate.scheduleAtFixedRate(task, initialDelay, period, unit);
595      return new ListenableScheduledTask<Void>(task, scheduled);
596    }
597
598    @Override
599    public ListenableScheduledFuture<?> scheduleWithFixedDelay(
600        Runnable command, long initialDelay, long delay, TimeUnit unit) {
601      NeverSuccessfulListenableFutureTask task =
602          new NeverSuccessfulListenableFutureTask(command);
603      ScheduledFuture<?> scheduled =
604          delegate.scheduleWithFixedDelay(task, initialDelay, delay, unit);
605      return new ListenableScheduledTask<Void>(task, scheduled);
606    }
607
608    private static final class ListenableScheduledTask<V>
609        extends SimpleForwardingListenableFuture<V>
610        implements ListenableScheduledFuture<V> {
611
612      private final ScheduledFuture<?> scheduledDelegate;
613
614      public ListenableScheduledTask(
615          ListenableFuture<V> listenableDelegate,
616          ScheduledFuture<?> scheduledDelegate) {
617        super(listenableDelegate);
618        this.scheduledDelegate = scheduledDelegate;
619      }
620
621      @Override
622      public boolean cancel(boolean mayInterruptIfRunning) {
623        boolean cancelled = super.cancel(mayInterruptIfRunning);
624        if (cancelled) {
625          // Unless it is cancelled, the delegate may continue being scheduled
626          scheduledDelegate.cancel(mayInterruptIfRunning);
627
628          // TODO(user): Cancel "this" if "scheduledDelegate" is cancelled.
629        }
630        return cancelled;
631      }
632
633      @Override
634      public long getDelay(TimeUnit unit) {
635        return scheduledDelegate.getDelay(unit);
636      }
637
638      @Override
639      public int compareTo(Delayed other) {
640        return scheduledDelegate.compareTo(other);
641      }
642    }
643
644    @GwtIncompatible("TODO")
645    private static final class NeverSuccessfulListenableFutureTask
646        extends AbstractFuture<Void>
647        implements Runnable {
648      private final Runnable delegate;
649
650      public NeverSuccessfulListenableFutureTask(Runnable delegate) {
651        this.delegate = checkNotNull(delegate);
652      }
653
654      @Override public void run() {
655        try {
656          delegate.run();
657        } catch (Throwable t) {
658          setException(t);
659          throw Throwables.propagate(t);
660        }
661      }
662    }
663  }
664
665  /*
666   * This following method is a modified version of one found in
667   * http://gee.cs.oswego.edu/cgi-bin/viewcvs.cgi/jsr166/src/test/tck/AbstractExecutorServiceTest.java?revision=1.30
668   * which contained the following notice:
669   *
670   * Written by Doug Lea with assistance from members of JCP JSR-166
671   * Expert Group and released to the public domain, as explained at
672   * http://creativecommons.org/publicdomain/zero/1.0/
673   * Other contributors include Andrew Wright, Jeffrey Hayes,
674   * Pat Fisher, Mike Judd.
675   */
676
677  /**
678   * An implementation of {@link ExecutorService#invokeAny} for {@link ListeningExecutorService}
679   * implementations.
680   */ static <T> T invokeAnyImpl(ListeningExecutorService executorService,
681      Collection<? extends Callable<T>> tasks, boolean timed, long nanos)
682          throws InterruptedException, ExecutionException, TimeoutException {
683    checkNotNull(executorService);
684    int ntasks = tasks.size();
685    checkArgument(ntasks > 0);
686    List<Future<T>> futures = Lists.newArrayListWithCapacity(ntasks);
687    BlockingQueue<Future<T>> futureQueue = Queues.newLinkedBlockingQueue();
688
689    // For efficiency, especially in executors with limited
690    // parallelism, check to see if previously submitted tasks are
691    // done before submitting more of them. This interleaving
692    // plus the exception mechanics account for messiness of main
693    // loop.
694
695    try {
696      // Record exceptions so that if we fail to obtain any
697      // result, we can throw the last exception we got.
698      ExecutionException ee = null;
699      long lastTime = timed ? System.nanoTime() : 0;
700      Iterator<? extends Callable<T>> it = tasks.iterator();
701
702      futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
703      --ntasks;
704      int active = 1;
705
706      for (;;) {
707        Future<T> f = futureQueue.poll();
708        if (f == null) {
709          if (ntasks > 0) {
710            --ntasks;
711            futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
712            ++active;
713          } else if (active == 0) {
714            break;
715          } else if (timed) {
716            f = futureQueue.poll(nanos, TimeUnit.NANOSECONDS);
717            if (f == null) {
718              throw new TimeoutException();
719            }
720            long now = System.nanoTime();
721            nanos -= now - lastTime;
722            lastTime = now;
723          } else {
724            f = futureQueue.take();
725          }
726        }
727        if (f != null) {
728          --active;
729          try {
730            return f.get();
731          } catch (ExecutionException eex) {
732            ee = eex;
733          } catch (RuntimeException rex) {
734            ee = new ExecutionException(rex);
735          }
736        }
737      }
738
739      if (ee == null) {
740        ee = new ExecutionException(null);
741      }
742      throw ee;
743    } finally {
744      for (Future<T> f : futures) {
745        f.cancel(true);
746      }
747    }
748  }
749
750  /**
751   * Submits the task and adds a listener that adds the future to {@code queue} when it completes.
752   */
753  @GwtIncompatible("TODO")
754  private static <T> ListenableFuture<T> submitAndAddQueueListener(
755      ListeningExecutorService executorService, Callable<T> task,
756      final BlockingQueue<Future<T>> queue) {
757    final ListenableFuture<T> future = executorService.submit(task);
758    future.addListener(new Runnable() {
759      @Override public void run() {
760        queue.add(future);
761      }
762    }, directExecutor());
763    return future;
764  }
765
766  /**
767   * Returns a default thread factory used to create new threads.
768   *
769   * <p>On AppEngine, returns {@code ThreadManager.currentRequestThreadFactory()}.
770   * Otherwise, returns {@link Executors#defaultThreadFactory()}.
771   *
772   * @since 14.0
773   */
774  @Beta
775  @GwtIncompatible("concurrency")
776  public static ThreadFactory platformThreadFactory() {
777    if (!isAppEngine()) {
778      return Executors.defaultThreadFactory();
779    }
780    try {
781      return (ThreadFactory) Class.forName("com.google.appengine.api.ThreadManager")
782          .getMethod("currentRequestThreadFactory")
783          .invoke(null);
784    } catch (IllegalAccessException e) {
785      throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e);
786    } catch (ClassNotFoundException e) {
787      throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e);
788    } catch (NoSuchMethodException e) {
789      throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e);
790    } catch (InvocationTargetException e) {
791      throw Throwables.propagate(e.getCause());
792    }
793  }
794
795  @GwtIncompatible("TODO")
796  private static boolean isAppEngine() {
797    if (System.getProperty("com.google.appengine.runtime.environment") == null) {
798      return false;
799    }
800    try {
801      // If the current environment is null, we're not inside AppEngine.
802      return Class.forName("com.google.apphosting.api.ApiProxy")
803          .getMethod("getCurrentEnvironment")
804          .invoke(null) != null;
805    } catch (ClassNotFoundException e) {
806      // If ApiProxy doesn't exist, we're not on AppEngine at all.
807      return false;
808    } catch (InvocationTargetException e) {
809      // If ApiProxy throws an exception, we're not in a proper AppEngine environment.
810      return false;
811    } catch (IllegalAccessException e) {
812      // If the method isn't accessible, we're not on a supported version of AppEngine;
813      return false;
814    } catch (NoSuchMethodException e) {
815      // If the method doesn't exist, we're not on a supported version of AppEngine;
816      return false;
817    }
818  }
819
820  /**
821   * Creates a thread using {@link #platformThreadFactory}, and sets its name to {@code name}
822   * unless changing the name is forbidden by the security manager.
823   */
824  @GwtIncompatible("concurrency")
825  static Thread newThread(String name, Runnable runnable) {
826    checkNotNull(name);
827    checkNotNull(runnable);
828    Thread result = platformThreadFactory().newThread(runnable);
829    try {
830      result.setName(name);
831    } catch (SecurityException e) {
832      // OK if we can't set the name in this environment.
833    }
834    return result;
835  }
836
837  // TODO(lukes): provide overloads for ListeningExecutorService? ListeningScheduledExecutorService?
838  // TODO(lukes): provide overloads that take constant strings? Function<Runnable, String>s to
839  // calculate names?
840
841  /**
842   * Creates an {@link Executor} that renames the {@link Thread threads} that its tasks run in.
843   *
844   * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
845   * right before each task is run.  The renaming is best effort, if a {@link SecurityManager}
846   * prevents the renaming then it will be skipped but the tasks will still execute.
847   *
848   *
849   * @param executor The executor to decorate
850   * @param nameSupplier The source of names for each task
851   */
852  @GwtIncompatible("concurrency")
853  static Executor renamingDecorator(final Executor executor, final Supplier<String> nameSupplier) {
854    checkNotNull(executor);
855    checkNotNull(nameSupplier);
856    if (isAppEngine()) {
857      // AppEngine doesn't support thread renaming, so don't even try
858      return executor;
859    }
860    return new Executor() {
861      @Override public void execute(Runnable command) {
862        executor.execute(Callables.threadRenaming(command, nameSupplier));
863      }
864    };
865  }
866
867  /**
868   * Creates an {@link ExecutorService} that renames the {@link Thread threads} that its tasks run
869   * in.
870   *
871   * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
872   * right before each task is run.  The renaming is best effort, if a {@link SecurityManager}
873   * prevents the renaming then it will be skipped but the tasks will still execute.
874   *
875   *
876   * @param service The executor to decorate
877   * @param nameSupplier The source of names for each task
878   */
879  @GwtIncompatible("concurrency")
880  static ExecutorService renamingDecorator(final ExecutorService service,
881      final Supplier<String> nameSupplier) {
882    checkNotNull(service);
883    checkNotNull(nameSupplier);
884    if (isAppEngine()) {
885      // AppEngine doesn't support thread renaming, so don't even try.
886      return service;
887    }
888    return new WrappingExecutorService(service) {
889      @Override protected <T> Callable<T> wrapTask(Callable<T> callable) {
890        return Callables.threadRenaming(callable, nameSupplier);
891      }
892      @Override protected Runnable wrapTask(Runnable command) {
893        return Callables.threadRenaming(command, nameSupplier);
894      }
895    };
896  }
897
898  /**
899   * Creates a {@link ScheduledExecutorService} that renames the {@link Thread threads} that its
900   * tasks run in.
901   *
902   * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
903   * right before each task is run.  The renaming is best effort, if a {@link SecurityManager}
904   * prevents the renaming then it will be skipped but the tasks will still execute.
905   *
906   *
907   * @param service The executor to decorate
908   * @param nameSupplier The source of names for each task
909   */
910  @GwtIncompatible("concurrency")
911  static ScheduledExecutorService renamingDecorator(final ScheduledExecutorService service,
912      final Supplier<String> nameSupplier) {
913    checkNotNull(service);
914    checkNotNull(nameSupplier);
915    if (isAppEngine()) {
916      // AppEngine doesn't support thread renaming, so don't even try.
917      return service;
918    }
919    return new WrappingScheduledExecutorService(service) {
920      @Override protected <T> Callable<T> wrapTask(Callable<T> callable) {
921        return Callables.threadRenaming(callable, nameSupplier);
922      }
923      @Override protected Runnable wrapTask(Runnable command) {
924        return Callables.threadRenaming(command, nameSupplier);
925      }
926    };
927  }
928
929  /**
930   * Shuts down the given executor gradually, first disabling new submissions and later cancelling
931   * existing tasks.
932   *
933   * <p>The method takes the following steps:
934   * <ol>
935   *  <li>calls {@link ExecutorService#shutdown()}, disabling acceptance of new submitted tasks.
936   *  <li>waits for half of the specified timeout.
937   *  <li>if the timeout expires, it calls {@link ExecutorService#shutdownNow()}, cancelling
938   *  pending tasks and interrupting running tasks.
939   *  <li>waits for the other half of the specified timeout.
940   * </ol>
941   *
942   * <p>If, at any step of the process, the calling thread is interrupted, the method calls {@link
943   * ExecutorService#shutdownNow()} and returns.
944   *
945   * @param service the {@code ExecutorService} to shut down
946   * @param timeout the maximum time to wait for the {@code ExecutorService} to terminate
947   * @param unit the time unit of the timeout argument
948   * @return {@code true} if the {@code ExecutorService} was terminated successfully, {@code false}
949   *     the call timed out or was interrupted
950   * @since 17.0
951   */
952  @Beta
953  @GwtIncompatible("concurrency")
954  public static boolean shutdownAndAwaitTermination(
955      ExecutorService service, long timeout, TimeUnit unit) {
956    checkNotNull(unit);
957    // Disable new tasks from being submitted
958    service.shutdown();
959    try {
960      long halfTimeoutNanos = TimeUnit.NANOSECONDS.convert(timeout, unit) / 2;
961      // Wait for half the duration of the timeout for existing tasks to terminate
962      if (!service.awaitTermination(halfTimeoutNanos, TimeUnit.NANOSECONDS)) {
963        // Cancel currently executing tasks
964        service.shutdownNow();
965        // Wait the other half of the timeout for tasks to respond to being cancelled
966        service.awaitTermination(halfTimeoutNanos, TimeUnit.NANOSECONDS);
967      }
968    } catch (InterruptedException ie) {
969      // Preserve interrupt status
970      Thread.currentThread().interrupt();
971      // (Re-)Cancel if current thread also interrupted
972      service.shutdownNow();
973    }
974    return service.isTerminated();
975  }
976}