Kanały w Go (Golang): buforowane, niebuforowane, close i range

Wysyłanie i odbieranie

Kanał to typowany przewód między gorutynami. ch <- v wysyła, <-ch odbiera. Kanał tworzy się przez make:

package main

import "fmt"

func main() {
	messages := make(chan string)

	go func() {
		messages <- "ping" // blocks until main receives
	}()

	msg := <-messages // blocks until the goroutine sends
	fmt.Println(msg)
}

Wartość zerowa typu kanałowego to nil, więc var ch chan string bez make daje kanał, który blokuje w nieskończoność. Zawsze twórz kanały przez make.

Kanały niebuforowane synchronizują

make(chan T) tworzy kanał niebuforowany. Wysłanie blokuje, dopóki odbiorca nie weźmie wartości, a odbiór blokuje, dopóki nadawca jej nie dostarczy. Obie gorutyny spotykają się w tym punkcie, dlatego kanał niebuforowany jest w równym stopniu narzędziem synchronizacji, co przewodem na dane. Kiedy poniżej <-done zwraca, wiadomo, że worker skończył wszystko, co było przed wysłaniem:

package main

import "fmt"

func main() {
	done := make(chan struct{})
	result := 0

	go func() {
		for i := 1; i <= 100; i++ {
			result += i
		}
		done <- struct{}{} // signal: finished
	}()

	<-done
	fmt.Println("sum:", result)
}

Odczyt result w main jest tu bezpieczny bez mutexa. Model pamięci Go gwarantuje, że wszystko, co worker zrobił przed wysłaniem, jest widoczne dla main po odpowiadającym mu odbiorze. chan struct{} to idiomatyczny typ dla czystego sygnału, bo struct{} nie zajmuje pamięci.

Kanały buforowane

make(chan T, n) daje kanałowi miejsce na n wartości. Wysyłanie działa bez odbiorcy, dopóki bufor się nie zapełni; odbieranie działa, dopóki bufor nie jest pusty. Wartości wychodzą w tej kolejności, w jakiej weszły.

package main

import "fmt"

func main() {
	ch := make(chan int, 3)

	ch <- 10
	ch <- 20
	fmt.Println("len:", len(ch), "cap:", cap(ch))

	ch <- 30 // buffer now full; a fourth send here would block forever

	fmt.Println(<-ch, <-ch, <-ch)
	fmt.Println("len:", len(ch))
}

Bufor oddziela nadawcę od odbiorcy, więc krótkie serie nie wstrzymują nadawcy. Nie naprawia jednak producenta, który jest stale szybszy od konsumenta; jedynie opóźnia moment, w którym nadawca się zablokuje. Rozmiar bufora dobieraj z konkretnego powodu (liczba nadawców, znany rozmiar partii), a nie po to, żeby zniknął deadlock.

len(ch) to migawka. Zanim na jej podstawie coś zrobisz, inna gorutyna mogła już ją zmienić, więc nie używaj jej do sprawdzania, czy wysłanie zablokuje. Do tego służy select z gałęzią default.

Close i range

close(ch) informuje odbiorców, że więcej wartości nie zostanie wysłanych. Po zamknięciu:

• wartości, które są już w buforze, nadal zostają dostarczone,
• potem każdy odbiór od razu zwraca wartość zerową,
• v, ok := <-ch zwraca ok == false,
• for v := range ch się kończy.

package main

import "fmt"

func produce(n int) <-chan int {
	ch := make(chan int)
	go func() {
		defer close(ch) // the sender closes, when it is done
		for i := range n {
			ch <- i * i
		}
	}()
	return ch
}

func main() {
	for v := range produce(5) {
		fmt.Println(v)
	}

	ch := make(chan string, 1)
	ch <- "last"
	close(ch)
	v, ok := <-ch
	fmt.Printf("%q %v\n", v, ok)
	v, ok = <-ch
	fmt.Printf("%q %v\n", v, ok)
}

Pierwszy odbiór po close nadal dostaje "last" z ok == true. Drugi dostaje wartość zerową "" i false.

Reguły, które kończą się panic:

• wysłanie do zamkniętego kanału wywołuje panic z komunikatem send on closed channel,
• zamknięcie już zamkniętego kanału wywołuje panic,
• zamknięcie kanału nil wywołuje panic.

Dlatego zamyka tylko strona wysyłająca i tylko raz. Przy kilku nadawcach żaden z nich nie wie, kiedy pozostali skończyli; niech osobna gorutyna poczeka na wszystkich nadawców (przez sync.WaitGroup) i zamknie kanał, gdy Wait zwróci. Zamykanie jest potrzebne tylko wtedy, gdy odbiorca czeka na koniec. Niezamknięty kanał, do którego nikt się nie odwołuje, jest sprzątany przez garbage collector jak każda inna wartość.

