Leveraging The Sync Package For Concurrency Control: Utilizing The Sync Package To Handle Complex Synchronization Problems

Overview

In this chapter, we will explore how Go’s sync package provides powerful tools to manage synchronization between goroutines, ensuring safe access to shared memory. You will learn how to utilize synchronization primitives like WaitGroups, Mutexes, RWMutexes, sync.Once, sync.Cond, and Semaphores to coordinate concurrent operations effectively. Additionally, we will cover common concurrency issues such as race conditions, deadlocks, and bugs, and demonstrate how to detect and fix them using Go’s race detector and other tools. We will also explore advanced patterns including one-time initialization, condition variables, and controlling concurrency with semaphores, comparing different approaches for throttling concurrent operations. This chapter will equip you with the knowledge to handle complex concurrency problems, making your Go programs more robust and efficient.

Synchronization Between Goroutines

Since goroutines run in the same address space, access to shared memory must be synchronized.

The sync package provides useful primitives to accomplish this, but there are also other primitives such as channels as well.

The Problem

Run the following program:


package main

import "fmt"

func main() {
	go sayHello()
}

func sayHello() {
	fmt.Println("hello")
}

Concurrency & Goroutines

Goroutines execute concurrently. In this case, main() and sayHello() were running at the same time. main() finished and sayHello() never got a chance to run!


package main

import "fmt"

func main() {
	go sayHello()
}

func sayHello() {
	fmt.Println("hello")
}

To solve this problem you need to use some sort of communication, or synchronization device, to signal that a goroutine is complete.

We typically do that with channels, or a WaitGroup, depending on whether you need to return a value from the function you’re running.

Using the go keyword in front of a function only schedules the function to be run, it does not run it immediately.

WaitGroup

A WaitGroup is defined in the sync package in the standard library. They’re a simple, and powerful, way to ensure that all of the goroutines you launch have completed before you move on to do other things.

https://golang.org/pkg/sync/#WaitGroup

Understanding A WaitGroup

A WaitGroup is, essentially, an atomic counter, that can be safely accessed concurrently. A WaitGroup had three methods:

  • Add(i int)
  • Done()
  • Wait()

When a new WaitGroup is created its internal counter is set to 0.

WaitGroup.Add

To add more items to a WaitGroup we can use the Add(n int) method. You can call Add as many times as you need to.

You can add one item at a time:

wg.Add(1)

Or add more than one:

wg.Add(42)

The number you add will almost always be the total number of goroutines you intend to launch.

If you create a new WaitGroup, and add 5 to it:

wg.Add(5)

And then later on add 42:

wg.Add(42)

The internal counter of the waitgroup will now be a total of 47.

It is important to remember that while you can call Add as many times as needed, once Wait is called, Add should no longer be called. Doing so may create a race condition in your code and will result in potentially non-deterministic code.

WaitGroup.Done

To remove an item from the group, we can call the Done() method. This method will decrement the counter by 1. So, if the WaitGroup contains 42 items, calling wg.Done(), will set the counter to 41.

You will almost always call Done() from within the Go Routine you launched. This signals to your program that the Go Routine has finished.

WaitGroup.Wait

The Wait() function on a WaitGroup will block execution until the internal counter is 0. This is always done after all Add’s have been called.

The common flow of using a WaitGroup will look like this:

  1. Declare your WaitGroup
  2. Call Add as many times as necessary and launch corresponding goroutines
  3. Call Wait from your current routine
  4. Call Done from inside each of the launched goroutines from your second step

Using the methods of a WaitGroup in this order will help ensure proper code syncronization.

WaitGroup Example


package main

import (
	"fmt"
	"sync"
)

var wg sync.WaitGroup

func main() {
	wg.Add(1)     // add 1 to the internal counter
	go sayHello() // schedule this to run in a goroutine
	wg.Wait()     // block here until we decrement the internal counter to 0
}

func sayHello() {
	fmt.Println("hello")
	wg.Done() // decrement the internal counter
}

Using Error Groups

For most work, a channel-based solution is simpler and the usual right choice for handling errors across goroutines: you collect results and errors yourself and can act on the first failure. It’s still worth knowing errgroup, because you’ll run across it in the wild.

The golang.org/x/sync/errgroup package wraps a WaitGroup with error handling, at the cost of running every goroutine you launch to completion even after one fails.


func main() {
	var wg errgroup.Group
	for i := 0; i < 100; i++ {
		x := i
		wg.Go(func() error {
			randx.RandomSleep(50)
			fmt.Printf("processing %d\n", x)
			return randx.RandomError(42)
		})
	}
	if err := wg.Wait(); err != nil {
		log.Fatal(err)
	}
}

Error Groups Behavior

While using an error group may simplify your program’s ability to retrieve the error, it will only report the first error. You won’t be able to determine if you had other errors, or what they were. This should be taken into account when using the errgroup.Group.


func main() {
	var wg errgroup.Group
	for i := 0; i < 5; i++ {
		i := i // capture the range variable
		wg.Go(func() error {
			fmt.Printf("processing %d\n", i)
			time.Sleep(10 * time.Duration(i) * time.Millisecond)
			fmt.Printf("finished %d\n", i)
			return fmt.Errorf("routine %d error'd out", i)
		})
	}
	if err := wg.Wait(); err != nil {
		fmt.Println(err)
	}
}

Output:


processing 4
processing 2
processing 0
finished 0
processing 1
processing 3
finished 1
finished 2
finished 3
finished 4
routine 0 error'd out

Context With ErrorGroup

The errgroup.WithContext function creates a new errgroup.Group and returns a context.Context that is canceled when the first goroutine returns an error.


func main() {
	ctx := context.Background()
	wg, ctx := errgroup.WithContext(ctx)

	go func() {
		// wait for errgroup to return an error
		// and cancel its context
		<-ctx.Done()
		fmt.Println("error returned from errgroup", ctx.Err())
	}()

	for i := 0; i < 10; i++ {
		x := i
		wg.Go(func() error {
			return doWork(ctx, x)
		})
	}

	// wait for errgroup to finish/error
	if err := wg.Wait(); err != nil {
		log.Fatal(err)
	}
}

Cancellation is only a signal: each goroutine still has to check ctx.Done() itself to stop early. If your work doesn’t, it runs to completion regardless. That is often why a channel-based design ends up simpler.

ErrorGroup Vs. WaitGroup

The question of when to use which is a matter of personal preference, as well as, the task at hand.

An errgroup.Group takes away a lot of the granular control you have with a WaitGroup, but gives you a simpler API and built-in error retrieval. A WaitGroup provides total control, but you plumb in your own error handling.

In practice, when you need error handling across goroutines, a channel-based solution is usually the clearest option. Reach for errgroup when its all-or-nothing model already fits the task.

Ergonomic WaitGroup Patterns

WaitGroup.Go() Pattern

The standard sync.WaitGroup requires explicitly calling Add(), Done(), and launching goroutines separately. This can be error-prone:

  • Forgetting to call Add() before launching a goroutine
  • Forgetting to call Done() (or not using defer)
  • Calling Add() after Wait() (race condition)

As of Go 1.25, sync.WaitGroup has a Go method that combines these operations for you, so you no longer have to manage the counter by hand.

WaitGroup.Go (Go 1.25)

As of Go 1.25, sync.WaitGroup has a Go method that manages Add and Done for you. You no longer need to wrap it or touch the counter directly:


func main() {
	var wg sync.WaitGroup

	for i := range 5 {
		wg.Go(func() {
			fmt.Printf("Worker %d starting\n", i)
			time.Sleep(time.Duration(i*100) * time.Millisecond)
			fmt.Printf("Worker %d finished\n", i)
		})
	}

	wg.Wait()
	fmt.Println("All workers completed")
}

wg.Go(fn) adds one to the counter, launches fn in a new goroutine, and calls Done when fn returns. That removes the manual Add/Done bookkeeping those mistakes come from.

Note: Go does not recover panics. fn must not panic, or it will crash the program just like any other goroutine.

What `.Go()` Does For You

