信号量，锁和 golang 相关源码分析

Posted 2021-04-05 Golang语言社区

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了信号量，锁和 golang 相关源码分析相关的知识，希望对你有一定的参考价值。

原创投稿：南瓜waniu

1资源同步

1.1 解决方案

2 信号量

2.1 共享变量
2.2 信号量

3 锁

3.1 死锁
3.2 活锁
3.3 饥饿

4 Golang sync 包

4.4.1 数据结构
4.4.2 NewCond函数
4.4.3 Wait方法
4.4.4 Singal方法
4.4.5 Broadcast方法
4.3.1 数据结构
4.3.2 Add和Done方法
4.3.3 Wait方法
4.2.1 常量和结构
4.2.1 RLock和RUnlock方法
4.2.2 Lock和Unlock方法
4.1.1 接口和结构
4.1.2 Lock 方法
4.1.3 Unlock方法
4.1 sync.mutex.go
4.2 sync.rwmutex.go
4.3 sync.waitgroup.go
4.4 sync.cond.go

1资源同步

并发已经成为现代程序设计中的重要考虑内容，但是并发涉及到一个很重要的内容就是资源同步。当两个或者多个线程访问同样的资源的时候，运行的结果取决于线程运行时精确的时序。这样导致结果与期望的结果大相径庭，因此我们需要对资源的访问顺序进行控制，已达到资源同步的目的。

1.1 解决方案

将共享资源（也就是共享内存）的程序片段成为临界区域，通过适当安排，使得两个者线程同时位于临界区域。对于临界区域访问解决方案，需要满足如下4个条件

任何两个线程不能同时位于临界区
不对CPU执行速度和时间做任何假设
临界区外运行的线程不阻塞其他线程
不能使线程无限期等待进入临界区

常见的互斥解决方案：

屏蔽中断线程的切换是由CPU中断机制提供的，如果一个线程进入临界区域后，CPU关闭中断响应；在离开临界区域后，再打开中断机制。那么在临界区域将不会有其他线程来竞争资源。当时将屏蔽中断权利交给用户空间执行是不明智的，而且对于多核CPU而言没有效果。

锁变量几乎每一个编程语言都提供了资源同步方式：锁机制。该机制通过对资源进行Lock和Unlock，以达到对关键资源有序访问。

严格轮换法线程不停的执行CPU时间，连续测试某一个值是否出现。但是如果认为等待的时间非常短，可以使用该方式浪费CPU时间，用于等待的锁也成为自旋锁。

2 信号量

2.1 共享变量

在理解信号量之前，先了解采用共享变量使用多线程会出现什么问题。下面是一个C代码片段

1for (i=0; i<niters; i++){
2    cnt ++;
3}

cnt为全局变量，一个线程执行该代码片段的时候的汇编代码如下：

 1   movq (%rdi), %rcx
 2    testq %rcx, %rcx
 3    jle .L2
 4    movl $0, %eax
 5.L3:
 6    movq cnt(%rip), %rdx
 7    addq %eax
 8    movq %eax, cnt(%rip)
 9    addq $1, %rax
10    cmpq %rcx, %rax
11    jne .L3
12.L2

2.2 信号量其中6-8行分别对应对应着加载cnt，更新cnt和存储cnt。将cnt变量从内存位置读出，加载到CPU寄存器中，在CPU运算器中加1，然后存储到cnt的内存位置。虽然代码中cnt++只有一行，但是转换为汇编代码的时候不只有一个操作，也就是说该语句不是原子操作。如果多个线程同时执行代码，按照之前的条件，不对CPU的执行顺序做任何假设，如果其中线程a在执行7行汇编代码，而线程b执行6行汇编代码，那么b将"看不到"线程a对全局变量cnt加1的操作，那么每次执行的结果cnt也不完全一致。

计算机领域先驱Dijkstra提出经典的解决上述问题的方法：信号量(semaphore)。它是一个非负整数的全局变量。而且该变量只能有两个特殊操作来处理: P和V。

