queue_work

【golang源码分析】chan底层原理——附带读写⽤户队列的环

形缓冲区

1环形缓冲区

1.1环形缓冲区结构

环形缓冲区通常有⼀个读指针和⼀个写指针。读指针指向环形缓冲区中可读的数据，写指针指向环形缓冲区中可写的缓冲区。通过移动

读指针和写指针就可以实现缓冲区的数据读取和写⼊。在通常情况下，环形缓冲区的读⽤户仅仅会影响读指针，⽽写⽤户仅仅会影响写指

针。如果仅仅有⼀个读⽤户和⼀个写⽤户，那么不需要添加互斥保护机制就可以保证数据的正确性。如果有多个读写⽤户访问环形缓冲区，

那么必须添加互斥保护机制来确保多个⽤户互斥访问环形缓冲区。

1.2环形缓冲区⼀种读写实现机制

⼀般的，圆形缓冲区需要4个指针

在内存中实际开始位置；

在内存中实际结束位置，也可以⽤缓冲区长度代替；

存储在缓冲区中的有效数据的开始位置（读指针）；

存储在缓冲区中的有效数据的结尾位置（写指针）。

缓冲区是满、或是空，都有可能出现读指针与写指针指向同⼀位置：

缓冲区中总是有⼀个存储单元保持未使⽤状态。缓冲区最多存⼊（size-1）个数据。如果读写指针指向同⼀位置，则缓冲区为空。如果写指

针位于读指针的相邻后⼀个位置，则缓冲区为满。

2chan内部数据结构

2.1chan的数据结构

chan实质是个环形缓冲区，外加⼀个接受者协程队列和⼀个发送者协程队列

buf：环形缓冲区

sendx：⽤于记录buf这个循环链表中的发送的index

recvx：⽤于记录buf这个循环链表中接收的index

recvq：接受者协程队列

sendq：发送者协程队列

lock：互斥锁

2.2有缓冲区和⽆缓冲区chan的区别

2.2.1⽆缓冲chan数据同步过程和sudog结构

1.创建⼀个发送者列表和接收者列表都为空的channel。

2.第⼀个协程向channel发送变量的值

l从池中获取⼀个sudog结构体变量，⽤于表⽰发送者。sudog结构体会保持对发送者所在协程的引⽤，以及发送变量的引

⽤。

4.发送者加⼊sendq队列。

5.发送者协程进⼊等待状态。

6.第⼆个协程将从channel中读取⼀个消息

l将sendq列表中等待状态的发送者出队列。

使⽤memmove函数将发送者要发送的值进⾏拷贝，包装⼊sudog结构体，再传递给channel接收者的接收变量。

9.在第五步中被挂起的第⼀个协程将恢复运⾏并释放第三步中获取的sudog结构体。

2.2.1有缓冲chan

有缓冲chan实质是使⽤了完整的环形缓冲区，只要缓冲区有空闲，则发送者⽆需进⼊等待队列，直接将数据放⼊环形缓冲区中，如果缓冲

区有数据，接受者⽆需进⼊等待队列，直接从环形缓冲区中获取数据。

3关键源码分析

3.1chan数据结构源码

typehchanstruct{

qcountuint//totaldatainthequeue

dataqsizuint//sizeofthecircularqueue

r//pointstoanarrayofdataqsizelements

elemsizeuint16

closeduint32

elemtype*_type//elementtype

sendxuint//sendindex

recvxuint//receiveindex

recvqwaitq//listofrecvwaiters

sendqwaitq//listofsendwaiters

//lockprotectsallfieldsinhchan,aswellasseveral

//fieldsinsudogsblockedonthischannel.

//

//DonotchangeanotherG'sstatuswhileholdingthislock

//(inparticular,donotreadyaG),asthiscandeadlock

//withstackshrinking.

lockmutex

}

3.2sudog数据结构源码

//sudogrepresentsaginawaitlist,suchasforsending/receiving

//onachannel.

//

//sudogisnecessarybecausetheg↔synchronizationobjectrelation

//eonmanywaitlists,sotheremaybe

