KAL
Menu

Pola Konkurensi

Bentuk kanonik fan-out/fan-in, pipeline, dan worker pool.

Daftar isi

Primitif goroutine dan channel tersusun menjadi beberapa bentuk yang berulang di hampir semua program Go serentak. Topik ini membahas tiga bentuk kanonik tersebut, yaitu pipeline sebagai rangkaian tahap yang dihubungkan channel, fan-out dan fan-in sebagai pembagian serta penggabungan pekerjaan, dan worker pool sebagai sekelompok pekerja berukuran tetap. Setiap bentuk disertai program utuh yang keluarannya sama pada setiap eksekusi. Setelah membacanya, ketiga bentuk dapat dipilih dan disusun tanpa menebak perilaku penutupan channelnya.

Pipeline#

Sebuah pipeline membagi pemrosesan menjadi tahap-tahap yang dihubungkan channel, dan setiap tahap menerima nilai dari tahap sebelumnya lalu mengirim hasilnya ke tahap berikutnya. Disiplin penutupan menjadi inti bentuk ini, yaitu pengirim menutup channel begitu selesai mengirim, sedangkan penerima membaca dengan range hingga channel tertutup. Program berikut menyusun dua tahap, yaitu sumber yang mengirim bilangan 1 sampai 5 dan tahap kuadrat yang memprosesnya.

go pipeline-kuadrat.go 
package main

import "fmt"

// sumber mengirim bilangan 1 sampai 5 lalu menutup channelnya.
func sumber(berangkat chan<- int) {
	for n := 1; n <= 5; n++ {
		berangkat <- n
	}
	close(berangkat)
}

// kuadrat menerima angka, mengkuadratkannya, lalu meneruskan hasilnya.
func kuadrat(masuk <-chan int, keluar chan<- int) {
	for n := range masuk {
		keluar <- n * n
	}
	close(keluar)
}

func main() {
	angka := make(chan int)
	hasil := make(chan int)

	go sumber(angka)
	go kuadrat(angka, hasil)

	for nilai := range hasil {
		fmt.Println(nilai)
	}
}
text Keluaran 
1
4
9
16
25

Nilai mengalir berurutan 1, 4, 9, 16, dan 25 karena pipeline mempertahankan urutan antarnilai pada satu jalur. Fungsi sumber menutup berangkat setelah lima nilai terkirim, sehingga perulangan range di tahap kuadrat berakhir dengan sendirinya dan kuadrat ikut menutup channel keluar. Penutupan berantai semacam ini membuat tahap tambahan dapat disisipkan tanpa menyentuh tahap lain, sekaligus mencegah goroutine menggantung karena menunggu nilai yang tidak pernah datang.

Setiap tahap pada pipeline dapat ditulis sebagai fungsi yang menerima channel masukan dan mengembalikan channel keluaran, sehingga tahap-tahap itu mudah dirangkai. Channel dengan buffer juga dapat disisipkan di antara dua tahap untuk memberi ruang bernapas, yaitu tahap hulu boleh berjalan lebih dahulu sebelum tahap hilir siap menerima. Kapasitas buffer itu sekaligus menjadi batas kerjanya, karena tahap hulu kembali terblokir ketika ruangnya penuh.

bash 
mkdir -p /tmp/go-pola-konkurensi
cp pipeline-kuadrat.go worker-pool.go /tmp/go-pola-konkurensi/
cd /tmp/go-pola-konkurensi
go run pipeline-kuadrat.go
text Keluaran 
1
4
9
16
25

Fan-out dan fan-in#

Satu tahap yang lambat dapat diduplikasi agar banyak pekerja menarik pekerjaan dari channel yang sama. Penduplikasi ini disebut fan-out, yaitu pembagian pekerjaan kepada beberapa goroutine pembaca, sedangkan fan-in adalah penggabungan beberapa channel hasil menjadi satu channel penerima. Kerangka penggabungan dua channel berbentuk berikut.

go Kerangka fan-in dua channel menjadi satu 
// tidak-dijalankan
func gabung(a, b <-chan int) <-chan int {
	keluar := make(chan int)
	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		for v := range a {
			keluar <- v
		}
	}()
	go func() {
		defer wg.Done()
		for v := range b {
			keluar <- v
		}
	}()
	go func() {
		wg.Wait()
		close(keluar)
	}()
	return keluar
}

Fungsi gabung meluncurkan satu goroutine pembaca untuk setiap channel masukan, lalu goroutine ketiga menutup channel keluaran setelah Wait kembali. Pola ini memadukan WaitGroup untuk menunggu seluruh pembaca selesai dengan penutupan channel sebagai sinyal berakhirnya aliran. Urutan nilai antarchannel tidak dijamin, sehingga penerima yang memerlukan urutan tertentu harus mengurutkan hasilnya sendiri.

Jumlah pembaca pada fan-out tidak harus dua, dan penggabungannya dapat diperumum dengan menerima slice channel sebagai masukan lalu meluncurkan satu goroutine pembaca untuk setiap elemennya. Jumlah pekerja yang paling tepat mengikuti sumber daya yang benar-benar dibatasi, misalnya jumlah inti pemroses untuk pekerjaan hitung murni atau jumlah koneksi basis data untuk pekerjaan yang menunggu jawaban jaringan.

Worker pool#

