「Go」- golang源码分析 - channel的底层实现

路径为：./src/runtime/chan.go 文件中，先看channel结构体:

type hchan struct {
    qcount   uint           // total data in the queue 当前队列中的数据的个数
    dataqsiz uint           // size of the circular queue   channel环形队列的大小
    buf      unsafe.Pointer // points to an array of dataqsiz elements  存放数据的环形队列的指针
    elemsize uint16     // channel 中存放的数据类型的大小|即每个元素的大小
    closed   uint32     // channel 是否关闭的标示
    elemtype *_type // element type channel中存放的元素的类型
    sendx    uint   // send index   当前发送元素指向channel环形队列的下标指针
    recvx    uint   // receive index 当前接收元素指向channel环形队列的下标指针
    recvq    waitq  // list of recv waiters 等待接收元素的goroutine队列
    sendq    waitq  // list of send waiters  等待发送元素的goroutine队列

    // 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.
    // 保持此锁定时不要更改另一个G的状态（特别是，没有准备好G），因为这可能会因堆栈收缩而死锁。
    lock mutex
}

以及waitq的结构体：

//等待发送及接收的等待接收元素的goroutine队列的结构体
type waitq struct {
    first *sudog
    last  *sudog
}

等待发送或接受goroutine链表的结构体sudog:

// sudog表示等待链表中的g，例如用于发送/接收在频道上。
// 一个G可以出现在许多等待列表中，因此一个G有许多sudog；许多G可能在等待相同的结果,同步对象，因此一个对象可能有多个sudog。
// sudog是从一个特殊的池中分配的。使用AcquireDog和
// 释放sudog来分配和释放它们。

type sudog struct {
    // 以下字段受hchan.lock的保护
    g *g // 绑定的goroutine
    isSelect bool   // isSelect的布尔值表示该线程是否正在进行操作channel
    next     *sudog // 指向下一个等待线程的指针地址
    prev     *sudog // 指向上一个等待线程的指针地址
    elem     unsafe.Pointer // data element (may point to stack) 数据对象（可能指向栈）
    // 当进行channel的send操作时，elem代表将要保存进channel的元素
    // 当进行channel的recv操作时, elem代表从channel接受的元素
    // G1执行ch<-task4的时候，G1会创建一个sudog然后将elem保存进入sendq队列

    // 从不同场景访问以下字段。
    // 对于channel，WaitLink只能由G访问。
    // 对于信号量，所有字段（包括上面的字段）只有在持有semaroot锁时才能访问。
    acquiretime int64 // 获取时间
    releasetime int64 // 释放时间
    ticket      uint32
    parent      *sudog // semaRoot binary tree
    waitlink    *sudog // g.waiting list or semaRoot
    waittail    *sudog // semaRoot
    c           *hchan // channel // 绑定channel
}

从以上三个结构体我们即可看出channel其实就是由一个环形数组实现的队列用于在确定大小的连续内存块进行数据元素的存储，用waitq以及链表sudog共同实现goroutine的等待队列，并在每个链表元素中存储待从channel中取出或拷贝进channel的数据元素,可以理解为每个等待线程都是channel的搬运工，负责运送数据.

其中hchan中的lock是 recvq 是读操作阻塞在 channel 的 goroutine 列表，sendq 是写操作阻塞在 channel 的 goroutine 列表。

qcount 和 dataqsiz 分别描述了该channel的当前使用量和最大容量。

接下来进行channel的每一个函数方法进行分析:

makechan:

func makechan(t *chantype, size int) *hchan {
    elem := t.elem

    // compiler checks this but be safe.
    // 判断定义的channel存储的每个元素大小是否在范围内
    if elem.size >= 1<<16 {
        throw("makechan: invalid channel element type")
    }

    if hchanSize%maxAlign != 0 || elem.align > maxAlign {
        throw("makechan: bad alignment")
    }

    // 计算channel所需要分配的内存大小
    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 {
    // 当计算channel的内存大小为0时创建不带buffer的channel
    case mem == 0:
        // Queue or element size is zero.
        c = (*hchan)(mallocgc(hchanSize, nil, true))
        // Race detector uses this location for synchronization.
        c.buf = c.raceaddr()
    // elem类型非指针
    // 当计算channel的内存大小为0时创建带buffer的channel
    // 分配连续的内存 （连续内存有利于提高内存使用效率）
    // 直接从栈中分配内存
    case elem.kind&kindNoPointers != 0:
        // 分配内存
        c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
        c.buf = add(unsafe.Pointer(c), hchanSize)
    // 当channel元素类型包含指针时分配离散的内存
    default:
        // Elements contain pointers.
        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, "; elemalg=", elem.alg, "; dataqsiz=", size, "\n")
    }
    return c
}

