Lock

1.条件等待

1.1.简介

• Go中由于channel的存在，sync.cond并发原语并不常用，但在一些开源组件还是可以看到，诸如kubernetes中的并发等待队列。

• --- Cond实现和定义的方法
      func NewCond(l Locker) *Cond
      func (c *Cond) Broadcast()
      func (c *Cond) Signal()
      func (c *Cond) Wait()
  

• sync.Cond是一个结构体，主要围绕条件并发设计，NewCond()函数接收一个互斥锁对象sync.Locker构造结构体指针。互斥锁对象sync.Locker是一个接口，sync.Mutex和sync.RWMutex都实现了此接口，这把互斥锁也是sync.Cond实现的关键。具体使用时，wait()可阻塞当前协程，直至其他协程调用Broadcast或Signal方法唤醒该协程。

1.2.初始化

• NewCond(locker)用于初始化sync.Cond，locker用于检查或修改条件时准入判断，持有锁的协程才能操作。

• type Cond struct {
  	noCopy noCopy  // 防止复制
  
  	L Locker
  
  	notify  notifyList
  	checker copyChecker
  }
  
  func NewCond(l Locker) *Cond {
  	return &Cond{L: l}
  }
  

• notify属性用于记录被阻塞等待的队列，主要维护通知列表，用于wait与Broadcast或Signal协作时高效协调协程的阻塞和唤醒。

• type notifyList struct {
  	wait   uint32           // 当前进入等待状态的协程数量
  	notify uint32			// 当前被通知可以继续的协程数量
  	lock   uintptr 			// 用于保护wait/notify的锁
  	head   unsafe.Pointer	// 等待队列的头指针
  	tail   unsafe.Pointer	// 等待队列的尾指针
  }
  

• checker用于防止sync.Cond结构体被复制，可以在运行时进行动态类型检查，nocopy防止复制主要作编译时的静态类型检查。

• type copyChecker uintptr
  
  // 检查是否被复制,出现复制时抛出panic
  func (c *copyChecker) check() {
  	if uintptr(*c) != uintptr(unsafe.Pointer(c)) &&
  		!atomic.CompareAndSwapUintptr((*uintptr)(c), 0, uintptr(unsafe.Pointer(c))) &&
  		uintptr(*c) != uintptr(unsafe.Pointer(c)) {
  		panic("sync.Cond is copied")
  	}
  }
  

• sync.Cond的实现很简略，主要是因为复杂的逻辑都放在了更底层的runtime实现，每个方法实现都会调用c.checker.check()检查对象cond是否被复制，然后才是主逻辑。

• func (c *Cond) Wait() {
  	c.checker.check()
  	t := runtime_notifyListAdd(&c.notify)
  	c.L.Unlock()
  	runtime_notifyListWait(&c.notify, t)
  	c.L.Lock()
  }
  
  func (c *Cond) Signal() {
  	c.checker.check()
  	runtime_notifyListNotifyOne(&c.notify)
  }
  
  func (c *Cond) Broadcast() {
  	c.checker.check()
  	runtime_notifyListNotifyAll(&c.notify)
  }
  

• wait方法会先调用runtime_notifyListAdd函数将调用者加入通知列表，再调用runtime_notifyListWait函数阻塞并等待通知，收到通知被唤醒后解锁。Signal方法内部调用runtime_notifyListNotifyOne通知唤醒一个通知列表中的调用者，然后将其从从通知列表删除。Broadcast则调用runtime_notifyListNotifyAll通知唤醒整个通知列表调用者，然后清空通知列表。

1.3.锁的定位

• wait的使用要求互斥锁和条件变量结合，锁的使用确保并发安全情况下对条件的检查，判断是否继续等待及唤醒时对条件的安全修改。底层的锁由调用方传入，调用方可以借助锁并发安全修改条件变量，wait方法内部则借助同一把锁加入等待队列及挂起当前协程。

• --- 使用
  	//	c.L.Lock()
      //	for !condition() {
      //	    c.Wait()
      //	}
      //	... make use of condition ...
      //	c.L.Unlock()
  	1.调用`wait`前加锁,调用后释放锁,基于`for`循环检测条件
  	2.基于锁并发安全修改条件变量
  	3.每次修改后调用`Signal/Broadcast`唤醒`wait`
  

1.4.示例

