gEvtCliBT.py

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

__facility__ = "Online"
__abstract__ = "Replacement evtCli class for event merging"
__author__   = "R. Claus <Claus@SLAC.Stanford.edu> SLAC - GLAST LAT I&T/Online"
__date__     = "6/29/2006"
__updated__  = ("$Date: 2006/07/19 16:24:56 $").split(' ')[1]
__version__  = "$Revision: 1.2 $"
__release__  = "$Name: R04-12-00 $"
__credits__  = "SLAC"

import LATTE.copyright_SLAC

import struct
from   array                              import array
from   select                             import select
import Queue
import time

import LDF

from   LATTE.client.gEvtCli               import EvtCli, EvtCliIO
from   LATTE.client.gSocket               import Socket

import LATTE.merger.geb                   as     geb
from   LATTE.merger.ldfContribution       import LdfContribution
from   LATTE.merger.ancillaryContribution import AncillaryContribution


class SocketClosed(Exception):  pass

class LdfContributor(geb.GebContributor):
  """!\brief Concrete contributor class specifically meant to work with EvtCli.
  """
  def __init__(self, evtCli, offset = 1L):
    geb.GebContributor.__init__(self)
    self.__socket    = evtCli           # Must be a Socket object
    self.__timeout   = 1                # Same as EvtCliIO.__timeout
    self.__offset    = offset

  def name(self):
    return "LdfContributor"

  def sequence(self, sequence):
    """!\brief Method for remapping the ancillary contribution's sequence number
    into the same space as the other contributions' sequence numbers.

    This method has the side effect that it can affect the next the value
    returned, so it assumes that it is called with sequential sequence numbers.
    This would be a problem if were called multiple times to retrieve the
    sequence number for the same sequence number.
    \todo Fix the side effect nature of this routine
    """
    # offset does double duty by compensating for the off-by-one nature of the
    # two event number spaces and by keeping track of the event number extention
    # needed to handle the raw 17 bit event number roll-over problem
    seq = sequence - self.__offset
    if sequence == 0x0001ffff:  self.__offset -= 0x00020000L
    return seq

  def open(self):
    # Wait for the socket to be created
    while self.__socket.socket() is None:
      time.sleep(0.1)

  def close(self):
    self.__quit = True

  def read(self):
    """!\brief Method to read an LDF contribution from a socket.
    """
    markerMsk   = 0x7 << 22
    sweepFld    = 0x5 << 22             # Marker 5
    hdrLen      = 8
    self.__quit = 0
    rd, wr, ex = select([self.__socket.socket()], [], [], self.__timeout)
    while not self.__quit:
      if rd:
        data   = self.__socket.recvall(4 + hdrLen) # "4 +" for leading status
        status, typeId, length = struct.unpack('>3L', data)
        if status != 0:
          # Don't try to continue if bad status
          raise KeyError, "LdfContributor.read: Bad ocsmanager status %08x" % (status)
        if typeId != LDF.LATdatagram.ID:
          # Don't try to continue if bad typeId: we're lost or it's undefined
          raise KeyError, "LdfContributor.read: Unrecognized contribution %08x"%(typeId)
        data = self.__socket.recvall(length - hdrLen) # Only the LATcontribution(s) portion
        summary = struct.unpack('>L', data[3*4 : 4*4])[0]
        if (summary & markerMsk) == sweepFld:
          return geb.Geb.Sweep, status, LdfContribution(self, data)
        else:
          return geb.Geb.Build,     status, LdfContribution(self, data)

      rd, wr, ex = select([self.__socket.socket()], [], [], self.__timeout)

    return geb.Geb.Terminate, None, None


