手摸手Go 并发编程基建Semaphore

用户3904122

发布于 2022-06-29 14:52:02

4631

发布于 2022-06-29 14:52:02

I leave uncultivated today, was precisely yesterday perishes tomorrow which person of the body implored。

前段时间剖析了一下sync.Mutex的源码，发现底层实现其实是基于Go的运行时Semaphore机制来实现的，虽然那篇文章也梳理了一下关于信号量的原理，但是感觉还是有些浅尝辄止，而且手摸手Go 并发包系列后面还打算写下sync.RWMutex所以索性这次彻底来搞清楚Semaphore。

sema.go中提供了Go语言中暴露的Semaphore实现，预期使用是在其他同步原语竞争情况下提供sleep和wakeup原语。因此它跟Linux的futex目标一致，只不过这里的语义更简单一些。也就是说不要将他们认为是信号量。把他们看作是一种实现sleep和wakeup原语的方式。这样，sleep和wakeup是成对出现，即使因为竞争原因，wakeup发生在sleep之前也是这样。翻看源码前，让我们先来搞清楚它的数据结构。

数据结构

sema.go中定义了一个全局变量，semtable数组。小为251，元素为一个匿名结构体。这里为了避免伪共享问题做了一下内存填充。

// Prime to not correlate with any user patterns.
const semTabSize = 251

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

每个元素持有的semaRoot为这个数据结构的核心。

// 为sync.Mutex准备的异步信号量
// golang.org/issue/17953 可以查看引入二级列表之前性能较差的程序示例test/locklinear.go
type semaRoot struct {
 lock  mutex
 treap *sudog // 平衡树的根节点
 nwait uint32 // Number of waiters. Read w/o the lock.
}

semaRoot的结构看上去并不复杂，每个semaRoot持有一个具有不同地址(sudog.elem)的sudog平衡树，每个sudog都可以通过s.waitlink依次指向一个相同地址等待的sudog列表，在具有相同等待地址的sudog内部列表上的操作时间复杂度都是O(1)。顶层semaRoot列表的扫描为O(logn),其中n是阻止goroutines的不同信号量地址的数量。既然节点都是sudog,那它是如何定义的？

type sudog struct {
 g *g
 next *sudog
 prev *sudog
 elem unsafe.Pointer //数据元素 (可能指向栈)
 // 下面的字段不会并发访问
 // 对于channels, waitlink 只被g访问
  // 对于semaphores, 所有自动(包括上面的)只有获取semaRoot的锁才能被访问
 acquiretime int64
 releasetime int64
 ticket      uint32
  //isSelect表示g正在参与一个select，因此必须对g.selectDone进行CAS才能赢得唤醒竞争
 isSelect bool
  //success表示channel c上的通信是否成功。如果goroutine因为在通道c上传递了一个值而被唤醒，则为true；
  //如果因为channel c关闭而被唤醒，则为false
 success bool
  
 parent   *sudog // semaRoot binary tree
 waitlink *sudog // g.waiting list or semaRoot
 waittail *sudog // semaRoot
 c        *hchan // channel
}

这里可能就涉及到了Go的运行时调度的知识

sudog是对goroutine的一种封装，比如当你使用channel时，goroutine在sending/receiving阻塞时是被封装成sudog放进阻塞队列进行等待。sudog是必需的，因为g和同步对象的关系是多对多的。一个g可以出现在许多等待列表上，因此一个g可能有很多个sudog。并且许多g可能正在等待同一个同步对象，因此一个对象可能有许多sudog sudog是从一个特殊的pool中分配。使用acquireSudog和releaseSudog来分配和释放他们。

其中的next、prev、parent字段构成了平衡树，waitlink和waittail构成了相同信号量地址的链表结构。

千言万语不如来张图:

sema

之前分析过sync.Mutex主要依赖runtime_SemacquireMutex和runtime_Semrelease对应于运行时的sync_runtime_SemacquireMutex和sync_runtime_Semrelease。那接下来我们细细剖析一下

源码分析

func runtime_Semacquire(s *uint32)
func runtime_SemacquireMutex(s *uint32, lifo bool, skipframes int)
func runtime_Semrelease(s *uint32, handoff bool, skipframes int)

