数据争用

什么是数据争用

数据争用就是多个协程同时访问相同的内存空间，并且至少有一个写操作的情况。

数据争用是高并发程序种最难排查的问题，原因在于其结果是不明确的，而且出错可能是在特定的条件下。这导致很难复现相同的错误，在测试阶段也不一定能测试出问题。

数据争用检查

检查工具race

在使用race之前，需要在系统中安装gcc，并且需要配置go环境变量开启CGO

apt install gcc
export CGO_ENABLED=1

race可以使用在多个go指令中，当检测器在程序中找到数据争用时，将打印报告。该报告包含发生race冲突的协程栈，以及此时正在运行的协程栈：

go test -race mypkg
go run -race mysrc.go
go build -race mycmd
go install -race mypkg

比如以下代码检测数据争用

package main

var count = 0

func add() {
	count++
}
func main() {
	go add()
	go add()
}
>> go run -race main.go
==================
WARNING: DATA RACE
Read at 0x00000054d5b8 by goroutine 7:
  main.add()
      /godemo/race/main.go:6 +0x29

Previous write at 0x00000054d5b8 by goroutine 6:
  main.add()
      /godemo/race/main.go:6 +0x44

Goroutine 7 (running) created at:
  main.main()
      /godemo/race/main.go:10 +0x35

Goroutine 6 (finished) created at:
  main.main()
      /godemo/race/main.go:9 +0x29
==================
Found 1 data race(s)
exit status 66
  • 竞争检测的成本因程序而异，典型情况下，内存可能增加5-10倍，执行时间增加2-20倍。
  • 同时竞争检测器为当前每个defer和recover语句额外分配8个字节，在goroutine退出之前，这些额外分配的字节不会被回收。这意味着如果有一个长期运行的goroutine并定期有defer和recover调用，则程序内存使用量可能无限增长。
  • 这些内存分配不会显示到runtime.ReadMemStats或runtime/pprof输出中

race工具原理

race工具借助了ThreadSanitizer，是google为了应对内部大量服务端C++代码的数据争用问题而开发的新一代工具。被Go语言通过CGO的形式调用。

race工具使用了矢量时钟(Vector Clock)技术，该技术在分布式系统中使用广泛，用于检测和确定分布式系统中事件的因果关系，也可以用于数据争用的检测。在Go程序中有n个协程就会有对应的n个逻辑时钟，而矢量时钟是所有这些逻辑时钟组成的数组。

矢量时钟简介：假如协程GA和协程GB初始化时都有一个逻辑时钟数组<0, 0>，假设指定第一个数字代表协程GA，第二个数字代表协程GB，每个特定的事件都会增加自己的逻辑时钟。例如当协程A完成count++操作，实际上执行了两个事件：（1）读取count内容，（2）写入数据到count。因此当协程GA结束时，其矢量时钟为<2, 0>。当加锁后，协程B能够观察到协程A释放了锁，其会更新内部对于协程A的逻辑时钟。如下图所示：

