--- id: 01933e13f963942e4c9bc8c3c89fbc2b author: snowmerak title: Go concurrency -aloituspaketti description: Tämä esittelee erilaisia työkaluja ja tekniikoita Go-kielen samanaikaisuuden hallintaan, kuten goroutine-säikeitä, kanavia ja mutex-lukkoja. language: fi date: 2024-10-18T05:51:30.010680581Z path: /blog/posts/go-concurrency-starter-pack-z221a3399 category: concurrency --- ## Yleiskatsaus ### Lyhyt johdanto Go-kielessä on monia työkaluja samanaikaisuuden hallintaan. Tässä artikkelissa esittelen osan niistä sekä joitakin temppuja. ### Goroutine? goroutine on Go-kielen tukema uusi muoto samanaikaisuusmallista. Yleensä ohjelma saa käyttöjärjestelmältä OS-säikeitä suorittaakseen useita tehtäviä samanaikaisesti, ja suorittaa niitä rinnakkain ytimien määrän verran. Pienemmän mittakaavan samanaikaisuuden toteuttamiseksi luodaan käyttäjätilassa green threadeja, jolloin useat green threadit suorittavat tehtäviä vuorotellen yhden OS-säikeen sisällä. Goroutinet ovat kuitenkin tehneet tästä green thread -muodosta entistä pienemmän ja tehokkaamman. Goroutinet käyttävät vähemmän muistia kuin säikeet, ja niitä voidaan luoda ja vaihtaa nopeammin kuin säikeitä. Goroutinea käytetään yksinkertaisesti avainsanalla `go`. Tämä mahdollistaa synkronisen koodin suorittamisen intuitiivisesti asynkronisena ohjelmoinnin aikana. ```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 {} } ``` Tämä koodi muuttaa yksinkertaisen synkronisen koodin, joka odottaa sekunnin ja tulostaa `Hello, World!`, asynkroniseksi kulkuun. Vaikka tämä esimerkki on yksinkertainen, kun monimutkaisempi koodi muutetaan synkronisesta asynkroniseksi, koodin luettavuus, näkyvyys ja ymmärrettävyys paranevat verrattuna perinteisiin tapoihin, kuten async await tai promise. Kuitenkin monissa tapauksissa, jos ei ymmärretä pelkästään synkronisen koodin kutsumista asynkronisesti ja `fork & join` -tyyppistä kulkua (joka muistuttaa divide and conquer -menetelmää), syntyy huonoja goroutine-koodeja. Tähän tilanteeseen voi varautua tässä artikkelissa esiteltävillä menetelmillä ja tekniikoilla. ## Samanaikaisuuden hallinta ### context Saattaa olla yllättävää, että ensimmäinen hallintatekniikka on `context`. Go-kielessä `context` ylittää kuitenkin pelkän peruutustoiminnon ja toimii erinomaisesti koko tehtäväpuun hallinnassa. Jos paketti on jollekulle tuntematon, selitän sen lyhyesti. ```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) } ``` Yllä oleva koodi käyttää `context`ia ja tulostaa `Context is done!` yhden sekunnin kuluttua. `context` tarjoaa tiedon peruutuksesta `Done()`-metodin kautta, ja tarjoaa erilaisia peruutustapoja metodien `WithCancel`, `WithTimeout`, `WithDeadline` ja `WithValue` kautta. Tehdään yksinkertainen esimerkki. Oletetaan, että haluat hakea dataa `aggregator`-mallilla ja kirjoittaa koodin, joka hakee `user`, `post` ja `comment`. Jos kaikkien pyyntöjen on tapahduttava 2 sekunnin sisällä, voit kirjoittaa sen seuraavasti. ```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!") } } ``` Yllä oleva koodi tulostaa `Timeout!`, jos kaikkia tietoja ei saada haettua 2 sekunnin sisällä, ja tulostaa `All data is fetched!`, jos kaikki tiedot haetaan. Näin `context`ia käyttämällä voidaan helposti hallita peruutuksia ja aikakatkaisuja myös koodissa, jossa useat goroutinet toimivat. Lisää context-funktioita ja -metodeja löytyy [godoc context](https://pkg.go.dev/context)-sivulta. Toivon, että opit yksinkertaiset asiat ja pystyt käyttämään niitä vaivattomasti. ### channel #### unbuffered channel `channel` on työkalu goroutinejen väliseen kommunikaatioon. `channel`in voi luoda `make(chan T)`-funktiolla. Tässä `T` on tyyppi, jota kanava välittää. Dataa lähetetään ja vastaanotetaan `<-`-operaattorilla, ja `channel` suljetaan `close`-funktiolla. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` Yllä oleva koodi tulostaa 1 ja 2 käyttäen `channel`ia. Tässä koodissa näytetään vain yksinkertainen arvon lähettäminen ja vastaanottaminen kanavaan. `channel` tarjoaa kuitenkin enemmän ominaisuuksia. Tutustutaan ensin `buffered channel`- ja `unbuffered channel`-kanaviin. Ennen kuin jatkamme, huomaa että yllä oleva esimerkki käyttää `unbuffered channel`ia, jossa kanavaan lähettäminen ja siitä vastaanottaminen on tapahduttava samanaikaisesti. Jos näin ei tapahdu, voi syntyä deadlock. #### buffered channel Entä jos yllä oleva koodi ei vain tulostaisi, vaan suorittaisi kaksi raskasta prosessia? Jos toinen prosessi lukee ja käsittelee dataa ja jumittuu pitkäksi aikaa, myös ensimmäinen prosessi pysähtyy siksi ajaksi. Tämän tilanteen välttämiseksi voimme käyttää `buffered channel`ia. ```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) } } ``` Yllä oleva koodi tulostaa 1 ja 2 käyttäen `buffered channel`ia. Tässä koodissa `buffered channel`in avulla kanavaan lähettäminen ja siitä vastaanottaminen eivät välttämättä tapahdu samanaikaisesti. Kun kanavalla on puskuri, siinä on varaa puskurin koon verran, joten myöhemmän tehtävän aiheuttamia viiveitä voidaan välttää. #### select Kun käsitellään useita kanavia, `select`-syntaksilla on helppo toteuttaa `fan-in`-rakenne. ```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) } } } ``` Yllä oleva koodi luo kolme kanavaa, jotka lähettävät säännöllisesti arvot 1, 2 ja 3, ja tulostaa kanavista vastaanotetut arvot `select`in avulla. Näin `select`illä voidaan vastaanottaa dataa useilta kanavilta samanaikaisesti ja käsitellä arvoja heti, kun ne saapuvat. #### for range `channel`ia voi lukea helposti `for range`-silmukalla. Kun `for range`ia käytetään kanavaan, se suoritetaan aina, kun kanavaan lisätään dataa, ja silmukka päättyy, kun kanava suljetaan. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` Yllä oleva koodi tulostaa 1 ja 2 käyttäen `channel`ia. Tässä koodissa `for range` vastaanottaa ja tulostaa dataa aina, kun kanavaan lisätään dataa. Kun kanava suljetaan, silmukka päättyy. Kuten edellä useaan kertaan kirjoitettiin, tätä syntaksia voidaan käyttää myös yksinkertaisena synkronointikeinona. ```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 {} } ``` Yllä oleva koodi odottaa yhden sekunnin ja tulostaa `Hello, World!`. Tässä koodissa synkroninen koodi on muutettu asynkroniseksi käyttämällä `channel`ia. Näin `channel`ia käyttämällä synkronisen koodin voi helposti muuttaa asynkroniseksi ja määrittää `join`-kohdan. #### etc 1. Jos nil-kanavaan lähetetään dataa tai siltä vastaanotetaan dataa, ohjelma voi jäädä ikuiseen silmukkaan ja deadlock voi syntyä. 2. Jos kanavaan lähetetään dataa sen jälkeen, kun se on suljettu, syntyy panic. 3. Vaikka kanavaa ei sulkisi erikseen, GC kerää sen ja sulkee kanavan. ### mutex #### spinlock `spinlock` on synkronointimenetelmä, joka yrittää lukita toistuvasti silmukassa. Go-kielessä spinlockin voi toteuttaa helposti käyttämällä osoittimia. ```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{} } ``` Yllä oleva koodi on `spinlock`-paketin toteutus. Tässä koodissa `SpinLock` on toteutettu `sync/atomic`-paketilla. `Lock`-metodi yrittää lukitusta `atomic.CompareAndSwapUintptr`-funktiolla, ja `Unlock`-metodi vapauttaa lukon `atomic.StoreUintptr`-funktiolla. Koska tämä menetelmä yrittää lukitusta jatkuvasti, se käyttää CPU:ta, kunnes lukko saadaan, ja voi joutua ikuiseen silmukkaan. Siksi `spinlock`ia kannattaa käyttää vain yksinkertaiseen synkronointiin tai lyhyeksi ajaksi. #### sync.Mutex `mutex` on työkalu goroutinejen väliseen synkronointiin. `sync`-paketin `mutex` tarjoaa metodit `Lock`, `Unlock`, `RLock`, `RUnlock` ja muita. `mutex` luodaan `sync.Mutex`-tyypillä, ja `sync.RWMutex`-tyypillä voi käyttää luku/kirjoitus-lukkoa. ```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) } ``` Yllä olevassa koodissa kaksi goroutinea käyttää lähes samanaikaisesti samaa `count`-muuttujaa. Kun `count`-muuttujaan käsiksi pääsevä koodi tehdään kriittiseksi alueeksi `mutex`in avulla, voidaan estää `count`-muuttujan samanaikainen käyttö. Silloin tämä koodi tulostaa aina `2`, riippumatta siitä, montako kertaa sen suorittaa. #### sync.RWMutex `sync.RWMutex` on `mutex`, jossa luku- ja kirjoituslukot voidaan erottaa toisistaan. Luku-lukko voidaan asettaa ja vapauttaa metodeilla `RLock` ja `RUnlock`. ```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 } ``` Yllä oleva koodi toteuttaa `ConcurrentMap`in käyttäen `sync.RWMutex`ia. Tässä koodissa `Get`-metodi asettaa luku-lukon ja `Set`-metodi kirjoitus-lukon, joten `data`-karttaan voidaan käsiksi turvallisesti lukea ja muokata. Luku-lukko on tarpeen, kun lukuoperaatioita on paljon: sen sijaan että kirjoitus-lukko asetettaisiin, asetetaan pelkkä luku-lukko, jolloin useat goroutinet voivat lukea samanaikaisesti. Näin, jos tilaa ei muuteta eikä kirjoitus-lukkoa tarvita, pelkkä luku-lukko parantaa suorituskykyä. #### fakelock `fakelock` on yksinkertainen temppu, joka toteuttaa `sync.Locker`in. Tämä rakenne tarjoaa samat metodit kuin `sync.Mutex`, mutta ei todellisuudessa tee mitään. ```go package fakelock type FakeLock struct{} func (f *FakeLock) Lock() {} func (f *FakeLock) Unlock() {} ``` Yllä oleva koodi on `fakelock`-paketin toteutus. Tämä paketti toteuttaa `sync.Locker`in ja tarjoaa `Lock`- ja `Unlock`-metodit, mutta ne eivät todellisuudessa tee mitään. Miksi tällaista koodia tarvitaan, selitän myöhemmin, kun siihen on sopiva tilaisuus. ### waitgroup #### sync.WaitGroup `sync.WaitGroup` on työkalu, joka odottaa, että kaikki goroutinejen tehtävät ovat valmiita. Se tarjoaa metodit `Add`, `Done` ja `Wait`. `Add`-metodilla lisätään goroutinejen määrää, `Done`-metodilla ilmoitetaan goroutinen tehtävän valmistumisesta, ja `Wait`-metodilla odotetaan, että kaikkien goroutinejen tehtävät ovat valmiita. ```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()) } ``` Yllä oleva koodi käyttää `sync.WaitGroup`ia, jolloin 100 goroutinea lisää samanaikaisesti arvoa muuttujaan `c`. Tässä koodissa `sync.WaitGroup`ia käytetään odottamaan, että kaikki goroutinet ovat valmiita, minkä jälkeen tulostetaan muuttujaan `c` kertynyt arvo. Jos `fork & join` koskee vain muutamaa tehtävää, pelkkä kanava riittää, mutta suuren määrän tehtäviä `fork & join`-kuviossa `sync.WaitGroup` on hyvä vaihtoehto. #### Käyttö slicen kanssa Kun sitä käytetään yhdessä slicen kanssa, `waitgroup` voi olla erinomainen työkalu samanaikaisten tehtävien hallintaan ilman lukkoja. ```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) } } ``` Yllä oleva koodi käyttää pelkkää `waitgroup`ia, jolloin jokainen goroutine luo samanaikaisesti satunnaisen kokonaisluvun ja tallentaa sen sille osoitettuun indeksiin. Tässä koodissa `waitgroup`ia käytetään odottamaan, että kaikki goroutinet ovat valmiita, minkä jälkeen tulostetaan `Done`. Näin `waitgroup`ia käyttämällä useat goroutinet voivat suorittaa tehtäviä samanaikaisesti, tallentaa dataa ilman lukkoa, ja tehdä yhteisen jälkikäsittelyn kaikkien valmistuttua. #### golang.org/x/sync/errgroup.ErrGroup `errgroup` on `sync.WaitGroup`in laajennus. Toisin kuin `sync.WaitGroup`, `errgroup` peruuttaa kaikki goroutinet ja palauttaa virheen, jos jokin goroutinen tehtävä aiheuttaa virheen. ```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) } } ``` Yllä oleva koodi luo `errgroup`ia käyttäen 10 goroutinea, joista viides aiheuttaa virheen. Aiheutin virheen tarkoituksella viidennessä goroutinessa, jotta näytän virhetilanteen. Käytännössä `errgroup`ia käytetään siten, että goroutineja luodaan ja jokaisessa goroutinessa voidaan toteuttaa erilaisia jälkikäsittelyjä virhetilanteissa. ### once Työkalu, joka suorittaa koodin, jonka pitäisi suoriutua vain kerran. Alla olevilla konstruktoreilla voit suorittaa siihen liittyvän koodin. ```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` mahdollistaa yksinkertaisesti, että funktio suoritetaan koko ohjelman ajan vain kerran. ```go package main import "sync" func main() { once := sync.OnceFunc(func() { println("Hello, World!") }) once() once() once() once() once() } ``` Yllä oleva koodi tulostaa `Hello, World!` käyttäen `sync.OnceFunc`ia. Tässä koodissa luodaan `once`-funktio `sync.OnceFunc`illa, ja vaikka `once`-funktiota kutsutaan useita kertoja, `Hello, World!` tulostuu vain kerran. #### OnceValue `OnceValue` ei yksinkertaisesti suorita funktiota vain kerran, vaan tallentaa funktion palautusarvon ja palauttaa tallennetun arvon, kun sitä kutsutaan uudelleen. ```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()) } ``` Yllä oleva koodi kasvattaa muuttujaa `c` yhdellä käyttäen `sync.OnceValue`a. Tässä koodissa luodaan `once`-funktio `sync.OnceValue`lla, ja vaikka `once`-funktiota kutsutaan useita kertoja, muuttuja `c` kasvaa vain kerran ja palautetaan arvo 1. #### OnceValues `OnceValues` toimii samoin kuin `OnceValue`, mutta voi palauttaa useita arvoja. ```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) } ``` Yllä oleva koodi kasvattaa muuttujaa `c` yhdellä käyttäen `sync.OnceValues`a. Tässä koodissa luodaan `once`-funktio `sync.OnceValues`lla, ja vaikka `once`-funktiota kutsutaan useita kertoja, muuttuja `c` kasvaa vain kerran ja palautetaan arvo 1. ### atomic `atomic`-paketti tarjoaa atomisia operaatioita. Se tarjoaa metodit `Add`, `CompareAndSwap`, `Load`, `Store`, `Swap` ja muita, mutta viime aikoina suositellaan `Int64`-, `Uint64`- ja `Pointer`-tyyppien käyttöä. ```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()) } ``` Tämä on esimerkki, joka käytettiin aiemmin. Koodi kasvattaa muuttujaa `c` atomisesti käyttäen `atomic.Int64`-tyyppiä. `Add`- ja `Load`-metodeilla voidaan kasvattaa muuttujaa atomisesti ja lukea sen arvo. Lisäksi `Store`-metodilla voidaan tallentaa arvo, `Swap`-metodilla vaihtaa arvo ja `CompareAndSwap`-metodilla verrata arvoa ja vaihtaa se, jos se sopii. ### cond #### sync.Cond `cond`-paketti tarjoaa ehtomuuttujia. `cond`-paketti voidaan luoda `sync.Cond`illa, ja se tarjoaa metodit `Wait`, `Signal` ja `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!") } ``` Yllä oleva koodi odottaa, että `ready`-muuttuja muuttuu arvoon `true`, käyttäen `sync.Cond`ia. Tässä koodissa `sync.Cond`ia käytetään odottamaan, kunnes `ready`-muuttuja on `true`, ja sen jälkeen tulostetaan `Ready!`. Näin `sync.Cond`ia käyttämällä useat goroutinet voivat odottaa, kunnes tietty ehto täyttyy. Tätä voidaan hyödyntää yksinkertaisen `queue`n toteuttamiseen. ```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 } ``` Kun `sync.Cond`ia käytetään tällä tavalla, sen sijaan että käytettäisiin paljon CPU:ta `spin-lock`illa, odotetaan tehokkaasti, ja kun ehto täyttyy, koodi voi jatkaa toimintaansa. ### semaphore #### golang.org/x/sync/semaphore.Semaphore `semaphore`-paketti tarjoaa semafooreja. `semaphore`-paketti voidaan luoda `golang.org/x/sync/semaphore.Semaphore`lla, ja se tarjoaa metodit `Acquire`, `Release` ja `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) } ``` Yllä oleva koodi luo semaforin käyttäen `semaphore`a, hankkii sen `Acquire`-metodilla ja vapauttaa `Release`-metodilla. Tässä koodissa näytettiin, miten semaforia hankitaan ja vapautetaan `semaphore`-paketilla. ## Lopuksi Perusasiat riittävät tähän asti, luulen. Toivon, että artikkelin sisällön perusteella ymmärrät, miten goroutinejen samanaikaisuutta hallitaan, ja pystyt käyttämään sitä käytännössä. Toivon, että tämä artikkeli on ollut sinulle hyödyllinen. Kiitos.