社区微信群开通啦,扫一扫抢先加入社区官方微信群
社区微信群
channel是golang中标志性的概念之一,很好很强大!
channel(通道),顾名思义,是一种通道,一种用于并发环境中数据传递的通道。通常结合golang中另一重要概念goroutine(go协程)使用,使得在golang中的并发编程变得清晰简洁同时又高效强大。
今天尝试着读读golang对channel的实现源码,拿起我生锈的水果刀,装模作样的解剖解剖这只大白老鼠。
type hchan struct {
qcount uint // total data in the queue
dataqsiz uint // size of the circular queue
buf unsafe.Pointer // points to an array of dataqsiz elements
elemsize uint16
closed uint32
elemtype *_type // element type
sendx uint // send index
recvx: uint // receive index
recvq waitq // list of recv waiters
sendq waitq // list of send waiters
// lock protects all fields in hchan, as well as several
// fields in sudogs blocked on this channel.
//
// Do not change another G's status while holding this lock
// (in particular, do not ready a G), as this can deadlock
// with stack shrinking.
lock mutex
}
hchan
结构就是channel的底层数据结构,看源码定义,可以说是非常清晰了。
其中关于recvq
和sendq
的两个列表所用的结构waitq
简单看下。
type waitq struct {
first *sudog
last *sudog
}
type sudog struct {
g *g
selectdone *uint32 // CAS to 1 to win select race (may point to stack)
next *sudog
prev *sudog
elem unsafe.Pointer // data element (may point to stack)
...
c *hchan // channel
}
可以看出waiq
是一个双向链表结构,链上的节点是sudog
。从sudog
的结构定义可以粗略看出,sudog
是对g
(即协程)的一个封装。用于记录一个等待在某个channel
上的协程g
、等待的元素elem
等信息。
func makechan(t *chantype, size int64) *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")
}
if size < 0 || int64(uintptr(size)) != size || (elem.size > 0 && uintptr(size) > (_MaxMem-hchanSize)/elem.size) {
panic(plainError("makechan: size out of range"))
}
var c *hchan
if elem.kind&kindNoPointers != 0 || size == 0 {
// Allocate memory in one call.
// Hchan does not contain pointers interesting for GC in this case:
// 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.
c = (*hchan)(mallocgc(hchanSize+uintptr(size)*elem.size, nil, true))
if size > 0 && elem.size != 0 {
c.buf = add(unsafe.Pointer(c), hchanSize)
} else {
// race detector uses this location for synchronization
// Also prevents us from pointing beyond the allocation (see issue 9401).
c.buf = unsafe.Pointer(c)
}
} else {
c = new(hchan)
c.buf = newarray(elem, int(size))
}
c.elemsize = uint16(elem.size)
c.elemtype = elem
c.dataqsiz = uint(size)
if debugChan {
print("makechan: chan=", c, "; elemsize=", elem.size, "; elemalg=", elem.alg, "; dataqsiz=", size, "n")
}
return c
}
第一部分的3个if
是对初始化参数的合法性检查。
_MaxMem
是可分配的堆大小。第二部分是具体的内存分配。
kindNoPointers
的时候,既非指针类型,则直接分配(hchanSize+uintptr(size)*elem.size)大小的连续空间。c.buf
指向hchan后面的elem队列首地址。c.buf
实际上是没有给他分配空间的kindNoPointers
,则channel的空间和buf的空间是分别分配的(这样做的原因待研究)// entry point for c <- x from compiled code
//go:nosplit
func chansend1(c *hchan, elem unsafe.Pointer) {
chansend(c, elem, true, getcallerpc(unsafe.Pointer(&c)))
}
channel发送,即协程向channel中发送数据,与此操作对应的go代码如c <- x
。
channel发送的实现源码中,通过chansend1()
,调用chansend()
,其中block
参数为true
。
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
if c == nil {
if !block {
return false
}
gopark(nil, nil, "chan send (nil chan)", traceEvGoStop, 2)
throw("unreachable")
}
...
}
chansend()
首先对c
进行判断, if c == nil
:即channel没有被初始化,这个时候会直接调用gopark
使得当前协程进入等待状态。而且用于唤醒的参数unlockf
传的nil
,即没有人来唤醒它,这样系统进入死锁。所以channel
必须被初始化之后才能使用,否则死锁。
接下来是正式的发送处理,且后续操作会加锁。
lock(&c.lock)
if c.closed != 0 {
unlock(&c.lock)
panic(plainError("send on closed channel"))
}
如果channel已经是closed
状态,解锁然后直接panic
。也就是说我们不可以向已经关闭的通道内在发送数据。
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
}
尝试从接收等待协程队列中取出一个协程,如果有则直接数据发给它。也就是说发送到channel的数据会优先检查接收等待队列,如果有协程等待取数,就直接给它。发完解锁,操作完成。
这里send()
方法会将数据写到从队列里取出来的sg
中,通过goready()
唤醒sg.g
(即等待的协程),进行后续处理。
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)
c.sendx++
if c.sendx == c.dataqsiz {
c.sendx = 0
}
c.qcount++
unlock(&c.lock)
return true
}
如果没有接收协程在等待,则去检查channel的缓存队列是否还有空位。如果有空位,则将数据放到缓存队列中。
通过c.sendx
游标找到队列中的空余位置,然后将数据存进去。移动游标,更新数据,然后解锁,操作完成。
if c.sendx == c.dataqsiz {
c.sendx = 0
}
通过这一段游标的处理可以看出,缓存队列是一个环形。
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.selectdone = nil
mysg.c = c
gp.waiting = mysg
gp.param = nil
c.sendq.enqueue(mysg)
goparkunlock(&c.lock, "chan send", traceEvGoBlockSend, 3)
如果缓存也慢了,这时候就只能阻塞住发送协程了, 等有合适的机会了,再将数据发送出去。getg()
获取当前协程对象g
的指针,acquireSudog()
生成一个sudog
,然后将当前协程及相关数据封装好链接到sendq
列表中。然年通过goparkunlock()
将其转为等待状态,并解锁。操作完成。
// entry points for <- c from compiled code
//go:nosplit
func chanrecv1(c *hchan, elem unsafe.Pointer) {
chanrecv(c, elem, true)
}
channel接收,即协程从channel中接收数据,与此操作对应的go代码如<- c
。
channel接收的实现源码中,通过chanrecv1()
,调用chanrecv()
,其中block
参数为true
。
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
...
if c == nil {
if !block {
return
}
gopark(nil, nil, "chan receive (nil chan)", traceEvGoStop, 2)
throw("unreachable")
}
...
}
同发送一样,接收也会首先检查c是否为nil,如果为nil,会调用gopark()
休眠当前协程,从而最终造成死锁。
接收操作同样先进行加锁,然后开始正式操作。
if c.closed != 0 && c.qcount == 0 {
if raceenabled {
raceacquire(unsafe.Pointer(c))
}
unlock(&c.lock)
if ep != nil {
typedmemclr(c.elemtype, ep)
}
return true, false
}
接收和发送略有不同,当channel关闭并且channel的缓存队列里没有数据了,那么接收动作会直接结束,但不会报错。
也就是说,允许从已关闭的channel中接收数据。
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
}
尝试从发送等待协程列表中取出一个等待协程,如果存在,则调用recv()
方法接收数据。
这里的recv()
方法比send()
方法稍微复杂一点,我们简单分析下。
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
if c.dataqsiz == 0 {
...
if ep != nil {
// copy data from sender
recvDirect(c.elemtype, sg, ep)
}
} else {
qp := chanbuf(c, c.recvx)
...
// copy data from queue to receiver
if ep != nil {
typedmemmove(c.elemtype, ep, qp)
}
// copy data from sender to queue
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)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(gp, skip+1)
}
recv()
的接收动作分为两种情况:
根据缓存的接收游标,从缓存队列中取出一个,拷贝给接受者
将发送协程中的数据,放到空出来的缓存位置中,游标下移。(即将新数据接到队列尾巴上)
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.selectdone = nil
mysg.c = c
gp.param = nil
c.recvq.enqueue(mysg)
goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)
没有协程等待发送,缓存中也没有数据了,那么之后阻塞接收协程,等待合适时机在接收数据。
同发送过程一样,将当前协程封装到sudog
中,链接到recvq
列表中。并休眠当前协程。
hchanSize
的计算maxAlign
参数的作用附注1:源码基于go1.9.2
附注2:文章中引用的源码...
处表示有删减
如果觉得我的文章对您有用,请随意打赏。你的支持将鼓励我继续创作!