package main
import (
"fmt"
)
func main(){
ch := make(chan int)
ch <- 1
fmt.Println(<-ch)
}
有人问我，上面的程序能不能正常运行？
我当时就生气了，感觉自己的智商受到了极大的侮辱。他让我冷静冷静，分析分析为啥？
我话不多说，上面先来一顿操作
0x0024 00036 (demo1.go:8) PCDATA $0, $0
0x0024 00036 (demo1.go:8) LEAQ type.chan int(SB), AX
0x002b 00043 (demo1.go:8) PCDATA $2, $0
0x002b 00043 (demo1.go:8) MOVQ AX, (SP)
0x002f 00047 (demo1.go:8) MOVQ $0, 8(SP)
0x0038 00056 (demo1.go:8) CALL runtime.makechan(SB)
0x003d 00061 (demo1.go:8) PCDATA $2, $1
0x003d 00061 (demo1.go:8) MOVQ 16(SP), AX
0x0042 00066 (demo1.go:8) PCDATA $0, $1
0x0042 00066 (demo1.go:8) MOVQ AX, "".ch+56(SP)
我们发现,创建chan实际上调用的是runtime.makechan。该函数提供了两个参数，第一个参数是type.chan int类型， 第二个参数是0，也就是说我们在make的时候并未提供第二个参数。返回值是一个hchan类型指针。之后对ch进行了赋值操作。 先看一眼hchan是个啥，混个眼熟
type hchan struct {
qcount uint // 队列中的已存数据长度
dataqsiz uint // 循环队列的大小
buf unsafe.Pointer // 指向循环队列数组
elemsize uint16 // 元素大小
closed uint32 // 当前channel是否关闭
elemtype *_type // 元素类型
sendx uint // 发送索引
recvx uint // 接受索引
recvq waitq // 等待接受的接受者的链表
sendq waitq // 等待发送的发送者链表
// 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
}
看一下该函数的原型
func makechan(t *chantype, size int) *hchan {
elem := t.elem
// compiler checks this but be safe.
if elem.size >= 1<<16 {
throw("makechan: invalid channel element type")
}
if hchanSize%maxAlign != 0 || elem.align > maxAlign {
throw("makechan: bad alignment")
}
mem, overflow := math.MulUintptr(elem.size, uintptr(size))
if overflow || mem > maxAlloc-hchanSize || size < 0 {
panic(plainError("makechan: size out of range"))
}
// 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 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()
case elem.kind&kindNoPointers != 0:
// Elements do not contain pointers.
// Allocate hchan and buf in one call.
c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
c.buf = add(unsafe.Pointer(c), hchanSize)
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
}
因为我们make时，提供的size大小为0，所以mem==0,hchan的dataqsiz也是0.在对ch变量赋值时，赋得是hchan的指针。
ch <- 1 这一步是向chan中写数据，该动作调用的是runtime.chansend1
0x0047 00071 (demo1.go:11) PCDATA $2, $0
0x0047 00071 (demo1.go:11) MOVQ AX, (SP)
0x004b 00075 (demo1.go:11) PCDATA $2, $1
0x004b 00075 (demo1.go:11) LEAQ "".statictmp_0(SB), AX
0x0052 00082 (demo1.go:11) PCDATA $2, $0
0x0052 00082 (demo1.go:11) MOVQ AX, 8(SP)
0x0057 00087 (demo1.go:11) CALL runtime.chansend1(SB)
看一下"".statictmp_0这个东西是什么，
"".statictmp_0 SRODATA size=8
0x0000 01 00 00 00 00 00 00 00 ........
如果我们往chan中写的是11，这个东西的值又是什么
"".statictmp_0 SRODATA size=8
0x0000 0b 00 00 00 00 00 00 00 ........
就是一个十六进制表示的数字，继续看
func chansend1(c *hchan, elem unsafe.Pointer) {
chansend(c, elem, true, getcallerpc())
}
/*
* generic single channel send/recv
* If block is not nil,
* then the protocol will not
* sleep but return if it could
* not complete.
*
* sleep can wake up with g.param == nil
* when a channel involved in the sleep has
* been closed. it is easiest to loop and re-run
* the operation; we'll see that it's now closed.
*/
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
if c == nil {
if !block {
return false
}
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)
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
}
// 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)
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
releaseSudog(mysg)
return true
}
第一个参数是上面创建的*hchan，第二个参数是"".statictmp_0(SB)地址。因为我们创建的是不带缓存的chan,其qcount和dataqsiz都是0，在对chan加锁后lock(&c.lock),阻塞在channel上，等待接受者可以帮我们完成任务，因为当前代码无接受者goroutine,导致死锁问题。
再看来一个示例
package main
import (
"fmt"
)
func main(){
var ch chan int
ch <- 11
fmt.Println(<-ch)
}
看一下汇编后的情况
0x0024 00036 (demo1.go:10) PCDATA $0, $1
0x0024 00036 (demo1.go:10) MOVQ $0, "".ch+56(SP)
0x002d 00045 (demo1.go:12) MOVQ $0, (SP)
0x0035 00053 (demo1.go:12) PCDATA $2, $1
0x0035 00053 (demo1.go:12) LEAQ "".statictmp_0(SB), AX
0x003c 00060 (demo1.go:12) PCDATA $2, $0
0x003c 00060 (demo1.go:12) MOVQ AX, 8(SP)
0x0041 00065 (demo1.go:12) CALL `runtime.chansend1`(SB)
runtime.chansend1函数的第一个参数赋个0.
该实例运行也是异常的，异常的原因都是死锁，只不过
➜ channeltest go run demo1.go
fatal error: all goroutines are asleep - deadlock!
goroutine 1 [chan send (nil chan)]:
main.main()
demo1.go:12 +0x3a
exit status 2
这次死锁的原因是，向nil channel发送数据，也就是说，var声明的ch是一个nil chanl,在chansend方法中
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
if c == nil {
if !block {
return false
}
gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
throw("unreachable")
}
...
}
检查到c==nil,之后，会将当前goroutine,gopark掉。park掉的原因是waitReasonChanSendNilChan
const (
waitReasonChanSendNilChan // "chan send (nil chan)"
)
要想让上面的示例，成功运行，该怎么办呢？
func main(){
// ch := make(chan int)
ch:= make(chan int, 1)
// var ch chan int
ch <- 11
fmt.Println(<-ch) //11
}
加上chan的缓存。上面的示例就能正常运行了，很稳，为啥呢？仔细看看chansend
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
}
上述代码节选自runtime.chansend函数，当我们首次将元素放到chan里面的时候，因为c.dataqsiz=1,c.qcount=0. qp := chanbuf(c, c.sendx)指向c.buf的第c.sendx个槽。typedmemmove(c.elemtype, qp, ep)该函数将ep的值复制到qp.之后c.sendx++将发送索引加一。如果发送索引达到了循环队列的最大长度，则从头开始。
c的数据量自增1.解锁在chansend中加的锁，并返回true. 继续往下走
0x0065 00101 (demo1.go:13) PCDATA $0, $0
0x0065 00101 (demo1.go:13) MOVQ "".ch+56(SP), AX
0x006a 00106 (demo1.go:13) PCDATA $2, $0
0x006a 00106 (demo1.go:13) MOVQ AX, (SP)
0x006e 00110 (demo1.go:13) PCDATA $2, $1
0x006e 00110 (demo1.go:13) LEAQ ""..autotmp_2+48(SP), AX
0x0073 00115 (demo1.go:13) PCDATA $2, $0
0x0073 00115 (demo1.go:13) MOVQ AX, 8(SP)
0x0078 00120 (demo1.go:13) CALL runtime.chanrecv1(SB)
0x007d 00125 (demo1.go:13) MOVQ ""..autotmp_2+48(SP), AX
下面的接受数据<-ch,涉及的函数调用是runtime.chanrecv1,其参数为第一个参数是ch,第二个参数是需要存放读取出来的数据的寄存器地址""..autotmp_2+48(SP)。没有返回值。因为我们并没有申明变量来接受数据。
// entry points for <- c from compiled code
//go:nosplit
func chanrecv1(c *hchan, elem unsafe.Pointer) {
chanrecv(c, elem, true)
}
// chanrecv 从channel中接受数据并将接受的数据写入到ep.
// ep可能为nil,这种情况下接受的数据将被忽略。
// 如果block == false并且没有元素可以使用，则返回(false, false)
// 否则，如果c已经关闭，零值的*ep并返回(true, false)
// 否则，使用元素填充ep并返回(true, true)
// 非空的ep必须指向堆或者调用者栈
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")
}
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)
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)
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
}
因为c.qcount>0,直接从队列中接受值，整个流程结束，其实上述的示例是比较简单的。更复杂一些的应用场景有哪些呢？
- 多个goroutine的同步控制
- 结合select使用
- 任务队列
关于第一个应用场景示例
import (
"time"
)
func worker(done chan bool) {
time.Sleep(time.Second)
done <- true
}
func main() {
done := make(chan bool, 1)
go worker(done)
<-done
}
只有在工作goroutine完成了任务处理之后，整个程序才能顺利结束.
第二个应用场景示例
package main
import (
"fmt"
)
func fibonacci(c, quit chan int) {
x, y := 0, 1
for {
select {
case c <- x:
x, y = y, x+y
case <-quit:
fmt.Println("fibonacci quit")
return
}
}
}
func main() {
c := make(chan int)
quit := make(chan int)
go func() {
for i := 0; i < 10; i++ {
fmt.Println(<-c)
}
quit <- 0
}()
fibonacci(c, quit)
}
第三个应用场景示例
package main
import (
"fmt"
)
func main() {
done := make(chan int, 10) // 带 10 个缓存
// 开N个后台打印线程
for i := 0; i < cap(done); i++ {
go func(){
fmt.Println("你好, 世界")
done <- 1
}()
}
// 等待N个后台线程完成
for i := 0; i < cap(done); i++ {
<-done
}
}
当然了，上述的示例，有更方便的替换方案实现：
package main
import (
"fmt"
"sync"
)
func main() {
var wg sync.WaitGroup
// 开N个后台打印线程
for i := 0; i < 10; i++ {
wg.Add(1)
go func() {
fmt.Println("你好, 世界")
wg.Done()
}()
}
// 等待N个后台线程完成
wg.Wait()
}
有关关闭channel的原则：
- 一个 sender，多个 receiver，由 sender 来关闭 channel，通知数据已发送完毕。
- 一旦 sender 有多个，可能就无法判断数据是否完毕了。这时候可以借助外部额外 channel 来做信号广播。这种做法类似于 done channel，或者 stop channel。
- 如果确定不会有 goroutine 在通信过程中被阻塞，也可以不关闭 channel，等待 GC 对其进行回收。
之所以出现类限制，主要是因为channel类型没办法多次关闭，在关闭的chan上调用close将panic,向close的chan发送数据也将panic.从close的chan上读取数据，将获取chan元素类型的零值，此种场景下要做第二个返回值的判断，以防止无法确切的知道到底是chan关闭了，还是从chan中读取的数就是零值。
至于有关chan作为参数时，是值传递还是引用传递，我们可以看一个小示例：
package main
import (
"fmt"
)
func worker(stop chan bool) {
for i:=0;i<10;i++ {
fmt.Println("干活....")
}
stop <- true
}
func main() {
stop := make(chan bool)
go worker(stop)
<- stop
}
检点看一下汇编
0x001d 00029 (main.go:17) PCDATA $0, $0
0x001d 00029 (main.go:17) LEAQ type.chan bool(SB), AX
0x0024 00036 (main.go:17) PCDATA $2, $0
0x0024 00036 (main.go:17) MOVQ AX, (SP)
0x0028 00040 (main.go:17) MOVQ $0, 8(SP)
0x0031 00049 (main.go:17) CALL runtime.makechan(SB)
0x0036 00054 (main.go:17) PCDATA $2, $1
0x0036 00054 (main.go:17) MOVQ 16(SP), AX
0x003b 00059 (main.go:17) PCDATA $0, $1
0x003b 00059 (main.go:17) MOVQ AX, "".stop+24(SP)
0x0040 00064 (main.go:18) MOVL $8, (SP)
0x0047 00071 (main.go:18) PCDATA $2, $2
0x0047 00071 (main.go:18) LEAQ "".worker·f(SB), CX
0x004e 00078 (main.go:18) PCDATA $2, $1
0x004e 00078 (main.go:18) MOVQ CX, 8(SP)
0x0053 00083 (main.go:18) PCDATA $2, $0
0x0053 00083 (main.go:18) MOVQ AX, 16(SP)
0x0058 00088 (main.go:18) CALL runtime.newproc(SB)
0x005d 00093 (main.go:19) PCDATA $2, $1
0x005d 00093 (main.go:19) PCDATA $0, $0
0x005d 00093 (main.go:19) MOVQ "".stop+24(SP), AX
0x0062 00098 (main.go:19) PCDATA $2, $0
0x0062 00098 (main.go:19) MOVQ AX, (SP)
0x0066 00102 (main.go:19) MOVQ $0, 8(SP)
0x006f 00111 (main.go:19) CALL runtime.chanrecv1(SB)
0x0074 00116 (main.go:20) MOVQ 32(SP), BP
我们在新起一个goroutine进行执行业务处理是，使用runtime.newproc该函数有两个参数，第一个参数是siz=8,第二个参数是一个函数地址
// 创建一个新的g运行fn，带有size字节的参数。
// 将它放在g的等待队列中。
// 编译器将go语句转化成对该方法的调用。
//
// Cannot split the stack because it assumes that the arguments
// are available sequentially after &fn; they would not be
// copied if a stack split occurred.
//go:nosplit
func newproc(siz int32, fn *funcval) {
argp := add(unsafe.Pointer(&fn), sys.PtrSize)
gp := getg()
pc := getcallerpc()
systemstack(func() {
newproc1(fn, (*uint8)(argp), siz, gp, pc)
})
}
本例中第三个参数是函数的参数，也就是我们makechan出来的*hchan,这也就是为何，我们可以在业务函数中可以修改chan的原因。
本系列文章：
有任何问题，欢迎留言