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
017package com.google.common.util.concurrent;
018
019import com.google.common.annotations.Beta;
020
021import java.util.Collections;
022import java.util.List;
023import java.util.concurrent.AbstractExecutorService;
024import java.util.concurrent.ExecutorService;
025import java.util.concurrent.Executors;
026import java.util.concurrent.RejectedExecutionException;
027import java.util.concurrent.ScheduledExecutorService;
028import java.util.concurrent.ScheduledThreadPoolExecutor;
029import java.util.concurrent.ThreadFactory;
030import java.util.concurrent.ThreadPoolExecutor;
031import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
032import java.util.concurrent.TimeUnit;
033import java.util.concurrent.locks.Condition;
034import java.util.concurrent.locks.Lock;
035import java.util.concurrent.locks.ReentrantLock;
036
037/**
038 * Factory and utility methods for {@link java.util.concurrent.Executor}, {@link
039 * ExecutorService}, and {@link ThreadFactory}.
040 *
041 * @author Eric Fellheimer
042 * @author Kyle Littlefield
043 * @author Justin Mahoney
044 * @since 3
045 */
046@Beta
047public final class MoreExecutors {
048  private MoreExecutors() {}
049
050  /**
051   * Converts the given ThreadPoolExecutor into an ExecutorService that exits
052   * when the application is complete.  It does so by using daemon threads and
053   * adding a shutdown hook to wait for their completion.
054   *
055   * <p>This is mainly for fixed thread pools.
056   * See {@link Executors#newFixedThreadPool(int)}.
057   *
058   * @param executor the executor to modify to make sure it exits when the
059   *        application is finished
060   * @param terminationTimeout how long to wait for the executor to
061   *        finish before terminating the JVM
062   * @param timeUnit unit of time for the time parameter
063   * @return an unmodifiable version of the input which will not hang the JVM
064   */
065  public static ExecutorService getExitingExecutorService(
066      ThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
067    executor.setThreadFactory(new ThreadFactoryBuilder()
068        .setDaemon(true)
069        .setThreadFactory(executor.getThreadFactory())
070        .build());
071
072    ExecutorService service = Executors.unconfigurableExecutorService(executor);
073
074    addDelayedShutdownHook(service, terminationTimeout, timeUnit);
075
076    return service;
077  }
078
079  /**
080   * Converts the given ScheduledThreadPoolExecutor into a
081   * ScheduledExecutorService that exits when the application is complete.  It
082   * does so by using daemon threads and adding a shutdown hook to wait for
083   * their completion.
084   *
085   * <p>This is mainly for fixed thread pools.
086   * See {@link Executors#newScheduledThreadPool(int)}.
087   *
088   * @param executor the executor to modify to make sure it exits when the
089   *        application is finished
090   * @param terminationTimeout how long to wait for the executor to
091   *        finish before terminating the JVM
092   * @param timeUnit unit of time for the time parameter
093   * @return an unmodifiable version of the input which will not hang the JVM
094   */
095  public static ScheduledExecutorService getExitingScheduledExecutorService(
096      ScheduledThreadPoolExecutor executor, long terminationTimeout,
097      TimeUnit timeUnit) {
098    executor.setThreadFactory(new ThreadFactoryBuilder()
099        .setDaemon(true)
100        .setThreadFactory(executor.getThreadFactory())
101        .build());
102
103    ScheduledExecutorService service =
104        Executors.unconfigurableScheduledExecutorService(executor);
105
106    addDelayedShutdownHook(service, terminationTimeout, timeUnit);
107
108    return service;
109  }
110
111  /**
112   * Add a shutdown hook to wait for thread completion in the given
113   * {@link ExecutorService service}.  This is useful if the given service uses
114   * daemon threads, and we want to keep the JVM from exiting immediately on
115   * shutdown, instead giving these daemon threads a chance to terminate
116   * normally.
117   * @param service ExecutorService which uses daemon threads
118   * @param terminationTimeout how long to wait for the executor to finish
119   *        before terminating the JVM
120   * @param timeUnit unit of time for the time parameter
121   */
122  public static void addDelayedShutdownHook(
123      final ExecutorService service, final long terminationTimeout,
124      final TimeUnit timeUnit) {
125    Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
126      @Override
127      public void run() {
128        try {
129          // We'd like to log progress and failures that may arise in the
130          // following code, but unfortunately the behavior of logging
131          // is undefined in shutdown hooks.
132          // This is because the logging code installs a shutdown hook of its
133          // own. See Cleaner class inside {@link LogManager}.
134          service.shutdown();
135          service.awaitTermination(terminationTimeout, timeUnit);
136        } catch (InterruptedException ignored) {
137          // We're shutting down anyway, so just ignore.
138        }
139      }
140    }));
141  }
142
143  /**
144   * Converts the given ThreadPoolExecutor into an ExecutorService that exits
145   * when the application is complete.  It does so by using daemon threads and
146   * adding a shutdown hook to wait for their completion.
147   *
148   * <p>This method waits 120 seconds before continuing with JVM termination,
149   * even if the executor has not finished its work.
150   *
151   * <p>This is mainly for fixed thread pools.
152   * See {@link Executors#newFixedThreadPool(int)}.
153   *
154   * @param executor the executor to modify to make sure it exits when the
155   *        application is finished
156   * @return an unmodifiable version of the input which will not hang the JVM
157   */
158  public static ExecutorService getExitingExecutorService(
159      ThreadPoolExecutor executor) {
160    return getExitingExecutorService(executor, 120, TimeUnit.SECONDS);
161  }
162
163  /**
164   * Converts the given ThreadPoolExecutor into a ScheduledExecutorService that
165   * exits when the application is complete.  It does so by using daemon threads
166   * and adding a shutdown hook to wait for their completion.
167   *
168   * <p>This method waits 120 seconds before continuing with JVM termination,
169   * even if the executor has not finished its work.
170   *
171   * <p>This is mainly for fixed thread pools.
172   * See {@link Executors#newScheduledThreadPool(int)}.
173   *
174   * @param executor the executor to modify to make sure it exits when the
175   *        application is finished
176   * @return an unmodifiable version of the input which will not hang the JVM
177   */
178  public static ScheduledExecutorService getExitingScheduledExecutorService(
179      ScheduledThreadPoolExecutor executor) {
180    return getExitingScheduledExecutorService(executor, 120, TimeUnit.SECONDS);
181  }
182
183  /**
184   * Creates an executor service that runs each task in the thread
185   * that invokes {@code execute/submit}, as in {@link CallerRunsPolicy}  This
186   * applies both to individually submitted tasks and to collections of tasks
187   * submitted via {@code invokeAll} or {@code invokeAny}.  In the latter case,
188   * tasks will run serially on the calling thread.  Tasks are run to
189   * completion before a {@code Future} is returned to the caller (unless the
190   * executor has been shutdown).
191   *
192   * <p>Although all tasks are immediately executed in the thread that
193   * submitted the task, this {@code ExecutorService} imposes a small
194   * locking overhead on each task submission in order to implement shutdown
195   * and termination behavior.
196   *
197   * <p>The implementation deviates from the {@code ExecutorService}
198   * specification with regards to the {@code shutdownNow} method.  First,
199   * "best-effort" with regards to canceling running tasks is implemented
200   * as "no-effort".  No interrupts or other attempts are made to stop
201   * threads executing tasks.  Second, the returned list will always be empty,
202   * as any submitted task is considered to have started execution.
203   * This applies also to tasks given to {@code invokeAll} or {@code invokeAny}
204   * which are pending serial execution, even the subset of the tasks that
205   * have not yet started execution.  It is unclear from the
206   * {@code ExecutorService} specification if these should be included, and
207   * it's much easier to implement the interpretation that they not be.
208   * Finally, a call to {@code shutdown} or {@code shutdownNow} may result
209   * in concurrent calls to {@code invokeAll/invokeAny} throwing
210   * RejectedExecutionException, although a subset of the tasks may already
211   * have been executed.
212   */
213  public static ExecutorService sameThreadExecutor() {
214    return new SameThreadExecutorService();
215  }
216
217  /*
218   * TODO(cpovirk): make this and other classes implement
219   * ListeningExecutorService?
220   */
221  // See sameThreadExecutor javadoc for behavioral notes.
222  private static class SameThreadExecutorService
223      extends AbstractExecutorService {
224    /**
225     * Lock used whenever accessing the state variables
226     * (runningTasks, shutdown, terminationCondition) of the executor
227     */
228    private final Lock lock = new ReentrantLock();
229
230    /** Signaled after the executor is shutdown and running tasks are done */
231    private final Condition termination = lock.newCondition();
232
233    /*
234     * Conceptually, these two variables describe the executor being in
235     * one of three states:
236     *   - Active: shutdown == false
237     *   - Shutdown: runningTasks > 0 and shutdown == true
238     *   - Terminated: runningTasks == 0 and shutdown == true
239     */
240    private int runningTasks = 0;
241    private boolean shutdown = false;
242
243    @Override
244    public void execute(Runnable command) {
245      startTask();
246      try {
247        command.run();
248      } finally {
249        endTask();
250      }
251    }
252
253    @Override
254    public boolean isShutdown() {
255      lock.lock();
256      try {
257        return shutdown;
258      } finally {
259        lock.unlock();
260      }
261    }
262
263    @Override
264    public void shutdown() {
265      lock.lock();
266      try {
267        shutdown = true;
268      } finally {
269        lock.unlock();
270      }
271    }
272
273    // See sameThreadExecutor javadoc for unusual behavior of this method.
274    @Override
275    public List<Runnable> shutdownNow() {
276      shutdown();
277      return Collections.emptyList();
278    }
279
280    @Override
281    public boolean isTerminated() {
282      lock.lock();
283      try {
284        return shutdown && runningTasks == 0;
285      } finally {
286        lock.unlock();
287      }
288    }
289
290    @Override
291    public boolean awaitTermination(long timeout, TimeUnit unit)
292        throws InterruptedException {
293      long nanos = unit.toNanos(timeout);
294      lock.lock();
295      try {
296        for (;;) {
297          if (isTerminated()) {
298            return true;
299          } else if (nanos <= 0) {
300            return false;
301          } else {
302            nanos = termination.awaitNanos(nanos);
303          }
304        }
305      } finally {
306        lock.unlock();
307      }
308    }
309
310    /**
311     * Checks if the executor has been shut down and increments the running
312     * task count.
313     *
314     * @throws RejectedExecutionException if the executor has been previously
315     *         shutdown
316     */
317    private void startTask() {
318      lock.lock();
319      try {
320        if (isShutdown()) {
321          throw new RejectedExecutionException("Executor already shutdown");
322        }
323        runningTasks++;
324      } finally {
325        lock.unlock();
326      }
327    }
328
329    /**
330     * Decrements the running task count.
331     */
332    private void endTask() {
333      lock.lock();
334      try {
335        runningTasks--;
336        if (isTerminated()) {
337          termination.signalAll();
338        }
339      } finally {
340        lock.unlock();
341      }
342    }
343  }
344}