并發(fā)編程的可見性

在 Go 官網(wǎng)上的內(nèi)存模型一文中，介紹了在 Go 并發(fā)編程下數(shù)據(jù)可見性問題，可見性是并發(fā)編程中一個(gè)重要概念，指的是在哪些條件下，可以保證一個(gè)線程中讀取某個(gè)變量時(shí)，可以觀察到另一個(gè)線程對(duì)該變量的寫入后的值，Go 語(yǔ)言中的 goroutine 也適用。

一般來說可見性屬于偏硬件和底層，因?yàn)樯婕暗蕉嗪?CPU 的 cache 讀寫和同步問題，開發(fā)者不需要關(guān)心細(xì)節(jié)，高級(jí)編程語(yǔ)言要么屏蔽掉了這些細(xì)節(jié)，要么會(huì)給出一些保證，承諾在確定的條件下就會(huì)得到確定的結(jié)果。

Go channel 有一個(gè)特性是在一個(gè)無(wú)緩沖的 channel 上發(fā)送和接收必須等待對(duì)方準(zhǔn)備好，才可以執(zhí)行，否則會(huì)被阻塞。實(shí)際上這就是一個(gè)同步保證，那么這個(gè)同步保證是如何實(shí)現(xiàn)的？下面看看官方文章中是如何解釋的。

先 send 后 receive

文中對(duì) channel 的描述有幾個(gè)原則，第一個(gè)是

A send on a channel is synchronized before the completion of the corresponding receive from that channel.

意思是：在一個(gè) channel 上的發(fā)送操作應(yīng)該發(fā)生在對(duì)應(yīng)的接收操作完成之前。說人話就是：要先發(fā)送數(shù)據(jù)，然后才能接收數(shù)據(jù)，否則就會(huì)阻塞。這也比較符合一般的認(rèn)知。

并用下面一段代碼舉例說明，這段代碼確保一定會(huì)輸出 "hello, world”。

var c = make(chan int, 10)
var a string

func f() {
    a = "hello, world"
    c <- 0
}

func main() {
    go f()
    <-c
    print(a)
}

f 函數(shù)負(fù)責(zé)給變量 a 賦值，main 函數(shù)負(fù)責(zé)打印變量 a。main 函數(shù)阻塞等待在 <- c 處，直到 f 函數(shù)對(duì) a 賦值之后并寫入數(shù)據(jù)到 c 中，main 函數(shù)才被喚醒繼續(xù)執(zhí)行，所以此時(shí)打印 a 必然會(huì)得到結(jié)果。

先 receive 后 send？

而下面這段描述有點(diǎn)反直覺

A receive from an unbuffered channel is synchronized before the completion of the corresponding send on that channel.

意思是在無(wú)緩沖 channel 上的接收操作發(fā)生在對(duì)應(yīng)的發(fā)送操作完成之前，說人話就是：要先接收數(shù)據(jù)，之后才可以發(fā)送數(shù)據(jù)，否則就會(huì)阻塞。這句話看上去與第一條相悖，因?yàn)榈谝粭l強(qiáng)調(diào)發(fā)送操作要在接收完成之前發(fā)生，而這一條強(qiáng)調(diào)接收操作要在發(fā)送完成之前發(fā)生，這樣相互等待對(duì)方的情況，不會(huì)陷入死鎖狀態(tài)嗎？

下面的示例代碼與前一個(gè)類似，區(qū)別是將 c 換成了無(wú)緩沖 channel，并把 c 的寫入和讀取調(diào)換了位置，這段代碼同樣可以保證輸出 "hello, world”。

var c = make(chan int)
var a string

func f() {
    a = "hello, world"
    <-c
}

func main() {
    go f()
    c <- 0
    print(a)
}

這兩段話到底是什么意思？為什么要相互等待但又不會(huì)死鎖？

接下來看看 runtime/chan.go 中是怎么實(shí)現(xiàn) channel 的發(fā)送和接收的。

channel 的結(jié)構(gòu)

首先看看 channel 的數(shù)據(jù)結(jié)構(gòu)

type hchan struct {
    qcount   uint           // 緩沖區(qū)元素?cái)?shù)量
    dataqsiz uint           // 緩沖區(qū)大小
    buf      unsafe.Pointer // 緩沖區(qū)起始指針
    elemsize uint16
    closed   uint32
    elemtype *_type
    sendx    uint   // 下一次發(fā)送的元素在隊(duì)列中的索引
    recvx    uint   // 下一個(gè)接收的元素在隊(duì)列中的索引
    recvq    waitq  // 當(dāng)隊(duì)列無(wú)數(shù)據(jù)時(shí)，receiver 阻塞等待的隊(duì)列
    sendq    waitq  // 當(dāng)隊(duì)列無(wú)空間時(shí)，sender 阻塞等待的隊(duì)列

    lock mutex
}

