Go言語のChannel：完全解説

Posted at 2025-01-16

Channel：Go言語における重要な機能とGo言語CSP並行モデルの重要な体現

ChannelはGo言語における非常に重要な機能であり、またGo言語のCSP並行モデルの重要な体現でもあります。簡単に言えば、goroutine間の通信はChannelを通じて行うことができます。

ChannelはGo言語において非常に重要であり、コード内で頻繁に使用されるため、その内部実装について興味を持つのも当然です。この記事では、Go 1.13のソースコードに基づいてChannelの内部実装原理を分析します。

Channelの基本的な使い方

正式にChannelの実装を分析する前に、まずChannelの最も基本的な使い方を見てみましょう。コードは以下の通りです：

package main
import "fmt"

func main() {
    c := make(chan int)

    go func() {
        c <- 1 // Channelに送信
    }()

    x := <-c // Channelから受信

    fmt.Println(x)
}

上記のコードでは、make(chan int)を使用してint型のChannelを作成しています。
あるgoroutineでは、c <- 1を使用してChannelにデータを送信しています。メインのgoroutineでは、x := <- cを通じてChannelからデータを読み取り、それをxに割り当てています。

上記のコードは、Channelの2つの基本的な操作に対応しています：

send操作c <- 1は、Channelにデータを送信することを意味します。
recv操作x := <- cは、Channelからデータを受信することを意味します。

また、Channelはバッファ付きChannelとバッファなしChannelに分けられます。上記のコードでは、バッファなしChannelを使用しています。バッファなしChannelの場合、現在他のgoroutineがChannelからデータを受信していない場合、送信者は送信文でブロックされます。

Channelを初期化するときに、バッファサイズを指定することができます。例えば、make(chan int, 2)はバッファサイズを2に指定します。バッファが満杯になる前は、送信者はブロックされることなくChannelにデータを送信でき、受信者が準備できるのを待つ必要はありません。ただし、バッファが満杯の場合、送信者は依然としてブロックされます。

Channelの基本的な使い方

Channelのソースコードを調べる前に、まずGo言語におけるChannelの具体的な実装がどこにあるのかを見つける必要があります。Channelを使用するときには<-記号を使用するのですが、Goのソースコードでその実装を直接見つけることはできません。しかし、Go言語のコンパイラは間違いなく<-記号を対応する基本的な実装に変換します。

我々はGoの組み込みコマンド：go tool compile -N -l -S hello.goを使用して、コードを対応するアセンブリ命令に変換することができます。

あるいは、オンラインツールのCompiler Explorerを直接使用することもできます。上記の例のコードの場合、このリンクで直接そのアセンブリ結果を表示することができます：go.godbolt.org/z/3xw5Cj。以下の図のように：

Channelのアセンブリ命令

上記の例のコードに対応するアセンブリ命令を注意深く調べると、以下の対応関係がわかります：

Channelの構築文make(chan int)は、runtime.makechan関数に対応しています。
送信文c <- 1は、runtime.chansend1関数に対応しています。
受信文x := <- cは、runtime.chanrecv1関数に対応しています。

上記の関数の実装はすべて、Goのソースコードのruntime/chan.goコードファイルにあります。次に、これらの関数を対象にしてChannelの実装を調べていきましょう。

Channelの構築

Channelの構築文make(chan int)は、Go言語のコンパイラによってruntime.makechan関数に変換され、その関数シグネチャは以下の通りです：

func makechan(t *chantype, size int) *hchan

ここで、t *chantypeはChannelを構築するときに渡される要素の型です。size intはユーザーが指定するChannelのバッファサイズで、指定されない場合は0です。この関数の戻り値は*hchanです。hchanはGo言語におけるChannelの内部実装です。その定義は以下の通りです：

type hchan struct {
        qcount   uint           // すでにバッファに配置された要素の数
        dataqsiz uint           // ユーザーがChannelを構築するときに指定するbufサイズ
        buf      unsafe.Pointer // バッファ
        elemsize uint16         // バッファ内の各要素のサイズ
        closed   uint32         // Channelが閉じられているかどうか、== 0は閉じられていないことを意味する
        elemtype *_type         // Channel要素の型情報
        sendx    uint           // バッファの送信インデックスにおける送信された要素のインデックス位置
        recvx    uint           // バッファの受信インデックスにおける受信された要素のインデックス位置
        recvq    waitq          // 受信待ちのgoroutineのリスト 受信待ちゴルーチン
        sendq    waitq          // 送信待ちのgoroutineのリスト 送信待ちゴルーチン

        lock mutex
}

