DataDistributor.py

Go to the documentation of this file.

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

__facility__ = "Online"
__abstract__ = "GLAST LAT data distribution wrapper classes"
__author__   = "R. Claus <Claus@SLAC.Stanford.edu> SLAC - GLAST LAT I&T/Online"
__date__     = "2005/07/23 00:08:27"
__updated__  = "$Date: 2005/09/22 00:29:31 $"
__version__  = "$Revision: 1.6 $"
__release__  = "$Name: HEAD $"
__credits__  = "SLAC"

import LICOS.copyright_SLAC


import socket
import struct


class DataDistributorServer(object):
  """!\brief Data Distributor Server.

  """
  def __init__(self, name = None, size = 32*1024):
    """Data Distributor server constructor.

    \param name The name of the type of data to be served.
                See the DataDistributorAddress class.
    \param size The size in bytes of the buffer to allocate for sending the
                data.  This is used to set the socket send buffer size.
    """
    self.__socket = None
    self.__group  = None
    self.__port   = DataDistributorAddress.port(name)
    self.__size   = size

  def connect(self, server, interface = socket.gethostname()):
    """!\brief Method to set up a socket and connect to it.

    \param server    The address of a socket on which to serve the data.  This
                     address must be the same for both server and client.  The
                     value can be in one of the following forms:
                     - A number of which only the two lower decimal digits are
                       used, i.e., in the range 0-99
                     - <port>@<multicast address>, where <port> is a value that
                       will override what is found from the data type name
                       translation, and <multicast address> is some valid
                       multicast IP address starting with the value 239.  A name
                       may be used for the <multicast address> value as well, as
                       long as it can be found in the name server and translates
                       to a valid multicast IP address.
                     - <multicast address>, as discussed in the previous bullet.
                       The port value is taken to be the one found from looking up
                       the data type name in the port table.
    \param interface The network interface to send multicast traffic on.
                     Defaults to 'the primary network interface', however
                     'primary' is defined.  See socket.gethostbyname_ex()
                     documentation.
    """
    if self.__socket is None:
      self.__port, self.__group = DataDistributorAddress.lookup(self.__port,
                                                                server)

      # Look up multicast group address in name server
      # (doesn't hurt if it is already in ddd.ddd.ddd.ddd format)
      group = socket.gethostbyname(self.__group)

      # Any chance this sort of looks like a valid multicast group?
      firstByte = int(group.split(".", 1)[0])
      if firstByte < 224 and firstByte > 239:
        raise RuntimeError, \
          "Group %s does not resolve to a multicast address" % self.__group

      s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
      self.__socket = s
      #
      # Make sure there is sufficient network buffering
      s.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, self.__size)
      #
      # Construct interface address
      # Use gethostbyname and gethostname to force the connection onto the
      # primary network interface.
      intf = socket.gethostbyname_ex(interface)[2][0]
      #
      # Specify the interface for outgoing multicast datagrams
      s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(intf) + socket.inet_aton('0.0.0.0'))
    else:
      raise RuntimeError, "Socket already in use"

  def disconnect(self):
    """!\brief Gracefully shut down the socket.

    """
    if self.__socket is not None:
      self.__socket.close()
      self.__socket = None
      self.__group  = None
      self.__port   = None
    else:
      raise RuntimeError, "No socket found"

  def publish(self, data):
    """!\brief Publish the data on the multicast socket.

    \param data The data to be sent out on the multicast socket.

    """
    if self.__socket is not None:
      self.__socket.sendto(data, (self.__group, self.__port))
    else:
      raise RuntimeError, "No socket found"


class DataDistributorClient(object):
  """!\brief Data distributor client class.

  """
  __timeout = 1

  def __init__(self, name = None, size = 64*1024):
    """!\brief Data Distributor client constructor.

    \param name The name of the type of data that the client wants to receive.
                See the DataDistributorAddress class.
    \param size The size in bytes of the buffer to allocate for receiving the
                data.  Defaults to 64 KB.
    """
    self.__socket     = None
    self.__group      = None
    self.__port       = DataDistributorAddress.port(name)
    self.__size       = size
    self.__returnNone = None

  # Open a UDP socket, bind it to a port and select a multicast group
  def connect(self, server):
    """!\brief Open a UDP socket, bind it to a port and select a multicast group.

    \param server The address the server is multicasting on.  This
                  address must be the same for both server and client.  The
                  value can be in one of the following forms:
                  - A number of which only the two lower decimal digits are
                    used, i.e., in the range 0-99
                  - <port>@<multicast address>, where <port> is a value that
                    will override what is found from the data type name
                    translation, and <multicast address> is some valid multicast
                    IP address starting with the value 239.  A name may be used
                    for the <multicast address> value as well, as long as it can
                    be found in the name server and translates to a valid
                    multicast IP address.
                  - <multicast address>, as discussed in the previous bullet.
                    The port value is taken to be the one found from looking up
                    the data type name in the port table.

    """
    if self.__socket is None:
      self.__port, self.__group = DataDistributorAddress.lookup(self.__port,
                                                                server)
      #print "port@group =", str(self.__port) + '@' + self.__group
      # Import modules used only here
      import string
      import struct
      #
      # Create a socket
      s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
      self.__socket = s
      #
      # Allow multiple copies of this program on one machine
      # (not strictly needed)
      s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
      try:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
      except AttributeError:
        pass # Some systems don't support SO_REUSEPORT
      #
      # Bind it to the port
      s.bind(('', self.__port))
      #
      # Look up multicast group address in name server
      # (doesn't hurt if it is already in ddd.ddd.ddd.ddd format)
      group = socket.gethostbyname(self.__group)
      #
      # Any chance this sort of looks like a valid multicast group?
      bytes = map(int, group.split("."))
      if bytes[0] < 224 and bytes[0] > 239:
        raise RuntimeError, \
          "Group %s does not resolve to a multicast address" % self.__group
      #
      # Make sure there is sufficient network buffering to receive stuff in
      s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, self.__size)
      #
      # Add group membership
      s.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP,
                   socket.inet_aton(group) + socket.inet_aton('0.0.0.0'))
    else:
      raise RuntimeError, "Socket already in use"

  def disconnect(self):
    """!\brief Gracefully shut down the socket.

    """
    s = self.__socket
    if s is not None:
      s.setsockopt(socket.IPPROTO_IP, socket.IP_DROP_MEMBERSHIP,
                   socket.inet_aton(self.__group) + socket.inet_aton('0.0.0.0'))
      s.close()
      self.__socket = None
      self.__group  = None
      self.__port   = None
    else:
      raise RuntimeError, "No socket found"

  def returnNone(self):
    """!\brief Method to signal the receive loop to exit.

    """
    self.__returnNone = 1

  def receive(self):
    """!\brief Receive multicast data and the sender's identification.

    \return A list of the data and the sender's identification string.
    """
    if self.__socket is not None:
      import select
      while self.__returnNone is None:
        rd, wr, ex = select.select([self.__socket], [], [], self.__timeout)
        if rd:
          data, sender = self.__socket.recvfrom(self.__size)
          return data, sender
        elif ex:
          return None, None
    else:
      raise RuntimeError, "No socket found"
    return None, None

