golang chan

chan是我们学习golang绕不开的一个话题，今天我就不讲基础的使用了，因为太多这种文章了，我讲一下channel底层的实现和它的数据结构

必须了解的数据结构

type hchan struct {
    qcount   uint           // 所有数据
    dataqsiz uint           // 数据size
    buf      unsafe.Pointer // 指向真实数据的指针
    elemsize uint16 
    closed   uint32
    elemtype *_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
}
//qcount：代表 chan 中已经接收但还没被取走的元素的个数。=len（ch）
//dataqsiz：队列的大小。chan 使用一个循环队列来存放元素
// buf：存放元素的循环队列的 buffer。
//elemtype 和 elemsize：chan 元素的类型和元素大小，一般分为普通类型和指针类型，在makechan函数中会判断是否为指针类型
//sendx：处理发送数据的指针在 buf 中的位置。一旦接收了新的数据，指针就会加上 elemsize，移向下一个位置。buf 的总大小是 elemsize 的整数倍，而且 
//buf 是一个循环列表。
//recvx：处理接收请求时的指针在 buf 中的位置。一旦取出数据，此指针会移动到下一个位置。
//recvq：chan 是多生产者多消费者的模式，如果消费者因为没有数据可读而阻塞了，就会被加入到 recvq 队列中。
//sendq：如果生产者因为 buf 满了而阻塞，会被加入到 sendq 队列中。

type waitq struct {
    first *sudog
    last  *sudog
}
// 就是gopark掉了的goroutine队列

//单个节点结构体
type sudog struct {
    g *g
    isSelect bool
    next     *sudog
    prev     *sudog
    elem     unsafe.Pointer // data element (may point to stack)
    acquiretime int64
    releasetime int64
    ticket      uint32
    parent      *sudog // semaRoot binary tree
    waitlink    *sudog // g.waiting list or semaRoot
    waittail    *sudog // semaRoot
    c           *hchan // channel
}

在说channel之前你应该了解的

用通俗一点的话来说，channel实际底层并发是靠锁lock实现的，数据写入buf和从buf读出来都有加锁的动作，
存储数据是靠一个循环队列来保存的，队列的大小就是属性buf指向的队列长度,队列长度（len）就是qcount,
实际存储数据是buf的指向的循环队列，然后sendx和recvx是来控制数据读取和删除的，顺序就是按照buf的
索引顺序来的。然后还会保存两个队列：sendq，recvq,这两个参数实际上是绑定wait状态下的g，也就是
被park掉的goroutine,要理解这句话就要有点gmp模型的概念。

1.我们创建一个buffer channel

image.png

2.此时的结构体应该是这样的

image.png

3.放入一个数据到chan

image.png

4.取走一个数据

image.png

5.那么简单的存放数据知道了，它是如何将goroutine阻塞的呢---先将一下GMP

因为这个图是网上扣下来的，不知道原图是谁的，感觉画的很不错

image.png image.png

这里在做个简单的解释：后面专门会出一个文章来讲解一下GMP调度模型。M是一个内核线程的一个映射（实际上的原理比这个复杂许多，涉及到线程模型，后面写篇文章细讲），P是我们的调度器，一般又称为schedule，正常情况下，我们在代码执行写了一个go func,那么这个go在代码编译的时候就会被丢入一个P的队列中：就像一堆排队的地鼠G去领钱，P就是包工头，第一个地鼠把自己的挖的洞的照片给P看，P看完就给地鼠拍拍灰（准备环境），就让G去M里面去领钱，（领钱就是执行），M呢是个小金库，但是每个地鼠只有10ms时间去领钱，过了就会被p强制的给退出来，有的地鼠不用10ms就能自己退出来

chan如何将goroutine阻塞的

image.png

这里需要了解一下gopark：将goroutine进行休眠

此时GMP应该是这个样子

image.png

此时G就没有人执行了，所以就看起来像卡住了一样，那么就有个问题了：怎么让它继续运行呢，我们不是将chan读出来goroutine又能继续执行了吗？

此时Chan的结构体变成了这个样子

image.png

哪些被游离的gorouine 就被chan的队列给抓住了，放到它的发送或者等待队列中。

此时这里sendq/recvq长这个样子

image.png

这里怎么读呢：我就简单解释一下，没这个图，我想自己画一个发现太丑了
此时buf的被别人取走了一个元素，那么就从sendqpop一个出来，是从头部开始pop，然后元素放到了buf，goroutine被设置成runable状态，然后放到P的runable队列中去继续执行下文。
上面是说的发送chan被占满

如果先来读chan被阻塞，chan内部长什么样子呢

image.png

那么chan怎么处理先读后写的这种场景

按照上面的套路，其实我们可以想，写的时候，先写到buf，然后recvq检测到buf有值了，将buf 的数据pop出来，将goroutine唤醒，数据写入elem指针指向的地址。但是有没有更好的套路呢？

其实根据场景：先有等待者，然后发送者来了，就像去银行存钱一样，取钱的人来了，但是银行没钱，后面存钱的人来了。最大的不同就是gorotuine不需要到chan，就想存钱取钱的人可以不用在银行操作，那么是不是存钱的人就可以直接把钱给取钱的人

