Fan-Out/Fan-In w Go: wzorzec dla równoległego przetwarzania danych

Wzorzec Fan-Out/Fan-In w Go: elastyczne równoległe przetwarzanie danych

Wzorzec Fan-Out/Fan-In zapewnia w Go potężny mechanizm do rozdzielania zadań między gorutyny i późniejszej agregacji wyników. W przeciwieństwie do Worker Pool, oferuje większą elastyczność dzięki indywidualnym pipeline'om przetwarzania dla każdego workera, co jest szczególnie przydatne w przypadku skomplikowanych łańcuchów transformacji danych.

Główne zasady wzorca

Wzorzec składa się z dwóch kluczowych faz. Faza Fan-Out odpowiada za rozdzielanie danych wejściowych między kilkoma workerami (gorutynami). Każdy worker otrzymuje swoje zadanie i przetwarza je niezależnie, wysyłając wynik w dedykowanym kanale. Faza Fan-In zbiera wyniki ze wszystkich indywidualnych kanałów workerów w jeden wspólny kanał wyjściowy do dalszej obróbki przez konsumenta.

Główna różnica w porównaniu z Worker Pool polega na architekturze kanałów:

• Worker Pool: wszyscy workerzy zazwyczaj czytają zadania z jednego kanału i piszą wyniki w drugim wspólnym kanale.
• Fan-Out/Fan-In: każdy worker ma swój własny kanał wyjściowy, co wymaga dodatkowego etapu łączenia (merge).

To podejście generuje pewne obciążenie z powodu konieczności dodatkowych gorutyn i kanałów do łączenia, ale zapewnia korzyści dla architektur pipeline'owych, gdzie kolejne etapy przetwarzania mogą wymagać oddzielnych strumieni danych.

Implementacja wzorca w praktyce

Rozważmy praktyczną implementację wzorca, która celowo różni się od klasycznego Worker Pool, aby zademonstrować kluczowe cechy.

Funkcja workera (Worker)

Worker w tej implementacji to funkcja, która uruchamia gorutynę i zwraca kanał dla wyników. Różni się to od podejścia, w którym worker przyjmuje kanał jako parametr.

func worker(x int, workerID int) chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        fmt.Printf("worker %d, job %d\n", workerID, x)
        out <- x * x
    }()
    return out
}

Ważne aspekty implementacji:

• Gorutyna zamyka swój kanał out za pomocą defer po wysłaniu wyniku.
• W celach demonstracyjnych praca jest trywialna – podnoszenie liczby do kwadratu.
• W kodzie produkcyjnym taka implementacja wymaga ulepszenia: konieczne jest zarządzanie cyklem życia poprzez kontekst dla graceful shutdown i obsługi timeoutów.

Faza Fan-Out (Rozdzielanie)

Rozdzielacz (producer) uruchamia po jednym workerze dla każdego elementu danych wejściowych. Kanały zwracane przez workerów są zapisywane w wycinku do późniejszego łączenia.

for i, x := range data {
    chnls = append(chnls, worker(x, i+1))
}

Kluczowa cecha: liczba workerów jest dynamicznie określana przez liczbę zadań, co nie jest obowiązkowe, ale pokazuje elastyczność wzorca. W rzeczywistych scenariuszach liczbę równoległych workerów zazwyczaj ogranicza się.

Faza Fan-In i funkcja łączenia (Merge)

Funkcja merge wykonuje krytycznie ważne zadanie – łączy dane z wielu kanałów w jeden. W tym celu:

• Tworzy wspólny niebuforowany kanał wyjściowy out.
• Dla każdego kanału wejściowego uruchamia gorutynę, która czyta dane i przekierowuje je do out.
• W oddzielnej gorutynie oczekuje na zakończenie wszystkich gorutyn-czytelników za pomocą sync.WaitGroup, a następnie zamyka kanał out.

func merge(chnls ...chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, ch := range chnls {
        wg.Add(1)
        go func(c chan int) {
            defer wg.Done()
            for x := range c {
                out <- x
            }
        }(ch)
    }

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

Techniczny szczegół: oczekiwanie na zakończenie (wg.Wait()) jest wykonywane w oddzielnej gorutynie. Jeśli wykonać je w głównym wątku, może powstać deadlock, ponieważ konsument może nie zdążyć rozpocząć czytania z kanału out, a operacja zapisu zostanie zablokowana.

Konsument (Consumer)

Konsument czyta wyniki z jednostkowego kanału, uzyskanego po łączeniu. Pętla for range automatycznie kończy się, gdy funkcja merge zamknie kanał out.

for x := range out {
    results = append(results, x)
}

Co jest ważne

• Elastyczność architektury: Indywidualne pipeline'y dla workerów umożliwiają łatwe wbudowanie dodatkowych etapów przetwarzania danych.
• Dynamiczne skalowanie: Teoretycznie wspierane jest tworzenie workera na każde zadanie, choć w praktyce wymagana jest kontrola nad liczbą równoległych gorutyn.
• Obciążenie: Wzorzec wymaga dodatkowych zasobów (kanały, gorutyny) dla etapu łączenia w porównaniu z prostszymi podejściami.
• Zarządzanie cyklem życia: Kod demonstracyjny nie używa kontekstów, co jest nieakceptowalne dla środowiska produkcyjnego, gdzie potrzebna jest kontrola nad anulowaniem operacji i timeoutami.
• Czytelność kodu: Użycie niebuforowanych kanałów i pętli for range sprawia, że kod jest kompaktowy i zrozumiały.

Zalecenia dla użytkowania w produkcji

Kod demonstracyjny służy celom edukacyjnym. Do eksploatacji przemysłowej konieczne są znaczące ulepszenia:

• Wprowadzenie kontekstu (context.Context): Obowiązkowo przekazuj kontekst workerom i funkcji merge dla obsługi anulowania operacji, deadline'ów i timeoutów.
• Ograniczenie równoległości: Uruchamianie nieograniczonej liczby gorutyn (po jednej na zadanie) może prowadzić do wyczerpania zasobów. Używaj semaforów lub puli workerów z ustalonym rozmiarem wewnątrz fazy Fan-Out.
• Obsługa błędów: Zaimplementuj mechanizm przekazywania błędów od workerów do konsumenta, np. przez kanał specjalnego typu lub strukturę wyniku.
• Buforowanie kanałów: W niektórych scenariuszach buforowane kanały mogą poprawić wydajność, zmniejszając blokady, ale ich użycie wymaga starannej analizy.

Wzorzec Fan-Out/Fan-In jest potężnym narzędziem w arsenale developera Go do budowania efektywnych i skalowalnych konwejerów równoległego przetwarzania danych, szczególnie gdy wymagana jest elastyczność w organizacji strumieni informacji między etapami obliczeń.

— Editorial Team