chan的理解

chan用于協(xié)程間通信，結(jié)構(gòu)體如下，代碼位置為go/src/runtime/chan.go

type hchan struct {
    qcount   uint           // total data in the queue；隊(duì)列中的數(shù)據(jù)數(shù)量
    dataqsiz uint           // size of the circular queue；chan的大小
    buf      unsafe.Pointer // points to an array of dataqsiz elements；存放數(shù)據(jù)的buffer
    elemsize uint16 //存放的數(shù)據(jù)類型大小
    closed   uint32//是否已關(guān)閉
    elemtype *_type // element type；存放數(shù)據(jù)的類型
    sendx    uint   // send index；發(fā)送數(shù)據(jù)時的buf位置
    recvx    uint   // receive index；讀取數(shù)據(jù)時的buf位置
    recvq    waitq  // list of recv waiters；讀取數(shù)據(jù)引起阻塞的go協(xié)程隊(duì)列
    sendq    waitq  // list of send waiters；寫數(shù)據(jù)引起阻塞的go協(xié)程隊(duì)列

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

type waitq struct {
    first *sudog
    last  *sudog
}
// sudog represents a g in a wait list, such as for sending/receiving
// on a channel.
//
// sudog is necessary because the g ? synchronization object relation
// is many-to-many. A g can be on many wait lists, so there may be
// many sudogs for one g; and many gs may be waiting on the same
// synchronization object, so there may be many sudogs for one object.
//
// sudogs are allocated from a special pool. Use acquireSudog and
// releaseSudog to allocate and free them.
type sudog struct {
    // The following fields are protected by the hchan.lock of the
    // channel this sudog is blocking on. shrinkstack depends on
    // this.

    g          *g//協(xié)程
    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)

    // The following fields are never accessed concurrently.
    // waitlink is only accessed by g.

    acquiretime int64
    releasetime int64
    ticket      uint32
    waitlink    *sudog // g.waiting list
    c           *hchan // channel
}

從結(jié)構(gòu)定義可以看出，chan包含了2個部分：1是讀寫協(xié)程等待隊(duì)列、數(shù)據(jù)存儲buffer。

chan操作

chan包含4類操作：make、read、write以及close

make chan

c := make(chan int,2)
編譯器會將make語句最終，指向

func reflect_makechan(t *chantype, size int64) *hchan {
    return makechan(t, size)
}

可以看得出來，返回的是一個hchan的指針
下面是實(shí)際的makechan的代碼

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
}

1、檢查待存的數(shù)據(jù)類型大小，大于1<<16時異常
2、檢查內(nèi)存對齊（降低尋址次數(shù)，提高內(nèi)存讀取速度），大于最大的內(nèi)存對齊字節(jié)數(shù)時，panic
3、檢查傳入的size大小，大于堆可分配的最大內(nèi)存時，panic，可以看出chan是在堆里面分配內(nèi)存的

    if size < 0 || int64(uintptr(size)) != size || (elem.size > 0 && uintptr(size) > (_MaxMem-hchanSize)/elem.size) {
        panic(plainError("makechan: size out of range"))
    }

4、存儲元素的類型沒有指針類型或者chan的大小為0時，分配連續(xù)地址空間（為什么這么做呢？），注意到size為0時，是不會為chan的buf malloc內(nèi)存空間的

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

send 即 c <- e

首先看chan為nil的情況

if c == nil {
  if !block {
      return false
  }
  gopark(nil, nil, "chan send (nil chan)", traceEvGoStop, 2)
  throw("unreachable")
}

chan為nil時，調(diào)用gopark進(jìn)入休眠狀態(tài)，并使用unlockf來喚醒，如下

// Puts the current goroutine into a waiting state and calls unlockf.
// If unlockf returns false, the goroutine is resumed.
// unlockf must not access this G's stack, as it may be moved between
// the call to gopark and the call to unlockf.
func gopark(unlockf func(*g, unsafe.Pointer) bool, lock unsafe.Pointer, reason string, traceEv byte, traceskip int) {
    mp := acquirem()
    gp := mp.curg
    status := readgstatus(gp)
    if status != _Grunning && status != _Gscanrunning {
        throw("gopark: bad g status")
    }
    mp.waitlock = lock
    mp.waitunlockf = *(*unsafe.Pointer)(unsafe.Pointer(&unlockf))
    gp.waitreason = reason
    mp.waittraceev = traceEv
    mp.waittraceskip = traceskip
    releasem(mp)
    // can't do anything that might move the G between Ms here.
    mcall(park_m)
}

