Golang通道channel的源碼分析

Posted on by

前言

channel是golang中標志性的概念之一，很好很強大！

channel（通道），顧名思義，是一種通道，一種用於並發環境中數據傳遞的通道。通常結合golang中另一重要概念goroutine（go協程）使用，使得在golang中的並發編程變得清晰簡潔同時又高效強大。

今天嘗試著讀讀golang對channel的實現源碼，本文主要是自己個人對於Channel源碼的學習筆記，需要的朋友可以參考以下內容，希望對大傢有幫助。

channel基礎結構

type hchan struct {
    qcount   uint           // total data in the queue
    dataqsiz uint           // size of the circular queue
    buf      unsafe.Pointer // points to an array of dataqsiz elements
    elemsize uint16
    closed   uint32
    elemtype *_type // element type
    sendx    uint   // send index
    recvx：    uint   // receive index
    recvq    waitq  // list of recv waiters
    sendq    waitq  // list of send waiters
    // lock protects all fields in hchan, as well as several
    // fields in sudogs blocked on this channel.
    //
    // Do not change another G's status while holding this lock
    // (in particular, do not ready a G), as this can deadlock
    // with stack shrinking.
    lock mutex
}

hchan結構就是channel的底層數據結構，看源碼定義，可以說是非常清晰瞭。

  • qcount：channel緩存隊列中已有的元素數量
  • dataqsiz：channel的緩存隊列大小（定義channel時指定的緩存大小，這裡channel用的是一個環形隊列）
  • buf：指向channel緩存隊列的指針
  • elemsize：通過channel傳遞的元素大小
  • closed：channel是否關閉的標志
  • elemtype：通過channel傳遞的元素類型
  • sendx：channel中發送元素在隊列中的索引
  • recvx：channel中接受元素在隊列中的索引
  • recvq：等待從channel中接收元素的協程列表
  • sendq：等待向channel中發送元素的協程列表
  • lock：channel上的鎖

其中關於recvq和sendq的兩個列表所用的結構waitq簡單看下。

type waitq struct {
    first *sudog
    last  *sudog
}
type sudog struct {
    g          *g
    selectdone *uint32 // CAS to 1 to win select race (may point to stack)
    next       *sudog
    prev       *sudog
    elem       unsafe.Pointer // data element (may point to stack)
...
    c           *hchan // channel
}

可以看出waiq是一個雙向鏈表結構，鏈上的節點是sudog。從sudog的結構定義可以粗略看出，sudog是對g（即協程）的一個封裝。用於記錄一個等待在某個channel上的協程g、等待的元素elem等信息。

channel初始化

func makechan(t *chantype, size int64) *hchan {
    elem := t.elem
    // compiler checks this but be safe.
    if elem.size >= 1<<16 {
        throw("makechan: invalid channel element type")
    }
    if hchanSize%maxAlign != 0 || elem.align > maxAlign {
        throw("makechan: bad alignment")
    }
    if size < 0 || int64(uintptr(size)) != size || (elem.size > 0 && uintptr(size) > (_MaxMem-hchanSize)/elem.size) {
        panic(plainError("makechan: size out of range"))
    }
    var c *hchan
    if elem.kind&kindNoPointers != 0 || size == 0 {
        // Allocate memory in one call.
        // Hchan does not contain pointers interesting for GC in this case:
        // buf points into the same allocation, elemtype is persistent.
        // SudoG's are referenced from their owning thread so they can't be collected.
        // TODO(dvyukov,rlh): Rethink when collector can move allocated objects.
        c = (*hchan)(mallocgc(hchanSize+uintptr(size)*elem.size, nil, true))
        if size > 0 && elem.size != 0 {
            c.buf = add(unsafe.Pointer(c), hchanSize)
        } else {
            // race detector uses this location for synchronization
            // Also prevents us from pointing beyond the allocation (see issue 9401).
            c.buf = unsafe.Pointer(c)
        }
    } else {
        c = new(hchan)
        c.buf = newarray(elem, int(size))
    }
    c.elemsize = uint16(elem.size)
    c.elemtype = elem
    c.dataqsiz = uint(size)
    if debugChan {
        print("makechan: chan=", c, "; elemsize=", elem.size, "; elemalg=", elem.alg, "; dataqsiz=", size, "\n")
    }
    return c
}

