はじめに

golangにはgoroutineというGoランタイムレベルでの軽量スレッドが実装されています。 Goランタイムは、以下の二つをgo文の記述によって簡単に実現します。

• 優秀なスケジューリングにより1つのマシンスレッド上でgoroutineを高速に切り替えて実行する 並行処理
• 複数マシンスレッド上でgoroutineを実行する 並列処理

go文は、非同期処理的な側面で使われることが多いかと思います。

同期的に処理が進行しているmain goroutineから、go文の宣言によって一部の処理が別goroutineに移譲され、main goroutineはそのgoroutineの完了を待たずに進行します。

処理を別goroutineに移し、main goroutineは進行する

とても便利なgo文ですが、使い方を誤るとgoroutine leakを誘発してしまう可能性があります。

A leaked goroutine is a goroutine blocked on some concurrency primitive (channels, sync.Mutex, sync.Cond, etc) that cannot possibly become unblocked.

https://go.dev/doc/go1.26 より

goroutine leakとはつまり、goroutineが終了できずに残り続け、メモリを占有し続ける状態を指します。 goroutineはプログラムが終了するまで解放されないため、じわじわとこれが問題になり、OOMの原因になり得ます。

個人的な考えですが、goroutine leakを起こしてしまうような書き方をしてしまう理由の一つとして

• 実行するgoroutineのライフタイムを開発者が認知できていない

というのが大きいのではないかなと思っていました。

そこで、このような問題への解決策として structured concurrency(構造的並行処理) という概念があることを知りました。

この記事では、構造的並行処理についてと、これをGoで表現するために筆者が作成したscopeというpackageについて紹介します。

構造的並行処理について

構造化プログラミングの文脈から、いかにして構造的並行処理という概念の必要性が謳われるようになったのかを簡単に説明していきます。

並行処理は構造化プログラミングを破っている？

構造的並行処理は、「構造化プログラミング(structured programming)」の概念を並行処理にも適用させたものになっています。

「構造化プログラミング」は、決められた制御構造の組み合わせだけでプログラムを組み立てる手法です。

制御構造というと、逐次的な実行である順次(sequential)、if-elseによる選択(selection)、whileやforによる繰り返し(repetition)の基本的な３つの要素でプログラムの流れを明確にしようというものですね。

上の3つの要素は、すべて「入口が1つ、出口が1つ」のブロックです。ブロックを抜けたということは、ブロック内の処理は完了しているということになります。

これにより、プログラムの処理の流れが上→下に固定されます。つまり、とても読みやすくなるんですね。

しかし、並行処理/非同期処理の呼び出しは「構造化プログラミング」による「制御フローの抽象化」を破壊しているのではないかというのです。

並行処理は上から下に流れる制御フローから外れている

以下のコードを見てください。hoge関数を実行した後、fuga()関数をgo文で実行しています。

func foo() {
  hoge()
  go fuga()  // 制御フローが横に飛び出す
  // ...
  return     // fuga はまだ生きているかもしれない
}

この場合、foo()がreturnして終了したとしても、並行処理としているfuga()が完了したかどうかわかりません。もしかしたら終了しているかもしれないし、まだ処理中かもしれません。

foo()の呼び出しもとは、foo()が完了した時点でfoo()ブロック内の処理は完了していると考え、後続の処理を続けるでしょう。しかし、goによる並行タスク実行によってそれが保証されない。関数呼び出しという一種の「抽象化」が破られています。これは、goroutineのメモリリーク、共有リソース操作の競合、goroutine内のエラーの握り潰しなど、バグの温床になります。

これを解決するための一種の並行処理/非同期処理のパラダイムとして提唱されているのが 構造的並行処理(structured concurrency) です。

構造的並行処理

構造的並行処理は以下3つのルールに集約されると言えます。

• 子スレッドのライフタイムは親スレッドのライフタイムに束縛される
• 子→親スレッドへエラーが伝播される
• タスクグループ内においてキャンセルが伝播される

子スレッドのライフタイムは親スレッドのライフタイムに束縛される

構造化プログラミングが目指すところである、「処理の流れを上から下に読めるようにする」ために重要なことは、処理の始点と終点を明確にすることにあると思います。ifやwhileは、ブロックの開き括弧が始点、閉じ括弧が終点です。