注意到調(diào)用gopark時傳入的unlockf為nil，會被一直休眠，recv也是同樣的做法，因此沒有初始化進(jìn)行同時讀寫時，會引起死鎖

var c chan int
    go func() {
        <-c
    }()
    c <- 1

這段代碼執(zhí)行會報錯：fatal error: all goroutines are asleep - deadlock!

疑問的一段？

// Fast path: check for failed non-blocking operation without acquiring the lock.
    //
    // After observing that the channel is not closed, we observe that the channel is
    // not ready for sending. Each of these observations is a single word-sized read
    // (first c.closed and second c.recvq.first or c.qcount depending on kind of channel).
    // Because a closed channel cannot transition from 'ready for sending' to
    // 'not ready for sending', even if the channel is closed between the two observations,
    // they imply a moment between the two when the channel was both not yet closed
    // and not ready for sending. We behave as if we observed the channel at that moment,
    // and report that the send cannot proceed.
    //
    // It is okay if the reads are reordered here: if we observe that the channel is not
    // ready for sending and then observe that it is not closed, that implies that the
    // channel wasn't closed during the first observation.
if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||
        (c.dataqsiz > 0 && c.qcount == c.dataqsiz)) {
        return false
    }

channel關(guān)閉后，再send時，直接panic

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

讀等待隊(duì)列中，有協(xié)程等待，這個時候直接將send的數(shù)據(jù)memmove到協(xié)程中elem元素中；goready喚醒阻塞的協(xié)程

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) })
        return true
    }

// send processes a send operation on an empty channel c.
// The value ep sent by the sender is copied to the receiver sg.
// The receiver is then woken up to go on its merry way.
// Channel c must be empty and locked.  send unlocks c with unlockf.
// sg must already be dequeued from c.
// ep must be non-nil and point to the heap or the caller's stack.
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func()) {
    if raceenabled {
        if c.dataqsiz == 0 {
            racesync(c, sg)
        } else {
            // Pretend we go through the buffer, even though
            // we copy directly. Note that we need to increment
            // the head/tail locations only when raceenabled.
            qp := chanbuf(c, c.recvx)
            raceacquire(qp)
            racerelease(qp)
            raceacquireg(sg.g, qp)
            racereleaseg(sg.g, qp)
            c.recvx++
            if c.recvx == c.dataqsiz {
                c.recvx = 0
            }
            c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
        }
    }
    if sg.elem != nil {
        sendDirect(c.elemtype, sg, ep)
        sg.elem = nil
    }
    gp := sg.g
    unlockf()
    gp.param = unsafe.Pointer(sg)
    if sg.releasetime != 0 {
        sg.releasetime = cputicks()
    }
    goready(gp, 4)
}

隊(duì)列buf沒有滿，將send數(shù)據(jù)寫入chan的buf中，并send指針后移，以及chan buf數(shù)據(jù)量增加

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
    }

chan隊(duì)列已滿，阻塞；將本協(xié)程放入等待協(xié)程中，同時休眠此協(xié)程

// 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.
    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)

send協(xié)程阻塞被喚醒：channel被close，panic；取到數(shù)據(jù)，正常返回

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

send 流程小結(jié)

chan為nil時，阻塞協(xié)程
chan closed時，panic
send有三種情況：1、等待隊(duì)列不為空，直接把數(shù)據(jù)發(fā)給等待協(xié)程 ；2、chan 的buf還有空間，把數(shù)據(jù)寫入buf；3、buf滿了，阻塞住協(xié)程，并放入chan的等待寫隊(duì)列

recv 也就是e := <-c

與send類似，整體流程如下：

chan為nil時，gopark阻塞協(xié)程
chan closed時，返回chan數(shù)據(jù)類型的默認(rèn)值，此時非阻塞

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
    }

recv有三種情況：1、等待隊(duì)列不為空，直接從等待寫協(xié)程 取出數(shù)據(jù)，并喚醒等待協(xié)程；2、chan 的buf還有數(shù)據(jù)，從buf中讀取數(shù)據(jù)；3、buf空，阻塞住協(xié)程，并放入chan的等待讀隊(duì)列

close

設(shè)置chan關(guān)閉標(biāo)志位，closed=1；取出chan的所有讀寫等待協(xié)程，改為就緒態(tài)，其中send協(xié)程，會panic；而recv協(xié)程會返回沒有被賦值的數(shù)據(jù)

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

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

    if raceenabled {
        callerpc := getcallerpc(unsafe.Pointer(&c))
        racewritepc(unsafe.Pointer(c), callerpc, funcPC(closechan))
        racerelease(unsafe.Pointer(c))
    }

    c.closed = 1

    var glist *g

    // release all readers
    for {
        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)
    }
}