并发安全之原子操作

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

func mutexAdd() {
 var a int32 =  0
 var wg sync.WaitGroup
 var mu sync.Mutex // 互斥锁
 start := time.Now()
 for i := 0; i < 100000000; i++ {
  wg.Add(1)
  go func() {
   defer wg.Done()
   mu.Lock()
   a += 1
   mu.Unlock()
  }()
 }
 wg.Wait()
 timeSpends := time.Now().Sub(start).Nanoseconds()
 fmt.Printf("use mutex a is %d, spend time: %v\n", a, timeSpends)
}

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

func AtomicAdd() {
 var a int32 =  0
 var wg sync.WaitGroup
 start := time.Now()
 for i := 0; i < 1000000; i++ {
  wg.Add(1)
  go func() {
   defer wg.Done()
   atomic.AddInt32(&a, 1)
  }()
 }
 wg.Wait()
 timeSpends := time.Now().Sub(start).Nanoseconds()
 fmt.Printf("use atomic a is %d, spend time: %v\n", atomic.LoadInt32(&a), timeSpends)
}

1
2
3

func CompareAndSwapInt32(addr *int32, old, new int32) (swapped bool)

func CompareAndSwapPointer(addr *unsafe.Pointer, old, new unsafe.Pointer) (swapped bool)

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

// 定义一个struct类型P
type P struct{ x, y, z int }
  
// 执行类型P的指针
var pP *P
  
func main() {
  
    // 定义一个执行unsafe.Pointer值的指针变量
    var unsafe1 = (*unsafe.Pointer)(unsafe.Pointer(&pP))
  
    // Old pointer
    var sy P
  
    // 为了演示效果先将unsafe1设置成Old Pointer
    px := atomic.SwapPointer(
        unsafe1, unsafe.Pointer(&sy))
  
    // 执行CAS操作，交换成功，结果返回true
    y := atomic.CompareAndSwapPointer(
        unsafe1, unsafe.Pointer(&sy), px)
  
    fmt.Println(y)
}

1
2
3
4

type Mutex struct {
 state int32
 sema  uint32
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

func (m *Mutex) Lock() {
   // Fast path: grab unlocked mutex.
   if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
       if race.Enabled {
           race.Acquire(unsafe.Pointer(m))
       }
       return
   }
   // Slow path (outlined so that the fast path can be inlined)
    m.lockSlow()
}

1
2
3
4
5
6
7

func StoreInt32(addr *int32, val int32)

func StoreInt64(addr *int64, val int64)

func StorePointer(addr *unsafe.Pointer, val unsafe.Pointer)

...

 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

type Demo struct {
	A int
	B int
}

var rect atomic.Value

func main() {
	wg := sync.WaitGroup{}
	var container *Demo = &Demo{}
	// 10 个协程并发更新
	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			container.A = i
			container.B = i + 1
			rect.Store(container)
		}()
	}
	wg.Wait()
	_r := rect.Load().(*Demo)

	fmt.Printf("rect.width=%d\nrect.length=%d\n", _r.A, _r.B)
}

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

type Pool interface {
    // Name returns the corresponding pool name.
    Name() string
    // SetCap sets the goroutine capacity of the pool.
    SetCap(cap int32)
    // Go executes f.
    Go(f func())
    // CtxGo executes f and accepts the context.
    CtxGo(ctx context.Context, f func())
    // SetPanicHandler sets the panic handler.
    SetPanicHandler(f func(context.Context, interface{}))
    // WorkerCount returns the number of running workers
    WorkerCount() int32
}

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

