1. 核心数据结构#

数据存储在一个环形数组中，并且维护了两个等待队列，分别用于存放因读写操作而阻塞的协程。

1.1 hchan#

1
type hchan struct {
2
    qcount   uint           // total data in the queue
3
    dataqsiz uint           // size of the circular queue
4
    buf      unsafe.Pointer // points to an array of dataqsiz elements
5
    elemsize uint16
6
    closed   uint32
7
    elemtype *_type // element type
8
    sendx    uint   // send index
9
    recvx    uint   // receive index
10
    recvq    waitq  // list of recv waiters
11
    sendq    waitq  // list of send waiters
12

13
    lock mutex
14
}

hchan：channel 数据结构。

qcount：当前 channel 中存在多少个元素；
dataqsize: 当前 channel 能存放的元素容量；
buf：channel 中用于存放元素的环形缓冲区；
elemsize：channel 元素类型的大小；
closed：标识 channel 是否关闭；
elemtype：channel 元素类型；
sendx：发送元素进入环形缓冲区的 index；
recvx：接收元素所处的环形缓冲区的 index；
recvq：因接收而陷入阻塞的协程队列；
sendq：因发送而陷入阻塞的协程队列。

1.2 waitq#

1
type waitq struct {
2
    first *sudog
3
    last  *sudog
4
}

waitq：阻塞的协程队列。

first：队列头部
last：队列尾部

1.3 sudog#

1
type sudog struct {
2
    g *g
3

4
    next *sudog
5
    prev *sudog
6
    elem unsafe.Pointer // data element (may point to stack)
7

8
    isSelect bool
9

10
    c        *hchan
11
}

sudog：用于包装协程的节点。

g：goroutine，协程；
next：队列中的下一个节点；
prev：队列中的前一个节点；
elem: 读取/写入 channel 的数据的容器;
isSelect：标识当前协程是否处在 select 多路复用的流程中；
c：标识与当前 sudog 交互的 chan.

2. 构造器函数#

几种类型的 channel

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

4
    // ...
5
    mem, overflow := math.MulUintptr(elem.size, uintptr(size))
6
    if overflow || mem > maxAlloc-hchanSize || size < 0 {
7
        panic(plainError("makechan: size out of range"))
8
    }
9

10
    var c *hchan
11
    switch {
12
    case mem == 0:
13
        // Queue or element size is zero.
14
        c = (*hchan)(mallocgc(hchanSize, nil, true))
15
        // Race detector uses this location for synchronization.
16
        c.buf = c.raceaddr()
17
    case elem.ptrdata == 0:
18
        // Elements do not contain pointers.
19
        // Allocate hchan and buf in one call.
20
        c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
21
        c.buf = add(unsafe.Pointer(c), hchanSize)
22
    default:
23
        // Elements contain pointers.
24
        c = new(hchan)
25
        c.buf = mallocgc(mem, elem, true)
26
    }
27

28
    c.elemsize = uint16(elem.size)
29
    c.elemtype = elem
30
    c.dataqsiz = uint(size)
31

32
    lockInit(&c.lock, lockRankHchan)
33

34
    return
35
}

判断申请内存空间大小是否越界，mem 大小为 element 类型大小与 element 个数相乘后得到，仅当无缓冲型 channel 时，因个数为 0 导致大小为 0；
根据类型，初始 channel，分为无缓冲型、有缓冲元素为 struct 型、有缓冲元素为 pointer 型 channel;
倘若为无缓冲型，则仅申请一个大小为默认值 96 的空间；
如若有缓冲的 struct 型，则一次性分配好 96 + mem 大小的空间，并且调整 chan 的 buf 指向 mem 的起始位置；
倘若为有缓冲的 pointer 型，则分别申请 chan 和 buf 的空间，两者无需连续；
对 channel 的其余字段进行初始化，包括元素类型大小、元素类型、容量以及锁的初始化。

3. 写流程#

3.1 两类异常情况处理#

1
func chansend1(c *hchan, elem unsafe.Pointer) {
2
    chansend(c, elem, true, getcallerpc())
3
}
4

