sdn

1. 网络拓扑生成python文件 demo.py

# 1. 导入相关的依赖
from mininet.net import Mininet
from mininet.node import Controller, RemoteController, OVSController
from mininet.node import CPULimitedHost, Host, Node
from mininet.node import OVSKernelSwitch, UserSwitch
from mininet.node import IVSSwitch
from mininet.cli import CLI
from mininet.log import setLogLevel, info
from mininet.link import TCLink, Intf
from subprocess import call

def myNetwork():
    # 定义网络，在IP基地址为10.0.0.0/8上
    net = Mininet(topo=None, build=False, ipBase='10.0.0.0/8')

    info('*** 添加远程控制器\n')
    # 因为我们在本地运行ryu-manager充当控制器，故这里为远程控制器
    # RemoteController且IP地址为本地IP
    c0 = net.addController(name='c0',
                           controller=RemoteController,
                           ip='127.0.0.1',
                           protocol='tcp',
                           port=6633)

    info('*** 添加交换机\n')
    # dpid为16位编号
    s1 = net.addSwitch('s1', cls=OVSKernelSwitch, dpid='0000000000000001')
    s2 = net.addSwitch('s2', cls=OVSKernelSwitch, dpid='0000000000000002')
    s3 = net.addSwitch('s3', cls=OVSKernelSwitch, dpid='0000000000000003')

    info('*** 添加3台主机\n')
    # 根据图片，只有 h1, h2, h3
    h1 = net.addHost('h1', cls=Host, ip='10.0.0.1', defaultRoute=None)
    h2 = net.addHost('h2', cls=Host, ip='10.0.0.2', defaultRoute=None)
    h3 = net.addHost('h3', cls=Host, ip='10.0.0.3', defaultRoute=None)

    info('*** 添加链路\n')
    # 主机与交换机的连接
    net.addLink(s1, h1)
    net.addLink(s2, h2)
    net.addLink(s3, h3)
    
    # 交换机级联（形成图片中的三角形环路）
    net.addLink(s1, s2)
    net.addLink(s2, s3)
    net.addLink(s3, s1)

    info('*** 启动网络\n')
    net.build()

    info('*** 启动控制器\n')
    for controller in net.controllers:
        controller.start()

    info('*** 启动交换机\n')
    net.get('s1').start([c0])
    net.get('s2').start([c0])
    net.get('s3').start([c0])

    info('*** 显示配置的交换机与主机信息\n')
    CLI(net)

    # 停止运行
    net.stop()

if __name__ == '__main__':
    setLogLevel('info')
    myNetwork()

2. 最短路径转发 shortest_ forwarding.py

# shortest_forwarding.py

from ryu.base import app_manager

from ryu.controller import ofp_event

from ryu.controller.handler import CONFIG_DISPATCHER, MAIN_DISPATCHER, DEAD_DISPATCHER

from ryu.controller.handler import set_ev_cls

from ryu.ofproto import ofproto_v1_3

from ryu.lib.packet import packet

from ryu.lib.packet import ethernet, arp, ether_types

from ryu.topology import event, switches

from ryu.topology.api import get_switch, get_link

import networkx as nx

  

class ShortestPathForwarding(app_manager.RyuApp):

OFP_VERSIONS = [ofproto_v1_3.OFP_VERSION]

  

def __init__(self, *args, **kwargs):

super(ShortestPathForwarding, self).__init__(*args, **kwargs)

# 初始化有向图

self.net = nx.DiGraph()

# 记录主机位置: mac -> (dpid, port)

self.mac_to_port = {}

# 记录 IP -> MAC 映射 (用于 ARP 代理)

self.arp_table = {}

  

@set_ev_cls(ofp_event.EventOFPSwitchFeatures, CONFIG_DISPATCHER)

def switch_features_handler(self, ev):

"""下发缺省流表，将无法匹配的包上报给控制器"""

datapath = ev.msg.datapath

ofproto = datapath.ofproto

parser = datapath.ofproto_parser

  

match = parser.OFPMatch()

