Mengingat Kembali Tentang Concurrency dan Parallelism

Goroutine adalah fungsi yang dibuat dan dijadwalkan untuk dijalankan secara independen oleh Go scheduler. Go scheduler bertanggung jawab atas pengelolaan dan eksekusi goroutine. Di balik layar, terdapat algoritma yang kompleks untuk bisa membuat goroutine bekerja. Untungnya, di Golang, kita dapat achieve operasi yang sangat kompleks ini dengan sederhana menggunakan keyword go.

Dalam snippet berikut, kita memiliki fungsi main yang memanggil fungsi say secara berurutan, dengan memberikan argumen "hello" dan "world":

func main() {
    say("hello")
    say("world")
}

Fungsi say menerima string sebagai parameter dan melakukan iterasi lima kali. Untuk setiap iterasi, kita membuat fungsi sleep selama 500 milidetik dan mencetak parameter s segera setelahnya:

func say(s string) {
    for i := 1; i < 5; i++ {
        time.Sleep(500 * time.Millisecond)
        fmt.Println(s)
    }
}

Ketika kita menjalankan program, program akan mencetak keluaran berikut:

hello
hello
hello
hello
hello
world
world
world
world
world

Sekarang, kita akan menggunakan kata kunci go tepat sebelum panggilan pertama ke fungsi say untuk memperkenalkan concurrency dalam program kita:

func main() {
    go say("hello")
    say("world")
}

Hasilnya seharusnya bergantian antara "hello" dan "world".

Jadi, kita dapat mencapai hasil yang sama jika kita membuat goroutine untuk panggilan fungsi kedua, kan yah?

func main() {
    say("hello")
    go say("world")
}

Mari kita lihat hasil programnya sekarang:

hello
hello
hello
hello

Ups! Ada yang salah di sini. Apa yang salah? Fungsi main dan goroutine tampak tidak sinkron.

Kita tidak melakukan kesalahan apa pun. Ya, Itu memang perilaku yang diharapkan. Ketika Anda melihat lebih dekat pada program pertama, goroutine dipicu, dan panggilan kedua dari say dieksekusi dalam konteks fungsi main secara berurutan.

Dengan kata lain, program harus menunggu fungsi untuk dihentikan untuk mencapai akhir fungsi main. Untuk program kedua, kita memiliki behavior yang berlawanan. Panggilan pertama adalah panggilan fungsi normal, jadi program mencetak lima kali seperti yang diharapkan, tetapi ketika goroutine kedua dipicu, tidak ada instruksi berikutnya pada fungsi main, sehingga program berhenti.

Meskipun perilakunya benar dari perspektif bagaimana program bekerja, ini bukanlah tujuan kita. Kita memerlukan cara untuk menyinkronkan penantian untuk semua goroutine dalam grup eksekusi ini sebelum memberikan fungsi main kesempatan untuk berhenti. Dalam situasi seperti ini, kita dapat memanfaatkan sesuatu dari package sync, yang disebut dengan WaitGroup.

WaitGroup

WaitGroup, seperti namanya, adalah mekanisme standard library Go yang memungkinkan kita untuk menunggu grup / kumpulan goroutine hingga selesai secara eksplisit.

Tidak ada factory function khusus untuk membuatnya, karena zero-value-nya sudah merupakan state yang valid dan dapat digunakan. Karena WaitGroup telah dibuat, kita perlu mengontrol berapa banyak goroutine yang sedang kita tunggu. Kita dapat menggunakan metode Add() .

Bagaimana kita dapat memberi tahu wg bahwa kita telah menyelesaikan salah satu routine? Caranya sangat intuitif. Kita dapat menggunakan method Done().

Dalam contoh berikut, kita menggunakan wait group untuk membuat program kita mengeluarkan pesan seperti yang diinginkan:

func main() {
    wg := sync.WaitGroup{}
    wg.Add(2)
    go say("world", &wg)
    go say("hello", &wg)
    wg.Wait()
}

Kita membuat WaitGroup (wg := sync.WaitGroup{}) dan menyatakan bahwa dua goroutine berpartisipasi dalam grup ini (wg.Add(2)).

