参考 https://zhuanlan.zhihu.com/p/74613114

channel 实现原理

  • 对chan的接收和发送都会在编译期间转换成为底层的发送接收函数。
  • channel 分为两种：带缓冲、不带缓冲。对不带缓冲的 channel 进行的操作实际上可以看作“同步模式”，带缓冲的则称为“异步模式”。
  • 同步模式下，发送接收方都要都要同步就绪，只有在双方都处于ready状态，数据才能在两者之间传输（实际上是内存拷贝）。否则，任意一方都会被挂起，等待另一方来唤醒。
  • 异步模式下，在缓冲区有剩余槽位时（有剩余容量），发送接收操作能顺利执行，不会被阻塞以致挂起。否则，操作的一方（如写入）同样会被挂起，直到出现相反操作（如接收）才会被唤醒。

chan 数据结构

  1. type hchan struct {
  2.     // chan 里元素数量
  3.     qcount   uint
  4.     // chan 底层循环数组的长度
  5.     dataqsiz uint
  6.     // 指向底层循环数组的指针
  7.     // 只针对有缓冲的 channel
  8.     buf      unsafe.Pointer
  9.     // chan 中元素大小
  10.     elemsize uint16
  11.     // chan 是否被关闭的标志
  12.     closed   uint32
  13.     // chan 中元素类型
  14.     elemtype *_type // element type
  16.     // 已发送元素在循环数组中的索引
  17.     sendx    uint   // send index
  18.     // 已接收元素在循环数组中的索引
  19.     recvx    uint   // receive index
  21.     // 等待接收的 goroutine 队列
  22.     recvq    waitq  // list of recv waiters
  23.     // 等待发送的 goroutine 队列
  24.     sendq    waitq  // list of send waiters
  26.     // 保护 hchan 中所有字段
  27.     lock mutex
  28. }
  • buf 指向底层循环数组，只有缓冲型的 channel 才有。
  • lock 用来保证每个读 channel 或写 channel 的操作都是原子的。
  • sendx，recvx 均指向底层循环数组，表示当前可以发送和接收的元素位置索引值（相对于底层数组）。
  • sendq，recvq 分别表示被阻塞的 goroutine，这些 goroutine 由于尝试读取 channel 或向 channel 发送数据而被阻塞。 
  1. type waitq struct {
  2.     first *sudog
  3.     last  *sudog
  4. }

waitq 是 sudog 的一个双向链表，而 sudog 实际上是对 goroutine 的一个封装。

1. Go内置数据结构 - 图1

操作方法实现

make

  1. const hchanSize = unsafe.Sizeof(hchan{}) + uintptr(-int(unsafe.Sizeof(hchan{}))&(maxAlign-1))
  3. func makechan(t *chantype, size int64) *hchan {
  4.     elem := t.elem
  6.     // 省略了检查 channel size，align 的代码
  7.     // ……
  9.     var c *hchan
  10.     // 如果元素类型不含指针 或者 size 大小为 0（无缓冲类型）
  11.     // 只进行一次内存分配
  12.     if elem.kind&kindNoPointers != 0 || size == 0 {
  13.         // 如果 hchan 结构体中不含指针，GC 就不会扫描 chan 中的元素
  14.         // 只分配 "hchan 结构体大小 + 元素大小*个数" 的内存
  15.         c = (*hchan)(mallocgc(hchanSize+uintptr(size)*elem.size, nil, true))
  16.         // 如果是缓冲型 channel 且元素大小不等于 0（大小等于 0的元素类型：struct{}）
  17.         if size > 0 && elem.size != 0 {
  18.             c.buf = add(unsafe.Pointer(c), hchanSize)
  19.         } else {
  20.             // race detector uses this location for synchronization
  21.             // Also prevents us from pointing beyond the allocation (see issue 9401).
  22.             // 1. 非缓冲型的，buf 没用，直接指向 chan 起始地址处
  23.             // 2. 缓冲型的，能进入到这里，说明元素无指针且元素类型为 struct{}，也无影响
  24.             // 因为只会用到接收和发送游标，不会真正拷贝东西到 c.buf 处（这会覆盖 chan的内容）
  25.             c.buf = unsafe.Pointer(c)
  26.         }
  27.     } else {
  28.         // 进行两次内存分配操作
  29.         c = new(hchan)
  30.         c.buf = newarray(elem, int(size))
  31.     }
  32.     c.elemsize = uint16(elem.size)
  33.     c.elemtype = elem
  34.     // 循环数组长度
  35.     c.dataqsiz = uint(size)
  37.     // 返回 hchan 指针
  38.     return c
  39. }

