GoでGoroutine Poolを実装する方法

Posted at 2025-02-18

Leapcell: The Next - Gen Serverless Platform for Golang app Hosting

0. はじめに

以前、GoのネイティブHTTPサーバがクライアント接続を処理する際に、各接続に対してgoroutineを生成するという、かなり無理やりなアプローチをとることが述べられました。より深く理解するために、Goのソースコードを見てみましょう。まず、以下のように最もシンプルなHTTPサーバを定義します。

func myHandler(w http.ResponseWriter, r *http.Request) {
    fmt.Fprintf(w, "Hello there!\n")
}

func main() {
    http.HandleFunc("/", myHandler)     // アクセスルートを設定
    log.Fatal(http.ListenAndServe(":8080", nil))
}

エントリーポイントのhttp.ListenAndServe関数を追いかけます。

// file: net/http/server.go
func ListenAndServe(addr string, handler Handler) error {
    server := &Server{Addr: addr, Handler: handler}
    return server.ListenAndServe()
}

func (srv *Server) ListenAndServe() error {
    addr := srv.Addr
    if addr == "" {
        addr = ":http"
    }
    ln, err := net.Listen("tcp", addr)
    if err!= nil {
        return err
    }
    return srv.Serve(tcpKeepAliveListener{ln.(*net.TCPListener)})        
}

func (srv *Server) Serve(l net.Listener) error {
    defer l.Close()
  ...
    for {
        rw, e := l.Accept()
        if e!= nil {
            // エラーハンドリング
            return e
        }
        tempDelay = 0
        c, err := srv.newConn(rw)
        if err!= nil {
            continue
        }
        c.setState(c.rwc, StateNew) // Serveが戻る前
        go c.serve()
    }
}

まず、net.Listenはネットワークポートを監視する役割を担っています。rw, e := l.Accept()はネットワークポートからTCP接続を取得し、go c.server()は各TCP接続に対してgoroutineを生成して処理します。また、fasthttpネットワークフレームワークはネイティブのnet/httpフレームワークよりもパフォーマンスが良いことも述べましたが、その理由の1つがgoroutineプールの使用です。では、自分たちでgoroutineプールを実装する場合はどうすればいいでしょうか？最もシンプルな実装から始めましょう。

1. 弱いバージョン

Goでは、goキーワードを使ってgoroutineを起動します。goroutineのリソースは一時的なオブジェクトプールとは異なり、戻して再取得することはできません。したがって、goroutineは継続的に実行されるべきです。必要なときに実行し、必要ないときにはブロックします。これは他のgoroutineのスケジューリングにほとんど影響を与えません。そして、goroutineのタスクはチャネルを通して渡すことができます。以下はシンプルな弱いバージョンです。

func Gopool() {
    start := time.Now()
    wg := new(sync.WaitGroup)
    data := make(chan int, 100)

    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            for _ = range data {
                fmt.Println("goroutine:", n, i)
            }
        }(i)
    }

    for i := 0; i < 10000; i++ {
        data <- i
    }
    close(data)
    wg.Wait()
    end := time.Now()
    fmt.Println(end.Sub(start))
}

上記のコードでは、プログラムの実行時間も計算しています。比較のために、プールを使用しないバージョンを以下に示します。

func Nopool() {
    start := time.Now()
    wg := new(sync.WaitGroup)

    for i := 0; i < 10000; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            //fmt.Println("goroutine", n)
        }(i)
    }
    wg.Wait()

    end := time.Now()
    fmt.Println(end.Sub(start))
}

最後に、実行時間を比較すると、goroutineプールを使用したコードはプールを使用しないコードの約2/3の時間で実行されます。もちろん、このテストはまだやや荒いものです。次に、reflectの記事で紹介されたGoのベンチマークテスト方法を使ってテストします。テストコードは以下の通りです（多くの関係のないコードを削除しています）。

package pool

import (
    "sync"
    "testing"
)

