Go并发编程

1atomic.CompareAndSwapInt32(&c,10,100)
2// 和下面是一个意思
3if c==10 {
4  c=100
5}

 1type semaRoot struct {
 2	lock  mutex
 3	treap *sudog // root of balanced tree of unique waiters. 等待者树
 4	nwait uint32 // Number of waiters. Read w/o the lock. 等待者数
 5}
 6
 7type sudog struct {
 8	// The following fields are protected by the hchan.lock of the
 9	// channel this sudog is blocking on. shrinkstack depends on
10	// this for sudogs involved in channel ops.
11
12	g *g
13
14	next *sudog
15	prev *sudog
16	elem unsafe.Pointer // data element (may point to stack)
17
18	// The following fields are never accessed concurrently.
19	// For channels, waitlink is only accessed by g.
20	// For semaphores, all fields (including the ones above)
21	// are only accessed when holding a semaRoot lock.
22
23	acquiretime int64
24	releasetime int64
25	ticket      uint32
26
27	// isSelect indicates g is participating in a select, so
28	// g.selectDone must be CAS'd to win the wake-up race.
29	isSelect bool
30
31	// success indicates whether communication over channel c
32	// succeeded. It is true if the goroutine was awoken because a
33	// value was delivered over channel c, and false if awoken
34	// because c was closed.
35	success bool
36
37	parent   *sudog // semaRoot binary tree
38	waitlink *sudog // g.waiting list or semaRoot
39	waittail *sudog // semaRoot
40	c        *hchan // channel
41}

 1//获得锁 -1 
 2func semacquire1(addr *uint32, lifo bool, profile semaProfileFlags, skipframes int) {}
 3
 4//=0 获得锁失败
 5func cansemacquire(addr *uint32) bool {
 6    for {
 7        v := atomic.Load(addr)
 8        if v == 0 {
 9            return false // =0 获得锁失败
10        }
11	    if atomic.Cas(addr, v, v-1) {
12            return true //获得锁成功
13        }
14    }
15}
16
17//释放锁  +1 
18func semrelease1(addr *uint32, handoff bool, skipframes int) {}

1//进入休眠
2// semacquire1
3goparkunlock(&root.lock, waitReasonSemacquire, traceEvGoBlockSync, 4+skipframes)

1// A Mutex must not be copied after first use.
2type Mutex struct {
3	state int32
4	sema  uint32 //值=0 当等待队列用
5}

1go run -race xxx.go
2go tool compile -S file.go #go语言查看汇编代码命令

1// A Mutex must not be copied after first use.
2type Mutex struct {
3	state int32
4	sema  uint32 // 默认0 
5}

