Mengimplementasikan Bounded Blocking Queue yang Thread-Safe

Masalah dan Kapan Masalah Ini Berlaku

Implementasikan BoundedBlockingQueue<E>. Konstruktornya menerima kapasitas positif. put(item) menambahkan elemen dalam urutan FIFO dan menunggu saat antrean penuh. take() menghapus elemen dari kepala antrean dan menunggu saat antrean kosong. Beberapa produsen dan konsumen dapat memanggil kedua metode secara bersamaan (concurrently). Tidak ada elemen yang boleh hilang, dikembalikan dua kali, atau dikembalikan setelah elemen yang masuk lebih awal. Jika sebuah thread diinterupsi saat memperoleh lock atau saat menunggu condition, metode tersebut akan melempar InterruptedException.

Latihan ini tidak mengizinkan pembungkusan (wrapping) ArrayBlockingQueue. API dasar tidak mencakup offer/poll non-blocking, batas waktu (timeouts), penghapusan massal (bulk removal), dan semantik shutdown, serta tidak menjanjikan keadilan ketat (strict fairness) di antara thread yang menunggu. Hal-hal tersebut merupakan pertanyaan lanjutan. Seperti BlockingQueue milik Java, implementasi ini menolak null, karena API antrean sering menggunakan null untuk menandakan bahwa tidak ada elemen yang tersedia.

Ini adalah masalah pengkodean struktur data konkuren. Pengindeksan array adalah bagian yang mudah. Ujian sebenarnya adalah apakah kandidat dapat menyatakan kontrak yang dapat diaudit: sinkronisasi mana yang melindungi shared state, mengapa notifikasi tidak dapat hilang, mengapa thread yang dibangunkan harus memeriksa ulang kondisinya, dan pada saat kapan suatu operasi berlaku bagi thread lain.

Apa yang Dievaluasi oleh Pewawancara

Sinyal pertama adalah apakah kandidat memisahkan keamanan (safety) dari kelangsungan hidup (liveness). Keamanan mengharuskan ukuran tetap berada dalam rentang [0, capacity], setiap elemen dihapus paling banyak satu kali, dan urutan penghapusan sesuai dengan urutan penyisipan. Kelangsungan hidup mengharuskan produsen mendapatkan kesempatan untuk melanjutkan ketika antrean penuh menjadi tidak penuh, dan konsumen mendapatkan kesempatan ketika antrean kosong menjadi tidak kosong. Interupsi juga harus dapat membatalkan proses tunggu.

Sinyal kedua adalah apakah primitif sinkronisasi sesuai dengan predikat state. Satu lock melindungi array, head, tail, dan size, sehingga pemeriksaan kondisi dan perubahan state terjadi di critical section yang sama. notFull mewakili size < capacity; notEmpty mewakili size > 0. Produsen hanya menunggu kondisi pertama, konsumen hanya menunggu kondisi kedua, dan perubahan state yang melewati batas memberi sinyal kepada peran yang berlawanan.

Sinyal ketiga adalah apakah kandidat dapat menjelaskan while alih-alih menggunakannya sebagai pola hafalan. Condition memungkinkan terjadinya spurious wakeup. Bahkan setelah signal yang nyata, thread pesaing lainnya mungkin memperoleh lock terlebih dahulu dan mengisi atau mengosongkan antrean lagi. Begitu thread yang terbangun memperoleh lock kembali, ia harus menguji ulang predikat tersebut. Menerima notifikasi tidak membuktikan bahwa kondisinya masih terpenuhi.

Terakhir, pewawancara dapat mendalami desain: bagaimana indeks ring mempertahankan FIFO, mengapa pembaruan ukuran termasuk dalam linearization point, mengapa satu lock menghindari deadlock urutan lock, kapan signal sudah cukup, dan mengapa batas waktu serta shutdown memerlukan semantik API baru.