構造的並行処理は、「並行タスクのライフサイクルを１つのブロックスコープ内に終始させる」ことでこれを実現します。このブロックスコープは、タスクグループとも呼ばれるようです。

図解で以下のように示せます。子スレッド（= 並行タスク）のライフタイムは、親スレッドによって明確に管理されます。

タスクグループを抜けた時点で子スレッドの完了が保証される

並行処理として呼び出した子スレッドは、タスクグループのブロックスコープ内で実行され、終了します。よって、その子スレッドの始点と終点が明確になっているんですね。

これにより構造化プログラミングの規約が満たされ、結果的に並行タスクのライフタイムが明確になります。

子→親スレッドへエラーが伝播される

普通、関数呼び出しはコールスタックという形式でツリー上につながっています。しかし、並行処理が子スレッドとして起動されると、そのコールスタックからは外れてしまいます。

以下のプログラムを実行してみます。

func main() {
  hoge()
  fmt.Println("main completed")
}

func hoge() {
  go panicTask()
  time.Sleep(time.Second)
}

func panicTask() {
  panic("something went wrong")
}

出力が以下のようになりました。

❯ go run main.go
panic: something went wrong

goroutine 7 [running]:
main.panicTask()
	workspace/main.go:19 +0x2c
created by main.hoge in goroutine 1
	workspace/main.go:14 +0x24
exit status 2

「panicTask()がhoge()にて作成された」というところから始まっています。

直感的にはmain()→hoge()→panicTask()というようにコールスタックが記録されているかと想像しますが、そうではないようです。

つまり、panicTask()のpanicはgoroutine 7の中で完結しており、hoge()やmain()のスタックには届きません。hoge()は子の失敗を知る術がありません。

コールスタックから外れるということは、エラーを伝播する経路そのものが存在しないということになります。

構造的並行処理では、エラーの伝播を保証します。

親スレッド上でタスクグループ内タスクのエラーを補足できる

タスクグループ内においてキャンセルが伝播される

タスクグループ内のタスクが失敗した場合、同タスクグループ内の他タスクの実行はキャンセルされます。

まず、一般的な関数のエラーハンドリングを見てみましょう。hoge()内で呼び出しているfooがエラーを返す、もしくはpanicを引き起こした時、fuga()は実行されないように記述するのが自然です。

func hoge() error {
  if err := foo(); err!=nil{
    return err // 後続の処理は行わずreturnする。panicが起きた場合はプログラムがクラッシュするので後続の処理には進まない。
  }
  fuga()
  // ...
  return nil
}

これと考えは同じで、タスクグループ内のいずれかのタスクが異常終了した場合は、他のタスクを続行するべきではありません。

fuga()の異常終了により、hoge()の実行がキャンセルされる

このように、構造的並行処理は「並行処理を一般的な構造化プログラミングとして扱えるようにしよう」という試みであるということがわかります。

Goの言語仕様には組み込まれていない

並行処理の実行のためのAPIとしてgo文が提供されているGoですが、構造的並行処理が言語仕様に組み込まれているわけではありません。しかし他言語に目を向けてみると、いくつかの言語では対応がされていました。

Goにはそれがありませんが、過去のissueを見ると、議論がなされていたことが垣間見えます。

内容は、Goの2系から構造的並行処理を言語仕様として導入することの提案です。

結果的に以下の理由でこの提案は棄却されています。

• 既存のコードをほぼ全て壊すため採用できない
• 準標準packageであるgolang.org/x/sync/errgroupと、標準のcontext.Contextの終了シグナルにより大体は実現可能なので具体的な言語変更は不要

提案自体には否定的な姿勢を示すだけではなく、一部賛同している部分もありそうでした。そのため、Goメンテナ側はさらなる議論が必要としています。その後この話題が進展している様子はwatchできていません。

Goで構造的並行処理を実現する選択肢

言語自体に組み込まれてはいないものの、標準/準標準/3rd party packageを利用すればある程度は実現することはできます。筆者が把握している範囲に限ってですが、以下のpackageについて簡単に機能性を整理します。

• golang.org/x/sync/errgroup
• sourcegraph/conc
• arunsworld/nursery

それぞれを構造的並行処理の3要件（ライフタイム束縛・エラー伝播・キャンセル伝播）の観点で整理してみました。

各packageへのリンクは以下です。

