Hedon

# 9. map

# 9.1 冲突

  • 开放寻址法
    • 线性探测法
    • 平方探测法
  • 拉链法
  • 再哈希法
  • 溢出表法

Go 是拉链法 + 溢出表法

# 9.2 底层

  • runtime/map.go

    // hmap
type hmap struct {
	count     int // # live cells == size of map.  Must be first (used by len() builtin)
	flags     uint8
	B         uint8  // 桶的个数 = 2^B
	noverflow uint16 // 
	hash0     uint32 // hash 种子，做哈希的时候需要用

	buckets    unsafe.Pointer // 桶，指向一个 []bmap
	oldbuckets unsafe.Pointer // 旧桶，也是指向一个 []bmap
	nevacuate  uintptr        // progress counter for evacuation (buckets less than this have been evacuated)

	extra *mapextra // 指向溢出桶
}

    // bmap
type bmap struct {
	tophash [bucketCnt]uint8   // key 的前8位哈希值
  // 编译时会动态生成，动态生成是为了在不限制 key 和 value 的类型的情况下节省空间
  // bucketCnt keys
  // bucketCnt elems
  // overflow pointer
}

image-20220812091839792

# 9.3 初始化

# 9.3.1 底层源码 makemap

  • make

    m := make(map[int]string, 10)

    底层：调用 runtime/map.go 中的 makemap()

    func makemap(t *maptype, hint int, h *hmap) *hmap {
  
  // 1. 计算预期的 map 大小
	mem, overflow := math.MulUintptr(uintptr(hint), t.bucket.size)
	if overflow || mem > maxAlloc {
		hint = 0
	}

	// 2. 创建一个新的 hmap
	if h == nil {
		h = new(hmap)
	}
	h.hash0 = fastrand()

	// 3. 计算 B
	B := uint8(0)
	for overLoadFactor(hint, B) {
		B++
	}
	h.B = B

	if h.B != 0 {
    // 4. 根据 B 创建桶和溢出桶
		var nextOverflow *bmap
		h.buckets, nextOverflow = makeBucketArray(t, h.B, nil)
		if nextOverflow != nil {
      // 5. 将溢出桶的数据存在 mapextra 中
			h.extra = new(mapextra)
			h.extra.nextOverflow = nextOverflow
		}
	}

	return h
}

# 9.3.2 总结图

image-20220812093130156

  • 字面量（底层还是先调用 makemap，然后再做赋值）
    • 元素少于 25 个时，一个一个简单赋值
    • 元素多个 25 个时，转为循环赋值

# 9.4 访问

# 9.4.1 底层源码 mapaccess

底层调用了 runtime/map.go 中的 mapaccess1() 或者 mapaccess2 方法：

  • v := m[k] 调用 mapaccess1()
  • v,k := m[k] 调用 mapaccess2()
func mapaccess1(t *maptype, h *hmap, key unsafe.Pointer) unsafe.Pointer {
	...
  
  // 1. 空 map 直接返回零值
	if h == nil || h.count == 0 {
		if t.hashMightPanic() {
			t.hasher(key, 0) // see issue 23734
		}
		return unsafe.Pointer(&zeroVal[0])
	}
  
  // 2. map 不允许并发访问，直接 panic
	if h.flags&hashWriting != 0 {
		throw("concurrent map read and map write")
	}
  
  // 3. 根据 hash0 计算 key 的哈希值
	hash := t.hasher(key, uintptr(h.hash0))
  // 4. 由 key 的哈希值和 B 确定桶
	m := bucketMask(h.B)
	b := (*bmap)(add(h.buckets, (hash&m)*uintptr(t.bucketsize)))
  // 5. 如果 oldbuckets 不为空，说明正在扩容
	if c := h.oldbuckets; c != nil {
    // 5.1 判断是否是等量扩容，不是的话就翻倍
		if !h.sameSizeGrow() {
			m >>= 1
		}
    // 5.2 寻找 key 在 oldbuckets 中的位置
		oldb := (*bmap)(add(c, (hash&m)*uintptr(t.bucketsize)))
    // 5.3 如果 oldbuckets 还未完成迁移，就在 oldbuckets 中寻找
		if !evacuated(oldb) {
			b = oldb
		}
	}
  
  // 6. 计算 tophash ：hash 的高八位
	top := tophash(hash)
  
  // 7. 遍历桶：先遍历正常桶，第一轮 for 循环不成功，再以此遍历 overflow 溢出桶
bucketloop:
	for ; b != nil; b = b.overflow(t) {
    // 7.1 一个桶里面有 8 个值，遍历这 8 个值，看看能不能找到对应的 tophash
		for i := uintptr(0); i < bucketCnt; i++ {
      // 7.1.1 找不到，再找
			if b.tophash[i] != top {
				if b.tophash[i] == emptyRest {
					break bucketloop
				}
				continue
			}
      // 7.1.2 找到了，就拿出 key 来比较
			k := add(unsafe.Pointer(b), dataOffset+i*uintptr(t.keysize))
			if t.indirectkey() {
				k = *((*unsafe.Pointer)(k))
			}
      // 7.1.3 如果 key 是对的，就取出值返回
			if t.key.equal(key, k) {
				e := add(unsafe.Pointer(b), dataOffset+bucketCnt*uintptr(t.keysize)+i*uintptr(t.elemsize))
				if t.indirectelem() {
					e = *((*unsafe.Pointer)(e))
				}
				return e
			}
      // 7.1.4 key 不对，就再找
		}
	}
  
  // 8. 找不到，返回零值
	return unsafe.Pointer(&zeroVal[0])
}