runtime_Semacquire 等待直到*s >0然后以原子的方法将其递减。它旨在作为一个简单的睡眠原语供同步库使用，但不要直接使用它
runtime_SemacquireMutex 类似于Semacquire，但是用于分析竞争的互斥对象。如果lifo为true,表示等待队列采用先进先出的模式，将等待者排在队列头部。skipframes表示从runtime_SemacquireMutex的调用者开始计算跟踪期间要忽略的帧数。
runtime_Semrelease 自动递增*s并通知等待在Semacquire上的goroutine。它旨在作为一个简单的唤醒语义供同步库使用，也不要直接使用它。如果handoff为true,则将计数直接传递给第一个等待者。skipframes表示从runtime_Semrelease的调用者开始计算跟踪期间要忽略的帧数。

但是这里只有方法声明，实际的代码实现部分，利用//go:linkname编译器指令转移到了sema.go`文件中，主要有如下方法

//go:linkname sync_runtime_Semacquire sync.runtime_Semacquire
func sync_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_Semacquire internal/poll.runtime_Semacquire
func poll_runtime_Semacquire(addr *uint32) {
 semacquire1(addr, false, semaBlockProfile, 0)
}
//go:linkname poll_runtime_Semrelease internal/poll.runtime_Semrelease
func poll_runtime_Semrelease(addr *uint32) {
 semrelease(addr)
}

sync_xxx主要用于支持并发包的实现，poll_xxx我目前看到的主要是用于管理fd的生命周期并且顺序访问Read、Write、Close操作。

操作方法

runtime_SemacquireMutex

sync_runtime_SemacquireMutex主要为sync.Mutex服务，实际调用semacquire1方法,实际sync_runtime_Semacquire、poll_runtime_Semacquire也都是调用semacquire1来实现。

func semacquire1(addr *uint32, lifo bool, profile semaProfileFlags, skipframes int) {
 gp := getg()
 if gp != gp.m.curg {
  throw("semacquire not on the G stack")
 }

 // 检查信号量大于0且CAS成功则直接返回
 if cansemacquire(addr) {
  return
 }

 // Harder case:
 // 增加等待者计数
 // 再次尝试cansemacquire 如果成功则返回
 // 将自己作为等待者入队
 // 休眠
 // (waiter descriptor is dequeued by signaler)
 s := acquireSudog() //获取一个sudog对象
 root := semroot(addr) //根据信号量地址hash到semtable中
 t0 := int64(0)
 s.releasetime = 0
 s.acquiretime = 0
 s.ticket = 0
 ... ...
 for {
  lockWithRank(&root.lock, lockRankRoot)
  // 将自己添加到nwait中来禁止semrelease中的easy case
  atomic.Xadd(&root.nwait, 1)
  // 检查cansemacquire 避免错过唤醒
  if cansemacquire(addr) {
   atomic.Xadd(&root.nwait, -1)
   unlock(&root.lock)
   break
  }
  // cansemacquire之后的所有semrelease都知道我们正在等待
  // (我们上面已经设置了nwait),所以进入休眠
  root.queue(addr, s, lifo)
  goparkunlock(&root.lock, waitReasonSemacquire, traceEvGoBlockSync, 4+skipframes)
  if s.ticket != 0 || cansemacquire(addr) {
   break
  }
 }
 if s.releasetime > 0 {
  blockevent(s.releasetime-t0, 3+skipframes)
 }
 releaseSudog(s)
}

大致逻辑：

获取当前的g并判断是否跟m上实际运行的g是否一致
循环判断信号量的值，若等于0，则直接返回false进入harder case;否则原子操作*addr -= 1成功则相当于拿到信号量直接返回

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

从当前P中获取sudog,用于保存需要阻塞的g的相关信息。

//go:nosplit
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
}

关于这里sudog获取使用了二级缓存，即P本地sudog缓存和全局的sched全局的sudog缓存。当本地的sudog缓存不足，则从全局缓存中获取；如果全局缓存也没有，则重新分配一个新的sudog。

递增nwait进而避免semrelease中的快速路径
再次检查cansemacquire避免错过wakeup，如果成功则nwait-1并返回
将当前g封装进sudog并放入等待队列