hchan内のすべての属性は大まかに3つのカテゴリーに分けることができます：

バッファ関連の属性：例えばbuf、dataqsiz、qcountなど。Channelのバッファサイズが0でない場合、バッファは受信待ちのデータを格納します。リングバッファを使用して実装されています。
waitq関連の属性：これは標準的なFIFOキューとして理解できます。その中で、recvqはデータを受信待ちのgoroutineを含み、sendqはデータを送信待ちのgoroutineを含みます。waitqは二重リンクリストを使用して実装されています。
その他の属性：例えばlock、elemtype、closedなど。

makechanの全体的なプロセスは基本的にいくつかの妥当性チェックと、バッファ、hchanその他の属性に対するメモリ割り当てであり、ここでは深く議論しません。興味のある方は直接ソースコードを参照してください。

hchanの属性を簡単に分析することで、2つの重要なコンポーネント、バッファとwaitqがあることがわかります。hchanのすべての動作と実装はこれら2つのコンポーネントを中心に展開されます。

Channelへのデータ送信

Channelの送信と受信のプロセスは非常に似ています。まずChannelの送信プロセス（例えばc <- 1）を分析してみましょう。これはruntime.chansend関数の実装に対応しています。

Channelにデータを送信しようとするとき、recvqキューが空でない場合、まずrecvqの先頭からデータを受信待ちのgoroutineを1つ取り出します。そして、データは直接このgoroutineに送信されます。コードは以下の通りです：

