select loop の小ネタ

Go の select loop で Chan を扱うときの小ネタ。

擬似乱数的な

case は上から評価されるけど、その結果実行可能な case が複数ある場合は、そのどれかが選ばれる。
どちらが選ばれるかは実行環境依存。

func main() {
    c := make(chan int)
    go func() {
        for {
            select {
            case i := <-c:
                print(i) // 0, 1 のどちらかがランダムで表示される
            }
        }
    }()

    for {
        select {
        case c <- 0:
        case c <- 1:
        }
    }
}

nil channel

channel の case が nil だと、そこに case が存在しないのと同じ扱いにある。
これを、 chan の読み取りをストップするなどに使うことができる。

1 秒ごとに読み出しを停止/再開する例。

func main() {
    stop := false
    c := make(chan int)
    interval := time.Tick(1 * time.Second)

    go func() {
        for {
            // 0.1 秒ごとに書き込む
            time.Sleep(100 * time.Millisecond)
            c <- 1
        }
    }()

    for {
        var read chan int
        if !stop {
            // ここを通らないと read は nil
            read = c
        }

        select {
        case j := <-read:
            // read が nil で無いときだけ読み出す
            log.Println(j)
        case <-interval:
            // 1 秒ごとに切り替える
            stop = !stop
        }
    }
}

channel buffer と select write

名前がよくわからないけど、以下のように worker にブロードキャストする時、読み出しで遅いのがいると詰まる。

func worker(t time.Duration) chan int {
    ch := make(chan int)
    go func() {
        for {
            time.Sleep(t * time.Millisecond)
            log.Println(<-ch)
        }
    }()
    return ch
}

func main() {
    workers := [](chan int){
        worker(10),
        worker(1000), // slow woker
        worker(10),
    }

    for i := 0; i < 100; i++ {
        time.Sleep(10 * time.Millisecond)
        for _, w := range workers {
            // slow worker のせいで詰まる
            w <- i
        }
    }
}

これは、書き込みを goroutine に飛ばすことで一応解決はする。

    for i := 0; i < 100; i++ {
        time.Sleep(10 * time.Millisecond)
        for _, w := range workers {
            // 書き込みを goroutine に飛ばす
            go func(i int) {
                w <- i
            }(i)
        }
    }

しかし、もし worker の読み出しが致命的に遅いと、ブロックした goroutine が大量にゾンビとなる可能性があるので、それを防止する仕組みが別途必要になる。

一つは chan に buffer を設定すること。以下では 20 個にしていて、これが MQ の役割になる。
しかし、ただ設定するだけだと、満杯になったとき結局ブロックするので、ここを switch で抜けるようにする。

func worker(t time.Duration) chan int {
    // chan にバッファを設定
    ch := make(chan int, 20)
    go func() {
        for {
            time.Sleep(t * time.Millisecond)
            log.Println(<-ch)
        }
    }()
    return ch
}

func main() {
    workers := [](chan int){
        worker(10),
        worker(1000), // slow woker
        worker(10),
    }

    for i := 0; i < 100; i++ {
        time.Sleep(10 * time.Millisecond)
        for _, w := range workers {
            select {
            case w <- i: // buffer が満杯だと実行されない
            default:
                // 空のデフォルトがあることでこの select を抜けられる。
                // もし無いと、 w が書き込み可能になるまで select がブロックする
            }
        }
    }

    // main を抜けないようにしてるだけ
    time.Sleep(10 * time.Second)
}

この場合は、 slow worker は 20 個だけ処理して。残りの 80 は受け取らない。