The method is conceptually equivalent to this small wrapper around sync.WaitGroup. There’s no magic, just Add/Done handled for you. You don’t need to write it anymore, but it’s worth seeing what it replaces:


// WaitGroup wraps sync.WaitGroup with a more ergonomic API
type WaitGroup struct {
	wg sync.WaitGroup
}

// Go launches a goroutine and automatically manages the waitgroup counter
func (w *WaitGroup) Go(fn func()) {
	w.wg.Add(1)
	go func() {
		defer w.wg.Done()
		fn()
	}()
}

// Wait blocks until all goroutines have finished
func (w *WaitGroup) Wait() {
	w.wg.Wait()
}

Comparison: Manual Vs WaitGroup.Go

Manual Add/Done (more verbose):


// Manual WaitGroup usage - you call Add and Done yourself
func standardWaitGroup() {
	var wg sync.WaitGroup

	for i := range 5 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			processItem(i)
		}()
	}

	wg.Wait()
}

WaitGroup.Go (more concise):


// WaitGroup.Go (Go 1.25) - Add and Done are handled for you
func ergonomicWaitGroup() {
	var wg sync.WaitGroup

	for i := range 5 {
		wg.Go(func() {
			processItem(i)
		})
	}

	wg.Wait()
}

Benefits Of The .Go() Pattern

Advantages:

  • Impossible to forget Add() - it’s called automatically
  • Impossible to forget Done() - defer is built-in
  • Less boilerplate - cleaner, more readable code
  • Harder to misuse - fewer opportunities for mistakes
  • Familiar API - similar to errgroup.Group.Go()

When to use:

  • Simple concurrent operations without error handling
  • When you want errgroup-like ergonomics without errors
  • Teaching/learning - helps prevent common mistakes

When To Still Reach For A Library

Now that sync.WaitGroup.Go covers the basic ergonomic case, a library earns its place only when it adds something the standard library doesn’t:

A channel-based solution - for most error-handling cases, collecting results and errors over a channel is simple, explicit, and usually the right choice. It also lets you act on the first failure and stop or cancel the remaining work, instead of paying for goroutines you no longer need.

golang.org/x/sync/errgroup - a convenient wrapper when its model fits: Go captures the first error, Wait returns it, and SetLimit bounds concurrency. Be aware it runs every goroutine you launch to completion even after one fails. WithContext gives you a cancellation signal, but each goroutine still has to check ctx.Done() itself to actually stop early, so it rarely saves you the work of a channel-based design.

sourcegraph/conc - adds ergonomics the standard library still lacks:

  • conc.WaitGroup - captures panics in child goroutines and re-raises them on Wait (or returns them from WaitAndRecover) instead of crashing the process; the native WaitGroup.Go does not
  • pool.Pool - bounded concurrency, with error- and result-collecting variants
  • iter.Map / iter.ForEach - parallel iteration over a slice (bounded by GOMAXPROCS, or configurable)
  • stream.Stream - run tasks concurrently while their results are handled in submission order

func main() {
	fmt.Println("=== conc.WaitGroup Example ===")

	var wg conc.WaitGroup

	for i := 0; i < 3; i++ {
		wg.Go(func() {
			fmt.Printf("worker %d starting\n", i)
			time.Sleep(time.Duration(i+1) * 50 * time.Millisecond)
			fmt.Printf("worker %d done\n", i)
		})
	}

	// conc.WaitGroup captures panics and returns them via WaitAndRecover.
	wg.Go(func() {
		fmt.Println("failing worker starting")
		time.Sleep(30 * time.Millisecond)
		panic("worker panic")
	})

	if rec := wg.WaitAndRecover(); rec != nil {
		fmt.Printf("recovered panic: %v\n", rec.Value)
	}

	fmt.Println("all tasks accounted for")
}

Rule of thumb: plain “run these and wait” reaches for the standard library WaitGroup.Go; handling errors across goroutines usually reaches for a channel; needing panic capture or structured parallel helpers reaches for conc. Note that conc is still pre-1.0 and has not shipped a release since v0.3.0 (2023), so weigh that before depending on it.

https://github.com/sourcegraph/conc

WaitGroup Pattern Summary

Pattern Use Case Error Handling
sync.WaitGroup (Add/Done) Full manual control Manual
sync.WaitGroup.Go (Go 1.25+) Simplest ergonomic API Manual
errgroup.Group Tasks that return errors Built-in ✓
conc.WaitGroup Panic capture, parallel helpers Panic capture ✓

Choose based on your needs:

  • Simple tasks → sync.WaitGroup.Go
  • Errors across goroutines → a channel (reach for errgroup only if its model fits)
  • Panic capture / structured helpers → sourcegraph/conc (pre-1.0; see note above)

One-Time Initialization With Sync.Once

Sync.Once Overview

The sync.Once type ensures that a function is executed exactly once, even when called from multiple goroutines concurrently.

This is particularly useful for:

  • Initializing shared resources (databases, caches, configs)
  • Implementing singleton patterns
  • Lazy initialization of expensive operations

In production Go, sync.Once is used as a field inside a struct, not as a package-level global variable. This keeps initialization scoped to the struct that owns it.

https://golang.org/pkg/sync/#Once

Sync.Once Basic Usage

sync.Once has a single method: Do(f func())

The function f will be called exactly once, no matter how many times Do is called. Use sync.Once as a struct field with a factory function:


type Config struct {
	once     sync.Once
	settings map[string]string
}

func NewConfig() *Config {
	return &Config{}
}

func (c *Config) Load() map[string]string {
	c.once.Do(func() {
		fmt.Println("Loading configuration...")
		c.settings = map[string]string{
			"database": "localhost:5432",
			"cache":    "redis:6379",
		}
		fmt.Println("Configuration loaded!")
	})
	return c.settings
}

Sync.Once Example

Here’s the complete example showing multiple goroutines all calling Load():


func main() {
	cfg := NewConfig()
	var wg sync.WaitGroup

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			settings := cfg.Load()
			fmt.Printf("Goroutine %d: database=%s\n", id, settings["database"])
		}(i)
	}

	wg.Wait()
	fmt.Println("All goroutines completed")
}

Output:


// Output (order will vary):
// Loading configuration...
// Configuration loaded!
// Goroutine 3: database=localhost:5432
// Goroutine 7: database=localhost:5432
// Goroutine 0: database=localhost:5432
// ...
// All goroutines completed

Notice that “Loading configuration…” is only printed once, even though 10 goroutines all called cfg.Load().

Database Connection With Sync.Once

A common use case for sync.Once is ensuring a database connection is established exactly once. The sync.Once lives as a field on the DB struct:


type DB struct {
	once      sync.Once
	conn      string
	connCount int
}

func New() *DB {
	return &DB{}
}

func (db *DB) Connect() {
	db.once.Do(func() {
		fmt.Println("Establishing database connection...")
		time.Sleep(100 * time.Millisecond)
		db.conn = "postgres://localhost:5432/mydb"
		db.connCount++
		fmt.Println("Database connected!")
	})
}

func (db *DB) Conn() string {
	return db.conn
}

Database Connection Example


func main() {
	db := New()
	var wg sync.WaitGroup

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			db.Connect()
			fmt.Printf("Goroutine %d: connections=%d\n", id, db.connCount)
		}(i)
	}

	wg.Wait()
	fmt.Printf("\nFinal check - Total connections: %d (should be 1)\n", db.connCount)
}

Output:


// Output:
// Establishing database connection...
// Database connected!
// Goroutine 0: connections=1
// Goroutine 9: connections=1
// ...
//
// Final check - Total connections: 1 (should be 1)

The database connection is established exactly once, regardless of how many goroutines call Connect().

Sync.Once Best Practices

Important considerations when using sync.Once:

  • Use sync.Once as a struct field, not a package-level global variable
  • Create structs with factory functions (NewConfig(), New()) that return the struct
  • The function passed to Do must not panic - if it panics, Do will consider it called and won’t retry
  • Once.Do is meant for initialization that cannot fail - for fallible initialization, use other patterns
  • Each sync.Once should only be used for a single initialization function
  • sync.Once is more efficient than mutex-based initialization checking

