Melaksanakan Bounded Blocking Queue yang Selamat daripada Thread

Masalah dan Bila Ia Digunakan

Laksanakan BoundedBlockingQueue<E>. Pembina (constructor) menerima kapasiti positif. put(item) menambah elemen mengikut susunan FIFO dan menunggu semasa queue penuh. take() mengeluarkan elemen dari bahagian kepala (head) dan menunggu semasa queue kosong. Berbilang pengeluar dan pengguna boleh memanggil kedua-dua kaedah ini secara serentak. Tiada elemen yang boleh hilang, dikembalikan dua kali, atau dikembalikan selepas elemen yang masuk lebih awal. Jika thread diganggu semasa memperoleh kunci atau menunggu syarat, kaedah tersebut akan melontarkan InterruptedException.

Latihan ini tidak membenarkan penggunaan pembungkus (wrapper) ArrayBlockingQueue. API asas tidak merangkumi offer/poll tidak menyekat, had masa tamat (timeouts), penyingkiran pukal, dan semantik penutupan (shutdown), serta tidak menjanjikan keadilan ketat (strict fairness) antara thread yang sedang menunggu. Perkara tersebut adalah soalan susulan. Seperti BlockingQueue dalam Java, pelaksanaan ini menolak null, kerana API queue sering menggunakan null untuk menandakan tiada elemen yang tersedia.

Ini merupakan masalah pengekodan struktur data serentak (concurrent data structure). Pengindeksan tatasusunan adalah bahagian yang mudah. Ujian sebenar adalah sama ada calon boleh menyatakan kontrak yang boleh diaudit: penyelarasan mana yang melindungi keadaan dikongsi (shared state), mengapa pemberitahuan tidak boleh hilang, mengapa thread yang dikejutkan mesti menyemak semula syaratnya, dan pada saat bila sesuatu operasi berkuat kuasa bagi thread lain.

Perkara yang Dinilai oleh Penemu Duga

Isyarat pertama adalah sama ada calon membezakan antara keselamatan (safety) dan keaktifan (liveness). Keselamatan memerlukan saiz kekal dalam [0, capacity], setiap elemen dikeluarkan paling banyak sekali, dan susunan penyingkiran sepadan dengan susunan pemasukan. Keaktifan memerlukan pengeluar mendapat peluang untuk meneruskan operasi apabila queue yang penuh tidak lagi penuh, dan pengguna mendapat peluang apabila queue yang kosong tidak lagi kosong. Gangguan juga mesti boleh membatalkan penantian.

Isyarat kedua adalah sama ada primitif penyelarasan sepadan dengan predikat keadaan. Satu kunci melindungi tatasusunan, head, tail, dan saiz, supaya penyemakan syarat dan perubahan keadaan berlaku dalam bahagian genting (critical section) yang sama. notFull mewakili size < capacity; notEmpty mewakili size > 0. Pengeluar hanya menunggu bagi syarat pertama, pengguna hanya menunggu bagi syarat kedua, dan perubahan keadaan yang merentasi sempadan memberi isyarat kepada peranan yang bertentangan.

Isyarat ketiga adalah sama ada calon boleh menerangkan while dan bukannya sekadar menggunakannya sebagai corak hafalan. Condition membenarkan spurious wakeups. Walaupun selepas signal sebenar, thread pesaing lain mungkin memperoleh semula kunci terlebih dahulu lalu mengisi atau mengosongkan queue semula. Sebaik sahaja thread yang dikejutkan memperoleh semula kunci, ia mesti menguji semula predikat tersebut. Menerima pemberitahuan tidak membuktikan syarat tersebut masih benar.

Akhir sekali, penemu duga boleh menguji reka bentuk secara mendalam: bagaimana indeks gelang mengekalkan FIFO, mengapa pengemaskinian saiz tergolong dalam titik linearisasi, mengapa satu kunci mengelakkan kebuntuan susunan kunci (lock-order deadlock), bila signal mencukupi, dan mengapa had masa tamat serta penutupan memerlukan semantik API baharu.

Soalan untuk Dijelaskan Sebelum Menjawab