函数接收两个参数，一个是channel里面保存的元素的数据类型，一个是缓冲的容量(如果为0表示是非缓冲buffer)，创建流程如下：

根据传递的缓冲大小size是否为零，分别创建不带buffer的channel或则带size大小的缓冲channel：

对于不带缓冲channel，申请一个hchan数据结构的内存大小；

对于带缓冲channel，new一个hchan对象，并初始化buffer内存；

对于包含指针带缓存的channel同样申请一个hchan数据结构的内存大小;

以及设置channel的属性。

带指针以及不带指针带内存申请区别可以看内存管理相关源码。

chanbuf:

//chanbuf(c, i) is pointer to the i'th slot in the buffer.
func chanbuf(c *hchan, i uint) unsafe.Pointer {
    return add(c.buf, uintptr(i)*uintptr(c.elemsize))
}

chanbuf的实现很简单，主要就是根据下标（sendx或recvx）以及每一个元素的大小还有环形队列的指针计算出该下标槽点内存地址并返回

chansend：

// 通用单通道发送/接收
// 如果阻塞不是nil，则将不会休眠，但如果无法完成则返回。
// 当睡眠中涉及的通道关闭时，睡眠可以通过g.param == nil唤醒。 最简单的循环和重新运行操作; 我们会 
// 看到它现在已经关闭了。   

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

    // 当 channel 未初始化或为 nil 时，向其中发送数据将会永久阻塞
    if c == nil {
        if !block {
            return false
        }
        // gopark 会使当前 goroutine 休眠，并通过 unlockf 唤醒，但是此时传入的 unlockf 为 nil, 因此，goroutine 会一直休眠
        gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
        throw("unreachable")
    }

    if debugChan {
        print("chansend: chan=", c, "\n")
    }

    // 如果开启了竞争检测
    if raceenabled {
        racereadpc(c.raceaddr(), callerpc, funcPC(chansend))
    }

    // Fast path: check for failed non-blocking operation without acquiring the lock.
    //
    // After observing that the channel is not closed, we observe that the channel is
    // not ready for sending. Each of these observations is a single word-sized read
    // (first c.closed and second c.recvq.first or c.qcount depending on kind of channel).
    // Because a closed channel cannot transition from 'ready for sending' to
    // 'not ready for sending', even if the channel is closed between the two observations,
    // they imply a moment between the two when the channel was both not yet closed
    // and not ready for sending. We behave as if we observed the channel at that moment,
    // and report that the send cannot proceed.
    //
    // It is okay if the reads are reordered here: if we observe that the channel is not
    // ready for sending and then observe that it is not closed, that implies that the
    // channel wasn't closed during the first observation.
    if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||
        (c.dataqsiz > 0 && c.qcount == c.dataqsiz)) {
        return false
    }

    var t0 int64
    //计时器
    if blockprofilerate > 0 {
        t0 = cputicks()
    }

    // 获取同步锁
    lock(&c.lock)


    // 向已经关闭的 channel 发送消息会产生 panic
    if c.closed != 0 {
        unlock(&c.lock)
        panic(plainError("send on closed channel"))
    }

    // CASE1: 当有 goroutine 在 recv 队列上等待时，跳过缓存队列，将消息直接发给 reciever goroutine
    // dequeue 从等待接受的线程队列链表获取一个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
    }

     // CASE2: 缓存队列未满，则将消息复制到缓存队列上并移动sendx下标
    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
    }
     // CASE3: 缓存队列已满，将goroutine 加入 send 队列
     // 创建 sudo
    // Block on the channel. Some receiver will complete our operation for us.
    //获取当前线程并绑定到sudog
    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.isSelect = false
    mysg.c = c
    gp.waiting = mysg
    gp.param = nil
    // 讲当前sudog放入等待发送的线程队列
    c.sendq.enqueue(mysg)
    // 休眠线程（即阻塞）
    // 通过调用goready（gp），goroutine可以再次运行。
    goparkunlock(&c.lock, waitReasonChanSend, traceEvGoBlockSend, 3)
    // 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.
    KeepAlive(ep)

    // 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
}