Code Walkthrough (Optional)

Now that we have a basic understanding of how a WaitGroup works, we can take a process that is slow and sequential, and make it fast and efficient.

What if We are asked to download the top 100 sites each day and monitor how long their response times are. We want this to process as fast as possible.

Given the following code, let’s work through the problem and use our new knowledge of concurrency:


package main

import (
	"log"
	"net/http"
)

func main() {
	args := []string{"http://youtube.com", "http://en.wikipedia.org", "http://facebook.com"}

	for _, a := range args {
		_, err := http.Get(a)
		if err != nil {
			log.Printf("failed to retrieve %s: %s\n", a, err)
			continue
		}
		log.Printf("Retrieved site %s\n", a)
	}
}

Refactor

To simplify the code, let’s go ahead and refactor the actual retrieving of the website into its own function:


package main

import (
	"log"
	"net/http"
)

func main() {
	args := []string{"http://youtube.com", "http://en.wikipedia.org", "http://facebook.com"}

	for _, a := range args {
		get(a)
	}
}

func get(s string) {
	_, err := http.Get(s)
	if err != nil {
		log.Printf("failed to retrieve %s: %s\n", s, err)
		return
	}
	log.Printf("Retrieved site %s\n", s)
}

Here is the output:


2019/10/09 16:54:12 Retrieved site http://youtube.com
2019/10/09 16:54:12 Retrieved site http://en.wikipedia.org
2019/10/09 16:54:13 Retrieved site http://facebook.com

Record Timing

Now that we have a basic structure, let’s record the time it takes for each site to be retrieved, as well as how much time it takes to run the entire process:


package main

import (
	"fmt"
	"log"
	"net/http"
	"time"
)

func main() {
	args := []string{"http://youtube.com", "http://en.wikipedia.org", "http://facebook.com"}

	now := time.Now()

	for _, a := range args {
		get(a)
	}

	fmt.Printf("It took %s for the entire process to finish\n", time.Since(now))
}

func get(s string) {
	now := time.Now()

	_, err := http.Get(s)
	if err != nil {
		log.Printf("failed to retrieve %s: %s\n", s, err)
		return
	}
	log.Printf("Retrieved site %s in %s\n", s, time.Since(now))
}

Here is the output:


2019/10/09 16:56:53 Retrieved site http://youtube.com in 662.923755ms
2019/10/09 16:56:54 Retrieved site http://en.wikipedia.org in 397.967866ms
2019/10/09 16:56:54 Retrieved site http://facebook.com in 459.329112ms
It took 1.520455229s for the entire process to finish

Because we are running everything sequentially, it takes longer than we would like to process these sites. Currently we are only doing 3 sites, so this problem will only get worse as we add more sites to process.

Add Concurrency

We are at a point now that we can add our concurrency primitives and make this code more efficient:


package main

import (
	"fmt"
	"log"
	"net/http"
	"sync"
	"time"
)

func main() {
	args := []string{"http://youtube.com", "http://en.wikipedia.org", "http://facebook.com"}

	wg := &sync.WaitGroup{}
	wg.Add(len(args))

	now := time.Now()

	for _, a := range args {
		go get(a, wg)
	}

	wg.Wait()

	fmt.Printf("It took %s for the entire process to finish\n", time.Since(now))
}

func get(s string, wg *sync.WaitGroup) {
	defer wg.Done()
	now := time.Now()

	_, err := http.Get(s)
	if err != nil {
		log.Printf("failed to retrieve %s: %s\n", s, err)
		return
	}
	log.Printf("Retrieved site %s in %s\n", s, time.Since(now))
}

Here is the output:


2019/10/09 17:02:40 Retrieved site http://en.wikipedia.org in 420.20124ms
2019/10/09 17:02:40 Retrieved site http://youtube.com in 474.16273ms
2019/10/09 17:02:41 Retrieved site http://facebook.com in 1.014454619s
It took 1.014520896s for the entire process to finish

Now the process only takes about as long as the longest site takes to retrieve. This is a great improvement.

Processing 100 Sites

Next, let’s process all the sites. We can use a list of the top 100 sites. Here is a sample of the list we will process:


http://youtube.com
http://en.wikipedia.org
http://facebook.com
http://twitter.com
http://amazon.com

First, we’ll have to get the arguments from the input now. We can use that by just asking for os.Args[1:]. We slice off the first argument, as that is always the name of the program that is running.

To process this file, I can use the following command:

go run main.go "$(< sites.txt)"


package main

import (
	"fmt"
	"log"
	"net/http"
	"os"
	"strings"
	"sync"
	"time"
)

func main() {
	args := os.Args[1:]
	// Args might be a single string with line returns to delimit sites, break them apart
	if len(args) == 1 {
		args = strings.Split(args[0], "\n")
	}

	wg := &sync.WaitGroup{}
	wg.Add(len(args))

	now := time.Now()

	for _, a := range args {
		go get(a, wg)
	}

	wg.Wait()

	fmt.Printf("It took %s for the entire process to finish\n", time.Since(now))
}

func get(s string, wg *sync.WaitGroup) {
	defer wg.Done()
	now := time.Now()

	_, err := http.Get(s)
	if err != nil {
		log.Printf("failed to retrieve %s: %s\n", s, err)
		return
	}
	log.Printf("Retrieved site %s in %s\n", s, time.Since(now))
}

Here is a sample of the output of all 100 sites:


2019/10/09 17:16:13 Retrieved site http://quizlet.com in 235.854997ms
2019/10/09 17:16:13 Retrieved site http://mail.yahoo.com in 330.833751ms
2019/10/09 17:16:13 Retrieved site http://go.com in 333.201946ms
2019/10/09 17:16:13 Retrieved site http://ufl.edu in 643.07295ms
2019/10/09 17:16:13 Retrieved site http://ign.com in 699.058387ms

2019/10/09 17:16:16 Retrieved site http://groupon.com in 3.339554149s
2019/10/09 17:16:16 Retrieved site http://medicalnewstoday.com in 3.559398518s
2019/10/09 17:16:16 Retrieved site http://craigslist.org in 3.676474407s
2019/10/09 17:16:25 Retrieved site http://washingtonpost.com in 12.931573753s
2019/10/09 17:26:20 failed to retrieve http://accuweather.com: Get http://accuweather.com: read tcp 192.168.1.36:59369->104.106.16.217:80: read: connection reset by peer
2019/10/09 17:26:20 failed to retrieve http://apartments.com: Get http://apartments.com: read tcp 192.168.1.36:59337->72.246.85.138:80: read: connection reset by peer
It took 10m8.077548804s for the entire process to finish

Outstanding Issues

While we were able to process all the files faster, we still have some issues we need to address to use this in a real production scenario:

  • Processing 100 sites may work, but what happens if we have 1,000, or 1,000,000? This solution won’t scale properly in that scenario. We’ll solve this scalability problem later in this module with semaphores.
  • Currently some sites don’t load and time out. The default timeout is 10 minutes, which we don’t want to wait for. We could control this with the http client, but using other concurrency primitives we can add our own timeout.
  • All the errors are happening inside the get function. Ideally we would want to pass these errors back to be properly handled upstream in our application.

WaitGroup Exercise (15 Min)

The following program currently works but could be improved by adding concurrency.

Make the following changes in the program:

  • Launch charCount in a go routine.
  • Use a sync.WaitGroup to track each time you launch a new go routine.
  • Make sure to wait for each go routine to finish before exiting the program.
  • Make sure to mark each go routine as Done.

Note: You can download this file and the files it uses with the Download Source button on the left.

Be sure to run the example from the same directory that main.go is in to ensure it can find the downloaded files.


package main

import (
	"bytes"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"time"
)

func main() {
	start := time.Now()

	root := "./files"
	err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
		if info != nil && !info.IsDir() {
			charCount(path)
		}
		return nil
	})
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("process took", time.Since(start))
}

func charCount(path string) {
	f, err := os.Open(path)
	if err != nil {
		fmt.Printf("error opening %s: %s\n", path, err)
		return
	}
	defer f.Close()
	var count int
	buff := make([]byte, bytes.MinRead)
	for {
		i, err := f.Read(buff)
		count += i
		if err == io.EOF {
			fmt.Printf("%s has %d characters\n", path, count)
			return
		}
	}
}