# 9.4.2 总结图

image-20220812171408648

# 9.5 写入

底层调用 runtime/map.go 的 mapassign() 方法，跟 mapaccess() 非常像，只不过：

  1. 先找找看 key 在不在，在的话，则覆盖新的 value；
  2. 如果 key 不在，则插入新的 key 和 value，这里则需要考虑是否需要扩容了；

# 9.6 扩容

# 9.6.1 原因

  • map 溢出桶太多时，会拉太多链，严重影响 map 的性能；
  • runtime.mapassign() 可能会触发扩容的情况：
    • 装载因子超过 6.5（平均每个槽 6.5 个 key）
    • 溢出桶数量超过了普通桶
  • 两种扩容：
    • 等量扩容：数据不多但是溢出桶太多了（整理）
    • 翻倍扩容：数据太多了

# 9.6.2 底层源码

# 9.6.2.1 hashGrow()

底层调用 runtime/map.go 的 hashGrow()。

func hashGrow(t *maptype, h *hmap) {
	// 1. 判断是等量扩容还是翻倍扩容
	bigger := uint8(1)
	if !overLoadFactor(h.count+1, h.B) {
		bigger = 0
		h.flags |= sameSizeGrow
	}
  // 2. oldbuckets 指向原来的桶
	oldbuckets := h.buckets
  // 3. buckets 指向新建的桶
	newbuckets, nextOverflow := makeBucketArray(t, h.B+bigger, nil)
	
	flags := h.flags &^ (iterator | oldIterator)
	if h.flags&iterator != 0 {
		flags |= oldIterator
	}
	// 4. 重置一些参数
	h.B += bigger
	h.flags = flags
	h.oldbuckets = oldbuckets
	h.buckets = newbuckets
	h.nevacuate = 0
	h.noverflow = 0
  
  // 5. 转移原来溢出桶的数据到 oldoverflow
	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
	}

  // 6. 并不立刻对桶数据进行迁移，只在写的时候，才迁移写到的那个桶的所有数据到 newbucktes，读还是读 oldbucktes
  // 		实际迁移的时候调用的是 growWork() 和 evacuate()。
}

# 9.6.2.2 evacuate()

func evacuate(t *maptype, h *hmap, oldbucket uintptr) {
  
  // 1. 计算 oldbuckets 指针
	b := (*bmap)(add(h.oldbuckets, oldbucket*uintptr(t.bucketsize)))
  // 2. 计算此次扩容之前桶数组的大小
	newbit := h.noldbuckets()
	if !evacuated(b) {
		// x 和 y 分别指代需要迁移的桶的两个目的地：一个的相对位置与旧桶的相对位置相同；如果是翻倍扩容：另一个桶 y 的相对位置则是旧桶的相对位置加上旧桶数组的大小。
		// 翻倍扩容时，旧桶的键值对将分散到两个桶中；
    // 等量扩容时，旧桶中所有值都将迁移到对应其位置的一个桶中。（如果有溢出，则在该桶后面添加溢出桶）
		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))
		
    // 3. 
		if !h.sameSizeGrow() {
			
			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))
		}

    // 4. 迁移正常桶及其下面的溢出桶
		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))
			}
		}

    // 5. 迁移完成，释放 oldbuckets 指针帮助 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)
		}
	}

  // 6. 维护迁移状态 nevacuate
  // 		h.nevacuate是迁移计数器，此指针之前的所有桶已被迁移。
	// 		如果当前迁移的桶是 h.nevacuate 指示的桶，则需要对 h.nevacuate 指针向前移动
  // 		如果 nevacuate 的指向已经超出旧桶数组的最高下标，说明迁移完成，可以释放对旧桶数组的引用
	if oldbucket == h.nevacuate {
		advanceEvacuationMark(h, t, newbit)
	}
}

