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