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