dpdk简单实现

DPDK是INTEL公司开发的一款高性能的网络驱动组件，旨在为数据面应用程序提供一个简单方便的，完整的，快速的数据包处理解决方案，主要技术有用户态、轮询取代中断、零拷贝、网卡RSS、访存DirectIO等。

应用场景

也是简单介绍，随着对dpdk认识的深入，对应用的理解也会深入。

• 高性能网关：可以使数据通过dpdk而不经过内核的协议栈直达应用层，用以提升网关性能。dns可以这么做。
• 虚拟交换机：利用dpdk优秀的性能，跑一个虚拟的交换机或是路由器，以提升性能。
• 防火墙：对数据的出入口做一些限制。

先简单入个门，这里介绍取代协议栈的方法。

编译配置

环境配置

这里使用的是虚拟机安装，使用物理机配置，会麻烦一些，这里就不说了。

1.添加网卡

虚拟机中，需要增加一个网卡，专门给dpdk使用。

这里看到有两个网卡，其中桥接网卡是 dpdk 运行的网卡，NAT 网卡是 ssh 连接的网卡。
修改网卡配置信息

2.修改网卡配置信息

在对应虚拟机目录中，打开虚拟机配置文件，后缀名是.vmx。

将 ethernet0.virtualDev 由 e1000 修改为 vmxnet3，因为 vmware 的 vmxnet3 支持多队列网卡。注意ethernet0是对应的网卡名，不同机器可能不同，不要生硬地去找ethernet0，根据自己的网卡配置来。
改成vmxnet3，代表支持多列队网卡，网卡可以出发多个中断。可以通过命令查看，还是以eth0为例，先查看是否支持多队列网卡。

cat /proc/interrupt | grep eth0
1

8行是因为8核cpu，给了8个终端号，竖列代表每个cpu的响应次数。

如果没有设置多队列网卡，就只会输出一行信息，只响应一个中断号。
说到中断，啰嗦几句。将中断名与cpu核心对应的掩码进行设置，进行cpu粘合，让指定中断号由特定cpu处理。

# 这里1代表是1号cpu，注意是掩码，所以2对应2号cpu，4对应3号，8对4号，以此类推
echo 1 > /proc/irq/57/smp_affinity
1
2

如果使用nginx，还可以使特定worker进程绑定指定cpu，在nginx.conf中设置。

worker_cpu affinity 00000001 00000010 ... # ...代表省略
1

这样一来，就实现了特定进程处理特定中断。

3.修改 ubuntu 系统的启动参数

在/etc/default/grub中增加对应字段，主要是修改巨页参数

default_hugepages=1G hugepagesz=2M hugepages=1024 isolcpus=0-2
1

执行update-grub重启生效。

编译安装

从官网上下载https://core.dpdk.org/download/
这里选择的版本是19.08.2，不同版本直接子系统接口会有差异。
接下来执行dpdk目录下/usertools/dpdk-setup.sh脚本

选择编译的环境，这里选择39，不同机器的环境不一定相同，不要无脑抄。
选择39，开始编译，如果缺少编译依赖的工具，会报错，根据提示安装缺少的工具。剧透一下，有gcc、g++、python、libnuma-dev、build-essential等。
编译成功的话，会多出 x86_64-native-linux-gcc 的文件夹

然后，可以配置环境变量了

export RTE_SDK=/home/dpdk  #自己的安装目录
export RTE_TARGET=x86_64-native-linux-gcc  # 就是刚才生成的目录名
1
2

最后的设置
再次执行./usertools/dpdk-setup.sh，选择43、44、49。
选择 43 插入 IGB_UIO 模块， 选择网卡为 vmxnet3 会加载此模块；
选择 44 插入 VFIO 模块，选择网卡为 e1000 会加载此模块；
选择 49 绑定 igb_uio 模块，输入网卡对应的pci地址，选择49一开始会打印，选择“VMXNET3”字样的。绑定之前要先把要绑定的网卡down掉

