Code snippets — Go Programming Concurrency

snippet-10.go

// WRONG: panics if close is called twice
close(ch)
close(ch)  // panic: close of closed channel

// RIGHT: use sync.Once
var once sync.Once
safeClose := func() { once.Do(func() { close(ch) }) }

snippet-2.go

// Buffered channel
ch := make(chan int, 10)  // Buffer of 10

// Producer
go func() {
    for i := 0; i < 5; i++ {
        ch <- i  // Send
    }
    close(ch)  // Signal done
}()

// Consumer
for val := range ch {  // Receive until closed
    fmt.Println(val)
}

snippet-3.go

func WorkerPool(jobs <-chan Job, numWorkers int) <-chan Result {
    results := make(chan Result, len(jobs))

    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for job := range jobs {
                result := processJob(job)
                results <- result
            }
        }()
    }

    go func() {
        wg.Wait()
        close(results)
    }()

    return results
}

// Usage
jobs := make(chan Job, 100)
results := WorkerPool(jobs, 10)  // 10 concurrent workers

// Feed jobs
go func() {
    for _, j := range myJobs {
        jobs <- j
    }
    close(jobs)
}()

// Collect results
for r := range results {
    handleResult(r)
}

snippet-4.go

func fanOut(input <-chan Work, n int) []<-chan Result {
    channels := make([]<-chan Result, n)
    for i := 0; i < n; i++ {
        channels[i] = worker(input)
    }
    return channels
}

func fanIn(channels ...<-chan Result) <-chan Result {
    merged := make(chan Result)
    var wg sync.WaitGroup

    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan Result) {
            defer wg.Done()
            for result := range c {
                merged <- result
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(merged)
    }()

    return merged
}

snippet-5.go

func fetchUserData(ctx context.Context, userID string) (*UserData, error) {
    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()

    // All downstream calls get the context
    user, err := userRepo.Get(ctx, userID)
    if err != nil {
        return nil, err
    }

    orders, err := orderRepo.GetByUser(ctx, userID)
    if err != nil {
        return nil, err
    }

    return &UserData{User: user, Orders: orders}, nil
}

// HTTP handler passes context from request
func handler(w http.ResponseWriter, r *http.Request) {
    data, err := fetchUserData(r.Context(), r.URL.Query().Get("id"))
    if err != nil {
        if errors.Is(err, context.DeadlineExceeded) {
            http.Error(w, "Timeout", http.StatusGatewayTimeout)
            return
        }
        http.Error(w, "Error", http.StatusInternalServerError)
        return
    }
    json.NewEncoder(w).Encode(data)
}

snippet-6.go

import "golang.org/x/sync/errgroup"

func loadDashboard(ctx context.Context, userID string) (*Dashboard, error) {
    g, ctx := errgroup.WithContext(ctx)

    var user *User
    var orders []Order
    var notifications []Notification

    g.Go(func() error {
        var err error
        user, err = userRepo.Get(ctx, userID)
        return err
    })

    g.Go(func() error {
        var err error
        orders, err = orderRepo.GetRecent(ctx, userID, 10)
        return err
    })

    g.Go(func() error {
        var err error
        notifications, err = notifRepo.GetUnread(ctx, userID)
        return err
    })

    if err := g.Wait(); err != nil {
        return nil, err
    }

    return &Dashboard{
        User:          user,
        RecentOrders:  orders,
        Notifications: notifications,
    }, nil
}

snippet-7.go

type SafeMap struct {
    mu sync.RWMutex
    m  map[string]string
}

func (sm *SafeMap) Get(key string) (string, bool) {
    sm.mu.RLock()   // Multiple readers allowed simultaneously
    defer sm.mu.RUnlock()
    val, ok := sm.m[key]
    return val, ok
}

func (sm *SafeMap) Set(key, val string) {
    sm.mu.Lock()    // Exclusive write lock
    defer sm.mu.Unlock()
    sm.m[key] = val
}

snippet-8.go

var (
    dbInstance *sql.DB
    once       sync.Once
)

func GetDB() *sql.DB {
    once.Do(func() {
        // This runs exactly once, even across goroutines
        db, err := sql.Open("postgres", os.Getenv("DATABASE_URL"))
        if err != nil {
            log.Fatal(err)
        }
        dbInstance = db
    })
    return dbInstance
}

snippet-9.go

// WRONG: goroutine leaks if nobody reads from ch
go func() {
    result := doExpensiveWork()
    ch <- result  // Blocks forever if receiver is gone
}()

// RIGHT: use context for cancellation
go func() {
    select {
    case ch <- doExpensiveWork():
    case <-ctx.Done():
        return  // Goroutine exits cleanly
    }
}()

snippet.go

func main() {
    // Start 10,000 goroutines with ease
    for i := 0; i < 10000; i++ {
        go func(id int) {
            time.Sleep(time.Second)
            fmt.Printf("Worker %d done\n", id)
        }(i)
    }
    time.Sleep(2 * time.Second)
}

snippet.sh

go test -race ./...