taskEngine.py

Go to the documentation of this file.

#!/usr/bin/env python
#
#                               Copyright 2005
#                                     by
#                        The Board of Trustees of the
#                     Leland Stanford Junior University.
#                            All rights reserved.
#

__facility__ = "Online"
__abstract__ = "GLAST LAT Task Engine class"
__author__   = "R. Claus <Claus@SLAC.Stanford.edu> SLAC - GLAST LAT I&T/Online"
__date__     = "2004/03/09 00:00:00"
__updated__  = "$Date: 2005/09/22 00:29:31 $"
__version__  = "$Revision: 1.5 $"
__release__  = "$Name: HEAD $"
__credits__  = "SLAC"

import LICOS.copyright_SLAC

from   Queue         import Queue
from   threading     import Thread, currentThread
import logging       as     log


class TaskEngine(Thread):
  """!\brief A class for setting up and queueing tasks to a task engine.

  A task engine is a thread that polls a task queue and runs tasks found on
  the queue.  Tasks are functions or class methods accompanied with their
  arguments and keyword arguments.  Each task must run to completion before
  the next task is pulled from the queue.  Returning a return value from the
  task to the caller is currently not supported (easy to add, if the need
  arises).  The queue is a FIFO queue and there is currently no way to jam
  a task to the head of the queue.
  """
  def __init__(self, name = None):
    Thread.__init__(self)

    if name is None:
      name = self.__class__.__name__
    self.__name  = name
    self.__queue = Queue()

  def name(self):
    """!\brief Method for querying the task engine's name

    \return The task engine's name
    """
    return self.__name

  def spawn(self, task, *args, **kwargs):
    """!\brief Method used to queue a task to the task engine.

    \param task    The task routine to queue
    \param args    The arguments for the task routine
    \param kwargs  The keyword arguments for the task routine
    """
    assert self.__queue is not None, (self.name + ".spawn: " + \
                                      "TaskEngine.__init__() was not called")
    self.__queue.put((task,) + args + (kwargs,))

  def run(self):
    """!\brief The task engine main routine.

    This method is invoked by the Thread class and should be considered
    'protected'.  Do not call it.
    """
    self.setName(self.name())

    while True:
      taskArgsKwargs = self.__queue.get(True) # Blocks
      if taskArgsKwargs[0] is None:  break    # Quit signal is None

      try:
        taskArgsKwargs[0](*taskArgsKwargs[1:-1], **taskArgsKwargs[-1])
      except:
        import types
        if isinstance(taskArgsKwargs[0], types.MethodType):
          log.exception(self.name()                               + \
                        ": Trapped exception from spawned task "  + \
                        taskArgsKwargs[0].im_class.__name__ + '.' + \
                        taskArgsKwargs[0].__name__ + '()')
        else:
          log.exception(self.name()                               + \
                        ": Trapped exception from spawned task "  + \
                        taskArgsKwargs[0].__name__ + '()')

      # Drop references to this object to allow it to be garbage collected
      taskArgsKwargs = None

    log.info(self.name() + ": terminating")

  def shutdown(self, tmo = 2.0):
    """!\brief Method for shutting down the task engine

    This method blocks (with a timeout) until the task engine has exited.
    It makes no sense to spawn this method to the task engine since it waits
    for the task engine to exit.

    \param tmo  The amount of time (seconds) to wait for the task engine to exit
    \return  0 for success,
           -1 if the task engine didn't exit,
           -2 if the task engine exited prematurely
    """
    # This method can not be called from the taskEngine thread on purpose.
    # If it could, there would be no way for the instigator to synchronize.
    if currentThread() is self:
      raise RuntimeError, "A task engine can't be shut down by its own thread"
    if self.isAlive():                  # Thread still there or did it crash?
      self.spawn(None)                  # Signal to quit
      self.join(tmo)                    # Let engine exit
      if self.isAlive():                # Timed out?
        log.error(self.name() + ".shutdown: " + \
                  "Task engine did not exit when expected")
        return -1
    else:
      log.warn(self.name() + ":.shutdown: " + \
               "Task engine unexpectedly disappeared")
      return -2

    return 0