• Apakah kapasiti dan elemen yang sah? Kapasiti mestilah lebih besar daripada sifar, dan elemen null ditolak.
• Patutkah penantian semasa penuh atau kosong menggunakan putaran (spin)? Tidak. Thread yang menunggu mesti melepaskan kunci dan menunggu pada syarat (condition). Ia tidak boleh menggunakan CPU dalam gelung putaran (spin loop) mahupun tidur sambil memegang kunci.
• Bagaimanakah gangguan harus bertindak? Kedua-dua put dan take menyebarkan InterruptedException. Ia tidak menelan gangguan atau mengubah queue selepas operasi yang diganggu gagal.
• Adakah keadilan ketat (strict fairness) diperlukan? Tidak dalam masalah asas. ReentrantLock tidak adil secara lalai mungkin membenarkan thread yang datang kemudian memperoleh kunci terlebih dahulu. Kunci adil mengubah daya pemprosesan (throughput) dan kelakuan penjadualan.
• Adakah queue memerlukan penutupan (shutdown)? Tidak dalam masalah asas. Jika perlu, tentukan sama ada item sedia ada boleh dikeluarkan sepenuhnya, sama ada thread yang menunggu menerima pengecualian atau hasil khas, dan siapa yang mengejutkan semua thread yang menunggu.
• Adakah elemen FIFO berlinear dan pemanggil FIFO yang menunggu merupakan jaminan yang sama? Tidak. Elemen keluar mengikut susunan linearisasi put yang berjaya. Ini tidak bermakna pengeluar atau pengguna yang disekat dibenarkan masuk mengikut susunan ketibaan mereka.
• Bolehkah kelas pustaka standard digunakan? Kod pengeluaran biasanya harus mengutamakan ArrayBlockingQueue yang telah diuji. Pelaksanaan secara manual di sini bertujuan khusus untuk menilai invarian keserentakan dan semantik penantian syarat.

Kerangka Jawapan 30 Saat

"Saya akan menggunakan tatasusunan tetap sebagai penimbal gelang, dengan head, tail, dan size mewakili kedudukan bacaan seterusnya, kedudukan penulisan seterusnya, dan bilangan elemen semasa. Satu ReentrantLock melindungi semua keadaan dikongsi, dan dua objek Condition mewakili tidak kosong dan tidak penuh. put menunggu di dalam while (size == capacity), memasukkan elemen serta meningkatkan size, kemudian memberi isyarat kepada seorang pengguna. take secara simetri menunggu bagi keadaan tidak kosong, mengosongkan head dan mengurangkan size, kemudian memberi isyarat kepada seorang pengeluar. Setiap semakan dan peralihan berlaku di bawah kunci yang sama. Kerana await melepaskan kunci secara atomik dan memperolehnya semula sebelum kembali, tiada jurang kehilangan pemberitahuan (lost-notification window) antara penyemakan dan penantian. Setiap operasi yang berjaya adalah O(1), dengan ruang O(capacity)."

Huraian Mendalam Langkah Demi Langkah

Terbitkan perwakilan daripada kekangan kaedah naif. Tatasusunan biasa yang mengalihkan elemen yang tinggal selepas setiap penyingkiran menyebabkan take menjadi O(n). Mengekalkan indeks bacaan yang hanya bertambah pula akan membazirkan ruang awalan yang telah dilepaskan. Tatasusunan gelang tetap menggunakan semula slot yang telah dilepaskan: head menunjuk ke kedudukan bacaan seterusnya, tail menunjuk ke kedudukan penulisan seterusnya, dan size adalah bilangan elemen semasa. Indeks berputar kembali ke sifar selepas mencapai penghujung, jadi pemasukan mahupun penyingkiran tidak mengalihkan elemen sedia ada.

Pelaksanaan ini mengekalkan empat invarian:

1. 0 <= size <= items.length.
2. Bermula pada head, slot size pertama dalam susunan gelang mengandungi jujukan FIFO yang belum digunakan.
3. tail == (head + size) % items.length. Apabila queue penuh, head == tail, jadi size membezakan keadaan penuh daripada kosong.
4. Setiap bacaan atau penulisan items, head, tail, dan size berlaku semasa memegang kunci yang sama.

Berikut ialah pelaksanaan teras:

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 dilaksanakan sebelum penguncian kerana ia hanya menyemak hujah dan tidak bergantung pada keadaan dikongsi. lockInterruptibly() menjadikan penantian untuk memperoleh kunci itu sendiri boleh diganggu. Selepas kaedah memasuki try, finally melepaskan kunci yang dipegang oleh thread semasa sama ada semasa pulangan normal, gangguan semasa penantian syarat, atau pengecualian masa larian (runtime exception).

Jaminan penting await() adalah ia melepaskan kunci yang berkaitan secara atomik dan menunggu, kemudian memperoleh semula kunci tersebut sebelum kembali. Pengeluar tidak boleh membuka kunci secara manual dan hanya mendaftarkan dirinya sebagai menunggu selepas itu. Ini mewujudkan jurang di mana pengguna boleh mengosongkan ruang dan menghantar pemberitahuan sebelum pengeluar benar-benar mula menunggu, menyebabkan pemberitahuan hilang secara kekal. Pemboleh ubah syarat menggabungkan "melepaskan dan mula menunggu" ke dalam satu tindakan penyelarasan dan menutup jurang tersebut.