5
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
6
    if c == nil {
7
        gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
8
        throw("unreachable")
9
    }
10

11
    lock(&c.lock)
12

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

18
    // ...
19
    // ...
20
}

对于未初始化的 chan，写入操作会引发死锁；
对于已关闭的 chan，写入操作会引发 panic。

3.2 case1：写时存在阻塞读协程#

直接写入阻塞读协程

1
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
2
    // ...
3
    lock(&c.lock)
4
    // ...
5
    if sg := c.recvq.dequeue(); sg != nil {
6
        // Found a waiting receiver. We pass the value we want to send
7
        // directly to the receiver, bypassing the channel buffer (if any).
8
        send(c, sg, ep, func() { unlock(&c.lock) }, 3)
9
        return true
10
    }
11
    // ...
12
}

加锁；
从阻塞度协程队列中取出一个 goroutine 的封装对象 sudog；
在 send 方法中，会基于 memmove 方法，直接将元素拷贝交给 sudog 对应的 goroutine；
在 send 方法中会完成解锁动作。

3.3 case2：写时无阻塞读协程但环形缓冲区仍有空间#

写入环形缓冲区

1
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
2
    // ...
3
    lock(&c.lock)
4
    // ...
5
    if c.qcount < c.dataqsiz {
6
        // Space is available in the channel buffer. Enqueue the element to send.
7
        qp := chanbuf(c, c.sendx)
8
        typedmemmove(c.elemtype, qp, ep)
9
        c.sendx++
10
        if c.sendx == c.dataqsiz {
11
            c.sendx = 0
12
        }
13
        c.qcount++
14
        unlock(&c.lock)
15
        return true
16
    }
17

18
    // ...
19
}

加锁；
将当前元素添加到环形缓冲区 sendx 对应的位置；
sendx++;
qcount++;
解锁，返回。

3.4 case3：写时无阻塞读协程且环形缓冲区无空间#

阻塞写协程

1
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
2
    // ...
3
    lock(&c.lock)
4

5
    // ...
6
    gp := getg()
7
    mysg := acquireSudog()
8
    mysg.elem = ep
9
    mysg.g = gp
10
    mysg.c = c
11
    gp.waiting = mysg
12
    c.sendq.enqueue(mysg)
13

14
    atomic.Store8(&gp.parkingOnChan, 1)
15
    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
16

17
    gp.waiting = nil
18
    closed := !mysg.success
19
    gp.param = nil
20
    mysg.c = nil
21
    releaseSudog(mysg)
22
    return true
23
}

加锁；
构造封装当前 goroutine 的 sudog 对象；
完成指针指向，建立 sudog、goroutine、channel 之间的指向关系；
把 sudog 添加到当前 channel 的阻塞写协程队列中；
park 当前协程；
倘若协程从 park 中被唤醒，则回收 sudog（sudog 能被唤醒，其对应的元素必然已经被读协程取走）；
解锁，返回。

3.5 写流程整体串联#

写流程串联

4. 读流程#

4.1 异常 case1：读空 channel#

1
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
2
    if c == nil {
3
        gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
4
        throw("unreachable")
5
    }
6
    // ...
7
}

park 挂起，引起死锁。

4.2 异常 case2：channel 已关闭且内部无元素#

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

3
    lock(&c.lock)
4

5
    if c.closed != 0 {
6
        if c.qcount == 0 {
7
            unlock(&c.lock)
8
            if ep != nil {
9
                typedmemclr(c.elemtype, ep)
10
            }
11
            return true, false
12
        }
13
        // The channel has been closed, but the channel's buffer have data.
14
    }
15

16
    // ...
17
}

直接解锁返回即可。

4.3 case3：读时有阻塞的写协程#

从阻塞写协程中读取

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

3
    lock(&c.lock)
4

5
    if sg := c.sendq.dequeue(); sg != nil {
6
        recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
7
        return true, true
8
     }
9
     // ...
10
}

