0%

计算机网络服务仿真简单示例

前言

想做一些关于计算机网络服务的仿真,因为不是特别了解其中的机理,所以想找一些参考资料,其中一篇文章Basic Network Simulations and Beyond in Python
刚好比较契合我的需要,所以转载过来,也与大家分享。

离散事件系统仿真计算机网络

计算机网络中的网络服务可以看作一个离散事件系统,例如请求的到达、服务的响应都是典型的离散事件,因此我们可以采用离散事件系统仿真进行模拟。

在计算机网络仿真钟,整个仿真系统需要执行大量的任务,例如数据包的生成、传输、排队等等。在现代计算机中,由于操作系统的进程、线程等机制,让这些任务有种同时执行的错觉

为了简化上述机制在仿真的实现,提高易用性和可扩展性,我们在Python中使用生成器来完成类似的功能。在这里,我们会使用曾经介绍过Python中用于离散仿真的库SimPy

仿真实体

在前面的文章中,我们曾经介绍过,仿真中需要用到各种临时实体和永久实体,在这个例子中,我们采用面向对象变成的思想,使用类来表示各种实体。

Packet类

Packet类是代表数据包的简单数据对象,由PacketGenerators类创建,关键字段包括:

  • generation_time 生成时间
  • size 数据包大小
  • flow_id 请求类型
  • packet_id 数据包ID
  • source 源
  • destination 目标

眼下,我们先不对更上层的协议建模,即数据包没有携带有效载荷(payload)。size参数用于确定传输的时间,flow_id用于确定处理请求占用的时间和资源。Packet类的主要代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class Packet(object):
""" A very simple class that represents a packet.
This packet will run through a queue at a switch output port.
We use a float to represent the size of the packet in bytes so that
we can compare to ideal M/M/1 queues.

Parameters
----------
time : float
the time the packet arrives at the output queue.
size : float
the size of the packet in bytes
id : int
an identifier for the packet
src, dst : int
identifiers for source and destination
flow_id : int
small integer that can be used to identify a flow
"""
def __init__(self, time, size, id, src="a", dst="z", flow_id=0):
self.time = time
self.size = size
self.id = id
self.src = src
self.dst = dst
self.flow_id = flow_id

def __repr__(self):
return "id: {}, src: {}, time: {}, size: {}".\
format(self.id, self.src, self.time, self.size)

PacketGenerator类

顾名思义,PacketGenerator类用来产生和发送具有指定到达时间间隔和大小分布的数据包,在生成时可以设置初始延迟和完成时间。此外,根据Packet类的字段可知,PacketGenerator类还需要设置数据包的sourceflow_id等信息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class PacketGenerator(object):
""" Generates packets with given inter-arrival time distribution.
Set the "out" member variable to the entity to receive the packet.

Parameters
----------
env : simpy.Environment
the simulation environment
adist : function
a no parameter function that returns the successive inter-arrival times of the packets
sdist : function
a no parameter function that returns the successive sizes of the packets
initial_delay : number
Starts generation after an initial delay. Default = 0
finish : number
Stops generation at the finish time. Default is infinite


"""
def __init__(self, env, id, adist, sdist, initial_delay=0, finish=float("inf"), flow_id=0):
self.id = id
self.env = env
self.adist = adist
self.sdist = sdist
self.initial_delay = initial_delay
self.finish = finish
self.out = None
self.packets_sent = 0
self.action = env.process(self.run()) # starts the run() method as a SimPy process
self.flow_id = flow_id

def run(self):
"""The generator function used in simulations.
"""
yield self.env.timeout(self.initial_delay)
while self.env.now < self.finish:
# wait for next transmission
yield self.env.timeout(self.adist())
self.packets_sent += 1
p = Packet(self.env.now, self.sdist(), self.packets_sent, src=self.id, flow_id=self.flow_id)
self.out.put(p)

PacketSink类