func Gopool() {
    wg := new(sync.WaitGroup)
    data := make(chan int, 100)

    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            for _ = range data {
            }
        }(i)
    }

    for i := 0; i < 10000; i++ {
        data <- i
    }
    close(data)
    wg.Wait()
}

func Nopool() {
    wg := new(sync.WaitGroup)

    for i := 0; i < 10000; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
        }(i)
    }
    wg.Wait()
}

func BenchmarkGopool(b *testing.B) {
    for i := 0; i < b.N; i++ {
        Gopool()
    }
}

func BenchmarkNopool(b *testing.B) {
    for i := 0; i < b.N; i++ {
        Nopool()
    }
}

最終的なテスト結果は以下の通りです。goroutineプールを使用したコードの方が実行時間が短いことが確認できます。

$ go test -bench='.' gopool_test.go
BenchmarkGopool-8            500       2696750 ns/op
BenchmarkNopool-8            500       3204035 ns/op
PASS

2. アップグレードバージョン

良いスレッドプールには、しばしばより多くの要件があります。最も切迫したニーズの1つは、goroutineが実行する関数をカスタマイズできることです。関数とは、関数アドレスと関数パラメータのことです。渡す関数の形式が異なる場合（異なるパラメータや戻り値）はどうすればいいでしょうか？比較的簡単な方法は、リフレクションを導入することです。

type worker struct {
    Func interface{}
    Args []reflect.Value
}

func main() {
    var wg sync.WaitGroup

    channels := make(chan worker, 10)
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for ch := range channels {
                reflect.ValueOf(ch.Func).Call(ch.Args)
            }
        }()
    }

    for i := 0; i < 100; i++ {
        wk := worker{
            Func: func(x, y int) {
                fmt.Println(x + y)
            },
            Args: []reflect.Value{reflect.ValueOf(i), reflect.ValueOf(i)},
        }
        channels <- wk
    }
    close(channels)
    wg.Wait()
}

しかし、リフレクションを導入すると、パフォーマンスの問題も発生します。goroutineプールはもともとパフォーマンスの問題を解決するために設計されたものですが、今度は新しいパフォーマンスの問題が発生してしまいました。では、どうすればいいでしょうか？クロージャを使います。

type worker struct {
    Func func()
}

func main() {
    var wg sync.WaitGroup

    channels := make(chan worker, 10)

    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for ch := range channels {
                //reflect.ValueOf(ch.Func).Call(ch.Args)
                ch.Func()
            }
        }()
    }

    for i := 0; i < 100; i++ {
        j := i
        wk := worker{
            Func: func() {
                fmt.Println(j + j)
            },
        }
        channels <- wk
    }
    close(channels)
    wg.Wait()
}

Goでは、クロージャを適切に使わないと簡単に問題が発生することに注意する必要があります。クロージャを理解する上での重要なポイントは、オブジェクトの参照であり、コピーではないということです。これはgoroutineプールの実装の簡略版にすぎません。実際に実装する際には、多くの詳細を考慮する必要があります。たとえば、プールを停止するための停止チャネルを設定するなどです。しかし、goroutineプールの核心はここにあります。

3. goroutineプールとCPUコアの関係

では、goroutineプール内のgoroutineの数とCPUコアの数には関係があるのでしょうか？これは実際にはケースごとに議論する必要があります。

1. goroutineプールが十分に活用されていない場合

これは、channel dataにデータがあるとすぐに、それがgoroutineによって取り出されることを意味します。この場合、もちろん、CPUがスケジューリングできる限り、つまり、プール内のgoroutineの数とCPUコアの数が最適です。テストによってこれが確認されています。

2. `channel data`内のデータがブロックされている場合

これは、goroutineが不十分であることを意味します。goroutineの実行タスクがCPU集中的でなく（ほとんどの場合そうではない）、I/Oによってのみブロックされる場合、一般的に、ある範囲内ではgoroutineが多ければ多いほど良いです。もちろん、具体的な範囲は具体的な状況に応じて分析する必要があります。