七叶笔记 » golang编程 » Go内置数据结构原理

Go内置数据结构原理

分类: golang编程 | 浏览: 268

作者：jackshi，腾讯 PCG 后台开发工程师

从C++切换到Go语言一年多了，有必要深入了解一下Go语言内置数据结构的实现原理，本文结合示例与Go源码深入到Go语言的底层实现。

数组

定义

数组是切片和映射的基础数据结构。

数组是一个长度固定的数据类型，用于存储一段具有相同类型元素的连续块。

声明与初始化 

 var arr [5]int //声明一个包含五个元素的数组
arr := [5]int{1, 2, 3, 4, 5} //字面量声明
arr := [...]int{1, 2, 3, 4, 5} //编译期自动推导数组长度
arr := [5]int{1:10,3:30} //初始化索引为1和3的元素，其他默认值为0

数组的使用 

 arr := [5]int{10, 20, 30, 40, 50}
arr[2] = 35 //修改索引为2的元素的值

var arr1 [3]string
arr2 := [3]string{"red", "yellow", "green"}
arr1 = arr2 //数组拷贝（深拷贝）

var arr3 [4]string
arr3 = arr2 //error 数组个数不同

var arr [4][2]int //多维数组
arr[1][0] = 20
arr[0] = [2]int{1, 2}

函数间传递数组 

 var arr [1e6]int

func foo1(arr [1e6]int) { //每次拷贝整个数组
    ...
}

func foo2(arr *[1e6]int) { //传递指针，效率更高
    ...
}

切片

切片是一种数据结构，便于使用和管理数据集合。切片是围绕动态数组的概念构建的，可以按需自动增长和缩小。

内部实现 

 type SliceHeader struct {
 Data uintptr
 Len  int
 Cap  int
}

编译期间的切片是 Slice 类型的，但是在运行时切片由 SliceHeader结构体表示，其中Data 字段是指向数组的指针，Len表示当前切片的长度，而 Cap表示当前切片的容量，也就是Data数组的大小。

创建和初始化 

 slice := make([]string, 5) //创建一个字符串切片，长度和容量都是5个元素

slice := make([]int, 3, 5) //创建一个整型切片，长度为3个元素，容量为5个元素

slice := make([]string, 5, 3) //error 容量需要小于长度

slice := []string{"Red", "Yellow", "Green"} //使用字面量声明

var slice []int //创建nil整型切片

slice := make([]int, 0) //使用make创建空的整型切片

slice := []int{} //使用切片字面量创建空的整型切片

不管是使用nil切片还是空切片，对其调用内置函数append、len、cap效果都一样，尽量使用len函数来判断切片是否为空

赋值和切片 

 slice := []int{1, 2, 3, 4, 5} //创建一个整型切片，长度和容量都是5

slice[2] = 30 //修改索引为2的元素的值，slice当前结果为{1, 2, 30, 4, 5}

newSlice := slice[1:3] //创建一个新切片，长度为2个元素，容量为4个元素

对底层数组容量是k的切片slice[i:j]来说：
长度：j - i
容量：k - i

newSlice[0] = 20 //由于newSlice与slice底层共享一个数组，因此slice当前结果为{1, 20, 30, 4, 5}

newSlice[3] = 40 //error 虽然容量足够，但超过长度限制，会触发panic

切片增长 

 slice := []int{1, 2, 3, 4, 5}
newSlice := slice[1:3] //cap(newSlice) : 4 

newSlice = append(newSlice, 60) // slice当前结果为{1, 2, 3, 60, 5} 

//newSlice的append操作影响了slice的结果，可以尝试 newSlice := slice[1:3:3]再看看结果

newSlice = append(newSlice, []int{70, 80}...) //slice当前结果仍为{1, 2, 3, 60, 5} len(newSlice) : 5 cap(newSlice) : 8


func growslice(et *_type, old slice, cap int) slice {
 ...
 newcap := old.cap
 doublecap := newcap + newcap
 if cap > doublecap {
  newcap = cap
 } else {
  if old.cap < 1024 {
   newcap = doublecap
  } else {
   for 0 < newcap && newcap < cap {
    newcap += newcap / 4
   }
   if newcap <= 0 {
    newcap = cap
   }
  }
 }
 ...
}

在分配内存空间之前需要先确定新的切片容量，Go 语言根据切片的当前容量选择不同的策略进行扩容：

如果期望容量大于当前容量的两倍就会使用期望容量； 如果当前切片的长度小于 1024 就会将容量翻倍； 如果当前切片的长度大于 1024 就会每次增加 25% 的容量，直到新容量大于期望容量； 这里只是确定切片的大致容量，接下来还需要根据切片中元素的大小对它们进行对齐，当数组中元素所占的字节大小为 1、8 或者 2 的倍数时，运行时会使用如下所示的代码对其内存 

 func growslice(et *_type, old slice, cap int) slice {
    ...
    
 var overflow bool
 var lenmem, newlenmem, capmem uintptr
    switch {
 case et.size == 1:
  lenmem = uintptr(old.len)
  newlenmem = uintptr(cap)
  capmem = roundupsize(uintptr(newcap))
  overflow = uintptr(newcap) > maxAlloc
  newcap = int(capmem)
 case et.size == sys.PtrSize:
  lenmem = uintptr(old.len) * sys.PtrSize
  newlenmem = uintptr(cap) * sys.PtrSize
  capmem = roundupsize(uintptr(newcap) * sys.PtrSize)
  overflow = uintptr(newcap) > maxAlloc/sys.PtrSize
  newcap = int(capmem / sys.PtrSize)
 case isPowerOfTwo(et.size):
  var shift uintptr
  if sys.PtrSize == 8 {
   // Mask shift for better code generation.
   shift = uintptr(sys.Ctz64(uint64(et.size))) & 63
  } else {
   shift = uintptr(sys.Ctz32(uint32(et.size))) & 31
  }
  lenmem = uintptr(old.len) << shift
  newlenmem = uintptr(cap) << shift
  capmem = roundupsize(uintptr(newcap) << shift)
  overflow = uintptr(newcap) > (maxAlloc >> shift)
  newcap = int(capmem >> shift)
 default:
  lenmem = uintptr(old.len) * et.size
  newlenmem = uintptr(cap) * et.size
  capmem, overflow = math.MulUintptr(et.size, uintptr(newcap))
  capmem = roundupsize(capmem)
  newcap = int(capmem / et.size)
 }
 ...
}