WaitGroup Solution


package main

import (
	"bytes"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"sync"
	"time"
)

func main() {
	start := time.Now()
	wg := &sync.WaitGroup{} // <- Add a waitgroup

	root := "./files"
	err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
		if info != nil && !info.IsDir() {
			wg.Add(1)              // <-- Track each new go routine
			go charCount(path, wg) // <-- Launch in a go routine, pass the WaitGroup
		}
		return nil
	})
	if err != nil {
		log.Fatal(err)
	}
	wg.Wait() // <-- Wait for all go routines to finish
	fmt.Println("process took", time.Since(start))
}

func charCount(path string, wg *sync.WaitGroup) { // <-- Accept the WaitGroup as an argument
	defer wg.Done() // <-- be sure to mark the go routine as done
	f, err := os.Open(path)
	if err != nil {
		fmt.Printf("error opening %s: %s\n", path, err)
		return
	}
	defer f.Close()
	var count int
	buff := make([]byte, bytes.MinRead)
	for {
		i, err := f.Read(buff)
		count += i
		if err == io.EOF {
			fmt.Printf("%s has %d characters\n", path, count)
			return
		}
	}
}

Protecting Data With A Mutex

Protecting Data With A Mutex

A Mutex allows you to synchronize goroutines for safe access to the same shared memory. Go provides two Mutex implementations, a full lock, and a read/write lock.

var m sync.Mutex // Make a new mutex

m.Lock() // lock the mutex

// Code in here that is protected by this mutex is safe to
// read or write for all goroutines observing this mutex.

m.Unlock() // unlock the mutex

https://golang.org/pkg/sync/#Mutex

https://golang.org/pkg/sync/#RWMutex

Running Without A Mutex

When trying to access a map concurrently, without a mutex, the app will panic and crash.

Here are two functions that access the same map:


var data = map[int]int{}

func get(i int) int {
	return data[i]
}

func set(i int) {
	data[i] = i
}

Call them concurrently without a mutex to synchronize access:


func main() {
	var wg sync.WaitGroup

	for i := 0; i < 100; i++ {
		wg.Add(2)
		go func(i int) {
			defer wg.Done()
			set(i)
		}(i)
		go func(i int) {
			defer wg.Done()
			fmt.Println(get(i))
		}(i)
	}
	wg.Wait()
}

If Go detects access to the same memory in different goroutines (specifically in a map), it will panic:

fatal error: 9
concurrent map writes

goroutine 54 [running]:
runtime.throw(0x10c4c4b, 0x15)
/usr/local/go/src/examples/runtime/panic.go:608 +0x72 fp=0xc0000b4760 sp=0xc0000b4730 pc=0x1027092
runtime.mapassign_fast64(0x10a7f00, 0xc000078120, 0x12, 0x0)

Synchronizing With A Mutex

By using a Mutex in the get and set functions that wrap the map, we are able to synchronize access to the map and prevent a panic in the application.


var data = map[int]int{}
var mu sync.Mutex

func get(i int) int {
	mu.Lock()
	defer mu.Unlock()
	return data[i]
}

func set(i int) {
	mu.Lock()
	defer mu.Unlock()
	data[i] = i
}

Synchronizing With A RWMutex

We can make use of a RWMutex as well so that we can have multiple read locks open allowing for less lock contention in a read heavy application:


var data = map[int]int{}
var mu sync.RWMutex

func get(i int) int {
	mu.RLock()
	defer mu.RUnlock()
	return data[i]
}

func set(i int) {
	mu.Lock()
	defer mu.Unlock()
	data[i] = i
}

Embedding A Mutex

You can embed a mutex in your own struct to synchronize access to internal data:


type Hits struct {
	sync.Mutex // embedding will promote all public methods
	n          int
}

func (h *Hits) Inc() {
	h.Lock()
	defer h.Unlock()
	h.n++
}

However, it’s more common to use a mutex as a field to prevent method promotion and ensure that only your struct can call the methods on the mutex.


type Hits struct {
	mu sync.Mutex // methods are no longer promoted and only accessible within this package now.
	n  int
}

func (h *Hits) Inc() {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.n++
}

Mutex Lock Order

Mutexes can run in two modes: normal and starvation.

  • In normal mode, waiters are queued in FIFO order.
  • In starvation mode, ownership of the mutex is directly handed off from the unlocking goroutine to the waiter at the front of the queue.

From the source:

// Mutex fairness.
//
// Mutex can be in 2 modes of operations: normal and starvation.
// In normal mode waiters are queued in FIFO order, but a woken up waiter
// does not own the mutex and competes with new arriving goroutines over
// the ownership. New arriving goroutines have an advantage -- they are
// already running on CPU and there can be lots of them, so a woken up
// waiter has good chances of losing. In such case it is queued at front
// of the wait queue. If a waiter fails to acquire the mutex for more than 1ms,
// it switches mutex to the starvation mode.
//
// In starvation mode ownership of the mutex is directly handed off from
// the unlocking goroutine to the waiter at the front of the queue.
// New arriving goroutines don't try to acquire the mutex even if it appears
// to be unlocked, and don't try to spin. Instead they queue themselves at
// the tail of the wait queue.
//
// If a waiter receives ownership of the mutex and sees that either
// (1) it is the last waiter in the queue, or (2) it waited for less than 1 ms,
// it switches mutex back to normal operation mode.
//
// Normal mode has considerably better performance as a goroutine can acquire
// a mutex several times in a row even if there are blocked waiters.
// Starvation mode is important to prevent pathological cases of tail latency.

Use Mutexes To Build A WaitGroup

Using a simple implementation of a WaitGroup we can demonstrate how a Mutex can help to prevent race conditions in our code.


type WaitGroup struct {
	quit    chan (struct{})
	counter int
}

func (wg *WaitGroup) Done() {
	wg.counter += -1
	if wg.counter <= 0 {
		if wg.quit != nil {
			close(wg.quit)
		}
	}
}

func (wg *WaitGroup) Add(n int) {
	wg.counter += n
}

func (wg *WaitGroup) Wait() {
	if wg.quit == nil {
		wg.quit = make(chan struct{})
	}
	<-wg.quit
}

Enabling Race Detection

Data races occur when multiple goroutines access the same shared memory. If you are writing concurrent code, you should always enable the race detector.

You can do this both via testing, or with the go run command.

To turn on the race detector, pass in the -race flag:

go run -race ./main.go

If a race is detected, it will both fail the test and output where the race was detected. The race detector makes one promise: it will never report a false positive. However, due to how the race detector is implemented, it will ONLY detect races that occur in the current running code.

Detecting Races

The current implementation of the waitgroup does not have a mutex to synchronize access to shared memory. This creates the potential for a data race.

We can see this by turning on the race detection with the -race flag:

go run -race ./broken.go

==================
WARNING: DATA RACE
Read at 0x00c000012058 by goroutine 9:
  main.(*WaitGroup).Done()
      main.go:16 +0x47
  main.main.func1()
      main.go:47 +0xcf

Previous write at 0x00c000012058 by main goroutine:
  main.main()
      main.go:25 +0xb3

Goroutine 9 (running) created at:
  main.main()
      main.go:43 +0xe1

Adding A Mutex

Adding a mutex to the WaitGroup will allow us to synchronize access to memory.


type WaitGroup struct {
	mu      sync.Mutex
	quit    chan (struct{})
	counter int
}

func (wg *WaitGroup) Done() {
	wg.mu.Lock()
	defer wg.mu.Unlock()
	wg.counter += -1
	if wg.counter <= 0 {
		if wg.quit != nil {
			close(wg.quit)
		}
	}
}

func (wg *WaitGroup) Add(n int) {
	wg.mu.Lock()
	defer wg.mu.Unlock()
	wg.counter += n
}

func (wg *WaitGroup) Wait() {
	wg.mu.Lock()
	if wg.quit == nil {
		wg.quit = make(chan struct{})
	}
	wg.mu.Unlock()
	<-wg.quit
}