Di baris terakhir program, kita secara eksplisit menahan eksekusi dengan metode Wait() untuk menghindari penghentian program.

Untuk membuat fungsi kita berinteraksi dengan WaitGroup, kita perlu mengirimkan reference ke grup ini. Setelah kita memiliki reference-nya, fungsi tersebut dapat menggunakan defer, memanggil Done(), untuk memastikan bahwa kita memberi sinyal dengan benar untuk grup kita setiap kali fungsi selesai.

Ini adalah fungsi say yang baru:

func say(s string, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := 0; i < 5; i++ {
        fmt.Println(s)
    }
}

Kita tidak perlu bergantung pada time.Sleep() lagi.

Sekarang, kita dapat mengontrol grup goroutine kita. Let’s deal with one central worrisome issue in concurrent programming – state.

Mengubah Shared State

Bayangkan sebuah skenario di mana dua pekerja yang rajin ditugaskan untuk mengemas barang ke dalam kotak di gudang. Setiap pekerja mengisi sejumlah barang tetap ke dalam paket, dan kita harus melacak jumlah total barang yang dikemas.

Tugas yang tampaknya mudah ini dapat dengan cepat menjadi mimpi buruk jika tidak ditangani dengan benar. Tanpa sinkronisasi yang tepat, para pekerja dapat menghindari gangguan yang disengaja terhadap pekerjaan satu sama lain, yang mengarah pada hasil yang salah dan perilaku yang tidak dapat diprediksi. Ini adalah contoh klasik dari data race, tantangan umum dalam pemrograman concurrent.

Kode berikut akan memberi analogi di mana dua pekerja gudang menghadapi masalah data race saat mengemas barang ke dalam kotak. Pertama-tama kita akan menyajikan kode tanpa sinkronisasi yang tepat, yang menunjukkan masalah data race. Kemudian, kita akan memodifikasi kode untuk mengatasi masalah tersebut, memastikan bahwa para pekerja berkolaborasi dengan lancar dan akurat.

Mari kita melangkah ke gudang yang ramai dan menyaksikan langsung tantangan concurrency dan pentingnya sinkronisasi dalam contoh ini:

package main

import (
    "fmt"
    "sync"
)

func main() {
    fmt.Println("Total Items Packed:", PackItems(0))
}

func PackItems(totalItems int) int {
    const workers = 2
    const itemsPerWorker = 1000
    var wg sync.WaitGroup
    itemsPacked := 0
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            // Simulate the worker packing items into boxes.
            for j := 0; j < itemsPerWorker; j++ {
                itemsPacked = totalItems
                // Simulate packing an item.
                itemsPacked++
                // Update the total items packed without proper
                // synchronization.
                totalItems = itemsPacked
            }
        }(i)
    }
    // Wait for all workers to finish.
    wg.Wait()
    return totalItems
}

Fungsi main dimulai dengan memanggil fungsi PackItems dengan nilai totalItems awal 0.

Dalam fungsi PackItems, ada dua konstanta yang didefinisikan:

• workers: Jumlah goroutine pekerja (diatur ke 2)

• itemsPerWorker: Jumlah barang yang harus dikemas setiap pekerja ke dalam kotak (diatur ke 1.000)

WaitGroup bernama wg dibuat untuk menunggu semua goroutine pekerja selesai sebelum mengembalikan nilai totalItems akhir.

Sebuah loop berjalan workers kali, di mana setiap iterasi memulai goroutine baru untuk mensimulasikan seorang pekerja yang mengemas barang ke dalam kotak. Di dalam goroutine, langkah-langkah berikut dilakukan:

1. Sebuah ID worker diteruskan ke goroutine sebagai argumen.

2. Pernyataan defer wg.Done() memastikan bahwa wait group dikurangi ketika goroutine keluar.

3. Variabel itemsPacked diinisialisasi dengan nilai totalItems saat ini untuk melacak barang yang dikemas oleh pekerja ini.

