# Linux 内核循环队列与内存屏障

笔者将通过本文，给混沌道友，展示内核对于 smp 屏障的使用技巧，请务必看过混沌专题之并发原理才看本文，不然会成为蒙B中的战斗机。

Linux 提供了许多用于实现循环缓冲区的函数。其中有两组这样的函数：

1、用于方便检测缓冲区大小是否为 2 的倍数的函数

2、缓冲区中对象的生产者和消费者不希望使用共享锁时使用的内存屏障

如下文所述，为了方便描述这些函数，样例中只包含一个生产者和一个消费者。我们可以通过序列化多个生产者和序列化多个消费者，来支持多生产者与多消费者的场景。

# 什么是循环缓冲区

循环缓冲区是一个大小固定的循环队列，包含如下两个指针变量：

1、head 指针（head index）：该下标用于生产者向缓冲区中插入数据

2、tail 指针（tail index）：该下标用于消费者从缓冲区中取出数据

此时：

1、当 tail 指针等于 head 指针时，缓冲区为空状态

2、当 head 指针等于 tail 指针 - 1 时，缓冲区为满状态

读者可以这么理解：初始化时 tail == head，生产者每放一个 head + 1，此时 tail 不变， head 向前移动，由于大小固定且循环，此时 head 终究会达到：head + 1 == tail 此时满状态。

循环缓冲区中，当生产者每次向其中添加数据，那么 head 指针递增，当数据被消费者移出时 tail 指针递减。tail 指针将从不会越过 head 指针，并且当两个指针达到缓冲区的末尾时，将其初始化为 0，此时正是实现环形的关键，正是由于此特性，我们称：循环缓冲区中可以保存无限数量的数据（消费者要不断消费！）。

通常情况下，放入循环缓冲区中的数据包含相同大小的数据，但这并不是严格要求的，我们可以使用如下技术，使得循环缓冲区可以包含不同大小的数据：

1、如果在缓冲区添加多个数据或可变大小的数据，head 指针可以递增超过 1，消费者亦是如此，但必须满足环形缓冲区的约束：两个指针不会互相跨越对方。实现这种可以包含不同大小数据的缓冲区，需要特别注意：存入的数据可能会因为到达缓冲区的末尾，从而被切割为两段（一段数据在缓冲区的尾部，一段数据在缓冲区的头部）。

# 缓冲区的大小为 2 的倍数

总所周知，2 的倍数是计算机中最喜欢使用数字，为什么呢？因为方便计算。比如这里的循环缓冲区中的大小亦是如此。考虑一下：如果我不使用 2 的倍数来作为缓冲区的大小，此时，当我需要计算下标时会怎么做？很明显，取余，此时需要 CPU 调用除法运算来求余数，这相对于使用与运算的操作慢得太多了，于是我们将缓冲区的大小设置为 2的倍数后，我们直接用 index 与 size - 1 执行与运算即可。

Linux 内核提供几组宏定义来支持循环缓冲区的 2的倍数的运算。我们可以引入 #include <linux/circ_buf.h> 头文件来使用它：

1、检测循环缓冲区中可以插入的数据个数：CIRC_SPACE(head_index, tail_index, buffer_size);

2、检测循环缓冲区中最大可以插入的连续空间数据个数（也即数据不会被切割成两部分：一部分在尾部，一部分在头部的空间）：CIRC_SPACE_TO_END(head_index, tail_index, buffer_size);

3、检测循环缓冲区中包含的数据个数：CIRC_CNT(head_index, tail_index, buffer_size);

4、检测循环缓冲区中包含不需要回退到缓冲区首部的数据个数：CIRC_CNT_TO_END(head_index, tail_index, buffer_size);

上述的每个宏定义都将返回一个介于 0 到缓冲区大小 - 1 的值，并且：

1、CIRC_SPACE 宏通常用于生产者进程。生产者可以通过该宏来获取一个操作 head index 指针的下界，因为在生产者进程操作缓冲区时，消费者可能在另一个CPU中执行消费动作不断推进 tail index

2、CIRC_CNT 宏通常用于消费者进程。同理，消费者可以通过该宏来获取一个操作 tail index 指针的下界，因为在消费者进程操作缓冲区时，生产者进程可能在另一个CPU中执行消费动作不断推进 head index

3、对于使用这些宏定义的其他进程来说，生产者和消费者对 tail index 和 head index 的写入顺序无法保证可见，因为它们是相互独立的，并且可能在不同的 CPU 上执行，因此，在这种情况下的这些宏定义的返回结果只是猜测值，甚至可能返回负数。

# 循环缓冲区中使用内存屏障

我们可以在循环缓冲区中使用内存屏障从而避免：

1、避免使用锁来保护循环缓冲区

2、改用原子性操作来代替锁

生产者不断向缓冲区中填充数据，消费中不断从缓冲区中获取数据消费。在同一时刻，只允许有一个生产者填充数据，在同一时刻也只允许有一个消费者消费数据，但是允许这一个生产者与这个消费者并行操作。

# 生产者样例代码

在代码中，我们做如下操作：

1、生产者和消费者各自一把自旋锁，此时保证同一时间只存在一个生产者和消费者，他们可以并行

2、使用 CAS 操作

spin_lock(&producer_lock); // 自旋锁保证同一时间，只存在一个生产者

unsigned long head = buffer->head;
unsigned long tail = ACCESS_ONCE(buffer->tail); // 使用该宏保证 buffer->tail 中的值从内存中读取最新值（编译器不使用寄存器缓存，详见 混沌专题之并发原理 描述） 

if (CIRC_SPACE(head, tail, buffer->size) >= 1) { // 缓冲区中存在空闲位置
    struct item *item = buffer[head]; // 向 head 指针处，写入数据
    produce_item(item);  // 填充该指针中的其他数据
    smp_wmb(); // 写屏障 保证 head 指针 提交到内存时，先让写入的数据可见
    buffer->head = (head + 1) & (buffer->size - 1); // 递增 head 指针
    wake_up(consumer); // 唤醒生产线程，该函数会保证 head 指针 对其他 CPU 可见
}

spin_unlock(&producer_lock);

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 消费者样例代码

该代码同生产者不再详述。

spin_lock(&consumer_lock); // 自旋锁保证同一时间，只存在一个消费者
unsigned long head = ACCESS_ONCE(buffer->head);
unsigned long tail = buffer->tail;

if (CIRC_CNT(head, tail, buffer->size) >= 1) { // 缓冲区中存在未消费数据
    smp_read_barrier_depends(); // 数据依赖屏障，保证读取tail指针前，一定能读取指针下标处的数据最新值
    struct item *item = buffer[tail]; // 从缓冲区中获取数据
    consume_item(item); // 消费数据
    smp_mb(); // 写屏障，保证 tail 指针修改前，数据已经被消费清空
    buffer->tail = (tail + 1) & (buffer->size - 1);
}

spin_unlock(&consumer_lock);

1
2
3
4
5
6
7
8
9
10
11
12
13

← Linux kill 信号原理 Linux Futex 快速用户空间互斥体原理 →