actions = [parser.OFPActionOutput(ofproto.OFPP_CONTROLLER,

ofproto.OFPCML_NO_BUFFER)]

self.add_flow(datapath, 0, match, actions)

  

def add_flow(self, datapath, priority, match, actions, buffer_id=None):

"""下发流表项"""

ofproto = datapath.ofproto

parser = datapath.ofproto_parser

  

inst = [parser.OFPInstructionActions(ofproto.OFPIT_APPLY_ACTIONS,

actions)]

if buffer_id:

mod = parser.OFPFlowMod(datapath=datapath, buffer_id=buffer_id,

priority=priority, match=match,

instructions=inst)

else:

mod = parser.OFPFlowMod(datapath=datapath, priority=priority,

match=match, instructions=inst)

datapath.send_msg(mod)

  

# ------------------- 拓扑发现部分 -------------------

@set_ev_cls(event.EventSwitchEnter)

def get_topology_data(self, ev):

"""交换机上线时，添加到图中"""

switch_list = get_switch(self, None)

switches = [switch.dp.id for switch in switch_list]

self.net.add_nodes_from(switches)

# print("Current Switches:", switches)

  

@set_ev_cls(event.EventLinkAdd)

def get_link_data(self, ev):

"""链路发现时，添加到图中"""

links_list = get_link(self, None)

# 清除旧边，重新添加，确保图是最新的

# self.net.clear_edges() # 视情况可选择是否全清

for link in links_list:

self.net.add_edge(link.src.dpid, link.dst.dpid, port=link.src.port_no)

self.net.add_edge(link.dst.dpid, link.src.dpid, port=link.dst.port_no)

# print("Link Added. Current Edges:", self.net.edges())

  

# ------------------- 核心逻辑部分 -------------------

@set_ev_cls(ofp_event.EventOFPPacketIn, MAIN_DISPATCHER)

def _packet_in_handler(self, ev):

msg = ev.msg

datapath = msg.datapath

ofproto = datapath.ofproto

parser = datapath.ofproto_parser

in_port = msg.match['in_port']

dpid = datapath.id

  

pkt = packet.Packet(msg.data)

eth = pkt.get_protocols(ethernet.ethernet)[0]

# 过滤 LLDP 和 IPv6 组播 (避免干扰和之前的风暴)

if eth.ethertype == ether_types.ETH_TYPE_LLDP:

return

if eth.ethertype == ether_types.ETH_TYPE_IPV6:

return

  

dst = eth.dst

src = eth.src

  

# 1. 学习源 MAC 位置

self.mac_to_port.setdefault(dpid, {})

self.mac_to_port[dpid][src] = in_port

# 将主机视为图中的节点 (可选，为了计算路径方便)

if src not in self.net:

self.net.add_node(src)

self.net.add_edge(dpid, src, port=in_port)

self.net.add_edge(src, dpid) # 双向连接

  

# 2. 处理 ARP (实现 ARP 代理，核心防风暴逻辑)

if eth.ethertype == ether_types.ETH_TYPE_ARP:

self.handle_arp(datapath, in_port, pkt)

return

  

# 3. 处理 IP 数据包 (最短路径转发)

if dst in self.net:

# 如果目的 MAC 在图中 (之前学过)，计算最短路径

self.compute_shortest_path(msg, src, dst, dpid, in_port)

else:

# 如果完全未知，只能泛洪 (在有环路且没有STP时这步有风险，

# 但因为我们拦截了ARP，大部分未知单播不会造成持续风暴)

out_port = ofproto.OFPP_FLOOD

actions = [parser.OFPActionOutput(out_port)]

data = None

if msg.buffer_id == ofproto.OFP_NO_BUFFER:

data = msg.data

out = parser.OFPPacketOut(datapath=datapath, buffer_id=msg.buffer_id,

in_port=in_port, actions=actions, data=data)

datapath.send_msg(out)

  

def handle_arp(self, datapath, in_port, pkt):

