Golang分享（一）：channel底层原理

type hchan struct {
  qcount   uint           // 循环数组中的元素数量，长度
  dataqsiz uint           // 循环数组的大小，容量
  // channel分为无缓冲和有缓冲channel两种
  // 有缓冲的channel使用ring buffer（环形缓冲区）来缓存写入的数据，本质是循环数组
  // 为什么是循环数组？普通数组容量固定、更适合指定的空间，且弹出元素时，元素需要全部前移
  buf      unsafe.Pointer // 指向底层循环数组的指针（环形缓冲区）
  elemsize uint16         // 元素的大小
  closed   uint32         // 是否关闭的标志，0：未关闭，1：已关闭
  elemtype *_type // channel中的元素类型
  // 当下标超过数组容量后会回到第一个位置，所以需要有两个字段记录当前读和当前写的下标位置
  sendx    uint   // 下一次写的位置
  recvx    uint   // 下一次读的位置
  // 尝试读/写channel时被阻塞的goroutine
  recvq    waitq  // 读等待队列
  sendq    waitq  // 写等待队列

  // 互斥锁，保证读写channel时的并发安全问题
  lock mutex
}

type waitq struct{
    first *sudog
    last  *sudog
}

//src/runtime/chan.go
type sudog struct{
    g *g //记录哪个协程在等待
    
    
    next *sudog
    prev *sudog
    elem unsafe.Pointer // 等待发送/接收的数据在哪里
    
    ...
    
    c *chan //等待的是哪个channel
}

func makechan(t *chantype, size int) *hchan {
  // 获取元素类型
  elem := t.elem

  // compiler checks this but be safe.
  // 元素的大小必须小于64K
  // 编译器已经检查了这一点，但是为了安全起见再次进行检查
  if elem.size >= 1 maxAlign {
    throw("makechan: bad alignment")
  }

  // 计算所需要的内存大小
  mem, overflow := math.MulUintptr(elem.size, uintptr(size))
  // 检查是否溢出，所需内存是否超过maxAlloc - hchanSize 或者 size是否小于0
  if overflow || mem > maxAlloc-hchanSize || size 

// 加锁
lock(&c.lock)
......
// 从接收者队列recvq中取出一个接收者，接收者不为空的情况下，直接将数据传递给该接收者
if sg := c.recvq.dequeue(); sg != nil {
  // c: channel
  // sg：从recvq中取出来的接收者
  send(c, sg, ep, func() { unlock(&c.lock) }, 3)
  return true
}

func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
  // 无缓存通道和有缓存通道的处理逻辑
  if raceenabled {
    // 无缓冲通道的处理逻辑
    if c.dataqsiz == 0 {
      racesync(c, sg)
    } else {
      racenotify(c, c.recvx, nil)
      racenotify(c, c.recvx, sg)
      // 假装经过channel
      // 相当于循环列表的rear指针向前进1
      c.recvx++
      // 队列数组中最后一个元素已经读取，则再次从头开始读取数据
      if c.recvx == c.dataqsiz {
        c.recvx = 0
      }
      c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
    }
  }
  // 将ep直接复制到接收者sg中
  if sg.elem != nil {
    // 复制数据到sg中
    sendDirect(c.elemtype, sg, ep)
    sg.elem = nil
  }
  // 接收者对应的goroutine
  gp := sg.g
  unlockf()
  gp.param = unsafe.Pointer(sg)
  sg.success = true
  if sg.releasetime != 0 {
    sg.releasetime = cputicks()
  }
  // 使接收者goroutine变成runnable状态，唤醒goroutine
  goready(gp, skip+1)
}

// 将src值复制到dst中
func typedmemmove(typ *_type, dst, src unsafe.Pointer) {
  if dst == src {
    return
  }
    ...
  memmove(dst, src, typ.size)
    ...
}

// 缓冲队列中的元素个数小于队列的大小
// 说明缓冲队列中还有空间
if c.qcount 

