一文讲解，Linux内核——Memory Barrier（内存屏障）

本文例子均在 Linux（g++）下验证通过，CPU 为 X86-64 处理器架构。所有罗列的 Linux 内核代码也均在（或只在）X86-64 下有效。

本文首先通过范例（以及内核代码）来解释 Memory Barrier，然后介绍一个利用 Memory Barrier 实现的无锁环形缓冲区。

Memory Barrier 简介

程序在运行时内存实际的访问顺序和程序代码编写的访问顺序不一定一致，这就是内存乱序访问。内存乱序访问行为出现的理由是为了提升程序运行时的性能。内存乱序访问主要发生在两个阶段：

编译时，编译器优化导致内存乱序访问（指令重排）
运行时，多 CPU 间交互引起内存乱序访问

Memory Barrier 能够让 CPU 或编译器在内存访问上有序。一个 Memory Barrier 之前的内存访问操作必定先于其之后的完成。Memory Barrier 包括两类：

编译器Memory Barrier
CPU Memory Barrier

很多时候，编译器和 CPU 引起内存乱序访问不会带来什么问题，但一些特殊情况下，程序逻辑的正确性依赖于内存访问顺序，这时候内存乱序访问会带来逻辑上的错误，例如：

// thread 1
while (!ok);
do(x);
// thread 2
x = 42;
ok = 1;

此段代码中，ok 初始化为 0，线程 1 等待 ok 被设置为 1 后执行 do 函数。假如说，线程 2 对内存的写操作乱序执行，也就是 x 赋值后于 ok 赋值完成，那么 do 函数接受的实参就很可能出乎程序员的意料，不为 42。

【文章福利】小编推荐自己的Linux内核技术交流群：【977878001】整理一些个人觉得比较好得学习书籍、视频资料共享在群文件里面，有需要的可以自行添加哦！！！前100进群领取，额外赠送一份价值699的内核资料包（含视频教程、电子书、实战项目及代码）

内核资料直通车：Linux内核源码技术学习路线+视频教程代码资料https://link.zhihu.com/?target=https%3A//docs.qq.com/doc/DUGZVQk1qWVBHTEl3

学习直通车：Linux内核源码/内存调优/文件系统/进程管理/设备驱动/网络协议栈-学习视频教程-腾讯课堂是不是学完操作系统原理后觉得纸上谈兵不过瘾？是不是面对浩若烟海的Linux内核源代码迷失在代码的海洋里不知所措？这门课可以带您用理论结合实践的方法一步一步抓住Linux内核最核心的部分代码，理解Linux操作系统运行的基本过程及涉及的核心机制。https://ke.qq.com/course/4032547?flowToken=1044374

编译时内存乱序访问

在编译时，编译器对代码做出优化时可能改变实际执行指令的顺序（例如 gcc 下 O2 或 O3 都会改变实际执行指令的顺序）：

// test.cpp
int x, y, r;
void f()
{
x = r;
y = 1;
}

编译器优化的结果可能导致 y = 1 在 x = r 之前执行完成。首先直接编译此源文件：

g++ -S test.cpp

得到相关的汇编代码如下：

movl r(%rip), %eax
movl %eax, x(%rip)
movl $1, y(%rip)

这里我们看到，x = r 和 y = 1 并没有乱序。现使用优化选项 O2（或 O3）编译上面的代码（g++ -O2 -S test.cpp），生成汇编代码如下：

movl r(%rip), %eax
movl $1, y(%rip)
movl %eax, x(%rip)

我们可以清楚的看到经过编译器优化之后 movl $1, y(%rip) 先于 movl %eax, x(%rip) 执行。避免编译时内存乱序访问的办法就是使用编译器 barrier（又叫优化 barrier）。Linux 内核提供函数 barrier() 用于让编译器保证其之前的内存访问先于其之后的完成。内核实现 barrier() 如下（X86-64 架构）：

#define barrier() __asm__ __volatile__("" ::: "memory")

现在把此编译器 barrier 加入代码中：

int x, y, r;
void f()
{
x = r;
__asm__ __volatile__("" ::: "memory");
y = 1;
}

这样就避免了编译器优化带来的内存乱序访问的问题了（如果有兴趣可以再看看编译之后的汇编代码）。本例中，我们还可以使用 volatile 这个关键字来避免编译时内存乱序访问（而无法避免后面要说的运行时内存乱序访问）。volatile 关键字能够让相关的变量之间在内存访问上避免乱序，这里可以修改 x 和 y 的定义来解决问题：