while mengendalikan dua kes berbeza. Pertama ialah spurious wakeup yang dibenarkan oleh spesifikasi. Satu lagi ialah persaingan biasa: dua pengguna mungkin dikejutkan, seorang memperoleh semula kunci terlebih dahulu dan mengeluarkan satu-satunya elemen, dan pengguna kedua mendapati queue kosong apabila ia akhirnya mendapat kunci. Pemberitahuan syarat hanya menyatakan bahawa keadaan mungkin telah berubah; predikat keadaan ialah penentu utama untuk meneruskan operasi.

Titik linearisasi bagi put yang berjaya ialah peralihan terkunci yang menambah elemen dan mengubah size daripada k kepada k + 1. Bagi take, ia adalah penyingkiran dan peralihan sepadan daripada k kepada k - 1. Kunci memastikan thread lain boleh memerhatikan sama ada keadaan lengkap sebelum peralihan atau keadaan lengkap selepasnya, tidak sekali-kali penulisan tatasusunan tanpa kemas kini saiz yang sepadan. Membuka kunci dan seterusnya memperoleh kunci yang sama juga mewujudkan keterlihatan memori (memory visibility), sepadan dengan matlamat happens-before yang ditentukan untuk menghantar elemen melalui queue serentak standard.

Mengapa menggunakan dua syarat? Dengan satu set penantian, take mungkin mengejutkan pengguna lain walaupun queue kekal kosong, manakala pengeluar yang boleh menggunakan slot baharu kekal tidur. Memisahkan notEmpty dan notFull membolehkan setiap peralihan hanya memberitahu peranan yang kini boleh meneruskan operasi. put atau take elemen tunggal asas hanya mencipta satu elemen atau slot baharu, jadi signal() sudah mencukupi; thread yang dikejutkan tetap menyemak semula dalam while. Jika satu operasi mengubah berbilang slot, atau penutupan memerlukan setiap thread yang menunggu untuk melihat keadaan baharu, pertimbangkan semula penggunaan signalAll().

Ketepatan dibuktikan secara aruhan pada invarian. Pada mulanya, head = tail = size = 0. Pemasukan hanya berjalan apabila size < capacity; ia menulis pada tail, menggerakkan tail ke hadapan, dan menambah saiz tepat sekali, jadi ia kekal dalam kapasiti dan menambah selepas semua elemen yang belum digunakan. Penyingkiran hanya berjalan apabila size > 0; ia membaca pada head, mengosongkan slot, menggerakkan head ke hadapan, dan mengurangkan saiz tepat sekali, jadi ia mengembalikan elemen belum digunakan yang paling awal. Kunci mensirikan peralihan ini, menjadikan setiap penjadualan berbilang pengeluar dan pengguna bersamaan dengan pelaksanaan berurutan yang sah.

Setiap put atau take yang berjaya melakukan bilangan operasi tatasusunan, integer, dan penyelarasan yang tetap, jadi kerjanya adalah O(1) tidak termasuk masa menunggu. Tatasusunan tetap menggunakan ruang O(capacity). Di bawah persaingan, penantian kunci dan pertukaran konteks (context switches) boleh mendominasi latensi; notasi asimptotik tidak menggambarkan kos tersebut. Reka bentuk satu kunci tidak mempunyai kitaran pelbagai kunci, tetapi pemanggil masih boleh mencipta masalah susunan kunci yang lebih besar jika mereka memanggil kaedah menyekat semasa memegang kunci lain.

Ujian tidak boleh berhenti pada contoh thread tunggal sahaja. Uji pertukaran penuh/kosong kapasiti satu, pelbagai putaran gelang lengkap (wraparounds), elemen berulang yang sama disimpan dan dikeluarkan secara berasingan, rentetan kosong diterima manakala null ditolak, nilai unik yang dihasilkan oleh beberapa pengeluar dan dikeluarkan oleh beberapa pengguna, tiada nilai yang hilang atau pendua dalam set akhir, susunan yang dikekalkan dalam setiap aliran pengeluar individu, dan gangguan kedua-dua operasi put dan take yang disekat. Kapasiti yang sangat besar memperuntukkan tatasusunan yang sama besarnya dalam pembina dan mungkin gagal kerana kekurangan memori; pelaksanaan asas bukan jenis malas (lazy) dan tidak boleh menyembunyikan kegagalan peruntukan sebagai queue kosong. Ujian keserentakan harus menyelaraskan permulaan dengan barrier atau latch dan menggunakan had masa test-runner untuk mengesan penyekatan kekal. sleep yang pendek bukanlah bukti bahawa thread telah mencapai keadaan menunggunya.

