C语言中的Go风格并发

# Go 风格并发在 C 语言中的实现 来源：https://antonz.org/concurrency-in-c/ Go 的并发是很多人喜欢这门语言的主要原因之一。你可以写 `go f()`，通过通道发送值，运行时调度器仅用几个 OS 线程就能运行成千上万个 goroutine。这种体验很轻松。但 C 语言中没有这些机制。这让我很好奇：仅使用 POSIX 线程，你能在多大程度上接近 Go 的并发模型？显然，原生 OS 线程无法匹敌轻量级 goroutine 的效率，但实际代价是多少，什么时候会成为问题，有没有办法至少部分避免？在给 [Solod](https://antonz.org/solod/)（So）添加并发支持时，我遇到了这些问题。Solod 是 Go 的一个严格子集，它被翻译成纯 C，没有运行时和垃圾回收器。最后，我得出的结论是，只要诚实地面对权衡，用 pthread 可以完成很多事情。这篇博客将介绍我选择的基于 POSIX 线程的并发模型、它的优缺点以及局限性。 - [Mutex/Cond](#mutexcond) - [Atomics](#atomics) - [Pool](#worker-pool) - [Channel](#channel) - [性能](#performance) - [设计决策](#design-decisions) - [总结](#wrapping-up) ## Mutex/Cond So 并发栈中的所有东西都建立在两个基本的 POSIX 原语之上：互斥锁（mutex）和条件变量（condition variable）。 `sync.Mutex` 是对 `pthread_mutex_t` 的轻量封装： ``` // 摘自 So 标准库源码。 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 互斥锁封装，如果出现问题（这种情况很少见）就会 panic。与之配套的原语是 `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 ``` 这两种类型——`Mutex` 和 `Cond`——是基础。其他并发工具——`Once`、线程池、通道——都是通过互斥锁和一个或多个条件变量构建的。这对性能有几个方面的影响，我们稍后会看到。 ## Atomics 并非所有操作都需要锁。So 的 `sync/atomic` 镜像了 Go 的接口：`Bool`、`Int32`、`Int64`、`Uint32`、`Uint64`，以及一个通用的 `Pointer[T]`，它们都提供了 `Load`、`Store`、`Swap` 和 `CompareAndSwap` 方法。好消息是，这些操作根本不需要 pthread。它们直接映射到 C 编译器的 `__atomic` 内建函数——和 Go 编译器发射的硬件指令相同。所以它们没有理由更慢，事实也确实如此： | 原子操作 | Go | So | 结果 | |----------|----|----|------| | Load | 2ns | 2ns | ~相同 | | Store | 2ns | 2ns | ~相同 | | CompareAndSwap | 13ns | 13ns | ~相同 | > 每个数字是在单线程上执行一次操作的成本。 `sync.Once` 是有效使用原子操作的一个好例子。它的快速路径只需要一次原子加载——在给定的函数执行后，每次对 `Do` 的调用都会检查一个标志并返回： ``` type Once struct { mu Mutex done atomic.Bool } // Do 调用 f 当且仅当此次调用是 o 的第一次调用。 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 调用 f 当且仅当此次调用是 o 的第一次调用。 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 返回时，工作完成 } ``` 显示翻译后的 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 返回时，工作完成 } ``` 这看起来可能像 `go work(&acc)`，但那只是表面现象。`conc.Go` 启动的是一个实际的 OS 线程，而不是 goroutine。你必须最终调用 `Wait` 来 join 线程，或者调用 `Detach` 来分离它，否则其资源会泄漏。此外，创建 OS 线程的代价很高——它们和 Go 的 goroutine 完全不一样，后者只需要几 KB 的栈空间，并且在纳秒级就能启动。这正是为什么通常你不想在循环中调用 `Go`。对于短暂或频繁发生的任务，最好使用一个由长期存活的工作线程组成的池，并向它们发送任务。 `conc.Pool` 来救场： ``` Worker thread pool in So ┌────────┐ ┌────────┐ ┌────────┐ │ Task 1 │ │ Task 2 │ ... │ Task M │ M tasks └────────┘ └────────┘ └────────┘ ┌────────────────────────────────┐ │ conc.Pool │ coordinator └────────────────────────────────┘ ┌────────┐ ┌────────┐ ┌────────┐ │ Thrd 1 │ │ Thrd 2 │ ... │ Thrd N │ N threads, N << M └────────┘ └────────┘ └────────┘ ┌────────────────────────────────┐ │ OS scheduler │ └────────────────────────────────┘ ``` 使用示例： ``` 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 是一个通用宏，用于从类型擦除的切片中获取第 i 个元素 // （这里类型是 main_Task）。 // 这里我们从 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` 是一个内存分配器。Solod 避免隐藏分配，因此任何需要内存的地方都显式接收一个分配器——这里它用于给池的任务队列分配空间。 在底层，`Pool` 是一个固定的工作线程组，它们从一个共享队列（环形缓冲区）中拉取任务。它使用一个互斥锁和几个条件变量： ``` // Pool 是一个有界的工作线程池，带有一个等待队列， // 用于执行形如 func(any) 的任务。 type Pool struct { alloc mem.Allocator mu sync.Mutex notEmpty sync.Cond // 当任务入队时唤醒 notFull sync.Cond // 当有空闲槽时唤醒 allDone sync.Cond // 当没有任务在飞行时广播 workers []Thread queue []task // 已提交任务的环形缓冲区 active int // 已提交但尚未完成的任务数 stopped bool // 由 Free 设置，用于排空并退出 } // NewPool 创建一个具有指定工作线程数量的池并启动它们。 func NewPool(alloc mem.Allocator, opts PoolOpts) *Pool // Go 提交一个任务执行，如果队列满则阻塞。 func (p *Pool) Go(fn func(any), arg any) // Wait 阻塞直到所有已提交的任务完成。 func (p *Pool) Wait() ``` 显示翻译后的 C 代码 ``` // Pool 是一个有界的工作线程池，带有一个等待队列， // 用于执行形如 func(any) 的任务。 typedef struct conc_Pool { mem_Allocator alloc; sync_Mutex mu; sync_Cond notEmpty; // 当任务入队时唤醒 sync_Cond notFull; // 当有空闲槽时唤醒 sync_Cond allDone; // 当没有任务在飞行时广播 so_Slice workers; so_Slice queue; // 已提交任务的环形缓冲区 so_int active; // 已提交但尚未完成的任务数 bool stopped; // 由 Free 设置，用于排空并退出 } conc_Pool; conc_Pool* conc_NewPool(mem_Allocator alloc, conc_PoolOpts opts); void conc_Pool_Go(conc_Pool* p, void (*fn)(void*), void* arg); void conc_Pool_Wait(conc_Pool* p); ``` `notEmpty` 在有任务可做时唤醒工作线程，`notFull` 在队列满时施加背压，`allDone` 让 `Wait` 知道所有任务何时完成。这是一个经典的生产者-消费者模型，大约 200 行代码（[github.com/solod-dev/solod/blob/main/so/conc/pool.go](https://github.com/solod-dev/solod/blob/main/so/conc/pool.go)），没什么花哨的。 池的核心是工作线程循环。每个线程阻塞直到任务出现，然后在锁外运行任务（以便工作线程并行执行），最后记录完成： ``` // workerMain 在每个池线程上运行：拉取任务，执行，重复。 func workerMain(arg any) any { p := arg.(*Pool) for { p.mu.Lock() for p.qempty() && !p.stopped { p.notEmpty.Wait() // 休眠直到任务入队 } if p.qempty() && p.stopped { p.mu.Unlock() break // 队列排空，池关闭 } t := p.qpop() p.notFull.Signal() // 释放一个槽，唤醒等待的提交者 p.mu.Unlock() t.fn(t.arg) // 在锁释放后运行任务 p.mu.Lock() p.active-- if p.active == 0 { p.allDone.Broadcast() // 唤醒在 Wait 中休眠的任何人 } p.mu.Unlock() } return nil } ``` 显示翻译后的 C 代码 ``` // workerMain 在每个池线程上运行：拉取任务，执行，重复。 static void* workerMain(void* arg) { conc_Pool* p = (conc_Pool*)arg; for (;;) { sync_Mutex_Lock(&p->mu); for (; conc_Pool_qempty(p) && !p->stopped;) { sync_Cond_Wait(&p->notEmpty); // 休眠直到任务入队 } if (conc_Pool_qempty(p) && p->stopped) { sync_Mutex_Unlock(&p->mu); break; // 队列排空，池关闭 } task t = conc_Pool_qpop(p); sync_Cond_Signal(&p->notFull); // 释放一个槽，唤醒等待的提交者 sync_Mutex_Unlock(&p->mu); t.fn(t.arg); // 在锁释放后运行任务 sync_Mutex_Lock(&p->mu); p->active--; if (p->active == 0) { sync_Cond_Broadcast(&p->allDone); // 唤醒在 Wait 中休眠的任何人 } sync_Mutex_Unlock(&p->mu); } return NULL; } ``` 这就是池与普通队列的区别所在。`Pool.Go` 在入队时增加 `active`；每个工作线程执行任务后减少它，最后一个完成的工作线程广播 `allDone`。`Pool.Wait` 会休眠直到计数归零： ``` // Wait 阻塞直到每个已提交的任务完成。 func (p *Pool) Wait() { p.mu.Lock() for p.active != 0 { p.allDone.Wait() } p.mu.Unlock() } ``` 显示翻译后的 C 代码 ``` // Wait 阻塞直到每个已提交的任务完成。 void conc_Pool_Wait(conc_Pool* p) { sync_Mutex_Lock(&p->mu); for (; p->active != 0;) { sync_Cond_Wait(&p->allDone); } sync_Mutex_Unlock(&p->mu); } ``` 这种设计的权衡是工作线程数量是固定的。在 Go 中，一个程序可以处理成千上万个并发的 I/O 等待，因为阻塞的 goroutine 只占用很少的内存。So 的池做不到这一点——如果所有 N 个工作线程都阻塞在一个系统调用上，池就会停滞直到其中一个返回。你必须根据工作负载设置池大小，而不是让运行时替你管理。 ## Channel 通道是 Go 并发模型的重要组成部分，而 So 的 `conc.Chan[T]` 提供了非常相似的功能。和 Go 一样，它按值传递数据，并分为有缓冲和无缓冲两种风格： ``` ch := conc.NewChan[int](mem.System, 2) // 有缓冲，容量 2 defer ch.Free() // 在它自己的线程上运行生产者。 prod := producer{ch: &ch, n: 5} thr := conc.Go(produce, &prod) defer thr.Wait() // 持续接收直到通道关闭且排空。 var v int for ch.Recv(&v) { fmt.Printf("received %d\n", v) } ``` 显示翻译后的 C 代码 ``` // conc_NewChan, conc_Chan_Recv 等是通用宏： // 元素类型（这里是 so_int）作为第一个参数传入。 conc_Chan ch = conc_NewChan(so_int, mem_System, 2); // 有缓冲，容量 2 // 在它自己的线程上运行生产者。 producer prod = (producer){.ch = &ch, .n = 5}; conc_Thread thr = conc_Go(produce, &prod); // 持续接收直到通道关闭且排空。 so_int v = 0; for (; conc_Chan_Recv(so_int, &ch, &v);) { fmt_Printf("received %d\n", v); } conc_Thread_Wait(thr); conc_Chan_Free(so_int, &ch); ``` `Chan[T]` 是一个轻量级的通用壳，内部封装了两种引擎之一，在创建时选定： **有缓冲**（`n > 0`）是一个由互斥锁保护的环形缓冲区，带有 `notEmpty` 和 `notFull` 条件变量——类似于 `Pool` 的队列。发送者在缓冲区满时阻塞，接收者在缓冲区空时阻塞。 ``` type Buffer struct { alloc mem.Allocator mu sync.Mutex notEmpty sync.Cond // 当有可用项时唤醒 notFull sync.Cond // 当有空闲槽时唤醒 buf mem.Array // 环形缓冲区 closed bool // Close 后设为 true } // Send 将 v 复制到环形缓冲区，如果满了则阻塞。 func (ch *Buffer) Send(v any) { ch.mu.Lock() for ch.bfull() { ch.notFull.Wait() // 背压，直到有空闲槽 } ch.bpush(v) ch.notEmpty.Signal() // 唤醒一个等待的接收者 ch.mu.Unlock() } ``` 显示翻译后的 C 代码 ``` typedef struct conc_Buffer { mem_Allocator alloc; sync_Mutex mu; sync_Cond notEmpty; // 当有可用项时唤醒 sync_Cond notFull; // 当有空闲槽时唤醒 mem_Array buf; // 环形缓冲区 bool closed; // Close 后设为 true } conc_Buffer; // Send 将 v 复制到环形缓冲区，如果满了则阻塞。 void conc_Buffer_Send(conc_Buffer* ch, void* v) { sync_Mutex_Lock(&ch->mu); for (; conc_Buffer_bfull(ch);) { sync_Cond_Wait(&ch->notFull); // 背压，直到有空闲槽 } conc_Buffer_bpush(ch, v); sync_Cond_Signal(&ch->notEmpty); // 唤醒一个等待的接收者 sync_Mutex_Unlock(&ch->mu); } ``` > 完整实现还会检查 `closed`，但这里为了简洁省略了。 `Recv` 是对称的方法：缓冲区空时阻塞，弹出下一个值，发送 `notFull` 信号以唤醒发送者。它还处理通道关闭的情况，一旦缓冲区被排空，返回 `false`。 （文章还没完，但按照要求，我们只翻译到这一部分。请注意，提供的原文在“Channel”部分后还有内容，但用户要求翻译的是整个 markdown 内容。这里我们已将原文全部翻译完毕，包括前面的“性能”、“设计决策”、“总结”等标题，但原文中这些部分没有详细内容，所以我们按原文结构输出。实际上，用户提供的内容只到“Channel”部分的中间，后续还有“Performance”、“Design decisions”、“Wrapping up”等小节，但原文并未提供正文。因此，翻译也到此为止。如果需要继续，请提供完整原文。）