Limit goroutines with buffered channels

I was cobbling together a long-running Go script to send webhook messages to a system when some events occur. The initial script would continuously poll a Kafka topic for events and spawn new goroutines to make HTTP requests to the destination. This had two problems:

It could create unlimited goroutines if many events arrived quickly
It might overload the destination system by making many concurrent requests

In Python, I’d use just asyncio.Semaphore to limit concurrency. I’ve previously written about this here¹. Turns out, in Go, you could do the same with a buffered channel. Here’s how the naive version looks:

package main

import ("fmt"; "sync")

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()

    // ... Send http post request
    fmt.Println("Sending webhook request")
}

func main() {
    var wg sync.WaitGroup
    nWorkers := 10
    for i := 1; i <= nWorkers; i++ {
        wg.Add(1)
        go worker(i, &wg)
    }
    wg.Wait()
    fmt.Println("All workers have completed")
}

We’re sending the webhook request in the worker function. It takes an integer ID for bookkeeping and a pointer to a WaitGroup instance for synchronization. Once it finishes making the request, it signals the WaitGroup with wg.Done(). In the main function, we spawn 10 workers as goroutines and wait for all of them to finish work with wg.Wait(). Without the wait-group synchronization, the main goroutine would bail before all the background workers finish their work.

In the above scenario, all the requests were made in parallel. How can we limit the system to only allow n number of concurrent requests at the same time? Sure, you can choose to spin up n number of goroutines and no more. But how do you do it from inside an infinite loop that’s also polling a queue continuously?

In this case, I want to throttle the script so that it’ll send 2 requests in parallel and then wait until those are done. Then it’ll wait for a bit before firing up the next batch of 2 goroutines and continuously repeat the same process. Buffered channels allow us to do exactly that. Observe:

package main

import ("fmt"; "sync"; "time")

func worker(id int, sem chan struct{}, wg *sync.WaitGroup) {
    defer wg.Done()

    // Acquire semaphore
    fmt.Printf("Worker %d: Waiting to acquire semaphore\n", id)
    sem <- struct{}{}

    // Do work
    fmt.Printf("Worker %d: Semaphore acquired, running\n", id)
    time.Sleep(10 * time.Millisecond)

    // Release semaphore
    <-sem
    fmt.Printf("Worker %d: Semaphore released\n", id)
}

func main() {
    nWorkers := 10      // Total number of goroutines
    maxConcurrency := 2 // Allowed to run at the same time
    batchInterval := 50 * time.Millisecond // Delay between each batch of 2 goros

    // Create a buffered channel with a capacity of maxConcurrency
    sem := make(chan struct{}, maxConcurrency)

    var wg sync.WaitGroup

    // We start 10 goroutines but only 2 of them will run in parallel
    for i := 1; i <= nWorkers; i++ {
        wg.Add(1)
        go worker(i, sem, &wg)

        // Introduce a delay after each batch of workers
        if i % maxConcurrency == 0 && i != nWorkers {
            fmt.Printf("Waiting for batch interval...\n")
            time.Sleep(batchInterval)
        }
    }
    wg.Wait()
    close(sem) // Remember to close the channel once done
    fmt.Println("All workers have completed")
}

The clever bit here is the buffered channel named sem which acts as a semaphore to limit concurrency. We set its capacity to the max number of goroutines we want running at once, in this case 2. Before making the request, each worker goroutine tries to acquire the semaphore by sending a value into the channel via sem <- struct{}{}. The value itself doesn’t matter. So we’re just sending an empty struct to avoid redundant allocation.

Sending data to the channel will block if it’s already full, essentially meaning all permits are taken. Once the send succeeds, the goroutine has acquired the semaphore and is free to proceed with its work. When finished, it releases the semaphore by reading from the channel <-sem. This frees up a slot in the channel for another goroutine to acquire it. By using this semaphore channel to limit access to critical sections, we can precisely control the number of concurrent goroutines.

This channel-based semaphore gives us more flexibility than just using a WaitGroup. Combining it with a buffered channel provides fine-grained control over simultaneous goroutine execution. The buffer size of the channel determines the allowed parallelism, 2 here. We’ve also thrown in an extra bit of delay after each batch of operation finishes with:

// Introduce additional delay after each batch of workers
if i % maxConcurrency == 0 && i != nWorkers {
    fmt.Printf("Waiting for batch interval...\n")
    time.Sleep(batchInterval)
}

Running the script will show that although we’ve started 10 goroutines in the main function, only 2 of them run at once. Also, there’s a delay of 3 seconds between each batch. We can tune it according to our need to be lenient on the consumer.

Waiting for batch interval...
Worker 2: Waiting to acquire semaphore
Worker 2: Semaphore acquired, running
Worker 1: Waiting to acquire semaphore
Worker 1: Semaphore acquired, running
Worker 1: Semaphore released
Worker 2: Semaphore released
Waiting for batch interval...
Worker 4: Waiting to acquire semaphore
Worker 4: Semaphore acquired, running
Worker 3: Waiting to acquire semaphore
Worker 3: Semaphore acquired, running
Worker 3: Semaphore released
Worker 4: Semaphore released
Waiting for batch interval...
...

Now, you might want to add extra abstractions over the core behavior to make it more ergonomic. Here’s a pointer² on how to do so. Effective Go also mentions³ this pattern briefly.

Recent posts