runtime.roundupsize函数会将待申请的内存向上取整，取整时会使用内存分配中介绍的runtime.class_to_size数组，使用该数组中的整数可以提高内存的分配效率并减少碎片。 

 func growslice(et *_type, old slice, cap int) slice {
    ...
    
    if overflow || capmem > maxAlloc {
  panic(errorString("growslice: cap out of range"))
 }

 var p unsafe.Pointer
 if et.ptrdata == 0 {
  p = mallocgc(capmem, nil, false)
  memclrNoHeapPointers(add(p, newlenmem), capmem-newlenmem)
 } else {
  p = mallocgc(capmem, et, true)
  if lenmem > 0 && writeBarrier.enabled {
   bulkBarrierPreWriteSrcOnly(uintptr(p), uintptr(old.array), lenmem-et.size+et.ptrdata)
  }
 }
 memmove(p, old.array, lenmem)

 return slice{p, old.len, newcap}
}

如果切片中元素不是指针类型，那么就会调用memclrNoHeapPointers将超出切片当前长度的位置清空并在最后使用memmove将原数组内存中的内容拷贝到新申请的内存中。

runtime.growslice函数最终会返回一个新的 slice 结构，其中包含了新的数组指针、大小和容量，这个返回的三元组最终会改变原有的切片，帮助 append 完成元素追加的功能。

示例： 

 var arr []int64
arr = append(arr, 1, 2, 3, 4, 5)
fmt.Println(cap(arr))

当我们执行如上所示的代码时，会触发 runtime.growslice 函数扩容 arr 切片并传入期望的新容量 5，这时期望分配的内存大小为 40 字节；不过因为切片中的元素大小等于 sys.PtrSize，所以运行时会调用 runtime.roundupsize 对内存的大小向上取整 48 字节，所以经过计算新切片的容量为 48 / 8 = 6 。

拷贝切片

切片的拷贝虽然不是一个常见的操作类型，但是却是我们学习切片实现原理必须要谈及的一个问题，当我们使用 copy(a, b) 的形式对切片进行拷贝时，编译期间的 cmd/compile/internal/gc.copyany 函数也会分两种情况进行处理，如果当前 copy 不是在运行时调用的，copy(a, b) 会被直接转换成下面的代码： 

 n := len(a)
if n > len(b) {
    n = len(b)
}
if a.ptr != b.ptr {
    memmove(a.ptr, b.ptr, n*sizeof(elem(a))) 
}

其中 memmove 会负责对内存进行拷贝，在其他情况下，编译器会使用 runtime.slicecopy 函数替换运行期间调用的 copy，例如：go copy(a, b)： 

 func slicecopy(to, fm slice, width uintptr) int {
 if fm.len == 0 || to.len == 0 {
  return 0
 }
 n := fm.len
 if to.len < n {
  n = to.len
 }
 if width == 0 {
  return n
 }
 ...

 size := uintptr(n) * width
 if size == 1 {
  *(*byte)(to.array) = *(*byte)(fm.array)
 } else {
  memmove(to.array, fm.array, size)
 }
 return n
}

上述函数的实现非常直接，两种不同的拷贝方式一般都会通过 memmove将整块内存中的内容拷贝到目标的内存区域中。

迭代切片 

 slice := []int{1, 2, 3, 4, 5}
for index, value := range slice {
    value += 10
    fmt.Printf("index:%d value:%d\n", index, value)
}
fmt.Println(slice) // {1, 2, 3, 4, 5}

==

var value int
for index := 0; index < len(slice); index ++ {
    value = slice[index]
    
}

函数间传递切片 

 slice := make([]int, 1e6)

slice = foo(slice)

func foo(slice []int) []int {
    ...
    return slice
}

在64位架构的机器上，一个切片需要24字节内存，数组指针、长度、容量分别占用8字节，在函数间传递24字节数据会非常快速简单，这也是切片效率高的原因。

映射

映射是一种数据结构，用于存储一系列无序的键值对。

映射是一个集合，可以使用类似处理数组和切片的方式迭代，但映射是无序的集合，每次迭代映射的时候顺序也可能不一样。

创建和初始化 

 dict := make(map[string]int) 

dict := map[string]string{"Red":"#da1337", "Orange":"#e95a22"} //字面量方式

使用映射 

 colors := make(map[string]string)  //创建一个空映射

colors["Red"] = "#da1337" //赋值

var colors map[string]string
colors["Red"] = "#da1337" //error

value, ok := colors["Blue"] //判断键是否存在
if ok {
    fmt.Println(value)
}

value := colors["Blue"] //判断读取到的值是否空值
if value != "" {
    fmt.Println(value)
}

函数间传递映射 

 func removeColor(colors map[string]string, key string) {
    delete(colors, key)
}

func main(){
    colors := map[string]string{
        "AliceBlue" : "#f0f8ff",
        "Coral"     : "#ff7F50",
    }
    
    for key, value := range colors {
        fmt.Printf("Key: %s Value: %s\n", key, value)
    }
    
    removeColor(colors, "Coral")
    
    for key, value := range colors {
        fmt.Printf("Key: %s Value: %s\n", key, value)
    }
}

内存模型

Go语言采用的是哈希查找表，并且使用链表解决哈希冲突。 

 // hashmap的简称
type hmap struct {
 count     int       //元素个数
  flags      uint8     //标记位
 B         uint8     // bucket s的对数 log_2
 noverflow uint16    //overflow的bucket的近似数
 hash0     uint32    //hash种子
 buckets    unsafe.Pointer //指向buckets数组的指针，数组个数为2^B
 oldbuckets unsafe.Pointer //扩容时使用，buckets长度是oldbuckets的两倍
 nevacuate  uintptr  //扩容进度，小于此地址的buckets已经迁移完成
 extra *mapextra     //扩展信息
}

//当map的key和value都不是指针，并且size都小于128字节的情况下，会把 bmap 标记为不含指针，这样可以避免gc时扫描整个hmap。但是，我们看bmap其实有一个overflow的字段，是指针类型的，破坏了bmap不含指针的设想，这时会把overflow移动到extra字段来。
type mapextra struct {
 overflow    *[]*bmap
 oldoverflow *[]*bmap

 nextOverflow *bmap
}

// bucket
type bmap struct {
 tophash [bucketCnt]uint8 //bucketCnt = 8
 // keys     [8]keytype
 // values   [8]valuetype
 // pad      uintptr
 // overflow uintptr
}

bmap就是桶的数据结构，每个桶最多存储8个key-value对，所有的key都是经过hash后有相同的尾部，在桶内，根据hash值的高8位来决定桶中的位置。