• // 队列接口
  type Interface interface {
  	Push(item any)    // 元素入队
  	Pop() (any, bool) // 元素出队
  	Len() int         // 队列长度
  	Close()           // 关闭队列
  	Closed() bool     // 队列是否关闭
  }
  
  // 并发等待队列(kubernetes中workqueue的精简实现)
  type Queue struct {
  	// 条件变量
  	cond *sync.Cond
  	// 队列
  	data []any
  	// 队列开启状态
  	shutdown bool
  }
  
  func NewQueue() *Queue {
  	return &Queue{
  		cond: sync.NewCond(&sync.Mutex{}),
  	}
  }
  
  // Len 获取队列长度
  func (q *Queue) Len() int {
  	q.cond.L.Lock()
  	defer q.cond.L.Unlock()
  
  	return len(q.data) // 返回队列当前长度
  }
  
  func (q *Queue) Push(item any) {
  	q.cond.L.Lock()
  	defer q.cond.L.Unlock()
  	// 检查队列开启状态
  	if q.shutdown {
  		return
  	}
  	// 追加及唤醒
  	q.data = append(q.data, item)
  	q.cond.Signal()
  }
  
  func (q *Queue) Pop() (any, bool) {
  	q.cond.L.Lock()
  	defer q.cond.L.Unlock()
  	// 队列为空且开启则阻塞当前协程,等待填充数据
  	if len(q.data) == 0 && !q.shutdown {
  		q.cond.Wait() // 如果队列为空且未关闭，阻塞等待队列中有数据时被唤醒
  	}
  	// 唤醒后队列依然为空结束(队列已经关闭了)
  	if len(q.data) == 0 {
  		return nil, true
  	}
  	// 出队
  	item := q.data[0]
  	// 清除引用帮助GC回收
  	q.data[0] = nil
  	// 出队
  	q.data = q.data[1:]
  	return item, false
  }
  
  func (q *Queue) Close() {
  	q.cond.L.Lock()
  	defer q.cond.L.Unlock()
  
  	q.shutdown = true
  
  	q.cond.Broadcast()
  }
  
  func (q *Queue) Closed() bool {
  	q.cond.L.Lock()
  	defer q.cond.L.Unlock()
  	return q.shutdown
  }
  

• 此处方法必须使用指针接收，否则由于值拷贝无法保证锁唯一

2.信号量

2.1.原理

• semaphore是Go中的另一个并发原语，用于通过信号控制协程之间协调，其方法定义如下。

•     func NewWeighted(n int64) *Weighted
      func (s *Weighted) Acquire(ctx context.Context, n int64) error
      func (s *Weighted) Release(n int64)
      func (s *Weighted) TryAcquire(n int64) bool
  

• semaphore.Weighted是一个表示信号量的结构体，NewWeighted是其构造函数，Acquire用于阻塞请求信号量，Release用于释放信号量，TryAcquire用于非阻塞请求信号量。

• type Weighted struct {
  	size    int64       // 资源总数量
  	cur     int64  		// 当前已经使用的资源数
  	mu      sync.Mutex	// 互斥锁
  	waiters list.List	// 等待者队列,用于列表实现
  }
  

• Weighted结构体包含4个字段，用于记录资源可用数量及等待者。

• func NewWeighted(n int64) *Weighted {
  	w := &Weighted{size: n}
  	return w
  }
  

• semaphore还定义了等待者结构体waiter，用于记录当前等待者请求资源数及准备好的资源管道。

• type waiter struct {
  	n     int64
  	ready chan<- struct{} // Closed when semaphore acquired.
  }
  

• Weighted最复杂的方法就是Acquire，它主要作请求检查、资源判断、等待者唤醒等主要工作。

• // 用于请求n个资源
  // 资源不足阻塞,直至资源就绪或ctx取消
  func (s *Weighted) Acquire(ctx context.Context, n int64) error {
  	s.mu.Lock()
      // 没有其他等待且资源充足,直接返回
  	if s.size-s.cur >= n && s.waiters.Len() == 0 {
  		s.cur += n
  		s.mu.Unlock()
  		return nil
  	}
  
      // 资源超出,结束ctx并返回error
  	if n > s.size {
  		// Don't make other Acquire calls block on one that's doomed to fail.
  		s.mu.Unlock()
  		<-ctx.Done()
  		return ctx.Err()
  	}
  
      // 加入等待队列
      // 这里创建channel作为等待者属性,用于后续通知其唤醒
  	ready := make(chan struct{})
  	w := waiter{n: n, ready: ready}
  	elem := s.waiters.PushBack(w)
  	s.mu.Unlock()
  
      // 使用select实现阻塞等待
  	select {
      // ctx被取消
  	case <-ctx.Done():
  		err := ctx.Err()
  		s.mu.Lock()
  		select {
          // 检查waiter是否被唤醒
  		case <-ready:
              // 清空err,被唤醒的协程继续做事(唤醒并分配了资源)
  			err = nil
  		default:
              // 当前waiter是否为第一个等待者
  			isFront := s.waiters.Front() == elem
              // 将当前waiter从等待者队列移除
  			s.waiters.Remove(elem)
  			// 如果当前waiter是队列的第一个等待者且存在剩余资源
  			if isFront && s.size > s.cur {
                  // 通知等待队列检查下一个waiter资源数是否充足
  				s.notifyWaiters()
  			}
  		}
  		s.mu.Unlock()
  		return err
  
  	case <-ready:
  		return nil
  	}
  }
  

• Release逻辑会归还资源数，归还时会检查是否超出，保证申请多少资源就释放多少资源，然后调用s.notifyWaiters()通知等待队列检查下一个waiter资源数是否满足。

