Go语言中常见100问题-#92 Writing concurrent code that leads to false ...

数据小冰

发布于 2023-12-13 14:24:58

1490

发布于 2023-12-13 14:24:58

编写伪共享的并发代码

在Go语言中常见100问题-#91 Not understanding CPU caches中讨论了缓存基本概念，可以看到一些特定的缓存(通常是 L1 和 L2)并不在所有逻辑内核之间共享，而是属于一个特定物理内核。这种特殊性会产生一些影响，比如并发时伪共享，这会导致性能显著下降。下面通过一个具体例子来说明什么是伪共享，以及如何预防这种情况发生。

定义两个结构体 Input 和 Result， count 函数接收一个 Input 切片，分别对切片中的a和b进行求和，求和结果存储在sumA和sumB中。

type Input struct {
 a int64
 b int64
}

type Result struct {
 sumA int64
 sumB int64
}

编写并发程序开启两个goroutine，分别对a和b进行求和，代码如下。

func count(inputs []Input) Result1 {
 wg := sync.WaitGroup{}
 wg.Add(2)

 result := Result1{}

 go func() {
  for i := 0; i < len(inputs); i++ {
   result.sumA += inputs[i].a
  }
  wg.Done()
 }()

 go func() {
  for i := 0; i < len(inputs); i++ {
   result.sumB += inputs[i].b
  }
  wg.Done()
 }()

 wg.Wait()
 return result
}

两个goroutine迭代的是不同字段，不存在数据竞争问题，求和的结果也是保存在不同的字段sumA和sumB中，但是上述存在伪共享问题，导致程序性能下降。

Result结构体在内存的中的布局如下图，因为sumA和sumB是相邻的字段，所以在内存中是连续分配的。在大多数情况下，它们会分配到同一个内存块，

现在假设运行上述程序的机器含有两个内核，在大多数情况下，应该在不同的内核上调度两个线程。因此，如果CPU决定将这个内存块复制到cache line时，会被复制两次，如下图，每个内存块都被复制到核心0核心1上的缓存行。

因为L1D（L1数据）是针对每个内核的，core 0 和 core 1 都要进行一次缓存行复制，上述程序中，每个goroutine更新自己的变量：一个更新sumA，另一个更新sumB. CPU需要保证这些缓存行的一致性，例如，如果一个goroutine更新sumA，而另一个读取sumA, 期望应用程序能够获取到最新的值。但是，本文中的程序每个goroutine访问的是各自的变量，而不是共享变量，由于CPU跟踪的粒度不是变量，而是缓存行。所以当一个缓存行在多个内核之间共享，只要有一个goroutine进行写操作，整个缓存行都会失效。即使更新在逻辑上是独立的，也会发生这种情况（例如这里的sumA和sumB）。这是伪共享，会降低程序性能。CPU内部使用MESI协议保证缓存一致性，它跟踪每个高速缓存行，标记它已修改、独占、共享或无效。

如何解决上述的伪共享问题呢？主要有两种方法，第一种方法是确保sumA和sumB不属于同一个缓存行。在结构体Result中的sumA和sumB字段中间添加填充，所谓填充就是额外分配内存，因为int64占用大小为8字节，缓存行大小为64字节，所以需要填充 64-8=56字节。

type Result struct {
 sumA int64
 _    [56]byte
 sumB int64
}

如下图，经过填充后，sumA和sumB不属于同一个缓存行。

编写性能测试代码，对填充前后进行基准测试，完整测试代码如下。

package main

import "sync"

type Input struct {
 a int64
 b int64
}

type Result1 struct {
 sumA int64
 sumB int64
}

type Result2 struct {
 sumA int64
 _    [56]byte
 sumB int64
}

func count1(inputs []Input) Result1 {
 wg := sync.WaitGroup{}
 wg.Add(2)

 result := Result1{}

 go func() {
  for i := 0; i < len(inputs); i++ {
   result.sumA += inputs[i].a
  }
  wg.Done()
 }()

 go func() {
  for i := 0; i < len(inputs); i++ {
   result.sumB += inputs[i].b
  }
  wg.Done()
 }()

 wg.Wait()
 return result
}

func count2(inputs []Input) Result2 {
 wg := sync.WaitGroup{}
 wg.Add(2)

 result := Result2{}

 go func() {
  for i := 0; i < len(inputs); i++ {
   result.sumA += inputs[i].a
  }
  wg.Done()
 }()

 go func() {
  for i := 0; i < len(inputs); i++ {
   result.sumB += inputs[i].b
  }
  wg.Done()
 }()

 wg.Wait()
 return result
}

基准测试代码

package main

import "testing"

const n = 1_000_000

var globalResult1 Result1

func BenchmarkCount1(b *testing.B) {
 var local Result1
 for i := 0; i < b.N; i++ {
  b.StopTimer()
  inputs := make([]Input, n)
  b.StartTimer()
  local = count1(inputs)
 }
 globalResult1 = local
}

var globalResult2 Result2

func BenchmarkCount2(b *testing.B) {
 var local Result2
 for i := 0; i < b.N; i++ {
  b.StopTimer()
  inputs := make([]Input, n)
  b.StartTimer()
  local = count2(inputs)
 }
 globalResult2 = local
}

测试结果如下，可以看到填充后相比填充前大约有30%的提升。