Go Chan 源碼解析

本篇文章內(nèi)容基于go1.14.2分析

golang的chan是一個(gè)內(nèi)置類型，作為csp編程的核心數(shù)據(jù)結(jié)構(gòu)，其底層數(shù)據(jù)結(jié)構(gòu)是一個(gè)叫hchan的struct：

type hchan struct {
    qcount   uint           // 隊(duì)列中的元素?cái)?shù)量
    dataqsiz uint           // （環(huán)形）隊(duì)列的大小
    buf      unsafe.Pointer // 隊(duì)列的指針
    elemsize uint16 // 元素大小
    closed   uint32 // 是否已close
    elemtype *_type // 元素類型
    sendx    uint   // 環(huán)形隊(duì)列中，send的位置
    recvx    uint   // 環(huán)形隊(duì)列中 recv的位置
    recvq    waitq  // 讀取等待隊(duì)列
    sendq    waitq  // 發(fā)送等待隊(duì)列
    lock mutex // 互斥鎖
}
image

如圖所示，chan最核心的部分由一個(gè)環(huán)形隊(duì)列和2個(gè)waitq組成，環(huán)形隊(duì)列用于存放數(shù)據(jù)（帶緩沖的情況下），waitq用于實(shí)現(xiàn)阻塞和恢復(fù)goroutine。

chan的相關(guān)操作

對(duì)chan的操作有：make、讀、寫、close，當(dāng)然還有select，這里只討論前面四個(gè)操作。

創(chuàng)建 chan

當(dāng)在代碼中使用make創(chuàng)建chan時(shí)，編譯器會(huì)根據(jù)情況自動(dòng)替換成makechan64 或者makechan，makechan64 其實(shí)還是調(diào)用了makechan函數(shù)。

func makechan(t *chantype, size int) *hchan {
    elem := t.elem
    
  // 確保元素類型的size < 2^16，
    if elem.size >= 1<<16 {
        throw("makechan: invalid channel element type")
    }
  // 檢查內(nèi)存對(duì)齊
    if hchanSize%maxAlign != 0 || elem.align > maxAlign {
        throw("makechan: bad alignment")
    }

  // 計(jì)算緩沖區(qū)所需分配內(nèi)存大小
    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:
        // 即不帶緩沖區(qū)的情況，只需要調(diào)用mallocgc分配
        c = (*hchan)(mallocgc(hchanSize, nil, true))
        // 理解為空地址
        c.buf = c.raceaddr()
    case elem.ptrdata == 0:
        // 元素類型不包含指針的情況
        c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
        c.buf = add(unsafe.Pointer(c), hchanSize)
    default:
        // 默認(rèn)情況下：包含指針
        c = new(hchan)
        c.buf = mallocgc(mem, elem, true)
    }

    c.elemsize = uint16(elem.size)
    c.elemtype = elem
    c.dataqsiz = uint(size)

    if debugChan {
        print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")
    }
    return c
}

chan 寫操作

當(dāng)對(duì)chan進(jìn)行寫入“ch <- interface{}” 時(shí)，會(huì)被編譯器替換成chansend1函數(shù)的調(diào)用,最終還是調(diào)用了chansend函數(shù)：

image
//elem 是待寫入元素的地址
func chansend1(c *hchan, elem unsafe.Pointer) {
    chansend(c, elem, true, getcallerpc())
}

先看看chansend的函數(shù)簽名，只需關(guān)注ep和block這個(gè)兩個(gè)參數(shù)即可，ep是要寫入數(shù)據(jù)的地址，block表示是否阻塞式的調(diào)用

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool

chansend有以下幾種處理流程：

  1. 當(dāng)對(duì)一個(gè)nil chan進(jìn)行寫操作時(shí)，如果是非阻塞調(diào)用，直接返回；否則將當(dāng)前協(xié)程掛起

    // chansend 對(duì)一個(gè) nil chan發(fā)送數(shù)據(jù)時(shí)，如果是非阻塞則直接返回，否則將當(dāng)前協(xié)程掛起
if c == nil {
     if !block {
         return false
     }
     gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
     throw("unreachable")
 }

  2. 非阻塞模式且chan未close，沒有緩沖區(qū)且沒有等待接收或者緩沖區(qū)滿的情況下，直接return false。

    // 1. 非阻塞模式且chan未close
  // 2. 沒有緩沖區(qū)且沒有等待接收 或者 緩沖區(qū)滿的情況下
  // 滿足以上條件直接return false
