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 com.google.errorprone.annotations.concurrent.GuardedBy;
031import java.lang.reflect.InvocationTargetException;
032import java.util.Collection;
033import java.util.Collections;
034import java.util.Iterator;
035import java.util.List;
036import java.util.concurrent.BlockingQueue;
037import java.util.concurrent.Callable;
038import java.util.concurrent.Delayed;
039import java.util.concurrent.ExecutionException;
040import java.util.concurrent.Executor;
041import java.util.concurrent.ExecutorService;
042import java.util.concurrent.Executors;
043import java.util.concurrent.Future;
044import java.util.concurrent.RejectedExecutionException;
045import java.util.concurrent.ScheduledExecutorService;
046import java.util.concurrent.ScheduledFuture;
047import java.util.concurrent.ScheduledThreadPoolExecutor;
048import java.util.concurrent.ThreadFactory;
049import java.util.concurrent.ThreadPoolExecutor;
050import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
051import java.util.concurrent.TimeUnit;
052import java.util.concurrent.TimeoutException;
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    /** Decrements the running task count. */
328    private void endTask() {
329      synchronized (lock) {
330        int numRunning = --runningTasks;
331        if (numRunning == 0) {
332          lock.notifyAll();
333        }
334      }
335    }
336  }
337
338  /**
339   * Creates an executor service that runs each task in the thread that invokes {@code
340   * execute/submit}, as in {@link CallerRunsPolicy} This applies both to individually submitted
341   * tasks and to collections of tasks submitted via {@code invokeAll} or {@code invokeAny}. In the
342   * latter case, tasks will run serially on the calling thread. Tasks are run to completion before
343   * a {@code Future} is returned to the caller (unless the executor has been shutdown).
344   *
345   * <p>Although all tasks are immediately executed in the thread that submitted the task, this
346   * {@code ExecutorService} imposes a small locking overhead on each task submission in order to
347   * implement shutdown and termination behavior.
348   *
349   * <p>The implementation deviates from the {@code ExecutorService} specification with regards to
350   * the {@code shutdownNow} method. First, "best-effort" with regards to canceling running tasks is
351   * implemented as "no-effort". No interrupts or other attempts are made to stop threads executing
352   * tasks. Second, the returned list will always be empty, as any submitted task is considered to
353   * have started execution. This applies also to tasks given to {@code invokeAll} or {@code
354   * invokeAny} which are pending serial execution, even the subset of the tasks that have not yet
355   * started execution. It is unclear from the {@code ExecutorService} specification if these should
356   * be included, and it's much easier to implement the interpretation that they not be. Finally, a
357   * call to {@code shutdown} or {@code shutdownNow} may result in concurrent calls to {@code
358   * invokeAll/invokeAny} throwing RejectedExecutionException, although a subset of the tasks may
359   * already have been executed.
360   *
361   * @since 18.0 (present as MoreExecutors.sameThreadExecutor() since 10.0)
362   */
363  @GwtIncompatible // TODO
364  public static ListeningExecutorService newDirectExecutorService() {
365    return new DirectExecutorService();
366  }
367
368  /**
369   * Returns an {@link Executor} that runs each task in the thread that invokes {@link
370   * Executor#execute execute}, as in {@link CallerRunsPolicy}.
371   *
372   * <p>This instance is equivalent to:
373   *
374   * <pre>{@code
375   * final class DirectExecutor implements Executor {
376   *   public void execute(Runnable r) {
377   *     r.run();
378   *   }
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   *
386   * @since 18.0
387   */
388  public static Executor directExecutor() {
389    return DirectExecutor.INSTANCE;
390  }
391
392  /** See {@link #directExecutor} for behavioral notes. */
393  private enum DirectExecutor implements Executor {
394    INSTANCE;
395
396    @Override
397    public void execute(Runnable command) {
398      command.run();
399    }
400
401    @Override
402    public String toString() {
403      return "MoreExecutors.directExecutor()";
404    }
405  }
406
407  /**
408   * Returns an {@link Executor} that runs each task executed sequentially, such that no two tasks
409   * are running concurrently. Submitted tasks have a happens-before order as defined in the Java
410   * Language Specification.
411   *
412   * <p>The executor uses {@code delegate} in order to {@link Executor#execute execute} each task in
413   * turn, and does not create any threads of its own.
414   *
415   * <p>After execution begins on a thread from the {@code delegate} {@link Executor}, tasks are
416   * polled and executed from a task queue until there are no more tasks. The thread will not be
417   * released until there are no more tasks to run.
418   *
419   * <p>If a task is submitted while a thread is executing tasks from the task queue, the thread
420   * will not be released until that submitted task is also complete.
421   *
422   * <p>If a task is {@linkplain Thread#interrupt interrupted} while a task is running:
423   *
424   * <ol>
425   *   <li>execution will not stop until the task queue is empty.
426   *   <li>tasks will begin execution with the thread marked as not interrupted - any interruption
427   *       applies only to the task that was running at the point of interruption.
428   *   <li>if the thread was interrupted before the SequentialExecutor's worker begins execution,
429   *       the interrupt will be restored to the thread after it completes so that its {@code
430   *       delegate} Executor may process the interrupt.
431   *   <li>subtasks are run with the thread uninterrupted and interrupts received during execution
432   *       of a task are ignored.
433   * </ol>
434   *
435   * <p>{@code RuntimeException}s thrown by tasks are simply logged and the executor keeps trucking.
436   * If an {@code Error} is thrown, the error will propagate and execution will stop until the next
437   * time a task is submitted.
438   *
439   * <p>When an {@code Error} is thrown by an executed task, previously submitted tasks may never
440   * run. An attempt will be made to restart execution on the next call to {@code execute}. If the
441   * {@code delegate} has begun to reject execution, the previously submitted tasks may never run,
442   * despite not throwing a RejectedExecutionException synchronously with the call to {@code
443   * execute}. If this behaviour is problematic, use an Executor with a single thread (e.g. {@link
444   * Executors#newSingleThreadExecutor}).
445   *
446   * @since 23.3 (since 23.1 as {@code sequentialExecutor})
447   */
448  @Beta
449  @GwtIncompatible
450  public static Executor newSequentialExecutor(Executor delegate) {
451    return new SequentialExecutor(delegate);
452  }
453
454  /**
455   * Creates an {@link ExecutorService} whose {@code submit} and {@code invokeAll} methods submit
456   * {@link ListenableFutureTask} instances to the given delegate executor. Those methods, as well
457   * as {@code execute} and {@code invokeAny}, are implemented in terms of calls to {@code
458   * delegate.execute}. All other methods are forwarded unchanged to the delegate. This implies that
459   * the returned {@code ListeningExecutorService} never calls the delegate's {@code submit}, {@code
460   * invokeAll}, and {@code invokeAny} methods, so any special handling of tasks must be implemented
461   * in the delegate's {@code execute} method or by wrapping the returned {@code
462   * ListeningExecutorService}.
463   *
464   * <p>If the delegate executor was already an instance of {@code ListeningExecutorService}, it is
465   * returned untouched, and the rest of this documentation does not apply.
466   *
467   * @since 10.0
468   */
469  @GwtIncompatible // TODO
470  public static ListeningExecutorService listeningDecorator(ExecutorService delegate) {
471    return (delegate instanceof ListeningExecutorService)
472        ? (ListeningExecutorService) delegate
473        : (delegate instanceof ScheduledExecutorService)
474            ? new ScheduledListeningDecorator((ScheduledExecutorService) delegate)
475            : new ListeningDecorator(delegate);
476  }
477
478  /**
479   * Creates a {@link ScheduledExecutorService} whose {@code submit} and {@code invokeAll} methods
480   * submit {@link ListenableFutureTask} instances to the given delegate executor. Those methods, as
481   * well as {@code execute} and {@code invokeAny}, are implemented in terms of calls to {@code
482   * delegate.execute}. All other methods are forwarded unchanged to the delegate. This implies that
483   * the returned {@code ListeningScheduledExecutorService} never calls the delegate's {@code
484   * submit}, {@code invokeAll}, and {@code invokeAny} methods, so any special handling of tasks
485   * must be implemented in the delegate's {@code execute} method or by wrapping the returned {@code
486   * ListeningScheduledExecutorService}.
487   *
488   * <p>If the delegate executor was already an instance of {@code
489   * ListeningScheduledExecutorService}, it is returned untouched, and the rest of this
490   * documentation does not apply.
491   *
492   * @since 10.0
493   */
494  @GwtIncompatible // TODO
495  public static ListeningScheduledExecutorService listeningDecorator(
496      ScheduledExecutorService delegate) {
497    return (delegate instanceof ListeningScheduledExecutorService)
498        ? (ListeningScheduledExecutorService) delegate
499        : new ScheduledListeningDecorator(delegate);
500  }
501
502  @GwtIncompatible // TODO
503  private static class ListeningDecorator extends AbstractListeningExecutorService {
504    private final ExecutorService delegate;
505
506    ListeningDecorator(ExecutorService delegate) {
507      this.delegate = checkNotNull(delegate);
508    }
509
510    @Override
511    public final boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
512      return delegate.awaitTermination(timeout, unit);
513    }
514
515    @Override
516    public final boolean isShutdown() {
517      return delegate.isShutdown();
518    }
519
520    @Override
521    public final boolean isTerminated() {
522      return delegate.isTerminated();
523    }
524
525    @Override
526    public final void shutdown() {
527      delegate.shutdown();
528    }
529
530    @Override
531    public final List<Runnable> shutdownNow() {
532      return delegate.shutdownNow();
533    }
534
535    @Override
536    public final void execute(Runnable command) {
537      delegate.execute(command);
538    }
539  }
540
541  @GwtIncompatible // TODO
542  private static final class ScheduledListeningDecorator extends ListeningDecorator
543      implements ListeningScheduledExecutorService {
544    @SuppressWarnings("hiding")
545    final ScheduledExecutorService delegate;
546
547    ScheduledListeningDecorator(ScheduledExecutorService delegate) {
548      super(delegate);
549      this.delegate = checkNotNull(delegate);
550    }
551
552    @Override
553    public ListenableScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
554      TrustedListenableFutureTask<Void> task = TrustedListenableFutureTask.create(command, null);
555      ScheduledFuture<?> scheduled = delegate.schedule(task, delay, unit);
556      return new ListenableScheduledTask<>(task, scheduled);
557    }
558
559    @Override
560    public <V> ListenableScheduledFuture<V> schedule(
561        Callable<V> callable, long delay, TimeUnit unit) {
562      TrustedListenableFutureTask<V> task = TrustedListenableFutureTask.create(callable);
563      ScheduledFuture<?> scheduled = delegate.schedule(task, delay, unit);
564      return new ListenableScheduledTask<V>(task, scheduled);
565    }
566
567    @Override
568    public ListenableScheduledFuture<?> scheduleAtFixedRate(
569        Runnable command, long initialDelay, long period, TimeUnit unit) {
570      NeverSuccessfulListenableFutureTask task = new NeverSuccessfulListenableFutureTask(command);
571      ScheduledFuture<?> scheduled = delegate.scheduleAtFixedRate(task, initialDelay, period, unit);
572      return new ListenableScheduledTask<>(task, scheduled);
573    }
574
575    @Override
576    public ListenableScheduledFuture<?> scheduleWithFixedDelay(
577        Runnable command, long initialDelay, long delay, TimeUnit unit) {
578      NeverSuccessfulListenableFutureTask task = new NeverSuccessfulListenableFutureTask(command);
579      ScheduledFuture<?> scheduled =
580          delegate.scheduleWithFixedDelay(task, initialDelay, delay, unit);
581      return new ListenableScheduledTask<>(task, scheduled);
582    }
583
584    private static final class ListenableScheduledTask<V>
585        extends SimpleForwardingListenableFuture<V> implements ListenableScheduledFuture<V> {
586
587      private final ScheduledFuture<?> scheduledDelegate;
588
589      public ListenableScheduledTask(
590          ListenableFuture<V> listenableDelegate, ScheduledFuture<?> scheduledDelegate) {
591        super(listenableDelegate);
592        this.scheduledDelegate = scheduledDelegate;
593      }
594
595      @Override
596      public boolean cancel(boolean mayInterruptIfRunning) {
597        boolean cancelled = super.cancel(mayInterruptIfRunning);
598        if (cancelled) {
599          // Unless it is cancelled, the delegate may continue being scheduled
600          scheduledDelegate.cancel(mayInterruptIfRunning);
601
602          // TODO(user): Cancel "this" if "scheduledDelegate" is cancelled.
603        }
604        return cancelled;
605      }
606
607      @Override
608      public long getDelay(TimeUnit unit) {
609        return scheduledDelegate.getDelay(unit);
610      }
611
612      @Override
613      public int compareTo(Delayed other) {
614        return scheduledDelegate.compareTo(other);
615      }
616    }
617
618    @GwtIncompatible // TODO
619    private static final class NeverSuccessfulListenableFutureTask extends AbstractFuture<Void>
620        implements Runnable {
621      private final Runnable delegate;
622
623      public NeverSuccessfulListenableFutureTask(Runnable delegate) {
624        this.delegate = checkNotNull(delegate);
625      }
626
627      @Override
628      public void run() {
629        try {
630          delegate.run();
631        } catch (Throwable t) {
632          setException(t);
633          throw Throwables.propagate(t);
634        }
635      }
636    }
637  }
638
639  /*
640   * This following method is a modified version of one found in
641   * http://gee.cs.oswego.edu/cgi-bin/viewcvs.cgi/jsr166/src/test/tck/AbstractExecutorServiceTest.java?revision=1.30
642   * which contained the following notice:
643   *
644   * Written by Doug Lea with assistance from members of JCP JSR-166 Expert Group and released to
645   * the public domain, as explained at http://creativecommons.org/publicdomain/zero/1.0/
646   *
647   * Other contributors include Andrew Wright, Jeffrey Hayes, Pat Fisher, Mike Judd.
648   */
649
650  /**
651   * An implementation of {@link ExecutorService#invokeAny} for {@link ListeningExecutorService}
652   * implementations.
653   */
654  @GwtIncompatible static <T> T invokeAnyImpl(
655      ListeningExecutorService executorService,
656      Collection<? extends Callable<T>> tasks,
657      boolean timed,
658      long timeout,
659      TimeUnit unit)
660      throws InterruptedException, ExecutionException, TimeoutException {
661    checkNotNull(executorService);
662    checkNotNull(unit);
663    int ntasks = tasks.size();
664    checkArgument(ntasks > 0);
665    List<Future<T>> futures = Lists.newArrayListWithCapacity(ntasks);
666    BlockingQueue<Future<T>> futureQueue = Queues.newLinkedBlockingQueue();
667    long timeoutNanos = unit.toNanos(timeout);
668
669    // For efficiency, especially in executors with limited
670    // parallelism, check to see if previously submitted tasks are
671    // done before submitting more of them. This interleaving
672    // plus the exception mechanics account for messiness of main
673    // loop.
674
675    try {
676      // Record exceptions so that if we fail to obtain any
677      // result, we can throw the last exception we got.
678      ExecutionException ee = null;
679      long lastTime = timed ? System.nanoTime() : 0;
680      Iterator<? extends Callable<T>> it = tasks.iterator();
681
682      futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
683      --ntasks;
684      int active = 1;
685
686      while (true) {
687        Future<T> f = futureQueue.poll();
688        if (f == null) {
689          if (ntasks > 0) {
690            --ntasks;
691            futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
692            ++active;
693          } else if (active == 0) {
694            break;
695          } else if (timed) {
696            f = futureQueue.poll(timeoutNanos, TimeUnit.NANOSECONDS);
697            if (f == null) {
698              throw new TimeoutException();
699            }
700            long now = System.nanoTime();
701            timeoutNanos -= now - lastTime;
702            lastTime = now;
703          } else {
704            f = futureQueue.take();
705          }
706        }
707        if (f != null) {
708          --active;
709          try {
710            return f.get();
711          } catch (ExecutionException eex) {
712            ee = eex;
713          } catch (RuntimeException rex) {
714            ee = new ExecutionException(rex);
715          }
716        }
717      }
718
719      if (ee == null) {
720        ee = new ExecutionException(null);
721      }
722      throw ee;
723    } finally {
724      for (Future<T> f : futures) {
725        f.cancel(true);
726      }
727    }
728  }
729
730  /**
731   * Submits the task and adds a listener that adds the future to {@code queue} when it completes.
732   */
733  @GwtIncompatible // TODO
734  private static <T> ListenableFuture<T> submitAndAddQueueListener(
735      ListeningExecutorService executorService,
736      Callable<T> task,
737      final BlockingQueue<Future<T>> queue) {
738    final ListenableFuture<T> future = executorService.submit(task);
739    future.addListener(
740        new Runnable() {
741          @Override
742          public void run() {
743            queue.add(future);
744          }
745        },
746        directExecutor());
747    return future;
748  }
749
750  /**
751   * Returns a default thread factory used to create new threads.
752   *
753   * <p>On AppEngine, returns {@code ThreadManager.currentRequestThreadFactory()}. Otherwise,
754   * returns {@link Executors#defaultThreadFactory()}.
755   *
756   * @since 14.0
757   */
758  @Beta
759  @GwtIncompatible // concurrency
760  public static ThreadFactory platformThreadFactory() {
761    if (!isAppEngine()) {
762      return Executors.defaultThreadFactory();
763    }
764    try {
765      return (ThreadFactory)
766          Class.forName("com.google.appengine.api.ThreadManager")
767              .getMethod("currentRequestThreadFactory")
768              .invoke(null);
769    } catch (IllegalAccessException | ClassNotFoundException | NoSuchMethodException e) {
770      throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e);
771    } catch (InvocationTargetException e) {
772      throw Throwables.propagate(e.getCause());
773    }
774  }
775
776  @GwtIncompatible // TODO
777  private static boolean isAppEngine() {
778    if (System.getProperty("com.google.appengine.runtime.environment") == null) {
779      return false;
780    }
781    try {
782      // If the current environment is null, we're not inside AppEngine.
783      return Class.forName("com.google.apphosting.api.ApiProxy")
784              .getMethod("getCurrentEnvironment")
785              .invoke(null)
786          != null;
787    } catch (ClassNotFoundException e) {
788      // If ApiProxy doesn't exist, we're not on AppEngine at all.
789      return false;
790    } catch (InvocationTargetException e) {
791      // If ApiProxy throws an exception, we're not in a proper AppEngine environment.
792      return false;
793    } catch (IllegalAccessException e) {
794      // If the method isn't accessible, we're not on a supported version of AppEngine;
795      return false;
796    } catch (NoSuchMethodException e) {
797      // If the method doesn't exist, we're not on a supported version of AppEngine;
798      return false;
799    }
800  }
801
802  /**
803   * Creates a thread using {@link #platformThreadFactory}, and sets its name to {@code name} unless
804   * changing the name is forbidden by the security manager.
805   */
806  @GwtIncompatible // concurrency
807  static Thread newThread(String name, Runnable runnable) {
808    checkNotNull(name);
809    checkNotNull(runnable);
810    Thread result = platformThreadFactory().newThread(runnable);
811    try {
812      result.setName(name);
813    } catch (SecurityException e) {
814      // OK if we can't set the name in this environment.
815    }
816    return result;
817  }
818
819  // TODO(lukes): provide overloads for ListeningExecutorService? ListeningScheduledExecutorService?
820  // TODO(lukes): provide overloads that take constant strings? Function<Runnable, String>s to
821  // calculate names?
822
823  /**
824   * Creates an {@link Executor} that renames the {@link Thread threads} that its tasks run in.
825   *
826   * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
827   * right before each task is run. The renaming is best effort, if a {@link SecurityManager}
828   * prevents the renaming then it will be skipped but the tasks will still execute.
829   *
830   *
831   * @param executor The executor to decorate
832   * @param nameSupplier The source of names for each task
833   */
834  @GwtIncompatible // concurrency
835  static Executor renamingDecorator(final Executor executor, final Supplier<String> nameSupplier) {
836    checkNotNull(executor);
837    checkNotNull(nameSupplier);
838    if (isAppEngine()) {
839      // AppEngine doesn't support thread renaming, so don't even try
840      return executor;
841    }
842    return new Executor() {
843      @Override
844      public void execute(Runnable command) {
845        executor.execute(Callables.threadRenaming(command, nameSupplier));
846      }
847    };
848  }
849
850  /**
851   * Creates an {@link ExecutorService} that renames the {@link Thread threads} that its tasks run
852   * in.
853   *
854   * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
855   * right before each task is run. The renaming is best effort, if a {@link SecurityManager}
856   * prevents the renaming then it will be skipped but the tasks will still execute.
857   *
858   *
859   * @param service The executor to decorate
860   * @param nameSupplier The source of names for each task
861   */
862  @GwtIncompatible // concurrency
863  static ExecutorService renamingDecorator(
864      final ExecutorService service, final Supplier<String> nameSupplier) {
865    checkNotNull(service);
866    checkNotNull(nameSupplier);
867    if (isAppEngine()) {
868      // AppEngine doesn't support thread renaming, so don't even try.
869      return service;
870    }
871    return new WrappingExecutorService(service) {
872      @Override
873      protected <T> Callable<T> wrapTask(Callable<T> callable) {
874        return Callables.threadRenaming(callable, nameSupplier);
875      }
876
877      @Override
878      protected Runnable wrapTask(Runnable command) {
879        return Callables.threadRenaming(command, nameSupplier);
880      }
881    };
882  }
883
884  /**
885   * Creates a {@link ScheduledExecutorService} that renames the {@link Thread threads} that its
886   * tasks run in.
887   *
888   * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
889   * right before each task is run. The renaming is best effort, if a {@link SecurityManager}
890   * prevents the renaming then it will be skipped but the tasks will still execute.
891   *
892   *
893   * @param service The executor to decorate
894   * @param nameSupplier The source of names for each task
895   */
896  @GwtIncompatible // concurrency
897  static ScheduledExecutorService renamingDecorator(
898      final ScheduledExecutorService service, final Supplier<String> nameSupplier) {
899    checkNotNull(service);
900    checkNotNull(nameSupplier);
901    if (isAppEngine()) {
902      // AppEngine doesn't support thread renaming, so don't even try.
903      return service;
904    }
905    return new WrappingScheduledExecutorService(service) {
906      @Override
907      protected <T> Callable<T> wrapTask(Callable<T> callable) {
908        return Callables.threadRenaming(callable, nameSupplier);
909      }
910
911      @Override
912      protected Runnable wrapTask(Runnable command) {
913        return Callables.threadRenaming(command, nameSupplier);
914      }
915    };
916  }
917
918  /**
919   * Shuts down the given executor service gradually, first disabling new submissions and later, if
920   * necessary, cancelling remaining tasks.
921   *
922   * <p>The method takes the following steps:
923   *
924   * <ol>
925   *   <li>calls {@link ExecutorService#shutdown()}, disabling acceptance of new submitted tasks.
926   *   <li>awaits executor service termination for half of the specified timeout.
927   *   <li>if the timeout expires, it calls {@link ExecutorService#shutdownNow()}, cancelling
928   *       pending tasks and interrupting running tasks.
929   *   <li>awaits executor service termination for the other half of the specified timeout.
930   * </ol>
931   *
932   * <p>If, at any step of the process, the calling thread is interrupted, the method calls {@link
933   * ExecutorService#shutdownNow()} and returns.
934   *
935   * @param service the {@code ExecutorService} to shut down
936   * @param timeout the maximum time to wait for the {@code ExecutorService} to terminate
937   * @param unit the time unit of the timeout argument
938   * @return {@code true} if the {@code ExecutorService} was terminated successfully, {@code false}
939   *     if the call timed out or was interrupted
940   * @since 17.0
941   */
942  @Beta
943  @CanIgnoreReturnValue
944  @GwtIncompatible // concurrency
945  public static boolean shutdownAndAwaitTermination(
946      ExecutorService service, long timeout, TimeUnit unit) {
947    long halfTimeoutNanos = unit.toNanos(timeout) / 2;
948    // Disable new tasks from being submitted
949    service.shutdown();
950    try {
951      // Wait for half the duration of the timeout for existing tasks to terminate
952      if (!service.awaitTermination(halfTimeoutNanos, TimeUnit.NANOSECONDS)) {
953        // Cancel currently executing tasks
954        service.shutdownNow();
955        // Wait the other half of the timeout for tasks to respond to being cancelled
956        service.awaitTermination(halfTimeoutNanos, TimeUnit.NANOSECONDS);
957      }
958    } catch (InterruptedException ie) {
959      // Preserve interrupt status
960      Thread.currentThread().interrupt();
961      // (Re-)Cancel if current thread also interrupted
962      service.shutdownNow();
963    }
964    return service.isTerminated();
965  }
966
967  /**
968   * Returns an Executor that will propagate {@link RejectedExecutionException} from the delegate
969   * executor to the given {@code future}.
970   *
971   * <p>Note, the returned executor can only be used once.
972   */
973  static Executor rejectionPropagatingExecutor(
974      final Executor delegate, final AbstractFuture<?> future) {
975    checkNotNull(delegate);
976    checkNotNull(future);
977    if (delegate == directExecutor()) {
978      // directExecutor() cannot throw RejectedExecutionException
979      return delegate;
980    }
981    return new Executor() {
982      boolean thrownFromDelegate = true;
983
984      @Override
985      public void execute(final Runnable command) {
986        try {
987          delegate.execute(
988              new Runnable() {
989                @Override
990                public void run() {
991                  thrownFromDelegate = false;
992                  command.run();
993                }
994              });
995        } catch (RejectedExecutionException e) {
996          if (thrownFromDelegate) {
997            // wrap exception?
998            future.setException(e);
999          }
1000          // otherwise it must have been thrown from a transitive call and the delegate runnable
1001          // should have handled it.
1002        }
1003      }
1004    };
1005  }
1006}