Lec5: Go, Threads, and Raft

[kmansei]

Code sample: http://nil.csail.mit.edu/6.824/2020/notes/go-concurrency.tar.gz Lecture: https://www.youtube.com/watch?v=UzzcUS2OHqo

[kmansei]

クロージャ(gpt)

Go言語（またはGolang）におけるクロージャ（Closure）は、関数とその関数が作成された環境（スコープ）を組み合わせた特殊な関数の形式です。クロージャは、自己完結した動作を実現し、外部の変数にアクセスできるようにする仕組みを提供します。

通常、関数が実行されると、その関数が使用する変数はスタック上に配置され、関数が終了するとスタックから削除されます。しかし、クロージャでは、その関数が終了しても、その関数内で使用されている変数へのアクセスが維持されます。これは、クロージャがその外部環境への参照を保持することによって実現されます。

[kmansei]

クロージャの特性が表れている例

func main() {
  var wg sync.WaitGroup
    for i := 0; i < 5; i++ {
      wg.Add(1)
        go func(x int) {
          sendRPC(x)
            wg.Done()
        }(i)
    }
  wg.Wait()
}

func sendRPC(i int) {
  println(i)
}

出力

$ go run goroutine/loop.go
1
0
2
3
4

func main() {
  var wg sync.WaitGroup
    for i := 0; i < 5; i++ {
      wg.Add(1)
        go func() {
          sendRPC(i)
            wg.Done()
        }()
    }
  wg.Wait()
}

func sendRPC(i int) {
  println(i)
}

出力

$ go run goroutine/bad.go
5
5
5
5
5

後者のプログラムのクロージャは、そのクロージャを作成したmain関数の変数iを参照している。 main関数は並列で動いているため、iが変わるとクロージャのiの値も変わる

[kmansei]

https://youtu.be/UzzcUS2OHqo?t=609

基本的にlockを獲得したら、returnする前にunlockすべきである。 ただ、今回はmainがdone=trueにした後は、lockを獲得することは無いので問題にはなっていない。

func main() {
  time.Sleep(1 * time.Second)
    println("started")
    go periodic()
    time.Sleep(5 * time.Second) // wait for a while so we can observe what ticker does
    mu.Lock()
    done = true
    mu.Unlock()
    println("cancelled")
    time.Sleep(3 * time.Second) // observe no output
}

func periodic() {
  for {
    println("tick")
    time.Sleep(1 * time.Second)
    mu.Lock()
    if done {
      mu.Unlock()  //これは入れた方が良くね
      return
    }
    mu.Unlock()
  }
}

[kmansei]

Synchronization Primitive

• Sync.WaitGroup
• Sync.Mutex
• Sync.NewCond

Go concurrency primitives Channel, Sync

[kmansei]

vote-count4.go

import (
    "math/rand"
    "sync"
    "time"
)

func main() {
    rand.Seed(time.Now().UnixNano())

    count := 0
    finished := 0
    var mu sync.Mutex
    cond := sync.NewCond(&mu)

    for i := 0; i < 10; i++ {
        go func() {
            vote := requestVote()
            mu.Lock()
            defer mu.Unlock()
            if vote {
                count++
            }
            finished++
            cond.Broadcast()  //cond.waitでwaitsetに入っているgoroutine全てを起こす
        }()
    }

    mu.Lock()
    for count < 5 && finished != 10 {
        cond.Wait()  //条件がTrueなら、mu.unlockして、他のgoroutineに呼ばれるまで待つ
                                      //signalやbroadcastで起こされたら、lockを獲得して再度条件を確認
    }
    if count >= 5 {
        println("received 5+ votes!")
    } else {
        println("lost")
    }
    mu.Unlock()
}

func requestVote() bool {
    time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
    return rand.Int()%2 == 0
}

[kmansei]

vote-count2.goのようなビジーループで処理を待つのは重い。

