Consumer/Producer (Concurrency) - Exception Handling

Question

I have a circular buffer (array/first in first out), a consumer and a producer. A producer puts random numbers into the array, and a consumer takes the first number and checks if it is relative prime.

My code works, and I think it works correct, but I want to improve it. I am not really sure about my "void run" methods. Should I anywhere else do the exception handling? Change the "infinite loop"? The method signatures should not be changed (they are predefined).

I would be happy about each suggestion to improve the code. (Don't care the visibility (public,...), and static things, I've just put it together in one file.

import java.math.BigInteger;
import java.util.Random;

public class ConsProd {

    static class CircularBuffer {

        private BigInteger[] buffer;
        //"pointers"
        private int read;
        private int write;

        public CircularBuffer(int size) {
            this.buffer = new BigInteger[size];
            this.read = 0;
            this.write = 0;
        }

        public boolean isFull() {
            for(int i = 0; i < buffer.length; i++) {
                if(buffer[i] == null)
                    return false;
            }
            return true;
        }

        public boolean isEmpty() {
            for(int i = 0; i < buffer.length; i++) {
                if(buffer[i] != null)
                    return false;
            }
            return true;    
        }

        public synchronized void put(BigInteger element) throws InterruptedException {
            while(isFull()){
                wait();
            }
            buffer[write] = element;
            write = (write+1)%buffer.length;
            notifyAll();
        }

        public synchronized BigInteger take() throws InterruptedException {
            while(isEmpty()){
                wait();
            }
            BigInteger temp = buffer[read];
            buffer[read] = null;
            read = (read+1)%buffer.length;
            notifyAll();
            return temp;
        }   
    }

    static class Consumer implements Runnable {

        private int id;
        private CircularBuffer buffer;

        public Consumer(int id, CircularBuffer b) {
            this.id = id;
            this.buffer = b;
        }

        private void consume(BigInteger e) {
            synchronized(e){
                System.out.println("consumer " + id + " retrieved: " + e);
                if (e.isProbablePrime(100)) {
                    System.out.println("     -----> probably prime!");
                }
            }
        }

        @Override
        public void run() {
            try { // TODO is this the right place to handle the exception? 
                while (!Thread.currentThread().isInterrupted()) {
                    BigInteger e = buffer.take();
                    consume(e);
                }
            } catch (InterruptedException e) { }
        }

    }

    static class Producer implements Runnable {

        private int id;
        private CircularBuffer buffer;

        public Producer(int id, CircularBuffer b) {
            this.id = id;
            this.buffer = b;
        }

        protected BigInteger produce() {
            BigInteger x = new BigInteger(10, new Random());
            System.out.println("producer " + id + " produced:  " + x.toString());
            return x;
        }

        @Override
        public void run() {

            try { // TODO is this the right place to handle the exception? 
                while (!Thread.currentThread().isInterrupted()) {
                    BigInteger e = produce();
                    buffer.put(e);
                }
            } catch (InterruptedException e) { }
        }
    }


    public static void main(String[] args) {

        CircularBuffer cbuf = new CircularBuffer(4);

        Thread t1 = new Thread(new Consumer(1, cbuf));
        Thread t2 = new Thread(new Consumer(2, cbuf));
        Thread t3 = new Thread(new Consumer(3, cbuf));

        Thread t4 = new Thread(new Producer(1, cbuf));
        Thread t5 = new Thread(new Producer(2, cbuf));

        t1.start();
        t2.start();
        t3.start();

        t4.start();
        t5.start();
    }
}

Answer 1

Below are my suggestions grouped by class and the code including these suggestions at the end.

CircularBuffer

I would advice using an implementation of BlockingQueue (for your case ArrayBlockingQueue seems appropriate). Nevertheless, below are my suggestions to improve this class :

• Use a specific variable for the number of available elements

You know the total number of elements of your array at construction and it does not change. Moreover you have one method that increases and one that decreases the number of available elements and they are both synchronized. So it is trivial to add a variable that keeps track of the number of available elements. This variable can be used to drastically simplify and improve the methods isFull or isEmpty.

• Use Lock for synchronisation

I am not a fan of synchronized keyword as it exposes your lock (the instance of the class). If, for some reason, someone synchronizes his code on an instance of CircularBuffer you might be in trouble.

To hide your lock, you can define a final parameter in your class like :

private final Object lock = new Object();

and then use synchronized(lock){...} blocks, as well as the wait and notifyAll methods of this object for synchronization. The method put will then be :

public void put(BigInteger element) throws InterruptedException {
    synchronized (lock) {
        while (isFull()) {
            lock.wait();
        }
        buffer[write] = element;
        write = (write + 1) % buffer.length;
        lock.notifyAll();
    }
}

But I would suggest the use of the framework in the package java.util.concurrent.locks. It offers more functionalities and specifically Condition: check its javadoc for a solution using conditions of the very same problem you faced with empty and full CircularBuffer.

Consumer

• Remove sychronization

In the method consume, you synchronize on the parameter of the method. I am not sure why. I assume this is to ensure that the BigInteger is not modified during the prime check. If that so, then the synchronization is useless: instances of BigInteger are immutable and therefore thread safe.

Moreover, synchronizing on method parameter does not seem to be a good idea. I lack arguments about that, but I feel this should be avoid (maybe someone could back me up one this part).

Consumer and Producer

• Do not swallow InterruptedException

They are not just annoying exceptions, they mean something and you need to take action accordingly. Someone did a better job than I would ever do to explain this issue, so check his answer on stackoverflow.

• Factorize code

This might not be completely relevant here, but you have some common code between these two classes : same members, same constructor and almost same method run. You could create an abstract class with all this common code and then extends Consumer and Producer from this class.

For all three classes

• Use the final keyword on appropriate class member

If a class member is defined at construction and should not change afterward, you should define it as final to increases readability.

This also true for variable in method...

• Use more explicit variable names

Try to use more specific variable names than 'b' or 'e'. In the CircularBuffer you have two variables with the very generic names read and write and a comment

//pointers

Why not calling these variables readPointer and writePointer directly. The code will be easier to read later.

Exception Handling

In you case you could simply log the exception and continue producing or consuming. If you want a more sophisticate exception handling, you need some more code but first you need to define what should happen when you encounter an exception: should you ignore it if it happens in a Consumer, should you stop everything if it happens in a Producer or should warn a third component that will decide what to do...

Below is the code with most of the suggestions above.

import java.math.BigInteger;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ConsProd {

    /**
     * Common class for procuder and consumer of a integer fifo.
     * The child class must implements the {@link #execute()} method as it would implement the {@link Runnable#run()}
     * method.
     */
    public static abstract class ConsProdRunnable implements Runnable {

        private final int id;
        private final BlockingQueue<BigInteger> integerQueue;

        public ConsProdRunnable(int id, BlockingQueue<BigInteger> integerQueue) {
            this.id = id;
            this.integerQueue = Objects.requireNonNull(integerQueue, "integerQueue must not be null");
        }

        protected int getId() {
            return id;
        }

        protected BigInteger takeIntegerFromQueue() throws InterruptedException {
            return integerQueue.take();
        }

        protected void putIntegerIntoQueue(final BigInteger value) throws InterruptedException {
            this.integerQueue.put(value);
        }

        protected abstract void execute() throws InterruptedException;

        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    execute();
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception exception) {
                    //the method execute should take care of its own exception
                    //if something is catched here, this is probably a bug.
                    //for the moment log it and continue
                    exception.printStackTrace(); //a Log framework would be better
                }
            }
        }
    }

    public static class Consumer extends ConsProdRunnable {

        public Consumer(int id, BlockingQueue<BigInteger> integerQueue) {
            super(id,integerQueue);
        }

        protected void execute() throws InterruptedException {
            final BigInteger value = takeIntegerFromQueue();
            System.out.println("consumer " + getId() + " retrieved: " + value);
            if (value.isProbablePrime(100)) {
                System.out.println("     "+value+" -----> probably prime!");
            }
        }
    }

    static class Producer extends ConsProdRunnable {

        public Producer(int id, BlockingQueue<BigInteger> integerQueue) {
            super(id, integerQueue);
        }

        @Override
        protected void execute() throws InterruptedException {
            final BigInteger randomValue = createRandomBigInteger();
            System.out.println("      producer " + getId() + " produced:  " + randomValue.toString());
            this.putIntegerIntoQueue(randomValue);
        }

        private BigInteger createRandomBigInteger() {
            return new BigInteger(10, new Random());
        }

    }


    public static void main(String[] args) {

        final BlockingQueue<BigInteger> queue = new ArrayBlockingQueue<>(4);

        Thread t1 = new Thread(new Consumer(1, queue));
        Thread t2 = new Thread(new Consumer(2, queue));
        Thread t3 = new Thread(new Consumer(3, queue));

        Thread t4 = new Thread(new Producer(1, queue));
        Thread t5 = new Thread(new Producer(2, queue));

        t1.start();
        t2.start();
        t3.start();

        t4.start();
        t5.start();
    }
}