• errgroup : https://pkg.go.dev/golang.org/x/sync/errgroup
• conc : https://pkg.go.dev/github.com/sourcegraph/conc
• nursery : https://pkg.go.dev/github.com/arunsworld/nursery

errgroup/concは3要件を概ね満たせるものの、Wait()の呼び忘れを構文レベルで防ぐことはできません。また、errgroup.Groupのインスタンスを宣言→Groupからgoroutineを実行→Wait()で合流待機といった、ある種 「手続的な」書き方がgoroutineのライフタイムをわかりにくくしている原因なのではないかなと個人的に考えています。

func main() {
  var g errgroup.Group

  g.Go(func() error {
    // ...
    return nil
  })
  g.Go(func() error {
    // ...
    return nil
  })

  if err := g.Wait(); err != nil {
    fmt.Println("error:", err)
  }
}

手続き的という特性には、その記述スタイルにある程度の自由性が付与されていると考えることができます。例えば、以下のように書くことも可能です。

func main() {
  var g errgroup.Group
  spawnTasks(&g)

  // 何らかの処理...

  if err := g.Wait(); err != nil {
    fmt.Println("error:", err)
  }
}

func spawnTasks(g *errgroup.Group) {
  g.Go(func() error {
    // ...
    return nil
  })
  g.Go(func() error {
    // ...
    return nil
  })
}

Go()でgoroutineを起動する場所とWait()で合流する場所をいくらでも分離できる訳ですね。このようにコーディングされた場合、コードを全体的に見渡さないと実行されているgoroutineがいつまで生き続けるのかを把握しづらくなります。

そこで、上のコード中に何らかの実装追加を以下のように行ったとします。

func main() {
  var g errgroup.Group
  spawnTasks(&g)

  // コード追加
  if err := somethingCheck(); err != nil {
    log.Println("error:", err)
    return // ← g.Wait()を通らない。task A/Bは走りっぱなしのまま終了
  }

  // 何らかの処理...

  if err := g.Wait(); err != nil {
    fmt.Println("error:", err)
  }
}

この時点で、spawnTasks()内で起動したgoroutineを待機するチェックポイントは通過しないルートが出てきてしまいました。 goroutineでファイル/ネットワーク通信処理が行われていた場合、main()の終了とともにgoroutineは強制終了となります(プログラムとして終了するため)。

また、これがHTTPハンドラー関数だとすると、goroutine leakの温床になります。handleRequest()がreturnしてもサーバープロセスは生き続けるので、goroutineは孤児として残り続ける可能性があります。

func handleRequest(w http.ResponseWriter, r *http.Request) {
  var g errgroup.Group
  spawnTasks(&g)

  // コード追加
  if err := somethingCheck(); err != nil {
    log.Println("error:", err)
    return // ← g.Wait()を通らない。task A/Bは走りっぱなしのまま終了
  }

  // 何らかの処理...

  if err := g.Wait(); err != nil {
    log.Println("error:", err)
  }
}

arunsworld/nurseryはその点、タスクグループを構文として強制している点が良く、構造的並行処理の3つの要件を満たします。

arunsworld/nurseryはPythonのtrioライブラリの一機能であるnurseryから着想をそのまま得ているものと思われます。 trioのnurseryは"Notes on structured concurrency, or: Go statement considered harmful"という構造的並行処理に関して強い影響力を与えたブログの著者であるNathaniel J. Smith氏が開発したものです。

arunsworld/nurseryは、他言語に提供されている機能性と比べると、ミニマムなものとなっています。

Kotlin coroutines / Swift Structured Concurrency / Python asyncio.TaskGroup / Java StructuredTaskScope などで一般的に提供されている機能には、以下のようなものがあります。

• panic / 例外回収
  • タスクの異常終了を補足して親に伝播
  • 例: Kotlin（コルーチンの例外は親 CoroutineScope に自動伝播）/ Swift withThrowingTaskGroup / Python TaskGroup (ExceptionGroup として再送出) / Java scope.throwIfFailed()
• 動的タスク追加
  • スコープ内で実行中のコードから新たなタスクを追加できる
  • 例: Kotlin launch / Swift TaskGroup.addTask / Python TaskGroup.create_task / Java scope.fork
