If you are not a medium member, you can still read this story here
Concurrency is a concept in computer science that involves structuring a program to perform multiple tasks at the same time, not necessarily simultaneously, but by breaking down tasks and managing them in a way that the program appears to be working on them concurrently.
Table of Contents
- Introduction to Concurrency
- Basics of Goroutines
- Synchronizing Goroutines
- Communication Between Goroutines with Channels
- Buffered vs. Unbuffered Channels
- Select Statement
- Mutexes in Go
- Best Practices for Go Concurrency
- Hands-on: Creating Worker Pools
- Conclusion
Prerequisites
- Basic Knowledge of Go: Familiarity with Go syntax, functions, and basic data structures.
- Basic Familiarity with Multithreading/Concurrency: Understanding of general concepts like threads and parallelism (not specific to Go).
“Concurrency is not Parallelism”’:
Go uses both concurrency and parallelism, but they are distinct concept. It’s important to distinguish concurrency from parallelism, though the two are often confused. Go is designed with concurrency as a core feature, making it easy to write programs that can handle multiple tasks simultaneously. Go can also leverage parallelism on multi-core systems, thanks to its runtime scheduler. By default, Go will use as many CPU cores as are available to the program. You can control this behavior using the GOMAXPROCS setting, which specifies the maximum number of OS threads that can run Go code simultaneously.
Concurrency is about dealing with multiple tasks, possibly overlapping in time, but not necessarily executing at the same moment. For example, a program might begin downloading a file (an I/O-bound task) while also allowing the user to interact with its interface (a CPU-bound task), switching between the two when appropriate.
Parallelism, on the other hand, refers to actual simultaneous execution. This usually happens on systems with multiple CPU cores, where each core can handle one task at the exact same time.
Concurrency is often implemented using threads, coroutines, or tasks, which execute portions of work independently. A scheduler or runtime system manages when each task runs, ensuring that no task monopolizes the system’s resources.
In Go, concurrency is achieved primarily through goroutines and channels, which are simpler and more lightweight compared to traditional threads. The Go runtime automatically schedules goroutines, allowing them to be executed concurrently.
In the Go community, Concurrency is not Parallelism is a topic of great debate. Watch the presentation by Rob Pike, one of the original authors of Golang here.
Basics of Goroutines:
A goroutine is a lightweight function or method that runs concurrently with other goroutines in Go. Goroutines are managed by the Go runtime, not the operating system, making them very lightweight compared to traditional OS threads. You can run thousands or even millions of goroutines without overwhelming system resources.
Goroutine Execution Model
Goroutines are multiplexed onto a smaller number of OS threads by the Go runtime, which allows it to handle scheduling efficiently. The Go scheduler is responsible for switching between goroutines, allowing them to run concurrently.
The Go runtime is responsible for managing goroutines and their execution. It includes a scheduler that efficiently multiplexes goroutines onto a limited number of OS threads. The goal is to make goroutines lightweight, fast, and easy to manage. The scheduler works to balance the execution of many goroutines across available CPU cores while minimizing overhead.
- Goroutines (G): These are lightweight, independent units of execution created using the
gokeyword. They are not OS-level threads but managed by the Go runtime. - OS Threads (M): The Go runtime uses a limited number of OS threads to execute goroutines. These threads are managed by the operating system, and Go’s scheduler maps multiple goroutines to these threads.
- Processors (P): The scheduler uses logical processors (P) to schedule goroutines. The number of
Ps is determined byGOMAXPROCS, which defines how many OS threads can execute Go code simultaneously. EachPis associated with an OS thread and is responsible for executing goroutines in its local queue. - Global Run Queue: When a new goroutine is created, it’s placed in the global run queue if there are no free processors available to handle it immediately. The global run queue serves as a pool of ready-to-execute goroutines that processors can pull from if they run out of tasks.
- Local Run Queue (per P): Each processor has its own local run queue, where goroutines are kept ready for execution. The scheduler tries to run goroutines from the local queue before looking into the global queue.
Step-by-Step Execution:
- Goroutine Creation: When a goroutine is created, it’s either placed directly into the local queue of a processor (P) or into the global queue if no processors are available.
- Processor Assignment: A logical processor (P) executes goroutines by pulling them from its local run queue. If the local queue is empty, it can pull goroutines from the global queue or steal from another processor’s local queue.
- Goroutine Execution: The scheduler assigns a goroutine (G) to an OS thread (M), and the thread runs that goroutine until it blocks (e.g., waiting for I/O) or yields its time slice.
- Preemption: The Go scheduler can preempt goroutines that are long-running or blocking, ensuring that other goroutines get a chance to execute. This prevents any goroutine from monopolizing the processor.
- Resuming Execution: Once a blocked goroutine becomes ready (e.g., after an I/O operation completes), it is placed back into the local or global run queue to be scheduled for execution again.
The Go scheduler maps multiple goroutines to these OS threads. The scheduler creates logical processors, which are responsible for managing goroutines in their local queues. The number of processors is defined by GOMAXPROCS. By using a limited number of OS threads and managing goroutines at the Go runtime level, the scheduler ensures efficient use of system resources.
Go Scheduler uses a model called as ‘work-stealing’. Here, a processor with no work looks for other processor’s threads and steals some. This is different from an another common model called ‘work-sharing’ where a processor with lots of threads hopes that some of the other under utilized processors pick them up.
Creating Goroutines
When you call a function normally, it runs synchronously. That is, the program waits for the function to return a value before continuing. Go routines, on the other hand, allow you to call functions asynchronously, meaning the program doesn’t have to wait for the function to complete before moving on to the next task.
A goroutine is created by using the go keyword followed by a function call. This function will run concurrently with the rest of the program.
package main
import (
"fmt"
"time"
)
func sayHello() {
fmt.Println("Hello, World!")
}
func main() {
go sayHello() // Run sayHello concurrently
time.Sleep(1 * time.Second)
}When you pass arguments to a goroutine, the arguments are passed by value at the moment the goroutine is created.
go func(msg string) {
fmt.Println(msg)
}("Hello, Go!")This main function itself runs as a Goroutine. Hence, all the Goroutines created from the main function are children of it.
In the above example, both the sayHello function and the main function run concurrently. However, since Go routines don’t wait for other routines to finish, the program might terminate before the Go routine completes. This leads us to a crucial aspect of Go routines: synchronization.
Synchronizing Go Routines:
Because Go routines run concurrently, one of the first challenges you’ll face is ensuring that the main Go routine (the main function) does not exit before other Go routines finish executing.
Using sync.WaitGroup:
The sync.WaitGroup is a common way to manage multiple Go routines and wait for them to finish. A WaitGroup keeps track of how many Go routines are running and lets you block the main program until all of them have finished.
package main
import (
"fmt"
"sync"
)
func sayHello(wg *sync.WaitGroup) {
fmt.Println("Hello from Go routine")
wg.Done() // Decrement the counter when the Go routine completes
}
func main() {
var wg sync.WaitGroup // Initialize a WaitGroup
wg.Add(1) // Increment the counter for the Go routine
go sayHello(&wg)
wg.Wait() // Block until all Go routines have completed
}How It Works:
- wg.Add(1): Adds one to the
WaitGroupcounter. - wg.Done(): Decrements the
WaitGroupcounter when the Go routine completes. - wg.Wait(): Blocks the program until the counter returns to zero.
Communication Between Go Routines with Channels:
While Go routines allow functions to run concurrently, channels provide a way for them to communicate safely. Channels allow Go routines to send and receive values to each other in a synchronized manner.
A channel is declared using the chan keyword:
ch := make(chan int)You can send and receive values to/from a channel using the <- operator. Here’s an example of two Go routines communicating via a channel:
package main
import "fmt"
func sendValues(ch chan<- int) { // Send-only channel
ch <- 42
}
func main() {
ch := make(chan int) // Create a new channel
go sendValues(ch) // Start a Go routine that sends a value
value := <-ch // Receive the value from the channel
fmt.Println("Received:", value)
}Channel Directions
- Send-only: A channel that can only send values (
chan<-). - Receive-only: A channel that can only receive values (
<-chan). - Bidirectional: A channel that can both send and receive values.
Closing a channel signals that no more values will be sent on it. This is important when using channels to control communication between multiple Go routines.
package main
import "fmt"
func sendValues(ch chan int) {
ch <- 42
close(ch) // Close the channel when done sending
}
func main() {
ch := make(chan int)
go sendValues(ch)
for value := range ch { // Loop over channel values until it is closed
fmt.Println("Received:", value)
}
}Buffered vs. Unbuffered Channels
By default, Go channels are unbuffered, meaning they block the sender until the receiver is ready to receive. Buffered channels, on the other hand, allow you to send multiple values before blocking.
Here’s how you can declare a buffered channel:
ch := make(chan int, 2) // Create a buffered channel with a capacity of 2You can now send two values before the channel blocks:
package main
import "fmt"
func main() {
ch := make(chan int, 2) // Buffered channel with capacity 2
ch <- 1 // No blocking
ch <- 2 // No blocking
fmt.Println(<-ch) // Receive first value
fmt.Println(<-ch) // Receive second value
}Select Statement
Go’s select statement allows you to listen on multiple channels simultaneously. This is useful for handling multiple concurrent operations, such as waiting for data on one channel or a timeout on another.
package main
import (
"fmt"
"time"
)
func main() {
ch1 := make(chan string)
ch2 := make(chan string)
go func() {
time.Sleep(1 * time.Second)
ch1 <- "from channel 1"
}()
go func() {
time.Sleep(2 * time.Second)
ch2 <- "from channel 2"
}()
select {
case msg1 := <-ch1:
fmt.Println(msg1)
case msg2 := <-ch2:
fmt.Println(msg2)
case <-time.After(3 * time.Second):
fmt.Println("timeout")
}
}Mutexes in Go
While Go routines allow you to perform concurrent tasks, sometimes those tasks need to access shared data, such as variables or resources. If multiple Go routines attempt to modify shared data simultaneously, it can lead to race conditions — an unpredictable state caused by the overlap in execution timing. To avoid these issues, Go provides several synchronization primitives, one of which is the mutex.
A mutex (short for mutual exclusion) is a tool used to manage concurrent access to shared resources, ensuring that only one Go routine can access the critical section of code at a time. This avoids race conditions by locking access to shared data while a Go routine is working on it and releasing the lock when it’s done.
How Mutexes Work
A mutex has two fundamental operations:
- Lock: Blocks other Go routines from accessing the shared resource until the current Go routine is done with it.
- Unlock: Allows other Go routines to access the shared resource after the current Go routine finishes.
In Go, mutexes are part of the sync package and are used as follows:
Lock()prevents other Go routines from entering the critical section.Unlock()releases the lock, allowing other Go routines to access the critical section.
Without Mutex (Race Condition):
package main
import (
"fmt"
"sync"
)
var counter = 0
func increment() {
for i := 0; i < 1000; i++ {
counter++
}
}
func main() {
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
increment()
}()
go func() {
defer wg.Done()
increment()
}()
wg.Wait()
fmt.Println("Final Counter:", counter)
}In the above code, we have two Go routines incrementing the same counter variable. Since both Go routines may access and modify the counter at the same time, we can’t guarantee that the final result will be 2000. This is an example of a race condition.
With Mutex (Avoiding Race Condition):
package main
import (
"fmt"
"sync"
)
var (
counter int
mutex sync.Mutex
)
func increment() {
for i := 0; i < 1000; i++ {
mutex.Lock() // Acquire the lock before modifying the counter
counter++
mutex.Unlock() // Release the lock after modification
}
}
func main() {
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
increment()
}()
go func() {
defer wg.Done()
increment()
}()
wg.Wait()
fmt.Println("Final Counter:", counter)
}In this version, we use a mutex to ensure that only one Go routine at a time can increment the counter. As a result, the final output will reliably be 2000.
Best Practices for Go Concurrency
Avoid Leaking Go Routines:
Always ensure that Go routines are properly synchronized, or that they exit when no longer needed. Use channels or context.Context to signal when a Go routine should stop.
Limit the Number of Go Routines:
Creating too many Go routines can lead to performance issues. Use worker pools or rate-limiting techniques to control the number of active Go routines.
Handle Panic Gracefully:
Use recover() in Go routines to catch panics and avoid crashing the entire program.
go func() {
defer func() {
if r := recover(); r != nil {
fmt.Println("Recovered from panic:", r)
}
}()
panic("something went wrong")
}()Hands-on: Creating Worker Pools (Optional):
Go routines and channels can be combined to create powerful concurrency patterns. One of those is Worker Pools. This pattern is widely used in the real world production systems.
A worker pool is a concurrency pattern where a fixed number of worker Go routines (the “pool”) handle a stream of tasks. Each worker processes tasks concurrently, improving performance and allowing for controlled parallelism. This pattern is highly useful when you have a large number of tasks but want to limit the number of active Go routines to avoid overwhelming your system’s resources.
In this hands-on guide, we’ll build a simple worker pool from scratch and see how it works step by step.
Problem Setup
Let’s say you need to process a list of tasks, and each task involves performing some I/O-bound or CPU-bound work. For simplicity, we’ll simulate this with a function that takes time to complete each task. Instead of spawning a new Go routine for each task (which can overwhelm the system), we’ll use a worker pool to process these tasks efficiently.
We’ll build a worker pool with:
- Workers: Go routines that take tasks from a shared channel and process them.
- Job Queue: A channel where tasks (jobs) are sent.
- Main Routine: A Go routine that distributes jobs to the worker pool.
Basic Worker Pool Architecture
We will start by implementing the following components:
- Worker Function: Each worker will pick up tasks from a job queue, process them, and then wait for the next task.
- Job Dispatcher: A central dispatcher will create and distribute tasks to the workers.
- Job Queue: A channel that feeds tasks into the worker pool.
Step 1: Define the Job and Worker Functions
package main
import (
"fmt"
"sync"
"time"
)
// Job represents the work that needs to be done
type Job struct {
ID int
WorkTime time.Duration // How long the job takes to complete
}
// Worker represents a single worker in the pool
func worker(id int, jobs <-chan Job, wg *sync.WaitGroup) {
defer wg.Done() // Signal the WaitGroup when done
for job := range jobs {
fmt.Printf("Worker %d started job %d\n", id, job.ID)
time.Sleep(job.WorkTime) // Simulate doing the job
fmt.Printf("Worker %d finished job %d\n", id, job.ID)
}
}Jobstruct: This represents a task, with an ID and the time it takes to complete.workerfunction: This function receives jobs from a channel and processes each one by sleeping for the time specified inWorkTime(simulating work). Each worker will be a Go routine.
Step 2: Create the Dispatcher and Main Function
package main
import (
"fmt"
"sync"
"time"
)
type Job struct {
ID int
WorkTime time.Duration
}
func worker(id int, jobs <-chan Job, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs {
fmt.Printf("Worker %d started job %d\n", id, job.ID)
time.Sleep(job.WorkTime)
fmt.Printf("Worker %d finished job %d\n", id, job.ID)
}
}
func main() {
// Number of workers in the pool
const numWorkers = 3
// Create a job queue (channel)
jobs := make(chan Job, 10)
var wg sync.WaitGroup
// Start the workers
for i := 1; i <= numWorkers; i++ {
wg.Add(1)
go worker(i, jobs, &wg)
}
// Create some jobs and send them to the workers
for j := 1; j <= 5; j++ {
jobs <- Job{ID: j, WorkTime: time.Second * 2}
fmt.Printf("Sent job %d to the job queue\n", j)
}
// Close the job channel to signal workers that no more jobs are coming
close(jobs)
// Wait for all workers to finish
wg.Wait()
fmt.Println("All workers finished their tasks")
}- Job Queue: The channel
jobsis the job queue, which is buffered to store a limited number of jobs. Here, we usejobs := make(chan Job, 10). - Workers: We start a fixed number of workers (
numWorkers). Each worker is a Go routine that will pull jobs from thejobschannel and process them. - Job Creation: We create and send 5 jobs, each with a 2-second processing time, to the job queue. Workers will pick up jobs as soon as they are available.
- Closing the Job Queue: After all jobs are sent, we close the channel using
close(jobs). This signals the workers that no more jobs will be sent, and they can finish once they complete the current tasks. - WaitGroup (
wg): We use async.WaitGroupto wait for all workers to finish before terminating the main routine.
Running the Code
When you run the above program, the output might look something like this:
Sent job 1 to the job queue
Sent job 2 to the job queue
Sent job 3 to the job queue
Sent job 4 to the job queue
Sent job 5 to the job queue
Worker 1 started job 1
Worker 2 started job 2
Worker 3 started job 3
Worker 1 finished job 1
Worker 1 started job 4
Worker 2 finished job 2
Worker 2 started job 5
Worker 3 finished job 3
Worker 1 finished job 4
Worker 2 finished job 5
All workers finished their tasksNotice that the three workers process jobs concurrently. For example, worker 1 processes job 1 while worker 2 processes job 2, and so on.
Once the workers complete their jobs, they pick up new jobs from the queue until all the jobs are finished.
Expanding the Worker Pool with Dynamic Jobs
Let’s modify the worker pool to handle an arbitrary number of jobs. We’ll simulate different processing times for each job:
package main
import (
"fmt"
"math/rand"
"sync"
"time"
)
type Job struct {
ID int
WorkTime time.Duration
}
func worker(id int, jobs <-chan Job, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs {
fmt.Printf("Worker %d started job %d\n", id, job.ID)
time.Sleep(job.WorkTime)
fmt.Printf("Worker %d finished job %d\n", id, job.ID)
}
}
func main() {
const numWorkers = 4
jobs := make(chan Job, 10)
var wg sync.WaitGroup
// Start workers
for i := 1; i <= numWorkers; i++ {
wg.Add(1)
go worker(i, jobs, &wg)
}
// Create 10 jobs with random processing times
for j := 1; j <= 10; j++ {
workTime := time.Duration(rand.Intn(3)+1) * time.Second
jobs <- Job{ID: j, WorkTime: workTime}
fmt.Printf("Sent job %d to the job queue (work time: %v)\n", j, workTime)
}
close(jobs)
wg.Wait()
fmt.Println("All workers have finished")
}Changes:
- Random Work Time: We use
rand.Intn()to randomly assign different work times (1 to 3 seconds) to each job. - More Jobs: We now create 10 jobs, and the workers will process them based on availability.
Sent job 1 to the job queue (work time: 2s)
Sent job 2 to the job queue (work time: 1s)
Sent job 3 to the job queue (work time: 3s)
...
Worker 1 started job 1
Worker 2 started job 2
...
Worker 1 finished job 1
Worker 1 started job 4
Worker 2 finished job 2
...
All workers have finishedhe worker pool ensures that only a limited number of Go routines (workers) are running at a time, avoiding overwhelming the system with too many concurrent Go routines. By limiting the number of workers, the worker pool pattern efficiently manages system resources like memory and CPU.
Wrapping Up: Go’s Approach to Concurrency
In this part, we have explored essential concepts and techniques for building efficient and scalable concurrent programs using Go. We also built a worker pool, a practical concurrency pattern where a fixed number of workers process a stream of tasks from a job queue.
Understanding and implementing these concurrency patterns and practices is essential for developing robust, high-performance applications in Go. By leveraging goroutines, channels, and synchronization primitives, you can build efficient concurrent systems that handle multiple tasks while avoiding common pitfalls like race conditions and resource exhaustion.
“You take many things from the world, but I wonder what you will give back in return?”
― Genzaburo Yoshino, How Do You Live?
