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 は受け取らない。