KAL
Menu

Agent dan channel

Agent MailboxProcessor untuk pesan berurutan dan channel dengan buffer antar tugas.

Daftar isi

Koordinasi antartugas membutuhkan cara bertukar pesan yang tidak menimbulkan kebutuhan kunci (lock) manual. Topik ini membahas agent MailboxProcessor yang menerima pesan satu per satu, pola pertanyaan-jawaban dengan PostAndAsyncReply, serta channel berbuffer terbatas dari System.Threading.Channels untuk aliran data antara produsen dan konsumen. Contohnya mencakup penghitung kejadian sensor, antrean tiket loket, dan pemindahan data melalui buffer berkapasitas dua, semuanya dijalankan sebagai skrip yang keluarannya dapat direproduksi.

Agent MailboxProcessor#

MailboxProcessor membentuk agent, yaitu objek yang menerima pesan dari sebuah antrean bernama kotak surat (mailbox) dan memprosesnya satu per satu sesuai urutan datang. Badannya adalah workflow async yang berulang, dan state program berpindah dari satu pesan ke pesan berikutnya melalui parameter rekursinya. Karena hanya satu pesan yang diproses pada satu waktu, state internal tidak pernah diakses dua tugas secara serentak, sehingga tidak ada kunci yang perlu ditulis. Berkas berikut mencatat kejadian dari dua sensor dengan state berupa Map<string, int> yang dipindah dari satu pesan ke pesan berikutnya.

fsharp penghitung.fsx 
// Agent penghitung kejadian: jumlah per sensor berpindah dari pesan ke pesan.
type Pesan =
    | Catat of sensor: string
    | Lihat of AsyncReplyChannel<Map<string, int>>

let agen =
    MailboxProcessor.Start(fun kotak ->
        let rec loop jumlah =
            async {
                let! pesan = kotak.Receive()
                match pesan with
                | Catat sensor ->
                    let n = jumlah |> Map.tryFind sensor |> Option.defaultValue 0
                    return! loop (Map.add sensor (n + 1) jumlah)
                | Lihat balas ->
                    balas.Reply jumlah
                    return! loop jumlah
            }
        loop Map.empty)

agen.Post (Catat "sensor-a")
agen.Post (Catat "sensor-b")
agen.Post (Catat "sensor-a")
agen.Post (Catat "sensor-a")

let jumlah = Async.RunSynchronously (agen.PostAndAsyncReply Lihat)

Map.iter (fun sensor n -> printfn "%s: %d kejadian" sensor n) jumlah
text Keluaran 
sensor-a: 3 kejadian
sensor-b: 1 kejadian

Pesan Catat membawa nama sensor dan memperbarui map, sedangkan Lihat membawa AsyncReplyChannel yang dijawab dengan map terkini. Keluaran menampilkan dua baris sesuai urutan kunci map, bukan urutan kedatangan pesan, karena Map.iter mengiterasi kunci secara terurut. Empat pesan Catat dikirim tanpa menunggu, dan semuanya tetap tercatat karena pesan yang belum diproses menunggu di dalam kotak suratnya.

Pertanyaan-jawaban dengan PostAndAsyncReply#

Pola pertanyaan-jawaban dibangun dengan PostAndAsyncReply, yang mengirim pesan berisi kanal jawab (reply channel) dan mengembalikan Async<'Reply> untuk menunggu jawabannya. Pesannya dibentuk oleh fungsi yang diberikan pemanggil, umumnya kasus discriminated union dengan field bertipe AsyncReplyChannel<'T>, dan agent mengisi jawabannya dengan metode Reply pada kanal tersebut. Pola ini mengubah agent dari penerima pesan searah menjadi pelayan yang mengembalikan hasil untuk setiap permintaan. Berkas berikut menyimulasikan loket tiket yang memberi nomor urut kepada setiap pemohon.

fsharp antrean-tiket.fsx 
// Loket tiket: setiap pemohon menerima nomor urut dari agent secara berurutan.
type PesanLoket =
    | AmbilTiket of nama: string * balas: AsyncReplyChannel<int>

let loket =
    MailboxProcessor.Start(fun kotak ->
        let rec loop nomorBerikut =
            async {
                let! pesan = kotak.Receive()
                match pesan with
                | AmbilTiket (nama, balas) ->
                    balas.Reply nomorBerikut
                    return! loop (nomorBerikut + 1)
            }
        loop 1)

let mintaTiket nama =
    loket.PostAndAsyncReply(fun balas -> AmbilTiket (nama, balas))

let daftar = [ "Ali"; "Budi"; "Citra"; "Dewi" ]

let tiket =
    daftar
    |> List.map mintaTiket
    |> Async.Parallel
    |> Async.RunSynchronously

Array.iter2
    (fun nama nomor -> printfn "%s menerima tiket nomor %d" nama nomor)
    (Array.ofList daftar)
    tiket
