0. Who am I?

👋 Hi, I’m Youssef Ameachaq
👀 I’m an engineering manager at Oracle.
🌱 I contribute to the htmx project.
💻 I try to build some cool stuff with Go and JavaScript.
📫 Reach me at youssefameachaq@gmail.com.
🐦 Follow me on Twitter: x.com/youssame0
🔗 Connect with me on LinkedIn: linkedin.com/in/youssef-ameachaq/

Intro

• Go provides a powerful and simple way to handle concurrency using goroutines and channels.

1. Worker Pool Pattern

• What it does
  • A worker pool uses multiple goroutines (workers) to process tasks from a shared channel. This pattern limits the number of concurrent tasks.

package main

import (
	"fmt"
	"time"
)

// worker processes tasks from the jobs channel
func worker(id int, jobs <-chan int, results chan<- int) {
	for job := range jobs {
		fmt.Printf("Worker %d started job %d\n", id, job)
		time.Sleep(time.Second) // Simulating work
		fmt.Printf("Worker %d finished job %d\n", id, job)
		results <- job * 2 // Send result back
	}
}

func main() {
	jobs := make(chan int, 5)    // Jobs channel
	results := make(chan int, 5) // Results channel

	// 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) // Close jobs channel when done sending

	// Collect results
	for r := 1; r <= 5; r++ {
		fmt.Println("Result:", <-results)
	}
}

2. Fan-Out, Fan-In Pattern

• What it does
  • Fan-Out: Distributes tasks to multiple goroutines for parallel processing.
  • Fan-In: Combines results from multiple goroutines into one channel.

Example

package main

import (
	"fmt"
	"time"
)

func worker(id int, data int, results chan<- int) {
	time.Sleep(time.Second) // Simulate work
	results <- data * 2 // Send back result
}

func main() {
	data := []int{1, 2, 3, 4, 5}
	results := make(chan int, len(data))

	// Fan-out: Start a goroutine for each data item
	for i, value := range data {
		go worker(i, value, results)
	}

	// Fan-in: Collect results
	for i := 0; i < len(data); i++ {
		fmt.Println("Result:", <-results)
	}
}

3. Pipelines

• What it does
  • Data flows through a series of stages, with each stage implemented by a goroutine.

Example

package main

import "fmt"

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

// Squares numbers
func square(input <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		for num := range input {
			out <- num * num
		}
		close(out)
	}()
	return out
}

func main() {
	nums := generator(1, 2, 3, 4, 5)
	squares := square(nums)

	// Consume the pipeline output
	for result := range squares {
		fmt.Println(result)
	}
}

4. Select Statement

• What it does
  • The select statement waits for one of multiple channel operations to proceed. This is useful for handling multiple concurrent tasks.

Example (Without default case)

package main

import (
	"fmt"
	"time"
)

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

	go func() {
		time.Sleep(1 * time.Second)
		ch1 <- "Channel 1 finished"
	}()

	go func() {
		time.Sleep(2 * time.Second)
		ch2 <- "Channel 2 finished"
	}()

	for i := 0; i < 2; i++ {
		select {
		case msg1 := <-ch1:
			fmt.Println(msg1)
		case msg2 := <-ch2:
			fmt.Println(msg2)
		}
	}
}

Using Default (Non-blocking select)

package main

import (
	"fmt"
	"time"
)

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

	go func() {
		time.Sleep(1 * time.Second)
		ch1 <- "Channel 1 finished"
	}()

	go func() {
		time.Sleep(2 * time.Second)
		ch2 <- "Channel 2 finished"
	}()

	for i := 0; i < 5; i++ {
		select {
		case msg1 := <-ch1:
			fmt.Println(msg1)
		case msg2 := <-ch2:
			fmt.Println(msg2)
		default:
			fmt.Println("No channel is ready yet")
			time.Sleep(500 * time.Millisecond) // Simulate doing other work
		}
	}
}

5. Mutex for Shared State

• What it does
  • A sync.Mutex ensures that only one goroutine can access shared data at a time, preventing race conditions.

Example

package main

import (
	"fmt"
	"sync"
)

func main() {
	var mu sync.Mutex
	counter := 0

	wg := sync.WaitGroup{}
	wg.Add(5)

	for i := 0; i < 5; i++ {
		go func(id int) {
			mu.Lock() // Lock before modifying shared state
			counter++
			fmt.Printf("Goroutine %d incremented counter to %d\n", id, counter)
			mu.Unlock() // Unlock after modification
			wg.Done()
		}(i)
	}

	wg.Wait()
	fmt.Println("Final Counter:", counter)
}

6. Context for Cancellation and Deadlines

• What it does
  • The context package is used to manage cancellations, timeouts, and deadlines for goroutines.

Example

package main

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

func worker(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			fmt.Println("Worker stopped")
			return
		default:
			fmt.Println("Worker running")
			time.Sleep(500 * time.Millisecond)
		}
	}
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	go worker(ctx)

	time.Sleep(3 * time.Second)
	fmt.Println("Main finished")
}

Wrap up

• Worker Pool: Process tasks with a fixed number of goroutines.
• Fan-Out/Fan-In: Spread tasks across goroutines and collect results.
• Pipelines: Pass data through stages of processing.
• Select: Handle multiple channels simultaneously.
• Mutex: Protect shared resources from concurrent access.
• Context: Manage cancellations and timeouts.

Thank you