• func (s *Weighted) Release(n int64) {
  	s.mu.Lock()
      // 释放当前资源
  	s.cur -= n
      // 资源超出panic
  	if s.cur < 0 {
  		s.mu.Unlock()
  		panic("semaphore: released more than held")
  	}
      // 通知唤醒其他等待者并申请资源
  	s.notifyWaiters()
  	s.mu.Unlock()
  }
  

• notifyWaiters内部会循环检查下一个waiter请求的资源是否满足，满足则出队，不满足则退出，notifyWaiters满足先入先出原则。唤醒检查waiter资源是否满足时，会按照队中顺序尝试申请，碰到第一个不满足的waiter就会退出，避免请求资源小的waiter总是先执行，造成某些waiter长时间饥饿。

• func (s *Weighted) notifyWaiters() {
  	for {
          // 获取下一个waiter
  		next := s.waiters.Front()
          // 下一个waiter不存在,说明不存在等待者,退出
  		if next == nil {
  			break
  		}
  		// 判断waiter申请资源是否充足
  		w := next.Value.(waiter)
  		if s.size-s.cur < w.n {
  			// Not enough tokens for the next waiter.  We could keep going (to try to
  			// find a waiter with a smaller request), but under load that could cause
  			// starvation for large requests; instead, we leave all remaining waiters
  			// blocked.
  			//
  			// Consider a semaphore used as a read-write lock, with N tokens, N
  			// readers, and one writer.  Each reader can Acquire(1) to obtain a read
  			// lock.  The writer can Acquire(N) to obtain a write lock, excluding all
  			// of the readers.  If we allow the readers to jump ahead in the queue,
  			// the writer will starve — there is always one token available for every
  			// reader.
  			break
  		}
  		// 占用资源
  		s.cur += w.n
          // 从等待队列移除waiter
  		s.waiters.Remove(next)
          // 关闭管道,唤醒协程
  		close(w.ready)
  	}
  }
  

• Acquire方法用于阻塞申请，TryAcquire则常用于非阻塞申请，判断资源充足且不存在其他等待者时就会占用资源，否则退出。

• func (s *Weighted) TryAcquire(n int64) bool {
  	s.mu.Lock()
      // 判断申请是否满足
  	success := s.size-s.cur >= n && s.waiters.Len() == 0
  	if success {
          // 满足则占用资源
  		s.cur += n
  	}
  	s.mu.Unlock()
  	return success
  }
  

2.2.示例

• func collatzSteps(n int) (steps int) {
  	if n <= 0 {
  		panic("none positive input")
  	}
  
  	for ; n > 1; steps++ {
  		if steps < 0 {
  			panic("too many steps")
  		}
  
  		if n%2 == 0 {
  			n /= 2
  			continue
  		}
  
  		const maxInt = int(^uint(0) >> 1)
  		if n > (maxInt-1)/3 {
  			panic("overflow")
  		}
  		n = 3*n + 1
  	}
  
  	return steps
  }
  
  func main() {
  	ctx := context.TODO()
  	// 定义协程池参数
  	var (
  		// 最大协程数
  		maxWorkers = runtime.GOMAXPROCS(0)
  		// 协程池
  		sem = semaphore.NewWeighted(int64(maxWorkers))
  		// 任务数
  		out = make([]int, 32)
  	)
  	// 一次最大启动maxWorkers个协程计算输出
  	for i := range out {
  		// 当最大协程耗尽Acquire会阻塞,直至有协程释放
  		if err := sem.Acquire(ctx, 1); err != nil {
  			log.Printf("Failed to acquire semaphore: %v", err)
  			break
  		}
  		// 开始协程执行计算任务
  		go func(i int) {
  			// 释放协程
  			defer sem.Release(1)
  			// 计算任务
  			out[i] = collatzSteps(i + 1)
  		}(i)
  	}
  	defer sem.Release(int64(maxWorkers))
  	// 回收所有协程
  	if err := sem.Acquire(ctx, int64(maxWorkers)); err != nil {
  		log.Printf("Failed to acquire semaphore: %v", err)
  
  	}
  	fmt.Println(out)
  }
  

3.syncMap

3.1.简介

• Go中的map类型不是并发安全的，进一步出现了sync.Map提供并发原语操作Map。sync.Map是Map的进一步包装，零值可用且支持并发安全，通过内部锁安全执行Store、Load和Delete操作。

• func currentHashMap() {
      var s sync.Map
  
  	// 存储键值对
  	s.Store("name", "bird")
  	s.Store("age", 1)
  	s.Store("location", "Beijing")
  
  	// 读取值
  	if value, ok := s.Load("name"); ok {
  		fmt.Println("name:", value)
  	}
  
  	// 删除一个键
  	s.Delete("age")
  
  	// 遍历 sync.Map
  	s.Range(func(key, value interface{}) bool {
  		fmt.Printf("%s: %s\n", key, value)
  		return true // 继续遍历
  	})
  }
  
  

