基于Simpy/Python的通信网络仿真工具（一）：数据包生成，转发和接收

前置知识

1. 通信网络中的流量测量基本指标参考：18张图带你了解衡量网络性能的四大指标：带宽、时延、抖动、丢包
2. Simpy是基于Python语言编写，实现离散时间仿真的库函数，关于Simpy的基础知识参考博客：python离散事件仿真库SimPy官方教程
3. 关于利用Simpy进行的通信网络仿真的基本原理可参考教程：Basic Network Simulations and Beyond
   in Python Introduction
4. 关于排队论知识可参考"wamg潇潇"的博客：排队论模型（一）：基本概念、输入过程与服务时间的常用概率分布

本系列博客内容主要是在“Basic Network Simulations and Beyond in Python Introduction”的基础上进行创作。
Simpy安装

pip install simpy

数据包生成和接收仿真

包生成器仿真拓扑如下
Pg: 包生成器；Ps: 包接收器

# 包生成器仿真
import simpy
import randomdef getPktSizes(env, host_num, min, max):""" 定义： 生成制定大小的数据包参数----env: 运行环境host_num: 到达的主机数量min: 数据包的最小值max: 数据包的最大值"""if env.now < 10:msg_list = []for __ in range(host_num):msg = random.randint(min, max)msg_list.append((msg))return msg_listclass Packet(object):""" 功能：定义数据包的相关信息参数-------time: 包发出时间arrive_time: 包到达时间size: 包大小id: 数据包idsrc: 源地址dst: 目的地址flow_id:  流id(用于大流量分包)"""def __init__(self, time, size, id, src="a", dst="z", flow_id=0):self.time = timeself.arrive_time = Noneself.size = sizeself.id = idself.src = srcself.dst = dstself.flow_id = flow_iddef __repr__(self):return "id: {}, src: {}, dst: {}, start time: {}, size: {}, arrive time: {}".\format(self.id, self.src, self.dst, self.time, self.size, self.arrive_time)class PacketGenerator(object):""" 定义: 包生成器，用于按规则发出数据包参数-------env : 仿真环境src : 源地址dst : 目的地址interval : 数据包的间隔发送时间initial_delay : 包生成器的间隔工作时间finish : 包生成器工作的终止时间"""def __init__(self, env, src, dst_list, interval, initial_delay=0, finish=float("inf"), flow_id=0):self.src = srcself.dst_list = dst_listself.env = envself.interval = intervalself.size_list = Noneself.initial_delay = initial_delayself.finish = finishself.out = Noneself.packets_sent = 0self.action = env.process(self.run())  # starts the run() method as a SimPy processself.flow_id = flow_iddef run(self):#仿真过程yield self.env.timeout(self.initial_delay)while self.env.now < self.finish:# 等待下一次开启传输yield self.env.timeout(self.interval()) self.size_list = getPktSizes(self.env, 1, 0, 200)if self.size_list is not None:for i in range(len(self.dst_list)):if self.size_list[i] > 0 : self.packets_sent += 1p = Packet(self.env.now, self.size_list[i], self.src + '-' + self.dst_list[i] + '-' +  str(self.packets_sent), src=self.src, dst = self.dst_list[i], flow_id=self.flow_id)self.out.put(p)class PacketSink(object):""" 定义: 包接收器，用于接收报文，并将延迟信息收集到等待列表中，并可以使用该列表查看流量的传输信息。参数-------env: 仿真环境debug : boolean 具体信息显示flagrec_arrivals : 记录到达时间absolute_arrivals : boolean 如果记录真实的绝对到达时间，否则记录连续到达之间的时间间隔。rec_waits : boolean 记录每个数据包的等待时间selector: 包选择器(规则匹配)"""def __init__(self, env, host, rec_arrivals=False, absolute_arrivals=False, rec_waits=True, debug=False, selector=None):self.store = simpy.Store(env)self.env = envself.hostName = host # 接收节点地址self.rec_waits = rec_waitsself.rec_arrivals = rec_arrivalsself.absolute_arrivals = absolute_arrivalsself.waits = []   # 记录等待时间self.arrivals = [] # 记录到达时间self.arrivals_id = [] # 记录到达的包名self.arrivals_src = [] # 记录到达的包源地址self.debug = debug  self.packets_rec = 0  # 记录接收包数量  self.bytes_rec = 0    # 记录接收字节数量self.selector = selector # 按条件选择接收包self.last_arrival = 0.0 # 最后到达的包def put(self, pkt):if not self.selector or self.selector(pkt):now = self.env.nowif self.rec_waits:self.waits.append(round(self.env.now - pkt.time, 3))self.arrivals_id.append(pkt.id)self.arrivals_src.append(pkt.src)if self.rec_arrivals:if self.absolute_arrivals:self.arrivals.append(now)else:self.arrivals.append(now - self.last_arrival)self.last_arrival = nowself.packets_rec += 1self.bytes_rec += pkt.sizepkt.arrive_time = nowif self.debug:print(self.hostName + " recv :", pkt)# host1的发包间隔
def constArrival1(): return 1.5# host2的发包间隔
def constArrival2():return 2.0# 定义包接收规则
def selector(pkt):return pkt.src in ["EE", "SS"] and pkt.dst == "MM"if __name__ == '__main__':until_time = 10env = simpy.Environment()  # 创建simpy仿真环境# 创建包生成器和包接收器ps = PacketSink(env=env, host="MM", debug=True, selector=selector)  # 开启信息打印pg1 = PacketGenerator(env=env, src="EE", dst_list = ["MM"],interval=constArrival1)pg2 = PacketGenerator(env=env, src="SS", dst_list = ["MM"], interval=constArrival2)# 连接包生成器和包接收器pg1.out = pspg2.out = ps# 环境运行，终止时间为10env.run(until=until_time)

