golang并发安全-sync.map

sync.map解决的问题

golang 原生map是存在并发读写的问题，在并发读写时候会抛出异常

func main() {
	mT := make(map[int]int)
	g1 := []int{1, 2, 3, 4, 5, 6}
	g2 := []int{4, 5, 6, 7, 8, 9}
	go func() {
		for i := range g1 {
			mT[i] = i
		}

	}()
	go func() {
		for i := range g2 {
			mT[i] = i
		}
	}()
	time.Sleep(3 * time.Second)
}

抛出异常 fatal error: concurrent map writes

如果将map换成sync.map 那么就不会出现这个问题，下面就简单说说syn.map怎么实现的

基本结构

Map结构体

// Map类型针对两种常见的用例进行了优化：1-当给定键的条目只写一次但读多次时，如在只增长的缓存中，2-当多个goroutine读取、写入和覆盖不相交的键集的条目时。在这两种情况下，与单独的Mutex或RWMutex配对的Go映射相比，使用Map可以显著减少锁争用。
type Map struct { 
	// 互斥锁mu，操作dirty需先获取mu 
    mu Mutex 
	// read是只读的数据结构，可安全并发访问部分，访问它无须加锁，sync.map的所有操作都优先读read 
    // read中存储结构体readOnly，readOnly中存着真实数据，储存数据时候需要加锁
    // read中可能会存在脏数据：即entry被标记为已删除
    read atomic.Value // readOnly
 	// dirty是可以同时读写的数据结构，访问它要加锁，新添加的key都会先放到dirty中 
    // dirty == nil的情况：
    // 1.被初始化 
    // 2.提升为read后，但它不能一直为nil，否则read和dirty会数据不一致。 
    // 当有新key来时，会用read中的数据(不是read中的全部数据，而是未被标记为已删除的数据，)填充dirty 
    // dirty != nil时它存着sync.map的全部数据（包括read中未被标记为已删除的数据和新来的数据）
    dirty map[interface{}]*entry 
 	// 统计访问read没有未命中然后穿透访问dirty的次数 
    // 若miss等于dirty的长度，dirty会提升成read，提升后可以增加read的命中率，减少加锁访问dirty的次数    
    misses int
}

?readOnly结构体

//第一点的结构read存的就是readOnly
type readOnly struct {
	m       map[any]*entry //m是一个map，key是interface，value是指针entry，其指向真实数据的地址，
	amended bool  // amended等于true代表dirty中有readOnly.m中不存在的entry。
}

entry结构体

type entry struct { 
    // p:
    //     expunged: 删除； nil: 逻辑删除但存在dirty; 数据  
    p unsafe.Pointer // *interface{}
}

Load方法

代码解说

Load：读取数据

// Load 返回 map 中key 对应的值，如果没有值，则返回 nil。
// ok 结果表示是否在 map 中找到了 value。
func (m *Map) Load(key any) (value any, ok bool) {
	read, _ := m.read.Load().(readOnly) // 从read 读取数据，并转换readonly
	e, ok := read.m[key]
	if !ok && read.amended { // readonly没有找到对应数据
		m.mu.Lock()
        // 双重检测:
        // 再检查一次readonly，以防中间有Map.dirty被替换为readonly
		read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
		if !ok && read.amended { // 去 dirty查找对应数据
			e, ok = m.dirty[key]
			// 无论Map.dirty中是否有这个key，miss都加一，
            // 若miss大小等于dirty的长度，dirty中的元素会被加到Map.read中 
			m.missLocked()
		}
		m.mu.Unlock()
	}
	if !ok {
		return nil, false
	}
	return e.load()// 若entry.p被删除（等于nil或expunged）返回nil和不存在（false），否则返回对应的值和存在（true）    
}

missLocked：dirty是如何提升为read

func (m *Map) missLocked() {
	m.misses++ // 每次misses+1
	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
}

流程图

?load： 会先从readOnly查找数据, 如果没有开启加锁，再次访问readOnly, 再次没有再去dirty去查。

Store方法

代码解说

store： 赋值