3.2.Store

• sync.Map结构体扩展了map的基本功能，提供额外的LoadAndDelete、LoadOrStore等复合操作。

• 	Load(interface{}) (interface{}, bool)
  	Store(key, value interface{})
  	LoadOrStore(key, value interface{}) (actual interface{}, loaded bool)
  	Delete(interface{})
  	Range(func(key, value interface{}) (shouldContinue bool))
  	LoadAndDelete(key any) (value any, loaded bool)
  

• sync.Map的设计类似缓存，其内部基于锁保护对dirty区操作，dirty区域是全量数据。read字段是一个原子类型的指针，指向readOnly结构体，readOnly内部也有一个map，用于只读操作。misses字段用于计数，记录read区读取失败次数，达到一定条件会将dirty提升为read。entry结构体是具体存储的value，用于记录值及键值对状态，内部会通过指针记录键值对是否在sync.Map中。expunged作为全局变量，值为任意指针，记录键值对是否从dirty区删除。

• // 指针,标记从dirty删除元素
  var expunged = unsafe.Pointer(new(any))
  
  type Map struct {
  	mu Mutex             	// 互斥锁,用于并发安全操作map
  	read atomic.Value 		// 原子类型指针,指向readOnly
  	dirty map[any]*entry	// 写map
  	misses int  			// read区数据未命中次数
  }
  
  type readOnly struct {
      // 存储数据的只读map
  	m       map[any]*entry
      // 记录是否存在新的key只存在于dirty
  	amended bool
  }
  
  type entry struct {
  	// 存储值的指针,支持原子操作
  	p unsafe.Pointer // *interface{}
  }
  

• Store方法用于存储键值对，可以交换指定键对应的值，也可以设置新键值对，存储时会判断key对应entry是否被标记为删除，通过标识状态更新或设置目标键值，减少entry被彻底删除的可能性。此外，由于某个key被彻底删除后有概率涉及到dirtyLocked操作，会遍历全量的read.m重新初始化dirty，很可能导致sync.Map的性能退化。

• func (e *entry) tryStore(i *any) bool {
  	for {
          // 获取key值(指针)
  		p := atomic.LoadPointer(&e.p)
          // key标记删除
  		if p == expunged {
  			return false
  		}
          // CAS更新key值
  		if atomic.CompareAndSwapPointer(&e.p, p, unsafe.Pointer(i)) {
  			return true
  		}
  	}
  }
  
  func (m *Map) dirtyLocked() {
      // dirty不为空
  	if m.dirty != nil {
  		return
  	}
  
      // dirty为空,基于read区未删除数据初始化dirty
  	read, _ := m.read.Load().(readOnly)
  	m.dirty = make(map[any]*entry, len(read.m))
  	for k, e := range read.m {
          // 将key值为空元素标记为删除,跳过标记为删除数据
  		if !e.tryExpungeLocked() {
  			m.dirty[k] = e
  		}
  	}
  }
  
  func (m *Map) Store(key, value any) {
      // 获取readMap
  	read, _ := m.read.Load().(readOnly)
      // key存在于read区尝试更新(未标记删除才更新,标记删除时dirty不存在,ready存在该key,更新可能造成数据不一致)
  	if e, ok := read.m[key]; ok && e.tryStore(&value) {
  		return
  	}
  
  	m.mu.Lock()
      // 重新获取(避免旧的被破坏)
  	read, _ = m.read.Load().(readOnly)
      // ready区尝试读取key
  	if e, ok := read.m[key]; ok {
          // 将标记删除的扭转为nil且存到dirty
  		if e.unexpungeLocked() {
  			m.dirty[key] = e
  		}
          // 存储值(dirty和read)中数据都会被更新
  		e.storeLocked(&value)
      // dirty区尝试读取key
  	} else if e, ok := m.dirty[key]; ok {
          // dirty区存在更新值
  		e.storeLocked(&value)
      // read区和dirty区都不存在
  	} else {
          // dirty区存在独占key,read区未标记
  		if !read.amended {
              // dirty为空时,将read区未标记删除数据放入dirty
  			m.dirtyLocked()
              // 标记read区缺数据,同步完整性
  			m.read.Store(readOnly{m: read.m, amended: true})
  		}
          // dirty区新增键值对
  		m.dirty[key] = newEntry(value)
  	}
  	m.mu.Unlock()
  }
  

• 

3.3.Load

• Load方法用于读取键值，会优先从readMap读取，此时不需要加锁，如果readMap没找到且read.amended为true，说明dirty包含新的键值，需要加锁从慢路径查找。慢路径查找时，会在加锁后双步检查，再次尝试从readMap读取，以应对dirty升级为read情况， 仍未找到时会将read.misses+1，同时继续从dirty查找。当read查找未命中次数大量dirty全量数据长度时，dirty升级为read并清空，这样作是为了下一次存值时有机会清空标记为删除的数据，将未删除数据重新加载到dirty，避免数据无限增长。

