手摸手Go 深入理解sync.Cond

package main

import (
 "fmt"
 "math/rand"
 "os"
 "os/signal"
 "sync"
 "time"
)

type FIFO struct {
 lock  sync.Mutex
 cond  *sync.Cond
 queue []int
}

type Queue interface {
 Pop() int
 Offer(num int) error
}

func (f *FIFO) Offer(num int) error {
 f.lock.Lock()
 defer f.lock.Unlock()
 f.queue = append(f.queue, num)
 f.cond.Broadcast()
 return nil
}
func (f *FIFO) Pop() int {
 f.lock.Lock()
 defer f.lock.Unlock()
 for {
  for len(f.queue) == 0 {
   f.cond.Wait()
  }
  item := f.queue[0]
  f.queue = f.queue[1:]
  return item
 }
}

func main() {
 l := sync.Mutex{}
 fifo := &FIFO{
  lock:  l,
  cond:  sync.NewCond(&l),
  queue: []int{},
 }
 go func() {
  for {
   fifo.Offer(rand.Int())
  }
 }()
 time.Sleep(time.Second)
 go func() {
  for {
   fmt.Println(fmt.Sprintf("goroutine1 pop-->%d", fifo.Pop()))
  }
 }()
 go func() {
  for {
   fmt.Println(fmt.Sprintf("goroutine2 pop-->%d", fifo.Pop()))
  }
 }()

 ch := make(chan os.Signal, 1)
 signal.Notify(ch, os.Interrupt)
 <-ch
}

    c.L.Lock()
    for !condition() {
        c.Wait()
    }
    ... make use of condition ...
   c.L.Unlock()

type Cond struct {
 noCopy noCopy // Cond使用后不允许拷贝
 // L is held while observing or changing the condition
 L Locker
  //通知列表调用wait()方法的goroutine会被放到notifyList中
 notify  notifyList
 checker copyChecker //检查Cond实例是否被复制
}

func (c *copyChecker) check() {
 if uintptr(*c) != uintptr(unsafe.Pointer(c)) &&
  !atomic.CompareAndSwapUintptr((*uintptr)(c), 0, uintptr(unsafe.Pointer(c))) &&
  uintptr(*c) != uintptr(unsafe.Pointer(c)) {
  panic("sync.Cond is copied")
 }
}

type notifyList struct {
 // wait is the ticket number of the next waiter. It is atomically
 // incremented outside the lock.
 wait uint32 // 等待goroutine操作的数量

 // notify is the ticket number of the next waiter to be notified. It can
 // be read outside the lock, but is only written to with lock held.
 //
 // Both wait & notify can wrap around, and such cases will be correctly
 // handled as long as their "unwrapped" difference is bounded by 2^31.
 // For this not to be the case, we'd need to have 2^31+ goroutines
 // blocked on the same condvar, which is currently not possible.
 notify uint32 // 唤醒goroutine操作的数量

 // List of parked waiters.
 lock mutex
 head *sudog
 tail *sudog
}

func (c *Cond) Wait() {
  //1. 检查cond是否被拷贝
 c.checker.check()
  //2. notifyList.wait+1
 t := runtime_notifyListAdd(&c.notify)
  //3. 释放锁 让出资源给其他goroutine
 c.L.Unlock()
  //4. 挂起goroutine
 runtime_notifyListWait(&c.notify, t)
  //5. 尝试获得锁
 c.L.Lock()
}

//go:linkname notifyListAdd sync.runtime_notifyListAdd
func notifyListAdd(l *notifyList) uint32 {
 // This may be called concurrently, for example, when called from
 // sync.Cond.Wait while holding a RWMutex in read mode.
 return atomic.Xadd(&l.wait, 1) - 1
}

// notifyListWait waits for a notification. If one has been sent since
// notifyListAdd was called, it returns immediately. Otherwise, it blocks.
//go:linkname notifyListWait sync.runtime_notifyListWait
func notifyListWait(l *notifyList, t uint32) {
 lockWithRank(&l.lock, lockRankNotifyList)

 // 如果已经被唤醒 则立即返回
 if less(t, l.notify) {
  unlock(&l.lock)
  return
 }

 // Enqueue itself.
 s := acquireSudog()
 s.g = getg()
  // 把等待递增序号赋值给s.ticket 为FIFO打基础
 s.ticket = t
 s.releasetime = 0
 t0 := int64(0)
 if blockprofilerate > 0 {
  t0 = cputicks()
  s.releasetime = -1
 }
  // 将当前goroutine插入到notifyList链表中
 if l.tail == nil {
  l.head = s
 } else {
  l.tail.next = s
 }
 l.tail = s
  // 最终调用gopark挂起当前goroutine
 goparkunlock(&l.lock, waitReasonSyncCondWait, traceEvGoBlockCond, 3)
 if t0 != 0 {
  blockevent(s.releasetime-t0, 2)
 }
  // goroutine被唤醒后释放sudog
 releaseSudog(s)
}

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

//go:linkname notifyListNotifyOne sync.runtime_notifyListNotifyOne
func notifyListNotifyOne(l *notifyList) {
  // wait==notify 说明没有等待的goroutine了
 if atomic.Load(&l.wait) == atomic.Load(&l.notify) {
  return
 }
 lockWithRank(&l.lock, lockRankNotifyList)
 // 锁下二次检查
 t := l.notify
 if t == atomic.Load(&l.wait) {
  unlock(&l.lock)
  return
 }

 // 更新下一个需要被唤醒的ticket number
 atomic.Store(&l.notify, t+1)

 // Try to find the g that needs to be notified.
 // If it hasn't made it to the list yet we won't find it,
 // but it won't park itself once it sees the new notify number.
 //
 // This scan looks linear but essentially always stops quickly.
 // Because g's queue separately from taking numbers,
 // there may be minor reorderings in the list, but we
 // expect the g we're looking for to be near the front.
 // The g has others in front of it on the list only to the
 // extent that it lost the race, so the iteration will not
 // be too long. This applies even when the g is missing:
 // it hasn't yet gotten to sleep and has lost the race to
 // the (few) other g's that we find on the list.
  //这里是FIFO实现的核心 其实就是遍历链表 sudog.ticket查找指定需要唤醒的节点
 for p, s := (*sudog)(nil), l.head; s != nil; p, s = s, s.next {
  if s.ticket == t {
   n := s.next
   if p != nil {
    p.next = n
   } else {
    l.head = n
   }
   if n == nil {
    l.tail = p
   }
   unlock(&l.lock)
   s.next = nil
   readyWithTime(s, 4)
   return
  }
 }
 unlock(&l.lock)
}

func (c *Cond) Broadcast() {
 c.checker.check()
 runtime_notifyListNotifyAll(&c.notify)
}
// notifyListNotifyAll notifies all entries in the list.
//go:linkname notifyListNotifyAll sync.runtime_notifyListNotifyAll
func notifyListNotifyAll(l *notifyList) {
 // Fast-path 无等待goroutine直接返回
 if atomic.Load(&l.wait) == atomic.Load(&l.notify) {
  return
 }

 lockWithRank(&l.lock, lockRankNotifyList)
 s := l.head
 l.head = nil
 l.tail = nil
 // 直接更新notify=wait
 atomic.Store(&l.notify, atomic.Load(&l.wait))
 unlock(&l.lock)

 // 依次调用goready唤醒goroutine
 for s != nil {
  next := s.next
  s.next = nil
  readyWithTime(s, 4)
  s = next
 }
}