go channel源码阅读
channel 介绍
channel是一个类型管道，通过它可以在groutine之间发送消息
核心数据结构
- channel内部数据结构是固定长度的双向循环列表
- 按顺序往里面写数据，写满之后又从0开始写
- chan中的两个重要组件是
buf和waitq，所有的行为和实现都是围绕着两个组件进行的
type hchan struct {
qcount uint // queue 里面有效用户元素，这个字段是在元素出对，入队改变的；
dataqsiz uint // 初始化的时候赋值，之后不再改变，指明数组 buffer 的大小；
buf unsafe.Pointer // 指明 buffer 数组的地址，初始化赋值，之后不会再改变；
elemsize uint16 // 指明元素的大小，和 dataqsiz 配合使用就能知道 buffer 内存块的大小了；
closed uint32
elemtype *_type // 元素类型，初始化赋值；
sendx uint // send index
recvx uint // receive index
recvq waitq // 等待 recv 响应的对象列表，抽象成 waiters
sendq waitq // 等待 sedn 响应的对象列表，抽象成 waiters
// 互斥资源的保护锁，官方特意说明，在持有本互斥锁的时候，绝对不要修改 Goroutine 的状态，不能很有可能在栈扩缩容的时候，出现死锁
lock mutex
}
// 等待读写的队列数据结构，保证先进先出
type waitq struct{
first *sudog
last *sudog
}
创建channel
// 对应的源码为 c := make(chan int, size)
// c := make(chan int) 这种情况下，size = 0
func makechan(t *chantype, size int) *hchan {
elem := t.elem
// 总共需要的buff大小 = channel中创建的这种元素类型的大小（elem.size）* size
mem, overflow := math.MulUintptr(elem.size, uintptr(size))
var c *hchan
// 下面是为buf创建并分配存储空间
switch {
case mem == 0:
// size为0，或者每个元素占用的大小为0
// 这时为buf分配大小时，只需要分配hchan结构体本身占用的大小即可
// hchanSize是一个常量，表示空的hchan需要占用的字节大小
// hchanSize = unsafe.Sizeof(hchan{}) + uintptr(-int(unsafe.Sizeof(hchan{}))&(maxAlign-1))
c = (*hchan)(mallocgc(hchanSize, nil, true))
// raceaddr内部实现为：return unsafe.Pointer(&c.buf)
c.buf = c.raceaddr()
case elem.ptrdata == 0:
// 如果队列中不存在指针，那么每个元素都需要被存储并占用空间，占用大小为前面乘法算出来的mem
// 同时还要加上hchan本身占用的空间大小，加起来就是整个hchan占用的空间大小
c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
// 把buf指针指向空的hchan占用空间大小的末尾
c.buf = add(unsafe.Pointer(c), hchanSize)
default:
// Elements contain pointers.
// 如果chan中的元素是指针类型的数据，为buf单独开辟mem大小的空间，用来保存所有的数据
c = new(hchan)
c.buf = mallocgc(mem, elem, true)
}
// 设置chan的总大小
c.elemsize = uint16(elem.size)
// 元素类型
c.elemtype = elem
// 环形队列的大小，即用户创建时设置的大小
c.dataqsiz = uint(size)
return c
}
发送数据到channel
发送数据到channel时，直观的理解是将数据放到chan的环形队列中，不过go做了一些优化：先判断是否有等待接收数据的groutine，如果有，直接将数据发给Groutine，唤醒groutine，就不放入队列中了。当然还有另外一种情况就是：队列如果满了，那就只能放到队列中等待，直到有数据被取走才能发送。
- 如果recq为空，将数据放入buf中
- 如果不为空 从recvq中取出一个等待接受数据的Groutine 将数据直接拷贝给Groutine
- 如果buf满 则将要发送的数据和当前的Groutine打包成Sudog对象放入sendq，并将groutine置为等待状态
发送数据源码
// ep指向要发送数据的首地址
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
// 先上锁
lock(&c.lock)
// 如果channel已经关闭，抛出错误
// 下面这个错误经常会遇到，都是对channel使用不当报出来的
if c.closed != 0 {
unlock(&c.lock)
panic(plainError("send on closed channel"))
}
// 从接收队列中取出元素，如果取到数据，就将数据传过去
if sg := c.recvq.dequeue(); sg != nil {
// 调用send方法，将值传过去
send(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true
}
// 走到这里，说明没有等待接收数据的Groutine
// 如果缓冲区没有满，直接将要发送的数据复制到缓冲区
if c.qcount < c.dataqsiz {
// c.sendx是已发送的索引位置，这个方法通过指针偏移找到索引位置
// 相当于执行c.buf(c.sendx)
qp := chanbuf(c, c.sendx)
if raceenabled {
raceacquire(qp)
racerelease(qp)
}
// 复制数据，内部调用了memmove，是用汇编实现的
// 通知接收方数据给你了，将接收方协程由等待状态改成可运行状态，
// 将当前协程加入协程队列，等待被调度
typedmemmove(c.elemtype, qp, ep)
// 数据索引前移，如果到了末尾，又从0开始
c.sendx++
if c.sendx == c.dataqsiz {
c.sendx = 0
}
// 元素个数加1，释放锁并返回
c.qcount++
unlock(&c.lock)
return true
}
// 走到这里，说明缓冲区也写满了
// 同步非阻塞的情况，直接返回
if !block {
unlock(&c.lock)
return false
}
// 以下为同步阻塞的情况
// 此时会将当前的Groutine以及要发送的数据放入到sendq队列中，并且切换出该Groutine
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
// 将Groutine放入sendq队列
c.sendq.enqueue(mysg)
// Groutine转入 waiting 状态，gopark是调度相关的代码
// 在用户看来，向channel发送数据的代码语句会阻塞
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
KeepAlive(ep)
// G被唤醒
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
// G被唤醒，状态改成可执行状态，从这里开始继续执行
releaseSudog(mysg)
return true
}
send函数
// 要发送的数据ep，被拷贝到接收者sg中，之后sg被唤醒继续执行
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
// 拷贝数据
if sg.elem != nil {
sendDirect(c.elemtype, sg, ep)
sg.elem = nil
}
gp := sg.g
unlockf()
gp.param = unsafe.Pointer(sg)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
// 放入调度队列，等待被调度
goready(gp, skip+1)
}
读取数据
从channel读取数据时，先判断是否有等待发送数据的groutine 如果有直接读取groutine的数据。如果没有，再从环形队列取数据
- 如果有等待发送数据的groutine，从sendq中取出一个等待发送数据的Groutine，取出数据。
- 如果没有等待的groutine，且环形队列中有数据，从队列中取出数据
- 如果没有等待的groutine，且环形队列中也没有数据，则阻塞该Groutine，并将groutine打包为sudogo加入到recevq等待队列中
读取数据源码
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
// 上锁
lock(&c.lock)
// 优先从发送队列中取数据，如果有等待发送数据的groutine,直接从发送数据的协程中取出数据
if sg := c.sendq.dequeue(); sg != nil {
recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true, true
}
// chan环形队列中如果有有数据
if c.qcount > 0 {
// 从接收数据的索引出取出数据
// 等价于 c.buf[c.recvx]
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++
// 环形队列，如果到了末尾，再从0开始
if c.recvx == c.dataqsiz {
c.recvx = 0
}
// 发送数据的索引移动位置
c.qcount--
unlock(&c.lock)
return true, true
}
// 同步非阻塞，协程直接返回
if !block {
unlock(&c.lock)
return false, false
}
// 同步阻塞
// 如果代码走到这，说明没有任何数据可以获取到，阻塞住协程，并加入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)
// 调度
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
// G被唤醒，从这里继续执行
releaseSudog(mysg)
return true, !closed
}
关闭channel
- 设置关闭状态
- 唤醒所有等待读取channel的协程
- 所有等待写入channel的协程 抛出异常
func closechan(c *hchan) {
// channel为空，抛出异常
if c == nil {
panic(plainError("close of nil channel"))
}
// 上锁
lock(&c.lock)
// 如果channel已经被关闭，抛出异常
if c.closed != 0 {
unlock(&c.lock)
panic(plainError("close of closed channel"))
}
// 设置关闭状态的值
c.closed = 1
// 申明一个存放g的list，把所有的groutine放进来
// 目的是尽快释放锁，因为队列中可能还有数据需要处理，可能用到锁
var glist gList
// release all readers
// 唤醒所有等待读取chanel数据的协程
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.
// 处理临时队列中所有的groutine
for !glist.empty() {
gp := glist.pop()
gp.schedlink = 0
// 放入调度队列，等待被调度
goready(gp, 3)
}
}