Go-flavored concurrency in C

Go's concurrency is one of the main reasons people like the language. You write go f(), send values through channels, and the runtime scheduler runs thousands of goroutines on just a few OS threads. It feels effortless.None of that machinery exists in C. Which made me wonder: how close can you get to Go's concurrency model using only POSIX threads? Obviously, native OS threads can't match the efficiency of lightweight goroutines, but what is the actual cost, when does it become a problem, and is there any way to at least partially avoid it?I ran into these questions while adding concurrency to Solod (So), a strict subset of Go that translates to plain C, with no runtime and no garbage collector. In the end, I came to the conclusion that you can do quite a lot with pthreads — as long as you're honest about the tradeoffs.This post is about the POSIX threads-based concurrency model I chose, the benefits it offers, and its limitations.Mutex/Cond • Atomics • Pool • Channel • Performance • Design • Wrapping upMutex/CondEverything in So's concurrency stack is built on two basic POSIX primitives: the mutex and the condition variable. sync.Mutex is a thin wrapper around pthread_mutex_t:// Extracted from So's stdlib source code. 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") } } Since So translates to C, this is basically a struct that holds a pthread_mutex_t and a function that calls pthread_mutex_lock. Here's the transpiler output:// The translated C code. 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"); } } That is the whole translation — the generated C is a near-mechanical mirror of the So code, only noisier. From here on, I'll mainly show the So version, but I'll also provide the C code for those who are interested.

There's nothing exciting here: sync.Mutex is a pthread mutex wrapper that panics if something goes wrong (which is rare).The companion primitive is sync.Cond, a wrapper around pthread_cond_t. It's the standard "wait until a condition holds" tool, associated with a mutex:type Cond struct // wraps pthread_cond_t + pthread_mutex_t func (c *Cond) Wait() // wraps pthread_cond_wait func (c *Cond) Signal() // wraps pthread_cond_signal func (c *Cond) Broadcast() // wraps pthread_cond_broadcast Show the translated C codetypedef struct sync_Cond { pthread_cond_t cond; sync_Mutex* mu; } sync_Cond;

void sync_Cond_Wait(sync_Cond* c); // wraps pthread_cond_wait void sync_Cond_Signal(sync_Cond* c); // wraps pthread_cond_signal void sync_Cond_Broadcast(sync_Cond* c); // wraps pthread_cond_broadcast These two types — Mutex and Cond — are the foundation. Other concurrency tools — Once, the thread pool, channels — are built using a mutex and one or more condition variables. This has several effects on performance, as we'll see later.AtomicsNot everything needs a lock. So's sync/atomic mirrors Go's: Bool, Int32, Int64, Uint32, Uint64, and a generic Pointer[T], all with Load, Store, Swap, and CompareAndSwap methods.The nice thing is that these don't need pthreads at all. They map directly to the C compiler's __atomic builtins — the same hardware instructions that Go's compiler emits. So there's no reason for them to be any slower, and they're not:Atomic opGoSoWinnerLoad2ns2ns~sameStore2ns2ns~sameCompareAndSwap13ns13ns~sameEach number is the cost of one operation on a single thread.sync.Once is a good example of using atomics effectively. Its fast path only needs a single atomic load — after the given function runs, every future call to Do checks a flag and returns:type Once struct { mu Mutex done atomic.

Bool }

// Do calls f if and only if Do is being called // for the first time for this o. func (o *Once) Do(f func()) { if o.done.Load() { // lock-free fast path return } // slow path... } Show the translated C codetypedef struct sync_Once { sync_Mutex mu; atomic_Bool done; } sync_Once;

// Do calls f if and only if Do is being called // for the first time for this o. void sync_Once_Do(sync_Once* o, void (*f)()) { if (atomic_Bool_Load(&o->done)) { // lock-free fast path return; } // slow path... } Worker poolTo actually run code concurrently, you need threads. The conc.Thread type wraps pthread_t and its related functions:type Thread struct // wraps pthread_t func (th Thread) Wait() any // wraps pthread_join func (th Thread) Detach() // wraps pthread_detach Show the translated C codetypedef struct conc_Thread { pthread_t t; } conc_Thread;