vote2-count.go

    for {
        mu.Lock()

        if count >= 5 || finished == 10 {
            break
        }
        mu.Unlock()
    }

vote-count3.goのようにsleepを入れればある程度、マシになるもののvote-count4.goのようにSync.condを使えば条件確認が必要なタイミングでロックを獲得、確認、解放を行ってくれるので便利かつ処理も軽い

vote3-count.go

    for {
        mu.Lock()
        if count >= 5 || finished == 10 {
            break
        }
        mu.Unlock()
        time.Sleep(50 * time.Millisecond)
    }

[kmansei]

vote-count4がやっていること

mu.Lock()
// do something that might affect the condition
cond.Broadcast()
mu.Unlock()

----

mu.Lock()
while condition == false {
    cond.Wait()
}
// now condition is true, and we have the lock
mu.Unlock()

[kmansei]

Goのチャンネルはキューのようなものだと思うかもしれないが、async/awaitのような同期をとる仕組みだと思った方が良い https://youtu.be/UzzcUS2OHqo?t=2147

unbuffered.go

func main() {
    c := make(chan bool)
    go func() {
        time.Sleep(1 * time.Second)
        <-c
    }()
    start := time.Now()
    c <- true // blocks until other goroutine receives
    fmt.Printf("send took %v\n", time.Since(start))
}

[kmansei]

チャンネルはsenderは別のgoroutineがreceiverするまで自身ブロックし、receiverは別のgoroutineが値をsendするまで自身をブロックするため、デッドロックが容易に起こる

package main

func main() {
    c := make(chan bool)
    c <- true
    <-c
}

$ go run channel/unbuffered-deadlock.go
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan send]:
main.main()
    /Users/s17421/Downloads/go-concurrency/channel/unbuffered-deadlock.go:5 +0x38
exit status

[kmansei]

上記のようにバグを検出して停止できているのはまだ良い例であり、下記のプログラムの場合はgoroutineが無限ループであるものの動いているため、プログラムは停止しない

https://youtu.be/UzzcUS2OHqo?t=2280

package main

func main() {
    go func() {
        for {
        }
    }()

    c := make(chan bool)
    c <- true
    <-c
}

[kmansei]

https://youtu.be/UzzcUS2OHqo?t=2360

バッファードチャンネルはチャンネルの容量が最大になるまでは、senderは自身をブロックしない

[kmansei]

https://youtu.be/UzzcUS2OHqo?t=2683

goroutineの同期を取りたいときは基本的にChannelを使うよりも、Synchronization Primitiveを使う用法が良い。

[kmansei]

https://youtu.be/UzzcUS2OHqo?t=2797 Raftを実装する上でありがちな二つのバグの紹介

[kmansei]

https://youtu.be/UzzcUS2OHqo?list=PLrw6a1wE39_tb2fErI4-WkMbsvGQk9_UB&t=3068

Raftのデッドロックのシナリオ

• サーバー(スレッド)は3つ
• サーバー0がAttemptElectionして、サーバー1と2にRequestVoteする
• サーバー1がAttemptElectionして、サーバー0と2にRequestVoteする
• サーバー2がサーバー1と0に返信する
• サーバー0とサーバー1がお互いのRequestVoteの返信を待っている
• 全てのスレッドが停止(deadlock)

プログラムレベル

RPCコールの間ずっとlockを獲得しているのが問題

[kmansei]

https://youtu.be/UzzcUS2OHqo?list=PLrw6a1wE39_tb2fErI4-WkMbsvGQk9_UB&t=3305

Leaderが同じtermで二つできてしまった実装例

[kmansei]

https://youtu.be/UzzcUS2OHqo?list=PLrw6a1wE39_tb2fErI4-WkMbsvGQk9_UB&t=3690

• DPrintf関数がdebugにおすすめ
  • デバッグログが仕込んでバグのスコープを狭めて特定
• ctrl+backslash
  • Goプログラムとgoroutineを止めて、stacktracewを出力