volatile int x, y;
int r;
void f()
{
x = r;
y = 1;
}

现加上了 volatile 关键字，这使得 x 相对于 y、y 相对于 x 在内存访问上有序。在 Linux 内核中，提供了一个宏 ACCESS_ONCE 来避免编译器对于连续的 ACCESS_ONCE 实例进行指令重排。其实 ACCESS_ONCE 实现源码如下：

#define ACCESS_ONCE(x) (*(volatile typeof(x) *)&(x))

此代码只是将变量 x 转换为 volatile 的而已。现在我们就有了第三个修改方案：

int x, y, r;
void f()
{
ACCESS_ONCE(x) = r;
ACCESS_ONCE(y) = 1;
}

到此基本上就阐述完了我们的编译时内存乱序访问的问题。下面开始介绍运行时内存乱序访问。

运行时内存乱序访问

在运行时，CPU 虽然会乱序执行指令，但是在单个 CPU 的上，硬件能够保证程序执行时所有的内存访问操作看起来像是按程序代码编写的顺序执行的，这时候 Memory Barrier 没有必要使用（不考虑编译器优化的情况下）。这里我们了解一下 CPU 乱序执行的行为。在乱序执行时，一个处理器真正执行指令的顺序由可用的输入数据决定，而非程序员编写的顺序。
早期的处理器为有序处理器（In-order processors），有序处理器处理指令通常有以下几步：

指令获取
如果指令的输入操作对象（input operands）可用（例如已经在寄存器中了），则将此指令分发到适当的功能单元中。如果一个或者多个操作对象不可用（通常是由于需要从内存中获取），则处理器会等待直到它们可用
指令被适当的功能单元执行
功能单元将结果写回寄存器堆（Register file，一个 CPU 中的一组寄存器）

相比之下，乱序处理器（Out-of-order processors）处理指令通常有以下几步：

指令获取
指令被分发到指令队列
指令在指令队列中等待，直到输入操作对象可用（一旦输入操作对象可用，指令就可以离开队列，即便更早的指令未被执行）
指令被分配到适当的功能单元并执行
执行结果被放入队列（而不立即写入寄存器堆）
只有所有更早请求执行的指令的执行结果被写入寄存器堆后，指令执行的结果才被写入寄存器堆（执行结果重排序，让执行看起来是有序的）

从上面的执行过程可以看出，乱序执行相比有序执行能够避免等待不可用的操作对象（有序执行的第二步）从而提高了效率。现代的机器上，处理器运行的速度比内存快很多，有序处理器花在等待可用数据的时间里已经可以处理大量指令了。
现在思考一下乱序处理器处理指令的过程，我们能得到几个结论：

对于单个 CPU 指令获取是有序的（通过队列实现）
对于单个 CPU 指令执行结果也是有序返回寄存器堆的（通过队列实现）

由此可知，在单 CPU 上，不考虑编译器优化导致乱序的前提下，多线程执行不存在内存乱序访问的问题。我们从内核源码也可以得到类似的结论（代码不完全的摘录）：

#ifdef CONFIG_SMP
#define smp_mb() mb()
#else
#define smp_mb() barrier()
#endif

这里可以看到，如果是 SMP 则使用 mb，mb 被定义为 CPU Memory barrier（后面会讲到），而非 SMP 时，直接使用编译器 barrier。

在多 CPU 的机器上，问题又不一样了。每个 CPU 都存在 cache（cache 主要是为了弥补 CPU 和内存之间较慢的访问速度），当一个特定数据第一次被特定一个 CPU 获取时，此数据显然不在 CPU 的 cache 中（这就是 cache miss）。此 cache miss 意味着 CPU 需要从内存中获取数据（这个过程需要 CPU 等待数百个周期），此数据将被加载到 CPU 的 cache 中，这样后续就能直接从 cache 上快速访问。当某个 CPU 进行写操作时，它必须确保其他的 CPU 已经将此数据从它们的 cache 中移除（以便保证一致性），只有在移除操作完成后此 CPU 才能安全的修改数据。显然，存在多个 cache 时，我们必须通过一个 cache 一致性协议来避免数据不一致的问题，而这个通讯的过程就可能导致乱序访问的出现，也就是这里说的运行时内存乱序访问。这里不再深入讨论整个细节，这是一个比较复杂的问题，有兴趣可以研究一文，其详细的分析了整个过程。