PacketSink类可以认为是数据包流转的终点,用来记录数据包到达的相关信息,可以是绝对的到达时间,也可以是到达间隔,还可以记录等待时间。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class PacketSink(object):
""" Receives packets and collects delay information into the
waits list. You can then use this list to look at delay statistics.

Parameters
----------
env : simpy.Environment
the simulation environment
debug : boolean
if true then the contents of each packet will be printed as it is received.
rec_arrivals : boolean
if true then arrivals will be recorded
absolute_arrivals : boolean
if true absolute arrival times will be recorded, otherwise the time between consecutive arrivals
is recorded.
rec_waits : boolean
if true waiting time experienced by each packet is recorded
selector: a function that takes a packet and returns a boolean
used for selective statistics. Default none.

"""
def __init__(self, env, rec_arrivals=False, absolute_arrivals=False, rec_waits=True, debug=False, selector=None):
self.store = simpy.Store(env)
self.env = env
self.rec_waits = rec_waits
self.rec_arrivals = rec_arrivals
self.absolute_arrivals = absolute_arrivals
self.waits = []
self.arrivals = []
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.now
if self.rec_waits:
self.waits.append(self.env.now - pkt.time)
if self.rec_arrivals:
if self.absolute_arrivals:
self.arrivals.append(now)
else:
self.arrivals.append(now - self.last_arrival)
self.last_arrival = now
self.packets_rec += 1
self.bytes_rec += pkt.size
if self.debug:
print(pkt)

SwitchPort类

模拟交换机或路由器的端口,采用先入先出(FIFO)的工作模式,可以设置输出端口的速率和队列容量限制(以字节为单位),并且可以跟踪接收的和丢弃的数据包。此外,我个人认为SwitchPort类也可以作为服务器的端口,设置更加复杂的功能和特性。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class SwitchPort(object):
""" Models a switch output port with a given rate and buffer size limit in bytes.
Set the "out" member variable to the entity to receive the packet.

Parameters
----------
env : simpy.Environment
the simulation environment
rate : float
the bit rate of the port
qlimit : integer (or None)
a buffer size limit in bytes or packets for the queue (including items
in service).
limit_bytes : If true, the queue limit will be based on bytes if false the
queue limit will be based on packets.

"""
def __init__(self, env, rate, qlimit=None, limit_bytes=True, debug=False):
self.store = simpy.Store(env)
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 # Current size of the queue in bytes
self.debug = debug
self.busy = 0 # Used to track if a packet is currently being sent
self.action = env.process(self.run()) # starts the run() method as a SimPy process

def run(self):
while True:
msg = (yield self.store.get())
self.busy = 1
self.byte_size -= msg.size
yield self.env.timeout(msg.size*8.0/self.rate)
self.out.put(msg)
self.busy = 0
if self.debug:
print(msg)

def put(self, pkt):
self.packets_rec += 1
tmp_byte_count = self.byte_size + pkt.size

if self.qlimit is None:
self.byte_size = tmp_byte_count
return self.store.put(pkt)
if self.limit_bytes and tmp_byte_count >= self.qlimit:
self.packets_drop += 1
return
elif not self.limit_bytes and len(self.store.items) >= self.qlimit-1:
self.packets_drop += 1
else:
self.byte_size = tmp_byte_count
return self.store.put(pkt)

PortMonitor类

用来监控SwitchPort类的队列长度。受PortMonitor类的启发,可以构造功能更丰富的Monitor类,用来输出仿真过程中系统内部的状态信息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class PortMonitor(object):
""" A monitor for an SwitchPort. Looks at the number of items in the SwitchPort
in service + in the queue and records that info in the sizes[] list. The
monitor looks at the port at time intervals given by the distribution dist.

Parameters
----------
env : simpy.Environment
the simulation environment
port : SwitchPort
the switch port object to be monitored.
dist : function
a no parameter function that returns the successive inter-arrival times of the
packets
"""
def __init__(self, env, port, dist, count_bytes=False):
self.port = port
self.env = env
self.dist = dist
self.count_bytes = count_bytes
self.sizes = []
self.action = env.process(self.run())

