Concurrency in Go 書本 - An Introduction to Concurrency 筆記

發表於 更新於 分類於 Disqus： 文章字數： 7.4k 所需閱讀時間 ≈ 13 分鐘

距離我上次寫文章，已經好幾個月了，最近終於忙完學校的事情，從八月開始算是要邁入社會新鮮人了，當然也希望我的寫 blog 的習慣可以一直持續下去。

今天這篇是開啟將一些我在看的實體書或是電子書的筆記記錄起來，所以之後應該會連續發好幾篇同一本書的筆記，為避免篇幅過長，我會將書本中的每一個章節都分別寫筆記記起來，會在 blog 寫出來的都是我覺得比較需要記起來的。

今天要提的是這本書：Concurrency in Go

個人認為這本書還不錯，它提供了很多寫 Concurrency 有哪一些 pattern 可以使用，之後應該也會將這些 concurrency pattern 放在我的 golib 上，這個 repo 是包含我平常寫專案常用的套件封裝，個人覺得還滿實用的，歡迎大家可以給個 star 並且發個 PR 提供更多好用的工具。

今天的主題是第一章的筆記：An Introduction to Concurrency

Why Is Concurrency Hard ?

Concurrency 的 code 一直是不容易 debug 的，原因在於一開始運作的可能如你預期的，而當運行越久後，可能資料量越多，導致資料讀寫非常頻繁，那麼如果 Concurrency 的 Code 沒寫好，這時候一些潛在的 bug 就可能會出現。

Race Conditions

那麼說到 bug，就必須要介紹何謂 Race Condition，這個名詞其實在我以前講 database 的 race condition 就解釋過了，有興趣的朋友可以再去翻來看。

簡單來說，race condition 指的是多個 operations 按照指定的順序執行，但是出來的結果卻不同於指定的順序執行該得到的結果。

我們來看個例子：

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

import "fmt"

var data int

func main() {
    go func() {
        data++
    }()

    if data == 0 {
        fmt.Printf("the value is %v.\n", data)
    }
}

這段 code 會有 race condition，因為根據這個例子，其執行結果就會有三種可能：

  • 沒有任何輸出
  • “the value is 0”
  • “the value is 1”

之所以會這樣，是因為我們可以知道這個例子有兩個 goroutine，分別是 main 與 匿名 func，而 data 這個 variable 由於被這兩個 goroutine 同時存取，那麼由於沒有所謂的 lock 機制去保護 data 在同一時間內只有一個 goroutine 去存取。

所以沒有任何輸出的結果來自於 匿名 func 先對 data++，接著 main 存取，所以 if condition 沒有成立，**"the value is 0" 的結果來自於 main 先存取 data，“the value is 1”** 的結果來自於 匿名 func 先存取 data。

OK，那麼怎麼避免 race condition？最快也是最爛的解決方式是這樣：

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

import (
    "fmt"
    "time"
)

var data int

func main() {
    go func() {
        data++
    }()
    time.Sleep(1 * time.Second)
    if data == 0 {
        fmt.Printf("the value is %v.\n", data)
    }
}

要知道這樣的做法，並沒有成功解決 race condition，而是降低 race condition 出現的機率而已。

Atomicity

那麼要真正的解決 race condition 的話，勢必也要介紹一個名詞 Atomicity 原子性。

原子性指的就是一個流程或操作之中，無法被外部因素影響，而造成中斷。而對具有原子性的操作而言，只有兩種可能：

  • 操作內所有的動作、變更都順利完成。
  • 操作失敗，內部的動作、變更全部不存在。

但是當一群原子性的操作執行的時候，也是會造成 race condition，為什麼？重點在於你當下執行的 context 為何。

舉例來說：

1
i++

這個指令看起來好像是原子性，但是實際上他執行起來會分為三個動作：

  • 存取 i 的值
  • 增加 i 的值
  • 將增加後的值存入 i 中

雖然上面三個動作都是原子性的，但是一起執行就未必會是原子性，要看你當下的 context 而定，如果你當下的情境是你的程式不是 concurrent 運作，那麼這個動作是原子性的，如果你當下的情境是這個操作執行時不會被其他 goroutine 存取到，那麼這個動作會是原子性的。

Deadlocks, Livelocks, and Starvation

OK，那麼我們知道要避免這個動作同時被許多 goroutine 存取，我們可以採用 lock 的方式，那這邊我們就必須要先介紹何謂 critical section：

When more than one processes access a same code segment that segment is known as critical section. Critical section contains shared variables or resources which are needed to be synchronized to maintain consistency of data variable.

