本篇文章介绍DPDK下数据包重组实例代码简单逻辑及使用时注意事项，对应DPDK官网链接为IP Reassembly Sample Application，详细实现方式可参看源代码梳理，代码路径为dpdk-20.11.3/examples/ip_reassembly。

操作系统版本：CentOS 8.4

DPDK版本：dpdk-20.11.3

1、样例功能

样例代码主要实现重组IP层分片报文，然后将重组后的报文转发出去，如类型为IPV4的数据包A（包总长度1434）、B（包总长度1434）、C（包总长度42）重组后得到数据包D（包总长度2842），集齐完毕之后将数据包A、B、C、D依次根据匹配规则转发到指定端口，流程结束。

对于解析匹配转发功能的实现可参考dpdk-l3fwd样例解析这篇文章，这里重点说一下IP重组功能的实现。

如果实现IP重组功能，理论上需要一个API接口，每次接收一个数据包，首先判断这个报文是否是分片包，如果是分片包则调用重组API接口，主要实现传入的数据包如果是最后一个包（就差它就重组完成了），那么将重组后的数据包作为返回值从API接口中获取到，如果不是最后一个包，那么将这个数据包插入到类似于HASH表或者链表中，流程到此结束，继续解析下一个数据包。

实际上DPDK的IP重组样例所实现流程就是这样，下面根据代码流程逐步进行了解。

2、IP重组模块

在此之前需要考虑几个问题：

1、是否需要考虑超时删除及如何确定超时时间？（时间太短容易缺失，也不可能将分片包一直保留，这样内存就爆了）

2、重组后的数据包是什么形式存在的？（已经重组好为一个数据包 或者 将分片的数据包有序链接为一个链表）

3、如何确定最多能够实现多少个分片包的重组？

一、分片节点哈希表初始化

从main函数开始，对于EAL初始化、接收发送队列初始化这些和重组功能不相干的这里不再赘述，首先，调用setup_queue_tbl对分片报文所需要的ip_frag_table表进行初始化，流程如下：

这里调用rte_ip_frag_table_create用来创建ip_frag_table，接口功能解释如下：

/**
 * Create a new IP fragmentation table.
 *
 * @param bucket_num
 *   Number of buckets in the hash table. // 哈希表桶数量
 * @param bucket_entries
 *   Number of entries per bucket (e.g. hash associativity).	// 每个桶下挂在的节点数量
 *   Should be power of two.	// 应该为2的幂次方
 * @param max_entries
 *   Maximum number of entries that could be stored in the table.	// 哈希表存储的最大节点数量
 *   The value should be less or equal then bucket_num * bucket_entries. // 该值需小于或等于 桶数量 * 桶节点数量
 * @param max_cycles
 *   Maximum TTL in cycles for each fragmented packet.	// 每个分片包最长超时时间（TTL)
 * @param socket_id
 *   The *socket_id* argument is the socket identifier in the case of	
 *   NUMA. The value can be *SOCKET_ID_ANY* if there is no NUMA constraints.
 * @return
 *   The pointer to the new allocated fragmentation table, on success. NULL on error.	// 返回值为哈希表的地址，创建失败则为NULL
 */
struct rte_ip_frag_tbl * rte_ip_frag_table_create(uint32_t bucket_num,
		uint32_t bucket_entries,  uint32_t max_entries,
		uint64_t max_cycles, int socket_id);

二、分片包重组接口

进入到循环处理接口main_loop，对接收到的数据包调用reassemble接口进行数据包重组。

这里重点介绍reassemble接口流程

static inline void
reassemble(struct rte_mbuf *m, uint16_t portid, uint32_t queue,
	struct lcore_queue_conf *qconf, uint64_t tms)
{
    
	struct rte_ether_hdr *eth_hdr;
	struct rte_ip_frag_tbl *tbl;
	struct rte_ip_frag_death_row *dr;
	struct rx_queue *rxq;
	void *d_addr_bytes;
	uint32_t next_hop;
	uint16_t dst_port;

