一、背景

sync.Mutex里面用了runtime_SemacquireMutex和runtime_Semrelease，所以看下这个runtime的信号量是如何实现的。

二、基础知识

2.1 信号量

信号量（英语：semaphore）又称为信号标，是一个同步对象，用于保持在0至指定最大值之间的一个计数值。当线程完成一次对该semaphore对象的等待（wait）时，该计数值减一；当线程完成一次对semaphore对象的释放（release）时，计数值加一。当计数值为0，则线程等待该semaphore对象不再能成功直至该semaphore对象变成signaled状态。semaphore对象的计数值大于0，为signaled状态；计数值等于0，为nonsignaled状态。

信号量的概念是由荷兰计算机科学家艾兹赫尔·戴克斯特拉（Edsger W. Dijkstra）发明的，广泛的应用于不同的操作系统中。在系统中，给予每一个进程一个信号量，代表每个进程目前的状态，未得到控制权的进程会在特定地方被强迫停下来，等待可以继续进行的信号到来。如果信号量是一个任意的整数，通常被称为计数信号量（Counting semaphore），或一般信号量（general semaphore）；如果信号量只有二进制的0或1，称为二进制信号量（binary semaphore）。

计数信号量具备两种操作动作，称为V（signal()）与P（wait()）（即部分参考书常称的PV操作）。V操作会增加信号标S的数值，P操作会减少它。

  • P原语：P是荷兰语Proberen(测试)的首字母。为阻塞原语，负责把当前进程由运行状态转换为阻塞状态，直到另外一个进程唤醒它。操作为：申请一个空闲资源(把信号量减1)，若成功，则退出；若失败，则该进程被阻塞；
  • V原语：V是荷兰语Verhogen(增加)的首字母。为唤醒原语，负责把一个被阻塞的进程唤醒，它有一个参数表，存放着等待被唤醒的进程信息。操作为：释放一个被占用的资源(把信号量加1)，如果发现有被阻塞的进程，则选择一个唤醒之。

Semaphore - wiki

2.2 Treap

Treap是Binary Search Tree+Heap的组合。

二叉查找树（Binary Search Tree），它或者是一棵空树，或者是具有下列性质的二叉树：若它的左子树不空，则左子树上所有结点的值均小于它的根结点的值；若它的右子树不空，则右子树上所有结点的值均大于它的根结点的值； 它的左、右子树也分别为二叉排序树。

堆有大顶堆和小顶堆：

  • 大顶堆:每个节点的值都大于或者等于他的左右孩子节点的值。
  • 小顶堆:每个结点的值都小于或等于其左孩子和右孩子结点的值。

Treap既是一棵二叉查找树，也是一个二叉堆。但是这两种数据结构貌是矛盾的存在，如果是二叉查找树，就不能是一个堆，如果是一个堆，那么必然不是二叉查找树。

所以Treap用了一个很巧妙的方式解决这个问题：给每个键值一个随机附加的优先级，让键值满足二叉查找树的结构，让优先级满足二叉堆的结构。

Treap它的最大优点就是实现简单，没有太多复杂的操作，但是我们前面也说了，它是通过随机的priority来控制树的平衡的，那么它显然无法做到完美平衡，只能做到不落入最坏的情况。

image.png

Treap——堆和二叉树的完美结合，性价比极值的搜索树

2.3 x/sync/semaphore

Go的 X-Repositories 提供了一种带权重的信号量实现方式 sync.semaphore。这个跟runtime.semaphore其实没太大关系。主要是提供了个high-level的信号量给Go开发者使用。实现方式如下：

type Weighted struct {
    size    int64 // 资源的总数，Acquire(n) 的时候会消耗这个资源，
    cur     int64 // 当前已申请资源数，Acquire(n)成功的话，cur=cur+n
    mu      sync.Mutex // 互斥锁，所有Acquire、Release 都要加锁
    waiters list.List // 阻塞的队列
}

type waiter struct {
    n     int64
    ready chan<- struct{} // 使用 channle 来通信
}

Weighted就是一个权重的信号量，主要提供Acquire(n)和Release(n)两个操作。实现逻辑比较简单。

Acquire(n)申请n个资源，申请成功的话会设置cur=cur+n，如果没有资源可以申请了，会new一个waiter，然后把这个waiter加到waiters这个等待的队列中，并阻塞在waiter.ready的读上面。

Release(n)释放n个资源，然后设置cur=cur-n，在waiters这个等待的队列中，循环取出取waiters头部的waiter（直到s.size-s.cur < w.n终止），调用close(waiter.ready)，这样阻塞在waiter.ready读上的goroutine会被唤醒。

下面写了个简单Demo：

