您好,登录后才能下订单哦!
这篇文章主要介绍了Golang通道channel使用源码分析的相关知识,内容详细易懂,操作简单快捷,具有一定借鉴价值,相信大家阅读完这篇Golang通道channel使用源码分析文章都会有所收获,下面我们一起来看看吧。
channel是golang中标志性的概念之一,很好很强大!
channel(通道),顾名思义,是一种通道,一种用于并发环境中数据传递的通道。通常结合golang中另一重要概念goroutine(go协程)使用,使得在golang中的并发编程变得清晰简洁同时又高效强大。
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的底层数据结构,看源码定义,可以说是非常清晰了。
qcount:channel缓存队列中已有的元素数量
dataqsiz:channel的缓存队列大小(定义channel时指定的缓存大小,这里channel用的是一个环形队列)
buf:指向channel缓存队列的指针
elemsize:通过channel传递的元素大小
closed:channel是否关闭的标志
elemtype:通过channel传递的元素类型
sendx:channel中发送元素在队列中的索引
recvx:channel中接受元素在队列中的索引
recvq:等待从channel中接收元素的协程列表
sendq:等待向channel中发送元素的协程列表
lock: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是对初始化参数的合法性检查。
if elem.size >= 1<<16:
检查channel元素大小,小于2字节
if hchanSize%maxAlign != 0 || elem.align > maxAlign
没看懂(对齐?)
if size < 0 || int64(uintptr(size)) != size || (elem.size > 0 && uintptr(size) > (_MaxMem-hchanSize)/elem.size)
第一个判断缓存大小需要大于等于0
int64(uintptr(size)) != size这一句实际是用于判断size是否为负数。由于uintptr实际是一个无符号整形,负数经过转换后会变成一个与原数完全不同的很大的正整数,而正数经过转换后并没有变化。
最后一句判断channel的缓存大小要小于heap中能分配的大小。_MaxMem是可分配的堆大小。
第二部分是具体的内存分配。
元素类型为kindNoPointers的时候,既非指针类型,则直接分配(hchanSize+uintptr(size)*elem.size)大小的连续空间。c.buf指向hchan后面的elem队列首地址。
如果channel缓存大小为0,则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)
close判断
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()休眠当前协程,从而最终造成死锁。
接收操作同样先进行加锁,然后开始正式操作。
close处理
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()的接收动作分为两种情况:
c.dataqsiz == 0:即当channel为无缓存channel时,直接将发送协程中的数据,拷贝给接收者。
c.dataqsiz != 0:如果channel有缓存,则:根据缓存的接收游标,从缓存队列中取出一个,拷贝给接受者。
关于“Golang通道channel使用源码分析”这篇文章的内容就介绍到这里,感谢各位的阅读!相信大家对“Golang通道channel使用源码分析”知识都有一定的了解,大家如果还想学习更多知识,欢迎关注亿速云行业资讯频道。
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。