Bagian 3: Channel dan Cancellation

Distribusi job dari Bagian 2 punya satu kelemahan senyap: for range pool.Results() berhenti hanya jika channel hasil ditutup, tetapi belum ada yang bertugas menutupnya. Bagian ini melengkapi mekanisme penutupan itu — pengawas yang menunggu worker habis — lalu membuka jalur darurat: pembatalan lewat context yang menghentikan seluruh pool di tengah pekerjaan tanpa meninggalkan satu goroutine pun.

Pengawas Penutup Channel Hasil #

Dua anggota pool bekerja berpasangan di pool.go:

go
// awasi menunggu worker habis lalu menutup results dan melepas context.
// Goroutine inilah yang mencegah bocor: selesai → channel ditutup.
func (p *Pool) awasi() {
        p.wg.Wait()
        close(p.results)
        p.cancel()
}

awasi dijalankan tepat satu kali oleh Start pada Bagian 2. Ketika penghitung wg mencapai nol — artinya setiap worker telah memanggil Done — pengawas menutup results dan membatalkan context turunan. Penutupan channel itulah yang mengakhiri for range di sisi pembaca; tanpa pengawas, pembaca menunggu selamanya meski semua pekerjaan telah usai. Pemanggilan cancel terakhir melepas resource context yang dialokasikan Start. Sisi pembaca hasil tinggal menunggu channel itu:

go
// Results mengembalikan channel hasil. Channel ini ditutup otomatis
// setelah seluruh worker keluar, sehingga `for range` berakhir pasti.
func (p *Pool) Results() <-chan Result {
        return p.results
}

Shutdown dengan sync.Once #

Jalur berhenti normal dimulai dari penutupan channel job:

go
// Shutdown menutup channel job. Worker menghabiskan sisa job yang sudah
// masuk, keluar, lalu pengawas menutup channel hasil. Shutdown aman
// dipanggil berapa pun kali — hanya pemanggil pertama yang menutup.
func (p *Pool) Shutdown() {
        p.sekali.Do(func() {
                close(p.jobs)
        })
}

Shutdown hanya menutup jobs dan seketika kembali — ia tidak menunggu. Worker yang sedang memegang job tetap menyelesaikannya; worker yang sedang menunggu job melihat channel tertutup dan keluar; pengawas kemudian menutup hasil. Penutupan channel pada Go berarti “tidak ada kiriman lagi”, bukan “buang yang tersisa”, sehingga tidak ada satu job yang sudah masuk akan hilang.

sync.Once menjaga penutupan terjadi satu kali. Menutup channel yang sudah tertutup membuat program panic, dan panggilan Shutdown kedua dari pemanggil berbeda adalah skenario yang wajar pada program nyata — misalnya satu dari handler sinyal dan satu dari alur utama.

Pembatalan Lewat Context #

Shutdown adalah jalur lembut: semua job yang masuk diselesaikan. Context adalah jalur darurat. Start membuat context turunan dari yang diberikan pemanggil, dan tombol Done()-nya dibaca di tiga tempat sekaligus:

Lokasi	Perilaku saat dibatalkan
`Submit`	Job dibuang; pengirim berhenti mengirim
Worker, sebelum ambil job	Worker keluar seketika
Worker, saat kirim hasil	Worker keluar tanpa menggantung pada pembaca

Rantai ini membuat pembatalan merambat ke seluruh penjuru pool sekali cancel() dipanggil — baik oleh pemanggil melalui context induk, maupun oleh awasi setelah penutupan normal. Siklus hidup lengkapnya tergambar berikut:

flowchart TD A["Start: n worker + pengawas"] --> B["Submit mengirim job"] B --> C{"worker mengambil job"} C --> D["fn memproses job"] D --> E["kirim Result"] E --> C C -- "channel jobs tertutup" --> F["worker keluar, wg nol"] A -.-> G["pengawas: wg.Wait"] G -- "nol" --> H["tutup results, cancel context"] H --> I["pembaca for range berhenti"] B -.-> J{"context dibatalkan?"} J -- "ya" --> K["Submit buang job, worker keluar"] K --> G

Demo Pembatalan Sebelum Pengiriman #

Program uji berikut menyalakan pool, membatalkan context lebih dulu, baru mengirim job. Seluruh Submit mendapati jalur masuk sudah tertutup:

go
package main

import (
        "context"
        "fmt"
        "runtime"
)