第一部分的3個if是對初始化參數的合法性檢查。

if elem.size >= 1<<16:

檢查channel元素大小，小於2字節

if hchanSize%maxAlign != 0 || elem.align > maxAlign

沒看懂（對齊？）

if size < 0 || int64(uintptr(size)) != size || (elem.size > 0 && uintptr(size) > (_MaxMem-hchanSize)/elem.size)

第一個判斷緩存大小需要大於等於0

int64(uintptr(size)) != size這一句實際是用於判斷size是否為負數。由於uintptr實際是一個無符號整形，負數經過轉換後會變成一個與原數完全不同的很大的正整數，而正數經過轉換後並沒有變化。

最後一句判斷channel的緩存大小要小於heap中能分配的大小。_MaxMem是可分配的堆大小。

第二部分是具體的內存分配。

元素類型為kindNoPointers的時候，既非指針類型，則直接分配（hchanSize+uintptr(size)*elem.size）大小的連續空間。c.buf指向hchan後面的elem隊列首地址。

如果channel緩存大小為0，則c.buf實際上是沒有給他分配空間的

如果類型為非kindNoPointers，則channel的空間和buf的空間是分別分配的。

channel發送

// entry point for c <- x from compiled code
//go:nosplit
func chansend1(c *hchan, elem unsafe.Pointer) {
    chansend(c, elem, true, getcallerpc(unsafe.Pointer(&c)))
}

channel發送，即協程向channel中發送數據，與此操作對應的go代碼如c <- x。

channel發送的實現源碼中，通過chansend1(),調用chansend()，其中block參數為true。

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
    if c == nil {
        if !block {
            return false
        }
        gopark(nil, nil, "chan send (nil chan)", traceEvGoStop, 2)
        throw("unreachable")
    }
... 
}

chansend()首先對c進行判斷， if c == nil：即channel沒有被初始化，這個時候會直接調用gopark使得當前協程進入等待狀態。而且用於喚醒的參數unlockf傳的nil，即沒有人來喚醒它，這樣系統進入死鎖。所以channel必須被初始化之後才能使用，否則死鎖。

接下來是正式的發送處理，且後續操作會加鎖。

lock(&c.lock)

close判斷

if c.closed != 0 {
        unlock(&c.lock)
        panic(plainError("send on closed channel"))
    }

如果channel已經是closed狀態，解鎖然後直接panic。也就是說我們不可以向已經關閉的通道內在發送數據。

將數據發給接收協程

if sg := c.recvq.dequeue(); sg != nil {
        // Found a waiting receiver. We pass the value we want to send
        // directly to the receiver, bypassing the channel buffer (if any).
        send(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true
    }

嘗試從接收等待協程隊列中取出一個協程，如果有則直接數據發給它。也就是說發送到channel的數據會優先檢查接收等待隊列，如果有協程等待取數，就直接給它。發完解鎖，操作完成。

這裡send()方法會將數據寫到從隊列裡取出來的sg中，通過goready()喚醒sg.g(即等待的協程)，進行後續處理。

數據放到緩存

if c.qcount < c.dataqsiz {
        // Space is available in the channel buffer. Enqueue the element to send.
        qp := chanbuf(c, c.sendx)
        if raceenabled {
            raceacquire(qp)
            racerelease(qp)
        }
        typedmemmove(c.elemtype, qp, ep)
        c.sendx++
        if c.sendx == c.dataqsiz {
            c.sendx = 0
        }
        c.qcount++
        unlock(&c.lock)
        return true
    }

如果沒有接收協程在等待，則去檢查channel的緩存隊列是否還有空位。如果有空位，則將數據放到緩存隊列中。

通過c.sendx遊標找到隊列中的空餘位置，然後將數據存進去。移動遊標，更新數據，然後解鎖，操作完成。

if c.sendx == c.dataqsiz {
        c.sendx = 0
    }

通過這一段遊標的處理可以看出，緩存隊列是一個環形。

阻塞發送協程

gp := getg()
    mysg := acquireSudog()
    mysg.releasetime = 0
    if t0 != 0 {
        mysg.releasetime = -1
    }
    // No stack splits between assigning elem and enqueuing mysg
    // on gp.waiting where copystack can find it.
    mysg.elem = ep
    mysg.waitlink = nil
    mysg.g = gp
    mysg.selectdone = nil
    mysg.c = c
    gp.waiting = mysg
    gp.param = nil
    c.sendq.enqueue(mysg)
    goparkunlock(&c.lock, "chan send", traceEvGoBlockSend, 3)

如果緩存也慢瞭，這時候就隻能阻塞住發送協程瞭， 等有合適的機會瞭，再將數據發送出去。

getg()獲取當前協程對象g的指針，acquireSudog()生成一個sudog，然後將當前協程及相關數據封裝好鏈接到sendq列表中。然年通過goparkunlock()將其轉為等待狀態，並解鎖。操作完成。

channel接收

// entry points for <- c from compiled code
//go:nosplit
func chanrecv1(c *hchan, elem unsafe.Pointer) {
    chanrecv(c, elem, true)
}

channel接收，即協程從channel中接收數據，與此操作對應的go代碼如<- c。

channel接收的實現源碼中，通過chanrecv1(),調用chanrecv()，其中block參數為true。

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
...
    if c == nil {
        if !block {
            return
        }
        gopark(nil, nil, "chan receive (nil chan)", traceEvGoStop, 2)
        throw("unreachable")
    }
...
}