Using A Read/write Mutex

Using sync.RWMutex we can lock access to data based on how that data is being used.

An sync.RWMutex is a reader/writer mutual exclusion lock. The lock can be held by an arbitrary number of readers or a single writer.


type WaitGroup struct {
	mu      sync.RWMutex
	quit    chan (struct{})
	counter int
}

func (wg *WaitGroup) Remaining() int {
	wg.mu.RLock()
	defer wg.mu.RUnlock()
	return wg.counter
}

Mutex Bug

The following code has a subtle bug that is common when using mutexes. Can you spot the bug?


func main() {
	var wg sync.WaitGroup
	wg.Add(1)
	var mu sync.Mutex

	var count int

	go func() {
		for i := 0; i < 5; i++ {
			mu.Lock()
			defer mu.Unlock()
			count = i
			fmt.Println(count)
		}
		wg.Done()
	}()

	wg.Wait()
}

Mutex Bug Solution

The defer keyword only executes once it exits the function it was called in. Because it is in a loop, the defer calls get put on the execution stack, but never gets called because the loop is running within the function scope.

In these scenarios, you will need to handle each unlock without the use of the defer keyword.


func main() {
	var wg sync.WaitGroup
	wg.Add(1)
	var mu sync.Mutex

	var count int

	go func() {
		for i := 0; i < 5; i++ {
			mu.Lock()
			count = i
			fmt.Println(count)
			mu.Unlock()
		}
		wg.Done()
	}()

	wg.Wait()
}

Mutex Bug Common Approach

It is much more common to launch many goroutines inside a for loop, than to launch a for loop in a goroutine.

The following is more idiomatic for handling the mutex with defers:


func main() {
	var wg sync.WaitGroup

	var mu sync.Mutex

	var count int

	wg.Add(5)
	for i := 0; i < 5; i++ {
		go func(i int) {
			mu.Lock()
			defer mu.Unlock()
			count = i
			fmt.Println(count)
			wg.Done()
		}(i)
	}

	wg.Wait()
}

Debugging Mutexes

On occasion, you may end up in a deadlock on a Mutex. They tend to be very difficult to track down.

The following packages can assist in debugging the offending lock. They are utilities you don’t need until you’ve done something totally wrong and desperately need help!

Debugging Our Bug

We can see that if we want to debug our offending code from the previous example, we can use the syncdebug package.

First, get the package:


go get go4.org/syncutil/syncdebug

Here is the code using the debug package:


func main() {
	var wg sync.WaitGroup
	wg.Add(1)
	var mu syncdebug.RWMutexTracker

	var count int

	go func() {
		for i := 0; i < 5; i++ {
			mu.Lock()
			defer mu.Unlock()
			count = i
			fmt.Println(count)
		}
		wg.Done()
	}()

	wg.Wait()
}

And here is the output when we run the program:


$ go run ./main.go
2020/06/04 08:46:26 Lock at goroutine 18 [running]:
go4.org/syncutil/syncdebug.(*RWMutexTracker).Lock(0xc0000ae180)
        /Users/corylanou/go/pkg/mod/go4.org@v0.0.0-20180809161055-417644f6feb5/syncutil/syncdebug/syncdebug.go:129 +0xf8
main.main.func1(0xc0000ae180, 0xc0000b2010, 0xc0000b2004)
        /Users/corylanou/projects/gopherguides/learn/_training/fundamentals/concurrency-sync/src/examples/mutex/bug/debug/main.go:20 +0xb2
created by main.main
        /Users/corylanou/projects/gopherguides/learn/_training/fundamentals/concurrency-sync/src/examples/mutex/bug/debug/main.go:18 +0xad
0
2020/06/04 08:46:27 Mutex 0xc0000ae180: waitW 1 haveW 1   waitR 0 haveR 0
2020/06/04 08:46:28 Mutex 0xc0000ae180: waitW 1 haveW 1   waitR 0 haveR 0
2020/06/04 08:46:29 Mutex 0xc0000ae180: waitW 1 haveW 1   waitR 0 haveR 0
2020/06/04 08:46:30 Mutex 0xc0000ae180: waitW 1 haveW 1   waitR 0 haveR 0
2020/06/04 08:46:31 Mutex 0xc0000ae180: waitW 1 haveW 1   waitR 0 haveR 0
2020/06/04 08:46:32 Mutex 0xc0000ae180: waitW 1 haveW 1   waitR 0 haveR 0
2020/06/04 08:46:33 Mutex 0xc0000ae180: waitW 1 haveW 1   waitR 0 haveR 0
2020/06/04 08:46:34 Mutex 0xc0000ae180: waitW 1 haveW 1   waitR 0 haveR 0
2020/06/04 08:46:35 Mutex 0xc0000ae180: waitW 1 haveW 1   waitR 0 haveR 0

Notice that it no longer deadlocks. This is because it starts up it’s own go routine to monitor how many locks are currently acquired and waiting.

Mutex Panics

Care must be taken to make sure not to try and unlock a Mutex that is not currently locked.


package main

import "sync"

func main() {
	var mu sync.Mutex
	mu.Unlock()
}
fatal error: sync: unlock of unlocked mutex

Advanced RWMutex Patterns

Let’s explore a practical example using RWMutex for tracking URL hits in a web application.

This demonstrates a read-heavy workload where RWMutex shines - many concurrent reads with occasional writes.


// Hits tracks URL hit counts with read-heavy access patterns
type Hits struct {
	mu     sync.RWMutex
	counts map[string]int
}

// NewHits creates a new Hits tracker
func NewHits() *Hits {
	return &Hits{
		counts: make(map[string]int),
	}
}

// Increment increments the hit count for a URL (write operation)
func (h *Hits) Increment(url string) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.counts[url]++
}

// Count returns the hit count for a URL (read operation)
func (h *Hits) Count(url string) int {
	h.mu.RLock()
	defer h.mu.RUnlock()
	return h.counts[url]
}

// Total returns the total number of hits across all URLs (read operation)
func (h *Hits) Total() int {
	h.mu.RLock()
	defer h.mu.RUnlock()
	total := 0
	for _, count := range h.counts {
		total += count
	}
	return total
}

// Stats returns a snapshot of all counts (read operation)
func (h *Hits) Stats() map[string]int {
	h.mu.RLock()
	defer h.mu.RUnlock()

	// Create a copy to avoid race conditions
	stats := make(map[string]int, len(h.counts))
	for url, count := range h.counts {
		stats[url] = count
	}
	return stats
}

RWMutex Hits Example

The Hits struct demonstrates several important patterns:

  • Write operations (Increment) use Lock()/Unlock()
  • Read operations (Count, Total, Stats) use RLock()/RUnlock()
  • Creating copies in Stats() prevents race conditions when data is used outside the lock
  • Multiple concurrent readers can execute simultaneously

func main() {
	hits := NewHits()
	var wg sync.WaitGroup

	urls := []string{
		"/api/users",
		"/api/products",
		"/api/orders",
	}

	// Simulate write operations (increment hits)
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(iteration int) {
			defer wg.Done()
			for _, url := range urls {
				hits.Increment(url)
				time.Sleep(10 * time.Millisecond)
			}
		}(i)
	}

	// Simulate read-heavy operations (checking counts)
	// Many concurrent reads can happen simultaneously with RWMutex
	for i := 0; i < 50; i++ {
		wg.Add(1)
		go func(iteration int) {
			defer wg.Done()
			for _, url := range urls {
				count := hits.Count(url)
				if iteration%10 == 0 {
					fmt.Printf("Read %d: %s has %d hits\n", iteration, url, count)
				}
			}
			time.Sleep(5 * time.Millisecond)
		}(i)
	}

	// Periodically check total
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func(iteration int) {
			defer wg.Done()
			time.Sleep(50 * time.Millisecond)
			total := hits.Total()
			fmt.Printf("Total hits at check %d: %d\n", iteration, total)
		}(i)
	}

	wg.Wait()

	// Final statistics
	fmt.Println("\n=== Final Statistics ===")
	stats := hits.Stats()
	for url, count := range stats {
		fmt.Printf("%s: %d hits\n", url, count)
	}
	fmt.Printf("Total: %d hits\n", hits.Total())
}

