Semaphore的原理和实现方法

本篇内容介绍了“Semaphore的原理和实现方法”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

Semaphore

数据结构

// Go 语言中暴露的 semaphore 实现 // 具体的用法是提供 sleep 和 wakeup 原语 // 以使其能够在其它同步原语中的竞争情况下使用 // 因此这里的 semaphore 和 linux 中的 futex 目标是一致的 // 只不过语义上更简单一些 // // 也就是说，不要认为这些是信号量 // 把这里的东西看作 sleep 和 wakeup 实现的一种方式 // 每一个 sleep 都会和一个 wakeup 配对 // 即使在发生 race 时，wakeup 在 sleep 之前时也是如此 // // See Mullender and Cox, ``Semaphores in Plan 9,'' // Http://swtch.com/semaphore.pdf  // 为 sync.Mutex 准备的异步信号量  // semaRoot 持有一棵 地址各不相同的 sudog(s.elem) 的平衡树 // 每一个 sudog 都反过来指向(通过 s.waitlink)一个在同一个地址上等待的其它 sudog 们 // 同一地址的 sudog 的内部列表上的操作时间复杂度都是 O(1)。顶层 semaRoot 列表的扫描 // 的时间复杂度是 O(log n)，n 是被哈希到同一个 semaRoot 的不同地址的总数，每一个地址上都会有一些 goroutine 被阻塞。 // 访问 golang.org/issue/17953 来查看一个在引入二级列表之前性能较差的程序样例，test/locklinear.go // 中有一个复现这个样例的测试 type semaRoot struct {     lock  mutex     treap *sudog // root of balanced tree of unique waiters.     nwait uint32 // Number of waiters. Read w/o the lock. }  // Prime to not correlate with any user patterns. const semTabSize = 251  var semtable [semTabSize]struct {     root semaRoot     pad  [sys.CacheLineSize - unsafe.Sizeof(semaRoot{})]byte }  func semroot(addr *uint32) *semaRoot {     return &semtable[(uintptr(unsafe.Pointer(addr))>>3)%semTabSize].root }

┌─────┬─────┬─────┬─────┬─────┬────────────────────────┬─────┐                  │  0  │  1  │  2  │  3  │  4  │         .....          │ 250 │                  └─────┴─────┴─────┴─────┴─────┴────────────────────────┴─────┘                     │                                                      │                        │                                                      │                        └──┐                                                   └─┐                         │                                                     │                         │                                                     │                         ▼                                                     ▼                    ┌─────────┐                                           ┌─────────┐               │ struct  │                                           │ struct  │               ├─────────┴─────────┐                                 ├─────────┴─────────┐     │   root semaRoot   │──┐                              │   root semaRoot   │──┐  ├───────────────────┤  │                              ├───────────────────┤  │  │        pad        │  │                              │        pad        │  │  └───────────────────┘  │                              └───────────────────┘  │                         │                                                     │        ┌────────────────┘                                    ┌────────────────┘        │                                                     │                         │                                                     │                         ▼                                                     ▼                   ┌──────────┐                                          ┌──────────┐              │ semaRoot │                                          │ semaRoot │              ├──────────┴────────┐                                 ├──────────┴────────┐     │    lock mutex     │                                 │    lock mutex     │     ├───────────────────┤                                 ├───────────────────┤     │   treap *sudog    │                                 │   treap *sudog    │     ├───────────────────┤                                 ├───────────────────┤     │   nwait uint32    │                                 │   nwait uint32    │     └───────────────────┘                                 └───────────────────┘

