深入Go并发原语：sync包与channel机制源码解析与实践

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

// sync.Map 的新实现基于 HashTrieMap
type Map struct {
    _ noCopy
    m isync.HashTrieMap[any, any]
}

// Load 返回存储在映射中的键值，如果不存在则返回nil
func (m *Map) Load(key any) (value any, ok bool) {
    return m.m.Load(key)
}

// Store 设置键的值
func (m *Map) Store(key, value any) {
    m.m.Store(key, value)
}

// LoadOrStore 返回键的现有值（如果存在）
// 否则，存储并返回给定值
// loaded结果为true表示值已加载，false表示已存储
func (m *Map) LoadOrStore(key, value any) (actual any, loaded bool) {
    return m.m.LoadOrStore(key, value)
}

// LoadAndDelete 删除键的值，返回之前的值（如果有）
// loaded结果报告键是否存在
func (m *Map) LoadAndDelete(key any) (value any, loaded bool) {
    return m.m.LoadAndDelete(key)
}

// Delete 删除键的值
// 如果键不在映射中，Delete什么也不做
func (m *Map) Delete(key any) {
    m.m.Delete(key)
}

// CompareAndSwap 如果映射中存储的值等于old，则交换old和new值
// old值必须是可比较类型
func (m *Map) CompareAndSwap(key, old, new any) (swapped bool) {
    return m.m.CompareAndSwap(key, old, new)
}

