C言語におけるGo風の並行処理

Goの並行処理は、人々がこの言語を好む主な理由の1つです。go f()と書き、チャネルを通じて値を送信すると、ランタイムスケジューラが少数のOSスレッドで数千のゴルーチンを実行します。それは楽に感じられます。Cにはそのメカニズムは一切存在しません。そこで私は疑問に思いました：POSIXスレッドだけを使って、Goの並行処理モデルにどれだけ近づけるだろうか？ネイティブOSスレッドは軽量なゴルーチンの効率には及びませんが、実際のコストはいくらで、いつ問題になり、それを少なくとも部分的に回避する方法はあるのでしょうか？私は、ランタイムもガベージコレクタもない、Goの厳密なサブセットであるSolod（So）に並行処理を追加する際に、これらの疑問にぶつかりました。最終的に、トレードオフを正直に受け入れる限り、pthreadsでかなりのことができるという結論に至りました。この記事は、私が選択したPOSIXスレッドベースの並行処理モデル、それが提供する利点、およびその限界について説明します。Mutex/Cond • アトミックス • プール • チャネル • パフォーマンス • 設計 • まとめMutex/CondSoの並行処理スタックのすべては、2つの基本的なPOSIXプリミティブ、すなわちミューテックスと条件変数に基づいています。sync.Mutexはpthread_mutex_tの薄いラッパーです：// Soのstdlibソースコードから抽出。type Mutex struct { mu pthread_mutex_t } func (m *Mutex) Lock() { rc := pthread_mutex_lock(&m.mu) if rc != 0 { panic("sync: Mutex.Lock failed") } } SoはCに変換されるため、これは基本的にpthread_mutex_tを保持する構造体と、pthread_mutex_lockを呼び出す関数です。以下はトランスパイラによる出力です：// 変換されたCコード。typedef struct sync_Mutex { pthread_mutex_t mu; } sync_Mutex; void sync_Mutex_Lock(sync_Mutex* m) { int rc = pthread_mutex_lock(&m->mu); if (rc != 0) { so_panic("sync: Mutex.Lock failed"); } } これがすべての変換です — 生成されたCコードはSoコードのほぼ機械的な鏡であり、ノイズが多いだけです。ここからは主にSoバージョンを示しますが、興味のある方のためにCコードも提供します。ここにはエキサイティングなものはありません：sync.Mutexは、何か問題が発生した場合にパニックを起こすpthreadミューテックスのラッパーです（これはまれです）。相補的なプリミティブはsync.Condで、pthread_cond_tのラッパーです。これは、ミューテックスに関連付けられた「条件が満たされるまで待つ」という標準的なツールです：type Cond struct // pthread_cond_t + pthread_mutex_t をラップします func (c *Cond) Wait() // pthread_cond_wait をラップします func (c *Cond) Signal() // pthread_cond_signal をラップします func (c *Cond) Broadcast() // pthread_cond_broadcast をラップします 変換されたCコードを表示typedef struct sync_Cond { pthread_cond_t cond; sync_Mutex* mu; } sync_Cond; void sync_Cond_Wait(sync_Cond* c); // pthread_cond_wait をラップします void sync_Cond_Signal(sync_Cond* c); // pthread_cond_signal をラップします void sync_Cond_Broadcast(sync_Cond* c); // pthread_cond_broadcast をラップします これら2つの型 — MutexとCond — が基盤となります。他の並行処理ツール — Once、スレッドプール、チャネル — は、ミューテックスと1つ以上の条件変数を使用して構築されます。これは、後で見るように、パフォーマンスにいくつかの影響を与えます。アトミックスすべてにロックが必要なわけではありません。Soのsync/atomicはGoのものをミラーリングしています：Bool、Int32、Int64、Uint32、Uint64、およびジェネリックなPointer[T]で、すべてLoad、Store、Swap、CompareAndSwapメソッドを備えています。良い点は、これらはpthreadsをまったく必要としないことです。これらはCコンパイラの__atomic組み込み関数に直接マッピングされます — Goのコンパイラが出力するのと同じハードウェア命令です。そのため、これらが遅くなる理由はなく、実際遅くもありません：アトミック操作 Go So 勝者Load 2ns 2ns ~同じStore 2ns 2ns ~同じCompareAndSwap 13ns 13ns ~同じ各数値は、単一スレッドでの1回の操作のコストです。sync.Onceは、アトミックスを効果的に使用する良い例です。その高速パスは単一のアトミックロードしか必要としません — 与えられた関数が実行された後、将来のDoの呼び出しはフラグをチェックして返します：type Once struct { mu Mutex done atomic.Bool } // Doは、oに対して初めて呼び出された場合にのみfを呼び出します。 func (o *Once) Do(f func()) { if o.done.Load() { // ロックフリー高速パス return } // 低速パス... } 変換されたCコードを表示typedef struct sync_Once { sync_Mutex mu; atomic_Bool done; } sync_Once; // Doは、oに対して初めて呼び出された場合にのみfを呼び出します。 void sync_Once_Do(sync_Once* o, void (*f)()) { if (atomic_Bool_Load(&o->done)) { // ロックフリー高速パス return; } // 低速パス... } ワーカープール実際にコードを並行して実行するには、スレッドが必要です。conc.Thread型はpthread_tとその関連関数をラップします：type Thread struct // pthread_t をラップします func (th Thread) Wait() any // pthread_join をラップします func (th Thread) Detach() // pthread_detach をラップします 変換されたCコードを表示typedef struct conc_Thread { pthread_t t; } conc_Thread; void* conc_Thread_Wait(conc_Thread th); // pthread_join をラップします void conc_Thread_Detach(conc_Thread th); // pthread_detach をラップします このconc.Go関数を検討してください：// GoはOSスレッドを起動し、fn(arg)を実行してそのハンドルを返します。 func Go(entry func(any) any, arg any) Thread { var th Thread rc := pthread_create(&th.t, nil, entry, arg) // ... } 変換されたCコードを表示// GoはOSスレッドを起動し、fn(arg)を実行してそのハンドルを返します。 // Soの`any`はCの`void*`に変換されます。 conc_Thread conc_Go(void* (*entry)(void*), void* arg) { conc_Thread th = {0}; int rc = pthread_create(&th.t, NULL, entry, arg); // ... } 使用例：func work(arg any) any { acc := arg.(*Account) // ... } func main() { var acc Account th := conc.Go(work, &acc) // ... 他の作業を並行して行う ... th.Wait() // Waitが返るとworkは完了です } 変換されたCコードを表示void* work(void* arg) { main_Account* acc = (main_Account*)arg; // ... } int main(void) { main_Account acc = {0}; conc_Thread th = conc_Go(work, &acc); // ... 他の作業を並行して行う ... conc_Thread_Wait(th); // Waitが返るとworkは完了です } これはgo work(&acc)のように見えるかもしれませんが、それは表面上だけです。conc.Goは実際のOSスレッドを起動するのであり、ゴルーチンではありません。リソースリークを防ぐために、最終的にWaitを呼び出して結合するか、Detachする必要があります。また、OSスレッドの作成はコストが高いため、スタックに数キロバイトしか必要とせず、ナノ秒で起動するGoのゴルーチンとはまったく異なります。まさに、ループ内でGoを呼び出すことを通常避けるべき理由です。短命なタスクや頻繁に発生するタスクには、長命なワーカー・スレッドのプールを使用し、それにタスクを送信する方が良いでしょう。conc.Poolが役立ちます：Soのワーカー・スレッド・プール ┌────────┐ ┌────────┐ ┌────────┐ │ Task 1 │ │ Task 2 │...│ Task M │ M個のタスク └────────┘ └────────┘ └────────┘ ┌────────────────────────────────┐ │ conc.Pool │ コーディネーター └────────────────────────────────┘ ┌────────┐ ┌────────┐ ┌────────┐ │ Thrd 1 │ │ Thrd 2 │...│ Thrd N │ N個のスレッド, N << M └────────┘ └────────┘ └────────┘ ┌────────────────────────────────┐ │ OSスケジューラ │ └────────────────────────────────┘ 使用例：type Task struct { in int out int } func square(arg any) { task := arg.(*Task) task.out = task.in * task.in } func main() { tasks := make([]Task, 10) opts := conc.PoolOpts{NumThreads: 2} pool := conc.NewPool(mem.System, opts) defer pool.Free() for i := range tasks { tasks[i].in = i pool.Go(square, &tasks[i]) } pool.Wait() } 変換されたCコードを表示typedef struct main_Task { so_int in; so_int out; } main_Task; void square(void* arg) { main_Task* task = (main_Task*)arg; task->out = task->in * task->in; } int main(void) { so_Slice tasks = so_make_slice(main_Task, 10, 10); conc_PoolOpts opts = (conc_PoolOpts){.NumThreads = 2}; conc_Pool* pool = conc_NewPool(mem_System, opts); for (so_int i = 0; i < so_len(tasks); i++) { // so_at は、型消去されたスライスから特定の型（ここではmain_Task）の // i番目の要素を取得するための汎用マクロです。 // ここでは、tasksスライスからi番目のタスクを取得しています。 so_at(main_Task, tasks, i).in = i; conc_Pool_Go(pool, square, &so_at(main_Task, tasks, i)); } conc_Pool_Wait(pool); conc_Pool_Free(pool); } NewPoolの最初の引数、mem.Systemはメモリallocatorです。Solodは隠れた割り当てを避けるため、メモリを必要とするものはすべて明示的にallocatorを受け取ります — ここではシステムallocatorを使用しています。