go-channel基础

结构

首先看一下channel的结构定义

type hchan struct {
	qcount   uint           // 队列长度，缓冲区中入队的数据长度
	dataqsiz uint           // 环形队列总长度，即缓冲区总长度
	buf      unsafe.Pointer // 队列中的数据
	elemsize uint16			//  chan中元素类型长度(unsafe.Sizeof(type))
	closed   uint32			//  是否关闭，0，1
	elemtype *_type // chan中元素类型
	sendx    uint   // 发送元素的索引 
	recvx    uint   // 接受元素的索引(配合sendx和buf，可以得到缓冲区真正的数据)
	recvq    waitq  // 等待接收数据的g队列
	sendq    waitq  // 等待发送数据的g队列

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

对于一个有缓冲的通道，有如下规则：

1. 发送操作会使通道复制被发送的元素。若因通道缓冲区已满而无法复制，则会阻塞发送操作的goroutine。复制的目的地有两种：①当通道缓冲区无数据且有接收方在等待接收数据时，数据将会被直接复制到接收方持有的内存地址；②否则，复制到通道的buf中
2. 接收操作会使通道给出一个已发送给它的元素值的副本，如果通道无数据，则会阻塞接收操作的goroutine
3. 对于同一个元素值来说，把它发送给某个通道的操作，一定会在从该通道接受它的操作完成之前完成

注意：

1. channel变量是引用类型，因此它的初始值为nil，向一个nil的channel发送或接收数据将会导致永久阻塞
2. 发送方向通道发送的数据会被复制，至少1次，最多两次。①1次的情形：A 已有接收方阻塞且通道无缓冲数据;B 无缓冲的channel ②2次情形：无缓冲channel不会复制2次，有缓冲channel只要缓冲区有数据，则都会发生两次

happen before

发送操作与接收操作的happen before

1. 无缓冲channel：接收操作完成都在发送操作完成后
2. 有缓冲channel：缓冲长度为k，第c个数据接收完成发送在第k+c个数据发送之前

发送操作前后代码的happen before

graph TD;
subgraph 协程 g1
1-->idch1(channel操作)
idch1-->A
A-->X
end
subgraph 协程 g2
2-->idch2(channel操作)
idch2-->B
B-->Y
end

解释：
如上图两个协程 g1 和 g2，channel操作指的是读或者写操作，这里两个协程一读一写。happen-before会使代码执行有以下顺序：

1. 1 -> B -> Y
2. 2 -> A -> X
3. 1 -> A -> X
4. 2 -> B -> Y
   其他顺序无法保证，比如： A与B、Y，B与A、X，1与2，整体顺序如下图：

   graph TD;
   subgraph 协程 g1
   1-->idch1(channel操作)
   idch1-->A
   A-->X
   end
   subgraph 协程 g2
   2-->idch2(channel操作)
   idch2-->B
   B-->Y
   end
   1-->idch2
   idch2-->A
   2-->idch1
   idch1-->B

数据读写

向通道写数据

1. 先判断是否有读channel的G阻塞到这里，如果有，则直接将数据复制到等待读数据的G的elem域，并将其标记为_Grunnable，结束

2. 否则，判断channel中已有数据是否小于缓冲大小，如果小于，则将数据复制到channel的循环队列中，结束

3. 否则，判断如果是非阻塞，则直接返回false，表示写数据失败

4. 否则，获取当前g，封装sudog，并把sudog放到channel的sendq域，然后将其标记为_Gwating，等有数据被读取，这里会被唤醒，并执行后续代码，最后返回true

   func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
   	
   	// ...
   	lock(&c.lock)
   	// ...
   	if sg := c.recvq.dequeue(); sg != nil {
   		// 1  
   		send(c, sg, ep, func() { unlock(&c.lock) }, 3)
   		return true
   	}
   	// 2
   	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
   	}
   	// 3
   	if !block {
   		unlock(&c.lock)
   		return false
   	}
       // 4  // 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
   	}
   	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)
   	
   	// ...
   	releaseSudog(mysg)
   	return true
   }

   从通道读数据

5. 如果channel已关闭，且缓冲区无数据，则直接返回

