--- id: 01933e13f963942e4c9bc8c3c89fbc2b author: snowmerak title: Go Concurrency Starter Pack description: This article introduces various tools and techniques for managing concurrency in Go, including goroutines, channels, and mutexes. language: en date: 2024-10-18T05:51:30.010680581Z path: /blog/posts/go-concurrency-starter-pack-z221a3399 category: concurrency --- ## 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. ```go package main import ( "fmt" "time" ) func main() { ch := make(chan struct{}) go func() { defer close(ch) time.Sleep(1 * time.Second) fmt.Println("Hello, World!") }() fmt.Println("Waiting for goroutine...") for range ch {} } ``` 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. ```go package main func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() go func() { <-ctx.Done() fmt.Println("Context is done!") }() time.Sleep(1 * time.Second) cancel() time.Sleep(1 * time.Second) } ``` 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. ```go package main func main() { ctx, cancel := context.WithTimeout(context.Background(), 2 * time.Second) defer cancel() ch := make(chan struct{}) go func() { defer close(ch) user := getUser(ctx) post := getPost(ctx) comment := getComment(ctx) fmt.Println(user, post, comment) }() select { case <-ctx.Done(): fmt.Println("Timeout!") case <-ch: fmt.Println("All data is fetched!") } } ``` 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](https://pkg.go.dev/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`. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` 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`. ```go package main func main() { ch := make(chan int, 2) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` 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. ```go package main import ( "fmt" "time" ) func main() { ch1 := make(chan int, 10) ch2 := make(chan int, 10) ch3 := make(chan int, 10) go func() { for { ch1 <- 1 time.Sleep(1 * time.Second) } }() go func() { for { ch2 <- 2 time.Sleep(2 * time.Second) } }() go func() { for { ch3 <- 3 time.Sleep(3 * time.Second) } }() for i := 0; i < 3; i++ { select { case v := <-ch1: fmt.Println(v) case v := <-ch2: fmt.Println(v) case v := <-ch3: fmt.Println(v) } } } ``` 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. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` 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. ```go package main func main() { ch := make(chan struct{}) go func() { defer close(ch) time.Sleep(1 * time.Second) fmt.Println("Hello, World!") }() fmt.Println("Waiting for goroutine...") for range ch {} } ``` 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 1. Sending data to or receiving data from a nil channel may cause an infinite loop and a deadlock. 2. Sending data after closing a channel causes a panic. 3. 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. ```go package spinlock import ( "runtime" "sync/atomic" ) type SpinLock struct { lock uintptr } func (s *SpinLock) Lock() { for !atomic.CompareAndSwapUintptr(&s.lock, 0, 1) { runtime.Gosched() } } func (s *SpinLock) Unlock() { atomic.StoreUintptr(&s.lock, 0) } func NewSpinLock() *SpinLock { return &SpinLock{} } ``` 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`. ```go package main import ( "sync" ) func main() { var mu sync.Mutex var count int go func() { mu.Lock() count++ mu.Unlock() }() mu.Lock() count++ mu.Unlock() println(count) } ``` 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. ```go package cmap import ( "sync" ) type ConcurrentMap[K comparable, V any] struct { sync.RWMutex data map[K]V } func (m *ConcurrentMap[K, V]) Get(key K) (V, bool) { m.RLock() defer m.RUnlock() value, ok := m.data[key] return value, ok } func (m *ConcurrentMap[K, V]) Set(key K, value V) { m.Lock() defer m.Unlock() m.data[key] = value } ``` 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. ```go package fakelock type FakeLock struct{} func (f *FakeLock) Lock() {} func (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. ```go package main import ( "sync" "sync/atomic" ) func main() { wg := sync.WaitGroup{} c := atomic.Int64{} for i := 0; i < 100 ; i++ { wg.Add(1) go func() { defer wg.Done() c.Add(1) }() } wg.Wait() println(c.Load()) } ``` 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. ```go package main import ( "fmt" "sync" "rand" ) func main() { var wg sync.WaitGroup arr := [10]int{} for i := 0; i < 10; i++ { wg.Add(1) go func(id int) { defer wg.Done() arr[id] = rand.Intn(100) }(i) } wg.Wait() fmt.Println("Done") for i, v := range arr { fmt.Printf("arr[%d] = %d\n", i, v) } } ``` 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. ```go package main import ( "context" "fmt" "golang.org/x/sync/errgroup" ) func main() { g, ctx := errgroup.WithContext(context.Background()) _ = ctx for i := 0; i < 10; i++ { i := i g.Go(func() error { if i == 5 { return fmt.Errorf("error") } return nil }) } if err := g.Wait(); err != nil { fmt.Println(err) } } ``` 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. ```go func OnceFunc(f func()) func() func OnceValue[T any](f func() T) func() T func 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. ```go package main import "sync" func main() { once := sync.OnceFunc(func() { println("Hello, World!") }) once() once() once() once() once() } ``` 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. ```go package main import "sync" func main() { c := 0 once := sync.OnceValue(func() int { c += 1 return c }) println(once()) println(once()) println(once()) println(once()) println(once()) } ``` 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. ```go package main import "sync" func main() { c := 0 once := sync.OnceValues(func() (int, int) { c += 1 return c, c }) a, b := once() println(a, b) a, b = once() println(a, b) a, b = once() println(a, b) a, b = once() println(a, b) a, b = once() println(a, b) } ``` 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. ```go package main import ( "sync" "sync/atomic" ) func main() { wg := sync.WaitGroup{} c := atomic.Int64{} for i := 0; i < 100 ; i++ { wg.Add(1) go func() { defer wg.Done() c.Add(1) }() } wg.Wait() println(c.Load()) } ``` 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. ```go package main import ( "sync" ) func main() { c := sync.NewCond(&sync.Mutex{}) ready := false go func() { c.L.Lock() ready = true c.Signal() c.L.Unlock() }() c.L.Lock() for !ready { c.Wait() } c.L.Unlock() println("Ready!") } ``` 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. ```go package queue import ( "sync" "sync/atomic" ) type Node[T any] struct { Value T Next *Node[T] } type Queue[T any] struct { sync.Mutex Cond *sync.Cond Head *Node[T] Tail *Node[T] Len int } func New[T any]() *Queue[T] { q := &Queue[T]{} q.Cond = sync.NewCond(&q.Mutex) return q } func (q *Queue[T]) Push(value T) { q.Lock() defer q.Unlock() node := &Node[T]{Value: value} if q.Len == 0 { q.Head = node q.Tail = node } else { q.Tail.Next = node q.Tail = node } q.Len++ q.Cond.Signal() } func (q *Queue[T]) Pop() T { q.Lock() defer q.Unlock() for q.Len == 0 { q.Cond.Wait() } node := q.Head q.Head = q.Head.Next q.Len-- return node.Value } ``` 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. ```go package main import ( "context" "fmt" "golang.org/x/sync/semaphore" ) func main() { s := semaphore.NewWeighted(1) if s.TryAcquire(1) { fmt.Println("Acquired!") } else { fmt.Println("Not Acquired!") } s.Release(1) } ``` 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.