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