BoundedBuffer

/**
 *
 * Proizvođač i potrošač se vrte u beskonačnim petljama. U svakoj iteraciji oni pokušavaju da stave,
 * odnosno preuzmu element iz bafera, nakon čega čekaju najviše 500 milisekundi pre prelaska na sledeću iteraciju.
 * U realnim situacijama procedure produce() i consume() bi implementirale ponašanje koje odgovara stvarnim potrebama
 * modeliranog problema. U našem primjeru, ovi metodi imaju samo simboličnu ulogu.
 *
 * Bafer je realizovan u potpunosti prateći definiciju ove strukture podataka.
 * Kritične oblasti su ostavarene upotrebom modifikatora synchronized pri deklaraciji metoda put() i get().
 * Na taj način  niti su onemogućene da istovremeno izvršavaju ove metode.
 * Pored toga, pre nego što zaista pristupi dodavanju elementa u bafer, metod put() provjerava da li je bafer pun,
 * i u slučaju da jeste blokira izvršavanje niti proizvođača. Nit proizvođača će ostati blokirana sve dok potrošač,
 * u cilju preuzimanja jednog elementa iz bafera, ne izvrši metod get(), na čijem kraju se nalazi poziv metoda notify().
 * Pozivom metoda notify(), nit proizvođača se budi i pristupa postavljanju novog elementa u bafer.
 * Na isti način je ostvarena sinhronizacija i u suprotnom smijeru.
 * Pre nego što zaista pristupi preuzimanju elemenata iz bafera, metod get provjerava da li je bafer prazan,
 * i u slučaju da jeste, blokira izvršavanje niti potrošača. Nit potrošača će ostati blokirana sve dok proizvođač,
 * u cilju dodavanja jednog elementa u bafer, ne izvrši metod put(), na čijem kraju se nalazi poziv metoda notify().
 * Pozivom metoda notify(), nit potrošača se budi i pristupa preuzimanju postavljenog elementa iz bafera.
 *
 */

public class BoundedBuffer {
    private volatile int size, head, tail;
    private volatile boolean empty;
    private volatile int[] array;

    public BoundedBuffer (int size) {
        this.array = new int[size];
        this.size = size;
        this.head = 0;
        this.tail = 0;
        this.empty = true;
    }

    public synchronized void put(int value) {
        if (increment(tail) == head)
            try {
                wait();
            } catch (InterruptedException e) {}

        if (empty)
            empty = false;
        else
            tail = increment(tail);

        array[tail] = value;

        System.out.println("Producted: " + value);
        notify();
    }

    public synchronized int get() {
        if (empty)
            try {
                wait();
            } catch(InterruptedException e) {}

        int value = array[head];

        if (head == tail)
            empty = true;
        else
            head = increment(head);

        System.out.println("Consumed: " + value);

        notify();
        return value;
    }

    private int increment(int value) {
        return (value + 1) % size;
    }

    public static void main(String[] args) {
        BoundedBuffer buffer = new BoundedBuffer(3);
        Producer producer = new Producer(buffer);
        Consumer consumer = new Consumer(buffer);
        producer.start();
        consumer.start();
    }
}

class Producer extends Thread {
    private BoundedBuffer buffer;

    public Producer (BoundedBuffer buffer) {
        this.buffer = buffer;
    }

    private int produce() {
        return (int) (100 * Math.random());
    }

    @Override
    public void run() {
        while (true) {
            buffer.put(produce());
            try {
                sleep((long) (500 * Math.random()));
            } catch (InterruptedException e) {}
        }
    }
}

class Consumer extends Thread {
    private BoundedBuffer buffer;

    public Consumer (BoundedBuffer buffer) {
        this.buffer = buffer;
    }

    private void consume (int value) {}

    @Override
    public void run() {
        while (true) {
            consume(buffer.get());
            try {
                sleep((long) (500 * Math.random()));
            } catch (InterruptedException e) {}
        }
    }
}