https://www.kernel.org/doc/html/v6.6/core-api/circular-buffers.html

循环缓冲区

作者

David Howells dhowells@redhat.com
Paul E. McKenney paulmck@linux.ibm.com

Linux提供了许多功能，可用于实现循环缓冲区。有两组这样的功能：

用于确定2的幂大小缓冲区信息的便利函数。
当缓冲区中的对象的生产者和消费者不想共享锁时，使用内存屏障。

要使用下面讨论的这些功能，只需要一个生产者和一个消费者。可以通过串行化来处理多个生产者，并通过串行化来处理多个消费者。

什么是循环缓冲区？

首先，什么是循环缓冲区？循环缓冲区是一个固定大小的缓冲区，其中有两个索引：

'头'索引 - 生产者将项目插入缓冲区的位置。
'尾'索引 - 消费者在缓冲区中找到下一个项目的位置。

通常情况下，当尾指针等于头指针时，缓冲区为空；当头指针比尾指针小1时，缓冲区为满。

当添加项目时，头索引会递增，而当移除项目时，尾索引会递增。尾索引不应超过头索引，并且当它们到达缓冲区末尾时，两个索引都应该被包装到0，从而允许无限量的数据流经过缓冲区。

通常，项目都是相同的单位大小，但并不严格要求使用下面的技术。如果要在缓冲区中包含多个项目或大小可变的项目，则可以逐渐增加索引，前提是两个索引都不会超过对方。但是，实施者必须小心，因为一个大于一个单位大小的区域可能会包装到缓冲区的末尾，并被分成两个段。

测量2的幂缓冲区

通常情况下，计算任意大小的循环缓冲区的占用量或剩余容量将是一个缓慢的操作，需要使用模数（除法）指令。但是，如果缓冲区的大小是2的幂，则可以使用更快的按位AND指令。

Linux提供了一组用于处理2的幂循环缓冲区的宏。可以通过以下方式使用这些宏：

#include <linux/circ_buf.h>

这些宏包括：

测量缓冲区中剩余的空间：
```
CIRC_SPACE(head_index, tail_index, buffer_size);
```
这将返回缓冲区中剩余的空间，可以插入项目。
测量缓冲区中最大的连续空间：
```
CIRC_SPACE_TO_END(head_index, tail_index, buffer_size);
```
这将返回缓冲区中最大的连续空间，可以立即插入项目，而无需回到缓冲区的开头。
测量缓冲区的占用量：
```
CIRC_CNT(head_index, tail_index, buffer_size);
```
这将返回当前占用缓冲区的项目数。
测量缓冲区中不会包装的占用量：
```
CIRC_CNT_TO_END(head_index, tail_index, buffer_size);
```
这将返回可以从缓冲区中提取的连续项目数，而无需回到缓冲区的开头。

这些宏中的每一个通常会返回一个介于0和buffer_size-1之间的值，但是：

CIRC_SPACE*() 用于生产者。对于生产者，它们将返回一个下限，因为生产者控制头索引，但是消费者可能仍在另一个CPU上耗尽缓冲区并移动尾索引。
对于消费者，它将显示一个上限，因为生产者可能正在忙于耗尽空间。
CIRC_CNT*() 用于消费者。对于消费者，它们将返回一个下限，因为消费者控制尾索引，但是生产者可能仍在另一个CPU上填充缓冲区并移动头索引。
对于生产者，它将显示一个上限，因为消费者可能正在忙于清空缓冲区。
对于第三方，无法保证生产者和消费者对索引的写入的可见性顺序，因为它们是独立的，可能在不同的CPU上进行 - 因此，在这种情况下的结果只是一个猜测，甚至可能是负数。

使用内存屏障与循环缓冲区

通过在循环缓冲区中使用内存屏障，可以避免以下需求：

使用单个锁来管理对缓冲区两端的访问，从而允许缓冲区同时被填充和清空。
使用原子计数操作。

这有两个方面：填充缓冲区的生产者和清空缓冲区的消费者。在任何时候，只应该有一件事在填充缓冲区，而只应该有一件事在清空缓冲区，但是两个方面可以同时操作。

生产者

生产者将类似于以下内容：

spin_lock(&producer_lock);

unsigned long head = buffer->head;
/* spin_unlock() 和下一个 spin_lock() 提供所需的排序。 */
unsigned long tail = READ_ONCE(buffer->tail);

if (CIRC_SPACE(head, tail, buffer->size) >= 1) {
        /* 将一个项目插入缓冲区 */
        struct item *item = buffer[head];

        produce_item(item);

        smp_store_release(buffer->head,
                          (head + 1) & (buffer->size - 1));

        /* wake_up() 确保在唤醒任何人之前，头部已经提交 */
        wake_up(consumer);
}

spin_unlock(&producer_lock);

这将指示CPU必须在将新项目的内容写入之前，将头索引使其对消费者可用，并且指示CPU必须在唤醒消费者之前，必须写入修订后的头索引。

请注意，wake_up() 不会保证任何类型的屏障，除非实际唤醒了某些内容。因此，我们不能依赖它进行排序。但是，数组中始终会留下一个元素为空。因此，生产者必须在可能损坏消费者当前正在读取的元素之前产生两个元素。因此，在连续调用消费者之间的解锁-锁对提供了所需的排序，用于指示消费者已经空出了给定元素的索引的读取，以及生产者对该相同元素的写入。

消费者

消费者将类似于以下内容：

spin_lock(&consumer_lock);

/* 在读取该索引的内容之前读取索引。 */
unsigned long head = smp_load_acquire(buffer->head);
unsigned long tail = buffer->tail;

if (CIRC_CNT(head, tail, buffer->size) >= 1) {

        /* 从缓冲区中提取一个项目 */
        struct item *item = buffer[tail];

        consume_item(item);

        /* 在增加尾部之前完成读取描述符。 */
        smp_store_release(buffer->tail,
                          (tail + 1) & (buffer->size - 1));
}

spin_unlock(&consumer_lock);

这将指示CPU在读取新项目之前，必须确保索引是最新的，然后必须确保CPU在写入新尾指针之前已经完成了读取项目，这将擦除该项目。

请注意，使用READ_ONCE() 和smp_load_acquire() 读取对立索引。这可以防止编译器丢弃并重新加载其缓存的值。如果可以确保对立索引只会被使用一次，则不严格需要这样做。此外，smp_load_acquire() 还强制CPU对后续内存引用进行排序。类似地，两个算法中都使用了smp_store_release() 来写入线程的索引。这说明了我们正在写入可以同时被读取的内容，防止编译器破坏存储，并强制对先前的访问进行排序。