Tomcat's Graceful Shutdown with Daemons and Shutdown Hooks

My last couple of blogs have talked about long polling and Spring's DeferredResult technique and to demonstrate these concepts I've shoehorned the code from my Producer Consumer project into a web application. Although the code demonstrates the points made by the blogs it does contain a large number of holes in its logic. Apart from the fact that in a real application you wouldn't use a simple LinkedBlockingQueue, but would choose JMS or some other industrial-strength messaging service, and the fact that only one user can get a hold of the match updates, there's also the problem that it spawns badly behaved threads that don't close down when the JVM terminates.

You may wonder why this should be a problem… well to you, as a developer, it isn't really a problem at all, it's only a little bit of sloppy programming. But, to one of your operations guys it can make life unnecessarily difficult. The reason for this is that if you have too many badly behaved threads then typing Tomcat's shutdown.sh command will have very little effect and you have to savagely kill your web server by typing something like:

ps -ef | grep java

to get the pid and then

kill -9 <<pid>>

…and when you have a field of Tomcat web servers to restart all this extra kerfuffle that becomes a severe pain. When you type shutdown.sh you want Tomcat to stop.

In my last couple of blogs the badly behaved threads I created had the following run() methods with the first of these, shown below, being really badly behaved:

  @Override
  public void run() {

    while (true) {
      try {

        DeferredResult<Message> result = resultQueue.take();
        Message message = queue.take();

        result.setResult(message);

      } catch (InterruptedException e) {
        throw new UpdateException("Cannot get latest update. " + e.getMessage(), e);
      }
    }
  }

In this code I've used an infinite while(true), which means that the thread will just keep running and never terminate.

  @Override
  public void run() {

    sleep(5); // Sleep to allow the reset of the app to load

    logger.info("The match has now started...");
    long now = System.currentTimeMillis();
    List<Message> matchUpdates = match.getUpdates();

    for (Message message : matchUpdates) {

      delayUntilNextUpdate(now, message.getTime());
      logger.info("Add message to queue: {}", message.getMessageText());
      queue.add(message);
    }
    start = true; // Game over, can restart
    logger.warn("GAME OVER");
  }

The second example, above, is also pretty badly behaved. It'll keep taking messages from MatchUpdates list and adding them to the message queue at the appropriate moment. Their only saving grace is that they may throw an InterruptedException, which if handled correctly would cause thread termination; however, this cannot be guaranteed.

There's a quick fix for this, really… all you need to do is to ensure that any threads you create are daemon threads. The definition of a daemon thread is a thread that doesn't prevent the JVM from exiting when the program finishes but, the thread is still running. The usual example of a daemon thread is the JVM's garbage collection thread. To turn your threads into daemon threads you simply call:

    thread.setDaemon(true);

...and when you type shutdown.sh then, WHAM, all your threads will disappear. There is, however, a problem with this. What if one of your daemon's threads was doing something important and chopping it down in its prime, resulting in the loss of some pretty important data?

What you need to do is to ensure that all your threads shut down gracefully, completing any work they may be currently undertaking. The rest of this blog demonstrates a fix for these errant threads, gracefully coordinating their shutdown by using a ShutdownHook. According to the documentation, a "shutdown hook is simply an initialized but unstarted thread. When the virtual machine begins its shutdown sequence it will start all registered shutdown hooks in some unspecified order and let them run concurrently." So, after reading the last sentence you may have guessed that what you need to do is to create a thread that has the responsibility of shutting down all your other threads and is passed to the JVM as a shutdown hook. All of this can be generically implemented in a couple of small classes and by performing some jiggery-pokery on your existing thread run() methods.

The two classes to create are a ShutdownService and a Hook. The Hook class, which I'll demonstrate first, is used to link the ShutdownService to your threads. The code for Hook is as follows:

public class Hook {

  private static final Logger logger = LoggerFactory.getLogger(Hook.class);

  private boolean keepRunning = true;

  private final Thread thread;

  Hook(Thread thread) {
    this.thread = thread;
  }

  /**
   * @return True if the daemon thread is to keep running
   */
  public boolean keepRunning() {
    return keepRunning;
  }

  /**
   * Tell the client daemon thread to shutdown and wait for it to close gracefully.
   */
  public void shutdown() {
    keepRunning = false;
    thread.interrupt();
    try {
      thread.join();
    } catch (InterruptedException e) {
      logger.error("Error shutting down thread with hook", e);
    }
  }
}

The Hook contains two instance variables: keepRunning and thread. thread is a reference to the thread that this instance of Hook is responsible for shutting down, while keepRunning tells the thread to… keep running.

Hook has two public methods: keepRunning() and shutdown(). keepRunning() is called by the thread to figure out whether it should keep running, and shutdown() is called by the ShutdownService's shutdown hook thread to get your thread to shut down. This is the most interesting of the two methods. Firstly it sets the keepRunning variable to false. It then calls thread.interrupt() to interrupt the thread forcing it to throw an InterruptedException. Lastly, it calls thread.join() and waits for the thread instance to shutdown.