treap 结构:

                                 ┌──────────┐                                                                 ┌─┬─▶│  sudog   │                                                                 │ │  ├──────────┴────────────┐                              ┌─────────────────────┼─┼──│      prev *sudog      │                              │                     │ │  ├───────────────────────┤                              │                     │ │  │      next *sudog      │────┐                         │                     │ │  ├───────────────────────┤    │                         │                     │ │  │     parent *sudog     │    │                         │                     │ │  ├───────────────────────┤    │                         │                     │ │  │  elem unsafe.Pointer  │    │                         │                     │ │  ├───────────────────────┤    │                         │                     │ │  │     ticket uint32     │    │                         │                     │ │  └───────────────────────┘    │                         │                     │ │                               │                         │                     │ │                               │                         │                     │ │                               │                         │                     │ │                               │                         │                     │ │                               │                         │                     │ │                               │                         ▼                     │ │                               ▼                   ┌──────────┐                │ │                         ┌──────────┐              │  sudog   │                │ │                         │  sudog   │              ├──────────┴────────────┐   │ │                         ├──────────┴────────────┐ │      prev *sudog      │   │ │                         │      prev *sudog      │ ├───────────────────────┤   │ │                         ├───────────────────────┤ │      next *sudog      │   │ │                         │      next *sudog      │ ├───────────────────────┤   │ │                         ├───────────────────────┤ │     parent *sudog     │───┘ └─────────────────────────│     parent *sudog     │ ├───────────────────────┤                               ├───────────────────────┤ │  elem unsafe.Pointer  │                               │  elem unsafe.Pointer  │ ├───────────────────────┤                               ├───────────────────────┤ │     ticket uint32     │                               │     ticket uint32     │ └───────────────────────┘                               └───────────────────────┘

在这个 treap 结构里，从 elem 的视角(其实就是 lock 的 addr)来看，这个结构是个二叉搜索树。从 ticket  的角度来看，整个结构就是一个小顶堆。

所以才叫树堆(treap)。

相同 addr，即对同一个 mutex 上锁的 g，会阻塞在同一个地址上。这些阻塞在同一个地址上的 goroutine 会被打包成  sudog，组成一个链表。用 sudog 的 waitlink 相连:

┌──────────┐                         ┌──────────┐                          ┌──────────┐              │  sudog   │                  ┌─────▶│  sudog   │                   ┌─────▶│  sudog   │              ├──────────┴────────────┐     │      ├──────────┴────────────┐      │      ├──────────┴────────────┐ │    waitlink *sudog    │─────┘      │    waitlink *sudog    │──────┘      │    waitlink *sudog    │ ├───────────────────────┤            ├───────────────────────┤             ├───────────────────────┤ │    waittail *sudog    │            │    waittail *sudog    │             │    waittail *sudog    │ └───────────────────────┘            └───────────────────────┘             └───────────────────────┘

中间的元素的 waittail 都会指向最后一个元素:

┌──────────┐                                                                                            │  sudog   │                                                                                            ├──────────┴────────────┐                                                                               │    waitlink *sudog    │                                                                               ├───────────────────────┤                                                                               │    waittail *sudog    │───────────────────────────────────────────────────────────┐                   └───────────────────────┘                                                           │                                                     ┌──────────┐                                      │                                                     │  sudog   │                                      │                                                     ├──────────┴────────────┐                         │                                                     │    waitlink *sudog    │                         │                                                     ├───────────────────────┤                         │                                                     │    waittail *sudog    │─────────────────────────┤                                                     └───────────────────────┘                         ▼                                                                                                 ┌──────────┐                                                                                            │  sudog   │                                                                                            ├──────────┴────────────┐                                                                               │    waitlink *sudog    │                                                                               ├───────────────────────┤                                                                               │    waittail *sudog    │                                                                               └───────────────────────┘

对外封装

在 sema.go 里实现的内容，用 go:linkname 导出给 sync、poll 库来使用，也是在链接期做了些手脚:

//go:linkname sync_runtime_SeMacquire sync.runtime_Semacquire func sync_runtime_Semacquire(addr *uint32) {     semacquire1(addr, false, semaBlockProfile) }  //go:linkname poll_runtime_Semacquire internal/poll.runtime_Semacquire func poll_runtime_Semacquire(addr *uint32) {     semacquire1(addr, false, semaBlockProfile) }  //go:linkname sync_runtime_Semrelease sync.runtime_Semrelease func sync_runtime_Semrelease(addr *uint32, handoff bool) {     semrelease1(addr, handoff) }  //go:linkname sync_runtime_SemacquireMutex sync.runtime_SemacquireMutex func sync_runtime_SemacquireMutex(addr *uint32, lifo bool) {     semacquire1(addr, lifo, semaBlockProfile|semaMutexProfile) }  //go:linkname poll_runtime_Semrelease internal/poll.runtime_Semrelease func poll_runtime_Semrelease(addr *uint32) {     semrelease(addr) }