在go语言中，每个协程在创建之初都会初始化矢量时钟，并在读取或写入事件时修改自己的逻辑时钟。触发race事件主要有2种方式：

  1. 在go语言运行时中大量（超过100处）注入触发事件，比如runtime/map.go种的mapaccess1函数

    func mapaccess1(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer {
	if raceenabled && h != nil {  // 检测是否开启race
		callerpc := getcallerpc()
		pc := abi.FuncPCABIInternal(mapaccess1)
		racereadpc(unsafe.Pointer(h), callerpc, pc)
		raceReadObjectPC(t.key, key, callerpc, pc)
	}
    ....
}

  2. 依靠编译器自动插入，当加上race指令后，编译器会在可能发生数据争用的地方插入race相关的指令。如下通过 go tool compile -S 命令反编译代码为汇编代码，可以看到调用了runtime.raceread和runtime.racewrite

    ➜ go tool compile -S -race main.go 
main.add STEXT size=106 args=0x0 locals=0x18 funcid=0x0 align=0x0
   		........
        0x0019 00025 (/godemo/race/main.go:5)   CALL    runtime.racefuncenter(SB)
        0x001e 00030 (/godemo/race/main.go:6)   LEAQ    main.count(SB), AX
        0x0025 00037 (/godemo/race/main.go:6)   CALL    runtime.raceread(SB)
        0x002a 00042 (/godemo/race/main.go:6)   MOVQ    main.count(SB), CX
        0x0031 00049 (/godemo/race/main.go:6)   MOVQ    CX, main..autotmp_2+8(SP)
        0x0036 00054 (/godemo/race/main.go:6)   LEAQ    main.count(SB), AX
        0x003d 00061 (/godemo/race/main.go:6)   NOP
        0x0040 00064 (/godemo/race/main.go:6)   CALL    runtime.racewrite(SB)

原子锁（自旋锁）

考虑一下count ++这样一个操作，看起来是一句代码，但其实底层经历了读取数据、更新cpu缓存、存入内存等一系列操作。所以我们代码的执行并不是原子性的，一旦多个协程并发执行就会可能出现严重的错误。

而且许多编译器和CPU处理器会通过调整指令顺序进行优化，因此指令执行顺序可能与代码中显示的不同。在同一个协程中，代码编译后顺序也可能发生变化。

因此需要有一种机制解决并发访问时数据冲突及内存操作乱序的问题，即提供一种原子性的操作。这通常依赖于硬件的支持：例如X86指令集的LOCK指令，对应go语言的sync/automic包。例如以下代码：

var count int64 = 0
func add() {
    atomic.AddInt64(&count, 1)   //  这句代码保证了原子性
}
func main() {
    go add()
    go add()
}

还有一个非常重要的操作CAS：CompareAndSwap，与元素值比较并替换。如下代码所示：使用一个for循环不断轮询原子操作，直到原子操作成功才获取该锁。

var count int64 = 0
var flag int64 = 0

func add() {
	for {
		if atomic.CompareAndSwapInt64(&flag, 0, 1) {
			count++
			atomic.StoreInt64(&flag, 0)
			return
		}
	}
}
func main() {
	go add()
	go add()
	go add()
	time.Sleep(time.Second * 1)
	fmt.Println(count)
}
// 输出：3

这种自旋锁的形式在Go源码中随处可见，通过原子操作可以构建许多同步原语：自旋锁、信号量、互斥锁等
但是原子锁虽然高效且简单，但并不是万能的：

  1. 如果一个协程长时间霸占锁，其它抢占协程将会一直自旋下去（忙等），导致无意义的消耗cpu资源；
  2. 当有许多正在获取锁的协程时，可能会有协程一直抢不到锁。

为了解决这种问题，操作系统的锁接口提供了终止与唤醒机制，在操作系统内部会构建起锁的等待队列，以便之后被唤醒。但是调用操作系统级别的锁会锁住整个线程使之无法运行，另外锁的抢占还会涉及线程之间的上下文切换。

因此go提供了一种比传统操作系统级别的锁更加轻量级的互斥锁：sync.Mutex，提供了更加复杂的机制避免自旋锁的争用问题。

互斥锁 sync.Mutex

互斥锁是一种混合锁，其实现的方式包含了自旋锁，同时参考了操作系统锁的实现。
sync.Mutex的结构体很简单：

type Mutex struct {
	state int32
	sema  uint32
}

  • state：通过位图的形式存储了当前锁的状态，其中包含是否为锁定状态、正在等待被唤醒的协程数量、两个和饥饿模式有段的标志；

  • sema：互斥锁中实现的信号量，作用会在后面讲解。

互斥锁加锁流程

第一阶段：

  1. 快速路径：先执行快速路径，使用原子锁快速抢占，如果抢占成功则立即返回，否则调用lockSlow，进入慢速路径；
  2. 慢速路径：在慢速路径，先用自旋锁抢占一段时间，而不会立即休眠，使得互斥锁在频繁加锁与释放时也能良好工作。出现下面4中情况自旋状态立即停止：
    （1）程序在单核CPU上运行。
    （2）逻辑处理器P小于等于1，同单核cpu道理一样。
    （3）当前协程所在的P的本地队列上有其它协程待运行。
    （4）自旋次数超过了设定的阈值（执行30次汇编PAUS指令）。

第二阶段：

信号量同步：自旋之后还没获取锁，进入信号量同步阶段：如果是加锁操作，信号量计数值减1；如果时解锁操作，信号量计数值加1。
当信号量计数大于0，说明有其它协程执行了解锁操作，这时加锁协程可以立即完成直接退出。
当信号量计数等于0，说明此时锁处于被占用状态，当前协程需要进入休眠状态。

第三阶段：

进入第三阶段后，所有锁的信息会根据锁的地址，计算出一个hash值，存储在全局变量semtable哈希表中。如果出现hash冲突，通过双向链表解决。

哈希桶中的链表还被构造成了特殊的treap树，treap树是一种引入了随机数的二叉搜索树。 如果查找到已存在的锁，则将当前协程添加到等待队列的尾部；如果不存在该锁，向当前treap树中添加一个新的锁。 注意：在访问哈希表时，仍然面临数据争用问题，因此也需要加锁。此时的方式先自旋一定次数，如果没有获取到，则调用操作系统级别的锁。

饥饿模式：当长时间无法获取锁时，当前的互斥锁会进入饥饿模式，在饥饿模式下，新的协程不会进入自旋状态，而是直接放入等待队列中。放入等待队列中的协程会主动让渡执行权力进入新的调度循环。在饥饿模式下，unlock会唤醒最先进入队列的协程，从而保证公平。

互斥锁释放流程

  1. 如果当前处于普通锁定状态，即没有进入饥饿模式和唤醒状态，也没有多个协程因抢占锁陷入阻塞，则立即释放退出。
  2. 如果锁未处于饥饿状态且当前mutexWorken已设置，说明有其它申请锁的协程准备从正常状态退出，这时锁释放后可以直接退出。
  3. 如果处于饥饿状态，则进入信号量同步阶段，到全局hash表中寻找当前锁的等待队列，以先入先出的顺序唤醒指定协程。将唤醒的协程放入当前协程所在的P的runext字段中，会被优先调度。当前协程主动让渡自己的执行权力，让被唤醒的协程直接运行。

读写锁 sync.RWMutex

在读多写少的情况下，如果长时间没有写操作，读取到的会是完全相同的值。因此完全不需要每次读取都获取互斥锁。
因此读写锁的优势是可以做到多个读并发。使用读写锁需要注意：

  1. 多个协程可以同时获得读锁并执行
  2. 要获取读锁，需要等待写锁的释放
  3. 要获取写锁，必须等待所有的读写锁释放

总之，读锁必须能观察到上一次写锁写入的值，写锁要等待之前所有的读锁释放才能写入。

读写锁原理

读写锁使用了互斥锁和信号量这两种机制：sync/rwmutex.go

type RWMutex struct {
	w           Mutex        // 互斥锁
	writerSem   uint32       // 信号量，写锁等待读取完成
	readerSem   uint32       // 信号量，读锁等待写入完成
	readerCount atomic.Int32 // 当前正在执行的读操作的数量
	readerWait  atomic.Int32 // 写操作被阻塞时，等待的读操作数量
}

读锁

  • 读操作先通过原子操作将readerCount加1，如果readerCount>=0,说明已有其它读协程，直接返回；
  • 当readerCount<0，说明有写锁，当前协程将借助信号量陷入等待状态，如果获取了信号量立即退出，获取不到的逻辑与互斥锁相似；
  • 读锁解锁时，如果当前没有写锁，其成本只有一个原子操作并直接退出。如果当前有写锁正在等待，则判断当前是否是最后一个被释放的读锁，如果是则需要增加信号量并唤醒写锁

写锁

  • 先获取互斥锁，接着readerCount - rwmutexMaxReaders，阻止后续的读操作；
  • 如果有其它协程持有读锁，那么当前协程进入全局等待队列并进入休眠状态，当最后一个读锁被释放时，会唤醒该协程；
  • 解锁时 readercount + rwmutexMaxReaders，表示不会阻塞后续的读锁。然后依次唤醒所有等待的读锁，最后释放互斥锁；