if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||
     (c.dataqsiz > 0 && c.qcount == c.dataqsiz)) {
     return false
 }

  3. c.recvq中有等待讀的接收者，將其出隊(duì)，將數(shù)據(jù)直接copy給接收者，并喚醒接收者。

    // 有等待的接收的goroutine
 // 出隊(duì)，傳遞數(shù)據(jù)
 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
 }

    recvq是一個(gè)雙向鏈表，每個(gè)sudog會(huì)關(guān)聯(lián)上一個(gè)reader（被阻塞的g）

    image

    當(dāng)sudog出隊(duì)后，會(huì)調(diào)用send方法，通過sendDirect 實(shí)現(xiàn)數(shù)據(jù)在兩個(gè)地址之間拷貝，最后調(diào)用goready喚醒reader（被阻塞的g）

    func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
 // ... 剔除無關(guān)代碼
 if sg.elem != nil {
     // 直接將數(shù)據(jù)拷貝到變量ep所在的地址
     sendDirect(c.elemtype, sg, ep)
     sg.elem = nil
 }
 gp := sg.g
 unlockf()
 gp.param = unsafe.Pointer(sg)
 if sg.releasetime != 0 {
     sg.releasetime = cputicks()
 }
 //將reader的goroutine喚起
 goready(gp, skip+1)
}

  4. 緩沖區(qū)未滿的情況下，數(shù)據(jù)放入環(huán)形緩沖區(qū)即可。

     // 緩沖區(qū)未滿
 // 將數(shù)據(jù)放到緩沖區(qū)
 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
 }

  1. 緩沖區(qū)已滿，阻塞模式下關(guān)聯(lián)一個(gè)sudog數(shù)據(jù)結(jié)構(gòu)并進(jìn)入c.sendq隊(duì)列，掛起當(dāng)前協(xié)程。

     // 阻塞的情況
 gp := getg() //拿到當(dāng)前g
 mysg := acquireSudog() // 獲取一個(gè)sudog
 mysg.releasetime = 0
 if t0 != 0 {
     mysg.releasetime = -1
 
 mysg.elem = ep //關(guān)聯(lián)ep，即待寫入的數(shù)據(jù)地址
 mysg.waitlink = nil
 mysg.g = gp
 mysg.isSelect = false
 mysg.c = c
 gp.waiting = mysg
 gp.param = nil
 c.sendq.enqueue(mysg) // 入隊(duì)
 // Signal to anyone trying to shrink our stack that we're about
 // to park on a channel. The window between when this G's status
 // changes and when we set gp.activeStackChans is not safe for
 // stack shrinking.
 atomic.Store8(&gp.parkingOnChan, 1)
 // 將g休眠，讓出cpu
  // gopark后，需等待reader來喚醒它
 gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
 // 喚醒過后
 // Ensure the value being sent is kept alive until the
 // receiver copies it out. The sudog has a pointer to the
 // stack object, but sudogs aren't considered as roots of the
 // stack tracer.
 // 保持?jǐn)?shù)據(jù)不被回收
 KeepAlive(ep)

 // someone woke us up.
 if mysg != gp.waiting {
     throw("G waiting list is corrupted")
 }
 gp.waiting = nil
 gp.activeStackChans = false
 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
 releaseSudog(mysg)
 return true

chan 讀操作

當(dāng)對(duì)chan進(jìn)行讀操作時(shí)，編譯器會(huì)替換成 chanrecv1或者chanrecv2函數(shù)，最終會(huì)調(diào)用chanrecv函數(shù)處理讀取

image
// v := <- ch
func chanrecv1(c *hchan, elem unsafe.Pointer) {
    chanrecv(c, elem, true)
}
// v, ok := <- ch
func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {
    _, received = chanrecv(c, elem, true)
    return
}

和chansend一樣，chanrecv也是支持非阻塞式的調(diào)用

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool)

