Go konkurencia kezelés gyorstalpaló

Mikor elkezdtem ezt a blogot írni, érdeklődésem előterében a Java alapú technológiák voltak. Rengeteg téma merült fel az eltelt 7 évben, amiről szívesen írtam volna, de mivel nem kapcsolódtak szorosan az eredeti tematikához, ezek a bejegyzések sosem kerültek implementálásra. Szakítva ezzel a (rossz) hagyománnyal szeretnék az informatika világának szélesebb spektrumával foglalkozni ezen túl. Első Java mentes bejegyzésem témájának pedig a Go nyelv konkurencia kezelését választottam.

A Go nyelv körül elég nagy a sürgés-forgás mostanában, nem tudom pontosan miért örvend ekkora népszerűségnek. Számtalan jó tulajdonsága van, de szerintem mind közül a legfontosabb szempont, hogy a nyelv egyszerű. Nincsenek mögötte "bonyolult" ideológiák mint például az objektum orientált programozásban, vagy nem lehet benne egy dolgot száz féleképpen csinálni mint mondjuk a funkcionális nyelvekben. Kutyaközönséges, de modern procedurális nyelv annak minden előnyével és hátrányával. Akit részletesebben érdekel ez a terület érdemes Rob Pike előadását meghallgatni. Témánkra rátérve ugyanez az egyszerűség jellemzi a Go konkurencia kezelését is. Vannak az un. goroutinok, ezek segítségével lehet párhozamosítani a végrehajtást, és vannak a chanelek, amiken keresztűl zajlik a kommunikáció. Utóbbiból van szinkron meg aszinkron.

	package main
	func main() {
	c := make(chan bool)
	go func() { c <- true }()
	println(<-c)
	}

view raw basic-concurrency.go hosted with ❤ by GitHub

A program létrehoz egy c szinkron chanelt, majd elindít egy goroutint, és kiolvassa a chanleből az értéket, amit a goroutin beletesz. Mivel szinkron módban működik egyik végrehajtás sem megy tovább amíg a kommunikáció meg nem történik. Aszinkron channelt a méretének megadásával tudunk létrehozni. Kész is, mindent tudunk :) Akit részletesebben érdekel, mi történik a motor-háztető alatt, annak ajánlom a Go memória kezeléséről szóló cikket. Természetesen ennyivel nem ússzuk meg a dolgot, komolyabb alkalmazásnál szükségünk van meg egy pár dologra, hogy biztosítani tudjuk programunk helyes működését. Két csomagot fogunk felhasználni, az egyik a sync, ami szinkronizációs problémákra nyújt megoldást, illetve a sync/atomic, melynek segítségével elemi műveleteket végezhetünk. A Go nyelv dokumentációját olvasva számtalan helyen olvashatjuk, hogy nincs garancia, hogy adott dolog hogyan viselkedik mikor több végrehajtó szál birizgálja egy időben (nem 1 az 1-hez a goroutine és szál leképzés, de ettől most tekintsünk el). Alapvetően érték szerint adódik át minden, így a legtöbb esetben nincs is szükség különösebb szinkronizációra, de ha mégis, magunknak kell gondoskodni róla.

Kitaláltam egy egyszerű kis feladatot, amin keresztűl bemutatok pár fontos eszközt működés közben.

• Legyen egy szál típus, ami 1000 véletlen számot generál, és a legnagyobbat elküldi feldolgozásra.
• Egy másik az előzőtől kapott értékből kiindulva generál még 500 véletlen számot, a legnagyobbat gyűjti egy tömbbe és tovább küldi.
• A harmadik feldolgozó nem csinál mást, mint számolja, hogy a kapott érték hány esetben volt az utolsó a tömbben.
• Szabályozható legyen a goroutinok száma, a generálóból hány példány futhat egyidőben, valamint hány értéket generáljon összesen.
• Mérjük a végrehajtás idejét a generálás kezdetétől.

A feladatnak semmi értelme, és biztos egyszerűbben is meg lehetett volna oldani, de igyekeztem minél több módszert belezsúfítani a megoldásba.

