信号量semaphore实现

const semTabSize = 251

// 信号量hash表
var semtable [semTabSize]struct {
 root semaRoot
 pad  [cpu.CacheLinePadSize - unsafe.Sizeof(semaRoot{})]byte
}

type semaRoot struct {
 lock mutex
 // 平衡树的树根
 treap *sudog 
 // 保存有多少个goroutine阻塞在该semaRoot上
 nwait uint32 
}

type sudog struct {
 // 保存goroutine对象的指针
 g *g

 // isSelect是给channel操作用的，如果它为true,表示有g正在参与一个select
 isSelect bool
 // next和prev可以理解为树的右和左两个分支，sudog构成一个树形结构
 next *sudog
 prev *sudog
 // 数据元素，treap树中根据elem的值构成二叉搜索树
 elem unsafe.Pointer 

 // 获取时间,goroutine进行休眠前的时间
 acquiretime int64
 // 释放时间，goroutine被唤醒后的时间
 releasetime int64
 // 权重值，treap树根据ticket值构成一个小根堆结构
 ticket      uint32
 // 指向父sudog节点
 parent      *sudog 
 // sudog列表，有相同elem值的g构成一个链表，它们之间通过waitlink串联起来
 waitlink    *sudog 
 // sudog列表中的最后一个元素，有字段，使得将goroutine插入链表尾的时间复杂度为O(1)
 waittail    *sudog 
 // 通道对象指针
 c           *hchan 
}

// 提供给sync包使用，例如在sync.WaithGroup用到了
func sync_runtime_Semacquire(addr *uint32) {
 semacquire1(addr, false, semaBlockProfile, 0)
}

// 提供给网络轮询poll使用
func poll_runtime_Semacquire(addr *uint32) {
 semacquire1(addr, false, semaBlockProfile, 0)
}

// 提供给sync包使用，例如在sync.Mutext，sync.RWMutex均有使用
func sync_runtime_Semrelease(addr *uint32, handoff bool, skipframes int) {
 semrelease1(addr, handoff, skipframes)
}

// 也是提供给sync包使用，在sync.Mutex中
func sync_runtime_SemacquireMutex(addr *uint32, lifo bool, skipframes int) {
 semacquire1(addr, lifo, semaBlockProfile|semaMutexProfile, skipframes)
}

func semacquire1(addr *uint32, lifo bool, profile semaProfileFlags, skipframes int) {
 // 获取当前运行的g
 gp := getg()
 // 判断当前的g是否是用户程序g,如果不是抛出异常
 if gp != gp.m.curg {
  throw("semacquire not on the G stack")
 }

 // 当addr指向的值大于0时，直接返回，否则继续下面的逻辑
 if cansemacquire(addr) {
  return
 }

 // 获取一个sudog对象s
 s := acquireSudog()
 // 根据addr的值(地址值)确定落在哪个hash桶中，返回桶中的semaRoot值
 root := semroot(addr)
 t0 := int64(0)
 // 初始化获取和释放goroutine时的时间，这两个值用于性能分析用，这里我们可以不关心
 s.releasetime = 0
 s.acquiretime = 0
 // ticket是权重值，sudog组成了一个treap结构，即tree+heap。从t.elem角度来看，
 // 此结构是一个二叉搜索树。从ticket角度看，此结构是一个小根堆。
 s.ticket = 0

 // 根据条件设置releasetime和acquiretime，用于pprof性能分析
 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 {
  lock(&root.lock)
  atomic.Xadd(&root.nwait, 1)
  // 加锁后再次检查*addr的值是否大于0，如果大于0，直接唤醒
  if cansemacquire(addr) {
   // 可以唤醒，前面nwait加了1，这里在减去1
   atomic.Xadd(&root.nwait, -1)
   unlock(&root.lock)
   break
  }
  // 将s加入到treap中，传入的参数有3个，根据addr的值决定挂载树上的哪个节点
  // s为要挂载的对象，lifo是后进先出的缩写(lat in first out)
  root.queue(addr, s, lifo)
  // 调用gopark将当前的goroutine休眠
  goparkunlock(&root.lock, waitReasonSemacquire, traceEvGoBlockSync, 4+skipframes)
  // 唤醒后检查，root.queue内部有修改s.ticket
  if s.ticket != 0 || cansemacquire(addr) {
   break
  }
 }
 if s.releasetime > 0 {
  blockevent(s.releasetime-t0, 3+skipframes)
 }
 // 将对象s归还到P中
 releaseSudog(s)
}

