Timeout dan mailbox

Mailbox adalah antrean tempat pesan menunggu hingga dibaca oleh receive, dan hampir semua kejanggalan program konkurensi bersumber dari cara antrean ini diperlakukan. Topik ini membahas perilaku pemilihan pesan oleh receive, pesan yang tertinggal di mailbox atau berlaku usang (message drift), pembacaan defensif dengan after 0, serta mailbox yang tumbuh tanpa batas. Hasil akhirnya adalah kemampuan membaca dan membersihkan mailbox dengan sengaja, alih-alih menganggap pesan yang datang lebih dahulu pasti dibaca lebih dahulu.

Mailbox sebagai antrean pesan#

Setiap process memiliki satu mailbox yang tidak terlihat dalam kodenya. Pesan yang dikirim send/2 diletakkan di ujung belakang antrean. BEAM menjamin urutan ini untuk setiap pasangan pengirim dan penerima: dua pesan dari pengirim yang sama menuju penerima yang sama ditambahkan ke mailbox dalam urutan asli pengirimannya. Antrean ini tidak memiliki batas ukuran, sehingga jumlah pesan yang dapat menunggu hanya dibatasi oleh memori yang tersedia.

Mailbox ada sepanjang umur processnya, dan hanya process pemiliknya yang dapat membacanya. Process lain hanya dapat mengirim pesan, sedangkan pembacaan antrean sepenuhnya milik process pemiliknya. Tidak ada pula pembersih otomatis yang membuang pesan lama, sehingga apa pun yang dibiarkan tanpa dibaca tetap tersimpan sampai processnya berhenti. Karena itu, pembersihan antrean selalu menjadi tugas kode, bukan pekerjaan BEAM.

Panjang antrean dapat diperiksa kapan pun melalui Process.info/2 dengan kunci :message_queue_len. Cara ini berguna untuk memastikan mailbox benar-benar kosong setelah serangkaian pengiriman:

send(self(), :pesan_satu)
send(self(), :pesan_dua)
IO.inspect(Process.info(self(), :message_queue_len), label: "panjang antrean mailbox")

panjang antrean mailbox: {:message_queue_len, 2}

Seleksi pesan dan pesan yang tertinggal#

receive/1 tidak selalu membaca pesan tertua. Saat dieksekusi, mailbox dipindai dari pesan tertua ke termuda, lalu pesan pertama yang cocok dengan salah satu pola klausa diambil keluar. Pesan-pesan yang tidak cocok dilewati dan tetap menunggu di mailbox dengan posisi urutannya terjaga. Perilaku ini disebut pemilihan pesan (selective receive) dan menjadi dasar penerapan prioritas: sebuah receive yang hanya mencocokkan pesan penting melewati pesan biasa yang datang lebih dahulu.

send(self(), {:tugas, :rendah})
send(self(), {:tugas, :tinggi})

receive do
  {:tugas, :tinggi} -> IO.puts("dipilih lebih dahulu: {:tugas, :tinggi}")
after
  100 -> IO.puts("tidak ada tugas tinggi")
end

dipilih lebih dahulu: {:tugas, :tinggi}

Pesan {:tugas, :rendah} pada contoh di atas tidak hilang, tetapi tertinggal di mailbox. Kondisi pesan yang dibiarkan lama inilah yang disebut drift pesan. Penyebabnya hampir selalu dua hal, yaitu pesan yang polanya tidak pernah dicocokkan oleh receive mana pun, atau pesan yang isinya sudah tidak relevan ketika akhirnya dibaca, misalnya hasil perhitungan yang diminta jauh sebelumnya. Karena pesan tertinggal tetap dibaca pada pencocokan berikutnya, drift pesan sering muncul sebagai balasan yang tertukar antarpermintaan tanpa ada error yang tercetak.

Pembacaan defensif dengan after 0#

Pola after 0 mengubah receive menjadi pemeriksaan instan: mailbox dipindai tanpa menunggu, dan apabila tidak ada pesan yang cocok, tubuh after langsung dijalankan. Dengan pola ini, mailbox dapat dikosongkan secara sengaja, sebuah kebiasaan yang disebut pembacaan defensif (defensive read). Skrip berikut menunjukkan pesan yang tertinggal, pembacaan defensif atas sisa mailbox, dan pembacaan kedua yang membuktikan antrean sudah kosong:

# mailbox.exs — pesan tertinggal, pembacaan defensif, dan pembersihan mailbox
defmodule Mailbox do
  # mengumpulkan seluruh sisa pesan hingga mailbox kosong, lalu mengembalikannya
  def kumpulkan(terkumpul) do
    receive do
      pesan -> kumpulkan([pesan | terkumpul])
    after
      0 -> Enum.reverse(terkumpul)
    end
  end
end