channel 內(nèi)部實(shí)現(xiàn)了一個(gè)環(huán)形隊(duì)列，通過 qcount dataqsiz buf sendx recvx 幾個(gè)部分組成。

另外 channel 還維護(hù)了兩個(gè)等待隊(duì)列，如果在執(zhí)行 <-c receive 操作時(shí)，此時(shí) channel 不滿足接收條件，receiver 會(huì)進(jìn)入 recvq 等待隊(duì)列；同樣的如果執(zhí)行 c<- send 操作時(shí)，此時(shí) channel 不滿足發(fā)送條件，sender 會(huì)進(jìn)入 sendq 等待隊(duì)列。

具體看代碼：

var c = make(chan int)
var a string

func f() {
    a = "hello, world"

    x := <-c    // 3
    fmt.Println("\nx:", x)
}

func main() {
    go f()      // 1
    c <- 123456 // 2

    print(a)
}

send 具體干了什么

當(dāng) main 函數(shù)執(zhí)行到 c<-123456 是，會(huì)執(zhí)行 runtime/chan.go 中的 chansend 函數(shù)，該函數(shù)首先會(huì)判斷當(dāng)前 channel c 的等待接收隊(duì)列是否有阻塞的 receiver

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  // ...省略部分代碼...

  // 是否有等待的 receiver 存在
    if sg := c.recvq.dequeue(); sg != nil {
        send(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true
    }

  // ...省略部分代碼...
}

如果有等待的 receiver 則彈出隊(duì)列，調(diào)用 send 函數(shù)，其中 sg 就表示 receiver，sg.elem 表示將數(shù)據(jù)接收到哪里去，這個(gè)地址也就對(duì)應(yīng)示例代碼中的變量 x 的地址。

func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
  // ...省略部分代碼...

    if sg.elem != nil {
        sendDirect(c.elemtype, sg, ep)
        sg.elem = nil
    }

  // ...省略部分代碼...
  // 將 goroutine 置為可執(zhí)行狀態(tài)
}

sendDirect 函數(shù)就是直接從 src 里面將數(shù)據(jù)復(fù)制到 dst 中。

// 直接拷貝數(shù)據(jù)
func sendDirect(t *_type, sg *sudog, src unsafe.Pointer) {
    dst := sg.elem
    typeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.size)
    memmove(dst, src, t.size)
}

回到 chansend 函數(shù)，如果沒有等待的 receiver，那么會(huì)查看當(dāng)前 buf 中是否有空間，如果有空間，則數(shù)據(jù)緩存到 buf 中。

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  // ...省略部分代碼...

  // 將數(shù)據(jù)緩存到 buf 中
    if c.qcount < c.dataqsiz {
        // Space is available in the channel buffer. Enqueue the element to send.
        qp := chanbuf(c, c.sendx)
        if raceenabled {
            racenotify(c, c.sendx, nil)
        }
        typedmemmove(c.elemtype, qp, ep)
        c.sendx++
        if c.sendx == c.dataqsiz {
            c.sendx = 0
        }
        c.qcount++
        unlock(&c.lock)
        return true
    }

  // ...省略部分代碼...
}

如果也沒有 buf 空間，那么就將 sender 本身放入到 sendq 等待隊(duì)列中。

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  // ...省略部分代碼...

  // 進(jìn)入 sendq 等待隊(duì)列
    gp := getg()
    mysg := acquireSudog()
    mysg.releasetime = 0
    if t0 != 0 {
        mysg.releasetime = -1
    }
    mysg.elem = ep
    mysg.waitlink = nil
    mysg.g = gp
    mysg.isSelect = false
    mysg.c = c
    gp.waiting = mysg
    gp.param = nil
    c.sendq.enqueue(mysg)

  // ...省略部分代碼...
}

總結(jié)起來 send 操作分三部分：

• 如果當(dāng)前 channel 上有等待的 receiver，則直接 copy 數(shù)據(jù)過去
• 否則如果當(dāng)前 buf 有空閑空間，則將數(shù)據(jù)存在 buf 中
• 否則將 sender 本身加入到 sendq 等待隊(duì)列中

receive 具體干了什么

相應(yīng)的與發(fā)送類似，執(zhí)行到示例代碼中第 (3) 步接收數(shù)據(jù)時(shí)，會(huì)調(diào)用 runtime/chan.go 中的 chanrecv 函數(shù)來處理接收，同樣是先看 sender 等待隊(duì)列是否有阻塞的 sender

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
  // ...省略部分代碼...

  // 從等待的 sender 取一個(gè)出來
    if sg := c.sendq.dequeue(); sg != nil {
        recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true, true
    }
  // ...省略部分代碼...
}

如果有的等待的 sender，那么將 sender 取出來，并復(fù)制數(shù)據(jù)。

