    001    /*
    002     * Copyright (C) 2007 The Guava Authors
    003     *
    004     * Licensed under the Apache License, Version 2.0 (the "License");
    005     * you may not use this file except in compliance with the License.
    006     * You may obtain a copy of the License at
    007     *
    008     * http://www.apache.org/licenses/LICENSE-2.0
    009     *
    010     * Unless required by applicable law or agreed to in writing, software
    011     * distributed under the License is distributed on an "AS IS" BASIS,
    012     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    013     * See the License for the specific language governing permissions and
    014     * limitations under the License.
    015     */
    016
    017    package com.google.common.util.concurrent;
    018
    019    import static com.google.common.base.Preconditions.checkNotNull;
    020
    021    import com.google.common.annotations.Beta;
    022
    023    import java.util.Collections;
    024    import java.util.List;
    025    import java.util.concurrent.Callable;
    026    import java.util.concurrent.ExecutorService;
    027    import java.util.concurrent.Executors;
    028    import java.util.concurrent.RejectedExecutionException;
    029    import java.util.concurrent.ScheduledExecutorService;
    030    import java.util.concurrent.ScheduledFuture;
    031    import java.util.concurrent.ScheduledThreadPoolExecutor;
    032    import java.util.concurrent.ThreadFactory;
    033    import java.util.concurrent.ThreadPoolExecutor;
    034    import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
    035    import java.util.concurrent.TimeUnit;
    036    import java.util.concurrent.locks.Condition;
    037    import java.util.concurrent.locks.Lock;
    038    import java.util.concurrent.locks.ReentrantLock;
    039
    040    /**
    041     * Factory and utility methods for {@link java.util.concurrent.Executor}, {@link
    042     * ExecutorService}, and {@link ThreadFactory}.
    043     *
    044     * @author Eric Fellheimer
    045     * @author Kyle Littlefield
    046     * @author Justin Mahoney
    047     * @since 3.0
    048     */
    049    public final class MoreExecutors {
    050      private MoreExecutors() {}
    051
    052      /**
    053       * Converts the given ThreadPoolExecutor into an ExecutorService that exits
    054       * when the application is complete.  It does so by using daemon threads and
    055       * adding a shutdown hook to wait for their completion.
    056       *
    057       * <p>This is mainly for fixed thread pools.
    058       * See {@link Executors#newFixedThreadPool(int)}.
    059       *
    060       * @param executor the executor to modify to make sure it exits when the
    061       *        application is finished
    062       * @param terminationTimeout how long to wait for the executor to
    063       *        finish before terminating the JVM
    064       * @param timeUnit unit of time for the time parameter
    065       * @return an unmodifiable version of the input which will not hang the JVM
    066       */
    067      @Beta
    068      public static ExecutorService getExitingExecutorService(
    069          ThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
    070        executor.setThreadFactory(new ThreadFactoryBuilder()
    071            .setDaemon(true)
    072            .setThreadFactory(executor.getThreadFactory())
    073            .build());
    074
    075        ExecutorService service = Executors.unconfigurableExecutorService(executor);
    076
    077        addDelayedShutdownHook(service, terminationTimeout, timeUnit);
    078
    079        return service;
    080      }
    081
    082      /**
    083       * Converts the given ScheduledThreadPoolExecutor into a
    084       * ScheduledExecutorService that exits when the application is complete.  It
    085       * does so by using daemon threads and adding a shutdown hook to wait for
    086       * their completion.
    087       *
    088       * <p>This is mainly for fixed thread pools.
    089       * See {@link Executors#newScheduledThreadPool(int)}.
    090       *
    091       * @param executor the executor to modify to make sure it exits when the
    092       *        application is finished
    093       * @param terminationTimeout how long to wait for the executor to
    094       *        finish before terminating the JVM
    095       * @param timeUnit unit of time for the time parameter
    096       * @return an unmodifiable version of the input which will not hang the JVM
    097       */
    098      @Beta
    099      public static ScheduledExecutorService getExitingScheduledExecutorService(
    100          ScheduledThreadPoolExecutor executor, long terminationTimeout,
    101          TimeUnit timeUnit) {
    102        executor.setThreadFactory(new ThreadFactoryBuilder()
    103            .setDaemon(true)
    104            .setThreadFactory(executor.getThreadFactory())
    105            .build());
    106
    107        ScheduledExecutorService service =
    108            Executors.unconfigurableScheduledExecutorService(executor);
    109
    110        addDelayedShutdownHook(service, terminationTimeout, timeUnit);
    111
    112        return service;
    113      }
    114
    115      /**
    116       * Add a shutdown hook to wait for thread completion in the given
    117       * {@link ExecutorService service}.  This is useful if the given service uses
    118       * daemon threads, and we want to keep the JVM from exiting immediately on
    119       * shutdown, instead giving these daemon threads a chance to terminate
    120       * normally.
    121       * @param service ExecutorService which uses daemon threads
    122       * @param terminationTimeout how long to wait for the executor to finish
    123       *        before terminating the JVM
    124       * @param timeUnit unit of time for the time parameter
    125       */
    126      @Beta
    127      public static void addDelayedShutdownHook(
    128          final ExecutorService service, final long terminationTimeout,
    129          final TimeUnit timeUnit) {
    130        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
    131          @Override
    132          public void run() {
    133            try {
    134              // We'd like to log progress and failures that may arise in the
    135              // following code, but unfortunately the behavior of logging
    136              // is undefined in shutdown hooks.
    137              // This is because the logging code installs a shutdown hook of its
    138              // own. See Cleaner class inside {@link LogManager}.
    139              service.shutdown();
    140              service.awaitTermination(terminationTimeout, timeUnit);
    141            } catch (InterruptedException ignored) {
    142              // We're shutting down anyway, so just ignore.
    143            }
    144          }
    145        }));
    146      }
    147
    148      /**
    149       * Converts the given ThreadPoolExecutor into an ExecutorService that exits
    150       * when the application is complete.  It does so by using daemon threads and
    151       * adding a shutdown hook to wait for their completion.
    152       *
    153       * <p>This method waits 120 seconds before continuing with JVM termination,
    154       * even if the executor has not finished its work.
    155       *
    156       * <p>This is mainly for fixed thread pools.
    157       * See {@link Executors#newFixedThreadPool(int)}.
    158       *
    159       * @param executor the executor to modify to make sure it exits when the
    160       *        application is finished
    161       * @return an unmodifiable version of the input which will not hang the JVM
    162       */
    163      @Beta
    164      public static ExecutorService getExitingExecutorService(
    165          ThreadPoolExecutor executor) {
    166        return getExitingExecutorService(executor, 120, TimeUnit.SECONDS);
    167      }
    168
    169      /**
    170       * Converts the given ThreadPoolExecutor into a ScheduledExecutorService that
    171       * exits when the application is complete.  It does so by using daemon threads
    172       * and adding a shutdown hook to wait for their completion.
    173       *
    174       * <p>This method waits 120 seconds before continuing with JVM termination,
    175       * even if the executor has not finished its work.
    176       *
    177       * <p>This is mainly for fixed thread pools.
    178       * See {@link Executors#newScheduledThreadPool(int)}.
    179       *
    180       * @param executor the executor to modify to make sure it exits when the
    181       *        application is finished
    182       * @return an unmodifiable version of the input which will not hang the JVM
    183       */
    184      @Beta
    185      public static ScheduledExecutorService getExitingScheduledExecutorService(
    186          ScheduledThreadPoolExecutor executor) {
    187        return getExitingScheduledExecutorService(executor, 120, TimeUnit.SECONDS);
    188      }
    189
    190      /**
    191       * Creates an executor service that runs each task in the thread
    192       * that invokes {@code execute/submit}, as in {@link CallerRunsPolicy}  This
    193       * applies both to individually submitted tasks and to collections of tasks
    194       * submitted via {@code invokeAll} or {@code invokeAny}.  In the latter case,
    195       * tasks will run serially on the calling thread.  Tasks are run to
    196       * completion before a {@code Future} is returned to the caller (unless the
    197       * executor has been shutdown).
    198       *
    199       * <p>Although all tasks are immediately executed in the thread that
    200       * submitted the task, this {@code ExecutorService} imposes a small
    201       * locking overhead on each task submission in order to implement shutdown
    202       * and termination behavior.
    203       *
    204       * <p>The implementation deviates from the {@code ExecutorService}
    205       * specification with regards to the {@code shutdownNow} method.  First,
    206       * "best-effort" with regards to canceling running tasks is implemented
    207       * as "no-effort".  No interrupts or other attempts are made to stop
    208       * threads executing tasks.  Second, the returned list will always be empty,
    209       * as any submitted task is considered to have started execution.
    210       * This applies also to tasks given to {@code invokeAll} or {@code invokeAny}
    211       * which are pending serial execution, even the subset of the tasks that
    212       * have not yet started execution.  It is unclear from the
    213       * {@code ExecutorService} specification if these should be included, and
    214       * it's much easier to implement the interpretation that they not be.
    215       * Finally, a call to {@code shutdown} or {@code shutdownNow} may result
    216       * in concurrent calls to {@code invokeAll/invokeAny} throwing
    217       * RejectedExecutionException, although a subset of the tasks may already
    218       * have been executed.
    219       *
    220       * @since 10.0 (<a href="http://code.google.com/p/guava-libraries/wiki/Compatibility"
    221       *        >mostly source-compatible</a> since 3.0)
    222       */
    223      public static ListeningExecutorService sameThreadExecutor() {
    224        return new SameThreadExecutorService();
    225      }
    226
    227      // See sameThreadExecutor javadoc for behavioral notes.
    228      private static class SameThreadExecutorService
    229          extends AbstractListeningExecutorService {
    230        /**
    231         * Lock used whenever accessing the state variables
    232         * (runningTasks, shutdown, terminationCondition) of the executor
    233         */
    234        private final Lock lock = new ReentrantLock();
    235
    236        /** Signaled after the executor is shutdown and running tasks are done */
    237        private final Condition termination = lock.newCondition();
    238
    239        /*
    240         * Conceptually, these two variables describe the executor being in
    241         * one of three states:
    242         *   - Active: shutdown == false
    243         *   - Shutdown: runningTasks > 0 and shutdown == true
    244         *   - Terminated: runningTasks == 0 and shutdown == true
    245         */
    246        private int runningTasks = 0;
    247        private boolean shutdown = false;
    248
    249        @Override
    250        public void execute(Runnable command) {
    251          startTask();
    252          try {
    253            command.run();
    254          } finally {
    255            endTask();
    256          }
    257        }
    258
    259        @Override
    260        public boolean isShutdown() {
    261          lock.lock();
    262          try {
    263            return shutdown;
    264          } finally {
    265            lock.unlock();
    266          }
    267        }
    268
    269        @Override
    270        public void shutdown() {
    271          lock.lock();
    272          try {
    273            shutdown = true;
    274          } finally {
    275            lock.unlock();
    276          }
    277        }
    278
    279        // See sameThreadExecutor javadoc for unusual behavior of this method.
    280        @Override
    281        public List<Runnable> shutdownNow() {
    282          shutdown();
    283          return Collections.emptyList();
    284        }
    285
    286        @Override
    287        public boolean isTerminated() {
    288          lock.lock();
    289          try {
    290            return shutdown && runningTasks == 0;
    291          } finally {
    292            lock.unlock();
    293          }
    294        }
    295
    296        @Override
    297        public boolean awaitTermination(long timeout, TimeUnit unit)
    298            throws InterruptedException {
    299          long nanos = unit.toNanos(timeout);
    300          lock.lock();
    301          try {
    302            for (;;) {
    303              if (isTerminated()) {
    304                return true;
    305              } else if (nanos <= 0) {
    306                return false;
    307              } else {
    308                nanos = termination.awaitNanos(nanos);
    309              }
    310            }
    311          } finally {
    312            lock.unlock();
    313          }
    314        }
    315
    316        /**
    317         * Checks if the executor has been shut down and increments the running
    318         * task count.
    319         *
    320         * @throws RejectedExecutionException if the executor has been previously
    321         *         shutdown
    322         */
    323        private void startTask() {
    324          lock.lock();
    325          try {
    326            if (isShutdown()) {
    327              throw new RejectedExecutionException("Executor already shutdown");
    328            }
    329            runningTasks++;
    330          } finally {
    331            lock.unlock();
    332          }
    333        }
    334
    335        /**
    336         * Decrements the running task count.
    337         */
    338        private void endTask() {
    339          lock.lock();
    340          try {
    341            runningTasks--;
    342            if (isTerminated()) {
    343              termination.signalAll();
    344            }
    345          } finally {
    346            lock.unlock();
    347          }
    348        }
    349      }
    350
    351      /**
    352       * Creates an {@link ExecutorService} whose {@code submit} and {@code
    353       * invokeAll} methods submit {@link ListenableFutureTask} instances to the
    354       * given delegate executor. Those methods, as well as {@code execute} and
    355       * {@code invokeAny}, are implemented in terms of calls to {@code
    356       * delegate.execute}. All other methods are forwarded unchanged to the
    357       * delegate. This implies that the returned {@code ListeningExecutorService}
    358       * never calls the delegate's {@code submit}, {@code invokeAll}, and {@code
    359       * invokeAny} methods, so any special handling of tasks must be implemented in
    360       * the delegate's {@code execute} method or by wrapping the returned {@code
    361       * ListeningExecutorService}.
    362       *
    363       * <p>If the delegate executor was already an instance of {@code
    364       * ListeningExecutorService}, it is returned untouched, and the rest of this
    365       * documentation does not apply.
    366       *
    367       * @since 10.0
    368       */
    369      public static ListeningExecutorService listeningDecorator(
    370          ExecutorService delegate) {
    371        return (delegate instanceof ListeningExecutorService)
    372            ? (ListeningExecutorService) delegate
    373            : (delegate instanceof ScheduledExecutorService)
    374            ? new ScheduledListeningDecorator((ScheduledExecutorService) delegate)
    375            : new ListeningDecorator(delegate);
    376      }
    377
    378      /**
    379       * Creates a {@link ScheduledExecutorService} whose {@code submit} and {@code
    380       * invokeAll} methods submit {@link ListenableFutureTask} instances to the
    381       * given delegate executor. Those methods, as well as {@code execute} and
    382       * {@code invokeAny}, are implemented in terms of calls to {@code
    383       * delegate.execute}. All other methods are forwarded unchanged to the
    384       * delegate. This implies that the returned {@code
    385       * SchedulingListeningExecutorService} never calls the delegate's {@code
    386       * submit}, {@code invokeAll}, and {@code invokeAny} methods, so any special
    387       * handling of tasks must be implemented in the delegate's {@code execute}
    388       * method or by wrapping the returned {@code
    389       * SchedulingListeningExecutorService}.
    390       *
    391       * <p>If the delegate executor was already an instance of {@code
    392       * ListeningScheduledExecutorService}, it is returned untouched, and the rest
    393       * of this documentation does not apply.
    394       *
    395       * @since 10.0
    396       */
    397      public static ListeningScheduledExecutorService listeningDecorator(
    398          ScheduledExecutorService delegate) {
    399        return (delegate instanceof ListeningScheduledExecutorService)
    400            ? (ListeningScheduledExecutorService) delegate
    401            : new ScheduledListeningDecorator(delegate);
    402      }
    403
    404      private static class ListeningDecorator
    405          extends AbstractListeningExecutorService {
    406        final ExecutorService delegate;
    407
    408        ListeningDecorator(ExecutorService delegate) {
    409          this.delegate = checkNotNull(delegate);
    410        }
    411
    412        @Override
    413        public boolean awaitTermination(long timeout, TimeUnit unit)
    414            throws InterruptedException {
    415          return delegate.awaitTermination(timeout, unit);
    416        }
    417
    418        @Override
    419        public boolean isShutdown() {
    420          return delegate.isShutdown();
    421        }
    422
    423        @Override
    424        public boolean isTerminated() {
    425          return delegate.isTerminated();
    426        }
    427
    428        @Override
    429        public void shutdown() {
    430          delegate.shutdown();
    431        }
    432
    433        @Override
    434        public List<Runnable> shutdownNow() {
    435          return delegate.shutdownNow();
    436        }
    437
    438        @Override
    439        public void execute(Runnable command) {
    440          delegate.execute(command);
    441        }
    442      }
    443
    444      private static class ScheduledListeningDecorator
    445          extends ListeningDecorator implements ListeningScheduledExecutorService {
    446        final ScheduledExecutorService delegate;
    447
    448        ScheduledListeningDecorator(ScheduledExecutorService delegate) {
    449          super(delegate);
    450          this.delegate = checkNotNull(delegate);
    451        }
    452
    453        @Override
    454        public ScheduledFuture<?> schedule(
    455            Runnable command, long delay, TimeUnit unit) {
    456          return delegate.schedule(command, delay, unit);
    457        }
    458
    459        @Override
    460        public <V> ScheduledFuture<V> schedule(
    461            Callable<V> callable, long delay, TimeUnit unit) {
    462          return delegate.schedule(callable, delay, unit);
    463        }
    464
    465        @Override
    466        public ScheduledFuture<?> scheduleAtFixedRate(
    467            Runnable command, long initialDelay, long period, TimeUnit unit) {
    468          return delegate.scheduleAtFixedRate(command, initialDelay, period, unit);
    469        }
    470
    471        @Override
    472        public ScheduledFuture<?> scheduleWithFixedDelay(
    473            Runnable command, long initialDelay, long delay, TimeUnit unit) {
    474          return delegate.scheduleWithFixedDelay(
    475              command, initialDelay, delay, unit);
    476        }
    477      }
    478    }