Producer-Consumer using low level synchronization

Question

I saw a tutorial from a site where it showed the use of ArrayBlockingQueue as a thread-safe concurrent data-structure . Just for learning purposes , i tried to build something similar using synchronized and wait-notify methods in java. I'm really very new to java.util.concurrent and i'm hoping someone could guide me to some improvements and point out the bad parts of this code i wrote :

import java.util.ArrayList;
import java.util.Random;

public class LowLevelProducerConsumer {
    ArrayList<Integer> list = new ArrayList<Integer>();
    private final int size = 10;
    private int elems = 0;
    Object lock = new Object();
    Random r = new Random(System.currentTimeMillis());

    public void producer() {
        while (true) {
            synchronized (lock) {
                try {
                    while (elems != size) {
                        list.add(r.nextInt(100));
                        elems++;
                    }
                } finally {
                    // allows consumer to remove an entry
                    lock.notify();
                }
            }
        }
    }

    public void consumer() throws InterruptedException {
        Thread.sleep(100); // any better way to ensure producer runs first ?
        int ran = 0;
        while (true) {
            // consumer tries to acquire lock here , but it can do so only after
            // producer has called notify
            synchronized (lock) {

                // do i need a lock.wait() here somewhere ?

                ran = r.nextInt(10);

                if (ran == 7) { // just an arbitrary condition

                    int loc = r.nextInt(list.size());
                    int data = list.remove(loc);
                    System.out.format(
                            "%d removed from index %d , size : %d\n",
                            data, loc, list.size());
                    // decrementing elems to let the producer work again
                    elems--;
                    Thread.sleep(300);
                }

                // release the lock so that producer can fill things up again
                lock.notify();
            }
        }

    }

    public static void main(String[] args) {
        final LowLevelProducerConsumer low = new LowLevelProducerConsumer();
        Thread t1 = new Thread(new Runnable() {
            @Override
            public void run() {
                low.producer();
            }
        });
        Thread t2 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    low.consumer();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        t1.start();
        t2.start();
        try {
            t1.join();
            t2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}

ps : i'v kept my understanding of how the code works in the comments. Any correction to them would be much appreciated.

Answer 1

Miscellaneous Remarks

• Your elems variable is just list.size(), so why not just use list.size() everywhere and eliminate elems?
• Your other size is a constant, and should be called SIZE by convention.
• Instead of spreading out int ran = 0; ran = r.nextInt(10);, and if (ran == 7) { ... } over three lines, you should just do if (r.nextInt(10) == 7) { ... }. Putting it all on one line takes less mental effort to follow the code.

Threading

• Both the producer and consumer threads are infinite loops. Neither thread will end normally, and therefore trying to .join() them is superfluous.
• Rather than ensuring that the producer runs first, you should just design the consumer to correctly handle the case when there are no more values to consume. As @DanielR pointed out, when list.size() is zero, r.nextInt(list.size()) throws an IllegalArgumentException.

• Your use of lock.notify() right now is superfluous, as there is no corresponding lock.wait() that expects a notification. A good place to insert a lock.wait() would be where you suspected. (Without the lock.wait(), the consumer would keep trying frantically to fetch a value, which could work, but is inefficient. Inserting lock.wait() makes the consumer yield the processor to the producer until the producer notifies the consumer that the values have been replenished.)

  public void consumer() throws InterruptedException {
      synchronized (lock) {
          while (list.isEmpty()) {
              // Nothing to consume for now; wait for more.
              lock.wait();
          }
          ...
      }
  }
  

• I think that the vocabulary is slightly off. To me, a "producer" and a "consumer" are both threads. I would say,

  Thread producer = new Thread(new Runnable() {
      @Override
      public void run() {
          low.produce();
      }
  });
  

  Therefore, I would rename your methods as verbs produce() and consume().

• In Java, every Object can act as a lock. There is no need to make a separate lock object; you could just synchronize on list or this instead. (It doesn't matter which object you choose as the lock, as long as the producer and consumer both agree on which object it is.) As a special case,

  public void method() {
      synchronized (this) {
          ...
      }
  }
  

  can be written as

  public synchronized void method() {
      ...
  }
  

  which is slightly more readable and also compiles to slightly tighter bytecode.

• I would take move the while-loops from produce() and consume() into the run() method. That would have the triple benefit of

  • keeping produce() and consume() simpler
  • making it more obvious that each Runnable is an infinite loop
  • moving the synchronized keyword into the method signatures, making it easier to see that produce() and consume() run mutually exclusively, and generating slightly tighter bytecode.

Proposed Solution

import java.util.ArrayList;
import java.util.Random;

public class LowLevelProducerConsumer {
    ArrayList<Integer> list = new ArrayList<Integer>();
    private final int SIZE = 10;
    Random r = new Random(System.currentTimeMillis());

    public synchronized void produce() {
        try {
            while (list.size() != SIZE) {
                list.add(r.nextInt(100));
            }
        } finally {
            // Notifies consumer that entries have been generated
            this.notify();
        }
    }

    public synchronized void consume() throws InterruptedException {
        while (list.isEmpty()) {
            // Nothing to consume for now; wait for more.
            // System.err.println("Consumer waiting...");
            this.wait();
        }

        if (r.nextInt(10) == 7) { // just an arbitrary condition
            int loc = r.nextInt(list.size());
            int data = list.remove(loc);
            System.out.format(
                    "%d removed from index %d , size : %d\n",
                    data, loc, list.size());
            Thread.sleep(300);
        }
    }

    public static void main(String[] args) {
        final LowLevelProducerConsumer low = new LowLevelProducerConsumer();
        Thread producer = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    low.produce();
                }
            }
        });
        Thread consumer = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        low.consume();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        producer.start();
        consumer.start();
    }
}

Answer 2

This line is unsafe:

int loc = r.nextInt(list.size());

an error is raised if list.size() is 0.

Your code implements some very strange producer consumer behaviour, especially the random value at the consumer without any waiting time. Please describe in words the intended behaviour.