仿真结果

MM recv : id: EE-MM-1, src: EE, dst: MM, start time: 1.5, size: 75, arrive time: 1.5
MM recv : id: SS-MM-1, src: SS, dst: MM, start time: 2.0, size: 84, arrive time: 2.0
MM recv : id: EE-MM-2, src: EE, dst: MM, start time: 3.0, size: 99, arrive time: 3.0
MM recv : id: SS-MM-2, src: SS, dst: MM, start time: 4.0, size: 7, arrive time: 4.0
MM recv : id: EE-MM-3, src: EE, dst: MM, start time: 4.5, size: 10, arrive time: 4.5
MM recv : id: SS-MM-3, src: SS, dst: MM, start time: 6.0, size: 94, arrive time: 6.0
MM recv : id: EE-MM-4, src: EE, dst: MM, start time: 6.0, size: 95, arrive time: 6.0
MM recv : id: EE-MM-5, src: EE, dst: MM, start time: 7.5, size: 103, arrive time: 7.5
MM recv : id: SS-MM-4, src: SS, dst: MM, start time: 8.0, size: 161, arrive time: 8.0
MM recv : id: EE-MM-6, src: EE, dst: MM, start time: 9.0, size: 183, arrive time: 9.0

由于没有设置包处理速度和传播速度，Ps接收包的时间和包发出时间一致。

单向端口转发仿真

拓扑连接关系图