4. Sebuah loop berjalan itemsPerWorker kali, mensimulasikan proses pengepakan barang ke dalam kotak. Namun, tidak ada pengepakan aktual yang terjadi; loop hanya menambah variabel itemsPacked.

5. Pada langkah terakhir dalam loop bagian dalam, totalItems menerima nilai itemsPacked yang diubah, yang berisi jumlah barang yang dikemas oleh pekerja.

Di sinilah masalah sinkronisasi terjadi. Pekerja mencoba memperbarui variabel totalItems dengan menambahkan nilai itemsPacked ke dalamnya.

Karena beberapa goroutine mencoba untuk memodifikasi totalItems secara concurrently tanpa sinkronisasi yang tepat, data race terjadi, yang mengarah pada hasil yang tidak dapat diprediksi dan salah.

Data Race Ilustration

Hasil Non-Deterministik

Pertimbangkan alternatif fungsi main berikut:

func main() {
    times := 0
    for {
        times++
        counter := PackItems(0)
        if counter != 2000 {
            log.Fatalf("it should be 2000 but found %d on execution %d", counter, times)
        }
    }
}

Program terus-menerus menjalankan fungsi PackItems hingga hasil yang diharapkan sebesar 2.000 tidak tercapai. Setelah ini terjadi, program akan menampilkan nilai salah yang dikembalikan oleh fungsi dan jumlah percobaan yang diperlukan untuk mencapai titik itu.

Karena sifat non-deterministik dari Go scheduler, hasilnya akan benar sebagian besar waktu. Kode ini akan membutuhkan banyak run untuk mengungkapkan kelemahan sinkronisasinya.

Dalam satu eksekusi, saya membutuhkan lebih dari 16.000 iterasi:

it should be 2000 but found 1170 on execution 16421

Giliran Anda!

Bereksperimenlah menjalankan kode di mesin Anda. Berapa banyak iterasi yang dibutuhkan kode Anda untuk gagal?

Jika Anda menggunakan komputer pribadi Anda, kemungkinan besar ada banyak tugas yang sedang dilakukan, tetapi mesin Anda mungkin memiliki banyak sumber daya yang tidak digunakan. Namun, penting untuk mempertimbangkan jumlah noise pada shared node dalam cluster jika Anda menjalankan program di lingkungan cloud dengan container. Dengan "noise," maksud saya pekerjaan yang dilakukan pada mesin host saat menjalankan program Anda. Mungkin saja idle seperti eksperimen lokal Anda. Namun, kemungkinan besar digunakan secara maksimal dalam skenario hemat biaya di mana setiap core dan memori dimanfaatkan.

Skenario kontes konstan untuk sumber daya ini membuat scheduler kita jauh lebih cenderung untuk memilih workload lain daripada hanya terus menjalankan goroutine kita.

Dalam contoh berikut, kita memanggil fungsi runtime.Gosched untuk meniru noise. Idenya adalah untuk memberikan petunjuk kepada Go scheduler, dengan mengatakan, "Hei! Mungkin ini adalah saat yang tepat untuk menjeda saya":

for j := 0; j < itemsPerWorker; j++ {
    itemsPacked = totalItems
    runtime.Gosched() // emulating noise!
    itemsPacked++
    totalItems = itemsPacked
}

Menjalankan fungsi main lagi, kita dapat melihat bahwa hasil yang salah terjadi jauh lebih cepat dari sebelumnya.

Dalam eksekusi saya, misalnya, saya hanya membutuhkan empat iterasi:

it should be 2000 but found 1507 on execution 4

Sayangnya, kode tersebut masih buggy. Bagaimana kita bisa mengantisipasi itu? Pada titik ini, Anda seharusnya sudah menebak bahwa tool Go memiliki jawabannya, dan Anda benar lagi. Kita dapat mengelola data race pada test kita.

Mengelola Data Race

Ketika beberapa goroutine mengakses shared data atau sumber daya secara concurrently, "race condition" dapat terjadi. Seperti yang dapat kita buktikan, bug concurrency jenis ini dapat menyebabkan perilaku yang tidak dapat diprediksi dan tidak diinginkan.