	rxq = &qconf->rx_queue_list[queue];

	eth_hdr = rte_pktmbuf_mtod(m, struct rte_ether_hdr *);

	dst_port = portid;

	/* if packet is IPv4 */ // 判断数据包类型IPV4 or IPV6
	if (RTE_ETH_IS_IPV4_HDR(m->packet_type)) {
    
		struct rte_ipv4_hdr *ip_hdr;
		uint32_t ip_dst;

		ip_hdr = (struct rte_ipv4_hdr *)(eth_hdr + 1);

		 /* if it is a fragmented packet, then try to reassemble. */ 
		if (rte_ipv4_frag_pkt_is_fragmented(ip_hdr)) {
     // 判断IPV4数据包是否时分片包
			struct rte_mbuf *mo;

			tbl = rxq->frag_tbl;
			dr = &qconf->death_row;

			/* prepare mbuf: setup l2_len/l3_len. */ // 获取数据链路层和IP层长度
			m->l2_len = sizeof(*eth_hdr);
			m->l3_len = sizeof(*ip_hdr);

			/* process this fragment. */	// 调用rte_ipv4_frag_reassemble_packet处理这个分片包
			mo = rte_ipv4_frag_reassemble_packet(tbl, dr, m, tms, ip_hdr);
			if (mo == NULL)	// 返回值为NULL 说明还没有重组完成，或者重组失败了
				/* no packet to send out. */
				return;

			/* we have our packet reassembled. */	// 不为空则说明已经重组完成
			if (mo != m) {
    	// 如果mo != m 说明最后一个进去的数据包m不是分片包的第一个
				m = mo; // 把重组完成的分片包链表头赋值给m
				eth_hdr = rte_pktmbuf_mtod(m,
					struct rte_ether_hdr *); // 获取链路层头
				ip_hdr = (struct rte_ipv4_hdr *)(eth_hdr + 1); // 获取IP层头
			}

			/* update offloading flags */
			m->ol_flags |= (PKT_TX_IPV4 | PKT_TX_IP_CKSUM); // 更新数据包的offloading flags
		}
		ip_dst = rte_be_to_cpu_32(ip_hdr->dst_addr); // 获取到目的IP（后面转发使用）

		/* Find destination port */ // 匹配获取转发端口ID
		if (rte_lpm_lookup(rxq->lpm, ip_dst, &next_hop) == 0 &&
				(enabled_port_mask & 1 << next_hop) != 0) {
    
			dst_port = next_hop;
		}
		eth_hdr->ether_type = rte_be_to_cpu_16(RTE_ETHER_TYPE_IPV4);

	} else if (RTE_ETH_IS_IPV6_HDR(m->packet_type)) {
     // IPV6同上，不在赘述
		/* if packet is IPv6 */
		struct ipv6_extension_fragment *frag_hdr;
		struct rte_ipv6_hdr *ip_hdr;

		ip_hdr = (struct rte_ipv6_hdr *)(eth_hdr + 1);

		frag_hdr = rte_ipv6_frag_get_ipv6_fragment_header(ip_hdr);

		if (frag_hdr != NULL) {
    
			struct rte_mbuf *mo;

			tbl = rxq->frag_tbl;
			dr  = &qconf->death_row;

			/* prepare mbuf: setup l2_len/l3_len. */
			m->l2_len = sizeof(*eth_hdr);
			m->l3_len = sizeof(*ip_hdr) + sizeof(*frag_hdr);

			mo = rte_ipv6_frag_reassemble_packet(tbl, dr, m, tms, ip_hdr, frag_hdr);
			if (mo == NULL)
				return;

			if (mo != m) {
    
				m = mo;
				eth_hdr = rte_pktmbuf_mtod(m,
							struct rte_ether_hdr *);
				ip_hdr = (struct rte_ipv6_hdr *)(eth_hdr + 1);
			}
		}

		/* Find destination port */
		if (rte_lpm6_lookup(rxq->lpm6, ip_hdr->dst_addr,
						&next_hop) == 0 &&
				(enabled_port_mask & 1 << next_hop) != 0) {
    
			dst_port = next_hop;
		}		
		eth_hdr->ether_type = rte_be_to_cpu_16(RTE_ETHER_TYPE_IPV6);
	}
	/* if packet wasn't IPv4 or IPv6, it's forwarded to the port it came from */