加锁；
从阻塞写协程队列中获取到一个写协程；
倘若 channel 无缓冲区，则直接读取写协程元素，并唤醒写协程；
倘若 channel 有缓冲区，则读取缓冲区头部元素，并将写协程元素写入缓冲区尾部后唤醒写写成；
解锁，返回。

4.4 case4：读时无阻塞写协程且缓冲区有元素#

从环形缓冲区读取

1
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
2
    // ...
3
    lock(&c.lock)
4
    // ...
5
    if c.qcount > 0 {
6
        // Receive directly from queue
7
        qp := chanbuf(c, c.recvx)
8
        if ep != nil {
9
            typedmemmove(c.elemtype, ep, qp)
10
        }
11
        typedmemclr(c.elemtype, qp)
12
        c.recvx++
13
        if c.recvx == c.dataqsiz {
14
            c.recvx = 0
15
        }
16
        c.qcount--
17
        unlock(&c.lock)
18
        return true, true
19
    }
20
    // ...
21
}

加锁；
获取到 recvx 对应位置的元素；
recvx++
qcount--
解锁，返回。

4.5 case5：读时无阻塞写协程且缓冲区无元素#

阻塞读协程

1
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
2
    // ...
3
    lock(&c.lock)
4
    // ...
5
    gp := getg()
6
    mysg := acquireSudog()
7
    mysg.elem = ep
8
    gp.waiting = mysg
9
    mysg.g = gp
10
    mysg.c = c
11
    gp.param = nil
12
    c.recvq.enqueue(mysg)
13
    atomic.Store8(&gp.parkingOnChan, 1)
14
    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)
15

16
    gp.waiting = nil
17
    success := mysg.success
18
    gp.param = nil
19
    mysg.c = nil
20
    releaseSudog(mysg)
21
    return true, success
22
}

加锁；
构造封装当前 goroutine 的 sudog 对象；
完成指针指向，建立 sudog、goroutine、channel 之间的指向关系；
把 sudog 添加到当前 channel 的阻塞读协程队列中；
park 当前协程；
倘若协程从 park 中被唤醒，则回收 sudog（sudog 能被唤醒，其对应的元素必然已经被写入）；
解锁，返回。

4.6 读流程整体串联#

读流程串联

5. 阻塞与非阻塞模式#

在上述源码分析流程中，均是以阻塞模式为主线进行讲述，忽略非阻塞模式的有关处理逻辑. 此处阐明两个问题：

非阻塞模式下，流程逻辑有何区别？
何时会进入非阻塞模式？

5.1 非阻塞模式逻辑区别#

非阻塞模式下，读/写 channel 方法通过一个 bool 型的响应参数，用以标识是否读取/写入成功.

所有需要使得当前 goroutine 被挂起的操作，在非阻塞模式下都会返回 false；
所有是的当前 goroutine 会进入死锁的操作，在非阻塞模式下都会返回 false；
所有能立即完成读取/写入操作的条件下，非阻塞模式下会返回 true。

5.2 何时进入非阻塞模式#

默认情况下，读/写 channel 都是阻塞模式，只有在 select 语句组成的多路复用分支中，与 channel 的交互会变成非阻塞模式：

1
ch := make(chan int)
2
select{
3
case <- ch:
4
default:
5
}

5.3 代码一览#

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

5
func selectnbrecv(elem unsafe.Pointer, c *hchan) (selected, received bool) {
6
    return chanrecv(c, elem, false)
7
}

在 select 语句包裹的多路复用分支中，读和写 channel 操作会被汇编为 selectnbrecv 和 selectnbsend 方法，底层同样复用 chanrecv 和 chansend 方法，但此时由于第三个入参 block 被设置为 false，导致后续会走进非阻塞的处理分支。

6. 两种读 channel 的协议#

读取 channel 时，可以根据第二个 bool 型的返回值用以判断当前 channel 是否已处于关闭状态：

1
ch := make(chan int, 2)
2
got1 := <- ch
3
got2,ok := <- ch

实现上述功能的原因是，两种格式下，读 channel 操作会被汇编成不同的方法：

