Channel

Channel adalah saluran bertipe untuk mengirim pesan antar fiber dan menjadi sarana utama pertukaran data dalam model konkurensi Crystal. Topik ini membahas channel tanpa buffer dan channel berbuffer, penutupan channel beserta akibatnya bagi penerima, pemeriksaan tipe oleh compiler, serta pola produsen-konsumen dan perulangan atas isi channel. Setiap program dijalankan dan keluarannya dicantumkan apa adanya.

Channel sebagai saluran pesan#

Channel adalah objek bertipe yang menghubungkan satu fiber dengan fiber lain melalui pengiriman pesan. Model ini mengikuti gaya Communicating Sequential Processes (CSP), yaitu model konkurensi yang menganjurkan pertukaran data melalui pesan alih-alih berbagi memori secara langsung. Tipe isi channel ditetapkan saat pembuatan, misalnya Channel(Int32) untuk channel yang hanya membawa bilangan bulat. Metode send mengirim satu nilai, sedangkan receive mengambil satu nilai dari channel.

Channel tanpa buffer, yaitu channel yang dibuat tanpa kapasitas, berperilaku seperti serah terima langsung. Pemanggilan send memblokir sampai ada fiber lain yang memanggil receive, sehingga kedua sisi bertemu pada saat pengiriman. Program berikut menunjukkan urutan kejadiannya.

channel = Channel(Int32).new

spawn do
  puts "pengirim: siap mengirim"
  channel.send(7)
  puts "pengirim: send selesai"
end

sleep 10.milliseconds
puts "utama: menerima nilai"
nilai = channel.receive
puts "utama: menerima #{nilai}"
sleep 10.milliseconds

# menjalankan program serah terima channel tanpa buffer
crystal run dasar-channel.cr

pengirim: siap mengirim
utama: menerima nilai
utama: menerima 7
pengirim: send selesai

Keluaran memperlihatkan serah terimanya secara berurutan. Fiber pengirim mencetak baris pertamanya, lalu menunggu di send karena belum ada penerima. Program utama mencetak utama: menerima nilai, dan pemanggilan receive baru selesai ketika nilai 7 berhasil diserahkan. Baris pengirim: send selesai muncul paling akhir karena fiber pengirim baru berjalan kembali setelah program utama menyerahkan giliran melalui sleep di akhir program.

Channel berbuffer#

Channel berbuffer menyimpan sejumlah nilai sebelum ada penerima. Kapasitasnya diberikan melalui parameter capacity, dan send tidak lagi memblokir selama buffer masih memiliki sisa ruang. Dengan buffer, pengirim dan penerima tidak perlu aktif pada saat yang sama.

channel = Channel(Int32).new(capacity: 3)

3.times do |i|
  channel.send(i * 10)
end
puts "tiga nilai terkirim tanpa ada penerima"

3.times do
  puts "menerima #{channel.receive}"
end

# menjalankan program channel berbuffer kapasitas 3
crystal run channel-buffer.cr

tiga nilai terkirim tanpa ada penerima
menerima 0
menerima 10
menerima 20

Program di atas mengirim tiga nilai tanpa satu pun penerima, dan pengiriman selesai tanpa hambatan karena kapasitas buffer tepat tiga. Nilai dikeluarkan kembali oleh receive sesuai urutan kedatangannya, yaitu 0, 10, lalu 20. Apabila nilai keempat dikirim ke channel yang sama, send memblokir sampai ada penerima yang mengosongkan buffer. Channel tanpa buffer dapat dilihat sebagai channel berbuffer dengan kapasitas nol.

Buffer bekerja seperti antrean, yaitu nilai pertama yang masuk adalah nilai pertama yang keluar. Urutan pada keluaran program di atas memperlihatkan perilaku tersebut. Sifat ini membuat channel berbuffer cocok untuk menyerap perbedaan laju antara pengirim yang cepat dan penerima yang lambat, selama penumpukannya tidak melebihi kapasitas.

Menutup channel#