现在通过一个例子来说明多 CPU 下内存乱序访问：

// test2.cpp
#include <pthread.h>
#include <assert.h>
// -------------------
int cpu_thread1 = 0;
int cpu_thread2 = 1;
volatile int x, y, r1, r2;
void start()
{
x = y = r1 = r2 = 0;
}
void end()
{
assert(!(r1 == 0 && r2 == 0));
}
void run1()
{
x = 1;
r1 = y;
}
void run2()
{
y = 1;
r2 = x;
}
// -------------------
static pthread_barrier_t barrier_start;
static pthread_barrier_t barrier_end;
static void* thread1(void*)
{
while (1) {
pthread_barrier_wait(&barrier_start);
run1();
pthread_barrier_wait(&barrier_end);
}
return NULL;
}
static void* thread2(void*)
{
while (1) {
pthread_barrier_wait(&barrier_start);
run2();
pthread_barrier_wait(&barrier_end);
}
return NULL;
}
int main()
{
assert(pthread_barrier_init(&barrier_start, NULL, 3) == 0);
assert(pthread_barrier_init(&barrier_end, NULL, 3) == 0);
pthread_t t1;
pthread_t t2;
assert(pthread_create(&t1, NULL, thread1, NULL) == 0);
assert(pthread_create(&t2, NULL, thread2, NULL) == 0);
cpu_set_t cs;
CPU_ZERO(&cs);
CPU_SET(cpu_thread1, &cs);
assert(pthread_setaffinity_np(t1, sizeof(cs), &cs) == 0);
CPU_ZERO(&cs);
CPU_SET(cpu_thread2, &cs);
assert(pthread_setaffinity_np(t2, sizeof(cs), &cs) == 0);
while (1) {
start();
pthread_barrier_wait(&barrier_start);
pthread_barrier_wait(&barrier_end);
end();
}
return 0;
}

这里创建了两个线程来运行测试代码（需要测试的代码将放置在 run 函数中）。我使用了 pthread barrier（区别于本文讨论的 Memory Barrier）主要为了让两个子线程能够同时运行它们的 run 函数。此段代码不停的尝试同时运行两个线程的 run 函数，以便得出我们期望的结果。在每次运行 run 函数前会调用一次 start 函数（进行数据初始化），run 运行后会调用一次 end 函数（进行结果检查）。run1 和 run2 两个函数运行在哪个 CPU 上则通过 cpu_thread1 和 cpu_thread2 两个变量控制。
先编译此程序：g++ -lpthread -o test2 test2.cpp（这里未优化，目的是为了避免编译器优化的干扰）。需要注意的是，两个线程运行在两个不同的 CPU 上（CPU 0 和 CPU 1）。只要内存不出现乱序访问，那么 r1 和 r2 不可能同时为 0，因此断言失败表示存在内存乱序访问。编译之后运行此程序，会发现存在一定概率导致断言失败。为了进一步说明问题，我们把 cpu_thread2 的值改为 0，换而言之就是让两个线程跑在同一个 CPU 下，再运行程序发现断言不再失败。

最后，我们使用 CPU Memory Barrier 来解决内存乱序访问的问题（X86-64 架构下）：

int cpu_thread1 = 0;
int cpu_thread2 = 1;
void run1()
{
x = 1;
__asm__ __volatile__("mfence" ::: "memory");
r1 = y;
}
void run2()
{
y = 1;
__asm__ __volatile__("mfence" ::: "memory");
r2 = x;
}

准备使用 Memory Barrier

Memory Barrier 常用场合包括：

实现同步原语（synchronization primitives）
实现无锁数据结构（lock-free data structures）
驱动程序

实际的应用程序开发中，开发者可能完全不知道 Memory Barrier 就可以开发正确的多线程程序，这主要是因为各种同步机制中已经隐含了 Memory Barrier（但和实际的 Memory Barrier 有细微差别），这就使得不直接使用 Memory Barrier 不会存在任何问题。但是如果你希望编写诸如无锁数据结构，那么 Memory Barrier 还是很有用的。

通常来说，在单个 CPU 上，存在依赖的内存访问有序：