	/* 02:00:00:00:00:xx */
	d_addr_bytes = &eth_hdr->d_addr.addr_bytes[0];
	*((uint64_t *)d_addr_bytes) = 0x000000000002 + ((uint64_t)dst_port << 40);

	/* src addr */
	rte_ether_addr_copy(&ports_eth_addr[dst_port], &eth_hdr->s_addr);

	send_single_packet(m, dst_port); // 将本身不是分片的数据包或者重组完成后的数据包转发到指定端接口
}

这里涉及到rte_ipv4_frag_reassemble_packet接口，接口功能如下：

/**
 * This function implements reassembly of fragmented IPv4 packets.
 * Incoming mbufs should have its l2_len/l3_len fields setup correctly.
 *
 * @param tbl
 *   Table where to lookup/add the fragmented packet. // 需要查找或者添加的分片哈希表
 * @param dr
 *   Death row to free buffers to	// 回收超时或者重组错误的分片包
 * @param mb
 *   Incoming mbuf with IPv4 fragment.	// 分片报文
 * @param tms
 *   Fragment arrival timestamp.	// 分片报文时间戳
 * @param ip_hdr
 *   Pointer to the IPV4 header inside the fragment.	// 分片包的IP层头
 * @return
 *   Pointer to mbuf for reassembled packet, or NULL if:  // 如果重组完成返回重组后的第一个mbuf数据包地址，可以理解为链表头节点， 如果返回值为NULL则说明重组失败，或者重组未完成（重组数据包不全）
 *   - an error occurred.
 *   - not all fragments of the packet are collected yet.
 */
struct rte_mbuf * rte_ipv4_frag_reassemble_packet(struct rte_ip_frag_tbl *tbl,
		struct rte_ip_frag_death_row *dr,
		struct rte_mbuf *mb, uint64_t tms, struct rte_ipv4_hdr *ip_hdr);

rte_ipv6_frag_reassemble_packet功能类似，不在赘述。

二、分片包回收

每次调用rte_ipv4_frag_reassemble_packet或者rte_ipv6_frag_reassemble_packetc重组数据包时，会将death_row实参传入进去，对于重组过程中由于超时或者其他原因失败的数据包，回传入到death_row表中进行回收释放

调用位置为：

至此，可以获取到问题1和2的答案，

问题1解释

1）、超时时间是在哈希表创建时，参数frag_cycles决定的，而frag_cycles的值，在代码中也有体现

frag_cycles = (rte_get_tsc_hz() + MS_PER_S - 1) / MS_PER_S * max_flow_ttl;

2）、超时及重组错误分片包数据回收调用rte_ip_frag_free_death_row接口实现。

问题2解释

1）、对于重组后的数据包，返回的类似于链表的头指针，struct rte_mbuf接口中next指针表示Next segment of scattered packet.

问题3解释

最后一个问题可以在代码中找到对应的数值

/* ip_fragmentation defines */
#define RTE_LIBRTE_IP_FRAG_MAX_FRAG 4

即ip分片默认最大为4个，下面通过执行程序进行几个简单的测试。

3、执行程序

这里测试和验证重组消息，将一个数据包拆分为4片、5片，一台服务器启动ip_reassembly程序，然后另外一台服务器发送这些分片包。

1、测试重组分片包超时

通过控制发包速率，测试分片包超时会不会重组成功。

2、测试重组后分片包的格式

通过打印重组成功后的分片包内容，测试分片包重组后的格式。

3、测试最大重组分片包个数

通过控制发送分片包的个数，测试最大能够重组的分片包数量。

然后代码中添加打印参数，用于打印重组成功后数据包的内容信息，以此可以用来测试以上3个问题。