Pertanyaan untuk Diklarifikasi Sebelum Menjawab

• Kapasitas dan elemen mana yang valid? Kapasitas harus lebih besar dari nol, dan elemen null ditolak.
• Apakah proses tunggu saat penuh atau kosong harus melakukan spin? Tidak. Thread yang menunggu harus melepaskan lock dan menunggu pada sebuah condition. Thread tidak boleh memakan CPU dalam loop spin maupun tidur sambil tetap menahan lock.
• Bagaimana perilaku interupsi yang diharapkan? Baik put maupun take meneruskan InterruptedException. Keduanya tidak menelan interupsi atau mengubah antrean setelah operasi yang diinterupsi gagal.
• Apakah keadilan ketat (strict fairness) diperlukan? Tidak dalam masalah dasar. ReentrantLock non-fair default dapat membiarkan thread yang datang belakangan memperoleh lock terlebih dahulu. Lock yang adil (fair lock) mengubah throughput dan perilaku penjadwalan.
• Apakah antrean memerlukan shutdown? Tidak dalam masalah dasar. Jika ya, definisikan apakah item yang ada boleh dikosongkan (drain), apakah thread yang menunggu menerima exception atau hasil khusus, dan siapa yang membangunkan semua thread yang menunggu.
• Apakah FIFO elemen yang terlinearisasi dan FIFO waiter merupakan jaminan yang sama? Tidak. Elemen keluar dalam urutan linearisasi dari operasi put yang berhasil. Itu tidak berarti produsen atau konsumen yang terblokir diterima sesuai urutan kedatangan.
• Bolehkah kelas pustaka standar digunakan? Kode produksi biasanya harus lebih memilih ArrayBlockingQueue yang teruji. Implementasi manual di sini secara khusus bertujuan untuk mengevaluasi invariansi konkurensi dan semantik condition-wait.

Kerangka Jawaban 30 Detik

“Saya akan menggunakan fixed array sebagai ring buffer, dengan head, tail, dan size yang mewakili posisi baca berikutnya, posisi tulis berikutnya, dan jumlah elemen saat ini. Satu ReentrantLock melindungi semua shared state, dan dua objek Condition mewakili kondisi tidak-kosong dan tidak-penuh. put menunggu di dalam while (size == capacity), menyisipkan dan menambahkan size, lalu memberi sinyal ke satu konsumen. take secara simetris menunggu kondisi tidak-kosong, mengosongkan head dan mengurangi size, lalu memberi sinyal ke satu produsen. Setiap pemeriksaan dan transisi terjadi di bawah lock yang sama. Karena await secara atomik melepaskan lock dan memperolehnya kembali sebelum kembali, tidak ada celah kehilangan notifikasi (lost-notification window) antara pemeriksaan dan tidur. Setiap operasi yang berhasil memiliki kompleksitas O(1), dengan ruang O(capacity).”

Pembahasan Mendalam Langkah demi Langkah

Turunkan representasi dari bottleneck pendekatan naif. Array biasa yang menggeser elemen-elemen yang tersisa setelah setiap penghapusan membuat take menjadi O(n). Mempertahankan indeks baca yang hanya terus bertambah justru membuang prefix yang telah dibebaskan. Ring array tetap menggunakan kembali slot yang dibebaskan: head menunjuk ke posisi baca berikutnya, tail menunjuk ke posisi tulis berikutnya, dan size adalah jumlah elemen saat ini. Indeks kembali ke nol setelah mencapai akhir, sehingga baik penyisipan maupun penghapusan tidak perlu menggeser elemen yang ada.

Implementasi ini mempertahankan empat invariansi:

1. 0 <= size <= items.length.
2. Dimulai dari head, size slot pertama dalam urutan ring berisi urutan FIFO yang belum dikonsumsi.
3. tail == (head + size) % items.length. Saat antrean penuh, head == tail, sehingga size membedakan kondisi penuh dari kosong.
4. Setiap pembacaan atau penulisan items, head, tail, dan size terjadi saat memegang lock yang sama.