• func (m *Map) missLocked() {
      // 更新misses
  	m.misses++
      // 未达到升级条件退出
  	if m.misses < len(m.dirty) {
  		return
  	}
      // dirty升级为read
  	m.read.Store(readOnly{m: m.dirty})
      // 清空dirty及未命中次数
  	m.dirty = nil
  	m.misses = 0
  }
  
  func (e *entry) load() (value any, ok bool) {
      // 原子加载entry
  	p := atomic.LoadPointer(&e.p)
      // 标记为删除返回空
  	if p == nil || p == expunged {
  		return nil, false
  	}
  	return *(*any)(p), true
  }
  
  func (m *Map) Load(key any) (value any, ok bool) {
      // 加载read区
  	read, _ := m.read.Load().(readOnly)
      // 获取key
  	e, ok := read.m[key]
      // 没找到且标记dirty存在新的键值
  	if !ok && read.amended {
  		m.mu.Lock()
  		// 再次加载read区
  		read, _ = m.read.Load().(readOnly)
          // 尝试查找key
  		e, ok = read.m[key]
          // 没找到且dirty存在新的键值
  		if !ok && read.amended {
              // dirty查找
  			e, ok = m.dirty[key]
              // 尝试将dirty升级为read
  			m.missLocked()
  		}
  		m.mu.Unlock()
  	}
  	if !ok {
  		return nil, false
  	}
  	return e.load()
  }
  

• 

3.4.Delete

• Delete用于删除指定键值对，其方法内部会调用LoadAndDelete执行删除动作。LoadAndDelete删除时依然遵循双检原则，先尝试从read找，找不到且dirty有新key则从dirty找，找到就删除，并执行entry的e.delete()删除值并返回。

• func (m *Map) LoadAndDelete(key any) (value any, loaded bool) {
      // 尝试从read读
  	read, _ := m.read.Load().(readOnly)
  	e, ok := read.m[key]
      // 不存在且dirty存在新键值对
  	if !ok && read.amended {
  		m.mu.Lock()
          // 再次尝试read读
  		read, _ = m.read.Load().(readOnly)
  		e, ok = read.m[key]
          // read不存在且dirty存在新键值对
  		if !ok && read.amended {
              // 从dirty获取
  			e, ok = m.dirty[key]
              // 从dirty删除key
  			delete(m.dirty, key)
  			// dirty尝试升级read
  			m.missLocked()
  		}
  		m.mu.Unlock()
  	}
      // 找到对应entry执行e.delete
      // 标记删除什么都不做,否则指针置为空
  	if ok {
  		return e.delete()
  	}
  	return nil, false
  }
  
  func (m *Map) Delete(key any) {
      // 代理给LoadAndDelete
  	m.LoadAndDelete(key)
  }
  

• 注意

  1.entry此时不会真正从map删除，只是将指向的值置为nil

  2.涉及到dirty升级为read，可能导致store时触发sync.Map性能退出

3.5.Range

• Range可以遍历sync.Map中所有的键值对，依次调用给定的f(key,value)函数，f()返回值为false时停止遍历。

• func (m *Map) Range(f func(key, value any) bool) {
  	// 加载当前read
  	read, _ := m.read.Load().(readOnly)
      // 如果read不是完整的,
  	if read.amended {
  		m.mu.Lock()
          // 再次加载read
  		read, _ = m.read.Load().(readOnly)
  		// read不是完整的
          if read.amended {
              // dirty升级为read
  			read = readOnly{m: m.dirty}
  			m.read.Store(read)
              // dirty置为nil
  			m.dirty = nil
              // 重置计数
  			m.misses = 0
  		}
  		m.mu.Unlock()
  	}
  
      // 遍历read
  	for k, e := range read.m {
          // 加载value
  		v, ok := e.load()
  		if !ok {
  			continue
  		}
          // 如果f返回false停止
  		if !f(k, v) {
  			break
  		}
  	}
  }
  

• 注意

  1.涉及到dirty升级为read，可能导致store时sync.Map性能退化

3.6.LoadOrStore

• LoadOrStore用于获取或保存一个键值对，key存在时返回当前值，或者存储并返回给定value，loaded返回值表示是否从map中加载的值。