RWMutex Performance Benefits

When to use RWMutex over Mutex:

  • Read-heavy workloads (90%+ reads)
  • Expensive read operations that would benefit from concurrency
  • Large data structures where copying is expensive

When to stick with Mutex:

  • Write-heavy workloads
  • Simple operations where RWMutex overhead isn’t worth it
  • When lock contention is low

Benchmark rule of thumb: If you have more than 5-10 concurrent readers per writer, RWMutex usually wins.

Race Condition Detection And Resolution

Race conditions are one of the most common and dangerous bugs in concurrent programs. Let’s see how to detect and fix them.

Example: A Common Race Condition

This counter has a subtle but serious race condition:


// Counter has a race condition - multiple goroutines access count without synchronization
type Counter struct {
	count int
}

// Increment has a race condition
func (c *Counter) Increment() {
	// This is NOT atomic - it's actually three operations:
	// 1. Read c.count
	// 2. Add 1
	// 3. Write back to c.count
	c.count++
}

// Value returns the current count (also has race condition with Increment)
func (c *Counter) Value() int {
	return c.count
}

The count++ operation is not atomic - it’s actually three separate operations that can be interleaved between goroutines.

Detecting The Race

Run the broken example with Go’s race detector:

go run -race ./src/examples/race/broken/main.go

The race detector will report:


// Output when run WITHOUT -race flag:
// Expected: 100000
// Actual:   93547  (or some other number less than 100000)
//
// Run with: go run -race main.go
// To see the race detector find this bug!
//
// Output when run WITH -race flag:
// ==================
// WARNING: DATA RACE
// Read at 0x00c00001a0a8 by goroutine 8:
//   main.(*Counter).Increment()
//       /path/to/main.go:18 +0x3a
// ...

Important: The race detector only finds races that actually occur during execution. Always run your tests with -race flag!

Fixing Race Conditions

There are two main approaches to fix this race condition:

1. Using a Mutex (works for any shared state):


// CounterMutex uses a mutex to prevent race conditions
type CounterMutex struct {
	mu    sync.Mutex
	count int
}

// Increment is now safe for concurrent use
func (c *CounterMutex) Increment() {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.count++
}

// Value is now safe for concurrent use
func (c *CounterMutex) Value() int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.count
}

2. Using Atomic Operations (best for simple numeric operations):


// CounterAtomic uses atomic operations for better performance
type CounterAtomic struct {
	count int64
}

// Increment uses atomic operation - no mutex needed
func (c *CounterAtomic) Increment() {
	atomic.AddInt64(&c.count, 1)
}

// Value uses atomic load
func (c *CounterAtomic) Value() int64 {
	return atomic.LoadInt64(&c.count)
}

Race Resolution Best Practices

Preventing race conditions:

  • Always use -race flag when testing concurrent code
  • Document synchronization - note which mutex protects which data
  • Minimize shared state - prefer message passing (channels) when possible
  • Use atomic operations for simple counters and flags
  • Review carefully - concurrent code needs extra scrutiny

Performance considerations:

  • Atomic operations are faster than mutexes for simple values
  • Mutexes provide more flexibility for complex operations
  • RWMutex is best for read-heavy workloads
  • Profile before optimizing - correctness first, performance second

Mutex Exercise (15 Mins)

The code below has a race condition. Your tasks:

  • Run the code with the race detector to confirm the race: go run -race ./main.go
  • Identify which shared data needs protection
  • Add the appropriate mutex type to prevent the race
  • Verify your fix by running with -race again

package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

type Numbers struct {
	data map[int]int
}

func (n *Numbers) Find(key int) (int, error) {
	k, ok := n.data[key]
	if !ok {
		return 0, fmt.Errorf("number not found: %d", key)
	}
	return k, nil
}

func (n *Numbers) Load(count int) {
	for i := range count {
		n.data[i] = i
		randomSleep(75)
	}
}

func NewNumbers(count int) *Numbers {
	n := &Numbers{
		data: map[int]int{},
	}
	go n.Load(count)
	return n
}

var wg sync.WaitGroup

func main() {
	n := NewNumbers(1000)

	wg.Add(1)
	go func() {
		defer wg.Done()
		for i := range 100 {
			for {
				_, err := n.Find(i)
				if err == nil {
					fmt.Printf("found %d\n", i)
					break
				}
				fmt.Println(err)
				randomSleep(25)
			}
		}
	}()
	wg.Wait()
}


func randomSleep(i int) {
	time.Sleep(time.Duration(rand.Intn(i)) * time.Millisecond)
}

Mutex Solution


package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

type Numbers struct {
	mu   sync.RWMutex
	data map[int]int
}

func (n *Numbers) Find(key int) (int, error) {
	n.mu.RLock()
	defer n.mu.RUnlock()
	k, ok := n.data[key]
	if !ok {
		return 0, fmt.Errorf("number not found: %d", key)
	}
	return k, nil
}

func (n *Numbers) Load(count int) {
	for i := range count {
		n.mu.Lock()
		n.data[i] = i
		n.mu.Unlock()
		randomSleep(75)
	}
}

func NewNumbers(count int) *Numbers {
	n := &Numbers{
		data: map[int]int{},
	}
	go n.Load(count)
	return n
}

var wg sync.WaitGroup

func main() {
	n := NewNumbers(1000)

	wg.Add(1)
	go func() {
		defer wg.Done()
		for i := range 100 {
			for {
				_, err := n.Find(i)
				if err == nil {
					fmt.Printf("found %d\n", i)
					break
				}
				fmt.Println(err)
				randomSleep(25)
			}
		}
	}()
	wg.Wait()
}


func randomSleep(i int) {
	time.Sleep(time.Duration(rand.Intn(i)) * time.Millisecond)
}

Condition Variables With Sync.Cond

Sync.Cond Overview

A sync.Cond (condition variable) lets one or more goroutines wait until some shared condition becomes true, instead of busy-looping and re-checking it themselves.

A Cond is always paired with a mutex, and the flow is always the same:

  1. Lock the mutex and check the shared condition
  2. If it isn’t satisfied yet, call Wait - this parks the goroutine
  3. Another goroutine changes the condition and calls Signal or Broadcast
  4. A parked goroutine wakes, re-locks the mutex, and re-checks

Key methods:

  • Wait() - atomically unlocks the mutex and suspends the goroutine; re-locks before returning
  • Signal() - wakes one waiting goroutine
  • Broadcast() - wakes all waiting goroutines

https://golang.org/pkg/sync/#Cond

How Wait, Signal, And Broadcast Work

The part that trips people up is Wait. It does three things as one atomic step:

  1. Unlocks the mutex, so other goroutines can run and eventually change the condition
  2. Suspends the goroutine until something wakes it
  3. Re-locks the mutex before returning

So you must already hold the lock when you call Wait, and you’ll hold it again once Wait returns.

Always wait in a for loop, never an if:

for !ready {      // re-check the condition on every wake
    cond.Wait()
}

A woken goroutine has no guarantee the condition is now true. Broadcast wakes every waiter, and another goroutine may have already consumed whatever you were waiting for. Re-checking in a loop is the only safe way to continue.

Use Signal to wake a single waiter, and Broadcast when one change could satisfy many waiters at once (like a gate opening for everyone).

Sync.Cond Basic Example

Five workers wait at a “gate” until the main goroutine opens it. The Cond, its mutex, and the open flag all live on a struct - the same struct-field discipline we used for sync.Once, not package globals:


// Gate lets many goroutines wait until it is opened exactly once.
// The sync.Cond, the mutex it coordinates with, and the state it
// guards (open) all live on the struct - never as package globals,
// the same discipline we used for sync.Once.
type Gate struct {
	mu   sync.Mutex
	cond *sync.Cond
	open bool
}

func NewGate() *Gate {
	g := &Gate{}
	// A Cond is always created with the lock it protects.
	g.cond = sync.NewCond(&g.mu)
	return g
}