"""ARP 代理逻辑：如果控制器知道 IP 对应的 MAC，直接回复，不泛洪"""

arp_pkt = pkt.get_protocol(arp.arp)

src_ip = arp_pkt.src_ip

src_mac = arp_pkt.src_mac

dst_ip = arp_pkt.dst_ip

  

# 记录 IP-MAC 映射

self.arp_table[src_ip] = src_mac

  

if arp_pkt.opcode == arp.ARP_REQUEST:

if dst_ip in self.arp_table:

# 命中缓存！控制器构造 ARP Reply 发回去

dst_mac = self.arp_table[dst_ip]

self.send_arp_reply(datapath, in_port, src_mac, src_ip, dst_mac, dst_ip)

else:

# 没命中，必须泛洪 (FLOOD)，让目标主机收到

# 注意：这里可能会有短暂的环路，但一旦收到 Reply，后续就不再泛洪

self.flood(datapath, in_port, pkt)

elif arp_pkt.opcode == arp.ARP_REPLY:

# ARP Reply 是单播，直接转发即可，或者让 shortest_path 处理

self.flood(datapath, in_port, pkt)

  

def send_arp_reply(self, datapath, port, src_mac, src_ip, dst_mac, dst_ip):

"""构造并发送 ARP Reply"""

ofproto = datapath.ofproto

parser = datapath.ofproto_parser

pkt = packet.Packet()

pkt.add_protocol(ethernet.ethernet(ethertype=ether_types.ETH_TYPE_ARP,

dst=src_mac, src=dst_mac))

pkt.add_protocol(arp.arp(opcode=arp.ARP_REPLY,

src_mac=dst_mac, src_ip=dst_ip,

dst_mac=src_mac, dst_ip=src_ip))

pkt.serialize()

actions = [parser.OFPActionOutput(port)]

out = parser.OFPPacketOut(datapath=datapath, buffer_id=ofproto.OFP_NO_BUFFER,

in_port=ofproto.OFPP_CONTROLLER,

actions=actions, data=pkt.data)

datapath.send_msg(out)

  

def flood(self, datapath, in_port, pkt):

ofproto = datapath.ofproto

parser = datapath.ofproto_parser

actions = [parser.OFPActionOutput(ofproto.OFPP_FLOOD)]

out = parser.OFPPacketOut(datapath=datapath, buffer_id=ofproto.OFP_NO_BUFFER,

in_port=in_port, actions=actions, data=pkt.data)

datapath.send_msg(out)

  

def compute_shortest_path(self, msg, src, dst, dpid, in_port):

"""使用 NetworkX 计算最短路径并下发流表"""

datapath = msg.datapath

ofproto = datapath.ofproto

parser = datapath.ofproto_parser

  

# 在图中计算 src 到 dst 的路径: [src_mac, switch1, switch2, ..., dst_mac]

try:

path = nx.shortest_path(self.net, src, dst)

# 当前交换机在路径中的位置

# path 结构示例: ['h1_mac', 1, 2, 3, 'h3_mac'] (假设 dpid 是整数)

# 找到当前 switch dpid 在 path 中的下一跳

if dpid not in path:

return

next_hop = path[path.index(dpid) + 1]

out_port = self.net[dpid][next_hop]['port']

# 下发流表：匹配 (src, dst) -> output: out_port

match = parser.OFPMatch(in_port=in_port, eth_dst=dst, eth_src=src)

actions = [parser.OFPActionOutput(out_port)]

self.add_flow(datapath, 1, match, actions, msg.buffer_id)

# 这里的 packet_out 已经在 add_flow 里被处理吗？不，packet_out 还是需要的

# 如果 buffer_id 是有效值，add_flow 会自动处理发包；如果不是，需要手动发

if msg.buffer_id == ofproto.OFP_NO_BUFFER:

out = parser.OFPPacketOut(datapath=datapath, buffer_id=msg.buffer_id,

in_port=in_port, actions=actions, data=msg.data)

datapath.send_msg(out)

except nx.NetworkXNoPath:

pass # 路径不可达

