【go源码分析】go源码之chan源码分析

柔情只为你懂 2022-04-17 01:52 359阅读 0赞

源码路径runtime/chan.go

## CSP 模型 ##

CSP 模型全称为 *communicating sequential processes*，CSP 模型由并发执行实体(进程，线程或协程)，和消息通道组成，实体之间通过消息通道发送消息

## channel读写 ##

ch := make(chan int)
    
    // write
    ch <- x
    
    // read
    x <- ch
    
    // another read
    x = <- ch

注意： channel 初始化后才能进行读写操作，否则阻塞

## 关闭 channel ##

ch := make(chan int)
    
    close(ch)

*  关闭未初始化 channel 会产生 panic
 *  重复关闭 channel 产生 panic
 *  向已关闭 channel 中发送会产生 panic
 *  从已关闭的 channel 读取不会 panic，能读出 channel 未被读取的，若消息均已读出则会读到类型的零值。从已关闭的 channel 中读取不会阻塞，会返回一个为 false 的 ok-idiom（判断 channel 是否关闭）
 *  关闭 channel 产生广播机制，所有读取channel的 goroutine 都会收到消息

# 结构体 #

###   hchan结构体 ###

Channel 实际上是个环形队列。

recvq和sendq分别用来保存对应的阻塞队列

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
    }

# 1. makechan函数 #

在堆上分配地址空间

###   1.1 验证机制 ###

*  数据类型大小，大于1<<16时异常
 *  内存对齐（降低寻址次数），大于最大的内存8字节数时异常
 *  传入的size大小，大于堆可分配的最大内存时异常

// 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")
    	}
    
    	if size < 0 || uintptr(size) > maxSliceCap(elem.size) || uintptr(size)*elem.size > maxAlloc-hchanSize {
    		panic(plainError("makechan: size out of range"))
    	}

###   1.2 分配地址空间 ###

*  size为0，分配hchan结构体空间，即无缓存channel
 *  不包括指针，分配连续地址空间，包括hchan结构体+数据
 *  包括指针，buf单独分配数据地址空间

// 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 size == 0 || elem.size == 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+uintptr(size)*elem.size, nil, true))
    		c.buf = add(unsafe.Pointer(c), hchanSize)
    	default:
    		// Elements contain pointers.
    		c = new(hchan)
    		c.buf = mallocgc(uintptr(size)*elem.size, elem, true)
    	}
    
    	c.elemsize = uint16(elem.size)
    	c.elemtype = elem
    	c.dataqsiz = uint(size)

# 2. chansend函数 #

操作如： ch <- x

###   2.1 未创始化为nil，向其中发送数据将会阻塞 ###

if c == nil {
    		if !block {
    			return false
    		}
    		gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
    		throw("unreachable")
    	}

###   2.2 这是队列满的情况 ###

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

###   2.3 已经关闭channel，向其中发送将panic ###

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

###   2.4 从当前channel等待队列取出gorouine然后调用 send（第3节讲解）方法 ###

goroutine 阻塞在 channel 上，直接将数据发送给该 goroutine，从当前 channel 的等待队列中取出等待的 goroutine，然后调用 send。goready 负责唤醒 goroutine，传说中的站茅坑不使用，被抢

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
    	}

###   2.5 缓存队列还有坑位的情况 ###

sendx代表队列index，直接加入

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
    	}

###   2.6 缓存队列么有坑位的情况 ###

那加入send队列

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

# 3. send函数 #

发送给goroutine，调用goready唤醒goroutine

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

# 4. closechan函数 #

###   4.1 close未初始化channel产生panic，关闭已经closed的channel将产生panic ###

if c == nil {
    		panic(plainError("close of nil channel"))
    	}
    
    	lock(&c.lock)
    	if c.closed != 0 {
    		unlock(&c.lock)
    		panic(plainError("close of closed channel"))
    	}

###   4.2 释放所有读取 ###

// release all readers
    	for {
    		sg := c.recvq.dequeue()
    		if sg == nil {
    			break
    		}
    		if sg.elem != nil {
    			typedmemclr(c.elemtype, sg.elem)
    			sg.elem = nil
    		}
    		if sg.releasetime != 0 {
    			sg.releasetime = cputicks()
    		}
    		gp := sg.g
    		gp.param = nil
    		if raceenabled {
    			raceacquireg(gp, c.raceaddr())
    		}
    		gp.schedlink.set(glist)
    		glist = gp
    	}

###   4.3 释放所有写 ###

// release all writers (they will panic)
    	for {
    		sg := c.sendq.dequeue()
    		if sg == nil {
    			break
    		}
    		sg.elem = nil
    		if sg.releasetime != 0 {
    			sg.releasetime = cputicks()
    		}
    		gp := sg.g
    		gp.param = nil
    		if raceenabled {
    			raceacquireg(gp, c.raceaddr())
    		}
    		gp.schedlink.set(glist)
    		glist = gp
    	}

4.4

// Ready all Gs now that we've dropped the channel lock.
    	for glist != nil {
    		gp := glist
    		glist = glist.schedlink.ptr()
    		gp.schedlink = 0
    		goready(gp, 3)
    	}

# 5. chanrecv函数  #

x :=  <-ch

###   5.1 未初始化channel，阻塞 ###

if c == nil {
    		if !block {
    			return
    		}
    		gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
    		throw("unreachable")
    	}

###   5.2 从已经关闭的且书记已经都读出，则返回数据类型的0值 ###

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
    	}

###   5.3 当前有发送 goroutine 阻塞在 channel 上，说明buf 已经满员 ###

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
    	}