6. 否则，判断通道是否有写数据的G阻塞，如果有，则调用recv()函数并直接返回
   2.1 recv() 如果缓冲区无数据，则直接将数据从sudog的G.elem中copy到ep
   2.2 recv() 否则，将缓冲区第一个数据copy到ep中，并将阻塞的G.elem数据copy到缓冲区队列
   2.3 recv() 将sudog.g的状态从_Gwaiting转为_Grunnable

7. 否则，若缓冲区队列已有数据，则直接从缓冲区复制，直接返回

8. 否则，若非阻塞，直接返回失败

9. 否则，获取当前g，初始化sudog，将sudog放至channel.recvq队列中，并将状态转至_Gwaiting，等待唤醒

10. 唤醒之后，执行后续逻辑，返回接收数据成功

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
	
	// ...
	lock(&c.lock)

	// 1
	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
	}
	// 2
	if sg := c.sendq.dequeue(); sg != nil {
		recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
		return true, true
	}
	// 3
	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
	}
	// 4
	if !block {
		unlock(&c.lock)
		return false, false
	}

	// 5
	// no sender available: block on this channel.
	gp := getg()
	mysg := acquireSudog()
	mysg.releasetime = 0
	if t0 != 0 {
		mysg.releasetime = -1
	}
	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)

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

select

// runtime/select.go

// A runtimeSelect is a single case passed to rselect.
// This must match ../reflect/value.go:/runtimeSelect
type runtimeSelect struct {
	dir selectDir      // selectSend、selectRecv、selectDefault 分别代表case里的事件 发送、接收、default
	typ unsafe.Pointer // channel type (not used here)
	ch  *hchan         // channel
	val unsafe.Pointer // ptr to data (SendDir) or ptr to receive buffer (RecvDir)
}


func reflect_rselect(cases []runtimeSelect) (int, bool) {
	if len(cases) == 0 {
		block()
	}
	sel := make([]scase, len(cases))
	order := make([]uint16, 2*len(cases))
	for i := range cases {
		rc := &cases[i]
		switch rc.dir {
		case selectDefault:
			sel[i] = scase{kind: caseDefault}
		case selectSend:
			sel[i] = scase{kind: caseSend, c: rc.ch, elem: rc.val}
		case selectRecv:
			sel[i] = scase{kind: caseRecv, c: rc.ch, elem: rc.val}
		}
		if raceenabled || msanenabled {
			selectsetpc(&sel[i])
		}
	}

	return selectgo(&sel[0], &order[0], len(cases))
}

// Select case descriptor.
// Known to compiler.
// Changes here must also be made in src/cmd/internal/gc/select.go's scasetype.
type scase struct {
	c           *hchan         // chan
	elem        unsafe.Pointer // data element
	kind        uint16
	pc          uintptr // race pc (for race detector / msan)
	releasetime int64
}

如上代码，初始时，会遍历select里的所有case，并将这些case放入scase数组，scase数组描述每个case的事件，最后将其作为参数调用selectgo方法

