一、线程的实现模型

二、Go的并发机制

type g struct {    stack      stack    // Goroutine的栈内存范围[stack.lo, stack.hi)    stackguard0    uintptr  // 用于调度器抢占式调度    m        *m    // Goroutine占用的线程    sched      gobuf    // Goroutine的调度相关数据    atomicstatus  uint32  // Goroutine的状态    ...}
type gobuf struct {    sp   uintptr    // 栈指针    pc   uintptr    // 程序计数器    g   guintptr    // gobuf对应的Goroutine    ret   sys.Uintewg  // 系统调用的返回值    ...}

type m struct {    g0      *g      // 一个特殊的goroutine，执行一些运行时任务    gsignal    *g      // 处理signal的G    curg    *g      // 当前M正在运行的G的指针    p      puintptr  // 正在与当前M关联的P    nextp    puintptr  // 与当前M潜在关联的P    oldp    puintptr  // 执行系统调用之前使用线程的P    spinning  bool    // 当前M是否正在寻找可运行的G    lockedg    *g      // 与当前M锁定的G}

type p struct {  // p 的状态  status   uint32    // 对应关联的 M  m        muintptr      // 可运行的Goroutine队列，可无锁访问  runqhead uint32  runqtail uint32  runq     [256]guintptr  // 缓存可立即执行的G  runnext    guintptr   // 可用的G列表，G状态等于Gdead   gFree struct {    gList    n int32  }  ...}

三、调度器

func schedule() {  _g_ := getg()
top:  var gp *g  var inheritTime bool
  if gp == nil {        // 为了公平，每调用schedule函数61次就要从全局可运行G队列中获取    if _g_.m.p.ptr().schedtick%61 == 0 && sched.runqsize > 0 {      lock(&sched.lock)      gp = globrunqget(_g_.m.p.ptr(), 1)      unlock(&sched.lock)    }  }    // 从P本地获取G任务  if gp == nil {    gp, inheritTime = runqget(_g_.m.p.ptr())  }    // 运行到这里表示从本地运行队列和全局运行队列都没有找到需要运行的G  if gp == nil {        // 阻塞地查找可用G    gp, inheritTime = findrunnable()  }    // 执行G任务函数  execute(gp, inheritTime)}

func execute(gp *g, inheritTime bool) {  _g_ := getg()
  // 将G绑定到当前M上  _g_.m.curg = gp  gp.m = _g_.m  // 将g正式切换为_Grunning状态  casgstatus(gp, _Grunnable, _Grunning)  gp.waitsince = 0  // 抢占信号  gp.preempt = false  gp.stackguard0 = gp.stack.lo + _StackGuard  if !inheritTime {    // 调度器调度次数增加1    _g_.m.p.ptr().schedtick++  }   ...     // gogo完成从g0到gp的切换  gogo(&gp.sched)}

MOVL gobuf_sp(BX), SP  // 将runtime.goexit函数的PC恢复到SP中MOVL gobuf_pc(BX), BX  // 获取待执行函数的程序计数器JMP  BX                // 开始执行

func goexit0(gp *g) {  _g_ := getg()  // 设置当前G状态为_Gdead  casgstatus(gp, _Grunning, _Gdead)   // 清理G  gp.m = nil  ...  gp.writebuf = nil  gp.waitreason = 0  gp.param = nil  gp.labels = nil  gp.timer = nil   // 解绑M和G  dropg()   ...  // 将G扔进gfree链表中等待复用  gfput(_g_.m.p.ptr(), gp)  // 再次进行调度  schedule()}

四、Channel

type hchan struct {    qcount    uint        // 循环列表元素个数    dataqsiz  uint        // 循环队列的大小    buf      unsafe.Pointer  // 循环队列的指针    elemsize  uint16      // chan中元素的大小    closed    uint32      // 是否已close    elemtype  *_type      // chan中元素类型    sendx    uint        // chan的发送操作处理到的位置    recvx    uint        // chan的接收操作处理到的位置    recvq    waitq        // 等待接收数据的Goroutine列表    sendq    waitq        // 等待发送数据的Goroutine列表        lock    mutex        // 互斥锁}
type waitq struct {        // 双向链表    first  *sudog    last  *sudog}

const (  maxAlign  = 8  hchanSize = unsafe.Sizeof(hchan{}) + uintptr(-int(unsafe.Sizeof(hchan{}))&(maxAlign-1)))
func makechan(t *chantype, size int) *hchan {  elem := t.elem        // 计算需要分配的buf空间大小  mem, overflow := math.MulUintptr(elem.size, uintptr(size))  if overflow || mem > maxAlloc-hchanSize || size < 0 {    panic(plainError("makechan: size out of range"))  }
  var c *hchan  switch {  case mem == 0:    // chan的大小或者elem的大小为0，不需要创建buf    c = (*hchan)(mallocgc(hchanSize, nil, true))    // Race detector uses this location for synchronization.    c.buf = c.raceaddr()  case elem.ptrdata == 0:    // elem不含指针，分配一块连续的内存给hchan数据结构和buf    c = (*hchan)(mallocgc(hchanSize+mem, nil, true))    c.buf = add(unsafe.Pointer(c), hchanSize)  default:    // elem包含指针，单独分配buf    c = new(hchan)    c.buf = mallocgc(mem, elem, true)  }
    // 更新hchan的elemsize、elemtype、dataqsiz字段  c.elemsize = uint16(elem.size)  c.elemtype = elem  c.dataqsiz = uint(size)      return c}

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {    if c == nil {    // 对于非阻塞的发送，直接返回    if !block {      return false    }    // 对于阻塞的通道，将goroutine挂起    gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)    throw("unreachable")  }    // 加锁  lock(&c.lock)    // channel已关闭，panic  if c.closed != 0 {    unlock(&c.lock)    panic(plainError("send on closed channel"))  }    ...}

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {    ...    // 从recvq中取出一个接收者  if sg := c.recvq.dequeue(); sg != nil {     // 如果接收者存在，直接向该接收者发送数据，绕过buf    send(c, sg, ep, func() { unlock(&c.lock) }, 3)    return true  }    ...}

func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {  ...  if sg.elem != nil {    // 直接把要发送的数据copy到接收者的栈空间    sendDirect(c.elemtype, sg, ep)    sg.elem = nil  }  gp := sg.g  unlockf()  gp.param = unsafe.Pointer(sg)  if sg.releasetime != 0 {    sg.releasetime = cputicks()  }  // 设置对应的goroutine为可运行状态  goready(gp, skip+1)}

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {  ...  // 如果缓冲区没有满，直接将要发送的数据复制到缓冲区  if c.qcount < c.dataqsiz {    // 找到buf要填充数据的索引位置    qp := chanbuf(c, c.sendx)    ...    // 将数据拷贝到buf中    typedmemmove(c.elemtype, qp, ep)    // 数据索引前移，如果到了末尾，又从0开始    c.sendx++    if c.sendx == c.dataqsiz {      c.sendx = 0    }    // 元素个数加1，释放锁并返回    c.qcount++    unlock(&c.lock)    return true  }  ...}

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {  ...  // 缓冲区没有空间了，对于非阻塞调用直接返回  if !block {    unlock(&c.lock)    return false  }  // 创建sudog对象  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  // 将sudog对象入队  c.sendq.enqueue(mysg)  // 进入等待状态  gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)  ...}

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {  if c == nil {        // 如果c为空且是非阻塞调用，直接返回    if !block {      return    }        // 阻塞调用直接等待    gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)    throw("unreachable")  }  ···  lock(&c.lock)    // 如果c已经关闭，并且c中没有数据，返回  if c.closed != 0 && c.qcount == 0 {    unlock(&c.lock)    if ep != nil {      typedmemclr(c.elemtype, ep)    }    return true, false  }    ···}

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {  ...  // 从发送者队列获取数据  if sg := c.sendq.dequeue(); sg != nil {     // 发送者队列不为空，直接从发送者那里提取数据    recv(c, sg, ep, func() { unlock(&c.lock) }, 3)    return true, true  }   ...}

func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {    // 如果是无缓冲区chan  if c.dataqsiz == 0 {    if ep != nil {            // 直接从发送者拷贝数据      recvDirect(c.elemtype, sg, ep)    }    // 有缓冲区chan  } else {        // 获取buf的存放数据指针    qp := chanbuf(c, c.recvx)        // 直接从缓冲区拷贝数据给接收者    if ep != nil {      typedmemmove(c.elemtype, ep, qp)    }        // 从发送者拷贝数据到缓冲区    typedmemmove(c.elemtype, qp, sg.elem)    c.recvx++    c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz  }  gp := sg.g  gp.param = unsafe.Pointer(sg)    // 设置对应的goroutine为可运行状态  goready(gp, skip+1)}

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {  ...    // 如果缓冲区中有数据  if c.qcount > 0 {    qp := chanbuf(c, c.recvx)        // 从缓冲区复制数据到ep    if ep != nil {      typedmemmove(c.elemtype, ep, qp)    }    typedmemclr(c.elemtype, qp)        // 接收数据的指针前移    c.recvx++        // 环形队列，如果到了末尾，再从0开始    if c.recvx == c.dataqsiz {      c.recvx = 0    }        // 缓冲区中现存数据减一    c.qcount--        unlock(&c.lock)    return true, true  }  ...}

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {  ...  // 非阻塞，直接返回  if !block {    unlock(&c.lock)    return false, false  }   // 创建sudog  gp := getg()  mysg := acquireSudog()  ···  gp.waiting = mysg  mysg.g = gp  mysg.isSelect = false  mysg.c = c  gp.param = nil  // 将sudog添加到等待接收队列中  c.recvq.enqueue(mysg)  // 阻塞Goroutine，等待被唤醒  gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)  ...}

func closechan(c *hchan) {    // 校验逻辑    ...    lock(&c.lock)    // 设置chan已关闭  c.closed = 1  var glist gList    // 获取所有接收者  for {    sg := c.recvq.dequeue()    if sg == nil {      break    }    if sg.elem != nil {      typedmemclr(c.elemtype, sg.elem)      sg.elem = nil    }    gp := sg.g    gp.param = nil    glist.push(gp)  }  // 获取所有发送者  for {    sg := c.sendq.dequeue()    ...  }    unlock(&c.lock)    // 唤醒所有glist中的goroutine  for !glist.empty() {    gp := glist.pop()    gp.schedlink = 0    goready(gp, 3)  }}

 作者简介