浅谈Golang两种线程安全的map

package main

import "time"

func main() {
	testMapReadWriteDiffKey()
}

func testMapReadWriteDiffKey() {
	m := make(map[int]int)
	go func() {
		for {
			m[100] = 100
		}
	}()
	go func() {
		for {
			_ = m[12]
		}
	}()
	select {}
}

var counter = struct{
    sync.RWMutex
    m map[string]int
}{m: make(map[string]int)}

counter.RLock()
n := counter.m["some_key"]
counter.RUnlock()
fmt.Println("some_key:", n)

counter.Lock()
counter.m["some_key"]++
counter.Unlock()

type Map struct {
 // 互斥锁mu，操作dirty需先获取mu
 mu Mutex 

 // read是只读的数据结构，访问它无须加锁，sync.map的所有操作都优先读read
 // read中存储结构体readOnly，readOnly中存着真实数据---entry（详见1.3），read是dirty的子集
 // read中可能会存在脏数据：即entry被标记为已删除(详见1.3)
read atomic.Value // readOnly

 // dirty是可以同时读写的数据结构，访问它要加锁，新添加的key都会先放到dirty中
 // dirty == nil的情况：1.被初始化 2.提升为read后，但它不能一直为nil，否则read和dirty会数据不一致。
 // 当有新key来时，会用read中的数据 (不是read中的全部数据，而是未被标记为已删除的数据，详见3.2)填充dirty
 // dirty != nil时它存着sync.map的全部数据（包括read中未被标记为已删除的数据和新来的数据）
 dirty map[interface{}]*entry 

 // 统计访问read没有未命中然后穿透访问dirty的次数
 // 若miss等于dirty的长度，dirty会提升成read，提升后可以增加read的命中率，减少加锁访问dirty的次数   
 misses int
}

type readOnly struct {
	m       map[interface{}]*entry
	amended bool 
}

type entry struct {
       // p == nil：entry已从readOnly中删除但存在于dirty中
       // p == expunged：entry已从Map中删除且不在dirty中
       // p == 其他值：entry为正常值
       p unsafe.Pointer // *interface{}
}

func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
    // 从m.read中换出readOnly，然后从里面找key，这个过程不加锁
    read, _ := m.read.Load().(readOnly)
    e, ok := read.m[key]

    // readOnly中不存在此key但Map.dirty可能存在
    if !ok && read.amended {
        // 加锁访问Map.dirty
        m.mu.Lock()
        // 双重检测:若加锁前Map.dirty被替换为readonly，则前面m.read.Load().(readOnly)无效，需         
        // 要再次检查
        read, _ = m.read.Load().(readOnly)
        e, ok = read.m[key]
        // read.m没有此key && dirty里有可能有（dirty中有read.m没有的数据）
        if !ok && read.amended {
            // 从dirty中获取key对应的entry
            e, ok = m.dirty[key]
            // 无论Map.dirty中是否有这个key，miss都加一，若miss大小等于dirty的长度，dirty中的元素会被
            // 加到Map.read中
            m.missLocked()
        }
        m.mu.Unlock()
    }
    if !ok {
        return nil, false
    }
    // 若entry.p被删除（等于nil或expunged）返回nil和不存在（false），否则返回对应的值和存在（true）
    return e.load()
}

func (m *Map) missLocked() {
        // 访问一次Map.dirty，misses就要加一
	m.misses++
	if m.misses < len(m.dirty) {
		return
	}
        // 当misses等于dirty的长度，m.dirty提升为readOnly，amended被默认赋值成false
	m.read.Store(readOnly{m: m.dirty})
	m.dirty = nil
	m.misses = 0
}

func (m *Map) Store(key, value interface{}) {
    // 把m.read转成结构体readOnly
    read, _ := m.read.Load().(readOnly)
    // 若key在readOnly.m中且entry.p不为expunged（没有标记成已删除）即key同时存在于readOnly.m和dirty
    // ,用CAS技术更新value 【注】：e.tryStore在entry.p == expunged时会立刻返回false，否则用CAS
    // 尝试更新对应的value, 更新成功会返回true
    if e, ok := read.m[key]; ok && e.tryStore(&value) {
        return
    }
    // key不存在于readOnly.m或者entry.p==expunged(entry被标记为已删除)，加锁访问dirty
    m.mu.Lock()
    // 双重检测：若加锁前Map.dirty被提升为readOnly，则前面的read.m[key]可能无效，所以需要再次检测key是
    // 否存在于readOnly中
    read, _ = m.read.Load().(readOnly)
    // 若key在于readOnly.m中
    if e, ok := read.m[key]; ok {
        // entry.p之前的状态是expunged，把它置为nil
        if e.unexpungeLocked() {
            // 之前dirty中没有此key，所以往dirty中添加此key  
            m.dirty[key] = e
        }
        // 更新（把value的地址原子赋值给指针entry.p）
        e.storeLocked(&value)  
      // 若key在dirty中
    } else if e, ok := m.dirty[key]; ok { 
        // 更新（把value的地址原子赋值给指针entry.p）
        e.storeLocked(&value)
      // 来了个新key
    } else { 
        // dirty中没有新数据，往dirty中添加第一个新key
        if !read.amended {  
            // 把readOnly中未标记为删除的数据拷贝到dirty中
            m.dirtyLocked()  
            // amended:true，因为现在dirty有readOnly中没有的key
            m.read.Store(readOnly{m: read.m, amended: true})
        }
        // 把这个新的entry加到dirty中
        m.dirty[key] = newEntry(value)
    }
    m.mu.Unlock()
}

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
		}
	}
}