class DataDistributorAddress(object):
  """!\brief Data distributor address class containing static methods that
  govern the picking of multicast ports and addresses.

  """
  # See http://www.iana.org/assignments/port-numbers on how to pick a port
  # I randomly chose the following from the dynamic ports section:
  __PORT_TEMPLATE = 54321
  # See http://www.iana.org/assignments/multicast-addresses on how to pick a
  # group BaBar uses 239.255.0.2- 239.255.0.130, so I chose the following:
  __GROUP_TEMPLATE = '239.255.1.%u'     # From "site-local" range (Stevens)

  __ports = {"LAT_EBFdata" : __PORT_TEMPLATE + 0,
             "LAT_LDFdata" : __PORT_TEMPLATE + 0, # Alias for EBFdata
             "LAT_CMDdata" : __PORT_TEMPLATE + 1,
             "LAT_HSKdata" : __PORT_TEMPLATE + 2,
             "LAT_MsgLog"  : __PORT_TEMPLATE + 3}

  def port(name):
    """!\brief Static method to select a physical multicast port to use based on the
       type of data that is desired.

    \param name The name of the type of data that is desired.  Choose from
                'LAT_EBFdata', 'LAT_CMDdata', 'LAT_HSKdata' and 'LAT_MsgLog'.
    """
    if name in DataDistributorAddress.__ports:
      return DataDistributorAddress.__ports[name]
    return None

  def lookup(port, server):
    """!\brief Static method for resolving the multicast port and group values.

    \param port   An integer containing the multicast port to use, e.g. from
                  the port() method above, or None to indicate that the value
                  will come from the supplied server string.
    \param server A string that contains either the machine ID (test stand
                  instance number), or <port>@<group IP or name>, or <group IP
                  or name> when a port number is also supplied.  These last two
                  can be used to override the values chosen by this class.  The
                  machine ID is a number from 0 - 999, where the hundreds
                  digit represents the site ID and remainder (0-99) is the
                  machine ID at the site.  This latter portion is what's used.
    \return A list made up of the resolved port and group values.
    """
    # Do a DNS like action to get the machine ID of the server
    # Until we get that built, let's do the following:

    # The 'server' argument is either a machine ID or a string of the form
    # port@<multicast group> (multicast group could be an IP address or a name)
    try:
      # Site machine IDs run from 0 to 99.  Remove the site ID to prevent > 255
      group = DataDistributorAddress.__GROUP_TEMPLATE % (int(server) % 100)
    except ValueError:                  # Can only be <port>@<group> or <group>
      try:
        port, group = server.split('@', 1)
        port = int(port)
      except ValueError:                # Must be <group>
        group = server
    return port, group

  port   = staticmethod(port)
  lookup = staticmethod(lookup)


if __name__ == "__main__":
  """Testing routine.

  Usage: Run DataDistributor (this file) in one session to start up the client
         and DataDistributor -s in another session to start up the server.
  """
  import time

  def sender():
    dds = DataDistributorServer('test')
    #dds.connect('54320@224.0.0.0')      # "node-local" according to Stevens
    dds.connect('54320@239.255.1.0')    # "site-local" according to Stevens
    #dds.connect('54320@239.255.1.0', "localhost")  # mcast to self via primary
                                         # interface doesn't seem to work on XP
                                         # with Checkpoint enabled
    i = 5
    while i > 0:  #1:
      i -= 1
      data = `time.time()`
      data += 50000*'-'                 # Make it a long string
      #print "data =", data
      dds.publish(data)
      time.sleep(1)
    dds.publish('Quit')
    dds.disconnect()

  def receiver():
    dds = DataDistributorClient('test', 64*1024)
    #dds.connect('54320@224.0.0.0')      # "node-local" according to Stevens
    dds.connect('54320@239.255.1.0')    # "site-local" according to Stevens
    #dds.connect('9875@sap.mcast.net')   # Didn't see any activity
    #dds.connect('123@ntp.mcast.net')    # Didn't see any activity

    # Loop, printing any data we receive
    while 1:
      data, sender = dds.receive()
      print sender, ':', `data[:40]`, len(data)
      if data == 'Quit':  break

    dds.disconnect()

  #---
  import sys

  flags = sys.argv[1:]

  if flags:
    sender()
  else:
    receiver()

---