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