Golang Cond源碼分析

cond的主要作用就是獲取鎖之后，wait()方法會等待一個通知，來進行下一步鎖釋放等操作，以此控制鎖合適釋放，釋放頻率,適用于在并發環境下goroutine的等待和通知。

針對Golang 1.9的sync.Cond，與Golang 1.10一樣。 源代碼位置：sync\cond.go。

結構體

type Cond struct {
    noCopy noCopy  // noCopy可以嵌入到結構中，在第一次使用后不可復制,使用go vet作為檢測使用

    // 根據需求初始化不同的鎖，如*Mutex 和 *RWMutex
    L Locker

    notify  notifyList  // 通知列表,調用Wait()方法的goroutine會被放入list中,每次喚醒,從這里取出
    checker copyChecker // 復制檢查,檢查cond實例是否被復制
}

再來看看等待隊列notifyList結構體：

type notifyList struct {
    wait   uint32
    notify uint32
    lock   uintptr
    head   unsafe.Pointer
    tail   unsafe.Pointer
}

函數

NewCond

相當于Cond的構造函數，用于初始化Cond。

參數為Locker實例初始化,傳參數的時候必須是引用或指針,比如&sync.Mutex{}或new(sync.Mutex)，不然會報異常:cannot use lock (type sync.Mutex) as type sync.Locker in argument to sync.NewCond。
大家可以想想為什么一定要是指針呢？ 知道的可以給我留言回答。

func NewCond(l Locker) *Cond {
    return &Cond{L: l}
}

Wait

等待自動解鎖c.L和暫停執行調用goroutine。恢復執行后,等待鎖c.L返回之前。與其他系統不同，等待不能返回，除非通過廣播或信號喚醒。

因為c。當等待第一次恢復時，L并沒有被鎖定，調用者通常不能假定等待返回時的條件是正確的。相反，調用者應該在循環中等待:

func (c *Cond) Wait() {
    // 檢查c是否是被復制的，如果是就panic
    c.checker.check()
    // 將當前goroutine加入等待隊列
    t := runtime_notifyListAdd(&c.notify)
    // 解鎖
    c.L.Unlock()
    // 等待隊列中的所有的goroutine執行等待喚醒操作
    runtime_notifyListWait(&c.notify, t)
    c.L.Lock()
}

判斷cond是否被復制。

type copyChecker uintptr

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")
    }
}

Signal

喚醒等待隊列中的一個goroutine，一般都是任意喚醒隊列中的一個goroutine，為什么沒有選擇FIFO的模式呢？這是因為FiFO模式效率不高，雖然支持，但是很少使用到。

func (c *Cond) Signal() {
    // 檢查c是否是被復制的，如果是就panic
    c.checker.check()
    // 通知等待列表中的一個 
    runtime_notifyListNotifyOne(&c.notify)
}

Broadcast

喚醒等待隊列中的所有goroutine。

func (c *Cond) Broadcast() {
    // 檢查c是否是被復制的，如果是就panic
    c.checker.check()
    // 喚醒等待隊列中所有的goroutine
    runtime_notifyListNotifyAll(&c.notify)
}

實例

package main

import (
    "fmt"
    "sync"
    "time"
)

var locker = new(sync.Mutex)
var cond = sync.NewCond(locker)

func main() {
    for i := 0; i < 40; i++ {
        go func(x int) {
            cond.L.Lock()         //獲取鎖
            defer cond.L.Unlock() //釋放鎖
            cond.Wait()           //等待通知,阻塞當前goroutine
            fmt.Println(x)
            time.Sleep(time.Second * 1)

        }(i)
    }
    time.Sleep(time.Second * 1)
    fmt.Println("Signal...")
    cond.Signal() // 下發一個通知給已經獲取鎖的goroutine
    time.Sleep(time.Second * 1)
    cond.Signal() // 3秒之后 下發一個通知給已經獲取鎖的goroutine
    time.Sleep(time.Second * 3)
    cond.Broadcast() //3秒之后 下發廣播給所有等待的goroutine
    fmt.Println("Broadcast...")
    time.Sleep(time.Second * 60)
}