src/python/oftest/dataplane.py

"""
OpenFlow Test Framework

DataPlane and DataPlanePort classes

Provide the interface to the control the set of ports being used
to stimulate the switch under test.

See the class dataplaneport for more details.  This class wraps
a set of those objects allowing general calls and parsing
configuration.

@todo Add "filters" for matching packets.  Actions supported
for filters should include a callback or a counter
"""

import sys
import os
import socket
import time
import netutils
from threading import Thread
from threading import Lock
from threading import Condition
import select
import logging
from oft_assert import oft_assert

##@todo Find a better home for these identifiers (dataplane)
RCV_SIZE_DEFAULT = 4096
ETH_P_ALL = 0x03
RCV_TIMEOUT = 10000

def match_exp_pkt(exp_pkt, pkt):
    """
    Compare the string value of pkt with the string value of exp_pkt,
    and return True iff they are identical.  If the length of exp_pkt is
    less than the minimum Ethernet frame size (60 bytes), then padding
    bytes in pkt are ignored.
    """
    e = str(exp_pkt)
    p = str(pkt)
    if len(e) < 60:
        p = p[:len(e)]
    return e == p


class DataPlanePort(Thread):
    """
    Class defining a port monitoring object.

    Control a dataplane port connected to the switch under test.
    Creates a promiscuous socket on a physical interface.
    Queues the packets received on that interface with time stamps.
    Inherits from Thread class as meant to run in background.  Also
    supports polling.
    Use accessors to dequeue packets for proper synchronization.

    Currently assumes a controlling 'parent' which maintains a
    common Lock object and a total packet-pending count.  May want
    to decouple that some day.
    """

    def __init__(self, interface_name, port_number, parent, max_pkts=1024):
        """
        Set up a port monitor object
        @param interface_name The name of the physical interface like eth1
        @param port_number The port number associated with this port
        @param parent The controlling dataplane object; for pkt wait CV
        @param max_pkts Maximum number of pkts to keep in queue
        """
        Thread.__init__(self)
        self.interface_name = interface_name
        self.max_pkts = max_pkts
        self.packets_total = 0
        self.packets = []
        self.packets_discarded = 0
        self.port_number = port_number
        logname = "dp-" + interface_name
        self.logger = logging.getLogger(logname)
        try:
            self.socket = self.interface_open(interface_name)
        except:
            self.logger.info("Could not open socket")
            sys.exit(1)
        self.logger.info("Openned port monitor socket")
        self.parent = parent
        self.pkt_sync = self.parent.pkt_sync

    def interface_open(self, interface_name):
        """
        Open a socket in a promiscuous mode for a data connection.
        @param interface_name port name as a string such as 'eth1'
        @retval s socket
        """
        s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW,
                          socket.htons(ETH_P_ALL))
        s.bind((interface_name, 0))
        netutils.set_promisc(s, interface_name)
        s.settimeout(RCV_TIMEOUT)
        return s

    def run(self):
        """
        Activity function for class
        """
        self.running = True
        self.socs = [self.socket]
        error_warned = False # Have we warned about error?
        while self.running:
            try:
                sel_in, sel_out, sel_err = \
                    select.select(self.socs, [], [], 1)
            except:
                print sys.exc_info()
                self.logger.error("Select error, exiting")
                break

            if not self.running:
                break

            if (sel_in is None) or (len(sel_in) == 0):
                continue

            try:
                rcvmsg = self.socket.recv(RCV_SIZE_DEFAULT)
            except socket.error:
                if not error_warned:
                    self.logger.info("Socket error on recv")
                    error_warned = True
                continue

            if len(rcvmsg) == 0:
                self.logger.info("Zero len pkt rcvd")
                self.kill()
                break

            rcvtime = time.clock()
            self.logger.debug("Pkt len " + str(len(rcvmsg)) +
                     " in at " + str(rcvtime) + " on port " +
                     str(self.port_number))

            # Enqueue packet
            self.pkt_sync.acquire()
            if len(self.packets) >= self.max_pkts:
                # Queue full, throw away oldest
                self.packets.pop(0)
                self.packets_discarded += 1
            else:
                self.parent.packets_pending += 1
            # Check if parent is waiting on this (or any) port
            drop_pkt = False
            if self.parent.want_pkt:
                if (not self.parent.want_pkt_port or
                        self.parent.want_pkt_port == self.port_number):
                    if self.parent.exp_pkt:
                        if not match_exp_pkt(self.parent.exp_pkt, rcvmsg):
                            drop_pkt = True
                    if not drop_pkt:
                        self.parent.got_pkt_port = self.port_number
                        self.parent.want_pkt = False
                        self.parent.pkt_sync.notify()
            if not drop_pkt:
                self.packets.append((rcvmsg, rcvtime))
                self.packets_total += 1
            self.pkt_sync.release()

        self.logger.info("Thread exit ")

    def kill(self):
        """
        Terminate the running thread
        """
        self.logger.debug("Port monitor kill")
        self.running = False
        try:
            self.socket.close()
        except:
            self.logger.info("Ignoring dataplane soc shutdown error")

    def dequeue(self, use_lock=True):
        """
        Get the oldest packet in the queue
        @param use_lock If True, acquires the packet sync lock (which is
        really the parent's lock)
        @return The pair packet, packet time-stamp
        """
        if use_lock:
            self.pkt_sync.acquire()
        if len(self.packets) > 0:
            pkt, pkt_time = self.packets.pop(0)
            self.parent.packets_pending -= 1
        else:
            pkt = pkt_time = None
        if use_lock:
            self.pkt_sync.release()
        return pkt, pkt_time

    def timestamp_head(self):
        """
        Return the timestamp of the head of queue or None if empty
        """
        rv = None
        try:
            rv = self.packets[0][1]
        except:
            rv = None
        return rv

    def flush(self):
        """
        Clear the packet queue
        """
        self.pkt_sync.acquire()
        self.packets_discarded += len(self.packets)
        self.parent.packets_pending -= len(self.packets)
        self.packets = []
        self.packet_times = []
        self.pkt_sync.release()


    def send(self, packet):
        """
        Send a packet to the dataplane port
        @param packet The packet data to send to the port
        @retval The number of bytes sent
        """
        return self.socket.send(packet)


    def register(self, handler):
        """
        Register a callback function to receive packets from this
        port.  The callback will be passed the packet, the
        interface name and the port number (if set) on which the
        packet was received.

        To be implemented
        """
        pass

    def show(self, prefix=''):
        print prefix + "Name:          " + self.interface_name
        print prefix + "Pkts pending:  " + str(len(self.packets))
        print prefix + "Pkts total:    " + str(self.packets_total)
        print prefix + "socket:        " + str(self.socket)