// selectgo 实现了select的功能.
//
// cas0 points to an array of type [ncases]scase, and order0 points to
// an array of type [2*ncases]uint16. Both reside on the goroutine's
// stack (regardless of any escaping in selectgo).
//
// selectgo 返回选中的nscase在数组cas0中的下标，它也是select调用语句中对应位置的序号
// 如果选中的scase事件是channel接收事件，第二个参数表示是否有值发送到接收方
func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool) {
	if debugSelect {
		print("select: cas0=", cas0, "\n")
	}

	cas1 := (*[1 << 16]scase)(unsafe.Pointer(cas0))
	order1 := (*[1 << 17]uint16)(unsafe.Pointer(order0))

	scases := cas1[:ncases:ncases]
	pollorder := order1[:ncases:ncases]
	lockorder := order1[ncases:][:ncases:ncases]

	// Replace send/receive cases involving nil channels with
	// caseNil so logic below can assume non-nil channel.
	for i := range scases {
		cas := &scases[i]
		if cas.c == nil && cas.kind != caseDefault {
			*cas = scase{}
		}
	}

	var t0 int64
	if blockprofilerate > 0 {
		t0 = cputicks()
		for i := 0; i < ncases; i++ {
			scases[i].releasetime = -1
		}
	}

	// The compiler rewrites selects that statically have
	// only 0 or 1 cases plus default into simpler constructs.
	// The only way we can end up with such small sel.ncase
	// values here is for a larger select in which most channels
	// have been nilled out. The general code handles those
	// cases correctly, and they are rare enough not to bother
	// optimizing (and needing to test).

	// generate permuted order
	for i := 1; i < ncases; i++ {
		j := fastrandn(uint32(i + 1))
		pollorder[i] = pollorder[j]
		pollorder[j] = uint16(i)
	}

	// sort the cases by Hchan address to get the locking order.
	// simple heap sort, to guarantee n log n time and constant stack footprint.
	for i := 0; i < ncases; i++ {
		j := i
		// Start with the pollorder to permute cases on the same channel.
		c := scases[pollorder[i]].c
		for j > 0 && scases[lockorder[(j-1)/2]].c.sortkey() < c.sortkey() {
			k := (j - 1) / 2
			lockorder[j] = lockorder[k]
			j = k
		}
		lockorder[j] = pollorder[i]
	}
	for i := ncases - 1; i >= 0; i-- {
		o := lockorder[i]
		c := scases[o].c
		lockorder[i] = lockorder[0]
		j := 0
		for {
			k := j*2 + 1
			if k >= i {
				break
			}
			if k+1 < i && scases[lockorder[k]].c.sortkey() < scases[lockorder[k+1]].c.sortkey() {
				k++
			}
			if c.sortkey() < scases[lockorder[k]].c.sortkey() {
				lockorder[j] = lockorder[k]
				j = k
				continue
			}
			break
		}
		lockorder[j] = o
	}

	if debugSelect {
		for i := 0; i+1 < ncases; i++ {
			if scases[lockorder[i]].c.sortkey() > scases[lockorder[i+1]].c.sortkey() {
				print("i=", i, " x=", lockorder[i], " y=", lockorder[i+1], "\n")
				throw("select: broken sort")
			}
		}
	}

	// lock all the channels involved in the select
	sellock(scases, lockorder)

	var (
		gp     *g
		sg     *sudog
		c      *hchan
		k      *scase
		sglist *sudog
		sgnext *sudog
		qp     unsafe.Pointer
		nextp  **sudog
	)

loop:
	// pass 1 - look for something already waiting
	var dfli int
	var dfl *scase
	var casi int
	var cas *scase
	var recvOK bool
	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
		}
	}

	if dfl != nil {
		selunlock(scases, lockorder)
		casi = dfli
		cas = dfl
		goto retc
	}

	// pass 2 - enqueue on all chans
	gp = getg()
	if gp.waiting != nil {
		throw("gp.waiting != nil")
	}
	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
		// No stack splits between assigning elem and enqueuing
		// sg on gp.waiting where copystack can find it.
		sg.elem = cas.elem
		sg.releasetime = 0
		if t0 != 0 {
			sg.releasetime = -1
		}
		sg.c = c
		// Construct waiting list in lock order.
		*nextp = sg
		nextp = &sg.waitlink

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

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

	// wait for someone to wake us up
	gp.param = nil
	gopark(selparkcommit, nil, waitReasonSelect, traceEvGoBlockSelect, 1)

	sellock(scases, lockorder)

	gp.selectDone = 0
	sg = (*sudog)(gp.param)
	gp.param = nil

	// pass 3 - dequeue from unsuccessful chans
	// otherwise they stack up on quiet channels
	// record the successful case, if any.
	// We singly-linked up the SudoGs in lock order.
	casi = -1
	cas = nil
	sglist = gp.waiting
	// Clear all elem before unlinking from gp.waiting.
	for sg1 := gp.waiting; sg1 != nil; sg1 = sg1.waitlink {
		sg1.isSelect = false
		sg1.elem = nil
		sg1.c = nil
	}
	gp.waiting = nil

	for _, casei := range lockorder {
		k = &scases[casei]
		if k.kind == caseNil {
			continue
		}
		if sglist.releasetime > 0 {
			k.releasetime = sglist.releasetime
		}
		if sg == sglist {
			// sg has already been dequeued by the G that woke us up.
			casi = int(casei)
			cas = k
		} else {
			c = k.c
			if k.kind == caseSend {
				c.sendq.dequeueSudoG(sglist)
			} else {
				c.recvq.dequeueSudoG(sglist)
			}
		}
		sgnext = sglist.waitlink
		sglist.waitlink = nil
		releaseSudog(sglist)
		sglist = sgnext
	}

	if cas == nil {
		// We can wake up with gp.param == nil (so cas == nil)
		// when a channel involved in the select has been closed.
		// It is easiest to loop and re-run the operation;
		// we'll see that it's now closed.
		// Maybe some day we can signal the close explicitly,
		// but we'd have to distinguish close-on-reader from close-on-writer.
		// It's easiest not to duplicate the code and just recheck above.
		// We know that something closed, and things never un-close,
		// so we won't block again.
		goto loop
	}

	c = cas.c

	if debugSelect {
		print("wait-return: cas0=", cas0, " c=", c, " cas=", cas, " kind=", cas.kind, "\n")
	}

	if cas.kind == caseRecv {
		recvOK = true
	}

	if raceenabled {
		if cas.kind == caseRecv && cas.elem != nil {
			raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc)
		} else if cas.kind == caseSend {
			raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc)
		}
	}
	if msanenabled {
		if cas.kind == caseRecv && cas.elem != nil {
			msanwrite(cas.elem, c.elemtype.size)
		} else if cas.kind == caseSend {
			msanread(cas.elem, c.elemtype.size)
		}
	}

	selunlock(scases, lockorder)
	goto retc