实现

sem 本身支持 acquire 和 release，其实就是 OS 里常说的 P 操作和 V 操作。

公共部分

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

acquire 过程

type semaProfileFlags int  const (     semaBlockProfile semaProfileFlags = 1 << iota     semaMutexProfile )  // Called from runtime. func semacquire(addr *uint32) {     semacquire1(addr, false, 0) }  func semacquire1(addr *uint32, lifo bool, profile semaProfileFlags) {     gp := getg()     if gp != gp.m.curg {         throw("semacquire not on the G stack")     }      // 低成本的情况     if cansemacquire(addr) {         return     }      // 高成本的情况:     //    增加 waiter count 的值     //    再尝试调用一次 cansemacquire，成本了就直接返回     //    没成功就把自己作为一个 waiter 入队     //    sleep     //    (之后 waiter 的 descriptor 被 signaler 用 dequeue 踢出)     s := acquireSudog()     root := semroot(addr)     t0 := int64(0)     s.releasetime = 0     s.acquiretime = 0     s.ticket = 0      for {         lock(&root.lock)         // 给 nwait 加一，这样后来的就不会在 semrelease 中进低成本的路径了         atomic.Xadd(&root.nwait, 1)         // 检查 cansemacquire 避免错过了唤醒         if cansemacquire(addr) {             atomic.Xadd(&root.nwait, -1)             unlock(&root.lock)             break         }         // 在 cansemacquire 之后的 semrelease 都可以知道我们正在等待         // (上面设置了 nwait)，所以会直接进入 sleep         // 注: 这里说的 sleep 其实就是 goparkunlock         root.queue(addr, s, lifo)         goparkunlock(&root.lock, "semacquire", traceEvGoBlockSync, 4)         if s.ticket != 0 || cansemacquire(addr) {             break         }     }     if s.releasetime > 0 {         blockevent(s.releasetime-t0, 3)     }     releaseSudog(s) }

release 过程