// Wait blocks the caller until the gate has been opened.
func (g *Gate) Wait(id int) {
	g.mu.Lock()
	defer g.mu.Unlock()

	// Re-check the condition in a for loop: a woken goroutine is not
	// guaranteed the gate is open, so we confirm it on every wake.
	for !g.open {
		fmt.Printf("Worker %d: waiting for the gate to open...\n", id)
		g.cond.Wait() // unlocks g.mu, sleeps, then re-locks g.mu on wake
	}
	fmt.Printf("Worker %d: gate is open, starting work!\n", id)
}

// Open opens the gate and wakes every goroutine waiting on it.
func (g *Gate) Open() {
	g.mu.Lock()
	g.open = true
	g.cond.Broadcast() // wake all waiters; Signal would wake only one
	g.mu.Unlock()
}

Each worker waits in a for !g.open loop; Open sets the flag and calls Broadcast to wake them all at once.

Sync.Cond Coordination Example


func main() {
	gate := NewGate()
	var wg sync.WaitGroup

	// Launch workers that will block until the gate opens.
	for i := 1; i <= 5; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			gate.Wait(id)

			time.Sleep(100 * time.Millisecond)
			fmt.Printf("Worker %d: work completed\n", id)
		}(i)
	}

	// Give the workers time to start waiting before we open the gate.
	time.Sleep(100 * time.Millisecond)

	fmt.Println("\nMain: opening the gate...")
	gate.Open()

	wg.Wait()
	fmt.Println("\nAll workers completed")
}

Output:


// Output:
// Worker 5: waiting for the gate to open...
// Worker 1: waiting for the gate to open...
// Worker 2: waiting for the gate to open...
// Worker 3: waiting for the gate to open...
// Worker 4: waiting for the gate to open...
//
// Main: opening the gate...
// Worker 5: gate is open, starting work!
// Worker 1: gate is open, starting work!
// Worker 2: gate is open, starting work!
// Worker 3: gate is open, starting work!
// Worker 4: gate is open, starting work!
// Worker 1: work completed
// Worker 2: work completed
// Worker 3: work completed
// Worker 4: work completed
// Worker 5: work completed
//
// All workers completed

Producer-Consumer With Sync.Cond

Here’s a thread-safe queue built on sync.Cond. Enqueue calls Signal to wake one waiting consumer; Dequeue blocks in a for len(q.items) == 0 loop until an item is available - the same re-check-in-a-loop rule from before:


// Queue is a thread-safe queue using sync.Cond
type Queue struct {
	mu    sync.Mutex
	cond  *sync.Cond
	items []string
}

// NewQueue creates a new queue
func NewQueue() *Queue {
	q := &Queue{
		items: make([]string, 0),
	}
	q.cond = sync.NewCond(&q.mu)
	return q
}

// Enqueue adds an item to the queue and signals waiting consumers
func (q *Queue) Enqueue(item string) {
	q.mu.Lock()
	defer q.mu.Unlock()

	q.items = append(q.items, item)
	fmt.Printf("Enqueued: %s (queue size: %d)\n", item, len(q.items))
	q.cond.Signal() // Wake up one waiting consumer
}

// Dequeue removes and returns an item from the queue
// Blocks if the queue is empty until an item is available
func (q *Queue) Dequeue() string {
	q.mu.Lock()
	defer q.mu.Unlock()

	// Wait while queue is empty
	for len(q.items) == 0 {
		fmt.Println("Queue empty, waiting for items...")
		q.cond.Wait()
	}

	// Get the first item
	item := q.items[0]
	q.items = q.items[1:]
	fmt.Printf("Dequeued: %s (queue size: %d)\n", item, len(q.items))
	return item
}

Queue Example Usage


func main() {
	q := NewQueue()
	var wg sync.WaitGroup

	// Start consumers first (they will wait)
	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for j := 0; j < 3; j++ {
				item := q.Dequeue()
				fmt.Printf("Consumer %d processed: %s\n", id, item)
				time.Sleep(50 * time.Millisecond)
			}
		}(i)
	}

	// Give consumers time to start waiting
	time.Sleep(100 * time.Millisecond)

	// Start producer
	wg.Add(1)
	go func() {
		defer wg.Done()
		for i := 1; i <= 9; i++ {
			item := fmt.Sprintf("item-%d", i)
			q.Enqueue(item)
			time.Sleep(100 * time.Millisecond)
		}
	}()

	wg.Wait()
	fmt.Println("\nAll producers and consumers completed")
}

Output:


// Output (order may vary):
// Queue empty, waiting for items...
// Queue empty, waiting for items...
// Queue empty, waiting for items...
// Enqueued: item-1 (queue size: 1)
// Dequeued: item-1 (queue size: 0)
// Consumer 1 processed: item-1
// Enqueued: item-2 (queue size: 1)
// Dequeued: item-2 (queue size: 0)
// Consumer 2 processed: item-2
// Enqueued: item-3 (queue size: 1)
// Dequeued: item-3 (queue size: 0)
// Consumer 3 processed: item-3
// ...

When To Use Sync.Cond

Use sync.Cond when:

  • Multiple goroutines need to wait for a specific condition
  • You need fine-grained control over waking specific goroutines
  • Implementing classical concurrency patterns (producer-consumer, barriers)

Consider channels instead when:

  • You need to pass data between goroutines (channels are more idiomatic)
  • You want simpler, more readable code
  • You don’t need the fine control that Cond provides

Note: Channels are generally preferred in Go for most synchronization needs. Use sync.Cond when you specifically need its unique capabilities.

Controlling Concurrency With Semaphores

Introduction To Semaphores

A semaphore is a synchronization mechanism that controls access to a common resource by limiting the number of goroutines that can access it simultaneously.

Think of it like a traffic light controlling the flow of cars - it prevents congestion by limiting how many vehicles can enter at once.

Key Purpose: Limit the number of goroutines running concurrently to prevent resource exhaustion.

Article: Controlling Concurrency in Go with the Semaphore Pattern

Solving The Scalability Problem

Earlier we saw our websites problem doesn’t scale well - launching 1,000,000 goroutines at once would be problematic.

The solution is to limit concurrent goroutines - process all sites, but only allow N to run simultaneously.

A semaphore gives us exactly that: it caps how many goroutines can run at once. Go’s golang.org/x/sync/semaphore package provides a ready-made implementation, which we’ll use throughout this section.

Why Use A Semaphore?

Bounding concurrency with a semaphore has several advantages:

  • Clear intent - the code reads as “limit concurrency to N”
  • One goroutine per task - no need to restructure work around a fixed set of workers
  • Standard pattern recognized by all Go developers
  • Easy to tune - the limit lives in one place

A semaphore makes your code’s purpose obvious at a glance: process everything, but only N at a time.

Real-World Scenarios

Semaphores are particularly useful for:

  • Rate limiting API requests (like our websites problem!)
  • Controlling database connection pools
  • Managing file I/O operations
  • Preventing resource exhaustion in high-load scenarios
  • Throttling concurrent downloads or uploads

Any time you need to limit concurrent operations, semaphores are the right tool.

The Golang.org/x/sync/semaphore Package

The standard library doesn’t ship a semaphore type, but the official package provides one. You create a semaphore with a capacity, then Acquire a slot before doing work and Release it after:


func main() {
	ctx := context.Background()
	maxConcurrent := int64(3)
	sem := semaphore.NewWeighted(maxConcurrent)

	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()

			// Acquire with context
			if err := sem.Acquire(ctx, 1); err != nil {
				fmt.Printf("worker %d: acquire failed: %v\n", id, err)
				return
			}
			defer sem.Release(1)

			fmt.Printf("worker %d running\n", id)
			time.Sleep(100 * time.Millisecond)
		}(i)
	}

	wg.Wait()
	fmt.Println("all workers done")
}

Benefits: Context support (an Acquire can be cancelled), weighted acquisition, and good composability.

https://pkg.go.dev/golang.org/x/sync/semaphore

Websites Problem With Semaphore Package

Using the official package, in the same go get(...) style as our earlier walkthrough:


