Golang通道channel的源碼分析
前言
channel是golang中標(biāo)志性的概念之一,很好很強(qiáng)大!
channel(通道),顧名思義,是一種通道,一種用于并發(fā)環(huán)境中數(shù)據(jù)傳遞的通道。通常結(jié)合golang中另一重要概念goroutine(go協(xié)程)使用,使得在golang中的并發(fā)編程變得清晰簡(jiǎn)潔同時(shí)又高效強(qiáng)大。
今天嘗試著讀讀golang對(duì)channel的實(shí)現(xiàn)源碼,本文主要是自己個(gè)人對(duì)于Channel源碼的學(xué)習(xí)筆記,需要的朋友可以參考以下內(nèi)容,希望對(duì)大家有幫助。
channel基礎(chǔ)結(jié)構(gòu)
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結(jié)構(gòu)就是channel的底層數(shù)據(jù)結(jié)構(gòu),看源碼定義,可以說是非常清晰了。
- qcount:channel緩存隊(duì)列中已有的元素?cái)?shù)量
- dataqsiz:channel的緩存隊(duì)列大?。ǘxchannel時(shí)指定的緩存大小,這里channel用的是一個(gè)環(huán)形隊(duì)列)
- buf:指向channel緩存隊(duì)列的指針
- elemsize:通過channel傳遞的元素大小
- closed:channel是否關(guān)閉的標(biāo)志
- elemtype:通過channel傳遞的元素類型
- sendx:channel中發(fā)送元素在隊(duì)列中的索引
- recvx:channel中接受元素在隊(duì)列中的索引
- recvq:等待從channel中接收元素的協(xié)程列表
- sendq:等待向channel中發(fā)送元素的協(xié)程列表
- lock:channel上的鎖
其中關(guān)于recvq和sendq的兩個(gè)列表所用的結(jié)構(gòu)waitq簡(jiǎn)單看下。
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是一個(gè)雙向鏈表結(jié)構(gòu),鏈上的節(jié)點(diǎn)是sudog。從sudog的結(jié)構(gòu)定義可以粗略看出,sudog是對(duì)g(即協(xié)程)的一個(gè)封裝。用于記錄一個(gè)等待在某個(gè)channel上的協(xié)程g、等待的元素elem等信息。
channel初始化
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個(gè)if是對(duì)初始化參數(shù)的合法性檢查。
if elem.size >= 1<<16:
檢查channel元素大小,小于2字節(jié)
if hchanSize%maxAlign != 0 || elem.align > maxAlign
沒看懂(對(duì)齊?)
if size < 0 || int64(uintptr(size)) != size || (elem.size > 0 && uintptr(size) > (_MaxMem-hchanSize)/elem.size)
第一個(gè)判斷緩存大小需要大于等于0
int64(uintptr(size)) != size這一句實(shí)際是用于判斷size是否為負(fù)數(shù)。由于uintptr實(shí)際是一個(gè)無符號(hào)整形,負(fù)數(shù)經(jīng)過轉(zhuǎn)換后會(huì)變成一個(gè)與原數(shù)完全不同的很大的正整數(shù),而正數(shù)經(jīng)過轉(zhuǎn)換后并沒有變化。
最后一句判斷channel的緩存大小要小于heap中能分配的大小。_MaxMem是可分配的堆大小。
第二部分是具體的內(nèi)存分配。
元素類型為kindNoPointers的時(shí)候,既非指針類型,則直接分配(hchanSize+uintptr(size)*elem.size)大小的連續(xù)空間。c.buf指向hchan后面的elem隊(duì)列首地址。
如果channel緩存大小為0,則c.buf實(shí)際上是沒有給他分配空間的
如果類型為非kindNoPointers,則channel的空間和buf的空間是分別分配的。
channel發(fā)送
// 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發(fā)送,即協(xié)程向channel中發(fā)送數(shù)據(jù),與此操作對(duì)應(yīng)的go代碼如c <- x。
channel發(fā)送的實(shí)現(xiàn)源碼中,通過chansend1(),調(diào)用chansend(),其中block參數(shù)為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()首先對(duì)c進(jìn)行判斷, if c == nil:即channel沒有被初始化,這個(gè)時(shí)候會(huì)直接調(diào)用gopark使得當(dāng)前協(xié)程進(jìn)入等待狀態(tài)。而且用于喚醒的參數(shù)unlockf傳的nil,即沒有人來喚醒它,這樣系統(tǒng)進(jìn)入死鎖。所以channel必須被初始化之后才能使用,否則死鎖。
接下來是正式的發(fā)送處理,且后續(xù)操作會(huì)加鎖。
lock(&c.lock)
close判斷
if c.closed != 0 { unlock(&c.lock) panic(plainError("send on closed channel")) }
如果channel已經(jīng)是closed狀態(tài),解鎖然后直接panic。也就是說我們不可以向已經(jīng)關(guān)閉的通道內(nèi)在發(fā)送數(shù)據(jù)。
將數(shù)據(jù)發(fā)給接收協(xié)程
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 }
嘗試從接收等待協(xié)程隊(duì)列中取出一個(gè)協(xié)程,如果有則直接數(shù)據(jù)發(fā)給它。也就是說發(fā)送到channel的數(shù)據(jù)會(huì)優(yōu)先檢查接收等待隊(duì)列,如果有協(xié)程等待取數(shù),就直接給它。發(fā)完解鎖,操作完成。
這里send()方法會(huì)將數(shù)據(jù)寫到從隊(duì)列里取出來的sg中,通過goready()喚醒sg.g(即等待的協(xié)程),進(jìn)行后續(xù)處理。
數(shù)據(jù)放到緩存
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 }
如果沒有接收協(xié)程在等待,則去檢查channel的緩存隊(duì)列是否還有空位。如果有空位,則將數(shù)據(jù)放到緩存隊(duì)列中。
通過c.sendx游標(biāo)找到隊(duì)列中的空余位置,然后將數(shù)據(jù)存進(jìn)去。移動(dòng)游標(biāo),更新數(shù)據(jù),然后解鎖,操作完成。
if c.sendx == c.dataqsiz { c.sendx = 0 }
通過這一段游標(biāo)的處理可以看出,緩存隊(duì)列是一個(gè)環(huán)形。
阻塞發(fā)送協(xié)程
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)
如果緩存也慢了,這時(shí)候就只能阻塞住發(fā)送協(xié)程了, 等有合適的機(jī)會(huì)了,再將數(shù)據(jù)發(fā)送出去。
getg()獲取當(dāng)前協(xié)程對(duì)象g的指針,acquireSudog()生成一個(gè)sudog,然后將當(dāng)前協(xié)程及相關(guān)數(shù)據(jù)封裝好鏈接到sendq列表中。然年通過goparkunlock()將其轉(zhuǎn)為等待狀態(tài),并解鎖。操作完成。
channel接收
// entry points for <- c from compiled code //go:nosplit func chanrecv1(c *hchan, elem unsafe.Pointer) { chanrecv(c, elem, true) }
channel接收,即協(xié)程從channel中接收數(shù)據(jù),與此操作對(duì)應(yīng)的go代碼如<- c。
channel接收的實(shí)現(xiàn)源碼中,通過chanrecv1(),調(diào)用chanrecv(),其中block參數(shù)為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") } ... }
同發(fā)送一樣,接收也會(huì)首先檢查c是否為nil,如果為nil,會(huì)調(diào)用gopark()休眠當(dāng)前協(xié)程,從而最終造成死鎖。
接收操作同樣先進(jìn)行加鎖,然后開始正式操作。
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 }
接收和發(fā)送略有不同,當(dāng)channel關(guān)閉并且channel的緩存隊(duì)列里沒有數(shù)據(jù)了,那么接收動(dòng)作會(huì)直接結(jié)束,但不會(huì)報(bào)錯(cuò)。
也就是說,允許從已關(guān)閉的channel中接收數(shù)據(jù)。
從發(fā)送等待協(xié)程中接收
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 }
嘗試從發(fā)送等待協(xié)程列表中取出一個(gè)等待協(xié)程,如果存在,則調(diào)用recv()方法接收數(shù)據(jù)。
這里的recv()方法比send()方法稍微復(fù)雜一點(diǎn),我們簡(jiǎn)單分析下。
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()的接收動(dòng)作分為兩種情況:
- c.dataqsiz == 0:即當(dāng)channel為無緩存channel時(shí),直接將發(fā)送協(xié)程中的數(shù)據(jù),拷貝給接收者。
- c.dataqsiz != 0:如果channel有緩存,則:根據(jù)緩存的接收游標(biāo),從緩存隊(duì)列中取出一個(gè),拷貝給接受者
小結(jié)
channel必須初始化后才能使用;
channel關(guān)閉后,不允許在發(fā)送數(shù)據(jù),但是還可以繼續(xù)從中接收未處理完的數(shù)據(jù)。所以盡量從發(fā)送端關(guān)閉channel;
無緩存的channel需要注意在一個(gè)協(xié)程中的操作不會(huì)造成死鎖;
到此這篇關(guān)于Golang通道channel的源碼分析的文章就介紹到這了,更多相關(guān)Golang通道channel內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Go語(yǔ)言協(xié)程處理數(shù)據(jù)有哪些問題
協(xié)程(coroutine)是Go語(yǔ)言中的輕量級(jí)線程實(shí)現(xiàn),由Go運(yùn)行時(shí)(runtime)管理。本文為大家詳細(xì)介紹了Go中的協(xié)程,協(xié)程不需要搶占式調(diào)度,可以有效提高線程的任務(wù)并發(fā)性,而避免多線程的缺點(diǎn)2023-02-02手把手帶你走進(jìn)Go語(yǔ)言之類型轉(zhuǎn)換
每個(gè)函數(shù)都可以強(qiáng)制將一個(gè)表達(dá)式轉(zhuǎn)換成某種特定數(shù)據(jù)類型,本文給大家介紹了在Go語(yǔ)言中類型轉(zhuǎn)換的具體用法,講述的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值2021-09-09Go并發(fā)控制Channel使用場(chǎng)景分析
使用channel來控制子協(xié)程的優(yōu)點(diǎn)是實(shí)現(xiàn)簡(jiǎn)單,缺點(diǎn)是當(dāng)需要大量創(chuàng)建協(xié)程時(shí)就需要有相同數(shù)量的channel,而且對(duì)于子協(xié)程繼續(xù)派生出來的協(xié)程不方便控制2021-07-07詳解Go語(yǔ)言中new和make關(guān)鍵字的區(qū)別
本篇文章來介紹一道非常常見的面試題,到底有多常見呢?可能很多面試的開場(chǎng)白就是由此開始的。那就是 new 和 make 這兩個(gè)內(nèi)置函數(shù)的區(qū)別,希望對(duì)大家有所幫助2023-03-03