func semrelease(addr *uint32) {     semrelease1(addr, false) }  func semrelease1(addr *uint32, handoff bool) {     root := semroot(addr)     atomic.Xadd(addr, 1)      // 低成本情况: 没有 waiter?     // 这个 atomic 的检查必须发生在 xadd 之前，以避免错误唤醒     // (具体参见 semacquire 中的循环)     if atomic.Load(&root.nwait) == 0 {         return     }      // 高成本情况: 搜索 waiter 并唤醒它     lock(&root.lock)     if atomic.Load(&root.nwait) == 0 {         // count 值已经被另一个 goroutine 消费了         // 所以我们不需要唤醒其它 goroutine 了         unlock(&root.lock)         return     }     s, t0 := root.dequeue(addr)     if s != nil {         atomic.Xadd(&root.nwait, -1)     }     unlock(&root.lock)     if s != nil { // 可能会很慢，所以先解锁         acquiretime := s.acquiretime         if acquiretime != 0 {             mutexevent(t0-acquiretime, 3)         }         if s.ticket != 0 {             throw("corrupted semaphore ticket")         }         if handoff && cansemacquire(addr) {             s.ticket = 1         }         readyWithTime(s, 5)     } }  func readyWithTime(s *sudog, traceskip int) {     if s.releasetime != 0 {         s.releasetime = cputicks()     }     goready(s.g, traceskip) }

treap 结构

sudog 按照地址 hash 到 251 个 bucket 中的其中一个，每一个 bucket 都是一棵 treap。而相同 addr 上的 sudog  会形成一个链表。

为啥同一个地址的 sudog 不需要展开放在 treap 中呢?显然，sudog 唤醒的时候，block 在同一个 addr 上的  goroutine，说明都是加的同一把锁，这些 goroutine 被唤醒肯定是一起被唤醒的，相同地址的 g  并不需要查找才能找到，只要决定是先进队列的被唤醒(fifo)还是后进队列的被唤醒(lifo)就可以了。

// queue 函数会把 s 添加到 semaRoot 上阻塞的 goroutine 们中 // 实际上就是把 s 添加到其地址对应的 treap 上 func (root *semaRoot) queue(addr *uint32, s *sudog, lifo bool) {     s.g = getg()     s.elem = unsafe.Pointer(addr)     s.next = nil     s.prev = nil      var last *sudog     pt := &root.treap     for t := *pt; t != nil; t = *pt {         if t.elem == unsafe.Pointer(addr) {             // Already have addr in list.             if lifo {                 // treap 中在 t 的位置用 s 覆盖掉 t                 *pt = s                 s.ticket = t.ticket                 s.acquiretime = t.acquiretime                 s.parent = t.parent                 s.prev = t.prev                 s.next = t.next                 if s.prev != nil {                     s.prev.parent = s                 }                 if s.next != nil {                     s.next.parent = s                 }                 // 把 t 放在 s 的 wait list 的第一个位置                 s.waitlink = t                 s.waittail = t.waittail                 if s.waittail == nil {                     s.waittail = t                 }                 t.parent = nil                 t.prev = nil                 t.next = nil                 t.waittail = nil             } else {                 // 把 s 添加到 t 的等待列表的末尾                 if t.waittail == nil {                     t.waitlink = s                 } else {                     t.waittail.waitlink = s                 }                 t.waittail = s                 s.waitlink = nil             }             return         }         last = t         if uintptr(unsafe.Pointer(addr)) < uintptr(t.elem) {             pt = &t.prev         } else {             pt = &t.next         }     }      // 把 s 作为树的新的叶子插入进去     // 平衡树使用 ticket 作为堆的权重值，这个 ticket 是随机生成的     // 也就是说，这个结构以元素地址来看的话，是一个二叉搜索树     // 同时用 ticket 值使其同时又是一个小顶堆，满足     // s.ticket <= both s.prev.ticket and s.next.ticket.     // https://en.wikipedia.org/wiki/Treap     // http://faculty.washington.edu/aragon/pubs/rst89.pdf     //     // s.ticket 会在一些地方和 0 相比，因此只设置最低位的 bit     // 这样不会明显地影响 treap 的质量?     s.ticket = fastrand() | 1     s.parent = last     *pt = s      // 按照 ticket 来进行旋转，以满足 treap 的性质     for s.parent != nil && s.parent.ticket > s.ticket {         if s.parent.prev == s {             root.rotateRight(s.parent)         } else {             if s.parent.next != s {                 panic("semaRoot queue")             }             root.rotateLeft(s.parent)         }     } }  // dequeue 会搜索到阻塞在 addr 地址的 semaRoot 中的第一个 goroutine // 如果这个 sudog 需要进行 profile，dequeue 会返回它被唤醒的时间(now)，否则的话 now 为 0 func (root *semaRoot) dequeue(addr *uint32) (found *sudog, now int64) {     ps := &root.treap     s := *ps     for ; s != nil; s = *ps {         if s.elem == unsafe.Pointer(addr) {             goto Found         }         if uintptr(unsafe.Pointer(addr)) < uintptr(s.elem) {             ps = &s.prev         } else {             ps = &s.next         }     }     return nil, 0  Found:     now = int64(0)     if s.acquiretime != 0 {         now = cputicks()     }     if t := s.waitlink; t != nil {         // 替换掉同样在 addr 上等待的 t。         *ps = t         t.ticket = s.ticket         t.parent = s.parent         t.prev = s.prev         if t.prev != nil {             t.prev.parent = t         }         t.next = s.next         if t.next != nil {             t.next.parent = t         }         if t.waitlink != nil {             t.waittail = s.waittail         } else {             t.waittail = nil         }         t.acquiretime = now         s.waitlink = nil         s.waittail = nil     } else {         // 向下旋转 s 到叶节点，以进行删除，同时要考虑优先级         for s.next != nil || s.prev != nil {             if s.next == nil || s.prev != nil && s.prev.ticket < s.next.ticket {                 root.rotateRight(s)             } else {                 root.rotateLeft(s)             }         }         // Remove s, now a leaf.         // 删除 s，现在是叶子节点了         if s.parent != nil {             if s.parent.prev == s {                 s.parent.prev = nil             } else {                 s.parent.next = nil             }         } else {             root.treap = nil         }     }     s.parent = nil     s.elem = nil     s.next = nil     s.prev = nil     s.ticket = 0     return s, now }

“Semaphore的原理和实现方法”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注编程网网站，小编将为大家输出更多高质量的实用文章！