Boru Hatları

Önceki ders kapatma sorumluluğunun kaç tarafa ait olması gerektiğini ölçtü: sıfır taraf sızıntıya, birden fazla taraf paniğe yol açıyordu, doğrusu tam olarak bir taraftı. Ama o üretim tek bir adımdan ibaretti — girdi alınır, karelenir, sonuç yazılır. Gerçek bir işleme genellikle birden fazla adımdan geçiyor ve her adımın kendi hızı, kendi çıktısı var. Bu ders bu adımları zincirliyor: aşamalı akış, yani boru hattı (pipeline) — her aşama önceki aşamanın çıktısını kendi girdisi olarak alan, kendi çıktısını üreten ve kendi çıktı kanalını kendisi kapatan bir zincir. Kapatma sorumluluğu artık tek bir yerde değil, zincirin her halkasında ayrı ayrı duruyor.

Önceki iki dersteki düzenlerle karşılaştırıldığında aşamalı akışın şekli farklı. İşçi havuzunda çok sayıda goroutine aynı kanaldan okuyordu (paylaşılan bir kuyruk); yaymada çok sayıda goroutine aynı kanala yazıyordu (paylaşılan bir hedef). Aşamalı akışta ise her kanal tam olarak bir yazıcı ve bir okuyucu arasında duruyor — paylaşım yok, yalnız zincirleme. Bu, önceki iki dersteki en zor sorunun (kapatmayı kaç tarafın üstlenmesi gerektiği) burada hiç ortaya çıkmadığı anlamına geliyor: her kanalın tam olarak bir sahibi var, o da kanalı açan aşamanın kendisi. Zincirin kendi iç yapısı belirlenimciliği bedavaya getiriyor; asıl soru bu dersin ikinci bölümünde ortaya çıkıyor — zincirin dışından, tüketiciden gelen bir vazgeçmeyle.

Aşamalı Akış: Her Aşama Kendi Kapatma Sorumluluğunu Taşır

Aşağıdaki ölçüm üç aşamalı bir hat kuruyor: uret bir sayı aralığı üretiyor, kareAl her değeri kareliyor, birEkle her değere bir ekliyor. Üçü de aynı biçimde yazılmış: bir kanal açıyor, kendi işini yapan bir goroutine başlatıyor, o goroutine işini bitirince kendi kanalını kendisi kapatıyor, ve kanalın kendisini çağıranına döndürüyor.

// asamalar.go — uc asamali boru hatti: her asama kendi ciktisini kendisi acar ve kendisi kapatir
package main

import "fmt"

// uret, 1..n araligindaki degerleri kendi ciktisina yazar ve isini bitirince kapatir.
func uret(n int) <-chan int {
	cikti := make(chan int)
	go func() {
		defer close(cikti)
		for i := 1; i <= n; i++ {
			cikti <- i
		}
	}()
	return cikti
}

// kareAl her degeri karesiyle degistirir; girdi kapaninca kendi ciktisini kapatir.
func kareAl(girdi <-chan int) <-chan int {
	cikti := make(chan int)
	go func() {
		defer close(cikti)
		for deger := range girdi {
			cikti <- deger * deger
		}
	}()
	return cikti
}

// birEkle her degere bir ekler; girdi kapaninca kendi ciktisini kapatir.
func birEkle(girdi <-chan int) <-chan int {
	cikti := make(chan int)
	go func() {
		defer close(cikti)
		for deger := range girdi {
			cikti <- deger + 1
		}
	}()
	return cikti
}

func main() {
	asama1 := uret(5)
	asama2 := kareAl(asama1)
	asama3 := birEkle(asama2)

	fmt.Println("-- uc asamali hat, tam tuketim --")
	sonuclar := make([]int, 0, 5)
	for s := range asama3 {
		sonuclar = append(sonuclar, s)
	}
	fmt.Printf("akan deger sayisi: %d\n", len(sonuclar))
	fmt.Println("degerler:", sonuclar)
	fmt.Println("her asamanin durma sebebi: bekledigi geldi (bir onceki asamanin kanali kapandi)")
}

-- uc asamali hat, tam tuketim --
akan deger sayisi: 5
degerler: [2 5 10 17 26]
her asamanin durma sebebi: bekledigi geldi (bir onceki asamanin kanali kapandi)