func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
    // ...省略部分代碼...
    if ep != nil {
        // copy data from sender
        recvDirect(c.elemtype, sg, ep)
    }
  // ...省略部分代碼...
}

func recvDirect(t *_type, sg *sudog, dst unsafe.Pointer) {
    src := sg.elem
    typeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.size)
    memmove(dst, src, t.size)
}

如果沒有等待的 sender，那么看 buf 中有沒有緩存的數(shù)據(jù)

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
  // ...省略部分代碼...
    if c.qcount > 0 {
        qp := chanbuf(c, c.recvx)
        if raceenabled {
            racenotify(c, c.recvx, nil)
        }
        if ep != nil {
            typedmemmove(c.elemtype, ep, qp)
        }
        typedmemclr(c.elemtype, qp)
        c.recvx++
        if c.recvx == c.dataqsiz {
            c.recvx = 0
        }
        c.qcount--
        unlock(&c.lock)
        return true, true
    }
  // ...省略部分代碼...
}

最后如果也沒有 buf 數(shù)據(jù)，那么久把自己加入到 receiver 等待隊(duì)列中 recvq

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
  // ...省略部分代碼...

    gp := getg()
    mysg := acquireSudog()
    mysg.releasetime = 0
    if t0 != 0 {
        mysg.releasetime = -1
    }
    mysg.elem = ep
    mysg.waitlink = nil
    gp.waiting = mysg
    mysg.g = gp
    mysg.isSelect = false
    mysg.c = c
    gp.param = nil
    c.recvq.enqueue(mysg)

  // ...省略部分代碼...
}

總結(jié)起來 receive 操作分三部分：

• 如果當(dāng)前 channel 上有等待的 sender，則直接 copy 數(shù)據(jù)過去
• 否則如果當(dāng)前 buf 有緩存的數(shù)據(jù)，則將讀取該數(shù)據(jù)
• 否則將 receiver 本身加入到 recvq 等待隊(duì)列中

小結(jié)

這樣一來就能夠理解前面的兩個(gè)原則了，在一個(gè)無(wú)緩沖的 channel 中，無(wú)論是 sender 先執(zhí)行，還是 receiver 先執(zhí)行，都會(huì)因?yàn)檎也坏綄?duì)方，并且沒有 buf 空間的情況下，將自己加入到等待隊(duì)列；當(dāng)對(duì)方開始執(zhí)行時(shí)就會(huì)檢查到已經(jīng)有對(duì)端正在阻塞，進(jìn)而拷貝數(shù)據(jù)，并喚醒阻塞的對(duì)象最終走完整個(gè)流程。

有一種說法是：sender 必須在 receiver 準(zhǔn)備好才能執(zhí)行，否則就會(huì)阻塞；而 receiver 必須在 sender 準(zhǔn)備好才能執(zhí)行，否則就會(huì)阻塞；這個(gè)說法沒錯(cuò)，但是太籠統(tǒng)了，什么叫準(zhǔn)備好？怎么算是準(zhǔn)備好？這是比較模糊的。而看過 send 和 receive 的流程之后，就更能理解整個(gè)過程了。

為什么要有無(wú)緩沖 channel

實(shí)際上兩個(gè) goroutine 相互等待對(duì)方到達(dá)某個(gè)狀態(tài)的效果，非常類似操作系統(tǒng)中的一種同步機(jī)制：屏障 barrier，同步屏障要求只有當(dāng)所有進(jìn)程都到達(dá)屏障后，才能一起執(zhí)行下一狀態(tài)，否則就阻塞在屏障處。

回到 channel 操作，即 sender 和 receiver 無(wú)論誰(shuí)先執(zhí)行，都必須等待對(duì)方也已經(jīng)執(zhí)行，兩者才可以繼續(xù)執(zhí)行。就像一塊電路板串聯(lián)有兩個(gè)開關(guān)，要想電路聯(lián)通，必須兩個(gè)開關(guān)都被打開才可以，而不管哪一個(gè)先打開，都必須等待另一個(gè)開關(guān)也打開，之后電流才可以接通電路也才聯(lián)通。

可以將無(wú)緩沖 channel 看做是一種同步屏障，同步屏障能夠讓多個(gè) goroutine 都達(dá)到某種狀態(tài)之后才可以繼續(xù)執(zhí)行，這是帶緩沖 channel 無(wú)法做到的。另外在無(wú)緩沖 channel 數(shù)據(jù)的交換更加簡(jiǎn)單快速，因?yàn)椴恍枰S護(hù)緩存 buf，實(shí)現(xiàn)邏輯也更簡(jiǎn)單，運(yùn)行更可靠。

以上就是Go channel發(fā)送方和接收方如何相互阻塞等待源碼解讀的詳細(xì)內(nèi)容，更多關(guān)于Go channel相互阻塞等待的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論