1
func chanrecv1(c *hchan, elem unsafe.Pointer) {
2
    chanrecv(c, elem, true)
3
}
4

5
//go:nosplit
6
func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {
7
    _, received = chanrecv(c, elem, true)
8
    return
9
}

7. 关闭#

关闭 channel 流程

1
func closechan(c *hchan) {
2
    if c == nil {
3
        panic(plainError("close of nil channel"))
4
    }
5

6
    lock(&c.lock)
7
    if c.closed != 0 {
8
        unlock(&c.lock)
9
        panic(plainError("close of closed channel"))
10
    }
11

12
    c.closed = 1
13

14
    var glist gList
15
    // release all readers
16
    for {
17
        sg := c.recvq.dequeue()
18
        if sg == nil {
19
            break
20
        }
21
        if sg.elem != nil {
22
            typedmemclr(c.elemtype, sg.elem)
23
            sg.elem = nil
24
        }
25
        gp := sg.g
26
        gp.param = unsafe.Pointer(sg)
27
        sg.success = false
28
        glist.push(gp)
29
    }
30

31
    // release all writers (they will panic)
32
    for {
33
        sg := c.sendq.dequeue()
34
        if sg == nil {
35
            break
36
        }
37
        sg.elem = nil
38
        gp := sg.g
39
        gp.param = unsafe.Pointer(sg)
40
        sg.success = false
41
        glist.push(gp)
42
    }
43
    unlock(&c.lock)
44

45
    // Ready all Gs now that we've dropped the channel lock.
46
    for !glist.empty() {
47
        gp := glist.pop()
48
        gp.schedlink = 0
49
        goready(gp, 3)
50
    }
51
}

关闭未初始化过的 channel 会 panic；
加锁；
重复关闭 channel 会 panic；
将阻塞读协程队列中的协程节点统一添加到 glist；
将阻塞写协程队列中的协程节点统一添加到 glist；
唤醒 glist 当中的所有协程。

1. 核心数据结构#

1.1 hchan#

1.2 waitq#

1.3 sudog#

2. 构造器函数#

3. 写流程#

3.1 两类异常情况处理#

3.2 case1：写时存在阻塞读协程#

3.3 case2：写时无阻塞读协程但环形缓冲区仍有空间#

3.4 case3：写时无阻塞读协程且环形缓冲区无空间#

3.5 写流程整体串联#

4. 读流程#

4.1 异常 case1：读空 channel#

4.2 异常 case2：channel 已关闭且内部无元素#

4.3 case3：读时有阻塞的写协程#

4.4 case4：读时无阻塞写协程且缓冲区有元素#

4.5 case5：读时无阻塞写协程且缓冲区无元素#

4.6 读流程整体串联#

5. 阻塞与非阻塞模式#

5.1 非阻塞模式逻辑区别#

5.2 何时进入非阻塞模式#

5.3 代码一览#

6. 两种读 channel 的协议#

7. 关闭#

文章分享

评论区

Channel 实现原理

1. 核心数据结构#

1.1 hchan#

1.2 waitq#

1.3 sudog#

2. 构造器函数#

3. 写流程#

3.1 两类异常情况处理#

3.2 case1：写时存在阻塞读协程#

3.3 case2：写时无阻塞读协程但环形缓冲区仍有空间#

3.4 case3：写时无阻塞读协程且环形缓冲区无空间#

3.5 写流程整体串联#

4. 读流程#

4.1 异常 case1：读空 channel#

4.2 异常 case2：channel 已关闭且内部无元素#

4.3 case3：读时有阻塞的写协程#

4.4 case4：读时无阻塞写协程且缓冲区有元素#

4.5 case5：读时无阻塞写协程且缓冲区无元素#

4.6 读流程整体串联#

5. 阻塞与非阻塞模式#

5.1 非阻塞模式逻辑区别#

5.2 何时进入非阻塞模式#

5.3 代码一览#

6. 两种读 channel 的协议#

7. 关闭#

文章分享

评论区