//manysudogsforoneg;andmanygsmaybewaitingonthesame

//synchronizationobject,sotheremaybemanysudogsforoneobject.

//

//uireSudogand

//releaseSudogtoallocateandfreethem.

typesudogstruct{

//the

//stackdependson

//thisforsudogsinvolvedinchannelops.

g*g

//isSelectindicatesgisparticipatinginaselect,so

//DonemustbeCAS'dtowinthewake-uprace.

isSelectbool

next*sudog

prev*sudog

r//dataelement(maypointtostack)

//Thefollowingfieldsareneveraccessedconcurrently.

//Forchannels,waitlinkisonlyaccessedbyg.

//Forsemaphores,allfields(includingtheonesabove)

//areonlyaccessedwhenholdingasemaRootlock.

acquiretimeint64

releasetimeint64

ticketuint32

parent*sudog//semaRootbinarytree

waitlink*sudog//glistorsemaRoot

waittail*sudog//semaRoot

c*hchan//channel

}

3.3chan的构造过程

funcmakechan(t*chantype,sizeint)*hchan{

elem:=

//compilerchecksthisbutbesafe.

>=1<<16{

throw("makechan:invalidchannelelementtype")

}

ifhchanSize%maxAlign!=0||>maxAlign{

throw("makechan:badalignment")

}

mem,overflow:=tptr(,uintptr(size))

ifoverflow||mem>maxAlloc-hchanSize||size<0{

panic(plainError("makechan:sizeoutofrange"))

}

//HchandoesnotcontainpointersinterestingforGCwhenelementsstoredinbufdonotcontainpointers.

//bufpointsintothesameallocation,elemtypeispersistent.

//SudoG'sarereferencedfromtheirowningthreadsotheycan'tbecollected.

//TODO(dvyukov,rlh):Rethinkwhencollectorcanmoveallocatedobjects.

varc*hchan

switch{

casemem==0:

//Queueorelementsizeiszero.

c=(*hchan)(mallocgc(hchanSize,nil,true))

//Racedetectorusesthislocationforsynchronization.

=dr()

a==0:

//Elementsdonotcontainpointers.

//Allocatehchanandbufinonecall.

c=(*hchan)(mallocgc(hchanSize+mem,nil,true))

=add(r(c),hchanSize)

default:

//Elementscontainpointers.

c=new(hchan)

=mallocgc(mem,elem,true)

}

ze=uint16()

pe=elem

iz=uint(size)

ifdebugChan{

print("makechan:chan=",c,";elemsize=",,";dataqsiz=",size,"n")

}

returnc

}

可以看到，如果不传⼊size或size=0，则没有为环形缓冲区分配内存，职位chan结构分配内存

3.4⽆缓冲收发

//Sendsandreceivesonunbufferedorempty-bufferedchannelsarethe

//onlyoperationswhereonerunninggoroutinewritestothestackof

//ssumesthatstackwritesonly

//happenwhenthegoroutineisrunningandareonlydonebythat

//writebarrierissufficienttomakeupfor

//violatingthatassumption,butthewritebarrierhastowork.

//typedmemmovewillcallbulkBarrierPreWrite,butthetargetbytes

//arenotintheheap,ngetocall

//memmoveandtypeBitsBulkBarrierinstead.

funcsendDirect(t*_type,sg*sudog,r){

//srcisonourstack,dstisaslotonanotherstack.

//tofsg,itwillnolonger

//beupdatedifthedestination'sstackgetscopied(shrunk).

//Somakesurethatnopreemptionpointscanhappenbetweenread&use.

dst:=

typeBitsBulkBarrier(t,uintptr(dst),uintptr(src),)

//Noneedforcgowritebarrierchecksbecausedstisalways

//Gomemory.

memmove(dst,src,)

}

funcrecvDirect(t*_type,sg*sudog,r){

//dstisonourstackortheheap,srcisonanotherstack.

//Thechannelislocked,sosrcwillnotmoveduringthis

//operation.

src:=

typeBitsBulkBarrier(t,uintptr(dst),uintptr(src),)

memmove(dst,src,)

}

参考：