except Exception as e:

print(f"Path Error: {e}")

3. 动态负载平衡 dynamic_load_balance.py

  

# dynamic_load_balance.py

from ryu.base import app_manager

from ryu.controller import ofp_event

from ryu.controller.handler import CONFIG_DISPATCHER, MAIN_DISPATCHER, DEAD_DISPATCHER

from ryu.controller.handler import set_ev_cls

from ryu.ofproto import ofproto_v1_3

from ryu.lib import hub

from ryu.lib.packet import packet, ethernet, arp, ether_types

from ryu.topology import event

from ryu.topology.api import get_switch, get_link

import networkx as nx

  

class DynamicLoadBalance(app_manager.RyuApp):

OFP_VERSIONS = [ofproto_v1_3.OFP_VERSION]

  

def __init__(self, *args, **kwargs):

super(DynamicLoadBalance, self).__init__(*args, **kwargs)

self.net = nx.DiGraph()

self.mac_to_port = {}

self.arp_table = {}

self.datapaths = {}

# --- 监控相关变量 ---

self.monitor_thread = hub.spawn(self._monitor) # 开启监控线程

self.port_stats = {} # 保存上一秒的统计信息 {(dpid, port): (bytes, timestamp)}

self.port_speed = {} # 保存计算出的速率

self.THRESHOLD = 1024 * 1024 * 1 # 阈值：1 MB/s (为了测试方便设得较低)

self.DEFAULT_WEIGHT = 1 # 默认链路权重

self.CONGESTED_WEIGHT = 100 # 拥堵链路权重

  

@set_ev_cls(ofp_event.EventOFPStateChange, [MAIN_DISPATCHER, DEAD_DISPATCHER])

def _state_change_handler(self, ev):

datapath = ev.datapath

if ev.state == MAIN_DISPATCHER:

if datapath.id not in self.datapaths:

self.datapaths[datapath.id] = datapath

elif ev.state == DEAD_DISPATCHER:

if datapath.id in self.datapaths:

del self.datapaths[datapath.id]

  

@set_ev_cls(ofp_event.EventOFPSwitchFeatures, CONFIG_DISPATCHER)

def switch_features_handler(self, ev):

datapath = ev.msg.datapath

ofproto = datapath.ofproto

parser = datapath.ofproto_parser

match = parser.OFPMatch()

actions = [parser.OFPActionOutput(ofproto.OFPP_CONTROLLER,

ofproto.OFPCML_NO_BUFFER)]

self.add_flow(datapath, 0, match, actions)

  

def add_flow(self, datapath, priority, match, actions, buffer_id=None):

ofproto = datapath.ofproto

parser = datapath.ofproto_parser

inst = [parser.OFPInstructionActions(ofproto.OFPIT_APPLY_ACTIONS, actions)]

if buffer_id:

mod = parser.OFPFlowMod(datapath=datapath, buffer_id=buffer_id,

priority=priority, match=match, instructions=inst)

else:

mod = parser.OFPFlowMod(datapath=datapath, priority=priority,

match=match, instructions=inst)

datapath.send_msg(mod)

  

# ------------------- 监控逻辑 -------------------

def _monitor(self):

"""每隔 3 秒向所有交换机请求端口统计信息"""

while True:

for dp in self.datapaths.values():

self._request_stats(dp)

hub.sleep(3)

  

def _request_stats(self, datapath):

ofproto = datapath.ofproto

parser = datapath.ofproto_parser

req = parser.OFPPortStatsRequest(datapath, 0, ofproto.OFPP_ANY)

datapath.send_msg(req)

  

@set_ev_cls(ofp_event.EventOFPPortStatsReply, MAIN_DISPATCHER)

def _port_stats_reply_handler(self, ev):

body = ev.msg.body

dpid = ev.msg.datapath.id

for stat in body:

port_no = stat.port_no

# 忽略 LOCAL 端口

if port_no > 1000:

continue

key = (dpid, port_no)

tmp_bytes = stat.tx_bytes + stat.rx_bytes # 统计收发总流量

  

