core.async

Topik ini membahas core.async, yaitu pustaka konkurensi Clojure yang mempertukarkan nilai antar bagian program melalui channel. Pembahasan mencakup pembuatan channel, operasi asinkron put! dan take!, operasi blocking >!! dan <!!, go block dengan operasi parking >! dan <!, serta buffer yang mengatur perilaku saat channel penuh. Setelah topik ini selesai, pola produksi dan konsumsi dapat ditulis secara deterministik, dan perbedaan antara parking dan blocking dapat ditegaskan dengan tepat.

Channel sebagai antrean nilai#

Channel adalah antrean thread-safe yang pengirim dan penerimanya tidak saling mengenal. Channel dibuat dengan chan, diisi dengan operasi pengiriman, dan dikosongkan dengan operasi pengambilan. Bentuk paling sederhana adalah put! dan take!, yaitu operasi asinkron yang segera kembali dan melaporkan hasilnya melalui callback. Channel tanpa buffer menahan nilai sampai ada penerima yang siap, sehingga pengiriman dan pengambilan bertemu seperti serah terima di loket.

Panggilan close! menandai bahwa ujung pengirim sudah selesai. Nilai yang masih tertinggal di dalam channel tetap dapat diambil, sedangkan pengambilan setelah channel benar-benar kosong menghasilkan nil. Nilai nil dengan demikian bukan data biasa pada channel, melainkan penanda bahwa ujungnya sudah ditutup.

Namespace clojure.core.async tidak disertakan dalam berkas jar Clojure, melainkan tersedia sebagai pustaka tersendiri dengan koordinat org.clojure/core.async. Dependensinya dinyatakan pada berkas deps.edn berikut, dan berkas yang sama menjadi bagian bundle topik ini. Pemanggilan clojure -M berkas.clj dari dalam direktori topik dengan demikian sudah membawa core.async pada classpath tanpa penyetelan tambahan.

{:deps {org.clojure/core.async {:mvn/version "1.9.865"}}}