Első lépésben inicializáljuk a konstansokat, valamint az eredmény tömböt. Mivel az eredménnyel több szál is fog egy időben dolgozni szükségünk lesz valami zárolási logikára. Tekintettel arra, hogy az egyik szál írja a másik pedig olvassa én a sync.RWMutex-et választottam.

	package main
	import (
	"fmt"
	"math"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
	)
	const (
	MAX = 500 // maximum darabszám
	PRODUCERS = 12 // generáló szálak
	CONSUMERS = 9 // feldolgozó szálak
	COUNTERS = 2 // statisztikázó szálak
	)
	func main() {
	results := make([]int, 0) // eredmény gyűjtő
	mtx := new(sync.RWMutex) // read-write lock

view raw concurrency-main.go hosted with ❤ by GitHub

A következő kódblockban implementálásra kerül a statisztikázó egység.

	var itsLast, notLast int32 // statisztika eredménye
	countChan := make(chan int, COUNTERS) // chanel, amin a számláló várja az értékeket
	counter := func() {
	for {
	e := <-countChan // kiolvassuk a következő értéket
	mtx.RLock() // olvasásra zárolunk és kiolvassuk az eredmenyből az utolsó elemet
	last := results[len(results)-1]
	mtx.RUnlock()
	if e != last { // eredmenytől függően elemi lépésben növeljük a megfelelő számlálót
	atomic.AddInt32(&notLast, 1)
	} else {
	atomic.AddInt32(&itsLast, 1)
	}
	}
	}
	for i := 0; i < COUNTERS; i++ { // megadott számban elindítjuk a számlálókat
	go counter()
	}

view raw concurrency-counter.go hosted with ❤ by GitHub

Majd megírjuk a feldolgozó egységet.

	appendChan := make(chan int, CONSUMERS) // chanel, amin a generálás eredményei jönek
	cons := func() {
	for {
	e := <-appendChan // kiolvassuk a következő generált értéket
	for i := 0; i < 500; i++ { // még 500szor probálkozunk nagyobbat generálni
	e = int(math.Min(float64(e), float64(rand.Int())))
	}
	mtx.Lock() // irásra zárolunk majd hozzáfűzzük az eredméyekhez
	results = append(results, e)
	mtx.Unlock()
	countChan <- e // tovabb küldjük az eredményt a számlálónak
	}
	}
	for i := 0; i < CONSUMERS; i++ { // megfelelő számban elindítjuk a feldolgozót
	go cons()
	}

view raw concurrency-append.go hosted with ❤ by GitHub

Na most kezd igazán érdekes lenni a dolog. A generátor esetében tudni kell szabályozni, hogy hány darab fut egyidőben. A legkézenfekvőbb döntés, ha létrehozunk egy chanelt adott méretben, és beleteszünk egy értéket amikor indítunk egy új generátort, és kiveszünk egyet ha végzett a generátor. Ennek eredméyeképpen amikor megtelik a chanel programunk várakozik.

	poolChan := make(chan bool, PRODUCERS) // egyidőben futó generátorok nyilvántartása
	prod := func() {
	// ...
	<-poolChan // kiolvasunk egy értéket
	}
	for i := 0; i < MAX; i++ { // megfelelő számban elindítjuk a generálót
	poolChan <- true // beteszünk egy értéket
	go prod() // elinditjuk a goroutint
	}

view raw concurrency-generator.go hosted with ❤ by GitHub

Miután elindítottuk a generáló szálakat szeretnénk megvárni amíg az összes végez. Ennek érdekében bevezetünk egy sync.WaitGroup-ot.

	prod := func(wg *sync.WaitGroup) {
	defer wg.Done() // a futás végeztével nyugtázunk
	// ...
	}
	wg := new(sync.WaitGroup) // új WaitGroup
	wg.Add(MAX) // beállítjuk hány nyugtázást várunk
	for i := 0; i < MAX; i++ {
	// ...
	go prod(wg)
	}
	wg.Wait() // bevárjuk az összes nyugtázást

view raw concurrency-waitgroup.go hosted with ❤ by GitHub

Egy utolsó dolog maradt a generátorral kapcsolatban, hogy szeretnénk az eltelt időt mérni a lehető legközelebb a generáláshoz. Ennek biztosítására én a sync.Once-ot választottam, ami garantálja ezt a működést.

	var start int64 // kezdés időbélyege
	once := new(sync.Once) // ez garantálja, hogy csak egyszer adjunk értéket az időbélyegnek
	poolChan := make(chan bool, PRODUCERS)
	prod := func(wg *sync.WaitGroup) {
	defer wg.Done()
	once.Do(func() { start = time.Now().UnixNano() }) // ertéket adunk
	e := rand.Int() // generálunk 1000 véletlenszámot
	for i := 0; i < 999; i++ {
	e = int(math.Min(float64(e), float64(rand.Int())))
	}
	appendChan <- e // tovább küldjük az eredményt
	<-poolChan
	}
	wg := new(sync.WaitGroup)
	wg.Add(MAX)
	for i := 0; i < MAX; i++ {
	poolChan <- true
	go prod(wg)
	}
	wg.Wait()

view raw concurrency-generator.go hosted with ❤ by GitHub

Nincs más dolgunk mint megvárni és kiírni az eredményt.

	var done bool
	for !done { // várunk amíg az összeset megszámolja a számláló
	done = atomic.LoadInt32(&itsLast)+atomic.LoadInt32(&notLast) == MAX
	}
	fmt.Printf("time spent: %dms\n", (time.Now().UnixNano()-start)/1000000)
	fmt.Printf("last:not ratio: %d:%d\n", itsLast, notLast)

view raw concurrency-final.go hosted with ❤ by GitHub

A teljes forráskód elérhető itt.

Programunk gyönyörűen fut, de vajon mi történik a háttérben? Fordítsuk le a programot a Go beépített verseny helyzet elemzőjével.

go build -race

Ez után futtatva valami hasonló kimenetet kapunk:

==================
WARNING: DATA RACE
Read at 0x00c42000c2e8 by main goroutine:
  main.main()
      /Users/rkovacs/asd/src/asd/main.go:92 +0x731

Previous write at 0x00c42000c2e8 by goroutine 7:
  sync/atomic.AddInt32()
      /usr/local/Cellar/go/1.7.1/libexec/src/runtime/race_amd64.s:269 +0xb
  main.main.func1()
      /Users/rkovacs/asd/src/asd/main.go:37 +0x191

Goroutine 7 (running) created at:
  main.main()
      /Users/rkovacs/asd/src/asd/main.go:42 +0x31b
==================
time spent: 1121ms
last:not ratio: 499:1
Found 1 data race(s)

Futás idejű analízisből kiderült, hogy volt értelme elemi műveletben növelni a számlálót, hiszen verseny helyzet alakult ki, és máskülönben megjósolhatatlan eredménnyel zárult volna a futás.

A téma még nem merült ki ennyivel, de remélem sikerült egy gyors képet adnom a Go konkurencia kezeléséről. Ajánlom továbbá figyelmetekbe a Concurrency is not parallelism és a Go concurrency patterns előadásokat.