# 单向Port仿真
import simpy
import functools
import randomclass SwitchPortIn(object):"""功能: 交换机包接收端口，通过加入一个出口选择器，实现根据源地址，目的地址转发数据参数----------env : 运行环境trans_list : 出口列表(list)valid_time : 流表有效时间(int)rec_arrivals : boolean  记录到达时间absolute_arrivals : boolean 如果为True，记录真实的绝对到达时间，否则记录连续到达之间的时间间隔。rec_waits : boolean  记录每个数据包的等待时间tables : 转发流表项"""def __init__(self, env, trans_list, valid_time, table, rec_arrivals=False, absolute_arrivals=False, rec_waits=True, debug=False):self.env = envself.out = [None for i in range(len(trans_list))]  # 出口列表self.trans_list = trans_listself.table = dict() if table is None else tableself.byte_size = 0  # 当前队列中的字节数self.debug = debugself.valid_time =  100 if valid_time == 0 or valid_time is None else valid_time # 等待时间为0, 或者为空，设置为100self.rec_waits = rec_waitsself.rec_arrivals = rec_arrivalsself.absolute_arrivals = absolute_arrivalsself.waits = []   # 记录等待时间self.sends_time = [] # 记录发送时间self.sends_id = [] # 记录发送的包名self.sends_src = [] # 记录发送的包源地址self.sends_dst = [] # 记录发送的包源地址self.packets_rec = 0  # 记录发送包数量  self.bytes_rec = 0    # 记录发送字节数量self.pkt_buffer = simpy.Store(env) # 包缓冲区self.transflow_buffer = simpy.Store(env) # 流表缓冲区self.action = env.process(self.waitTable()) # 执行转发def transPkt(self, pkt):pkt_key = pkt.src + '-' +pkt.dstnow = self.env.nowif self.rec_waits:self.waits.append(self.env.now - pkt.time)self.sends_id.append(pkt.id)self.sends_src.append(pkt.src)self.sends_dst.append(pkt.dst)if self.rec_arrivals:if self.absolute_arrivals:self.sends_time.append(now)else:self.sends_time.append(now - self.last_arrival)if pkt_key not in self.table:  # 流表项不存在，将包放入缓冲区return self.pkt_buffer.put(pkt)target_port = self.trans_list.index(self.table[pkt_key]) # 通过包的源地址，目的地址查询流表中对应的转发端口self.out[target_port].put(pkt) self.packets_rec += 1self.bytes_rec += pkt.sizeif self.debug:print("Port_in : ", pkt)""" 流结构 : "src-dst" : [port_id, gen_time] 目标端口，生成时间"""def put(self, pkt):pkt_key = pkt.src + '-' +pkt.dstif pkt_key in self.table:  # 流表项存在now = self.env.nowif now - self.table[pkt_key] >= self.valid_time: # 流超时print("流超时 : {}, time : {}".format(self.table[pkt_key], round(now, 3)))# 删除超时流self.table.pop(pkt_key)else:# 执行转发self.transPkt(pkt)else:self.pkt_buffer.put(pkt)def waitTable(self):while True:flow = yield self.transflow_buffer.get()print(flow)self.table[flow.pkt_key] = flowpkt = yield self.pkt_buffer.get()print(pkt)pkt_key = pkt.src + '-' +pkt.dstif pkt_key == flow.pkt_key:self.transPkt(pkt)else:self.pkt_buffer.put(pkt)def __repr__(self):return "switch: {}".format(self.switch.address)class SwitchPortOut(object):""" 功能: 交换机发送端口，将数据包从当前端口传输到目的地参数----------env : 运行环境rate : float 发送速率qlimit : integer (or None) 缓存队列大小limit_bytes :  是否丢包debug : 是否打印输出"""def __init__(self, env, address, rate, qlimit=None, limit_bytes=True, debug=False):self.store = simpy.Store(env) # 包缓冲区self.address = address # port地址self.rate = rate       # 发送速率self.env = env         # 运行环境self.out = None        # 目的连接方self.packets_rec = 0   # 包接收数量self.packets_drop = 0  # 包丢失数量self.qlimit = qlimit   # 队列大小self.limit_bytes = limit_bytes self.byte_size = 0     # 当前处理的数据量self.debug = debug     self.busy = 0          # 忙绿标志位 self.queue = []        # 记录队列self.out_drop = []     # 记录丢包self.action = env.process(self.run())  # starts the run() method as a SimPy processdef run(self):while True:pkt = (yield self.store.get()) self.busy = 1self.byte_size -= pkt.size # 这段时间能够处理的数据量yield self.env.timeout(pkt.size*8.0/self.rate) # 发送时延# yield self.env.timeout(distance/(2/3*c)) # 传播时延self.out.put(pkt)self.queue.remove(pkt)self.busy = 0if self.debug:print("Port_out : ", pkt)def put(self, pkt):self.packets_rec += 1tmp_byte_count = self.byte_size + pkt.size# 无限长队列（无丢包）处理if self.qlimit is None:self.byte_size = tmp_byte_countreturn self.store.put(pkt)# 丢包处理，情况1：包大小超限制，直接丢弃if self.limit_bytes and tmp_byte_count >= self.qlimit:self.packets_drop += 1self.out_drop.append(pkt)return#          情况2：队列满情况，直接丢弃elif not self.limit_bytes and len(self.store.items) >= self.qlimit-1:self.packets_drop += 1self.out_drop.append(pkt)#  正常转发else:self.byte_size = tmp_byte_countself.queue.append(pkt)return self.store.put(pkt)class Port(object):""" 功能: 接收和转发全功能端口，实现端口的接收和转发参数------env: 运行环境port_id: 端口的idtrans_list: 内部端口的连接列表, A_in --> B_out, C_outtable: 转发规则表, src-dst : port_outvalid_time: 流表有效时间send_rate: out端口的传输速率queue_limit: 队列最大长度is_limit: 是否丢包debug: 输出执行信息"""def __init__(self, env, port_id, trans_list, table=None, valid_time=None,send_rate=0, queue_limit=None, is_limit=True, debug=False) -> None:self.env = envself.switch_id = Noneself.port_id = port_id self.queue = []   # 端口队列记录self.drop = []    # 端口丢包记录self.in_port = SwitchPortIn(env=env, trans_list=trans_list, valid_time=valid_time,table=table, debug=debug)self.out_port = SwitchPortOut(env=env, address=None, rate=send_rate, qlimit=queue_limit,limit_bytes=is_limit, debug=debug)# 设置转发表（判断包应该转发到哪个port的out）def setTable(self, table):self.in_port.table = table# 设置out端口的连接对象def setOutObj(self, out):self.out_port.out = out# 设置in端口的连接对象def setTransPort(self, target_list):for i in range(len(target_list)):self.in_port.out[i] = target_list[i].out_port# 获取out端口的队列def getQueue(self):return self.out_port.queue# 输出out端口的丢包情况def getOutDrop(self):return self.out_port.out_dropdef __repr__(self):return "switch_address :{}, port_id :{}".format(self.switch_id, self.port_id)def constArrival():return 1.0   # time interval
def selector(pkt):return pkt.src in ["EE"] and pkt.dst == "MM"
def getTransList1(): # 2 代表 目标Port的编号 从1开始return [2]
def getTransList2():return [1]
def getTable1():return {"EE-MM": 2}
def getTable2():return {"MM-EE": 1}
if __name__=='__main__':samp_dist = functools.partial(random.expovariate, 1.0)env = simpy.Environment()  # Create the SimPy environmentps = PacketSink(env=env, host="MM", debug=True, rec_arrivals=True,absolute_arrivals=True, selector=selector)  # debugging enable for simple outputpg = PacketGenerator(env=env, src="EE", dst_list=["MM"],interval=constArrival)  # 不加括号表示，调用函数体，加括号表示调用结果pt1 = Port(env=env, port_id = 1, trans_list=getTransList1(), table=getTable1(),send_rate=400, queue_limit=301, debug=False)pt2 = Port(env=env, port_id = 2, trans_list=getTransList2(), table=getTable2(),send_rate=400, queue_limit=301, debug=False)  # rate 是波特率pg.out = pt1.in_port # pg.out --> pt1.inpt1.setTransPort(target_list = [pt2]) # pt1.in --> pt2.outpt2.setOutObj(ps)   # pt2.out --> ps.inenv.run(until=15)print("waits: {}".format(ps.waits))print("sink arrival times: {}".format(ps.arrivals))print("received: {}, dropped {}, sent {}".format(ps.packets_rec, pt2.out_port.packets_drop, pg.packets_sent))print("received ids: {}".format(ps.arrivals_id))print("queue: {}".format([pkt.id for pkt in pt2.out_port.queue]))print("drop : {}".format([pkt.id for pkt in pt2.out_port.out_drop])) 

