我们一起聊聊 Go 内存模型

Go 内存模型​指定了在什么条件下，一个 goroutine​ 对变量的写操作可以被另一个 goroutine 读取到。

建议

当一份数据同时被多个 goroutine 读取时，在修改这份数据时，必须序列化读取顺序。

要序列化读取，请使用通道或其他同步原语（例如 sync 包）来保护数据。

内存模型

内存模型​ 描述了对程序执行的要求，这些要求由 goroutine​ 执行组成，而 goroutine 执行又由内存操作组成。

内存操作由四个细节概括：

• • 类型: 如普通数据读取、普通数据写入、同步操作、原子数据读取、互斥操作或通道操作
• • 代码中的位置
• • 正在读取的内存位置
• • 正在读取或写入的值

goroutine 的执行过程被抽象为一组内存操作。

同步

happens before

在一个 goroutine 内部，即使 CPU 或者编译器进行了指令重排，程序执行顺序依旧和代码指定的顺序一致。

对于多个 goroutine 之间的通信，则需要依赖于 Go 保证的 happens before 规则约束。

下面来介绍不同场景下的 happens before 规则。

初始化

程序初始化在单个 goroutine​ 中运行 (main goroutine​)，但是 main goroutine​ 可能会并发运行其他 goroutine。

如果包 p 导入包 q，则 q 的 init 函数在 p 的 init 函数之前完成。

所有 init​ 函数在主函数 main.main 开始之前完成。

goroutine 构造

内存模型​ 保证启动新 goroutine​ 的 go​ 语句会在 goroutine 内部语句执行之前完成。

如下代码所示:

package main

var a string

func f() {
    print(a)
}

func hello() {
    a = "hello, world"
    go f()
}

调用 hello 将在将来某一时刻打印 "hello, world"（也可能在 hello 返回之后打印）。

goroutine 销毁

内存模型​ 不保证 goroutine 的退出发生在程序中的任何事件之前。

如下代码所示:

package main

var a string

func hello() {
    go func() { a = "hello" }()
    print(a)
}

对 a 的赋值之后没有任何同步事件，因此不能保证任何其他 goroutine​ 都看到更新后的 a 的值。事实上，激进的编译器可能会删除整个 go func() ... 语句。

如果一个 goroutine​ 的结果必须被另一个 goroutine 看到，必须使用同步机制（例如锁或通道通信）来建立相对顺序。

channel 通信

channel​ 通信是 goroutine​ 之间同步的主要方式，特定 channel​ 上的发送和接收是一一对应的，通常在不同的 goroutine 上进行。

channel 的发送操作发生在对应的接收操作完成之前 (happens before)。

如下代码所示:

package main

var c = make(chan int, 10)
var a string

func f() {
    a = "hello, world"
    c <- 0
}

func main() {
    go f()
    <-c
    print(a)
}

程序确保打印 "hello world" :

• • 对 a 的写操作发生在 c 的发送操作之前 (happens before)
• • c 的发送操作发生在 c 的接收操作完成之前 (happens before)
• • c 的接收操作发生在 print 之前 (happens before)

对 channel​ 的 close​ 操作发生在 channel​ 的接收操作之前 (happens before)，且由于 channel 被关闭，接收方将会收到一个零值。

在前面的示例中，将 c <- 0 替换为 close(c), 程序的行为不会发生变化。

unbuffered channel 的接收操作发生在发送操作完成之前。

如下代码所示（和上面的程序差不多，但交换了发送和接收语句并使用了一个 unbuffered channel）：

package main

var c = make(chan int)
var a string

func f() {
    a = "hello, world"
    <-c
}

func main() {
    go f()
    c <- 0
    print(a)
}

这段代码同样能保证最终输出 "hello, world":

• • 对 a 的写入发生在 c 的之前 (happens before)
• • c 的接收操作发生在 c 的发送操作完成之前
• • c 的发送操作发生在 print 操作之前

如果 channel​ 为缓冲（例如，c = make(chan int, 1)），那么程序将不能保证打印 "hello, world" （它可能会打印空字符串、崩溃或执行其他操作）。

容量为 C 的 channel 上的第 k 个接收操作发生在第 k+C 个发送操作之前。

这条规则可以视为对上面规则的拓展，(当 c = 0 时就是一个 unbuffered channel​ 了)，可以使用 buffered channel​ 封装出一个信号量 (semaphore)， 用 channel​ 里面的元素数量来代表当前正在使用的资源数量，channel​ 的容量表示同时可以使用的最大资源数量。当申请信号量时，就往 channel​ 中发送一个元素， 释放信号量时就从 channel 中接收一个元素，这是限制并发的常用操作。