(require '[clojure.core.async :as a :refer [chan put! take! close!]])

(def kanal (chan))

(put! kanal "tiket A")
(put! kanal "tiket B")
(close! kanal)

(take! kanal (fn [nilai] (println "diterima:" nilai)))
(take! kanal (fn [nilai]
               (println "diterima:" nilai)
               (System/exit 0)))

diterima: tiket A
diterima: tiket B

Dua nilai dimasukkan sebelum channel ditutup, dan kedua callback take! menghabiskannya sesuai urutan serah terima. Panggilan (System/exit 0) di dalam callback terakhir menghentikan JVM, karena thread pool milik core.async bukan daemon dan program akan terus berjalan tanpa perintah itu. Penempatan pada callback terakhir juga menjamin seluruh pencetakan sempat dijalankan.

Operasi blocking >!! dan <!!#

Operator >!! dan <!! berbeda dari put! dan take! dalam satu hal, yaitu thread yang memanggilnya menunggu sampai operasi selesai. Pemanggil >!! membeku sampai nilai diterima, sedangkan pemanggil <!! membeku sampai nilai tersedia. Kata blocking digunakan dalam arti yang paling harfiah, yaitu thread asli berhenti sepenuhnya dan tidak dapat menjalankan pekerjaan lain.

Blocking menjadi wajar bila channel dilengkapi buffer. Channel (chan 2) menampung dua nilai sebelum pengirimnya perlu menunggu, sehingga pengiriman singkat tidak membekukan thread yang mengirimnya. Contoh berikut menggunakan buffer berkapasitas dua, membaca isinya, lalu memperlihatkan hasil pengambilan setelah channel ditutup.

(require '[clojure.core.async :as a :refer [chan >!! <!! close!]])

(def kanal (chan 2))

(>!! kanal "nomor 1")
(>!! kanal "nomor 2")
(println (<!! kanal))
(println (<!! kanal))
(close! kanal)
(println (<!! kanal))
(System/exit 0)

nomor 1
nomor 2
nil

Dua pengiriman pertama segera selesai karena masuk ke buffer. Baris nil pada akhir keluaran memperlihatkan semantik close!: pengambilan atas channel yang kosong dan tertutup mengembalikan nil tanpa menunggu apa pun. Apabila buffer penuh dan tidak ada penerima, >!! akan membeku sampai satu nilai diambil, dan perilaku itulah yang membedakannya dari >!.

go block dan parking#

go block adalah blok kode yang dikompilasi core.async menjadi mesin keadaan, yaitu rangkaian langkah yang dapat dihentikan dan dilanjutkan kembali. Di dalamnya, operator >! dan <! bekerja seperti padanan blocking-nya, tetapi yang berhenti bukan thread, melainkan langkah eksekusi go block. Go thread yang dipinjam dari dispatch pool segera kembali ke pool, dan eksekusi dilanjutkan pada thread mana pun ketika operasi sudah bisa selesai. Istilahnya adalah parking, yakni memarkir kelanjutan kode tanpa menyandera thread sungguhan.

Contoh berikut membentuk pola produksi dan konsumsi. Go pertama memproduksi tiga nilai lalu menutup channel, go kedua mengumpulkan seluruh nilai dengan perulangan recur sampai <! menghasilkan nil, dan thread utama membaca hasil akhir dengan <!! karena benar-benar perlu menunggu.

(require '[clojure.core.async :as a :refer [chan go >! <! <!! close!]])

(def kanal (chan))

(go
  (>! kanal "satu")
  (>! kanal "dua")
  (>! kanal "tiga")
  (close! kanal))

(def hasil
  (go
    (loop [terkumpul []]
      (if-let [nilai (<! kanal)]
        (recur (conj terkumpul nilai))
        terkumpul))))

(prn (<!! hasil))
(System/exit 0)

["satu" "dua" "tiga"]

Go block mengembalikan channel, sehingga nilai akhir perulangan dibaca dengan <!! pada thread utama. Ketiga nilai tiba dalam urutan produksinya karena kedua go block bertukar nilai melalui channel yang sama, dan pengumpulan berhenti dengan rapi ketika channel ditutup.

Buffer channel#

Buffer memutus ketergantungan langsung antara pengirim dan penerima. Selain buffer tetap (chan n), core.async menyediakan sliding-buffer yang membuang nilai tertua bila penuh, serta dropping-buffer yang membuang nilai baru bila penuh. Keduanya menjamin pengiriman tidak pernah membekukan pengirim, dengan konsekuensi sebagian nilai tidak pernah sampai. Pilihan buffer mengikuti kebutuhan. Catatan sensor suhu yang mengutamakan nilai terbaru, misalnya, cocok dengan sliding-buffer.

(require '[clojure.core.async :as a :refer [chan >!! <!!]])

(def muat-terus (chan (a/sliding-buffer 3)))

(>!! muat-terus 10)
(>!! muat-terus 20)
(>!! muat-terus 30)
(>!! muat-terus 40)

(prn (<!! muat-terus))
(prn (<!! muat-terus))
(prn (<!! muat-terus))
(System/exit 0)

20
30
40

Nilai 10 terbuang begitu 40 tiba, karena buffer hanya menampung tiga nilai terakhir. Untuk pemrosesan yang menuntut seluruh nilai diproses tanpa kehilangan, fungsi pipeline meneruskan isi satu channel ke channel lain melalui fungsi transduksi dengan tingkat paralelisme yang dapat diatur. Dengan demikian, perulangan konsumsi seperti pada contoh go block tidak perlu ditulis ulang untuk setiap kebutuhan.

Parking versus blocking#

Perbedaan parking dan blocking adalah inti penggunaan core.async yang benar. Blocking membekukan thread asli, dan thread itu tidak dapat digunakan siapa pun sampai operasi selesai. Parking menghentikan kelanjutan kode go block tanpa menyandera thread, sedangkan thread yang dipinjamnya kembali ke dispatch pool dan mengerjakan go block lain. Tabel berikut merangkum ketiga kelompok operasi.

Aturan praktisnya mengikuti tempat kode dijalankan. Di dalam go block, gunakan >! dan <! sehingga parkir tidak menyandera thread. Di thread biasa, termasuk thread utama program, <!! sah digunakan untuk menunggu hasil, sedangkan >!! cocok pada thread produksi yang memang khusus menulis. Perpaduan pada contoh produksi dan konsumsi memperlihatkan pola ini, yaitu parking pada kedua go block dan satu pembacaan blocking pada thread utama.

Kesalahan umum#

Memanggil >!! atau <!! di dalam go block. Operasi blocking membekukan thread dari dispatch pool yang jumlahnya sedikit, sehingga seluruh go block lain ikut kehabisan penggerak. Perbaikannya dengan menggunakan >! dan <! di dalam go block, dan membatasi operasi blocking hanya untuk thread biasa seperti thread utama program.

Memperlakukan nil sebagai nilai data. Pengiriman nil ke channel melempar error karena nil berarti channel ditutup, dan pembacaan <!! yang menghasilkan nil sering salah ditafsirkan sebagai data kosong. Perbaikannya dengan menjadikan nil hanya sebagai sinyal akhir, dan menguji keluaran <!! sebelum diproses.

Mengabaikan penutupan channel. Sisi produksi yang tidak pernah memanggil close! membuat sisi konsumsi menunggu selamanya pada <!! atau <!. Perbaikannya dengan menutup channel segera setelah produksi selesai, sebagaimana go block pada contoh produksi dan konsumsi yang memanggil close! sebagai langkah terakhirnya.

Ringkasan#

• Channel adalah antrean thread-safe yang diisi dan dikosongkan melalui tiga kelompok operasi, yaitu asinkron put! dan take!, blocking >!! dan <!!, serta parking >! dan <! di dalam go block.
• close! menandai ujung produksi selesai, nilai tersisa tetap dapat diambil, dan pengambilan atas channel kosong yang tertutup menghasilkan nil.
• go block dikompilasi menjadi mesin keadaan pada dispatch pool, sehingga parking tidak menyandera thread sungguhan seperti blocking.
• Buffer tetap, sliding-buffer, dan dropping-buffer mengatur perilaku channel penuh, sedangkan pipeline meneruskan nilai antar channel dengan paralelisme yang diatur.

Referensi#

• core.async — Clojure — Repositori resmi core.async beserta dokumentasi channel dan go block.
• Walkthrough of core.async — Clojure — Contoh beranotasi dari penggunaan channel, buffer, dan operasinya.