5
\$\begingroup\$

Please suggest improvements in the following java program I've written for producer-consumer scenario. The program seems to be working fine. Does it suffer from possible deadlock scenarios? How better I could have done this? Since I am using Stack read/write (push/pop) already been synchronized? What if they do not?

import java.util.Stack;

import logger.CustomLogger;

public class TestProducerConsumer {

    private Stack<Integer> buffer;
    public static final int MAX_SIZE = 10;
    public int count;

    public TestProducerConsumer(){

        buffer = new Stack<Integer>();
        count = 0;
    }

    public Stack<Integer> getBuffer(){
        return buffer;
    }

    public void addToBuffer(Integer i) throws StackException{

        if(buffer.size() < MAX_SIZE){
            buffer.push(i);
            CustomLogger.logger.info("pushed "+i);
        }else
            throw new StackException("Stack Over Flow");
    }

    public Integer removeFromBuffer() throws StackException{

        if(buffer.size() == 0)
            throw new StackException("Buffer Empty");
        else 
            return buffer.pop();
    }

    public static void main(String[] args) {

        TestProducerConsumer pd = new TestProducerConsumer();
        Producer p1 = new Producer(pd);
        Producer p2 = new Producer(pd);
        Producer p3 = new Producer(pd);

        Consumer c1 = new Consumer(pd);
        Consumer c2 = new Consumer(pd);
        Consumer c3 = new Consumer(pd);
        Consumer c4 = new Consumer(pd);
        Consumer c5 = new Consumer(pd);

        Thread tp1 = new Thread(p1);
        Thread tp2 = new Thread(p2);
        Thread tp3 = new Thread(p3);

        Thread tc1 = new Thread(c1);
        Thread tc2 = new Thread(c2);
        Thread tc3 = new Thread(c3);
        Thread tc4 = new Thread(c4);
        Thread tc5 = new Thread(c5);

        tp1.start();
        tc1.start();
        tc2.start();
        tc3.start();
        tc4.start();
        tc5.start();
        tp2.start();
        tp3.start();
    }
}

class Producer implements Runnable{

    private TestProducerConsumer pc;

    public Producer(){

    }
    public Producer(TestProducerConsumer pc){
        this.pc = pc;
    }