// 获取当前的goroutine，用于绑定给一个sudog
  gp := getg()
  // 获取一个sudog
  mysg := acquireSudog()
  mysg.releasetime = 0
  if t0 != 0 {
    mysg.releasetime = -1
  }
  // 设置sudog发送的数据
  mysg.elem = ep
  mysg.waitlink = nil
  // 设置sudog绑定的goroutine
  mysg.g = gp
  mysg.isSelect = false
  // 设置sudog绑定的channel
  mysg.c = c
  gp.waiting = mysg
  gp.param = nil
  // 将发送者入队sendq
  c.sendq.enqueue(mysg)
  gp.parkingOnChan.Store(true)
  // 调用gopark方法挂起当前goroutine，状态为waitReasonChanSend，阻塞等待channel。
  gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
  // 最后，KeepAlive()确保发送的值保持活跃状态，直到接收者将其复制出来
  KeepAlive(ep)

  // someone woke us up.
  if mysg != gp.waiting {
    throw("G waiting list is corrupted")
  }
  gp.waiting = nil
  gp.activeStackChans = false
  closed := !mysg.success
  gp.param = nil
  if mysg.releasetime > 0 {
    blockevent(mysg.releasetime-t0, 2)
  }
  mysg.c = nil
  releaseSudog(mysg)
  if closed {
    if c.closed == 0 {
      throw("chansend: spurious wakeup")
    }
    panic(plainError("send on closed channel"))
  }
  return true

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, abi.FuncPCABIInternal(chansend))
   }

   // 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 full()).
   // 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. However, nothing here
   // guarantees forward progress. We rely on the side effects of lock release in
   // chanrecv() and closechan() to update this thread's view of c.closed and full().
   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  0 {
      blockevent(mysg.releasetime-t0, 2)
   }
   mysg.c = nil
   releaseSudog(mysg)
   if closed {
      if c.closed == 0 {
         throw("chansend: spurious wakeup")
      }
      panic(plainError("send on closed channel"))
   }
   return true
}

// 如果发送队列中存在发送者，返回(true,true)
lock(&c.lock)

if sg := c.sendq.dequeue(); sg != nil {
  // 找到一个等待的发送者，如果缓冲区大小为0，则直接接收发送者的值
  // 否则，从队列头部接收，并将发送者的值添加到队列的尾部
  recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
  return true, true
}

func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
  // 创建的channel是无缓存buf
  if c.dataqsiz == 0 {
    if raceenabled {
      racesync(c, sg)
    }
    // 直接将数据从发送者复制过去，即直接接收发送者的数据
    if ep != nil {
      // copy data from sender
      recvDirect(c.elemtype, sg, ep)
    }
  } else {
    // 返回buf中待被接收的数据的指针
    qp := chanbuf(c, c.recvx)
    if raceenabled {
      racenotify(c, c.recvx, nil)
      racenotify(c, c.recvx, sg)
    }
    // 复制队列中的数据给接收者
    if ep != nil {
      typedmemmove(c.elemtype, ep, qp)
    }
    // 将发送者的数据加入到buf中
    typedmemmove(c.elemtype, qp, sg.elem)
    // 更新recvx
    c.recvx++
    if c.recvx == c.dataqsiz {
      c.recvx = 0
    }
    // 更新sendx
    c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
  }
  sg.elem = nil
  gp := sg.g
  // 释放锁
  unlockf()
  gp.param = unsafe.Pointer(sg)
  sg.success = true
  if sg.releasetime != 0 {
    sg.releasetime = cputicks()
  }
  // 唤醒发送者，下一轮参与调度
  goready(gp, skip+1)
}

//返回buffer中下标为i的指针
func chanbuf(c *hchan, i uint) unsafe.Pointer {
  return add(c.buf, uintptr(i)*uintptr(c.elemsize))
}

if c.qcount > 0 {
    // 待接收数据的指针
    qp := chanbuf(c, c.recvx)
    if raceenabled {
      racenotify(c, c.recvx, nil)
    }
    // 直接从队列中接收数据
    if ep != nil {
      typedmemmove(c.elemtype, ep, qp)
    }
    typedmemclr(c.elemtype, qp)
    // 更新recvx
    c.recvx++
    if c.recvx == c.dataqsiz {
      c.recvx = 0
    }
    // 更新qcount
    c.qcount--
    // 释放锁
    unlock(&c.lock)
    // 返回(true,true)
    return true, true
  }