Worker pool adalah bentuk lengkap dari fan-out dan fan-in, yaitu sejumlah pekerja berukuran tetap melayani sejumlah beban yang mungkin jauh lebih besar. Ukuran pekerja mengatur beban sistem, misalnya jumlah koneksi basis data yang boleh terbuka, sedangkan jumlah bebannya bebas berubah. Program berikut melayani delapan beban dengan tiga pekerja, mengumpulkan hasilnya lewat satu channel, lalu mencetaknya terurut setelah semua pekerja selesai.

go worker-pool.go 
package main

import (
	"fmt"
	"sort"
	"sync"
)

// hasil menyimpan pasangan beban dan nilai pemrosesannya agar urutan
// cetak dapat diurutkan setelah semua pekerja selesai.
type hasil struct {
	beban int
	nilai int
}

// pekerja menarik beban dari channel tugas sampai channelnya ditutup.
func pekerja(tugas <-chan int, keluar chan<- hasil, wg *sync.WaitGroup) {
	defer wg.Done()
	for beban := range tugas {
		keluar <- hasil{beban: beban, nilai: beban * beban}
	}
}

func main() {
	tugas := make(chan int)
	keluar := make(chan hasil)

	// fan-out: tiga pekerja berbagi satu channel tugas
	var wg sync.WaitGroup
	for i := 0; i < 3; i++ {
		wg.Add(1)
		go pekerja(tugas, keluar, &wg)
	}

	// pengirim tugas: delapan beban, lalu channel ditutup
	go func() {
		for beban := 1; beban <= 8; beban++ {
			tugas <- beban
		}
		close(tugas)
	}()

	// fan-in: tunggu seluruh pekerja selesai, lalu tutup channel hasil
	go func() {
		wg.Wait()
		close(keluar)
	}()

	var daftar []hasil
	for h := range keluar {
		daftar = append(daftar, h)
	}
	sort.Slice(daftar, func(i, j int) bool {
		return daftar[i].beban < daftar[j].beban
	})

	for _, h := range daftar {
		fmt.Printf("beban %d dikerjakan menjadi %d\n", h.beban, h.nilai)
	}
}
text Keluaran 
beban 1 dikerjakan menjadi 1
beban 2 dikerjakan menjadi 4
beban 3 dikerjakan menjadi 9
beban 4 dikerjakan menjadi 16
beban 5 dikerjakan menjadi 25
beban 6 dikerjakan menjadi 36
beban 7 dikerjakan menjadi 49
beban 8 dikerjakan menjadi 64

Tiga pekerja menarik beban dari channel tugas tanpa perlu mengetahui beban mana yang dikerjakan oleh siapa, karena channel sendiri yang mengatur pembagiannya. Goroutine penghitung menutup tugas setelah delapan beban terkirim, sehingga perulangan range pada setiap pekerja berakhir. Goroutine terakhir menutup channel keluar setelah Wait kembali, dan fungsi main mengumpulkan seluruh hasil sebelum mengurutkannya menurut beban. Pengurutan ini diperlukan karena waktu penyelesaian antarbeban tidak tertib, padahal keluaran yang dapat dipercaya harus sama pada setiap eksekusi.

Alternatif pengumpulan hasil adalah channel berbuffer sebesar jumlah beban, sehingga pekerja tidak pernah terblokir saat mengirim dan goroutine penutupnya tidak perlu menunggu tahap penerimaan selesai. Bentuk ini memperpendek rantai sinkronisasi, tetapi membutuhkan taksiran jumlah hasil sejak awal, sedangkan channel tanpa buffer pada program di atas bekerja untuk jumlah beban apa pun.

bash 
cd /tmp/go-pola-konkurensi
go run worker-pool.go
text Keluaran 
beban 1 dikerjakan menjadi 1
beban 2 dikerjakan menjadi 4
beban 3 dikerjakan menjadi 9
beban 4 dikerjakan menjadi 16
beban 5 dikerjakan menjadi 25
beban 6 dikerjakan menjadi 36
beban 7 dikerjakan menjadi 49
beban 8 dikerjakan menjadi 64

Kesalahan umum#

  • Mengirim ke channel yang sudah ditutup. Pengiriman setelah close memicu panic dan menghentikan program. Perbaikannya, hak menutup channel melekat pada pengirimnya, dan penutupan dilakukan setelah seluruh pengiriman selesai.
  • Menutup channel dua kali. Pemanggilan close kedua juga memicu panic. Perbaikannya, serahkan penutupan kepada satu goroutine yang jelas, misalnya goroutine yang menunggu dengan WaitGroup.
  • Membiarkan channel tanpa penerima. Goroutine yang terblokir mengirim ke channel yang tidak dibaca tidak pernah berhenti dan menumpuk sepanjang umur program. Perbaikannya, pastikan setiap channel yang dikirim juga dibaca sampai ditutup atau dibatalkan.

Ringkasan#

  • Pipeline menghubungkan tahap lewat channel dengan disiplin pengirim menutup dan penerima membaca hingga tertutup.
  • Fan-out menduplikasi pembaca satu channel, sedangkan fan-in menggabungkan beberapa channel hasil dengan bantuan WaitGroup.
  • Worker pool menyempurnakan bentuk tersebut dengan pekerja berukuran tetap dan hasil yang dikumpulkan lalu diurutkan.
  • Hak menutup channel melekat pada pengirim, dan penutupan dilakukan tepat satu kali.

Referensi#