#-------------------------------------------------------------------------------
# Unit testing follows:

if __debug__ and __name__ == '__main__':
  import unittest
  import time

  results = [None, None, None, None, None, None]
  status  = None

  class TaskEngineTestCase(unittest.TestCase):
    def __init__(self, arg):
      unittest.TestCase.__init__(self, arg)

    def checkInstantiation(self):
      """!\brief Test class instantiation.

      """
      engine = None
      engine = TaskEngine("TestEngine")
      assert engine != None, "Could not instantiate a TaskEngine"

    def checkStartup(self):
      """!\brief Test whether the thread starts up.

      """
      engine = TaskEngine("TestEngine")
      engine.start()
      time.sleep(0.1)                   # Let it start
      assert engine.isAlive(), "Could not start TaskEngine"
      engine.shutdown()
      time.sleep(.1)

    def __task_1(self, a, b, c, x, y, z):
      self.__a = a
      self.__b = b
      self.__c = c
      self.__x = x
      self.__y = y
      self.__z = z

    def checkSpawn(self):
      """!\brief Test whether we can spawn a class method task to the engine and
      have the arguments handled correctly.

      """
      self.__a = None
      self.__b = None
      self.__c = None
      self.__x = None
      self.__y = None
      self.__z = None

      a = "Hello world"
      b = 3.1415926
      c = 0xabadcafeL
      engine = TaskEngine("TestEngine")
      engine.start()
      time.sleep(0.1)                   # Let it start
      engine.spawn(self.__task_1, a, b, c, z = a, x = b, y = c)
      time.sleep(0.1)                   # Let it run
      assert engine.isAlive(), "TaskEngine prematurely disappeared"
      assert self.__a is a, "TaskEngine did not correctly handle args[0]"
      assert self.__b is b, "TaskEngine did not correctly handle args[1]"
      assert self.__c is c, "TaskEngine did not correctly handle args[2]"
      assert self.__x is b, "TaskEngine did not correctly handle kwargs[0]"
      assert self.__y is c, "TaskEngine did not correctly handle kwargs[1]"
      assert self.__z is a, "TaskEngine did not correctly handle kwargs[2]"
      engine.shutdown()
      time.sleep(.1)

    def checkUnboundSpawn(self):
      """!\brief Test whether we can spawn an unbound function task to the engine and
      have the arguments handled correctly.

      """
      a = "Test of an unbound function"
      b = 2.718281828
      c = 0xdeadbeefL

      engine = TaskEngine("TestEngine")
      engine.start()
      time.sleep(0.1)                   # Let it start
      engine.spawn(_unbound1, a, b, c, y = a, z = b, x = c)
      time.sleep(0.1)                   # Let it run
      assert engine.isAlive(), "TaskEngine prematurely disappeared"
      assert results[0] is a, "TaskEngine did not correctly handle args[0]"
      assert results[1] is b, "TaskEngine did not correctly handle args[1]"
      assert results[2] is c, "TaskEngine did not correctly handle args[2]"
      assert results[3] is c, "TaskEngine did not correctly handle kwargs[0]"
      assert results[4] is a, "TaskEngine did not correctly handle kwargs[1]"
      assert results[5] is b, "TaskEngine did not correctly handle kwargs[2]"
      engine.shutdown()
      time.sleep(.1)

    def __doException(self):  # 2nd level to give traceback some depth
      raise RuntimeError, "Raising RuntimeError to test exception handling"
      return True

    def __task_2(self):
      self.__status = False
      self.__status = self.__doException()

    def checkException1(self):
      """!\brief Test that the engine doesn't die if a class method task throws an
      exception.

      """
      print ""
      print "##############################################################################"
      print "#                                                                            #"
      print "#  This test should cause a 'Trapped exception from spawned task' exception  #"
      print "#                                                                            #"
      print "##############################################################################"
      engine = TaskEngine("TestEngine")
      engine.start()
      time.sleep(0.1)                   # Let it start
      engine.spawn(self.__task_2)
      time.sleep(0.1)                   # Let it run
      assert engine.isAlive(), "TaskEngine died from exception"
      assert self.__status == False, "Exception didn't abort task as expected"
      engine.shutdown()
      time.sleep(.1)

    def checkException2(self):
      """!\brief Test that the engine doesn't die if an unbound function task throws an
      exception.

      """
      print ""     # Expecting an exception: add space for readability
      print "##############################################################################"
      print "#                                                                            #"
      print "#  This test should cause a 'Trapped exception from spawned task' exception  #"
      print "#                                                                            #"
      print "##############################################################################"
      engine = TaskEngine("TestEngine")
      engine.start()
      time.sleep(0.1)                   # Let it start
      engine.spawn(_unbound2)
      time.sleep(0.1)                   # Let it run
      assert engine.isAlive(), "TaskEngine died from exception"
      assert status == False, "Exception didn't abort task as expected"
      engine.shutdown()
      time.sleep(.1)

    def checkSelfShutdown(self):
      """!\brief Test that attempts to shut down an engine via a task fails.

      """
      print ""     # Expecting an exception: add space for readability
      print "##############################################################################"
      print "#                                                                            #"
      print "#  This test should cause a 'Trapped exception from spawned task' exception  #"
      print "#                                                                            #"
      print "##############################################################################"
      engine = TaskEngine("TestEngine")
      engine.start()
      time.sleep(0.1)                   # Let it start
      engine.spawn(engine.shutdown)
      time.sleep(0.1)                   # Let it run
      assert engine.isAlive(), "TaskEngine died from exception"
      engine.shutdown()
      time.sleep(.1)

    def checkShutdown1(self):
      """!\brief Test that attempting to shut down an engine from an unrelated thread
      succeeds.

      """
      engine = TaskEngine("TestEngine")
      engine.start()
      time.sleep(0.1)                   # Let it start
      status = engine.shutdown()
      time.sleep(0.1)                   # Let it shut down
      assert not engine.isAlive(), "TaskEngine didn't shut down"
      assert status == 0, "Bad status %d returned from shutdown()" % (status)

    def checkShutdown2(self):
      """!\brief Test that attempts to shut down an engine that wasn't started fails.

      """
      print ""
      print "#############################################################################"
      print "#                                                                           #"
      print "#  This test should cause a 'Task engine unexpectedly disappeared' warning  #"
      print "#                                                                           #"
      print "#############################################################################"
      engine = TaskEngine("TestEngine")
      status = engine.shutdown()
      assert status != 0, "TaskEngine didn't shut down as expected"

  def _unbound1(a, b, c, x, y, z):
    results[0] = a
    results[1] = b
    results[2] = c
    results[3] = x
    results[4] = y
    results[5] = z

  def _doException():
    raise RuntimeError, "Raising RuntimeError to test exception handling"
    return True

  def _unbound2():
    global status
    status = False
    status = _doException()

  def suite():
    loader = unittest.TestLoader()
    testSuite = loader.loadTestsFromNames(
      ["checkInstantiation", "checkStartup", "checkSpawn", "checkUnboundSpawn",
       "checkException1", "checkException2", "checkSelfShutdown",
       "checkShutdown1", "checkShutdown2"], TaskEngineTestCase)
    return testSuite

  def test():
    print "########################################################"
    print "#                                                      #"
    print "#  Exceptions are expected from this test.  These can  #"
    print "#  be ignored if the test completes with status 'OK'   #"
    print "#                                                      #"
    print "########################################################"
    print ""

    # Initialize logging
    log.basicConfig()

    runner = unittest.TextTestRunner()
    runner.run(suite())

    # Shut down the logger
    log.shutdown()

  # Now run the tests
  test()


#-------------------------------------------------------------------------------

---