アトミック操作とは何ですか? Go におけるアトミック操作の詳細な分析

func TestAtomic(t *testing.T) {
	var sum = 0
	var wg sync.WaitGroup
	wg.Add(1000)

	// 启动 1000 个协程，每个协程对 sum 做加法操作
	for i := 0; i < 1000; i++ {
		go func() {
			defer wg.Done()
			sum++
		}()
	}

	// 等待所有的协程都执行完毕
	wg.Wait()
	fmt.Println(sum) // 这里输出多少呢？
}

func TestAtomic1(t *testing.T) {
	// 将 sum 的类型改成 int32，因为原子操作只能针对 int32、int64、uint32、uint64、uintptr 这几种类型
	var sum int32 = 0
	var wg sync.WaitGroup
	wg.Add(1000)

    // 启动 1000 个协程，每个协程对 sum 做加法操作
	for i := 0; i < 1000; i++ {
		go func() {
			defer wg.Done()
			// 将 sum++ 改成下面这样
			atomic.AddInt32(&sum, 1)
		}()
	}

	wg.Wait()
	fmt.Println(sum) // 输出 1000
}

func AddInt32(addr *int32, delta int32) (new int32)
func AddInt64(addr *int64, delta int64) (new int64)
func AddUint32(addr *uint32, delta uint32) (new uint32)
func AddUint64(addr *uint64, delta uint64) (new uint64)
func AddUintptr(addr *uintptr, delta uintptr) (new uintptr)

func CompareAndSwapInt32(addr *int32, old, new int32) (swapped bool)
func CompareAndSwapInt64(addr *int64, old, new int64) (swapped bool)
func CompareAndSwapUint32(addr *uint32, old, new uint32) (swapped bool)
func CompareAndSwapUint64(addr *uint64, old, new uint64) (swapped bool)
func CompareAndSwapUintptr(addr *uintptr, old, new uintptr) (swapped bool)
func CompareAndSwapPointer(addr *unsafe.Pointer, old, new unsafe.Pointer) (swapped bool)

func LoadInt32(addr *int32) (val int32)
func LoadInt64(addr *int64) (val int64)
func LoadUint32(addr *uint32) (val uint32)
func LoadUint64(addr *uint64) (val uint64)
func LoadUintptr(addr *uintptr) (val uintptr)
func LoadPointer(addr *unsafe.Pointer) (val unsafe.Pointer)

func StoreInt32(addr *int32, val int32)
func StoreInt64(addr *int64, val int64)
func StoreUint32(addr *uint32, val uint32)
func StoreUint64(addr *uint64, val uint64)
func StoreUintptr(addr *uintpre, val uintptr)
func StorePointer(addr *unsafe.Pointer, val unsafe.Pointer)

func SwapInt32(addr *int32, new int32) (old int32)
func SwapInt64(addr *int64, new int64) (old int64)
func SwapUint32(addr *uint32, new uint32) (old uint32)
func SwapUint64(addr *uint64, new uint64) (old uint64)
func SwapUintptr(addr *uintptr, new uintptr) (old uintptr)
func SwapPointer(addr *unsafe.Pointer, new unsafe.Pointer) (old unsafe.Pointer)

// Value 提供一致类型值的原子加载和存储。
type Value struct {
	v any
}

// Load 返回由最近的 Store 设置的值。
func (v *Value) Load() (val any) {
	// atomic.Value 转换为 efaceWords
	vp := (*efaceWords)(unsafe.Pointer(v))

	// 判断 atomic.Value 的类型
	typ := LoadPointer(&vp.typ)
	// 第一次 Store 还没有完成，直接返回 nil
	if typ == nil || typ == unsafe.Pointer(&firstStoreInProgress) {
		// firstStoreInProgress 是一个特殊的变量，存储到 typ 中用来表示第一次 Store 还没有完成
		return nil
	}

	// 获取 atomic.Value 的值
	data := LoadPointer(&vp.data)
	// 将 val 转换为 efaceWords 类型
	vlp := (*efaceWords)(unsafe.Pointer(&val))
	// 分别赋值给 val 的 typ 和 data
	vlp.typ = typ
	vlp.data = data
	return
}

// 表示一个 interface{}/any 类型
type efaceWords struct {
	typ  unsafe.Pointer
	data unsafe.Pointer
}

