go-concurrency.md

Go Concurrency

Concurrency is one of Go's most powerful features, built into the language from the ground up. Go provides goroutines and channels as first-class language constructs that make concurrent programming simple and efficient. Understanding Go's concurrency patterns is essential for building high-performance, scalable applications.

What is Concurrency

Concurrency is the ability to handle multiple tasks by switching between them, not necessarily executing them simultaneously. Go's concurrency model is based on Communicating Sequential Processes (CSP), which emphasizes passing messages between independent processes rather than sharing memory.

graph TD
    A[Main Program] --> B[Goroutine 1]
    A --> C[Goroutine 2]
    A --> D[Goroutine 3]
    B <-.Channel.-> C
    C <-.Channel.-> D
    B <-.Channel.-> D

Goroutines

Goroutines are lightweight threads managed by the Go runtime. They are much cheaper than operating system threads - thousands of goroutines can run concurrently in a single program.

Creating Goroutines

package main

import (
    "fmt"
    "time"
)

func sayHello() {
    fmt.Println("Hello from goroutine")
}

func main() {
    // Start goroutine
    go sayHello()
    
    // Start anonymous function as goroutine
    go func() {
        fmt.Println("Anonymous goroutine")
    }()
    
    // Main must wait or goroutines won't execute
    time.Sleep(time.Second)
    fmt.Println("Main function")
}

Goroutine Communication

func printNumbers() {
    for i := 1; i <= 5; i++ {
        time.Sleep(100 * time.Millisecond)
        fmt.Printf("%d ", i)
    }
}

func printLetters() {
    for i := 'a'; i < 'f'; i++ {
        time.Sleep(150 * time.Millisecond)
        fmt.Printf("%c ", i)
    }
}

func main() {
    go printNumbers()
    go printLetters()
    
    time.Sleep(2 * time.Second)
    fmt.Println("\nDone")
}

sequenceDiagram
    participant Main
    participant Goroutine1
    participant Goroutine2

    Main->>Goroutine1: Start
    Main->>Goroutine2: Start
    Goroutine1->>Goroutine1: Execute
    Goroutine2->>Goroutine2: Execute
    Main->>Main: Continue

Channels

Channels are typed conduits for communication between goroutines. They provide a safe way to send and receive data.

Creating and Using Channels

// Create channel
ch := make(chan int)

// Send to channel
ch <- 42

// Receive from channel
value := <-ch

// Buffered channel
buffered := make(chan int, 3) // capacity of 3

Unbuffered Channels

Unbuffered channels block until both sender and receiver are ready.

func sendData(ch chan int) {
    fmt.Println("Sending data...")
    ch <- 42
    fmt.Println("Data sent")
}

func main() {
    ch := make(chan int)
    
    go sendData(ch)
    
    fmt.Println("Waiting for data...")
    data := <-ch
    fmt.Println("Received:", data)
}

Buffered Channels

Buffered channels allow sending multiple values without blocking until the buffer is full.

func main() {
    ch := make(chan int, 2) // buffer size 2
    
    ch <- 1 // doesn't block
    ch <- 2 // doesn't block
    // ch <- 3 // would block until someone receives
    
    fmt.Println(<-ch) // 1
    fmt.Println(<-ch) // 2
}

Channel Direction

Specify channel direction in function parameters for type safety.

// Send-only channel
func sendOnly(ch chan<- int) {
    ch <- 42
}

// Receive-only channel
func receiveOnly(ch <-chan int) {
    value := <-ch
    fmt.Println(value)
}

func main() {
    ch := make(chan int)
    
    go sendOnly(ch)
    receiveOnly(ch)
}

Channel Operations

Closing Channels

func producer(ch chan int) {
    for i := 0; i < 5; i++ {
        ch <- i
    }
    close(ch) // Signal no more values
}

func main() {
    ch := make(chan int)
    
    go producer(ch)
    
    // Range over channel until closed
    for value := range ch {
        fmt.Println(value)
    }
    
    // Check if channel is closed
    value, ok := <-ch
    if !ok {
        fmt.Println("Channel closed")
    }
}

Select Statement

