// Copyright 2011 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package sync
import (
"internal/race"
"sync/atomic"
"unsafe"
)
// A WaitGroup waits for a collection of goroutines to finish.
WaitGroup等待一组goroutine完成。
// The main goroutine calls Add to set the number of
goroutine通过调用 Add(),增加等待的goroutines的数量。
// goroutines to wait for. Then each of the goroutines
goroutines组中的goroutine等执行完成后，会调用Done()。
// runs and calls Done when finished. At the same time,
同时，
// Wait can be used to block until all goroutines have finished.
// 可以用wait()等待所有goroutines完成。
// A WaitGroup must not be copied after first use.
WaitGroup 不要被复制使用。
type WaitGroup struct {
// noCopy可以嵌入到不能复制的结构中
//首次使用后
noCopy noCopy
// 64-bit value: high 32 bits are counter, low 32 bits are waiter count.
64 字节：高32位是counter, 低32位是等待者的数量。
// 64-bit atomic operations require 64-bit alignment, but 32-bit
64位院子操作需要64位对齐，但是32位系统编译器无法保证它，
// compilers do not ensure it. So we allocate 12 bytes and then use
所以我们使用12字节，然后使用8字节作为状态，
// the aligned 8 bytes in them as state, and the other 4 as storage
其余四个字节用在放信号。
// for the sema.
state1 [3]uint32 // 12字节
}
// state returns pointers to the state and sema fields stored within wg.state1.
// state()用来返回，state,sema(信号)。
// 区分32位系统和64为系统
// uintptr is an integer type that is large enough to hold the bit pattern of
// any pointer.
func (wg *WaitGroup) state() (statep *uint64, semap *uint32) {
// unsafe.Pointer 可以包含任何变量的地址。
if uintptr(unsafe.Pointer(&wg.state1))%8 == 0 {
//这里涉及到了对齐。
return (*uint64)(unsafe.Pointer(&wg.state1)), &wg.state1[2]
} else {
return (*uint64)(unsafe.Pointer(&wg.state1[1])), &wg.state1[0]
}
}
// Add adds delta, which may be negative, to the WaitGroup counter.
// If the counter becomes zero, all goroutines blocked on Wait are released.
// If the counter goes negative, Add panics.
//
// Note that calls with a positive delta that occur when the counter is zero
// must happen before a Wait. Calls with a negative delta, or calls with a
// positive delta that start when the counter is greater than zero, may happen
// at any time.
// Typically this means the calls to Add should execute before the statement
// creating the goroutine or other event to be waited for.
// If a WaitGroup is reused to wait for several independent sets of events,
// new Add calls must happen after all previous Wait calls have returned.
// See the WaitGroup example.
// 增加 delata
func (wg *WaitGroup) Add(delta int) {
statep, semap := wg.state()
if race.Enabled {
_ = *statep // trigger nil deref early
if delta < 0 {
// Synchronize decrements with Wait.
race.ReleaseMerge(unsafe.Pointer(wg))
}
race.Disable()
defer race.Enable()
}
// delta左移动32(低32位是sema)
state := atomic.AddUint64(statep, uint64(delta)<<32)
v := int32(state >> 32)//去掉低位就是值
w := uint32(state)// 去掉高位就是，就是等待的数量
if race.Enabled && delta > 0 && v == int32(delta) {
// The first increment must be synchronized with Wait.
// Need to model this as a read, because there can be
// several concurrent wg.counter transitions from 0.
race.Read(unsafe.Pointer(semap))
}
if v < 0 {//只有 add() done（） 会改变v ,v最小值是0
panic("sync: negative WaitGroup counter")
}
if w != 0 && delta > 0 && v == int32(delta) {
panic("sync: WaitGroup misuse: Add called concurrently with Wait")
}
// add 成功后，并且wait是0 直接返回
if v > 0 || w == 0 {
return
}
//运行到这里，说明v==0,并且w>0
// 到这一步，只能调用done(waitgroup add -1)
// This goroutine has set counter to 0 when waiters > 0.
// Now there can't be concurrent mutations of state:
// - Adds must not happen concurrently with Wait,
// - Wait does not increment waiters if it sees counter == 0.
// Still do a cheap sanity check to detect WaitGroup misuse.
if *statep != state {
panic("sync: WaitGroup misuse: Add called concurrently with Wait")
}
// Reset waiters count to 0.
*statep = 0
for ; w != 0; w-- {
// 信号量释放
runtime_Semrelease(semap, false)
}
}
// Done decrements the WaitGroup counter by one.
func (wg *WaitGroup) Done() {
wg.Add(-1)
}
// Wait blocks until the WaitGroup counter is zero.
func (wg *WaitGroup) Wait() {
statep, semap := wg.state()
if race.Enabled {
_ = *statep // trigger nil deref early
race.Disable()
}
for {
state := atomic.LoadUint64(statep)
// 获取v,w
v := int32(state >> 32)
w := uint32(state)
if v == 0 {
// Counter is 0, no need to wait.
if race.Enabled {
race.Enable()
race.Acquire(unsafe.Pointer(wg))
}
return
}
// Increment waiters count.
// 通过 cas 将state +1 ,也就是wait +1
// 也就是wait可以调用多个
if atomic.CompareAndSwapUint64(statep, state, state+1) {
if race.Enabled && w == 0 {
// Wait must be synchronized with the first Add.
// Need to model this is as a write to race with the read in Add.
// As a consequence, can do the write only for the first waiter,
// otherwise concurrent Waits will race with each other.
race.Write(unsafe.Pointer(semap))
}
//获取信号，相当于订阅
runtime_Semacquire(semap)
if *statep != 0 {
panic("sync: WaitGroup is reused before previous Wait has returned")
}
if race.Enabled {
race.Enable()
race.Acquire(unsafe.Pointer(wg))
}
return
}
}
}
