type hchan struct {
qcount uint // 队列中得所有数据数
dataqsiz uint // 环形队列的大小
buf unsafe.Pointer // 指向大小为 dataqsiz 的数组 。ring buf 数据结构，循环队列
elemsize uint16 // 元素大小
closed uint32 // 是否关闭
elemtype *_type // 元素类型
sendx uint // 发送索引
recvx uint // 接受索引
recvq waitq // recv 等待队列，即（<-chan）
sendq waitq // send 等待队列，即（ch<-）
lock mutex // lock 保护了 hchan 的所有字段，以及在此 channel 上阻塞的 sudog 的一些字段。当持有此锁时不改变其他 goroutine 的状态（不ready 的 goroutine），因为他会在栈收缩时发送死锁
type sudog struct {
g *g // 由 sudog 阻塞的通道的hchan.lock 进行保护
isSelect bool // 表示g正在参与 一个select，因此 g.selectDone 必须已 CAS 的方式避免唤醒时候的 data race
next *sudog // 后指针
prev *sudog // 前指针
elem unsafe.Pointer // 数据元素，（可能指向栈）
// 下面的字段永远不会被并发访问，只有 waitlink 会被 g 访问
// 对于 semaphores，所有的字段（包括上面的）只会在持有 semaRoot 锁时被访问
acquiretime int64
releasetime int64
ticket uint32
isSelect bool
parent *sudog // semaRoot 二叉树
waitlink *sudog // g.waiting 列表 或 semaRoot
waittail *sudog // semaRoot
c *hchan // channel
}
ch := make(chan int,5)
ch := make(chan int)
make（chan interface{}，4）
func makechan64(t *chantype, size int64) *hchan {
if int64(int(size)) != size {
panic(plainError("makechan: size out of range"))
}
return makechan(t, int(size))
}
申请内存时控制传入的缓冲大小
func makechan(t *chantype, size int) *hchan {
elem := t.elem
// compiler checks this but be safe.
// 编译器相关的检查
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"))
}
// Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.
// buf points into the same allocation, elemtype is persistent.
// SudoG's are referenced from their owning thread so they can't be collected.
// TODO(dvyukov,rlh): Rethink when collector can move allocated objects.
var c *hchan
switch {
case mem == 0:
// Queue or element size is zero.
// 通过 mallocgc 申请内存，mallocgc 是 go的运行时内存分配器
// 无缓冲的channel
c = (*hchan)(mallocgc(hchanSize, nil, true))
// Race detector uses this location for synchronization.
c.buf = c.raceaddr()
case elem.ptrdata == 0:
// Elements do not contain pointers.
// Allocate hchan and buf in one call.
// hchanSize chan的大小加上 ring buf 的大小（mem）。连续内存
// 元素不包含指针。指针这块是关于gc的处理
c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
c.buf = add(unsafe.Pointer(c), hchanSize)
default:
// Elements contain pointers.
// 元素包含指针
c = new(hchan)
c.buf = mallocgc(mem, elem, true)
}
// 初始化一些字段
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
}
ch <-1
<-chan
func chansend1(c *hchan, elem unsafe.Pointer) {
chansend(c, elem, true, getcallerpc())
}
本质还是调用了 chansend
// callerpc 运行时调试用的，不需太关注
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
// fast path
if c == nil {
if !block {
return false
}
// 向一个已经关闭的channel 发送数据，会死锁
gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
throw("unreachable")
}
if debugChan {
print("chansend: chan=", c, "\n")
}
if raceenabled {
racereadpc(c.raceaddr(), callerpc, funcPC(chansend))
}
if !block && c.closed == 0 && full(c) {
return false
}
var t0 int64
if blockprofilerate > 0 {
t0 = cputicks()
}
// 持有锁
lock(&c.lock)
// 获取锁之后需要再检查是否被 close
if c.closed != 0 {
unlock(&c.lock)
panic(plainError("send on closed channel"))
}
// 判断是否有正在 recvq 中 阻塞的接收协程，有的话直接进行数据的拷贝
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
}
// 如果buf 中有剩余空间，直接将数据写入到 buf中
if c.qcount < c.dataqsiz {
// Space is available in the channel buffer. Enqueue the element to send.
// 入队操作
qp := chanbuf(c, c.sendx)
if raceenabled {
raceacquire(qp)
racerelease(qp)
}
typedmemmove(c.elemtype, qp, ep)
// 修改 sendx
c.sendx++
if c.sendx == c.dataqsiz {
c.sendx = 0
}
c.qcount++
// 释放锁
unlock(&c.lock)
return true
}
if !block {
unlock(&c.lock)
return false
}
// Block on the channel. Some receiver will complete our operation for us.
// 阻塞在 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
mysg.g = gp
mysg.isSelect = false
mysg.c = c
gp.waiting = mysg
gp.param = nil
// 放入到 sendq 中
c.sendq.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.
atomic.Store8(&gp.parkingOnChan, 1)
// park 结束之后 解锁，增加了回调
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
// Ensure the value being sent is kept alive until the
// receiver copies it out. The sudog has a pointer to the
// stack object, but sudogs aren't considered as roots of the
// stack tracer.
// 防止gc 清除要使用的数据
KeepAlive(ep)
// someone woke us up.
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
releaseSudog(mysg)
return true
}
这种情况下的接收数据没有 buf 的参与，和无缓存的channel 类似
func chanrecv1(c *hchan, elem unsafe.Pointer) {
chanrecv(c, elem, true)
}
//go:nosplit
func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {
_, received = chanrecv(c, elem, true)
return
}
一个处理了返回值，本质还是调用了通用的 chanrecv
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) {
if atomic.Load(&c.closed) == 0 {
return
}
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 && c.qcount == 0 {
if raceenabled {
raceacquire(c.raceaddr())
}
unlock(&c.lock)
if ep != nil {
typedmemclr(c.elemtype, ep)
}
return true, false
}
// 看看有没有发送方阻塞在这上面。如果有的话，直接向这个goroutine 拿数据即可，不需要经过 ring buf
if sg := c.sendq.dequeue(); sg != nil {
recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true, true
}
// 如果buf 中海油数据，直接从buf 中拿
if c.qcount > 0 {
// Receive directly from queue
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++
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.
// 如果接收的时候buf 为空，且没人发数据，就入队 recvq，阻塞等待被唤醒
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)
atomic.Store8(&gp.parkingOnChan, 1)
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
releaseSudog(mysg)
return true, !closed
}
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"))
}
if raceenabled {
callerpc := getcallerpc()
racewritepc(c.raceaddr(), callerpc, funcPC(closechan))
racerelease(c.raceaddr())
}
// 关闭 channel
c.closed = 1
var glist gList
// 将 sendq 和 recvq 中所有的g 都临时存放在 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 = 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)
}
// 将所有的G都存放在gList 中时，就可释放锁
// 拿着锁去read 比较费性能
unlock(&c.lock)
// Ready all Gs now that we've dropped the channel lock.
// 依次将 所有的g 就绪
for !glist.empty() {
gp := glist.pop()
gp.schedlink = 0
goready(gp, 3)
}
}