healthy thread registry

thread_registry.py

import threading

from twitter.common import log
from twitter.common.exceptions import ExceptionalThread


class ThreadRegistry(ExceptionalThread):
  DEFAULT_WAIT_INTERVAL_SECS = 1.0

  def __init__(self, wait_interval=DEFAULT_WAIT_INTERVAL_SECS):
    self.__threads = []
    super(ThreadRegistry, self).__init__()
    self.daemon = True
    self._stopped = threading.Event()
    self.dead = threading.Event()
    self.lock = threading.Lock()
    self.wait_interval = wait_interval

  def register(self, thread_object):
    with self.lock:
      self.__threads.append(thread_object)

  def unregister(self, thread_object):
    with self.lock:
      self.__threads.remove(thread_object)

  def stop(self):
    self._stopped.set()

  def run(self):
    while not self.dead.is_set() and not self._stopped.wait(timeout=self.wait_interval):
      with self.lock:
        for thread in self.__threads:
          if not thread.is_alive():
            log.error('Thread %s died unexpectedly!' % thread)
            self.dead.set()