--- id: 01933e13f963942e4c9bc8c3c89fbc2b author: snowmerak title: Go párhuzamossági kezdőcsomag description: Bemutatjuk a Go nyelvben a párhuzamosság kezelésére szolgáló különböző eszközöket és technikákat, például a goroutine-okat, a csatornákat (channel) és a mutexeket. language: hu date: 2024-10-18T05:51:30.010680581Z path: /blog/posts/go-concurrency-starter-pack-z221a3399 category: concurrency --- ## Áttekintés ### Rövid bevezetés A Go nyelvben számos eszköz áll rendelkezésre a párhuzamosság kezelésére. Ebben a cikkben ezek közül néhányat és néhány fogást szeretnék bemutatni. ### Goroutine? A goroutine a Go nyelv által támogatott, új típusú párhuzamossági modell. Általában a program az egyidejű több feladat végrehajtásához OS-szálakat kap az operációs rendszertől, és ezeket a processzormagok számával megegyező mértékben párhuzamosan futtatja. Kisebb egységű párhuzamosság eléréséhez pedig felhasználói térben zöld szálakat hozunk létre, így egyetlen OS-szálon belül több zöld szál váltakozva hajtja végre a feladatokat. A goroutine-ok esetében azonban ezt a zöld szál jellegű megoldást még kisebbé és hatékonyabbá tették. Ezek a goroutine-ok kevesebb memóriát használnak, mint a szálak, és gyorsabban hozhatók létre és cserélhetők le, mint azok. A goroutine használatához egyszerűen csak a `go` kulcsszót kell alkalmazni. Ez a program írása során intuitív módon teszi lehetővé, hogy a szinkron kódot aszinkron módon futtassuk. ```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 {} } ``` Ez a kód egy 1 másodpercig tartó szünet után kiírja a `Hello, World!` üzenetet, és a szinkron kódot aszinkron folyamattá alakítja. A jelenlegi példa egyszerű, de ha valamivel bonyolultabb kódot alakítunk át szinkron kódból aszinkron kóddá, a kód olvashatósága, átláthatósága és érthetősége jobb lesz, mint a hagyományos async await vagy promise jellegű megoldásokkal. Sok esetben azonban, ha nem értjük meg az ilyen szinkron kódok egyszerű aszinkron hívásának folyamatát, valamint a `fork & join` jellegű folyamatot (amely hasonló a divide and conquer, azaz oszd meg és uralkodj módszerhez), rossz goroutine-kód születhet. Ezekre az esetekre készülve ebben a cikkben bemutatok néhány módszert és technikát. ## Párhuzamosság kezelése ### context Talán meglepő, hogy az első kezelési technika a `context`. A Go nyelvben azonban a `context` a puszta megszakítási funkciónál többet nyújt, és kiválóan alkalmas a teljes feladatfa kezelésére. Ha valakinek nem ismerős a csomag, röviden bemutatom. ```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) } ``` A fenti kód a `context` segítségével 1 másodperc elteltével kiírja a `Context is done!` üzenetet. A `context` a `Done()` metóduson keresztül ellenőrizhető, és a `WithCancel`, `WithTimeout`, `WithDeadline`, `WithValue` stb. metódusokon keresztül különböző megszakítási módokat biztosít. Készítsünk egy egyszerű példát. Tegyük fel, hogy egy `aggregator` mintát alkalmazva írunk kódot, amely a `user`, `post` és `comment` adatokat kéri le, és az összes kérésnek 2 másodpercen belül kell megtörténnie. Ekkor a következőképpen írhatjuk meg: ```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!") } } ``` A fenti kód, ha 2 másodpercen belül nem sikerül lekérni az összes adatot, kiírja a `Timeout!` üzenetet, ha pedig minden adat megérkezik, a `All data is fetched!` üzenetet. Ilyen módon a `context` használatával több goroutine-t futtató kódban is egyszerűen kezelhetők a megszakítások és az időtúllépések. A context-hez kapcsolódó különféle függvények és metódusok a [godoc context](https://pkg.go.dev/context) oldalon tekinthetők meg. Remélem, hogy az egyszerűbb részeket megtanulva kényelmesen tudják majd használni. ### channel #### unbuffered channel A `channel` a goroutine-ok közötti kommunikáció eszköze. A `channel` a `make(chan T)` segítségével hozható létre. Itt a `T` az adott `channel` által továbbított adat típusa. A `channel`en a `<-` operátorral lehet adatot küldeni és fogadni, a `channel` pedig a `close` függvénnyel zárható le. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` A fenti kód a `channel` segítségével kiírja az 1-et és a 2-t. Ebben a kódban csupán egyszerű küldést és fogadást mutattam be. A `channel` ennél több funkciót is kínál. Először a `buffered channel` és az `unbuffered channel` fogalmát ismertetem. Mielőtt továbbmennénk, megjegyzem, hogy a fenti példa `unbuffered channel`, amelyben az adat küldésének és fogadásának egyidejűleg kell megtörténnie. Ha ezek a műveletek nem egyszerre történnek meg, deadlock léphet fel. #### buffered channel Mi a helyzet akkor, ha a fenti kód nem egyszerű kiírást végez, hanem két nehéz feladatot végző folyamatról van szó? Ha a második folyamat a beolvasás után hosszú ideig tart a feldolgozással, akkor az első folyamat is megáll erre az időre. Az ilyen helyzet elkerülésére használhatunk `buffered channel`t. ```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) } } ``` A fenti kód a `buffered channel` segítségével kiírja az 1-et és a 2-t. Ebben a kódban a `buffered channel` használatával elértük, hogy a `channel`be történő küldés és a fogadás nem kell, hogy egyidejűleg történjen meg. Ha a csatornán puffer van, akkor a pufferhossznak megfelelő tartalék keletkezik, így megelőzhető a későbbi feladatok hatására fellépő munkakésedelem. #### select Több csatorna kezelésekor a `select` szintaxis használatával könnyen megvalósítható a `fan-in` szerkezet. ```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) } } } ``` A fenti kód három csatornát hoz létre, amelyek rendszeres időközönként az 1, 2 és 3 értékeket továbbítják, a `select` segítségével pedig a csatornákról érkező értékeket kiírja. Ilyen módon a `select` használatával egyszerre több csatornán érkező adatokat fogadhatunk, és az értékeket abban a sorrendben dolgozhatjuk fel, ahogyan megérkeznek. #### for range A `channel` a `for range` segítségével egyszerűen fogadhat adatokat. Ha a `for range`et csatornára alkalmazzuk, akkor minden alkalommal lefut, amikor a csatornára új adat kerül, a ciklus pedig a csatorna lezárásakor ér véget. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` A fenti kód a `channel` segítségével kiírja az 1-et és a 2-t. Ebben a kódban a `for range` használatával minden alkalommal fogadjuk és kiírjuk az adatot, amikor új adat kerül a csatornára. A csatorna lezárásakor pedig a ciklus véget ér. Ahogy korábban többször leírtam, ez a szintaxis egyszerű szinkronizációs eszközként is használható. ```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 {} } ``` A fenti kód 1 másodpercnyi szünet után kiírja a `Hello, World!` üzenetet. Ebben a kódban a `channel` segítségével a szinkron kódot aszinkron kóddá alakítottuk. Ilyen módon a `channel` használatával könnyen átalakíthatjuk a szinkron kódot aszinkron kóddá, és kijelölhetjük a `join` pontot. #### etc 1. Ha nil channelbe küldünk vagy abból fogadunk adatot, végtelen ciklusba kerülhetünk, ami deadlockot okozhat. 2. Ha a csatornát lezárás után küldünk bele adatot, panic lép fel. 3. Nem feltétlenül kell lezárni a csatornát, mivel a GC összegyűjtéskor lezárja azt. ### mutex #### spinlock A `spinlock` egy olyan szinkronizációs módszer, amely ciklusban folyamatosan megpróbálja megszerezni a zárat. A Go nyelvben mutatók segítségével egyszerűen implementálhatunk spinlockot. ```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{} } ``` A fenti kód a `spinlock` csomag megvalósítása. Ebben a kódban a `sync/atomic` csomag segítségével implementáltuk a `SpinLock`ot. A `Lock` metódusban az `atomic.CompareAndSwapUintptr` használatával próbáljuk megszerezni a zárat, az `Unlock` metódusban pedig az `atomic.StoreUintptr` segítségével szabadítjuk fel. Ez a módszer megállás nélkül próbálkozik a zár megszerzésével, ezért a zár megszerzéséig folyamatosan terheli a CPU-t, és végtelen ciklusba kerülhet. Ezért a `spinlock`ot célszerű egyszerű szinkronizációra vagy csak rövid ideig tartó használatra alkalmazni. #### sync.Mutex A `mutex` a goroutine-ok közötti szinkronizáció eszköze. A `sync` csomag által biztosított `mutex` a `Lock`, `Unlock`, `RLock`, `RUnlock` stb. metódusokat kínálja. A `mutex` `sync.Mutex` típussal hozható létre, a `sync.RWMutex` típussal pedig olvasási/írási zár is használható. ```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) } ``` A fenti kódban szinte egyidejűleg két goroutine fér hozzá ugyanahhoz a `count` változóhoz. Ha a `count` változóhoz történő hozzáférést tartalmazó kódot `mutex` segítségével kritikus szakaszként jelöljük, megakadályozhatjuk a `count` változó egyidejű elérését. Ekkor a kód bárhányszor futtatva mindig `2`-t fog kiírni. #### sync.RWMutex A `sync.RWMutex` olyan `mutex`, amely az olvasási és az írási zárakat megkülönböztetve használható. Az `RLock` és `RUnlock` metódusokkal olvasási zárat helyezhetünk el és oldhatunk fel. ```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 } ``` A fenti kód a `sync.RWMutex` segítségével valósítja meg a `ConcurrentMap`ot. Ebben a kódban a `Get` metódus olvasási zárat, a `Set` metódus írási zárat alkalmaz, így a `data` map biztonságosan érhető el és módosítható. Az olvasási zárra azért van szükség, mert ha sok az egyszerű olvasási művelet, akkor íráscsak olvasási zárat alkalmazva több goroutine is egyidejűleg végezhet olvasást. Így azokban az esetekben, amikor az állapot megváltoztatása nem szükséges, és nem kell írási zárat alkalmazni, csupán olvasási zárral javítható a teljesítmény. #### fakelock A `fakelock` a `sync.Locker` interfészt megvalósító egyszerű trükk. Ez a struktúra ugyanazokat a metódusokat biztosítja, mint a `sync.Mutex`, de valójában nem végez semmilyen műveletet. ```go package fakelock type FakeLock struct{} func (f *FakeLock) Lock() {} func (f *FakeLock) Unlock() {} ``` A fenti kód a `fakelock` csomag megvalósítása. Ez a csomag implementálja a `sync.Locker` interfészt, és biztosítja a `Lock` és `Unlock` metódusokat, de valójában egyiknek sincs műveleti hatása. Azt, hogy miért van szükség ilyen kódra, alkalmi jelleggel később fogom leírni. ### waitgroup #### sync.WaitGroup A `sync.WaitGroup` olyan eszköz, amely megvárja, amíg a goroutine-ok feladatai mind befejeződnek. Az `Add`, `Done`, `Wait` metódusokat biztosítja: az `Add` metódussal növelhetjük a goroutine-ok számát, a `Done` metódussal jelezzük, hogy egy goroutine feladata befejeződött. A `Wait` metódussal pedig megvárjuk, amíg az összes goroutine feladata befejeződik. ```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()) } ``` A fenti kód a `sync.WaitGroup` segítségével 100 goroutine-t indít, amelyek egyidejűleg adnak hozzá értéket a `c` változóhoz. Ebben a kódban a `sync.WaitGroup` segítségével megvárjuk, amíg az összes goroutine befejeződik, majd kiírjuk a `c` változóba összegyűjtött értéket. Néhány feladat `fork & join` jellegű kezelésére egyszerűen elegendő csak a csatornákat használni, nagyszámú feladat `fork & join` kezelésénél azonban a `sync.WaitGroup` használata is jó választás. #### with slice Ha szelettel (slice) együtt használjuk, a `waitgroup` zár nélkül is kiváló eszköze lehet a párhuzamosan futó feladatok kezelésének. ```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) } } ``` A fenti kód kizárólag `waitgroup` használatával valósítja meg, hogy minden goroutine egyidejűleg generál egy-egy véletlen egész számot 10 darabot, és ezeket a kiosztott indexre menti. Ebben a kódban a `waitgroup` segítségével megvárjuk, amíg az összes goroutine befejeződik, majd kiírjuk a `Done` szót. Ilyen módon a `waitgroup` használatával a goroutine-ok egyidejűleg dolgozhatnak, zár nélkül menthetik az adatokat, a feladatok befejezése után pedig egységesen elvégezhetjük az utófeldolgozást. #### golang.org/x/sync/errgroup.ErrGroup Az `errgroup` a `sync.WaitGroup` továbbfejlesztett csomagja. Az `errgroup` a `sync.WaitGroup`tól eltérően, ha a goroutine-ok feladatai közül bármelyikben hiba lép fel, megszakítja az összes goroutine-t, és visszaadja a hibát. ```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) } } ``` A fenti kód az `errgroup` segítségével 10 goroutine-t hoz létre, és az 5. goroutine-ban hibát idéz elő. A szándékosan az ötödik goroutine-ban előidézett hiba segítségével mutattam be a hibás működés esetét. Valós használat során az `errgroup`-pal hozunk létre goroutine-okat, és az egyes goroutine-okban fellépő hibák esetén különféle utófeldolgozást végezhetünk. ### once Olyan eszköz, amely azt a kódot futtatja, amelynek csak egyszer szabad lefutnia. Az alábbi konstruktorokkal futtathatjuk a kapcsolódó kódot. ```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 Az `OnceFunc` egyszerűen biztosítja, hogy az adott függvény az egész futás során pontosan egyszer hajtódjon végre. ```go package main import "sync" func main() { once := sync.OnceFunc(func() { println("Hello, World!") }) once() once() once() once() once() } ``` A fenti kód a `sync.OnceFunc` segítségével a `Hello, World!` üzenetet írja ki. Ebben a kódban a `sync.OnceFunc` használatával létrehozunk egy `once` függvényt, és a `once` függvényt többszöri hívása esetén is a `Hello, World!` csak egyszer íródik ki. #### OnceValue Az `OnceValue` nem egyszerűen biztosítja, hogy az adott függvény az egész futás során pontosan egyszer hajtódjon végre, hanem eltárolja a függvény visszatérési értékét, és újrahíváskor az eltárolt értéket adja vissza. ```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()) } ``` A fenti kód a `sync.OnceValue` segítségével a `c` változót eggyel növeli. Ebben a kódban a `sync.OnceValue` használatával létrehozunk egy `once` függvényt, és a `once` függvényt többszöri hívása esetén is a `c` változó csak egyszer növekszik, így az mindig az 1-es értéket adja vissza. #### OnceValues Az `OnceValues` az `OnceValue`-hoz hasonlóan működik, de több értéket is képes visszaadni. ```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) } ``` A fenti kód a `sync.OnceValues` segítségével a `c` változót eggyel növeli. Ebben a kódban a `sync.OnceValues` használatával létrehozunk egy `once` függvényt, és a `once` függvényt többszöri hívása esetén is a `c` változó csak egyszer növekszik, így az mindig az 1-es értéket adja vissza. ### atomic Az `atomic` csomag atomi műveleteket biztosító csomag. Az `atomic` csomag az `Add`, `CompareAndSwap`, `Load`, `Store`, `Swap` stb. metódusokat kínálja, az újabb verziókban azonban az `Int64`, `Uint64`, `Pointer` stb. típusok használata javasolt. ```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()) } ``` Ez a példa már korábban is szerepelt. Az `atomic.Int64` típus segítségével a `c` változót atomikusan növeljük. Az `Add` és `Load` metódusokkal atomikusan növelhetjük a változót, illetve beolvashatjuk annak értékét. Emellett a `Store` metódussal értéket tárolhatunk, a `Swap` metódussal értéket cserélhetünk, a `CompareAndSwap` metódussal pedig értéket hasonlíthatunk össze, és ha az megfelel, lecserélhetjük. ### cond #### sync.Cond A `cond` csomag feltételváltozókat biztosító csomag. A `cond` csomag `sync.Cond` típussal hozható létre, és a `Wait`, `Signal`, `Broadcast` metódusokat kínálja. ```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!") } ``` A fenti kód a `sync.Cond` segítségével addig vár, amíg a `ready` változó értéke `true` nem lesz. Ebben a kódban a `sync.Cond` használatával addig várunk, amíg a `ready` változó értéke `true` nem lesz, majd kiírjuk a `Ready!` üzenetet. Ilyen módon a `sync.Cond` használatával több goroutine-t is várakoztathatunk addig, amíg azok egy adott feltételnek meg nem felelnek. Ennek felhasználásával egyszerű `queue` is megvalósítható. ```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 } ``` Ha így a `sync.Cond`-ot használjuk, a `spin-lock` által okozott nagy processzorhasználat helyett hatékonyan várakozhatunk, és ha a feltétel teljesül, újra folytathatjuk a működést. ### semaphore #### golang.org/x/sync/semaphore.Semaphore A `semaphore` csomag szemafort biztosító csomag. A `semaphore` csomag a `golang.org/x/sync/semaphore.Semaphore` típussal hozható létre, és az `Acquire`, `Release`, `TryAcquire` metódusokat kínálja. ```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) } ``` A fenti kód a `semaphore` segítségével szemafort hoz létre, és az `Acquire` metódussal megszerzi, a `Release` metódussal pedig felszabadítja a szemafort. Ebben a kódban a `semaphore` használatával bemutattam a szemafor megszerzésének és felszabadításának módját. ## Összegzés Az alapvető tartalom ennyi elegendő lehet. A cikk anyagai alapján remélem, hogy megérti, hogyan kezelheti a párhuzamosságot goroutine-ok segítségével, és képes lesz azt a gyakorlatban is alkalmazni. Remélem, hogy ez a cikk hasznos volt az Ön számára. Köszönöm a figyelmet.