func main() {
	ctx := context.Background()
	sites := []string{
		"https://golang.org",
		"https://google.com",
		"https://github.com",
	}

	sem := semaphore.NewWeighted(30) // max 30 concurrent

	wg := &sync.WaitGroup{}
	wg.Add(len(sites))

	for _, site := range sites {
		go get(ctx, site, wg, sem)
	}

	wg.Wait()
	fmt.Println("all requests finished")
}

func get(ctx context.Context, site string, wg *sync.WaitGroup, sem *semaphore.Weighted) {
	defer wg.Done()

	if err := sem.Acquire(ctx, 1); err != nil {
		log.Printf("failed to acquire semaphore for %s: %s\n", site, err)
		return
	}
	defer sem.Release(1)

	resp, err := http.Get(site)
	if err != nil {
		log.Printf("failed to retrieve %s: %s\n", site, err)
		return
	}
	defer resp.Body.Close()

	fmt.Printf("fetched %s: %d\n", site, resp.StatusCode)
}

Weighted Semaphores

Some operations “cost” more than others:


func fetchSiteWeighted(ctx context.Context, site string,
	sem *semaphore.Weighted) error {
	// Different sites "cost" different amounts
	weight := calculateWeight(site)

	if err := sem.Acquire(ctx, weight); err != nil {
		return err
	}
	defer sem.Release(weight)

	return fetch(site)
}

This allows for more sophisticated resource management.

Practical Patterns And Best Practices

Choosing the Right Limit


func main() {
	// CPU-bound tasks
	cpuBound := runtime.NumCPU()

	// I/O-bound tasks (like our websites!)
	ioBound := runtime.NumCPU() * 10

	// External API with rate limits
	apiRateLimit := 100
	requestsPerSecond := 10
	apiLimit := apiRateLimit / requestsPerSecond

	fmt.Printf("CPU-bound limit:      %d\n", cpuBound)
	fmt.Printf("I/O-bound limit:      %d\n", ioBound)
	fmt.Printf("API rate limit:       %d\n", apiLimit)
}

For websites: 30 concurrent requests is reasonable, but tune based on your network.

Error Handling With Semaphores

Always release semaphores, even on errors:


func processWithErrors(ctx context.Context, sem *semaphore.Weighted) error {
	if err := sem.Acquire(ctx, 1); err != nil {
		// Context cancelled or deadline exceeded
		return fmt.Errorf("semaphore acquire failed: %w", err)
	}
	defer sem.Release(1) // Always release!

	if err := doWork(); err != nil {
		return fmt.Errorf("work failed: %w", err)
	}

	return nil
}

Use defer to ensure semaphores are released.

Anti-Patterns To Avoid

Common mistakes when using semaphores:

  • ❌ Forgetting to release semaphore (use defer)
  • ❌ Acquiring inside tight loops (acquire outside)
  • ❌ Setting limits too low (underutilizing resources)
  • ❌ Setting limits too high (defeating the purpose)
  • ❌ Not using context for cancellation

Always use defer to release, and always support context cancellation.

Putting It All Together

Here’s the websites fetcher with everything applied: a semaphore capping concurrency at 30, the Acquire error checked, defer for Release, and a context.WithTimeout that bounds the entire run. If the deadline hits, goroutines still waiting in Acquire return an error instead of starting new work, and because each request is built with http.NewRequestWithContext, in-flight fetches are cancelled too:


package main

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"os"
	"strings"
	"sync"
	"time"

	"golang.org/x/sync/semaphore"
)

func main() {
	args := os.Args[1:]
	// Args might be a single string with line returns to delimit sites, break them apart
	if len(args) == 1 {
		args = strings.Split(args[0], "\n")
	}

	// Create a context with timeout for the entire operation
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	wg := &sync.WaitGroup{}
	wg.Add(len(args))

	// Create a semaphore to limit concurrent requests to 30
	sem := semaphore.NewWeighted(30)

	now := time.Now()

	for _, a := range args {
		go get(ctx, a, wg, sem)
	}

	wg.Wait()

	fmt.Printf("It took %s for the entire process to finish\n", time.Since(now))
}

func get(ctx context.Context, s string, wg *sync.WaitGroup, sem *semaphore.Weighted) {
	defer wg.Done()

	// Acquire semaphore
	if err := sem.Acquire(ctx, 1); err != nil {
		log.Printf("failed to acquire semaphore for %s: %s\n", s, err)
		return
	}
	defer sem.Release(1)

	now := time.Now()

	// Attach the context to the request so an in-flight fetch is also
	// cancelled when the deadline hits, not just goroutines waiting in Acquire.
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, s, nil)
	if err != nil {
		log.Printf("failed to build request for %s: %s\n", s, err)
		return
	}

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		log.Printf("failed to retrieve %s: %s\n", s, err)
		return
	}
	defer resp.Body.Close()

	log.Printf("Retrieved site %s in %s\n", s, time.Since(now))
}

Performance Considerations

Memory Usage

  • Each goroutine uses ~2KB stack (can grow)
  • 10,000 goroutines = ~20MB minimum
  • Semaphore overhead is minimal (just a counter)

For Websites Example:

  • 100 sites × 2KB = ~200KB goroutines
  • 30 concurrent semaphore = negligible overhead
  • Actual memory usage depends on HTTP client buffers

Always profile your application before optimization.

Tuning Guidelines

  1. Profile first - Don’t guess at optimization
  2. Start conservative - Increase gradually
  3. Monitor resources - CPU, memory, network
  4. Consider external limits - API rates, DB connections
  5. Test under load - Peak conditions matter

For Websites Example:

  • Tune based on network bandwidth
  • Monitor timeout rates
  • Adjust based on success/failure ratio
  • Consider target site rate limits

Semaphore Exercise (20 Min)

Take the websites fetcher from earlier and refactor it to throttle concurrency with golang.org/x/sync/semaphore.

Tasks:

  • Add the dependency to your module: go get golang.org/x/sync/semaphore
  • Use a weighted semaphore to limit concurrent requests to 30
  • Acquire a slot before each request and Release it with defer
  • Check the error returned by Acquire
  • Test with the provided sites.txt file

package main

import (
	"fmt"
	"log"
	"net/http"
	"os"
	"strings"
	"sync"
	"time"
)

// TODO: Use golang.org/x/sync/semaphore to limit concurrency to 30

func main() {
	args := os.Args[1:]
	if len(args) == 1 {
		args = strings.Split(args[0], "\n")
	}

	wg := &sync.WaitGroup{}
	wg.Add(len(args))

	now := time.Now()

	for _, a := range args {
		go get(a, wg)
	}

	wg.Wait()

	fmt.Printf("It took %s for the entire process to finish\n", time.Since(now))
}

func get(s string, wg *sync.WaitGroup) {
	defer wg.Done()

	now := time.Now()

	_, err := http.Get(s)
	if err != nil {
		log.Printf("failed to retrieve %s: %s\n", s, err)
		return
	}
	log.Printf("Retrieved site %s in %s\n", s, time.Since(now))
}

Semaphore Solution


package main

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"os"
	"strings"
	"sync"
	"time"

	"golang.org/x/sync/semaphore"
)

func main() {
	args := os.Args[1:]
	if len(args) == 1 {
		args = strings.Split(args[0], "\n")
	}

	wg := &sync.WaitGroup{}
	wg.Add(len(args))

	now := time.Now()

	sem := semaphore.NewWeighted(30)

	for _, a := range args {
		go get(a, wg, sem)
	}

	wg.Wait()

	fmt.Printf("It took %s for the entire process to finish\n", time.Since(now))
}

func get(s string, wg *sync.WaitGroup, sem *semaphore.Weighted) {
	defer wg.Done()

	if err := sem.Acquire(context.Background(), 1); err != nil {
		log.Printf("failed to acquire semaphore for %s: %s\n", s, err)
		return
	}
	defer sem.Release(1)

	now := time.Now()

	_, err := http.Get(s)
	if err != nil {
		log.Printf("failed to retrieve %s: %s\n", s, err)
		return
	}
	log.Printf("Retrieved site %s in %s\n", s, time.Since(now))
}