GoLang channel底层代码分析详解_Golang_

type hchan struct { qcount uint // buf当前元素的数量 dataqsiz uint // buf的容量 buf unsafe.Pointer // channel缓冲区，一个循环数组 elemsize uint16 // 元素大小 closed uint32 // channel关闭标记 elemtype *_type // element type sendx uint // 当下一次发送数据到channel时，数据存放到buf中的哪个index recvx uint // 当下一次从channel接收数据时，从buf的哪个index获取数据 recvq waitq // 等待接收数据的goroutine列表，双向链表 sendq waitq // 等待发送数据的goroutine列表，双向链表 lock mutex // 互斥锁，发送和接收操作前需要获取的锁，所以channel的发送和接收操作是互斥的 }

type waitq struct { first *sudog last *sudog } 

// sudog表示等待队列中的一个g，例如在一个channel中的发送/接收。 // sudog是必要的，因为g和同步对象的关系是多对多的，一个g可以在多个等待队列中，因此一个g会有很多个sudog， // 很多g可能在等待着同一个同步对象，因此一个对象可能有多个sudog。 // sudog是从一个特殊的池中分配的，使用acquireSudog和releaseSudog分配和释放它们。 type sudog struct { // 以下字段受此sudog阻塞的channel的hchan.lock保护 g *g next *sudog prev *sudog elem unsafe.Pointer // data element (may point to stack) // 以下字段永远不会被同时访问 // 对于channel，waitlink只能被g访问 // 对于信号量，所有字段（包括上述字段）只有在持有semaRoot锁时才能访问。 acquiretime int64 releasetime int64 ticket uint32 // isSelect表示g正在参与选择，因此g.selectDone必须经过CAS处理，才能被唤醒 isSelect bool // success表示通过channel c的通信是否成功。 // 如果goroutine因为通过channel c传递了一个值而被唤醒，则为true // 如果因为c被关闭而唤醒，则为false success bool parent *sudog // semaRoot binary tree waitlink *sudog // g.waiting list or semaRoot waittail *sudog // semaRoot c *hchan // channel }

// 无缓冲channel ch := make(chan int) // 缓冲大小为5的channel ch2 := make(chan int, 5) 

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: // 队列或元素大小为0，即无缓冲channel c = (*hchan)(mallocgc(hchanSize, nil, true)) // Race detector uses this location for synchronization. c.buf = c.raceaddr() case elem.ptrdata == 0: // 元素不包含指针类型，只进行一次 hchan 和 buf 的内存分配 // 当存储在buf中的元素不包含指针时，GC就不会扫描hchan中的元素 c = (*hchan)(mallocgc(hchanSize+mem, nil, true)) c.buf = add(unsafe.Pointer(c), hchanSize) default: // 元素中包含指针类型，进行2次内存分配操作 // 用new分配内存返回的是指针 c = new(hchan) c.buf = mallocgc(mem, elem, true) } // 初始化channel数据 c.elemsize = uint16(elem.size) c.elemtype = elem c.dataqsiz = uint(size) lockInit(&c.lock, lockRankHchan) if debugChan { print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n") } // 返回 hchan 的指针类型 return c }

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool { // 如果当前channel是nil if c == nil { // 如果不阻塞，则直接返回false if !block { return false } // 挂起当前goroutine gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2) throw("unreachable") } if debugChan { print("chansend: chan=", c, "\n") } if raceenabled { racereadpc(c.raceaddr(), callerpc, funcPC(chansend)) } // 这里访问了hchan结构中的closed, full函数内部访问了dataqsiz,recvq,qcount字段，这里没有加锁，是为什么呢？ // 先说说这里判断的含义：如果不阻塞，且channel没有被关闭，且buf已满，则快速返回false，表示数据发送失败。 // 因为没有加锁，假如在判断c.closed == 0之后结果为true，在判断full之前，这时channel被其他goroutine关闭了， // 然后full函数返回了true，那么它会直接return false，这样子会有什么影响呢？ // 其实并没有什么影响，在这种情况下返回false也是合理的，因为都是表示在不阻塞的情况下发送数据失败。 // 所以这里访问hchan里面的数据就没有加锁了 if !block && c.closed == 0 && full(c) { return false } var t0 int64 if blockprofilerate > 0 { t0 = cputicks() } // 锁住channel，可见channel是并发安全的 lock(&c.lock) // 如果channel已关闭，则panic if c.closed != 0 { unlock(&c.lock) panic(plainError("send on closed channel")) } // 如果recvq等待接收队列中有值，则直接把值传给等待接收的goroutine，这样可以减少一次内存拷贝 if sg := c.recvq.dequeue(); sg != nil { send(c, sg, ep, func() { unlock(&c.lock) }, 3) return true } // 如果recvq等待接收队列中没有值，且为有缓冲channel，则把数据copy到buf中 if c.qcount < c.dataqsiz { // Space is available in the channel buffer. Enqueue the element to send. qp := chanbuf(c, c.sendx) if raceenabled { racenotify(c, c.sendx, nil) } typedmemmove(c.elemtype, qp, ep) c.sendx++ // 因为buf是环形数组，所以如果sendx超出了最大index，就要归0 if c.sendx == c.dataqsiz { c.sendx = 0 } c.qcount++ unlock(&c.lock) return true } // 如果recvq等待接收队列中没有值，且为无缓冲channel，且不阻塞，则直接返回false if !block { unlock(&c.lock) return false } // 接下来做阻塞当前goroutine的一些准备工作，构造一个sudog // 获取当前goroutine的指针 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 // 把构建好的 sudog 加到 sendq 发送等待队列中 c.sendq.enqueue(mysg) // Signal to anyone trying to shrink our stack that we're about // to park on a channel. The window between when this G's status // changes and when we set gp.activeStackChans is not safe for // stack shrinking. atomic.Store8(&gp.parkingOnChan, 1) // 挂起当前goroutine gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2) // 如果当前 goroutine 被唤醒后，会在这里继续执行 // 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 gp.activeStackChans = false closed := !mysg.success gp.param = nil if mysg.releasetime > 0 { blockevent(mysg.releasetime-t0, 2) } mysg.c = nil releaseSudog(mysg) if closed { if c.closed == 0 { throw("chansend: spurious wakeup") } // 如果唤醒后，发现 channel 被关闭，则关闭 panic(plainError("send on closed channel")) } return true }

// 判断channel中是否还有位置存放数据 func full(c *hchan) bool { // 如果是非缓冲channel if c.dataqsiz == 0 { // 如果 recvq 中没有等待接收数据的 goroutine，则返回 true，表示已满，否则返回 false return c.recvq.first == nil } // 如果是有缓冲 channel，则判断buf是否已满 return c.qcount == c.dataqsiz }

// 在一个空的 channel c 中完成发送操作 // 把数据 ep 从发送者复制到接收者 sg 中 // 最后接收的 goroutine 会被唤醒 // channel c 一定是空的且被锁住的 // sg 一定是已经从 c 的 recvq 中出队了 // eq 一定是不等于 nil 的，且指向堆或者是调用者的栈 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. racenotify(c, c.recvx, nil) racenotify(c, c.recvx, sg) c.recvx++ if c.recvx == c.dataqsiz { c.recvx = 0 } c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz } } // sg.elem 指向接收者存放接收数据的存放的位置 if sg.elem != nil { // 直接内存拷贝，从发送者拷贝到接收者内存 sendDirect(c.elemtype, sg, ep) sg.elem = nil } gp := sg.g // 解锁 unlockf() gp.param = unsafe.Pointer(sg) sg.success = true if sg.releasetime != 0 { sg.releasetime = cputicks() } // 唤醒接收数据的goroutine goready(gp, skip+1) }