chanrecv有以下幾種處理流程：

  1. 讀nil chan，如果是非阻塞，直接返回；如果是阻塞式，將當(dāng)前協(xié)程掛起。

     // 讀阻塞
 if c == nil {
     if !block {
         return
     }
     gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
     throw("unreachable")
 }

  2. 非阻塞模式下，沒有緩沖區(qū)且沒有等待寫的writer或者緩沖區(qū)沒數(shù)據(jù)，直接返回。

     if !block && (c.dataqsiz == 0 && c.sendq.first == nil ||
     c.dataqsiz > 0 && atomic.Loaduint(&c.qcount) == 0) &&
     atomic.Load(&c.closed) == 0 {
     return
 }

  3. chan已經(jīng)被close，并且隊(duì)列中沒有數(shù)據(jù)時(shí)，會(huì)將存放值的變量清零，然后返回。

     // c已經(jīng)被close 并且 沒有數(shù)據(jù)
 // 清除ep指針
 if c.closed != 0 && c.qcount == 0 {
     if raceenabled {
         raceacquire(c.raceaddr())
     }
     unlock(&c.lock)
     if ep != nil {
         typedmemclr(c.elemtype, ep)
     }
     return true, false
 }

  4. sendq中有等待的writer，writer出隊(duì)，并調(diào)用recv函數(shù)

    // 從sendq中取出sender
 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).
     // 從sender中讀取數(shù)據(jù)
     recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
     return true, true
 }
    image

    recv在這分兩種處理：如果ch不帶緩沖區(qū)的話，直接將writer的sg.elem數(shù)據(jù)拷貝到ep；如果帶緩沖區(qū)的話，此時(shí)緩沖區(qū)肯定滿了，那么就從緩沖區(qū)隊(duì)列頭部取出數(shù)據(jù)拷貝至ep，然后將writer的sg.elem數(shù)據(jù)拷貝到緩沖區(qū)中，最后喚醒writer（g）

    func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
   // 不帶緩沖區(qū)的情況
   // 直接copy from sender
   if c.dataqsiz == 0 {
      if raceenabled {
         racesync(c, sg)
      }
      if ep != nil {
         // copy data from sender
         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.
      // 隊(duì)列已滿
      // 隊(duì)列元素出隊(duì)
      qp := chanbuf(c, c.recvx)
      if raceenabled {
         raceacquire(qp)
         racerelease(qp)
         raceacquireg(sg.g, qp)
         racereleaseg(sg.g, qp)
      }
      // copy data from queue to receiver
      // 數(shù)據(jù)拷貝給ep
      if ep != nil {
         typedmemmove(c.elemtype, ep, qp)
      }
      // copy data from sender to queue
      // 將sender的數(shù)據(jù)拷貝到這個(gè)槽中
      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()
   }
   // 喚醒sender協(xié)程
   goready(gp, skip+1)
}

  5. 直接從緩沖隊(duì)列中讀數(shù)。

     // 帶緩沖區(qū)
 if c.qcount > 0 {
     // Receive directly from queue
     // 直接buf中取
     qp := chanbuf(c, c.recvx)
     if raceenabled {
         raceacquire(qp)
         racerelease(qp)
     }
     // 拷貝數(shù)據(jù)到ep指針
     if ep != nil {
         typedmemmove(c.elemtype, ep, qp)
     }
     // 清除qp
     typedmemclr(c.elemtype, qp)
     c.recvx++
     if c.recvx == c.dataqsiz {
         c.recvx = 0
     }
     c.qcount--
     unlock(&c.lock)
     return true, true
 }

  6. 阻塞的情況，緩沖區(qū)沒有數(shù)據(jù)，且沒有writer

    
 // 阻塞
 gp := getg() //拿到當(dāng)前的goroutine
 mysg := acquireSudog() // 獲取一個(gè)sudog
 mysg.releasetime = 0
 if t0 != 0 {
     mysg.releasetime = -1
 }
 
 //sudog 關(guān)聯(lián)
 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) //入隊(duì)
 // Signal to anyone trying to shrink our stack that we're about
 // to park on a channel. The window between when this G's status
 // changes and when we set gp.activeStackChans is not safe for
 // stack shrinking.
 atomic.Store8(&gp.parkingOnChan, 1)
  // 掛起當(dāng)前goroutine，等待writer喚醒
 gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 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)
 }
 closed := gp.param == nil
 gp.param = nil
 // sudog解除關(guān)聯(lián)
 mysg.c = nil
  // 釋放sudog
 releaseSudog(mysg)

close 關(guān)閉操作

當(dāng)close一個(gè)chan時(shí)，編譯器會(huì)替換成對(duì)closechan函數(shù)的調(diào)用，將closed字段置為1，并將recvq和sendq中的goroutine釋放喚醒，對(duì)sendq中未寫入的數(shù)據(jù)做清除，且writer會(huì)發(fā)生panic異常。

func closechan(c *hchan) {
    if c == nil {
        panic(plainError("close of nil channel"))
    }
    
  // 加鎖
    lock(&c.lock)
  // 不可重復(fù)close
    if c.closed != 0 {
        unlock(&c.lock)
        panic(plainError("close of closed channel"))
    }

    if raceenabled {
        callerpc := getcallerpc()
        racewritepc(c.raceaddr(), callerpc, funcPC(closechan))
        racerelease(c.raceaddr())
    }

    c.closed = 1

    var glist gList

    // 釋放所有的
    for {
        // 出隊(duì)
        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
        if raceenabled {
            raceacquireg(gp, c.raceaddr())
        }
        glist.push(gp)
    }

    // 釋放所有writer
    for {
        // 出隊(duì)
        sg := c.sendq.dequeue()
        if sg == nil {
            break
        }
        // 丟棄數(shù)據(jù)
        sg.elem = nil
        if sg.releasetime != 0 {
            sg.releasetime = cputicks()
        }
        gp := sg.g
        gp.param = nil
        if raceenabled {
            raceacquireg(gp, c.raceaddr())
        }
        glist.push(gp)
    }
    unlock(&c.lock)

    // 喚醒所有g(shù)
    for !glist.empty() {
        gp := glist.pop()
        gp.schedlink = 0
        goready(gp, 3)
    }
}