// Store 将 Value 的值设置为 val。
func (v *Value) Store(val any) {
	// 不能存储 nil 值
	if val == nil {
		panic("sync/atomic: store of nil value into Value")
	}
	// atomic.Value 转换为 efaceWords
	vp := (*efaceWords)(unsafe.Pointer(v))
	// val 转换为 efaceWords
	vlp := (*efaceWords)(unsafe.Pointer(&val))
	
	// 自旋进行原子操作，这个过程不会很久，开销相比互斥锁小
	for {
		// LoadPointer 可以保证获取到的是最新的
		typ := LoadPointer(&vp.typ)
		// 第一次 store 的时候 typ 还是 nil，说明是第一次 store
		if typ == nil {
			// 尝试开始第一次 Store。
			// 禁用抢占，以便其他 goroutines 可以自旋等待完成。
			// （如果允许抢占，那么其他 goroutine 自旋等待的时间可能会比较长，因为可能会需要进行协程调度。）
			runtime_procPin()
			// 抢占失败，意味着有其他 goroutine 成功 store 了，允许抢占，再次尝试 Store
			// 这也是一个原子操作。
			if !CompareAndSwapPointer(&vp.typ, nil, unsafe.Pointer(&firstStoreInProgress)) {
				runtime_procUnpin()
				continue
			}
			// 完成第一次 store
			// 因为有 firstStoreInProgress 标识的保护，所以下面的两个原子操作是安全的。
			StorePointer(&vp.data, vlp.data) // 存储值（原子操作）
			StorePointer(&vp.typ, vlp.typ)   // 存储类型（原子操作）
			runtime_procUnpin()              // 允许抢占
			return
		}

		// 另外一个 goroutine 正在进行第一次 Store。自旋等待。
		if typ == unsafe.Pointer(&firstStoreInProgress) {
			continue
		}

		// 第一次 Store 已经完成了，下面不是第一次 Store 了。
		// 需要检查当前 Store 的类型跟第一次 Store 的类型是否一致，不一致就 panic。
		if typ != vlp.typ {
			panic("sync/atomic: store of inconsistently typed value into Value")
		}

		// 后续的 Store 只需要 Store 值部分就可以了。
		// 因为 atomic.Value 只能保存一种类型的值。
		StorePointer(&vp.data, vlp.data)
		return
	}
}

// Swap 将 Value 的值设置为 new 并返回旧值。
func (v *Value) Swap(new any) (old any) {
	// 不能存储 nil 值
	if new == nil {
		panic("sync/atomic: swap of nil value into Value")
	}

	// atomic.Value 转换为 efaceWords
	vp := (*efaceWords)(unsafe.Pointer(v))
	// new 转换为 efaceWords
	np := (*efaceWords)(unsafe.Pointer(&new))
	
	// 自旋进行原子操作，这个过程不会很久，开销相比互斥锁小
	for {
		// 下面这部分代码跟 Store 一样，不细说了。
		// 这部分代码是进行第一次存储的代码。
		typ := LoadPointer(&vp.typ)
		if typ == nil {
			runtime_procPin()
			if !CompareAndSwapPointer(&vp.typ, nil, unsafe.Pointer(&firstStoreInProgress)) {
				runtime_procUnpin()
				continue
			}
			StorePointer(&vp.data, np.data)
			StorePointer(&vp.typ, np.typ)
			runtime_procUnpin()
			return nil
		}
		if typ == unsafe.Pointer(&firstStoreInProgress) {
			continue
		}
		if typ != np.typ {
			panic("sync/atomic: swap of inconsistently typed value into Value")
		}

		// ---- 下面是 Swap 的特有逻辑 ----
		// op 是返回值
		op := (*efaceWords)(unsafe.Pointer(&old))
		// 返回旧的值
		op.typ, op.data = np.typ, SwapPointer(&vp.data, np.data)
		return old
	}
}

