Yayma ve Toplama

Önceki ders, sabit sayıda işçinin aynı iş kanalını dinlediği bir kurulumu ölçtü: tek kaynak, çok tüketici, ve durma sinyali hepsine tek bir kapatmayla ulaşıyordu. Ama iş her zaman işçilerin kendi başına yetiştirebileceği hızda gelmiyor. Bazen üretilen değerlerin kendisi birden fazla işlemciye dağıtılmalı — bu yayma — ve o dağıtılmış işin sonuçları tekrar tek bir akışta toplanmalı — bu toplama. Önceki dersin sorduğu “kanalı kim kapatır” sorusu burada da geçerli, ama yanıtı farklı bir yerde duruyor: işçi havuzunda kapatan taraf işi dağıtan tarafken, burada kapatılması gereken kanala birden fazla taraf yazıyor — ve bu, “kim kapatır” sorusunu ilk kez gerçek bir soruna çeviriyor.

Yayma ve toplama iki ayrı yönü kapsıyor. Yayma, bir üretim kaynağının (bir kuyruk, bir kanal) birden fazla işlemciye açılması — önceki dersteki iş kanalıyla neredeyse aynı düzenek, çünkü paylaşılan bir kanaldan okuyan sabit sayıda goroutine zaten bir yaymadır. Toplama ise tersi yönde çalışıyor: birden fazla kaynağın tek bir kanalda birleşmesi. İki yön birlikte kurulduğunda ortaya yeni bir soru çıkıyor — yaymada kapatan taraf bellidir (kaynağı yöneten tek taraf), ama toplamada kapatılması gereken kanala kaç tarafın yazdığı, kapatma sorumluluğunu da o kadar taraf arasında paylaştırma riski taşıyor.

Bir Kaynağın Çoğa Yayılması, Sonuçların Tek Kanalda Toplanması

Aşağıdaki ölçüm altı değeri üç üretici goroutine’ine yayıyor; her üretici kendi payına düşen değerleri kareleyip tek bir toplama kanalına yazıyor. Toplama kanalını kimin kapatacağı burada zaten çözülmüş: ayrı bir goroutine, bir bekleme grubunun (sync.WaitGroup) bütün üreticilerin bittiğini bildirmesini bekliyor ve ancak o zaman toplamayı kapatıyor. Tek sahiplik burada bir üreticide değil, üreticilerin dışında kuruluyor.

// yaymatoplama.go — kaynak uc uretene yayiliyor, sonuclar tek kanalda toplaniyor; toplama kanalini bekleme grubu kapatiyor
package main

import (
	"fmt"
	"sort"
	"sync"
)

func uretici(girdi <-chan int, toplama chan<- int, grup *sync.WaitGroup) {
	defer grup.Done()
	for deger := range girdi {
		toplama <- deger * deger
	}
}

func main() {
	const isSayisi = 6
	const ureticiSayisi = 3

	girdi := make(chan int, isSayisi)
	toplama := make(chan int, isSayisi)
	var grup sync.WaitGroup

	for id := 1; id <= ureticiSayisi; id++ {
		grup.Add(1)
		go uretici(girdi, toplama, &grup)
	}

	for i := 1; i <= isSayisi; i++ {
		girdi <- i
	}
	close(girdi)

	// Toplayici: butun ureticiler bitince toplama kanalini KENDISI kapatiyor;
	// hicbir uretici kendi basina kapatmiyor.
	go func() {
		grup.Wait()
		close(toplama)
	}()

	fmt.Println("-- kaynak 3 uretene yayiliyor, sonuclar tek kanalda toplaniyor --")
	sonuclar := make([]int, 0, isSayisi)
	for s := range toplama {
		sonuclar = append(sonuclar, s)
	}
	sort.Ints(sonuclar)

	toplamDeger := 0
	for _, s := range sonuclar {
		toplamDeger += s
	}
	fmt.Printf("yayilan is sayisi: %d\n", isSayisi)
	fmt.Printf("toplanan sonuc sayisi: %d, toplam: %d\n", len(sonuclar), toplamDeger)
	fmt.Println("toplayicinin durma sebebi: bekledigi geldi (toplama kanali kapandi)")
}

-- kaynak 3 uretene yayiliyor, sonuclar tek kanalda toplaniyor --
yayilan is sayisi: 6
toplanan sonuc sayisi: 6, toplam: 91
toplayicinin durma sebebi: bekledigi geldi (toplama kanali kapandi)

KL6. girdi kanalı önceki dersteki isler ile aynı rolü oynuyor — üç üretici arasında paylaşılan bir kuyruk. Farklı olan toplama kanalı: üç üretici de aynı anda ona yazıyor, tek bir işçi değil. Sonuç sayısının altıya, toplamın doksan bire (kareler 1+4+9+16+25+36) çıkması hangi üreticinin hangi değeri işlediğinden bağımsız; sıralama (sort.Ints) da bu yüzden gerekiyor — üç üreticinin toplama kanalına yazma sırası koşumdan koşuma değişebilir, ama kümenin kendisi değişmiyor.