那我們直接來看個例子：

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
package main

import (
    "fmt"
    "sync"
)

var memoryAccess sync.Mutex
var data int

func main() {
    go func() {
        memoryAccess.Lock()
        data++
        memoryAccess.Unlock()
    }()

    memoryAccess.Lock()
    if data == 0 {
        fmt.Printf("the value is %v.\n", data)
    } else {
        fmt.Printf("the value is %v.\n", data)
    }
    memoryAccess.Unlock()
}

在這個例子中，採用 lock 來避免多個 goroutine 同時進入 critical section。所以我們可以知道，總共有三處 critical section 需要被保護：

  • Line 14：增加 data 的數值
  • Line 19：if 判斷，需要存取 data 的數值
  • Line 20 & 22：fmt.Printf 需要存取 data 的數值

為了讓程式不要有 race condition 發生，必須採用 Memory Access Synchronization 的方式來解決，那這邊就是透過 Go 的 sync.Mutex 產生 exclusive lock 的效果。

Deadlocks

但是當我們加了所謂的 lock，又可能會衍生出一個問題，那就是 deadlock。deadlock 指的是：

A deadlocked program is one in which all concurrent processes are waiting on one another. In the state, the program will never recover without outside intervention.

那麼在 Go 中如果遇到 deadlock 就會出現 all goroutines must be blocked or “asleep” 的錯誤訊息。

我們來看個例子：

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
package main

import (
    "fmt"
    "sync"
    "time"
)

type value struct {
    mu sync.Mutex
    value int
}

var wg sync.WaitGroup

func main() {
    printSum := func(v1, v2 *value) {
        defer wg.Done()
        v1.mu.Lock()
        defer v1.mu.Unlock()

        time.Sleep(2 * time.Second)
        v2.mu.Lock()
        defer v2.mu.Unlock()

        fmt.Printf("sum=%v\n", v1.value + v2.value)
    }

    var a, b value
    wg.Add(2)
    go printSum(&a, &b)
    go printSum(&b, &a)
    wg.Wait()
}

為什麼會造成 deadlock？原因在於兩個 goroutine printSum 對於 a, b 的 加鎖與解鎖相反了，也就是說當第一個 printSum 在 a 上加了鎖，然後存取 b，但是此時另一個 printSum 已經在 b 上加了鎖，並且想要存取 a ，如此一來造就兩個 goroutine 在互相等待對方的資源釋放出來，那麼這樣就造成一個無止盡的循環，形成了 deadlock 無法跳出。

那麼根據著名的 **deadlock ** 定律：Coffman Conditions

Deadlock 發生必須符合以下四項充要條件 :

  • Mutual exclusion：某些資源在同一個時間點最多只能被一個 process 使用
  • Hold and wait：某 process 持有部分資源，並等待其他 process 正在持有的資源
  • No preemption：process 不可以任意強奪其他 process 所持有的資源
  • Circular waiting：系統中存在一組 processes P=P0,P1,…,PnP=P0,P1,…,Pn，其中 P0 等待 P1 所持有的資源 … Pn 等待 P0 所持有的資源，形成循環式等待。

根據我們前面的 deadlock 的範例，可以知道確實符合這四個條件，所以會有 deadlock。

Livelock

有死鎖，那麼有活鎖嗎？有的：

Livelocks are programs that are actively performing concurrent operations, but these operations do nothing to move the state of the program forward.

直接用一個情境來想像，今天你走在人行道上，遇到了另外一個行人，你選擇讓他過，於是你往左邊站，但是另外一個行人心裡想著也是要讓你，於是他往他的右邊站，boom！你跟他又撞在一起，兩人又不約而同地往另外一邊站，這樣形成一個循環，這就是 livelock ，一樣會無法跳出。

那麼以專業的話來說就是：

活結是行程彼此釋放資源又同時占用對方釋放的資源。當此情況持續發生時，儘管資源的狀態不斷改變，但每個行程都無法取得所需資源，使得事情沒有任何進展。

我們來看個例子：

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
package main