• Timeout / Deadline
  • スコープ全体に時間制限を設ける
  • 例: Kotlin withTimeout / Python asyncio.timeout / Java joinUntil (Swift はコアAPI無し、Task.sleep + group.cancelAll() で自作が慣用っぽい)
• ネスト子スコープ
  • 親スコープから子スコープを派生させる明示API
  • 例: Kotlin nested coroutineScope / Swift nested withTaskGroup / Python nested asyncio.TaskGroup / Java nested StructuredTaskScope
• 型付き結果取得
  • タスクの戻り値を型安全に受け取る
  • 例: Kotlin Deferred<T> / Swift async let / Python asyncio.Task[T].result() / Java Subtask<T>.get()
• エラー集約
  • 全タスクのエラーをまとめて取得
  • 例: Python ExceptionGroup (PEP 654) / Java カスタムポリシー
• Supervisor mode: 一部のタスクが失敗しても他のタスクをキャンセルしない
  • 例: Kotlin supervisorScope / SupervisorJob (Java は JEP 505 以降の Joiner.awaitAll() で同等のことが可能)
• 最初の成功で他をキャンセル
  • いずれかが成功した時点で他のタスクを打ち切る
  • 例: Java StructuredTaskScope.ShutdownOnSuccess

そこで、構造的並行処理の概念に忠実であり、かつ一定の機能性を兼ね備えたpackageを目指して筆者が実装したのが、次に紹介するscopepackageになります。

scope packageの紹介

コンセプト / 機能性の観点から簡単に紹介します。

コンセプト

最も基本的な設計コンセプトは、

• goroutineのライフタイムをレキシカルスコープによって「宣言的に」束縛する

です。

その点は、他言語の構造的並行処理の実装と比べると特異点はありませんし、arunsworld/nurseryと志向は同じです。

goroutineのライフタイムはscopeに閉じている

scopeでは、タスクグループをscope.Runという一つの関数呼び出しで表現します。scope.Runのbody（第二引数のコールバック）の中だけが「並行処理を起動して良いスコープ」であり、scope.Runがreturnする時点で、その内側で起動したgoroutineは必ず完了またはキャンセル済みであることが保証されます。

errgroupで挙げた例をscopeで書き直すと以下のようになります。

func main() {
    err := scope.Run(context.Background(), func(s *scope.Scope) error {
        s.Go(func(ctx context.Context) error {
            // ...
            return nil
        })
        s.Go(func(ctx context.Context) error {
            // ...
            return nil
        })
        return nil
    })
    if err != nil {
        fmt.Println("error:", err)
    }
}

scope.Runのbodyの開き括弧がタスクグループの始点、閉じ括弧が終点です。Wait()に相当する明示的な合流処理は存在せず、scope.Runがreturnする時点で合流が完了します。

これにより、

• 「Wait()の呼び忘れ」を構文レベルで不可能にする
• 起動と合流の場所が分離されない（Scopeインスタンスがscope.Runの外に漏れない）
• bodyの途中でreturnしても、起動済みのgoroutineはscope.Runで待機・回収される

といった効果が得られます。

機能性

他言語の構造的並行処理に実装された一般的な機能性に倣う形で、scopeでは以下を提供しています。

panic / 例外回収

scope.Run配下で起動されたgoroutine内のpanicは自動でrecoverされ、scope: panic recovered: ...という形式のerrorに変換されてscope.Runの戻り値として返ります。プロセスは落ちません。

err := scope.Run(ctx, func(s *scope.Scope) error {
    s.Go(func(ctx context.Context) error {
        panic("something bad")
    })
    return nil
})
// err: scope: panic recovered: something bad (+ stack trace)

動的タスク追加

Scope.Goはbody内に限らず、既に起動済みのgoroutine内からも呼び出せます。事前にタスクを配列で列挙しておく必要はありません。

scope.Run(ctx, func(s *scope.Scope) error {
    s.Go(func(ctx context.Context) error {
        for _, item := range fetchItems(ctx) {
            s.Go(func(ctx context.Context) error {
                return process(ctx, item)
            })
        }
        return nil
    })
    return nil
})

Timeout / Deadline

scope.WithTimeout(d)オプションでスコープ全体に時間制限を設けられます。期限超過時はスコープのcontextがキャンセルされ、context.DeadlineExceededが返ります。