注意到key和value是各自在一起的，不是key/value/key/value/…的方式，这样的好处是某些情况下可以省略padding字段，节省内存空间

如map[int64]int8，按照key/value/key/value/… 这样的模式存储，每一个key/value对之后都要额外 padding7个字节；而将所有的key，value 分别绑定到一起，key/key/…/value/value/…，只需在最后添加padding。

每个bucket设计成最多只能放8个key-value对，如果有第9个 key-value落入当前的bucket，那就需要再构建一个bucket，通过overflow指针连接起来。

map的扩容

装载因子：loadFactor := count / (2^B) count是map中元素的个数，2^B表示bucket数量

符合下面两个条件会触发扩容： 1）装载因子超过阈值：源码里定义的阈值为6.5 2）overflow的bucket数量过多：当B>15时，overflow的bucket数量超过2^15^，否则超过2^B^

增加桶搬迁示例： 

 const (
 bucketCntBits = 3
 bucketCnt     = 1 << bucketCntBits
 
 loadFactorNum = 13
 loadFactorDen = 2
)

func bucketShift(b uint8) uintptr {
 return uintptr(1) << (b & (sys.PtrSize*8 - 1))
}

func overLoadFactor(count int, B uint8) bool {
 return count > bucketCnt && uintptr(count) > loadFactorNum*(bucketShift(B)/loadFactorDen)
}

func tooManyOverflowBuckets(noverflow uint16, B uint8) bool {
 if B > 15 {
  B = 15
 }
 return noverflow >= uint16(1)<<(B&15)
}

func hashGrow(t *maptype, h *hmap) {
 bigger := uint8(1)
 if !overLoadFactor(h.count+1, h.B) {
  bigger = 0
  h.flags |= sameSizeGrow
 }
 oldbuckets := h.buckets
 newbuckets, nextOverflow := makeBucketArray(t, h.B+bigger, nil)

 flags := h.flags &^ (iterator | oldIterator)
 if h.flags&iterator != 0 {
  flags |= oldIterator
 }
 // commit the grow (atomic wrt gc)
 h.B += bigger
 h.flags = flags
 h.oldbuckets = oldbuckets
 h.buckets = newbuckets
 h.nevacuate = 0
 h.noverflow = 0

 if h.extra != nil && h.extra.overflow != nil {
  if h.extra.oldoverflow != nil {
   throw("oldoverflow is not nil")
  }
  h.extra.oldoverflow = h.extra.overflow
  h.extra.overflow = nil
 }
 if nextOverflow != nil {
  if h.extra == nil {
   h.extra = new(mapextra)
  }
  h.extra.nextOverflow = nextOverflow
 }

 // the actual copying of the hash table data is done incrementally
 // by growWork() and evacuate().
}

// 返回是否在搬迁中
func (h *hmap) growing() bool {
 return h.oldbuckets != nil
}

// 执行搬迁
func growWork(t *maptype, h *hmap, bucket uintptr) {
 // make sure we evacuate the oldbucket corresponding
 // to the bucket we're about to use
 evacuate(t, h, bucket&h.oldbucketmask()) //搬迁一个桶

 // evacuate one more oldbucket to make progress on growing
 if h.growing() {
  evacuate(t, h, h.nevacuate)
 }
}


// evacDst 搬迁结构体
type evacDst struct {
 b *bmap          // current destination bucket
 i int            // key/elem index into b
 k unsafe.Pointer // pointer to current key storage
 e unsafe.Pointer // pointer to current elem storage
}
// 执行搬迁
func evacuate(t *maptype, h *hmap, oldbucket uintptr) {
 b := (*bmap)(add(h.oldbuckets, oldbucket*uintptr(t.bucketsize)))
 newbit := h.noldbuckets()
 if !evacuated(b) {
  var xy [2]evacDst
  x := &xy[0]
  x.b = (*bmap)(add(h.buckets, oldbucket*uintptr(t.bucketsize)))
  x.k = add(unsafe.Pointer(x.b), dataOffset)
  x.e = add(x.k, bucketCnt*uintptr(t.keysize))

  if !h.sameSizeGrow() {
   // Only calculate y pointers if we're growing bigger.
   // Otherwise GC can see bad pointers.
   y := &xy[1]
   y.b = (*bmap)(add(h.buckets, (oldbucket+newbit)*uintptr(t.bucketsize)))
   y.k = add(unsafe.Pointer(y.b), dataOffset)
   y.e = add(y.k, bucketCnt*uintptr(t.keysize))
  }

  for ; b != nil; b = b.overflow(t) {
   k := add(unsafe.Pointer(b), dataOffset)
   e := add(k, bucketCnt*uintptr(t.keysize))
   for i := 0; i < bucketCnt; i, k, e = i+1, add(k, uintptr(t.keysize)), add(e, uintptr(t.elemsize)) {
    top := b.tophash[i]
    if isEmpty(top) {
     b.tophash[i] = evacuatedEmpty
     continue
    }
    if top < minTopHash {
     throw("bad map state")
    }
    k2 := k
    if t.indirectkey() {
     k2 = *((*unsafe.Pointer)(k2))
    }
    var useY uint8
    if !h.sameSizeGrow() {
     
     hash := t.hasher(k2, uintptr(h.hash0))
     if h.flags&iterator != 0 && !t.reflexivekey() && !t.key.equal(k2, k2) {
      useY = top & 1
      top = tophash(hash)
     } else {
      if hash&newbit != 0 {
       useY = 1
      }
     }
    }

    if evacuatedX+1 != evacuatedY || evacuatedX^1 != evacuatedY {
     throw("bad evacuatedN")
    }

    b.tophash[i] = evacuatedX + useY // evacuatedX + 1 == evacuatedY
    dst := &xy[useY]                 // evacuation destination

    if dst.i == bucketCnt {
     dst.b = h.newoverflow(t, dst.b)
     dst.i = 0
     dst.k = add(unsafe.Pointer(dst.b), dataOffset)
     dst.e = add(dst.k, bucketCnt*uintptr(t.keysize))
    }
    dst.b.tophash[dst.i&(bucketCnt-1)] = top // mask dst.i as an optimization, to avoid a bounds check
    if t.indirectkey() {
     *(*unsafe.Pointer)(dst.k) = k2 // copy pointer
    } else {
     typedmemmove(t.key, dst.k, k) // copy elem
    }
    if t.indirectelem() {
     *(*unsafe.Pointer)(dst.e) = *(*unsafe.Pointer)(e)
    } else {
     typedmemmove(t.elem, dst.e, e)
    }
    dst.i++
    dst.k = add(dst.k, uintptr(t.keysize))
    dst.e = add(dst.e, uintptr(t.elemsize))
   }
  }
  // Unlink the overflow buckets & clear key/elem to help GC.
  if h.flags&oldIterator == 0 && t.bucket.ptrdata != 0 {
   b := add(h.oldbuckets, oldbucket*uintptr(t.bucketsize))
   ptr := add(b, dataOffset)
   n := uintptr(t.bucketsize) - dataOffset
   memclrHasPointers(ptr, n)
  }
 }

 if oldbucket == h.nevacuate {
  advanceEvacuation(h, t, newbit)
 }
}