# 9.6.2.3 growWork()

func growWork(t *maptype, h *hmap, bucket uintptr) {
	// 迁移即将使用的桶
	evacuate(t, h, bucket&h.oldbucketmask())

	// 每次都多迁移一个桶，加快迁移的进程，这样可以早点释放 oldbuckets
	if h.growing() {
		evacuate(t, h, h.nevacuate)
	}
}

# 9.6.3 总结图

image-20220812182343329

# 9.7 并发

# 9.7.1 问题

前面分析 map 的访问的时候，我们已经知道 map 明确严禁并发读写。比如：

  • 一个协程在读 map，另一个协程在驱逐，就可能出现问题。

所以如果我们非要在并发情况下使用 map 的话，就需要用 mutex 加锁了，但是这样 map 的性能非常差。

# 9.7.2 解决 —— sync.Map

# 9.7.2.1 底层

  • Map

    type Map struct {
	mu Mutex							// 锁
	read atomic.Value 		// 指向一个 readOnly 结构体的值
	dirty map[any]*entry	// 指向一个 map
	misses int						// 没有命中的个数，即在 read 中读不到的次数
}

  • readOnly

    // readOnly is an immutable struct stored atomically in the Map.read field.
type readOnly struct {
	m       map[any]*entry	// 存储 map 数据
	amended bool 						// 当 dirtyMap 中有 m 没有的元素的时候，amended 值为 true
}

  • entry

    type entry struct {
	p unsafe.Pointer // 万能指针，指向 value
}

image-20220812192409927

# 9.7.2.2 正常读写

  • 走 read，读出 value 或者覆盖 value

image-20220812193406236

# 9.7.2.3 追加

func (m *Map) Store(key, value interface{}) {
	// 1. 先尝试在 read map 中进行写
  read, _ := m.read.Load().(readOnly)
	if e, ok := read.m[key]; ok && e.tryStore(&value) {
		return
	}
	
  // 2. read 中没有对应的 key，那就只能追加了，上锁操作 dirty map
	m.mu.Lock()
  // 3. 再读一遍 read map，因为有可能在我们上锁之前的一瞬间，别的协程将 dirty 提升了
	read, _ = m.read.Load().(readOnly)
  // 4. read map 中有了，说明已经被其他协程 dirty 提升了，
	if e, ok := read.m[key]; ok {
    // 4-1. 判断读出来的 entry 是否已经被标记为 unexpunged(已删除)
		if e.unexpungeLocked() {
			// 4-2. 该 enrty 已被标记为删除，那么就需要将其放到 dirty 中
			m.dirty[key] = e
		}
    // 4-3 读出来
		e.storeLocked(&value)
	} else if e, ok := m.dirty[key]; ok {
    // 5. read map 中还是没有，那就读 dirty map
		e.storeLocked(&value)
	} else {
    // 6. dirty map 中还是没有，那就只能往 dirty map 中追加了
		if !read.amended {
			m.dirtyLocked()
			m.read.Store(readOnly{m: read.m, amended: true})
		}
		m.dirty[key] = newEntry(value)
	}
  // 7. 追加完，解锁
	m.mu.Unlock()
}

// unexpungeLocked 可确保 entry 未标记为已清除。
//
// 如果该 entry 已经被标记为删除了，则必须在解锁 m.mu 之前将其添加到 dirty map 中。
func (e *entry) unexpungeLocked() (wasExpunged bool) {
	return atomic.CompareAndSwapPointer(&e.p, expunged, nil)
}

image-20220812193459329

# 9.7.2.4 追加后的读

func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
  // 1. 先在 read 中找
	read, _ := m.read.Load().(readOnly)
	e, ok := read.m[key]
  // 2. read 中找不到，就在 dirty 中找
	if !ok && read.amended {
    // 4. 上锁
		m.mu.Lock()
    // 5. 再读一次 read map，因为有可能在我们上锁之前的一瞬间，别的协程将 dirty 提升了
		read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
    // 6. 还是没在 read 中找到，就只能在 dirty 中找了
		if !ok && read.amended {
			e, ok = m.dirty[key]
      // 7. misses ++ 并判断是否需要 dirty 提升
			m.missLocked()
		}
    // 8. 解锁
		m.mu.Unlock()
	}
	if !ok {
		return nil, false
	}
	return e.load()
}

