Goroutine ve Channel: Go'nun Concurrency Modeli

Sürüm notu: Bu yazıdaki kodlar Go 1.27.4 ile derlenip çalıştırılmıştır; module'ün go satırı 1.26.0'dır ve kod Go 1.26.11 ile de derlenir. Ölçümler bu yazıyı hazırladığım makinede, GOMAXPROCS=4 ile alındı; kendi makinende sayılar farklı çıkar.

Önce Tahmin Et

Sitedeki serilerin ortak örnek uygulaması Bookstore'da (bir online kitapçı) bir kitabın stok adedini bir channel üzerinden taşımak istiyoruz. C#'ta bir queue'ya önce yazıp sonra okumak gibi düşünürsen şu program sorunsuz görünür:

// Lab 07: kimsenin almadığı unbuffered bir send.
package main

import "fmt"

func main() {
    stock := make(chan int)
    stock <- 12 // başka bir goroutine alana kadar bloklar
    fmt.Println(<-stock)
}

Ekrana 12 mi yazar, yoksa bir şey mi olur?

go run ./labs/07-deadlock

fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan send]:
main.main()
    /home/user/bookstore/labs/07-deadlock/main.go:8 +0x38
exit status 2

make(chan int) ile açılan channel'ın buffer'ı yok. Böyle bir channel'a yapılan send, başka bir goroutine o değeri alana kadar bekler. Programda başka goroutine olmadığı için main 8. satırda sonsuza kadar bekleyecekti; runtime bütün goroutine'lerin uyuduğunu görüp programı durdurdu. Channel bir queue değil, iki goroutine'in buluştuğu bir nokta. Bu yazıda goroutine'leri, onları çalıştıran scheduler'ı ve channel'ları Bookstore'un stok kontrolü üzerinden göreceğiz.

Goroutine: Ucuz Bir Fonksiyon Çağrısı

Bir fonksiyonu go anahtar kelimesiyle çağırınca, fonksiyon yeni bir goroutine'de çalışmaya başlar ve çağıran beklemeden devam eder. Effective Go goroutine'i "lightweight, costing little more than the allocation of stack space" diye tanımlıyor. Ne kadar ucuz olduğunu ölçtüm: 100.000 goroutine başlatıp hepsini bir channel'da bekletiyorum.

// Lab 07: park edilmiş 100.000 goroutine'in maliyeti.
package main

import (
    "fmt"
    "runtime"
    "sync"
    "time"
)

func main() {
    const n = 100_000
    fmt.Println("GOMAXPROCS:", runtime.GOMAXPROCS(0))

    var before runtime.MemStats
    runtime.ReadMemStats(&before)

    start := time.Now()
    release := make(chan struct{})
    var wg sync.WaitGroup
    for range n {
        wg.Add(1)
        go func() {
            defer wg.Done()
            <-release // main channel'ı kapatana kadar park et
        }()
    }
    fmt.Printf("started %d goroutines in %v\n", n, time.Since(start).Round(time.Millisecond))

    var after runtime.MemStats
    runtime.ReadMemStats(&after)
    fmt.Println("goroutines:", runtime.NumGoroutine())
    fmt.Printf("stack in use: %.1f MB (%.1f KB per goroutine)\n",
        float64(after.StackInuse-before.StackInuse)/1e6,
        float64(after.StackInuse-before.StackInuse)/n/1e3)

    close(release) // tek bir close bütün alıcıları uyandırır
    wg.Wait()
    fmt.Println("after close:", runtime.NumGoroutine())
}

GOMAXPROCS=4 go run ./labs/07-goroutines

GOMAXPROCS: 4
started 100000 goroutines in 77ms
goroutines: 100001
stack in use: 204.9 MB (2.0 KB per goroutine)
after close: 1