func advanceEvacuation(h *hmap, t *maptype, newbit uintptr) {
 h.nevacuate++
 stop := h.nevacuate + 1024
 if stop > newbit {
  stop = newbit
 }
 for h.nevacuate != stop && bucketEvacuated(t, h, h.nevacuate) {
  h.nevacuate++
 }
 if h.nevacuate == newbit { // newbit == # of oldbuckets
  // 搬迁完成，释放旧的桶数据
  h.oldbuckets = nil
  if h.extra != nil {
   h.extra.oldoverflow = nil
  }
  h.flags &^= sameSizeGrow
 }
}

map的创建 

 //注：go:nosplit该指令指定文件中声明的下一个函数不得包含堆栈溢出检查。简单来讲，就是这个函数跳过堆栈溢出的检查

//go:nosplit 快速得到随机数 参考论文：
func fastrand() uint32 {
 mp := getg().m
 s1, s0 := mp.fastrand[0], mp.fastrand[1]
 s1 ^= s1 << 17
 s1 = s1 ^ s0 ^ s1>>7 ^ s0>>16
 mp.fastrand[0], mp.fastrand[1] = s0, s1
 return s0 + s1
}

func makemap(t *maptype, hint int, h *hmap) *hmap {
 mem, overflow := math.MulUintptr(uintptr(hint), t.bucket.size)
 if overflow || mem > maxAlloc {
  hint = 0
 }

 // 初始化
 if h == nil {
  h = new(hmap)
 }
 h.hash0 = fastrand()

 // 计算合适的桶个数B
 B := uint8(0)
 for overLoadFactor(hint, B) {
  B++
 }
 h.B = B

 // 分配哈希表
 if h.B != 0 {
  var nextOverflow *bmap
  h.buckets, nextOverflow = makeBucketArray(t, h.B, nil)
  if nextOverflow != nil {
   h.extra = new(mapextra)
   h.extra.nextOverflow = nextOverflow
  }
 }

 return h
}

// makeBucketArray initializes a backing array for map buckets.
func makeBucketArray(t *maptype, b uint8, dirtyalloc unsafe.Pointer) (
    buckets unsafe.Pointer, nextOverflow *bmap) {
    
 base := bucketShift(b)
 nbuckets := base
 // 对于比较小的b，不额外申请过载空间
 if b >= 4 {
  nbuckets += bucketShift(b - 4)
  sz := t.bucket.size * nbuckets
  up := roundupsize(sz)
  if up != sz {
   nbuckets = up / t.bucket.size
  }
 }

 if dirtyalloc == nil {
  buckets = newarray(t.bucket, int(nbuckets))
 } else {
  buckets = dirtyalloc
  size := t.bucket.size * nbuckets
  if t.bucket.ptrdata != 0 {
   memclrHasPointers(buckets, size)
  } else {
   memclrNoHeapPointers(buckets, size)
  }
 }

 if base != nbuckets {
  nextOverflow = (*bmap)(add(buckets, base*uintptr(t.bucketsize)))
  last := (*bmap)(add(buckets, (nbuckets-1)*uintptr(t.bucketsize)))
  last.setoverflow(t, (*bmap)(buckets))
 }
 return buckets, nextOverflow
}

key的定位

key经过hash后得到一个64位整数，使用最后B位判断该key落入到哪个桶中，如果桶相同，说明有hash碰撞，在该桶中遍历寻找，采用hash值的高8位来快速比较，来看一个示例：

再来看一下源码： 

 const (
 emptyRest      = 0 // this cell is empty, and there are no more non-empty cells at higher indexes or overflows.
 emptyOne       = 1 // this cell is empty
 evacuatedX     = 2 // key/elem is valid.  Entry has been evacuated to first half of larger table.
 evacuatedY     = 3 // same as above, but evacuated to second half of larger table.
 evacuatedEmpty = 4 // cell is empty, bucket is evacuated.
 minTopHash     = 5 // minimum tophash for a normal filled cell
)

//计算hash值的高8位
func tophash(hash uintptr) uint8 {
 top := uint8(hash >> (sys.PtrSize*8 - 8))
 if top < minTopHash {
  top += minTopHash
 }
 return top
}

//p指针偏移x位
func add(p unsafe.Pointer, x uintptr) unsafe.Pointer {
 return unsafe.Pointer(uintptr(p) + x)
}

const PtrSize = 4 << (^uintptr(0) >> 63)           // 64位环境下是 8

//获取下一个桶节点
func (b *bmap) overflow(t *maptype) *bmap {
 return *(**bmap)(add(unsafe.Pointer(b), uintptr(t.bucketsize)-sys.PtrSize))
}

//判断是否搬迁完成
func evacuated(b *bmap) bool {
 h := b.tophash[0]
 return h > emptyOne && h < minTopHash
}

func mapaccess1(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer{}
func mapaccess2(t *maptype, h *hmap, key unsafe.Pointer) (unsafe.Pointer, bool){}

func mapaccess1(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer {
    // ......
    //h为空或数量为0返回空值
 if h == nil || h.count == 0 {
  if t.hashMightPanic() {
   t.hasher(key, 0) // see issue 
  }
  return unsafe.Pointer(&zeroVal[0])
 }
 //读写冲突
 if h.flags&hashWriting != 0 {
  throw("concurrent map read and map write")
 }
 //对请求key计算hash值
 hash := t.hasher(key, uintptr(h.hash0))
 //获取桶的掩码，如B=5,m=31即二进制（11111）
 m := bucketMask(h.B)
 //定位到对应的桶
 b := (*bmap)(add(h.buckets, (hash&m)*uintptr(t.bucketsize)))
 //判断是否处于扩容中
 if c := h.oldbuckets; c != nil {
  if !h.sameSizeGrow() {
   // There used to be half as many buckets; mask down one more power of two.
   m >>= 1
  }
  oldb := (*bmap)(add(c, (hash&m)*uintptr(t.bucketsize)))
  if !evacuated(oldb) {
   b = oldb
  }
 }

 //计算hash值的前8位
 top := tophash(hash)
bucketloop:
 for ; b != nil; b = b.overflow(t) {
  for i := uintptr(0); i < bucketCnt; i++ {
   if b.tophash[i] != top {
    if b.tophash[i] == emptyRest {
     break bucketloop
    }
    continue
   }
   //定位到key的位置
   k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))
   if t.indirectkey() { //对于指针需要解引用
    k = *((*unsafe.Pointer)(k))
   }
   if t.key.equal(key, k) {
    //定位到value的位置
    e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))
    if t.indirectelem() { //对于指针需要解引用
     e = *((*unsafe.Pointer)(e))
    }
    return e
   }
  }
 }
 return unsafe.Pointer(&zeroVal[0])
}

