一、什么是Ring buffer？
Ring buffer是网络收发包流程中的缓冲区，网络收包流程可以简单概括为如下几个环节：
- 网卡收到数据之后，通过DMA将数据写入到内存的ring buffer中。
- 数据写入完成后，由网卡发起一个硬中断，CPU收到中断后，执行相应的中断处理函数，将网卡驱动中的napi_struct传到的poll_queue中，并置位软中断标志位。
- 在ksoftirqd线程中通过查询软中断标志位，识别软中断类型，调用对应的注册函数，将sk_buff从Ring Buff中取下，并申请新的sk_buff中挂在Ring Buff上。
- 将取下sk_buff交由上层协议栈处理，并在协议栈处理完后释放内存空间。
二、Ring buffer是如何实现的
为防止混淆，先声明，上一节图片中的Ring buffer，在本章中节中称为ring，而ring buffer在本节中特指图片中存放packet的内存空间，之所以这样表示，是因为驱动程序中的变量命名。
以openeuler-22.04-sp2源码中自带的e1000网卡驱动为例，网卡的每个队列，都有一个rx ring和一个tx ring组成。本篇文章以rx ring为例，每个rx ring都由结构体一个e1000_rx_ring进行管理，该结构体在e1000_alloc_queues函数中申请内存空间，调用链为
--e1000_probe
--e1000_sw_init
--e1000_alloc_queues
该结构体主要维护以下rx ring相关信息
struct e1000_rx_ring{
void *desc; /*指向ring_desc数组的指针
dma_addr_t dma; /*ring_desc数组的dma地址
unsigned int count /*ring_desc数组的元素个数
unsigned int next_to_use; /*下一个用于分配dma内存地址的rx_buffer
unsigned int next_to_clean; /*下一个用于摘取数据包的内存地址
struct e1000_rx_buffer *buffer_info; /*指向rx_buffer相关信息数组的指针
…
}
ring_desc数组主要用硬件和驱动程序负责填写并维护，buffer_info则由驱动程序维护，二者的单个元素保存的都是和数据包相关的信息，如数据包在内存中的地址，而不是数据包本身。
Rx ring中ring_desc数组和rx_buffer数组的内存空间在e1000_open函数中分配，调用链为
--e1000_open
--e1000_setup_all_rx_resources
--e1000_setup_rx_resources
在e1000_setup_rx_resources中，申请了分配了两个内存空间
static int e1000_setup_rx_resources(struct e1000_adapter *adapter, struct e1000_rx_ring *rxdr){
…
size = sizeof(struct e1000_rx_buffer) * rxdr->count;
rxdr->buffer_info = vzalloc(size);
…
desc_len = sizeof(struct e1000_rx_desc);
rxdr->size = rxdr->count * desc_len;
rxdr->size = ALIGN(rxdr->size, 4096);
rxdr->desc = dma_alloc_coherent(&pdev->dev, rxdr->size, &rxdr->dma, GFP_KERNEL);
…
}
根据rxdr->count，分配相应数量的e1000_rx_buffer和e1000_rx_desc结构体
struct e1000_rx_buffer {
union {
struct page *page; /* jumbo: alloc_page */
u8 *data; /* else, netdev_alloc_frag */
} rxbuf;
dma_addr_t dma;
}
用 dma_alloc_coherent 分配的地址，当cpu读取或写入段地址禁止使用cache，为了避免dma将io设备的数据写入内存时，cpu在此时读写映射到这段内存的cache，导致读取旧数据到cache中，或是将cache中的旧数据到内存里，这部分内容涉及一致性dma和流式dma，不做继续深入了，如有兴趣可以根据关键字检索相关文章。
分配内存的e1000_rx_desc数据结构如下，这段内存完全由网卡通过dma填写，保存了网卡dma写入的数据包在内存中的地址，长度，校验值，该描述符的状态等信息。
struct e1000_rx_desc {
__le64 buffer_addr; /* Address of the descriptor's data buffer */
__le16 length; /* Length of data DMAed into data buffer */
__le16 csum; /* Packet checksum */
u8 status; /* Descriptor status */
u8 errors; /* Descriptor Errors */
__le16 special;
};
至此只是申请了由dma填写的rx_buffer描述符的内存空间，和保存buffer地址的e1000_rx_buffer结构体内存空间
随后在e1000_open的e1000_configure_rx函数中，注册e1000_clean_rx_irq和e1000_alloc_rx_buffers函数，调用链如下
--e1000_open
--e1000_configure
--e1000_configure_rx
注册内容如下
rdlen = adapter->rx_ring[0].count * sizeof(struct e1000_rx_desc);
adapter->clean_rx = e1000_clean_rx_irq;
adapter->alloc_rx_buf = e1000_alloc_rx_buffers;
且将申请的e1000_rx_desc内存地址配置到网卡的寄存器中
rdlen = adapter->rx_ring[0].count *sizeof(struct e1000_rx_desc);
rdba = adapter->rx_ring[0].dma;
ew32(RDLEN, rdlen);
ew32(RDBAH, (rdba >> 32));
ew32(RDBAL, (rdba & 0x00000000ffffffffULL));
e1000_configure_rx返回后，继续在e1000_configure函数中调用刚注册adapter->alloc_rx_buf函数分配为rx_buffer内存空间
for (i = 0; i < adapter->num_rx_queues; i++) {
struct e1000_rx_ring *ring = &adapter->rx_ring[i];
adapter->alloc_rx_buf(adapter, ring,
E1000_DESC_UNUSED(ring));
}
对于每个队列的每个rx_ring，都会分配和描述符个数相同的rx_buffer内存空间，每个rx_buffer内存空间的大小由网卡的rx_buffer_len参数指定，该值默认为最大的vlan帧长度1522字节，初始化过程中会根据网卡型号进行更改。实际分配过程中，还会加上NET_SKB_PAD和结构体skb_shared_info的长度，并进行内存对齐，这部分涉及内核处理skb结构体的相关知识，不做继续深入了
i = rx_ring->next_to_use
…
data = e1000_alloc_frag(adapter);
buffer_info->dma = dma_map_single(&pdev->dev,
data,
adapter->rx_buffer_len,
DMA_FROM_DEVICE);
rx_desc = E1000_RX_DESC(*rx_ring, i);
rx_desc->buffer_addr = cpu_to_le64(buffer_info->dma);
…
if (unlikely(++i == rx_ring->count))
i = 0;
rx_ring->next_to_use = i
writel(i, hw->hw_addr + rx_ring->rdt)
每次申请完rx_buffer内存空间后，会对这片内存区域和网络设备的fifo队列进行dma映射，并在buffer_info->dma和rx_desc->buffer_addr中保存dma映射的物理地址。此时rx_ring->next_to_use指向环形队列的队尾，表示网络设备传输时的最多只能传输到这个rx_buffer的前一个，这个rx_buffer还没有申请内存空间，或是还有数据包没有取出，不能使用。随后调用writel将该值写入到网络设备的寄存器中。
前面的部分大部分内容都是在初始化ring buffer。总结下，在初始化流程中我们先后完成了以下操作
1、为每个网卡队列定义了e1000_rx_ring结构体变量，该变量包括了rx_ring的基本信息，还包括了指向e1000_rx_desc和e1000_rx_buffer数组的指针。
2、为rx_ring每个rx_buffer申请了对应的e1000_rx_desc和e1000_rx_buffer结构体的内存空间，并把e1000_rx_desc数组的物理地址写入网卡的寄存器中。这两个结构体都仅存放数据包的相关信息，如指向存放数据包内容内存地址的指针。
3、调用e1000_configure_rx为每个rx_buffer申请存放数据包的内存空间，并将该内存空间的地址写入e1000_rx_desc和e1000_rx_buffer结构体中。
前面注册的e1000_clean_rx_irq才是操作rx_buff完成收包功能的函数。在e1000_clean_rx_irq函数被调用前，网络设备会通过e1000_rx_desc得知存放数据包的地址，并通过dma，将数据包传输到该地址上，传输完成先后触发硬中断和软中断。
e1000_clean_rx_irq在e1000_clean函数中调用，e1000_clean函数在上面提到的e1000_open中被注册为网络设备的poll函数，poll函数将会在软中断内被napi_poll调用。
netif_napi_add(netdev, &adapter->napi, e1000_clean, 64);
这部分内容涉及linux的napi接口，可以自行检索，我们的重点放在e1000_clean_rx_irq的功能实现上。
首先从rx ring取出序号为next to clean的rx_desc和rx_buffer_info元素的地址
i = rx_ring->next_to_clean;
rx_desc = E1000_RX_DESC(*rx_ring, i);
buffer_info = &rx_ring->buffer_info[i];
然后检查该序号的rx_buffer是否已经完成的dma传输
while (rx_desc->status & E1000_RXD_STAT_DD)
对于小数据包，采用直接拷贝的方式，新建一个buffer长度为数据包大小的skb，将内容拷贝到新申请的内存空间中，具体通过在e1000_copybreak函数中调用e1000_alloc_rx_skb和skb_put_data完成，同时，由于数据被拷贝出去了，所以原来申请的dma内存空间可以重复使用，因此不需要释放，也不需要将指向该地址的指针赋为空值。
skb = e1000_alloc_rx_skb(adapter, length);
skb_put_data(skb, data, length);
如果是较大的数据包，则将rx_buffer的地址直接添加到skb，省去了重新申请内存空间并拷贝数据的环节，但代价是，如果数据包没有填满整个帧，则多余的空间也被留在skb中，造成了内存的浪费。
unsigned int frag_len = e1000_frag_len(adapter);
skb = build_skb(data - E1000_HEADROOM, frag_len);
skb_reserve(skb, E1000_HEADROOM);
对于第二种情况，将rx ring中的数据取下并挂在skb上后，取消这段数据对应内存空间的dma映射，然后清空buffer_info数组中的dma地址
dma_unmap_single(&pdev->dev, buffer_info->dma,
adapter->rx_buffer_len,
DMA_FROM_DEVICE);
buffer_info->dma = 0;
buffer_info->rxbuf.data = NULL;
将这rx buffer的数据取下后，检查这个rx buffer的数据是否是单个数据包，如果是一个数据包分为多个帧发送，装载在多个rx buffer中，则需要将多个rx buffer的数据整合为一个包，但是在e1000的驱动中只是单纯的丢弃了这类数据包。
/* All receives must fit into a single buffer */
netdev_dbg(netdev, "Receive packet consumed multiple buffers\n");
dev_kfree_skb(skb);
if (status & E1000_RXD_STAT_EOP)
adapter->discarding = false;
goto next_desc;
最后对skb数据包中的内容作校验，然后将skb上传到上层协议栈中，调用栈为
--e1000_receive_skb(adapter, status, rx_desc->special, skb);
--napi_gro_receive(&adapter->napi, skb);
由上层协议栈处理完数据后，释放skb的内存空间
将足够的数据包传输到上层协议栈后，准备退出这次软中断处理函数e1000_clean_rx_irq，在退出前最后一步，统计rx ring中已经交付上层，但还没有重新申请内存空间的rx buffer个数，并调用alloc_rx_buf一次性为这些rx buffer申请内存空间
cleaned_count = E1000_DESC_UNUSED(rx_ring);
if (cleaned_count)
adapter->alloc_rx_buf(adapter, rx_ring, cleaned_count);
至此，rx ring完成了一轮收包中的全部工作。
希望看到这里，能让您对rx ring主要相关的数据结构，操作函数和工作原理有更深刻的理解。
三、如何更改Ring buffer相关参数进行调优？
使用指令 ethtool -G eth0 rx/tx 可以改变Ring buffer中buffer的数量，该值也等同于rx_desc和buffer_info数组中元素的数量，当Rng buffer空间不足时，多余的数据包会被丢弃。但过大的Ring buffer空间会造成内存浪费，同时会降低cpu cache预取的命中率。
使用指令 ethtool -L eth0 combined 可以改变网卡队列数，即Ring buffer的Rx ring的数量，该值也等同于Tx ring的数量。
使用指令 ethtool –config-ntuple eth0 可以改变网络数据包的哈希规则，确定收到的数据包被发送到哪个网卡队列上，即哪个Rx ring上。
上述的调优都需要根据具体场景选择合适的值，简单的调整为最大值或最小值，可能会起到反效果。
点赞
收藏
回复
分享
举报
一、什么是Ring buffer？
Ring buffer是网络收发包流程中的缓冲区，网络收包流程可以简单概括为如下几个环节：
二、Ring buffer是如何实现的
为防止混淆，先声明，上一节图片中的Ring buffer，在本章中节中称为ring，而ring buffer在本节中特指图片中存放packet的内存空间，之所以这样表示，是因为驱动程序中的变量命名。
以openeuler-22.04-sp2源码中自带的e1000网卡驱动为例，网卡的每个队列，都有一个rx ring和一个tx ring组成。本篇文章以rx ring为例，每个rx ring都由结构体一个e1000_rx_ring进行管理，该结构体在e1000_alloc_queues函数中申请内存空间，调用链为
--e1000_probe
--e1000_sw_init
--e1000_alloc_queues
该结构体主要维护以下rx ring相关信息
struct e1000_rx_ring{
void *desc; /*指向ring_desc数组的指针
dma_addr_t dma; /*ring_desc数组的dma地址
unsigned int count /*ring_desc数组的元素个数
unsigned int next_to_use; /*下一个用于分配dma内存地址的rx_buffer
unsigned int next_to_clean; /*下一个用于摘取数据包的内存地址
struct e1000_rx_buffer *buffer_info; /*指向rx_buffer相关信息数组的指针
…
}
ring_desc数组主要用硬件和驱动程序负责填写并维护，buffer_info则由驱动程序维护，二者的单个元素保存的都是和数据包相关的信息，如数据包在内存中的地址，而不是数据包本身。
Rx ring中ring_desc数组和rx_buffer数组的内存空间在e1000_open函数中分配，调用链为
--e1000_open
--e1000_setup_all_rx_resources
--e1000_setup_rx_resources
在e1000_setup_rx_resources中，申请了分配了两个内存空间
static int e1000_setup_rx_resources(struct e1000_adapter *adapter, struct e1000_rx_ring *rxdr){
…
size = sizeof(struct e1000_rx_buffer) * rxdr->count;
rxdr->buffer_info = vzalloc(size);
…
desc_len = sizeof(struct e1000_rx_desc);
rxdr->size = rxdr->count * desc_len;
rxdr->size = ALIGN(rxdr->size, 4096);
rxdr->desc = dma_alloc_coherent(&pdev->dev, rxdr->size, &rxdr->dma, GFP_KERNEL);
…
}
根据rxdr->count，分配相应数量的e1000_rx_buffer和e1000_rx_desc结构体
struct e1000_rx_buffer {
union {
struct page *page; /* jumbo: alloc_page */
u8 *data; /* else, netdev_alloc_frag */
} rxbuf;
dma_addr_t dma;
}
用 dma_alloc_coherent 分配的地址，当cpu读取或写入段地址禁止使用cache，为了避免dma将io设备的数据写入内存时，cpu在此时读写映射到这段内存的cache，导致读取旧数据到cache中，或是将cache中的旧数据到内存里，这部分内容涉及一致性dma和流式dma，不做继续深入了，如有兴趣可以根据关键字检索相关文章。
分配内存的e1000_rx_desc数据结构如下，这段内存完全由网卡通过dma填写，保存了网卡dma写入的数据包在内存中的地址，长度，校验值，该描述符的状态等信息。
struct e1000_rx_desc {
__le64 buffer_addr; /* Address of the descriptor's data buffer */
__le16 length; /* Length of data DMAed into data buffer */
__le16 csum; /* Packet checksum */
u8 status; /* Descriptor status */
u8 errors; /* Descriptor Errors */
__le16 special;
};
至此只是申请了由dma填写的rx_buffer描述符的内存空间，和保存buffer地址的e1000_rx_buffer结构体内存空间
随后在e1000_open的e1000_configure_rx函数中，注册e1000_clean_rx_irq和e1000_alloc_rx_buffers函数，调用链如下
--e1000_open
--e1000_configure
--e1000_configure_rx
注册内容如下
rdlen = adapter->rx_ring[0].count * sizeof(struct e1000_rx_desc);
adapter->clean_rx = e1000_clean_rx_irq;
adapter->alloc_rx_buf = e1000_alloc_rx_buffers;
且将申请的e1000_rx_desc内存地址配置到网卡的寄存器中
rdlen = adapter->rx_ring[0].count *sizeof(struct e1000_rx_desc);
rdba = adapter->rx_ring[0].dma;
ew32(RDLEN, rdlen);
ew32(RDBAH, (rdba >> 32));
ew32(RDBAL, (rdba & 0x00000000ffffffffULL));
e1000_configure_rx返回后，继续在e1000_configure函数中调用刚注册adapter->alloc_rx_buf函数分配为rx_buffer内存空间
for (i = 0; i < adapter->num_rx_queues; i++) {
struct e1000_rx_ring *ring = &adapter->rx_ring[i];
adapter->alloc_rx_buf(adapter, ring,
E1000_DESC_UNUSED(ring));
}
对于每个队列的每个rx_ring，都会分配和描述符个数相同的rx_buffer内存空间，每个rx_buffer内存空间的大小由网卡的rx_buffer_len参数指定，该值默认为最大的vlan帧长度1522字节，初始化过程中会根据网卡型号进行更改。实际分配过程中，还会加上NET_SKB_PAD和结构体skb_shared_info的长度，并进行内存对齐，这部分涉及内核处理skb结构体的相关知识，不做继续深入了
i = rx_ring->next_to_use
…
data = e1000_alloc_frag(adapter);
buffer_info->dma = dma_map_single(&pdev->dev,
data,
adapter->rx_buffer_len,
DMA_FROM_DEVICE);
rx_desc = E1000_RX_DESC(*rx_ring, i);
rx_desc->buffer_addr = cpu_to_le64(buffer_info->dma);
…
if (unlikely(++i == rx_ring->count))
i = 0;
rx_ring->next_to_use = i
writel(i, hw->hw_addr + rx_ring->rdt)
每次申请完rx_buffer内存空间后，会对这片内存区域和网络设备的fifo队列进行dma映射，并在buffer_info->dma和rx_desc->buffer_addr中保存dma映射的物理地址。此时rx_ring->next_to_use指向环形队列的队尾，表示网络设备传输时的最多只能传输到这个rx_buffer的前一个，这个rx_buffer还没有申请内存空间，或是还有数据包没有取出，不能使用。随后调用writel将该值写入到网络设备的寄存器中。
前面的部分大部分内容都是在初始化ring buffer。总结下，在初始化流程中我们先后完成了以下操作
1、为每个网卡队列定义了e1000_rx_ring结构体变量，该变量包括了rx_ring的基本信息，还包括了指向e1000_rx_desc和e1000_rx_buffer数组的指针。
2、为rx_ring每个rx_buffer申请了对应的e1000_rx_desc和e1000_rx_buffer结构体的内存空间，并把e1000_rx_desc数组的物理地址写入网卡的寄存器中。这两个结构体都仅存放数据包的相关信息，如指向存放数据包内容内存地址的指针。
3、调用e1000_configure_rx为每个rx_buffer申请存放数据包的内存空间，并将该内存空间的地址写入e1000_rx_desc和e1000_rx_buffer结构体中。
前面注册的e1000_clean_rx_irq才是操作rx_buff完成收包功能的函数。在e1000_clean_rx_irq函数被调用前，网络设备会通过e1000_rx_desc得知存放数据包的地址，并通过dma，将数据包传输到该地址上，传输完成先后触发硬中断和软中断。
e1000_clean_rx_irq在e1000_clean函数中调用，e1000_clean函数在上面提到的e1000_open中被注册为网络设备的poll函数，poll函数将会在软中断内被napi_poll调用。
netif_napi_add(netdev, &adapter->napi, e1000_clean, 64);
这部分内容涉及linux的napi接口，可以自行检索，我们的重点放在e1000_clean_rx_irq的功能实现上。
首先从rx ring取出序号为next to clean的rx_desc和rx_buffer_info元素的地址
i = rx_ring->next_to_clean;
rx_desc = E1000_RX_DESC(*rx_ring, i);
buffer_info = &rx_ring->buffer_info[i];
然后检查该序号的rx_buffer是否已经完成的dma传输
while (rx_desc->status & E1000_RXD_STAT_DD)
对于小数据包，采用直接拷贝的方式，新建一个buffer长度为数据包大小的skb，将内容拷贝到新申请的内存空间中，具体通过在e1000_copybreak函数中调用e1000_alloc_rx_skb和skb_put_data完成，同时，由于数据被拷贝出去了，所以原来申请的dma内存空间可以重复使用，因此不需要释放，也不需要将指向该地址的指针赋为空值。
skb = e1000_alloc_rx_skb(adapter, length);
skb_put_data(skb, data, length);
如果是较大的数据包，则将rx_buffer的地址直接添加到skb，省去了重新申请内存空间并拷贝数据的环节，但代价是，如果数据包没有填满整个帧，则多余的空间也被留在skb中，造成了内存的浪费。
unsigned int frag_len = e1000_frag_len(adapter);
skb = build_skb(data - E1000_HEADROOM, frag_len);
skb_reserve(skb, E1000_HEADROOM);
对于第二种情况，将rx ring中的数据取下并挂在skb上后，取消这段数据对应内存空间的dma映射，然后清空buffer_info数组中的dma地址
dma_unmap_single(&pdev->dev, buffer_info->dma,
adapter->rx_buffer_len,
DMA_FROM_DEVICE);
buffer_info->dma = 0;
buffer_info->rxbuf.data = NULL;
将这rx buffer的数据取下后，检查这个rx buffer的数据是否是单个数据包，如果是一个数据包分为多个帧发送，装载在多个rx buffer中，则需要将多个rx buffer的数据整合为一个包，但是在e1000的驱动中只是单纯的丢弃了这类数据包。
/* All receives must fit into a single buffer */
netdev_dbg(netdev, "Receive packet consumed multiple buffers\n");
dev_kfree_skb(skb);
if (status & E1000_RXD_STAT_EOP)
adapter->discarding = false;
goto next_desc;
最后对skb数据包中的内容作校验，然后将skb上传到上层协议栈中，调用栈为
--e1000_receive_skb(adapter, status, rx_desc->special, skb);
--napi_gro_receive(&adapter->napi, skb);
由上层协议栈处理完数据后，释放skb的内存空间
将足够的数据包传输到上层协议栈后，准备退出这次软中断处理函数e1000_clean_rx_irq，在退出前最后一步，统计rx ring中已经交付上层，但还没有重新申请内存空间的rx buffer个数，并调用alloc_rx_buf一次性为这些rx buffer申请内存空间
cleaned_count = E1000_DESC_UNUSED(rx_ring);
if (cleaned_count)
adapter->alloc_rx_buf(adapter, rx_ring, cleaned_count);
至此，rx ring完成了一轮收包中的全部工作。
希望看到这里，能让您对rx ring主要相关的数据结构，操作函数和工作原理有更深刻的理解。
三、如何更改Ring buffer相关参数进行调优？
使用指令 ethtool -G eth0 rx/tx 可以改变Ring buffer中buffer的数量，该值也等同于rx_desc和buffer_info数组中元素的数量，当Rng buffer空间不足时，多余的数据包会被丢弃。但过大的Ring buffer空间会造成内存浪费，同时会降低cpu cache预取的命中率。
使用指令 ethtool -L eth0 combined 可以改变网卡队列数，即Ring buffer的Rx ring的数量，该值也等同于Tx ring的数量。
使用指令 ethtool –config-ntuple eth0 可以改变网络数据包的哈希规则，确定收到的数据包被发送到哪个网卡队列上，即哪个Rx ring上。
上述的调优都需要根据具体场景选择合适的值，简单的调整为最大值或最小值，可能会起到反效果。