select allows waiting on multiple channel operations.

func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)
    
    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- "from ch1"
    }()
    
    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "from ch2"
    }()
    
    // Wait for whichever is ready first
    select {
    case msg1 := <-ch1:
        fmt.Println(msg1)
    case msg2 := <-ch2:
        fmt.Println(msg2)
    }
    
    // Select with default (non-blocking)
    select {
    case msg := <-ch1:
        fmt.Println(msg)
    default:
        fmt.Println("No data available")
    }
}

graph TD
    A[Select Statement] --> B{Channel 1 Ready?}
    A --> C{Channel 2 Ready?}
    A --> D{Default Case?}
    B -->|Yes| E[Receive from Ch1]
    C -->|Yes| F[Receive from Ch2]
    D -->|None Ready| G[Execute Default]

Common Concurrency Patterns

Worker Pool Pattern

func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d processing job %d\n", id, job)
        time.Sleep(time.Second)
        results <- job * 2
    }
}

func main() {
    jobs := make(chan int, 10)
    results := make(chan int, 10)
    
    // Start 3 workers
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }
    
    // Send 5 jobs
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    close(jobs)
    
    // Collect results
    for a := 1; a <= 5; a++ {
        <-results
    }
}

Fan-Out, Fan-In Pattern

// Fan-out: distribute work to multiple goroutines
func fanOut(input <-chan int, workers int) []<-chan int {
    channels := make([]<-chan int, workers)
    
    for i := 0; i < workers; i++ {
        ch := make(chan int)
        channels[i] = ch
        
        go func(out chan<- int) {
            for value := range input {
                out <- value * 2
            }
            close(out)
        }(ch)
    }
    
    return channels
}

// Fan-in: combine results from multiple channels
func fanIn(channels ...<-chan int) <-chan int {
    out := make(chan int)
    
    for _, ch := range channels {
        go func(c <-chan int) {
            for value := range c {
                out <- value
            }
        }(ch)
    }
    
    return out
}

Pipeline Pattern

// Stage 1: Generate numbers
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

// Stage 2: Square numbers
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

// Stage 3: Print numbers
func print(in <-chan int) {
    for n := range in {
        fmt.Println(n)
    }
}

func main() {
    // Pipeline: generate -> square -> print
    nums := generate(2, 3, 4, 5)
    squared := square(nums)
    print(squared)
}

graph LR
    A[Generate] -->|Channel| B[Square]
    B -->|Channel| C[Print]

Synchronization Primitives

WaitGroup

WaitGroup waits for a collection of goroutines to finish.

import "sync"

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done() // Decrement counter when done
    
    fmt.Printf("Worker %d starting\n", id)
    time.Sleep(time.Second)
    fmt.Printf("Worker %d done\n", id)
}

func main() {
    var wg sync.WaitGroup
    
    for i := 1; i <= 5; i++ {
        wg.Add(1) // Increment counter
        go worker(i, &wg)
    }
    
    wg.Wait() // Block until counter is 0
    fmt.Println("All workers done")
}

Mutex

Mutex provides exclusive access to shared data.

import "sync"

type Counter struct {
    mu    sync.Mutex
    value int
}

func (c *Counter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.value++
}

func (c *Counter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.value
}

func main() {
    counter := &Counter{}
    var wg sync.WaitGroup
    
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter.Increment()
        }()
    }
    
    wg.Wait()
    fmt.Println("Counter:", counter.Value())
}

RWMutex

RWMutex allows multiple readers or one writer.

import "sync"

type Cache struct {
    mu    sync.RWMutex
    data  map[string]string
}

func (c *Cache) Get(key string) string {
    c.mu.RLock() // Read lock
    defer c.mu.RUnlock()
    return c.data[key]
}

func (c *Cache) Set(key, value string) {
    c.mu.Lock() // Write lock
    defer c.mu.Unlock()
    c.data[key] = value
}

Context Package

Context carries deadlines, cancellation signals, and request-scoped values.

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

func operation(ctx context.Context) error {
    // Check for cancellation
    select {
    case <-ctx.Done():
        return ctx.Err()
    case <-time.After(2 * time.Second):
        fmt.Println("Operation complete")
        return nil
    }
}