运行结果：

MM recv : id: EE-MM-1, src: EE, dst: MM, start time: 1.0, size: 5, arrive time: 1.1
MM recv : id: EE-MM-2, src: EE, dst: MM, start time: 2.0, size: 140, arrive time: 4.8
MM recv : id: EE-MM-3, src: EE, dst: MM, start time: 3.0, size: 143, arrive time: 7.66
MM recv : id: EE-MM-5, src: EE, dst: MM, start time: 5.0, size: 137, arrive time: 10.4
MM recv : id: EE-MM-6, src: EE, dst: MM, start time: 6.0, size: 70, arrive time: 11.8
MM recv : id: EE-MM-7, src: EE, dst: MM, start time: 7.0, size: 7, arrive time: 11.94
MM recv : id: EE-MM-8, src: EE, dst: MM, start time: 8.0, size: 149, arrive time: 14.92
waits: [0.1, 2.8, 4.66, 5.4, 5.8, 4.94, 6.92]
sink arrival times: [1.1, 4.8, 7.66, 10.4, 11.8, 11.94, 14.92]
received: 7, dropped 1, sent 9
received ids: ['EE-MM-1', 'EE-MM-2', 'EE-MM-3', 'EE-MM-5', 'EE-MM-6', 'EE-MM-7', 'EE-MM-8']
queue: ['EE-MM-9']
drop : ['EE-MM-4']

在数据包生成与接收的基础上，增加端口的转发和拓扑连接，实现数据包的成功转发和丢包，排队等。
欢迎通过企鹅(488128665)进行交流！
以上仿真内容代码下载：基于Simpy/Python的通信网络仿真工具