func main() {
    var (
        wg  sync.WaitGroup
        ctx = context.Background()
    )

    cpuNum := runtime.GOMAXPROCS(0)
    sem := semaphore.NewWeighted(int64(cpuNum)) // 设置 goroutine 最大并发数 = cpuNum
    for i := 0; i < 100; i++ {
        wg.Add(1)
        err := sem.Acquire(ctx, 1)
        if err != nil {
            panic(err)
        }

        go func(i int) {
            fmt.Println("ng: ", runtime.NumGoroutine(), " i = ", i)
            sem.Release(1)
            wg.Done()
        }(i)
    }
    wg.Wait()
}

2.4 sudog

sudog表示为了一个等待队列中的goroutine，比如因为读写channel阻塞，或者Lock导致阻塞（Lock底层其实就是semaphore）等的goroutine，由于这些阻塞的goroutine不确定什么时候能被调度（取消阻塞比如unlock、读写channel），所以这种阻塞的goroutine不适合一直放在p的本地运行队列中，这个时候会把阻塞的gorutine打包成一个sudog里面会有一些这个g运行的上下文。然后存到另外一个地方，比如channel是存在 recvq 和 sendq 中，而阻塞在信号量上的goroutine是存在 semTable 中。

具体sudog的结构体如下：

// sudog 代表一个处于等待队列的g，比如阻塞在读写 channle上 的 goroutine
// sduog 和 g 是多对多的关系，一个 g 可以在多个 wait lists 上，所以一个 g 可以对应多个 sduog
// sudogs 会有两级缓存，优先去p的sudogcache取，取不到则去全局的sudogcache取一批（直到本地容量达到50%）
// sudogs 通过 acquireSudog 和 releaseSudog 去申请或释放
type sudog struct {
    // 在channel的场景中sudog中所有字段都受hchan.lock保护
    g *g

    next *sudog // 双向链表，指向下一个 sduog
    prev *sudog // 双向链表，指向上一个 sduog
    
    // channel场景存的是，读写的数据。
    // semaphore 场景存的是信号量的地址。
    elem unsafe.Pointer 

    // 下面这些字段不会被并发访问
    // For channels, waitlink is only accessed by g.
    // For semaphores, 所有字段需要拿到semaRoot的lock才能访问

    acquiretime int64 // semaphore 场景使用的，记录获取信号量时间
    releasetime int64 // 释放时间
    ticket      uint32 // treap 里面堆用的随机的权重

    // 只在select 场景使用的字段，表明当前g是否被选中，然后唤醒
    isSelect bool

    // 只在channel场景使用的
    success bool

    // 只在 semaphore 场景使用
    parent   *sudog // semaRoot binary tree
    waitlink *sudog // semaRoot 节点对应的等待列表
    waittail *sudog // semaRoot 等待列表的尾部节点
    c        *hchan // 只在channel场景使用的，关联的channel
}

从sodug结构体看出来，sudog里面的字段分别跟channel、semaphore、select几个场景有关。某些字段只有在特点场景才会用到，感觉全部都耦合在一个Struct不够优雅。

sudog on channel

2.5 semtable

semTable 是一个长度为251的全局数组，每个semaRoot指向一个treap，主要用于存放阻塞在信号量(semaphore)上的sudog

var semtable [semTabSize]struct {
    root semaRoot
    pad  [cpu.CacheLinePadSize - unsafe.Sizeof(semaRoot{})]byte // 防止 flase sharing
}

semaRoot最早是双向链表，在某些场景下性能比较查，所以优化成了treap，具体可以看 CR37103

优化之后semtable存储的结构大概是这样：

image.png

2.6 mutex

runtime包里面的锁都是使用的runtime内部实现的mutex，具体是使用CAS+futex来实现的。更多详见Go源码——runtime.mutex

三、semaphore 源码分析

semaphore基本操作，在 src/sync/runtime.go 定义了下面几个方法：

// Semacquire等待*s > 0，然后原子递减它。
// 它是一个简单的睡眠原语，用于同步库使用。
func runtime_Semacquire(s *uint32)

// SemacquireMutex类似于Semacquire,用来阻塞互斥的对象
// 如果lifo为true，waiter将会被插入到队列的头部
// skipframes是跟踪过程中要省略的帧数
func runtime_SemacquireMutex(s *uint32, lifo bool, skipframes int)

// Semrelease会自动增加*s并通知一个被Semacquire阻塞的等待的goroutine
// 它是一个简单的唤醒原语，用于同步库
// 如果handoff为true, 传递信号到队列头部的waiter
// skipframes是跟踪过程中要省略的帧数，从这里开始计算
func runtime_Semrelease(s *uint32, handoff bool, skipframes int)

这个几个函数具体的实现在 src/runtime/sema.go。

