深刻理解go-channel和select的原理
Go最吸引人的兩個地方，除了goroutine，也就是channel了，同時，我一直很納悶，select究竟是怎麼實現的？跟我以前的文章同樣，部分無關的代碼直接省略golang
1. 結構概覽
1.1. hchan
這個就是channel的結構體了數組
type hchan struct {
qcount uint // 隊列中數據總量
dataqsiz uint // 環形隊列的大小，> 0表示有緩衝，= 0表示無緩衝
buf unsafe.Pointer // 指向元素數組的指針
elemsize uint16 // 單個元素的大小
closed uint32 // 代表是否close了
elemtype *_type // 元素類型，後面寫interface的時候再具體介紹
sendx uint // send數組的索引, c <- i
recvx uint // receive 數組的索引 <- c
recvq waitq // 等待recv 數據的goroutine的鏈表
sendq waitq // 等待send數據的goroutine鏈表
lock mutex
}
1.2. waitq
type waitq struct {
first *sudog
last *sudog
}
1.3. sudog
sudog 表明了一個在等待中的g緩存
type sudog struct {
g *g
// isSelect indicates g is participating in a select, so
// g.selectDone must be CAS'd to win the wake-up race.
isSelect bool
next *sudog
prev *sudog
elem unsafe.Pointer // 數據元素， c <- 1, 此時就是 1
// The following fields are never accessed concurrently.
// For channels, waitlink is only accessed by g.
// For semaphores, all fields (including the ones above)
// are only accessed when holding a semaRoot lock.
acquiretime int64
releasetime int64
ticket uint32
parent *sudog // semaRoot binary tree
waitlink *sudog // g.waiting list or semaRoot
waittail *sudog // semaRoot
c *hchan // channel
}
1.4. hcase
這個是 select 中一個case生成的結構體app
type scase struct {
c *hchan // chan
elem unsafe.Pointer // data element
kind uint16 // 當前case的類型，nil recv send 仍是 default
pc uintptr // race pc (for race detector / msan)
releasetime int64
}
經過上面的結構，咱們能夠看出，channel的內部實質就是一個緩衝池+兩個隊列（send recv），那麼數據是如何交互的呢，網上有個示意圖，展現的仍是比較形象的異步
1.5. 圖示
1.5.1. 無緩衝（同步）
1.5.2. 帶緩衝（異步）
綜合 上面的結構和圖示，大概能夠推測出 channel 的send recv流程函數
-
若是是recv（<-channel ）請求，則先去判斷一個sendq隊列裏有沒有人等待這放數據工具
- 若是sendq隊列不爲空且緩衝池不爲空，那麼這個sendq隊列是在等待着放數據，recv的這個g從緩衝池拿數據，而後把sendq的第一個g攜帶的數據放入到buf緩衝池裏面便可
- 若是sendq不爲空可是緩衝池爲空，那麼這個是不帶緩衝池的chan，我從sendq裏面拿第一個g的數據就ok了
- 若是sendq爲空，那就去緩衝池看看，緩衝池有數據，那就拿了就走了
- 若是sendq爲空，緩衝池也沒有數據，那就在這等着吧
- 若是send，流程跟recv是同樣的
- 若是此時 channel 被close了，喚醒全部等待的隊列 （sendq 或 recvq）裏面的等待的g，告訴他們channel.close = true
接下來就是跟蹤源碼，證實及糾正猜測了oop
2. 源碼分析
2.1. 收發
2.1.1. main
咱們使用 go tool 工具分析一下，channel 生成， c <- i， <- c 在底層都是經過什麼方法實現的源碼分析
func main() {
c1 := make(chan int)
c2 := make(chan int, 2)
go func() {
c1 <- 1
c2 <- 2
}()
<-c1
<-c2
close(c1)
close(c2)
}
go build -gcflags=all="-N -l" main.gogo tool objdump -s "main.main" main學習
咱們把 CALL 過濾出來後
▶ go tool objdump -s "main\.main" main | grep CALL main.go:4 0x4548d5 e806fbfaff CALL runtime.makechan(SB) main.go:5 0x4548f8 e8e3fafaff CALL runtime.makechan(SB) main.go:6 0x454929 e822a1fdff CALL runtime.newproc(SB) main.go:10 0x454940 e81b08fbff CALL runtime.chanrecv1(SB) main.go:11 0x454957 e80408fbff CALL runtime.chanrecv1(SB) main.go:12 0x454965 e82605fbff CALL runtime.closechan(SB) main.go:13 0x454973 e81805fbff CALL runtime.closechan(SB) main.go:3 0x454982 e8d981ffff CALL runtime.morestack_noctxt(SB) main.go:7 0x454a32 e899fcfaff CALL runtime.chansend1(SB) main.go:8 0x454a4c e87ffcfaff CALL runtime.chansend1(SB) main.go:6 0x454a5b e80081ffff CALL runtime.morestack_noctxt(SB)
- makechan: 建立channel的函數，有無緩衝區的都是同樣的
- chanrecv1: <- c1 時，調用的函數
- closechan: close(c1) 時調用的函數，關閉channel使用
- chansend1: c1 <- 1 時，也就是發送數據用到的函數
2.1.2. makechan
建立channel這一塊主要就是給結構體和bug緩衝池分配內存，而後初始化一下hchan的結構體
func makechan(t *chantype, size int) *hchan {
elem := t.elem
// compiler checks this but be safe.
// 校驗elem的大小限制
if elem.size >= 1<<16 {
throw("makechan: invalid channel element type")
}
// 對齊限制
if hchanSize%maxAlign != 0 || elem.align > maxAlign {
throw("makechan: bad alignment")
}
// size，即make(chan int, 2)中的2，默認不傳爲0， 判斷size的上限和下限
if size < 0 || uintptr(size) > maxSliceCap(elem.size) || uintptr(size)*elem.size > maxAlloc-hchanSize {
panic(plainError("makechan: size out of range"))
}
var c *hchan
switch {
case size == 0 || elem.size == 0:
// 隊列或者元素size爲0，不分配緩衝池
// Queue or element size is zero.
c = (*hchan)(mallocgc(hchanSize, nil, true))
// Race detector uses this location for synchronization.
// buf指向自身，沒有分配內存
c.buf = c.raceaddr()
case elem.kind&kindNoPointers != 0:
// Elements do not contain pointers.
// Allocate hchan and buf in one call.
// 分配一整塊內存，用於存儲hchan和 buf
c = (*hchan)(mallocgc(hchanSize+uintptr(size)*elem.size, nil, true))
c.buf = add(unsafe.Pointer(c), hchanSize)
default:
// Elements contain pointers.
// 是指針類型，那正常分配hchan結構體便可，buf單獨分配
c = new(hchan)
c.buf = mallocgc(uintptr(size)*elem.size, elem, true)
}
// 初始化 hchan的屬性
c.elemsize = uint16(elem.size)
c.elemtype = elem
c.dataqsiz = uint(size)
return c
}
2.1.3. chanrecv1
chanrecv1 調用了chanrecv 實現，chanrecv 監聽channel並接收 channel裏面的數據，並寫入到 ep 裏面
func chanrecv1(c *hchan, elem unsafe.Pointer) {
chanrecv(c, elem, true)
}
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
lock(&c.lock)
if c.closed != 0 && c.qcount == 0 {
unlock(&c.lock)
if ep != nil {
// 清空地址裏面的數據值，但不會改變類型
typedmemclr(c.elemtype, ep)
}
return true, false
}
if sg := c.sendq.dequeue(); sg != nil {
// 獲取一個等待send的sudog，而後判斷channel是否有緩衝區，若是有無緩衝區，獲取sudog裏面的數據便可， 若是channel有緩衝區，則獲取緩衝區的頭元素，把獲取到的sudog的元素添加到緩衝區的隊尾
recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true, true
}
if c.qcount > 0 {
// Receive directly from queue
// 緩衝區有數據，且send隊列沒有等待發送數據的sudog，（異步且緩衝區剛滿或未滿的狀況），根據recvx索引，獲取數據
qp := chanbuf(c, c.recvx)
// 若是ep不爲nil，拷貝 gp 到 ep
if ep != nil {
typedmemmove(c.elemtype, ep, qp)
}
// gp地址裏的數據清除
typedmemclr(c.elemtype, qp)
// 更新下一次recv的索引
c.recvx++
if c.recvx == c.dataqsiz {
c.recvx = 0
}
// 更新 qcount計數
c.qcount--
unlock(&c.lock)
return true, true
}
if !block {
unlock(&c.lock)
return false, false
}
// no sender available: block on this channel.
// 找不到send 的sudog，緩衝區也沒有數據，須要阻塞
gp := getg()
// 獲取一個sudog的結構，並更新這個sudog的屬性
mysg := acquireSudog()
mysg.releasetime = 0
// 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
// 把這個sudog放入到recv的隊列
c.recvq.enqueue(mysg)
// 休眠這個g，當g被喚醒後，從這裏繼續執行
goparkunlock(&c.lock, waitReasonChanReceive, traceEvGoBlockRecv, 3)
// someone woke us up
if mysg != gp.waiting {
throw("G waiting list is corrupted")
}
gp.waiting = nil
if mysg.releasetime > 0 {
blockevent(mysg.releasetime-t0, 2)
}
closed := gp.param == nil
gp.param = nil
mysg.c = nil
// 清理完sudog的屬性後，把sudog釋放
releaseSudog(mysg)
return true, !closed
}
經過上面的邏輯，能夠看出來數據傳輸的四種可能
- sendq隊列不爲空，可是buf爲空（同步有阻塞g的狀況）: 獲取sendq隊列頭的sudog，並把sudog.elem數據拷貝目標地址 ep
- sendq隊列不爲空，buf也不爲空（異步有阻塞g的狀況）：把buf頭元素拷貝到目標地址ep， 獲取sendq隊列頭的sudog，而後把sudog.elem的數據拷貝到buf隊尾，釋放sudog
- sendq隊列爲空，可是buf不爲空（異步無阻塞g的狀況）：把buf頭元素拷貝到目標地址ep便可
- sendq隊列爲空，buf也爲空（同步無阻塞g的狀況）：這時候就須要就須要阻塞自身，獲取一個sudog的結構，放到channel的recvq隊列裏，等待send的g來喚醒本身，並把本身的數據拷貝到目標地址
這裏細想一下，其實會發現一個問題，在上面L66 goparkunlock(&c.lock, waitReasonChanReceive, traceEvGoBlockRecv, 3) 休眠g後，g被喚醒後從這裏開始繼續往下執行，好像沒有什麼邏輯顯示，這個recv g獲取到了數據，這個g阻塞在這裏是爲了等數據來的，可是下面的邏輯，居然沒有一個是操做數據的？
接下來分析的 recv 這個方法就能理解了
2.1.3.1. recv
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
// 若是是無緩衝區的channel
if c.dataqsiz == 0 {
if ep != nil {
// copy data from sender
// 直接在兩個g之間進行數據拷貝
recvDirect(c.elemtype, sg, ep)
}
} else {
// 這裏是有緩衝區纔會走到的邏輯
// Queue is full. Take the item at the
// head of the queue. Make the sender enqueue
// its item at the tail of the queue. Since the
// queue is full, those are both the same slot.
// 由於在sendq隊列獲取到了等待發送數據的sudog，因此說明緩衝區已經滿了，根據rcvx獲取buf裏面隊列首元素的地址
qp := chanbuf(c, c.recvx)
// copy data from queue to receiver
if ep != nil {
// 把buf裏面的數據拷貝到ep裏面
typedmemmove(c.elemtype, ep, qp)
}
// copy data from sender to queue
// 把從sendq隊列獲取到的sudog的數據拷貝到剛剛的buf地址裏面，並更新buf裏面recvx的索引，也就是表名，buf隊列的首元素地址後移
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
}
// 清空sudog的數據
sg.elem = nil
gp := sg.g
unlockf()
gp.param = unsafe.Pointer(sg)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
// 喚醒sendq裏面獲取的sugog對應的g
goready(gp, skip+1)
}
結合上面的邏輯就發現，g在被喚醒以前，跟g相關的sudog的數據就已經被channel使用掉了，因此當g被喚醒時，無需處理跟數據傳輸相關的邏輯了
2.1.3.2. acquireSudog
獲取一個sudog的結構，這裏跟cache和scheduler調度待運行g的隊列同樣，使用了 p sched 的兩級緩存，也就是本地緩存一個sudog的數組，同時在全局的 sched結構上面也維護了一個sudogcache的鏈表，當p本地的sudog不足或者過多的時候，就去跟全局的sched 進行平衡
func acquireSudog() *sudog {
// 加鎖
mp := acquirem()
pp := mp.p.ptr()
// 若是當前緩存的沒有sudog了，則去全局的sched中批量拉取一些sudog緩存到當前p
if len(pp.sudogcache) == 0 {
lock(&sched.sudoglock)
// First, try to grab a batch from central cache.
for len(pp.sudogcache) < cap(pp.sudogcache)/2 && sched.sudogcache != nil {
s := sched.sudogcache
sched.sudogcache = s.next
s.next = nil
pp.sudogcache = append(pp.sudogcache, s)
}
unlock(&sched.sudoglock)
// If the central cache is empty, allocate a new one.
if len(pp.sudogcache) == 0 {
pp.sudogcache = append(pp.sudogcache, new(sudog))
}
}
// 從本地緩存的sudog裏面，獲取第一個返回，並更新sudogcache slice
n := len(pp.sudogcache)
s := pp.sudogcache[n-1]
pp.sudogcache[n-1] = nil
pp.sudogcache = pp.sudogcache[:n-1]
if s.elem != nil {
throw("acquireSudog: found s.elem != nil in cache")
}
// 去鎖
releasem(mp)
return s
}
2.1.3.3. releaseSudog
releaseSudog 就是釋放當前使用的sudog，並平衡p本地緩存的sudog和全局隊列的sudog
func releaseSudog(s *sudog) {
mp := acquirem() // avoid rescheduling to another P
pp := mp.p.ptr()
// 若是 p本地緩存的sudog的數量超出這個slice的最大長度，則平衡通常的sudog到全局的sched上面
if len(pp.sudogcache) == cap(pp.sudogcache) {
// Transfer half of local cache to the central cache.
var first, last *sudog
for len(pp.sudogcache) > cap(pp.sudogcache)/2 {
n := len(pp.sudogcache)
p := pp.sudogcache[n-1]
pp.sudogcache[n-1] = nil
pp.sudogcache = pp.sudogcache[:n-1]
if first == nil {
first = p
} else {
last.next = p
}
last = p
}
lock(&sched.sudoglock)
last.next = sched.sudogcache
sched.sudogcache = first
unlock(&sched.sudoglock)
}
// 把釋放的sudog放到本地緩存的slice裏面
pp.sudogcache = append(pp.sudogcache, s)
releasem(mp)
}
2.1.4. chansend1
發送邏輯跟接收的邏輯差很少
func chansend1(c *hchan, elem unsafe.Pointer) {
chansend(c, elem, true, getcallerpc())
}
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
lock(&c.lock)
// 從recvq隊列獲取一個 sudog
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
}
// 若是qcount < dataqsiz，說明這個channel是帶buf的channel，並且buf沒有滿，直接把數據ep添加到buf隊尾便可
if c.qcount < c.dataqsiz {
// Space is available in the channel buffer. Enqueue the element to send.
qp := chanbuf(c, c.sendx)
typedmemmove(c.elemtype, qp, ep)
c.sendx++
if c.sendx == c.dataqsiz {
c.sendx = 0
}
// 更新qcount
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.
// 走到這裏說明，buf滿了或者沒有buf，並且recvq隊列爲空，就須要阻塞當前的g，等待有其餘的g接收數據
gp := getg()
// 獲取一個sudog，並初始化相關屬性
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
// 把sudog入隊sendq
c.sendq.enqueue(mysg)
// 休眠當前g，等待其餘的g recv數據，recv數據後，喚醒這個g
goparkunlock(&c.lock, waitReasonChanSend, traceEvGoBlockSend, 3)
// someone woke us up.
if mysg != gp.waiting {
throw("G waiting list is corrupted")
}
gp.waiting = nil
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
// 釋放sudog
releaseSudog(mysg)
return true
}
2.1.4.1. send
send 跟 recv 的邏輯也是大體相同的，並且由於從recvq裏面拿到了一個sudog，因此說明緩衝區爲空，那麼send方法就不須要考慮往緩衝區添加數據了，send比recv更加簡單，只須要交換數據、喚醒g便可
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
if sg.elem != nil {
sendDirect(c.elemtype, sg, ep)
sg.elem = nil
}
gp := sg.g
unlockf()
gp.param = unsafe.Pointer(sg)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(gp, skip+1)
}
2.1.5. closechan
收發數據已經結束了，最後就是關閉channel了
func closechan(c *hchan) {
// nil chan 檢查
if c == nil {
panic(plainError("close of nil channel"))
}
lock(&c.lock)
// closed chan 檢查
if c.closed != 0 {
unlock(&c.lock)
panic(plainError("close of closed channel"))
}
// 設置c爲closed狀態
c.closed = 1
var glist *g
// release all readers
// 遍歷 recvq，清除sudog的數據，並把recvq中sudog對應的g串成一個鏈表
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
gp.schedlink.set(glist)
glist = gp
}
// release all writers (they will panic)
// 遍歷sendq，清除sudog的數據，並把sendq中的sudog中的g和recvq中的sudog一塊兒串成一個鏈表
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())
}
gp.schedlink.set(glist)
glist = gp
}
unlock(&c.lock)
// Ready all Gs now that we've dropped the channel lock.
// 喚醒上面收集的全部的g
for glist != nil {
gp := glist
glist = glist.schedlink.ptr()
gp.schedlink = 0
goready(gp, 3)
}
}
chan close以後，全部阻塞的recvq 和 sendq（recvq和sendq只有有一個隊列存在）中的sudog，清除sudog的一些數據和狀態，設置 gp.param = nil， 讓上層邏輯知道這是由於 close chan致使的
喚醒全部的g以後，g就會 繼續執行 chansend 或者 chanrecv 中剩餘的邏輯，也就是釋放sudog（這也就是爲何 closechan 不須要釋放sudog的緣由）
2.1.6. 總結
語言的表述老是蒼白的，在網上找資料的時候正好看到了兩張流程圖，能夠結合着來看
發送流程（send）
接收流程（recv）
2.2. select
2.2.1. main
channel的收發流程在上面已經追蹤了，流程也已經清晰了，可是跟channel一塊兒使用的還有一個select，那select的流程又是什麼呢
咱們仍是用go tool工具分析一下
func main() {
c1 := make(chan int)
c2 := make(chan int)
go func() {
time.Sleep(time.Second)
<-c2
c1 <- 1
}()
select {
case v := <-c1:
fmt.Printf("%d <- c1", v)
case c2 <- 1:
fmt.Println("c2 <- 1")
}
}
分析結果過濾一下CALL
main.go:9 0x4a05c6 e81542f6ff CALL runtime.makechan(SB) main.go:10 0x4a05ec e8ef41f6ff CALL runtime.makechan(SB) main.go:11 0x4a0620 e82b3bf9ff CALL runtime.newproc(SB) main.go:16 0x4a0654 e82c94fbff CALL 0x459a85 main.go:16 0x4a06e3 e8d8b7f9ff CALL runtime.selectgo(SB) main.go:18 0x4a074c e8df8df6ff CALL runtime.convT2E64(SB) main.go:18 0x4a07ec e8cf89ffff CALL fmt.Printf(SB) main.go:18 0x4a0806 e8f587fbff CALL runtime.gcWriteBarrier(SB) main.go:20 0x4a088c e87f8bffff CALL fmt.Println(SB) main.go:8 0x4a0898 e85369fbff CALL runtime.morestack_noctxt(SB) main.go:12 0x4a0945 e8868efaff CALL time.Sleep(SB) main.go:13 0x4a095c e8ff4bf6ff CALL runtime.chanrecv1(SB) main.go:14 0x4a0976 e85541f6ff CALL runtime.chansend1(SB) main.go:11 0x4a0985 e86668fbff CALL runtime.morestack_noctxt(SB)
能夠看出來，select 的實現是靠 selectgo 函數的
覺得就這樣嗎，而後咱們就開始分析 selectgo 函數了，不，在我手賤的時候還發現了另外一種狀況
func main() {
c1 := make(chan int)
go func() {
time.Sleep(time.Second)
c1 <- 1
}()
select {
case <-c1:
fmt.Printf("c1 <- 1")
default:
fmt.Println("default")
}
}
分析結果以下：
main.go:9 0x49eca8 e8335bf6ff CALL runtime.makechan(SB) main.go:11 0x49eccf e85c54f9ff CALL runtime.newproc(SB) main.go:17 0x49ece6 e83570f6ff CALL runtime.selectnbrecv(SB) main.go:18 0x49ed1c e88f8bffff CALL fmt.Printf(SB) main.go:22 0x49ed8f e86c8dffff CALL fmt.Println(SB) main.go:8 0x49ed96 e8556cfbff CALL runtime.morestack_noctxt(SB) main.go:12 0x49ee35 e87692faff CALL time.Sleep(SB) main.go:13 0x49ee4f e87c5cf6ff CALL runtime.chansend1(SB) main.go:11 0x49ee5e e88d6bfbff CALL runtime.morestack_noctxt(SB)
能夠看到，這裏 select 的實現是依靠底層的 selectnbrecv 的函數的，若是，既然有 selectnbrecv 函數，會不會有 selectnbsend 函數呢，繼續試驗一下
func main() {
c1 := make(chan int)
go func() {
time.Sleep(time.Second)
<- c1
}()
select {
case c1 <- 1:
fmt.Printf("c1 <- 1")
default:
fmt.Println("default")
}
}
分析j結果
main.go:9 0x49ecb3 e8285bf6ff CALL runtime.makechan(SB) main.go:11 0x49ecda e85154f9ff CALL runtime.newproc(SB) main.go:17 0x49ed05 e81670f6ff CALL runtime.selectnbsend(SB) main.go:18 0x49ed3b e8708bffff CALL fmt.Printf(SB) main.go:22 0x49edb4 e8478dffff CALL fmt.Println(SB) main.go:8 0x49edbb e8306cfbff CALL runtime.morestack_noctxt(SB) main.go:12 0x49ee65 e84692faff CALL time.Sleep(SB) main.go:13 0x49ee7c e8df66f6ff CALL runtime.chanrecv1(SB) main.go:11 0x49ee8b e8606bfbff CALL runtime.morestack_noctxt(SB)
這裏就是用 selectnbsend 函數實現了 select 語句，而後繼續試驗，得出結論以下：
- 若是select語句中只有一個 case在等待從channel中接收數據，則調用
selectnbrecv實現 - 若是select語句中只有一個 case在等待向channel發送數據，則調用
selectnbsend實現 - 若是select語句中有多個case，在等待向一個或多個channel發送或接收數據，則調用
selectgo實現
好了，咱們開始從 selectgo 開始跟蹤了，可是跟蹤selectgo以前，咱們須要選跟蹤一下 reflect_rselect ， 否則看着 selectgo 函數的參數，徹底就是一臉懵逼啊
2.2.2. reflect_rselect
func reflect_rselect(cases []runtimeSelect) (int, bool) {
// 若是沒有case的select，休眠當前goroutine
if len(cases) == 0 {
block()
}
sel := make([]scase, len(cases))
order := make([]uint16, 2*len(cases))
for i := range cases {
rc := &cases[i]
switch rc.dir {
case selectDefault:
sel[i] = scase{kind: caseDefault}
case selectSend:
// 若是是發送的話，c <- 1, rc.val 就是1的地址
sel[i] = scase{kind: caseSend, c: rc.ch, elem: rc.val}
case selectRecv:
// 若是是接收的話，v:= <- c, rc.val 就是v的地址
sel[i] = scase{kind: caseRecv, c: rc.ch, elem: rc.val}
}
}
return selectgo(&sel[0], &order[0], len(cases))
}
2.2.3. selectgo
func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool) {
cas1 := (*[1 << 16]scase)(unsafe.Pointer(cas0))
order1 := (*[1 << 17]uint16)(unsafe.Pointer(order0))
// order是 2*ncases長度的slice，而後把 order[0-ncases] 給 pollorder用，order[ncases-2ncases] 給lockorder用
scases := cas1[:ncases:ncases]
pollorder := order1[:ncases:ncases]
lockorder := order1[ncases:][:ncases:ncases]
// Replace send/receive cases involving nil channels with
// caseNil so logic below can assume non-nil channel.
for i := range scases {
cas := &scases[i]
if cas.c == nil && cas.kind != caseDefault {
*cas = scase{}
}
}
// The compiler rewrites selects that statically have
// only 0 or 1 cases plus default into simpler constructs.
// The only way we can end up with such small sel.ncase
// values here is for a larger select in which most channels
// have been nilled out. The general code handles those
// cases correctly, and they are rare enough not to bother
// optimizing (and needing to test).
// generate permuted order
// 肯定輪詢的順序
for i := 1; i < ncases; i++ {
j := fastrandn(uint32(i + 1))
pollorder[i] = pollorder[j]
pollorder[j] = uint16(i)
}
// sort the cases by Hchan address to get the locking order.
// simple heap sort, to guarantee n log n time and constant stack footprint.
// 經過hchan的地址來肯定加鎖順序，使用堆排序減小時間複雜度
for i := 0; i < ncases; i++ {
j := i
// Start with the pollorder to permute cases on the same channel.
c := scases[pollorder[i]].c
for j > 0 && scases[lockorder[(j-1)/2]].c.sortkey() < c.sortkey() {
k := (j - 1) / 2
lockorder[j] = lockorder[k]
j = k
}
lockorder[j] = pollorder[i]
}
for i := ncases - 1; i >= 0; i-- {
o := lockorder[i]
c := scases[o].c
lockorder[i] = lockorder[0]
j := 0
for {
k := j*2 + 1
if k >= i {
break
}
if k+1 < i && scases[lockorder[k]].c.sortkey() < scases[lockorder[k+1]].c.sortkey() {
k++
}
if c.sortkey() < scases[lockorder[k]].c.sortkey() {
lockorder[j] = lockorder[k]
j = k
continue
}
break
}
lockorder[j] = o
}
// lock all the channels involved in the select
// 根據上面肯定的加鎖順序 lockorder，來逐個對case進行加鎖
sellock(scases, lockorder)
var (
gp *g
sg *sudog
c *hchan
k *scase
sglist *sudog
sgnext *sudog
qp unsafe.Pointer
nextp **sudog
)
loop:
// pass 1 - look for something already waiting
var dfli int
var dfl *scase
var casi int
var cas *scase
var recvOK bool
for i := 0; i < ncases; i++ {
// 根據pollorder，獲取當前輪詢到的case
casi = int(pollorder[i])
cas = &scases[casi]
c = cas.c
switch cas.kind {
// nil類型的case，無視，繼續下一個
case caseNil:
continue
case caseRecv:
// recv類型的case，判斷sendq的隊列中有沒有等待發送數據的sudog，若是獲取到的話，跳轉到 recv
sg = c.sendq.dequeue()
if sg != nil {
goto recv
}
// 沒有sudog在sendq隊列排隊，而後檢查buf裏面是否有數據，若是buf裏有，則跳轉到bufrecv
if c.qcount > 0 {
goto bufrecv
}
// 最後 sendq buf都拿不到數據，則判斷這個channel是否爲關閉狀態了
// 因此 能夠看出來，若是咱們關閉一個帶buf的channel，在關閉以後仍是能把以前存儲的數據讀完的
if c.closed != 0 {
goto rclose
}
case caseSend:
// send 類型的case，首先確認channel是否關閉
if c.closed != 0 {
goto sclose
}
// 而後判斷，recvq隊列裏面有沒有等待接收數據的sudog，有則跳轉到 send 標籤
sg = c.recvq.dequeue()
if sg != nil {
goto send
}
// 判斷是否有空餘的buf位置，可讓本身把數據放上去，若是有，則跳轉到bufsend標籤
if c.qcount < c.dataqsiz {
goto bufsend
}
case caseDefault:
// 更新並記錄 case的索引及地址
dfli = casi
dfl = cas
}
}
// 根據 dfl 來判斷是否有 default，而且走到了
// 在全部 case遍歷完成後，若是不須要等待，都會跳轉到相應的標籤，例如 recv bufrecv send等，若是走到這裏，說明全部的case都沒法直接獲取或發送數據，等待另外一個g的就緒
if dfl != nil {
selunlock(scases, lockorder)
casi = dfli
cas = dfl
// 若是有default，直接執行default
goto retc
}
// pass 2 - enqueue on all chans
// 流程執行到這裏，全部的case都須要等待，且沒有default執行
gp = getg()
if gp.waiting != nil {
throw("gp.waiting != nil")
}
nextp = &gp.waiting
// 按照lockorder，對每一個case，建立相應的sudog並放入case對應的channel的recvq或sendq隊列
for _, casei := range lockorder {
casi = int(casei)
cas = &scases[casi]
if cas.kind == caseNil {
continue
}
c = cas.c
// 每個case獲取一個sudog，綁定到case對應的cahnnel的sendq或recvq隊列
sg := acquireSudog()
sg.g = gp
sg.isSelect = true
// No stack splits between assigning elem and enqueuing
// sg on gp.waiting where copystack can find it.
sg.elem = cas.elem
sg.releasetime = 0
if t0 != 0 {
sg.releasetime = -1
}
sg.c = c
// Construct waiting list in lock order.
// 按照lockorder，把這些sudog，依賴sudog.waitlink串聯起來
*nextp = sg
nextp = &sg.waitlink
switch cas.kind {
case caseRecv:
// 若是recv，放入到recvq隊列
c.recvq.enqueue(sg)
case caseSend:
// 若是是send，放入到sendq隊列
c.sendq.enqueue(sg)
}
}
// wait for someone to wake us up
// 休眠等待喚醒
gp.param = nil
gopark(selparkcommit, nil, waitReasonSelect, traceEvGoBlockSelect, 1)
//
sellock(scases, lockorder)
gp.selectDone = 0
sg = (*sudog)(gp.param)
gp.param = nil
// pass 3 - dequeue from unsuccessful chans
// otherwise they stack up on quiet channels
// record the successful case, if any.
// We singly-linked up the SudoGs in lock order.
casi = -1
cas = nil
sglist = gp.waiting
// Clear all elem before unlinking from gp.waiting.
// 在解散waiting這個隊列前，先把數據清空，由於執行到這列，確定是由於另外一個goroutine在recv或send 某個channel，而且拿到數據致使的，因此，執行到這裏後，數據都沒用了
for sg1 := gp.waiting; sg1 != nil; sg1 = sg1.waitlink {
sg1.isSelect = false
sg1.elem = nil
sg1.c = nil
}
gp.waiting = nil
for _, casei := range lockorder {
k = &scases[casei]
if k.kind == caseNil {
continue
}
if sglist.releasetime > 0 {
k.releasetime = sglist.releasetime
}
if sg == sglist {
// sg has already been dequeued by the G that woke us up.
// 肯定這個sudog致使的自身被喚醒
casi = int(casei)
cas = k
} else {
// 把其餘還在等待的sudog從等待隊列中移除
c = k.c
if k.kind == caseSend {
c.sendq.dequeueSudoG(sglist)
} else {
c.recvq.dequeueSudoG(sglist)
}
}
sgnext = sglist.waitlink
sglist.waitlink = nil
releaseSudog(sglist)
sglist = sgnext
}
if cas == nil {
// 若是cas爲nil，說明有可能由於其餘因素被喚醒，再循環一次
goto loop
}
c = cas.c
if cas.kind == caseRecv {
recvOK = true
}
selunlock(scases, lockorder)
goto retc
bufrecv:
// can receive from buffer
// recv操做，並buf不爲空，從buf中獲取數據便可
recvOK = true
qp = chanbuf(c, c.recvx)
if cas.elem != nil {
typedmemmove(c.elemtype, cas.elem, qp)
}
typedmemclr(c.elemtype, qp)
// 更新buf中recvx的索引
c.recvx++
if c.recvx == c.dataqsiz {
c.recvx = 0
}
// 更新buf中數據的數量
c.qcount--
// 解鎖當前case
selunlock(scases, lockorder)
goto retc
bufsend:
// can send to buffer
// send操做，且buf有空餘位置存儲，把本身的數據拷貝到buf隊尾
typedmemmove(c.elemtype, chanbuf(c, c.sendx), cas.elem)
// 更新buf中sendx的索引
c.sendx++
if c.sendx == c.dataqsiz {
c.sendx = 0
}
// 更新buf中數據的數量
c.qcount++
// 解鎖當前case
selunlock(scases, lockorder)
goto retc
recv:
// can receive from sleeping sender (sg)
// recv操做，可是sendq中有sudog在等待，經過recv方法，獲取數據
recv(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2)
recvOK = true
goto retc
rclose:
// read at end of closed channel
// recv 操做，可是這個channel已經close了
selunlock(scases, lockorder)
recvOK = false
if cas.elem != nil {
typedmemclr(c.elemtype, cas.elem)
}
goto retc
send:
// can send to a sleeping receiver (sg)
// send操做，可是recvq隊列中有在等待的sudog
send(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2)
goto retc
retc:
// 返回
return casi, recvOK
sclose:
// send on closed channel
selunlock(scases, lockorder)
panic(plainError("send on closed channel"))
}
2.2.4. selectnbrecv
當一個select裏面只有一個 case，且這個case 是接收數據的操做的時候，select就會調用 selectnbrecv 函數來實現
func selectnbrecv(elem unsafe.Pointer, c *hchan) (selected bool) {
selected, _ = chanrecv(c, elem, false)
return
}
這裏就會發現 selectnbrecv 就是調用了 chanrecv 來實現，也就是咱們上面解析的 <- c1 是同樣的，就至關於 select 退變 成單獨的 <- c 的表達了
2.2.5. selectnbsend
同 selectnbrecv 同樣，當select只有一個case，且這個case是發送數據到channel的，就會退變成 c <- 1 的表達了
func selectnbsend(c *hchan, elem unsafe.Pointer) (selected bool) {
return chansend(c, elem, false, getcallerpc())
}
2.2.6. 總結
因此，select的流程大體以下
- 對每一個case進行收發判斷，是否須要阻塞，不須要，直接跳轉執行
- 若是每一個case的收發操做都須要阻塞等待，則判斷有沒有default，若是有，執行default
- 若是每一個case的收發操做都須要阻塞等待，且沒有default，那就爲每一個case建立一個sudog，綁定到case對應的channel的sendq或recvq隊列
- 若是某個sudog被臨幸，而後被喚醒了，清空全部sudog的數據等屬性，並把其餘的sudog從隊列中移除
- 至此，一個select操做結束
3. 總結
我仍是很像吐槽一下，selectgo 函數華麗麗的寫了300多行，裏面還使用了若干的 goto 去進行跳轉，真的不能夠分拆一下嗎，不過大神的代碼，仍是真的須要膜拜的
4. 參考文檔
- 《Go語言學習筆記》-- 雨痕
- Go Channel 源碼剖析
- 深刻理解 Go Channel
- Go channel實現原理剖析
相關文章
- 1. 深刻理解go-channel和select的原理
- 2. 【Runloop】深刻理解原理和機制
- 3. 深刻理解Session和cookie原理
- 4. 深刻理解VLAN原理和實驗
- 5. 深刻理解MVC原理
- 6. 深刻理解Ajax原理
- 7. 深刻理解 Dubbo 原理
- 8. 深刻理解mybatis原理
- 9. View.post() 原理深刻理解
- 10. 深刻理解CyclicBarrier原理
- 更多相關文章...
- • MyBatis的工作原理 - MyBatis教程
- • BASE原理與最終一致性 - NoSQL教程
- • ☆技術問答集錦（13）Java Instrument原理
- • Java Agent入門實戰（三）-JVM Attach原理與使用
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 深刻理解go-channel和select的原理
- 2. 【Runloop】深刻理解原理和機制
- 3. 深刻理解Session和cookie原理
- 4. 深刻理解VLAN原理和實驗
- 5. 深刻理解MVC原理
- 6. 深刻理解Ajax原理
- 7. 深刻理解 Dubbo 原理
- 8. 深刻理解mybatis原理
- 9. View.post() 原理深刻理解
- 10. 深刻理解CyclicBarrier原理