Random walk to my blog

my blog for sharing my knowledge,experience and viewpoint

0%

Golang channel 源码分析

Golang中,在接受和发送数据的同时,channel决定一个Goroutine是执行还是阻塞。关于Golang的调度器,可以看这里

channel结构

Golang中,channel结构体是用来进行在Goroutine中进行信息传递的结构体。

1
ch := make(chan int, 8)

运行时,它是这样的:

hchan 结构体

当使用make(chan int,8)时,channel是从hchan创建的。hchan代码.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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
}

type waitq struct {
first *sudog
last *sudog
}
  • dataqsize是buffer的大小,也就是make(chan T, N)中的N。
  • elemsize是channel中单个元素的大小
  • buf是带缓冲的channel(buffered channel)中用来保存数据的循环队列
  • closed表示channel是否关闭。0->打开,1->关闭
  • sendxrecvx表示循环队列接受和发送数据的下标
  • recvqsendq是保存阻塞的Goroutine的等待队列,recvq保存读取数据而阻塞的Goroutine,sendq保存写入数据而阻塞的Goroutine
  • lock是在每个读写操作对channel的锁

sudog 结构体

sudog表示Goroutine

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
type sudog struct {
// The following fields are protected by the hchan.lock of the
// channel this sudog is blocking on. shrinkstack depends on
// this for sudogs involved in channel ops.

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 // data element (may point to stack)

// 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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
func goRoutineA(a <-chan int) {
val := <-a
fmt.Println("goRoutineA received the data", val)
}

func goRoutineB(b <-chan int) {
val := <-b
fmt.Println("goRoutineB received the data", val)
}

func TestGoRoutine(t *testing.T) {
ch := make(chan int)
go goRoutineA(ch)
go goRoutineB(ch)
ch <- 3
time.Sleep(time.Second * 1)
}

运行程序时,在ch <- 3加个断点,看一下这时channel的内部结构:
channeld send
可以看到,recvq保存着因为读取数据而阻塞的两个Goroutine

recvqsendq是一个链表:
recvq_structure

channel的发送数据操作

  1. nil channel发送

    1
    2
    3
    4
    5
    if c == nil {
    ...
    gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
    throw("unreachable")
    }

    向nil channel发送数据,当前的Goroutine会暂停操作。

  2. closed channel 发送

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

    如果channel已经关闭,还要发送数据,会引发panic

  3. 在channel上阻塞的Goroutine,会直接向其发送数据

    1
    2
    3
    4
    5
    6
    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上有等待接收数据的Goroutine,当前的写操作会直接向其发送数据。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    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)
    }

    这里,通过调用goready(gp, skip+1),使得阻塞的Goroutine变的可执行(runnable)。

  4. 带缓冲的channel,如果还有hchan.buf有空间,将数据放在缓冲区

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17

    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
    }

    chanbuf(c, c.sendx)会操作对应的内存区域

  5. hchan.buf已经满了

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    // Block on the channel. Some receiver will complete our operation for us.
    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
    c.sendq.enqueue(mysg)
    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)

    在当前的栈上创建一个GoroutineacquireSudog将当前的goroutine设置成park状态,然后将它放在channel的sendq队列。

发送操作总结

  1. lock锁住整个channel结构体
  2. 确定写操作。尝试从recvq中取出一个正在等待的goroutine,然后直接将数据写在里面。
  3. 如果recvq为空,确定buffer是否有空间。如果有,将数据放在缓冲区。
  4. 如果缓冲区没有空间,将数据保存在当前执行的goroutine,然后将这个goroutine放在sendq队列中,然后这个goroutine的执行暂停。

在第4点中,缓冲区不足的channel,或者没有缓存的channel,数据会保存在sudog结构体的elem

channel的读取数据操作

与channel的发送数据操作类似

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
...
if c == nil {
if !block {
return
}
gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
throw("unreachable")
}

...

lock(&c.lock)

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
}

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
}

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
}

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

...
// 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)
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)

// someone woke us up
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
mysg.c = nil
releaseSudog(mysg)
return true, !closed
}

select操作

select的源代码在runtime/select.go

select的使用例子:

1
2
3
4
5
6
7
8
9
10
11
12
func TestSelect(t *testing.T) {
cInt := make(chan int, 5)
cString := make(chan string, 5)
select {
case msg := <-cInt:
fmt.Println("receive msg", msg)
case msg := <-cString:
fmt.Println("receive msg", msg)
default:
fmt.Println("no msg received")
}
}
  1. 操作是互斥的,使所以需要获得参与select的channel的锁,先按hchane的地址排序,然后顺序获得锁,所以并不是同时获得所有参与select的channel的锁。

    1
    sellock(scases, lockorder)

    scases数组上的每一个scase结构体,包含着当前的操作类型,和正在操作的channel。

    1
    2
    3
    4
    5
    6
    7
    type scase struct {
    c *hchan // chan
    elem unsafe.Pointer // data element
    kind uint16
    pc uintptr // race pc (for race detector / msan)
    releasetime int64
    }

    kind是当前操作类型的case,可以是CaseRecv, CaseSend 或者 CaseDefault.

  2. select的选取顺序是,以伪随机数的方法,将参与select的channel顺序打乱。所以select的顺序和程序声明的顺序是不一样的。

    1
    2
    3
    4
    5
    6
    // generate permuted order
    for i := 1; i < ncases; i++ {
    j := fastrandn(uint32(i + 1))
    pollorder[i] = pollorder[j]
    pollorder[j] = uint16(i)
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
for i := 0; i < ncases; i++ {
casi = int(pollorder[i])
cas = &scases[casi]
c = cas.c

switch cas.kind {
case caseNil:
continue

case caseRecv:
sg = c.sendq.dequeue()
if sg != nil {
goto recv
}
if c.qcount > 0 {
goto bufrecv
}
if c.closed != 0 {
goto rclose
}

case caseSend:
if raceenabled {
racereadpc(c.raceaddr(), cas.pc, chansendpc)
}
if c.closed != 0 {
goto sclose
}
sg = c.recvq.dequeue()
if sg != nil {
goto send
}
if c.qcount < c.dataqsiz {
goto bufsend
}

case caseDefault:
dfli = casi
dfl = cas
}
}
  1. 只要有不阻塞的channelselect就会返回,不会等待每一个参与select的channel都才准备好。

  2. 如果当前没有channel回应,并且没有default语句,当前的g就会对应的等待队列悬停。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    // pass 2 - enqueue on all chans
    gp = getg()
    ...
    nextp = &gp.waiting
    for _, casei := range lockorder {
    casi = int(casei)
    cas = &scases[casi]
    if cas.kind == caseNil {
    continue
    }
    c = cas.c
    sg := acquireSudog()
    sg.g = gp
    sg.isSelect = true
    ...

    switch cas.kind {
    case caseRecv:
    c.recvq.enqueue(sg)

    case caseSend:
    c.sendq.enqueue(sg)
    }
    }

    sg.isSelect 表示goroutineselect语句中

我的公众号:lyp分享的地方

我的知乎专栏

我的博客