func main() {
        sebelum := runtime.NumGoroutine()

        ctx, cancel := context.WithCancel(context.Background())
        pool := New(3, prosesJob)
        pool.Start(ctx)

        // Batalkan context sebelum satu job pun dikirim.
        cancel()

        go func() {
                for i := 1; i <= 6; i++ {
                        pool.Submit(Job{ID: i, Payload: fmt.Sprintf("tugas-%02d", i)})
                }
        }()

        jumlah := 0
        for range pool.Results() {
                jumlah++
        }

        fmt.Println("pool dibatalkan sebelum job dikirim")
        fmt.Printf("jumlah hasil: %d\n", jumlah)
        fmt.Println("channel hasil tertutup — pool berhenti bersih")

        tungguGoroutine(sebelum)
        sesudah := runtime.NumGoroutine()
        fmt.Printf("goroutine sebelum=%d, sesudah=%d\n", sebelum, sesudah)
}

func prosesJob(job Job) (string, error) {
        return "TUGAS-" + job.Payload, nil
}

// tungguGoroutine menunggu jumlah goroutine kembali ke nilai dasar.
func tungguGoroutine(dasar int) {
        for i := 0; i < 1000 && runtime.NumGoroutine() > dasar; i++ {
                runtime.Gosched()
        }
}

Output yang Dihasilkan:

text
pool dibatalkan sebelum job dikirim
jumlah hasil: 0
channel hasil tertutup — pool berhenti bersih
goroutine sebelum=1, sesudah=1

Empat baris itu memuat seluruh janji pembatalan: Submit keenam kali pun tidak menggantung (job dibuang), channel hasil tetap ditutup pengawas (loop for range berakhir), dan hitungan goroutine kembali seperti semula — tidak ada pekerja yang terjebak. Program ini dijalankan sebagai module uji terpisah yang menyalin worker.go dan pool.go proyek apa adanya; bukti yang sama diuji otomatis oleh TestPoolPembatalanContext pada Bagian 4.

Rujukan Materi #

Context — pola WithCancel, Done, dan Err secara utuh.
Mencegah Goroutine Leak — galat klasik goroutine yang menunggu channel tak berpembaca.
Select — pemilihan banyak operasi channel sekaligus sebagai dasar jalur darurat.
Dokumentasi context — referensi resmi paket context.

Channel menutup dengan sopan dan pembatalan bekerja bersih. Satu tugas terakhir menutup proyek: membuktikan semuanya dengan race detector dan pengujian yang menekan jalur-jalur berbahaya.

Berkas Sumber

Berkas pendukung halaman ini dapat diunduh dan dijalankan langsung sesuai penjelasan di atas.

go
package main

import (
	"context"
	"fmt"
	"runtime"
	"slices"
	"strings"
)

const (
	jumlahWorker = 3
	jumlahJob    = 10
)

// prosesJob adalah pekerjaan yang dijalankan setiap worker: mengubah
// payload menjadi huruf besar. Pada kasus nyata fungsi ini bisa berisi
// kalkulasi berat, pemanggilan API, atau pembacaan berkas.
func prosesJob(job Job) (string, error) {
	return strings.ToUpper(job.Payload), nil
}

func main() {
	fmt.Println("=== Demo Worker Pool ===")
	fmt.Printf("pool: %d worker, %d job\n", jumlahWorker, jumlahJob)

	sebelum := runtime.NumGoroutine()

	pool := New(jumlahWorker, prosesJob)
	pool.Start(context.Background())

	// Pengiriman job berjalan di goroutine terpisah karena kedua channel
	// pool tanpa buffer: pengirim dan pembaca hasil harus aktif bersamaan,
	// jika tidak keduanya saling menunggu selamanya (deadlock).
	kirimSelesai := make(chan struct{})
	go func() {
		defer close(kirimSelesai)
		for i := 1; i <= jumlahJob; i++ {
			pool.Submit(Job{ID: i, Payload: fmt.Sprintf("tugas-%02d", i)})
		}
		pool.Shutdown()
	}()

	// Kumpulkan seluruh hasil; perulangan berhenti saat pengawas pool
	// menutup channel hasil.
	hasil := []Result{}
	for r := range pool.Results() {
		hasil = append(hasil, r)
	}
	<-kirimSelesai

	// Urutan kedatangan hasil dari beberapa goroutine tidak menentu,
	// jadi hasil dikumpulkan dulu lalu diurutkan sebelum dicetak agar
	// output selalu sama setiap eksekusi.
	slices.SortFunc(hasil, func(a, b Result) int {
		return a.JobID - b.JobID
	})

	fmt.Println("hasil (terurut berdasarkan job):")
	for _, h := range hasil {
		fmt.Printf("job %02d → %s\n", h.JobID, h.Output)
	}

	// Keluarnya goroutine tidak bisa diamati lewat channel mana pun,
	// jadi beri penjadwal kesempatan menjalankan epilogue goroutine
	// yang sudah menyelesaikan tugasnya sebelum menghitung ulang.
	tungguGoroutine(sebelum)

	sesudah := runtime.NumGoroutine()
	fmt.Printf("goroutine sebelum=%d, sesudah=%d\n", sebelum, sesudah)
}

