--- /dev/null
+# Copyright (C) 2013 Nippon Telegraph and Telephone Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import struct
+
+from ryu.base import app_manager
+from ryu.controller import event
+from ryu.controller import ofp_event
+from ryu.controller.handler import DEAD_DISPATCHER
+from ryu.controller.handler import MAIN_DISPATCHER
+from ryu.controller.handler import set_ev_cls
+from ryu.ofproto import ether
+from ryu.ofproto import inet
+from ryu.ofproto import ofproto_v1_0
+from ryu.ofproto import ofproto_v1_2
+from ryu.ofproto import ofproto_v1_3
+from ryu.lib import addrconv
+from ryu.lib import hub
+from ryu.lib.dpid import dpid_to_str
+from ryu.lib.packet import packet
+from ryu.lib.packet import ethernet
+from ryu.lib.packet import ipv4
+from ryu.lib.packet import igmp
+
+
+class EventPacketIn(event.EventBase):
+ """a PacketIn event class using except IGMP."""
+
+ def __init__(self, msg):
+ """initialization."""
+ super(EventPacketIn, self).__init__()
+ self.msg = msg
+
+
+MG_GROUP_ADDED = 1
+MG_MEMBER_CHANGED = 2
+MG_GROUP_REMOVED = 3
+
+
+class EventMulticastGroupStateChanged(event.EventBase):
+ """a event class that notifies the changes of the statuses of the
+ multicast groups."""
+
+ def __init__(self, reason, address, src, dsts):
+ """
+ ========= =====================================================
+ Attribute Description
+ ========= =====================================================
+ reason why the event occurs. use one of MG_*.
+ address a multicast group address.
+ src a port number in which a querier exists.
+ dsts a list of port numbers in which the members exist.
+ ========= =====================================================
+ """
+ super(EventMulticastGroupStateChanged, self).__init__()
+ self.reason = reason
+ self.address = address
+ self.src = src
+ self.dsts = dsts
+
+
+class IgmpLib(app_manager.RyuApp):
+ """IGMP snooping library."""
+
+ # -------------------------------------------------------------------
+ # PUBLIC METHODS
+ # -------------------------------------------------------------------
+ def __init__(self):
+ """initialization."""
+ super(IgmpLib, self).__init__()
+ self.name = 'igmplib'
+ self._querier = IgmpQuerier()
+ self._snooper = IgmpSnooper(self.send_event_to_observers)
+
+ def set_querier_mode(self, dpid, server_port):
+ """set a datapath id and server port number to the instance
+ of IgmpQuerier.
+
+ ============ ==================================================
+ Attribute Description
+ ============ ==================================================
+ dpid the datapath id that will operate as a querier.
+ server_port the port number linked to the multicasting server.
+ ============ ==================================================
+ """
+ self._querier.set_querier_mode(dpid, server_port)
+
+ # -------------------------------------------------------------------
+ # PUBLIC METHODS ( EVENT HANDLERS )
+ # -------------------------------------------------------------------
+ @set_ev_cls(ofp_event.EventOFPPacketIn, MAIN_DISPATCHER)
+ def packet_in_handler(self, evt):
+ """PacketIn event handler. when the received packet was IGMP,
+ proceed it. otherwise, send a event."""
+ msg = evt.msg
+ dpid = msg.datapath.id
+
+ req_pkt = packet.Packet(msg.data)
+ req_igmp = req_pkt.get_protocol(igmp.igmp)
+ if req_igmp:
+ if self._querier.dpid == dpid:
+ self._querier.packet_in_handler(req_igmp, msg)
+ else:
+ self._snooper.packet_in_handler(req_pkt, req_igmp, msg)
+ else:
+ self.send_event_to_observers(EventPacketIn(msg))
+
+ @set_ev_cls(ofp_event.EventOFPStateChange,
+ [MAIN_DISPATCHER, DEAD_DISPATCHER])
+ def state_change_handler(self, evt):
+ """StateChange event handler."""
+ datapath = evt.datapath
+ assert datapath is not None
+ if datapath.id == self._querier.dpid:
+ if evt.state == MAIN_DISPATCHER:
+ self._querier.start_loop(datapath)
+ elif evt.state == DEAD_DISPATCHER:
+ self._querier.stop_loop()
+
+
+class IgmpBase(object):
+ """IGMP abstract class library."""
+
+ # -------------------------------------------------------------------
+ # PUBLIC METHODS
+ # -------------------------------------------------------------------
+ def __init__(self):
+ self._set_flow_func = {
+ ofproto_v1_0.OFP_VERSION: self._set_flow_entry_v1_0,
+ ofproto_v1_2.OFP_VERSION: self._set_flow_entry_v1_2,
+ ofproto_v1_3.OFP_VERSION: self._set_flow_entry_v1_2,
+ }
+ self._del_flow_func = {
+ ofproto_v1_0.OFP_VERSION: self._del_flow_entry_v1_0,
+ ofproto_v1_2.OFP_VERSION: self._del_flow_entry_v1_2,
+ ofproto_v1_3.OFP_VERSION: self._del_flow_entry_v1_2,
+ }
+
+ # -------------------------------------------------------------------
+ # PROTECTED METHODS ( RELATED TO OPEN FLOW PROTOCOL )
+ # -------------------------------------------------------------------
+ def _set_flow_entry_v1_0(self, datapath, actions, in_port, dst,
+ src=None):
+ ofproto = datapath.ofproto
+ parser = datapath.ofproto_parser
+
+ match = parser.OFPMatch(
+ dl_type=ether.ETH_TYPE_IP, in_port=in_port,
+ nw_src=self._ipv4_text_to_int(src),
+ nw_dst=self._ipv4_text_to_int(dst))
+ mod = parser.OFPFlowMod(
+ datapath=datapath, match=match, cookie=0,
+ command=ofproto.OFPFC_ADD, actions=actions)
+ datapath.send_msg(mod)
+
+ def _set_flow_entry_v1_2(self, datapath, actions, in_port, dst,
+ src=None):
+ ofproto = datapath.ofproto
+ parser = datapath.ofproto_parser
+
+ match = parser.OFPMatch(
+ eth_type=ether.ETH_TYPE_IP, in_port=in_port, ipv4_dst=dst)
+ if src is not None:
+ match.append_field(ofproto.OXM_OF_IPV4_SRC, src)
+ inst = [parser.OFPInstructionActions(
+ ofproto.OFPIT_APPLY_ACTIONS, actions)]
+ mod = parser.OFPFlowMod(
+ datapath=datapath, command=ofproto.OFPFC_ADD,
+ priority=65535, match=match, instructions=inst)
+ datapath.send_msg(mod)
+
+ def _set_flow_entry(self, datapath, actions, in_port, dst, src=None):
+ """set a flow entry."""
+ set_flow = self._set_flow_func.get(datapath.ofproto.OFP_VERSION)
+ assert set_flow
+ set_flow(datapath, actions, in_port, dst, src)
+
+ def _del_flow_entry_v1_0(self, datapath, in_port, dst, src=None):
+ ofproto = datapath.ofproto
+ parser = datapath.ofproto_parser
+
+ match = parser.OFPMatch(
+ dl_type=ether.ETH_TYPE_IP, in_port=in_port,
+ nw_src=self._ipv4_text_to_int(src),
+ nw_dst=self._ipv4_text_to_int(dst))
+ mod = parser.OFPFlowMod(
+ datapath=datapath, match=match, cookie=0,
+ command=ofproto.OFPFC_DELETE)
+ datapath.send_msg(mod)
+
+ def _del_flow_entry_v1_2(self, datapath, in_port, dst, src=None):
+ ofproto = datapath.ofproto
+ parser = datapath.ofproto_parser
+
+ match = parser.OFPMatch(
+ eth_type=ether.ETH_TYPE_IP, in_port=in_port, ipv4_dst=dst)
+ if src is not None:
+ match.append_field(ofproto.OXM_OF_IPV4_SRC, src)
+ mod = parser.OFPFlowMod(
+ datapath=datapath, command=ofproto.OFPFC_DELETE,
+ out_port=ofproto.OFPP_ANY, out_group=ofproto.OFPG_ANY,
+ match=match)
+ datapath.send_msg(mod)
+
+ def _del_flow_entry(self, datapath, in_port, dst, src=None):
+ """remove a flow entry."""
+ del_flow = self._del_flow_func.get(datapath.ofproto.OFP_VERSION)
+ assert del_flow
+ del_flow(datapath, in_port, dst, src)
+
+ def _do_packet_out(self, datapath, data, in_port, actions):
+ """send a packet."""
+ ofproto = datapath.ofproto
+ parser = datapath.ofproto_parser
+
+ out = parser.OFPPacketOut(
+ datapath=datapath, buffer_id=ofproto.OFP_NO_BUFFER,
+ data=data, in_port=in_port, actions=actions)
+ datapath.send_msg(out)
+
+ # -------------------------------------------------------------------
+ # PROTECTED METHODS ( OTHERS )
+ # -------------------------------------------------------------------
+ def _ipv4_text_to_int(self, ip_text):
+ """convert ip v4 string to integer."""
+ if ip_text is None:
+ return None
+ assert isinstance(ip_text, str)
+ return struct.unpack('!I', addrconv.ipv4.text_to_bin(ip_text))[0]
+
+
+class IgmpQuerier(IgmpBase):
+ """IGMP querier emulation class library.
+
+ this querier is a simplified implementation, and is not based on RFC,
+ for example as following points:
+ - ignore some constant values
+ - does not send a specific QUERY in response to LEAVE
+ - and so on
+ """
+
+ # -------------------------------------------------------------------
+ # PUBLIC METHODS
+ # -------------------------------------------------------------------
+ def __init__(self):
+ """initialization."""
+ super(IgmpQuerier, self).__init__()
+ self.name = "IgmpQuerier"
+ self.logger = logging.getLogger(self.name)
+ self.dpid = None
+ self.server_port = None
+
+ self._datapath = None
+ self._querier_thread = None
+
+ # the structure of self._macst
+ #
+ # +-------+------------------+
+ # | group | port: True/False |
+ # | +------------------+
+ # | |... |
+ # +-------+------------------+
+ # | ... |
+ # +--------------------------+
+ #
+ # group multicast address.
+ # port a port number which connect to the group member.
+ # the value indicates that whether a flow entry
+ # was registered.
+ self._mcast = {}
+
+ self._set_logger()
+
+ def set_querier_mode(self, dpid, server_port):
+ """set the datapath to work as a querier. note that you can set
+ up only the one querier. when you called this method several
+ times, only the last one becomes effective."""
+ self.dpid = dpid
+ self.server_port = server_port
+ if self._querier_thread:
+ hub.kill(self._querier_thread)
+ self._querier_thread = None
+
+ def packet_in_handler(self, req_igmp, msg):
+ """the process when the querier received IGMP."""
+ ofproto = msg.datapath.ofproto
+ if ofproto.OFP_VERSION == ofproto_v1_0.OFP_VERSION:
+ in_port = msg.in_port
+ else:
+ in_port = msg.match['in_port']
+ if (igmp.IGMP_TYPE_REPORT_V1 == req_igmp.msgtype or
+ igmp.IGMP_TYPE_REPORT_V2 == req_igmp.msgtype):
+ self._do_report(req_igmp, in_port, msg)
+ elif igmp.IGMP_TYPE_LEAVE == req_igmp.msgtype:
+ self._do_leave(req_igmp, in_port, msg)
+
+ def start_loop(self, datapath):
+ """start QUERY thread."""
+ self._datapath = datapath
+ self._querier_thread = hub.spawn(self._send_query)
+ self.logger.info("started a querier.")
+
+ def stop_loop(self):
+ """stop QUERY thread."""
+ hub.kill(self._querier_thread)
+ self._querier_thread = None
+ self._datapath = None
+ self.logger.info("stopped a querier.")
+
+ # -------------------------------------------------------------------
+ # PRIVATE METHODS ( RELATED TO IGMP )
+ # -------------------------------------------------------------------
+ def _send_query(self):
+ """ send a QUERY message periodically."""
+ timeout = 60
+ ofproto = self._datapath.ofproto
+ parser = self._datapath.ofproto_parser
+ if ofproto_v1_0.OFP_VERSION == ofproto.OFP_VERSION:
+ send_port = ofproto.OFPP_NONE
+ else:
+ send_port = ofproto.OFPP_ANY
+
+ # create a general query.
+ res_igmp = igmp.igmp(
+ msgtype=igmp.IGMP_TYPE_QUERY,
+ maxresp=igmp.QUERY_RESPONSE_INTERVAL * 10,
+ csum=0,
+ address='0.0.0.0')
+ res_ipv4 = ipv4.ipv4(
+ total_length=len(ipv4.ipv4()) + len(res_igmp),
+ proto=inet.IPPROTO_IGMP, ttl=1,
+ src='0.0.0.0',
+ dst=igmp.MULTICAST_IP_ALL_HOST)
+ res_ether = ethernet.ethernet(
+ dst=igmp.MULTICAST_MAC_ALL_HOST,
+ src=self._datapath.ports[ofproto.OFPP_LOCAL].hw_addr,
+ ethertype=ether.ETH_TYPE_IP)
+ res_pkt = packet.Packet()
+ res_pkt.add_protocol(res_ether)
+ res_pkt.add_protocol(res_ipv4)
+ res_pkt.add_protocol(res_igmp)
+ res_pkt.serialize()
+
+ flood = [parser.OFPActionOutput(ofproto.OFPP_FLOOD)]
+
+ while True:
+ # reset reply status.
+ for status in self._mcast.values():
+ for port in status.keys():
+ status[port] = False
+
+ # send a general query to the host that sent this message.
+ self._do_packet_out(
+ self._datapath, res_pkt.data, send_port, flood)
+ hub.sleep(igmp.QUERY_RESPONSE_INTERVAL)
+
+ # QUERY timeout expired.
+ del_groups = []
+ for group, status in self._mcast.items():
+ del_ports = []
+ actions = []
+ for port in status.keys():
+ if not status[port]:
+ del_ports.append(port)
+ else:
+ actions.append(parser.OFPActionOutput(port))
+ if len(actions) and len(del_ports):
+ self._set_flow_entry(
+ self._datapath, actions, self.server_port, group)
+ if not len(actions):
+ self._del_flow_entry(
+ self._datapath, self.server_port, group)
+ del_groups.append(group)
+ if len(del_ports):
+ for port in del_ports:
+ self._del_flow_entry(self._datapath, port, group)
+ for port in del_ports:
+ del status[port]
+ for group in del_groups:
+ del self._mcast[group]
+
+ rest_time = timeout - igmp.QUERY_RESPONSE_INTERVAL
+ hub.sleep(rest_time)
+
+ def _do_report(self, report, in_port, msg):
+ """the process when the querier received a REPORT message."""
+ datapath = msg.datapath
+ ofproto = datapath.ofproto
+ parser = datapath.ofproto_parser
+
+ if ofproto.OFP_VERSION == ofproto_v1_0.OFP_VERSION:
+ size = 65535
+ else:
+ size = ofproto.OFPCML_MAX
+
+ update = False
+ self._mcast.setdefault(report.address, {})
+ if in_port not in self._mcast[report.address]:
+ update = True
+ self._mcast[report.address][in_port] = True
+
+ if update:
+ actions = []
+ for port in self._mcast[report.address]:
+ actions.append(parser.OFPActionOutput(port))
+ self._set_flow_entry(
+ datapath, actions, self.server_port, report.address)
+ self._set_flow_entry(
+ datapath,
+ [parser.OFPActionOutput(ofproto.OFPP_CONTROLLER, size)],
+ in_port, report.address)
+
+ def _do_leave(self, leave, in_port, msg):
+ """the process when the querier received a LEAVE message."""
+ datapath = msg.datapath
+ parser = datapath.ofproto_parser
+
+ self._mcast.setdefault(leave.address, {})
+ if in_port in self._mcast[leave.address]:
+ self._del_flow_entry(
+ datapath, in_port, leave.address)
+ del self._mcast[leave.address][in_port]
+ actions = []
+ for port in self._mcast[leave.address]:
+ actions.append(parser.OFPActionOutput(port))
+ if len(actions):
+ self._set_flow_entry(
+ datapath, actions, self.server_port, leave.address)
+ else:
+ self._del_flow_entry(
+ datapath, self.server_port, leave.address)
+
+ # -------------------------------------------------------------------
+ # PRIVATE METHODS ( OTHERS )
+ # -------------------------------------------------------------------
+ def _set_logger(self):
+ """change log format."""
+ self.logger.propagate = False
+ hdl = logging.StreamHandler()
+ fmt_str = '[querier][%(levelname)s] %(message)s'
+ hdl.setFormatter(logging.Formatter(fmt_str))
+ self.logger.addHandler(hdl)
+
+
+class IgmpSnooper(IgmpBase):
+ """IGMP snooping class library."""
+
+ # -------------------------------------------------------------------
+ # PUBLIC METHODS
+ # -------------------------------------------------------------------
+ def __init__(self, send_event):
+ """initialization."""
+ super(IgmpSnooper, self).__init__()
+ self.name = "IgmpSnooper"
+ self.logger = logging.getLogger(self.name)
+ self._send_event = send_event
+
+ # the structure of self._to_querier
+ #
+ # +------+--------------+
+ # | dpid | 'port': port |
+ # | +--------------+
+ # | | 'ip': ip |
+ # | +--------------+
+ # | | 'mac': mac |
+ # +------+--------------+
+ # | ... |
+ # +---------------------+
+ #
+ # dpid datapath id.
+ # port a port number which connect to the querier.
+ # ip IP address of the querier.
+ # mac MAC address of the querier.
+ self._to_querier = {}
+
+ # the structure of self._to_hosts
+ #
+ # +------+-------+---------------------------------+
+ # | dpid | group | 'replied': True/False |
+ # | | +---------------------------------+
+ # | | | 'leave': leave |
+ # | | +-----------+--------+------------+
+ # | | | 'ports' | portno | 'out': out |
+ # | | | | +------------+
+ # | | | | | 'in': in |
+ # | | | +--------+------------+
+ # | | | | ... |
+ # | +-------+-----------+---------------------+
+ # | | ... |
+ # +------+-----------------------------------------+
+ # | ... |
+ # +------------------------------------------------+
+ #
+ # dpid datapath id.
+ # group multicast address.
+ # replied the value indicates whether a REPORT message was
+ # replied.
+ # leave a LEAVE message.
+ # portno a port number which has joined to the multicast
+ # group.
+ # out the value indicates whether a flow entry for the
+ # packet outputted to the port was registered.
+ # in the value indicates whether a flow entry for the
+ # packet inputted from the port was registered.
+ self._to_hosts = {}
+
+ self._set_logger()
+
+ def packet_in_handler(self, req_pkt, req_igmp, msg):
+ """the process when the snooper received IGMP."""
+ dpid = msg.datapath.id
+ ofproto = msg.datapath.ofproto
+ if ofproto.OFP_VERSION == ofproto_v1_0.OFP_VERSION:
+ in_port = msg.in_port
+ else:
+ in_port = msg.match['in_port']
+
+ log = "SW=%s PORT=%d IGMP received. " % (
+ dpid_to_str(dpid), in_port)
+ self.logger.debug(str(req_igmp))
+ if igmp.IGMP_TYPE_QUERY == req_igmp.msgtype:
+ self.logger.info(log + "[QUERY]")
+ (req_ipv4, ) = req_pkt.get_protocols(ipv4.ipv4)
+ (req_eth, ) = req_pkt.get_protocols(ethernet.ethernet)
+ self._do_query(req_igmp, req_ipv4, req_eth, in_port, msg)
+ elif (igmp.IGMP_TYPE_REPORT_V1 == req_igmp.msgtype or
+ igmp.IGMP_TYPE_REPORT_V2 == req_igmp.msgtype):
+ self.logger.info(log + "[REPORT]")
+ self._do_report(req_igmp, in_port, msg)
+ elif igmp.IGMP_TYPE_LEAVE == req_igmp.msgtype:
+ self.logger.info(log + "[LEAVE]")
+ self._do_leave(req_igmp, in_port, msg)
+ elif igmp.IGMP_TYPE_REPORT_V3 == req_igmp.msgtype:
+ self.logger.info(log + "V3 is not supported yet.")
+ self._do_flood(in_port, msg)
+ else:
+ self.logger.info(log + "[unknown type:%d]",
+ req_igmp.msgtype)
+ self._do_flood(in_port, msg)
+
+ # -------------------------------------------------------------------
+ # PRIVATE METHODS ( RELATED TO IGMP )
+ # -------------------------------------------------------------------
+ def _do_query(self, query, iph, eth, in_port, msg):
+ """the process when the snooper received a QUERY message."""
+ datapath = msg.datapath
+ dpid = datapath.id
+ ofproto = datapath.ofproto
+ parser = datapath.ofproto_parser
+
+ # learn the querier.
+ self._to_querier[dpid] = {
+ 'port': in_port,
+ 'ip': iph.src,
+ 'mac': eth.src
+ }
+
+ # set the timeout time.
+ timeout = igmp.QUERY_RESPONSE_INTERVAL
+ if query.maxresp:
+ timeout = query.maxresp / 10
+
+ self._to_hosts.setdefault(dpid, {})
+ if query.address == '0.0.0.0':
+ # general query. reset all reply status.
+ for group in self._to_hosts[dpid].values():
+ group['replied'] = False
+ group['leave'] = None
+ else:
+ # specific query. reset the reply status of the specific
+ # group.
+ group = self._to_hosts[dpid].get(query.address)
+ if group:
+ group['replied'] = False
+ group['leave'] = None
+
+ actions = [parser.OFPActionOutput(ofproto.OFPP_FLOOD)]
+ self._do_packet_out(
+ datapath, msg.data, in_port, actions)
+
+ # wait for REPORT messages.
+ hub.spawn(self._do_timeout_for_query, timeout, datapath)
+
+ def _do_report(self, report, in_port, msg):
+ """the process when the snooper received a REPORT message."""
+ datapath = msg.datapath
+ dpid = datapath.id
+ ofproto = datapath.ofproto
+ parser = datapath.ofproto_parser
+
+ if ofproto.OFP_VERSION == ofproto_v1_0.OFP_VERSION:
+ size = 65535
+ else:
+ size = ofproto.OFPCML_MAX
+
+ # check whether the querier port has been specified.
+ outport = None
+ value = self._to_querier.get(dpid)
+ if value:
+ outport = value['port']
+
+ # send a event when the multicast group address is new.
+ self._to_hosts.setdefault(dpid, {})
+ if not self._to_hosts[dpid].get(report.address):
+ self._send_event(
+ EventMulticastGroupStateChanged(
+ MG_GROUP_ADDED, report.address, outport, []))
+ self._to_hosts[dpid].setdefault(
+ report.address,
+ {'replied': False, 'leave': None, 'ports': {}})
+
+ # set a flow entry from a host to the controller when
+ # a host sent a REPORT message.
+ if not self._to_hosts[dpid][report.address]['ports'].get(
+ in_port):
+ self._to_hosts[dpid][report.address]['ports'][
+ in_port] = {'out': False, 'in': False}
+ self._set_flow_entry(
+ datapath,
+ [parser.OFPActionOutput(ofproto.OFPP_CONTROLLER, size)],
+ in_port, report.address)
+
+ if not self._to_hosts[dpid][report.address]['ports'][
+ in_port]['out']:
+ self._to_hosts[dpid][report.address]['ports'][
+ in_port]['out'] = True
+
+ if not outport:
+ self.logger.info("no querier exists.")
+ return
+
+ # set a flow entry from a multicast server to hosts.
+ if not self._to_hosts[dpid][report.address]['ports'][
+ in_port]['in']:
+ actions = []
+ ports = []
+ for port in self._to_hosts[dpid][report.address]['ports']:
+ actions.append(parser.OFPActionOutput(port))
+ ports.append(port)
+ self._send_event(
+ EventMulticastGroupStateChanged(
+ MG_MEMBER_CHANGED, report.address, outport, ports))
+ self._set_flow_entry(
+ datapath, actions, outport, report.address)
+ self._to_hosts[dpid][report.address]['ports'][
+ in_port]['in'] = True
+
+ # send a REPORT message to the querier if this message arrived
+ # first after a QUERY message was sent.
+ if not self._to_hosts[dpid][report.address]['replied']:
+ actions = [parser.OFPActionOutput(outport, size)]
+ self._do_packet_out(datapath, msg.data, in_port, actions)
+ self._to_hosts[dpid][report.address]['replied'] = True
+
+ def _do_leave(self, leave, in_port, msg):
+ """the process when the snooper received a LEAVE message."""
+ datapath = msg.datapath
+ dpid = datapath.id
+ ofproto = datapath.ofproto
+ parser = datapath.ofproto_parser
+
+ # check whether the querier port has been specified.
+ if not self._to_querier.get(dpid):
+ self.logger.info("no querier exists.")
+ return
+
+ # save this LEAVE message and reset the condition of the port
+ # that received this message.
+ self._to_hosts.setdefault(dpid, {})
+ self._to_hosts[dpid].setdefault(
+ leave.address,
+ {'replied': False, 'leave': None, 'ports': {}})
+ self._to_hosts[dpid][leave.address]['leave'] = msg
+ self._to_hosts[dpid][leave.address]['ports'][in_port] = {
+ 'out': False, 'in': False}
+
+ # create a specific query.
+ timeout = igmp.LAST_MEMBER_QUERY_INTERVAL
+ res_igmp = igmp.igmp(
+ msgtype=igmp.IGMP_TYPE_QUERY,
+ maxresp=timeout * 10,
+ csum=0,
+ address=leave.address)
+ res_ipv4 = ipv4.ipv4(
+ total_length=len(ipv4.ipv4()) + len(res_igmp),
+ proto=inet.IPPROTO_IGMP, ttl=1,
+ src=self._to_querier[dpid]['ip'],
+ dst=igmp.MULTICAST_IP_ALL_HOST)
+ res_ether = ethernet.ethernet(
+ dst=igmp.MULTICAST_MAC_ALL_HOST,
+ src=self._to_querier[dpid]['mac'],
+ ethertype=ether.ETH_TYPE_IP)
+ res_pkt = packet.Packet()
+ res_pkt.add_protocol(res_ether)
+ res_pkt.add_protocol(res_ipv4)
+ res_pkt.add_protocol(res_igmp)
+ res_pkt.serialize()
+
+ # send a specific query to the host that sent this message.
+ actions = [parser.OFPActionOutput(ofproto.OFPP_IN_PORT)]
+ self._do_packet_out(datapath, res_pkt.data, in_port, actions)
+
+ # wait for REPORT messages.
+ hub.spawn(self._do_timeout_for_leave, timeout, datapath,
+ leave.address, in_port)
+
+ def _do_flood(self, in_port, msg):
+ """the process when the snooper received a message of the
+ outside for processing. """
+ datapath = msg.datapath
+ ofproto = datapath.ofproto
+ parser = datapath.ofproto_parser
+
+ actions = [parser.OFPActionOutput(ofproto.OFPP_FLOOD)]
+ self._do_packet_out(datapath, msg.data, in_port, actions)
+
+ def _do_timeout_for_query(self, timeout, datapath):
+ """the process when the QUERY from the querier timeout expired."""
+ dpid = datapath.id
+
+ hub.sleep(timeout)
+ outport = self._to_querier[dpid]['port']
+
+ remove_dsts = []
+ for dst in self._to_hosts[dpid]:
+ if not self._to_hosts[dpid][dst]['replied']:
+ # if no REPORT message sent from any members of
+ # the group, remove flow entries about the group and
+ # send a LEAVE message if exists.
+ self._remove_multicast_group(datapath, outport, dst)
+ remove_dsts.append(dst)
+
+ for dst in remove_dsts:
+ del self._to_hosts[dpid][dst]
+
+ def _do_timeout_for_leave(self, timeout, datapath, dst, in_port):
+ """the process when the QUERY from the switch timeout expired."""
+ parser = datapath.ofproto_parser
+ dpid = datapath.id
+
+ hub.sleep(timeout)
+ outport = self._to_querier[dpid]['port']
+
+ if self._to_hosts[dpid][dst]['ports'][in_port]['out']:
+ return
+
+ del self._to_hosts[dpid][dst]['ports'][in_port]
+ self._del_flow_entry(datapath, in_port, dst)
+ actions = []
+ ports = []
+ for port in self._to_hosts[dpid][dst]['ports']:
+ actions.append(parser.OFPActionOutput(port))
+ ports.append(port)
+
+ if len(actions):
+ self._send_event(
+ EventMulticastGroupStateChanged(
+ MG_MEMBER_CHANGED, dst, outport, ports))
+ self._set_flow_entry(
+ datapath, actions, outport, dst)
+ self._to_hosts[dpid][dst]['leave'] = None
+ else:
+ self._remove_multicast_group(datapath, outport, dst)
+ del self._to_hosts[dpid][dst]
+
+ def _remove_multicast_group(self, datapath, outport, dst):
+ """remove flow entries about the group and send a LEAVE message
+ if exists."""
+ ofproto = datapath.ofproto
+ parser = datapath.ofproto_parser
+ dpid = datapath.id
+
+ self._send_event(
+ EventMulticastGroupStateChanged(
+ MG_GROUP_REMOVED, dst, outport, []))
+ self._del_flow_entry(datapath, outport, dst)
+ for port in self._to_hosts[dpid][dst]['ports']:
+ self._del_flow_entry(datapath, port, dst)
+ leave = self._to_hosts[dpid][dst]['leave']
+ if leave:
+ if ofproto.OFP_VERSION == ofproto_v1_0.OFP_VERSION:
+ in_port = leave.in_port
+ else:
+ in_port = leave.match['in_port']
+ actions = [parser.OFPActionOutput(outport)]
+ self._do_packet_out(
+ datapath, leave.data, in_port, actions)
+
+ # -------------------------------------------------------------------
+ # PRIVATE METHODS ( OTHERS )
+ # -------------------------------------------------------------------
+ def _set_logger(self):
+ """change log format."""
+ self.logger.propagate = False
+ hdl = logging.StreamHandler()
+ fmt_str = '[snoop][%(levelname)s] %(message)s'
+ hdl.setFormatter(logging.Formatter(fmt_str))
+ self.logger.addHandler(hdl)
projects / vsorcdistro / .git / blobdiff