实际上的代码也是这么来处理的，就是send 的goroutine直接写了recv的stack，就减少了lock的启动和释放，提高了性能。但是也只有only operations in runtime where this happen：运行时发生这样的情况。

• 基本上所有的流程就是上面这个图片，下面是比较重要的一些方法，可以按照上面理解去下面看看，不过最好能用goland调试一下，看代码很难看懂，我在看源码的时候，确认chan 数据类型是指针还是普通数据就踩坑了，初次创建的时候调用下面的makechan，我是直接跑到runtime里面去看，发现断点没有走预期路线，我还调试了好久，才发现它的上层是反复调用这个方法，调用了3次，才返回实例，这里面我就不细细的讲了，大家有什么建议也可以评论一下，大家共同学习进步

创建chan:makechan

func makechan(t *chantype, size int) *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 {
        thro("makechan: bad alignment")
    }
    mem, overflow := math.MulUintptr(elem.size, uintptr(size))
    if overflow || mem > maxAlloc-hchanSize || size < 0 {
        panic(plainError("makechan: size out of range"))
    }
    // 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 {
    case mem == 0:
        // Queue or element size is zero.
        c = (*hchan)(mallocgc(hchanSize, nil, true))
        // Race detector uses this location for synchronization.
        c.buf = c.raceaddr()
    case elem.ptrdata == 0:
        // Elements do not contain pointers.
        // Allocate hchan and buf in one call.
        c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
        c.buf = add(unsafe.Pointer(c), hchanSize)
    default:
        // Elements contain pointers.
        c = new(hchan)
        c.buf = mallocgc(mem, elem, true)
    }
    c.elemsize = uint16(elem.size)
    c.elemtype = elem
    c.dataqsiz = uint(size)
    lockInit(&c.lock, lockRankHchan)
    if debugChan {
        print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")

    }
    return c
}

chansend

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
    if c == nil {
        if !block {
            return false
        }
        gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
        throw("unreachable")
    }
    if debugChan {
        print("chansend: chan=", c, "\n")
    }
    if raceenabled {
        racereadpc(c.raceaddr(), callerpc, funcPC(chansend))
    }
    if !block && c.closed == 0 && full(c) {
        return false
    }
    var t0 int64
    if blockprofilerate > 0 {
        t0 = cputicks()
    }
    lock(&c.lock)
    if c.closed != 0 {
        unlock(&c.lock)
        panic(plainError("send on closed channel"))
    }
    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
    }
    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
    }
    if !block {
        unlock(&c.lock)
        return false
    }
    // 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.isSelect = false
    mysg.c = c
    gp.waiting = mysg
    gp.param = nil
    c.sendq.enqueue(mysg)
    // Signal to anyone trying to shrink our stack that we're about
    // to park on a channel. The window between when this G's status
    // changes and when we set gp.activeStackChans is not safe for
    // stack shrinking.
    atomic.Store8(&gp.parkingOnChan, 1)
    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
    // Ensure the value being sent is kept alive until the
    // receiver copies it out. The sudog has a pointer to the
    // stack object, but sudogs aren't considered as roots of the
    // stack tracer.
    KeepAlive(ep)
    // someone woke us up.
    if mysg != gp.waiting {
        throw("G waiting list is corrupted")
    }
    gp.waiting = nil
    gp.activeStackChans = false
    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
}

close chan

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()
        racewritepc(c.raceaddr(), callerpc, funcPC(closechan))
        racerelease(c.raceaddr())
    }
    c.closed = 1
    var glist gList
    // 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, c.raceaddr())
        }
        glist.push(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, c.raceaddr())
        }
        glist.push(gp)
    }
    unlock(&c.lock)
    // Ready all Gs now that we've dropped the channel lock.
    for !glist.empty() {
        gp := glist.pop()
        gp.schedlink = 0
        goready(gp, 3)
    }
}

chanrecv方法

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")
    }
    if c == nil {
        if !block {
            return
        }
        gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
        throw("unreachable")
    }
    // Fast path: check for failed non-blocking operation without acquiring the lock.
    if !block && empty(c) {
        if atomic.Load(&c.closed) == 0 {
            return
        }
        if empty(c) {
            // The channel is irreversibly closed and empty.
            if raceenabled {
                raceacquire(c.raceaddr())
            }
            if ep != nil {
                typedmemclr(c.elemtype, ep)
            }
            return true, false
        }
    }
    var t0 int64
    if blockprofilerate > 0 {
        t0 = cputicks()
    }
    lock(&c.lock)
    if c.closed != 0 && c.qcount == 0 {
        if raceenabled {
            raceacquire(c.raceaddr())
        }
        unlock(&c.lock)
        if ep != nil {
            typedmemclr(c.elemtype, ep)
        }
        return true, false
    }
    if sg := c.sendq.dequeue(); sg != nil {
        recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true, true
    }
    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
    }
    // 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)
    atomic.Store8(&gp.parkingOnChan, 1)
    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)
    // someone woke us up
    if mysg != gp.waiting {
        throw("G waiting list is corrupted")
    }
    gp.waiting = nil
    gp.activeStackChans = false
    if mysg.releasetime > 0 {
        blockevent(mysg.releasetime-t0, 2)
    }
    closed := gp.param == nil
    gp.param = nil
    mysg.c = nil
    releaseSudog(mysg)
    return true, !closed
}