map的赋值 

 const hashWriting  = 4 // a goroutine is writing to the map


// 判断是否是空节点
func isEmpty(x uint8) bool {
 return x <= emptyOne // x <= 1
}

func mapassign(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer {
 if h == nil {
  panic(plainError("assignment to entry in nil map"))
 }
 
 //省略条件判断
 
 if h.flags&hashWriting != 0 {
  throw("concurrent map writes")
 }
 hash := t.hasher(key, uintptr(h.hash0))

 // Set hashWriting after calling t.hasher, since t.hasher may panic,
 // in which case we have not actually done a write.
 h.flags ^= hashWriting

 if h.buckets == nil {
  h.buckets = newobject(t.bucket) // newarray(t.bucket, 1)
 }

again:
 bucket := hash & bucketMask(h.B)
 if h.growing() {
  growWork(t, h, bucket)
 }
 b := (*bmap)(add(h.buckets, bucket*uintptr(t.bucketsize)))
 top := tophash(hash)

 var inserti *uint8
 var insertk unsafe.Pointer
 var elem unsafe.Pointer
bucketloop:
 for {
  for i := uintptr(0); i < bucketCnt; i++ {
   if b.tophash[i] != top {
    if isEmpty(b.tophash[i]) && inserti == nil {
     inserti = &b.tophash[i]
     insertk = add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))
     elem = add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))
    }
    if b.tophash[i] == emptyRest {
     break bucketloop
    }
    continue
   }
   k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))
   if t.indirectkey() {
    k = *((*unsafe.Pointer)(k))
   }
   if !t.key.equal(key, k) {
    continue
   }
   // already have a mapping for key. Update it.
   if t.needkeyupdate() {
    typedmemmove(t.key, k, key)
   }
   elem = add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))
   goto done
  }
  ovf := b.overflow(t)
  if ovf == nil {
   break
  }
  b = ovf
 }

 //没有找到key，需要分配空间，先判断是否需要扩容，需扩容则重新计算
 if !h.growing() && (overLoadFactor(h.count+1, h.B) || tooManyOverflowBuckets(h.noverflow, h.B)) {
  hashGrow(t, h)
  goto again // Growing the table invalidates everything, so try again
 }

 if inserti == nil {
  newb := h.newoverflow(t, b)
  inserti = &newb.tophash[0]
  insertk = add(unsafe.Pointer(newb), dataOffset)
  elem = add(insertk, bucketCnt*uintptr(t.keysize))
 }

 // store new key/elem at insert position
 if t.indirectkey() {
  kmem := newobject(t.key)
  *(*unsafe.Pointer)(insertk) = kmem
  insertk = kmem
 }
 if t.indirectelem() {
  vmem := newobject(t.elem)
  *(*unsafe.Pointer)(elem) = vmem
 }
 typedmemmove(t.key, insertk, key)
 *inserti = top
 h.count++

done:
 if h.flags&hashWriting == 0 {
  throw("concurrent map writes")
 }
 h.flags &^= hashWriting
 if t.indirectelem() {
  elem = *((*unsafe.Pointer)(elem))
 }
 return elem
}

map的删除 

 func mapdelete(t *maptype, h *hmap, key unsafe.Pointer) {
 //... 忽略检查
    
 if h == nil || h.count == 0 {
  if t.hashMightPanic() {
   t.hasher(key, 0) // see issue 23734
  }
  return
 }
 if h.flags&hashWriting != 0 {
  throw("concurrent map writes")
 }

 hash := t.hasher(key, uintptr(h.hash0))

 // Set hashWriting after calling t.hasher, since t.hasher may panic,
 // in which case we have not actually done a write (delete).
 h.flags ^= hashWriting

 bucket := hash & bucketMask(h.B)
 if h.growing() {
  growWork(t, h, bucket)
 }
 b := (*bmap)(add(h.buckets, bucket*uintptr(t.bucketsize)))
 bOrig := b
 top := tophash(hash)
search:
 for ; b != nil; b = b.overflow(t) {
  for i := uintptr(0); i < bucketCnt; i++ {
   if b.tophash[i] != top {
    if b.tophash[i] == emptyRest { //遍历到最后仍未找到
     break sear ch 
    }
    continue
   }
   k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))
   k2 := k
   if t.indirectkey() {
    k2 = *((*unsafe.Pointer)(k2))
   }
   if !t.key.equal(key, k2) { //比较key是否一致
    continue
   }
   // Only clear key if there are pointers in it.
   if t.indirectkey() {
    *(*unsafe.Pointer)(k) = nil
   } else if t.key.ptrdata != 0 {
    memclrHasPointers(k, t.key.size)
   }
   e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))
   if t.indirectelem() {
    *(*unsafe.Pointer)(e) = nil
   } else if t.elem.ptrdata != 0 {
    memclrHasPointers(e, t.elem.size)
   } else {
    memclrNoHeapPointers(e, t.elem.size)
   }
   //将tophash标记置为空值
   b.tophash[i] = emptyOne
   
   if i == bucketCnt-1 {
    if b.overflow(t) != nil && b.overflow(t).tophash[0] != emptyRest {
     goto notLast
    }
   } else {
    if b.tophash[i+1] != emptyRest {
     goto notLast
    }
   }
   for {
    b.tophash[i] = emptyRest
    if i == 0 {
     if b == bOrig {
      break // beginning of initial bucket, we're done.
     }
     // Find previous bucket, continue at its last entry.
     c := b
     for b = bOrig; b.overflow(t) != c; b = b.overflow(t) {
     }
     i = bucketCnt - 1
    } else {
     i--
    }
    if b.tophash[i] != emptyOne {
     break
    }
   }
  notLast:
   h.count--
   // Reset the hash seed to make it more difficult for attackers to
   // repeatedly trigger hash collisions. See issue 25237.
   if h.count == 0 {
    h.hash0 = fastrand()
   }
   break search
  }
 }

 if h.flags&hashWriting == 0 {
  throw("concurrent map writes")
 }
 h.flags &^= hashWriting
}

