Go Map 1.24新实现

底层结构

源码：src/internal/runtime/maps

使用Directory管理多个table，Directory是 Table的数组 ，大小为2^globalDepth。如果globalDepth=2，那Directory最多有4个表，分为0x00、0x01、0x10、0x11。Go通过key的hash值的前 globalDepth 个bit来选择table。这是一种“extendible hashing”，这是一种动态哈希技术，其核心特点是通过动态调整使用的哈希位数(比如上面提到的globalDepth)来实现渐进式扩容。比如：初始可能只用1位哈希值来区分，需要时可以扩展到用2位，再需要时可以扩展到用3位，以此类推。

关键源码：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39


type Map struct {
	// 已填充槽位的数量（即所有表中的元素总数）。
    // 不包含已删除的槽位。
    // 必须放在结构体的第一个字段
    //（编译器已知，用于 len() 内建函数）。
	used uint64

	// 哈希种子
	seed uintptr

	// 通常dirPtr指向一个tables数组
    // directory [] *Table
    // 该数组的长度（dirLen）为 `1 << globalDepth`。
    // 多个目录项可能指向同一个表。
    // 小 map 优化：如果 map 中的元素数量始终不超过abi.SwissMapGroupSlots=8，则可以完全放入一个单独的 group 中。
    // 在这种情况下，dirPtr 直接指向一个 group。
    // 此时，dirLen 为 0。used 表示该 group 中已使用的槽位数量。
    // 注意：小 map 永远不会有已删除的槽位
    //（因为不存在需要维护的探测序列）。
	dirPtr unsafe.Pointer  
	dirLen int

	// 用于表目录（table directory）查找的比特数。
	globalDepth uint8

	// 在进行目录（directory）查找时，需要从哈希值中右移（丢弃）的比特数。
    // 在 64 位系统上，该值为 64 - globalDepth。
	globalShift uint8

	// writing 是一个标志位，在 map 被写入时通过异或 1（XOR 1）来翻转。
    // 通常在写入期间它被置为 1；
    // 但如果存在多个并发写入者，持续翻转该标志
    // 可以提高双方检测到竞争条件（data race）的概率。
	writing uint8

	// clearSeq 是对 Clear 调用次数的序列计数器。
    // 用于在迭代过程中检测 map 是否被清空（clear）。
	clearSeq uint64
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32


type table struct {
	// 已填充槽位的数量（即表中元素的数量）。
	used uint16

	// 槽位的总数量（始终为 2 的 N 次幂）。
    // 等于 `(groups.lengthMask + 1) * abi.SwissMapGroupSlots`。
	capacity uint16

	// 在无需 rehash 的前提下仍可使用的槽位数。
    // 当 used + tombstones > loadFactor * capacity 时触发 rehash，
    // 计入 tombstones，以防止哈希表被墓碑项过度填充。
    // 该字段表示距离下一次 rehash 还剩余的空槽位数量。
	growthLeft uint16

	// 用于定位到该表的目录查找所需的比特位数。
    // 注意：当目录已经增长但该表还未被分裂时，
    // 该值可能小于 globalDepth。
	localDepth uint8

	// 该表在 Map 目录中的索引。这里指的是目录中指向该表的
    // 第一个索引位置；该表可能对应目录中的多个连续索引。
    // 若该表已过期（不再存在于目录中），则 index 取值为 -1。
	index int

	// groups 为槽位组数组。每个组包含 abi.SwissMapGroupSlots = 8 个
    // 键/值槽位及其控制字节。表的 groups 数组大小固定，
    // 当需要扩容时，会在 rehash 时用新表替换。
    // TODO(prattmic)：键和值交错存储可提高缓存局部性，
    // 但对某些类型会浪费空间（如 uint8 键 + uint64 值）。
    // 可考虑在这些情况下将键集中存放以节省空间。
	groups groupsReference
}

1
2
3
4


type groupsReference struct {
    data unsafe.Pointer //group数组
    lengthMask uint64 //len(group)-1，方便快速求模
}

1
2
3
4
5
6
7
8


type group struct {
    ctrl  ctrlGroup        // 8 字节
    slots [8]slot
}

// 控制字
type ctrlGroup uint64
// 每个control byte 包含状态(1byte---空、删除、占用)+h2(哈希低7位)

特性

多个 directory index 可以指向同一个 table
directory 里存的只是指针

初始化流程

关键源码：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66


func NewMap(mt *abi.SwissMapType, hint uintptr, m *Map, maxAlloc uintptr) *Map {
	if m == nil {
		m = new(Map)
	}

	m.seed = uintptr(rand())

	if hint <= abi.SwissMapGroupSlots {
		// A small map can fill all 8 slots, so no need to increase
		// target capacity.
		//
		// In fact, since an 8 slot group is what the first assignment
		// to an empty map would allocate anyway, it doesn't matter if
		// we allocate here or on the first assignment.
		//
		// Thus we just return without allocating. (We'll save the
		// allocation completely if no assignment comes.)

		// Note that the compiler may have initialized m.dirPtr with a
		// pointer to a stack-allocated group, in which case we already
		// have a group. The control word is already initialized.

		return m
	}

	// Full size map.

	// Set initial capacity to hold hint entries without growing in the
	// average case.
	targetCapacity := (hint * abi.SwissMapGroupSlots) / maxAvgGroupLoad
	if targetCapacity < hint { // overflow
		return m // return an empty map.
	}

	dirSize := (uint64(targetCapacity) + maxTableCapacity - 1) / maxTableCapacity
	dirSize, overflow := alignUpPow2(dirSize)
	if overflow || dirSize > uint64(math.MaxUintptr) {
		return m // return an empty map.
	}

	// Reject hints that are obviously too large.
	groups, overflow := math.MulUintptr(uintptr(dirSize), maxTableCapacity)
	if overflow {
		return m // return an empty map.
	} else {
		mem, overflow := math.MulUintptr(groups, mt.GroupSize)
		if overflow || mem > maxAlloc {
			return m // return an empty map.
		}
	}

	m.globalDepth = uint8(sys.TrailingZeros64(dirSize))
	m.globalShift = depthToShift(m.globalDepth)

	directory := make([]*table, dirSize)

	for i := range directory {
		// TODO: Think more about initial table capacity.
		directory[i] = newTable(mt, uint64(targetCapacity)/dirSize, i, m.globalDepth)
	}

	m.dirPtr = unsafe.Pointer(&directory[0])
	m.dirLen = len(directory)

	return m
}

源码解读：初始化时会根据 hint 的值进入不同的初始化逻辑：

hint<=8，小map

对于小map，指定容量hint<=8，做了优化，不再分配一个table数组，dirPtr直接引用分配 一个group
hint>8

a> 首先计算目标容量 targetCapacity，算出需要多少slot，才能在“平均负载”下容纳 hint 个元素

b> 计算 directory 大小dirSize（table 数）（需要多少个 table，才能容纳 targetCapacity ）

c> 计算总的 group 数量和总内存。

d> 设置 globalDepth / globalShift

e> 分配directory并 创建table（初始化时table内部的localdepth=globalDepth）

f> directory 写入Map（unsafe），最后返回Map。

添加流程

关键源码：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49


func (m *Map) PutSlot(typ *abi.SwissMapType, key unsafe.Pointer) unsafe.Pointer {
	if m.writing != 0 {
		fatal("concurrent map writes")
	}

	hash := typ.Hasher(key, m.seed)

	// Set writing after calling Hasher, since Hasher may panic, in which
	// case we have not actually done a write.
	m.writing ^= 1 // toggle, see comment on writing

	if m.dirPtr == nil {
		m.growToSmall(typ)
	}

	if m.dirLen == 0 {
		if m.used < abi.SwissMapGroupSlots {
			elem := m.putSlotSmall(typ, hash, key)

			if m.writing == 0 {
				fatal("concurrent map writes")
			}
			m.writing ^= 1

			return elem
		}

		// Can't fit another entry, grow to full size map.
		//
		// TODO(prattmic): If this is an update to an existing key then
		// we actually don't need to grow.
		m.growToTable(typ)
	}

	for {
		idx := m.directoryIndex(hash)
		elem, ok := m.directoryAt(idx).PutSlot(typ, m, hash, key)
		if !ok {
			continue
		}

		if m.writing == 0 {
			fatal("concurrent map writes")
		}
		m.writing ^= 1

		return elem
	}
}

源码解读：

首先进行并发写检测，保障线程安全。
然后计算key的hash，并标记写入状态，防止并发写。
判断当前map是否已经分配了，如果没有分配，会先分配一个group（小map）。

a. 如果当前map是小map
1. 如果是小表且未满，直接插入，返回元素槽指针。
2. 如果小表已满，升级为大表（分配table并迁移数据）
b. 如果当前map是大map
1. 根据H1计算目录索引，定位table（与查找相同）
2. 调用大map的插入函数，若table扩容则重试
3. 插入成功返回与元素槽指针
再次检测并发写，防止并发写破坏数据。

小map写入细节

首先获取group的引用，并结合H2对所有控制字进行比对。
若存在相同的，比对是否存在相同key，如果存在则返回对应value槽指针。
不存在相同的，查找空槽，准备写入新key，如果没有空槽，说明并发写入或者逻辑错误。
写入key和value。
更新控制字和used计数。
返回value槽指针。

大map写入细节

首先根据hash的高位H1和group的数量定位探查序列。
定义变量，记录第一个已删除槽，以便后续使用。
探查循环，遍历group，查找与H2匹配的槽。
若有匹配的槽，对比key，若key需要更新，则拷贝新key
无匹配槽，查找空槽或已删除槽（tombstone）。若有空槽且growthleft>0，优先复用空槽，设置控制字，更新计数。若growthleft耗尽（表已满），触发rehash（扩容或分裂），返回nil，false，提示上层重试。

搜索流程

关键源码：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65


// Get performs a lookup of the key that key points to. It returns a pointer to
// the element, or false if the key doesn't exist.
func (m *Map) Get(typ *abi.SwissMapType, key unsafe.Pointer) (unsafe.Pointer, bool) {
	return m.getWithoutKey(typ, key)
}

func (m *Map) getWithoutKey(typ *abi.SwissMapType, key unsafe.Pointer) (unsafe.Pointer, bool) {
	if m.Used() == 0 {
		return nil, false
	}

	if m.writing != 0 {
		fatal("concurrent map read and map write")
	}

	hash := typ.Hasher(key, m.seed)

	if m.dirLen == 0 {
		_, elem, ok := m.getWithKeySmall(typ, hash, key)
		return elem, ok
	}

	idx := m.directoryIndex(hash)
	return m.directoryAt(idx).getWithoutKey(typ, hash, key)
}

func (m *Map) directoryIndex(hash uintptr) uintptr {
	if m.dirLen == 1 {
		return 0
	}
	return hash >> (m.globalShift & 63)
}

func (t *table) getWithoutKey(typ *abi.SwissMapType, hash uintptr, key unsafe.Pointer) (unsafe.Pointer, bool) {
	seq := makeProbeSeq(h1(hash), t.groups.lengthMask)
	for ; ; seq = seq.next() {
		g := t.groups.group(typ, seq.offset)

		match := g.ctrls().matchH2(h2(hash))

		for match != 0 {
			i := match.first()

			slotKey := g.key(typ, i)
			if typ.IndirectKey() {
				slotKey = *((*unsafe.Pointer)(slotKey))
			}
			if typ.Key.Equal(key, slotKey) {
				slotElem := g.elem(typ, i)
				if typ.IndirectElem() {
					slotElem = *((*unsafe.Pointer)(slotElem))
				}
				return slotElem, true
			}
			match = match.removeFirst()
		}

		match = g.ctrls().matchEmpty()
		if match != 0 {
			// Finding an empty slot means we've reached the end of
			// the probe sequence.
			return nil, false
		}
	}
}

源码解读：

根据map的 used 字段快速判断，如果是空map，直接返回nil。
其次对map的 并发读写 进行检测，存在并发读写直接报错。
计算key的hash（H1-高57位，用于定位group。H2-低7位，用于对比控制字）值。
根据dirLen判断是否是小map。

a. 如果是小map
1. 进入小map的搜索流程，直接从初始化中的单个group中进行查找
2. 获取当前hash的H2，然后遍历group，逐个对比控制字中的H2部分
3. 不匹配就继续下一个，匹配之后，再继续对key校验（只用7位进行校验，存在hash冲突的可能性，虽然很低，1/128）
4. 若key相同，找到val直接返回，若遍历结束也没有匹配项，返回nil结束。
b. 如果不是小map
1. 先根据key的hash的高位（globalDepth）计算出当前 map 的“目录数组”(directory)中的索引，从而定位到应该查找或操作的 table（分表）。（如果目录长度为 1，说明 map 只有一个 table（未分表），所有 key 都在这一个 table 里。否则说明已经分表，用hash的高位来选择table。例如globalDepth = 3，则目录长度为8，globalShift = 61，目录索引为hash»61，即hash的最高3位）
2. 根据hash的 H1 部分计算出table内部遍历的 group的起始位置。（起始位置offset：uint64(H1) & mask，用 h1(hash) 的低若干位（由 mask 决定）作为 group 的初始索引。其中mask = group 数量 - 1。例如group数量为8，mask=0b111，则只取H1的最低3位。）
3. 然后开始查找group内的元素，通过H2与group的8个控制字做并行匹配（复制H2为8份），如果存在匹配的，则继续对key进行比较，如果遇到空槽，说明key不存在，返回nil,false，如果没有找到且没有空槽，继续查找下一个group。

关键点

q：为什么遇到空槽代表不存在key？
a：在插入新key时会沿着探查序列查找第一个空槽或者已删除的空槽，查找时同样沿着探查序列查找，如果key曾经被插入过，一定会在遇到空槽之前被找到（插入时遇到空槽就插入了）

删除过程

关键源码：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37


func (m *Map) Delete(typ *abi.SwissMapType, key unsafe.Pointer) {
	if m == nil || m.Used() == 0 {
		if err := mapKeyError(typ, key); err != nil {
			panic(err) // see issue 23734
		}
		return
	}

	if m.writing != 0 {
		fatal("concurrent map writes")
	}

	hash := typ.Hasher(key, m.seed)

	// Set writing after calling Hasher, since Hasher may panic, in which
	// case we have not actually done a write.
	m.writing ^= 1 // toggle, see comment on writing

	if m.dirLen == 0 {
		m.deleteSmall(typ, hash, key)
	} else {
		idx := m.directoryIndex(hash)
		m.directoryAt(idx).Delete(typ, m, hash, key)
	}

	if m.used == 0 {
		// Reset the hash seed to make it more difficult for attackers
		// to repeatedly trigger hash collisions. See
		// https://go.dev/issue/25237.
		m.seed = uintptr(rand())
	}

	if m.writing == 0 {
		fatal("concurrent map writes")
	}
	m.writing ^= 1
}

源码解读：

判断当前map是否为空，如果为空，则删除操作无效。同时检查key的类型是否合法，如果不合法直接 panic。
检测并发写状态，保证线程安全。
计算key的hash，如果key类型不匹配则panic。
设置并发写标志。
判断map的大小。

a. 如果是小map
1. 获取group引用，匹配H2。
2. 若有匹配的，再对比key。
3. 找到之后，计数-1，然后清理key和value。
4. 标记slot为空，小map没有探查序列，不需要tombstone直接将控制字设为empty。
5. 如果没有匹配的key，则什么都不做。
b. 如果是大map
1. 通过H1和表容量构造探查序列。
2. 结合H2遍历group，若存在H2匹配的，再对比key。
3. 清理key和value。
4. 设置控制字，如果当前group还有空槽，直接将ctrl（控制字）设置为empty，并回收growthleft（可插入槽数）。否则，设置为deleted（墓碑），墓碑会在rehash时被清理。
5. 如果group没找到目标key，遇到空槽则终止。如果group有空槽，说明key不存在，直接返回。

扩容过程：

小map扩容

扩容节点：小map只能存放8个元素，当插入第9个元素时会触发扩容。调用growToTable

关键源码：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38


func (m *Map) growToTable(typ *abi.SwissMapType) {
	tab := newTable(typ, 2*abi.SwissMapGroupSlots, 0, 0)

	g := groupReference{
		data: m.dirPtr,
	}

	for i := uintptr(0); i < abi.SwissMapGroupSlots; i++ {
		if (g.ctrls().get(i) & ctrlEmpty) == ctrlEmpty {
			// Empty
			continue
		}

		key := g.key(typ, i)
		if typ.IndirectKey() {
			key = *((*unsafe.Pointer)(key))
		}

		elem := g.elem(typ, i)
		if typ.IndirectElem() {
			elem = *((*unsafe.Pointer)(elem))
		}

		hash := typ.Hasher(key, m.seed)

		tab.uncheckedPutSlot(typ, hash, key, elem)
	}

	directory := make([]*table, 1)

	directory[0] = tab

	m.dirPtr = unsafe.Pointer(&directory[0])
	m.dirLen = len(directory)

	m.globalDepth = 0
	m.globalShift = depthToShift(m.globalDepth)
}

源码解读：

分配新表，新表容量为16，原来2倍，localdepth\globaldepth都为0，index为0。
遍历原来的8个slot，如果非空，重新计算hash，插入到新表。
构建目录指向新表，长度为1。
map进入表模式，后续插入查找都走table逻辑。

大map扩容

扩容节点：在插入流程中，如果当前为大表，且在插入时发现没有空槽，就会出发rehash。

关键源码：

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121


func (t *table) rehash(typ *abi.SwissMapType, m *Map) {
	newCapacity := 2 * t.capacity
	if newCapacity <= maxTableCapacity {
		t.grow(typ, m, newCapacity)
		return
	}
	t.split(typ, m)
}


func (t *table) grow(typ *abi.SwissMapType, m *Map, newCapacity uint16) {
	newTable := newTable(typ, uint64(newCapacity), t.index, t.localDepth)

	if t.capacity > 0 {
		for i := uint64(0); i <= t.groups.lengthMask; i++ {
			g := t.groups.group(typ, i)
			for j := uintptr(0); j < abi.SwissMapGroupSlots; j++ {
				if (g.ctrls().get(j) & ctrlEmpty) == ctrlEmpty {
					// Empty or deleted
					continue
				}

				key := g.key(typ, j)
				if typ.IndirectKey() {
					key = *((*unsafe.Pointer)(key))
				}

				elem := g.elem(typ, j)
				if typ.IndirectElem() {
					elem = *((*unsafe.Pointer)(elem))
				}

				hash := typ.Hasher(key, m.seed)

				newTable.uncheckedPutSlot(typ, hash, key, elem)
			}
		}
	}

	newTable.checkInvariants(typ, m)
	m.replaceTable(newTable)
	t.index = -1
}

func (t *table) split(typ *abi.SwissMapType, m *Map) {
	localDepth := t.localDepth
	localDepth++

	// TODO: is this the best capacity?
	left := newTable(typ, maxTableCapacity, -1, localDepth)
	right := newTable(typ, maxTableCapacity, -1, localDepth)

	// Split in half at the localDepth bit from the top.
	mask := localDepthMask(localDepth)

	for i := uint64(0); i <= t.groups.lengthMask; i++ {
		g := t.groups.group(typ, i)
		for j := uintptr(0); j < abi.SwissMapGroupSlots; j++ {
			if (g.ctrls().get(j) & ctrlEmpty) == ctrlEmpty {
				// Empty or deleted
				continue
			}

			key := g.key(typ, j)
			if typ.IndirectKey() {
				key = *((*unsafe.Pointer)(key))
			}

			elem := g.elem(typ, j)
			if typ.IndirectElem() {
				elem = *((*unsafe.Pointer)(elem))
			}

			hash := typ.Hasher(key, m.seed)
			var newTable *table
			if hash&mask == 0 {
				newTable = left
			} else {
				newTable = right
			}
			newTable.uncheckedPutSlot(typ, hash, key, elem)
		}
	}

	m.installTableSplit(t, left, right)
	t.index = -1
}

func (m *Map) installTableSplit(old, left, right *table) {
	if old.localDepth == m.globalDepth {
		// No room for another level in the directory. Grow the
		// directory.
		newDir := make([]*table, m.dirLen*2)
		for i := range m.dirLen {
			t := m.directoryAt(uintptr(i))
			newDir[2*i] = t
			newDir[2*i+1] = t
			// t may already exist in multiple indicies. We should
			// only update t.index once. Since the index must
			// increase, seeing the original index means this must
			// be the first time we've encountered this table.
			if t.index == i {
				t.index = 2 * i
			}
		}
		m.globalDepth++
		m.globalShift--
		//m.directory = newDir
		m.dirPtr = unsafe.Pointer(&newDir[0])
		m.dirLen = len(newDir)
	}

	// N.B. left and right may still consume multiple indicies if the
	// directory has grown multiple times since old was last split.
	left.index = old.index
	m.replaceTable(left)

	entries := 1 << (m.globalDepth - left.localDepth)
	right.index = left.index + entries
	m.replaceTable(right)
}

源码解读：

首先判断预期的新容量（旧容量的两倍）是否达到了单表的最大group容量（maxTableCapacity=1024）
如果没超过，则进入扩容流程（grow）
如果超过了，则进入分裂流程（split）

扩容流程详解（grow）

首先创建新table（但还未替换）。
判断旧table是否为空，迁移所有元素（新建table时capacity=0，判断避免空表扫描）。
遍历所有group，遍历每个group内的8个slot，遇到delete或者empty的直接跳过，获取存活的key\elem，重新计算hash。
插入新table（unchecked）（新table是一定有足够空间的）。
完整性校验（checkInvariants）（used数量、ctrl状态、 growthLeft 、 map.used 一致性）。
替换旧table（replaceTable）。
标记旧table失效（t.index = -1）。

分裂流程详解（split）（ Extendible Hashing ）

当一个table已经到达最大容量（ maxTableCapacity ），但还是要继续插入时会进入split。
table的localdepth加1（localdepth表示当前table负责的hash范围，区别globaldepth表示当前directory使用的hash的范围）。
创建左右两个新table。
计算分流掩码mask（用于后续决定元素流向哪个table）。
扫描旧table，逐个slot进行分流。跳过delete\empty，取key\elem并重新hash。
根据mask，用新增的那一位hash来决定该slot流入哪个table（见上方源码）。
插入新table，不需要检查，保证成功。
将split的结果加入到directory中（如果directory的globalDepth位数不够用会扩容）。
废弃当前table。

分裂过程中第8步详解

判断当前directory的globalDepth位数是否够用。（split会使得localdepth+1，如果超过的globalDepth位数则需要扩容directory）

如果需要翻倍，创建新的dir数组，原来的2倍容量。

a. 复制并维护index。

b. 更新 globalDepth / globalShift 。

c. 原子替换 directory（将原来的dir指针指向新的数组）。

left覆盖原来old table负责的起始位置。

计算right table的index偏移量并设置right index。