//go:linkname sync_runtime_Semacquire sync.runtime_Semacquire
func sync_runtime_Semacquire(addr *uint32) {
    semacquire1(addr, false, semaBlockProfile, 0)
}

//go:linkname poll_runtime_Semacquire internal/poll.runtime_Semacquire
func poll_runtime_Semacquire(addr *uint32) {
    semacquire1(addr, false, semaBlockProfile, 0)
}

//go:linkname sync_runtime_Semrelease sync.runtime_Semrelease
func sync_runtime_Semrelease(addr *uint32, handoff bool, skipframes int) {
    semrelease1(addr, handoff, skipframes)
}

//go:linkname sync_runtime_SemacquireMutex sync.runtime_SemacquireMutex
func sync_runtime_SemacquireMutex(addr *uint32, lifo bool, skipframes int) {
    semacquire1(addr, lifo, semaBlockProfile|semaMutexProfile, skipframes)
}

//go:linkname poll_runtime_Semrelease internal/poll.runtime_Semrelease
func poll_runtime_Semrelease(addr *uint32) {
    semrelease(addr)
}

3.1 获取信号量

semaphore获取信号量操作步骤如下：

  1. 调用runtime_SemacquireMutex （比如sync.Mutex.Lock()场景）
  2. sync_runtime_SemacquireMutex
  3. semacquire1
  4. CAS(addr, v, v-1)状态成功就返回，失败继续往下
  5. 缓存池拿一个sudog，或者new一个sudog（acquireSudog）
  6. 把g相关的数据存到sudog中。
  7. 循环
    • 对当前semaRoot加锁
    • nwait++
    • cansemacquire/CAS(addr, v, v-1)
    • sudog加到semaRoot的treap中/root.queue()
    • 可能要调整树的结构（左旋rotateRight/右旋rotateLeft）防止树退化为链表
    • goparkunlock让出当前g的执行
    • 被唤醒
    • CAS成功或者s.ticket != 0（当前没有其他竞争者了） 认为成功
    • 否则继续循环
  8. 最后释放sudog/releaseSudog

具体源码如下：

func semacquire1(addr *uint32, lifo bool, profile semaProfileFlags, skipframes int) {
    gp := getg()
    if gp != gp.m.curg {// 判断下g是不是当前m绑定的g
        throw("semacquire not on the G stack")
    }

    // CAS(addr, v, v-1) 成功就直接成功否则一直循环，如果 *addr = 0 返回 false 走下面 slowpath
    if cansemacquire(addr) {
        return
    }
    
    // 走到这里表示当前g要阻塞
    // 下面逻辑，就是把g封装成sudog,然后存到semTable中。
    // 最后调用 gopark 让出当前g


    s := acquireSudog() // 这个先去 p.sudogcache 拿，没拿到去全局sudohgcache拿
    root := semroot(addr) // 根据sema的地址，算出用到semTable中哪个semaRoot
    t0 := int64(0)
    s.releasetime = 0
    s.acquiretime = 0
    s.ticket = 0
    if profile&semaBlockProfile != 0 && blockprofilerate > 0 {
        t0 = cputicks()
        s.releasetime = -1
    }
    if profile&semaMutexProfile != 0 && mutexprofilerate > 0 {
        if t0 == 0 {
            t0 = cputicks()
        }
        s.acquiretime = t0
    }
    for {
        lockWithRank(&root.lock, lockRankRoot) // 加锁，方面下面修改 semaRoot的属性
        // 对等待的计数加1，这样sema_release时候不会走快路径
        atomic.Xadd(&root.nwait, 1)
        // 看下是否有其他的goroutine调用了sema_release
        // 在尝试 CAS(addr, v, v-1) 试下
        if cansemacquire(addr) {
            atomic.Xadd(&root.nwait, -1)
            unlock(&root.lock)
            break
        }
        
        // 这里，就是这个新的 sudog 加到 semaTable中的
        root.queue(addr, s, lifo)
        goparkunlock(&root.lock, waitReasonSemacquire, traceEvGoBlockSync, 4+skipframes) // 这你会让出当前的goroutine
        
        
        // goroutine 被调度回来了，表示有 sema_release 以后唤醒了这个 sema
        // s.ticket != 0 表示是等待队列头部的 sudog，当前队列只有一个sudog了，所以直接结束
        // CAS(addr, v, v-1) 成功也结束
        if s.ticket != 0 || cansemacquire(addr) {
            break
        }
    }
    if s.releasetime > 0 {
        blockevent(s.releasetime-t0, 3+skipframes)
    }
    releaseSudog(s) // 释放 sudog
}


func cansemacquire(addr *uint32) bool {
    for {
        v := atomic.Load(addr)
        if v == 0 {
            return false
        }
        if atomic.Cas(addr, v, v-1) {
            return true
        }
    }
}