ifconfig eth0 down
1

对环境进行测试

tsetpmd是一个使用DPDK软件包分发的参考应用程序，其主要目的是在网络接口的以太网端口之间转发数据包。我们在这里使用的目的，是测一下环境配通没有，不大量使用它。
还是执行脚本./usertools/dpdk-setup.sh，选择53执行testpmd。
这里bitmask填7，用3个cpu测，再输入命令show port info 0返回端口信息

testpmd>show port info 0
1

其实执行这个命令，不是真为了查看信息，是为了测试环境，打印出来信息，说明配置成功。

运行demo测试

环境通了，跑一跑代码。dpdk提供了一些样例，进入 example/helloworld目录，编译

gcc -o helloword main.c -I /usr/local/include/dpdk/ -ldpdk -lpthread -lnuma -ldl
1

其实，也可以make……
执行程序

./helloworld -l 0-7 -n 8 # 使用8核
1

出现如下字样，说明测试成功。

hello from core 1
hello from core 2
hello from core 3
hello from core 4
hello from core 5
hello from core 6
hello from core 7
hello from core 0
1
2
3
4
5
6
7
8

简单实现数据收发

环境跑通了，下面该写代码了。这里先实现最简单的数据收发功能。
直接贴代码。

#include 
#include 
#include 

#include 
#include 

//内存池的块大小设置
//不设置2的n次方，减1的好处，是如果分配整块（4096），就另外去放到大块中，不再放到小块这个内存池中
#define NUM_MBUFS (4096-1)
#define BURST_SIZE	32

int gDpdkPortId = 0;  //此端口非彼端口

//配置信息结构体，这里只需要设置一个成员
static const struct rte_eth_conf port_conf_default = {
	.rxmode = {.max_rx_pkt_len = RTE_ETHER_MAX_LEN }
};

static void ng_init_port(struct rte_mempool *mbuf_pool) {

	uint16_t nb_sys_ports= rte_eth_dev_count_avail();  //返回值是49绑定的网卡数
	if (nb_sys_ports == 0) {
		rte_exit(EXIT_FAILURE, "No Supported eth found\n");
	}

	struct rte_eth_dev_info dev_info;
	rte_eth_dev_info_get(gDpdkPortId, &dev_info);  //获取端口的信息
	
	const int num_rx_queues = 1;  //读队列个数
	const int num_tx_queues = 1;  //写队列个数
	struct rte_eth_conf port_conf = port_conf_default;
	rte_eth_dev_configure(gDpdkPortId, num_rx_queues, num_tx_queues, &port_conf);

    //设置读队列，参数0号队列，队列最多接收128个数据包
	if (rte_eth_rx_queue_setup(gDpdkPortId, 0 , 128, 
		rte_eth_dev_socket_id(gDpdkPortId),NULL, mbuf_pool) < 0) {

		rte_exit(EXIT_FAILURE, "Could not setup RX queue\n");

	}

	if (rte_eth_tx_queue_setup(gDpdkPortId, 0 , 128, 
		rte_eth_dev_socket_id(gDpdkPortId),NULL, mbuf_pool) < 0) {

		rte_exit(EXIT_FAILURE, "Could not setup RX queue\n");

	}

	if (rte_eth_dev_start(gDpdkPortId) < 0 ) {
		rte_exit(EXIT_FAILURE, "Could not start\n");
	}

	rte_eth_promiscuous_enable(gDpdkPortId);	

}


