并发编程之Map

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

type Map struct {
	// 当涉及到脏数据（dirty）操作时，需要用锁
	mu Mutex

	read atomic.Value // readOnly

	// dirty 包含部分map键值对，如果需要操作需要mutex获取锁
	// 最后 dirty 中的元素会被全部提升到read里的map去
	dirty map[interface{}]*entry

	// 计数器，用于记录read中没有的而在dirty中有的数据的数量。
	// 也就是说如果read不包含这个数据，会从dirty中读取，misses+1
	// 当misses的数量等于dirty的长度，就会将dirty中的数据迁移到read中
	misses int
}

1
2
3

type entry struct {
    p unsafe.Pointer // *interface{}
}

1
2
3
4

type readOnly struct {
	m map[interface{}]*entry
	amended bool // true if the dirty map contains some key not in m.
}

 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

func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
	// 从read的map中查找
	read, _ := m.read.Load().(readOnly)
	e, ok := read.m[key]
	// 如果没有找到，并且read.amended=true，说明dirty中有新数据，从dirty中查找，需要加锁
	if !ok && read.amended {
		m.mu.Lock()
		// 在read中再检查一遍，防止加锁的时候dirty已经迁移到read中（二次检查）
		read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
		if !ok && read.amended {
			e, ok = m.dirty[key]
			// 不管是否存在，记录misses+1，满足条件后将dirty中数据迁移到read中
			m.missLocked()
		}
		m.mu.Unlock()
	}
	if !ok {
		return nil, false
	}
	return e.load()
}

func (e *entry) load() (value interface{}, ok bool) {
	p := atomic.LoadPointer(&e.p)
	if p == nil || p == expunged {
		return nil, false
	}
	return *(*interface{})(p), true
}

func (m *Map) missLocked() {
	m.misses++
	if m.misses < len(m.dirty) { // misses小于dirty的长度，就不迁移数据
		return
	}
	m.read.Store(readOnly{m: m.dirty})
	m.dirty = nil
	m.misses = 0
}

 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

func (m *Map) Store(key, value interface{}) {
	// 从read map中如果能找到key，update
	read, _ := m.read.Load().(readOnly)
	if e, ok := read.m[key]; ok && e.tryStore(&value) {
		return
	}

	// 上锁之后重新check一下read map的内容(二次读写)
	m.mu.Lock()
	read, _ = m.read.Load().(readOnly)
	// 场景一：如果在read中，而key对应的value标识是已删除（需要先在dirty中添加），更新这个值
	if e, ok := read.m[key]; ok {
		if e.unexpungeLocked() {
			// The entry was previously expunged, which implies that there is a
			// non-nil dirty map and this entry is not in it.
			m.dirty[key] = e
		}
		e.storeLocked(&value)
		// 场景二：如果不在read中，在dirty中，直接更新值
	} else if e, ok := m.dirty[key]; ok {
		e.storeLocked(&value)
		// 场景三：都不存在，那么把数据存在dirty中，并且修改read中的标识
	} else {
		if !read.amended { // dirty没有新数据
			// We're adding the first new key to the dirty map.
			// Make sure it is allocated and mark the read-only map as incomplete.
			// 从read中复制未删除的数据
			m.dirtyLocked()
			// 更改状态
			m.read.Store(readOnly{m: read.m, amended: true})
		}
		m.dirty[key] = newEntry(value) // 加入dirty中
	}
	m.mu.Unlock()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

func (e *entry) tryStore(i *interface{}) bool {
	for {
		p := atomic.LoadPointer(&e.p)
		if p == expunged {
			return false
		}
		if atomic.CompareAndSwapPointer(&e.p, p, unsafe.Pointer(i)) {
			return true
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

func (m *Map) Delete(key interface{}) {
	read, _ := m.read.Load().(readOnly)
	e, ok := read.m[key]
	if !ok && read.amended { // 不存在read中，并且dirty有新数据
		m.mu.Lock()
		read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
		if !ok && read.amended {
			delete(m.dirty, key) // 加锁删除dirty的数据
		}
		m.mu.Unlock()
	}
	if ok {
		// 在read中存在key
		e.delete()
	}
}