看过管道的底层源码吗

题目来源： 答案：

  • Channel原理
    channel的底层是一个hchan的结构体 基于锁实现的
  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
  15.     // 已发送元素在循环数组中的索引
  16.     sendx    uint   // send index
  17.     // 已接收元素在循环数组中的索引
  18.     recvx    uint   // receive index
  19.     // 等待接收的 goroutine 队列
  20.     recvq    waitq  // list of recv waiters
  21.     // 等待发送的 goroutine 队列
  22.     sendq    waitq  // list of send waiters
  24.     // 保护 hchan 中所有字段
  25.     lock mutex
  26. }
  • Channel 发送
  1. 如果当前 Channel 的 recvq 上存在已经被阻塞的 Goroutine，那么会直接将数据发送给当前 Goroutine 并将其设置成下一个运行的 Goroutine；
  2. 如果 Channel 存在缓冲区并且其中还有空闲的容量，我们会直接将数据存储到缓冲区 sendx 所在的位置上；
  3. 如果不满足上面的两种情况，会创建一个 runtime.sudog 结构并将其加入 Channel 的 sendq 队列中，当前 Goroutine 也会陷入阻塞等待其他的协程从 Channel 接收数据；
  1. func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  2.     // 如果 channel 是 nil
  3.     if c == nil {
  4.         // 不能阻塞，直接返回 false，表示未发送成功
  5.         if !block {
  6.             return false
  7.         }
  8.         // 当前 goroutine 被挂起
  9.         gopark(nil, nil, "chan send (nil chan)", traceEvGoStop, 2)
  10.         throw("unreachable")
  11.     }
  13.     // 省略 debug 相关……
  15.     // 对于不阻塞的 send，快速检测失败场景
  16.     //
  17.     // 如果 channel 未关闭且 channel 没有多余的缓冲空间。这可能是：
  18.     // 1. channel 是非缓冲型的，且等待接收队列里没有 goroutine
  19.     // 2. channel 是缓冲型的，但循环数组已经装满了元素
  20.     if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||
  21.         (c.dataqsiz > 0 && c.qcount == c.dataqsiz)) {
  22.         return false
  23.     }
  25.     var t0 int64
  26.     if blockprofilerate > 0 {
  27.         t0 = cputicks()
  28.     }
  30.     // 锁住 channel，并发安全
  31.     lock(&c.lock)
  33.     // 如果 channel 关闭了
  34.     if c.closed != 0 {
  35.         // 解锁
  36.         unlock(&c.lock)
  37.         // 直接 panic
  38.         panic(plainError("send on closed channel"))
  39.     }
  41.     // 如果接收队列里有 goroutine，直接将要发送的数据拷贝到接收 goroutine
  42.     if sg := c.recvq.dequeue(); sg != nil {
  43.         send(c, sg, ep, func() { unlock(&c.lock) }, 3)
  44.         return true
  45.     }
  47.     // 对于缓冲型的 channel，如果还有缓冲空间
  48.     if c.qcount < c.dataqsiz {
  49.         // qp 指向 buf 的 sendx 位置
  50.         qp := chanbuf(c, c.sendx)
  52.         // ……
  54.         // 将数据从 ep 处拷贝到 qp
  55.         typedmemmove(c.elemtype, qp, ep)
  56.         // 发送游标值加 1
  57.         c.sendx++
  58.         // 如果发送游标值等于容量值，游标值归 0
  59.         if c.sendx == c.dataqsiz {
  60.             c.sendx = 0
  61.         }
  62.         // 缓冲区的元素数量加一
  63.         c.qcount++
  65.         // 解锁
  66.         unlock(&c.lock)
  67.         return true
  68.     }
  70.     // 如果不需要阻塞，则直接返回错误
  71.     if !block {
  72.         unlock(&c.lock)
  73.         return false
  74.     }
  76.     // channel 满了，发送方会被阻塞。接下来会构造一个 sudog
  78.     // 获取当前 goroutine 的指针
  79.     gp := getg()
  80.     mysg := acquireSudog()
  81.     mysg.releasetime = 0
  82.     if t0 != 0 {
  83.         mysg.releasetime = -1
  84.     }
  86.     mysg.elem = ep
  87.     mysg.waitlink = nil
  88.     mysg.g = gp
  89.     mysg.selectdone = nil
  90.     mysg.c = c
  91.     gp.waiting = mysg
  92.     gp.param = nil
  94.     // 当前 goroutine 进入发送等待队列
  95.     c.sendq.enqueue(mysg)
  97.     // 当前 goroutine 被挂起
  98.     goparkunlock(&c.lock, "chan send", traceEvGoBlockSend, 3)
  100.     // 从这里开始被唤醒了（channel 有机会可以发送了）
  101.     if mysg != gp.waiting {
  102.         throw("G waiting list is corrupted")
  103.     }
  104.     gp.waiting = nil
  105.     if gp.param == nil {
  106.         if c.closed == 0 {
  107.             throw("chansend: spurious wakeup")
  108.         }
  109.         // 被唤醒后，channel 关闭了。坑爹啊，panic
  110.         panic(plainError("send on closed channel"))
  111.     }
  112.     gp.param = nil
  113.     if mysg.releasetime > 0 {
  114.         blockevent(mysg.releasetime-t0, 2)
  115.     }
  116.     // 去掉 mysg 上绑定的 channel
  117.     mysg.c = nil
  118.     releaseSudog(mysg)
  119.     return true
  120. }
  • Channel 接收
  1. 如果Channel为空，那么会直接调用 runtime.gopark 挂起当前 Goroutine；
  2. 如果Channel已经关闭并且缓冲区没有任何数据，runtime.chanrecv 会直接返回；
  3. 如果Channel的sendq队列中存在挂起的Goroutine，会将recvx索引所在的数据拷贝到接收变量所在的内存空间上并将 sendq 队列中 Goroutine 的数据拷贝到缓冲区；
  4. 如果 Channel 的缓冲区中包含数据，那么直接读取 recvx 索引对应的数据；
  5. 在默认情况下会挂起当前的 Goroutine，将 runtime.sudog 结构加入 recvq 队列并陷入休眠等待调度器的唤醒；
  1. func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
  2.     // 省略 debug 内容 …………
  4.     // 如果是一个 nil 的 channel
  5.     if c == nil {
  6.         // 如果不阻塞，直接返回 (false, false)
  7.         if !block {
  8.             return
  9.         }
  10.         // 否则，接收一个 nil 的 channel，goroutine 挂起
  11.         gopark(nil, nil, "chan receive (nil chan)", traceEvGoStop, 2)
  12.         // 不会执行到这里
  13.         throw("unreachable")
  14.     }
  16.     // 在非阻塞模式下，快速检测到失败，不用获取锁，快速返回
  17.     // 当我们观察到 channel 没准备好接收：
  18.     // 1. 非缓冲型，等待发送列队 sendq 里没有 goroutine 在等待
  19.     // 2. 缓冲型，但 buf 里没有元素
  20.     // 之后，又观察到 closed == 0，即 channel 未关闭。
  21.     // 因为 channel 不可能被重复打开，所以前一个观测的时候 channel 也是未关闭的，
  22.     // 因此在这种情况下可以直接宣布接收失败，返回 (false, false)
  23.     if !block && (c.dataqsiz == 0 && c.sendq.first == nil ||
  24.         c.dataqsiz > 0 && atomic.Loaduint(&c.qcount) == 0) &&
  25.         atomic.Load(&c.closed) == 0 {
  26.         return
  27.     }
  29.     var t0 int64
  30.     if blockprofilerate > 0 {
  31.         t0 = cputicks()
  32.     }
  34.     // 加锁
  35.     lock(&c.lock)
  37.     // channel 已关闭，并且循环数组 buf 里没有元素
  38.     // 这里可以处理非缓冲型关闭 和 缓冲型关闭但 buf 无元素的情况
  39.     // 也就是说即使是关闭状态，但在缓冲型的 channel，
  40.     // buf 里有元素的情况下还能接收到元素
  41.     if c.closed != 0 && c.qcount == 0 {
  42.         if raceenabled {
  43.             raceacquire(unsafe.Pointer(c))
  44.         }
  45.         // 解锁
  46.         unlock(&c.lock)
  47.         if ep != nil {
  48.             // 从一个已关闭的 channel 执行接收操作，且未忽略返回值
  49.             // 那么接收的值将是一个该类型的零值
  50.             // typedmemclr 根据类型清理相应地址的内存
  51.             typedmemclr(c.elemtype, ep)
  52.         }
  53.         // 从一个已关闭的 channel 接收，selected 会返回true
  54.         return true, false
  55.     }
  57.     // 等待发送队列里有 goroutine 存在，说明 buf 是满的
  58.     // 这有可能是：
  59.     // 1. 非缓冲型的 channel
  60.     // 2. 缓冲型的 channel，但 buf 满了
  61.     // 针对 1，直接进行内存拷贝（从 sender goroutine -> receiver goroutine）
  62.     // 针对 2，接收到循环数组头部的元素，并将发送者的元素放到循环数组尾部
  63.     if sg := c.sendq.dequeue(); sg != nil {
  64.         // Found a waiting sender. If buffer is size 0, receive value
  65.         // directly from sender. Otherwise, receive from head of queue
  66.         // and add sender's value to the tail of the queue (both map to
  67.         // the same buffer slot because the queue is full).
  68.         recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
  69.         return true, true
  70.     }
  72.     // 缓冲型，buf 里有元素，可以正常接收
  73.     if c.qcount > 0 {
  74.         // 直接从循环数组里找到要接收的元素
  75.         qp := chanbuf(c, c.recvx)
  77.         // …………
  79.         // 代码里，没有忽略要接收的值，不是 "<- ch"，而是 "val <- ch"，ep 指向 val
  80.         if ep != nil {
  81.             typedmemmove(c.elemtype, ep, qp)
  82.         }
  83.         // 清理掉循环数组里相应位置的值
  84.         typedmemclr(c.elemtype, qp)
  85.         // 接收游标向前移动
  86.         c.recvx++
  87.         // 接收游标归零
  88.         if c.recvx == c.dataqsiz {
  89.             c.recvx = 0
  90.         }
  91.         // buf 数组里的元素个数减 1
  92.         c.qcount--
  93.         // 解锁
  94.         unlock(&c.lock)
  95.         return true, true
  96.     }
  98.     if !block {
  99.         // 非阻塞接收，解锁。selected 返回 false，因为没有接收到值
  100.         unlock(&c.lock)
  101.         return false, false
  102.     }
  104.     // 接下来就是要被阻塞的情况了
  105.     // 构造一个 sudog
  106.     gp := getg()
  107.     mysg := acquireSudog()
  108.     mysg.releasetime = 0
  109.     if t0 != 0 {
  110.         mysg.releasetime = -1
  111.     }
  113.     // 待接收数据的地址保存下来
  114.     mysg.elem = ep
  115.     mysg.waitlink = nil
  116.     gp.waiting = mysg
  117.     mysg.g = gp
  118.     mysg.selectdone = nil
  119.     mysg.c = c
  120.     gp.param = nil
  121.     // 进入channel 的等待接收队列
  122.     c.recvq.enqueue(mysg)
  123.     // 将当前 goroutine 挂起
  124.     goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)
  126.     // 被唤醒了，接着从这里继续执行一些扫尾工作
  127.     if mysg != gp.waiting {
  128.         throw("G waiting list is corrupted")
  129.     }
  130.     gp.waiting = nil
  131.     if mysg.releasetime > 0 {
  132.         blockevent(mysg.releasetime-t0, 2)
  133.     }
  134.     closed := gp.param == nil
  135.     gp.param = nil
  136.     mysg.c = nil
  137.     releaseSudog(mysg)
  138.     return true, !closed
  139. }