下面的程序为 work​ 列表中的每个元素启动一个 goroutine​，并使用名字 limit 的 channel 来协调协程，保证同一时刻最多有三个方法在执行 。

package main

var limit = make(chan int, 3)

func main() {
    for _, w := range work {
        go func(w func()) {
            limit <- 1
            w()
            <-limit
        }(w)
    }
    select{}
}

锁

sync​ 包实现了两种锁数据类型，sync.Mutex​ 和 sync.RWMutex。

**对于任何 sync.Mutex​ 或 sync.RWMutex​ 变量 l，在 n < m 的条件下，对 l.Unlock()​ 的第 n 次调用发生在 l.Lock() 的第 m 次调用的返回之前 (happens before)**。

如下代码所示:

package main

var l sync.Mutex
var a string

func f() {
    a = "hello, world"
    l.Unlock()
}

func main() {
    l.Lock()
    go f()
    l.Lock()
    print(a)
}

上面的代码保证会输出 "hello, world":

• • l.Unlock()​ 的第一次调用 (在 f() 内) 发生在第二次调用 l.lock() 返回之前 (在 main) (happens before)
• • 第二次调用 l.lock() 发生在 print(a) 之前 (happens before)

对于类型为 sync.RWMutex​ 的变量 l，对任何一次 l.RLock()​ 的调用，都会存在一个 n，使得 l.RLock()​ 发生在第 n 次调用 l.Unlock()​ 之后， 并发生在第 n + 1 次 l.Lock 之前。

sync.Once

sync​ 包通过使用 Once​ 类型在存在多个 goroutine 的情况下提供了一种安全的初始化机制。多个线程可以为特定的 f 执行一次 Do(f)， 但只有一个会运行 f()，而其他调用将阻塞直到 f() 返回。

once.Do(f)​ 中 f() 将会在所有的 once.Do(f) 返回之前返回 (happens before)。

如下代码所示:

package main

var a string
var once sync.Once

func setup() {
    a = "hello, world"
}

func doprint() {
    once.Do(setup)
    print(a)
}

func twoprint() {
    go doprint()
    go doprint()
}

twoprint​ 只会调用一次 setup​。 setup 将在调用 print 之前完成，结果是 "hello, world" 被打印两次。

Atomic Values

sync/atomic​ 包中的 API 统称为 "原子操作"，可用于同步不同 goroutine 的执行。如果原子操作 A 的效果被原子操作 B 读取到， 那么 A 在 B 之前同步，程序中执行的所有原子操作的行为就像顺序执行一样。

Finalizers

runtime​ 包提供了一个 SetFinalizer​ 函数，该函数添加了一个终结器，当程序不再可以读取特定对象时将调用该终结器，在完成调用 f(x) 之前同步调用 SetFinalizer(x, f)。

其他同步原语

sync​ 包还提供了其他同步原语，包括 sync.Cond​, sync.Map​, sync.Pool​, sync.WaitGroup。

错误的同步

存在竞态的程序是不正确的，并且可以表现出非顺序一致的执行。

如下代码所示:

package main

var a, b int

func f() {
    a = 1
    b = 2
}

func g() {
    print(b)
    print(a)
}

func main() {
    go f()
    g()
}

g() 可能会发生先输出 2 再输出 0 的情况。

双重检查锁定是一种避免同步开销的尝试，如下代码所示，twoprint 程序可能被错误地写成：

package main

var a string
var done bool

func setup() {
    a = "hello, world"
    done = true
}

func doprint() {
    if !done {
        once.Do(setup)
    }
    print(a)
}

func twoprint() {
    go doprint()
    go doprint()
}

在 doprint 内，即使读取到了 done 变量被更新为 true，也并不能保证 a 变量被更新为 "hello, world" 了。因此上面的程序可能会打印出一个空字符串。

下面是一段忙等待的代码，它的原本目的是：无限循环，直至变量 a 被赋值。

package main

var a string
var done bool

func setup() {
    a = "hello, world"
    done = true
}

func main() {
    go setup()
    for !done {
    }
    print(a)
}

和上面一样，读取到 done 的写操作并不能表示能读取到对 a 的写操作，所以这段代码也可能会打印出一个空白的字符串。更糟的是， 由于不能保证 done 的写操作一定会被 main 读取到，main 可能会进入无限循环。

如下代码所示:

package main

type T struct {
    msg string
}

