零拷貝技術研究與實現
 

--------------------------------------------------------------------------------
 
梁健（firstdot） 時間：2003-12-6 8:24:58 
   

感謝王超、史曉龍的共同研究與大力幫助
一．基本概念
零拷貝（zero-copy）基本思想是：數據報從網絡設備到用戶程序空間傳遞的過程中，減少數據拷貝次數，減少系統調用，實現CPU的零參與，徹底消除CPU在這方面的負載。實現零拷貝用到的最主要技術是DMA數據傳輸技術和內存區域映射技術。如圖1所示，傳統的網絡數據報處理，需要經過網絡設備到操作系統內存空間，系統內存空間到用戶應用程序空間這兩次拷貝，同時還需要經歷用戶向系統發出的系統調用。而零拷貝技術則首先利用DMA技術將網絡數據報直接傳遞到系統內核預先分配的地址空間中，避免CPU的參與；同時，將系統內核中存儲數據報的內存區域映射到檢測程序的應用程序空間（還有一種方式是在用戶空間建立一緩存，并將其映射到內核空間，類似于linux系統下的kiobuf技術），檢測程序直接對這塊內存進行訪問，從而減少了系統內核向用戶空間的內存拷貝，同時減少了系統調用的開銷，實現了真正的“零拷貝”。


圖1 傳統數據處理與零拷貝技術之比較
二．實現
在redhat7.3上通過修改其內核源碼中附帶的8139too.c完成零拷貝的試驗，主要想法是：在8139too網卡驅動模塊啟動時申請一內核緩存，并建立一數據結構對其進行管理，然后試驗性的向該緩存寫入多個字符串數據，最后通過proc文件系統將該緩存的地址傳給用戶進程；用戶進程通過讀proc文件系統取得緩存地址并對該緩存進行地址映射，從而可以從其中讀取數據。哈哈，為了偷懶，本文只是對零拷貝思想中的地址映射部分進行試驗，而沒有實現DMA數據傳輸（太麻煩了，還得了解硬件），本試驗并不是一個IDS產品中抓包模塊的一部分，要想真正在IDS中實現零拷貝，除了DMA外，還有一些問題需考慮，詳見本文第三節的分析。以下為實現零拷貝的主要步驟，詳細代碼見附錄。

步驟一：修改網卡驅動程序
a．在網卡驅動程序中申請一塊緩存：由于在linux2.4.X內核中支持的最大可分配連續緩存大小為2M，所以如果需要存儲更大量的網絡數據報文，則需要分配多塊非連續的緩存，并使用鏈表、數組或hash表來對這些緩存進行管理。

#define PAGES_ORDER 9
unsigned long su1_2
su1_2 = __get_free_pages(GFP_KERNEL,PAGES_ORDER);

b.向緩存中寫入數據：真正IDS產品中的零拷貝實現應該是使用DMA數據傳輸把網卡硬件接收到的包直接寫入該緩存。作為試驗，我只是向該緩存中寫入幾個任意的字符串，如果不考慮DMA而又想向緩存中寫入真正的網絡數據包，可以在8139too.c的rtl8139_rx_interrupt()中調用netif_rx()后插入以下代碼：

//put_pkt2mem_n++; //包個數
//put_mem(skb->data,pkt_size);
其中put_pkt2mem_n變量和put_mem函數見附錄。

c.把該緩存的物理地址傳到用戶空間：由于在內核中申請的緩存地址為虛擬地址，而在用戶空間需要得到的是該緩存的物理地址，所以首先要進行虛擬地址到物理地址的轉換，在linux系統中可以使用內核虛擬地址減3G來獲得對應的物理地址。把緩存的地址傳到用戶空間需要在內核與用戶空間進行少量數據傳輸，這可以使用字符驅動、proc文件系統等方式實現，在這里采用了proc文件系統方式。

int read_procaddr(char *buf,char **start,off_t offset,int count,int *eof,void *data)
{
    sprintf(buf,"%u\n",__pa(su1_2));
    *eof = 1;
    return 9;
}
create_proc_read_entry("nf_addr",0,NULL,read_procaddr,NULL);