P(s)：如果s非零，那么P将s减1，并且立即返回。如果s为零，那么就挂起这个线程，知道s为非零。
V(s): V操作将s加1。如果有任何线程阻塞在P操作等待s非零，那么V将重启其中线程中的一个。

Posix标准定义需要操作信号量的函数

1#include <semaphore.h>
2int sem_init(sem_t *sem, 0, unsigned int value);
3int sem_wait(sem_t *s); /*P(s)*/
4int sem_post(sem_t *s); /*P(s)*/

那么如何使用信号量是的2.1小节出现同步问题解决呢？首先定义全局信号量

1volatile long cnt = 0; /* global variable */
2sem_t mutex; /*global semaphore*/

初始化信号量，在这里初始值为1

1sem_init(&mutex, 0, 1);

最后使用信号量操作函数将临界区域代码包含起来

1for (i =0; i<niters; i++){
2    sem_wait(&mutex);
3    cnt++;
4    sem_post(&mutex);
5}

3 锁

3.1 死锁

首先看一下死锁的规范定义:

如果一个线程（进程）集合中的每一个线程（进程）都在等待只能由该线程（进程）集合中的其他线程（进程）才能引发的事件，那么该线程（进程）集合是死锁的。

举一个例子，如果线程 a 和线程 b 同是执行，线程a获取了资源r1，等待获取资源r2；而线程b获取了资源r2，等待获取资源r1。那么线程a和线程b组成的集合是死锁的。

预防死锁

破坏占有等待条件
对于需要获取多个资源的线程，一次性获取全部资源，而不是依次获取各个资源。
破坏环路等待条件
死锁集合的线程按照等占有线程和等待线程可以组成有向环图。那么如果对所有资源进行排序，所有线程按照资源顺序获取资源。

3.2 活锁

在某些情况下，当线程意识它不能获下一个资源的时候，它会“礼貌性”地释放已经获得的资源，然后等待1ms，在尝试一次。如果另一个线程也在相同的时候做了相同的操作，那么同步的步调将导致两个线程都无法前进。

3.3 饥饿

在信号量小节中，当执行V操作后，将恢复挂起线程中的一个，那么问题出现了：如果有多个线程被挂起，那么选择哪个线程恢复呢？如果随机选择一个线程恢复，如果源源不断的线程到达临界区域并且挂起，那么很有可能出现某一个线程一直等待资源，而导致"饥饿"。当然也有好的FILO调度策略来解决调用问题。当时问题在于刚刚到达的线程有很好的局部性，也就是CPU的寄存器、缓存等包含了该线程的局部变量，如果程获得资源锁，很好的避免了线程上下文切换，对性能提高很有帮助。

在go语言的互斥锁中采用结合上述两种策略，接下来小节中，将会仔细分析源码。

4 Golang sync 包

4.1 sync.mutex.go

4.1.1 接口和结构

包含了Locker接口和Mutex结构：

1type Locker interface {
2    Lock()
3    Unlock()
4}
5type Mutex struct {
6    state int32
7    sema  uint32
8}

Mutex实现了Locker接口，该结构包含了state的字段，用来表示该锁当前状态；sema则为一个信号量。state是一个32位的整数，不同比特位包含了不同的意义，其中源码中的有很详细的注释，该注释很好解释mutex如何工作：

互斥锁有两种状态：正常状态和饥饿状态。在正常状态下，所有挂起等待的goroutine按照FIFO顺序等待。唤醒的goroutine将会和刚刚到达的goroutine竞争互斥锁的拥有权，因为刚刚到达的goroutine具有优势：它刚刚正在CPU上执行，所以刚刚唤醒的goroutine有很大可能在锁竞争中失败。如果一个等待的goroutine超过1ms没有获取互斥锁，那么它将会把互斥锁转变为饥饿模式。在饥饿模式下，互斥锁的所有权将移交给等待队列中的第一个。新来的goroutine将不会尝试去获得互斥锁，也不会去尝试自旋操作，而是放在队列的最后一个。如果一个等待的goroutine获取的互斥锁，如何它满足一下其中的任何一个条件：(1)它是队列中的最后一个；(2)它等待的时候小于1ms。它会将互斥锁的转台转换为正常状态。正常状态有很好的性能表现，饥饿模式也是非常重要的的，因为它能阻止尾部延迟的现象。

