Goroutine とは？

Go が抽象化した軽量スレッドであり、アプリケーションレベルでスケジューリングを行うことで、簡単に並行処理を行うことができます。

goキーワードを使用することで起動できます。コンパイラは cmd/compile/internal/gc.state.stmt と cmd/compile/internal/gc.state.call という 2 つのメソッドを通して、このキーワードを runtime.newproc 関数の呼び出しに変換します。

新しい Goroutine を起動してタスクを実行する際は、runtime.newproc を通して g を初期化し、コルーチンを実行します。

Goroutine はどれくらいのリソースを消費するか？

メモリの消費

コルーチンを起動してブロックさせることで、前後のメモリ使用量の変化を観察します。

func getGoroutineMemConsume() {
  var c chan int
  var wg sync.WaitGroup
  const goroutineNum = 1000000

  memConsumed := func() uint64 {
    runtime.GC() // GCを実行してオブジェクトの影響を除外
    var memStat runtime.MemStats
    runtime.ReadMemStats(&memStat)
    return memStat.Sys
  }

  noop := func() {
    wg.Done()
    <-c // Goroutineが終了しないようにし、メモリが解放されないようにする
  }

  wg.Add(goroutineNum)
  before := memConsumed() // Goroutine作成前のメモリを取得
  for i := 0; i < goroutineNum; i++ {
    go noop()
  }
  wg.Wait()
  after := memConsumed() // Goroutine作成後のメモリを取得
  fmt.Println(runtime.NumGoroutine())
  fmt.Printf("%.3f KB bytes\n", float64(after-before)/goroutineNum/1024)
}

結果の分析：

各コルーチンは少なくとも 2KB の空間を消費します。つまり、もしマシンのメモリが 2GB であれば、最大で 2GB / 2KB = 100 万個のコルーチンを同時に存在させることが可能です。

CPU の消費

1 つの Goroutine がどれくらい CPU を消費するかは、実行される関数のロジックに大きく依存します。もしその関数が CPU 集約型の計算で長時間実行されるようなものであれば、この時点で CPU がボトルネックになります。

どれくらいの数の Goroutine を同時に実行できるかは、実際にはプログラム内で何をしているかによります。もし大量のメモリを消費するネットワーク操作などであれば、数個の Goroutine でプログラムがクラッシュする可能性もあります。

結論

実際に起動できる Goroutine の数は、処理内容によって消費される CPU やメモリに依存します。
もし処理内容が何も行わないような空の操作であれば、理論上はメモリが最初にボトルネックになります。この場合、2GB のメモリが使い切られると、プログラムにエラーが発生します。
逆に CPU 集約型の処理であれば、2〜3 個の Goroutine だけでもプログラムが異常をきたす可能性があります。

Goroutine が多すぎると発生する一般的な問題

too many open files：これは開かれているファイルやソケットが多すぎることによる問題です。
out of memory：メモリ不足によるクラッシュです。

実際の業務での応用

どのように並行数を制御するか？

runtime.NumGoroutine() を使えば、Goroutine の数を監視できます。

1. タスクを 1 つの Goroutine でのみ実行する

API などで並列実行が必要なケースでは、アプリケーション層で並行数の制御を行う必要があります。
例えば、Goroutine を使ってリソースを初期化する処理は、一度だけ実行すればよく、複数の Goroutine で同時に初期化を行う必要はありません。
このような場合には、running フラグを用いて、現在初期化中であるかどうかを判断できます。

// SingerConcurrencyRunner はタスクが一度に一つだけ実行されることを保証する
type SingerConcurrencyRunner struct {
  isRunning bool
  sync.Mutex
}

func NewSingerConcurrencyRunner() *SingerConcurrencyRunner {
  return &SingerConcurrencyRunner{}
}

func (c *SingerConcurrencyRunner) markRunning() (ok bool) {
  c.Lock()
  defer c.Unlock()
  // ダブルチェック：外部のチェックが成功した後、他のGoroutineが割り込まないようにする
  if c.isRunning {
    return false
  }
  // 実行中としてマーク
  c.isRunning = true
  return true
}

func (c *SingerConcurrencyRunner) unmarkRunning() (ok bool) {
  c.Lock()
  defer c.Unlock()
  if !c.isRunning {
    return false
  }
  // 実行中マークを解除
  c.isRunning = false
  return true
}