def run(self):
while True:
yield self.env.timeout(self.dist())
if self.count_bytes:
total = self.port.byte_size
else:
total = len(self.port.store.items) + self.port.busy
self.sizes.append(total)

仿真案例

例1: 两个数据生成器和一个接收器

在这个例子中,我们设计固定的时间间隔分别为1.5h和2的两个数据包生成器,数据包的大小服从指数分布。我们直接使用PacketSink作为接收对象,意味着数据产生以后立刻就会被记录下到达时间等信息(没有考虑任何传输和处理的耗时)。我们将仿真的时长设定为20,具体代码和结果如下。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from random import expovariate
import simpy
from SimComponents import PacketGenerator, PacketSink, SwitchPort

def constArrival(): # Constant arrival distribution for generator 1
return 1.5

def constArrival2():
return 2.0

def distSize():
return expovariate(0.01)

env = simpy.Environment() # Create the SimPy environment
# Create the packet generators and sink
ps = PacketSink(env, debug=True) # debugging enable for simple output
pg = PacketGenerator(env, "Generator 1", constArrival, distSize)
pg2 = PacketGenerator(env, "Generator 2", constArrival2, distSize)
# Wire packet generators and sink together
pg.out = ps
pg2.out = ps
env.run(until=10)
id: 1, src: Generator 1, time: 1.5, size: 235.32925861364745
id: 1, src: Generator 2, time: 2.0, size: 65.96294829880948
id: 2, src: Generator 1, time: 3.0, size: 136.2031141150086
id: 2, src: Generator 2, time: 4.0, size: 31.698004682491476
id: 3, src: Generator 1, time: 4.5, size: 112.40261662960079
id: 3, src: Generator 2, time: 6.0, size: 150.7985723713098
id: 4, src: Generator 1, time: 6.0, size: 97.96121191425672
id: 5, src: Generator 1, time: 7.5, size: 38.67744218881105
id: 4, src: Generator 2, time: 8.0, size: 4.688240030858553
id: 6, src: Generator 1, time: 9.0, size: 4.535071796124653

例2:转发端口超负荷

这个例子是对转发端口性能的仿真,对于每一个SwitchPort类,rate参数代表该端口的数据转发能力,qlimt参数代表该端口的排队容量限制,在仿真中设定参数如下:

  • 数据包大小为100bytes;
  • 数据到达间隔为1.5;
  • 端口的处理能力为200bps,也就是200/8=25Bps;
  • 队列长度限制为为300bytes;
  • 仿真时长为20。
    可以看到,在整个仿真过程中一共产生了13个数据包,其中有4个被转发给负责最终接收的PacketSink对象,有6个数据包被丢弃。那还有3个呢?另外的3个还在转发端口的队列中排队。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
def constSize():
return 100.0 # bytes

env = simpy.Environment() # Create the SimPy environment
ps = PacketSink(env, debug=True) # debug: every packet arrival is printed
pg = PacketGenerator(env, "SJSU", constArrival, constSize)
switch_port = SwitchPort(env, rate=200.0, qlimit=300)
# Wire packet generators and sinks together
pg.out = switch_port
switch_port.out = ps
env.run(until=20)
print("waits: {}".format(ps.waits))
print("received: {}, dropped {}, sent {}".format(ps.packets_rec,
switch_port.packets_drop, pg.packets_sent))
id: 1, src: SJSU, time: 1.5, size: 100.0
id: 2, src: SJSU, time: 3.0, size: 100.0
id: 3, src: SJSU, time: 4.5, size: 100.0
id: 4, src: SJSU, time: 6.0, size: 100.0
waits: [4.0, 6.5, 9.0, 11.5]
received: 4, dropped 6, sent 13

例3: M/M/1排队系统

M/M/1排队系统,也叫单服务台是排队论中最典型、应用领域最广的模型,M/M/1的具体条件如下:

  • 请求到达为泊松过程;
  • 服务时间附送指数分布;
  • 单服务台,FIFO;
  • 队列长度无限;
  • 请求数目无限。

