Golang channel 初探

Goroutine和channel是Golang實現高並發的基礎。深入理解其背後的實現，寫起代碼來才不慌-_-

首先我們定義如下代碼，來看看Golang底層是如何實現channel的。

package main

import (
        "fmt"
)

func main() {
        c := make(chan int, 5)
        c <- 1

        g := <-c
        fmt.Println(g)

}

編譯後我們看下相關函數

go build -gcflags "-N -l" -o chan chan.go
go tool objdump -s "main\.main" chan
 

可以看到初始化調用了runtime.makechan，

寫channel調用了runtime.chansend1，

讀channel調用了runtime.chanrecv1.

在runtime/chan.go中我們找到對應函數

func makechan(t *chantype, size int) *hchan 

返回一個hchan結構，我們先看下channel的結構，我們為一眼可以理解的加點註釋。

channel結構

type hchan struct {
   qcount   uint           // total data in the queue 隊列中存在的個數
   dataqsiz uint           // size of the circular queue buffer大小 實現看起來是個循環數組
   buf      unsafe.Pointer // points to an array of dataqsiz elements 數組指針
   elemsize uint16       //channel類型的大小
   closed   uint32      //channel是否關閉
   elemtype *_type // element type //channel 類型
   sendx    uint   // send index  //發送index
   recvx    uint   // receive index //接收index
   recvq    waitq  // list of recv waiters //接收鏈表 即讀channel的goroutine
   sendq    waitq  // list of send waiters //發送鏈表 即寫channel的goroutine

   // 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
}
 

接下來來看下到底如何初始化的。

channel初始化

func makechan(t *chantype, size int) *hchan {
    elem := t.elem
   ...
    //一些合法判斷 
    

    // Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.
    // 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.
    var c *hchan
    switch {
    //channel buffer大小為0 或者類型大小為0
    case size == 0 || elem.size == 0:
        // Queue or element size is zero.
        c = (*hchan)(mallocgc(hchanSize, nil, true))
        // Race detector uses this location for synchronization.
        c.buf = unsafe.Pointer(c)
        
        //channel非指針
    case elem.kind&kindNoPointers != 0:
        // Elements do not contain pointers.
        // Allocate hchan and buf in one call.
        c = (*hchan)(mallocgc(hchanSize+uintptr(size)*elem.size, nil, true))
        c.buf = add(unsafe.Pointer(c), hchanSize)
    default:
        // Elements contain pointers.
        c = new(hchan)
        c.buf = mallocgc(uintptr(size)*elem.size, elem, true)
    }

    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
}
 

可以看出主要是根據channel類型以及buffer大小申請hcan.buf的內存，同時設置對應的datasiz、elemsize等，比較簡單。

那麽寫channel是怎麽實現的呢

寫channel

c<-1這種形式的寫channel會調用chansend1

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

看源碼還有select的selectnbsend，reflect的reflect_chansend，

它們最後都會調用chansend，所以只用看chansend的實現，它們只是參數不一樣而已。

/*
 * generic single channel send/recv
 * If block is not nil,
 * then the protocol will not
 * sleep but return if it could
 * not complete.
 *
 * sleep can wake up with g.param == nil
 * when a channel involved in the sleep has
 * been closed.  it is easiest to loop and re-run
 * the operation; we'll see that it's now closed.
 */
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
    //如果設置了block為true 且channel為nil，goroutine將會死在這
    if c == nil {
        if !block {
            return false
        }
        gopark(nil, nil, "chan send (nil chan)", traceEvGoStop, 2)
        throw("unreachable")
    }
    //忽略一些無用代碼
    .....
    //block false  channel沒buffer或者已滿 直接返回
    if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||
        (c.dataqsiz > 0 && c.qcount == c.dataqsiz)) {
        return false
    }

    ...
    
    lock(&c.lock)
    //寫已經關閉的channel 將會panic
    if c.closed != 0 {
        unlock(&c.lock)
        panic(plainError("send on closed channel"))
    }

    //優先檢查讀隊列是否有等待的goroutine，有的話直接調用並返回
    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有buffer可以寫
    if c.qcount < c.dataqsiz {
        // Space is available in the channel buffer. Enqueue the element to send.
        //根據sendx計算該數據在數組的位置
        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
    }
    //沒buffer 且block false直接返回
    if !block {
        unlock(&c.lock)
        return false
    }
   //沒buffer寫 初始化一個sudog結構
   
    // Block on the channel. Some receiver will complete our operation for us.
    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.
    
    //將該數據放在sudog的elem裏，所以沒buffer的channel數據其實在調用的goroutine裏
    mysg.elem = ep
    mysg.waitlink = nil
    mysg.g = gp
    mysg.isSelect = false
    mysg.c = c
    gp.waiting = mysg
    gp.param = nil
    
    //放到channel的寫隊列，並阻塞
    c.sendq.enqueue(mysg)
    goparkunlock(&c.lock, "chan send", traceEvGoBlockSend, 3)

    // someone woke us up.
    if mysg != gp.waiting {
        throw("G waiting list is corrupted")
    }
    gp.waiting = nil
    if gp.param == nil {
        if c.closed == 0 {
            throw("chansend: spurious wakeup")
        }
        panic(plainError("send on closed channel"))
    }
    gp.param = nil
    if mysg.releasetime > 0 {
        blockevent(mysg.releasetime-t0, 2)
    }
    mysg.c = nil
    releaseSudog(mysg)
    return true

