Golang channel 的实现原理

Golang channel 的实现原理

Channel 是golang语言自身提供的一种非常重要的语言特性， 它是实现任务执行队列、 协程间消息传递、高并发框架的基础。关于channel的用法的文章已经很多， 本文从channel源码的实现的角度， 讨论一下其实现原理。

关于channel放在： src/runtime/chan.go
channel的关键的结构体放在hchan里面， 它记录了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
}

创建channel
用法： ch := make(chan TYPE， size int)；
这里， type是channel里面传递的elem的类型；
size是channel缓存的大小：
如果为1， 代表非缓冲的channel， 表明channel里面最多有一个elem， 剩余的只能在channel外排队等待；
如果为0，
其实现代码如下：

func makechan(t *chantype, size int) *hchan { 
// Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers. 
// 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. 
var c *hchan 
switch { 
case size == 0 || elem.size == 0: 
// Queue or element size is zero. 
c = (*hchan)(mallocgc(hchanSize, nil, true)) 
// Race detector uses this location for synchronization. 
c.buf = unsafe.Pointer(c) 
case elem.kind&kindNoPointers != 0: 
// Elements do not contain pointers. 
// Allocate hchan and buf in one call. 
c = (*hchan)(mallocgc(hchanSize+uintptr(size)*elem.size, nil, true)) 
c.buf = add(unsafe.Pointer(c), hchanSize) 
default: 
// Elements contain pointers. 
c = new(hchan) 
c.buf = mallocgc(uintptr(size)*elem.size, elem, true) 
}

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

}

写入channel elem
用法： ch <- elem
channel是阻塞时的管道， 从channel读取的时候，可能发生如下3种情况：
channel 已经关闭， 发生panic；
从已经收到的队列内读取一个elem；
从缓存队列内读取elem；
lock(&c.lock)

    if c.closed != 0 {
        unlock(&c.lock)
        panic(plainError("send on closed channel"))
    }

    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
    }

    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
    }

    if !block {
        unlock(&c.lock)
        return false
    }

从channel elem读取elem
用法： elem, ok := <- ch
if !ok {
fmt.Println(“ch has been closed”)
}
3种可能返回值：

chan已经被关闭， OK为false， elem为该类型的空值；
如果chan此时没有值存在， 该读取语句会一直等待直到有值；
如果chan此时有值， 读取正确的值， ok为true；
代码实现：

// chanrecv receives on channel c and writes the received data to ep.
// ep may be nil, in which case received data is ignored.
// If block == false and no elements are available, returns (false, false).
// Otherwise, if c is closed, zeros *ep and returns (true, false).
// Otherwise, fills in *ep with an element and returns (true, true).
// A non-nil ep must point to the heap or the caller's stack.
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
          // 正常的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
    }

 // 从缓存的elem队列读取一个elem  
  if c.qcount > 0 {
        // Receive directly from queue
        qp := chanbuf(c, c.recvx)
        if raceenabled {
            raceacquire(qp)
            racerelease(qp)
        }
        if ep != nil {
            typedmemmove(c.elemtype, ep, qp)
        }
        typedmemclr(c.elemtype, qp)
        c.recvx++
        if c.recvx == c.dataqsiz {
            c.recvx = 0
        }
        c.qcount--
        unlock(&c.lock)
        return true, true
    }   

  // channel没有可以读取的elem， 更新channel内部状态， 等待新的elem；   
}

关闭channel
用法： close(ch)
作用: 关闭channel， 并将channel内缓存的elem清除；

代码实现：

func closechan(c *hchan) {
    var glist *g

    // release all readers
    for {
        sg := c.recvq.dequeue()
        gp := sg.g
        gp.param = nil
        gp.schedlink.set(glist)
        glist = gp
    }

    // release all writers (they will panic)
    for {
        sg := c.sendq.dequeue()
        sg.elem = nil
        gp := sg.g
        gp.param = nil
        gp.schedlink.set(glist)
        glist = gp
    }
    unlock(&c.lock)

    // Ready all Gs now that we've dropped the channel lock.
    for glist != nil {
        gp := glist
        glist = glist.schedlink.ptr()
        gp.schedlink = 0
        goready(gp, 3)
    }
}

      每天坚持学习1小时Go语言，大家加油，我是彬哥，下期见！如果文章中不同观点、意见请文章下留言或者关注下方订阅号反馈！

社区交流群：221273219
Golang语言社区论坛 ：
www.Golang.Ltd
LollipopGo游戏服务器地址：
https://github.com/Golangltd/LollipopGo
社区视频课程课件GIT地址：
https://github.com/Golangltd/codeclass

Golang语言社区