Contoh Jawapan Berkualiti Tinggi

"Mula-mula saya akan mengehadkan skop kontrak asas kepada kapasiti positif, elemen bukan nol, put/take menyekat, FIFO, berbilang pengeluar dan pengguna, serta penantian yang boleh diganggu. Had masa tamat, penutupan, dan keadilan ketat memerlukan nilai pulangan dan semantik keadaan tambahan, jadi saya akan mengecualikannya daripada pelaksanaan teras.

Perwakilannya ialah tatasusunan gelang tetap. head ialah kedudukan bacaan seterusnya, tail kedudukan penulisan seterusnya, dan size menghapuskan kekaburan penuh berbanding kosong apabila head == tail. Satu ReentrantLock melindungi keempat-empat medan yang dikongsi. Predikat untuk notEmpty ialah size > 0, dan predikat untuk notFull ialah size < capacity.

put memperoleh kunci secara boleh diganggu, menunggu bagi keadaan tidak penuh di dalam while, menulis pada tail dan menambah size, kemudian memberi isyarat kepada seorang pengguna. take secara simetri menunggu bagi keadaan tidak kosong, mengeluarkan head, mengosongkan rujukan dan mengurangkan size, kemudian memberi isyarat kepada seorang pengeluar. Gelung diperlukan kerana penantian syarat mungkin mengalami spurious wakeups dan thread lain mungkin menjadikan predikat itu palsu semula semasa thread yang dikejutkan bersaing untuk memperoleh semula kunci.

await melepaskan kunci secara atomik dan mula menunggu, yang mengelakkan kehilangan pemberitahuan selepas menyemak tetapi sebelum tidur. Titik linearisasi bagi operasi yang berjaya ialah peralihan keadaan terkunci yang mengubah keahlian queue dan size. Kunci yang sama memastikan thread lain melihat keadaan sebelum atau selepas yang lengkap. Aruhan ke atas kapasiti, FIFO gelang, dan invarian kunci yang sama membuktikan bahawa pelaksanaan ini tidak melimpah (overflow), tidak menduplikasi penyingkiran, atau mengubah susunan elemen.

Tidak termasuk masa menyekat, kedua-dua kaedah adalah O(1), dan ruang adalah O(capacity). Saya akan memulakan berbilang pengeluar dan pengguna menggunakan latch, mengesahkan set pengecam unik dan susunan bagi setiap pengeluar, serta menguji laluan penuh, kosong, dan gangguan secara berasingan."

Kesilapan Biasa

• Menyemak penuh atau kosong dengan if → syarat mungkin masih palsu selepas spurious wakeup atau persaingan kunci → uji semula predikat keadaan dalam while.
• Membuka kunci secara manual selepas semakan dan kemudian menunggu → perubahan keadaan boleh berlaku sebelum pendaftaran penunggu, menyebabkan kehilangan pemberitahuan → gunakan Condition.await() yang dikaitkan dengan kunci yang sama.
• Menyelaraskan tatasusunan, indeks, dan size secara berasingan → thread lain boleh melihat keadaan perantaraan yang bercanggah → lindungi keseluruhan semakan dan peralihan dengan satu kunci.
• Menggunakan satu set penantian dengan signal sembarangan → isyarat mungkin mengejutkan thread dengan peranan yang sama yang tidak boleh maju → kekalkan syarat notEmpty dan notFull secara berasingan.
• Membuat putaran (spinning) atau tidur semasa memegang kunci → thread yang boleh mengubah syarat tidak dapat memperoleh kunci → penantian syarat mesti melepaskan kunci.
• Menelan InterruptedException → pemanggil tidak dapat membatalkan kerja dan thread mungkin kekal tersekat tanpa had masa → isytiharkan dan sebarkan gangguan, dan buka kunci dalam finally.
• Hanya menggunakan head == tail untuk mengesan kedua-dua keadaan → keadaan gelang penuh dan kosong tidak dapat dibezakan → kekalkan size yang dilindungi kunci.
• Membiarkan rujukan yang dikeluarkan di dalam tatasusunan → tatasusunan mengekalkan objek yang telah digunakan lebih lama daripada yang sepatutnya → tetapkan slot kepada null selepas membacanya.
• Menyamakan elemen FIFO dengan keadilan thread → kunci lalai tidak menyelesaikan panggilan menunggu mengikut susunan ketibaan → huraikan susunan elemen dan dasar penjadualan secara berasingan.
• Mendakwa sokongan penutupan (shutdown) dalam pelaksanaan asas → penunggu tidak mempunyai keadaan tertutup yang boleh diperhatikan dan mungkin tidak akan bangun → tentukan kontrak penutupan sebelum menambah keadaan dan pemberitahuan siaran (broadcast notification).