同發送一樣，接收也會首先檢查c是否為nil，如果為nil，會調用gopark()休眠當前協程，從而最終造成死鎖。

接收操作同樣先進行加鎖，然後開始正式操作。

close處理

if c.closed != 0 && c.qcount == 0 {
        if raceenabled {
            raceacquire(unsafe.Pointer(c))
        }
        unlock(&c.lock)
        if ep != nil {
            typedmemclr(c.elemtype, ep)
        }
        return true, false
    }

接收和發送略有不同，當channel關閉並且channel的緩存隊列裡沒有數據瞭，那麼接收動作會直接結束，但不會報錯。

也就是說，允許從已關閉的channel中接收數據。

從發送等待協程中接收

if sg := c.sendq.dequeue(); sg != nil {
        // Found a waiting sender. If buffer is size 0, receive value
        // directly from sender. Otherwise, receive from head of queue
        // and add sender's value to the tail of the queue (both map to
        // the same buffer slot because the queue is full).
        recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true, true
    }

嘗試從發送等待協程列表中取出一個等待協程，如果存在，則調用recv()方法接收數據。

這裡的recv()方法比send()方法稍微復雜一點，我們簡單分析下。

func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
    if c.dataqsiz == 0 {
        ...
        if ep != nil {
            // copy data from sender
            recvDirect(c.elemtype, sg, ep)
        }
    } else {
        qp := chanbuf(c, c.recvx)
        ...
        // copy data from queue to receiver
        if ep != nil {
            typedmemmove(c.elemtype, ep, qp)
        }
        // copy data from sender to queue
        typedmemmove(c.elemtype, qp, sg.elem)
        c.recvx++
        if c.recvx == c.dataqsiz {
            c.recvx = 0
        }
        c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
    }
    sg.elem = nil
    gp := sg.g
    unlockf()
    gp.param = unsafe.Pointer(sg)
    if sg.releasetime != 0 {
        sg.releasetime = cputicks()
    }
    goready(gp, skip+1)
}

recv()的接收動作分為兩種情況：

  • c.dataqsiz == 0：即當channel為無緩存channel時，直接將發送協程中的數據，拷貝給接收者。
  • c.dataqsiz != 0：如果channel有緩存，則：根據緩存的接收遊標，從緩存隊列中取出一個，拷貝給接受者

小結

channel必須初始化後才能使用；

channel關閉後，不允許在發送數據，但是還可以繼續從中接收未處理完的數據。所以盡量從發送端關閉channel；

無緩存的channel需要註意在一個協程中的操作不會造成死鎖；

到此這篇關於Golang通道channel的源碼分析的文章就介紹到這瞭,更多相關Golang通道channel內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：