Bu ölçümde uretici işlevi, önceki dersteki isci işleviyle neredeyse aynı gövdeye sahip: ikisi de bir girdi kanalından okuyor, bir çıktı kanalına yazıyor. Aradaki fark işlevin kendi kodunda değil, çağrıldığı bağlamda — isci sabit sayıda tüketiciyi aynı işi paylaşacak biçimde kurarken, uretici aynı işlevi bu kez birden fazla yazıcının aynı hedefe yazması bağlamında çalıştırıyor. Kalıbın adı değişiyor (işçi havuzundan yaymaya), ama alttaki kanal düzeneği aynı kalıyor; değişen yalnız kapatma sorumluluğunun nerede durduğu.

KL7. Kapatan goroutine burada üçüncü bir taraf: ne üreticilerden biri, ne toplayıcının kendisi. grup.Wait() üç üreticinin üçü de grup.Done()e ulaşana kadar dönmüyor, ve ancak o zaman close(toplama) çalışıyor — bu, toplama kanalına yapılacak son yazmanın zaten gerçekleşmiş olduğunu garanti ediyor. girdi ve toplama kanallarının ikisi de isSayisi kapasitesiyle tamponlu açılıyor, aynı önceki dersteki gerekçeyle: üreticiler kendi paylarını alıp yazarken engellenmeden çalışabilsin diye. Bu üçüncü taraf yaklaşımı bir sonraki bölümün sorusunu doğuruyor: ya bu ayrı goroutine hiç olmasaydı, ve üreticilerin her biri kendi işini bitirince kanalı kendisi kapatmaya çalışsaydı?

Toplama Kanalını Kim Kapatır

Bir kanalı kapatmak Go’da tek seferlik bir işlem: aynı kanalı iki kez kapatmak panikler. Üç üretici de aynı kanalı paylaşıyorsa ve her biri kendi işini bitirince “ben bittim, kanalı kapatayım” derse, ilk kapatan başarılı olur ama ondan sonraki her kapatma denemesi panikler. Aşağıdaki ölçüm bunu kuruyor: her üretici önce kendi değerini gönderiyor, sonra (bir bekleme noktasıyla bütün üreticilerin göndermeyi bitirdiğinden emin olunduktan sonra) kendi başına close(toplama) deniyor.

// naifkapatma.go — her uretici kendi basina kapatmayi denerse, ilkinden sonrakiler panikler
package main

import (
	"fmt"
	"sync"
)

func ureticiNaif(id int, toplama chan int, senkron *sync.WaitGroup, sonuc chan<- bool, bitis *sync.WaitGroup) {
	defer bitis.Done()
	toplama <- id
	senkron.Done()
	senkron.Wait() // butun ureticiler kendi degerini gonderene kadar bekle
	panikledi := false
	func() {
		defer func() {
			if r := recover(); r != nil {
				panikledi = true
			}
		}()
		close(toplama)
	}()
	sonuc <- panikledi
}

func main() {
	const ureticiSayisi = 3
	toplama := make(chan int, ureticiSayisi)
	sonuc := make(chan bool, ureticiSayisi)
	var senkron sync.WaitGroup
	var bitis sync.WaitGroup
	senkron.Add(ureticiSayisi)
	bitis.Add(ureticiSayisi)

	for id := 1; id <= ureticiSayisi; id++ {
		go ureticiNaif(id, toplama, &senkron, sonuc, &bitis)
	}
	bitis.Wait()
	close(sonuc)

	basarili, panikleyen := 0, 0
	for panikledi := range sonuc {
		if panikledi {
			panikleyen++
		} else {
			basarili++
		}
	}
	fmt.Println("-- her uretici once kendi degerini gonderiyor, sonra kapatmayi deniyor --")
	fmt.Printf("basariyla kapatan uretici sayisi: %d / %d\n", basarili, ureticiSayisi)
	fmt.Printf("panikleyen uretici sayisi: %d / %d\n", panikleyen, ureticiSayisi)
}

-- her uretici once kendi degerini gonderiyor, sonra kapatmayi deniyor --
basariyla kapatan uretici sayisi: 1 / 3
panikleyen uretici sayisi: 2 / 3