1type RWMutex struct {
2	w           Mutex  // held if there are pending writers 互斥锁作为写锁,加写锁才用的
3	writerSem   uint32 // semaphore for writers to wait for completing readers 作为写协程队列
4	readerSem   uint32 // semaphore for readers to wait for completing writers 作为读协程队列
5	readerCount int32  // number of pending readers   正直：正在读的协程  负值：加了写锁  
6	readerWait  int32  // number of departing readers 写锁加锁前要等待多少个读协程的个数 
7}

 1// A WaitGroup must not be copied after first use.
 2type WaitGroup struct {
 3	noCopy noCopy // noCopy 的辅助字段，主要就是辅助 vet 工具检查是否通过 copy 赋值这个 WaitGroup 实例。
 4
 5	// 64-bit value: high 32 bits are counter, low 32 bits are waiter count.
 6	// 64-bit atomic operations require 64-bit alignment, but 32-bit
 7	// compilers do not ensure it. So we allocate 12 bytes and then use
 8	// the aligned 8 bytes in them as state, and the other 4 as storage
 9	// for the sema.
10	state1 [3]uint32
11}

 1// Wait blocks until the WaitGroup counter is zero.
 2func (wg *WaitGroup) Wait() {
 3	statep, semap := wg.state() // statep 就是 count+waiter 
 4	if race.Enabled {
 5		_ = *statep // trigger nil deref early
 6		race.Disable()
 7	}
 8	for {
 9		state := atomic.LoadUint64(statep)
10		v := int32(state >> 32)  //拿到 counter
11		w := uint32(state)       //拿到waiter 
12		
13		if v == 0 {
14			// Counter is 0, no need to wait.   被等待协程数是0 ，不需要等待
15			if race.Enabled {
16				race.Enable()
17				race.Acquire(unsafe.Pointer(wg))
18			}
19			return
20		}
21		// Increment waiters count.
22		if atomic.CompareAndSwapUint64(statep, state, state+1) {
23			if race.Enabled && w == 0 {
24				// Wait must be synchronized with the first Add.
25				// Need to model this is as a write to race with the read in Add.
26				// As a consequence, can do the write only for the first waiter,
27				// otherwise concurrent Waits will race with each other.
28				race.Write(unsafe.Pointer(semap))
29			}
30			runtime_Semacquire(semap)
31			if *statep != 0 {
32				panic("sync: WaitGroup is reused before previous Wait has returned")
33			}
34			if race.Enabled {
35				race.Enable()
36				race.Acquire(unsafe.Pointer(wg))
37			}
38			return
39		}
40	}
41}

 1package sync
 2
 3import (
 4	"sync/atomic"
 5)
 6
 7// Once is an object that will perform exactly one action.
 8type Once struct {
 9	// done indicates whether the action has been performed.
10	// It is first in the struct because it is used in the hot path.
11	// The hot path is inlined at every call site.
12	// Placing done first allows more compact instructions on some architectures (amd64/x86),
13	// and fewer instructions (to calculate offset) on other architectures.
14	done uint32
15	m    Mutex
16}
17
18// Do calls the function f if and only if Do is being called for the
19// first time for this instance of Once. In other words, given
20// 	var once Once
21// if once.Do(f) is called multiple times, only the first call will invoke f,
22// even if f has a different value in each invocation. A new instance of
23// Once is required for each function to execute.
24//
25// Do is intended for initialization that must be run exactly once. Since f
26// is niladic, it may be necessary to use a function literal to capture the
27// arguments to a function to be invoked by Do:
28// 	config.once.Do(func() { config.init(filename) })
29//
30// Because no call to Do returns until the one call to f returns, if f causes
31// Do to be called, it will deadlock.
32//
33// If f panics, Do considers it to have returned; future calls of Do return
34// without calling f.
35//
36func (o *Once) Do(f func()) {
37	// Note: Here is an incorrect implementation of Do:
38	//
39	//	if atomic.CompareAndSwapUint32(&o.done, 0, 1) {
40	//		f()
41	//	}
42	//
43	// Do guarantees that when it returns, f has finished.
44	// This implementation would not implement that guarantee:
45	// given two simultaneous calls, the winner of the cas would
46	// call f, and the second would return immediately, without
47	// waiting for the first's call to f to complete.
48	// This is why the slow path falls back to a mutex, and why
49	// the atomic.StoreUint32 must be delayed until after f returns.
50
51	if atomic.LoadUint32(&o.done) == 0 {
52		// Outlined slow-path to allow inlining of the fast-path.
53		o.doSlow(f)
54	}
55}
56
57func (o *Once) doSlow(f func()) {
58	o.m.Lock()
59	defer o.m.Unlock()
60	// double check  都进入doSlow  可能有并发问题。 o.Done 被改了 导致重复执行f() 。 
61	if o.done == 0 {
62		defer atomic.StoreUint32(&o.done, 1)
63		f()
64}

 1
 2func poolCleanup() {
 3	// This function is called with the world stopped, at the beginning of a garbage collection.
 4	// It must not allocate and probably should not call any runtime functions.
 5
 6	// Because the world is stopped, no pool user can be in a
 7	// pinned section (in effect, this has all Ps pinned).
 8
 9	// Drop victim caches from all pools.
10	
11	//丢弃当前victim ，stop the world 不用加锁
12	for _, p := range oldPools {
13		p.victim = nil
14		p.victimSize = 0
15	}
16
17	// Move primary cache to victim cache.
18	//将local 复制给 victim，原来的local置成nil 
19	for _, p := range allPools {
20		p.victim = p.local
21		p.victimSize = p.localSize
22		p.local = nil
23		p.localSize = 0
24	}
25
26	// The pools with non-empty primary caches now have non-empty
27	// victim caches and no pools have primary caches.
28	oldPools, allPools = allPools, nil
29}

1type poolLocalInternal struct {
2	private any       // Can be used only by the respective P.
3	shared  poolChain // Local P can pushHead/popHead; any P can popTail.
4}

 1func (p *Pool) Get() any {
 2	if race.Enabled {
 3		race.Disable()
 4	}
 5	//把当前groutine 固定在当前P上
 6	l, pid := p.pin()
 7	x := l.private  //优先从 private 字段取，快速
 8	l.private = nil
 9	if x == nil { // private 不存在
10		// Try to pop the head of the local shard. We prefer
11		// the head over the tail for temporal locality of
12		// reuse.   
13		//从当前local shared 弹出一个，注意从head 读取并移除
14		x, _ = l.shared.popHead()
15		if x == nil {
16			//如果没有 取其他P上偷一个
17			x = p.getSlow(pid)
18		}
19	}
20	runtime_procUnpin()
21	if race.Enabled {
22		race.Enable()
23		if x != nil {
24			race.Acquire(poolRaceAddr(x))
25		}
26	}
27	
28	//都没有拿到，尝试New 函数去生产一个新的
29	if x == nil && p.New != nil {
30		x = p.New()
31	}
32	return x
33}

 1func (p *Pool) getSlow(pid int) any {
 2	// See the comment in pin regarding ordering of the loads.
 3	size := runtime_LoadAcquintptr(&p.localSize) // load-acquire
 4	locals := p.local                            // load-consume
 5	// Try to steal one element from other procs.
 6	
 7	//从其他procs 偷取一个元素
 8	for i := 0; i < int(size); i++ {
 9		l := indexLocal(locals, (pid+i+1)%int(size))
10		if x, _ := l.shared.popTail(); x != nil {
11			return x
12		}
13	}
14
15	// Try the victim cache. We do this after attempting to steal
16	// from all primary caches because we want objects in the
17	// victim cache to age out if at all possible.
18	
19	//如果其他procs 没有可用元素，尝试从 victim 中获取
20	size = atomic.LoadUintptr(&p.victimSize)
21	if uintptr(pid) >= size {
22		return nil
23	}
24	locals = p.victim
25	l := indexLocal(locals, pid)
26	if x := l.private; x != nil { //同样的逻辑 先从 victim的local private中获取
27		l.private = nil
28		return x
29	}
30	for i := 0; i < int(size); i++ {  //从victim 其他proc 偷取
31		l := indexLocal(locals, (pid+i)%int(size))
32		if x, _ := l.shared.popTail(); x != nil {
33			return x
34		}
35	}
36
37	// Mark the victim cache as empty for future gets don't bother
38	// with it.
39	//标记 这个victim为空，以后查找快速跳过
40	atomic.StoreUintptr(&p.victimSize, 0)
41
42	return nil

 1// Put adds x to the pool.
 2func (p *Pool) Put(x any) {
 3	if x == nil { // nil 直接丢弃
 4		return
 5	}
 6	if race.Enabled {
 7		if fastrandn(4) == 0 {
 8			// Randomly drop x on floor.
 9			return
10		}
11		race.ReleaseMerge(poolRaceAddr(x))
12		race.Disable()
13	}
14	l, _ := p.pin()
15	if l.private == nil { //如果本地private 为空， 则直接设置这个指即可
16		l.private = x
17		x = nil
18	}
19	if x != nil { // 否则加入到本地队列
20		l.shared.pushHead(x)
21	}
22	runtime_procUnpin()
23	if race.Enabled {
24		race.Enable()
25	}
26}

 11)	func (c *Cond) Wait()  wait 之前加上锁
 2该函数的作用可归纳为如下三点：
 3a)	阻塞等待条件变量满足	
 4b)	释放已掌握的互斥锁相当于cond.L.Unlock()。 注意：两步为一个原子操作。（原子操作，一起获得cpu ）
 5c)	当被唤醒时，Wait()函数返回时，解除阻塞并重新获取互斥锁。相当于cond.L.Lock()
 6
 72)	func (c *Cond) Signal()
 8	单发通知，给一个正等待（阻塞）在该条件变量上的goroutine（线程）发送通知。
 93)	func (c *Cond) Broadcast()
10广播通知，给正在等待（阻塞）在该条件变量上的所有goroutine（线程）发送通知。

 1
 2type Cond struct {
 3    noCopy noCopy
 4
 5    // 当观察或者修改等待条件的时候需要加锁
 6    L Locker
 7
 8    // 等待队列
 9    notify  notifyList
10    checker copyChecker
11}
12
13func NewCond(l Locker) *Cond {
14    return &Cond{L: l}
15}
16
17func (c *Cond) Wait() {
18    c.checker.check()
19    // 增加到等待队列中
20    t := runtime_notifyListAdd(&c.notify)
21    c.L.Unlock()
22    // 阻塞休眠直到被唤醒
23    runtime_notifyListWait(&c.notify, t)
24    c.L.Lock()
25}
26
27func (c *Cond) Signal() {
28    c.checker.check()
29    runtime_notifyListNotifyOne(&c.notify)
30}
31
32func (c *Cond) Broadcast() {
33    c.checker.check()
34    runtime_notifyListNotifyAll(&c.notify）
35}