--- id: 01933e13f963942e4c9bc8c3c89fbc2b author: snowmerak title: Pakiet startowy do współbieżności w Go description: Przedstawiamy różnorodne narzędzia i techniki zarządzania współbieżnością w Go, w tym gorutyny, kanały oraz muteksy. language: pl date: 2024-10-18T05:51:30.010680581Z path: /blog/posts/go-concurrency-starter-pack-z221a3399 category: concurrency --- ## Przegląd ### Krótkie wprowadzenie Język Go posiada wiele narzędzi do zarządzania współbieżnością. W tym artykule przedstawimy niektóre z nich oraz kilka sztuczek. ### Goroutine? goroutine to nowy rodzaj modelu współbieżności obsługiwanego przez język Go. Zwykle program otrzymuje od systemu operacyjnego wątki systemu operacyjnego (OS threads), aby równolegle wykonywać wiele zadań, w liczbie odpowiadającej liczbie rdzeni. Aby realizować współbieżność na mniejszą skalę, w przestrzeni użytkownika (userland) tworzy się zielone wątki (green threads), dzięki czemu wiele zielonych wątków wykonuje zadania naprzemiennie w obrębie jednego wątku systemu operacyjnego. Goroutine sprawiają jednak, że ta forma zielonych wątków jest jeszcze mniejsza i bardziej wydajna. Goroutine zużywają mniej pamięci niż wątki i mogą być tworzone oraz przełączane szybciej niż wątki. Aby użyć goroutine, wystarczy zastosować słowo kluczowe `go`. Pozwala to w sposób intuicyjny podczas pisania programu wykonywać kod synchroniczny jako kod asynchroniczny. ```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 {} } ``` Ten kod w prosty sposób zmienia synchroniczny kod, który czeka 1 sekundę, a następnie wypisuje `Hello, World!`, na przepływ asynchroniczny. Obecny przykład jest prosty, ale jeśli nieco bardziej złożony kod zostanie przekształcony z synchronicznego na asynchroniczny, czytelność, przejrzystość i zrozumiałość kodu będą lepsze niż w dotychczasowych podejściach, takich jak async await czy promise. Jednak w wielu przypadkach, bez zrozumienia przepływu polegającego na prostym wywoływaniu takiego kodu synchronicznego w sposób asynchroniczny oraz przepływu przypominającego `fork & join` (podobnego do dziel i zwyciężaj), powstaje słaby kod z goroutine. W tym artykule przedstawimy kilka metod i technik, które pomagają przygotować się na takie przypadki. ## Zarządzanie współbieżnością ### context To, że pierwszą technikę zarządzania stanowi `context`, może być zaskoczeniem. Jednak w języku Go `context` wykracza poza prostą funkcję anulowania i doskonale sprawdza się w zarządzaniu całym drzewem zadań. Dla osób, które go nie znają, krótko opiszę ten pakiet. ```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) } ``` Powyższy kod za pomocą `context` wypisuje `Context is done!` po 1 sekundzie. `context` pozwala sprawdzić, czy nastąpiło anulowanie, za pomocą metody `Done()`, a metody takie jak `WithCancel`, `WithTimeout`, `WithDeadline` i `WithValue` zapewniają różne sposoby anulowania. Stwórzmy prosty przykład. Załóżmy, że chcesz napisać kod pobierający dane `user`, `post` i `comment` z wykorzystaniem wzorca `aggregator`. Jeśli wszystkie żądania muszą zakończyć się w ciągu 2 sekund, możesz napisać to następująco. ```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!") } } ``` Powyższy kod wypisuje `Timeout!`, jeśli nie uda się pobrać wszystkich danych w ciągu 2 sekund, a jeśli wszystkie dane zostaną pobrane, wypisuje `All data is fetched!`. Używając `context` w ten sposób, możesz łatwo zarządzać anulowaniem i limitami czasu w kodzie, w którym działa wiele goroutine. Różne funkcje i metody związane z context są dostępne w [godoc context](https://pkg.go.dev/context). Mam nadzieję, że po przyswojeniu prostszych elementów można będzie z nich wygodnie korzystać. ### channel #### unbuffered channel `channel` to narzędzie do komunikacji między goroutine. `channel` można utworzyć za pomocą `make(chan T)`. Tutaj `T` oznacza typ danych, które dany `channel` będzie przesyłał. Dane wysyła się i odbiera za pomocą `<-`, a `channel` zamyka się za pomocą `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) } } ``` Powyższy kod za pomocą `channel` wypisuje 1 i 2. W tym kodzie pokazano jedynie wysyłanie i odbieranie wartości przez `channel`. Jednak `channel` oferuje więcej możliwości. Najpierw przyjrzyjmy się `buffered channel` i `unbuffered channel`. Zanim zaczniemy, zauważmy, że przedstawiony przykład używa `unbuffered channel`, w którym wysłanie danych do kanału i ich odebranie muszą nastąpić jednocześnie. Jeśli te działania nie zajdą w tym samym czasie, może dojść do deadlocka. #### buffered channel A co, jeśli powyższy kod wykonuje nie proste wypisywanie, lecz dwa procesy wykonujące ciężkie zadania? Jeśli drugi proces odczytuje i przetwarza dane, a to trwa długo, pierwszy proces również zostanie zatrzymany na ten czas. Aby zapobiec takiej sytuacji, można użyć `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) } } ``` Powyższy kod za pomocą `buffered channel` wypisuje 1 i 2. W tym kodzie `buffered channel` sprawia, że wysłanie danych do `channel` i ich odebranie nie muszą nastąpić jednocześnie. Dzięki buforowi w kanale powstaje zapas o takiej długości, co pozwala zapobiegać opóźnieniom pracy wynikającym z wpływu zadań o niższym priorytecie. #### select Pracując z wieloma kanałami, składnia `select` pozwala łatwo zaimplementować strukturę `fan-in`. ```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) } } } ``` Powyższy kod tworzy 3 kanały, które cyklicznie przekazują wartości 1, 2 i 3, a następnie za pomocą `select` odbiera wartości z kanałów i je wypisuje. Dzięki `select` można jednocześnie odbierać dane z wielu kanałów i przetwarzać je w miarę ich nadejścia. #### for range `channel` można łatwo odczytywać za pomocą `for range`. Gdy `for range` zostanie użyte z kanałem, wykonuje się za każdym razem, gdy do kanału zostanie dodana wartość, a po zamknięciu kanału pętla się kończy. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` Powyższy kod za pomocą `channel` wypisuje 1 i 2. W tym kodzie `for range` odbiera i wypisuje dane za każdym razem, gdy zostaną dodane do kanału. Gdy kanał zostanie zamknięty, pętla się kończy. Jak wspomniano kilka razy wyżej, tę składnię można także użyć jako prostego mechanizmu synchronizacji. ```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 {} } ``` Powyższy kod czeka 1 sekundę, a następnie wypisuje `Hello, World!`. W tym kodzie `channel` został użyty do przekształcenia kodu synchronicznego w asynchroniczny. W ten sposób za pomocą `channel` można łatwo przekształcić kod synchroniczny w asynchroniczny oraz ustawić punkt `join`. #### etc 1. Wysłanie lub odebranie danych z nil channel może spowodować nieskończoną pętlę i deadlock. 2. Wysłanie danych do zamkniętego kanału powoduje panic. 3. Nawet jeśli kanału nie zamkniemy jawnie, GC go usunie, zamykając go przy okazji. ### mutex #### spinlock `spinlock` to metoda synchronizacji, która w pętli wielokrotnie próbuje zdobyć blokadę. W języku Go można łatwo zaimplementować spinlock z użyciem wskaźników. ```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{} } ``` Powyższy kod to implementacja pakietu `spinlock`. Wykorzystano w nim pakiet `sync/atomic` do zaimplementowania `SpinLock`. Metoda `Lock` próbuje zdobyć blokadę za pomocą `atomic.CompareAndSwapUintptr`, a metoda `Unlock` zwalnia blokadę za pomocą `atomic.StoreUintptr`. Ponieważ ta metoda bez przerwy próbuje zdobyć blokadę, do czasu jej uzyskania stale zużywa procesor i może wpaść w nieskończoną pętlę. Dlatego `spinlock` najlepiej stosować do prostej synchronizacji lub tylko przez krótki czas. #### sync.Mutex `mutex` to narzędzie do synchronizacji między goroutine. `mutex` dostarczany przez pakiet `sync` oferuje metody `Lock`, `Unlock`, `RLock`, `RUnlock` i inne. `mutex` można utworzyć jako `sync.Mutex`, a do blokady odczytu/zapisu można użyć `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) } ``` W powyższym kodzie dwie goroutine niemal jednocześnie uzyskują dostęp do tej samej zmiennej `count`. Jeśli kod uzyskujący dostęp do zmiennej `count` ujmiemy w sekcję krytyczną za pomocą `mutex`, można zapobiec równoczesnemu dostępowi do `count`. Wtedy ten kod za każdym razem wypisze `2`, niezależnie od liczby uruchomień. #### sync.RWMutex `sync.RWMutex` to `mutex`, który pozwala rozróżniać blokadę odczytu i blokadę zapisu. Za pomocą metod `RLock` i `RUnlock` można założyć i zwolnić blokadę odczytu. ```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 } ``` Powyższy kod implementuje `ConcurrentMap` z użyciem `sync.RWMutex`. W tym kodzie metoda `Get` zakłada blokadę odczytu, a metoda `Set` blokadę zapisu, dzięki czemu można bezpiecznie uzyskiwać dostęp do mapy `data` i ją modyfikować. Blokada odczytu jest potrzebna, gdy wykonuje się wiele prostych operacji odczytu: zamiast blokować zapis, zakłada się tylko blokadę odczytu, aby wiele goroutine mogło jednocześnie czytać. Dzięki temu w przypadkach, gdy stan nie ulega zmianie i nie trzeba zakładać blokady zapisu, samo założenie blokady odczytu pozwala poprawić wydajność. #### fakelock `fakelock` to prosty trik polegający na implementacji `sync.Locker`. Ta struktura udostępnia te same metody co `sync.Mutex`, ale faktycznie nic nie robi. ```go package fakelock type FakeLock struct{} func (f *FakeLock) Lock() {} func (f *FakeLock) Unlock() {} ``` Powyższy kod to implementacja pakietu `fakelock`. Pakiet implementuje `sync.Locker` i udostępnia metody `Lock` oraz `Unlock`, ale w rzeczywistości nie wykonują one żadnych działań. Dlaczego taki kod jest potrzebny, wyjaśnię innym razem. ### waitgroup #### sync.WaitGroup `sync.WaitGroup` to narzędzie, które czeka, aż wszystkie zadania goroutine zostaną zakończone. Udostępnia metody `Add`, `Done` i `Wait`: metodą `Add` dodaje się liczbę goroutine, metodą `Done` sygnalizuje się zakończenie zadania goroutine, a metodą `Wait` czeka się, aż zakończą się wszystkie goroutine. ```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()) } ``` Powyższy kod za pomocą `sync.WaitGroup` pozwala 100 goroutine jednocześnie dodawać wartości do zmiennej `c`. W tym kodzie `sync.WaitGroup` czeka, aż wszystkie goroutine zakończą pracę, a następnie wypisuje wartość dodaną do `c`. W przypadku prostego `fork & join` kilku zadań wystarczy sam kanał, ale przy `fork & join` dużej liczby zadań dobrym wyborem jest `sync.WaitGroup`. #### with slice W połączeniu z sliceem `waitgroup` może stać się doskonałym narzędziem do zarządzania równoległym wykonywaniem zadań bez użycia blokad. ```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) } } ``` Powyższy kod za pomocą samego `waitgroup` pozwala każdej z goroutine jednocześnie wygenerować 10 losowych liczb całkowitych i zapisać je pod przydzielonym indeksem. W tym kodzie `waitgroup` czeka, aż wszystkie goroutine zakończą pracę, a następnie wypisuje `Done`. W ten sposób za pomocą `waitgroup` wiele goroutine może jednocześnie wykonywać zadania, zapisywać dane bez blokad, aż wszystkie się zakończą, i dopiero potem zbiorczo przeprowadzić dalsze przetwarzanie. #### golang.org/x/sync/errgroup.ErrGroup `errgroup` to pakiet rozszerzający `sync.WaitGroup`. W przeciwieństwie do `sync.WaitGroup`, `errgroup` w razie wystąpienia błędu w choćby jednym zadaniu goroutine anuluje wszystkie goroutine i zwraca błąd. ```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) } } ``` Powyższy kod za pomocą `errgroup` tworzy 10 goroutine, z których piąta zgłasza błąd. Celowo wywołano błąd w piątej goroutine, aby pokazać sytuację, w której błąd występuje. W praktyce `errgroup` używa się do tworzenia goroutine i stosowania różnorodnych dalszych działań w przypadku wystąpienia błędu w każdej z nich. ### once Narzędzie do wykonywania kodu, który powinien zostać uruchomiony tylko raz. Powiązany kod można uruchomić za pomocą poniższych konstruktorów. ```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` po prostu zapewnia, że dana funkcja zostanie wykonana dokładnie raz w całym programie. ```go package main import "sync" func main() { once := sync.OnceFunc(func() { println("Hello, World!") }) once() once() once() once() once() } ``` Powyższy kod za pomocą `sync.OnceFunc` wypisuje `Hello, World!`. W tym kodzie `sync.OnceFunc` tworzy funkcję `once`, a mimo wielokrotnego wywołania `once` tekst `Hello, World!` zostaje wypisany tylko raz. #### OnceValue `OnceValue` nie tylko zapewnia, że dana funkcja wykona się dokładnie raz w całym programie, ale także zapisuje jej wartość zwrotną i przy kolejnych wywołaniach zwraca zapisaną wartość. ```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()) } ``` Powyższy kod za pomocą `sync.OnceValue` zwiększa zmienną `c` o 1. W tym kodzie `sync.OnceValue` tworzy funkcję `once`, a mimo wielokrotnego jej wywołania zmienna `c` zwiększa się tylko raz i zwracana jest wartość 1. #### OnceValues `OnceValues` działa tak samo jak `OnceValue`, ale może zwracać wiele wartości. ```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) } ``` Powyższy kod za pomocą `sync.OnceValues` zwiększa zmienną `c` o 1. W tym kodzie `sync.OnceValues` tworzy funkcję `once`, a mimo wielokrotnego jej wywołania zmienna `c` zwiększa się tylko raz i zwracana jest wartość 1. ### atomic Pakiet `atomic` udostępnia operacje atomowe. Oferuje metody `Add`, `CompareAndSwap`, `Load`, `Store`, `Swap` i inne, ale ostatnio zaleca się korzystanie z typów takich jak `Int64`, `Uint64`, `Pointer`. ```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()) } ``` To przykład, który pojawił się wcześniej. Używa typu `atomic.Int64`, aby atomowo zwiększać zmienną `c`. Metodami `Add` i `Load` można atomowo zwiększać zmienną i ją odczytywać. Ponadto metodą `Store` zapisuje się wartość, metodą `Swap` zamienia się wartość, a metodą `CompareAndSwap` porównuje się wartość i, jeśli pasuje, ją podmienia. ### cond #### sync.Cond Pakiet `cond` dostarcza zmienne warunkowe. Można go utworzyć jako `sync.Cond` i udostępnia metody `Wait`, `Signal` i `Broadcast`. ```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!") } ``` Powyższy kod za pomocą `sync.Cond` czeka, aż zmienna `ready` przyjmie wartość `true`. W tym kodzie `sync.Cond` czeka, aż `ready` stanie się `true`, a następnie wypisuje `Ready!`. Dzięki `sync.Cond` wiele goroutine może czekać, aż zostanie spełniony określony warunek. Wykorzystując to, można zaimplementować prostą `queue`. ```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 } ``` Dzięki `sync.Cond` zamiast zużywać dużo procesora w `spin-lock`, można efektywnie czekać, a po spełnieniu warunku wznowić działanie. ### semaphore #### golang.org/x/sync/semaphore.Semaphore Pakiet `semaphore` dostarcza semafory. Można go utworzyć jako `golang.org/x/sync/semaphore.Semaphore`, a udostępnia metody `Acquire`, `Release` i `TryAcquire`. ```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) } ``` Powyższy kod tworzy semafor za pomocą `semaphore`, a następnie pozyskuje go metodą `Acquire` i zwalnia metodą `Release`. W tym kodzie pokazano, jak pozyskiwać i zwalniać semafor za pomocą `semaphore`. ## Zakończenie Na tym etapie wystarczy podstawowa wiedza. Mam nadzieję, że na podstawie tego artykułu zrozumiesz, jak zarządzać współbieżnością za pomocą goroutine, i będziesz mógł/mogła z tego korzystać w praktyce. Mam nadzieję, że ten artykuł był dla Ciebie pomocny. Dziękuję.