ICode9

精准搜索请尝试: 精确搜索
首页 > 编程语言> 文章详细

Token Bucket在QoS中入门级介绍python示例

2021-04-13 18:03:33  阅读:25  来源: 互联网

标签:Current QoS 示例 python self Add tokens token time


Token Bucket为令牌桶算法,常被用于流量整形,或对客户端的限速。

假设家里是200M的宽带, 为了保证打游戏的流畅,我要把电视设置为10M。那么实现上可以如下:

 

方案一:

累加电视的所有接收/发出的数据包长度,如果超出10M,则丢弃报文;每秒更新一次累加值为0。

思路非常简单,也非常实用;从长时间统计上看,结果上偏差不大。

但可能限制出来流量图就会如下,不均匀,像抽风似的。如果设置了10条限速规则对10个设备进行限速时,则每秒开始的0.1秒,CPU压力很大,然后0.9秒很闲。

 

这种漏桶似的算法,利用率和平衡性较差。

方案二:

如果改进一下,将统计时间细化0.1秒,将效果会好不少。再进一步,将时间细化到每个报文接收,则效果就达到极致。

这就是Token Bucket做的事情。

 

流程出下:

1、当收到数据报文时,先试图获取token,如果剩余token足够则放行;

2、不够则更新token,等待token足够再继续。token按指定速率添加到bucket中,超过其容量则设置为最大值。

预期效果图如下:

 

示例代码实现如下:假设允许的速率每秒100,模拟包长在50~100随机,统计一下发送时间和平均速率。


 

# -- coding: utf-8 --
# Good luck.

import time
import random

class tb():
    def __init__(self, token_per_sec):
        self.captoken = token_per_sec
        self.curr_token = 0
        self.timestamp = time.time()

    def refill(self):
        if time.time() > self.timestamp:
            refill_token = self.captoken*(time.time() - self.timestamp)
            self.curr_token = min(self.curr_token + refill_token, self.captoken)
            print "\tRefill tokens Current: %d. (Add %d)" %(self.curr_token, refill_token)
            self.timestamp = time.time()
        else:
            print "\tNo refill...."

    def consume(self, num):
        ret = False
        if self.curr_token > 0: # Control precision
            self.curr_token -=  num
            print "\tGet token %d : remain %d" %(num, self.curr_token)
            ret = True
        self.refill()
        return ret


test = tb(100)
run_time = time.time()
all_len = 0;
times = 0

while True:
    t = 0.1
    ts = time.time()
    send_len = random.randint(10, 50)
    print "Need send %d" %(send_len)
    while not test.consume(send_len):
        time.sleep(t)
    te = time.time()
    print "Use Time:%s Send:%d " %(te-ts, send_len)
    all_len += send_len
    times += 1
    print "Avg = %s (times=%d)" %(all_len/(ts - run_time+0.1), times)
    time.sleep(t)

运行效果

C:\Python27\python.exe X:/GIT/wxy_code_backup/Flash_tools/tokenbucket.py

Need send 44

No refill....

Refill tokens Current: 9. (Add 9)

Get token 44 : remain -34

Refill tokens Current: -23. (Add 10)

Use Time:0.200999975204 Send:44

Avg = 440.0 (times=1)

Need send 27

Refill tokens Current: -13. (Add 9)

Refill tokens Current: -3. (Add 10)

Refill tokens Current: 6. (Add 10)

Get token 27 : remain -20

Refill tokens Current: -10. (Add 10)

Use Time:0.302000045776 Send:27

Avg = 177.057409665 (times=2)

Need send 33

Refill tokens Current: 0. (Add 10)

Refill tokens Current: 9. (Add 10)

Get token 33 : remain -23

Refill tokens Current: -13. (Add 10)

Use Time:0.200999975204 Send:33

Avg = 129.51431022 (times=3)

Need send 42

Refill tokens Current: -3. (Add 10)

Refill tokens Current: 6. (Add 10)

Get token 42 : remain -35

Refill tokens Current: -25. (Add 10)

Use Time:0.201000213623 Send:42

Avg = 132.126711657 (times=4)

Need send 16

Refill tokens Current: -15. (Add 9)

Refill tokens Current: -5. (Add 9)

Refill tokens Current: 4. (Add 10)

Get token 16 : remain -11

Refill tokens Current: -1. (Add 9)

Use Time:0.300999879837 Send:16

Avg = 115.22048411 (times=5)

......

Need send 21

Refill tokens Current: -4. (Add 10)

Refill tokens Current: 5. (Add 9)

Get token 21 : remain -15

Refill tokens Current: -5. (Add 10)

Use Time:0.200999975204 Send:21

Avg = 100.389763573 (times=138)

Need send 41

Refill tokens Current: 4. (Add 9)

Get token 41 : remain -36

Refill tokens Current: -26. (Add 10)

Use Time:0.101000070572 Send:41

Avg = 100.66413934 (times=139)

Need send 18

Refill tokens Current: -15. (Add 10)

Refill tokens Current: -5. (Add 10)

