go-channel底层原理

type hchan struct {
	qcount   uint// 队列中所有数据总数
	dataqsiz uint// 循环队列大小
	buf      unsafe.Pointer       // 指向循环队列的指针
	elemsize uint16// 循环队列中元素的大小
	closed   uint32// chan是否关闭的标识
	elemtype *_type               // 循环队列中元素的类型
	sendx    uint// 已发送元素在循环队列中的位置
	recvx    uint// 已接收元素在循环队列中的位置
	recvq    waitq                // 等待接收的goroutine的等待队列
	sendq    waitq                // 等待发送的goroutine的等待队列
	lock mutex                    // 控制chan并发访问的互斥锁
}

type waitq struct {
	first *sudog
	last  *sudog
}

funcmakechan(t *chantype, size int) *hchan {
	elem := t.elem
// 检查数据项大小不能超过 64KB
if elem.size >= 1<<16 {
		throw("makechan: invalid channel element type")
	}
// 检查内存对齐是否正确
if hchanSize%maxAlign != 0 || elem.align > maxAlign {
		throw("makechan: bad alignment")
	}
// 缓冲区大小检查，判断是否溢出
	mem, overflow := math.MulUintptr(elem.size, uintptr(size))
if overflow || mem > maxAlloc-hchanSize || size < 0 {
panic(plainError("makechan: size out of range"))
	}
var c *hchan
switch {
case mem == 0:
// 队列或者元素大小为 zero 时，无须创建buf环形队列.
		c = (*hchan)(mallocgc(hchanSize, nil, true))
// 竞态检查，利用这个地址进行同步操作.
		c.buf = c.raceaddr()
case elem.ptrdata == 0:
// 元素不是指针，分配一块连续的内存给hchan数据结构和缓冲区buf
		c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
// 表示hchan后面在内存里紧跟着就是buf环形队列
		c.buf = add(unsafe.Pointer(c), hchanSize)
default:
// 元素包含指针，单独分配环形队列buf
		c = new(hchan)
		c.buf = mallocgc(mem, elem, true)
	}
// 设置元素个数、元素类型给创建的chan
	c.elemsize = uint16(elem.size)
	c.elemtype = elem
	c.dataqsiz = uint(size)
	lockInit(&c.lock, lockRankHchan)
if debugChan {
print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")
	}
return c
}

func chansend1(c *hchan, elem unsafe.Pointer) {
chansend(c, elem, true, getcallerpc())
}

funcchansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr)bool {
// 如果chan为nil
if c == nil {
// 对于select这种非阻塞的发送,直接返回
if !block {
returnfalse
		}
// 对于阻塞的通道，将 goroutine 挂起
		gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
		throw("unreachable")
	}
        ......
}

if !block && c.closed == 0 && full(c) {
returnfalse
}
.......
funcfull(c *hchan)bool {
// 如果循环队列大小为0
if c.dataqsiz == 0 {
// 假设指针读取是近似原子性的，这里用来判断没有接收者
return c.recvq.first == nil
	}
// 队列满了
return c.qcount == c.dataqsiz
}

// 对chan加锁
lock(&c.lock)
// 检查chan是否关闭
if c.closed !=0 {
    unlock(&c.lock)
    panic(plainError("send on closed channel"))
}
// 从 recvq 中取出一个接收者
if sg := c.recvq.dequeue(); sg !=nil {
// 如果接收者存在，直接向该接收者发送数据，绕过循环队列buf
    send(c, sg, ep, func() { unlock(&c.lock) }, 3)
returntrue
}

func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
	......
if sg.elem != nil {
// 直接把要发送的数据拷贝到receiver的内存地址
sendDirect(c.elemtype, sg, ep)
		sg.elem = nil
	}
	gp := sg.g
unlockf()
	gp.param = unsafe.Pointer(sg)
	sg.success = true
if sg.releasetime != 0 {
		sg.releasetime = cputicks()
	}
// 唤醒等待的接收者goroutine
goready(gp, skip+1)
}

// 如果缓冲区没有满，直接将要发送的数据复制到缓冲区
if c.qcount < c.dataqsiz {
// 找到要发送数据到循环队列buf的索引位置
    qp := chanbuf(c, c.sendx)
    ......
// 数据拷贝到循环队列中
typedmemmove(c.elemtype, qp, ep)
// 将待发送数据索引加1，由于是循环队列，如果到了末尾，从0开始
    c.sendx++
if c.sendx == c.dataqsiz {
        c.sendx = 0
    }
// chan中元素个数加1，释放锁返回true
    c.qcount++
unlock(&c.lock)
    return true
}

if !block {
unlock(&c.lock)
		return false
	}
// 下面的逻辑是将当前goroutine挂起
// 调用 getg()方法获取当前goroutine的指针，用于绑定给一个 sudog
	gp := getg()
// 调用 acquireSudog()方法获取一个 sudog，可能是新建的 sudog，也有可能是从缓存中获取的。设置好sudog要发送的数据和状态。比如发送的 Channel、是否在 select 中和待发送数据的内存地址等等。
	mysg := acquireSudog()
	mysg.releasetime = 0
if t0 != 0 {
		mysg.releasetime = -1
	}
	mysg.elem = ep
	mysg.waitlink = nil
	mysg.g = gp
	mysg.isSelect = false
	mysg.c = c
	gp.waiting = mysg
	gp.param = nil
// 调用 c.sendq.enqueue 方法将配置好的 sudog 加入待发送的等待队列
	c.sendq.enqueue(mysg)
	atomic.Store8(&gp.parkingOnChan, 1)