class AncillaryContributor(geb.GebContributor):
  """!\brief Concrete contributor class specifically meant to work with EvtCli.
  """
  __typeIds = [AncillaryContribution.HeaderId,
               AncillaryContribution.TrailerId,
               AncillaryContribution.EventId]
  def __init__(self, (host, port), offset = 0L):
    geb.GebContributor.__init__(self)
    self.__socket    = Socket(port)
    self.__timeout   = 1                # Same as EvtCliIO.__timeout
    self.__host      = host
    self.__port      = port
    self.__offset    = offset

  def name(self):
    return "AncContributor"

  def sequence(self, sequence):
    """!\brief Method for remapping the ancillary contribution's sequence number
    into the same space as the other contributions' sequence numbers.
    """
    # offset does double duty by compensating for the off-by-one nature of the
    # two event number spaces and by keeping track of the event number extention
    # needed to handle the raw 17 bit event number roll-over problem
    seq = sequence - self.__offset
    if sequence == 0x0fffffff:  self.__offset -= 0x10000000L
    return seq

  def open(self):
    """!\brief Method to open and connect to the server socket.
    It retries until a connection can be established, or the quit flag
    is seen to be set.
    """
    self.__quit = False
    while not self.__quit:
      try:
        self.__socket.connect(self.__host)
        break
      except Exception, e:
        import logging
        logging.debug("AncillaryContributor.open: %s" % (e))
        time.sleep(1.0)                 # Don't retry too quickly
    import logging
    logging.debug("AncillaryContributor.open: socket opened")

  def close(self):
    """!\brief Method to signal the read loop to exit.
    """
    self.__quit = True

  def read(self):
    """!\brief Method to read an ancillary contribution from a socket.

    This method deals with reconnecting when the server restarted.
    """
    self.__quit = 0
    while not self.__quit:
      try:
        return self.__read()
      except SocketClosed:
        import logging
        logging.debug("AncillaryContributor.read: socket closed")
        self.__socket.disconnect()
        self.__socket = Socket(self.__port)
        self.open()
      except Exception, e:
        if "Connection reset by peer" in str(e):
          import logging
          logging.debug("AncillaryContributor.read: socket closed")
          self.__socket.disconnect()
          self.__socket = Socket(self.__port)
          self.open()
        else:
          import logging
          logging.exception("AncillaryContributor.read: %s" % (str(e)))
          # Do we need to do something here?
          break

    return geb.Geb.Terminate, None, None

  def __read(self):
    """!\brief Method to read an ancillary contribution from a socket.
    """
    self.__quit = 0
    rd, wr, ex = select([self.__socket.socket()], [], [], self.__timeout)
    initLen = 8
    while not self.__quit:
      if rd:
        data            = self.__socket.recvall(initLen)
        if len(data) != initLen:
          # This will happen only when the socket closed
          raise SocketClosed
        vTypeId, length = struct.unpack('<2L', data)
        typeId = vTypeId & 0x000fffff
        if typeId not in AncillaryContributor.__typeIds:
          # Don't try to continue if bad typeId: we're lost or it's undefined
          raise KeyError, "AncillaryContributor.read: Unrecognized contribution %08x" % (typeId)
        length &= 0x0fffffff
        data  += self.__socket.recvall(length - initLen)
        data = array('I', data)
        data.byteswap()                 # Get it into big endian form
        data = data.tostring()

        status = 0
        if typeId == AncillaryContribution.EventId:
          return geb.Geb.Build,     status, AncillaryContribution(self, data)
        else:
          return geb.Geb.DontBuild, status, AncillaryContribution(self, data)

      rd, wr, ex = select([self.__socket.socket()], [], [], self.__timeout)

    self.__socket.disconnect()
    return geb.Geb.Terminate, None, None


class Merger(geb.Geb):
  def __init__(self, contributors, queue):
    geb.Geb.__init__(self, contributors)
    self.__queue = queue

  def dispose(self, buffer):
    """!\brief Method called by the base class to dispose of an event.
    """
    self.__queue.put(buffer)


def getHostPort(hostPort):
  host_port = hostPort.split(':')
  host = host_port[0]
  if len(host_port) == 1:
    port = 2345
  else:
    port = int(host_port[1])
  return host, port


class EvtCliBT(EvtCli):
  AncHostPort = "localhost:2345"
  def __init__(self, debug = 0,
               port        = EvtCliIO.PortNumber,
               ancHostPort = "localhost:2345"):  #EvtCliBT.AncHostPort):
    EvtCli.__init__(self, debug, port)
    self.__timeout = 1                  # same as EvtCliIO.__timeout

    ldfContributor = LdfContributor(self)
    ancContributor = AncillaryContributor(getHostPort(ancHostPort))
    self.__queue   = Queue.Queue()
    self.__merger  = Merger([ldfContributor, ancContributor], self.__queue)
    #self.__merger  = Merger([ldfContributor], self.__queue)
    self.__merger.start()

  def readEvent(self):
    """\brief Method used to read an event from the event merging code.
    """
    self.__abort = 0
    while not self.__abort:
      try:
        return self.__queue.get(block = True, timeout = self.__timeout)
      except Queue.Empty:
        if self.isTimeoutEnabled():
          raise IOError, "EvtCliBT.readEvent: Event read timeout"
    # Abort
    raise RuntimeError, "EvtCliBT.readEvent: Aborted"

  def flushAll(self):
    self.__merger.flushAll()

  def abort(self):
    """\brief Abort out of a readEvent().

    Can be used from another thread
    """
    self.__abort = True

  def shutdown(self):
    """\brief Method used to shut down the event merging threads.

    This must be called from a different thread than the one in which the
    merger runs.
    """
    self.__merger.shutdown()
    self.__merger.join()   # Wait for the Event Builder thread to exit

---