Goにおける並行処理 - channel編

はじめに

これはGoにおける並行処理 - goroutine編 - oinume journalの続きの記事。goroutineに引き続き、Goの並行処理を支える重要な概念であるchannelについて説明する。

channelとは？

channelはメモリに対するアクセスを同期するためやgoroutine間の通信(データの受け渡し)として使うものである。Go言語による並行処理では

水が流れる川のように、チャネルは情報の流れの水路として機能します。値はチャネルに沿って 渡され、そこから下流に読み込まれます。

と説明されている。例えば以下のプログラムでは、main関数からHello WorldをChannel messageChに書き込んで、それを無名goroutineが読み取って出力している(Playground)

package main

import (
    "fmt"
)

func main() {
    messageCh := make(chan string)
    go func() {
        fmt.Println(<-messageCh)
    }()
    messageCh <- "Hello World" 
}

channelを使う上で覚えておくべき基本的なものとしては以下である。

1. channelにデータを書き込む送信側とデータを受け取る受信側が存在している必要がある
   • 例えば上のプログラムで fmt.Println(<-messageCh) がない場合、 fatal error: all goroutines are asleep - deadlock! というエラーになる
2. 受信処理はchannelに書き込みがあるまでブロックされる
   • 上のプログラムで messageCh <- "Hello World" の書き込みがない場合は、goroutineとして実行されている無名関数がずっとブロックされ何も起こらずにプログラムが終了する。
3. 送信処理はchannelがいっぱいの場合、空きができるまでブロックされる
4. channelは一般的にはmakeで生成する
   • make(chan, 3) のように、バッファをつけて生成することができる。この例だと3個のバッファがあるため、4個目を書き込む時にブロックされる。
5. 送信専用channelと受信専用channelがある
6. channelをcloseする

channelのブロック

通常であれば、goroutineはsync.WaitGroupを使って起動元のgoroutineにjoinするようにしないと、タイミングにもよるがほぼ確実に実行されずにmain関数が終了されてしまう。しかし、上のHello Worldを出力するgoroutineでは、必ずHello Worldと出力される。これはいったいなぜだろうか？

理由は、channelからデータを受信する場合データが送信されるまでブロックするからである。つまり、上のプログラムの実行順序は以下のようになる。

func main() {
    messageCh := make(chan string)  // (1) channelを宣言
    go func() { // (2) goroutineの実行
        fmt.Println(<-messageCh) // (4) channelからデータを受信(送信されるまでブロックされている)
    }()
    messageCh <- "Hello World"      // (3) channelへデータを送信
}

つまりchannelとgoroutineを使うことで、何かデータを受信するまで待って、受信したら処理を開始する というようなWorker処理を簡単に実装することができる。

channelのバッファ

ch := make(chan int, 3)

のようにmakeの第2引数にバッファを指定することができる。バッファを指定することで、channelの読み込みが一度も行われなくても、キャパシティが3のバッファ付きchannelであればgoroutineは3回まで書き込みが可能になる。バッファを指定しない場合は1回書き込むとブロックされる。

送信・受信専用channel

channelを宣言する時に何も指定しなければ書き込みも読み込みも可能になるが、以下のように宣言することで書き込み専用・読み込み専用のchannelを作ることができる。

• chan<- int: 読み込み専用
• <-chan int: 書き込み専用

この文法は、関数の引数にchannelを受け取る場合や戻り値としてchannelを返す場合に「これは書き込みしかできない」「これは読み込みしかできない」ということを明示できるため有用である。

channelのclose

channelは使い終わったらcloseすることができる。closeされたchannelからはデータを読み込むことはできない。例えば以下のプログラムは

• goroutineでchに書き込み、最後にchannelをcloseしている
• mainの中のforループでチャネルchからデータを読み込んでいる

というものであるが、closeされているのでその段階でchannelからの読み込みはブロックされずにforループが終了する。

package main

import "fmt"

func main() {
    ch := make(chan int)
    go func() {
        defer close(ch)
        for i := 1; i <= 5; i++ {
            ch <- i
        }
    }()

    for value := range ch {
        fmt.Printf("%v ", value)
    }
}

実践的な例

最後に実践的なchannelの使い方を載せておく。上で紹介した以下のchannelの特徴を使っている。

• バッファ
• 送信・受信専用channel
• close

このプログラムが行っていることは、URLをchannelで送って、http.Getした結果をchannelで返すというものである。main関数の中で以下のgoroutineを立ち上げて、最後にchannel resultsを読み込んで標準出力にurl, status, errを出力している。

• httpGetを実行する(channel urlsを読み込んで、resultsに書き込む)
• channel urlsに書き込む

package main

import (
    "fmt"
    "net/http"
)

type result struct {
    url    string
    status int
    err    error
}

