geb.py

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

__facility__ = "Online"
__abstract__ = "Generic event builder"
__author__   = "R. Claus <Claus@SLAC.Stanford.edu> SLAC - GLAST LAT I&T/Online"
__date__     = "2/27/2006"
__updated__  = ("$Date: 2006/07/19 16:06:44 $").split(' ')[1]
__version__  = "$Revision: 1.2 $"
__release__  = "$Name: R04-12-00 $"
__credits__  = "SLAC"

import LATTE.copyright_SLAC

from   Queue         import Queue
from   threading     import Thread
import time
import struct

import LDF


class GebContribution(object):
  def __init__(self, contributor):
    self.__contributor = contributor

  def identity(self):
    """!\brief Method to return the identity (LDF TypeId) of a contribution to
    an event.  Meant to be overridden.
    """
    raise NotImplementedError, "identity() method must be implemented in a subclass"

  def length(self):
    """!\brief Method to return the length of a contribution to an event.
    Meant to be overridden.
    """
    raise NotImplementedError, "sequence() method must be implemented in a subclass"

  def sequence(self):
    """!\brief Method to return the sequence value of the contribution.
    Meant to be overridden.

    This method must return a value for which the equality and greater
    than operators are defined.  The sequence values must be common
    amoungst all contributors such that a if a contribution's sequence
    value is equal to another contributions sequence value, the two
    contributions belong together as part of the same event.  A
    contribution A's sequence value that is greater than another
    contribution B's sequence value is interpreted to mean that A is
    newer than B.  This information is used to flush out partially
    built older events.
    """
    raise NotImplementedError, "sequence() method must be implemented in a subclass"

  def contributor(self):
    return self.__contributor


class GebContributor(Thread):
  # Set up one common blocking queue for all contributors
  __contributionQueue = Queue()

  def __init__(self):
    Thread.__init__(self)
    self.setName(self.name())

  def contributionQueue():
    """!\brief Method to get the common contribution queue.

    \returns A Queue object
    """
    return GebContributor.__contributionQueue

  contributionQueue = staticmethod(contributionQueue)

  def read(self):
    """!\brief Method to read a contribution from somewhere.
    Meant to be overridden.
    """
    raise NotImplementedError, "read() method must be implemented in a subclass"

  def name(self):
    """!\brief Method to return the name of the contribution type.
    Meant to be overridden.
    """
    raise NotImplementedError, "name() method must be implemented in a subclass"

  def sequence(self, sequence):
    """!\brief Optional method for remapping the ancillary contribution's
    sequence number into the same space as the other contributions' sequence
    numbers.  If this method is not overridden, the supplied sequence number is
    returned.
    """
    return sequence

  def open(self):
    """!\brief Method to open the contributor.
    Meant to be overridden.
    """
    raise NotImplementedError, "open() method must be implemented in a subclass"

  def close(self):
    """!\brief Optional method to close the contributor.
    Meant to be overridden.
    """
    pass

  def idMask(self, id = None):
    """!\brief Method for building the contributor mask
    """
    if id is not None:
      self.__idMask = id
    return self.__idMask

  def run(self):
    """!\brief Loop that continuously reads from the data source
    """
    contributionQueue = self.__contributionQueue
    reason = None
    try:
      while reason != Geb.Terminate:
        reason, status, contribution = self.read()
        contributionQueue.put((reason, status, contribution))

        # Allow other threads to run by forcing a reschedule
        time.sleep(0.0)
    except Exception, e:
      import logging
      logging.exception(self.name() + ": Trapped exception:\n" + str(e))

    print self.name() + ": terminating"
    import logging
    logging.info(self.name() + ": terminating")