?

? 整個寫channel的邏輯還是很清晰的。 參照這個其實讀channel差不多可以推斷出是個對應的邏輯。

讀channel

我們還是在runtime/chan.go中找到相關代碼

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

讀channel還有chanrecv2，selectnbrecv，selectnbrecv2，reflect_chanrecv。跟寫channel一樣，

它們之間也是參數的區別。

我們繼續看chanrecv

// chanrecv receives on channel c and writes the received data to ep.
// ep may be nil, in which case received data is ignored.
// If block == false and no elements are available, returns (false, false).
// Otherwise, if c is closed, zeros *ep and returns (true, false).
// Otherwise, fills in *ep with an element and returns (true, true).
// A non-nil ep must point to the heap or the caller's stack.
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
    // raceenabled: don't need to check ep, as it is always on the stack
    // or is new memory allocated by reflect.

    if debugChan {
        print("chanrecv: chan=", c, "\n")
    }
    //block true的時候讀nil的channel 將會永遠阻塞
    if c == nil {
        if !block {
            return
        }
        gopark(nil, nil, "chan receive (nil chan)", traceEvGoStop, 2)
        throw("unreachable")
    }
    
    //block false的時候 判斷channel buffer 已滿或者沒buffer 沒有讀的goroutine就直接返回
    if !block && (c.dataqsiz == 0 && c.sendq.first == nil ||
        c.dataqsiz > 0 && atomic.Loaduint(&c.qcount) == 0) &&
        atomic.Load(&c.closed) == 0 {
        return
    }

    

    lock(&c.lock)
     //如果channel已空 且沒有讀的數據 清除並退出
    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
    }
    //還是優先檢查寫隊列 拿出阻塞的goroutine
    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
    }
    //buffer中有數據 跟寫對應處理
    if c.qcount > 0 {
        // Receive directly from queue
        qp := chanbuf(c, c.recvx)
        if raceenabled {
            raceacquire(qp)
            racerelease(qp)
        }
        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
    }

    if !block {
        unlock(&c.lock)
        return false, false
    }
    //沒有數據讀 初始化一個sudog並阻塞
    // no sender available: block on this channel.
    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
    gp.waiting = mysg
    mysg.g = gp
    mysg.isSelect = false
    mysg.c = c
    gp.param = nil
    
    //把自己放到讀隊列
    c.recvq.enqueue(mysg)
    goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)

    // someone woke us up
    if mysg != gp.waiting {
        throw("G waiting list is corrupted")
    }
    gp.waiting = nil
    if mysg.releasetime > 0 {
        blockevent(mysg.releasetime-t0, 2)
    }
    closed := gp.param == nil
    gp.param = nil
    mysg.c = nil
    releaseSudog(mysg)
    return true, !closed
}

關channel

最後我們來看看close channel

func closechan(c *hchan) {
//關nil channel直接panic
    if c == nil {
        panic(plainError("close of nil channel"))
    }

    lock(&c.lock)
    //重復關閉 panic
    if c.closed != 0 {
        unlock(&c.lock)
        panic(plainError("close of closed channel"))
    }

    if raceenabled {
        callerpc := getcallerpc()
        racewritepc(unsafe.Pointer(c), callerpc, funcPC(closechan))
        racerelease(unsafe.Pointer(c))
    }
 //設置channel標記
    c.closed = 1

    var glist *g

    // release all readers
    for {
       //通知所有的讀goroutine
        sg := c.recvq.dequeue()
        if sg == nil {
            break
        }
        if sg.elem != nil {
            typedmemclr(c.elemtype, sg.elem)
            sg.elem = nil
        }
        if sg.releasetime != 0 {
            sg.releasetime = cputicks()
        }
        gp := sg.g
        gp.param = nil
        if raceenabled {
            raceacquireg(gp, unsafe.Pointer(c))
        }
        gp.schedlink.set(glist)
        glist = gp
    }

    // release all writers (they will panic)
    for {
        sg := c.sendq.dequeue()
        if sg == nil {
            break
        }
        sg.elem = nil
        if sg.releasetime != 0 {
            sg.releasetime = cputicks()
        }
        gp := sg.g
        gp.param = nil
        if raceenabled {
            raceacquireg(gp, unsafe.Pointer(c))
        }
        gp.schedlink.set(glist)
        glist = gp
    }
    unlock(&c.lock)

    // Ready all Gs now that we've dropped the channel lock.
    for glist != nil {
        gp := glist
        glist = glist.schedlink.ptr()
        gp.schedlink = 0
        goready(gp, 3)
    }
}

Golang channel實現