// Store 设置key value
func (m *Map) Store(key, value any) {
	read, _ := m.read.Load().(readOnly) // 转换readOnly
    // 若key在readOnly.m中且 e.tryStore 不为 false（没有逻辑删除）
	if e, ok := read.m[key]; ok && e.tryStore(&value) {
		return
	}

	m.mu.Lock()
    // 双重检测:
    // 再检查一次readonly，以防中间有Map.dirty被替换为readonly
	read, _ = m.read.Load().(readOnly)
	if e, ok := read.m[key]; ok {
        // entry.p状态是expunged置为nil
        // 如果是逻辑删除就需要清除标记了
		if e.unexpungeLocked() {
			// 之前dirty中没有此key，所以往dirty中添加此key              
			m.dirty[key] = e
		}
        // cas: 赋值
		e.storeLocked(&value)
	} else if e, ok := m.dirty[key]; ok {
		e.storeLocked(&value)
	} 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})
		}
		m.dirty[key] = newEntry(value)
	}
	m.mu.Unlock()
}

tryStore：尝试写入数据

func (e *entry) tryStore(i *any) bool {
    for {   
        p := atomic.LoadPointer(&e.p)    
        if p == expunged {   // 如果逻辑删除就返回false    
            return false   
        }    

        // 不是就将value写入
        if atomic.CompareAndSwapPointer(&e.p, p, unsafe.Pointer(i)) {      
            return true    
        }  
    }
}

dirtyLocked： 将readOnly 未删除的放到dirty

func (m *Map) dirtyLocked() { 
    if m.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未被删除时返回false，反之返回true    
        if !e.tryExpungeLocked() {    // entry没被删除    
            m.dirty[k] = e 
            
        }  
    }
}

流程图

sync.map不适合用于频繁插入新key-value的场景，因为此操作会频繁加锁访问dirty会导致性能下降。更新操作在key存在于readOnly中且值没有被标记为删除(expunged)的场景下会用无锁操作CAS进行性能优化，否则也会加锁访问dirty。

Delete方法

代码解说

LoadAndDelete：查找删除

func (m *Map) LoadAndDelete(key any) (value any, loaded bool) {
    read, _ := m.read.Load().(readOnly) 
    e, ok := read.m[key]
    if !ok && read.amended { // readOnly不存在此key，但dirty中可能存在               
        // 加锁访问dirty   
        m.mu.Lock()                
        // 双重检测 
        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()   // 判断dirty是否可以转换readOnly，可以就转换
        }   
        m.mu.Unlock()  
    }  
    if ok {                
        // 如果entry.p不为nil或者expunged，则把逻辑删除（标记为nil）    
        return e.delete()  
    } 
    return nil, false
}

delete：逻辑删除

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

流程图

Range方法

代码解说

Range：轮训元素

func (m *Map) Range(f func(key, value any) bool) {
    read, _ := m.read.Load().(readOnly)     
    if read.amended { // 如果dirty存在数据
        m.mu.Lock()         
        // 双重检测      
        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         
        }     
    }
}

流程图?

Range：全部key都存在于readOnly中时，是无锁遍历的，性能最优。如果readOnly只存在Map中的部分key时，会一次性加锁拷贝dirty的元素到readOnly，减少多次加锁访问dirty中的数据。

总结

1- sync.map 结构体加了readOnly 和 dirty 来实现读写分离，load，store, delete，range 每次都会优先访问read，后面访问dirty都会双重检测以防加锁前Map.dirty可能已被提升为read

2- sync.map不适合写多读少，从store 代码中可以看出会频繁加锁访问dirty，双重检测等等，这些都会导致性能下降

3- sync.map 没有提供对read， dirty 的长度方法，这个对象使用在于并发场景下，会额外带来锁竞争的问题

4- misses 是 统计访问read没有未命中然后穿透访问dirty的次数 ，如果等于dirty会转换readOnly

5- entry 有三种类型 expunged: 删除； nil: 逻辑删除但存在dirty; 数据 。其中expunged 会在 unexpungeLocked 方法中进行赋值（在store时候会加锁访问dirty，把readOnly中的未被标记为删除的所有entry指针放到dirty，之前被delete方法标记为删除状态的entry=nil都变为expunged，那这些被标记为expunged的entry将不会出现在dirty中。）