零拷贝技术研究与实现

一．基本概念

零拷贝（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*
4000
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