首页 > 其他分享 >Go的Channel底层实现细节

Go的Channel底层实现细节

时间:2022-10-05 17:32:40浏览次数:83  
标签:队列 buf elem hchan Go 底层 data Channel size

相信大家在开发的过程中经常会使用到go中并发利器channelchannelCSP并发模型中最重要的一个组件,两个独立的并发实体通过共享的通讯channel进行通信。大多数人只是会用这么个结构很少有人讨论它底层实现,这篇文章讲写写channel的底层实现。

channel

channel的底层实现是一个结构体,源代码如下:

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
}

可能看源代码不是很好看得懂,这里我个人画了一张图方便大家查看,我在上面标注了不同颜色,并且注释其作用。

hchan struct

通道像一个传送带或者队列,总是遵循FIFO的规则,保证收发数据的顺序,通道是goroutine间重要通信的方式,是并发安全的。

buf

hchan结构体中的buf指向一个循环队列,用来实现循环队列,sendx是循环队列的队尾指针,recvx是循环队列的队头指针,dataqsize是缓存型通道的大小,qcount是记录通道内元素个数。

在日常开发过程中用的最多就是ch := make(chan int, 10)这样的方式创建一个通道,如果这要声明初始化的话,这个通道就是有缓冲区的,也是图上紫色的bufbuf是在make的时候程序创建的,它有元素大小*元素个数组成一个循环队列,可以看做成一个环形结构,buf则是一个指针指向这个环。

ring

上图对应的代码那就是ch = make(chan int,6)buf指向这个环在heap上的地址。

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:
        // 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.ptrdata == 0:
        // Elements do not contain pointers.
        // Allocate hchan and buf in one call.
        c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
        c.buf = add(unsafe.Pointer(c), hchanSize)
    default:
        // Elements contain pointers.
        c = new(hchan)
        c.buf = mallocgc(mem, elem, true)
    }

    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")
    }
    return c
}

上面就是对应的代码实现,上来它会检查你一系列参数是否合法,然后在通过mallocgc在内存开辟这块空间,然后返回。

sendx and recvx

下面我手动模拟一个ring实现的代码:

// Queue cycle buffer
type CycleQueue struct {
    data                  []interface{} // 存放元素的数组,准确来说是切片
    frontIndex, rearIndex int           // frontIndex 头指针,rearIndex 尾指针
    size                  int           // circular 的大小
}

// NewQueue Circular Queue
func NewQueue(size int) (*CycleQueue, error) {
    if size <= 0 || size < 10 {
      return nil, fmt.Errorf("initialize circular queue size fail,%d not legal,size >= 10", size)
    }
    cq := new(CycleQueue)
    cq.data = make([]interface{}, size)
    cq.size = size
    return cq, nil
}

// Push  add data to queue
func (q *CycleQueue) Push(value interface{}) error {
    if (q.rearIndex+1)%cap(q.data) == q.frontIndex {
        return errors.New("circular queue full")
    }
    q.data[q.rearIndex] = value
    q.rearIndex = (q.rearIndex + 1) % cap(q.data)
    return nil
}

// Pop return queue a front element
func (q *CycleQueue) Pop() interface{} {
    if q.rearIndex == q.frontIndex {
        return nil
    }
    v := q.data[q.frontIndex]
    q.data[q.frontIndex] = nil // 拿除元素 位置就设置为空
    q.frontIndex = (q.frontIndex + 1) % cap(q.data)
    return v
}

循环队列一般使用空余单元法来解决队空和队满时候都存在font=rear带来的二义性问题,但这样会浪费一个单元。golangchannel中是通过增加qcount字段记录队列长度来解决二义性,一方面不会浪费一个存储单元,另一方面当使用len函数查看队列长度时候,可以直接返回qcount字段,一举两得。

对应的源代码

当我们需要读取的数据的时候直接从recvx指针上的元素取,而写就从sendx位置写入元素,如图:

读写指针

sendq and recvq

当写入数据的如果缓冲区已经满或者读取的缓冲区已经没有数据的时候,就会发生协程阻塞。

send

如果写阻塞的时候会把当前的协程加入到sendq的队列中,直到有一个recvq发起了一个读取的操作,那么写的队列就会被程序唤醒进行工作。

全部挂起态

当缓冲区满了所有的g-w则被加入sendq队列等待g-r有操作就被唤醒g-w,继续工作,这种设计和操作系统的里面thread5种状态很接近了,可以看出go的设计者在可能参考过操作系统的thread设计。

当然上面只是我简述整个个过程,实际上go还做了其他细节优化,sendq不为空的时候,并且没有缓冲区,也就是无缓冲区通道,此时会从sendq第一个协程中拿取数据,有兴趣的gopher可以去自己查看源代码,本文也是最近笔者在看到这块源代码的笔记总结。

标签:队列,buf,elem,hchan,Go,底层,data,Channel,size
From: https://blog.51cto.com/u_13482767/5732780

相关文章

  • Django的urls的配置
    在一个请求到达的时候,最先达到的就是视图层,然后根据url映射到视图函数。这一部分我们来说明url的配置。概述为了给一个应用设计URL,你需要创建一个Python模块,通常称为URL......
  • Django 之ModelAdmin对象
    一、后台管理系统配置1在该表对应的models类里面添加一个Meta类classMeta:verbose_name_plural='用户表'#指定该表在admin后台的名字为:用户表#verbo......
  • go 反射
             ......
  • go 泛型
            ......
  • go学习
    一、接口定义概念:Interface类型可以定义一组方法,不需要实现,并且不能包含任何变量,称之为接口。接口不需要显式的实现,只需要一个变量,含有接口类型中的所有方法,那......
  • Django3.2整站开发
    DTL模版引擎模版配置第一步:在项目配置文件中配置模版引擎相关的配置,Django默认引擎为DTL。TEMPLATES=[{ #配置模版引擎,默认使用DjangoTemplateLangua......
  • Django项目的创建、启动、停止
    1.首先安装Django的包pipinstallDjango的命令安装包 2.配置环境变量,必须配置,后边需要用到django的命令,这个包和别的包不太一样的地方C:\Users**********Python\P......
  • Google – Reviews
    前言继上一篇 Facebook–Reviews(GraphAPI) 后,这篇继续介绍另一个Reviews大平台GoogleReviews.想通过API获取GoogleReviews难度比FacebookReviews高......
  • mongo操作工具类(索引创建等)
    importcn.hutool.core.date.DateUtil;importcom.alibaba.fastjson.JSON;importcom.ft.monitoring.management.OnlineApplication;importcom.mongodb.BasicDBObject......
  • Django使用uwsgi部署教程
     一直想用uwsgi部署Django,因为uwsgi性能更好而且可以多进程跑(pythonmanager.pyrunserver是单进程)。但在windows开发机上实验的时候死活装不了uwsgi(有人说可以......