type pool struct {
    // The name of the pool
    name string

    // capacity of the pool, the maximum number of goroutines that are actually working
    cap int32 // 也就是限制 worker 的数量

    // Configuration information
    config *Config // gopool提供默认配置，当任务数大于1时，就新开一个goroutine

    // 真正放函数的地方，所有的函数都放在这里
    // linked list of tasks
    taskHead  *task      // 头节点
    taskTail  *task      // 指向最后一个节点，实现一个O(1)的复杂度添加
    taskLock  sync.Mutex // 保证并发安全
    taskCount int32      // 任务数量

    // Record the number of running workers
    workerCount int32 // 正在跑的worker

    // This method will be called when the worker panic
    panicHandler func(context.Context, 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
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

func (p *pool) Name() string {
    return p.name
}

// SetCap 通过CPU指令集保证是原子的方式存到指定地址，保证并发安全
func (p *pool) SetCap(cap int32) {
    // 将值原子地设置到p里
    atomic.StoreInt32(&p.cap, cap)
}

func (p *pool) Go(f func()) {
    p.CtxGo(context.Background(), f)
}

func (p *pool) CtxGo(ctx context.Context, f func()) {
    // 池里取空的对象
    t := taskPool.Get().(*task)
    // 置新
    t.ctx = ctx
    t.f = f
    // 下面代码逻辑上锁
    p.taskLock.Lock()
    // 将 task 放到任务链表里，这个过程由于是多个Goroutine操作并发操作，都在往任务链表里塞 task ，
    // 因此，需要将task入链表的代码锁住，防止并发问题
    if p.taskHead == nil { // 链表为空
        p.taskHead = t
        p.taskTail = t
    } else { //
        p.taskTail.next = t
        p.taskTail = t
    }
    p.taskLock.Unlock()
    // 然后原子更改任务状态，不需要加锁（避免上下文切换），这样可以获得更好的性能
    atomic.AddInt32(&p.taskCount, 1)
    // The following two conditions are met:
    // 1. the number of tasks is greater than the threshold.
    // 2. The current number of workers is less than the upper limit p.cap.
    // or there are currently no workers.
    // 根据默认的配置：worker数大于等于 p.cap时就会，就不会新建 worker了，此时就会复用已有的worker来执行；
    // 在小于p.cap时就会每个task新建一个goroutine
    if (atomic.LoadInt32(&p.taskCount) >= p.config.ScaleThreshold && p.WorkerCount() < atomic.LoadInt32(&p.cap)) || p.WorkerCount() == 0 {
        // 新开一个 worker
        p.incWorkerCount()
        // 池里拿
        w := workerPool.Get().(*worker)
        // p放到worker的pool里
        w.pool = p
        // 再开一个协程跑，取这个worker的任务并执行
        w.run()
    }
}

// SetPanicHandler the func here will be called after the panic has been recovered.
func (p *pool) SetPanicHandler(f func(context.Context, interface{})) {
    p.panicHandler = f
}

func (p *pool) WorkerCount() int32 {
    return atomic.LoadInt32(&p.workerCount)
}

func (p *pool) incWorkerCount() {
    atomic.AddInt32(&p.workerCount, 1)
}

func (p *pool) decWorkerCount() {
    atomic.AddInt32(&p.workerCount, -1)
}

1
2
3

type worker struct {
    pool *pool // 多个worker都指向这个pool
}

 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

func (w *worker) run() {
    go func() {
        for {
            var t *task

            // 必须上锁，因为可能有多个 worker（也就是多个协程） 同时再对这个pool里的task链表做取出操作，并执行这个task
            w.pool.taskLock.Lock()

            // 不断拿出 task
            if w.pool.taskHead != nil {
                t = w.pool.taskHead
                w.pool.taskHead = w.pool.taskHead.next
                // 原子的减少任务数
                atomic.AddInt32(&w.pool.taskCount, -1)
            }

            // 任务取完了
            if t == nil {
                // if there's no task to do, exit
                w.close()
                w.pool.taskLock.Unlock()
                // 回收
                w.Recycle()
                return
            }

            // 释放时机到了
            w.pool.taskLock.Unlock()

            // 匿名函数调用，退一个栈从而形成独立环境，不让当前的defer去捕获外面函数里的异常，只是捕获这个执行函数的panic
            // 确保这个defer一定是捕获的task的panic
            func() {
                // 处理panic
                defer func() {
                    if r := recover(); r != nil {
                        if w.pool.panicHandler != nil {
                            w.pool.panicHandler(t.ctx, r)
                        } else {
                            msg := fmt.Sprintf("GOPOOL: panic in pool: %s: %v: %s", w.pool.name, r, debug.Stack())
                            logger.CtxErrorf(t.ctx, msg)
                        }
                    }
                }()
                // 真正地执行函数
                t.f()
            }()
            // 回收task
            t.Recycle()
        }
    }()
}

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

// 协程数减一
func (w *worker) close() {
    w.pool.decWorkerCount()
}

func (w *worker) zero() {
    w.pool = nil
}

// Recycle 回收worker
func (w *worker) Recycle() {
    w.zero()
    // 放回池里
    workerPool.Put(w)
}

1
2
3
4
5
6

type task struct {
    ctx context.Context
    f   func()

    next *task
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

func (t *task) zero() {
    t.ctx = nil
    t.f = nil
    t.next = nil
}

// Recycle 回收到 sync.Pool里
func (t *task) Recycle() {
    // task 置空
    t.zero()
    // 放到池里
    taskPool.Put(t)
}