Skip to content

A buffered channel as a semaphore

Level: 201 · you need to limit how many goroutines do something at once, without keeping a pool

One line: make(chan struct{}, n) is a counting semaphore: a send takes a slot and blocks while all n are taken, and a receive gives one back, so no more than n goroutines are ever between the two. With a barrier that holds the first three inside until main has seen a fourth kept out, the most inside at once is exactly 3, on every run.

The output

Verified output of a_buffered_channel_as_a_semaphore_go.go — regenerated by tools/run_examples.py, never hand-typed.

slots:                   3
tasks started:           5
tasks inside:            3
slots taken:             3 of 3
started, not let in:     2
a 4th acquire:           would block
main opens the barrier
tasks finished:          5
slots taken:             0 of 3
most inside at once:     3
a_buffered_channel_as_a_semaphore_go.go

a_buffered_channel_as_a_semaphore_go.go in full — pasted here by tools/run_examples.py from the file CI runs.

// A buffered channel with room for 3 is a semaphore with 3 slots: a send takes
// a slot, and blocks while all 3 are taken; a receive gives one back. Five
// resize tasks share it. The tasks that get in wait at a barrier, which main
// opens only after it has seen 3 inside and shown that a 4th acquire would
// block, so the most tasks inside at once is exactly 3, on every run.
//
//  go build a_buffered_channel_as_a_semaphore_go.go && ./a_buffered_channel_as_a_semaphore_go
package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

const (
    tasks = 5
    slots = 3
)

func main() {
    sem := make(chan struct{}, slots) // the semaphore: its buffer holds the taken slots

    barrier := make(chan struct{}) // closed by main: until then, no task inside may leave
    entered := make(chan struct{}) // a task reports here once it holds a slot

    var started, all sync.WaitGroup
    var inside, mostInside atomic.Int32

    started.Add(tasks)
    for range tasks {
        all.Go(func() {
            started.Done()

            sem <- struct{}{} // acquire: blocks while every slot is taken
            n := inside.Add(1)
            for m := mostInside.Load(); n > m && !mostInside.CompareAndSwap(m, n); m = mostInside.Load() {
            }
            entered <- struct{}{}

            <-barrier // the resize would happen here

            inside.Add(-1)
            <-sem // release: gives the slot back
        })
    }

    started.Wait()
    for range slots {
        <-entered
    }
    fmt.Printf("slots:                   %d\n", cap(sem))
    fmt.Printf("tasks started:           %d\n", tasks)
    fmt.Printf("tasks inside:            %d\n", inside.Load())
    fmt.Printf("slots taken:             %d of %d\n", len(sem), cap(sem))
    fmt.Printf("started, not let in:     %d\n", tasks-int(inside.Load()))

    select {
    case sem <- struct{}{}:
        fmt.Println("a 4th acquire:           succeeded") // cannot happen: the buffer is full
        <-sem
    default:
        fmt.Println("a 4th acquire:           would block")
    }

    fmt.Println("main opens the barrier")
    close(barrier)
    for range tasks - slots {
        <-entered
    }
    all.Wait()

    fmt.Printf("tasks finished:          %d\n", tasks)
    fmt.Printf("slots taken:             %d of %d\n", len(sem), cap(sem))
    fmt.Printf("most inside at once:     %d\n", mostInside.Load())
}

Reading the output

Why a channel can count. The spec says a send on a buffered channel succeeds without blocking only while the buffer is not full (spec: Channel types ↗). So the values in the buffer are the taken slots: len(sem) is how many are taken, cap(sem) is how many there are, and the send that would make n+1 waits for a receive. The element type is struct{} because the value carries no meaning, and the spec gives a struct with no fields a size of zero (spec: Size and alignment guarantees ↗). Effective Go ↗ uses the same trick to limit the number of requests being processed at once.

tasks inside: 3 and started, not let in: 2. main waited until all five goroutines had started (the started WaitGroup) and three of them had reported on entered from inside the semaphore. Those three are waiting at the barrier, a channel that main has not closed yet, so none of them can give back its slot. With all three slots taken, the other two cannot get in. The first lines of the output describe that moment, and it is the same moment on every run.

a 4th acquire: would block. main tries to take a slot itself, in a select with a default case. The spec says that when none of a select's communications can proceed and there is a default, the default runs (spec: Select statements ↗; Select waits on many). So this is a try-acquire, and it fails.

most inside at once: 3. When main closes the barrier, the three leave, the other two take their slots, and all five finish with every slot given back (0 of 3). Each task counted itself in with an atomic counter and recorded the highest count (Atomic counters). The semaphore keeps that count at 3 or below, and the barrier makes sure it reaches 3. Without the barrier, tasks this short could each finish before the next one got in, and the key would record whatever the scheduler happened to do that run.

Where the acquire goes. Here each goroutine acquires its own slot, so all five goroutines exist and two of them sit blocked, which costs little (Goroutines are cheap). Acquiring in the loop, before the go statement, limits the number of goroutines as well, at the price of blocking the loop.

A semaphore or a pool. A worker pool keeps n goroutines and feeds them jobs through a channel. A semaphore keeps no goroutines: any number of them may exist, and it lets n at a time into the part of the work that needs limiting.

The packaged version. golang.org/x/sync/semaphore ↗ has a Weighted semaphore. Its Acquire ↗ takes a context and gives up when the context is done, and takes a weight, so that one caller can hold several slots. errgroup's SetLimit ↗ puts the same kind of limit on a group of goroutines. Both live outside the standard library, so no program here uses them.

What to do

  • Acquire with a send and release with a receive, on every path: sem <- struct{}{}, then defer func() { <-sem }().
  • Use struct{} as the element type. Only the count matters.
  • To give up instead of waiting, acquire in a select: with a default case to turn work away when the semaphore is full, or with case <-ctx.Done() to wait no longer than the request lives.
  • Test a limit with a barrier, not a sleep. Hold the tasks inside until the limit is reached, as this program does, and the maximum becomes exact.

In other languages

  • Java's Semaphore ↗, Python's threading.Semaphore ↗ and C++20's std::counting_semaphore ↗ are the named type this channel stands in for: a count of permits, with acquire and release. Rust's std::sync ↗ lists no semaphore. The Concurrency library's chapter 04 Waiting for each other, where semaphores and barriers will be compared across languages, is planned.
  • The Rust library's Channels ↗ shows mpsc::sync_channel(2) blocking on the third send: the same full buffer that this page counts slots with.

Sources