if sg := c.recvq.dequeue(); sg!= nil {
        send(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true
}

recvqはデータを受信待ちのgoroutineを含みます。あるgoroutineがrecv操作（例えばx := <- c）を使用するとき、このときChannelのキャッシュにデータがなく、他のgoroutineが送信待ちでもない場合（つまりsendqが空の場合）、このgoroutineと受信するデータのアドレスがsudogオブジェクトにパッケージ化され、recvqに入れられます。

上記のコードを続けますが、このときrecvqが空でない場合、send関数が呼び出され、データが対応するgoroutineのスタックにコピーされます。

send関数の実装は主に2点です：

memmove(dst, src, t.size)はデータ転送を行い、基本的にはメモリコピーです。
goready(gp, skip+1) goreadyの機能は対応するgoroutineを起動することです。

recvqキューが空の場合、つまりこのときデータを受信待ちのgoroutineがいない場合、Channelはデータをバッファに入れようと試みます。コードは以下の通りです：

if c.qcount < c.dataqsiz {
        // 等価于 c.buf[c.sendx]
        qp := chanbuf(c, c.sendx)
        // データをバッファにコピー
        typedmemmove(c.elemtype, qp, ep)
        c.sendx++
        if c.sendx == c.dataqsiz {
                c.sendx = 0
        }
        c.qcount++
        unlock(&c.lock)
        return true
}

上記のコードの機能は実際とても単純で、つまりデータをバッファに入れるだけです。このプロセスはリングバッファの操作を伴い、dataqsizはユーザーが指定するChannelのバッファサイズを表し、指定されない場合はデフォルトで0になります。他の具体的な詳細な操作は後のリングバッファのセクションで詳しく説明します。

ユーザーがバッファなしChannelを使用している場合、またはこのときバッファが満杯の場合、条件c.qcount < c.dataqsizは満たされず、上記のプロセスは実行されません。このとき、現在のgoroutineと送信するデータがsendqキューに入れられ、同時にこのgoroutineは切り替えられます。全体のプロセスは以下のコードに対応しています：

gp := getg()
mysg := acquireSudog()
mysg.releasetime = 0
if t0!= 0 {
        mysg.releasetime = -1
}
mysg.elem = ep
mysg.waitlink = nil
mysg.g = gp
mysg.isSelect = false
mysg.c = c
gp.waiting = mysg
gp.param = nil
c.sendq.enqueue(mysg)
// goroutineを待機状態に切り替えてロック解除
goparkunlock(&c.lock, waitReasonChanSend, traceEvGoBlockSend, 3)

上記のコードでは、goparkunlockは入力されたミューテックスをロック解除し、このgoroutineを切り出し、このgoroutineを待機状態に設定します。上記のgoparkとgoreadyは互いに対応し、逆の操作です。goparkとgoreadyはruntimeのソースコードでよく出会うもので、goroutineのスケジューリングプロセスに関係していますが、ここでは深く議論しません。後で別の記事で書こうと思います。

goparkを呼び出した後、ユーザーの目線では、Channelにデータを送信するコード文がブロックされることになります。

上記のプロセスはChannelの送信文（例えばc <- 1）の内部ワークフローであり、全体の送信プロセスはc.lockを使用してロックを行い、並行セキュリティを保証しています。

簡単に言えば、全体のプロセスは以下の通りです：

recvqが空かどうかをチェックします。空でない場合、recvqの先頭からgoroutineを1つ取り出し、それにデータを送信し、対応するgoroutineを起動します。
recvqが空の場合、データをバッファに入れます。
バッファが満杯の場合、送信するデータと現在のgoroutineをsudogオブ体にパッケージ化してsendqに入れます。そして現在のgoroutineを待機状態に設定します。

チャネルからのデータ受信プロセス

チャネルからデータを受信するプロセスは、基本的に送信プロセスと似ており、ここでは繰り返しません。受信プロセスに関わる特定のバッファ関連の操作については、後で詳しく説明します。

ここで留意すべきは、チャネルの送信と受信の全体プロセスではruntime.mutexを使用してロックを行っていることです。runtime.mutexは、ランタイム関連のソースコードで一般的に使用される軽量のロックです。全体のプロセスは最も効率的な無ロックアプローチではありません。Go言語にはgo/issues#8899という問題があり、これは無ロックチャネルのソリューションを示しています。

チャネルのリングバッファの実装

チャネルは書き込まれたデータをキャッシュするためにリングバッファを使用しています。リングバッファには多くの利点があり、固定長のFIFOキューの実装に非常に適しています。

チャネルにおけるリングバッファの実装は以下の通りです：

チャネルにおけるリングバッファの実装

hchanにはバッファに関連する2つの変数があります：recvxとsendx。その中で、sendxはバッファ内の書き込み可能なインデックスを表し、recvxはバッファ内の読み取り可能なインデックスを表します。recvxとsendxの間の要素は、正常にバッファに配置されたデータを表します。

我々は直接buf[recvx]を使ってキューの最初の要素を読み取り、buf[sendx] = xを使ってキューの末尾に要素を配置することができます。

バッファの書き込み

バッファが満杯でない場合、バッファにデータを入れる操作は以下の通りです：

qp := chanbuf(c, c.sendx)
// データをバッファにコピー
typedmemmove(c.elemtype, qp, ep)
c.sendx++
if c.sendx == c.dataqsiz {
    c.sendx = 0
}
c.qcount++

ここで、chanbuf(c, c.sendx)はc.buf[c.sendx]と等価です。上記のプロセスは非常に単純で、つまりデータをバッファのsendxの位置にコピーするだけです。

そして、sendxを次の位置に移動します。sendxが最後の位置に達した場合、0に設定します。これは典型的な先頭と末尾をつなぐ方法です。

バッファの読み取り

バッファが満杯でない場合、このときsendqも必ず空でなければなりません（なぜなら、バッファが満杯でない場合、データを送信するgoroutineはキューに入らず、直接バッファにデータを入れるからです。具体的なロジックは、上記のチャネルへのデータ送信のセクションを参照してください）。このとき、チャネルの読み取りプロセスchanrecvは比較的単純で、データを直接バッファから読み取ることができます。これもrecvxを移動するプロセスです。基本的には上記のバッファの書き込みと同じです。

sendqに待機中のgoroutineがある場合、このときバッファは必ず満杯です。このときのチャネルの読み取りロジックは以下の通りです：

// 等価于 c.buf[c.recvx]
qp := chanbuf(c, c.recvx)
// キューから受信者にデータをコピー
if ep!= nil {
    typedmemmove(c.elemtype, ep, qp)
}
// 送信者からキューにデータをコピー
typedmemmove(c.elemtype, qp, sg.elem)
c.recvx++
if c.recvx == c.dataqsiz {
    c.recvx = 0
}
c.sendx = c.recvx // c.sendx = (c.sendx + 1) % c.dataqsiz

上記のコードで、epはデータを受け取る変数に対応するアドレスです。例えば、x := <- cでは、epは変数xのアドレスを表します。
そしてsgはsendqから取り出された最初のsudogを表します。そして：

typedmemmove(c.elemtype, ep, qp)は、バッファ内の現在読み取り可能な要素を受信変数のアドレスにコピーすることを意味します。
typedmemmove(c.elemtype, qp, sg.elem)は、sendq内のgoroutineが送信待ちのデータをバッファにコピーすることを意味します。後でrecv++が実行されるので、これはsendq内のデータをキューの末尾に配置することと等価です。

簡単に言えば、ここではチャネルがバッファ内の最初のデータを対応する受信変数にコピーし、同時にsendq内の要素をキューの末尾にコピーすることで、データをFIFO（First In First Out）で処理することができます。