Berikut adalah implementasi intinya:

import java.util.Objects;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public final class BoundedBlockingQueue<E> {
  private final Object[] items;
  private final ReentrantLock lock = new ReentrantLock();
  private final Condition notEmpty = lock.newCondition();
  private final Condition notFull = lock.newCondition();

  private int head;
  private int tail;
  private int size;

  public BoundedBlockingQueue(int capacity) {
    if (capacity <= 0) {
      throw new IllegalArgumentException("capacity must be positive");
    }
    items = new Object[capacity];
  }

  public void put(E item) throws InterruptedException {
    Objects.requireNonNull(item, "item");
    lock.lockInterruptibly();
    try {
      while (size == items.length) {
        notFull.await();
      }

      items[tail] = item;
      tail = (tail + 1) % items.length;
      size++;
      notEmpty.signal();
    } finally {
      lock.unlock();
    }
  }

  @SuppressWarnings("unchecked")
  public E take() throws InterruptedException {
    lock.lockInterruptibly();
    try {
      while (size == 0) {
        notEmpty.await();
      }

      E item = (E) items[head];
      items[head] = null;
      head = (head + 1) % items.length;
      size--;
      notFull.signal();
      return item;
    } finally {
      lock.unlock();
    }
  }
}

Objects.requireNonNull berjalan sebelum locking karena hanya memeriksa argumen dan tidak bergantung pada shared state. lockInterruptibly() membuat proses menunggu untuk memperoleh lock itu sendiri dapat diinterupsi. Setelah metode memasuki blok try, blok finally melepaskan lock yang dipegang oleh thread saat ini saat pengembalian normal, interupsi selama condition wait, atau exception runtime.

Jaminan penting dari await() adalah bahwa metode ini secara atomik melepaskan lock terkait dan menunggu, lalu memperoleh kembali lock tersebut sebelum kembali. Produsen tidak boleh membuka lock secara manual lalu baru mendaftarkan dirinya sebagai waiter. Hal itu menciptakan celah di mana konsumen dapat membebaskan ruang dan mengirim notifikasi sebelum produsen benar-benar mulai menunggu, yang menyebabkan hilangnya notifikasi secara permanen. Condition variable menggabungkan “lepas dan mulai menunggu” menjadi satu aksi sinkronisasi dan menutup celah tersebut.

while menangani dua kasus berbeda. Pertama adalah spurious wakeup yang diizinkan oleh spesifikasi. Kasus kedua adalah persaingan biasa: dua konsumen mungkin dibangunkan, satu memperoleh lock terlebih dahulu dan menghapus satu-satunya elemen, dan konsumen kedua mendapati antrean kosong ketika akhirnya mendapatkan lock. Notifikasi condition hanya menyatakan bahwa state mungkin telah berubah; predikat state adalah otoritas mutlak untuk melanjutkan proses.

Linearization point dari put yang berhasil adalah transisi terkunci yang menambahkan elemen dan mengubah size dari k menjadi k + 1. Untuk take, poin tersebut adalah penghapusan terkait dan transisi dari k menjadi k - 1. Lock memastikan thread lain dapat mengamati state lengkap sebelum transisi atau state lengkap setelahnya, dan tidak akan pernah mengamati penulisan array tanpa pembaruan ukuran yang sesuai. Membuka lock dan kemudian memperoleh lock yang sama juga menetapkan visibilitas memori, mencocokkan tujuan happens-before yang ditentukan untuk meneruskan elemen melalui antrean konkuren standar.