KL8. Sonuç şaşırtıcı olmayan bir biçimde tek yanıtlı: her koşumda tam olarak bir üretici başarıyla kapatıyor, kalan ikisi panikliyor. Hangi üreticinin başardığı koşumdan koşuma değişebilir (üç üretici de kapatmayı aynı anda denemeye başlıyor), ama kaç üreticinin başardığı hiç değişmiyor — Go’nun çalışma zamanı bir kanalın kapatılmasını iç kilitle koruyor, bu yüzden birden fazla goroutine aynı anda close çağırsa bile yalnız biri kanalı gerçekten kapatıyor, geri kalanı zaten kapanmış bir kanalı kapatmaya çalıştığını görüp panikliyor. Ders bu yüzden kimin başardığını değil, kaçının başardığını yazdırıyor.

Bu iç kilit, closein kendisinin atomik bir işlem olduğu anlamına geliyor: bir kanalın “kapalı mı değil mi” durumu, birden fazla goroutine aynı anda sorup değiştirmeye çalışsa bile, her zaman tek bir sıralı geçmişe sahip. Üç goroutine close(toplama)yı gerçekten aynı anda çağırsa da çalışma zamanı bunları görünmez bir sırayla işliyor — ilk işlenen kapatma gerçekleşiyor, ondan sonraki her biri zaten kapalı bir kanalı bulup panikliyor. Bu davranış paylaşılan bir değişkene korumasız yazmaktan çok farklı: paylaşılan bir sayaca aynı anda iki goroutine yazarsa sonuç tanımsız bir yarış koşuludur (ölçülmüyor, yalnız kavram olarak M01/K05 İşletim Sistemi Kavramları’nda kuruldu), ama kanal kapatma dilin kendisinin garanti ettiği, yarışa açık olmayan bir işlem — kazanan kim olursa olsun tam olarak bir kazanan çıkıyor.

KL9. İkinci senkronizasyon noktası (senkron) burada bilinçli bir tasarım kararı: ölçüm her üreticinin önce kendi değerini gönderdiğinden, sonra kapatmayı denediğinden emin oluyor. Bu ayrım olmasaydı, bir üretici henüz kendi değerini göndermeden bir başkası kanalı kapatabilir, ve o üreticinin gönderimi (toplama <- id) kapatma satırından önce değil panik olarak sonuçlanırdı — bu da hangi hattın hangi sebeple paniklediğini karıştırırdı. senkron.Wait(), önceki derste görülen grup.Wait() gibi, sırayı çizelgeleyiciye bırakmadan iki aşamayı (gönderme, kapatma) birbirinden ayırıyor.

Bu ayrımın kendisi bir gözlemi gizliyor: bir kapalı kanalla ilgili panik aslında iki ayrı biçimde tetiklenebiliyor — kapalı bir kanalı tekrar kapatmak (close of closed channel) ve kapalı bir kanala göndermek (send on closed channel). Yukarıdaki senkron engeli olmasaydı, bir üretici henüz kendi değerini göndermeden (toplama <- id) bir başkası kanalı kapatmış olabilirdi, ve o gönderim satırı kendi recover bloğunun dışında kaldığı için panik yakalanmadan program çöker, ders kod_dogrula.pyye hiç ulaşmadan durur. senkron.Wait() engeli tam olarak bunu önlüyor: bütün gönderimlerin bittiğinden emin olunmadan hiçbir üretici kapatmayı denemiyor, bu yüzden gözlenen panik türü hep aynı (kapalı kanalı tekrar kapatma), hiçbir zaman diğeri değil.

Bu ölçüm önceki bölümdeki doğru düzenin neden doğru olduğunu gösteriyor: toplama kanalını kapatma sorumluluğu üreticilerin hiçbirine ait değil, çünkü hiçbiri diğerlerinin ne zaman biteceğini bilmiyor. Sorumluluk, ancak bütün üreticilerin bittiğini bilen üçüncü bir tarafa (bir bekleme grubunu bekleyen ayrı bir goroutine’e) verilebiliyor.

Kapatılmayan Toplama Kanalı

Önceki iki kurulum, kapatma sorumluluğunun sıfır ile birden fazla taraf arasında paylaştırılmasının iki farklı sonucunu gösterdi: birinci kurulumda sorumluluk tam olarak bir tarafa aitti ve her şey sessizce bitti; ikinci kurulumda sorumluluk (yanlışlıkla) üç tarafa aitti ve iki panik ortaya çıktı. Geriye üçüncü olasılık kalıyor: sorumluluk hiç kimseye verilmezse ne olur? Üçüncü taraf goroutine (yukarıdaki bekleme grubunu bekleyip toplama kanalını kapatan ayrı goroutine) hiç kurulmazsa, toplama kanalı hiç kapanmıyor. Aşağıdaki ölçüm bunu en yalın haliyle kuruyor: bir toplayıcı goroutine toplama kanalında aralık yapıyor, ama kanalı kapatacak hiç kimse yok.

// sizintitoplama.go — toplama kanali hic kapatilmazsa toplayici hic durmaz
package main

import (
	"fmt"
	"runtime"
)