func (root *semaRoot) queue(addr *uint32, s *sudog, lifo bool) {
 // 设置s的g为当前的goroutine
 s.g = getg()
 // 将地址addr的值给s的elem
 s.elem = unsafe.Pointer(addr)
 // 清理干净next和prev
 s.next = nil
 s.prev = nil

 var last *sudog
 // pt为 **sudog类型，二次指针
 pt := &root.treap
 // 遍历treap，将s放进树中的合适位置
 for t := *pt; t != nil; t = *pt {
  // 地址已经在treap中存在，相同地址的sudog对象构成了一个链表
  if t.elem == unsafe.Pointer(addr) {
   // lifo为true,表示插入到链表的表头
   if lifo {
    // 将treap中原来是t的节点替换为新的节点s
    *pt = s
    // 将原来t的节点的值拷贝给s
    s.ticket = t.ticket
    s.acquiretime = t.acquiretime
    // 将s和treap互相关联绑定
    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
    }
    // 以s为开头的sudog对象构成了一个链表，执行s.waitlink=t
    // 将s和原来的链表串起来
    s.waitlink = t
    // 设置s的指向的尾节点值
    s.waittail = t.waittail
    // 如果s.waittail为nil,说明t.waittail的值为nil
    // 即之前链表中只有一个元素就是t
    if s.waittail == nil {
     // 设置s的尾节点为t,现在只要2个元素s和t
     s.waittail = t
    }
    // 解除旧节点t与treap的关联
    t.parent = nil
    t.prev = nil
    t.next = nil
    t.waittail = nil
   } else {
    // else逻辑将节点s添加到链表尾部，处理来说比上面加入到头节点简单很多
    // 如果t.waittail为nil,说明链表中只有一个元素就是t
    if t.waittail == nil {
     // t的下一个节点指向s
     t.waitlink = s
    } else {
     // t的最后一个节点的下一个节点指向s
     t.waittail.waitlink = s
    }
    // 更新t的尾节点为当前新加入的节点s
    // NOTE:这里只是更新了链表头节点指向的尾部节点为s,那链表中间的节点的waittail值不更新，
    // 将它指向s吗？需要更新，不过是渐进式的，在每次链表头结点出队dequeue的时候，会更新头
    // 节点下一个节点，将它的waittail指向正确的位置
    t.waittail = s
    // 将s的waitlink指向空
    s.waitlink = nil
   }
   // 已将s加入到treap中，直接返回
   return
  }
  last = t
  // treap是根据elem构成的二叉搜索树，所以如果addr比当前t.elem小
  // 向左t.prev搜索，否则向右搜索
  if uintptr(unsafe.Pointer(addr)) < uintptr(t.elem) {
   pt = &t.prev
  } else {
   pt = &t.next
  }
 }
    
 // 走到这里，说明addr的值在treap中不存在，所以需要将当前的节点s插入到treap中合适的位置
 // last是一个叶子节点，将当前的节点s挂载到last的子节点上(可能是last.prev也可能是last.next的)

 // 产生一个随机值赋值给s.ticket,下面要根据ticket进行顺序调整，构成一个random search tree
 s.ticket = fastrand() | 1
 // 设置s的parent为last
 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)
  }
 }
}

func semrelease1(addr *uint32, handoff bool, skipframes int) {
 // 更加addr的值定位hash表中桶的位置,并获得treap的根节点root
 root := semroot(addr)
 // 将 *addr的值加1，相当于V操作，在semacquire1中有减1，这里反过来加1
 atomic.Xadd(addr, 1)

 // nwait表示阻塞休眠的goroutine的数量，如果为0，说明没有goroutine休眠，
 // 也就不用进行下面的唤醒逻辑
 if atomic.Load(&root.nwait) == 0 {
  return
 }

 lock(&root.lock)
 // 这里加锁后再次检查nwait，如果为0，处理同上
 if atomic.Load(&root.nwait) == 0 {
  unlock(&root.lock)
  return
 }
 // 找到与addr关联的sudog对象s,将其从treap中移除
 s, t0 := root.dequeue(addr)
 if s != nil {
  // 休眠的goroutine数量减1
  atomic.Xadd(&root.nwait, -1)
 }
 unlock(&root.lock)
 // 如果s不等于nil,需要将s中绑定的g唤醒
 if s != nil { 
  acquiretime := s.acquiretime
  if acquiretime != 0 {
   // 调试诊断
   mutexevent(t0-acquiretime, 3+skipframes)
  }
  if s.ticket != 0 {
   throw("corrupted semaphore ticket")
  }
  // handoff为true表示需要剥离当前goroutine运行，立即运行P中runnext中的g
  if handoff && cansemacquire(addr) {
   s.ticket = 1
  }
  // 调用goready将休眠的goroutine唤醒
  readyWithTime(s, 5+skipframes)
  if s.ticket == 1 && getg().m.locks == 0 {
   // 前面执行readyWithTime，已经将休眠的goroutine放入到P的runnext中了，这里调用
   // goyield让调度程序立即运行runnext中的g. runnext中的g运行会继承当前的时间片。
   // handoff会在sync.Mutex中饥饿状态下会设置为true,避免高度竞争的信号量过度占用P.
   goyield()
  }
 }
}