func main() {
    urls := make(chan string, 3)
    results := make(chan result, 3)

    // Consumer
    go httpGet(urls, results)

    // Producer
    go func() {
        targetURLs := []string{
            "https://journal.lampetty.net/entry/review-2019-07",
            "https://journal.lampetty.net/entry/review-2019-06",
            "https://journal.lampetty.net/entry/review-2019-05",
            "https://journal.lampetty.net/entry/review-2019-04",
            "https://journal.lampetty.net/entry/review-2019-03",
            "https://journal.lampetty.net/entry/review-2019-02",
            "https://journal.lampetty.net/entry/review-2019-01",
        }
        for _, url := range targetURLs {
            urls <- url
        }
        close(urls)
    }()

    for r := range results {
        fmt.Printf("url = %v, status = %v, err = %v\n", r.url, r.status, r.err)
    }
}

func httpGet(urls <-chan string, results chan<- result) {
    for url := range urls {
        resp, err := http.Get(url)
        if err != nil {
            results <- result{
                url: url,
                err: err,
            }
            return
        }
        results <- result{
            url:    url,
            status: resp.StatusCode,
        }
        _ = resp.Body.Close()
    }
    close(results)
}

まとめ

Goの並行処理の基礎となっているchannelについて説明した。次のパートでは、応用編としてgoroutineとchannelを使った並行処理の実践的なパターンを紹介したいと思う。

Go言語による並行処理

Goにおける並行処理 - goroutine編

はじめに

Goでは、goroutineとchannelが言語仕様として組み込まれているため、他の言語に比べてとても並行処理のコードが書きやすい。この2つの基本的な動作原理についてまとめた自分用のメモである。(channelについては別の記事で書く予定)

goroutineとは？

• goroutine(ゴールーチン)とは、他のコードに対して並行に実行している関数のこと
• goroutineはgoというキーワードを関数呼び出しの前に置くことで起動できる。簡単!
• あらゆるGoのプログラムでは、main関数を動かすためのgoroutineが必ず存在する。つまりGoを支えるとてもプリミティブなものである。

goキーワードを使ってgoroutineを動かすとても簡単な例としては以下になる。

package main

import "fmt"

func main() {
    
go sayHello() 
   // 他の処理を続ける
} 

func sayHello() {
    fmt.Println("hello") 
}

このコードは以下の2つのgoroutineで動作している。

1. main関数を動かしているgoroutine
2. main関数から分岐した、sayHello関数を動かしているgoroutine

goroutineの正体

ではgoroutineとは何なのだろうか？Javaなどの他の言語をやってきた人からすると、これはスレッドとして実行されているのか？と気になるところだが、実態はスレッドではない。Go言語による並行処理 P.38では以下のように説明されている。

ゴルーチンは(他の言語にも似た並行処理のプリミティブは存在しますが)Go 特有のものです。ゴルーチンはOSスレッドではなく、また必ずしもグリーンスレッド――言語のランタイムにより管理されるスレッド――ではありません。ゴルーチンはコルーチン(coroutine)として知られる高水準の抽象化です。コルーチンは単に「プリエンプティブでない」並行処理のサブルーチン(Goでは関数、クロージャー、メソッドに相応)です。つまり、割り込みをされることがないということです。かわりに、コルーチンには一時停止や再エントリーを許す複数のポイントがあります。

ゴルーチンが独特なのは、ゴルーチンが Go のランタイムと密結合していることです。ゴルーチンは 一時停止や再エントリーのポイントを定義していません。Go のランタイムはゴルーチンの実行時の振る舞いを観察し、ゴルーチンがブロックしたら自動的に一時停止し、ブロックが解放されたら再開します。これによってある意味ゴルーチンをプリエンプティブにしていますが、ゴルーチンがブロックしたときにしか割り込みません。このようにランタイムとゴルーチンのロジックには美しい関係性がありま す。以上のことから、ゴルーチンは特殊なコルーチンと考えられます。

goroutineはGoのランタイムにより管理されるため、Goのコードから一時停止したりレジュームすることはできない(channelやcontextを使えば動作を止めるようにキャンセルすることは可能)。これがコルーチンとは違う点だと思う。また、goroutineとして関数を実行してもスレッドが新しく作られるわけではないので、スレッドよりも軽量である。

Go がゴルーチンをホストする機構は、いわゆる M:N スケジューラーと呼ばれる実装になっています。 これは M個のグリーンスレッドを N個の OS スレッドに対応させるものです。ゴルーチンはグリーンスレッドにスケジュールされます。グリーンスレッドの数よりも多い数のゴルーチンがある場合には、スケジューラーはゴルーチンを利用可能なグリーンスレッドに割り振って、これらのゴルーチンがブロックした場合には他のゴルーチンを実行するようにしています。

fork-joinモデル

Goではfork-joinモデルと呼ばれる並行処理のモデルに従っている。

• fork: プログラムの特定の場所で子供の処理を分岐させて、親と平行に実行させること
• join: 分岐した時点より以降に、平行処理されている親と子が再び合流すること

図で表すと以下のようになる。

ここで先ほどのsayHelloを使ったgoroutineの例を見てみる。

func main() {
    
go sayHello() 
} 

func sayHello() {
    fmt.Println("hello") 
}

