GoSudaGo Concurrency Starter Pack
Overview
Brief Introduction
The Go language provides many tools for managing concurrency. In this article, I would like to introduce some of them along with a few tricks.
Goroutines?
A goroutine is a new form of concurrency model supported by the Go language. Typically, a program obtains OS threads from the OS to perform multiple tasks simultaneously, executing work in parallel up to the number of cores. To achieve concurrency at a smaller granularity, green threads are created in userland, allowing multiple green threads to take turns executing work within a single OS thread. However, goroutines make this form of green thread even smaller and more efficient. Goroutines use less memory than threads and can be created and switched more quickly than threads.
To use a goroutine, one simply needs to use the go keyword. This allows synchronous code to be executed as asynchronous code in an intuitive manner while writing a program.
1package main
2
3import (
4 "fmt"
5 "time"
6)
7
8func main() {
9 ch := make(chan struct{})
10 go func() {
11 defer close(ch)
12 time.Sleep(1 * time.Second)
13 fmt.Println("Hello, World!")
14 }()
15
16 fmt.Println("Waiting for goroutine...")
17 for range ch {}
18}
This code converts synchronous code that waits for 1 second and then prints Hello, World! into an asynchronous flow. Although the current example is simple, when converting more complex code from synchronous to asynchronous, its readability, visibility, and comprehensibility become better than with existing approaches such as async await or promises.
However, in many cases, without an understanding of the flow of simply calling synchronous code asynchronously and flows similar to fork & join (a flow resembling divide and conquer), poor goroutine code can be produced. To prepare for such cases, this article will introduce several methods and techniques.
Concurrency Management
context
It may be surprising that context appears as the first management technique. However, in the Go language, context goes beyond simple cancellation and plays an outstanding role in managing the entire tree of work. For those who may not be familiar, I will briefly explain this package.
1package main
2
3func main() {
4 ctx, cancel := context.WithCancel(context.Background())
5 defer cancel()
6
7 go func() {
8 <-ctx.Done()
9 fmt.Println("Context is done!")
10 }()
11
12 time.Sleep(1 * time.Second)
13
14 cancel()
15
16 time.Sleep(1 * time.Second)
17}
The above code uses context to print Context is done! after 1 second. context allows you to check whether cancellation has occurred through the Done() method, and it provides various cancellation methods through WithCancel, WithTimeout, WithDeadline, WithValue, and other methods.
Let us create a simple example. Suppose you write code that uses the aggregator pattern to fetch user, post, and comment data, and all requests must complete within 2 seconds. You could write it as follows.
1package main
2
3func main() {
4 ctx, cancel := context.WithTimeout(context.Background(), 2 * time.Second)
5 defer cancel()
6
7 ch := make(chan struct{})
8 go func() {
9 defer close(ch)
10 user := getUser(ctx)
11 post := getPost(ctx)
12 comment := getComment(ctx)
13
14 fmt.Println(user, post, comment)
15 }()
16
17 select {
18 case <-ctx.Done():
19 fmt.Println("Timeout!")
20 case <-ch:
21 fmt.Println("All data is fetched!")
22 }
23}
The above code prints Timeout! if all data is not fetched within 2 seconds, and prints All data is fetched! if all data is fetched. In this way, using context makes it easy to manage cancellation and timeouts even in code where multiple goroutines are running.
Various context-related functions and methods can be found in godoc context. I hope you will learn the simpler ones so that you can use them comfortably.
channel
unbuffered channel
channel is a tool for communication between goroutines. A channel can be created with make(chan T). Here, T is the type of data that the channel will transfer. Data can be sent and received on a channel using <-, and a channel can be closed using close.
1package main
2
3func main() {
4 ch := make(chan int)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
The above code uses a channel to print 1 and 2. This code only shows the simple act of sending and receiving values through a channel. However, channel provides more functionality than this. First, let us look at buffered channel and unbuffered channel. Before starting, the example written above is an unbuffered channel, where the act of sending data to the channel and the act of receiving data must occur simultaneously. If these actions do not occur at the same time, a deadlock may occur.
buffered channel
What if the above code, instead of simply printing, consisted of two processes performing heavy work? If the second process reads and processes data and is delayed for a long time, the first process will also be stopped during that time. To prevent such a situation, we can use a buffered channel.
1package main
2
3func main() {
4 ch := make(chan int, 2)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
The above code uses a buffered channel to print 1 and 2. In this code, a buffered channel is used so that the act of sending data to the channel and the act of receiving data do not have to occur simultaneously. By placing a buffer in the channel, there is room equal to its length, which can prevent task delays caused by the effects of lower-priority tasks.
select
When dealing with multiple channels, the select syntax can be used to easily implement a fan-in structure.
1package main
2
3import (
4 "fmt"
5 "time"
6)
7
8func main() {
9 ch1 := make(chan int, 10)
10 ch2 := make(chan int, 10)
11 ch3 := make(chan int, 10)
12
13 go func() {
14 for {
15 ch1 <- 1
16 time.Sleep(1 * time.Second)
17 }
18 }()
19 go func() {
20 for {
21 ch2 <- 2
22 time.Sleep(2 * time.Second)
23 }
24 }()
25 go func() {
26 for {
27 ch3 <- 3
28 time.Sleep(3 * time.Second)
29 }
30 }()
31
32 for i := 0; i < 3; i++ {
33 select {
34 case v := <-ch1:
35 fmt.Println(v)
36 case v := <-ch2:
37 fmt.Println(v)
38 case v := <-ch3:
39 fmt.Println(v)
40 }
41 }
42}
The above code creates three channels that periodically deliver 1, 2, and 3, and uses select to receive values from the channels and print them. In this way, using select allows you to receive data from multiple channels simultaneously and process values as they are received from the channels.
for range
A channel can easily receive data using for range. When for range is used on a channel, it operates each time data is added to that channel, and it terminates the loop when the channel is closed.
1package main
2
3func main() {
4 ch := make(chan int)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
The above code uses a channel to print 1 and 2. In this code, for range is used to receive and print data each time data is added to the channel. When the channel is closed, the loop terminates.
As written several times above, this syntax can also be used as a simple synchronization means.
1package main
2
3func main() {
4 ch := make(chan struct{})
5 go func() {
6 defer close(ch)
7 time.Sleep(1 * time.Second)
8 fmt.Println("Hello, World!")
9 }()
10
11 fmt.Println("Waiting for goroutine...")
12 for range ch {}
13}
The above code waits for 1 second and then prints Hello, World!. In this code, a channel is used to convert synchronous code into asynchronous code. In this way, using a channel makes it easy to convert synchronous code into asynchronous code and to set a join point.
etc
- Sending data to or receiving data from a nil channel may cause an infinite loop and a deadlock.
- Sending data after closing a channel causes a panic.
- Even if you do not explicitly close a channel, the GC collects it, which effectively closes the channel.
mutex
spinlock
A spinlock is a synchronization method that repeatedly attempts to acquire a lock in a loop. In the Go language, a spinlock can easily be implemented using pointers.
1package spinlock
2
3import (
4 "runtime"
5 "sync/atomic"
6)
7
8type SpinLock struct {
9 lock uintptr
10}
11
12func (s *SpinLock) Lock() {
13 for !atomic.CompareAndSwapUintptr(&s.lock, 0, 1) {
14 runtime.Gosched()
15 }
16}
17
18func (s *SpinLock) Unlock() {
19 atomic.StoreUintptr(&s.lock, 0)
20}
21
22func NewSpinLock() *SpinLock {
23 return &SpinLock{}
24}
The above code implements the spinlock package. In this code, SpinLock is implemented using the sync/atomic package. The Lock method attempts to acquire the lock using atomic.CompareAndSwapUintptr, and the Unlock method releases the lock using atomic.StoreUintptr. Because this approach attempts to acquire the lock without pause, it keeps consuming CPU until the lock is obtained, and it can fall into an infinite loop. Therefore, a spinlock is best used for simple synchronization or only for cases where it is held for a short period.
sync.Mutex
A mutex is a tool for synchronization between goroutines. The mutex provided by the sync package offers methods such as Lock, Unlock, RLock, and RUnlock. A mutex can be created with sync.Mutex, and a read/write lock can be used with sync.RWMutex.
1package main
2
3import (
4 "sync"
5)
6
7func main() {
8 var mu sync.Mutex
9 var count int
10
11 go func() {
12 mu.Lock()
13 count++
14 mu.Unlock()
15 }()
16
17 mu.Lock()
18 count++
19 mu.Unlock()
20
21 println(count)
22}
In the above code, two goroutines access the same count variable at almost the same time. At this point, if the code that accesses the count variable is made a critical section using a mutex, simultaneous access to the count variable can be prevented. Then this code will always print 2 no matter how many times it is executed.
sync.RWMutex
sync.RWMutex is a mutex that can distinguish between read locks and write locks. A read lock can be acquired and released using the RLock and RUnlock methods.
1package cmap
2
3import (
4 "sync"
5)
6
7type ConcurrentMap[K comparable, V any] struct {
8 sync.RWMutex
9 data map[K]V
10}
11
12func (m *ConcurrentMap[K, V]) Get(key K) (V, bool) {
13 m.RLock()
14 defer m.RUnlock()
15
16 value, ok := m.data[key]
17 return value, ok
18}
19
20func (m *ConcurrentMap[K, V]) Set(key K, value V) {
21 m.Lock()
22 defer m.Unlock()
23
24 m.data[key] = value
25}
The above code implements ConcurrentMap using sync.RWMutex. In this code, the Get method acquires a read lock, and the Set method acquires a write lock, so the data map can be safely accessed and modified. The reason for using a read lock is that when there are many simple read operations, multiple goroutines can perform read operations simultaneously by acquiring only a read lock instead of a write lock. Through this, performance can be improved by acquiring only a read lock in cases where there is no state change and a write lock is unnecessary.
fakelock
fakelock is a simple trick that implements sync.Locker. This struct provides the same methods as sync.Mutex, but it does not actually perform any operation.
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
The above code implements the fakelock package. This package implements sync.Locker and provides the Lock and Unlock methods, but it actually does nothing. I will explain why such code is needed when the opportunity arises.
waitgroup
sync.WaitGroup
sync.WaitGroup is a tool that waits until all of a goroutine's work has finished. It provides the Add, Done, and Wait methods. The Add method increases the number of goroutines, the Done method notifies that a goroutine's work has finished, and the Wait method waits until all goroutines' work has finished.
1package main
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8func main() {
9 wg := sync.WaitGroup{}
10 c := atomic.Int64{}
11
12 for i := 0; i < 100 ; i++ {
13 wg.Add(1)
14 go func() {
15 defer wg.Done()
16 c.Add(1)
17 }()
18 }
19
20 wg.Wait()
21 println(c.Load())
22}
The above code uses sync.WaitGroup so that 100 goroutines simultaneously add values to the c variable. In this code, sync.WaitGroup is used to wait until all goroutines have finished, and then the accumulated value of the c variable is printed. When performing fork & join on a small number of tasks, using only channels is sufficient, but when performing fork & join on a large number of tasks, using sync.WaitGroup is also a good choice.
with slice
When used together with a slice, waitgroup can become a tool for managing concurrent work without locks.
1package main
2
3import (
4 "fmt"
5 "sync"
6 "rand"
7)
8
9func main() {
10 var wg sync.WaitGroup
11 arr := [10]int{}
12
13 for i := 0; i < 10; i++ {
14 wg.Add(1)
15 go func(id int) {
16 defer wg.Done()
17
18 arr[id] = rand.Intn(100)
19 }(i)
20 }
21
22 wg.Wait()
23 fmt.Println("Done")
24
25 for i, v := range arr {
26 fmt.Printf("arr[%d] = %d\n", i, v)
27 }
28}
The above code uses only waitgroup so that each goroutine concurrently generates a random integer between 0 and 99 and stores it at its assigned index. In this code, waitgroup is used to wait until all goroutines have finished, and then Done is printed. In this way, using waitgroup allows multiple goroutines to perform work concurrently, store data without locks until all goroutines finish, and then perform post-processing collectively after the work is complete.
golang.org/x/sync/errgroup.ErrGroup
errgroup is a package that extends sync.WaitGroup. Unlike sync.WaitGroup, if an error occurs in any of the goroutines' work, errgroup cancels all goroutines and returns the error.
1package main
2
3import (
4 "context"
5 "fmt"
6 "golang.org/x/sync/errgroup"
7)
8
9func main() {
10 g, ctx := errgroup.WithContext(context.Background())
11 _ = ctx
12
13 for i := 0; i < 10; i++ {
14 i := i
15 g.Go(func() error {
16 if i == 5 {
17 return fmt.Errorf("error")
18 }
19 return nil
20 })
21 }
22
23 if err := g.Wait(); err != nil {
24 fmt.Println(err)
25 }
26}
The above code uses errgroup to create 10 goroutines, and an error is raised in the 5th goroutine. An error was intentionally raised in the fifth goroutine to demonstrate the case where an error occurs. However, in actual use, errgroup is used to create goroutines, and various post-processing can be carried out for cases where an error occurs in each goroutine.
once
This is a tool for executing code that must run only once. Related code can be executed through the constructors below.
1func OnceFunc(f func()) func()
2func OnceValue[T any](f func() T) func() T
3func OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2)
OnceFunc
OnceFunc simply ensures that the given function is executed exactly once across all calls.
1package main
2
3import "sync"
4
5func main() {
6 once := sync.OnceFunc(func() {
7 println("Hello, World!")
8 })
9
10 once()
11 once()
12 once()
13 once()
14 once()
15}
The above code uses sync.OnceFunc to print Hello, World!. In this code, sync.OnceFunc is used to create the once function, and even though the once function is called several times, Hello, World! is printed only once.
OnceValue
OnceValue does not simply ensure that the given function is executed exactly once; it also stores the return value of that function and returns the stored value when called again.
1package main
2
3import "sync"
4
5func main() {
6 c := 0
7 once := sync.OnceValue(func() int {
8 c += 1
9 return c
10 })
11
12 println(once())
13 println(once())
14 println(once())
15 println(once())
16 println(once())
17}
The above code uses sync.OnceValue to increment the c variable by 1. In this code, sync.OnceValue is used to create the once function, and even though the once function is called several times, the c variable is incremented only once and returns 1.
OnceValues
OnceValues works the same as OnceValue, but it can return multiple values.
1package main
2
3import "sync"
4
5func main() {
6 c := 0
7 once := sync.OnceValues(func() (int, int) {
8 c += 1
9 return c, c
10 })
11
12 a, b := once()
13 println(a, b)
14 a, b = once()
15 println(a, b)
16 a, b = once()
17 println(a, b)
18 a, b = once()
19 println(a, b)
20 a, b = once()
21 println(a, b)
22}
The above code uses sync.OnceValues to increment the c variable by 1. In this code, sync.OnceValues is used to create the once function, and even though the once function is called several times, the c variable is incremented only once and returns 1.
atomic
The atomic package is a package that provides atomic operations. It provides methods such as Add, CompareAndSwap, Load, Store, and Swap, but recently the use of types such as Int64, Uint64, and Pointer is recommended.
1package main
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8func main() {
9 wg := sync.WaitGroup{}
10 c := atomic.Int64{}
11
12 for i := 0; i < 100 ; i++ {
13 wg.Add(1)
14 go func() {
15 defer wg.Done()
16 c.Add(1)
17 }()
18 }
19
20 wg.Wait()
21 println(c.Load())
22}
This is the example used earlier. It is code that atomically increments the c variable using the atomic.Int64 type. The Add and Load methods atomically increment the variable and read it. In addition, the Store method stores a value, the Swap method replaces a value, and the CompareAndSwap method compares a value and replaces it if it matches.
cond
sync.Cond
The cond package is a package that provides condition variables. The cond package can be created with sync.Cond and provides the Wait, Signal, and Broadcast methods.
1package main
2
3import (
4 "sync"
5)
6
7func main() {
8 c := sync.NewCond(&sync.Mutex{})
9 ready := false
10
11 go func() {
12 c.L.Lock()
13 ready = true
14 c.Signal()
15 c.L.Unlock()
16 }()
17
18 c.L.Lock()
19 for !ready {
20 c.Wait()
21 }
22 c.L.Unlock()
23
24 println("Ready!")
25}
The above code uses sync.Cond to wait until the ready variable becomes true. In this code, sync.Cond is used to wait until the ready variable becomes true, and then Ready! is printed. In this way, using sync.Cond allows multiple goroutines to wait until a specific condition is satisfied.
Using this, a simple queue can be implemented.
1package queue
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8type Node[T any] struct {
9 Value T
10 Next *Node[T]
11}
12
13type Queue[T any] struct {
14 sync.Mutex
15 Cond *sync.Cond
16 Head *Node[T]
17 Tail *Node[T]
18 Len int
19}
20
21func New[T any]() *Queue[T] {
22 q := &Queue[T]{}
23 q.Cond = sync.NewCond(&q.Mutex)
24 return q
25}
26
27func (q *Queue[T]) Push(value T) {
28 q.Lock()
29 defer q.Unlock()
30
31 node := &Node[T]{Value: value}
32 if q.Len == 0 {
33 q.Head = node
34 q.Tail = node
35 } else {
36 q.Tail.Next = node
37 q.Tail = node
38 }
39 q.Len++
40 q.Cond.Signal()
41}
42
43func (q *Queue[T]) Pop() T {
44 q.Lock()
45 defer q.Unlock()
46
47 for q.Len == 0 {
48 q.Cond.Wait()
49 }
50
51 node := q.Head
52 q.Head = q.Head.Next
53 q.Len--
54 return node.Value
55}
By utilizing sync.Cond in this way, instead of consuming a large amount of CPU with a spin-lock, it can wait efficiently and resume operation when the condition is satisfied.
semaphore
golang.org/x/sync/semaphore.Semaphore
The semaphore package is a package that provides semaphores. The semaphore package can be created with golang.org/x/sync/semaphore.Semaphore and provides the Acquire, Release, and TryAcquire methods.
1package main
2
3import (
4 "context"
5 "fmt"
6 "golang.org/x/sync/semaphore"
7)
8
9func main() {
10 s := semaphore.NewWeighted(1)
11
12 if s.TryAcquire(1) {
13 fmt.Println("Acquired!")
14 } else {
15 fmt.Println("Not Acquired!")
16 }
17
18 s.Release(1)
19}
The above code creates a semaphore using semaphore, acquires it with the Acquire method, and releases it with the Release method. In this code, the method of acquiring and releasing a semaphore using semaphore is demonstrated.
Conclusion
It seems that this is enough for the basic content. Based on the contents of this article, I hope you will understand how to manage concurrency using goroutines and be able to use them in practice. I hope this article has been helpful to you. Thank you.