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