Metode close menandai sebuah channel selesai digunakan. Penutupan memberi tahu sisi penerima bahwa tidak akan ada nilai baru, sehingga penerima dapat berhenti menunggu. Setelah channel ditutup, receive? mengembalikan nil sebagai tanda tidak ada nilai lagi, sedangkan receive melempar error Channel::ClosedError.

channel = Channel(Int32).new

spawn do
  channel.send(1)
  channel.send(2)
  channel.close
end

puts channel.receive
puts channel.receive
puts channel.receive?.inspect

begin
  channel.receive
rescue error : Channel::ClosedError
  puts "error: #{error.class}: #{error.message}"
end

# menjalankan program penutupan channel
crystal run tutup-channel.cr

1
2
nil
error: Channel::ClosedError: Channel is closed

Program di atas menerima dua nilai yang dikirim sebelum penutupan. Pemanggilan receive? berikutnya mengembalikan nil yang dicetak melalui inspect agar terlihat jelas dan bukan keluaran kosong. Pemanggilan receive pada channel yang sudah ditutup memicu Channel::ClosedError dengan pesan Channel is closed, sehingga penanganannya dibungkus blok begin dan rescue.

Pilihan di antara keduanya menyesuaikan kebutuhan, receive? untuk perulangan yang berhenti secara wajar dan receive untuk menegaskan bahwa penerimaan setelah penutupan adalah kondisi tidak wajar. Penutupan juga tidak membuang nilai yang masih tertahan di buffer. Nilai yang terkirim sebelum close tetap dapat diterima setelahnya, dan receive? baru mengembalikan nil setelah buffer benar-benar kosong.

Tipe pesan diperiksa compiler#

Compiler memeriksa tipe setiap nilai yang dikirim ke channel. Channel yang dideklarasikan sebagai Channel(Int32) menolak nilai bertipe lain pada waktu kompilasi, sebelum program pernah dijalankan. Program berikut sengaja mengirim string ke channel bertipe Int32.

# tidak-dijalankan
channel = Channel(Int32).new
channel.send("teks")

# mengompilasi program salah tipe untuk melihat error compiler
crystal build salah-tipe.cr

Showing last frame. Use --error-trace for full trace.

In salah-tipe.cr:3:14

 3 | channel.send("teks")
                  ^
Error: expected argument #1 to 'Channel(Int32)#send' to be Int32, not String

Overloads are:
 - Channel(T)#send(value : T)

Error compiler menyebutkan secara eksplisit bahwa argumen pertama send harus Int32 dan bukan String. Sisi penerima mendapat manfaat yang sama, karena nilai kembalian receive bertipe sesuai deklarasi channel, yaitu Int32 untuk Channel(Int32). Pemeriksaan pada kedua sisi ini membuat ketidakcocokan tipe pada pengiriman pesan tertangkap sebelum program dijalankan. Deklarasi Channel(String) untuk pesan teks atau Channel({Int32, String}) untuk pesan yang menggabungkan dua tipe diperlakukan dengan aturan yang sama. Batas tipe antar fiber dengan demikian terjaga tanpa pemeriksaan manual di kedua sisinya.

Pola produsen-konsumen#

Pola produsen-konsumen (producer-consumer) adalah penerapan channel yang lazim, yaitu satu fiber memproduksi nilai dan fiber lain mengonsumsinya. Konsumen tidak perlu mengetahui jumlah nilai yang akan datang apabila produsen menutup channel setelah selesai mengirim. Perulangan dengan while dan receive? menerima nilai sampai nil, yaitu sampai channel ditutup.

channel = Channel(Int32).new

spawn do
  5.times do |i|
    channel.send(i * i)
  end
  channel.close
end

while nilai = channel.receive?
  puts "memproses #{nilai}"
end

puts "channel ditutup, konsumen berhenti"

# menjalankan program produsen konsumen
crystal run produsen-konsumen.cr

memproses 0
memproses 1
memproses 4
memproses 9
memproses 16
channel ditutup, konsumen berhenti