1const (
2    mutexLocked = 1 << iota // mutex is locked
3    mutexWoken
4    mutexStarving
5    mutexWaiterShift = iota
6    starvationThresholdNs = 1e6
7)

mutexLocked

该值为1, 第一位比特位1，代表了该是否该互斥锁已经被锁住。mutex.state与它进行&操作，如果为1表示已经锁住，0则表示未被锁住。

mutexWoken

该值为2，第二位比特位1，代表了该互斥锁是否被唤醒，mutex.state与它进行&操作，如果为1表示已经被唤醒，0代表未被唤醒

mutexStarving

该值为4,第三位比特为1，代表了该互斥锁是否处于饥饿状态，mutex.state与它进行&操作，如果为1表示处于饥饿转态，0表示处于正常状态。

mutexWaiterShift

该值为3，表示mutex.state右移3位后为等待的goroutine的数量。

starvationThresholdNs

goroutine将互斥锁转换状态的时间等待的临界值：一百万纳秒，也就是1ms。

4.1.2 Lock 方法

1if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
2        if race.Enabled {
3            race.Acquire(unsafe.Pointer(m))
4        }
5        return
6}

CompareAndSwapInt32是一个原子操作，它判断是一个参数的值是否等于第二个参数，如果相等，则将第一个参数设置为第三个参数，并返回true；否则对一个参数不做任何操作并且返回false。这一段是代码是处理第一次goroutine进行尝试Lock操作，如果一切都是初始状态，则m.state为.....0000001并且返回，进入临界区域代码，否则代码继续往下走。

1var waitStartTime int64
2starving := false
3awoke := false
4iter := 0
5old := m.state

首先定义了一下变量：goroutine等待时间，是否饥饿转台，是否唤醒和自旋迭代次数和保存当前互斥锁状态。接下来是一个for循环，只有退出循环才能进入临界区域代码，纵观代码只有两处使用break来退出循环。

 1for {
 2    if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
 3            if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
 4                atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
 5                awoke = true
 6            }
 7            runtime_doSpin()
 8            iter++
 9            old = m.state
10            continue
11        }
12}

