Python网络质量测试工具增加乱序统计

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了Python网络质量测试工具增加乱序统计，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含3816字，纯文字阅读大概需要6分钟。

内容图文

半月月前，我用Python写了一个工具，可以测试网络的纯丢包率以及探测网络路径中的队列情况，经过一些使用者的反馈，还算比较好用，关于这个工具，请参见《动手写一个探测网络质量(丢包率/RTT/队形等)的工具》。
??????? 但是我觉得这个少了关于乱序度的测试功能，于是补充之。其实，在Linux的TC工具上，除了队列，丢包率，延迟之外，乱序度也是一个非常重要的配置参数，不过请记住，Linux不是全部，对于程序员而言，除了抓包之外，了解一点Linux之外的东西，比如Cisco，运营商之类的，还是必要的。我至今依然记得网吧的网管搞定了一个东软高级专家搞不定的问题的场景，这很正常，因为虽然网吧的网管可能没什么基础知识，程序员同样也没有，网管的优势在于，人家善于使用工具。两个领域，方法论自然不同。
??????? 事实上关于乱序的测试统计非常简单，这基于一个预期，就是“应答序列号一定大于等于发送序列号”，一旦不满足这个预期，说明有数据包乱序，在这个程序中，我主要统计两个变量，第一是总的乱序度数量，另一个是总的乱序次数。
统计逻辑如下：
变量：1).当前应答的最高序列号hseq；2).当前应答的序列号seq；3).乱序度reorder；4).乱序次数reorder_cnt。
1>.如果seq大于等于hseq，则更新hseq为seq；
2>.否则计算delta=hseq-seq，redorder+=delta，reorder+=1。
这样在程序的最后会统计出整个网络的乱序情况。详细情况请看代码：

#!/usr/local/bin/python

import sys
import time
from time import sleep,ctime

import signal
import threading
from scapy.all import *

target = sys.argv[1]
tot = int(sys.argv[2])
tot_per = int(sys.argv[3])
vl = int(sys.argv[4])
flt = "host " + target + " and icmp"

handle = open("/dev/null", 'w')

out_list = []
in_list = []

global pos
global curr
global tot_reorder
global reorder

pos = 0
curr = 0
tot_reorder = 0
reorder = 0

def output():
	global pos
	global curr
	global tot_reorder
	global reorder
	all = out_list + in_list
	all.sort(lambda x,y:cmp(x[3],y[3]))
	for item in all:
		print item[0], item[1], item[2], item[3]*10
	sys.stdout.flush()
    	handle.write("\nReorder:" + str(reorder) + " Reorder cnt:" + str(tot_reorder) + "\n")
        os._exit(0)

def signal_handler(signal, frame):
    	handle.write("\nExit:" + ctime() + '\n')
	output()

class ThreadWraper(threading.Thread):
	def __init__(self,func,args,name=''):
		threading.Thread.__init__(self)
		self.name=name
		self.func=func
		self.args=args

	def run(self):
		apply(self.func,self.args)

def printrecv(pktdata):
	global pos
	global curr
	global tot_reorder
	global reorder
	if ICMP in pktdata and pktdata[ICMP]:
		seq = str(pktdata[ICMP].seq)
		if seq == tot_per + 2:
			return
		if str(pktdata[IP].dst) == target:
    			handle.write('*')
    			handle.flush()
			out_list.append(('+', 1, seq, time.clock()))
		else:
			if vl == 2:
    				handle.write('.')
			else:
    				handle.write('\b \b')
    			handle.flush()
			in_list.append(('-', 0, seq, time.clock()))
			
			curr = int(seq)
			if curr >= pos:
				pos = curr
			else:
				delta = pos - curr
				tot_reorder += 1
				reorder += delta
				

def checkstop(pktdata):
	if ICMP in pktdata and pktdata[ICMP]:
		seq = str(pktdata[ICMP].seq)
		if int(seq) == tot_per + 2 and str(pktdata[IP].src) == target:
    			handle.write("\nExit:" + ctime() + '\n')
			output()
			return True
	return False
	
def send_packet():
	times = 0
	global pos
	global curr
	while times < tot:
		times += 1
    		send(IP(dst = target)/ICMP(seq = (0, tot_per))/"test", verbose = 0, loop = 1, count = 1)
		pos = 0
		curr = 0
		#out_list.append(('++++++++', 1, -1, str(time.clock())))
    	send(IP(dst = target)/ICMP(seq = tot_per+2)/"bye", verbose = 0)


def recv_packet():
	sniff(prn = printrecv, store = 1, filter = flt, stop_filter = checkstop)

def startup():
    	handle.write("Start:" + ctime() + '\n')

	send_thread = ThreadWraper(send_packet,(),send_packet.__name__)
	send_thread.setDaemon(True)  
	send_thread.start()

	recv_thread = ThreadWraper(recv_packet,(),recv_packet.__name__)
	recv_thread.setDaemon(True)  
	recv_thread.start()

	signal.pause()

if __name__ == '__main__':
	if vl != 0:
		handle.close()
		handle = sys.stderr
	signal.signal(signal.SIGINT, signal_handler)
	startup()

相关的更新已经更新到了github

再分享一下我老师大神的人工智能教程吧。零基础！通俗易懂！风趣幽默！还带黄段子！希望你也加入到我们人工智能的队伍中来！https://blog.csdn.net/jiangjunshow

内容总结

以上是互联网集市为您收集整理的Python网络质量测试工具增加乱序统计全部内容，希望文章能够帮你解决Python网络质量测试工具增加乱序统计所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

---