util/concurrent/Executors.java

/*
 * Copyright (C) 2007 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.google.common.util.concurrent;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Factory and utility methods for {@link java.util.concurrent.Executor}, {@link
 * ExecutorService}, and {@link ThreadFactory}.
 *
 * @author Eric Fellheimer
 * @author Kyle Littlefield
 * @author Justin Mahoney
 * @since 2009.09.15 <b>tentative</b>
 */
public class Executors {

  /**
   * Converts the given ThreadPoolExecutor into an ExecutorService that exits
   * when the application is complete.  It does so by using daemon threads and
   * adding a shutdown hook to wait for their completion.
   *
   * <p>This is mainly for fixed thread pools.
   * See {@link java.util.concurrent.Executors#newFixedThreadPool(int)}.
   *
   * @param executor the executor to modify to make sure it exits when the
   *        application is finished
   * @param terminationTimeout how long to wait for the executor to
   *        finish before terminating the JVM
   * @param timeUnit unit of time for the time parameter
   * @return an unmodifiable version of the input which will not hang the JVM
   */
  public static ExecutorService getExitingExecutorService(
      ThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
    executor.setThreadFactory(daemonThreadFactory(executor.getThreadFactory()));

    ExecutorService service = java.util.concurrent.Executors
        .unconfigurableExecutorService(executor);

    addDelayedShutdownHook(service, terminationTimeout, timeUnit);

    return service;
  }

  /**
   * Converts the given ScheduledThreadPoolExecutor into a
   * ScheduledExecutorService that exits when the application is complete.  It
   * does so by using daemon threads and adding a shutdown hook to wait for
   * their completion.
   *
   * <p>This is mainly for fixed thread pools.
   * See {@link java.util.concurrent.Executors#newScheduledThreadPool(int)}.
   *
   * @param executor the executor to modify to make sure it exits when the
   *        application is finished
   * @param terminationTimeout how long to wait for the executor to
   *        finish before terminating the JVM
   * @param timeUnit unit of time for the time parameter
   * @return an unmodifiable version of the input which will not hang the JVM
   */
  public static ScheduledExecutorService getExitingScheduledExecutorService(
      ScheduledThreadPoolExecutor executor, long terminationTimeout,
      TimeUnit timeUnit) {
    executor.setThreadFactory(daemonThreadFactory(executor.getThreadFactory()));

    ScheduledExecutorService service = java.util.concurrent.Executors
        .unconfigurableScheduledExecutorService(executor);

    addDelayedShutdownHook(service, terminationTimeout, timeUnit);

    return service;
  }

