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