一般情况下,大家更加关注服务达到稳态时的一些指标例如:

MM1_formula

在这里我们假设:

  • 单位时间的请求到达速率为λ=0.5\lambda=0.5;
  • 请求数据包的大小服从参数为0.01的指数分布,也就是其均值为100bytes;
  • 服务台的处理能力为1000 bps,也就是125Bps,也就是单位时间平均处理μ=1.25\mu=1.25个数据包。

容易计算得到以下理论值:

  • 缓冲效用为ρ=λμ=0.4\rho=\frac{\lambda}{\mu}=0.4
  • 平均等待时间为ρμλ=1.333\frac{\rho}{\mu-\lambda}=1.333
  • 平均队列长度(包含正在服务对象)为ρ1ρ=0.667\frac{\rho}{1-\rho}=0.667

可以看到,仿真结果与理论值较为接近。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import random
import functools
import simpy
import matplotlib.pyplot as plt
from SimComponents import PacketGenerator, PacketSink, SwitchPort, PortMonitor

adist = functools.partial(random.expovariate, 0.5)
sdist = functools.partial(random.expovariate, 0.01) # mean size 100 bytes
samp_dist = functools.partial(random.expovariate, 1.0)
port_rate = 1000.0

env = simpy.Environment() # Create the SimPy environment
# Create the packet generators and sink
ps = PacketSink(env, debug=False, rec_arrivals=True)
pg = PacketGenerator(env, "Greg", adist, sdist)
switch_port = SwitchPort(env, port_rate, qlimit=10000)
# Using a PortMonitor to track queue sizes over time
pm = PortMonitor(env, switch_port, samp_dist)
# Wire packet generators, switch ports, and sinks together
pg.out = switch_port
switch_port.out = ps
# Run it
env.run(until=100000)
print("Last 10 waits: " + ", ".join(["{:.3f}".format(x) for x in ps.waits[-10:]]))
print("Last 10 queue sizes: {}".format(pm.sizes[-10:]))
print("Last 10 sink arrival times: " + ", ".join(["{:.3f}".format(x) for x in ps.arrivals[-10:]]))
print("average wait = {:.3f}".format(sum(ps.waits)/len(ps.waits)))
print("received: {}, dropped {}, sent {}".format(switch_port.packets_rec, switch_port.packets_drop, pg.packets_sent))
print("loss rate: {}".format(float(switch_port.packets_drop)/switch_port.packets_rec))
print("average queue size: {:.3f}".format(float(sum(pm.sizes))/len(pm.sizes)))
Last 10 waits: 2.757, 2.366, 2.094, 1.655, 2.725, 0.679, 2.731, 3.078, 0.820, 0.177
Last 10 queue sizes: [1, 1, 0, 1, 1, 2, 2, 2, 2, 2]
Last 10 sink arrival times: 2.352, 0.682, 0.021, 0.727, 3.607, 4.632, 2.316, 0.799, 0.116, 0.175
average wait = 1.334
received: 50240, dropped 0, sent 50240
loss rate: 0.0
average queue size: 0.668
1
2
3
4
5
6
7
fig, axis = plt.subplots(figsize=(16,10))
axis.hist(ps.arrivals, bins=100, density=True)
axis.set_title("Histogram for Sink Interarrival times")
axis.set_xlabel("time")
axis.set_ylabel("normalized frequency of occurrence")
# fig.savefig("ArrivalHistogram.png")
plt.show()

output_26_0.png

除了这些例子以外,还可以模拟排队网络、加权公平排队等更加复杂的情景,此处不再一一介绍。

小结

文章介绍了模拟计算机网络服务时可能用到的几种实体,并且在几种比较基本或者典型的场景下进行了仿真。后续的话,将在次基础上继续丰富网络仿真的功能。

参考资料

  1. https://www.grotto-networking.com/DiscreteEventPython.html
张da统帅 wechat
扫码订阅我的微信公众号
走过路过,切莫错过!