Refill tokens Current: 4. (Add 10)

Get token 18 : remain -13

Refill tokens Current: -3. (Add 10)

Use Time:0.301999807358 Send:18

Avg = 100.607594694 (times=140)

 

稳定后效果还算可以,从发包点上也相对均匀。当然初始仅为波动而已,可以通过调整优化掉,但从长时间看,效果是一致的。

 

方案三:

方案一、二都缺少缓冲的支持。如果报文来了,但是现在资源不够,不能发送,直接丢弃,则客户端再重发,这个过程太浪费。

因此增加缓冲队列,根据系统的内存,设定队列的大小,对于不能发送的报文,先入队。

如果队列满了呢? 那只能丢弃了,毫无办法。(不考虑共享宽带的情况)。

从Linux收发包角度,每个以太网设备有一个发送队列,如果满了,则设置为状态不可用,则不再收包。道理是一样的。

 

关于更新Token时间的设计,如上例程序,仅为演示,我没有加线程额外处理。

在存在缓冲队列的情况下,则必须增加更新线程,因为存在收发异常的情况。

 

示例如下,不过缓冲不过是增加一些弹性,丢包依然不可避免。结果可预知:由于缓冲的存在,平均流量会在一段时间后,才能趋向限制值。

# -- coding: utf-8 --
# Good luck.

import time
import random
import threading

class tb():
    def __init__(self, token_per_sec):
        self.captoken = token_per_sec
        self.curr_token = 0
        self.timestamp = time.time()
        self.queue = []
        self.q_max = 5

    def refill(self):
        if time.time() > self.timestamp:
            refill_token = self.captoken*(time.time() - self.timestamp)
            self.curr_token = min(self.curr_token + refill_token, self.captoken)
            print "\tRefill tokens Current: %d. (Add %d)" %(self.curr_token, refill_token)
            self.timestamp = time.time()
        else:
            print "\tNo refill...."

    def consume(self, num):
        ret = False
        if self.curr_token > 0: # Control precision
            self.curr_token -=  num
            print "\tGet token %d : remain %d" %(num, self.curr_token)
            ret = True
        elif len(self.queue) < self.q_max:
            # Enter queue
            self.queue.append(num)
            print "Enqueue Action."
            ret = True
        else:
            # Drop it.
            ret = False
        self.refill()
        return ret

    def dequeue(self):
        if len(self.queue) and self.curr_token > 0:
            tmp = self.queue.pop()
            self.curr_token -= tmp
            print "Dequeue Action : %d (remain %d)" %(tmp, self.curr_token)


def update_bucket(args):
    print "Args=%s" %(args.captoken)
    while True:
        time.sleep(0.1)
        args.refill()
        args.dequeue()

test = tb(100)
run_time = time.time()
all_len = 0
times = 0
tx_queue = []

# update bucket thread.
update_t = threading.Thread(target=update_bucket, args=(test,))
update_t.start()

while True:
    t = 0.2
    ts = time.time()
    send_len = random.randint(40, 50)
    print "Need send %d" %(send_len)

    if test.consume(send_len):
        te = time.time()
        print "Use Time:%s Send:%d " %(te-ts, send_len)
        all_len += send_len
    else:
        print "Drop Action."

    times += 1
    print "Avg = %s (times=%d)" %(all_len/(ts - run_time+0.1), times)
    time.sleep(t)

运行结果:

Args=100

Need send 46

Enqueue Action.

Refill tokens Current: 0. (Add 0)

Use Time:0.0 Send:46

Avg = 455.44587139 (times=1)

Refill tokens Current: 10. (Add 9)

Dequeue Action : 46 (remain -35)

Need send 41

Enqueue Action.

Refill tokens Current: -25. (Add 10)

Use Time:0.0 Send:41

Avg = 289.036568661 (times=2)

Refill tokens Current: -25. (Add 0)

Refill tokens Current: -15. (Add 10)

Need send 47

Enqueue Action.

Refill tokens Current: -5. (Add 9)

Use Time:0.0 Send:47

Avg = 266.932297286 (times=3)

Refill tokens Current: -5. (Add 0)

Refill tokens Current: 4. (Add 10)

Dequeue Action : 47 (remain -42)

 

......

Drop Action.

Avg = 106.217083485 (times=170)

Refill tokens Current: -9. (Add 2)

Refill tokens Current: 0. (Add 10)

Dequeue Action : 42 (remain -41)

Need send 50

Enqueue Action.

Refill tokens Current: -34. (Add 7)

Use Time:0.0 Send:50

Avg = 107.055107617 (times=171)

Refill tokens Current: -31. (Add 2)

 

在有针对多个用户的多个限速规则,必然存在多个队列后,就需要有一定的调度算法,常用的RR、DRR、WRR,后面使用一些示例,展开讨论一下。

 

 

 

 

 

 

标签:Current,QoS,示例,python,self,Add,tokens,token,time
来源: https://blog.51cto.com/u_7777817/2703856

专注分享技术,共同学习,共同进步。侵权联系[admin#icode9.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有