bufrecv:
	// can receive from buffer
	if raceenabled {
		if cas.elem != nil {
			raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc)
		}
		raceacquire(chanbuf(c, c.recvx))
		racerelease(chanbuf(c, c.recvx))
	}
	if msanenabled && cas.elem != nil {
		msanwrite(cas.elem, c.elemtype.size)
	}
	recvOK = true
	qp = chanbuf(c, c.recvx)
	if cas.elem != nil {
		typedmemmove(c.elemtype, cas.elem, qp)
	}
	typedmemclr(c.elemtype, qp)
	c.recvx++
	if c.recvx == c.dataqsiz {
		c.recvx = 0
	}
	c.qcount--
	selunlock(scases, lockorder)
	goto retc

bufsend:
	// can send to buffer
	if raceenabled {
		raceacquire(chanbuf(c, c.sendx))
		racerelease(chanbuf(c, c.sendx))
		raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc)
	}
	if msanenabled {
		msanread(cas.elem, c.elemtype.size)
	}
	typedmemmove(c.elemtype, chanbuf(c, c.sendx), cas.elem)
	c.sendx++
	if c.sendx == c.dataqsiz {
		c.sendx = 0
	}
	c.qcount++
	selunlock(scases, lockorder)
	goto retc

recv:
	// can receive from sleeping sender (sg)
	recv(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2)
	if debugSelect {
		print("syncrecv: cas0=", cas0, " c=", c, "\n")
	}
	recvOK = true
	goto retc

rclose:
	// read at end of closed channel
	selunlock(scases, lockorder)
	recvOK = false
	if cas.elem != nil {
		typedmemclr(c.elemtype, cas.elem)
	}
	if raceenabled {
		raceacquire(c.raceaddr())
	}
	goto retc

send:
	// can send to a sleeping receiver (sg)
	if raceenabled {
		raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc)
	}
	if msanenabled {
		msanread(cas.elem, c.elemtype.size)
	}
	send(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2)
	if debugSelect {
		print("syncsend: cas0=", cas0, " c=", c, "\n")
	}
	goto retc

retc:
	if cas.releasetime > 0 {
		blockevent(cas.releasetime-t0, 1)
	}
	return casi, recvOK

sclose:
	// send on closed channel
	selunlock(scases, lockorder)
	panic(plainError("send on closed channel"))
}