// send processes a send operation on an empty channel c.
// The value ep sent by the sender is copied to the receiver sg.
// The receiver is then woken up to go on its merry way.
// Channel c must be empty and locked.  send unlocks c with unlockf.
// sg must already be dequeued from c.
// ep must be non-nil and point to the heap or the caller's stack.
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
    if raceenabled {
        if c.dataqsiz == 0 {
            racesync(c, sg)
        } else {
            // Pretend we go through the buffer, even though
            // we copy directly. Note that we need to increment
            // the head/tail locations only when raceenabled.
            qp := chanbuf(c, c.recvx)
            raceacquire(qp)
            racerelease(qp)
            raceacquireg(sg.g, qp)
            racereleaseg(sg.g, qp)
            c.recvx++
            if c.recvx == c.dataqsiz {
                c.recvx = 0
            }
            c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
        }
    }
    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)
}

send 有以下四种情况：【都是对不为nil的chan的情况】

1.向已经close的chan写数据，抛panic。

2.有 goroutine 阻塞在 channel recv 队列上，此时缓存队列( hchan.buf)为空(即缓冲区内无元素)，直接将消息发送给 reciever goroutine,只产生一次复制,从当前 channel 的等待队列中取出等待的 goroutine，然后调用 send。goready 负责唤醒 goroutine。

3.当 channel 缓存队列( hchan.buf )有剩余空间时，将数据放到队列里，等待接收，接收后总共产生两次复制

4.当 channel 缓存队列( hchan.buf )已满时，将当前 goroutine 加入 send 队列并阻塞

receive：

// 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) {
    // raceenabled: don't need to check ep, as it is always on the stack
    // or is new memory allocated by reflect.

    if debugChan {
        print("chanrecv: chan=", c, "\n")
    }

    // 从 nil 的 channel 中接收消息，永久阻塞
    if c == nil {
        if !block {
            return
        }
        gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
        throw("unreachable")
    }

    // Fast path: check for failed non-blocking operation without acquiring the lock.
    //
    // After observing that the channel is not ready for receiving, we observe that the
    // channel is not closed. Each of these observations is a single word-sized read
    // (first c.sendq.first or c.qcount, and second c.closed).
    // Because a channel cannot be reopened, the later observation of the channel
    // being not closed implies that it was also not closed at the moment of the
    // first observation. We behave as if we observed the channel at that moment
    // and report that the receive cannot proceed.
    //
    // The order of operations is important here: reversing the operations can lead to
    // incorrect behavior when racing with a close.
    if !block && (c.dataqsiz == 0 && c.sendq.first == nil ||
        c.dataqsiz > 0 && atomic.Loaduint(&c.qcount) == 0) &&
        atomic.Load(&c.closed) == 0 {
        return
    }

    var t0 int64
    if blockprofilerate > 0 {
        t0 = cputicks()
    }

    lock(&c.lock)

     // CASE1: 从已经 close 且为空的 channel recv 数据，返回空值
    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
    }

    // CASE2: send 队列不为空,直接从channel队列中获取
    // sg是sends 线程队列
    // 从sends 线程队列获取一个sudog并唤醒让其将元素推入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
    }
    // CASE3: 缓存队列不为空，此时只有可能是缓存队列已满，从队列头取出元素，
    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)
        //移动channel的recvx下标
        c.recvx++
        if c.recvx == c.dataqsiz {
            c.recvx = 0
        }
        c.qcount--
        unlock(&c.lock)
        return true, true
    }

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


    // CASE4: 缓存队列为空，将 goroutine 加入 recv 队列，并阻塞
    // no sender available: block on this channel.
    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
    gp.waiting = mysg
    mysg.g = gp
    mysg.isSelect = false
    mysg.c = c
    gp.param = nil
    c.recvq.enqueue(mysg)
    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
    releaseSudog(mysg)
    return true, !closed
}

从代码上可以很明显的看出

receive和send的四种情况相互配合相互对应实现一存一拿的执行顺序

close channel 的工作

整个channel的流程结构：

WechatIMG1513.jpeg

15141548494179_.pic_hd.jpg