recv

  1. // entry points for <- c from compiled code
  2. func chanrecv1(c *hchan, elem unsafe.Pointer) {
  3.     chanrecv(c, elem, true)
  4. }
  6. func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {
  7.     _, received = chanrecv(c, elem, true)
  8.     return
  9. }

chanrecv1 函数处理不带 “ok” 的情形，chanrecv2 则通过返回 “received” 这个字段来反应 channel 是否被关闭。接收值则比较特殊，会“放到”参数 elem 所指向的地址了，这很像 C/C++ 里的写法。如果代码里忽略了接收值，这里的 elem 为 nil。

无论如何，最终转向了 chanrecv 函数：

  1. // 位于 src/runtime/chan.go
  3. // chanrecv 函数接收 channel c 的元素并将其写入 ep 所指向的内存地址。
  4. // 如果 ep 是 nil，说明忽略了接收值。
  5. // 如果 block == false，即非阻塞型接收，在没有数据可接收的情况下，返回 (false, false)
  6. // 否则，如果 c 处于关闭状态，将 ep 指向的地址清零，返回 (true, false)
  7. // 否则，用返回值填充 ep 指向的内存地址。返回 (true, true)
  8. // 如果 ep 非空，则应该指向堆或者函数调用者的栈
  10. func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
  11.     // 省略 debug 内容 …………
  13.     // 如果是一个 nil 的 channel
  14.     if c == nil {
  15.         // 如果不阻塞，直接返回 (false, false)
  16.         if !block {
  17.             return
  18.         }
  19.         // 否则，接收一个 nil 的 channel，goroutine 挂起
  20.         gopark(nil, nil, "chan receive (nil chan)", traceEvGoStop, 2)
  21.         throw("unreachable") // 不会执行到这
  22.     }
  24.     // 非阻塞，buf没元素，
  25.     if !block && empty(c) {
  26.         // 未关闭，返回 (false, false)
  27.         if atomic.Load(&c.closed) == 0 {
  28.             return
  29.         }
  30.            // 已关闭。 因为 channel 不可能被重复打开。
  31.         // 它可能在上面的空检查和关闭检查之间到达，重新检查通道是否有待处理的数据接收，
  32.         if empty(c) {
  33.             if ep != nil {
  34.                 // 未忽略返回值，那么接收的值将是一个该类型的零值
  35.                 // typedmemclr 根据类型清理相应地址的内存，标为零值
  36.                 typedmemclr(c.elemtype, ep)
  37.             }
  38.             return true, false
  39.         }
  40.     }
  42.     var t0 int64
  43.     if blockprofilerate > 0 {
  44.         t0 = cputicks()
  45.     }
  47.     // 加锁
  48.     lock(&c.lock)
  50.     // 已关闭 && 循环数组 buf 里没有元素
  51.     // 这里可以处理 非缓冲型关闭 和 缓冲型 buf 无元素关闭 两种情况
  52.     // 也就是说即使是关闭状态，但在缓冲型的 channel，buf 里有元素的情况下还能接收到元素
  53.     if c.closed != 0 && c.qcount == 0 {
  54.         if raceenabled {
  55.             raceacquire(unsafe.Pointer(c))
  56.         }
  57.         // 解锁
  58.         unlock(&c.lock)
  59.         if ep != nil {
  60.             // 从一个已关闭的并且空的 channel 执行接收操作，且未忽略返回值
  61.             // 那么接收的值将是一个该类型的零值
  62.             // typedmemclr 根据类型清理相应地址的内存，标为零值
  63.             typedmemclr(c.elemtype, ep)
  64.         }
  65.         // 从一个已关闭的 channel 接收，selected 会返回true
  66.         return true, false
  67.     }
  69.     // 等待发送队列里有 goroutine 存在，说明 buf 是满的
  70.     // 这有可能是：
  71.     // 1. 非缓冲型的 channel
  72.     // 2. 缓冲型的 channel，但 buf 满了
  73.     // 针对 1，直接进行内存拷贝（从 sender goroutine -> receiver goroutine）
  74.     // 针对 2，接收到循环数组头部的元素，并将发送者的元素放到循环数组尾部
  75.     if sg := c.sendq.dequeue(); sg != nil {
  76.         // Found a waiting sender. If buffer is size 0, receive value
  77.         // directly from sender. Otherwise, receive from head of queue
  78.         // and add sender's value to the tail of the queue (both map to
  79.         // the same buffer slot because the queue is full).
  80.         recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
  81.         return true, true
  82.     }
  84.     // 下面逻辑是 没有阻塞在sendq 的 g
  86.     // 缓冲型，buf 里有元素，可以正常接收
  87.     if c.qcount > 0 {
  88.         // 直接从循环数组里找到要接收的元素
  89.         qp := chanbuf(c, c.recvx)
  91.         // …………
  93.         // 代码里，没有忽略要接收的值，不是 "<- ch"，而是 "val <- ch"，ep 指向 val
  94.         if ep != nil {
  95.             typedmemmove(c.elemtype, ep, qp)
  96.         }
  97.         // 清理掉循环数组里相应位置的值
  98.         typedmemclr(c.elemtype, qp)
  99.         // 接收游标向前移动
  100.         c.recvx++
  101.         // 接收游标归零
  102.         if c.recvx == c.dataqsiz {
  103.             c.recvx = 0
  104.         }
  105.         // buf 数组里的元素个数减 1
  106.         c.qcount--
  107.         // 解锁
  108.         unlock(&c.lock)
  109.         return true, true
  110.     }
  112.     if !block {
  113.         // 非阻塞接收，解锁。selected 返回 false，因为没有接收到值
  114.         unlock(&c.lock)
  115.         return false, false
  116.     }
  118.     // 接下来就是要被阻塞接收的情况了
  119.     // 构造一个 sudog
  120.     gp := getg()            // p上运行的g
  121.     mysg := acquireSudog() // 新创建的g
  122.     mysg.releasetime = 0
  123.     if t0 != 0 {
  124.         mysg.releasetime = -1
  125.     }
  127.     // 待接收数据的地址保存下来
  128.     mysg.elem = ep
  129.     mysg.waitlink = nil
  130.     gp.waiting = mysg
  131.     mysg.g = gp
  132.     mysg.selectdone = nil
  133.     mysg.c = c
  134.     gp.param = nil
  135.     // 进入channel 的等待接收队列
  136.     c.recvq.enqueue(mysg)
  137.     // 将当前 goroutine 挂起
  138.     goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)
  140.     // 被唤醒了，接着从这里继续执行一些扫尾工作
  141.     if mysg != gp.waiting {
  142.         throw("G waiting list is corrupted")
  143.     }
  144.     gp.waiting = nil
  145.     if mysg.releasetime > 0 {
  146.         blockevent(mysg.releasetime-t0, 2)
  147.     }
  148.     closed := gp.param == nil
  149.     gp.param = nil
  150.     mysg.c = nil
  151.     releaseSudog(mysg)
  152.     return true, !closed
  153. }