100.000 goroutine 77 ms'de başladı ve her biri 2 KB stack kullandı. Bu sayı runtime kaynak kodundaki stackMin = 2048 sabitiyle aynı. Goroutine'in stack'i sabit değil; derin bir çağrı zincirinde runtime onu daha büyük bir alana kopyalayarak büyütür. Karşılaştırma için: Windows'ta bir thread için varsayılan olarak 1 MB stack ayrılıyor. Aynı işi 100.000 thread ile yapmak düşünülmez bile; Go'da bir HTTP isteği, bir arka plan işi ya da bir channel okuyucusu için goroutine başlatmak sıradan bir iş.

Tek bir close(release) bütün alıcıları uyandırdı ve sayı 1'e, yani main'e döndü. sync.WaitGroup ile goroutine'lerin bitmesini bekledik; onu ve Go 1.25'teki WaitGroup.Go'yu 08'de ayrıntılı göreceğiz.

Scheduler: G, M ve P

Goroutine'ler OS thread'i değil. Runtime, çok sayıda goroutine'i (G) az sayıda OS thread'ine (M) dağıtıyor; buna M:N scheduling deniyor. Aradaki üçüncü parça P (processor): Go kodu çalıştırmak için bir M'nin bir P tutması gerekiyor ve P sayısı GOMAXPROCS'a eşit. Varsayılan değer makinedeki mantıksal CPU sayısı; Go 1.25'ten beri Linux'ta container'ın cgroup CPU limiti daha düşükse o kullanılıyor.

Her P'nin kendi local run queue'su var; bir de bütün P'lerin baktığı global run queue var. Bir P'nin kuyruğu boşalınca başka bir P'nin kuyruğundan iş çalıyor (work stealing). Bir goroutine channel'da beklemeye başladığında park ediliyor ve aynı M, kuyruktaki bir sonraki goroutine'i çalıştırıyor; thread boşta beklemiyor. Network I/O'da da aynısı oluyor: socket'ten veri bekleyen goroutine network poller'a devrediliyor, veri gelince tekrar kuyruğa giriyor. Go 1.14'ten beri uzun süren bir döngü de sonsuza kadar CPU'yu tutamıyor; runtime goroutine'leri asenkron olarak durdurabiliyor (preemption).

async/await Olmadan Asenkron: Fonksiyon Rengi Yok

C# Asenkron Programlama serisinin "C# async/await Nasıl Çalışır: State Machine ve Continuation" yazısında, derleyicinin her async method'u bir state machine'e çevirdiğini ve await'te thread'i serbest bıraktığını görmüştük. Bunun bedeli, Bob Nystrom'un 2015 tarihli "What Color is Your Function?" yazısında anlattığı ayrım: async bir method'u çağıran method da async olmak zorunda, Task call stack boyunca yukarı yayılıyor ve senkron bir interface'in arkasına async kod koymak zorlaşıyor.

Go'da bu ayrım yok. http.Get, conn.Read, bir channel receive ya da veritabanı sorgusu, kodda sıradan bloklayan çağrılar gibi yazılır. Bloklayan şey thread değil goroutine'dir; scheduler o sırada aynı thread'de başka goroutine'leri çalıştırır. C#'taki state machine'in yaptığı işi Go'da runtime yapıyor, sen sadece düz kod yazıyorsun. net/http server'ı da bu yüzden her bağlantı için bir goroutine başlatıyor; 05'te handler'daki panic'in neden sadece o bağlantıyı etkilediğini de bu açıklıyor. Bunun karşılığında, aynı anda çalışacak işi açıkça go ile başlatıp sonucunu bir channel ya da WaitGroup ile toplaman gerekiyor; C#'taki Task.WhenAll'ın hazır bir karşılığı dilde yok.

Channel'lar: Unbuffered, Buffered, close ve range

Go'nun concurrency modeli Tony Hoare'ın 1978 tarihli CSP (Communicating Sequential Processes; birbirine mesajla bağlanan sıralı süreçler modeli) çalışmasına dayanıyor. Effective Go bunu tek cümleyle söylüyor: "Do not communicate by sharing memory; instead, share memory by communicating." Bir veriyi birden çok goroutine'in eriştiği bir değişkende tutmak yerine, onu bir channel ile sahibinden diğerine geçirirsin. Modelin teorisi Concurrency Teorisi serisinin "Concurrency Modelleri ve Concurrent Kodu Doğrulamak" yazısında; burada pratiği.