// Range 对映射中存在的每个键值对顺序调用f
// 如果f返回false，Range停止迭代
func (m *Map) Range(f func(key, value any) bool) {
    m.m.Range(f)
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200

// Pool 对象池结构体，用于存储和复用临时对象，减少内存分配开销
type Pool struct {
    noCopy noCopy  // 防止Pool被复制的标记，确保Pool只能通过指针传递
    
    // ==================== 本地池管理 ====================
    local     unsafe.Pointer // 指向Per-P池数组的指针，每个P都有自己的本地池
                            // 类型为[]poolLocal，但使用unsafe.Pointer避免GC扫描
    localSize uintptr        // 本地池数组的大小，等于GOMAXPROCS的值
                            // 用于边界检查和数组访问
    
    // ==================== 受害者缓存机制 ====================  
    victim     unsafe.Pointer // 指向上一轮GC前的本地池数组
                             // 当本地池为空时，会尝试从victim中获取对象
                             // 这是一种两代回收机制，提高对象复用率
    victimSize uintptr        // 受害者池数组的大小
                             // 可能与当前localSize不同，因为GOMAXPROCS可能发生变化
    
    // ==================== 对象创建函数 ====================
    // New可选地指定一个函数来生成新对象
    // 当Get方法无法从池中获取对象时调用此函数创建新对象
    // 如果New为nil，Get方法将返回nil
    New func() any
}

// poolLocalInternal Per-P池的内部结构，包含实际的对象存储
type poolLocalInternal struct {
    // ==================== 私有对象槽 ====================
    private any       // 私有对象槽，只能被对应的P访问
                     // 这是最快的获取路径，无需任何同步操作
                     // 当Put时如果private为空，直接存储到这里
                     // 当Get时优先从这里获取
    
    // ==================== 共享对象队列 ====================                 
    shared  poolChain // 共享对象队列，支持多P并发访问
                     // 本地P可以pushHead/popHead（LIFO操作，保证局部性）
                     // 其他P只能popTail（工作窃取，避免竞争）
                     // 使用无锁链式队列实现，支持动态扩容
}

// poolLocal Per-P池的完整结构，包含缓存行对齐
type poolLocal struct {
    poolLocalInternal  // 嵌入内部结构
    
    // ==================== 缓存行对齐填充 ====================
    // 防止在广泛使用的平台上出现false sharing（伪共享）
    // 128字节对齐确保每个poolLocal占用独立的缓存行
    // 128 mod (缓存行大小) = 0，适配主流CPU的缓存行大小（64/128字节）
    pad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte
}

// Put 将对象x添加到池中，供后续Get操作复用
// 这是对象池的核心存储操作，优先使用私有槽，其次使用共享队列
func (p *Pool) Put(x any) {
    // 空对象检查：nil对象没有复用价值，直接返回
    if x == nil {
        return
    }
    
    // ==================== 获取本地池 ====================
    // pin()操作：
    // 1. 禁用抢占，确保当前goroutine不会被调度到其他P
    // 2. 获取当前P对应的poolLocal
    // 3. 返回poolLocal指针和P的ID
    l, _ := p.pin()
    
    // ==================== 优先使用私有槽 ====================
    if l.private == nil {
        // 私有槽为空，直接存储对象
        // 这是最快的路径，无需任何同步操作
        // 下次同一个P上的Get操作可以直接获取
        l.private = x
    } else {
        // ==================== 使用共享队列 ====================
        // 私有槽已被占用，将对象推入共享队列的头部
        // pushHead操作：
        // 1. 使用LIFO策略，保证时间局部性
        // 2. 新对象更可能被快速复用
        // 3. 支持无锁并发操作
        l.shared.pushHead(x)
    }
    
    // ==================== 恢复抢占 ====================
    // 重新启用抢占，允许goroutine被调度
    // 必须与pin()配对使用，确保临界区的完整性
    runtime_procUnpin()
}

// Get 从池中获取一个对象，将其从池中移除并返回给调用者
// 获取顺序：私有槽 -> 本地共享队列 -> 工作窃取 -> 受害者缓存 -> 创建新对象
// Get可能选择忽略池并将其视为空，这是一种性能优化策略
func (p *Pool) Get() any {
    // ==================== 获取本地池 ====================
    // pin()禁用抢占并获取当前P的poolLocal和P的ID
    l, pid := p.pin()
    
    // ==================== 第一优先级：私有槽 ====================
    // 尝试从私有槽获取对象，这是最快的路径
    x := l.private      // 读取私有槽中的对象
    l.private = nil     // 立即清空私有槽，为下次Put做准备
    
    if x == nil {
        // ==================== 第二优先级：本地共享队列 ====================
        // 私有槽为空，尝试从本地共享队列的头部弹出对象
        // 我们更喜欢头部而不是尾部，以实现重用的时间局部性
        // popHead操作：
        // 1. LIFO策略，最近Put的对象最先被Get
        // 2. 保证缓存友好性，提高性能
        x, _ = l.shared.popHead()
        
        if x == nil {
            // ==================== 第三优先级：慢路径 ====================
            // 本地队列也为空，进入慢路径
            // getSlow会尝试工作窃取和受害者缓存
            x = p.getSlow(pid)
        }
    }
    
    // ==================== 恢复抢占 ====================
    // 重新启用抢占，必须与pin()配对
    runtime_procUnpin()
    
    // ==================== 最后手段：创建新对象 ====================
    if x == nil && p.New != nil {
        // 所有获取途径都失败，调用New函数创建新对象
        // 这确保了Get操作总是能返回可用的对象（如果New不为nil）
        x = p.New()
    }
    
    return x
}

// getSlow 实现工作窃取和受害者缓存访问的慢路径
// 当本地池（私有槽和共享队列）都为空时调用此方法
func (p *Pool) getSlow(pid int) any {
    // ==================== 工作窃取阶段 ====================
    // 获取当前池的大小和本地池数组
    size := runtime_LoadAcquintptr(&p.localSize)  // 原子加载，确保读取到最新的大小
    locals := p.local                             // 获取本地池数组指针
    
    // 尝试从其他P的本地池中窃取对象
    // 窃取策略：从下一个P开始，循环遍历所有其他P的共享队列
    for i := 0; i < int(size); i++ {
        // 计算目标P的索引：(pid+i+1) % size
        // +1确保不会窃取自己的队列，i确保遍历所有其他P
        l := indexLocal(locals, (pid+i+1)%int(size))
        
        // 从目标P的共享队列尾部窃取对象
        // 使用popTail而不是popHead，实现真正的工作窃取：
        // - 本地Get使用popHead（LIFO）
        // - 远程窃取使用popTail（FIFO）
        // 这样可以减少竞争，提高并发性能
        if x, _ := l.shared.popTail(); x != nil {
            return x  // 窃取成功，返回对象
        }
    }
    
    // ==================== 受害者缓存阶段 ====================
    // 工作窃取失败，尝试从受害者缓存中获取对象
    // 受害者缓存是上一轮GC周期中保存的池内容
    
    // 获取受害者缓存的大小
    size = atomic.LoadUintptr(&p.victimSize)  // 原子加载受害者缓存大小
    if uintptr(pid) >= size {
        // 当前P的ID超出受害者缓存范围，直接返回nil
        // 这可能发生在GOMAXPROCS动态调整的情况下
        return nil
    }
    
    // 获取受害者缓存数组
    locals = p.victim
    
    // ==================== 受害者私有槽 ====================
    // 首先尝试从当前P对应的受害者私有槽获取
    l := indexLocal(locals, pid)
    if x := l.private; x != nil {
        l.private = nil  // 清空私有槽
        return x         // 返回找到的对象
    }
    
    // ==================== 受害者共享队列窃取 ====================
    // 私有槽为空，尝试从所有受害者共享队列中窃取
    for i := 0; i < int(size); i++ {
        // 从当前P开始，循环遍历所有受害者池的共享队列
        l := indexLocal(locals, (pid+i)%int(size))
        
        // 从受害者共享队列尾部窃取对象
        if x, _ := l.shared.popTail(); x != nil {
            return x  // 窃取成功，返回对象
        }
    }

    // ==================== 清理受害者缓存 ====================
    // 受害者缓存已经完全耗尽，标记为空
    // 这样可以避免后续无意义的受害者缓存访问
    atomic.StoreUintptr(&p.victimSize, 0)
    
    // 所有获取途径都失败，返回nil
    // 调用者会检查这个结果，并可能调用New函数创建新对象
    return nil
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225

// WaitGroup 等待一组goroutine完成的同步原语
// 使用场景：主goroutine需要等待多个子goroutine完成后再继续执行
// 
// 典型用法：
// 1. 主goroutine调用Add(n)设置要等待的goroutine数量
// 2. 启动n个goroutine，每个goroutine在完成时调用Done()
// 3. 主goroutine调用Wait()阻塞，直到所有goroutine完成
//
// 注意：Add()调用必须在Wait()之前，通常在启动goroutine之前调用
type WaitGroup struct {
    noCopy noCopy  // 防止WaitGroup被复制，确保语义正确性
    
    // ==================== 核心状态字段 ====================
    // state是一个64位原子值，使用位字段布局存储多个信息：
    // 
    // 位字段布局 (从低位到高位):
    //   bits[0:32]   counter 计数器 - 记录还有多少个goroutine未完成
    //   bits[32]     flag: synctest bubble membership - 测试标志位
    //   bits[33:64]  wait count 等待计数 - 记录有多少个goroutine在Wait()中阻塞
    //
    // 使用64位原子操作的原因：
    // 1. 需要同时原子地操作计数器和等待者计数
    // 2. 避免竞态条件，确保Add/Done/Wait操作的一致性
    // 3. 单次原子操作比多次原子操作更高效
    //
    // 内存对齐考虑：
    // 64位原子操作需要64位对齐，Go编译器会自动处理对齐问题
    state atomic.Uint64  // 组合的计数器、标志位和等待者状态
    
    // ==================== 信号量字段 ====================
    // sema是用于阻塞和唤醒等待goroutine的信号量
    // 当计数器降为0时，会通过信号量唤醒所有等待的goroutine
    // 使用runtime的信号量实现，比channel更轻量级且性能更好
    sema  uint32         // 信号量，用于Wait()的阻塞和唤醒机制
}

// Add 将delta（可能为负）添加到WaitGroup任务计数器
// 这是WaitGroup的核心操作，用于增加或减少待完成的任务数量
// 
// 参数说明：
// - delta > 0: 增加待完成任务数（通常在启动goroutine前调用）
// - delta < 0: 减少待完成任务数（Done()内部调用Add(-1)）
// - delta = 0: 无操作，直接返回
//
// 重要规则：
// 1. 如果计数器变为零，所有在Wait()上阻塞的goroutine都会被释放
// 2. 如果计数器变为负数，Add会panic
// 3. Add()必须在Wait()之前调用，否则会panic
func (wg *WaitGroup) Add(delta int) {
    // ==================== 原子更新状态 ====================
    // 将delta左移32位，因为计数器位于state的高32位
    // 原子地将delta添加到计数器部分
    state := wg.state.Add(uint64(delta) << 32)
    
    // 解析state的各个字段
    v := int32(state >> 32)              // 提取计数器值（高32位）
    w := uint32(state & 0x7fffffff)      // 提取等待者数量（低31位，排除flag位）
    
    // ==================== 错误检查 ====================
    if v < 0 {
        // 计数器不能为负，这通常意味着Done()调用次数超过了Add()
        panic("sync: negative WaitGroup counter")
    }
    
    // 检查并发使用错误：Add()和Wait()不能并发调用
    // 条件解释：
    // - w != 0: 有goroutine在等待
    // - delta > 0: 这是一个增加计数的Add调用
    // - v == int32(delta): 计数器之前为0，现在等于delta
    // 这种情况表明在有goroutine等待时又调用了Add()，这是错误的使用方式
    if w != 0 && delta > 0 && v == int32(delta) {
        panic("sync: WaitGroup misuse: Add called concurrently with Wait")
    }
    
    // ==================== 常规情况：无需唤醒 ====================
    if v > 0 || w == 0 {
        // 情况1：计数器仍大于0，还有任务未完成，无需唤醒等待者
        // 情况2：计数器为0但没有等待者，无需唤醒
        return
    }
    
    // ==================== 唤醒等待者 ====================
    // 执行到这里说明：v == 0 && w > 0
    // 即计数器为0且有等待者，需要唤醒所有等待的goroutine
    
    // 此时状态不能再改变，因为：
    // - Add不能与Wait并发（上面已检查）
    // - Wait不能增加等待者计数，因为它们会看到counter==0
    // 
    // 额外的安全检查：确保状态没有被其他goroutine修改
    // 这是一个防御性检查，正常情况下不应该失败
    if wg.state.Load() != state {
        panic("sync: WaitGroup misuse: Add called concurrently with Wait")
    }
    
    // 重置state为0，清除所有状态
    // 这样可以让WaitGroup重新使用（虽然不推荐）
    wg.state.Store(0)
    
    // 逐个释放所有等待的goroutine
    // 使用信号量机制唤醒每个在Wait()中阻塞的goroutine
    for ; w != 0; w-- {
        // 参数说明：
        // - &wg.sema: 信号量地址
        // - false: 不是handoff模式
        // - 1: 跳过等待队列中的1个waiter（即释放1个等待者）
        runtime_Semrelease(&wg.sema, false, 1)
    }
}

// Done 减少WaitGroup计数器1，标记一个任务完成
// 这是Add(-1)的便捷方法，通常在goroutine完成工作时调用
// 
// 使用模式：
//   defer wg.Done()  // 在goroutine开始时设置，确保无论如何都会调用
//
// 注意事项：
// 1. Done()调用次数不能超过Add()增加的总数，否则会panic
// 2. 建议使用defer确保Done()一定会被调用，避免死锁
// 3. Done()内部调用Add(-1)，会触发相同的唤醒逻辑
func (wg *WaitGroup) Done() {
    // 直接调用Add(-1)来减少计数器
    // 这样可以复用Add()中的所有逻辑，包括：
    // - 原子操作保证并发安全
    // - 错误检查（防止计数器变负）
    // - 唤醒等待者（当计数器变为0时）
    wg.Add(-1)
}

// Wait 阻塞直到WaitGroup计数器为零
// 这个方法会一直阻塞，直到所有通过Add()添加的任务都调用了Done()
//
// 使用场景：
// - 主goroutine等待所有工作goroutine完成
// - 确保所有并发任务完成后再进行下一步操作
//
// 重要注意事项：
// 1. Wait()必须在所有Add()调用之后调用
// 2. 不要在Add()和Wait()之间有竞态条件
// 3. WaitGroup可以重用，但必须等待上一轮Wait()完全返回
func (wg *WaitGroup) Wait() {
    // ==================== 循环等待逻辑 ====================
    // 使用循环是因为CompareAndSwap可能失败，需要重试
    for {
        // 原子加载当前状态
        state := wg.state.Load()
        
        // 解析状态字段
        v := int32(state >> 32)              // 提取计数器值（高32位）
        w := uint32(state & 0x7fffffff)      // 提取等待者数量（低31位，排除flag位）
        
        // ==================== 快速路径：无需等待 ====================
        if v == 0 {
            // 计数器为0，说明所有任务都已完成，无需等待
            return
        }
        
        // ==================== 慢路径：需要阻塞等待 ====================
        // 尝试原子地增加等待者计数（+1）
        // 使用CompareAndSwap确保在增加等待者计数时状态没有被其他goroutine修改
        if wg.state.CompareAndSwap(state, state+1) {
            // ==================== 成功增加等待者计数 ====================
            // 现在当前goroutine已经注册为等待者，可以安全地阻塞
            
            // 在信号量上阻塞等待，直到被Add()中的Semrelease唤醒
            // 参数说明：
            // - &wg.sema: 信号量地址
            // - false: 不是lifo模式，按FIFO顺序唤醒
            runtime_SemacquireWaitGroup(&wg.sema, false)
            
            // ==================== 被唤醒后的检查 ====================
            // 正常情况下，被唤醒时state应该为0（所有任务完成）
            // 如果state不为0，说明WaitGroup在Wait()返回前就被重用了
            // 这是一个使用错误，需要panic
            if wg.state.Load() != 0 {
                panic("sync: WaitGroup is reused before previous Wait has returned")
            }
            
            // 等待完成，返回
            return
        }
        
        // ==================== CompareAndSwap失败，重试 ====================
        // 如果CompareAndSwap失败，说明其他goroutine修改了state
        // 继续循环，重新读取state并尝试
        // 这种情况可能发生在：
        // 1. 其他goroutine调用了Add()或Done()
        // 2. 其他goroutine也在调用Wait()
    }
}

// Go 在新goroutine中调用f并将该任务添加到WaitGroup
// 这是一个便捷方法，自动处理Add()和Done()的调用
// 当f返回时，任务从WaitGroup中移除
//
// 使用场景：
// - 简化并发任务的启动和管理
// - 避免手动调用Add()和Done()的繁琐操作
// - 确保Done()一定会被调用（通过defer）
//
// 等价于手动操作：
//   wg.Add(1)
//   go func() {
//       defer wg.Done()
//       f()
//   }()
func (wg *WaitGroup) Go(f func()) {
    // ==================== 增加任务计数 ====================
    // 在启动goroutine之前先增加计数器
    // 这样可以确保Wait()不会在goroutine启动前就返回
    wg.Add(1)
    
    // ==================== 启动goroutine ====================
    go func() {
        // 使用defer确保Done()一定会被调用
        // 无论f()是正常返回还是panic，Done()都会执行
        // 这避免了因为panic导致的死锁问题
        defer wg.Done()
        
        // 执行用户提供的函数
        // 如果f()发生panic，defer会确保Done()仍然被调用
        // panic会继续向上传播，不会被这里捕获
        f()
    }()
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203

// Cond 条件变量，用于在共享资源状态改变时通知等待的goroutine
// 条件变量必须与互斥锁配合使用，实现"等待某个条件成立"的同步模式
//
// 典型使用场景：
// - 生产者-消费者模式：消费者等待队列非空
// - 资源池：等待资源可用
// - 状态同步：等待某个状态变化
//
// 重要原则：
// 1. 条件检查必须在循环中进行（防止虚假唤醒）
// 2. Wait()调用前必须持有锁
// 3. Signal/Broadcast调用时建议持有锁（虽然不强制）
type Cond struct {
    noCopy noCopy  // 防止Cond被复制，确保语义正确性
    
    // ==================== 互斥锁字段 ====================
    // L是与条件变量关联的锁，用于保护共享状态
    // 在观察或更改条件时必须持有此锁
    // 通常是*sync.Mutex或*sync.RWMutex
    L Locker       // 关联的锁，保护条件变量相关的共享状态
    
    // ==================== 通知机制字段 ====================
    // notify是等待队列的实现，管理所有等待此条件的goroutine
    // 使用runtime的notifyList实现高效的等待和唤醒
    notify  notifyList  // 等待队列，存储所有在此条件上等待的goroutine
    
    // ==================== 复制检查字段 ====================
    // checker用于检测Cond是否被非法复制
    // 复制Cond会导致不可预期的行为，因为等待队列会被复制
    checker copyChecker // 复制检测器，防止Cond被意外复制
}

// NewCond 返回一个带有Locker l的新Cond
// 创建条件变量时必须提供一个锁，这个锁将用于保护条件相关的共享状态
//
// 参数说明：
// - l: 实现了Locker接口的锁（通常是*sync.Mutex或*sync.RWMutex）
//
// 返回值：
// - 新创建的Cond实例指针
//
// 使用示例：
//   var mu sync.Mutex
//   cond := sync.NewCond(&mu)
func NewCond(l Locker) *Cond {
    // 创建新的Cond实例，只需要设置关联的锁
    // notify和checker字段会使用零值初始化
    // - notify: 空的等待队列
    // - checker: 初始状态的复制检测器
    return &Cond{L: l}
}

// Wait 原子地解锁c.L并暂停调用goroutine的执行
// 这是条件变量的核心操作，实现"等待条件成立"的语义
//
// 操作流程：
// 1. 原子地释放锁并加入等待队列
// 2. 阻塞当前goroutine，直到被Signal或Broadcast唤醒
// 3. 被唤醒后重新获取锁
// 4. 返回给调用者（此时持有锁）
//
// 重要特性：
// - 与其他系统不同，Wait不能返回，除非被Broadcast或Signal唤醒
// - Wait返回时不保证条件为真（可能存在虚假唤醒）
// - 调用Wait前必须持有c.L锁
//
// 标准使用模式：
//   c.L.Lock()
//   for !condition() {  // 必须使用循环检查条件
//       c.Wait()        // 等待条件成立
//   }
//   // ... 使用条件 ...
//   c.L.Unlock()
//
// 为什么需要循环：
// 1. 防止虚假唤醒（spurious wakeup）
// 2. 多个goroutine可能同时被唤醒，但只有一个能满足条件
// 3. 条件可能在唤醒后被其他goroutine改变
func (c *Cond) Wait() {
    // ==================== 复制检查 ====================
    // 检查Cond是否被非法复制
    // 如果检测到复制，会panic
    c.checker.check()
    
    // ==================== 加入等待队列 ====================
    // 将当前goroutine添加到等待队列，并返回一个ticket
    // ticket用于标识这个等待者，在唤醒时使用
    // 这个操作是原子的，确保在释放锁前已经加入队列
    t := runtime_notifyListAdd(&c.notify)
    
    // ==================== 释放锁 ====================
    // 释放关联的锁，允许其他goroutine修改共享状态
    // 这是条件变量的关键：在等待时不持有锁
    c.L.Unlock()
    
    // ==================== 阻塞等待 ====================
    // 在等待队列上阻塞，直到被Signal或Broadcast唤醒
    // 参数说明：
    // - &c.notify: 等待队列
    // - t: 等待者的ticket，用于标识和唤醒
    // 这个调用会阻塞当前goroutine，直到被唤醒
    runtime_notifyListWait(&c.notify, t)
    
    // ==================== 重新获取锁 ====================
    // 被唤醒后，重新获取锁
    // 这确保了Wait返回时调用者持有锁，可以安全地检查条件
    c.L.Lock()
    
    // 此时goroutine已被唤醒并重新持有锁
    // 调用者应该重新检查条件，因为：
    // 1. 可能是虚假唤醒
    // 2. 条件可能已被其他goroutine改变
}

// Signal 唤醒一个等待c的goroutine（如果有的话）
// 这是条件变量的通知操作，用于唤醒单个等待者
//
// 使用场景：
// - 资源变为可用时，只需要唤醒一个等待者
// - 生产者-消费者模式中，生产了一个项目
// - 状态改变时，只需要一个goroutine处理
//
// 重要特性：
// - 如果没有goroutine在等待，Signal是无操作的
// - 只唤醒一个等待者，具体是哪个是不确定的
// - 调用者在调用期间持有c.L是允许的但不是必需的
// - 建议在持有锁时调用，确保状态一致性
//
// 使用模式：
//   c.L.Lock()
//   // 修改共享状态
//   condition = true
//   c.Signal()  // 通知一个等待者
//   c.L.Unlock()
func (c *Cond) Signal() {
    // ==================== 复制检查 ====================
    // 检查Cond是否被非法复制
    // 如果检测到复制，会panic
    c.checker.check()
    
    // ==================== 唤醒一个等待者 ====================
    // 从等待队列中唤醒一个goroutine
    // 如果等待队列为空，这个调用是无操作的
    // 被唤醒的goroutine会从Wait()中的runtime_notifyListWait返回
    // 然后重新获取锁并继续执行
    runtime_notifyListNotifyOne(&c.notify)
}

// Broadcast 唤醒所有等待c的goroutine
// 这是条件变量的广播操作，用于唤醒所有等待者
//
// 使用场景：
// - 全局状态改变，所有等待者都需要重新检查条件
// - 程序关闭时，唤醒所有等待的goroutine
// - 配置更新时，所有相关的goroutine都需要知道
// - 资源池扩容时，多个等待者可能都能获得资源
//
// 重要特性：
// - 唤醒所有在等待队列中的goroutine
// - 如果没有goroutine在等待，Broadcast是无操作的
// - 所有被唤醒的goroutine都会竞争锁
// - 调用者在调用期间持有c.L是允许的但不是必需的
// - 建议在持有锁时调用，确保状态一致性
//
// 使用模式：
//   c.L.Lock()
//   // 修改全局状态
//   globalCondition = true
//   c.Broadcast()  // 通知所有等待者
//   c.L.Unlock()
//
// 注意事项：
// - Broadcast后，所有等待者会竞争锁，只有一个能立即执行
// - 其他goroutine需要等待锁释放后才能继续
// - 每个被唤醒的goroutine都应该重新检查条件
func (c *Cond) Broadcast() {
    // ==================== 复制检查 ====================
    // 检查Cond是否被非法复制
    // 如果检测到复制，会panic
    c.checker.check()
    
    // ==================== 唤醒所有等待者 ====================
    // 从等待队列中唤醒所有goroutine
    // 如果等待队列为空，这个调用是无操作的
    // 所有被唤醒的goroutine会从Wait()中的runtime_notifyListWait返回
    // 它们会竞争重新获取锁，然后继续执行
    runtime_notifyListNotifyAll(&c.notify)
}

// copyChecker 持有指向自身的后向指针以检测对象复制
type copyChecker uintptr

func (c *copyChecker) check() {
    // 分三步检查c是否已被复制：
    // 1. 第一次比较是快速路径。如果c已初始化且未复制，这将立即返回
    // 2. 确保c已初始化。如果CAS成功，我们就完成了
    // 3. 再次执行步骤1。现在c肯定已初始化，如果失败，c被复制了
    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")
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

// 生产者-消费者模式
type Buffer struct {
    mu   sync.Mutex
    cond *sync.Cond
    data []int
    size int
}

func NewBuffer(size int) *Buffer {
    b := &Buffer{size: size}
    b.cond = sync.NewCond(&b.mu)
    return b
}

// 生产者
func (b *Buffer) Put(item int) {
    b.mu.Lock()
    defer b.mu.Unlock()
    
    // 等待缓冲区有空间
    for len(b.data) >= b.size {
        b.cond.Wait()
    }
    
    b.data = append(b.data, item)
    b.cond.Broadcast() // 通知消费者
}

// 消费者
func (b *Buffer) Get() int {
    b.mu.Lock()
    defer b.mu.Unlock()
    
    // 等待缓冲区有数据
    for len(b.data) == 0 {
        b.cond.Wait()
    }
    
    item := b.data[0]
    b.data = b.data[1:]
    b.cond.Broadcast() // 通知生产者
    return item
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

// hchan 是channel的运行时表示，包含了channel的所有状态信息
// 这是Go语言channel机制的核心数据结构
type hchan struct {
    // ==================== 缓冲区管理字段 ====================
    qcount   uint           // 当前缓冲区中的元素数量
                           // 对于无缓冲channel，始终为0
                           // 对于有缓冲channel，表示buf中实际存储的元素个数
    
    dataqsiz uint           // 缓冲区容量，即make(chan T, N)中的N
                           // 0表示无缓冲channel
                           // >0表示有缓冲channel的最大容量
    
    buf      unsafe.Pointer // 指向缓冲区数组的指针
                           // 缓冲区是一个循环队列，存储dataqsiz个元素
                           // 对于无缓冲channel，buf为nil
    
    elemsize uint16         // 单个元素的字节大小
                           // 用于计算缓冲区偏移和内存分配
                           // 例如：chan int的elemsize为8（64位系统）
    
    // ==================== 状态管理字段 ====================
    closed   uint32         // channel关闭状态标志
                           // 0: 未关闭，1: 已关闭
                           // 使用原子操作访问，确保并发安全
    
    timer    *timer         // 关联的定时器（用于select with timeout等场景）
                           // 大多数情况下为nil，只在特定场景下使用
    
    elemtype *_type         // 元素类型信息，用于GC和反射
                           // 包含类型的大小、对齐、GC信息等
    
    // ==================== 循环队列索引 ====================
    sendx    uint           // 下一个发送元素在buf中的索引位置
                           // 发送时使用，发送后递增（模dataqsiz）
                           // 与recvx配合实现循环队列
    
    recvx    uint           // 下一个接收元素在buf中的索引位置
                           // 接收时使用，接收后递增（模dataqsiz）
                           // 当sendx == recvx时，缓冲区为空或满
    
    // ==================== 等待队列 ====================
    recvq    waitq          // 等待接收的goroutine队列
                           // 当channel为空时，接收操作会阻塞在此队列
                           // 按FIFO顺序排列等待的goroutine
    
    sendq    waitq          // 等待发送的goroutine队列
                           // 当channel满时，发送操作会阻塞在此队列
                           // 按FIFO顺序排列等待的goroutine
    
    // ==================== 互斥锁 ====================
    // lock保护hchan中的所有字段，以及在此通道上阻塞的sudog中的几个字段
    // 
    // 重要约束：
    // - 持有此锁时不要更改另一个G的状态（特别是不要ready一个G）
    // - 这可能与栈收缩产生死锁
    // - 所有channel操作都必须在此锁保护下进行
    lock mutex              // 保护channel所有字段的互斥锁
}

// waitq 等待队列，用于管理阻塞在channel上的goroutine
// 这是一个双向链表结构，按FIFO顺序管理等待者
type waitq struct {
    first *sudog  // 队列头部，指向第一个等待的sudog
                  // 新的等待者会被添加到队列尾部
                  // 唤醒时从头部开始
    
    last  *sudog  // 队列尾部，指向最后一个等待的sudog
                  // 用于快速在队列尾部添加新的等待者
                  // 当队列为空时，first和last都为nil
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106

// chansend 在通道c上发送数据
// ep指向要发送的数据
// block表示是否阻塞
// callerpc用于竞态检测
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
    if c == nil {
        if !block {
            return false
        }
        // 向nil通道发送会永远阻塞
        gopark(nil, nil, waitReasonChanSendNilChan, traceBlockForever, 2)
        throw("unreachable")
    }
    
    // 快速路径：检查非阻塞操作是否失败，无需获取锁
    if !block && c.closed == 0 && full(c) {
        return false
    }
    
    lock(&c.lock)
    
    if c.closed != 0 {
        unlock(&c.lock)
        panic(plainError("send on closed channel"))
    }
    
    // 情况1：找到等待的接收者，直接发送
    if sg := c.recvq.dequeue(); sg != nil {
        send(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true
    }
    
    // 情况2：缓冲区有空间，入队
    if c.qcount < c.dataqsiz {
        qp := chanbuf(c, c.sendx)
        typedmemmove(c.elemtype, qp, ep)
        c.sendx++
        if c.sendx == c.dataqsiz {
            c.sendx = 0
        }
        c.qcount++
        unlock(&c.lock)
        return true
    }
    
    if !block {
        unlock(&c.lock)
        return false
    }
    
    // 情况3：阻塞在通道上，等待接收者完成操作
    gp := getg()
    mysg := acquireSudog()
    mysg.releasetime = 0
    mysg.elem = ep
    mysg.waitlink = nil
    mysg.g = gp
    mysg.isSelect = false
    mysg.c = c
    gp.waiting = mysg
    gp.param = nil
    c.sendq.enqueue(mysg)
    
    // 标记即将在通道上park
    gp.parkingOnChan.Store(true)
    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceBlockChanSend, 2)
    
    // 被唤醒
    KeepAlive(ep)
    
    if mysg != gp.waiting {
        throw("G waiting list is corrupted")
    }
    gp.waiting = nil
    gp.activeStackChans = false
    closed := !mysg.success
    gp.param = nil
    mysg.c = nil
    releaseSudog(mysg)
    
    if closed {
        if c.closed == 0 {
            throw("chansend: spurious wakeup")
        }
        panic(plainError("send on closed channel"))
    }
    return true
}

// send 处理空通道c上的发送操作
// 发送者发送的值ep被复制到接收者sg
// 然后唤醒接收者继续执行
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
    if sg.elem != nil {
        sendDirect(c.elemtype, sg, ep)
        sg.elem = nil
    }
    gp := sg.g
    unlockf()
    gp.param = unsafe.Pointer(sg)
    sg.success = true
    if sg.releasetime != 0 {
        sg.releasetime = cputicks()
    }
    goready(gp, skip+1)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

// chanrecv 从通道c接收数据并将其写入ep
// ep可能为nil，在这种情况下接收的值被忽略
// 如果block == false且没有元素可用，返回(false, false)
// 否则，如果c关闭，将*ep清零并返回(true, false)
// 否则，用元素填充*ep并返回(true, true)
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
    if c == nil {
        if !block {
            return
        }
        // 从nil通道接收会永远阻塞
        gopark(nil, nil, waitReasonChanReceiveNilChan, traceBlockForever, 2)
        throw("unreachable")
    }
    
    // 快速路径：检查非阻塞操作是否失败，无需获取锁
    if !block && empty(c) {
        if atomic.Load(&c.closed) == 0 {
            return
        }
        if empty(c) {
            if ep != nil {
                typedmemclr(c.elemtype, ep)
            }
            return true, false
        }
    }
    
    lock(&c.lock)
    
    if c.closed != 0 {
        if c.qcount == 0 {
            unlock(&c.lock)
            if ep != nil {
                typedmemclr(c.elemtype, ep)
            }
            return true, false
        }
        // 通道已关闭，但缓冲区有数据
    } else {
        // 找到等待的发送者且通道未关闭
        if sg := c.sendq.dequeue(); sg != nil {
            recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
            return true, true
        }
    }
    
    // 情况1：缓冲区有数据，直接接收
    if c.qcount > 0 {
        qp := chanbuf(c, c.recvx)
        if ep != nil {
            typedmemmove(c.elemtype, ep, qp)
        }
        typedmemclr(c.elemtype, qp)
        c.recvx++
        if c.recvx == c.dataqsiz {
            c.recvx = 0
        }
        c.qcount--
        unlock(&c.lock)
        return true, true
    }
    
    if !block {
        unlock(&c.lock)
        return false, false
    }
    
    // 情况2：没有发送者可用，阻塞在此通道上
    gp := getg()
    mysg := acquireSudog()
    mysg.releasetime = 0
    mysg.elem = ep
    mysg.waitlink = nil
    gp.waiting = mysg
    mysg.g = gp
    mysg.isSelect = false
    mysg.c = c
    gp.param = nil
    c.recvq.enqueue(mysg)
    
    // 标记即将在通道上park
    gp.parkingOnChan.Store(true)
    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceBlockChanRecv, 2)
    
    // 被唤醒
    if mysg != gp.waiting {
        throw("G waiting list is corrupted")
    }
    gp.waiting = nil
    gp.activeStackChans = false
    success := mysg.success
    gp.param = nil
    mysg.c = nil
    releaseSudog(mysg)
    return true, success
}