func (c *SingerConcurrencyRunner) Run(f func()) {
  // すでに実行中のGoroutineがある場合、実行せずに戻る。メモリ消費を抑えるため。
  if c.isRunning {
    return
  }
  if !c.markRunning() {
    // 競合に負けた場合は戻る
    return
  }

  // 実際の処理を非同期で実行
  go func() {
    defer func() {
      if err := recover(); err != nil {
        // ログなど
      }
    }()
    f()
    c.unmarkRunning()
  }()
}

信頼性テストとして、同時に 2 つ以上の Goroutine が走っていないことを確認：

func TestConcurrency(t *testing.T) {
  runner := NewConcurrencyRunner()
  for i := 0; i < 100000; i++ {
    runner.Run(f)
  }
}

func f() {
        // この中ではGoroutineの数が3を超えることはない
  if runtime.NumGoroutine() > 3 {
    fmt.Println(">3", runtime.NumGoroutine())
  }
}

2. 指定された数の Goroutine でタスクを実行

他の Goroutine は待機に入るか、適切なタイムアウトを設定し、あるいは古いデータで処理を返すことも可能です。
Tunny を使用すると、Goroutine 数の制御が可能になります。
もし Worker がすべて使用中であれば、WorkRequest を受け取って処理することはなく、代わりに reqChan に書き込まれて待機状態になります。

func (w *workerWrapper) run() {
//...
        for {
    // NOTE: ここでブロックすると、ワーカーのシャットダウンを防ぐことになる。
    w.worker.BlockUntilReady()
    select {
    case w.reqChan <- workRequest{
      jobChan:       jobChan,
      retChan:       retChan,
      interruptFunc: w.interrupt,
    }:
      select {

      case payload := <-jobChan:
        result := w.worker.Process(payload)
        select {
        case retChan <- result:
        case <-w.interruptChan:
          w.interruptChan = make(chan struct{})
        }
//...
    }
  }
//...
}

ここでの実装方法は、常駐する Goroutine を使って制御しています。
Size を変更すると、新たな Worker が追加されて処理を実行します。
もう一つの実装方法としては、chan を使って Goroutine を起動するかどうかを制御する方式があります。
もしバッファがすでに満杯であれば、新しい Goroutine は起動できません。

type ProcessFunc func(ctx context.Context, param interface{})

type MultiConcurrency struct {
  ch chan struct{}
  f  ProcessFunc
}

func NewMultiConcurrency(size int, f ProcessFunc) *MultiConcurrency {
  return &MultiConcurrency{
    ch: make(chan struct{}, size),
    f:  f,
  }
}

func (m *MultiConcurrency) Run(ctx context.Context, param interface{}) {
  // バッファが満杯なら処理を開始しない
  m.ch <- struct{}{}
  go func() {
    defer func() {
      // バッファを解放
      <-m.ch
      if err := recover(); err != nil {
        fmt.Println(err)
      }
    }()
    m.f(ctx, param)
  }()
}

テストコードでは、Goroutine の数が 13 を超えないことを確認します：

func mockFunc(ctx context.Context, param interface{}) {
  fmt.Println(param)
}

func TestNewMultiConcurrency_Run(t *testing.T) {
  concurrency := NewMultiConcurrency(10, mockFunc)
  for i := 0; i < 1000; i++ {
    concurrency.Run(context.Background(), i)
    if runtime.NumGoroutine() > 13 {
      fmt.Println("goroutine", runtime.NumGoroutine())
    }
  }
}

このような方法を使えば、実行中の Goroutine を大量に常駐させる必要がなくなります。
ただし、仮に 100 個の Goroutine が常駐していたとしても、それによるメモリ消費は
2KB × 100 = 200KB に過ぎず、基本的には無視できるレベルです。

私たちはLeapcell、Rustプロジェクトのホスティングの最適解です。

Leapcellは、Webホスティング、非同期タスク、Redis向けの次世代サーバーレスプラットフォームです：

複数言語サポート

Node.js、Python、Go、Rustで開発できます。

無制限のプロジェクトデプロイ

使用量に応じて料金を支払い、リクエストがなければ料金は発生しません。

比類のないコスト効率

使用量に応じた支払い、アイドル時間は課金されません。
例: $25で6.94Mリクエスト、平均応答時間60ms。

洗練された開発者体験

直感的なUIで簡単に設定できます。
完全自動化されたCI/CDパイプラインとGitOps統合。
実行可能なインサイトのためのリアルタイムのメトリクスとログ。

簡単なスケーラビリティと高パフォーマンス

高い同時実行性を容易に処理するためのオートスケーリング。
ゼロ運用オーバーヘッド — 構築に集中できます。

ドキュメントで詳細を確認！

Xでフォローする：@LeapcellHQ

ブログでこの記事を読む