• func (e *entry) tryLoadOrStore(i any) (actual any, loaded, ok bool) {
      // 加载entry
  	p := atomic.LoadPointer(&e.p)
      // 标记删除提前结束
  	if p == expunged {
  		return nil, false, false
  	}
      // 当前值存在直接返回
  	if p != nil {
  		return *(*any)(p), true, true
  	}
  
  	ic := i
  	for {
          // CAS尝试设置新值
  		if atomic.CompareAndSwapPointer(&e.p, nil, unsafe.Pointer(&ic)) {
  			return i, false, true
  		}
          // 加载entry
  		p = atomic.LoadPointer(&e.p)
          // 标记删除退出
  		if p == expunged {
  			return nil, false, false
  		}
          // 返回已经设置的值
  		if p != nil {
  			return *(*any)(p), true, true
  		}
  	}
  }
  
  func (m *Map) LoadOrStore(key, value any) (actual any, loaded bool) {
  	// 加载read区
  	read, _ := m.read.Load().(readOnly)
  	// 从read查找
      if e, ok := read.m[key]; ok {
          // 尝试设置entry值
  		actual, loaded, ok := e.tryLoadOrStore(value)
  		// 找到
          if ok {
  			return actual, loaded
  		}
  	}
  
  	m.mu.Lock()
      // 再次加载read区
  	read, _ = m.read.Load().(readOnly)
      // read存在key
  	if e, ok := read.m[key]; ok {
  		// 将标记删除的entry指向nil
          if e.unexpungeLocked() {
              // entry加入dirty,以便后续设置值
  			m.dirty[key] = e
  		}
          // 读取key值
  		actual, loaded, _ = e.tryLoadOrStore(value)
      // dirty存在key
  	} else if e, ok := m.dirty[key]; ok {
          // 尝试设置新值
  		actual, loaded, _ = e.tryLoadOrStore(value)
  		// 记录misses,dirty尝试升级为read
          m.missLocked()
      // 都不存在
  	} else {
          // read数据完整
  		if !read.amended {
  			// 尝试初始化dirty,为空时将read中未标记删除数据加入dirty
  			m.dirtyLocked()
              // 标记read数据不完整
  			m.read.Store(readOnly{m: read.m, amended: true})
  		}
          // dirty新增键值对
  		m.dirty[key] = newEntry(value)
  		actual, loaded = value, false
  	}
  	m.mu.Unlock()
  
  	return actual, loaded
  }
  

4.ConcurrentMap

4.1.简介

• map不是线程安全的，对一个map进行并发读写会引发panic，sync.Map解决了并发安全问题，专为append-only场景设计，适用于读多写少场景。但写多读少场景，sync.Map性能会出现瓶颈，concurrentMap是更高效的解决方案。concurrentMap是一个并发安全的哈希表，允许多个协程同时进行map读写操作，无需显示的锁或同步原语。concurrentMap的核心原理是使用分片锁，将哈希表划分成多个小的哈希表片段，每个片段占用独立的锁。此外，concurrentMap额外使用了一些优化策略，包括缓存哈希值和桶的地址，减少计算和查找时间，提高读写性能。
• 

4.2.定义

• concurrentMap核心在分片Segment，通过将数据分散到多个Segment独立管理分片数据，基于位运算替代取模，通过segmentShift和segmentMask快速定位键对应的分片。

• type ConcurrentMap struct {
      // 分片数组
      segments []*Segment
      // 用于哈希引擎的单次初始化
  	engChecker *Once
      // 哈希引擎
  	eng        unsafe.Pointer
  	// 键的类型信息
  	kind unsafe.Pointer
  	// 用于分片计算哈希索引的掩码
  	segmentMask int
  	// 用于定位分片的哈希值右移位数
  	segmentShift uint
  }
  
  type Segment struct {
      // 互斥锁
      lock *sync.Mutex
      
      // 当前分片的数据数量(原子操作)
      count int32
      
      // 修改计数器(用于检测并发修改)
      modCount int32
      
      // 扩容阈值(capacity * loadFactor)
      threshold int32
      
      // 指向entry的指针数组(哈希表)
      pTable unsafe.Pointer
      
      // 负载因子
      loadFactor float32
      
      // 指向外层Map
  	m *ConcurrentMap
  }
  
  type Entry struct {
  	// 键
      key   interface{}
  	// 哈希值
      hash  uint32
      // 值(原子读写)
      value unsafe.Pointer
      // 链表指针
  	next  *Entry
  }
  

• 核心实现

  1.concurrentMap将数据分散到多个Segment，每个分片独立管理一部分数据

  2.segmentShift和segmentMask可以快速定位键对应的分片

  3.Segment使用独立的锁管理片内数据，内部的pTable采用链表存储键值对

  4.pTable中存储的其实是entry，entry的value通过unsafe.Pointer和原子操作保证并发安全

  5.key和hash初始化后不可变，确保哈希查找的正确性

  6.cmp的默认容量16，负载因子0.75，默认分片数量16，最大容量2^30，最大分片2^16，默认无锁重试次数2

4.3.初始化

• NewConcurrentMap用于concurrentMap初始化，支持传入自定义容量、负载因子、分片数，未指定时将使用默认值，代理给newConcurrentMap3初始化。