func main() {
    // Context with timeout
    ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
    defer cancel()
    
    if err := operation(ctx); err != nil {
        fmt.Println("Error:", err)
    }
    
    // Context with cancellation
    ctx2, cancel2 := context.WithCancel(context.Background())
    go func() {
        time.Sleep(500 * time.Millisecond)
        cancel2() // Cancel the context
    }()
    
    if err := operation(ctx2); err != nil {
        fmt.Println("Cancelled:", err)
    }
}

sequenceDiagram
    participant Main
    participant Context
    participant Operation

    Main->>Context: Create with timeout
    Main->>Operation: Start (pass context)
    Operation->>Context: Check Done channel
    Context->>Operation: Timeout signal
    Operation->>Main: Return error

Rate Limiting

import (
    "fmt"
    "time"
)

func main() {
    requests := make(chan int, 5)
    
    // Fill requests
    for i := 1; i <= 5; i++ {
        requests <- i
    }
    close(requests)
    
    // Rate limiter: 1 per 200ms
    limiter := time.Tick(200 * time.Millisecond)
    
    for req := range requests {
        <-limiter // Wait for tick
        fmt.Println("Request", req, time.Now())
    }
}

// Bursty rate limiter
func burstLimiter() {
    requests := make(chan int, 5)
    for i := 1; i <= 5; i++ {
        requests <- i
    }
    close(requests)
    
    // Allow bursts of 3
    burstyLimiter := make(chan time.Time, 3)
    for i := 0; i < 3; i++ {
        burstyLimiter <- time.Now()
    }
    
    // Refill one per 200ms
    go func() {
        for t := range time.Tick(200 * time.Millisecond) {
            burstyLimiter <- t
        }
    }()
    
    for req := range requests {
        <-burstyLimiter
        fmt.Println("Request", req, time.Now())
    }
}

Practical Example: Concurrent Web Scraper

package main

import (
    "fmt"
    "io/ioutil"
    "net/http"
    "sync"
    "time"
)

type Result struct {
    URL    string
    Status int
    Size   int
    Error  error
}

func fetchURL(url string, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    
    client := http.Client{
        Timeout: 10 * time.Second,
    }
    
    resp, err := client.Get(url)
    if err != nil {
        results <- Result{URL: url, Error: err}
        return
    }
    defer resp.Body.Close()
    
    body, err := ioutil.ReadAll(resp.Body)
    if err != nil {
        results <- Result{URL: url, Error: err}
        return
    }
    
    results <- Result{
        URL:    url,
        Status: resp.StatusCode,
        Size:   len(body),
    }
}

func main() {
    urls := []string{
        "https://golang.org",
        "https://github.com",
        "https://stackoverflow.com",
    }
    
    results := make(chan Result, len(urls))
    var wg sync.WaitGroup
    
    // Start workers
    for _, url := range urls {
        wg.Add(1)
        go fetchURL(url, results, &wg)
    }
    
    // Wait and close results
    go func() {
        wg.Wait()
        close(results)
    }()
    
    // Collect results
    for result := range results {
        if result.Error != nil {
            fmt.Printf("%s: Error - %v\n", result.URL, result.Error)
        } else {
            fmt.Printf("%s: Status %d, Size %d bytes\n",
                result.URL, result.Status, result.Size)
        }
    }
}

Key Takeaways

Goroutines are lightweight threads that enable concurrent execution with minimal overhead
Channels provide type-safe communication between goroutines following CSP principles
Unbuffered channels block until both sender and receiver are ready
Buffered channels allow asynchronous communication up to the buffer capacity
The select statement enables multiplexing across multiple channel operations
Worker pool pattern distributes work efficiently across a fixed number of goroutines
Pipeline pattern chains processing stages using channels
WaitGroup synchronizes multiple goroutines waiting for completion
Mutex and RWMutex protect shared data from race conditions
Context enables cancellation, timeouts, and request-scoped values
Rate limiting controls the frequency of operations
Always close channels when done sending to signal completion to receivers
Prefer channels for communication, mutexes for protecting shared state