Concurrency

go fetch(url) // returns immediately; fetch runs on its own

var wg sync.WaitGroup
for _, url := range urls {
	wg.Add(1)          // one more to wait for
	go func() {
		defer wg.Done() // this one is finished
		fetch(url)
	}()
}
wg.Wait() // blocks until the counter is back to zero

ch := make(chan int)
go func() { ch <- 42 }() // send
v := <-ch                // receive: waits until a value arrives

v, ok := <-ch      // ok is false once the channel is closed and empty
for v := range ch { // loops until the channel is closed
	fmt.Println(v)
}

func produce(out chan<- int) { ... } // can only send
func consume(in <-chan int)  { ... } // can only receive

select {
case msg := <-results:
	fmt.Println(msg)
case <-time.After(500 * time.Millisecond):
	fmt.Println("timeout")
}

var mu sync.Mutex
mu.Lock()
count++
mu.Unlock()

package main
 
import (
	"fmt"
	"sync"
)
 
func player(name string, myTurn <-chan struct{}, next chan<- struct{}, out chan<- string, n int, wg *sync.WaitGroup) {
	defer wg.Done()
	for range n {
		<-myTurn           // wait until it is our turn
		out <- name        // blocks until the printer receives it (unbuffered)
		next <- struct{}{} // hand the turn to the other player
	}
}
 
func main() {
	out := make(chan string)
	pingTurn := make(chan struct{}, 1) // buffered, so the final hand-off never blocks
	pongTurn := make(chan struct{}, 1)
	pingTurn <- struct{}{} // ping goes first
 
	var wg sync.WaitGroup
	wg.Add(2)
	go player("ping", pingTurn, pongTurn, out, 3, &wg)
	go player("pong", pongTurn, pingTurn, out, 3, &wg)
	go func() {
		wg.Wait()
		close(out) // both players are done: nothing more will arrive
	}()
 
	for msg := range out {
		fmt.Println(msg)
	}
}

ping
pong
ping
pong
ping
pong

package main
 
import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)
 
func slowSquare(n int) int {
	time.Sleep(time.Duration(10-n) * time.Millisecond) // later jobs finish sooner
	return n * n
}
 
func main() {
	// 1. A timeout with select.
	result := make(chan int, 1)
	go func() { result <- slowSquare(1) }() // takes about 9ms
	select {
	case v := <-result:
		fmt.Println("got", v)
	case <-time.After(2 * time.Millisecond):
		fmt.Println("timed out waiting")
	}
 
	// 2. A pool of 3 workers processing 8 jobs.
	type job struct{ idx, n int }
	jobs := make(chan job)
	out := make([]int, 8)
	var wg sync.WaitGroup
	var busy, peak atomic.Int32
	for range 3 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := range jobs {
				now := busy.Add(1)
				for p := peak.Load(); now > p && !peak.CompareAndSwap(p, now); p = peak.Load() {
				}
				out[j.idx] = slowSquare(j.n) // each worker writes a different index: no race
				busy.Add(-1)
			}
		}()
	}
	for i := range 8 {
		jobs <- job{i, i + 1}
	}
	close(jobs)
	wg.Wait()
	fmt.Println("results:", out)
	fmt.Println("never more than 3 at once:", peak.Load() <= 3)
}

timed out waiting
results: [1 4 9 16 25 36 49 64]
never more than 3 at once: true

package main
 
import (
	"context"
	"fmt"
	"sync"
	"time"
)
 
// Stage 1: send 1..n, then close.
func generate(n int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for i := 1; i <= n; i++ {
			out <- i
		}
	}()
	return out
}
 
// Stage 2: a pool of workers squaring numbers.
func squareAll(in <-chan int, workers int) <-chan int {
	out := make(chan int, 10)
	var wg sync.WaitGroup
	for range workers {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for v := range in {
				out <- v * v
			}
		}()
	}
	go func() {
		wg.Wait()
		close(out) // only after every worker is done sending
	}()
	return out
}
 
func slowAnswer(ctx context.Context) (int, error) {
	select {
	case <-time.After(2 * time.Second):
		return 42, nil
	case <-ctx.Done():
		return 0, ctx.Err()
	}
}
 
func main() {
	// Pipeline + worker pool
	sum := 0
	for sq := range squareAll(generate(9), 3) {
		sum += sq
	}
	fmt.Println("sum of squares:", sum)
 
	// Mutex
	var mu sync.Mutex
	count := 0
	var wg sync.WaitGroup
	for range 1000 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			mu.Lock()
			count++
			mu.Unlock()
		}()
	}
	wg.Wait()
	fmt.Println("count:", count)
 
	// Buffered channel as a queue
	queue := make(chan string, 2)
	queue <- "a"
	queue <- "b" // no receiver needed: buffer has room
	fmt.Println(len(queue), cap(queue), <-queue, <-queue)
 
	// select with timeout and default
	results := make(chan string)
	select {
	case r := <-results:
		fmt.Println(r)
	default:
		fmt.Println("nothing ready yet")
	}
	go func() {
		time.Sleep(50 * time.Millisecond)
		results <- "fast result"
	}()
	select {
	case r := <-results:
		fmt.Println(r)
	case <-time.After(time.Second):
		fmt.Println("timeout")
	}
 
	// Cancellation with context
	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer cancel()
	_, err := slowAnswer(ctx)
	fmt.Println("slowAnswer:", err)
}

sum of squares: 285
count: 1000
2 2 a b
nothing ready yet
fast result
slowAnswer: context deadline exceeded