func (e *entry) unexpungeLocked() (wasExpunged bool) {
	return atomic.CompareAndSwapPointer(&e.p, expunged, nil)
}

func (m *Map) dirtyLocked() {
	if m.dirty != nil {  // 只要调用dirtyLocked，此时dirty肯定等于nil
		return
	}
        // dirty为nil时，把readOnly中没被标记成删除的entry添加到dirty
	read, _ := m.read.Load().(readOnly)
	m.dirty = make(map[interface{}]*entry, len(read.m))
	for k, e := range read.m {
                // tryExpungeLocked函数在entry未被删除时【e.p!=expunged&&e.p!=nil】返回false，在
                // e.p==nil时会将其置为expunged并返回true
		if !e.tryExpungeLocked() {  
			m.dirty[k] = e  // entry没被删除，把它添加到dirty中
		}
	}
}

func (m *Map) LoadAndDelete(key interface{}) (value interface{}, loaded bool) {
        // 从m.read中换出readOnly，然后从里面找key，此过程不加锁
	read, _ := m.read.Load().(readOnly)
	e, ok := read.m[key]

        // readOnly不存在此key，但dirty中可能存在
	if !ok && read.amended {
                // 加锁访问dirty
		m.mu.Lock()
                // 双重检测:若加锁前Map.dirty被替换为readonly，则前面m.read.Load().(readOnly)无 
                // 效，需要再次检查
		read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
                // readOnly不存在此key，但是dirty中可能存在
		if !ok && read.amended {
			e, ok = m.dirty[key]
			delete(m.dirty, key)
			m.missLocked()
		}
		m.mu.Unlock()
	}
	if ok {
                // 如果entry.p不为nil或者expunged，则把entry.p软删除（标记为nil）
		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
		}
                // e.p是真实值，把它置为nil
		if atomic.CompareAndSwapPointer(&e.p, p, nil) {
			return *(*interface{})(p), true
		}
	}
}

小结：

func (m *Map) Range(f func(key, value interface{}) bool) { 
     read, _ := m.read.Load().(readOnly) 
     if read.amended { // dirty存在readOnly中不存在的元素 
         // 加锁访问dirty
         m.mu.Lock() 
         // 再次检测read.amended，因为加锁前它可能已由true变成false
         read, _ = m.read.Load().(readOnly) 
         if read.amended { 
             // readOnly.amended被默认赋值成false 
             read = readOnly{m: m.dirty} 
             m.read.Store(read) 
             m.dirty = nil 
             m.misses = 0 
        } 
        m.mu.Unlock() 
    } 
    // 遍历readOnly.m
    for k, e := range read.m { 
         v, ok := e.load() 
         if !ok {
             continue 
         } 
         if !f(k, v) { 
             break 
         } 
    } 
}

// SHARD_COUNT 分片大小
var SHARD_COUNT = 32

type ConcurrentMap []*ConcurrentMapShared

// ConcurrentMapShared 分片的并发map
type ConcurrentMapShared struct {
	items        map[string]interface{}
	sync.RWMutex // 访问内部map都需要先获取读写锁
}

// New 创建一个concurrent map.
func New() ConcurrentMap {
	m := make(ConcurrentMap, SHARD_COUNT)
	for i := 0; i < SHARD_COUNT; i++ {
		m[i] = &ConcurrentMapShared{items: make(map[string]interface{})}
	}
	return m
}

// 先hash拿到key对应的分区号，然后加锁，读取值，最后释放锁和返回
func (m ConcurrentMap) Get(key string) (interface{}, bool) {
	// Get shard
	shard := m.GetShard(key)
	shard.RLock()
	// Get item from shard.
	val, ok := shard.items[key]
	shard.RUnlock()
	return val, ok
}

// 先hash拿到key对应的分区号，然后加锁，设置新值，最后释放锁
func (m ConcurrentMap) Set(key string, value interface{}) {
	// Get map shard.
	shard := m.GetShard(key)
	shard.Lock()
	shard.items[key] = value
	shard.Unlock()
}

// 先hash拿到key对应的分区号，然后加锁，删除key，最后释放锁
func (m ConcurrentMap) Remove(key string) {
	// Try to get shard.
	shard := m.GetShard(key)
	shard.Lock()
	delete(shard.items, key)
	shard.Unlock()
}

// 分别拿到每个分片map中的元素数量，然后汇总后返回
func (m ConcurrentMap) Count() int {
	count := 0
	for i := 0; i < SHARD_COUNT; i++ {
		shard := m[i]
		shard.RLock()
		count += len(shard.items)
		shard.RUnlock()
	}
	return count
}

// 先hash拿到key对应的分区号，然后加锁，如果key存在就更新其value，否则插入新的k-v，释放锁并返回
func (m ConcurrentMap) Upsert(key string, value interface{}, cb UpsertCb) (res interface{}) {
	shard := m.GetShard(key)
	shard.Lock()
	v, ok := shard.items[key]
	res = cb(ok, v, value)
	shard.items[key] = res
	shard.Unlock()
	return res
}