sayHelloはgoroutineとして実行された後にjoinしていないので、helloと表示されずにこのプログラムは終わってしまう。ではmainに合流させるにはどうしたらよいだろうか？

一番簡単なやり方は以下のようにsync.WaitGroupを使うことである。

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup
    wg.Add(1) // 1. 起動するgoroutineの数を伝えるため、カウンターに1を足す
    go func() {
        defer wg.Done() // 2. カウンターから1をマイナスする
        sayHello()
    }()
    wg.Wait() // 3. ここでsayHelloのgoroutineが終わるまで待つ
    fmt.Println("finished")
}

func sayHello() {
    fmt.Println("hello") 
}

https://play.golang.org/p/-uPFeh_37mB

まとめ

これがgoroutineの基本的なことの全てである。goというキーワードを使うだけで並行処理が簡単に書けるということがわかるはず。

Go言語による並行処理

2019年7月の振り返り

アウトプット

7月も小ネタだけどブログを書いた。

• xargsでファイルの内容をコマンドの引数として渡す - oinume journal
• Intellij IDEAを2019.2にアップグレードするとコピペや入力がおかしくなる問題のワークアラウンド - oinume journal

path-shrinker

path-shrinker というものを作った。ターミナルのプロンプトの /Users/go/src/github.com を /U/g/s/github.com みたいにするやつ。zshのoh-my-zshではプラグインがあるんだけど、Bashでは見当たらなかったので作ってみた。

Bashでもzshやfishみたいにターミナルのプロンプトのディレクトリをシュッと短くしたかったので作った。https://t.co/d3k5hWP6Gp

— oinume (@oinume) July 14, 2019

Cloud Next'19

GoogleのCloud Nextに参加した。AWSも含めてこういうクラウド系のイベントには参加したことがなかったのだけど、GoogleのSREの人のセッションは参考・刺激になったので自分も実践していきたいと思った。AWSに比べてGCPはServerlessがかなりいけてるという印象を持っていたのだけど、それが再確認できたのも良かった。

OpenCensus

US regionから日本のサーバにアクセスしていて、どうしてもレイテンシが気になってチューニングしたいので、手始めにhttptrace.ClientTrace + OpenCensusでtracingしてみてる。connectが遅いのか、TTFBが遅いのかなどがわかるので良い。

— oinume (@oinume) July 28, 2019

lekcijeのパフォーマンスチューニングのためにOpenCensusを導入して、Stackdriver Traceでlatencyを見れるようにした。もともとボトルネックはわかっていたのだけど、httptrace.ClientTrace + OpenCensusを導入してHTTPクライアントでどこが遅いかも視覚化できたのはいい勉強になったと思う。けっきょくHTTP KeepAliveを長めに設定して少しだけパフォーマンスが良くなった。

睡眠

1日6時間以上寝る、ということを今年の目標にしている。ヘルスケアアプリによると6月は平均6時間25分だった。ただ、毎日6時間以上の睡眠ができているわけではないので、引き続きこれは課題がある。特に平日は6時間寝れている日がほとんど皆無。

英語

IELTSの攻略本を見ると、語彙力がすべてのパートにおいて重要なので、頑張って英単語を地味に覚えるようにしている。

アルゴリズム

特に進捗なし

Go, gRPC, SQLの復習

引き続き Go言語による並行処理 を読んでいる。

まとめ

7月は体調を崩してしまって会社を2日を休んでしまったのと、あまりアウトプットはできてないので頑張っていきたい。

Intellij IDEAを2019.2にアップグレードするとコピペや入力がおかしくなる問題のワークアラウンド

先日IntelliJ IDEAを2019.2にアップグレードしたところ、以下の問題に遭遇した。

• エディタでのコピー＆ペーストすると同じものが2回ペーストされる
• 文字の入力に取りこぼしがある(publicって入力してもpubcみたいなる)

バグトラッカーを見たところ、Copied text is pasted twice via Cmd+V (with Japanese input)が不具合として上がっていたが修正されそうな気配がなくて途方にくれていたところ、こんなワークアラウンドがあることをTwitter上で発見したので試してみた。

JBR11というJava Runtimeが原因とのことなので、これを古いバージョンにすればよいらしい。

自分は jbrx-8u202 をわざわざダウンロードするのが面倒だったので

$ brew cask install homebrew/cask-versions/adoptopenjdk8

してから

$ /usr/libexec/java_home -v "1.8"
/Library/Java/JavaVirtualMachines/adoptopenjdk-8.jdk/Contents/Home

で出力されたディレクトリをSwitch Boot JDKで指定したところ問題が起きなくなった。めでたしめでたし。

IntelliJ IDEAハンズオン――基本操作からプロジェクト管理までマスター

xargsでファイルの内容をコマンドの引数として渡す

dates.txt

2019-01-01
2019-02-01
2019-03-01

みたいなファイルがあった時に、このファイルの行をxargsコマンド経由で任意のコマンドの引数として渡す方法。

$ cat dates.txt | xargs -I{} 'echo {}'