树莓派CAN(FD) 测试

双向压力测试

#!/usr/bin/env python3# -*- coding: utf-8 -*-importosimporttimeimportsocketimportstructimportselectimportthreadingimportsubprocessimportargparse# ===== CAN 常量 =====SOL_CAN_RAW=getattr(socket,"SOL_CAN_RAW",101)CAN_RAW_FILTER=getattr(socket,"CAN_RAW_FILTER",1)CAN_RAW_LOOPBACK=getattr(socket,"CAN_RAW_LOOPBACK",3)CAN_RAW_RECV_OWN_MSGS=getattr(socket,"CAN_RAW_RECV_OWN_MSGS",4)CAN_ECHO_FLAG=0x20000000STD_MASK=0x7FFCAN_ID=0x123PAYLOAD=bytes([0x11,0x22,0x33,0x44,0x55,0x66,0x77,0x88])CAN_FRAME_FMT="=IB3x8s"# can_id, dlc, pad, data# ===== 工具函数 =====defrun(cmd):subprocess.run(cmd,shell=True,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)defsetup_can():run("sudo ifconfig can0 down")run("sudo ifconfig can1 down")run("sudo ip link set can0 up type can bitrate 1000000")run("sudo ip link set can1 up type can bitrate 1000000 dbitrate 1000000 restart-ms 1000 berr-reporting on fd on")run("sudo ifconfig can0 txqueuelen 65536")run("sudo ifconfig can1 txqueuelen 65536")defopen_can(iface,is_tx):s=socket.socket(socket.AF_CAN,socket.SOCK_RAW,socket.CAN_RAW)s.bind((iface,))flt=struct.pack("=II",CAN_ID,STD_MASK)s.setsockopt(SOL_CAN_RAW,CAN_RAW_FILTER,flt)ifis_tx:s.setsockopt(SOL_CAN_RAW,CAN_RAW_LOOPBACK,0)else:s.setsockopt(SOL_CAN_RAW,CAN_RAW_RECV_OWN_MSGS,0)returnsdefpack_frame():returnstruct.pack(CAN_FRAME_FMT,CAN_ID,8,PAYLOAD)deffmt_frame():return"ID=0x123 DATA=11 22 33 44 55 66 77 88"# ===== 发送线程 =====defsend_loop(sock,rate,end_t,stat,key,tag):interval=1.0/rate next_t=time.perf_counter()frame=pack_frame()printed=Falsewhiletime.perf_counter()<end_t:now=time.perf_counter()ifnow<next_t:time.sleep(min(0.0005,next_t-now))continuetry:sock.send(frame)stat[key]+=1ifnotprinted:print(f"[{tag}TX OK]{fmt_frame()}")printed=TrueexceptOSError:passnext_t+=interval# ===== 接收线程 =====defrecv_loop(sock,end_t,stat,key,tag):printed=Falsewhiletime.perf_counter()<end_t:r,_,_=select.select([sock],[],[],0.05)ifnotr:continueframe=sock.recv(16)can_id,dlc,data=struct.unpack(CAN_FRAME_FMT,frame)ifcan_id&CAN_ECHO_FLAG:continueif(can_id&STD_MASK)==CAN_IDanddata[:8]==PAYLOAD:stat[key]+=1ifnotprinted:print(f"[{tag}RX OK]{fmt_frame()}")printed=True# ===== 主程序 =====defmain():parser=argparse.ArgumentParser()parser.add_argument("--no-setup",action="store_true",help="跳过 can 配置")parser.add_argument("--duration",type=float,default=3.0,help="每档测试秒数")parser.add_argument("--rates",default="10,50,100,200,500,1000,2000,3000,4000,5000")args=parser.parse_args()ifnotargs.no_setup:setup_can()tx0=open_can("can0",True)tx1=open_can("can1",True)rx0=open_can("can0",False)rx1=open_can("can1",False)rates=[int(x)forxinargs.rates.split(",")]print("\n=== CAN0 <-> CAN1 双向收发测试 ===")print("ID=0x123 DATA=11 22 33 44 55 66 77 88\n")print("rate(Hz) | can0->can1 sent/recv drop% | can1->can0 sent/recv drop%")forrateinrates:stat={"s0":0,"r1":0,"s1":0,"r0":0}start=time.perf_counter()send_end=start+args.duration recv_end=send_end+0.5ts0=threading.Thread(target=send_loop,args=(tx0,rate,send_end,stat,"s0","can0"))ts1=threading.Thread(target=send_loop,args=(tx1,rate,send_end,stat,"s1","can1"))tr0=threading.Thread(target=recv_loop,args=(rx0,recv_end,stat,"r0","can0"))tr1=threading.Thread(target=recv_loop,args=(rx1,recv_end,stat,"r1","can1"))tr0.start();tr1.start()ts0.start();ts1.start()ts0.join();ts1.join()tr0.join();tr1.join()d01=0ifstat["s0"]==0else(stat["s0"]-stat["r1"])/stat["s0"]*100d10=0ifstat["s1"]==0else(stat["s1"]-stat["r0"])/stat["s1"]*100print(f"{rate:7d}|{stat['s0']:5d}/{stat['r1']:5d}{d01:6.2f}%"f" |{stat['s1']:5d}/{stat['r0']:5d}{d10:6.2f}%")ifd01>1ord10>1:print(">> 已明显开始丢包，再提速意义不大了")breakprint("\n=== 测试结束 ===")if__name__=="__main__":ifos.geteuid()!=0:print("请 sudo 运行")exit(1)main()