// CompareAndSwap 比较并交换。
func (v *Value) CompareAndSwap(old, new any) (swapped bool) {
	// 注意：old 是可以为 nil 的，new 不能为 nil。
	// old 是 nil 表示是第一次进行 Store 操作。
	if new == nil {
		panic("sync/atomic: compare and swap of nil value into Value")
	}

	// atomic.Value 转换为 efaceWords
	vp := (*efaceWords)(unsafe.Pointer(v))
	// new 转换为 efaceWords
	np := (*efaceWords)(unsafe.Pointer(&new))
	// old 转换为 efaceWords
	op := (*efaceWords)(unsafe.Pointer(&old))

	// old 和 new 类型必须一致，且不能为 nil
	if op.typ != nil && np.typ != op.typ {
		panic("sync/atomic: compare and swap of inconsistently typed values")
	}

	// 自旋进行原子操作，这个过程不会很久，开销相比互斥锁小
	for {
		// LoadPointer 可以保证获取到的 typ 是最新的
		typ := LoadPointer(&vp.typ)
		if typ == nil { // atomic.Value 是 nil，还没 Store 过
			// 准备进行第一次 Store，但是传递进来的 old 不是 nil，compare 这一步就失败了。直接返回 false
			if old != nil {
				return false
			}

			// 下面这部分代码跟 Store 一样，不细说了。 
			// 这部分代码是进行第一次存储的代码。
			runtime_procPin()
			if !CompareAndSwapPointer(&vp.typ, nil, unsafe.Pointer(&firstStoreInProgress)) {
				runtime_procUnpin()
				continue
			}
			StorePointer(&vp.data, np.data)
			StorePointer(&vp.typ, np.typ)
			runtime_procUnpin()
			return true
		}
		if typ == unsafe.Pointer(&firstStoreInProgress) {
			continue
		}
		if typ != np.typ {
			panic("sync/atomic: compare and swap of inconsistently typed value into Value")
		}

		// 通过运行时相等性检查比较旧版本和当前版本。
		// 这允许对值类型进行比较，这是包函数所没有的。
		// 下面的 CompareAndSwapPointer 仅确保 vp.data 自 LoadPointer 以来没有更改。
		data := LoadPointer(&vp.data)
		var i any
		(*efaceWords)(unsafe.Pointer(&i)).typ = typ
		(*efaceWords)(unsafe.Pointer(&i)).data = data
		if i != old { // atomic.Value 跟 old 不相等
			return false
		}
		// 只做 val 部分的 cas 操作
		return CompareAndSwapPointer(&vp.data, data, np.data)
	}
}

// An Int32 is an atomic int32. The zero value is zero.
type Int32 struct {
	_ noCopy
	v int32
}

// Load atomically loads and returns the value stored in x.
func (x *Int32) Load() int32 { return LoadInt32(&x.v) }

// Store atomically stores val into x.
func (x *Int32) Store(val int32) { StoreInt32(&x.v, val) }

// Swap atomically stores new into x and returns the previous value.
func (x *Int32) Swap(new int32) (old int32) { return SwapInt32(&x.v, new) }

// CompareAndSwap executes the compare-and-swap operation for x.
func (x *Int32) CompareAndSwap(old, new int32) (swapped bool) {
	return CompareAndSwapInt32(&x.v, old, new)
}

// 系统信息 cpu: Intel(R) Core(TM) i7-8700 CPU @ 3.20GHz
// 10.13 ns/op
func BenchmarkMutex(b *testing.B) {
   var mu sync.Mutex

   for i := 0; i < b.N; i++ {
      mu.Lock()
      mu.Unlock()
   }
}

// 5.849 ns/op
func BenchmarkAtomic(b *testing.B) {
   var sum atomic.Uint64

   for i := 0; i < b.N; i++ {
      sum.Add(uint64(1))
   }
}

// 删除 entry
func (e *entry) delete() (value any, ok bool) {
	for {
		p := e.p.Load()
		// 已经被删除了，不需要再删除
		if p == nil || p == expunged {
			return nil, false
		}
		// 删除成功
		if e.p.CompareAndSwap(p, nil) {
			return *p, true
		}
	}
}

// 如果条目尚未删除，trySwap 将交换一个值。
func (e *entry) trySwap(i *any) (*any, bool) {
	for {
		p := e.p.Load()
		// 已经被删除了
		if p == expunged {
			return nil, false
		}
		// swap 成功
		if e.p.CompareAndSwap(p, i) {
			return p, true
		}
	}
}

// WaitGroup 的 Add 函数：高 32 位加上 delta
state := wg.state.Add(uint64(delta) << 32)

// WaitGroup 的 Wait 函数：低 32 位加 1
// 等待者的数量加 1
wg.state.CompareAndSwap(state, state+1)

func TestCas(t *testing.T) {
	var sum int32 = 0
	var wg sync.WaitGroup
	wg.Add(1000)

	for i := 0; i < 1000; i++ {
		go func() {
			defer wg.Done()
			// 这一行是有可能会失败的
			atomic.CompareAndSwapInt32(&sum, sum, sum+1)
		}()
	}

	wg.Wait()
	fmt.Println(sum) // 不是 1000
}

func TestCas(t *testing.T) {
	var sum int32 = 0
	var wg sync.WaitGroup
	wg.Add(1000)

	for i := 0; i < 1000; i++ {
		go func() {
			defer wg.Done()
			
			// cas 失败的时候，重新获取 sum 的值进行计算。
			// cas 成功则返回。
			for {
				if atomic.CompareAndSwapInt32(&sum, sum, sum+1) {
					return
				}
			}
		}()
	}

	wg.Wait()
	fmt.Println(sum)
}