text Keluaran 
Ali menerima tiket nomor 1
Budi menerima tiket nomor 2
Citra menerima tiket nomor 3
Dewi menerima tiket nomor 4

Keempat permintaan dikirim serentak melalui Async.Parallel, tetapi nomor tiket tetap 1 sampai 4 sesuai urutan kedatangan pesan karena agent memprosesnya satu per satu. Hasil Async.Parallel dirakit sesuai urutan pemohon, sehingga pasangan nama dan nomor tetap terpasang dengan benar pada saat dicetak. Kombinasi pemrosesan berurutan di agent dan perakitan berurutan di pemanggil menjaga keluarannya tetap deterministik tanpa satu kunci pun yang ditulis. Kanal jawab bersifat sekali pakai untuk satu permintaan, dan jawabannya tersimpan sampai pemanggil menunggu, sehingga agent tidak menumpahkan hasil kepada pihak yang belum siap. Bentuk PostAndReply menunggu jawaban secara langsung, sedangkan PostAndAsyncReply menerima batas waktu opsional untuk menjaga agar pemanggil tidak menunggu tanpa ujung.

Buffer terbatas antartugas#

Channel adalah saluran data satu arah antara produsen dan konsumen dengan buffer yang dapat dibatasi kapasitasnya. Namespace System.Threading.Channels sudah tersedia pada .NET 10 tanpa paket tambahan, sehingga Channel.CreateBounded langsung dapat digunakan di fsi. Pada kanal berkapasitas dua, panggilan WriteAsync menunggu setiap kali buffer penuh sampai konsumen mengambil satu item, sehingga produsen yang lebih cepat otomatis tertahan mengikuti laju konsumen. Mekanisme tertahan ini disebut backpressure, yaitu tekanan yang mengalir mundur dari konsumen ke produsen. Berkas berikut memindahkan empat nama melalui kanal berkapasitas dua.

fsharp kanal-buffer.fsx 
// Kanal berkapasitas dua: penulisan menunggu setiap kali buffer penuh.
open System.Threading.Channels
open System.Threading.Tasks

let kanal = Channel.CreateBounded<string>(2)

let produsen =
    task {
        for nama in [ "Ali"; "Budi"; "Citra"; "Dewi" ] do
            do! kanal.Writer.WriteAsync nama
        kanal.Writer.Complete()
    }

let konsumen =
    task {
        let hasil = ResizeArray<string>()
        let mutable berjalan = true
        while berjalan do
            let! ada = kanal.Reader.WaitToReadAsync()
            if ada then
                let mutable adaItem = true
                while adaItem do
                    match kanal.Reader.TryRead() with
                    | true, nama -> hasil.Add nama
                    | false, _ -> adaItem <- false
            else
                berjalan <- false
        return List.ofSeq hasil
    }

printfn "urutan terbaca: %A" konsumen.Result
text Keluaran 
urutan terbaca: ["Ali"; "Budi"; "Citra"; "Dewi"]

Produsen menulis empat nama lalu menandai kanal selesai dengan Complete, sedangkan konsumen membaca dengan WaitToReadAsync dan TryRead sampai kanal kosong dan selesai. Keluaran memperlihatkan urutan baca yang sama dengan urutan tulis karena kanal mengantre item secara first in, first out. Alternatif yang lebih lama dengan perilaku serupa adalah BlockingCollection dari System.Collections.Concurrent, yang menunggu dengan memblokir thread, sedangkan Channel menunggu secara asinkron tanpa menahan thread.

Memilih antara agent dan channel#

Agent dipilih ketika ada state yang harus diperbarui satu per satu, seperti penghitung kejadian, antrean tiket loket, atau pembatas akses ke satu sumber daya. Channel dipilih ketika persoalannya adalah aliran data dari pihak yang menghasilkan ke pihak yang memproses, terutama apabila lajunya perlu dibatasi oleh kapasitas buffer. Keduanya dapat dipadukan, misalnya agent yang membaca item dari sebuah channel lalu memperbarui state-nya untuk setiap item yang masuk. Kesamaan keduanya terletak pada komunikasi yang berjalan berurutan pada satu titik, sehingga state yang dilindungi tidak diakses oleh dua tugas secara serentak dan tidak ada kunci manual yang perlu dipelihara. Titik komunikasinya juga mudah diperiksa, karena seluruh pergantian state dan pengiriman data melewati satu jenis pesan atau satu kanal yang tipenya nyata pada kode.

Ringkasan#

  • MailboxProcessor membentuk agent yang memproses pesan satu per satu dari kotak suratnya, sehingga state internal aman tanpa kunci manual.
  • PostAndAsyncReply melengkapi agent dengan pola pertanyaan-jawaban melalui AsyncReplyChannel, termasuk ketika banyak pemohon dikirim bersamaan lewat Async.Parallel.
  • Channel.CreateBounded menyediakan buffer terbatas yang menahan produsen saat penuh, dengan urutan baca yang mengikuti urutan tulis.

Referensi#