@Service
public class ShutdownService {

  private static final Logger logger = LoggerFactory.getLogger(ShutdownService.class);

  private final List<Hook> hooks;

  public ShutdownService() {
    logger.debug("Creating shutdown service");
    hooks = new ArrayList<Hook>();
    createShutdownHook();
  }

  /**
   * Protected for testing
   */
  @VisibleForTesting
  protected void createShutdownHook() {
    ShutdownDaemonHook shutdownHook = new ShutdownDaemonHook();
    Runtime.getRuntime().addShutdownHook(shutdownHook);
  }

  protected class ShutdownDaemonHook extends Thread {

    /**
     * Loop and shutdown all the daemon threads using the hooks
     *
     * @see java.lang.Thread#run()
     */
    @Override
    public void run() {

      logger.info("Running shutdown sync");

      for (Hook hook : hooks) {
        hook.shutdown();
      }
    }
  }

  /**
   * Create a new instance of the hook class
   */
  public Hook createHook(Thread thread) {

    thread.setDaemon(true);
    Hook retVal = new Hook(thread);
    hooks.add(retVal);
    return retVal;
  }

  @VisibleForTesting
  List<Hook> getHooks() {
    return hooks;
  }
}

    Runtime.getRuntime().addShutdownHook(shutdownHook);

@Service("DeferredService")
public class DeferredResultService implements Runnable {

  private static final Logger logger = LoggerFactory.getLogger(DeferredResultService.class);

  private final BlockingQueue<DeferredResult<Message>> resultQueue = new LinkedBlockingQueue<>();

  private Thread thread;

  private volatile boolean start = true;

  @Autowired
  private ShutdownService shutdownService;

  private Hook hook;

  @Autowired
  @Qualifier("theQueue")
  private LinkedBlockingQueue<Message> queue;

  @Autowired
  @Qualifier("BillSkyes")
  private MatchReporter matchReporter;

  public void subscribe() {
    logger.info("Starting server");
    matchReporter.start();
    startThread();
  }

  private void startThread() {

    if (start) {
      synchronized (this) {
        if (start) {
          start = false;
          thread = new Thread(this, "Studio Teletype");
          hook = shutdownService.createHook(thread);
          thread.start();
        }
      }
    }
  }

  @Override
  public void run() {

    logger.info("DeferredResultService - Thread running");
    while (hook.keepRunning()) {
      try {

        DeferredResult<Message> result = resultQueue.take();
        Message message = queue.take();

        result.setResult(message);

      } catch (InterruptedException e) {
        System.out.println("Interrupted when waiting for latest update. " + e.getMessage());
      }
    }
    System.out.println("DeferredResultService - Thread ending");
  }

  public void getUpdate(DeferredResult<Message> result) {
    resultQueue.add(result);
  }

}

          thread = new Thread(this, "Studio Teletype");
          hook = shutdownService.createHook(thread);
          thread.start();

    while (hook.keepRunning()) {

public class MatchReporter implements Runnable {

  private static final Logger logger = LoggerFactory.getLogger(MatchReporter.class);

  private final Match match;

  private final Queue<Message> queue;

  private volatile boolean start = true;

  @Autowired
  private ShutdownService shutdownService;

  private Hook hook;

  public MatchReporter(Match theBigMatch, Queue<Message> queue) {
    this.match = theBigMatch;
    this.queue = queue;
  }

  /**
   * Called by Spring after loading the context. Will "kick off" the match...
   */
  public void start() {

    if (start) {
      synchronized (this) {
        if (start) {
          start = false;
          logger.info("Starting the Match Reporter...");
          String name = match.getName();
          Thread thread = new Thread(this, name);
          hook = shutdownService.createHook(thread);

          thread.start();
        }
      }
    } else {
      logger.warn("Game already in progress");
    }
  }

  /**
   * The main run loop
   */
  @Override
  public void run() {

    sleep(5); // Sleep to allow the reset of the app to load

    logger.info("The match has now started...");
    long now = System.currentTimeMillis();
    List<Message> matchUpdates = match.getUpdates();

    for (Message message : matchUpdates) {

      delayUntilNextUpdate(now, message.getTime());
      if (!hook.keepRunning()) {
        break;
      }
      logger.info("Add message to queue: {}", message.getMessageText());
      queue.add(message);
    }
    start = true; // Game over, can restart
    logger.warn("GAME OVER");
  }

  private void sleep(int deplay) {
    try {
      TimeUnit.SECONDS.sleep(10);
    } catch (InterruptedException e) {
      logger.info("Sleep interrupted...");
    }
  }

  private void delayUntilNextUpdate(long now, long messageTime) {

    while (System.currentTimeMillis() < now + messageTime) {

      try {
        Thread.sleep(100);
      } catch (InterruptedException e) {
        logger.info("MatchReporter Thread interrupted...");
      }
    }
  }

}