  /**
   * Add a shutdown hook to wait for thread completion in the given
   * {@link ExecutorService service}.  This is useful if the given service uses
   * daemon threads, and we want to keep the JVM from exiting immediately on
   * shutdown, instead giving these daemon threads a chance to terminate
   * normally.
   * @param service ExecutorService which uses daemon threads
   * @param terminationTimeout how long to wait for the executor to finish
   *        before terminating the JVM
   * @param timeUnit unit of time for the time parameter
   */
  public static void addDelayedShutdownHook(
      final ExecutorService service, final long terminationTimeout,
      final TimeUnit timeUnit) {
    Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
      public void run() {
        try {
          // We'd like to log progress and failures that may arise in the
          // following code, but unfortunately the behavior of logging
          // is undefined in shutdown hooks.
          // This is because the logging code installs a shutdown hook of its
          // own. See Cleaner class inside {@link LogManager}.
          service.shutdown();
          service.awaitTermination(terminationTimeout, timeUnit);
        } catch (InterruptedException ignored) {
          // We're shutting down anyway, so just ignore.
        }
      }
    }));
  }

  /**
   * Converts the given ThreadPoolExecutor into an ExecutorService that exits
   * when the application is complete.  It does so by using daemon threads and
   * adding a shutdown hook to wait for their completion.
   *
   * <p>This method waits 120 seconds before continuing with JVM termination,
   * even if the executor has not finished its work.
   *
   * <p>This is mainly for fixed thread pools.
   * See {@link java.util.concurrent.Executors#newFixedThreadPool(int)}.
   *
   * @param executor the executor to modify to make sure it exits when the
   *        application is finished
   * @return an unmodifiable version of the input which will not hang the JVM
   */
  public static ExecutorService getExitingExecutorService(
      ThreadPoolExecutor executor) {
    return getExitingExecutorService(executor, 120, TimeUnit.SECONDS);
  }

  /**
   * Converts the given ThreadPoolExecutor into a ScheduledExecutorService that
   * exits when the application is complete.  It does so by using daemon threads
   * and adding a shutdown hook to wait for their completion.
   *
   * <p>This method waits 120 seconds before continuing with JVM termination,
   * even if the executor has not finished its work.
   *
   * <p>This is mainly for fixed thread pools.
   * See {@link java.util.concurrent.Executors#newScheduledThreadPool(int)}.
   *
   * @param executor the executor to modify to make sure it exits when the
   *        application is finished
   * @return an unmodifiable version of the input which will not hang the JVM
   */
  public static ScheduledExecutorService getExitingScheduledExecutorService(
      ScheduledThreadPoolExecutor executor) {
    return getExitingScheduledExecutorService(executor, 120, TimeUnit.SECONDS);
  }

  /**
   * Returns a {@link ThreadFactory} which creates daemon threads. This is
   * implemented by wrapping {@link
   * java.util.concurrent.Executors#defaultThreadFactory()}, marking all new
   * threads as daemon threads
   *
   * @return a {@link ThreadFactory} which creates daemon threads
   */
  public static ThreadFactory daemonThreadFactory() {
    return daemonThreadFactory(
        java.util.concurrent.Executors.defaultThreadFactory());
  }

  /**
   * Wraps another {@link ThreadFactory}, making all new threads daemon threads.
   *
   * @param factory the {@link ThreadFactory} used to generate new threads
   * @return a new {@link ThreadFactory} backed by {@code factory} whose created
   *         threads are all daemon threads
   */
  public static ThreadFactory daemonThreadFactory(ThreadFactory factory) {
    return new DaemonThreadFactory(factory);
  }

  /**
   * Creates an executor service that runs each task in the thread
   * that invokes {@code execute/submit}, as in {@link CallerRunsPolicy}  This
   * applies both to individually submitted tasks and to collections of tasks
   * submitted via {@code invokeAll} or {@code invokeAny}.  In the latter case,
   * tasks will run serially on the calling thread.  Tasks are run to
   * completion before a {@code Future} is returned to the caller (unless the
   * executor has been shutdown).
   *
   * <p>Although all tasks are immediately executed in the thread that
   * submitted the task, this {@code ExecutorService} imposes a small
   * locking overhead on each task submission in order to implement shutdown
   * and termination behavior.
   *
   * <p>The implementation deviates from the {@code ExecutorService}
   * specification with regards to the {@code shutdownNow} method.  First,
   * "best-effort" with regards to canceling running tasks is implemented
   * as "no-effort".  No interrupts or other attempts are made to stop
   * threads executing tasks.  Second, the returned list will always be empty,
   * as any submitted task is considered to have started execution.
   * This applies also to tasks given to {@code invokeAll} or {@code invokeAny}
   * which are pending serial execution, even the subset of the tasks that
   * have not yet started execution.  It is unclear from the
   * {@code ExecutorService} specification if these should be included, and
   * it's much easier to implement the interpretation that they not be.
   * Finally, a call to {@code shutdown} or {@code shutdownNow} may result
   * in concurrent calls to {@code invokeAll/invokeAny} throwing
   * RejectedExecutionException, although a subset of the tasks may already
   * have been executed.
   */
  public static ExecutorService sameThreadExecutor() {
    return new SameThreadExecutorService();
  }

  // See sameThreadExecutor javadoc for behavioral notes.
  private static class SameThreadExecutorService extends AbstractExecutorService {

    /**
     * Lock used whenever accessing the state variables
     * (runningTasks, shutdown, terminationCondition) of the executor
     */
    private final Lock lock = new ReentrantLock();

    /** Signaled after the executor is shutdown and running tasks are done */
    private final Condition termination = lock.newCondition();

    /*
     * Conceptually, these two variables describe the executor being in
     * one of three states:
     *   - Active: shutdown == false
     *   - Shutdown: runningTasks > 0 and shutdown == true
     *   - Terminated: runningTasks == 0 and shutdown == true
     */
    private int runningTasks = 0;
    private boolean shutdown = false;

    /*@Override*/
    public void execute(Runnable command) {
      startTask();
      try {
        command.run();
      } finally {
        endTask();
      }
    }

    /*@Override*/
    public boolean isShutdown() {
      lock.lock();
      try {
        return shutdown;
      } finally {
        lock.unlock();
      }
    }

    /*@Override*/
    public void shutdown() {
      lock.lock();
      try {
        shutdown = true;
      } finally {
        lock.unlock();
      }
    }

    // See sameThreadExecutor javadoc for unusual behavior of this method.
    /*@Override*/
    public List<Runnable> shutdownNow() {
      shutdown();
      return Collections.emptyList();
    }

    /*@Override*/
    public boolean isTerminated() {
      lock.lock();
      try {
        return shutdown && runningTasks == 0;
      } finally {
        lock.unlock();
      }
    }

    /*@Override*/
    public boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException {
      long nanos = unit.toNanos(timeout);
      lock.lock();
      try {
        for (;;) {
          if (isTerminated()) {
            return true;
          } else if (nanos <= 0) {
            return false;
          } else {
            nanos = termination.awaitNanos(nanos);
          }
        }
      } finally {
        lock.unlock();
      }
    }

    /**
     * Checks if the executor has been shut down and increments the running
     * task count.
     *
     * @throws RejectedExecutionException if the executor has been previously
     *         shutdown
     */
    private void startTask() {
      lock.lock();
      try {
        if (isShutdown()) {
          throw new RejectedExecutionException("Executor already shutdown");
        }
        runningTasks++;
      } finally {
        lock.unlock();
      }
    }

    /**
     * Decrements the running task count.
     */
    private void endTask() {
      lock.lock();
      try {
        runningTasks--;
        if (isTerminated()) {
          termination.signalAll();
        }
      } finally {
        lock.unlock();
      }
    }
  }
}