// tungguGoroutine menunggu jumlah goroutine kembali ke nilai dasar
// dengan batas putaran tetap. Ini sinkronisasi eksplisit untuk demo;
// pengujian memakai fungsi yang sama.
func tungguGoroutine(dasar int) {
	for i := 0; i < 1000 && runtime.NumGoroutine() > dasar; i++ {
		runtime.Gosched()
	}
}

go
package main

import (
	"context"
	"fmt"
	"runtime"
	"slices"
	"strings"
	"testing"
)

func fungsiUji(job Job) (string, error) {
	return strings.ToUpper(job.Payload), nil
}

func TestPoolHasilLengkapDanTerurut(t *testing.T) {
	pool := New(3, fungsiUji)
	pool.Start(context.Background())

	kirimSelesai := make(chan struct{})
	go func() {
		defer close(kirimSelesai)
		for i := 1; i <= 10; i++ {
			pool.Submit(Job{ID: i, Payload: fmt.Sprintf("tugas-%02d", i)})
		}
		pool.Shutdown()
	}()

	hasil := []Result{}
	for r := range pool.Results() {
		hasil = append(hasil, r)
	}
	<-kirimSelesai

	slices.SortFunc(hasil, func(a, b Result) int {
		return a.JobID - b.JobID
	})

	if len(hasil) != 10 {
		t.Fatalf("jumlah hasil = %d, ingin 10", len(hasil))
	}
	for i, h := range hasil {
		want := fmt.Sprintf("TUGAS-%02d", i+1)
		if h.JobID != i+1 || h.Output != want || h.Err != nil {
			t.Errorf("hasil[%d] = %+v, ingin JobID %d Output %q", i, h, i+1, want)
		}
	}
}

func TestPoolShutdownBersihTanpaBocor(t *testing.T) {
	dasar := runtime.NumGoroutine()

	pool := New(3, fungsiUji)
	pool.Start(context.Background())

	go func() {
		for i := 1; i <= 6; i++ {
			pool.Submit(Job{ID: i, Payload: fmt.Sprintf("tugas-%02d", i)})
		}
		pool.Shutdown()
	}()

	jumlah := 0
	for range pool.Results() {
		jumlah++
	}

	if jumlah != 6 {
		t.Fatalf("jumlah hasil = %d, ingin 6", jumlah)
	}

	// Shutdown dipanggil kedua kali untuk memastikan pool tetap aman.
	pool.Shutdown()

	tungguGoroutine(dasar)
	if sesudah := runtime.NumGoroutine(); sesudah > dasar {
		t.Errorf("goroutine tersisa: sebelum=%d sesudah=%d", dasar, sesudah)
	}
}

func TestPoolPembatalanContext(t *testing.T) {
	dasar := runtime.NumGoroutine()

	ctx, cancel := context.WithCancel(context.Background())
	pool := New(3, fungsiUji)
	pool.Start(ctx)

	kirimSelesai := make(chan struct{})
	go func() {
		defer close(kirimSelesai)
		for i := 1; i <= 6; i++ {
			pool.Submit(Job{ID: i, Payload: fmt.Sprintf("tugas-%02d", i)})
		}
		// Batalkan di tengah jalan: sebagian job mungkin sudah diproses,
		// sebagian dibuang Submit. Yang diuji adalah pool berhenti tanpa
		// menggantung dan tanpa goroutine tersisa.
		cancel()
	}()

	hasil := []Result{}
	for r := range pool.Results() {
		hasil = append(hasil, r)
	}
	<-kirimSelesai

	if len(hasil) > 6 {
		t.Errorf("jumlah hasil = %d, tidak boleh melebihi 6", len(hasil))
	}

	tungguGoroutine(dasar)
	if sesudah := runtime.NumGoroutine(); sesudah > dasar {
		t.Errorf("goroutine tersisa setelah pembatalan: sebelum=%d sesudah=%d", dasar, sesudah)
	}
}