// 调用gopark方法挂起当前goroutine，状态为waitReasonChanSend，阻塞等待channel接收者的激活
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
// 最后，KeepAlive() 确保发送的值保持活动状态，直到接收者将其复制出来
KeepAlive(ep)

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
        ......
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
}

i <- ch
i, ok <- ch

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
         ......
} 

funcchanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
	......
// 如果c为空
if c == nil {
// 如果c为空且是非阻塞调用，那么直接返回 (false,false)
if !block {
return
		}
//阻塞当前的goroutine
		gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
		throw("unreachable")
	}
        ......
} 

 //非阻塞调用，通过empty()判断是无缓冲chan或者是chan中没有数据
	if !block && empty(c) {
		// 如果chan没有关闭，则直接返回 (false,false)
		if atomic.Load(&c.closed) ==0 {
			return
		}
                // 如果chan关闭, 为了防止检查期间的状态变化，二次调用empty()进行原子检查，如果是无缓冲chan或者是chan中没有数据，返回 (true,false)
		if empty(c) {
			if raceenabled {
				raceacquire(c.raceaddr())
			}
			if ep != nil {
				typedmemclr(c.elemtype, ep)
			}
			return true,false
		}
	}
func empty(c *hchan) bool {
	// c.dataqsiz 是不可变的
	if c.dataqsiz ==0 {
		return atomic.Loadp(unsafe.Pointer(&c.sendq.first)) == nil
	}
	return atomic.Loaduint(&c.qcount) ==0
}

......
// 对chan加锁
lock(&c.lock)
// 如果已经关闭，并且chan中没有数据，返回 (true,false)
if c.closed !=0&& c.qcount ==0 {
ifraceenabled {
			raceacquire(c.raceaddr())
		}
		unlock(&c.lock)
if ep !=nil {
			typedmemclr(c.elemtype, ep)
		}
returntrue, false
	}
......

 ......
// 从发送者队列获取等待发送的 goroutine	
if sg := c.sendq.dequeue(); sg != nil {
//在 channel 的发送队列中找到了等待发送的 goroutine，取出队头等待的 goroutine。如果缓冲区的大小为 0，则直接从发送方接收值。否则，对应缓冲区满的情况，从队列的头部接收数据，发送者的值添加到队列的末尾（此时队列已满，因此两者都映射到缓冲区中的同一个下标）
		recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
returntrue, true
	}
funcrecv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
if c.dataqsiz == 0 {
if raceenabled {
			racesync(c, sg)
		}
if ep != nil {
// 从发送者sender里面拷贝数据
			recvDirect(c.elemtype, sg, ep)
		}
	} else {
// 队列是满的
		qp := chanbuf(c, c.recvx)
if raceenabled {
			racenotify(c, c.recvx, nil)
			racenotify(c, c.recvx, sg)
		}
// 从缓冲区拷贝数据给接收者receiver
if ep != nil {
			typedmemmove(c.elemtype, ep, qp)
		}
// 从发送者sender拷贝数据到缓冲区
		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)
	sg.success = true
if sg.releasetime != 0 {
		sg.releasetime = cputicks()
	}
// 唤醒发送者
	goready(gp, skip+1)
}

// 如果缓冲区有数据
if c.qcount > 0 {
// 直接从缓冲区接收数据
	qp := chanbuf(c, c.recvx)
if raceenabled {
racenotify(c, c.recvx, nil)
	}
// 接收数据地址ep不为空，直接从缓冲区复制数据到ep
if ep != nil {
typedmemmove(c.elemtype, ep, qp)
	}
typedmemclr(c.elemtype, qp)
// 待接收索引加1
	c.recvx++
// 循环队列，如果到了末尾，从0开始
if c.recvx == c.dataqsiz {
		c.recvx = 0
	}
// 缓冲区数据减1
	c.qcount--
unlock(&c.lock)
	return true, true
}

// 如果是select非阻塞读取的情况，直接返回(false, false)
if !block {
unlock(&c.lock)
	return false, false
}
// 没有发送者，挂起当前goroutine
// 获取当前 goroutine 的指针，用于绑定给一个 sudog
gp := getg()
// 调用 acquireSudog() 方法获取一个 sudog，可能是新建的 sudog，也有可能是从缓存中获取的。设置好 sudog 要发送的数据和状态
mysg := acquireSudog()
mysg.releasetime = 0
if t0 != 0 {
	mysg.releasetime = -1
}
mysg.elem = ep
mysg.waitlink = nil
gp.waiting = mysg
mysg.g = gp
mysg.isSelect = false
mysg.c = c
gp.param = nil
// 将配置好的 sudog 加入待发送的等待队列
c.recvq.enqueue(mysg)
atomic.Store8(&gp.parkingOnChan, 1)
// 挂起当前 goroutine
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
	......
// 当前goroutine被唤醒，完成chan数据的接收，之后进行参数检查，解除chan绑定，并释放sudog
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 closechan(c *hchan) {
// 如果chan为空，此时关闭它会panic
if c == nil {
panic(plainError("close of nil channel"))
	}
// 加锁
lock(&c.lock)
// 如果chan已经关闭了，再次关闭它会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())
	}
// 设置chan的closed状态为关闭
	c.closed = 1
// 申明一个存放所有接收者和发送者goroutine的list
var glist gList
//获取recvq里的所有接收者
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中
		glist.push(gp)
	}
// 获取所有发送者
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中
		glist.push(gp)
	}
unlock(&c.lock)
// 唤醒所有的glist中的goroutine 
for !glist.empty() {
		gp := glist.pop()
		gp.schedlink = 0
goready(gp, 3)
	}
}