send

  1. func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  2.     if c == nil {
  3.         // 不能阻塞，直接返回 false，表示未发送成功
  4.         if !block {
  5.             return false
  6.         }
  7.         // 当前 goroutine 被挂起
  8.         gopark(nil, nil, "chan send (nil chan)", traceEvGoStop, 2)
  9.         throw("unreachable")
  10.     }
  12.     // 省略 debug 相关……
  14.     // 对于不阻塞的 send，快速检测失败场景
  15.     //
  16.     // 如果 channel 未关闭且 channel 没有多余的缓冲空间。这可能是：
  17.     // 1. channel 是非缓冲型的，且等待接收队列里没有 goroutine
  18.     // 2. channel 是缓冲型的，但循环数组已经装满了元素
  19.     if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||
  20.         (c.dataqsiz > 0 && c.qcount == c.dataqsiz)) {
  21.         return false
  22.     }
  24.     var t0 int64
  25.     if blockprofilerate > 0 {
  26.         t0 = cputicks()
  27.     }
  29.     // 锁住 channel，并发安全
  30.     lock(&c.lock)
  32.     // 如果 channel 关闭了
  33.     if c.closed != 0 {
  34.         unlock(&c.lock)
  35.         panic(plainError("send on closed channel"))
  36.     }
  38.     // 如果接收队列里有 goroutine，直接将要发送的数据拷贝到接收 goroutine
  39.     if sg := c.recvq.dequeue(); sg != nil {
  40.         send(c, sg, ep, func() { unlock(&c.lock) }, 3)
  41.         return true
  42.     }
  44.     // 对于缓冲型的 channel，如果还有缓冲空间
  45.     if c.qcount < c.dataqsiz {
  46.         // qp 指向 buf 的 sendx 位置
  47.         qp := chanbuf(c, c.sendx)
  49.         // ……
  51.         // 将数据从 ep 处拷贝到 qp
  52.         typedmemmove(c.elemtype, qp, ep)
  53.         // 发送游标值加 1
  54.         c.sendx++
  55.         // 如果发送游标值等于容量值，游标值归 0
  56.         if c.sendx == c.dataqsiz {
  57.             c.sendx = 0
  58.         }
  59.         // 缓冲区的元素数量加一
  60.         c.qcount++
  62.         unlock(&c.lock)
  63.         return true
  64.     }
  66.     // 如果不需要阻塞，则直接返回错误
  67.     if !block {
  68.         unlock(&c.lock)
  69.         return false
  70.     }
  72.     // channel 满了，发送方会被阻塞。接下来会构造一个 sudog
  74.     // 获取当前 goroutine 的指针
  75.     gp := getg()
  76.     mysg := acquireSudog()
  77.     mysg.releasetime = 0
  78.     if t0 != 0 {
  79.         mysg.releasetime = -1
  80.     }
  82.     mysg.elem = ep
  83.     mysg.waitlink = nil
  84.     mysg.g = gp
  85.     mysg.selectdone = nil
  86.     mysg.c = c
  87.     gp.waiting = mysg
  88.     gp.param = nil
  90.     // 当前 goroutine 进入发送等待队列
  91.     c.sendq.enqueue(mysg)
  93.     // 当前 goroutine 被挂起
  94.     goparkunlock(&c.lock, "chan send", traceEvGoBlockSend, 3)
  96.     // 从这里开始被唤醒了（channel 有机会可以发送了）
  97.     if mysg != gp.waiting {
  98.         throw("G waiting list is corrupted")
  99.     }
  100.     gp.waiting = nil
  101.     if gp.param == nil {
  102.         if c.closed == 0 {
  103.             throw("chansend: spurious wakeup")
  104.         }
  105.         // 被唤醒后，channel 关闭了。坑爹啊，panic
  106.         panic(plainError("send on closed channel"))
  107.     }
  108.     gp.param = nil
  109.     if mysg.releasetime > 0 {
  110.         blockevent(mysg.releasetime-t0, 2)
  111.     }
  112.     // 去掉 mysg 上绑定的 channel
  113.     mysg.c = nil
  114.     releaseSudog(mysg)
  115.     return true
  116. }
  • 如果检测到 channel 是nil的，当前 goroutine 会被挂起。
  • 对于不阻塞的发送操作，如果 channel 未关闭并且没有多余的缓冲空间（说明：a. channel 是非缓冲型的，且等待接收队列里没有 goroutine；b. channel 是缓冲型的，但循环数组已经装满了元素），会返回false

