本文源码版本基于 go1.21.13
Golang 中使用 CSP 中 channel 的概念。channel 是被单独创建并且可以在进程之间传递，它的通信模式类似于 boss-worker 模式，一个实体通过将消息发送到 channel 中，然后由监听这个 channel 的实体处理，两个实体之间是没有互相感知的，这个就实现实体之间的解耦。
type hchan struct { | |
qcount uint | |
dataqsiz uint | |
buf unsafe.Pointer | |
elemsize uint16 | |
closed uint32 | |
elemtype *_type | |
sendx uint | |
recvx uint | |
recvq waitq | |
sendq waitq | |
lock mutex | |
} | |
type waitq struct { | |
first *sudog | |
last *sudog | |
} |
如果当前 channel 中不存在缓冲区，那么就只会为 hchan 分配一段内存空间；
如果当前 channel 中存储的类型不是指针类型，就会直接为当前的 Channel 和底层的数组分配一块连续的内存空间；
在默认情况下会单独为 hchan 和 buff 分配内存；
func makechan(t *chantype, size int) *hchan { | |
elem := t.Elem | |
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: | |
c = (*hchan)(mallocgc(hchanSize, nil, true)) | |
c.buf = c.raceaddr() | |
case elem.PtrBytes == 0: | |
c = (*hchan)(mallocgc(hchanSize+mem, nil, true)) | |
c.buf = add(unsafe.Pointer(c), hchanSize) | |
default: | |
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 | |
} |
func (c *hchan) raceaddr() unsafe.Pointer { | |
return unsafe.Pointer(&c.buf) | |
} |
func empty(c *hchan) bool { | |
if c.dataqsiz == 0 { | |
return atomic.Loadp(unsafe.Pointer(&c.sendq.first)) == nil | |
} | |
return atomic.Loaduint(&c.qcount) == 0 | |
} |
type waitq struct { | |
first *sudog | |
last *sudog | |
} | |
func (q *waitq) enqueue(sgp *sudog) { | |
sgp.next = nil | |
x := q.last | |
if x == nil { | |
sgp.prev = nil | |
q.first = sgp | |
q.last = sgp | |
return | |
} | |
sgp.prev = x | |
x.next = sgp | |
q.last = sgp | |
} | |
func (q *waitq) dequeue() *sudog { | |
for { | |
sgp := q.first | |
if sgp == nil { | |
return nil | |
} | |
y := sgp.next | |
if y == nil { | |
q.first = nil | |
q.last = nil | |
} else { | |
y.prev = nil | |
q.first = y | |
sgp.next = nil | |
} | |
if sgp.isSelect && !sgp.g.selectDone.CompareAndSwap(0, 1) { | |
continue | |
} | |
return sgp | |
} | |
} |
当 channel 的 sendq 队列中包含处于等待状态的 goroutine 时，取出队列头的 goroutine，直接调用 send。
当有缓冲区时，并且缓冲区未满时，写入缓冲区。
当没有缓冲区时或者缓冲区已满时，阻塞接收。
向 已经 close 的 channel 写入数据会导致 panic
func chansend1(c *hchan, elem unsafe.Pointer) { | |
chansend(c, elem, true, getcallerpc()) | |
} | |
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool { | |
if c == nil { | |
if !block { | |
return false | |
} | |
gopark(nil, nil, waitReasonChanSendNilChan, traceBlockForever, 2) | |
throw("unreachable") | |
} | |
if debugChan { | |
print("chansend: chan=", c, "\n") | |
} | |
if raceenabled { | |
racereadpc(c.raceaddr(), callerpc, abi.FuncPCABIInternal(chansend)) | |
} | |
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 { | |
send(c, sg, ep, func() { unlock(&c.lock) }, 3) | |
return true | |
} | |
if c.qcount < c.dataqsiz { | |
qp := chanbuf(c, c.sendx) | |
if raceenabled { | |
racenotify(c, c.sendx, nil) | |
} | |
typedmemmove(c.elemtype, qp, ep) | |
c.sendx++ | |
if c.sendx == c.dataqsiz { | |
c.sendx = 0 | |
} | |
c.qcount++ | |
unlock(&c.lock) | |
return true | |
} | |
if !block { | |
unlock(&c.lock) | |
return false | |
} | |
gp := getg() | |
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(mysg) | |
gp.parkingOnChan.Store(true) | |
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceBlockChanSend, 2) | |
KeepAlive(ep) | |
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 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) | |
c.recvx++ | |
if c.recvx == c.dataqsiz { | |
c.recvx = 0 | |
} | |
c.sendx = c.recvx | |
} | |
} | |
if sg.elem != nil { | |
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() | |
} | |
goready(gp, skip+1) | |
} |
func sendDirect(t *_type, sg *sudog, src unsafe.Pointer) { | |
dst := sg.elem | |
typeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.Size_) | |
memmove(dst, src, t.Size_) | |
} |
当 channel 被关闭时，缓冲区中没数据，返回
当 channel 被关闭时，sendq 队列中有处于等待状态的 goroutine 取出队列头的 goroutine，直接调用 recv
当 channel 的缓冲区中有数据时，读取缓冲区数据
当 channel 没有被关闭，缓冲区没有数据，阻塞接收。
func chanrecv1(c *hchan, elem unsafe.Pointer) { | |
chanrecv(c, elem, true) | |
} | |
func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) { | |
_, received = chanrecv(c, elem, true) | |
return | |
} | |
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) { | |
if debugChan { | |
print("chanrecv: chan=", c, "\n") | |
} | |
if c == nil { | |
if !block { | |
return | |
} | |
gopark(nil, nil, waitReasonChanReceiveNilChan, traceBlockForever, 2) | |
throw("unreachable") | |
} | |
if !block && empty(c) { | |
if atomic.Load(&c.closed) == 0 { | |
return | |
} | |
if empty(c) { | |
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 | |
} | |
} else { | |
if sg := c.sendq.dequeue(); sg != nil { | |
recv(c, sg, ep, func() { unlock(&c.lock) }, 3) | |
return true, true | |
} | |
} | |
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) | |
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 | |
} | |
gp := getg() | |
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 | |
c.recvq.enqueue(mysg) | |
gp.parkingOnChan.Store(true) | |
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceBlockChanRecv, 2) | |
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 recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) { | |
if c.dataqsiz == 0 { | |
if raceenabled { | |
racesync(c, sg) | |
} | |
if ep != nil { | |
recvDirect(c.elemtype, sg, ep) | |
} | |
} else { | |
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) | |
} | |
typedmemmove(c.elemtype, qp, sg.elem) | |
c.recvx++ | |
if c.recvx == c.dataqsiz { | |
c.recvx = 0 | |
} | |
c.sendx = c.recvx | |
} | |
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) | |
} |
在函数执行的最后会为所有被阻塞的 Goroutine 调用 goready 函数重新对这些协程进行调度.
close nil channel 和 close 已经关闭的 channel 会导致 panic
func reflect_chanclose(c *hchan) { | |
closechan(c) | |
} | |
func closechan(c *hchan) { | |
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 | |
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) | |
} | |
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) | |
for !glist.empty() { | |
gp := glist.pop() | |
gp.schedlink = 0 | |
goready(gp, 3) | |
} | |
} |
https://github.com/LeoYang90/Golang-Internal-Notes/blob/master/Go Channel.md
http://legendtkl.com/2017/07/30/understanding-golang-channel/
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。