Mengapa menggunakan dua condition? Dengan satu waiting set, sebuah take mungkin membangunkan konsumen lain meskipun antrean tetap kosong, sementara produsen yang dapat menggunakan slot baru tersebut tetap tertidur. Memisahkan notEmpty dan notFull memungkinkan setiap transisi hanya memberi tahu peran yang kini dapat melanjutkan eksekusi. Operasi tunggal dasar put atau take hanya membuat satu elemen atau slot baru, sehingga signal() sudah cukup; thread yang dibangunkan tetap memeriksa ulang di dalam while. Jika satu operasi mengubah beberapa slot, atau shutdown mengharuskan setiap waiter mengamati state baru, pertimbangkan kembali signalAll().

Kebenaran dibuktikan melalui induksi pada invariansi. Awalnya, head = tail = size = 0. Penyisipan hanya berjalan ketika size < capacity; operasi ini menulis di tail, memajukan tail, dan menambah size tepat satu kali, sehingga tetap berada dalam kapasitas dan menambahkan elemen setelah semua elemen yang belum dikonsumsi. Penghapusan hanya berjalan ketika size > 0; operasi ini membaca di head, mengosongkan slot, memajukan head, dan mengurangi size tepat satu kali, sehingga mengembalikan elemen paling awal yang belum dikonsumsi. Lock menserialisasi transisi ini, membuat setiap penjadwalan dari banyak produsen dan konsumen setara dengan suatu eksekusi sekuensial yang legal.

Setiap operasi put atau take yang berhasil melakukan sejumlah operasi array, integer, dan sinkronisasi yang tetap, sehingga beban kerjanya adalah O(1) tidak termasuk waktu tunggu. Fixed array menggunakan ruang O(capacity). Di bawah kondisi contention tinggi, waktu tunggu lock dan context switch dapat mendominasi latensi; notasi asimptotik tidak menangkap biaya tersebut. Desain satu lock tidak memiliki siklus multi-lock, tetapi pemanggil masih dapat menciptakan masalah lock-order yang lebih besar jika mereka memanggil metode pemblokiran sambil menahan lock lainnya.

Pengujian tidak boleh berhenti pada contoh single-threaded saja. Cakup pengujian pergantian penuh/kosong pada kapasitas satu, beberapa kali ring wraparound, elemen berulang yang sama disimpan dan dihapus secara terpisah, string kosong diterima sementara null ditolak, nilai-nilai unik yang diproduksi oleh beberapa produsen dan dikosongkan oleh beberapa konsumen, tidak ada nilai yang hilang atau duplikat pada set akhir, urutan yang dipertahankan dalam setiap stream produsen individual, serta interupsi pada operasi put yang terblokir maupun take yang terblokir. Kapasitas yang sangat besar akan mengalokasikan array yang sama besarnya di konstruktor dan dapat gagal karena kekurangan memori; implementasi dasar tidak bersifat lazy dan tidak boleh menyamarkan kegagalan alokasi sebagai antrean kosong. Uji konkuren harus mengoordinasikan awal mulai dengan barrier atau latch dan menggunakan batas waktu test-runner untuk mendeteksi pemblokiran permanen. Pemanggilan sleep yang singkat bukanlah bukti bahwa sebuah thread telah mencapai status menunggunya.

Contoh Jawaban Berkualitas Tinggi

“Pertama-tama, saya akan membatasi cakupan kontrak dasar pada kapasitas positif, elemen non-null, pemblokiran put/take, FIFO, banyak produsen dan konsumen, serta proses tunggu yang dapat diinterupsi. Batas waktu, shutdown, dan keadilan ketat memerlukan nilai kembalian dan semantik state tambahan, jadi saya tidak akan memasukkannya ke dalam implementasi inti.

Representasinya adalah sebuah ring array tetap. head adalah posisi baca berikutnya, tail adalah posisi tulis berikutnya, dan size menghilangkan ambiguitas penuh versus kosong ketika head == tail. Satu ReentrantLock melindungi keempat shared field. Predikat untuk notEmpty adalah size > 0, dan predikat untuk notFull adalah size < capacity.

