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