KL11. Beş değer ( 1..5 ) sırasıyla kareleniyor (1,4,9,16,25) ve her birine bir ekleniyor (2,5,10,17,26) — sonuç listesi bu sırayla basılıyor ve bu sıra hiçbir yerde sıralanmadan ortaya çıkıyor. Sebep önceki iki dersteki yayma/toplama düzenlerinden farklı: orada birden fazla üretici aynı kanala aynı anda yazıyordu, bu yüzden sıra kaybolabiliyordu ve toplanan sonuçların sıralanması gerekiyordu. Burada her kanalın tek bir yazıcısı ve tek bir okuyucusu var (uretin tek goroutine’i, kareAlin tek goroutine’i, ana goroutine’in kendisi); tek yazıcı/tek okuyucu ilişkisinde bir kanal gönderme sırasını her zaman koruyor, çünkü yarışacak ikinci bir taraf yok.

Ana goroutine’in kendisi de zincirin bir parçası: asama3ün tek okuyucusu o, ve tüketim döngüsü birEklein goroutine’i kendi çıktısını kapatana kadar sürüyor. Zincirin başından sonuna kadar toplam dört goroutine çalışıyor — uret, kareAl, birEklein üçü ve ana goroutine’in kendisi — ama hiçbiri diğerini beklemek için özel bir senkronizasyon aracına (bir bekleme grubuna, bir selecte) ihtiyaç duymuyor; kanalların kendisi hem veri taşıyor hem sırayı dayatıyor.

KL12. Her aşamanın durma sebebi de aynı: kareAl ve birEkle, kendi girdi kanalları kapandığında (for deger := range girdi biterken) kendi çıktı kanallarını kapatıyor — defer close(cikti) bunu garanti ediyor, girdi tüketilirken panik olsa bile. uretin kendi goroutine’i ise beş değeri gönderdikten sonra döngüsü bitince aynı deferle kapanıyor. Zincirin her halkası, önceki derste kurulan “kapatma sorumluluğu tam olarak bir tarafa aittir” ilkesini kendi kanalı için ayrı ayrı uyguluyor — zincirdeki üç kanalın üçü de kendi üreticisi tarafından kapatılıyor, hiçbiri paylaşılmıyor.

Bu düzenin ölçeklenmesi de aynı ilkeye dayanıyor: zincire dördüncü, beşinci bir aşama eklense, her yeni aşama yalnız bir önceki aşamanın çıktısını girdi olarak alıp kendi kanalını açar — asama4 := birBaskaDonusum(asama3) gibi bir çağrı, önceki üç aşamanın hiçbirine dokunmadan zincire eklenebilir. Belirlenimcilik zincirin uzunluğundan bağımsız, çünkü her ek halka de aynı tek-yazıcı/tek-okuyucu kısıtına uyuyor; üç aşamalı bir hatta doğru olan (sıranın korunması, kapatmanın kendiliğinden yayılması) on aşamalı bir hatta da doğru kalıyor.

boru hattı terimi burada bir niteleyiciyle kullanılıyor: bu, aşamalı akış boru hattı — kabuk programlamadaki komut boru hattından ya da sürekli teslimat hattındaki adımlardan ayrı bir kavram, yalnız adı aynı. Ortak olan şey, her ikisinde de bir çıktının bir sonraki adımın girdisi olması; farklı olan, buradaki “aşama”nın bir kabuk süreci ya da bir dağıtım basamağı değil, bir kanal çifti etrafında kurulmuş bir goroutine olması.

Her aşama işlevinin dönüş tipi de bunu yansıtıyor: uret, kareAl ve birEkle üçü de <-chan int döndürüyor — yalnız alınabilen, gönderilemeyen bir kanal. Bu, çağıranın yanlışlıkla döndürülen kanala kendi değerini göndermesini derleme zamanında engelliyor; kanalın tek yazıcısı onu oluşturan işlevin kendi goroutine’i, ve dönüş tipi bu tekliği tipe kodluyor. Aynı kısıtlama girdi parametrelerinde de var: kareAl ve birEkle, girdi <-chan int alıyor — yalnız kendi kanallarından okuyabilir, oraya yazamazlar. Bu küçük tip ayrıntısı, “her kanalın tek bir sahibi var” ilkesini yalnız bir tasarım niyeti olarak değil, derleyicinin denetlediği bir kural olarak kuruyor.