// Lab 07: unbuffered ve buffered channel'lar, close ve range, select.
package main

import (
    "fmt"
    "time"
)

var start = time.Now()

func logf(format string, args ...any) {
    fmt.Printf("%4dms  "+format+"\n", append([]any{time.Since(start).Milliseconds()}, args...)...)
}

// reserve n tane rezervasyon ID'si gönderir ve bitince channel'ı kapatır.
func reserve(out chan<- string, n int) {
    for i := 1; i <= n; i++ {
        out <- fmt.Sprintf("RES-%d", i)
        logf("sent RES-%d", i)
    }
    close(out) // channel'ı sadece gönderen kapatır
}

func main() {
    // 1) Unbuffered: her send bir receive'i bekler.
    unbuffered := make(chan string)
    go reserve(unbuffered, 3)
    time.Sleep(100 * time.Millisecond)
    for id := range unbuffered { // channel kapanıp boşalınca döngü biter
        logf("got  %s", id)
        time.Sleep(100 * time.Millisecond)
    }

    // 2) Buffered: yer olduğu sürece send beklemez.
    start = time.Now()
    buffered := make(chan string, 3)
    go reserve(buffered, 3)
    time.Sleep(100 * time.Millisecond)
    for id := range buffered {
        logf("got  %s", id)
    }

    // 3) Kapalı channel'dan okumak: zero value ve ok == false.
    id, ok := <-buffered
    fmt.Printf("closed: %q %t\n", id, ok)

    // 4) select: hangisi önce hazır olursa onu al ya da vazgeç.
    slow := make(chan string)
    go func() {
        time.Sleep(300 * time.Millisecond)
        slow <- "stock: 12"
    }()
    select {
    case v := <-slow:
        fmt.Println(v)
    case <-time.After(100 * time.Millisecond):
        fmt.Println("inventory did not answer in 100ms")
    }

    // 5) default, select'i bloklamayan hâle getirir.
    select {
    case v := <-slow:
        fmt.Println("late answer:", v)
    default:
        fmt.Println("nothing ready yet")
    }
    fmt.Println("late answer:", <-slow) // goroutine hâlâ göndermek için bekliyor
}

go run ./labs/07-channels

 102ms  got  RES-1
 102ms  sent RES-1
 214ms  got  RES-2
 217ms  sent RES-2
 316ms  got  RES-3
 316ms  sent RES-3
   0ms  sent RES-1
   0ms  sent RES-2
   0ms  sent RES-3
 102ms  got  RES-1
 102ms  got  RES-2
 102ms  got  RES-3
closed: "" false
inventory did not answer in 100ms
nothing ready yet
late answer: stock: 12

İlk blokta reserve, RES-1'i 0. milisaniyede göndermeye çalıştı ama main 100 ms uyuduğu için "sent" satırı 102. milisaniyede geldi. Unbuffered channel'da send ve receive aynı anda tamamlanır; değer doğrudan bir goroutine'den diğerine geçer. Sonraki iki rezervasyon da alıcının hızına bağlı kaldı. Figürün sağ tarafı bu buluşmayı gösteriyor. Kapasitesi 3 olan buffered channel'da ise üç send de 0. milisaniyede bitti; gönderici, buffer dolana kadar alıcıyı beklemiyor.

chan<- string ve <-chan string yön kısıtlı channel tipleri. reserve sadece gönderebiliyor; yanlışlıkla okumaya çalışırsa derleme hatası alır. close, "bundan sonra değer gelmeyecek" demek; for range channel kapanıp boşalınca biter, kapalı bir channel'dan okumak ise beklemeden zero value ve ok == false döndürür. Channel'ı her zaman gönderen taraf kapatır.

