Go中由WaitGroup引发对内存对齐思考-技术圈

本文使用的go的源码是1.14.4

WaitGroup介绍

WaitGroup 提供了三个方法：

    func (wg *WaitGroup) Add(delta int)
    func (wg *WaitGroup) Done()
    func (wg *WaitGroup) Wait()

Add，用来设置 WaitGroup 的计数值；
Done，用来将 WaitGroup 的计数值减 1，其实就是调用了 Add(-1)；
Wait，调用这个方法的 goroutine 会一直阻塞，直到 WaitGroup 的计数值变为 0。

例子我就不举了，网上是很多的，下面我们直接进入正题。

解析

type noCopy struct{}

type WaitGroup struct {
    // 避免复制使用的一个技巧，可以告诉vet工具违反了复制使用的规则
 noCopy noCopy
 // 一个复合值，用来表示waiter数、计数值、信号量
 state1 [3]uint32
}
// 获取state的地址和信号量的地址
func (wg *WaitGroup) state() (statep *uint64, semap *uint32) {
 if uintptr(unsafe.Pointer(&wg.state1))%8 == 0 {
  // 如果地址是64bit对齐的，数组前两个元素做state，后一个元素做信号量
  return (*uint64)(unsafe.Pointer(&wg.state1)), &wg.state1[2]
 } else {
  // 如果地址是32bit对齐的，数组后两个元素用来做state，它可以用来做64bit的原子操作，第一个元素32bit用来做信号量
  return (*uint64)(unsafe.Pointer(&wg.state1[1])), &wg.state1[0]
 }
}

这里刚开始，WaitGroup就秀了一把肌肉，让我们看看大牛是怎么写代码的，思考一个原子操作在不同架构平台上是怎么操作的，在看state方法里面为什么要这么做之前，我们先来看看内存对齐。

内存对齐

在维基百科https://en.wikipedia.org/wiki/Data_structure_alignment上我们可以看到对于内存对齐的定义：

A memory address a is said to be n-byte aligned when a is a multiple of n bytes (where n is a power of 2).

简而言之，现在的CPU访问内存的时候是一次性访问多个bytes，比如32位架构一次访问4bytes，该处理器只能从地址为4的倍数的内存开始读取数据，所以要求数据在存放的时候首地址的值是4的倍数存放，这就是所谓的内存对齐。

由于找不到Go语言的对齐规则，我对照了一下C语言的内存对齐的规则，可以和Go语言匹配的上，所以先参照下面的规则。

内存对齐遵循下面三个原则：

结构体变量的起始地址能够被其最宽的成员大小整除；
结构体每个成员相对于起始地址的偏移能够被其自身大小整除，如果不能则在前一个成员后面补充字节；
结构体总体大小能够被最宽的成员的大小整除，如不能则在后面补充字节；

通过下面的例子来实操一下内存对齐：

在32位架构中，int8占1byte，int32占4bytes，int16占2bytes。

type A struct {
 a int8
 b int32
 c int16
}

type B struct {
 a int8
 c int16
 b int32
}

func main() {

 fmt.Printf("arrange fields to reduce size:\n"+
  "A align: %d, size: %d\n" ,
  unsafe.Alignof(A{}), unsafe.Sizeof(A{}) )

 fmt.Printf("arrange fields to reduce size:\n"+
  "B align: %d, size: %d\n" ,
  unsafe.Alignof(B{}), unsafe.Sizeof(B{}) )
}

//output:
//arrange fields to reduce size:
//A align: 4, size: 12
//arrange fields to reduce size:
//B align: 4, size: 8

下面以在32位的架构中运行为例子：

在32位架构的系统中默认的对齐大小是4bytes。

假设结构体A中a的起始地址为0x0000，能够被最宽的数据成员大小4bytes（int32）整除，所以从0x0000开始存放占用一个字节即0x0000~0x0001；b是int32，占4bytes，所以要满足条件2，需要在a后面padding3个byte，从0x0004开始；c是int16，占2bytes故从0x0008开始占用两个字节，即0x0008~0x0009；此时整个结构体占用的空间是0x0000~0x0009占用10个字节，10%4 ！= 0, 不满足第三个原则，所以需要在后面补充两个字节，即最后内存对齐后占用的空间是0x0000~0x000B，一共12个字节。

同理，相比结构体B则要紧凑些：

WaitGroup中state方法的内存对齐

在讲之前需要注意的是noCopy是一个空的结构体，大小为0，不需要做内存对齐，所以大家在看的时候可以忽略这个字段。

