网站首页 返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

GO源码学习之channel

2020-06-10 11:39:44 admin 2053

# 前言

channel是golang中标志性的概念之一，很好很强大！  
channel（通道），顾名思义，是一种通道，一种用于并发环境中数据传递的通道。通常结合golang中另一重要概念goroutine（go协程）使用，使得在golang中的并发编程变得清晰简洁同时又高效强大。  
今天尝试着读读golang对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()`的接收动作分为两种情况：

1. c.dataqsiz == 0：即当channel为无缓存channel时，直接将发送协程中的数据，拷贝给接收者。
  2. c.dataqsiz != 0：如果channel有缓存，则：

* 根据缓存的接收游标，从缓存队列中取出一个，拷贝给接受者

* 将发送协程中的数据，放到空出来的缓存位置中，游标下移。(即将新数据接到队列尾巴上)

* 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.selectdone = nil
        mysg.c = c
        gp.param = nil
        c.recvq.enqueue(mysg)
        goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)

没有协程等待发送，缓存中也没有数据了，那么之后阻塞接收协程，等待合适时机在接收数据。  
同发送过程一样，将当前协程封装到`sudog`中，链接到`recvq`列表中。并休眠当前协程。

# 总结

* channel必须初始化后才能使用
  * channel关闭后，不允许在发送数据，但是还可以继续从中接收未处理完的数据。所以尽量从发送端关闭channel
  * 无缓存的channel需要注意在一个协程中的操作不会造成死锁

# 遗留问题

* `hchanSize`的计算
  * `maxAlign`参数的作用
  * 内存分配
  * 设计思想的梳理

> 附注1：源码基于go1.9.2  
>  附注2：文章中引用的源码`...`处表示有删减

转载文章，原文链接： GO源码学习之channel

关键字词go 源码

分享到：

上一篇：Kubernetes、Docker 等容器技术相关博客汇总

下一篇：Go语言实现RPC

如需留言，请登录，没有账号？请注册

0 条评论 0 人参与

网站首页 返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

GO源码学习之channel

admin 1年前

zhou 2年前

zhou 2年前

zhou 2年前

zhou 2年前

网站首页 返回列表 像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

GO源码学习之channel

admin 官方 1年前

zhou 2年前

zhou 2年前

zhou 2年前

zhou 2年前

网站首页返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

admin 1年前