Go 基础知识与框架体系 系列十：Golang 并发包 sync 系列

1
2
3
4
5
6

type Map struct {
	mu Mutex
	read atomic.Value // readOnly
	dirty map[interface{}]*entry
	misses int
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

func (m *Map) Load(key interface{}) (value interface{}, ok bool) {
  // 首先从只读字段读取内容
	read, _ := m.read.Load().(readOnly)
  e, ok := read.m[key]
  // 如果没读到，并且dirty有read没有数据则从dirty中读取
	if !ok && read.amended {
		m.mu.Lock()
		// 在从dirty读取前需要加锁后再做一次验证，防止期间read突然有数据，也就是double check
		read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
		if !ok && read.amended {
			e, ok = m.dirty[key]
      // 将此次记录记录添加到miss中，可以看到这里没对dirty的取值做判断，也就是说不管是否
      // 取到miss都会添加一次
			m.missLocked()
		}
		m.mu.Unlock()
	}
	if !ok {
		return nil, false
	}
	return e.load()
}

 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

func (m *Map) Store(key, value interface{}) {
  // 存储之前先从只读字段读取要存储的值，如果存在，则用CAS的方式将新的值存储进去
  read, _ := m.read.Load().(readOnly)
  // tryStore 会检查dirty的keys值是否已经删除，
  // 如果没有删除标记，则直接采用CAS方式存储entry
	if e, ok := read.m[key]; ok && e.tryStore(&value) {
		return
	}

  m.mu.Lock()
  // double check
	read, _ = m.read.Load().(readOnly)
	if e, ok := read.m[key]; ok {
    // double check 如果read存在，调用 unexpungeLocked 将 expunged 设置为 nil，
    // 然后更新dirty，expunged 表示dirty中记录的删除标识（read没同步），由于有新的值存储需要
    // 将删除标识更新。
		if e.unexpungeLocked() {
			m.dirty[key] = e
		}
		e.storeLocked(&value)
	} else if e, ok := m.dirty[key]; ok {
    // 如果read中没有对应的键，从dirty中有则直接更新dirty中的键
		e.storeLocked(&value)
	} else {
    // dirty 和 read 都不存在这个键的情况
		if !read.amended {
      // amended为true标识dirty包含read没有的key，由于dirty是最全的数据，amend为false只有两种
      // 情况，一种就是 dirty 的键值等于 read 的键值，一种是dirty为空的时候，所以这里只有可能是
			// 第二种，也就是dirty为空，因此再store 之前先判断一下 dirty map 是否为空，如果为空，就把 read map 浅拷贝一次。
			m.dirtyLocked()
			m.read.Store(readOnly{m: read.m, amended: true})
    }
    // 如果dirty数据和read的key不同步数据，直接将值写入dirty
		m.dirty[key] = newEntry(value)
	}
	m.mu.Unlock()
}

 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

func (m *Map) Delete(key interface{}) {
	m.LoadAndDelete(key)
}

func (m *Map) LoadAndDelete(key interface{}) (value interface{}, loaded bool) {
	read, _ := m.read.Load().(readOnly)
	e, ok := read.m[key]
	if !ok && read.amended {
		m.mu.Lock()
		read, _ = m.read.Load().(readOnly)
		e, ok = read.m[key]
		if !ok && read.amended {
			e, ok = m.dirty[key]
			m.missLocked()
		}
		m.mu.Unlock()
	}
	if ok {
		return e.delete()
	}
	return nil, false
}

func (e *entry) delete() (value interface{}, ok bool) {
	for {
		p := atomic.LoadPointer(&e.p)
		if p == nil || p == expunged {
			return nil, false
		}
		if atomic.CompareAndSwapPointer(&e.p, p, nil) {
			return *(*interface{})(p), true
		}
	}
}

 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

func (m *Map) Range(f func(key, value interface{}) bool) {
  // 如果amend为ture，说明dirty包含了read所有的key，将dirty提升为read，
  // 并将dirty设置为nil，之后用Store存储新的值的时候再拷贝回来
  // 最后对read进行遍历即可
	read, _ := m.read.Load().(readOnly)
	if read.amended {
		m.mu.Lock()
		read, _ = m.read.Load().(readOnly)
		if read.amended {
			read = readOnly{m: m.dirty}
			m.read.Store(read)
			m.dirty = nil
			m.misses = 0
		}
		m.mu.Unlock()
	}

	for k, e := range read.m {
		v, ok := e.load()
		if !ok {
			continue
		}
		if !f(k, v) {
			break
		}
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

type Pool struct {
  // noCopy 表示不支持值拷贝，如果出现值拷贝用 go vet 编译检查的时候会报错
	noCopy noCopy

  // [P]poolLocal，表示每个local的P池
	local     unsafe.Pointer
  // local的长度
	localSize uintptr

  // 也是[P]poolLocal，表示上一个生命周期的local
	victim     unsafe.Pointer
  // victim的长度
	victimSize uintptr

  // 用于创建新对象方法，get获取不到就会调用创建一个新对象，一般由用户传入
	New func() interface{}
}

 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

func TestPoolNew(t *testing.T) {
    // disable GC so we can control when it happens.
    defer debug.SetGCPercent(debug.SetGCPercent(-1))

    i := 0
    p := Pool{
        New: func() interface{} {
            i++
            return i
        },
    }
    if v := p.Get(); v != 1 {
        t.Fatalf("got %v; want 1", v)
    }
    if v := p.Get(); v != 2 {
        t.Fatalf("got %v; want 2", v)
    }

    // Make sure that the goroutine doesn't migrate to another P
    // between Put and Get calls.
    Runtime_procPin()
    p.Put(42)
    if v := p.Get(); v != 42 {
        t.Fatalf("got %v; want 42", v)
    }
    Runtime_procUnpin()

    if v := p.Get(); v != 3 {
        t.Fatalf("got %v; want 3", v)
    }
}

1
2
3

runtime.SetFinalizer(v, func(vv *string) {
    atomic.AddUint32(&fin, 1)
})

1
2
3
4
5
6

// poolDequeue testing.
type PoolDequeue interface {
    PushHead(val interface{}) bool
    PopHead() (interface{}, bool)
    PopTail() (interface{}, bool)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

type poolLocal struct {
    poolLocalInternal

    // 将 poolLocal 补齐至两个缓存行的倍数，防止 false sharing,
    // 每个缓存行具有 64 bytes，即 512 bit
    // 目前我们的处理器一般拥有 32 * 1024 / 64 = 512 条缓存行
    // 伪共享，仅占位用，防止在 cache line 上分配多个 poolLocalInternal
    pad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte
}

// Local per-P Pool appendix.
type poolLocalInternal struct {
    // P 的私有缓存区，使用时无需要加锁
    private interface{}
    // 公共缓存区。本地 P 可以 pushHead/popHead；其他 P 则只能 popTail
    shared  poolChain
}

 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

type poolChain struct {
    // 只有生产者会 push to，不用加锁
    head *poolChainElt

    // 读写需要原子控制。 pop from
    tail *poolChainElt
}

type poolChainElt struct {
    poolDequeue

    // next 被 producer 写，consumer 读。所以只会从 nil 变成 non-nil
    // prev 被 consumer 写，producer 读。所以只会从 non-nil 变成 nil
    next, prev *poolChainElt
}

type poolDequeue struct {
    // The head index is stored in the most-significant bits so
    // that we can atomically add to it and the overflow is
    // harmless.
    // headTail 包含一个 32 位的 head 和一个 32 位的 tail 指针。这两个值都和 len(vals)-1 取模过。
    // tail 是队列中最老的数据，head 指向下一个将要填充的 slot
    // slots 的有效范围是 [tail, head)，由 consumers 持有。
    headTail uint64

    // vals 是一个存储 interface{} 的环形队列，它的 size 必须是 2 的幂
    // 如果 slot 为空，则 vals[i].typ 为空；否则，非空。
    // 一个 slot 在这时宣告无效：tail 不指向它了，vals[i].typ 为 nil
    // 由 consumer 设置成 nil，由 producer 读
    vals []eface
}

1
2
3
4
5
6
7

// Prime to not correlate with any user patterns.
const semTabSize = 251

var semtable [semTabSize]struct {
 root semaRoot
 pad  [cpu.CacheLinePadSize - unsafe.Sizeof(semaRoot{})]byte
}

1
2
3
4
5
6
7

// 为sync.Mutex准备的异步信号量
// golang.org/issue/17953 可以查看引入二级列表之前性能较差的程序示例test/locklinear.go
type semaRoot struct {
 lock  mutex
 treap *sudog // 平衡树的根节点
 nwait uint32 // Number of waiters. Read w/o the lock.
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

type sudog struct {
 g *g
 next *sudog
 prev *sudog
 elem unsafe.Pointer //数据元素 (可能指向栈)
 // 下面的字段不会并发访问
 // 对于channels, waitlink 只被g访问
  // 对于semaphores, 所有自动(包括上面的)只有获取semaRoot的锁才能被访问
 acquiretime int64
 releasetime int64
 ticket      uint32
  //isSelect表示g正在参与一个select，因此必须对g.selectDone进行CAS才能赢得唤醒竞争
 isSelect bool
  //success表示channel c上的通信是否成功。如果goroutine因为在通道c上传递了一个值而被唤醒，则为true；
  //如果因为channel c关闭而被唤醒，则为false
 success bool
  
 parent   *sudog // semaRoot binary tree
 waitlink *sudog // g.waiting list or semaRoot
 waittail *sudog // semaRoot
 c        *hchan // channel
}

1
2
3
4

type Once struct {
   done uint32 // 初始值为0表示还未执行过，1表示已经执行过
   m    Mutex 
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

func (o *Once) Do(f func()) {
   // 判断done是否为0，若为0，表示未执行过，调用doSlow()方法初始化
   if atomic.LoadUint32(&o.done) == 0 {
      // Outlined slow-path to allow inlining of the fast-path.
      o.doSlow(f)
   }
}


// 加载资源
func (o *Once) doSlow(f func()) {
   o.m.Lock()
   defer o.m.Unlock()
   // 采用双重检测机制 加锁判断done是否为零
   if o.done == 0 {
      // 执行完f()函数后，将done值设置为1
      defer atomic.StoreUint32(&o.done, 1)
      // 执行传入的f()函数
      f()
   }