⚡ Chapter 15 of 20

Concurrency

Go's concurrency model — goroutines, channels, and the GMP scheduler — is one of its greatest strengths. The mantra: "Do not communicate by sharing memory; instead, share memory by communicating."

⏱ —📊 Advanced

🎯 Learning Objectives

✅ After this chapter you will:

Launch goroutines and understand the GMP scheduler
Use buffered and unbuffered channels
Multiplex channels with select
Understand and prevent data races
Use sync.WaitGroup, sync.Mutex, sync.Once
Implement the worker pool and fan-out/fan-in patterns
Use context for cancellation and timeouts

🔬 The GMP Scheduler

📊 GMP Model: Goroutines → Machines → Processors

G (Goroutines) — lightweight user-space threads (~2KB initial stack)
P (Processors)  — logical CPUs (GOMAXPROCS, defaults to physical core count)
M (Machines)    — OS threads (created/destroyed dynamically by the runtime)

  G G G G        G G G G G       (goroutines — millions possible)
  │ │ │ │        │ │ │ │ │
 ┌─────────┐    ┌─────────┐
 │  P (run │    │  P (run │      (P holds a local run queue of goroutines)
 │  queue) │    │  queue) │
 └────┬────┘    └────┬────┘
      │              │
      M              M           (M is an OS thread, pinned to a P)
      │              │
 [OS kernel]    [OS kernel]

When a goroutine blocks (syscall/channel):
  → The P detaches from M and attaches to another M
  → Unblocked goroutine rejoins a P's run queue
  → Result: no wasted OS threads while waiting for I/O!

package main

import (
    "fmt"
    "runtime"
    "sync"
    "time"
)

func main() {
    fmt.Println("CPUs:", runtime.NumCPU())
    fmt.Println("GOMAXPROCS:", runtime.GOMAXPROCS(0))  // 0 = query only

    // Goroutine — lightweight concurrency primitive
    // Stack starts at 2KB and grows/shrinks dynamically (up to 1GB by default)
    var wg sync.WaitGroup

    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(id int) {  // capture id by value — avoid closure variable capture bug!
            defer wg.Done()
            time.Sleep(time.Duration(id) * 10 * time.Millisecond)
            fmt.Printf("goroutine %d done\n", id)
        }(i)
    }

    wg.Wait()  // blocks until all goroutines call Done()
    fmt.Println("all goroutines finished")

    // Current goroutine count
    fmt.Println("goroutines:", runtime.NumGoroutine())
}

✅ Example Output (order may vary)

CPUs: 8 GOMAXPROCS: 8 goroutine 0 done goroutine 1 done goroutine 2 done goroutine 3 done goroutine 4 done all goroutines finished goroutines: 1

📡 Channels

Channels are typed conduits for communicating between goroutines. They provide synchronisation with communication.

package main

import (
    "fmt"
    "time"
)

func producer(ch chan<- int) {  // send-only channel
    for i := 1; i <= 5; i++ {
        fmt.Println("sending", i)
        ch <- i  // send blocks until receiver is ready (unbuffered)
        time.Sleep(50 * time.Millisecond)
    }
    close(ch)  // signal no more values — receivers see zero value after close
}

func consumer(ch <-chan int) {  // receive-only channel
    for v := range ch {  // range receives until channel is closed
        fmt.Println("received", v)
    }
}

func main() {
    // Unbuffered channel — every send blocks until there is a receive
    ch := make(chan int)  // capacity = 0
    go producer(ch)
    consumer(ch)

    fmt.Println("--- buffered channel ---")

    // Buffered channel — sends block only when buffer is full
    bch := make(chan string, 3)  // capacity = 3
    bch <- "a"  // doesn't block
    bch <- "b"  // doesn't block
    bch <- "c"  // doesn't block
    // bch <- "d"  // ← would block! buffer full
    fmt.Println("buffered:", len(bch), "/", cap(bch))

    fmt.Println(<-bch)  // a
    fmt.Println(<-bch)  // b
    fmt.Println(<-bch)  // c

    // Done channel pattern — signal completion
    done := make(chan struct{})
    go func() {
        time.Sleep(100 * time.Millisecond)
        fmt.Println("work complete")
        close(done)  // broadcast to all receivers
    }()
    <-done  // wait for signal
}

✅ Output

sending 1 received 1 sending 2 received 2 ... --- buffered channel --- buffered: 3 / 3 a b c work complete

🔀 Select — Multiplexing Channels

package main

import (
    "context"
    "fmt"
    "time"
)

func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)

    go func() { time.Sleep(100*time.Millisecond); ch1 <- "one" }()
    go func() { time.Sleep( 50*time.Millisecond); ch2 <- "two" }()

    // select waits for any case to be ready
    for i := 0; i < 2; i++ {
        select {
        case msg := <-ch1:
            fmt.Println("ch1:", msg)
        case msg := <-ch2:
            fmt.Println("ch2:", msg)
        }
    }

    // select with timeout
    ch := make(chan int)
    select {
    case v := <-ch:
        fmt.Println("received:", v)
    case <-time.After(200 * time.Millisecond):
        fmt.Println("timeout — no message received")
    }

    // select with context cancellation (production pattern)
    ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
    defer cancel()

    workCh := make(chan string)
    go func() {
        time.Sleep(300 * time.Millisecond)
        workCh <- "result"
    }()

    select {
    case result := <-workCh:
        fmt.Println("got result:", result)
    case <-ctx.Done():
        fmt.Println("context cancelled:", ctx.Err())
    }

    // Non-blocking select with default
    ch3 := make(chan int, 1)
    select {
    case v := <-ch3:
        fmt.Println("received:", v)
    default:
        fmt.Println("no message ready — non-blocking")
    }
}