select, birden fazla channel işleminden hangisi önce hazır olursa onu seçer. time.After belli bir süre sonra değer gönderen bir channel döndürdüğü için basit bir timeout'a dönüşüyor; default ise hiçbir case hazır değilse beklemeden çalışıyor. HTTP isteklerinde timeout'u context ile vereceğiz; o da 08'in konusu.

nil channel'ın sonsuza kadar bloklaması işe yarar bir özellik: select içinde bir channel değişkenini nil yaparsan o case bir daha seçilmez. İki kaynaktan okuyup biri bitince onu devreden çıkarmak için kullanılan yaygın bir yöntem.

C# Channels ile Kıyas

C# Asenkron Programlama serisinin "C# Channels ile Producer/Consumer: Backpressure, Bounded Channel ve Pipeline" yazısındaki System.Threading.Channels kavram olarak Go channel'larına çok yakın. Channel.CreateBounded<T>(3) bir buffered channel'ın, Writer.Complete() close'un, await foreach (var x in reader.ReadAllAsync()) de for range'in karşılığı. Bounded channel'ın dolunca yazanı bekletmesi, buffer'ı dolan Go channel'ının send'i bekletmesiyle aynı backpressure.

Farklar da var. C#'ta channel bir library; Go'da dilin parçası ve select ile birden çok channel'ı aynı anda beklemek dilin kendi syntax'ı. Go'da make(chan T) varsayılan olarak kapasitesi sıfır olan, yani göndereni alanla buluşturan bir channel açar; C#'ta kapasiteyi her zaman sen seçersin (CreateBounded ya da CreateUnbounded). Kapalı channel'a yazmak C#'ta TryWrite'ın false döndürmesi ya da bir exception iken Go'da panic. Lab'da üç panic'i de tek tek denedim:

func try(name string, f func()) {
    defer func() { fmt.Printf("%-16s %v\n", name, recover()) }()
    f()
}

func main() {
    try("send on closed", func() {
        ch := make(chan int, 1)
        close(ch)
        ch <- 1
    })
    try("close twice", func() {
        ch := make(chan int)
        close(ch)
        close(ch)
    })
    try("close nil", func() {
        var ch chan int
        close(ch)
    })

go run ./labs/07-panics

send on closed   send on closed channel
close twice      close of closed channel
close nil        close of nil channel

Pipeline, Fan-out ve Fan-in: Sepetin Stoğunu Kontrol Etmek

Bookstore'da sepetteki her kitap için envanter servisine sormamız gerekiyor ve her çağrı 50 ms sürüyor. Altı kitap için sırayla sormak 300 ms eder. Bunu üç aşamalı bir pipeline ile yazdım: generate ISBN'leri (kitaplara verilen uluslararası standart numaralar) bir channel'a koyuyor, üç check worker'ı aynı channel'dan okuyor (fan-out), merge da üç sonuç channel'ını tek bir channel'da topluyor (fan-in).

// generate ilk aşama: bir slice'ı channel'a çevirir.
func generate(isbns []string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for _, isbn := range isbns {
            out <- isbn
        }
    }()
    return out
}

// check bir worker: in kapanana kadar ISBN okur.
func check(in <-chan string) <-chan result {
    out := make(chan result)
    go func() {
        defer close(out)
        for isbn := range in {
            b, err := lookup(isbn)
            out <- result{isbn: isbn, stock: b.Stock, err: err}
        }
    }()
    return out
}

// merge fan-in yapar: her giriş channel'ını tek bir çıkış channel'ına kopyalar.
func merge(ins ...<-chan result) <-chan result {
    out := make(chan result)
    var wg sync.WaitGroup
    for _, in := range ins {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for r := range in {
                out <- r
            }
        }()
    }
    go func() {
        wg.Wait()
        close(out) // bütün kopyalayıcılar bitince tek bir kez kapat
    }()
    return out
}