Soalan Susulan dan Respons

Susulan 1: Bagaimanakah anda menambah offer dan poll dengan had masa?

Tukarkan baki tempoh masa kepada nanosaat dan panggil awaitNanos(remaining) di dalam gelung while predikat yang sama. Selepas setiap pulangan, teruskan dengan baki masa yang dilaporkan. Jika nilainya bukan positif dan predikat masih palsu, kembalikan kegagalan. Menggunakan semula had masa penuh selepas setiap spurious wakeup boleh memanjangkan penantian sebenar tanpa had. API juga mesti membezakan masa tamat daripada elemen null, satu lagi sebab untuk menolak null.

Susulan 2: Bagaimanakah anda melaksanakan shutdown()?

Tentukan mesin keadaan (state machine) terlebih dahulu. Contohnya, tolak put baharu selepas penutupan tetapi benarkan item sedia ada dikeluarkan; apabila ditutup dan kosong, take melontarkan pengecualian khas atau mengembalikan hasil eksplisit. shutdown mesti menukar bendera tertutup di bawah kunci yang sama dan memanggil signalAll() pada kedua-dua syarat supaya setiap penunggu boleh memperoleh semula kunci dan melihat penutupan tersebut. Setiap predikat gelung penantian memerlukan cabang keadaan tertutup; hanya menambah medan boolean adalah tidak mencukupi.

Susulan 3: Mengapa menggunakan signal() di sini, dan bila anda akan menggunakan signalAll()?

Satu pemasukan asas mencipta satu elemen yang boleh digunakan, dan satu penyingkiran mencipta satu slot kosong. Mengejutkan satu thread dengan peranan bertentangan sudah cukup untuk membolehkan kemajuan dan mengurangkan persaingan yang sia-sia. Pemasukan pukal, penyingkiran pukal, perubahan kapasiti dinamik, atau penutupan boleh menyebabkan beberapa penunggu layak pada masa yang sama, jadi ia biasanya memerlukan signalAll() atau bilangan pemberitahuan yang sepadan dengan perubahan keadaan. Walau berapa banyak pun yang dikejutkan, semakan semula while tetap wajib.

Susulan 4: Bagaimanakah anda menyediakan keadilan (fairness)?

new ReentrantLock(true) menjadikan pemerolehan kunci mengutamakan thread yang paling lama menunggu, tetapi ia masih tidak memberikan susunan penyiapan masa nyata yang mutlak untuk put/take; gangguan dan penjadualan juga memain peranan. Dasar yang adil secara amnya mengurangkan barging dan risiko kebuluran (starvation), tetapi mungkin mengurangkan daya pemprosesan. Bayar dan sahkan kos tersebut hanya apabila kontrak pemanggil benar-benar memerlukan susunan penunggu.

Susulan 5: Bolehkah ini dijadikan lock-free queue?

Lock-free bounded MPMC queue biasanya memerlukan nombor jujukan atomik, CAS, dan pembuktian susunan memori yang jauh lebih rumit. Tingkah laku "menyekat" masih memerlukan mekanisme letak (parking) dan bangun (wakeup); putaran CAS semata-mata bukanlah blocking queue. Reka bentuk ini menambah masalah ABA, perkongsian palsu (false sharing), jaminan kemajuan, dan kebimbangan model memori platform. Melainkan pengukuran menunjukkan kunci tunggal adalah punca kesesakan (bottleneck) dan pasukan mampu mengekalkan bukti serta ujian tekanan, kelas pustaka standard atau pelaksanaan berasaskan kunci yang jelas adalah lebih selamat.

Susulan 6: Mengapa tidak menggunakan dua semaphore secara langsung?

Satu counting semaphore boleh mewakili slot kosong dan satu lagi mewakili elemen yang tersedia, tetapi kemas kini kepada head/tail gelang masih memerlukan saling eksklusif (mutual exclusion). Memperoleh beberapa primitif penyelarasan juga memerlukan pengendalian pengecualian, gangguan, dan pengunduran permit (permit-rollback) yang teliti. Penyelesaian semaphore boleh jadi betul, tetapi ia tidak semestinya lebih pendek daripada satu kunci ditambah dua syarat. Mana-mana reka bentuk mesti membuktikan bahawa kiraan permit dan keadaan tatasusunan sebenar tidak pernah menyimpang.