62 / 73

Pipelines

Goroutine'lar va kanallardan foydalanishni o'rgangach, endi ularni birgalikda ma'lumotlarni oqimli qayta ishlovchi concurrent konveyerlar (pipeline) ko'rinishida yig'ishni ko'rib chiqamiz!

Bob mundarijasi:

Goroutine sizib chiqishi (Leaked goroutine)
Bekor qilish kanali (Cancel channel)
Bekor qilish va Bajarildi (Cancel vs. Done)
Kanallarni birlashtirish: Ketma-ket (Merging: sequentially)
Kanallarni birlashtirish: Bir vaqtda (Merging: concurrently)
Kanallarni birlashtirish: select orqali (Merging: select)
Konveyer (Pipeline) arxitekturasi
Topshiriq 1: Ko'p bosqichli konveyer
Goroutine sizib chiqishining oldini olish
Topshiriq 2: Xavfsiz generator konveyeri
Konveyerda xatolarni boshqarish (Error handling)
Topshiriq 3: Xatoliklarni uzatuvchi konveyer
Xulosa va keyingi qadamlar

Goroutine sizib chiqishi

Berilgan oraliqdagi sonlarni kanalga uzatuvchi oddiy funksiyani ko'rib chiqamiz:

example.go

func rangeGen(start, stop int) <-chan int {
	out := make(chan int)
	go func() {
		for i := start; i < stop; i++ {
			out <- i
		}
		close(out)
	}()
	return out
}

U normal holatda kutilganidek ishlaydi:

example.go

func main() {
	generated := rangeGen(41, 46)
	for val := range generated {
		fmt.Println(val)
	}
}

Natija:

text

Ammo agar qabul qiluvchi sikldan muddatidan oldin chiqib ketsa nima bo'ladi?

example.go

func main() {
	generated := rangeGen(41, 46)
	for val := range generated {
		fmt.Println(val)
		if val == 42 {
			break // muddatidan oldin to'xtatildi!
		}
	}
}

Natija:

text

41
42

Bir qarashda dastur to'g'ri ishlagandek ko'rinadi. Ammo aslida rangeGen() ichidagi goroutine abadiy tiqilib qoldi (stuck):

example.go

func rangeGen(start, stop int) <-chan int {
	out := make(chan int)
	go func() {
		for i := start; i < stop; i++ { // (1)
			out <- i                    // (2)
		}
		close(out)
	}()
	return out
}

Chunki main() siklni 42 sonida to'xtatdi va generated kanalidan boshqa hech narsa o'qimay qo'ydi. Natijada rangeGen() ichidagi goroutine ➊ 43 sonini out kanaliga yuborishga urinib ➋, abadiy bloklanib qoldi. Kanal yopilmadi va goroutine xotiradan o'chirilmadi!

Ushbu sodda misolda bu unchalik xavfli ko'rinmasligi mumkin (chunki main() tugashi bilan dastur to'xtaydi). Ammo agar bu uzoq ishlaydigan veb-server bo'lsa va rangeGen() doimiy chaqirilib tursa, har bir erta to'xtashda tizimda bittadan o'lik goroutine qolib ketadi. Bu holat goroutine leak (goroutine sizib chiqishi) deb ataladi. Go'ning Garbage Collector'i faol bloklangan goroutine'larni tozalay olmaydi va vaqt o'tishi bilan xotira to'lib (OOM) dastur qulaydi.

Bizga goroutine'ni tashqaridan muddatidan oldin xavfsiz to'xtatish yo'li kerak.

Bekor qilish kanali

Buning uchun alohida cancel channel (bekor qilish kanali) yaratamiz. main() funksiyasi ushbu kanal orqali rangeGen()ga to'xtash signalini uzatadi:

example.go

func main() {
	cancel := make(chan struct{}) // (1)
	defer close(cancel)           // (2)

	generated := rangeGen(cancel, 41, 46) // (3)
	for val := range generated {
		fmt.Println(val)
		if val == 42 {
			break
		}
	}
}

