Fan-out / Fan-in — işi paylamaq və nəticələri toplamaq

1. Nəyə görə fan-out və fan-in lazımdır

Paralel kod yazmağa yenicə başlayanda demək olar ki, həmişə bir sadəlövh plan yaranır: “N qorin işə salım, nəsə hesablasınlar, sonra nəticələri birtəhər toplayaram”. Buradakı sadəlövhlük “birtəhər” sözündədir: paralellikdə “birtəhər” adətən “cümə axşamı axşamı ilişib qalan proqramın qarşısında oturub bunun karma olduğunu düşünəcəksiniz” deməkdir.

Fan-out/fan-in — sizə aydın sxem verən iki çox praktik şablondur: müstəqil işi bir neçə qorinə necə paylamaq (fan-out), və sonra onların cavablarını bir yerə necə toplamaq (fan-in). Bu, nə sehrdir, nə də framework; sadəcə intizamdır: kim yazır, kim oxuyur, kim kanalı bağlayır və kimsə “çıxıb gedəndə” nə baş verir.

Bunu başda saxlamaq üçün restoranı təsəvvür edin. Fan-out — siz sifarişləri müxtəlif aşpazlara paylayırsınız. Fan-in — hazır yeməkləri bir buraxılış masasında toplayırsınız. Vacibdir: masa bağlanıbsa, aşpazlar boşqabla əbədi dayanmalı deyillər.

Ümumi sxemi çəkək:

flowchart LR
    A[Input data] -->|distribute| B1[worker 1]
    A -->|distribute| B2[worker 2]
    A -->|distribute| B3[worker 3]
    B1 -->|result| C[shared results]
    B2 -->|result| C
    B3 -->|result| C
    C --> D[collector / main]

2. Fan-out: işi qorinlərə paylayırıq

Fan-out adətən sadə bir nöqtədən başlayır: bizim tapşırıqlar siyahımız var (məsələn, ədədlər, sətirlər, elementlər). Hər elementi müstəqil şəkildə emal etmək istəyirik. Emal məntiqi istənilən şey ola bilər: hesablamaq, parse etmək, yoxlamaq, sətiri “normallaşdırmaq”, hash hesablamaq və s. Tədris nümunələrində sadə bir şey edəcəyik, amma mühəndis kimi düşünəcəyik: “emalçı tez/yavaş işləsə, nəticə oxuyucusu da fərqli sürətdə olsa, nə baş verər?”

Ən kiçik fan-out: ədədlərin kvadratları

Budur “skelet”: results yaradırıq, hər element üçün bir qorin işə salırıq, sonra N nəticəni oxuyuruq.

package main

import (
	"fmt"
)

func main() {
	values := []int{2, 3, 4}
	results := make(chan int, len(values)) // bütün nəticələr üçün buffer

	for _, v := range values {
		go func(v int) {
			results <- v * v
		}(v) // vacibdir: v parametr kimi ötürüldü
	}

	for i := 0; i < len(values); i++ {
		fmt.Println(<-results) // sıra zəmanət verilmir
	}
}

İki vacib ideyaya diqqət yetirin.

Birincisi, results len(values) qədər buferləşdirilib. Bu o deməkdir: hər bir qorin nəticəni “yerləşdirə” və bitirə bilər, hətta main hələ oxumağa başlamasa belə. Bu, təsadüfi bloklanmalardan çox faydalı bir sığortadır.

İkincisi, v-ni anonim funksiyaya parametr kimi ötürürük: }(v). Go 1.22+ versiyasında belə bir çox tipik range tələləri yaxşılaşıb, amma “qorinin girişini açıq etmək” vərdişi — kəmər bağlamaq kimi bir şeydir. Bəlkə siz onsuz da çatarsınız, amma bağlı halda daha rahatdır.

Nəyə görə nəticə göndərilməsi bloklana bilər

İndi vacib məqam: results <- x göndərilməsi də sinxronizasiyadır. Kanal bufersizdirsə (və ya buffer doludursa), göndərən oxuyucunu gözləyəcək. Oxuyucu yoxdursa və olmayacaqsa — qorin ilişib qalır.

Proqram həyatı baxımından bu belə görünür: kodunuz “sanki bitdi”, amma proses nədənsə onlarla/ yüzlərlə/ minlərlə qorini saxlayır, onlar isə göndərmə nöqtəsində dayanır. Bu, qorin sızmasına aparan ən tipik yollardan biridir.

3. Fan-in: nəticələri toplayırıq

Fan-in — hekayənin əks tərəfidir. Nəticələrin bir kanal boyunca axıb main və ya “collector” tərəfindən oxunmasını istəyirik. Praktikada fan-in iki variantda görünür.

Birinci variant: artıq tək bir ümumi results var (kvadrat nümunəsində olduğu kimi) və biz sadəcə ondan N dəyər oxuyuruq. Bu, əgər N-i bilirik və hər worker-in dəqiq bir nəticə göndərdiyinə zəmanət veririksa, dürüst işləyir.