// 获取当前的goroutine，用于绑定给一个sudog
  gp := getg()
  // 返回一个sudog
  mysg := acquireSudog()
  mysg.releasetime = 0
  if t0 != 0 {
    mysg.releasetime = -1
  }
  // sudog绑定接收的数据
  mysg.elem = ep
  mysg.waitlink = nil
  gp.waiting = mysg
  // sudog绑定当前的goroutine
  mysg.g = gp
  mysg.isSelect = false
  // sudog绑定当前的channel
  mysg.c = c
  gp.param = nil
  // 将接收者入队recvq中
  c.recvq.enqueue(mysg)
  gp.parkingOnChan.Store(true)
  // 挂起当前的goroutine
  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)
  }
  success := mysg.success
  gp.param = nil
  mysg.c = nil
  releaseSudog(mysg)
  // 进行接收
  return true, success

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) {
    // After observing that the channel is not ready for receiving, we observe whether the
    // channel is closed.
    //
    // Reordering of these checks could lead to incorrect behavior when racing with a close.
    // For example, if the channel was open and not empty, was closed, and then drained,
    // reordered reads could incorrectly indicate "open and empty". To prevent reordering,
    // we use atomic loads for both checks, and rely on emptying and closing to happen in
    // separate critical sections under the same lock.  This assumption fails when closing
    // an unbuffered channel with a blocked send, but that is an error condition anyway.
    if atomic.Load(&c.closed) == 0 {
      // Because a channel cannot be reopened, the later observation of the channel
      // being not closed implies that it was also not closed at the moment of the
      // first observation. We behave as if we observed the channel at that moment
      // and report that the receive cannot proceed.
      return
    }
    // The channel is irreversibly closed. Re-check whether the channel has any pending data
    // to receive, which could have arrived between the empty and closed checks above.
    // Sequential consistency is also required here, when racing with such a send.
    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 {
    if c.qcount == 0 {
      if raceenabled {
        raceacquire(c.raceaddr())
      }
      unlock(&c.lock)
      if ep != nil {
        typedmemclr(c.elemtype, ep)
      }
      return true, false
    }
    // The channel has been closed, but the channel's buffer have data.
  } else {
    // Just found waiting sender with not closed.
    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
    }
  }

  if c.qcount > 0 {
    // Receive directly from queue
    qp := chanbuf(c, c.recvx)
    if raceenabled {
      racenotify(c, c.recvx, nil)
    }
    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)
  // 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.
  gp.parkingOnChan.Store(true)
  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)
  }
  success := mysg.success
  gp.param = nil
  mysg.c = nil
  releaseSudog(mysg)
  return true, success
}

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, abi.FuncPCABIInternal(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 = unsafe.Pointer(sg)
    sg.success = false
    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 = unsafe.Pointer(sg)
    sg.success = false
    if raceenabled {
      raceacquireg(gp, c.raceaddr())
    }
    glist.push(gp)
  }
  unlock(&c.lock)

func closechan(c *hchan) {
  if c == nil {
    panic(plainError("close of nil channel"))
  }
  
  // 获取锁
  lock(&c.lock)
  // 如果通道已关闭，则panic
  if c.closed != 0 {
    unlock(&c.lock)
    panic(plainError("close of closed channel"))
  }

  if raceenabled {
    callerpc := getcallerpc()
    racewritepc(c.raceaddr(), callerpc, abi.FuncPCABIInternal(closechan))
    racerelease(c.raceaddr())
  }
  // 关闭channel
  c.closed = 1

  // 创建待唤醒列表
  var glist gList

  // 释放所有的接收者
  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 {
      // 设置释放时间为当前的CPU时间
      sg.releasetime = cputicks()
    }
    gp := sg.g
    gp.param = unsafe.Pointer(sg)
    sg.success = false
    if raceenabled {
      raceacquireg(gp, c.raceaddr())
    }
    // 将接受者的goroutine添加到待唤醒列表中
    glist.push(gp)
  }

  // 释放所有的发送者（这些发送者将会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 = unsafe.Pointer(sg)
    sg.success = false
    if raceenabled {
      raceacquireg(gp, c.raceaddr())
    }
    glist.push(gp)
  }
  unlock(&c.lock)

  // 在释放通道锁之后，唤醒所有的goroutine
  for !glist.empty() {
    gp := glist.pop()
    gp.schedlink = 0
    goready(gp, 3)
  }
}

func main() {
  runtime.GC()
  stats := runtime.MemStats{}
  runtime.ReadMemStats(&stats)
  println(stats.HeapInuse)

  run()

  runtime.GC()
  runtime.ReadMemStats(&stats)
  println(stats.HeapInuse)
}

func run() {
  c := make(chan int, 10)
  for i := 0; i