Proyek: Concurrent Worker Pool

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()
	}
}

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")
	}
}

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)
	})
}

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))
	}
}

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}:
			}
		}
	}
}