map的遍历

先来看个示例： 从红色位置开始，先遍历3号桶的e，然后是3号桶链接的f和g，接着遍历到0号桶，由于0号桶未搬迁，只遍历其中属于0号桶的b和c，接着到1号桶的h，最后是2号桶的a和d。

遍历结果：e->f->g->b->c->h->a->d

源码执行流程图： 

 // 哈希遍历结构体
type hiter struct {
 key         unsafe.Pointer // Must be in first position.  Write nil to indicate iteration end (see cmd/compile/internal/gc/range.go).
 elem        unsafe.Pointer // Must be in second position (see cmd/compile/internal/gc/range.go).
 t           *maptype
 h           *hmap
 buckets     unsafe.Pointer // bucket ptr at hash_iter initialization time
 bptr        *bmap          // current bucket
 overflow    *[]*bmap       // keeps overflow buckets of hmap.buckets alive
 oldoverflow *[]*bmap       // keeps overflow buckets of hmap.oldbuckets alive
 startBucket uintptr        // bucket iteration started at
 offset      uint8          // intra-bucket offset to start from during iteration (should be big enough to hold bucketCnt-1)
 wrapped     bool           // already wrapped around from end of bucket array to beginning
 B           uint8
 i           uint8
 bucket      uintptr
 checkBucket uintptr
}

func mapiterinit(t *maptype, h *hmap, it *hiter) {
 if raceenabled && h != nil {
  callerpc := getcallerpc()
  racereadpc(unsafe.Pointer(h), callerpc, funcPC(mapiterinit))
 }

 if h == nil || h.count == 0 {
  return
 }

 if unsafe.Sizeof(hiter{})/sys.PtrSize != 12 {
  throw("hash_iter size incorrect") // see cmd/compile/internal/gc/reflect.go
 }
 it.t = t
 it.h = h

 it.B = h.B
 it.buckets = h.buckets
 if t.bucket.ptrdata == 0 {
  h.createOverflow()
  it.overflow = h.extra.overflow
  it.oldoverflow = h.extra.oldoverflow
 }

 // 定位到随机位置
 r := uintptr(fastrand())
 if h.B > 31-bucketCntBits {
  r += uintptr(fastrand()) << 31
 }
 it.startBucket = r & bucketMask(h.B)
 it.offset = uint8(r >> h.B & (bucketCnt - 1))

 it.bucket = it.startBucket

 if old := h.flags; old&(iterator|oldIterator) != iterator|oldIterator {
  atomic.Or8(&h.flags, iterator|oldIterator)
 }

 mapiternext(it)
}

func mapiternext(it *hiter) {
 h := it.h
 if raceenabled {
  callerpc := getcallerpc()
  racereadpc(unsafe.Pointer(h), callerpc, funcPC(mapiternext))
 }
 if h.flags&hashWriting != 0 {
  throw("concurrent map iteration and map write")
 }
 t := it.t
 bucket := it.bucket
 b := it.bptr
 i := it.i
 checkBucket := it.checkBucket

next:
 if b == nil {
  if bucket == it.startBucket && it.wrapped {
   // end of iteration
   it.key = nil
   it.elem = nil
   return
  }
  if h.growing() && it.B == h.B {
   oldbucket := bucket & it.h.oldbucketmask()
   b = (*bmap)(add(h.oldbuckets, oldbucket*uintptr(t.bucketsize)))
   if !evacuated(b) {
    checkBucket = bucket
   } else {
    b = (*bmap)(add(it.buckets, bucket*uintptr(t.bucketsize)))
    checkBucket = noCheck
   }
  } else {
   b = (*bmap)(add(it.buckets, bucket*uintptr(t.bucketsize)))
   checkBucket = noCheck
  }
  bucket++
  if bucket == bucketShift(it.B) {
   bucket = 0
   it.wrapped = true
  }
  i = 0
 }
 for ; i < bucketCnt; i++ {
  offi := (i + it.offset) & (bucketCnt - 1)
  if isEmpty(b.tophash[offi]) || b.tophash[offi] == evacuatedEmpty {
   continue
  }
  k := add(unsafe.Pointer(b), dataOffset+uintptr(offi)*uintptr(t.keysize))
  if t.indirectkey() {
   k = *((*unsafe.Pointer)(k))
  }
  e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+uintptr(offi)*uintptr(t.elemsize))
  if checkBucket != noCheck && !h.sameSizeGrow() {
   if t.reflexivekey() || t.key.equal(k, k) {
    hash := t.hasher(k, uintptr(h.hash0))
    if hash&bucketMask(it.B) != checkBucket {
     continue
    }
   } else {
    if checkBucket>>(it.B-1) != uintptr(b.tophash[offi]&1) {
     continue
    }
   }
  }
  if (b.tophash[offi] != evacuatedX && b.tophash[offi] != evacuatedY) ||
   !(t.reflexivekey() || t.key.equal(k, k)) {
   it.key = k
   if t.indirectelem() {
    e = *((*unsafe.Pointer)(e))
   }
   it.elem = e
  } else {
   rk, re := mapaccessK(t, h, k)
   if rk == nil {
    continue // key has been deleted
   }
   it.key = rk
   it.elem = re
  }
  it.bucket = bucket
  if it.bptr != b { // avoid unnecessary write barrier; see issue 14921
   it.bptr = b
  }
  it.i = i + 1
  it.checkBucket = checkBucket
  return
 }
 b = b.overflow(t)
 i = 0
 goto next
}

管道

基本概念

不要通过共享内存来通信，而要通过通信来实现内存共享。 Do not communicate by sharing memory; instead, share memory by communicating. 

 chan T // 声明一个双向通道
chan<- T // 声明一个只能用于发送的通道
<-chan T // 声明一个只能用于接收的通道

ch := make(chan int) //初始化一个无缓冲区的int类型通道
ch := make(chan int, 3) //初始化一个容量为3有缓冲区的int类型通道

示例 

 func goRoutineA(a <-chan int) {
    val := <-a
    fmt.Println("goRoutineA received the data", val)
}
func main() {
    ch := make(chan int)
    go goRoutineA(ch)
    time.Sleep(time.Second * 1)
}