chan使用小技巧

  1. 避免read、write一個(gè)nil chan

    func main() {
 ch := make(chan int,1)

 go func() {
     time.Sleep(1*time.Second)
     ch = nil
 }()

 ch<-1 
 ch<-1 // 協(xié)程直接掛起
}

  2. 從chan中read時(shí)，使用帶指示的訪問方式，讀取的時(shí)候無法感知到close的關(guān)閉

    func main() {
 ch := make(chan int)

 go func() {
     ch <- 10
     close(ch)
 }()

 for {
     select {
      // case i, ok := <-ch:
      // if ok {
      //  break
      //}
         case i := <-ch:
             fmt.Println(i)
             time.Sleep(100 * time.Millisecond)
     }
 }
}

  3. 從chan中read時(shí)，不要使用已存在變量接收, chan close之后，緩沖區(qū)沒有數(shù)據(jù)的話，使用存在變量讀取時(shí)，會(huì)將變量清零

    func main() {
 a := 10
 ch := make(chan int,1)

 fmt.Println("before close a is: ", a) // a is 10
 close(ch)
 a = <-ch 
 fmt.Println("after close a is: ", a) // a is 0
}

  1. 使用select+default可以實(shí)現(xiàn) chan的無阻塞讀取

    // 使用select反射包實(shí)現(xiàn)無阻塞讀寫
func tryRead(ch chan int) (int, bool) {
 var cases []reflect.SelectCase
 caseRead := reflect.SelectCase{
     Dir:  reflect.SelectRecv,
     Chan: reflect.ValueOf(ch),
 }

 cases = append(cases, caseRead)
 cases = append(cases, reflect.SelectCase{
     Dir: reflect.SelectDefault,
 })

 _, v, ok := reflect.Select(cases)

 if ok {

     return (v.Interface()).(int), ok
 }

 return 0, ok
}

func tryWrite(ch chan int, data int) bool {
 var cases []reflect.SelectCase
 caseWrite := reflect.SelectCase{
     Dir:  reflect.SelectSend,
     Chan: reflect.ValueOf(ch),
     Send: reflect.ValueOf(data),
 }

 cases = append(cases, caseWrite)
 cases = append(cases, reflect.SelectCase{
     Dir: reflect.SelectDefault,
 })
 chosen, _, _ := reflect.Select(cases)

 return chosen == 0
}

// 使用select + default實(shí)現(xiàn)無阻塞讀寫
func tryRead2(ch chan int) (int, bool) {
 select {
 case v, ok := <-ch:
     return v, ok
 default:
     return 0, false
 }
}

func tryWrite2(ch chan int, data int) bool {
 select {
 case ch <- data:
     return true
 default:
     return false
 }
}

    原因是如果select的case中存在default，對(duì)chan的讀寫會(huì)使用無阻塞的方法

    func selectnbsend(c *hchan, elem unsafe.Pointer) (selected bool) {
 return chansend(c, elem, false, getcallerpc())
}

func selectnbrecv(elem unsafe.Pointer, c *hchan) (selected bool) {
 selected, _ = chanrecv(c, elem, false)
 return
}
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

  • 簡(jiǎn)介 熟悉Go的人都知道，它提倡著不要通過共享內(nèi)存來通訊，而要通過通訊來共享內(nèi)存。Go提供了一種獨(dú)特的并發(fā)同步技術(shù)...
    marsjhe閱讀 3,074評(píng)論 0贊 2
  • 設(shè)計(jì)原理 目前的 Channel 收發(fā)操作均遵循了先進(jìn)先出的設(shè)計(jì)，具體規(guī)則如下： 先從 Channel 讀取數(shù)據(jù)的...
    Xuenqlve閱讀 2,451評(píng)論 0贊 0
  • channel是golang中特有的一種數(shù)據(jù)結(jié)構(gòu)，通常與goroutine一起使用，下面我們就介紹一下這種數(shù)據(jù)結(jié)構(gòu)...
    cfanbo閱讀 333評(píng)論 0贊 0
  • 前言 Golang在并發(fā)編程上有兩大利器，分別是channel和goroutine，這篇文章我們先聊聊channe...
    即將禿頭的Java程序員閱讀 1,219評(píng)論 0贊 2
  • 簡(jiǎn)介(js) 通道(channel) 是Go實(shí)現(xiàn)CSP并發(fā)模型的關(guān)鍵， 鼓勵(lì)用通信來實(shí)現(xiàn)數(shù)據(jù)共享。 Dont' c...
    darcyaf閱讀 324評(píng)論 0贊 0

友情鏈接更多精彩內(nèi)容