• func NewConcurrentMap(paras ...interface{}) (m *ConcurrentMap) {
  	ok := false
  	cap := DEFAULT_INITIAL_CAPACITY
  	factor := DEFAULT_LOAD_FACTOR
  	concurrent_lvl := DEFAULT_CONCURRENCY_LEVEL
  
      // 初始化容量
  	if len(paras) >= 1 {
  		if cap, ok = paras[0].(int); !ok {...}
  	}
  
      // 初始化负载因子
  	if len(paras) >= 2 {
  		if factor, ok = paras[1].(float32); !ok {...}
  	}
  
      // 初始化分片
  	if len(paras) >= 3 {
  		if concurrent_lvl, ok = paras[2].(int); !ok {...}
  	}
  
      // 调用newConcurrentMap3创建Map
  	m = newConcurrentMap3(cap, factor, concurrent_lvl)
  	return
  }
  
  func newConcurrentMap3(initialCapacity int,
  	loadFactor float32, concurrencyLevel int) (m *ConcurrentMap) {
  	m = &ConcurrentMap{}
  
  	...
  
      // 纠正最大分片
  	if concurrencyLevel > MAX_SEGMENTS {
  		concurrencyLevel = MAX_SEGMENTS
  	}
  
  	// 计算偏移及掩码
  	sshift := 0
  	ssize := 1
      // 找到第一个比分片数大的掩码(2^k)
  	for ssize < concurrencyLevel {
  		sshift++
  		ssize = ssize << 1
  	}
  
      // 计算需要向右偏移位数
  	m.segmentShift = uint(32) - uint(sshift)
      // 计算掩码
  	m.segmentMask = ssize - 1
  
      // 初始化分片数
  	m.segments = make([]*Segment, ssize)
  
      // 纠正最大容量
  	if initialCapacity > MAXIMUM_CAPACITY {
  		initialCapacity = MAXIMUM_CAPACITY
  	}
  
      // 计算每个分片容量(向上取整)
  	c := initialCapacity / ssize
  	if c*ssize < initialCapacity {
  		c++
  	}
      // 分片容量取2^k
  	cap := 1
  	for cap < c {
  		cap <<= 1
  	}
  
      // 初始化每个分片容量及负载因子
  	for i := 0; i < len(m.segments); i++ {
  		m.segments[i] = m.newSegment(cap, loadFactor)
  	}
      // 初始化哈希引擎单词操作的回检器
  	m.engChecker = new(Once)
  	return
  }
  

4.4.获取数据

• Get用于根据key获取数据，获取时先计算key的hash定位分片位置，然后从分片中根据key和hash获取value。

• func (this *Segment) get(key interface{}, hash uint32) interface{} {
      // 原子检查是否存在元素
  	if atomic.LoadInt32(&this.count) != 0 
      	// 获取hash对应的第一个entry
  		e := this.getFirst(hash)
  		for e != nil {
              // hash一致且key一致返回
  			if e.hash == hash && equals(e.key, key) {
  				v := e.Value()
  				if v != nil {
  					//return
  					return v
  				}
                  // 返回再次加载的值(避免被修改)
  				return this.readValueUnderLock(e) 
  			}
              // 向后查找(哈希碰撞)
  			e = e.next
  		}
  	}
  	return nil
  }
  
  func (this *ConcurrentMap) Get(key interface{}) (value interface{}, err error) {
  	...
      // 计算hash值
      // 基础类型直接计算
      // 如果是实现了HashTable接口的包装类型,根据HashTable接口调用计算
      // 如果是其他包装类型,根据类型信息初始化哈希引擎计算
  	if hash, e := hashKey(key, this, false); e != nil {
  		err = e
  	} else {
          // hash计算后定位分片(hash>>shift&mask),从分片中根据key及hash值从链表查找
  		value = this.segmentFor(hash).get(key, hash)
  	}
  	return
  }
  

• 注意

  1.分片中数据采用切片+链表形式，以便根据哈希与位运算快速定位第一个哈希匹配entry

  2.hash计算时，涉及到基础类型的直接计算、实现接口类型的函数调用计算、包装类型的解包装计算

  3.由于可能出现哈希冲突，hash匹配的第一个entry不一定满足，后移继续找

4.5.增加数据

• Put用于将指定key/value存储到concurrentMap，存储时会根据hash定位分片，然后调用put将key/value落分片链表。

• func (this *ConcurrentMap) Put(key interface{}, value interface{}) (oldVal interface{}, err error) {
  	...
      // 计算哈希值(桶Get)
  	if hash, e := hashKey(key, this, false); e != nil {
  		err = e
  	} else {
  		// 写入key/value
  		oldVal = this.segmentFor(hash).put(key, hash, value, false, nil)
  	}
  	return
  }
  

• put方法会判断分片哈希表容量是否溢出，溢出时涉及到扩容及rehash，该操作会把当前分片容量扩容，然后把之前的key/value重新计算哈希放入对应位置。如果未溢出，根据hash计算第一个命中的entry数据，向右遍历查找key是否存在。