class DataPlane:
    """
    Class defining access primitives to the data plane
    Controls a list of DataPlanePort objects
    """
    def __init__(self, config=None):
        self.port_list = {}
        # pkt_sync serves double duty as a regular top level lock and
        # as a condition variable
        self.pkt_sync = Condition()

        # These are used to signal async pkt arrival for polling
        self.want_pkt = False
        self.exp_pkt = None
        self.want_pkt_port = None # What port required (or None)
        self.got_pkt_port = None # On what port received?
        self.packets_pending = 0 # Total pkts in all port queues
        self.logger = logging.getLogger("dataplane")

        if config is None:
            self.config = {}
        else:
            self.config = config; 

        ############################################################
        #
        # We use the DataPlanePort class defined here by 
        # default for all port traffic:
        #
        self.dppclass = DataPlanePort

        ############################################################
        #
        # The platform/config can provide a custom DataPlanePort class
        # here if you have a custom implementation with different
        # behavior. 
        #
        # Set config.dataplane.portclass = MyDataPlanePortClass
        # where MyDataPlanePortClass has the same interface as the class
        # DataPlanePort defined here. 
        #
        if "dataplane" in self.config:
            if "portclass" in self.config["dataplane"]:
                self.dppclass = self.config["dataplane"]["portclass"]

        if self.dppclass == None:
            raise Exception("Problem determining DataPlanePort class.")


    def port_add(self, interface_name, port_number):
        """
        Add a port to the dataplane
        TBD:  Max packets for queue?
        @param interface_name The name of the physical interface like eth1
        @param port_number The port number used to refer to the port
        """

        self.port_list[port_number] = self.dppclass(interface_name, 
                                                    port_number, self); 

        self.port_list[port_number].start()



    def send(self, port_number, packet):
        """
        Send a packet to the given port
        @param port_number The port to send the data to
        @param packet Raw packet data to send to port
        """
        self.logger.debug("Sending %d bytes to port %d" %
                          (len(packet), port_number))
        bytes = self.port_list[port_number].send(packet)
        if bytes != len(packet):
            self.logger.error("Unhandled send error, length mismatch %d != %d" %
                     (bytes, len(packet)))
        return bytes

    def flood(self, packet):
        """
        Send a packet to all ports
        @param packet Raw packet data to send to port
        """
        for port_number in self.port_list.keys():
            bytes = self.port_list[port_number].send(packet)
            if bytes != len(packet):
                self.logger.error("Unhandled send error" +
                         ", port %d, length mismatch %d != %d" %
                         (port_number, bytes, len(packet)))

    def _oldest_packet_find(self):
        # Find port with oldest packet
        min_time = 0
        min_port = -1
        for port_number in self.port_list.keys():
            ptime = self.port_list[port_number].timestamp_head()
            if ptime:
                if (min_port == -1) or (ptime < min_time):
                    min_time = ptime
                    min_port = port_number
        oft_assert(min_port != -1, "Could not find port when pkts pending")

        return min_port

    def poll(self, port_number=None, timeout=None, exp_pkt=None):
        """
        Poll one or all dataplane ports for a packet

        If port_number is given, get the oldest packet from that port.
        Otherwise, find the port with the oldest packet and return
        that packet.

        If exp_pkt is true, discard all packets until that one is found

        @param port_number If set, get packet from this port
        @param timeout If positive and no packet is available, block
        until a packet is received or for this many seconds
        @param exp_pkt If not None, look for this packet and ignore any
        others received.  Note that if port_number is None, all packets
        from all ports will be discarded until the exp_pkt is found
        @return The triple port_number, packet, pkt_time where packet
        is received from port_number at time pkt_time.  If a timeout
        occurs, return None, None, None
        """


        if exp_pkt and not port_number:
            self.logger.warn("Dataplane poll with exp_pkt but no port number")

        self.pkt_sync.acquire()

        # Check if requested specific port and it has a packet
        if port_number and len(self.port_list[port_number].packets) != 0:
            while len(self.port_list[port_number].packets) != 0:
                pkt, time = self.port_list[port_number].dequeue(use_lock=False)
                if not exp_pkt:
                    break
                if match_exp_pkt(exp_pkt, pkt):
                    break
                pkt = None # Discard silently
            if pkt:
                self.pkt_sync.release()
                oft_assert(pkt, "Poll: packet not found on port " +
                           str(port_number))
                return port_number, pkt, time

        # Check if requested any port and some packet pending
        if not port_number:
            while self.packets_pending != 0:
                port = self._oldest_packet_find()
                pkt, time = self.port_list[port].dequeue(use_lock=False)
                self.pkt_sync.release()
                oft_assert(pkt, "Poll: oldest packet not found")
                if not exp_pkt or match_exp_pkt(exp_pkt, pkt):
                    return port, pkt, time

        # No packet pending; blocking call requested?
        if not timeout:
            self.pkt_sync.release()
            return None, None, None

        # Desired packet isn't available and timeout is specified
        # Already holding pkt_sync; wait on pkt_sync variable
        self.want_pkt = True
        self.exp_pkt = exp_pkt
        self.want_pkt_port = port_number
        self.got_pkt_port = None
        self.pkt_sync.wait(timeout)
        self.want_pkt = False
        self.exp_pkt = None
        if self.got_pkt_port:
            pkt, time = \
                self.port_list[self.got_pkt_port].dequeue(use_lock=False)
            self.pkt_sync.release()
            oft_assert(pkt, "Poll: pkt reported, but not found at " +
                       str(self.got_pkt_port))
            return self.got_pkt_port, pkt, time

        self.pkt_sync.release()
        self.logger.debug("Poll time out, no packet from " + str(port_number))

        return None, None, None

    def kill(self, join_threads=True):
        """
        Close all sockets for dataplane
        @param join_threads If True call join on each thread
        """
        for port_number in self.port_list.keys():
            self.port_list[port_number].kill()
            if join_threads:
                self.logger.debug("Joining " + str(port_number))
                self.port_list[port_number].join()

        self.logger.info("DataPlane shutdown")

    def show(self, prefix=''):
        print prefix + "Dataplane Controller"
        print prefix + "Packets pending" + str(self.packets_pending)
        for pnum, port in self.port_list.items():
            print prefix + "OpenFlow Port Number " + str(pnum)
            port.show(prefix + '  ')