put memperoleh lock secara interruptible, menunggu kondisi tidak-penuh di dalam while, menulis di tail dan menambah size, lalu memberi sinyal ke satu konsumen. take secara simetris menunggu kondisi tidak-kosong, menghapus head, mengosongkan referensi dan mengurangi size, lalu memberi sinyal ke satu produsen. Loop diperlukan karena condition wait dapat bangun secara palsu (spurious) dan karena thread lain dapat membuat predikat menjadi false lagi saat thread yang terbangun bersaing untuk memperoleh lock kembali.

await secara atomik melepaskan lock dan mulai menunggu, yang menghindari hilangnya notifikasi setelah pemeriksaan tetapi sebelum tidur. Linearization point dari operasi yang berhasil adalah transisi state terkunci yang mengubah keanggotaan antrean dan size. Penggunaan lock yang sama memastikan thread lain melihat state sebelum atau state sesudah yang lengkap. Induksi atas kapasitas, ring FIFO, dan invariansi lock yang sama membuktikan bahwa implementasi tidak mengalami overflow, tidak menduplikasi penghapusan, atau tidak mengubah urutan elemen.

Di luar waktu pemblokiran, kedua metode memiliki kompleksitas O(1), dan ruangnya adalah O(capacity). Saya akan menjalankan beberapa produsen dan konsumen secara bersamaan menggunakan latch, memverifikasi set pengenal unik dan urutan per-produsen, serta menguji jalur penuh, kosong, dan interupsi secara terpisah.”

Kesalahan Umum

• Memeriksa kondisi penuh atau kosong dengan if → kondisi mungkin masih bernilai false setelah spurious wakeup atau perebutan lock → uji ulang predikat state di dalam while.
• Membuka lock secara manual setelah pemeriksaan lalu menunggu → perubahan state dapat terjadi sebelum pendaftaran waiter, yang menyebabkan notifikasi hilang → gunakan Condition.await() yang terkait dengan lock yang sama.
• Menyinkronkan array, indeks, dan size secara terpisah → thread lain dapat mengamati intermediate state yang kontradiktif → lindungi seluruh pemeriksaan dan transisi dengan satu lock.
• Menggunakan satu wait set dengan signal sembarangan → sinyal dapat membangunkan thread dengan peran yang sama yang tidak dapat melanjutkan eksekusi → pertahankan condition notEmpty dan notFull secara terpisah.
• Melakukan spin atau tidur sambil menahan lock → thread yang seharusnya dapat mengubah kondisi tidak dapat memperoleh lock → condition wait harus melepaskan lock.
• Menelan InterruptedException → pemanggil tidak dapat membatalkan pekerjaan dan thread dapat tertahan tanpa batas waktu → deklarasikan dan teruskan interupsi, serta buka lock di dalam blok finally.
• Hanya menggunakan head == tail untuk mendeteksi kedua state → state ring penuh dan kosong tidak dapat dibedakan → pertahankan variabel size yang dilindungi oleh lock.
• Membiarkan referensi yang dihapus tetap berada di dalam array → array menahan objek yang telah dikonsumsi lebih lama dari yang diperlukan → atur slot ke null setelah membacanya.
• Menyamakan FIFO elemen dengan keadilan thread → lock default tidak menyelesaikan pemanggilan yang menunggu berdasarkan urutan kedatangan → jelaskan pengurutan elemen dan kebijakan penjadwalan secara terpisah.
• Mengklaim dukungan shutdown pada implementasi dasar → waiter tidak memiliki state penutupan yang dapat diamati dan mungkin tidak akan pernah bangun → definisikan kontrak shutdown sebelum menambahkan state dan notifikasi broadcast.

Pertanyaan Lanjutan dan Tanggapan

Lanjutan 1: Bagaimana Anda menambahkan offer dan poll dengan batas waktu?