Tool test Go memiliki fitur bawaan yang disebut Go race detection yang dapat mendeteksi dan mengidentifikasi race condition dalam kode Go Anda.

Jadi, mari kita buat file main_test.go dengan test case sederhana:

Go

package main

import (
    "testing"
)

func TestPackItems(t *testing.T) {
    totalItems := PackItems(2000)
    expectedTotal := 2000
    if totalItems != expectedTotal {
        t.Errorf("Expected total: %d, Actual total: %d",
            expectedTotal, totalItems)
    }
}

Sekarang, mari kita gunakan race detector:

Bash

go test -race

Hasil di console akan menjadi seperti ini:

==================
WARNING: DATA RACE
Read at 0x00c00000e288 by goroutine 9:
  example1.PackItems.func1()
      /tmp/main.go:35 +0xa8
  example1.PackItems.func2()
      /tmp/main.go:45 +0x47
Previous write at 0x00c00000e288 by goroutine 8:
  example1.PackItems.func1()
      /tmp/main.go:39 +0xba
  example1.PackItems.func2()
      /tmp/main.go:45 +0x47
// Other lines omitted for brevity

Keluaran mungkin tampak menakutkan pada pandangan pertama, tetapi informasi yang paling penting pada awalnya adalah pesan WARNING: DATA RACE.

Untuk memperbaiki masalah sinkronisasi dalam kode ini, kita harus menggunakan mekanisme sinkronisasi untuk melindungi akses ke variabel totalItems. Tanpa sinkronisasi yang tepat, penulisan concurrent ke shared data dapat menyebabkan race condition dan hasil yang tidak terduga.

Kita telah menggunakan WaitGroup dari package sync. Mari kita jelajahi lebih banyak mekanisme sinkronisasi untuk memastikan kebenaran program.

Operasi Atomik

Sangat disayangkan bahwa istilah "atomik" di Go tidak melibatkan manipulasi atom secara fisik, seperti dalam fisika atau kimia. Akan sangat menarik untuk memiliki kemampuan itu dalam pemrograman; sebagai gantinya, operasi atomik di Go difokuskan untuk menyinkronkan dan mengelola concurrency di antara goroutine menggunakan package sync/atomic.

Go menawarkan operasi atomik untuk memuat, menyimpan, menambah, dan CAS (compare and swap) untuk jenis tertentu, seperti int32, int64, uint32, uint64, uintptr, float32, dan float64. Operasi atomik tidak dapat dilakukan secara langsung pada struktur data arbitrer.

Mari kita ubah program kita menggunakan package atomic. Pertama, kita harus mengimpornya:

Go

import (
    "fmt"
    "sync"
    "sync/atomic"
)

Alih-alih memperbarui totalItems secara langsung, kita akan memanfaatkan fungsi AddInt32 untuk menjamin sinkronisasi:

Go

for j := 0; j < itemsPerWorker; j++ {
    atomic.AddInt32(&totalItems, int32(itemsPacked))
}

Jika kita memeriksa data race lagi, tidak ada masalah yang akan dilaporkan.

Struktur atomik sangat bagus ketika kita perlu menyinkronkan satu operasi, tetapi ketika kita ingin menyinkronkan blok kode, tool lain lebih cocok, seperti mutex.

Mutexes

Ah, mutexes! Mereka seperti bouncer di pesta untuk goroutine. Bayangkan sekelompok makhluk Go kecil ini mencoba menari dengan shared data. Semuanya menyenangkan dan permainan sampai kekacauan terjadi, dan Anda memiliki kemacetan lalu lintas goroutine dengan data yang tumpah di mana-mana!

Jangan khawatir, karena mutexes datang seperti pengawas lantai dansa, memastikan bahwa hanya satu goroutine groovy yang dapat melakukan gerakan di bagian critical section pada satu waktu. Mereka seperti penjaga ritme concurrency, memastikan bahwa setiap orang bergiliran dan tidak ada yang menginjak jari kaki satu sama lain.

Anda dapat membuat mutex dengan mendeklarasikan variabel bertipe sync.Mutex. Sebuah mutex memungkinkan