import (
    "bytes"
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

func main() {
    cadence := sync.NewCond(&sync.Mutex{})
    go func() {
        for range time.Tick(1 * time.Millisecond) {
            cadence.Broadcast()
        }
    }()

    takeStep := func() {
        cadence.L.Lock()
        cadence.Wait()
        cadence.L.Unlock()
    }

    tryDir := func(dirName string, dir *int32, out *bytes.Buffer) bool {
        fmt.Fprintf(out, " %v", dirName)
        atomic.AddInt32(dir, 1)
        takeStep()
        if atomic.LoadInt32(dir) == 1 {
            fmt.Fprintf(out, ". Success!")
            return true
        }
        takeStep()
        atomic.AddInt32(dir, -1)
        return false
    }

    var left, right int32
    tryLeft := func(out *bytes.Buffer) bool { return tryDir("left", &left, out)}
    tryRight := func(out *bytes.Buffer) bool { return tryDir("right", &right, out)}

    walk := func(walking *sync.WaitGroup, name string) {
        var out bytes.Buffer
        defer func() {fmt.Println(out.String())}()
        defer walking.Done()
        fmt.Fprintf(&out, "%v is trying to scoot:", name)
        for i := 0; i < 10; i++ {
            if tryLeft(&out) || tryRight(&out) {
                return
            }
        }
        fmt.Fprintf(&out, "\n%v tosses her hands up in exasperation!", name)
    }

    var peopleInHallway sync.WaitGroup
    peopleInHallway.Add(2)
    go walk(&peopleInHallway, "Alice")
    go walk(&peopleInHallway, "Barbara")
    peopleInHallway.Wait()
}

這邊用到了 sync.Condition 這個作用是，結合互斥與條件的作用，為了在對應的共享數據的狀態發生變化時，通知其他因此而被阻塞的 goroutine。互斥為共享數據的訪問提供互斥支持，而條件變量可以就共享數據的狀態的變化向相關線程發出通知。

所以根據 Line 19 ~ Line 23 可以知道，假設這兩個人往左站往右站需要花費的時間，那麼會透過 cadence.Wait() 而被 block 住，透過 Line 13 ~ Line 17 來固定速度來通知被 lock 的 goroutine 可以動作。

實際運行這段程式碼就可以知道，兩個 go walk 會因為雙方都是同時向左同時向右，因此卡住了，造成 livelock。

對於 sync.Condition 不了解的話，推薦可以看這個介紹文：https://zhuanlan.zhihu.com/p/349411905

那麼要怎麼 detect livelock 呢？通常會看 CPU 使用率，如果 CPU 使用率很高，但是卻發現沒有執行任何操作，也許可能遇到 livelock 了。

Starvation

那麼搶不到鎖呢？那就會造成 goroutine 會飢餓。也就是說如果有些過度強勢的 goroutine 存在，你會發現某些 goroutine 一直無法拿到資源來工作，用專業的話來解釋：

starvation usually implies that there are one or more greedy concurrent process that are unfairly preventing one or more concurrent processes from accomplishing work as efficiently as possible, or maybe at all

我們來看個例子：

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
package main

import (
    "fmt"
    "sync"
    "time"
)

func main() {
    var wg sync.WaitGroup
    var shardLock sync.Mutex
    const runtime = 1 * time.Second

    greedyWorker := func() {
        defer wg.Done()

        var count int
        for begin := time.Now(); time.Since(begin) <= runtime; {
            shardLock.Lock()
            time.Sleep(3 * time.Nanosecond)
            shardLock.Unlock()
            count++
        }
        fmt.Printf("Greedy worker was able tto execute %v work loop\n", count)
    }

    politeWorker := func() {
        defer wg.Done()

        var count int
        for begin := time.Now(); time.Since(begin) <= runtime; {
            shardLock.Lock()
            time.Sleep(1 * time.Nanosecond)
            shardLock.Unlock()

            shardLock.Lock()
            time.Sleep(1 * time.Nanosecond)
            shardLock.Unlock()

            shardLock.Lock()
            time.Sleep(1 * time.Nanosecond)
            shardLock.Unlock()

            count++
        }
        fmt.Printf("Polite worker was able to execute %v work loop\n", count)
    }

    wg.Add(2)
    go greedyWorker()
    go politeWorker()

    wg.Wait()
}

greedyWorker 拿到 lock 就鎖了 three nanosecond，而相較於 **politeWorker ** 則是分了三次拿 lock，這樣的方式會導致 greedyWorker 拿到 lock 的次數 多於 politeWorker 三倍，也可以想成是多 loop 了三倍的次數。

那麼通常 detect starvation 的方式是，可以對每一個 worker 完成工作後 ，下 logger 後看每一個 worker 完成的速度或是做的次數，去找出是否有 starvation

總結

從 Concurrency bug 我們可以知道會有 race condition 那麼解決的方式會用到 atomicity 的原則，但是要看你的情境是不是會符合原子性，那麼要怎麼做到？可以使用 lock，使用 lock 要注意可能會遇到 Deadlock, Livelock, Starvation。