void* conc_Thread_Wait(conc_Thread th); // wraps pthread_join void conc_Thread_Detach(conc_Thread th); // wraps pthread_detach Consider this conc.Go function:// Go launches an OS thread that runs fn(arg) and returns a handle to it. func Go(entry func(any) any, arg any) Thread { var th Thread rc := pthread_create(&th.t, nil, entry, arg) // ... } Show the translated C code// Go launches an OS thread that runs fn(arg) and returns a handle to it. // `any` in So translates to `void*` in C. conc_Thread conc_Go(void* (*entry)(void*), void* arg) { conc_Thread th = {0}; int rc = pthread_create(&th.t, NULL, entry, arg); // ... } Usage example:func work(arg any) any { acc := arg.(*Account) // ... }

func main() { var acc Account th := conc.Go(work, &acc) // ... do other work concurrently ... th.

Wait() // work is complete once Wait returns } Show the translated C codevoid* work(void* arg) { main_Account* acc = (main_Account*)arg; // ... }

int main(void) { main_Account acc = {0}; conc_Thread th = conc_Go(work, &acc); // ... do other work concurrently ... conc_Thread_Wait(th); // work is complete once Wait returns } It might look like go work(&acc), but that's just on the surface. conc.Go starts an actual OS thread, not a goroutine. You have to eventually call Wait to join or Detach it, or else its resources will leak. Also, OS threads are expensive to create — they're nothing like Go's goroutines, which only need a few kilobytes of stack and start up in nanoseconds.That's exactly why you usually don't want to call Go inside a loop. For tasks that are short-lived or happen often, it's better to use a pool of long-lived worker threads and send tasks to them.conc.Pool to the rescue: 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 │ └────────────────────────────────┘ Usage example: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() } Show the translated C codetypedef 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 is a generic macro to get the i-th element of a // specific type (main_Task here) from a type-erased slice. // Here we're getting the i-th task from the tasks slice. 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); } The first argument to NewPool, mem.System, is a memory allocator. Solod avoids hidden allocations, so anything that needs memory takes an allocator explicitly — here it backs the pool's task queue.Under the hood, a Pool is a fixed group of worker threads that pull tasks from a shared queue (a ring buffer). It uses one mutex and a few condition variables:// Pool is a bounded pool of worker threads with a wait queue // which execute tasks of the form func(any). type Pool struct { alloc mem.Allocator

mu sync.Mutex notEmpty sync.Cond // signaled when a task is enqueued notFull sync.Cond // signaled when a slot frees allDone sync.

Cond // broadcast when no task is in flight

workers []Thread queue []task // ring buffer of submitted tasks active int // tasks submitted but not yet finished stopped bool // set by Free to drain and exit }

// NewPool creates a pool with a given number // of worker threads and starts them. func NewPool(alloc mem.Allocator, opts PoolOpts) *Pool

// Go submits a task for execution, blocking while the queue is full. func (p *Pool) Go(fn func(any), arg any)

// Wait blocks until all submitted tasks finish. func (p *Pool) Wait() Show the translated C code// Pool is a bounded pool of worker threads with a wait queue // which execute tasks of the form func(any). typedef struct conc_Pool { mem_Allocator alloc;

sync_Mutex mu; sync_Cond notEmpty; // signaled when a task is enqueued sync_Cond notFull; // signaled when a slot frees sync_Cond allDone; // broadcast when no task is in flight

so_Slice workers; so_Slice queue; // ring buffer of submitted tasks so_int active; // tasks submitted but not yet finished bool stopped; // set by Free to drain and exit } 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 wakes up a worker when there are tasks to do, notFull applies back-pressure when the queue is full, and allDone lets Wait know when everything is finished. It's a classic producer-consumer setup, about 200 lines of code, and there's nothing fancy about it.The heart of the pool is the worker loop. Each thread blocks until a task appears, runs it outside the lock so workers execute in parallel, then records that it finished:// workerMain runs on every pool thread: pull a task, run it, repeat. func workerMain(arg any) any { p := arg.(*Pool) for { p.mu.Lock() for p.qempty() && !