close

  1. func closechan(c *hchan) {
  2.     // 关闭一个 nil channel，panic
  3.     if c == nil {
  4.         panic(plainError("close of nil channel"))
  5.     }
  7.     // 上锁
  8.     lock(&c.lock)
  9.     // 如果 channel 已经关闭
  10.     if c.closed != 0 {
  11.         unlock(&c.lock)
  12.         // panic
  13.         panic(plainError("close of closed channel"))
  14.     }
  16.     // …………
  18.     // 修改关闭状态
  19.     c.closed = 1
  21.     var glist *g
  23.     // 将 channel 所有等待接收队列的里 sudog 释放
  24.     for {
  25.         // 从接收队列里出队一个 sudog
  26.         sg := c.recvq.dequeue()
  27.         // 出队完毕，跳出循环
  28.         if sg == nil {
  29.             break
  30.         }
  32.         // 如果 elem 不为空，说明此 receiver 未忽略接收数据
  33.         // 给它赋一个相应类型的零值
  34.         if sg.elem != nil {
  35.             typedmemclr(c.elemtype, sg.elem)
  36.             sg.elem = nil
  37.         }
  38.         if sg.releasetime != 0 {
  39.             sg.releasetime = cputicks()
  40.         }
  41.         // 取出 goroutine
  42.         gp := sg.g
  43.         gp.param = nil
  44.         if raceenabled {
  45.             raceacquireg(gp, unsafe.Pointer(c))
  46.         }
  47.         // 相连，形成链表
  48.         gp.schedlink.set(glist)
  49.         glist = gp
  50.     }
  52.     // 将 channel 等待发送队列里的 sudog 释放
  53.     // 如果存在，这些 goroutine 将会 panic
  54.     for {
  55.         // 从发送队列里出队一个 sudog
  56.         sg := c.sendq.dequeue()
  57.         if sg == nil {
  58.             break
  59.         }
  61.         // 发送者会 panic
  62.         sg.elem = nil
  63.         if sg.releasetime != 0 {
  64.             sg.releasetime = cputicks()
  65.         }
  66.         gp := sg.g
  67.         gp.param = nil
  68.         if raceenabled {
  69.             raceacquireg(gp, unsafe.Pointer(c))
  70.         }
  71.         // 形成链表
  72.         gp.schedlink.set(glist)
  73.         glist = gp
  74.     }
  75.     // 解锁
  76.     unlock(&c.lock)
  78.     // Ready all Gs now that we've dropped the channel lock.
  79.     // 遍历链表
  80.     for glist != nil {
  81.         // 取最后一个
  82.         gp := glist
  83.         // 向前走一步，下一个唤醒的 g
  84.         glist = glist.schedlink.ptr()
  85.         gp.schedlink = 0
  86.         // 唤醒相应 goroutine
  87.         goready(gp, 3)
  88.     }
  89. }