Biz cancel kanalini ochamiz ➊ va darhol defer close(cancel) deb e'lon qilamiz ➋. Bu ajoyib andoza bo'lib, funksiya qanday tugashidan qat'i nazar (oddiy return, break yoki hatto panic) kanal yopilishini va barcha qaram goroutine'lar xavfsiz tozalanishini ta'minlaydi.

Endi rangeGen ichida cancel signali kelgan-kelmaganini tekshirish kerak:

example.go

func rangeGen(cancel <-chan struct{}, start, stop int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for i := start; i < stop; i++ {
			select {
			case <-cancel:
				return // bekor qilish signali olindi, darhol chiqish!
			case out <- i:
				// qiymat muvaffaqiyatli uzatildi, davom etamiz
			}
		}
	}()
	return out
}

select operatori tufayli:

Agar cancel yopilgan bo'lsa, goroutine darhol ishini tugatadi va defer close(out) orqali chiqish kanalini ham yopadi;
Agar cancel yopilmagan bo'lsa va out kanali qabul qilishga tayyor bo'lsa, navbatdagi son yuboriladi.

Endi hech qanday goroutine sizib chiqishi (leak) yuz bermaydi!

Bekor qilish va Bajarildi

cancel va done kanallari bir xil chan struct{} turiga ega bo'lsa-da, ularning semantik maqsadi farqlanadi:

cancel kanali: "Iltimos, ishingni muddatidan oldin to'xtat, natijang endi kerak emas" degan signal;
done kanali: "Vazifa muvaffaqiyatli tugallandi" degan xabar.

Ikkala holatda ham kanalni close() qilish bir vaqtning o'zida barcha tinglovchilarga broadcast (ommaviy) xabar yetkazishning eng samarali usulidir.

Kanallarni birlashtirish: Ketma-ket

Tasavvur qiling, sizda bir nechta kirish kanallari bor va ulardagi barcha ma'lumotlarni bitta umumiy chiqish kanaliga yig'ishingiz kerak (fan-in andozasi).

Eng sodda usul — ularni navbatma-navbat (ketma-ket) o'qib, bitta kanalga yozish:

example.go

func mergeSeq[T any](cancel <-chan struct{}, channels ...<-chan T) <-chan T {
	out := make(chan T)
	go func() {
		defer close(out)
		for _, ch := range channels {
			for val := range ch {
				select {
				case <-cancel:
					return
				case out <- val:
				}
			}
		}
	}()
	return out
}

Bu usul oson, ammo sekin: u birinchi kanal to'liq yopilmaguncha ikkinchi kanalga o'tmaydi.

Kanallarni birlashtirish: Bir vaqtda

Agar barcha kirish kanallari parallel ishlayotgan bo'lsa, ularni parallel (concurrent) birlashtirish kerak. Buning uchun har bir kirish kanali uchun alohida nusxa ko'chiruvchi goroutine ochiladi va sync.WaitGroup orqali ularning barchasi tugashi kutiladi:

example.go

func merge[T any](cancel <-chan struct{}, channels ...<-chan T) <-chan T {
	out := make(chan T)
	var wg sync.WaitGroup

	// Har bir kanal uchun alohida goroutine:
	forward := func(ch <-chan T) {
		defer wg.Done()
		for val := range ch {
			select {
			case <-cancel:
				return
			case out <- val:
			}
		}
	}

	wg.Add(len(channels))
	for _, ch := range channels {
		go forward(ch)
	}

	// Barcha forwarderlar tugagach out kanalini yopuvchi kuzatuvchi goroutine:
	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

Bu klassik Go Fan-In (birlashtirish) andozasidir.

Konveyer (Pipeline) arxitekturasi

Konveyer (pipeline) — bu bir bosqichning chiqishi keyingi bosqichning kirishi bo'lib xizmat qiladigan zanjirdir:

kod

[Generator] ---> (ch1) ---> [Filtr] ---> (ch2) ---> [Transformator] ---> (ch3) ---> [Iste'molchi]

Keling, to'liq amaliy konveyer quramiz:

Sonlar generatori (gen)
Faqat toq sonlarni qoldiruvchi filtr (filterOdd)
Har bir sonni kvadratga ko'taruvchi bosqich (squarer)

example.go

package main

import "fmt"

func gen(cancel <-chan struct{}, nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			select {
			case <-cancel:
				return
			case out <- n:
			}
		}
	}()
	return out
}