kareAl ve birEklein gövdeleri kelimesi kelimesine aynı kalıba uyuyor: bir çıktı kanalı aç, bir goroutine başlat, girdi kanalında aralık yap, her değeri dönüştürüp çıktıya yaz, girdi biterken kendi çıktını kapat. Bu tekrar tesadüf değil — herhangi bir dönüştürücü aşama aynı üç adımı izliyor, değişen yalnız aradaki dönüşüm (deger * deger ya da deger + 1). uret bu kalıbın bir ucu (girdi kanalı yok, yalnız üretim) ve ana goroutine’deki tüketim döngüsü öbür ucu (çıktı kanalı yok, yalnız tüketim); aradaki her aşama ikisinin ortasında, hem bir girdisi hem bir çıktısı olan aynı şablonu tekrarlıyor. Bir hatta kaç aşama olursa olsun, kapatma sorumluluğu her zaman bu şablonun içinde, defer close(cikti) satırında duruyor.

Erken Bırakma: Önceki Aşamalar Hiç Durmaz

Yukarıdaki ölçüm zincirin sonuna kadar tüketti — asama3teki bütün değerler alınana kadar döngü sürdü. Ama bir tüketici her zaman zincirin sonuna kadar gitmeyebilir: aradığı ilk sonucu bulduğunda durabilir, bir hata alabilir, ya da başka bir sebeple vazgeçebilir — örneğin bir arama işlemi ilk eşleşmeyi bulduğunda kalan değerleri hiç istemeyebilir. Aşağıdaki ölçüm bu durumu en yalın haliyle kuruyor: iki aşamalı bir hat açılıyor ve tüketici ikinci değerden sonra vazgeçiyor, for range döngüsünden break ile çıkıyor.

// erkenbirak.go — tuketici erken vazgecerse onceki asamalar hic durmaz (cozum sonraki derste)
package main

import (
	"fmt"
	"runtime"
)

func uret(n int) <-chan int {
	cikti := make(chan int)
	go func() {
		defer close(cikti)
		for i := 1; i <= n; i++ {
			cikti <- i
		}
	}()
	return cikti
}

func kareAl(girdi <-chan int) <-chan int {
	cikti := make(chan int)
	go func() {
		defer close(cikti)
		for deger := range girdi {
			cikti <- deger * deger
		}
	}()
	return cikti
}

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

	asama1 := uret(5)
	asama2 := kareAl(asama1)

	alinan := 0
	for range asama2 {
		alinan++
		if alinan == 2 {
			break // tuketici iki degerden sonra vazgeciyor
		}
	}

	fmt.Println("-- tuketici 2 degerden sonra vazgeciyor --")
	fmt.Printf("alinan deger sayisi: %d\n", alinan)
	fmt.Printf("durmayan asama kaldi mi: %v\n", runtime.NumGoroutine() > taban)
}

-- tuketici 2 degerden sonra vazgeciyor --
alinan deger sayisi: 2
durmayan asama kaldi mi: true

KL13. İki değer alındıktan sonra break ile döngüden çıkılıyor — ve bu, zincirin geri kalanına hiç haber verilmiyor. Zincirde tam olarak nerede takılma olduğu zamanlamaya bağlı, ama olası uçları saymak öğretici: kareAlin goroutine’i ya kendi kareleme sonucunu asama2ye göndermeye çalışırken takılıyor — çünkü ana goroutine artık onu almıyor — ya da bir sonraki değeri asama1den almaya çalışırken bekliyor. Eğer kareAl gönderirken takılıysa, uretin goroutine’i de bir sonraki değerini asama1e göndermeye çalıştığında takılıyor, çünkü kareAl artık onu okuyamayacak kadar meşgul. Zincir kısa olduğu için bu iki olası durumun her ikisi de aynı sonuca varıyor: en az bir, çoğunlukla iki goroutine kalıcı olarak bloke oluyor. Tam olarak hangi durumun gerçekleştiği koşumdan koşuma değişebilir (tamponsuz kanallarda tam sınır zamanlamaya bağlı), ama en az bir aşamanın takılı kaldığı hiç değişmiyor — bu yüzden ölçüm yalnız runtime.NumGoroutine() karşılaştırmasını, kaç goroutine’in kaldığını değil, yazdırıyor.

KL14. Bu, önceki iki derste görülen kapatılmayan kanal sızıntılarından farklı bir sebeple ortaya çıkıyor. Orada sorun kapatmanın hiç yapılmamasıydı; burada her aşama kendi kapatma sorumluluğunu doğru taşıyor — uret de, kareAl da kendi çıktısını işi bitince kapatacak. Sorun, işin hiç bitmemesi: uretin “bitmesi” beşinci değeri göndermesine bağlı, ama üçüncü değerden sonra kimse almıyorsa gönderim hiç tamamlanmıyor, dolayısıyla defer close satırına da hiç ulaşılmıyor. Doğru kurulmuş bir kapatma sorumluluğu, zincirin her halkasının birbirini beklediği varsayımına dayanıyor; bu varsayım tüketici erken çıktığında bozuluyor ve hiçbir aşama bunu fark edecek bir mekanizmaya sahip değil.