可以看出，如果重组失败或者重组未完成，流程不会到添加的打印代码段，因此可以这些打印即可判断慢速发送分片数据包导致分片包超时删除导致重组失败，又可以判断数据包分片过多（4个以上）导致数据包重组失败，又可以哦按段已经数据包重组后的格式是否符合预期。

启动程序命令

./ip_reassembly_app-static -l 1-2 --log-level 8 -n 4 -- -p 0x1

[root@LFTF dpdk-ip_reassembly]# ./ip_reassembly_app-static -l 1-2 --log-level 8 -n 4 -- -p 0x1
EAL: Detected 40 lcore(s)
EAL: Detected 2 NUMA nodes
EAL: Multi-process socket /var/run/dpdk/rte/mp_socket
EAL: Selected IOVA mode 'VA'
EAL: No available hugepages reported in hugepages-2048kB
EAL: Probing VFIO support...
EAL: VFIO support initialized
EAL: DPDK is running on a NUMA system, but is compiled without NUMA support.
EAL: This will have adverse consequences for performance and usability.
EAL: Please use --legacy-mem option, or recompile with NUMA support.
EAL:   using IOMMU type 1 (Type 1)
EAL: Ignore mapping IO port bar(2)
EAL: Probe PCI driver: net_ixgbe (8086:10fb) device: 0000:05:00.0 (socket 0)
EAL: Ignore mapping IO port bar(2)
EAL: Probe PCI driver: net_ixgbe (8086:10fb) device: 0000:05:00.1 (socket 0)
EAL: No legacy callbacks, legacy socket not created
IP_RSMBL: Creating LPM table on socket 0
IP_RSMBL: Creating LPM6 table on socket 0
USER1: rte_ip_frag_table_create: allocated of 33554560 bytes at socket 0
Initializing port 0 ... Port 0 modified RSS hash function based on hardware support,requested:0xa38c configured:0x8104
 Address:AC:F9:70:83:B6:63
txq=1,0,0 txq=2,1,0 

Skipping disabled port 1

IP_RSMBL: Socket 0: adding route 100.10.0.0/16 (port 0)
IP_RSMBL: Socket 0: adding route 100.20.0.0/16 (port 1)
IP_RSMBL: Socket 0: adding route 100.30.0.0/16 (port 2)
IP_RSMBL: Socket 0: adding route 100.40.0.0/16 (port 3)
IP_RSMBL: Socket 0: adding route 100.50.0.0/16 (port 4)
IP_RSMBL: Socket 0: adding route 100.60.0.0/16 (port 5)
IP_RSMBL: Socket 0: adding route 100.70.0.0/16 (port 6)
IP_RSMBL: Socket 0: adding route 100.80.0.0/16 (port 7)
IP_RSMBL: Socket 0: adding route 0101:0101:0101:0101:0101:0101:0101:0101/48 (port 0)
IP_RSMBL: Socket 0: adding route 0201:0101:0101:0101:0101:0101:0101:0101/48 (port 1)
IP_RSMBL: Socket 0: adding route 0301:0101:0101:0101:0101:0101:0101:0101/48 (port 2)
IP_RSMBL: Socket 0: adding route 0401:0101:0101:0101:0101:0101:0101:0101/48 (port 3)
IP_RSMBL: Socket 0: adding route 0501:0101:0101:0101:0101:0101:0101:0101/48 (port 4)
IP_RSMBL: Socket 0: adding route 0601:0101:0101:0101:0101:0101:0101:0101/48 (port 5)
IP_RSMBL: Socket 0: adding route 0701:0101:0101:0101:0101:0101:0101:0101/48 (port 6)
IP_RSMBL: Socket 0: adding route 0801:0101:0101:0101:0101:0101:0101:0101/48 (port 7)

Checking link status
done
Port 0 Link up at 10 Gbps FDX Autoneg
IP_RSMBL: lcore 2 has nothing to do
IP_RSMBL: entering main loop on lcore 1
IP_RSMBL:  -- lcoreid=1 portid=0