    public void run() {

        Stack<Integer> buf = pc.getBuffer();


        while(true){    
            synchronized(pc){               
                if(buf.size() < pc.MAX_SIZE){
                    try {
                        pc.addToBuffer(new Integer((pc.count)++));
                        if(buf.size() == 1){
                            CustomLogger.logger.info("Wake up consumer");
                            pc.notifyAll();
                        }
                    } catch (StackException e) {
                        CustomLogger.logger.info(e.getError());
                        break;              
                    }
                } else{
                    try {
                        CustomLogger.logger.info("Producer sleeping");
                        pc.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }

      }
    }

}

class Consumer implements Runnable{
    private TestProducerConsumer pc;

    public Consumer(TestProducerConsumer pc){
        this.pc = pc;
    }

    public void run(){

        Stack<Integer> buf = pc.getBuffer();
        int i;

        while(true){

            synchronized(pc){

                if(buf.size() == 0){

                    try {
                        CustomLogger.logger.info("Consumer Sleeping");
                        pc.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else
                {
                    try {
                        i = pc.removeFromBuffer();
                        CustomLogger.logger.info("poped "+i);

                        if(buf.size() == 0){
                            CustomLogger.logger.info("Wake up Producer");
                            pc.notifyAll();
                        }
                    } catch (StackException e) {
                        System.out.println(e.getError());
                        break;
                    }
                }
            }
        }
    }
}

class StackException extends Exception{
    private String reason;

    public StackException(){
        super();
    }
    public StackException(String reason){
        super(reason);
        this.reason = reason;
    }

    public String getError(){
        return reason;
    }

}
\$\endgroup\$
  • 4
    \$\begingroup\$ I wouldn't use a Stack for Producer/Consumer. I would just use a Queue or even better an ExecutorService. \$\endgroup\$ – Peter Lawrey May 20 '11 at 13:07
  • 2
    \$\begingroup\$ Let me discourage you from having a getBuffer() method which directly exposes the underlying data structure. If you want to control how the collection is modified through your own addToBuffer and removeFromBuffer methods, then you're completely sacrificing any guarantee of that by exposing buffer directly (allowing some outside code to call getBuffer().push). \$\endgroup\$ – Dan Tao May 20 '11 at 13:16
4
\$\begingroup\$

You could try using a queue as they are designed for this sort of thing. The code is much shorter.

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public class TestProducerConsumer {
    public static final int MAX_SIZE = 10;
    private final BlockingQueue<Integer> tasks = new ArrayBlockingQueue<Integer>(MAX_SIZE);
    public final ExecutorService executor = Executors.newCachedThreadPool();
    public final AtomicInteger count = new AtomicInteger();

    public static final int POISON_VALUE = -1;

    public void addToBuffer(Integer i) {
        try {
            tasks.put(i);
        } catch (InterruptedException e) {
            throw new AssertionError(e);
        }
    }

    public Integer removeFromBuffer() {
        try {
            return tasks.take();
        } catch (InterruptedException e) {
            throw new AssertionError(e);
        }
    }

    public static void main(String... args) {
        TestProducerConsumer pd = new TestProducerConsumer();
        pd.new Producer();
        pd.new Producer();
        pd.new Producer();

        pd.new Consumer();
        pd.new Consumer();
        pd.new Consumer();
        pd.new Consumer();
        pd.new Consumer();
    }

    class Producer implements Runnable {
        public Producer() {
            executor.execute(this);
        }

        public void run() {
            while (count.get() >= 0) {
                addToBuffer(count.getAndIncrement());
            }
            addToBuffer(TestProducerConsumer.POISON_VALUE);
        }
    }

    class Consumer implements Runnable {
        public Consumer() {
            executor.execute(this);
        }

        public void run() {
            Integer num;
            while ((num = removeFromBuffer()) != TestProducerConsumer.POISON_VALUE) {
                System.out.println("popped " + num);
            }
        }
    }
}
\$\endgroup\$
  • \$\begingroup\$ This implementation does not rise an exception if the queue is full, it blocks the request! (same for removeFromBuffer for an empty queue). -- I don't know if this is an improvement or not - any way it should be mentioned, that this implementation does not behave like the "original" one. \$\endgroup\$ – Ralph May 20 '11 at 14:18
  • \$\begingroup\$ It uses a queue instead of a Stack so it processes work in the order it is created. That is a more important difference. The StackException would only be thrown when there is a thread safety issue which won't happen for a BlockingQueue. \$\endgroup\$ – Peter Lawrey May 20 '11 at 14:33
3
\$\begingroup\$

Your program is not thread-safe.

For example, if two threads invoke addToBuffer(Integer i) at the same time, both can pass the if(buffer.size() < MAX_SIZE) check before one of them puts an item into the stack, therefore the stack can have more items than MAX_SIZE!

The same is for removeFromBuffer(), and combinations of add and remove will have there own unexpected behaviour.


In general instead of implementing it by your own have a look at the java.util.concurrent package

Queues

The java.util.concurrent ConcurrentLinkedQueue class supplies an efficient scalable thread-safe non-blocking FIFO queue. Five implementations in java.util.concurrent support the extended BlockingQueue interface, that defines blocking versions of put and take: LinkedBlockingQueue, ArrayBlockingQueue, SynchronousQueue, PriorityBlockingQueue, and DelayQueue. The different classes cover the most common usage contexts for producer-consumer, messaging, parallel tasking, and related concurrent designs. The BlockingDeque interface extends BlockingQueue to support both FIFO and LIFO (stack-based) operations. Class LinkedBlockingDeque provides an implementation.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.