Q = P;
D = *Q;

这里内存操作有序。然而在 Alpha CPU 上，存在依赖的内存读取操作不一定有序，需要使用数据依赖 barrier（由于 Alpha 不常见，这里就不详细解释了）。

在 Linux 内核中，除了前面说到的编译器 barrier — barrier() 和 ACCESS_ONCE()，还有 CPU Memory Barrier：

通用 barrier，保证读写操作有序的，mb() 和 smp_mb()
写操作 barrier，仅保证写操作有序的，wmb() 和 smp_wmb()
读操作 barrier，仅保证读操作有序的，rmb() 和 smp_rmb()

注意，所有的 CPU Memory Barrier（除了数据依赖 barrier 之外）都隐含了编译器 barrier。这里的 smp 开头的 Memory Barrier 会根据配置在单处理器上直接使用编译器 barrier，而在 SMP 上才使用 CPU Memory Barrier（也就是 mb()、wmb()、rmb()，回忆上面相关内核代码）。

最后需要注意一点的是，CPU Memory Barrier 中某些类型的 Memory Barrier 需要成对使用，否则会出错，详细来说就是：一个写操作 barrier 需要和读操作（或数据依赖）barrier 一起使用（当然，通用 barrier 也是可以的），反之依然。

Memory Barrier 的范例

读内核代码进一步学习 Memory Barrier 的使用。
Linux 内核实现的无锁（只有一个读线程和一个写线程时）环形缓冲区 kfifo 就使用到了 Memory Barrier，实现源码如下：

/*
* A simple kernel FIFO implementation.
*
* Copyright (C) 2004 Stelian Pop <stelian@popies.net>
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 2 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
*
*/
#include <linux/kernel.h>
#include <linux/module.h>
#include <linux/slab.h>
#include <linux/err.h>
#include <linux/kfifo.h>
#include <linux/log2.h>
/**
* kfifo_init - allocates a new FIFO using a preallocated buffer
* @buffer: the preallocated buffer to be used.
* @size: the size of the internal buffer, this have to be a power of 2.
* @gfp_mask: get_free_pages mask, passed to kmalloc()
* @lock: the lock to be used to protect the fifo buffer
*
* Do NOT pass the kfifo to kfifo_free() after use! Simply free the
* &struct kfifo with kfree().
*/
struct kfifo *kfifo_init(unsigned char *buffer, unsigned int size,
gfp_t gfp_mask, spinlock_t *lock)
{
struct kfifo *fifo;
/* size must be a power of 2 */
BUG_ON(!is_power_of_2(size));
fifo = kmalloc(sizeof(struct kfifo), gfp_mask);
if (!fifo)
return ERR_PTR(-ENOMEM);
fifo->buffer = buffer;
fifo->size = size;
fifo->in = fifo->out = 0;
fifo->lock = lock;
return fifo;
}
EXPORT_SYMBOL(kfifo_init);
/**
* kfifo_alloc - allocates a new FIFO and its internal buffer
* @size: the size of the internal buffer to be allocated.
* @gfp_mask: get_free_pages mask, passed to kmalloc()
* @lock: the lock to be used to protect the fifo buffer
*
* The size will be rounded-up to a power of 2.
*/
struct kfifo *kfifo_alloc(unsigned int size, gfp_t gfp_mask, spinlock_t *lock)
{
unsigned char *buffer;
struct kfifo *ret;
/*
* round up to the next power of 2, since our 'let the indices
* wrap' technique works only in this case.
*/
if (!is_power_of_2(size)) {
BUG_ON(size > 0x80000000);
size = roundup_pow_of_two(size);
}
buffer = kmalloc(size, gfp_mask);
if (!buffer)
return ERR_PTR(-ENOMEM);
ret = kfifo_init(buffer, size, gfp_mask, lock);
if (IS_ERR(ret))
kfree(buffer);
return ret;
}
EXPORT_SYMBOL(kfifo_alloc);
/**
* kfifo_free - frees the FIFO
* @fifo: the fifo to be freed.
*/
void kfifo_free(struct kfifo *fifo)
{
kfree(fifo->buffer);
kfree(fifo);
}
EXPORT_SYMBOL(kfifo_free);
/**
* __kfifo_put - puts some data into the FIFO, no locking version
* @fifo: the fifo to be used.
* @buffer: the data to be added.
* @len: the length of the data to be added.
*
* This function copies at most @len bytes from the @buffer into
* the FIFO depending on the free space, and returns the number of
* bytes copied.
*
* Note that with only one concurrent reader and one concurrent
* writer, you don't need extra locking to use these functions.
*/
unsigned int __kfifo_put(struct kfifo *fifo,
const unsigned char *buffer, unsigned int len)
{
unsigned int l;
len = min(len, fifo->size - fifo->in + fifo->out);
/*
* Ensure that we sample the fifo->out index -before- we
* start putting bytes into the kfifo.
*/
smp_mb();
/* first put the data starting from fifo->in to buffer end */
l = min(len, fifo->size - (fifo->in & (fifo->size - 1)));
memcpy(fifo->buffer + (fifo->in & (fifo->size - 1)), buffer, l);
/* then put the rest (if any) at the beginning of the buffer */
memcpy(fifo->buffer, buffer + l, len - l);
/*
* Ensure that we add the bytes to the kfifo -before
* we update the fifo->in index.
*/
smp_wmb();
fifo->in += len;
return len;
}
EXPORT_SYMBOL(__kfifo_put);
/**
* __kfifo_get - gets some data from the FIFO, no locking version
* @fifo: the fifo to be used.
* @buffer: where the data must be copied.
* @len: the size of the destination buffer.
*
* This function copies at most @len bytes from the FIFO into the
* @buffer and returns the number of copied bytes.
*
* Note that with only one concurrent reader and one concurrent
* writer, you don't need extra locking to use these functions.
*/
unsigned int __kfifo_get(struct kfifo *fifo,
unsigned char *buffer, unsigned int len)
{
unsigned int l;
len = min(len, fifo->in - fifo->out);
/*
* Ensure that we sample the fifo->in index -before- we
* start removing bytes from the kfifo.
*/
smp_rmb();
/* first get the data from fifo->out until the end of the buffer */
l = min(len, fifo->size - (fifo->out & (fifo->size - 1)));
memcpy(buffer, fifo->buffer + (fifo->out & (fifo->size - 1)), l);
/* then get the rest (if any) from the beginning of the buffer */
memcpy(buffer + l, fifo->buffer, len - l);
/*
* Ensure that we remove the bytes from the kfifo -before-
* we update the fifo->out index.
*/
smp_mb();
fifo->out += len;
return len;
}
EXPORT_SYMBOL(__kfifo_get);