首先判断锁状态是被锁而且不是处于饥饿模式，加上还能自旋额次数，进入下一层判断。如果当前goroutine没有被唤醒，其他goroutine也没有被唤醒，等待的goroutine超过1和可以将m.state设置为唤醒转态四个条件同时满足，将awoke设置true`。然后进行自旋操作，进行一轮循环。

 1new := old
 2if old&mutexStarving == 0 {
 3            new |= mutexLocked
 4}
 5if old&(mutexLocked|mutexStarving) != 0 {
 6        new += 1 << mutexWaiterShift
 7}
 8if starving && old&mutexLocked != 0 {
 9            new |= mutexStarving
10}

这三个判断条件做了如下工作：如果当前的mutex.state处于正常模式，则将new的锁位设置为1，如果当前锁锁状态为锁定状态或者处于饥饿模式，则将等待的线程数量+1。如果starving变量为true并且处于锁定状态，则new的饥饿状态位打开。

1if awoke {
2    if new&mutexWoken == 0 {
3                throw("sync: inconsistent mutex state")
4    }
5    new &^= mutexWoken
6}

如果 goroutine已经被唤醒，则清空new的唤醒位。

1if atomic.CompareAndSawpInt32(&m.state, old, new){
2    //...
3}else{
4    //...
5}

如果更新m.state成功

1if old&(mutexLocked|mutexStarving) == 0 {
2                break 
3}

如果未被锁定并且并不是出于饥饿状态，到达第一个break，进入代码临界区域。

1queueLifo := waitStartTime != 0
2if waitStartTime == 0 {
3    waitStartTime = runtime_nanotime()
4}
5runtime_SemacquireMutex(&m.sema, queueLifo)

runtime_SemacquireMutex(s *uint32, lifo bool)函数类似P操作，如果lifo为true则将等待goroutine插入到队列的前面。在这里，对于每一个到达的goroutine，如果CompareAndSawpInt32成功，并且到达时候如果锁出于锁定状态，那么将该goroutine插入到等待队列的最后，否则插入到最前面。此时goroutine将会被挂起，等待Unlock的V操作，将唤醒goroutines

 1starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
 2old = m.state
 3if old&mutexStarving != 0 {
 4                if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
 5                    throw("sync: inconsistent mutex state")
 6                }
 7                delta := int32(mutexLocked - 1<<mutexWaiterShift)
 8                if !starving || old>>mutexWaiterShift == 1 {
 9                    delta -= mutexStarving
10                }
11                atomic.AddInt32(&m.state, delta)
12                break
13}
14

判断被唤醒的线程是否为达到饥饿状态，也就是等待时间超过1ms，如果之前的m.state不是饥饿状态，继续循环，给新到来goroutine让出互斥锁。如果已经饥饿状态，则修改等待goroutine数量和饥饿状态位，并返回进入临界代码区域。

4.1.3 Unlock方法

1new := atomic.AddInt32(&m.state, -mutexLocked)

首先创建变量new，该变量的锁位为0。接下来是饥饿状态判断

 1if new&mutexStarving == 0 {
 2        old := new
 3        for {
 4                if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
 5                return
 6            }
 7            new = (old - 1<<mutexWaiterShift) | mutexWoken
 8            if atomic.CompareAndSwapInt32(&m.state, old, new) {
 9                runtime_Semrelease(&m.sema, false)
10                return
11            }
12            old = m.state
13        }
14    } else {
15        runtime_Semrelease(&m.sema, true)
16    }

如果是正常状态，则判断如果等待的goroutine为零，或者已经被锁定、唤醒、或者已经变成饥饿状态，返回，不需要唤醒任何其他被挂起的goroutine，因为互斥锁已经被其他goroutine抢占。否则更新new值（修改等待的goroutine数量）并设置唤醒为，如果CompareAndSwapInt32成功，则通过runtime_Semrelease(&m.sema, false)恢复挂起的goroutine.r如果为 true表明将唤醒第一个阻塞的goroutine，这第一点在else饥饿的分支中体现。

4.2 sync.rwmutex.go

读写锁也是一种常见的锁机制，它允许多个线程读共享资源，只有一个线程写共享资源，接下来看看go中如何实现读写锁。

4.2.1 常量和结构

1type RWMutex struct {
2    w           Mutex  
3    writerSem   uint32 
4    readerSem   uint32 
5    readerCount int32  
6    readerWait  int32 
7}
8const rwmutexMaxReaders = 1 << 30

RWMutex结构包含了如下的字段

goroutine数量。

4.2.1 RLock和RUnlock方法

1func (rw *RWMutex) RLock() {
2    // [...]
3    if atomic.AddInt32(&rw.readerCount, 1) < 0 {
4        runtime_Semacquire(&rw.readerSem)
5    }
6//[...]
7}

首先是readerCount值+1, 如果小于零，则挂起goroutine等待readerSem。是不是很奇怪，为什么会小于零判断呢？在这里先卖一个关子，接下来会看到为什么是这样的设计逻辑。

 1func (rw *RWMutex) RUnlock() {
 2    //[...]
 3    if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
 4        if r+1 == 0 || r+1 == -rwmutexMaxReaders {
 5            race.Enable()
 6            throw("sync: RUnlock of unlocked RWMutex")
 7        }
 8        if atomic.AddInt32(&rw.readerWait, -1) == 0 {
 9            runtime_Semrelease(&rw.writerSem, false)
10        }
11    }
12    //[...]
13}

首先将readerCount减去1，如果小于零，再讲readWait减去1，如果是离开读的goroutine数量为零，则对writerSem信号量进行V操作。

4.2.2 Lock和Unlock方法

1func (rw *RWMutex) Lock() {
2    //[...]
3    rw.w.Lock()
4    r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
5    if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
6        runtime_Semacquire(&rw.writerSem)
7    }
8    //[...]
9}

首先rw.w.Lock操作，来防止其他goroutine对共享资源的写访问。然后将readerCount减去rwmutexMaxReaders，表明还剩多少goroutine可以进行读访问，这也解释在RLock中小于零的判断，如果还可以还可以进行读访问，则必须获得readerSem信号量。在Lock中接下来是对readWait判断，如果该数量不为零，则需要对writerSem进行P操作，而V操作只在RUnlock方法中，如果最后一个读goroutine离开，则进行V操作。

 1func (rw *RWMutex) Unlock() {
 2    //[...]
 3    r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
 4    if r >= rwmutexMaxReaders {
 5        race.Enable()
 6        throw("sync: Unlock of unlocked RWMutex")
 7    }
 8    for i := 0; i < int(r); i++ {
 9        runtime_Semrelease(&rw.readerSem, false)
10    }
11    //[...]
12}

首先恢复readCounter为正数，然后对readerSem信号量进行r次V操作，唤醒在RLock中被挂起的goroutine。

4.3 sync.waitgroup.go

WaitGroup通常用在等待一组goroutine全部完成。调用Add方法指明要等待的goroutine的数量，调用Done方法说明该goroutine已经完成，而Wait方法是阻塞等待的goroutine。

4.3.1 数据结构

1type WaitGroup struct {
2    noCopy noCopy
3    state1 [12]byte
4    sema   uint32
5}

noCopy字段说明WaitGroup不允许拷贝，而state1字段是一个非常tricky的方法，用其中的8个字节(64bit)来保存一些状态。高位的32bit用来表示需要等待的goroutine的数量，地位的32bit用来表示被挂起的goroutine的数量。至于为什么不直接使用64bit的数据主要是为了考虑32为编译器无法保证64位对齐。sema则是一个信号量。

1func (wg *WaitGroup) state() *uint64 {
2    if uintptr(unsafe.Pointer(&wg.state1))%8 == 0 {
3        return (*uint64)(unsafe.Pointer(&wg.state1))
4    } else {
5        return (*uint64)(unsafe.Pointer(&wg.state1[4]))
6    }
7}

该方法是一个辅助方法，用来获取state，一个64为的无符号整数。

4.3.2 Add和Done方法

1func (wg *WaitGroup) Done() {
2    wg.Add(-1)
3}

Done方法其实调用了Add(-1)方法，所以我们着重讨论Add方法。

 1func (wg *WaitGroup) Add(delta int) {
 2    statep := wg.state()
 3    //[...]
 4    state := atomic.AddUint64(statep, uint64(delta)<<32)
 5    v := int32(state >> 32)
 6    w := uint32(state)
 7    if race.Enabled && delta > 0 && v == int32(delta) {
 8            race.Read(unsafe.Pointer(&wg.sema))
 9    }
10    if v < 0 {
11        panic("sync: negative WaitGroup counter")
12    }
13    if w != 0 && delta > 0 && v == int32(delta) {
14        panic("sync: WaitGroup misuse: Add called concurrently with Wait")
15    }
16    if v > 0 || w == 0 {
17        return
18    }
19    if *statep != state {
20        panic("sync: WaitGroup misuse: Add called concurrently with Wait")
21    }
22    // Reset waiters count to 0.
23    *statep = 0
24    for ; w != 0; w-- {
25        runtime_Semrelease(&wg.sema, false)
26    }
27}

首先是获取state，然后将delta右移32位，加上等待的goroutine数量。v和w分别代表了需要等待的goroutine和被阻塞的goroutine的数量。接下来v==int32(delta)判断条件表明如果是第一次Add操作，则必须与等待的goroutine同步，在Wait方法中可以看到同样的操作。接下来是一些抛异常操作，如果等待的数量为负数，如何第一次Add操作没有同步。if >0 || w==0条件表明如何v没有降到零，或者被阻塞的goroutine数量为零，直接返回。如何v为零，则按照w的数量，依次对信号量ws.sema进行V操作。

4.3.3 Wait方法

 1func (wg *WaitGroup) Wait() {
 2    //[...]
 3    for {
 4        state := atomic.LoadUint64(statep)
 5        v := int32(state >> 32)
 6        w := uint32(state)
 7        //[...]
 8        // Increment waiters count.
 9        if atomic.CompareAndSwapUint64(statep, state, state+1) {
10            if race.Enabled && w == 0 {
11                race.Write(unsafe.Pointer(&wg.sema))
12            }
13            runtime_Semacquire(&wg.sema)
14            if *statep != 0 {
15                panic("sync: WaitGroup is reused before previous Wait has returned")
16            }
17            //[...]
18            return
19        }
20    }
21}

Wait方法同样也是CAS算法，首先获取需要等待的goroutine的数量v和阻塞的goroutine数量w, 然后将阻塞的goroutine数量+1，如果之前的w为零，表示是第一次等待，则与Add操作进行同步，最后后对信号量wg.sema进行P操作。

4.4 sync.cond.go

在编程中使用Cond也叫管程(monitor)，它可以用来使不同线程完成互斥条件，也可以使某个线程等待某个条件的发生。常见的使用模式如下：

 1var locker = new(sync.Mutex)
 2var cond = sync.NewCond(locker)
 3var condition = true
 4// goroutine A
 5cond.L.Lock()
 6for condition {
 7    cond.Wait()
 8}
 9// ...
10cond.L.Unlock()
11
12//goroutine B
13condiiton = false
14cond.Signal()

为什么使用for循环作为判断进入Wait的条件而不是if呢？主要是防止为被唤醒的goroutine在返回Wait调用的时候，恰好有别的goroutine修改了conditon的值，所以需要使用for循环作为条件判断。

4.4.1 数据结构

1type Cond struct {
2    noCopy noCopy
3    L Locker
4    notify  notifyList
5    checker copyChecker
6}

Cond结构不允许拷贝，包含了Locker的接口字段，和一个notifyList的集合字段。

4.4.2 NewCond函数

1func NewCond(l Locker) *Cond {
2    return &Cond{L: l}
3}

实现Locker接口的类型都可以，一般为Mutex和RWMutex

4.4.3 Wait方法

1func (c *Cond) Wait() {
2    c.checker.check()
3    t := runtime_notifyListAdd(&c.notify)
4    c.L.Unlock()
5    runtime_notifyListWait(&c.notify, t)
6    c.L.Lock()
7}

在使用Wait方法之前，要调用c.L.Lock来进入临界区域，将当前等待的goroutine加入到通知队列中，然后调用c.L.Unlock()来退出临界区域，以便让其他goroutine可以进入等待区域。紧接着挂起goroutine，等待消息。

4.4.4 Singal方法

1func (c *Cond) Signal() {
2    c.checker.check()
3    runtime_notifyListNotifyOne(&c.notify)}

runtime_notifyListNotifyOne唤起其中的等待的goroutine。

4.4.5 Broadcast方法

1func (c *Cond) Broadcast() {
2    c.checker.check()
3    runtime_notifyListNotifyAll(&c.notify)
4}

runtime_notifyListNotifyAll唤起全部等待的goroutine。

版权申明：内容来源网络，版权归原创者所有。除非无法确认，我们都会标明作者及出处，如有侵权烦请告知，我们会立即删除并表示歉意。谢谢。

Golang语言社区

ID：Golangweb

www.bytedancing.com

游戏服务器架构丨分布式技术丨大数据丨游戏算法学习

以上是关于信号量，锁和 golang 相关源码分析的主要内容，如果未能解决你的问题，请参考以下文章

CountDownLatch源码分析

AQS基础源码分析

源码分析-Semaphor

Golang的Context介绍及其源码分析

Golang-ReverseProxy源码分析