static void create_eth_ip_udp(uint8_t *msg, size_t total_len, uint8_t *dst_mac, 
	uint32_t src_ip, uint32_t dst_ip, uint16 src_port, uint16 dst_port) {

	struct rte_ether_addr src_mac;

	struct rte_ether_hdr *eth = (struct rte_ether_hdr *)msg;
	rte_memcpy(eth->d_addr.addr_bytes, dst_mac, RTE_ETHER_ADDR_LEN);

	rte_eth_macaddr_get(gDpdkPortId, &src_mac);
	rte_memcpy(eth->s_addr.addr_bytes, src_mac, RTE_ETHER_ADDR_LEN);
	eth->ether_type = htons(RTE_ETHER_TYPE_IPV4);

	struct rte_ipv4_hdr *ip = (struct rte_ipv4_hdr *)(eth+1);
	ip->version_ihl = 0x45;
	ip->type_of_service = 0;
	ip->total_length = htons(total_len - sizeof(struct rte_ether_hdr));
}

int main(int argc, char *argv[]) {

	if (rte_eal_init(argc, argv) < 0) {
		rte_exit(EXIT_FAILURE, "Error with EAL init\n");
		
	}

	struct rte_mempool *mbuf_pool = rte_pktmbuf_pool_create("mbuf pool", NUM_MBUFS,
		0, 0, RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id());
	if (mbuf_pool == NULL) {
		rte_exit(EXIT_FAILURE, "Could not create mbuf pool\n");
	}

	ng_init_port(mbuf_pool);

	while (1) {

		struct rte_mbuf *mbufs[BURST_SIZE];
		//接收数据
		unsigned num_recvd = rte_eth_rx_burst(gDpdkPortId, 0, mbufs, BURST_SIZE);
		if (num_recvd > BURST_SIZE) {
			rte_exit(EXIT_FAILURE, "Error receiving from eth\n");
		}
			
		unsigned i = 0;
		for (i = 0;i < num_recvd;i ++) {
            //就是拆包了
			struct rte_ether_hdr *ehdr = rte_pktmbuf_mtod(mbufs[i], struct rte_ether_hdr*);
			if (ehdr->ether_type != rte_cpu_to_be_16(RTE_ETHER_TYPE_IPV4)) {
				rte_pktmbuf_free(mbufs[i]);
				continue;
			}

			struct rte_ipv4_hdr *iphdr =  rte_pktmbuf_mtod_offset(mbufs[i], struct rte_ipv4_hdr *, 
				sizeof(struct rte_ether_hdr));
			
			if (iphdr->next_proto_id == IPPROTO_UDP) {

				struct rte_udp_hdr *udphdr = 
					(struct rte_udp_hdr *)((unsigned char*)iphdr + sizeof(struct rte_ipv4_hdr));

				if (ntohs( udphdr->dst_port) == 8888 ) {
					uint16_t length = ntohs(udphdr->dgram_len);
					*(((char*)udphdr) + length) = '\0';

					struct in_addr addr;
					addr.s_addr = iphdr->src_addr;
					printf("src: %s:%d, ", inet_ntoa(addr), ntohs(udphdr->src_port));

					addr.s_addr = iphdr->dst_addr;
					printf("dst: %s:%d, length:%d --> %s\n", inet_ntoa(addr), 
						ntohs(udphdr->src_port), length, (char *)(udphdr+1));
				}

				rte_pktmbuf_free(mbufs[i]);
			}
			
		}

	}

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

这里编译可以改掉实例的makefile，也可以自己gcc。
执行程序接收数据，但是这里有问题，给这个网卡发数据，以现在的配置，是收不到的，为什么？是因为没有在发送端配置arp信息，发送端不知道该往哪发。这里给出win端的配置。

arp -s 192.168.0.120 00-0c-29-85-2e-88
1

这样做无法指定添加到哪个网段，可能添加到了其他网段中，所以不推荐。
推荐使用如下方式

# 查看网卡信息，主要是要找到对应网段的网卡的Idx值
netsh i i show in
# 绑定ip和mac地址，这个23就是刚才查的Idx
netsh -c i i add neighbors 23 192.168.0.120 00-0c-29-85-2e-88
# 解除绑定
netsh -c "i i" delete neighbors 23
1
2
3
4
5
6

绑定以后，可以通过命令arp -a查看是否绑定成功。
绑定成功，发送数据，dpdk端可以收到了。