func TestPoolJobGagalDilaporkan(t *testing.T) {
	fn := func(job Job) (string, error) {
		if job.Payload == "" {
			return "", fmt.Errorf("job %d: payload kosong", job.ID)
		}
		return strings.ToUpper(job.Payload), nil
	}

	pool := New(2, fn)
	pool.Start(context.Background())

	go func() {
		pool.Submit(Job{ID: 1, Payload: "tugas-01"})
		pool.Submit(Job{ID: 2, Payload: ""})
		pool.Shutdown()
	}()

	hasil := []Result{}
	for r := range pool.Results() {
		hasil = append(hasil, r)
	}

	slices.SortFunc(hasil, func(a, b Result) int {
		return a.JobID - b.JobID
	})

	if len(hasil) != 2 {
		t.Fatalf("jumlah hasil = %d, ingin 2", len(hasil))
	}
	if hasil[0].Err != nil || hasil[0].Output != "TUGAS-01" {
		t.Errorf("hasil[0] = %+v, ingin sukses TUGAS-01", hasil[0])
	}
	if hasil[1].Err == nil {
		t.Errorf("hasil[1].Err = nil, ingin galat payload kosong")
	}
}

go
package main

import (
	"context"
	"sync"
)

// Pool membatasi jumlah worker yang memproses job secara bersamaan.
// Job masuk lewat channel jobs, keluar sebagai Result lewat channel
// results; keduanya tanpa buffer sehingga pengirim, worker, dan pembaca
// hasil bertemu langsung tanpa antrean tersembunyi.
type Pool struct {
	jumlahWorker int
	fn           func(Job) (string, error)
	jobs         chan Job
	results      chan Result
	ctx          context.Context
	cancel       context.CancelFunc
	wg           sync.WaitGroup
	sekali       sync.Once
}

// New membuat pool ber-jumlahWorker worker yang memproses job dengan
// fungsi fn. Pool belum berjalan sampai Start dipanggil.
func New(jumlahWorker int, fn func(Job) (string, error)) *Pool {
	return &Pool{
		jumlahWorker: jumlahWorker,
		fn:           fn,
		jobs:         make(chan Job),
		results:      make(chan Result),
	}
}

// Start menjalankan seluruh worker plus satu goroutine pengawas.
// Context yang dibatalkan menghentikan semua worker di tengah pekerjaan;
// Shutdown tetap perlu dipanggil untuk jalur berhenti yang normal.
func (p *Pool) Start(ctx context.Context) {
	p.ctx, p.cancel = context.WithCancel(ctx)
	for i := 0; i < p.jumlahWorker; i++ {
		p.wg.Add(1)
		go worker(p.ctx, p.jobs, p.results, p.fn, &p.wg)
	}
	// Pengawas menutup channel hasil setelah semua worker keluar.
	// Pembaca hasil berhenti bereputar tepat pada saat itu — tanpa
	// pengawas, pembaca tidak pernah tahu pool sudah selesai.
	go p.awasi()
}

// awasi menunggu worker habis lalu menutup results dan melepas context.
// Goroutine inilah yang mencegah bocor: selesai → channel ditutup.
func (p *Pool) awasi() {
	p.wg.Wait()
	close(p.results)
	p.cancel()
}

// Submit mengirim satu job ke worker yang tersedia dan memblokir saat
// semua worker sedang sibuk (backpressure). Job dibuang begitu context
// pool sudah dibatalkan, jadi pemanggil harus berhenti mengirim setelah
// pembatalan terjadi.
func (p *Pool) Submit(job Job) {
	select {
	case p.jobs <- job:
	case <-p.ctx.Done():
		// Pool sudah berhenti; job tidak dikirim.
	}
}

// Results mengembalikan channel hasil. Channel ini ditutup otomatis
// setelah seluruh worker keluar, sehingga `for range` berakhir pasti.
func (p *Pool) Results() <-chan Result {
	return p.results
}

// Shutdown menutup channel job. Worker menghabiskan sisa job yang sudah
// masuk, keluar, lalu pengawas menutup channel hasil. Shutdown aman
// dipanggil berapa pun kali — hanya pemanggil pertama yang menutup.
func (p *Pool) Shutdown() {
	p.sekali.Do(func() {
		close(p.jobs)
	})
}

go
package main

import (
	"context"
	"errors"
	"strings"
	"sync"
	"testing"
)