func main() {
    cart := []string{"9780441172719", "9780547928227", "9780141439549",
        "9780141439587", "9780141439686", "9780000000000"}

    start := time.Now()
    for _, isbn := range cart {
        if _, err := lookup(isbn); err != nil {
            fmt.Println("sequential:", err)
        }
    }
    fmt.Println("sequential:", time.Since(start).Round(10*time.Millisecond))

    start = time.Now()
    isbns := generate(cart)
    workers := []<-chan result{check(isbns), check(isbns), check(isbns)} // fan-out: tek channel, 3 okuyucu
    for r := range merge(workers...) {
        if r.err != nil {
            fmt.Println("  error:", r.err)
            continue
        }
        fmt.Printf("  %s stock=%d\n", r.isbn, r.stock)
    }
    fmt.Println("pipeline with 3 workers:", time.Since(start).Round(10*time.Millisecond))
}

go run ./labs/07-stockcheck

sequential: isbn 9780000000000: book not found
sequential: 310ms
  9780141439549 stock=4
  9780547928227 stock=0
  9780441172719 stock=12
  9780141439686 stock=3
  error: isbn 9780000000000: book not found
  9780141439587 stock=7
pipeline with 3 workers: 100ms

Üç worker ile süre 310 ms'den 100 ms'ye indi: altı iş, üçer üçer iki turda bitti. Sonuçların sırası ise sepetteki sıra değil; hangi worker önce bitirirse onun sonucu önce geliyor. Sıra önemliyse sonucu index ile taşıyıp sonradan sıralaman gerekir. merge'deki ayrıntıya dikkat et: çıkış channel'ını, bütün kopyalayıcıların bitmesini bekleyen ayrı bir goroutine kapatıyor. Kopyalayıcılardan biri kapatsaydı diğerleri kapalı channel'a yazıp panic olurdu. Döngüdeki go func()'ın in'i doğrudan kullanabilmesi de Go 1.22'deki döngü değişkeni değişikliği sayesinde; 03'te konuşmuştuk.

Goroutine Leak: Hiç Bitmeyen Goroutine'ler

Garbage collector bloklu bir goroutine'i toplamaz. Bir goroutine kimsenin okumadığı bir channel'a yazmaya çalışıyorsa, programın sonuna kadar bellekte kalır. En sık sebep, sonuçları toplayan tarafın erken dönmesi:

// firstMissing her ISBN için bir goroutine başlatır ve ilk hatada döner.
func firstMissing(isbns []string) error {
    errs := make(chan error) // unbuffered: her gönderici bir alıcı bekler
    for _, isbn := range isbns {
        go func() {
            _, err := catalog.Store{}.ByISBN(isbn)
            if err == nil {
                time.Sleep(50 * time.Millisecond) // bilinen kitaplar envanter servisine de sorulur
            }
            errs <- err
        }()
    }
    for range isbns {
        if err := <-errs; err != nil {
            return err // kalan göndericiler hiçbir zaman alıcı bulamaz
        }
    }
    return nil
}

go run ./labs/07-leak

call 1: book not found, goroutines=4
call 2: book not found, goroutines=7
call 3: book not found, goroutines=10

Her çağrı ilk hatayı alıp döndü ve geride üç goroutine bıraktı. Bu kod bir HTTP handler'ında olsaydı her istek üç goroutine sızdırırdı; günler sonra bellek grafiği yavaş yavaş yükselir ve sebebi kodda kolayca görünmezdi. Bu örnekteki çözüm basit: channel'a her gönderici için yer açmak.

errs := make(chan error, len(isbns)) // her gönderici için yer var

Bu değişiklikle üç çağrıdan sonra da goroutines=1 kaldı. Daha genel çözüm, goroutine'lere "artık sonuç istemiyorum" diyebilmek; Go'da bunun yolu context ile iptal. 08'de onu ve Go 1.27'de genel kullanıma açılan goroutineleak profilini göreceğiz.

Üretimde Dikkat

Bir goroutine'deki panic bütün process'i durdurur. 05'teki Recover middleware'i sadece handler'ın kendi goroutine'ini korur; handler'ın içinde başlattığın bir goroutine panic olursa API tamamen kapanır:

go run ./labs/07-panics crash

send on closed   send on closed channel
close twice      close of closed channel
close nil        close of nil channel
panic: assignment to entry in nil map