数据结构 

 // 管道
type hchan struct {
 qcount   uint           // 元素的数量
 dataqsiz uint           // 缓冲区的大小
 buf      unsafe.Pointer // 指向缓冲区的指针
 elemsize uint16 // 每个元素的大小
 closed   uint32 // 管道状态
 elemtype *_type // 元素类型
 sendx    uint   // 发送的索引
 recvx    uint   // 接收的索引
 recvq    waitq  // 等待接收的队列
 sendq    waitq  // 等待发送的队列

 lock mutex //互斥锁
}

// 等待队列
type waitq struct {
 first *sudog
 last  *sudog
}


type sudog struct {
 g *g

 next *sudog
 prev *sudog
 elem unsafe.Pointer // data element (may point to stack)
        ...

 isSelect bool
 
 ...
 c        *hchan // channel
}

func goroutineA(a <-chan int) {
    val := <- a
    fmt.Println("G1 received data: ", val)
    return
}

func goroutineB(b <-chan int) {
    val := <- b
    fmt.Println("G2 received data: ", val)
    return
}

func main() {
    ch := make(chan int)
    go goroutineA(ch)
    go goroutineB(ch)
    ch <- 3
    time.Sleep(time.Second)
}

chan的创建 

 const (
 maxAlign  = 8
 hchanSize = unsafe.Sizeof(hchan{}) + uintptr(-int(unsafe.Sizeof(hchan{}))&(maxAlign-1))
)

func makechan(t *chantype, size int) *hchan {
 elem := t.elem

 // ... 忽略检查和对齐
        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
  c = (*hchan)(mallocgc(hchanSize, nil, true))
  // 用于竞态检测
  c.buf = c.raceaddr()
 case elem.ptrdata == 0:
  // 元素不包含指针，一次调用分配chan和缓冲区
  c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
  c.buf = add(unsafe.Pointer(c), hchanSize)
 default:
  // 元素包含指针，两次调用分配
  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)

 return c
}

chan的接收 

 ch := make(chan int, 3)
ch <- 1
x := <-ch
fmt.Println(x)

ch <- 2
y, ok := <-ch
if ok {
    fmt.Println(y)
}

//go:nosplit
func chanrecv1(c *hchan, elem unsafe.Pointer) {
 chanrecv(c, elem, true)
}

//go:nosplit
func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {
 _, received = chanrecv(c, elem, true)
 return
}

// 判断管道是否为空
// 1. 非缓冲型，等待发送列队sendq里没有goroutine在等待
// 2. 缓冲型，但 buf 里没有元素
func empty(c *hchan) bool {
 if c.dataqsiz == 0 {
  return atomic.Loadp(unsafe.Pointer(&c.sendq.first)) == nil
 }
 return atomic.Loaduint(&c.qcount) == 0
}

// 找到缓冲区第i个位置
func chanbuf(c *hchan, i uint) unsafe.Pointer {
 return add(c.buf, uintptr(i)*uintptr(c.elemsize))
}

// 管道接收
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
 if c == nil {
     // 如果不阻塞，直接返回 (false, false)
  if !block {
   return
  }
  // 否则，接收一个nil的 channel，goroutine挂起
  gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
  // 不会执行到这里
  throw("unreachable")
 }
 
        // 在非阻塞模式下，快速检测到失败，不用获取锁，快速返回
 if !block && empty(c) {
     // 如果block == false，且没有元素就绪，且管道未关闭，则返回(false, false)
  if atomic.Load(&c.closed) == 0 {
   return
  }
  
  // 如果c关闭了，将ep置0，返回(true, false)
  if empty(c) {
   // ... 忽略竞态检查
   
   if ep != nil {
       // 从一个已关闭的 channel 执行接收操作，且未忽略返回值
                // 那么接收的值将是一个该类型的零值
                // typedmemclr 根据类型清理相应地址的内存
    typedmemclr(c.elemtype, ep)
   }
   // 从一个已关闭的 channel 接收，selected 会返回true
   return true, false
  }
 }

 var t0 int64
 if blockprofilerate > 0 {
  t0 = cputicks()
 }
 
 lock(&c.lock)
 if c.closed != 0 && c.qcount == 0 {
  // ... 忽略竞态检查
  unlock(&c.lock)
  if ep != nil {
   typedmemclr(c.elemtype, ep)
  }
  return true, false
 }
 //等待发送队列里有goroutine存在，调用recv函数处理
 if sg := c.sendq.dequeue(); sg != nil {
  recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
  return true, true
 }

        // 缓冲型，buf里有元素，可以正常接收
 if c.qcount > 0 {
  // 直接从循环数组里找到要接收的元素
  qp := chanbuf(c, c.recvx)
  // ... 忽略竞态检查
  if ep != nil {
   typedmemmove(c.elemtype, ep, qp)
  }
  // 清理掉循环数组里相应位置的值
  typedmemclr(c.elemtype, qp)
  // 接收游标向前移动
  c.recvx++
  // 接收游标归零
  if c.recvx == c.dataqsiz {
   c.recvx = 0
  }
  // buf 数组里的元素个数减 1
  c.qcount--
  unlock(&c.lock)
  return true, true
 }
 if !block {
     // 非阻塞接收，解锁。selected返回false，因为没有接收到值
  unlock(&c.lock)
  return false, false
 }

 // 接下来就是处理要被阻塞的情况，构造一个sudog
 gp := getg()
 mysg := acquireSudog()
 mysg.releasetime = 0
 if t0 != 0 {
  mysg.releasetime = -1
 }
 // 待接收数据的地址保存下来
 mysg.elem = ep
 mysg.waitlink = nil
 gp.waiting = mysg
 mysg.g = gp
 mysg.isSelect = false
 mysg.c = c
 gp.param = nil
 
 // 进入channel 的等待接收队列
 c.recvq.enqueue(mysg)
 
 // 将当前 goroutine 挂起
 atomic.Store8(&gp.parkingOnChan, 1)
 gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)

 // 被唤醒了，接着从这里继续执行一些扫尾工作
 if mysg != gp.waiting {
  throw("G waiting list is corrupted")
 }
 gp.waiting = nil
 gp.activeStackChans = false
 if mysg.releasetime > 0 {
  blockevent(mysg.releasetime-t0, 2)
 }
 success := mysg.success
 gp.param = nil
 mysg.c = nil
 releaseSudog(mysg)
 return true, success
}