// queue adds s to the blocked goroutines in semaRoot.
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 {//先进先出的话 将新节点放到链表的第一位
    // 用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的等待链表的第一位
    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放到等待列表的末尾
    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作为一个新的叶子节点加入到唯一地址树中
 // 平衡树是一个treap树，使用ticket作为随机堆优先级
 // 也就是说，它是根据elem地址排序的二叉树
 // 但是在代表这些地址的可能的二叉树空间中，是通过ticket满足s.ticket均 <=s.prev.ticket 和 s.next.ticket来维护堆
  // 的顺序，从而平均得保持平衡。
 // https://en.wikipedia.org/wiki/Treap
 // https://faculty.washington.edu/aragon/pubs/rst89.pdf
 // s.ticket在几个地方与零比较，因此设置了最低位
 // 这不会明显影响treap的质量
 s.ticket = fastrand() | 1
 s.parent = last
 *pt = s

 // 根据ticket翻转树
 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)
  }
 }
}

这里入队的树结构是一个treap,故名思义，treap=tree+heap，即拥有tree的特性，又有heap的特性。主要思想是在二叉搜索树的基础上，给每个节点一个随机权重(这里是一个随机值ticket)，然后通过旋转在不破坏二叉搜索树性质的前提下将所有节点根据权重重新组织，使其满足堆的性质。由于权重的随机性，所以可以认为treap能在增删操作下相对平衡，不会退化为链表。

这个treap是根据elem地址排序的二叉树，又根据随机值ticket作为权重值，来维护其平衡(ticket满足s.ticket均 <=s.prev.ticket 和 s.next.ticket)，即：

当当前节点的权重值小于根节点的权重值则旋转

如果当前节点是根的左孩子则右旋
如果当前节点是根的右孩子则左旋

rotate

所以从elem的角度，这个treap是个二叉搜索树，从ticket来看是个小顶堆。

其实最早并不是treap结构而是linked list,可以看看(https://github.com/golang/go/issues/17953)

调用gopark挂起当前g
当前g被唤醒后释放sudog

//go:nosplit
func releaseSudog(s *sudog) {
... ...
 gp := getg()
 if gp.param != nil {
  throw("runtime: releaseSudog with non-nil gp.param")
 }
 mp := acquirem() // 设置P禁止抢占
 pp := mp.p.ptr()
 if len(pp.sudogcache) == cap(pp.sudogcache) {
  // 将本地一半的sudog缓存放回全局缓存
  var first, last *sudog
  for len(pp.sudogcache) > cap(pp.sudogcache)/2 {
   n := len(pp.sudogcache)
   p := pp.sudogcache[n-1]
   pp.sudogcache[n-1] = nil
   pp.sudogcache = pp.sudogcache[:n-1]
   if first == nil {
    first = p
   } else {
    last.next = p
   }
   last = p
  }
  lock(&sched.sudoglock)
  last.next = sched.sudogcache
  sched.sudogcache = first
  unlock(&sched.sudoglock)
 }
 pp.sudogcache = append(pp.sudogcache, s)
 releasem(mp)
}

道理很简单，为了保证sudog的复用，当goroutine被唤醒，当前的sudog需要回收到缓存中以备后续使用。刚刚提到这里涉及到P和sched的二级缓存。所以归还sudog时，如果本地sudog已经满了，会将本地的一半缓存交还回全局缓存。

runtime_Semrelease

runtime_Semrelease实际调用semrelease1完成了wakeup的语义。

//go:linkname sync_runtime_Semrelease sync.runtime_Semrelease
func sync_runtime_Semrelease(addr *uint32, handoff bool, skipframes int) {
 semrelease1(addr, handoff, skipframes)
}
func semrelease1(addr *uint32, handoff bool, skipframes int) {
 root := semroot(addr)
 atomic.Xadd(addr, 1)

 // 快速路径：没有等待者？
 // 检查必须发生在xadd之后，避免错过wakeup
 // (详见semacquire中的循环).
 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
  }
    readyWithTime(s, 5+skipframes) //goready(s.g,5)标记runnable 等待被重新调度
  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.
   goyield()
  }
 }
}

大致逻辑：

根据信号量地址偏移取模&semtable[(uintptr(unsafe.Pointer(addr))>>3)%semTabSize].root拿到semaRoot
信号量原子递增+1，这样semacquire1阻塞的goroutine就可能通过cansemacquire操作
通过原子判断root.nwait的值是否为0，为0表示当前不存在阻塞的goroutine。这里的检查必须发生在semacquire1中的atomic.Xadd(&root.nwait, 1)，防止错过唤醒操作。
加锁再次检查root.nwait的值，没有阻塞的goroutine 则直接返回。否则，从treap中出队当前信号量上的sudog。