func TestWorkerMemprosesJob(t *testing.T) {
	jobs := make(chan Job, 1)
	results := make(chan Result, 1)
	var wg sync.WaitGroup

	fn := func(job Job) (string, error) {
		return strings.ToUpper(job.Payload), nil
	}

	wg.Add(1)
	go worker(context.Background(), jobs, results, fn, &wg)

	jobs <- Job{ID: 7, Payload: "tugas-07"}
	close(jobs)

	res := <-results
	wg.Wait()

	if res.JobID != 7 {
		t.Errorf("JobID hasil = %d, ingin 7", res.JobID)
	}
	if res.Output != "TUGAS-07" {
		t.Errorf("Output hasil = %q, ingin TUGAS-07", res.Output)
	}
	if res.Err != nil {
		t.Errorf("Err hasil = %v, ingin nil", res.Err)
	}
}

func TestWorkerJobGagal(t *testing.T) {
	jobs := make(chan Job, 1)
	results := make(chan Result, 1)
	var wg sync.WaitGroup

	galatUji := errors.New("payload kosong")
	fn := func(job Job) (string, error) {
		return "", galatUji
	}

	wg.Add(1)
	go worker(context.Background(), jobs, results, fn, &wg)

	jobs <- Job{ID: 1, Payload: ""}
	close(jobs)

	res := <-results
	wg.Wait()

	if res.Err == nil {
		t.Fatal("Err hasil = nil, ingin galat")
	}
	if !errors.Is(res.Err, galatUji) {
		t.Errorf("Err hasil = %v, ingin galat uji", res.Err)
	}
	if res.Output != "" {
		t.Errorf("Output hasil = %q, ingin string kosong saat gagal", res.Output)
	}
}

func TestWorkerBerhentiSaatContextDibatalkan(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	// Batalkan sebelum worker sempat mengambil job apa pun.
	cancel()

	jobs := make(chan Job, 1)
	results := make(chan Result, 1)
	var wg sync.WaitGroup

	fn := func(job Job) (string, error) {
		return strings.ToUpper(job.Payload), nil
	}

	wg.Add(1)
	go worker(ctx, jobs, results, fn, &wg)

	jobs <- Job{ID: 1, Payload: "tugas-01"}
	close(jobs)
	wg.Wait()

	select {
	case res := <-results:
		t.Errorf("worker memproses job setelah pembatalan: %+v", res)
	default:
		// Tidak ada hasil: worker keluar tanpa memproses apa pun.
	}
}

func TestWorkerChannelTertutupMenghentikanWorker(t *testing.T) {
	jobs := make(chan Job, 2)
	results := make(chan Result, 2)
	var wg sync.WaitGroup

	fn := func(job Job) (string, error) {
		return strings.ToUpper(job.Payload), nil
	}

	wg.Add(1)
	go worker(context.Background(), jobs, results, fn, &wg)

	jobs <- Job{ID: 1, Payload: "satu"}
	jobs <- Job{ID: 2, Payload: "dua"}
	close(jobs)
	wg.Wait()

	if len(results) != 2 {
		t.Fatalf("jumlah hasil = %d, ingin 2", len(results))
	}
}

go
package main

import (
	"context"
	"sync"
)

// Job adalah satu unit pekerjaan yang dikirim ke pool.
// ID menjaga identitas job sehingga hasil bisa diurutkan kembali,
// Payload membawa data yang diproses fungsi kerja.
type Job struct {
	ID      int
	Payload string
}

// Result adalah hasil pemrosesan satu Job. Err tidak kosong ketika
// fungsi kerja gagal memproses Payload.
type Result struct {
	JobID  int
	Output string
	Err    error
}

// worker mengonsumsi Job dari channel jobs sampai channel ditutup atau
// context dibatalkan, lalu menuliskan Result ke channel results.
// wg.Done via defer menjamin pool tahu worker ini sudah keluar.
func worker(ctx context.Context, jobs <-chan Job, results chan<- Result, fn func(Job) (string, error), wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		// Periksa pembatalan lebih dulu agar worker berhenti seketika
		// ketika context sudah dibatalkan sebelum job berikutnya diambil.
		if ctx.Err() != nil {
			return
		}
		select {
		case <-ctx.Done():
			return
		case job, ok := <-jobs:
			if !ok {
				return
			}
			output, err := fn(job)
			// Pengiriman hasil juga harus siap kalah dari pembatalan;
			// tanpa case ini worker bisa menggantung selamanya menunggu
			// pembaca hasil yang sudah tidak ada.
			select {
			case <-ctx.Done():
				return
			case results <- Result{JobID: job.ID, Output: output, Err: err}:
			}
		}
	}
}

Bagian 3: Channel dan Cancellation

Pengawas Penutup Channel Hasil #

Shutdown dengan sync.Once #

Pembatalan Lewat Context #

Demo Pembatalan Sebelum Pengiriman #

Rujukan Materi #

Berkas Sumber

main.go

pool_test.go

pool.go

worker_test.go

worker.go