image-20220813121131695

# 9.7.2.5 dirty 提升

当 meisses = len(dirty) 的时候，就砍掉 read，将 dirty 提升到 read 的位置。

func (m *Map) missLocked() {
  // 1. 每在 dirty 中查一次，就 misses++
	m.misses++
	if m.misses < len(m.dirty) {
		return
	}
  // 2. 当 misses = len(m.dirty) 的时候，就 dirty 提升
  // 3. 将 dirtymap 赋值给 read map
  //     这里没有指出 amended，但是因为默认零值是 false，所以这里也将 amended 置为 false 了
	m.read.Store(readOnly{m: m.dirty})
  // 4. dirty 先为 nil，当要追加的时候，再来复制 read map
	m.dirty = nil
  // 5. 重置 misses
	m.misses = 0
}

image-20220813120422714

image-20220813123004358

func (m *Map) Store(key, value interface{}) {
	...
	if e, ok := read.m[key]; ok {
		...
	} else if e, ok := m.dirty[key]; ok {
		...
	} else {
    // 第一次追加到 dirty map 的时候，需要判断看 dirty map 之前是否已经被提升了，可能为 nil
    // 如果是 nil 的话，就需要复制 read map
		if !read.amended {
			m.dirtyLocked()
			m.read.Store(readOnly{m: read.m, amended: true})
		}
    // 追加 key
		m.dirty[key] = newEntry(value)
	}
	m.mu.Unlock()
}

// 当 dirty map 为 nil 的时候
// 负责将 read map 复制到 dirty map
func (m *Map) dirtyLocked() {
	if m.dirty != nil {
		return
	}

	read, _ := m.read.Load().(readOnly)
	m.dirty = make(map[interface{}]*entry, len(read.m))
	for k, e := range read.m {
		if !e.tryExpungeLocked() {
			m.dirty[k] = e
		}
	}
}

image-20220813172157448

# 9.7.2.6 删除

  • 正常删除

    // Delete deletes the value for a key.
func (m *Map) Delete(key interface{}) {
	m.LoadAndDelete(key)
}

// LoadAndDelete deletes the value for a key, returning the previous value if any.
// The loaded result reports whether the key was present.
func (m *Map) LoadAndDelete(key interface{}) (value interface{}, loaded bool) {
	read, _ := m.read.Load().(readOnly)
  // 1. 先从 read map 中找
	e, ok := read.m[key]
	...
  // 2. 找到了，就直接在 read map 中删除
	if ok {
		return e.delete()
	}
	return nil, false
}

func (e *entry) delete() (value interface{}, ok bool) {
	for {
		p := atomic.LoadPointer(&e.p)
		if p == nil || p == expunged {
			return nil, false
		}
    // 3. 直接将 *entry 的 Pointer 置为空
		if atomic.CompareAndSwapPointer(&e.p, p, nil) {
			return *(*interface{})(p), true
		}
	}
}

    image-20220813172136456

  • 追加后删除

    func (m *Map) LoadAndDelete(key interface{}) (value interface{}, loaded bool) {
  // 1. 先在 read map 中找
	read, _ := m.read.Load().(readOnly)
	e, ok := read.m[key]
	if !ok && read.amended {
    // 2. 找不到，就上锁，然后在 dirty map 中找
		m.mu.Lock()
    // 3. 找的时候一样，还是再次在 read map 中查一遍，防止有 dirty 提升
		read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
		if !ok && read.amended {
      // 4. read map 中还是没找到，那就在 dirty map 中找，删的时候，还是 *entry.Pointer = nil
			e, ok = m.dirty[key]
			delete(m.dirty, key)
      // 5. misses ++
			m.missLocked()
		}
		m.mu.Unlock()
	}
	if ok {
		return e.delete()
	}
	return nil, false
}

    image-20220813171708610

  • 删除后提升 dirty

    func (m *Map) dirtyLocked() {
	if m.dirty != nil {
		return
	}

	read, _ := m.read.Load().(readOnly)
	m.dirty = make(map[interface{}]*entry, len(read.m))
	for k, e := range read.m {
    // 不复制标记为 expunged 的
		if !e.tryExpungeLocked() {
			m.dirty[k] = e
		}
	}
}

    image-20220813172111143image-20220813172332026

# 9.3.4 总结

  • sync.Map 使用了两个 map，将“普通读写”和“追加”进行分离；
  • 不会引发扩容的操作（查、改）使用 read map；
  • 可能引起扩容的操作（增）使用 dirty map；
上次更新: 8/13/2022, 5:30:58 PM

← 10. interface →