Producer Consumer Problem

Setelah membaca dan mem-postkan Strategy Pattern, saya ingin intermezzo dulu. Capek juga euy baca topik tentang design pattern.

Beberapa hari lalu saya melihat postingan mengenai masalah Producer Consumer di sini. Teman saya yang mem-post-kan. Ternyata merupakan implementasi dari pseudocode soal di ujian sistem operasinya. Wah rajin banget euy!

Di situ, terdapat kode producer consumer dalam bahasa java. Katanya, pseudocode-nya tidak aman. Dan benar juga, ketika dijalankan, outputnya menyatakan terjadi buffer overflow. Ini dia tampilannya saat saya jalankan di komputer.

Perkiraan awal saya, mungkin ini ada variabel yang di-share bersama. Jadi kalau tidak dilindungi, bisa gawat. Kemudian saya trace program-nya. Oh, di situ ada variabel count. Mungkin ini yang jadi masalah, karena diakses oleh producer dan consumer. Kemudian di java ada modifier synchronized untuk mencegah ada lebih dari satu objek mengakses method secara bersamaan. Saya coba synchronize-kan saja method yang menggunakan variabel count.

Kemudian saya compile dan jalankan. Arrgh…. masih terjadi buffer overflow. Saya cek lagi program-nya. Terus saya lihat ternyata producer dan consumer tidak mengakses method yang sama. Ketika mengubah variabel count, producer menggunakan method incCount(). Sedangkan consumer menggunakan decCount(). Dengan begitu, tidak ada gunanya di-synchronize.

Saya trace lagi, dan menemukan bahwa pengaksesan variable tidak ada yang jadi masalah. Akhirnya pencerahan itu datang. Letak masalahnya ada di statement

if (buffer.getCount() == 0) {
                try {
                    Thread.sleep(tts);
                } catch (InterruptedException e) {
                    //e.printStackTrace();
                }
            }

Saya sebelumnya mengira, statement ini menyatakan consumer akan terus sleep ketika buffer-nya masih kosong. Tapi asumsi saya salah. Ia hanya mengecek sekali, kemudian sleep beberapa lama, setelah itu wake dan melakukan remove item, meskipun bisa saja item pada buffer belum tersedia alias buffer-nya masih kosong. Mungkin di saat awal tidak terjadi masalah. Tapi beberapa cycle kemudian, timing-nya tidak bisa menjamin ketika consumer wake lagi, buffer-nya pasti ada item-nya.

Sebelum melakukan remove item dari buffer, consumer harus memastikan dulu buffer-nya ada item-nya. Ini bisa dilakukan dengan mengganti pengecekannya dengan

while( buffer.getCount()==0) { }.

Jadi selama masih kosong, do nothing. Cara itu untuk mencegah buffer underflow.
Untuk mencegah buffer overflow, pengecekan pada producer dapat diganti menjadi

while (buffer.getCount() == buffer.getBufferSize()) { }

Selama masih penuh, do nothing.

Yah, sebenarnya kurang efektif sih pengecekan seperti itu. Kalau bisa processor langsung pindah dari consumer ketika buffer masih kosong, dan bukan berada terus di consumer dengan tidak melakukan apa-apa. Menyia-nyiakan waktu quantum saja!

Tapi untuk sementara itu dulu solusinya. Yang penting tidak ada buffer overflow atau buffer underflow .

Ini dia listing kode yang sedikit saya modifikasi. Bedanya, dengan mengimplementasikan runnable, kemudian waktu main-nya membuat thread dari producer dan consumer. Nggak ada alasan khusus sih. Saya cuma tahu cara buat thread seperti ini.

import java.util.*;

public class Buffer {
    private Vector buffer;
    private int bufferSize;
    private int count;

    public Buffer(int bufferSize) {
        this.buffer = new Vector(bufferSize);
        this.bufferSize = bufferSize;
        this.count = 0;
    }

    public synchronized void enterItem(String item) {
        buffer.add(item);
        System.out.println(item + " added.");
        if (buffer.size()>bufferSize) {
            System.err.println("\tBuffer overflow!!...");
        }
    }

    public synchronized String removeItem() {
        Object result;
        try {
        result = buffer.remove(0);
        } catch (Exception e) {
            System.err.println("\tBuffer underflow!!...");
            return "";
        }

        System.out.println("" + result + " removed.");
        return "" + result;
    }

    public void incCount() {
        count++;
    }

    public void decCount() {
        count--;
    }

    public int getBufferSize() {
        return bufferSize;
    }

    public int getCount() {
        return count;
    }

    public synchronized String toString() {
        StringBuffer sb = new StringBuffer(buffer.size());
        for (Object o:buffer) {
            if (o!=null) {
                sb.append("# ");
            }
        }

        return "" + count + ": " + sb.toString();
    }
}

public class Producer implements Runnable {
    private Buffer buffer;
    private long tts;

    private Consumer consumer;

    //--
    String item;
    private int msgId = 0;

    public Producer(Buffer buffer, long tts) {
        this.buffer = buffer;
        this.tts = tts;

        //setDaemon(false);
    }

    public void setConsumer(Consumer consumer) {
        this.consumer = consumer;
    }

    public void run() {
        while (true) {
            item = "msg " + ++msgId;

//	    if (buffer.getCount() == buffer.getBufferSize()) {
//                try {
//                    Thread.sleep(tts);
//                } catch (InterruptedException e) {
//                    //e.printStackTrace();
//                }
//            }

	while (buffer.getCount() == buffer.getBufferSize())
			{
			}

            buffer.enterItem(item);
            buffer.incCount();

            if (buffer.getCount() == 1)
                synchronized (consumer) {
                    consumer.notifyAll();
                }
        }
    }
}

public class Consumer implements Runnable {

    private Buffer buffer;
    private long tts;

    private Producer producer;

    //--
    private String item;

    public Consumer(Buffer buffer, long tts) {
        this.buffer = buffer;
        this.tts = tts;

        //setDaemon(false);
    }

    public void setProducer(Producer producer) {
        this.producer = producer;
    }

    public void run() {
        while (true) {

//           if (buffer.getCount() == 0) {
//               try {
//                   Thread.sleep(tts);
//               } catch (InterruptedException e) {
//                   //e.printStackTrace();
//               }
//           }

	   while (buffer.getCount() == 0)
			{
			}

            item = buffer.removeItem();
            buffer.decCount();

            if (buffer.getCount() == buffer.getBufferSize() - 1) {
                synchronized (producer) {
                    producer.notifyAll();
                }
            }

            System.out.println("Consumer consumes " + item);
        }
    }
}

public class ProducerConsumerDemo {

	public static void main(String[] args) {
		// Contoh kasus
		// buffernya hanya berkapasitas
		// satu item
		Buffer buffer = new Buffer(1);

		Producer producer = new Producer(buffer, 500);
		Consumer consumer = new Consumer(buffer, 500);
		producer.setConsumer(consumer);
		consumer.setProducer(producer);

		Thread one = new Thread(producer);
		Thread two = new Thread(consumer);

		one.start();
		two.start();
	}
}

Tampilan outputnya:

Setelah message ke 7798, syukurlah tidak ada pesan buffer overflow atau buffer underflow.

Mungkin ada teman yang punya solusi yang lebih baik. Share dong!