在WaitGroup里面，使用了uint32的数组来构造state1字段，然后根据系统的位数的不同构造不同的返回值，下面我面先来说说怎么通过sate1这个字段构建waiter数、计数值、信号量的。

首先unsafe.Pointer来获取state1的地址值然后转换成uintptr类型的，然后判断一下这个地址值是否能被8整除，这里通过地址 mod 8的方式来判断地址是否是64位对齐。

因为有内存对齐的存在，在64位架构里面WaitGroup结构体state1起始的位置肯定是64位对齐的，所以在64位架构上用state1前两个元素并成uint64来表示statep，state1最后一个元素表示semap；

那么64位架构上面获取state1的时候能不能第一个元素表示semap，后两个元素拼成64位返回呢？

答案自然是不可以，因为uint32的对齐保证是4bytes，64位架构中一次性处理事务的一个固定长度是8bytes，如果用state1的后两个元素表示一个64位字的字段的话CPU需要读取内存两次，不能保证原子性。

但是在32位架构里面，一个字长是4bytes，要操作64位的数据分布在两个数据块中，需要两次操作才能完成访问。如果两次操作中间有可能别其他操作修改，不能保证原子性。

同理32位架构想要原子性的操作8bytes，需要由调用方保证其数据地址是64位对齐的，否则原子访问会有异常，我们在这里https://golang.org/pkg/sync/atomic/#pkg-note-BUG可以看到描述：

On ARM, x86-32, and 32-bit MIPS, it is the caller's responsibility to arrange for 64-bit alignment of 64-bit words accessed atomically. The first word in a variable or in an allocated struct, array, or slice can be relied upon to be 64-bit aligned.

所以为了保证64位字对齐，只能让变量或开辟的结构体、数组和切片值中的第一个64位字可以被认为是64位字对齐。但是在使用WaitGroup的时候会有嵌套的情况，不能保证总是让WaitGroup存在于结构体的第一个字段上，所以我们需要增加填充使它能对齐64位字。

在32位架构中，WaitGroup在初始化的时候，分配内存地址的时候是随机的，所以WaitGroup结构体state1起始的位置不一定是64位对齐，可能会是：uintptr(unsafe.Pointer(&wg.state1))%8 = 4，如果出现这样的情况，那么就需要用state1的第一个元素做padding，用state1的后两个元素合并成uint64来表示statep。

小结

这里小结一下，因为为了完成上面的这篇内容实在是查阅了很多资料，才得出这样的结果。所以这里小结一下，在64位架构中，CPU每次操作的字长都是8bytes，编译器会自动帮我们把结构体的第一个字段的地址初始化成64位对齐的，所以64位架构上用state1前两个元素并成uint64来表示statep，state1最后一个元素表示semap；

然后在32位架构中，在初始化WaitGroup的时候，编译器只能保证32位对齐，不能保证64位对齐，所以通过uintptr(unsafe.Pointer(&wg.state1))%8判断是否等于0来看state1内存地址是否是64位对齐，如果是，那么也和64位架构一样，用state1前两个元素并成uint64来表示statep，state1最后一个元素表示semap，否则用state1的第一个元素做padding，用state1的后两个元素合并成uint64来表示statep。

如果我说错了，欢迎来diss我，我觉得我需要学习的地方还有很多。

Add 方法

func (wg *WaitGroup) Add(delta int) {
 // 获取状态值
 statep, semap := wg.state()
 ...
 // 高32bit是计数值v，所以把delta左移32，增加到计数上
 state := atomic.AddUint64(statep, uint64(delta)<<32)
 // 获取计数器的值
 v := int32(state >> 32)
 // 获取waiter的值
 w := uint32(state)
 ...
 // 任务计数器不能为负数
 if v < 0 {
  panic("sync: negative WaitGroup counter")
 }
 // wait不等于0说明已经执行了Wait，此时不容许Add
 if w != 0 && delta > 0 && v == int32(delta) {
  panic("sync: WaitGroup misuse: Add called concurrently with Wait")
 }
 // 计数器的值大于或者没有waiter在等待,直接返回
 if v > 0 || w == 0 {
  return
 } 
 if *statep != state {
  panic("sync: WaitGroup misuse: Add called concurrently with Wait")
 }
 // 此时，counter一定等于0，而waiter一定大于0
 // 先把counter置为0，再释放waiter个数的信号量
 *statep = 0
 for ; w != 0; w-- {
  //释放信号量，执行一次释放一个，唤醒一个等待者
  runtime_Semrelease(semap, false, 0)
 }
}