İkinci variant: bir neçə giriş kanalımız var (məsələn, müxtəlif data mənbələri) və biz onları bir çıxış kanalında “birləşdirmək” istəyirik. Məhz burada klassik merge funksiyası yaranır.

merge: iki kanalı birində birləşdiririk

Aşağıda minimal versiya var: iki qorin dəyərləri “ötürür”, ayrıca koordinasiya qorini isə hər iki ötürmə bitəndə out-u bağlayır.

package main

import (
	"sync"
)

func merge(a, b <-chan int) <-chan int {
	out := make(chan int)
	var wg sync.WaitGroup
	wg.Add(2)

	forward := func(in <-chan int) {
		defer wg.Done()
		for v := range in {
			out <- v
		}
	}

	go forward(a)
	go forward(b)

	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

Burada əsas intizam budur: out-u bir koordinatordan başqa heç kim bağlamır. “Kim yoruldu”, “kim birinci tapdı”, “gəlin hamımız bağlayaq, sonra baxarıq” — yox. Go-da kanalın bağlanması iki dəfə edilə bilməyən bir hadisədir.

Kodu sadəcə “köçürmək” yox, protokolu anlamaq vacib olduğuna görə, bunu sözlə də formalaşdıraq.

forward in-i range vasitəsilə oxuyur. Bu o deməkdir: forward yalnız giriş kanalı bağlananda bitəcək. Deməli, in-in bağlanması onu yazan tərəfin (producer) məsuliyyətidir. Hər iki forward bitəndə koordinatör close(out) edir, və out oxuyucusu range ilə oxunu bitirə bilər.

4. Şablonları birləşdirir və genişləndiririk

Fan-out + fan-in: tapşırıqların paralel emalçısı

İndi hər iki şablonu real tətbiqlərdə olan işlərə bənzəyən kiçik bir hekayədə birləşdirək. Tutaq ki, bizim “tapşırıqlar” siyahımız var (tədris üslubumuzda bunlar sadəcə ədədlərdir) və onları paralel emal edib sonra nəticələri toplamaq istəyirik.

Daha aydın olsun deyə nəticə tipi yaradaq.

package main

import (
	"fmt"
	"time"
)

type Result struct {
	Input int
	Value int
}

func work(x int) Result {
	time.Sleep(20 * time.Millisecond) // işi imitasiya edirik
	return Result{Input: x, Value: x * x}
}

func main() {
	values := []int{2, 3, 4}
	results := make(chan Result, len(values))

	for _, v := range values {
		go func(v int) {
			results <- work(v)
		}(v)
	}

	for i := 0; i < len(values); i++ {
		r := <-results
		fmt.Println(r.Input, "->", r.Value) // məsələn: 3 -> 9
	}
}

Bu artıq tam fan-out-dur (hər tapşırıq üçün bir qorin) və fan-in-dir (cavabları toplayan bir results kanalı).

Diqqət yetirin: çıxış sırası “necə alındısa elə” olacaq. Əgər sizə sabit sıra lazımdırsa — adətən nəticələri sonra sıralayırsınız, ya da başqa protokoldan istifadə edirsiniz (məsələn, nəticədə indeks saxlamaqla). Paralellik nadir hallarda “həm sürətli, həm də mənə xoş olan sırada hər şey” verir, ona görə “sıra gözləməmək” mühəndis vərdişi əsəblərə qənaət edir.

İlk nəticə qalib gəlir

Bəzən sizə bütün nəticələr lazım olmur. Sizə ilk uğurlu/ilk gələn nəticə lazımdır. Tipik nümunə: “bir neçə mənbədən soruşaq və daha tez gələn cavabı götürək”.

Go-da bu, adətən tutumu 1 olan kanal və select ilə default vasitəsilə bloklanmayan göndərmə ilə edilir. Bloklanmayan göndərmə qorinlərin ilişib qalmamasına zəmanət verir, hətta oxuyucu artıq “çıxıb gedibsə” və ya nəticə artıq lazım deyilsə belə.

package main

import (
	"fmt"
)

func main() {
	values := []int{2, 3, 4}
	out := make(chan int, 1) // yalnız bir cavab lazımdır

	for _, v := range values {
		go func(v int) {
			r := v * v
			select {
			case out <- r:
				// qalib nəticəni göndərdi
			default:
				// kanal artıq məşğuldur: uduzan sadəcə çıxır
			}
		}(v)
	}

	fmt.Println(<-out) // bunlardan biri: 4, 9, 16
}

Burada nəyə görə 1 buffer vacibdir? Çünki əks halda yarış vəziyyətinə düşə bilərsiniz: “qalib” göndərməyə çalışdı, amma oxuyucu hələ oxumağa başlamayıb. Bu şablonun ideyası budur: buffer ilk göndərənə dəyəri yerləşdirmək üçün yer verir, default isə digərlərinin ilişməməsini təmin edir.

Burada, yeri gəlmişkən, çox aydın görünür ki, default təhlükəli alətdir. O, əməliyyatı bloklanmayan edir, amma bunun bahasına siz məlumat itirə bilərsiniz. Bu şablonda məlumat itkisi müqavilə üzrə xüsusiyyətdir (“bizə bir cavab lazımdır”). Əgər müqaviləyə görə itki yolverilməzdirsə, default istifadə etmək olmaz.

5. Protokollar: kanalı kim və nə vaxt bağlayır

Paralellikdə problem çox vaxt sintaksisdə olmur, insanların protokolu açıq danışmamasında olur. Gəlin bir neçə qaydanı kiçik cədvəl formasında sabitləyək. Bu, “yeganə doğru həqiqət” deyil, amma çox yaxşı başlanğıc müqaviləsidir.

Bu, çox writer olan fan-out-da praktikada necə görünür? Çox vaxt WaitGroup istifadə olunur və ayrıca bir qorin results-i bütün writer-lər bitəndə bağlayır. Bu, demək olar ki, merge-də gördüyümüz eyni üsuldur.

package main

import (
	"sync"
)

func fanOutSquares(values []int) <-chan int {
	out := make(chan int)
	var wg sync.WaitGroup
	wg.Add(len(values))

	for _, v := range values {
		go func(v int) {
			defer wg.Done()
			out <- v * v
		}(v)
	}

	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

İndi istehlakçı belə yaza bilər:

for x := range fanOutSquares([]int{2, 3, 4}) {
	_ = x
}

Bu rahatdır, çünki “axın bitənə qədər oxumaq” adətən “dəqiq N dəfə oxumaq”dan daha asandır. Amma yenə də: bu yalnız kanalı düzgün bağladığınız halda işləyir.

6. Tipik səhvlər

Səhv №1: unbuffered results + “sonra oxuyaram” → göndərənlər ilişib qalır.
Yeni başlayan çox vaxt results := make(chan int) yazır və onlarla qorin işə salır, hər biri nəticə yazır. Nəticələri isə “axırda”, bütün qorinlər işə salındıqdan sonra oxuyur. Amma unbuffered kanal eyni vaxtda oxuyucu tələb edir: qorinlər ilk göndərişlərdə artıq results <- ... üzərində bloklanmağa başlayacaq. Bunun həlli ya gözlənilən say qədər buffer, ya oxuyucunun dərhal oxumağa başlaması, ya da ayrıca collector-dur.

Səhv №2: “ilk nəticə qalib gəlir”, amma göndəriş adi out <- r kimi yazılıb.
İlk baxışda məntiqli görünür: “bir nəticə oxuyaram və gedərəm”. Problem budur ki, uduzan qorinlər yenə göndərməyə çalışacaq və ilişəcək, çünki artıq oxuyucu yoxdur. Düzgün protokol — ya select { case out <- r: default: } vasitəsilə bloklanmayan göndəriş, ya da ayrıca ləğv siqnalıdır.

Səhv №3: out-u bir neçə qorin bağlayır.
Bu, ən çox rast gəlinən “heç yerdən panic”lərdən biridir. merge fan-in-də adamın ağlına gəlir: “girişim bitən kimi çıxışı bağlayım”. Amma giriş iki dənədirsə, çıxışı iki dəfə bağlamağa çalışacaqlar. Həll sadə və darıxdırıcıdır (və buna görə də doğrudur): çıxışı bütün iştirakçıların bitməsini gözləyən bir koordinatör bağlayır (məsələn, WaitGroup vasitəsilə).

Səhv №4: “kanal boşdur” ilə “kanal bağlıdır”ı qarışdırırlar.
Bəzən bloklanmayan oxu edirlər və düşünürlər: “oxunmadısa — deməli bağlıdır”. Yox: “oxunmadı” o deməkdir ki, “hazırda dəyər yoxdur”. Bağlanma ya v, ok := <-ch ilə müəyyən olunur (ok == false olduqda), ya da range-in bitməsi ilə. Əgər məntiqinizi yanlış fərq üzərində qursanız, davranış qeyri-sabit olacaq: bir gün işləyir, bir gün işləmir — paralel səhvlərin klassikası.

Səhv №5: nəticələrin sırasına ümid edirlər.
Fan-out demək olar ki, “orijinal siyahıdakı sıra”nın saxlanmayacağına zəmanət verir. Sıra vacibdirsə, onu məlumatın bir hissəsi etməlisiniz (məsələn, Result-a indeks əlavə edin) və sonra sıralayın, ya da nəticələri indeksə görə strukturda toplayın. Və bəli, bu, çoxlarının ilk dəfə anladığı yerdir ki, paralellik “tez et” düyməsi deyil, “paralel et, amma düşün” müqaviləsidir.