// 如果semacquire1中设置了对sudog进行概要分析，dequeue计算到现在为止唤醒goroutine的时间作为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) {//查找到指定信号量地址上的sudog
   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 {
  // 用t替换唯一addrs的根树中的s
  *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 {//该信号量地址上 只有一个sudog时
  // 将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)
   }
  }
  // s当前为叶子节点，移除s
  if s.parent != nil {
   if s.parent.prev == s {//为根节点的左孩子
    s.parent.prev = nil
   } else {//为根节点的右孩子
    s.parent.next = nil
   }
  } else {//当前treap只有s一个节点
   root.treap = nil
  }
 }
 s.parent = nil
 s.elem = nil
 s.next = nil
 s.prev = nil
 s.ticket = 0
 return s, now
}

查找semaRoot中阻塞在指定信号量addr上的第一个goroutine。熟悉了treap结构及queue的逻辑后这里dequeue就比较简单：

查找treap中指定addr的sudog节点
- 若链表长度大于1，则将头节点弹出，返回弹出的sudog
- 若链表长度等于1，即需要移除treap树的节点，这时候需要通过循环旋转将节点根据权重保持平衡,将目标节点旋转为叶子节点，然后删除
- 如果未找到则返回nil,0
- 如果找到，判断节点的等待链表

rotate

如果出队的sudog不为空，将root.nwait原子-1，并释放锁()，让其他goroutine可以继续执行
readyWithTime将sudog中的g唤醒，并放到当前P本地队列的下一个执行位置

func readyWithTime(s *sudog, traceskip int) {
 if s.releasetime != 0 {
  s.releasetime = cputicks()
 }
 goready(s.g, traceskip)
}
func goready(gp *g, traceskip int) {
 systemstack(func() { //切换到系统堆栈
  ready(gp, traceskip, true)
 })
}
// 标记 gp准备run
func ready(gp *g, traceskip int, next bool) {
 if trace.enabled {
  traceGoUnpark(gp, traceskip)
 }

 status := readgstatus(gp)

 // Mark runnable.
 _g_ := getg()
 mp := acquirem() // 设置禁止P抢占
 if status&^_Gscan != _Gwaiting {
  dumpgstatus(gp)
  throw("bad g->status in ready")
 }

 // status is Gwaiting or Gscanwaiting, make Grunnable and put on runq
 casgstatus(gp, _Gwaiting, _Grunnable)
  // 将g放到P的本地队列，注意这里next=true即放到本地队列的下一个执行位置
  // 否则放到对尾
 runqput(_g_.m.p.ptr(), gp, next)
 wakep()
 releasem(mp)//解除抢占
}

饥饿状态下，调用goyield()让出当前时间片，由等待的g继承时间片，避免无限的争夺信号量。因为readyWithTime已经将等待的G放到P本地队列下一个位置，所以调度器会立即执行s.g

func goyield() {
 checkTimeouts()
 mcall(goyield_m)
}
func goyield_m(gp *g) {
 if trace.enabled {
  traceGoPreempt()
 }
 pp := gp.m.p.ptr()
 casgstatus(gp, _Grunning, _Grunnable)//让出时间片
 dropg()
 runqput(pp, gp, false)//将当前g放到P本地队列尾部
 schedule()//触发调度
}

这也是sync.Mutex饥饿模式下，等待goroutine能优先获得锁的原因。

总结

semacquire和semrelease成对出现，实现了简单的sleep和wakeup原语。主要解决并发场景的资源争用问题，显然他们一定是在两个不同的m上执行的场景发生。我们不妨假设m1和m2

当m1上的g1执行到semacquire1时，如果快速路径cansemacquire成功，则说明g1抢到锁，能够继续执行。但一旦失败且在Harder Case下依然抢不到锁，则会进入goparkunlock，将当前g1放到等待队列中，进而让m1切换并执行其他的g。
当m2上的g2开始调用semrelease1时，将等待的g1放回P的本地调度队列中，若当前为饥饿模式(handoff=ture)则让当前等待继承时间片立刻执行，如果成功则semacquire1中会归还sudog。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-03-01，如有侵权请联系 cloudcommunity@tencent.com 删除

bash 指令