class Geb(Thread):
  Build     = 0
  DontBuild = 1
  Terminate = 2
  Sweep     = 3
  def __init__(self, contributors):
    Thread.__init__(self)
    self.setName(self.name())
    self.__idMask = 0L
    idMask        = 0x1L
    for contributor in contributors:
      # Make sure that the contributors were inherited from GebContributor
      if not isinstance(contributor, GebContributor):
        raise TypeError, "Contributor must be a subclass of GebContributor"
      self.__idMask |= contributor.idMask(idMask)
      idMask <<= 1
    self.__contributors      = contributors
    self.__contributionQueue = GebContributor.contributionQueue()
    self.__contributionsDict = {}

  def name(self):
    """!\brief Method to return the name of the event builder.
    May be overridden.
    """
    return "GEB"

  def shutdown(self):
    """!\brief Method to shutdown the builder
    """
    # Shutdown the contributors
    for contributor in self.__contributors:
      contributor.close()

  def run(self):
    """!\brief Method run by the Thread base class.  Use the start() method
    to start event building.
    """
    try:
      self.build()
    except Exception, e:
      import logging
      logging.exception(self.name() + ":\n" + str(e))

  def build(self):
    """!\brief Call this method to start event building
    """
    contributionQueue = self.__contributionQueue
    contributionsDict = self.__contributionsDict
    contributors      = self.__contributors
    dTime             = 0.0

    # Open the contributors
    for contributor in contributors:
      contributor.open()

    # Start the contributor threads up
    for contributor in contributors:
      contributor.start()

    reason, status, contribution = contributionQueue.get() # Blocks
    tStart = time.time()

    n = 0
    while reason != Geb.Terminate:
      if reason == Geb.Build:
        contributor = contribution.contributor()
        sequence    = contribution.sequence()
        try:
          contributionsDict[sequence].append((status, contribution))
          contributionsDict[sequence][0] |= contributor.idMask()
        except KeyError: # i.e., sequence not in contributionsDict.keys():
          contributionsDict[sequence] = [contributor.idMask(), (status, contribution)]

        if contributionsDict[sequence][0] == self.__idMask:
          self.__flush(sequence)
          self.__dispose(contributionsDict.pop(sequence)[1:])
          if n % 100 == 0:
            print "......Geb.build: seq %08x built and disposed" % (sequence)
          n += 1
      elif reason == Geb.Sweep:
        sequence = contribution.sequence()
        contributionsDict[sequence] = [self.__idMask, (status, contribution)]

        self.__flush(sequence)
        self.__dispose(contributionsDict.pop(sequence)[1:])
        print "......Geb.build: seq %08x disposed (sweep)" % (sequence)
      else:
        print "......Geb.build: id %08x disposed" % (contribution.identity())

        self.__dispose([(status, contribution)])

      dTime += time.time() - tStart
      reason, status, contribution = contributionQueue.get() # Blocks
      tStart = time.time()

    if n     == 0:  n     = 1.0e-99
    if dTime == 0:  dTime = 1.0e-99
    print "\nGeb.build: N = %d, T = %f, T/N = %f, N/T = %f" % \
          (n, dTime, dTime / n, float(n) / dTime)

    # Close the contributors
    for contributor in contributors:
      contributor.close()    # Redundant - normally done by shutdown()
      contributor.join()     # Wait for the contributor threads to exit

    # Let the subclass clean up and close
    self.close()

    print "\nGeb.build: terminating\n"
    import logging
    logging.info("Geb.build: terminating")

  def flushAll(self):
    """!\brief Method for flushing all stale events from the system.
    """
    contribsDict = self.__contributionsDict
    sequences    = contribsDict.keys()
    for staleSequence in sequences:
      contributions = contribsDict.pop(staleSequence)[1:]
      self.__dispose(contributions)
      ids = ""
      for status, contribution in contributions:
        ids += hex(contribution.identity()) + ", "
      import logging
      logging.warning("...Geb.flushAll: seq %08x flushed, ids = %s" % (staleSequence, ids))

  def __flush(self, sequence):
    """!\brief Method for flushing stale events from the system.

    Stale events are generally events that are missing contributions and will
    never be completed.  Since we require that events are disposed of in
    chronological order, incomplete events older than the most recently
    completed event are deemed stale.

    \param sequence - The sequence number of the most recently completed event
    """
    contribsDict = self.__contributionsDict
    sequences    = contribsDict.keys()
    for staleSequence in sequences:
      if staleSequence < sequence:
        contributions = contribsDict.pop(staleSequence)[1:]
        self.__dispose(contributions)
        ids = ""
        for status, contribution in contributions:
          ids += hex(contribution.identity()) + ", "
        import logging
        logging.warning("......Geb.flush: seq %08x flushed, ids = %s" % (staleSequence, ids))

  def __dispose(self, contributions):
    """!\brief Internal method to concatenate the contributions into a single
    event and call the subclass to dispose of it.
    """
    length   = 0
    buffer   = ''
    statuses = ()
    for status, contribution in contributions:
      statuses += (status,)
      length   += contribution.length()
      buffer   += str(contribution.buffer())

    length += 8
    header  = struct.pack('>2L', LDF.LATdatagram.ID, length)

    self.dispose((statuses, header + buffer))

  def close(self):
    """!\brief Optional method to close down the derived class.
    Meant to be overridden.
    """
    return

---