func (root *semaRoot) dequeue(addr *uint32) (found *sudog, now int64) {
 // ps为二级指针，指向根*sudog
 ps := &root.treap
 // s指向根sudog对象
 s := *ps
 // 遍历treap树，查找是否有元素的elem值与addr相对
 for ; s != nil; s = *ps {
  // 找到了值相等的节点
  if s.elem == unsafe.Pointer(addr) {
   goto Found
  }
  // 二叉搜索，当前的值大于addr，向左查找，否则向右查找
  if uintptr(unsafe.Pointer(addr)) < uintptr(s.elem) {
   ps = &s.prev
  } else {
   ps = &s.next
  }
 }
 // 走到这里，表示没有找到，返回nil
 return nil, 0

Found:
 // 找到了值相同的节点s
 now = int64(0)
 if s.acquiretime != 0 {
  now = cputicks()
 }
 // 多个值相同的sudog对象会组成一个链表结构，这里判断以s开头的链接元素的个数
 // s.waitlink不为nil,表示链表中有多个元素，s会从treap中移除，将链表中第二个节点
 // 移动到s的位置
 if t := s.waitlink; t != nil {
  // 将原来s节点替换为t节点(链表的第二个节点)
  *ps = t
  // 将节点s的ticket值赋值给t
  t.ticket = s.ticket
  // 建立节点t与它父节点关系
  t.parent = s.parent
  // t的prev指向s.prev
  t.prev = s.prev
  if t.prev != nil {
   // 更新t.prev的父节点为当前的t节点
   t.prev.parent = t
  }
  // t的next指向s.next
  t.next = s.next
  if t.next != nil {
   // 更新t.next的父节点为当前的t节点
   t.next.parent = t
  }
  // 更新t节点指向链表的尾节点，这里采用了惰性更新，在入队enqueue的时候
  // 链表中间的节点的waittail还没有设置指向链表尾节点
  if t.waitlink != nil {
   t.waittail = s.waittail
  } else {
   // 如果除了t链表中没有其他元素了，设置t.waittail为nil
   t.waittail = nil
  }
  // 更新节点t的acquiretime值为当前时间
  t.acquiretime = now
  // 清理掉要移除的节点s的waitlink和waittail值
  s.waitlink = nil
  s.waittail = nil
 } else {
  // treap中只有一个节点的elem值与addr相等，则移除s，需要调整以s为根节点的treap
  // 保持random search tree性质，通过左旋或右旋来实现调整
  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从treap树中移除掉
  if s.parent != nil {
   if s.parent.prev == s {
    s.parent.prev = nil
   } else {
    s.parent.next = nil
   }
  } else {
   // 移除的s节点是treap的根节点，需要特殊处理
   root.treap = nil
  }
 }
 // 清理节点s的字段值
 s.parent = nil
 s.elem = nil
 s.next = nil
 s.prev = nil
 s.ticket = 0
 return s, now
}

type p struct {
    ...
 sudogcache []*sudog
 sudogbuf   [128]*sudog
    ...
}

func acquireSudog() *sudog {
 // Delicate dance: 信号量的实现调用acquireSudog，然后acquireSudog调用new(sudog)
 // new调用malloc, malloc调用垃圾收集器，垃圾收集器在stopTheWorld调用信号量
 // 通过在new(sudog)周围执行acquirem/releasem来打破循环
 // acquirem/releasem在new(sudog)期间增加m.locks，防止垃圾收集器被调用
 
 // 获取当前运行的工作线程M
 mp := acquirem()
 // 获取M绑定的P(pp)
 pp := mp.p.ptr()
 // 检查pp的sudogcache 切片是否为空
 if len(pp.sudogcache) == 0 {
  lock(&sched.sudoglock)
  // First, try to grab a batch from central cache.
  // 从全局sched的sudogcache切片中获取sudog加入到pp中，直到全局切片为空或者达到
  // pp切片容量的一半为止
  for len(pp.sudogcache) < cap(pp.sudogcache)/2 && sched.sudogcache != nil {
   s := sched.sudogcache
   sched.sudogcache = s.next
   s.next = nil
   // 将从全局shed中获取的sudog对象加入到本地pp的sudogcache中
   pp.sudogcache = append(pp.sudogcache, s)
  }
  unlock(&sched.sudoglock)
  // 走到这里还是空的，说明全局缓存中也没有sudog对象，这时new一个新的
  if len(pp.sudogcache) == 0 {
   pp.sudogcache = append(pp.sudogcache, new(sudog))
  }
 }
 // 将pp.sudogcache切片中最后一个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
}

func releaseSudog(s *sudog) {
 ... 异常检查
 // 获取当前的工作线程M
 mp := acquirem() // avoid rescheduling to another P
 // 获取与M绑定的P
 pp := mp.p.ptr()
 // 如果本地缓存切片满了，则会将本地缓存后一半的sudog对象转移到全局缓存中
 if len(pp.sudogcache) == cap(pp.sudogcache) {
  // Transfer half of local cache to the central cache.
  var first, last *sudog
  // 先将本地缓存中后一半的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)
 }
 // 将s加入到本地缓存中
 pp.sudogcache = append(pp.sudogcache, s)
 releasem(mp)
}