# tiga pesan masuk berurutan ke mailbox process utama
send(self(), :pesan_lama)
send(self(), :peringatan)
send(self(), :hasil)

# receive memilih pesan yang cocok dengan polanya, bukan pesan yang paling awal
receive do
  :hasil -> IO.puts("dibaca lebih dahulu: :hasil")
after
  1000 -> IO.puts("waktu habis")
end

# dua pesan lain masih tertinggal di mailbox, lalu dibaca dengan after 0
IO.inspect(Mailbox.kumpulkan([]), label: "sisa pesan di mailbox")

# mailbox sudah kosong, pembacaan defensif tidak menemukan apa pun
IO.inspect(Mailbox.kumpulkan([]), label: "pembacaan kedua")

dibaca lebih dahulu: :hasil
sisa pesan di mailbox: [:pesan_lama, :peringatan]
pembacaan kedua: []

Fungsi Mailbox.kumpulkan/1 bekerja dengan menumpuk setiap pesan yang ditemukan, lalu berhenti saat after 0 tidak lagi menemukan apa pun. Daftar hasil dibalik dengan Enum.reverse/1 agar urutannya sama dengan urutan kedatangan pesan. Pembacaan kedua menghasilkan daftar kosong, dan itulah tanda mailbox dalam keadaan bersih. Pola ini digunakan sebelum memulai sesi kerja baru, setelah batas waktu tercapai, atau pada titik ketika pesan lama tidak lagi bermakna.

Timeout sebagai pengaman#

Klausa after pada receive menentukan lama menunggu dalam milidetik. Apabila tidak ada pesan yang cocok hingga batas itu tercapai, tubuh after dijalankan dan receive selesai tanpa mengambil pesan apa pun. Nilai after 0 berarti memeriksa lalu melanjutkan, sedangkan atom :infinity menandakan penungguan tanpa batas, yang perilakunya sama dengan receive tanpa klausa after:

receive do
  :tidak_akan_datang -> IO.puts("pesan diterima")
after
  300 -> IO.puts("waktu tunggu 300 milidetik habis")
end

waktu tunggu 300 milidetik habis

Batas waktu juga menjadi cara paling sederhana untuk menangani drift pesan. Apabila balasan yang diharapkan tidak tiba tepat waktu, receive menghasilkan nilai seperti :waktu_habis, dan sisa pesan dapat dibersihkan dengan after 0 sebelum permintaan berikutnya dikirim. Dengan demikian, kegagalan satu permintaan tidak memengaruhi permintaan lain. Pendekatan yang lebih teliti menempelkan penanda unik pada setiap permintaan dan hanya menerima balasan yang penandanya cocok, sehingga balasan lain tetap tertinggal dan tidak tertukar.

Mailbox yang tumbuh tanpa batas#

Mailbox tidak memberi peringatan ketika isinya menumpuk. Apabila laju pengiriman terus melebihi laju pembacaan, antrean tumbuh tanpa batas, konsumsi memori ikut naik, dan process penerima semakin lambat karena setiap pemindaian receive harus melewati pesan yang menumpuk. Kondisi ini jarang memunculkan error yang jelas, padahal akar masalahnya adalah penerima yang tidak mampu mengimbangi pengirimnya.

Penanganannya bertumpu pada kebiasaan membaca. Penerima yang membaca mailbox secara rutin, termasuk membaca pesan yang tak dikenal untuk dibuang, menjaga antrean tetap pendek. Pengiriman yang berlebihan dapat dikendalikan dengan mengatur laju di sisi pengirim, dan pesan yang mengandung penanda waktu dapat dibuang melalui pembacaan defensif apabila usang. Pemeriksaan :message_queue_len secara berkala menjadi alat deteksi dini sebelum antrean benar-benar membengkak.

Ringkasan#

• Mailbox setiap process adalah antrean tanpa batas ukuran, dan BEAM menjamin urutan pesan untuk setiap pasangan pengirim dan penerima.
• receive memilih pesan pertama yang cocok dengan polanya, sehingga pesan yang tidak cocok tertinggal di mailbox dan dapat menimbulkan drift pesan.
• Pola after 0 menjadikan receive sebagai pemeriksaan instan untuk membaca dan membersihkan sisa mailbox secara defensif.
• Antrean yang terus tumbuh melemahkan process penerima, sehingga laju pembacaan, pembacaan defensif, dan pemeriksaan :message_queue_len digunakan bersama sebagai pengaman antrean.

Referensi#

• Processes — Erlang Reference Manual — spesifikasi antrean pesan dan pemilihan pesan pada level Erlang.
• Kernel — Hexdocs Elixir — dokumentasi bentuk khusus receive beserta klausa after.
• Modul Process — Hexdocs Elixir — dokumentasi Process.info/2 untuk pemeriksaan panjang antrean.