启动程序之后对端开始打包，加上打印代码之后，起初测试的时候发现了几个问题，仅能重组成功2个分片包，分片包超过2个的时候不能重组成功，后来摸索发现，当时没有考虑到超时问题，对端打包采用的命令如下：

[root@LF pcap]# tcpreplay -i enp2s -l 1 -p 4 ip_fragement.pcap

可以看出是一个一个的打包，因此导致处理解析第二个分片包是，第一个分片包超时删除了，导致重组失败。

后面我把打包命令改成如下全速发包：

[root@LF pcap]# tcpreplay -i enp2s -t -p 4 ip_fragement.pcap

这样就可以正常完成重组4个数据包了。但是后面又发现重组4个以上数据包时重组失败，这个时候查询代码发现了RTE_LIBRTE_IP_FRAG_MAX_FRAG值设置为4，然后把这个值改成8之后，再次发送5个分片的数据包仍然重组失败，此时认为可能是超时时间太短导致，因此将frag_cycles值改为frag_cycles *= 100之后再次编译代码仍然重组失败，再次查询RTE_LIBRTE_IP_FRAG_MAX_FRAG发现其它库函数也间接使用到了这个值，而且RTE_LIBRTE_IP_FRAG_MAX_FRAG值在rte_config.h定义，因此感觉需要重新编译DPDK才能解决这个问题，根据这个思路重新编译了RTE_LIBRTE_IP_FRAG_MAX_FRAG值为8的版本库，再次编译ip_reassembly样例程序，执行程序，对端打包，然后发现这次重组成功了。打印如下：

IPV6 Pkt_len = 1862

Data_len = 350
Data: [7a7ac0a8c80100000000003386dd6000000007101140fe800000000000000000000000000033fe8000000000000000000000000000010035003507105d98adadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadad]


Data_len = 296
Data: [adadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadad]


Data_len = 296
Data: [adadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadad]


Data_len = 296
Data: [adadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadad]


Data_len = 296
Data: [adadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadad]


Data_len = 296
Data: [adadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadad]


Data_len = 32
Data: [adadadadadadadadadadadadadadadadadadadadadadadadadadadadadadadad]

可以看到，这一次成功重组了7个分片包，至此ip_reassembly测试结束。

下面是我做的测试及推断出的结论：

测试一： 慢速发送4个分片数据包，查看是否重组成功（现象：重组失败。 结论：超时原因导致重组失败）

测试二： 修改代码中frag_cycles值，将超时时间扩大100倍，然后再根据测试一的条件测试（现象：重组成功。 结论：超时时间理应控制很短，这里仅作为测试）

测试三： 不修改修改代码中frag_cycles值，快速发送4个分片数据包，查看是否重组成功（现象：重组成功。 结论：不超时的前提下可以重组4个分片）

测试四： 修改代码中frag_cycles值， 将超时时间扩大100倍，快速发送5个数据包，查看是否重组成功（现象：重组失败。 结论：分片数量过多导致分片失败）

测试五： 修改代码中frag_cycles值， 将超时时间扩大100倍，修改RTE_LIBRTE_IP_FRAG_MAX_FRAG值为8，快速发送5个数据包，查看是否重组成功（现象：重组失败。 结论：分片数量过多导致分片失败，进修改头文件中RTE_LIBRTE_IP_FRAG_MAX_FRAG值不能改正重组数据包个数）

测试六: 修改RTE_LIBRTE_IP_FRAG_MAX_FRAG值为8，重新编译DPDK，修改样例代码中frag_cycles值， 将超时时间扩大100倍，然后再编译测试代码，慢速发送5个数据包，查看是否重组成功（现象：重组成功。 结论：分片数量过多导致分片失败，需要修改RTE_LIBRTE_IP_FRAG_MAX_FRAG值之后重新编译dpdk）

分片数据包下载链接：fragemetPcap

4、总结

实际项目中可能用不到IP组包功能，即使用到了实际上分片包的个数一般也就2个，很少情况下超过4个，文中的非正常测试也属于自己的胡思乱想，瞎搞一通！！