步驟二：在用戶程序中實現對共享緩存的訪問
a.讀取緩存地址：通過直接讀取proc文件的方式便可獲得。

char addr[9];
int fd_procaddr;
unsigned long ADDR;
fd_procaddr = open("/proc/nf_addr",O_RDONLY);
read(fd_procaddr,addr,9);
ADDR = atol(addr);

b.把緩存映射到用戶進程空間中：在用戶進程中打開/dev/mem設備(相當于物理內存），使用mmap把網卡驅動程序申請的緩存映射到自己的進程空間，然后就可以從中讀取所需要的網絡數據包了。

char *su1_2;
int fd;
fd=open("/dev/mem",O_RDWR);    
su1_2 = mmap(0,PAGES*4*1024, PROT_READ|PROT_WRITE, MAP_SHARED, fd, ADDR);

三．分析
    零拷貝中存在的最關鍵問題是同步問題，一邊是處于內核空間的網卡驅動向緩存中寫入網絡數據包，一邊是用戶進程直接對緩存中的數據包進行分析（注意，不是拷貝后再分析），由于兩者處于不同的空間，這使得同步問題變得更加復雜。緩存被分成多個小塊，每一塊存儲一個網絡數據包并用一數據結構表示，本試驗在包數據結構中使用標志位來標識什么時候可以進行讀或寫，當網卡驅動向包數據結構中填入真實的包數據后便標識該包為可讀，當用戶進程對包數據結構中的數據分析完后便標識該包為可寫，這基本解決了同步問題。然而，由于IDS的分析進程需要直接對緩存中的數據進行入侵分析，而不是將數據拷貝到用戶空間后再進行分析，這使得讀操作要慢于寫操作，有可能造成網卡驅動無緩存空間可以寫，從而造成一定的丟包現象，解決這一問題的關鍵在于申請多大的緩存，太小的緩存容易造成丟包，太大的緩存則管理麻煩并且對系統性能會有比較大的影響。

四．附錄
a.    8139too.c中加入的代碼

/*add_by_liangjian for zero_copy*/
#include <linux/wrapper.h>
#include <asm/page.h>
#include <linux/slab.h>
#include <linux/proc_fs.h>
#define PAGES_ORDER 9
#define PAGES 512
#define MEM_WIDTH    1500
/*added*/

/*add_by_liangjian for zero_copy*/
struct MEM_DATA
{
    //int key;
    unsigned short width;/*緩沖區寬度*/
    unsigned short length;/*緩沖區長度*/
    //unsigned short wtimes;/*寫進程記數,預留，為以后可以多個進程寫*/
    //unsigned short rtimes;/*讀進程記數,預留，為以后可以多個進程讀*/
    unsigned short wi;/*寫指針*/
    unsigned short ri;/*讀指針*/
} * mem_data;
struct MEM_PACKET
{
    unsigned int len;
    unsigned char packetp[MEM_WIDTH - 4];/*sizeof(unsigned int) == 4*/
};
unsigned long su1_2;/*緩存地址*/
/*added*/

/*add_by_liangjian for zero_copy*/
//刪除緩存
void del_mem()
{
    int pages = 0;
    char *addr;
    addr = (char *)su1_2;
    while (pages <=PAGES -1)
    {
        mem_map_unreserve(virt_to_page(addr));
        addr = addr + PAGE_SIZE;
        pages++;
    }
    free_pages(su1_2,PAGES_ORDER);    
}    
void init_mem()
/********************************************************
*                  初始化緩存
*       輸入:   aMode:    緩沖區讀寫模式:  r,w        *
*       返回:   00:     失敗                        *
*               >0:     緩沖區地址                  *
********************************************************/
{
    int i;
    int pages = 0;
    char *addr;
    char *buf;
    struct MEM_PACKET * curr_pack;
    
    su1_2 = __get_free_pages(GFP_KERNEL,PAGES_ORDER);
    printk("[%x]\n",su1_2);
    addr = (char *)su1_2;
    while (pages <= PAGES -1)
    {
        mem_map_reserve(virt_to_page(addr));//需使緩存的頁面常駐內存
        addr = addr + PAGE_SIZE;
        pages++;
    }
    mem_data = (struct MEM_DATA *)su1_2;
    mem_data[0].ri = 1;
          mem_data[0].wi = 1;
          mem_data[0].length = PAGES*4*1024 / MEM_WIDTH;
          mem_data[0].width = MEM_WIDTH;
    /* initial su1_2 */
    for(i=1;i<=mem_data[0].length;i++)
    {
        buf = (void *)((char *)su1_2 + MEM_WIDTH * i);
        curr_pack = (struct MEM_PACKET *)buf;
        curr_pack->len = 0;
    }    
}
int put_mem(char *aBuf,unsigned int pack_size)
/****************************************************************
*                 寫緩沖區子程序                                *
*       輸入參數    :   aMem:   緩沖區地址                      *
*                       aBuf:   寫數據地址                      *
*       輸出參數    :   <=00 :  錯誤                            *
*                       XXXX :  數據項序號                      *
*****************************************************************/
{
    register int s,i,width,length,mem_i;
    char *buf;
    struct MEM_PACKET * curr_pack;

    s = 0;
    mem_data = (struct MEM_DATA *)su1_2;
    width  = mem_data[0].width;
    length = mem_data[0].length;
    mem_i  = mem_data[0].wi;
    buf = (void *)((char *)su1_2 + width * mem_i);

    for (i=1;i<length;i++){
        curr_pack = (struct MEM_PACKET *)buf;
            if  (curr_pack->len == 0){
                    memcpy(curr_pack->packetp,aBuf,pack_size);
                    curr_pack->len = pack_size;;
                s = mem_i;
            mem_i++;
                    if  (mem_i >= length)
                        mem_i = 1;
                mem_data[0].wi = mem_i;
                break;
            }
            mem_i++;
            if  (mem_i >= length){
                    mem_i = 1;
                    buf = (void *)((char *)su1_2 + width);
            }
            else buf = (char *)su1_2 + width*mem_i;
        }

    if(i >= length)
            s = 0;
    return s;
}
// proc文件讀函數
int read_procaddr(char *buf,char **start,off_t offset,int count,int *eof,void *data)
{
    sprintf(buf,"%u\n",__pa(su1_2));
    *eof = 1;
    return 9;
}
/*added*/

在8139too.c的rtl8139_init_module()函數中加入以下代碼：
/*add_by_liangjian for zero_copy*/
    put_pkt2mem_n = 0;
    init_mem();
    put_mem("data1dfadfaserty",16);
    put_mem("data2zcvbnm",11);
    put_mem("data39876543210poiuyt",21);
    create_proc_read_entry("nf_addr",0,NULL,read_procaddr,NULL);
/*added */    

在8139too.c的rtl8139_cleanup_module()函數中加入以下代碼：
/*add_by_liangjian for zero_copy*/
    del_mem();
    remove_proc_entry("nf_addr",NULL);
/*added*/    

b．用戶空間讀取緩存代碼

#include <stdio.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/mman.h>
#include <fcntl.h>
#define PAGES 512
#define MEM_WIDTH 1500
struct MEM_DATA
{
    //int key;
    unsigned short width;/*緩沖區寬度*/
    unsigned short length;/*緩沖區長度*/
    //unsigned short wtimes;/*寫進程記數,預留，為以后可以多個進程寫*/
    //unsigned short rtimes;/*讀進程記數,預留，為以后可以多個進程讀*/
    unsigned short wi;/*寫指針*/
    unsigned short ri;/*讀指針*/
} * mem_data;

struct MEM_PACKET
{
    unsigned int len;
    unsigned char packetp[MEM_WIDTH - 4];/*sizeof(unsigned int) == 4*/
};

int get_mem(char *aMem,char *aBuf,unsigned int *size)
/****************************************************************
*                 讀緩沖區子程序                                *
*       輸入參數    :   aMem:   緩沖區地址                      *
*                       aBuf:   返回數據地址, 其數據區長度應大于*
*                               緩沖區寬度                      *
*       輸出參數    :   <=00 :  錯誤                            *
*                       XXXX :  數據項序號                      *
*****************************************************************/
{
    register int i,s,width,length,mem_i;
    char     *buf;
    struct MEM_PACKET * curr_pack;

    s = 0;
    mem_data = (void *)aMem;
    width  = mem_data[0].width;
    length = mem_data[0].length;
    mem_i  = mem_data[0].ri;
    buf = (void *)(aMem + width * mem_i);

    curr_pack = (struct MEM_PACKET *)buf;
    if  (curr_pack->len != 0){/*第一個字節為0說明該部分為空*/
            memcpy(aBuf,curr_pack->packetp,curr_pack->len);
            *size = curr_pack->len;
            curr_pack->len = 0;
            s = mem_data[0].ri;
            mem_data[0].ri++;
            if(mem_data[0].ri >= length)
                    mem_data[0].ri = 1;
            goto ret;
        }
    
    for (i=1;i<length;i++){
            mem_i++;/*繼續向后找，最糟糕的情況是把整個緩沖區都找一遍*/
            if  (mem_i >= length)
                mem_i = 1;
            buf = (void *)(aMem + width*mem_i);
            curr_pack = (struct MEM_PACKET *)buf;
            if  (curr_pack->len == 0)
                    continue;
            memcpy(aBuf,curr_pack->packetp,curr_pack->len);
            *size = curr_pack->len;
            curr_pack->len = 0;
            s = mem_data[0].ri = mem_i;
            mem_data[0].ri++;
            if(mem_data[0].ri >= length)
            mem_data[0].ri = 1;
            break;
        }

    ret:
    return s;
}

int main()
{
    char *su1_2;
    char receive[1500];
    int i,j;
    int fd;
    int fd_procaddr;
    unsigned int size;
    char addr[9];
    unsigned long ADDR;
    
    j = 0;
    /*open device 'mem' as a media to access the RAM*/
    fd=open("/dev/mem",O_RDWR);    
    fd_procaddr = open("/proc/nf_addr",O_RDONLY);
    read(fd_procaddr,addr,9);
    ADDR = atol(addr);
    close(fd_procaddr);
    printf("%u[%8lx]\n",ADDR,ADDR);
    /*Map the address in kernel to user space, use mmap function*/
    su1_2 = mmap(0,PAGES*4*1024, PROT_READ|PROT_WRITE, MAP_SHARED, fd, ADDR);
    perror("mmap");
    while(1)
    {
        bzero(receive,1500);
        i = get_mem(su1_2,receive,&size);
        if (i != 0)
        {
            j++;
            printf("%d:%s[size = %d]\n",j,receive,size);
        }    
        else 
        {
            printf("there have no data\n");
            munmap(su1_2,PAGES*4*1024);
            close(fd);
            break;
        }
    }
    while(1);
}

五．參考文獻
1．CHRISTIAN KURMANN, FELIX RAUCH ,THOMAS M. STRICKER. 
Speculative Defragmentation - Leading Gigabit Ethernet to True Zero-Copy Communication
2．ALESSANDRO RUBINI,JONATHAN CORBET.《LINUX DEVICE DRIVERS 2》,O’Reilly & Associates 2002.
3．胡希明,毛德操.《LINUX 內核源代碼情景分析》,浙江大學出版社 2001


關于作者：梁健，華北計算技術研究所在讀碩士研究生，研究方向：信息安全。論文開題為《基于系統調用分析的主機異常入侵檢測與防御》。對IDS有兩年多的研究經驗，熟悉linux內核，熟悉linux c/c++編程、win32 API編程，對網絡和操作系統安全感興趣。