add方法首先会调用state方法获取statep、semap的值。statep是一个uint64类型的值，高32位用来记录add方法传入的delta值之和；低32位用来表示调用wait方法等待的goroutine的数量，也就是waiter的数量。如下：

add方法会调用atomic.AddUint64方法将传入的delta左移32位，也就是将counter加上delta的值；
因为计数器counter可能为负数，所以int32来获取计数器的值，waiter不可能为负数，所以使用uint32来获取；
接下来就是一系列的校验，v不能小于零表示任务计数器不能为负数，否则会panic；w不等于，并且v的值等于delta表示wait方法先于add方法执行，此时也会panic，因为waitgroup不允许调用了Wait方法后还调用add方法；
v大于零或者w等于零直接返回，说明这个时候不需要释放waiter，所以直接返回；
*statep != state到了这个校验这里，状态只能是waiter大于零并且counter为零。当waiter大于零的时候是不允许再调用add方法，counter为零的时候也不能调用wait方法，所以这里使用state的值和内存的地址值进行比较，查看是否调用了add或者wait导致state变动，如果有就是非法调用会引起panic；
最后将statep值重置为零，然后释放所有的waiter；

Wait方法

func (wg *WaitGroup) Wait() {
 statep, semap := wg.state()
 ...
 for {
  state := atomic.LoadUint64(statep)
  // 获取counter
  v := int32(state >> 32)
  // 获取waiter
  w := uint32(state)
  // counter为零,不需要等待直接返回
  if v == 0 {
   ...
   return
  }
  // 使用CAS将waiter加1
  if atomic.CompareAndSwapUint64(statep, state, state+1) {
   ...
   // 挂起等待唤醒
   runtime_Semacquire(semap)
   // 唤醒之后statep不为零,表示WaitGroup又被重复使用,这回panic
   if *statep != 0 {
    panic("sync: WaitGroup is reused before previous Wait has returned")
   }
   ...
          // 直接返回   
   return
  }
 }
}

Wait方法首先也是调用state方法获取状态值；
进入for循环之后Load statep的值，然后分别获取counter和counter；
如果counter已经为零了，那么直接返回不需要等待；
counter不为零，那么使用CAS将waiter加1，由于CAS可能失败，所以for循环会再次的回到这里进行CAS，直到成功；
调用runtime_Semacquire挂起等待唤醒；
*statep != 0唤醒之后statep不为零,表示WaitGroup又被重复使用,这会panic。需要注意的是waitgroup并不是不让重用，而是不能在wait方法还没运行完就开始重用。

waitgroup使用小结

看完了waitgroup的add方法与wait方法，我们发现里面有很多校验，使用不当会导致panic，所以我们需要总结一下如何正确使用：

不能将计数器设置为负数，否则会发生panic；注意有两种方式会导致计数器为负数，一是调用 Add 的时候传递一个负数，第二是调用 Done 方法的次数过多，超过了 WaitGroup 的计数值；
在使用 WaitGroup 的时候，一定要等所有的 Add 方法调用之后再调用 Wait，否则就可能导致 panic；
wait还没结束就重用 WaitGroup。WaitGroup是可以重用的，但是需要等上一批的goroutine 都调用wait完毕后才能继续重用WaitGroup；

总结

waitgroup里面的代码实际上是非常的简单的，这篇文章主要是由waitgroup引入了内存对齐这个概念。由waitgroup带我们看了在实际的代码中是如何利用内存对齐这个概念的，以及如何在32为操作系统中原子性的操作64位长的字段。

除了内存对齐的概念以外通过源码我们也了解到了使用waitgroup的时候需要怎么做才是符合规范的，不会引发panic。

Reference

http://blog.newbmiao.com/2020/02/10/dig101-golang-struct-memory-align.html

https://gfw.go101.org/article/memory-layout.html

https://golang.org/pkg/sync/atomic/#pkg-note-BUG

https://en.wikipedia.org/wiki/Data_structure_alignment

https://www.zhihu.com/question/27862634

推荐阅读

使用context、WaitGroup优雅处理goroutine

福利

我为大家整理了一份从入门到进阶的Go学习资料礼包，包含学习建议：入门看什么，进阶看什么。关注公众号「polarisxu」，回复 ebook 获取；还可以回复「进群」，和数万 Gopher 交流学习。