err := scope.Run(ctx, func(s *scope.Scope) error {
    s.Go(func(ctx context.Context) error {
        <-ctx.Done()
        return ctx.Err()
    })
    return nil
}, scope.WithTimeout(10*time.Millisecond))

ネスト子スコープ

Scope.Scopeで子スコープを派生できます。子スコープのcontextは親のcontextから派生しており、子スコープ内のgoroutineは子スコープを抜ける時点で完了が保証されます。

scope.Run(ctx, func(s *scope.Scope) error {
    s.Scope(func(child *scope.Scope) error {
        child.Go(func(ctx context.Context) error {
            // child scope 内のタスク
            return nil
        })
        return nil
    })
    // ここに到達した時点で child scope の全タスクは完了済み
    return nil
})

型付き結果取得

scope.GoFuture[T]でジェネリックな結果を返すgoroutineを起動でき、Future[T].Wait()で型安全に値を取り出せます。chan Tを別途用意したりクロージャで外部変数に書き込んだりせずに、戻り値として直接受け取れます。タスク間に依存がある場面（DAG的なフロー）で有用です。

scope.Run(ctx, func(s *scope.Scope) error {
    aF := scope.GoFuture(s, func(ctx context.Context) (int, error) {
        return 9, nil
    })
    bF := scope.GoFuture(s, func(ctx context.Context) (int, error) {
        a, err := aF.Wait()
        if err != nil {
            return 0, err
        }
        return a + 1, nil
    })
    s.Go(func(ctx context.Context) error {
        v, _ := bF.Wait()
        fmt.Println(v) // 10
        return nil
    })
    return nil
})

エラー集約

デフォルトは「最初のerrorのみを返す（first-error-wins）」挙動ですが、scope.WithErrAggregation()を渡すと全goroutineのerrorをerrors.Joinでまとめて返します。

err := scope.Run(ctx, func(s *scope.Scope) error {
    s.Go(func(ctx context.Context) error { return errA })
    s.Go(func(ctx context.Context) error { return errB })
    return nil
}, scope.WithErrAggregation())
// errors.Is(err, errA) && errors.Is(err, errB)

Supervisor mode

scope.WithSupervisor()を渡すと、同スコープ内のgoroutineの失敗が他の兄弟goroutineのcontextをキャンセルしなくなります。「片方が落ちても他は走らせ続けたい」というケース向けです。

scope.Run(ctx, func(s *scope.Scope) error {
    s.Go(func(ctx context.Context) error {
        return errors.New("task failed")
    })
    s.Go(func(ctx context.Context) error {
        // ctx はキャンセルされない
        return doWork(ctx)
    })
    return nil
}, scope.WithSupervisor())

最初の成功で他をキャンセル

scope.WithCancelOnSuccess()を渡すと、いずれかのgoroutineがnil errorで返った時点でスコープのcontextをキャンセルします。ヘッジリクエスト（同じ問い合わせを複数経路に投げて速い方を採用する）パターンに利用できます。

err := scope.Run(ctx, func(s *scope.Scope) error {
    s.Go(func(ctx context.Context) error { return queryA(ctx) })
    s.Go(func(ctx context.Context) error { return queryB(ctx) })
    return nil
}, scope.WithCancelOnSuccess())

最大並行数の制限

これは構造的並行処理に必須の機能ではないですが、実用的な機能としてscope.WithMaxConcurrency(n)を提供しています。スコープ内で同時に走るgoroutineの数を制限でき、上限に達した状態でのScope.Goは空きが出るまで待機します。外部API呼び出しのレートリミット対策や、リソース消費の抑制に使えます。

scope.Run(ctx, func(s *scope.Scope) error {
    for _, item := range items {
        s.Go(func(ctx context.Context) error {
            return process(ctx, item)
        })
    }
    return nil
}, scope.WithMaxConcurrency(3))

s.Go(func(ctx context.Context) error {
    for _, item := range items {
        select {
        case <-ctx.Done():
            return ctx.Err() // キャンセル通知を受けて早期終了
        default:
        }
        // itemに対する処理
        process(item)
    }
    return nil
})

詳しくは、scope packageのドキュメントをご参照ください。

おわりに

scopepackage はまだ、構造的並行処理の記述を理想的なインターフェースに落としこみつつ、他言語を参考に機能を実装したという段階です。

ドッグフーディング的に使いつつ、育てていければと思います。

参考