Bu, önceki dersteki “doğru kurulum bile üreticilerin sonunda biteceğini varsayıyor” gözleminin zincire yayılmış hâli. Orada tek bir varsayım vardı (üreticiler bitecek); burada zincirdeki her aşama bir öncekinin ve bir sonrakinin normal çalıştığını varsayıyor, ve bu varsayımların zincirleme etkisi zincir uzadıkça büyüyor. Üç aşamalı bir hatta bir tüketicinin erken çıkması iki goroutine’i etkileyebiliyorken, on aşamalı bir hatta aynı erken çıkış dokuz goroutine’i bloke edebilir — sızıntının büyüklüğü zincirin uzunluğuyla birlikte artıyor, tüketicinin ne kadar erken vazgeçtiğinden bağımsız olarak.

Üç dersin sızıntı ölçümlerini yan yana koymak farkı netleştiriyor. İşçi havuzunda sızıntının sebebi kapatmanın unutulmasıydı — close(isler) satırı hiç yazılmamıştı. Yaymada sızıntının sebebi sorumluluğun hiç kimseye verilmemesiydi — üçüncü taraf goroutine hiç kurulmamıştı. Burada sorumluluk hem yazılmış hem doğru tarafa verilmiş, ama tetiklenmesi gereken olay (girdinin tükenmesi) hiç gerçekleşmiyor, çünkü girdiyi tüketecek olan zaten aradan çekilmiş. Üçü de aynı boolean ölçütle (runtime.NumGoroutine() > taban) yakalanıyor, ama üçünün kök sebebi üç ayrı yerde duruyor: eksik kapatma, sahipsiz kapatma, tetiklenmeyen kapatma.

Bu sınır, ders içinde çözülmüyor — yalnız ölçülüyor. Bir aşamayı “tüketici artık dinlemiyor” bilgisiyle donatmak, o aşamanın kendi döngüsünün dışından gelen bir sinyali denetlemesini gerektiriyor; her aşamanın for deger := range girdi döngüsüne ek bir dal eklemesi, ve bu dalın bütün aşamalara aynı anda ulaşan tek bir kaynaktan gelmesi gerekiyor. Bu tam olarak sıradaki dersin konusu.

Özet

• Aşamalı akışta (boru hattı) her aşama kendi çıktı kanalını kendisi açar ve kendisi kapatır; kapatma sorumluluğu tek bir yerde değil, zincirin her halkasında ayrı ayrı durur.
• Zincirdeki her kanalın tek bir yazıcısı ve tek bir okuyucusu olduğu için sıra hiçbir sıralama aracına gerek kalmadan korunur — yayma/toplamadaki paylaşılan kanallardan farklı.
• boru hattı terimi burada aşamalı akış anlamında, kabuk ya da teslimat hattı boru hatlarından ayrı bir kavram olarak kullanılıyor.
• Tüketici zincirden erken çıktığında önündeki aşamalar bunu hiç bilmiyor ve kalıcı olarak bloke oluyor; sızıntı yine yalnız bir boolean olarak ölçülüyor.
• Bu sızıntının kaynağı, önceki derslerdeki kapatılmayan kanallardan farklı: kapatma sorumluluğu doğru kurulu, ama işin kendisi hiç bitmiyor.
• Sızıntının büyüklüğü zincirin uzunluğuyla birlikte artıyor: bir tüketicinin erken çıkışı kısa bir hatta bir goroutine’i, uzun bir hatta zincirin geri kalanını etkileyebilir.

Sonraki Adım

Bu derste görülen sızıntının çözümü, bir aşamanın yalnız kendi girdi kanalını değil, dışarıdan gelen bir vazgeçme sinyalini de dinlemesini gerektiriyor — ve bu sinyalin zincirin başından sonuna kadar bütün aşamalara aynı anda ulaşması gerekiyor, tek tek her aşamaya elle iletilmeden. Sıradaki ders bu sinyali taşıyan aracı ölçüyor: bir iptalin, zincirdeki bütün bekleyenlere nasıl aynı anda ulaştığını, ve bir goroutine’in bu sinyali hiç denetlemediğinde ne olduğunu.