✅ Output

ch2: two ch1: one timeout — no message received got result: result no message ready — non-blocking

⚠️ Data Races & Synchronisation

package main

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

func main() {
    // ❌ DATA RACE — counter is shared without synchronisation
    // Run with: go run -race main.go
    // counter := 0
    // var wg sync.WaitGroup
    // for i := 0; i < 1000; i++ {
    //     wg.Add(1)
    //     go func() { defer wg.Done(); counter++ }()  // RACE!
    // }
    // wg.Wait()

    // ✅ Fix 1: sync.Mutex
    var mu sync.Mutex
    counter := 0
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            mu.Lock()
            counter++
            mu.Unlock()
        }()
    }
    wg.Wait()
    fmt.Println("mutex counter:", counter)  // always 1000

    // ✅ Fix 2: atomic (fastest for simple counters)
    var atomicCounter int64
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            atomic.AddInt64(&atomicCounter, 1)  // lock-free, CPU instruction
        }()
    }
    wg.Wait()
    fmt.Println("atomic counter:", atomicCounter)  // always 1000

    // sync.Once — initialise exactly once (thread-safe singleton)
    var once sync.Once
    for i := 0; i < 5; i++ {
        once.Do(func() {
            fmt.Println("initialised exactly once")
        })
    }
}

✅ Output

mutex counter: 1000 atomic counter: 1000 initialised exactly once

🏭 Worker Pool Pattern

package main

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

type Job struct{ ID int }
type Result struct{ JobID, Value int }

func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        // Simulate work
        time.Sleep(10 * time.Millisecond)
        results <- Result{JobID: job.ID, Value: job.ID * job.ID}
        fmt.Printf("worker %d processed job %d\n", id, job.ID)
    }
}

func main() {
    const numWorkers = 3
    const numJobs = 9

    jobs := make(chan Job, numJobs)
    results := make(chan Result, numJobs)

    // Start worker pool
    var wg sync.WaitGroup
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }

    // Send all jobs
    for j := 1; j <= numJobs; j++ {
        jobs <- Job{ID: j}
    }
    close(jobs)  // signal workers: no more jobs

    // Close results channel when all workers are done
    go func() {
        wg.Wait()
        close(results)
    }()

    // Collect results
    total := 0
    for r := range results {
        total += r.Value
    }
    fmt.Printf("Sum of squares 1..9 = %d\n", total)  // 285
}

✅ Example Output

worker 3 processed job 1 worker 1 processed job 2 worker 2 processed job 3 ... Sum of squares 1..9 = 285

🎤 Interview Questions

What is a goroutine? How does it differ from an OS thread?

▶

Detailed Answer

A goroutine is a lightweight concurrently executing function managed by the Go runtime. Key differences from OS threads:

Memory: Goroutine starts at ~2KB stack (grows dynamically up to 1GB); OS thread typically uses 1–8MB fixed stack.
Creation cost: Goroutines are cheap (~microseconds); OS threads are expensive (~milliseconds).
Scheduling: Goroutines are scheduled by Go's GMP scheduler in user space; OS threads are scheduled by the kernel.
Multiplexing: Thousands of goroutines run on a handful of OS threads via the GMP model.
Blocking: When a goroutine blocks on I/O, the runtime parks it and runs another goroutine on the same OS thread. OS threads block and waste the kernel thread.

A typical Go server can handle 100,000+ concurrent goroutines on commodity hardware.

What is a data race? How do you detect and fix it?

▶

Detailed Answer

A data race occurs when two goroutines access the same memory location concurrently AND at least one access is a write, without synchronisation. The Go memory model guarantees that reads and writes to shared memory must be synchronised — otherwise results are undefined.

Detection: Run with go run -race or go test -race. The built-in race detector instruments memory accesses at compile time and reports races at runtime with the full goroutine stack trace.

Fixes: (1) sync.Mutex: protect shared data with Lock/Unlock; (2) sync/atomic: for simple counters and flags; (3) Channels: communicate rather than share — pass ownership of data through a channel; (4) sync.RWMutex: multiple readers, exclusive writer.

What happens when you send to a closed channel? Receive from a closed channel?

▶

Detailed Answer

Send to closed channel: Panics immediately — "send on closed channel". Never send to a channel you don't own or that may be closed by another goroutine. The producer should be the one who closes channels.

Receive from closed channel: Always succeeds immediately, returning the zero value of the element type and false for the ok flag: v, ok := <-ch. With range, the loop terminates cleanly when the channel is closed. This is the primary mechanism for signalling that a producer is done.

Rule: Only close a channel from the sender side. Never close from the receiver side. Use a done channel or context to signal cancellation in the opposite direction.

📌 Cheat Sheet

📌 Concurrency — Quick Reference

Goroutine

go func() { ... }()

Channel (unbuf)

ch := make(chan T)

Channel (buf)

ch := make(chan T, cap)

Send

ch <- val

Receive

v := <-ch

close(ch) // from sender only

Range channel

for v := range ch { }

Select

select { case v:=<-ch: ... }

WaitGroup

wg.Add(1); go...; wg.Done(); wg.Wait()

Mutex

mu.Lock(); defer mu.Unlock()

Atomic

atomic.AddInt64(&n, 1)

Race detect

go run -race / go test -race

🎯 Learning Objectives

🔬 The GMP Scheduler

📡 Channels

🔀 Select — Multiplexing Channels

⚠️ Data Races & Synchronisation

🏭 Worker Pool Pattern

🎤 Interview Questions

🧪 Quiz

📌 Cheat Sheet