深入理解Go语音的sync.Map

fatal error: concurrent map read and map write

    var m1 sync.Map   // 也可以使用 m := new(sync.Map) 来创建一个sync.Map实例

value, ok := m.Load(key)
if ok {
    // key存在，value是对应的值
} else {
    // key不存在
}

m.Store(key, value)

actual, loaded := m.LoadOrStore(key, value)
if loaded {
    // 键已经存在，actual是已存在的值
} else {
    // 键不存在，已存储提供的值，actual是提供的值
}

m.Delete(key)

m.Range(func(key, value interface{}) bool {
    // 使用key和value
    // 如果要停止迭代，返回false
    return true
})

// sync/map.go
type Map struct {
    mu Mutex          // 互斥锁，用于保护dirty字段和misses字段。
    read atomic.Value // readOnly, 一个atomic.Value类型的字段，存储了一个readOnly结构体，用于存储只读数据。
    dirty map[interface{}]*entry // 一个map，存储了所有可写的键值对。
    misses int    // 一个计数器，记录了从read读取失败的次数，用于触发将数据从dirty迁移到read的决策。
}

type readOnly struct {
    m       map[interface{}]*entry    // 实际存储键值对的map。
    amended bool // 标记位，如果dirty中有read中没有的键，那么为true
}

type entry struct {
    p unsafe.Pointer // *interface{} 一个指向实际数据的指针。
}

var expunged = unsafe.Pointer(new(any))

// Load returns the value stored in the map for a key, or nil if no
// value is present.
// The ok result indicates whether value was found in the map.
func (m *Map) Load(key any) (value any, ok bool) {
	// 尝试从 Map.read 中查找数据
	read, _ := m.read.Load().(readOnly)
	e, ok := read.m[key]

	// Map.read 中找不到 Key， 且dirty中有新增的key， 则尝试从 Map.dirty 中查找
	if !ok && read.amended {
		m.mu.Lock()
	  // 避免并发时，在上一次读 m.read.Load() 和 m.mu.Lock() 的时间区间内发生 dirty=>read 的升级
		read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
		if !ok && read.amended {
			e, ok = m.dirty[key]
			// 计数器加1， 达到阈值后，触发dirty=>read的升级
			m.missLocked()
		}
		m.mu.Unlock()
	}
	if !ok {
		return nil, false
	}
	return e.load()
}

func (e *entry) load() (value any, ok bool) {
	p := atomic.LoadPointer(&e.p)
	// p == nil || p == expunged 都可以确定entry已经被删除，
	if p == nil || p == expunged {
		return nil, false
	}
	return *(*any)(p), true
}

// Store sets the value for a key.
func (m *Map) Store(key, value any) {
	read, _ := m.read.Load().(readOnly)
	// 如果 key 在 m.read.m 中存在，可以直接先尝试通过m.read.m的entry来修改value。
	// 这里有一个例外是 p == expunged，因为此时这个key在dirty中已经不存在了。 而我们需要保证dirty中持有全量的key。 详见tryStore函数
  // 因而在这种 p == expunged 需要进一步操作m.dirty
	if e, ok := read.m[key]; ok && e.tryStore(&value) {
		return
	}

	m.mu.Lock() // 加锁
	read, _ = m.read.Load().(readOnly)
	if e, ok := read.m[key]; ok {
	  // 属于前面说的p == expunged的例外情况，需要先将p设置为nil，然后在m.dirty中补齐当前key
		if e.unexpungeLocked() {
			m.dirty[key] = e
		}
		e.storeLocked(&value) // 此时，m.read.m中的p为nil了，重新进行store。
	} else if e, ok := m.dirty[key]; ok {
		e.storeLocked(&value)
	} else {
		if !read.amended {
			// We're adding the first new key to the dirty map.
			// Make sure it is allocated and mark the read-only map as incomplete.
			m.dirtyLocked()
			m.read.Store(readOnly{m: read.m, amended: true})
		}
		m.dirty[key] = newEntry(value)
	}
	m.mu.Unlock()
}

// tryStore stores a value if the entry has not been expunged.
//
// If the entry is expunged, tryStore returns false and leaves the entry
// unchanged.
func (e *entry) tryStore(i *any) bool {
	for {
		p := atomic.LoadPointer(&e.p)
		if p == expunged {
			return false
		}
		if atomic.CompareAndSwapPointer(&e.p, p, unsafe.Pointer(i)) {
			return true
		}
	}
}

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

	read, _ := m.read.Load().(readOnly)
	m.dirty = make(map[any]*entry, len(read.m))
	// 进行一次 m.read.m 到 m.dirty 同步
	for k, e := range read.m {  
		if !e.tryExpungeLocked() {
			m.dirty[k] = e
		}
	}
}

// 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 any) (value any, loaded bool) {
	read, _ := m.read.Load().(readOnly)
	e, ok := read.m[key]

  // m.read 中找不到key，且 m.dirty中存在新key
	if !ok && read.amended {
		m.mu.Lock()

		// 排除并发干扰，上一次读之后发生了 dirty=>read 的数据同步，导致 m.read变化，因而需要重新判断
		read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
		if !ok && read.amended {
			e, ok = m.dirty[key]
			delete(m.dirty, key)
			// 计数器 加1 
			m.missLocked()
		}
		m.mu.Unlock()
	}
	if ok {
		return e.delete()
	}
	return nil, false
}

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

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