chanrecv接收管道c，将接收到的数据写入到ep 如果ep为空，则忽略接收到的数据 如果block == false，且没有元素就绪，返回(false, false) 否则，如果c关闭了，将ep置0，返回(true, false) 否则，将接收到的数据填充到ep中，返回(true, true) 

 func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
 if c.dataqsiz == 0 {
  // ... 忽略竞态检查
  if ep != nil {
   // 直接拷贝数据，从 sender goroutine -> receiver goroutine
   recvDirect(c.elemtype, sg, ep)
  }
 } else {
  // 缓冲型的 channel，但 buf 已满。
         // 将循环数组 buf 队首的元素拷贝到接收数据的地址
         // 将发送者的数据入队。实际上这时 revx 和 sendx 值相等
         // 找到接收游标
  qp := chanbuf(c, c.recvx)
  // ... 忽略竞态检查
  // 将接收游标处的数据拷贝给接收者
  if ep != nil {
   typedmemmove(c.elemtype, ep, qp)
  }
  // 将发送者数据拷贝到 buf
  typedmemmove(c.elemtype, qp, sg.elem)
  // 更新游标值
  c.recvx++
  if c.recvx == c.dataqsiz {
   c.recvx = 0
  }
  c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
 }
 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)
}

func recvDirect(t *_type, sg *sudog, dst unsafe.Pointer) {
 src := sg.elem
 typeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.size)
 memmove(dst, src, t.size)
}

管道的发送 

 //判断缓冲区是否满了，有两种可能
// 1. channel 是非缓冲型的，且等待接收队列里没有 goroutine
// 2. channel 是缓冲型的，但循环数组已经装满了元素
func full(c *hchan) bool {
 if c.dataqsiz == 0 {
  return c.recvq.first == nil
 }
 return c.qcount == c.dataqsiz
}

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
 if c == nil {
  if !block {
   return false
  }
  gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
  throw("unreachable")
 }

 // ... 忽略竞态检查

 // 对于不阻塞的 send，快速检测失败场景
 // 如果channel未关闭且channel没有多余的缓冲空间，返回false
 if !block && c.closed == 0 && full(c) {
  return false
 }

 var t0 int64
 if blockprofilerate > 0 {
  t0 = cputicks()
 }

 //加锁
 lock(&c.lock)

        //尝试向已经关闭的channel发送数据会触发panic
 if c.closed != 0 {
  unlock(&c.lock)
  panic(plainError("send on closed channel"))
 }

        // 如果接收队列里有goroutine，直接将要发送的数据拷贝到接收goroutine
 if sg := c.recvq.dequeue(); sg != nil {
  send(c, sg, ep, func() { unlock(&c.lock) }, 3)
  return true
 }

        // 对于缓冲型的channel，如果还有缓冲空间
 if c.qcount < c.dataqsiz {
  // qp指向buf的sendx位置
  qp := chanbuf(c, c.sendx)
  
  // ... 忽略竞态检查
  // 将数据从ep处拷贝到qp
  typedmemmove(c.elemtype, qp, ep)
  // 发送游标值加 1
  c.sendx++
  // 如果发送游标值等于容量值，游标值归0
  if c.sendx == c.dataqsiz {
   c.sendx = 0
  }
  // 缓冲区的元素数量加一
  c.qcount++
  // 解锁
  unlock(&c.lock)
  return true
 }

 if !block {
  unlock(&c.lock)
  return false
 }

 // channel满了，发送方会被阻塞。接下来会构造一个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
 
 // 当前 goroutine 进入发送等待队列
 c.sendq.enqueue(mysg)

        // 当前 goroutine 被挂起
 atomic.Store8(&gp.parkingOnChan, 1)
 gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
 
 // 保持活跃
 KeepAlive(ep)

 // 从这里开始被唤醒了（channel有机会可以发送了）
 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
  panic(plainError("send on closed channel"))
 }
 return true
}

func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
 // ... 忽略竞态检查
 
 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()
 }
 goready(gp, skip+1)
}

func sendDirect(t *_type, sg *sudog, src unsafe.Pointer) {
 // src 在当前 goroutine 的栈上，dst 是另一个 goroutine 的栈

        // 直接进行内存"搬迁"
        // 如果目标地址的栈发生了栈收缩，当我们读出了 sg.elem 后
        // 就不能修改真正的 dst 位置的值了
        // 因此需要在读和写之前加上一个屏障
 dst := sg.elem
 typeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.size)
 memmove(dst, src, t.size)
}

管道的关闭 

 func closechan(c *hchan) {
    // 关闭空channel会触发panic
 if c == nil {
  panic(plainError("close of nil channel"))
 }

 lock(&c.lock)
 if c.closed != 0 {
  unlock(&c.lock)
  // 关闭已经关闭的channel会触发panic
  panic(plainError("close of closed channel"))
 }

 // ... 忽略竞态检查

        //设置管道已关闭
 c.closed = 1

 var glist gList

 // 将channel所有等待接收队列的里sudog释放
 for {
  sg := c.recvq.dequeue()
  if sg == nil {
   break
  }
  
  // 如果elem不为空，说明此receiver未忽略接收数据
                // 给它赋一个相应类型的零值
  if sg.elem != nil {
   typedmemclr(c.elemtype, sg.elem)
   sg.elem = nil
  }
  if sg.releasetime != 0 {
   sg.releasetime = cputicks()
  }
  
  gp := sg.g
  gp.param = unsafe.Pointer(sg)
  sg.success = false
  // ... 忽略竞态检查
  
  glist.push(gp)
 }

 // 将channel所有等待发送队列的里sudog释放
 for {
  sg := c.sendq.dequeue()
  if sg == nil {
   break
  }
  sg.elem = nil
  if sg.releasetime != 0 {
   sg.releasetime = cputicks()
  }
  gp := sg.g
  gp.param = unsafe.Pointer(sg)
  sg.success = false
  // ... 忽略竞态检查
  glist.push(gp)
 }
 unlock(&c.lock)

 // 遍历链表执行释放
 for !glist.empty() {
  gp := glist.pop()
  gp.schedlink = 0
  goready(gp, 3)
 }
}

管道的使用

参考文献

  • 《Go语言实战》
  • 《Go语言学习笔记》
  • 《Go并发编程实战》
  • 《Go语言核心编程》
  • 《Go程序设计语言》
  • go语言源码
  • 深度解密Go语言之slice
  • 深度解密Go语言之map
  • 深度解密Go语言之channel
  • Go 语言设计与实现
  • Diving Deep Into The Golang Channels
  • 并发与并行 · Concurrency in Go 中文笔记 · 看云

出处:

相关文章