if key in self.port_stats:

# 计算速率：(当前字节 - 上次字节) / 时间间隔(这里约等于 sleep 时间)

# 简化处理，直接看增量

prev_bytes = self.port_stats[key]

speed = (tmp_bytes - prev_bytes) / 3.0 # byte/s

self.port_speed[key] = speed

# --- 动态调整权重 ---

self._adjust_link_weight(dpid, port_no, speed)

# 更新记录

self.port_stats[key] = tmp_bytes

  

def _adjust_link_weight(self, dpid, port_no, speed):

"""如果流量超过阈值，增加图中对应边的权重"""

# 找到该端口连接的邻居节点

# 注意：这里需要遍历图来找到 (u, v) 其中 u=dpid 且 port=port_no

for u, v, data in self.net.edges(data=True):

if u == dpid and data['port'] == port_no:

# 检查是否超过阈值

if speed > self.THRESHOLD:

if data['weight'] == self.DEFAULT_WEIGHT:

print(f"!!! 拥塞告警 [Switch {dpid} Port {port_no}] 速率: {speed/1024:.2f} KB/s. 切换线路!")

self.net[u][v]['weight'] = self.CONGESTED_WEIGHT

# 删除旧流表，强制重新寻路

self.del_flow(self.datapaths[dpid], port_no)

else:

# 流量恢复正常，恢复权重

if data['weight'] == self.CONGESTED_WEIGHT:

print(f"*** 拥塞解除 [Switch {dpid} Port {port_no}]. 恢复默认线路.")

self.net[u][v]['weight'] = self.DEFAULT_WEIGHT

# 可选：也删除流表让流量切回来（或者等流表自然超时）

self.del_flow(self.datapaths[dpid], port_no)

break

  

def del_flow(self, datapath, out_port):

"""删除指定输出端口的流表，迫使下一包触发 Packet-In 重新计算路径"""

ofproto = datapath.ofproto

parser = datapath.ofproto_parser

match = parser.OFPMatch() # 也可以更精确匹配，这里为了演示简单直接删掉该端口所有相关的流

# 注意：这里删除的是 action 输出到 out_port 的流，Ryu 标准库 match 不能直接匹配 out_port action

# 实际操作：通常我们删除所有流或者让流表由 idle_timeout 自动老化。

# 为了演示效果，我们发送一个删除指令，匹配所有包

mod = parser.OFPFlowMod(datapath=datapath, command=ofproto.OFPFC_DELETE,

out_port=out_port, out_group=ofproto.OFPG_ANY,

priority=1, match=match)

datapath.send_msg(mod)

  
  

# ------------------- 拓扑与转发逻辑 -------------------

@set_ev_cls(event.EventSwitchEnter)

def get_topology_data(self, ev):

switch_list = get_switch(self, None)

switches = [switch.dp.id for switch in switch_list]

self.net.add_nodes_from(switches)

@set_ev_cls(event.EventLinkAdd)

def get_link_data(self, ev):

links_list = get_link(self, None)

for link in links_list:

# 初始化权重为 1

self.net.add_edge(link.src.dpid, link.dst.dpid,

port=link.src.port_no, weight=self.DEFAULT_WEIGHT)

self.net.add_edge(link.dst.dpid, link.src.dpid,

port=link.dst.port_no, weight=self.DEFAULT_WEIGHT)

  

@set_ev_cls(ofp_event.EventOFPPacketIn, MAIN_DISPATCHER)

def _packet_in_handler(self, ev):

msg = ev.msg

datapath = msg.datapath

dpid = datapath.id

parser = datapath.ofproto_parser

in_port = msg.match['in_port']

pkt = packet.Packet(msg.data)

eth = pkt.get_protocols(ethernet.ethernet)[0]

  

if eth.ethertype == ether_types.ETH_TYPE_LLDP or eth.ethertype == ether_types.ETH_TYPE_IPV6:

return

  

dst = eth.dst

src = eth.src

# 拓扑学习：记录主机位置