func acquireSudog() *sudog {
    // 设置禁止抢占
    mp := acquirem()
    pp := mp.p.ptr()
    //当前本地sudog缓存没有了，则去全局缓存中拉取一批
    if len(pp.sudogcache) == 0 {
        lock(&sched.sudoglock)
        // 首先尝试从全局缓存中获取sudog，直到本地容量达到50%
        for len(pp.sudogcache) < cap(pp.sudogcache)/2 && sched.sudogcache != nil {
            s := sched.sudogcache
            sched.sudogcache = s.next
            s.next = nil
            pp.sudogcache = append(pp.sudogcache, s)
        }
        unlock(&sched.sudoglock)
        // 如果全局缓存为空，则分配创建一个新的sudog
        if len(pp.sudogcache) == 0 {
            pp.sudogcache = append(pp.sudogcache, new(sudog))
        }
    }
    n := len(pp.sudogcache)
    s := pp.sudogcache[n-1]
    pp.sudogcache[n-1] = nil
    pp.sudogcache = pp.sudogcache[:n-1]
    if s.elem != nil {
        throw("acquireSudog: found s.elem != nil in cache")
    }
    //解除抢占限制
    releasem(mp)
    return s
}

3.2 释放信号量

semaphore释放信号量操作步骤如下：

  1. 调用runtime_Semrelease，比如sync.Mutex.Unlock()场景。
  2. sync_runtime_Semrelease
  3. semrelease1
  4. 原子*addr++
  5. nwait=0，表示没有阻塞在这个信号量上的g直接返回。
  6. 有阻塞的g在semTable中找到对应的semaRoot，然后对semaRoot`加锁。
  7. 再次check下nwait=0，等于0直接返回。
  8. 拿到sema的addres在semTable中对应的队列头部的seamRoot。
  9. dequeue是否需要调整左旋rotateLeft或者右旋rotateRight调整树结构。
  10. readyWithTime，调用goread唤醒sudog绑定的g。
  11. goyield

semrelease源码如下：

func semrelease1(addr *uint32, handoff bool, skipframes int) {
    root := semroot(addr)
    atomic.Xadd(addr, 1)

    // 没有等待者直接返回
    if atomic.Load(&root.nwait) == 0 {
        return
    }

    //查找一个等待着并唤醒它
    lockWithRank(&root.lock, lockRankRoot)
    if atomic.Load(&root.nwait) == 0 {
        //计数已经被其他goroutine消费，所以不需要唤醒其他goroutine
        unlock(&root.lock)
        return
    }
    s, t0 := root.dequeue(addr) //查找第一个出现的addr
    if s != nil {
        atomic.Xadd(&root.nwait, -1)
    }
    unlock(&root.lock)
    if s != nil { // 可能比较慢 甚至被挂起所以先unlock
        acquiretime := s.acquiretime
        if acquiretime != 0 {
            mutexevent(t0-acquiretime, 3+skipframes)
        }
        if s.ticket != 0 {
            throw("corrupted semaphore ticket")
        }
        if handoff && cansemacquire(addr) {
            s.ticket = 1
        }
        //goready(s.g,5)标记runnable 等待被重新调度

        readyWithTime(s, 5+skipframes)
        if s.ticket == 1 && getg().m.locks == 0 {
            // 直接切换G
            // readyWithTime已经将等待的G作为runnext放到当前的P
            // 我们现在调用调度器可以立即执行等待的G
            // 注意waiter继承了我们的时间片：这是希望避免在P上无限得进行激烈的信号量竞争
            // goyield类似于Gosched，但是它是发送“被强占”的跟踪事件，更重要的是，将当前G放在本地runq
            // 而不是全局队列。
            // 我们仅在饥饿状态下执行此操作(handoff=true),因为非饥饿状态下，当我们yielding/scheduling时，
            // 其他waiter可能会获得信号量，这将是浪费的。我们等待进入饥饿状体，然后开始进行ticket和P的手递手交接
            // See issue 33747 for discussion.
            // https://go-review.googlesource.com/c/go/+/206180
            goyield()
        }
    }
}

四、总结

获取信号量操作主要尝试把sema地址CAS方式原子减1，成就直接返回，失败以后会把当前g打包成sudog然后保存到semTable，然后调用gopark让出当前的goroutine。

释放信号量操作就是吧sema地址加1，然后看有没有等待中的g，没有直接返回，有的话去semaTable的等待队列取出然后调用goready唤醒对应的g。

主要理解semaTable里面存储sudog的方式就好了。

参考资料

手摸手Go 并发编程基建Semaphore

一文读懂go中semaphore(信号量)源码