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