var g *T

func setup() {
    t := new(T)
    t.msg = "hello, world"
    g = t
}

func main() {
    go setup()
    for g == nil {
    }
    print(g.msg)
}

即使 main 读取到 g != nil 并退出循环，也不能保证它会读取到 g.msg 的初始化值。

在刚才所有这些示例代码中，问题的解决方案都是相同的：使用显式同步。

错误的编译

Go 内存模型 限制编译器优化，就像限制 Go 程序一样，一些在单线程程序中有效的编译器优化并非在所有 Go 程序中都有效。 尤其是编译器不得引入原始程序中不存在的写入操作，不得允许单个操作读取多个值，并且不得允许单次操作写入多个值。

以下所有示例均假定 "*p" 和 "*q" 指的是多个 goroutine 都可读取的指针变量。

不将数据竞态引入无竞态程序，意味着不将写入操作从它们出现的条件语句中移出。如下代码所示，编译器不得反转此程序中的条件：

*p = 1
if cond {
    *p = 2
}

也就是说，编译器不得将程序重写为以下代码：

*p = 2
if !cond {
    *p = 1
}

如果 cond​ 为 false，另一个 goroutine​ 正在读取 *p，那么在原来的程序中，另一个 goroutine​ 只能读取到 *p​ 等于 1。 在改写后的程序中，另一个 goroutine 可以读取到 2，这在以前是不可能的。

不引入数据竞态也意味着假设循环不会终止，如下代码所示，编译器通常不得在该程序的循环之前移动对 *p​ 或 *q 的读取顺序：

n := 0
for e := list; e != nil; e = e.next {
    n++
}
i := *p
*q = 1

如果 list 指向环形链表，那么原始程序将永远不会读取 *p​ 或 *q，但重写后的代码可以读取到。

不引入数据竞态也意味着，假设被调用的函数可能不返回或没有同步操作，如下代码所示，编译器不得在该程序中的函数调用之前移动对 *p​ 或 *q 的读取。

f()
i := *p
*q = 1

如果调用永远不会返回，那么原始程序将永远读取不到 *p​ 或 *q，但重写后的代码可以读取到。

不允许单次读取多个值，意味着不从共享内存中重新加载局部变量。如下代码所示，编译器不得丢弃变量 i 并再次从 *p 重新加载。

i := *p
if i < 0 || i >= len(funcs) {
    panic("invalid function index")
}
... complex code ...
// compiler must NOT reload i = *p here
funcs[i]()

如果复杂代码需要很多寄存器，单线程程序的编译器可以丢弃变量 i 而不保存副本，然后在 funcs[i]()​ 调用之前重新加载 i = *p。

不允许单次写入多个值，意味着在写入之前不使用局部变量的内存作为临时变量，如下代码所示，编译器不得在此程序中使用 *p 作为临时变量：

*p = i + *p/2

也就是说，它不能将程序改写成这个：

*p /= 2
*p += i

如果 i 和 *p​ 开始等于 2，则原始代码确实 *p​ = 3，因此一个执行较快线程只能从 *p​ 中读取 2 或 3。 重写的代码执行 *p​ = 1，然后 *p = 3，从而允许竞态线程也读取 1。

请注意，所有这些优化在 C/C++ 编译器中都是允许的：与 C/C++ 编译器共享后端的 Go 编译器必须注意禁用对 Go 无效的优化。 如果编译器可以证明数据竞态不会影响目标平台上的正确执行，则不需要禁止引入数据竞态。如下代码所示，在大多数 CPU 上重写都是有效的。

n := 0
for i := 0; i < m; i++ {
    n += *shared
}

重写为

n := 0
local := *shared
for i := 0; i < m; i++ {
    n += local
}

前提是可以证明 *shared 不会出现读取出错，因为潜在的读取不会影响任何现有的并发读取或写入。

结论

当谈到有竞态的程序时，程序员和编译器都应该记住这个建议：不要自作聪明。

笔者寄语

本文翻译自官方 博客原文[1], 希望读者在读完本文后，能够深入理解 happens before 在各场景下的规则，写出更加健壮的并发程序。

Reference

• • The Go Memory Model[2]
• • Memory Reordering[3]
• • Updating the Go Memory Model[4]

引用链接

[1]​ 博客原文: https://go.dev/ref/mem[2]​ The Go Memory Model: https://go.dev/ref/mem[3]​ Memory Reordering: https://cch123.github.io/ooo/[4]​ Updating the Go Memory Model: https://research.swtch.com/gomm