Typy kierunkowe

Funkcja może zadeklarować, że na kanale tylko wysyła albo tylko odbiera. Kompilator odrzuci wtedy drugą operację.

chan T konwertuje się niejawnie na każdy z typów ograniczonych, gdy przekazujesz go do funkcji. Dlatego produce powyżej może zwrócić <-chan int: wywołujący mogą po nim iterować przez range, ale nie mogą do niego wysyłać ani go zamknąć. Używaj typów kierunkowych w każdym parametrze funkcji, w którym mają sens. Dokumentują własność i zamieniają niewłaściwe użycie w błąd kompilacji.

Deadlock

Jeśli każda gorutyna jest zablokowana i nic nie może żadnej z nich obudzić, środowisko uruchomieniowe przerywa program:

package main

import "fmt"

func main() {
	ch := make(chan int)
	ch <- 1 // no other goroutine exists to receive
	fmt.Println(<-ch)
}

fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan send]:
main.main()
	/tmp/main.go:7 +0x38
exit status 2

Zrzut gorutyn pokazuje, która operacja utknęła (tu chan send, gdzie indziej chan receive, sync.WaitGroup.Wait, select). Typowe przyczyny:

• wysłanie do kanału niebuforowanego, gdy nie działa żaden odbiorca,
• range po kanale, który nigdy nie zostaje zamknięty,
• licznik WaitGroup, który nigdy nie spada do zera,
• dwie gorutyny, z których każda czeka na drugą.

Środowisko uruchomieniowe wykrywa tylko sytuację, w której utknęły wszystkie gorutyny. W serwerze z innymi żywymi gorutynami (listener HTTP, ticker) ten sam błąd niczego nie wywraca; zablokowana gorutyna po prostu wycieka.

Kanały nil

Wysyłanie i odbieranie na kanale nil blokuje w nieskończoność. Brzmi to bezużytecznie, ale to standardowa sztuczka w select: ustawienie zmiennej kanału na nil wyłącza jej gałąź. Ten kod łączy dwa kanały i przestaje nasłuchiwać każdego z nich, gdy zostanie zamknięty:

package main

import "fmt"

func main() {
	a := make(chan int)
	b := make(chan int)
	go func() { defer close(a); a <- 1; a <- 2 }()
	go func() { defer close(b); b <- 10 }()

	sum := 0
	for a != nil || b != nil {
		select {
		case v, ok := <-a:
			if !ok {
				a = nil // a is finished: never select it again
				continue
			}
			sum += v
		case v, ok := <-b:
			if !ok {
				b = nil
				continue
			}
			sum += v
		}
	}
	fmt.Println("sum:", sum)
}

Bez przypisań nil zamknięty kanał jest zawsze gotowy i pętla kręciłaby się w kółko na wartościach zerowych.

Pipeline

Kanały składają się w pipeline'y: każdy etap to gorutyna, która odbiera z jednego kanału i wysyła do następnego, a zamyka swoje wyjście, gdy skończy się jej wejście.

package main

import "fmt"

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

func square(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			out <- n * n
		}
	}()
	return out
}

func main() {
	total := 0
	for v := range square(square(generate(1, 2, 3))) {
		fmt.Println(v)
		total += v
	}
	fmt.Println("total:", total)
}

Każdy etap działa współbieżnie, a kolejność wyników jest deterministyczna (1, 16, 81), bo każdy etap to jedna gorutyna zachowująca kolejność. Zamknięcia przechodzą kaskadowo: generate zamyka kanał, co kończy range w square, które zamyka swoje wyjście, i tak dalej aż do main.

Słaby punkt tego pipeline'u: gdyby main przestało czytać wcześniej, etapy zablokowałyby się na wysyłaniu na zawsze. Prawdziwe pipeline'y przyjmują context.Context albo kanał done i obok każdego wysłania robią na nim select.

Kanał czy mutex

Kanały służą do przekazywania własności danych i sygnalizowania zdarzeń. sync.Mutex jest prostszy do ochrony wspólnego stanu, który wiele gorutyn czyta i zmienia w miejscu, jak cache albo licznik. Struktura z mutexem w środku jest często czytelniejsza niż gorutyna, która posiada stan i obsługuje żądania przez kanały. Wybierz to, co daje krótszy kod i oczywistą własność.

Ściągawka