if src not in self.net:

self.net.add_node(src)

self.net.add_edge(dpid, src, port=in_port, weight=0)

self.net.add_edge(src, dpid, weight=0)

  

# 处理 ARP

if eth.ethertype == ether_types.ETH_TYPE_ARP:

self.handle_arp(datapath, in_port, pkt)

return

  

# 处理 IP 转发 (核心：带权重的最短路径)

if dst in self.net:

try:

# weight='weight' 告诉算法考虑我们动态调整的拥塞权重

path = nx.shortest_path(self.net, src, dst, weight='weight')

if dpid not in path: return

next_hop = path[path.index(dpid) + 1]

out_port = self.net[dpid][next_hop]['port']

match = parser.OFPMatch(in_port=in_port, eth_dst=dst, eth_src=src)

actions = [parser.OFPActionOutput(out_port)]

# idle_timeout=5: 设短一点，方便快速适应网络变化

self.add_flow(datapath, 1, match, actions, msg.buffer_id)

if msg.buffer_id == datapath.ofproto.OFP_NO_BUFFER:

out = parser.OFPPacketOut(datapath=datapath, buffer_id=msg.buffer_id,

in_port=in_port, actions=actions, data=msg.data)

datapath.send_msg(out)

except:

pass

# (复用之前的 handle_arp 和 add_flow 辅助函数，此处省略未变部分以节省空间，请补全)

# 务必将之前的 handle_arp, send_arp_reply, flood 函数复制过来放到这里

def handle_arp(self, datapath, in_port, pkt):

"""ARP 代理逻辑：如果控制器知道 IP 对应的 MAC，直接回复，不泛洪"""

arp_pkt = pkt.get_protocol(arp.arp)

src_ip = arp_pkt.src_ip

src_mac = arp_pkt.src_mac

dst_ip = arp_pkt.dst_ip

  

# 记录 IP-MAC 映射

self.arp_table[src_ip] = src_mac

  

if arp_pkt.opcode == arp.ARP_REQUEST:

if dst_ip in self.arp_table:

# 命中缓存！控制器构造 ARP Reply 发回去

dst_mac = self.arp_table[dst_ip]

self.send_arp_reply(datapath, in_port, src_mac, src_ip, dst_mac, dst_ip)

else:

# 没命中，必须泛洪 (FLOOD)，让目标主机收到

# 注意：这里可能会有短暂的环路，但一旦收到 Reply，后续就不再泛洪

self.flood(datapath, in_port, pkt)

elif arp_pkt.opcode == arp.ARP_REPLY:

# ARP Reply 是单播，直接转发即可，或者让 shortest_path 处理

self.flood(datapath, in_port, pkt)

  

def send_arp_reply(self, datapath, port, src_mac, src_ip, dst_mac, dst_ip):

"""构造并发送 ARP Reply"""

ofproto = datapath.ofproto

parser = datapath.ofproto_parser

pkt = packet.Packet()

pkt.add_protocol(ethernet.ethernet(ethertype=ether_types.ETH_TYPE_ARP,

dst=src_mac, src=dst_mac))

pkt.add_protocol(arp.arp(opcode=arp.ARP_REPLY,

src_mac=dst_mac, src_ip=dst_ip,

dst_mac=src_mac, dst_ip=src_ip))

pkt.serialize()

actions = [parser.OFPActionOutput(port)]

out = parser.OFPPacketOut(datapath=datapath, buffer_id=ofproto.OFP_NO_BUFFER,

in_port=ofproto.OFPP_CONTROLLER,

actions=actions, data=pkt.data)

datapath.send_msg(out)

  

def flood(self, datapath, in_port, pkt):

ofproto = datapath.ofproto

parser = datapath.ofproto_parser

actions = [parser.OFPActionOutput(ofproto.OFPP_FLOOD)]

out = parser.OFPPacketOut(datapath=datapath, buffer_id=ofproto.OFP_NO_BUFFER,

in_port=in_port, actions=actions, data=pkt.data)

datapath.send_msg(out)