func filterOdd(cancel <-chan struct{}, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			if n%2 != 0 {
				select {
				case <-cancel:
					return
				case out <- n:
				}
			}
		}
	}()
	return out
}

func squarer(cancel <-chan struct{}, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			select {
			case <-cancel:
				return
			case out <- n * n:
			}
		}
	}()
	return out
}

func main() {
	cancel := make(chan struct{})
	defer close(cancel)

	numbers := gen(cancel, 1, 2, 3, 4, 5, 6, 7, 8, 9)
	odds := filterOdd(cancel, numbers)
	squares := squarer(cancel, odds)

	for res := range squares {
		fmt.Printf("%d ", res) // 1 9 25 49 81
	}
	fmt.Println()
}

Topshiriq 1: Ko'p bosqichli konveyer

Berilgan satrlar oqimidan barcha bo'sh satrlarni olib tashlaydigan va qolgan satrlarni katta harflarga (UPPERCASE) o'giradigan ikki bosqichli konveyer yozing.

example.go

package main

import (
	"fmt"
	"strings"
)

func cleanAndUpper(cancel <-chan struct{}, in <-chan string) <-chan string {
	out := make(chan string)
	// TODO: in kanalidan satrlarni o'qing, bo'sh bo'lmaganlarini strings.ToUpper qilib out ga yuboring
	return out
}

func main() {
	cancel := make(chan struct{})
	defer close(cancel)

	in := make(chan string)
	go func() {
		defer close(in)
		for _, s := range []string{"go", "", "concurrency", " ", "pipeline", ""} {
			in <- s
		}
	}()

	for res := range cleanAndUpper(cancel, in) {
		fmt.Println(res)
	}
}

Konveyerda xatolarni boshqarish

Konveyer bosqichlarida xatolik yuz berganda (masalan, fayl topilmadi yoki tarmoq xatosi), butun konveyerni to'xtatish yoki xatolikni keyingi bosqichga xavfsiz yetkazish kerak bo'ladi.

Buning eng toza usuli — kanal orqali faqat T qiymatini emas, balki qiymat va xatolik juftligini o'z ichiga olgan struktura yuborishdir:

example.go

type Result[T any] struct {
	Value T
	Err   error
}

func parseNumbers(cancel <-chan struct{}, in <-chan string) <-chan Result[int] {
	out := make(chan Result[int])
	go func() {
		defer close(out)
		for s := range in {
			val, err := strconv.Atoi(s)
			res := Result[int]{Value: val, Err: err}
			select {
			case <-cancel:
				return
			case out <- res:
			}
		}
	}()
	return out
}

Bu andoza Go'da juda keng qo'llaniladi, chunki u hech qanday panic chiqarmaydi va xatolarni tartibli boshqarishga imkon beradi.

Xulosa va keyingi qadamlar

Ushbu bobda biz konveyerlarning professional arxitekturasini qurdik:

Goroutine leak xavfi va uning oldini oluvchi cancel kanali;
Bir nechta kanallarni bitta kanalga parallel birlashtirish (Fan-in);
Bosqichma-bosqich mustaqil modullardan konveyer zanjirini tuzish;
Result[T] orqali xatolarni oqim bo'ylab toza uzatish.

Keyingi bob: 04. Time (Taymerlar, tanaffuslar va vaqt boshqaruvi)!

Oldingi

Keyingi