为了更好的理解上面的源码，这里顺带说一下此实现使用到的一些和本文主题无关的技巧：

使用与操作来求取环形缓冲区的下标，相比取余操作来求取下标的做法效率要高不少。使用与操作求取下标的前提是环形缓冲区的大小必须是 2 的 N 次方，换而言之就是说环形缓冲区的大小为一个仅有一个 1 的二进制数，那么 index & (size – 1) 则为求取的下标（这不难理解）
使用了 in 和 out 两个索引且 in 和 out 是一直递增的（此做法比较巧妙），这样能够避免一些复杂的条件判断（某些实现下，in == out 时还无法区分缓冲区是空还是满）

这里，索引 in 和 out 被两个线程访问。in 和 out 指明了缓冲区中实际数据的边界，也就是 in 和 out 同缓冲区数据存在访问上的顺序关系，由于未使用同步机制，那么保证顺序关系就需要使用到 Memory barrier 了。索引 in 和 out 都分别只被一个线程修改，而被两个线程读取。__kfifo_put 先通过 in 和 out 来确定可以向缓冲区中写入数据量的多少，这时，out 索引应该先被读取后才能真正的将用户 buffer 中的数据写入缓冲区，因此这里使用到了 smp_mb()，对应的，__kfifo_get 也使用 smp_mb() 来确保修改 out 索引之前缓冲区中数据已经被成功读取并写入用户 buffer 中了。对于 in 索引，在 __kfifo_put 中，通过 smp_wmb() 保证先向缓冲区写入数据后才修改 in 索引，由于这里只需要保证写入操作有序，故选用写操作 barrier，在 __kfifo_get 中，通过 smp_rmb() 保证先读取了 in 索引（这时候 in 索引用于确定缓冲区中实际存在多少可读数据）才开始读取缓冲区中数据（并写入用户 buffer 中），由于这里只需要保证读取操作有序，故选用读操作 barrier。