Konversikan durasi yang tersisa ke nanodetik dan panggil awaitNanos(remaining) di dalam loop predikat while yang sama. Setelah setiap pengembalian, lanjutkan dengan sisa waktu yang dilaporkannya. Jika nilainya non-positif dan predikat masih false, kembalikan status gagal. Menggunakan kembali batas waktu penuh setelah setiap spurious wakeup dapat memperpanjang waktu tunggu sebenarnya tanpa batas. API juga harus membedakan batas waktu dari elemen null, alasan lain untuk menolak null.

Lanjutan 2: Bagaimana Anda mengimplementasikan shutdown()?

Definisikan state machine terlebih dahulu. Misalnya, tolak operasi put baru setelah penutupan tetapi biarkan item yang ada dikosongkan (drain); saat tertutup dan kosong, take melempar exception khusus atau mengembalikan hasil eksplisit. shutdown harus mengubah flag penutupan di bawah lock yang sama dan memanggil signalAll() pada kedua condition sehingga setiap waiter dapat memperoleh lock kembali dan mengamati penutupan tersebut. Setiap predikat wait-loop memerlukan cabang untuk state tertutup; hanya menambahkan field boolean saja tidaklah cukup.

Lanjutan 3: Mengapa menggunakan signal() di sini, dan kapan Anda akan menggunakan signalAll()?

Satu penyisipan dasar membuat satu elemen yang dapat dikonsumsi, dan satu penghapusan membuat satu slot kosong. Membangunkan satu thread dengan peran berlawanan sudah cukup untuk membuat progres dan mengurangi persaingan yang sia-sia. Penyisipan massal, penghapusan massal, perubahan kapasitas dinamis, atau shutdown dapat membuat beberapa waiter memenuhi syarat sekaligus, sehingga biasanya memerlukan signalAll() atau jumlah notifikasi yang sesuai dengan perubahan state. Terlepas dari berapa banyak yang dibangunkan, pemeriksaan ulang pada while tetap wajib dilakukan.

Lanjutan 4: Bagaimana Anda memberikan keadilan (fairness)?

new ReentrantLock(true) membuat perolehan lock memprioritaskan thread yang menunggu paling lama, tetapi ini tetap tidak memberikan urutan penyelesaian real-time mutlak untuk put/take; interupsi dan penjadwalan juga berpengaruh. Kebijakan yang adil umumnya mengurangi risiko barging dan starvation tetapi dapat menurunkan throughput. Bayar dan verifikasi biaya tersebut hanya jika kontrak pemanggil memang memerlukan pengurutan waiter.

Lanjutan 5: Bisakah ini dijadikan antrean lock-free?

Antrean MPMC bounded yang lock-free biasanya memerlukan nomor urut atomik (atomic sequence numbers), CAS, dan pembuktian memory-order yang jauh lebih rumit. Perilaku “blocking”-nya masih membutuhkan mekanisme parking dan wakeup; spin CAS saja bukanlah sebuah blocking queue. Desain tersebut menambah masalah ABA, false sharing, jaminan progres, dan pertimbangan memory-model platform. Kecuali jika pengukuran menunjukkan bahwa lock tunggal adalah bottleneck dan tim dapat memelihara pembuktian serta stress test-nya, kelas pustaka standar atau implementasi berbasis lock yang jelas akan lebih aman.

Lanjutan 6: Mengapa tidak langsung menggunakan dua semaphore?

Satu counting semaphore dapat mewakili slot kosong dan yang lainnya mewakili elemen yang tersedia, tetapi pembaruan pada head/tail ring tetap memerlukan mutual exclusion. Memperoleh beberapa primitif sinkronisasi juga memerlukan penanganan exception, interupsi, dan rollback permit yang cermat. Solusi semaphore bisa saja benar, tetapi tidak secara otomatis lebih ringkas daripada satu lock ditambah dua condition. Kedua desain harus membuktikan bahwa jumlah permit dan state array sebenarnya tidak pernah berbeda.