Produsen mengirim lima nilai kuadrat lalu menutup channel. Konsumen mencetak setiap nilai segera setelah diterima, dan perulangan berhenti ketika receive? mengembalikan nil setelah penutupan. Perulangan while nilai = channel.receive? menugaskan sekaligus menguji nilai yang diterima dalam satu baris, sehingga perulangan berhenti tepat ketika nil muncul. Keluaran bernilai pasti karena hanya ada satu produsen, sehingga urutan kedatangan nilai sama dengan urutan pengirimannya. Pola ini menjadi dasar bagi kasus dengan banyak produsen, yang urutan kedatangannya bergantung pada penjadwal.

Channel sebagai Iterator#

Sejak Crystal 1.21, Channel menyertakan modul Iterator, sehingga isi channel dapat diulang dengan metode perulangan yang biasa digunakan untuk koleksi. Metode each menerima nilai satu per satu dan berhenti ketika channel ditutup, sedangkan to_a mengumpulkan seluruh nilai menjadi sebuah array. Bentuk perulangan ini menggantikan while manual pada pola produsen-konsumen.

channel = Channel(Int32).new

spawn do
  {2, 4, 6}.each { |n| channel.send(n) }
  channel.close
end

channel.each do |nilai|
  puts "menerima #{nilai}"
end

puts "perulangan selesai"

# menjalankan program perulangan channel dengan each
crystal run channel-each.cr

menerima 2
menerima 4
menerima 6
perulangan selesai

Metode each menerima nilai selama channel masih terbuka dan berhenti dengan sendirinya setelah produsen menutup channel, tanpa pemeriksaan nil yang eksplisit. Perbedaannya dengan while pada bagian sebelumnya terletak pada penulisan. each menyembunyikan pemeriksaan nil di dalam metode, sehingga kode konsumen hanya berisi pekerjaan terhadap nilainya, sedangkan perilaku berhentinya tetap sama. Metode to_a bekerja dengan cara yang sama dan mengembalikan seluruh nilai sebagai array setelah channel ditutup. Kedua metode membawa kemudahan perulangan koleksi ke dalam aliran data antar fiber.

Kesalahan umum#

Tiga kesalahan berikut lazim muncul ketika channel baru pertama kali digunakan.

• Mengirim ke channel yang sudah ditutup. send pada channel yang telah ditutup juga melempar Channel::ClosedError dengan pesan Channel is closed. Tutup channel hanya dari sisi produsen, dan hanya setelah seluruh nilai terkirim.
• Program menggantung tanpa akhir. Pada channel tanpa buffer, send memblokir sampai ada penerima, sehingga pengiriman tanpa penerima menghentikan seluruh kemajuan program. Kondisi saling menunggu ini disebut deadlock. Pastikan ada fiber yang memanggil receive, atau berikan buffer dengan kapasitas yang cukup.
• Mengandalkan urutan keluaran dari banyak produsen. Nilai dari beberapa produsen pada satu channel tiba sesuai urutan yang dipilih penjadwal, bukan urutan tertentu. Sertakan identitas atau nomor urut di dalam pesan apabila urutan kedatangan perlu diketahui.

Ringkasan#

• Channel bertipe mengirim pesan antar fiber melalui send dan receive, dan tipenya diperiksa compiler sebelum program dijalankan.
• Channel tanpa buffer menyerahkan nilai langsung kepada penerima yang menunggu, sedangkan channel berbuffer menyimpan nilai hingga kapasitasnya.
• Setelah close, receive? mengembalikan nil sedangkan receive melempar Channel::ClosedError dengan pesan Channel is closed.
• Sejak Crystal 1.21, channel menjadi Iterator sehingga isinya dapat diulang dengan each atau dikumpulkan dengan to_a.

Referensi#

• Channel — Crystal API — dokumentasi kelas Channel beserta metode send, receive, dan close.
• Concurrency guide — Crystal — panduan resmi konkurensi Crystal yang memperkenalkan channel dan spawn.