func main() {
	taban := runtime.NumGoroutine()

	toplama := make(chan int)
	go func() {
		// Kimse toplama kanalini kapatmadigi icin bu aralik hic bitmeyecek.
		for range toplama {
		}
	}()

	fmt.Println("-- toplama kanali hic kapatilmadi --")
	fmt.Printf("durmayan toplayici kaldi mi: %v\n", runtime.NumGoroutine() > taban)
}

-- toplama kanali hic kapatilmadi --
durmayan toplayici kaldi mi: true

KL10. Bu, önceki dersteki kapatılmayan iş kanalı ölçümünün toplama tarafındaki aynası: orada işçiler sonsuza kadar bekliyordu, burada tek bir toplayıcı sonsuza kadar bekliyor. Fark, sızıntının yerinde: işçi havuzunda sızan taraf çok sayıda tüketiciydi, burada sızan taraf tek bir toplayıcı — ama ikisinde de sebep aynı, kimsenin kapatmadığı bir kanalda aralık. Toplayıcının kendisi de bir goroutine olduğu için, onun sızması da işçi sızıntısından farklı bir araçla değil, aynı runtime.NumGoroutine() karşılaştırmasıyla ölçülüyor.

Bu üç kurulumun ortak noktası, hiçbirinin sorunu kaynağında çözmemesi: doğru kurulum bile bütün üreticilerin sonunda biteceğini varsayıyor. Üreticilerden biri kendi işini hiç bitirmezse (örneğin sonsuz bir döngüye girerse), grup.Wait() de hiç dönmez ve toplama kanalı yine hiç kapanmaz — bu kurulum kapatma sorumluluğunu doğru bir tarafa veriyor, ama o tarafın kendisi de bir koşula bağlı kalıyor. Bir üreticinin kendi işini bitirmemesi olasılığını sınırlamak (belirli bir süre ya da dışarıdan gelen bir sinyalle) bu dersin konusu değil; o soruyu bağlam paketiyle iptal yayma ölçen bir sonraki ders ele alıyor.

Üç ölçüm birlikte konunun tezini tamamlıyor: doğru kurulan toplama (bir kişide toplanan sorumluluk) belirlenimci ve sessiz durdu; yanlış kurulan toplama (herkesin kendi başına kapatmaya çalışması) belirlenimci ama gürültülü durdu — paniklerle; ve hiç kurulmayan toplama (kimsenin kapatmaması) hiç durmadı. Üçü de aynı sorunun farklı yanıtları: toplama kanalını kapatma sorumluluğu tam olarak bir tarafa ait olmalı, ne az ne çok. Bu ilke işçi havuzundaki tek kapatan tarafla (işi dağıtan taraf) aynı ilkenin bir genellemesi: kaç taraf bir kanala yazarsa yazsın, kapatma her zaman ayrı ve tekil bir sorumluluk olarak kalıyor.

Özet

• Bir kaynak sabit sayıda üreticiye yayıldığında (yayma), sonuçlar tek bir toplama kanalında yeniden birleşiyor (toplama); toplanan sonuç sayısı ve toplamı üretici sayısından bağımsız iki sabit değer.
• Toplama kanalını kapatma sorumluluğu üreticilerin hiçbirine ait değil — bir bekleme grubunun bütün üreticilerin bittiğini bildirdiği üçüncü bir taraf kapatıyor.
• Her üretici kendi başına kapatmayı denerse, tam olarak bir tanesi başarır, kalanı panikler; hangi üreticinin başardığı değişse de kaçının başardığı hiç değişmiyor.
• Toplama kanalı hiç kapatılmazsa toplayıcı sonsuza kadar bekler; sızıntı yine yalnız bir boolean olarak ölçülüyor.
• Kapatma sorumluluğunun tam olarak bir tarafa ait olması gerekiyor: sıfır taraf sızıntıya, birden fazla taraf paniğe yol açıyor.
• Kanal kapatma dilin kendisinin garanti ettiği atomik bir işlem; birden fazla goroutine aynı anda close çağırsa bile tam olarak biri başarır, geri kalanı panikler.

Sonraki Adım

Bu derste bir kaynak çoğa yayıldı ve sonuçlar tek bir kanalda toplandı — ama akış tek bir adımdı: her üretici girdiyi aldı, kareledi, sonucu yazdı. Gerçek bir işleme genellikle birden fazla adımdan geçiyor ve her adımın kendi hızı, kendi kapatma sorumluluğu var. Sıradaki ders bu adımları zincirliyor — aşamalı akış bir boru hattı kuruyor, ve her aşama kendi çıktı kanalını kendisi kapatıyor. Soru bu sefer tek bir kapatmanın kimde durduğu değil: zincirdeki bir aşama erken durursa, önündeki aşamalara ne oluyor?