• func (this *Segment) put(key interface{}, hash uint32, value interface{}, onlyIfAbsent bool, action func(oldValue interface{}) (newVal interface{})) (oldValue interface{}) {
  	this.lock.Lock()
  	defer this.lock.Unlock()
  
      // 根据容量判断是否需要扩容
  	c := this.count
  	if c > this.threshold {
  		this.rehash()
  	}
  
      // 获取pTable
  	tab := this.table()
      // 计算当前key落点
  	index := hash & uint32(len(tab)-1)
      // 获取第一个匹配hash的entry
  	first := (*Entry)(tab[index])
  	e := first
  	
      // 查找key是否出现过
  	for e != nil && (e.hash != hash || !equals(e.key, key)) {
  		e = e.next
  	}
  
  	if action == nil {
          // 出现过,更新值
  		if e != nil {
  			oldValue = e.fastValue()
  			if !onlyIfAbsent {
                  // 存储新值
  				e.storeValue(&value)
  			}
  		} else {
              // 容量增1
  			c++
  			oldValue = nil
              // 记录修改次数
  			this.modCount++
              // 保存`key/value`,作为新的链表头节点
  			tab[index] = unsafe.Pointer(&Entry{key, hash, unsafe.Pointer(&value), first})
  			// 更新容量
              atomic.StoreInt32(&this.count, c)
  		}
  	} else {
  		if e != nil {
  			oldValue = e.fastValue()
  		} else {
  			c++
  			oldValue = nil
  		}
  
          // 根据action回调转换值
  		newVal := action(oldValue)
  		if newVal != nil {
  			if oldValue == nil {
                  // 新增`key/value`作为新的链表头节点
  				e = &Entry{key, hash, unsafe.Pointer(&value), first}
  				tab[index] = unsafe.Pointer(e)
  				this.modCount++
  				atomic.StoreInt32(&this.count, c) // atomic write 这里可以保证对modCount和tab的修改不会被reorder到this.count之后
  			}
              // 更新value
  			e.storeValue(&newVal)
  		} else if e != nil {
  			// 当前位置entry不是空的回退容量
  			c--
  			this.modCount++
              // 重新获取第一个匹配hash元素
  			newFirst := e.next
  			for p := first; p != e; p = p.next {
  				newFirst = &Entry{p.key, p.hash, p.value, newFirst}
  			}
              // 设置当前key值
  			tab[index] = unsafe.Pointer(newFirst)
  			atomic.StoreInt32(&this.count, c) //this.count = c
  		}
  	}
  	return
  }
  

4.6.存在判断

• ContainsKey用于判断key是否存在，其内部调用segment.containsKey实现，主要根据hash确认分片内落点后遍历链表，判断key对应entry是否存在。

• func (this *ConcurrentMap) ContainsKey(key interface{}) (found bool, err error) {
  	...
  	// 计算hash值
  	if hash, e := hashKey(key, this, false); e != nil {
  		err = e
  	} else {
  		// 调用segment.containsKey判断是否存在
  		found = this.segmentFor(hash).containsKey(key, hash)
  	}
  	return
  }
  

• containsKey会获取链表根节点，然后遍历所有节点判断key对应的entry是否存在。

• func (this *Segment) containsKey(key interface{}, hash uint32) bool {
      // 原子加载容量
  	if atomic.LoadInt32(&this.count) != 0 { 
          // 获取头节点
  		e := this.getFirst(hash)
  		for e != nil {
              // 判断key对应entry是否存在
  			if e.hash == hash && equals(e.key, key) {
  				return true
  			}
              // 向后移动
  			e = e.next
  		}
  	}
  	return false
  }
  

4.7.移除数据

• Remove用于移除指定key，移除过程中，会将部分节点翻转到头节点，直至遇到待删除节点。

• func (this *ConcurrentMap) Remove(key interface{}) (oldVal interface{}, err error) {
  	...
  	// 计算hash值
  	if hash, e := hashKey(key, this, false); e != nil {
  		err = e
  	} else {
  		// 移除分片内key对应entry
  		oldVal = this.segmentFor(hash).remove(key, hash, nil)
  	}
  	return
  }
  
  func (this *Segment) remove(key interface{}, hash uint32, value interface{}) (oldValue interface{}) {
  	this.lock.Lock()
  	defer this.lock.Unlock()
  
  	c := this.count - 1
  	tab := this.table()
      // 计算key落点,拿到头节点
  	index := hash & uint32(len(tab)-1)
  	first := (*Entry)(tab[index])
  	e := first
  	// 从头节点开始找key对应节点
  	for e != nil && (e.hash != hash || !equals(e.key, key)) {
  		e = e.next
  	}
  
  	if e != nil {
          // 找到后再校验value
  		v := e.fastValue()
  		if value == nil || value == v {
  			oldValue = v
              // 非key对应节点移至头节点(此处实现未修改e>next指针,可能造成内存泄漏?)
  			this.modCount++
  			newFirst := e.next
  			for p := first; p != e; p = p.next {
  				newFirst = &Entry{p.key, p.hash, p.value, newFirst}
  			}
              // 重新保存key落点链表
  			tab[index] = unsafe.Pointer(newFirst)
  			atomic.StoreInt32(&this.count, c)
  		}
  	}
  	return
  }