goroutine 19 [running]:
main.main.func5()
    /home/user/bookstore/labs/07-panics/main.go:37 +0x34
created by main.main in goroutine 1
    /home/user/bookstore/labs/07-panics/main.go:34 +0xd0
exit status 2

main'deki defer/recover hiçbir şey yapamadı, çünkü recover sadece kendi goroutine'indeki panic'i yakalar. Arka planda iş başlatan her goroutine, gerekiyorsa kendi recover'ını taşımalı.

Sınırsız goroutine başlatma. Goroutine ucuz ama arkasındaki iş ucuz değil. Gelen her sipariş için bir goroutine açıp her birinin envanter servisine gitmesi, o servisi ve veritabanı connection pool'unu bir anda tüketebilir. Lab'daki gibi sabit sayıda worker kullanmak paralelliği sınırlamanın en basit yolu; 08'de errgroup.SetLimit ile aynı işi daha az kodla yapacağız.

Container'da GOMAXPROCS. Go 1.25 öncesinde, CPU limiti 2 olan bir container'da 64 çekirdekli bir host üzerinde çalışan Go servisi 64 P açıyor ve limit yüzünden sürekli kısılıyordu. 1.25 ve sonrasında runtime Linux'ta cgroup limitini dikkate alıyor; daha eski bir sürümle derlenmiş servisleri container'a koyarken bu farkı hatırla.

Sık Düşülen Tuzaklar

Channel'ı queue sanmak. Önce Tahmin Et'teki deadlock bu: unbuffered bir channel'a aynı goroutine'de yazıp okuyamazsın. Buffer'ı "sorun çözülsün" diye büyütmek de genellikle hatayı sadece erteler.

Alıcı tarafta close çağırmak. Alıcı channel'ı kapatırsa, hâlâ göndermeye çalışan bir goroutine panic olur. Kapatma sorumluluğu her zaman gönderende; birden fazla gönderen varsa merge'deki gibi hepsini bekleyen tek bir yer kapatır.

Sonucu okunmayan goroutine. Erken return, okunmayan bir channel ve iptal mekanizması olmayan worker'lar goroutine leak'in klasik üçlüsü. Her go satırında "bu goroutine ne zaman ve nasıl bitecek?" diye sor.

Sıralamaya güvenmek. Fan-in'den gelen sonuçlar işlerin bitiş sırasıyla gelir. Test bir gün geçip ertesi gün kalıyorsa ilk şüpheli bu.

Paylaşılan değişkeni korumasız kullanmak. Goroutine'ler aynı map'e ya da sayaca channel olmadan yazıyorsa bir data race var. Bookstore'un Service.nextID sayacı ve memstore tam olarak bu durumda; bir sonraki yazıda race detector ile bulup düzelteceğiz.

Küçük Bir Deneme

★☆☆ labs/07-deadlock'u iki farklı şekilde düzelt: önce channel'ı make(chan int, 1) yap, sonra buffer'sız bırakıp send'i ayrı bir goroutine'e taşı. İki çözümün farkını bir cümleyle yaz.

★★☆ labs/07-stockcheck'teki worker sayısını 1, 2, 6 ve 12 yapıp süreleri karşılaştır. 6'dan sonra neden hızlanma durdu?

★★★ merge'i, sonuçları sepetteki sırayla döndürecek şekilde değiştir. İpucu: generate, ISBN ile birlikte index'i de göndersin; sonuçları index'e göre bir slice'a yaz ve channel kapanınca o slice'ı yazdır.

Goroutine ve channel'ları gördük, ama gerçek bir serviste eksik bir parça var: bir isteğin iptal edildiğini goroutine'lere nasıl söyleyeceğiz? Bir sonraki yazıda context.Context, sync.Mutex ve WaitGroup, errgroup ile worker pool, race detector ve goroutine leak tespitiyle devam ediyoruz. Bookstore'un eşzamanlı kullanımda bozulan sipariş numarası sayacını ve memstore'u da orada düzelteceğiz.