這一年關於PCIE高速採集卡的業務量激增,究其緣由,發現百度「xilinx pcie dma」,出來的都是本人的博客。前期的博文主要以教程爲主,教你們如何理解PCIE協議以及如何正確使用PCIE相關的IP核,由於涉及到商業道德,本人不能將公司自研的IP核以及相關工程應用放到網上。但爲了知足你們對PCIE高速採集卡這塊的業務需求,博主特意利用業餘時間,使用XDMA這個xilinx官方IP,配合xilinx提供的linux驅動,在KC705開發板上實現了一套高速採集系統,該系統可對前端ADC產生的不大於2GB/s的連續或非連續數據進行實時採集,同時該採集卡具有數據發送功能,能夠將用戶文件或者內存中的數據寫到FPGA的發送FIFO中,速率約爲2GB/s,該採集卡具有上位機讀寫FPGA用戶寄存器的功能,讀寫接口爲local bus接口,方便易用。固然,若是您的高速採集卡須要大於2GB/s的採集速率,那博主只能拿出壓箱底的另外一套QDMA採集系統了,該系統在VC709上具有6.1GB/s的連續採集能力,要知道VC709的PCIE理論帶寬都只有6.4GB/s,高達95%的傳輸效率真的很恐怖了,固然這套QDMA採集系統能有如此威力,主要拜FPGA大神馬克傑所賜,馬哥寫的驅動充分發揮了系統的最大性能,吊打Xilinx的官方驅動。html
一、XILINX KC705開發板前端
二、pg195-pcie-dma.pdflinux
三、Vivado2018.2套件算法
四、X86主機一臺,安裝64位centos7.4 1708操做系統windows
五、XDMA linux驅動2018版本,GitHub上有下載。centos
採集卡系統框圖api
從左到右從上到下依次介紹模塊以及相應功能緩存
1.data_genpost
此模塊模擬ADC產生的流數據,在本系統中,採樣時鐘250M,模擬AD數據位寬64位,故AD實時採樣速率爲2GB/s,可經過Send_En隨時停止或繼續數據產生。性能
2.axis data width converter
此模塊將流數據64位位寬轉換成128位位寬,時鐘250M。
3.axis data fifo
此模塊爲流數據緩衝FIFO,深度不大,128足矣,真正的緩存要靠ddr完成。
4.YDMA
此模塊爲博主本身寫的採集卡DMA控制器,該控制器的功能主要分四塊:一,將收到的ST數據(axis接口)轉換成MM數據(axi接口)寫入DDR3;二,將須要發送的MM數據(axi接口)從DDR3中取出後轉換成ST數據(axis接口)供用戶使用;三,將XDMA輸出的BYPASS接口轉換成local_bus接口供用戶讀寫寄存器使用;四,中斷控制器,將寫DDR和讀DDR產生的中斷送給XDMA,用戶可設置包大小,中斷包個數,中斷超時時間。
5.user_reg_define
此模塊爲用戶寄存器讀寫模塊,讀寫接口爲local bus接口,此用例中咱們用它來配置Send_En。
6.axis_data_check
此模塊用來校驗上位機發下來的數據。
7.XDMA
此模塊由上位機驅動控制,經過PCIE以SG_DMA的方式讀寫DDR3中的數據。
8.memory interface generator
此模塊爲DDR3控制器,使用AXI接口。
綜上,整個採集卡包含兩個方向的數據流向:FPGA>>PC:
data_gen->data_fifo->YDMA->DDR3->XDMA
PC>>FPGA:XDMA->DDR3->YDMA->data_check
固然FPGA邏輯部分最大的難點就在YDMA上,爲了知足對任意包長、任意間隔的連續或非連續數據進行實時採集,須要產生大量的中斷以及與之相對應的ddr緩存地址和緩存長度等中斷信息,但XDMA驅動最大的bug偏偏出在中斷上,爲了規避XDMA的中斷bug,又要提高總體的採集性能,須要對中斷控制作精細設計。同時,對於那些突發的情況,好比採集數據忽然中斷的狀況、急停急起的狀況,都須要經過邏輯和軟件的相互配合,才能跑出使人滿意的採集效果。至於PC往FPGA發數據這個功能對於採集卡來講是錦上添花而已。由於有客戶提出,須要將採集到的數據作處理,處理完後經過FPGA再發到另外一個設備上,故我在YDMA上作了一個發數據的功能,用戶接口也是你們最熟悉易用的axis(fifo)接口。由於YDMA包含必定技術含量,故該採集系統不能免費提供給你們,須要的用戶能夠聯繫我談價格。
XDMA的驅動是官方提供的,這裏不作詳細解讀,總之XDMA驅動就是把PCIE DMA包成了多種字符設備:xdma_h2c,xdma_c2h,xdma_user,xdma_control,xdma_bypass,xdma_events
通過本人測試使用,我只推薦使用xdma_h2c,xdma_c2h,xdma_bypass,xdma_events這四個字符設備。xdma_h2c用來把數據從內存寫到FPGA的DDR,xdma_c2h用來把數據從FPGA的DDR讀到內存,xdma_bypass用來配置FPGA的用戶寄存器,xdma_events用來讀取用戶中斷。
下面咱們來看看採集卡的測試程序咱們是怎麼寫的,裏面給出了詳細的註釋:
#define _BSD_SOURCE #define _XOPEN_SOURCE 500 #include <assert.h> #include <time.h> #include <fcntl.h> #include <getopt.h> #include <stdint.h> #include <string.h> #include <unistd.h> #include <stdio.h> #include <stdlib.h> #include <fcntl.h> #include <memory.h> #include <sys/stat.h> #include <sys/types.h> #include <sys/ioctl.h> #include <sys/time.h> #include <sys/resource.h> #include <sys/wait.h> #include <pthread.h> #include <sched.h> #include <semaphore.h> #include <sys/mman.h> #include <errno.h> //#include "dma_utils.c" #define FATAL do { fprintf(stderr, "Error at line %d, file %s \n", __LINE__, __FILE__); exit(1); } while(0) #define DEVICE_NAME_H2C "/dev/xdma0_h2c_0" #define DEVICE_NAME_C2H "/dev/xdma0_c2h_0" #define DEVICE_NAME_REG "/dev/xdma0_bypass" #define MAP_SIZE (1*1024*1024) #define MAP_MASK (MAP_SIZE - 1) #define RCV_EN_CMD 0 #define RX_DM_RST 1 struct timezone tz_time; struct timeval tv_time3; struct timeval tv_time4; pthread_t rcv_tid ; pthread_t print_sta_id; pthread_t event_thread; int work = 0 ; int lxcj =0; int int_rc; unsigned int lastData = 0; unsigned int rcvPktNum = 0; unsigned long rcvBytes = 0; unsigned long rcvBytes_l = 0; unsigned int errnum = 0 ; int c2h_fd ; int h2c_fd ; int control_fd; int interrupt_fd; void *control_base; static sem_t int_sem_rx; static sem_t int_sem_tx; char *device_c2h = DEVICE_NAME_C2H; char *device_h2c = DEVICE_NAME_H2C; char *device_reg = DEVICE_NAME_REG; static void write_control(void *base_addr,int offset,uint32_t val);//寫用戶寄存器 static uint32_t read_control(void *base_addr,int offset);//讀用戶寄存器 /*開中斷*/ int open_event(char *devicename) { int fd; fd=open(devicename,O_RDWR|O_SYNC ); if(fd==-1) {printf("open event error\n"); return -1;} return fd; } /*獲取用戶中斷*/ int read_event(int fd) { int val; read(fd,&val,4); return val; } /*打開字符設備*/ static int open_control(char *filename) { int fd; fd = open(filename, O_RDWR | O_SYNC); if(fd == -1) { printf("open control error\n"); return -1; } return fd; } /*獲取設備對應的內存映射地址*/ static void *mmap_control(int fd,long mapsize) { void *vir_addr; vir_addr = mmap(0, mapsize, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0); return vir_addr; } /*寫用戶寄存器*/ static void write_control(void *base_addr,int offset,uint32_t val) { //uint32_t writeval = htoll(val); *((uint32_t *)(base_addr+offset)) = val; } /*讀用戶寄存器*/ static uint32_t read_control(void *base_addr,int offset) { uint32_t read_result = *((uint32_t *)(base_addr+offset)); //read_result = ltohl(read_result); return read_result; } /*打印進程,5秒打印一次統計信息,包含收到的包個數,錯誤包個數,以及當前的採集速率*/ void *printStatus() { unsigned int lostNum = 0 ; unsigned int allNum = 0 ; while( work ==1) { sleep(5); printf("rcvPkt[%8x], err[%8x] , rate[%d]MBps\n", rcvPktNum, errnum , (rcvBytes-rcvBytes_l)/5/1000000 ); rcvBytes_l = rcvBytes ; } } /*數據校驗,由於模擬ADC數據是累加數,故收到的當前包的第一個數應該是上一次包的第一個數加上上次包的長度*/ void checkData(unsigned int *add, unsigned int len) { if(lastData != add[0] & lastData!=0 ) { errnum ++; if(errnum<20 )printf("l[%8x], n[%8x][%8x], [%8x], p[%x] , len[%d]\n", lastData , add[0], add[1], add[0] - lastData , (add[0] - lastData)/len , len) ; } rcvBytes = rcvBytes + len ; lastData = add[0] + len/8 ; } /*ADC連續數據採集處理進程*/ void *procPkt( ) { int i; int rxint_rc;//接收中斷信息寄存器返回值 unsigned int * rxBuf;//接收數據存放的地址 int rxlen; //接收數據的長度 int c2h0_inbuf =0; c2h_fd= open(device_c2h, O_RDWR | O_NONBLOCK);//打開xdma_c2h字符設備 posix_memalign((void **)&rxBuf, 1024, 8*1024*1024);//開一個8M的內存空間用於暫存接收數據 printf("procAD up. \n" ); while( work ==1 ) { if(lxcj==1) { //assert(c2h_fd >= 0); sem_wait(&int_sem_rx); //等待用戶接收中斷 rxint_rc=read_control(control_base,0x10020);//從接收中斷寄存器中獲取接收中斷相關的中斷信息 int icnt; if((rxint_rc&0x80000000)>>31) icnt= rxint_rc &0x00ffffff;//接收中斷寄存器bit31表示是否有接收中斷,bit23-bit0表示有幾個中斷包 else continue; write_control(control_base,0x10020,icnt);//清中斷寄存器,寫入的內容爲即將要處理的中斷包個數 for(i=0;i<icnt;i++) //處理中斷包 { int count = read_control(control_base,0x10018);//讀接收中斷狀態FIFO,獲取當前中斷包的實際長度 off_t off = lseek(c2h_fd, c2h0_inbuf, SEEK_SET);//和FPGA協商好從DDR3的0地址開始存放接收數據,故軟件從0地址開始取數據 rxlen = read(c2h_fd, rxBuf, count);//從DDR中取出數據放入rxbuffer write_control(control_base,0x10018,1); //清接收中斷FIFO c2h0_inbuf = c2h0_inbuf + 0x400000 ;//本測試用例中設置的中斷包最大長度爲4M if(c2h0_inbuf==0x40000000) c2h0_inbuf = 0;//當DDR3偏移達到1G的時候從新歸零 if(rxlen > 0) rcvPktNum ++ ;//統計接收包個數 checkData(rxBuf , rxlen) ;//校驗接受到的包是否爲連續數 } } } pthread_exit(0); } /*寫數據進程,此用例中爲發送任意大小文件*/ void h2c_process(char *filename) { h2c_fd= open(device_h2c, O_RDWR | O_NONBLOCK);//打開xdma_h2c字符設備 assert(h2c_fd>0); uint32_t send_len;//單次數據包發送長度,本測試用例中以4M爲單位 uint32_t send_addr=0x0; int rc; int file_fd; uint64_t size;//發送文件的實際大小 uint64_t snd_cnt; struct stat fileStat; file_fd = open(filename, O_RDONLY);//打開發送文件 assert(file_fd >= 0); rc= stat(filename, &fileStat ); size = fileStat.st_size ; //獲取發送文件的大小 snd_cnt =size; char *sendbuff = NULL; posix_memalign((void **)&sendbuff, 1024/*alignment*/, 8*1024*1024);//開一塊8M的內存空間 gettimeofday(&tv_time3, &tz_time); while(size!=0) { sem_wait(&int_sem_tx);//等待發送中斷信號量,該信號量初始值爲9 if(size>0x400000) send_len = 0x400000; else send_len = size; off_t off_file = lseek(file_fd, send_addr, SEEK_SET); rc = read(file_fd, sendbuff, send_len);//將數據從文件中讀到sendbuffer off_t off_h2c = lseek(h2c_fd, send_addr, SEEK_SET); //printf("send_len=%d\n,sendbuff=%x\n",send_len,sendbuff); rc = write(h2c_fd, sendbuff, send_len);//將sendbuffer中的數據發送到DDR3 write_control(control_base,0x11000,send_addr);//將DDR3數據緩存地址寫入FPGA端的發送地址寄存器 write_control(control_base,0x11010,send_len); //將DDR3數據緩存長度寫入FPGA端的發送長度寄存器 //printf("rc=%d\n",rc); assert(rc == send_len); size = size - send_len; send_addr = send_addr + send_len; if(send_addr==0x40000000) send_addr = 0;//發送數據的DDR3緩存偏移地址爲1G時歸零 } gettimeofday(&tv_time4, &tz_time); printf("write done\n"); printf(" 時間 %ld useconds\n", (tv_time4.tv_sec - tv_time3.tv_sec) * 1000000 + tv_time4.tv_usec - tv_time3.tv_usec); printf(" 數據量 %ld 字節\n", snd_cnt); printf(" 帶寬 %ld MB/s\n", snd_cnt / ((tv_time4.tv_sec - tv_time3.tv_sec) * 1000000 + tv_time4.tv_usec - tv_time3.tv_usec)); if (file_fd >= 0) close(file_fd); free(sendbuff); } /*中斷處理進程*/ void *event_process() { int i; int txint_rc; interrupt_fd = open_event("/dev/xdma0_events_0"); //打開用戶中斷 while(work==1) { read_event(interrupt_fd); //獲取用戶中斷 int_rc=read_control(control_base,0x00000); //讀總中斷寄存器 switch(int_rc) { case 1: //接收中斷 sem_post(&int_sem_rx); break; case 2: //發送中斷 txint_rc=read_control(control_base,0x11020); //從發送中斷寄存器中獲取發送中斷相關的中斷信息 int txicnt; if((txint_rc&0x80000000)>>31) txicnt= txint_rc &0x00ffffff;//發送中斷寄存器bit31表示是否有發送中斷,bit23-bit0表示發出了幾個中斷包 else break; write_control(control_base,0x11020,txicnt);//清中斷寄存器,寫入的內容爲即將要處理的中斷包個數 for(i=0;i<txicnt;i++) sem_post(&int_sem_tx); //爲每一個發出的中斷包釋放一個信號量 break; default: break; } } pthread_exit(0); } int main(int argc, char *argv[]) { ssize_t rc; char inp ; unsigned int * rxBuf; posix_memalign((void **)&rxBuf, 1024, 1024*1024*1024); control_fd = open_control("/dev/xdma0_bypass");//打開bypass字符設備 control_base = mmap_control(control_fd,MAP_SIZE);//獲取bypass映射的內存地址 //c2h_fd= open(device_c2h, O_RDWR | O_NONBLOCK); //h2c_fd= open(device_h2c, O_RDWR | O_NONBLOCK); sem_init(&int_sem_rx, 0, 0); sem_init(&int_sem_tx, 0, 9); work =1 ; pthread_create(&rcv_tid , NULL, procPkt, NULL); pthread_create(&print_sta_id, NULL, printStatus, NULL ); pthread_create(&event_thread, NULL, event_process, NULL); write_control(control_base,0x10028,0xFFFFFF08);//寫接收中斷控制寄存器,bit31-bit8爲中斷超時時間,bit7-bit0爲多少個包產生一次中斷 write_control(control_base,0x11028,0xFFFFFF00);//寫發送中斷控制寄存器,bit31-bit8爲中斷超時時間,bit7-bit0爲多少個包產生一次中斷 char *file_write = "/run/media/root/software/CentOS-7-x86_64-Everything-1708/CentOS-7-x86_64-Everything-1708.iso"; while(inp!='o') { inp = getchar(); switch(inp) { case 'w': h2c_process(file_write); break; case 'r': write_control(control_base,0x10030,4);//復位接收DMA rc=read( c2h_fd, rxBuf, 1*1024); printf("rc=%x\n",rc); break; case 's': write_control(control_base,0x10030,4);//復位接收DMA lxcj=1; write_control(control_base,0x10038,1);//使能接收 break; case 'e': write_control(control_base,0x10038,0);//中止接收 sleep(2); lxcj=0; break; case 't': write_control(control_base,0x30008,1);//使能模擬ADC數據發送 break; case 'p': write_control(control_base,0x30008,0);//暫停模擬ADC數據發送 break; case 'o': write_control(control_base,0x10030,1);//復位接收DMA write_control(control_base,0x11030,1);//復位發送DMA break; default: break; } } work =0 ; out: close(c2h_fd); close(h2c_fd); return rc; }
本採集系統測試環境爲X86主機,CPU爲Intel 酷睿i7 8700K,FPGA選用xilinx公司的KC705開發板,操做系統爲centos7.4 1708,內核版本3.10.0-693,博主最近會在windows上作一版測試程序,到時候分享給須要的朋友。
本博文展現的PCIE高速採集系統主要面向有這方面工程應用需求的朋友,不建議初學者做爲學習使用。本人從事高速總線接口已七年有餘,積累了大量總線相關的FPGA設計經驗,主要涉及FC、rapidio、千兆、萬兆以太網、lvds、mlvds、can、42二、1553B。同時也可承接算法加速或者視頻圖像處理等相關項目。最後放上一段基於QDMA(非xilinx的官方IP)的PCIe高速採集卡在VC709上的測試結果,致敬前輩馬哥!